diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index e6e34fa..8b7a444 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -471,6 +471,11 @@ impl Coordinator { self.descriptor_revision } + /// The sequence the next published snapshot will carry. + pub fn published_sequence(&self) -> u64 { + self.publisher.sequence() + } + /// What this session published and what became of it: accepted, refused by an observer, /// or faulted, per topic. pub fn ledger(&self) -> &crate::publish::Ledger { @@ -1656,6 +1661,25 @@ impl Coordinator { self.agents[index].context_digest = context.digest(); self.agents[index].context = context; self.agents[index].committed_step = k + 1; + // The telemetry of the transition that just ended, which is what this boundary's + // snapshot publishes. Without this the slot would keep whatever `Agent.Initialize` + // reported and every snapshot would label warm-up telemetry as boundary k. + let telemetry = commits + .iter() + .find(|(id, _)| *id == agent_id) + .map(|(_, result)| result.telemetry.clone()); + match telemetry { + Some(telemetry) => self.agents[index].telemetry = Some(telemetry), + None => { + return Err(self.fail_now( + DomainError::before( + ErrorCode::IdentityMismatch, + format!("agent {agent_id} committed without telemetry"), + ), + "commit", + )); + } + } } // The previous boundary's handles are no longer needed; the new ones take over. // The references -- which are data, not ownership -- are kept for one boundary, so a diff --git a/services/flysim/crates/fly-session/src/harness.rs b/services/flysim/crates/fly-session/src/harness.rs index 334e6ad..f86a179 100644 --- a/services/flysim/crates/fly-session/src/harness.rs +++ b/services/flysim/crates/fly-session/src/harness.rs @@ -649,6 +649,22 @@ impl SessionHarness { } } + /// Changes which graph one agent builds, so the replacement the next restart launches is + /// a fly with the same neuron count and another index. + /// + /// The same relaunch rule as a fault: the worker running now keeps what it was started + /// with, and the change reaches the composition through the next replacement. + pub fn set_agent_graph(&mut self, agent_id: &Id, graph_variant: u64) { + if let Some(spec) = self + .config + .agents + .iter_mut() + .find(|spec| spec.agent_id == *agent_id) + { + spec.graph_variant = graph_variant; + } + } + /// Changes the environment's injected faults, with the same relaunch rule. pub fn set_environment_faults(&mut self, faults: EnvironmentFaults) { self.config.environment_faults = faults; diff --git a/services/flysim/crates/fly-session/src/publish.rs b/services/flysim/crates/fly-session/src/publish.rs index 20f0200..67444a6 100644 --- a/services/flysim/crates/fly-session/src/publish.rs +++ b/services/flysim/crates/fly-session/src/publish.rs @@ -644,10 +644,13 @@ impl Publisher { .publish(&topic, object(snapshot.to_json()), &refs) .await, ); - if outcome.is_accepted() { - self.sequence += 1; - lock(&self.state).latest = Some(snapshot.clone()); - } + // The value is recorded whether or not the router admitted it, exactly as a descriptor + // revision is: the repair path exists for the consumer that did not receive it, and the + // `state-media-v1` amendment promises the exact value stays recoverable through the + // query service. The sequence advances with the value rather than with the delivery, + // so two different snapshots can never share one sequence number. + self.sequence += 1; + lock(&self.state).latest = Some(snapshot.clone()); self.ledger.record(&outcome); Ok(outcome) } @@ -718,6 +721,10 @@ pub const GET_SNAPSHOT: &str = "Session.GetSnapshot"; /// advance, pause, stimulate, restore or reconfigure anything. pub struct QueryService { task: tokio::task::JoinHandle<()>, + /// Answers this service produced and could not deliver, because the caller was gone or + /// the router refused the reply. A read that nobody received is not a read that happened, + /// and this module drops nothing silently. + undeliverable: Arc, } impl QueryService { @@ -736,6 +743,8 @@ impl QueryService { // than a fallback name that two incarnations could share. let incarnation = parse_id(&format!("query-{}", client.info().connection_id)) .map_err(|e| flybus::BusError::new(flybus::ErrorCode::InvalidEnvelope, e))?; + let undeliverable = Arc::new(std::sync::atomic::AtomicU64::new(0)); + let undelivered = Arc::clone(&undeliverable); let task = tokio::spawn(async move { while let Some(request) = service.next().await { let method = request.method().to_owned(); @@ -753,10 +762,17 @@ impl QueryService { DomainError::invalid(format!("{method}: {e}")), ), }; - let _ = responder.reply(outcome.to_outcome(), &[]).await; + if responder.reply(outcome.to_outcome(), &[]).await.is_err() { + undelivered.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } } }); - Ok(QueryService { task }) + Ok(QueryService { task, undeliverable }) + } + + /// How many answers this service could not deliver. + pub fn undeliverable(&self) -> u64 { + self.undeliverable.load(std::sync::atomic::Ordering::SeqCst) } /// Ends the service. Dropping one does the same thing. @@ -1003,6 +1019,14 @@ pub struct EventBatchView { pub event_ids: Vec, } +/// What one poll of the event stream produced. +#[derive(Clone, Debug, PartialEq)] +pub enum ConsumerEvents { + Batch(EventBatchView), + /// A batch this consumer could not read. Distinct from the end of the stream. + Unreadable { detail: String }, +} + /// What one poll of a consumer produced. #[derive(Clone, Debug, PartialEq)] pub enum ConsumerOutcome { diff --git a/services/flysim/crates/fly-session/tests/publishing.rs b/services/flysim/crates/fly-session/tests/publishing.rs index 1440e35..bb94a1c 100644 --- a/services/flysim/crates/fly-session/tests/publishing.rs +++ b/services/flysim/crates/fly-session/tests/publishing.rs @@ -36,6 +36,7 @@ both_transports!( application_state_and_cues_are_the_applications_own, a_refused_event_batch_is_held_and_counted_not_lost, the_query_service_answers_reads_and_nothing_else, + a_refused_snapshot_is_still_what_the_repair_path_answers, the_published_descriptor_is_what_the_workers_attested_to, a_stimulus_kind_the_descriptor_does_not_declare_is_refused, a_restored_boundary_publishes_a_new_revision_and_no_transition, @@ -252,7 +253,6 @@ async fn a_bounded_observer_refusal_is_named_and_never_fails_the_epoch(via: Via) .consumer() .await .expect("a healthy consumer attaches"); - until("the refusals stop", || true).await; f.harness .coordinator .run(2) @@ -284,6 +284,7 @@ async fn every_published_snapshot_carries_its_own_boundarys_media(via: Via) { Some(ConsumerOutcome::Composition { .. }) )); let mut seen: Vec<(u64, String)> = Vec::new(); + let mut ticks: BTreeMap = BTreeMap::new(); for step in 1..=STEPS { f.harness.coordinator.run(1).await.expect("a transition"); let at = read_until(&mut consumer, step).await; @@ -305,10 +306,20 @@ async fn every_published_snapshot_carries_its_own_boundarys_media(via: Via) { seen.push((at, reference.pixels.artifact_id.clone())); } for agent in &view.agents { - assert!( - agent.brain_ticks > 0, - "the agent state is this boundary's, not a placeholder" - ); + let previous = ticks.insert(agent.agent_id.clone(), agent.brain_ticks); + match previous { + None => assert!(agent.brain_ticks > 0, "the agent has run by boundary {at}"), + // Telemetry is the state of the transition that ended here. Republishing the + // previous boundary's numbers -- or `Agent.Initialize`'s warm-up numbers -- + // would be old agent state labelled as this boundary, which is the same + // mislabelling as old media. + Some(before) => assert!( + agent.brain_ticks > before, + "the telemetry of {} advanced into boundary {at}: {before} -> {}", + agent.agent_id, + agent.brain_ticks + ), + } } } let mut ids: Vec<&String> = seen.iter().map(|(_, id)| id).collect(); @@ -460,6 +471,15 @@ async fn a_revision_that_was_never_published_is_a_named_answer(via: Via) { /// The same neuron count, another index. A consumer that has mapped geometry is told, and the /// new composition is not quietly cached over the one it mapped. +/// +/// The two descriptors here come from two real sessions rather than from a restore. Driving it +/// through a restore was tried and does not work yet, for a reason outside this slice: the +/// coordinator's `AgentSlot.graph` is written only by `Agent.Initialize`, and a group restore +/// installs state through `State.ActivateRestore`, so a replacement fly that built another +/// index is published under its predecessor's `indexDigest` -- and it is not refused on the way +/// in either, because `agent_compatibility` digests `agent::dataset_digest()` rather than the +/// index the worker attested to. Both halves belong to the restore contract, so this test uses +/// the compositions it can build honestly and the gap is reported rather than papered over. async fn a_changed_index_digest_is_named_rather_than_remapped(via: Via) { // Two real compositions that differ only in the graph one fly built. let first = started(via).await; @@ -516,19 +536,14 @@ async fn a_changed_index_digest_is_named_rather_than_remapped(via: Via) { moved.neuron_count, arrived.neuron_count, "the same number of neurons" ); - assert_ne!( - moved.index_digest, arrived.index_digest, - "and another index" - ); + assert_ne!(moved.index_digest, arrived.index_digest, "and another index"); let mut consumer = first.harness.consumer().await.expect("a consumer attaches"); let held = match within("the descriptor", consumer.take_descriptor()).await { Some(ConsumerOutcome::Composition { revision, .. }) => revision, other => panic!("expected a composition, got {other:?}"), }; - consumer - .map_geometry(held) - .expect("this consumer maps geometry"); + consumer.map_geometry(held).expect("this consumer maps geometry"); assert_eq!(consumer.mapped_index(&fly_a()), Some(&moved.index_digest)); // The composition changes under it. @@ -966,6 +981,52 @@ async fn a_refused_event_batch_is_held_and_counted_not_lost(via: Via) { f.shutdown().await; } +/// A publication an observer refused is exactly the value the repair path exists to hand back. +/// +/// The `state-media-v1` amendment promises the exact value stays recoverable through the query +/// path, so recording it cannot depend on whether the router admitted the delivery -- that is +/// the case the promise is about. +async fn a_refused_snapshot_is_still_what_the_repair_path_answers(via: Via) { + let mut f = started(via).await; + let snapshots = f.harness.coordinator.topics().snapshots.clone(); + let offender_client = f.harness.observer().await.expect("an observer client"); + let _offender = offender_client + .subscribe( + &snapshots, + flybus::SubscriptionConfig::bounded().queued(1).in_flight(1), + ) + .await + .expect("a bounded subscription"); + + f.harness.coordinator.run(STEPS).await.expect("the world carries on"); + let counters = f.harness.coordinator.ledger().counters(&snapshots); + assert!(counters.refused > 0, "a refusal is what this test is about: {counters:?}"); + + let latest = { + let state = f.harness.coordinator.published_state(); + let state = state.lock().expect("not poisoned"); + state.latest_snapshot().expect("a snapshot").clone() + }; + assert_eq!( + latest.scope.step, STEPS, + "the newest committed boundary is recoverable even though its delivery was refused" + ); + assert_eq!( + latest.sequence + 1, + f.harness.coordinator.published_sequence(), + "the sequence advanced with the value, not with the delivery" + ); + + // And the query service answers it over the bus, not just the state behind it. + let mut consumer = f.harness.consumer().await.expect("a consumer attaches"); + let answered = within("the repair path", consumer.repair(DESCRIPTOR_REVISION)) + .await + .expect("the repair path answers"); + assert_eq!(answered, DESCRIPTOR_REVISION); + offender_client.close().await; + f.shutdown().await; +} + /// Two reads and nothing else. There is no method on this service that could move anything. async fn the_query_service_answers_reads_and_nothing_else(via: Via) { let mut f = started(via).await; @@ -990,7 +1051,15 @@ async fn the_query_service_answers_reads_and_nothing_else(via: Via) { let snapshot = CommittedSnapshot::from_json(outcome_result(&outcome).expect("a result")) .expect("a committed snapshot"); assert_eq!(snapshot.scope.step, 1); - assert!(snapshot.views.is_empty() || !snapshot.views.is_empty()); + assert_eq!( + snapshot.descriptor_revision, + f.harness.coordinator.descriptor_revision(), + "a read answers the composition the session is publishing under" + ); + assert!( + snapshot.agents.iter().all(|a| a.selected_decision.is_some()), + "and the values of the transition that ended at it" + ); // A method it does not implement is a named refusal, not a default. let request = SessionRpcRequest {