Failure Recovery — Expired Leases, Lost Messages, Timers and Partitions

Every component you built assumes things can fail. This page adds the loops that notice failures and repair them. They all run inside every coordinator, and every one of them is safe to run on many replicas at the same time — no leader election needed.

#Step 1: Start the background loops

In pkg/coord/coordinator.go, define the intervals:

// pkg/coord/coordinator.go

const (
    // recoveryInterval is how often the coordinator sweeps for tasks whose
    // worker died (expired lease) or whose dispatch message was lost.
    recoveryInterval = 30 * time.Second
    // timerInterval is how often due WAIT timers and event timeouts are fired.
    timerInterval = time.Second
    // partitionInterval is how often partitions are created ahead / expired.
    partitionInterval = 6 * time.Hour
    // partitionMonthsAhead is how many future monthly partitions are kept ready.
    partitionMonthsAhead = 3
    // lostDispatchAge is how long a PENDING task may sit without an outbox
    // entry before it is assumed lost and re-dispatched.
    lostDispatchAge = 10 * time.Minute
)

And Run, which prepares storage, then starts everything:

// Run prepares storage, starts the coordinator's background loops and returns.
func (c *Coordinator) Run(ctx context.Context) error {
    c.Logger.Info().Int("stream_shards", c.Config.StreamShards).Msg("starting coordinator")

    // Partitions must exist before any run is inserted.
    if err := c.MaintainPartitions(ctx); err != nil {
        return err
    }
    // Workers read the shard count to know which streams to consume.
    if err := c.Redis.Client().Set(ctx, stream.ShardsKey, c.Config.StreamShards, 0).Err(); err != nil {
        return fmt.Errorf("publish stream shard count: %w", err)
    }

    // Recover before the outbox starts so re-dispatched tasks go out in its first poll.
    c.RecoverTasks(ctx)

    go c.Outbox.Run(ctx)
    go c.every(ctx, recoveryInterval, c.RecoverTasks)
    go c.every(ctx, timerInterval, c.fireTimers)
    go c.every(ctx, partitionInterval, func(ctx context.Context) {
        if err := c.MaintainPartitions(ctx); err != nil {
            c.Logger.Error().Err(err).Msg("partition maintenance failed")
        }
    })
    go c.every(ctx, 15*time.Second, c.sampleMetrics)
    return nil
}

// every calls fn on each tick until ctx is done.
func (c *Coordinator) every(ctx context.Context, interval time.Duration, fn func(context.Context)) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            fn(ctx)
        }
    }
}

The order at startup matters: partitions first (no run can be inserted without one), then the shard count (workers need it), then one recovery pass (so anything orphaned while the cluster was down goes out in the outbox's very first batch).

#Step 2: Recover tasks whose worker died

A worker that crashes mid-task leaves the task RUNNING with a lease that will never be renewed. Two safety nets exist:

1. Another worker's reclaim loop takes over the stream message (previous page). Its Acquire succeeds because the lease has expired.
2. The coordinator's lease sweeper, below, handles the cases the stream cannot — for example, the message was already acked by a worker that then lost its lease.

The sweeper resets expired tasks to PENDING and writes fresh outbox rows, in one statement:

// pkg/coord/repository_dispatch.go

// dispatchInsert is the outbox INSERT … SELECT shared by the recovery
// statements below; they supply FROM with task rows "t" and their run "r",
// and the shard count as parameter $1.
const dispatchInsert = `
    INSERT INTO outbox (id, task_id, stream, payload, type, status, next_attempt_at)
    SELECT gen_random_uuid()::text, t.id,
           '` + stream.TaskStreamPrefix + `' || COALESCE(t.queue, lower(t.node_type)) || '.'
               || (((hashtextextended(t.id, 0) % $1) + $1) % $1)::text,
           jsonb_build_object(
               'task_id', t.id, 'run_id', t.run_id,
               'workflow_id', r.workflow_definition_id, 'workflow_version', r.workflow_version,
               'node_id', t.node_id, 'node_type', t.node_type,
               'attempt', t.attempt, 'input', t.input_json::text),
           'TASK_EXECUTION', 'PENDING', NOW()`

// RequeueExpiredLeases resets RUNNING tasks whose lease expired — their worker
// died or lost contact — and re-dispatches them. Safe to run concurrently from
// several coordinators: each task row is claimed by exactly one UPDATE.
func (r repository) RequeueExpiredLeases(ctx context.Context, q db.Querier) (int64, error) {
    tag, err := q.Exec(ctx, `
        WITH expired AS (
            UPDATE task_instances
               SET state = 'PENDING', lease_owner = NULL, lease_expires_at = NULL,
                   started_at = NULL, updated_at = NOW()
             WHERE state = 'RUNNING' AND (lease_expires_at IS NULL OR lease_expires_at < NOW())
            RETURNING id, run_id, node_id, node_type, queue, attempt, input_json
        )`+dispatchInsert+`
          FROM expired t JOIN workflow_runs r ON r.id = t.run_id`,
        r.shards())
    if err != nil {
        return 0, fmt.Errorf("requeue expired leases: %w", err)
    }
    return tag.RowsAffected(), nil
}

The UPDATE … RETURNING inside a CTE feeds straight into the outbox INSERT. Because it is one statement, the reset and the re-dispatch are atomic. Because each UPDATE locks the rows it changes, two coordinators running this at once claim disjoint sets of tasks.

Notice the attempt does not change. The task is not being retried — its execution was interrupted. The same attempt is dispatched again, so if the "dead" worker was only slow and reports later, its report still matches the attempt and is applied once; the second execution's report then finds the task already COMPLETED and is ignored. Either way, exactly one result is applied.

The shard expression ((hash % n) + n) % n keeps the result non-negative, because hashtextextended can return negative numbers.

#Step 3: Recover lost dispatch messages

What if a message is lost entirely — Redis restarted without persistence, say? Then a task sits PENDING with no outbox row and no stream entry. Nobody will ever run it.

Detect exactly that shape: a PENDING task of a RUNNING run, untouched for ten minutes, with no outbox row:

// RedispatchPendingTasks re-dispatches PENDING tasks of running workflows that
// have no outbox entry and were not touched for olderThan — their stream
// message was lost. Touching updated_at limits each task to one re-dispatch per
// interval; a task still queued in the stream may run twice, which the lease
// deduplicates.
func (r repository) RedispatchPendingTasks(ctx context.Context, q db.Querier, olderThan time.Duration) (int64, error) {
    tag, err := q.Exec(ctx, `
        WITH lost AS (
            UPDATE task_instances t
               SET updated_at = NOW()
              FROM workflow_runs r
             WHERE r.id = t.run_id
               AND r.status = 'RUNNING'
               AND t.state = 'PENDING'
               AND t.updated_at < NOW() - make_interval(secs => $2)
               AND NOT EXISTS (SELECT 1 FROM outbox o WHERE o.task_id = t.id)
            RETURNING t.id, t.run_id, t.node_id, t.node_type, t.queue, t.attempt, t.input_json
        )`+dispatchInsert+`
          FROM lost t JOIN workflow_runs r ON r.id = t.run_id`,
        r.shards(), olderThan.Seconds())
    if err != nil {
        return 0, fmt.Errorf("redispatch pending tasks: %w", err)
    }
    return tag.RowsAffected(), nil
}

This is why the outbox row stores task_id: "does this task still have a pending dispatch?" becomes one indexed lookup.

Ten minutes is deliberately generous. A task might legitimately be PENDING because the workers are busy and its message is waiting in the stream. Re-dispatching it then would only create a duplicate — harmless thanks to the lease, but wasteful.

Both sweeps run together:

// RecoverTasks re-dispatches tasks orphaned by a dead worker or a lost stream
// message. Every coordinator runs it; the statements are safe to race.
func (c *Coordinator) RecoverTasks(ctx context.Context) {
    if n, err := c.repo.RequeueExpiredLeases(ctx, c.DB); err != nil {
        c.Logger.Error().Err(err).Msg("recovery: requeue expired leases failed")
    } else if n > 0 {
        c.Logger.Warn().Int64("count", n).Msg("recovery: re-dispatched tasks with expired leases")
    }

    if n, err := c.repo.RedispatchPendingTasks(ctx, c.DB, lostDispatchAge); err != nil {
        c.Logger.Error().Err(err).Msg("recovery: redispatch pending tasks failed")
    } else if n > 0 {
        c.Logger.Warn().Int64("count", n).Msg("recovery: re-dispatched pending tasks with lost messages")
    }
}

#Step 4: Fire timers

WAIT nodes and WAIT_FOR_EVENT timeouts store a wake_at. Once a second, every coordinator looks for due ones and applies TimerFired:

// pkg/coord/engine.go

// FireDueWaits applies TimerFired to every WAITING task whose wake-up time has
// passed and returns how many timers fired. Every coordinator runs it; the run
// lock makes concurrent calls safe (the loser's event is ignored).
func (c *Coordinator) FireDueWaits(ctx context.Context) (int, error) {
    due, err := c.repo.DueWaits(ctx, c.DB, timerBatch)
    if err != nil {
        return 0, err
    }
    fired := 0
    for _, w := range due {
        logger := c.Logger.With().Str("run_id", w.RunID).Str("node_id", w.NodeID).Str("task_id", w.TaskID).Logger()
        tr, err := c.handle(ctx, w.RunID, logger, statemachine.TimerFired{TaskID: w.TaskID, NodeID: w.NodeID})
        switch {
        case err != nil:
            logger.Error().Err(err).Msg("failed to fire timer")
        case tr.Ignored == "":
            fired++
        }
    }
    return fired, nil
}

// pkg/coord/coordinator.go
func (c *Coordinator) fireTimers(ctx context.Context) {
    if _, err := c.FireDueWaits(ctx); err != nil {
        c.Logger.Error().Err(err).Msg("firing due waits failed")
    }
}

Two coordinators may pick the same due wait. That is fine:

The state machine's guard in onTimerFired (task must still be WAITING) turns the duplicate into a no-op. No leader, no coordination protocol — just the run lock and an idempotent rule.

A timer resolution is a normal event, so a WAIT that completes routes forward and dispatches its successors in the same transaction, like any other completion.

#Step 5: Maintain partitions

// MaintainPartitions creates the monthly partitions of run-scoped tables ahead
// of time and, with RetentionDays set, drops months past retention.
func (c *Coordinator) MaintainPartitions(ctx context.Context) error {
    var created int
    if err := c.DB.QueryRow(ctx, `SELECT composer_ensure_partitions($1)`, partitionMonthsAhead).Scan(&created); err != nil {
        return fmt.Errorf("ensure partitions: %w", err)
    }
    if created > 0 {
        c.Logger.Info().Int("created", created).Msg("created run partitions")
    }
    if c.Config.RetentionDays <= 0 {
        return nil
    }
    rows, err := c.DB.Query(ctx,
        `SELECT composer_drop_partitions(NOW() - make_interval(days => $1))`, c.Config.RetentionDays)
    if err != nil {
        return fmt.Errorf("drop expired partitions: %w", err)
    }
    dropped, err := pgx.CollectRows(rows, pgx.RowTo[string])
    if err != nil {
        return fmt.Errorf("drop expired partitions: %w", err)
    }
    if len(dropped) > 0 {
        c.Logger.Info().Strs("months", dropped).Msg("dropped expired run partitions")
    }
    return nil
}

The SQL functions take an advisory lock, so many coordinators calling this together is safe.

#Step 6: Check every failure

Now walk through each failure and confirm the system heals:

Failure	What happens
Coordinator crashes mid-event	The transaction rolls back. The worker's ReportTaskResult fails, the message is not acked, and it is reclaimed and retried. Nothing is half-applied.
Worker dies mid-task	Its lease expires. Another worker reclaims the message and acquires the lease, and every coordinator's sweeper re-dispatches expired leases every 30 s.
Worker is slow, not dead	Its renewal fails once the lease is taken over; it cancels execution and discards the result.
Duplicate or late report	Ignored: results must match the task's current (task_id, attempt).
Enqueuer crashes after XADD, before DELETE	Row returns to RETRY after 2 minutes and is published again; the duplicate is dropped by the lease.
Redis loses messages	Tasks still PENDING without an outbox row are re-dispatched after 10 minutes.
Redis is down	Publishes fail, rows go to RETRY with backoff, and dispatch resumes when Redis returns.
A message keeps crashing workers	After MaxDeliveries it is reported as failed and the node's retry policy applies.
Step fails	Retry with backoff; when attempts run out the node fails, and the run fails once nothing else is active.

Every row ends in one of two places: the task completes once, or the task fails visibly. Nothing hangs.

#Step 7: Test recovery against real infrastructure

Unit tests cannot prove that leases and SQL sweeps work together. Use Testcontainers to start real PostgreSQL and Redis. In tests/engine_test.go, simulate a worker that leased a task and died:

// tests/engine_test.go
func TestExpiredLeaseIsRecovered(t *testing.T) {
    e := newEngine(t)
    runID := e.run(model.WorkflowDefinitionSpec{Nodes: []model.Node{e.httpNode("orphan")}}, `{}`)

    // Simulate a worker that leased the task and then died.
    var taskID string
    if err := e.infra.DBPool.QueryRow(e.ctx,
        `SELECT id FROM task_instances WHERE run_id = $1`, runID).Scan(&taskID); err != nil {
        t.Fatal(err)
    }
    if ok, err := e.coord.LeaseManager.Acquire(e.ctx, runID, taskID, "dead-worker", time.Second); err != nil || !ok {
        t.Fatalf("acquire: ok=%v err=%v", ok, err)
    }
    if ok, _ := e.coord.LeaseManager.Acquire(e.ctx, runID, taskID, "dead-worker", time.Second); ok {
        t.Fatal("an owner must not re-acquire a lease it still holds")
    }

    time.Sleep(1500 * time.Millisecond)
    e.coord.RecoverTasks(e.ctx)
    e.startWorker("survivor", 2, 10)

    if got := e.waitForRun(runID, 30*time.Second); got != model.WorkflowInstanceStatusSucceeded {
        t.Fatalf("status %s, tasks %v", got, e.nodeStates(runID))
    }
    if calls := e.svc.callCount("/ok/orphan"); calls != 1 {
        t.Fatalf("orphaned task executed %d times, want 1", calls)
    }
}

And prove NOTIFY really drives dispatch — set the poll interval to one minute, so a three-step run can only finish quickly if notifications work:

func TestOutboxIsWokenByNotify(t *testing.T) {
    e := newEngine(t, func(c *coord.Config) { c.OutboxConfig.PollInterval = time.Minute })
    e.startWorker("w1", 2, 10)
    time.Sleep(500 * time.Millisecond) // let the listener connect

    start := time.Now()
    runID := e.run(model.WorkflowDefinitionSpec{
        Nodes: []model.Node{e.httpNode("a"), e.httpNode("b"), e.httpNode("c")},
        Edges: []model.Edge{{From: "a", To: "b"}, {From: "b", To: "c"}},
    }, `{}`)
    if got := e.waitForRun(runID, 20*time.Second); got != model.WorkflowInstanceStatusSucceeded {
        t.Fatalf("status %s, tasks %v", got, e.nodeStates(runID))
    }
    if elapsed := time.Since(start); elapsed > 5*time.Second {
        t.Fatalf("three dispatches took %v; the outbox was not woken by NOTIFY", elapsed)
    }
}

go test ./tests/ -run 'TestExpiredLeaseIsRecovered|TestOutboxIsWokenByNotify' -v

The engine is complete. The last page runs the whole stack and drives a real workflow through it.