feat: Send incremental state as checkpoints after the resources it covers

[disq]

Summary

Adds source checkpoints and destination flushes, so the CLI can persist state only after its resources commit.

Sources opt in with state.ClientOptions{Checkpoints: res}, destinations with plugin.WithFlushSupport().
Behaviour change: with checkpoints on, a failed batch write is reported on the next flush, with the tables it failed for, so the CLI stops saving state for that sync instead of only logging the failure.
After a cancelled sync, including a quota stop, Flush returns state.ErrSyncAborted for keys set in table resolvers and does not save them, as the scheduler may have dropped their rows.
scheduler.Sync now returns the cancel cause, not nil, when the context is cancelled while the batcher closes.
Also fixes NewConnectedClientWithOptions with a zero MaxMsgSizeInBytes, which set a 0-byte gRPC message limit.
Depends on cloudquery/plugin-pb#42 and cloudquery/plugin-pb-go#710: go.mod replaces plugin-pb-go with that branch until it is released.

[erezrokah]

Found two issues in the streaming flush path:

• Flush can report success after startWorker rejects an insert, allowing a checkpoint to advance past unwritten rows. Dispatch errors should propagate to the flush result.
• Cancellation during commit abandons the unbuffered acknowledgement channel, leaving the worker and Close blocked indefinitely. A buffered or cancellation-aware acknowledgement would avoid this.

Both reproduced in focused tests.

[disq]

@erezrokah Thanks, both reproduced and fixed in ca3497e:

• A message startWorker rejects now fails every later flush in that Write call, so a checkpoint cannot pass unwritten rows (TestFlushMessageReportsRejectedMessages).
• The commit acknowledgement channel is buffered, so a cancelled flush no longer blocks the worker or Close (TestFlushMessageCancelledDuringCommit).

[erezrokah]

The original issues are fixed. I reproduced two remaining problems:

• A later flush of a shared cursor can overtake an earlier resolver’s unfinished rows because flush clears its pending dependencies.
• Shuffle-queue can combine valid checkpoints into one oversized message, causing the sync to fail.

Could we preserve shared-key dependencies across flushes and split coalesced checkpoints to respect the message limit?

[disq]

@erezrokah Both fixed in 363038e:

• The state client keeps each key's last checkpoint until it is sent, and the next checkpoint of that key waits for it, so a later flush of a shared cursor can no longer overtake an earlier resolver's unsent rows (TestStateClientCheckpointWaitsForPreviousCheckpointOfKey). Checkpoints merged during a shuffle-queue sync count as sent once the merged one is sent.
• The server now splits an oversized checkpoint across several messages, the same way it splits inserts, so a merged shuffle-queue checkpoint or a single large Flush stays within the message limit (TestSendRecordSplitsOversizedCheckpoint).

[erezrokah]

All four earlier issues are fixed. I found one regression: dependency barriers bypass shuffle-queue coalescing, retaining every intermediate cursor record until sync completion.

The same state-client test produces 1 checkpoint for 100 flushes on the previous revision, but 100 checkpoints now. Could we preserve coalescing when tracking prior-checkpoint dependencies and cover the actual SetKey/Flush path in a regression test?

[disq]

@erezrokah Fixed in b2202a5: a scope can now say a checkpoint passed to it is never sent before one it already holds, and the shuffle-queue scope does so for checkpoints merged into its held state. A repeated flush of a key in a shuffle-queue sync then skips the dependency barrier and is merged again. TestShuffleQueueCoalescesFlushedCheckpoints covers the SetKey/Flush path: 100 flushes produce one checkpoint, sent after every row.