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

kevy_embedded/
ops_pipeline.rs
