diff --git a/cmd/sip-go-agent/agent_cleanup.go b/cmd/sip-go-agent/agent_cleanup.go new file mode 100644 index 0000000..e8768bf --- /dev/null +++ b/cmd/sip-go-agent/agent_cleanup.go @@ -0,0 +1,26 @@ +package main + +import ( + "context" + "log/slog" + + "git.ipao.vip/rogee/go-sip/internal/agent" + "git.ipao.vip/rogee/go-sip/internal/daily" +) + +// This owns only local delivered-file cleanup, never RPC/cloud recovery. +func startAgentMidnightCleanup(parent context.Context, root string) func() { + ctx, cancel := context.WithCancel(parent) + done := make(chan struct{}) + go func() { + defer close(done) + if err := daily.Run(ctx, func(ctx context.Context) error { + count, err := (&agent.RecordingRecovery{Root: root}).CleanupDelivered(ctx) + slog.InfoContext(ctx, "Agent midnight cleanup finished", "completed_recordings", count, "error", err) + return err + }); err != nil { + slog.Error("Agent midnight cleanup timer failed", "error", err) + } + }() + return func() { cancel(); <-done } +} diff --git a/cmd/sip-go-agent/agent_command.go b/cmd/sip-go-agent/agent_command.go index 39f3de6..843d1ee 100644 --- a/cmd/sip-go-agent/agent_command.go +++ b/cmd/sip-go-agent/agent_command.go @@ -113,6 +113,10 @@ func newAgentCommand() *cobra.Command { defer listener.Close() server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)), grpc.ChainUnaryInterceptor(logging.UnaryServerInterceptor("agent", "dispatcher"))) agentpb.RegisterAgentControlServiceServer(server, handler) + if settings.RecoveryRoot != "" { + stopCleanup := startAgentMidnightCleanup(cmd.Context(), settings.RecoveryRoot) + defer stopCleanup() + } if mode == "nonprod-real" { startRealAgentRecovery(cmd.Context(), settings, recoveryClient, handler) } diff --git a/docs/evidence/sqlite-cleanup-v0.1-local.md b/docs/evidence/sqlite-cleanup-v0.1-local.md index 0b0d835..93227e1 100644 --- a/docs/evidence/sqlite-cleanup-v0.1-local.md +++ b/docs/evidence/sqlite-cleanup-v0.1-local.md @@ -47,6 +47,15 @@ - RejectData 写入和清理共享生命周期排除；清理结束后同源再来的新拒绝保留新原件和未确认回执，不误删为旧记录。移除了此前 admitted rejected 的“只要整个拒绝目录非空就保留”保护，改为精确本组资格。 - MQ 确认后的真实控制器入口已验证三类拒绝，未产生执行或伪造最终结果；多原件、两阶段失败、损坏组与独立组、既有文件、未知临时文件和并发写入回归通过。`go test -race ./internal/store ./internal/dispatcher ./internal/rpc` 通过。 +### 每日午夜补扫 + +- 采用固定 Asia/Shanghai 日历边界，启动只安排下一次午夜，不在启动时额外清理、不在 12:00 运行、不按消息年龄删除；每次重新计算下一日期。系统时钟回退不会提前执行，清理失败等待下一午夜，不形成高频重试。 +- Dispatcher 使用独立且可取消的后台任务，分页扫描 completed 候选并复用 MQ 确认后相同的逐条事务/原件资格；单条失败继续其他记录，不变更准入、任务状态、停止状态或发布消息。停服取消并等待清理任务结束。 +- 配置录音恢复根目录的 Agent 启动独立午夜任务，复用已有扫描和 delivered 删除；其他状态只校验保留，不申请授权、PUT 或上报结果。既有十秒真实恢复的用途和节奏不变。 +- 每条日志记录身份、保留原因或错误、实际文件数量与字节；Dispatcher 汇总扫描、保留、失败、删除行、逻辑数据字节、文件数量/字节及耗时。没有永久扫描进度或二次确认记录。 +- UTC/上海转换、12:00 不清理、午夜启动安排次日、时钟回退、失败等待次日、取消计时器，以及 205 条跨页候选、未确认/待拨邻居保留、取消后续处理、只清 delivered 不重试其他录音验证通过。 +- `go test -race ./internal/daily ./internal/store ./internal/dispatcher ./internal/agent ./cmd/sip-go-agent` 通过。 + ## 尚待完成 -午夜补扫，以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。 +全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。 diff --git a/internal/agent/recording_cleanup_test.go b/internal/agent/recording_cleanup_test.go index 416b2af..e531b21 100644 --- a/internal/agent/recording_cleanup_test.go +++ b/internal/agent/recording_cleanup_test.go @@ -238,6 +238,28 @@ func TestRecordingCleanupLeavesNoPermanentInMemoryLocks(t *testing.T) { } } +func TestMidnightDeliveredOnlySweepPreservesUnfinishedFiles(t *testing.T) { + r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} + completed := deliveredRecordingFixture(t, r) + pending := failedRecordingFixture() + pending.SourceEventID = "execute-pending" + pending.ObjectKey = "tenant/call/pending.wav" + if _, err := r.SaveFailure([]byte("pending audio"), pending, &UploadHTTPError{StatusCode: 503}); err != nil { + t.Fatal(err) + } + count, err := r.CleanupDelivered(context.Background()) + if err != nil || count != 1 { + t.Fatalf("midnight result=%d err=%v", count, err) + } + _, _, completedInfo, _ := r.paths(completed.Bucket, completed.ObjectKey) + if _, err := os.Stat(completedInfo); !errors.Is(err, os.ErrNotExist) { + t.Fatal("delivered info remains") + } + if entry, err := r.Load(pending.Bucket, pending.ObjectKey); err != nil || entry.State != "retry_pending" { + t.Fatalf("unfinished entry changed: %+v %v", entry, err) + } +} + func TestMissingAudioDoesNotRelaxUnfinishedRecordingValidation(t *testing.T) { r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")} entry := failedRecordingFixture() diff --git a/internal/agent/recording_scan.go b/internal/agent/recording_scan.go index d8cbe70..49ba767 100644 --- a/internal/agent/recording_scan.go +++ b/internal/agent/recording_scan.go @@ -16,7 +16,16 @@ import ( // RecoverDue discovers failure-only recording pairs after an Agent restart. // Blocked/corrupt entries remain untouched and are returned as errors while // other independent entries continue; none can produce a fabricated result. -func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter) (processedCount int, scanErr error) { +func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter) (int, error) { + return r.scanRecovery(ctx, grant, report, false) +} + +// CleanupDelivered is a local midnight-only sweep, never an upload retry. +func (r *RecordingRecovery) CleanupDelivered(ctx context.Context) (int, error) { + return r.scanRecovery(ctx, nil, nil, true) +} + +func (r *RecordingRecovery) scanRecovery(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter, deliveredOnly bool) (processedCount int, scanErr error) { defer func() { if scanErr != nil { slog.ErrorContext(ctx, "Recording recovery scan failed", "stage", "recovery_scan_failed", "processed", processedCount, "error", scanErr) @@ -106,6 +115,9 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, addBlocked(dir, "recovery_load", err) return nil } + if deliveredOnly && before.State != "delivered" { + return nil + } after, err := r.Retry(ctx, bucket, objectKey, grant, report) if errors.Is(err, ErrRecordingNotDue) { return nil @@ -116,6 +128,12 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, } if before.State != "delivered" && after.State == "delivered" { processed++ + } else if deliveredOnly && after.State == "delivered" { + if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) { + processed++ + } else if err != nil { + addBlocked(path, "delivered_cleanup_stat", err) + } } return nil }) diff --git a/internal/daily/midnight.go b/internal/daily/midnight.go new file mode 100644 index 0000000..68ddabd --- /dev/null +++ b/internal/daily/midnight.go @@ -0,0 +1,58 @@ +// Package daily provides the project's fixed Shanghai midnight calendar timer. +// It is not configurable and does not track permanent sweep history. +package daily + +import ( + "context" + "errors" + "log/slog" + "time" + _ "time/tzdata" +) + +func NextMidnight(now time.Time) (time.Time, error) { + zone, err := time.LoadLocation("Asia/Shanghai") + if err != nil { + return time.Time{}, err + } + local := now.In(zone) + return time.Date(local.Year(), local.Month(), local.Day()+1, 0, 0, 0, 0, zone), nil +} + +func Run(ctx context.Context, sweep func(context.Context) error) error { + return run(ctx, sweep, time.Now, func(delay time.Duration) (<-chan time.Time, func()) { + timer := time.NewTimer(delay) + return timer.C, func() { timer.Stop() } + }) +} + +func run(ctx context.Context, sweep func(context.Context) error, now func() time.Time, newTimer func(time.Duration) (<-chan time.Time, func())) error { + if ctx == nil || sweep == nil || now == nil || newTimer == nil { + return errors.New("midnight cleanup requires context and a sweep") + } + for { + if ctx.Err() != nil { + return nil + } + current := now() + next, err := NextMidnight(current) + if err != nil { + return err + } + slog.InfoContext(ctx, "Midnight cleanup scheduled", "timezone", "Asia/Shanghai", "next_at", next) + tick, stop := newTimer(next.Sub(current)) + select { + case <-ctx.Done(): + stop() + return nil + case <-tick: + stop() + } + if now().Before(next) { + continue + } // a wall-clock rollback must not run early + if err := sweep(ctx); err != nil && ctx.Err() == nil { + slog.ErrorContext(ctx, "Midnight cleanup incomplete; retry scheduled for next midnight", "error", err) + } + } +} diff --git a/internal/daily/midnight_test.go b/internal/daily/midnight_test.go new file mode 100644 index 0000000..1a5eb55 --- /dev/null +++ b/internal/daily/midnight_test.go @@ -0,0 +1,72 @@ +package daily + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestNextMidnightUsesShanghaiCalendarNotNoonOrElapsed24Hours(t *testing.T) { + for _, test := range []struct{ now, want string }{ + {"2026-10-10T03:59:00Z", "2026-10-11T00:00:00+08:00"}, + {"2026-10-10T04:00:00Z", "2026-10-11T00:00:00+08:00"}, + {"2026-10-10T15:59:59Z", "2026-10-11T00:00:00+08:00"}, + {"2026-10-10T16:00:00Z", "2026-10-12T00:00:00+08:00"}, + } { + now, _ := time.Parse(time.RFC3339, test.now) + got, err := NextMidnight(now) + if err != nil || got.Format(time.RFC3339) != test.want { + t.Fatalf("now=%s got=%s want=%s err=%v", test.now, got, test.want, err) + } + } +} + +func TestWallClockRollbackDoesNotTriggerEarlyCleanup(t *testing.T) { + now, _ := time.Parse(time.RFC3339, "2026-10-10T15:59:00Z") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + calls, waits := 0, []time.Duration{} + err := run(ctx, func(context.Context) error { calls++; return nil }, func() time.Time { return now }, func(delay time.Duration) (<-chan time.Time, func()) { + waits = append(waits, delay) + ch := make(chan time.Time, 1) + switch len(waits) { + case 1: + now = now.Add(delay - time.Hour) + ch <- now + case 2: + now = now.Add(delay) + ch <- now + default: + cancel() + } + return ch, func() {} + }) + if err != nil || calls != 1 || len(waits) != 3 || waits[1] != time.Hour { + t.Fatalf("rollback ran early: calls=%d waits=%v err=%v", calls, waits, err) + } +} + +func TestMidnightFailureRetriesOnlyTomorrowAndCancellationStopsTimer(t *testing.T) { + now, _ := time.Parse(time.RFC3339, "2026-10-10T15:59:00Z") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + calls, stopped := 0, 0 + durations := []time.Duration{} + clock := func() time.Time { return now } + timer := func(delay time.Duration) (<-chan time.Time, func()) { + durations = append(durations, delay) + ch := make(chan time.Time, 1) + if len(durations) <= 2 { + now = now.Add(delay) + ch <- now + } else { + cancel() + } + return ch, func() { stopped++ } + } + err := run(ctx, func(context.Context) error { calls++; return errors.New("one record failed") }, clock, timer) + if err != nil || calls != 2 || stopped != 3 || len(durations) != 3 || durations[0] != time.Minute || durations[1] != 24*time.Hour { + t.Fatalf("unexpected daily behavior: calls=%d stopped=%d delays=%v err=%v", calls, stopped, durations, err) + } +} diff --git a/internal/dispatcher/cleanup.go b/internal/dispatcher/cleanup.go index 47860ad..3b215be 100644 --- a/internal/dispatcher/cleanup.go +++ b/internal/dispatcher/cleanup.go @@ -9,9 +9,7 @@ import ( // 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 +func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event store.OutboxEvent) (result store.CleanupResult, err error) { switch event.EventType { case "call.execute", "call.execute.result": result, err = c.Store.CleanupCall(c.DispatcherID, event.EventID) @@ -26,16 +24,17 @@ func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event case "sip.config": result, err = c.Store.CleanupRejectedFeedback(c.DispatcherID, event.EventType, event.EventID) default: - return + return result, nil } 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, "deleted_files", result.DeletedFiles, "file_bytes", result.FileBytes, "error", err) - return + return result, err } 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) + logger.Info("Completed data retained", "reason", result.RetainedReason) } + return result, nil } diff --git a/internal/dispatcher/cleanup_scan.go b/internal/dispatcher/cleanup_scan.go new file mode 100644 index 0000000..4765766 --- /dev/null +++ b/internal/dispatcher/cleanup_scan.go @@ -0,0 +1,63 @@ +package dispatcher + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "git.ipao.vip/rogee/go-sip/internal/store" +) + +// SweepCompleted uses exactly the per-record qualifications used after MQ +// confirmation. It never publishes, changes admission, or rewrites task state. +func (c *ExecuteController) SweepCompleted(ctx context.Context) (total store.CleanupResult, sweepErr error) { + if c == nil || c.Store == nil || c.DispatcherID == "" { + return total, errors.New("cleanup sweep requires durable Dispatcher identity") + } + started := time.Now() + scanned, retained, failed := 0, 0, 0 + defer func() { + slog.InfoContext(ctx, "Completed data sweep finished", "dispatcher_id", c.DispatcherID, "scanned", scanned, "retained", retained, "failed", failed, "deleted_rows", total.DeletedRows, "data_bytes", total.DataBytes, "deleted_files", total.DeletedFiles, "file_bytes", total.FileBytes, "elapsed_ms", time.Since(started).Milliseconds()) + }() + cursorType, cursorID := "", "" + var firstFailure error + for { + if err := ctx.Err(); err != nil { + return total, errors.Join(firstFailure, err) + } + batch, err := c.Store.ListCleanupCandidates(c.DispatcherID, cursorType, cursorID) + if err != nil { + return total, errors.Join(firstFailure, err) + } + for _, event := range batch { + if err := ctx.Err(); err != nil { + return total, errors.Join(firstFailure, err) + } + scanned++ + result, err := c.cleanupConfirmedFeedback(ctx, event) + total.DeletedRows += result.DeletedRows + total.DataBytes += result.DataBytes + total.DeletedFiles += result.DeletedFiles + total.FileBytes += result.FileBytes + if err != nil { + failed++ + if firstFailure == nil { + firstFailure = err + } + } else if result.RetainedReason != "" { + retained++ + } + } + if len(batch) == 0 { + break + } + last := batch[len(batch)-1] + cursorType, cursorID = last.EventType, last.EventID + } + if failed != 0 { + return total, fmt.Errorf("%d independent cleanup failures; first: %w", failed, firstFailure) + } + return total, nil +} diff --git a/internal/dispatcher/cleanup_scan_test.go b/internal/dispatcher/cleanup_scan_test.go new file mode 100644 index 0000000..b556ac8 --- /dev/null +++ b/internal/dispatcher/cleanup_scan_test.go @@ -0,0 +1,66 @@ +package dispatcher + +import ( + "context" + "errors" + "fmt" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/store" +) + +func TestMidnightSweepUsesRealTimeConditionsAndKeepsUnconfirmedNeighbors(t *testing.T) { + c, originator, _, s := newExecuteFixture(t) + for i := 0; i < 205; i++ { + id := fmt.Sprintf("night-confirmed-%03d", i) + raw := []byte(`{"event_id":"` + id + `","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{}}`) + if err := s.RejectData(c.DispatcherID, raw, errors.New("invalid original")); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(c.DispatcherID, "call.execute", id); err != nil { + t.Fatal(err) + } + } + raw := []byte(`{"event_id":"night-unconfirmed","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{}}`) + if err := s.RejectData(c.DispatcherID, raw, errors.New("invalid original")); err != nil { + t.Fatal(err) + } + pending := store.ExecuteCommand{DispatcherID: c.DispatcherID, EventID: "night-pending", TenantID: 1001, TaskID: "task-asr", Callee: "15003164745", IssuedAt: "2026-09-21T01:00:00Z"} + if _, _, err := s.RecordExecute(pending); err != nil { + t.Fatal(err) + } + result, err := c.SweepCompleted(context.Background()) + if err != nil || result.DeletedRows != 205 || result.DeletedFiles != 205 || len(originator.calls) != 0 { + t.Fatalf("sweep=%+v err=%v", result, err) + } + outbox, err := s.ListPendingOutbox(c.DispatcherID) + if err != nil || len(outbox) != 1 || outbox[0].EventID != "night-unconfirmed" { + t.Fatalf("unconfirmed proof lost: %v %v", outbox, err) + } + commands, err := s.ListPendingExecute(c.DispatcherID) + if err != nil || len(commands) != 1 || commands[0].EventID != pending.EventID { + t.Fatalf("pending neighbor changed: %v %v", commands, err) + } + if result, err := c.SweepCompleted(context.Background()); err != nil || result.DeletedRows != 0 { + t.Fatalf("repeat=%+v %v", result, err) + } +} + +func TestMidnightSweepCancellationDoesNotInventCompletion(t *testing.T) { + c, _, _, s := newExecuteFixture(t) + body := []byte(`{"event_id":"night-cancelled","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{}}`) + if err := s.RejectData(c.DispatcherID, body, errors.New("invalid")); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(c.DispatcherID, "call.execute", "night-cancelled"); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if result, err := c.SweepCompleted(ctx); !errors.Is(err, context.Canceled) || result.DeletedRows != 0 { + t.Fatalf("cancelled sweep=%+v %v", result, err) + } + if result, err := c.SweepCompleted(context.Background()); err != nil || result.DeletedRows != 1 { + t.Fatalf("resumed sweep=%+v %v", result, err) + } +} diff --git a/internal/dispatcher/runtime.go b/internal/dispatcher/runtime.go index f5208e8..331516f 100644 --- a/internal/dispatcher/runtime.go +++ b/internal/dispatcher/runtime.go @@ -11,6 +11,7 @@ import ( "git.ipao.vip/rogee/go-sip/internal/configread" "git.ipao.vip/rogee/go-sip/internal/contract" + "git.ipao.vip/rogee/go-sip/internal/daily" "git.ipao.vip/rogee/go-sip/internal/mq" "git.ipao.vip/rogee/go-sip/internal/store" "git.ipao.vip/rogee/go-sip/internal/tenant" @@ -75,6 +76,19 @@ func (r *Runtime) Serve(ctx context.Context) (result error) { } }() + cleanupContext, stopCleanup := context.WithCancel(ctx) + cleanupDone := make(chan struct{}) + go func() { + defer close(cleanupDone) + if err := daily.Run(cleanupContext, func(ctx context.Context) error { + _, err := r.Execute.SweepCompleted(ctx) + return err + }); err != nil { + slog.Error("Dispatcher midnight cleanup timer failed", "error", err) + } + }() + defer func() { stopCleanup(); <-cleanupDone }() + r.Broker.SetInvalidMessageHandler(r.rejectInvalidData) if err := r.Execute.FlushOutbox(ctx); err != nil { return err diff --git a/internal/store/cleanup_scan.go b/internal/store/cleanup_scan.go new file mode 100644 index 0000000..6d5040e --- /dev/null +++ b/internal/store/cleanup_scan.go @@ -0,0 +1,27 @@ +package store + +// ListCleanupCandidates snapshots one bounded, stable-key page. Pending and +// unknown executions are not candidates; missing confirmations stay protected +// by the same transactional qualification used for real-time cleanup. +func (s *Store) ListCleanupCandidates(dispatcherID, afterType, afterID string) ([]OutboxEvent, error) { + rows, err := s.db.Query(`SELECT event_type,event_id FROM ( + SELECT 'call.execute' AS event_type,event_id FROM dispatcher_inbox + WHERE dispatcher_id=? AND status IN ('finished','rejected','suppressed') + UNION + SELECT event_type,event_id FROM dispatcher_outbox WHERE dispatcher_id=? AND confirmed=1 + AND event_type IN ('call.execute','call.execute.result','task.control','sip.config') + ) WHERE event_type>? OR (event_type=? AND event_id>?) ORDER BY event_type,event_id LIMIT ?`, dispatcherID, dispatcherID, afterType, afterType, afterID, cleanupBatchSize) + if err != nil { + return nil, err + } + defer rows.Close() + var candidates []OutboxEvent + for rows.Next() { + event := OutboxEvent{} + if err := rows.Scan(&event.EventType, &event.EventID); err != nil { + return nil, err + } + candidates = append(candidates, event) + } + return candidates, rows.Err() +}