You've already forked agentic-coding-workflow
chore(jobs): align log keys + add prune no-op test + down-migration comment
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.
This commit is contained in:
@@ -75,7 +75,7 @@ func (r *WorkspaceFetch) Run(ctx context.Context) {
|
|||||||
func (r *WorkspaceFetch) fetchOne(ctx context.Context, t WorkspaceFetchTarget) {
|
func (r *WorkspaceFetch) fetchOne(ctx context.Context, t WorkspaceFetchTarget) {
|
||||||
ok, err := r.repo.MarkSyncing(ctx, t.ID)
|
ok, err := r.repo.MarkSyncing(ctx, t.ID)
|
||||||
if err != nil {
|
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
|
return
|
||||||
}
|
}
|
||||||
if !ok {
|
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 {
|
if fetchErr := r.gitr.Fetch(ctx, t.MainPath, t.ID); fetchErr != nil {
|
||||||
msg := truncate(fetchErr.Error(), 2000)
|
msg := truncate(fetchErr.Error(), 2000)
|
||||||
if err := r.repo.MarkFetchResult(ctx, t.ID, false, msg); err != nil {
|
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
|
return
|
||||||
}
|
}
|
||||||
if derr := r.disp.Dispatch(ctx, notify.Message{
|
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(),
|
CreatedAt: time.Now().UTC(),
|
||||||
}); derr != nil {
|
}); 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
|
return
|
||||||
}
|
}
|
||||||
if err := r.repo.MarkFetchResult(ctx, t.ID, true, ""); err != nil {
|
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())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -75,7 +75,7 @@ func (r *WorktreePrune) warningPhase(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
for _, w := range rows {
|
for _, w := range rows {
|
||||||
if err := r.repo.MarkPruneWarning(ctx, w.ID); err != nil {
|
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
|
continue
|
||||||
}
|
}
|
||||||
if derr := r.disp.Dispatch(ctx, notify.Message{
|
if derr := r.disp.Dispatch(ctx, notify.Message{
|
||||||
@@ -90,7 +90,7 @@ func (r *WorktreePrune) warningPhase(ctx context.Context) {
|
|||||||
},
|
},
|
||||||
CreatedAt: time.Now().UTC(),
|
CreatedAt: time.Now().UTC(),
|
||||||
}); derr != nil {
|
}); 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
|
continue
|
||||||
}
|
}
|
||||||
if rmErr := r.gitr.WorktreeRemove(ctx, started.MainPath, started.Path); rmErr != nil {
|
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 {
|
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
|
continue
|
||||||
}
|
}
|
||||||
if err := r.repo.FinishPruneSuccess(ctx, w.ID); err != nil {
|
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())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -155,6 +155,23 @@ func TestWorktreePrune_ExecutionFailureRollsBack(t *testing.T) {
|
|||||||
assert.Equal(t, "rollback", repo.finalized[id])
|
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) {
|
func TestWorktreePrune_CASLostSkipsSilently(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
id := uuid.New()
|
id := uuid.New()
|
||||||
|
|||||||
+10
-10
@@ -86,15 +86,15 @@ func (w *Worker) pollOnce(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (w *Worker) execute(ctx context.Context, job Job) {
|
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()
|
start := w.opts.Clock.Now()
|
||||||
|
|
||||||
h, ok := w.handlers[job.Type]
|
h, ok := w.handlers[job.Type]
|
||||||
if !ok {
|
if !ok {
|
||||||
if derr := w.repo.DeadLetter(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type)); derr != nil {
|
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)))
|
w.fireOnDeadLetter(ctx, job, errors.New("unknown job type: "+string(job.Type)))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -102,10 +102,10 @@ func (w *Worker) execute(ctx context.Context, job Job) {
|
|||||||
err := w.runHandler(ctx, h, job)
|
err := w.runHandler(ctx, h, job)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
if cerr := w.repo.Complete(ctx, job.ID); cerr != 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",
|
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())
|
"duration_ms", w.opts.Clock.Now().Sub(start).Milliseconds())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -114,10 +114,10 @@ func (w *Worker) execute(ctx context.Context, job Job) {
|
|||||||
if IsPermanent(err) || job.Attempts >= job.MaxAttempts {
|
if IsPermanent(err) || job.Attempts >= job.MaxAttempts {
|
||||||
msg := truncate(err.Error(), 2000)
|
msg := truncate(err.Error(), 2000)
|
||||||
if derr := w.repo.DeadLetter(ctx, job.ID, msg); derr != nil {
|
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",
|
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)
|
w.fireOnDeadLetter(ctx, job, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -127,10 +127,10 @@ func (w *Worker) execute(ctx context.Context, job Job) {
|
|||||||
next := w.opts.Clock.Now().Add(delay)
|
next := w.opts.Clock.Now().Add(delay)
|
||||||
msg := truncate(err.Error(), 2000)
|
msg := truncate(err.Error(), 2000)
|
||||||
if ferr := w.repo.FailWithRetry(ctx, job.ID, next, msg); ferr != nil {
|
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",
|
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))
|
"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() {
|
defer func() {
|
||||||
if r := recover(); r != nil {
|
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)
|
w.opts.OnDeadLetter(ctx, job, err)
|
||||||
|
|||||||
@@ -1,2 +1,4 @@
|
|||||||
ALTER TABLE workspace_worktrees DROP COLUMN IF EXISTS prune_warning_at;
|
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;
|
DROP TABLE IF EXISTS jobs;
|
||||||
|
|||||||
Reference in New Issue
Block a user