acyclic_stream/
lib.rs

1#![doc = include_str!("../README.md")]
2
3use std::{fmt, sync::Arc};
4
5use async_trait::async_trait;
6use bytes::Bytes;
7use futures::stream::BoxStream;
8use thiserror::Error;
9
10pub mod conformance;
11#[cfg(feature = "grpc")]
12pub mod grpc;
13#[cfg(feature = "local")]
14mod local;
15mod memory;
16#[cfg(any(feature = "grpc", feature = "local"))]
17mod wire_codec;
18
19/// Generated canonical Stream v2 protocol.
20#[allow(missing_docs, clippy::pedantic, clippy::too_many_lines)]
21pub mod wire {
22    include!(concat!(env!("OUT_DIR"), "/acyclic.stream.v2.rs"));
23}
24/// Canonical public descriptor set used by compatibility gates.
25pub const FILE_DESCRIPTOR_SET: &[u8] = include_bytes!("../proto/stream/v2/stream_descriptor.bin");
26#[cfg(feature = "local")]
27pub use local::{LocalDurability, LocalStream, LocalStreamError, LocalStreamLimits};
28pub use memory::{MemoryLimits, MemoryStream};
29
30/// Maximum opaque record body.
31pub const MAX_RECORD_BYTES: usize = 64 * 1024;
32/// Maximum records, participants, mutations, or path segments in one request.
33pub const MAX_ITEMS: usize = 1_024;
34/// Maximum canonical application command, including metadata.
35pub const MAX_COMMAND_BYTES: usize = 1024 * 1024 + 8 * 1024;
36/// Minimum durable replay window required from a provider.
37pub const MIN_IDEMPOTENCY_RETENTION_SECS: u64 = 24 * 60 * 60;
38/// Maximum caller retry-identity width.
39pub const MAX_IDEMPOTENCY_KEY_BYTES: usize = 256;
40/// Maximum canonical path text accepted by the single-command wire format.
41pub const MAX_PATH_BYTES: usize = u16::MAX as usize;
42
43/// Permanent account-relative slash-separated ASCII path.
44#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
45pub struct StreamPath(Arc<str>);
46
47impl StreamPath {
48    /// Validates and owns one path.
49    pub fn new(path: impl AsRef<str>) -> Result<Self, StreamError> {
50        let path = path.as_ref();
51        if path.is_empty()
52            || path.len() > MAX_PATH_BYTES
53            || !path.is_ascii()
54            || path.starts_with('/')
55            || path.ends_with('/')
56        {
57            return Err(StreamError::InvalidPath);
58        }
59        let mut count = 0_usize;
60        for segment in path.split('/') {
61            count = count.checked_add(1).ok_or(StreamError::LimitExceeded)?;
62            if segment.is_empty()
63                || segment == "."
64                || segment == ".."
65                || segment
66                    .bytes()
67                    .any(|byte| byte.is_ascii_control() || byte.is_ascii_whitespace())
68                || segment.as_bytes().contains(&b'\\')
69            {
70                return Err(StreamError::InvalidPath);
71            }
72        }
73        if count > MAX_ITEMS {
74            return Err(StreamError::LimitExceeded);
75        }
76        Ok(Self(Arc::from(path)))
77    }
78
79    /// Canonical path text.
80    #[must_use]
81    pub fn as_str(&self) -> &str {
82        &self.0
83    }
84
85    /// Immediate parent, if any.
86    #[must_use]
87    #[allow(
88        clippy::string_slice,
89        reason = "`new` rejects any path that is not `is_ascii()`, so every byte offset from \
90                  `rfind` is a valid char boundary and this can never panic"
91    )]
92    pub fn parent(&self) -> Option<Self> {
93        self.0
94            .rfind('/')
95            .map(|index| Self(Arc::from(&self.0[..index])))
96    }
97}
98
99impl fmt::Display for StreamPath {
100    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
101        formatter.write_str(self.as_str())
102    }
103}
104
105/// Opaque content-bound identity of one committed envelope.
106#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
107pub struct CommitId([u8; 32]);
108
109impl CommitId {
110    /// Constructs an ID from exact bytes.
111    #[must_use]
112    pub const fn from_bytes(bytes: [u8; 32]) -> Self {
113        Self(bytes)
114    }
115
116    /// Returns exact bytes.
117    #[must_use]
118    pub const fn as_bytes(&self) -> &[u8; 32] {
119        &self.0
120    }
121}
122
123/// Account-scoped stable retry identity.
124#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
125pub struct IdempotencyKey(Bytes);
126
127impl IdempotencyKey {
128    /// Constructs a nonempty bounded key.
129    pub fn new(value: impl Into<Bytes>) -> Result<Self, StreamError> {
130        let value = value.into();
131        if value.is_empty() || value.len() > MAX_IDEMPOTENCY_KEY_BYTES {
132            return Err(StreamError::InvalidArgument);
133        }
134        Ok(Self(value))
135    }
136
137    /// Exact caller bytes.
138    #[must_use]
139    pub fn as_bytes(&self) -> &[u8] {
140        &self.0
141    }
142}
143
144/// One immutable record.
145#[derive(Clone, Debug, Eq, PartialEq)]
146pub struct Record {
147    /// Zero-based sequence.
148    pub sequence: u64,
149    /// Shared opaque value.
150    pub value: Bytes,
151    /// Envelope that introduced the record.
152    pub commit_id: CommitId,
153}
154
155/// Successful contiguous append.
156#[derive(Clone, Debug, Eq, PartialEq)]
157pub struct AppendReceipt {
158    /// First sequence.
159    pub start: u64,
160    /// Exclusive end sequence.
161    pub end: u64,
162    /// Resulting tail.
163    pub tail: u64,
164    /// Immutable envelope identity.
165    pub commit_id: CommitId,
166}
167
168/// Tail-CAS append outcome.
169#[derive(Clone, Debug, Eq, PartialEq)]
170pub enum AppendOutcome {
171    /// Entire batch committed.
172    Committed(AppendReceipt),
173    /// Tail differed and nothing changed.
174    TailConflict {
175        /// Linearizable tail observed by the provider.
176        actual_tail: u64,
177    },
178}
179
180/// One atomic append request.
181#[derive(Clone, Debug, Eq, PartialEq)]
182pub struct AppendRequest {
183    /// Permanent destination.
184    pub path: StreamPath,
185    /// Nonempty contiguous batch.
186    pub records: Vec<Bytes>,
187    /// Optional exact tail condition.
188    pub if_tail: Option<u64>,
189    /// Optional stable recovery identity.
190    pub idempotency_key: Option<IdempotencyKey>,
191}
192
193/// Successful O(1) immutable-prefix fork.
194#[derive(Clone, Debug, Eq, PartialEq)]
195pub struct ForkReceipt {
196    /// Source path.
197    pub source: StreamPath,
198    /// New destination path.
199    pub destination: StreamPath,
200    /// Exclusive inherited prefix end.
201    pub forked_at: u64,
202    /// Initial destination tail.
203    pub tail: u64,
204    /// Immutable envelope identity.
205    pub commit_id: CommitId,
206}
207
208/// One fork request. Absence of `at_tail` selects the source tail atomically.
209#[derive(Clone, Debug, Eq, PartialEq)]
210pub struct ForkRequest {
211    /// Existing source.
212    pub source: StreamPath,
213    /// Absent destination.
214    pub destination: StreamPath,
215    /// Optional exact retained prefix.
216    pub at_tail: Option<u64>,
217    /// Optional stable recovery identity.
218    pub idempotency_key: Option<IdempotencyKey>,
219}
220
221/// Successful monotonic logical trim.
222#[derive(Clone, Debug, Eq, PartialEq)]
223pub struct TrimReceipt {
224    /// Trimmed path.
225    pub path: StreamPath,
226    /// New earliest readable sequence.
227    pub trim_point: u64,
228    /// Immutable envelope identity.
229    pub commit_id: CommitId,
230}
231
232/// Successful permanent logical deletion.
233#[derive(Clone, Debug, Eq, PartialEq)]
234pub struct DeleteReceipt {
235    /// Retired permanent path.
236    pub path: StreamPath,
237    /// Immutable envelope identity.
238    pub commit_id: CommitId,
239}
240
241/// Required-bounds finite read.
242#[derive(Clone, Debug, Eq, PartialEq)]
243pub struct ReadRequest {
244    /// Stream path.
245    pub path: StreamPath,
246    /// First sequence, inclusive.
247    pub from: u64,
248    /// Nonzero record bound.
249    pub limit: u32,
250}
251
252/// One immutable direct child.
253#[derive(Clone, Debug, Eq, PartialEq)]
254pub struct Child {
255    /// Exact child path.
256    pub path: StreamPath,
257}
258
259/// Fixed-snapshot direct-child page request.
260#[derive(Clone, Debug, Eq, PartialEq)]
261pub struct ChildrenRequest {
262    /// Parent, or `None` for top-level paths.
263    pub parent: Option<StreamPath>,
264    /// Nonzero result bound.
265    pub limit: u32,
266}
267
268/// Append fact retained in a committed envelope.
269#[derive(Clone, Debug, Eq, PartialEq)]
270pub struct CommittedAppend {
271    /// Destination.
272    pub path: StreamPath,
273    /// First sequence.
274    pub start: u64,
275    /// Exclusive end.
276    pub end: u64,
277    /// Resulting tail.
278    pub tail: u64,
279    /// Records carrying the envelope ID.
280    pub records: Vec<Record>,
281}
282
283/// Fork fact retained in a committed envelope.
284#[derive(Clone, Debug, Eq, PartialEq)]
285pub struct CommittedFork {
286    /// Source.
287    pub source: StreamPath,
288    /// Destination.
289    pub destination: StreamPath,
290    /// Exclusive inherited prefix end.
291    pub forked_at: u64,
292    /// Initial destination tail.
293    pub tail: u64,
294}
295
296/// Logical trim fact retained in a committed envelope.
297#[derive(Clone, Debug, Eq, PartialEq)]
298pub struct CommittedTrim {
299    /// Trimmed path.
300    pub path: StreamPath,
301    /// New earliest readable sequence.
302    pub trim_point: u64,
303}
304
305/// Permanent deletion fact retained in a committed envelope.
306#[derive(Clone, Debug, Eq, PartialEq)]
307pub struct CommittedDelete {
308    /// Retired path.
309    pub path: StreamPath,
310}
311
312/// Mutation in a successful immutable envelope.
313#[derive(Clone, Debug, Eq, PartialEq)]
314pub enum CommittedMutation {
315    /// Contiguous append.
316    Append(CommittedAppend),
317    /// Immutable-prefix fork.
318    Fork(CommittedFork),
319    /// Logical prefix trim.
320    Trim(CommittedTrim),
321    /// Permanent path retirement.
322    Delete(CommittedDelete),
323}
324
325/// Complete immutable successful mutation envelope.
326#[derive(Clone, Debug, Eq, PartialEq)]
327pub struct CommittedEnvelope {
328    /// Content-bound identity.
329    pub commit_id: CommitId,
330    /// Canonically ordered mutation facts.
331    pub mutations: Vec<CommittedMutation>,
332}
333
334/// Exact optimistic condition.
335#[derive(Clone, Debug, Eq, PartialEq)]
336pub enum CommitCondition {
337    /// Existing tail must match.
338    Tail {
339        /// Existing path.
340        path: StreamPath,
341        /// Required exact tail.
342        expected: u64,
343    },
344    /// Path must never have existed or been retired.
345    Absent {
346        /// Permanently named path that must be unused.
347        path: StreamPath,
348    },
349}
350
351/// Mutation in one coordinated commit.
352#[derive(Clone, Debug, Eq, PartialEq)]
353pub enum CommitMutation {
354    /// Append one nonempty batch.
355    Append {
356        /// Destination.
357        path: StreamPath,
358        /// Opaque records.
359        records: Vec<Bytes>,
360    },
361    /// Fork one pre-commit prefix.
362    Fork {
363        /// Source.
364        source: StreamPath,
365        /// Destination.
366        destination: StreamPath,
367        /// Exact source prefix end.
368        at_tail: u64,
369    },
370    /// Advance one path's logical trim point.
371    Trim {
372        /// Existing path.
373        path: StreamPath,
374        /// New earliest readable sequence.
375        before: u64,
376    },
377    /// Permanently retire one path.
378    Delete {
379        /// Existing path without live descendants.
380        path: StreamPath,
381    },
382}
383
384/// One bounded all-or-nothing optimistic commit.
385#[derive(Clone, Debug, Eq, PartialEq)]
386pub struct CommitRequest {
387    /// Exact conditions for every participant.
388    pub conditions: Vec<CommitCondition>,
389    /// Nonempty mutation set.
390    pub mutations: Vec<CommitMutation>,
391    /// Required stable recovery identity.
392    pub idempotency_key: IdempotencyKey,
393}
394
395/// Trusted clock used by providers to evaluate publication deadlines at the
396/// same linearization point as a coordinated commit.
397pub trait UnixMillisClock: Send + Sync + 'static {
398    /// Current Unix time in milliseconds.
399    fn now_unix_millis(&self) -> u64;
400}
401
402/// Production wall clock for deadline-aware providers.
403#[derive(Clone, Copy, Debug, Default)]
404pub struct SystemUnixMillisClock;
405
406impl UnixMillisClock for SystemUnixMillisClock {
407    fn now_unix_millis(&self) -> u64 {
408        std::time::SystemTime::now()
409            .duration_since(std::time::UNIX_EPOCH)
410            .map_or(0, |duration| {
411                u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
412            })
413    }
414}
415
416/// Failed exact condition.
417#[derive(Clone, Debug, Eq, PartialEq)]
418pub enum CommitConflict {
419    /// Tail differed.
420    Tail {
421        /// Path.
422        path: StreamPath,
423        /// Required tail.
424        expected: u64,
425        /// Observed tail, or absence when the path does not exist.
426        actual: Option<u64>,
427    },
428    /// Requested absent path exists.
429    Exists {
430        /// Path that already exists.
431        path: StreamPath,
432    },
433    /// Requested absent path or ancestor is retired.
434    Retired {
435        /// Retired path or descendant of a retired path.
436        path: StreamPath,
437    },
438}
439
440/// Coordinated commit outcome.
441#[derive(Clone, Debug, Eq, PartialEq)]
442pub enum CommitOutcome {
443    /// Every mutation committed.
444    Committed(CommittedEnvelope),
445    /// Nothing changed.
446    Conflict(Vec<CommitConflict>),
447}
448
449/// Terminal result retained under one account-scoped retry identity.
450#[derive(Clone, Debug, Eq, PartialEq)]
451pub enum IdempotencyOutcome {
452    /// Append success or tail conflict.
453    Append(AppendOutcome),
454    /// Successful immutable-prefix fork.
455    Fork(ForkReceipt),
456    /// Successful logical trim.
457    Trim(TrimReceipt),
458    /// Successful permanent deletion.
459    Delete(DeleteReceipt),
460    /// Coordinated commit success or exact conflict.
461    Commit(CommitOutcome),
462}
463
464/// Exact retained recovery fact for one caller-owned retry identity.
465#[derive(Clone, Debug, Eq, PartialEq)]
466pub struct IdempotencyObservation {
467    /// Account-scoped identity supplied by the caller.
468    pub idempotency_key: IdempotencyKey,
469    /// Digest of the canonical request arguments bound to the identity.
470    pub request_digest: [u8; 32],
471    /// Original terminal result.
472    pub outcome: IdempotencyOutcome,
473}
474
475/// Backpressured finite read or long-lived follow.
476pub type RecordStream = BoxStream<'static, Result<Record, StreamError>>;
477/// Backpressured fixed-snapshot direct-child listing.
478pub type ChildStream = BoxStream<'static, Result<Child, StreamError>>;
479
480/// Canonical provider contract. Placement and transport remain invisible.
481#[async_trait]
482pub trait StreamProvider: Send + Sync + 'static {
483    /// Reads the retained terminal result for one caller-owned retry identity.
484    async fn inspect_idempotency(
485        &self,
486        idempotency_key: IdempotencyKey,
487    ) -> Result<Option<IdempotencyObservation>, StreamError>;
488    /// Current next sequence.
489    async fn tail(&self, path: StreamPath) -> Result<u64, StreamError>;
490    /// Atomic append or tail conflict.
491    async fn append(&self, request: AppendRequest) -> Result<AppendOutcome, StreamError>;
492    /// Atomic immutable-prefix fork.
493    async fn fork(&self, request: ForkRequest) -> Result<ForkReceipt, StreamError>;
494    /// Monotonically advances a path's logical trim point.
495    async fn trim(
496        &self,
497        path: StreamPath,
498        before: u64,
499        idempotency_key: IdempotencyKey,
500    ) -> Result<TrimReceipt, StreamError>;
501    /// Permanently retires a path with no live descendants.
502    async fn delete(
503        &self,
504        path: StreamPath,
505        idempotency_key: IdempotencyKey,
506    ) -> Result<DeleteReceipt, StreamError>;
507    /// Opens a bounded finite read.
508    async fn read(&self, request: ReadRequest) -> Result<RecordStream, StreamError>;
509    /// Replays and then remains live without a handoff gap.
510    async fn follow(&self, path: StreamPath, from: u64) -> Result<RecordStream, StreamError>;
511    /// Lists one fixed-snapshot direct-child page.
512    async fn children(&self, request: ChildrenRequest) -> Result<ChildStream, StreamError>;
513    /// Executes one all-or-nothing optimistic commit.
514    async fn commit(&self, request: CommitRequest) -> Result<CommitOutcome, StreamError>;
515    /// Executes one coordinated commit only if the provider's trusted clock is
516    /// strictly before `deadline_unix_millis` at the linearization point.
517    ///
518    /// Exact idempotent replay is resolved before the deadline. Providers that
519    /// cannot enforce this atomically fail closed with [`StreamError::Unsupported`].
520    async fn commit_before(
521        &self,
522        request: CommitRequest,
523        deadline_unix_millis: u64,
524    ) -> Result<CommitOutcome, StreamError> {
525        let _ = (request, deadline_unix_millis);
526        Err(StreamError::Unsupported)
527    }
528    /// Reads one complete immutable successful envelope.
529    async fn read_commit(&self, commit_id: CommitId) -> Result<CommittedEnvelope, StreamError>;
530}
531
532/// Minimal provider-bound client.
533pub struct StreamClient<P> {
534    provider: Arc<P>,
535}
536
537impl<P> Clone for StreamClient<P> {
538    fn clone(&self) -> Self {
539        Self {
540            provider: Arc::clone(&self.provider),
541        }
542    }
543}
544
545impl<P: StreamProvider> StreamClient<P> {
546    /// Binds one already authenticated provider.
547    #[must_use]
548    pub fn new(provider: Arc<P>) -> Self {
549        Self { provider }
550    }
551
552    /// Opens one path handle after local validation.
553    pub fn stream(&self, path: impl AsRef<str>) -> Result<Stream<P>, StreamError> {
554        Ok(Stream {
555            client: Self::new(Arc::clone(&self.provider)),
556            path: StreamPath::new(path)?,
557        })
558    }
559
560    /// Reads the retained terminal result for one caller-owned retry identity.
561    pub async fn inspect_idempotency(
562        &self,
563        idempotency_key: IdempotencyKey,
564    ) -> Result<Option<IdempotencyObservation>, StreamError> {
565        self.provider.inspect_idempotency(idempotency_key).await
566    }
567
568    /// Lists direct children from one fixed snapshot.
569    pub async fn children(
570        &self,
571        parent: Option<&str>,
572        limit: u32,
573    ) -> Result<ChildStream, StreamError> {
574        self.provider
575            .children(ChildrenRequest {
576                parent: parent.map(StreamPath::new).transpose()?,
577                limit,
578            })
579            .await
580    }
581
582    /// Executes a coordinated commit.
583    pub async fn commit(&self, request: CommitRequest) -> Result<CommitOutcome, StreamError> {
584        self.provider.commit(request).await
585    }
586
587    /// Executes one coordinated commit under the provider's trusted deadline.
588    pub async fn commit_before(
589        &self,
590        request: CommitRequest,
591        deadline_unix_millis: u64,
592    ) -> Result<CommitOutcome, StreamError> {
593        self.provider
594            .commit_before(request, deadline_unix_millis)
595            .await
596    }
597
598    /// Reads a committed envelope.
599    pub async fn read_commit(&self, commit_id: CommitId) -> Result<CommittedEnvelope, StreamError> {
600        self.provider.read_commit(commit_id).await
601    }
602}
603
604#[cfg(feature = "grpc")]
605impl StreamClient<grpc::Client> {
606    /// Connects the high-level API to an authenticated managed or customer-hosted endpoint.
607    pub async fn connect(
608        endpoint: impl AsRef<str>,
609        bearer_token: impl AsRef<str>,
610    ) -> Result<Self, grpc::ConnectError> {
611        Ok(Self::new(Arc::new(
612            grpc::Client::connect(endpoint, bearer_token).await?,
613        )))
614    }
615
616    /// Connects the high-level API through ambient roots plus one caller-supplied private CA.
617    pub async fn connect_with_ca_certificate(
618        endpoint: impl AsRef<str>,
619        bearer_token: impl AsRef<str>,
620        certificate_pem: impl AsRef<[u8]>,
621    ) -> Result<Self, grpc::ConnectError> {
622        Ok(Self::new(Arc::new(
623            grpc::Client::connect_with_ca_certificate(endpoint, bearer_token, certificate_pem)
624                .await?,
625        )))
626    }
627}
628
629/// Handle to one permanent Stream path.
630pub struct Stream<P> {
631    client: StreamClient<P>,
632    path: StreamPath,
633}
634
635impl<P> Clone for Stream<P> {
636    fn clone(&self) -> Self {
637        Self {
638            client: self.client.clone(),
639            path: self.path.clone(),
640        }
641    }
642}
643
644impl<P: StreamProvider> Stream<P> {
645    /// Permanent path.
646    #[must_use]
647    pub const fn path(&self) -> &StreamPath {
648        &self.path
649    }
650
651    /// Current tail.
652    pub async fn tail(&self) -> Result<u64, StreamError> {
653        self.client.provider.tail(self.path.clone()).await
654    }
655
656    /// Unconditionally appends one record.
657    pub async fn append(&self, value: impl Into<Bytes>) -> Result<AppendOutcome, StreamError> {
658        self.append_batch(vec![value.into()], None, None).await
659    }
660
661    /// Appends one record only at the exact tail.
662    pub async fn append_at(
663        &self,
664        value: impl Into<Bytes>,
665        if_tail: u64,
666    ) -> Result<AppendOutcome, StreamError> {
667        self.append_batch(vec![value.into()], Some(if_tail), None)
668            .await
669    }
670
671    /// Atomically appends a contiguous batch.
672    pub async fn append_batch(
673        &self,
674        records: Vec<Bytes>,
675        if_tail: Option<u64>,
676        idempotency_key: Option<IdempotencyKey>,
677    ) -> Result<AppendOutcome, StreamError> {
678        self.client
679            .provider
680            .append(AppendRequest {
681                path: self.path.clone(),
682                records,
683                if_tail,
684                idempotency_key: Some(idempotency_key.unwrap_or_else(new_idempotency_key)),
685            })
686            .await
687    }
688
689    /// Forks the current or selected retained prefix.
690    pub async fn fork(
691        &self,
692        destination: impl AsRef<str>,
693        at_tail: Option<u64>,
694        idempotency_key: Option<IdempotencyKey>,
695    ) -> Result<ForkReceipt, StreamError> {
696        self.client
697            .provider
698            .fork(ForkRequest {
699                source: self.path.clone(),
700                destination: StreamPath::new(destination)?,
701                at_tail,
702                idempotency_key: Some(idempotency_key.unwrap_or_else(new_idempotency_key)),
703            })
704            .await
705    }
706
707    /// Monotonically advances the earliest readable sequence.
708    pub async fn trim(
709        &self,
710        before: u64,
711        idempotency_key: Option<IdempotencyKey>,
712    ) -> Result<TrimReceipt, StreamError> {
713        self.client
714            .provider
715            .trim(
716                self.path.clone(),
717                before,
718                idempotency_key.unwrap_or_else(new_idempotency_key),
719            )
720            .await
721    }
722
723    /// Permanently retires this path.
724    pub async fn delete(
725        &self,
726        idempotency_key: Option<IdempotencyKey>,
727    ) -> Result<DeleteReceipt, StreamError> {
728        self.client
729            .provider
730            .delete(
731                self.path.clone(),
732                idempotency_key.unwrap_or_else(new_idempotency_key),
733            )
734            .await
735    }
736
737    /// Reads at most `limit` records from `from`.
738    pub async fn read(&self, from: u64, limit: u32) -> Result<RecordStream, StreamError> {
739        self.client
740            .provider
741            .read(ReadRequest {
742                path: self.path.clone(),
743                from,
744                limit,
745            })
746            .await
747    }
748
749    /// Replays from `from`, then remains live.
750    pub async fn follow(&self, from: u64) -> Result<RecordStream, StreamError> {
751        self.client.provider.follow(self.path.clone(), from).await
752    }
753}
754
755fn new_idempotency_key() -> IdempotencyKey {
756    IdempotencyKey(Bytes::copy_from_slice(uuid::Uuid::new_v4().as_bytes()))
757}
758
759/// Stable public failures. CAS and absence conflicts are values instead.
760#[derive(Clone, Debug, Eq, Error, PartialEq)]
761pub enum StreamError {
762    /// Invalid path syntax.
763    #[error("invalid stream path")]
764    InvalidPath,
765    /// Invalid operation shape.
766    #[error("invalid stream argument")]
767    InvalidArgument,
768    /// Count or byte ceiling exceeded.
769    #[error("stream limit exceeded")]
770    LimitExceeded,
771    /// Path does not exist.
772    #[error("stream not found")]
773    NotFound,
774    /// Destination already exists.
775    #[error("stream already exists")]
776    AlreadyExists,
777    /// Path or ancestor is permanently retired.
778    #[error("stream path retired")]
779    Retired,
780    /// Requested source prefix is not retained.
781    #[error("stream prefix not retained")]
782    PrefixNotRetained,
783    /// Requested sequence is beyond the current tail.
784    #[error("stream sequence out of range")]
785    OutOfRange,
786    /// Retry identity was reused with different arguments.
787    #[error("idempotency mismatch")]
788    IdempotencyMismatch,
789    /// Provider's bounded retained state is exhausted.
790    #[error("stream capacity exhausted")]
791    Capacity,
792    /// Access denied.
793    #[error("stream access denied")]
794    AccessDenied,
795    /// Required authority is unavailable.
796    #[error("stream unavailable")]
797    Unavailable,
798    /// A provider-evaluated commit deadline elapsed before linearization.
799    #[error("stream commit deadline elapsed")]
800    DeadlineElapsed,
801    /// The provider cannot supply a required semantic capability.
802    #[error("stream capability unsupported")]
803    Unsupported,
804}
acyclic_stream/lib.rs

acyclic_stream/
lib.rs
