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, so outputDir = 'cas://lab' sends every published file through the plugin instead of onto a results/ tree, and it provides a lineage store for lineage.store.location = 'cas://lab', so Nextflow's own lid:// records live in the same store. It is built and tested against released Nextflow 26.04.6. A run writes DAG-CBOR lineage records into the store beside Nextflow's own lid:// records, and keeps a SQLite index of them; another pipeline reads the outputs back with the fromStore channel factory, and nf-blocks:explore browses them and saves Selections.

Status: beta (0.8.0-beta.1)

This plugin supports:

• Cloud environments with any member as a local directory or S3 location
• nf-core pipelines with lineage recording from workflow outputs
• Retention policies with release, pin, and sweep operations
• Dependency on 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.8.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 = 'anonymous'          // optional opaque label; never the OS user 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.

Important: When adding nf-blocks to someone else's pipeline with a -c config file, a plugins { } block replaces the pipeline's own plugin list. Repeat the pipeline's pinned plugins beside nf-blocks, or Nextflow installs the latest release of each one.

Examples

Basic Usage

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

Publishing into S3

A writable S3 member takes an s3://<bucket>[/<prefix>] location:

plugins {
    id 'nf-blocks@0.8.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'
}

Reading Outputs in Another Pipeline

A second pipeline reads a run's outputs back out of the store with fromStore:

plugins {
    id 'nf-blocks@0.8.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']
}

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

Using fromStore in Typed Scripts

In a typed script, use nextflow.Channel.fromStore() 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()
}

Browsing and Managing Store Content

Start the explorer to browse configured members:

nextflow plugin nf-blocks:explore

Serve the store at http://127.0.0.1:<port>/, with port --port or ephemeral.

Managing Selections from Command Line

Query and save items as named Selections:

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

Space Management

Release, pin, and prune runs:

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 --budget '10 GB'

nextflow plugin nf-blocks:untrash <cid>

Plugin Development

Build the plugin:

make assemble

Run tests:

make test    # unit tests
make check   # unit tests plus dependency check and memory bound
make smoke   # builds, installs and runs a real pipeline

Publish to the Nextflow Registry:

make release

License

Apache License 2.0. See the COPYING file for details.

What's new

Status: beta (0.8.0-beta.1) — Pin the version in plugins { }, as the plugin is still in beta. The on-disk format (block kinds, the index, the Store Log) may still change before 1.0. A change to the blocks will come with a migration, since a block's address is its content and existing stores keep their blocks.

Supported features include:

• Milestone 4 (cloud): any member, writable included, is a local directory or S3 location
• Milestone 5 (nf-core pipelines): lineage recording from workflow outputs only
• Milestone 6 (retention): content is a root by default and kept until released
• Node hashing under Fusion to avoid head-node processing
• Index Snapshot seeding for cold head nodes
• Selections for curating Output Items across runs
• S3 lifecycle rules and bucket policies for immutability and cost management