add few status log messages during recovery finish

Author: kelvich

src/include/logger.h

@@ -57,13 +57,15 @@ typedef enum MtmLogTag
 	ProtoTraceTx			= DEBUG1,
 	ProtoTraceMode			= LOG,
 	ProtoTraceMessage		= LOG,
+	ProtoTraceState			= LOG,
 
 	/* receiver */
 	MtmReceiverMode			= LOG,
 	MtmReceiverFilter		= LOG,
 	MtmApplyMessage			= LOG,
 	MtmApplyTrace			= DEBUG2,
 	MtmApplyError			= LOG,
+	MtmApplyBgwFinish		= LOG,
 
 	/* state */
 	MtmStateSwitch			= LOG,

src/multimaster.c

@@ -1857,6 +1857,8 @@ MtmAllApplyWorkersFinished()
 		ntasks = Mtm->nodes[i].pool.active + Mtm->nodes[i].pool.pending;
 		SpinLockRelease(&Mtm->nodes[i].pool.lock);
 
+		mtm_log(MtmApplyBgwFinish, "MtmAllApplyWorkersFinished %d tasks not finished", ntasks);
+
 		if (ntasks != 0)
 			false;
 	}

src/pglogical_output.c

@@ -479,9 +479,11 @@ pg_decode_caughtup(LogicalDecodingContext *ctx)
 		{
 			LWLockAcquire(MtmReceiverBarrier, LW_EXCLUSIVE);
 			hooks_data->recovery_done = true;
+			mtm_log(ProtoTraceState, "Start building safe point to finish recovery");
 		}
 		else if (MtmAllApplyWorkersFinished())
 		{
+			mtm_log(ProtoTraceState, "Recovery finish safe point done");
 			MtmOutputPluginPrepareWrite(ctx, true, true);
 			data->api->write_caughtup(ctx->out, data, ctx->reader->EndRecPtr);
 			MtmOutputPluginWrite(ctx, true, true);

tests/docker-entrypoint.sh

@@ -2,6 +2,7 @@
 
 if [ "$1" = 'postgres' ]; then
 	mkdir -p "$PGDATA"
+	mkdir -p /pg/archive/
 	mkdir -p /pg/src/src/test/regress/testtablespace
 
 	# look specifically for PG_VERSION, as it is expected in the DB dir
@@ -48,31 +49,24 @@ if [ "$1" = 'postgres' ]; then
 
 
 		cat <<-EOF >> $PGDATA/postgresql.conf
-			listen_addresses='*' 
+			listen_addresses='*'
+			log_line_prefix = '%m [%p]: '
+			archive_mode = on
+			archive_command = 'cp %p /pg/archive/%f'
+
+			fsync = on
+
 			max_prepared_transactions = 100
-			synchronous_commit = on
-			fsync = off
 			wal_level = logical
 			max_worker_processes = 50
 			max_replication_slots = 10
 			max_wal_senders = 10
-			shared_preload_libraries = 'multimaster'
-			log_line_prefix = '%m [%p]: '
-			wal_writer_delay = 500ms
-			archive_mode = on
-			archive_command = '/bin/false'
-			# log_statement = all
 
+			shared_preload_libraries = 'multimaster'
 			multimaster.max_nodes = 3
 			multimaster.heartbeat_recv_timeout = 1100
 			multimaster.heartbeat_send_timeout = 250
-			multimaster.max_recovery_lag = 1GB
-			multimaster.min_recovery_lag = 10kB
-			multimaster.preserve_commit_order = on
-
 			multimaster.volkswagen_mode = on
-			multimaster.ignore_tables_without_pk = on
-			# partition_backend = 'internal'
 		EOF
 
 		if [ -n "$NODE_ID" ]; then

tests/lib/test_helper.py

@@ -39,7 +39,7 @@ def awaitCommit(self, node_id):
         total_sleep = 0
 
         while total_sleep <= TEST_MAX_RECOVERY_TIME:
-            aggs = self.client.get_aggregates(clean=False, _print=False)
+            aggs = self.client.get_aggregates(clean=False, _print=True)
             print('=== ',aggs[node_id]['transfer']['finish'])
             if ('commit' in aggs[node_id]['transfer']['finish'] and
                     aggs[node_id]['transfer']['finish']['commit'] > 10):

tests/test_recovery.py

@@ -138,7 +138,7 @@ def test_node_bicrash(self):
 
         for _ in range(10):
             aggs_failure, aggs = self.performFailure(CrashRecoverNode('node3'),
-                node_wait_for_commit=2, stop_load=True)
+                node_wait_for_online="dbname=regression user=postgres host=127.0.0.1 port=15434", stop_load=True)
 
             self.assertCommits(aggs_failure[:2])
             self.assertNoCommits(aggs_failure[2:])
@@ -148,7 +148,7 @@ def test_node_bicrash(self):
             self.assertIsolation(aggs)
 
             aggs_failure, aggs = self.performFailure(CrashRecoverNode('node1'),
-                node_wait_for_commit=0, stop_load=True)
+                node_wait_for_online="dbname=regression user=postgres host=127.0.0.1 port=15432", stop_load=True)
 
             self.assertNoCommits(aggs_failure[0:1])  # [1]
             self.assertCommits(aggs_failure[1:]) # [2, 3]