Minor fixes and improvements for progress tracking

- Refactor apply_worker_get_progress() to return only the timestamp needed. Return value instead of pointer to entire struct.
- Lock sooner in spock_group_resource_dump() because of getting the number of entries
- Properly handle progress updates after COPY operations in replication set data sync and avoid potential memory leak
- Address compiler warnings

Author: mason-sharp

include/spock_group.h

@@ -156,7 +156,7 @@ extern void spock_group_detach(void);
 extern bool spock_group_progress_update(const SpockApplyProgress *sap);
 extern void spock_group_progress_update_ptr(SpockGroupEntry *entry,
 											const SpockApplyProgress *sap);
-extern SpockApplyProgress *apply_worker_get_progress(void);
+extern TimestampTz apply_worker_get_prev_remote_ts(void);
 
 extern void spock_group_resource_dump(void);
 extern void spock_checkpoint_hook(XLogRecPtr checkPointRedo, int flags);

src/spock_apply.c

@@ -242,7 +242,7 @@ wait_for_previous_transaction(void)
 		 * loop and process this transaction. Otherwise, wait for the
 		 * predecessor to commit.
 		 */
-		if (apply_worker_get_progress()->prev_remote_ts == required_commit_ts ||
+		if (apply_worker_get_prev_remote_ts() == required_commit_ts ||
 			required_commit_ts == 0)
 		{
 			break;
@@ -263,7 +263,7 @@ wait_for_previous_transaction(void)
 		elog(DEBUG1, "SPOCK: slot-group '%s' WAIT for ts [current proccessed"
 			 ", required] [" INT64_FORMAT ", " INT64_FORMAT "]",
 			 MySubscription->slot_name,
-			 apply_worker_get_progress()->prev_remote_ts,
+			 apply_worker_get_prev_remote_ts(),
 			 required_commit_ts);
 
 		/* Latch */

src/spock_executor.c

@@ -247,7 +247,6 @@ spock_object_access(ObjectAccessType access,
 	if (access == OAT_DROP)
 	{
 		ObjectAccessDrop *drop_arg = (ObjectAccessDrop *) arg;
-		ObjectAddress object;
 		DropBehavior behavior;
 
 		/* No need to check for internal deletions. */

src/spock_group.c

@@ -308,7 +308,7 @@ progress_update_struct(SpockApplyProgress *dest, const SpockApplyProgress *src)
 	 * Value of the received_lsn potentially can exceed remote_insert_lsn
 	 * because it is reported more frequently (by keepalive messages).
 	 */
-	Assert(!(dest->remote_commit_ts == 0 ^ dest->last_updated_ts == 0));
+	Assert(!((dest->remote_commit_ts == 0) ^ (dest->last_updated_ts == 0)));
 	Assert(dest->remote_commit_ts >= 0 && dest->last_updated_ts >= 0);
 }
 
@@ -392,22 +392,23 @@ spock_group_progress_update_ptr(SpockGroupEntry *e,
 }
 
 /*
- * apply_worker_get_progress
+ * apply_worker_get_prev_remote_ts
+ *
+ * Get the previous remote timestamp for our apply worker
  *
- * Return a pointer to the snapshot of the current apply worker's progress.
  */
-SpockApplyProgress *
-apply_worker_get_progress(void)
+TimestampTz
+apply_worker_get_prev_remote_ts(void)
 {
-	static SpockApplyProgress sap;
+	TimestampTz	prev_remote_ts;
 
 	Assert(MyApplyWorker != NULL);
 	Assert(MyApplyWorker->apply_group != NULL);
 
 	if (MyApplyWorker && MyApplyWorker->apply_group)
 	{
 		LWLockAcquire(SpockCtx->apply_group_master_lock, LW_SHARED);
-		sap = MyApplyWorker->apply_group->progress;
+		prev_remote_ts = MyApplyWorker->apply_group->progress.prev_remote_ts;
 		LWLockRelease(SpockCtx->apply_group_master_lock);
 	}
 	else
@@ -417,7 +418,7 @@ apply_worker_get_progress(void)
 		 */
 		elog(ERROR, "apply worker has not been fully initialised yet");
 
-	return &sap;
+	return prev_remote_ts;
 }
 
 /* Iterate all groups */
@@ -507,12 +508,13 @@ spock_group_resource_dump(void)
 	hdr.version = SPOCK_RES_VERSION;
 	hdr.system_identifier = GetSystemIdentifier();
 	hdr.flags = 0;
+
+	/* Acquire lock before reading hash table to ensure consistency */
+	LWLockAcquire(SpockCtx->apply_group_master_lock, LW_SHARED);
 	hdr.entry_count = hash_get_num_entries(SpockGroupHash);
 
 	write_buf(fd, &hdr, sizeof(hdr), SPOCK_RES_DUMPFILE "(header)");
 
-	LWLockAcquire(SpockCtx->apply_group_master_lock, LW_SHARED);
-
 	dctx.fd = fd;
 	dctx.count = 0;
 
@@ -662,7 +664,7 @@ spock_group_progress_update_list(List *lst)
 
 		elog(LOG, "SPOCK: adjust spock.progress %d->%d to "
 			 "remote_commit_ts='%s' "
-			 "remote_commit_lsn=%llX remote_insert_lsn=%llX",
+			 "remote_commit_lsn=%lX remote_insert_lsn=%lX",
 			 sap->key.remote_node_id, MySubscription->target->id,
 			 timestamptz_to_str(sap->remote_commit_ts),
 			 sap->remote_commit_lsn, sap->remote_insert_lsn);

src/spock_sync.c

@@ -958,6 +958,7 @@ copy_replication_sets_data(SpockSubscription *sub, const char *origin_dsn,
 	PGconn	   *origin_conn;
 	PGconn	   *target_conn;
 	List	   *tables;
+	List	   *progress_entries_list = NIL;
 	ListCell   *lc;
 
 	/* Connect to origin node. */
@@ -1021,12 +1022,19 @@ copy_replication_sets_data(SpockSubscription *sub, const char *origin_dsn,
 		CHECK_FOR_INTERRUPTS();
 	}
 
-	adjust_progress_info(origin_conn);
+	progress_entries_list = adjust_progress_info(origin_conn);
 
 	/* Finish the transactions and disconnect. */
 	finish_copy_origin_tx(origin_conn);
 	finish_copy_target_tx(target_conn);
 
+	/*
+	 * Match handling in copy_tables_data().
+	 * Update replication progress. We must do it after commit of the COPY.
+	 * Call below will free progress_entries_list
+	 */
+	spock_group_progress_update_list(progress_entries_list);
+
 	return tables;
 }