Blob staging

A migration is naturally two halves: read from the source (extract) and write to the destination (load). Keeping them a single in-memory pass is simplest, and Setu does exactly that for small direct runs. But the moment you want the two halves to happen in different processes — an agent extracting inside your network, the cloud worker loading into a warehouse — you need a durable place to hand records over. That place is blob storage, and it is deliberately the thinnest abstraction in the codebase.

The Blob interface

A blob store is a keyed object store with slash-separated paths and four methods. That's the entire contract — small enough that a new backend is an isolated adapter, and general enough that it maps onto every object store that matters.

// Blob is a keyed object store. Keys are "/"-separated paths.
type Blob interface {
	// Put stores an object at key, overwriting any existing object. size may
	// be -1 when unknown (the backend buffers as needed).
	Put(ctx context.Context, key string, r io.Reader, size int64) error
	// Get opens an object for reading; returns ErrNotFound if absent.
	Get(ctx context.Context, key string) (io.ReadCloser, error)
	// List returns objects whose key starts with prefix.
	List(ctx context.Context, prefix string) ([]ObjectInfo, error)
	// Delete removes an object. Deleting a missing key is not an error.
	Delete(ctx context.Context, key string) error
}

Two contract choices make this interface safe to build on. Get returns a sentinel ErrNotFound rather than a backend-specific error, so callers branch on absence portably (the sink's resume logic depends on exactly this). And Delete of a missing key is a no-op, not an error, so the garbage collector is idempotent — re-running retention over an already-cleaned prefix is harmless.

Two backends

There are two implementations behind that interface:

• fs — a filesystem store rooted at a directory, for dev, tests, and single-host deployments. Keys map to paths under the root, with a safePath guard that rejects any key trying to escape the root via .. or absolute components.
• s3 — an S3-compatible store built on aws-sdk-go-v2. One implementation serves AWS S3, MinIO, Cloudflare R2, Backblaze B2, and Supabase Storage. The SDK's BaseEndpoint handles endpoints that carry a path prefix (which older clients like minio-go can't), and ForcePathStyle covers MinIO and Supabase.

The fs backend's safePath is the security-critical line: a run's stream slug is derived from a stream name, which ultimately comes from user-controlled discovery, so a hostile or malformed key must never write outside the root. It cleans the joined path and confirms it is still under the root before any I/O.

// safePath resolves key to an absolute path under root, rejecting any key that
// would escape root via ".." or absolute components.
func (b *FS) safePath(key string) (string, error) {
	root := filepath.Clean(b.root)
	p := filepath.Clean(filepath.Join(root, filepath.FromSlash(key)))
	if p != root && !strings.HasPrefix(p, root+string(filepath.Separator)) {
		return "", fmt.Errorf("blob: key %q escapes root", key)
	}
	return p, nil
}

fs.List walks the whole tree under the root and filters by key prefix, which is O(files) but fine for the single-host scale it targets. The s3 backend has two details worth calling out. Put buffers the entire body into a bytes.Reader before uploading — SigV4 needs a seekable body and a known content length, and because staged parts are bounded (see the part cap below) this is cheap and avoids the deprecated transfer manager. List uses the ListObjectsV2 paginator so it survives buckets with more than 1,000 objects. And Get's not-found detection is deliberately broad: it matches the typed NoSuchKey/NotFound and the generic 404 that some S3-compatible stores return, so ErrNotFound is reliable across providers.

// S3Config configures an S3-compatible backend. Endpoint is the FULL base URL
// including scheme and any path prefix (e.g.
// "https://<proj>.supabase.co/storage/v1/s3" for Supabase,
// "http://minio:9000" for MinIO); empty defaults to AWS S3.
// ForcePathStyle is required for MinIO and Supabase Storage.
type S3Config struct {
	Endpoint       string
	Region         string
	Bucket         string
	AccessKey      string
	SecretKey      string
	ForcePathStyle bool
}

Config: how both sides stage to the same place

A backend is selected from SETU_BLOB_* environment variables via FromEnv. That works fine for the cloud worker, which shares the control plane's environment. The agent runs elsewhere — so the control plane serializes the same configuration into a Spec and ships it inside the job, and the agent rebuilds the identical backend with FromSpec. This is the crux: both processes end up writing to and reading from the same bucket. Spec mirrors the env vars exactly, so SpecFromEnv and FromEnv produce byte-identical backends — the only difference is where the config came from.

// FromEnv builds the backend from SETU_BLOB_* env vars (worker side).
func FromEnv() (Blob, error)

// Spec is a serializable blob configuration the control plane hands to an
// agent so both stage to the SAME backend (the cross-process seam needs it).
// NOTE: for s3 this carries access/secret keys, so only include it in
// responses to trusted, authenticated agents.
type Spec struct {
	Backend        string `json:"backend,omitempty"`
	FSRoot         string `json:"fsRoot,omitempty"`
	Endpoint       string `json:"endpoint,omitempty"`
	Region         string `json:"region,omitempty"`
	Bucket         string `json:"bucket,omitempty"`
	AccessKey      string `json:"accessKey,omitempty"`
	SecretKey      string `json:"secretKey,omitempty"`
	ForcePathStyle bool   `json:"forcePathStyle,omitempty"`
}

func SpecFromEnv() Spec              // worker: serialize its config
func (s Spec) Empty() bool           // agent: fall back to its own FromEnv
func FromSpec(s Spec) (Blob, error)  // agent: rebuild the identical backend

The staged layout

A run's data lives under a deterministic key prefix so both sides — and the garbage collector — can address it without shared state. The prefix is ws/<workspace>/pl/<pipeline>/run/<run>/, and within it each stream gets a folder holding a manifest and its gzipped JSONL parts.

The stream folder is keyed by a filesystem-safe slug of the qualified stream name — Slug replaces any run of characters outside [a-zA-Z0-9_-] with _, so public.orders becomes public_orders. Because the prefix is fully determined by workspace / pipeline / run / slug, either process can compute any key without consulting the other. Part keys are zero-padded (%06d) so a lexical listing is also a numeric ordering.

func streamPrefix(prefix, slug string) string { return prefix + "stream/" + slug + "/" }
func manifestKey(prefix, slug string) string  { return streamPrefix(prefix, slug) + "_manifest.json" }
func partKey(prefix, slug string, n int) string {
	return fmt.Sprintf("%spart-%06d.jsonl.gz", streamPrefix(prefix, slug), n)
}

// Manifest is the per-stream contract Load consumes. The schema is stored
// once here so chunk rows stay compact.
type Manifest struct {
	Stream      string         `json:"stream"`      // qualified name
	Schema      *record.Schema `json:"schema"`
	PrimaryKey  []string       `json:"primaryKey,omitempty"`  // upsert/merge target
	Namespace   string         `json:"namespace,omitempty"`
	StreamType  string         `json:"streamType,omitempty"`
	Status      string         `json:"status"`      // "extracting" | "complete"
	Parts       []Part         `json:"parts"`
	RowsTotal   int64          `json:"rowsTotal"`
	CursorField string         `json:"cursorField,omitempty"`
	CreatedAt   time.Time      `json:"createdAt"`
	CompletedAt time.Time      `json:"completedAt,omitempty"`
}

// Part is one chunk object; MinCursor/MaxCursor bound its cursor values.
type Part struct {
	Key       string `json:"key"`
	Rows      int64  `json:"rows"`
	Bytes     int64  `json:"bytes"`
	MinCursor string `json:"minCursor,omitempty"`
	MaxCursor string `json:"maxCursor,omitempty"`
}

The manifest is the whole contract between extract and load. It persists the source stream's identity — PrimaryKey (the conflict target an upsert/merge sink requires), Namespace, and StreamType — so the loader can faithfully reconstruct the connector.Stream without ever touching the source again. The schema lives here exactly once rather than on every row, which is why the record model's single-schema batch spills so cleanly. And each part records its cursor MinCursor/MaxCursor, so a direct incremental run can learn the delta's high-water mark from the manifest alone.

Staging is just another connector

The elegant part: staging doesn't need special engine support. A blob Sink and Source implement the same connector interfaces as Postgres or CSV, so runner.Copy drives them unchanged. Extract is Copy(source → blob sink); load is Copy(blob source → destination sink).

const (
	maxPartRows  = 50000
	maxPartBytes = 64 << 20 // 64 MiB
)

// Write appends rows to the current part's buffer, flushing at either cap.
func (s *Sink) Write(ctx, stream, batch, _ WriteMode) (WriteResult, error) {
	w := s.streams[stream.Qualified()]
	for _, row := range batch.Rows {
		line, _ := encodeRow(w.schema, row)
		w.buf.Write(line); w.buf.WriteByte('\n'); w.rows++
		if w.cursorField != "" { /* fold cursor into part min/max */ }
		if w.rows >= maxPartRows || w.buf.Len() >= maxPartBytes {
			s.flushPart(ctx, w) // gzip, Put, append Part, rewrite manifest
		}
	}
	return WriteResult{RowsWritten: int64(batch.Len())}, nil
}

The blob sink captures raw records to gzipped JSONL parts (it ignores WriteMode — extract is a raw capture), flushing a part at the row or byte cap, whichever comes first. The subtle, important behavior is that flushPart persists an updated manifest with status: "extracting" after every landed part. A crash therefore leaves durable, discoverable, resumable parts and a manifest that names them — never orphaned objects the loader can't find. FinishStream (the StreamFinalizer hook the runner calls at the end) flushes the tail buffer and rewrites the manifest as complete with a CompletedAt.

Resume works from both ends of the seam, driven entirely by the manifest:

• Extract resume. On restart, the sink reads any existing manifest for the stream; if found, it continues part numbering at len(Parts), so already-landed parts are kept and new parts simply append. It is careful never to clobber a previously recorded PrimaryKey — an explicit cursor field wins, but stream identity is only backfilled when the loaded manifest lacks it.
• Load resume. The blob source replays a run's manifests as batches, exactly one batch per part. A resume map (qualified stream → already-committed part count) tells it where to start, so a retried load skips the parts it already wrote to the destination and picks up mid-stream.

// Read opens an iterator over the stream's parts, skipping resume[stream].
func (s *Source) Read(ctx, stream, _ string) (RecordIterator, error) {
	m, err := ReadManifest(ctx, s.b, manifestKey(s.prefix, Slug(stream.Qualified())))
	if err != nil { return nil, err }
	start := s.resume[stream.Qualified()]      // already-committed parts
	if start > len(m.Parts) { start = len(m.Parts) }
	return &partIterator{b: s.b, schema: m.Schema, parts: m.Parts, next: start}, nil
}

A hybrid run, end to end

Putting the pieces together: in a hybrid run the agent and the worker never talk directly. They rendezvous on the blob prefix. The agent runs Copy with the real source and a blob sink; the worker later runs Copy with a blob source and the real destination. The manifest's status is the signal that a stream is safe to load.

Direct vs. staged

Staging buys durability, resume, and the cross-process seam — but a blob round-trip is pure overhead when the payload is tiny. So the worker sizes the delta first: a small incremental delta (at or below SETU_DIRECT_MAX_ROWS, default 50,000 — about one part) streams directly source → operators → destination with no blob, while larger and hybrid runs stage. A run that resumes at the load phase can't go direct (the source rows only exist in blob), so only a fresh, source-sized delta takes the direct path. Same runner.Copy, different wiring — and, because the codec round-trips types exactly, identical results either way.