SPOC-311: Fix data loss when adding a node under write load

When a new node is added to a running cluster while the existing
nodes are handling writes, the new node can end up permanently
missing rows or accumulating stale updates.

Root cause: in the fallback path (Case B) of create_slot_with_progress(),
the resume LSN (P_snap) was derived from ros_remote_lsn read after the
slot-creation retry loop.  By that time the position could have advanced
past the actual COPY boundary, creating a gap of permanently lost rows
or allowing duplicate transactions to replay and corrupt balances.

Fix: capture v_pre_ros_lsn just before each slot-creation attempt and
use it as the preferred P_snap fallback.  This value is guaranteed to
be at or before the COPY boundary.  Add a MAX-by-LSN guard in
spock_group_progress_force_set_list() to prevent overwriting a live
apply worker's higher LSN with a stale P_snap value.

Also add the create_slot_with_progress() function to the upgrade SQL,
update zodan.sql procedures, and adjust the TAP test timing.

Author: Ibrar Ahmed

.github/workflows/zodan_sync.yml

@@ -15,10 +15,9 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest]
         pgver: [15, 16, 17, 18]
 
-    runs-on: ${{ matrix.os }}
+    runs-on: ubuntu-24.04-x64
 
     steps:
       - name: Checkout spock

include/spock_group.h

@@ -161,5 +161,6 @@ extern TimestampTz apply_worker_get_prev_remote_ts(void);
 extern void spock_group_resource_dump(void);
 extern void spock_checkpoint_hook(XLogRecPtr checkPointRedo, int flags);
 extern void spock_group_progress_update_list(List *lst);
+extern void spock_group_progress_force_set_list(List *lst);
 
 #endif							/* SPOCK_GROUP_H */

samples/Z0DAN/zodan.sql

@@ -1346,9 +1346,79 @@ BEGIN
                 RAISE EXCEPTION '    ✗ %', rpad('Creating replication slot ' || slot_name || ' on node ' || rec.node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
         END;
 
+        -- Wait for the source node to have committed all changes from this
+        -- "other" node up to L_slot, ensuring P_snap >= L_slot when Phase 5
+        -- takes the snapshot (prevents data loss in the [P_snap, L_slot) gap).
+        BEGIN
+            DECLARE
+                src_progress_lsn  pg_lsn;
+                wait_iters        integer := 0;
+                max_wait_iters    integer := 2400;  -- 20 minutes (0.5s per iter)
+                progress_sql      text;
+            BEGIN
+                progress_sql := format(
+                    'SELECT p.remote_commit_lsn '
+                    'FROM spock.progress p '
+                    'JOIN spock.node n ON n.node_id = p.remote_node_id '
+                    'WHERE p.node_id = (SELECT node_id FROM spock.node_info()) '
+                    '  AND n.node_name = %L',
+                    rec.node_name);
+
+                RAISE NOTICE '    - Waiting for source node % to commit % changes up to slot LSN %...',
+                             src_node_name, rec.node_name, _commit_lsn;
+
+                LOOP
+                    SELECT * FROM dblink(src_dsn, progress_sql)
+                        AS t(lsn pg_lsn) INTO src_progress_lsn;
+
+                    EXIT WHEN src_progress_lsn IS NOT NULL
+                              AND src_progress_lsn >= _commit_lsn;
+
+                    IF wait_iters >= max_wait_iters THEN
+                        RAISE WARNING '    Timeout waiting for source node commit catchup (last seen: %)', src_progress_lsn;
+                        EXIT;
+                    END IF;
+
+                    PERFORM pg_sleep(0.5);
+                    wait_iters := wait_iters + 1;
+                END LOOP;
+
+                RAISE NOTICE '    OK: %', rpad(
+                    'Source node ' || src_node_name || ' committed ' || rec.node_name
+                    || ' changes up to ' || COALESCE(src_progress_lsn::text, 'unknown')
+                    || ' (needed >= ' || _commit_lsn || ')', 120, ' ');
+            END;
+        EXCEPTION
+            WHEN OTHERS THEN
+                RAISE WARNING '    Could not verify source commit catchup for %: %', rec.node_name, SQLERRM;
+        END;
+
         -- Create disabled subscription on new node from "other" node
         BEGIN
 			sub_name := 'sub_' || rec.node_name || '_' || new_node_name;
+            -- Drop stale replication origin from a previous add_node cycle
+            -- so create_sub starts fresh at 0/0 (avoids stale-LSN data loss).
+            BEGIN
+                PERFORM dblink_exec(
+                    new_node_dsn,
+                    format($dsql$
+                        DO $x$
+                        BEGIN
+                            IF EXISTS (SELECT 1 FROM pg_replication_origin
+                                       WHERE roname = %L) THEN
+                                PERFORM pg_replication_origin_drop(%L);
+                            END IF;
+                        END $x$
+                        $dsql$,
+                        slot_name, slot_name)
+                );
+                RAISE NOTICE '    OK: Dropped stale origin % on new node (if existed)',
+                    slot_name;
+            EXCEPTION
+                WHEN OTHERS THEN
+                    RAISE WARNING '    Could not drop stale origin % on new node: %',
+                        slot_name, SQLERRM;
+            END;
             CALL spock.create_sub(
                 new_node_dsn,                                 -- Create on new node
                 sub_name, 									  -- sub_<new_node>_<other_node>
@@ -1362,12 +1432,12 @@ BEGIN
                 false,                                        -- force_text_transfer
                 verb                                          -- verbose
             );
-            RAISE NOTICE '    ✓ %', rpad('Creating initial subscription ' || sub_name || ' on node ' || rec.node_name, 120, ' ');
+            RAISE NOTICE '    ✓ %', rpad('Creating initial subscription ' || sub_name || ' on new node ' || new_node_name || ' (provider: ' || rec.node_name || ')', 120, ' ');
             PERFORM pg_sleep(5);
             subscription_count := subscription_count + 1;
         EXCEPTION
             WHEN OTHERS THEN
-                RAISE NOTICE '    ✗ %', rpad('Creating initial subscription ' || sub_name || ' on node ' || rec.node_name || ' (error: ' || SQLERRM || ')', 120, ' ');
+                RAISE NOTICE '    ✗ %', rpad('Creating initial subscription ' || sub_name || ' on new node ' || new_node_name || ' (provider: ' || rec.node_name || ') (error: ' || SQLERRM || ')', 120, ' ');
         END;
     END LOOP;
 
@@ -1429,7 +1499,7 @@ BEGIN
 
                         -- Wait for this sync event on the new node where the subscription exists
                         PERFORM * FROM dblink(new_node_dsn,
-                            format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s)',
+                            format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true)',
                                    src_node_name, sync_lsn, timeout_ms)) AS t(result text);
 
                         IF verb THEN
@@ -1450,7 +1520,8 @@ BEGIN
             CALL spock.verify_subscription_replicating(
                 new_node_dsn,
                 'sub_' || src_node_name || '_' || new_node_name,
-                verb
+                verb,
+				1200
             );
 
             RAISE NOTICE '    ✓ %', rpad('Enabling subscription ' || sub_name || '...', 120, ' ');
@@ -1497,7 +1568,7 @@ BEGIN
 
                         -- Wait for this sync event on the new node where the subscription exists
                         PERFORM * FROM dblink(new_node_dsn,
-                            format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s)',
+                            format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true)',
                                    rec.node_name, sync_lsn, timeout_ms)) AS t(result text);
 
                         IF verb THEN
@@ -1512,7 +1583,8 @@ BEGIN
                 CALL spock.verify_subscription_replicating(
                     new_node_dsn,
                     'sub_'|| rec.node_name || '_' || new_node_name,
-                    verb
+                    verb,
+					1200
                 );
 
                 RAISE NOTICE '    ✓ %', rpad('Enabling subscription ' || sub_name || '...', 120, ' ');
@@ -1694,7 +1766,7 @@ BEGIN
 
         -- Wait for sync event on source node
         BEGIN
-            remotesql := format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s);',
+            remotesql := format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true);',
                                rec.node_name, sync_lsn, timeout_ms);
             IF verb THEN
                 RAISE NOTICE '    Remote SQL for waiting sync event: %', remotesql;
@@ -1726,12 +1798,101 @@ DECLARE
     slot_name text;
     dbname text;
     remotesql text;
+    src_rec RECORD;
+    src_commit_lsn pg_lsn;
+    src_slot_name text;
+    src_dbname text;
+    current_lsn pg_lsn;
+    target_lsn pg_lsn;
+    v_sub_name name;
 BEGIN
     RAISE NOTICE 'Phase 7: Checking commit timestamp and advancing replication slot';
 
+    -- Wait for src->new COPY to complete so P_snap is written to spock.progress.
+    v_sub_name := ('sub_' || src_node_name || '_' || new_node_name)::name;
+    RAISE NOTICE '    - Waiting for subscription % to reach READY...', v_sub_name;
+    BEGIN
+        PERFORM spock.sub_wait_for_sync(v_sub_name);
+        RAISE NOTICE '    - Subscription % is READY', v_sub_name;
+    EXCEPTION
+        WHEN OTHERS THEN
+            RAISE WARNING '    - sub_wait_for_sync(%) failed: %; proceeding anyway', v_sub_name, SQLERRM;
+    END;
+
+    -- Check src->new slot; only advance if it is NOT active (defensive).
+    BEGIN
+        RAISE NOTICE '    - Checking source-to-new subscription slot...';
+        
+        -- Get source node info and extract dbname
+        FOR src_rec IN SELECT * FROM temp_spock_nodes WHERE node_name = src_node_name LOOP
+            SELECT spock.extract_dbname_from_dsn(src_rec.dsn) INTO src_dbname;
+            IF src_dbname IS NOT NULL THEN
+                src_dbname := TRIM(BOTH '''' FROM src_dbname);
+            END IF;
+            IF src_dbname IS NULL THEN 
+                src_dbname := 'pgedge'; 
+            END IF;
+
+            -- Generate slot name: spk_<dbname>_<src>_sub_<src>_<new>
+            src_slot_name := spock.spock_gen_slot_name(
+                src_dbname, src_node_name, 
+                'sub_' || src_node_name || '_' || new_node_name);
+            
+            RAISE NOTICE '    Looking for slot % on source node', src_slot_name;
+
+            -- Check if slot exists on source node and whether it is active
+            DECLARE
+                v_slot_active boolean;
+            BEGIN
+                remotesql := format(
+                    'SELECT restart_lsn, active FROM pg_replication_slots WHERE slot_name = %L',
+                    src_slot_name);
+                SELECT * FROM dblink(src_dsn, remotesql)
+                    AS t(lsn pg_lsn, active boolean)
+                    INTO current_lsn, v_slot_active;
+
+                IF current_lsn IS NULL THEN
+                    RAISE NOTICE '    Slot % not found on source node', src_slot_name;
+                ELSIF v_slot_active THEN
+                    -- Subscription is running; slot and origin managed by the apply worker
+                    RAISE NOTICE '    Slot % is active (subscription running) — no advance needed', src_slot_name;
+                ELSE
+                    -- Slot exists but is not active (unusual).  Advance defensively.
+                    RAISE NOTICE '    Slot % found at LSN % (inactive)', src_slot_name, current_lsn;
+
+                    SELECT p.remote_commit_lsn INTO target_lsn
+                    FROM spock.progress p
+                    JOIN spock.node n ON n.node_id = p.remote_node_id
+                    WHERE n.node_name = src_node_name;
+
+                    IF target_lsn IS NOT NULL AND target_lsn > current_lsn THEN
+                        RAISE NOTICE '    Snapshot LSN for %: %', src_node_name, target_lsn;
+                        remotesql := format('SELECT pg_replication_slot_advance(%L, %L::pg_lsn)', src_slot_name, target_lsn);
+                        PERFORM * FROM dblink(src_dsn, remotesql) AS t(result text);
+                        RAISE NOTICE '    OK: Advanced slot % on source node from % to %', src_slot_name, current_lsn, target_lsn;
+
+                        IF NOT EXISTS (
+                            SELECT 1 FROM pg_replication_origin WHERE roname = src_slot_name
+                        ) THEN
+                            RAISE WARNING '    Origin % not found on new node; creating it now', src_slot_name;
+                            PERFORM pg_replication_origin_create(src_slot_name);
+                        END IF;
+                        PERFORM pg_replication_origin_advance(src_slot_name, target_lsn);
+                        RAISE NOTICE '    OK: Advanced replication origin % on new node to %', src_slot_name, target_lsn;
+                    ELSE
+                        RAISE NOTICE '    Slot % already at or beyond snapshot LSN', src_slot_name;
+                    END IF;
+                END IF;
+            END;
+        END LOOP;
+    EXCEPTION
+        WHEN OTHERS THEN
+            RAISE WARNING 'Could not check source-to-new slot: %', SQLERRM;
+    END;
+
     -- Check if this is a 2-node scenario (only source and new node)
     IF (SELECT count(*) FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name) = 0 THEN
-        RAISE NOTICE '    - No other nodes exist, skipping commit timestamp check';
+        RAISE NOTICE '    - No other nodes exist, skipping additional commitment checks';
         RETURN;
     END IF;
 
@@ -1793,9 +1954,21 @@ BEGIN
                     CONTINUE;
                 END IF;
 
-                target_lsn := commit_lsn;
+                -- Advance the slot to P_snap: the last commit from this node
+                -- that N1 had applied at snapshot time (stored in N3's spock.progress).
+                SELECT p.remote_commit_lsn INTO target_lsn
+                FROM spock.progress p
+                JOIN spock.node n ON n.node_id = p.remote_node_id
+                WHERE n.node_name = rec.node_name;
+
+                IF target_lsn IS NULL THEN
+                    RAISE NOTICE '    WARNING: No spock.progress entry for %, falling back to pg_current_wal_lsn()', rec.node_name;
+                    remotesql := 'SELECT pg_current_wal_lsn()';
+                    SELECT * FROM dblink(rec.dsn, remotesql) AS t(lsn pg_lsn) INTO target_lsn;
+                END IF;
+
                 IF target_lsn IS NULL OR target_lsn <= current_lsn THEN
-                    RAISE NOTICE '    - Slot % already at or beyond target LSN (current: %, target: %)', slot_name, current_lsn, target_lsn;
+                    RAISE NOTICE '    - Slot % already at or beyond P_snap LSN (current: %, target: %)', slot_name, current_lsn, target_lsn;
                     CONTINUE;
                 END IF;
 
@@ -1807,6 +1980,17 @@ BEGIN
 
                 PERFORM * FROM dblink(rec.dsn, remotesql) AS t(result text);
                 RAISE NOTICE '    OK: %', rpad('Advanced slot ' || slot_name || ' from ' || current_lsn || ' to ' || target_lsn, 120, ' ');
+
+                -- Advance the replication origin on the new node (subscriber side)
+                -- directly, not via dblink; the origin is local to the new node.
+                IF NOT EXISTS (
+                    SELECT 1 FROM pg_replication_origin WHERE roname = slot_name
+                ) THEN
+                    RAISE WARNING '    Origin % not found on new node; creating it now (was it created in Phase 3?)', slot_name;
+                    PERFORM pg_replication_origin_create(slot_name);
+                END IF;
+                PERFORM pg_replication_origin_advance(slot_name, target_lsn);
+                RAISE NOTICE '    OK: %', rpad('Advanced replication origin ' || slot_name || ' on new node to ' || target_lsn, 120, ' ');
             END;
         EXCEPTION
             WHEN OTHERS THEN
@@ -1851,7 +2035,7 @@ BEGIN
 
     -- Wait for sync event on new node
     BEGIN
-        remotesql := format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s);', src_node_name, sync_lsn, timeout_ms);
+        remotesql := format('CALL spock.wait_for_sync_event(true, %L, %L::pg_lsn, %s, true);', src_node_name, sync_lsn, timeout_ms);
         IF verb THEN
             RAISE NOTICE '    Remote SQL for wait_for_sync_event on new node %: %', new_node_name, remotesql;
         END IF;
@@ -1877,7 +2061,7 @@ DECLARE
     sub_rec  RECORD;
     rec RECORD;
     wait_count integer := 0;
-    max_wait_count integer := 300; -- Wait up to 300 seconds
+    max_wait_count integer := 1200; -- Wait up to 1200 seconds
 BEGIN
     -- Let remote subscriptions update their subscription's state.
     COMMIT;