# feat\(netflow-db\): add merge-shards and a cluster launcher for day-sharded reprocessing · PR \#106 · flambo/atlantis

[View on GitCafe](https://git.cafe/flambo/atlantis/pull/106)

Repository: [flambo/atlantis](https://git.cafe/flambo/atlantis)

Visibility: public

State: merged

Source repository: flambo/atlantis

Source branch: feat/cluster-shard-merge

Target branch: maad/07-compact-storage

Head commit: fd760d7d2c3ca9173f56bb1c25dc8a4ac7e84b28

[Discussion and reviews](https://git.cafe/flambo/atlantis/pull/106)

[Commits](https://git.cafe/flambo/atlantis/pull/106/commits?format=markdown)

[Files changed](https://git.cafe/flambo/atlantis/pull/106/diffs?format=markdown)

## Description

---

> [!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 (#119).** This PR is stacked on #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.