Save persistent winner decision locally and fix highlighted problem with recovery from disabled node

Author: kelvich

multimaster--1.0.sql

@@ -114,6 +114,8 @@ LANGUAGE C;
 
 CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema name, rel_name name, primary key(rel_schema, rel_name));
 
+CREATE TABLE mtm.referee_decision(key text primary key not null, node_id int);
+
 CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS
 $$
 DECLARE

multimaster.c

@@ -256,6 +256,7 @@ bool  MtmPreserveCommitOrder;
 bool  MtmVolksWagenMode; /* Pretend to be normal postgres. This means skip some NOTICE's and use local sequences */
 bool  MtmMajorNode;
 char* MtmRefereeConnStr;
+bool  MtmEnforceLocalTx;
 
 static char* MtmConnStrs;
 static char* MtmRemoteFunctionsList;
@@ -1110,7 +1111,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	ts->procno = MyProc->pgprocno;
 	ts->votingCompleted = false;
 	ts->participantsMask = (((nodemask_t)1 << Mtm->nAllNodes) - 1) & ~Mtm->disabledNodeMask & ~((nodemask_t)1 << (MtmNodeId-1));
-    ts->isLocal = x->isReplicated || !x->containsDML || (ts->participantsMask == 0);
+	ts->isLocal = x->isReplicated || !x->containsDML || (ts->participantsMask == 0) || MtmEnforceLocalTx;
 	ts->nConfigChanges = Mtm->nConfigChanges;
 	ts->votedMask = 0;
 	ts->nSubxids = xactGetCommittedChildren(&subxids);
@@ -3713,8 +3714,14 @@ MtmReplicationTxnFilterHook(struct PGLogicalTxnFilterArgs* args)
 	 * unless we are performing recovery of disabled node
 	 * (in this case all transactions should be sent)
 	 */
-	bool res = Mtm->status != MTM_RECOVERY
-		&& (args->origin_id == InvalidRepOriginId
+	/*
+	 * I removed (Mtm->status != MTM_RECOVERY) here since in major
+	 * mode we need to recover from offline node too. Also it seems
+	 * that with amount of nodes >= 3 we also need that. --sk
+	 *
+	 * On a first look this works fine.
+	 */
+	bool res = (args->origin_id == InvalidRepOriginId
 			|| MtmIsRecoveredNode(MtmReplicationNodeId));
 	if (!res) {
 		MTM_LOG2("Filter transaction with origin_id=%d", args->origin_id);

multimaster.h

@@ -392,6 +392,7 @@ extern MtmConnectionInfo* MtmConnections;
 extern bool MtmMajorNode;
 extern bool MtmBackgroundWorker;
 extern char* MtmRefereeConnStr;
+extern bool  MtmEnforceLocalTx;
 
 
 extern void  MtmArbiterInitialize(void);

state.c

@@ -2,6 +2,10 @@
 #include "miscadmin.h" /* PostmasterPid */
 #include "multimaster.h"
 #include "state.h"
+#include "executor/spi.h"
+#include "utils/snapmgr.h"
+#include "nodes/makefuncs.h"
+#include "catalog/namespace.h"
 
 char const* const MtmNeighborEventMnem[] =
 {
@@ -27,6 +31,9 @@ char const* const MtmEventMnem[] =
 
 static int  MtmRefereeGetWinner(void);
 static bool MtmRefereeClearWinner(void);
+static int  MtmRefereeReadSaved(void);
+
+static bool mtm_state_initialized;
 
 // XXXX: allocate in context and clean it
 static char *
@@ -434,6 +441,13 @@ MtmRefreshClusterStatus()
 	{
 		int winner_node_id = MtmRefereeGetWinner();
 
+		/* We also can have old value. Do that only from single mtm-monitor process */
+		if (winner_node_id <= 0 && !mtm_state_initialized)
+		{
+			winner_node_id = MtmRefereeReadSaved();
+			mtm_state_initialized = true;
+		}
+
 		if (winner_node_id > 0)
 		{
 			Mtm->refereeWinnerId = winner_node_id;
@@ -555,13 +569,70 @@ MtmRefreshClusterStatus()
 	MtmUnlock();
 }
 
+/*
+ * Referee caches decision in mtm.referee_decision
+ */
+static bool
+MtmRefereeHasLocalTable()
+{
+	RangeVar   *rv;
+	Oid			rel_oid;
+
+	StartTransactionCommand();
+	rv = makeRangeVar(MULTIMASTER_SCHEMA_NAME, "referee_decision", -1);
+	rel_oid = RangeVarGetRelid(rv, NoLock, true);
+	CommitTransactionCommand();
+
+	return OidIsValid(rel_oid);
+}
+
+static int
+MtmRefereeReadSaved(void)
+{
+	int winner = -1;
+	int rc;
+
+	if (!MtmRefereeHasLocalTable())
+		return -1;
+
+	/* Save result locally */
+	StartTransactionCommand();
+	SPI_connect();
+	PushActiveSnapshot(GetTransactionSnapshot());
+	rc = SPI_execute("select node_id from mtm.referee_decision where key = 'winner';", true, 0);
+	if (rc != SPI_OK_SELECT)
+	{
+		MTM_ELOG(WARNING, "Failed to load referee decision");
+	}
+	else if (SPI_processed > 0)
+	{
+		bool isnull;
+		winner = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1, &isnull));
+		Assert(SPI_processed == 1);
+		Assert(!isnull);
+	}
+	else
+	{
+		/* no saved decision found */
+		Assert(SPI_processed == 0);
+	}
+	SPI_finish();
+	PopActiveSnapshot();
+	CommitTransactionCommand();
+
+	MTM_LOG1("Read saved referee decision, winner=%d.", winner);
+	return winner;
+}
+
 static int
 MtmRefereeGetWinner(void)
 {
 	PGconn* conn;
 	PGresult *res;
 	char sql[128];
 	int  winner_node_id;
+	int  old_winner = -1;
+	int  rc;
 
 	conn = PQconnectdb_safe(MtmRefereeConnStr, 5);
 	if (PQstatus(conn) != CONNECTION_OK)
@@ -599,6 +670,48 @@ MtmRefereeGetWinner(void)
 	/* Ok, we finally got it! */
 	PQclear(res);
 	PQfinish(conn);
+
+	/* Save result locally */
+	if (MtmRefereeHasLocalTable())
+	{
+		MtmEnforceLocalTx = true;
+		StartTransactionCommand();
+		SPI_connect();
+		PushActiveSnapshot(GetTransactionSnapshot());
+		/* Check old value if any */
+		rc = SPI_execute("select node_id from mtm.referee_decision where key = 'winner';", true, 0);
+		if (rc != SPI_OK_SELECT)
+		{
+			MTM_ELOG(WARNING, "Failed to load previous referee decision");
+		}
+		else if (SPI_processed > 0)
+		{
+			bool isnull;
+			old_winner = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1, &isnull));
+			Assert(SPI_processed == 1);
+			Assert(!isnull);
+		}
+		else
+		{
+			/* no saved decision found */
+			Assert(SPI_processed == 0);
+		}
+		/* Update actual key */
+		sprintf(sql,
+			"insert into mtm.referee_decision values ('winner', %d) on conflict(key) do nothing;",
+			winner_node_id);
+		rc = SPI_execute(sql, false, 0);
+		SPI_finish();
+		if (rc < 0)
+			MTM_ELOG(WARNING, "Failed to save referee decision, but proceeding anyway");
+		PopActiveSnapshot();
+		CommitTransactionCommand();
+		MtmEnforceLocalTx = false;
+
+		if (old_winner > 0 && old_winner != winner_node_id)
+			MTM_LOG1("WARNING Overriding old referee decision (%d) with new one (%d)", old_winner, winner_node_id);
+	}
+
 	MTM_LOG1("Got referee response, winner node_id=%d.", winner_node_id);
 	return winner_node_id;
 }
@@ -609,6 +722,32 @@ MtmRefereeClearWinner(void)
 	PGconn* conn;
 	PGresult *res;
 	char *response;
+	int rc;
+
+	/*
+	 * Delete result locally first.
+	 *
+	 * If we delete decision from referee but fail to delete local cached
+	 * that will be pretty bad -- on the next reboot we can read
+	 * stale referee decision and on next failure end up with two masters.
+	 * So just delete local cache first.
+	 */
+	if (MtmRefereeHasLocalTable())
+	{
+		StartTransactionCommand();
+		SPI_connect();
+		PushActiveSnapshot(GetTransactionSnapshot());
+		rc = SPI_execute("delete from mtm.referee_decision where key = 'winner'", false, 0);
+		SPI_finish();
+		PopActiveSnapshot();
+		CommitTransactionCommand();
+		if (rc < 0)
+		{
+			MTM_ELOG(WARNING, "Failed to save referee decision, proceeding anyway");
+			return false;
+		}
+	}
+
 
 	conn = PQconnectdb_safe(MtmRefereeConnStr, 5);
 	if (PQstatus(conn) != CONNECTION_OK)

tests2/test_referee.py

@@ -102,6 +102,76 @@ def test_double_failure_referee(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
+    def test_saved_referee_decision(self):
+        print('### test_saved_referee_decision ###')
+        docker_api = docker.from_env()
+
+        print('#### down on(winner) || on')
+        print('###########################')
+        aggs_failure, aggs = self.performFailure(StopNode('node1'))
+
+        self.assertNoCommits(aggs_failure[:1])
+        self.assertCommits(aggs_failure[1:])
+        self.assertIsolation(aggs_failure)
+
+        self.assertNoCommits(aggs[:1])
+        self.assertCommits(aggs[1:])
+        self.assertIsolation(aggs)
+
+        print('#### down restart(winner) || down')
+        print('###########################')
+        docker_api.containers.get('referee').stop()
+        aggs_failure, aggs = self.performFailure(RestartNode('node2'), node_wait_for_commit=1)
+
+        # without saved decision node2 will be endlessy disabled here
+
+        self.assertNoCommits(aggs_failure)
+        self.assertIsolation(aggs_failure)
+
+        self.assertNoCommits(aggs[:1])
+        self.assertCommits(aggs[1:])
+        self.assertIsolation(aggs)
+
+        print('#### up down(winner) || down')
+        print('###########################')
+        docker_api.containers.get('node2').stop()
+        docker_api.containers.get('node1').start()
+        aggs_failure, aggs = self.performFailure(NoFailure())
+
+        self.assertNoCommits(aggs_failure)
+        self.assertIsolation(aggs_failure)
+        self.assertNoCommits(aggs)
+        self.assertIsolation(aggs)
+
+        print('#### up down(winner) || up')
+        print('###########################')
+        docker_api.containers.get('referee').start()
+        aggs_failure, aggs = self.performFailure(NoFailure())
+
+        self.assertNoCommits(aggs_failure)
+        self.assertIsolation(aggs_failure)
+        self.assertNoCommits(aggs)
+        self.assertIsolation(aggs)
+
+        print('#### up up(winner) || up')
+        print('###########################')
+        docker_api.containers.get('node2').start()
+        self.awaitCommit(0)
+
+        # give it time to clean old decision
+        time.sleep(5)
+
+        print('#### check that decision is cleaned')
+        print('###########################')
+        con = psycopg2.connect("dbname=regression user=postgres host=127.0.0.1 port=15435")
+        con.autocommit = True
+        cur = con.cursor()
+        cur.execute("select node_id into winner_id from referee.decision where key = 'winner'")
+        decisions_count = cur.rowcount
+        cur.close()
+        con.close()
+
+        self.assertEqual(decisions_count, 0)
 
     def test_winner_restart(self):
         print('### test_winner_restart ###')