diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md index 9b8ee00..219af28 100644 --- a/services/flysim/crates/fly-session/README.md +++ b/services/flysim/crates/fly-session/README.md @@ -218,6 +218,22 @@ The durable store is `state`, over the `FLYSESS1` layout the contract crate owns derived from the epoch, so a resumed run's behaviour is compared through `EpochRebase`, which rewrites exactly those and fails on anything it does not recognise. +## Before you add a check to a reply + +Ask which kind of reply it is. Is the far side **reporting what it did**, in which case a +subset or an empty answer is permitted and must be accepted? Or is it **being held to a +requirement**, in which case exactness is the rule and must be enforced? `Worker.Acknowledge` +is the only reply of the first kind in this crate, because `ipc-v1` section 5 explicitly makes +it idempotent -- "Already released/unknown IDs are ignored" -- so a second one legitimately +releases nothing, and the section 6 resolution turns any slow Acknowledge into exactly that +second one. Demanding the whole list back there fenced healthy sessions until +`an_acknowledge_that_releases_nothing_is_not_a_failure` was written. + +The commit and batch checks are the second kind and must stay exact: `commit_all` requires +every agent (`step-v1` section 3 phase D, section 7) and `check_batch` requires every declared +port (`workers-v1` section 3). Loosening those in the name of tolerance is the same mistake +pointing the other way -- they are what make a partial commit and an incomplete batch fail. + ## Where this crate narrows or adds to the contract crate - **Required views.** `WorldObservation::validate_against` checks the views a result carries diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs index ab2e1a1..5e94143 100644 --- a/services/flysim/crates/fly-session/src/agent.rs +++ b/services/flysim/crates/fly-session/src/agent.rs @@ -199,6 +199,9 @@ pub struct AgentFaults { /// Refuse `State.ActivateRestore` after this worker has already staged, so a group meets /// a failure halfway through activation. pub fail_activate_restore: bool, + /// Add this id to every `Worker.Acknowledge` reply, so the caller meets a worker + /// reporting about an id it was never asked about. + pub acknowledge_extra_id: Option, } /// One fake agent worker's configuration. @@ -698,6 +701,10 @@ impl WorkerEndpoint for FakeAgentWorker { self.config.worker_threads as u64 } + fn acknowledge_extra_id(&self) -> Option { + self.config.faults.acknowledge_extra_id.clone() + } + fn methods(&self) -> Vec<&'static str> { vec![ "Agent.Initialize", diff --git a/services/flysim/crates/fly-session/src/cli.rs b/services/flysim/crates/fly-session/src/cli.rs index 603e375..dcb0df7 100644 --- a/services/flysim/crates/fly-session/src/cli.rs +++ b/services/flysim/crates/fly-session/src/cli.rs @@ -213,6 +213,9 @@ fn serve(role: &str, options: &Options) -> Result<(), String> { commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?, fail_stage_restore: options.flag(flags::FAIL_STAGE_RESTORE)?, fail_activate_restore: options.flag(flags::FAIL_ACTIVATE_RESTORE)?, + // A worker process is never asked to misbehave this way: the subset refusal + // is a caller-side check and its test runs the worker in-process. + acknowledge_extra_id: None, }, client_id: client_id.clone(), service: service.clone(), diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 7d68aa1..7683ec4 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -63,6 +63,10 @@ pub struct Injections { pub substituted_published_handle: bool, /// Ask an agent to apply a stimulus kind its published descriptor does not declare. pub undeclared_stimulus: bool, + /// Acknowledge the lifecycle replies twice, which is what the `ipc-v1` section 6 + /// resolution does to any Acknowledge whose first reply outran the probe. The second one + /// legitimately releases nothing, and bootstrap must accept it. + pub duplicate_lifecycle_acknowledge: bool, } /// What an injection produced, for a test to assert on. @@ -346,6 +350,8 @@ pub struct Coordinator { pub resolutions: u64, /// How the last resolution ended, so a test or a supervisor can tell which bound fired. pub last_resolution: Option, + /// How many attempts the last resolution spent. Counted, not inferred from the clock. + pub last_resolution_attempts: u32, /// The caller-side failure-detection budgets of `ipc-v1` section 6. pub deadlines: Deadlines, /// Per-method and critical-path latency samples. Local synthetic timings, never a @@ -435,6 +441,7 @@ impl Coordinator { in_progress_replies: 0, resolutions: 0, last_resolution: None, + last_resolution_attempts: 0, deadlines: Deadlines::default(), metrics: Metrics::default(), blame: None, @@ -966,23 +973,71 @@ impl Coordinator { .push(request_id); } for (worker, ids) in by_worker.into_values() { - let params = AcknowledgeParams { request_ids: ids.clone() }; - let reply = self - .call(&worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[]) - .await?; - let result: AcknowledgeResult = - reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?; - if result.acknowledged.len() != ids.len() { - return Err(self.fail_now( - DomainError::invalid("a worker did not acknowledge every lifecycle reply"), - "acknowledge", - )); + if self.injections.duplicate_lifecycle_acknowledge { + // Release them first, out of sight, so the call this method then makes and + // checks is already the *second* one -- which is the shape the section 6 + // resolution produces when an Acknowledge's first reply outruns the probe, + // and the shape the original defect fenced a healthy session on. Adding a + // second call after the checked one would not reproduce it: the first reply + // is always complete, so a length check on it would pass. + let first = self.acknowledge_replies(&worker, &ids).await?; + if first.len() != ids.len() { + return Err(self.fail_now( + DomainError::invalid("the first Acknowledge did not release everything"), + "acknowledge", + )); + } } + self.acknowledge_replies(&worker, &ids).await?; } self.audit.push("acknowledge.lifecycle".to_owned()); Ok(()) } + /// Releases a worker's retained lifecycle replies, and accepts a short answer. + /// + /// **`ipc-v1` section 5: "Already released/unknown IDs are ignored."** The reply lists what + /// *this* call released, which is not always everything it asked about, and the contract + /// type already holds that list to a subset of the request. So a second Acknowledge of the + /// same ids answers with an empty list by design, and an empty list is success. + /// + /// This matters beyond tidiness. The `ipc-v1` section 6 resolution turns any Acknowledge + /// whose reply is slower than the probe into a second Acknowledge of the same ids, so the + /// short answer is not an edge case -- it is what the contract produces on an ordinarily + /// slow worker. Requiring the whole list back made the contract's own idempotence a failed + /// epoch, which is what + /// `an_acknowledge_that_releases_nothing_is_not_a_failure` guards against. + /// + /// A short list is accepted; a list about something else is not. The worker reports what + /// *it* released, so fewer ids than asked for is success -- but it is still only entitled + /// to report about the ids it was asked about, and an id outside the request is a worker + /// talking about another caller's cache. That half is exact-demanded, and + /// `AcknowledgeResult::validate_against` is what says so. + /// + /// Returns the ids the worker actually released. + pub async fn acknowledge_replies( + &mut self, + worker: &WorkerRef, + request_ids: &[DomainRequestId], + ) -> Outcome> { + let params = AcknowledgeParams { request_ids: request_ids.to_vec() }; + let reply = self + .call(worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[]) + .await?; + let result: AcknowledgeResult = + reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?; + if let Err(e) = result.validate_against(¶ms) { + return Err(self.fail_now( + DomainError::before( + ErrorCode::IdentityMismatch, + format!("a worker acknowledged an id this session never asked about: {e}"), + ), + "acknowledge", + )); + } + Ok(result.acknowledged) + } + /// Queries one worker's status without waiting for its current mutation. pub async fn status(&mut self, worker: &WorkerRef) -> Outcome { let reply = self @@ -1274,16 +1329,22 @@ impl Coordinator { let attempts = self.deadlines.resolve_attempts; let started = Instant::now(); self.resolutions += 1; + // Both, together: a resolution that ends before its first attempt would otherwise + // report the previous one's count. self.last_resolution = None; + self.last_resolution_attempts = 0; self.audit.push(format!("resolve:{}:{method}", worker.worker_id)); // The budget is the working limit and the attempt count is a guard; whichever runs // out is recorded, so "it gave up" is never an unexplained number. let mut end = ResolutionEnd::AttemptsExhausted; + let mut spent = 0u32; for _ in 0..attempts { if started.elapsed() >= budget { end = ResolutionEnd::BudgetExpired; break; } + spent += 1; + self.last_resolution_attempts = spent; let outcome = call_owned( self.bus.clone(), worker.clone(), diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs index f778858..65827b6 100644 --- a/services/flysim/crates/fly-session/src/launcher.rs +++ b/services/flysim/crates/fly-session/src/launcher.rs @@ -1519,6 +1519,9 @@ mod flag_tests { commit_delay_ms: 2, fail_stage_restore: true, fail_activate_restore: true, + // Argv carries the injected faults a worker process can have; this one is + // in-process only, because the check it drives is the caller's. + acknowledge_extra_id: None, }, client_id: "worker-fly-a".to_owned(), service: "agent.fly-a".to_owned(), diff --git a/services/flysim/crates/fly-session/src/worker.rs b/services/flysim/crates/fly-session/src/worker.rs index d717878..e361af3 100644 --- a/services/flysim/crates/fly-session/src/worker.rs +++ b/services/flysim/crates/fly-session/src/worker.rs @@ -194,6 +194,13 @@ pub trait WorkerEndpoint: Send + 'static { /// allocation" can read the allocation instead of being told it out of band. fn worker_threads(&self) -> u64; + /// An id this worker will add to every `Worker.Acknowledge` reply, for a test that needs a + /// worker reporting about something it was never asked about. `None` for a worker that + /// behaves. + fn acknowledge_extra_id(&self) -> Option { + None + } + /// The domain methods this endpoint implements, beyond the common `Worker.*` set. /// Anything else returns UNSUPPORTED without entering the endpoint. fn methods(&self) -> Vec<&'static str>; @@ -281,7 +288,8 @@ async fn run( ) { // Identity and capabilities are fixed for the endpoint's lifetime, so the shell reads them // once and never takes the endpoint mutex to answer Hello or Status. - let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads) = { + #[allow(clippy::type_complexity)] + let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads, extra_ack) = { let e = endpoint.lock().await; ( e.worker_id(), @@ -292,6 +300,7 @@ async fn run( e.status_cell(), e.methods(), e.worker_threads(), + e.acknowledge_extra_id(), ) }; let mut running: Vec> = Vec::new(); @@ -345,7 +354,7 @@ async fn run( continue; } "Worker.Acknowledge" => { - let outcome = match acknowledge(&request, &cache).await { + let outcome = match acknowledge(&request, &cache, extra_ack.as_ref()).await { Ok(result) => success(&request, &worker_id, &incarnation_id, result), Err(e) => { failure(&request.request_id, &worker_id, &incarnation_id, request.scope.clone(), e) @@ -717,16 +726,24 @@ fn hello( async fn acknowledge( request: &SessionRpcRequest, cache: &Arc>, + extra: Option<&Id>, ) -> DomainResult> { let params: AcknowledgeParams = AcknowledgeParams::from_json(&request.params) .map_err(|e| DomainError::invalid(format!("Worker.Acknowledge: {e}")))?; if params.request_ids.is_empty() || params.request_ids.len() > MAX_ACKNOWLEDGE { return Err(DomainError::invalid("Worker.Acknowledge takes 1..=16 request ids")); } - let acknowledged = { + let mut acknowledged = { let mut c = cache.lock().await; c.acknowledge(¶ms.request_ids) }; + // A deliberately misbehaving worker, for the caller-side subset check to refuse. + if let Some(extra) = extra + && let Ok(id) = DomainRequestId::parse(extra) + && !params.request_ids.contains(&id) + { + acknowledged.push(id); + } let result = AcknowledgeResult { acknowledged }; Ok(object(result.to_json())) } diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs index 0eaa267..76f853d 100644 --- a/services/flysim/crates/fly-session/tests/processes.rs +++ b/services/flysim/crates/fly-session/tests/processes.rs @@ -16,13 +16,15 @@ use common::{at, count, fly_a, fly_b, mode_fixture, within}; use fly_session::agent::AgentFaults; use fly_session::coordinator::{DispatchOrder, Injections}; use fly_session::environment::EnvironmentFaults; -use fly_session::harness::{ExecutionMode, HarnessConfig, Via}; +use fly_session::harness::{AgentSpec, ExecutionMode, HarnessConfig, Via}; use fly_session::launcher::{ReapOutcome, ThreadBudget}; use fly_session::ResolutionEnd; use fly_session::phase::Phase; use fly_session::types::*; all_modes!( + an_acknowledge_that_releases_nothing_is_not_a_failure, + bootstrap_survives_the_second_acknowledge_its_resolution_makes, a_slow_participant_is_resolved_rather_than_failed, a_resolution_says_which_of_its_two_bounds_ended_it, a_delayed_one_agent_result_holds_the_world, @@ -84,6 +86,111 @@ async fn sequential_reversed_and_parallel_completion_agree() { } } +// ------------------------------------------------------------------------------------------- +// ipc-v1 section 5: an Acknowledge that releases nothing is success + +/// `ipc-v1` section 5: "Already released/unknown IDs are ignored." +/// +/// A second `Worker.Acknowledge` of ids the worker has already released answers with an empty +/// list. That is the contract working, not a worker misbehaving, and the coordinator must +/// accept it and carry on. The session's own bootstrap releases every lifecycle reply, so +/// asking again for the same ids is exactly that case -- driven directly here rather than by +/// making something slow, because it is a rule about the reply and not about timing. +/// +/// The rule has teeth because of section 6: any Acknowledge whose reply outruns the probe is +/// resolved, and the resolution *is* a second Acknowledge of the same ids. A coordinator that +/// demands the whole list back therefore fences a healthy session the first time a worker is +/// slow to answer. It did, on this branch's parent; this test fails if that check returns. +async fn an_acknowledge_that_releases_nothing_is_not_a_failure(mode: ExecutionMode) { + let mut f = mode_fixture(mode, two_agents(mode)).await; + // Bootstrap acknowledges every lifecycle reply, so afterwards the worker holds none. + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap(); + + // The ids bootstrap already released. The worker ignores them and releases nothing. + let already: Vec = + (1..=3).map(DomainRequestId::from_serial).collect(); + let released = within( + "acknowledge", + f.harness.coordinator.acknowledge_replies(&worker, &already), + ) + .await + .expect("a second Acknowledge of released ids is success, not a failed epoch"); + assert!( + released.is_empty(), + "already released ids are ignored, so this call released nothing: {released:?}" + ); + + // The session is untouched by it: not fenced, still at its boundary, and still plays. + assert!(!f.harness.coordinator.is_fenced(), "an empty acknowledgment is not a fault"); + assert_eq!(f.harness.coordinator.phase(), Phase::Ready(0)); + let report = within("step", f.harness.coordinator.step()) + .await + .expect("the session continues after an Acknowledge that released nothing"); + assert_eq!(report.boundary, 1); + assert_eq!(f.harness.coordinator.stats().advances, 1); + f.shutdown().await; +} + +/// The same rule, on the path `bootstrap` actually uses. +/// +/// The test above calls `acknowledge_replies` directly, which guards the check where it lives +/// now but not where it lived before: a length check reintroduced into `acknowledge_lifecycle` +/// after that call would leave it green. This one drives bootstrap itself, with the +/// `duplicate_lifecycle_acknowledge` injection doing exactly what the section 6 resolution +/// does -- the same ids again, to a worker that has already released them -- so the second, +/// empty answer has to be accepted by every check on bootstrap's path. +async fn bootstrap_survives_the_second_acknowledge_its_resolution_makes(mode: ExecutionMode) { + let mut f = mode_fixture(mode, two_agents(mode)).await; + f.harness.coordinator.injections = Injections { + duplicate_lifecycle_acknowledge: true, + ..Injections::default() + }; + within("bootstrap", f.harness.coordinator.bootstrap()) + .await + .expect("bootstrap accepts the second, empty acknowledgment of its own lifecycle ids"); + assert!(!f.harness.coordinator.is_fenced()); + assert_eq!(f.harness.coordinator.phase(), Phase::Ready(0)); + let report = within("step", f.harness.coordinator.step()).await.expect("and still plays"); + assert_eq!(report.boundary, 1); + f.shutdown().await; +} + +/// The other half of the rule: a short list is accepted, an id outside the request is not. +/// +/// A worker reports what *it* released, so fewer ids than asked for is success -- but it is +/// only entitled to report about the ids it was asked about. An id from outside the request is +/// a worker talking about another caller's cache, and +/// `AcknowledgeResult::validate_against` is what refuses it. Without this, dropping the length +/// check left nothing checking the reply against the request at all. +/// +/// Not generated per mode, deliberately. The check is the *caller's*, so the mode of the +/// worker that misbehaves is irrelevant to it, and the alternative -- carrying the +/// misbehaviour to a separate process over argv -- would put a flag in the shipped binary +/// whose only purpose is to make a worker lie about its acknowledgments. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_acknowledged_id_outside_the_request_is_refused() { + let mode = ExecutionMode::InProcess; + let mut config = two_agents(mode); + // This worker adds an id nobody asked about to every acknowledgment. + config.agents[0].faults = AgentFaults { + acknowledge_extra_id: Some(id("req-9999")), + ..AgentFaults::default() + }; + let mut f = mode_fixture(mode, config).await; + let failure = within("bootstrap", f.harness.coordinator.bootstrap()) + .await + .expect_err("a worker may not acknowledge an id this session never asked about"); + assert_eq!(failure.error.code, ErrorCode::IdentityMismatch); + assert!( + failure.error.message.contains("never asked about"), + "the refusal says what was wrong: {failure}" + ); + assert_eq!(failure.detail, "acknowledge"); + assert_eq!(failure.error.mutation, MutationCertainty::None, "refused before any mutation"); + f.shutdown().await; +} + // ------------------------------------------------------------------------------------------- // ipc-v1 section 6: an uncertain call is resolved, not failed @@ -114,17 +221,21 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode) config.environment_faults = EnvironmentFaults { advance_delay_ms: 500, ..EnvironmentFaults::default() }; let mut f = mode_fixture(mode, config).await; + // Bootstrap first, at ordinary deadlines: its lifecycle calls are not what this test is + // about, and squeezing them through the probe below only tests the machine's luck. + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); // A probe well inside both delays, and a resolution budget well outside them: the point is - // a call that expires and an operation that is nevertheless fine. + // a call that expires and an operation that is nevertheless fine. The guard is out of + // reach so the budget is the only bound in play, and the budget is far above what the + // delays need, so neither ends this resolution -- the answer does. f.harness.coordinator.deadlines = fly_session::Deadlines { probe: Duration::from_millis(120), - resolve: Duration::from_secs(20), - resolve_attempts: 4096, + resolve: Duration::from_secs(15), + resolve_attempts: u32::MAX, boot: Duration::from_secs(30), capture: Duration::from_secs(30), durable: Duration::from_secs(60), }; - within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); let reports = within("run", f.harness.coordinator.run(2)) .await .expect("a slow participant is resolved, not failed"); @@ -176,28 +287,46 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode) f.shutdown().await; } +/// One agent, one port, and a participant that will not answer this side of the test's own +/// timeout. The composition for the bound tests: one participant means one possible name in +/// the failure, so which agent is blamed is not a race. +fn one_silent_agent(mode: ExecutionMode) -> HarnessConfig { + HarnessConfig { + agents: vec![AgentSpec { + // Ten minutes. The suite's own `within` gives up at twenty seconds, so if the step + // returns at all, a bound ended it and not the participant. That is a claim about + // the code rather than about how fast this machine happens to be. + faults: AgentFaults { prepare_delay_ms: 600_000, ..AgentFaults::default() }, + ..AgentSpec::new("fly-a", "p1", 7) + }], + mode, + ..HarnessConfig::default() + } +} + /// The resolution has two bounds, and which one ended it is never left to be guessed. /// -/// `resolve` is the working limit at the default values -- the attempt guard is over sixteen -/// seconds of pauses against an eight-second budget -- so an unresponsive participant runs the -/// budget out. Setting the guard low instead ends the same resolution the other way, and the -/// failure says so both in `last_resolution` and in its own message. +/// Both halves are arranged so the bound under test is the only one that *can* fire: the +/// other is set orders of magnitude out of reach, so no amount of scheduling delay flips them. +/// The claim is the contract's -- a resolution ends by budget or by guard, records which, and +/// names it in the failure -- and nothing here is timed. +/// +/// The deadlines are installed after `bootstrap`, deliberately. Bootstrap makes lifecycle +/// calls of its own, and squeezing them through a fifty-millisecond probe tests the harness's +/// luck rather than the resolution. async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode) { - // The budget is what ends it at ordinary settings: a generous attempt guard, a short - // budget, and a participant far slower than either. - let mut config = two_agents(mode); - config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() }; - let mut f = mode_fixture(mode, config).await; + // Half one: the budget fires, because the guard cannot. `u32::MAX` attempts at the two + // millisecond pause is over ninety days; the budget is a fifth of a second. + let mut f = mode_fixture(mode, one_silent_agent(mode)).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); f.harness.coordinator.deadlines = fly_session::Deadlines { probe: Duration::from_millis(50), - resolve: Duration::from_millis(300), - resolve_attempts: 8192, + resolve: Duration::from_millis(200), + resolve_attempts: u32::MAX, boot: Duration::from_secs(30), capture: Duration::from_secs(30), durable: Duration::from_secs(60), }; - within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); - let started = Instant::now(); let failure = within("step", f.harness.coordinator.step()) .await .expect_err("a participant that never answers exhausts the resolution"); @@ -206,38 +335,42 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode) failure.error.message.contains("resolution budget"), "the message names the bound that fired: {failure}" ); - assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str())); + // The budget ended it with attempts still in hand, which is what makes it the budget. A + // 200 ms budget at a 50 ms probe cannot spend more than a handful, and `u32::MAX` was + // never in reach; asserting against the guard's own size would be vacuous. + let spent = f.harness.coordinator.last_resolution_attempts; + assert!(spent >= 1, "the resolution made at least one attempt"); + assert!(spent < 100, "and nowhere near its guard: {spent}"); + assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str())); assert_eq!(failure.error.mutation, MutationCertainty::Unknown); - assert!( - started.elapsed() < Duration::from_secs(20), - "the budget, not the 30-second participant, is what ended it" - ); assert!(f.harness.coordinator.is_fenced()); f.shutdown().await; - // The guard is what ends it when it is set below the budget: three attempts against a - // budget the participant could never reach anyway. - let mut config = two_agents(mode); - config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() }; - let mut f = mode_fixture(mode, config).await; + // Half two: the guard fires, because the budget cannot. Three attempts against an hour. + let mut f = mode_fixture(mode, one_silent_agent(mode)).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); f.harness.coordinator.deadlines = fly_session::Deadlines { probe: Duration::from_millis(50), - resolve: Duration::from_secs(600), + resolve: Duration::from_secs(3_600), resolve_attempts: 3, boot: Duration::from_secs(30), capture: Duration::from_secs(30), durable: Duration::from_secs(60), }; - within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); let failure = within("step", f.harness.coordinator.step()) .await .expect_err("three attempts are not enough to resolve a silent participant"); assert_eq!(f.harness.coordinator.last_resolution, Some(ResolutionEnd::AttemptsExhausted)); assert!( - failure.error.message.contains("attempt guard") && failure.error.message.contains("3 attempts"), + failure.error.message.contains("attempt guard") + && failure.error.message.contains("3 attempts"), "the message names the bound that fired and its size: {failure}" ); - assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str())); + // Counted, not timed: the guard was spent exactly, and the hour never came near. + assert_eq!(f.harness.coordinator.last_resolution_attempts, 3); + assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str())); + assert_eq!(failure.error.mutation, MutationCertainty::Unknown); + assert!(f.harness.coordinator.is_fenced(), "an exhausted guard fences the epoch too"); f.shutdown().await; } @@ -294,6 +427,35 @@ async fn a_delayed_one_agent_result_holds_the_world(mode: ExecutionMode) { // ------------------------------------------------------------------------------------------- // Acceptance: worker or helper death has a bounded diagnosed outcome +/// Waits until `worker` is provably inside the operation, then kills it. +/// +/// Sleeping a fixed time before the kill asserts a race: under load the kill can land before +/// the call is even dispatched, and then `MutationCertainty::None` is the *correct* answer +/// because the participant never received anything. The certainty the death rows are about -- +/// `unknown`, because the participant died with work in its hands -- only holds if the work +/// reached it, so the test waits for the worker's own status to say so rather than guessing +/// from the clock. +async fn kill_once_it_is_working( + launcher: &mut fly_session::Launcher, + worker: &Id, + inside: impl Fn(&StatusResult) -> bool, +) -> ReapOutcome { + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if let Ok(status) = launcher.health_check(worker).await + && inside(&status) + { + break; + } + assert!( + Instant::now() < deadline, + "{worker} never reported itself inside the operation" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } + launcher.kill(worker).await +} + /// One agent dies in the middle of its Prepare. The epoch fails with a typed cause naming /// that agent, within the caller's own budget, and nothing continues on the remainder. async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { @@ -301,22 +463,21 @@ async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { config.agents[1].faults = AgentFaults { prepare_delay_ms: 5_000, ..AgentFaults::default() }; let mut f = mode_fixture(mode, config).await; within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); - let started = Instant::now(); + let victim = fly_b(); let (coordinator, launcher) = f.harness.parts(); let (stepped, reaped) = tokio::join!( async { within("step", coordinator.step()).await }, - async { - tokio::time::sleep(Duration::from_millis(80)).await; - launcher.kill(&fly_b()).await - } + // Killed once it has the Prepare in its hands, not after a fixed sleep: the row is + // about a participant that dies *with work*, so the work has to have reached it. + kill_once_it_is_working(launcher, &victim, |status| { + status.state == WorkerState::Preparing && status.active_request_id.is_some() + }) ); assert_eq!(reaped, ReapOutcome::Terminated); + // Boundedness is the suite's own `within` above: the participant is five seconds slow and + // `within` gives up at twenty, so returning at all is the claim. let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one"); - assert!( - started.elapsed() < Duration::from_secs(20), - "the outcome must be bounded, not a hang" - ); assert_eq!( failure.participant.as_deref(), Some(fly_b().as_str()), @@ -350,19 +511,20 @@ async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { let mut f = mode_fixture(mode, config).await; within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); let environment = f.harness.environment_id(); - let started = Instant::now(); let (coordinator, launcher) = f.harness.parts(); let (stepped, reaped) = tokio::join!( async { within("step", coordinator.step()).await }, - async { - tokio::time::sleep(Duration::from_millis(200)).await; - launcher.kill(&environment).await - } + // Killed once the world has recorded the batch, which the arena does before its + // injected delay. So the Advance provably reached it and the certainty is `unknown` + // rather than `none`; a fixed sleep could land before dispatch under load, and then + // `none` would be right and this row would be asserting a race. + kill_once_it_is_working(launcher, &environment, |status| { + status.last_batch_id.is_some() + }) ); assert_eq!(reaped, ReapOutcome::Terminated); let failure = stepped.expect_err("a dead world is a failed epoch"); - assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang"); assert_eq!( failure.participant.as_deref(), Some(environment.as_str()),