Distributed hardening: soak harness + coordinator recovery + backpressure (M5 E2/E3)#9
Merged
Conversation
…y (PRD E2) Adds an in-process soak harness (conduit-distributed/tests/soak.rs) that drives the real Coordinator under continuous load while crashing and replacing workers mid-flight, asserting the §8.2 SLOs: exactly-once completion, drain-to-zero, and bounded state, and reporting throughput + p50/p99 latency. Fast fixed burst by default (a real CI regression, ~1s); CONDUIT_SOAK_SECS=<n> runs a continuous soak (nightly can set hours). The harness surfaced a real coordinator gap: a task dispatched to a worker in the window between its disconnect and handle_worker_disconnect computing the orphan list is stranded in `inflight` for a removed worker with no recovery path. Fixed with a reconciliation sweep in the periodic health-check that requeues any in-flight task whose worker is no longer connected (idempotent, safe every tick). Covered by a focused unit test plus the soak harness itself; verified exactly-once holds over 4600+ tasks and 76 crash/replace cycles with zero duplicates or losses.
…store (PRD E3)
Persists the coordinator's authoritative in-flight state so a restarted
coordinator reconstructs it instead of silently orphaning every running
task.
- New AssignmentStore trait with two backends in conduit-distributed:
InMemoryAssignmentStore (default — original non-durable behaviour) and
RocksAssignmentStore (RocksDB, keyed by assignment_id). Both pass one
conformance suite; the Rocks backend also has a reopen-survival test.
- Coordinator records each dispatched assignment (both the immediate and
drained-from-queue paths) and removes it on terminal result. New
Coordinator::with_store + async recover(): recover() loads persisted
assignments and re-queues them as pending (the old workers are gone;
they re-register and pick the work up). An is_recovering() gate queues
new submissions during recovery so recovered work keeps its place.
- DistributedExecutor::with_persistence(config, path) opens the durable
store at {state_dir}/coordinator_assignments and recovers on startup;
new() stays in-memory and unchanged.
- Tests: a full restart scenario (3 in-flight tasks survive a coordinator
drop+reopen and re-dispatch to a fresh worker) and a
completed-tasks-are-not-recovered durability check.
80 distributed tests pass; the soak harness still holds exactly-once
with persistence recording active on every dispatch.
submit_task previously logged an error and silently DROPPED a task when the queue was full — lost work the caller thought succeeded. It now returns a SubmitOutcome (Dispatched | Queued | Rejected): a full queue yields Rejected without accepting the task, so the caller applies backpressure instead of losing it. New CoordinatorConfig.high_water_mark (default 80% of max_queue_size) + Coordinator::is_under_backpressure() let producers throttle proactively before the queue fills. DistributedExecutor::dispatch surfaces the outcome (logging Rejected) and exposes is_under_backpressure(). Two unit tests cover full-queue rejection + the dispatched/backpressure signals. 82 distributed tests pass; soak still holds exactly-once.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Distributed-runtime hardening — M5 E2 and E3 from
docs/PRD_USER_READINESS.md. The soak harness drove out three real reliability gaps in the coordinator; each is fixed and covered. All green: 82 distributed tests,clippy --all-targets -D warningsclean,fmtclean.E2 — soak / chaos harness (+ a bug it found)
conduit-distributed/tests/soak.rsdrives the real coordinator under continuous load while crashing and replacing workers mid-flight, asserting the §8.2 SLOs — exactly-once completion, drain-to-zero, bounded state — and reporting throughput + p50/p99. Fast fixed burst by default (a real CI regression, ~1s);CONDUIT_SOAK_SECS=<n>runs a continuous soak (nightly can set hours). Verified exactly-once over 4,600+ tasks and 76 crash/replace cycles.Bug it surfaced: a task dispatched to a worker in the window between its disconnect and
handle_worker_disconnectcomputing the orphan list was stranded ininflightforever. Fixed with a reconciliation sweep in the periodic health-check that requeues any in-flight task whose worker is no longer connected. Covered by a focused unit test + the soak itself.E3 — coordinator crash recovery
A restarted coordinator previously orphaned every running task. Now it survives a restart:
AssignmentStoretrait, two backends:InMemoryAssignmentStore(default, non-durable — original behaviour) andRocksAssignmentStore(RocksDB). One shared conformance suite + a reopen-survival test.with_store+ asyncrecover()reloads persisted assignments and re-queues them (old workers are gone; they re-register and pick the work up); anis_recovering()gate queues new submissions during recovery so recovered work keeps its place.DistributedExecutor::with_persistence({state_dir}/coordinator_assignments);new()stays in-memory and unchanged.E3 — real backpressure (was drop-on-full)
submit_taskused to log an error and silently drop a task when the queue was full — lost work the caller thought succeeded. It now returnsSubmitOutcome::{Dispatched,Queued,Rejected}: a full queue yieldsRejectedwithout accepting the task, so the caller applies backpressure. Newhigh_water_markconfig +is_under_backpressure()let producers throttle before the queue fills. Two unit tests.Everything composes: the soak still holds exactly-once with persistence recording active on every dispatch.
Remaining M5 (not in this PR): D7/D8 — SQLGlot differential oracle + lineage resolver hardening (
conduit-lineage).🤖 Generated with Claude Code