Flush logical messages with snapshots and replication origin (#382)

Co-authored-by: Konstantin Knizhnik <knizhnik@neon.tech>

Author: 2 people

src/backend/replication/logical/origin.c

@@ -692,7 +692,7 @@ CheckPointReplicationOrigin(void)
 		chkp_size += sizeof(crc);
 
 		/* NEON specific: persist snapshot in storage using logical message */
-		LogLogicalMessage("neon-file:pg_logical/replorigin_checkpoint", buf, chkp_size, false);
+		XLogFlush(LogLogicalMessage("neon-file:pg_logical/replorigin_checkpoint", buf, chkp_size, false));
 	}
 	pfree(buf);
 

src/backend/replication/logical/snapbuild.c

@@ -1756,7 +1756,7 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
 
 	/* NEON specific: persist snapshot in storage using logical message */
 	snprintf(prefix, sizeof(prefix), "neon-file:%s", path);
-	LogLogicalMessage(prefix, (char*)ondisk, needed_length, false);
+	XLogFlush(LogLogicalMessage(prefix, (char*)ondisk, needed_length, false));
 
 	errno = 0;
 	pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
@@ -2122,7 +2122,7 @@ CheckPointSnapBuild(void)
 
 			/* NEON specific: delete file from storage using logical message */
 			snprintf(prefix, sizeof(prefix), "neon-file:%s", path);
-			LogLogicalMessage(prefix, NULL, 0, false);
+			XLogFlush(LogLogicalMessage(prefix, NULL, 0, false));
 
 			/*
 			 * It's not particularly harmful, though strange, if we can't