nodedb_lite/nodedb/collection/
kv_sync.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Sync recording for the public KV API, and the KV counter.
4//!
5//! With sync on, a public put or delete records itself in the KV outbound
6//! queue before it applies locally. The push loop sends each record to
7//! Origin as a `KvPushMsg`. Every public write holds the write-order lock
8//! across the record and the apply, so the queue order is the apply order.
9//!
10//! A full outbound queue refuses the write with `Backpressure` before it
11//! changes local state.
12
13use nodedb_physical::kv_atomic::compute;
14use nodedb_physical::physical_plan::KvCounterShape;
15use nodedb_types::Namespace;
16use nodedb_types::error::{NodeDbError, NodeDbResult};
17
18use super::super::{LockExt, NodeDbLite};
19use super::kv::{decode_value, is_expired, kv_key};
20use crate::query::kv_ops::writes::atomic_error;
21use crate::storage::engine::StorageEngine;
22
23impl<S: StorageEngine> NodeDbLite<S> {
24    /// KV INCREMENT: add `delta` to the counter at `key` and return the new
25    /// value.
26    ///
27    /// The counter follows the rules the SQL `INCR` follows:
28    /// - a raw value is decimal text in and decimal text out;
29    /// - a typed row moves its first integer column in key order;
30    /// - an absent or expired key starts at 0 and stores decimal text;
31    /// - the key keeps its expiry.
32    ///
33    /// A value that is not a counter, or a sum outside `i64`, is an error
34    /// and writes nothing. The new value syncs to Origin like any put.
35    pub async fn kv_increment(&self, collection: &str, key: &str, delta: i64) -> NodeDbResult<i64> {
36        self.kv_check_pressure()?;
37        let _order = self.kv_local.write_order.lock().await;
38        let (current, deadline) = self.kv_live_entry(collection, key).await?;
39        let (value, written) = compute::incr(current.as_deref(), delta, &KvCounterShape::Raw)
40            .map_err(|e| NodeDbError::from(atomic_error(collection, e)))?;
41        self.kv_record_stored(collection, key.as_bytes(), &written, deadline)
42            .await?;
43        self.kv_buffer_put(collection, key, &written, deadline)
44            .await?;
45        Ok(value)
46    }
47
48    /// The live value at `key` and its expiry deadline, reading the write
49    /// buffer first. An absent or expired key reads as `(None, 0)`.
50    async fn kv_live_entry(
51        &self,
52        collection: &str,
53        key: &str,
54    ) -> NodeDbResult<(Option<Vec<u8>>, u64)> {
55        let rkey = kv_key(collection, key.as_bytes());
56        let buffered: Option<Option<Vec<u8>>> = {
57            let buf = self.kv_local.write_buf.lock_or_recover();
58            buf.overlay.get(&rkey).cloned()
59        };
60        let stored = match buffered {
61            Some(entry) => entry,
62            None => self
63                .storage
64                .get(Namespace::Kv, &rkey)
65                .await
66                .map_err(NodeDbError::storage)?,
67        };
68        let Some(raw) = stored else {
69            return Ok((None, 0));
70        };
71        let (deadline, body) = decode_value(&raw).ok_or_else(|| {
72            NodeDbError::storage(format!(
73                "corrupt KV entry for key '{key}' in collection '{collection}'"
74            ))
75        })?;
76        if is_expired(deadline) {
77            Ok((None, 0))
78        } else {
79            Ok((Some(body.to_vec()), deadline))
80        }
81    }
82
83    /// Record a put of the raw value `value` for sync. A no-op with sync
84    /// off.
85    pub(super) async fn kv_record_put(
86        &self,
87        collection: &str,
88        key: &[u8],
89        value: &[u8],
90        deadline_ms: u64,
91    ) -> NodeDbResult<()> {
92        #[cfg(not(target_arch = "wasm32"))]
93        if let Some(queue) = &self.kv_outbound {
94            let op = crate::sync::PendingKvOp::put_raw(key, value, deadline_ms)?;
95            queue
96                .enqueue(&crate::sync::PendingKvWrite::new(collection, key, op))
97                .await?;
98        }
99        #[cfg(target_arch = "wasm32")]
100        let _ = (collection, key, value, deadline_ms);
101        Ok(())
102    }
103
104    /// Record a put of the stored body `body` for sync. A typed row body
105    /// is sent as its columns. A no-op with sync off.
106    async fn kv_record_stored(
107        &self,
108        collection: &str,
109        key: &[u8],
110        body: &[u8],
111        deadline_ms: u64,
112    ) -> NodeDbResult<()> {
113        #[cfg(not(target_arch = "wasm32"))]
114        if let Some(queue) = &self.kv_outbound {
115            let op = crate::sync::PendingKvOp::put_stored(key, body, deadline_ms)?;
116            queue
117                .enqueue(&crate::sync::PendingKvWrite::new(collection, key, op))
118                .await?;
119        }
120        #[cfg(target_arch = "wasm32")]
121        let _ = (collection, key, body, deadline_ms);
122        Ok(())
123    }
124
125    /// Record a delete of `key` for sync. A no-op with sync off.
126    pub(super) async fn kv_record_delete(&self, collection: &str, key: &[u8]) -> NodeDbResult<()> {
127        #[cfg(not(target_arch = "wasm32"))]
128        if let Some(queue) = &self.kv_outbound {
129            queue
130                .enqueue(&crate::sync::PendingKvWrite::new(
131                    collection,
132                    key,
133                    crate::sync::PendingKvOp::Delete,
134                ))
135                .await?;
136        }
137        #[cfg(target_arch = "wasm32")]
138        let _ = (collection, key);
139        Ok(())
140    }
141}
142
143#[cfg(test)]
144mod tests {
145    use std::sync::Arc;
146
147    use nodedb_types::value::Value;
148
149    use crate::config::LiteConfig;
150    use crate::storage::pagedb_storage::PagedbStorageMem;
151    use crate::sync::{PendingKvOp, PendingKvWrite};
152
153    use super::*;
154
155    async fn open_db(sync_enabled: bool) -> Arc<NodeDbLite<PagedbStorageMem>> {
156        let storage = PagedbStorageMem::open_in_memory().await.expect("storage");
157        let config = LiteConfig {
158            sync_enabled,
159            ..LiteConfig::default()
160        };
161        NodeDbLite::open_with_config(storage, config)
162            .await
163            .expect("open")
164    }
165
166    async fn queued(db: &NodeDbLite<PagedbStorageMem>) -> Vec<PendingKvWrite> {
167        let queue = db.kv_outbound.as_ref().expect("sync on opens the queue");
168        queue
169            .drain(usize::MAX)
170            .await
171            .expect("drain")
172            .into_iter()
173            .map(|(_, write)| write)
174            .collect()
175    }
176
177    fn row_value(write: &PendingKvWrite, column: &str) -> Option<Value> {
178        match &write.op {
179            PendingKvOp::Put { row, .. } => match nodedb_types::value_from_msgpack(row) {
180                Ok(Value::Object(map)) => map.get(column).cloned(),
181                other => panic!("queued row is not a map: {other:?}"),
182            },
183            PendingKvOp::Delete => None,
184        }
185    }
186
187    #[tokio::test]
188    async fn a_put_and_a_delete_queue_in_write_order() {
189        let db = open_db(true).await;
190        db.kv_put("cfg", "k1", b"v1").await.expect("put");
191        db.kv_delete("cfg", "k1").await.expect("delete");
192
193        let writes = queued(&db).await;
194        assert_eq!(writes.len(), 2);
195        assert_eq!(writes[0].collection, "cfg");
196        assert_eq!(writes[0].key, b"k1".to_vec());
197        assert_eq!(
198            row_value(&writes[0], "value"),
199            Some(Value::String("v1".into()))
200        );
201        assert_eq!(writes[1].op, PendingKvOp::Delete);
202        assert_eq!(writes[0].seq, 0, "a seq is assigned at first send");
203    }
204
205    #[tokio::test]
206    async fn a_put_with_ttl_carries_its_deadline() {
207        let db = open_db(true).await;
208        db.kv_put_with_ttl("cfg", "k1", b"v1", 60_000)
209            .await
210            .expect("put");
211        let writes = queued(&db).await;
212        match &writes[0].op {
213            PendingKvOp::Put { expire_at_ms, .. } => assert!(*expire_at_ms > 0),
214            PendingKvOp::Delete => panic!("expected a put"),
215        }
216    }
217
218    #[tokio::test]
219    async fn an_increment_stores_decimal_text_and_queues_the_new_value() {
220        let db = open_db(true).await;
221        assert_eq!(db.kv_increment("ctr", "hits", 5).await.expect("incr"), 5);
222        assert_eq!(db.kv_increment("ctr", "hits", -2).await.expect("incr"), 3);
223        assert_eq!(
224            db.kv_get("ctr", "hits").await.expect("get"),
225            Some(b"3".to_vec())
226        );
227
228        let writes = queued(&db).await;
229        assert_eq!(writes.len(), 2);
230        assert_eq!(
231            row_value(&writes[1], "value"),
232            Some(Value::String("3".into()))
233        );
234    }
235
236    #[tokio::test]
237    async fn an_increment_of_a_non_counter_writes_nothing() {
238        let db = open_db(true).await;
239        db.kv_put("ctr", "name", b"alice").await.expect("put");
240        assert!(db.kv_increment("ctr", "name", 1).await.is_err());
241        assert_eq!(queued(&db).await.len(), 1, "only the put is queued");
242    }
243
244    #[tokio::test]
245    async fn sync_off_queues_nothing() {
246        let db = open_db(false).await;
247        assert!(db.kv_outbound.is_none());
248        db.kv_put("cfg", "k1", b"v1").await.expect("put");
249        assert_eq!(db.kv_increment("ctr", "n", 1).await.expect("incr"), 1);
250        assert_eq!(
251            db.kv_get("cfg", "k1").await.expect("get"),
252            Some(b"v1".to_vec())
253        );
254    }
255}
nodedb_lite/nodedb/collection/kv_sync.rs

nodedb_lite/nodedb/collection/
kv_sync.rs
