From a3c1c125cd454dc190e3ac5a3f66472a200ebfdc Mon Sep 17 00:00:00 2001 From: acamilo Date: Tue, 22 Sep 2026 16:00:13 +0000 Subject: [PATCH] session: negative fixtures for the allocation, and honest resolution bounds Two review notes. The launcher allocation became a required field of HelloResult.limits with no negative fixture behind it. Three rows now cover it: missing, above maxWorkerThreads, and zero. Both readers reject all three, and no derived fixture moved, because invalid.json is not one of them. The resolution's two bounds disagreed. 512 attempts at a 2 ms pause give up near 1.5 s, so the attempt count silently pre-empted the 8 s budget the doc comment advertised. The budget is now the working limit and says so: the guard is 8192 attempts, over sixteen seconds of pauses against an eight-second budget, so at the default values the budget is always what fires. Which one did is no longer arithmetic either -- ResolutionEnd records it, the failure message names the bound and its size, and the code, the doc comment and the README all state the same numbers. a_resolution_says_which_of_its_two_bounds_ended_it drives each bound to the end in every execution mode. --- .../fly-session-types/fixtures/invalid.json | 65 +++++++++++++++ services/flysim/crates/fly-session/README.md | 26 +++--- .../crates/fly-session/src/coordinator.rs | 79 +++++++++++++++---- services/flysim/crates/fly-session/src/lib.rs | 2 +- .../crates/fly-session/tests/processes.rs | 68 ++++++++++++++++ 5 files changed, 216 insertions(+), 24 deletions(-) diff --git a/services/flysim/crates/fly-session-types/fixtures/invalid.json b/services/flysim/crates/fly-session-types/fixtures/invalid.json index 7305201..cbce2b4 100644 --- a/services/flysim/crates/fly-session-types/fixtures/invalid.json +++ b/services/flysim/crates/fly-session-types/fixtures/invalid.json @@ -2852,6 +2852,71 @@ }, "reason": "v1 selects major 1" }, + { + "name": "hello result without its launcher allocation", + "type": "HelloResult", + "value": { + "selectedMajor": 1, + "selectedMinor": 0, + "workerId": "fly-a", + "incarnationId": "inc-1", + "role": "agent", + "buildDigest": "44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66", + "contractDigest": "cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf", + "capabilities": [ + "agent-step-v1" + ], + "limits": { + "maxAgents": 4, + "maxPorts": 4 + } + }, + "reason": "limits.workerThreads is required by the 2026-09-22 workers-v1 amendment" + }, + { + "name": "hello result promising more threads than a launcher may allocate", + "type": "HelloResult", + "value": { + "selectedMajor": 1, + "selectedMinor": 0, + "workerId": "fly-a", + "incarnationId": "inc-1", + "role": "agent", + "buildDigest": "44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66", + "contractDigest": "cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf", + "capabilities": [ + "agent-step-v1" + ], + "limits": { + "maxAgents": 4, + "maxPorts": 4, + "workerThreads": 4097 + } + }, + "reason": "limits.workerThreads is at most maxWorkerThreads" + }, + { + "name": "hello result reporting no threads at all", + "type": "HelloResult", + "value": { + "selectedMajor": 1, + "selectedMinor": 0, + "workerId": "fly-a", + "incarnationId": "inc-1", + "role": "agent", + "buildDigest": "44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66", + "contractDigest": "cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf", + "capabilities": [ + "agent-step-v1" + ], + "limits": { + "maxAgents": 4, + "maxPorts": 4, + "workerThreads": 0 + } + }, + "reason": "limits.workerThreads is at least one" + }, { "name": "hello params with no supported majors", "type": "HelloParams", diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md index f5de173..8e114e2 100644 --- a/services/flysim/crates/fly-session/README.md +++ b/services/flysim/crates/fly-session/README.md @@ -115,15 +115,23 @@ fly-session measure --steps 300 --agents 1,2,4 participant it is attributed to, and failing fences the session: the committed boundary stops moving, the artifact handles are dropped, and no further transition or publication is allowed. Lifting the fence is a coherent group restore, which is STATE-01's. -- **The `ipc-v1` section 6 procedure, on the path that reaches it.** A call that goes without - a terminal reply for the probe budget is *uncertain*, not failed. The coordinator then - queries the same operation -- a fresh bus call carrying the original domain request id and - body, pinned to the same incarnation, with its retained attachments -- for a bounded number - of attempts within a bounded budget, absorbing `IN_PROGRESS` while the original is still - running. Only when that ends without a definite answer, or the incarnation is gone, or the - retained result expired, is the epoch failed. A merely slow participant therefore finishes - its step, and `step-v1` section 7's "query/retransmit same request to same incarnation; - never new batch" is the same code path for a slow Advance. +- **The `ipc-v1` section 6 procedure, on the path that reaches it.** A call that goes two + seconds without a terminal reply is *uncertain*, not failed. The coordinator then queries + the same operation -- a fresh bus call carrying the original domain request id and body, + pinned to the same incarnation, with its retained attachments -- absorbing `IN_PROGRESS` + while the original is still running. Only when that ends without a definite answer, or the + incarnation is gone, or the retained result expired, is the epoch failed. A merely slow + participant therefore finishes its step, and `step-v1` section 7's "query/retransmit same + request to same incarnation; never new batch" is the same code path for a slow Advance. + + The procedure has two explicit bounds, and they do not mean the same thing. **`resolve`, 8 + seconds, is the working limit**: two to notice plus eight to resolve is section 6's ten + seconds without progress. **`resolve_attempts`, 8192, is a guard**, not the limit -- the + procedure pauses 2 ms between attempts, so the guard is over sixteen seconds of pauses + alone, twice the budget, and an attempt whose call expires costs a whole probe on top. At + these values the budget is always what fires. Which one did is recorded in + `Coordinator::last_resolution` and named in the failure's own message, so an exhausted + resolution never has to be explained by arithmetic. - **A bounded diagnosed outcome.** Those budgets are the coordinator's own, on its own clock, so a participant that dies or stops answering produces a typed failure naming it rather than a hang. An expired deadline is `unknown`, never `none`: a caller-side timeout is not diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index b72d4b8..12a7938 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -118,28 +118,53 @@ type Outcome = Result; /// answer, or the incarnation is gone, or the retained result expired, is the epoch failed. /// /// The prototype values follow section 6: probe at two seconds without a reply, give up at -/// ten seconds without progress, with a separate budget for a long boot. They are -/// failure-detection values, not a gameplay latency goal. The attempt count is explicit -/// because section 6 forbids filling this gap with an implicit best-effort policy. +/// ten seconds without progress -- two to notice plus eight to resolve -- with a separate +/// budget for a long boot. They are failure-detection values, not a gameplay latency goal. +/// +/// **Which bound ends a resolution.** `resolve` is the one that does, at these values. +/// Between attempts the procedure sleeps [`RESOLVE_PAUSE`], so the fastest the attempt count +/// can be spent is `resolve_attempts * RESOLVE_PAUSE`; the default 8192 attempts is over +/// sixteen seconds of pauses alone, twice the eight-second budget, and an attempt whose call +/// expires costs a whole `probe` on top. `resolve_attempts` is therefore a second, coarser +/// stop for a pathological loop that costs nothing per turn, not the working limit. Both are +/// explicit because section 6 forbids filling this gap with an implicit best-effort policy, +/// and [`ResolutionEnd`] says which of them fired. #[derive(Clone, Copy, Debug)] pub struct Deadlines { /// Without a terminal reply for this long, the call is uncertain. pub probe: Duration, - /// The resolution's own budget, measured from its first attempt. + /// The resolution's own budget, measured from its first attempt. The working limit. pub resolve: Duration, - /// How many times the resolution may re-ask. Bounded, and never a retry of the operation: - /// every attempt carries the original request id and body. + /// How many times the resolution may re-ask, as a guard rather than the working limit. + /// Never a retry of the operation: every attempt carries the original request id and body. pub resolve_attempts: u32, /// A separate, larger budget for `Worker.Hello` and the `Initialize` methods. pub boot: Duration, } +/// How long the resolution waits between attempts. +pub const RESOLVE_PAUSE: Duration = Duration::from_millis(2); + +/// What ended a resolution, so a caller can tell an exhausted budget from an exhausted +/// attempt count rather than reading one number out of a message. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ResolutionEnd { + /// A matching terminal result arrived. + Answered, + /// The `resolve` budget ran out. At the default values this is the one that fires. + BudgetExpired, + /// The `resolve_attempts` guard ran out first, which needs a pause short enough or an + /// attempt count low enough for it to be reached before the budget. + AttemptsExhausted, +} + impl Default for Deadlines { fn default() -> Deadlines { Deadlines { probe: Duration::from_secs(2), resolve: Duration::from_secs(8), - resolve_attempts: 512, + // Over sixteen seconds of pauses: the budget above is what terminates. + resolve_attempts: 8192, boot: Duration::from_secs(30), } } @@ -253,6 +278,8 @@ pub struct Coordinator { pub in_progress_replies: u64, /// How many uncertain calls ran the `ipc-v1` section 6 resolution. pub resolutions: u64, + /// How the last resolution ended, so a test or a supervisor can tell which bound fired. + pub last_resolution: Option, /// 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 @@ -312,6 +339,7 @@ impl Coordinator { injection_log: Vec::new(), in_progress_replies: 0, resolutions: 0, + last_resolution: None, deadlines: Deadlines::default(), metrics: Metrics::default(), blame: None, @@ -886,12 +914,18 @@ async fn call_owned( /// /// Deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof /// that nothing was mutated. -fn unresolved(method: &str, worker: &WorkerRef, budget: Duration) -> DomainError { +fn unresolved(method: &str, worker: &WorkerRef, end: ResolutionEnd, bound: &str) -> DomainError { + let why = match end { + ResolutionEnd::BudgetExpired => "its resolution budget", + ResolutionEnd::AttemptsExhausted => "its resolution attempt guard", + ResolutionEnd::Answered => "its resolution", + }; DomainError::new( ErrorCode::BackendFailure, format!( - "{method}: {} never resolved within {:?}; the operation's outcome is unknown", - worker.worker_id, budget + "{method}: {} never resolved; {why} of {bound} ran out and the operation's \ +outcome is unknown", + worker.worker_id ), MutationCertainty::Unknown, ) @@ -1065,9 +1099,14 @@ impl Coordinator { let attempts = self.deadlines.resolve_attempts; let started = Instant::now(); self.resolutions += 1; + self.last_resolution = None; 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; for _ in 0..attempts { if started.elapsed() >= budget { + end = ResolutionEnd::BudgetExpired; break; } let outcome = call_owned( @@ -1087,7 +1126,7 @@ impl Coordinator { // Still no answer. The original may simply be slow; asking again is the // procedure, and the request id it carries is unchanged. CallOutcome::Expired => { - tokio::time::sleep(std::time::Duration::from_millis(2)).await; + tokio::time::sleep(RESOLVE_PAUSE).await; continue; } // Step 4: routes or ownership lost, or the incarnation is gone. @@ -1097,17 +1136,24 @@ impl Coordinator { match reply.result() { Ok(_) => { self.blame(None); + self.last_resolution = Some(ResolutionEnd::Answered); return Ok(reply); } Err(e) if e.code == ErrorCode::InProgress => { self.in_progress_replies += 1; - tokio::time::sleep(std::time::Duration::from_millis(2)).await; + tokio::time::sleep(RESOLVE_PAUSE).await; } // A terminal refusal, including RESULT_EXPIRED: definite, so the epoch fails. Err(e) => return Err(self.fail_now(e, method)), } } - Err(self.fail_now(unresolved(method, worker, budget), method)) + self.last_resolution = Some(end); + let bound = match end { + ResolutionEnd::AttemptsExhausted => format!("{attempts} attempts"), + _ => format!("{budget:?}"), + }; + let error = unresolved(method, worker, end, &bound); + Err(self.fail_now(error, method)) } /// Sends the same domain request again on a fresh bus call and reports what came back, @@ -1198,7 +1244,12 @@ impl Coordinator { match outcome { CallOutcome::Answered(reply) => reply.result().cloned(), CallOutcome::Refused(e) => Err(e), - CallOutcome::Expired => Err(unresolved(method, worker, self.deadlines.probe)), + CallOutcome::Expired => Err(unresolved( + method, + worker, + ResolutionEnd::BudgetExpired, + &format!("{:?}", self.deadlines.probe), + )), } } diff --git a/services/flysim/crates/fly-session/src/lib.rs b/services/flysim/crates/fly-session/src/lib.rs index 93aea5e..5d9aae1 100644 --- a/services/flysim/crates/fly-session/src/lib.rs +++ b/services/flysim/crates/fly-session/src/lib.rs @@ -42,7 +42,7 @@ pub mod types; pub use fly_session_types; pub use coordinator::{ - Coordinator, Deadlines, DispatchOrder, Injections, SessionFailure, StepReport, + Coordinator, Deadlines, DispatchOrder, Injections, ResolutionEnd, SessionFailure, StepReport, }; pub use launcher::{ExecutionMode, Launcher, ReapOutcome, ThreadBudget, Via}; pub use phase::{Phase, PhaseMachine}; diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs index 1c3f6a4..d2dcd53 100644 --- a/services/flysim/crates/fly-session/tests/processes.rs +++ b/services/flysim/crates/fly-session/tests/processes.rs @@ -18,11 +18,13 @@ use fly_session::coordinator::{DispatchOrder, Injections}; use fly_session::environment::EnvironmentFaults; use fly_session::harness::{ExecutionMode, HarnessConfig, Via}; use fly_session::launcher::{ReapOutcome, ThreadBudget}; +use fly_session::ResolutionEnd; use fly_session::phase::Phase; use fly_session::types::*; all_modes!( 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, a_worker_death_has_a_bounded_diagnosed_outcome, a_helper_death_has_a_bounded_diagnosed_outcome, @@ -137,6 +139,11 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode) f.harness.coordinator.in_progress_replies > 0, "the resolution must have met the original still running" ); + assert_eq!( + f.harness.coordinator.last_resolution, + Some(ResolutionEnd::Answered), + "the resolution ended by being answered, not by running out of anything" + ); // No second operation anywhere: one advance per transition, one batch id per transition, // and the same behaviour as the run that never timed out. @@ -167,6 +174,67 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode) f.shutdown().await; } +/// 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. +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; + f.harness.coordinator.deadlines = fly_session::Deadlines { + probe: Duration::from_millis(50), + resolve: Duration::from_millis(300), + resolve_attempts: 8192, + boot: Duration::from_secs(30), + }; + 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"); + assert_eq!(f.harness.coordinator.last_resolution, Some(ResolutionEnd::BudgetExpired)); + assert!( + 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())); + 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; + f.harness.coordinator.deadlines = fly_session::Deadlines { + probe: Duration::from_millis(50), + resolve: Duration::from_secs(600), + resolve_attempts: 3, + boot: Duration::from_secs(30), + }; + 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"), + "the message names the bound that fired and its size: {failure}" + ); + assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str())); + f.shutdown().await; +} + // ------------------------------------------------------------------------------------------- // Acceptance: a delayed one-agent result holds the world