drasi_core/interface/
live_results_writer.rs

1// Copyright 2025 The Drasi Authors.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Persistent live results writer trait for reaction recovery.
16//!
17//! The live results store maintains a snapshot of the current query result set.
18//! Each row is keyed by its `row_signature` (a stable u64 hash). On restart,
19//! reactions can read the full snapshot to rebuild their downstream state without
20//! re-processing the entire event history.
21//!
22//! ## Key semantics
23//!
24//! - **apply_mutations**: Atomic batch of upserts (signature → data) and deletes
25//!   (signature → None). Corresponds to the `ResultDiff` entries in a `QueryResult`.
26//! - **read_snapshot**: Returns all currently live rows.
27//! - **clear**: Removes all rows (used during `AutoReset` recovery).
28
29use async_trait::async_trait;
30
31use super::IndexError;
32
33/// A single row mutation: upsert if data is `Some`, delete if `None`.
34#[derive(Debug, Clone)]
35pub struct RowMutation<'a> {
36    /// Stable row identifier (hash of the row's key columns).
37    pub row_signature: u64,
38    /// Serialized row data for upsert, or `None` for delete.
39    pub data: Option<&'a [u8]>,
40}
41
42/// Persistent live results storage for query snapshot recovery.
43///
44/// Implementations store serialized row data keyed by `(query_id, row_signature)`.
45/// The `lib` layer handles serialization/deserialization of row data.
46#[async_trait]
47pub trait LiveResultsWriter: Send + Sync {
48    /// Apply a batch of row mutations atomically.
49    ///
50    /// - `data = Some(bytes)`: Upsert the row (insert or replace).
51    /// - `data = None`: Delete the row if it exists.
52    ///
53    /// Implementations should apply all mutations in a single batch/transaction
54    /// when possible for efficiency.
55    async fn apply_mutations(
56        &self,
57        query_id: &str,
58        mutations: &[RowMutation<'_>],
59    ) -> Result<(), IndexError>;
60
61    /// Read all live rows for a query.
62    ///
63    /// Returns `(row_signature, serialized_data)` pairs in unspecified order.
64    /// Returns an empty vec if no rows exist for this query.
65    async fn read_snapshot(&self, query_id: &str) -> Result<Vec<(u64, Vec<u8>)>, IndexError>;
66
67    /// Delete all live rows for a query.
68    ///
69    /// Used during `AutoReset` recovery and reaction deprovisioning.
70    async fn clear(&self, query_id: &str) -> Result<(), IndexError>;
71
72    /// Count the number of live rows for a query.
73    ///
74    /// Useful for diagnostics and capacity monitoring.
75    async fn row_count(&self, query_id: &str) -> Result<usize, IndexError>;
76}
drasi_core/interface/live_results_writer.rs

drasi_core/interface/
live_results_writer.rs
