diff --git a/AGENTS.md b/AGENTS.md index 0fd75b6..51bfa26 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,7 +79,7 @@ ## 当前范围与权威入口 -- **SQLite cleanup v0.1 首期（开发分支，本地验收）**：`feat/sqlite-cleanup-v0.1` 仅调整完成数据的本地保留周期。源 `event_id`（含 numberID）原样回传，不新增 `execution_id`，SaaS 如何处理重复结果不在范围；共享合同仍固定 `ef3009d`，接口/类型/路由不变，无共享发布。Agent 不再读写永久 `.approved`；活动 TaskCalls 与 Dispatcher 未决/未知保护保留。已派发呼叫需终结/结果成功及派发回执、最终结果分别 MQ confirm；未发起拒绝只需拒绝确认；取消记录需 durable stopped 和成功 stop confirm。清理事务不删共享配置/任务停止状态，布局仍为 **4**，无迁移/重置。录音仅 delivered 且无未完成 `.results` 引用后删音频、信息、空叶，删除中断凭 delivered 依据续处理，不再 PUT/报告；拒绝原件按完整回执/原身份/内容精确关联，先全部原件后回执。新原件文件名为回执 SHA-256 与完整记录 SHA-256，仅磁盘关联，不是业务编号；既有完整文件校验读取、不迁移，不能关联的损坏旧文件和潜在依据保留待人工。两角色独立 Asia/Shanghai 每日 24:00 补扫复用实时资格，无 12:00/年龄删除/VACUUM/永久清理历史，清理错误不关闭业务准入。仅测试目录验证 `.approved` 退役，没有实际主机退役、部署、真实数据/OSS 删除、拨号、抓包、AI/OSS 调用或 main 合入/推送；本地通过不代表真实服务验收。检查和逐项证据见 [`docs/evidence/sqlite-cleanup-v0.1-local.md`](docs/evidence/sqlite-cleanup-v0.1-local.md)。此条是开发分支状态，不改变下述线上事实。 +- **SQLite cleanup v0.1 首期（开发分支，本地验收）**：`feat/sqlite-cleanup-v0.1` 仅调整完成数据的本地保留周期。源 `event_id`（含 numberID）原样回传，不新增 `execution_id`，SaaS 如何处理重复结果不在范围；用户在复审中批准仅必要一句文字对齐，共享合同 `7b4448f` 已发布并固定本项目引用，限定首次事实保护为本地执行/回执仍保留期间；接口/Schema/类型/路由不变，不涉及 SaaS 重复处理或等待适配。Agent 不再读写永久 `.approved`；活动 TaskCalls 与 Dispatcher 未决/未知保护保留。已派发呼叫需终结/结果成功及派发回执、最终结果分别 MQ confirm；未发起拒绝只需拒绝确认；取消记录需 durable stopped 和成功 stop confirm。清理事务不删共享配置/任务停止状态，布局仍为 **4**，无迁移/重置。录音仅 delivered 且无未完成 `.results` 引用后删音频、信息、空叶；删除前在私有 `.recording-cleanup` 保存原 delivered 信息临时副本，精确空叶删除并同步父目录后才删副本。info/空叶删除之间硬退出也可按该依据续处理，不泛扫未知空目录、不留永久完成历史；同 OSS 目标在副本清完前拒绝新恢复写入，不再 PUT/报告；拒绝原件按完整回执/原身份/内容精确关联，先全部原件后回执。新原件文件名为回执 SHA-256 与完整记录 SHA-256，仅磁盘关联，不是业务编号；既有完整文件校验读取、不迁移，不能关联的损坏旧文件和潜在依据保留待人工。两角色独立 Asia/Shanghai 每日 24:00 补扫复用实时资格，无 12:00/年龄删除/VACUUM/永久清理历史，清理错误不关闭业务准入。仅测试目录验证 `.approved` 退役，没有实际主机退役、部署、真实数据/OSS 删除、拨号、抓包、AI/OSS 调用或 main 合入/推送；本地通过不代表真实服务验收。检查和逐项证据见 [`docs/evidence/sqlite-cleanup-v0.1-local.md`](docs/evidence/sqlite-cleanup-v0.1-local.md)。此条是开发分支状态，不改变下述线上事实。 - **2026-10-10 event_id 修复获批部署与重置现场**：使用者在要求直接部署后最终同意必要私密备份与测试运行数据重置。修复 `f7aa766` 已合入并推送 main，安装至登记 `server.sip`，共享合同固定 `ef3009d`。先停 Dispatcher 关准入，核实零通道、无未完成/未知执行、无待确认反馈/Agent 恢复；MQ 控制及任务3–5队列为空/消费者0，任务1–2队列不存在，无发布/消费/ACK/purge。私密备份与原29文件 bytes/清单完全一致，恢复验证包含 WAL/SHM，保留原版本3、8条完成及25条确认事实；不改写旧结果。新运行库布局4、inbox/outbox0，不继承旧事件防重历史，不主动重投旧命令。实际独立复现二进制 SHA-256 `6886987795b208e79964053328262c450a3249ee3c3f7e9e75bc7ca1c10c0586`；Agent2910678 enabled+active、0重启、Debug开启/Capture关闭。Dispatcher enabled且实际auto-restart，最新结构化错误为SaaS HTTP502（初始独立探测404），`business_ready=false`，不把failed/数据库/证书/未知错误算等待成功。Asterisk777461保持不变、0通道；未重启、抓包、主动投递/试拨或额外AI/OSS，未清外部SaaS/MQ/OSS。程序安装与获批重置完成，但线上event_id反馈/真实通话/生产未验证。证据与源制品边界见 [`docs/evidence/event-id-lifecycle-20261010.md`](docs/evidence/event-id-lifecycle-20261010.md) 文末与 `.local/event-id-deployment/`。本条新授权与现场事实优先于下方本地未部署的历史阶段。 diff --git a/contracts/cleanup_scope_test.go b/contracts/cleanup_scope_test.go new file mode 100644 index 0000000..7b1f21e --- /dev/null +++ b/contracts/cleanup_scope_test.go @@ -0,0 +1,27 @@ +package contracts + +import ( + "os" + "strings" + "testing" +) + +func TestCleanupScopeRetainsIdentityProtectionUntilLocalCompletion(t *testing.T) { + data, err := os.ReadFile("schema/saas-dispatcher.md") + if err != nil { + t.Fatal(err) + } + text := string(data) + for _, required := range []string{ + "执行或回执仍被本地保留期间，同类型同身份只保留首次事实", + "完成并满足本地清理条件后，再收到同 `event_id` 的命令可按新执行处理", + "源 `event_id` 仍原样保留", + "本条不规定 SaaS 对重复反馈的处理", + "不拼接、清洗或哈希改号", + "`(dispatcher_id,event_type,event_id)`", + } { + if !strings.Contains(text, required) { + t.Fatalf("cleanup scope or identity protection missing: %q", required) + } + } +} diff --git a/contracts/schema b/contracts/schema index ef3009d..7b4448f 160000 --- a/contracts/schema +++ b/contracts/schema @@ -1 +1 @@ -Subproject commit ef3009df3a90f6e941ca0b24e46134aef17a374d +Subproject commit 7b4448f3586104cde91509a5d5951575c37a8b2c diff --git a/docs/evidence/sqlite-cleanup-v0.1-local.md b/docs/evidence/sqlite-cleanup-v0.1-local.md index d98c27a..6010e45 100644 --- a/docs/evidence/sqlite-cleanup-v0.1-local.md +++ b/docs/evidence/sqlite-cleanup-v0.1-local.md @@ -2,9 +2,9 @@ ## 范围与状态 -- 开发分支：`feat/sqlite-cleanup-v0.1`；首期本地开发及规定验收完成。 +- 开发分支：`feat/sqlite-cleanup-v0.1`；首期本地开发及复审根因修复完成；修复后完整 `make check`、覆盖率和Python测试通过，最终干净制品检查记录见文末。 - 用户确认：仅本项目内部数据保留周期调整，不新增 `execution_id`，源 `event_id`（包括 numberID）原样回传；SaaS 如何处理再次返回的同 ID 结果不在范围内。 -- 共享合同固定 `ef3009d`，`make contracts-update` 的 43 个 JSON 检查与 8 项合同测试通过；接口字段、消息类型、路由、共享文件及 submodule 指针不变，没有共享发布。 +- 初轮固定 `ef3009d`。复审后用户明确批准仅必要一句文字及共享发布，`7b4448f3586104cde91509a5d5951575c37a8b2c` 已发布并固定本项目引用，43个JSON和8项合同检查通过；Schema、接口字段、消息类型、路由不变。 - 没有合入或推送本项目 main、部署、操作现存运行数据库/恢复目录、退役主机 `.approved`、拨号、抓包或调用真实 AI/OSS。 ## 已完成的分批验证 @@ -32,7 +32,7 @@ - 先添加反例：原逻辑不能读取音频已删但 `info.json` 仍为 delivered 的合法中间状态，且不会删除已交付文件；实现后通过。 - 仅持久 delivered、上传成功回执和原身份/长度/校验和等信息有效、且原事件无未完成 `.results` 引用时删除。prepared、ready、end_pending 引用都保留原恢复文件，引用完成后可以继续收尾。 -- 依次删除本条 `recording.wav`、同步目录、删除 `info.json`、同步并删除准确且空的叶目录；共享父目录、`.results`、未知侧文件与 OSS 对象不删除。信息或叶目录删除失败保留/恢复 delivered 依据，下次启动只补删除，不申请授权、PUT 或重报。 +- 删除前先把原 delivered 信息原样持久保存至私有 `.recording-cleanup` 临时副本；依次删除本条 `recording.wav`、同步目录、删除 `info.json`、同步并删除准确且空的叶目录，同步父目录后才删副本。共享父目录、`.results`、未知侧文件/空目录与 OSS 对象不删除；返回错误及硬退出均保留原信息或副本，下次启动只补删除，不申请授权、PUT 或重报。 - delivered 且原音频已经缺失时，只放行这一已确认删除的中间状态；其他状态缺音频仍报错。原校验和严格保持有效的小写 SHA-256，上传回执、长度和身份校验不弱化。 - 同一录音跨不同 recovery 实例使用临时按路径互斥；结果引用写入/确认与删除协调，锁使用结束即移除，不形成永久身份历史。并发实例仍只有一次 PUT、一次回报；完成后不存在该记录，后续直接调用明确返回不存在而非虚构已完成历史。 - 既有恢复扫描能续处理 delivered 中间状态，避免清理后同次遍历把已删除音频误报成孤立文件；单条坏目录不妨碍其他合法记录。没有新增十秒全目录清理任务，既有恢复调度保持原用途。 @@ -56,16 +56,34 @@ - UTC/上海转换、12:00 不清理、午夜启动安排次日、时钟回退、失败等待次日、取消计时器，以及 205 条跨页候选、未确认/待拨邻居保留、取消后续处理、只清 delivered 不重试其他录音验证通过。 - `go test -race ./internal/daily ./internal/store ./internal/dispatcher ./internal/agent ./cmd/sip-go-agent` 通过。 -## 最终验收 +## 初轮固定提交验证（复审修复前的历史记录） - `make check`：格式、合同固定引用/43 JSON及8合同测试、Proto来源、`go vet ./...`、`go test -race ./...`、二进制构建及隔离 MQ 检查通过。 -- `make release-check-local`：在干净固定代码提交 `afa4fe2` 重新通过，`source dirty=False`、`production_approval=False`；制品 SHA-256 为 `e03c42d8f6b7b2cfd6c95e810c4ec1461b2f06d515b1f53c17719a1c7b9978ac`，合同固定 `ef3009df3a90f6e941ca0b24e46134aef17a374d`。同提交 `make check` 和 `make coverage` 重新通过，业务覆盖率72.9%。本文后续只补证据记录，不把最初 dirty 检查当作固定提交制品，也不将本地制品批准为生产。 +- `make release-check-local`：在干净固定代码提交 `afa4fe2` 重新通过，`source dirty=False`、`production_approval=False`；制品 SHA-256 为 `e03c42d8f6b7b2cfd6c95e810c4ec1461b2f06d515b1f53c17719a1c7b9978ac`，合同固定 `ef3009df3a90f6e941ca0b24e46134aef17a374d`。同提交 `make check` 和 `make coverage` 重新通过，业务覆盖率72.9%。此制品及覆盖率属于复审修复前，不能作为修复后最终代码的制品证据，也不将本地制品批准为生产。 - 实际运行的隔离 RabbitMQ + HTTPS + gRPC mTLS 端到端通过。更新旧“完成后永不重用”断言：收到原 ID 最终结果、两个消息类型分别确认、本地清理后，再通过 MQ 投递同一源 ID，Agent 再次接纳且 ID 不变。没有以孤立单元测试代替 MQ/HTTPS/mTLS 运行事实。 - 业务覆盖率 **72.9%**（新增纯测试后复核），高于 65%；Python生产部署 **26** 项、脚本 **9** 项通过。 - SQLite增长：4轮各40通、同源 numberID 重用及重启，完成后 inbox/outbox/recordings 都为0，共享task/config仍各1；预热后 page_count 不继续增长，不执行 VACUUM、不要求数据库文件立即缩小。 - 录音/拒绝文件增长：各3轮×20条重复身份，完成后原件/回执或录音叶目录不存在；保留共享目录，不留 `.approved` 或内存永久完成身份历史。 - 没有修改部署主机或现有运行数据库/文件，未 main 合入/推送、部署、重置、迁移、拨号、抓包、实际 AI/OSS 调用或 OSS 对象删除。SaaS收到、判重、覆盖和真实通话/生产验收不在本次范围。 +## 复审根因修复 + +- 独立审核发现初轮 info 删除后、空叶删除前的硬退出窗口没有依据；不是普通删除返回错误。新增子进程 `os.Exit(23)` 先复现：info 删除后空叶仍留存。修复后音频、info、叶、副本四个实际删除边界 × 启动/午夜两入口均通过；无 defer、恢复信息回写或有序退出掩盖窗口。 +- `.recording-cleanup/<原bucket/objectKey路径SHA>.json` 仅是现有 delivered 信息的私有临时副本，不是外部 ID、二次确认或永久完成历史。副本在删除任一原文件前持久保存；空叶 durably gone 后才删副本。没有根据前缀或“空目录”推测完成，读取原 delivered 身份/上传回执/长度/校验和并校验精确副本路径。 +- 副本缺失/损坏/身份冲突/错名/非delivered/假上传回执不能清原件；原副本持久化失败时音频/info都保留。info已删只允许从有效副本续处理，且音频必须已不存在；其他缺失/损坏仍报错。 +- 空叶已删、副本删除失败时仍保留依据；同OSS目标新恢复写入明确拒绝，避免旧副本误删新记录。独立合法记录照常清理，下一实例重试最终只删副本，目标随后可复用；副本清完不留完成文件。 +- `go test -race ./contracts ./internal/agent` 通过。`contracts/cleanup_scope_test.go` 固定文字测试保留原值/tuple/尚未清理的防重规则；机器Schema无改动。 + +### 修复后全量检查 + +- `make check` 重新通过，含格式、固定合同43 JSON及8测试、Proto、vet、全量race、构建和实际隔离RabbitMQ/HTTPS/mTLS；新合同范围回归与四个删除边界硬退出的启动/午夜续收尾均包含在全量测试中。 +- `make coverage`：业务覆盖率 **73.0%**。生产Python **26** 项、脚本Python **9** 项均重新通过。日志 `/tmp/go-sip-cleanup-audit-final-{check,coverage,python,script-tests}.log`。 +- 两项审核缺口均有对应根因修复及测试，不以普通返回错误模拟硬退出，不忽略共享首次事实的必要文字。 + +### SaaS开发同步说明（不等待外部适配） + +共享版本：`7b4448f3586104cde91509a5d5951575c37a8b2c`。仅把“同类型同身份保留首次事实”明确为 Dispatcher 本地执行/回执仍保留期间；完成且按条件清理后，同源 `event_id` 可再次执行，回执/结果仍原值，不增加 `execution_id`。接口字段、Schema、类型、路由不变。此项不规定 SaaS 重复结果接收/判重/覆盖/区分规则，也不以 SaaS 处理或验收为开发条件；本说明供开发同步，不声称对方已收到、正确处理或完成适配。 + ### 计划验收逐项对应 | 项目 | 本地证据 | @@ -82,7 +100,7 @@ | 两阶段删除失败/并发新捕获不会提前清回执 | 原件半途失败、回执SQL触发器失败、capture-race及独立坏组回归 | | .results正常成功收尾、SIP来源、既有临时行为不退化 | 全量agent/result_journal、store/sip_source及cmd/rpc回归；未知.pending保持 | | 实时与午夜同条件；0点非12点；故障/取消不关准入 | daily/midnight_test.go、dispatcher/cleanup_scan_test.go、Agent delivered-only回归 | -| 同源ID原值重用，不新增字段，不验证SaaS重复处理 | controller单元与隔离MQ最终结果/再次执行；Schema/Proto/submodule均未改 | +| 同源ID原值重用，不新增字段，不验证SaaS重复处理 | controller单元与隔离MQ最终结果/再次执行；Schema/Proto均未改，submodule仅获批文字更新 | ### A01–A12 / K01–K16 范围核对 diff --git a/docs/plan-sqlite-cleanup-v0.1.md b/docs/plan-sqlite-cleanup-v0.1.md index d632390..3c0baaf 100644 --- a/docs/plan-sqlite-cleanup-v0.1.md +++ b/docs/plan-sqlite-cleanup-v0.1.md @@ -1,13 +1,13 @@ # Dispatcher SQLite、本地文件清理与配置缓存调整计划（v0.1） -- **状态：** 首期本地实施与验收完成；未合入/推送 main、未部署。逐项证据见 `docs/evidence/sqlite-cleanup-v0.1-local.md` +- **状态：** 首期实施与复审修复完成，全量本地验收通过；未合入/推送 main、未部署。逐项证据见 `docs/evidence/sqlite-cleanup-v0.1-local.md` - **日期：** 2026-10-10 - **首期范围：** Dispatcher SQLite 内已满足条件的单次呼叫执行记录及回执；Agent `.approved` 持久执行标记、已完成交付的录音恢复文件；Dispatcher 已确认拒绝的原始数据文件。遗留临时文件仅发现并报告，不自动删除。 - **后续独立范围：** 当前配置内存化；不作为首期清理的开发或验收前置条件。 -- **执行授权：** 已确认首期本地开发、测试和相关文档更新；不授权部署、操作现存运行数据或发布共享合同改动 +- **执行授权：** 已确认首期本地开发、测试和相关文档更新；不授权部署或操作现存运行数据；复审后用户另行明确批准仅必要一句共享文字调整及其发布 - **决策来源：** 记录本轮较新的用户决定；若与旧项目计划中的清理门槛不一致，以本文件作为本地目标方案，不视作共享合同已变更。 -旧计划 [`plan-dispatcher-state-v0.1.md`](plan-dispatcher-state-v0.1.md) 对确认边界有不同的暂定描述。本文件不修改旧计划或共享合同；实施前需按 [`contracts.md`](contracts.md) 更新并核验共享合同，再同步整理相关项目文档。 +旧计划 [`plan-dispatcher-state-v0.1.md`](plan-dispatcher-state-v0.1.md) 对确认边界有不同的暂定描述。旧计划作为历史保留；复审后获批的共享文字调整已发布为 `7b4448f` 并固定本项目引用，仅明确本地记录保留期间的首次事实保护，不改机器 Schema、字段、类型或路由。合同同步仍按 [`contracts.md`](contracts.md) 执行。 ## 1. 目标 @@ -40,13 +40,13 @@ - 对已持久化为终态 `rejected`、且能证明呼叫未发起的 `call.execute`，只需其拒绝回执获 publisher confirm；不要求不存在的 Agent 结果 RPC 或 `call.execute.result`。拒绝回执未确认时保留相关记录。 - 对因任务停止而取消、且确认未发起的 `suppressed` 记录，所属任务持久状态为 `stopped`，且对应成功停止回执已获 publisher confirm 后，按任务清理；不等待 Agent 最终结果，不补造逐通拒绝或最终结果回执。停止回执须保留到相关 `suppressed` 记录清理完成，任务停止状态继续保留。 - 满足各自清理条件后，删除该次呼叫的 inbox、执行快照、录音上传元数据和对应 outbox 数据。按本地执行规则，清理完成后再次收到相同 `event_id`，按新执行处理；不保留永久 Dispatcher 防重记录，也不要求 SaaS 必须改用新 `event_id`。 -- **本次范围澄清：**不新增 `execution_id` 或其他外部编号；源 `event_id`（包括 numberID）在回执及最终结果中原样保留。SaaS 如何接收、判重、覆盖或区分同 ID 多次结果不在本次范围，也不作为开发和验收前置条件。本次只调整已完成数据的本地保留周期，不修改共享 Schema、消息类型、路由或发布共享合同改动；现有数据尚未清理期间的防重及未知执行保护继续有效，不保留永久完成历史。 +- **本次范围澄清：**不新增 `execution_id` 或其他外部编号；源 `event_id`（包括 numberID）在回执及最终结果中原样保留。SaaS 如何接收、判重、覆盖或区分同 ID 多次结果不在本次范围，也不作为开发和验收前置条件。本次只调整已完成数据的本地保留周期，不修改共享 Schema、字段、消息类型或路由；获批共享发布仅含本地保留周期的一句必要文字；现有数据尚未清理期间的防重及未知执行保护继续有效，不保留永久完成历史。 - 其他消息类型按各自所需回执处理：需要一份回执的，等该回执确认；需要多份的，等全部确认。 - 活动、未知、待执行、派发状态不确定、已派发但结果 RPC 未成功或任何必需 MQ 回执未确认的数据均保留。不得把未知执行当作失败或自动释放额度。 ## 3. 当前 SQLite 数据与清理候选 -当前 Store 使用单个 SQLite 数据库，包含以下运行数据。代码中没有通用 `DELETE` 清理器或保留期限；本节的候选资格是**未来实现目标**，不表示现有代码已证明或执行了删除。 +当前 Store 使用单个 SQLite 数据库，包含以下运行数据。首期已按本节资格实现专用逐通/逐回执清理，没有通用 `DELETE` 清理器或按年龄保留期限；逐项证据见本地迭代记录。 | 表/数据 | 当前用途 | 计划处理 | | --- | --- | --- | @@ -62,7 +62,7 @@ 以下文件不属于 SQLite 表。首期只自动清理已完成交付的录音恢复文件和已确认拒绝的原始数据；临时文件仅记录发现，不自动清理。可删除对象仍须满足各自完成条件，不按目录整体或文件年龄清空。 -| 文件/目录 | 当前行为 | 计划处理 | +| 文件/目录 | 改动前行为 | 首期处理 | | --- | --- | --- | | Agent 录音恢复目录内的 `recording.wav`、`info.json` | 上传失败后保存；恢复上传和结果上报成功后只写入 `delivered`，仍保留文件 | 原上传与结果交付成功已持久确认、且无未完成恢复引用后，删除该条录音和恢复信息，再删除空的自有末级目录；保留共享目录及其他记录。 | | Dispatcher `rejected-data/*.json` | 在拒绝回执写入 SQLite 前保存原始消息及反馈，当前没有删除路径 | 准确关联的拒绝回执已持久确认且无恢复引用后删除；原始文件清理完成前保留该回执的确认依据。 | @@ -115,8 +115,8 @@ Agent `.results` 成功上报后的文件、已处理完成的 SIP 来源文件 ### 已完成交付的 Agent 录音恢复文件 - 清理条件是：原上传已确认成功、原最终结果 RPC 成功返回、`delivered` 状态已持久保存，且不存在仍需使用这些文件的未完成恢复引用。仅上传成功或内存状态已改变不足以删除；`retry_pending`、`put_in_flight`、`outcome_unknown`、`uploaded_unreported`、`expired` 及损坏记录继续保留。 -- 在持久记录 `delivered` 后触发清理。以该条记录为单位协调恢复、写入与删除，先删除 `recording.wav` 并确保持久完成，再删除作为完成依据的 `info.json`，最后只删除确实为空且属于该条记录的末级目录。不递归删除恢复根目录、共享 bucket 目录或其他调用的文件，不触碰 OSS 对象。 -- 文件删除不是一个可回滚的数据库事务。若中途失败或重启，保留尚在的 `delivered` 信息，后续只继续删除，不能重新 PUT 或重报已确认结果。恢复扫描须能核验持久 `delivered` 后处理“音频已删、信息仍在”的合法中间状态，不能先要求音频完整而把清理永远阻断；非 `delivered` 缺音频仍是错误。只在该已确认清理路径中将目标已不存在视为完成，其他读取、权限和删除错误必须报出。 +- 在持久记录 `delivered` 后触发清理。以该条记录为单位协调恢复、写入与删除，先在私有 `.recording-cleanup` 命名空间持久保存原 delivered 信息的临时副本，再删除 `recording.wav` 并确保持久完成，删除 `info.json`，最后只删除确实为空且属于该条记录的末级目录。末级目录删除并同步父目录后才删除该副本；不是新确认协议或永久清理历史。不递归删除恢复根目录、共享 bucket 目录或其他调用的文件，不触碰 OSS 对象。 +- 文件删除不是一个可回滚的数据库事务。若中途失败或进程硬退出，保留原 `delivered` 信息或已同步的临时副本，后续只继续删除，不能重新 PUT 或重报已确认结果。信息已删而空叶仍在、空叶已删而副本仍在的窗口均按精确目标继续收尾，不泛扫无依据空目录；副本删除前禁止同目标的新恢复写入。恢复扫描须能核验持久 `delivered` 后处理“音频已删、信息仍在”的合法中间状态，不能先要求音频完整而把清理永远阻断；非 `delivered` 缺音频仍是错误。只在该已确认清理路径中将目标已不存在视为完成，其他读取、权限和删除错误必须报出。 - 此项会改变现有“恢复成功后仍保留录音”的行为，须同步调整对应测试；不把当前实现或仅有 `delivered` 写入说成已经完成清理。 ### Dispatcher 拒绝原始数据文件 @@ -159,12 +159,12 @@ Agent `.results` 成功上报后的文件、已处理完成的 SIP 来源文件 本地文件专项验收还须覆盖： - 原录音上传、结果上报及 `delivered` 持久化都成功后清除该条音频、恢复信息和空末级目录；任一条件缺失、仍有恢复引用或状态未知/过期时不删，其他记录及 OSS 对象不受影响。 -- 音频删除后、恢复信息删除前中断或重启，仍能依据持久完成状态继续清理，不重新上传或重报结果；删除失败保持完成依据且可重试，非已交付记录缺音频仍明确报错。 +- 在音频、恢复信息、空叶和临时完成副本每步删除后，子进程硬退出并重启，启动恢复及午夜入口均能依据持久完成状态继续清理，不重新上传或重报结果；删除失败保持完成依据且可重试，非已交付记录缺音频仍明确报错。 - 拒绝原始文件仅在正确关联回执确认后删除；通用 outbox 清理不能先丢弃依据。覆盖一份回执关联多份原始文件、仅落盘但回执事务失败、文件删除中断和回执删除失败，确保未确认或关联不明的数据保留。 - 临时文件仅做现有行为回归：正常写入收尾仍删除自身临时文件；异常遗留、正在写入、唯一未提交副本及未知文件均不会被新增清理路径或 24:00 补扫误删，现有检查发现的问题可见。首期不要求实现或验收自动冗余识别。 - 文件删除与恢复/写入并发时不误删；24:00 补扫可完成遗漏，重复执行无副作用；已存在的 `.results` 和 SIP 来源文件成功清理行为不退化，不新增永久清理历史或按年龄删除未知记录。 -共享合同仍按 `docs/contracts.md` 执行 `make contracts-update` 和固定引用核验，不为本次内部清理新增接口字段或发布合同改动。SaaS 对重复结果的处理不属于本次验收，不宣称 SaaS 已收到或能区分多次执行结果。若改动 SQLite 表结构，按项目规则评估新的数据库布局版本；不自动迁移或重置现有数据库。 +共享合同仍按 `docs/contracts.md` 执行 `make contracts-update` 和固定引用核验，不为本次内部清理新增接口字段；仅发布用户批准的本地保留周期文字 `7b4448f`，本项目引用、实现、测试及说明同步。SaaS 对重复结果的处理不属于本次验收，不宣称 SaaS 已收到或能区分多次执行结果。若改动 SQLite 表结构，按项目规则评估新的数据库布局版本；不自动迁移或重置现有数据库。 ## 7. Agent 本地恢复与全局停接边界（只读核查） diff --git a/internal/agent/recording_cleanup.go b/internal/agent/recording_cleanup.go index ec5b97c..9ff3334 100644 --- a/internal/agent/recording_cleanup.go +++ b/internal/agent/recording_cleanup.go @@ -15,8 +15,8 @@ type RecordingCleanupResult struct { RetainedReason string } -// cleanupDelivered requires the recording path lock. The original delivered -// info remains the completion proof until its audio deletion is durable. +// cleanupDelivered requires the recording path lock. A temporary durable copy +// of the delivered info remains outside the leaf until that leaf is gone. func (r *RecordingRecovery) cleanupDelivered(entry RecordingRecoveryEntry) (result RecordingCleanupResult, cleanupErr error) { defer func() { attrs := append(recordingRecoveryAttrs(entry), "stage", "delivered_recording_cleanup", "deleted_files", result.DeletedFiles, "deleted_bytes", result.DeletedBytes) @@ -35,27 +35,38 @@ func (r *RecordingRecovery) cleanupDelivered(entry RecordingRecoveryEntry) (resu if err != nil { return result, err } - // Never follow a directory link or remove shared parents/unknown sidecars. + proofPath, err := r.cleanupProofPath(entry.Bucket, entry.ObjectKey) + if err != nil { + return result, err + } + _, proofErr := r.readCleanupProof(proofPath) + if proofErr != nil && !errors.Is(proofErr, os.ErrNotExist) { + return result, proofErr + } + proved := proofErr == nil + leafExists := true current := r.Root for _, segment := range append([]string{"", entry.Bucket}, strings.Split(entry.ObjectKey, "/")...) { if segment != "" { current = filepath.Join(current, segment) } - stat, err := os.Lstat(current) + if err := privateCleanupDirectory(current); err != nil { + if current == dir && proved && errors.Is(err, os.ErrNotExist) { + leafExists = false + break + } + return result, err + } + } + if leafExists { + items, err := os.ReadDir(dir) if err != nil { return result, err } - if !stat.IsDir() || stat.Mode()&os.ModeSymlink != 0 || stat.Mode().Perm()&0077 != 0 { - return result, errors.New("recording cleanup target is not an owned private directory") - } - } - items, err := os.ReadDir(dir) - if err != nil { - return result, err - } - for _, item := range items { - if item.Name() != "recording.wav" && item.Name() != "info.json" { - return result, errors.New("recording cleanup target contains an unrecognized file") + for _, item := range items { + if item.Name() != "recording.wav" && item.Name() != "info.json" { + return result, errors.New("recording cleanup target contains an unrecognized file") + } } } journal := ResultJournal{Root: r.Root} @@ -79,7 +90,6 @@ func (r *RecordingRecovery) cleanupDelivered(entry RecordingRecoveryEntry) (resu if !errors.Is(err, os.ErrNotExist) { return result, fmt.Errorf("inspect original recording result reference: %w", err) } - // Only this already-confirmed deletion path accepts an absent owned file. remove := func(path string) error { stat, err := os.Lstat(path) if errors.Is(err, os.ErrNotExist) { @@ -105,43 +115,50 @@ func (r *RecordingRecovery) cleanupDelivered(entry RecordingRecoveryEntry) (resu } return nil } - // Validate the completion proof's file before removing either member. stat, err := os.Lstat(info) - if err != nil || !stat.Mode().IsRegular() || stat.Mode().Perm()&0077 != 0 { - return result, errors.Join(errors.New("recording delivery proof is not a private regular file"), err) + if !(proved && errors.Is(err, os.ErrNotExist)) { + if err != nil || !stat.Mode().IsRegular() || stat.Mode().Perm()&0077 != 0 { + return result, errors.Join(errors.New("recording delivery proof is not a private regular file"), err) + } + } + // Validate any existing receipt before deleting either original member, and + // persist its copy before info unlink. Hard exits cannot lose the last proof. + proofPath, err = r.preserveCleanupProof(entry) + if err != nil { + return result, fmt.Errorf("preserve delivered cleanup receipt: %w", err) } if err := remove(audio); err != nil { return result, fmt.Errorf("delete delivered recording audio: %w", err) } - if err := syncDirectory(dir); err != nil { - return result, fmt.Errorf("persist delivered audio deletion: %w", err) - } - filesBeforeInfo, bytesBeforeInfo := result.DeletedFiles, result.DeletedBytes - restoreProof := func() error { - err := r.persistState(info, entry) - if err == nil { - result.DeletedFiles, result.DeletedBytes = filesBeforeInfo, bytesBeforeInfo + if leafExists { + if err := syncDirectory(dir); err != nil { + return result, fmt.Errorf("persist delivered audio deletion: %w", err) } - return err } if err := remove(info); err != nil { return result, fmt.Errorf("delete delivered recording info: %w", err) } - if err := syncDirectory(dir); err != nil { - return result, errors.Join(fmt.Errorf("persist delivered info deletion: %w", err), restoreProof()) - } - if r.removeFile != nil { - err = r.removeFile(dir) - } else { - err = os.Remove(dir) - } - if err != nil && !errors.Is(err, os.ErrNotExist) { - // The audio is gone, but a restored delivered receipt lets the next - // recovery pass retry this exact leaf without replaying PUT/report. - return result, errors.Join(fmt.Errorf("remove completed recording leaf: %w", err), restoreProof()) + if leafExists { + if err := syncDirectory(dir); err != nil { + return result, fmt.Errorf("persist delivered info deletion: %w", err) + } + if r.removeFile != nil { + err = r.removeFile(dir) + } else { + err = os.Remove(dir) + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + return result, fmt.Errorf("remove completed recording leaf: %w", err) + } } if err := syncDirectory(filepath.Dir(dir)); err != nil { return result, fmt.Errorf("persist recording leaf deletion: %w", err) } + if err := remove(proofPath); err != nil { + return result, fmt.Errorf("remove completed cleanup receipt: %w", err) + } + if err := syncDirectory(filepath.Dir(proofPath)); err != nil { + return result, fmt.Errorf("persist cleanup receipt deletion: %w", err) + } return result, nil } diff --git a/internal/agent/recording_cleanup_proof.go b/internal/agent/recording_cleanup_proof.go new file mode 100644 index 0000000..f63b620 --- /dev/null +++ b/internal/agent/recording_cleanup_proof.go @@ -0,0 +1,159 @@ +package agent + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" +) + +// This temporary copy of an existing delivered receipt survives the final +// info unlink and rmdir. It is removed after the exact owned leaf is durably +// gone; it is not a completion history or an acknowledgment protocol. +const recordingCleanupDirectory = ".recording-cleanup" + +func (r *RecordingRecovery) cleanupProofPath(bucket, objectKey string) (string, error) { + if _, _, _, err := r.paths(bucket, objectKey); err != nil { + return "", err + } + digest := sha256.Sum256([]byte(bucket + "/" + objectKey)) + return filepath.Join(r.Root, recordingCleanupDirectory, fmt.Sprintf("%x.json", digest)), nil +} + +func privateCleanupDirectory(path string) error { + stat, err := os.Lstat(path) + if err != nil { + return err + } + if !stat.IsDir() || stat.Mode()&os.ModeSymlink != 0 || stat.Mode().Perm()&0077 != 0 { + return errors.New("recording cleanup directory is not an owned private directory") + } + return nil +} + +func (r *RecordingRecovery) readCleanupProof(path string) (RecordingRecoveryEntry, error) { + for _, dir := range []string{r.Root, filepath.Join(r.Root, recordingCleanupDirectory)} { + if err := privateCleanupDirectory(dir); err != nil { + return RecordingRecoveryEntry{}, err + } + } + stat, err := os.Lstat(path) + if err != nil { + return RecordingRecoveryEntry{}, err + } + if !stat.Mode().IsRegular() || stat.Mode().Perm()&0077 != 0 { + return RecordingRecoveryEntry{}, errors.New("recording cleanup proof is not a private regular file") + } + data, err := os.ReadFile(path) + if err != nil { + return RecordingRecoveryEntry{}, err + } + var identity RecordingRecoveryEntry + if err := json.Unmarshal(data, &identity); err != nil { + return identity, err + } + entry, err := decodeRecordingInfo(data, identity.Bucket, identity.ObjectKey) + if err != nil { + return entry, err + } + if entry.State != "delivered" { + return entry, errors.New("recording cleanup proof is not delivered") + } + expected, err := r.cleanupProofPath(entry.Bucket, entry.ObjectKey) + if err != nil { + return entry, err + } + if filepath.Clean(path) != filepath.Clean(expected) { + return entry, errors.New("recording cleanup proof filename conflicts with its original target") + } + return entry, nil +} + +func (r *RecordingRecovery) preserveCleanupProof(entry RecordingRecoveryEntry) (string, error) { + path, err := r.cleanupProofPath(entry.Bucket, entry.ObjectKey) + if err != nil { + return "", err + } + if err := privateCleanupDirectory(r.Root); err != nil { + return "", err + } + dir := filepath.Dir(path) + if err := os.Mkdir(dir, 0700); err == nil { + if err := syncDirectory(r.Root); err != nil { + return "", err + } + } else if !errors.Is(err, os.ErrExist) { + return "", err + } + if err := privateCleanupDirectory(dir); err != nil { + return "", err + } + before, err := r.readCleanupProof(path) + if err == nil { + original, marshalErr := json.Marshal(entry) + persisted, beforeErr := json.Marshal(before) + if marshalErr != nil || beforeErr != nil { + return "", errors.Join(marshalErr, beforeErr) + } + if !bytes.Equal(original, persisted) { + return "", errors.New("recording cleanup proof conflicts with current recording") + } + return path, nil + } + if !errors.Is(err, os.ErrNotExist) { + return "", err + } + if err := r.persistState(path, entry); err != nil { + return "", err + } + return path, nil +} + +// Scan only explicitly proved targets, never arbitrary empty directories. +func (r *RecordingRecovery) resumeCleanupProofs(ctx context.Context) (int, error) { + dir := filepath.Join(r.Root, recordingCleanupDirectory) + if err := privateCleanupDirectory(dir); errors.Is(err, os.ErrNotExist) { + return 0, nil + } else if err != nil { + return 0, err + } + items, err := os.ReadDir(dir) + if err != nil { + return 0, err + } + count := 0 + var blocked []error + for _, item := range items { + if err := ctx.Err(); err != nil { + return count, errors.Join(append(blocked, err)...) + } + path := filepath.Join(dir, item.Name()) + entry, err := r.readCleanupProof(path) + if errors.Is(err, os.ErrNotExist) { + continue + } // independent scanner completed it + if err == nil { + _, err = r.Retry(ctx, entry.Bucket, entry.ObjectKey, nil, nil) + if errors.Is(err, os.ErrNotExist) { + continue + } + } + if err != nil { + digest := sha256.Sum256([]byte(item.Name())) + slog.ErrorContext(ctx, "Delivered recording cleanup proof retained", "stage", "delivered_cleanup_resume", "proof_digest", fmt.Sprintf("%x", digest[:6]), "error", err) + blocked = append(blocked, fmt.Errorf("recording cleanup proof %x: %w", digest[:6], err)) + continue + } + if _, err := os.Lstat(path); errors.Is(err, os.ErrNotExist) { + count++ + } else if err != nil { + blocked = append(blocked, err) + } + } + return count, errors.Join(blocked...) +} diff --git a/internal/agent/recording_cleanup_proof_test.go b/internal/agent/recording_cleanup_proof_test.go new file mode 100644 index 0000000..83a2806 --- /dev/null +++ b/internal/agent/recording_cleanup_proof_test.go @@ -0,0 +1,128 @@ +package agent + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestCleanupProofMustPersistBeforeEitherOriginalIsDeleted(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + _, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + r.writeState = func(path string, value any) error { + if filepath.Dir(path) == filepath.Join(r.Root, recordingCleanupDirectory) { + return errors.New("proof disk failure") + } + return writeJSONAtomic(path, value) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err == nil { + t.Fatal("proof failure swallowed") + } + for _, path := range []string{audio, info} { + if _, err := os.Stat(path); err != nil { + t.Fatalf("original lost before durable proof: %v", err) + } + } + r.writeState = nil + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } +} + +func TestCleanupProofBlocksTargetReuseUntilFinalRemovalAndIsolatesFailures(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + proof, _ := r.cleanupProofPath(e.Bucket, e.ObjectKey) + r.removeFile = func(path string) error { + if path == proof { + return errors.New("proof removal failure") + } + return os.Remove(path) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err == nil { + t.Fatal("proof deletion failure swallowed") + } + dir, _, _, _ := r.paths(e.Bucket, e.ObjectKey) + if _, err := os.Stat(dir); !errors.Is(err, os.ErrNotExist) { + t.Fatal("original leaf deletion did not finish") + } + fresh := failedRecordingFixture() + if _, err := r.SaveFailure([]byte("new recording"), fresh, &UploadHTTPError{StatusCode: 503}); err == nil { + t.Fatal("new target overwritten before old proof cleanup") + } + other := fresh + other.SourceEventID = "independent-delivered" + other.ObjectKey = "tenant/call/independent.wav" + if _, err := r.SaveFailure([]byte("independent recording"), other, &UploadHTTPError{StatusCode: 503}); err != nil { + t.Fatal(err) + } + other, err := r.Load(other.Bucket, other.ObjectKey) + if err != nil { + t.Fatal(err) + } + other.State = "delivered" + other.Uploaded = &UploadResult{StatusCode: 200, SizeBytes: other.SizeBytes, SHA256: other.SHA256} + _, _, info, _ := r.paths(other.Bucket, other.ObjectKey) + if err := r.persistState(info, other); err != nil { + t.Fatal(err) + } + if count, err := r.CleanupDelivered(context.Background()); err == nil || count != 1 { + t.Fatalf("independent cleanup blocked: count=%d err=%v", count, err) + } + if _, err := os.Stat(proof); err != nil { + t.Fatal("failure lost completion receipt") + } + restarted := &RecordingRecovery{Root: r.Root} + if count, err := restarted.CleanupDelivered(context.Background()); err != nil || count != 1 { + t.Fatalf("final proof not resumed: count=%d err=%v", count, err) + } + if _, err := restarted.SaveFailure([]byte("new recording"), fresh, &UploadHTTPError{StatusCode: 503}); err != nil { + t.Fatal("cleaned identity cannot be reused", err) + } +} + +func TestCorruptOrConflictingCleanupProofNeverDeletesOriginals(t *testing.T) { + for _, damage := range []string{"identity", "filename", "state", "receipt", "json"} { + t.Run(damage, func(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + proof, err := r.preserveCleanupProof(e) + if err != nil { + t.Fatal(err) + } + _, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + switch damage { + case "identity": + e.SourceEventID = "wrong-original" + err = r.persistState(proof, e) + case "filename": + renamed := filepath.Join(filepath.Dir(proof), strings.Repeat("0", 64)+".json") + err = os.Rename(proof, renamed) + proof = renamed + case "state": + e.State = "uploaded_unreported" + err = r.persistState(proof, e) + case "receipt": + e.Uploaded.StatusCode = 500 + err = r.persistState(proof, e) + case "json": + err = os.WriteFile(proof, []byte("broken original proof"), 0600) + } + if err != nil { + t.Fatal(err) + } + if _, err := r.resumeCleanupProofs(context.Background()); err == nil { + t.Fatal("invalid proof accepted") + } + for _, path := range []string{audio, info, proof} { + if _, err := os.Stat(path); err != nil { + t.Fatalf("invalid association deleted a file: %v", err) + } + } + }) + } +} diff --git a/internal/agent/recording_cleanup_test.go b/internal/agent/recording_cleanup_test.go index 4f68956..1fbbaac 100644 --- a/internal/agent/recording_cleanup_test.go +++ b/internal/agent/recording_cleanup_test.go @@ -3,7 +3,10 @@ package agent import ( "context" "errors" + "fmt" + "io/fs" "os" + "os/exec" "path/filepath" "testing" "time" @@ -31,6 +34,89 @@ func deliveredRecordingFixture(t *testing.T, r *RecordingRecovery) RecordingReco return e } +func TestDeliveredCleanupHardExitHelper(t *testing.T) { + root := os.Getenv("RECORDING_CLEANUP_EXIT_ROOT") + if root == "" { + return + } + r := &RecordingRecovery{Root: root} + e := failedRecordingFixture() + dir, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + r.removeFile = func(path string) error { + if err := os.Remove(path); err != nil { + return err + } + step := os.Getenv("RECORDING_CLEANUP_EXIT_STEP") + if (step == "audio" && path == audio) || (step == "info" && path == info) || (step == "leaf" && path == dir) || (step == "proof" && filepath.Dir(path) == filepath.Join(root, recordingCleanupDirectory)) { + if err := syncDirectory(filepath.Dir(path)); err != nil { + return err + } + os.Exit(23) // no defers, proof restoration, or orderly shutdown + } + return nil + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } + t.Fatal("hard-exit injection did not run") +} + +func TestDeliveredCleanupResumesAfterHardExitAtEveryFinalDeletion(t *testing.T) { + for _, step := range []string{"audio", "info", "leaf", "proof"} { + for _, midnight := range []bool{false, true} { + t.Run(step+"/midnight="+fmt.Sprint(midnight), func(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + dir, _, _, _ := r.paths(e.Bucket, e.ObjectKey) + cmd := exec.Command(os.Args[0], "-test.run=^TestDeliveredCleanupHardExitHelper$") + cmd.Env = append(os.Environ(), "RECORDING_CLEANUP_EXIT_ROOT="+r.Root, "RECORDING_CLEANUP_EXIT_STEP="+step) + if err := cmd.Run(); err == nil || cmd.ProcessState.ExitCode() != 23 { + t.Fatalf("helper did not exit at boundary: %v", err) + } + unknown := filepath.Join(r.Root, "unowned-empty-leaf") + if err := os.Mkdir(unknown, 0700); err != nil { + t.Fatal(err) + } + restarted := &RecordingRecovery{Root: r.Root} + calls := 0 + var err error + if midnight { + _, err = restarted.CleanupDelivered(context.Background()) + } else { + _, err = restarted.RecoverDue(context.Background(), func(context.Context, RecordingRecoveryEntry) (RecoveryTarget, error) { + calls++ + return RecoveryTarget{}, errors.New("no grant") + }, func(context.Context, RecordingRecoveryEntry) error { calls++; return errors.New("no report") }) + } + if err != nil || calls != 0 { + t.Fatalf("restart replayed work: calls=%d err=%v", calls, err) + } + if _, err := os.Stat(dir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("completed leaf remains after hard exit: %v", err) + } + if _, err := os.Stat(unknown); err != nil { + t.Fatal("unknown empty directory was removed") + } + files := 0 + if err := filepath.WalkDir(r.Root, func(_ string, item fs.DirEntry, err error) error { + if err != nil { + return err + } + if !item.IsDir() { + files++ + } + return nil + }); err != nil { + t.Fatal(err) + } + if files != 0 { + t.Fatalf("completion proof or files accumulated: %d", files) + } + }) + } + } +} + func TestDeliveredRecordingCleanupAfterRestartWithoutAudio(t *testing.T) { r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private"), Now: func() time.Time { return time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC) }} e := deliveredRecordingFixture(t, r) diff --git a/internal/agent/recording_recovery.go b/internal/agent/recording_recovery.go index 241ded3..a3d9e57 100644 --- a/internal/agent/recording_recovery.go +++ b/internal/agent/recording_recovery.go @@ -71,6 +71,15 @@ func (r *RecordingRecovery) SaveFailure(audio []byte, entry RecordingRecoveryEnt return RecordingRecoveryEntry{}, err } defer release() + proofPath, err := r.cleanupProofPath(entry.Bucket, entry.ObjectKey) + if err != nil { + return RecordingRecoveryEntry{}, err + } + if _, err := os.Lstat(proofPath); err == nil { + return RecordingRecoveryEntry{}, errors.New("previous delivered recording cleanup is unfinished for this target") + } else if !errors.Is(err, os.ErrNotExist) { + return RecordingRecoveryEntry{}, err + } sha := sha256.Sum256(audio) digest := hex.EncodeToString(sha[:]) if entry.SHA256 != "" && !strings.EqualFold(entry.SHA256, digest) { @@ -135,6 +144,20 @@ func (r *RecordingRecovery) load(bucket, objectKey string) (RecordingRecoveryEnt } data, err := os.ReadFile(infoPath) if errors.Is(err, os.ErrNotExist) { + proof, proofErr := r.cleanupProofPath(bucket, objectKey) + if proofErr != nil { + return RecordingRecoveryEntry{}, proofErr + } + entry, proofErr := r.readCleanupProof(proof) + if proofErr == nil { + if _, audioErr := os.Lstat(audioPath); !errors.Is(audioErr, os.ErrNotExist) { + return RecordingRecoveryEntry{}, errors.Join(ErrRecordingRecoveryIncomplete, audioErr) + } + return entry, nil + } + if !errors.Is(proofErr, os.ErrNotExist) { + return RecordingRecoveryEntry{}, proofErr + } if _, audioErr := os.Stat(audioPath); audioErr == nil { return RecordingRecoveryEntry{}, ErrRecordingRecoveryIncomplete } @@ -142,6 +165,25 @@ func (r *RecordingRecovery) load(bucket, objectKey string) (RecordingRecoveryEnt if err != nil { return RecordingRecoveryEntry{}, err } + entry, err := decodeRecordingInfo(data, bucket, objectKey) + if err != nil { + return RecordingRecoveryEntry{}, err + } + audio, err := os.ReadFile(audioPath) + if errors.Is(err, os.ErrNotExist) && entry.State == "delivered" { + return entry, nil // legal interruption after confirmed audio deletion + } + if err != nil { + return RecordingRecoveryEntry{}, fmt.Errorf("%w: read original audio: %v", ErrRecordingRecoveryIncomplete, err) + } + sha := sha256.Sum256(audio) + if int64(len(audio)) != entry.SizeBytes || !strings.EqualFold(hex.EncodeToString(sha[:]), entry.SHA256) { + return RecordingRecoveryEntry{}, ErrRecordingRecoveryIncomplete + } + return entry, nil +} + +func decodeRecordingInfo(data []byte, bucket, objectKey string) (RecordingRecoveryEntry, error) { var entry RecordingRecoveryEntry decoder := json.NewDecoder(bytes.NewReader(data)) decoder.DisallowUnknownFields() @@ -171,22 +213,11 @@ func (r *RecordingRecovery) load(bucket, objectKey string) (RecordingRecoveryEnt } else if entry.Uploaded != nil { return RecordingRecoveryEntry{}, ErrRecordingRecoveryIncomplete } - audio, err := os.ReadFile(audioPath) - if errors.Is(err, os.ErrNotExist) && entry.State == "delivered" { - return entry, nil // legal interruption after confirmed audio deletion - } - if err != nil { - return RecordingRecoveryEntry{}, fmt.Errorf("%w: read original audio: %v", ErrRecordingRecoveryIncomplete, err) - } - sha := sha256.Sum256(audio) - if int64(len(audio)) != entry.SizeBytes || !strings.EqualFold(hex.EncodeToString(sha[:]), entry.SHA256) { - return RecordingRecoveryEntry{}, ErrRecordingRecoveryIncomplete - } return entry, nil } func (r *RecordingRecovery) paths(bucket, objectKey string) (dir, audioPath, infoPath string, err error) { - if strings.TrimSpace(r.Root) == "" || validateName(bucket) != nil || objectKey == "" || strings.ContainsAny(objectKey, "\\\x00") { + if strings.TrimSpace(r.Root) == "" || validateName(bucket) != nil || bucket == recordingCleanupDirectory || objectKey == "" || strings.ContainsAny(objectKey, "\\\x00") { return "", "", "", errors.New("recovery root, bucket or object key is invalid") } for _, segment := range strings.Split(objectKey, "/") { diff --git a/internal/agent/recording_scan.go b/internal/agent/recording_scan.go index 49ba767..0af1ea2 100644 --- a/internal/agent/recording_scan.go +++ b/internal/agent/recording_scan.go @@ -44,9 +44,16 @@ func (r *RecordingRecovery) scanRecovery(ctx context.Context, grant RecoveryGran if err := checkRecoveryDirectory(r.Root); err != nil { return 0, err } + resumed, proofErr := r.resumeCleanupProofs(ctx) processed := 0 + if deliveredOnly { + processed = resumed + } handled := make(map[string]bool) var blocked []error + if proofErr != nil { + blocked = append(blocked, proofErr) + } addBlocked := func(dir, stage string, cause error, identity ...any) { digest := sha256.Sum256([]byte(dir)) attrs := append([]any{"stage", stage, "recovery_digest", fmt.Sprintf("%x", digest[:6]), "error", cause}, identity...) @@ -78,7 +85,7 @@ func (r *RecordingRecovery) scanRecovery(ctx context.Context, grant RecoveryGran addBlocked(path, "recovery_directory", err) return filepath.SkipDir } - if path == filepath.Join(r.Root, ".results") { + if path == filepath.Join(r.Root, ".results") || path == filepath.Join(r.Root, recordingCleanupDirectory) { return filepath.SkipDir // result-only files have their own scanner } return nil