bugfix

internal/jobs/domain.go

@@ -83,6 +83,14 @@ func IsPermanent(err error) bool {
 	return errors.As(err, &pe)
 }
 
+// ErrLeaseLost is returned by the terminal/retry write-back methods
+// (Complete, FailWithRetry, DeadLetter) when the fencing predicate matches no
+// row: the job is no longer leased by this worker (e.g. the reaper re-leased it
+// to another worker after the lease timeout). The caller must treat this as a
+// no-op — the write was correctly dropped to avoid clobbering the new run's
+// state — and must NOT surface it as a failure that triggers further retries.
+var ErrLeaseLost = errors.New("job lease lost; write-back skipped")
+
 // RunnerConfig is the per-runner configuration loaded from config.yaml.
 // Zero Interval means "use the runner's documented default".
 type RunnerConfig struct {

internal/jobs/module.go

@@ -54,15 +54,17 @@ func NewModule(cfg Config, deps ModuleDeps) *Module {
 		sch.Add(name, entry.interval, entry.enabled, fn)
 	}
 
+	handlerTimeout := handlerTimeoutFor(cfg.JobReaper)
 	workers := make([]*Worker, cfg.Workers)
 	for i := 0; i < cfg.Workers; i++ {
 		workers[i] = NewWorker(deps.Repo, deps.Handlers, WorkerOptions{
-			ID:           workerID(i),
-			PollInterval: 200 * time.Millisecond,
-			Backoff:      deps.WorkerBackoff,
-			Clock:        clk,
-			Log:          log,
-			OnDeadLetter: deps.OnDeadLetter,
+			ID:             workerID(i),
+			PollInterval:   200 * time.Millisecond,
+			Backoff:        deps.WorkerBackoff,
+			Clock:          clk,
+			Log:            log,
+			HandlerTimeout: handlerTimeout,
+			OnDeadLetter:   deps.OnDeadLetter,
 		})
 	}
 	return &Module{cfg: cfg, scheduler: sch, workers: workers, log: log}
@@ -112,6 +114,19 @@ func (m *Module) Stop(ctx context.Context) {
 
 func workerID(i int) string { return "worker#" + strconv.Itoa(i) }
 
+// handlerTimeoutFor derives the per-job handler timeout from the reaper's lease
+// timeout, leaving a margin below it so a healthy long handler finishes (or is
+// canceled) before the reaper re-leases the job and double-runs it. Returns 0
+// (no timeout) when reaping is disabled or no positive lease timeout is set.
+func handlerTimeoutFor(rc JobReaperConfig) time.Duration {
+	if !rc.Enabled || rc.LeaseTimeout <= 0 {
+		return 0
+	}
+	// 90% of the lease window leaves headroom for the write-back to land before
+	// the reaper's cutoff.
+	return rc.LeaseTimeout - rc.LeaseTimeout/10
+}
+
 type schedEntry struct {
 	interval time.Duration
 	enabled  bool

internal/jobs/queries/jobs.sql

@@ -20,28 +20,31 @@ UPDATE jobs
  )
 RETURNING *;
 
--- name: CompleteJob :exec
+-- name: CompleteJob :execrows
+-- 围栏谓词：只有持有当前租约的 worker 才能写回，防止被 reaper 重新租出后旧 worker 覆盖新一轮状态。
 UPDATE jobs
    SET status='completed', completed_at=now(),
        leased_at=NULL, leased_by=NULL, last_error=NULL,
        updated_at=now()
- WHERE id=$1;
+ WHERE id=$1 AND leased_by=$2 AND status='running';
 
--- name: FailJobWithRetry :exec
+-- name: FailJobWithRetry :execrows
+-- 围栏谓词同上。
 UPDATE jobs
    SET status='pending',
        scheduled_at=$2,
        leased_at=NULL, leased_by=NULL,
        last_error=$3,
        updated_at=now()
- WHERE id=$1;
+ WHERE id=$1 AND leased_by=$4 AND status='running';
 
--- name: DeadLetterJob :exec
+-- name: DeadLetterJob :execrows
+-- 围栏谓词同上。
 UPDATE jobs
    SET status='dead', completed_at=now(),
        leased_at=NULL, leased_by=NULL, last_error=$2,
        updated_at=now()
- WHERE id=$1;
+ WHERE id=$1 AND leased_by=$3 AND status='running';
 
 -- name: GetJob :one
 SELECT * FROM jobs WHERE id=$1;

internal/jobs/repository.go

@@ -22,9 +22,13 @@ import (
 type Repository interface {
 	Enqueue(ctx context.Context, typ JobType, payload json.RawMessage, scheduledAt time.Time, maxAttempts int32) (*Job, error)
 	Lease(ctx context.Context, leasedBy string) (*Job, error)
-	Complete(ctx context.Context, id uuid.UUID) error
-	FailWithRetry(ctx context.Context, id uuid.UUID, scheduledAt time.Time, lastErr string) error
-	DeadLetter(ctx context.Context, id uuid.UUID, lastErr string) error
+	// Complete, FailWithRetry and DeadLetter take the leasing worker id and apply
+	// a fencing predicate (id + leased_by + status='running'); if the lease has
+	// been lost (e.g. the reaper re-leased the job) they return ErrLeaseLost so
+	// the caller drops the write instead of clobbering the new run's state.
+	Complete(ctx context.Context, id uuid.UUID, leasedBy string) error
+	FailWithRetry(ctx context.Context, id uuid.UUID, leasedBy string, scheduledAt time.Time, lastErr string) error
+	DeadLetter(ctx context.Context, id uuid.UUID, leasedBy string, lastErr string) error
 	Get(ctx context.Context, id uuid.UUID) (*Job, error)
 	ReapStuckRunningJobs(ctx context.Context, leasedBefore time.Time) ([]uuid.UUID, error)
 	PurgeCompleted(ctx context.Context, completedBefore time.Time) (int64, error)
@@ -90,11 +94,20 @@ func (r *PgRepository) Lease(ctx context.Context, leasedBy string) (*Job, error)
 	return rowToJob(row), nil
 }
 
-// Complete marks a job as completed and clears its lease/error fields.
-func (r *PgRepository) Complete(ctx context.Context, id uuid.UUID) error {
-	if err := r.q.CompleteJob(ctx, toPgUUID(id)); err != nil {
+// Complete marks a job as completed and clears its lease/error fields. The
+// update is fenced on the leasing worker id and status='running'; if no row
+// matches (lease lost to the reaper) it returns ErrLeaseLost as a no-op signal.
+func (r *PgRepository) Complete(ctx context.Context, id uuid.UUID, leasedBy string) error {
+	n, err := r.q.CompleteJob(ctx, jobssqlc.CompleteJobParams{
+		ID:       toPgUUID(id),
+		LeasedBy: &leasedBy,
+	})
+	if err != nil {
 		return errs.Wrap(err, errs.CodeInternal, "complete job")
 	}
+	if n == 0 {
+		return ErrLeaseLost
+	}
 	return nil
 }
 
@@ -102,25 +115,37 @@ func (r *PgRepository) Complete(ctx context.Context, id uuid.UUID) error {
 // the last error. Attempts is preserved (it was incremented on lease) so the
 // worker can compare against MaxAttempts to decide whether to retry or
 // dead-letter on the next failure.
-func (r *PgRepository) FailWithRetry(ctx context.Context, id uuid.UUID, scheduledAt time.Time, lastErr string) error {
-	if err := r.q.FailJobWithRetry(ctx, jobssqlc.FailJobWithRetryParams{
+func (r *PgRepository) FailWithRetry(ctx context.Context, id uuid.UUID, leasedBy string, scheduledAt time.Time, lastErr string) error {
+	n, err := r.q.FailJobWithRetry(ctx, jobssqlc.FailJobWithRetryParams{
 		ID:          toPgUUID(id),
 		ScheduledAt: pgtype.Timestamptz{Time: scheduledAt, Valid: true},
 		LastError:   &lastErr,
-	}); err != nil {
+		LeasedBy:    &leasedBy,
+	})
+	if err != nil {
 		return errs.Wrap(err, errs.CodeInternal, "fail job with retry")
 	}
+	if n == 0 {
+		return ErrLeaseLost
+	}
 	return nil
 }
 
-// DeadLetter terminally marks a job as dead and stores the last error.
-func (r *PgRepository) DeadLetter(ctx context.Context, id uuid.UUID, lastErr string) error {
-	if err := r.q.DeadLetterJob(ctx, jobssqlc.DeadLetterJobParams{
+// DeadLetter terminally marks a job as dead and stores the last error. Fenced
+// on the leasing worker id and status='running'; returns ErrLeaseLost on a
+// 0-row match so a stale worker does not overwrite a re-leased run.
+func (r *PgRepository) DeadLetter(ctx context.Context, id uuid.UUID, leasedBy string, lastErr string) error {
+	n, err := r.q.DeadLetterJob(ctx, jobssqlc.DeadLetterJobParams{
 		ID:        toPgUUID(id),
 		LastError: &lastErr,
-	}); err != nil {
+		LeasedBy:  &leasedBy,
+	})
+	if err != nil {
 		return errs.Wrap(err, errs.CodeInternal, "dead-letter job")
 	}
+	if n == 0 {
+		return ErrLeaseLost
+	}
 	return nil
 }
 

internal/jobs/repository_test.go

@@ -125,7 +125,7 @@ func TestComplete_ClearsLease(t *testing.T) {
 	require.NoError(t, err)
 	leased, err := repo.Lease(ctx, "h")
 	require.NoError(t, err)
-	require.NoError(t, repo.Complete(ctx, leased.ID))
+	require.NoError(t, repo.Complete(ctx, leased.ID, "h"))
 	got, err := repo.Get(ctx, job.ID)
 	require.NoError(t, err)
 	assert.Equal(t, jobs.StatusCompleted, got.Status)
@@ -133,6 +133,49 @@ func TestComplete_ClearsLease(t *testing.T) {
 	assert.Nil(t, got.LeasedBy)
 }
 
+// TestComplete_LeaseLostIsNoOp verifies the fencing predicate: a write-back
+// from a worker that no longer holds the lease (e.g. the reaper re-leased the
+// job) matches 0 rows and returns ErrLeaseLost without clobbering the new run.
+func TestComplete_LeaseLostIsNoOp(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	repo, pool := setupRepo(t)
+	_, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
+	require.NoError(t, err)
+
+	// Worker A leases the job, then its handler stalls past the lease timeout.
+	leased, err := repo.Lease(ctx, "workerA")
+	require.NoError(t, err)
+	// Reaper resets it to pending; worker B re-leases it (now leased_by=workerB).
+	_, err = pool.Exec(ctx, `UPDATE jobs SET leased_at = now() - INTERVAL '10 minutes' WHERE id=$1`, leased.ID)
+	require.NoError(t, err)
+	ids, err := repo.ReapStuckRunningJobs(ctx, time.Now().Add(-5*time.Minute))
+	require.NoError(t, err)
+	require.Len(t, ids, 1)
+	reLeased, err := repo.Lease(ctx, "workerB")
+	require.NoError(t, err)
+	require.Equal(t, leased.ID, reLeased.ID)
+
+	// Stale worker A's terminal write-backs must be no-ops (ErrLeaseLost) and
+	// must not change the row that workerB now owns.
+	assert.ErrorIs(t, repo.Complete(ctx, leased.ID, "workerA"), jobs.ErrLeaseLost)
+	assert.ErrorIs(t, repo.FailWithRetry(ctx, leased.ID, "workerA", time.Now(), "stale"), jobs.ErrLeaseLost)
+	assert.ErrorIs(t, repo.DeadLetter(ctx, leased.ID, "workerA", "stale"), jobs.ErrLeaseLost)
+
+	got, err := repo.Get(ctx, leased.ID)
+	require.NoError(t, err)
+	assert.Equal(t, jobs.StatusRunning, got.Status, "row must still be owned by workerB's run")
+	require.NotNil(t, got.LeasedBy)
+	assert.Equal(t, "workerB", *got.LeasedBy)
+	assert.Nil(t, got.LastError, "stale write must not have set an error")
+
+	// Worker B (the true owner) completes successfully.
+	require.NoError(t, repo.Complete(ctx, leased.ID, "workerB"))
+	got, err = repo.Get(ctx, leased.ID)
+	require.NoError(t, err)
+	assert.Equal(t, jobs.StatusCompleted, got.Status)
+}
+
 func TestFailWithRetry_ReschedulesAndKeepsAttempts(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -142,7 +185,7 @@ func TestFailWithRetry_ReschedulesAndKeepsAttempts(t *testing.T) {
 	leased, err := repo.Lease(ctx, "h")
 	require.NoError(t, err)
 	next := time.Now().Add(30 * time.Second)
-	require.NoError(t, repo.FailWithRetry(ctx, leased.ID, next, "boom"))
+	require.NoError(t, repo.FailWithRetry(ctx, leased.ID, "h", next, "boom"))
 	got, err := repo.Get(ctx, leased.ID)
 	require.NoError(t, err)
 	assert.Equal(t, jobs.StatusPending, got.Status)
@@ -160,7 +203,7 @@ func TestDeadLetter_FinalState(t *testing.T) {
 	require.NoError(t, err)
 	leased, err := repo.Lease(ctx, "h")
 	require.NoError(t, err)
-	require.NoError(t, repo.DeadLetter(ctx, leased.ID, "max attempts"))
+	require.NoError(t, repo.DeadLetter(ctx, leased.ID, "h", "max attempts"))
 	got, err := repo.Get(ctx, leased.ID)
 	require.NoError(t, err)
 	assert.Equal(t, jobs.StatusDead, got.Status)
@@ -204,7 +247,10 @@ func TestPurge_DeletesOldCompletedAndDead(t *testing.T) {
 	deadOld := mkOld(jobs.StatusDead)
 	freshComp, err := repo.Enqueue(ctx, jobs.TypeWebhookDeliver, json.RawMessage(`{}`), time.Now(), 5)
 	require.NoError(t, err)
-	require.NoError(t, repo.Complete(ctx, freshComp.ID))
+	leasedFresh, err := repo.Lease(ctx, "h")
+	require.NoError(t, err)
+	require.Equal(t, freshComp.ID, leasedFresh.ID)
+	require.NoError(t, repo.Complete(ctx, freshComp.ID, "h"))
 
 	rows, err := repo.PurgeCompleted(ctx, time.Now().Add(-7*24*time.Hour))
 	require.NoError(t, err)

internal/jobs/sqlc/jobs.sql.go

@@ -11,35 +11,49 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 )
 
-const completeJob = `-- name: CompleteJob :exec
+const completeJob = `-- name: CompleteJob :execrows
 UPDATE jobs
    SET status='completed', completed_at=now(),
        leased_at=NULL, leased_by=NULL, last_error=NULL,
        updated_at=now()
- WHERE id=$1
+ WHERE id=$1 AND leased_by=$2 AND status='running'
 `
 
-func (q *Queries) CompleteJob(ctx context.Context, id pgtype.UUID) error {
-	_, err := q.db.Exec(ctx, completeJob, id)
-	return err
+type CompleteJobParams struct {
+	ID       pgtype.UUID `json:"id"`
+	LeasedBy *string     `json:"leased_by"`
 }
 
-const deadLetterJob = `-- name: DeadLetterJob :exec
+// 围栏谓词：只有持有当前租约的 worker 才能写回，防止被 reaper 重新租出后旧 worker 覆盖新一轮状态。
+func (q *Queries) CompleteJob(ctx context.Context, arg CompleteJobParams) (int64, error) {
+	result, err := q.db.Exec(ctx, completeJob, arg.ID, arg.LeasedBy)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected(), nil
+}
+
+const deadLetterJob = `-- name: DeadLetterJob :execrows
 UPDATE jobs
    SET status='dead', completed_at=now(),
        leased_at=NULL, leased_by=NULL, last_error=$2,
        updated_at=now()
- WHERE id=$1
+ WHERE id=$1 AND leased_by=$3 AND status='running'
 `
 
 type DeadLetterJobParams struct {
 	ID        pgtype.UUID `json:"id"`
 	LastError *string     `json:"last_error"`
+	LeasedBy  *string     `json:"leased_by"`
 }
 
-func (q *Queries) DeadLetterJob(ctx context.Context, arg DeadLetterJobParams) error {
-	_, err := q.db.Exec(ctx, deadLetterJob, arg.ID, arg.LastError)
-	return err
+// 围栏谓词同上。
+func (q *Queries) DeadLetterJob(ctx context.Context, arg DeadLetterJobParams) (int64, error) {
+	result, err := q.db.Exec(ctx, deadLetterJob, arg.ID, arg.LastError, arg.LeasedBy)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected(), nil
 }
 
 const enqueueJob = `-- name: EnqueueJob :one
@@ -83,25 +97,35 @@ func (q *Queries) EnqueueJob(ctx context.Context, arg EnqueueJobParams) (Job, er
 	return i, err
 }
 
-const failJobWithRetry = `-- name: FailJobWithRetry :exec
+const failJobWithRetry = `-- name: FailJobWithRetry :execrows
 UPDATE jobs
    SET status='pending',
        scheduled_at=$2,
        leased_at=NULL, leased_by=NULL,
        last_error=$3,
        updated_at=now()
- WHERE id=$1
+ WHERE id=$1 AND leased_by=$4 AND status='running'
 `
 
 type FailJobWithRetryParams struct {
 	ID          pgtype.UUID        `json:"id"`
 	ScheduledAt pgtype.Timestamptz `json:"scheduled_at"`
 	LastError   *string            `json:"last_error"`
+	LeasedBy    *string            `json:"leased_by"`
 }
 
-func (q *Queries) FailJobWithRetry(ctx context.Context, arg FailJobWithRetryParams) error {
-	_, err := q.db.Exec(ctx, failJobWithRetry, arg.ID, arg.ScheduledAt, arg.LastError)
-	return err
+// 围栏谓词同上。
+func (q *Queries) FailJobWithRetry(ctx context.Context, arg FailJobWithRetryParams) (int64, error) {
+	result, err := q.db.Exec(ctx, failJobWithRetry,
+		arg.ID,
+		arg.ScheduledAt,
+		arg.LastError,
+		arg.LeasedBy,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected(), nil
 }
 
 const getJob = `-- name: GetJob :one

internal/jobs/worker.go

@@ -20,6 +20,12 @@ type WorkerOptions struct {
 	Backoff      []time.Duration
 	Clock        clock.Clock
 	Log          *slog.Logger
+	// HandlerTimeout bounds a single handler invocation. It must be set slightly
+	// below the JobReaper's LeaseTimeout so a healthy long-running handler
+	// finishes (or is canceled) before the reaper re-leases the job to another
+	// worker and double-runs it. When <= 0 no per-job timeout is applied (the
+	// module always sets it from JobReaper.LeaseTimeout when reaping is enabled).
+	HandlerTimeout time.Duration
 	// OnDeadLetter, if non-nil, is invoked once after a job is moved to dead-letter
 	// (either because IsPermanent(err) or attempts exhausted). The hook runs in
 	// the worker goroutine; panics are recovered and logged but do not stop the
@@ -91,7 +97,9 @@ func (w *Worker) execute(ctx context.Context, job Job) {
 
 	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 {
+		if derr := w.repo.DeadLetter(ctx, job.ID, w.opts.ID, fmt.Sprintf("unknown job type: %s", job.Type)); w.leaseLost(derr, job, "dead_letter") {
+			return
+		} else if derr != nil {
 			w.opts.Log.Error("job.dead_letter_write_failed", "job_id", job.ID, "err", derr.Error())
 		}
 		w.opts.Log.Error("job.dead_letter", "job_id", job.ID, "job_type", job.Type, "reason", "unknown_type")
@@ -101,7 +109,9 @@ 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 {
+		if cerr := w.repo.Complete(ctx, job.ID, w.opts.ID); w.leaseLost(cerr, job, "complete") {
+			return
+		} else if cerr != nil {
 			w.opts.Log.Error("job.complete_write_failed", "job_id", job.ID, "err", cerr.Error())
 		}
 		w.opts.Log.Info("job.complete",
@@ -113,7 +123,9 @@ func (w *Worker) execute(ctx context.Context, job Job) {
 	// Permanent or attempts exhausted -> dead-letter
 	if IsPermanent(err) || job.Attempts >= job.MaxAttempts {
 		msg := truncate(err.Error(), 2000)
-		if derr := w.repo.DeadLetter(ctx, job.ID, msg); derr != nil {
+		if derr := w.repo.DeadLetter(ctx, job.ID, w.opts.ID, msg); w.leaseLost(derr, job, "dead_letter") {
+			return
+		} else if derr != nil {
 			w.opts.Log.Error("job.dead_letter_write_failed", "job_id", job.ID, "err", derr.Error())
 		}
 		w.opts.Log.Error("job.dead_letter",
@@ -126,7 +138,9 @@ func (w *Worker) execute(ctx context.Context, job Job) {
 	delay := backoffFor(w.opts.Backoff, int(job.Attempts))
 	next := w.opts.Clock.Now().Add(delay)
 	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, w.opts.ID, next, msg); w.leaseLost(ferr, job, "fail_retry") {
+		return
+	} else if ferr != nil {
 		w.opts.Log.Error("job.fail_retry_write_failed", "job_id", job.ID, "err", ferr.Error())
 	}
 	w.opts.Log.Warn("job.fail_retry",
@@ -134,8 +148,30 @@ func (w *Worker) execute(ctx context.Context, job Job) {
 		"err", msg, "next_at", next.Format(time.RFC3339))
 }
 
-// runHandler wraps Handle with panic recovery, converting panics into retryable errors.
+// leaseLost reports whether writeErr is ErrLeaseLost. The fencing predicate on
+// the terminal/retry updates matched no row, meaning this worker's lease was
+// stolen (e.g. its handler outran LeaseTimeout and the reaper re-leased the
+// job). The write is correctly dropped: we only log and drop, and must NOT
+// treat it as a failure that schedules further work — the new run owns the job.
+func (w *Worker) leaseLost(writeErr error, job Job, phase string) bool {
+	if errors.Is(writeErr, ErrLeaseLost) {
+		w.opts.Log.Warn("job.lease_lost",
+			"job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts, "phase", phase)
+		return true
+	}
+	return false
+}
+
+// runHandler wraps Handle with panic recovery (converting panics into retryable
+// errors) and bounds the handler to HandlerTimeout. Deriving a per-job timeout
+// below the reaper's LeaseTimeout ensures a healthy long-running handler is
+// canceled before the reaper re-leases the job, so it is never double-run.
 func (w *Worker) runHandler(ctx context.Context, h Handler, job Job) (err error) {
+	if w.opts.HandlerTimeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, w.opts.HandlerTimeout)
+		defer cancel()
+	}
 	defer func() {
 		if r := recover(); r != nil {
 			err = fmt.Errorf("handler panic: %v", r)