TD-2041

src/inc/taosdef.h

@@ -431,6 +431,8 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
 #define TSDB_PORT_HTTP       11 
 #define TSDB_PORT_ARBITRATOR 12
 
+#define TSDB_MAX_WAL_SIZE    (1024*1024)
+
 typedef enum {
   TAOS_QTYPE_RPC   = 0,
   TAOS_QTYPE_FWD   = 1,

src/inc/taoserror.h

@@ -237,7 +237,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_QRY_NOT_READY,                0, 0x0707, "Query not
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_HAS_RSP,                  0, 0x0708, "Query should response")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_IN_EXEC,                  0, 0x0709, "Multiple retrieval of this query")
 TAOS_DEFINE_ERROR(TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW,      0, 0x070A, "Too many time window in query")
-TAOS_DEFINE_ERROR(TSDB_CODE_QRY_NOT_ENOUGH_BUFFER,      0, 0x070B, "Query buffer limit has reached")
+TAOS_DEFINE_ERROR(TSDB_CODE_QRY_NOT_ENOUGH_BUFFER,        0, 0x070B, "Query buffer limit has reached")
 
 // grant
 TAOS_DEFINE_ERROR(TSDB_CODE_GRANT_EXPIRED,                0, 0x0800, "License expired")
@@ -261,6 +261,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_VERSION,          0, 0x0902, "Invalid Sy
 // wal
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR,                0, 0x1000, "Unexpected generic error in wal")
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_FILE_CORRUPTED,           0, 0x1001, "WAL file is corrupted")
+TAOS_DEFINE_ERROR(TSDB_CODE_WAL_SIZE_LIMIT,               0, 0x1002, "WAL size exceeds limit")
 
 // http
 TAOS_DEFINE_ERROR(TSDB_CODE_HTTP_SERVER_OFFLINE,          0, 0x1100, "http server is not onlin")

src/sync/inc/syncInt.h

@@ -35,6 +35,8 @@ extern "C" {
 #define TAOS_SMSG_SYNC_MUST    6
 #define TAOS_SMSG_STATUS       7
 
+#define SYNC_MAX_SIZE (TSDB_MAX_WAL_SIZE + sizeof(SWalHead) + sizeof(SSyncHead) + 16)
+
 #define nodeRole    pNode->peerInfo[pNode->selfIndex]->role
 #define nodeVersion pNode->peerInfo[pNode->selfIndex]->version
 #define nodeSStatus pNode->peerInfo[pNode->selfIndex]->sstatus

src/sync/src/syncMain.c

@@ -79,7 +79,7 @@ int32_t syncInit() {
   info.numOfThreads = tsSyncTcpThreads;
   info.serverIp = 0;
   info.port = tsSyncPort;
-  info.bufferSize = 640000;
+  info.bufferSize = SYNC_MAX_SIZE;
   info.processBrokenLink = syncProcessBrokenLink;
   info.processIncomingMsg = syncProcessPeerMsg;
   info.processIncomingConn = syncProcessIncommingConnection;
@@ -850,7 +850,7 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) {
   SSyncNode *pNode = pPeer->pSyncNode;
   SWalHead * pHead = (SWalHead *)cont;
 
-  sDebug("%s, forward is received, ver:%" PRIu64, pPeer->id, pHead->version);
+  sDebug("%s, forward is received, hver:%" PRIu64 ", len:%d", pPeer->id, pHead->version, pHead->len);
 
   if (nodeRole == TAOS_SYNC_ROLE_SLAVE) {
     // nodeVersion = pHead->version;
@@ -859,7 +859,7 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) {
     if (nodeSStatus != TAOS_SYNC_STATUS_INIT) {
       syncSaveIntoBuffer(pPeer, pHead);
     } else {
-      sError("%s, forward discarded, ver:%" PRIu64, pPeer->id, pHead->version);
+      sError("%s, forward discarded, hver:%" PRIu64, pPeer->id, pHead->version);
     }
   }
 }
@@ -890,7 +890,7 @@ static int32_t syncReadPeerMsg(SSyncPeer *pPeer, SSyncHead *pHead, char *cont) {
 
   // head.len = htonl(head.len);
   if (pHead->len < 0) {
-    sError("%s, invalid pkt length, len:%d", pPeer->id, pHead->len);
+    sError("%s, invalid pkt length, hlen:%d", pPeer->id, pHead->len);
     return -1;
   }
 

src/sync/src/syncRetrieve.c

@@ -244,7 +244,7 @@ static int32_t syncCheckLastWalChanges(SSyncPeer *pPeer, uint32_t *pEvent) {
 }
 
 static int32_t syncRetrieveLastWal(SSyncPeer *pPeer, char *name, uint64_t fversion, int64_t offset, uint32_t *pEvent) {
-  SWalHead *pHead = malloc(640000);
+  SWalHead *pHead = malloc(SYNC_MAX_SIZE);
   int32_t   code = -1;
   int32_t   bytes = 0;
   int32_t   sfd;

src/sync/src/tarbitrator.c

@@ -86,7 +86,7 @@ int32_t main(int32_t argc, char *argv[]) {
   info.numOfThreads = 1;
   info.serverIp = 0;
   info.port = tsArbitratorPort;
-  info.bufferSize = 640000;
+  info.bufferSize = SYNC_MAX_SIZE;
   info.processBrokenLink = arbProcessBrokenLink;
   info.processIncomingMsg = arbProcessPeerMsg;
   info.processIncomingConn = arbProcessIncommingConnection;

src/vnode/src/vnodeWrite.c

@@ -217,6 +217,11 @@ int32_t vnodeWriteToWQueue(void *vparam, void *wparam, int32_t qtype, void *rpar
     if (code != TSDB_CODE_SUCCESS) return code;
   }
 
+  if (pHead->len > TSDB_MAX_WAL_SIZE) {
+    vError("vgId:%d, wal len:%d exceeds limit, hver:%" PRIu64, pVnode->vgId, pHead->len, pHead->version);
+    return TSDB_CODE_WAL_SIZE_LIMIT;
+  }
+
   int32_t size = sizeof(SVWriteMsg) + sizeof(SWalHead) + pHead->len;
   SVWriteMsg *pWrite = taosAllocateQitem(size);
   if (pWrite == NULL) {

src/wal/inc/walInt.h

@@ -34,7 +34,7 @@ extern int32_t wDebugFlag;
 #define WAL_PREFIX     "wal"
 #define WAL_PREFIX_LEN 3
 #define WAL_REFRESH_MS 1000
-#define WAL_MAX_SIZE   (1024 * 1024)
+#define WAL_MAX_SIZE   (TSDB_MAX_WAL_SIZE + sizeof(SWalHead) + 16)
 #define WAL_SIGNATURE  ((uint32_t)(0xFAFBFDFE))
 #define WAL_PATH_LEN   (TSDB_FILENAME_LEN + 12)
 #define WAL_FILE_LEN   (TSDB_FILENAME_LEN + 32)

tests/script/general/http/sync.sim

@@ -0,0 +1,124 @@
+system sh/stop_dnodes.sh
+
+system sh/deploy.sh -n dnode1 -i 1
+system sh/deploy.sh -n dnode2 -i 2
+system sh/deploy.sh -n dnode3 -i 3
+
+system sh/cfg.sh -n dnode1 -c numOfMnodes -v 3
+system sh/cfg.sh -n dnode2 -c numOfMnodes -v 3
+system sh/cfg.sh -n dnode3 -c numOfMnodes -v 3
+
+system sh/cfg.sh -n dnode1 -c mnodeEqualVnodeNum -v 4
+system sh/cfg.sh -n dnode2 -c mnodeEqualVnodeNum -v 4
+system sh/cfg.sh -n dnode3 -c mnodeEqualVnodeNum -v 4
+
+system sh/cfg.sh -n dnode1 -c http -v 1
+system sh/cfg.sh -n dnode2 -c http -v 1
+system sh/cfg.sh -n dnode3 -c http -v 1
+
+system sh/cfg.sh -n dnode1 -c maxTablesPerVnode -v 20000
+system sh/cfg.sh -n dnode2 -c maxTablesPerVnode -v 20000
+system sh/cfg.sh -n dnode3 -c maxTablesPerVnode -v 20000
+
+system sh/cfg.sh -n dnode1 -c replica -v 3
+system sh/cfg.sh -n dnode2 -c replica -v 3
+system sh/cfg.sh -n dnode3 -c replica -v 3
+
+system sh/cfg.sh -n dnode1 -c maxSQLLength -v 940032
+system sh/cfg.sh -n dnode2 -c maxSQLLength -v 940032
+system sh/cfg.sh -n dnode3 -c maxSQLLength -v 940032
+
+print ============== deploy
+
+system sh/exec.sh -n dnode1 -s start 
+sleep 5001
+sql connect
+
+sql create dnode $hostname2
+sql create dnode $hostname3
+system sh/exec.sh -n dnode2 -s start
+system sh/exec.sh -n dnode3 -s start 
+
+print  =============== step1
+$x = 0
+show1: 
+	$x = $x + 1
+	sleep 2000
+	if $x == 5 then
+	  return -1
+	endi
+sql show mnodes -x show1
+$mnode1Role = $data2_1
+print mnode1Role $mnode1Role
+$mnode2Role = $data2_2
+print mnode2Role $mnode2Role
+$mnode3Role = $data2_3
+print mnode3Role $mnode3Role
+
+if $mnode1Role != master then
+  goto show1
+endi
+if $mnode2Role != slave then
+  goto show1
+endi
+if $mnode3Role != slave then
+  goto show1
+endi
+
+print ===============  step2
+sql create database d1 replica 3
+sql use d1
+
+sql create table table_rest (ts timestamp, i int) 
+print sql length is 870KB
+restful d1 table_rest 1591072800 30000
+restful d1 table_rest 1591172800 30000
+restful d1 table_rest 1591272800 30000
+restful d1 table_rest 1591372800 30000
+restful d1 table_rest 1591472800 30000
+restful d1 table_rest 1591572800 30000
+restful d1 table_rest 1591672800 30000
+restful d1 table_rest 1591772800 30000
+restful d1 table_rest 1591872800 30000
+restful d1 table_rest 1591972800 30000
+
+sql select * from table_rest;
+print rows: $rows
+if $rows != 300000 then
+  return -1
+endi
+
+print ===============  step3
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+sleep 5000
+sql select * from table_rest;
+print rows: $rows
+if $rows != 300000 then
+  return -1
+endi
+system sh/exec.sh -n dnode1 -s start -x SIGINT
+sleep 5000
+
+print ===============  step4
+system sh/exec.sh -n dnode2 -s stop -x SIGINT
+sleep 5000
+sql select * from table_rest;
+print rows: $rows
+if $rows != 300000 then
+  return -1
+endi
+system sh/exec.sh -n dnode2 -s start -x SIGINT
+sleep 5000
+
+print ===============  step5
+system sh/exec.sh -n dnode3 -s stop -x SIGINT
+sleep 5000
+sql select * from table_rest;
+print rows: $rows
+if $rows != 300000 then
+  return -1
+endi
+
+system sh/exec.sh -n dnode1 -s stop  -x SIGINT
+system sh/exec.sh -n dnode2 -s stop  -x SIGINT
+system sh/exec.sh -n dnode3 -s stop  -x SIGINT

tests/script/jenkins/basic.txt

@@ -89,6 +89,7 @@ cd ../../../debug; make
 ./test.sh -f general/http/telegraf.sim
 ./test.sh -f general/http/grafana_bug.sim
 ./test.sh -f general/http/grafana.sim
+./test.sh -f general/http/sync.sim
 
 ./test.sh -f general/import/basic.sim
 ./test.sh -f general/import/commit.sim