feat(netflow-db): add merge-shards and a cluster launcher for day-sharded reprocessing

[flamboh]

Note

🤖 Claude Opus 5.5 on behalf of Oliver

ELI5

A full reprocess takes days on one machine. This lets several machines each build a slice of days and then stitches the slices into one database, checking that the result matches a single-machine run.

Why

pipeline is one process with one SQLite writer. A production day takes ~16 min and uses only 4–5 of 24 cores, so a year of data is ~100 h single-process. Local days are independent: rollups max out at 1d, and every rollup bucket is aligned to local midnight. Day ranges can therefore be built separately and merged.

What changes

• netflow-db merge-shards --output OUT SHARD... merges day-sharded products built with identical flags and an explicit --end-date into a new product. It copies completion markers, so running pipeline on the merged product is a no-op for the merged days.
• --consume keeps peak disk near the output size plus one shard. It moves the first shard into the temporary output and deletes each other shard right after its rows commit.
• scripts/netflow-db-cluster.sh splits a date range across host:N slots. It runs each shard detached over plain SSH, copies snapshots back, merges with --consume by default and runs verify. All hosts must use the same --remote-dir path.
• Compact MAAD tables (feat: store MAAD results as compact f32 rows #119). This PR is stacked on feat: store MAAD results as compact f32 rows #119. address_maad_stats rows are day-owned and merged like other stats. maad_q_grid is shared: every shard must hold identical (ip_version, q_min, q_step, q_count) rows before the merge keeps the first shard's copy. With MAAD disabled, all shards hold an empty grid. The merge checks the table contract with storage::product_schema() and no longer keeps its own copy. Any table that is neither day-owned nor shared is refused.
• Docs: a cluster section in docs/user/setup-pipeline.md, and "Day-sharded products" in docs/code/pipeline-contract.md.

Review path

Setup: any nfcapd tree with at least three days and a dataset registry entry, e.g. datasets.json.example. Build each shard with identical flags and an explicit --start-date/--end-date.

1. Plain merge. Build a.sqlite (day 1) and b.sqlite (days 2–3), then run netflow-db merge-shards --output m.sqlite b.sqlite a.sqlite.
   • Expect a per-table row summary.
   • netflow-db verify m.sqlite --dataset-id <id> --require-data --require-maad-data --require-processed --require-rollup-parity --require-no-raw-ip prints OK.
   • Rerunning pipeline over days 1–3 on m.sqlite reports Published five-minute buckets: 0.
2. Refusals. Merge overlapping shards, shards built with a different MAAD setting, or a shard whose maad_q_grid was edited (for example UPDATE maad_q_grid SET q_step = q_step * 2).
   • Expect a refusal and no output or temporary file left behind.
3. Consuming merge and recovery. Build one shard per day, then make the day-3 shard fail its insert, e.g. with a BEFORE INSERT trigger on traffic_stats. Run merge-shards --consume with the shards in the order day 2, day 1, day 3.
   • The error names the day-1 and day-2 shards as consumed. It keeps a partial .merge.tmp product and prints a resume command.
   • The partial's datasets.default_start_date is day 1.
   • After removing the trigger, the printed resume command produces a product identical to a single run.
4. Launcher allocation. Run the launcher with --hosts nodeA:4,nodeB over 6 days, using real hosts or the fake ssh/scp in tools/netflow-db/tests/cluster_launcher.rs.
   • The plan is 5 shards: 2+1+1+1+1 days on nodeA, nodeB, nodeA, nodeA, nodeA.
   • Over 3 days the plan is nodeA, nodeB, nodeA.

Decisions and edge cases

• Launcher split. The range splits into min(days, slots) contiguous shards whose lengths differ by at most one day. Slots are ordered round-robin across hosts, so every listed host gets work whenever the range has at least as many days as there are hosts. host:N gives a host N slots, so leaving :N off a busier host gives it a lighter share instead of skipping it.

• default_start_date policy. For each dataset:

  • If every shard stores the same date, it is kept. This covers configured dates.
  • Otherwise, every shard must store the date inferred from its own earliest five-minute traffic, or the fallback date if it has no traffic. Any other mismatch is refused.
  • The merged product then takes the date of its earliest merged traffic, and the fallback only if the whole product has no traffic. A traffic-less shard's fallback date no longer wins.
  • The date is recomputed inside each shard's insert transaction, so a partial product from a failed consuming merge always carries the date of the days it holds, whatever the shard order.

• What counts as consumed. A shard counts as consumed the moment its data is inside the temporary output: after the first shard's rename (or cross-filesystem copy and sync), or after a later shard's commit. Any later failure keeps the partial and lists that shard as consumed, so the resume command never re-includes it. This covers a directory sync, DETACH, or deleting the shard's files. A failed file deletion is reported as its own cleanup error that names the leftover files, which are safe to delete.

• Failure after publication. If the directory sync fails after the output is renamed into place, the error says the product is published and asks you to run verify. It does not report the temporary path, which no longer exists.

• Recovery guarantees, as tested. Tests cover:

  • an insert failure midway, with shards in reverse-chronological order;
  • a deletion failure right after the first shard's rename;
  • a deletion failure right after a later shard's commit.

  In each case no data is lost, the partial holds exactly the consumed shards, and resuming produces a product identical to a single run. Durability across a crash or power loss rests on SQLite's rollback journal with synchronous=FULL and explicit file and directory syncs, and is not tested. The post-publication sync failure is not covered by a test.

• The nfdump path is part of product identity, so every host installs it under the same absolute --remote-dir.

• processed_inputs.file_device differs per node on network filesystems, and resume decisions ignore it.

• A launcher rerun re-copies every shard from the hosts and deletes a stale partial merge first. Remote shards survive an SSH drop or a launcher interrupt, and a rerun reattaches to shards that are still running.

Verification

• Automated:
  • bun run format, bun run lint, bun run typecheck, bun run test:db, and cargo fmt --all --check.
  • cargo clippy --workspace --all-targets --all-features --locked -- -D warnings.
  • Merge tests:
    • equality with a single run on dual-stack traffic, where every IPv4 and IPv6 addresses/packets/bytes row stores blob curves and 2 grid rows, plus verify and resume as a no-op;
    • refusal when a maad_q_grid q_min, q_step, or q_count differs or a row is missing, and a MAAD-disabled merge with empty grid and MAAD tables;
    • refusals for identity, overlap, out-of-range rows, CSV, schema, missing zero-fill, and mismatched configured start dates;
    • temp cleanup on a write failure, consuming success, the three recovery cases above, and a traffic-less shard not setting the start date.
  • Launcher tests with fake ssh/scp: --consume by default and off with --keep-shards, plus shard allocation for nodeA:4,nodeB over 6 and 3 days and nodeA,nodeB over 7 days.
• Real data, from before the review fixes: 3 production days sharded across 3 nodes matched a single-process run table by table (row counts and SHA-256), apart from file_device. A 2-node launcher run built, merged and verified OK, and survived a mid-run TERM and rerun.
• Remaining manual checks:
  • one launcher run on the cluster with an uneven host list, confirming every host receives a shard;
  • the first production-scale consuming merge.

---

Made by Claude Opus 5.5 via Claude Code.