fix(connectors): defer postgres source progress until ack

[rohankumardubey]

Which issue does this PR address?

Closes #3635

Rationale

The PostgreSQL source advanced tracking offsets, deleted or marked rows, and consumed CDC changes before Iggy confirmed delivery. A failed send could therefore permanently skip source records.

What changed?

PostgreSQL polling now stages cursor updates and row operations until the runtime reports a successful batch acknowledgment. NACK discards the staged work, while ACK commits the state and performs the pending delete or mark operations.

CDC now peeks logical-slot changes and advances the slot only after acknowledgment. A deterministic regression test stops Iggy during delivery and verifies that the PostgreSQL rows are redelivered after restart.

Local Execution

• Passed cargo fmt --all -- --check
• Passed cargo clippy -p iggy_connector_postgres_source -p integration --all-features --all-targets -- -D warnings
• Passed all 70 PostgreSQL source unit tests
• Passed all 7 PostgreSQL polling integration tests
• Passed both PostgreSQL CDC integration tests
• Passed the deterministic kill-server regression test
• Passed git diff --check

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[codecov]

Codecov Report

❌ Patch coverage is 89.72222% with 37 lines in your changes missing coverage. Please review.
✅ Project coverage is 59.06%. Comparing base (412014a) to head (e345553).

Files with missing lines	Patch %	Lines
core/connectors/sources/postgres_source/src/lib.rs	88.92%	22 Missing and 15 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3957       +/-   ##
=============================================
- Coverage     85.25%   59.06%   -26.19%     
  Complexity     1402     1402               
=============================================
  Files          1236     1235        -1     
  Lines        182594   150607    -31987     
  Branches     148898   116913    -31985     
=============================================
- Hits         155664    88956    -66708     
- Misses        22886    57806    +34920     
+ Partials       4044     3845      -199     

Components	Coverage Δ
Rust Core	51.99% <89.72%> (-34.14%)	⬇️
Java SDK	67.29% <ø> (ø)
C# SDK	76.33% <ø> (-0.10%)	⬇️
Python SDK	90.06% <ø> (ø)
PHP SDK	85.65% <ø> (ø)
Node SDK	96.22% <ø> (-0.03%)	⬇️
Go SDK	69.35% <ø> (+0.03%)	⬆️

Files with missing lines	Coverage Δ
core/connectors/runtime/src/stream.rs	91.15% <100.00%> (+1.05%)	⬆️
core/connectors/sources/postgres_source/src/lib.rs	76.12% <88.92%> (+2.80%)	⬆️

... and 421 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[rohankumardubey]

/ready

[rohankumardubey]

/request-review @hubcio

[hubcio]

a few things without diff lines to hang them on:

• .claude/skills/connector-source/SKILL.md still teaches the pre-ack pattern this pr removes (the "matches poll_tables" snippet writes cursors during poll, "always return state in every ProducedMessages" no longer holds for empty polls, and state-serialization failure is now a hard poll error, lib.rs:288-296) and doesn't mention on_batch_result at all - needs an update to the new contract.
• mark_or_delete_processed_rows maps db errors to Error::InvalidRecord (lib.rs:732, :751) - misleading now that these errors surface through the ack path; Error::Connection like advance_replication_slot uses would fit better.
• README.md:15 still promises offset tracking "avoid duplicates" - at-least-once redelivery means duplicates are possible; and README.md:57 documents the poll_interval default as 1s while the code fallback is 10s (lib.rs:180).

[hubcio]

@rohankumardubey please write
/ready
when PR is ready for review. at any top level comment, in the beginning of any line. this will let know maintainers that something is ready for review by flipping gh labels.

[rohankumardubey]

/ready

[hubcio]

looks like CI is failing.

[rohankumardubey]

The failure came from a merge-resolution type mismatch in the TLS connection-string branch. Thanks @hubcio

[rohankumardubey]

/ready

[hubcio]

the staging model is right and closes the old send-failure hole. two things block: 55006 isn't transient, so any slot collision at ack time stops the source for good, and an empty cdc peek never advances the slot, so an idle database retains wal without bound. the tracking <= boundary guard also regresses custom_query, null and numeric tracking users, and the new cdc test races the connector on the slot.

outside this diff, two pre-existing bugs in build_polling_query worth their own issue: $now is replaced before $now_unix so $now_unix never expands, and $offset is substituted raw into custom_query from row data.

[rohankumardubey]

@hubcio Agreed, I’ll track staged-operation persistence and replay, along with the two pre-existing custom_query bugs, in separate follow-up issues.

[hubcio]

@rohankumardubey sure thing. if you plan to do that, please fill-in the TODOs in code so that it won't be forgotten (in relevant places).

[rohankumardubey]

/ready