Fix incorrect subscription name in Phase 9 of ZODAN add_node (SPOC-503)

create_sub_on_new_node_to_src_node generated sub_{subscriber}_{provider}
instead of the expected sub_{provider}_{subscriber} convention, causing
remove-node to fail when looking up subscriptions by their expected name.

Introduce spock.gen_sub_name(provider_node, subscriber_node) helper and
replace all inline 'sub_' || concatenations to prevent future mismatches.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: mason-sharp

samples/Z0DAN/zodan.sql

@@ -28,6 +28,22 @@
 
 -- ============================================================================
 
+-- ============================================================================
+-- Function: gen_sub_name
+-- Purpose : Generate a subscription name following the sub_{provider}_{subscriber}
+--           convention (i.e. sub_{remote}_{local}).
+-- Arguments:
+--   provider_node   - The node supplying data (remote/provider).
+--   subscriber_node - The node receiving data (local/subscriber).
+-- Returns  : text, e.g. 'sub_n1_n2' where n1 is the provider and n2 is the subscriber.
+-- ============================================================================
+CREATE OR REPLACE FUNCTION spock.gen_sub_name(
+    provider_node text,
+    subscriber_node text
+) RETURNS text LANGUAGE sql IMMUTABLE AS $$
+    SELECT 'sub_' || provider_node || '_' || subscriber_node;
+$$;
+
 -- ============================================================================
 -- Procedure: check_spock_version_compatibility
 -- Purpose: Verify all nodes have the same Spock version before adding a node
@@ -1217,7 +1233,7 @@ BEGIN
 
             slot_name := spock.spock_gen_slot_name(
 							dbname, rec.node_name,
-							'sub_' || rec.node_name || '_' || new_node_name);
+							spock.gen_sub_name(rec.node_name, new_node_name));
 
             CALL spock.create_replication_slot(
                 rec.dsn,
@@ -1312,7 +1328,7 @@ BEGIN
 
 			slot_name := spock.spock_gen_slot_name(
 							dbname, rec.node_name,
-							'sub_' || rec.node_name || '_' || new_node_name);
+							spock.gen_sub_name(rec.node_name, new_node_name));
 
             remotesql := format('SELECT slot_name, lsn FROM pg_create_logical_replication_slot(%L, ''spock_output'');', slot_name);
             IF verb THEN
@@ -1415,7 +1431,7 @@ BEGIN
 
         -- Create disabled subscription on new node from "other" node
         BEGIN
-			sub_name := 'sub_' || rec.node_name || '_' || new_node_name;
+			sub_name := spock.gen_sub_name(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
@@ -1488,7 +1504,7 @@ BEGIN
     IF (SELECT count(*) FROM temp_spock_nodes WHERE node_name != src_node_name AND node_name != new_node_name) = 0 THEN
         -- 2-node scenario: enable the disabled subscription from source to new node
 
-		sub_name := 'sub_' || src_node_name || '_' || new_node_name;
+		sub_name := spock.gen_sub_name(src_node_name, new_node_name);
 
         BEGIN
             CALL spock.enable_sub( new_node_dsn, sub_name, verb, true);
@@ -1548,7 +1564,7 @@ BEGIN
             -- Verify subscription is replicating after enabling (2-node scenario)
             CALL spock.verify_subscription_replicating(
                 new_node_dsn,
-                'sub_' || src_node_name || '_' || new_node_name,
+                spock.gen_sub_name(src_node_name, new_node_name),
                 verb,
 				180
             );
@@ -1575,7 +1591,7 @@ BEGIN
                     CONTINUE;  -- Skip new node to avoid self-subscription
                 END IF;
 
-				sub_name := 'sub_'|| rec.node_name || '_' || new_node_name;
+				sub_name := spock.gen_sub_name(rec.node_name, new_node_name);
 
                 CALL spock.enable_sub(new_node_dsn, sub_name, verb, true);
 
@@ -1619,7 +1635,7 @@ BEGIN
                 -- Verify subscription is replicating after enabling
                 CALL spock.verify_subscription_replicating(
                     new_node_dsn,
-                    'sub_'|| rec.node_name || '_' || new_node_name,
+                    spock.gen_sub_name(rec.node_name, new_node_name),
                     verb,
 					180
                 );
@@ -1661,7 +1677,7 @@ BEGIN
 
     -- For each existing node (excluding new node), create subscription TO the new node
     FOR rec IN SELECT * FROM temp_spock_nodes WHERE node_name != new_node_name LOOP
-	    sub_name := 'sub_' || rec.node_name || '_' || new_node_name;
+	    sub_name := spock.gen_sub_name(new_node_name, rec.node_name);
         BEGIN
             CALL spock.create_sub(
                 rec.dsn,                                      -- Create on existing node
@@ -1704,7 +1720,7 @@ CREATE OR REPLACE PROCEDURE spock.create_new_to_source_subscription(
     verb boolean
 ) LANGUAGE plpgsql AS $$
 DECLARE
-    sub_name text := 'sub_' || new_node_name || '_' || src_node_name;
+    sub_name text := spock.gen_sub_name(new_node_name, src_node_name);
 BEGIN
     RAISE NOTICE 'Phase 10: Creating new to source node subscription';
 
@@ -1738,7 +1754,7 @@ CREATE OR REPLACE PROCEDURE spock.create_source_to_new_subscription(
     verb            boolean  -- Verbose flag
 ) LANGUAGE plpgsql AS $$
 DECLARE
-    sub_name text := 'sub_' || src_node_name || '_' || new_node_name;
+    sub_name text := spock.gen_sub_name(src_node_name, new_node_name);
 BEGIN
     RAISE NOTICE 'Phase 4: Creating source to new node subscription';
 
@@ -1857,7 +1873,7 @@ BEGIN
     RAISE NOTICE 'Phase 7: Checking commit timestamp and advancing replication slot';
 
     -- Wait for src->new COPY to complete so resume_lsn is written to spock.progress.
-    v_sub_name := ('sub_' || src_node_name || '_' || new_node_name)::name;
+    v_sub_name := spock.gen_sub_name(src_node_name, new_node_name)::name;
     RAISE NOTICE '    - Waiting for subscription % to reach READY...', v_sub_name;
     BEGIN
         -- Avoid rare hangs in C-level sub_wait_for_sync by using a bounded SQL loop.
@@ -1918,7 +1934,7 @@ BEGIN
             -- 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);
+                spock.gen_sub_name(src_node_name, new_node_name));
 
             RAISE NOTICE '    Looking for slot % on source node', src_slot_name;
 
@@ -2017,7 +2033,7 @@ BEGIN
             END IF;
             IF dbname IS NULL THEN dbname := 'pgedge'; END IF;
 
-			slot_name := spock.spock_gen_slot_name(dbname, rec.node_name, 'sub_' || rec.node_name || '_' || new_node_name);
+			slot_name := spock.spock_gen_slot_name(dbname, rec.node_name, spock.gen_sub_name(rec.node_name, new_node_name));
 
             -- First check if slot exists and get current LSN
             remotesql := format('SELECT restart_lsn FROM pg_replication_slots WHERE slot_name = %L', slot_name);