From 523ac84b2c746921cff32430785ebf99df9694ae Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 11 Oct 2026 18:48:22 +0800 Subject: [PATCH] feat(dispatcher): clean confirmed completed call state transactionally --- docs/evidence/sqlite-cleanup-v0.1-local.md | 33 +++ internal/dispatcher/cleanup.go | 33 +++ internal/dispatcher/cleanup_test.go | 160 ++++++++++++ internal/dispatcher/execute.go | 15 +- internal/dispatcher/runtime.go | 5 +- internal/store/calls.go | 14 ++ internal/store/cleanup.go | 275 +++++++++++++++++++++ internal/store/cleanup_control.go | 172 +++++++++++++ internal/store/cleanup_control_test.go | 121 +++++++++ internal/store/cleanup_test.go | 211 ++++++++++++++++ internal/store/store.go | 6 +- 11 files changed, 1041 insertions(+), 4 deletions(-) create mode 100644 docs/evidence/sqlite-cleanup-v0.1-local.md create mode 100644 internal/dispatcher/cleanup.go create mode 100644 internal/dispatcher/cleanup_test.go create mode 100644 internal/store/cleanup.go create mode 100644 internal/store/cleanup_control.go create mode 100644 internal/store/cleanup_control_test.go create mode 100644 internal/store/cleanup_test.go diff --git a/docs/evidence/sqlite-cleanup-v0.1-local.md b/docs/evidence/sqlite-cleanup-v0.1-local.md new file mode 100644 index 0000000..52f4b2e --- /dev/null +++ b/docs/evidence/sqlite-cleanup-v0.1-local.md @@ -0,0 +1,33 @@ +# SQLite cleanup v0.1 本地迭代记录 + +## 范围与状态 + +- 开发分支：`feat/sqlite-cleanup-v0.1`；首期开发尚未全部完成。 +- 用户确认：仅本项目内部数据保留周期调整，不新增 `execution_id`，源 `event_id`（包括 numberID）原样回传；SaaS 如何处理再次返回的同 ID 结果不在范围内。 +- 共享合同固定 `ef3009d`，`make contracts-update` 的 43 个 JSON 检查与 8 项合同测试通过；接口字段、消息类型、路由、共享文件及 submodule 指针不变，没有共享发布。 +- 没有合入或推送本项目 main、部署、操作现存运行数据库/恢复目录、退役主机 `.approved`、拨号、抓包或调用真实 AI/OSS。 + +## 已完成的分批验证 + +### Agent 永久标记退役（`f07bf90`） + +- 先新增并运行反例，原实现因创建 `.approved`、依赖旧标记及完成后不允许重用而失败；删除相关生产读写代码后通过。 +- 验证同身份活动期间不能重叠执行，结束后可再次执行；旧 `.approved` 文件/目录不读取、不改写、不删除，新运行不重建该目录。 +- 仅在临时目录验证私密备份字节、长度和 SHA-256 一致后删除准确标记路径，相邻会话、`.results`、`.executions`、录音、上传恢复、锁及 Dispatcher 数据均保留。 +- `go test -race ./internal/rpc ./internal/agent ./internal/store ./internal/dispatcher` 通过。 + +### SQLite 实时清理 + +- 新增测试先因缺少清理入口失败，实现后全部通过。 +- 已派发呼叫需要已确认终结、通过现有结果 RPC 验证后持久提交的结果，以及派发回执、最终结果各自的 MQ publisher confirm。结果提交之后没有额外可能失败的业务验收步骤；不增加第二次 Agent 确认，不把 RPC 响应的网络到达或 MQ confirm 当作 SaaS 应用收讫。 +- 删除事务重新检查类型、源 ID、归属、号码、线路、冻结快照及上传结果；矛盾证据保留并报错。SQLite 结构仍为 4，没有新表或迁移。 +- 未发起拒绝仅清理已确认拒绝；原始拒绝文件仍须由后续文件收尾步骤保护其回执。活动、pending、dispatching、dispatched、unknown、任一确认缺失均保留。 +- stopped 的取消记录逐条事务、分批处理；单条失败不阻止其他记录，停止回执到剩余取消/待拨记录清完后才删，任务停止状态不删。205 条取消记录的批次、取消中断和 SQL 故障恢复测试通过。 +- 持有待拨指令或原派发调用期间排除清理；重新核对缓冲指令，停止/删除/替换后的旧批次不再拨号。极速终结与原派发返回并发时，先完成原调用再清理。 +- MQ publisher confirm 后触发清理；清理故障记录身份与原因，不作为业务准入故障返回。 +- 原 `event_id` 清理后可再次接纳、执行和回传；只删除本通数据，共享配置和任务状态保留。报告的是删除的逻辑行/正文字节，不声称 SQLite 文件立即变小。 +- `go test -race ./internal/store ./internal/dispatcher ./internal/rpc` 通过。 + +## 尚待完成 + +已交付录音、拒绝原始文件的准确关联与续处理、午夜补扫，以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。 diff --git a/internal/dispatcher/cleanup.go b/internal/dispatcher/cleanup.go new file mode 100644 index 0000000..5a89ec2 --- /dev/null +++ b/internal/dispatcher/cleanup.go @@ -0,0 +1,33 @@ +package dispatcher + +import ( + "context" + "log/slog" + + "git.ipao.vip/rogee/go-sip/internal/store" +) + +// A cleanup failure is local to completed data. Publication has already been +// confirmed; it must not close admission or prevent independent deliveries. +func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event store.OutboxEvent) { + var result store.CleanupResult + var err error + switch event.EventType { + case "call.execute", "call.execute.result": + result, err = c.Store.CleanupCall(c.DispatcherID, event.EventID) + case "task.control": + result, err = c.Store.CleanupControlReply(ctx, c.DispatcherID, event.EventID) + default: + return + } + logger := slog.Default().With("dispatcher_id", c.DispatcherID, "event_type", event.EventType, "event_id", event.EventID, "stage", "confirmed_data_cleanup") + if err != nil { + logger.Error("Completed data cleanup failed; evidence retained", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes, "error", err) + return + } + if result.DeletedRows != 0 { + logger.Info("Completed data cleaned", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes) + } else { + logger.Debug("Completed data retained", "reason", result.RetainedReason) + } +} diff --git a/internal/dispatcher/cleanup_test.go b/internal/dispatcher/cleanup_test.go new file mode 100644 index 0000000..cc5fd87 --- /dev/null +++ b/internal/dispatcher/cleanup_test.go @@ -0,0 +1,160 @@ +package dispatcher + +import ( + "context" + "encoding/json" + "testing" + "time" + + "git.ipao.vip/rogee/go-sip/internal/store" +) + +func cleanerResultPayload() []byte { + return []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:01Z","ended_at":"2026-09-21T01:30:06Z","duration_ms":5000,"outcome":"no_answer","reason_code":486,"reason_message":"busy","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`) +} + +func TestConfirmedResultCleansCallAndPreservesSourceIDForReuse(t *testing.T) { + controller, originator, publisher, s := newExecuteFixture(t) + id := "call:numberID:000042" + body := executeBody(t, id, "15003164745") + if err := controller.ProcessExecute(context.Background(), body); err != nil { + t.Fatal(err) + } + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + if err := s.FinishExecute(controller.DispatcherID, id); err != nil { + t.Fatal(err) + } + if _, _, err := s.RecordCallResult(controller.DispatcherID, id, cleanerResultPayload()); err != nil { + t.Fatal(err) + } + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + if err := s.RequireReservedCall(controller.DispatcherID, id, 1001); err == nil { + t.Fatal("completed call was retained after both confirmations") + } + if err := controller.ProcessExecute(context.Background(), body); err != nil { + t.Fatal(err) + } + if len(originator.calls) != 2 { + t.Fatalf("same source event was not reusable: calls=%d", len(originator.calls)) + } + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + if len(publisher.bodies) != 3 { + t.Fatalf("feedback count=%d", len(publisher.bodies)) + } + for _, event := range publisher.bodies { + var feedback struct { + EventID string `json:"event_id"` + EventType string `json:"event_type"` + } + if err := json.Unmarshal(event, &feedback); err != nil || feedback.EventID != id { + t.Fatalf("source event ID changed: %+v err=%v", feedback, err) + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(event, &fields); err != nil { + t.Fatal(err) + } + if _, added := fields["execution_id"]; added { + t.Fatal("cleanup added an external execution identity") + } + } +} + +func TestConfirmedStopCleansCancelledBatchWithoutDispatchError(t *testing.T) { + controller, originator, _, s := newExecuteFixture(t) + cmd := store.ExecuteCommand{DispatcherID: controller.DispatcherID, EventID: "cancelled-buffer", TenantID: 1001, TaskID: "task-asr", Callee: "15003164745", IssuedAt: "2026-09-21T01:00:00Z"} + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + batch, err := s.ListPendingExecute(controller.DispatcherID) + if err != nil || len(batch) != 1 { + t.Fatalf("batch=%v err=%v", batch, err) + } + if err := s.PrepareControl(controller.DispatcherID, 1001, "task-asr", "stop"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(controller.DispatcherID, 1001, "task-asr", "stop", "stop-buffer"); err != nil { + t.Fatal(err) + } + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + if err := controller.dispatchPending(context.Background(), batch[0]); err != nil { + t.Fatalf("cancelled buffered item closed normal processing: %v", err) + } + if len(originator.calls) != 0 { + t.Fatal("cancelled buffered instruction originated") + } + if admitted, err := s.CanAdmit(controller.DispatcherID, 1001, "task-asr"); err != nil || admitted { + t.Fatalf("durable stop state lost: %t %v", admitted, err) + } +} + +type blockedCleanupOriginator struct{ started, release chan struct{} } + +func (o *blockedCleanupOriginator) LoadedTrunks(context.Context) (map[string]int64, error) { + return map[string]int64{"trunk-mock": 8}, nil +} +func (o *blockedCleanupOriginator) Originate(ctx context.Context, _ CallSpec) error { + close(o.started) + select { + case <-o.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func TestCleanupWaitsForOriginalDispatchHandlerToFinish(t *testing.T) { + controller, _, _, s := newExecuteFixture(t) + originator := &blockedCleanupOriginator{started: make(chan struct{}), release: make(chan struct{})} + controller.Originator = originator + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + body := executeBody(t, "rapid-end", "15003164745") + dispatched := make(chan error, 1) + go func() { dispatched <- controller.ProcessExecute(ctx, body) }() + select { + case <-originator.started: + case <-ctx.Done(): + t.Fatal("originator did not start") + } + if err := s.FinishExecute(controller.DispatcherID, "rapid-end"); err != nil { + t.Fatal(err) + } + if _, _, err := s.RecordCallResult(controller.DispatcherID, "rapid-end", cleanerResultPayload()); err != nil { + t.Fatal(err) + } + flushed := make(chan error, 1) + go func() { flushed <- controller.FlushOutbox(ctx) }() + select { + case err := <-flushed: + t.Fatalf("cleanup overtook active original dispatch: %v", err) + case <-time.After(30 * time.Millisecond): + } + close(originator.release) + select { + case err := <-dispatched: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("dispatch did not finish") + } + select { + case err := <-flushed: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal("cleanup did not resume") + } + if err := s.RequireReservedCall(controller.DispatcherID, "rapid-end", 1001); err == nil { + t.Fatal("completed call remained after protected handler exit") + } +} diff --git a/internal/dispatcher/execute.go b/internal/dispatcher/execute.go index 58a75a5..9e740e4 100644 --- a/internal/dispatcher/execute.go +++ b/internal/dispatcher/execute.go @@ -65,6 +65,7 @@ func (c *ExecuteController) ProcessExecute(ctx context.Context, body []byte) err if err := c.validate(); err != nil { return err } + defer c.Store.HoldCallLifecycle()() if err := contract.ValidateCurrent("mq", body); err != nil { return fmt.Errorf("invalid incoming call.execute: %w", err) } @@ -128,6 +129,7 @@ func (c *ExecuteController) ProcessPending(ctx context.Context) error { if err := c.validate(); err != nil { return err } + defer c.Store.HoldCallLifecycle()() commands, err := c.Store.ListPendingExecute(c.DispatcherID) if err != nil { return err @@ -163,6 +165,14 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu logger.Error("Call admission or execution failed", "stage", phase, "error", dispatchErr, "duration_ms", time.Since(started).Milliseconds()) } }() + current, err := c.Store.IsCurrentPending(cmd) + if err != nil { + return fmt.Errorf("recheck pending instruction %q: %w", cmd.EventID, err) + } + if !current { + logger.Info("Buffered call instruction no longer pending", "stage", "pending_batch_skip") + return nil + } logger.Info("Call admission check", "stage", "select_line") at := c.Now() issued, err := time.Parse(time.RFC3339Nano, cmd.IssuedAt) @@ -302,7 +312,8 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu } // FlushOutbox uses the same event identity/body on every retry. Confirming a -// bound queue is not a SaaS application receipt; confirmed rows remain stored. +// bound queue is not a SaaS application receipt. Completed local data may be +// cleaned only after its independent completion conditions are satisfied. func (c *ExecuteController) FlushOutbox(ctx context.Context) error { if err := c.validate(); err != nil { return err @@ -319,7 +330,9 @@ func (c *ExecuteController) FlushOutbox(ctx context.Context) error { } if err := c.Store.MarkOutboxConfirmed(c.DispatcherID, event.EventType, event.EventID); err != nil { failures = append(failures, fmt.Errorf("persist queue confirmation for %q/%q: %w", event.EventType, event.EventID, err)) + continue } + c.cleanupConfirmedFeedback(ctx, event) } return errors.Join(failures...) } diff --git a/internal/dispatcher/runtime.go b/internal/dispatcher/runtime.go index 9882196..f5208e8 100644 --- a/internal/dispatcher/runtime.go +++ b/internal/dispatcher/runtime.go @@ -290,7 +290,10 @@ func (r *Runtime) processPending(ctx context.Context) error { } for _, command := range pending { cmd := command - if err := r.withTask(cmd.TaskID, func() error { return r.Execute.dispatchPending(ctx, cmd) }); err != nil { + if err := r.withTask(cmd.TaskID, func() error { + defer r.Bootstrap.Store.HoldCallLifecycle()() + return r.Execute.dispatchPending(ctx, cmd) + }); err != nil { return fmt.Errorf("retry eligible pending instruction %q: %w", cmd.EventID, err) } } diff --git a/internal/store/calls.go b/internal/store/calls.go index e55680a..bdf8f1f 100644 --- a/internal/store/calls.go +++ b/internal/store/calls.go @@ -117,6 +117,20 @@ func (s *Store) RecordExecute(cmd ExecuteCommand) (Execution, bool, error) { return found, affected == 1, nil } +// IsCurrentPending rejects a detached batch item that has been cancelled, +// completed, cleaned, or replaced since the batch was read. +func (s *Store) IsCurrentPending(cmd ExecuteCommand) (bool, error) { + var current ExecuteCommand + err := s.db.QueryRow(`SELECT dispatcher_id,event_id,tenant_id,task_id,callee,issued_at FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=? AND status='pending'`, cmd.DispatcherID, cmd.EventID).Scan(¤t.DispatcherID, ¤t.EventID, ¤t.TenantID, ¤t.TaskID, ¤t.Callee, ¤t.IssuedAt) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("inspect current pending call: %w", err) + } + return current == cmd, nil +} + // ReserveExecute is the final transactional quota/fencing boundary before // sending a single originate. A crash or timeout after this point leaves an // occupied dispatching/unknown execution, never an automatic redial. diff --git a/internal/store/cleanup.go b/internal/store/cleanup.go new file mode 100644 index 0000000..300a9b3 --- /dev/null +++ b/internal/store/cleanup.go @@ -0,0 +1,275 @@ +package store + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + + "git.ipao.vip/rogee/go-sip/internal/contract" +) + +// CleanupResult reports deleted logical data, not a reduction in SQLite's +// physical size. Completed pages may be reused without shrinking the file. +type CleanupResult struct { + DeletedRows int + DataBytes int64 + RetainedReason string +} + +// HoldCallLifecycle protects in-flight handlers and detached pending batches +// from cleanup. It does not create a lasting execution identity or journal. +func (s *Store) HoldCallLifecycle() func() { + s.cleanupMu.RLock() + return s.cleanupMu.RUnlock +} + +type cleanupFeedback struct { + EventID string `json:"event_id"` + EventType string `json:"event_type"` + DispatcherID string `json:"dispatcher_id"` + TenantID int64 `json:"tenant_id"` + IssuedAt string `json:"issued_at"` + Payload struct { + Status string `json:"status"` + TaskID string `json:"task_id"` + Action string `json:"action"` + TaskState string `json:"task_state"` + Callee string `json:"callee"` + TrunkID string `json:"trunk_id"` + Recording json.RawMessage `json:"recording"` + } `json:"payload"` +} + +func confirmedCleanupFeedback(tx *sql.Tx, dispatcherID, eventType, eventID string) (cleanupFeedback, []byte, bool, error) { + var body []byte + var confirmed int + var confirmedAt sql.NullString + err := tx.QueryRow(`SELECT body,confirmed,confirmed_at FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, eventType, eventID).Scan(&body, &confirmed, &confirmedAt) + if errors.Is(err, sql.ErrNoRows) { + return cleanupFeedback{}, nil, false, nil + } + if err != nil { + return cleanupFeedback{}, nil, false, fmt.Errorf("inspect cleanup feedback: %w", err) + } + if confirmed != 1 { + return cleanupFeedback{}, nil, false, nil + } + if !confirmedAt.Valid || confirmedAt.String == "" { + return cleanupFeedback{}, nil, false, errors.New("confirmed feedback lacks its confirmation time") + } + if err := contract.ValidateCurrent("mq", body); err != nil { + return cleanupFeedback{}, nil, false, fmt.Errorf("invalid cleanup feedback: %w", err) + } + var event cleanupFeedback + if err := json.Unmarshal(body, &event); err != nil { + return event, nil, false, err + } + if event.DispatcherID != dispatcherID || event.EventID != eventID || event.EventType != eventType { + return event, nil, false, errors.New("cleanup feedback envelope differs from its durable identity") + } + return event, body, true, nil +} + +// CleanupCall rechecks completion in the deletion transaction. The final-result +// row is committed only after all fallible ReportCallResult validation succeeds; +// no new Agent acknowledgment protocol or database layout is required. +func (s *Store) CleanupCall(dispatcherID, eventID string) (result CleanupResult, err error) { + if dispatcherID == "" || eventID == "" { + return result, errors.New("cleanup requires Dispatcher and call identity") + } + s.cleanupMu.Lock() + defer s.cleanupMu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return result, fmt.Errorf("begin call cleanup: %w", err) + } + defer tx.Rollback() + var tenantID int64 + var state, taskID, callee, trunkID, issuedAt string + var snapshot []byte + err = tx.QueryRow(`SELECT tenant_id,task_id,callee,status,COALESCE(selected_trunk_id,''),issued_at,snapshot_json FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&tenantID, &taskID, &callee, &state, &trunkID, &issuedAt, &snapshot) + if errors.Is(err, sql.ErrNoRows) { + result.RetainedReason = "no_retained_execution" + return result, nil + } + if err != nil { + return result, fmt.Errorf("inspect call cleanup identity: %w", err) + } + if state != "finished" && state != "rejected" && state != "suppressed" { + result.RetainedReason = "execution_" + state + return result, nil + } + var ackBody, resultBody []byte + if state == "suppressed" { + if trunkID != "" || len(snapshot) != 0 { + return result, errors.New("suppressed call contains evidence of an issued reservation") + } + var references int + if err := tx.QueryRow(`SELECT (SELECT COUNT(*) FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?)+(SELECT COUNT(*) FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=? AND event_type IN ('call.execute','call.execute.result'))`, dispatcherID, eventID, dispatcherID, eventID).Scan(&references); err != nil { + return result, err + } + if references != 0 { + return result, errors.New("suppressed call contains conflicting upload or call feedback") + } + ready, err := confirmedStoppedTask(tx, dispatcherID, tenantID, taskID) + if err != nil { + return result, err + } + if !ready { + result.RetainedReason = "stop_not_confirmed" + return result, nil + } + } else { + ack, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "call.execute", eventID) + if err != nil { + return result, err + } + if !ready { + result.RetainedReason = "call_receipt_not_confirmed" + return result, nil + } + if ack.TenantID != tenantID || ack.IssuedAt != issuedAt { + return result, errors.New("call cleanup receipt differs from its accepted command") + } + ackBody = body + if state == "rejected" { + if ack.Payload.Status != "rejected" { + return result, errors.New("rejected call has a non-rejection receipt") + } + var conflicting int + if err := tx.QueryRow(`SELECT (SELECT COUNT(*) FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?)+(SELECT COUNT(*) FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=? AND event_type='call.execute.result')`, dispatcherID, eventID, dispatcherID, eventID).Scan(&conflicting); err != nil { + return result, err + } + if conflicting != 0 { + return result, errors.New("rejected call has recording or final-result evidence") + } + // File evidence cannot lose its confirmed receipt before its own + // coordinated cleanup has completed. + entries, err := os.ReadDir(filepath.Join(filepath.Dir(s.path), "rejected-data")) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return result, fmt.Errorf("inspect rejection evidence before call cleanup: %w", err) + } + if len(entries) != 0 { + result.RetainedReason = "rejection_files_require_cleanup" + return result, nil + } + } else { + if ack.Payload.Status != "dispatched" || trunkID == "" || len(snapshot) == 0 { + return result, errors.New("finished call lacks its frozen dispatch evidence") + } + final, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "call.execute.result", eventID) + if err != nil { + return result, err + } + if !ready { + result.RetainedReason = "final_result_not_confirmed" + return result, nil + } + if final.TenantID != tenantID || final.Payload.TaskID != taskID || final.Payload.Callee != callee || final.Payload.TrunkID != trunkID { + return result, errors.New("final cleanup result differs from the finished call") + } + if err := validateCleanupRecording(tx, dispatcherID, eventID, final.Payload.Recording); err != nil { + return result, err + } + resultBody = body + } + } + queries := []struct { + query string + args []any + }{ + {`DELETE FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?`, []any{dispatcherID, eventID}}, + {`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=? AND event_type IN ('call.execute','call.execute.result')`, []any{dispatcherID, eventID}}, + {`DELETE FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=? AND status=?`, []any{dispatcherID, eventID, state}}, + } + for _, deletion := range queries { + affected, err := tx.Exec(deletion.query, deletion.args...) + if err != nil { + return CleanupResult{}, fmt.Errorf("delete completed call data: %w", err) + } + n, err := affected.RowsAffected() + if err != nil { + return CleanupResult{}, err + } + result.DeletedRows += int(n) + } + if err := tx.Commit(); err != nil { + return CleanupResult{}, fmt.Errorf("commit completed call cleanup: %w", err) + } + result.DataBytes = int64(len(snapshot) + len(ackBody) + len(resultBody)) + return result, nil +} + +func confirmedStoppedTask(tx *sql.Tx, dispatcherID string, tenantID int64, taskID string) (bool, error) { + var stopped int + if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND control_state='stopped'`, dispatcherID, tenantID, taskID).Scan(&stopped); err != nil { + return false, err + } + if stopped == 0 { + return false, nil + } + rows, err := tx.Query(`SELECT event_id FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND confirmed=1 + AND CASE WHEN json_valid(body) THEN json_extract(body,'$.tenant_id') END=? + AND CASE WHEN json_valid(body) THEN json_extract(body,'$.payload.task_id') END=?`, dispatcherID, tenantID, taskID) + if err != nil { + return false, err + } + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return false, err + } + ids = append(ids, id) + } + err = rows.Err() + rows.Close() + if err != nil { + return false, err + } + for _, id := range ids { + event, _, confirmed, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", id) + if err != nil { + return false, err + } + if confirmed && event.TenantID == tenantID && event.Payload.TaskID == taskID && event.Payload.Action == "stop" && event.Payload.Status == "applied" && event.Payload.TaskState == "stopped" { + return true, nil + } + } + return false, nil +} + +func validateCleanupRecording(tx *sql.Tx, dispatcherID, eventID string, raw json.RawMessage) error { + grant, err := loadRecordingUpload(tx.QueryRow(`SELECT dispatcher_id,source_event_id,upload_id,recording_id,bucket,object_key,checksum_sha256,size_bytes,format,channels,sample_rate_hz,duration_ms,COALESCE(confirmed_at,'') FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?`, dispatcherID, eventID)) + var recording struct { + Status string `json:"status"` + Bucket string `json:"bucket"` + ObjectKey string `json:"object_key"` + Checksum string `json:"checksum_sha256"` + Size int64 `json:"size_bytes"` + Format string `json:"format"` + Channels int `json:"channels"` + SampleRate int `json:"sample_rate_hz"` + Duration int64 `json:"duration_ms"` + } + if decodeErr := json.Unmarshal(raw, &recording); decodeErr != nil { + return fmt.Errorf("decode final cleanup recording: %w", decodeErr) + } + if errors.Is(err, ErrUploadNotFound) { + if recording.Status != "" { + return errors.New("final cleanup recording has no original upload binding") + } + return nil + } + if err != nil { + return err + } + if grant.ConfirmedAt == "" || recording.Status != "uploaded" || recording.Bucket != grant.Bucket || recording.ObjectKey != grant.ObjectKey || recording.Checksum != grant.ChecksumSHA256 || recording.Size != grant.SizeBytes || recording.Format != grant.Format || recording.Channels != grant.Channels || recording.SampleRate != grant.SampleRateHz || recording.Duration != grant.DurationMS { + return errors.New("cleanup recording differs from its confirmed original upload") + } + return nil +} diff --git a/internal/store/cleanup_control.go b/internal/store/cleanup_control.go new file mode 100644 index 0000000..ebc3066 --- /dev/null +++ b/internal/store/cleanup_control.go @@ -0,0 +1,172 @@ +package store + +import ( + "context" + "errors" + "fmt" + "log/slog" +) + +const cleanupBatchSize = 100 + +// CleanupControlReply leaves durable task state untouched. Rejection evidence +// has its own file/receipt cleanup; a successful stop first drains cancelled rows. +func (s *Store) CleanupControlReply(ctx context.Context, dispatcherID, eventID string) (CleanupResult, error) { + event, ready, err := s.cleanupControlIdentity(dispatcherID, eventID) + if err != nil { + return CleanupResult{}, err + } + if !ready { + return CleanupResult{RetainedReason: "control_reply_not_confirmed"}, nil + } + if event.Payload.Status != "applied" { + return CleanupResult{RetainedReason: "rejection_files_require_cleanup"}, nil + } + if event.Payload.Action == "stop" { + return s.CleanupStoppedTask(ctx, dispatcherID, eventID) + } + s.cleanupMu.Lock() + defer s.cleanupMu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return CleanupResult{}, err + } + defer tx.Rollback() + current, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", eventID) + if err != nil { + return CleanupResult{}, err + } + if !ready { + return CleanupResult{RetainedReason: "control_reply_not_confirmed"}, nil + } + if current.TenantID != event.TenantID || current.Payload.TaskID != event.Payload.TaskID || current.Payload.Action != event.Payload.Action || current.Payload.Status != event.Payload.Status || current.Payload.TaskState != event.Payload.TaskState { + return CleanupResult{}, errors.New("control reply changed during cleanup") + } + deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=? AND confirmed=1`, dispatcherID, eventID) + if err != nil { + return CleanupResult{}, err + } + n, err := deleted.RowsAffected() + if err != nil { + return CleanupResult{}, err + } + if err := tx.Commit(); err != nil { + return CleanupResult{}, err + } + return CleanupResult{DeletedRows: int(n), DataBytes: int64(len(body))}, nil +} + +// CleanupStoppedTask deletes each cancelled call in its own transaction. The +// stop receipt remains durable across interrupted batches and individual faults. +func (s *Store) CleanupStoppedTask(ctx context.Context, dispatcherID, stopEventID string) (total CleanupResult, cleanupErr error) { + if err := ctx.Err(); err != nil { + return total, err + } + event, ready, err := s.cleanupControlIdentity(dispatcherID, stopEventID) + if err != nil { + return total, err + } + if !ready { + total.RetainedReason = "stop_not_confirmed" + return total, nil + } + if event.Payload.Action != "stop" || event.Payload.Status != "applied" || event.Payload.TaskState != "stopped" { + return total, errors.New("cancelled-call cleanup requires an applied stop receipt") + } + var failures []error + cursor := "" + for { + if err := ctx.Err(); err != nil { + return total, errors.Join(append(failures, err)...) + } + rows, err := s.db.Query(`SELECT event_id FROM dispatcher_inbox WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status='suppressed' AND event_id>? ORDER BY event_id LIMIT ?`, dispatcherID, event.TenantID, event.Payload.TaskID, cursor, cleanupBatchSize) + if err != nil { + return total, errors.Join(append(failures, err)...) + } + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return total, errors.Join(append(failures, err)...) + } + ids = append(ids, id) + } + err = rows.Err() + rows.Close() + if err != nil { + return total, errors.Join(append(failures, err)...) + } + for _, id := range ids { + if err := ctx.Err(); err != nil { + return total, errors.Join(append(failures, err)...) + } + result, err := s.CleanupCall(dispatcherID, id) + total.DeletedRows += result.DeletedRows + total.DataBytes += result.DataBytes + if err != nil { + slog.Error("Cancelled call cleanup failed; stop proof retained", "dispatcher_id", dispatcherID, "event_id", id, "task_id", event.Payload.TaskID, "error", err) + failures = append(failures, fmt.Errorf("cancelled call %q: %w", id, err)) + } + } + if len(ids) < cleanupBatchSize { + break + } + cursor = ids[len(ids)-1] + } + s.cleanupMu.Lock() + defer s.cleanupMu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return total, errors.Join(append(failures, err)...) + } + defer tx.Rollback() + current, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", stopEventID) + if err != nil { + return total, errors.Join(append(failures, err)...) + } + if !ready { + total.RetainedReason = "stop_not_confirmed" + return total, errors.Join(failures...) + } + if current.TenantID != event.TenantID || current.Payload.TaskID != event.Payload.TaskID || current.Payload.Action != "stop" || current.Payload.Status != "applied" || current.Payload.TaskState != "stopped" { + return total, errors.Join(append(failures, errors.New("stop receipt changed during cancelled-call cleanup"))...) + } + var stopped, remaining int + if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND control_state='stopped'`, dispatcherID, event.TenantID, event.Payload.TaskID).Scan(&stopped); err != nil { + return total, errors.Join(append(failures, err)...) + } + if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_inbox WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status IN ('suppressed','pending')`, dispatcherID, event.TenantID, event.Payload.TaskID).Scan(&remaining); err != nil { + return total, errors.Join(append(failures, err)...) + } + if stopped != 1 || remaining != 0 { + total.RetainedReason = "cancelled_calls_not_fully_cleaned" + return total, errors.Join(failures...) + } + deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=? AND confirmed=1`, dispatcherID, stopEventID) + if err != nil { + return total, errors.Join(append(failures, err)...) + } + n, err := deleted.RowsAffected() + if err != nil { + return total, errors.Join(append(failures, err)...) + } + if err := tx.Commit(); err != nil { + return total, errors.Join(append(failures, err)...) + } + total.DeletedRows += int(n) + total.DataBytes += int64(len(body)) + return total, errors.Join(failures...) +} + +func (s *Store) cleanupControlIdentity(dispatcherID, eventID string) (cleanupFeedback, bool, error) { + s.cleanupMu.Lock() + defer s.cleanupMu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return cleanupFeedback{}, false, err + } + defer tx.Rollback() + event, _, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", eventID) + return event, ready, err +} diff --git a/internal/store/cleanup_control_test.go b/internal/store/cleanup_control_test.go new file mode 100644 index 0000000..159f768 --- /dev/null +++ b/internal/store/cleanup_control_test.go @@ -0,0 +1,121 @@ +package store + +import ( + "context" + "fmt" + "testing" +) + +func TestCleanupStoppedTaskBatchesAndKeepsDurableStopState(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("cancelled-000") + for i := 0; i < 205; i++ { + cmd.EventID = fmt.Sprintf("cancelled-%03d", i) + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + } + if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-batch"); err != nil { + t.Fatal(err) + } + confirmCleanupFeedback(t, s, "stop-batch", "task.control") + result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-batch") + if err != nil || result.DeletedRows != 206 { + t.Fatalf("batched cleanup=%+v err=%v", result, err) + } + requireCleanupCount(t, s, "dispatcher_inbox", 0) + requireCleanupCount(t, s, "dispatcher_outbox", 0) + requireCleanupCount(t, s, "dispatcher_tasks", 1) + if admitted, err := s.CanAdmit(cmd.DispatcherID, cmd.TenantID, cmd.TaskID); err != nil || admitted { + t.Fatalf("stopped task reopened: %t %v", admitted, err) + } +} + +func TestCleanupStoppedTaskKeepsStopProofAcrossPartialFailure(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("cancelled-bad") + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + cmd.EventID = "cancelled-good" + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-partial"); err != nil { + t.Fatal(err) + } + confirmCleanupFeedback(t, s, "stop-partial", "task.control") + if _, err := s.db.Exec(`CREATE TRIGGER cleanup_one_fault BEFORE DELETE ON dispatcher_inbox WHEN OLD.event_id='cancelled-bad' BEGIN SELECT RAISE(ABORT,'one call failure'); END`); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-partial"); err == nil || result.DeletedRows != 1 { + t.Fatalf("independent cancelled call was not cleaned: %+v %v", result, err) + } + requireCleanupCount(t, s, "dispatcher_inbox", 1) + requireCleanupCount(t, s, "dispatcher_outbox", 1) + if _, err := s.db.Exec(`DROP TRIGGER cleanup_one_fault`); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-partial"); err != nil || result.DeletedRows != 2 { + t.Fatalf("partial stop cleanup did not resume: %+v %v", result, err) + } +} + +func TestCleanupStoppedTaskPreservesAcknowledgmentWhenInterrupted(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("cancelled-context") + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-context"); err != nil { + t.Fatal(err) + } + confirmCleanupFeedback(t, s, "stop-context", "task.control") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := s.CleanupStoppedTask(ctx, cmd.DispatcherID, "stop-context"); err == nil { + t.Fatal("interruption was ignored") + } + requireCleanupCount(t, s, "dispatcher_inbox", 1) + requireCleanupCount(t, s, "dispatcher_outbox", 1) + if result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-context"); err != nil || result.DeletedRows != 2 { + t.Fatalf("restart-style retry=%+v err=%v", result, err) + } +} + +func TestCleanupPendingBatchSeesCancellationAndChangedCommand(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("buffered-stop") + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + pending, err := s.ListPendingExecute(cmd.DispatcherID) + if err != nil || len(pending) != 1 { + t.Fatalf("pending=%v err=%v", pending, err) + } + if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil { + t.Fatal(err) + } + if current, err := s.IsCurrentPending(pending[0]); err != nil || current { + t.Fatalf("cancelled batch remained eligible: %t %v", current, err) + } + if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-buffered"); err != nil { + t.Fatal(err) + } + confirmCleanupFeedback(t, s, "stop-buffered", "task.control") + if _, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-buffered"); err != nil { + t.Fatal(err) + } + if current, err := s.IsCurrentPending(pending[0]); err != nil || current { + t.Fatalf("cleaned batch remained eligible: %t %v", current, err) + } +} diff --git a/internal/store/cleanup_test.go b/internal/store/cleanup_test.go new file mode 100644 index 0000000..c6cffc3 --- /dev/null +++ b/internal/store/cleanup_test.go @@ -0,0 +1,211 @@ +package store + +import ( + "database/sql" + "errors" + "testing" + "time" +) + +func confirmCleanupFeedback(t *testing.T, s *Store, id, eventType string) { + t.Helper() + if err := s.MarkOutboxConfirmed(currentCall(id).DispatcherID, eventType, id); err != nil { + t.Fatal(err) + } +} + +func requireCleanupCount(t *testing.T, s *Store, table string, expected int) { + t.Helper() + var count int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil || count != expected { + t.Fatalf("%s rows=%d expected=%d err=%v", table, count, expected, err) + } +} + +func TestCleanupCallNeedsSuccessfulResultAndBothTypedConfirms(t *testing.T) { + for _, scenario := range []struct { + name string + result, ackConfirm, resultConfirm bool + deleted int + }{ + {"no_result_rpc", false, true, false, 0}, + {"no_dispatch_confirm", true, false, true, 0}, + {"no_result_confirm", true, true, false, 0}, + {"all_complete", true, true, true, 3}, + } { + t.Run(scenario.name, func(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "numberID-cleanup", true) + if scenario.result { + if _, _, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, currentResultPayload(t)); err != nil { + t.Fatal(err) + } + } + if scenario.ackConfirm { + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute") + } + if scenario.resultConfirm { + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute.result") + } + result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID) + if err != nil || result.DeletedRows != scenario.deleted { + t.Fatalf("cleanup=%+v err=%v", result, err) + } + requireCleanupCount(t, s, "dispatcher_tasks", 1) + requireCleanupCount(t, s, "dispatcher_configs", 1) + if scenario.deleted == 0 { + requireCleanupCount(t, s, "dispatcher_inbox", 1) + if result.RetainedReason == "" { + t.Fatal("retained call has no reason") + } + return + } + requireCleanupCount(t, s, "dispatcher_inbox", 0) + requireCleanupCount(t, s, "dispatcher_outbox", 0) + if repeated, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || repeated.DeletedRows != 0 { + t.Fatalf("repeat cleanup=%+v err=%v", repeated, err) + } + if _, inserted, err := s.RecordExecute(cmd); err != nil || !inserted { + t.Fatalf("same event ID was not reusable: inserted=%t err=%v", inserted, err) + } + }) + } +} + +func TestCleanupRejectedNeedsOnlyItsConfirmedRefusal(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("rejected-cleanup") + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + if err := s.RejectExecute(cmd.DispatcherID, cmd.EventID, "not issued"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 { + t.Fatalf("unconfirmed refusal removed: %+v %v", result, err) + } + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute") + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 2 { + t.Fatalf("refusal cleanup=%+v err=%v", result, err) + } + requireCleanupCount(t, s, "dispatcher_inbox", 0) + requireCleanupCount(t, s, "dispatcher_outbox", 0) +} + +func TestCleanupNeverRemovesUnfinishedOrUnknownCalls(t *testing.T) { + for _, state := range []string{"pending", "dispatching", "dispatched", "unknown"} { + t.Run(state, func(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("retain-" + state) + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + if state != "pending" { + if err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, currentReservation(), time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)); err != nil { + t.Fatal(err) + } + } + if state == "dispatched" { + if err := s.MarkExecuteDispatched(cmd.DispatcherID, cmd.EventID); err != nil { + t.Fatal(err) + } + } + if state == "unknown" { + if err := s.MarkExecuteUnknown(cmd.DispatcherID, cmd.EventID); err != nil { + t.Fatal(err) + } + } + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 || result.RetainedReason == "" { + t.Fatalf("unfinished cleanup=%+v err=%v", result, err) + } + requireCleanupCount(t, s, "dispatcher_inbox", 1) + }) + } +} + +func TestCleanupCallRollbackPreservesRecordingAndOriginalFacts(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "cleanup-rollback", true) + binding := currentUploadBinding(cmd) + if _, _, err := s.BindRecordingUpload(binding); err != nil { + t.Fatal(err) + } + if _, _, err := s.RecordUploadedCallResult(cmd.DispatcherID, cmd.EventID, currentUploadedResultPayload(t, binding), currentUploadProof(binding)); err != nil { + t.Fatal(err) + } + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute") + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute.result") + if _, err := s.db.Exec(`CREATE TRIGGER cleanup_fault BEFORE DELETE ON dispatcher_inbox BEGIN SELECT RAISE(ABORT,'cleanup interrupted'); END`); err != nil { + t.Fatal(err) + } + if _, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err == nil { + t.Fatal("cleanup swallowed SQLite deletion failure") + } + requireCleanupCount(t, s, "dispatcher_recordings", 1) + requireCleanupCount(t, s, "dispatcher_outbox", 2) + requireCleanupCount(t, s, "dispatcher_inbox", 1) + if _, err := s.db.Exec(`DROP TRIGGER cleanup_fault`); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 4 { + t.Fatalf("retry cleanup=%+v err=%v", result, err) + } + requireCleanupCount(t, s, "dispatcher_recordings", 0) +} + +func TestCleanupSuppressedRequiresCompletedConfirmedStop(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentCall("cancelled-cleanup") + if _, _, err := s.RecordExecute(cmd); err != nil { + t.Fatal(err) + } + if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 { + t.Fatalf("stopping task cleaned early: %+v %v", result, err) + } + if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-cleanup"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 { + t.Fatalf("unconfirmed stop cleaned early: %+v %v", result, err) + } + confirmCleanupFeedback(t, s, "stop-cleanup", "task.control") + if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 1 { + t.Fatalf("cancelled cleanup=%+v err=%v", result, err) + } + requireCleanupCount(t, s, "dispatcher_outbox", 1) // stop proof belongs to the task sweep + requireCleanupCount(t, s, "dispatcher_tasks", 1) + if admitted, err := s.CanAdmit(cmd.DispatcherID, cmd.TenantID, cmd.TaskID); err != nil || admitted { + t.Fatalf("stop barrier lost: %t %v", admitted, err) + } + if _, _, err := s.RecordExecute(cmd); !errors.Is(err, ErrStopped) { + t.Fatalf("cleaned stopped command was readmitted: %v", err) + } +} + +func TestCleanupCallRefusesMismatchedFinalIdentity(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "cleanup-corrupt", true) + if _, _, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, currentResultPayload(t)); err != nil { + t.Fatal(err) + } + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute") + confirmCleanupFeedback(t, s, cmd.EventID, "call.execute.result") + if _, err := s.db.Exec(`UPDATE dispatcher_outbox SET body=json_set(body,'$.payload.task_id','other-task') WHERE event_type='call.execute.result'`); err != nil { + t.Fatal(err) + } + if _, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err == nil { + t.Fatal("conflicting result identity was deleted") + } + requireCleanupCount(t, s, "dispatcher_outbox", 2) + requireCleanupCount(t, s, "dispatcher_inbox", 1) + if _, _, err := s.RecordExecute(currentCall("neighbor")); err != nil { + t.Fatal(err) + } + var state string + if err := s.db.QueryRow(`SELECT status FROM dispatcher_inbox WHERE event_id='neighbor'`).Scan(&state); errors.Is(err, sql.ErrNoRows) || err != nil || state != "pending" { + t.Fatalf("neighbor changed: state=%s err=%v", state, err) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index be3837d..49cae72 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "strings" + "sync" "git.ipao.vip/rogee/go-sip/internal/configread" _ "modernc.org/sqlite" @@ -17,8 +18,9 @@ import ( // Store refuses an incompatible database instead of clearing or // replaying data whose owner and message identity cannot be proved safe. type Store struct { - db *sql.DB - path string + db *sql.DB + path string + cleanupMu sync.RWMutex } var (