pagesift_api/jobs/
mirror.rs

1//! 任务记录的复制：多服务器网关把任务存储的每次写入同时交给 [`JobLog`]（账本日志，随日志传送到温备），
2//! 温备用 [`apply_op`] 写进自己的存储，接管后照常从存储恢复任务（结束的能查、进行中的从断点继续）。
3//!
4//! 写入编码成操作：KV 写入/删除、带序号的列表追加、弹出队首、集合加入、清空。按日志顺序重放的结果与原写入
5//! 一致；列表追加按序号去重。账本保证每条记录在温备上只应用一次（已同步的温备跳过快照，
6//! 重新同步的从快照起应用），弹出队首因此不会重复。webhook 待发队列也走这里：重放后即使多留了
7//! 已投过的事件，投递时也按已确认游标跳过，不重发。
8//! 写存储与交给日志在同一把锁里完成，日志顺序就是写入顺序；快照拿同一把锁，得到一致的切面。
9//!
10//! 单主机没有 `JobMirror`，任务存储的写法不变。
11
12use std::sync::{Arc, Mutex};
13
14use pagesift_store::{Store, Tasks};
15
16use super::store::{list_key, meta_key};
17
18/// 复制写入的去处（网关：追加进账本日志）。
19pub trait JobLog: Send + Sync {
20    fn append(&self, op: Vec<u8>);
21}
22
23pub struct JobMirror {
24    log: Arc<dyn JobLog>,
25    lock: Mutex<()>,
26}
27
28// [协议] 本模块自定的操作标签
29const KV_PUT: u8 = 1;
30const KV_DEL: u8 = 2; // [协议] 同上
31const LIST_PUSH: u8 = 3; // [协议] 同上
32const LIST_CLEAR: u8 = 4; // [协议] 同上
33const SET_INSERT: u8 = 5; // [协议] 同上
34const SET_CLEAR: u8 = 6; // [协议] 同上
35const LIST_POP: u8 = 7; // [协议] 同上
36
37/// 任务的列表（与 `store.rs` 的布局一致）。
38const LISTS: &[&str] = &["pages", "errors", "frontier"];
39
40fn put_bytes(out: &mut Vec<u8>, b: &[u8]) {
41    out.extend_from_slice(&(b.len() as u64).to_le_bytes());
42    out.extend_from_slice(b);
43}
44
45fn op(tag: u8, a: &[u8], n: Option<u64>, b: Option<&[u8]>) -> Vec<u8> {
46    let mut out = vec![tag];
47    put_bytes(&mut out, a);
48    if let Some(n) = n {
49        out.extend_from_slice(&n.to_le_bytes());
50    }
51    if let Some(b) = b {
52        put_bytes(&mut out, b);
53    }
54    out
55}
56
57struct Reader<'a>(&'a [u8]);
58
59impl<'a> Reader<'a> {
60    fn u64(&mut self) -> Option<u64> {
61        let (h, rest) = self.0.split_first_chunk::<{ size_of::<u64>() }>()?;
62        self.0 = rest;
63        Some(u64::from_le_bytes(*h))
64    }
65
66    fn bytes(&mut self) -> Option<&'a [u8]> {
67        let n = usize::try_from(self.u64()?).ok()?;
68        let (b, rest) = (self.0.get(..n)?, self.0.get(n..)?);
69        self.0 = rest;
70        Some(b)
71    }
72}
73
74/// 记录写入的任务存储视图：每个写操作先写本地存储，再记下对应的复制操作。
75pub(crate) struct Writer<'a> {
76    t: Tasks<'a>,
77    ops: Option<Vec<Vec<u8>>>,
78}
79
80impl Writer<'_> {
81    pub fn kv_get(&self, key: &str) -> pagesift_store::Result<Option<Vec<u8>>> {
82        self.t.kv_get(key)
83    }
84
85    pub fn kv_put(&mut self, key: &str, value: &[u8]) -> pagesift_store::Result<()> {
86        self.t.kv_put(key, value)?;
87        self.note(|| op(KV_PUT, key.as_bytes(), None, Some(value)));
88        Ok(())
89    }
90
91    pub fn kv_delete(&mut self, key: &str) -> pagesift_store::Result<()> {
92        self.t.kv_delete(key)?;
93        self.note(|| op(KV_DEL, key.as_bytes(), None, None));
94        Ok(())
95    }
96
97    pub fn list_push(&mut self, list: &str, item: &[u8]) -> pagesift_store::Result<()> {
98        let i = self.t.list_push(list, item)?;
99        self.note(|| op(LIST_PUSH, list.as_bytes(), Some(i), Some(item)));
100        Ok(())
101    }
102
103    pub fn list_clear(&mut self, list: &str) -> pagesift_store::Result<()> {
104        self.t.list_clear(list)?;
105        self.note(|| op(LIST_CLEAR, list.as_bytes(), None, None));
106        Ok(())
107    }
108
109    pub fn set_insert(&mut self, set: &str, member: &[u8]) -> pagesift_store::Result<()> {
110        self.t.set_insert(set, member)?;
111        self.note(|| op(SET_INSERT, set.as_bytes(), None, Some(member)));
112        Ok(())
113    }
114
115    pub fn list_pop_front(&mut self, list: &str) -> pagesift_store::Result<()> {
116        self.t.list_pop_front(list)?;
117        self.note(|| op(LIST_POP, list.as_bytes(), None, None));
118        Ok(())
119    }
120
121    pub fn set_clear(&mut self, set: &str) -> pagesift_store::Result<()> {
122        self.t.set_clear(set)?;
123        self.note(|| op(SET_CLEAR, set.as_bytes(), None, None));
124        Ok(())
125    }
126
127    fn note(&mut self, f: impl FnOnce() -> Vec<u8>) {
128        if let Some(ops) = &mut self.ops {
129            ops.push(f());
130        }
131    }
132}
133
134/// 在任务存储上做一组写入；有 `mirror` 时在它的锁里写，并把这组写入按顺序交给日志。
135pub(crate) fn write<R>(
136    store: &Store,
137    mirror: Option<&JobMirror>,
138    f: impl FnOnce(&mut Writer<'_>) -> pagesift_store::Result<R>,
139) -> pagesift_store::Result<R> {
140    let Some(m) = mirror else {
141        return f(&mut Writer {
142            t: store.tasks(),
143            ops: None,
144        });
145    };
146    let _g = m.lock.lock().unwrap_or_else(|e| e.into_inner());
147    let mut w = Writer {
148        t: store.tasks(),
149        ops: Some(Vec::new()),
150    };
151    let r = f(&mut w);
152    // 失败的一组里已经写进本地的部分也照样复制，两边保持一致
153    for o in w.ops.take().unwrap_or_default() {
154        m.log.append(o);
155    }
156    r
157}
158
159impl JobMirror {
160    pub fn new(log: Arc<dyn JobLog>) -> Arc<Self> {
161        Arc::new(Self {
162            log,
163            lock: Mutex::new(()),
164        })
165    }
166
167    /// 存储里全部任务记录，编码成复制操作（快照用）：每个列表、集合先清空再按序号写入，
168    /// 温备据此重建，不受它原有内容的影响。
169    pub fn dump(&self, store: &Store) -> Vec<Vec<u8>> {
170        let _g = self.lock.lock().unwrap_or_else(|e| e.into_inner());
171        let t = store.tasks();
172        let mut out = Vec::new();
173        for key in t.kv_keys(meta_key("")).unwrap_or_default() {
174            let Ok(key) = String::from_utf8(key) else {
175                continue;
176            };
177            let id = &key[meta_key("").len()..];
178            if id.contains('/') {
179                continue;
180            }
181            let Ok(Some(meta)) = t.kv_get(&key) else {
182                continue;
183            };
184            out.push(op(KV_PUT, key.as_bytes(), None, Some(&meta)));
185            for &what in LISTS {
186                let list = list_key(id, what);
187                out.push(op(LIST_CLEAR, list.as_bytes(), None, None));
188                for (i, item) in t
189                    .list_range(&list, 0, usize::MAX)
190                    .unwrap_or_default()
191                    .into_iter()
192                    .enumerate()
193                {
194                    out.push(op(LIST_PUSH, list.as_bytes(), Some(i as u64), Some(&item)));
195                }
196            }
197            let hook = crate::webhook::outbox_key(id);
198            out.push(op(LIST_CLEAR, hook.as_bytes(), None, None));
199            for (i, item) in t
200                .list_range(&hook, 0, usize::MAX)
201                .unwrap_or_default()
202                .into_iter()
203                .enumerate()
204            {
205                out.push(op(LIST_PUSH, hook.as_bytes(), Some(i as u64), Some(&item)));
206            }
207            for k in [crate::webhook::seq_key(id), crate::webhook::ack_key(id)] {
208                match t.kv_get(&k) {
209                    Ok(Some(v)) => out.push(op(KV_PUT, k.as_bytes(), None, Some(&v))),
210                    _ => out.push(op(KV_DEL, k.as_bytes(), None, None)),
211                }
212            }
213            let done = list_key(id, "done");
214            out.push(op(SET_CLEAR, done.as_bytes(), None, None));
215            for m in t.set_members(&done).unwrap_or_default() {
216                out.push(op(SET_INSERT, done.as_bytes(), None, Some(&m)));
217            }
218        }
219        out
220    }
221}
222
223/// 温备：把一条复制操作写进本地存储。列表追加按序号去重（已有这个序号就跳过）。
224pub fn apply_op(store: &Store, raw: &[u8]) -> pagesift_store::Result<()> {
225    let Some((&tag, rest)) = raw.split_first() else {
226        return Ok(());
227    };
228    let mut r = Reader(rest);
229    let Some(a) = r.bytes() else {
230        return Ok(());
231    };
232    let t = store.tasks();
233    match tag {
234        KV_PUT => {
235            if let Some(v) = r.bytes() {
236                t.kv_put(a, v)?;
237            }
238        }
239        KV_DEL => {
240            t.kv_delete(a)?;
241        }
242        LIST_PUSH => {
243            if let (Some(i), Some(v)) = (r.u64(), r.bytes())
244                && t.list_next(a) <= i
245            {
246                t.list_push(a, v)?;
247            }
248        }
249        LIST_CLEAR => {
250            t.list_clear(a)?;
251        }
252        SET_INSERT => {
253            if let Some(v) = r.bytes() {
254                t.set_insert(a, v)?;
255            }
256        }
257        SET_CLEAR => {
258            t.set_clear(a)?;
259        }
260        LIST_POP => {
261            t.list_pop_front(a)?;
262        }
263        _ => {}
264    }
265    Ok(())
266}
267
268#[cfg(test)]
269mod tests {
270    use super::*;
271
272    struct Vec2(Mutex<Vec<Vec<u8>>>);
273
274    impl JobLog for Vec2 {
275        fn append(&self, op: Vec<u8>) {
276            self.0.lock().unwrap().push(op);
277        }
278    }
279
280    /// 网关的写入复制到另一个存储后逐项相同；重放两遍、再叠加一次快照，结果不变。
281    #[test]
282    fn replicated_writes_rebuild_the_same_job_records() {
283        let (a, b) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
284        let (src, dst) = (
285            Store::open_dir(a.path()).unwrap(),
286            Store::open_dir(b.path()).unwrap(),
287        );
288        let log = Arc::new(Vec2(Mutex::default()));
289        let m = JobMirror::new(log.clone());
290        write(&src, Some(&m), |w| {
291            w.kv_put(&meta_key("j1"), b"{\"meta\":1}")?;
292            for i in 0..3u8 {
293                w.list_push(&list_key("j1", "pages"), &[i])?;
294                w.set_insert(&list_key("j1", "done"), &[i])?;
295            }
296            w.list_push(&list_key("j1", "frontier"), b"u")
297        })
298        .unwrap();
299        let ops = log.0.lock().unwrap().clone();
300        for _ in 0..2 {
301            for o in &ops {
302                apply_op(&dst, o).unwrap();
303            }
304        }
305        for o in m.dump(&src) {
306            apply_op(&dst, &o).unwrap();
307        }
308        let (s, d) = (src.tasks(), dst.tasks());
309        assert_eq!(
310            d.kv_get(meta_key("j1")).unwrap(),
311            s.kv_get(meta_key("j1")).unwrap()
312        );
313        for &what in LISTS {
314            let l = list_key("j1", what);
315            assert_eq!(
316                d.list_range(&l, 0, usize::MAX).unwrap(),
317                s.list_range(&l, 0, usize::MAX).unwrap(),
318                "{what}"
319            );
320        }
321        assert_eq!(d.set_len(list_key("j1", "done")), 3);
322    }
323}
pagesift_api/jobs/mirror.rs

pagesift_api/jobs/
mirror.rs
