From 3014c66965c91c50dee40b28b6af0c42a06a6028 Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Wed, 26 Jun 2024 11:21:31 +0800
Subject: [PATCH] fix(streamMeta): commit tdb after load tasks

---
 source/libs/stream/src/streamMeta.c | 29 +++++++++++++++++------------
 1 file changed, 17 insertions(+), 12 deletions(-)

diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
index 03c7b93f91..08e7c97150 100644
--- a/source/libs/stream/src/streamMeta.c
+++ b/source/libs/stream/src/streamMeta.c
@@ -151,7 +151,7 @@ int32_t streamMetaCheckBackendCompatible(SStreamMeta* pMeta) {
   int8_t ret = STREAM_STATA_COMPATIBLE;
   TBC*   pCur = NULL;
 
-  if (tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL) < 0) { // no task info, no stream
+  if (tdbTbcOpen(pMeta->pTaskDb, &pCur, NULL) < 0) {  // no task info, no stream
     return ret;
   }
 
@@ -557,7 +557,7 @@ int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
     return -1;
   }
 
-  if (pTask->ver < SSTREAM_TASK_SUBTABLE_CHANGED_VER){
+  if (pTask->ver < SSTREAM_TASK_SUBTABLE_CHANGED_VER) {
     pTask->ver = SSTREAM_TASK_VER;
   }
 
@@ -907,7 +907,7 @@ void streamMetaLoadAllTasks(SStreamMeta* pMeta) {
     if (p == NULL) {
       code = pMeta->buildTaskFn(pMeta->ahandle, pTask, pTask->chkInfo.checkpointVer + 1);
       if (code < 0) {
-        stError("failed to expand s-task:0x%"PRIx64", code:%s, continue", id.taskId, tstrerror(terrno));
+        stError("failed to expand s-task:0x%" PRIx64 ", code:%s, continue", id.taskId, tstrerror(terrno));
         tFreeStreamTask(pTask);
         continue;
       }
@@ -958,6 +958,8 @@ void streamMetaLoadAllTasks(SStreamMeta* pMeta) {
           pMeta->numOfStreamTasks, pMeta->numOfPausedTasks);
 
   taosArrayDestroy(pRecycleList);
+
+  (void)streamMetaCommit(pMeta);
 }
 
 static bool waitForEnoughDuration(SMetaHbInfo* pInfo) {
@@ -1012,7 +1014,7 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
   for (int32_t i = 0; i < numOfTasks; ++i) {
     SStreamTaskId* pId = taosArrayGet(pMeta->pTaskList, i);
 
-    STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
+    STaskId       id = {.streamId = pId->streamId, .taskId = pId->taskId};
     SStreamTask** pTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
     if (pTask == NULL) {
       continue;
@@ -1052,12 +1054,14 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
     }
 
     if ((*pTask)->chkInfo.pActiveInfo->activeId != 0) {
-      entry.checkpointInfo.failed = ((*pTask)->chkInfo.pActiveInfo->failedId >= (*pTask)->chkInfo.pActiveInfo->activeId) ? 1 : 0;
+      entry.checkpointInfo.failed =
+          ((*pTask)->chkInfo.pActiveInfo->failedId >= (*pTask)->chkInfo.pActiveInfo->activeId) ? 1 : 0;
       entry.checkpointInfo.activeId = (*pTask)->chkInfo.pActiveInfo->activeId;
       entry.checkpointInfo.activeTransId = (*pTask)->chkInfo.pActiveInfo->transId;
 
       if (entry.checkpointInfo.failed) {
-        stInfo("s-task:%s set kill checkpoint trans in hb, transId:%d", (*pTask)->id.idStr, (*pTask)->chkInfo.pActiveInfo->transId);
+        stInfo("s-task:%s set kill checkpoint trans in hb, transId:%d", (*pTask)->id.idStr,
+               (*pTask)->chkInfo.pActiveInfo->transId);
       }
     }
 
@@ -1384,7 +1388,7 @@ int32_t streamMetaStartAllTasks(SStreamMeta* pMeta, __stream_task_expand_fn expa
   int64_t now = taosGetTimestampMs();
 
   int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);
-  stInfo("vgId:%d start to check all %d stream task(s) downstream status, start ts:%"PRId64, vgId, numOfTasks, now);
+  stInfo("vgId:%d start to check all %d stream task(s) downstream status, start ts:%" PRId64, vgId, numOfTasks, now);
 
   if (numOfTasks == 0) {
     stInfo("vgId:%d no tasks to be started", pMeta->vgId);
@@ -1513,8 +1517,8 @@ bool streamMetaAllTasksReady(const SStreamMeta* pMeta) {
   int32_t num = taosArrayGetSize(pMeta->pTaskList);
   for (int32_t i = 0; i < num; ++i) {
     SStreamTaskId* pId = taosArrayGet(pMeta->pTaskList, i);
-    STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
-    SStreamTask** ppTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
+    STaskId        id = {.streamId = pId->streamId, .taskId = pId->taskId};
+    SStreamTask**  ppTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
     if (ppTask == NULL) {
       continue;
     }
@@ -1598,7 +1602,7 @@ int32_t streamMetaAddTaskLaunchResult(SStreamMeta* pMeta, int64_t streamId, int3
   if (pStartInfo->startAllTasks != 1) {
     int64_t el = endTs - startTs;
     stDebug("vgId:%d not start all task(s), not record status, s-task:0x%x launch succ:%d elapsed time:%" PRId64 "ms",
-           pMeta->vgId, taskId, ready, el);
+            pMeta->vgId, taskId, ready, el);
     streamMetaWUnLock(pMeta);
     return 0;
   }
@@ -1725,7 +1729,8 @@ void streamMetaAddIntoUpdateTaskList(SStreamMeta* pMeta, SStreamTask* pTask, SSt
     taosHashPut(pMeta->updateInfo.pTasks, &hEntry, sizeof(hEntry), NULL, 0);
 
     stDebug("s-task:%s vgId:%d transId:%d task nodeEp update completed, streamTask/hTask closed, elapsed:%" PRId64
-            " ms", id, vgId, transId, el);
+            " ms",
+            id, vgId, transId, el);
   } else {
     stDebug("s-task:%s vgId:%d transId:%d task nodeEp update completed, streamTask closed, elapsed time:%" PRId64 "ms",
             id, vgId, transId, el);
@@ -1740,4 +1745,4 @@ void streamMetaClearUpdateTaskList(SStreamMeta* pMeta) {
 void streamMetaInitUpdateTaskList(SStreamMeta* pMeta, int32_t transId) {
   taosHashClear(pMeta->updateInfo.pTasks);
   pMeta->updateInfo.transId = transId;
-}
\ No newline at end of file
+}