1use std::fs::{File, OpenOptions};
43use std::io::{BufRead, BufReader, Write};
44use std::path::{Path, PathBuf};
45
46use serde::{Deserialize, Serialize};
47
48use crate::delegation::Delegation;
49use crate::error::CoreError;
50use crate::gate::DenyReason;
51use crate::intent::SpendIntent;
52
53#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
55#[serde(rename_all = "snake_case")]
56pub enum WalDecision {
57 Allow,
58 Deny,
59}
60
61#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(tag = "kind", rename_all = "snake_case")]
64pub enum WalRecord {
65 RegisterDelegation { ts: u64, delegation: Delegation },
67 Revoke { ts: u64, delegation_id: String },
69 Decide {
71 ts: u64,
72 decision: WalDecision,
73 delegation_id: String,
74 intent: SpendIntent,
75 #[serde(skip_serializing_if = "Option::is_none")]
77 reason: Option<DenyReason>,
78 budget_after_cents: u64,
80 },
81}
82
83impl WalRecord {
84 pub fn ts(&self) -> u64 {
86 match self {
87 WalRecord::RegisterDelegation { ts, .. }
88 | WalRecord::Revoke { ts, .. }
89 | WalRecord::Decide { ts, .. } => *ts,
90 }
91 }
92
93 pub fn kind(&self) -> &'static str {
95 match self {
96 WalRecord::RegisterDelegation { .. } => "register_delegation",
97 WalRecord::Revoke { .. } => "revoke",
98 WalRecord::Decide { .. } => "decide",
99 }
100 }
101}
102
103#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
108pub struct WalLine {
109 pub seq: u64,
111 pub prev: u64,
113 pub rec: WalRecord,
115}
116
117pub(crate) fn chain_value(prev: u64, seq: u64, rec_json: &str) -> u64 {
123 let mut bytes = Vec::with_capacity(16 + rec_json.len());
124 bytes.extend_from_slice(&prev.to_le_bytes());
125 bytes.extend_from_slice(&seq.to_le_bytes());
126 bytes.extend_from_slice(rec_json.as_bytes());
127 fnv1a_64(&bytes)
128}
129
130pub(crate) fn fnv1a_64(bytes: &[u8]) -> u64 {
132 const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
133 const PRIME: u64 = 0x0000_0100_0000_01b3;
134 let mut hash = OFFSET_BASIS;
135 for byte in bytes {
136 hash ^= u64::from(*byte);
137 hash = hash.wrapping_mul(PRIME);
138 }
139 hash
140}
141
142#[derive(Debug)]
145pub struct Wal {
146 file: File,
147 path: PathBuf,
148 lines: u64,
149 chain: u64,
151 _lock: WalLock,
153}
154
155pub fn single_writer_lock_path(wal_path: impl AsRef<Path>) -> PathBuf {
160 let wal_path = wal_path.as_ref();
161 let mut name = wal_path
162 .file_name()
163 .map(|n| n.to_os_string())
164 .unwrap_or_default();
165 name.push(".lock");
166 wal_path.with_file_name(name)
167}
168
169#[derive(Debug)]
191pub struct WalLock {
192 path: PathBuf,
193}
194
195impl WalLock {
196 pub fn acquire(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
198 let wal_path = wal_path.as_ref();
199 let lock_path = single_writer_lock_path(wal_path);
200 match OpenOptions::new()
201 .write(true)
202 .create_new(true)
203 .open(&lock_path)
204 {
205 Ok(mut file) => {
206 let written = writeln!(file, "pid={}", std::process::id())
207 .and_then(|()| writeln!(file, "wal={}", wal_path.display()))
208 .and_then(|()| file.flush());
209 if let Err(e) = written {
210 let _ = std::fs::remove_file(&lock_path);
212 return Err(CoreError::WalIo(format!(
213 "写单写者锁 {lock_path:?} 失败(fail-closed): {e}"
214 )));
215 }
216 Ok(Self { path: lock_path })
217 }
218 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
219 let holder = std::fs::read_to_string(&lock_path).unwrap_or_default();
220 let holder = holder.trim();
221 let holder_note = if holder.is_empty() {
222 "锁文件为空(持锁方刚创建,极可能是并发启动竞争)".to_string()
223 } else {
224 format!("持锁信息: {holder}")
225 };
226 Err(CoreError::WalLocked {
227 path: lock_path.display().to_string(),
228 message: format!(
229 "同一份审计日志已有另一个 Wanning 进程在写({holder_note});\
230 确认没有别的闸在跑后,删除该锁文件即可恢复\
231 (默认 WAL 在 target/ 下,cargo clean 亦可)"
232 ),
233 })
234 }
235 Err(e) => Err(CoreError::WalIo(format!(
236 "创建单写者锁 {lock_path:?} 失败: {e}"
237 ))),
238 }
239 }
240}
241
242impl Drop for WalLock {
243 fn drop(&mut self) {
244 let _ = std::fs::remove_file(&self.path);
246 }
247}
248
249impl Wal {
250 pub fn open(path: impl AsRef<Path>) -> Result<Self, CoreError> {
256 let path = path.as_ref().to_path_buf();
257 let _lock = WalLock::acquire(&path)?;
258 let (existing_lines, chain) = if path.exists() {
260 let verified = read_verified(&path)?;
261 (verified.records.len() as u64, verified.tail)
262 } else {
263 (0, 0)
264 };
265 let file = OpenOptions::new()
266 .create(true)
267 .append(true)
268 .read(false)
269 .open(&path)
270 .map_err(|e| CoreError::WalIo(format!("打开 WAL {path:?} 失败: {e}")))?;
271 Ok(Self {
272 file,
273 path,
274 lines: existing_lines,
275 chain,
276 _lock,
277 })
278 }
279
280 pub fn path(&self) -> &Path {
281 &self.path
282 }
283
284 pub fn line_count(&self) -> u64 {
286 self.lines
287 }
288
289 pub fn chain_tail(&self) -> u64 {
291 self.chain
292 }
293
294 pub fn append(&mut self, record: &WalRecord) -> Result<u64, CoreError> {
299 let seq = self.lines + 1;
300 let rec_json = serde_json::to_string(record)
301 .map_err(|e| CoreError::WalIo(format!("WAL 记录序列化失败: {e}")))?;
302 let mut line = serde_json::to_string(&WalLine {
303 seq,
304 prev: self.chain,
305 rec: record.clone(),
306 })
307 .map_err(|e| CoreError::WalIo(format!("WAL 记录序列化失败: {e}")))?;
308 line.push('\n');
309 let path = self.path.clone();
310 self.file
311 .write_all(line.as_bytes())
312 .and_then(|()| self.file.flush())
313 .map_err(|e| CoreError::WalIo(format!("写 WAL {path:?} 失败: {e}")))?;
314 self.lines = seq;
315 self.chain = chain_value(self.chain, seq, &rec_json);
316 Ok(self.lines)
317 }
318}
319
320pub fn raw_lines(path: impl AsRef<Path>) -> Result<Vec<String>, CoreError> {
322 let path = path.as_ref();
323 let file =
324 File::open(path).map_err(|e| CoreError::WalIo(format!("读 WAL {path:?} 失败: {e}")))?;
325 BufReader::new(file)
326 .lines()
327 .collect::<Result<Vec<_>, _>>()
328 .map_err(|e| CoreError::WalIo(format!("读 WAL {path:?} 失败: {e}")))
329}
330
331#[derive(Clone, Debug, PartialEq, Eq)]
333pub struct VerifiedLog {
334 pub records: Vec<(u64, WalRecord)>,
336 pub links: Vec<WalChainLink>,
340 pub tail: u64,
343}
344
345#[derive(Clone, Copy, Debug, PartialEq, Eq)]
347pub struct WalChainLink {
348 pub seq: u64,
350 pub prev: u64,
352 pub value: u64,
354}
355
356pub fn read_verified(path: impl AsRef<Path>) -> Result<VerifiedLog, CoreError> {
367 let mut records = Vec::new();
368 let mut links = Vec::new();
369 let mut chain = 0u64;
370 for (idx, line) in raw_lines(path)?.into_iter().enumerate() {
371 let line_no = idx as u64 + 1;
372 if line.trim().is_empty() {
373 return Err(CoreError::WalBadLine {
374 line: line_no,
375 message: "空行(WAL 不允许空行)".to_string(),
376 });
377 }
378 let parsed: WalLine = match serde_json::from_str(&line) {
379 Ok(parsed) => parsed,
380 Err(e) => return Err(parse_failure(line_no, &line, e)),
381 };
382 if parsed.seq != line_no {
383 return Err(CoreError::WalChainBroken {
384 line: line_no,
385 message: format!(
386 "seq={} 与物理行号 {line_no} 不一致——删行/重排/复制的痕迹",
387 parsed.seq
388 ),
389 });
390 }
391 if parsed.prev != chain {
392 return Err(CoreError::WalChainBroken {
393 line: line_no,
394 message: format!(
395 "prev={} 与按前文重算的链值 {chain} 不符——本行或之前的行被改过,\
396 且后续整条链未重算",
397 parsed.prev
398 ),
399 });
400 }
401 let rec_json = serde_json::to_string(&parsed.rec).map_err(|e| CoreError::WalBadLine {
402 line: line_no,
403 message: format!("记录重序列化失败: {e}"),
404 })?;
405 chain = chain_value(chain, line_no, &rec_json);
406 records.push((line_no, parsed.rec));
407 links.push(WalChainLink {
408 seq: line_no,
409 prev: parsed.prev,
410 value: chain,
411 });
412 }
413 Ok(VerifiedLog {
414 records,
415 links,
416 tail: chain,
417 })
418}
419
420fn parse_failure(line_no: u64, line: &str, error: serde_json::Error) -> CoreError {
423 let legacy_hint = if serde_json::from_str::<WalRecord>(line).is_ok() {
424 ";该行是 W-21 引入完整性链之前的旧格式(裸记录,无 seq/prev 完整性链)。\
425 新旧格式不互通:旧文件原样保留、绝不迁移改写;确认旧日志已留档后,\
426 可将其改名/移走,让闸从一份新日志重新开始"
427 } else {
428 ""
429 };
430 CoreError::WalBadLine {
431 line: line_no,
432 message: format!("JSON 解析失败: {error}{legacy_hint}"),
433 }
434}
435
436pub fn read_records(path: impl AsRef<Path>) -> Result<Vec<(u64, WalRecord)>, CoreError> {
438 Ok(read_verified(path)?.records)
439}
440
441#[cfg(test)]
442mod tests {
443 use super::*;
444
445 fn tmp_path(tag: &str) -> PathBuf {
446 let dir = std::env::temp_dir().join("wanning-wal-tests");
447 std::fs::create_dir_all(&dir).expect("建临时目录");
448 dir.join(format!("{tag}-{}.jsonl", std::process::id()))
449 }
450
451 fn sample_record(ts: u64) -> WalRecord {
452 WalRecord::Decide {
453 ts,
454 decision: WalDecision::Allow,
455 delegation_id: "d1".into(),
456 intent: SpendIntent::new("d1", 1, 500, "jd:shop-1", "grocery", "测试"),
457 reason: None,
458 budget_after_cents: 500,
459 }
460 }
461
462 #[test]
463 fn append_is_one_json_per_line_and_counts_lines() {
464 let path = tmp_path("append");
465 let mut wal = Wal::open(&path).expect("打开");
466 assert_eq!(wal.line_count(), 0);
467 assert_eq!(wal.chain_tail(), 0, "空日志链尾 = 创世值 0");
468 assert_eq!(wal.append(&sample_record(1)).expect("写"), 1);
469 assert_eq!(wal.append(&sample_record(2)).expect("写"), 2);
470 drop(wal);
471
472 let lines = raw_lines(&path).expect("读");
473 assert_eq!(lines.len(), 2);
474 assert!(!lines[0].ends_with('\n'), "行内不含换行");
475 let line: WalLine = serde_json::from_str(&lines[0]).expect("逐行可解析");
476 assert_eq!(line.seq, 1, "首行 seq = 物理行号");
477 assert_eq!(line.prev, 0, "首行 prev = 创世值 0");
478 assert_eq!(line.rec.ts(), 1);
479 assert_eq!(line.rec.kind(), "decide");
480 let second: WalLine = serde_json::from_str(&lines[1]).expect("逐行可解析");
481 assert_eq!(second.seq, 2);
482 assert_ne!(second.prev, 0, "第二行 prev 必须是第一行的链值");
483 }
484
485 #[test]
486 fn chain_tail_matches_independent_recompute_and_survives_reopen() {
487 let path = tmp_path("chain-tail");
489 let mut wal = Wal::open(&path).expect("打开");
490 for ts in 1..=3 {
491 wal.append(&sample_record(ts)).expect("写");
492 }
493 let live_tail = wal.chain_tail();
494 drop(wal);
495
496 let verified = read_verified(&path).expect("读回验链");
497 assert_eq!(verified.records.len(), 3);
498 assert_eq!(verified.tail, live_tail, "读侧重算链尾 == 写侧链尾");
499 assert_ne!(live_tail, 0, "三条记录后链尾非 0");
500
501 let mut wal = Wal::open(&path).expect("重开(历史完整)");
503 assert_eq!(wal.line_count(), 3);
504 assert_eq!(wal.chain_tail(), live_tail, "重开后链尾从历史接续");
505 wal.append(&sample_record(4)).expect("续写");
506 let verified = read_verified(&path).expect("续写后读回验链");
507 assert_eq!(verified.records.len(), 4);
508 assert_eq!(verified.tail, wal.chain_tail());
509 }
510
511 #[test]
512 fn read_verified_reports_per_line_chain_links() {
513 let path = tmp_path("links");
516 let mut wal = Wal::open(&path).expect("打开");
517 for ts in 1..=4 {
518 wal.append(&sample_record(ts)).expect("写");
519 }
520 drop(wal);
521
522 let verified = read_verified(&path).expect("读回验链");
523 assert_eq!(
524 verified.links.len(),
525 verified.records.len(),
526 "逐行链与记录一一对应"
527 );
528 for (idx, link) in verified.links.iter().enumerate() {
529 assert_eq!(link.seq, idx as u64 + 1, "link.seq = 物理行号");
530 if idx == 0 {
531 assert_eq!(link.prev, 0, "首行 prev = 创世值 0");
532 } else {
533 assert_eq!(
534 link.prev,
535 verified.links[idx - 1].value,
536 "本行 prev = 前行链值"
537 );
538 }
539 }
540 assert_eq!(
541 verified.links.last().map(|link| link.value),
542 Some(verified.tail),
543 "尾行链值 = 链尾"
544 );
545 }
546
547 #[test]
548 fn empty_wal_has_no_chain_links() {
549 let path = tmp_path("empty-links");
550 std::fs::write(&path, "").expect("写空文件");
551 let verified = read_verified(&path).expect("空文件是合法状态");
552 assert!(verified.links.is_empty(), "空日志无链节");
553 }
554
555 #[test]
556 fn empty_wal_verifies_to_genesis_chain() {
557 let path = tmp_path("empty-chain");
558 std::fs::write(&path, "").expect("写空文件");
559 let verified = read_verified(&path).expect("空文件是合法状态");
560 assert!(verified.records.is_empty());
561 assert_eq!(verified.tail, 0, "空日志链尾 = 创世值 0");
562 }
563
564 #[test]
565 fn open_is_append_only_never_truncates() {
566 let path = tmp_path("append-only");
567 {
568 let mut wal = Wal::open(&path).expect("打开");
569 wal.append(&sample_record(1)).expect("写");
570 }
571 {
572 let mut wal = Wal::open(&path).expect("重开不得截断");
573 assert_eq!(wal.line_count(), 1, "重开必须看到历史行");
574 wal.append(&sample_record(2)).expect("追加");
575 }
576 assert_eq!(raw_lines(&path).expect("读").len(), 2, "历史行必须保留");
577 }
578
579 #[test]
580 fn decide_record_roundtrip_shape() {
581 let deny = WalRecord::Decide {
583 ts: 7,
584 decision: WalDecision::Deny,
585 delegation_id: "d1".into(),
586 intent: SpendIntent::new("d1", 2, 9000, "jd:shop-1", "x", ""),
587 reason: Some(DenyReason::OverBudget),
588 budget_after_cents: 500,
589 };
590 let json = serde_json::to_string(&deny).unwrap();
591 assert!(json.contains("\"kind\":\"decide\""));
592 assert!(json.contains("\"decision\":\"deny\""));
593 assert!(json.contains("\"reason\":\"over_budget\""));
594 let back: WalRecord = serde_json::from_str(&json).unwrap();
595 assert_eq!(back, deny);
596
597 let allow_json = serde_json::to_string(&sample_record(1)).unwrap();
598 assert!(!allow_json.contains("reason"), "Allow 不应带 reason 字段");
599 }
600
601 #[test]
602 fn read_records_fails_closed_on_half_line() {
603 let path = tmp_path("corrupt");
604 std::fs::write(&path, "{\"kind\":\"revoke\",\"ts\":1,\"deleg\n").expect("写坏行");
605 let err = read_records(&path).unwrap_err();
606 assert!(
607 matches!(err, CoreError::WalBadLine { line: 1, .. }),
608 "半行 JSON 必须 fail-closed 报错: {err:?}"
609 );
610 }
611
612 #[test]
613 fn read_records_fails_closed_on_blank_line() {
614 let path = tmp_path("blank");
615 std::fs::write(&path, "\n").expect("写空行");
616 let err = read_records(&path).unwrap_err();
617 assert!(
618 matches!(err, CoreError::WalBadLine { line: 1, .. }),
619 "{err:?}"
620 );
621 }
622
623 #[test]
624 fn read_records_fails_closed_on_unknown_shape() {
625 let path = tmp_path("unknown");
626 std::fs::write(&path, "{\"kind\":\"mystery\",\"ts\":1}\n").expect("写");
627 let err = read_records(&path).unwrap_err();
628 assert!(
629 matches!(err, CoreError::WalBadLine { line: 1, .. }),
630 "{err:?}"
631 );
632 }
633
634 #[test]
635 fn read_records_reports_failing_line_number() {
636 let path = tmp_path("line3");
637 let mut wal = Wal::open(&path).expect("打开");
638 wal.append(&sample_record(1)).expect("写");
639 wal.append(&sample_record(2)).expect("写");
640 drop(wal);
641 let mut content = raw_lines(&path).expect("读").join("\n");
642 content.push_str("\n{\"kind\":\"decide\",\"ts\":3\n");
643 std::fs::write(&path, content).expect("追加坏行");
644
645 match read_records(&path) {
646 Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 3, "报错必须指到坏行"),
647 other => panic!("应报 WalBadLine,实际 {other:?}"),
648 }
649 }
650}