From f6089ac5fb474400e838fb3e9d9e7a16da90965f Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Wed, 6 May 2026 11:52:17 +0800 Subject: [PATCH] chore(jobs): align log keys + add prune no-op test + down-migration comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-cutting review polish: - worker.go: id/type → job_id/job_type; one stray attempts → attempt for consistency with the other 4 sites in the same file. - runners/workspace_fetch.go: ws_id → workspace_id (matches the dispatched notify.Message metadata key). - runners/worktree_prune.go: id → worktree_id (same reason). - runners/worktree_prune_test.go: add TestWorktreePrune_NoRowsIsNoOp for symmetry with the no-op tests in the other 4 runner test files. - migrations/0005_jobs.down.sql: comment that DROP TABLE jobs cascades to drop the three idx_jobs_* indexes automatically. --- internal/jobs/runners/workspace_fetch.go | 10 +++++----- internal/jobs/runners/worktree_prune.go | 10 +++++----- internal/jobs/runners/worktree_prune_test.go | 17 +++++++++++++++++ internal/jobs/worker.go | 20 ++++++++++---------- migrations/0005_jobs.down.sql | 2 ++ 5 files changed, 39 insertions(+), 20 deletions(-) diff --git a/internal/jobs/runners/workspace_fetch.go b/internal/jobs/runners/workspace_fetch.go index 591f9cf..f3a9c36 100644 --- a/internal/jobs/runners/workspace_fetch.go +++ b/internal/jobs/runners/workspace_fetch.go @@ -75,7 +75,7 @@ func (r *WorkspaceFetch) Run(ctx context.Context) { func (r *WorkspaceFetch) fetchOne(ctx context.Context, t WorkspaceFetchTarget) { ok, err := r.repo.MarkSyncing(ctx, t.ID) if err != nil { - r.log.Error("workspace_fetch.mark_syncing", "ws_id", t.ID, "err", err.Error()) + r.log.Error("workspace_fetch.mark_syncing", "workspace_id", t.ID, "err", err.Error()) return } if !ok { @@ -85,7 +85,7 @@ func (r *WorkspaceFetch) fetchOne(ctx context.Context, t WorkspaceFetchTarget) { if fetchErr := r.gitr.Fetch(ctx, t.MainPath, t.ID); fetchErr != nil { msg := truncate(fetchErr.Error(), 2000) if err := r.repo.MarkFetchResult(ctx, t.ID, false, msg); err != nil { - r.log.Error("workspace_fetch.mark_fail", "ws_id", t.ID, "err", err.Error()) + r.log.Error("workspace_fetch.mark_fail", "workspace_id", t.ID, "err", err.Error()) return } if derr := r.disp.Dispatch(ctx, notify.Message{ @@ -101,13 +101,13 @@ func (r *WorkspaceFetch) fetchOne(ctx context.Context, t WorkspaceFetchTarget) { }, CreatedAt: time.Now().UTC(), }); derr != nil { - r.log.Error("workspace_fetch.dispatch", "ws_id", t.ID, "err", derr.Error()) + r.log.Error("workspace_fetch.dispatch", "workspace_id", t.ID, "err", derr.Error()) } - r.log.Warn("workspace_fetch.failed", "ws_id", t.ID, "err", msg) + r.log.Warn("workspace_fetch.failed", "workspace_id", t.ID, "err", msg) return } if err := r.repo.MarkFetchResult(ctx, t.ID, true, ""); err != nil { - r.log.Error("workspace_fetch.mark_ok", "ws_id", t.ID, "err", err.Error()) + r.log.Error("workspace_fetch.mark_ok", "workspace_id", t.ID, "err", err.Error()) } } diff --git a/internal/jobs/runners/worktree_prune.go b/internal/jobs/runners/worktree_prune.go index 79295aa..41826b7 100644 --- a/internal/jobs/runners/worktree_prune.go +++ b/internal/jobs/runners/worktree_prune.go @@ -75,7 +75,7 @@ func (r *WorktreePrune) warningPhase(ctx context.Context) { } for _, w := range rows { if err := r.repo.MarkPruneWarning(ctx, w.ID); err != nil { - r.log.Error("worktree_prune.mark_warning", "id", w.ID, "err", err.Error()) + r.log.Error("worktree_prune.mark_warning", "worktree_id", w.ID, "err", err.Error()) continue } if derr := r.disp.Dispatch(ctx, notify.Message{ @@ -90,7 +90,7 @@ func (r *WorktreePrune) warningPhase(ctx context.Context) { }, CreatedAt: time.Now().UTC(), }); derr != nil { - r.log.Error("worktree_prune.dispatch", "id", w.ID, "err", derr.Error()) + r.log.Error("worktree_prune.dispatch", "worktree_id", w.ID, "err", derr.Error()) } } } @@ -109,14 +109,14 @@ func (r *WorktreePrune) executionPhase(ctx context.Context) { continue } if rmErr := r.gitr.WorktreeRemove(ctx, started.MainPath, started.Path); rmErr != nil { - r.log.Error("worktree_prune.remove_failed", "id", w.ID, "err", rmErr.Error()) + r.log.Error("worktree_prune.remove_failed", "worktree_id", w.ID, "err", rmErr.Error()) if rbErr := r.repo.FinishPruneRollback(ctx, w.ID); rbErr != nil { - r.log.Error("worktree_prune.rollback", "id", w.ID, "err", rbErr.Error()) + r.log.Error("worktree_prune.rollback", "worktree_id", w.ID, "err", rbErr.Error()) } continue } if err := r.repo.FinishPruneSuccess(ctx, w.ID); err != nil { - r.log.Error("worktree_prune.finish_success", "id", w.ID, "err", err.Error()) + r.log.Error("worktree_prune.finish_success", "worktree_id", w.ID, "err", err.Error()) } } } diff --git a/internal/jobs/runners/worktree_prune_test.go b/internal/jobs/runners/worktree_prune_test.go index b6d5240..2ba30f4 100644 --- a/internal/jobs/runners/worktree_prune_test.go +++ b/internal/jobs/runners/worktree_prune_test.go @@ -155,6 +155,23 @@ func TestWorktreePrune_ExecutionFailureRollsBack(t *testing.T) { assert.Equal(t, "rollback", repo.finalized[id]) } +func TestWorktreePrune_NoRowsIsNoOp(t *testing.T) { + t.Parallel() + repo := &fakeWorktreeRepo{} // both lists nil + gitr := &fakePruner{} + disp := &recordingDispatcher{} + cfg := runners.WorktreePruneConfig{IdleThreshold: 7 * 24 * time.Hour, WarningLead: 3 * 24 * time.Hour} + + r := runners.NewWorktreePrune(repo, gitr, disp, cfg, discardLogger()) + r.Run(context.Background()) + + assert.Empty(t, repo.markedWarning) + assert.Empty(t, repo.tryStartCalled) + assert.Empty(t, repo.finalized) + assert.Zero(t, gitr.calls) + assert.Empty(t, disp.msgs) +} + func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) { t.Parallel() id := uuid.New() diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index 45d2a8d..6116e8b 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -86,15 +86,15 @@ func (w *Worker) pollOnce(ctx context.Context) { } func (w *Worker) execute(ctx context.Context, job Job) { - w.opts.Log.Info("job.start", "id", job.ID, "type", job.Type, "attempt", job.Attempts) + w.opts.Log.Info("job.start", "job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts) start := w.opts.Clock.Now() h, ok := w.handlers[job.Type] if !ok { if derr := w.repo.DeadLetter(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type)); derr != nil { - w.opts.Log.Error("job.dead_letter_write_failed", "id", job.ID, "err", derr.Error()) + w.opts.Log.Error("job.dead_letter_write_failed", "job_id", job.ID, "err", derr.Error()) } - w.opts.Log.Error("job.dead_letter", "id", job.ID, "type", job.Type, "reason", "unknown_type") + w.opts.Log.Error("job.dead_letter", "job_id", job.ID, "job_type", job.Type, "reason", "unknown_type") w.fireOnDeadLetter(ctx, job, errors.New("unknown job type: "+string(job.Type))) return } @@ -102,10 +102,10 @@ func (w *Worker) execute(ctx context.Context, job Job) { err := w.runHandler(ctx, h, job) if err == nil { if cerr := w.repo.Complete(ctx, job.ID); cerr != nil { - w.opts.Log.Error("job.complete_write_failed", "id", job.ID, "err", cerr.Error()) + w.opts.Log.Error("job.complete_write_failed", "job_id", job.ID, "err", cerr.Error()) } w.opts.Log.Info("job.complete", - "id", job.ID, "type", job.Type, "attempt", job.Attempts, + "job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts, "duration_ms", w.opts.Clock.Now().Sub(start).Milliseconds()) return } @@ -114,10 +114,10 @@ func (w *Worker) execute(ctx context.Context, job Job) { if IsPermanent(err) || job.Attempts >= job.MaxAttempts { msg := truncate(err.Error(), 2000) if derr := w.repo.DeadLetter(ctx, job.ID, msg); derr != nil { - w.opts.Log.Error("job.dead_letter_write_failed", "id", job.ID, "err", derr.Error()) + w.opts.Log.Error("job.dead_letter_write_failed", "job_id", job.ID, "err", derr.Error()) } w.opts.Log.Error("job.dead_letter", - "id", job.ID, "type", job.Type, "attempts", job.Attempts, "err", msg) + "job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts, "err", msg) w.fireOnDeadLetter(ctx, job, err) return } @@ -127,10 +127,10 @@ func (w *Worker) execute(ctx context.Context, job Job) { next := w.opts.Clock.Now().Add(delay) msg := truncate(err.Error(), 2000) if ferr := w.repo.FailWithRetry(ctx, job.ID, next, msg); ferr != nil { - w.opts.Log.Error("job.fail_retry_write_failed", "id", job.ID, "err", ferr.Error()) + w.opts.Log.Error("job.fail_retry_write_failed", "job_id", job.ID, "err", ferr.Error()) } w.opts.Log.Warn("job.fail_retry", - "id", job.ID, "type", job.Type, "attempt", job.Attempts, + "job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts, "err", msg, "next_at", next.Format(time.RFC3339)) } @@ -164,7 +164,7 @@ func (w *Worker) fireOnDeadLetter(ctx context.Context, job Job, err error) { } defer func() { if r := recover(); r != nil { - w.opts.Log.Error("job.dead_letter_hook_panic", "id", job.ID, "panic", r) + w.opts.Log.Error("job.dead_letter_hook_panic", "job_id", job.ID, "panic", r) } }() w.opts.OnDeadLetter(ctx, job, err) diff --git a/migrations/0005_jobs.down.sql b/migrations/0005_jobs.down.sql index 31d6438..a5e0eb7 100644 --- a/migrations/0005_jobs.down.sql +++ b/migrations/0005_jobs.down.sql @@ -1,2 +1,4 @@ ALTER TABLE workspace_worktrees DROP COLUMN IF EXISTS prune_warning_at; +-- DROP TABLE jobs cascades to drop idx_jobs_status_scheduled, idx_jobs_leased, +-- and idx_jobs_completed_at automatically. DROP TABLE IF EXISTS jobs;