krabka_client_streams/store/
api.rs

1//! Store traits.
2//!
3//! `StateStore` is object-safe, so the registry holds it erased. It carries the
4//! changelog hooks. Every #3 store is changelog-logged, so the erased registry
5//! can restore and drain through `&mut dyn StateStore`. `KeyValueStore<K,V>` is
6//! the typed get/put/delete surface.
7
8use std::any::Any;
9
10use async_trait::async_trait;
11
12/// Lifecycle, identity, and changelog hooks for any store.
13#[async_trait]
14pub trait StateStore: Any + Send {
15    fn name(&self) -> &str;
16    /// Flush pending state. This is a no-op for in-memory, where the changelog
17    /// gives durability.
18    async fn flush(&mut self);
19    fn close(&mut self);
20    /// Typed downcast hook that `get_state_store` uses.
21    fn as_any_mut(&mut self) -> &mut dyn Any;
22    /// The store's changelog topic (`<app>-<store>-changelog`).
23    fn changelog_topic(&self) -> &str;
24    /// Drain the buffered changelog entries: key bytes, and value bytes or None
25    /// for a tombstone.
26    fn take_changelog(&mut self) -> Vec<(bytes::Bytes, Option<bytes::Bytes>)>;
27    /// Like `take_changelog`, but each entry also carries an optional explicit
28    /// changelog RECORD timestamp. `None` means use the producer default, which
29    /// is send-time. Versioned stores override this method to emit the version
30    /// timestamp (KIP-889). The default wraps `take_changelog` with `None`.
31    fn take_changelog_ts(&mut self) -> Vec<(bytes::Bytes, Option<bytes::Bytes>, Option<i64>)> {
32        self.take_changelog()
33            .into_iter()
34            .map(|(k, v)| (k, v, None))
35            .collect()
36    }
37    /// Apply a changelog record during restore. It updates state and does NOT
38    /// re-log.
39    async fn apply_changelog(&mut self, key: bytes::Bytes, value: Option<bytes::Bytes>);
40    /// Like `apply_changelog`, but it carries the changelog record's timestamp.
41    /// The default ignores the timestamp and delegates. Versioned stores override
42    /// this method to insert the version at this timestamp.
43    async fn apply_changelog_ts(
44        &mut self,
45        key: bytes::Bytes,
46        value: Option<bytes::Bytes>,
47        _timestamp: i64,
48    ) {
49        self.apply_changelog(key, value).await;
50    }
51    /// Toggle changelog logging. It is off during restore and on during
52    /// processing.
53    fn set_logging(&mut self, on: bool);
54    /// IQ read view, if this store is interactively queryable. Default `None`.
55    fn as_iq(&self) -> Option<&dyn crate::store::iq::IqQueryable> {
56        None
57    }
58    /// Stash the current record's context, so a caching store can attach it to
59    /// the next write. The flush forwards that context with the deduped `Change`.
60    /// The default is a no-op.
61    fn set_record_context(&mut self, _ctx: crate::processor::record::RecordContext) {}
62    /// Erased record-cache hook.
63    ///
64    /// This method wraps the store's backend in the supplied internal
65    /// `NamedCache`, which is registered in the task's `ThreadCache`. It returns
66    /// `true` when this store kind is cache-aware. It lets `instantiate` enable
67    /// caching on a materialized KV store and never learn the store's `K` or `V`.
68    /// The default is `false`, which means not cacheable. Window and session
69    /// stores keep the default until their caching lands. KV stores override this
70    /// method and delegate to their typed `enable_cache`.
71    ///
72    /// `NamedCache` is crate-internal store plumbing. This method is reachable on
73    /// the `pub` trait, but no caller outside the crate can use it.
74    #[allow(private_interfaces)]
75    fn enable_cache_erased(
76        &mut self,
77        _cache: std::sync::Arc<std::sync::Mutex<crate::store::cache::named::NamedCache>>,
78    ) -> bool {
79        false
80    }
81    /// Erased query for whether this store's record cache is enabled.
82    ///
83    /// The cache is enabled when
84    /// [`enable_cache_erased`](Self::enable_cache_erased) has wrapped the
85    /// backend. A materializing processor uses this method to decide whether to
86    /// suppress its immediate downstream forward, and it never learns `K` or `V`.
87    /// The cache flush forwards the deduped `Change`. The default is `false`,
88    /// which means not cached and not cache-aware. KV stores override this
89    /// method.
90    fn is_cached_erased(&self) -> bool {
91        false
92    }
93    /// Flush this store's record cache, if it has one.
94    ///
95    /// This method writes the dirty entries through to the underlying store,
96    /// buffers their changelog records, and pushes the deduped downstream
97    /// `Record<K, Change<V>>` into `buffer`. It pushes one boxed copy PER child
98    /// in `children`, which matches the per-child clone in
99    /// `ProcessorContext::forward`. The default has no cache, so it is a no-op.
100    ///
101    /// `ErasedRecord` is crate-internal graph plumbing. This method is reachable
102    /// on the `pub` trait, but no caller outside the crate can use it.
103    #[allow(private_interfaces)]
104    async fn flush_cache_into(
105        &mut self,
106        _buffer: &mut std::collections::VecDeque<(usize, crate::processor::erased::ErasedRecord)>,
107        _children: &[usize],
108    ) {
109    }
110    /// Wipe every entry and any buffered changelog. The EOS rollback path uses
111    /// this method to reset the store to a clean slate before it re-restores from
112    /// the committed changelog.
113    async fn clear(&mut self);
114    /// Serialize the whole store into one barrier-snapshot payload.
115    ///
116    /// The task calls this method for every store it owns when a barrier fires,
117    /// and it puts the payloads into the frozen snapshot container under the
118    /// store names. See [`crate::store::snapshot`].
119    ///
120    /// The payload holds every piece of state that
121    /// [`restore_snapshot`](Self::restore_snapshot) needs to rebuild the store,
122    /// which for some stores is more than the key-value entries. A suppress
123    /// buffer also carries its sequence counter and its byte-size total.
124    ///
125    /// The payload format is Rust's own. The container around it is shared with
126    /// `krabka-streams-java` and `krabka-streams-go`, the payload inside it is
127    /// not.
128    async fn snapshot(&mut self) -> bytes::Bytes;
129    /// Replace the whole store with a payload that
130    /// [`snapshot`](Self::snapshot) produced.
131    ///
132    /// The store is wiped first, so the restored state is the snapshot and
133    /// nothing else. This method does NOT re-log, exactly as `apply_changelog`
134    /// does not.
135    ///
136    /// # Errors
137    ///
138    /// Returns [`StreamsClientError::Snapshot`](crate::StreamsClientError::Snapshot)
139    /// when the payload does not match the format this store writes.
140    async fn restore_snapshot(
141        &mut self,
142        data: bytes::Bytes,
143    ) -> Result<(), crate::error::StreamsClientError>;
144}
145
146/// A keyed store. The in-memory store implements it, and a processor gets this
147/// typed view from `ProcessorContext::get_state_store`.
148///
149/// `K` needs `Send + Sync` because `get` and `delete` take `&K`, and the boxed
150/// store future must be `Send`. The whole execution chain runs inside
151/// `tokio::spawn`: `Graph::pipe`, then the processors, then the store ops. A
152/// `&K` is only `Send` when `K: Sync`.
153#[async_trait]
154pub trait KeyValueStore<K: Send + Sync, V: Send>: StateStore {
155    async fn get(&self, key: &K) -> Option<V>;
156    async fn put(&mut self, key: K, value: V);
157    async fn delete(&mut self, key: &K) -> Option<V>;
158    /// Half-open range scan `[lo, hi)` in memcmp (lexicographic) key order.
159    async fn range(&self, lo: &K, hi: &K) -> Vec<(K, V)>;
160}
krabka_client_streams/store/api.rs

krabka_client_streams/store/
api.rs
