Scheduler config now supports multiple impls by allada · Pull Request #141 · TraceMachina/nativelink

allada · 2023-07-07T04:55:10Z

This is a step in the direction to support different scheduler impls.

towards: #140

This change is

This is a step in the direction to support different scheduler impls. To upgrade, you'll need to add a "local" property to the "scheduler" configs. See example files that where changed in this commit. towards: #140

chrisstaite-menlo

Reviewed 17 of 18 files at r1, all commit messages.
Reviewable status: 17 of 18 files reviewed, all discussions resolved

chrisstaite-menlo

Reviewed 1 of 18 files at r1.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @allada)

…epoch change Worker-side: generate process-lifetime boot_epoch_id (random u64 from UUIDv4 first 8 bytes) lazily in OnceLock; send in every ConnectWorkerRequest. Server-side (WorkerApiServer): track per-endpoint state {boot_epoch, owner_worker_id} under a parking_lot::Mutex. On connect, wipe locality_map[endpoint] iff prev epoch differs from the new one (or new == 0 for legacy). On disconnect, only the connection that still owns the endpoint runs remove_endpoint — suppresses the documented race where an old connection's cleanup task wipes a newer worker's just-registered entries. Atomicity: the wipe + owner update is performed under the endpoint_state mutex while holding locality_map.write(); the worker waits on ConnectionResult before sending BlobsAvailable, so the new worker's first registrations cannot interleave with the wipe. The disconnect-cleanup task takes the same mutex around its own remove_endpoint, preventing TOCTOU between owner check and wipe. Tests (nativelink-service/tests/worker_api_server_test.rs): - boot_epoch_same_preserves_locality_entries_test - boot_epoch_different_wipes_locality_entries_test - boot_epoch_zero_legacy_wipes_locality_entries_test - boot_epoch_new_blobs_survive_old_disconnect_cleanup_test Mutation: forcing 'always wipe' fails same_preserves; forcing 'no owner check' on cleanup fails new_blobs_survive_old_cleanup + same_preserves.

…#141 review nit)

…ality_map on epoch change) Workers send their boot_epoch_id in ConnectWorkerRequest. The server tracks the last-seen epoch per endpoint and atomically wipes that endpoint's locality_map entries on a new epoch (or on epoch=0 = legacy). This closes the lost-eviction race where a worker restart could leave the server with stale entries pointing at blobs the worker no longer has. Three commits from worktree-agent-a6873bf3 (rebased onto 71a067f): - WIP failing tests + proto field - server wipes locality_map on epoch change - clippy is_some_and review nit Includes manual pb.rs codegen for the new proto field (regen via canonical bazel toolchain queued as TraceMachina#87 follow-up). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…her leak When a worker reconnects with a new boot_epoch BEFORE OLD's disconnect-cleanup runs, OLD's dispatcher.worker_txs[(endpoint, prev_epoch)] and per-(worker, prev_epoch) queues leak forever — OLD's later cleanup hits the ownership-check guard (NEW owns the endpoint) and SKIPS its dispatcher.unregister_worker / unpin_on_disconnect calls. Symmetric to the locality_map wipe directly above; see also TraceMachina#141 (which fixed locality_map for the same race class). The leak is dormant under `small_blob_mirror_enabled=false` (the production default) and becomes an active leak the moment the flag flips. Tracker TraceMachina#174 owns the actual fix (call dispatcher.unregister_worker(endpoint, prev_epoch) + unpin_on_disconnect here while endpoint_state lock is held). This commit only adds the TODO comment so the next reader sees the gap. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…hina#97 PR1 server arm) Implements the server-side state for TraceMachina#97 (BIS chunking + ack): - `broadcast_blobs_in_stable_storage_chunked(digests)`: splits the digest list into `BIS_DIGESTS_PER_CHUNK = 4096` chunks via the shared `chunk_iter` helper from TraceMachina#98, dispatches each chunk as `Update::ChunkedMessage(BlobsInStableStorageChunk)` with a unique per-broadcast `broadcast_id`, and adds the chunks to the per-worker resend buffer keyed by `cas_endpoint`. Direct-merge per TraceMachina#98: no per-broadcast worker-side buffer; chunks unpin immediately. - `replay_bis_chunks_to_worker(cas_endpoint, tx)`: called from `add_worker` whenever a worker (re)connects. Replays every chunk still in the per-endpoint resend buffer to the new tx. Same-epoch reconnects automatically pick up unacked chunks; epoch changes are handled below. - `clear_bis_resend_buffer_for_endpoint(cas_endpoint)`: drops every buffered chunk for an endpoint. Called by `worker_api_server` on `boot_epoch_id` change (a fresh worker process means the pin state died with the old process — replaying unpins for blobs that no longer exist would just waste memory). - `bis_ack_received(worker_id, broadcast_id, sequence)`: drops the matching chunk from the per-endpoint resend buffer. Idempotent. When the buffer hits zero, the per-endpoint entry is removed entirely (so a long-lived worker doesn't accumulate empty buffers). State lives on `ApiWorkerSchedulerImpl`: - `bis_resend_buffers: HashMap<String, BisResendBuffer>` — per-endpoint. - `next_bis_broadcast_id: u64` — monotonic allocator, reset at process start (workers tolerate the reset because broadcast_ids are not stable across server restarts; see proto doc-comment). Trait impl: - `WorkerScheduler::bis_ack_received` — both the trait method and the forwarding `ApiWorkerScheduler` / `SimpleScheduler` impls (already added in the worker-arm commit; this commit fills them in). Resend-buffer keying choice (cas_endpoint vs worker_id): the worker_id is regenerated on every connect (see `inner_connect_worker`), so it cannot serve as a stable identity across reconnects. `cas_endpoint` is the worker's own identity; pairing it with `boot_epoch_id` (per TraceMachina#141) distinguishes a fresh process restart from a transient reconnect, and the boot_epoch wipe path clears the buffer atomically with the locality_map wipe. Tests (`#[cfg(test)] mod tests` in the same file): - `bis_chunked_ack_acknowledged`: 100K digests → ~25 chunks; every chunk lands in the buffer; acking each drains the buffer to zero. - `bis_chunked_ack_lost_resend`: 10K digests → ~3 chunks; ack only half; reconnect with same endpoint replays exactly the unacked half (verified by sequence equality). - `bis_chunked_boot_epoch_change_clears_buffer`: after `clear_bis_resend_buffer_for_endpoint`, reconnect receives zero replayed chunks. - `bis_chunked_full_ack_drops_endpoint_entry`: full-ack removes the per-endpoint HashMap entry entirely. TDD verification: - Mutation 1 (suppress `buf.add(chunk.clone())`) — both the buffer count assertion and the lost_resend partial-ack assertion fail with specific messages. Reverted. - Mutation 2 (suppress `replay_bis_chunks_to_worker` call from `add_worker`) — `bis_chunked_ack_lost_resend` fails with the specific "reconnect must replay" assertion. Reverted. Production wiring NOT in this commit: - `worker_api_server` boot-epoch-change clear_buffer call. - `nativelink.rs` switch from `broadcast_blobs_in_stable_storage` → `_chunked` in the BIS broadcast loop. - TLA+ spec. Those land next.

…ent leak When a worker reconnects with a NEW boot_epoch BEFORE the OLD's disconnect-cleanup runs, the OLD epoch's `dispatcher.worker_txs[(endpoint, OLD_epoch)]` and per-(worker, OLD_epoch) queues leak forever — OLD's later cleanup hits the ownership-check guard (NEW owns the endpoint) and SKIPS its `dispatcher.unregister_worker` / `dispatcher.unpin_on_disconnect` calls. Symmetric to the locality_map wipe (TraceMachina#141) that the same boot-epoch wipe block already performs. Currently dormant in production behind `small_blob_mirror_enabled=false`; becomes an active leak the moment the flag flips. Fix: while `endpoint_state` lock is held in the wipe block, also call `dispatcher.unregister_worker(endpoint, prev_epoch)` and `dispatcher.unpin_on_disconnect(endpoint, prev_epoch)`. The lock guarantees the OLD disconnect task observes a fully wiped dispatcher state OR an unwiped one, never a partial state that could lose entries from EITHER epoch. Tests (production composition, both directions per asymmetric- contract-coverage rule): - `boot_epoch_wipe_clears_dispatcher_state` (under-action): seeds worker_tx + queue + pin entries for (E, OLD), reconnects with (E, NEW), asserts (E, OLD) state is cleared. Wrapped in 5 s `tokio::time::timeout` deadlock detector. - `boot_epoch_wipe_does_not_clear_new_epoch_state` (over-action): reconnects with (E, NEW), asserts (E, NEW) state survives. Guards against an over-broad regression that nukes both epochs. Adds two diagnostic accessors on SmallBlobDispatcher: `has_worker_tx(endpoint, boot_epoch_id)` and `queue_count_for_worker(endpoint, boot_epoch_id)` — used only by tests; production callers should rely on send-error results. Mutation step verified: commenting out the new `unregister_worker` + `unpin_on_disconnect` calls makes `boot_epoch_wipe_clears_dispatcher_state` panic with "TraceMachina#174 leak: boot-epoch wipe failed to call unregister_worker". Sibling-bug audit (read-only — out of scope for this PR): `api_worker_scheduler.rs` has the same race class on `endpoint_to_worker` and `remove_prefetch_for_endpoint`, both keyed by endpoint with no boot-epoch ownership check. Owned by TraceMachina#97 fixup worktree. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…hina#279 sub-item 3, sibling-of-TraceMachina#141) worker_api_server.rs:531 calls ac_pin_registry.wipe_endpoint(...) on a worker reconnect that flips boot_epoch_id. NO production-composition test asserted this wiring works on the boot-epoch flip path: the existing worker_disconnect_wipes_ac_registry covers ONLY the disconnect cleanup at :912. A regression that drops the :531 call would ship silently — exactly the bug class TraceMachina#141 fixed for CAS locality_map. Three new tests, all using the same multi-connect harness pattern as worker_api_server_test.rs::boot_epoch_*_locality_entries_test: 1. boot_epoch_different_wipes_ac_registry (under-action): hold the OLD stream alive across the reconnect so the wipe MUST come from the registration path, not from the disconnect cleanup. Asserts within tokio::time::timeout(5s) — bespoke message "boot-epoch wipe MUST clear AC pin registry — sibling-of-TraceMachina#141 regression". 2. boot_epoch_wipe_of_endpoint_a_does_not_clear_endpoint_b (over-action): reconnecting endpoint A with a new boot_epoch_id MUST NOT collateral-wipe endpoint B's pins. Bespoke "boot-epoch wipe of A MUST NOT clear B — over-action: cross-endpoint wipe leaked". 3. boot_epoch_wipe_does_not_block_repopulation: after the wipe, a fresh BlobsAvailable on the new connection MUST re-populate the registry — confirms the wipe didn't break re-registration. mutation step (ac_pin_registry.wipe_endpoint at worker_api_server.rs:531 commented out): all three tests RED-fail with the bespoke messages. Restored: all 8 tests in this file PASS. build: cargo build --release --bin nativelink --features quic,pprof,chunked_fast_slow exit 0 (6m 44s). test: cargo test --release -p nativelink-service --test worker_api_ac_pin_registry_test exit 0 (8/8 pass in 0.25s post-restore).

…raceMachina#2 ordering (TraceMachina#279 reviewer fixups) Address three reviewer findings on commit ee0ca1ea (the boot-epoch wipe regression suite): 1. **code-reviewer FIXES-REQUIRED — test 3 race** `boot_epoch_wipe_does_not_block_repopulation` dropped tx1/stream1 BEFORE the reconnect, allowing the disconnect-cleanup at `worker_api_server.rs:912` to race the registration-path wipe at `:531`. If `:912` ran first, it called `state.remove(&endpoint)` — the reconnect's `:531` then saw `prev = None`, `needs_wipe = false`, and the `:531` wipe DID NOT FIRE. Under that interleaving, commenting out `:531` would still leave the test green via the `:912`-already-cleared path. Fix: hold tx1/stream1 alive across the reconnect, mirroring tests TraceMachina#1 and TraceMachina#2's mechanic. The wipe MUST now come from `:531` exclusively. 2. **testing-czar Important TraceMachina#1 — test 2 assertion ordering** `boot_epoch_wipe_of_endpoint_a_does_not_clear_endpoint_b` asserted A drained BEFORE B survives. Under the "wipe missing" mutation the under-action half (A drained) timed out FIRST and the over-action half (B survives) was never reached — a future reader running only that mutation would conclude the over-action contract is redundant with `boot_epoch_different_wipes_ac_registry`. Fix: invert the order so B-survives is asserted first. Both halves now exercise their respective contracts under their own mutations: under-action mutation ("comment out wipe call") trips A's drain timeout; over-action mutation ("wipe ALL endpoints") trips B's survival assertion. 3. **testing-czar Important TraceMachina#2 — test 3 polling predicate** `send_ac_blobs_and_wait` polls for `len() == expected`. With the OLD stream alive (per fix TraceMachina#1) the registry holds d_old (len=1) until the wipe fires; under the wipe-missing mutation the registry would hold d_old=1 while the d_new tick is in flight, and a `len() == 1` predicate would silently exit with `snap = [d_old]` then catch the bug only via the `assert_eq!(snap[0].1, d_new)` (a generic diff, not a bespoke message). Fix: in test TraceMachina#3 send the post-reconnect `BlobsAvailable` directly (bypassing `send_ac_blobs_and_wait`'s ambiguous predicate) and poll for `d_new`'s PRESENCE specifically. The bespoke message is now reachable on every mutation interleaving. Mutation step verified (post-fixup): * `worker_api_server.rs:531` commented out → all 3 boot-epoch tests RED-fail with bespoke messages: - `boot_epoch_different_wipes_ac_registry`: "boot-epoch wipe MUST clear AC pin registry — sibling-of-TraceMachina#141 regression: Elapsed(())" - `boot_epoch_wipe_of_endpoint_a_does_not_clear_endpoint_b`: "boot-epoch wipe of A MUST clear A's AC pin entries — under-action half: Elapsed(())" - `boot_epoch_wipe_does_not_block_repopulation`: "post-wipe registry MUST hold ONLY d_new — stale (pre-wipe) digest leaked through wipe" * `:531` replaced with "wipe ALL endpoints" loop → only test 2's over-action half red-fails: "boot-epoch wipe of A MUST NOT clear B — over-action: cross-endpoint wipe leaked" * Restored: 8/8 green in 0.25s. Module-level `//!` doc-block updated to enumerate tests 5–7 with the asymmetric-coverage framing and an explicit "wipe ALL endpoints" mutation entry that surfaces the over-action contract. Reviewer reports addressed: - .claude/reviews/279-drain-hardening-ee0ca1ea/code-review.md (MAJOR) - .claude/reviews/279-drain-hardening-ee0ca1ea/testing.md (Important TraceMachina#1 + TraceMachina#2) - .claude/reviews/279-drain-hardening-ee0ca1ea/red-team.md (Q5/Q7) - .claude/reviews/279-drain-hardening-ee0ca1ea/distributed-systems.md (MINOR-1 mutation-step verification) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…epoch reconnects Adds a sliding-window flap detector to `WorkerApiServer::inner_connect_worker` that fires `warn!("worker reconnect storm ...")` and bumps the new `worker_flap_warns_total` counter when the same `cas_endpoint` undergoes `FLAP_THRESHOLD = 3` boot_epoch changes within `FLAP_WINDOW = 300s`. `FLAP_COOLDOWN = 60s` suppresses re-warn so a sustained flap doesn't drown the log. Mechanism: extends `EndpointState` (`worker_api_server.rs:223-260`) with `epoch_changes: VecDeque<Duration>` + `last_flap_warn_at: Option<Duration>`. Both fields share the pre-existing `endpoint_state: parking_lot::Mutex<...>` already held across the `TraceMachina#141` wipe path (`:520-660`) — zero new lock cost. Timestamps come from the server's injectable `now_fn`, not `SystemTime::now()`, so tests drive the cooldown/window logic deterministically. The flap deque only pushes inside the `needs_wipe == true` branch, so same-epoch reconnects (transient stream drops) don't inflate the count. Window-eviction at push-time keeps the deque bounded at FLAP_THRESHOLD entries per endpoint for a flapping worker (UNBOUNDED-OK: not attacker-controlled — `TraceMachina#216` build_sha allowlist gates connects). Counter rides on `WorkerApiMetrics` (already wired via `#[metric(group = "worker_api")]` at `:139-140`), so it's scrapable today — unlike `TraceMachina#386` which can only lean on `warn!` until `TraceMachina#380` lands. The `warn!` is still the load-bearing operator signal in journald. Tests (`worker_api_server_test.rs`): - `worker_flap_detection_fires_warn_and_respects_cooldown_test`: 3 flips within window fire exactly one warn; 4th within cooldown does NOT re-warn; flip past cooldown re-warns. Mutation step: commenting out the warn+counter red-fails with bespoke "worker-flap warn missing — operator-blind to whole-process restart loop" assertion (verified). - `worker_flap_window_evicts_stale_entries_test`: 10 flips at 400s intervals (beyond FLAP_WINDOW=300s) never accumulate to threshold — sliding-window eviction. - `worker_flap_same_epoch_reconnects_dont_count_test`: 5 reconnects at the SAME epoch never count as flaps. All 3 new tests pass; 6 existing `TraceMachina#141` boot_epoch tests still green. Audit: `.claude/audits/384-exec-log-2026-05-10/server-worker-oom-detection.md` Q2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…y` map so it survives disconnect-cleanup Addresses BLOCK-FIX-FIRST from distributed-systems-reviewer review at `.claude/reviews/387-first-pass/distributed-systems-reviewer.md`. The previous commit (`bf809900`) folded the flap deque + cooldown into `EndpointState`. That made the detector silent in the exact scenario it was built to surface: production OOM-loop pattern is `worker process dies → kernel RST → server's per-connection background task observes the stream break in ms → disconnect-cleanup runs `state.remove(...)` at :1108-1166 while the worker takes seconds to relaunch`. After cleanup, the next connect saw `prev = None` → `needs_wipe = false` → flap-deque push gated out. The detector only fired on the rare race-flap where the new connection arrived BEFORE the old's cleanup — the OPPOSITE of the OOM-loop pattern. Fix shape: SEPARATE `parking_lot::Mutex<HashMap<String, FlapHistory>>` field on `WorkerApiServer` named `flap_history`. Holds `last_seen_epoch + epoch_changes deque + last_flap_warn_at`. NOT touched by disconnect-cleanup (which only takes `endpoint_state`). Lock-order rule: connect path takes `endpoint_state` first (for the TraceMachina#141 wipe + insert), drops it via `drop(state)`, THEN takes `flap_history`. Disconnect path takes only `endpoint_state` and never touches `flap_history`. Two separate locks; never held simultaneously. Picked two-Mutex shape (DSR's recommended option (a)) over single-Mutex covering both maps because: - disconnect-cleanup must NOT touch `flap_history` (the whole point); co-locating would force a deliberate skip and re-introduce the foot-gun - lock contention is ignorable at connect frequency (one per worker reconnect, fleet-wide) Detector now fires when `flap_history[endpoint].last_seen_epoch != new_boot_epoch || new_boot_epoch == 0`, regardless of whether `endpoint_state` has an entry. First-ever connect records `last_seen_epoch` without pushing (no prior epoch to differ from). Invariant: when the same `cas_endpoint` undergoes `FLAP_THRESHOLD` boot_epoch changes within `FLAP_WINDOW`, exactly one `worker_flap_warns_total` increment fires WHETHER OR NOT disconnect-cleanup runs between the changes. Mechanism that violates it (now removed): flap state on `EndpointState` wiped by `state.remove(&cas_endpoint)` at `:1160`. Mechanism that re-establishes it: `flap_history` is a separate map never touched by disconnect-cleanup. Composite invariants: TraceMachina#141 boot_epoch wipe (still gated on `endpoint_state.boot_epoch`, unchanged); TraceMachina#174 dispatcher state wipe (still gated on `endpoint_state`, unchanged); flap detection now orthogonal to both. Test that proves it: `worker_flap_detection_survives_disconnect_cleanup_test`. Tests: - New: `worker_flap_detection_survives_disconnect_cleanup_test` drives disconnect-cleanup to completion (polls `endpoint_state_is_empty_for_testing` under `tokio::time::timeout(5s)` deadlock-detector) BEFORE every connect, then asserts the flap counter still reaches threshold across 4 fully-cleaned-up flips. Bespoke assertion message: "flap detector wiped by disconnect cleanup — operator-blind to OOM-loop pattern". - Mutation step verified: uncomment the `if !_prev_was_some { self.flap_history.lock().remove(...); }` block at `:768-770` to simulate the original BLOCK-FIX-FIRST defect. The new test red-fails with the bespoke message above; the prior 3 flap tests still pass (they exercise the race-flap path where `_prev_was_some = true`). - All 6 prior TraceMachina#141 boot_epoch tests + all 29 tests in `worker_api_server_test.rs` still pass. Doc-comment on `FlapHistory` names the lifecycle explicitly (per DSR MAJOR-2 follow-up). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ted reload (server durability bundle, directive-3) Adds graceful-shutdown Phase 3.5 (persist the digest->worker locality map to NVMe) and startup reload behind a Bazel-REAPI readiness gate, on top of directive-1's flush + directive-2's worker-pull+quiesce. After a restart the server reloads the locality INDEX so it can peer-fetch a worker-only blob the instant a worker reconnects -- before that worker's first full BlobsAvailable snapshot -- closing cause-(b) of the 2026-06-23 NotFound storm for the blobs directive-2's byte-pull does not localize (pull-skipped, LRU-evicted post-pull, small/Redis). Design: .claude/audits/persist-locality-map-design-2026-06-23.md. == Mechanism == - PERSIST (Phase 3.5, src/bin/nativelink.rs SIGTERM handler, AFTER the dir-2 pull, BEFORE eviction): LocalityPersister::persist_to_disk snapshots the locality map JOINED with endpoint_state (for boot_epoch) into a PersistedLocalityMap and writes /srv/nativelink/locality-map.bin via tokio::fs::write(tmp) + tokio::fs::rename -- ATOMIC RENAME, ZERO fsync. The bincode encode runs in spawn_blocking (never blocks a tokio worker). On-disk shape: raw 6-byte frame (u32 magic LE + u16 version LE) validated BEFORE any bincode decode (so a corrupt header is rejected up front, not via a stray length-prefix giant-alloc), then a bincode payload (persisted_at + entries). Snapshot is CAPPED AT digest_count() -- index only, blob bytes never collected, dropped after write, shutdown-local (not a network path). - READINESS GATE (bazel_ready_tx watch::channel, the inverse of dir-2's accept_stop_tx): every listener dir-2 flags quiesce_on_shutdown (the public Bazel-REAPI listeners) binds its socket immediately but does NOT accept() until the startup reload flips the gate. The worker_api (:50061) + worker-CAS (:50071/:50072) listeners (NOT quiesce-flagged) stay UNGATED so workers reconnect + reconcile DURING the reload. Reuses dir-2's exact listener discriminator (quiesce_this_listener), not the design's old worker_api.is_none() heuristic. FAILS OPEN on any reload error + a 30s watchdog backstop flips the gate regardless -- a reload failure can never wedge Bazel serving. - RECONCILIATION (the crux): reload_from_disk primes BOTH the locality_map AND endpoint_state (with the persisted boot_epoch + a sentinel owner) so the EXISTING TraceMachina#141 connect-path wipe (worker_api_server.rs:763) does the reconciliation for free -- same boot_epoch on reconnect keeps the entries until the first full snapshot; different boot_epoch (rebooted worker) fires remove_endpoint and drops the stale entries before any BlobsAvailable. Never-reconnect -> bounded-TTL sweep (sweep_unconfirmed, default 600s grace) drops endpoints still carrying the sentinel. Stale-but-up -> existing should_evict_locality_on_peer_error covers it (no new code). == Invariant-walk == Invariant being violated: after a server restart, a CAS blob that exists only on a worker is not FINDABLE by the server (the digest->worker index is empty) until that worker reconnects AND sends its first full BlobsAvailable snapshot, causing a NotFound window even for blobs the server could peer-fetch the instant the worker's TCP connection is up. Mechanism that violates it: the locality map (blob_locality_map.rs) is in-memory only; on SIGTERM workers are evicted (worker_api_server.rs:768 remove_endpoint) wiping the map, which is never persisted; at startup it is constructed empty (new_shared_blob_locality_map). Mechanism that re-establishes it: Phase 3.5 (src/bin/nativelink.rs) snapshots endpoint_blobs JOINED with endpoint_state and writes (cas_endpoint, boot_epoch) -> digests to /srv/nativelink/locality-map.bin (atomic rename, no fsync) BEFORE eviction; startup reload_from_disk repopulates the live BlobLocalityMap and primes endpoint_state with the persisted boot_epochs, gating Bazel-REAPI accept (bazel_ready_tx) until reload completes (worker-API accept ungated); the existing TraceMachina#141 wipe reconciles each reloaded endpoint on reconnect; a bounded-TTL sweep drops never-reconnect endpoints; should_evict_locality_on_peer_error drops reloaded-but-stale entries on first failed peer-fetch. Composite invariants this fix interacts with: (1) TraceMachina#141 boot_epoch wipe (worker_api_server.rs:763) -- REUSED by priming endpoint_state; missing the priming makes the mismatch-wipe silently no-op (the trap, guarded by reloaded_entry_wiped_on_boot_epoch_mismatch_via_primed_endpoint_state). (2) is_full_snapshot wipe-and-replace -- bounds how long a reloaded entry lives after reconnect (until the first full snapshot); self-heals. (3) should_evict_locality_on_peer_error -- reactive staleness path; reloaded entries flow through it unchanged. (4) directive-2 shutdown ordering -- Phase 3.5 runs after the pull, before eviction; both before the evict-before-drain d204e1b order (preserved). (5) accept_stop_tx quiesce gate -- dir-3 adds a SEPARATE inverse-phased bazel_ready_tx; independent watch channels (startup-readiness vs shutdown-quiesce), both can be active at shutdown without conflict. (6) >=2-replica / BIS -- untouched; dir-3 is read-path findability only, never marks a blob stable, never acks BIS, never unpins. Test that proves the fix: reloaded_entry_is_findable_without_rebackfill (a reloaded entry resolves lookup_workers with zero re-backfill) + reloaded_entry_wiped_on_boot_epoch_mismatch_via_primed_endpoint_state (the priming-trap guard). == Behavior changes == This is an ARCHITECTURAL change (new startup ordering + a readiness gate on the Bazel-REAPI listeners + a new shutdown phase). Runtime behavior changes against their production composition: - src/bin/nativelink.rs startup: Bazel-REAPI listeners (quiesce_on_shutdown: true; production :50051) now BIND immediately but do not accept() until the locality reload flips bazel_ready_tx (or the 30s watchdog fires). Worker-API (:50061) + worker-CAS (:50071/:50072) listeners (not quiesce-flagged) are UNGATED -- behavior unchanged, they accept immediately. In production the reload is a fast local NVMe read so the gate opens near-instantly; on a fresh boot with no file it fails open immediately. - src/bin/nativelink.rs SIGTERM: new Phase 3.5 persist runs between the pull and eviction. Adds one atomic file write to the shutdown path; on a config with no worker_api entry the handle is None and the phase is a logged no-op. - worker_api_server.rs: NEW LocalityPersister handle + reload primes endpoint_state with a sentinel owner_worker_id "__reloaded_unconfirmed__". The TraceMachina#141 connect-path wipe is UNCHANGED -- it now simply observes a primed prev on the first reconnect after a reload-restart (previously prev was None). No change to any path for a fleet that never restarts with a persist file (no file -> empty reload -> identical to today). - blob_locality_map.rs: NEW persist/reload serialization types (PersistedLocalityMap, PersistedEndpoint, ReloadedLocalitySummary) + snapshot_endpoint_blobs(). Pure additions; no change to any existing register/lookup/evict/remove behavior. - nativelink-util/Cargo.toml: adds bincode (already a workspace dep in nativelink-store) for the persist serialization. == Constraints == - NO fsync: atomic rename only (torn-write avoidance, not durability); durability = ZFS txg + best-effort worker re-announce. Audited: zero sync_all/sync_data/File::sync_* in the write path. - Never blocks a tokio worker: bincode encode/decode in spawn_blocking; file I/O via tokio::fs. - Snapshot bounded (CAPPED AT digest_count()) + annotated; not a network path. == Drift reported to parent == - DigestInfo serializes as a "{hex}-{size}" STRING (custom serde impl, common.rs:220), so on-disk cost is ~73 B/digest (string) not the design's 40 B/binary estimate -- still bounded by digest_count(), just a larger constant. Functionally correct (round-trip test passes). - The bin-level accept-gate (bazel_ready_tx) is verified by code-read + the full bin build, not an automated test (it lives inside inner_main's listener spawn). The reload contract the gate waits on IS fully tested + mutation- proven; the gate's "skip reload -> empty map -> NotFound" regression is proven via the reload-skip mutation. - PRE-EXISTING on the base (NOT directive-3): several sibling test files reference removed proto fields swap_pressured/swap_pressure_rate_per_sec (compile-fail in nativelink-service tests); and `cargo clippy` fails in the untouched nativelink-error crate (std-vs-core import lint, lib.rs:712-734). Neither blocks directive-3: the full bin built clean and the directive-3 test target builds + passes in isolation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

…eep + reload clobber guard + bounded decode (Tier-3 convergent FIXES-REQUIRED) Addresses the Tier-3 cadre's convergent must-fix findings on the server durability bundle (reviews at .claude/reviews/server-bundle-7a8b1816/). A) Pull confirms DURABILITY, not has==Some (red-team + code-reviewer): the worker-pull lands blobs in the FSS fast tier + spawns the slow write async; has_with_results returns Some from in_flight_slow_writes (NOT the durable tier). Added Phase 3.6 in src/bin/nativelink.rs: re-run the unbounded flush_slow_writes after the pull, before persist/eviction, so pull-landed blobs are durable on the slow tier. Bounded+idempotent: the flush slow.has-skips already-persisted blobs (CAS content-addressed). B) Scheduled the never-reconnect sweep + honored grace (distsys MAJOR-1 + code-reviewer): one-shot sweep task in inner_main fires at LOCALITY_PERSIST_RECONNECT_GRACE_SECS (600s) after boot; sweep_unconfirmed now gates on a shared reload-baseline Instant (no-op until reload is >= grace old) instead of `let _ = grace;`. C) Reload state.insert clobber (distsys MAJOR-2): reload priming is now `entry().or_insert_with` so a worker that reconnected DURING the reload window keeps its live owner (never re-stamped to the sentinel → never swept). register_blobs stays additive; TraceMachina#141 priming-trap guard preserved for endpoints that did NOT reconnect (absent → or_insert primes them). D) Bounded the bincode body decode (security MEDIUM): added .with_limit::<4GiB>() on the reload decode as cumulative-allocation defense-in-depth. DRIFT: the security review's "Vec::with_capacity(u64::MAX) abort" premise does NOT hold on the bincode-2.0.1-serde path (serde's cautious cap + slice-bounded decode → UnexpectedEnd, not abort; empirically confirmed). The load-bearing fail-open is the Err→Ok(empty) mapping, which the new test pins. E) Removed the PersistedPayload twin (code-simplifier S1): derive Serialize/Deserialize directly on PersistedLocalityMap; drops the full entries.clone() on the SIGTERM persist path + the silent-drift hazard. Wire format byte-identical (same two fields, same order, same standard()); the raw magic+version frame is unchanged. F) Doc fixes: ~40 B/digest → ~72 B/pair (string-encoded DigestInfo), CAPPED AT pair_count at the 3 annotation sites; bazel_reapi_quiesce module doc now describes the explicit quiesce_on_shutdown flag (set ONLY on :50051), not the REJECTED worker_api.is_none() heuristic, and clarifies :50071/:50072 are NOT quiesced. Invariant being violated: an acked/pulled large CAS blob must be durable on the slow tier before the process exits (small class already Redis-durable via smallack); the pull declared "pulled" on volatile fast-tier presence. Mechanism that violates it: src/bin/nativelink.rs pull phase + fast_slow_store has_with_results reading in_flight_slow_writes — has==Some is satisfied by the volatile in-flight/fast tier, and the only flush ran BEFORE the pull. Mechanism that re-establishes it after the fix: src/bin/nativelink.rs Phase 3.6 re-runs StoreManager::flush_slow_writes (unbounded Phase-2 fast→slow drain) after the pull and before persist/eviction. Composite invariants this fix interacts with: ≥2-replica (unchanged); no fsync (the post-pull flush + persist stay no-fsync); slow-writes-async business rule (UNCHANGED — the flush reads fast → writes slow at shutdown, not a per-write sync); TraceMachina#141 boot_epoch reconciliation priming (preserved by or_insert); never-reconnect TTL sweep (now scheduled + grace-gated); locality_map eventual consistency (reload + sweep + TraceMachina#141 wipe). Test that proves the fix re-establishes the invariant: post_pull_flush_makes_pulled_blob_durable_on_slow_tier (item A); sweep_honors_grace_no_sweep_before_grace_elapses (B); reload_does_not_clobber_live_reconnected_endpoint (C); reload_valid_frame_hostile_body_fails_open (D). Behavior changes (each enumerated against its production composition): - SIGTERM sequence: ADDS Phase 3.6 (post-pull flush_slow_writes) between the worker-pull (Phase 3) and the locality persist (Phase 3.5). On a server with a worker_api entry + a FastSlowStore CAS, pull-landed blobs are now driven to the slow tier (Redis/FilesystemStore) before persist/eviction/exit. No-op when STORE_MANAGER is unset. Unbounded (same TimeoutStopSec=infinity dependency as Phase 2). - Startup: ADDS a one-shot sweep task that fires once at 600s after boot, removing reloaded locality entries for endpoints whose worker never reconnected. No-op when there is no worker_api/locality_map, or when no reload ran (no baseline / no sentinel entries). - Reload priming: endpoint_state insert is now conditional (or_insert). For an endpoint with NO live connection (the common reload case) behavior is identical (sentinel primed). For an endpoint that reconnected during the reload window, the live owner is now PRESERVED (was: clobbered with the sentinel) — strictly safer (the live worker is no longer at risk of a spurious sweep / stale boot_epoch). - Reload decode: now bounded by a 4 GiB limit. For every realistic file (≤ ~200 MB) behavior is identical. A pathological > 4 GiB cumulative decode now returns LimitExceeded → fail-open empty map (was: unbounded decode). - Persist serialize: drops one full entries.clone() (perf/memory only; wire bytes identical). No runtime-behavior change to the persisted file. - bazel_reapi_quiesce + ~40B doc edits: documentation only; no behavior change. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

… grace-after-boot distsys MAJOR-1 (re-opened in refix-distsys.md): the never-reconnect TTL sweep was SCHEDULED but its single one-shot fire ALWAYS swept 0. The sweep task slept `grace` (600s) from BOOT, but `reload_baseline` is stamped at the END of `reload_from_disk` (after the async read+decode = boot + reload_dur). At fire `baseline.elapsed() = grace - reload_dur < grace`, and the strict `<` time-gate in `sweep_unconfirmed` returned 0 every boot. The sentinel producer shipped; its sole consumer was a guaranteed no-op → reloaded never-reconnect locality entries leaked forever (degrades peer-fetch; bounded best-effort index per invariant TraceMachina#6, NOT durability loss). Fix: the sweep task now awaits the `bazel_ready` readiness gate (which the reload task flips `true` ONLY after `reload_from_disk` returns, i.e. after the baseline is stamped) BEFORE sleeping `grace`, so `baseline.elapsed() >= grace` holds at the single fire. The await->sleep->sweep sequence is extracted into `LocalityPersister::run_never_reconnect_sweep(reload_done, grace)` so the bin path and the production-timing test exercise the SAME scheduling logic. `reload_baseline` switched from `std::time::Instant` to `tokio::time::Instant` (3 sites, grace-gate only) so the sweep's `sleep(grace)` and the gate's `baseline.elapsed()` read ONE clock — required for the production-timing test to virtualize the 600s window under `tokio::time::pause()`. Production (never paused) is identical: `tokio::time::Instant` delegates to the real monotonic clock. Invariant being violated: a reloaded endpoint whose worker never reconnects within `grace` must have its sentinel locality entries swept (design §4.4 — no "hold-forever"). Mechanism that violates it: src/bin/nativelink.rs sweep task slept `grace` from boot while sweep_unconfirmed's gate (worker_api_server.rs:1806 region) measures `grace` from the strictly-later baseline -> elapsed < grace -> 0. Mechanism that re-establishes it: nativelink.rs sweep task awaits the bazel_ready gate (flips after reload_from_disk returns + stamps baseline) then run_never_reconnect_sweep sleeps grace -> baseline.elapsed() >= grace at fire. Composite invariants this fix interacts with: (a) item C or_insert guard (sweep must not drop a live reconnected endpoint — unchanged, still gated on the sentinel identity); (b) bazel_ready fail-open (gate flips even on reload error/timeout — sweep then no-ops on a None/late baseline, correct); (c) TraceMachina#141 boot-epoch wipe (sentinel identity, orthogonal to the timing gate). Test that proves the fix re-establishes the invariant: never_reconnect_sweep_fires_grace_after_baseline_not_boot. Behavior changes: - nativelink (bin) sweep task: prior -> the one-shot sweep slept grace from boot and was a guaranteed no-op (swept 0 every boot); new -> awaits the reload-done readiness gate, then sleeps grace, then sweeps once -> swept N (the actual never-reconnect endpoints). Verified safe in compositions: the production worker_api entry (locality_persister Some), where the reload task + Bazel-REAPI listeners + this sweep task share `bazel_ready_tx`; a no-worker_api entry yields None persister -> task returns early (no-op), unchanged. - LocalityPersister.reload_baseline / WorkerApiServer.locality_reload_baseline: std::time::Instant -> tokio::time::Instant. No runtime behavior change in production (un-paused tokio::time::Instant == real monotonic clock); enables virtual-time testing only. - No async<->sync flip, no buffer moved, no spawn removed, no new network buffer, no fsync/sync-write primitive. The slow-write decoupling and the shutdown sequence are untouched. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

allada requested a review from chrisstaite-menlo July 7, 2023 04:55

allada force-pushed the scheduler-config-supports-different-impls branch from 96ebec5 to 6228e48 Compare July 7, 2023 05:15

[BreakingChange] Scheduler config now supports multiple impls

8fb40d0

This is a step in the direction to support different scheduler impls. To upgrade, you'll need to add a "local" property to the "scheduler" configs. See example files that where changed in this commit. towards: #140

allada force-pushed the scheduler-config-supports-different-impls branch from 6228e48 to 8fb40d0 Compare July 7, 2023 05:37

chrisstaite-menlo reviewed Jul 7, 2023

View reviewed changes

chrisstaite-menlo approved these changes Jul 7, 2023

View reviewed changes

allada merged commit 384f14e into master Jul 7, 2023

allada deleted the scheduler-config-supports-different-impls branch July 7, 2023 18:08

rejuvenile added a commit to rejuvenile/nativelink that referenced this pull request Aug 19, 2026

WIP TraceMachina#141 boot_epoch_id one-way: failing tests + proto field

6b75750

rejuvenile added a commit to rejuvenile/nativelink that referenced this pull request Aug 19, 2026

worker_api_server: clippy is_some_and instead of map_or (TraceMachina…

7637cfb

…#141 review nit)

Navigation Menu

Repository navigation

Scheduler config now supports multiple impls - #141

Scheduler config now supports multiple impls#141
allada merged 1 commit into
masterfrom
scheduler-config-supports-different-impls

allada commented Jul 7, 2023 •

edited

Loading

Uh oh!

chrisstaite-menlo left a comment

Uh oh!

chrisstaite-menlo left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

allada commented Jul 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chrisstaite-menlo left a comment

Choose a reason for hiding this comment

Uh oh!

chrisstaite-menlo left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

allada commented Jul 7, 2023 •

edited

Loading