[source-postgres] CDC initial snapshot can skip unfinished CTID snapshot after socket state checksum rejection

### Connector Name

source-postgres

### Connector Version

3.8.1

### What step the error happened?

During the sync

### Relevant information

## Summary
In PostgreSQL CDC mode, concurrent CTID-based initial snapshot partitions can emit records ahead of serial CTID checkpoint accounting. In socket mode this can produce GLOBAL state messages whose `sourceStats.recordCount` does not match the destination/platform-tracked records for the state partition. The destination rejects the CTID snapshot state as invalid, while heartbeat/WAL CDC shared state may still be accepted.
On retry, Airbyte can then have CDC shared state but missing or empty per-stream CTID snapshot state. The source-postgres CDC partition factory interprets `maybeCtid == null` as “snapshot complete” and resumes CDC-only, skipping the unfinished snapshot.
## Expected behavior
If an initial CDC snapshot is not durably checkpointed, the next retry should resume or restart the CTID snapshot before advancing to CDC-only reading.
State messages emitted during concurrent snapshot reads should have source record counts that match the records associated with the same destination/platform-tracked partition.
## Actual behavior
The sync emits invalid GLOBAL state messages like:
`Source state message checksum is invalid: state record count ... does not equal platform tracked record count ...`
The rejected state contains CTID snapshot stream state, but the accepted state can still contain CDC shared WAL/heartbeat state.
After retry, source-postgres sees CDC shared state with empty or missing CTID stream state and skips the remaining initial snapshot.
## Root cause analysis
There appear to be two related issues.
1. In the Bulk CDK socket path, JDBC partition readers emit records concurrently with a socket `partition_id`, but `FeedReader` applies partition checkpoints serially later. `StateManager.GlobalStateManager.checkpoint()` can emit GLOBAL state with `sourceStats.recordCount` from a stream partition without propagating the matching top-level `partition_id`. `FeedReader.maybeCheckpoint()` then generates a random top-level `partition_id`, so destination state validation sees a record-count mismatch.
Relevant files:
`airbyte-cdk/bulk/core/extract/src/main/kotlin/io/airbyte/cdk/read/FeedReader.kt`
`airbyte-cdk/bulk/core/extract/src/main/kotlin/io/airbyte/cdk/read/StateManager.kt`
`airbyte-cdk/bulk/toolkits/extract-jdbc/src/main/kotlin/io/airbyte/cdk/read/JdbcPartitionReader.kt`
2. In source-postgres CDC mode, an empty stream state `{}` or missing CTID state is treated the same as completed CTID snapshot state because `maybeCtid == null`.
Relevant file:
`airbyte-integrations/connectors/source-postgres/src/main/kotlin/io/airbyte/integrations/source/postgres/PostgresSourceJdbcPartitionFactory.kt`

### Relevant log output

```shell
2026-06-29 09:36:04 replication-orchestrator ERROR Source state message checksum is invalid: state record count 1963104.0 does not equal platform tracked record count 1.5631028E7. No hash collisions were observed.
2026-06-29 09:36:04 replication-orchestrator ERROR Raw state message with bad count {"type":"GLOBAL","global":{"shared_state":{"state":{"[\"loan_service_database\",{\"server\":\"loan_service_database\"}]":"{\"lsn_proc\":152584644074640,\"messageType\":\"MESSAGE\",\"lsn_commit\":152584644074640,\"lsn\":152584644109888,\"txId\":4139155009,\"ts_usec\":1782736281930442}"}},"stream_states":[{"stream_descriptor":{"name":"installment","namespace":"public"},"stream_state":{"version":3,"state_type":"ctid_based","stream_name":"","stream_namespace":"","ctid":"(32332,1)","cursors":{},"relation_filenode":255076647}},{"stream_descriptor":{"name":"loan_task","namespace":"public"},"stream_state":{}}]},"sourceStats":{"recordCount":1963104.0},"id":3}
2026-06-29 09:36:05 replication-orchestrator ERROR Source state message checksum is invalid: state record count 1940873.0 does not equal platform tracked record count 0.0. No hash collisions were observed.
2026-06-29 09:36:05 replication-orchestrator ERROR Raw state message with bad count {"type":"GLOBAL","global":{"shared_state":{"state":{"[\"loan_service_database\",{\"server\":\"loan_service_database\"}]":"{\"lsn_proc\":152584644074640,\"messageType\":\"MESSAGE\",\"lsn_commit\":152584644074640,\"lsn\":152584644109888,\"txId\":4139155009,\"ts_usec\":1782736281930442}"}},"stream_states":[{"stream_descriptor":{"name":"installment","namespace":"public"},"stream_state":{"version":3,"state_type":"ctid_based","stream_name":"","stream_namespace":"","ctid":"(64664,1)","cursors":{},"relation_filenode":255076647}},{"stream_descriptor":{"name":"loan_task","namespace":"public"},"stream_state":{}}]},"sourceStats":{"recordCount":1940873.0},"id":4}
2026-06-29 09:37:01 replication-orchestrator ERROR Source state message checksum is invalid: state record count 2361380.0 does not equal platform tracked record count 2457837.0. No hash collisions were observed.
2026-06-29 09:37:01 replication-orchestrator ERROR Raw state message with bad count {"type":"GLOBAL","global":{"shared_state":{"state":{"[\"loan_service_database\",{\"server\":\"loan_service_database\"}]":"{\"lsn_proc\":152584644074640,\"messageType\":\"MESSAGE\",\"lsn_commit\":152584644074640,\"lsn\":152584644109888,\"txId\":4139155009,\"ts_usec\":1782736281930442}"}},"stream_states":[{"stream_descriptor":{"name":"installment","namespace":"public"},"stream_state":{"version":3,"state_type":"ctid_based","stream_name":"","stream_namespace":"","ctid":"(161660,1)","cursors":{},"relation_filenode":255076647}},{"stream_descriptor":{"name":"loan_task","namespace":"public"},"stream_state":{"version":3,"state_type":"ctid_based","stream_name":"","stream_namespace":"","ctid":"(70095,1)","cursors":{},"relation_filenode":255076689}}]},"sourceStats":{"recordCount":2361380.0},"id":8}
```

### Contribute

- [x] Yes, I want to contribute

---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/13013