pagesift_store/
lib.rs

1#![forbid(unsafe_code)]
2//! pagesift 的内嵌存储：只追加的分段日志 + 内存哈希索引（PRD 6.8、F-6.4）。
3//!
4//! 三个命名空间共用一套日志：
5//! - [`Cache`]：抓取结果缓存，语义对齐 Firecrawl 的 `maxAge`、`minAge`、`storeInCache`；
6//! - [`Snapshots`]：原始 HTML 快照，zstd 压缩，按内容哈希去重；
7//! - [`Tasks`]：爬取任务状态用的 KV、追加列表和集合。
8//!
9//! # 并发与异步
10//!
11//! [`Store`] 是 `Clone + Send + Sync` 的句柄，多读单写：读只拿读锁并用 pread 读盘，
12//! 写操作串行。所有方法都是同步的，单次调用通常只有一次系统调用；
13//! 在 tokio 里可以用 [`Store::blocking`]（`tokio` 特性，默认开启）放进 `spawn_blocking`。
14//!
15//! # 持久性
16//!
17//! 每条记录带长度和 CRC32。重新打开时逐条校验，遇到写了一半的尾部就截断到最后一条完整记录。
18//! 何时 fsync 由 [`SyncPolicy`] 决定。同一个目录同一时间只能被一个进程打开。
19
20mod cache;
21mod engine;
22mod index;
23mod items;
24mod record;
25mod snapshot;
26mod tasks;
27#[cfg(test)]
28mod tests;
29pub mod warc;
30
31use std::fmt;
32use std::io;
33use std::path::PathBuf;
34use std::time::{Duration, SystemTime, UNIX_EPOCH};
35
36pub use cache::{Cache, CacheEntry, CacheLookup, CachePolicy};
37pub use engine::Store;
38pub use items::Items;
39pub use snapshot::{ContentHash, SnapshotMeta, SnapshotPut, Snapshots};
40pub use tasks::Tasks;
41pub use warc::{
42    AssetSniffer, RecordType, SubresourceAsset, WarcDigests, WarcRecord, WarcWriter,
43    WebpageSnapshot,
44};
45
46/// 写入后何时调用 fsync。
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum SyncPolicy {
49    /// 每次写入都 fsync，最安全也最慢。
50    Always,
51    /// 距上次 fsync 超过这个间隔时，在下一次写入后 fsync。进程崩溃不丢数据（已写入内核），
52    /// 断电最多丢这个间隔内的写入。
53    Interval(Duration),
54    /// 交给操作系统。
55    Never,
56}
57
58/// 打开存储的配置，用 [`Config::new`] 创建后按需修改字段。
59#[derive(Debug, Clone)]
60#[non_exhaustive]
61pub struct Config {
62    pub dir: PathBuf,
63    pub sync: SyncPolicy,
64    /// 单个段文件的大小上限，超过就换新段。不能超过 2GiB。
65    pub max_segment_bytes: u64,
66    /// 存储空间上限（F-8.4）。超过后按 LRU 淘汰缓存和快照内容，任务状态不会被淘汰。
67    pub max_bytes: Option<u64>,
68    /// 淘汰到上限的这个比例为止。
69    pub low_watermark: f64,
70    /// 无效数据（被覆盖、删除）占比达到这个值时后台压缩。
71    pub compact_ratio: f64,
72    /// 总大小低于这个值时不自动压缩。
73    pub compact_min_bytes: u64,
74    /// 是否在后台线程自动压缩和淘汰；关掉后需要自己调用 [`Store::gc`]、[`Store::compact`]。
75    pub background: bool,
76    /// 单个快照（压缩前）的大小上限。
77    pub snapshot_max_bytes: usize,
78    pub snapshot_zstd_level: i32,
79    /// 测试用：把哈希截短以制造冲突。
80    pub(crate) hash_mask: u64,
81}
82
83impl Config {
84    pub fn new(dir: impl Into<PathBuf>) -> Self {
85        Self {
86            dir: dir.into(),
87            sync: SyncPolicy::Interval(Duration::from_secs(1)),
88            max_segment_bytes: 64 << 20,
89            max_bytes: None,
90            low_watermark: 0.9,
91            compact_ratio: 0.5,
92            compact_min_bytes: 16 << 20,
93            background: true,
94            snapshot_max_bytes: 512 << 10,
95            snapshot_zstd_level: 3,
96            hash_mask: u64::MAX,
97        }
98    }
99}
100
101#[derive(Debug)]
102#[non_exhaustive]
103pub enum Error {
104    Io(io::Error),
105    /// 段文件不是本存储的格式。
106    Corrupt(String),
107    /// 键超过 65535 字节。
108    KeyTooLarge(usize),
109    /// 值或快照超过上限。
110    ValueTooLarge {
111        size: usize,
112        limit: usize,
113    },
114}
115
116impl Error {
117    /// 目录正被另一个进程（或同进程里另一个打开的实例）使用。
118    pub fn is_locked(&self) -> bool {
119        matches!(self, Error::Io(e) if e.kind() == io::ErrorKind::WouldBlock)
120    }
121}
122
123impl fmt::Display for Error {
124    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
125        match self {
126            Error::Io(e) => write!(f, "store io error: {e}"),
127            Error::Corrupt(m) => write!(f, "store corrupt: {m}"),
128            Error::KeyTooLarge(n) => write!(f, "key too large: {n} bytes"),
129            Error::ValueTooLarge { size, limit } => {
130                write!(f, "value too large: {size} bytes (limit {limit})")
131            }
132        }
133    }
134}
135
136impl std::error::Error for Error {
137    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
138        match self {
139            Error::Io(e) => Some(e),
140            _ => None,
141        }
142    }
143}
144
145impl From<io::Error> for Error {
146    fn from(e: io::Error) -> Self {
147        Error::Io(e)
148    }
149}
150
151pub type Result<T, E = Error> = std::result::Result<T, E>;
152
153/// 打开时重建索引的情况。
154#[derive(Debug, Clone, Default)]
155#[non_exhaustive]
156pub struct OpenReport {
157    pub segments: usize,
158    pub records: u64,
159    /// 因为写了一半或校验失败被截掉的字节数。
160    pub truncated_bytes: u64,
161    pub elapsed: Duration,
162}
163
164#[derive(Debug, Clone, Default)]
165#[non_exhaustive]
166pub struct Stats {
167    pub entries: u64,
168    /// 索引堆内存估算值（字节）。
169    pub index_bytes: u64,
170    pub disk_bytes: u64,
171    pub live_bytes: u64,
172    pub segments: usize,
173    pub open: OpenReport,
174    /// 后台维护最近一次失败的原因。
175    pub last_error: Option<String>,
176}
177
178#[derive(Debug, Clone, Default)]
179#[non_exhaustive]
180pub struct GcReport {
181    pub evicted_entries: u64,
182    pub evicted_bytes: u64,
183    pub compacted_segments: usize,
184    pub disk_bytes_before: u64,
185    pub disk_bytes_after: u64,
186}
187
188pub(crate) fn to_ms(t: SystemTime) -> u64 {
189    t.duration_since(UNIX_EPOCH)
190        .map_or(0, |d| d.as_millis() as u64)
191}
192
193pub(crate) fn from_ms(ms: u64) -> SystemTime {
194    UNIX_EPOCH + Duration::from_millis(ms)
195}
196
197pub(crate) fn now_secs() -> u32 {
198    (to_ms(SystemTime::now()) / 1000) as u32
199}
pagesift_store/lib.rs

pagesift_store/
lib.rs
