feat: Send state as checkpoints after the resources it covers

[jpollard-cs]

Summary

Brings the source side of state checkpoints (cloudquery/plugin-pb#42, cloudquery/plugin-sdk#2632, cloudquery/cloudquery#23546) to the Python SDK. When the CLI sets Sync.Request.supports_checkpoints, flushed state goes on the sync stream as Sync.MessageCheckpoint instead of being written straight to the backend. The CLI then persists it only after every destination commits the records synced before it.

Why we want this: some of our Python syncs run for many hours, often limited by provider rate limits. A transient failure late in the sync currently throws away all the progress, because state is only safe to write once the sync ends. With checkpoints, a failed sync keeps the cursors whose records were committed, and the next run resumes from there.

Opt-in, and unchanged unless both sides opt in:

self._scheduler = Scheduler(concurrency=...)
self._state_client = StateClientBuilder.build(
    backend_options=options.backend_options,
    checkpoints=self._scheduler.checkpoints,  # new, optional
)
self._scheduler.set_post_sync_hook(self._state_client.flush)

# in a table resolver, with one cursor per form
key = state_key("responses", client_id=client.id(), form_id=form["id"])
for item in api.list_responses(form["id"], since=state_client.get_key(key)):
    yield item
    state_client.set_key(key, item["updated_at"])
    state_client.flush()  # a checkpoint, sent once this item and its relations are on the stream

Without checkpoints, or when the CLI doesn't support them, flush() writes to the backend as it does today.

Ordering

This follows the Go SDK's semantics:

• A state change made inside a table resolver is tied to that resolver call's scope. The scheduler records the scope in a thread-local while the resolver runs.
• On flush(), each group of keys becomes a checkpoint. It goes on the scheduler's queue once every resource those scopes produced before the flush is on it, including the relations those resources spawned.
  • Child resolvers run asynchronously on the depth pools, so the checkpoint waits for every child scope that was still open at flush time.
  • Unlike Go, the Python scheduler has no batcher: resources reach the stream in queue order, so nothing needs flushing first.
• A checkpoint also waits for earlier unsent checkpoints of the same keys. The state table has no version column, so the CLI relies on stream order.
• Changes made outside table resolvers wait for nothing. Flush them after the resources they summarize, for example in the post-sync hook. Checkpoints flushed by the post-sync hook are now sent before Scheduler.sync returns.
• Flushing in checkpoint mode while no sync is running raises an error instead of dropping the state.

Granularity

Two things set how much a checkpointing sync writes, and today plugin authors and users control only one of them:

• How many keys change between commits. That's the plugin's choice: one cursor per table, or one per parent, account or other partition. This PR adds state_key(table, *, client_id=None, **partition), so plugins choose the granularity without each inventing a key format. Keys are deterministic, partition fields are sorted, and separators are escaped, so tables, clients and partitions never share a key. Each key is one state row, and a partition's checkpoint waits only for the resolver that set it. Finer keys mean a resumed sync re-fetches less, at the cost of more rows per commit.
• How often the CLI commits. The CLI merges checkpoints and commits at most every 30 seconds (checkpointInterval in cli/cmd/sync_v3_checkpoints.go). Each commit flushes every destination, then writes every key changed since the last commit to the state backend. The interval is hard-coded.

We'd like the commit interval to be configurable per sync, for example in the source spec next to backend_options, including an option to commit only at the end. The right value depends on the destination and the sync, not on the plugin:

• Destination batches: each flush ends the destination's batch, so a 30-second interval caps batch size by time. For S3 it also writes a new object per table every 30 seconds.
• State backend load: a sync with per-partition keys writes up to one row per partition per commit to a relational state backend.
• Sync length: for a sync that runs many hours against rate-limited APIs, committing every few minutes loses little on failure and writes far less. A short sync gains nothing from mid-sync commits.

Keeping the plugin-side flush cheap matters too, and it is: flush() in checkpoint mode only puts a message on the sync stream, and nothing touches a database until the CLI commits.

Other changes

• State client: now thread-safe. It clears changed keys after a successful flush (previously every flush rewrote every key changed since the start), and it keeps them after a failed one so a later flush retries them.
• Backend options: BackendOptions gains supports_checkpoints. The server now builds BackendOptions from the request instead of passing the protobuf message through.
• Server guard: the server refuses to send a checkpoint when the CLI didn't ask for them.
• Key helper: state_key is exported from cloudquery.sdk.stateclient.

Depends on

The plugin-pb-python bindings with the new messages (cloudquery/plugin-pb-python#250), which aren't released yet (release PR cloudquery/plugin-pb-python#225). pyproject.toml requires >=0.0.55, but uv.lock still needs updating after that release, so CI will fail until then. That's why this is a draft.

Open questions

1. Failed relations. Like Go, a parent's checkpoint still goes out when one of its child resolvers fails. For a parent cursor such as updated_since, that can skip the failed parent's children for good. Would you take an opt-in to hold back a scope's checkpoints when anything in its subtree errored?
2. version column. Should the Python state table get the version column, to match the Go client?
3. Python destinations. Python destination plugins don't implement Write.MessageFlush yet. I left that for a separate PR, since the CLI only enables checkpoints when every destination supports flushing.
4. Full syncs. Checkpoints only help tables that keep cursors. Resuming long full (non-incremental) syncs, and the interaction with delete-stale, is a bigger question. I'd like to raise it in a separate issue.

Testing

• New tests in tests/scheduler/checkpoints.py, tests/serve/checkpoints.py and tests/stateclient/keys.py. They cover:
  • parent checkpoints following all child resources of the parents they cover, with randomized child latency
  • child checkpoints
  • per-key order under concurrent flushes
  • post-sync hook checkpoints
  • backend writes without checkpoints
  • failed-flush retry
  • the builder and server gating
  • state_key ordering, escaping and uniqueness
• Removing either the wait on child scopes or the per-key ordering makes the parent-checkpoint test fail.
• The full suite passes locally against plugin-pb-python main, except test_plugin_package, which builds a Docker image and fails the same way locally on main.