feat(storage): S3 manifest with CAS-coordinated writes and stale-refresh reads

[pepicrft]

What changed

Introduce a per-(tenant, signal) manifest.json at tenants/<tenant>/v2/logs/manifest.json as the coordination surface Pulso's architecture doc calls for. Every accepted append is registered in the manifest via a per-tenant owner process that coalesces concurrent segment registrations into a single S3 conditional PUT per flush window; every query reads the manifest from a lock-free ETS cache, prunes by time bounds, and fetches only the survivors.

Concretely:

• NIF (native/pulso_object_store/src/lib.rs): new put_if_none_match, put_if_match, get_if_none_match. put now returns {:ok, etag} so callers can prime a CAS (compare-and-swap, Wikipedia) cache without a follow-up GET. A shared stream_body_into helper keeps every GET path at one full-payload allocation streamed straight into a Rustler NewBinary, no Rust-side intermediate. Error mapping now covers :not_modified, :already_exists, :precondition_failed.
• Manifest data model (Pulso.Storage.S3.Manifest + Segment): compact JSON on the wire (1-2 character keys), an in-memory invariant of segments sorted by max_ts descending, and an O(n) merge/2 with per-batch key dedup.
• Cache (Pulso.Storage.S3.ManifestCache): public ETS table with read_concurrency: true. Query path does a single :ets.lookup/2 — no message passing, no locks under load.
• Owner (Pulso.Storage.S3.ManifestOwner): Registry-addressed GenServer that owns writes for one manifest, batches pending segments into one CAS per flush window (default 10ms) or immediately at batch cap (default 256), retries on :precondition_failed with a reload-and-rebuild, and does a first-write LIST rebuild when no manifest exists yet.
• Supervision (Pulso.Storage.S3.ManifestSupervision): Registry + DynamicSupervisor + Cache, wired into Pulso.Application only when the S3 adapter is active.
• Storage wire-up (Pulso.Storage.S3): append/3 writes the segment then CAS-registers it via the owner and returns :ok only after both durability points land. query/2 reads the manifest from the cache and, when the cached entry ages past refresh_stale_ms (default 1s), refreshes it with a conditional GET so cross-node writes become visible without a local write kicking the cache.

Why

The current storage adapter discovers segments by LIST-ing the tenant prefix on every query, which is fine for a few objects but does not scale. docs/architecture.md is explicit that S3 is the only shared coordination surface for Pulso and that manifests updated via conditional PUT are the load-bearing durability point. Landing the manifest layer first is what unlocks the follow-up work (Parquet segments, compaction, alerting, cross-node freshness gossip) without redesigning coordination each time.

Optimization goals for the initial cut were minimizing copies at the NIF boundary, maximum write throughput per tenant, a lock-free query hot path, and a compact wire form.

Approach

The tradeoffs I made explicitly:

• Coalesce writes at the owner, not at the ingester. One owner per tenant, one CAS per flush window regardless of concurrent ingesters. Under sustained load this pins S3 CAS rate to 1 / flush_interval_ms per tenant. Ingesters block in GenServer.call/3 and get their reply once the CAS lands, so the ingester layer does not have to know about the CAS itself.
• ETS cache on reads with read_concurrency: true. Every query is one lock-free lookup. Writes to the cache go only through the owner (single writer per tenant), so I do not pay for write_concurrency.
• Compact JSON with short keys on the wire. The manifest is transferred on every cold-cache load, and a fleet of thousands of segments per tenant is realistic. Short keys keep the payload small without leaving the manifest.json shape the architecture doc names.
• Stale-refresh via conditional GET. A node whose local writer is idle would otherwise never learn about segments another node commits. The refresh path issues a small conditional GET (returning 304 in the common case, with no body transfer) once a cache entry ages past refresh_stale_ms. This is the bound on cross-node freshness in the absence of gossip.
• Explicit backpressure. register_segments/5 probes the owner's mailbox length before enqueuing and returns {:error, :owner_overloaded} at the cap rather than allowing unbounded mailbox growth; the GenServer.call/3 timeout is translated into {:error, :timeout} so the typed spec holds in every failure mode.

Impact

• API compatibility: Pulso.ObjectStore.put/3 now returns {:ok, etag} instead of :ok. Only callers inside this repo are affected and all are updated in this change.
• Runtime: adds one Registry, one DynamicSupervisor, and one ETS-backed cache to the supervision tree when the S3 adapter is configured. No new processes when the memory adapter is active (unit-test path).
• Object-store state: an existing tenant with segments but no manifest gets its manifest rebuilt lazily on the first read after this lands, from a LIST of that tenant's v2/logs/ prefix. Non-.ndjson objects under the prefix are skipped at rebuild time.
• Configuration: Pulso.Storage.S3 config accepts two new optional keys — refresh_stale_ms (default 1000) for the cross-node freshness window, and max_mailbox (default 512) for the backpressure cap.

Validation

Codex adversarial review addressed three concrete findings before merge:

1. Cross-node cache staleness (high) — cached manifests were served without ever re-checking object storage, so a node whose local writer was idle could not learn about segments another node committed. Fixed with the stale-refresh conditional GET on the query path.
2. Rebuild published unloadable manifests (high) — rebuild_from_prefix used to include every object under the signal prefix. A sidecar or malformed key became a segment with nil bounds, which the next Manifest.decode/1 rejected, permanently breaking the tenant. Rebuild now skips any key that does not parse to integer time bounds.
3. Unbounded mailbox and untyped timeout (medium) — the owner mailbox was unbounded during a slow CAS or repeated :precondition_failed retries, and callers hit exit({:timeout, {GenServer, :call, _}}) instead of the specified error tuple. register_segments/5 now probes the mailbox length and returns {:error, :owner_overloaded} at the cap; the timeout exit is translated into {:error, :timeout} so the @spec holds in every failure mode.

Commands run locally:

• mix precommit — clean; 84 unit tests pass.
• PULSO_INTEGRATION=1 mix test --only integration — clean; 25 integration tests pass against the local RustFS (RustFS is the S3-compatible object store the repo bundles via docker-compose.yml).

Regression tests were added for each Codex finding: stale cache refreshes via conditional GET so cross-node writes become visible, rebuild skips non-segment objects (sidecars, junk under the prefix), and rejects appends with :owner_overloaded when the mailbox is at the cap.

🤖 Generated with Claude Code