Write ingest output to multiple buckets

[macpie]

Every file ingest writes to output_bucket can now be copied, byte for byte and under the same key, to any number of additional buckets. The motivating case is mirroring an S3 primary to Cloudflare R2, so the two buckets do not share a provider or a key pair.

output_bucket = "ingest-bucket"

[additional_output_buckets.r2]
bucket = "ingest-bucket-mirror"
endpoint = "https://<account_id>.r2.cloudflarestorage.com"
region = "auto"
access_key_id = "<r2 access key id>"
secret_access_key = "<r2 secret access key>"

[file_store]
region = "us-west-2"

Defaults to empty, so existing deployments are unaffected. FileUpload::new and from_bucket_client keep their signatures, so price, mobile-verifier and mobile-packet-verifier are untouched.

Why a map and not a list

config builds nested maps from INGEST__-prefixed environment variables but cannot build sequences. A Vec would have stranded every mirror's credentials in the settings file with no way to inject them as env vars — verified in ingest/tests/settings_env.rs, which configures a whole mirror, secret included, from the environment:

INGEST__ADDITIONAL_OUTPUT_BUCKETS__R2__SECRET_ACCESS_KEY=<secret>

(Note the separator is __ after the prefix too — INGEST_MODE is silently ignored.) The map key also names the bucket in log lines and config errors.

Why entries inherit nothing from [file_store]

Each entry is self-contained. Lending an R2 bucket the S3 account's region or key pair only builds a client that authenticates against neither.

For the same reason ingest now refuses to start on an entry carrying only one half of a credential pair: the S3 client installs static credentials only when it has both, and otherwise silently falls back to the ambient AWS credential chain — which cannot succeed against R2. A typo in one of the two key names would have passed startup and failed every upload at runtime.

No changes were needed in new_client for R2: it already applies force_path_style when an endpoint is set, and already forces request_checksum_calculation(WhenRequired), which is what keeps R2 from choking on aws-chunked trailing checksums. Clients are cached on the full credential tuple, so the two stay distinct.

What happens when one bucket fails

Buckets are attempted concurrently, each with its own retry loop (~60s). A local file is deleted only once every bucket has it, so a failing bucket never costs the others their copy. A file left behind is retried on the next restart, when FileSink::init rescans the cache directory.

Nothing retries it in-process, so a mirror that stays broken accumulates files on the ingest host's disk. That is the deliberate tradeoff — never lose a copy — but worth knowing before enabling a mirror. A periodic cache rescan would make it self-healing; happy to add that if reviewers want it now.

Observability

Adds file_store_upload{bucket,status} — one increment per file per bucket at its terminal outcome, so status="error" counts files that bucket actually dropped:

rate(file_store_upload{status="error"}[5m]) > 0

Both series are seeded to zero per bucket at startup so an alert can tell "no failures" from "not reporting at all". Without the metric, a bucket that quietly stops accepting files is invisible: the uploader keeps the local copy, logs an error, and the service goes on looking healthy.

Two fixes that fell out of testing

A roll_time shorter than 60s was silently ignored. Time-based rolling happens only on a rollover timer whose period was a flat SINK_CHECK_MILLIS = 60_000; the write path checks size alone (will_fit). So any configured roll time under a minute was rounded up to one — which is why file_store carried a #[cfg(test)] 50ms override, and why the new ingest integration test initially hung. The interval is now min(roll_time, 60s), floored at 1ms since interval panics on zero.

Production behaviour is unchanged: every deployment configures ≥ 60s (3 min default, 15 for ingest), so min always picks 60s. The cfg(test) override is gone and file_store's own sink tests now pass against the real constant. mobile_packet_verifier's integration tests set roll_time(100ms) with a comment saying it "ensures the file is committed promptly" — that only became true with this change; all 78 still pass.

AwsLocal::delete_bucket could not delete an empty bucket. It always sent DeleteObjects, which S3 rejects with no keys (MalformedXML, HTTP 400), so a bucket created but never written to could not be torn down at all.

Tests

All end to end against the RustFS in docker-compose.yml, on buckets AwsLocal creates and deletes per run — no init.sh change, no cross-run collisions.

• ingest/tests/multi_output_bucket.rs drives the real grpc_server from a real settings file through the real sinks and uploader: gRPC submission → rolling sink → both buckets. Asserts the same key, byte-identical content, and that those bytes decode back to the submitted report. Unlike mobile_ingest.rs, nothing is stubbed.
• file_store/tests/file_upload.rs covers the fan-out, the keep-on-partial-failure behaviour, a mirror under an independent credential pair, and the metric's label shape.
• ingest/src/settings.rs unit tests cover the config shape, the credential-pair guard, and that a settings dump omits mirror credentials — main.rs logs the whole struct as JSON at startup, and the skip_serializing on the credential pair now sits behind a serde(flatten).

Each assertion was mutation-checked (dropping the mirrors, removing the metric increment, removing the seeding) to confirm it fails when the behaviour breaks.

CI

Verified against the pinned 1.94 toolchain, not the newer one on my machine — cargo fmt --all --check and cargo clippy --all-targets --locked -- -Dclippy::all -D warnings are both clean workspace-wide. Full suites pass for file-store, ingest, mobile-packet-verifier, mobile-verifier, price and file-store-oracles.

One unrelated flake to be aware of: file_info_poller::poller_without_idle_timeout_does_not_exit has a 100ms timeout on an S3 round-trip and fails intermittently on a loaded worker. It predates this branch.

metrics-util is added as a dev-dependency only (already in the lockfile transitively); enabling its debugging feature pulls a few new crates into Cargo.lock, so the first CI run fetches them rather than getting them pre-warmed from the base image.