diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 825a366..f9d74c5 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -839,24 +839,41 @@ 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?; - // The reply lists what this call released, which is not always everything it - // asked about: `ipc-v1` section 5 says "Already released/unknown IDs are - // ignored", and the contract type already holds the list to a subset of the - // request. A second Acknowledge therefore answers with an empty list by design -- - // and the section 6 resolution produces exactly that second Acknowledge whenever - // the first one's reply was slow. Demanding the whole list back turned a safe, - // contract-sanctioned retry into a failed epoch. - let _: AcknowledgeResult = - reply.parse().map_err(|e| self.fail_now(e, "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. + /// + /// 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"))?; + 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 diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs index 77014c5..6cdd3b5 100644 --- a/services/flysim/crates/fly-session/tests/processes.rs +++ b/services/flysim/crates/fly-session/tests/processes.rs @@ -23,6 +23,7 @@ use fly_session::phase::Phase; use fly_session::types::*; all_modes!( + an_acknowledge_that_releases_nothing_is_not_a_failure, 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 +85,52 @@ 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; +} + // ------------------------------------------------------------------------------------------- // ipc-v1 section 6: an uncertain call is resolved, not failed