---
title: "Run sync pipeline (read → write)"
method: POST
path: "/pipeline_sync"
tags: ["Stateless Sync API"]
---

# Run sync pipeline (read → write)

`POST /pipeline_sync`

Reads from the source connector and writes to the destination (backfill mode).

## Request body

- object
  - `pipeline` PipelineConfig, required
    - `source` SourceConfig, required — unresolved $ref
    - `destination` union, required
      - object
        - `type` 'postgres', required
        - `postgres` DestinationPostgresConfig, required
          - `url` string — Postgres connection string
          - `connection_string` string — Deprecated alias for url; prefer url
          - `schema` string — Target schema name (e.g. "stripe")
          - `batch_size` number — Records to buffer before flushing
          - `aws` object — AWS RDS IAM authentication config
            - `host` string, required — Postgres host for RDS IAM auth
            - `port` number — Postgres port for RDS IAM auth
            - `database` string, required — Database name for RDS IAM auth
            - `user` string, required — Database user for RDS IAM auth
            - `region` string, required — AWS region for RDS instance
            - `role_arn` string — IAM role ARN to assume (cross-account)
            - `external_id` string — External ID for STS AssumeRole
          - `ssl_ca_pem` string — PEM-encoded CA certificate for SSL verification (required for verify-ca / verify-full with a private CA)
      - object
        - `type` 'google_sheets', required
        - `google_sheets` DestinationGoogleSheetsConfig, required
          - `client_id` string — Google OAuth2 client ID (env: GOOGLE_CLIENT_ID)
          - `client_secret` string — Google OAuth2 client secret (env: GOOGLE_CLIENT_SECRET)
          - `access_token` string, nullable
          - `refresh_token` string, required — OAuth2 refresh token
          - `spreadsheet_id` string — Target spreadsheet ID (created if omitted)
          - `spreadsheet_title` string — Title when creating a new spreadsheet
          - `batch_size` number — Rows per Sheets API append call
      - object
        - `type` 'stripe', required
        - `stripe` union, required
          - object
            - `api_key` string, required — Stripe API key (sk_test_... or sk_live_...)
            - `base_url` string, uri — Override the Stripe API base URL (e.g. http://localhost:12111 for tests)
            - `max_retries` integer — Retries for 429/5xx/network errors
            - `api_version` 'unsafe-development', required
            - `object` 'custom_object', required
            - `write_mode` 'create', required
            - `streams` object, required — Per-source-stream Custom Object write configuration.
          - object
            - `api_key` string, required — Stripe API key (sk_test_... or sk_live_...)
            - `base_url` string, uri — Override the Stripe API base URL (e.g. http://localhost:12111 for tests)
            - `max_retries` integer — Retries for 429/5xx/network errors
            - `api_version` '2026-03-25.dahlia' | '2026-02-25.clover' | '2026-01-28.clover' | '2025-12-15.clover' | '2025-11-17.clover' | '2025-10-29.clover' | '2025-09-30.clover' | '2025-08-27.basil' | '2025-07-30.basil' | '2025-06-30.basil' | '2025-05-28.basil' | '2025-04-30.basil' | '2025-03-31.basil' | '2025-02-24.acacia' | '2025-01-27.acacia' | '2024-12-18.acacia' | '2024-11-20.acacia' | '2024-10-28.acacia' | '2024-09-30.acacia' | '2024-06-20' | '2024-04-10' | '2024-04-03' | '2023-10-16' | '2023-08-16' | '2022-11-15' | '2022-08-01' | '2020-08-27' | '2020-03-02' | '2019-12-03' | '2019-11-05' | '2019-10-17' | '2019-10-08' | '2019-09-09' | '2019-08-14' | '2019-05-16' | '2019-03-14' | '2019-02-19' | '2019-02-11' | '2018-11-08' | '2018-10-31' | '2018-09-24' | '2018-09-06' | '2018-08-23' | '2018-07-27' | '2018-05-21' | '2018-02-28' | '2018-02-06' | '2018-02-05' | '2018-01-23' | '2017-12-14' | '2017-08-15', required
            - `object` 'standard_object', required
            - `write_mode` 'create', required
            - `streams` object, required — Per-source-stream standard Stripe object create configuration.
      - object
        - `type` 'redis', required
        - `redis` DestinationRedisConfig, required
          - `url` string — Redis connection URL (redis://host:port)
          - `host` string — Redis host (default: localhost)
          - `port` number — Redis port (default: 6379)
          - `password` string — Redis password
          - `db` number — Redis database number (default: 0)
          - `tls` boolean — Enable TLS
          - `key_prefix` string — Prefix for all Redis keys (default: empty)
          - `batch_size` number — Records to buffer before flushing via pipeline
    - `streams` object[]
      - `name` string, required — Stream (table) name to sync.
      - `sync_mode` 'incremental' | 'full_refresh' — How the source reads this stream. Defaults to full_refresh.
      - `fields` string[] — If set, only these fields are synced.
      - `backfill_limit` integer — Cap backfill to this many records, then mark the stream complete.
  - `time_limit` number — Stop streaming after N seconds.
  - `soft_time_limit` number — Soft wall-clock deadline in seconds. Stops reading from the source between messages; the destination continues to drain and flush until time_limit fires.
  - `run_id` string — Optional sync run identifier used to track bounded sync progress.
  - `stdin` Message[] — Optional array of input messages (push mode). Without stdin, reads from the source connector (backfill mode). — unresolved $ref
  - `state` SyncState — Full sync checkpoint with separate sections for source, destination, and sync run. Connectors only see their own section; the engine manages routing.
    - `source` SourceState, required — Source connector state — cursors, backfill progress, events cursors.
      - `streams` object, required — Per-stream checkpoint data, keyed by stream name.
      - `global` object, required — Source-wide state shared across all streams.
    - `destination` object, required — Destination connector state.
    - `sync_run` object, required — Engine-managed run state — run_id, time_ceiling, accumulated progress.
      - `run_id` string — Identifies a finite backfill run. Omit for continuous sync.
      - `time_ceiling` string — Frozen upper bound (ISO 8601). Set on first invocation when run_id is present; reused on continuation.
      - `progress` ProgressPayload, required — Periodic sync progress emitted by the engine as a top-level message. Each emission is a full replacement.
        - `started_at` string, required — When this sync started (ISO 8601); generally equals time_ceiling.
        - `elapsed_ms` integer, required — Wall-clock milliseconds since the sync run started.
        - `global_state_count` integer, required — Total source_state messages observed so far.
        - `connection_status` object — Set when source or destination emits connection_status: failed.
          - `status` 'succeeded' | 'failed', required — Whether the connection check passed.
          - `message` string — Human-readable explanation of the check result.
        - `derived` object, required — Computed aggregates.
          - `status` 'started' | 'succeeded' | 'failed', required — succeeded = all streams completed/skipped; failed = connection_status failed OR any stream errored.
          - `records_per_second` number, required — Overall throughput for the entire run.
          - `states_per_second` number, required — State checkpoints per second.
          - `total_record_count` integer, required — Total records across all streams.
          - `total_state_count` integer, required — Total source_state messages across all streams.
        - `streams` object, required — Per-stream progress, keyed by stream name.

## Response `200`

NDJSON stream of sync messages

## Other responses

- `400` — Invalid params

## Changes

- **2026-07-17** `7f4c6ed52875` — 2 breaking, 5 info
  - removed `subschema #1, subschema #2` from the `stdin/items/oneOf[#/components/schemas/ControlMessage]/control` request property `oneOf` list
  - added `subschema #1, subschema #2` to the `oneOf[#/components/schemas/ControlMessage]/control` response property `oneOf` list for the response status `200`
  - added the new optional request property `stdin/items/oneOf[#/components/schemas/RecordMessage]/record/recordDeleted`
  - added `subschema #1, subschema #2` to the `stdin/items/oneOf[#/components/schemas/ControlMessage]/control` request property `oneOf` list
  - …3 more
- **2026-04-30** `4b2cc3fc8044` — 6 info
  - added the new optional request property `pipeline/source/oneOf[subschema #2]/metronome/webhook_port`
  - added the new optional request property `pipeline/source/oneOf[subschema #2]/metronome/webhook_secret`
  - added the new optional request property `stdin/items/oneOf[#/components/schemas/ControlMessage]/control/oneOf[subschema #1]/source_config/oneOf[#/components/schemas/SourceMetronomeConfig]/webhook_port`
  - added the new optional request property `stdin/items/oneOf[#/components/schemas/ControlMessage]/control/oneOf[subschema #1]/source_config/oneOf[#/components/schemas/SourceMetronomeConfig]/webhook_secret`
  - …2 more
- …earlier changes not shown

[Full history](https://skmtc.dev/stripe/apis/stripe-sync-engine/changes/pipeline_sync/post.md)

---

[API](https://skmtc.dev/stripe/apis/stripe-sync-engine.md) · [All operations](https://skmtc.dev/stripe/apis/stripe-sync-engine/llms.txt) · [OpenAPI document](https://skmtc-service-production.skmtc.workers.dev/v1/apis/stripe/stripe-sync-engine/revisions/7f4c6ed52875/schema)