diff --git a/docs/design/session-framework/checkpoint-envelope-v1.md b/docs/design/session-framework/checkpoint-envelope-v1.md index 69bdf4d..6e5017a 100644 --- a/docs/design/session-framework/checkpoint-envelope-v1.md +++ b/docs/design/session-framework/checkpoint-envelope-v1.md @@ -111,7 +111,9 @@ Without it a replacement fly that built another graph -- the same dataset, the s count, another index -- passed the group check and was then published under its predecessor's `indexDigest`, which is a graph identity crossing a recovery and exactly what section 5's rules exist to prevent. It is recorded from the worker's attestation rather than recomputed from the -dataset, because the point is that the two can disagree. +dataset, because the point is that the two can disagree. `envelopeVersion` stays `1`, which the +required-manifest-field rule below allows only while no production `FLYSESS1` file exists; once +one does, adding a required manifest field must bump it. **Amendment, 2026-09-22 (STATE-01).** The table above names a holder for every payload except the environment's own, although section 6's fixture has one (`world`) and a group install has diff --git a/services/flysim/crates/fly-session/src/publish.rs b/services/flysim/crates/fly-session/src/publish.rs index 67444a6..8390298 100644 --- a/services/flysim/crates/fly-session/src/publish.rs +++ b/services/flysim/crates/fly-session/src/publish.rs @@ -1275,26 +1275,42 @@ impl PresentationConsumer { } /// Takes the next bounded event batch. - pub async fn take_events(&mut self) -> Option { + /// + /// `None` is the end of the stream and nothing else. A batch this consumer cannot read + /// comes back as [`ConsumerEvents::Unreadable`], because "there are no more events" and + /// "that one made no sense" are different facts and one value cannot carry both: a + /// consumer that saw the second as the first would stop reading a live stream. + pub async fn take_events(&mut self) -> Option { let message = self.events.next().await?; let payload = message.payload().clone(); drop(message); - let epoch = payload.get("epoch").and_then(Value::as_str).map(id)?; - let dropped_before = payload + let unreadable = |what: &str| { + Some(ConsumerEvents::Unreadable { + detail: format!("an event batch has no readable {what}"), + }) + }; + let Some(epoch) = payload.get("epoch").and_then(Value::as_str).map(id) else { + return unreadable("epoch"); + }; + let Some(dropped_before) = payload .get("droppedBefore") .and_then(Value::as_str) - .and_then(|t| t.parse::().ok())?; - let event_ids = payload - .get("events") - .and_then(Value::as_array)? + .and_then(|t| t.parse::().ok()) + else { + return unreadable("droppedBefore"); + }; + let Some(events) = payload.get("events").and_then(Value::as_array) else { + return unreadable("events array"); + }; + let event_ids = events .iter() .filter_map(|e| e.get("id").and_then(Value::as_str).map(str::to_owned)) .collect(); - Some(EventBatchView { + Some(ConsumerEvents::Batch(EventBatchView { epoch, dropped_before, event_ids, - }) + })) } /// Records that this consumer has mapped geometry against the agents of `revision`. diff --git a/services/flysim/crates/fly-session/tests/publishing.rs b/services/flysim/crates/fly-session/tests/publishing.rs index 12f7af1..b5e7b93 100644 --- a/services/flysim/crates/fly-session/tests/publishing.rs +++ b/services/flysim/crates/fly-session/tests/publishing.rs @@ -14,8 +14,9 @@ use common::{Fixture, default_fixture, fixture, fly_a, fly_b, mode_fixture, with use fly_session::coordinator::{DESCRIPTOR_REVISION, Injections}; use fly_session::harness::{AgentSpec, ExecutionMode, HarnessConfig, Via}; use fly_session::publish::{ - ApplicationChannel, ConsumerOutcome, Delivery, EVENT_BATCH_DEPTH, EventOutbox, GET_SNAPSHOT, - PresentationConsumer, PublicationOutcome, TopicPolicy, check_publication, query_service, + ApplicationChannel, ConsumerEvents, ConsumerOutcome, Delivery, EVENT_BATCH_DEPTH, EventOutbox, + GET_SNAPSHOT, PresentationConsumer, PublicationOutcome, TopicPolicy, check_publication, + query_service, }; use fly_session::types::*; use serde_json::json; @@ -38,6 +39,7 @@ both_transports!( the_query_service_answers_reads_and_nothing_else, a_refused_snapshot_is_still_what_the_repair_path_answers, a_replacement_that_built_another_index_cannot_install_the_checkpoint, + a_malformed_event_batch_is_unreadable_and_not_the_end_of_the_stream, 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, @@ -1079,6 +1081,55 @@ async fn a_refused_snapshot_is_still_what_the_repair_path_answers(via: Via) { f.shutdown().await; } +/// An event batch a consumer cannot read is reported as unreadable, and the stream goes on. +/// +/// The two facts differ: a consumer that read "that one made no sense" as "there are no more +/// events" would stop reading a live stream, which is the silent default this module exists to +/// refuse. The malformed batch is published by a second publisher on the session's own address, +/// because a well-behaved session cannot produce one. +async fn a_malformed_event_batch_is_unreadable_and_not_the_end_of_the_stream(via: Via) { + let mut f = started(via).await; + let mut consumer = f.harness.consumer().await.expect("a consumer attaches"); + let events_topic = f.harness.coordinator.topics().events.clone(); + let intruder = f.harness.publisher().await.expect("a publishing client"); + + // A batch with no droppedBefore: readable JSON, unreadable as a batch. + intruder + .publish( + &events_topic, + object(json!({ + "sessionId": f.harness.config.session_id.as_str(), + "epoch": f.harness.config.epoch.as_str(), + "events": [], + })), + &[], + ) + .await + .expect("the malformed batch publishes"); + + match within("the malformed batch", consumer.take_events()).await { + Some(ConsumerEvents::Unreadable { detail }) => { + assert!( + detail.contains("droppedBefore"), + "the report names what it could not read: {detail}" + ); + } + other => panic!("a malformed batch must not read as {other:?}"), + } + + // The stream did not end: the next real batch still arrives and reads. + f.harness.coordinator.run(1).await.expect("a transition"); + match within("the next batch", consumer.take_events()).await { + Some(ConsumerEvents::Batch(batch)) => { + assert_eq!(batch.epoch, f.harness.config.epoch); + assert!(!batch.event_ids.is_empty(), "the transition produced events"); + } + other => panic!("expected a readable batch after the malformed one, got {other:?}"), + } + intruder.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;