Merge pull request #28872 from taosdata/fix/TS-5669

fix double compress when retry

Author: guanshengliang

source/libs/transport/inc/transComm.h

@@ -452,6 +452,7 @@ void    transPrintEpSet(SEpSet* pEpSet);
 void    transFreeMsg(void* msg);
 int32_t transCompressMsg(char* msg, int32_t len);
 int32_t transDecompressMsg(char** msg, int32_t* len);
+int32_t transDecompressMsgExt(char const* msg, int32_t len, char** out, int32_t* outLen);
 
 int32_t transOpenRefMgt(int size, void (*func)(void*));
 void    transCloseRefMgt(int32_t refMgt);

source/libs/transport/src/transCli.c

@@ -1334,28 +1334,49 @@ static void cliBatchSendCb(uv_write_t* req, int status) {
   }
 }
 bool cliConnMayAddUserInfo(SCliConn* pConn, STransMsgHead** ppHead, int32_t* msgLen) {
+  int32_t   code = 0;
   SCliThrd* pThrd = pConn->hostThrd;
   STrans*   pInst = pThrd->pInst;
   if (pConn->userInited == 1) {
     return false;
   }
   STransMsgHead* pHead = *ppHead;
-  STransMsgHead* tHead = taosMemoryCalloc(1, *msgLen + sizeof(pInst->user));
+  int32_t        len = *msgLen;
+  char*          oriMsg = NULL;
+  int32_t        oriLen = 0;
+
+  if (pHead->comp == 1) {
+    int32_t msgLen = htonl(pHead->msgLen);
+    code = transDecompressMsgExt((char*)(pHead), msgLen, &oriMsg, &oriLen);
+    if (code < 0) {
+      tError("failed to decompress since %s", tstrerror(code));
+      return false;
+    } else {
+      tDebug("decompress msg and resent, compress size %d, raw size %d", msgLen, oriLen);
+    }
+
+    pHead = (STransMsgHead*)oriMsg;
+    len = oriLen;
+  }
+  STransMsgHead* tHead = taosMemoryCalloc(1, len + sizeof(pInst->user));
   if (tHead == NULL) {
     return false;
   }
   memcpy((char*)tHead, (char*)pHead, TRANS_MSG_OVERHEAD);
   memcpy((char*)tHead + TRANS_MSG_OVERHEAD, pInst->user, sizeof(pInst->user));
 
   memcpy((char*)tHead + TRANS_MSG_OVERHEAD + sizeof(pInst->user), (char*)pHead + TRANS_MSG_OVERHEAD,
-         *msgLen - TRANS_MSG_OVERHEAD);
+         len - TRANS_MSG_OVERHEAD);
 
   tHead->withUserInfo = 1;
   *ppHead = tHead;
-  *msgLen += sizeof(pInst->user);
+  *msgLen = len + sizeof(pInst->user);
 
   pConn->pInitUserReq = tHead;
   pConn->userInited = 1;
+  if (oriMsg != NULL) {
+    taosMemoryFree(oriMsg);
+  }
   return true;
 }
 int32_t cliBatchSend(SCliConn* pConn, int8_t direct) {
@@ -1421,9 +1442,8 @@ int32_t cliBatchSend(SCliConn* pConn, int8_t direct) {
       pReq->contLen = 0;
     }
 
-    int32_t msgLen = transMsgLenFromCont(pReq->contLen);
-
     STransMsgHead* pHead = transHeadFromCont(pReq->pCont);
+    int32_t        msgLen = transMsgLenFromCont(pReq->contLen);
 
     char*   content = pReq->pCont;
     int32_t contLen = pReq->contLen;

source/libs/transport/src/transComm.c

@@ -77,16 +77,48 @@ int32_t transDecompressMsg(char** msg, int32_t* len) {
   STransMsgHead* pNewHead = (STransMsgHead*)buf;
   int32_t        decompLen = LZ4_decompress_safe(pCont + sizeof(STransCompMsg), (char*)pNewHead->content,
                                                  tlen - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen);
+
+  if (decompLen != oriLen) {
+    taosMemoryFree(buf);
+    return TSDB_CODE_INVALID_MSG;
+  }
   memcpy((char*)pNewHead, (char*)pHead, sizeof(STransMsgHead));
 
   *len = oriLen + sizeof(STransMsgHead);
   pNewHead->msgLen = htonl(oriLen + sizeof(STransMsgHead));
 
   taosMemoryFree(pHead);
   *msg = buf;
+  return 0;
+}
+int32_t transDecompressMsgExt(char const* msg, int32_t len, char** out, int32_t* outLen) {
+  STransMsgHead* pHead = (STransMsgHead*)msg;
+  char*          pCont = transContFromHead(pHead);
+
+  STransCompMsg* pComp = (STransCompMsg*)pCont;
+  int32_t        oriLen = htonl(pComp->contLen);
+
+  int32_t tlen = len;
+  char*   buf = taosMemoryCalloc(1, oriLen + sizeof(STransMsgHead));
+  if (buf == NULL) {
+    return terrno;
+  }
+
+  STransMsgHead* pNewHead = (STransMsgHead*)buf;
+  int32_t        decompLen = LZ4_decompress_safe(pCont + sizeof(STransCompMsg), (char*)pNewHead->content,
+                                                 tlen - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen);
   if (decompLen != oriLen) {
+    tError("msgLen:%d, originLen:%d, decompLen:%d", len, oriLen, decompLen);
+    taosMemoryFree(buf);
     return TSDB_CODE_INVALID_MSG;
   }
+  memcpy((char*)pNewHead, (char*)pHead, sizeof(STransMsgHead));
+
+  *out = buf;
+  *outLen = oriLen + sizeof(STransMsgHead);
+  pNewHead->msgLen = *outLen;
+  pNewHead->comp = 0;
+
   return 0;
 }
 

tests/parallel_test/cases.task

@@ -1300,6 +1300,7 @@
 #,,y,script,./test.sh -f tsim/mnode/basic3.sim
 ,,y,script,./test.sh -f tsim/mnode/basic4.sim
 ,,y,script,./test.sh -f tsim/mnode/basic5.sim
+,,y,script,./test.sh -f tsim/mnode/basic6.sim
 ,,y,script,./test.sh -f tsim/show/basic.sim
 ,,y,script,./test.sh -f tsim/table/autocreate.sim
 ,,y,script,./test.sh -f tsim/table/basic1.sim