ai_store_sync/
sink.rs

1//! Synchronous [`ProjectionSink`] bridge.
2//!
3//! [`SyncProjectionSink`] is the sibling of [`ProjectionSink`] with sync
4//! method signatures. [`BlockingSink`] wraps an implementation of
5//! [`SyncProjectionSink`] and exposes it to the async facade.
6//!
7//! ## Two dispatch modes
8//!
9//! - [`Dispatch::SpawnBlocking`] (default via [`BlockingSink::new`]) hands
10//!   each `commit` / `on_label_set` off to `tokio::task::spawn_blocking`. Use
11//!   this whenever the sync method may block for more than a few hundred
12//!   microseconds — file I/O, `fsync`, database drivers with only sync APIs.
13//!   Requires a runtime with a blocking pool (any `tokio::runtime` built via
14//!   `Builder::new_multi_thread()` or `Builder::new_current_thread()` with
15//!   `enable_all()` has one).
16//!
17//! - [`Dispatch::Inline`] (via [`BlockingSink::inline`]) runs the sync method
18//!   directly on the async worker. Correct only for fast in-memory
19//!   bookkeeping — anything else stalls the runtime.
20//!
21//! ## Idempotence
22//!
23//! [`SyncProjectionSink`] inherits the same idempotence contract as
24//! [`ProjectionSink`]: replaying the same `(stream, seq)` must produce the
25//! same effect as the first application, because `Store::catch_up` and
26//! `Store::rebuild` may re-drive events after a crash or configuration
27//! change.
28
29use std::sync::Arc;
30
31use ai_store_core::{Event, Label, ProjectionSink, Seq, StoreError, StreamId};
32use async_trait::async_trait;
33use serde_json::Value;
34
35/// Synchronous variant of [`ProjectionSink`]. Implement this when the sink's
36/// body is inherently blocking (file I/O, synchronous database drivers,
37/// stdlib `println!`), then wrap in [`BlockingSink`] to plug into the async
38/// facade.
39pub trait SyncProjectionSink: Send + Sync + 'static {
40    /// Stable identifier used as the checkpoint key. See
41    /// [`ProjectionSink::id`].
42    fn id(&self) -> &str;
43
44    /// Apply one committed event. Must be idempotent under retries.
45    fn commit(
46        &self,
47        stream: &StreamId,
48        seq: Seq,
49        state: &Value,
50        event: &Event,
51    ) -> Result<(), StoreError>;
52
53    /// React to a label being pinned or moved. Default is a no-op, matching
54    /// the async trait's default.
55    fn on_label_set(
56        &self,
57        _stream: &StreamId,
58        _label: &Label,
59        _at: Seq,
60        _state: &Value,
61    ) -> Result<(), StoreError> {
62        Ok(())
63    }
64}
65
66/// How [`BlockingSink`] hands off calls to the wrapped
67/// [`SyncProjectionSink`].
68#[derive(Debug, Clone, Copy, PartialEq, Eq)]
69pub enum Dispatch {
70    /// Run the sync method inline on the async worker.
71    ///
72    /// Safe only when the method is guaranteed to return in a few hundred
73    /// microseconds or less. Anything else blocks the runtime.
74    Inline,
75    /// Offload to `tokio::task::spawn_blocking`.
76    ///
77    /// The default. Correct for any sink that touches the file system,
78    /// synchronous DB drivers, or otherwise blocking code.
79    SpawnBlocking,
80}
81
82/// Adapter turning a [`SyncProjectionSink`] into a [`ProjectionSink`].
83pub struct BlockingSink<T: SyncProjectionSink> {
84    inner: Arc<T>,
85    dispatch: Dispatch,
86}
87
88impl<T: SyncProjectionSink> BlockingSink<T> {
89    /// Wrap `inner` in `spawn_blocking` dispatch — the safe default for any
90    /// sink that may block.
91    pub fn new(inner: T) -> Self {
92        Self {
93            inner: Arc::new(inner),
94            dispatch: Dispatch::SpawnBlocking,
95        }
96    }
97
98    /// Wrap `inner` in inline dispatch. Use only for fast in-memory sinks;
99    /// see [`Dispatch::Inline`].
100    pub fn inline(inner: T) -> Self {
101        Self {
102            inner: Arc::new(inner),
103            dispatch: Dispatch::Inline,
104        }
105    }
106
107    /// Access the wrapped sink. Useful for tests that need to peek at
108    /// accumulated state after the async dispatch loop advanced the
109    /// checkpoint.
110    pub fn inner(&self) -> &Arc<T> {
111        &self.inner
112    }
113}
114
115#[async_trait]
116impl<T: SyncProjectionSink> ProjectionSink for BlockingSink<T> {
117    fn id(&self) -> &str {
118        // `id()` on the trait returns a borrowed `&str` from `T`. Since
119        // `self.inner: Arc<T>` is stable for the lifetime of `&self`, this
120        // borrow is sound.
121        self.inner.id()
122    }
123
124    async fn commit(
125        &self,
126        stream: &StreamId,
127        seq: Seq,
128        state: &Value,
129        event: &Event,
130    ) -> Result<(), StoreError> {
131        match self.dispatch {
132            Dispatch::Inline => self.inner.commit(stream, seq, state, event),
133            Dispatch::SpawnBlocking => {
134                let inner = Arc::clone(&self.inner);
135                let stream = stream.clone();
136                let state = state.clone();
137                let event = event.clone();
138                tokio::task::spawn_blocking(move || {
139                    inner.commit(&stream, seq, &state, &event)
140                })
141                .await
142                .map_err(|e| StoreError::Backend(format!("blocking sink join: {e}")))?
143            }
144        }
145    }
146
147    async fn on_label_set(
148        &self,
149        stream: &StreamId,
150        label: &Label,
151        at: Seq,
152        state: &Value,
153    ) -> Result<(), StoreError> {
154        match self.dispatch {
155            Dispatch::Inline => self.inner.on_label_set(stream, label, at, state),
156            Dispatch::SpawnBlocking => {
157                let inner = Arc::clone(&self.inner);
158                let stream = stream.clone();
159                let label = label.clone();
160                let state = state.clone();
161                tokio::task::spawn_blocking(move || {
162                    inner.on_label_set(&stream, &label, at, &state)
163                })
164                .await
165                .map_err(|e| StoreError::Backend(format!("blocking sink join: {e}")))?
166            }
167        }
168    }
169}
ai_store_sync/sink.rs

ai_store_sync/
sink.rs
