修复initLastTaskLsn在集群重启后无法正确初始化归档起点的问题

src/gausskernel/process/postmaster/pgarch.cpp

@@ -109,6 +109,7 @@ typedef bool(*doArchive)(XLogRecPtr);
 static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun);
 static void archKill(int code, Datum arg);
 static void initLastTaskLsn();
+static void initArchiveCxt();
 
 AlarmCheckResult DataInstArchChecker(Alarm* alarm, AlarmAdditionalParam* additionalParam)
 {
@@ -227,6 +228,7 @@ NON_EXEC_STATIC void PgArchiverMain()
     init_ps_display("archiver process", "", "", "");
     setObsArchLatch(&t_thrd.arch.mainloop_latch);
     initLastTaskLsn();
+    initArchiveCxt();
     pgarch_MainLoop();
 
     gs_thread_exit(0);
@@ -689,14 +691,28 @@ static void pgarch_ArchiverObsCopyLoop(XLogRecPtr flushPtr, doArchive fun)
         }
 
         uint32 size = isObsSlot() ? OBS_XLOG_SLICE_BLOCK_SIZE : NAS_XLOG_FILE_SIZE;
-        targetLsn = Min(t_thrd.arch.pitr_task_last_lsn + size - (t_thrd.arch.pitr_task_last_lsn % size) - 1,
-                        flushPtr);
+        if (flushPtr == InvalidXLogRecPtr) {
+            targetLsn = t_thrd.arch.pitr_task_last_lsn + size - (t_thrd.arch.pitr_task_last_lsn % size) - 1;
+        } else {
+            targetLsn = Min(t_thrd.arch.pitr_task_last_lsn + size - (t_thrd.arch.pitr_task_last_lsn % size) - 1,
+                            flushPtr);
+        }
 
         /* The previous slice has been archived, switch to the next. */
         if (t_thrd.arch.pitr_task_last_lsn == targetLsn) {
             targetLsn = Min(targetLsn + size, flushPtr);
         }
 
+        if (!XlogFileIsExisted(t_thrd.proc_cxt.DataDir, targetLsn, DEFAULT_TIMELINE_ID)) {
+            ereport(WARNING,
+                (errmsg("transaction log file \"%X/%X\" does not existed, it may be archived by the old primary.", 
+                    (uint32)(targetLsn >> 32), (uint32)(targetLsn))));
+            gettimeofday(&tv, NULL);
+            t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv);
+            t_thrd.arch.pitr_task_last_lsn = targetLsn;
+            continue;
+        }
+
         if (fun(targetLsn) == false) {
             ereport(WARNING,
                 (errmsg("transaction log file \"%X/%X\" could not be archived: try again", 
@@ -1102,14 +1118,14 @@ static void initLastTaskLsn()
     gettimeofday(&tv,NULL);
     XLogRecPtr targetLsn;
     t_thrd.arch.last_arch_time = TIME_GET_MILLISEC(tv);
-    ReplicationSlot* obs_archive_slot = getObsReplicationSlot();
+    volatile ReplicationSlot* obs_archive_slot = getObsReplicationSlot();
     if (obs_archive_slot != NULL && !IsServerModeStandby()) {
-        ArchiveXlogMessage obs_archive_info;
-        if (obs_replication_get_last_xlog(&obs_archive_info) == 0) {
-            t_thrd.arch.pitr_task_last_lsn = obs_archive_info.targetLsn;
+        XLogRecPtr targetLsn = obs_archive_slot->data.restart_lsn;
+        if (targetLsn != InvalidXLogRecPtr) {
+            t_thrd.arch.pitr_task_last_lsn = targetLsn;
             advanceObsSlot(t_thrd.arch.pitr_task_last_lsn);
             ereport(LOG,
-                (errmsg("initLastTaskLsn update lsn to  %X/%X from obs", (uint32)(t_thrd.arch.pitr_task_last_lsn >> 32), 
+                (errmsg("initLastTaskLsn update lsn to  %X/%X from slot", (uint32)(t_thrd.arch.pitr_task_last_lsn >> 32), 
                     (uint32)(t_thrd.arch.pitr_task_last_lsn))));
         } else {
             targetLsn = GetFlushRecPtr();
@@ -1128,3 +1144,12 @@ static void initLastTaskLsn()
                 (uint32)(t_thrd.arch.pitr_task_last_lsn))));
     }
 }
+
+/* 
+ * The global variables related to archiving tasks must be initialized before entering the mainloop.
+ */
+static void initArchiveCxt() {
+    g_instance.archive_obs_cxt.pitr_finish_result = false;
+    g_instance.archive_obs_cxt.archive_task.targetLsn = InvalidXLogRecPtr;
+    g_instance.archive_obs_cxt.pitr_task_status = PITR_TASK_NONE;
+}

src/gausskernel/process/threadpool/knl_thread.cpp

@@ -626,7 +626,9 @@ static void knl_t_arch_init(knl_t_arch_context* arch)
     arch->last_sigterm_time = 0;
     arch->pitr_task_last_lsn = 0;
     arch->task_wait_interval = 1000;
+    arch->advance_slot_wait_interval = 1000;
     arch->last_arch_time = 0;
+    arch->last_advance_slot_time = 0;
     arch->sync_walsender_idx = -1;
 }
 

src/gausskernel/storage/replication/slot.cpp

@@ -52,6 +52,7 @@
 #include "postmaster/postmaster.h"
 #include "utils/builtins.h"
 
+#define TIME_GET_MILLISEC(t) (((long)(t).tv_sec * 1000) + ((long)(t).tv_usec) / 1000)
 
 extern bool PMstateIsRun(void);
 
@@ -1903,6 +1904,7 @@ ReplicationSlot *getObsReplicationSlot()
             g_instance.archive_obs_cxt.archive_slot->archive_obs->obs_sk = pstrdup_ext(slot->archive_obs->obs_sk);
             g_instance.archive_obs_cxt.archive_slot->archive_obs->obs_prefix = pstrdup_ext(slot->archive_obs->obs_prefix);
             g_instance.archive_obs_cxt.archive_slot->archive_obs->media_type = slot->archive_obs->media_type;
+            g_instance.archive_obs_cxt.archive_slot->data.restart_lsn = slot->data.restart_lsn;
             MemoryContextSwitchTo(curr);
             SpinLockRelease(&slot->mutex);
             *slot_idx = slotno;
@@ -1924,12 +1926,20 @@ void advanceObsSlot(XLogRecPtr restart_pos)
         SpinLockAcquire(&slot->mutex);
         if (slot->in_use == true && slot->archive_obs != NULL) {
             slot->data.restart_lsn = restart_pos;
+            SpinLockRelease(&slot->mutex);
+            struct timeval tv;
+            gettimeofday(&tv, NULL);
+            long diff = TIME_GET_MILLISEC(tv) - t_thrd.arch.last_advance_slot_time;
+            if (!RecoveryInProgress() && diff > t_thrd.arch.advance_slot_wait_interval) {
+                log_slot_advance(&slot->data);
+                t_thrd.arch.last_advance_slot_time = TIME_GET_MILLISEC(tv);
+            }
         } else {
             ereport(WARNING,
                 (errcode_for_file_access(), errmsg("slot idx not valid, obs slot %X/%X not advance ",
                     (uint32)(restart_pos >> 32), (uint32)(restart_pos))));
+            SpinLockRelease(&slot->mutex);
         }
-        SpinLockRelease(&slot->mutex);
     }
 }
 

src/gausskernel/storage/replication/slotfuncs.cpp

@@ -849,9 +849,11 @@ void redo_slot_advance(const ReplicationSlotPersistentData *slotInfo)
      * If logical replication slot is active on the current standby, the current
      * standby notify the primary to advance the logical replication slot.
      * Thus, we do not redo the slot_advance log.
+     * for logical replication slot, the Slot->data.database is VALID.
      */
 #ifndef ENABLE_MULTIPLE_NODES
-    if (IsReplicationSlotActive(NameStr(slotInfo->name))) {
+    if (IsReplicationSlotActive(NameStr(slotInfo->name)) && t_thrd.slot_cxt.MyReplicationSlot != NULL && 
+        t_thrd.slot_cxt.MyReplicationSlot->data.database != InvalidOid) {
         return;
     }
 #endif

src/gausskernel/storage/replication/walsender.cpp

@@ -3078,7 +3078,9 @@ static int WalSndLoop(WalSndSendDataCallback send_data)
         }
 
         volatile unsigned int *pitr_archive_flag = &t_thrd.walsender_cxt.MyWalSnd->archive_flag;
-        /* standby can't parse the archive message when the standby is building,
+
+        /* 
+         * standby can't parse the archive message when the standby is building,
          * so we can't send the archive message.
          */
         if (unlikely(pg_atomic_read_u32(pitr_archive_flag) == 1) &&

src/include/knl/knl_thread.h

@@ -1188,9 +1188,12 @@ typedef struct knl_t_arch_context {
     XLogRecPtr pitr_task_last_lsn;
     /* millsecond */
     int task_wait_interval;
+    long advance_slot_wait_interval;
     int sync_walsender_idx;
     /* for standby millsecond*/
     long last_arch_time;
+    /* for advance slot millsecond*/
+    long last_advance_slot_time;
 } knl_t_arch_context;
 
 /* Maximum length of a timezone name (not including trailing null) */