Data flows

Every run is the same shape: extract reads the source into blob, load replays blob into the destination. What changes between runs is where each stage executes. A connection placed agent pins its stage to that agent (its secrets live at the edge and never reach the cloud); anything else runs in the cloud worker. Two endpoints × two placements = four flows. Toggle between them:

The rest of this page is a first-principles walkthrough of the real code path: how a POST …/run becomes a placement decision, how that decision picks an executor, how the worker's Work loop drives a run to a terminal status, when the blob is skipped entirely, and how cancel and failure propagate across the cloud/edge boundary. Everything below cites the source that implements it — no invented APIs.

Placement is a pure function

On POST /pipelines/{id}/run, runPipeline creates a run row and calls orchestrateRun, which loads the source and destination connections and derives placement from their placement columns. It is a pure function — no I/O, no clock, no randomness — so a manual run and a scheduled run route identically. The same two calls (extract side, load side) collapse to one of four outcomes:

// A connection placed "agent" pins its stage to that agent; anything else → cloud.
func derivePlacement(srcPlacement string, srcAgent uuid.UUID, dstPlacement string, dstAgent uuid.UUID) runPlacement {
	return runPlacement{
		Extract: stageFor(srcPlacement, srcAgent), // reads the source
		Load:    stageFor(dstPlacement, dstAgent), // writes the destination
	}
}

func stageFor(placement string, agent uuid.UUID) stagePlacement {
	if placement == "agent" {
		return stagePlacement{Cloud: false, AgentID: agent}
	}
	return stagePlacement{Cloud: true}
}

Two tests pin the top and bottom rows of that table directly against runPipeline: TestRunPipelineCloudEnqueuesMigrate asserts a fully-cloud run enqueues exactly one migrate job and creates no agent job, while TestRunPipelineAgentSourceDispatchesExtract asserts an agent-source run creates exactly one extract job (carrying the run + source connection, payload = pipeline id) and enqueues nothing on the worker.

From run row to executor

runPipeline is deliberately thin: validate, create the pending run, orchestrate, then re-read the run before responding. That re-read matters — orchestrateRun may have already terminalized the run (e.g. a bound runtime is offline), so returning the stale pending row from CreateRun would lie to the caller.

run, _ := s.q.CreateRun(r.Context(), sqlc.CreateRunParams{PipelineID: id, DryRun: dryRun})
if err := s.orchestrateRun(r.Context(), ws, pipe, run); err != nil {
	s.fail(w, http.StatusInternalServerError, err)
	return
}
// orchestrateRun may have already terminated the run (e.g. a bound runtime is
// offline). Re-read so the response reflects the real status, not the stale
// pending row from CreateRun.
if fresh, err := s.q.GetRun(r.Context(), sqlc.GetRunParams{ID: run.ID, WorkspaceID: ws}); err == nil {
	run = fresh
}
writeJSON(w, http.StatusAccepted, toRunDTO(run))

orchestrateRun resolves placement, then does a reachability pre-flight on whichever stage runs first on an agent (extract if the source is agent-placed, else load). A job dispatched to an offline runtime would sit un-leased forever — nothing polls for it — so a manual run fails fast with a fixable message instead of hanging. Only after that guard does it branch to a dispatch:

placement := derivePlacement(src.Placement, uuid.UUID(src.AgentID.Bytes), dst.Placement, uuid.UUID(dst.AgentID.Bytes))

// Fail fast if the first edge stage's runtime is offline (else the job hangs un-leased).
if !placement.Extract.Cloud {
	if reachable, name := s.agentReachable(ctx, ws, placement.Extract.AgentID); !reachable {
		return s.failRunUnreachableAgent(ctx, run.ID, name) // marks run failed; returns nil
	}
} else if !placement.Load.Cloud {
	if reachable, name := s.agentReachable(ctx, ws, placement.Load.AgentID); !reachable {
		return s.failRunUnreachableAgent(ctx, run.ID, name)
	}
}

if !placement.Extract.Cloud {
	payload, _ := json.Marshal(map[string]string{"pipelineId": pipe.ID.String()})
	s.q.CreateAgentJob(ctx, sqlc.CreateAgentJobParams{
		AgentID: placement.Extract.AgentID, WorkspaceID: ws, Kind: "extract",
		RunID:        pgtype.UUID{Bytes: run.ID, Valid: true},
		ConnectionID: pgtype.UUID{Bytes: pipe.SourceConnectionID, Valid: true},
		Payload:      payload, // IDs only, never secrets
	})
	s.runEvent(ctx, run.ID, "info", "", "Dispatched extract to the source runtime")
	return nil // NOT enqueued to the worker — the agent drives the extract
}
return s.enq.EnqueueMigrate(ctx, run.ID) // extract runs in the cloud

Reachability is a recency check, not a live ping: agentReachable returns true only when the agent is active and checked in within agentOnlineWindow (90 seconds). A healthy runtime long-polls every few seconds, so the window is generous but still catches a runtime that fell over.

cloud → cloud

Both endpoints are reachable from Setu, so orchestrateRun takes the simple branch: EnqueueMigrate. The River worker does the whole run in one Work call — extract to blob, flip the run's phase to load, load from blob, mark it succeeded.

if run.Phase == "extract" {
	if _, err := w.runExtract(runCtx, q, runID, src, streams, res, workspaceID, run.PipelineID, blb); err != nil {
		return runner.Stats{}, err
	}
	q.SetRunPhase(ctx, sqlc.SetRunPhaseParams{ID: runID, Phase: "load"})
	w.event(ctx, q, runID, "info", "", "extract complete; staged to blob")
	// (loadOnAgent is false here, so we fall through and load in-process)
}
// Load phase replays the staged blob through the operator chain into the destination.
stats, err := w.runLoad(runCtx, ctx, q, runID, dst, res, run.DryRun, run.RowsTotal, blb, workspaceID, run.PipelineID, cancel)

The phase column is what makes this resumable. It is persisted the instant extract finishes, so a crash-and-retry re-enters execute at phase == "load" and never re-reads the source. That same guard is what lets an agent-staged run load in the cloud (see agent → cloud).

The worker's run loop

Work is the state machine that turns a queued job into a terminal run status. Four details carry the correctness weight:

1. 1
   The deadline is disabled. Timeout returns -1. River defaults to a 60-second per-job deadline, which would cancel a multi-million-row load mid-write and retry it forever. A migration instead runs to completion or is stopped cooperatively via the cancel flag.
2. 2
   MarkRunRunning is conditional. It only matches a run still eligible to start. If the run was cancelled while pending, it no longer matches, returns pgx.ErrNoRows, and the worker skips the job rather than resurrecting a dead run.
3. 3
   The terminal write is detached from cancellation. persistCtx = context.WithoutCancel(ctx) keeps the status write alive even when the job's own context was cancelled (River shutdown / timeout) — otherwise a cancelled job leaves the run orphaned in running forever.
4. 4
   Failure resumes, it doesn't restart. While attempts remain the error is returned so River retries; the run stays running at its persisted phase, so the retry resumes from where it stopped. Only the last attempt writes MarkRunFailed.

persistCtx := context.WithoutCancel(ctx) // terminal write survives ctx cancellation

if _, err := q.MarkRunRunning(ctx, runID); err != nil {
	if errors.Is(err, pgx.ErrNoRows) {
		w.log.Info("run already cancelled, skipping", "runId", runID) // cancelled while pending
		return nil
	}
	return fmt.Errorf("mark running: %w", err)
}

runCtx, cancel := context.WithCancel(ctx) // cancelable copy context
defer cancel()
stats, err := w.execute(runCtx, ctx, q, runID, cancel)

if errors.Is(err, errHandedOffToAgent) {         // cloud staged; edge will load → not done
	return nil                                    // run stays 'running' until the agent posts its result
}
if err != nil {
	if errors.Is(err, context.Canceled) || w.cancelRequested(persistCtx, q, runID) {
		q.MarkRunCancelled(persistCtx, runID)     // a cancelled copy is cancelled, not failed
		return nil
	}
	if job.Attempt < job.MaxAttempts {
		return err                                // River retries; execute resumes from the persisted phase
	}
	q.MarkRunFailed(persistCtx, sqlc.MarkRunFailedParams{ID: runID, Error: &msg})
	return nil
}
q.MarkRunSucceeded(persistCtx, /* reconciliation counts */)

Staged vs direct execution

The blob is the default seam, but for a small incremental delta the round-trip is pure overhead. execute chooses direct — streaming source → operators → destination in one pass with no blob (phase=direct) — under two conditions, and only when the run is still extracting in the cloud:

// Only a fresh (still-extracting) cloud-loading run can go direct. A run resumed at
// the load phase already committed to staged, and an agent load has no live source here.
if run.Phase == "extract" && !loadOnAgent {
	forceDirect := !stagedEnabled()  // kill switch: SETU_STAGED_EXECUTION=off
	autoDirect := false
	if res.Incremental {
		plans, todo, total, sized, _ := w.directDelta(ctx, src, res, streams, run.PipelineID, ecs)
		directPlans, directStreams, directTotal = plans, todo, total
		autoDirect = sized && total <= directMaxRows()  // small delta → skip the blob
	}
	if forceDirect || autoDirect {
		return w.runDirect(runCtx, ctx, q, runID, src, dst, directStreams, res, run.DryRun,
			directPlans, directTotal, ecs, run.PipelineID, cancel)
	}
}
// …otherwise STAGE to blob (extract → phase=load → load).

Direct trades resume for speed: it sets phase=direct, writes no run_artifact, and a retry re-reads the whole delta (safe because the load upsert is idempotent). That is precisely why the auto-decision only picks it for small deltas — a large direct run that crashed would have to re-read everything. The staged path, by contrast, checkpoints per blob part on the load side and per stream on the extract side, so a retry resumes mid-flight.

On the staged load side, whether checkpointing is even valid depends on the operator chain. Per-part load checkpoints assume a streaming chain (identity or row-local ops) where one batch == one durably-written part. A blocking operator (an operator.Flusher, e.g. aggregate) buffers across the whole stream and emits from Finish(), firing extra drain batches — so the worker detects a blocking chain and disables checkpointing for it, resuming from 0 (safe: upsert is idempotent).

agent → cloud (the headline seam)

The source is behind a firewall; the destination is a cloud warehouse. The cloud can't reach the source, so extract has to run at the edge — but the agent opens no inbound ports, so the cloud can't push it work either. The resolution is the whole point of the hybrid model: orchestrateRun doesn't enqueue a worker job — it creates an extract agent_job and returns. The agent picks it up on its own outbound long-poll, stages to the shared blob, and posts a result; the apiserver then enqueues the cloud load. See long-polling & leasing for how the agent gets that job.

The agent never receives a credential. The extract-spec it fetches carries the connection config verbatim with ${secret.X} references intact — the agent resolves them from its own edge secret store — and the same blob backend the cloud reads from (SpecFromEnv), so the staged parts land where the cloud load will look:

type extractSpecDTO struct {
	RunID      uuid.UUID       `json:"runId"`
	BlobPrefix string          `json:"blobPrefix"` // ws/{ws}/pl/{pipe}/run/{run}/
	Provider   string          `json:"provider"`
	Config     json.RawMessage `json:"config"`     // secret refs intact — NEVER resolved
	Streams    []string        `json:"streams"`    // selected datasets, or [] = discover all
	Blob       engineblob.Spec `json:"blob"`       // SAME shared backend the cloud reads back from
}

// Config is returned VERBATIM (secret refs intact). Never resolve here.
cfg := json.RawMessage(conn.Config)
writeJSON(w, http.StatusOK, extractSpecDTO{
	RunID: runID, BlobPrefix: runBlobPrefix(ws, run.PipelineID, runID),
	Provider: conn.Provider, Config: cfg, Streams: streams, Blob: engineblob.SpecFromEnv(),
})

When the agent posts success, onAgentExtractResult records the staged artifact, advances the run to load, sets the run total from the staged row count (so the monitor has a determinate progress denominator the cloud never computed), and dispatches the load. Here the destination is cloud, so it enqueues the worker:

s.q.CreateRunArtifact(ctx, sqlc.CreateRunArtifactParams{RunID: runID, BlobPrefix: res.BlobPrefix, RowsTotal: res.Rows, /* … */})
s.q.SetRunPhase(ctx, sqlc.SetRunPhaseParams{ID: runID, Phase: "load"})
s.q.SetRunTotal(ctx, sqlc.SetRunTotalParams{ID: runID, RowsTotal: res.Rows})

// Where does the load run? Resolve the run's destination placement.
load, dstID, ok := s.runLoadPlacement(ctx, job.WorkspaceID, runID)
if load.Cloud {
	s.enq.EnqueueMigrate(ctx, runID) // the worker loads from the agent-staged blob
	s.runEvent(ctx, runID, "info", "", "Loading staged data on Setu cloud")
}

The worker resumes at phase == "load": it never builds the source (its edge secrets aren't in the cloud) and loads purely from blob. That guard is what makes an agent-extracted run loadable in the cloud:

// A run already past extract (phase "load") — staged by an agent — loads purely
// from blob and must not touch the source: its edge secrets never reach the cloud.
var src connector.Source
if run.Phase == "extract" {
	src, err = w.reg.BuildSource(ctx, srcConn.Provider, srcCfg)
	// …
}

This flow crosses three endpoints on the agent's side, all outbound. The agent leases with a bounded long-poll (wait capped at 30s), holds the job under a 60-second lease it renews via /heartbeat, and posts a terminal result. A reaper requeues any lease that expires without a result, so a crashed agent's job is retried rather than lost.

cloud → agent

Mirror image: the source is a cloud warehouse, the destination is private. The cloud extracts to blob, but it can't reach the destination — so after extract it hands the load off to the destination's bound agent. The worker signals this with a sentinel error, errHandedOffToAgent, which Work treats as a clean, non-retrying completion of the cloud's part; the run stays running at phase load until the agent posts its result.

// dstConn.Placement == "agent": don't build the sink here — the load runs at the edge.
loadOnAgent := dstConn.Placement == "agent"
// … after staging the extract:
if loadOnAgent {
	q.CreateAgentJob(ctx, sqlc.CreateAgentJobParams{AgentID: uuid.UUID(dstConn.AgentID.Bytes), Kind: "load", RunID: /* … */})
	return runner.Stats{}, errHandedOffToAgent
}

// In Work, the sentinel is not a failure:
if errors.Is(err, errHandedOffToAgent) {
	w.event(persistCtx, q, runID, "info", "", "staged to blob; handed off to runtime for load")
	return nil // run stays 'running' until onAgentLoadResult marks it succeeded
}

The agent's load-spec is richer than the extract one: besides the destination provider + config (secret refs intact) and the blob prefix, it carries the pipeline Document and the write mode, so the edge compiles the same interior operator chain the cloud would have applied at load. The transform semantics are identical whether the load runs in the cloud or at the edge.

type loadSpecDTO struct {
	RunID      uuid.UUID       `json:"runId"`
	BlobPrefix string          `json:"blobPrefix"`
	Provider   string          `json:"provider"`
	Config     json.RawMessage `json:"config"` // secret refs intact
	Mode       string          `json:"mode"`   // append / upsert / overwrite / merge
	Spec       json.RawMessage `json:"spec"`   // pipeline Document → same operator chain the cloud compiles
	Blob       engineblob.Spec `json:"blob"`   // same shared backend the cloud staged into
}

Completion runs through onAgentLoadResult, which simply marks the run succeeded with the reported row counts — the terminal transition for both agent load flows.

agent → agent

Both endpoints are private — possibly on different agents. Extract runs on the source's agent; the result triggers onAgentExtractResult, whose runLoadPlacement now resolves to a non-cloud load, so it dispatches a load job to the destination's agent. The cloud never touches the data — the shared blob is the durable relay between the two edges.

load, dstID, ok := s.runLoadPlacement(ctx, job.WorkspaceID, runID)
if !load.Cloud {
	// Dest is agent-placed: hand off to that agent's load job.
	s.q.CreateAgentJob(ctx, sqlc.CreateAgentJobParams{
		AgentID: load.AgentID, Kind: "load",
		RunID: pgtype.UUID{Bytes: runID, Valid: true},
		ConnectionID: pgtype.UUID{Bytes: dstID, Valid: true},
	})
	s.runEvent(ctx, runID, "info", "", "Loading staged data on the destination runtime")
	return
}

Discover & preview at the edge

Not everything is a full run. When you browse a source's datasets or preview rows, the cloud normally runs the op itself — but it can't reach an agent-placed connection. For those, the apiserver dispatches a connection-scoped agent job (test, discover, or preview) and blocks for the result, so the interactive call still returns a synchronous answer.

// Enqueue a connection-scoped agent job and BLOCK for its result, so on-demand
// source ops the cloud can't run against an edge connection still return synchronously.
func (s *Server) runConnJobSync(ctx context.Context, c sqlc.Connection, kind string, payload map[string]any) (json.RawMessage, error) {
	// … verify the agent is active, default payload provider/config (secret refs intact) …
	job, err := s.q.CreateAgentJob(ctx, sqlc.CreateAgentJobParams{AgentID: agentID, Kind: kind, ConnectionID: /* … */, Payload: pl})
	if err != nil { return nil, err }
	return s.awaitAgentJob(ctx, c.WorkspaceID, job.ID) // polls every 500ms until terminal
}

These jobs have no run — the caller polls the job directly. A successful discover is bridged by onAgentDiscoverResult, which upserts the enumerated streams into the catalog with the same reconcile-and-missing-sweep the cloud discover uses, applied to a connection the cloud cannot reach. Because it is connection-scoped, onAgentJobResult routes it separately: it never touches a run, and the runIsActive guard that protects extract/load results doesn't apply to it.

Cancel & failure propagation

Cancel is cooperative. cancelRun sets the run's cancel_requested flag (RequestRunCancel, which only matches a pending/running run — so a repeat is an idempotent no-op). A running cloud copy stops because the worker polls that flag once per batch and cancels the copy context:

// In the per-batch progress callback:
if w.cancelRequested(ctx, q, runID) { // reads IsRunCancelRequested
	w.event(ctx, q, runID, "warn", p.Stream, "cancellation requested; stopping run")
	cancel() // aborts the copy context → runner returns context.Canceled
}

// In Work, a cancelled copy is recorded as cancelled, not failed:
if errors.Is(err, context.Canceled) || w.cancelRequested(persistCtx, q, runID) {
	q.MarkRunCancelled(persistCtx, runID)
	return nil
}

When the work runs on an agent the worker isn't in the loop, so cancelRun abandons the run's in-flight agent jobs (CancelRunAgentJobs) and terminalizes the run itself — but only when nothing is actively copying: a pending run (not yet started) or an agent run whose jobs were just abandoned. A running cloud run is left to the worker's own cancel poll.

run, err := s.q.RequestRunCancel(ctx, sqlc.RequestRunCancelParams{ID: id, WorkspaceID: ws}) // sets cancel_requested
// … pgx.ErrNoRows → already finished/cancelled/not ours → idempotent {cancelling:false}
agentJobs, _ := s.q.CancelRunAgentJobs(ctx, pgtype.UUID{Bytes: id, Valid: true})
if run.Status == "pending" || agentJobs > 0 {
	s.q.MarkRunCancelled(ctx, id)
}

Failure propagates the other way. A failed extract/load agent job must fail its run, or the run hangs in its phase forever: onAgentJobFailed appends an error run_event and calls MarkRunFailed (guarded by runIsActive). Connection-scoped kinds (test/discover/preview) have no run, so they are no-ops here — their caller sees the failure via the blocking awaitAgentJob poll instead.

func (s *Server) onAgentJobFailed(ctx context.Context, job sqlc.AgentJob, jobErr *string) {
	if !job.RunID.Valid { return }
	switch job.Kind {
	case "extract", "load": // only run-scoped kinds touch a run
	default:
		return
	}
	if !s.runIsActive(ctx, uuid.UUID(job.RunID.Bytes)) { return } // don't overwrite a cancelled/finished run
	prefixed := fmt.Sprintf("runtime %s: %s", job.Kind, msg)
	s.runEvent(ctx, runID, "error", "", prefixed)
	s.q.MarkRunFailed(ctx, sqlc.MarkRunFailedParams{ID: runID, Error: &prefixed})
}

Because agent-executed runs bypass the worker's own event stream, the apiserver emits runEvent rows at every dispatch/result transition — “Dispatched extract to the source runtime”, “Loading staged data on Setu cloud”, and so on. Without those the run monitor would show a blank timeline for the entire edge portion of the flow. The run's executedOn field ({kind:"agent", name} vs {kind:"cloud"}) is derived from whether an extract agent_job exists — the two tests TestGetRunExecutedOnAgent and TestGetRunExecutedOnCloud pin that surface.