From 7fe3d429f0163c52da20e45e9719b2b8b8efbd20 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 20 Jul 2026 11:40:05 -0500 Subject: [PATCH] close compaction parity --- README.md | 4 +- docs/grafana-dashboard.md | 22 +- docs/semantic-parity.md | 2 +- docs/upstream-compaction-spec.md | 72 +++---- justfile | 1 + src/internal/archive.zig | 26 ++- src/internal/bootstrap/lifecycle.zig | 19 +- src/internal/cold.zig | 1 + src/internal/compact/pass.zig | 290 ++++++++++++++++++++++++--- src/internal/compact/steady.zig | 198 +++++++++++++++--- src/internal/metrics.zig | 98 ++++++++- src/internal/segment_rewrite.zig | 59 +++++- src/internal/tombstone.zig | 130 +++++++++--- src/main.zig | 1 + 14 files changed, 785 insertions(+), 138 deletions(-) diff --git a/README.md b/README.md index 1493830..0ef9b06 100644 --- a/README.md +++ b/README.md @@ -136,5 +136,5 @@ fail-closed metric contract. The dashboard and process receipts are fully offline: run `just dashboard-test`, `just process-metrics-contract`, `just http-metrics-contract`, then `just dashboard-contract http://127.0.0.1:6008/metrics` against a local -candidate. The final command intentionally remains red until every upstream -lifecycle family has a real producer. +candidate. `just process-metrics-contract` also runs that complete dashboard +family check against the ReleaseSafe binary it launches. diff --git a/docs/grafana-dashboard.md b/docs/grafana-dashboard.md index 1ac69b4..9694589 100644 --- a/docs/grafana-dashboard.md +++ b/docs/grafana-dashboard.md @@ -165,6 +165,20 @@ The following dashboard families already have non-placeholder producers: - `jetstream_import_segments_examined_total` - `jetstream_import_segments_patched_total` - `jetstream_import_bytes_rewritten_total` +- `jetstream_compaction_passes_total` +- `jetstream_compaction_pass_duration_seconds` +- `jetstream_compaction_passes_early_total` +- `jetstream_compaction_tombstones_collected_total` +- `jetstream_compaction_tombstone_set_entries` +- `jetstream_compaction_tombstone_set_bytes` +- `jetstream_compaction_segments_examined_total` +- `jetstream_compaction_segments_rewritten_total` +- `jetstream_compaction_segments_skipped_clean_total` +- `jetstream_compaction_manifest_reconciled_total` +- `jetstream_compaction_rows_dropped_total` +- `jetstream_compaction_bytes_rewritten_total` +- `jetstream_compaction_watermark_seq` +- `jetstream_compaction_watermark_lag_seconds` The backfill families are measured at Stream's actual page checkpoint: worker results enter the completion queue after download/handling, archive bytes are @@ -174,6 +188,8 @@ upstream's writer-hook completion batcher without manufacturing a second queue. Offline loopback receipts cover a terminal RepoNotFound, a generated complete CAR, redirected rate limiting, and a restart-safe parked-host skip. -The checksum-pinned deployment contract remains red until the remaining -compaction families are backed by their real lifecycle boundaries. The current -binary is not a dashboard-complete release. +Compaction metrics are emitted at the actual fold, worker rewrite, manifest +refresh, watermark commit, live-set, and timer boundaries. The production +offline receipt now reports `required=104 present=104 missing=0`; presence is +paired with real-JSS unit receipts so registered zero series cannot substitute +for an implementation. diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 24a85b5..2b19894 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -20,7 +20,7 @@ this document or `bootstrap-semantic-parity.md` is open. | Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. The writer emits `sync` then bounded 1,024-row `create_resync` batches and stages the fetched/pending chain checkpoints under the archive lock before any matching fsync. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. | | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | **closed at the implementation boundary.** Durable v5 repo rows distinguish initial Backfill.Rev from latest Rev/UpdatedAt, retain resolved Handle/PDS and reserved record/byte fields, and atomically maintain `handle/`. Merge source cursor commits refresh latest revisions without mutating backfill watermarks. Same-batch `host/` totals, active/status counts, cumulative error classes, and five bounded recent samples survive restart; host moves and active flips are atomic with the source repo row. `/status?tab=hosts` reads those rows without a whole-network scan and matches upstream ordering and error presentation. Account queries preserve resolver-first handle semantics, the legacy `did`/`handle` aliases, and missing-identity hydration, then reconstruct records last-writer-wins from checksum-verified, DID-bloom-pruned sealed JSS plus rotation-safe active/pending and bootstrap-live snapshots. The resulting canonical MST is compared with the PDS commit root through real Sync 1.1 `getLatestCommit`/`getBlocks`; the page renders the same match fields and error state as upstream. `HEAD` performs lookup without verification; successful status responses are `no-store` and carry a generation timestamp. Expensive verification uses upstream's exact 4/source-IP/minute fixed-window limiter, 4,096-entry bound, stale pruning, oldest eviction, and explicit disable option. The offline receipt uses a real HTTP Stream route, real RocksDB row, real JSS archive, signed commit, and loopback PLC/PDS; four separate connections verify successfully, a fifth changing only its ephemeral port is blocked without another upstream request, HEAD remains side-effect-free, and disabling the limiter performs a new authoritative verification. | -| Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract; canonical build/live-firehose/archive-append/drop/verifier/backfill/timestamp-import producers are present. The complete canonical subscriber family set is exported from real scan, delivery, rejection, cursor, and disconnect boundaries; `none`, `deflate`, and `zstd` labels now arise only from the actual negotiated delivery path. Live frame inspection distinguishes gaps, bounded-label server errors, forward-compatible unknowns, and malformed CBOR at the raw transport boundary; a Zat reconnect callback measures attempts rather than connection failures. The only dashboard divergence is now a deterministic, checksum-pinned process-runtime row: same-semantics start/CPU/RSS/FD measurements, real VM/thread/page-fault/context-switch signals, no `go_*` aliases, and no host-network counters mislabeled as process I/O. A production-binary loopback receipt proves values and cumulative monotonicity on independent scrapes. The hot readable log now matches upstream's durability pin: rows above the archive watermark may exceed the byte budget but cannot be evicted, publication releases their pinned-byte accounting, and only then may retention advance the floor; all three canonical readable-log gauges come from that exact state. Data-directory free bytes are collected fresh from the archive directory's filesystem at scrape time, and the canonical seal histogram measures only successful flush-through-footer/header-fsync work with upstream's exact 15 slow-latency buckets. HTTP duration uses upstream's public-mux boundary and stable labels, including inline WebSocket lifetime and the single `xrpc/` subtree; debug and unmatched routes remain unobserved. getBlock outcomes, duration, and served bytes come from actual response completion, with bytes counted only for complete 200 responses. A ReleaseSafe offline receipt proves 200/206/304/400/404/416 behavior and WebSocket close accounting. Sealed headers and block indexes are now genuinely resident: startup and verified refresh latency, loaded count, and lookup hits/misses use upstream's exact boundaries. Refcounted snapshots remain memory-safe across refresh, while fresh-file offsets prevent cross-generation splicing; the pinned official client proves cold replay produces hits. RocksDB point reads, durable point writes/deletes, and atomic batch commits populate upstream's exact `{op,status}` store histogram and fast-latency buckets; real-database and production-startup receipts prove the non-error paths. Durable lifecycle phase/transition metrics, all six cutover-state timings, and merge counters are wired to the actual bootstrap/merge state machine. Merge now performs one cached Backfill.Rev lookup per DID, including misses; real JSS/RocksDB and eight-crash production receipts prove counter and restart behavior. Backfill discovery, active flips, handler timing, completion queue/durability, current-run progress, forced checkpoints, failure persistence, and steady retry scans now use Stream's real worker/page/RocksDB boundaries. Generated-CAR, RepoNotFound, redirected-429, durable-host-park, and production crash receipts prevent zero-series or API-shape substitutions. **Only the compaction dashboard families remain open.** | +| Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | **closed.** The exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract. Canonical build, live-firehose, archive, verifier, subscribe, backfill, timestamp-import, and compaction families all have same-semantics producers. Compaction measures pass success/failure/duration, early triggers, live tombstone size, folded reasons, worker examination/rewrite/clean outcomes, dropped-row reasons, bytes, manifest heals, durable watermark, and witnessed-time lag at the corresponding real boundaries. Real JSS tests cover bloom avoidance, bounded concurrent rewrites, malformed account failure, reason accounting, manifest-before-watermark ordering, and durable eviction. The ReleaseSafe production receipt reports `required=104 present=104 missing=0`; the runtime-only dashboard row remains an intentional checksum-pinned Zig/process adaptation with no synthetic `go_*` aliases. See `grafana-dashboard.md`. | | Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **open — current crash matrix is necessary but substantially smaller than upstream's oracle** | ## Experiment admission diff --git a/docs/upstream-compaction-spec.md b/docs/upstream-compaction-spec.md index 6c0b6ca..d44b6d6 100644 --- a/docs/upstream-compaction-spec.md +++ b/docs/upstream-compaction-spec.md @@ -108,8 +108,8 @@ boundary-seq rules: crashpoints straddle both sides (`AfterCompactionRewriteBeforeWatermark`, `AfterCompactionChunkWatermark`). -failure policy: a failed steady pass logs and retries next tick — it must -NEVER tear down the daemon (would turn a transient IO error into an +failure policy: a failed steady pass logs and retries at the next scheduled +pass — it must NEVER tear down the daemon (would turn a transient IO error into an ingestion outage). `watermark_lag_seconds` (tip witnessed-at minus watermark witnessed-at, floored to the span of uncompacted data so an empty watermark doesn't read as a 50-year spike) is the operator paging signal. @@ -202,43 +202,31 @@ observes lag and continues. tombstones collected by reason, manifest reconciles, early passes, tombstone set bytes/len. -## port notes - -maps naturally onto stream: -- `src/internal/manifest.zig` now owns sorted sealed headers and refcounted - block indexes. Seal publishes a new entry; compaction and timestamp import - verify header/footer checksum and atomically replace it before their durable - lifecycle boundary advances. Blooms and collection indexes remain on-disk - planner work and are not mislabeled as resident yet. -- jss sealed segments already carry per-block indexes and blooms, so the - window fold (block seq-bound pruning) and candidate-DID bloom prefilter - translate directly; keep the 100k-DID narrowing cutoff and the - "block bounds are historical supersets" invariant when rewriting. -- rewrite = tmp + fsync + rename + fsync dir is already house style; keep - block topology (empty blocks stay) so block numbering survives rewrites. -- watermark: no pebble — a versioned uint64 file (write-tmp-rename) beside - the segments dir matches our file-based metadata precedent; the floor - init (`nextSeq - 1`) is trivial. -- in-memory tombstone set is a plain hashmap pair under a mutex; the fold/ - merge/evict/rebuild functions are small and pure — good first unit-test - surface. clone keys out of decoded block buffers, same aliasing hazard. - -needs a design decision: -- cold-path decoded-block cache invalidation: if/when stream grows a - decoded-block LRU (`cold.zig` / `lru.zig`), decide between generation - counters in the cache key vs keying purely on segment checksum; upstream - needs both (generation closes the in-flight race, checksum covers missed - invalidations). single-threaded readers may let us skip generations. -- rewrite lock: upstream needs it because timestamp import shares the - rewrite path; stream has no second rewriter yet — decide whether to - introduce the mutex now (cheap) or document single-compactor as an - invariant. -- who owns force-rotate: upstream threads the live writer into the pass; - stream's ingest/segment_writer split needs an explicit seam so the - compactor can seal the active segment and rely on the "tombstone observed - before rotate returns" ordering. -- account-delete detection decodes the #account CBOR payload per row during - fold; decide whether to precompute a flag at ingest (a row-kind or header - bit) instead of re-decoding CBOR on every rebuild/pass. -- chunking cap and trigger cap defaults (32M entries) assume go map sizes; - re-derive for zig hashmap overhead before copying the constant. +## Stream implementation receipt + +Stream implements this contract directly over its JSS archive and shared +RocksDB metadata store: + +- account CBOR is decoded during observation/fold and malformed markers fail + the append or pass; tombstone keys are cloned, strict drop reasons survive + through physical rewrite, and the live set publishes upstream-equivalent + entry and approximate-byte gauges; +- every pass folds its bounded window from sealed JSS blocks, narrows real + segment reads with the existing Gloom DID bloom up to the 100,000-DID + cutoff, and fans all sealed segments across `min(CPU, 8)` workers; +- delete and timestamp rewrites share the archive rewrite mutex. Cleanup is + locked independently and mutation is locked per chunk, allowing safe import + interleaving only between chunks; +- rewritten files preserve block topology and historical envelopes, then use + tmp write/fsync/rename/directory-fsync. Manifest refresh completes before + the RocksDB watermark commits; steady passes heal checksum mismatches and + merge-tail passes perform the required one-shot reconcile before serving; +- the archive hook is fallible and runs after sequence assignment but before + any flush or seal. Force rotation therefore cannot expose a sealed event + whose tombstone observation has not succeeded; +- a zero compaction interval disables both merge-tail and steady compaction; + the merge destination is still sealed because serving requires sealed JSS; +- all canonical compaction counters, gauges, reason labels, and the exact + 16-bucket pass histogram are populated at these real boundaries. Offline + tests exercise real segment files, concurrent workers, bloom avoidance, + malformed input, manifest refresh, and durable watermark advancement. diff --git a/justfile b/justfile index b44c72e..c57f05d 100644 --- a/justfile +++ b/justfile @@ -44,6 +44,7 @@ process-metrics-contract: sleep 0.1 done STREAM_BASE_URL="http://127.0.0.1:$PORT" python3 tests/process_metrics_contract.py + python3 tools/check_dashboard.py --metrics "http://127.0.0.1:$PORT/metrics" # Seed a real sealed archive and prove the upstream HTTP middleware and # getBlock metric families against actual HTTP and WebSocket connections. diff --git a/src/internal/archive.zig b/src/internal/archive.zig index 2c54e2d..0cac7be 100644 --- a/src/internal/archive.zig +++ b/src/internal/archive.zig @@ -61,7 +61,7 @@ pub const Archive = struct { /// so a force-rotate returns only after every sealed row was observed /// (the live-tombstone ordering guarantee) on_append_ctx: ?*anyopaque = null, - on_append: ?*const fn (*anyopaque, segment.Event) void = null, + on_append: ?*const fn (*anyopaque, segment.Event) anyerror!void = null, /// Upstream's timestamp Stamper: called under `mu`, before sequence /// assignment and buffering. It mutates the caller's row so archive bytes /// and immediate subscriber delivery use the identical display timestamp. @@ -257,7 +257,7 @@ pub const Archive = struct { }; self.next_seq += 1; if (self.stats) |stats| _ = stats.ingest_events_appended_total.fetchAdd(1, .monotonic); - if (self.on_append) |hook| hook(self.on_append_ctx.?, event.*); + if (self.on_append) |hook| try hook(self.on_append_ctx.?, event.*); return seq; } @@ -267,6 +267,28 @@ pub const Archive = struct { return self.next_seq - 1; } + /// Heal resident metadata after an atomic segment rewrite. Returns the + /// number of missing or checksum-mismatched entries refreshed from disk. + pub fn reconcileManifest(self: *Archive) !u64 { + var reconciled: u64 = 0; + var it = self.dir.iterate(); + while (try it.next(self.io)) |entry| { + const idx = parseSegmentIndex(entry.name) orelse continue; + var file = try self.dir.openFile(self.io, entry.name, .{}); + defer file.close(self.io); + const header = segment.readHeaderFile(self.io, &file) catch |err| switch (err) { + error.ActiveSegment => continue, + else => return err, + }; + if (self.manifest.segmentByIndex(idx)) |resident| { + if (resident.header.checksum == header.checksum) continue; + } + try self.manifest.refresh(self.dir, idx, true); + reconciled += 1; + } + return reconciled; + } + pub fn lockRewrites(self: *Archive) void { self.rewrite_mu.lockUncancelable(self.io); } diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index a67ad4d..66bdd6c 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -54,6 +54,9 @@ pub const Options = struct { max_repos: usize = 0, workers: usize = engine.default_workers, max_inflight_bytes: usize = engine.default_max_inflight_bytes, + /// Matches upstream: a zero interval disables both merge-tail and + /// steady delete compaction. The destination is still sealed to serve. + compaction_enabled: bool = true, }; /// drive the lifecycle to steady_state. no-op if already there. the @@ -144,16 +147,28 @@ pub fn run( // nothing is serving yet. idempotent, so the restart-after-cleanup // path re-runs it too. try dst.rotate(); - { + if (opts.compaction_enabled) { + const compaction_started = Io.Timestamp.now(io, .awake); var seg_path_buf: [256]u8 = undefined; const seg_path = try std.fmt.bufPrint(&seg_path_buf, "{s}/segments", .{opts.data_dir}); var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io); - const cs = try compact_pass.run(allocator, io, seg_dir, &wm, null, .{}); + const cs = compact_pass.run(allocator, io, seg_dir, &wm, null, .{ + .instrumentation = stats, + .measure_pass = false, + }) catch |err| { + stats.observeCompactionPass(.error_result, elapsedUs(io, compaction_started)); + return err; + }; + stats.observeCompactionPass(.ok, elapsedUs(io, compaction_started)); log.info("merge-tail compaction: watermark {d}, {d} chunks, {d}/{d} segments rewritten, {d} rows dropped", .{ cs.target_watermark, cs.chunks, cs.segments_rewritten, cs.segments_examined, cs.rows_dropped, }); } + // Merge-tail rewriting is deliberately manifest-oblivious. Before + // serving can ungate, every resident header/index must match disk. + const reconciled = try dst.reconcileManifest(); + _ = stats.compaction_manifest_reconciled_total.fetchAdd(reconciled, .monotonic); crashpoint.hit(.after_seal_before_discovery); if (have_live) { diff --git a/src/internal/cold.zig b/src/internal/cold.zig index e88f708..fb8ff07 100644 --- a/src/internal/cold.zig +++ b/src/internal/cold.zig @@ -583,6 +583,7 @@ test "cursor combines resident envelopes with fresh generation offsets" { io, archive.dir, "seg_0000000000.jss", + .{}, {}, Keep.row, ); diff --git a/src/internal/compact/pass.zig b/src/internal/compact/pass.zig index ce437ea..a940852 100644 --- a/src/internal/compact/pass.zig +++ b/src/internal/compact/pass.zig @@ -13,14 +13,13 @@ //! overlapping segment the chunk watermark commits, then the live set //! evicts. a chunk cap bounds snapshot memory. //! -//! not yet ported: the candidate-DID bloom prefilter (we exact-scan -//! every overlapping segment — what upstream does anyway when the -//! snapshot exceeds its 100k-DID narrowing cutoff, as merge-tail passes -//! always do) and the min(NumCPU,8) rewrite worker fan-out (serial). +//! candidate-DID segment blooms avoid unrelated decodes while the set is +//! reasonably small; rewrites fan out across min(CPU, 8) workers. const std = @import("std"); const archive_mod = @import("../archive.zig"); const meta_store = @import("../meta_store.zig"); +const metrics = @import("../metrics.zig"); const rewrite_mod = @import("../segment_rewrite.zig"); const segment = @import("../segment.zig"); const tombstone = @import("../tombstone.zig"); @@ -39,8 +38,17 @@ pub const Options = struct { /// snapshot entry cap per chunk; reaching it closes the chunk at the /// current segment's max_seq tombstone_cap: usize = default_tombstone_cap, + bloom_narrow_max_dids: usize = 100_000, + rewrite_workers: usize = 0, + instrumentation: ?*metrics.Stats = null, + measure_pass: bool = true, on_rewritten_ctx: ?*anyopaque = null, on_rewritten: ?*const fn (?*anyopaque, u64) anyerror!void = null, + rewrite_lock_ctx: ?*anyopaque = null, + rewrite_lock: ?*const fn (?*anyopaque) void = null, + rewrite_unlock: ?*const fn (?*anyopaque) void = null, + on_committed_ctx: ?*anyopaque = null, + on_committed: ?*const fn (?*anyopaque, u64) void = null, }; pub const Stats = struct { @@ -49,6 +57,9 @@ pub const Stats = struct { segments_examined: u64 = 0, segments_rewritten: u64 = 0, rows_dropped: u64 = 0, + rows_dropped_by_reason: tombstone.Counts = .{}, + segments_skipped_clean: u64 = 0, + bytes_rewritten: u64 = 0, tombstones_folded: u64 = 0, }; @@ -56,6 +67,9 @@ const SegMeta = struct { idx: u64, min_seq: u64, max_seq: u64, + event_count: u32, + min_witnessed_at: i64, + max_witnessed_at: i64, }; /// run one pass. `seg_dir` is /segments (iterable). `live` is the @@ -70,7 +84,32 @@ pub fn run( live: ?*tombstone.Snapshot, opts: Options, ) !Stats { - try rewrite_mod.cleanStaleTmp(allocator, io, seg_dir); + const started = Io.Timestamp.now(io, .awake); + const result = runInner(allocator, io, seg_dir, wm, live, opts) catch |err| { + if (opts.measure_pass) { + if (opts.instrumentation) |s| s.observeCompactionPass(.error_result, elapsedUs(io, started)); + } + return err; + }; + if (opts.measure_pass) { + if (opts.instrumentation) |s| s.observeCompactionPass(.ok, elapsedUs(io, started)); + } + return result; +} + +fn runInner( + allocator: Allocator, + io: Io, + seg_dir: Io.Dir, + wm: *watermark.Store, + live: ?*tombstone.Snapshot, + opts: Options, +) !Stats { + { + if (opts.rewrite_lock) |lock| lock(opts.rewrite_lock_ctx); + defer if (opts.rewrite_unlock) |unlock| unlock(opts.rewrite_lock_ctx); + try rewrite_mod.cleanStaleTmp(allocator, io, seg_dir); + } var metas: std.ArrayList(SegMeta) = .empty; defer metas.deinit(allocator); @@ -78,11 +117,20 @@ pub fn run( var stats: Stats = .{}; var target: u64 = 0; - for (metas.items) |m| target = @max(target, m.max_seq); + for (metas.items) |m| if (m.event_count > 0) { + target = @max(target, m.max_seq); + }; stats.target_watermark = target; var current = wm.load(io) orelse 0; - if (metas.items.len == 0 or target <= current) return stats; + if (opts.instrumentation) |s| { + s.compaction_watermark.store(current, .monotonic); + s.compaction_watermark_lag_us.store(watermarkLagUs(metas.items, current), .monotonic); + } + if (metas.items.len == 0 or target <= current) { + if (opts.instrumentation) |s| s.compaction_watermark_lag_us.store(0, .monotonic); + return stats; + } while (current < target) { // fold this chunk's tombstones from disk in (current, chunk_end] @@ -97,27 +145,81 @@ pub fn run( break; } } + const collected = snap.countsByReason(); stats.tombstones_folded += snap.len(); - const ctx: Decide = .{ .chunk_end = chunk_end, .snap = &snap }; - for (metas.items) |m| { - if (m.min_seq > chunk_end) continue; - stats.segments_examined += 1; - var name_buf: [64]u8 = undefined; - const name = archive_mod.formatSegmentName(&name_buf, m.idx); - const r = try rewrite_mod.rewriteFile(allocator, io, seg_dir, name, ctx, decide); - if (r.rewritten) { - if (opts.on_rewritten) |refresh| try refresh(opts.on_rewritten_ctx, m.idx); - stats.segments_rewritten += 1; - stats.rows_dropped += r.dropped; - log.info("compaction: rewrote {s} ({d} rows dropped)", .{ name, r.dropped }); + if (snap.len() > 0) { + // Match upstream's lock scope: one mutating chunk. Timestamp + // import may safely interleave between chunks, never within one. + if (opts.rewrite_lock) |lock| lock(opts.rewrite_lock_ctx); + defer if (opts.rewrite_unlock) |unlock| unlock(opts.rewrite_lock_ctx); + const candidates = try snap.candidateDids(allocator, opts.bloom_narrow_max_dids); + defer if (candidates) |items| allocator.free(items); + const outcomes = try allocator.alloc(RewriteOutcome, metas.items.len); + defer allocator.free(outcomes); + @memset(outcomes, .{}); + var work: RewriteWork = .{ + .allocator = allocator, + .io = io, + .seg_dir = seg_dir, + .metas = metas.items, + .snap = &snap, + .chunk_end = chunk_end, + .candidate_dids = candidates, + .outcomes = outcomes, + .instrumentation = opts.instrumentation, + }; + var group: Io.Group = .init; + const cpu = std.Thread.getCpuCount() catch 1; + const configured = if (opts.rewrite_workers == 0) @min(cpu, 8) else opts.rewrite_workers; + const worker_count = @min(@max(configured, 1), metas.items.len); + for (0..worker_count) |_| try group.concurrent(io, RewriteWork.runWorker, .{&work}); + group.await(io) catch |err| return err; + if (work.fatal_error) |err| return err; + + for (metas.items, outcomes) |m, outcome| { + stats.segments_examined += 1; + const r = outcome.result; + if (r.rewritten) { + if (opts.on_rewritten) |refresh| try refresh(opts.on_rewritten_ctx, m.idx); + stats.segments_rewritten += 1; + stats.rows_dropped += r.dropped; + stats.rows_dropped_by_reason.record += outcome.drops.record; + stats.rows_dropped_by_reason.sync += outcome.drops.sync; + stats.rows_dropped_by_reason.account += outcome.drops.account; + stats.bytes_rewritten += r.bytes_rewritten; + if (opts.instrumentation) |s| { + _ = s.compaction_segments_rewritten_total.fetchAdd(1, .monotonic); + _ = s.compaction_rows_dropped.getPtr(.record).fetchAdd(outcome.drops.record, .monotonic); + _ = s.compaction_rows_dropped.getPtr(.sync).fetchAdd(outcome.drops.sync, .monotonic); + _ = s.compaction_rows_dropped.getPtr(.account).fetchAdd(outcome.drops.account, .monotonic); + _ = s.compaction_bytes_rewritten_total.fetchAdd(r.bytes_rewritten, .monotonic); + } + log.info("compaction: rewrote segment {d} ({d} rows dropped)", .{ m.idx, r.dropped }); + } else { + stats.segments_skipped_clean += 1; + if (opts.instrumentation) |s| _ = s.compaction_segments_skipped_clean_total.fetchAdd(1, .monotonic); + } } } + if (opts.instrumentation) |s| { + _ = s.compaction_tombstones_collected.getPtr(.record).fetchAdd(collected.record, .monotonic); + _ = s.compaction_tombstones_collected.getPtr(.sync).fetchAdd(collected.sync, .monotonic); + _ = s.compaction_tombstones_collected.getPtr(.account).fetchAdd(collected.account, .monotonic); + } + // durable commit, then eviction (rebuild-from-disk makes a crash // between the two immaterial) try wm.save(io, chunk_end); - if (live) |l| l.evict(chunk_end); + if (opts.instrumentation) |s| { + s.compaction_watermark.store(chunk_end, .monotonic); + s.compaction_watermark_lag_us.store(watermarkLagUs(metas.items, chunk_end), .monotonic); + } + if (opts.on_committed) |committed| + committed(opts.on_committed_ctx, chunk_end) + else if (live) |l| + l.evict(chunk_end); stats.chunks += 1; current = chunk_end; } @@ -140,8 +242,14 @@ fn sweep(allocator: Allocator, io: Io, seg_dir: Io.Dir, metas: *std.ArrayList(Se error.ActiveSegment => continue, else => return err, }; - if (header.event_count == 0) continue; - try metas.append(allocator, .{ .idx = idx, .min_seq = header.min_seq, .max_seq = header.max_seq }); + try metas.append(allocator, .{ + .idx = idx, + .min_seq = header.min_seq, + .max_seq = header.max_seq, + .event_count = header.event_count, + .min_witnessed_at = header.min_witnessed_at, + .max_witnessed_at = header.max_witnessed_at, + }); } std.mem.sort(SegMeta, metas.items, {}, struct { fn lt(_: void, a: SegMeta, b: SegMeta) bool { @@ -181,13 +289,90 @@ pub fn foldSegment( const Decide = struct { chunk_end: u64, snap: *const tombstone.Snapshot, + drops: *tombstone.Counts, }; /// keep rows above the chunk (a later chunk's fold covers them), drop /// covered materialization rows below it -fn decide(ctx: Decide, ev: segment.Event) bool { +fn decide(ctx: *Decide, ev: segment.Event) bool { if (ev.seq > ctx.chunk_end) return true; - return !ctx.snap.shouldDrop(ev); + const reason = ctx.snap.dropReason(ev) orelse return true; + switch (reason) { + .record => ctx.drops.record += 1, + .sync => ctx.drops.sync += 1, + .account => ctx.drops.account += 1, + } + return false; +} + +const RewriteOutcome = struct { + result: rewrite_mod.Result = .{}, + drops: tombstone.Counts = .{}, +}; + +const RewriteWork = struct { + allocator: Allocator, + io: Io, + seg_dir: Io.Dir, + metas: []const SegMeta, + snap: *const tombstone.Snapshot, + chunk_end: u64, + candidate_dids: ?[]const []const u8, + outcomes: []RewriteOutcome, + instrumentation: ?*metrics.Stats, + next: std.atomic.Value(usize) = .init(0), + fatal_mu: Io.Mutex = .init, + fatal_error: ?anyerror = null, + failed: std.atomic.Value(bool) = .init(false), + + fn runWorker(self: *RewriteWork) void { + while (true) { + if (self.failed.load(.acquire)) return; + const index = self.next.fetchAdd(1, .monotonic); + if (index >= self.metas.len) return; + self.process(index) catch |err| { + self.fatal_mu.lockUncancelable(self.io); + if (self.fatal_error == null) self.fatal_error = err; + self.fatal_mu.unlock(self.io); + self.failed.store(true, .release); + return; + }; + } + } + + fn process(self: *RewriteWork, index: usize) !void { + if (self.instrumentation) |s| _ = s.compaction_segments_examined_total.fetchAdd(1, .monotonic); + const m = self.metas[index]; + var name_buf: [64]u8 = undefined; + const name = archive_mod.formatSegmentName(&name_buf, m.idx); + var outcome: RewriteOutcome = .{}; + var ctx: Decide = .{ .chunk_end = self.chunk_end, .snap = self.snap, .drops = &outcome.drops }; + outcome.result = try rewrite_mod.rewriteFile(self.allocator, self.io, self.seg_dir, name, .{ + .candidate_dids = self.candidate_dids, + }, &ctx, decide); + self.outcomes[index] = outcome; + } +}; + +fn watermarkLagUs(metas: []const SegMeta, watermark_seq: u64) u64 { + if (metas.len == 0) return 0; + var tip_witness: i64 = 0; + var oldest_witness: i64 = std.math.maxInt(i64); + var watermark_witness: ?i64 = null; + for (metas) |m| { + if (m.event_count == 0) continue; + tip_witness = @max(tip_witness, m.max_witnessed_at); + oldest_witness = @min(oldest_witness, m.min_witnessed_at); + if (m.max_seq <= watermark_seq) + watermark_witness = @max(watermark_witness orelse m.max_witnessed_at, m.max_witnessed_at); + } + const base = watermark_witness orelse oldest_witness; + if (tip_witness <= base) return 0; + return @intCast(tip_witness - base); +} + +fn elapsedUs(io: Io, started: Io.Timestamp) u64 { + return @intCast(@max(0, started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds())); } // === tests === @@ -248,13 +433,27 @@ test "pass: drops superseded rows, keeps markers, commits watermark, idempotent" var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io); - const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{}); + var instrumentation: metrics.Stats = .{}; + const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{ .instrumentation = &instrumentation }); try testing.expectEqual(@as(u64, 6), stats.target_watermark); try testing.expectEqual(@as(u64, 1), stats.chunks); try testing.expectEqual(@as(u64, 1), stats.segments_rewritten); // dropped: create a/k1 (seq 1 < delete seq 4), create b/k1 (seq 3 < account seq 5) try testing.expectEqual(@as(u64, 2), stats.rows_dropped); + try testing.expectEqual(@as(u64, 1), stats.rows_dropped_by_reason.record); + try testing.expectEqual(@as(u64, 1), stats.rows_dropped_by_reason.account); + try testing.expect(stats.bytes_rewritten > 0); try testing.expectEqual(@as(?u64, 6), wm.load(io)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_tombstones_collected.getPtrConst(.record).load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_tombstones_collected.getPtrConst(.account).load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_rows_dropped.getPtrConst(.record).load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_rows_dropped.getPtrConst(.account).load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_segments_examined_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_segments_rewritten_total.load(.monotonic)); + try testing.expect(instrumentation.compaction_bytes_rewritten_total.load(.monotonic) > 0); + try testing.expectEqual(@as(u64, 6), instrumentation.compaction_watermark.load(.monotonic)); + try testing.expectEqual(@as(u64, 0), instrumentation.compaction_watermark_lag_us.load(.monotonic)); // surviving rows: creates k2/k3 + both markers; block topology intact const bytes = try seg_dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20)); @@ -273,9 +472,10 @@ test "pass: drops superseded rows, keeps markers, commits watermark, idempotent" try testing.expectEqualSlices(u64, &.{ 2, 4, 5, 6 }, seqs.items); // second pass: nothing above the watermark → no-op - const again = try run(testing.allocator, io, seg_dir, &wm, null, .{}); + const again = try run(testing.allocator, io, seg_dir, &wm, null, .{ .instrumentation = &instrumentation }); try testing.expectEqual(@as(u64, 0), again.chunks); try testing.expectEqual(@as(u64, 0), again.segments_rewritten); + try testing.expectEqual(@as(u64, 2), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic)); } test "pass: tombstone cap splits the pass into chunks across segments" { @@ -311,9 +511,45 @@ test "pass: tombstone cap splits the pass into chunks across segments" { var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); defer seg_dir.close(io); - const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{ .tombstone_cap = 1 }); + const stats = try run(testing.allocator, io, seg_dir, &wm, null, .{ .tombstone_cap = 1, .rewrite_workers = 2 }); try testing.expectEqual(@as(u64, 2), stats.chunks); try testing.expectEqual(@as(u64, 2), stats.segments_rewritten); try testing.expectEqual(@as(u64, 2), stats.rows_dropped); + try testing.expectEqual(@as(u64, 4), stats.segments_examined); + try testing.expectEqual(@as(u64, 2), stats.segments_skipped_clean); try testing.expectEqual(@as(?u64, 4), wm.load(io)); } + +test "pass: malformed account marker records failure without advancing watermark" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const n = try tmp.dir.realPath(io, &path_buf); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]}); + { + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + _ = try archive.append(row(.account, "did:plc:a", "", "", "\xa1"), 1); + try archive.rotate(); + } + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var wm = watermark.Store.init(&meta); + defer wm.deinit(io); + var seg_path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const seg_path = try std.fmt.bufPrint(&seg_path_buf, "{s}/segments", .{data_dir}); + var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true }); + defer seg_dir.close(io); + var instrumentation: metrics.Stats = .{}; + + try testing.expectError(error.UnexpectedEof, run(testing.allocator, io, seg_dir, &wm, null, .{ + .instrumentation = &instrumentation, + })); + try testing.expectEqual(@as(?u64, null), wm.load(io)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.error_result).load(.monotonic)); + try testing.expectEqual(@as(u64, 0), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic)); +} diff --git a/src/internal/compact/steady.zig b/src/internal/compact/steady.zig index 9105175..0d6bfca 100644 --- a/src/internal/compact/steady.zig +++ b/src/internal/compact/steady.zig @@ -9,8 +9,8 @@ //! under the writer lock, so by rotate's return every sealed row's //! tombstone is observed and later rows land above the pass target. the //! live set is used ONLY for the cap trigger; passes fold their drop -//! windows from disk. a failed pass logs and retries next tick — it must -//! never tear down the daemon. +//! windows from disk. a failed pass logs and retries at the next scheduled +//! pass — it must never tear down the daemon. //! //! every successful rewrite refreshes the resident header/block-index entry //! before its watermark may commit. In-flight readers retain the old @@ -23,6 +23,7 @@ const metrics = @import("../metrics.zig"); const meta_store = @import("../meta_store.zig"); const pass = @import("pass.zig"); const segment = @import("../segment.zig"); +const segment_rewrite = @import("../segment_rewrite.zig"); const tombstone = @import("../tombstone.zig"); const watermark = @import("watermark.zig"); @@ -68,14 +69,12 @@ pub const Compactor = struct { } /// archive on_append hook; runs under the archive writer lock - pub fn onAppend(ctx: *anyopaque, ev: segment.Event) void { + pub fn onAppend(ctx: *anyopaque, ev: segment.Event) !void { const self: *Compactor = @ptrCast(@alignCast(ctx)); self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); - self.live.observe(ev) catch |err| { - log.warn("live tombstone observe failed: {s}", .{@errorName(err)}); - return; - }; + try self.live.observe(ev); + self.publishTombstoneGauges(); if (self.live.len() >= self.tombstone_cap) self.trigger.store(true, .release); } @@ -102,6 +101,7 @@ pub const Compactor = struct { }; } if (self.stats) |s| s.compaction_watermark.store(w, .monotonic); + self.publishTombstoneGauges(); log.info("live tombstones rebuilt above watermark {d}: {d} entries", .{ w, self.live.len() }); } @@ -109,61 +109,86 @@ pub const Compactor = struct { pub fn run(self: *Compactor) void { if (self.interval_s == 0) return; var since_pass: u64 = 0; + var have_pass = false; while (!self.stop_flag.load(.acquire)) { self.io.sleep(Io.Duration.fromSeconds(1), .awake) catch break; since_pass += 1; - const due = since_pass >= self.interval_s or - (self.trigger.load(.acquire) and since_pass >= self.min_trigger_spacing_s); + const periodic_due = since_pass >= self.interval_s; + // Consume the coalesced trigger even when spacing suppresses it, + // exactly like upstream's one-buffered channel. A later append + // above the cap re-arms it; the first-ever trigger is immediate. + const triggered = self.trigger.swap(false, .acq_rel); + const cap_due = triggered and (!have_pass or since_pass >= self.min_trigger_spacing_s); + const due = periodic_due or cap_due; if (!due) continue; - self.trigger.store(false, .release); - const stats = self.runOnce() catch |err| { - // failure policy: never tear down the daemon; retry next tick + const early = cap_due and !periodic_due; + const result = self.runOnce(); + if (early) { + if (self.stats) |s| _ = s.compaction_passes_early_total.fetchAdd(1, .monotonic); + } + const stats = result catch |err| { + // failure policy: never tear down the daemon; retry later log.warn("compaction pass failed: {s} (retrying next interval)", .{@errorName(err)}); since_pass = 0; + have_pass = true; continue; }; log.info("compaction pass: watermark {d}, {d} chunks, {d}/{d} segments rewritten, {d} rows dropped", .{ stats.target_watermark, stats.chunks, stats.segments_rewritten, stats.segments_examined, stats.rows_dropped, }); since_pass = 0; + have_pass = true; } } /// one force-rotate + pass + eviction; also the test seam pub fn runOnce(self: *Compactor) !pass.Stats { + const started = Io.Timestamp.now(self.io, .awake); + const result = self.runOnceInner() catch |err| { + if (self.stats) |s| s.observeCompactionPass(.error_result, elapsedUs(self.io, started)); + return err; + }; + if (self.stats) |s| s.observeCompactionPass(.ok, elapsedUs(self.io, started)); + return result; + } + + fn runOnceInner(self: *Compactor) !pass.Stats { // rows deleted while their segment was active get compacted this // pass; on_append ordering makes this race-free (module doc) try self.archive.rotate(); - // Timestamp import uses the same lock for its cold patch phase. The - // stronger whole-pass scope here is safe: live append uses Archive.mu, - // not rewrite_mu, and an atomic patch/delete rewrite can no longer - // lose the other operation's already-durable generation. - self.archive.lockRewrites(); - defer self.archive.unlockRewrites(); - var wm = watermark.Store.init(self.meta); defer wm.deinit(self.io); var seg_dir = try self.openSegDir(); defer seg_dir.close(self.io); + const reconciled = try self.archive.reconcileManifest(); + if (self.stats) |s| _ = s.compaction_manifest_reconciled_total.fetchAdd(reconciled, .monotonic); // the pass gets no live-set handle: it folds from disk, and we // evict once after the final chunk commit (observe and evict then // never race; entries <= W only cover physically-compacted rows) const stats = try pass.run(self.allocator, self.io, seg_dir, &wm, null, .{ .tombstone_cap = self.tombstone_cap, - .on_rewritten_ctx = self.archive, + .instrumentation = self.stats, + .measure_pass = false, + .on_rewritten_ctx = self, .on_rewritten = refreshManifest, + .rewrite_lock_ctx = self.archive, + .rewrite_lock = lockRewrites, + .rewrite_unlock = unlockRewrites, + .on_committed_ctx = self, + .on_committed = onCommitted, }); - if (wm.load(self.io)) |w| { - self.mu.lockUncancelable(self.io); - defer self.mu.unlock(self.io); - self.live.evict(w); - if (self.stats) |s| s.compaction_watermark.store(w, .monotonic); - } return stats; } + fn publishTombstoneGauges(self: *Compactor) void { + if (self.stats) |s| { + s.compaction_tombstone_set_entries.store(self.live.len(), .monotonic); + s.compaction_tombstone_set_bytes.store(self.live.approxBytes(), .monotonic); + } + } + fn openSegDir(self: *Compactor) !Io.Dir { var buf: [256]u8 = undefined; const seg_path = try std.fmt.bufPrint(&buf, "{s}/segments", .{self.data_dir}); @@ -171,11 +196,34 @@ pub const Compactor = struct { } fn refreshManifest(ctx: ?*anyopaque, idx: u64) !void { + const self: *Compactor = @ptrCast(@alignCast(ctx.?)); + try self.archive.manifest.refresh(self.archive.dir, idx, true); + if (self.stats) |s| _ = s.compaction_manifest_reconciled_total.fetchAdd(1, .monotonic); + } + + fn lockRewrites(ctx: ?*anyopaque) void { + const archive: *archive_mod.Archive = @ptrCast(@alignCast(ctx.?)); + archive.lockRewrites(); + } + + fn unlockRewrites(ctx: ?*anyopaque) void { const archive: *archive_mod.Archive = @ptrCast(@alignCast(ctx.?)); - try archive.manifest.refresh(archive.dir, idx, true); + archive.unlockRewrites(); + } + + fn onCommitted(ctx: ?*anyopaque, watermark_seq: u64) void { + const self: *Compactor = @ptrCast(@alignCast(ctx.?)); + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + self.live.evict(watermark_seq); + self.publishTombstoneGauges(); } }; +fn elapsedUs(io: Io, started: Io.Timestamp) u64 { + return @intCast(@max(0, started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds())); +} + // === tests === const testing = std.testing; @@ -194,6 +242,11 @@ fn row(kind: segment.Kind, did: []const u8, rkey: []const u8, payload: []const u }; } +const KeepAtOrAbove = struct { cutoff: u64 }; +fn keepAtOrAbove(ctx: KeepAtOrAbove, event: segment.Event) bool { + return event.seq >= ctx.cutoff; +} + test "steady: hook feeds live set, runOnce force-rotates, compacts, evicts" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); @@ -212,6 +265,8 @@ test "steady: hook feeds live set, runOnce force-rotates, compacts, evicts" { var c = Compactor.init(testing.allocator, io, &a, &meta, data_dir); defer c.deinit(); + var instrumentation: metrics.Stats = .{}; + c.stats = &instrumentation; a.on_append_ctx = &c; a.on_append = Compactor.onAppend; @@ -220,6 +275,8 @@ test "steady: hook feeds live set, runOnce force-rotates, compacts, evicts" { _ = try a.append(row(.create, "did:plc:a", "k2", "x"), 2); _ = try a.append(row(.delete, "did:plc:a", "k1", ""), 3); try testing.expectEqual(@as(usize, 1), c.live.len()); + try testing.expectEqual(@as(usize, 1), instrumentation.compaction_tombstone_set_entries.load(.monotonic)); + try testing.expect(instrumentation.compaction_tombstone_set_bytes.load(.monotonic) > 64); // force-rotate seals them; the pass drops the superseded create const stats = try c.runOnce(); @@ -233,6 +290,32 @@ test "steady: hook feeds live set, runOnce force-rotates, compacts, evicts" { try testing.expectEqual(@as(u32, 2), resident_blocks.entries[0].event_count); // eviction cleared the covered entry (delete seq 3 <= watermark 3) try testing.expectEqual(@as(usize, 0), c.live.len()); + try testing.expectEqual(@as(usize, 0), instrumentation.compaction_tombstone_set_entries.load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.ok).load(.monotonic)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_manifest_reconciled_total.load(.monotonic)); + + // Model rewrite-succeeded/refresh-failed: mutate the sealed file through + // the real atomic rewriter without notifying the manifest. The next + // steady pass must heal the checksum before its no-op watermark check. + const old_checksum = resident.header.checksum; + const direct = try segment_rewrite.rewriteFile( + testing.allocator, + io, + a.dir, + "seg_0000000000.jss", + .{}, + KeepAtOrAbove{ .cutoff = 3 }, + keepAtOrAbove, + ); + try testing.expect(direct.rewritten); + var rewritten_file = try a.dir.openFile(io, "seg_0000000000.jss", .{}); + defer rewritten_file.close(io); + const rewritten_header = try segment.readHeaderFile(io, &rewritten_file); + try testing.expect(old_checksum != rewritten_header.checksum); + const healed = try c.runOnce(); + try testing.expectEqual(@as(u64, 0), healed.chunks); + try testing.expectEqual(rewritten_header.checksum, a.manifest.segmentByIndex(0).?.header.checksum); + try testing.expectEqual(@as(u64, 2), instrumentation.compaction_manifest_reconciled_total.load(.monotonic)); // rebuild from disk reproduces an empty live set above the watermark var c2 = Compactor.init(testing.allocator, io, &a, &meta, data_dir); @@ -268,6 +351,67 @@ test "steady: cap sets the coalesced trigger" { try testing.expect(c.trigger.load(.acquire)); } +test "steady: first cap trigger runs an early real pass and resets live gauges" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const n = try tmp.dir.realPath(io, &path_buf); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]}); + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var compactor = Compactor.init(testing.allocator, io, &archive, &meta, data_dir); + defer compactor.deinit(); + var instrumentation: metrics.Stats = .{}; + compactor.stats = &instrumentation; + compactor.interval_s = 3600; + compactor.tombstone_cap = 1; + archive.on_append_ctx = &compactor; + archive.on_append = Compactor.onAppend; + var future = try io.concurrent(Compactor.run, .{&compactor}); + defer { + compactor.requestStop(); + _ = future.cancel(io); + } + + _ = try archive.append(row(.delete, "did:plc:a", "k", ""), 1); + for (0..3000) |_| { + if (instrumentation.compaction_passes_early_total.load(.acquire) == 1) break; + try io.sleep(.fromMilliseconds(1), .awake); + } + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes_early_total.load(.acquire)); + try testing.expectEqual(@as(u64, 1), instrumentation.compaction_passes.getPtrConst(.ok).load(.acquire)); + try testing.expectEqual(@as(usize, 0), instrumentation.compaction_tombstone_set_entries.load(.acquire)); +} + +test "steady: malformed account marker fails the real archive append" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const n = try tmp.dir.realPath(io, &path_buf); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]}); + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var compactor = Compactor.init(testing.allocator, io, &archive, &meta, data_dir); + defer compactor.deinit(); + archive.on_append_ctx = &compactor; + archive.on_append = Compactor.onAppend; + + try testing.expectError(error.UnexpectedEof, archive.append(row(.account, "did:plc:a", "", "\xa1"), 1)); + try testing.expectError(error.ArchivePoisoned, archive.append(row(.create, "did:plc:a", "k", "x"), 2)); +} + test "steady: timestamp rewrite lock excludes the real compaction pass without blocking append" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index f8c335b..caeb7c2 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -41,6 +41,8 @@ pub const OrchestratorState = enum { merge, write_phase_steady, }; +pub const CompactionPassResult = enum { ok, error_result }; +pub const CompactionReason = enum { record, sync, account }; const http_series_count = @typeInfo(HttpHandler).@"enum".fields.len * @typeInfo(HttpMethod).@"enum".fields.len * @typeInfo(HttpCode).@"enum".fields.len; @@ -116,8 +118,22 @@ pub const Stats = struct { livestream_stream_error_other_total: Counter = .init(0), livestream_stream_error_missing_total: Counter = .init(0), livestream_last_seen_upstream_event_timestamp_seconds: std.atomic.Value(i64) = .init(0), - /// highest physically-compacted seq (lag = stream_upstream_seq - this) compaction_watermark: Counter = .init(0), + compaction_watermark_lag_us: Counter = .init(0), + compaction_passes: std.enums.EnumArray(CompactionPassResult, Counter) = .initFill(.init(0)), + compaction_pass_duration_count: Counter = .init(0), + compaction_pass_duration_sum_us: Counter = .init(0), + compaction_pass_duration_buckets: [16]Counter = @splat(.init(0)), + compaction_passes_early_total: Counter = .init(0), + compaction_tombstones_collected: std.enums.EnumArray(CompactionReason, Counter) = .initFill(.init(0)), + compaction_tombstone_set_entries: std.atomic.Value(usize) = .init(0), + compaction_tombstone_set_bytes: std.atomic.Value(usize) = .init(0), + compaction_segments_examined_total: Counter = .init(0), + compaction_segments_rewritten_total: Counter = .init(0), + compaction_segments_skipped_clean_total: Counter = .init(0), + compaction_manifest_reconciled_total: Counter = .init(0), + compaction_rows_dropped: std.enums.EnumArray(CompactionReason, Counter) = .initFill(.init(0)), + compaction_bytes_rewritten_total: Counter = .init(0), retry_resynced_total: Counter = .init(0), retry_failed_total: Counter = .init(0), backfill_discovered_total: Counter = .init(0), @@ -295,6 +311,16 @@ pub const Stats = struct { duration_us, ); } + + pub fn observeCompactionPass(self: *Stats, result: CompactionPassResult, duration_us: u64) void { + _ = self.compaction_passes.getPtr(result).fetchAdd(1, .monotonic); + _ = self.compaction_pass_duration_count.fetchAdd(1, .monotonic); + _ = self.compaction_pass_duration_sum_us.fetchAdd(duration_us, .monotonic); + inline for (0..16) |index| { + const upper_us = @as(u64, 100_000) << @intCast(index); + if (duration_us <= upper_us) _ = self.compaction_pass_duration_buckets[index].fetchAdd(1, .monotonic); + } + } }; fn observeSlowHistogram(count: *Counter, sum_us: *Counter, buckets: *[15]Counter, duration_us: u64) void { @@ -900,6 +926,48 @@ pub fn format( w.print( \\# TYPE stream_compaction_watermark_seq gauge \\stream_compaction_watermark_seq {d} + \\# TYPE jetstream_compaction_watermark_seq gauge + \\jetstream_compaction_watermark_seq {d} + \\# TYPE jetstream_compaction_watermark_lag_seconds gauge + \\jetstream_compaction_watermark_lag_seconds {d}.{d:0>6} + \\# TYPE jetstream_compaction_passes_total counter + \\jetstream_compaction_passes_total{{result="ok"}} {d} + \\jetstream_compaction_passes_total{{result="error"}} {d} + \\# TYPE jetstream_compaction_passes_early_total counter + \\jetstream_compaction_passes_early_total {d} + \\# TYPE jetstream_compaction_tombstone_set_entries gauge + \\jetstream_compaction_tombstone_set_entries {d} + \\# TYPE jetstream_compaction_tombstone_set_bytes gauge + \\jetstream_compaction_tombstone_set_bytes {d} + \\# TYPE jetstream_compaction_segments_examined_total counter + \\jetstream_compaction_segments_examined_total {d} + \\# TYPE jetstream_compaction_segments_rewritten_total counter + \\jetstream_compaction_segments_rewritten_total {d} + \\# TYPE jetstream_compaction_segments_skipped_clean_total counter + \\jetstream_compaction_segments_skipped_clean_total {d} + \\# TYPE jetstream_compaction_manifest_reconciled_total counter + \\jetstream_compaction_manifest_reconciled_total {d} + \\# TYPE jetstream_compaction_bytes_rewritten_total counter + \\jetstream_compaction_bytes_rewritten_total {d} + \\ + , .{ + stats.compaction_watermark.load(.monotonic), + stats.compaction_watermark.load(.monotonic), + stats.compaction_watermark_lag_us.load(.monotonic) / std.time.us_per_s, + stats.compaction_watermark_lag_us.load(.monotonic) % std.time.us_per_s, + stats.compaction_passes.getPtrConst(.ok).load(.monotonic), + stats.compaction_passes.getPtrConst(.error_result).load(.monotonic), + stats.compaction_passes_early_total.load(.monotonic), + stats.compaction_tombstone_set_entries.load(.monotonic), + stats.compaction_tombstone_set_bytes.load(.monotonic), + stats.compaction_segments_examined_total.load(.monotonic), + stats.compaction_segments_rewritten_total.load(.monotonic), + stats.compaction_segments_skipped_clean_total.load(.monotonic), + stats.compaction_manifest_reconciled_total.load(.monotonic), + stats.compaction_bytes_rewritten_total.load(.monotonic), + }) catch {}; + + w.print( \\# TYPE stream_retry_repos_total counter \\stream_retry_repos_total{{result="resynced"}} {d} \\stream_retry_repos_total{{result="failed"}} {d} @@ -952,7 +1020,6 @@ pub fn format( \\stream_backfill_inflight_peak_bytes {d} \\ , .{ - stats.compaction_watermark.load(.monotonic), stats.retry_resynced_total.load(.monotonic), stats.retry_failed_total.load(.monotonic), stats.backfill_discovered_total.load(.monotonic), @@ -981,6 +1048,33 @@ pub fn format( stats.backfill_inflight_peak_bytes.load(.monotonic), }) catch {}; + w.print("# TYPE jetstream_compaction_tombstones_collected_total counter\n", .{}) catch {}; + w.print("# TYPE jetstream_compaction_rows_dropped_total counter\n", .{}) catch {}; + inline for (comptime std.enums.values(CompactionReason)) |reason| { + w.print("jetstream_compaction_tombstones_collected_total{{kind=\"{s}\"}} {d}\n", .{ + @tagName(reason), stats.compaction_tombstones_collected.getPtrConst(reason).load(.monotonic), + }) catch {}; + w.print("jetstream_compaction_rows_dropped_total{{reason=\"{s}\"}} {d}\n", .{ + @tagName(reason), stats.compaction_rows_dropped.getPtrConst(reason).load(.monotonic), + }) catch {}; + } + const compaction_count = stats.compaction_pass_duration_count.load(.monotonic); + const compaction_sum_us = stats.compaction_pass_duration_sum_us.load(.monotonic); + w.print("# TYPE jetstream_compaction_pass_duration_seconds histogram\n", .{}) catch {}; + inline for (0..16) |index| { + const upper_us = @as(u64, 100_000) << @intCast(index); + w.print("jetstream_compaction_pass_duration_seconds_bucket{{le=\"{d}.{d:0>6}\"}} {d}\n", .{ + upper_us / std.time.us_per_s, upper_us % std.time.us_per_s, + stats.compaction_pass_duration_buckets[index].load(.monotonic), + }) catch {}; + } + w.print( + \\jetstream_compaction_pass_duration_seconds_bucket{{le="+Inf"}} {d} + \\jetstream_compaction_pass_duration_seconds_sum {d}.{d:0>6} + \\jetstream_compaction_pass_duration_seconds_count {d} + \\ + , .{ compaction_count, compaction_sum_us / std.time.us_per_s, compaction_sum_us % std.time.us_per_s, compaction_count }) catch {}; + const backfill_slow_bounds = [_][]const u8{ "0.01", "0.02", "0.04", "0.08", "0.16", "0.32", "0.64", "1.28", "2.56", "5.12", "10.24", "20.48", "40.96", "81.92", "163.84", diff --git a/src/internal/segment_rewrite.zig b/src/internal/segment_rewrite.zig index 71d8c33..fec0d9f 100644 --- a/src/internal/segment_rewrite.zig +++ b/src/internal/segment_rewrite.zig @@ -16,6 +16,7 @@ const std = @import("std"); const footer_mod = @import("segment_footer.zig"); +const gloom = @import("gloom.zig"); const segment = @import("segment.zig"); const writer_mod = @import("segment_writer.zig"); const zstd = @import("zstd.zig"); @@ -28,6 +29,13 @@ pub const Result = struct { rewritten: bool = false, kept: u64 = 0, dropped: u64 = 0, + bytes_rewritten: u64 = 0, +}; + +pub const Options = struct { + /// null means full scan; a non-null slice enables the segment-level + /// candidate-DID bloom prefilter used by upstream compaction. + candidate_dids: ?[]const []const u8 = null, }; /// rewrite sealed bytes in memory. returns the new sealed file bytes @@ -37,12 +45,28 @@ pub fn rewrite( allocator: Allocator, old_bytes: []const u8, result: *Result, + opts: Options, ctx: anytype, comptime keep: fn (@TypeOf(ctx), segment.Event) bool, ) !?[]u8 { var sealed = try segment.Sealed.parse(allocator, old_bytes); defer sealed.deinit(allocator); + if (opts.candidate_dids) |candidates| { + if (sealed.header.block_did_bloom_offset < sealed.header.did_bloom_offset) + return error.BadFooter; + var filter = try gloom.unmarshal(allocator, old_bytes[sealed.header.did_bloom_offset..sealed.header.block_did_bloom_offset]); + defer filter.deinit(allocator); + var possible = false; + for (candidates) |did| { + if (filter.contains(did)) { + possible = true; + break; + } + } + if (!possible) return null; + } + var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const scratch = arena.allocator(); @@ -106,7 +130,9 @@ pub fn rewrite( .max_witnessed_at = sealed.header.max_witnessed_at, .event_count = result.kept, }); - return try out.toOwnedSlice(allocator); + const owned = try out.toOwnedSlice(allocator); + result.bytes_rewritten = owned.len; + return owned; } /// rewrite one sealed segment file in place (tmp + fsync + rename + @@ -118,6 +144,7 @@ pub fn rewriteFile( io: Io, dir: Io.Dir, name: []const u8, + opts: Options, ctx: anytype, comptime keep: fn (@TypeOf(ctx), segment.Event) bool, ) !Result { @@ -125,7 +152,7 @@ pub fn rewriteFile( defer allocator.free(old_bytes); var result: Result = .{}; - const new_bytes = (try rewrite(allocator, old_bytes, &result, ctx, keep)) orelse return result; + const new_bytes = (try rewrite(allocator, old_bytes, &result, opts, ctx, keep)) orelse return result; defer allocator.free(new_bytes); var tmp_buf: [128]u8 = undefined; @@ -198,19 +225,23 @@ fn keepAbove(ctx: DropBelow, ev: segment.Event) bool { return ev.seq >= ctx.cutoff; } +fn unreachableKeep(_: void, _: segment.Event) bool { + @panic("segment bloom false negative: predicate must not run"); +} + test "rewrite: clean segment returns null, file layout preserved on drops" { const old = try buildSealed(testing.allocator); defer testing.allocator.free(old); // clean: nothing drops var clean_res: Result = .{}; - try testing.expectEqual(@as(?[]u8, null), try rewrite(testing.allocator, old, &clean_res, DropBelow{ .cutoff = 0 }, keepAbove)); + try testing.expectEqual(@as(?[]u8, null), try rewrite(testing.allocator, old, &clean_res, .{}, DropBelow{ .cutoff = 0 }, keepAbove)); try testing.expect(!clean_res.rewritten); try testing.expectEqual(@as(u64, 30), clean_res.kept); // drop the whole first block (seqs 1..10) plus half the second var res: Result = .{}; - const new_bytes = (try rewrite(testing.allocator, old, &res, DropBelow{ .cutoff = 16 }, keepAbove)).?; + const new_bytes = (try rewrite(testing.allocator, old, &res, .{}, DropBelow{ .cutoff = 16 }, keepAbove)).?; defer testing.allocator.free(new_bytes); try testing.expect(res.rewritten); try testing.expectEqual(@as(u64, 15), res.dropped); @@ -244,6 +275,22 @@ test "rewrite: clean segment returns null, file layout preserved on drops" { try testing.expectEqual(@as(usize, 15), total); } +test "rewrite: unrelated candidate DIDs skip the real segment decode" { + const old = try buildSealed(testing.allocator); + defer testing.allocator.free(old); + var result: Result = .{}; + const candidates = [_][]const u8{"did:plc:not-in-this-segment"}; + try testing.expectEqual(@as(?[]u8, null), try rewrite( + testing.allocator, + old, + &result, + .{ .candidate_dids = &candidates }, + {}, + unreachableKeep, + )); + try testing.expect(!result.rewritten); +} + test "rewriteFile: atomic in-place rewrite; stale tmp reclaim" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); @@ -268,7 +315,7 @@ test "rewriteFile: atomic in-place rewrite; stale tmp reclaim" { try cleanStaleTmp(testing.allocator, io, dir); try testing.expectError(error.FileNotFound, dir.access(io, "seg_0000000000.jss.tmp", .{})); - const res = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", DropBelow{ .cutoff = 21 }, keepAbove); + const res = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", .{}, DropBelow{ .cutoff = 21 }, keepAbove); try testing.expect(res.rewritten); try testing.expectEqual(@as(u64, 20), res.dropped); @@ -280,6 +327,6 @@ test "rewriteFile: atomic in-place rewrite; stale tmp reclaim" { try testing.expectEqual(@as(u32, 3), sealed.header.block_count); // second pass over the compacted file is clean - const res2 = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", DropBelow{ .cutoff = 21 }, keepAbove); + const res2 = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", .{}, DropBelow{ .cutoff = 21 }, keepAbove); try testing.expect(!res2.rewritten); } diff --git a/src/internal/tombstone.zig b/src/internal/tombstone.zig index 8c37d5c..d89bfcd 100644 --- a/src/internal/tombstone.zig +++ b/src/internal/tombstone.zig @@ -22,6 +22,7 @@ const segment = @import("segment.zig"); const Allocator = std.mem.Allocator; pub const DidReason = enum { sync, account }; +pub const DropReason = enum { record, sync, account }; pub const DidTombstone = struct { seq: u64, @@ -39,20 +40,22 @@ pub const Counts = struct { /// used only for the cap trigger, never for drop decisions. pub const Snapshot = struct { allocator: Allocator, - /// owns all keys - arena: std.heap.ArenaAllocator, + /// both maps own their individually allocated keys so eviction releases + /// real resident memory rather than retaining an arena high-water mark. records: std.StringHashMapUnmanaged(u64) = .empty, dids: std.StringHashMapUnmanaged(DidTombstone) = .empty, - counts: Counts = .{}, pub fn init(allocator: Allocator) Snapshot { - return .{ .allocator = allocator, .arena = std.heap.ArenaAllocator.init(allocator) }; + return .{ .allocator = allocator }; } pub fn deinit(self: *Snapshot) void { + var records = self.records.keyIterator(); + while (records.next()) |key| self.allocator.free(key.*); + var dids = self.dids.keyIterator(); + while (dids.next()) |key| self.allocator.free(key.*); self.records.deinit(self.allocator); self.dids.deinit(self.allocator); - self.arena.deinit(); } pub fn len(self: *const Snapshot) usize { @@ -68,16 +71,19 @@ pub const Snapshot = struct { const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return; const gop = try self.records.getOrPut(self.allocator, key); if (!gop.found_existing) { - gop.key_ptr.* = try self.arena.allocator().dupe(u8, key); + const owned = self.allocator.dupe(u8, key) catch |err| { + _ = self.records.remove(key); + return err; + }; + gop.key_ptr.* = owned; gop.value_ptr.* = ev.seq; - self.counts.record += 1; } else if (ev.seq > gop.value_ptr.*) { gop.value_ptr.* = ev.seq; } }, .sync => try self.observeDid(ev.did, ev.seq, .sync), .account => { - if (accountIsDelete(self.allocator, ev.payload)) + if (try accountIsDelete(self.allocator, ev.payload)) try self.observeDid(ev.did, ev.seq, .account); }, .create, .create_resync, .identity => {}, @@ -87,12 +93,12 @@ pub const Snapshot = struct { fn observeDid(self: *Snapshot, did: []const u8, seq: u64, reason: DidReason) !void { const gop = try self.dids.getOrPut(self.allocator, did); if (!gop.found_existing) { - gop.key_ptr.* = try self.arena.allocator().dupe(u8, did); + const owned = self.allocator.dupe(u8, did) catch |err| { + _ = self.dids.remove(did); + return err; + }; + gop.key_ptr.* = owned; gop.value_ptr.* = .{ .seq = seq, .reason = reason }; - switch (reason) { - .sync => self.counts.sync += 1, - .account => self.counts.account += 1, - } } else if (seq > gop.value_ptr.seq) { gop.value_ptr.* = .{ .seq = seq, .reason = reason }; } @@ -102,19 +108,74 @@ pub const Snapshot = struct { /// materialization row strictly below a covering tombstone. marker rows /// are never dropped. pub fn shouldDrop(self: *const Snapshot, ev: segment.Event) bool { + return self.dropReason(ev) != null; + } + + /// Upstream attributes each physically removed row to the covering + /// tombstone. DID-wide tombstones win over record tombstones. + pub fn dropReason(self: *const Snapshot, ev: segment.Event) ?DropReason { switch (ev.kind) { .create, .update, .create_resync => {}, - else => return false, + else => return null, } if (self.dids.get(ev.did)) |t| { - if (t.seq > ev.seq) return true; + if (t.seq > ev.seq) return switch (t.reason) { + .sync => .sync, + .account => .account, + }; } var key_buf: [1024]u8 = undefined; - const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return false; + const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return null; if (self.records.get(key)) |t_seq| { - if (t_seq > ev.seq) return true; + if (t_seq > ev.seq) return .record; } - return false; + return null; + } + + /// Exact counterpart of upstream Snapshot's metric accounting. Counts + /// are derived from the final map so a later account tombstone replacing + /// a sync tombstone is attributed to account, not its insertion history. + pub fn countsByReason(self: *const Snapshot) Counts { + var result: Counts = .{ .record = self.records.count() }; + var it = self.dids.iterator(); + while (it.next()) |entry| switch (entry.value_ptr.reason) { + .sync => result.sync += 1, + .account => result.account += 1, + }; + return result; + } + + /// Key bytes plus upstream's conservative 64-byte map-entry overhead. + pub fn approxBytes(self: *const Snapshot) usize { + var result: usize = 0; + var rit = self.records.iterator(); + while (rit.next()) |entry| result += entry.key_ptr.len - 2 + 64; + var dit = self.dids.iterator(); + while (dit.next()) |entry| result += entry.key_ptr.len + 64; + return result; + } + + /// Distinct DIDs represented by this snapshot. Returns null once the + /// upstream narrowing cutoff is exceeded, signaling a full scan. + pub fn candidateDids(self: *const Snapshot, allocator: Allocator, max: usize) !?[][]const u8 { + var set: std.StringHashMapUnmanaged(void) = .empty; + defer set.deinit(allocator); + var records = self.records.iterator(); + while (records.next()) |entry| { + const nul = std.mem.indexOfScalar(u8, entry.key_ptr.*, 0).?; + try set.put(allocator, entry.key_ptr.*[0..nul], {}); + if (set.count() > max) return null; + } + var dids = self.dids.iterator(); + while (dids.next()) |entry| { + try set.put(allocator, entry.key_ptr.*, {}); + if (set.count() > max) return null; + } + const result = try allocator.alloc([]const u8, set.count()); + var index: usize = 0; + var it = set.keyIterator(); + while (it.next()) |did| : (index += 1) result[index] = did.*; + return result; } /// drop entries with seq <= watermark (they cover only rows already @@ -122,11 +183,19 @@ pub const Snapshot = struct { pub fn evict(self: *Snapshot, watermark: u64) void { var rit = self.records.iterator(); while (rit.next()) |e| { - if (e.value_ptr.* <= watermark) self.records.removeByPtr(e.key_ptr); + if (e.value_ptr.* <= watermark) { + const key = e.key_ptr.*; + self.records.removeByPtr(e.key_ptr); + self.allocator.free(key); + } } var dit = self.dids.iterator(); while (dit.next()) |e| { - if (e.value_ptr.seq <= watermark) self.dids.removeByPtr(e.key_ptr); + if (e.value_ptr.seq <= watermark) { + const key = e.key_ptr.*; + self.dids.removeByPtr(e.key_ptr); + self.allocator.free(key); + } } } }; @@ -140,10 +209,10 @@ fn recordKey(buf: []u8, did: []const u8, collection: []const u8, rkey: []const u /// does an #account payload mark the account deleted? upstream decodes the /// envelope per fold; matches active == false && status == "deleted". -pub fn accountIsDelete(allocator: Allocator, payload: []const u8) bool { +pub fn accountIsDelete(allocator: Allocator, payload: []const u8) !bool { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); - const env = zat.cbor.decodeAll(arena.allocator(), payload) catch return false; + const env = try zat.cbor.decodeAll(arena.allocator(), payload); const active = env.getBool("active") orelse return false; if (active) return false; const status = env.getString("status") orelse return false; @@ -199,8 +268,9 @@ test "did tombstones: sync and account-delete cover the whole did" { try snap.observe(row(.sync, 20, "did:plc:b", "", "", "")); try snap.observe(row(.account, 30, "did:plc:c", "", "", account_deleted)); try snap.observe(row(.account, 40, "did:plc:d", "", "", account_active)); // not a delete - try testing.expectEqual(@as(u64, 1), snap.counts.sync); - try testing.expectEqual(@as(u64, 1), snap.counts.account); + const counts = snap.countsByReason(); + try testing.expectEqual(@as(u64, 1), counts.sync); + try testing.expectEqual(@as(u64, 1), counts.account); try testing.expect(snap.shouldDrop(row(.create, 19, "did:plc:b", "app.bsky.feed.post", "k", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 20, "did:plc:b", "app.bsky.feed.post", "k", "x"))); @@ -219,3 +289,15 @@ test "evict removes entries at or below the watermark" { try testing.expect(snap.shouldDrop(row(.create, 14, "did:plc:a", "app.bsky.feed.post", "k2", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 4, "did:plc:a", "app.bsky.feed.post", "k1", "x"))); } + +test "account tombstones reject malformed CBOR and metrics reflect final reason" { + var snap = Snapshot.init(testing.allocator); + defer snap.deinit(); + try testing.expectError(error.UnexpectedEof, snap.observe(row(.account, 1, "did:plc:a", "", "", "\xa1"))); + try snap.observe(row(.sync, 2, "did:plc:a", "", "", "")); + try snap.observe(row(.account, 3, "did:plc:a", "", "", account_deleted)); + const counts = snap.countsByReason(); + try testing.expectEqual(@as(u64, 0), counts.sync); + try testing.expectEqual(@as(u64, 1), counts.account); + try testing.expectEqual(@as(usize, "did:plc:a".len + 64), snap.approxBytes()); +} diff --git a/src/main.zig b/src/main.zig index bd514e0..beee1ef 100644 --- a/src/main.zig +++ b/src/main.zig @@ -296,6 +296,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .selected_repos = backfill_repos, .workers = backfill_workers, .max_inflight_bytes = backfill_max_inflight_bytes, + .compaction_enabled = compaction_interval_s != 0, }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats); hub.serving.store(true, .release); } -- 2.51.2