[disq]

Summary

When every destination supports flush, the CLI flushes them before it writes each source state checkpoint to the backend.

Tested with a local SQLite sync: after a source crash, the stored cursor never passed the committed rows.
Uses the protocol from plugin-pb-go v1.28.2.
Checkpoints stay off with transformers or destinations without flush support. The sync logs whether it requested checkpoints, or why not, and the sync command reference documents the feature and its failure policy.
If a destination or the state backend reports a failed flush, the CLI saves no more state for that source, finishes the sync and exits with an error, which the sync summary also gives as state_error. The next sources still sync.
A destination that ends its stream after it reports the failure still lets the sync finish. A writer that stops on a failed write before it reports it on a flush, such as the mixed batch writer, still ends the whole run, as on main.
At the end of the sync, the CLI flushes failed destinations again, so it knows every table they lost records of.
If the source fails after a checkpoint, the CLI still saves the state that checkpoint covers.
Side effects of this policy:

• A row that a destination always rejects stops every incremental cursor of the source, not only the cursor of its table.
• Delete-stale is skipped for the relation trees of the failed tables on every such run, or for every table if the destination does not name them or ends its stream.
• In append mode, each rerun appends again the rows synced after the last saved state.

Also, with or without checkpoints, the CLI now closes every destination before it returns a close error.

[erezrokah]

Can we flush pending checkpoints while the source is idle and split large checkpoint batches to stay within the gRPC message limit?

[disq]

@erezrokah Both done in 118eac4:

• The CLI now receives sync messages on a separate goroutine, so pending state is persisted when the 30s interval elapses even while the source sends nothing (TestCheckpointerCommitsWhileSourceIsIdle).
• State written to the backend is split into records of about 64 MiB, well under the 200 MiB gRPC limit (TestCheckpointerSplitsLargeState).

Checkpoint messages from the source are not split yet: a single Flush larger than the SDK's 100 MiB message limit still fails the sync. I can split those in cloudquery/plugin-sdk#2632 if you want.

[erezrokah]

Could we apply checkpointCommitError to the receive-error path too? When a destination write fails while the source is idle, the receiver can report only context canceled, hiding the actual destination error.

[disq]

@erezrokah Done in 30ee29e: the receive-error path now goes through the same check (renamed to unlessFailedElsewhere), so when a pipeline's destination write fails while the source is idle, the sync returns the destination error instead of context canceled. TestSyncReportsDestinationErrorWhileSourceIsIdle covers it end to end.

[greptile-apps]

Confidence Score: 5/5

[Critical impact] The latest changes appear safe to merge and address both previous findings.

Summary

The CLI saves source checkpoints only after all destinations commit the preceding records. Failed commits stop further state saves while allowing later sources to run.

• Sync checkpoints wait until destinations commit earlier records.
• Failed commits stop further state saves without stopping the sync.
• The sync closes every destination before returning close errors.

Diagram

sequenceDiagram
  participant Source
  participant CLI
  participant Destinations
  participant Backend
  Source->>CLI: Records and checkpoint
  CLI->>Destinations: Deliver preceding records
  CLI->>Destinations: Flush
  Destinations-->>CLI: Commit result
  alt Every destination committed
    CLI->>Backend: Save checkpoint
  else A commit failed
    CLI->>CLI: Stop saving state
  end
  CLI->>Destinations: Close every stream and plugin
  CLI->>CLI: Write final file summaries and logs

Loading

_{Reviews (2) · Last reviewed commit: "fix: Write the summary files once the de..." · Reviewed by Greptile}

[disq]

@greptileai retry