use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use crate::delegation::Delegation;
use crate::error::CoreError;
use crate::gate::DenyReason;
use crate::intent::SpendIntent;
use crate::pending::PendingOutcome;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WalDecision {
Allow,
Deny,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum WalRecord {
RegisterDelegation { ts: u64, delegation: Delegation },
Revoke { ts: u64, delegation_id: String },
Decide {
ts: u64,
decision: WalDecision,
delegation_id: String,
intent: SpendIntent,
#[serde(skip_serializing_if = "Option::is_none")]
reason: Option<DenyReason>,
budget_after_cents: u64,
},
Pending {
ts: u64,
pending_id: String,
delegation_id: String,
intent: SpendIntent,
approved_amount_cents: u64,
expires_ts: u64,
},
Confirm {
ts: u64,
pending_id: String,
amount_cents: u64,
proof: String,
},
Terminal {
ts: u64,
pending_id: String,
outcome: PendingOutcome,
},
}
impl WalRecord {
pub fn ts(&self) -> u64 {
match self {
WalRecord::RegisterDelegation { ts, .. }
| WalRecord::Revoke { ts, .. }
| WalRecord::Decide { ts, .. }
| WalRecord::Pending { ts, .. }
| WalRecord::Confirm { ts, .. }
| WalRecord::Terminal { ts, .. } => *ts,
}
}
pub fn kind(&self) -> &'static str {
match self {
WalRecord::RegisterDelegation { .. } => "register_delegation",
WalRecord::Revoke { .. } => "revoke",
WalRecord::Decide { .. } => "decide",
WalRecord::Pending { .. } => "pending",
WalRecord::Confirm { .. } => "confirm",
WalRecord::Terminal { .. } => "terminal",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WalLine {
pub seq: u64,
pub prev: u64,
pub rec: WalRecord,
}
pub(crate) fn chain_value(prev: u64, seq: u64, rec_json: &str) -> u64 {
let mut bytes = Vec::with_capacity(16 + rec_json.len());
bytes.extend_from_slice(&prev.to_le_bytes());
bytes.extend_from_slice(&seq.to_le_bytes());
bytes.extend_from_slice(rec_json.as_bytes());
fnv1a_64(&bytes)
}
pub(crate) fn fnv1a_64(bytes: &[u8]) -> u64 {
const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
const PRIME: u64 = 0x0000_0100_0000_01b3;
let mut hash = OFFSET_BASIS;
for byte in bytes {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(PRIME);
}
hash
}
#[derive(Debug)]
pub struct Wal {
file: File,
path: PathBuf,
lines: u64,
chain: u64,
_lock: WalLock,
}
pub fn single_writer_lock_path(wal_path: impl AsRef<Path>) -> PathBuf {
let wal_path = wal_path.as_ref();
let mut name = wal_path
.file_name()
.map(|n| n.to_os_string())
.unwrap_or_default();
name.push(".lock");
wal_path.with_file_name(name)
}
#[derive(Debug)]
pub struct WalLock {
path: PathBuf,
}
impl WalLock {
pub fn acquire(wal_path: impl AsRef<Path>) -> Result<Self, CoreError> {
let wal_path = wal_path.as_ref();
let lock_path = single_writer_lock_path(wal_path);
match OpenOptions::new()
.write(true)
.create_new(true)
.open(&lock_path)
{
Ok(mut file) => {
let written = writeln!(file, "pid={}", std::process::id())
.and_then(|()| writeln!(file, "wal={}", wal_path.display()))
.and_then(|()| file.flush());
if let Err(e) = written {
let _ = std::fs::remove_file(&lock_path);
return Err(CoreError::WalIo(format!(
"写单写者锁 {lock_path:?} 失败(fail-closed): {e}"
)));
}
Ok(Self { path: lock_path })
}
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
let holder = std::fs::read_to_string(&lock_path).unwrap_or_default();
let holder = holder.trim();
let holder_note = if holder.is_empty() {
"锁文件为空(持锁方刚创建,极可能是并发启动竞争)".to_string()
} else {
format!("持锁信息: {holder}")
};
Err(CoreError::WalLocked {
path: lock_path.display().to_string(),
message: format!(
"同一份审计日志已有另一个 Wanning 进程在写({holder_note});\
确认没有别的闸在跑后,删除该锁文件即可恢复\
(默认 WAL 在 target/ 下,cargo clean 亦可)"
),
})
}
Err(e) => Err(CoreError::WalIo(format!(
"创建单写者锁 {lock_path:?} 失败: {e}"
))),
}
}
}
impl Drop for WalLock {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.path);
}
}
impl Wal {
pub fn open(path: impl AsRef<Path>) -> Result<Self, CoreError> {
let path = path.as_ref().to_path_buf();
crate::paths::ensure_wal_parent(&path)?;
let _lock = WalLock::acquire(&path)?;
let (existing_lines, chain) = if path.exists() {
let verified = read_verified(&path)?;
(verified.records.len() as u64, verified.tail)
} else {
(0, 0)
};
let file = OpenOptions::new()
.create(true)
.append(true)
.read(false)
.open(&path)
.map_err(|e| CoreError::WalIo(format!("打开 WAL {path:?} 失败: {e}")))?;
Ok(Self {
file,
path,
lines: existing_lines,
chain,
_lock,
})
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn line_count(&self) -> u64 {
self.lines
}
pub fn chain_tail(&self) -> u64 {
self.chain
}
pub fn append(&mut self, record: &WalRecord) -> Result<u64, CoreError> {
let seq = self.lines + 1;
let rec_json = serde_json::to_string(record)
.map_err(|e| CoreError::WalIo(format!("WAL 记录序列化失败: {e}")))?;
let mut line = serde_json::to_string(&WalLine {
seq,
prev: self.chain,
rec: record.clone(),
})
.map_err(|e| CoreError::WalIo(format!("WAL 记录序列化失败: {e}")))?;
line.push('\n');
let path = self.path.clone();
self.file
.write_all(line.as_bytes())
.and_then(|()| self.file.flush())
.map_err(|e| CoreError::WalIo(format!("写 WAL {path:?} 失败: {e}")))?;
self.lines = seq;
self.chain = chain_value(self.chain, seq, &rec_json);
Ok(self.lines)
}
}
pub fn raw_lines(path: impl AsRef<Path>) -> Result<Vec<String>, CoreError> {
let path = path.as_ref();
let file =
File::open(path).map_err(|e| CoreError::WalIo(format!("读 WAL {path:?} 失败: {e}")))?;
BufReader::new(file)
.lines()
.collect::<Result<Vec<_>, _>>()
.map_err(|e| CoreError::WalIo(format!("读 WAL {path:?} 失败: {e}")))
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct VerifiedLog {
pub records: Vec<(u64, WalRecord)>,
pub links: Vec<WalChainLink>,
pub tail: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WalChainLink {
pub seq: u64,
pub prev: u64,
pub value: u64,
}
pub fn read_verified(path: impl AsRef<Path>) -> Result<VerifiedLog, CoreError> {
let mut records = Vec::new();
let mut links = Vec::new();
let mut chain = 0u64;
for (idx, line) in raw_lines(path)?.into_iter().enumerate() {
let line_no = idx as u64 + 1;
if line.trim().is_empty() {
return Err(CoreError::WalBadLine {
line: line_no,
message: "空行(WAL 不允许空行)".to_string(),
});
}
let parsed: WalLine = match serde_json::from_str(&line) {
Ok(parsed) => parsed,
Err(e) => return Err(parse_failure(line_no, &line, e)),
};
if parsed.seq != line_no {
return Err(CoreError::WalChainBroken {
line: line_no,
message: format!(
"seq={} 与物理行号 {line_no} 不一致——删行/重排/复制的痕迹",
parsed.seq
),
});
}
if parsed.prev != chain {
return Err(CoreError::WalChainBroken {
line: line_no,
message: format!(
"prev={} 与按前文重算的链值 {chain} 不符——本行或之前的行被改过,\
且后续整条链未重算",
parsed.prev
),
});
}
let rec_json = serde_json::to_string(&parsed.rec).map_err(|e| CoreError::WalBadLine {
line: line_no,
message: format!("记录重序列化失败: {e}"),
})?;
chain = chain_value(chain, line_no, &rec_json);
records.push((line_no, parsed.rec));
links.push(WalChainLink {
seq: line_no,
prev: parsed.prev,
value: chain,
});
}
Ok(VerifiedLog {
records,
links,
tail: chain,
})
}
fn parse_failure(line_no: u64, line: &str, error: serde_json::Error) -> CoreError {
let legacy_hint = if serde_json::from_str::<WalRecord>(line).is_ok() {
";该行是 W-21 引入完整性链之前的旧格式(裸记录,无 seq/prev 完整性链)。\
新旧格式不互通:旧文件原样保留、绝不迁移改写;确认旧日志已留档后,\
可将其改名/移走,让闸从一份新日志重新开始"
} else {
""
};
CoreError::WalBadLine {
line: line_no,
message: format!("JSON 解析失败: {error}{legacy_hint}"),
}
}
pub fn read_records(path: impl AsRef<Path>) -> Result<Vec<(u64, WalRecord)>, CoreError> {
Ok(read_verified(path)?.records)
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp_path(tag: &str) -> PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static SEQ: AtomicU64 = AtomicU64::new(0);
let dir = std::env::temp_dir().join("wanning-wal-tests");
std::fs::create_dir_all(&dir).expect("建临时目录");
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
dir.join(format!(
"{tag}-{}-{}-{nanos}.jsonl",
std::process::id(),
SEQ.fetch_add(1, Ordering::SeqCst)
))
}
fn sample_record(ts: u64) -> WalRecord {
WalRecord::Decide {
ts,
decision: WalDecision::Allow,
delegation_id: "d1".into(),
intent: SpendIntent::new("d1", 1, 500, "jd:shop-1", "grocery", "测试"),
reason: None,
budget_after_cents: 500,
}
}
#[test]
fn append_is_one_json_per_line_and_counts_lines() {
let path = tmp_path("append");
let mut wal = Wal::open(&path).expect("打开");
assert_eq!(wal.line_count(), 0);
assert_eq!(wal.chain_tail(), 0, "空日志链尾 = 创世值 0");
assert_eq!(wal.append(&sample_record(1)).expect("写"), 1);
assert_eq!(wal.append(&sample_record(2)).expect("写"), 2);
drop(wal);
let lines = raw_lines(&path).expect("读");
assert_eq!(lines.len(), 2);
assert!(!lines[0].ends_with('\n'), "行内不含换行");
let line: WalLine = serde_json::from_str(&lines[0]).expect("逐行可解析");
assert_eq!(line.seq, 1, "首行 seq = 物理行号");
assert_eq!(line.prev, 0, "首行 prev = 创世值 0");
assert_eq!(line.rec.ts(), 1);
assert_eq!(line.rec.kind(), "decide");
let second: WalLine = serde_json::from_str(&lines[1]).expect("逐行可解析");
assert_eq!(second.seq, 2);
assert_ne!(second.prev, 0, "第二行 prev 必须是第一行的链值");
}
#[test]
fn chain_tail_matches_independent_recompute_and_survives_reopen() {
let path = tmp_path("chain-tail");
let mut wal = Wal::open(&path).expect("打开");
for ts in 1..=3 {
wal.append(&sample_record(ts)).expect("写");
}
let live_tail = wal.chain_tail();
drop(wal);
let verified = read_verified(&path).expect("读回验链");
assert_eq!(verified.records.len(), 3);
assert_eq!(verified.tail, live_tail, "读侧重算链尾 == 写侧链尾");
assert_ne!(live_tail, 0, "三条记录后链尾非 0");
let mut wal = Wal::open(&path).expect("重开(历史完整)");
assert_eq!(wal.line_count(), 3);
assert_eq!(wal.chain_tail(), live_tail, "重开后链尾从历史接续");
wal.append(&sample_record(4)).expect("续写");
let verified = read_verified(&path).expect("续写后读回验链");
assert_eq!(verified.records.len(), 4);
assert_eq!(verified.tail, wal.chain_tail());
}
#[test]
fn read_verified_reports_per_line_chain_links() {
let path = tmp_path("links");
let mut wal = Wal::open(&path).expect("打开");
for ts in 1..=4 {
wal.append(&sample_record(ts)).expect("写");
}
drop(wal);
let verified = read_verified(&path).expect("读回验链");
assert_eq!(
verified.links.len(),
verified.records.len(),
"逐行链与记录一一对应"
);
for (idx, link) in verified.links.iter().enumerate() {
assert_eq!(link.seq, idx as u64 + 1, "link.seq = 物理行号");
if idx == 0 {
assert_eq!(link.prev, 0, "首行 prev = 创世值 0");
} else {
assert_eq!(
link.prev,
verified.links[idx - 1].value,
"本行 prev = 前行链值"
);
}
}
assert_eq!(
verified.links.last().map(|link| link.value),
Some(verified.tail),
"尾行链值 = 链尾"
);
}
#[test]
fn empty_wal_has_no_chain_links() {
let path = tmp_path("empty-links");
std::fs::write(&path, "").expect("写空文件");
let verified = read_verified(&path).expect("空文件是合法状态");
assert!(verified.links.is_empty(), "空日志无链节");
}
#[test]
fn empty_wal_verifies_to_genesis_chain() {
let path = tmp_path("empty-chain");
std::fs::write(&path, "").expect("写空文件");
let verified = read_verified(&path).expect("空文件是合法状态");
assert!(verified.records.is_empty());
assert_eq!(verified.tail, 0, "空日志链尾 = 创世值 0");
}
#[test]
fn open_is_append_only_never_truncates() {
let path = tmp_path("append-only");
{
let mut wal = Wal::open(&path).expect("打开");
wal.append(&sample_record(1)).expect("写");
}
{
let mut wal = Wal::open(&path).expect("重开不得截断");
assert_eq!(wal.line_count(), 1, "重开必须看到历史行");
wal.append(&sample_record(2)).expect("追加");
}
assert_eq!(raw_lines(&path).expect("读").len(), 2, "历史行必须保留");
}
#[test]
fn decide_record_roundtrip_shape() {
let deny = WalRecord::Decide {
ts: 7,
decision: WalDecision::Deny,
delegation_id: "d1".into(),
intent: SpendIntent::new("d1", 2, 9000, "jd:shop-1", "x", ""),
reason: Some(DenyReason::OverBudget),
budget_after_cents: 500,
};
let json = serde_json::to_string(&deny).unwrap();
assert!(json.contains("\"kind\":\"decide\""));
assert!(json.contains("\"decision\":\"deny\""));
assert!(json.contains("\"reason\":\"over_budget\""));
let back: WalRecord = serde_json::from_str(&json).unwrap();
assert_eq!(back, deny);
let allow_json = serde_json::to_string(&sample_record(1)).unwrap();
assert!(!allow_json.contains("reason"), "Allow 不应带 reason 字段");
}
#[test]
fn read_records_fails_closed_on_half_line() {
let path = tmp_path("corrupt");
std::fs::write(&path, "{\"kind\":\"revoke\",\"ts\":1,\"deleg\n").expect("写坏行");
let err = read_records(&path).unwrap_err();
assert!(
matches!(err, CoreError::WalBadLine { line: 1, .. }),
"半行 JSON 必须 fail-closed 报错: {err:?}"
);
}
#[test]
fn read_records_fails_closed_on_blank_line() {
let path = tmp_path("blank");
std::fs::write(&path, "\n").expect("写空行");
let err = read_records(&path).unwrap_err();
assert!(
matches!(err, CoreError::WalBadLine { line: 1, .. }),
"{err:?}"
);
}
#[test]
fn read_records_fails_closed_on_unknown_shape() {
let path = tmp_path("unknown");
std::fs::write(&path, "{\"kind\":\"mystery\",\"ts\":1}\n").expect("写");
let err = read_records(&path).unwrap_err();
assert!(
matches!(err, CoreError::WalBadLine { line: 1, .. }),
"{err:?}"
);
}
#[test]
fn read_records_reports_failing_line_number() {
let path = tmp_path("line3");
let mut wal = Wal::open(&path).expect("打开");
wal.append(&sample_record(1)).expect("写");
wal.append(&sample_record(2)).expect("写");
drop(wal);
let mut content = raw_lines(&path).expect("读").join("\n");
content.push_str("\n{\"kind\":\"decide\",\"ts\":3\n");
std::fs::write(&path, content).expect("追加坏行");
match read_records(&path) {
Err(CoreError::WalBadLine { line, .. }) => assert_eq!(line, 3, "报错必须指到坏行"),
other => panic!("应报 WalBadLine,实际 {other:?}"),
}
}
}