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 published files through the plugin instead of to a traditional results/ directory tree. The plugin provides a lineage store for lineage.store.location = 'cas://lab', so Nextflow's own lid:// records live in the same store. It supports content-addressed storage with DAG-CBOR lineage records, SQLite indexing, and features like the fromStore channel factory for reading outputs in downstream pipelines and an explorer for browsing stored content and managing Selections.

Status: beta (version 0.11.0-beta.1). The plugin supports Milestone 4 (cloud) with local directories or S3 locations, Milestone 5 (nf-core pipelines) with workflow output lineage recording, and Milestone 6 (retention) with configurable content retention and garbage collection. It requires the nf-amazon plugin (≥3.9.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.11.0-beta.1'
}

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. A value of anonymous counts as unset. See docs/identity.md for the order and what each source reads.

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.

To add nf-blocks to someone else's pipeline, such as an nf-core one, with a -c config file: a plugins { } block in that file replaces the pipeline's own plugin list rather than adding to it. Repeat the pipeline's pinned plugins beside nf-blocks, or Nextflow installs the latest release of each one the pipeline includes.

Examples

Any pipeline with a workflow output block publishes into the store. With the configuration above, running a pipeline 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", ...}}

A second pipeline reads a run's outputs back out of the store with fromStore. Its config names the producer's store as a read-only member beside its own writable one:

plugins {
    id 'nf-blocks@0.11.0-beta.1'
}

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']
}

fromStore is used in a pipeline script:

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()
}

Each item arrives in the shape the producer published it, here [meta, bam], with every file a cas:// path Nextflow stages like any other.

For publishing into S3:

plugins {
    id 'nf-blocks@0.11.0-beta.1'
}

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'
}

Browse a store with:

nextflow plugin nf-blocks:explore

This serves a live view of every configured member at http://127.0.0.1:<port>/.

Command-line tools for managing content and retention:

nextflow plugin nf-blocks:items aligned nested.kit=truseq --run latest --pipeline cas-test-pipeline

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> ...]

For typed scripts:

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()
}

License

Apache License 2.0. See the COPYING file for details.