Workers and Leases — Read, Claim, Execute, Report

Tasks are in the streams. Now build the worker: a process that reads task messages, makes sure it is the only one running each task, executes it, and reports back.

You build the lease first, because everything on the worker depends on it.

#Step 1: The lease manager

The lease lives on the task row: lease_owner and lease_expires_at. Acquiring it is a single conditional UPDATE, so the database decides who wins — no distributed lock service needed.

Create pkg/lease/lease.go:

// pkg/lease/lease.go

// Package lease implements the database-backed task lease that guards against
// two workers executing the same task at the same time.
package lease

// Manager handles task lease operations.
type Manager struct {
    db db.DB
}

// NewManager creates a new lease manager.
func NewManager(db db.DB) *Manager {
    return &Manager{db: db}
}

// Acquire claims a task for owner and moves it to RUNNING. It succeeds when the
// task is waiting to run, or when it is RUNNING under a lease that has expired
// (the previous worker died). An owner never re-acquires a lease it still
// holds, so a redelivered message cannot start a second concurrent execution.
//
// runID is the task's run; it confines the lookup to the run's partition.
func (m *Manager) Acquire(ctx context.Context, runID, taskID, owner string, ttl time.Duration) (bool, error) {
    return m.exec(ctx, `
        UPDATE task_instances
           SET state = 'RUNNING',
               lease_owner = $1,
               lease_expires_at = NOW() + make_interval(secs => $2),
               started_at = NOW(),
               updated_at = NOW()
         WHERE run_id = $3 AND id = $4
           AND (state IN ('PENDING', 'QUEUED')
                OR (state = 'RUNNING' AND lease_expires_at < NOW()))
        RETURNING id`,
        owner, ttl.Seconds(), runID, taskID)
}

The WHERE clause is the whole policy:

Task state	Acquire?
PENDING / QUEUED	yes (first execution, or a retry)
RUNNING, lease still valid	no (someone is working on it)
RUNNING, lease expired	yes (previous worker died — take over)
COMPLETED / FAILED / SKIPPED	no (already done — the message is stale)

Notice the third row of "no": the same worker holding a valid lease is also refused. If a message is redelivered to the worker already running it, the second copy is dropped instead of starting a parallel execution.

Renewing only works for the current owner of a running task:

// Renew extends a lease that owner currently holds. It returns false once the
// lease was lost (expired and taken over, or the task finished), which tells
// the worker to abandon the execution.
func (m *Manager) Renew(ctx context.Context, runID, taskID, owner string, ttl time.Duration) (bool, error) {
    return m.exec(ctx, `
        UPDATE task_instances
           SET lease_expires_at = NOW() + make_interval(secs => $1),
               updated_at = NOW()
         WHERE run_id = $2 AND id = $3 AND state = 'RUNNING' AND lease_owner = $4
        RETURNING id`,
        ttl.Seconds(), runID, taskID, owner)
}

func (m *Manager) exec(ctx context.Context, query string, args ...any) (bool, error) {
    var id string
    err := m.db.QueryRow(ctx, query, args...).Scan(&id)
    if errors.Is(err, pgx.ErrNoRows) {
        return false, nil
    }
    if err != nil {
        return false, fmt.Errorf("task lease: %w", err)
    }
    return true, nil
}

Both use NOW() from the database, never the worker's clock. Clock skew between worker machines cannot break the lease.

#Step 2: Serve leases over gRPC

The worker cannot reach PostgreSQL, so the coordinator exposes the lease. Add to pkg/api/grpc/handlers.go:

// AcquireTaskLease lets a worker claim a task before executing it.
func (s *WorkflowService) AcquireTaskLease(ctx context.Context, req *workflowpb.AcquireTaskLeaseRequest) (*workflowpb.AcquireTaskLeaseResponse, error) {
    return s.lease(ctx, req, false)
}

// RenewTaskLease extends a lease the worker holds while the task runs.
func (s *WorkflowService) RenewTaskLease(ctx context.Context, req *workflowpb.AcquireTaskLeaseRequest) (*workflowpb.AcquireTaskLeaseResponse, error) {
    return s.lease(ctx, req, true)
}

func (s *WorkflowService) lease(ctx context.Context, req *workflowpb.AcquireTaskLeaseRequest, renew bool) (*workflowpb.AcquireTaskLeaseResponse, error) {
    if req.GetTaskId() == "" || req.GetRunId() == "" || req.GetOwner() == "" {
        return nil, status.Error(codes.InvalidArgument, "task_id, run_id and owner are required")
    }
    ttl := time.Duration(req.GetTtlSeconds()) * time.Second
    if ttl <= 0 {
        ttl = defaultLeaseTTL
    }

    leases := s.coordinator.LeaseManager
    op := leases.Acquire
    if renew {
        op = leases.Renew
    }
    acquired, err := op(ctx, req.GetRunId(), req.GetTaskId(), req.GetOwner(), ttl)
    if err != nil {
        return nil, status.Error(codes.Internal, err.Error())
    }
    return &workflowpb.AcquireTaskLeaseResponse{Acquired: acquired}, nil
}

The lease is deliberately not a state-machine event. It does not change routing — it only says who may execute. Keeping it a single fast UPDATE, outside the run lock, means leasing never waits behind other events of the same run.

#Step 3: The coordinator client

On the worker side, wrap the gRPC calls. Create worker/coordinator_client.go:

// worker/coordinator_client.go

// AcquireTaskLease asks the coordinator to claim a task for owner.
func (c *CoordinatorClient) AcquireTaskLease(ctx context.Context, msg *TaskMessage, owner string, ttlSeconds int64) (bool, error) {
    return c.lease(ctx, c.client.AcquireTaskLease, msg, owner, ttlSeconds)
}

// RenewTaskLease extends a lease owner holds; false means the lease was lost.
func (c *CoordinatorClient) RenewTaskLease(ctx context.Context, msg *TaskMessage, owner string, ttlSeconds int64) (bool, error) {
    return c.lease(ctx, c.client.RenewTaskLease, msg, owner, ttlSeconds)
}

type leaseRPC func(context.Context, *workflowpb.AcquireTaskLeaseRequest, ...grpc.CallOption) (*workflowpb.AcquireTaskLeaseResponse, error)

func (c *CoordinatorClient) lease(ctx context.Context, rpc leaseRPC, msg *TaskMessage, owner string, ttlSeconds int64) (bool, error) {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    resp, err := rpc(ctx, &workflowpb.AcquireTaskLeaseRequest{
        TaskId:     msg.TaskID,
        RunId:      msg.RunID,
        Owner:      owner,
        TtlSeconds: ttlSeconds,
    })
    if err != nil {
        return false, fmt.Errorf("task lease rpc: %w", err)
    }
    return resp.GetAcquired(), nil
}

// ReportTaskResult reports the outcome of one task attempt. errMsg is nil on success.
func (c *CoordinatorClient) ReportTaskResult(ctx context.Context, msg *TaskMessage, output json.RawMessage, errMsg *string) error {
    req := &workflowpb.ReportTaskResultRequest{
        TaskId:  msg.TaskID,
        RunId:   msg.RunID,
        NodeId:  msg.NodeID,
        Attempt: int32(msg.Attempt),
        Status:  "completed",
        Output:  output,
    }
    if errMsg != nil {
        req.Status = "failed"
        req.Error = *errMsg
    }

    ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    resp, err := c.client.ReportTaskResult(ctx, req)
    if err != nil {
        return fmt.Errorf("failed to report task result: %w", err)
    }
    if !resp.GetAcknowledged() {
        return fmt.Errorf("coordinator did not acknowledge task result")
    }
    return nil
}

The report carries Attempt straight from the message — the fencing token travels the whole loop: state machine → outbox → stream → worker → report → state machine.

#Step 4: The worker structure

Create worker/worker.go. The design goal: a worker with Concurrency = 10 should hold at most 10 messages, no matter how many streams it reads.

// worker/worker.go
package worker

const (
    // taskTimeout bounds a single task execution.
    taskTimeout = 90 * time.Second
    // defaultConcurrency is used when WorkerConfig.Concurrency is unset.
    defaultConcurrency = 1
    // defaultMaxDeliveries is used when WorkerConfig.MaxDeliveries is unset.
    defaultMaxDeliveries = 5
)

// errLeaseLost cancels an execution whose lease was taken over or released.
var errLeaseLost = errors.New("task lease lost")

// Worker executes tasks from its streams and reports results to the coordinator.
//
// One reader goroutine per stream pulls messages into a bounded hand-off
// channel, and Concurrency executor goroutines drain it. Readers only fetch as
// many messages as the channel has room for, so a reader blocked on an idle
// stream holds no execution capacity, and messages never pile up in memory.
type Worker struct {
    cfg           WorkerConfig
    router        *Router
    coordClient   *CoordinatorClient
    langClientMgr *LangClientManager

    deliveries chan delivery
    space      chan struct{} // signalled when an executor takes a delivery
}

// delivery is a message together with the stream it must be acked on.
type delivery struct {
    msg  stream.Message
    from stream.Consumer
}

// TaskMessage is the fields we expect in each stream message.
type TaskMessage struct {
    TaskID          string          `json:"task_id"`
    RunID           string          `json:"run_id"`
    NodeID          string          `json:"node_id"`
    NodeType        string          `json:"node_type"` // e.g. "http", "python", "js"
    WorkflowID      string          `json:"workflow_id"`
    WorkflowVersion int             `json:"workflow_version"`
    Attempt         int             `json:"attempt"`
    Input           json.RawMessage `json:"input"`
}

deliveries has capacity Concurrency, so the channel is the bound:

#Step 5: Run the loops

// Run reads and executes tasks until ctx is canceled, then waits for
// in-flight tasks to finish. Messages still in the hand-off channel stay
// pending in their stream and are reclaimed by another worker.
func (w *Worker) Run(ctx context.Context) {
    w.cfg.Logger.Info().
        Str("consumer", w.cfg.ConsumerName).
        Int("streams", len(w.cfg.Streams)).
        Int("concurrency", w.cfg.Concurrency).
        Msg("starting worker")

    var readers, executors sync.WaitGroup
    for _, s := range w.cfg.Streams {
        readers.Add(2)
        go func() { defer readers.Done(); w.readLoop(ctx, s) }()
        go func() { defer readers.Done(); w.reclaimPendingLoop(ctx, s) }()
    }
    for range w.cfg.Concurrency {
        executors.Add(1)
        go func() { defer executors.Done(); w.executeLoop(ctx) }()
    }

    readers.Wait()
    w.cfg.Logger.Info().Msg("context canceled, waiting for in-flight tasks")
    executors.Wait()
}

Each reader asks Redis only for as many messages as there are free slots:

// readLoop fetches new messages from one stream while the hand-off channel has room.
func (w *Worker) readLoop(ctx context.Context, s stream.Consumer) {
    for ctx.Err() == nil {
        free := cap(w.deliveries) - len(w.deliveries)
        if free <= 0 {
            w.waitForSpace(ctx)
            continue
        }
        msgs, err := s.ReadNew(ctx, int64(free))
        if err != nil {
            if ctx.Err() == nil {
                w.cfg.Logger.Error().Err(err).Msg("read new messages failed")
                time.Sleep(time.Second)
            }
            continue
        }
        for _, msg := range msgs {
            if !w.hand(ctx, delivery{msg: msg, from: s}) {
                return
            }
        }
    }
}

// waitForSpace blocks until an executor frees a slot (or briefly, since
// several readers share one signal).
func (w *Worker) waitForSpace(ctx context.Context) {
    select {
    case <-w.space:
    case <-time.After(100 * time.Millisecond):
    case <-ctx.Done():
    }
}

// hand passes a delivery to the executors; false if ctx was canceled first.
func (w *Worker) hand(ctx context.Context, d delivery) bool {
    select {
    case w.deliveries <- d:
        return true
    case <-ctx.Done():
        return false
    }
}

func (w *Worker) executeLoop(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        case d := <-w.deliveries:
            select {
            case w.space <- struct{}{}:
            default:
            }
            w.handleMessage(ctx, d)
        }
    }
}

Why does this matter? If readers fetched freely, one busy worker could pull hundreds of messages into memory while other workers sat idle — and if it crashed, all of those would wait for reclaim. Fetching only what you can run keeps work spread evenly across the fleet.

#Step 6: Handle one message

// handleMessage processes one message and acks it once the outcome is durable
// at the coordinator (or the message turned out to be stale or invalid).
func (w *Worker) handleMessage(ctx context.Context, d delivery) {
    taskMsg, err := parseTaskMessage(d.msg.Values)
    if err != nil {
        w.cfg.Logger.Error().Err(err).Msg("invalid task message; acking and discarding")
        _ = d.from.Ack(ctx, d.msg.ID)
        return
    }

    if err := w.processMessage(ctx, &taskMsg); err != nil {
        w.cfg.Logger.Error().Err(err).Str("task_id", taskMsg.TaskID).Msg("task processing failed")
        return // not acked — redelivered via reclaim
    }
    _ = d.from.Ack(ctx, d.msg.ID)
}

The ack rule is the key to at-least-once: ack only after the coordinator has durably accepted the result. If reporting fails, the message stays in the PEL and gets reclaimed.

Now the core of the worker:

func (w *Worker) processMessage(ctx context.Context, msg *TaskMessage) error {
    logger := w.cfg.Logger.With().
        Str("task_id", msg.TaskID).
        Str("node_id", msg.NodeID).
        Str("run_id", msg.RunID).
        Int("attempt", msg.Attempt).
        Logger()

    acquired, err := w.coordClient.AcquireTaskLease(ctx, msg, w.cfg.ConsumerName, w.cfg.LeaseTTLSeconds)
    if err != nil {
        return fmt.Errorf("acquire task lease: %w", err)
    }
    if !acquired {
        // Running elsewhere under a live lease, already finished, or superseded.
        logger.Debug().Msg("task lease not acquired, dropping message")
        return nil
    }

    execCtx, cancel := context.WithTimeout(ctx, taskTimeout)
    defer cancel()
    leaseCtx, stopRenewal := context.WithCancelCause(execCtx)
    defer stopRenewal(nil)
    go w.renewLeaseLoop(leaseCtx, msg, stopRenewal, logger)

    // Embed the task ID so executors can use task_instances.id as an
    // idempotency key (e.g. Idempotency-Key header on HTTP requests).
    start := time.Now()
    output, execErr := w.router.Route(executors.WithTaskID(leaseCtx, msg.TaskID), msg.NodeType, msg.Input)
    stopRenewal(nil)

    if errors.Is(context.Cause(leaseCtx), errLeaseLost) {
        // Another worker owns the task now; its result is the one that counts.
        logger.Warn().Msg("lease lost during execution, discarding result")
        return nil
    }
    if ctx.Err() != nil {
        // Shutting down: don't report the interruption as a task failure. The
        // lease expires and the coordinator re-dispatches the task.
        return fmt.Errorf("worker stopped during execution: %w", ctx.Err())
    }

    var errStr *string
    if execErr != nil {
        s := execErr.Error()
        errStr = &s
    }
    logger.Info().Dur("duration", time.Since(start)).Bool("success", execErr == nil).Msg("task executed")

    if err := w.coordClient.ReportTaskResult(ctx, msg, output, errStr); err != nil {
        return fmt.Errorf("report result: %w", err)
    }
    return nil
}

Walk through the three exits after execution:

Situation	What the worker does	Why
Lease lost while running	Discard result, ack	Another worker owns the task; two results for one attempt would be confusing.
Worker shutting down	Return error, no ack, no report	An interruption is not a task failure. The lease expires and the task is re-dispatched.
Normal finish	Report success or failure, then ack	The coordinator decides retry vs fail.

context.WithCancelCause lets the renewal loop cancel the execution with a reason, so the worker can tell "lease lost" apart from "timed out".

#Step 7: Renew the lease

// renewLeaseLoop renews the task lease every TTL/3 until ctx is canceled. If
// the lease is lost it cancels the execution with errLeaseLost.
func (w *Worker) renewLeaseLoop(ctx context.Context, msg *TaskMessage, cancel context.CancelCauseFunc, logger zerolog.Logger) {
    renewInterval := max(time.Duration(w.cfg.LeaseTTLSeconds/3)*time.Second, 5*time.Second)
    ticker := time.NewTicker(renewInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            renewed, err := w.coordClient.RenewTaskLease(ctx, msg, w.cfg.ConsumerName, w.cfg.LeaseTTLSeconds)
            switch {
            case err != nil:
                // Transient; the lease survives until its TTL, so keep trying.
                logger.Warn().Err(err).Msg("failed to renew task lease")
            case !renewed:
                cancel(errLeaseLost)
                return
            }
        }
    }
}

Renewing at TTL/3 gives two retries before the lease can expire. One slow network round trip never costs you the task.

#Step 8: Reclaim orphans and dead-letter poison messages

When a worker dies, its delivered-but-unacked messages sit in its PEL. Every worker runs a reclaim loop per stream that takes over messages idle longer than the lease TTL:

// reclaimPendingLoop periodically reclaims messages of one stream that have
// been pending too long (e.g. from crashed workers).
func (w *Worker) reclaimPendingLoop(ctx context.Context, s stream.Consumer) {
    reclaimInterval := max(time.Duration(w.cfg.LeaseTTLSeconds/2)*time.Second, 10*time.Second)
    minIdleTime := time.Duration(w.cfg.LeaseTTLSeconds) * time.Second

    ticker := time.NewTicker(reclaimInterval)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            w.reclaimPendingMessages(ctx, s, minIdleTime)
        }
    }
}

// reclaimPendingMessages reclaims orphaned messages and enforces the DLQ limit.
// Messages whose DeliveryCount exceeds MaxDeliveries are dead-lettered instead
// of being reprocessed.
func (w *Worker) reclaimPendingMessages(ctx context.Context, s stream.Consumer, minIdleTime time.Duration) {
    msgs, err := s.ReclaimOrphaned(ctx, minIdleTime)
    if err != nil {
        w.cfg.Logger.Error().Err(err).Msg("reclaim orphaned messages failed")
        return
    }
    if len(msgs) > 0 {
        w.cfg.Logger.Info().Int("reclaimed", len(msgs)).Msg("reclaimed orphaned pending messages")
    }

    for _, msg := range msgs {
        if msg.DeliveryCount > w.cfg.MaxDeliveries {
            w.deadLetter(ctx, delivery{msg: msg, from: s})
            continue
        }
        if !w.hand(ctx, delivery{msg: msg, from: s}) {
            return
        }
    }
}

A message delivered more than MaxDeliveries times is probably crashing workers (out of memory on a huge input, say). Instead of letting it take down the fleet one worker at a time, report it as failed:

// deadLetter reports a message as permanently failed to the coordinator and
// removes it from the stream so it is never redelivered.
// If the report fails the message is not acked so the next reclaim cycle
// retries the DLQ report.
func (w *Worker) deadLetter(ctx context.Context, d delivery) {
    logger := w.cfg.Logger.With().Str("msg_id", d.msg.ID).Int64("delivery_count", d.msg.DeliveryCount).Logger()

    taskMsg, err := parseTaskMessage(d.msg.Values)
    if err != nil {
        logger.Error().Err(err).Msg("dead-letter: unparseable message payload, discarding")
        _ = d.from.Ack(ctx, d.msg.ID)
        return
    }

    logger.Error().Str("task_id", taskMsg.TaskID).Msg("dead-lettering task: exceeded max delivery attempts")
    errStr := fmt.Sprintf("task exceeded max delivery attempts (%d)", w.cfg.MaxDeliveries)
    if err := w.coordClient.ReportTaskResult(ctx, &taskMsg, nil, &errStr); err != nil {
        logger.Error().Err(err).Msg("failed to report dead-lettered task; will retry on next reclaim cycle")
        return
    }
    _ = d.from.Ack(ctx, d.msg.ID)
}

The failure goes through the normal state machine, so the node's retry policy and the run's failure handling apply as usual.

#Step 9: Route to an executor

Create worker/executors/executor.go with the executor contract and a way to pass the task ID down:

// worker/executors/executor.go

// Executor defines the interface for task executors
type Executor interface {
    // Execute runs a task with the given input and returns output or error
    Execute(ctx context.Context, input json.RawMessage) (json.RawMessage, error)

    // Type returns the executor type name (e.g., "http")
    Type() string
}

type contextKey int

const taskIDKey contextKey = iota

// WithTaskID returns a context carrying the task instance ID.
// Executors read this to set Idempotency-Key headers or pass the ID to
// language servers, so all execution is traceable back to the originating
// task_instances row.
func WithTaskID(ctx context.Context, taskID string) context.Context {
    return context.WithValue(ctx, taskIDKey, taskID)
}

// TaskIDFromContext returns the task instance ID stored in ctx, or "" if none.
func TaskIDFromContext(ctx context.Context) string {
    v, _ := ctx.Value(taskIDKey).(string)
    return v
}

The router in worker/router.go sends language runtimes to a language server and everything else to a built-in executor:

// Route routes a task to the appropriate executor based on node type
// nodeType can be:
// - For CODE nodes: runtime (python, js, node) - routes to language executor
// - For other nodes: type (HTTP) - routes to built-in executor
// - Control flow nodes (SWITCH, PARALLEL/FORK, JOIN) are handled by coordinator, not worker
func (r *Router) Route(ctx context.Context, nodeType string, input json.RawMessage) (json.RawMessage, error) {
    logger := r.logger.With().Str("node_type", nodeType).Logger()

    // Check if it's a language runtime (python, js, node, etc.)
    // These come from CODE type nodes where runtime is specified
    if isLanguageRuntime(nodeType) {
        if r.langClientMgr == nil {
            return nil, fmt.Errorf("language client manager not configured")
        }
        langClient, err := r.langClientMgr.GetClientForRuntime(nodeType)
        if err != nil {
            return nil, fmt.Errorf("language client not configured for runtime: %s: %w", nodeType, err)
        }
        return r.executeLanguageTaskWithClient(ctx, nodeType, input, langClient, logger)
    }

    // Check for built-in executor (HTTP, etc.)
    exec, exists := r.executors[nodeType]
    if !exists {
        return nil, fmt.Errorf("no executor found for node type: %s", nodeType)
    }

    logger.Debug().Msg("routing to built-in executor")
    return exec.Execute(ctx, input)
}

#Step 10: The HTTP executor

Create worker/executors/http_executor.go. One shared, pooled http.Client reuses TCP and TLS connections across thousands of tasks:

// NewHTTPExecutor creates a new HTTP executor with a shared connection pool.
func NewHTTPExecutor() *HTTPExecutor {
    transport := &http.Transport{
        DialContext: (&net.Dialer{
            Timeout:   30 * time.Second,
            KeepAlive: 30 * time.Second,
        }).DialContext,
        MaxIdleConns:          100,
        MaxIdleConnsPerHost:   20,
        IdleConnTimeout:       90 * time.Second,
        TLSHandshakeTimeout:   10 * time.Second,
        ExpectContinueTimeout: 1 * time.Second,
    }
    return &HTTPExecutor{
        // No global Timeout — each request carries its own context deadline.
        client: &http.Client{Transport: transport},
    }
}

The interesting part of Execute is the idempotency key:

    method := params.Method
    if method == "POST" || method == "PUT" || method == "PATCH" {
        if req.Header.Get("Content-Type") == "" && len(params.Body) > 0 {
            req.Header.Set("Content-Type", "application/json")
        }
        // Use task_instances.id as the idempotency key so services that honour
        // the Idempotency-Key header (Stripe, many REST APIs) deduplicate
        // retried requests automatically.  The header is only set when the task
        // ID is present in the context and the caller has not already set one.
        if taskID := TaskIDFromContext(ctx); taskID != "" && req.Header.Get("Idempotency-Key") == "" {
            req.Header.Set("Idempotency-Key", taskID)
        }
    }

    resp, err := e.client.Do(req)
    if err != nil {
        return nil, fmt.Errorf("HTTP request failed: %w", err)
    }
    defer resp.Body.Close()

    out, err := io.ReadAll(resp.Body)
    if err != nil {
        return nil, fmt.Errorf("failed to read response body: %w", err)
    }

    if resp.StatusCode >= 400 {
        return nil, fmt.Errorf("HTTP error %d: %s", resp.StatusCode, string(out))
    }

    return normalizeResponseBody(out)

The task ID stays the same across every attempt and every redelivery of a task. So when at-least-once execution sends the same POST /charges twice, a payment API that honors Idempotency-Key charges the card only once.

Responses are normalized so the next node's templates always get JSON:

// normalizeResponseBody ensures the output is always valid JSON.
// Valid JSON is returned as-is. Non-JSON and empty bodies are wrapped.
func normalizeResponseBody(body []byte) (json.RawMessage, error) {
    if len(body) == 0 {
        return json.RawMessage("{}"), nil
    }
    if json.Valid(body) {
        return json.RawMessage(body), nil
    }
    // Wrap plain text / HTML / binary in a JSON object so downstream
    // CEL expressions can access it via .body
    wrapped, err := json.Marshal(map[string]string{"body": string(body)})
    if err != nil {
        return nil, fmt.Errorf("failed to encode non-JSON response body: %w", err)
    }
    return json.RawMessage(wrapped), nil
}

#Step 11: Code steps run in language servers

CODE nodes run user-written Python or JavaScript. Running untrusted code inside the Go worker would be dangerous, so the worker forwards it over gRPC to a separate language server:

// worker/langclient.go

// ExecuteInlineCode executes code on remote language worker with a timeout.
func (c *LangClient) ExecuteInlineCode(ctx context.Context, taskID, runtime, code, filename, inputJSON string, timeoutSec int) (*pb.ExecuteResponse, error) {
    if c == nil {
        return nil, ErrNoLangClient
    }
    req := &pb.ExecuteRequest{
        TaskId:         taskID,
        Runtime:        runtime,
        InputJson:      inputJSON,
        TimeoutSeconds: int32(timeoutSec),
        Metadata:       map[string]string{"from": "worker"},
        Payload: &pb.ExecuteRequest_InlineCode{
            InlineCode: &pb.InlineCode{Code: code, Filename: filename},
        },
    }
    ctx2, cancel := context.WithTimeout(ctx, time.Duration(timeoutSec+5)*time.Second)
    defer cancel()
    return c.client.Execute(ctx2, req)
}

The language servers stack several sandbox layers:

Layer	Python	JavaScript
Language	RestrictedPython: no _-prefixed attributes, no eval/exec; import allowlist	V8 isolate: no require, process, timers or Node.js APIs
Interpreter	PEP 578 audit hook: no file writes, no network, no subprocesses, no ctypes	—
Process	per-execution child process with memory, CPU, file and process limits; empty environment	isolate memory limit, timeout, capped logs and output
Container	non-root, read-only filesystem, no capabilities, no-new-privileges, PID limit	same

Separating them also lets you scale code execution independently of HTTP-heavy workers.

#Step 12: Wire the worker binary

In cmd/worker/main.go, build one consumer per shard of every queue the worker serves, then start:

// cmd/worker/main.go

    // One consumer per shard stream of every queue this worker serves.
    shards := streamShards(ctx, redisStream, logger)
    streamNames := stream.QueueStreams(queues, shards)
    consumers := make([]stream.Consumer, len(streamNames))
    for i, name := range streamNames {
        consumers[i] = redisstream.NewConsumer(redisstream.ConsumerConfig{
            StreamName:    name,
            ConsumerGroup: consumerGroup,
            ConsumerName:  workerID,
        }, redisStream, logger)
    }

    // Create worker
    wcfg := worker.WorkerConfig{
        Streams:         consumers,
        CoordinatorAddr: coordAddr,
        LangWorkerAddrs: langWorkerAddrs,
        ConsumerName:    workerID,
        LeaseTTLSeconds: leaseTTLSeconds,
        Concurrency:     int(concurrency),
        Logger:          logger,
    }

Which queues? WORKER_QUEUES if set, otherwise http plus one per configured language runtime:

// workerQueues returns the task queues to consume: WORKER_QUEUES if set,
// otherwise "http" plus one queue per configured language runtime.
func workerQueues(langWorkerAddrs map[string]string) []string {
    if v := os.Getenv("WORKER_QUEUES"); v != "" {
        var queues []string
        for _, q := range strings.Split(v, ",") {
            if q = strings.ToLower(strings.TrimSpace(q)); q != "" {
                queues = append(queues, q)
            }
        }
        return queues
    }
    queues := []string{"http"}
    for runtime := range langWorkerAddrs {
        queues = append(queues, runtime)
    }
    return queues
}

Variable	Default	Meaning
COORDINATOR_ADDR	localhost:50051	coordinator gRPC address
REDIS_ADDR	127.0.0.1:6379	Redis address
LANG_WORKER_ADDRS	—	e.g. python:host:50053,js:host:50052
WORKER_QUEUES	http + each runtime	queues to consume
WORKER_CONCURRENCY	10	tasks executed in parallel
LEASE_TTL_SECONDS	90	lease TTL, renewed every TTL/3
WORKER_ID	hostname + random suffix	consumer name and lease owner; must be unique

WORKER_ID must be unique: it is both the Redis consumer name and the lease owner. Two workers with the same ID would each think they own the other's lease.

The full loop now works: decide → outbox → stream → lease → execute → report → decide. The last piece is making sure nothing gets stuck when machines fail.