exoware_server/
engine.rs

1//! Storage callbacks for the store services.
2//!
3//! Implement the capability traits your component serves. String errors are surfaced to clients as
4//! internal RPC failures (message only; keep messages safe to expose if you rely on that). `Ingest`
5//! additionally lets a backend mark a write failure transient via [`IngestError`].
6
7use std::collections::HashMap;
8use std::future::Future;
9
10use buffa::Message;
11use bytes::Bytes;
12use exoware_sdk::common::kv::v1::Entry;
13use exoware_sdk::log::stream::v1::GetResponse as StreamGetResponse;
14use exoware_sdk::prune_policy::PrunePolicyDocument;
15
16/// Backend-defined query metadata.
17///
18/// Keep this lightweight: streaming query RPCs may emit detail on every frame.
19pub type QueryExtra = HashMap<String, buffa_types::google::protobuf::Value>;
20
21#[derive(Clone, Debug, Default)]
22pub struct RangeScanBatch {
23    /// Rows read by this cursor pull.
24    pub rows: Vec<(Bytes, Bytes)>,
25    /// Backend-specific query metadata after reading these rows.
26    pub extra: QueryExtra,
27}
28
29/// Owned pull-based range cursor for query RPCs.
30///
31/// Implementations own any state needed to produce batches, allowing query
32/// handlers to pull rows lazily without borrowing the engine.
33pub trait RangeScan: Send {
34    /// Pull up to `max_items` rows. Returning an empty `rows` batch marks EOF.
35    /// EOF may carry non-empty `extra` with final query metadata.
36    /// `extra` is emitted with the response frame built from the same batch.
37    fn next_batch(
38        &mut self,
39        max_items: usize,
40    ) -> impl Future<Output = Result<RangeScanBatch, String>> + Send;
41}
42
43/// Local sequence frontier visible to this process.
44pub trait Sequence: Send + Sync + 'static {
45    /// Highest sequence number this process can currently serve.
46    fn current_sequence(&self) -> u64;
47}
48
49/// Why an ingest write was not accepted.
50///
51/// Backends choose the variant; the Connect layer maps it to the wire code and retry details.
52#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
53pub enum IngestError {
54    /// The write cannot currently be accepted; clients may retry with backoff.
55    #[error("unavailable: {message}")]
56    Unavailable { message: String },
57
58    /// The write failed in a way retries will not fix.
59    #[error("internal: {message}")]
60    Internal { message: String },
61}
62
63/// Ingest write capability.
64pub trait Ingest: Send + Sync + 'static {
65    /// Persist key-value pairs atomically and return the global sequence number that includes this
66    /// write. Backends may coalesce concurrent writes and return the same sequence number to each
67    /// coalesced caller.
68    fn put_batch(
69        &self,
70        kvs: Vec<(Bytes, Bytes)>,
71    ) -> impl Future<Output = Result<u64, IngestError>> + Send;
72}
73
74/// Query read capability.
75pub trait Query: Sequence {
76    type RangeScan: RangeScan + 'static;
77
78    /// Fetch the value for a single key plus backend-specific query metadata.
79    /// Returns `None` when the key does not exist.
80    fn get(
81        &self,
82        key: Bytes,
83    ) -> impl Future<Output = Result<(Option<Bytes>, QueryExtra), String>> + Send;
84
85    /// Cursor over keys in `[start, end]` (inclusive) when `end` is non-empty;
86    /// empty `end` means unbounded above. Matches `store.query.v1.RangeRequest`
87    /// / `ReduceRequest` on the wire. `limit` caps rows yielded.
88    fn range_scan(
89        &self,
90        start: Bytes,
91        end: Bytes,
92        limit: usize,
93        forward: bool,
94    ) -> impl Future<Output = Result<Self::RangeScan, String>> + Send;
95
96    /// Batch-get plus backend-specific query metadata. Returns `(key, Option<value>)`
97    /// for each input key, preserving order.
98    fn get_many(
99        &self,
100        keys: Vec<Bytes>,
101    ) -> impl Future<Output = Result<(Vec<(Bytes, Option<Bytes>)>, QueryExtra), String>> + Send;
102}
103
104/// Prune mutation capability.
105pub trait Prune: Send + Sync + 'static {
106    /// Apply a validated prune policy document sequentially.
107    fn apply_prune_policies(
108        &self,
109        document: PrunePolicyDocument,
110    ) -> impl Future<Output = Result<(), String>> + Send;
111}
112
113/// Pre-encoded `log.stream.v1.GetResponse` stored for a retained sequence batch.
114#[derive(Clone, Debug)]
115pub struct LogBatch {
116    sequence_number: u64,
117    response_bytes: Bytes,
118}
119
120impl LogBatch {
121    /// Wrap protobuf bytes already encoded as `log.stream.v1.GetResponse`.
122    pub fn from_response_bytes(sequence_number: u64, response_bytes: impl Into<Bytes>) -> Self {
123        Self {
124            sequence_number,
125            response_bytes: response_bytes.into(),
126        }
127    }
128
129    /// Build a log batch from key-value pairs.
130    pub fn from_entries(sequence_number: u64, kvs: Vec<(Bytes, Bytes)>) -> Self {
131        Self::from_response(StreamGetResponse {
132            sequence_number,
133            entries: kvs
134                .into_iter()
135                .map(|(key, value)| Entry {
136                    key: key.to_vec(),
137                    value,
138                    ..Default::default()
139                })
140                .collect(),
141            ..Default::default()
142        })
143    }
144
145    /// Build a log batch from an owned response.
146    pub fn from_response(response: StreamGetResponse) -> Self {
147        Self {
148            sequence_number: response.sequence_number,
149            response_bytes: Bytes::from(response.encode_to_vec()),
150        }
151    }
152
153    /// Sequence number under which this batch was loaded.
154    pub fn sequence_number(&self) -> u64 {
155        self.sequence_number
156    }
157
158    /// Consume the batch into its pre-encoded response bytes.
159    pub fn into_response_bytes(self) -> Bytes {
160        self.response_bytes
161    }
162
163    /// Decode the stored response for paths that need to inspect entries.
164    pub fn decode_response(&self) -> Result<StreamGetResponse, String> {
165        StreamGetResponse::decode_from_slice(&self.response_bytes)
166            .map_err(|err| format!("failed to decode sequence log value: {err}"))
167    }
168}
169
170/// Retained per-sequence batch-log access for stream replay and lookups.
171pub trait Log: Sequence {
172    /// Return the pre-encoded `GetResponse` for the `put_batch` call that was
173    /// assigned `sequence_number`. Return `Ok(None)` when the batch is not
174    /// available from this log.
175    ///
176    /// Engines that don't retain a log return `Ok(None)` unconditionally,
177    /// which disables `GetBatch` and since-cursored `Subscribe` for that
178    /// deployment.
179    ///
180    /// The stream service maps unavailable batches to `BATCH_NOT_FOUND` when
181    /// they are beyond the visible sequence frontier and `BATCH_EVICTED`
182    /// otherwise.
183    fn get_batch(
184        &self,
185        sequence_number: u64,
186    ) -> impl Future<Output = Result<Option<LogBatch>, String>> + Send;
187
188    /// Lowest retained batch sequence number, or `None` when the log is
189    /// empty. Surfaced in `BATCH_EVICTED` error details so clients know where
190    /// to resume from.
191    fn oldest_retained_batch(&self) -> impl Future<Output = Result<Option<u64>, String>> + Send;
192}
193
194/// Compatibility facade for backends that serve every store capability.
195pub trait StoreEngine: Ingest + Query + Prune + Log {}
196
197impl<T: Ingest + Query + Prune + Log> StoreEngine for T {}
exoware_server/engine.rs

exoware_server/
engine.rs
