feat(resync): Add support to update snapshot config (max_parallel_workers, num_tables, rows_per_partition) during resync signal

[masterashu]

No description provided.

[claude]

Code review

No issues found. Checked for bugs and CLAUDE.md compliance.

[claude]

Code review

Bug: Snapshot parameter overrides lost in resync-during-resync path (flow/workflows/cdc_flow.go, line 652-656). overrideSnapshotParametersInState(val, state) correctly updates the workflow state with the new snapshot parameters, but uploadConfigToCatalog(ctx, cfg) (line 653) and state.DropFlowInput.FlowConnectionConfigs = cfg (line 656) both use the raw cfg which still has the old snapshot parameter values. When DropFlowWorkflow later calls ContinueAsNew with input.FlowConnectionConfigs and nil state, NewCDCFlowWorkflowState initializes snapshot params from the config which never received the overrides. The snapshot parameter changes are silently discarded. Compare with the other resync paths (e.g., lines 516-524 and lines 889-897) which correctly use syncStateToConfigProtoInCatalog(ctx, cfg, state) to copy state values into the config. Suggested fix: After overrideSnapshotParametersInState(val, state), manually copy snapshot parameters from state to cfg before uploading to catalog.

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
2225	2	2223	205

View the top 3 failed test(s) by shortest run time

github.com/PeerDB-io/peerdb/flow/e2e::TestGenericPG

Stack Traces | 0s run time

=== RUN   TestGenericPG
=== PAUSE TestGenericPG
=== CONT  TestGenericPG
--- FAIL: TestGenericPG (0.00s)
2026/05/05 16:05:18 INFO Received AWS credentials from peer for connector: ci x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}
2026/05/05 16:05:18 INFO Received AWS credentials from peer for connector: clickhouse x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}
2026/05/05 16:05:18 INFO Executing and processing query x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart query="SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id"
2026/05/05 16:05:18 INFO Executing and processing query stream x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart query="SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id"
2026/05/05 16:05:18 INFO [pg_query_executor] declared cursor x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart cursorQuery="DECLARE peerdb_cursor_732811130957147718 CURSOR FOR SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id" args=[]
2026/05/05 16:05:18 INFO [pg_query_executor] fetching rows start x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart query="SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id" channelLen=0
2026/05/05 16:05:18 INFO [pg_query_executor] fetching from cursor x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart cursor=peerdb_cursor_732811130957147718
2026/05/05 16:05:18 INFO processed row stream x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart cursor=peerdb_cursor_732811130957147718 records=8 bytes=168 channelLen=7
2026/05/05 16:05:18 INFO [pg_query_executor] fetched rows x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart query="SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id" rows=8 bytes=168 channelLen=7
2026/05/05 16:05:18 INFO [pg_query_executor] fetching from cursor x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart cursor=peerdb_cursor_732811130957147718
2026/05/05 16:05:18 INFO processed row stream x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart cursor=peerdb_cursor_732811130957147718 records=0 bytes=0 channelLen=0
2026/05/05 16:05:18 INFO [pg_query_executor] fetched rows x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart query="SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id" rows=0 bytes=0 channelLen=0
2026/05/05 16:05:18 INFO [pg_query_executor] committing transaction x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart
2026/05/05 16:05:18 INFO [pg_query_executor] committed transaction for query x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN} partitionId=testpart query="SELECT id,name,created_at FROM e2e_test_bq_hlxratn1_20260505160435.\"test_inheritance_dynconf\" ORDER BY id" rows=8 bytes=168 channelLen=0

github.com/PeerDB-io/peerdb/flow/e2e::TestApiMongo

Stack Traces | 0.01s run time

=== RUN   TestApiMongo
=== PAUSE TestApiMongo
=== CONT  TestApiMongo
--- FAIL: TestApiMongo (0.01s)
2026/05/05 16:09:32 INFO Received AWS credentials from peer for connector: ci x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}

github.com/PeerDB-io/peerdb/flow/e2e::TestPeerFlowE2ETestSuitePG_CH

Stack Traces | 0.01s run time

=== RUN   TestPeerFlowE2ETestSuitePG_CH
=== PAUSE TestPeerFlowE2ETestSuitePG_CH
=== CONT  TestPeerFlowE2ETestSuitePG_CH
--- FAIL: TestPeerFlowE2ETestSuitePG_CH (0.01s)

github.com/PeerDB-io/peerdb/flow/e2e::TestPeerFlowE2ETestSuitePG_CH/Test_Partition_Key_Timestamp

Stack Traces | 5.09s run time

=== RUN   TestPeerFlowE2ETestSuitePG_CH/Test_Partition_Key_Timestamp
=== PAUSE TestPeerFlowE2ETestSuitePG_CH/Test_Partition_Key_Timestamp
=== CONT  TestPeerFlowE2ETestSuitePG_CH/Test_Partition_Key_Timestamp
2026/05/05 16:34:34 INFO Received AWS credentials from peer for connector: ci x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}
2026/05/05 16:34:34 INFO Received AWS credentials from peer for connector: clickhouse x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}
    clickhouse_test.go:2866: Failed to connect temporal client: failed reaching server: context deadline exceeded
    clickhouse.go:107: begin tearing down postgres schema pgch_qq2uzrms
--- FAIL: TestPeerFlowE2ETestSuitePG_CH/Test_Partition_Key_Timestamp (5.09s)

github.com/PeerDB-io/peerdb/flow/e2e::TestApiMongo/TestCancelTableAddition_NoRemovalAssumed

Stack Traces | 15.2s run time

=== RUN   TestApiMongo/TestCancelTableAddition_NoRemovalAssumed
=== PAUSE TestApiMongo/TestCancelTableAddition_NoRemovalAssumed
=== CONT  TestApiMongo/TestCancelTableAddition_NoRemovalAssumed
2026/05/05 16:08:05 INFO Received AWS credentials from peer for connector: ci x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}
2026/05/05 16:08:05 INFO Received AWS credentials from peer for connector: clickhouse x-peerdb-additional-metadata={Operation:FLOW_OPERATION_UNKNOWN}
    cancel_table_addition_test.go:361: WaitFor wait for initial load to finish 2026-05-05 16:08:13.211056166 +0000 UTC m=+271.711890443
    cancel_table_addition_test.go:365: WaitFor t1 initial load 2026-05-05 16:08:13.21162912 +0000 UTC m=+271.712463407
    cancel_table_addition_test.go:366: WaitFor t2 initial load 2026-05-05 16:08:13.216754673 +0000 UTC m=+271.717588970
    cancel_table_addition_test.go:367: WaitFor t3 initial load 2026-05-05 16:08:13.221819643 +0000 UTC m=+271.722653920
    cancel_table_addition_test.go:375: WaitFor wait for pause for add table 2026-05-05 16:08:13.230047754 +0000 UTC m=+271.730882040
    cancel_table_addition_test.go:376: UNEXPECTED ERROR unable to establish connection with catalog: FATAL: terminating connection due to administrator command (SQLSTATE 57P01)
    api_test.go:49: begin tearing down postgres schema api_hmpnsjyv
--- FAIL: TestApiMongo/TestCancelTableAddition_NoRemovalAssumed (15.15s)

github.com/PeerDB-io/peerdb/flow/e2e::TestGenericPG/Test_Custom_Replication_Slot_Starting_With_Numbers_CDC_Only

Stack Traces | 31.4s run time

=== RUN   TestGenericPG/Test_Custom_Replication_Slot_Starting_With_Numbers_CDC_Only
=== PAUSE TestGenericPG/Test_Custom_Replication_Slot_Starting_With_Numbers_CDC_Only
=== CONT  TestGenericPG/Test_Custom_Replication_Slot_Starting_With_Numbers_CDC_Only
    generic_test.go:1162: Inserted 5 initial rows before creating replication slot
    generic_test.go:1168: Created custom replication slot: 112_custom_slot_nj3q2v4t
    generic_test.go:1176: Inserted 5 initial rows after creating replication slot
    generic_test.go:1197: UNEXPECTED STATUS TIMEOUT STATUS_SETUP
    postgres.go:92: begin tearing down postgres schema pg_4bf4xpdq
--- FAIL: TestGenericPG/Test_Custom_Replication_Slot_Starting_With_Numbers_CDC_Only (31.37s)

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[github-actions]

🔄 Flaky Test Detected

Analysis: Two e2e tests failed due to flaky conditions: a CDC setup phase timeout (UNEXPECTED STATUS TIMEOUT STATUS_SETUP) and a Postgres catalog connection killed by a parallel test's teardown (SQLSTATE 57P01: terminating connection due to administrator command).
Confidence: 0.92

✅ Automatically retrying the workflow

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: The e2e test Test_Partition_Key_Timestamp failed with a transient context deadline exceeded error when connecting to the Temporal server, indicating a CI infrastructure timeout rather than a code regression.
Confidence: 0.92

✅ Automatically retrying the workflow

View workflow run

[masterashu]

moved to #4276