Merge pull request #4835 from taosdata/feature/sim

Feature/sim

src/mnode/src/mnodeSdb.c

@@ -1051,7 +1051,10 @@ static int32_t sdbWriteFwdToQueue(int32_t vgId, void *wparam, int32_t qtype, voi
   memcpy(pRow->pHead, pHead, sizeof(SWalHead) + pHead->len);
   pRow->rowData = pRow->pHead->cont;
 
-  return sdbWriteToQueue(pRow, qtype);
+  int32_t code = sdbWriteToQueue(pRow, qtype);
+  if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) code = 0;
+
+  return code;
 }
 
 static int32_t sdbWriteRowToQueue(SSdbRow *pInputRow, int32_t action) {

src/sync/src/syncRestore.c

@@ -195,7 +195,11 @@ static int32_t syncRestoreWal(SSyncPeer *pPeer, uint64_t *wver) {
     }
     lastVer = pHead->version;
 
-    (*pNode->writeToCache)(pNode->vgId, pHead, TAOS_QTYPE_WAL, NULL);
+    ret = (*pNode->writeToCache)(pNode->vgId, pHead, TAOS_QTYPE_WAL, NULL);
+    if (ret != 0) {
+      sError("%s, failed to restore record since %s, hver:%" PRIu64, pPeer->id, tstrerror(ret), pHead->version);
+      break;
+    }
   }
 
   if (code < 0) {

src/vnode/inc/vnodeStatus.h

@@ -37,6 +37,7 @@ bool vnodeSetResetStatus(SVnodeObj* pVnode);
 
 bool vnodeInInitStatus(SVnodeObj* pVnode);
 bool vnodeInReadyStatus(SVnodeObj* pVnode);
+bool vnodeInReadyOrUpdatingStatus(SVnodeObj* pVnode);
 bool vnodeInClosingStatus(SVnodeObj* pVnode);
 bool vnodeInResetStatus(SVnodeObj* pVnode);
 

src/vnode/src/vnodeStatus.c

@@ -135,6 +135,18 @@ bool vnodeInReadyStatus(SVnodeObj* pVnode) {
   return in;
 }
 
+bool vnodeInReadyOrUpdatingStatus(SVnodeObj* pVnode) {
+  bool in = false;
+  pthread_mutex_lock(&pVnode->statusMutex);
+
+  if (pVnode->status == TAOS_VN_STATUS_READY || pVnode->status == TAOS_VN_STATUS_UPDATING) {
+    in = true;
+  }
+
+  pthread_mutex_unlock(&pVnode->statusMutex);
+  return in;
+}
+
 bool vnodeInClosingStatus(SVnodeObj* pVnode) {
   bool in = false;
   pthread_mutex_lock(&pVnode->statusMutex);

src/vnode/src/vnodeWrite.c

@@ -245,17 +245,18 @@ static int32_t vnodeWriteToWQueueImp(SVWriteMsg *pWrite) {
   if (pWrite->qtype == TAOS_QTYPE_RPC) {
     int32_t code = vnodeCheckWrite(pVnode);
     if (code != TSDB_CODE_SUCCESS) {
+      vError("vgId:%d, failed to write into vwqueue since %s", pVnode->vgId, tstrerror(code));
       taosFreeQitem(pWrite);
       vnodeRelease(pVnode);
       return code;
     }
   }
 
-  if (!vnodeInReadyStatus(pVnode)) {
-    vDebug("vgId:%d, vnode status is %s, refCount:%d pVnode:%p", pVnode->vgId, vnodeStatus[pVnode->status],
-           pVnode->refCount, pVnode);
+  if (!vnodeInReadyOrUpdatingStatus(pVnode)) {
+    vError("vgId:%d, failed to write into vwqueue, vstatus is %s, refCount:%d pVnode:%p", pVnode->vgId,
+           vnodeStatus[pVnode->status], pVnode->refCount, pVnode);
     taosFreeQitem(pWrite);
-    vnodeRelease(pVnode);       
+    vnodeRelease(pVnode);
     return TSDB_CODE_APP_NOT_READY;
   }