feat(netflow-db): add coordinated active-source subsets

.gitignore

@@ -44,3 +44,8 @@ scripts/local/ugr16-csv.pipeline-v2.json
 **/playwright-mcp
 data/**
 .codex
+
+# Local native helper artifacts
+/tools/netflow-db/maad_fast
+/tools/netflow-db/nfdump_reducer
+/tools/netflow-db/nfdump_reducer.build-id

docs/code/pipeline-contract.md

@@ -26,6 +26,49 @@ Coverage is observed before selection. Thus, selected-out buckets remain as dens
 
 Native nfcapd input pushes the IP prefix condition into the nfdump filter. Visibility conditions apply before statistics accumulate.
 
+`daily_active_sources` is a separate, fixed selection policy for the UOregon `/16` candidate
+products. It accepts exactly one IPv4 `/16` and exactly one `nfcapd_tree` input. For each complete
+local calendar day, it makes two bounded passes over every unique physical member:
+
+1. Select anonymized IPv4 source traffic in the `/16` that uses TCP or UDP and source port 1024 or
+   greater. Sum flows, packets, and bytes by exact source address across physical members.
+2. Mark sources active at the inclusive thresholds of 3 flows, 20 packets, and 2,000 bytes. Publish
+   only the same qualifying-flow population from active sources into the existing five-minute and
+   rollup contracts.
+
+There is no destination-port filter and no TCP-flag or SYN filter. Overlapping logical sources do
+not double-count activity because the first pass deduplicates their physical members. A day with
+any missing physical capture is not published. Activity resets at local midnight, including DST
+days.
+
+The normalized product identity records the entire fixed policy. It is not compatible with an old
+prefix-only database. A late or changed input can alter the active set for every five-minute bucket
+in a day, so repair requires a whole-day `--force` rebuild inside the existing day transaction.
+
+## Coordinated subset runs
+
+Repeat `--dataset` for two or more registry entries to build coordinated subset products. Each entry
+supplies its own root and source configuration. Entries do not point to a parent dataset or a
+`source_dataset`.
+
+Multi mode supports only `daily_active_sources`. The selected entries must resolve to the same
+nfcapd root and logical source layout, and one whole local-day window. The command rejects `--config`,
+`--database-path`, partial time bounds, and ambiguous selection overrides. It takes each output path
+from the corresponding registry entry.
+
+The run coordinates one daily eligibility scan and one publication scan across the subsets. These
+remain two physical phases because qualification needs the complete local day before any bucket can
+publish. Local-day completeness is shared, so an incomplete required day blocks publication for every
+subset.
+
+Each subset keeps its own immutable product database, product identity, transactions, resume state,
+and MAAD configuration. Active sets are still resolved independently. Overlapping subsets may both
+receive the same qualifying flow.
+
+Outputs commit sequentially rather than as one cross-database transaction. If the process stops
+between product commits, sibling databases can differ by at most the local day that was in flight.
+The next run sees the missing completion marker and rebuilds that day for each unfinished product.
+
 ## Input identity
 
 Each input records an exact revision. The revision contains a SHA-256 content identity and a canonical decoder fingerprint.
@@ -81,6 +124,16 @@ The fork's stdout is the private `atlantis-flow-stream-v1` binary contract. The
 
 The pipeline stops when the fork executable is absent or incompatible.
 
+Request resolution stores the executable's canonical path, one SHA-256 content identity, and a
+cheap device/inode/size/timestamp snapshot. The binary identity is part of the product config and
+the native input decoder fingerprint, so replacing the executable requires a fresh product and
+cannot mix revisions in a resumed database. The snapshot is rechecked around activity and decode
+scans and immediately before native publication commits; a change rolls that transaction back.
+
+Before creating an output directory, lock, or database, the pipeline runs one bounded probe against
+an isolated empty `-R` directory. The probe requires the exact empty Atlantis stream (header,
+terminator, and EOF), including for incomplete-day requests.
+
 To update the fork, rebase its `atlantis-binary-v1` branch onto a reviewed upstream nfdump tag and run the fork's serial test suite. Then advance this repository's submodule pointer and run `./vendor/scripts/compile-nfdump.sh`. Treat a protocol or normalization change as a versioned wire-contract change. Update the Rust decoder and the provenance revision in the same change.
 
 ## Analysis-window exports

docs/user/datasets.md

@@ -61,6 +61,41 @@ A logical source combines the captures from more than one collector directory. E
 }
 ```
 
+## Define coordinated subsets
+
+Give each subset its own registry entry. Repeat those dataset IDs in one pipeline command. Each entry
+defines its own logical sources and `daily_active_sources` selection.
+
+```json
+[
+  {
+    "dataset_id": "campus-a",
+    "root_path": "/data/netflow/campus",
+    "source_ids": ["router-a"],
+    "selection": {
+      "kind": "daily_active_sources",
+      "ip_prefix": "0.220.0.0/16"
+    },
+    "db_path": "data/campus-a/netflow.sqlite"
+  },
+  {
+    "dataset_id": "campus-b",
+    "root_path": "/data/netflow/campus",
+    "source_ids": ["router-a"],
+    "selection": {
+      "kind": "daily_active_sources",
+      "ip_prefix": "0.221.0.0/16"
+    },
+    "db_path": "data/campus-b/netflow.sqlite"
+  }
+]
+```
+
+The entries share a capture root and the same logical source layout. Their active sets remain independent,
+so one flow may publish to both products when the selections overlap. Do not add a parent or
+`source_dataset` relation. The multi-dataset command infers each subset from the selected entry and
+uses each entry's `db_path`.
+
 ## Required fields
 
 | Field        | Purpose                                         |
@@ -80,8 +115,11 @@ A logical source combines the captures from more than one collector directory. E
 | `source_ids`         | None                               | Simple source names for datasets without member directories.                                         |
 | `discovery_mode`     | `static`                           | `live` marks a dataset that continues to receive new captures. `static` marks a complete dataset.    |
 | `sort_order`         | `0`                                | The dataset order in the dashboard. Lower values sort first.                                         |
+| `selection`          | All flows                          | A normalized flow-selection object applied automatically by dataset-mode pipeline runs.              |
 
 Set `db_path` only for a database that must stay separate, such as a [flow selection](setup-pipeline.md#select-flows) product.
+Persist `selection` with a dedicated `db_path` when the dataset is itself a selected product. Command-line
+selection flags may only override it when `--database-path` names a different output product.
 
 Each pipeline run calculates `default_start_date` again. A run that adds earlier days moves the date back. Set the field to hold the dashboard at one date.
 

docs/user/setup-pipeline.md

@@ -64,6 +64,44 @@ Dataset mode calculates MAAD statistics by default. MAAD statistics describe the
 
 If a command fails, read [Troubleshooting](troubleshooting.md).
 
+## Process coordinated subsets
+
+Repeat `--dataset` for two or more registry entries that select subsets of one nfcapd tree.
+Native runs must name the pinned ATLANTIS nfdump fork explicitly:
+
+```bash
+./scripts/netflow-db.sh pipeline \
+  --nfdump target/nfdump/libexec/nfdump \
+  --dataset campus-a \
+  --dataset campus-b \
+  --start-date <YYYY-MM-DD> \
+  --end-date <YYYY-MM-DD>
+```
+
+Coordinated mode accepts only registry-backed `daily_active_sources` products. It rejects a run
+unless all selected entries have compatible inputs and execution settings:
+
+- Dataset IDs and output database paths must be unique.
+- Every entry must use the same canonical nfcapd root, logical source layout, and timezone.
+- Every entry must use the same whole local-day window, force setting, MAAD setting, coverage
+  setting, and nfdump executable revision.
+- Each entry supplies its own `daily_active_sources` prefix and `db_path`.
+- Output databases, locks, and sidecars must not overlap one another or the capture tree.
+
+Do not combine repeated `--dataset` with `--config`, `--database-path`, `--start-time`, `--end-time`,
+or command-line selection flags. Configure selection and output paths in `datasets.json`. The
+command has no parent dataset or `source_dataset` relation.
+
+The pipeline discovers the capture plan once, scans each day once, and fans the decoded flow stream
+out to the selected products. A missing required capture leaves that day unpublished for every
+product. Each product still has its own identity, active-source set, transaction, and completion
+marker, so overlapping prefixes may contain the same qualifying flow.
+
+After a successful run, repeating the exact command is a no-op. The report says
+`Published five-minute buckets: 0`, and the pipeline does not rewrite completed days. If a previous
+run stopped between product commits, the next run rebuilds only the unfinished day for the affected
+product.
+
 ## Select flows
 
 Selection conditions use AND logic. The IP prefix can match the source endpoint or the destination endpoint.
@@ -79,13 +117,40 @@ Selection conditions use AND logic. The IP prefix can match the source endpoint
 ```
 
 A selected population is a different database product. Thus, selection options require an explicit `--database-path`.
+Dataset registry entries may instead persist a `selection` beside their dedicated `db_path`; dataset
+mode applies that selection automatically.
 
 Available selection options are:
 
 - `--ip-prefix`
+- `--daily-active-sources`
 - `--src-visibility literal|anonymized`
 - `--dst-visibility literal|anonymized`
 
+`--daily-active-sources` applies the fixed active-user definition used to choose the UOregon
+candidate subnets. It requires an IPv4 `/16` and cannot be combined with the visibility flags:
+
+```bash
+./scripts/netflow-db.sh pipeline \
+  --dataset example \
+  --start-date <YYYY-MM-DD> \
+  --end-date <YYYY-MM-DD> \
+  --database-path data/example-active/netflow.sqlite \
+  --ip-prefix 0.220.0.0/16 \
+  --daily-active-sources
+```
+
+For each complete local day, the pipeline sums qualifying traffic by exact source address across
+each unique physical capture member. A source is active when it has at least 3 flows, 20 packets,
+and 2,000 bytes that day. Qualifying traffic is IPv4 TCP or UDP from an anonymized source in the
+target `/16`, with source port at least 1024. Destination ports and TCP flags are unrestricted.
+Only that qualifying traffic from active sources is published.
+
+This mode supports exactly one `nfcapd_tree` input and whole local days. A day missing any expected
+physical capture is skipped rather than published as zero. If input evidence changes after a day
+was published, rebuild the whole day with `--force`; a single five-minute repair is not safe because
+it can change the active-source set for every bucket in that day.
+
 ## Use a pipeline configuration
 
 Configuration mode supports CSV input, nfcapd input, and mixed input. Explicit `csv` and `nfcapd` inputs and `csv_tree` and `nfcapd_tree` discovery inputs go in the top-level `inputs` list.
@@ -109,6 +174,27 @@ Put flow selection in the top-level `selection` object:
 }
 ```
 
+The equivalent active-source selection is deliberately a named policy rather than configurable
+thresholds:
+
+```json
+{
+  "selection": {
+    "kind": "daily_active_sources",
+    "ip_prefix": "0.220.0.0/16"
+  },
+  "inputs": [
+    {
+      "input_kind": "nfcapd_tree",
+      "root_path": "/path/to/captures",
+      "source_ids": ["gateway-a", "gateway-b"],
+      "start_date": "2025-06-01",
+      "end_date": "2026-06-29"
+    }
+  ]
+}
+```
+
 On the native path, nfcapd input needs the fork path: set the top-level `"nfdump"` value to `"target/nfdump/libexec/nfdump"`, or pass `--nfdump` when the configuration does not set it.
 
 ## Common options

tools/netflow-db/Cargo.toml

@@ -26,6 +26,7 @@ fs2 = "0.4"
 ipnet = { version = "2", features = ["serde"] }
 jiff = "0.2"
 libc = "0.2"
+nix = { version = "0.27", default-features = false, features = ["fs"] }
 parquet = { version = "59.2", default-features = false, features = ["arrow", "zstd"] }
 rayon = "1"
 regex = "1"