Update.

Author: pgelephant2025

include/pgraft_core.h

@@ -80,6 +80,19 @@ typedef struct
 	int			port;
 	WORKER_STATUS status;
 
+	/*
+	 * Protects the command/status/apply circular buffers and their
+	 * head/tail/count indices below.  Client backends (producers) and the
+	 * background worker (consumer) mutate these concurrently, so every
+	 * head/tail/count update must be done while holding this spinlock.
+	 */
+	slock_t		mutex;
+
+	/* Monotonic id source used to uniquely identify queued commands
+	 * (replaces the old 1-second-granularity time(NULL) key, which collided
+	 * for commands queued within the same second). */
+	int64_t		next_command_id;
+
 	/* Fixed-size circular buffer for commands */
 	pgraft_command_t commands[MAX_COMMANDS];
 	int			command_head;		/* Index of next command to process */

src/pgraft.c

@@ -452,10 +452,16 @@ pgraft_main(Datum main_arg)
 						{
 							/* Create JSON data for Raft replication using json-c */
 							if (pgraft_json_create_kv_operation(PGRAFT_KV_PUT, cmd.kv_key, cmd.kv_value, cmd.kv_client_id, json_data, sizeof(json_data)) != 0) {
-								elog(ERROR, "pgraft: failed to create JSON for KV PUT operation");
-								continue;
+								/* Fail just this command; do NOT elog(ERROR) here,
+								 * which would abort (and restart) the whole worker. */
+								cmd.status = COMMAND_STATUS_FAILED;
+								snprintf(cmd.error_message, sizeof(cmd.error_message),
+										"Failed to create JSON for KV PUT operation");
+								elog(WARNING, "pgraft: %s", cmd.error_message);
+								pgraft_update_command_status(cmd.timestamp, cmd.status, cmd.error_message);
+								break;
 							}
-							
+
 							elog(LOG, "pgraft: calling pgraft_go_append_log with data=%s", json_data);
 
 							/* Replicate through Raft */
@@ -498,10 +504,16 @@ pgraft_main(Datum main_arg)
 						{
 							/* Create JSON data for Raft replication using json-c */
 							if (pgraft_json_create_kv_operation(PGRAFT_KV_DELETE, cmd.kv_key, NULL, cmd.kv_client_id, json_data, sizeof(json_data)) != 0) {
-								elog(ERROR, "pgraft: failed to create JSON for KV DELETE operation");
-								continue;
+								/* Fail just this command; do NOT elog(ERROR) here,
+								 * which would abort (and restart) the whole worker. */
+								cmd.status = COMMAND_STATUS_FAILED;
+								snprintf(cmd.error_message, sizeof(cmd.error_message),
+										"Failed to create JSON for KV DELETE operation");
+								elog(WARNING, "pgraft: %s", cmd.error_message);
+								pgraft_update_command_status(cmd.timestamp, cmd.status, cmd.error_message);
+								break;
 							}
-							
+
 							elog(LOG, "pgraft: calling pgraft_go_append_log with data=%s", json_data);
 
 							/* Replicate through Raft */
@@ -550,6 +562,45 @@ pgraft_main(Datum main_arg)
 			}
 		}
 
+		/*
+		 * Apply committed Raft entries that the Go layer has replicated to
+		 * this node.  applyEntry() in the Go library enqueues every committed
+		 * EntryNormal via pgraft_enqueue_for_apply_from_go(); here we drain
+		 * that queue and apply each entry locally.  Each apply is wrapped in
+		 * PG_TRY so that a single malformed entry cannot abort (and thereby
+		 * restart) the whole background worker.
+		 *
+		 * NOTE: this worker holds BGWORKER_SHMEM_ACCESS only (no database
+		 * connection), so only shared-memory key/value operations are applied
+		 * here.  SQL/DDL replication would additionally require
+		 * BackgroundWorkerInitializeConnection() and a transaction.
+		 */
+		{
+			pgraft_apply_entry_t apply_entry;
+			MemoryContext apply_loop_ctx = CurrentMemoryContext;
+
+			while (pgraft_dequeue_apply_entry(&apply_entry))
+			{
+				PG_TRY();
+				{
+					pgraft_apply_entry_to_postgres(apply_entry.raft_index,
+												   apply_entry.data,
+												   apply_entry.data_len);
+				}
+				PG_CATCH();
+				{
+					/* Recover and keep the worker alive; the entry is dropped
+					 * but the error is reported to the log. */
+					MemoryContextSwitchTo(apply_loop_ctx);
+					EmitErrorReport();
+					FlushErrorState();
+					elog(WARNING, "pgraft: error applying raft entry %lu, skipping",
+						 (unsigned long) apply_entry.raft_index);
+				}
+				PG_END_TRY();
+			}
+		}
+
 		pg_usleep(100000);
 		sleep_count++;
 
@@ -562,6 +613,14 @@ pgraft_main(Datum main_arg)
 
 	state->status = WORKER_STATUS_STOPPED;
 	elog(LOG, "pgraft: background worker stopped");
+
+	/*
+	 * Exit cleanly (status 0) in response to an explicit shutdown command.
+	 * A background worker that exits with status 0 is de-registered and NOT
+	 * restarted by the postmaster, so the shutdown is durable.  (bgw_restart_time
+	 * remains a finite interval so that an actual *crash* is still recovered.)
+	 */
+	proc_exit(0);
 }
 
 /*
@@ -583,14 +642,24 @@ pgraft_worker_get_state(void)
 			worker_state->port = 0;
 			strlcpy(worker_state->address, "127.0.0.1", sizeof(worker_state->address));
 			worker_state->status = WORKER_STATUS_STOPPED;
-			
+
+			/* Initialize the queue spinlock and command id source */
+			SpinLockInit(&worker_state->mutex);
+			worker_state->next_command_id = 1;
+
 			/* Initialize circular buffers */
 			worker_state->command_head = 0;
 			worker_state->command_tail = 0;
 			worker_state->command_count = 0;
 			worker_state->status_head = 0;
 			worker_state->status_tail = 0;
 			worker_state->status_count = 0;
+
+			/* Initialize apply queue */
+			worker_state->apply_head = 0;
+			worker_state->apply_tail = 0;
+			worker_state->apply_count = 0;
+			worker_state->last_applied_index = 0;
 		}
 	}
 	return worker_state;
@@ -839,6 +908,8 @@ pgraft_update_shared_memory_from_go(void)
 			elog(LOG, "pgraft: DEBUG - Got nodes JSON from Go: '%s'", nodes_json ? nodes_json : "NULL");
 			if (nodes_json && strcmp(nodes_json, "[]") != 0)
 			{
+				/* NOTE: nodes_json is freed unconditionally below (see the
+				 * matching pgraft_go_free_string after this block). */
 				/* Parse JSON using separate module */
 				int node_count;
 				int32_t node_ids[16];
@@ -870,9 +941,13 @@ pgraft_update_shared_memory_from_go(void)
 				{
 					elog(LOG, "pgraft: DEBUG - No valid nodes parsed from JSON");
 				}
-				
-				pgraft_go_free_string(nodes_json);
 			}
+
+			/* Free the Go-allocated string in ALL cases (including the common
+			 * "[]" steady-state path), otherwise the worker leaks one string
+			 * every iteration. */
+			if (nodes_json)
+				pgraft_go_free_string(nodes_json);
 		}
 	}
 

src/pgraft_apply.c

@@ -48,6 +48,15 @@ pgraft_apply_entry_to_postgres(uint64 raft_index, const char *data, size_t len)
 
 	elog(LOG, "pgraft: applying raft entry %lu to PostgreSQL (len=%zu)", raft_index, len);
 
+	/* Defensive: nothing to apply for empty/NULL payloads (e.g. Raft no-op
+	 * entries committed on leader election).  Avoid dereferencing data[0]. */
+	if (data == NULL || len == 0)
+	{
+		elog(DEBUG1, "pgraft: skipping empty raft entry %lu", raft_index);
+		pgraft_record_applied_index(raft_index);
+		return 0;
+	}
+
 	/* Create a memory context for this operation */
 	apply_context = AllocSetContextCreate(CurrentMemoryContext,
 										  "PgRaft Apply Context",
@@ -236,9 +245,9 @@ pgraft_serialize_log_entry(PgRaftLogEntry *entry, size_t *out_len)
 	result = (char *) palloc(8192);
 
 	/* Format: index|term|op|database|schema|sql */
-	len = snprintf(result, 8192, "%lu|%lu|%d|%s|%s|%s",
-				   (unsigned long) entry->index,
-				   (unsigned long) entry->term,
+	len = snprintf(result, 8192, UINT64_FORMAT "|" UINT64_FORMAT "|%d|%s|%s|%s",
+				   entry->index,
+				   entry->term,
 				   entry->op,
 				   entry->database,
 				   entry->schema,
@@ -257,26 +266,19 @@ void
 pgraft_record_applied_index(uint64 index)
 {
 	pgraft_worker_state_t *worker_state;
-	pgraft_cluster_t *shm_cluster;
-	
-	/* Get shared memory references */
-	shm_cluster = pgraft_core_get_shared_memory();
-	if (!shm_cluster) {
-		elog(WARNING, "pgraft: failed to get cluster state for recording applied index");
-		return;
-	}
-	
+
 	worker_state = pgraft_worker_get_state();
 	if (!worker_state) {
 		elog(WARNING, "pgraft: failed to get worker state for recording applied index");
 		return;
 	}
-	
-	/* Update the last applied index in shared memory */
-	SpinLockAcquire(&shm_cluster->mutex);
+
+	/* Guard the worker-state field with the worker state's OWN mutex (not the
+	 * cluster mutex, which lives in a different shared-memory struct). */
+	SpinLockAcquire(&worker_state->mutex);
 	worker_state->last_applied_index = index;
-	SpinLockRelease(&shm_cluster->mutex);
-	
+	SpinLockRelease(&worker_state->mutex);
+
 	elog(DEBUG2, "pgraft: recorded applied index %lu in shared memory", index);
 }
 
@@ -287,27 +289,19 @@ uint64
 pgraft_get_applied_index(void)
 {
 	pgraft_worker_state_t *worker_state;
-	pgraft_cluster_t *shm_cluster;
 	uint64 last_applied;
-	
-	/* Get shared memory references */
-	shm_cluster = pgraft_core_get_shared_memory();
-	if (!shm_cluster) {
-		elog(WARNING, "pgraft: failed to get cluster state for getting applied index");
-		return 0;
-	}
-	
+
 	worker_state = pgraft_worker_get_state();
 	if (!worker_state) {
 		elog(WARNING, "pgraft: failed to get worker state for getting applied index");
 		return 0;
 	}
-	
-	/* Get the last applied index from shared memory */
-	SpinLockAcquire(&shm_cluster->mutex);
+
+	/* Guard with the worker state's own mutex (matches pgraft_record_applied_index). */
+	SpinLockAcquire(&worker_state->mutex);
 	last_applied = worker_state->last_applied_index;
-	SpinLockRelease(&shm_cluster->mutex);
-	
+	SpinLockRelease(&worker_state->mutex);
+
 	return last_applied;
 }
 

src/pgraft_core.c

@@ -442,10 +442,26 @@ pgraft_core_get_shared_memory(void)
 		cluster = (pgraft_cluster_t *) ShmemInitStruct("pgraft_cluster",
 													   sizeof(pgraft_cluster_t),
 													   &found);
-		if (!found)
+		if (!found && cluster)
 		{
-			/* Initialize shared memory when first accessed */
-			pgraft_core_init_shared_memory();
+			/*
+			 * We are the first to touch the segment.  Initialize it in place.
+			 * Do NOT delegate to pgraft_core_init_shared_memory() here: that
+			 * would call ShmemInitStruct() a second time, observe found==true,
+			 * and skip SpinLockInit(), leaving the mutex uninitialized.
+			 */
+			memset(cluster, 0, sizeof(pgraft_cluster_t));
+			SpinLockInit(&cluster->mutex);
+			cluster->initialized = false;
+			cluster->node_id = -1;
+			cluster->current_term = 0;
+			cluster->leader_id = -1;
+			strncpy(cluster->state, "stopped", sizeof(cluster->state) - 1);
+			cluster->state[sizeof(cluster->state) - 1] = '\0';
+			cluster->num_nodes = 0;
+			cluster->messages_processed = 0;
+			cluster->heartbeats_sent = 0;
+			cluster->elections_triggered = 0;
 		}
 	}
 	return cluster;