zerodds_durability_store/
model.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! Data model at the store boundary: samples, selectors, paginated pages.
5
6use alloc::string::String;
7use alloc::vec::Vec;
8use std::time::SystemTime;
9
10/// A stored durability sample. The `(instance_key, sequence)` pair is the
11/// stable ordering key within a topic.
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct DurabilitySample {
14    /// Topic name.
15    pub topic: String,
16    /// Instance KeyHash (RTPS §9.6.4.8, 16 bytes).
17    pub instance_key: [u8; 16],
18    /// Writer-assigned monotonic sequence number.
19    pub sequence: u64,
20    /// Encoded payload (XCDR2 serialized body).
21    pub payload: Vec<u8>,
22    /// XCDR representation of `payload` from the encapsulation header (RTPS 2.5
23    /// §10.5): `0` = XCDR1 (CDR/PL_CDR), `1` = XCDR2 (CDR2/D_CDR2/PL_CDR2). The
24    /// replay writer re-declares it so a late-joiner reads the body with the
25    /// right alignment.
26    pub representation: u8,
27    /// Wire byte order of `payload`: `false` = little-endian, `true` =
28    /// big-endian. A big-endian peer's stored sample holds big-endian body
29    /// bytes; the replay writer must emit a matching `_BE` encap header
30    /// (otherwise a late-joiner mis-decodes). Defaults to little-endian.
31    pub big_endian: bool,
32    /// Source/creation wall-clock timestamp.
33    pub created_at: SystemTime,
34    /// GUID of the writer that originally emitted this sample (16 bytes). With
35    /// `source_sequence` this is the globally-unique source identity a
36    /// durability service uses to dedup a writer's history against its live
37    /// stream, and across a service restart (O2 P5). All-zero when unknown.
38    pub source_guid: [u8; 16],
39    /// The source writer's RTPS sequence number for this sample (DDSI-RTPS
40    /// §8.3.5.4). `-1` when the ingest path had no source sequence (intra-runtime
41    /// or pre-P5 stored samples), in which case it does not participate in
42    /// source-identity dedup.
43    pub source_sequence: i64,
44}
45
46/// Pagination cursor: the `(instance_key, sequence)` of the last sample of a
47/// [`Page`]. The next page starts strictly after this key in
48/// `(instance_key, sequence)` lexicographic order.
49pub type Cursor = ([u8; 16], u64);
50
51/// A read selector for [`crate::DurabilityStore::query`]. All filters are
52/// optional and combine with AND. `after` + `limit` drive pagination.
53///
54/// An empty selector (`Selector::default()`) plus paging therefore yields the
55/// full contracted history for a topic, ordered by `(instance_key, sequence)`.
56#[derive(Debug, Clone, Default)]
57pub struct Selector {
58    /// Restrict to a single instance.
59    pub instance_key: Option<[u8; 16]>,
60    /// Inclusive lower sequence bound.
61    pub seq_from: Option<u64>,
62    /// Inclusive upper sequence bound.
63    pub seq_to: Option<u64>,
64    /// Inclusive lower `created_at` bound.
65    pub time_from: Option<SystemTime>,
66    /// Inclusive upper `created_at` bound.
67    pub time_to: Option<SystemTime>,
68    /// Return only samples strictly after this `(instance_key, sequence)`.
69    pub after: Option<Cursor>,
70    /// Maximum samples in the returned page. `None` = adapter default.
71    pub limit: Option<usize>,
72}
73
74impl Selector {
75    /// Builds the selector for the next page after a returned [`Cursor`],
76    /// preserving the filter predicate but advancing the pagination anchor.
77    #[must_use]
78    pub fn after_cursor(&self, cursor: Cursor) -> Self {
79        Self {
80            after: Some(cursor),
81            ..self.clone()
82        }
83    }
84
85    /// Returns `true` if `sample` matches the filter predicate (ignores
86    /// `after`/`limit`, which are pagination, not predicate).
87    #[must_use]
88    pub fn matches(&self, sample: &DurabilitySample) -> bool {
89        if let Some(k) = self.instance_key {
90            if sample.instance_key != k {
91                return false;
92            }
93        }
94        if let Some(lo) = self.seq_from {
95            if sample.sequence < lo {
96                return false;
97            }
98        }
99        if let Some(hi) = self.seq_to {
100            if sample.sequence > hi {
101                return false;
102            }
103        }
104        if let Some(t0) = self.time_from {
105            if sample.created_at < t0 {
106                return false;
107            }
108        }
109        if let Some(t1) = self.time_to {
110            if sample.created_at > t1 {
111                return false;
112            }
113        }
114        // Pagination: strictly after `(instance_key, sequence)`.
115        if let Some((ck, cs)) = self.after {
116            if (sample.instance_key, sample.sequence) <= (ck, cs) {
117                return false;
118            }
119        }
120        true
121    }
122}
123
124/// One page of a paginated [`crate::DurabilityStore::query`] result. Samples
125/// are ordered by `(instance_key, sequence)`. `next` is `Some` when more
126/// samples may follow — feed it back via [`Selector::after_cursor`].
127#[derive(Debug, Clone, Default)]
128pub struct Page {
129    /// The samples in this page, ordered by `(instance_key, sequence)`.
130    pub samples: Vec<DurabilitySample>,
131    /// Cursor to continue after, or `None` when the result is exhausted.
132    pub next: Option<Cursor>,
133}
134
135/// Aggregate counters for a topic (diagnostics, limit accounting, tiering).
136#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
137pub struct StoreStats {
138    /// Number of stored samples for the topic.
139    pub samples: usize,
140    /// Number of distinct instances for the topic.
141    pub instances: usize,
142    /// Total payload bytes stored for the topic.
143    pub bytes: u64,
144}
zerodds_durability_store/model.rs

zerodds_durability_store/
model.rs
