//! The publication boundary of `publishing-v1`: what leaves the session, and what a refusal is. //! //! This is the PUBLISH-01 slice. It is an *internal* boundary on the *same* bus: there is no //! second transport, no gateway process, no codec and no show or tournament service. What it //! adds is the separation the contract asks for: //! //! ```text //! session..descriptor retained latest framework: what this composition is //! session..snapshots retained latest framework: the values of one committed boundary //! session..events bounded framework: scoped domain events, not a log //! .state retained latest application-owned, application-shaped //! .cues bounded application-owned, under a declared policy //! session..query RPC, read-only the repair path for a missed descriptor //! ``` //! //! Three rules decide everything below. //! //! 1. **A publication outcome is named.** [`PublicationOutcome`] is `Accepted`, //! `RefusedByObserver` or `Faulted`; nothing is dropped, retried or defaulted silently. //! `bus-v1` section 6 says plainly that "bounded event subscriptions can reject publication; //! latest spectator subscriptions cannot hold a required session transaction indefinitely", //! so a refusal is a thing the contract expects and this module counts, not an error to //! swallow. Only `BACKPRESSURE` is an observer's refusal. A store quota, a lost router or an //! unreadable payload is the session's own fault and fails the epoch. //! 2. **An observer never moves the world.** Publication happens after the committed boundary //! is established. A refusal changes no phase, takes no step and releases no handle, and a //! latest observation topic supersedes the refused value at the next boundary, so a slow or //! bounded observer costs its own delivery and one boundary of a stream whose contract is //! "latest" -- never a tick, never a stall. The repair path recovers the exact value. //! 3. **Agent state and media are one statement.** [`check_publication`] refuses a snapshot //! whose media does not belong to the boundary its agent state belongs to, before anything //! is published, so "future agent state with old media" is a failure rather than a frame. //! //! What this module deliberately does **not** contain: the approved public v2 wire schemas and //! the stage adapters that speak them. `implementation.md` sequences those after this slice and //! together with each other, and inventing a public byte format here would be exactly the //! unapproved contract that ordering exists to prevent. use std::collections::{BTreeMap, VecDeque}; use std::sync::{Arc, Mutex}; use serde_json::{Map, Value, json}; // `crate::types` is this crate's facade over the shared `fly-session-types` crate; the glob // keeps the contract's own names in sight instead of restating them. use crate::types::*; /// How many events one bounded batch may hold before the oldest are explicitly dropped. pub const EVENT_BATCH_DEPTH: usize = 64; /// The in-flight credits a framework consumer takes on an observation topic. pub const SPECTATOR_CREDITS: u32 = 2; // ---------------------------------------------------------------------------------------------- // Policy /// The delivery a topic is published under. Declared once, never inferred per message. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Delivery { /// One replaceable value. A consumer that falls behind loses intermediate values and /// nothing else; that loss is the contract of the topic, not an accident of load. LatestValue, /// A bounded batch. Nothing is coalesced: when the batch is full the oldest entries are /// dropped by an explicit count that travels with the next accepted batch. BoundedBatch { depth: usize }, } impl Delivery { /// Whether the router retains the last value for a late subscriber. /// /// A latest observation is retained, because a consumer that arrives mid-session must be /// able to start; a bounded event stream is not, because it is "not a durable log" /// (publishing-v1 section 2) and a retained tail would look like one. pub fn retained(self) -> flybus::Retained { match self { Delivery::LatestValue => flybus::Retained::Latest, Delivery::BoundedBatch { .. } => flybus::Retained::None, } } /// The subscription a consumer of this topic takes. pub fn subscription(self) -> flybus::SubscriptionConfig { match self { Delivery::LatestValue => flybus::SubscriptionConfig::latest() .in_flight(SPECTATOR_CREDITS) .replay(true), Delivery::BoundedBatch { depth } => { flybus::SubscriptionConfig::bounded().queued(depth as u32) } } } } /// One published address and the delivery it was declared under. #[derive(Clone, Debug)] pub struct TopicPolicy { pub topic: String, pub delivery: Delivery, } impl TopicPolicy { pub fn latest(topic: impl Into) -> TopicPolicy { TopicPolicy { topic: topic.into(), delivery: Delivery::LatestValue, } } pub fn bounded(topic: impl Into, depth: usize) -> TopicPolicy { TopicPolicy { topic: topic.into(), delivery: Delivery::BoundedBatch { depth }, } } } // ---------------------------------------------------------------------------------------------- // Outcomes /// What became of one publication. Every path through [`Publisher`] returns one of these. #[derive(Clone, Debug, PartialEq)] pub enum PublicationOutcome { /// The router admitted it. `replaced` counts the queued values it coalesced away, which /// is the only loss a latest subscriber can suffer and is reported, not hidden. Accepted { topic: String, topic_sequence: u64, subscribers: u64, replaced: u64, }, /// A subscriber refused it. `bus-v1` section 5 rejects the whole publish for a bounded /// subscriber's full queue, so this is an observer's doing: the world is untouched, the /// value is recoverable through the query service, and the offender is named. RefusedByObserver { topic: String, detail: String }, /// The session's own resource or identity fault. This one fails the epoch. Faulted { topic: String, detail: String }, } impl PublicationOutcome { pub fn topic(&self) -> &str { match self { PublicationOutcome::Accepted { topic, .. } | PublicationOutcome::RefusedByObserver { topic, .. } | PublicationOutcome::Faulted { topic, .. } => topic, } } pub fn is_accepted(&self) -> bool { matches!(self, PublicationOutcome::Accepted { .. }) } pub fn is_refused(&self) -> bool { matches!(self, PublicationOutcome::RefusedByObserver { .. }) } /// The domain error a *fault* carries, or `None` for an accepted or refused publication. /// /// The certainty is `none`: publication happens after the boundary is committed and /// mutates no participant, so a failed publish has changed nothing in the world. pub fn fault(&self) -> Option { match self { PublicationOutcome::Faulted { topic, detail } => Some(DomainError::new( ErrorCode::BackendFailure, format!("publishing {topic}: {detail}"), MutationCertainty::None, )), _ => None, } } fn from_bus(topic: &str, result: Result) -> Self { match result { Ok(receipt) => PublicationOutcome::Accepted { topic: topic.to_owned(), topic_sequence: receipt.topic_sequence, subscribers: receipt.subscribers, replaced: receipt.replaced, }, // The one code an observer can cause. Everything else is ours. Err(e) if e.code == flybus::ErrorCode::Backpressure => { PublicationOutcome::RefusedByObserver { topic: topic.to_owned(), detail: e.message, } } Err(e) => PublicationOutcome::Faulted { topic: topic.to_owned(), detail: format!("{:?}: {}", e.code, e.message), }, } } } /// Per-topic publication counters, for assertions and for an operator. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct TopicCounters { pub accepted: u64, pub refused: u64, pub faulted: u64, /// Values a latest subscriber's queue coalesced away, as the router reported them. pub replaced: u64, } /// What this session published and what happened to it. #[derive(Clone, Debug, Default)] pub struct Ledger { topics: BTreeMap, /// Events dropped from the bounded batch because the batch was full, cumulative. pub events_dropped: u64, /// Batches held for a later boundary because an observer refused them. pub events_held: u64, last: Option, } impl Ledger { pub fn counters(&self, topic: &str) -> TopicCounters { self.topics.get(topic).cloned().unwrap_or_default() } pub fn last(&self) -> Option<&PublicationOutcome> { self.last.as_ref() } /// Every refusal this session has met, by topic. pub fn refusals(&self) -> u64 { self.topics.values().map(|c| c.refused).sum() } fn record(&mut self, outcome: &PublicationOutcome) { let entry = self.topics.entry(outcome.topic().to_owned()).or_default(); match outcome { PublicationOutcome::Accepted { replaced, .. } => { entry.accepted += 1; entry.replaced += replaced; } PublicationOutcome::RefusedByObserver { .. } => entry.refused += 1, PublicationOutcome::Faulted { .. } => entry.faulted += 1, } self.last = Some(outcome.clone()); } } // ---------------------------------------------------------------------------------------------- // Published state and the repair path /// Everything the query service can answer from: the descriptor revisions this session has /// published, and the latest committed snapshot. /// /// "Latest retained descriptors accelerate startup; RPC querying remains the repair path" /// (publishing-v1 section 2). A consumer that meets a revision it does not hold asks here /// instead of guessing the shape of the values it is reading. #[derive(Clone, Debug, Default)] pub struct PublishedState { descriptors: BTreeMap, latest: Option, } impl PublishedState { pub fn descriptor(&self, revision: u64) -> Option<&SessionDescriptor> { self.descriptors.get(&revision) } pub fn newest_descriptor(&self) -> Option<&SessionDescriptor> { self.descriptors.values().next_back() } pub fn latest_snapshot(&self) -> Option<&CommittedSnapshot> { self.latest.as_ref() } pub fn revisions(&self) -> Vec { self.descriptors.keys().copied().collect() } } /// The shared handle the publisher writes and the query service reads. pub type SharedState = Arc>; fn lock(state: &SharedState) -> std::sync::MutexGuard<'_, PublishedState> { state .lock() .expect("the published state is never held across a panic") } // ---------------------------------------------------------------------------------------------- // The bounded event batch /// The pending event batch of `publishing-v1` section 6: bounded, explicit about loss. /// /// Events are "scoped domain events; not a durable log" (section 2), and "bus publish /// acceptance and delivery consumption are not durable acknowledgments" (section 6). So this /// keeps a bounded batch and says exactly what it could not keep: `droppedBefore` travels with /// the next accepted batch, so a subscriber reads a count rather than inferring a gap. What it /// never does is grow without bound, retry forever or forget quietly. #[derive(Clone, Debug, Default)] pub struct EventOutbox { depth: usize, pending: VecDeque<(u64, TaskEvent)>, dropped_since_accepted: u64, } impl EventOutbox { pub fn new(depth: usize) -> EventOutbox { EventOutbox { depth, pending: VecDeque::new(), dropped_since_accepted: 0, } } /// Adds this boundary's events, dropping the oldest if the batch is over its depth. /// Returns how many were dropped by this call. pub fn offer(&mut self, source_step: u64, events: &[TaskEvent]) -> u64 { for event in events { self.pending.push_back((source_step, event.clone())); } let mut dropped = 0; while self.pending.len() > self.depth { self.pending.pop_front(); dropped += 1; } self.dropped_since_accepted += dropped; dropped } pub fn is_empty(&self) -> bool { self.pending.is_empty() } pub fn len(&self) -> usize { self.pending.len() } pub fn dropped_since_accepted(&self) -> u64 { self.dropped_since_accepted } fn payload(&self, session_id: &Id, epoch: &Id) -> Map { let events: Vec = self .pending .iter() .map(|(source_step, event)| { let mut value = event.to_json(); if let Value::Object(map) = &mut value { map.insert("sourceStep".to_owned(), source_step.to_string().into()); } value }) .collect(); object(json!({ "sessionId": session_id.as_str(), "epoch": epoch.as_str(), "droppedBefore": self.dropped_since_accepted.to_string(), "events": Value::Array(events), })) } fn accepted(&mut self) { self.pending.clear(); self.dropped_since_accepted = 0; } } // ---------------------------------------------------------------------------------------------- // Coherence /// Refuses a snapshot whose media does not belong to the boundary its agent state belongs to. /// /// This is the "future agent state is never mixed with old media" rule, checked on the way out /// and again on the way in. Four things must agree: /// /// 1. Each published view's `producedStep` is exactly what its declared delay implies for this /// boundary. A frame from an older boundary is a refusal, not a substitution. /// 2. Each published audio chunk's sample range ends where the stream's next chunk begins, so /// a chunk from a previous transition cannot ride along under a new boundary. /// 3. Every referenced view and chunk has its attachment, and no attachment is present that /// the payload does not reference. An extra handle is an old boundary's frame. /// 4. The snapshot agrees with the descriptor it names: revision, agent set, port assignment, /// rate roles. pub fn check_publication( descriptor: &SessionDescriptor, snapshot: &CommittedSnapshot, attachments: &[(String, ArtifactRef)], audio_next_sample: &BTreeMap, ) -> DomainResult<()> { snapshot .validate_against(descriptor) .map_err(|e| coherence(format!("snapshot and descriptor disagree: {e}")))?; check_views(descriptor, snapshot)?; check_attachments(snapshot, attachments)?; for chunk in &snapshot.audio { let end = chunk .first_sample .checked_add(chunk.sample_frames) .ok_or_else(|| { coherence(format!( "audio stream {} overflows its sample position", chunk.stream_id )) })?; match audio_next_sample.get(&chunk.stream_id) { Some(next) if *next == end => {} Some(next) => { return Err(coherence(format!( "audio stream {} covers samples {}..{end} and this boundary ends at {next}", chunk.stream_id, chunk.first_sample ))); } None => { return Err(coherence(format!( "audio stream {} has no accepted position at this boundary", chunk.stream_id ))); } } } Ok(()) } /// Every published view comes from exactly the boundary its declared delay implies. pub fn check_views( descriptor: &SessionDescriptor, snapshot: &CommittedSnapshot, ) -> DomainResult<()> { let boundary = snapshot.scope.step; for view in &snapshot.views { let declared = descriptor .environment .views .iter() .find(|v| v.view_id == view.view_id) .ok_or_else(|| coherence(format!("view {} is not declared", view.view_id)))?; let want = declared.required_produced_step(boundary); if view.produced_step != want { return Err(coherence(format!( "view {} at boundary {boundary} was produced at {}, and its declared delay of {} requires {want}", view.view_id, view.produced_step, declared.observation_delay_steps ))); } } Ok(()) } /// The handles and the references are the same media: no missing frame, no extra one, and /// each handle is the artifact its reference names. /// /// The identity comparison is the half that matters: an attachment set that matches by *name* /// while one handle is the previous boundary's object is exactly "old media under new agent /// state", and only the `ArtifactRef` sees it. pub fn check_attachments( snapshot: &CommittedSnapshot, attachments: &[(String, ArtifactRef)], ) -> DomainResult<()> { let mut want: Vec<(String, &ArtifactRef)> = snapshot .views .iter() .map(|v| (crate::media::view_attachment(&v.view_id), &v.pixels)) .chain( snapshot .audio .iter() .map(|a| (crate::media::audio_attachment(&a.stream_id), &a.samples)), ) .collect(); want.sort_by(|a, b| a.0.cmp(&b.0)); let mut given: Vec<(String, &ArtifactRef)> = attachments.iter().map(|(n, r)| (n.clone(), r)).collect(); given.sort_by(|a, b| a.0.cmp(&b.0)); if want.len() != given.len() || want.iter().zip(&given).any(|(w, g)| w.0 != g.0) { let want: Vec<&String> = want.iter().map(|(n, _)| n).collect(); let given: Vec<&String> = given.iter().map(|(n, _)| n).collect(); return Err(coherence(format!( "the published handles {given:?} are not the media the snapshot references {want:?}" ))); } for ((name, reference), (_, handle)) in want.iter().zip(&given) { if reference != handle { return Err(coherence(format!( "the handle published as {name} is artifact {} generation {}, and the snapshot references {} generation {}", handle.artifact_id, handle.generation, reference.artifact_id, reference.generation ))); } } Ok(()) } fn coherence(message: impl std::fmt::Display) -> DomainError { // Nothing was published, so nothing downstream saw a mixed boundary. DomainError::new(ErrorCode::BufferInvalid, message, MutationCertainty::None) } // ---------------------------------------------------------------------------------------------- // The publisher /// The session's publication path. One bus client, three framework topics, named outcomes. pub struct Publisher { bus: flybus::Client, session_id: Id, epoch: Id, descriptor_topic: TopicPolicy, snapshot_topic: TopicPolicy, event_topic: TopicPolicy, /// The STATE-01 checkpoint stream. A stream of distinct facts, so it is a bounded /// delivery and never a latest value: a "committed" that replaced a "queued" would erase /// the distinction the durable commit rules are built on. checkpoint_topic: TopicPolicy, outbox: EventOutbox, state: SharedState, ledger: Ledger, sequence: u64, } impl Publisher { pub fn new( bus: flybus::Client, session_id: &Id, epoch: &Id, topics: &crate::coordinator::Topics, ) -> Publisher { Publisher { bus, session_id: session_id.clone(), epoch: epoch.clone(), descriptor_topic: TopicPolicy::latest(&topics.descriptor), snapshot_topic: TopicPolicy::latest(&topics.snapshots), event_topic: TopicPolicy::bounded(&topics.events, EVENT_BATCH_DEPTH), checkpoint_topic: TopicPolicy::bounded(&topics.checkpoints, EVENT_BATCH_DEPTH), outbox: EventOutbox::new(EVENT_BATCH_DEPTH), state: Arc::new(Mutex::new(PublishedState::default())), ledger: Ledger::default(), sequence: 0, } } /// The declared policies, in publication order. pub fn policies(&self) -> Vec { vec![ self.descriptor_topic.clone(), self.snapshot_topic.clone(), self.event_topic.clone(), self.checkpoint_topic.clone(), ] } pub fn ledger(&self) -> &Ledger { &self.ledger } pub fn state(&self) -> SharedState { Arc::clone(&self.state) } pub fn outbox(&self) -> &EventOutbox { &self.outbox } /// The next publication sequence, which is monotonic within this publisher incarnation. pub fn sequence(&self) -> u64 { self.sequence } pub fn incarnation(&self) -> String { self.bus.info().connection_id.clone() } /// Declares every framework topic under its policy. A topic already declared compatibly /// is accepted; a conflicting declaration is a fault here, not a silent reuse. pub async fn declare(&self) -> DomainResult<()> { for policy in self.policies() { self.bus .declare_topic(&policy.topic, policy.delivery.retained()) .await .map_err(|e| { DomainError::new( ErrorCode::BackendFailure, format!("declaring {}: {}", policy.topic, e.message), MutationCertainty::None, ) })?; } Ok(()) } /// Publishes a descriptor revision and records it as answerable by the query service. /// /// The revision is recorded even when the publication is refused: the repair path exists /// exactly for the consumer that did not receive it. pub async fn publish_descriptor( &mut self, descriptor: &SessionDescriptor, ) -> DomainResult { descriptor.validate().map_err(DomainError::invalid)?; { let mut state = lock(&self.state); if let Some(previous) = state.descriptor(descriptor.revision) && previous != descriptor { return Err(DomainError::before( ErrorCode::IdentityMismatch, format!( "descriptor revision {} was already published with another composition", descriptor.revision ), )); } state .descriptors .insert(descriptor.revision, descriptor.clone()); } let topic = self.descriptor_topic.topic.clone(); let outcome = PublicationOutcome::from_bus( &topic, self.bus .publish(&topic, object(descriptor.to_json()), &[]) .await, ); self.ledger.record(&outcome); Ok(outcome) } /// Publishes one committed boundary with the media handles it references. /// /// The coherence check runs first, so a snapshot that mixes boundaries never reaches a /// subscriber; the sequence advances only on a publication the router accepted, so /// "monotonic within publisherIncarnation" counts published values and not attempts. pub async fn publish_snapshot( &mut self, descriptor: &SessionDescriptor, snapshot: &CommittedSnapshot, attachments: &[(String, flybus::Artifact)], audio_next_sample: &BTreeMap, ) -> DomainResult { let named: Vec<(String, ArtifactRef)> = attachments .iter() .map(|(name, artifact)| (name.clone(), artifact.reference().clone())) .collect(); check_publication(descriptor, snapshot, &named, audio_next_sample)?; let refs: Vec<(&str, &flybus::Artifact)> = attachments.iter().map(|(n, a)| (n.as_str(), a)).collect(); let topic = self.snapshot_topic.topic.clone(); let outcome = PublicationOutcome::from_bus( &topic, self.bus .publish(&topic, object(snapshot.to_json()), &refs) .await, ); // 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) } /// Offers this boundary's events to the bounded batch and publishes what it holds. /// /// An empty batch publishes nothing and is not an outcome. A refused batch is held for the /// next boundary and counted; what the depth pushed out is counted too and travels with /// the next accepted batch as `droppedBefore`. pub async fn publish_events( &mut self, source_step: u64, events: &[TaskEvent], ) -> Option { let dropped = self.outbox.offer(source_step, events); self.ledger.events_dropped += dropped; if self.outbox.is_empty() { return None; } let payload = self.outbox.payload(&self.session_id, &self.epoch); let topic = self.event_topic.topic.clone(); let outcome = PublicationOutcome::from_bus(&topic, self.bus.publish(&topic, payload, &[]).await); match &outcome { PublicationOutcome::Accepted { .. } => self.outbox.accepted(), PublicationOutcome::RefusedByObserver { .. } => self.ledger.events_held += 1, PublicationOutcome::Faulted { .. } => {} } self.ledger.record(&outcome); Some(outcome) } } impl Publisher { /// Publishes one checkpoint fact on the checkpoint stream. /// /// It goes through the same named outcomes as everything else: a durable-commit fact that /// an observer refuses is counted and does not fail the session, because the durable /// acknowledgment is the store's, not the subscriber's -- "bus publish acceptance and /// delivery consumption are not durable acknowledgments" (publishing-v1 section 6). pub async fn publish_checkpoint(&mut self, payload: Map) -> PublicationOutcome { let topic = self.checkpoint_topic.topic.clone(); let outcome = PublicationOutcome::from_bus(&topic, self.bus.publish(&topic, payload, &[]).await); self.ledger.record(&outcome); outcome } } // ---------------------------------------------------------------------------------------------- // The query service: the repair path /// The read-only service name a session answers descriptor queries on. pub fn query_service(session_id: &Id) -> String { format!("session.{session_id}.query") } /// `Session.GetDescriptor`: one published revision, or the newest. pub const GET_DESCRIPTOR: &str = "Session.GetDescriptor"; /// `Session.GetSnapshot`: the latest committed snapshot, without its media handles. pub const GET_SNAPSHOT: &str = "Session.GetSnapshot"; /// A running query service. /// /// It is the repair path of `publishing-v1` section 2 and nothing else: two read methods over /// state the publisher already published. It takes no parameters that select a participant, it /// mutates nothing, and it is not a controller API -- there is no method here that could /// 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 { pub async fn start( client: flybus::Client, session_id: &Id, state: SharedState, ) -> Result { let name = query_service(session_id); let mut service = client .register(&name, flybus::ServiceConfig::default()) .await?; let worker_id = session_id.clone(); // The bus connection id names this publisher incarnation. It has to be an `Id` to // travel in a reply, and a connection id that is not one is a refusal here rather // 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(); let responder = request.responder(); let parsed = SessionRpcRequest::from_json(&Value::Object(request.payload().clone())); drop(request); let outcome = match parsed { Ok(parsed) => answer(&parsed, &method, &worker_id, &incarnation, &state), Err(e) => failure_outcome( &DomainRequestId::from_serial(0), &worker_id, &incarnation, None, DomainError::invalid(format!("{method}: {e}")), ), }; if responder.reply(outcome.to_outcome(), &[]).await.is_err() { undelivered.fetch_add(1, std::sync::atomic::Ordering::SeqCst); } } }); 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. pub fn stop(self) { drop(self); } } impl Drop for QueryService { /// A dropped session leaves no task reading a service it no longer answers for. fn drop(&mut self) { self.task.abort(); } } fn answer( request: &SessionRpcRequest, method: &str, worker_id: &Id, incarnation: &Id, state: &SharedState, ) -> SessionRpcOutcome { let result = match method { GET_DESCRIPTOR => get_descriptor(request, state), GET_SNAPSHOT => get_snapshot(request, state), other => Err(DomainError::before( ErrorCode::Unsupported, format!("{other} is not a method of the session query service"), )), }; match result { Ok(result) => success_outcome( &request.request_id, worker_id, incarnation, request.scope.clone(), result, ), Err(error) => failure_outcome( &request.request_id, worker_id, incarnation, request.scope.clone(), error, ), } } fn get_descriptor( request: &SessionRpcRequest, state: &SharedState, ) -> DomainResult> { let wanted = match request.params.get("revision") { None | Some(Value::Null) => None, Some(Value::String(text)) => Some( text.parse::() .map_err(|_| DomainError::invalid("revision is a decimal U64 string"))?, ), Some(_) => return Err(DomainError::invalid("revision is a decimal U64 string")), }; let state = lock(state); let descriptor = match wanted { Some(revision) => state.descriptor(revision).ok_or_else(|| { // A revision this session never published is an answer, not an empty result. DomainError::before( ErrorCode::IdentityMismatch, format!( "descriptor revision {revision} was never published; this session has {:?}", state.revisions() ), ) })?, None => state.newest_descriptor().ok_or_else(|| { DomainError::before(ErrorCode::InvalidPhase, "no descriptor has been published") })?, }; Ok(object(descriptor.to_json())) } fn get_snapshot( request: &SessionRpcRequest, state: &SharedState, ) -> DomainResult> { if !request.params.as_object().is_some_and(Map::is_empty) { return Err(DomainError::invalid( "Session.GetSnapshot takes no parameters", )); } let state = lock(state); let snapshot = state.latest_snapshot().ok_or_else(|| { DomainError::before(ErrorCode::InvalidPhase, "no snapshot has been published") })?; Ok(object(snapshot.to_json())) } // ---------------------------------------------------------------------------------------------- // Application-owned state and cues /// An application's own publication channel: its schema, its topics, its delivery policy. /// /// The framework supplies the channel and nothing about what travels on it. "Application state /// carries whatever the experience needs ... It is developed with its presentation, not forced /// into a framework-wide show state/tournament schema" (publishing-v1 section 4). So the topics /// are named by the application, the values are `TypedValue`s under the application's own /// namespaced schema, and this module never looks inside one. There is no director here, no /// bracket, no cast and no game. pub struct ApplicationChannel { bus: flybus::Client, state_topic: TopicPolicy, cue_topic: TopicPolicy, ledger: Ledger, revision: u64, } impl ApplicationChannel { /// `prefix` is the application's own address root, for example `app.counter`. pub fn new(bus: flybus::Client, prefix: &str, cue_depth: usize) -> ApplicationChannel { ApplicationChannel { bus, state_topic: TopicPolicy::latest(format!("{prefix}.state")), cue_topic: TopicPolicy::bounded(format!("{prefix}.cues"), cue_depth), ledger: Ledger::default(), revision: 0, } } pub fn state_topic(&self) -> &str { &self.state_topic.topic } pub fn cue_topic(&self) -> &str { &self.cue_topic.topic } pub fn ledger(&self) -> &Ledger { &self.ledger } pub async fn declare(&self) -> DomainResult<()> { for policy in [&self.state_topic, &self.cue_topic] { self.bus .declare_topic(&policy.topic, policy.delivery.retained()) .await .map_err(|e| { DomainError::new( ErrorCode::BackendFailure, format!("declaring {}: {}", policy.topic, e.message), MutationCertainty::None, ) })?; } Ok(()) } /// Publishes application state at a committed boundary it names. /// /// The boundary travels with it because a consumer combines this with the framework /// snapshot, and cross-topic ordering is not guaranteed: "a subscriber receiving an unknown /// descriptor revision must fetch it ... or buffer a bounded number of snapshots, not infer /// shape" (publishing-v1 section 2). A named boundary is how the two are joined. pub async fn publish_state(&mut self, boundary: u64, value: &TypedValue) -> PublicationOutcome { self.revision += 1; let payload = object(json!({ "boundary": boundary.to_string(), "revision": self.revision.to_string(), "state": value.to_json(), })); let topic = self.state_topic.topic.clone(); let outcome = PublicationOutcome::from_bus(&topic, self.bus.publish(&topic, payload, &[]).await); self.ledger.record(&outcome); outcome } /// Publishes a presentation cue. Cues are presentation data: "presentation cues may be /// immediate; simulation effects apply at declared boundaries" (section 7). Nothing here /// reaches a worker, a controller or the world. pub async fn publish_cue( &mut self, boundary: u64, kind: &Id, value: &TypedValue, ) -> PublicationOutcome { let payload = object(json!({ "boundary": boundary.to_string(), "kind": kind.as_str(), "cue": value.to_json(), })); let topic = self.cue_topic.topic.clone(); let outcome = PublicationOutcome::from_bus(&topic, self.bus.publish(&topic, payload, &[]).await); self.ledger.record(&outcome); outcome } } // ---------------------------------------------------------------------------------------------- // The fake multi-agent consumer /// One agent's committed values as a consumer reads them. /// /// A presentation consumer is a *multi-agent* consumer: one snapshot carries the whole /// composition, so there is no per-fly stream to join and no "current fly" to be stale. #[derive(Clone, Debug, PartialEq)] pub struct AgentView { pub agent_id: Id, /// The index the descriptor says this agent's rates and geometry belong to. A consumer /// that maps anything spatial compares this, not `neuronCount`. pub index_digest: Digest, pub brain_ticks: u64, pub rates: Vec<(Id, f64)>, /// The decision of the transition that *ended* at this boundary, null only at boundary 0. pub decision: Option, pub controls: Option, } /// One committed boundary as a consumer reads it, with the media handles it still owns. pub struct SnapshotView { pub boundary: u64, pub descriptor_revision: u64, pub sequence: u64, /// How many undelivered snapshots the router coalesced away before this one. pub replaced: u64, pub agents: Vec, pub views: Vec, pub audio: Vec, /// The extracted handles, held after the message is dropped. pub artifacts: BTreeMap, } impl SnapshotView { pub fn agent(&self, agent_id: &Id) -> Option<&AgentView> { self.agents.iter().find(|a| a.agent_id == *agent_id) } } /// One bounded event batch as a consumer reads it. #[derive(Clone, Debug, PartialEq)] pub struct EventBatchView { pub epoch: Id, /// How many events the publisher's bounded batch dropped before this one. A count, never /// a gap the consumer has to infer. pub dropped_before: u64, 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 { /// A descriptor this consumer now holds, and the agents it describes. Composition { revision: u64, agents: Vec }, /// A snapshot whose descriptor revision this consumer holds and which agrees with it. Read { boundary: u64, agents: Vec }, /// A snapshot naming a descriptor revision this consumer has never seen. Nothing is /// inferred from it: the consumer repairs through the query service and reads it again. UnknownRevision { revision: u64 }, /// The composition changed under an agent this consumer had already mapped: the same /// neuron count, another index. Visible, named, and never silently remapped. IndexChanged { agent_id: Id, from: Digest, to: Digest, }, /// The snapshot does not agree with the descriptor it names, or its media does not belong /// to its boundary. Incoherent { detail: String }, } /// A presentation-side consumer of one session, over the same bus. /// /// It is a regular bus subscriber (publishing-v1 section 5): latest subscriptions with finite /// credits on the observation topics, a bounded subscription on events, and one read-only RPC /// for repair. It is deliberately *not* a gateway: it resolves no artifact into a browser /// transport, encodes nothing and holds no private owner token on anyone's behalf. pub struct PresentationConsumer { bus: flybus::Client, query: String, descriptors: flybus::Subscription, snapshots: flybus::Subscription, events: flybus::Subscription, held: Vec, cached: BTreeMap, /// The index digest this consumer has mapped geometry against, per agent. mapped: BTreeMap, /// Where each audio stream's last accepted chunk ended. audio_end: BTreeMap, last: Option, repairs: u64, serial: u64, coalesced: u64, } impl PresentationConsumer { pub async fn attach( bus: flybus::Client, session_id: &Id, topics: &crate::coordinator::Topics, ) -> Result { let descriptors = bus .subscribe(&topics.descriptor, Delivery::LatestValue.subscription()) .await?; let snapshots = bus .subscribe(&topics.snapshots, Delivery::LatestValue.subscription()) .await?; let events = bus .subscribe( &topics.events, Delivery::BoundedBatch { depth: EVENT_BATCH_DEPTH, } .subscription(), ) .await?; Ok(PresentationConsumer { bus, query: query_service(session_id), descriptors, snapshots, events, held: Vec::new(), cached: BTreeMap::new(), mapped: BTreeMap::new(), audio_end: BTreeMap::new(), last: None, repairs: 0, serial: 0, coalesced: 0, }) } /// How many descriptor revisions this consumer holds. pub fn revisions(&self) -> Vec { self.cached.keys().copied().collect() } pub fn descriptor(&self, revision: u64) -> Option<&SessionDescriptor> { self.cached.get(&revision) } /// How many times this consumer had to ask the query service for a descriptor. pub fn repairs(&self) -> u64 { self.repairs } /// How many snapshots the router coalesced away in this consumer's own queue. pub fn coalesced(&self) -> u64 { self.coalesced } pub fn last(&self) -> Option<&SnapshotView> { self.last.as_ref() } /// Takes the next descriptor from the retained-latest topic. /// /// A composition that moves an agent this consumer has already mapped geometry against is /// reported here and *not* cached: remapping silently is the one thing a consumer holding /// a spatial mapping must not do. (publishing-v1 section 3) pub async fn take_descriptor(&mut self) -> Option { let message = self.descriptors.next().await?; let descriptor = match SessionDescriptor::from_json(&Value::Object(message.payload().clone())) { Ok(descriptor) => descriptor, Err(e) => { return Some(ConsumerOutcome::Incoherent { detail: e.to_string(), }); } }; drop(message); for agent in &descriptor.agents { if let Some(mapped) = self.mapped.get(&agent.agent_id) && *mapped != agent.index_digest { return Some(ConsumerOutcome::IndexChanged { agent_id: agent.agent_id.clone(), from: mapped.clone(), to: agent.index_digest.clone(), }); } } let revision = descriptor.revision; let agents = descriptor .agents .iter() .map(|a| a.agent_id.clone()) .collect(); self.cached.insert(revision, descriptor); Some(ConsumerOutcome::Composition { revision, agents }) } /// Asks the query service for a revision this consumer does not hold. /// /// This is the repair path, and it is an ordinary RPC on the same bus. A revision the /// session never published comes back as a named error rather than an empty answer. pub async fn repair(&mut self, revision: u64) -> DomainResult { self.serial += 1; self.repairs += 1; let request = SessionRpcRequest { request_id: DomainRequestId::from_serial(self.serial), scope: None, params: json!({ "revision": revision.to_string() }), }; let mut pending = self .bus .call( &self.query, None, GET_DESCRIPTOR, object(request.to_json()), &[], ) .await .map_err(|e| { DomainError::new( ErrorCode::BackendFailure, e.message, MutationCertainty::None, ) })?; let result = pending.result().await.map_err(|e| { DomainError::new( ErrorCode::BackendFailure, e.message, MutationCertainty::None, ) })?; let outcome = SessionRpcOutcome::from_json(&Value::Object(result.outcome().clone())) .map_err(DomainError::invalid)?; drop(result); let value = outcome_result(&outcome)?; let descriptor = SessionDescriptor::from_json(value).map_err(DomainError::invalid)?; if descriptor.revision != revision { return Err(DomainError::before( ErrorCode::IdentityMismatch, "the query service answered with another revision", )); } self.cached.insert(revision, descriptor); Ok(revision) } /// Takes the next committed snapshot and reads it against the descriptor it names. pub async fn take_snapshot(&mut self) -> Option { let message = self.snapshots.next().await?; self.coalesced += message.replaced(); Some(self.read(message)) } /// Takes a snapshot without reading or releasing it, which is what a viewer that stopped /// rendering does. Its credits run out; nothing else in the session notices. pub async fn hold_snapshot(&mut self) -> bool { match self.snapshots.next().await { Some(message) => { self.coalesced += message.replaced(); self.held.push(message); true } None => false, } } /// Holds a snapshot if one is queued right now, and answers `false` if none is. /// /// The bounded form: a test that means "take whatever is there" must not be able to /// block on a stream that is deliberately not producing. pub fn try_hold_snapshot(&mut self) -> bool { match self.snapshots.try_next() { Some(message) => { self.coalesced += message.replaced(); self.held.push(message); true } None => false, } } /// Reads a snapshot if one is queued right now. pub fn try_take_snapshot(&mut self) -> Option { let message = self.snapshots.try_next()?; self.coalesced += message.replaced(); Some(self.read(message)) } /// Releases everything this consumer was holding, returning its credits. pub fn release(&mut self) { self.held.clear(); } pub fn held(&self) -> usize { self.held.len() } /// Takes the next bounded event batch. /// /// `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 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()) 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(ConsumerEvents::Batch(EventBatchView { epoch, dropped_before, event_ids, })) } /// Records that this consumer has mapped geometry against the agents of `revision`. /// /// A consumer that never calls this never claims a mapping, and a later index change is /// simply a new descriptor. One that does claim it gets [`ConsumerOutcome::IndexChanged`] /// when the composition moves under it. pub fn map_geometry(&mut self, revision: u64) -> DomainResult<()> { let descriptor = self.cached.get(&revision).ok_or_else(|| { DomainError::before( ErrorCode::IdentityMismatch, format!("revision {revision} is not held by this consumer"), ) })?; for agent in &descriptor.agents { self.mapped .insert(agent.agent_id.clone(), agent.index_digest.clone()); } Ok(()) } pub fn mapped_index(&self, agent_id: &Id) -> Option<&Digest> { self.mapped.get(agent_id) } fn read(&mut self, message: flybus::Message) -> ConsumerOutcome { let sequence = message.topic_sequence(); let replaced = message.replaced(); let snapshot = match CommittedSnapshot::from_json(&Value::Object(message.payload().clone())) { Ok(snapshot) => snapshot, Err(e) => { return ConsumerOutcome::Incoherent { detail: e.to_string(), }; } }; let Some(descriptor) = self.cached.get(&snapshot.descriptor_revision).cloned() else { // Not an error and not a guess: the consumer buffers nothing and infers nothing, // it repairs. (publishing-v1 section 2) return ConsumerOutcome::UnknownRevision { revision: snapshot.descriptor_revision, }; }; for agent in &descriptor.agents { if let Some(mapped) = self.mapped.get(&agent.agent_id) && *mapped != agent.index_digest { return ConsumerOutcome::IndexChanged { agent_id: agent.agent_id.clone(), from: mapped.clone(), to: agent.index_digest.clone(), }; } } let names: Vec = message.attachment_names().map(str::to_owned).collect(); let mut handles = Vec::new(); for name in &names { match message.artifact(name) { Ok(artifact) => handles.push((name.clone(), artifact.reference().clone())), Err(e) => { return ConsumerOutcome::Incoherent { detail: format!("attachment {name}: {}", e.message), }; } } } if let Err(e) = check_received(&descriptor, &snapshot, &self.audio_end, &handles) { return ConsumerOutcome::Incoherent { detail: e.to_string(), }; } // The renderer keeps its handles after the delivery is gone; the bytes stay alive // because the extracted handle owns them. (publishing-v1 section 5) let mut artifacts = BTreeMap::new(); for name in &names { match message.artifact(name) { Ok(artifact) => { artifacts.insert(name.clone(), artifact); } Err(e) => { return ConsumerOutcome::Incoherent { detail: format!("attachment {name}: {}", e.message), }; } } } drop(message); for chunk in &snapshot.audio { self.audio_end.insert( chunk.stream_id.clone(), chunk.first_sample + chunk.sample_frames, ); } let agents: Vec = snapshot .agents .iter() .map(|agent| { let declared = descriptor .agents .iter() .find(|a| a.agent_id == agent.agent_id) .expect("validate_against checked the agent set"); AgentView { agent_id: agent.agent_id.clone(), index_digest: declared.index_digest.clone(), brain_ticks: agent.telemetry.brain_ticks, rates: agent .telemetry .rates .iter() .map(|r| (r.role_id.clone(), r.hz)) .collect(), decision: agent.selected_decision.clone(), controls: agent.applied_controls.clone(), } }) .collect(); let outcome = ConsumerOutcome::Read { boundary: snapshot.scope.step, agents: agents.iter().map(|a| a.agent_id.clone()).collect(), }; self.last = Some(SnapshotView { boundary: snapshot.scope.step, descriptor_revision: snapshot.descriptor_revision, sequence, replaced, agents, views: snapshot.views.clone(), audio: snapshot.audio.clone(), artifacts, }); outcome } } /// The receiving half of the coherence rule. /// /// A consumer does not trust that the publisher checked: it checks the same statement from the /// other side, against the audio position it last accepted rather than the one the publisher /// holds. That is what catches a chunk from an earlier transition arriving under a later /// boundary, which is the one shape of "old media" a publisher-side check cannot see. pub fn check_received( descriptor: &SessionDescriptor, snapshot: &CommittedSnapshot, audio_end: &BTreeMap, attachments: &[(String, ArtifactRef)], ) -> DomainResult<()> { snapshot .validate_against(descriptor) .map_err(|e| coherence(format!("snapshot and descriptor disagree: {e}")))?; check_views(descriptor, snapshot)?; check_attachments(snapshot, attachments)?; for chunk in &snapshot.audio { match audio_end.get(&chunk.stream_id) { // A stream this consumer has not heard yet, or one that says it skipped: both are // declared states, not assumptions. None => {} Some(_) if chunk.discontinuity => {} // Forward is the latest subscription's own contract: a consumer that fell behind // was told so by `replaced`, and the boundaries in between are values it chose // not to receive. Backwards or overlapping is old media under a new boundary, // which no delivery policy explains. Some(end) if chunk.first_sample >= *end => {} Some(end) => { return Err(coherence(format!( "audio stream {} starts at {} and the last chunk this consumer read ended at {end}", chunk.stream_id, chunk.first_sample ))); } } } Ok(()) }