Retries and replays

Providers fail, networks drop and workers restart. Synchronoid retries transient failures for you, pauses sources that keep failing, and lets you replay any run from the raw layer. This guide explains how each part works and what your steps must guarantee.

Retry policies

A retry policy is a sy.Retry object. You can attach it to an asset, a step or a whole pipeline. The most specific policy wins: a step policy overrides an asset policy, and an asset policy overrides the project default in synchronoid.yaml.

import synchronoid as sy

policy = sy.Retry(
    max_attempts=5,
    backoff="exponential",
    initial_delay="2s",
    max_delay="5m",
    jitter=True,
    retry_on=[sy.errors.Timeout, sy.errors.HttpStatus(429, 500, 502, 503, 504)],
)

@sy.asset(layer="staged", source=companies, retry=policy)
def company_profiles(ctx, raw):
    ctx.log.info("attempt %s", ctx.attempt)
    ...

Errors that retrying cannot fix, such as an HTTP 401 response, schema violations and exceptions raised by your own code, are not retried unless you list them in retry_on.

How a retry behaves

When a step fails with a retryable error, the run moves to Retrying in the control room and the step is scheduled again after the backoff delay. Only the failed step is repeated. Steps that already finished keep their results, because steps pass references to stored data and not the data itself. If a worker dies in the middle of a step, the run resumes from the last finished step.

Automatic source pausing

A source that keeps failing wastes budget and may get your account blocked by the provider. Each source has a circuit breaker. When the failure rate crosses a threshold, Synchronoid pauses the source and alerts an operator.

sources:
  company_registry:
    auto_pause:
      after_failures: 10        # consecutive failed runs or failed attempts
      window: "30m"
      notify: ["oncall-pagerduty"]

While a source is paused, schedules that depend on it are skipped, and sensors stop polling. Existing data stays available in all layers. After you fix the cause, resume the source:

$ syn sources pause company_registry
$ syn sources resume company_registry

Resuming runs one probe request first. If it fails, the source pauses again and no queued work starts.

Replaying a run

Every request payload is stored untouched in the raw layer. A replay runs your current code against that stored data, so it does not call the provider again. Use it after you fix a bug in a transformation or add a field to a staged asset.

$ syn runs list --status failed --limit 5
$ syn runs replay run_8f3a21 --from-layer raw
Replay run_c91d07 started from run_8f3a21 (raw, 4 assets)

$ syn runs replay run_8f3a21 --asset company_profiles --dry-run

A replay creates a new run. The original run is never changed, so its history stays accurate. Provenance on replayed values keeps the original fetch time and source and adds the id of the replay run.

Idempotency requirements

Retries and replays only work if running a step twice has the same effect as running it once. Synchronoid enforces this for built-in operations such as merge, and you must keep it for your own code.

• Write by key. Use merge(key=...) or upserts. Do not append rows without a key.
• Do not read the clock. Use ctx.run_time and ctx.partition instead of datetime.now().
• Make side effects safe. When you call an outside service, send an idempotency key such as ctx.idempotency_key.
• Do not depend on other runs. A step should produce the same output from the same input.

@sy.step(retry=sy.Retry(max_attempts=3))
def post_summary(ctx, summary):
    requests.post(
        CRM_URL,
        json=summary,
        headers={"Idempotency-Key": ctx.idempotency_key},
    )

Next steps

• Run large historical loads safely with backfills.
• Send failures to the right people with alerts.
• Look up error codes in Troubleshooting.