poll-cycle.d2

direction: down

start: "Poll Cycle Start" { shape: oval }
snapshot: "current_snapshot()" { shape: step }
no_snap: "No snapshot?\nSleep & retry" { shape: diamond }
load: "Load cursors from\n_viaduck_state" { shape: step }
group: "Group destinations\nby last_snapshot_id" { shape: step }

mode: "key_columns\nconfigured?" { shape: diamond }
cdc_changes: "table_changes(\nstart, end,\nfilter_expr)" { shape: step }
cdc_insert: "table_insertions(\nstart, end,\nfilter_expr)" { shape: step }

empty: "Empty\nchangeset?" { shape: diamond }

phase1: "Phase 1: Resolve Preimages\n• pair by rowid\n• cross-tenant → delete\n• orphaned → delete" { shape: step }
route: "Route: split_and_count()\nArrow split by routing field" { shape: step }
phase2: "Phase 2: Resolve Conflicts\n• insert+delete same rowid → cancel\n• postimage+delete → drop postimage" { shape: step }
phase3: "Phase 3: Apply (in txn)\n• delete matching rows\n• upsert inserts + postimages" { shape: step }

append: "append() to each\ndestination" { shape: step }

advance: "advance_cursor()\n(in transaction)" { shape: step }
fail: "Write failed?" { shape: diamond }
retry: "Retry with\nbackoff (3x)" { shape: step }
error: "record_error()\nevict connection" { shape: step }
advance_empty: "advance_cursors()\n(no-data advance)" { shape: step }
metrics: "Update lag metrics" { shape: step }
done: "Sleep\npoll.interval_seconds" { shape: oval }

start -> snapshot
snapshot -> no_snap
no_snap -> done: "yes"
no_snap -> load: "no"
load -> group
group -> mode: "for each snapshot group"

mode -> cdc_changes: "yes (full CDC)"
mode -> cdc_insert: "no (append-only)"

cdc_changes -> empty
cdc_insert -> empty

empty -> advance_empty: "yes (0 rows)"
empty -> phase1: "no (full CDC)"
empty -> route: "no (append-only)"

phase1 -> route
route -> phase2: "full CDC"
route -> append: "append-only"
phase2 -> phase3

phase3 -> fail
append -> fail

fail -> advance: "no"
fail -> retry: "yes"
retry -> error: "exhausted"
retry -> advance: "succeeded"
advance -> metrics
advance_empty -> metrics
error -> metrics
metrics -> done