Feat/336/wal-recover

include/common/tglobal.h

@@ -291,7 +291,9 @@ extern int64_t tsmaDataDeleteMark;
 // wal
 extern int64_t tsWalFsyncDataSizeLimit;
 extern bool    tsWalForceRepair;
-extern bool    tsWalDeleteOnCorruption;
+// WAL recovery policy (only affects single replica)
+// 0 = refuse to start (default), 1 = delete corrupted and start
+extern int32_t tsWalRecoveryPolicy;
 
 // internal
 extern bool    tsDiskIDCheckEnabled;

include/libs/sync/sync.h

@@ -303,6 +303,8 @@ const char* syncStr(ESyncState state);
 
 int32_t syncNodeGetConfig(int64_t rid, SSyncCfg* cfg);
 
+int32_t syncNotifyWalTruncated(int32_t vgId, int64_t truncatedVer);
+
 // util
 int32_t syncSnapInfoDataRealloc(SSnapshot* pSnap, int32_t size);
 

include/libs/wal/wal.h

@@ -166,7 +166,7 @@ int32_t walInit(stopDnodeFn stopDnode);
 void    walCleanUp();
 
 // handle open and ctl
-SWal   *walOpen(const char *path, SWalCfg *pCfg);
+SWal   *walOpen(const char *path, SWalCfg *pCfg, int32_t replica);
 int32_t walAlter(SWal *, SWalCfg *pCfg);
 int32_t walPersist(SWal *);
 void    walClose(SWal *);

source/common/src/tglobal.c

@@ -356,7 +356,7 @@ bool    tsStartUdfd = true;
 // wal
 int64_t tsWalFsyncDataSizeLimit = (100 * 1024 * 1024L);
 bool    tsWalForceRepair = 0;
-bool    tsWalDeleteOnCorruption = false;
+int32_t tsWalRecoveryPolicy = 0;  // Default: refuse to start for single replica
 
 // ttl
 bool    tsTtlChangeOnWrite = false;  // if true, ttl delete time changes on last write
@@ -981,7 +981,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   TAOS_CHECK_RETURN(cfgAddInt64(pCfg, "syncApplyQueueSize", tsSyncApplyQueueSize, 32, 2048, CFG_SCOPE_SERVER, CFG_DYN_SERVER,CFG_CATEGORY_GLOBAL));
   TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "syncRoutineReportInterval", tsRoutineReportInterval, 5, 600, CFG_SCOPE_SERVER, CFG_DYN_SERVER,CFG_CATEGORY_LOCAL));
   TAOS_CHECK_RETURN(cfgAddBool(pCfg, "syncLogHeartbeat", tsSyncLogHeartbeat, CFG_SCOPE_SERVER, CFG_DYN_SERVER,CFG_CATEGORY_LOCAL));
-  TAOS_CHECK_RETURN(cfgAddBool(pCfg, "walDeleteOnCorruption", tsWalDeleteOnCorruption, CFG_SCOPE_SERVER, CFG_DYN_NONE,CFG_CATEGORY_LOCAL));
+  TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "walRecoveryPolicy", tsWalRecoveryPolicy, 0, 1, CFG_SCOPE_SERVER, CFG_DYN_NONE, CFG_CATEGORY_LOCAL));
 
   TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "syncTimeout", tsSyncTimeout, 0, 60 * 24 * 2 * 1000, CFG_SCOPE_SERVER, CFG_DYN_SERVER,CFG_CATEGORY_GLOBAL));
 
@@ -2143,8 +2143,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsRpcRecvLogThreshold = pItem->i32;
   // GRANT_CFG_GET;
 
-  TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "walDeleteOnCorruption");
-  tsWalDeleteOnCorruption = pItem->bval;
+  TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "walRecoveryPolicy");
+  tsWalRecoveryPolicy = pItem->i32;
 
   TAOS_RETURN(TSDB_CODE_SUCCESS);
 }

source/dnode/mnode/impl/src/mndMain.c

@@ -632,7 +632,7 @@ static int32_t mndInitWal(SMnode *pMnode) {
   }
 #endif
 
-  pMnode->pWal = walOpen(path, &cfg);
+  pMnode->pWal = walOpen(path, &cfg, pMnode->syncMgmt.numOfReplicas);
   if (pMnode->pWal == NULL) {
     code = TSDB_CODE_MND_RETURN_VALUE_NULL;
     if (terrno != 0) code = terrno;

source/dnode/vnode/src/vnd/vnodeOpen.c

@@ -490,7 +490,7 @@ SVnode *vnodeOpen(const char *path, int32_t diskPrimary, STfs *pTfs, SMsgCb msgC
   TAOS_UNUSED(ret);
 
   vInfo("vgId:%d, start to open vnode wal", TD_VID(pVnode));
-  pVnode->pWal = walOpen(tdir, &(pVnode->config.walCfg));
+  pVnode->pWal = walOpen(tdir, &(pVnode->config.walCfg), pVnode->config.syncCfg.replicaNum);
   if (pVnode->pWal == NULL) {
     vError("vgId:%d, failed to open vnode wal since %s. wal:%s", TD_VID(pVnode), tstrerror(terrno), tdir);
     goto _err;

source/libs/sync/inc/syncInt.h

@@ -341,6 +341,8 @@ int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode);
 bool    syncNodeIsMnode(SSyncNode* pSyncNode);
 int32_t syncNodePeerStateInit(SSyncNode* pSyncNode);
 
+int32_t syncNotifyWalTruncated(int32_t vgId, int64_t truncatedVer);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/sync/src/syncMain.c

@@ -4090,3 +4090,8 @@ bool syncNodeCanChange(SSyncNode* pSyncNode) {
   return true;
 }
 #endif
+
+int32_t syncNotifyWalTruncated(int32_t vgId, int64_t truncatedVer) {
+  sInfo("vgId:%d, notified sync module: WAL truncated to ver:%" PRId64, vgId, truncatedVer);
+  return TSDB_CODE_SUCCESS;
+}

source/libs/wal/inc/walInt.h

@@ -168,7 +168,7 @@ int32_t walRemoveMeta(SWal* pWal);
 int32_t walRollImpl(SWal* pWal);
 int32_t walRollFileInfo(SWal* pWal);
 int32_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx, int64_t* lastVer);
-int32_t walCheckAndRepairMeta(SWal* pWal);
+int32_t walCheckAndRepairMeta(SWal* pWal, int32_t replica);
 int64_t walChangeWrite(SWal* pWal, int64_t ver);
 
 int32_t walCheckAndRepairIdx(SWal* pWal);

source/libs/wal/src/walMeta.c

@@ -431,7 +431,143 @@ static int32_t walRenameCorruptedDir(SWal* pWal) {
   TAOS_RETURN(code);
 }
 
-static int32_t walLogEntriesComplete(SWal* pWal) {
+static int32_t walTruncateCorruptedFiles(SWal* pWal, int32_t fileIdx, int32_t replica, int64_t lastValidVer) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  int32_t lino = 0;
+  bool shouldRecover = false;
+
+  if (replica == 3) {
+    shouldRecover = true;
+    SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
+    wInfo("vgId:%d, WAL corrupted at ver:%" PRId64 ", auto-recovery enabled for replica=3",
+          pWal->cfg.vgId, pFileInfo->firstVer);
+  } else {
+    shouldRecover = (tsWalRecoveryPolicy == 1);
+    if (shouldRecover) {
+      SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
+      wWarn("vgId:%d, WAL corrupted at ver:%" PRId64 ", force recovery enabled by walRecoveryPolicy=1",
+            pWal->cfg.vgId, pFileInfo->firstVer);
+    } else {
+      SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
+      wError("vgId:%d, WAL corrupted at ver:%" PRId64 ", refusing to start to prevent data loss",
+             pWal->cfg.vgId, pFileInfo->firstVer);
+      wError("vgId:%d, corrupted WAL files are preserved for manual inspection", pWal->cfg.vgId);
+      wError("vgId:%d, to force recovery with data loss, set 'walRecoveryPolicy 1' in taos.cfg and restart",
+             pWal->cfg.vgId);
+      TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
+    }
+  }
+
+  if (!shouldRecover) {
+    TAOS_RETURN(TSDB_CODE_WAL_FILE_CORRUPTED);
+  }
+
+  wInfo("vgId:%d, truncating WAL at corrupted file index %d, lastValidVer:%" PRId64, pWal->cfg.vgId, fileIdx, lastValidVer);
+
+  // Truncate current corrupted file if there's a valid version
+  if (fileIdx < taosArrayGetSize(pWal->fileInfoSet) && lastValidVer >= 0) {
+    SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
+
+    // Only truncate if this file has valid entries before corruption
+    if (lastValidVer >= pFileInfo->firstVer) {
+      char logName[WAL_FILE_LEN];
+      char idxName[WAL_FILE_LEN];
+      walBuildLogName(pWal, pFileInfo->firstVer, logName);
+      walBuildIdxName(pWal, pFileInfo->firstVer, idxName);
+
+      // Read the idx file to find the offset of lastValidVer + 1
+      TdFilePtr pIdxFile = taosOpenFile(idxName, TD_FILE_READ);
+      if (pIdxFile != NULL) {
+        int64_t idxEntries = lastValidVer - pFileInfo->firstVer + 1;
+        int64_t idxFileSize = idxEntries * sizeof(SWalIdxEntry);
+
+        // Truncate idx file
+        (void)taosCloseFile(&pIdxFile);
+        pIdxFile = taosOpenFile(idxName, TD_FILE_READ | TD_FILE_WRITE);
+        if (pIdxFile != NULL) {
+          if (taosFtruncateFile(pIdxFile, idxFileSize) == 0) {
+            wInfo("vgId:%d, truncated idx file %s to size:%" PRId64, pWal->cfg.vgId, idxName, idxFileSize);
+          } else {
+            wWarn("vgId:%d, failed to truncate idx file %s", pWal->cfg.vgId, idxName);
+          }
+          (void)taosCloseFile(&pIdxFile);
+        }
+
+        // Read the last valid entry's offset from idx, then read log entry to get size
+        pIdxFile = taosOpenFile(idxName, TD_FILE_READ);
+        if (pIdxFile != NULL && idxEntries > 0) {
+          SWalIdxEntry idxEntry;
+          int64_t readPos = (idxEntries - 1) * sizeof(SWalIdxEntry);
+          if (taosLSeekFile(pIdxFile, readPos, SEEK_SET) >= 0) {
+            if (taosReadFile(pIdxFile, &idxEntry, sizeof(SWalIdxEntry)) == sizeof(SWalIdxEntry)) {
+              // Read log entry header to get the size
+              TdFilePtr pLogFile = taosOpenFile(logName, TD_FILE_READ | TD_FILE_WRITE);
+              if (pLogFile != NULL) {
+                SWalCkHead ckHead;
+                if (taosLSeekFile(pLogFile, idxEntry.offset, SEEK_SET) >= 0) {
+                  if (taosReadFile(pLogFile, &ckHead, sizeof(SWalCkHead)) == sizeof(SWalCkHead)) {
+                    int32_t cryptedBodyLen = ckHead.head.bodyLen;
+                    if (pWal->cfg.encryptAlgorithm == DND_CA_SM4) {
+                      cryptedBodyLen = ENCRYPTED_LEN(cryptedBodyLen);
+                    }
+                    int64_t logTruncateOffset = idxEntry.offset + sizeof(SWalCkHead) + cryptedBodyLen;
+
+                    // Truncate log file
+                    if (taosFtruncateFile(pLogFile, logTruncateOffset) == 0) {
+                      wInfo("vgId:%d, truncated log file %s to offset:%" PRId64, pWal->cfg.vgId, logName, logTruncateOffset);
+                      pFileInfo->lastVer = lastValidVer;
+                      pFileInfo->fileSize = logTruncateOffset;
+                    } else {
+                      wWarn("vgId:%d, failed to truncate log file %s", pWal->cfg.vgId, logName);
+                    }
+                  }
+                }
+                (void)taosCloseFile(&pLogFile);
+              }
+            }
+          }
+          (void)taosCloseFile(&pIdxFile);
+        }
+      }
+
+      // Don't delete this file, move to next
+      fileIdx++;
+    }
+  }
+
+  // Delete all files from fileIdx onwards
+  for (int32_t i = fileIdx; i < taosArrayGetSize(pWal->fileInfoSet); i++) {
+    SWalFileInfo* pDelFileInfo = taosArrayGet(pWal->fileInfoSet, i);
+    char delLogName[WAL_FILE_LEN];
+    char delIdxName[WAL_FILE_LEN];
+
+    walBuildLogName(pWal, pDelFileInfo->firstVer, delLogName);
+    walBuildIdxName(pWal, pDelFileInfo->firstVer, delIdxName);
+
+    if (taosRemoveFile(delLogName) != 0) {
+      wWarn("vgId:%d, failed to remove corrupted log file %s", pWal->cfg.vgId, delLogName);
+    } else {
+      wInfo("vgId:%d, removed corrupted log file %s", pWal->cfg.vgId, delLogName);
+    }
+
+    if (taosRemoveFile(delIdxName) != 0) {
+      wWarn("vgId:%d, failed to remove corrupted idx file %s", pWal->cfg.vgId, delIdxName);
+    } else {
+      wInfo("vgId:%d, removed corrupted idx file %s", pWal->cfg.vgId, delIdxName);
+    }
+  }
+
+  // Remove deleted files from fileInfoSet
+  if (fileIdx < taosArrayGetSize(pWal->fileInfoSet)) {
+    taosArrayRemoveBatch(pWal->fileInfoSet, fileIdx, taosArrayGetSize(pWal->fileInfoSet) - fileIdx, NULL);
+  }
+
+  wInfo("vgId:%d, WAL truncated successfully", pWal->cfg.vgId);
+
+  TAOS_RETURN(TSDB_CODE_SUCCESS);
+}
+
+static int32_t walLogEntriesComplete(SWal* pWal, int32_t replica) {
   int32_t sz = taosArrayGetSize(pWal->fileInfoSet);
   bool    complete = true;
   int32_t fileIdx = -1;
@@ -451,13 +587,10 @@ static int32_t walLogEntriesComplete(SWal* pWal) {
 
   if (!complete) {
     wError("vgId:%d, WAL log entries incomplete in range [%" PRId64 ", %" PRId64 "], index:%" PRId64
-           ", snaphot index:%" PRId64,
-           pWal->cfg.vgId, pWal->vers.firstVer, pWal->vers.lastVer, index, pWal->vers.snapshotVer);
-    if (tsWalDeleteOnCorruption) {
-      TAOS_RETURN(walRenameCorruptedDir(pWal));
-    } else {
-      TAOS_RETURN(TSDB_CODE_WAL_LOG_INCOMPLETE);
-    }
+           ", snaphot index:%" PRId64 ", fileIdx:%d",
+           pWal->cfg.vgId, pWal->vers.firstVer, pWal->vers.lastVer, index, pWal->vers.snapshotVer, fileIdx);
+    int64_t lastValidVer = (index > 0) ? (index - 1) : -1;
+    TAOS_RETURN(walTruncateCorruptedFiles(pWal, fileIdx, replica, lastValidVer));
   } else {
     TAOS_RETURN(TSDB_CODE_SUCCESS);
   }
@@ -517,7 +650,7 @@ void walRegfree(regex_t* ptr) {
   regfree(ptr);
 }
 
-int32_t walCheckAndRepairMeta(SWal* pWal) {
+int32_t walCheckAndRepairMeta(SWal* pWal, int32_t replica) {
   // load log files, get first/snapshot/last version info
   int32_t     code = 0;
   int32_t     lino = 0;
@@ -526,6 +659,8 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
   regex_t     logRegPattern, idxRegPattern;
   TdDirPtr    pDir = NULL;
   SArray*     actualLog = NULL;
+  bool        walTruncated = false;
+  int64_t     truncatedVer = -1;
 
   wInfo("vgId:%d, begin to repair meta, wal path:%s, first index:%" PRId64 ", last index:%" PRId64
         ", snapshot index:%" PRId64,
@@ -607,22 +742,35 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
     if (lastVer < 0) {
       if (code != TSDB_CODE_WAL_LOG_NOT_EXIST) {
         wError("vgId:%d, failed to scan wal last index since %s", pWal->cfg.vgId, tstrerror(code));
-        if (tsWalDeleteOnCorruption) {
-          TAOS_RETURN(walRenameCorruptedDir(pWal));
+
+        // Calculate last valid version from previous file
+        int64_t lastValidVer = (fileIdx > 0) ? ((SWalFileInfo*)taosArrayGet(pWal->fileInfoSet, fileIdx - 1))->lastVer : -1;
+
+        code = walTruncateCorruptedFiles(pWal, fileIdx, replica, lastValidVer);
+        if (code != TSDB_CODE_SUCCESS) {
+          goto _exit;
         }
-        goto _exit;
-      }
-      // empty log file
-      lastVer = pFileInfo->firstVer - 1;
 
-      code = TSDB_CODE_SUCCESS;
+        // After truncation, set lastVer based on remaining files
+        lastVer = lastValidVer;
+        wInfo("vgId:%d, WAL truncated, new lastVer:%" PRId64, pWal->cfg.vgId, lastVer);
+        updateMeta = true;
+        code = TSDB_CODE_SUCCESS;
+      } else {
+        // empty log file
+        lastVer = pFileInfo->firstVer - 1;
+        code = TSDB_CODE_SUCCESS;
+      }
     }
-    wInfo("vgId:%d, repaired file %s, last index:%" PRId64 ", fileSize:%" PRId64 ", fileSize in meta:%" PRId64,
-          pWal->cfg.vgId, fnameStr, lastVer, fileSize, pFileInfo->fileSize);
 
-    // update lastVer
-    pFileInfo->lastVer = lastVer;
-    totSize += pFileInfo->fileSize;
+    if (code == TSDB_CODE_SUCCESS && lastVer >= 0) {
+      wInfo("vgId:%d, repaired file %s, last index:%" PRId64 ", fileSize:%" PRId64 ", fileSize in meta:%" PRId64,
+            pWal->cfg.vgId, fnameStr, lastVer, fileSize, pFileInfo->fileSize);
+
+      // update lastVer
+      pFileInfo->lastVer = lastVer;
+      totSize += pFileInfo->fileSize;
+    }
   }
 
   // reset vers info and so on
@@ -647,7 +795,7 @@ int32_t walCheckAndRepairMeta(SWal* pWal) {
     TAOS_CHECK_EXIT(walSaveMeta(pWal));
   }
 
-  TAOS_CHECK_EXIT(walLogEntriesComplete(pWal));
+  TAOS_CHECK_EXIT(walLogEntriesComplete(pWal, replica));
 
   wInfo("vgId:%d, success to repair meta, wal path:%s, first index:%" PRId64 ", last index:%" PRId64
         ", snapshot index:%" PRId64,

source/libs/wal/src/walMgmt.c

@@ -129,7 +129,7 @@ int32_t walInitWriteFileForSkip(SWal *pWal) {
   TAOS_RETURN(code);
 }
 
-SWal *walOpen(const char *path, SWalCfg *pCfg) {
+SWal *walOpen(const char *path, SWalCfg *pCfg, int32_t replica) {
   int32_t code = 0;
   SWal   *pWal = taosMemoryCalloc(1, sizeof(SWal));
   if (pWal == NULL) {
@@ -206,7 +206,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
     wWarn("vgId:%d, failed to load meta, code:0x%x", pWal->cfg.vgId, code);
   }
   if (pWal->cfg.level != TAOS_WAL_SKIP) {
-    code = walCheckAndRepairMeta(pWal);
+    code = walCheckAndRepairMeta(pWal, replica);
     if (code < 0) {
       wError("vgId:%d, cannot open wal since repair meta file failed since %s", pWal->cfg.vgId, tstrerror(code));
       goto _err;