周末关闭的显示屏与消失的零食日志

目标

恢复周末期间关闭的零食库存电子屏。如果日志的保留范围已经结束，就安装同一时点的快照，并从下一个事件开始追赶。

为什么重要

即使连接恢复了，已经删除的事件也不会回来。如果库存与游标指向不同的时点，即使重放成功，也会显示错误的数字。这次要亲自实现保留边界、代次、同一时点读取和原子安装，并通过真实终止之后的磁盘状态来确认。

预计需要 100 分钟。比默认会话更长，所以请在到期之前点击“＋时间”按钮延长。会话结束后文件会消失，需要的代码请另行保存。开始前应已了解前面模块中的 Python 异常、SQLite 事务、游标，以及 outbox 恢复。

数据契约

交付物是 /root/snapshot/worker.py 这一个文件。源和副本是具有相同的下列 schema、相互独立的一次性本地 SQLite 文件。它们由检查器创建并清理，所以不要把 DB 路径写死。所有文件都是 schema 正常的可信本地文件，借给函数的 con 在开始时没有未结束的事务。除 open_store 和 sync_files 之外的函数不会关闭借来的连接，且无论成功还是失败，都不留下事务。

CREATE TABLE checkpoint (id INTEGER PRIMARY KEY CHECK(id=1),
    epoch TEXT NOT NULL, last INTEGER NOT NULL, floor INTEGER NOT NULL);
CREATE TABLE stock (id INTEGER PRIMARY KEY CHECK(id=1), total INTEGER NOT NULL);
CREATE TABLE events (seq INTEGER PRIMARY KEY, delta INTEGER NOT NULL);

checkpoint 的初始行是 (1, 传入的 epoch, -1, 0)，stock 的初始行是 (1,0)。在同一个代次内不复用编号，retained events 从 floor 到 last 连续。全部被截断时 floor=last+1。total 是从初始 0 开始、累加各个 delta 得到的业务状态。expected_epoch 取自已信任的连接配置，不会从 snapshot 本身提取并当作期望值。

实验中的写入是依次逐个执行的。检查时点隔离的方式，是在读事务期间让其他连接提交写入。这并不是使用单独线程的并发 checkpoint，也不是多 writer 的性能测试。对所有连接都设置 wal_autocheckpoint=0，并在写入结束之后依次关闭连接。不要把它原样用作长期运行的 WAL 管理策略。

步骤

保存代次与恢复状态——在 worker.py 中定义 Exception 的子类 Conflict、ResyncRequired、Gap、CoveredBySnapshot，并实现 validate_snapshot(snapshot) 和 open_store(path,epoch)。快照是只含 version、epoch、last、total 的 dict。version 是严格的 int 1，epoch 是由 ASCII 字母、数字、下划线、连字符组成的 1–64 个字符，last 是不含 bool 的 int，范围 -1 到 2147483647，total 是不含 bool 的 int，其绝对值不超过 1000*(last+1)。有效时返回新的 dict 副本，否则为 ValueError。open_store 在验证 epoch 之后，仅在不存在时以原子方式创建下面的三张表和初始行。以 isolation_level=None、timeout=1、WAL、synchronous=FULL、wal_autocheckpoint=0 打开，并返回 sqlite3.Connection。不覆盖已有内容，初始化失败时关闭连接。
同时确定库存与下一个编号——append(con,delta,fault=None) 追加一个不含 bool、范围 -1000 到 1000 的 int 变化。在 BEGIN IMMEDIATE 内检查下一个 seq=last+1，并依次进行：events 插入 → fault('after-event') → 把 delta 加到 stock.total 并更新 checkpoint.last → fault('after-state') → COMMIT → fault('after-commit')。seq 范围是 0 到 2147483647，超出范围为 ValueError。fault 存在时才调用，并返回 seq。提交之前的错误整体回滚，提交之后的错误保留已确定的状态，并传递原始错误。
导出同一时点的快照——export_snapshot(con,between=None) 在 BEGIN 读事务中依次执行：查询 epoch 和 last → between() → 查询 total → COMMIT，并返回 version=1 的快照。钩子只在存在时调用一次，且必须允许其他连接的写入在这里提交。失败时清理读事务，并传递原始错误。不要在两个 SELECT 之间使用 COMMIT 或 BEGIN IMMEDIATE。
比较保留范围与重放游标——trim(con,through) 接收不含 bool、范围 -1 到 2147483647 的 int，并在一个写事务中检查 floor-1 <= through <= last。否则为 ValueError，符合则只删除 seq<=through 的事件并更新 floor=through+1，返回 None。replay(con,epoch,last,limit=16) 会验证 epoch 格式、不含 bool 且范围 -1 到 2147483647 的 last，以及 1 到 16 的 limit。在同一个读事务中，其他代次为 ResyncRequired，比服务器更靠前的游标为 ValueError，lastlast 的 (seq,delta) tuple 构成的 list，最多 limit 个。要区分空结果与遗漏错误。
一次性安装快照——install_snapshot(con,snapshot,expected_epoch,fault=None) 会验证 snapshot 与 expected_epoch 的格式。snapshot.epoch 与期望代次不同则为 ResyncRequired。在一个写事务中，同一代次中更小的 last，或相同 last 但不同 total，为 Conflict；相同 last 和 total 则不做修改并返回 False。除此之外，依次执行：删除全部 events → fault('after-clear') → 替换 total 并更新 checkpoint 的 epoch、last、floor=last+1 → fault('after-install') → COMMIT → fault('after-commit')，然后返回 True。提交之前的错误整体回滚，提交之后的错误保留已确定的状态。所允许的其他代次，即使编号更小也可以替换。
应用可验证的重复与新批次——apply_batch(con,epoch,events,fault=None) 先验证 epoch 和 events。events 是最多 16 个的 list，空 list 也允许。元素是由 seq、delta 两个值构成的 tuple 或 list，seq 是不含 bool、范围 0 到 2147483647 的 int，delta 是不含 bool、范围 -1000 到 1000 的 int。格式错误为 ValueError，批次内部的编号不是逐个加 1 时为 Gap。在一个写事务中，代次不一致为 ResyncRequired。seq<=last 时，如果保留的 delta 相同则无效果，delta 不同为 Conflict，没有单个明细则为 CoveredBySnapshot。新的 seq 只允许 last+1，否则为 Gap。每次更新新事件、库存和游标时调用 fault('after-one')，整体 COMMIT 之后调用 fault('after-commit')。返回本次新效果的个数，并且不修改输入。提交之前的错误使整个批次回滚，之后的错误保留已确定的状态。
让一次恢复尝试有限地结束——sync_once(source,replica,expected_epoch,after_snapshot=None) 会验证期望代次，如果与 source 的实际代次不同，则抛出 ResyncRequired。用 replica 的 epoch 和 last 请求一次 replay。只有在出现 ResyncRequired 时，才通过 export_snapshot(source) → install_snapshot(replica,...,expected_epoch) → after_snapshot() → 快照 last 之后的 replay 来恢复。钩子只在存在时调用。批次最多 16 个，在 apply_batch 之后返回 export_snapshot(replica)。如果第二次 replay 再次被截断，就原样传递错误，并保留已安装的有效状态。其他错误也要传递，并且内部不做无限重试。
在真实终止之后，从独立文件恢复——sync_files(source_path,replica_path,expected_epoch) 会验证期望代次，如果本地源文件不存在则为 FileNotFoundError。如果 realpath 相同，或已存在的两个路径是 samefile，则在打开之前就是 ValueError。用 open_store 打开相互独立的源和副本，执行一次 sync_once 并返回其结果，并在所有路径上关闭自己拥有的连接。即使打不开副本，也要关闭源。检查器会在 append、install、batch 的提交前后，让真实的子进程以退出码 73 结束，并用新连接检查磁盘状态。

参考

分步骤诊断：python3 -B /opt/fixtures/snapshot/check.py 8 /root/snapshot/worker.py。把 8 改成当前步骤，就会检查到该步骤。评分上限为 12 秒，不会修改提交的文件。
只使用标准库和本地文件。不需要互联网安装、外部 DB 或额外的 capability。检查数据很小且有限。
检查真实的子进程终止，以及在同一块磁盘上的恢复。断电、磁盘丢失、恶意替换文件的竞争、服务器之间的共识、TLS 和用户权限，不在本实验的验证范围内。
SQLite 版本及 WAL-reset 的修复情况，请结合官方文档和发行版的补丁历史一起确认。不要把通过这个单写入实验，当作多写入生产环境安全性的保证。

保存代次与恢复状态

在 worker.py 中定义 Exception 的子类 Conflict、ResyncRequired、Gap、CoveredBySnapshot，并实现 validate_snapshot(snapshot) 和 open_store(path,epoch)。快照是只含 version、epoch、last、total 的 dict。version 是严格的 int 1，epoch 是由 ASCII 字母、数字、下划线、连字符组成的 1–64 个字符，last 是不含 bool 的 int，范围 -1 到 2147483647，total 是不含 bool 的 int，其绝对值不超过 1000*(last+1)。有效时返回新的 dict 副本，否则为 ValueError。open_store 在验证 epoch 之后，仅在不存在时以原子方式创建下面的三张表和初始行。以 isolation_level=None、timeout=1、WAL、synchronous=FULL、wal_autocheckpoint=0 打开，并返回 sqlite3.Connection。不覆盖已有内容，初始化失败时关闭连接。

last=-1 时 total 只能是 0。要把 CREATE IF NOT EXISTS 与初始行插入分开，并且在所有 int 契约中排除 bool。

同时确定库存与下一个编号

append(con,delta,fault=None) 追加一个不含 bool、范围 -1000 到 1000 的 int 变化。在 BEGIN IMMEDIATE 内检查下一个 seq=last+1，并依次进行：events 插入 → fault('after-event') → 把 delta 加到 stock.total 并更新 checkpoint.last → fault('after-state') → COMMIT → fault('after-commit')。seq 范围是 0 到 2147483647，超出范围为 ValueError。fault 存在时才调用，并返回 seq。提交之前的错误整体回滚，提交之后的错误保留已确定的状态，并传递原始错误。

即使截断了日志，编号也是从 last 接着往下排。不要把 MAX(events.seq) 当作下一个编号的依据。

导出同一时点的快照

export_snapshot(con,between=None) 在 BEGIN 读事务中依次执行：查询 epoch 和 last → between() → 查询 total → COMMIT，并返回 version=1 的快照。钩子只在存在时调用一次，且必须允许其他连接的写入在这里提交。失败时清理读事务，并传递原始错误。不要在两个 SELECT 之间使用 COMMIT 或 BEGIN IMMEDIATE。

评分对象不是单个值，而是值的组合。在钩子内产生的新库存，应当在下一次 export 中能看到。

比较保留范围与重放游标

trim(con,through) 接收不含 bool、范围 -1 到 2147483647 的 int，并在一个写事务中检查 floor-1 <= through <= last。否则为 ValueError，符合则只删除 seq<=through 的事件并更新 floor=through+1，返回 None。replay(con,epoch,last,limit=16) 会验证 epoch 格式、不含 bool 且范围 -1 到 2147483647 的 last，以及 1 到 16 的 limit。在同一个读事务中，其他代次为 ResyncRequired，比服务器更靠前的游标为 ValueError，lastlast 的 (seq,delta) tuple 构成的 list，最多 limit 个。要区分空结果与遗漏错误。

即使全部删除之后，last 和 floor 也必须保留。不等号中的一个等号，就可能把已处理的事件再次包含进来。

一次性安装快照

install_snapshot(con,snapshot,expected_epoch,fault=None) 会验证 snapshot 与 expected_epoch 的格式。snapshot.epoch 与期望代次不同则为 ResyncRequired。在一个写事务中，同一代次中更小的 last，或相同 last 但不同 total，为 Conflict；相同 last 和 total 则不做修改并返回 False。除此之外，依次执行：删除全部 events → fault('after-clear') → 替换 total 并更新 checkpoint 的 epoch、last、floor=last+1 → fault('after-install') → COMMIT → fault('after-commit')，然后返回 True。提交之前的错误整体回滚，提交之后的错误保留已确定的状态。所允许的其他代次，即使编号更小也可以替换。

不要问快照自己是不是允许的代次。expected_epoch 是来自被验证对象之外的基准。

应用可验证的重复与新批次

apply_batch(con,epoch,events,fault=None) 先验证 epoch 和 events。events 是最多 16 个的 list，空 list 也允许。元素是由 seq、delta 两个值构成的 tuple 或 list，seq 是不含 bool、范围 0 到 2147483647 的 int，delta 是不含 bool、范围 -1000 到 1000 的 int。格式错误为 ValueError，批次内部的编号不是逐个加 1 时为 Gap。在一个写事务中，代次不一致为 ResyncRequired。seq<=last 时，如果保留的 delta 相同则无效果，delta 不同为 Conflict，没有单个明细则为 CoveredBySnapshot。新的 seq 只允许 last+1，否则为 Gap。每次更新新事件、库存和游标时调用 fault('after-one')，整体 COMMIT 之后调用 fault('after-commit')。返回本次新效果的个数，并且不修改输入。提交之前的错误使整个批次回滚，之后的错误保留已确定的状态。

在验证全部输入之后，以原子方式处理一个小批次。无法从快照的总和反推出单个过去的 delta。

让一次恢复尝试有限地结束

sync_once(source,replica,expected_epoch,after_snapshot=None) 会验证期望代次，如果与 source 的实际代次不同，则抛出 ResyncRequired。用 replica 的 epoch 和 last 请求一次 replay。只有在出现 ResyncRequired 时，才通过 export_snapshot(source) → install_snapshot(replica,...,expected_epoch) → after_snapshot() → 快照 last 之后的 replay 来恢复。钩子只在存在时调用。批次最多 16 个，在 apply_batch 之后返回 export_snapshot(replica)。如果第二次 replay 再次被截断，就原样传递错误，并保留已安装的有效状态。其他错误也要传递，并且内部不做无限重试。

第一次遗漏是选择恢复路径的信号，而恢复过程中的第二次遗漏则是这次尝试尚未结束的信号。

在真实终止之后，从独立文件恢复

sync_files(source_path,replica_path,expected_epoch) 会验证期望代次，如果本地源文件不存在则为 FileNotFoundError。如果 realpath 相同，或已存在的两个路径是 samefile，则在打开之前就是 ValueError。用 open_store 打开相互独立的源和副本，执行一次 sync_once 并返回其结果，并在所有路径上关闭自己拥有的连接。即使打不开副本，也要关闭源。检查器会在 append、install、batch 的提交前后，让真实的子进程以退出码 73 结束，并用新连接检查磁盘状态。

with sqlite3.Connection 管理的是事务，与关闭连接是两回事。要明确由谁拥有已打开的连接。