Fix/spoc 106 session reset feedback lsn (#80)

* fix: avoid redundant replication session resets

Removed explicit replorigin_session_reset() calls from the apply
worker. These resets were unnecessary, as PostgreSQL handles
replication session cleanup via on-exit callbacks. Even in retry
scenarios (e.g. within PG_CATCH), a reset followed by re-setup is not
required, since we are not exiting but simply retrying the same
transaction in the same session context.

* fix: prevent flush LSN advancement from 'w' message feedback

- Introduced maybe_send_feedback() to centralize logic for deciding
  when to send feedback. Feedback is now sent only after a threshold
  number of 'w' messages or after wal_sender_timeout / 2 has elapsed.

- Ensured that feedback does not update the flush LSN unless actual
  changes have be en applied. This avoids the risk of acknowledging
  WAL positions that haven't been safely replayed.

Author: rasifr

spock_apply.c

@@ -288,15 +288,14 @@ static void update_progress_entry(Oid target_node_id,
 static void check_and_update_progress(XLogRecPtr last_received_lsn,
 								TimestampTz timestamp);
 
-static ApplyReplayEntry *
-apply_replay_entry_create(int r, char *buf);
-static void
-apply_replay_entry_free(ApplyReplayEntry *entry);
-static void
-apply_replay_queue_reset(void);
-
+static ApplyReplayEntry *apply_replay_entry_create(int r, char *buf);
+static void apply_replay_entry_free(ApplyReplayEntry *entry);
+static void apply_replay_queue_reset(void);
+static void maybe_send_feedback(PGconn *applyconn, XLogRecPtr lsn_to_send,
+								TimestampTz *last_receive_timestamp);
 static void append_feedback_position(XLogRecPtr recvpos);
-static void get_feedback_position(XLogRecPtr *recvpos, XLogRecPtr *writepos, XLogRecPtr *flushpos, XLogRecPtr *max_recvpos);
+static void get_feedback_position(XLogRecPtr *recvpos, XLogRecPtr *writepos,
+								  XLogRecPtr *flushpos, XLogRecPtr *max_recvpos);
 
 
 /*
@@ -1868,7 +1867,9 @@ spock_apply_worker_shmem_exit(int code, Datum arg)
 	 * on_proc_exit because the backend may also clean up the origin
 	 * in certain cases, and we want to avoid duplicate cleanup.
 	 */
-	replorigin_session_reset();
+	replorigin_session_origin = InvalidRepOriginId;
+	replorigin_session_origin_lsn = InvalidXLogRecPtr;
+	replorigin_session_origin_timestamp = 0;
 }
 
 /*
@@ -2740,7 +2741,6 @@ apply_work(PGconn *streamConn)
 	TimestampTz last_receive_timestamp = GetCurrentTimestamp();
 	bool		need_replay;
 	ErrorData  *edata;
-	RepOriginId originid;
 
 	applyconn = streamConn;
 	fd = PQsocket(applyconn);
@@ -2763,7 +2763,6 @@ apply_work(PGconn *streamConn)
 	if (MyApplyWorker->apply_group == NULL)
 		spock_apply_worker_attach(); /* Attach this worker. */
 
-	originid = replorigin_session_origin;
 stream_replay:
 
 	need_replay = false;
@@ -2921,38 +2920,26 @@ apply_work(PGconn *streamConn)
 				{
 					XLogRecPtr start_lsn;
 					XLogRecPtr end_lsn;
-					static int w_message_count = 0;
 
 					start_lsn = pq_getmsgint64(msg);
 					end_lsn = pq_getmsgint64(msg);
 					pq_getmsgint64(msg); /* sendTime */
 
+					/*
+					 * Call maybe_send_feedback before last_received is updated.
+					 * This ordering guarantees that feedback LSN never advertises
+					 * a position beyond what has actually been received and processed.
+					 * Prevents skipping over unapplied changes due to premature flush LSN.
+					 */
+					maybe_send_feedback(applyconn, last_received,
+										&last_receive_timestamp);
+
 					if (last_received < start_lsn)
 						last_received = start_lsn;
 
 					if (last_received < end_lsn)
 						last_received = end_lsn;
 
-					w_message_count++;
-
-					/*
-					 * Send feedback if wal_sender_timeout/2 has passed or after 10 'w' messages.
-					 */
-					if (TimestampDifferenceExceeds(last_receive_timestamp, GetCurrentTimestamp(), wal_sender_timeout / 2) ||
-						w_message_count >= 10)
-					{
-						elog(DEBUG2, "SPOCK %s: force sending feedback after %d 'w' messages or timeout",
-							 MySubscription->name, w_message_count);
-						/*
-						 * We need to send feedback to the walsender process
-						 * to avoid remote wal_sender_timeout.
-						 */
-						send_feedback(applyconn, last_received, GetCurrentTimestamp(), true);
-						last_receive_timestamp = GetCurrentTimestamp();
-						w_message_count = 0;
-					}
-
-
 					/*
 					 * Append the entry to the end of the replay queue
 					 * if we read it from the stream but check for overflow.
@@ -3162,8 +3149,6 @@ apply_work(PGconn *streamConn)
 		MemoryContextSwitchTo(MessageContext);
 		elog(LOG, "SPOCK: caught initial exception - %s", edata->message);
 
-		/* reset replication session to avoid reuse of it after error. */
-		replorigin_session_reset();
 		FlushErrorState();
 
 		MemoryContextReset(MessageContext);
@@ -3184,10 +3169,6 @@ apply_work(PGconn *streamConn)
 	{
 		MyApplyWorker->use_try_block = true;
 
-		/* Its possible that origin session may have been reset above */
-		replorigin_session_setup(originid);
-		replorigin_session_origin = originid;
-
 		goto stream_replay;
 	}
 
@@ -4173,3 +4154,34 @@ apply_replay_queue_reset(void)
 
 	MemoryContextReset(ApplyReplayContext);
 }
+
+/*
+ * Check if we should send feedback based on message count or timeout.
+ * Resets internal state if feedback is sent.
+ */
+static void
+maybe_send_feedback(PGconn *applyconn, XLogRecPtr lsn_to_send,
+					TimestampTz *last_receive_timestamp)
+{
+	static int w_message_count = 0;
+	TimestampTz now = GetCurrentTimestamp();
+
+	w_message_count++;
+
+	/*
+	 * Send feedback if wal_sender_timeout/2 has passed or after 10 'w' messages.
+	 */
+	if (TimestampDifferenceExceeds(*last_receive_timestamp, now, wal_sender_timeout / 2) ||
+		w_message_count >= 10)
+	{
+		elog(DEBUG2, "SPOCK %s: force sending feedback after %d 'w' messages or timeout",
+				MySubscription->name, w_message_count);
+		/*
+		 * We need to send feedback to the walsender process
+		 * to avoid remote wal_sender_timeout.
+		 */
+		send_feedback(applyconn, lsn_to_send, now, true);
+		*last_receive_timestamp = now;
+		w_message_count = 0;
+	}
+}