salamander/
stream.rs

1//! First-class stream and append-batch contracts.
2
3use std::collections::HashMap;
4
5use crate::format::{
6    BatchId, BranchId, EventId, EventType, Metadata, OwnedStoredRecord, StreamId, StreamRevision,
7};
8use crate::{Result, SalamanderError};
9
10/// The user-facing name of a stream within a branch (UTF-8, non-empty, no
11/// NUL bytes, at most [`StreamName::MAX_BYTES`]). The engine maps it to a
12/// stable compact `StreamId` internally.
13#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
14pub struct StreamName(String);
15
16impl StreamName {
17    /// Maximum length of a stream name, in bytes.
18    pub const MAX_BYTES: usize = 1024;
19
20    /// Validates and constructs a stream name, rejecting empty, oversized,
21    /// or NUL-containing input.
22    pub fn new(value: impl Into<String>) -> Result<Self> {
23        let value = value.into();
24        if value.is_empty() {
25            return Err(SalamanderError::InvalidArgument(
26                "stream name must not be empty".into(),
27            ));
28        }
29        if value.len() > Self::MAX_BYTES {
30            return Err(SalamanderError::ResourceLimit {
31                resource: "stream name bytes",
32                actual: value.len() as u64,
33                maximum: Self::MAX_BYTES as u64,
34            });
35        }
36        if value.as_bytes().contains(&0) {
37            return Err(SalamanderError::InvalidArgument(
38                "stream name must not contain NUL".into(),
39            ));
40        }
41        Ok(Self(value))
42    }
43
44    /// The name as a string slice.
45    pub fn as_str(&self) -> &str {
46        &self.0
47    }
48}
49
50impl TryFrom<&str> for StreamName {
51    type Error = SalamanderError;
52
53    fn try_from(value: &str) -> Result<Self> {
54        Self::new(value)
55    }
56}
57
58/// Optimistic-concurrency expectation checked against a stream's current
59/// revision, in the same critical section that assigns positions. A failed
60/// expectation appends nothing.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub enum ExpectedRevision {
63    /// Append unconditionally.
64    Any,
65    /// Require that the stream does not yet exist.
66    NoStream,
67    /// Require the stream's current revision to equal this value.
68    Exact(StreamRevision),
69}
70
71/// A caller-supplied key identifying an append command, scoped to a
72/// database and branch, so that a retry with identical content is
73/// idempotent (non-empty, at most [`IdempotencyKey::MAX_BYTES`]).
74#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
75pub struct IdempotencyKey(Vec<u8>);
76
77impl IdempotencyKey {
78    /// Maximum length of an idempotency key, in bytes.
79    pub const MAX_BYTES: usize = 1024;
80
81    /// Validates and constructs an idempotency key, rejecting empty or
82    /// oversized input.
83    pub fn new(value: impl Into<Vec<u8>>) -> Result<Self> {
84        let value = value.into();
85        if value.is_empty() {
86            return Err(SalamanderError::InvalidArgument(
87                "idempotency key must not be empty".into(),
88            ));
89        }
90        if value.len() > Self::MAX_BYTES {
91            return Err(SalamanderError::ResourceLimit {
92                resource: "idempotency key bytes",
93                actual: value.len() as u64,
94                maximum: Self::MAX_BYTES as u64,
95            });
96        }
97        Ok(Self(value))
98    }
99
100    /// The key as a byte slice.
101    pub fn as_bytes(&self) -> &[u8] {
102        &self.0
103    }
104}
105
106/// The durability level requested for an append. A stronger level includes
107/// the guarantees of the weaker ones; see the crate's durability contract
108/// for the per-platform survival matrix.
109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110pub enum Durability {
111    /// Return after bytes are accepted into engine-managed memory; no
112    /// promise of surviving process or power loss.
113    Buffered,
114    /// Submit all bytes to the operating system before returning; not
115    /// promised to survive OS or power loss.
116    Flush,
117    /// Invoke the strongest supported file-data synchronization before
118    /// returning; survives abrupt process termination under the documented
119    /// model.
120    Sync,
121}
122
123/// The durability actually achieved for an append, as reported on the
124/// [`AppendReceipt`]. The receipt states this rather than leaving the
125/// caller to infer it from the method used.
126#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
127pub enum ReceiptDurability {
128    /// Accepted into memory only.
129    Buffered,
130    /// Handed to the operating system.
131    Flushed,
132    /// Synchronized to durable storage.
133    Synced,
134}
135
136/// One event to be appended, carrying a typed payload `B`.
137#[derive(Debug, Clone)]
138pub struct NewEvent<B> {
139    /// Caller-supplied event identity; if `None`, the engine generates one.
140    pub event_id: Option<EventId>,
141    /// The event's type tag.
142    pub event_type: EventType,
143    /// Schema version of the payload under `event_type`.
144    pub schema_version: u32,
145    /// Engine metadata attached to the event (reserved keys are prefixed
146    /// `salamander.`).
147    pub metadata: Metadata,
148    /// The application payload.
149    pub body: B,
150}
151
152impl<B> NewEvent<B> {
153    /// Creates an event with a generated ID, schema version 1, and no
154    /// metadata.
155    pub fn new(event_type: EventType, body: B) -> Self {
156        Self {
157            event_id: None,
158            event_type,
159            schema_version: 1,
160            metadata: Metadata::new(),
161            body,
162        }
163    }
164}
165
166/// One append command: a batch of events for a single stream, appended
167/// atomically (all-or-nothing) under an optimistic-concurrency expectation
168/// and a durability level.
169#[derive(Debug, Clone)]
170pub struct AppendRequest<B> {
171    /// Branch to append to.
172    pub branch: BranchId,
173    /// Target stream within the branch.
174    pub stream: StreamName,
175    /// Optimistic-concurrency expectation validated before any write.
176    pub expected: ExpectedRevision,
177    /// Optional key making a retry of this command idempotent.
178    pub idempotency_key: Option<IdempotencyKey>,
179    /// The events, in order; must be non-empty and at most
180    /// [`AppendRequest::MAX_EVENTS`].
181    pub events: Vec<NewEvent<B>>,
182    /// Durability level to satisfy before returning the receipt.
183    pub durability: Durability,
184}
185
186impl<B> AppendRequest<B> {
187    /// Maximum number of events in a single batch.
188    pub const MAX_EVENTS: usize = 4096;
189
190    /// Checks the batch is non-empty and within [`Self::MAX_EVENTS`].
191    pub fn validate(&self) -> Result<()> {
192        if self.events.is_empty() {
193            return Err(SalamanderError::InvalidArgument(
194                "append batch must contain at least one event".into(),
195            ));
196        }
197        if self.events.len() > Self::MAX_EVENTS {
198            return Err(SalamanderError::ResourceLimit {
199                resource: "batch event count",
200                actual: self.events.len() as u64,
201                maximum: Self::MAX_EVENTS as u64,
202            });
203        }
204        Ok(())
205    }
206}
207
208/// The result of a successful append: the assigned positions, resulting
209/// stream revision, and the durability that was achieved.
210#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
211pub struct AppendReceipt {
212    /// Stable identity of the appended batch.
213    pub batch_id: BatchId,
214    /// Global position of the first event in the batch.
215    pub first_position: u64,
216    /// Global position of the last event in the batch.
217    pub last_position: u64,
218    /// Stable engine identity of the target stream.
219    pub stream_id: StreamId,
220    /// The stream's revision before this batch, or `None` if it was new.
221    pub previous_revision: Option<StreamRevision>,
222    /// The stream's revision after this batch.
223    pub current_revision: StreamRevision,
224    /// Durability guaranteed at the time the receipt was returned.
225    pub durability: ReceiptDurability,
226}
227
228#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
229pub(crate) struct StreamCatalog {
230    streams: HashMap<(BranchId, String), StreamEntry>,
231    event_ids: HashMap<EventId, (u32, BatchId)>,
232    idempotency: HashMap<(BranchId, Vec<u8>), IdempotencyEntry>,
233    receipts: HashMap<BatchId, AppendReceipt>,
234}
235
236#[derive(Clone, Copy, serde::Serialize, serde::Deserialize)]
237struct StreamEntry {
238    id: StreamId,
239    last_revision: StreamRevision,
240}
241
242#[derive(Clone, serde::Serialize, serde::Deserialize)]
243struct IdempotencyEntry {
244    digest: u32,
245    receipt: AppendReceipt,
246}
247
248impl StreamCatalog {
249    pub(crate) fn rebuild(
250        records: impl Iterator<Item = Result<OwnedStoredRecord>>,
251    ) -> Result<Self> {
252        let mut catalog = Self::default();
253        let mut batches: HashMap<BatchId, AppendReceipt> = HashMap::new();
254        for item in records {
255            let record = item?;
256            let name = stream_name(&record)?;
257            let key = (record.envelope.branch_id, name.clone());
258            let entry = catalog.streams.entry(key).or_insert(StreamEntry {
259                id: record.envelope.stream_id,
260                last_revision: record.envelope.stream_revision,
261            });
262            if entry.id != record.envelope.stream_id {
263                return Err(SalamanderError::Corrupt {
264                    offset: record.position,
265                    reason: "stream name maps to multiple stream IDs".into(),
266                });
267            }
268            if record.envelope.stream_revision > entry.last_revision {
269                entry.last_revision = record.envelope.stream_revision;
270            }
271            let event_digest = event_fingerprint(&record);
272            if let Some(previous) = catalog.event_ids.insert(
273                record.envelope.event_id,
274                (event_digest, record.envelope.batch_id),
275            ) {
276                if previous.0 != event_digest {
277                    return Err(SalamanderError::EventIdConflict);
278                }
279            }
280
281            let receipt =
282                batches
283                    .entry(record.envelope.batch_id)
284                    .or_insert_with(|| AppendReceipt {
285                        batch_id: record.envelope.batch_id,
286                        first_position: record.position,
287                        last_position: record.position,
288                        stream_id: record.envelope.stream_id,
289                        previous_revision: record
290                            .envelope
291                            .stream_revision
292                            .0
293                            .checked_sub(1)
294                            .map(StreamRevision),
295                        current_revision: record.envelope.stream_revision,
296                        durability: ReceiptDurability::Buffered,
297                    });
298            receipt.last_position = record.position;
299            receipt.current_revision = record.envelope.stream_revision;
300
301            if let (Some(idempotency_key), Some(digest_bytes)) = (
302                record.envelope.metadata.get("salamander.idempotency_key"),
303                record.envelope.metadata.get("salamander.request_digest"),
304            ) {
305                if digest_bytes.len() == 4 {
306                    let digest = u32::from_le_bytes(digest_bytes.as_slice().try_into().unwrap());
307                    catalog.idempotency.insert(
308                        (record.envelope.branch_id, idempotency_key.clone()),
309                        IdempotencyEntry {
310                            digest,
311                            receipt: receipt.clone(),
312                        },
313                    );
314                }
315            }
316        }
317        catalog.receipts = batches;
318        Ok(catalog)
319    }
320
321    pub(crate) fn revision(&self, branch: BranchId, stream: &StreamName) -> Option<StreamRevision> {
322        self.streams
323            .get(&(branch, stream.as_str().to_string()))
324            .map(|entry| entry.last_revision)
325    }
326
327    pub(crate) fn stream_id(&self, branch: BranchId, stream: &StreamName) -> Option<StreamId> {
328        self.streams
329            .get(&(branch, stream.as_str().to_string()))
330            .map(|entry| entry.id)
331    }
332
333    pub(crate) fn event_digest(&self, id: EventId) -> Option<u32> {
334        self.event_ids.get(&id).map(|entry| entry.0)
335    }
336
337    pub(crate) fn event_receipt(&self, id: EventId) -> Option<(u32, AppendReceipt)> {
338        let (digest, batch) = self.event_ids.get(&id)?;
339        Some((*digest, self.receipts.get(batch)?.clone()))
340    }
341
342    pub(crate) fn batch_receipt(&self, id: BatchId) -> Option<AppendReceipt> {
343        self.receipts.get(&id).cloned()
344    }
345
346    pub(crate) fn idempotent(
347        &self,
348        branch: BranchId,
349        key: &IdempotencyKey,
350    ) -> Option<(u32, AppendReceipt)> {
351        self.idempotency
352            .get(&(branch, key.as_bytes().to_vec()))
353            .map(|entry| (entry.digest, entry.receipt.clone()))
354    }
355
356    pub(crate) fn record_batch(
357        &mut self,
358        branch: BranchId,
359        stream: &StreamName,
360        stream_id: StreamId,
361        event_digests: impl IntoIterator<Item = (EventId, u32)>,
362        idempotency: Option<(&IdempotencyKey, u32)>,
363        receipt: AppendReceipt,
364    ) {
365        self.streams.insert(
366            (branch, stream.as_str().to_string()),
367            StreamEntry {
368                id: stream_id,
369                last_revision: receipt.current_revision,
370            },
371        );
372        self.event_ids.extend(
373            event_digests
374                .into_iter()
375                .map(|(id, digest)| (id, (digest, receipt.batch_id))),
376        );
377        self.receipts.insert(receipt.batch_id, receipt.clone());
378        if let Some((key, digest)) = idempotency {
379            self.idempotency.insert(
380                (branch, key.as_bytes().to_vec()),
381                IdempotencyEntry { digest, receipt },
382            );
383        }
384    }
385}
386
387fn stream_name(record: &OwnedStoredRecord) -> Result<String> {
388    let bytes = record
389        .envelope
390        .metadata
391        .get("salamander.stream_name")
392        .ok_or_else(|| SalamanderError::Corrupt {
393            offset: record.position,
394            reason: "event is missing stream name metadata".into(),
395        })?;
396    std::str::from_utf8(bytes)
397        .map(str::to_owned)
398        .map_err(|_| SalamanderError::Corrupt {
399            offset: record.position,
400            reason: "stream name metadata is not UTF-8".into(),
401        })
402}
403
404pub(crate) fn event_fingerprint(record: &OwnedStoredRecord) -> u32 {
405    let mut bytes = Vec::new();
406    bytes.extend_from_slice(record.envelope.event_type.as_str().as_bytes());
407    bytes.extend_from_slice(&record.envelope.schema_version.to_le_bytes());
408    for (key, value) in &record.envelope.metadata {
409        if !key.starts_with("salamander.") {
410            bytes.extend_from_slice(key.as_bytes());
411            bytes.extend_from_slice(value);
412        }
413    }
414    bytes.extend_from_slice(&record.payload);
415    crc32c::crc32c(&bytes)
416}
417
418#[cfg(test)]
419mod tests {
420    use super::*;
421
422    #[test]
423    fn names_and_idempotency_keys_validate_before_storage() {
424        assert!(StreamName::new("").is_err());
425        assert!(StreamName::new("a\0b").is_err());
426        assert!(StreamName::new("a".repeat(StreamName::MAX_BYTES + 1)).is_err());
427        assert!(IdempotencyKey::new(Vec::<u8>::new()).is_err());
428        assert!(IdempotencyKey::new(vec![0; IdempotencyKey::MAX_BYTES + 1]).is_err());
429    }
430
431    #[test]
432    fn empty_and_oversized_batches_are_rejected() {
433        let request: AppendRequest<()> = AppendRequest {
434            branch: BranchId::ZERO,
435            stream: StreamName::new("s").unwrap(),
436            expected: ExpectedRevision::Any,
437            idempotency_key: None,
438            events: Vec::new(),
439            durability: Durability::Buffered,
440        };
441        assert!(request.validate().is_err());
442
443        let mut request = request;
444        request.events = (0..=AppendRequest::<()>::MAX_EVENTS)
445            .map(|_| NewEvent::new(EventType::new("test").unwrap(), ()))
446            .collect();
447        assert!(request.validate().is_err());
448    }
449}
salamander/stream.rs

salamander/
stream.rs
