diff --git a/docs/evidence/sqlite-cleanup-v0.1-local.md b/docs/evidence/sqlite-cleanup-v0.1-local.md index 4734fee..0b0d835 100644 --- a/docs/evidence/sqlite-cleanup-v0.1-local.md +++ b/docs/evidence/sqlite-cleanup-v0.1-local.md @@ -38,6 +38,15 @@ - 既有恢复扫描能续处理 delivered 中间状态,避免清理后同次遍历把已删除音频误报成孤立文件;单条坏目录不妨碍其他合法记录。没有新增十秒全目录清理任务,既有恢复调度保持原用途。 - 信息/目录删除失败后重启、未完成引用、无音频的坏回执、未知文件、跨实例并发及互斥清退测试通过;`go test -race ./internal/agent ./internal/rpc ./cmd/sip-go-agent` 通过。 +### 已确认拒绝原始文件 + +- 新增资格/事务/文件失败反例后实现。call.execute、task.control、sip.config 结构拒绝须有对应 typed publisher confirm;无回执、未确认、非拒绝及已接纳执行的冲突不会删原始文件。 +- 按接收 Dispatcher、原类型/ID、租户及完整回执字节关联,验证原消息身份、记录结构、私有普通文件和完整记录 SHA-256。关联或内容冲突保留本组原件及完成依据。 +- 新文件名仅增加回执正文 SHA-256 前缀,保证即使本条记录无法解析也能保留正确回执、而不阻止独立组;这是内部磁盘索引,不新增业务编号或协议字段。现有仅记录 SHA-256 命名的文件仍以完整内容验证,不改写或迁移;旧文件若连关联身份也无法读取,保留文件及潜在完成依据,日志明确要求人工核验,不猜测删除。 +- 一回执多原件先逐条删除并同步目录,全部完成后才删回执;文件删除半途失败和原件删完后回执事务失败都能补收尾。未生成回执的孤立原件、未确认组、未知名称和既存 .pending 临时文件不删。 +- RejectData 写入和清理共享生命周期排除;清理结束后同源再来的新拒绝保留新原件和未确认回执,不误删为旧记录。移除了此前 admitted rejected 的“只要整个拒绝目录非空就保留”保护,改为精确本组资格。 +- MQ 确认后的真实控制器入口已验证三类拒绝,未产生执行或伪造最终结果;多原件、两阶段失败、损坏组与独立组、既有文件、未知临时文件和并发写入回归通过。`go test -race ./internal/store ./internal/dispatcher ./internal/rpc` 通过。 + ## 尚待完成 -拒绝原始文件的准确关联与续处理、午夜补扫,以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成,不代表真实 SaaS、通话或生产验收。 +午夜补扫,以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成,不代表真实 SaaS、通话或生产验收。 diff --git a/internal/dispatcher/cleanup.go b/internal/dispatcher/cleanup.go index 5a89ec2..47860ad 100644 --- a/internal/dispatcher/cleanup.go +++ b/internal/dispatcher/cleanup.go @@ -15,18 +15,26 @@ func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event switch event.EventType { case "call.execute", "call.execute.result": result, err = c.Store.CleanupCall(c.DispatcherID, event.EventID) + if err == nil && event.EventType == "call.execute" && result.RetainedReason == "no_retained_execution" { + result, err = c.Store.CleanupRejectedFeedback(c.DispatcherID, event.EventType, event.EventID) + } case "task.control": result, err = c.Store.CleanupControlReply(ctx, c.DispatcherID, event.EventID) + if err == nil && result.RetainedReason == "rejection_files_require_cleanup" { + result, err = c.Store.CleanupRejectedFeedback(c.DispatcherID, event.EventType, event.EventID) + } + case "sip.config": + result, err = c.Store.CleanupRejectedFeedback(c.DispatcherID, event.EventType, 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) + logger.Error("Completed data cleanup failed; evidence retained", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes, "deleted_files", result.DeletedFiles, "file_bytes", result.FileBytes, "error", err) return } - if result.DeletedRows != 0 { - logger.Info("Completed data cleaned", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes) + if result.DeletedRows != 0 || result.DeletedFiles != 0 { + logger.Info("Completed data cleaned", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes, "deleted_files", result.DeletedFiles, "file_bytes", result.FileBytes) } else { logger.Debug("Completed data retained", "reason", result.RetainedReason) } diff --git a/internal/dispatcher/cleanup_test.go b/internal/dispatcher/cleanup_test.go index cc5fd87..89119ca 100644 --- a/internal/dispatcher/cleanup_test.go +++ b/internal/dispatcher/cleanup_test.go @@ -3,6 +3,7 @@ package dispatcher import ( "context" "encoding/json" + "errors" "testing" "time" @@ -65,6 +66,34 @@ func TestConfirmedResultCleansCallAndPreservesSourceIDForReuse(t *testing.T) { } } +func TestConfirmedStructuralRejectionIsCleanedWithoutCreatingExecution(t *testing.T) { + for _, eventType := range []string{"call.execute", "task.control", "sip.config"} { + t.Run(eventType, func(t *testing.T) { + controller, originator, publisher, s := newExecuteFixture(t) + body := []byte(`{"event_id":"structural-original","event_type":"` + eventType + `","dispatcher_id":"` + controller.DispatcherID + `","tenant_id":1001,"payload":{}}`) + if err := s.RejectData(controller.DispatcherID, body, errors.New("invalid original")); err != nil { + t.Fatal(err) + } + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + pending, err := s.ListPendingOutbox(controller.DispatcherID) + if err != nil || len(pending) != 0 || len(originator.calls) != 0 || len(publisher.bodies) != 1 { + t.Fatalf("structural cleanup changed execution or delivery: pending=%v err=%v", pending, err) + } + // Repeating this source creates a new receipt after cleanup, not a + // permanent completed-id record or a synthetic final result. + if err := s.RejectData(controller.DispatcherID, body, errors.New("invalid original")); err != nil { + t.Fatal(err) + } + pending, err = s.ListPendingOutbox(controller.DispatcherID) + if err != nil || len(pending) != 1 { + t.Fatalf("source could not be reused after local cleanup: %v %v", pending, err) + } + }) + } +} + 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"} diff --git a/internal/store/cleanup.go b/internal/store/cleanup.go index 300a9b3..6e539b8 100644 --- a/internal/store/cleanup.go +++ b/internal/store/cleanup.go @@ -5,8 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "os" - "path/filepath" "git.ipao.vip/rogee/go-sip/internal/contract" ) @@ -16,6 +14,8 @@ import ( type CleanupResult struct { DeletedRows int DataBytes int64 + DeletedFiles int + FileBytes int64 RetainedReason string } @@ -146,15 +146,10 @@ func (s *Store) CleanupCall(dispatcherID, eventID string) (result CleanupResult, 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 + files, err := s.cleanupRejectionFiles(ack, ackBody) + result.DeletedFiles, result.FileBytes = files.DeletedFiles, files.FileBytes + if err != nil { + return result, err } } else { if ack.Payload.Status != "dispatched" || trunkID == "" || len(snapshot) == 0 { @@ -185,20 +180,22 @@ func (s *Store) CleanupCall(dispatcherID, eventID string) (result CleanupResult, {`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}}, } + deletedRows := 0 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) + return result, fmt.Errorf("delete completed call data: %w", err) } n, err := affected.RowsAffected() if err != nil { - return CleanupResult{}, err + return result, err } - result.DeletedRows += int(n) + deletedRows += int(n) } if err := tx.Commit(); err != nil { - return CleanupResult{}, fmt.Errorf("commit completed call cleanup: %w", err) + return result, fmt.Errorf("commit completed call cleanup: %w", err) } + result.DeletedRows = deletedRows result.DataBytes = int64(len(snapshot) + len(ackBody) + len(resultBody)) return result, nil } diff --git a/internal/store/cleanup_rejected.go b/internal/store/cleanup_rejected.go new file mode 100644 index 0000000..70de5bb --- /dev/null +++ b/internal/store/cleanup_rejected.go @@ -0,0 +1,219 @@ +package store + +import ( + "bytes" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strings" + + "git.ipao.vip/rogee/go-sip/internal/contract" +) + +// CleanupRejectedFeedback covers structural rejection without an admitted call. +// Admitted rejected calls use CleanupCall so their inbox and receipt stay atomic. +func (s *Store) CleanupRejectedFeedback(dispatcherID, eventType, eventID string) (result CleanupResult, cleanupErr error) { + if eventType != "call.execute" && eventType != "task.control" && eventType != "sip.config" { + return result, errors.New("unsupported rejected feedback type") + } + s.cleanupMu.Lock() + defer s.cleanupMu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return result, err + } + defer tx.Rollback() + event, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, eventType, eventID) + if err != nil { + return result, err + } + if !ready { + result.RetainedReason = "rejection_not_confirmed" + return result, nil + } + if event.Payload.Status != "rejected" { + result.RetainedReason = "feedback_is_not_rejection" + return result, nil + } + if eventType == "call.execute" { + var state string + err := tx.QueryRow(`SELECT status FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&state) + if err == nil { + if state != "rejected" { + return result, errors.New("rejection conflicts with a retained execution") + } + result.RetainedReason = "admitted_call_cleanup_required" + return result, nil + } + if !errors.Is(err, sql.ErrNoRows) { + return result, err + } + } + result, err = s.cleanupRejectionFiles(event, body) + if err != nil { + return result, err + } + deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=? AND confirmed=1`, dispatcherID, eventType, eventID) + if err != nil { + return result, fmt.Errorf("delete confirmed rejection receipt after original files: %w", err) + } + n, err := deleted.RowsAffected() + if err != nil { + return result, err + } + if err := tx.Commit(); err != nil { + return result, err + } + result.DeletedRows = int(n) + result.DataBytes = int64(len(body)) + return result, nil +} + +// The caller holds the cleanup write lock and the receipt's DB transaction. +// This excludes RejectData from adding another original before receipt deletion. +func (s *Store) cleanupRejectionFiles(event cleanupFeedback, body []byte) (result CleanupResult, cleanupErr error) { + dir := filepath.Join(filepath.Dir(s.path), "rejected-data") + stat, err := os.Lstat(dir) + if errors.Is(err, os.ErrNotExist) { + return result, nil + } + if err != nil { + return result, err + } + if !stat.IsDir() || stat.Mode()&os.ModeSymlink != 0 || stat.Mode().Perm()&0077 != 0 { + return result, errors.New("rejected evidence directory is not private and physical") + } + files, err := os.ReadDir(dir) + if err != nil { + return result, err + } + var failures []error + receiptSum := sha256.Sum256(body) + receiptKey := hex.EncodeToString(receiptSum[:]) + for _, file := range files { + name := file.Name() + parts := strings.Split(strings.TrimSuffix(name, ".json"), ".") + indexed := len(parts) == 2 + if indexed && parts[0] != receiptKey { + continue + } + if len(parts) != 1 && !indexed { + continue + } + stem := parts[len(parts)-1] + digest, hashErr := hex.DecodeString(stem) + if !strings.HasSuffix(name, ".json") || hashErr != nil || len(digest) != sha256.Size || hex.EncodeToString(digest) != stem { + // Unrecognized/temporary artifacts are outside this cleanup. Do not + // open them, derive a source identity, or delete them. + continue + } + path := filepath.Join(dir, name) + stat, err := os.Lstat(path) + if err != nil { + failures = append(failures, fmt.Errorf("inspect rejection evidence %s: %w", stem, err)) + continue + } + if !stat.Mode().IsRegular() { + failures = append(failures, fmt.Errorf("rejection evidence %s is not a regular file; completion proof retained", stem)) + continue + } + encoded, err := os.ReadFile(path) + if err != nil { + failures = append(failures, fmt.Errorf("read rejection evidence %s: %w", stem, err)) + continue + } + var evidence struct { + Original []byte `json:"Original"` + InvalidSource []byte `json:"InvalidSource"` + Feedback json.RawMessage `json:"Feedback"` + } + decode := json.NewDecoder(bytes.NewReader(encoded)) + decode.DisallowUnknownFields() + decodeErr := decode.Decode(&evidence) + var header cleanupFeedback + feedbackErr := json.Unmarshal(evidence.Feedback, &header) + var original struct { + EventID string `json:"event_id"` + EventType string `json:"event_type"` + DispatcherID string `json:"dispatcher_id"` + TenantID int64 `json:"tenant_id"` + Payload json.RawMessage `json:"payload"` + } + originalErr := json.Unmarshal(evidence.Original, &original) + candidate := indexed || (feedbackErr == nil && header.DispatcherID == event.DispatcherID && header.EventType == event.EventType && header.EventID == event.EventID) || + (originalErr == nil && original.DispatcherID == event.DispatcherID && original.EventType == event.EventType && original.EventID == event.EventID) + if !candidate { + if decodeErr != nil || feedbackErr != nil || originalErr != nil { + slog.Warn("Unidentifiable rejection evidence retained", "file_sha256", stem, "reason", "unknown_source_identity") + failures = append(failures, fmt.Errorf("legacy rejection evidence %s cannot be associated; completion proof retained", stem)) + } + continue + } + fail := func(reason string) { + failures = append(failures, fmt.Errorf("rejection evidence %s: %s", stem, reason)) + } + if decodeErr != nil { + fail("invalid evidence structure") + continue + } + if err := decode.Decode(new(any)); !errors.Is(err, io.EOF) { + fail("trailing evidence data") + continue + } + actualDigest := sha256.Sum256(encoded) + if !bytes.Equal(actualDigest[:], digest) || stat.Mode().Perm()&0077 != 0 { + fail("evidence digest or private permissions differ") + continue + } + if feedbackErr != nil || originalErr != nil || !bytes.Equal(evidence.Feedback, body) { + fail("feedback differs from its confirmed rejection") + continue + } + if original.EventID != event.EventID || original.EventType != event.EventType || original.DispatcherID != event.DispatcherID || (event.EventType != "sip.config" && original.TenantID != event.TenantID) { + fail("original identity differs from its rejection") + continue + } + if event.EventType != "sip.config" && (original.TenantID <= 0 || len(evidence.Original) == 0) { + fail("original tenant identity is incomplete") + continue + } + if contract.ValidateCurrent("mq", evidence.Original) == nil { + var payload map[string]json.RawMessage + if err := json.Unmarshal(original.Payload, &payload); err != nil { + fail("original direction cannot be verified") + continue + } + if _, receipt := payload["status"]; receipt { + fail("original is an outbound receipt, not a command") + continue + } + } + if s.removeRejectedFile != nil { + err = s.removeRejectedFile(path) + } else { + err = os.Remove(path) + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + fail("original file deletion failed: " + err.Error()) + continue + } + if err == nil { + result.DeletedFiles++ + result.FileBytes += stat.Size() + } + slog.Info("Confirmed rejection original removed", "dispatcher_id", event.DispatcherID, "event_type", event.EventType, "event_id", event.EventID, "tenant_id", event.TenantID, "file_sha256", stem, "file_bytes", stat.Size()) + } + // Even a retry with no remaining matched files persists earlier unlink + // operations before allowing the DB receipt transaction to commit. + if err := syncPrivateDirectory(dir); err != nil { + failures = append(failures, fmt.Errorf("persist rejected original deletion: %w", err)) + } + return result, errors.Join(failures...) +} diff --git a/internal/store/cleanup_rejected_test.go b/internal/store/cleanup_rejected_test.go new file mode 100644 index 0000000..a9ab9df --- /dev/null +++ b/internal/store/cleanup_rejected_test.go @@ -0,0 +1,297 @@ +package store + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func rejectedRaw(id, eventType, extra string) []byte { + return []byte(`{"event_id":"` + id + `","event_type":"` + eventType + `","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{},"extra":"` + extra + `"}`) +} + +func rejectionFiles(t *testing.T, s *Store) []string { + t.Helper() + files, err := filepath.Glob(filepath.Join(filepath.Dir(s.path), "rejected-data", "*.json")) + if err != nil { + t.Fatal(err) + } + return files +} + +func TestCleanupRejectedFeedbackNeedsConfirmAndHandlesMultipleOriginals(t *testing.T) { + for _, kind := range []string{"call.execute", "sip.config"} { + t.Run(kind, func(t *testing.T) { + s := preparedCurrentCallStore(t) + for _, extra := range []string{"first", "second"} { + if err := s.RejectData(currentDispatcherID, rejectedRaw("rejection-multiple", kind, extra), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + } + if len(rejectionFiles(t, s)) != 2 { + t.Fatal("fixture needs two originals") + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, kind, "rejection-multiple"); err != nil || result.DeletedFiles != 0 { + t.Fatalf("unconfirmed originals cleaned: %+v %v", result, err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, kind, "rejection-multiple"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, kind, "rejection-multiple"); err != nil || result.DeletedFiles != 2 || result.DeletedRows != 1 || result.FileBytes == 0 { + t.Fatalf("matching rejection cleanup=%+v err=%v", result, err) + } + if len(rejectionFiles(t, s)) != 0 { + t.Fatal("confirmed originals remain") + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, kind, "rejection-multiple"); err != nil || result.DeletedFiles != 0 || result.DeletedRows != 0 { + t.Fatalf("repeated cleanup=%+v %v", result, err) + } + }) + } +} + +func TestCleanupRejectedFeedbackKeepsOrphanOriginalAndIndependentConfirmedGroup(t *testing.T) { + s := preparedCurrentCallStore(t) + if _, err := s.db.Exec(`CREATE TRIGGER reject_insert_fault BEFORE INSERT ON dispatcher_outbox WHEN NEW.event_id='orphan-original' BEGIN SELECT RAISE(ABORT,'receipt failure'); END`); err != nil { + t.Fatal(err) + } + if err := s.RejectData(currentDispatcherID, rejectedRaw("orphan-original", "call.execute", "orphan"), errors.New("invalid command")); err == nil { + t.Fatal("fixture did not fail receipt commit") + } + if _, err := s.db.Exec(`DROP TRIGGER reject_insert_fault`); err != nil { + t.Fatal(err) + } + if len(rejectionFiles(t, s)) != 1 { + t.Fatal("orphan original was lost") + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "orphan-original"); err != nil || result.DeletedFiles != 0 { + t.Fatalf("orphan cleanup=%+v %v", result, err) + } + if err := s.RejectData(currentDispatcherID, rejectedRaw("independent-original", "call.execute", "complete"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "independent-original"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "independent-original"); err != nil || result.DeletedFiles != 1 || result.DeletedRows != 1 { + t.Fatalf("independent group blocked by orphan: %+v %v", result, err) + } + if len(rejectionFiles(t, s)) != 1 { + t.Fatal("orphan original was deleted") + } +} + +func TestCleanupRejectedFilesBeforeReceiptAndRetryAfterBothFailures(t *testing.T) { + s := preparedCurrentCallStore(t) + for _, extra := range []string{"first", "second"} { + if err := s.RejectData(currentDispatcherID, rejectedRaw("rejection-resume", "call.execute", extra), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "rejection-resume"); err != nil { + t.Fatal(err) + } + removed := 0 + s.removeRejectedFile = func(path string) error { + if removed == 1 { + return errors.New("second original deletion failed") + } + removed++ + return os.Remove(path) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "rejection-resume"); err == nil || result.DeletedFiles != 1 { + t.Fatalf("partial original deletion=%+v %v", result, err) + } + requireCleanupCount(t, s, "dispatcher_outbox", 1) + if len(rejectionFiles(t, s)) != 1 { + t.Fatal("failure did not preserve remaining original") + } + s.removeRejectedFile = nil + if _, err := s.db.Exec(`CREATE TRIGGER reject_cleanup_fault BEFORE DELETE ON dispatcher_outbox BEGIN SELECT RAISE(ABORT,'receipt deletion failure'); END`); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "rejection-resume"); err == nil || result.DeletedFiles != 1 { + t.Fatalf("receipt deletion failure=%+v %v", result, err) + } + if len(rejectionFiles(t, s)) != 0 { + t.Fatal("originals were not deleted before receipt") + } + requireCleanupCount(t, s, "dispatcher_outbox", 1) + if _, err := s.db.Exec(`DROP TRIGGER reject_cleanup_fault`); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "rejection-resume"); err != nil || result.DeletedRows != 1 { + t.Fatalf("retry after file removal failed: %+v %v", result, err) + } +} + +func TestCleanupRejectedOpaqueFileIsBoundToOnlyItsReceipt(t *testing.T) { + s := preparedCurrentCallStore(t) + if err := s.RejectData(currentDispatcherID, rejectedRaw("opaque-source", "call.execute", "bad"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + file := rejectionFiles(t, s)[0] + if err := os.WriteFile(file, []byte("damaged evidence without readable identity"), 0600); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "opaque-source"); err != nil { + t.Fatal(err) + } + if _, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "opaque-source"); err == nil { + t.Fatal("opaque evidence lost its receipt basis") + } + if err := s.RejectData(currentDispatcherID, rejectedRaw("opaque-independent", "call.execute", "good"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "opaque-independent"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "opaque-independent"); err != nil || result.DeletedRows != 1 { + t.Fatalf("damaged other record blocked independent source: %+v %v", result, err) + } + if _, err := os.Stat(file); err != nil { + t.Fatal("opaque evidence was deleted") + } + requireCleanupCount(t, s, "dispatcher_outbox", 1) +} + +func TestCleanupExistingValidEvidenceWithoutRenamingOrMigration(t *testing.T) { + s := preparedCurrentCallStore(t) + if err := s.RejectData(currentDispatcherID, rejectedRaw("existing-source", "call.execute", "old"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + file := rejectionFiles(t, s)[0] + body, err := os.ReadFile(file) + if err != nil { + t.Fatal(err) + } + digest := sha256.Sum256(body) + old := filepath.Join(filepath.Dir(file), hex.EncodeToString(digest[:])+".json") + if err := os.Rename(file, old); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "existing-source"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "existing-source"); err != nil || result.DeletedFiles != 1 || result.DeletedRows != 1 { + t.Fatalf("existing original not verified correctly: %+v %v", result, err) + } +} + +func TestCleanupRejectedControlAndUnrecognizedArtifacts(t *testing.T) { + s := preparedCurrentCallStore(t) + if err := s.RejectData(currentDispatcherID, rejectedRaw("control-source", "task.control", "bad"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + file := rejectionFiles(t, s)[0] + if len(strings.Split(filepath.Base(file), ".")) != 3 { + t.Fatal("original lacks an immutable receipt binding") + } + unknown := filepath.Join(filepath.Dir(file), "operator-note") + temporary := filepath.Join(filepath.Dir(file), ".pending-existing") + for _, path := range []string{unknown, temporary} { + if err := os.WriteFile(path, []byte("retained"), 0600); err != nil { + t.Fatal(err) + } + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "control-source"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "task.control", "control-source"); err != nil || result.DeletedFiles != 1 || result.DeletedRows != 1 { + t.Fatalf("control rejection=%+v %v", result, err) + } + for _, path := range []string{unknown, temporary} { + if _, err := os.Stat(path); err != nil { + t.Fatal("unrecognized artifact was touched") + } + } +} + +func TestRejectionCaptureCannotRaceOriginalDeletionAndReceiptCommit(t *testing.T) { + s := preparedCurrentCallStore(t) + if err := s.RejectData(currentDispatcherID, rejectedRaw("capture-race", "call.execute", "first"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "capture-race"); err != nil { + t.Fatal(err) + } + entered, release := make(chan struct{}), make(chan struct{}) + s.removeRejectedFile = func(path string) error { close(entered); <-release; return os.Remove(path) } + cleaned := make(chan error, 1) + go func() { + _, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "capture-race") + cleaned <- err + }() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("cleanup did not enter deletion") + } + captured := make(chan error, 1) + go func() { + captured <- s.RejectData(currentDispatcherID, rejectedRaw("capture-race", "call.execute", "next"), errors.New("invalid command")) + }() + select { + case err := <-captured: + close(release) + t.Fatalf("capture passed cleanup's receipt boundary: %v", err) + case <-time.After(30 * time.Millisecond): + } + close(release) + for _, result := range []chan error{cleaned, captured} { + select { + case err := <-result: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("coordinated operation did not finish") + } + } + if len(rejectionFiles(t, s)) != 1 { + t.Fatal("new original was erased with its predecessor") + } + pending, err := s.ListPendingOutbox(currentDispatcherID) + if err != nil || len(pending) != 1 { + t.Fatalf("new original lost its unconfirmed receipt: %v %v", pending, err) + } +} + +func TestCleanupRejectedEvidenceConflictPreservesItsProofNotOtherRecords(t *testing.T) { + s := preparedCurrentCallStore(t) + if err := s.RejectData(currentDispatcherID, rejectedRaw("rejection-corrupt", "call.execute", "corrupt"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + files := rejectionFiles(t, s) + body, err := os.ReadFile(files[0]) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(files[0], append(body, ' '), 0600); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "rejection-corrupt"); err != nil { + t.Fatal(err) + } + if _, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "rejection-corrupt"); err == nil { + t.Fatal("corrupted evidence was deleted") + } + if len(rejectionFiles(t, s)) != 1 { + t.Fatal("corrupt evidence was lost") + } + if err := s.RejectData(currentDispatcherID, rejectedRaw("rejection-good", "call.execute", "complete"), errors.New("invalid command")); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "call.execute", "rejection-good"); err != nil { + t.Fatal(err) + } + if result, err := s.CleanupRejectedFeedback(currentDispatcherID, "call.execute", "rejection-good"); err != nil || result.DeletedRows != 1 { + t.Fatalf("independent proof blocked: %+v %v", result, err) + } + requireCleanupCount(t, s, "dispatcher_outbox", 1) +} diff --git a/internal/store/rejected_data.go b/internal/store/rejected_data.go index f58c87d..6aa84b1 100644 --- a/internal/store/rejected_data.go +++ b/internal/store/rejected_data.go @@ -32,6 +32,7 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er if id.EventType != "task.control" && id.EventType != "call.execute" && id.EventType != "sip.config" { return ErrUnidentifiableData } + defer s.HoldCallLifecycle()() // A structurally valid receipt is not an original command identity. Never // turn a bounced outbound receipt into another receipt. if contract.ValidateCurrent("mq", original) == nil { @@ -106,8 +107,11 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er return err } sum := sha256.Sum256(evidence) + receiptSum := sha256.Sum256(body) + // A receipt fingerprint associates even damaged evidence with its proof; + // it changes only an internal filename, never a business event identity. dir := filepath.Join(filepath.Dir(s.path), "rejected-data") - if err := persistPrivateData(dir, hex.EncodeToString(sum[:])+".json", evidence); err != nil { + if err := persistPrivateData(dir, hex.EncodeToString(receiptSum[:])+"."+hex.EncodeToString(sum[:])+".json", evidence); err != nil { return fmt.Errorf("persist rejected-data evidence: %w", err) } route, err := tenant.ResultRoute(dispatcherID) diff --git a/internal/store/store.go b/internal/store/store.go index 49cae72..349dd90 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -18,9 +18,10 @@ 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 - cleanupMu sync.RWMutex + db *sql.DB + path string + cleanupMu sync.RWMutex + removeRejectedFile func(string) error // package-test deletion fault injection } var (