State persistence

When a scheduled pipeline restarts — because of a deploy, an OOM, or your process supervisor — you usually don't want to refetch everything from scratch. State persistence keeps a small cursor file on disk that the input reads at boot and writes after every successful batch.

The three persistable cursors

Most inputs support a mix. You pick the ones that match the source API's pagination semantics.

httpPolling example

input:
  type: httpPolling
  schedule: "*/10 * * * *"
  endpoint: https://source.example.com/api/events
  dataField: events
  statePersistence:
    timestamp:
      enabled: true
      field: event.timestamp      # which record field is the timestamp
      queryParam: updated_after   # how the source API expects it
    id:
      enabled: true
      field: event.id             # which record field is the ID
      queryParam: after_id
    storagePath: ./.cannectors-state

After each successful batch, Cannectors writes the max-seen values into ./.cannectors-state/<pipeline-name>.json. On the next run, it reads that file and appends the relevant query parameters to the request.

Reading the state yourself

queryParam is the one-line shortcut, and it only reaches a query parameter. When the source filters incrementally by another route — an OData $filter, a path segment, an If-Modified-Since header, a POST search body — read the state directly instead. It is a top-level template variable, with the same two names in every input — httpPolling's endpoint, headers, queryParams and body, the soapPolling body, and the database parameter expressions:

input:
  type: httpPolling
  endpoint: https://source.example.com/api/orders?$filter=LastModifiedDate gt {{ state.lastTimestamp }}
  headers:
    X-Since: "{{ state.lastTimestamp }}"
  queryParams:
    $orderby: LastModifiedDate asc
  statePersistence:
    timestamp:
      enabled: true

The two mechanisms combine: queryParam still applies, and its value wins over a queryParams entry of the same name, because it is computed for the run at hand.

database example

For database inputs, the cursor is a SQL column instead of a query parameter:

input:
  type: database
  connectionStringRef: ${SOURCE_DATABASE_URL}
  query: |
    SELECT id, updated_at, payload
    FROM events
    WHERE updated_at > $1
    ORDER BY updated_at ASC
    LIMIT 500
  parameters:
    - state.lastTimestamp
  statePersistence:
    timestamp:
      enabled: true
      field: updated_at
    storagePath: ./.cannectors-state

The $1 placeholder is bound to the state.lastTimestamp parameter expression at runtime. On the first run, when no state exists yet, it evaluates to the epoch (1970-01-01T00:00:00Z) so the query returns everything.

Where to put storagePath

If the storage disappears between runs, the pipeline restarts from scratch — possibly re-processing already-shipped records.

Delivery guarantee and crashes

Cannectors delivers at least once. A record can reach its destination more than once; it should not silently fail to arrive. Plan for that on the receiving side — an idempotent upsert keyed on a business id is the usual answer, not a plain insert.

Two things cause a repeat:

• A retry replays the whole batch. When a batch fails and is retried, every record in it is sent again, including those the destination had already accepted before the failure.
• A crash between delivery and the state write. The state is saved after the output succeeds, so a process killed in that window restarts from the previous cursor and re-sends the last batch.

The state file itself is written atomically — to a temporary file, then renamed — so a crash leaves either the previous state or the new one, never a half-written file. That holds under SIGKILL, which the lab exercises at several points of a run.

Backfills

To force a fresh full sync, delete the relevant state file:

rm ./.cannectors-state/<pipeline-name>.json

The next run will treat itself as a fresh start. There is no "backfill from date X" knob in the YAML — instead, write to the state file manually if you really need to seed it:

echo '{"timestamp": "2026-01-01T00:00:00Z"}' \
  > ./.cannectors-state/<pipeline-name>.json

Cross-references