行动日志
为Node.js所嵌入的事件源文档存储. 零原属地.
每一次突变都被记录为一个仅附加日志中的操作. 当前状态是通过从最新的快照中重放操作而得到的. 你得到崩溃安全,撤消,审计线索, 和同步准备,没有数据库服务器.
安装
npm install @backloghq/opslog
使用量
import { Store } from "@backloghq/opslog";
const store = new Store<{ name: string; status: string }>();
await store.open("./data");
// Create
await store.set("task-1", { name: "Build API", status: "active" });
// Read
const task = store.get("task-1");
// Update
await store.set("task-1", { ...task, status: "done" });
// Delete
await store.delete("task-1");
// Undo the delete
await store.undo();
// Query
const active = store.filter((r) => r.status === "active");
const count = store.count((r) => r.status === "done");
// Batch (single disk write for multiple operations)
await store.batch(() => {
store.set("a", { name: "A", status: "active" });
store.set("b", { name: "B", status: "active" });
store.set("c", { name: "C", status: "active" });
});
// Close (checkpoints automatically)
await store.close();
国家幸存下来的重启——重新打开同一个目录和所有东西.
如何运作
data/
manifest.json # Points to current snapshot + ops file(s)
snapshots/
snap-<timestamp>.jsonl # Immutable full-state capture (JSONL: header + one line per record)
ops/
ops-<timestamp>.jsonl # Append-only operation log (single-writer)
agent-<id>-<timestamp>.jsonl # Per-agent operation log (multi-writer)
archive/
archive-<period>.json # Old records, lazy-loaded
写作 将操作( 1 JSON 线) 附加到操作文件 . 读取 来源于由最新快照 + ops 重放所构建的内存地图. 检查站 实现当前状态为新的不可改变的快照.
每个操作记录上一个值,所以 撤销 弹出上一个操作并恢复旧状态. 操作日志作为 审计线索 和自然单位 同步 在节点之间.
API 密码
生命周期
await store.open(dir, options?) // Load state from directory
await store.close() // Checkpoint and close
拥挤
store.get(id) // Get record by ID
await store.set(id, value) // Create or update
await store.delete(id) // Remove (throws if not found)
store.has(id) // Check existence
查询
store.all() // All records
store.entries() // All [id, record] pairs
store.filter(predicate) // Records matching predicate
store.count(predicate?) // Count (all or matching)
store.getManifest() // Read-only ManifestInfo (snapshot/WAL paths, stats)
(供外部消费者使用)
// Stream snapshot records without loading all into memory
for await (const [id, record] of store.streamSnapshot()) {
process(id, record);
}
// Read WAL operations (optionally since a timestamp)
for await (const op of store.getWalOps(sinceTimestamp?)) {
// op: { ts, op: "set"|"delete", id, data?, prev }
}
批次
await store.batch(() => { // Multiple ops, single disk write
store.set("a", valueA); // Rolls back all on error
store.set("b", valueB);
});
空批次( 无) set编号 :delete ——没有执行I/O.
历史
await store.undo() // Undo last operation
store.getHistory(id) // All operations for a record
store.getOps(since?) // Operations since timestamp
维修
await store.compact() // Create new snapshot, clear ops
await store.archive(predicate) // Move matching records to archive
await store.loadArchive(segment) // Lazy-load archived records
store.listArchiveSegments() // List available archive files
store.stats() // { activeRecords, opsCount, archiveSegments }
await store.refresh() // Reload from all agent WALs (multi-writer only)
Blob 存储( 通过存储后端)
await backend.writeBlob("data/file.jsonl", buffer); // Write at relative path
const buf = await backend.readBlob("data/file.jsonl"); // Read full file
const range = await backend.readBlobRange("data/file.jsonl", 1024, 256); // Byte-range read
const names = await backend.listBlobs("data"); // List files under prefix
await backend.deleteBlob("data/file.jsonl"); // Delete file
readBlobRange 可以在 JSONL 唱片商店中实现 O(1) 点查—— 寻求字节相抵, 读取精确长度 。 FsBackend 用途 fs.read 页:1 S3Backend使用HTTP Range头.
选项
await store.open(dir, {
checkpointThreshold: 100, // Auto-checkpoint after N ops (default: 100)
checkpointOnClose: true, // Checkpoint when close() is called (default: true)
version: 1, // Schema version
migrate: (record, fromVersion) => record, // Migration function
readOnly: false, // Open in read-only mode (default: false)
skipLoad: false, // Skip loading snapshot/WAL into memory (default: false)
writeMode: "immediate", // "immediate" (default), "group" (~12x faster), or "async" (~50x faster, lossy on crash)
groupCommitSize: 50, // Group: flush after N ops (default: 50)
groupCommitMs: 100, // Group: flush after N ms (default: 100)
agentId: "agent-A", // Enable multi-writer mode (optional)
backend: new FsBackend(), // Custom storage backend (optional, default: FsBackend)
});
组提交
缓冲写入内存并冲出为单盘写出. 对持续写作来说.
const store = new Store();
await store.open("./data", {
writeMode: "group", // Buffer ops, flush periodically
groupCommitSize: 50, // Flush when buffer has 50 ops
groupCommitMs: 100, // Or after 100ms idle
});
// Writes are buffered — no fsync per op
await store.set("a", valueA);
await store.set("b", valueB); // Both flushed together
// Explicit flush if needed
await store.flush();
// close() always flushes before shutting down
await store.close();
同步模式
对于最大写入吞吐量, 使用 writeMode: "async"。在缓冲内存后立即写入解析——热路没有磁盘 I/O. ~50x比即时模式快.
const store = new Store();
await store.open("./data", {
writeMode: "async",
groupCommitSize: 50,
groupCommitMs: 100,
});
await store.set("a", valueA); // Returns instantly — buffered in memory
await store.set("b", valueB); // Same
// Ensure durability before exit
await store.sync();
await store.close();
崩溃语义 : 数据缓冲自上次冲出是 丢失 不干净的关闭。 打电话 sync() 在进程退出后可持久. close() 总是冲来冲去.
安全性: 强迫 "immediate" 何时 agentId 设置(多写器模式)。 其他特工无法看到缓冲行动,因此团体/同步承诺只是单写.
何时使用哪种模式 :
"immediate"(默认)——每作都耐用. 当数据丢失是不可接受的时使用 ."group"- 写作是批量的,但呼叫者仍在等待冲浪。 ~12x快一点. 崩溃损失最多groupCommitMs数据数."async"- 马上回信 ~50x快点. 崩溃失去所有未流出的数据 。 最适合高通量、耐久性、耐撞性的工作.
多写器模式
多个特工可以同时给同一家商店写信. 每个代理都得到自己的WAL文件——没有写争议.
// Agent A (process 1 / machine 1)
const storeA = new Store<Task>();
await storeA.open("./data", { agentId: "agent-A" });
await storeA.set("task-1", { title: "Build API", status: "active" });
await storeA.close();
// Agent B (process 2 / machine 2)
const storeB = new Store<Task>();
await storeB.open("./data", { agentId: "agent-B" });
// B sees A's writes on open
storeB.get("task-1"); // { title: "Build API", status: "active" }
await storeB.set("task-2", { title: "Write tests", status: "active" });
await storeB.close();
如何运作
- 每个代理都给
ops/agent-{id}-{timestamp}.jsonl——单独文件,写入不需要锁定 - 业务携带a 兰波特钟 用于订购
- 打开
open(),所有代理 WAL 文件由(clock, agentId)确定总顺序 - 冲突(两个代理写同样的密钥)通过 最后写手胜出 按时钟值
undo()只取消调用代理的最后一次操作compact()获取压缩锁, 抓取合并状态, 并重置所有 WAL 文件refresh()重新阅读所有代理 WALs 来取取其他代理的写作
解决冲突
当两个特工修改了同一个密钥时,使用更高一等的Lamport时钟的操作获胜. 如果时钟相等,则词典上更高的代理ID获胜. 这是决定性的——所有特工都到达同一状态,无论重播命令如何.
// Agent A sets "shared" (clock=1)
await storeA.set("shared", { value: "from-A" });
// Agent B opens (sees clock=1), sets "shared" (clock=2)
await storeB.set("shared", { value: "from-B" });
// B wins — higher clock
store.get("shared"); // { value: "from-B" }
自定义存储后端
opslog 使用可插件 StorageBackend 用于所有 I/O 的接口。 默认是 FsBackend (本地文件系统). 您可以为 S3 、 数据库或其他存储系统执行自己的后端.
import { Store, FsBackend } from "@backloghq/opslog";
import type { StorageBackend } from "@backloghq/opslog";
// Use the default filesystem backend (implicit)
const store = new Store();
await store.open("./data");
// Or pass a custom backend explicitly
const store = new Store();
await store.open("./data", { backend: new FsBackend() });
// Or implement your own
class S3Backend implements StorageBackend {
// ... implement all methods
}
const store = new Store();
await store.open("s3://bucket/prefix", { backend: new S3Backend() });
仅读模式
在不获取写锁的情况下打开商店阅读. 可用于仪表板,备份进程,或多个阅读器与单个写作机并列.
const reader = new Store();
await reader.open("./data", { readOnly: true });
// All reads work
const tasks = reader.all();
const active = reader.filter((t) => t.status === "active");
// All mutations throw
await reader.set("x", value); // Error: Store is read-only
只读存储在打开时加载最新快照并重放操作. 他们不关闭检查站。 多个只读商店可以同时打开同一个目录和一个写手同时打开.
货币
所有州级业务(set, (中文(简体) ), delete, (中文(简体) ), batch, (中文(简体) ), undo, (中文(简体) ), compact, (中文(简体) ), archive)通过内部活性突变被序列化. 这样可以防止同时发生的突变——例如, compact() 交换操作文件时 set() 正在附加,或 undo() 切换时 set() 正在写作.
读取操作( E)get, (中文(简体) ), all, (中文(简体) ), filter, (中文(简体) ), count, (中文(简体) ), has, (中文(简体) ), entries)是同步和无锁的.
咨询目录写入锁( N).lock 有 PID 的文件)防止两个进程打开同一商店。 从已坠毁的流程中自动取出 Stale 锁.
崩溃安全
- 操作文件: 只写附件 。 坠机中继器最多输掉最后一次操作. 恢复时跳过错误的行 .
- 抓图:不可改变. 写入临时文件,然后在解剖学上重新命名.
- 声明:通过临时文件重命名进行解剖替换. 总是指向有效的快照 .
- 撤销: 使用
ftruncate()单原子 POSIX syscall. O(1) 无论文件大小.
没有崩溃的数据腐败 。 最多一次飞行行动失败.
Schema 迁移
const store = new Store();
await store.open(dir, {
version: 2,
migrate: (record, fromVersion) => {
if (fromVersion < 2) return { ...record, newField: "default" };
return record;
},
});
记录在打开时移入内存。 下一个检查站继续维持迁移状态.
发展
npm run build # Compile TypeScript
npm run lint # ESLint
npm test # Run tests
npm run test:coverage # Tests with coverage
许可证
麻省理工学院