pagesift_api/jobs/
mirror.rs

1//! 任务记录的复制：多服务器网关把任务存储的每次写入同时交给 [`JobLog`]（账本日志，随日志传送到温备），
2//! 温备用 [`apply_op`] 写进自己的存储，接管后照常从存储恢复任务（结束的能查、进行中的从断点继续）。
3//!
4//! 写入编码成操作：KV 写入/删除、带序号的列表追加、弹出队首、集合加入、清空。按日志顺序重放的结果与原写入
5//! 一致；列表追加按序号去重。账本保证每条记录在温备上只应用一次（已同步的温备跳过快照，
6//! 重新同步的从快照起应用），弹出队首因此不会重复。webhook 待发队列也走这里：重放后即使多留了
7//! 已投过的事件，投递时也按已确认游标跳过，不重发。
8//! 写存储与交给日志在同一把锁里完成，日志顺序就是写入顺序；快照拿同一把锁，得到一致的切面。
9//!
10//! 交给日志失败（账本日志写不进去）时 [`write`] 返回错误：本地已经写进去的部分不撤回，但调用方知道
11//! 这组写入没有复制到温备（任务因此失败，见 `jobs::sink`），不会像 0.90 那样只在日志里记一笔警告。
12//! 镜像同时记下失败次数（[`JobMirror::failures`]）。
13//!
14//! 单主机没有 `JobMirror`，任务存储的写法不变。
15
16use std::sync::{Arc, Mutex};
17
18use pagesift_store::{Store, Tasks};
19
20use super::store::{list_key, meta_key};
21
22/// 复制写入的去处（网关：追加进账本日志）。失败时返回原因。
23pub trait JobLog: Send + Sync {
24    fn append(&self, op: Vec<u8>) -> Result<(), String>;
25    /// 一条操作的编码字节上限（超过的写入不复制，例如很大的附件）。默认不限。
26    fn max_op(&self) -> usize {
27        usize::MAX
28    }
29}
30
31pub struct JobMirror {
32    log: Arc<dyn JobLog>,
33    lock: Mutex<()>,
34    failures: std::sync::atomic::AtomicU64,
35}
36
37// [协议] 本模块自定的操作标签
38const KV_PUT: u8 = 1;
39const KV_DEL: u8 = 2; // [协议] 同上
40const LIST_PUSH: u8 = 3; // [协议] 同上
41const LIST_CLEAR: u8 = 4; // [协议] 同上
42const SET_INSERT: u8 = 5; // [协议] 同上
43const SET_CLEAR: u8 = 6; // [协议] 同上
44const LIST_POP: u8 = 7; // [协议] 同上
45
46/// 任务的列表（与 `store.rs` 的布局一致）。
47const LISTS: &[&str] = &["pages", "errors", "frontier", "notes"];
48
49fn put_bytes(out: &mut Vec<u8>, b: &[u8]) {
50    out.extend_from_slice(&(b.len() as u64).to_le_bytes());
51    out.extend_from_slice(b);
52}
53
54fn op(tag: u8, a: &[u8], n: Option<u64>, b: Option<&[u8]>) -> Vec<u8> {
55    let mut out = vec![tag];
56    put_bytes(&mut out, a);
57    if let Some(n) = n {
58        out.extend_from_slice(&n.to_le_bytes());
59    }
60    if let Some(b) = b {
61        put_bytes(&mut out, b);
62    }
63    out
64}
65
66struct Reader<'a>(&'a [u8]);
67
68impl<'a> Reader<'a> {
69    fn u64(&mut self) -> Option<u64> {
70        let (h, rest) = self.0.split_first_chunk::<{ size_of::<u64>() }>()?;
71        self.0 = rest;
72        Some(u64::from_le_bytes(*h))
73    }
74
75    fn bytes(&mut self) -> Option<&'a [u8]> {
76        let n = usize::try_from(self.u64()?).ok()?;
77        let (b, rest) = (self.0.get(..n)?, self.0.get(n..)?);
78        self.0 = rest;
79        Some(b)
80    }
81}
82
83/// 记录写入的任务存储视图：每个写操作先写本地存储，再记下对应的复制操作。
84pub(crate) struct Writer<'a> {
85    t: Tasks<'a>,
86    ops: Option<Vec<Vec<u8>>>,
87}
88
89impl Writer<'_> {
90    pub fn kv_get(&self, key: &str) -> pagesift_store::Result<Option<Vec<u8>>> {
91        self.t.kv_get(key)
92    }
93
94    pub fn kv_put(&mut self, key: &str, value: &[u8]) -> pagesift_store::Result<()> {
95        self.t.kv_put(key, value)?;
96        self.note(|| op(KV_PUT, key.as_bytes(), None, Some(value)));
97        Ok(())
98    }
99
100    pub fn kv_delete(&mut self, key: &str) -> pagesift_store::Result<()> {
101        self.t.kv_delete(key)?;
102        self.note(|| op(KV_DEL, key.as_bytes(), None, None));
103        Ok(())
104    }
105
106    pub fn list_push(&mut self, list: &str, item: &[u8]) -> pagesift_store::Result<()> {
107        let i = self.t.list_push(list, item)?;
108        self.note(|| op(LIST_PUSH, list.as_bytes(), Some(i), Some(item)));
109        Ok(())
110    }
111
112    pub fn list_clear(&mut self, list: &str) -> pagesift_store::Result<()> {
113        self.t.list_clear(list)?;
114        self.note(|| op(LIST_CLEAR, list.as_bytes(), None, None));
115        Ok(())
116    }
117
118    pub fn set_insert(&mut self, set: &str, member: &[u8]) -> pagesift_store::Result<()> {
119        self.t.set_insert(set, member)?;
120        self.note(|| op(SET_INSERT, set.as_bytes(), None, Some(member)));
121        Ok(())
122    }
123
124    pub fn list_pop_front(&mut self, list: &str) -> pagesift_store::Result<()> {
125        self.t.list_pop_front(list)?;
126        self.note(|| op(LIST_POP, list.as_bytes(), None, None));
127        Ok(())
128    }
129
130    pub fn set_clear(&mut self, set: &str) -> pagesift_store::Result<()> {
131        self.t.set_clear(set)?;
132        self.note(|| op(SET_CLEAR, set.as_bytes(), None, None));
133        Ok(())
134    }
135
136    fn note(&mut self, f: impl FnOnce() -> Vec<u8>) {
137        if let Some(ops) = &mut self.ops {
138            ops.push(f());
139        }
140    }
141}
142
143/// 在任务存储上做一组写入；有 `mirror` 时在它的锁里写，并把这组写入按顺序交给日志。
144pub(crate) fn write<R>(
145    store: &Store,
146    mirror: Option<&JobMirror>,
147    f: impl FnOnce(&mut Writer<'_>) -> pagesift_store::Result<R>,
148) -> pagesift_store::Result<R> {
149    let Some(m) = mirror else {
150        return f(&mut Writer {
151            t: store.tasks(),
152            ops: None,
153        });
154    };
155    let _g = m.lock.lock().unwrap_or_else(|e| e.into_inner());
156    let mut w = Writer {
157        t: store.tasks(),
158        ops: Some(Vec::new()),
159    };
160    let r = f(&mut w);
161    // 失败的一组里已经写进本地的部分也照样复制，两边保持一致；一条交不出去就不再交后面的
162    // （温备按序号追加列表，跳过一条后后面的会错位）
163    let mut failed = None;
164    for o in w.ops.take().unwrap_or_default() {
165        if let Err(e) = m.log.append(o) {
166            m.failures
167                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
168            failed = Some(e);
169            break;
170        }
171    }
172    match (r, failed) {
173        (Ok(_), Some(e)) => Err(pagesift_store::Error::Io(std::io::Error::other(
174            Unreplicated(e),
175        ))),
176        (r, _) => r,
177    }
178}
179
180/// 本地已经写进去、只是没能交给复制日志：[`write`] 的这种错误里装的是它（[`unreplicated`] 判断）。
181#[derive(Debug)]
182struct Unreplicated(String);
183
184impl std::fmt::Display for Unreplicated {
185    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        write!(f, "replicating to the standby failed: {}", self.0)
187    }
188}
189
190impl std::error::Error for Unreplicated {}
191
192/// `e` 是不是"本地写成了、复制失败"（见 [`write`]）。
193pub(crate) fn unreplicated(e: &pagesift_store::Error) -> bool {
194    matches!(e, pagesift_store::Error::Io(io) if io.get_ref().is_some_and(|x| x.is::<Unreplicated>()))
195}
196
197impl JobMirror {
198    pub fn new(log: Arc<dyn JobLog>) -> Arc<Self> {
199        Arc::new(Self {
200            log,
201            lock: Mutex::new(()),
202            failures: std::sync::atomic::AtomicU64::new(0),
203        })
204    }
205
206    /// 一条复制操作的字节上限（见 [`JobLog::max_op`]）。
207    pub fn max_op(&self) -> usize {
208        self.log.max_op()
209    }
210
211    /// 交给日志失败的写入组数（自进程启动）。
212    pub fn failures(&self) -> u64 {
213        self.failures.load(std::sync::atomic::Ordering::Relaxed)
214    }
215
216    /// 存储里全部任务记录，编码成复制操作（快照用）：每个列表、集合先清空再按序号写入，
217    /// 温备据此重建，不受它原有内容的影响。
218    pub fn dump(&self, store: &Store) -> Vec<Vec<u8>> {
219        let _g = self.lock.lock().unwrap_or_else(|e| e.into_inner());
220        let t = store.tasks();
221        let mut out = Vec::new();
222        for key in t.kv_keys(meta_key("")).unwrap_or_default() {
223            let Ok(key) = String::from_utf8(key) else {
224                continue;
225            };
226            let id = &key[meta_key("").len()..];
227            if id.contains('/') {
228                continue;
229            }
230            let Ok(Some(meta)) = t.kv_get(&key) else {
231                continue;
232            };
233            out.push(op(KV_PUT, key.as_bytes(), None, Some(&meta)));
234            for &what in LISTS {
235                let list = list_key(id, what);
236                out.push(op(LIST_CLEAR, list.as_bytes(), None, None));
237                for (i, item) in t
238                    .list_range(&list, 0, usize::MAX)
239                    .unwrap_or_default()
240                    .into_iter()
241                    .enumerate()
242                {
243                    out.push(op(LIST_PUSH, list.as_bytes(), Some(i as u64), Some(&item)));
244                }
245            }
246            let hook = crate::webhook::outbox_key(id);
247            out.push(op(LIST_CLEAR, hook.as_bytes(), None, None));
248            for (i, item) in t
249                .list_range(&hook, 0, usize::MAX)
250                .unwrap_or_default()
251                .into_iter()
252                .enumerate()
253            {
254                out.push(op(LIST_PUSH, hook.as_bytes(), Some(i as u64), Some(&item)));
255            }
256            for k in [crate::webhook::seq_key(id), crate::webhook::ack_key(id)] {
257                match t.kv_get(&k) {
258                    Ok(Some(v)) => out.push(op(KV_PUT, k.as_bytes(), None, Some(&v))),
259                    _ => out.push(op(KV_DEL, k.as_bytes(), None, None)),
260                }
261            }
262            let done = list_key(id, "done");
263            out.push(op(SET_CLEAR, done.as_bytes(), None, None));
264            for m in t.set_members(&done).unwrap_or_default() {
265                out.push(op(SET_INSERT, done.as_bytes(), None, Some(&m)));
266            }
267        }
268        // 附件（内容和到期时刻）：先内容后到期时刻，与在线写入的顺序相同；超过单条上限的不复制
269        for (data, exp) in crate::attachments::stored(&t) {
270            if data.1.len() <= self.max_op() {
271                out.push(op(KV_PUT, data.0.as_bytes(), None, Some(&data.1)));
272                out.push(op(KV_PUT, exp.0.as_bytes(), None, Some(&exp.1)));
273            }
274        }
275        out
276    }
277}
278
279/// 温备：把一条复制操作写进本地存储。列表追加按序号去重（已有这个序号就跳过）。
280pub fn apply_op(store: &Store, raw: &[u8]) -> pagesift_store::Result<()> {
281    let Some((&tag, rest)) = raw.split_first() else {
282        return Ok(());
283    };
284    let mut r = Reader(rest);
285    let Some(a) = r.bytes() else {
286        return Ok(());
287    };
288    let t = store.tasks();
289    match tag {
290        KV_PUT => {
291            if let Some(v) = r.bytes() {
292                t.kv_put(a, v)?;
293            }
294        }
295        KV_DEL => {
296            t.kv_delete(a)?;
297        }
298        LIST_PUSH => {
299            if let (Some(i), Some(v)) = (r.u64(), r.bytes())
300                && t.list_next(a) <= i
301            {
302                t.list_push(a, v)?;
303            }
304        }
305        LIST_CLEAR => {
306            t.list_clear(a)?;
307        }
308        SET_INSERT => {
309            if let Some(v) = r.bytes() {
310                t.set_insert(a, v)?;
311            }
312        }
313        SET_CLEAR => {
314            t.set_clear(a)?;
315        }
316        LIST_POP => {
317            t.list_pop_front(a)?;
318        }
319        _ => {}
320    }
321    Ok(())
322}
323
324#[cfg(test)]
325mod tests {
326    use super::*;
327
328    struct Vec2(Mutex<Vec<Vec<u8>>>);
329
330    impl JobLog for Vec2 {
331        fn append(&self, op: Vec<u8>) -> Result<(), String> {
332            self.0.lock().unwrap().push(op);
333            Ok(())
334        }
335    }
336
337    /// 日志写不进去（第二条起）：写入报错、计数加一，不再交后面的操作。
338    struct Broken(Mutex<usize>);
339
340    impl JobLog for Broken {
341        fn append(&self, _op: Vec<u8>) -> Result<(), String> {
342            let mut n = self.0.lock().unwrap();
343            *n += 1;
344            if *n > 1 {
345                Err("journal full".into())
346            } else {
347                Ok(())
348            }
349        }
350    }
351
352    #[test]
353    fn a_failed_append_is_reported_to_the_writer() {
354        let a = tempfile::tempdir().unwrap();
355        let src = Store::open_dir(a.path()).unwrap();
356        let log = Arc::new(Broken(Mutex::new(0)));
357        let m = JobMirror::new(log.clone());
358        let r = write(&src, Some(&m), |w| {
359            w.list_push(&list_key("j1", "pages"), b"a")?;
360            w.list_push(&list_key("j1", "pages"), b"b")?;
361            w.set_insert(&list_key("j1", "done"), b"a")
362        });
363        let e = r.unwrap_err();
364        assert!(unreplicated(&e));
365        let e = e.to_string();
366        assert!(e.contains("journal full"), "{e}");
367        assert_eq!(m.failures(), 1);
368        assert_eq!(*log.0.lock().unwrap(), 2);
369    }
370
371    /// 网关的写入复制到另一个存储后逐项相同；重放两遍、再叠加一次快照，结果不变。
372    #[test]
373    fn replicated_writes_rebuild_the_same_job_records() {
374        let (a, b) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
375        let (src, dst) = (
376            Store::open_dir(a.path()).unwrap(),
377            Store::open_dir(b.path()).unwrap(),
378        );
379        let log = Arc::new(Vec2(Mutex::default()));
380        let m = JobMirror::new(log.clone());
381        write(&src, Some(&m), |w| {
382            w.kv_put(&meta_key("j1"), b"{\"meta\":1}")?;
383            for i in 0..3u8 {
384                w.list_push(&list_key("j1", "pages"), &[i])?;
385                w.set_insert(&list_key("j1", "done"), &[i])?;
386            }
387            w.list_push(&list_key("j1", "frontier"), b"u")
388        })
389        .unwrap();
390        let ops = log.0.lock().unwrap().clone();
391        for _ in 0..2 {
392            for o in &ops {
393                apply_op(&dst, o).unwrap();
394            }
395        }
396        for o in m.dump(&src) {
397            apply_op(&dst, &o).unwrap();
398        }
399        let (s, d) = (src.tasks(), dst.tasks());
400        assert_eq!(
401            d.kv_get(meta_key("j1")).unwrap(),
402            s.kv_get(meta_key("j1")).unwrap()
403        );
404        for &what in LISTS {
405            let l = list_key("j1", what);
406            assert_eq!(
407                d.list_range(&l, 0, usize::MAX).unwrap(),
408                s.list_range(&l, 0, usize::MAX).unwrap(),
409                "{what}"
410            );
411        }
412        assert_eq!(d.set_len(list_key("j1", "done")), 3);
413    }
414}
pagesift_api/jobs/mirror.rs

pagesift_api/jobs/
mirror.rs
