diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs index 22b5ac5..0802c44 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. @@ -689,6 +692,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 bbd6bbf..92d426d 100644 --- a/services/flysim/crates/fly-session/src/cli.rs +++ b/services/flysim/crates/fly-session/src/cli.rs @@ -212,6 +212,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 78f7858..eae9fb4 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -54,6 +54,10 @@ pub struct Injections { pub altered_advance_controls: bool, /// Read and release the Advance result's frame, then replay the same operation. pub consume_advance_artifact_then_retry: 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. @@ -895,6 +899,21 @@ impl Coordinator { .push(request_id); } for (worker, ids) in by_worker.into_values() { + 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()); @@ -915,6 +934,12 @@ impl Coordinator { /// 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, @@ -927,6 +952,15 @@ impl Coordinator { .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) } @@ -1221,7 +1255,10 @@ 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. diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs index 43d6705..a19ab4d 100644 --- a/services/flysim/crates/fly-session/src/launcher.rs +++ b/services/flysim/crates/fly-session/src/launcher.rs @@ -1511,6 +1511,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 980ef3d..76f853d 100644 --- a/services/flysim/crates/fly-session/tests/processes.rs +++ b/services/flysim/crates/fly-session/tests/processes.rs @@ -24,6 +24,7 @@ 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, @@ -131,6 +132,65 @@ async fn an_acknowledge_that_releases_nothing_is_not_a_failure(mode: ExecutionMo 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 @@ -275,10 +335,12 @@ 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!( - f.harness.coordinator.last_resolution_attempts < u32::MAX, - "the budget ended it with attempts still in hand, which is what makes it the budget" - ); + // 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!(f.harness.coordinator.is_fenced()); @@ -308,6 +370,7 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode) 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; } @@ -400,7 +463,6 @@ 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(); @@ -413,10 +475,9 @@ async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { }) ); assert_eq!(reaped, ReapOutcome::Terminated); - let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one"); // 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 _ = started; + let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one"); assert_eq!( failure.participant.as_deref(), Some(fly_b().as_str()), @@ -450,7 +511,6 @@ 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!( @@ -465,7 +525,6 @@ async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) { ); assert_eq!(reaped, ReapOutcome::Terminated); let failure = stepped.expect_err("a dead world is a failed epoch"); - let _ = started; assert_eq!( failure.participant.as_deref(), Some(environment.as_str()),