GameCult/Idunn
Watch
9
0
Fork
You've already forked Idunn
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

B3 fix: pin what cargo-mutants found unpinned, delete what it found redundant

Browse source 
First cargo-mutants --in-diff pass over a61540b..aa1889b left 16 misses.

Deleted:
- the candidate challenge's own route-id check: rendering the Expected next
  holds the same binding, so it could never decide anything.
- odin_can_observe in advance_warming: is_odin && !(!is_odin || observed) is
  is_odin && !observed; two of the misses were operators in that detour.

Moved: which transactions are held for their evidence is now
DeploymentTransaction::held_disagreement, so "finished or aborting work is
never held" is a value with a test, not a local in resume_candidate.

Pinned: the admitted-generation topology refresh (taken once, then not
again), WarmingEvidence shape checks per source, the PresenceDisagrees text,
the plan's readiness declaration needing the shared-infrastructure kind, TCP
framed challenges (exact answer, bound, empty, non-CultNet, silence), RUDP
silence, and ChallengeFailure's text and source.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 15:24:45 +02:00
commit 86ee21dac5
3 changed files with 265 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

157
src/control_plane.rs
View file

@ -1438,6 +1438,18 @@ impl DeploymentTransaction {
.err()
}
/// The disagreement that holds this transaction: evidence collected under
/// another class is held and reported, not advanced and not repaired, since
/// no step can make it the right kind and running one anyway only fails it
/// every tick. What is finished, or being finished (a completion, or an
/// abort under way), never reads the class, so it is left to run.
fn held_disagreement(&self) -> Option<ReadinessDisagreement> {
let finishing = self.completion.is_some()
|| self.pre_fencing_abort.is_some()
|| self.post_fencing_abort.is_some();
self.readiness_disagreement().filter(|_| !finishing)
}
fn validate(&self) -> Result<()> {
ensure!(
self.schema_version == DEPLOYMENT_TRANSACTION_SCHEMA,
@ -4530,14 +4542,7 @@ impl Engine {
{
return Ok(false);
}
// Evidence collected under another class is held and reported, not
// advanced and not repaired: no step can make it the right kind, and
// running one anyway only fails it every tick. An abort already under
// way does not read the class, so it is left to finish.
let aborting = current.value.completion.is_some()
|| current.value.pre_fencing_abort.is_some()
|| current.value.post_fencing_abort.is_some();
if let Some(disagreement) = current.value.readiness_disagreement().filter(|_| !aborting) {
if let Some(disagreement) = current.value.held_disagreement() {
self.note_fault("holds transaction", id, &anyhow::Error::new(disagreement));
return Ok(false);
}
@ -5585,8 +5590,7 @@ impl Engine {
} else {
None
};
let odin_can_observe = !is_odin || odin_observation.is_some();
let odin_observes_itself = is_odin && !odin_can_observe;
let odin_observes_itself = is_odin && odin_observation.is_none();
if odin_observes_itself {
ensure!(
expected.write_lease_required,
@ -10384,6 +10388,59 @@ mod tests {
Ok(())
}
#[test]
fn warming_evidence_is_shape_checked_whatever_its_source() {
let good = |evidence: &ReadinessEvidence| match evidence {
ReadinessEvidence::RouteProof { evidence } => evidence.clone(),
ReadinessEvidence::OdinCorrelated { .. } => unreachable!("a route-proof fixture"),
};
let presence = good(&route_proof_evidence());
for warming in [
WarmingEvidence::FirstOdinDirect { evidence: presence.clone() },
WarmingEvidence::RouteProofDirect { evidence: presence.clone() },
WarmingEvidence::OdinTopology { evidence: topology(3, 1) },
] {
warming.validate_shape().expect("well-formed evidence");
}
let mut damaged = presence;
damaged.canonical_bytes.push(1);
for warming in [
WarmingEvidence::FirstOdinDirect { evidence: damaged.clone() },
WarmingEvidence::RouteProofDirect { evidence: damaged },
WarmingEvidence::OdinTopology {
evidence: TopologyEvidence {
canonical_sha256: "sha256:not-the-digest".into(),
..topology(3, 1)
},
},
] {
assert!(warming.validate_shape().is_err(), "{warming:?}");
}
}
#[test]
fn a_presence_disagreement_names_each_disagreement() {
let shortfall = PresenceDisagrees {
disagreements: vec![
OdinTopologyDisagreement {
code: "capacity-below-minimum".into(),
expected: Some("2".into()),
observed: Some("1".into()),
},
OdinTopologyDisagreement {
code: "state".into(),
expected: None,
observed: None,
},
],
};
assert_eq!(
shortfall.to_string(),
"route proof answered with a runtime that disagrees with current authority: \
capacity-below-minimum (expected 2, observed 1); state (expected none, observed none)"
);
}
fn route_proof_evidence() -> ReadinessEvidence {
let canonical_bytes = vec![7, 7, 7];
ReadinessEvidence::RouteProof {
@ -11925,6 +11982,51 @@ mod tests {
.clone())
}
#[test]
fn an_admitted_odin_correlated_generation_takes_a_newer_odin_reading_once() -> Result<()> {
let world = EngineFixture::with_workload(Arc::new(StillWorkload))?;
let seeded = transaction_at(&world, DeploymentPhase::Fencing)?;
let first = seeded_sequence(None) + 1;
odin_reports_ready(&world, &seeded, first)?;
drive(&world, |transaction| transaction.phase == DeploymentPhase::Complete)?;
assert!(world.engine.retire_one_terminal_transaction()?);
let admitted = ControlSnapshot::read(&world.state_store)?
.admitted_for("service")
.context("commit wrote no admitted generation")?
.value
.clone();
assert_eq!(admitted.readiness(), Ok(ReadinessClass::OdinCorrelated));
assert_eq!(admitted.odin_publisher_sequence_cursor, first);
// Nothing newer than what is held: nothing is written.
let snapshot = ControlSnapshot::read(&world.state_store)?;
let current = snapshot.admitted_for("service").context("no generation")?;
let envelope = current.envelope.clone();
assert!(!world.engine.refresh_admitted_topology(&snapshot, current)?);
assert_eq!(incumbent_envelope(&world)?, envelope);
// A newer reading is adopted, once.
odin_republishes(&world, &seeded, first + 4)?;
let snapshot = ControlSnapshot::read(&world.state_store)?;
let current = snapshot.admitted_for("service").context("no generation")?;
assert!(world.engine.refresh_admitted_topology(&snapshot, current)?);
let refreshed = ControlSnapshot::read(&world.state_store)?
.admitted_for("service")
.context("no generation")?
.value
.clone();
assert_eq!(refreshed.odin_publisher_sequence_cursor, first + 4);
assert_eq!(
refreshed.latest_odin_observation.as_ref().map(|latest| latest.publisher_sequence),
Some(first + 4)
);
assert_eq!(refreshed.ready.odin().map(|ready| ready.publisher_sequence), Some(first + 4));
let snapshot = ControlSnapshot::read(&world.state_store)?;
let current = snapshot.admitted_for("service").context("no generation")?;
assert!(!world.engine.refresh_admitted_topology(&snapshot, current)?);
Ok(())
}
/// Change the incumbent in the store, compare-and-swap.
fn edit_incumbent(
world: &EngineFixture,
@ -15086,6 +15188,41 @@ mod tests {
Ok(())
}
#[test]
fn only_a_transaction_that_still_reads_its_class_is_held_for_its_evidence() -> Result<()> {
let routed = routed_world(Odin::Unreachable, 1, DeploymentPhase::Fencing, false)?;
routed.stub.set_state("active");
routed.run_to_routing()?;
let at_routing = routed.transaction()?;
// What pre-B3 Idunn would have collected: an Odin receipt.
let mut legacy = at_routing.clone();
legacy.ready = Some(ReadinessEvidence::OdinCorrelated {
evidence: odin_receipt_for(&routed, &at_routing, 9)?,
});
let disagreement = legacy.readiness_disagreement();
assert!(disagreement.is_some());
assert_eq!(legacy.held_disagreement(), disagreement);
// Finished work, and an abort under way, never read the class.
let mut completed = legacy.clone();
completed.completion = Some(TransactionCompletion::Admitted {
generation_id: "generation-1".into(),
});
assert_eq!(completed.held_disagreement(), None);
let mut aborting = legacy.clone();
aborting.pre_fencing_abort = Some(PreFencingAbort {
error: "operator abort".into(),
candidate_cleanup: CleanupEvidence::Pending,
topology_reconciliation: CleanupEvidence::Pending,
source_cleanup: CleanupEvidence::Pending,
});
assert_eq!(aborting.held_disagreement(), None);
let mut aborting = legacy;
aborting.post_fencing_abort = Some(post_fencing_abort_intent(&aborting, "operator abort"));
assert_eq!(aborting.held_disagreement(), None);
Ok(())
}
#[test]
fn a_stored_record_that_declares_no_way_to_prove_readiness_is_held() -> Result<()> {
let world = EngineFixture::new()?;

4
src/deployment_plan.rs
View file

@ -1585,12 +1585,14 @@ nodes = ["yggdrasil"]
let provides_odin = |recipe: &str| {
recipe.replace("capability = \"service.runtime\"", "capability = \"odin.verse-rendezvous\"")
};
let cases: [(&str, String, &str, Option<u16>, Result<ReadinessClass, ()>); 5] = [
let cases: [(&str, String, &str, Option<u16>, Result<ReadinessClass, ()>); 6] = [
("depends on Odin, routed", RECIPE.to_owned(), BINDING, Some(18001), Ok(ReadinessClass::OdinCorrelated)),
("routed, no Odin", without_odin.clone(), BINDING, Some(18001), Ok(ReadinessClass::RouteProof)),
("unrouted, no Odin", unrouted(&without_odin), &unrouted_binding, None, Err(())),
("provides Odin, unrouted", provides_odin(&unrouted(&without_odin)), &unrouted_binding, None, Ok(ReadinessClass::OdinSelf)),
("depends on Odin, unrouted", unrouted(RECIPE), &unrouted_binding, None, Ok(ReadinessClass::OdinCorrelated)),
// Only the shared-infrastructure declaration is the declaration.
("requires the Odin capability, routed", RECIPE.replace("kind = \"shared-infrastructure\"", "kind = \"required\""), BINDING, Some(18001), Ok(ReadinessClass::RouteProof)),
];
for (name, recipe, binding, port, outcome) in cases {
let providers = [ready_odin_provider("odin", "odin-yggdrasil", 1)];

124
src/drivers.rs
View file

@ -5468,10 +5468,7 @@ impl NginxRouteDriver {
.as_ref()
.context("candidate challenge has no Expected route")
.map_err(refused)?;
ensure_or_refuse(
route.route_id == self.binding.route_id,
"candidate challenge binding differs from Expected",
)?;
// Rendering holds the binding to the Expected route, id and endpoints.
self.render(expected).map_err(refused)?;
self.request_runtime_presence_at(expected, message_id, &route.candidate_endpoint)
}
@ -8205,20 +8202,31 @@ mod tests {
fn challenge(
port: u16,
adjust: impl FnOnce(&mut IdunnExpectedIncarnationRecord),
) -> Result<RouteSnapshotResponse, ChallengeFailure> {
challenge_over("http", RouteDriver::NginxStreamTcp, port, adjust)
}
/// The same challenge over another transport: `transport` names both the
/// Expected's transport and the endpoint scheme.
fn challenge_over(
transport: &str,
driver: RouteDriver,
port: u16,
adjust: impl FnOnce(&mut IdunnExpectedIncarnationRecord),
) -> Result<RouteSnapshotResponse, ChallengeFailure> {
let temp = tempfile::tempdir().expect("a temporary directory");
let mut candidate = expected();
candidate.route = Some(cultnet_rs::IdunnExpectedRoute {
route_id: "service".into(),
transport: "http".into(),
stable_endpoint: "http://127.0.0.1:17999".into(),
candidate_endpoint: format!("http://127.0.0.1:{port}"),
transport: transport.into(),
stable_endpoint: format!("{transport}://127.0.0.1:17999"),
candidate_endpoint: format!("{transport}://127.0.0.1:{port}"),
});
adjust(&mut candidate);
NginxRouteDriver::new(RouteBinding {
driver: RouteDriver::NginxStreamTcp,
driver,
route_id: "service".into(),
stable_endpoint: "http://127.0.0.1:17999".into(),
stable_endpoint: format!("{transport}://127.0.0.1:17999"),
private_host: "127.0.0.1".into(),
private_port_start: port,
private_port_end: port,
@ -8227,6 +8235,104 @@ mod tests {
})
.request_candidate_runtime_presence(&candidate, "candidate-probe")
}
/// A TCP peer that reads one framed snapshot request and writes back the
/// bytes `reply` builds from the challenge id, verbatim.
fn framed_peer(reply: impl FnOnce(&str) -> Vec<u8> + Send + 'static) -> u16 {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("a free port");
let port = listener.local_addr().expect("a bound port").port();
thread::spawn(move || {
let Ok((mut stream, _)) = listener.accept() else {
return;
};
let mut header = [0_u8; 4];
if stream.read_exact(&mut header).is_err() {
return;
}
let mut request = vec![0_u8; u32::from_be_bytes(header) as usize];
if stream.read_exact(&mut request).is_err() {
return;
}
let Ok(CultNetMessage::SnapshotRequest { message_id, .. }) =
decode_cultnet_message_from_slice(&request, CultNetWireContract::CultNetSchemaV0)
else {
return;
};
let _ = stream.write_all(&reply(&message_id));
});
port
}
/// The body of an `honest` HTTP answer, framed for TCP.
fn framed_honest(message_id: &str) -> Vec<u8> {
let response = honest(message_id);
let body = response
.windows(4)
.position(|window| window == b"\r\n\r\n")
.map(|end| &response[end + 4..])
.expect("an HTTP answer");
encode_frame(body).expect("a frame")
}
fn tcp_challenge(port: u16) -> Result<RouteSnapshotResponse, ChallengeFailure> {
challenge_over("tcp", RouteDriver::NginxStreamTcp, port, |_| {})
}
#[test]
fn a_tcp_challenge_takes_the_exact_framed_answer_and_waits_only_on_silence() {
let answer = tcp_challenge(framed_peer(framed_honest)).expect("the exact answer");
assert_eq!(answer.message_id, "candidate-probe");
assert_eq!(answer.canonical_presence, [1, 2, 3]);
assert_silent(tcp_challenge(framed_peer(|_| Vec::new())), "a peer that hangs up");
assert_silent(
tcp_challenge(framed_peer(|_| vec![0, 0, 0, 9, 1])),
"a frame that never finishes",
);
// A frame that claims more than the observation bound, or nothing, is
// an answer that is wrong, not one that is late.
assert_refused(
tcp_challenge(framed_peer(|_| {
((ROUTE_SNAPSHOT_MAX_BYTES + 1) as u32).to_be_bytes().to_vec()
})),
"a frame over the bound",
);
assert_refused(
tcp_challenge(framed_peer(|_| 0_u32.to_be_bytes().to_vec())),
"an empty frame",
);
assert_refused(
tcp_challenge(framed_peer(|_| encode_frame(b"not cultnet").expect("a frame"))),
"a frame that is not CultNet",
);
}
#[test]
fn a_rudp_challenge_that_nothing_answers_is_silence() {
let socket = std::net::UdpSocket::bind("127.0.0.1:0").expect("a free port");
let port = socket.local_addr().expect("a bound port").port();
// Bound but never read: nothing answers, and nothing refuses either.
assert_silent(
challenge_over("rudp", RouteDriver::NginxStreamUdp, port, |_| {}),
"a rudp peer that never answers",
);
drop(socket);
}
/// A challenge that cannot be made says why, whatever the message.
#[test]
fn a_challenge_failure_shows_and_chains_its_cause() {
for failure in [
silent(anyhow!("cause of silence")),
refused(anyhow!("cause of refusal")),
] {
let shown = failure.to_string();
assert!(shown.starts_with("cause of "), "{shown}");
let source = std::error::Error::source(&failure).expect("the underlying error");
assert_eq!(source.to_string(), shown);
}
}
fn assert_silent(outcome: Result<RouteSnapshotResponse, ChallengeFailure>, why: &str) {
assert!(