Log Message:
-----------
Let failoverSet_int() and moveSet_int() communicate the exact ev_seqno
of the FAILOVER_SET or MOVE_SET as 4th parameter of ACCEPT_SET, named
wait_seqno. The ACCEPT_SET processing in slon then waits for exactly that
event to be processed before continuing with the ACCEPT_SET. This avoids
a faulty interpretation of an older MOVE_SET to let ACCEPT_SET continue
too early.

Jan

Modified Files:
--------------
    slony1-engine/src/backend:
        slony1_funcs.sql (r1.96 -> r1.97)
    slony1-engine/src/slon:
        remote_worker.c (r1.122 -> r1.123)

-------------- next part --------------
Index: slony1_funcs.sql
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/backend/slony1_funcs.sql,v
retrieving revision 1.96
retrieving revision 1.97
diff -Lsrc/backend/slony1_funcs.sql -Lsrc/backend/slony1_funcs.sql -u -w -r1.96 -r1.97
--- src/backend/slony1_funcs.sql
+++ src/backend/slony1_funcs.sql
@@ -1338,7 +1338,7 @@
 	notify "_ at CLUSTERNAME at _Restart";
 
 	perform @NAMESPACE at .failoverSet_int(p_failed_node,
-			p_backup_node, p_set_id);
+			p_backup_node, p_set_id, p_ev_seqfake);
 
 	return p_ev_seqfake;
 end;
@@ -1351,17 +1351,18 @@
 fake the FAILOVER_SET event.';
 
 -- ----------------------------------------------------------------------
--- FUNCTION failoverSet_int (failed_node, backup_node, set_id)
+-- FUNCTION failoverSet_int (failed_node, backup_node, set_id, wait_seqno)
 --
 --	Finish failover for one set.
 -- ----------------------------------------------------------------------
-create or replace function @NAMESPACE at .failoverSet_int (int4, int4, int4)
+create or replace function @NAMESPACE at .failoverSet_int (int4, int4, int4, int8)
 returns int4
 as '
 declare
 	p_failed_node		alias for $1;
 	p_backup_node		alias for $2;
 	p_set_id			alias for $3;
+	p_wait_seqno		alias for $4;
 	v_row				record;
 	v_last_sync			int8;
 begin
@@ -1399,12 +1400,13 @@
 		insert into @NAMESPACE at .sl_event
 				(ev_origin, ev_seqno, ev_timestamp,
 				ev_minxid, ev_maxxid, ev_xip,
-				ev_type, ev_data1, ev_data2, ev_data3)
+				ev_type, ev_data1, ev_data2, ev_data3, ev_data4)
 				values
 				(p_backup_node, "pg_catalog".nextval(''@NAMESPACE at .sl_event_seq''), CURRENT_TIMESTAMP,
 				''0'', ''0'', '''',
 				''ACCEPT_SET'', p_set_id::text,
-				p_failed_node::text, p_backup_node::text);
+				p_failed_node::text, p_backup_node::text,
+				p_wait_seqno::text);
 	else
 		delete from @NAMESPACE at .sl_subscribe
 				where sub_set = p_set_id
@@ -1455,8 +1457,8 @@
 	return p_failed_node;
 end;
 ' language plpgsql;
-comment on function @NAMESPACE at .failoverSet_int (int4, int4, int4) is
-'FUNCTION failoverSet_int (failed_node, backup_node, set_id)
+comment on function @NAMESPACE at .failoverSet_int (int4, int4, int4, int8) is
+'FUNCTION failoverSet_int (failed_node, backup_node, set_id, wait_seqno)
 
 Finish failover for one set.';
 
@@ -2144,7 +2146,7 @@
 	-- Reconfigure everything
 	-- ----
 	perform @NAMESPACE at .moveSet_int(p_set_id, v_local_node_id,
-			p_new_origin);
+			p_new_origin, 0);
 
 	perform @NAMESPACE at .RebuildListenEntries();
 
@@ -2173,17 +2175,18 @@
 Generate MOVE_SET event to request that the origin for set set_id be moved to node new_origin';
 
 -- ----------------------------------------------------------------------
--- FUNCTION moveSet_int (set_id, old_origin, new_origin)
+-- FUNCTION moveSet_int (set_id, old_origin, new_origin, wait_seqno)
 --
 --	Process the MOVE_SET event.
 -- ----------------------------------------------------------------------
-create or replace function @NAMESPACE at .moveSet_int (int4, int4, int4)
+create or replace function @NAMESPACE at .moveSet_int (int4, int4, int4, int8)
 returns int4
 as '
 declare
 	p_set_id			alias for $1;
 	p_old_origin		alias for $2;
 	p_new_origin		alias for $3;
+	p_wait_seqno		alias for $4;
 	v_local_node_id		int4;
 	v_tab_row			record;
 	v_sub_row			record;
@@ -2223,7 +2226,7 @@
 		-- finalize the setsync status.
 		perform @NAMESPACE at .createEvent(''_ at CLUSTERNAME@'', ''SYNC'', NULL);
 		perform @NAMESPACE at .createEvent(''_ at CLUSTERNAME@'', ''ACCEPT_SET'', 
-			p_set_id, p_old_origin, p_new_origin);
+			p_set_id, p_old_origin, p_new_origin, p_wait_seqno);
 	end if;
 
 	-- ----
@@ -2365,8 +2368,8 @@
 	return p_set_id;
 end;
 ' language plpgsql;
-comment on function @NAMESPACE at .moveSet_int(int4, int4, int4) is 
-'moveSet(set_id, old_origin, new_origin)
+comment on function @NAMESPACE at .moveSet_int(int4, int4, int4, int8) is 
+'moveSet(set_id, old_origin, new_origin, wait_seqno)
 
 Process MOVE_SET event to request that the origin for set set_id be
 moved from old_origin to node new_origin';
Index: remote_worker.c
===================================================================
RCS file: /usr/local/cvsroot/slony1/slony1-engine/src/slon/remote_worker.c,v
retrieving revision 1.122
retrieving revision 1.123
diff -Lsrc/slon/remote_worker.c -Lsrc/slon/remote_worker.c -u -w -r1.122 -r1.123
--- src/slon/remote_worker.c
+++ src/slon/remote_worker.c
@@ -1119,8 +1119,8 @@
 			{
 				int			set_id,
 							old_origin,
-							new_origin,
-							event_no;
+							new_origin;
+				char	   *wait_seqno;
 				PGresult   *res;
 
 				slon_log(SLON_DEBUG2, "start processing ACCEPT_SET\n");
@@ -1130,8 +1130,8 @@
 				slon_log(SLON_DEBUG2, "ACCEPT: old origin=%d\n", old_origin);
 				new_origin = (int)strtol(event->ev_data3, NULL, 10);
 				slon_log(SLON_DEBUG2, "ACCEPT: new origin=%d\n", new_origin);
-				event_no = event->ev_seqno;
-				slon_log(SLON_DEBUG2, "ACCEPT: move set seq=%d\n", event_no);
+				wait_seqno = event->ev_data4;
+				slon_log(SLON_DEBUG2, "ACCEPT: move set seq=%s\n", wait_seqno);
 
 				slon_log(SLON_DEBUG2, "got parms ACCEPT_SET\n");
 
@@ -1151,20 +1151,22 @@
 								 "select 1 from %s.sl_event "
 								 "where "
 								 "     (ev_origin = %d and "
+								 "      ev_seqno = %s and "
 								 "      ev_type = 'MOVE_SET' and "
 								 "      ev_data1 = '%d' and "
 								 "      ev_data2 = '%d' and "
 								 "      ev_data3 = '%d') "
 								 "or "
 								 "     (ev_origin = %d and "
+								 "      ev_seqno = %s and "
 								 "      ev_type = 'FAILOVER_SET' and "
 								 "      ev_data1 = '%d' and "
 								 "      ev_data2 = '%d' and "
 								 "      ev_data3 = '%d'); ",
 
 								 rtcfg_namespace,
-								 old_origin, set_id, old_origin, new_origin,
-								 old_origin, old_origin, new_origin, set_id);
+								 old_origin, wait_seqno, set_id, old_origin, new_origin,
+								 old_origin, wait_seqno, old_origin, new_origin, set_id);
 
 					res = PQexec(local_dbconn, dstring_data(&query2));
 					while (PQntuples(res) == 0)
@@ -1184,13 +1186,15 @@
 					 */
 					slon_appendquery(&query1,
 								"update %s.sl_setsync "
-								"    set ssy_seqno = '%s', "
+								"    set ssy_origin = %d, "
+								"        ssy_seqno = '%s', "
 								"        ssy_minxid = '%s', "
 								"        ssy_maxxid = '%s', "
 								"        ssy_xip = '%q', "
 								"        ssy_action_list = '' "
 								"    where ssy_setid = %d; ",
 								rtcfg_namespace,
+								new_origin,
 								seqbuf,
 								event->ev_minxid_c,
 								event->ev_maxxid_c,
@@ -1233,9 +1237,9 @@
 				 */
 
 				slon_appendquery(&query1,
-								 "select %s.moveSet_int(%d, %d, %d); ",
+								 "select %s.moveSet_int(%d, %d, %d, %s); ",
 								 rtcfg_namespace,
-								 set_id, old_origin, new_origin);
+								 set_id, old_origin, new_origin, seqbuf);
 				if (query_execute(node, local_dbconn, &query1) < 0)
 					slon_retry();
 
@@ -1277,9 +1281,9 @@
 				rtcfg_storeSet(set_id, backup_node, NULL);
 
 				slon_appendquery(&query1,
-								 "select %s.failoverSet_int(%d, %d, %d); ",
+								 "select %s.failoverSet_int(%d, %d, %d, %s); ",
 								 rtcfg_namespace,
-								 failed_node, backup_node, set_id);
+								 failed_node, backup_node, set_id, seqbuf);
 
 				if (archive_dir)
 				{