Merge fix/sf-resolution-bound-flake: a short acknowledgement is the contract, and two timing bets become claims
This commit is contained in:
acamilo
commit
954b9f4db5
7 changed files with 330 additions and 61 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<Id>,
|
||||
}
|
||||
|
||||
/// 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<Id> {
|
||||
self.config.faults.acknowledge_extra_id.clone()
|
||||
}
|
||||
|
||||
fn methods(&self) -> Vec<&'static str> {
|
||||
vec![
|
||||
"Agent.Initialize",
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -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<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
|
||||
|
|
@ -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<Vec<DomainRequestId>> {
|
||||
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<StatusResult> {
|
||||
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(),
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -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<Id> {
|
||||
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<E: WorkerEndpoint>(
|
|||
) {
|
||||
// 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: WorkerEndpoint>(
|
|||
e.status_cell(),
|
||||
e.methods(),
|
||||
e.worker_threads(),
|
||||
e.acknowledge_extra_id(),
|
||||
)
|
||||
};
|
||||
let mut running: Vec<tokio::task::JoinHandle<()>> = Vec::new();
|
||||
|
|
@ -345,7 +354,7 @@ async fn run<E: WorkerEndpoint>(
|
|||
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<tokio::sync::Mutex<ResultCache>>,
|
||||
extra: Option<&Id>,
|
||||
) -> DomainResult<Map<String, Value>> {
|
||||
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()))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<DomainRequestId> =
|
||||
(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()),
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue