enh: adjust size limit of appyQ and negotiationWin

include/util/taoserror.h

@@ -547,6 +547,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SYN_INVALID_SNAPSHOT_MSG      TAOS_DEF_ERROR_CODE(0, 0x0915)  // internal
 #define TSDB_CODE_SYN_BUFFER_FULL               TAOS_DEF_ERROR_CODE(0, 0x0916)
 #define TSDB_CODE_SYN_WRITE_STALL               TAOS_DEF_ERROR_CODE(0, 0x0917)
+#define TSDB_CODE_SYN_NEGO_WIN_EXCEEDED         TAOS_DEF_ERROR_CODE(0, 0X0918)
 #define TSDB_CODE_SYN_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x09FF)
 
 // tq

include/util/tdef.h

@@ -285,8 +285,11 @@ typedef enum ELogicConditionType {
 #define TSDB_DNODE_ROLE_VNODE 2
 
 #define TSDB_MAX_REPLICA               5
+
 #define TSDB_SYNC_LOG_BUFFER_SIZE      4096
-#define TSDB_SYNC_LOG_BUFFER_RETENTION (TSDB_SYNC_LOG_BUFFER_SIZE >> 4)
+#define TSDB_SYNC_LOG_BUFFER_RETENTION 256
+#define TSDB_SYNC_APPLYQ_SIZE_LIMIT    512
+#define TSDB_SYNC_NEGOTIATION_WIN      512
 
 #define TSDB_TBNAME_COLUMN_INDEX     (-1)
 #define TSDB_MULTI_TABLEMETA_MAX_NUM 100000  // maximum batch size allowed to load table meta

source/libs/sync/src/syncPipeline.c

@@ -53,8 +53,15 @@ int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt
     goto _err;
   }
 
+  if (pNode->restoreFinish && index - pBuf->commitIndex >= TSDB_SYNC_NEGOTIATION_WIN) {
+    terrno = TSDB_CODE_SYN_NEGO_WIN_EXCEEDED;
+    sError("vgId:%d, failed to append since %s, index:%" PRId64 ", commit-index:%" PRId64, pNode->vgId, terrstr(),
+           index, pBuf->commitIndex);
+    goto _err;
+  }
+
   SyncIndex appliedIndex = pNode->pFsm->FpAppliedIndexCb(pNode->pFsm);
-  if (pNode->restoreFinish && pBuf->commitIndex - appliedIndex >= pBuf->size) {
+  if (pNode->restoreFinish && pBuf->commitIndex - appliedIndex >= TSDB_SYNC_APPLYQ_SIZE_LIMIT) {
     terrno = TSDB_CODE_SYN_WRITE_STALL;
     sError("vgId:%d, failed to append since %s. index:%" PRId64 ", commit-index:%" PRId64 ", applied-index:%" PRId64,
            pNode->vgId, terrstr(), index, pBuf->commitIndex, appliedIndex);
@@ -83,6 +90,7 @@ int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt
 _err:
   syncLogBufferValidate(pBuf);
   taosThreadMutexUnlock(&pBuf->mutex);
+  taosMsleep(1);
   return -1;
 }
 

source/util/src/terror.c

@@ -426,6 +426,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_RESTORING,                "Sync leader is restor
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_SNAPSHOT_MSG,     "Sync invalid snapshot msg")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_BUFFER_FULL,              "Sync buffer is full")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_WRITE_STALL,              "Sync write stall")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NEGO_WIN_EXCEEDED,        "Sync negotiation win exceeded")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR,           "Sync internal error")
 
 //tq