Repository navigation
Refactor ExistanceStore - #418
Conversation
allada
left a comment
There was a problem hiding this comment.
Reviewable status: 0 of 10 files reviewed, all discussions resolved (waiting on @blakehatch)
b00cce4 to
04a11ab
Compare
blakehatch
left a comment
There was a problem hiding this comment.
Reviewed 10 of 10 files at r1, all commit messages.
Reviewable status: all files reviewed, 2 unresolved discussions (waiting on @allada)
native-link-store/src/existence_store.rs line 88 at r1 (raw file):
self.pin_inner() .has_with_results(¬_cached_digests, &mut inner_results) .await?;
Nit: .err_tip(|| "In ExistenceStore::inner_has_with_results")
native-link-util/src/evicting_map.rs line 33 at r1 (raw file):
#[derive(Serialize, Deserialize, PartialEq, Debug, Clone)] pub struct SerializedLRU { pub data: Vec<(DigestInfo, i32)>,
Nit: Is there a way to use evicting_map for it's pruning capabilities without having to store any data as existence is just being checked? Probably an over-optimization especially since this approach limits the size of the data stored and by extension the upside of an optimization like this but something to consider.
allada
left a comment
There was a problem hiding this comment.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @allada)
native-link-util/src/evicting_map.rs line 33 at r1 (raw file):
Previously, blakehatch (Blake Hatch) wrote…
Nit: Is there a way to use evicting_map for it's pruning capabilities without having to store any data as existence is just being checked? Probably an over-optimization especially since this approach limits the size of the data stored and by extension the upside of an optimization like this but something to consider.
Storing a () (ie: void) costs 1 byte (in C, so probably the same in rust), so the overhead is so small it is negligible.
Long answer: Yes, but it would be a pain to refactor and use.
Refactor Existance Store to be compatible with store API.
04a11ab to
4053335
Compare
allada
left a comment
There was a problem hiding this comment.
Reviewable status: 9 of 10 files reviewed, all discussions resolved (waiting on @blakehatch)
native-link-store/src/existence_store.rs line 88 at r1 (raw file):
Previously, blakehatch (Blake Hatch) wrote…
Nit:
.err_tip(|| "In ExistenceStore::inner_has_with_results")
Done.
blakehatch
left a comment
There was a problem hiding this comment.
Reviewed 1 of 1 files at r2, all commit messages.
Reviewable status:complete! all files reviewed, all discussions resolved (waiting on @allada)
Refactor Existance Store to be compatible with store API.
Refactor Existance Store to be compatible with store API.
…d corrupt StreamState (TraceMachina#418) Production observation 2026-05-12: post-TraceMachina#413-Option-A deploy 21:55:38 PDT, 22 follow-on `Code::Internal` failures concentrated on 2 large blobs (`38f6b374...-155605824` 155 MB, 9 retries; `bff125c2 ...-218429440` 218 MB, 11 retries — bff125c2 ABANDONED by client without successful commit). Two error variants observed depending on retry timing: `"Failed to write to data, receiver disconnected"` (buf_channel.rs:215-221, on first send-after-resume) and `"Tried to send while stream is closed"` (line 185, on subsequent sends). Both share root cause: `ActiveStreamGuard::Drop` recycles broken `StreamState` (dead `tx`, completed-with-Err `store_update_fut`) back into `IdleStream`, so Bazel client retries within the 60-sec sweeper TTL resume the same broken state. **Lifecycle invariant** (per CLAUDE.md "Admission/Eviction/Pin Composability" — `active_uploads` triangle: gate=IdleStream-resume, eviction=60s sweeper, pin=active_uploads-entry-while-in-use): a `StreamState` is RESUMABLE iff its future is paused mid-await AND its tx is connected to a live rx. Pre-fix: gate active + sweeper TTL too long for retry budget + no compensating cleanup → wedge. Post-fix: corrupt state arm closes the gate at Drop time. Discriminator is paused-vs-completed future state: - **Paused** (legitimate Bazel TCP-flap recovery): `tokio::try_join!` cancels the inner `store_update_fut` mid-await — but does NOT drop it; the future survives in `StreamState`, paused at `reader.recv().await` with `rx` still live. The wrapped body's `if result.is_err()` line never runs, so `store_errored` stays false. Recycle into `IdleStream` → next QueryWriteStatus + retry resumes cleanly. Existing tests `resume_write_success`, `restart_write_success`, `restart_mid_stream_write_success`, `resumable_write_reconnect_same_uuid`, `partial_write_bytes_counter_tracks_idle_and_resume`, `memory_pressure_evicts_oldest_idle_stream(s)` cover this path. - **Completed-with-Err** (the cascade-cancel-from-store case this fix addresses): `store.update()` returns Err, the wrapped future body completes, `store_errored` flips true, scope ends, `rx` drops. StreamState now unrecoverable; ANY retry hitting `tx.send` returns `Internal: receiver disconnected` (or, on subsequent send, `stream is closed`). Drop discards the entry. `tx.is_pipe_broken()` belt-and-suspenders is kept as future-regression defense (currently unreachable as an independent signal — every path that closes `tx` also sets `store_errored`). **Discriminator implementation:** - New `Arc<AtomicBool> store_errored` on `StreamState`. Set ONLY when the wrapped `store_update_fut` body returns Err (i.e. the future ran to completion). Cancellation does NOT set it (the future is dropped at its current suspension point and the assignment never runs). - Memory ordering: `Release` on store, `Acquire` on load (Relaxed would suffice since both happen in the same task context, but Acquire/Release is the safer choice and not measurably costly). - `ActiveStreamGuard::Drop` reads both `store_errored.load(Acquire)` and `tx.is_pipe_broken()`; if either, REMOVE the entry from `active_uploads` instead of recycling into `IdleStream`. New `warn!("TraceMachina#418: discarding corrupt StreamState ...")` for production observability. **Test:** `cancelled_chunked_write_replaced_not_recycled_on_retry` in `tests/bytestream_server_test.rs`. Production composition (real `ByteStreamServer::write` over real gRPC + real chunked store via `FailAfterNBytesStore` test fixture). `tokio::time::timeout(5s)` deadlock detector + bespoke `STREAM_CLOSED_BUG_MSG` panic message (NOT generic `is_err()` — guards against the `Elapsed` mask trap per CLAUDE.md). Test tolerates `Ok | Code::Unavailable | Code::InvalidArgument` on retry (any of these is a valid fix-path outcome); FORBIDS `Code::Internal: "Tried to send while stream is closed"` (the bug shape). Mutation: gate the corrupt-state arm to `false` → red-fails with the specific `STREAM_CLOSED_BUG_MSG`. Restored; test re-runs green. **Sibling-bug audit:** enumerated all `impl Drop for` in `bytestream_server.rs`. `InFlightWritesGuard::Drop` (TraceMachina#402) always REMOVES, no recycling — safe. `LoggingReadStream::Drop` logging only — safe. `ActiveStreamGuard::Drop` fixed. `ExpectedDropRx::Drop` is read-half mark — unrelated. `IdleStreamSweeper`: with corrupt entries removed at Drop, sweeper sees only healthy entries — unchanged behavior. Reviewer cadre (4): - distributed-systems-reviewer: APPROVE-WITH-FINDINGS — the `bad_client_input` framing concern from prior bundle reframed as "lifecycle invariant: paused vs completed" (this commit message reflects). Memory entry filed (`feedback_paused_vs_completed_future_state.md`). - code-reviewer: APPROVE-WITH-FINDINGS — doc reframe applied (this commit), `bytes_consumed` dead instrumentation removed (this commit), atomic ordering Acquire/Release confirmed correct. - testing-czar: APPROVE-WITH-FINDINGS — 152/152 crate sweep PASS, mutation independently re-verified RED-fail with bespoke message, asymmetric coverage (under-action by new test + over-action by 7 existing tests). - red-team: APPROVE-WITH-CONCERNS — flagged that the discriminator misses panic-in-`store.update()` (panic unwinds before the flag- set line). Filed as TraceMachina#422 (`TraceMachina#418 sibling: panic-safety`) for follow-up via `scopeguard::defer!` or PanicGuard RAII. Verdict: ship the named-Err-class fix; panic case is separate scope. Build: `cargo check -p nativelink-service` clean. Test: 34/34 bytestream_server_test pass; new test PASSES with mutation verified. Follow-ups: - TraceMachina#421 composability harness: assert_no_dead_state_after_drop<G> (testing-czar M3 — defer until 3rd RAII-guard member surfaces) - TraceMachina#422 panic-safety on the discriminator (red-team Q9) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Mechanical sweep demoting audited high-volume log sites. No behavior changes
to data flow, error handling, or metrics.
Behavior changes (mandatory per .claude/rules/architectural-changes.md):
WARN → debug:
buf_channel.rs:228: warn→debug, "channel backpressure (>1s wait)", 1,373×/12h
buf_channel.rs:694: warn→debug/warn (debug for <60s gap/wait; warn kept for
≥60s gap or ≥60s recv wait — previously fired at >5s), 39,424×/12h
buf_channel.rs:963: warn→debug, "Receiver dropped mid-stream", 463×/12h
stall_detector.rs:302: warn→debug (benign SuppressDumpProducerDriven branch
only; stack-dump branch unchanged), 3,049×/12h
proto_stream_utils.rs:87: warn→debug, "GrpcStore::write made no progress", 477×/12h
proto_stream_utils.rs:182: warn→debug, "read path made no progress", 122×/12h
bytestream_server.rs:616: warn→debug, "slow read completion (>=5s, <1MB/s)", 153×/12h
bytestream_server.rs:2148: warn→debug, "mirror channel closed, dropping mirror", 433×/12h
grpc_store.rs:1788: warn→debug, "GrpcStore::write succeeded but took > 60s", 410×/12h
fast_slow_store.rs:5525: warn→debug, "background slow write task was delayed
before starting" (streaming update path), 337×/12h (both paths combined)
fast_slow_store.rs:5874: warn→debug, "update_oneshot: background slow write
task was delayed before starting"
fast_slow_store.rs:4913,4939,4963: warn→debug, "slow lock acquire" (3 sites,
has_with_results local_only_reads: in_flight/chunked/mirror), 65×/12h
fast_slow_store.rs:4998,5035,5065: warn→debug, "slow lock acquire" (3 sites,
has_with_results server: in_flight/chunked/mirror)
filesystem_store.rs:1229: warn→debug, "update_file slow phases (>100ms)", 67×/12h
filesystem_store.rs:2162: warn→debug, "update slow write (>100ms)", 115×/12h
cas_server.rs:1011: warn→debug, "GetTree: slow BFS level (>100ms)", 85×/12h
fast_slow_store.rs:4918-ish: see slow lock acquire above
streaming_blob.rs:1323: warn→debug, "streaming blob reader slow notify wakeup", 39×/12h
api_worker_scheduler.rs:3756: warn→debug, "tree resolution slow directory fetch", 69×/12h
WARN → trace:
fs.rs:571: warn→trace, "read_file_to_channel: slow read syscall (>100ms)", 311×/12h
INFO → debug:
grpc_store.rs:1445: info→debug, "TraceMachina#59 GrpcStore::write: starting ByteStream write"
grpc_store.rs:1673: info→debug, "TraceMachina#59 GrpcStore::write: RPC returned Ok (normal success)"
grpc_store.rs:1777: info→debug, "TraceMachina#59 GrpcStore::write: completed successfully"; combined ~464K/12h
bytestream_server.rs:660: info→debug, "ByteStream::read: CAS read completed", ~243K/12h
api_worker_scheduler.rs:1931: info→debug, "scoring: pre-match state", ~207K/12h
bytestream_server.rs:3092: info→debug, "TraceMachina#62 ByteStream::Write: already-exists short-circuit", ~155K/12h
chunked_write_handler.rs:1013: info→debug, "WriteChunked RPC entry"
chunked_write_handler.rs:1317: info→debug, "in_flight handler-local entry inserted (write_chunked_inner)"
chunked_write_handler.rs:2901: info→debug, "chunked dispatch: blob committed (reaper)"
chunked_write_handler.rs:3072: info→debug, "in_flight handler-local entry inserted (dispatch_chunks_to_driver)"
chunked_write_handler.rs:3119: info→debug, "producer-arrival probe armed (TraceMachina#394/TraceMachina#413 Phase 1)"
chunked_write_handler.rs:3827: info→debug, "chunked_in_flight registered"
chunked_write_handler.rs:3877: info→debug, "chunked_in_flight guard disarmed"
chunked_write_handler.rs:3934: info→debug, "chunked_in_flight removed"; combined ~120K/12h
chunked_write_handler.rs:1517: "WriteChunked: blob committed" KEPT at info (terminal state transition)
fast_slow_store.rs:1854: info→debug, "FastSlowStore::update (chunked): admission accepted", ~54K/12h
fast_slow_store.rs:2093: info→debug, "FastSlowStore::update (chunked): commit complete"
fast_slow_store.rs:1830: info→debug, "data_stream_fut step", ~16K/12h
existence_cache_store.rs:848: info→debug, "returning NotFound to caller", ~17K/12h
cas_server.rs:1067: info→debug, "GetTree: resolved directory tree", ~16K/12h
api_worker_scheduler.rs:2384: info→debug, "inline tree resolution complete, caching"
api_worker_scheduler.rs:2396: info→debug, "tree cache byte-bounded eviction"
api_worker_scheduler.rs:2643: info→debug, "prefetch: spawning batched push"
api_worker_scheduler.rs:2827: info→debug, "prefetch: completed batched push"; combined ~51K/12h
api_worker_scheduler.rs:1263: info→debug, "phase6 scheduler dispatching StartAction"
worker_api_server.rs:1493: info→debug, "phase6 scheduler received action-N completion"
NOT touched (kept at warn per spec):
bytestream_server.rs:841 "TraceMachina#418: discarding corrupt StreamState" — KEEP warn
proto_stream_utils.rs aggregate warn (rate-limited burst warn) — KEEP warn
chunked per-chunk back-edge lines (TraceMachina#47 active) — KEEP warn
redis_store slow-pipeline/slow-command warns — KEEP warn
No tests required for pure level changes.
cargo check -p nativelink-util -p nativelink-store -p nativelink-service -p nativelink-scheduler — clean (exit 0). Log: /tmp/logsweep-check1.log.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…dary test + nits Tier-3 cadre on c09ad35 returned unanimous SHIP; this addresses the convergent MINOR-1 + the boundary-test request + doc/comment nits before deploy. Reviews: .claude/reviews/c09ad356-zombie-reaper/. Invariant being violated (MINOR-1): a REAPED zombie's in-flight slot + partial store write must be released IMMEDIATELY, not after the ~60s idle TTL; a NON-reaped idle stream must still recycle normally for legit Bazel-flap resume. Mechanism that violated it: when the select! cancel arm fired, `ActiveStreamGuard::Drop` saw store_errored=false + tx not pipe-broken (the cancel dropped process_client_stream mid-await without setting either), so it fell through to the recycle-to-IdleStream path (bytestream_server.rs Drop). The dead IdleStream lived ~60s until Pass-1 TTL eviction AND its partial bytes were added to partial_write_bytes — under the real burst (auditor-confirmed 20+ simultaneous zombies, recurring) that backlog feeds Pass-2 memory-pressure eviction of LEGITIMATE idle streams (red-team pre-mortem). The error text also misleadingly claimed "in-flight slot released" when it was not. Mechanism that re-establishes it: a new `zombie_reaped: Arc<AtomicBool>` on ActiveStreamGuard. The cancel arm sets it (cloned before the &mut stream_state borrow) before returning Err(Aborted); Drop checks it FIRST (before the TraceMachina#418 corrupt-state arm) and does an immediate `active_uploads.remove(uuid)` + metric decrement + return — dropping stream_state cancels the incomplete store_update_fut (correct: incomplete blob must not be stored). Does NOT add to partial_write_bytes (no IdleStream created). Error message corrected to "in-flight slot and StallGuard removed immediately, incomplete store write cancelled". Composite invariants this fix interacts with: - NON-reaped idle recycle: unchanged. A client mid-flap (Bazel TCP reconnect) whose write was NOT reaped still falls through to the recycle path and resumes via QueryWriteStatus → into_active_stream. Verified by the unchanged 44-test bytestream_server_test suite (resume_write_* family). - TraceMachina#418 corrupt-state arm: unchanged; zombie arm is checked first but only fires when zombie_reaped is set, which only the cancel arm sets. - QueryWriteStatus after reap: entry gone → store.has()=None (partial never committed) → committed_size 0, complete=false → Bazel restarts from 0 (correct). Verified by reaped_zombie_slot_freed_immediately_not_after_idle_ttl. Test that proves the fix: reaped_zombie_slot_freed_immediately_not_after_idle_ttl — after a reap, QueryWriteStatus returns committed_size=0 (entry gone NOW), not the partial 16 of a lingering IdleStream. Mutation: disable the Drop zombie arm → entry recycles → QueryWriteStatus returns committed_size=16 → test red-fails with its bespoke message. Verified. Also in this commit: - Boundary test reap_boundary_just_under_survives_just_over_reaps (convergent pair-a B1 + pair-b M3): one run exercises both sides of the >= boundary — progress every (threshold - 1s) sustained across >2 thresholds is NOT reaped, then just-over silence IS reaped. Pins the boundary against future threshold drift. (Its just-under phase depends on the byte-progress reset — the reset mutation red-fails it too, re-confirmed.) - M2 doc-fix: progressed_then_silent_zombie_is_reaped's mutation actually fails at the contains("zombie write reaped") assert (writer returns "client closed write stream" when frame_tx drops at t≈14s), NOT at the outer timeout. Test doc-comment corrected to match the real failure mode. - TRIVIAL: empty-blob comment now reads "current_bytes >= expected_size (== 0)". - M4 (OPTIONAL, left conservative): added a comment noting the plain-try_join! else branch is defensively unreachable today (all guard constructors set zombie_reap_rx: Some(...)); not removed to avoid scope creep. Re-verified all 3 original mutations still catch after the polish: - cancel arm disabled → all 4 reap-dependent tests fail, 2 over-action pass. - MAJOR-2 >= expected_size guard disabled → all_bytes test fails (bespoke msg). - byte-progress reset disabled → progressed_then_silent + boundary fail, slow_alive correctly unaffected. Behavior changes: - A reaped zombie's active_uploads entry is removed IMMEDIATELY (was: recycled to IdleStream for ~60s). Production composition (VerifyStore → ExistenceCache → SizePartitioning → MemoryStore/FilesystemStore + WPS mirror tee): frees the in-flight slot at reap time and stops dead IdleStreams from inflating partial_write_bytes / triggering Pass-2 eviction of legit idle streams. Incomplete store_update_fut is cancelled (correct — partial garbage not stored). - Reap error message text changed (Code unchanged = Aborted; clients still retry). - ActiveStreamGuard grows by one Arc<AtomicBool> (8B ptr + heap AtomicBool) per in-flight write. Negligible; bounded by h2 concurrency. - NON-reaped paths (graceful finish, TraceMachina#418 corrupt-state, idle recycle): unchanged. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Refactor Existance Store to be compatible with store API.
This change is