kevy_embedded/
ops_pipeline.rs

1//! Non-atomic batched-fsync write queue: `Store::pipeline`.
2//!
3//! Builder-style queue: enqueue any number of writes via fluent
4//! methods, then `commit()` applies them in queue order. Per-shard
5//! AOF appends are batched into one fsync per shard at commit
6//! time, cutting fsync cost from `N` to `min(N, shard_count)`.
7//!
8//! `Pipeline` is not atomic — each op acquires its own per-shard
9//! write lock as it is applied, so other writers see intermediate
10//! states. For transactional semantics use
11//! [`Store::atomic`](crate::Store::atomic).
12
13use crate::KevyResult;
14
15use crate::store::Store;
16
17/// Builder-style write queue. Returned by [`Store::pipeline`]; call
18/// fluent methods to enqueue + `commit()` to apply with batched
19/// AOF fsync.
20///
21/// ```
22/// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
23/// let queued = s.pipeline().set(b"a", b"1").incr(b"hits").incr(b"hits");
24/// assert_eq!(queued.len(), 3);
25/// assert_eq!(s.get(b"hits")?, None, "nothing applies before commit");
26/// queued.commit()?;
27/// assert_eq!(s.get(b"hits")?.as_deref(), Some(&b"2"[..]));
28/// # Ok::<(), kevy_embedded::KevyError>(())
29/// ```
30#[derive(Debug)]
31pub struct Pipeline<'a> {
32    store: &'a Store,
33    ops: Vec<PendingOp>,
34}
35
36#[derive(Debug)]
37enum PendingOp {
38    Set { key: Vec<u8>, value: Vec<u8> },
39    Del { keys: Vec<Vec<u8>> },
40    Incr { key: Vec<u8> },
41    IncrBy { key: Vec<u8>, delta: i64 },
42    HSet { key: Vec<u8>, pairs: Vec<(Vec<u8>, Vec<u8>)> },
43    HDel { key: Vec<u8>, fields: Vec<Vec<u8>> },
44    HIncrBy { key: Vec<u8>, field: Vec<u8>, delta: i64 },
45    ZAdd { key: Vec<u8>, pairs: Vec<(f64, Vec<u8>)> },
46    ZAddFlags { key: Vec<u8>, pairs: Vec<(f64, Vec<u8>)>, flags: kevy_store::ZaddFlags },
47    ZRem { key: Vec<u8>, members: Vec<Vec<u8>> },
48    ZIncrBy { key: Vec<u8>, delta: f64, member: Vec<u8> },
49    SAdd { key: Vec<u8>, members: Vec<Vec<u8>> },
50    SRem { key: Vec<u8>, members: Vec<Vec<u8>> },
51    LPush { key: Vec<u8>, values: Vec<Vec<u8>> },
52    RPush { key: Vec<u8>, values: Vec<Vec<u8>> },
53}
54
55impl<'a> Pipeline<'a> {
56    pub(crate) fn new(store: &'a Store) -> Self {
57        Self { store, ops: Vec::new() }
58    }
59
60    /// Number of ops queued so far.
61    pub fn len(&self) -> usize {
62        self.ops.len()
63    }
64
65    /// `true` when no ops are queued.
66    pub fn is_empty(&self) -> bool {
67        self.ops.is_empty()
68    }
69
70    // ---- fluent enqueue --------------------------------------------
71
72    /// Queue `SET key value`. Nothing is applied (or AOF-logged)
73    /// until [`commit`](Self::commit) — true of every enqueue method.
74    pub fn set(mut self, key: &[u8], value: &[u8]) -> Self {
75        self.ops.push(PendingOp::Set { key: key.to_vec(), value: value.to_vec() });
76        self
77    }
78
79    /// Queue `DEL key [key ...]`.
80    pub fn del(mut self, keys: &[&[u8]]) -> Self {
81        self.ops.push(PendingOp::Del { keys: keys.iter().map(|k| k.to_vec()).collect() });
82        self
83    }
84
85    /// Queue `INCR key` (commit errors if the value is not an integer).
86    pub fn incr(mut self, key: &[u8]) -> Self {
87        self.ops.push(PendingOp::Incr { key: key.to_vec() });
88        self
89    }
90
91    /// Queue `INCRBY key delta`; negative `delta` does DECR-style work.
92    pub fn incr_by(mut self, key: &[u8], delta: i64) -> Self {
93        self.ops.push(PendingOp::IncrBy { key: key.to_vec(), delta });
94        self
95    }
96
97    /// Queue `HSET key field value [field value ...]`.
98    pub fn hset(mut self, key: &[u8], pairs: &[(&[u8], &[u8])]) -> Self {
99        self.ops.push(PendingOp::HSet {
100            key: key.to_vec(),
101            pairs: pairs.iter().map(|(f, v)| (f.to_vec(), v.to_vec())).collect(),
102        });
103        self
104    }
105
106    /// Queue `HDEL key field [field ...]`.
107    pub fn hdel(mut self, key: &[u8], fields: &[&[u8]]) -> Self {
108        self.ops.push(PendingOp::HDel {
109            key: key.to_vec(),
110            fields: fields.iter().map(|f| f.to_vec()).collect(),
111        });
112        self
113    }
114
115    /// Queue `HINCRBY key field delta`.
116    pub fn hincrby(mut self, key: &[u8], field: &[u8], delta: i64) -> Self {
117        self.ops.push(PendingOp::HIncrBy { key: key.to_vec(), field: field.to_vec(), delta });
118        self
119    }
120
121    /// Queue `ZADD key score member [score member ...]`.
122    pub fn zadd(mut self, key: &[u8], pairs: &[(f64, &[u8])]) -> Self {
123        self.ops.push(PendingOp::ZAdd {
124            key: key.to_vec(),
125            pairs: pairs.iter().map(|(s, m)| (*s, m.to_vec())).collect(),
126        });
127        self
128    }
129
130    /// Flags-aware `ZADD` — e.g. the `GT` monotonic-heal form.
131    pub fn zadd_flags(
132        mut self,
133        key: &[u8],
134        pairs: &[(f64, &[u8])],
135        flags: kevy_store::ZaddFlags,
136    ) -> Self {
137        self.ops.push(PendingOp::ZAddFlags {
138            key: key.to_vec(),
139            pairs: pairs.iter().map(|(s, m)| (*s, m.to_vec())).collect(),
140            flags,
141        });
142        self
143    }
144
145    /// Queue `ZREM key member [member ...]`.
146    pub fn zrem(mut self, key: &[u8], members: &[&[u8]]) -> Self {
147        self.ops.push(PendingOp::ZRem {
148            key: key.to_vec(),
149            members: members.iter().map(|m| m.to_vec()).collect(),
150        });
151        self
152    }
153
154    /// Queue `ZINCRBY key delta member`.
155    pub fn zincrby(mut self, key: &[u8], delta: f64, member: &[u8]) -> Self {
156        self.ops.push(PendingOp::ZIncrBy { key: key.to_vec(), delta, member: member.to_vec() });
157        self
158    }
159
160    /// Queue `SADD key member [member ...]`.
161    pub fn sadd(mut self, key: &[u8], members: &[&[u8]]) -> Self {
162        self.ops.push(PendingOp::SAdd {
163            key: key.to_vec(),
164            members: members.iter().map(|m| m.to_vec()).collect(),
165        });
166        self
167    }
168
169    /// Queue `SREM key member [member ...]`.
170    pub fn srem(mut self, key: &[u8], members: &[&[u8]]) -> Self {
171        self.ops.push(PendingOp::SRem {
172            key: key.to_vec(),
173            members: members.iter().map(|m| m.to_vec()).collect(),
174        });
175        self
176    }
177
178    /// Queue `LPUSH key value [value ...]`.
179    pub fn lpush(mut self, key: &[u8], values: &[&[u8]]) -> Self {
180        self.ops.push(PendingOp::LPush {
181            key: key.to_vec(),
182            values: values.iter().map(|v| v.to_vec()).collect(),
183        });
184        self
185    }
186
187    /// Queue `RPUSH key value [value ...]`.
188    pub fn rpush(mut self, key: &[u8], values: &[&[u8]]) -> Self {
189        self.ops.push(PendingOp::RPush {
190            key: key.to_vec(),
191            values: values.iter().map(|v| v.to_vec()).collect(),
192        });
193        self
194    }
195
196    /// Apply every queued op in order. Each op acquires its own
197    /// per-shard write lock — other writers see intermediate states
198    /// between ops; for transactional semantics use [`Store::atomic`]
199    /// instead. AOF appends batch into one fsync per shard.
200    pub fn commit(mut self) -> KevyResult<()> {
201        let ops = std::mem::take(&mut self.ops);
202        for op in ops {
203            self.apply_one(op)?;
204        }
205        Ok(())
206    }
207
208    // fn-length exemption: pure data-driven op match table — one flat
209    // arm per PendingOp variant, only arg plumbing + one store call.
210    // LOC-WAIVER: data-driven op dispatch table — one store-call arm per PendingOp variant.
211    fn apply_one(&self, op: PendingOp) -> KevyResult<()> {
212        match op {
213            PendingOp::Set { key, value } => {
214                self.store.set(&key, &value)?;
215            }
216            PendingOp::Del { keys } => {
217                let refs: Vec<&[u8]> = keys.iter().map(|k| k.as_slice()).collect();
218                self.store.del(&refs)?;
219            }
220            PendingOp::Incr { key } => {
221                self.store.incr(&key)?;
222            }
223            PendingOp::IncrBy { key, delta } => {
224                self.store.incr_by(&key, delta)?;
225            }
226            PendingOp::HSet { key, pairs } => {
227                let refs: Vec<(&[u8], &[u8])> =
228                    pairs.iter().map(|(f, v)| (f.as_slice(), v.as_slice())).collect();
229                self.store.hset(&key, &refs)?;
230            }
231            PendingOp::HDel { key, fields } => {
232                let refs: Vec<&[u8]> = fields.iter().map(|f| f.as_slice()).collect();
233                self.store.hdel(&key, &refs)?;
234            }
235            PendingOp::HIncrBy { key, field, delta } => {
236                self.store.hincrby(&key, &field, delta)?;
237            }
238            PendingOp::ZAdd { key, pairs } => {
239                let refs: Vec<(f64, &[u8])> =
240                    pairs.iter().map(|(s, m)| (*s, m.as_slice())).collect();
241                self.store.zadd(&key, &refs)?;
242            }
243            PendingOp::ZAddFlags { key, pairs, flags } => {
244                let refs: Vec<(f64, &[u8])> =
245                    pairs.iter().map(|(s, m)| (*s, m.as_slice())).collect();
246                self.store.zadd_flags(&key, &refs, flags)?;
247            }
248            PendingOp::ZRem { key, members } => {
249                let refs: Vec<&[u8]> = members.iter().map(|m| m.as_slice()).collect();
250                self.store.zrem(&key, &refs)?;
251            }
252            PendingOp::ZIncrBy { key, delta, member } => {
253                self.store.zincrby(&key, delta, &member)?;
254            }
255            PendingOp::SAdd { key, members } => {
256                let refs: Vec<&[u8]> = members.iter().map(|m| m.as_slice()).collect();
257                self.store.sadd(&key, &refs)?;
258            }
259            PendingOp::SRem { key, members } => {
260                let refs: Vec<&[u8]> = members.iter().map(|m| m.as_slice()).collect();
261                self.store.srem(&key, &refs)?;
262            }
263            PendingOp::LPush { key, values } => {
264                let refs: Vec<&[u8]> = values.iter().map(|v| v.as_slice()).collect();
265                self.store.lpush(&key, &refs)?;
266            }
267            PendingOp::RPush { key, values } => {
268                let refs: Vec<&[u8]> = values.iter().map(|v| v.as_slice()).collect();
269                self.store.rpush(&key, &refs)?;
270            }
271        }
272        Ok(())
273    }
274}
275
276impl Store {
277    /// Begin a [`Pipeline`] — fluent write queue. Add ops via
278    /// `.set(...).hset(...).zadd(...)` then call `.commit()`.
279    pub fn pipeline(&self) -> Pipeline<'_> {
280        Pipeline::new(self)
281    }
282}
283
284/// Parity manifest: command names `Pipeline` implements.
285/// Cross-checked against `kevy_resp::ops_table` in
286/// `store_tests_op_table.rs` — update BOTH when adding an op.
287#[cfg_attr(not(test), allow(dead_code))]
288pub(crate) const PIPELINE_OPS: &[&str] = &[
289    "SET", "DEL", "INCR", "INCRBY", "HSET", "HDEL", "HINCRBY", "ZADD", "ZREM", "ZINCRBY", "SADD",
290    "SREM", "LPUSH", "RPUSH",
291];
kevy_embedded/ops_pipeline.rs

kevy_embedded/
ops_pipeline.rs
