From 04f7e915cf4ff0140a162c0803272d5d53fc308f Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 11 Oct 2026 19:14:28 +0800 Subject: [PATCH] feat(agent): clean delivered recording files with resumable deletion --- docs/evidence/sqlite-cleanup-v0.1-local.md | 12 +- internal/agent/private_state_lock.go | 44 ++++ internal/agent/recording_cleanup.go | 147 ++++++++++++ internal/agent/recording_cleanup_test.go | 254 +++++++++++++++++++++ internal/agent/recording_recovery.go | 31 ++- internal/agent/recording_retry.go | 21 +- internal/agent/recording_retry_test.go | 10 +- internal/agent/recording_scan.go | 23 +- internal/agent/recording_scan_test.go | 16 +- internal/agent/result_journal.go | 33 +++ internal/agent/upload.go | 3 +- 11 files changed, 569 insertions(+), 25 deletions(-) create mode 100644 internal/agent/private_state_lock.go create mode 100644 internal/agent/recording_cleanup.go create mode 100644 internal/agent/recording_cleanup_test.go diff --git a/docs/evidence/sqlite-cleanup-v0.1-local.md b/docs/evidence/sqlite-cleanup-v0.1-local.md index 52f4b2e..4734fee 100644 --- a/docs/evidence/sqlite-cleanup-v0.1-local.md +++ b/docs/evidence/sqlite-cleanup-v0.1-local.md @@ -28,6 +28,16 @@ - 原 `event_id` 清理后可再次接纳、执行和回传；只删除本通数据，共享配置和任务状态保留。报告的是删除的逻辑行/正文字节，不声称 SQLite 文件立即变小。 - `go test -race ./internal/store ./internal/dispatcher ./internal/rpc` 通过。 +### 已交付录音文件 + +- 先添加反例：原逻辑不能读取音频已删但 `info.json` 仍为 delivered 的合法中间状态，且不会删除已交付文件；实现后通过。 +- 仅持久 delivered、上传成功回执和原身份/长度/校验和等信息有效、且原事件无未完成 `.results` 引用时删除。prepared、ready、end_pending 引用都保留原恢复文件，引用完成后可以继续收尾。 +- 依次删除本条 `recording.wav`、同步目录、删除 `info.json`、同步并删除准确且空的叶目录；共享父目录、`.results`、未知侧文件与 OSS 对象不删除。信息或叶目录删除失败保留/恢复 delivered 依据，下次启动只补删除，不申请授权、PUT 或重报。 +- delivered 且原音频已经缺失时，只放行这一已确认删除的中间状态；其他状态缺音频仍报错。原校验和严格保持有效的小写 SHA-256，上传回执、长度和身份校验不弱化。 +- 同一录音跨不同 recovery 实例使用临时按路径互斥；结果引用写入/确认与删除协调，锁使用结束即移除，不形成永久身份历史。并发实例仍只有一次 PUT、一次回报；完成后不存在该记录，后续直接调用明确返回不存在而非虚构已完成历史。 +- 既有恢复扫描能续处理 delivered 中间状态，避免清理后同次遍历把已删除音频误报成孤立文件；单条坏目录不妨碍其他合法记录。没有新增十秒全目录清理任务，既有恢复调度保持原用途。 +- 信息/目录删除失败后重启、未完成引用、无音频的坏回执、未知文件、跨实例并发及互斥清退测试通过；`go test -race ./internal/agent ./internal/rpc ./cmd/sip-go-agent` 通过。 + ## 尚待完成 -已交付录音、拒绝原始文件的准确关联与续处理、午夜补扫，以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。 +拒绝原始文件的准确关联与续处理、午夜补扫，以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。 diff --git a/internal/agent/private_state_lock.go b/internal/agent/private_state_lock.go new file mode 100644 index 0000000..7650576 --- /dev/null +++ b/internal/agent/private_state_lock.go @@ -0,0 +1,44 @@ +package agent + +import ( + "path/filepath" + "sync" +) + +// Recovery and result-journal instances share a root but are constructed by +// different call/recovery paths. Locks exist only while a path is in use; they +// are not a permanent identity history and do not serialize independent files. +type privateStateLock struct { + mu sync.Mutex + users int +} + +var privateStateLocks = struct { + sync.Mutex + paths map[string]*privateStateLock +}{paths: make(map[string]*privateStateLock)} + +func holdPrivateState(path string) (func(), error) { + key, err := filepath.Abs(path) + if err != nil { + return nil, err + } + privateStateLocks.Lock() + lock := privateStateLocks.paths[key] + if lock == nil { + lock = &privateStateLock{} + privateStateLocks.paths[key] = lock + } + lock.users++ + privateStateLocks.Unlock() + lock.mu.Lock() + return func() { + lock.mu.Unlock() + privateStateLocks.Lock() + lock.users-- + if lock.users == 0 { + delete(privateStateLocks.paths, key) + } + privateStateLocks.Unlock() + }, nil +} diff --git a/internal/agent/recording_cleanup.go b/internal/agent/recording_cleanup.go new file mode 100644 index 0000000..ec5b97c --- /dev/null +++ b/internal/agent/recording_cleanup.go @@ -0,0 +1,147 @@ +package agent + +import ( + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" +) + +type RecordingCleanupResult struct { + DeletedFiles int + DeletedBytes int64 + RetainedReason string +} + +// cleanupDelivered requires the recording path lock. The original delivered +// info remains the completion proof until its audio deletion is durable. +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) + if cleanupErr != nil { + slog.Error("Delivered recording cleanup failed; unfinished evidence retained", append(attrs, "error", cleanupErr)...) + } else if result.RetainedReason != "" { + slog.Info("Delivered recording retained", append(attrs, "reason", result.RetainedReason)...) + } else { + slog.Info("Delivered recording files cleaned", attrs...) + } + }() + if entry.State != "delivered" || entry.Uploaded == nil { + return result, errors.New("recording cleanup requires its persisted delivery receipt") + } + dir, audio, info, err := r.paths(entry.Bucket, entry.ObjectKey) + if err != nil { + return result, err + } + // Never follow a directory link or remove shared parents/unknown sidecars. + 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 != 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") + } + } + journal := ResultJournal{Root: r.Root} + resultPath, err := journal.path(entry.SourceEventID) + if err != nil { + return result, err + } + release, err := holdPrivateState(resultPath) + if err != nil { + return result, err + } + defer release() + pending, _, err := journal.load(entry.SourceEventID) + if err == nil { + if entry.DispatcherID != "" && (pending.DispatcherID != entry.DispatcherID || pending.TenantID != entry.TenantID) { + return result, errors.New("original result reference conflicts with delivered recording ownership") + } + result.RetainedReason = "uncompleted_original_result_reference" + return result, nil + } + 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) { + return nil + } + if err != nil { + return err + } + if !stat.Mode().IsRegular() || stat.Mode().Perm()&0077 != 0 { + return errors.New("recording cleanup file is not an owned private regular file") + } + if r.removeFile != nil { + err = r.removeFile(path) + } else { + err = os.Remove(path) + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + if err == nil { + result.DeletedFiles++ + result.DeletedBytes += stat.Size() + } + 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 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 + } + 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 err := syncDirectory(filepath.Dir(dir)); err != nil { + return result, fmt.Errorf("persist recording leaf deletion: %w", err) + } + return result, nil +} diff --git a/internal/agent/recording_cleanup_test.go b/internal/agent/recording_cleanup_test.go new file mode 100644 index 0000000..416b2af --- /dev/null +++ b/internal/agent/recording_cleanup_test.go @@ -0,0 +1,254 @@ +package agent + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" +) + +func deliveredRecordingFixture(t *testing.T, r *RecordingRecovery) RecordingRecoveryEntry { + t.Helper() + e := failedRecordingFixture() + if _, err := r.SaveFailure([]byte("recording bytes"), e, &UploadHTTPError{StatusCode: 503}); err != nil { + t.Fatal(err) + } + e, err := r.Load(e.Bucket, e.ObjectKey) + if err != nil { + t.Fatal(err) + } + e.State = "delivered" + e.Uploaded = &UploadResult{StatusCode: 200, SizeBytes: e.SizeBytes, SHA256: e.SHA256} + _, _, info, err := r.paths(e.Bucket, e.ObjectKey) + if err != nil { + t.Fatal(err) + } + if err := r.persistState(info, e); err != nil { + t.Fatal(err) + } + return e +} + +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) + dir, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + if err := os.Remove(audio); err != nil { + t.Fatal(err) + } + restarted := &RecordingRecovery{Root: r.Root} + if loaded, err := restarted.Load(e.Bucket, e.ObjectKey); err != nil || loaded.State != "delivered" { + t.Fatalf("valid deletion intermediate state unreadable: %+v %v", loaded, err) + } + calls := 0 + after, err := restarted.Retry(context.Background(), e.Bucket, e.ObjectKey, + func(context.Context, RecordingRecoveryEntry) (RecoveryTarget, error) { + calls++ + return RecoveryTarget{}, errors.New("must not request a grant") + }, + func(context.Context, RecordingRecoveryEntry) error { + calls++ + return errors.New("must not report again") + }) + if err != nil || after.State != "delivered" || calls != 0 { + t.Fatalf("delivered cleanup replayed work: %+v calls=%d err=%v", after, calls, err) + } + for _, path := range []string{dir, audio, info} { + if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("completed recording remains at %q: %v", path, err) + } + } + if _, err := os.Stat(filepath.Dir(dir)); err != nil { + t.Fatal("cleanup deleted a shared parent") + } +} + +func TestDeliveredRecordingRetainedWhileResultReferenceUnfinished(t *testing.T) { + for _, state := range []string{"prepared", "ready", "end_pending"} { + t.Run(state, func(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + j := ResultJournal{Root: r.Root} + entry := ResultEntry{SourceEventID: e.SourceEventID, DispatcherID: "dispatcher-1", TenantID: 1001, Payload: []byte(`{}`)} + if err := j.save(entry, state); err != nil { + t.Fatal(err) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } + _, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + for _, path := range []string{audio, info} { + if _, err := os.Stat(path); err != nil { + t.Fatalf("pending reference lost recovery file: %v", err) + } + } + if state == "prepared" { + if err := j.DiscardPrepared(e.SourceEventID); err != nil { + t.Fatal(err) + } + } else { + if state == "end_pending" { + if err := j.MarkEnded(e.SourceEventID); err != nil { + t.Fatal(err) + } + } + if err := j.Ack(e.SourceEventID); err != nil { + t.Fatal(err) + } + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(info); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("reference-free delivered recording remains: %v", err) + } + }) + } +} + +func TestDeliveredCleanupFailureResumesWithoutPUTOrReport(t *testing.T) { + for _, failedStep := range []string{"info", "directory"} { + t.Run(failedStep, func(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + dir, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + r.removeFile = func(path string) error { + if (failedStep == "info" && path == info) || (failedStep == "directory" && path == dir) { + return errors.New("injected cleanup interruption") + } + return os.Remove(path) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err == nil { + t.Fatal("deletion interruption was swallowed") + } + if _, err := os.Stat(audio); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("original audio deletion did not finish: %v", err) + } + if loaded, err := r.Load(e.Bucket, e.ObjectKey); err != nil || loaded.State != "delivered" { + t.Fatalf("interrupted cleanup lost its completion proof: %+v %v", loaded, err) + } + restarted := &RecordingRecovery{Root: r.Root} + if _, err := restarted.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(dir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("restart did not finish exact-leaf cleanup: %v", err) + } + }) + } +} + +func TestDeliveredMissingAudioStillValidatesOriginalReceipt(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + _, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + if err := os.Remove(audio); err != nil { + t.Fatal(err) + } + e.SHA256 = "abc" + e.Uploaded.SHA256 = "abc" + if err := r.persistState(info, e); err != nil { + t.Fatal(err) + } + if _, err := r.Load(e.Bucket, e.ObjectKey); !errors.Is(err, ErrRecordingRecoveryIncomplete) { + t.Fatalf("malformed original receipt accepted after audio deletion: %v", err) + } + if _, err := os.Stat(info); err != nil { + t.Fatal("malformed evidence was removed") + } +} + +func TestDeliveredCleanupPreservesUnexpectedFilesAndNeighborRecords(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + dir, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + unknown := filepath.Join(dir, "operator-notes") + if err := os.WriteFile(unknown, []byte("retained"), 0600); err != nil { + t.Fatal(err) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err == nil { + t.Fatal("unknown leaf file was silently ignored") + } + for _, path := range []string{unknown, audio, info} { + if _, err := os.Stat(path); err != nil { + t.Fatalf("unknown leaf cleanup deleted data: %v", err) + } + } + if err := os.Remove(unknown); err != nil { + t.Fatal(err) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } +} + +func TestDeliveredRecoveryScanFinishesPartialDeletionDespiteIndependentBadDirectory(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + _, audio, info, _ := r.paths(e.Bucket, e.ObjectKey) + if err := os.Remove(audio); err != nil { + t.Fatal(err) + } + blockedDir := filepath.Join(r.Root, "blocked-bucket") + if err := os.Mkdir(blockedDir, 0700); err != nil { + t.Fatal(err) + } + if err := os.Chmod(blockedDir, 0755); err != nil { + t.Fatal(err) + } + calls := 0 + processed, err := r.RecoverDue(context.Background(), + func(context.Context, RecordingRecoveryEntry) (RecoveryTarget, error) { + calls++ + return RecoveryTarget{}, errors.New("must not request a grant") + }, + func(context.Context, RecordingRecoveryEntry) error { + calls++ + return errors.New("must not report again") + }) + if err == nil || processed != 0 || calls != 0 { + t.Fatalf("independent bad directory handling: processed=%d calls=%d err=%v", processed, calls, err) + } + if _, err := os.Stat(info); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("delivered intermediate state was not cleaned: %v", err) + } + if _, err := os.Stat(blockedDir); err != nil { + t.Fatal("unknown directory was removed") + } +} + +func TestRecordingCleanupLeavesNoPermanentInMemoryLocks(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + e := deliveredRecordingFixture(t, r) + dir, _, _, _ := r.paths(e.Bucket, e.ObjectKey) + resultPath, err := (ResultJournal{Root: r.Root}).path(e.SourceEventID) + if err != nil { + t.Fatal(err) + } + if _, err := r.Retry(context.Background(), e.Bucket, e.ObjectKey, nil, nil); err != nil { + t.Fatal(err) + } + privateStateLocks.Lock() + recordingHeld, resultHeld := privateStateLocks.paths[dir] != nil, privateStateLocks.paths[resultPath] != nil + privateStateLocks.Unlock() + if recordingHeld || resultHeld { + t.Fatalf("completed entry kept permanent path locks: recording=%t result=%t", recordingHeld, resultHeld) + } +} + +func TestMissingAudioDoesNotRelaxUnfinishedRecordingValidation(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + entry := failedRecordingFixture() + if _, err := r.SaveFailure([]byte("recording bytes"), entry, &UploadHTTPError{StatusCode: 503}); err != nil { + t.Fatal(err) + } + _, audio, _, _ := r.paths(entry.Bucket, entry.ObjectKey) + if err := os.Remove(audio); err != nil { + t.Fatal(err) + } + if _, err := r.Load(entry.Bucket, entry.ObjectKey); !errors.Is(err, ErrRecordingRecoveryIncomplete) { + t.Fatalf("unfinished missing audio was accepted: %v", err) + } +} diff --git a/internal/agent/recording_recovery.go b/internal/agent/recording_recovery.go index e62ffdf..241ded3 100644 --- a/internal/agent/recording_recovery.go +++ b/internal/agent/recording_recovery.go @@ -11,7 +11,6 @@ import ( "os" "path/filepath" "strings" - "sync" "time" ) @@ -48,13 +47,11 @@ type RecordingRecovery struct { Now func() time.Time Upload UploadClient - mu sync.Mutex writeState func(string, any) error // deterministic failure injection in package tests + removeFile func(string) error // deletion failure injection in package tests } func (r *RecordingRecovery) SaveFailure(audio []byte, entry RecordingRecoveryEntry, cause error) (RecordingRecoveryEntry, error) { - r.mu.Lock() - defer r.mu.Unlock() var rejected *UploadHTTPError if !errors.Is(cause, ErrUploadOutcomeUnknown) && (!errors.As(cause, &rejected) || rejected.StatusCode < 300) { return RecordingRecoveryEntry{}, errors.New("only a failed OSS PUT may create recording recovery files") @@ -69,6 +66,11 @@ func (r *RecordingRecovery) SaveFailure(audio []byte, entry RecordingRecoveryEnt if err != nil { return RecordingRecoveryEntry{}, err } + release, err := holdPrivateState(dir) + if err != nil { + return RecordingRecoveryEntry{}, err + } + defer release() sha := sha256.Sum256(audio) digest := hex.EncodeToString(sha[:]) if entry.SHA256 != "" && !strings.EqualFold(entry.SHA256, digest) { @@ -113,6 +115,20 @@ func (r *RecordingRecovery) SaveFailure(audio []byte, entry RecordingRecoveryEnt } func (r *RecordingRecovery) Load(bucket, objectKey string) (RecordingRecoveryEntry, error) { + dir, _, _, err := r.paths(bucket, objectKey) + if err != nil { + return RecordingRecoveryEntry{}, err + } + release, err := holdPrivateState(dir) + if err != nil { + return RecordingRecoveryEntry{}, err + } + defer release() + return r.load(bucket, objectKey) +} + +// load requires the caller's per-recording lock when used by recovery/cleanup. +func (r *RecordingRecovery) load(bucket, objectKey string) (RecordingRecoveryEntry, error) { _, audioPath, infoPath, err := r.paths(bucket, objectKey) if err != nil { return RecordingRecoveryEntry{}, err @@ -139,6 +155,10 @@ func (r *RecordingRecovery) Load(bucket, objectKey string) (RecordingRecoveryEnt if entry.Bucket != bucket || entry.ObjectKey != objectKey || entry.CallID == "" || entry.SourceEventID == "" || entry.RecordingID == "" || entry.UploadID == "" || entry.SizeBytes <= 0 || entry.SHA256 == "" || !json.Valid(entry.ResultPayload) || entry.Attempts < 1 || entry.SavedAt.IsZero() || !entry.ExpiresAt.Equal(entry.SavedAt.Add(recordingRetryWindow)) { return RecordingRecoveryEntry{}, ErrRecordingRecoveryIncomplete } + digest, decodeErr := hex.DecodeString(entry.SHA256) + if decodeErr != nil || len(digest) != sha256.Size || hex.EncodeToString(digest) != entry.SHA256 { + return RecordingRecoveryEntry{}, fmt.Errorf("%w: invalid original audio checksum", ErrRecordingRecoveryIncomplete) + } switch entry.State { case "retry_pending", "outcome_unknown", "put_in_flight", "uploaded_unreported", "delivered", "expired": default: @@ -152,6 +172,9 @@ func (r *RecordingRecovery) Load(bucket, objectKey string) (RecordingRecoveryEnt 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) } diff --git a/internal/agent/recording_retry.go b/internal/agent/recording_retry.go index 8f0aa73..9886ccf 100644 --- a/internal/agent/recording_retry.go +++ b/internal/agent/recording_retry.go @@ -54,9 +54,16 @@ func (r *RecordingRecovery) Retry(ctx context.Context, bucket, objectKey string, slog.InfoContext(ctx, "Recording recovery transition complete", attrs...) } }() - r.mu.Lock() - defer r.mu.Unlock() - entry, err := r.Load(bucket, objectKey) + dir, _, _, err := r.paths(bucket, objectKey) + if err != nil { + return RecordingRecoveryEntry{}, err + } + release, err := holdPrivateState(dir) + if err != nil { + return RecordingRecoveryEntry{}, err + } + defer release() + entry, err := r.load(bucket, objectKey) if err != nil { return RecordingRecoveryEntry{}, err } @@ -71,8 +78,9 @@ func (r *RecordingRecovery) Retry(ctx context.Context, bucket, objectKey string, } switch entry.State { case "delivered": - phase = "recovery_already_delivered" - return entry, nil + phase = "delivered_recording_cleanup" + _, err := r.cleanupDelivered(entry) + return entry, err case "expired": return entry, ErrRecordingRetryExpired case "outcome_unknown", "put_in_flight": @@ -184,7 +192,8 @@ func (r *RecordingRecovery) reportUploaded(ctx context.Context, infoPath string, return entry, fmt.Errorf("persist delivered result receipt: %w", err) } slog.InfoContext(ctx, "Original result recovery acknowledged", append(recordingRecoveryAttrs(entry), "stage", "original_result_acknowledged")...) - return entry, nil + _, err := r.cleanupDelivered(entry) + return entry, err } func (r *RecordingRecovery) persistState(infoPath string, entry RecordingRecoveryEntry) error { diff --git a/internal/agent/recording_retry_test.go b/internal/agent/recording_retry_test.go index 2dd0e22..75cfca0 100644 --- a/internal/agent/recording_retry_test.go +++ b/internal/agent/recording_retry_test.go @@ -85,12 +85,12 @@ func TestRecordingRecoveryBackoffAndResultRetryDoNotRepeatSuccessfulPUT(t *testi return RecoveryTarget{}, errors.New("duplicate grant") } processed, err := restarted.RecoverDue(context.Background(), neverGrant, report) - finished, loadErr := restarted.Load(entry.Bucket, entry.ObjectKey) - if err != nil || loadErr != nil || processed != 1 || finished.State != "delivered" || puts.Load() != 2 || grants.Load() != 2 || reports.Load() != 2 { - t.Fatalf("restart must resend only original result: recovered=%d state=%s puts=%d grants=%d reports=%d err=%v load=%v", processed, finished.State, puts.Load(), grants.Load(), reports.Load(), err, loadErr) + _, loadErr := restarted.Load(entry.Bucket, entry.ObjectKey) + if err != nil || !errors.Is(loadErr, os.ErrNotExist) || processed != 1 || puts.Load() != 2 || grants.Load() != 2 || reports.Load() != 2 { + t.Fatalf("restart must resend only original result then clean delivered files: recovered=%d puts=%d grants=%d reports=%d err=%v load=%v", processed, puts.Load(), grants.Load(), reports.Load(), err, loadErr) } - if _, err := os.Stat(filepath.Join(root, "mock-bucket", "tenant", "call", "rec.wav", "recording.wav")); err != nil { - t.Fatalf("failed-path original audio must not be cleared automatically: %v", err) + if _, err := os.Stat(filepath.Join(root, "mock-bucket", "tenant", "call", "rec.wav", "recording.wav")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("delivered original audio remains: %v", err) } } diff --git a/internal/agent/recording_scan.go b/internal/agent/recording_scan.go index a656456..d8cbe70 100644 --- a/internal/agent/recording_scan.go +++ b/internal/agent/recording_scan.go @@ -36,6 +36,7 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, return 0, err } processed := 0 + handled := make(map[string]bool) var blocked []error addBlocked := func(dir, stage string, cause error, identity ...any) { digest := sha256.Sum256([]byte(dir)) @@ -51,28 +52,40 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, } walkErr := filepath.WalkDir(r.Root, func(path string, item fs.DirEntry, err error) error { if err != nil { - return err + if path == r.Root { + return err + } + addBlocked(path, "recovery_walk", err) + if item != nil && item.IsDir() { + return filepath.SkipDir + } + return nil } if err := ctx.Err(); err != nil { return err } if item.IsDir() { + if err := checkRecoveryDirectory(path); err != nil { + addBlocked(path, "recovery_directory", err) + return filepath.SkipDir + } if path == filepath.Join(r.Root, ".results") { - if err := checkRecoveryDirectory(path); err != nil { - return err - } return filepath.SkipDir // result-only files have their own scanner } - return checkRecoveryDirectory(path) + return nil } dir := filepath.Dir(path) switch item.Name() { case "recording.wav": + if handled[dir] { + return nil + } // info processing may have cleaned this pair if _, err := os.Stat(filepath.Join(dir, "info.json")); err != nil { addBlocked(dir, "recovery_validate", fmt.Errorf("%w: original audio has no call information", ErrRecordingRecoveryIncomplete)) } return nil case "info.json": + handled[dir] = true default: addBlocked(dir, "recovery_validate", fmt.Errorf("%w: unexpected recovery file", ErrRecordingRecoveryIncomplete)) return nil diff --git a/internal/agent/recording_scan_test.go b/internal/agent/recording_scan_test.go index d7cb255..c9c6924 100644 --- a/internal/agent/recording_scan_test.go +++ b/internal/agent/recording_scan_test.go @@ -99,7 +99,7 @@ func TestRecordingRecoveryScannerReportsOrphanInsteadOfInventingResult(t *testin } } -func TestRecordingRecoveryConcurrentRetryOnlyPutsOnce(t *testing.T) { +func TestRecordingRecoveryConcurrentInstancesOnlyPutOnce(t *testing.T) { root := privateRecoveryRoot(t) now := time.Date(2026, 10, 1, 10, 0, 0, 0, time.UTC) recovery := RecordingRecovery{Root: root, Now: func() time.Time { return now }, Upload: UploadClient{AllowInsecureHTTP: true, Now: func() time.Time { return now }}} @@ -122,20 +122,30 @@ func TestRecordingRecoveryConcurrentRetryOnlyPutsOnce(t *testing.T) { var wg sync.WaitGroup errs := make(chan error, 2) for i := 0; i < 2; i++ { + instance := &RecordingRecovery{Root: recovery.Root, Now: recovery.Now, Upload: recovery.Upload} wg.Add(1) go func() { defer wg.Done() - _, err := recovery.Retry(context.Background(), entry.Bucket, entry.ObjectKey, grant, report) + _, err := instance.Retry(context.Background(), entry.Bucket, entry.ObjectKey, grant, report) errs <- err }() } wg.Wait() close(errs) + completed, removed := 0, 0 for err := range errs { - if err != nil { + switch { + case err == nil: + completed++ + case errors.Is(err, os.ErrNotExist): + removed++ // the first caller removed its confirmed entry; no tombstone remains + default: t.Fatalf("serialized retry failed: %v", err) } } + if completed != 1 || removed != 1 { + t.Fatalf("concurrent callers invented a retained completion history: completed=%d removed=%d", completed, removed) + } if puts.Load() != 1 || reports.Load() != 1 { t.Fatalf("concurrent retry repeated a completed side effect: puts=%d reports=%d", puts.Load(), reports.Load()) } diff --git a/internal/agent/result_journal.go b/internal/agent/result_journal.go index 8dcd313..ab8d4fa 100644 --- a/internal/agent/result_journal.go +++ b/internal/agent/result_journal.go @@ -39,6 +39,14 @@ func (j ResultJournal) path(id string) (string, error) { return filepath.Join(j.Root, ".results", hex.EncodeToString(sum[:])+".json"), nil } +func (j ResultJournal) hold(id string) (func(), error) { + path, err := j.path(id) + if err != nil { + return nil, err + } + return holdPrivateState(path) +} + func (j ResultJournal) prepare() (string, error) { if j.Root == "" { return "", errors.New("private result journal root is required") @@ -64,6 +72,11 @@ func (j ResultJournal) save(entry ResultEntry, state string) error { if err != nil { return err } + release, err := holdPrivateState(path) + if err != nil { + return err + } + defer release() dir, err := j.prepare() if err != nil { return err @@ -128,6 +141,11 @@ func (j ResultJournal) load(id string) (ResultEntry, string, error) { } func (j ResultJournal) MarkEnded(id string) error { + release, err := j.hold(id) + if err != nil { + return err + } + defer release() entry, path, err := j.load(id) if err != nil { return err @@ -140,6 +158,11 @@ func (j ResultJournal) MarkEnded(id string) error { } func (j ResultJournal) MarkUploaded(id string, observation *agentpb.UploadObservation) error { + release, err := j.hold(id) + if err != nil { + return err + } + defer release() entry, path, err := j.load(id) if err != nil { return err @@ -153,6 +176,11 @@ func (j ResultJournal) MarkUploaded(id string, observation *agentpb.UploadObserv } func (j ResultJournal) Ack(id string) error { + release, err := j.hold(id) + if err != nil { + return err + } + defer release() entry, path, err := j.load(id) if err != nil { return err @@ -169,6 +197,11 @@ func (j ResultJournal) Ack(id string) error { // DiscardPrepared is only called after the durable audio+result recovery entry // has itself reported the same final result. Old entries have no journal. func (j ResultJournal) DiscardPrepared(id string) error { + release, err := j.hold(id) + if err != nil { + return err + } + defer release() entry, path, err := j.load(id) if err != nil { if errors.Is(err, os.ErrNotExist) { diff --git a/internal/agent/upload.go b/internal/agent/upload.go index ee5a008..9750ea5 100644 --- a/internal/agent/upload.go +++ b/internal/agent/upload.go @@ -23,7 +23,8 @@ import ( // UploadClient performs the Agent-direct data-plane upload using a restricted // Dispatcher grant. It never sends file bytes through Dispatcher or writes an // OSS credential to logs. It never deletes or moves the source asset; the -// lifecycle owner retains it until upload-fact delivery to MQ is durably recorded. +// lifecycle owner retains recovery files until the Dispatcher accepts the final +// result and the Agent durably records that delivery; it never deletes OSS data. type UploadClient struct { HTTPClient *http.Client AllowedHosts map[string]struct{}