use std::collections::VecDeque;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use zenoh::bytes::Encoding;
use zenoh::key_expr::OwnedKeyExpr;
use zenoh::sample::Sample;
use crate::RetiredTimelines;
use crate::abi::{CodecId, encoding_string, parse_encoding_string};
use crate::codec::{Codec, MessagePack};
use crate::contract::{
CommandContract, ContractBody, DiagnosticContract, MeasurementContract, StateContract,
WorldClockContract,
};
use crate::error::{BusError, Result};
use crate::identity::TimelineId;
use crate::metadata::BusMetadata;
use crate::query::{QueryError, QueryFailure};
use crate::runtime_metrics::RuntimeMetricHandle;
use crate::session::Bus;
use crate::session::OUTBOUND_CAPACITY;
use crate::time::{CaptureStamp, LocalInstant, RobotInstant, TimeWindow};
use crate::topic::{AskQuery, Publish, Subscribe, Topic};
pub const DEFAULT_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
const PENDING_TIMELINE_CAPACITY: usize = 4;
mod sealed {
pub trait Sealed {}
}
pub trait StepStamp: sealed::Sealed {
fn instant(&self) -> RobotInstant;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct StepToken {
at: RobotInstant,
}
impl StepToken {
#[doc(hidden)]
pub const fn __mint(at: RobotInstant) -> Self {
StepToken { at }
}
}
impl sealed::Sealed for StepToken {}
impl StepStamp for StepToken {
fn instant(&self) -> RobotInstant {
self.at
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WorldStepToken {
at: RobotInstant,
}
impl sealed::Sealed for WorldStepToken {}
impl StepStamp for WorldStepToken {
fn instant(&self) -> RobotInstant {
self.at
}
}
pub struct TimelineAuthority {
timeline: TimelineId,
}
static TIMELINE_AUTHORITY_HELD: AtomicBool = AtomicBool::new(false);
impl TimelineAuthority {
#[doc(hidden)]
pub fn __mint(timeline: TimelineId) -> Result<Self> {
if TIMELINE_AUTHORITY_HELD.swap(true, Ordering::AcqRel) {
return Err(BusError::Namespace(
"a second timeline authority was requested; exactly one participant may own a \
timeline"
.to_string(),
));
}
Ok(TimelineAuthority { timeline })
}
pub const fn timeline(&self) -> TimelineId {
self.timeline
}
pub fn replace_timeline(&mut self, timeline: TimelineId) {
self.timeline = timeline;
}
pub const fn completed_step(&self, ticks: u64) -> WorldStepToken {
WorldStepToken {
at: RobotInstant::new(self.timeline, ticks),
}
}
}
impl Drop for TimelineAuthority {
fn drop(&mut self) {
TIMELINE_AUTHORITY_HELD.store(false, Ordering::Release);
}
}
struct Outbox<B> {
bus: Bus,
key: String,
metric: RuntimeMetricHandle,
_body: PhantomData<fn() -> B>,
}
impl<B> Clone for Outbox<B> {
fn clone(&self) -> Self {
Outbox {
bus: self.bus.clone(),
key: self.key.clone(),
metric: self.metric.clone(),
_body: PhantomData,
}
}
}
impl<B: ContractBody> Outbox<B> {
fn new(bus: Bus, topic: &Topic<Publish<B>>) -> Result<Self> {
let topic_key = topic.publish_key()?;
let metric = bus
.runtime_metrics()
.register_outbound(topic_key, OUTBOUND_CAPACITY);
let key = bus.full_key(topic_key);
Ok(Outbox {
bus,
key,
metric,
_body: PhantomData,
})
}
fn emit(&self, produced_at: Option<TimeWindow>, body: B) -> Result<()> {
let payload = MessagePack::encode(&body)?;
let metadata = self.bus.metadata(produced_at);
let encoding = encoding_string(MessagePack::ID);
self.bus.enqueue(
self.key.clone(),
encoding,
metadata.encode(),
payload,
self.metric.clone(),
)
}
}
macro_rules! role_publisher {
($name:ident, $bound:ident, $doc:literal) => {
#[doc = $doc]
pub struct $name<B: $bound>(Outbox<B>);
impl<B: $bound> Clone for $name<B> {
fn clone(&self) -> Self {
$name(self.0.clone())
}
}
impl<B: $bound> $name<B> {
#[doc(hidden)]
pub fn new(bus: Bus, topic: &Topic<Publish<B>>) -> Result<Self> {
Ok($name(Outbox::new(bus, topic)?))
}
}
};
}
role_publisher!(
StatePublisher,
StateContract,
"Publishes state at a logical step.\n\nThe step instant comes from a \
framework-minted [`StepToken`] or [`WorldStepToken`], so a participant \
cannot publish state at a time it did not reach. Non-blocking, so it is \
safe to call from the step loop (D35/D43e). The framework's own \
world-clock contract is deliberately NOT a `StateContract` and so cannot \
be named here; see [`WorldClockPublisher`]."
);
role_publisher!(
MeasurementPublisher,
MeasurementContract,
"Publishes a sensor observation with its capture stamp.\n\nThe driver owns \
mapping its device clock into robot time - including reset, drift, \
wraparound, batching, and exposure-versus-readout semantics - and says so \
honestly through [`CaptureStamp`], which can represent an untranslated \
capture rather than inventing an instant."
);
role_publisher!(
CommandPublisher,
CommandContract,
"Sends a command.\n\nA command is a request, not an observation: it \
expresses no robot time. The owning service stamps its own observation and \
applies the result at a logical step."
);
role_publisher!(
DiagnosticPublisher,
DiagnosticContract,
"Publishes an output that describes the participant rather than the world \
(health, logs, runtime evidence). It expresses no robot time."
);
pub struct WorldClockPublisher<B: WorldClockContract>(Outbox<B>);
impl<B: WorldClockContract> Clone for WorldClockPublisher<B> {
fn clone(&self) -> Self {
WorldClockPublisher(self.0.clone())
}
}
impl<B: StateContract> StatePublisher<B> {
pub fn publish(&self, step: &impl StepStamp, body: B) -> Result<()> {
self.0.emit(Some(TimeWindow::exact(step.instant())), body)
}
}
impl<B: WorldClockContract> WorldClockPublisher<B> {
#[doc(hidden)]
pub fn __mint(bus: Bus, topic: &Topic<Publish<B>>) -> Result<Self> {
Ok(WorldClockPublisher(Outbox::new(bus, topic)?))
}
pub fn publish(&self, step: &impl StepStamp, body: B) -> Result<()> {
self.0.emit(Some(TimeWindow::exact(step.instant())), body)
}
}
impl<B: MeasurementContract> MeasurementPublisher<B> {
pub fn publish(&self, stamp: CaptureStamp, body: B) -> Result<()> {
self.0.emit(stamp.into_window(), body)
}
}
impl<B: CommandContract> CommandPublisher<B> {
pub fn send(&self, body: B) -> Result<()> {
self.0.emit(None, body)
}
}
impl<B: DiagnosticContract> DiagnosticPublisher<B> {
pub fn publish(&self, body: B) -> Result<()> {
self.0.emit(None, body)
}
}
pub struct Querier<Req, Resp> {
bus: Bus,
key: String,
timeout: Duration,
_p: PhantomData<fn() -> (Req, Resp)>,
}
impl<Req, Resp> Clone for Querier<Req, Resp> {
fn clone(&self) -> Self {
Querier {
bus: self.bus.clone(),
key: self.key.clone(),
timeout: self.timeout,
_p: PhantomData,
}
}
}
impl<Req, Resp> Querier<Req, Resp>
where
Req: ContractBody,
Resp: ContractBody,
{
#[doc(hidden)]
pub fn new(bus: Bus, topic: &Topic<AskQuery<Req, Resp>>, timeout: Duration) -> Result<Self> {
let key = bus.full_key(topic.publish_key()?);
Ok(Querier {
bus,
key,
timeout,
_p: PhantomData,
})
}
pub async fn query(&self, request: Req) -> std::result::Result<Resp, QueryError> {
let payload =
MessagePack::encode(&request).map_err(|e| QueryError::Protocol(e.to_string()))?;
let metadata = self.bus.metadata(None);
let key = OwnedKeyExpr::new(self.key.clone())
.map_err(|e| QueryError::Protocol(format!("invalid query key '{}': {e}", self.key)))?;
let replies = self
.bus
.session()
.get(key)
.payload(payload)
.encoding(Encoding::from(encoding_string(MessagePack::ID)))
.attachment(metadata.encode())
.target(zenoh::query::QueryTarget::All)
.consolidation(zenoh::query::ConsolidationMode::None)
.await
.map_err(|e| QueryError::Protocol(e.to_string()))?;
let deadline = tokio::time::Instant::now() + self.timeout;
let mut outcome: Option<std::result::Result<Resp, QueryError>> = None;
loop {
match tokio::time::timeout_at(deadline, replies.recv_async()).await {
Ok(Ok(reply)) => {
if outcome.is_some() {
return Err(QueryError::TooManyResponders);
}
outcome = Some(decode_reply_result::<Resp>(reply.into_result()));
}
Ok(Err(_)) => break, Err(_elapsed) => {
return outcome.unwrap_or_else(|| {
Err(QueryError::Timeout(QueryFailure::deadline_exceeded(
"query deadline exceeded",
)))
});
}
}
}
outcome.unwrap_or(Err(QueryError::Unavailable))
}
}
fn decode_reply_result<Resp: ContractBody>(
result: std::result::Result<Sample, zenoh::query::ReplyError>,
) -> std::result::Result<Resp, QueryError> {
match result {
Ok(sample) => decode_reply::<Resp>(&sample),
Err(reply_error) => {
let bytes = reply_error.payload().to_bytes();
match crate::query::QueryFailure::decode(bytes.as_ref()) {
Ok(failure) => Err(QueryError::Server(failure)),
Err(e) => Err(QueryError::Protocol(format!("malformed error reply: {e}"))),
}
}
}
}
fn decode_reply<Resp: ContractBody>(sample: &Sample) -> std::result::Result<Resp, QueryError> {
match decode_sample::<Resp>(sample, Resp::TOPIC) {
Ok((body, _)) => Ok(body),
Err(e) => Err(QueryError::Decode(e.to_string())),
}
}
#[derive(Clone, Debug)]
pub struct Observed<B> {
pub body: B,
pub metadata: BusMetadata,
pub observed_at: LocalInstant,
}
impl<B> Observed<B> {
pub fn timeline(&self) -> Option<TimelineId> {
self.metadata.produced_at.map(TimeWindow::timeline)
}
pub fn age(&self, now: LocalInstant) -> Duration {
now.saturating_duration_since(self.observed_at)
}
}
pub struct Latest<B> {
state: Arc<Mutex<LatestState<B>>>,
metric: RuntimeMetricHandle,
_guard: Arc<SubscriptionGuard>,
}
struct LatestState<B> {
active_timeline: Option<TimelineId>,
observed: Option<Arc<Observed<B>>>,
pending: VecDeque<Arc<Observed<B>>>,
retired_timelines: RetiredTimelines,
}
enum LatestIngest {
Active {
overwrote: bool,
},
Pending {
timeline: TimelineId,
new_timeline: bool,
filtered: u64,
},
Filtered,
}
impl<B> LatestState<B> {
fn ingest(&mut self, observed: Observed<B>) -> LatestIngest {
let timeline = observed.timeline();
let observed = Arc::new(observed);
let (Some(timeline), Some(active_timeline)) = (timeline, self.active_timeline) else {
return LatestIngest::Active {
overwrote: self.observed.replace(observed).is_some(),
};
};
if timeline == active_timeline {
return LatestIngest::Active {
overwrote: self.observed.replace(observed).is_some(),
};
}
if self.retired_timelines.contains(timeline) {
return LatestIngest::Filtered;
}
if let Some(candidate) = self
.pending
.iter_mut()
.find(|candidate| candidate.timeline() == Some(timeline))
{
*candidate = observed;
return LatestIngest::Pending {
timeline,
new_timeline: false,
filtered: 1,
};
}
let filtered = if self.pending.len() == PENDING_TIMELINE_CAPACITY {
self.pending.pop_front();
1
} else {
0
};
self.pending.push_back(observed);
LatestIngest::Pending {
timeline,
new_timeline: true,
filtered,
}
}
fn retain_timeline(&mut self, timeline: TimelineId) -> (u64, bool) {
if self.active_timeline == Some(timeline) {
return (0, self.observed.is_some());
}
if let Some(previous) = self.active_timeline.replace(timeline) {
self.retired_timelines.retire(previous);
}
self.retired_timelines.activate(timeline);
let mut filtered = 0_u64;
let active = self
.observed
.take()
.filter(|observed| {
let keep = observed.timeline().is_none_or(|line| line == timeline);
filtered += u64::from(!keep);
keep
})
.or_else(|| {
let index = self
.pending
.iter()
.position(|observed| observed.timeline() == Some(timeline))?;
self.pending.remove(index)
});
filtered = filtered.saturating_add(u64::try_from(self.pending.len()).unwrap_or(u64::MAX));
self.pending.clear();
self.observed = active;
(filtered, self.observed.is_some())
}
}
impl<B> Clone for Latest<B> {
fn clone(&self) -> Self {
Latest {
state: Arc::clone(&self.state),
metric: self.metric.clone(),
_guard: Arc::clone(&self._guard),
}
}
}
impl<B: ContractBody> Latest<B> {
#[doc(hidden)]
pub async fn new(bus: &Bus, topic: &Topic<Subscribe<B>>) -> Result<Self> {
let state = Arc::new(Mutex::new(LatestState {
active_timeline: None,
observed: None,
pending: VecDeque::with_capacity(PENDING_TIMELINE_CAPACITY),
retired_timelines: RetiredTimelines::default(),
}));
let store = Arc::clone(&state);
let metric = bus.runtime_metrics().register_latest(topic.key());
let observe = metric.clone();
let topic_owned = topic.key().to_string();
let guard = spawn_subscription::<B, _>(
bus,
topic.key(),
move |observed| {
let mut state = store.lock().expect("latest mutex poisoned");
match state.ingest(observed) {
LatestIngest::Active { overwrote } => observe.record_latest(overwrote),
LatestIngest::Pending {
timeline,
new_timeline,
filtered,
} => {
observe.record_pending_latest();
observe.record_timeline_filtered(filtered);
if new_timeline {
tracing::warn!(
target: "phoxal.bus",
topic = %topic_owned,
%timeline,
"quarantining sample from a foreign timeline pending its clock"
);
}
}
LatestIngest::Filtered => observe.record_timeline_filtered(1),
}
},
metric.clone(),
)
.await?;
Ok(Latest {
state,
metric,
_guard: Arc::new(guard),
})
}
pub fn observed(&self) -> Option<Observed<B>> {
let observed = self
.state
.lock()
.expect("latest mutex poisoned")
.observed
.clone();
observed.map(|observed| Observed {
body: observed.body.clone(),
metadata: observed.metadata.clone(),
observed_at: observed.observed_at,
})
}
pub fn latest(&self) -> Option<B> {
self.observed().map(|observed| observed.body)
}
#[doc(hidden)]
pub fn __retain_timeline(&self, timeline: TimelineId) {
let mut state = self.state.lock().expect("latest mutex poisoned");
let (filtered, occupied) = state.retain_timeline(timeline);
self.metric.record_timeline_filtered(filtered);
self.metric.record_latest_depth(occupied);
}
}
pub struct Subscriber<B> {
ring: Arc<Ring<B>>,
_guard: Arc<SubscriptionGuard>,
}
impl<B> Clone for Subscriber<B> {
fn clone(&self) -> Self {
Subscriber {
ring: Arc::clone(&self.ring),
_guard: Arc::clone(&self._guard),
}
}
}
impl<B: ContractBody> Subscriber<B> {
#[doc(hidden)]
pub async fn new(bus: &Bus, topic: &Topic<Subscribe<B>>, depth: usize) -> Result<Self> {
let depth = depth.max(1);
let metric = bus
.runtime_metrics()
.register_subscriber(topic.key(), depth);
let ring = Arc::new(Ring::new(depth, metric.clone()));
let push = Arc::clone(&ring);
let drops = bus.clone();
let topic_owned = topic.key().to_string();
let guard = spawn_subscription::<B, _>(
bus,
topic.key(),
move |observed| {
let outcome = push.push(observed);
if !outcome.accepted {
return;
}
if let Some(timeline) = outcome.new_pending_timeline {
tracing::warn!(
target: "phoxal.bus",
topic = %topic_owned,
%timeline,
"quarantining samples from a foreign timeline pending its clock"
);
}
if outcome.evicted {
drops.health().inbound_drops.fetch_add(1, Ordering::Relaxed);
}
},
metric.clone(),
)
.await?;
Ok(Subscriber {
ring,
_guard: Arc::new(guard),
})
}
pub async fn recv(&self) -> Result<Observed<B>> {
let (observed, _current_depth) = self.ring.recv().await;
Ok(observed)
}
pub fn try_recv(&self) -> Option<Observed<B>> {
self.ring
.try_pop()
.map(|(observed, _current_depth)| observed)
}
pub fn dropped(&self) -> u64 {
self.ring.dropped.load(Ordering::Relaxed)
}
#[doc(hidden)]
pub fn __retain_timeline(&self, timeline: TimelineId) {
self.ring.retain_timeline(timeline);
}
}
struct Ring<B> {
state: Mutex<RingState<B>>,
notify: Notify,
cap: usize,
dropped: AtomicU64,
metric: RuntimeMetricHandle,
}
struct RingState<B> {
active_timeline: Option<TimelineId>,
buf: VecDeque<Observed<B>>,
pending: VecDeque<PendingTimeline<B>>,
retired_timelines: RetiredTimelines,
}
struct PendingTimeline<B> {
timeline: TimelineId,
buf: VecDeque<Observed<B>>,
}
struct RingPush {
accepted: bool,
evicted: bool,
new_pending_timeline: Option<TimelineId>,
}
impl<B> Ring<B> {
fn new(cap: usize, metric: RuntimeMetricHandle) -> Self {
Ring {
state: Mutex::new(RingState {
active_timeline: None,
buf: VecDeque::with_capacity(cap),
pending: VecDeque::with_capacity(PENDING_TIMELINE_CAPACITY),
retired_timelines: RetiredTimelines::default(),
}),
notify: Notify::new(),
cap,
dropped: AtomicU64::new(0),
metric,
}
}
fn push(&self, item: Observed<B>) -> RingPush {
let mut state = self.state.lock().expect("ring mutex poisoned");
let timeline = item.timeline();
if let (Some(timeline), Some(active_timeline)) = (timeline, state.active_timeline)
&& timeline != active_timeline
{
if state.retired_timelines.contains(timeline) {
self.metric.record_timeline_filtered(1);
return RingPush {
accepted: false,
evicted: false,
new_pending_timeline: None,
};
}
let mut new_pending_timeline = None;
let pending_index = state
.pending
.iter()
.position(|pending| pending.timeline == timeline);
let pending_index = match pending_index {
Some(index) => index,
None => {
if state.pending.len() == PENDING_TIMELINE_CAPACITY
&& let Some(removed) = state.pending.pop_front()
{
self.metric.record_timeline_filtered(
u64::try_from(removed.buf.len()).unwrap_or(u64::MAX),
);
}
state.pending.push_back(PendingTimeline {
timeline,
buf: VecDeque::with_capacity(self.cap),
});
new_pending_timeline = Some(timeline);
state.pending.len() - 1
}
};
let pending = &mut state.pending[pending_index];
if pending.buf.len() == self.cap {
pending.buf.pop_front();
self.metric.record_timeline_filtered(1);
}
pending.buf.push_back(item);
self.metric.record_pending_subscriber();
return RingPush {
accepted: true,
evicted: false,
new_pending_timeline,
};
}
let mut dropped = false;
if state.buf.len() == self.cap {
state.buf.pop_front();
dropped = true;
self.dropped.fetch_add(1, Ordering::Relaxed);
}
state.buf.push_back(item);
let depth = state.buf.len();
self.metric.record_subscriber(dropped, depth);
drop(state);
self.notify.notify_one();
RingPush {
accepted: true,
evicted: dropped,
new_pending_timeline: None,
}
}
fn try_pop(&self) -> Option<(Observed<B>, usize)> {
let mut state = self.state.lock().expect("ring mutex poisoned");
let item = state.buf.pop_front()?;
let depth = state.buf.len();
self.metric.record_subscriber_pop(depth);
Some((item, depth))
}
fn retain_timeline(&self, timeline: TimelineId) {
let mut state = self.state.lock().expect("ring mutex poisoned");
if state.active_timeline == Some(timeline) {
return;
}
if let Some(previous) = state.active_timeline.replace(timeline) {
state.retired_timelines.retire(previous);
}
state.retired_timelines.activate(timeline);
let mut filtered = 0_u64;
state.buf.retain(|observed| {
let keep = observed.timeline().is_none_or(|line| line == timeline);
filtered += u64::from(!keep);
keep
});
if let Some(index) = state
.pending
.iter()
.position(|pending| pending.timeline == timeline)
{
let mut promoted = state
.pending
.remove(index)
.expect("pending timeline index must remain valid")
.buf;
if state.buf.is_empty() {
state.buf = promoted;
} else {
while let Some(item) = promoted.pop_front() {
if state.buf.len() == self.cap {
state.buf.pop_front();
filtered = filtered.saturating_add(1);
}
state.buf.push_back(item);
}
}
}
filtered = filtered.saturating_add(state.pending.iter().fold(0_u64, |total, pending| {
total.saturating_add(u64::try_from(pending.buf.len()).unwrap_or(u64::MAX))
}));
state.pending.clear();
self.metric.record_timeline_filtered(filtered);
self.metric.record_subscriber_pop(state.buf.len());
let notify = !state.buf.is_empty();
drop(state);
if notify {
self.notify.notify_waiters();
}
}
async fn recv(&self) -> (Observed<B>, usize) {
loop {
let notified = self.notify.notified();
if let Some(item) = self.try_pop() {
return item;
}
notified.await;
}
}
}
struct SubscriptionGuard {
task: JoinHandle<()>,
}
impl Drop for SubscriptionGuard {
fn drop(&mut self) {
self.task.abort();
}
}
async fn spawn_subscription<B, F>(
bus: &Bus,
topic_key: &str,
mut on_sample: F,
metric: RuntimeMetricHandle,
) -> Result<SubscriptionGuard>
where
B: ContractBody,
F: FnMut(Observed<B>) + Send + 'static,
{
let full_key = bus.full_key(topic_key);
let key_expr = OwnedKeyExpr::new(full_key.clone())
.map_err(|e| BusError::Namespace(format!("invalid subscribe key '{full_key}': {e}")))?;
let subscriber = bus
.session()
.declare_subscriber(key_expr)
.await
.map_err(|e| BusError::Transport(e.to_string()))?;
let topic_owned = topic_key.to_string();
let health_bus = bus.clone();
let task = tokio::spawn(async move {
while let Ok(sample) = subscriber.recv_async().await {
let Some(observed_at) = LocalInstant::try_now() else {
tracing::error!(
target: "phoxal.bus",
topic = %topic_owned,
"dropped inbound sample: the host boot clock could not be read"
);
continue;
};
match decode_sample::<B>(&sample, &topic_owned) {
Ok((body, metadata)) => on_sample(Observed {
body,
metadata,
observed_at,
}),
Err(err) => {
metric.record_decode_error();
health_bus
.health()
.decode_errors
.fetch_add(1, Ordering::Relaxed);
tracing::warn!(target: "phoxal.bus", topic = %topic_owned, error = %err, "dropped inbound sample");
}
}
}
});
Ok(SubscriptionGuard { task })
}
pub(crate) fn decode_sample<B: ContractBody>(
sample: &Sample,
topic: &str,
) -> Result<(B, BusMetadata)> {
let encoding =
parse_encoding_string(&sample.encoding().to_string()).map_err(|e| BusError::Metadata {
topic: topic.to_string(),
detail: format!("malformed encoding string: {e}"),
})?;
match encoding.codec_id() {
Some(CodecId::MessagePack) => {}
None => {
return Err(BusError::UnsupportedCodec(
encoding.codec,
topic.to_string(),
));
}
}
let attachment = sample.attachment().ok_or_else(|| BusError::Metadata {
topic: topic.to_string(),
detail: "missing BusMetadata attachment".to_string(),
})?;
let metadata =
BusMetadata::decode(attachment.to_bytes().as_ref()).map_err(|e| BusError::Metadata {
topic: topic.to_string(),
detail: format!("malformed BusMetadata: {e}"),
})?;
if metadata.codec != encoding.codec {
return Err(BusError::Metadata {
topic: topic.to_string(),
detail: format!(
"encoding/BusMetadata codec mismatch: encoding codec={}, metadata codec={}",
encoding.codec, metadata.codec
),
});
}
match metadata.codec_id() {
Some(CodecId::MessagePack) => {}
None => {
return Err(BusError::UnsupportedCodec(
metadata.codec,
topic.to_string(),
));
}
}
let body = MessagePack::decode::<B>(sample.payload().to_bytes().as_ref())?;
Ok((body, metadata))
}
#[cfg(test)]
mod subscriber_ring_tests {
use super::*;
use crate::identity::ProducerId;
use std::sync::Barrier;
fn timeline(value: u64) -> TimelineId {
TimelineId::from_raw(value).expect("test timeline must be nonzero")
}
fn observed(body: u8, line: Option<u64>) -> Observed<u8> {
Observed {
body,
metadata: BusMetadata {
codec: CodecId::MessagePack.as_u8(),
producer: ProducerId::mint(),
sequence: u64::from(body),
produced_at: line
.map(|line| TimeWindow::exact(RobotInstant::new(timeline(line), 0))),
participant: "test".to_string(),
},
observed_at: LocalInstant::try_now().expect("test host clock"),
}
}
#[test]
fn ring_counts_each_drop_oldest_eviction_cumulatively() {
let metrics = crate::runtime_metrics::RuntimeMetrics::default();
let metric = metrics.register_subscriber("v0.1/test/state", 1);
let ring = Ring::new(1, metric);
let first = ring.push(observed(1, None));
assert!(first.accepted);
assert!(!first.evicted);
let second = ring.push(observed(2, None));
assert!(second.accepted);
assert!(second.evicted);
let third = ring.push(observed(3, None));
assert!(third.accepted);
assert!(third.evicted);
assert_eq!(ring.dropped.load(Ordering::Relaxed), 2);
let (observed, depth) = ring.try_pop().unwrap();
assert_eq!(observed.body, 3);
assert_eq!(depth, 0);
let row = metrics.take().pop().unwrap();
assert_eq!(row.count, 3);
assert_eq!(row.drops, 2);
assert_eq!(row.bounded_evictions, 2);
assert_eq!(row.current_depth, 0);
assert_eq!(row.high_water_depth, 1);
}
#[test]
fn a_sample_expressing_no_robot_time_survives_every_timeline_barrier() {
let metrics = crate::runtime_metrics::RuntimeMetrics::default();
let metric = metrics.register_subscriber("v0.1/test/command", 4);
let ring = Ring::new(4, metric);
ring.retain_timeline(timeline(1));
assert!(ring.push(observed(1, None)).accepted);
ring.retain_timeline(timeline(2));
assert_eq!(ring.try_pop().map(|(sample, _)| sample.body), Some(1));
}
#[test]
fn latest_quarantines_a_replacement_timeline_until_atomic_activation() {
let mut state = LatestState {
active_timeline: None,
observed: None,
pending: VecDeque::with_capacity(PENDING_TIMELINE_CAPACITY),
retired_timelines: RetiredTimelines::default(),
};
assert!(matches!(
state.ingest(observed(1, Some(1))),
LatestIngest::Active { overwrote: false }
));
assert_eq!(state.retain_timeline(timeline(1)), (0, true));
assert!(matches!(
state.ingest(observed(2, Some(2))),
LatestIngest::Pending {
new_timeline: true,
filtered: 0,
..
}
));
assert_eq!(state.observed.as_ref().map(|sample| sample.body), Some(1));
assert_eq!(state.retain_timeline(timeline(2)), (1, true));
assert_eq!(state.observed.as_ref().map(|sample| sample.body), Some(2));
assert!(matches!(
state.ingest(observed(3, Some(1))),
LatestIngest::Filtered
));
assert_eq!(state.observed.as_ref().map(|sample| sample.body), Some(2));
}
#[test]
fn latest_activation_is_safe_when_replacement_ingress_races_the_clock() {
let state = Arc::new(Mutex::new(LatestState {
active_timeline: Some(timeline(1)),
observed: Some(Arc::new(observed(1, Some(1)))),
pending: VecDeque::with_capacity(PENDING_TIMELINE_CAPACITY),
retired_timelines: RetiredTimelines::default(),
}));
let barrier = Arc::new(Barrier::new(3));
let ingress_state = Arc::clone(&state);
let ingress_barrier = Arc::clone(&barrier);
let ingress = std::thread::spawn(move || {
ingress_barrier.wait();
ingress_state
.lock()
.expect("latest mutex poisoned")
.ingest(observed(2, Some(2)));
});
let clock_state = Arc::clone(&state);
let clock_barrier = Arc::clone(&barrier);
let clock = std::thread::spawn(move || {
clock_barrier.wait();
clock_state
.lock()
.expect("latest mutex poisoned")
.retain_timeline(timeline(2));
});
barrier.wait();
ingress.join().expect("ingress thread should join");
clock.join().expect("clock thread should join");
let mut state = state.lock().expect("latest mutex poisoned");
assert_eq!(state.active_timeline, Some(timeline(2)));
assert_eq!(state.observed.as_ref().map(|sample| sample.body), Some(2));
assert!(matches!(
state.ingest(observed(3, Some(1))),
LatestIngest::Filtered
));
assert_eq!(state.observed.as_ref().map(|sample| sample.body), Some(2));
}
#[test]
fn subscriber_activation_is_safe_when_replacement_ingress_races_the_clock() {
let metrics = crate::runtime_metrics::RuntimeMetrics::default();
let metric = metrics.register_subscriber("v0.1/test/state", 4);
let ring = Arc::new(Ring::new(4, metric));
assert!(ring.push(observed(1, Some(1))).accepted);
ring.retain_timeline(timeline(1));
assert_eq!(ring.try_pop().map(|(sample, _)| sample.body), Some(1));
let barrier = Arc::new(Barrier::new(3));
let ingress_ring = Arc::clone(&ring);
let ingress_barrier = Arc::clone(&barrier);
let ingress = std::thread::spawn(move || {
ingress_barrier.wait();
assert!(ingress_ring.push(observed(2, Some(2))).accepted);
});
let clock_ring = Arc::clone(&ring);
let clock_barrier = Arc::clone(&barrier);
let clock = std::thread::spawn(move || {
clock_barrier.wait();
clock_ring.retain_timeline(timeline(2));
});
barrier.wait();
ingress.join().expect("ingress thread should join");
clock.join().expect("clock thread should join");
assert_eq!(ring.try_pop().map(|(sample, _)| sample.body), Some(2));
assert!(!ring.push(observed(3, Some(1))).accepted);
assert!(ring.try_pop().is_none());
assert_eq!(metrics.take().pop().unwrap().timeline_filtered, 1);
}
#[test]
fn only_one_timeline_authority_exists_at_a_time() {
let first = TimelineAuthority::__mint(timeline(1)).expect("first authority should mint");
assert!(
TimelineAuthority::__mint(timeline(2)).is_err(),
"a second authority must be rejected at startup"
);
assert_eq!(first.completed_step(50).instant().ticks(), 50);
drop(first);
TimelineAuthority::__mint(timeline(3)).expect("the slot is released on drop");
}
}