From 67853b4a5461d72739f49f853f1f6abaa3f85b12 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 26 Jan 2025 18:19:37 +0800
Subject: [PATCH 01/16] enh(stream): add checkpoint queue for source tasks.

---
 include/libs/stream/tstream.h        |   1 +
 source/libs/stream/inc/streamInt.h   |   2 +
 source/libs/stream/src/streamQueue.c | 136 ++++++++++++++++++++++-----
 3 files changed, 113 insertions(+), 26 deletions(-)

diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index c4c0aaf742..5188e3c667 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -136,6 +136,7 @@ enum {
 enum {
   STREAM_QUEUE__SUCESS = 1,
   STREAM_QUEUE__FAILED,
+  STREAM_QUEUE__CHKPTFAILED,
   STREAM_QUEUE__PROCESSING,
 };
 
diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h
index d9778a6a05..0e5aaac58c 100644
--- a/source/libs/stream/inc/streamInt.h
+++ b/source/libs/stream/inc/streamInt.h
@@ -144,6 +144,8 @@ struct SStreamQueue {
   STaosQall*  qall;
   void*       qItem;
   int8_t      status;
+  STaosQueue* pChkptQueue;
+  void*       qChkptItem;
 };
 
 struct SStreamQueueItem {
diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 401aa7530d..2a1332c0c4 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -47,7 +47,9 @@ static void streamQueueCleanup(SStreamQueue* pQueue) {
 
 int32_t streamQueueOpen(int64_t cap, SStreamQueue** pQ) {
   *pQ = NULL;
+
   int32_t code = 0;
+  int32_t lino = 0;
 
   SStreamQueue* pQueue = taosMemoryCalloc(1, sizeof(SStreamQueue));
   if (pQueue == NULL) {
@@ -55,24 +57,26 @@ int32_t streamQueueOpen(int64_t cap, SStreamQueue** pQ) {
   }
 
   code = taosOpenQueue(&pQueue->pQueue);
-  if (code) {
-    taosMemoryFreeClear(pQueue);
-    return code;
-  }
+  TSDB_CHECK_CODE(code, lino, _error);
 
   code = taosAllocateQall(&pQueue->qall);
-  if (code) {
-    taosCloseQueue(pQueue->pQueue);
-    taosMemoryFree(pQueue);
-    return code;
-  }
+  TSDB_CHECK_CODE(code, lino, _error);
+
+  code = taosOpenQueue(&pQueue->pChkptQueue);
+  TSDB_CHECK_CODE(code, lino, _error);
 
   pQueue->status = STREAM_QUEUE__SUCESS;
+
   taosSetQueueCapacity(pQueue->pQueue, cap);
   taosSetQueueMemoryCapacity(pQueue->pQueue, cap * 1024);
 
   *pQ = pQueue;
   return code;
+
+_error:
+  streamQueueClose(pQueue, 0);
+  stError("failed to open stream queue at line:%d, code:%s", lino, tstrerror(code));
+  return code;
 }
 
 void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) {
@@ -82,6 +86,11 @@ void streamQueueClose(SStreamQueue* pQueue, int32_t taskId) {
 
   taosFreeQall(pQueue->qall);
   taosCloseQueue(pQueue->pQueue);
+  pQueue->pQueue = NULL;
+
+  taosCloseQueue(pQueue->pChkptQueue);
+  pQueue->pChkptQueue = NULL;
+
   taosMemoryFree(pQueue);
 }
 
@@ -94,6 +103,46 @@ void streamQueueNextItem(SStreamQueue* pQueue, SStreamQueueItem** pItem) {
   } else {
     pQueue->qItem = NULL;
     (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
+
+    if (pQueue->qItem == NULL) {
+      (void) taosReadAllQitems(pQueue->pQueue, pQueue->qall);
+      (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
+    }
+
+    *pItem = streamQueueCurItem(pQueue);
+  }
+}
+
+void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem, ETaskStatus status) {
+  *pItem = NULL;
+  int8_t flag = atomic_exchange_8(&pQueue->status, STREAM_QUEUE__PROCESSING);
+
+  if (flag == STREAM_QUEUE__CHKPTFAILED) {
+    *pItem = pQueue->qChkptItem;
+  } else {
+    pQueue->qChkptItem = NULL;
+    taosReadQitem(pQueue->pChkptQueue, (void**) &pQueue->qChkptItem);
+    if (pQueue->qChkptItem != NULL) {
+      stDebug("read data from checkpoint queue, status:%d", status);
+
+      *pItem = pQueue->qChkptItem;
+      return;
+    }
+
+    // if in checkpoint status, not read data from ordinary input q.
+    if (status == TASK_STATUS__CK) {
+      stDebug("in checkpoint status, not ready data in normal queue");
+      return;
+    }
+  }
+
+  // let's try the ordinary input q
+  if (flag == STREAM_QUEUE__FAILED) {
+    *pItem = streamQueueCurItem(pQueue);
+  } else {
+    pQueue->qItem = NULL;
+    (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
+
     if (pQueue->qItem == NULL) {
       (void) taosReadAllQitems(pQueue->pQueue, pQueue->qall);
       (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
@@ -110,6 +159,7 @@ void streamQueueProcessSuccess(SStreamQueue* queue) {
   }
 
   queue->qItem = NULL;
+  queue->qChkptItem = NULL;
   atomic_store_8(&queue->status, STREAM_QUEUE__SUCESS);
 }
 
@@ -121,6 +171,14 @@ void streamQueueProcessFail(SStreamQueue* queue) {
   atomic_store_8(&queue->status, STREAM_QUEUE__FAILED);
 }
 
+void streamQueueGetSourceChkptFailed(SStreamQueue* pQueue) {
+  if (atomic_load_8(&pQueue->status) != STREAM_QUEUE__PROCESSING) {
+    stError("invalid queue status:%d, expect:%d", atomic_load_8(&pQueue->status), STREAM_QUEUE__PROCESSING);
+    return;
+  }
+  atomic_store_8(&pQueue->status, STREAM_QUEUE__CHKPTFAILED);
+}
+
 bool streamQueueIsFull(const SStreamQueue* pQueue) {
   int32_t numOfItems = streamQueueGetNumOfItems(pQueue);
   if (numOfItems >= STREAM_TASK_QUEUE_CAPACITY) {
@@ -175,8 +233,9 @@ const char* streamQueueItemGetTypeStr(int32_t type) {
 
 EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInput, int32_t* numOfBlocks,
                                              int32_t* blockSize) {
-  const char* id = pTask->id.idStr;
-  int32_t     taskLevel = pTask->info.taskLevel;
+  const char*   id = pTask->id.idStr;
+  int32_t       taskLevel = pTask->info.taskLevel;
+  SStreamQueue* pQueue = pTask->inputq.queue;
 
   *pInput = NULL;
   *numOfBlocks = 0;
@@ -189,13 +248,19 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
   }
 
   while (1) {
-    if (streamTaskShouldPause(pTask) || streamTaskShouldStop(pTask)) {
-      stDebug("s-task:%s task should pause, extract input blocks:%d", id, *numOfBlocks);
+    ETaskStatus status = streamTaskGetStatus(pTask).state;
+    if (status == TASK_STATUS__PAUSE || status == TASK_STATUS__STOP || status == TASK_STATUS__STOP) {
+      stDebug("s-task:%s task should pause/stop, extract input blocks:%d", id, *numOfBlocks);
       return EXEC_CONTINUE;
     }
 
     SStreamQueueItem* qItem = NULL;
-    streamQueueNextItem(pTask->inputq.queue, (SStreamQueueItem**)&qItem);
+    if (taskLevel == TASK_LEVEL__SOURCE) {
+      streamQueueNextItemInSourceQ(pQueue, &qItem, status);
+    } else {
+      streamQueueNextItem(pQueue, &qItem);
+    }
+
     if (qItem == NULL) {
       // restore the token to bucket
       if (*numOfBlocks > 0) {
@@ -225,14 +290,19 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
         *numOfBlocks = 1;
         *pInput = qItem;
         return EXEC_CONTINUE;
-      } else {  // previous existed blocks needs to be handle, before handle the checkpoint msg block
+      } else {  // previous existed blocks needs to be handled, before handle the checkpoint msg block
         stDebug("s-task:%s %s msg extracted, handle previous blocks, numOfBlocks:%d", id, p, *numOfBlocks);
         *blockSize = streamQueueItemGetSize(*pInput);
         if (taskLevel == TASK_LEVEL__SINK) {
           streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);
         }
 
-        streamQueueProcessFail(pTask->inputq.queue);
+        if ((type == STREAM_INPUT__CHECKPOINT_TRIGGER || type == STREAM_INPUT__CHECKPOINT) &&
+            (taskLevel == TASK_LEVEL__SOURCE)) {
+          streamQueueGetSourceChkptFailed(pQueue);
+        } else {
+          streamQueueProcessFail(pQueue);
+        }
         return EXEC_CONTINUE;
       }
     } else {
@@ -252,7 +322,7 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
             streamTaskConsumeQuota(pTask->outputInfo.pTokenBucket, *blockSize);
           }
 
-          streamQueueProcessFail(pTask->inputq.queue);
+          streamQueueProcessFail(pQueue);
           return EXEC_CONTINUE;
         }
 
@@ -260,7 +330,7 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
       }
 
       *numOfBlocks += 1;
-      streamQueueProcessSuccess(pTask->inputq.queue);
+      streamQueueProcessSuccess(pQueue);
 
       if (*numOfBlocks >= MAX_STREAM_EXEC_BATCH_NUM) {
         stDebug("s-task:%s batch size limit:%d reached, start to process blocks", id, MAX_STREAM_EXEC_BATCH_NUM);
@@ -279,6 +349,7 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
 int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem) {
   int8_t      type = pItem->type;
   STaosQueue* pQueue = pTask->inputq.queue->pQueue;
+  int32_t     level = pTask->info.taskLevel;
   int32_t     total = streamQueueGetNumOfItems(pTask->inputq.queue) + 1;
 
   if (type == STREAM_INPUT__DATA_SUBMIT) {
@@ -326,15 +397,28 @@ int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem)
     stDebug("s-task:%s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr, total, size);
   } else if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__CHECKPOINT_TRIGGER ||
              type == STREAM_INPUT__TRANS_STATE || type == STREAM_INPUT__DATA_RETRIEVE) {
-    int32_t code = taosWriteQitem(pQueue, pItem);
-    if (code != TSDB_CODE_SUCCESS) {
-      streamFreeQitem(pItem);
-      return code;
-    }
 
-    double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
-    stDebug("s-task:%s level:%d %s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr,
-            pTask->info.taskLevel, streamQueueItemGetTypeStr(type), total, size);
+    int32_t code = 0;
+    if ((type == STREAM_INPUT__CHECKPOINT_TRIGGER || type == STREAM_INPUT__CHECKPOINT) && level == TASK_LEVEL__SOURCE) {
+      STaosQueue* pChkptQ = pTask->inputq.queue->pChkptQueue;
+      code = taosWriteQitem(pChkptQ, pItem);
+
+      double  size = SIZE_IN_MiB(taosQueueMemorySize(pChkptQ));
+      int32_t num = taosQueueItemSize(pChkptQ);
+
+      stDebug("s-task:%s level:%d %s checkpoint enqueue ctrl queue, total in queue:%d, size:%.2fMiB, data queue:%d",
+              pTask->id.idStr, pTask->info.taskLevel, streamQueueItemGetTypeStr(type), num, size, (total - 1));
+    } else {
+      code = taosWriteQitem(pQueue, pItem);
+      if (code != TSDB_CODE_SUCCESS) {
+        streamFreeQitem(pItem);
+        return code;
+      }
+
+      double size = SIZE_IN_MiB(taosQueueMemorySize(pQueue));
+      stDebug("s-task:%s level:%d %s blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr,
+              pTask->info.taskLevel, streamQueueItemGetTypeStr(type), total, size);
+    }
   } else if (type == STREAM_INPUT__GET_RES) {
     // use the default memory limit, refactor later.
     int32_t code = taosWriteQitem(pQueue, pItem);

From 24c6fe727ff72ce32ff771b168d02f35cdfdc547 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 26 Jan 2025 21:50:51 +0800
Subject: [PATCH 02/16] refactor(stream): do some internal refactor.

---
 source/libs/stream/src/streamQueue.c | 52 +++++++++++++++-------------
 1 file changed, 28 insertions(+), 24 deletions(-)

diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 2a1332c0c4..6a9a1ac880 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -119,37 +119,41 @@ void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem
 
   if (flag == STREAM_QUEUE__CHKPTFAILED) {
     *pItem = pQueue->qChkptItem;
-  } else {
-    pQueue->qChkptItem = NULL;
-    taosReadQitem(pQueue->pChkptQueue, (void**) &pQueue->qChkptItem);
-    if (pQueue->qChkptItem != NULL) {
-      stDebug("read data from checkpoint queue, status:%d", status);
+    ASSERT(status != TASK_STATUS__CK && pQueue->qItem == NULL);
+    return;
+  }
 
-      *pItem = pQueue->qChkptItem;
-      return;
-    }
+  if (flag == STREAM_QUEUE__FAILED) {
+    *pItem = pQueue->qItem;
+    ASSERT(status != TASK_STATUS__CK && pQueue->qChkptItem == NULL);
+    return;
+  }
 
-    // if in checkpoint status, not read data from ordinary input q.
-    if (status == TASK_STATUS__CK) {
-      stDebug("in checkpoint status, not ready data in normal queue");
-      return;
-    }
+  pQueue->qChkptItem = NULL;
+  taosReadQitem(pQueue->pChkptQueue, (void**)&pQueue->qChkptItem);
+  if (pQueue->qChkptItem != NULL) {
+    stDebug("read data from checkpoint queue, status:%d", status);
+
+    *pItem = pQueue->qChkptItem;
+    return;
+  }
+
+  // if in checkpoint status, not read data from ordinary input q.
+  if (status == TASK_STATUS__CK) {
+    stDebug("in checkpoint status, not ready data in normal queue");
+    return;
   }
 
   // let's try the ordinary input q
-  if (flag == STREAM_QUEUE__FAILED) {
-    *pItem = streamQueueCurItem(pQueue);
-  } else {
-    pQueue->qItem = NULL;
-    (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
+  pQueue->qItem = NULL;
+  (void)taosGetQitem(pQueue->qall, &pQueue->qItem);
 
-    if (pQueue->qItem == NULL) {
-      (void) taosReadAllQitems(pQueue->pQueue, pQueue->qall);
-      (void) taosGetQitem(pQueue->qall, &pQueue->qItem);
-    }
-
-    *pItem = streamQueueCurItem(pQueue);
+  if (pQueue->qItem == NULL) {
+    (void)taosReadAllQitems(pQueue->pQueue, pQueue->qall);
+    (void)taosGetQitem(pQueue->qall, &pQueue->qItem);
   }
+
+  *pItem = streamQueueCurItem(pQueue);
 }
 
 void streamQueueProcessSuccess(SStreamQueue* queue) {

From f56aeaf1bd06d3a60da5fa0a7d929257d1da2348 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 26 Jan 2025 22:57:06 +0800
Subject: [PATCH 03/16] fix(stream): fix error in deciding exec should quit or
 not.

---
 source/libs/stream/src/streamExec.c  | 28 +++++++++++++++++++++++++---
 source/libs/stream/src/streamQueue.c |  9 ++++-----
 2 files changed, 29 insertions(+), 8 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 054f88ec3d..ef3fb11866 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -915,8 +915,31 @@ bool streamTaskReadyToRun(const SStreamTask* pTask, char** pStatus) {
   }
 }
 
+static bool shouldNotCont(SStreamTask* pTask) {
+  int32_t       level = pTask->info.taskLevel;
+  SStreamQueue* pQueue = pTask->inputq.queue;
+  ETaskStatus   status = streamTaskGetStatus(pTask).state;
+
+  // 1. task should jump out
+  bool quit = (status == TASK_STATUS__STOP) || (status == TASK_STATUS__PAUSE) || (status == TASK_STATUS__DROPPING);
+
+  // 2. checkpoint procedure, the source task's checkpoint queue is empty, not read from ordinary queue
+  bool notCkCont =
+      (taosQueueItemSize(pQueue->pChkptQueue) == 0) && (level == TASK_LEVEL__SOURCE) && (status == TASK_STATUS__CK);
+
+  // 3. no data in ordinary queue
+  int32_t numOfItems = streamQueueGetNumOfItems(pQueue);
+
+  if ((numOfItems == 0) || quit || notCkCont) {
+    return true;
+  } else {
+    return false;
+  }
+}
+
 int32_t streamResumeTask(SStreamTask* pTask) {
   const char* id = pTask->id.idStr;
+  int32_t     level = pTask->info.taskLevel;
   int32_t     code = 0;
 
   if (pTask->status.schedStatus != TASK_SCHED_STATUS__ACTIVE) {
@@ -929,11 +952,10 @@ int32_t streamResumeTask(SStreamTask* pTask) {
     if (code) {
       stError("s-task:%s failed to exec stream task, code:%s, continue", id, tstrerror(code));
     }
-    // check if continue
+
     streamMutexLock(&pTask->lock);
 
-    int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);
-    if ((numOfItems == 0) || streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) {
+    if (shouldNotCont(pTask)) {
       atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
       streamTaskClearSchedIdleInfo(pTask);
       streamMutexUnlock(&pTask->lock);
diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 6a9a1ac880..3d0d01ce0e 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -113,7 +113,7 @@ void streamQueueNextItem(SStreamQueue* pQueue, SStreamQueueItem** pItem) {
   }
 }
 
-void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem, ETaskStatus status) {
+void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem, ETaskStatus status, const char* id) {
   *pItem = NULL;
   int8_t flag = atomic_exchange_8(&pQueue->status, STREAM_QUEUE__PROCESSING);
 
@@ -132,15 +132,14 @@ void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem
   pQueue->qChkptItem = NULL;
   taosReadQitem(pQueue->pChkptQueue, (void**)&pQueue->qChkptItem);
   if (pQueue->qChkptItem != NULL) {
-    stDebug("read data from checkpoint queue, status:%d", status);
-
+    stDebug("s-task:%s read data from checkpoint queue, status:%d", id, status);
     *pItem = pQueue->qChkptItem;
     return;
   }
 
   // if in checkpoint status, not read data from ordinary input q.
   if (status == TASK_STATUS__CK) {
-    stDebug("in checkpoint status, not ready data in normal queue");
+    stDebug("s-task:%s in checkpoint status, not read data in block queue, status:%d", id, status);
     return;
   }
 
@@ -260,7 +259,7 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
 
     SStreamQueueItem* qItem = NULL;
     if (taskLevel == TASK_LEVEL__SOURCE) {
-      streamQueueNextItemInSourceQ(pQueue, &qItem, status);
+      streamQueueNextItemInSourceQ(pQueue, &qItem, status, id);
     } else {
       streamQueueNextItem(pQueue, &qItem);
     }

From ed03b3a22cc3e70e7919e9d81c61d462b32893fb Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 27 Jan 2025 00:50:36 +0800
Subject: [PATCH 04/16] fix(stream): try starting task even the inputQ is full.

---
 source/dnode/vnode/src/tq/tqStreamTask.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
index b34ea78f64..88bbc30b7b 100644
--- a/source/dnode/vnode/src/tq/tqStreamTask.c
+++ b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -317,9 +317,10 @@ bool taskReadyForDataFromWal(SStreamTask* pTask) {
     return false;
   }
 
-  // check if input queue is full or not
+  // check whether input queue is full or not
   if (streamQueueIsFull(pTask->inputq.queue)) {
-    tqTrace("s-task:%s input queue is full, do nothing", pTask->id.idStr);
+    tqTrace("s-task:%s input queue is full, launch task without scanning wal", pTask->id.idStr);
+    streamTrySchedExec(pTask);
     return false;
   }
 

From cb9a1a852d58aee1bb743e1cf1cd901ce03e28da Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Tue, 28 Jan 2025 01:38:16 +0800
Subject: [PATCH 05/16] refactor: update some logs.

---
 source/libs/stream/src/streamDispatch.c | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
index b7039d372d..b093f808c0 100644
--- a/source/libs/stream/src/streamDispatch.c
+++ b/source/libs/stream/src/streamDispatch.c
@@ -1098,6 +1098,8 @@ static void chkptReadyMsgSendMonitorFn(void* param, void* tmrId) {
   pActiveInfo = pTask->chkInfo.pActiveInfo;
   pTmrInfo = &pActiveInfo->chkptReadyMsgTmr;
 
+  stDebug("s-task:%s acquire task, refId:%" PRId64, id, taskRefId);
+
   // check the status every 100ms
   if (streamTaskShouldStop(pTask)) {
     streamCleanBeforeQuitTmr(pTmrInfo, param);

From 34a439d78bb322ad2ec5c3814eae072053855430 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 5 Feb 2025 10:59:29 +0800
Subject: [PATCH 06/16] fix(stream): fix error in the check of continuing
 execution condition.

---
 source/libs/stream/src/streamExec.c | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index ef3fb11866..3555515f75 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -924,16 +924,20 @@ static bool shouldNotCont(SStreamTask* pTask) {
   bool quit = (status == TASK_STATUS__STOP) || (status == TASK_STATUS__PAUSE) || (status == TASK_STATUS__DROPPING);
 
   // 2. checkpoint procedure, the source task's checkpoint queue is empty, not read from ordinary queue
-  bool notCkCont =
-      (taosQueueItemSize(pQueue->pChkptQueue) == 0) && (level == TASK_LEVEL__SOURCE) && (status == TASK_STATUS__CK);
+  bool emptyCkQueue = (taosQueueItemSize(pQueue->pChkptQueue) == 0) && (level == TASK_LEVEL__SOURCE);
 
   // 3. no data in ordinary queue
-  int32_t numOfItems = streamQueueGetNumOfItems(pQueue);
+  int32_t emptyBlockQueue = (streamQueueGetNumOfItems(pQueue) == 0);
 
-  if ((numOfItems == 0) || quit || notCkCont) {
+  if (quit) {
     return true;
   } else {
-    return false;
+    if (status == TASK_STATUS__CK) {
+      // in checkpoint procedure, we only check whether the controller queue is empty or not
+      return emptyCkQueue;
+    } else { // otherwise, if the block queue is empty, not continue.
+      return emptyBlockQueue;
+    }
   }
 }
 

From 732a9d4b32b9cb36be64b3bbe674e242f1a90307 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 5 Feb 2025 13:35:23 +0800
Subject: [PATCH 07/16] fix(stream): fix error in the check of continuing
 execution condition.

---
 source/libs/stream/src/streamExec.c  | 8 ++++----
 source/libs/stream/src/streamQueue.c | 2 +-
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 3555515f75..1015917f61 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -924,19 +924,19 @@ static bool shouldNotCont(SStreamTask* pTask) {
   bool quit = (status == TASK_STATUS__STOP) || (status == TASK_STATUS__PAUSE) || (status == TASK_STATUS__DROPPING);
 
   // 2. checkpoint procedure, the source task's checkpoint queue is empty, not read from ordinary queue
-  bool emptyCkQueue = (taosQueueItemSize(pQueue->pChkptQueue) == 0) && (level == TASK_LEVEL__SOURCE);
+  bool emptyCkQueue = (taosQueueItemSize(pQueue->pChkptQueue) == 0);
 
   // 3. no data in ordinary queue
-  int32_t emptyBlockQueue = (streamQueueGetNumOfItems(pQueue) == 0);
+  bool emptyBlockQueue = (streamQueueGetNumOfItems(pQueue) == 0);
 
   if (quit) {
     return true;
   } else {
-    if (status == TASK_STATUS__CK) {
+    if (status == TASK_STATUS__CK && level == TASK_LEVEL__SOURCE) {
       // in checkpoint procedure, we only check whether the controller queue is empty or not
       return emptyCkQueue;
     } else { // otherwise, if the block queue is empty, not continue.
-      return emptyBlockQueue;
+      return emptyBlockQueue && emptyCkQueue;
     }
   }
 }
diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 3d0d01ce0e..f68dd1452f 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -402,7 +402,7 @@ int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem)
              type == STREAM_INPUT__TRANS_STATE || type == STREAM_INPUT__DATA_RETRIEVE) {
 
     int32_t code = 0;
-    if ((type == STREAM_INPUT__CHECKPOINT_TRIGGER || type == STREAM_INPUT__CHECKPOINT) && level == TASK_LEVEL__SOURCE) {
+    if ((type == STREAM_INPUT__CHECKPOINT_TRIGGER || type == STREAM_INPUT__CHECKPOINT) && (level == TASK_LEVEL__SOURCE)) {
       STaosQueue* pChkptQ = pTask->inputq.queue->pChkptQueue;
       code = taosWriteQitem(pChkptQ, pItem);
 

From e60321196bf7f360ac5a1630648048a4f075371a Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 9 Feb 2025 22:29:41 +0800
Subject: [PATCH 08/16] fix(stream): remove invalid assert.

---
 source/libs/stream/src/streamQueue.c | 2 --
 1 file changed, 2 deletions(-)

diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index f68dd1452f..c7b0bc8f11 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -119,13 +119,11 @@ void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem
 
   if (flag == STREAM_QUEUE__CHKPTFAILED) {
     *pItem = pQueue->qChkptItem;
-    ASSERT(status != TASK_STATUS__CK && pQueue->qItem == NULL);
     return;
   }
 
   if (flag == STREAM_QUEUE__FAILED) {
     *pItem = pQueue->qItem;
-    ASSERT(status != TASK_STATUS__CK && pQueue->qChkptItem == NULL);
     return;
   }
 

From 3e55f8edfed67b32a5ccdcc584a329b7ba606ef2 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 23 Feb 2025 10:43:56 +0800
Subject: [PATCH 09/16] fix(stream): follower nodes not restart tasks.

---
 include/dnode/vnode/tqCommon.h             |  2 +-
 source/dnode/snode/src/snode.c             |  2 +-
 source/dnode/vnode/src/tq/tq.c             |  3 ++-
 source/dnode/vnode/src/tqCommon/tqCommon.c | 31 ++++++++++++++--------
 4 files changed, 24 insertions(+), 14 deletions(-)

diff --git a/include/dnode/vnode/tqCommon.h b/include/dnode/vnode/tqCommon.h
index 4d5e18520c..d4ca0aba62 100644
--- a/include/dnode/vnode/tqCommon.h
+++ b/include/dnode/vnode/tqCommon.h
@@ -19,7 +19,7 @@
 // message process
 int32_t tqStreamTaskStartAsync(SStreamMeta* pMeta, SMsgCb* cb, bool restart);
 int32_t tqStreamStartOneTaskAsync(SStreamMeta* pMeta, SMsgCb* cb, int64_t streamId, int32_t taskId);
-int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pMsg, bool restored);
+int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pMsg, bool restored, bool isLeader);
 int32_t tqStreamTaskProcessDispatchReq(SStreamMeta* pMeta, SRpcMsg* pMsg);
 int32_t tqStreamTaskProcessDispatchRsp(SStreamMeta* pMeta, SRpcMsg* pMsg);
 int32_t tqStreamTaskProcessRetrieveReq(SStreamMeta* pMeta, SRpcMsg* pMsg);
diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c
index 6eee8c510b..b15ac447b0 100644
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
@@ -157,7 +157,7 @@ int32_t sndProcessWriteMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg *pRsp) {
     case TDMT_STREAM_TASK_DROP:
       return tqStreamTaskProcessDropReq(pSnode->pMeta, pMsg->pCont, pMsg->contLen);
     case TDMT_VND_STREAM_TASK_UPDATE:
-      return tqStreamTaskProcessUpdateReq(pSnode->pMeta, &pSnode->msgCb, pMsg, true);
+      return tqStreamTaskProcessUpdateReq(pSnode->pMeta, &pSnode->msgCb, pMsg, true, true);
     case TDMT_VND_STREAM_TASK_RESET:
       return tqStreamTaskProcessTaskResetReq(pSnode->pMeta, pMsg->pCont);
     case TDMT_STREAM_TASK_PAUSE:
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 63727e5c45..e1298b11ea 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1364,7 +1364,8 @@ int32_t tqProcessTaskCheckpointReadyMsg(STQ* pTq, SRpcMsg* pMsg) {
 }
 
 int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg) {
-  return tqStreamTaskProcessUpdateReq(pTq->pStreamMeta, &pTq->pVnode->msgCb, pMsg, pTq->pVnode->restored);
+  return tqStreamTaskProcessUpdateReq(pTq->pStreamMeta, &pTq->pVnode->msgCb, pMsg,
+                                      pTq->pVnode->restored, (pTq->pStreamMeta->role == NODE_ROLE_LEADER));
 }
 
 int32_t tqProcessTaskResetReq(STQ* pTq, SRpcMsg* pMsg) {
diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c
index 2c48ada0fa..fb3582c5ff 100644
--- a/source/dnode/vnode/src/tqCommon/tqCommon.c
+++ b/source/dnode/vnode/src/tqCommon/tqCommon.c
@@ -139,7 +139,7 @@ int32_t tqStreamStartOneTaskAsync(SStreamMeta* pMeta, SMsgCb* cb, int64_t stream
 }
 
 // this is to process request from transaction, always return true.
-int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pMsg, bool restored) {
+int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pMsg, bool restored, bool isLeader) {
   int32_t      vgId = pMeta->vgId;
   char*        msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t      len = pMsg->contLen - sizeof(SMsgHead);
@@ -298,14 +298,19 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
   int32_t numOfTasks = streamMetaGetNumOfTasks(pMeta);
   int32_t updateTasks = taosHashGetSize(pMeta->updateInfo.pTasks);
 
-  if (restored) {
+  if (restored && isLeader) {
     tqDebug("vgId:%d s-task:0x%x update epset transId:%d, set the restart flag", vgId, req.taskId, req.transId);
     pMeta->startInfo.tasksWillRestart = 1;
   }
 
   if (updateTasks < numOfTasks) {
-    tqDebug("vgId:%d closed tasks:%d, unclosed:%d, all tasks will be started when nodeEp update completed", vgId,
-            updateTasks, (numOfTasks - updateTasks));
+    if (isLeader) {
+      tqDebug("vgId:%d closed tasks:%d, unclosed:%d, all tasks will be started when nodeEp update completed", vgId,
+              updateTasks, (numOfTasks - updateTasks));
+    } else {
+      tqDebug("vgId:%d closed tasks:%d, unclosed:%d, follower not restart tasks", vgId, updateTasks,
+              (numOfTasks - updateTasks));
+    }
   } else {
     if ((code = streamMetaCommit(pMeta)) < 0) {
       // always return true
@@ -316,17 +321,21 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
 
     streamMetaClearSetUpdateTaskListComplete(pMeta);
 
-    if (!restored) {
-      tqDebug("vgId:%d vnode restore not completed, not start all tasks", vgId);
-    } else {
-      tqDebug("vgId:%d all %d task(s) nodeEp updated and closed, transId:%d", vgId, numOfTasks, req.transId);
+    if (isLeader) {
+      if (!restored) {
+        tqDebug("vgId:%d vnode restore not completed, not start all tasks", vgId);
+      } else {
+        tqDebug("vgId:%d all %d task(s) nodeEp updated and closed, transId:%d", vgId, numOfTasks, req.transId);
 #if 0
       taosMSleep(5000);// for test purpose, to trigger the leader election
 #endif
-      code = tqStreamTaskStartAsync(pMeta, cb, true);
-      if (code) {
-        tqError("vgId:%d async start all tasks, failed, code:%s", vgId, tstrerror(code));
+        code = tqStreamTaskStartAsync(pMeta, cb, true);
+        if (code) {
+          tqError("vgId:%d async start all tasks, failed, code:%s", vgId, tstrerror(code));
+        }
       }
+    } else {
+      tqDebug("vgId:%d follower nodes not restart tasks", vgId);
     }
   }
 

From 14ef25101dddd37307b6fc75d6ee10f84eb887e5 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 23 Feb 2025 15:37:59 +0800
Subject: [PATCH 10/16] fix(stream): adjust log.

---
 source/dnode/vnode/src/tq/tqStreamTask.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
index 88bbc30b7b..57579eecb4 100644
--- a/source/dnode/vnode/src/tq/tqStreamTask.c
+++ b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -131,7 +131,7 @@ static void doStartScanWal(void* param, void* tmrId) {
   }
 
   if (pMeta->startInfo.startAllTasks) {
-    tqTrace("vgId:%d in restart procedure, not ready to scan wal", vgId);
+    tqDebug("vgId:%d in restart procedure, not ready to scan wal", vgId);
     goto _end;
   }
 

From 951b469b10c6a0e306662a35d904b9638df611bb Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 23 Feb 2025 16:18:19 +0800
Subject: [PATCH 11/16] fix(stream): add some logs.

---
 source/dnode/vnode/src/tq/tqStreamTask.c | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
index 57579eecb4..1e021a8d5b 100644
--- a/source/dnode/vnode/src/tq/tqStreamTask.c
+++ b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -87,6 +87,8 @@ static void doStartScanWal(void* param, void* tmrId) {
   tmr_h                  pTimer = NULL;
   SBuildScanWalMsgParam* pParam = (SBuildScanWalMsgParam*)param;
 
+  tqDebug("start to do scan wal in tmr, metaRid:%" PRId64, pParam->metaId);
+
   SStreamMeta* pMeta = taosAcquireRef(streamMetaRefPool, pParam->metaId);
   if (pMeta == NULL) {
     tqError("metaRid:%" PRId64 " not valid now, stream meta has been freed", pParam->metaId);

From 6e1206bbc11ef6e2bb0f7da5284c9a86bc14b731 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 23 Feb 2025 16:18:49 +0800
Subject: [PATCH 12/16] fix(stream): add some logs.

---
 source/dnode/vnode/src/tq/tqStreamTask.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
index 1e021a8d5b..c8862b03ad 100644
--- a/source/dnode/vnode/src/tq/tqStreamTask.c
+++ b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -156,7 +156,7 @@ static void doStartScanWal(void* param, void* tmrId) {
     goto _end;
   }
 
-  tqTrace("vgId:%d create msg to start wal scan, numOfTasks:%d", vgId, numOfTasks);
+  tqDebug("vgId:%d create msg to start wal scan, numOfTasks:%d", vgId, numOfTasks);
 
    #if 0
   //  wait for the vnode is freed, and invalid read may occur.

From b23f20a4507c78673b848175946acc5c4a3c3576 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Tue, 25 Feb 2025 14:29:49 +0800
Subject: [PATCH 13/16] fix(stream): fix memory leak in checkpt queue.

---
 source/libs/stream/src/streamQueue.c | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index c7b0bc8f11..300b3bdb4d 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -32,11 +32,12 @@ typedef struct SQueueReader {
 static bool streamTaskExtractAvailableToken(STokenBucket* pBucket, const char* id);
 static void streamTaskPutbackToken(STokenBucket* pBucket);
 static void streamTaskConsumeQuota(STokenBucket* pBucket, int32_t bytes);
+static void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem, ETaskStatus status, const char* id);
 
 static void streamQueueCleanup(SStreamQueue* pQueue) {
   SStreamQueueItem* qItem = NULL;
   while (1) {
-    streamQueueNextItem(pQueue, &qItem);
+    streamQueueNextItemInSourceQ(pQueue, &qItem, TASK_STATUS__READY, NULL);
     if (qItem == NULL) {
       break;
     }
@@ -250,7 +251,7 @@ EExtractDataCode streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueIte
 
   while (1) {
     ETaskStatus status = streamTaskGetStatus(pTask).state;
-    if (status == TASK_STATUS__PAUSE || status == TASK_STATUS__STOP || status == TASK_STATUS__STOP) {
+    if (status == TASK_STATUS__PAUSE || status == TASK_STATUS__STOP) {
       stDebug("s-task:%s task should pause/stop, extract input blocks:%d", id, *numOfBlocks);
       return EXEC_CONTINUE;
     }

From 00e02280b81eed943e3843f3d8feb3b160b26cd7 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Fri, 28 Feb 2025 14:01:26 +0800
Subject: [PATCH 14/16] fix(stream): adjust the error code position.

---
 include/libs/stream/tstream.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 5188e3c667..0fa32acfbb 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -136,8 +136,8 @@ enum {
 enum {
   STREAM_QUEUE__SUCESS = 1,
   STREAM_QUEUE__FAILED,
-  STREAM_QUEUE__CHKPTFAILED,
   STREAM_QUEUE__PROCESSING,
+  STREAM_QUEUE__CHKPTFAILED,
 };
 
 typedef enum EStreamTaskEvent {

From 3892a98e2ee61619fadbcf1f30ff74c4f0c6a824 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Fri, 28 Feb 2025 14:09:45 +0800
Subject: [PATCH 15/16] fix(stream): check return values.

---
 source/libs/stream/src/streamQueue.c | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 300b3bdb4d..954e41f288 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -144,11 +144,21 @@ void streamQueueNextItemInSourceQ(SStreamQueue* pQueue, SStreamQueueItem** pItem
 
   // let's try the ordinary input q
   pQueue->qItem = NULL;
-  (void)taosGetQitem(pQueue->qall, &pQueue->qItem);
+  int32_t code = taosGetQitem(pQueue->qall, &pQueue->qItem);
+  if (code) {
+    stError("s-task:%s failed to get item in inputq, code:%s", id, tstrerror(code));
+  }
 
   if (pQueue->qItem == NULL) {
-    (void)taosReadAllQitems(pQueue->pQueue, pQueue->qall);
-    (void)taosGetQitem(pQueue->qall, &pQueue->qItem);
+    code = taosReadAllQitems(pQueue->pQueue, pQueue->qall);
+    if (code) {
+      stError("s-task:%s failed to get all items in inputq, code:%s", id, tstrerror(code));
+    }
+
+    code = taosGetQitem(pQueue->qall, &pQueue->qItem);
+    if (code) {
+      stError("s-task:%s failed to get item in inputq, code:%s", id, tstrerror(code));
+    }
   }
 
   *pItem = streamQueueCurItem(pQueue);

From 5f6ecab85443cd161ea13d39fc00911e0c6144ad Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Fri, 28 Feb 2025 15:05:00 +0800
Subject: [PATCH 16/16] fix(stream): check the return values.

---
 source/dnode/vnode/src/tq/tqStreamTask.c | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
index c8862b03ad..7c65f805e9 100644
--- a/source/dnode/vnode/src/tq/tqStreamTask.c
+++ b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -322,7 +322,10 @@ bool taskReadyForDataFromWal(SStreamTask* pTask) {
   // check whether input queue is full or not
   if (streamQueueIsFull(pTask->inputq.queue)) {
     tqTrace("s-task:%s input queue is full, launch task without scanning wal", pTask->id.idStr);
-    streamTrySchedExec(pTask);
+    int32_t code = streamTrySchedExec(pTask);
+    if (code) {
+      tqError("s-task:%s failed to start task while inputQ is full", pTask->id.idStr);
+    }
     return false;
   }