Execution flow

This is the exact sequence the runtime walks every time run fires. Useful to read alongside the source — most of it lives under internal/runtime/.

Steps

1. parse YAML        →  config tree
2. validate          →  errors or OK
3. resolve env vars  →  concrete config
4. build runtime     →  input, filters[], output, scheduler
5. tick              →  scheduler fires (CRON or one-shot)
6. fetch             →  input emits a batch of records
7. for each record:
   7a. for each filter in order:
       → mutate / drop / branch
   7b. queue to output
8. flush             →  output sends (batch) or per-record (single)
9. persist state     →  cursors written
10. log result       →  summary on stdout
11. loop or exit     →  scheduled? back to step 5. else exit 0.

Step-by-step

1. Parse

internal/config/parser.go opens the file, normalizes whitespace, and converts to YAML AST. JSON works too — gopkg.in/yaml.v3 parses both. Output is a Go struct tree.

2. Validate

internal/config/validator.go runs the parsed tree through the JSON Schema. Validation is recursive — each module type's schema is walked, with allOf and oneOf resolved to pick the right shape.

Errors include line + column from the original YAML so the user sees useful messages.

3. Resolve env vars

internal/config/env.go walks the tree, substituting every ${VAR_NAME} with the corresponding value from os.Environ(). Substitution happens once at startup, so a rotated secret needs a process restart.

If a ${VAR} doesn't resolve, the runtime exits before step 4 with a clear error.

4. Build runtime

internal/runtime/builder.go walks the validated config and instantiates one input, the filter chain, and one output by calling each module's New(config) constructor. The constructor returns a typed module value (e.g. *HTTPPollingInput).

The builder also wires the shared StateStore and the scheduler.

Modules are built once per pipeline, not once per run. The chain is held for the life of the process and every tick executes against the same instances, which is what lets a database pool, an HTTP connection pool, a compiled script, and the http_call / sql_call caches survive from one run to the next. Rebuilding them per tick meant reopening a connection pool on every tick — a meaningful share of a short tick's duration, and thousands of database sessions a day for nothing.

Three consequences follow from that, and they are the ones to keep in mind when writing a module:

• State inside a module persists across runs unless you clear it. The caches are the deliberate case; anything you accumulate in a module field is the accidental one. The script filter instantiates a fresh JavaScript runtime per execution for exactly this reason — a script appending to a global would otherwise grow without bound.
• A failed run drops the chain. When an execution returns an error the modules are closed and rebuilt on the next tick, so a pipeline that fails repeatedly behaves as it did before modules were reused. A run that merely logs or skips bad records succeeds, and keeps its modules.
• Whoever builds a chain that outlives a run owns closing it. The executor closes modules only when it was handed them for a single run; a caller that reuses them calls RetainModules() and closes them itself at shutdown.

5. Tick

If the input has a schedule, the scheduler runs the pipeline on every CRON match. Without a schedule, it runs exactly once.

If the runtime is still processing when the next tick fires, the runtime skips that tick and logs a warning. Runs never overlap.

6. Fetch

The input's Fetch(ctx) method runs. Implementations:

• httpPolling: builds the URL (incl. pagination + state query params), sends GET, parses the JSON, returns dataField as the records.
• webhook: returns the current delivery (one per HTTP POST).
• database: opens a connection, runs the query, scans rows.

Each Fetch returns []Record plus an opaque state delta that the runtime will persist later.

7. Filter chain

For each record, the executor calls each filter's Process(ctx, record) in declared order. Filters return:

• A modified record (continue down the chain),
• nil (drop the record),
• An error (handled by onError).

The chain stops at the first drop. A record only reaches the output if every filter passed it through.

8. Flush

The output's Send(ctx, records) runs. In batch mode, this is one call with all records that survived. In single mode, it's one call per record.

Retries happen inside Send. The runtime above only sees success or the post-retry error.

9. Persist state

After a successful batch, the runtime asks the input for its state delta and writes it to <storagePath>/<pipeline-name>.json. The write is atomic (tempfile + rename) — a crash mid-write leaves the previous state intact.

10. Log

A summary line goes to stdout:

{"time":"…","level":"info","pipeline":"sync-orders","msg":"complete","records":93,"duration_ms":1840}

11. Loop or exit

If the input has a schedule and the process hasn't received a stop signal, the runtime sleeps until the next tick. Otherwise it exits 0.

Graceful shutdown

SIGINT / SIGTERM cancels the context passed through every layer. The runtime then:

1. Stops accepting new ticks (scheduled mode) or new webhook deliveries (listener mode).
2. Finishes the current batch — Fetch and Process calls already in flight see a cancelled context and decide whether to abort cleanly or finish what they started.
3. Flushes the output for completed records.
4. Persists state.
5. Closes connections (DB pools, HTTP transports).
6. Exits 0.

TimeoutStopSec=30s (systemd) or terminationGracePeriodSeconds: 60 (Kubernetes) gives the runtime enough time for steps 2–5.

See also