liminal/channel/
record.rs

1//! What a recording publish saw (SRV-012, the tap's stations and hops): whom
2//! its fan-out queued it for and when, and when a durable channel's append
3//! completed. Filled by [`ChannelHandle::publish_recording`] into a record
4//! its caller keeps and reuses.
5//!
6//! [`ChannelHandle::publish_recording`]: crate::channel::ChannelHandle::publish_recording
7
8use std::time::Instant;
9
10/// One recipient: the subscriber pid, and the instant its inbox queued the
11/// envelope.
12pub type Recipient = (u64, Instant);
13
14/// What a recording publish saw. Cleared at the start of each recording
15/// publish and its allocation reused, so a caller that keeps one record
16/// allocates only while it grows to the widest fan-out.
17#[derive(Debug, Default)]
18pub struct PublishRecord {
19    pub(crate) recipients: Vec<Recipient>,
20    pub(crate) appended_at: Option<Instant>,
21}
22
23impl PublishRecord {
24    /// An empty record with room for `recipients` before it grows.
25    #[must_use]
26    pub fn with_capacity(recipients: usize) -> Self {
27        Self {
28            recipients: Vec::with_capacity(recipients),
29            appended_at: None,
30        }
31    }
32
33    /// How many recipients it holds before it grows.
34    #[must_use]
35    pub fn capacity(&self) -> usize {
36        self.recipients.capacity()
37    }
38
39    /// Each subscriber whose inbox queued the envelope, with the instant it
40    /// did, in fan-out order: exactly
41    /// [`ChannelDelivery::delivered_count`](crate::channel::ChannelDelivery::delivered_count)
42    /// of them, never one that matched and refused.
43    #[must_use]
44    pub fn recipients(&self) -> &[Recipient] {
45        &self.recipients
46    }
47
48    /// When a durable channel's append of the envelope completed, before its
49    /// fan-out; `None` for a channel that is not durable.
50    #[must_use]
51    pub const fn appended_at(&self) -> Option<Instant> {
52        self.appended_at
53    }
54
55    /// Empties the record, keeping its allocation.
56    pub fn clear(&mut self) {
57        self.recipients.clear();
58        self.appended_at = None;
59    }
60}
61
62/// The instant a recording publish reads: only ever called while a record
63/// is lent, so a publish with none reads no clock. A test build counts the
64/// reads, so a pin can show it.
65pub fn recorded_now() -> Instant {
66    #[cfg(test)]
67    CLOCK_READS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
68    Instant::now()
69}
70
71/// Clock reads [`recorded_now`] has made in this process. Test-only.
72#[cfg(test)]
73pub static CLOCK_READS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
liminal/channel/record.rs

liminal/channel/
record.rs
