The Coordinator — Lock, Decide, Persist, Commit

You have a state machine that decides and a repository that remembers. The coordinator joins them. Every input — an API call, a worker report, a fired timer — becomes an event and goes through the same four steps:

Because the decision and its effects commit together, nobody ever sees a half-applied transition. Because of the lock, events for one run never interleave. You write this in pkg/coord/engine.go, tx.go and persist.go.

#Step 1: A transaction with after-commit hooks

Some side effects must happen only if the transaction commits: "task completed" metrics, "workflow failed" logs. If you emit them inside the transaction and it rolls back, your dashboards lie.

Create pkg/coord/tx.go:

// pkg/coord/tx.go
package coord

// txn is a transaction plus side effects (metrics, logs) that must only
// happen if it commits.
type txn struct {
    pgx.Tx
    afterCommit []func()
}

func (t *txn) onCommit(fn func()) {
    t.afterCommit = append(t.afterCommit, fn)
}

// inTx runs fn in a transaction and commits it if fn succeeds.
func (c *Coordinator) inTx(ctx context.Context, fn func(t *txn) error) error {
    tx, err := c.DB.Begin(ctx)
    if err != nil {
        return fmt.Errorf("begin transaction: %w", err)
    }
    defer tx.Rollback(ctx) //nolint:errcheck // no-op after commit

    t := &txn{Tx: tx}
    if err := fn(t); err != nil {
        return err
    }
    if err := tx.Commit(ctx); err != nil {
        return fmt.Errorf("commit transaction: %w", err)
    }
    for _, f := range t.afterCommit {
        f()
    }
    return nil
}

txn embeds pgx.Tx, so it satisfies db.Querier and can be passed straight into every repository method.

#Step 2: The handle loop

Create pkg/coord/engine.go. The core is handleWith: lock and load the run, let the caller pick the definition and event, then decide and persist.

// pkg/coord/engine.go

// handle applies ev to a run on the definition version the run is pinned to.
func (c *Coordinator) handle(ctx context.Context, runID string, logger zerolog.Logger, ev statemachine.Event) (*statemachine.Transition, error) {
    return c.handleWith(ctx, runID, logger, func(run *statemachine.Run) (*model.WorkflowDAG, statemachine.Event, error) {
        dag, err := c.loadDAG(ctx, run.WorkflowID, run.WorkflowVersion)
        return dag, ev, err
    })
}

// handleWith locks and loads the run, lets prepare pick the definition and
// event, then decides and persists — all in one transaction.
func (c *Coordinator) handleWith(ctx context.Context, runID string, logger zerolog.Logger,
    prepare func(*statemachine.Run) (*model.WorkflowDAG, statemachine.Event, error)) (*statemachine.Transition, error) {

    var tr *statemachine.Transition
    err := c.inTx(ctx, func(t *txn) error {
        run, err := c.repo.LoadRun(ctx, t, runID)
        if err != nil {
            return err
        }
        dag, ev, err := prepare(run)
        if err != nil {
            return err
        }
        tr, err = c.decide(ctx, t, logger, dag, run, ev, time.Now())
        return err
    })
    if err != nil {
        logger.Error().Err(err).Msg("event not applied")
    }
    return tr, err
}

Why the prepare callback? Most events use the definition version the run is pinned to (run.WorkflowVersion). A retry may move the run to a different version, and it can only know which one after reading the locked run. The callback keeps that choice inside the lock.

decide builds a fresh machine, applies the event, and persists:

// decide runs the state machine for one event and persists its transition.
func (c *Coordinator) decide(ctx context.Context, t *txn, logger zerolog.Logger, dag *model.WorkflowDAG,
    run *statemachine.Run, ev statemachine.Event, scheduledAt time.Time) (*statemachine.Transition, error) {

    machine := statemachine.New(dag, run, statemachine.Deps{
        Templates: c.Template,
        Store:     runStore{ctx: ctx, q: t, repo: c.repo, runID: run.ID},
        Now:       time.Now,
        NewID:     newTaskID,
    })
    tr, err := machine.Apply(ev)
    if err != nil {
        return nil, err
    }
    logger.Debug().Str("event", fmt.Sprintf("%T", ev)).Int("actions", len(tr.Actions)).Str("ignored", tr.Ignored).
        Msg("event applied")
    if tr.Ignored != "" {
        return tr, nil
    }
    p := &persister{c: c, ctx: ctx, t: t, logger: logger, run: run, scheduledAt: scheduledAt}
    return tr, p.persist(tr.Actions)
}

Here the real dependencies are plugged in: time.Now for the clock, UUIDv7 for IDs, and a runStore that reads inside the same locked transaction:

// runStore is the machine's on-demand read access to the locked run.
type runStore struct {
    ctx   context.Context
    q     db.Querier
    repo  repository
    runID string
}

func (s runStore) Outputs(nodeIDs []string) (map[string]json.RawMessage, error) {
    return s.repo.LoadOutputs(s.ctx, s.q, s.runID, nodeIDs)
}

func (s runStore) PendingSignal(name string) (*statemachine.Signal, error) {
    return s.repo.PendingSignal(s.ctx, s.q, s.runID, name)
}

// newTaskID returns a time-ordered task ID, which keeps index inserts local.
func newTaskID() string {
    id, err := uuid.NewV7()
    if err != nil {
        return uuid.NewString()
    }
    return id.String()
}

Time-ordered task IDs mean new index entries land at the right edge of the B-tree, instead of at random pages all over it.

#Step 3: Turn actions into SQL

Create pkg/coord/persist.go. The persister walks the actions in order and calls one repository method per action:

// pkg/coord/persist.go

// errStaleSnapshot means a write did not find the row state the machine
// decided on. Under the run lock this cannot happen unless something wrote
// outside the engine; the transaction is rolled back and the event fails.
var errStaleSnapshot = errors.New("run changed under the state machine")

// persister writes one transition's actions inside the run's transaction.
type persister struct {
    c           *Coordinator
    ctx         context.Context
    t           *txn
    logger      zerolog.Logger
    run         *statemachine.Run
    scheduledAt time.Time
}

func (p *persister) persist(actions []statemachine.Action) error {
    for _, a := range actions {
        if err := p.apply(a); err != nil {
            return fmt.Errorf("persist %T: %w", a, err)
        }
    }
    return nil
}

The big type switch maps each action to its write:

func (p *persister) apply(action statemachine.Action) error {
    repo, ctx, q, runID := p.c.repo, p.ctx, p.t, p.run.ID

    switch a := action.(type) {
    case statemachine.CreateTask:
        inserted, err := repo.InsertTask(ctx, q, runID, a, p.scheduledAt)
        if err != nil {
            return err
        }
        if !inserted {
            return errStaleSnapshot
        }
        if a.Status != model.TaskInstanceStatusSkipped {
            nodeType := a.Node.GetNodeTypeForRouting()
            p.t.onCommit(func() { util.RecordTaskCreated(ctx, a.TaskID, runID, a.Node.NodeID, nodeType) })
        }
        return nil

    case statemachine.DispatchTask:
        return repo.EnqueueDispatch(ctx, q, p.dispatch(a.TaskID, a.Node.NodeID, a.Node.GetNodeTypeForRouting(),
            a.Node.Queue(), a.Attempt, a.Input), a.NotBefore)

    case statemachine.CompleteTask:
        task, ok, err := repo.CompleteTask(ctx, q, p.ref(a.TaskID, a.NodeID, a.Attempt), a.Output)
        if err != nil || !ok {
            return orStale(err)
        }
        p.t.onCommit(func() {
            util.RecordTaskCompleted(ctx, a.TaskID, runID, a.NodeID, task.NodeType, sinceOrZero(task.StartedAt))
        })
        return nil

    case statemachine.FailTask:
        task, ok, err := repo.FailTask(ctx, q, p.ref(a.TaskID, a.NodeID, a.Attempt), a.Reason)
        if err != nil || !ok {
            return orStale(err)
        }
        p.t.onCommit(func() {
            util.RecordTaskFailed(ctx, a.TaskID, runID, a.NodeID, task.NodeType, sinceOrZero(task.StartedAt), a.Reason)
            p.logger.Warn().Str("node_id", a.NodeID).Str("reason", a.Reason).Msg("task failed")
        })
        return nil

    case statemachine.RetryTask:
        d, ok, err := repo.RequeueTask(ctx, q, p.ref(a.TaskID, a.NodeID, a.Attempt), a.Reason)
        if err != nil || !ok {
            return orStale(err)
        }
        d.WorkflowID, d.WorkflowVersion = p.run.WorkflowID, p.run.WorkflowVersion
        p.t.onCommit(func() {
            p.logger.Warn().Str("node_id", a.NodeID).Int("next_attempt", d.Attempt).
                Time("not_before", a.NotBefore).Msg("task failed, retry scheduled")
        })
        return repo.EnqueueDispatch(ctx, q, d, a.NotBefore)

    case statemachine.ResolveWait:
        ok, err := repo.ResolveWait(ctx, q, runID, a)
        if err != nil || !ok {
            return orStale(err)
        }
        return nil

    case statemachine.BufferSignal:
        return repo.BufferSignal(ctx, q, runID, a)

    case statemachine.ConsumeSignal:
        return repo.ConsumeSignal(ctx, q, runID, a.SignalID)

    case statemachine.ReopenRun:
        if err := repo.ResetRunForRetry(ctx, q, runID, a.Version, a.Context.Input); err != nil {
            return err
        }
        if err := repo.SaveRunContext(ctx, q, runID, a.Context); err != nil {
            return err
        }
        return repo.DeleteTasks(ctx, q, runID, a.ResetNodes)

    case statemachine.CancelTasks:
        return repo.CancelRunTasks(ctx, q, runID, a.Reason)

    case statemachine.FinishRun:
        duration, changed, err := repo.FinishRun(ctx, q, runID, a.Status, a.Reason)
        if err != nil || !changed {
            return orStale(err)
        }
        workflowID := p.run.WorkflowID
        p.t.onCommit(func() {
            if a.Status == model.WorkflowInstanceStatusFailed {
                util.RecordWorkflowFailed(ctx, workflowID, runID, duration, a.Reason)
                p.logger.Error().Str("reason", a.Reason).Msg("workflow failed")
            } else {
                util.RecordWorkflowCompleted(ctx, workflowID, runID, duration)
                p.logger.Info().Dur("duration", duration).Msg("workflow succeeded")
            }
        })
        return nil

    default:
        return fmt.Errorf("unknown action %T", action)
    }
}

Read it as a table:

Action	SQL
CreateTask	INSERT task_instances … ON CONFLICT DO NOTHING
DispatchTask	INSERT outbox (next page)
CompleteTask / FailTask	fenced UPDATE task_instances
RetryTask	fenced UPDATE (attempt+1) and INSERT outbox with next_attempt_at = backoff
ResolveWait	UPDATE … WHERE state = 'WAITING'
BufferSignal / ConsumeSignal	INSERT / UPDATE workflow_events
ReopenRun	reset run row, save context, DELETE reset tasks
FinishRun	UPDATE workflow_runs … WHERE status IN ('RUNNING','PENDING')

Any write that matches zero rows becomes errStaleSnapshot, which rolls the whole transaction back:

// orStale turns "no row matched" (err == nil) into errStaleSnapshot.
func orStale(err error) error {
    if err == nil {
        return errStaleSnapshot
    }
    return err
}

A retry's backoff costs nothing to implement: the outbox row is written with next_attempt_at in the future, and the outbox enqueuer simply will not publish it until then.

#Step 4: Turn inputs into events

Now write one small public method per input. Each one builds an event and calls handle. Start with worker reports:

// TaskResult is a worker's report for one task attempt.
type TaskResult struct {
    TaskID  string
    RunID   string
    NodeID  string
    Attempt int
    Output  json.RawMessage
    Error   string // non-empty for failures
}

// HandleNodeCompleted applies a worker's success report.
func (c *Coordinator) HandleNodeCompleted(ctx context.Context, res TaskResult) (err error) {
    ctx, span := util.StartTaskSpan(ctx, "completed", res.TaskID, res.RunID, res.NodeID)
    defer func() { util.EndSpan(span, err) }()
    _, err = c.handle(ctx, res.RunID, c.taskLogger(res), statemachine.NodeCompleted{
        TaskID: res.TaskID, NodeID: res.NodeID, Attempt: res.Attempt, Output: res.Output,
    })
    return err
}

// HandleNodeFailed applies a worker's failure report.
func (c *Coordinator) HandleNodeFailed(ctx context.Context, res TaskResult) (err error) {
    ctx, span := util.StartTaskSpan(ctx, "failed", res.TaskID, res.RunID, res.NodeID)
    defer func() { util.EndSpan(span, err) }()
    _, err = c.handle(ctx, res.RunID, c.taskLogger(res).With().Str("error", res.Error).Logger(), statemachine.NodeFailed{
        TaskID: res.TaskID, NodeID: res.NodeID, Attempt: res.Attempt, Error: res.Error,
    })
    return err
}

Signals, retries and cancels follow the same pattern:

// SignalWorkflow applies SignalReceived: an external event for the run's
// WAIT_FOR_EVENT nodes. It reports whether a waiting node consumed it (false:
// it was buffered for a node the run has not reached yet).
func (c *Coordinator) SignalWorkflow(ctx context.Context, runID, event string, payload json.RawMessage) (bool, error) {
    logger := c.Logger.With().Str("run_id", runID).Str("event", event).Logger()
    tr, err := c.handle(ctx, runID, logger, statemachine.SignalReceived{Name: event, Payload: payload})
    if err != nil {
        return false, err
    }
    return tr.SignalDelivered, nil
}

// RetryWorkflowInstance applies RunRetried: the run restarts from a node,
// optionally on another definition version (0: the deployed one) and with new
// input or execution options.
func (c *Coordinator) RetryWorkflowInstance(ctx context.Context, p *model.RetryWorkflowRequest) (err error) {
    ctx, span := util.StartWorkflowSpan(ctx, "retry", "", p.RunID)
    defer func() { util.EndSpan(span, err) }()
    logger := c.Logger.With().Str("run_id", p.RunID).Str("node_id", p.NodeID).Logger()

    _, err = c.handleWith(ctx, p.RunID, logger, func(run *statemachine.Run) (*model.WorkflowDAG, statemachine.Event, error) {
        version, err := c.resolveRetryVersion(ctx, run.WorkflowID, p.WorkflowVersion)
        if err != nil {
            return nil, nil, err
        }
        dag, err := c.loadDAG(ctx, run.WorkflowID, version)
        return dag, statemachine.RunRetried{FromNodeID: p.NodeID, Version: version,
            Input: p.Input, Config: p.ExecutionOptions}, err
    })
    return err
}

// CancelWorkflowInstance applies RunCancelled.
func (c *Coordinator) CancelWorkflowInstance(ctx context.Context, runID, reason string) error {
    _, err := c.handle(ctx, runID, c.Logger.With().Str("run_id", runID).Logger(), statemachine.RunCancelled{Reason: reason})
    return err
}

#Starting a run

Starting is the one case with no existing row to lock. Insert the run and apply RunStarted in the same transaction, so a run never exists without its first tasks:

// StartWorkflow creates a run and applies RunStarted, in one transaction.
func (c *Coordinator) StartWorkflow(ctx context.Context, p *model.StartWorkflowRequest) (err error) {
    if p.RunID, err = runIDFor(p.RunID); err != nil {
        return err
    }
    ctx, span := util.StartWorkflowSpan(ctx, "start", p.WorkflowID, p.RunID)
    defer func() { util.EndSpan(span, err) }()
    logger := c.Logger.With().Str("workflow_id", p.WorkflowID).Int("workflow_version", p.WorkflowVersion).
        Str("run_id", p.RunID).Logger()

    dag, err := c.loadDAG(ctx, p.WorkflowID, p.WorkflowVersion)
    if err != nil {
        return err
    }
    scheduledAt := time.Now()
    if p.StartedAtUnixMs > 0 {
        scheduledAt = time.UnixMilli(p.StartedAtUnixMs)
    }

    rc := model.NewRunContext(p.Input, p.ExecutionOptions)
    err = c.inTx(ctx, func(t *txn) error {
        if err := c.repo.InsertRun(ctx, t, p, rc); err != nil {
            return err
        }
        run := statemachine.NewRun(p.RunID, p.WorkflowID, p.WorkflowVersion, rc)
        _, err := c.decide(ctx, t, logger, dag, run, statemachine.RunStarted{StartNodeID: p.StartNodeID}, scheduledAt)
        return err
    })
    if err != nil {
        logger.Error().Err(err).Msg("failed to start workflow")
        return err
    }
    util.RecordWorkflowStarted(ctx, p.WorkflowID, p.RunID)
    logger.Info().Msg("workflow started")
    return nil
}

Until this transaction commits, the new run row is invisible to every other transaction, so no other event can touch the run in between.

#Validate client run IDs

Clients may pass their own run ID (handy for idempotent starts). Because tables are partitioned by run ID, it must be a UUIDv7 close to now:

// Client-supplied run IDs must be UUIDv7 close to the current time, so they
// land in an existing partition (run-scoped tables are partitioned by run ID).
const (
    runIDMaxAge  = 24 * time.Hour
    runIDMaxSkew = time.Hour
    // timerBatch is how many due waits one FireDueWaits call resolves.
    timerBatch = 200
)

// runIDFor returns a new UUIDv7 run ID, or validates a client-supplied one.
func runIDFor(requested string) (string, error) {
    if requested == "" {
        id, err := uuid.NewV7()
        return id.String(), err
    }
    id, err := uuid.Parse(requested)
    if err != nil || id.Version() != 7 {
        return "", fmt.Errorf("%w: run_id must be a UUIDv7", ErrInvalid)
    }
    sec, nsec := id.Time().UnixTime()
    if age := time.Since(time.Unix(sec, nsec)); age > runIDMaxAge || age < -runIDMaxSkew {
        return "", fmt.Errorf("%w: run_id timestamp must be within the last %v", ErrInvalid, runIDMaxAge)
    }
    return id.String(), nil
}

#Step 5: Cache definitions

Every event needs the run's WorkflowDAG. Parsing JSON and building indexes per event would be wasteful, and definition versions are immutable — a perfect cache key. Create pkg/coord/wf_cache.go:

// pkg/coord/wf_cache.go

// workflowCache wraps a theine cache for WorkflowDAG instances.
// W-TinyLFU admission policy gives better hit rates than a plain LRU/TTL map
// under skewed access patterns (a few hot workflows hit much more than others).
type workflowCache struct {
    c   *theine.Cache[string, *model.WorkflowDAG]
    ttl time.Duration
}

func wfCacheKey(workflowID string, version int) string {
    return fmt.Sprintf("%s@v%d", workflowID, version)
}

func (w *workflowCache) get(workflowID string, version int) (*model.WorkflowDAG, bool) {
    return w.c.Get(wfCacheKey(workflowID, version))
}

func (w *workflowCache) set(workflowID string, version int, dag *model.WorkflowDAG) {
    w.c.SetWithTTL(wfCacheKey(workflowID, version), dag, 1, w.ttl)
}

And the loader in engine.go:

// loadDAG returns the definition a run is pinned to, using the in-process
// cache. Deleted definitions still load so in-flight runs can finish.
func (c *Coordinator) loadDAG(ctx context.Context, workflowID string, version int) (*model.WorkflowDAG, error) {
    if dag, ok := c.wfCache.get(workflowID, version); ok {
        return dag, nil
    }
    var specJSON json.RawMessage
    err := c.DB.QueryRow(ctx,
        `SELECT spec_json FROM workflow_definitions WHERE id = $1 AND version = $2`,
        workflowID, version).Scan(&specJSON)
    if errors.Is(err, pgx.ErrNoRows) {
        return nil, fmt.Errorf("%w: workflow definition %s v%d", ErrNotFound, workflowID, version)
    }
    if err != nil {
        return nil, fmt.Errorf("load workflow definition: %w", err)
    }
    var spec model.WorkflowDefinitionSpec
    if err := json.Unmarshal(specJSON, &spec); err != nil {
        return nil, fmt.Errorf("parse workflow spec: %w", err)
    }
    dag := model.BuildDAG(&spec)
    c.wfCache.set(workflowID, version, dag)
    return dag, nil
}

WorkflowDAG is read-only after BuildDAG, so many goroutines can share one cached instance safely.

#Step 6: Expose the coordinator to workers over gRPC

Workers report results through gRPC. Define the service in pkg/api/proto/workflow.proto:

// ReportTaskResultRequest contains task execution results
message ReportTaskResultRequest {
  string task_id = 1;
  string status = 2;  // "completed", "failed"
  bytes output = 3;
  string error = 4;
  map<string, string> metadata = 5;  // legacy: run_id / node_id for older workers
  string run_id = 6;
  string node_id = 7;
  int32 attempt = 8;  // fencing token: must match the task's current attempt
}

// AcquireTaskLeaseRequest asks coordinator to claim a lease on a task instance
message AcquireTaskLeaseRequest {
  string task_id = 1;
  string owner = 2;
  int64 ttl_seconds = 3;
  string run_id = 4;  // routes the lookup to the run's partition
}

Generate code with make proto. Then implement the handler in pkg/api/grpc/handlers.go. It validates, converts, and calls the coordinator:

// pkg/api/grpc/handlers.go

// ReportTaskResult applies a worker's result for one task attempt.
func (s *WorkflowService) ReportTaskResult(ctx context.Context, req *workflowpb.ReportTaskResultRequest) (*workflowpb.ReportTaskResultResponse, error) {
    res := coord.TaskResult{
        TaskID:  req.GetTaskId(),
        RunID:   firstNonEmpty(req.GetRunId(), req.GetMetadata()["run_id"]),
        NodeID:  firstNonEmpty(req.GetNodeId(), req.GetMetadata()["node_id"]),
        Attempt: int(req.GetAttempt()),
        Output:  req.GetOutput(),
    }
    if res.TaskID == "" || res.RunID == "" || res.NodeID == "" {
        return nil, status.Error(codes.InvalidArgument, "task_id, run_id and node_id are required")
    }

    var err error
    switch req.GetStatus() {
    case "completed":
        err = s.coordinator.HandleNodeCompleted(ctx, res)
    case "failed":
        res.Error = firstNonEmpty(req.GetError(), "task execution failed")
        err = s.coordinator.HandleNodeFailed(ctx, res)
    default:
        return nil, status.Error(codes.InvalidArgument, "status must be 'completed' or 'failed'")
    }
    if err != nil {
        return nil, status.Error(codes.Internal, err.Error())
    }
    return &workflowpb.ReportTaskResultResponse{Acknowledged: true}, nil
}

A stale report is not an error. The machine ignores it, the transaction commits nothing, and the worker gets Acknowledged: true — which tells it to ack the message and move on. Only real failures (database down) return an error, so the worker keeps the message and tries again.

#What one event looks like end to end

The task t8 now exists and has an outbox row. Nothing has reached Redis yet. Getting it there quickly and reliably is the next page.