refactor(datastream): cleanup
Trim the datastream crate to its dumb-pipe core: drop the frame-timing sidecar, make mux positions gap-free for accepted frames, and simplify the endpoint fanout. - mux: defer position assignment from submit to drain so an overflowed submission no longer consumes a position (no synthetic gaps); submit now returns bool and the Mutex<Receiver> is removed since positions are assigned only to accepted frames - endpoint/mux: switch from std::sync::mpsc to crossbeam-channel and drop per-event event_matches_request filtering — request filters now apply only to the initial catalog snapshot, and future events broadcast to all subscribers - endpoint (DeliveryFanout): snapshot sender handles under the lock and deliver outside it via FanoutTarget/FanoutReport, so large batches or slow subscribers no longer block subscribe/snapshot control-plane ops - emit/endpoint/producer: drop set_frame_timing_enabled/frame_timing_enabled and the Position return from submit_record/submit_text/submit_bytes, and add submit_text_owned taking owned String - timing/lib/spec: delete the timing module and FRAME_TIME_CHANNEL/FRAME_TIME_CHANNEL_ID/FrameTimeSample re-exports (including the auto-registered timing channel in ChannelCatalogState) and renumber the DATASTREAM_SPEC.md section references across frame/ingest/store/mux - tests: remove the 567-line shared datastream_support/mod.rs harness Signed-off-by: Zachery Aaron Shores-Chmielewski <zacheryasc@gmail.com>
This commit is contained in:
parent
a185c8068c
commit
b06583f598
21 changed files with 1101 additions and 1385 deletions
3
Cargo.lock
generated
3
Cargo.lock
generated
|
|
@ -937,6 +937,7 @@ dependencies = [
|
|||
name = "datastream"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"crossbeam-channel",
|
||||
"iroh",
|
||||
"libc",
|
||||
"serde",
|
||||
|
|
@ -2104,6 +2105,7 @@ dependencies = [
|
|||
name = "iroh-driver"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"crossbeam-channel",
|
||||
"datastream",
|
||||
"distribution",
|
||||
"iroh",
|
||||
|
|
@ -4342,6 +4344,7 @@ name = "swactor-process"
|
|||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"crossbeam-queue",
|
||||
"datastream",
|
||||
"libc",
|
||||
"proptest",
|
||||
"proptest-state-machine",
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ pub mod view;
|
|||
use std::sync::Arc;
|
||||
|
||||
use datastream::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
|
||||
use parking_lot::Mutex;
|
||||
use serde::Serialize;
|
||||
use tokio::sync::broadcast;
|
||||
|
||||
|
|
@ -86,10 +85,32 @@ pub struct DashboardHandle {
|
|||
store: Arc<DashboardStore>,
|
||||
views: Arc<ViewRegistry>,
|
||||
shutdown_notify: Arc<tokio::sync::Notify>,
|
||||
standalone_rt: Mutex<Option<tokio::runtime::Runtime>>,
|
||||
}
|
||||
|
||||
impl DashboardHandle {
|
||||
/// Create the datastream dashboard state.
|
||||
///
|
||||
/// The HTTP server is not started until `start_http` or `start_http_standalone`
|
||||
/// is called.
|
||||
pub fn new(config: DashboardConfig) -> Self {
|
||||
let views = Arc::new(ViewRegistry::new());
|
||||
views.register(Arc::new(live_explorer::LiveDatastreamExplorer::default()));
|
||||
views.register(Arc::new(hardware_view::HardwareDashboardView::default()));
|
||||
views.register(swactor::worker_view());
|
||||
let store = Arc::new(DashboardStore::new(
|
||||
config.raw_frame_history,
|
||||
Arc::clone(&views),
|
||||
));
|
||||
let (frames, _) = broadcast::channel(config.frame_buffer.max(1));
|
||||
Self {
|
||||
port: config.port,
|
||||
frames,
|
||||
store,
|
||||
views,
|
||||
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Register a read-only view. External crates can keep their interpretation
|
||||
/// code beside their component and plug it into this registry.
|
||||
pub fn register_view(&self, view: Arc<dyn DashboardView>) {
|
||||
|
|
@ -113,8 +134,8 @@ impl DashboardHandle {
|
|||
self.shutdown_notify.notify_waiters();
|
||||
}
|
||||
|
||||
/// Start the HTTP server on an existing Tokio runtime.
|
||||
pub fn start_http(&self, handle: tokio::runtime::Handle) {
|
||||
/// Build the HTTP server future for an embedding runtime to poll directly.
|
||||
pub fn http_server(&self) -> impl Future<Output = ()> + Send + 'static {
|
||||
let state = server::AppState {
|
||||
frames: self.frames.clone(),
|
||||
store: Arc::clone(&self.store),
|
||||
|
|
@ -122,44 +143,13 @@ impl DashboardHandle {
|
|||
shutdown_notify: Arc::clone(&self.shutdown_notify),
|
||||
};
|
||||
let port = self.port;
|
||||
handle.spawn(async move {
|
||||
async move {
|
||||
server::run_server(state, port).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the HTTP server on a standalone Tokio runtime.
|
||||
pub fn start_http_standalone(&self) {
|
||||
let rt = tokio::runtime::Builder::new_multi_thread()
|
||||
.worker_threads(1)
|
||||
.enable_all()
|
||||
.build()
|
||||
.expect("failed to create tokio runtime for dashboard HTTP");
|
||||
let handle = rt.handle().clone();
|
||||
*self.standalone_rt.lock() = Some(rt);
|
||||
self.start_http(handle);
|
||||
}
|
||||
}
|
||||
|
||||
/// Create the datastream dashboard state.
|
||||
///
|
||||
/// The HTTP server is not started until `start_http` or `start_http_standalone`
|
||||
/// is called.
|
||||
pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
|
||||
let views = Arc::new(ViewRegistry::new());
|
||||
views.register(Arc::new(live_explorer::LiveDatastreamExplorer::default()));
|
||||
views.register(Arc::new(hardware_view::HardwareDashboardView::default()));
|
||||
views.register(swactor::worker_view());
|
||||
let store = Arc::new(DashboardStore::new(
|
||||
config.raw_frame_history,
|
||||
Arc::clone(&views),
|
||||
));
|
||||
let (frames, _) = broadcast::channel(config.frame_buffer.max(1));
|
||||
DashboardHandle {
|
||||
port: config.port,
|
||||
frames,
|
||||
store,
|
||||
views,
|
||||
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
|
||||
standalone_rt: Mutex::new(None),
|
||||
/// Spawn the HTTP server on an existing Tokio runtime and return its task handle.
|
||||
pub fn spawn_http(&self, handle: &tokio::runtime::Handle) -> tokio::task::JoinHandle<()> {
|
||||
handle.spawn(self.http_server())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ swactor-transport = { path = "../transport" }
|
|||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
iroh = "0.98"
|
||||
crossbeam-channel = "0.5"
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
libc = "0.2"
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -7,14 +7,14 @@ use swactor::actor::ActorAddress;
|
|||
use swactor::process_observer::ProcessOutputObserver;
|
||||
use swactor::runtime::Runtime;
|
||||
|
||||
use super::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
|
||||
use super::frame::{ChannelId, Frame, Lifetime, NodeId, StreamId};
|
||||
use super::mux::Mux;
|
||||
use super::record::Record;
|
||||
use super::wire::{DatastreamFrame, encode_delivery};
|
||||
|
||||
/// Where assembled frames go once the mux has ordered them.
|
||||
/// Legacy sink for frames after mux drain has assigned positions.
|
||||
pub trait FrameSink: Send {
|
||||
/// Ship one ordered frame for `stream`. Best-effort: a sink may drop.
|
||||
/// Ship one positioned frame for `stream`. Best-effort: a sink may drop.
|
||||
fn ship(&mut self, stream: &StreamId, frame: &Frame);
|
||||
}
|
||||
|
||||
|
|
@ -73,26 +73,22 @@ impl DatastreamEmitter {
|
|||
self.mux.dropped()
|
||||
}
|
||||
|
||||
pub fn set_frame_timing_enabled(&self, enabled: bool) {
|
||||
self.mux.set_frame_timing_enabled(enabled);
|
||||
}
|
||||
|
||||
pub fn frame_timing_enabled(&self) -> bool {
|
||||
self.mux.frame_timing_enabled()
|
||||
}
|
||||
|
||||
pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> Position {
|
||||
pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool {
|
||||
self.mux.submit(channel, record.encode())
|
||||
}
|
||||
|
||||
pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> Position {
|
||||
pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool {
|
||||
self.mux.submit(channel, text.as_ref().to_vec())
|
||||
}
|
||||
|
||||
pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> Position {
|
||||
pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool {
|
||||
self.mux.submit(channel, bytes)
|
||||
}
|
||||
|
||||
pub fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool {
|
||||
self.mux.submit(channel, text.into_bytes())
|
||||
}
|
||||
|
||||
pub fn process_observer_with<F>(&self, channel_for: F) -> Arc<dyn ProcessOutputObserver>
|
||||
where
|
||||
F: Fn(&str, bool) -> ChannelId + Send + Sync + 'static,
|
||||
|
|
@ -123,17 +119,21 @@ pub struct DatastreamEventSink {
|
|||
}
|
||||
|
||||
impl DatastreamEventSink {
|
||||
pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> Position {
|
||||
pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool {
|
||||
self.mux.submit(channel, record.encode())
|
||||
}
|
||||
|
||||
pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> Position {
|
||||
pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool {
|
||||
self.mux.submit(channel, text.as_ref().to_vec())
|
||||
}
|
||||
|
||||
pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> Position {
|
||||
pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool {
|
||||
self.mux.submit(channel, bytes)
|
||||
}
|
||||
|
||||
pub fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool {
|
||||
self.mux.submit(channel, text.into_bytes())
|
||||
}
|
||||
}
|
||||
|
||||
/// A sink that drops everything.
|
||||
|
|
|
|||
|
|
@ -6,7 +6,11 @@
|
|||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex, mpsc};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use crossbeam_channel::{
|
||||
Receiver, RecvError, RecvTimeoutError, Sender, TryRecvError, TrySendError, bounded,
|
||||
};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use swactor::process_observer::ProcessOutputObserver;
|
||||
|
|
@ -14,12 +18,10 @@ use swactor::stats::{ActorSnapshot, StatsHook};
|
|||
|
||||
use crate::frame::{
|
||||
ChannelContent, ChannelDescriptor, ChannelFilter, ChannelId, ChannelRef, DatastreamEvent,
|
||||
FrameDelivery, Position, SourceFilter, StreamDescriptor, StreamId, StreamOrigin,
|
||||
SubscriptionRequest,
|
||||
FrameDelivery, SourceFilter, StreamDescriptor, StreamId, StreamOrigin, SubscriptionRequest,
|
||||
};
|
||||
use crate::mux::Mux;
|
||||
use crate::record::Record;
|
||||
use crate::timing::{FRAME_TIME_CHANNEL, FRAME_TIME_CHANNEL_ID};
|
||||
use crate::transport::Delivery;
|
||||
|
||||
const DEFAULT_MUX_CAPACITY: usize = 4096;
|
||||
|
|
@ -51,7 +53,7 @@ pub struct SubscriberSnapshot {
|
|||
pub dropped: u64,
|
||||
}
|
||||
|
||||
/// Current catalog snapshot delivered at subscription time.
|
||||
/// Current catalog snapshot delivered at subscription time, filtered by request.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct DatastreamSnapshot {
|
||||
pub streams: Vec<StreamDescriptor>,
|
||||
|
|
@ -70,7 +72,7 @@ pub struct DatastreamSubscription {
|
|||
name: String,
|
||||
request: SubscriptionRequest,
|
||||
snapshot: DatastreamSnapshot,
|
||||
rx: mpsc::Receiver<DatastreamEvent>,
|
||||
rx: Receiver<DatastreamEvent>,
|
||||
}
|
||||
|
||||
impl DatastreamSubscription {
|
||||
|
|
@ -90,18 +92,18 @@ impl DatastreamSubscription {
|
|||
&self.snapshot
|
||||
}
|
||||
|
||||
pub fn try_recv(&self) -> Result<DatastreamEvent, mpsc::TryRecvError> {
|
||||
pub fn try_recv(&self) -> Result<DatastreamEvent, TryRecvError> {
|
||||
self.rx.try_recv()
|
||||
}
|
||||
|
||||
pub fn recv(&self) -> Result<DatastreamEvent, mpsc::RecvError> {
|
||||
pub fn recv(&self) -> Result<DatastreamEvent, RecvError> {
|
||||
self.rx.recv()
|
||||
}
|
||||
|
||||
pub fn recv_timeout(
|
||||
&self,
|
||||
timeout: std::time::Duration,
|
||||
) -> Result<DatastreamEvent, mpsc::RecvTimeoutError> {
|
||||
) -> Result<DatastreamEvent, RecvTimeoutError> {
|
||||
self.rx.recv_timeout(timeout)
|
||||
}
|
||||
|
||||
|
|
@ -116,17 +118,27 @@ impl DatastreamSubscription {
|
|||
|
||||
struct SubscriberSlot {
|
||||
name: String,
|
||||
request: SubscriptionRequest,
|
||||
tx: mpsc::SyncSender<DatastreamEvent>,
|
||||
tx: Sender<DatastreamEvent>,
|
||||
dropped: u64,
|
||||
}
|
||||
|
||||
struct FanoutTarget {
|
||||
id: SubscriptionId,
|
||||
tx: Sender<DatastreamEvent>,
|
||||
}
|
||||
|
||||
struct FanoutReport {
|
||||
id: SubscriptionId,
|
||||
dropped: u64,
|
||||
disconnected: bool,
|
||||
}
|
||||
|
||||
struct FanoutState {
|
||||
next_id: u64,
|
||||
subscribers: BTreeMap<SubscriptionId, SubscriberSlot>,
|
||||
}
|
||||
|
||||
/// Local event fanout used by endpoints and collectors.
|
||||
/// Local fanout; future events are broadcast to every subscriber without request filtering.
|
||||
pub struct DeliveryFanout {
|
||||
default_capacity: usize,
|
||||
state: Mutex<FanoutState>,
|
||||
|
|
@ -168,7 +180,7 @@ impl DeliveryFanout {
|
|||
capacity: usize,
|
||||
) -> DatastreamSubscription {
|
||||
let name = name.into();
|
||||
let (tx, rx) = mpsc::sync_channel(capacity.max(1));
|
||||
let (tx, rx) = bounded(capacity.max(1));
|
||||
let mut state = self.state.lock().expect("datastream fanout poisoned");
|
||||
let id = SubscriptionId(state.next_id);
|
||||
state.next_id = state.next_id.wrapping_add(1).max(1);
|
||||
|
|
@ -176,7 +188,6 @@ impl DeliveryFanout {
|
|||
id,
|
||||
SubscriberSlot {
|
||||
name: name.clone(),
|
||||
request: request.clone(),
|
||||
tx,
|
||||
dropped: 0,
|
||||
},
|
||||
|
|
@ -212,23 +223,34 @@ impl DeliveryFanout {
|
|||
.collect()
|
||||
}
|
||||
|
||||
pub fn publish(&self, event: DatastreamEvent, catalog: &CatalogSnapshot) -> EndpointTick {
|
||||
self.publish_batch(std::iter::once(event), catalog)
|
||||
pub fn publish(&self, event: DatastreamEvent) -> EndpointTick {
|
||||
self.publish_batch(std::iter::once(event))
|
||||
}
|
||||
|
||||
pub fn publish_batch(
|
||||
&self,
|
||||
events: impl IntoIterator<Item = DatastreamEvent>,
|
||||
catalog: &CatalogSnapshot,
|
||||
) -> EndpointTick {
|
||||
pub fn publish_batch(&self, events: impl IntoIterator<Item = DatastreamEvent>) -> EndpointTick {
|
||||
let events: Vec<DatastreamEvent> = events.into_iter().collect();
|
||||
if events.is_empty() {
|
||||
return EndpointTick::default();
|
||||
}
|
||||
|
||||
let mut state = self.state.lock().expect("datastream fanout poisoned");
|
||||
let subscribers = state.subscribers.len();
|
||||
if subscribers == 0 {
|
||||
// Snapshot sender handles while holding the subscriber map lock, then
|
||||
// deliver outside the lock so large batches or slow subscribers do not
|
||||
// block subscribe/snapshot control-plane operations.
|
||||
let (targets, subscribers) = {
|
||||
let state = self.state.lock().expect("datastream fanout poisoned");
|
||||
let subscribers = state.subscribers.len();
|
||||
let targets = state
|
||||
.subscribers
|
||||
.iter()
|
||||
.map(|(id, slot)| FanoutTarget {
|
||||
id: *id,
|
||||
tx: slot.tx.clone(),
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
(targets, subscribers)
|
||||
};
|
||||
|
||||
if targets.is_empty() {
|
||||
return EndpointTick {
|
||||
drained: events.len(),
|
||||
subscribers: 0,
|
||||
|
|
@ -237,36 +259,48 @@ impl DeliveryFanout {
|
|||
}
|
||||
|
||||
let mut delivered = 0;
|
||||
let mut dropped = 0;
|
||||
let mut disconnected = Vec::new();
|
||||
for (id, slot) in state.subscribers.iter_mut() {
|
||||
let mut reports = Vec::new();
|
||||
for target in targets {
|
||||
let mut dropped = 0;
|
||||
let mut disconnected = false;
|
||||
for event in &events {
|
||||
if !event_matches_request(event, &slot.request, catalog) {
|
||||
continue;
|
||||
}
|
||||
match slot.tx.try_send(event.clone()) {
|
||||
match target.tx.try_send(event.clone()) {
|
||||
Ok(()) => delivered += 1,
|
||||
Err(mpsc::TrySendError::Full(_)) => {
|
||||
slot.dropped = slot.dropped.saturating_add(1);
|
||||
Err(TrySendError::Full(_)) => dropped += 1,
|
||||
Err(TrySendError::Disconnected(_)) => {
|
||||
dropped += 1;
|
||||
}
|
||||
Err(mpsc::TrySendError::Disconnected(_)) => {
|
||||
slot.dropped = slot.dropped.saturating_add(1);
|
||||
dropped += 1;
|
||||
disconnected.push(*id);
|
||||
break;
|
||||
disconnected = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if dropped > 0 || disconnected {
|
||||
reports.push(FanoutReport {
|
||||
id: target.id,
|
||||
dropped,
|
||||
disconnected,
|
||||
});
|
||||
}
|
||||
}
|
||||
for id in disconnected {
|
||||
state.subscribers.remove(&id);
|
||||
|
||||
let dropped_for_subscribers = reports.iter().map(|report| report.dropped).sum::<u64>();
|
||||
if !reports.is_empty() {
|
||||
let mut state = self.state.lock().expect("datastream fanout poisoned");
|
||||
for report in &reports {
|
||||
if let Some(slot) = state.subscribers.get_mut(&report.id) {
|
||||
slot.dropped = slot.dropped.saturating_add(report.dropped);
|
||||
}
|
||||
}
|
||||
for report in &reports {
|
||||
if report.disconnected {
|
||||
state.subscribers.remove(&report.id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
EndpointTick {
|
||||
drained: events.len(),
|
||||
delivered,
|
||||
dropped_for_subscribers: dropped,
|
||||
dropped_for_subscribers: usize::try_from(dropped_for_subscribers).unwrap_or(usize::MAX),
|
||||
subscribers,
|
||||
}
|
||||
}
|
||||
|
|
@ -279,6 +313,7 @@ pub struct CatalogSnapshot {
|
|||
}
|
||||
|
||||
impl CatalogSnapshot {
|
||||
/// Apply a subscription request to the initial metadata snapshot only.
|
||||
pub fn datastream_snapshot(&self, request: &SubscriptionRequest) -> DatastreamSnapshot {
|
||||
let channels: Vec<ChannelDescriptor> = self
|
||||
.channels
|
||||
|
|
@ -320,24 +355,12 @@ struct ChannelCatalogState {
|
|||
|
||||
impl ChannelCatalogState {
|
||||
fn new(stream: StreamDescriptor) -> Self {
|
||||
let mut state = Self {
|
||||
stream: stream.clone(),
|
||||
Self {
|
||||
stream,
|
||||
by_id: BTreeMap::new(),
|
||||
by_name: BTreeMap::new(),
|
||||
next_channel: 1,
|
||||
};
|
||||
let timing = ChannelDescriptor {
|
||||
stream: stream.stream.clone(),
|
||||
id: FRAME_TIME_CHANNEL_ID,
|
||||
name: FRAME_TIME_CHANNEL.to_owned(),
|
||||
label: Some("frame construction time".to_owned()),
|
||||
content: ChannelContent::JsonRecord {
|
||||
schema: Some("datastream.frame_time.v1".to_owned()),
|
||||
},
|
||||
};
|
||||
state.by_name.insert(timing.name.clone(), timing.id);
|
||||
state.by_id.insert(timing.id, timing);
|
||||
state
|
||||
}
|
||||
}
|
||||
|
||||
fn snapshot(&self) -> CatalogSnapshot {
|
||||
|
|
@ -456,14 +479,6 @@ impl DatastreamEndpoint {
|
|||
.snapshot()
|
||||
}
|
||||
|
||||
pub fn set_frame_timing_enabled(&self, enabled: bool) {
|
||||
self.mux.set_frame_timing_enabled(enabled);
|
||||
}
|
||||
|
||||
pub fn frame_timing_enabled(&self) -> bool {
|
||||
self.mux.frame_timing_enabled()
|
||||
}
|
||||
|
||||
pub fn producer(&self) -> DatastreamProducer {
|
||||
DatastreamProducer {
|
||||
mux: Arc::clone(&self.mux),
|
||||
|
|
@ -530,26 +545,36 @@ impl DatastreamEndpoint {
|
|||
self.fanout.subscriber_snapshots()
|
||||
}
|
||||
|
||||
/// Drain the mux and fan out catalog-aware future events.
|
||||
/// Drain the mux once and broadcast future frame events; with no subscribers, drained frames are bitbucketed.
|
||||
pub fn tick(&self) -> EndpointTick {
|
||||
let frames = self.mux.drain();
|
||||
if frames.is_empty() {
|
||||
let events = self.drain_events();
|
||||
if events.is_empty() {
|
||||
return EndpointTick::default();
|
||||
}
|
||||
let drained = frames.len();
|
||||
self.drained.fetch_add(drained as u64, Ordering::Relaxed);
|
||||
let events = frames.into_iter().map(|frame| {
|
||||
DatastreamEvent::Frame(FrameDelivery {
|
||||
channel: ChannelRef {
|
||||
stream: self.stream.clone(),
|
||||
channel: frame.channel,
|
||||
},
|
||||
position: frame.position,
|
||||
payload: frame.payload,
|
||||
self.publish_events(events)
|
||||
}
|
||||
|
||||
fn drain_events(&self) -> Vec<DatastreamEvent> {
|
||||
self.mux
|
||||
.drain()
|
||||
.into_iter()
|
||||
.map(|frame| {
|
||||
DatastreamEvent::Frame(FrameDelivery {
|
||||
channel: ChannelRef {
|
||||
stream: self.stream.clone(),
|
||||
channel: frame.channel,
|
||||
},
|
||||
position: frame.position,
|
||||
payload: frame.payload,
|
||||
})
|
||||
})
|
||||
});
|
||||
let catalog = self.catalog_snapshot();
|
||||
let tick = self.fanout.publish_batch(events, &catalog);
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn publish_events(&self, events: Vec<DatastreamEvent>) -> EndpointTick {
|
||||
let drained = events.len();
|
||||
self.drained.fetch_add(drained as u64, Ordering::Relaxed);
|
||||
let tick = self.fanout.publish_batch(events);
|
||||
if tick.subscribers == 0 {
|
||||
self.bitbucketed
|
||||
.fetch_add(drained as u64, Ordering::Relaxed);
|
||||
|
|
@ -620,24 +645,20 @@ impl DatastreamProducer {
|
|||
.id_for_name(name)
|
||||
}
|
||||
|
||||
pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> Position {
|
||||
pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool {
|
||||
self.mux.submit(channel, record.encode())
|
||||
}
|
||||
|
||||
pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> Position {
|
||||
pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool {
|
||||
self.mux.submit(channel, text.as_ref().to_vec())
|
||||
}
|
||||
|
||||
pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> Position {
|
||||
pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool {
|
||||
self.mux.submit(channel, bytes)
|
||||
}
|
||||
|
||||
pub fn set_frame_timing_enabled(&self, enabled: bool) {
|
||||
self.mux.set_frame_timing_enabled(enabled);
|
||||
}
|
||||
|
||||
pub fn frame_timing_enabled(&self) -> bool {
|
||||
self.mux.frame_timing_enabled()
|
||||
pub fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool {
|
||||
self.mux.submit(channel, text.into_bytes())
|
||||
}
|
||||
|
||||
pub fn process_observer_with<F>(&self, channel_for: F) -> Arc<dyn ProcessOutputObserver>
|
||||
|
|
@ -674,29 +695,25 @@ fn register_channel(
|
|||
name: String,
|
||||
content: ChannelContent,
|
||||
) -> Result<ChannelId, ChannelRegistrationError> {
|
||||
let (id, event, snapshot) = {
|
||||
// New channel declarations publish a future event immediately; existing-name
|
||||
// reuse only returns the prior id.
|
||||
let (id, event) = {
|
||||
let mut catalog = catalog.lock().expect("datastream catalog poisoned");
|
||||
match catalog.try_register_channel(name.clone(), content)? {
|
||||
Some(descriptor) => {
|
||||
let id = descriptor.id;
|
||||
let snapshot = catalog.snapshot();
|
||||
(
|
||||
id,
|
||||
Some(DatastreamEvent::ChannelDeclared(descriptor)),
|
||||
snapshot,
|
||||
)
|
||||
}
|
||||
Some(descriptor) => (
|
||||
descriptor.id,
|
||||
Some(DatastreamEvent::ChannelDeclared(descriptor)),
|
||||
),
|
||||
None => {
|
||||
let id = catalog
|
||||
.id_for_name(&name)
|
||||
.expect("duplicate channel name remains registered");
|
||||
let snapshot = catalog.snapshot();
|
||||
(id, None, snapshot)
|
||||
(id, None)
|
||||
}
|
||||
}
|
||||
};
|
||||
if let Some(event) = event {
|
||||
let _ = fanout.publish(event, &snapshot);
|
||||
let _ = fanout.publish(event);
|
||||
}
|
||||
Ok(id)
|
||||
}
|
||||
|
|
@ -773,33 +790,6 @@ struct RuntimeMessageTypeCount<'a> {
|
|||
count: u64,
|
||||
}
|
||||
|
||||
fn event_matches_request(
|
||||
event: &DatastreamEvent,
|
||||
request: &SubscriptionRequest,
|
||||
catalog: &CatalogSnapshot,
|
||||
) -> bool {
|
||||
match event {
|
||||
DatastreamEvent::StreamDeclared(descriptor) => {
|
||||
source_matches(&descriptor.stream, Some(descriptor), &request.sources)
|
||||
&& (matches!(request.channels, ChannelFilter::All)
|
||||
|| catalog.channels.values().any(|channel| {
|
||||
channel.stream == descriptor.stream
|
||||
&& descriptor_matches_request(channel, request, catalog)
|
||||
}))
|
||||
}
|
||||
DatastreamEvent::ChannelDeclared(descriptor) => {
|
||||
descriptor_matches_request(descriptor, request, catalog)
|
||||
}
|
||||
DatastreamEvent::Frame(delivery) => catalog
|
||||
.descriptor_for(&delivery.channel)
|
||||
.map(|descriptor| descriptor_matches_request(descriptor, request, catalog))
|
||||
.unwrap_or_else(|| matches!(request.channels, ChannelFilter::All)),
|
||||
DatastreamEvent::StreamEnded(stream) => {
|
||||
source_matches(stream, catalog.stream_descriptor(stream), &request.sources)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn descriptor_matches_request(
|
||||
descriptor: &ChannelDescriptor,
|
||||
request: &SubscriptionRequest,
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
//! Core data model: the framed, channel-multiplexed stream (spec §4).
|
||||
//! Core data model: the framed, channel-multiplexed stream (spec §2).
|
||||
//!
|
||||
//! A stream is identified by the producing node and lifetime. Frames carry a
|
||||
//! stream-local numeric channel id plus the mux-assigned position and opaque
|
||||
|
|
@ -10,12 +10,12 @@ use std::sync::Arc;
|
|||
|
||||
use serde::{Deserialize, Deserializer, Serialize, Serializer};
|
||||
|
||||
/// A position assigned by a node's mux (spec §5.2).
|
||||
/// A position assigned by a node's mux during drain (spec §2.3).
|
||||
///
|
||||
/// Positions are **monotonic** and **gap-free** within a single node's stream:
|
||||
/// the mux never reuses one and never skips one in its numbering. A position
|
||||
/// that is assigned but never delivered surfaces downstream as a missing
|
||||
/// position — a detectable gap (spec §5.3, §7.5).
|
||||
/// Assignment is monotonic and gap-free for accepted frames: the mux never
|
||||
/// reuses one and never skips one while draining. A frame assigned a position
|
||||
/// can still be lost by transport and later surface downstream as a detectable
|
||||
/// gap.
|
||||
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize, Deserialize)]
|
||||
pub struct Position(pub u64);
|
||||
|
||||
|
|
@ -27,9 +27,9 @@ impl fmt::Display for Position {
|
|||
|
||||
/// Stream-local numeric channel id.
|
||||
///
|
||||
/// `ChannelId(0)` is reserved for the datastream frame-timing sidecar. Every
|
||||
/// other id is allocated by the stream owner and is meaningful only with the
|
||||
/// corresponding [`StreamId`]. Consumers resolve frames by `(stream, channel)`.
|
||||
/// A raw `ChannelId` is meaningful only together with its [`StreamId`]. The
|
||||
/// public endpoint allocator currently starts at `ChannelId(1)`, leaving
|
||||
/// `ChannelId(0)` unallocated by normal registration.
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)]
|
||||
pub struct ChannelId(pub u32);
|
||||
|
||||
|
|
@ -73,7 +73,7 @@ pub enum StreamOrigin {
|
|||
RemoteNode,
|
||||
}
|
||||
|
||||
/// The stable identity of a node that produces a stream (spec §4.4, §8.1).
|
||||
/// The stable identity of a node that produces a stream (spec §2.2, §7.1).
|
||||
#[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
|
||||
pub struct NodeId(Arc<str>);
|
||||
|
||||
|
|
@ -131,7 +131,7 @@ impl<'de> Deserialize<'de> for NodeId {
|
|||
}
|
||||
}
|
||||
|
||||
/// A lifetime discriminator distinguishing a node's incarnations (spec §8.4).
|
||||
/// A lifetime discriminator distinguishing a node's incarnations (spec §2.2).
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)]
|
||||
pub struct Lifetime(pub u64);
|
||||
|
||||
|
|
@ -168,7 +168,7 @@ pub struct StreamDescriptor {
|
|||
pub origin: StreamOrigin,
|
||||
}
|
||||
|
||||
/// Channel metadata declared by the stream owner.
|
||||
/// Catalog metadata declared by the stream owner; frames store only `id` and payload.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ChannelDescriptor {
|
||||
pub stream: StreamId,
|
||||
|
|
@ -236,7 +236,7 @@ impl SubscriptionRequest {
|
|||
}
|
||||
}
|
||||
|
||||
/// The unit the mux emits (spec §4.1): bytes tagged with a channel and a position.
|
||||
/// The unit the mux emits (spec §2.1): bytes tagged with a channel and a position.
|
||||
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct Frame {
|
||||
/// The stream-local lane these bytes belong to.
|
||||
|
|
|
|||
|
|
@ -1,24 +1,23 @@
|
|||
//! Consumer ingest: reconstruct each node's stream from deliveries (spec
|
||||
//! §8.1).
|
||||
//! Consumer ingest: reconstruct each stream from deliveries (spec §7.1).
|
||||
//!
|
||||
//! The consumer receives frames from many nodes, in any order, some never
|
||||
//! arriving, and reconstructs each node's stream — keyed by stream id,
|
||||
//! ordered by position — into the [`Store`]. Ingest is deliberately thin:
|
||||
//! it routes a delivery to its stream and records the frame whole. It MUST
|
||||
//! NOT thin, aggregate, decode-and-discard, or truncate (spec §8.2), and it
|
||||
//! never inspects a channel or a payload, so a channel it cannot decode is
|
||||
//! retained exactly like any other (spec §8.3).
|
||||
//! arriving, and reconstructs each stream — keyed by stream id, ordered by
|
||||
//! position — into the [`Store`]. Ingest is deliberately thin: it routes a
|
||||
//! delivery to its stream and records the frame whole. It MUST NOT thin,
|
||||
//! aggregate, decode-and-discard, or truncate (spec §7.1, §7.2), and it never
|
||||
//! inspects a channel or payload, so an undecoded channel is retained exactly
|
||||
//! like any other (spec §7.2, §8.3).
|
||||
//!
|
||||
//! Reconstruction is by position, not arrival: out-of-order deliveries land
|
||||
//! in order in the store, and a position delivered twice collapses to one
|
||||
//! (spec §7.5). Two lives of one node are different stream ids and never
|
||||
//! merge (spec §8.4).
|
||||
//! Reconstruction is by position, not arrival: out-of-order deliveries land in
|
||||
//! order in the store, and a position delivered twice collapses to one (spec
|
||||
//! §7.2). Two lives of one node are different stream ids and never merge (spec
|
||||
//! §2.2).
|
||||
|
||||
use super::store::Store;
|
||||
use super::transport::Delivery;
|
||||
|
||||
/// The single consumer toward which all telemetry flows (spec §2). It owns
|
||||
/// the stored streams and grows them as deliveries arrive.
|
||||
/// A store-owning ingest fold for deliveries. It grows stored streams as
|
||||
/// deliveries arrive.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct Consumer {
|
||||
store: Store,
|
||||
|
|
@ -33,7 +32,7 @@ impl Consumer {
|
|||
/// Accept one delivery: route it to its stream and record the frame.
|
||||
/// Returns `true` if the frame was new (a duplicate position is
|
||||
/// ignored, keeping the first — the carrier cannot fabricate content,
|
||||
/// spec §9).
|
||||
/// spec §9.1).
|
||||
pub fn accept(&mut self, delivery: Delivery) -> bool {
|
||||
let Delivery { stream, frame } = delivery;
|
||||
self.store.stream_mut(&stream).record(frame)
|
||||
|
|
@ -46,7 +45,7 @@ impl Consumer {
|
|||
}
|
||||
}
|
||||
|
||||
/// The stored streams — the source of truth for every view (spec §8.2).
|
||||
/// The stored streams — frame truth for raw storage and view projections (spec §7.3, §8.1).
|
||||
pub fn store(&self) -> &Store {
|
||||
&self.store
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,34 +1,35 @@
|
|||
//! The per-node telemetry **datastream** (see `DATASTREAM_SPEC.md`).
|
||||
//!
|
||||
//! A deliberately dumb pipe: producers dump bytes tagged by channel, a
|
||||
//! single per-node mux interleaves them into one ordered stream, a
|
||||
//! best-effort transport carries that stream to the one consumer, ingest
|
||||
//! reconstructs each node's stream by position, and views are read-time
|
||||
//! projections over the stored stream. Nothing between a producer and a
|
||||
//! A deliberately dumb pipe: producers dump bytes tagged by stream-local
|
||||
//! channel id, a single per-node mux accepts those bytes and assigns canonical
|
||||
//! positions during drain, the endpoint broadcasts catalog-aware events to
|
||||
//! subscribers, ingest reconstructs streams by position, and views are
|
||||
//! read-time projections over stored frames. Nothing between a producer and a
|
||||
//! view interprets the payload.
|
||||
//!
|
||||
//! ```text
|
||||
//! producers (caller-owned records + text)
|
||||
//! │ bytes tagged by channel → [`record::Record`]
|
||||
//! │ bytes tagged by registered ChannelId
|
||||
//! ▼
|
||||
//! per-node MUX → [`mux::Mux`]
|
||||
//! │ one ordered stream of [`Frame`]s
|
||||
//! endpoint / catalog → [`endpoint`]
|
||||
//! │ channel metadata + producer handles
|
||||
//! ▼
|
||||
//! best-effort transport → [`transport`]
|
||||
//! │ delivery: frames, maybe dropped/reordered/delayed
|
||||
//! per-node MUX → [`mux::Mux`]
|
||||
//! │ positioned [`frame::Frame`]s
|
||||
//! ▼
|
||||
//! consumer INGEST → [`ingest::Consumer`]
|
||||
//! │ complete stream, stored whole
|
||||
//! endpoint fanout → [`endpoint::DeliveryFanout`]
|
||||
//! │ catalog-aware events, maybe dropped per subscriber
|
||||
//! ▼
|
||||
//! stored STREAM (truth) → [`store`]
|
||||
//! │ read-time only
|
||||
//! ingest / store → [`ingest`], [`store`]
|
||||
//! │ position-keyed frame truth
|
||||
//! ▼
|
||||
//! VIEWS → [`views`]
|
||||
//! views → [`views`]
|
||||
//! ```
|
||||
//!
|
||||
//! The data model ([`frame`]), extension contract ([`record`]), and wire
|
||||
//! envelope ([`wire`]) are the seams a test observes. Channel meanings live in
|
||||
//! producer/consumer crates, not in a datastream-wide catalog.
|
||||
//! The data model ([`frame`]), extension contract ([`record`]), endpoint/fanout
|
||||
//! seam ([`endpoint`]), and compatibility wire helpers ([`wire`]) are the seams
|
||||
//! tests observe. Channel meanings live in producer/consumer crates, not in a
|
||||
//! datastream-wide global registry.
|
||||
|
||||
pub mod emit;
|
||||
pub mod endpoint;
|
||||
|
|
@ -41,7 +42,6 @@ pub mod publisher_actor;
|
|||
pub mod record;
|
||||
pub mod sink_actor;
|
||||
pub mod store;
|
||||
pub mod timing;
|
||||
pub mod transport;
|
||||
pub mod views;
|
||||
pub mod wire;
|
||||
|
|
@ -65,6 +65,5 @@ pub use publisher_actor::{
|
|||
pub use record::{ChannelKind, ChannelRegistry, Record};
|
||||
pub use sink_actor::{DATASTREAM_SINK_NAME, DatastreamSink};
|
||||
pub use store::{GapSpan, Store, StoredStream};
|
||||
pub use timing::{FRAME_TIME_CHANNEL, FRAME_TIME_CHANNEL_ID, FrameTimeSample};
|
||||
pub use transport::{Delivery, Reorder, ScriptedTransport, StreamScript};
|
||||
pub use views::{Body, LogEntry, MergedFrame};
|
||||
|
|
|
|||
|
|
@ -1,42 +1,39 @@
|
|||
//! The per-node mux: the single ordering authority (spec §5).
|
||||
//! The per-stream mux: the single ordering authority (spec §4.4).
|
||||
//!
|
||||
//! Every producer on a node submits bytes tagged with a stream-local channel id
|
||||
//! to one mux, and the mux assigns a single monotonic position sequence across
|
||||
//! all channels. A drop consumes a position and is therefore visible downstream
|
||||
//! as a gap.
|
||||
//! Producers submit bytes tagged with a stream-local channel id. The mux accepts
|
||||
//! payloads into a bounded queue first, then assigns a single monotonic position
|
||||
//! sequence while draining accepted payloads.
|
||||
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::mpsc::{Receiver, SyncSender, TryRecvError, TrySendError, sync_channel};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use crossbeam_channel::{Receiver, Sender, TryRecvError, TrySendError, bounded};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
|
||||
use super::frame::{ChannelId, Frame, Position, StreamId};
|
||||
use super::record::Record;
|
||||
use super::timing::FRAME_TIME_CHANNEL_ID;
|
||||
use super::timing::FrameTimeSample;
|
||||
|
||||
struct PendingFrame {
|
||||
channel: ChannelId,
|
||||
payload: Vec<u8>,
|
||||
}
|
||||
|
||||
/// A node's single position authority and outgoing telemetry queue.
|
||||
pub struct Mux {
|
||||
stream: StreamId,
|
||||
next: AtomicU64,
|
||||
dropped: AtomicU64,
|
||||
frame_timing_enabled: AtomicBool,
|
||||
tx: SyncSender<Frame>,
|
||||
rx: Mutex<Receiver<Frame>>,
|
||||
tx: Sender<PendingFrame>,
|
||||
rx: Receiver<PendingFrame>,
|
||||
}
|
||||
|
||||
impl Mux {
|
||||
/// Create a mux for `stream` with a bounded outgoing queue.
|
||||
pub fn new(stream: StreamId, capacity: usize) -> Self {
|
||||
let capacity = capacity.max(1).min(1_048_576);
|
||||
let (tx, rx) = sync_channel(capacity);
|
||||
let (tx, rx) = bounded(capacity);
|
||||
Mux {
|
||||
stream,
|
||||
next: AtomicU64::new(0),
|
||||
dropped: AtomicU64::new(0),
|
||||
frame_timing_enabled: AtomicBool::new(true),
|
||||
tx,
|
||||
rx: Mutex::new(rx),
|
||||
rx,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -45,88 +42,50 @@ impl Mux {
|
|||
Mux::new(stream, usize::MAX)
|
||||
}
|
||||
|
||||
/// The stream this mux produces (spec §8.4 ingest key).
|
||||
/// The stream this mux produces (spec §2.2, §7.1 ingest key).
|
||||
pub fn stream_id(&self) -> &StreamId {
|
||||
&self.stream
|
||||
}
|
||||
|
||||
/// Submit opaque bytes on a registered channel id.
|
||||
pub fn submit(&self, channel: ChannelId, payload: Vec<u8>) -> Position {
|
||||
let position = Position(self.next.fetch_add(1, Ordering::Relaxed));
|
||||
let timing_sample = self
|
||||
.frame_timing_enabled
|
||||
.load(Ordering::Relaxed)
|
||||
.then(|| now_unix_ns())
|
||||
.filter(|_| channel != FRAME_TIME_CHANNEL_ID)
|
||||
.map(|created_at_unix_ns| FrameTimeSample::new(position, created_at_unix_ns));
|
||||
let frame = Frame {
|
||||
channel,
|
||||
position,
|
||||
payload,
|
||||
};
|
||||
|
||||
match self.tx.try_send(frame) {
|
||||
Ok(()) => {
|
||||
if let Some(sample) = timing_sample {
|
||||
self.push_timing_sample_if_room(sample);
|
||||
}
|
||||
}
|
||||
pub fn submit(&self, channel: ChannelId, payload: Vec<u8>) -> bool {
|
||||
match self.tx.try_send(PendingFrame { channel, payload }) {
|
||||
Ok(()) => true,
|
||||
Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
|
||||
self.dropped.fetch_add(1, Ordering::Relaxed);
|
||||
false
|
||||
}
|
||||
}
|
||||
position
|
||||
}
|
||||
|
||||
/// Pull all currently queued frames, sorted by mux position.
|
||||
/// Pull all currently queued frames in mux queue order.
|
||||
pub fn drain(&self) -> Vec<Frame> {
|
||||
let rx = self.rx.lock().expect("mux receiver poisoned");
|
||||
let mut frames = Vec::new();
|
||||
loop {
|
||||
match rx.try_recv() {
|
||||
Ok(frame) => frames.push(frame),
|
||||
Err(TryRecvError::Empty) => break,
|
||||
Err(TryRecvError::Disconnected) => break,
|
||||
match self.rx.try_recv() {
|
||||
Ok(pending) => {
|
||||
// Position is consumed only after a pending frame has left
|
||||
// the queue; failed submit never reaches this point.
|
||||
let position = Position(self.next.fetch_add(1, Ordering::Relaxed));
|
||||
frames.push(Frame {
|
||||
channel: pending.channel,
|
||||
position,
|
||||
payload: pending.payload,
|
||||
});
|
||||
}
|
||||
Err(TryRecvError::Empty) | Err(TryRecvError::Disconnected) => break,
|
||||
}
|
||||
}
|
||||
frames.sort_by_key(|frame| frame.position);
|
||||
frames
|
||||
}
|
||||
|
||||
/// Enable or disable optional sidecar timing samples for newly submitted frames.
|
||||
pub fn set_frame_timing_enabled(&self, enabled: bool) {
|
||||
self.frame_timing_enabled.store(enabled, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Whether this mux currently emits sidecar frame timing samples.
|
||||
pub fn frame_timing_enabled(&self) -> bool {
|
||||
self.frame_timing_enabled.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn push_timing_sample_if_room(&self, sample: FrameTimeSample) {
|
||||
let position = Position(self.next.fetch_add(1, Ordering::Relaxed));
|
||||
let frame = Frame {
|
||||
channel: FRAME_TIME_CHANNEL_ID,
|
||||
position,
|
||||
payload: sample.encode(),
|
||||
};
|
||||
let _ = self.tx.try_send(frame);
|
||||
}
|
||||
|
||||
/// How many positions have been assigned — the gap-free high-water mark.
|
||||
/// How many positions have been assigned while draining accepted frames.
|
||||
pub fn assigned(&self) -> u64 {
|
||||
self.next.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// How many data frames have been dropped on overflow.
|
||||
/// How many submissions have been dropped before entering the mux.
|
||||
pub fn dropped(&self) -> u64 {
|
||||
self.dropped.load(Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
fn now_unix_ns() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|duration| u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX))
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,15 +1,15 @@
|
|||
//! `DatastreamSink` — the cluster-side consumer of [`DatastreamFrame`] messages.
|
||||
//! Legacy `DatastreamSink` for [`DatastreamFrame`] actor messages.
|
||||
//!
|
||||
//! This is the counterpart to [`ClusterFrameSink`](super::emit::ClusterFrameSink):
|
||||
//! a node ships its ordered telemetry as `DatastreamFrame` actor messages over the
|
||||
//! regular swactor transport, and this actor — registered under a well-known name
|
||||
//! on the collector (e.g. the orchestrator) — receives them, decodes each back
|
||||
//! into a `(StreamId, Frame)` delivery, and hands it to a caller-supplied fold.
|
||||
//! New live transports should prefer catalog-aware [`crate::DatastreamEvent`]
|
||||
//! streams. This actor remains as the counterpart to
|
||||
//! [`ClusterFrameSink`](super::emit::ClusterFrameSink): a node ships positioned
|
||||
//! telemetry as legacy `DatastreamFrame` actor messages over regular Swactor
|
||||
//! transport, and this actor receives them, decodes each back into a
|
||||
//! `(StreamId, Frame)` delivery, and hands it to a caller-supplied fold.
|
||||
//!
|
||||
//! It deliberately knows nothing about any view layer. The actor owns an opaque
|
||||
//! callback so binaries can wire the decoded deliveries into whichever fold they
|
||||
//! need. Malformed payloads are dropped silently — the same best-effort tolerance
|
||||
//! the UDP ingest had.
|
||||
//! callback so binaries can wire decoded deliveries into whichever fold they
|
||||
//! need. Malformed payloads are dropped silently.
|
||||
|
||||
use swactor::actor::ActorInterface;
|
||||
use swactor::runtime::Ctx;
|
||||
|
|
@ -17,9 +17,9 @@ use swactor::runtime::Ctx;
|
|||
use super::frame::{Frame, StreamId};
|
||||
use super::wire::{DatastreamFrame, decode_delivery};
|
||||
|
||||
/// Receives [`DatastreamFrame`] cluster messages and folds each decoded delivery
|
||||
/// through `on_frame`. Spawn it, then publish its address under
|
||||
/// [`DATASTREAM_SINK_NAME`] so emitters can resolve and ship to it.
|
||||
/// Receives legacy [`DatastreamFrame`] cluster messages and folds each decoded
|
||||
/// delivery through `on_frame`. Spawn it, then publish its address under
|
||||
/// [`DATASTREAM_SINK_NAME`] so legacy emitters can resolve and ship to it.
|
||||
pub struct DatastreamSink {
|
||||
on_frame: Box<dyn FnMut(StreamId, Frame) + Send>,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,15 +1,15 @@
|
|||
//! The stored stream — the consumer's source of truth (spec §8).
|
||||
//! The stored stream — the consumer's frame source of truth (spec §7).
|
||||
//!
|
||||
//! Storage holds each node's **complete** stream, whole and append-only.
|
||||
//! Nothing is thinned, aggregated, decoded-and-discarded, or truncated at
|
||||
//! ingest (spec §8.2); everything a view ever shows is derived from here
|
||||
//! (spec §9.1). Frames on channels the consumer cannot decode are kept as
|
||||
//! opaque bytes, in order, alongside the rest (spec §8.3) — the store never
|
||||
//! looks at a channel or a payload.
|
||||
//! Storage holds each node's complete stream, whole and append-only. Nothing is
|
||||
//! thinned, aggregated, decoded-and-discarded, or truncated at ingest (spec
|
||||
//! §7.1, §7.2); everything a view shows is derived from here (spec §8.1).
|
||||
//! Frames on channels the consumer cannot decode are kept as opaque bytes,
|
||||
//! alongside the rest (spec §7.2, §8.3) — the store never looks at a channel or
|
||||
//! payload.
|
||||
//!
|
||||
//! A [`StoredStream`] is keyed in the [`Store`] by [`StreamId`] — node plus
|
||||
//! lifetime — so a re-incarnated node does not append to its prior life
|
||||
//! (spec §8.4).
|
||||
//! lifetime — so a re-incarnated node does not append to its prior life (spec
|
||||
//! §2.2, §7.1).
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
|
|
@ -19,8 +19,8 @@ use super::frame::{Frame, Position, StreamId};
|
|||
///
|
||||
/// Backed by a position-keyed map so out-of-order arrivals land in order
|
||||
/// and a position seen twice collapses to one (the carrier may not
|
||||
/// fabricate content, spec §9). Gaps are not stored — they are *derived*
|
||||
/// at read time from the positions that are present (spec §9.1).
|
||||
/// fabricate content, spec §9.1). Gaps are not stored — they are *derived*
|
||||
/// at read time from the positions that are present (spec §7.3, §8.1).
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct StoredStream {
|
||||
frames: BTreeMap<u64, Frame>,
|
||||
|
|
@ -34,7 +34,7 @@ impl StoredStream {
|
|||
|
||||
/// Record a delivered frame. Idempotent by position: the first frame
|
||||
/// seen for a position wins and is never mutated (append-only,
|
||||
/// spec §8.2). Returns `true` if this was the first time the position
|
||||
/// spec §7.2). Returns `true` if this was the first time the position
|
||||
/// was seen.
|
||||
pub fn record(&mut self, frame: Frame) -> bool {
|
||||
match self.frames.entry(frame.position.0) {
|
||||
|
|
@ -73,19 +73,19 @@ impl StoredStream {
|
|||
}
|
||||
|
||||
/// The **interior** gaps — runs of positions assigned between the first
|
||||
/// and last delivered frame but never delivered (spec §7.5, §8), each as
|
||||
/// one [`GapSpan`].
|
||||
/// and last delivered frame but never delivered (spec §7.3), each as one
|
||||
/// [`GapSpan`].
|
||||
///
|
||||
/// Cost is O(stored frames), never O(gap size): it walks adjacent stored
|
||||
/// positions and reads each span's endpoints from them, rather than
|
||||
/// enumerating the (possibly enormous) range in between. A stream that
|
||||
/// brackets a huge interior gap — what a long consumer outage produces
|
||||
/// (spec §7.4), or a single wild position from a corrupt datagram — still
|
||||
/// surfaces in work proportional to the frames held, not to `u64::MAX`.
|
||||
/// (spec §6.3, §7.3), or a single wild position from a corrupt datagram —
|
||||
/// still surfaces in work proportional to the frames held, not to `u64::MAX`.
|
||||
///
|
||||
/// Only interior gaps are knowable: a position lost *after* the last
|
||||
/// delivered frame leaves no bracketing frame to reveal it, so it shows
|
||||
/// up as the stream simply ending (spec §7.4 node death), not a gap.
|
||||
/// Leading and trailing losses are not derivable from stored frames because
|
||||
/// no bracketing position exists. A position lost after the last delivered
|
||||
/// frame shows up as the stream simply ending (spec §5.7, §7.3), not a gap.
|
||||
pub fn gap_spans(&self) -> Vec<GapSpan> {
|
||||
let mut spans = Vec::new();
|
||||
let mut prev: Option<u64> = None;
|
||||
|
|
@ -105,7 +105,7 @@ impl StoredStream {
|
|||
}
|
||||
|
||||
/// A contiguous run of missing positions surfaced in a stored stream
|
||||
/// (spec §7.5). Inclusive on both ends.
|
||||
/// (spec §7.3). Inclusive on both ends.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct GapSpan {
|
||||
/// First missing position.
|
||||
|
|
@ -121,7 +121,7 @@ impl GapSpan {
|
|||
}
|
||||
}
|
||||
|
||||
/// All stored streams at the consumer, keyed by [`StreamId`] (spec §8.4).
|
||||
/// All stored streams at the consumer, keyed by [`StreamId`] (spec §2.2, §7.1).
|
||||
///
|
||||
/// Two streams with the same node but different lifetime are distinct keys
|
||||
/// and never merge.
|
||||
|
|
|
|||
|
|
@ -1,32 +0,0 @@
|
|||
//! Optional frame-construction timing sidecar records.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::frame::ChannelId;
|
||||
|
||||
use crate::frame::Position;
|
||||
use crate::record::Record;
|
||||
|
||||
/// Reserved channel carrying optional timing samples for frames in the same stream.
|
||||
pub const FRAME_TIME_CHANNEL: &str = "datastream.frame_time";
|
||||
pub const FRAME_TIME_CHANNEL_ID: ChannelId = ChannelId(0);
|
||||
|
||||
/// Sidecar timing sample keyed by the target frame's stream-local position.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct FrameTimeSample {
|
||||
pub target_position: u64,
|
||||
pub created_at_unix_ns: u64,
|
||||
}
|
||||
|
||||
impl FrameTimeSample {
|
||||
pub fn new(target_position: Position, created_at_unix_ns: u64) -> Self {
|
||||
Self {
|
||||
target_position: target_position.0,
|
||||
created_at_unix_ns,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Record for FrameTimeSample {
|
||||
const CHANNEL: &'static str = FRAME_TIME_CHANNEL;
|
||||
}
|
||||
|
|
@ -1,28 +1,26 @@
|
|||
//! Transport: best-effort carriage of a node's stream to the one consumer
|
||||
//! (spec §7), and a scripted in-process carrier for offline tests (testing
|
||||
//! spec §2, §9).
|
||||
//! Legacy transport/test seam for carrying positioned frames into ingest.
|
||||
//!
|
||||
//! A [`Delivery`] is the value on the transport→ingest seam: which stream a
|
||||
//! frame belongs to, and the frame. A real carrier rides the connections
|
||||
//! the system already maintains (spec §7.1); a test replaces it with the
|
||||
//! [`ScriptedTransport`] here, whose faults are chosen by the scenario and
|
||||
//! stay inside the **envelope** (testing spec §9): a carrier may *deliver*,
|
||||
//! *drop*, *reorder*, or *delay*, and it MUST NOT corrupt a payload,
|
||||
//! fabricate a frame, or alter a position.
|
||||
//! The live endpoint path now fans out catalog-aware [`DatastreamEvent`] values;
|
||||
//! this module keeps the older [`Delivery`] shape used by ingest, storage tests,
|
||||
//! and scripted conformance checks.
|
||||
//!
|
||||
//! Under position-ordering a *delay* is indistinguishable from a *reorder*
|
||||
//! (a delayed frame simply arrives later), so the envelope's delay is
|
||||
//! covered by [`Reorder`]. Everything the scripted carrier produces is a
|
||||
//! reordered subsequence of what was sent — never a superset, never a
|
||||
//! mutation — which is exactly the property the real-transport conformance
|
||||
//! check pins (testing spec §9).
|
||||
//! A [`Delivery`] pairs the producing stream id with one frame. A real carrier
|
||||
//! rides connections the system already maintains; tests can replace it with
|
||||
//! [`ScriptedTransport`], whose faults stay inside the transport envelope: it
|
||||
//! may *deliver*, *drop*, *reorder*, or *delay*, and it MUST NOT corrupt a
|
||||
//! payload, fabricate a frame, or alter a position.
|
||||
//!
|
||||
//! Under position-ordering a *delay* is indistinguishable from a *reorder*:
|
||||
//! a delayed frame simply arrives later. Everything the scripted carrier
|
||||
//! produces is a reordered subsequence of what was sent — never a superset and
|
||||
//! never a mutation.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
use super::frame::{Frame, StreamId};
|
||||
|
||||
/// A frame as the consumer receives it from the transport (testing spec §2
|
||||
/// seam): tagged with the stream it belongs to.
|
||||
/// A frame as the legacy transport/ingest seam receives it: tagged with the
|
||||
/// stream it belongs to.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Delivery {
|
||||
/// Which node's life produced the frame (spec §8.1 ingest key).
|
||||
|
|
@ -38,9 +36,9 @@ impl Delivery {
|
|||
}
|
||||
}
|
||||
|
||||
/// How the surviving frames of a stream are reordered on arrival. This is
|
||||
/// the envelope's *reorder* (and *delay*) axis (testing spec §9); each
|
||||
/// variant is a permutation of the survivors, never adding or dropping.
|
||||
/// How the surviving frames of a stream are reordered on arrival. This is the
|
||||
/// envelope's *reorder* and *delay* axis (spec §6.3, §9.1); each variant is a
|
||||
/// permutation of the survivors, never adding or dropping.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub enum Reorder {
|
||||
/// Delivered in the order sent.
|
||||
|
|
@ -57,7 +55,7 @@ pub enum Reorder {
|
|||
Permutation(Vec<usize>),
|
||||
}
|
||||
|
||||
/// The faults a scripted carrier applies to one stream (testing spec §9
|
||||
/// The faults a scripted carrier applies to one stream (spec §6.3, §9.1
|
||||
/// envelope). Drops and reorders only — payloads and positions are never
|
||||
/// touched.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
|
|
@ -91,10 +89,9 @@ impl StreamScript {
|
|||
}
|
||||
}
|
||||
|
||||
/// A scripted, in-process transport (testing spec §2). It is a pure,
|
||||
/// deterministic transform from what a node *sent* to what the consumer is
|
||||
/// *delivered* — the entanglement of real wires replaced by a script so a
|
||||
/// run completes in microseconds and returns the same result every time.
|
||||
/// A scripted, in-process transport for conformance scenarios (spec §9.1). It
|
||||
/// is a pure, deterministic transform from what a node sent to what the
|
||||
/// consumer receives, replacing real wire behavior with a fast test script.
|
||||
pub struct ScriptedTransport;
|
||||
|
||||
impl ScriptedTransport {
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
//! Views: read-time projections over a stored stream (spec §9).
|
||||
//! Views: read-time projections over a stored stream (spec §8).
|
||||
|
||||
use std::fmt;
|
||||
|
||||
|
|
@ -70,6 +70,7 @@ where
|
|||
C: ChannelClassifier + ?Sized,
|
||||
R: Fn(ChannelId) -> Option<String>,
|
||||
{
|
||||
// Capacity covers stored frames; surfaced gaps may add extra log entries.
|
||||
let mut out = Vec::with_capacity(stream.len());
|
||||
let mut prev: Option<u64> = None;
|
||||
for frame in stream.frames() {
|
||||
|
|
@ -114,7 +115,8 @@ where
|
|||
timeline_with_resolver(stream, classifier, resolve_name)
|
||||
}
|
||||
|
||||
/// Transitional merged log using numeric channel ids as strings for classifier lookup.
|
||||
/// Transitional helper for callers that still classify by rendered numeric ids;
|
||||
/// named decoding should use [`merged_log_with_names`].
|
||||
pub fn merged_log_with<C: ChannelClassifier + ?Sized>(
|
||||
stream: &StoredStream,
|
||||
classifier: &C,
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ pub fn decode_delivery(buf: &[u8]) -> Result<(StreamId, Frame), WireError> {
|
|||
Ok((stream, frame))
|
||||
}
|
||||
|
||||
/// One datastream event, addressed to a local/cluster datastream actor.
|
||||
/// Legacy actor-message payload wrapper for datastream bytes.
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct DatastreamFrame {
|
||||
pub payload: Vec<u8>,
|
||||
|
|
|
|||
|
|
@ -1,567 +0,0 @@
|
|||
//! Shared support for the datastream tests (testing spec §2, §3).
|
||||
//!
|
||||
//! Two things live here, both deliberately separate from the system under
|
||||
//! test so a test never asserts the code against itself:
|
||||
//!
|
||||
//! * the **payload library** — realistic record shapes and log lines, the
|
||||
//! bytes the pipe will actually carry (testing spec §5: "never
|
||||
//! placeholder text"); and
|
||||
//! * the **reference model** — the spec's rules restated as small, total
|
||||
//! functions over sequences (testing spec §3). It is the trusted oracle:
|
||||
//! every test's `expected` is derived from it, never captured from a run.
|
||||
//! It is written naively on purpose (collect, sort, scan) so a reader can
|
||||
//! confirm it against the spec by eye, while the pipe computes the same
|
||||
//! answers the long way.
|
||||
//!
|
||||
//! This module is `#[path]`-included into more than one test binary, so
|
||||
//! some items are unused in some of them.
|
||||
#![allow(dead_code)]
|
||||
|
||||
use datastream::mux::Mux;
|
||||
use datastream::store::GapSpan;
|
||||
pub use datastream::transport::Delivery;
|
||||
use datastream::{ChannelId, Frame, Position, Record, StreamId};
|
||||
|
||||
pub mod schema {
|
||||
use datastream::{ChannelId, Record};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
pub const IDENTITY: &str = "identity";
|
||||
pub const HOST_RESOURCE: &str = "host.resource";
|
||||
pub const TRANSPORT_INTERNALS: &str = "transport.internals";
|
||||
pub const MEMBERSHIP: &str = "membership";
|
||||
pub const RUNTIME_STATS: &str = "runtime.stats";
|
||||
pub const DIST_STATE: &str = "dist.state";
|
||||
pub const RUNTIME_ACTORS: &str = "runtime.actors";
|
||||
pub const RUNTIME_WORKERS: &str = "runtime.workers";
|
||||
pub const DATASTREAM_HEALTH: &str = "datastream.health";
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum ProcStream {
|
||||
Stdout,
|
||||
Stderr,
|
||||
}
|
||||
|
||||
impl ProcStream {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
ProcStream::Stdout => "stdout",
|
||||
ProcStream::Stderr => "stderr",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn process_output(label: &str, stream: ProcStream) -> ChannelId {
|
||||
ChannelId::new(format!("proc.{label}.{}", stream.as_str()))
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct IdentityRecord {
|
||||
pub node: String,
|
||||
#[serde(default)]
|
||||
pub life: u64,
|
||||
#[serde(default)]
|
||||
pub node_name: String,
|
||||
#[serde(default)]
|
||||
pub listen_addr: String,
|
||||
#[serde(default)]
|
||||
pub relay_url: String,
|
||||
#[serde(default)]
|
||||
pub version: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct ResourceSample {
|
||||
#[serde(default)]
|
||||
pub cpu_pct: f32,
|
||||
#[serde(default)]
|
||||
pub mem_used_mb: u32,
|
||||
#[serde(default)]
|
||||
pub mem_total_mb: u32,
|
||||
#[serde(default)]
|
||||
pub gpu_pct: f32,
|
||||
#[serde(default)]
|
||||
pub disk_used_gb: u32,
|
||||
#[serde(default)]
|
||||
pub net_rx_kbps: u32,
|
||||
#[serde(default)]
|
||||
pub net_tx_kbps: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct TransportInternals {
|
||||
#[serde(default)]
|
||||
pub relay_connected: bool,
|
||||
#[serde(default)]
|
||||
pub direct_peers: u32,
|
||||
#[serde(default)]
|
||||
pub relay_peers: u32,
|
||||
#[serde(default)]
|
||||
pub rtt_ms_p50: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct MembershipTransition {
|
||||
pub peer: String,
|
||||
pub from: String,
|
||||
pub to: String,
|
||||
#[serde(default)]
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RuntimeStats {
|
||||
#[serde(default)]
|
||||
pub actors_live: u32,
|
||||
#[serde(default)]
|
||||
pub mailbox_depth: u32,
|
||||
#[serde(default)]
|
||||
pub scheduled_tasks: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct DistributionState {
|
||||
#[serde(default)]
|
||||
pub cache_size: u32,
|
||||
#[serde(default)]
|
||||
pub cache_entries: Vec<CacheEntryRec>,
|
||||
#[serde(default)]
|
||||
pub directory_route_count: u32,
|
||||
#[serde(default)]
|
||||
pub registry_size: u32,
|
||||
#[serde(default)]
|
||||
pub registry_tombstones: u32,
|
||||
#[serde(default)]
|
||||
pub registry_entries: Vec<RegistryEntryRec>,
|
||||
#[serde(default)]
|
||||
pub recent_probe_targets: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub peer_auth_mode: String,
|
||||
#[serde(default)]
|
||||
pub authorized_peer_count: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct CacheEntryRec {
|
||||
#[serde(default)]
|
||||
pub actor_addr: String,
|
||||
#[serde(default)]
|
||||
pub node_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RegistryEntryRec {
|
||||
#[serde(default)]
|
||||
pub name: String,
|
||||
#[serde(default)]
|
||||
pub actor_addr: String,
|
||||
#[serde(default)]
|
||||
pub node_id: String,
|
||||
#[serde(default)]
|
||||
pub tombstone: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct WorkerCounters {
|
||||
#[serde(default)]
|
||||
pub num_workers: u32,
|
||||
#[serde(default)]
|
||||
pub scheduled_tasks: u32,
|
||||
#[serde(default)]
|
||||
pub local_sends: u64,
|
||||
#[serde(default)]
|
||||
pub cross_sends: u64,
|
||||
#[serde(default)]
|
||||
pub inbox_sends: u64,
|
||||
#[serde(default)]
|
||||
pub type_mismatches: u64,
|
||||
#[serde(default)]
|
||||
pub panics: u64,
|
||||
#[serde(default)]
|
||||
pub messages_dropped: u64,
|
||||
#[serde(default)]
|
||||
pub restarts: u64,
|
||||
#[serde(default)]
|
||||
pub stops: u64,
|
||||
#[serde(default)]
|
||||
pub messages_processed: u64,
|
||||
#[serde(default)]
|
||||
pub tick_p50_us: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct DatastreamHealth {
|
||||
#[serde(default)]
|
||||
pub assigned: u64,
|
||||
#[serde(default)]
|
||||
pub dropped: u64,
|
||||
#[serde(default)]
|
||||
pub loss_rate_ppm: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ActorRuntimeDetail {
|
||||
#[serde(default)]
|
||||
pub actors: Vec<ActorRec>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ActorRec {
|
||||
#[serde(default)]
|
||||
pub address: String,
|
||||
#[serde(default)]
|
||||
pub name: String,
|
||||
#[serde(default)]
|
||||
pub mailbox_depth: u32,
|
||||
#[serde(default)]
|
||||
pub messages_processed: u64,
|
||||
#[serde(default)]
|
||||
pub last_msg_type: String,
|
||||
#[serde(default)]
|
||||
pub poisoned: bool,
|
||||
#[serde(default)]
|
||||
pub message_type_counts: Vec<(String, u64)>,
|
||||
}
|
||||
|
||||
impl Record for IdentityRecord {
|
||||
const CHANNEL: &'static str = IDENTITY;
|
||||
}
|
||||
impl Record for ResourceSample {
|
||||
const CHANNEL: &'static str = HOST_RESOURCE;
|
||||
}
|
||||
impl Record for TransportInternals {
|
||||
const CHANNEL: &'static str = TRANSPORT_INTERNALS;
|
||||
}
|
||||
impl Record for MembershipTransition {
|
||||
const CHANNEL: &'static str = MEMBERSHIP;
|
||||
}
|
||||
impl Record for RuntimeStats {
|
||||
const CHANNEL: &'static str = RUNTIME_STATS;
|
||||
}
|
||||
impl Record for DistributionState {
|
||||
const CHANNEL: &'static str = DIST_STATE;
|
||||
}
|
||||
impl Record for ActorRuntimeDetail {
|
||||
const CHANNEL: &'static str = RUNTIME_ACTORS;
|
||||
}
|
||||
impl Record for WorkerCounters {
|
||||
const CHANNEL: &'static str = RUNTIME_WORKERS;
|
||||
}
|
||||
impl Record for DatastreamHealth {
|
||||
const CHANNEL: &'static str = DATASTREAM_HEALTH;
|
||||
}
|
||||
}
|
||||
|
||||
use schema::*;
|
||||
|
||||
/// An in-process node (testing spec §2 — the faked machine boundary): a
|
||||
/// *real* mux plus the producers that feed it. Producers push realistic
|
||||
/// bytes in at the producer seam; [`Node::sent`] takes the mux's ordered
|
||||
/// output stream — the value on the mux→transport seam. Only the machine
|
||||
/// boundary is faked; the mux is the production code.
|
||||
pub struct Node {
|
||||
stream: StreamId,
|
||||
mux: Mux,
|
||||
}
|
||||
|
||||
impl Node {
|
||||
/// Stand up a node for a given stream (node identity + lifetime).
|
||||
pub fn new(stream: StreamId) -> Self {
|
||||
Node {
|
||||
mux: Mux::unbounded(stream.clone()),
|
||||
stream,
|
||||
}
|
||||
}
|
||||
|
||||
/// The stream this node produces.
|
||||
pub fn stream_id(&self) -> &StreamId {
|
||||
&self.stream
|
||||
}
|
||||
|
||||
/// A producer emits a typed record on its own channel (spec §6.1).
|
||||
pub fn emit<R: Record>(&self, record: &R) -> Position {
|
||||
self.mux.submit(R::channel(), record.encode())
|
||||
}
|
||||
|
||||
/// A producer emits a line of raw process output (spec §6.2).
|
||||
pub fn emit_text(&self, label: &str, stream: ProcStream, line: &str) -> Position {
|
||||
self.mux
|
||||
.submit(process_output(label, stream), line.as_bytes().to_vec())
|
||||
}
|
||||
|
||||
/// A producer emits bytes on a channel the consumer may not know
|
||||
/// (spec §6.3) — opaque to everything until a view learns the channel.
|
||||
pub fn emit_opaque(&self, channel: &str, bytes: &[u8]) -> Position {
|
||||
self.mux.submit(ChannelId::new(channel), bytes.to_vec())
|
||||
}
|
||||
|
||||
/// Take the node's ordered output stream (drains the mux).
|
||||
pub fn sent(&self) -> Vec<Frame> {
|
||||
self.mux.drain()
|
||||
}
|
||||
}
|
||||
|
||||
/// Realistic payloads, drawn on by both the verified vectors (testing spec
|
||||
/// §5) and the deployment scenario (testing spec §8). Nothing here is
|
||||
/// placeholder text.
|
||||
pub mod payloads {
|
||||
use super::*;
|
||||
|
||||
/// A boot/identity record for a node, including the descriptive fields a
|
||||
/// node fills once known (name, listen addr, embedded relay, build version).
|
||||
pub fn identity(node: &str, life: u64) -> IdentityRecord {
|
||||
IdentityRecord {
|
||||
node: node.to_string(),
|
||||
life,
|
||||
node_name: format!("swift-{node}"),
|
||||
listen_addr: format!("{node}.iroh:4242"),
|
||||
relay_url: "https://relay.example:4443/".to_string(),
|
||||
version: "ds-inference @ abc1234".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// A consolidated distribution-subsystem state; `tick` nudges the values so
|
||||
/// a series is not constant.
|
||||
pub fn dist_state(tick: u64) -> DistributionState {
|
||||
DistributionState {
|
||||
cache_size: 3 + (tick % 4) as u32,
|
||||
cache_entries: vec![CacheEntryRec {
|
||||
actor_addr: format!("actor-{}", tick % 5),
|
||||
node_id: format!("node-{}", tick % 3),
|
||||
}],
|
||||
directory_route_count: 5 + (tick % 7) as u32,
|
||||
registry_size: 8 + (tick % 3) as u32,
|
||||
registry_tombstones: (tick % 2) as u32,
|
||||
registry_entries: vec![RegistryEntryRec {
|
||||
name: format!("svc-{}", tick % 4),
|
||||
actor_addr: format!("actor-{}", tick % 5),
|
||||
node_id: format!("node-{}", tick % 3),
|
||||
tombstone: tick.is_multiple_of(2),
|
||||
}],
|
||||
recent_probe_targets: vec![format!("peer-{}", tick % 6)],
|
||||
peer_auth_mode: if tick.is_multiple_of(2) {
|
||||
"open".into()
|
||||
} else {
|
||||
"allow-list".into()
|
||||
},
|
||||
authorized_peer_count: (tick % 5) as u32,
|
||||
}
|
||||
}
|
||||
|
||||
/// A per-actor runtime-detail record (the real actor table).
|
||||
pub fn actor_detail(tick: u64) -> ActorRuntimeDetail {
|
||||
ActorRuntimeDetail {
|
||||
actors: vec![
|
||||
ActorRec {
|
||||
address: format!("{:064x}", tick),
|
||||
name: "SwimActor".to_string(),
|
||||
mailbox_depth: (tick % 5) as u32,
|
||||
messages_processed: 100 + tick,
|
||||
last_msg_type: "swactor_dist::Ping".to_string(),
|
||||
poisoned: false,
|
||||
message_type_counts: vec![("swactor_dist::Ping".to_string(), 40 + tick)],
|
||||
},
|
||||
ActorRec {
|
||||
address: format!("{:064x}", tick + 1),
|
||||
name: "RegistryActor".to_string(),
|
||||
mailbox_depth: 0,
|
||||
messages_processed: 10 + tick % 3,
|
||||
last_msg_type: "Tick".to_string(),
|
||||
poisoned: false,
|
||||
message_type_counts: vec![("Tick".to_string(), 10 + tick % 3)],
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
/// A plausible resource sample; `tick` nudges the values so a series is
|
||||
/// not constant.
|
||||
pub fn resource(tick: u64) -> ResourceSample {
|
||||
ResourceSample {
|
||||
cpu_pct: 12.5 + (tick % 7) as f32 * 3.0,
|
||||
mem_used_mb: 2048 + (tick % 5) as u32 * 128,
|
||||
mem_total_mb: 16384,
|
||||
gpu_pct: (tick % 4) as f32 * 25.0,
|
||||
disk_used_gb: 40 + (tick % 3) as u32,
|
||||
net_rx_kbps: 900 + (tick % 11) as u32 * 30,
|
||||
net_tx_kbps: 300 + (tick % 13) as u32 * 20,
|
||||
}
|
||||
}
|
||||
|
||||
/// A transport-internals snapshot.
|
||||
pub fn transport(tick: u64) -> TransportInternals {
|
||||
TransportInternals {
|
||||
relay_connected: !tick.is_multiple_of(9),
|
||||
direct_peers: 2 + (tick % 3) as u32,
|
||||
relay_peers: 1,
|
||||
rtt_ms_p50: 18 + (tick % 5) as u32 * 4,
|
||||
}
|
||||
}
|
||||
|
||||
/// A membership transition between two peers' states.
|
||||
pub fn membership(peer: &str, from: &str, to: &str) -> MembershipTransition {
|
||||
MembershipTransition {
|
||||
peer: peer.to_string(),
|
||||
from: from.to_string(),
|
||||
to: to.to_string(),
|
||||
reason: "probe timeout".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// A runtime-stats record.
|
||||
pub fn runtime(tick: u64) -> RuntimeStats {
|
||||
RuntimeStats {
|
||||
actors_live: 30 + (tick % 6) as u32,
|
||||
mailbox_depth: (tick % 17) as u32,
|
||||
scheduled_tasks: 4 + (tick % 3) as u32,
|
||||
}
|
||||
}
|
||||
|
||||
/// Aggregated worker-runtime counters (the `runtime.workers` channel).
|
||||
pub fn worker_counters(tick: u64) -> WorkerCounters {
|
||||
WorkerCounters {
|
||||
num_workers: 4,
|
||||
scheduled_tasks: 4 + (tick % 3) as u32,
|
||||
local_sends: 100 + tick,
|
||||
cross_sends: 20 + tick,
|
||||
inbox_sends: tick,
|
||||
messages_processed: 1000 + tick * 7,
|
||||
tick_p50_us: 50 + tick,
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// Datastream self-health (the `datastream.health` channel). `assigned`
|
||||
/// tracks the seed directly so a scenario can pick a frame out by its value.
|
||||
pub fn datastream_health(tick: u64) -> DatastreamHealth {
|
||||
DatastreamHealth {
|
||||
assigned: tick,
|
||||
dropped: tick % 4,
|
||||
loss_rate_ppm: (tick % 4) as u32,
|
||||
}
|
||||
}
|
||||
|
||||
/// A realistic line of process output (without trailing newline).
|
||||
pub fn log_line(label: &str, tick: u64) -> String {
|
||||
format!(
|
||||
"[{label}] step {tick} loss=0.{:03} lr=3e-4",
|
||||
250 - (tick % 200)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// The reference model: the spec's rules as plain total functions over
|
||||
/// sequences. No transport, no storage, no concurrency, no time.
|
||||
pub mod reference {
|
||||
use super::*;
|
||||
|
||||
/// The frames a consumer was delivered for one stream, in arrival
|
||||
/// order — the raw material reconstruction works over.
|
||||
pub fn delivered_frames(stream: &StreamId, deliveries: &[Delivery]) -> Vec<Frame> {
|
||||
deliveries
|
||||
.iter()
|
||||
.filter(|d| &d.stream == stream)
|
||||
.map(|d| d.frame.clone())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Reconstruction (spec §8.1, testing spec §3): "keep the frames that
|
||||
/// were delivered, in position order." Duplicates of a position
|
||||
/// collapse to one (the carrier may not fabricate content, spec §9, so
|
||||
/// a repeat carries identical bytes).
|
||||
pub fn reconstruct(delivered: &[Frame]) -> Vec<Frame> {
|
||||
let mut frames: Vec<Frame> = Vec::new();
|
||||
for f in delivered {
|
||||
if !frames.iter().any(|seen| seen.position == f.position) {
|
||||
frames.push(f.clone());
|
||||
}
|
||||
}
|
||||
frames.sort_by_key(|f| f.position);
|
||||
frames
|
||||
}
|
||||
|
||||
/// The surfaced gaps as spans: the **interior** runs of positions missing
|
||||
/// between the first and last delivered position (spec §7.5, §8). A
|
||||
/// consumer can only detect gaps it has bracketing frames for; positions
|
||||
/// lost after the last delivered frame are invisible and manifest as the
|
||||
/// stream ending (spec §7.4 node death = truncation, not a gap).
|
||||
///
|
||||
/// Walks the sorted delivered positions — O(frames), never the gap size —
|
||||
/// so the oracle agrees with the store on the cheap path even across a
|
||||
/// near-`u64::MAX` gap.
|
||||
pub fn gap_spans(delivered: &[Frame]) -> Vec<GapSpan> {
|
||||
let recon = reconstruct(delivered);
|
||||
let mut spans = Vec::new();
|
||||
let mut prev: Option<u64> = None;
|
||||
for f in &recon {
|
||||
let p = f.position.0;
|
||||
if let Some(q) = prev
|
||||
&& p > q + 1
|
||||
{
|
||||
spans.push(GapSpan {
|
||||
start: q + 1,
|
||||
end: p - 1,
|
||||
});
|
||||
}
|
||||
prev = Some(p);
|
||||
}
|
||||
spans
|
||||
}
|
||||
|
||||
/// One structural item on the merged timeline (testing spec §3: "all
|
||||
/// stored frames in position order, channels interleaved"). This is the
|
||||
/// *structure* of the merged log — order and surfaced gaps — decoupled
|
||||
/// from how each payload is rendered for display, which is a separate
|
||||
/// §9.3 concern the tests assert on its own.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum TimelineItem {
|
||||
Frame { position: u64, channel: String },
|
||||
Gap { start: u64, end: u64 },
|
||||
}
|
||||
|
||||
/// The merged log oracle (spec §9.2): every delivered frame in position
|
||||
/// order, channels interleaved, with an interior gap surfaced wherever a
|
||||
/// position is missing. Written naively so it is obviously the spec.
|
||||
pub fn merged_log(delivered: &[Frame]) -> Vec<TimelineItem> {
|
||||
let frames = reconstruct(delivered);
|
||||
let mut out = Vec::new();
|
||||
let mut prev: Option<u64> = None;
|
||||
for f in &frames {
|
||||
let pos = f.position.0;
|
||||
if let Some(p) = prev
|
||||
&& pos > p + 1
|
||||
{
|
||||
out.push(TimelineItem::Gap {
|
||||
start: p + 1,
|
||||
end: pos - 1,
|
||||
});
|
||||
}
|
||||
out.push(TimelineItem::Frame {
|
||||
position: pos,
|
||||
channel: f.channel.to_string(),
|
||||
});
|
||||
prev = Some(pos);
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/// Convenience: build a frame on a typed channel from a record.
|
||||
pub fn typed_frame<R: Record>(record: &R, position: u64) -> Frame {
|
||||
Frame::new(R::channel(), Position(position), record.encode())
|
||||
}
|
||||
|
||||
/// Convenience: build a frame on a raw-text process-output channel.
|
||||
pub fn text_frame(label: &str, stream: ProcStream, line: &str, position: u64) -> Frame {
|
||||
Frame::new(
|
||||
process_output(label, stream),
|
||||
Position(position),
|
||||
line.as_bytes().to_vec(),
|
||||
)
|
||||
}
|
||||
|
||||
/// Convenience: a frame on a channel id the consumer does not know — an
|
||||
/// opaque channel (spec §6.3).
|
||||
pub fn opaque_frame(id: &str, payload: &[u8], position: u64) -> Frame {
|
||||
Frame::new(ChannelId::new(id), Position(position), payload.to_vec())
|
||||
}
|
||||
|
|
@ -7,8 +7,7 @@ use datastream::transport::{Delivery, Reorder, ScriptedTransport, StreamScript};
|
|||
use datastream::views::{self, Body, LogEntry};
|
||||
use datastream::wire::{decode_delivery, encode_delivery};
|
||||
use datastream::{
|
||||
ChannelId, ChannelKind, ChannelRegistry, FRAME_TIME_CHANNEL_ID, Frame, FrameTimeSample,
|
||||
Lifetime, NodeId, Position, Record, StreamId,
|
||||
ChannelId, ChannelKind, ChannelRegistry, Frame, Lifetime, NodeId, Position, Record, StreamId,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
|
|
@ -58,67 +57,81 @@ fn record_codecs_round_trip_without_global_catalog() {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn mux_numbers_monotonic_and_gap_free() {
|
||||
fn mux_assigns_positions_when_drained() {
|
||||
let mux = Mux::unbounded(stream());
|
||||
mux.set_frame_timing_enabled(false);
|
||||
|
||||
for i in 0..64u64 {
|
||||
let position = mux.submit(RESOURCE_CHANNEL, resource(i as u32).encode());
|
||||
assert_eq!(position, Position(i));
|
||||
assert!(mux.submit(RESOURCE_CHANNEL, resource(i as u32).encode()));
|
||||
}
|
||||
|
||||
assert_eq!(mux.assigned(), 64);
|
||||
assert_eq!(mux.assigned(), 0);
|
||||
assert_eq!(mux.dropped(), 0);
|
||||
let positions: Vec<u64> = mux.drain().iter().map(|frame| frame.position.0).collect();
|
||||
let mut positions: Vec<u64> = mux.drain().iter().map(|frame| frame.position.0).collect();
|
||||
positions.sort_unstable();
|
||||
assert_eq!(positions, (0..64).collect::<Vec<_>>());
|
||||
assert_eq!(mux.assigned(), 64);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mux_queue_preserves_position_order_across_producers() {
|
||||
fn mux_concurrent_producers_assign_unique_positions_on_drain() {
|
||||
let mux = Arc::new(Mux::unbounded(stream()));
|
||||
mux.set_frame_timing_enabled(false);
|
||||
|
||||
let mut threads = Vec::new();
|
||||
for producer in 0..4u8 {
|
||||
let mux = Arc::clone(&mux);
|
||||
threads.push(thread::spawn(move || {
|
||||
let mut accepted = 0;
|
||||
for seq in 0..32u8 {
|
||||
mux.submit(LOG_CHANNEL, vec![producer, seq]);
|
||||
if mux.submit(LOG_CHANNEL, vec![producer, seq]) {
|
||||
accepted += 1;
|
||||
}
|
||||
}
|
||||
accepted
|
||||
}));
|
||||
}
|
||||
for thread in threads {
|
||||
thread.join().expect("producer thread completes");
|
||||
}
|
||||
let accepted: usize = threads
|
||||
.into_iter()
|
||||
.map(|thread| thread.join().expect("producer thread completes"))
|
||||
.sum();
|
||||
|
||||
assert_eq!(accepted, 128);
|
||||
assert_eq!(mux.assigned(), 0);
|
||||
let frames = mux.drain();
|
||||
assert_eq!(frames.len(), 128);
|
||||
for (expected, frame) in frames.iter().enumerate() {
|
||||
assert_eq!(frame.position, Position(expected as u64));
|
||||
}
|
||||
let mut positions: Vec<u64> = frames.iter().map(|frame| frame.position.0).collect();
|
||||
positions.sort_unstable();
|
||||
assert_eq!(positions, (0..128).collect::<Vec<_>>());
|
||||
assert_eq!(mux.assigned(), 128);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mux_timing_sidecar_uses_reserved_numeric_channel_and_does_not_recurse() {
|
||||
let mux = Mux::unbounded(stream());
|
||||
assert!(mux.frame_timing_enabled());
|
||||
fn mux_full_queue_drops_without_consuming_position() {
|
||||
let mux = Mux::new(stream(), 1);
|
||||
|
||||
let data_position = mux.submit(RESOURCE_CHANNEL, resource(0).encode());
|
||||
let timing_position = mux.submit(
|
||||
FRAME_TIME_CHANNEL_ID,
|
||||
FrameTimeSample::new(Position(42), 123).encode(),
|
||||
);
|
||||
assert!(mux.submit(LOG_CHANNEL, b"first".to_vec()));
|
||||
assert!(!mux.submit(LOG_CHANNEL, b"second".to_vec()));
|
||||
assert_eq!(mux.dropped(), 1);
|
||||
assert_eq!(mux.assigned(), 0);
|
||||
|
||||
let frames = mux.drain();
|
||||
assert_eq!(frames.len(), 1);
|
||||
assert_eq!(frames[0].channel, LOG_CHANNEL);
|
||||
assert_eq!(frames[0].position, Position(0));
|
||||
assert_eq!(frames[0].payload, b"first");
|
||||
assert_eq!(mux.assigned(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mux_one_submit_one_drained_frame_without_timing_sidecar() {
|
||||
let mux = Mux::unbounded(stream());
|
||||
|
||||
assert!(mux.submit(RESOURCE_CHANNEL, resource(0).encode()));
|
||||
let frames = mux.drain();
|
||||
|
||||
assert_eq!(data_position, Position(0));
|
||||
assert_eq!(timing_position, Position(2));
|
||||
assert_eq!(mux.assigned(), 3);
|
||||
assert_eq!(frames.len(), 3);
|
||||
assert_eq!(frames.len(), 1);
|
||||
assert_eq!(frames[0].channel, RESOURCE_CHANNEL);
|
||||
assert_eq!(frames[1].channel, FRAME_TIME_CHANNEL_ID);
|
||||
assert_eq!(frames[2].channel, FRAME_TIME_CHANNEL_ID);
|
||||
let sample = FrameTimeSample::decode(&frames[1].payload).expect("timing decodes");
|
||||
assert_eq!(sample.target_position, data_position.0);
|
||||
assert_eq!(frames[0].position, Position(0));
|
||||
assert_eq!(frames[0].payload, resource(0).encode());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -2,8 +2,8 @@ use std::time::Duration;
|
|||
|
||||
use datastream::{
|
||||
ChannelContent, ChannelContentKind, ChannelFilter, ChannelId, DatastreamEndpoint,
|
||||
DatastreamEvent, FRAME_TIME_CHANNEL_ID, FrameDelivery, FrameTimeSample, Lifetime, NodeId,
|
||||
Position, Record, SourceFilter, StreamId, SubscriptionRequest,
|
||||
DatastreamEvent, FrameDelivery, Lifetime, NodeId, Position, Record, SourceFilter, StreamId,
|
||||
SubscriptionRequest,
|
||||
};
|
||||
use serde_json::Value;
|
||||
use swactor::actor::ActorAddress;
|
||||
|
|
@ -37,7 +37,6 @@ fn frame_event(event: &DatastreamEvent) -> &FrameDelivery {
|
|||
fn endpoint_without_subscribers_drains_to_bitbucket() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
producer.set_frame_timing_enabled(false);
|
||||
let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
|
||||
|
||||
producer.submit_text(log, "before");
|
||||
|
|
@ -63,12 +62,12 @@ fn channel_registration_allocates_numeric_ids() {
|
|||
assert_eq!(stderr, ChannelId(2));
|
||||
assert_eq!(runtime, ChannelId(3));
|
||||
let catalog = endpoint.catalog_snapshot();
|
||||
let removed_timing_name = ["datastream", "frame_time"].join(".");
|
||||
assert!(
|
||||
catalog
|
||||
!catalog
|
||||
.channels
|
||||
.values()
|
||||
.any(|descriptor| descriptor.id == FRAME_TIME_CHANNEL_ID
|
||||
&& descriptor.name == "datastream.frame_time")
|
||||
.any(|descriptor| descriptor.name == removed_timing_name)
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -118,7 +117,6 @@ fn subscription_snapshot_contains_stream_and_channel_metadata() {
|
|||
fn subscription_receives_only_future_matching_frames() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
producer.set_frame_timing_enabled(false);
|
||||
let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
|
||||
|
||||
producer.submit_text(log, "pre-subscription");
|
||||
|
|
@ -138,10 +136,9 @@ fn subscription_receives_only_future_matching_frames() {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn subscription_filters_textstream_channels() {
|
||||
fn subscription_snapshot_filters_but_future_fanout_broadcasts() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
producer.set_frame_timing_enabled(false);
|
||||
let stdout = producer.register_channel("stdout", ChannelContent::TextStream);
|
||||
let json = producer.register_record::<RuntimeRecord>();
|
||||
let text_subscription = endpoint.subscribe(
|
||||
|
|
@ -152,22 +149,26 @@ fn subscription_filters_textstream_channels() {
|
|||
},
|
||||
);
|
||||
|
||||
assert_eq!(text_subscription.snapshot().channels.len(), 1);
|
||||
assert_eq!(text_subscription.snapshot().channels[0].id, stdout);
|
||||
|
||||
producer.submit_text(stdout, "line");
|
||||
producer.submit_record(json, &RuntimeRecord { value: 5 });
|
||||
endpoint.tick();
|
||||
|
||||
let events = text_subscription.drain_available();
|
||||
assert_eq!(events.len(), 1);
|
||||
let delivery = frame_event(&events[0]);
|
||||
assert_eq!(delivery.channel.channel, stdout);
|
||||
assert_eq!(delivery.payload, b"line");
|
||||
assert_eq!(events.len(), 2);
|
||||
let channels: Vec<ChannelId> = events
|
||||
.iter()
|
||||
.map(|event| frame_event(event).channel.channel)
|
||||
.collect();
|
||||
assert_eq!(channels, vec![stdout, json]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn endpoint_fans_out_ordered_frames_to_multiple_subscribers() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
producer.set_frame_timing_enabled(false);
|
||||
let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
|
||||
let left = endpoint.subscribe_all("left");
|
||||
let right = endpoint.subscribe_all("right");
|
||||
|
|
@ -188,7 +189,6 @@ fn endpoint_fans_out_ordered_frames_to_multiple_subscribers() {
|
|||
fn slow_subscriber_drops_without_blocking_fast_subscriber() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
producer.set_frame_timing_enabled(false);
|
||||
let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
|
||||
let slow = endpoint.subscribe_all_with_capacity("slow", 1);
|
||||
let fast = endpoint.subscribe_all_with_capacity("fast", 8);
|
||||
|
|
@ -212,35 +212,56 @@ fn slow_subscriber_drops_without_blocking_fast_subscriber() {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn endpoint_producer_timing_sidecars_are_fanned_out_when_enabled() {
|
||||
fn channel_declared_is_broadcast_to_filtered_subscribers() {
|
||||
let endpoint = endpoint();
|
||||
let subscription = endpoint.subscribe(
|
||||
"text-only",
|
||||
SubscriptionRequest {
|
||||
sources: SourceFilter::All,
|
||||
channels: ChannelFilter::Content(ChannelContentKind::TextStream),
|
||||
},
|
||||
);
|
||||
|
||||
let channel = endpoint.register_channel(
|
||||
"runtime.json",
|
||||
ChannelContent::JsonRecord {
|
||||
schema: Some("runtime.json".to_owned()),
|
||||
},
|
||||
);
|
||||
|
||||
let events = subscription.drain_available();
|
||||
assert_eq!(events.len(), 1);
|
||||
match &events[0] {
|
||||
DatastreamEvent::ChannelDeclared(descriptor) => {
|
||||
assert_eq!(descriptor.id, channel);
|
||||
assert_eq!(descriptor.name, "runtime.json");
|
||||
}
|
||||
other => panic!("expected channel declaration, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn submit_text_owned_queues_owned_string() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
|
||||
let sub = endpoint.subscribe_all("test");
|
||||
|
||||
producer.set_frame_timing_enabled(true);
|
||||
let data_position = producer.submit_text(log, "visible");
|
||||
assert!(producer.submit_text_owned(log, String::from("hello")));
|
||||
let tick = endpoint.tick();
|
||||
|
||||
assert_eq!(data_position, Position(0));
|
||||
assert_eq!(tick.drained, 2);
|
||||
let events = sub.drain_available();
|
||||
assert_eq!(events.len(), 2);
|
||||
assert_eq!(frame_event(&events[0]).channel.channel, log);
|
||||
assert_eq!(frame_event(&events[0]).payload, b"visible");
|
||||
assert_eq!(
|
||||
frame_event(&events[1]).channel.channel,
|
||||
FRAME_TIME_CHANNEL_ID
|
||||
);
|
||||
let sample = FrameTimeSample::decode(&frame_event(&events[1]).payload).expect("timing decodes");
|
||||
assert_eq!(sample.target_position, data_position.0);
|
||||
assert_eq!(tick.drained, 1);
|
||||
assert_eq!(tick.delivered, 1);
|
||||
let event = sub.recv_timeout(Duration::from_millis(50)).unwrap();
|
||||
let delivery = frame_event(&event);
|
||||
assert_eq!(delivery.channel.channel, log);
|
||||
assert_eq!(delivery.payload, b"hello");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn process_observer_adapter_submits_configured_channels() {
|
||||
let endpoint = endpoint();
|
||||
let producer = endpoint.producer();
|
||||
producer.set_frame_timing_enabled(false);
|
||||
let stdout = producer.register_channel("proc.trainer.stdout", ChannelContent::TextStream);
|
||||
let stderr = producer.register_channel("proc.trainer.stderr", ChannelContent::TextStream);
|
||||
let observer = producer.process_observer_with(
|
||||
|
|
@ -300,8 +321,10 @@ fn stats_hook_adapter_submits_worker_snapshot_json() {
|
|||
}
|
||||
|
||||
fn positions(events: &[DatastreamEvent]) -> Vec<u64> {
|
||||
events
|
||||
let mut positions: Vec<u64> = events
|
||||
.iter()
|
||||
.map(|event| frame_event(event).position.0)
|
||||
.collect()
|
||||
.collect();
|
||||
positions.sort_unstable();
|
||||
positions
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,7 +28,6 @@ impl Record for ResourceSample {
|
|||
fn build_stream() -> (StreamId, Vec<Frame>) {
|
||||
let id = StreamId::new(NodeId::new("node-real"), Lifetime(1));
|
||||
let mux = Mux::unbounded(id.clone());
|
||||
mux.set_frame_timing_enabled(false);
|
||||
for tick in 0..20 {
|
||||
mux.submit(
|
||||
RESOURCE_CHANNEL,
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ mod datastream_records {
|
|||
//! the mux.
|
||||
|
||||
use datastream::frame::{Lifetime, NodeId, StreamId};
|
||||
use datastream::{Mux, Record};
|
||||
use datastream::{ChannelId, Mux, Position, Record};
|
||||
use distribution::telemetry::{
|
||||
CacheEntryRec, DIST_STATE, DistributionState, MembershipTransition, RegistryEntryRec,
|
||||
};
|
||||
|
|
@ -49,18 +49,18 @@ mod datastream_records {
|
|||
fn distribution_emits_owned_channel_through_datastream_mux() {
|
||||
let stream = StreamId::new(NodeId::new("dist-node"), Lifetime(1));
|
||||
let mux = Mux::unbounded(stream);
|
||||
mux.set_frame_timing_enabled(false);
|
||||
let state = DistributionState {
|
||||
registry_size: 9,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let pos = mux.submit(DistributionState::channel(), state.encode());
|
||||
let channel = ChannelId(1);
|
||||
assert!(mux.submit(channel, state.encode()));
|
||||
let frames = mux.drain();
|
||||
|
||||
assert_eq!(pos.0, 0);
|
||||
assert_eq!(frames.len(), 1);
|
||||
assert_eq!(frames[0].channel.as_str(), DIST_STATE);
|
||||
assert_eq!(frames[0].channel, channel);
|
||||
assert_eq!(frames[0].position, Position(0));
|
||||
assert_eq!(
|
||||
DistributionState::decode(&frames[0].payload)
|
||||
.unwrap()
|
||||
|
|
|
|||
Loading…
Reference in a new issue