Pipelines and steps

A pipeline is an ordered group of steps that you run as one unit. Assets cover most needs. Pipelines add explicit control over order, branching and retries for work that does not fit into a single asset.

Steps

A step is the smallest unit of work in a run. It is a function decorated with @sy.step. Synchronoid records the start, end, inputs and outputs of every step, and it never runs a finished step twice for the same inputs.

import synchronoid as sy

@sy.step(retry=sy.Retry(max_attempts=3, backoff="exponential"))
def fetch_listings(ctx, region):
    return ctx.source.get("/listings", params={"region": region})

@sy.step
def normalize(ctx, payload):
    return [clean(item) for item in payload.records()]

Pipelines

A pipeline connects steps. Calling a step inside a pipeline does not run it immediately. It adds the step to the plan, and the return value is a reference to the future result.

@sy.pipeline(name="listings_sync")
def listings_sync(regions):
    for region in regions:
        payload = fetch_listings(region)
        rows = normalize(payload)
        publish(rows)

The three steps for each region form a chain, and the chains for different regions run in parallel. Because steps pass references, large payloads are never copied between workers.

Durable runs

A run is one execution of a pipeline or of one or more assets. Runs are durable: Synchronoid writes the result of each finished step before it starts the next. If a worker crashes, restarts or is replaced during a deploy, the run continues on another worker from the last finished step.

Why steps must be idempotent

Because a step can run more than once, running it twice has to be safe. Synchronoid helps in two ways. Writes to storage are keyed by run id, step name and record key, so a second write replaces the first. Inputs are immutable snapshots, so a step that is run again sees the same data.

Retries

Set a retry policy on a step, an asset, or a whole source. The most specific setting wins.

retry = sy.Retry(
    max_attempts=5,
    backoff="exponential",    # 2s, 4s, 8s, 16s ...
    retry_on=[sy.errors.Timeout, sy.errors.HttpStatus(429, 503)],
)

When a source keeps failing across many runs, Synchronoid pauses it automatically and alerts an operator, so a broken provider does not use up your quota. See Retries and replays.

Replaying a run

You can replay any run from the control room or the CLI. A replay re-executes the steps using the stored inputs, so it does not call the source again.

$ syn runs list --status failed
$ syn runs replay run_8f2c1a

Choosing between assets and pipelines

• Use assets when each stage produces data you want to keep and inspect.
• Use a pipeline when you need fan-out over a list, conditional branches, or several steps that produce one asset.
• You can mix both. A pipeline can read assets, and an asset can be built by a pipeline.