alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

session: a retried Acknowledge is not a failed epoch

Browse source 
The bound test failed two runs in thirty under load, and not on the
bound it was testing. Both failures were bootstrap: "a worker did not
acknowledge every lifecycle reply".

ipc-v1 section 5 says already released or unknown ids are ignored, and
the contract type already holds the acknowledged list to a subset of
the request. So the second Acknowledge of the same ids answers with an
empty list by design -- and the section 6 resolution produces exactly
that second Acknowledge whenever the first reply is slower than the
probe. Demanding the whole list back turned a safe, contract-sanctioned
retry into a failed epoch, which is a defect in the coordinator rather
than in the test: a slow lifecycle reply would do it to a real session
too.

The test made itself easy to hit by installing a fifty-millisecond
probe before bootstrap, so bootstrap's own lifecycle calls ran under a
budget meant for the step under test. It now bootstraps at ordinary
deadlines and tightens them afterwards.

The two bounds are also separated by construction rather than by clock.
Each half puts the bound it is not testing out of reach -- u32::MAX
attempts against a fifth of a second, three attempts against an hour --
so no scheduling delay can flip which one fires, and the silent
participant is ten minutes slow against a twenty-second test timeout,
so returning at all proves a bound ended it. The wall-clock assertion
is gone and the attempt count is asserted instead, which
last_resolution_attempts now records. One agent per composition, so the
participant the failure names is not a race either.
This commit is contained in:
acamilo 2026-09-22 18:47:42 +00:00
parent e76b3d1063
commit 6d67fa7ed2
2 changed files with 68 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
services/flysim/crates/fly-session/src/coordinator.rs
View file

@ -290,6 +290,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<ResolutionEnd>,
/// 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
@ -357,6 +359,7 @@ impl Coordinator {
in_progress_replies: 0,
resolutions: 0,
last_resolution: None,
last_resolution_attempts: 0,
deadlines: Deadlines::default(),
metrics: Metrics::default(),
blame: None,
@ -840,14 +843,15 @@ impl Coordinator {
let reply = self
.call(&worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[])
.await?;
let result: AcknowledgeResult =
// 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"))?;
if result.acknowledged.len() != ids.len() {
return Err(self.fail_now(
DomainError::invalid("a worker did not acknowledge every lifecycle reply"),
"acknowledge",
));
}
}
self.audit.push("acknowledge.lifecycle".to_owned());
Ok(())
@ -1147,11 +1151,14 @@ impl Coordinator {
// 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(),

85
services/flysim/crates/fly-session/tests/processes.rs
View file

@ -16,7 +16,7 @@ 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;
@ -114,15 +114,19 @@ 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),
};
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");
@ -174,26 +178,44 @@ 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),
};
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");
@ -202,36 +224,37 @@ 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()));
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"
f.harness.coordinator.last_resolution_attempts < u32::MAX,
"the budget ended it with attempts still in hand, which is what makes it the budget"
);
assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
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),
};
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);
f.shutdown().await;
}