No lock renewal for in-progress work items — long activities lose their lock; timeout must exceed worst-case activity duration

[erma07]

A work item's LockExpiration is set once at dequeue (GetOrchestrationWorkItem sqlite.go:754/postgres.go:766; GetActivityWorkItem sqlite.go:847/postgres.go:856) and never extended during processing — the only hint is a TODO:

backend/sqlite/sqlite.go:250: // TODO: Support for stickiness, which would extend the LockExpiration
backend/postgres/postgres.go:253: // TODO: Support for stickiness, which would extend the LockExpiration
Complete*/Abandon* only clear the lock. So the lock timeout has to be set above the worst-case activity wall-clock; otherwise long activities shed locks (duplicate execution + ErrWorkItemLockLost). That couples crash-recovery latency to max activity duration.

Proposed approach

Add to the Backend interface (backend/backend.go):

// Extend the lock on a work item currently held by this worker.
// Returns ErrWorkItemLockLost if the lock is no longer held.
RenewOrchestrationWorkItemLock(ctx context.Context, wi OrchestrationWorkItem) error
RenewActivityWorkItemLock(ctx context.Context, wi ActivityWorkItem) error
ActivityWorkItem is keyed by SequenceNumber (backend/workitem.go:49) + LockedBy; OrchestrationWorkItem by InstanceID + LockedBy — same keys Complete/Abandon already use.

SQLite impl (backend/sqlite/sqlite.go), mirroring the existing optimistic guard:

func (be *sqliteBackend) RenewActivityWorkItemLock(ctx context.Context, wi *backend.ActivityWorkItem) error {
if err := be.ensureDB(); err != nil { return err }
newExp := time.Now().UTC().Add(be.options.ActivityLockTimeout)
res, err := be.db.ExecContext(ctx,
"UPDATE NewTasks SET [LockExpiration] = ? WHERE [SequenceNumber] = ? AND [LockedBy] = ?",
newExp, wi.SequenceNumber, wi.LockedBy)
if err != nil { return fmt.Errorf("renew activity lock: %w", err) }
if n, _ := res.RowsAffected(); n == 0 { return backend.ErrWorkItemLockLost }
return nil
}

func (be *sqliteBackend) RenewOrchestrationWorkItemLock(ctx context.Context, wi *backend.OrchestrationWorkItem) error {
if err := be.ensureDB(); err != nil { return err }
newExp := time.Now().UTC().Add(be.options.OrchestrationLockTimeout)
res, err := be.db.ExecContext(ctx,
"UPDATE Instances SET [LockExpiration] = ? WHERE [InstanceID] = ? AND [LockedBy] = ?",
newExp, string(wi.InstanceID), wi.LockedBy)
if err != nil { return fmt.Errorf("renew orchestration lock: %w", err) }
if n, _ := res.RowsAffected(); n == 0 { return backend.ErrWorkItemLockLost }
return nil
}
Postgres impl (backend/postgres/postgres.go) — same, with $N placeholders + pgxpool:

func (be *postgresBackend) RenewActivityWorkItemLock(ctx context.Context, wi *backend.ActivityWorkItem) error {
if err := be.ensureDB(); err != nil { return err }
newExp := time.Now().UTC().Add(be.options.ActivityLockTimeout)
tag, err := be.db.Exec(ctx,
UPDATE NewTasks SET LockExpiration = $1 WHERE SequenceNumber = $2 AND LockedBy = $3,
newExp, wi.SequenceNumber, wi.LockedBy)
if err != nil { return fmt.Errorf("renew activity lock: %w", err) }
if tag.RowsAffected() == 0 { return backend.ErrWorkItemLockLost }
return nil
}

func (be *postgresBackend) RenewOrchestrationWorkItemLock(ctx context.Context, wi *backend.OrchestrationWorkItem) error {
if err := be.ensureDB(); err != nil { return err }
newExp := time.Now().UTC().Add(be.options.OrchestrationLockTimeout)
tag, err := be.db.Exec(ctx,
UPDATE Instances SET LockExpiration = $1 WHERE InstanceID = $2 AND LockedBy = $3,
newExp, string(wi.InstanceID), wi.LockedBy)
if err != nil { return fmt.Errorf("renew orchestration lock: %w", err) }
if tag.RowsAffected() == 0 { return backend.ErrWorkItemLockLost }
return nil
}
In the work-item processing loop (backend/worker.go), run a renewal ticker at ≈ lockTimeout/3 for the duration of each item's processing; on ErrWorkItemLockLost, cancel the in-flight execution’s context. With renewal, the timeout governs crash detection rather than max activity duration.