Summary

nf-blocks is a Nextflow plugin that publishes pipeline outputs into a content-addressed store and records their lineage there. It owns the cas URI scheme, allowing users to configure outputDir = 'cas://lab' to send every published file through the plugin instead of onto a results/ tree. It provides a lineage store for lineage.store.location = 'cas://lab', so Nextflow's own lid:// records live in the same store. The plugin is built and tested against Nextflow 26.04.6 and writes DAG-CBOR lineage records into the store alongside Nextflow's lid:// records, maintaining a SQLite index. Another pipeline can read outputs back with the fromStore channel factory, and nf-blocks:explore provides a browser interface to explore outputs and save Selections. Status: beta (0.10.0-beta.2).

Get Started

Enable the plugin and point both the lineage store and the output directory at the same store alias:

plugins {
    id 'nf-blocks@0.10.0-beta.2'
}

lineage.enabled = true
lineage.store.location = 'cas://lab'   // alias of the writable member
outputDir = 'cas://lab'                // optional: unset, it is the lineage alias

cas {
    stores {
        lab {
            location = '/data/cas'     // the writable member
        }
    }
    resolve = ['lab']                  // optional; default is every alias, writable first
    asserted_by = 'ada@example.org'    // optional; who the blocks say asserted them
}

asserted_by is optional. Left unset, nf-blocks uses your Seqera Platform identity, then NXF_BLOCKS_USER, then git config user.email, and warns if none of them gives a name.

An alias matches ^[a-z][a-z0-9_-]{0,31}$ and must not parse as a content address. outputDir may be left out: unset, it means the alias in lineage.store.location. When set, it must name that alias, or the run aborts.

Nextflow downloads the plugin from the Nextflow Registry the first time the pipeline runs. Note that a plugins { } block in a config file replaces the pipeline's own plugin list rather than adding to it, so repeat the pipeline's pinned plugins beside nf-blocks to avoid Nextflow installing the latest release of each one.

Examples

Any pipeline with a workflow output block publishes into the store. With the configuration above, running it writes each published path under the store's coordinate tree:

/data/cas/coords/aligned/A/A.bam
/data/cas/coords/qc/A/A_qc/summary.txt
/data/cas/nf/<run hash>/aligned/A/A.bam/.data.json

The lineage record for a published file names its coordinate rather than a work directory path:

{"kind":"FileOutput","spec":{"path":"cas://lab/aligned/A/A.bam", ...}}

For publishing into S3, configure a writable S3 member:

plugins {
    id 'nf-blocks@0.10.0-beta.2'
}

lineage.enabled = true
lineage.store.location = 'cas://lab'

workDir = 's3://my-work-bucket/work'
process.executor = 'awsbatch'
process.queue = 'my-queue'

aws {
    region = 'us-east-1'
    profile = 'my-profile'
}

cas {
    stores {
        lab { location = 's3://my-cas-bucket/cas' }
    }
    tmpDir = '/scratch/nf-blocks'
}

For reading outputs in another pipeline, configure the producer's store as a read-only member:

plugins {
    id 'nf-blocks@0.10.0-beta.2'
}

manifest.name = 'downstream'

lineage.enabled = true
lineage.store.location = 'cas://mine'

cas {
    stores {
        mine { location = '/data/cas-downstream' }
        lab  { location = '/data/cas' }
    }
    resolve = ['mine', 'lab']
}

Use fromStore to retrieve outputs:

include { fromStore } from 'plugin/nf-blocks'

workflow {
    main:
    picked  = channel.fromStore(selection: '<selection address>')
    aligned = channel.fromStore(run: 'latest', pipeline: 'cas-test-pipeline', output: 'aligned', where: [sample: 'B'])

    picked.mix(aligned).view()
}

For typed scripts, call through nextflow.Channel with records: true:

nextflow.enable.types = true

include { fromStore } from 'plugin/nf-blocks'

record Sample {
    sample: String
    lane: Integer
}

process COUNT_LINES {
    input:
    tuple(meta: Sample, bam: Path)

    output:
    stdout()

    script:
    """
    printf '%s\\t' '${meta.sample}'
    wc -l < '${bam}'
    """
}

workflow {
    COUNT_LINES(nextflow.Channel.fromStore(selection: '<selection address>', records: true))
        .view()
}

Browse a store with:

nextflow plugin nf-blocks:explore
nextflow plugin nf-blocks:snapshot
nextflow plugin nf-blocks:reindex

Manage retention and deletion:

nextflow plugin nf-blocks:prune --keep-last 3
nextflow plugin nf-blocks:prune --keep-last 3 --apply true
nextflow plugin nf-blocks:sweep
nextflow plugin nf-blocks:sweep --apply true
nextflow plugin nf-blocks:untrash <cid> [<cid> ...]

Create and manage Selections from the command line:

nextflow plugin nf-blocks:items aligned nested.kit=truseq --run latest --pipeline cas-test-pipeline
nextflow -q plugin nf-blocks:items aligned lane=2 --run lid://<run hash> --format selection \
  | nextflow -q plugin nf-blocks:put /dev/stdin --name lane-2

License

Apache License 2.0. See the COPYING file for details.