1use std::fs::{File, OpenOptions};
53use std::io::{BufRead, BufReader, Write};
54use std::path::{Path, PathBuf};
55
56use serde::{Deserialize, Serialize};
57
58use crate::delegation::Delegation;
59use crate::error::CoreError;
60use crate::gate::DenyReason;
61use crate::intent::SpendIntent;
62use crate::pending::PendingOutcome;
63
64#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
66#[serde(rename_all = "snake_case")]
67pub enum WalDecision {
68 Allow,
69 Deny,
70}
71
72#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
74#[serde(tag = "kind", rename_all = "snake_case")]
75pub enum WalRecord {
76 RegisterDelegation { ts: u64, delegation: Delegation },
78 Revoke { ts: u64, delegation_id: String },
80 Decide {
82 ts: u64,
83 decision: WalDecision,
84 delegation_id: String,
85 intent: SpendIntent,
86 #[serde(skip_serializing_if = "Option::is_none")]
88 reason: Option<DenyReason>,
89 budget_after_cents: u64,
91 },
92 Pending {
94 ts: u64,
95 pending_id: String,
96 delegation_id: String,
97 intent: SpendIntent,
99 approved_amount_cents: u64,
101 expires_ts: u64,
103 },
104 Confirm {
106 ts: u64,
107 pending_id: String,
108 amount_cents: u64,
110 proof: String,
112 },
113 Terminal {
115 ts: u64,
116 pending_id: String,
117 outcome: PendingOutcome,
118 },
119}
120
121impl WalRecord {
122 pub fn ts(&self) -> u64 {
124 match self {
125 WalRecord::RegisterDelegation { ts, .. }
126 | WalRecord::Revoke { ts, .. }
127 | WalRecord::Decide { ts, .. }
128 | WalRecord::Pending { ts, .. }
129 | WalRecord::Confirm { ts, .. }
130 | WalRecord::Terminal { ts, .. } => *ts,
131 }
132 }
133
134 pub fn kind(&self) -> &'static str {
136 match self {
137 WalRecord::RegisterDelegation { .. } => "register_delegation",
138 WalRecord::Revoke { .. } => "revoke",
139 WalRecord::Decide { .. } => "decide",
140 WalRecord::Pending { .. } => "pending",
141 WalRecord::Confirm { .. } => "confirm",
142 WalRecord::Terminal { .. } => "terminal",
143 }
144 }
145}
146
147#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
152pub struct WalLine {
153 pub seq: u64,
155 pub prev: u64,
157 pub rec: WalRecord,
159}
160
161pub(crate) fn chain_value(prev: u64, seq: u64, rec_json: &str) -> u64 {
167 let mut bytes = Vec::with_capacity(16 + rec_json.len());
168 bytes.extend_from_slice(&prev.to_le_bytes());
169 bytes.extend_from_slice(&seq.to_le_bytes());
170 bytes.extend_from_slice(rec_json.as_bytes());
171 fnv1a_64(&bytes)
172}
173
174pub(crate) fn fnv1a_64(bytes: &[u8]) -> u64 {
176 const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
177 const PRIME: u64 = 0x0000_0100_0000_01b3;
178 let mut hash = OFFSET_BASIS;
179 for byte in bytes {
180 hash ^= u64::from(*byte);
181 hash = hash.wrapping_mul(PRIME);
182 }
183 hash
184}
185
186#[derive(Debug)]
189pub struct Wal {
190 file: File,
191 path: PathBuf,
192 lines: u64,
193 chain: u64,
195 _lock: WalLock,
197}
198
199pub fn single_writer_lock_path(wal_path: impl AsRef<Path>) -> PathBuf {
204 let wal_path = wal_path.as_ref();
205 let mut name = wal_path
206 .file_name()
207 .map(|n| n.to_os_string())
208 .unwrap_or_default();
209 name.push(".lock");
210 wal_path.with_file_name(name)
211}
212
213#[derive(Debug)]
235pub struct WalLock {
236 path: PathBuf,
237}
238
239impl WalLock {
240 pub fn acquire(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
242 let wal_path = wal_path.as_ref();
243 let lock_path = single_writer_lock_path(wal_path);
244 match OpenOptions::new()
245 .write(true)
246 .create_new(true)
247 .open(&lock_path)
248 {
249 Ok(mut file) => {
250 let written = writeln!(file, "pid={}", std::process::id())
251 .and_then(|()| writeln!(file, "wal={}", wal_path.display()))
252 .and_then(|()| file.flush());
253 if let Err(e) = written {
254 let _ = std::fs::remove_file(&lock_path);
256 return Err(CoreError::WalIo(format!(
257 "写单写者锁 {lock_path:?} 失败(fail-closed): {e}"
258 )));
259 }
260 Ok(Self { path: lock_path })
261 }
262 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
263 let holder = std::fs::read_to_string(&lock_path).unwrap_or_default();
264 let holder = holder.trim();
265 let holder_note = if holder.is_empty() {
266 "锁文件为空(持锁方刚创建,极可能是并发启动竞争)".to_string()
267 } else {
268 format!("持锁信息: {holder}")
269 };
270 Err(CoreError::WalLocked {
271 path: lock_path.display().to_string(),
272 message: format!(
273 "同一份审计日志已有另一个 Wanning 进程在写({holder_note});\
274 确认没有别的闸在跑后,删除该锁文件即可恢复\
275 (默认 WAL 在 target/ 下,cargo clean 亦可)"
276 ),
277 })
278 }
279 Err(e) => Err(CoreError::WalIo(format!(
280 "创建单写者锁 {lock_path:?} 失败: {e}"
281 ))),
282 }
283 }
284}
285
286impl Drop for WalLock {
287 fn drop(&mut self) {
288 let _ = std::fs::remove_file(&self.path);
290 }
291}
292
293impl Wal {
294 pub fn open(path: impl AsRef<Path>) -> Result<Self, CoreError> {
302 let path = path.as_ref().to_path_buf();
303 crate::paths::ensure_wal_parent(&path)?;
304 let _lock = WalLock::acquire(&path)?;
305 let (existing_lines, chain) = if path.exists() {
307 let verified = read_verified(&path)?;
308 (verified.records.len() as u64, verified.tail)
309 } else {
310 (0, 0)
311 };
312 let file = OpenOptions::new()
313 .create(true)
314 .append(true)
315 .read(false)
316 .open(&path)
317 .map_err(|e| CoreError::WalIo(format!("打开 WAL {path:?} 失败: {e}")))?;
318 Ok(Self {
319 file,
320 path,
321 lines: existing_lines,
322 chain,
323 _lock,
324 })
325 }
326
327 pub fn path(&self) -> &Path {
328 &self.path
329 }
330
331 pub fn line_count(&self) -> u64 {
333 self.lines
334 }
335
336 pub fn chain_tail(&self) -> u64 {
338 self.chain
339 }
340
341 pub fn append(&mut self, record: &WalRecord) -> Result<u64, CoreError> {
346 let seq = self.lines + 1;
347 let rec_json = serde_json::to_string(record)
348 .map_err(|e| CoreError::WalIo(format!("WAL 记录序列化失败: {e}")))?;
349 let mut line = serde_json::to_string(&WalLine {
350 seq,
351 prev: self.chain,
352 rec: record.clone(),
353 })
354 .map_err(|e| CoreError::WalIo(format!("WAL 记录序列化失败: {e}")))?;
355 line.push('\n');
356 let path = self.path.clone();
357 self.file
358 .write_all(line.as_bytes())
359 .and_then(|()| self.file.flush())
360 .map_err(|e| CoreError::WalIo(format!("写 WAL {path:?} 失败: {e}")))?;
361 self.lines = seq;
362 self.chain = chain_value(self.chain, seq, &rec_json);
363 Ok(self.lines)
364 }
365}
366
367pub fn raw_lines(path: impl AsRef<Path>) -> Result<Vec<String>, CoreError> {
369 let path = path.as_ref();
370 let file =
371 File::open(path).map_err(|e| CoreError::WalIo(format!("读 WAL {path:?} 失败: {e}")))?;
372 BufReader::new(file)
373 .lines()
374 .collect::<Result<Vec<_>, _>>()
375 .map_err(|e| CoreError::WalIo(format!("读 WAL {path:?} 失败: {e}")))
376}
377
378#[derive(Clone, Debug, PartialEq, Eq)]
380pub struct VerifiedLog {
381 pub records: Vec<(u64, WalRecord)>,
383 pub links: Vec<WalChainLink>,
387 pub tail: u64,
390}
391
392#[derive(Clone, Copy, Debug, PartialEq, Eq)]
394pub struct WalChainLink {
395 pub seq: u64,
397 pub prev: u64,
399 pub value: u64,
401}
402
403pub fn read_verified(path: impl AsRef<Path>) -> Result<VerifiedLog, CoreError> {
414 let mut records = Vec::new();
415 let mut links = Vec::new();
416 let mut chain = 0u64;
417 for (idx, line) in raw_lines(path)?.into_iter().enumerate() {
418 let line_no = idx as u64 + 1;
419 if line.trim().is_empty() {
420 return Err(CoreError::WalBadLine {
421 line: line_no,
422 message: "空行(WAL 不允许空行)".to_string(),
423 });
424 }
425 let parsed: WalLine = match serde_json::from_str(&line) {
426 Ok(parsed) => parsed,
427 Err(e) => return Err(parse_failure(line_no, &line, e)),
428 };
429 if parsed.seq != line_no {
430 return Err(CoreError::WalChainBroken {
431 line: line_no,
432 message: format!(
433 "seq={} 与物理行号 {line_no} 不一致——删行/重排/复制的痕迹",
434 parsed.seq
435 ),
436 });
437 }
438 if parsed.prev != chain {
439 return Err(CoreError::WalChainBroken {
440 line: line_no,
441 message: format!(
442 "prev={} 与按前文重算的链值 {chain} 不符——本行或之前的行被改过,\
443 且后续整条链未重算",
444 parsed.prev
445 ),
446 });
447 }
448 let rec_json = serde_json::to_string(&parsed.rec).map_err(|e| CoreError::WalBadLine {
449 line: line_no,
450 message: format!("记录重序列化失败: {e}"),
451 })?;
452 chain = chain_value(chain, line_no, &rec_json);
453 records.push((line_no, parsed.rec));
454 links.push(WalChainLink {
455 seq: line_no,
456 prev: parsed.prev,
457 value: chain,
458 });
459 }
460 Ok(VerifiedLog {
461 records,
462 links,
463 tail: chain,
464 })
465}
466
467fn parse_failure(line_no: u64, line: &str, error: serde_json::Error) -> CoreError {
470 let legacy_hint = if serde_json::from_str::<WalRecord>(line).is_ok() {
471 ";该行是 W-21 引入完整性链之前的旧格式(裸记录,无 seq/prev 完整性链)。\
472 新旧格式不互通:旧文件原样保留、绝不迁移改写;确认旧日志已留档后,\
473 可将其改名/移走,让闸从一份新日志重新开始"
474 } else {
475 ""
476 };
477 CoreError::WalBadLine {
478 line: line_no,
479 message: format!("JSON 解析失败: {error}{legacy_hint}"),
480 }
481}
482
483pub fn read_records(path: impl AsRef<Path>) -> Result<Vec<(u64, WalRecord)>, CoreError> {
485 Ok(read_verified(path)?.records)
486}
487
488#[cfg(test)]
489mod tests {
490 use super::*;
491
492 fn tmp_path(tag: &str) -> PathBuf {
493 use std::sync::atomic::{AtomicU64, Ordering};
494 static SEQ: AtomicU64 = AtomicU64::new(0);
495 let dir = std::env::temp_dir().join("wanning-wal-tests");
496 std::fs::create_dir_all(&dir).expect("建临时目录");
497 let nanos = std::time::SystemTime::now()
499 .duration_since(std::time::UNIX_EPOCH)
500 .map(|d| d.as_nanos())
501 .unwrap_or(0);
502 dir.join(format!(
503 "{tag}-{}-{}-{nanos}.jsonl",
504 std::process::id(),
505 SEQ.fetch_add(1, Ordering::SeqCst)
506 ))
507 }
508
509 fn sample_record(ts: u64) -> WalRecord {
510 WalRecord::Decide {
511 ts,
512 decision: WalDecision::Allow,
513 delegation_id: "d1".into(),
514 intent: SpendIntent::new("d1", 1, 500, "jd:shop-1", "grocery", "测试"),
515 reason: None,
516 budget_after_cents: 500,
517 }
518 }
519
520 #[test]
521 fn append_is_one_json_per_line_and_counts_lines() {
522 let path = tmp_path("append");
523 let mut wal = Wal::open(&path).expect("打开");
524 assert_eq!(wal.line_count(), 0);
525 assert_eq!(wal.chain_tail(), 0, "空日志链尾 = 创世值 0");
526 assert_eq!(wal.append(&sample_record(1)).expect("写"), 1);
527 assert_eq!(wal.append(&sample_record(2)).expect("写"), 2);
528 drop(wal);
529
530 let lines = raw_lines(&path).expect("读");
531 assert_eq!(lines.len(), 2);
532 assert!(!lines[0].ends_with('\n'), "行内不含换行");
533 let line: WalLine = serde_json::from_str(&lines[0]).expect("逐行可解析");
534 assert_eq!(line.seq, 1, "首行 seq = 物理行号");
535 assert_eq!(line.prev, 0, "首行 prev = 创世值 0");
536 assert_eq!(line.rec.ts(), 1);
537 assert_eq!(line.rec.kind(), "decide");
538 let second: WalLine = serde_json::from_str(&lines[1]).expect("逐行可解析");
539 assert_eq!(second.seq, 2);
540 assert_ne!(second.prev, 0, "第二行 prev 必须是第一行的链值");
541 }
542
543 #[test]
544 fn chain_tail_matches_independent_recompute_and_survives_reopen() {
545 let path = tmp_path("chain-tail");
547 let mut wal = Wal::open(&path).expect("打开");
548 for ts in 1..=3 {
549 wal.append(&sample_record(ts)).expect("写");
550 }
551 let live_tail = wal.chain_tail();
552 drop(wal);
553
554 let verified = read_verified(&path).expect("读回验链");
555 assert_eq!(verified.records.len(), 3);
556 assert_eq!(verified.tail, live_tail, "读侧重算链尾 == 写侧链尾");
557 assert_ne!(live_tail, 0, "三条记录后链尾非 0");
558
559 let mut wal = Wal::open(&path).expect("重开(历史完整)");
561 assert_eq!(wal.line_count(), 3);
562 assert_eq!(wal.chain_tail(), live_tail, "重开后链尾从历史接续");
563 wal.append(&sample_record(4)).expect("续写");
564 let verified = read_verified(&path).expect("续写后读回验链");
565 assert_eq!(verified.records.len(), 4);
566 assert_eq!(verified.tail, wal.chain_tail());
567 }
568
569 #[test]
570 fn read_verified_reports_per_line_chain_links() {
571 let path = tmp_path("links");
574 let mut wal = Wal::open(&path).expect("打开");
575 for ts in 1..=4 {
576 wal.append(&sample_record(ts)).expect("写");
577 }
578 drop(wal);
579
580 let verified = read_verified(&path).expect("读回验链");
581 assert_eq!(
582 verified.links.len(),
583 verified.records.len(),
584 "逐行链与记录一一对应"
585 );
586 for (idx, link) in verified.links.iter().enumerate() {
587 assert_eq!(link.seq, idx as u64 + 1, "link.seq = 物理行号");
588 if idx == 0 {
589 assert_eq!(link.prev, 0, "首行 prev = 创世值 0");
590 } else {
591 assert_eq!(
592 link.prev,
593 verified.links[idx - 1].value,
594 "本行 prev = 前行链值"
595 );
596 }
597 }
598 assert_eq!(
599 verified.links.last().map(|link| link.value),
600 Some(verified.tail),
601 "尾行链值 = 链尾"
602 );
603 }
604
605 #[test]
606 fn empty_wal_has_no_chain_links() {
607 let path = tmp_path("empty-links");
608 std::fs::write(&path, "").expect("写空文件");
609 let verified = read_verified(&path).expect("空文件是合法状态");
610 assert!(verified.links.is_empty(), "空日志无链节");
611 }
612
613 #[test]
614 fn empty_wal_verifies_to_genesis_chain() {
615 let path = tmp_path("empty-chain");
616 std::fs::write(&path, "").expect("写空文件");
617 let verified = read_verified(&path).expect("空文件是合法状态");
618 assert!(verified.records.is_empty());
619 assert_eq!(verified.tail, 0, "空日志链尾 = 创世值 0");
620 }
621
622 #[test]
623 fn open_is_append_only_never_truncates() {
624 let path = tmp_path("append-only");
625 {
626 let mut wal = Wal::open(&path).expect("打开");
627 wal.append(&sample_record(1)).expect("写");
628 }
629 {
630 let mut wal = Wal::open(&path).expect("重开不得截断");
631 assert_eq!(wal.line_count(), 1, "重开必须看到历史行");
632 wal.append(&sample_record(2)).expect("追加");
633 }
634 assert_eq!(raw_lines(&path).expect("读").len(), 2, "历史行必须保留");
635 }
636
637 #[test]
638 fn decide_record_roundtrip_shape() {
639 let deny = WalRecord::Decide {
641 ts: 7,
642 decision: WalDecision::Deny,
643 delegation_id: "d1".into(),
644 intent: SpendIntent::new("d1", 2, 9000, "jd:shop-1", "x", ""),
645 reason: Some(DenyReason::OverBudget),
646 budget_after_cents: 500,
647 };
648 let json = serde_json::to_string(&deny).unwrap();
649 assert!(json.contains("\"kind\":\"decide\""));
650 assert!(json.contains("\"decision\":\"deny\""));
651 assert!(json.contains("\"reason\":\"over_budget\""));
652 let back: WalRecord = serde_json::from_str(&json).unwrap();
653 assert_eq!(back, deny);
654
655 let allow_json = serde_json::to_string(&sample_record(1)).unwrap();
656 assert!(!allow_json.contains("reason"), "Allow 不应带 reason 字段");
657 }
658
659 #[test]
660 fn read_records_fails_closed_on_half_line() {
661 let path = tmp_path("corrupt");
662 std::fs::write(&path, "{\"kind\":\"revoke\",\"ts\":1,\"deleg\n").expect("写坏行");
663 let err = read_records(&path).unwrap_err();
664 assert!(
665 matches!(err, CoreError::WalBadLine { line: 1, .. }),
666 "半行 JSON 必须 fail-closed 报错: {err:?}"
667 );
668 }
669
670 #[test]
671 fn read_records_fails_closed_on_blank_line() {
672 let path = tmp_path("blank");
673 std::fs::write(&path, "\n").expect("写空行");
674 let err = read_records(&path).unwrap_err();
675 assert!(
676 matches!(err, CoreError::WalBadLine { line: 1, .. }),
677 "{err:?}"
678 );
679 }
680
681 #[test]
682 fn read_records_fails_closed_on_unknown_shape() {
683 let path = tmp_path("unknown");
684 std::fs::write(&path, "{\"kind\":\"mystery\",\"ts\":1}\n").expect("写");
685 let err = read_records(&path).unwrap_err();
686 assert!(
687 matches!(err, CoreError::WalBadLine { line: 1, .. }),
688 "{err:?}"
689 );
690 }
691
692 #[test]
693 fn read_records_reports_failing_line_number() {
694 let path = tmp_path("line3");
695 let mut wal = Wal::open(&path).expect("打开");
696 wal.append(&sample_record(1)).expect("写");
697 wal.append(&sample_record(2)).expect("写");
698 drop(wal);
699 let mut content = raw_lines(&path).expect("读").join("\n");
700 content.push_str("\n{\"kind\":\"decide\",\"ts\":3\n");
701 std::fs::write(&path, content).expect("追加坏行");
702
703 match read_records(&path) {
704 Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 3, "报错必须指到坏行"),
705 other => panic!("应报 WalBadLine,实际 {other:?}"),
706 }
707 }
708}