Skip to content

Refactor ExistanceStore - #418

Merged
blakehatch merged 1 commit into
mainfrom
fix-existance-store
Nov 27, 2023
Merged

blakehatch merged 1 commit into
mainfrom
fix-existance-store

Conversation

@allada

@allada allada commented Nov 26, 2023 •

Copy link
Copy Markdown
Contributor

Refactor Existance Store to be compatible with store API.


This change is Reviewable

@allada allada left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+@blakehatch

Reviewable status: 0 of 10 files reviewed, all discussions resolved (waiting on @blakehatch)

@allada
allada force-pushed the fix-existance-store branch from b00cce4 to 04a11ab Compare November 26, 2023 23:54

@blakehatch blakehatch left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

:lgtm:

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(&not_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 allada left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.
@allada
allada force-pushed the fix-existance-store branch from 04a11ab to 4053335 Compare November 27, 2023 01:10

@allada allada left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 blakehatch left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed 1 of 1 files at r2, all commit messages.
Reviewable status: :shipit: complete! all files reviewed, all discussions resolved (waiting on @allada)

@blakehatch
blakehatch merged commit 08b2954 into main Nov 27, 2023
@blakehatch
blakehatch deleted the fix-existance-store branch November 27, 2023 01:14
TripleKai pushed a commit to TripleKai/turbo-cache that referenced this pull request Nov 29, 2023
Refactor Existance Store to be compatible with store API.
MarcusSorealheis pushed a commit to MarcusSorealheis/nativelink that referenced this pull request Nov 3, 2025
Refactor Existance Store to be compatible with store API.
rejuvenile added a commit to rejuvenile/nativelink that referenced this pull request Aug 19, 2026
…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>
rejuvenile added a commit to rejuvenile/nativelink that referenced this pull request Aug 19, 2026
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>
rejuvenile added a commit to rejuvenile/nativelink that referenced this pull request Aug 19, 2026
…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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants