enh: check completeness of WAL log entries after repairing

include/util/taoserror.h

@@ -451,6 +451,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_WAL_OUT_OF_MEMORY             TAOS_DEF_ERROR_CODE(0, 0x1004)
 #define TSDB_CODE_WAL_LOG_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x1005)
 #define TSDB_CODE_WAL_CHKSUM_MISMATCH           TAOS_DEF_ERROR_CODE(0, 0x1006)
+#define TSDB_CODE_WAL_LOG_INCOMPLETE            TAOS_DEF_ERROR_CODE(0, 0x1007)
 
 // tfs
 #define TSDB_CODE_FS_INVLD_CFG                  TAOS_DEF_ERROR_CODE(0, 0x2201)

source/libs/wal/src/walMeta.c

@@ -282,6 +282,33 @@ void walAlignVersions(SWal* pWal) {
   pWal->vers.appliedVer = TMIN(pWal->vers.commitVer, pWal->vers.appliedVer);
 }
 
+bool walLogEntriesComplete(const SWal* pWal) {
+  int32_t sz = taosArrayGetSize(pWal->fileInfoSet);
+  bool    complete = true;
+  int32_t fileIdx = -1;
+  int64_t index = pWal->vers.firstVer;
+
+  while (++fileIdx < sz) {
+    SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
+    if (pFileInfo->firstVer != index) {
+      break;
+    }
+    index = pFileInfo->lastVer + ((fileIdx + 1 < sz) ? 1 : 0);
+  }
+  // empty is regarded as complete
+  if (sz != 0) {
+    complete = (index == pWal->vers.lastVer);
+  }
+
+  if (!complete) {
+    wError("vgId:%d, WAL log entries incomplete in range [%" PRId64 ", %" PRId64 "], aligned with snaphotVer:%" PRId64,
+           pWal->cfg.vgId, pWal->vers.firstVer, pWal->vers.lastVer, pWal->vers.snapshotVer);
+    terrno = TSDB_CODE_WAL_LOG_INCOMPLETE;
+  }
+
+  return complete;
+}
+
 int walCheckAndRepairMeta(SWal* pWal) {
   // load log files, get first/snapshot/last version info
   const char* logPattern = "^[0-9]+.log$";
@@ -395,6 +422,11 @@ int walCheckAndRepairMeta(SWal* pWal) {
   if (updateMeta) {
     (void)walSaveMeta(pWal);
   }
+
+  if (!walLogEntriesComplete(pWal)) {
+    return -1;
+  }
+
   return 0;
 }
 

source/util/src/terror.c

@@ -452,6 +452,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_WAL_INVALID_VER,              "WAL use invalid versi
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_OUT_OF_MEMORY,            "WAL out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_LOG_NOT_EXIST,            "WAL log not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_CHKSUM_MISMATCH,          "WAL checksum mismatch")
+TAOS_DEFINE_ERROR(TSDB_CODE_WAL_LOG_INCOMPLETE,           "WAL log incomplete")
 
 // tfs
 TAOS_DEFINE_ERROR(TSDB_CODE_FS_INVLD_CFG,                 "tfs invalid mount config")