• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #3561

19 Dec 2024 03:15AM UTC coverage: 58.812% (-1.3%) from 60.124%
#3561

push

travis-ci

web-flow
Merge pull request #29213 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

130770 of 287658 branches covered (45.46%)

Branch coverage included in aggregate %.

32 of 78 new or added lines in 4 files covered. (41.03%)

7347 existing lines in 166 files now uncovered.

205356 of 283866 relevant lines covered (72.34%)

7187865.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.64
/source/dnode/mnode/impl/src/mndTrans.c
1
/*
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3
 *
4
 * This program is free software: you can use, redistribute, and/or modify
5
 * it under the terms of the GNU Affero General Public License, version 3
6
 * or later ("AGPL"), as published by the Free Software Foundation.
7
 *
8
 * This program is distributed in the hope that it will be useful, but WITHOUT
9
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10
 * FITNESS FOR A PARTICULAR PURPOSE.
11
 *
12
 * You should have received a copy of the GNU Affero General Public License
13
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14
 */
15

16
#define _DEFAULT_SOURCE
17
#include "mndDb.h"
18
#include "mndPrivilege.h"
19
#include "mndShow.h"
20
#include "mndStb.h"
21
#include "mndSubscribe.h"
22
#include "mndSync.h"
23
#include "mndTrans.h"
24
#include "mndUser.h"
25

26
#define TRANS_VER1_NUMBER  1
27
#define TRANS_VER2_NUMBER  2
28
#define TRANS_ARRAY_SIZE   8
29
#define TRANS_RESERVE_SIZE 44
30

31
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
32
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *OldTrans, STrans *pOld);
33
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc);
34

35
static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
36
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction);
37
static void    mndTransDropLogs(SArray *pArray);
38
static void    mndTransDropActions(SArray *pArray);
39

40
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf);
41
static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans, bool topHalf);
42
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans, bool topHalf);
43
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
44
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
45
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
46
static bool    mndTransPerformRedoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
47
static bool    mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
48
static bool    mndTransPerformUndoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
49
static bool    mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
50
static bool    mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
51
static bool    mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
52
static bool    mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
53
static bool    mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
54

55
static inline bool mndTransIsInSyncContext(bool topHalf) { return !topHalf; }
385,588✔
56

57
static bool mndCannotExecuteTrans(SMnode *pMnode, bool topHalf) {
316,234✔
58
  bool isLeader = mndIsLeader(pMnode);
316,234✔
59
  bool ret = (!pMnode->deploy && !isLeader) || mndTransIsInSyncContext(topHalf);
316,234✔
60
  if (ret) mDebug("cannot execute trans action, deploy:%d, isLeader:%d, topHalf:%d", pMnode->deploy, isLeader, topHalf);
316,234✔
61
  return ret;
316,234✔
62
}
63

64
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
374,107✔
65

66
static void    mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
67
static int32_t mndProcessTransTimer(SRpcMsg *pReq);
68
static int32_t mndProcessTtl(SRpcMsg *pReq);
69
static int32_t mndProcessKillTransReq(SRpcMsg *pReq);
70

71
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
72
static void    mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
73

74
static int32_t tsMaxTransId = 0;
75

76
int32_t mndInitTrans(SMnode *pMnode) {
1,366✔
77
  SSdbTable table = {
1,366✔
78
      .sdbType = SDB_TRANS,
79
      .keyType = SDB_KEY_INT32,
80
      .encodeFp = (SdbEncodeFp)mndTransEncode,
81
      .decodeFp = (SdbDecodeFp)mndTransDecode,
82
      .insertFp = (SdbInsertFp)mndTransActionInsert,
83
      .updateFp = (SdbUpdateFp)mndTransActionUpdate,
84
      .deleteFp = (SdbDeleteFp)mndTransDelete,
85
  };
86

87
  mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransTimer);
1,366✔
88
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
1,366✔
89

90
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndRetrieveTrans);
1,366✔
91
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndCancelGetNextTrans);
1,366✔
92
  return sdbSetTable(pMnode->pSdb, table);
1,366✔
93
}
94

95
void mndCleanupTrans(SMnode *pMnode) {}
1,365✔
96

97
static int32_t mndTransGetActionsSize(SArray *pArray) {
427,440✔
98
  int32_t actionNum = taosArrayGetSize(pArray);
427,440✔
99
  int32_t rawDataLen = 0;
427,440✔
100

101
  for (int32_t i = 0; i < actionNum; ++i) {
1,398,099✔
102
    STransAction *pAction = taosArrayGet(pArray, i);
970,659✔
103
    if (pAction->actionType == TRANS_ACTION_RAW) {
970,659✔
104
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
607,316✔
105
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
363,343!
106
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
363,343✔
107
    } else {
108
      // empty
109
    }
110
    rawDataLen += sizeof(int8_t);
970,659✔
111
  }
112

113
  return rawDataLen;
427,440✔
114
}
115

116
static int32_t mndTransEncodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionsNum) {
427,440✔
117
  int32_t code = 0;
427,440✔
118
  int32_t lino = 0;
427,440✔
119
  int32_t dataPos = *offset;
427,440✔
120
  int8_t  unused = 0;
427,440✔
121
  int32_t ret = -1;
427,440✔
122

123
  for (int32_t i = 0; i < actionsNum; ++i) {
1,398,099✔
124
    STransAction *pAction = taosArrayGet(pActions, i);
970,659✔
125
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
970,659!
126
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
970,659!
127
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
970,659!
128
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
970,659!
129
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
970,659!
130
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
970,659!
131
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
970,659!
132
    if (pAction->actionType == TRANS_ACTION_RAW) {
970,659✔
133
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
607,316✔
134
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->rawWritten*/, _OVER)
607,316!
135
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
607,316!
136
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
607,316!
137
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
363,343!
138
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
363,343!
139
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
363,343!
140
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgSent*/, _OVER)
363,343!
141
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgReceived*/, _OVER)
363,343!
142
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
363,343!
143
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
363,343!
144
    } else {
145
      // nothing
146
    }
147
  }
148
  ret = 0;
427,440✔
149

150
_OVER:
427,440✔
151
  *offset = dataPos;
427,440✔
152
  return ret;
427,440✔
153
}
154

155
SSdbRaw *mndTransEncode(STrans *pTrans) {
106,860✔
156
  int32_t code = 0;
106,860✔
157
  int32_t lino = 0;
106,860✔
158
  terrno = TSDB_CODE_INVALID_MSG;
106,860✔
159
  int8_t sver = taosArrayGetSize(pTrans->prepareActions) ? TRANS_VER2_NUMBER : TRANS_VER1_NUMBER;
106,860✔
160

161
  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
106,860✔
162
  rawDataLen += mndTransGetActionsSize(pTrans->prepareActions);
106,860✔
163
  rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
106,860✔
164
  rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
106,860✔
165
  rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
106,860✔
166

167
  SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, sver, rawDataLen);
106,860✔
168
  if (pRaw == NULL) {
106,860!
169
    mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr());
×
170
    return NULL;
×
171
  }
172

173
  int32_t dataPos = 0;
106,860✔
174
  SDB_SET_INT32(pRaw, dataPos, pTrans->id, _OVER)
106,860!
175
  SDB_SET_INT8(pRaw, dataPos, pTrans->stage, _OVER)
106,860!
176
  SDB_SET_INT8(pRaw, dataPos, pTrans->policy, _OVER)
106,860!
177
  SDB_SET_INT8(pRaw, dataPos, pTrans->conflict, _OVER)
106,860!
178
  SDB_SET_INT8(pRaw, dataPos, pTrans->exec, _OVER)
106,860!
179
  SDB_SET_INT8(pRaw, dataPos, pTrans->oper, _OVER)
106,860!
180
  SDB_SET_INT8(pRaw, dataPos, 0, _OVER)
106,860!
181
  SDB_SET_INT16(pRaw, dataPos, pTrans->originRpcType, _OVER)
106,860!
182
  SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
106,860!
183
  SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
106,860!
184
  SDB_SET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
106,860!
185
  SDB_SET_INT32(pRaw, dataPos, pTrans->actionPos, _OVER)
106,860!
186

187
  int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions);
106,860✔
188
  int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
106,860✔
189
  int32_t undoActionNum = taosArrayGetSize(pTrans->undoActions);
106,860✔
190
  int32_t commitActionNum = taosArrayGetSize(pTrans->commitActions);
106,860✔
191

192
  if (sver > TRANS_VER1_NUMBER) {
106,860✔
193
    SDB_SET_INT32(pRaw, dataPos, prepareActionNum, _OVER)
33,392!
194
  }
195
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
106,860!
196
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
106,860!
197
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
106,860!
198

199
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
106,860!
200
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
106,860!
201
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
106,860!
202
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
106,860!
203

204
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
106,860!
205
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
106,860!
206
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
106,860!
207
  if (pTrans->param != NULL) {
106,860!
208
    SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER)
×
209
  }
210

211
  SDB_SET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER)
106,860!
212

213
  int32_t arbGroupNum = taosHashGetSize(pTrans->arbGroupIds);
106,860✔
214
  SDB_SET_INT32(pRaw, dataPos, arbGroupNum, _OVER)
106,860!
215
  void *pIter = NULL;
106,860✔
216
  pIter = taosHashIterate(pTrans->arbGroupIds, NULL);
106,860✔
217
  while (pIter) {
106,878✔
218
    int32_t arbGroupId = *(int32_t *)pIter;
18✔
219
    SDB_SET_INT32(pRaw, dataPos, arbGroupId, _OVER)
18!
220
    pIter = taosHashIterate(pTrans->arbGroupIds, pIter);
18✔
221
  }
222

223
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
106,860!
224
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
106,860!
225

226
  terrno = 0;
106,860✔
227

228
_OVER:
106,860✔
229
  if (terrno != 0) {
106,860!
230
    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
×
231
           dataPos, terrstr());
232
    sdbFreeRaw(pRaw);
×
233
    return NULL;
×
234
  }
235

236
  mTrace("trans:%d, encode to raw:%p, row:%p len:%d", pTrans->id, pRaw, pTrans, dataPos);
106,860✔
237
  return pRaw;
106,860✔
238
}
239

240
static int32_t mndTransDecodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionNum) {
788,904✔
241
  int32_t      code = 0;
788,904✔
242
  int32_t      lino = 0;
788,904✔
243
  STransAction action = {0};
788,904✔
244
  int32_t      dataPos = *offset;
788,904✔
245
  int8_t       unused = 0;
788,904✔
246
  int8_t       stage = 0;
788,904✔
247
  int8_t       actionType = 0;
788,904✔
248
  int32_t      dataLen = 0;
788,904✔
249
  int32_t      ret = -1;
788,904✔
250

251
  for (int32_t i = 0; i < actionNum; ++i) {
2,519,492✔
252
    memset(&action, 0, sizeof(action));
1,730,588✔
253
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
1,730,588!
254
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
1,730,588!
255
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
1,730,588!
256
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
1,730,588!
257
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
1,730,588!
258
    action.actionType = actionType;
1,730,588✔
259
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
1,730,588!
260
    action.stage = stage;
1,730,588✔
261
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
1,730,588!
262
    if (action.actionType == TRANS_ACTION_RAW) {
1,730,588✔
263
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.rawWritten*/, _OVER)
1,068,937!
264
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
1,068,937!
265
      action.pRaw = taosMemoryMalloc(dataLen);
1,068,937!
266
      if (action.pRaw == NULL) goto _OVER;
1,068,937!
267
      mTrace("raw:%p, is created", action.pRaw);
1,068,937✔
268
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
1,068,937!
269
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
1,068,937!
270
      action.pRaw = NULL;
1,068,937✔
271
    } else if (action.actionType == TRANS_ACTION_MSG) {
661,651!
272
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
661,651!
273
      tmsgUpdateDnodeEpSet(&action.epSet);
661,651✔
274
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
661,651!
275
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgSent*/, _OVER)
661,651!
276
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgReceived*/, _OVER)
661,651!
277
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
661,651!
278
      action.pCont = taosMemoryMalloc(action.contLen);
661,651!
279
      if (action.pCont == NULL) goto _OVER;
661,651!
280
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
661,651!
281
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
661,651!
282
      action.pCont = NULL;
661,651✔
283
    } else {
284
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
×
285
    }
286
  }
287
  ret = 0;
788,904✔
288

289
_OVER:
788,904✔
290
  *offset = dataPos;
788,904✔
291
  taosMemoryFreeClear(action.pCont);
788,904!
292
  return ret;
788,904✔
293
}
294

295
SSdbRow *mndTransDecode(SSdbRaw *pRaw) {
197,226✔
296
  terrno = TSDB_CODE_INVALID_MSG;
197,226✔
297
  int32_t  code = 0;
197,226✔
298
  int32_t  lino = 0;
197,226✔
299
  SSdbRow *pRow = NULL;
197,226✔
300
  STrans  *pTrans = NULL;
197,226✔
301
  char    *pData = NULL;
197,226✔
302
  int32_t  dataLen = 0;
197,226✔
303
  int8_t   sver = 0;
197,226✔
304
  int32_t  prepareActionNum = 0;
197,226✔
305
  int32_t  redoActionNum = 0;
197,226✔
306
  int32_t  undoActionNum = 0;
197,226✔
307
  int32_t  commitActionNum = 0;
197,226✔
308
  int32_t  dataPos = 0;
197,226✔
309
  int32_t  arbgroupIdNum = 0;
197,226✔
310

311
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
197,226!
312

313
  if (sver != TRANS_VER1_NUMBER && sver != TRANS_VER2_NUMBER) {
197,226!
314
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
315
    goto _OVER;
×
316
  }
317

318
  pRow = sdbAllocRow(sizeof(STrans));
197,226✔
319
  if (pRow == NULL) goto _OVER;
197,226!
320

321
  pTrans = sdbGetRowObj(pRow);
197,226✔
322
  if (pTrans == NULL) goto _OVER;
197,226!
323

324
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
197,226!
325

326
  int8_t stage = 0;
197,226✔
327
  int8_t policy = 0;
197,226✔
328
  int8_t conflict = 0;
197,226✔
329
  int8_t exec = 0;
197,226✔
330
  int8_t oper = 0;
197,226✔
331
  int8_t reserved = 0;
197,226✔
332
  int8_t actionType = 0;
197,226✔
333
  SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
197,226!
334
  SDB_GET_INT8(pRaw, dataPos, &policy, _OVER)
197,226!
335
  SDB_GET_INT8(pRaw, dataPos, &conflict, _OVER)
197,226!
336
  SDB_GET_INT8(pRaw, dataPos, &exec, _OVER)
197,226!
337
  SDB_GET_INT8(pRaw, dataPos, &oper, _OVER)
197,226!
338
  SDB_GET_INT8(pRaw, dataPos, &reserved, _OVER)
197,226!
339
  pTrans->stage = stage;
197,226✔
340
  pTrans->policy = policy;
197,226✔
341
  pTrans->conflict = conflict;
197,226✔
342
  pTrans->exec = exec;
197,226✔
343
  pTrans->oper = oper;
197,226✔
344
  SDB_GET_INT16(pRaw, dataPos, &pTrans->originRpcType, _OVER)
197,226!
345
  SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
197,226!
346
  SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
197,226!
347
  SDB_GET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
197,226!
348
  SDB_GET_INT32(pRaw, dataPos, &pTrans->actionPos, _OVER)
197,226!
349

350
  if (sver > TRANS_VER1_NUMBER) {
197,226✔
351
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
70,158!
352
  }
353
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
197,226!
354
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
197,226!
355
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
197,226!
356

357
  pTrans->prepareActions = taosArrayInit(prepareActionNum, sizeof(STransAction));
197,226✔
358
  pTrans->redoActions = taosArrayInit(redoActionNum, sizeof(STransAction));
197,226✔
359
  pTrans->undoActions = taosArrayInit(undoActionNum, sizeof(STransAction));
197,226✔
360
  pTrans->commitActions = taosArrayInit(commitActionNum, sizeof(STransAction));
197,226✔
361

362
  if (pTrans->prepareActions == NULL) goto _OVER;
197,226!
363
  if (pTrans->redoActions == NULL) goto _OVER;
197,226!
364
  if (pTrans->undoActions == NULL) goto _OVER;
197,226!
365
  if (pTrans->commitActions == NULL) goto _OVER;
197,226!
366

367
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
197,226!
368
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
197,226!
369
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
197,226!
370
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
197,226!
371

372
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
197,226!
373
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
197,226!
374
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
197,226!
375
  if (pTrans->paramLen != 0) {
197,226!
376
    pTrans->param = taosMemoryMalloc(pTrans->paramLen);
×
377
    if (pTrans->param == NULL) goto _OVER;
×
378
    SDB_GET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER);
×
379
  }
380

381
  SDB_GET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER);
197,226!
382

383
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
197,226✔
384

385
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
197,226!
386
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
197,256✔
387
    int32_t arbGroupId = 0;
30✔
388
    SDB_GET_INT32(pRaw, dataPos, &arbGroupId, _OVER)
30!
389
    if ((terrno = taosHashPut(pTrans->arbGroupIds, &arbGroupId, sizeof(int32_t), NULL, 0)) != 0) goto _OVER;
30!
390
  }
391

392
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
197,226!
393

394
  terrno = 0;
197,226✔
395

396
_OVER:
197,226✔
397
  if (terrno != 0 && pTrans != NULL) {
197,226!
398
    mError("trans:%d, failed to parse from raw:%p since %s", pTrans->id, pRaw, terrstr());
×
399
    mndTransDropData(pTrans);
×
400
    taosMemoryFreeClear(pRow);
×
401
    return NULL;
×
402
  }
403

404
  if (pTrans != NULL) {
197,226!
405
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
197,226✔
406
  }
407
  return pRow;
197,226✔
408
}
409

410
static const char *mndTransStr(ETrnStage stage) {
1,444,837✔
411
  switch (stage) {
1,444,837!
412
    case TRN_STAGE_PREPARE:
89,993✔
413
      return "prepare";
89,993✔
414
    case TRN_STAGE_REDO_ACTION:
691,445✔
415
      return "redoAction";
691,445✔
416
    case TRN_STAGE_ROLLBACK:
10✔
417
      return "rollback";
10✔
418
    case TRN_STAGE_UNDO_ACTION:
74✔
419
      return "undoAction";
74✔
420
    case TRN_STAGE_COMMIT:
171,032✔
421
      return "commit";
171,032✔
422
    case TRN_STAGE_COMMIT_ACTION:
237,479✔
423
      return "commitAction";
237,479✔
424
    case TRN_STAGE_FINISH:
254,794✔
425
      return "finished";
254,794✔
426
    case TRN_STAGE_PRE_FINISH:
10✔
427
      return "pre-finish";
10✔
428
    default:
×
429
      return "invalid";
×
430
  }
431
}
432

433
static void mndSetTransLastAction(STrans *pTrans, STransAction *pAction) {
449,979✔
434
  if (pAction != NULL) {
449,979✔
435
    pTrans->lastAction = pAction->id;
360,847✔
436
    pTrans->lastMsgType = pAction->msgType;
360,847✔
437
    pTrans->lastEpset = pAction->epSet;
360,847✔
438
    pTrans->lastErrorNo = pAction->errCode;
360,847✔
439
  } else {
440
    pTrans->lastAction = 0;
89,132✔
441
    pTrans->lastMsgType = 0;
89,132✔
442
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
89,132✔
443
    pTrans->lastErrorNo = 0;
89,132✔
444
  }
445
}
449,979✔
446

447
static void mndTransTestStartFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
448
  mInfo("test trans start, param:%s, len:%d", (char *)param, paramLen);
×
449
}
×
450

451
static void mndTransTestStopFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
452
  mInfo("test trans stop, param:%s, len:%d", (char *)param, paramLen);
×
453
}
×
454

455
static TransCbFp mndTransGetCbFp(ETrnFunc ftype) {
1,784✔
456
  switch (ftype) {
1,784!
457
    case TRANS_START_FUNC_TEST:
×
458
      return mndTransTestStartFunc;
×
459
    case TRANS_STOP_FUNC_TEST:
×
460
      return mndTransTestStopFunc;
×
461
    case TRANS_START_FUNC_MQ_REB:
892✔
462
      return mndRebCntInc;
892✔
463
    case TRANS_STOP_FUNC_MQ_REB:
892✔
464
      return mndRebCntDec;
892✔
465
    default:
×
466
      return NULL;
×
467
  }
468
}
469

470
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
37,287✔
471
  mInfo("trans:%d, perform insert action, row:%p stage:%s, callfunc:1, startFunc:%d", pTrans->id, pTrans,
37,287!
472
        mndTransStr(pTrans->stage), pTrans->startFunc);
473

474
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
37,287✔
475

476
  if (pTrans->startFunc > 0) {
37,287✔
477
    TransCbFp fp = mndTransGetCbFp(pTrans->startFunc);
892✔
478
    if (fp) {
892!
479
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
892✔
480
    }
481
    // pTrans->startFunc = 0;
482
  }
483

484
  if (pTrans->stage == TRN_STAGE_COMMIT) {
37,287!
UNCOV
485
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
×
UNCOV
486
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pTrans->id);
×
487
  }
488

489
  if (pTrans->stage == TRN_STAGE_ROLLBACK) {
37,287!
UNCOV
490
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
×
UNCOV
491
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pTrans->id);
×
492
  }
493

494
  if (pTrans->stage == TRN_STAGE_PRE_FINISH) {
37,287!
495
    pTrans->stage = TRN_STAGE_FINISH;
×
496
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pTrans->id);
×
497
  }
498

499
  return 0;
37,287✔
500
}
501

502
void mndTransDropData(STrans *pTrans) {
229,717✔
503
  if (pTrans->prepareActions != NULL) {
229,717!
504
    mndTransDropActions(pTrans->prepareActions);
229,717✔
505
    pTrans->prepareActions = NULL;
229,717✔
506
  }
507
  if (pTrans->redoActions != NULL) {
229,717!
508
    mndTransDropActions(pTrans->redoActions);
229,717✔
509
    pTrans->redoActions = NULL;
229,717✔
510
  }
511
  if (pTrans->undoActions != NULL) {
229,717!
512
    mndTransDropActions(pTrans->undoActions);
229,717✔
513
    pTrans->undoActions = NULL;
229,717✔
514
  }
515
  if (pTrans->commitActions != NULL) {
229,717!
516
    mndTransDropActions(pTrans->commitActions);
229,717✔
517
    pTrans->commitActions = NULL;
229,717✔
518
  }
519
  if (pTrans->arbGroupIds != NULL) {
229,717!
520
    taosHashCleanup(pTrans->arbGroupIds);
229,717✔
521
  }
522
  if (pTrans->pRpcArray != NULL) {
229,717✔
523
    taosArrayDestroy(pTrans->pRpcArray);
32,491✔
524
    pTrans->pRpcArray = NULL;
32,491✔
525
  }
526
  if (pTrans->rpcRsp != NULL) {
229,717✔
527
    taosMemoryFree(pTrans->rpcRsp);
9,331!
528
    pTrans->rpcRsp = NULL;
9,331✔
529
    pTrans->rpcRspLen = 0;
9,331✔
530
  }
531
  if (pTrans->param != NULL) {
229,717!
532
    taosMemoryFree(pTrans->param);
×
533
    pTrans->param = NULL;
×
534
    pTrans->paramLen = 0;
×
535
  }
536
  (void)taosThreadMutexDestroy(&pTrans->mutex);
229,717✔
537
}
229,717✔
538

539
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc) {
117,245✔
540
  mInfo("trans:%d, perform delete action, row:%p stage:%s callfunc:%d, stopFunc:%d", pTrans->id, pTrans,
117,245!
541
        mndTransStr(pTrans->stage), callFunc, pTrans->stopFunc);
542

543
  if (pTrans->stopFunc > 0 && callFunc) {
117,245✔
544
    TransCbFp fp = mndTransGetCbFp(pTrans->stopFunc);
892✔
545
    if (fp) {
892!
546
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
892✔
547
    }
548
    // pTrans->stopFunc = 0;
549
  }
550

551
  mndTransDropData(pTrans);
117,245✔
552
  return 0;
117,245✔
553
}
554

555
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
170,936✔
556
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
576,721✔
557
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
405,785✔
558
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
405,785✔
559
    pOldAction->rawWritten = pNewAction->rawWritten;
405,785✔
560
    pOldAction->msgSent = pNewAction->msgSent;
405,785✔
561
    pOldAction->msgReceived = pNewAction->msgReceived;
405,785✔
562
    pOldAction->errCode = pNewAction->errCode;
405,785✔
563
  }
564
}
170,936✔
565

566
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
42,734✔
567
  mInfo("trans:%d, perform update action, old row:%p stage:%s create:%" PRId64 ", new row:%p stage:%s create:%" PRId64,
42,734!
568
        pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage), pNew->createdTime);
569

570
  if (pOld->createdTime != pNew->createdTime) {
42,734!
UNCOV
571
    mError("trans:%d, failed to perform update action since createTime not match, old row:%p stage:%s create:%" PRId64
×
572
           ", new row:%p stage:%s create:%" PRId64,
573
           pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage),
574
           pNew->createdTime);
575
    // only occured while sync timeout
UNCOV
576
    TAOS_RETURN(TSDB_CODE_MND_TRANS_SYNC_TIMEOUT);
×
577
  }
578

579
  mndTransUpdateActions(pOld->prepareActions, pNew->prepareActions);
42,734✔
580
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
42,734✔
581
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
42,734✔
582
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
42,734✔
583
  pOld->stage = pNew->stage;
42,734✔
584
  pOld->actionPos = pNew->actionPos;
42,734✔
585

586
  if (pOld->stage == TRN_STAGE_COMMIT) {
42,734✔
587
    pOld->stage = TRN_STAGE_COMMIT_ACTION;
37,222✔
588
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pNew->id);
37,222!
589
  }
590

591
  if (pOld->stage == TRN_STAGE_ROLLBACK) {
42,734✔
592
    pOld->stage = TRN_STAGE_UNDO_ACTION;
2✔
593
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pNew->id);
2!
594
  }
595

596
  if (pOld->stage == TRN_STAGE_PRE_FINISH) {
42,734✔
597
    pOld->stage = TRN_STAGE_FINISH;
2✔
598
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pNew->id);
2!
599
  }
600

601
  return 0;
42,734✔
602
}
603

604
STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
196,898✔
605
  STrans *pTrans = sdbAcquire(pMnode->pSdb, SDB_TRANS, &transId);
196,898✔
606
  if (pTrans == NULL) {
196,898✔
607
    terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
3,304✔
608
  }
609
  return pTrans;
196,898✔
610
}
611

612
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
193,595✔
613
  SSdb *pSdb = pMnode->pSdb;
193,595✔
614
  if (pTrans != NULL) mInfo("vgId:1, trans:%d, release transaction", pTrans->id);
193,595!
615
  sdbRelease(pSdb, pTrans);
193,595✔
616
}
193,595✔
617

618
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
32,491✔
619
                       const char *opername) {
620
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
32,491!
621
  if (pTrans == NULL) {
32,491!
622
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
623
    mError("failed to create transaction since %s", terrstr());
×
624
    return NULL;
×
625
  }
626

627
  if (opername != NULL) {
32,491!
628
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
32,491✔
629
  }
630

631
  int32_t sdbMaxId = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
32,491✔
632
  sdbReadLock(pMnode->pSdb, SDB_TRANS);
32,491✔
633
  pTrans->id = TMAX(sdbMaxId, tsMaxTransId + 1);
32,491✔
634
  sdbUnLock(pMnode->pSdb, SDB_TRANS);
32,491✔
635
  pTrans->stage = TRN_STAGE_PREPARE;
32,491✔
636
  pTrans->policy = policy;
32,491✔
637
  pTrans->conflict = conflict;
32,491✔
638
  pTrans->exec = TRN_EXEC_PARALLEL;
32,491✔
639
  pTrans->createdTime = taosGetTimestampMs();
32,491✔
640
  pTrans->prepareActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
32,491✔
641
  pTrans->redoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
32,491✔
642
  pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
32,491✔
643
  pTrans->commitActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
32,491✔
644
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
32,491✔
645
  pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
32,491✔
646
  pTrans->mTraceId = pReq ? TRACE_GET_ROOTID(&pReq->info.traceId) : tGenIdPI64();
32,491✔
647
  taosInitRWLatch(&pTrans->lockRpcArray);
32,491✔
648
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
32,491✔
649

650
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
32,491!
651
      pTrans->pRpcArray == NULL) {
32,491!
652
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
653
    mError("failed to create transaction since %s", terrstr());
×
654
    mndTransDrop(pTrans);
×
655
    return NULL;
×
656
  }
657

658
  if (pReq != NULL) {
32,491✔
659
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
44,660!
660
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
661
      return NULL;
×
662
    }
663
    pTrans->originRpcType = pReq->msgType;
22,330✔
664
  }
665

666
  mInfo("trans:%d, create transaction:%s, origin:%s", pTrans->id, pTrans->opername, opername);
32,491!
667

668
  mTrace("trans:%d, local object is created, data:%p", pTrans->id, pTrans);
32,491✔
669
  return pTrans;
32,491✔
670
}
671

672
static void mndTransDropActions(SArray *pArray) {
918,868✔
673
  int32_t size = taosArrayGetSize(pArray);
918,868✔
674
  for (int32_t i = 0; i < size; ++i) {
2,933,197✔
675
    STransAction *pAction = taosArrayGet(pArray, i);
2,014,329✔
676
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,014,329✔
677
      taosMemoryFreeClear(pAction->pRaw);
1,257,133!
678
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
757,196!
679
      taosMemoryFreeClear(pAction->pCont);
757,196!
680
    } else {
681
      // nothing
682
    }
683
  }
684

685
  taosArrayDestroy(pArray);
918,868✔
686
}
918,868✔
687

688
void mndTransDrop(STrans *pTrans) {
32,849✔
689
  if (pTrans != NULL) {
32,849✔
690
    mndTransDropData(pTrans);
32,491✔
691
    mTrace("trans:%d, local object is freed, data:%p", pTrans->id, pTrans);
32,491✔
692
    taosMemoryFreeClear(pTrans);
32,491!
693
  }
694
}
32,849✔
695

696
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
283,741✔
697
  pAction->id = taosArrayGetSize(pArray);
283,741✔
698

699
  void *ptr = taosArrayPush(pArray, pAction);
283,741✔
700
  if (ptr == NULL) {
283,741!
701
    TAOS_RETURN(terrno);
×
702
  }
703

704
  return 0;
283,741✔
705
}
706

707
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
1,599✔
708
  STransAction action = {
1,599✔
709
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
1,599✔
710
  return mndTransAppendAction(pTrans->redoActions, &action);
1,599✔
711
}
712

713
int32_t mndTransAppendNullLog(STrans *pTrans) {
×
714
  STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_NULL};
×
715
  return mndTransAppendAction(pTrans->redoActions, &action);
×
716
}
717

718
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
10,993✔
719
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
10,993✔
720
  return mndTransAppendAction(pTrans->undoActions, &action);
10,993✔
721
}
722

723
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
159,356✔
724
  STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
159,356✔
725
  return mndTransAppendAction(pTrans->commitActions, &action);
159,356✔
726
}
727

728
int32_t mndTransAppendPrepareLog(STrans *pTrans, SSdbRaw *pRaw) {
16,248✔
729
  STransAction action = {
16,248✔
730
      .pRaw = pRaw, .stage = TRN_STAGE_PREPARE, .actionType = TRANS_ACTION_RAW, .mTraceId = pTrans->mTraceId};
16,248✔
731
  return mndTransAppendAction(pTrans->prepareActions, &action);
16,248✔
732
}
733

734
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction) {
71,929✔
735
  pAction->stage = TRN_STAGE_REDO_ACTION;
71,929✔
736
  pAction->actionType = TRANS_ACTION_MSG;
71,929✔
737
  pAction->mTraceId = pTrans->mTraceId;
71,929✔
738
  return mndTransAppendAction(pTrans->redoActions, pAction);
71,929✔
739
}
740

741
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction) {
23,616✔
742
  pAction->stage = TRN_STAGE_UNDO_ACTION;
23,616✔
743
  pAction->actionType = TRANS_ACTION_MSG;
23,616✔
744
  return mndTransAppendAction(pTrans->undoActions, pAction);
23,616✔
745
}
746

747
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen) {
9,331✔
748
  pTrans->rpcRsp = pCont;
9,331✔
749
  pTrans->rpcRspLen = contLen;
9,331✔
750
}
9,331✔
751

752
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
882✔
753
  pTrans->startFunc = startFunc;
882✔
754
  pTrans->stopFunc = stopFunc;
882✔
755
  pTrans->param = param;
882✔
756
  pTrans->paramLen = paramLen;
882✔
757
}
882✔
758

759
int32_t mndSetRpcInfoForDbTrans(SMnode *pMnode, SRpcMsg *pMsg, EOperType oper, const char *dbname) {
×
760
  STrans *pTrans = NULL;
×
761
  void   *pIter = NULL;
×
762
  int32_t code = -1;
×
763

764
  while (1) {
765
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
×
766
    if (pIter == NULL) break;
×
767

768
    if (pTrans->oper == oper) {
×
769
      if (strcasecmp(dbname, pTrans->dbname) == 0) {
×
770
        mInfo("trans:%d, db:%s oper:%d matched with input", pTrans->id, dbname, oper);
×
771
        taosWLockLatch(&pTrans->lockRpcArray);
×
772
        if (pTrans->pRpcArray == NULL) {
×
773
          pTrans->pRpcArray = taosArrayInit(4, sizeof(SRpcHandleInfo));
×
774
        }
775
        if (pTrans->pRpcArray != NULL && taosArrayPush(pTrans->pRpcArray, &pMsg->info) != NULL) {
×
776
          code = 0;
×
777
        }
778
        taosWUnLockLatch(&pTrans->lockRpcArray);
×
779

780
        sdbRelease(pMnode->pSdb, pTrans);
×
781
        break;
×
782
      }
783
    }
784

785
    sdbRelease(pMnode->pSdb, pTrans);
×
786
  }
787
  return code;
×
788
}
789

790
void mndTransSetDbName(STrans *pTrans, const char *dbname, const char *stbname) {
23,101✔
791
  if (dbname != NULL) {
23,101!
792
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
23,101✔
793
  }
794
  if (stbname != NULL) {
23,101✔
795
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
16,913✔
796
  }
797
}
23,101✔
798

799
void mndTransAddArbGroupId(STrans *pTrans, int32_t groupId) {
6✔
800
  if (taosHashPut(pTrans->arbGroupIds, &groupId, sizeof(int32_t), NULL, 0) != 0) {
6!
801
    mError("trans:%d, failed to put groupid into hash, groupId:%d", pTrans->id, groupId);
×
802
  }
803
}
6✔
804

805
void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
1,217✔
806

807
void mndTransSetParallel(STrans *pTrans) { pTrans->exec = TRN_EXEC_PARALLEL; }
×
808

809
void mndTransSetChangeless(STrans *pTrans) { pTrans->changeless = true; }
33✔
810

811
void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }
3,175✔
812

813
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
69,512✔
814
  int32_t  code = 0;
69,512✔
815
  SSdbRaw *pRaw = mndTransEncode(pTrans);
69,512✔
816
  if (pRaw == NULL) {
69,512!
817
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
818
    if (terrno != 0) code = terrno;
×
819
    mError("trans:%d, failed to encode while sync trans since %s", pTrans->id, tstrerror(code));
×
820
    TAOS_RETURN(code);
×
821
  }
822
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
69,512!
823

824
  mInfo("trans:%d, sync to other mnodes, stage:%s createTime:%" PRId64, pTrans->id, mndTransStr(pTrans->stage),
69,512!
825
        pTrans->createdTime);
826
  code = mndSyncPropose(pMnode, pRaw, pTrans->id);
69,512✔
827
  if (code != 0) {
69,512✔
828
    mError("trans:%d, failed to sync, errno:%s code:0x%x createTime:%" PRId64 " saved trans:%d", pTrans->id,
14!
829
           tstrerror(code), code, pTrans->createdTime, pMnode->syncMgmt.transId);
830
    sdbFreeRaw(pRaw);
14✔
831
    TAOS_RETURN(code);
14✔
832
  }
833

834
  sdbFreeRaw(pRaw);
69,498✔
835
  mInfo("trans:%d, sync finished, createTime:%" PRId64, pTrans->id, pTrans->createdTime);
69,498!
836
  TAOS_RETURN(code);
69,498✔
837
}
838

839
static bool mndCheckDbConflict(const char *conflict, STrans *pTrans) {
229✔
840
  if (conflict[0] == 0) return false;
229!
841
  if (strcasecmp(conflict, pTrans->dbname) == 0) return true;
229✔
842
  return false;
6✔
843
}
844

845
static bool mndCheckStbConflict(const char *conflict, STrans *pTrans) {
793✔
846
  if (conflict[0] == 0) return false;
793!
847
  if (strcasecmp(conflict, pTrans->stbname) == 0) return true;
793!
848
  return false;
793✔
849
}
850

851
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
1,022✔
852
  if (conflict) {
1,022✔
853
    mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with trans:%d db:%s stb:%s type:%d", pNew->id,
223!
854
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
855
    *globalConflict = true;
223✔
856
  } else {
857
    mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with trans:%d db:%s stb:%s type:%d", pNew->id, pNew->dbname,
799!
858
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
859
  }
860
}
1,022✔
861

862
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
92,845✔
863
  STrans *pTrans = NULL;
92,845✔
864
  void   *pIter = NULL;
92,845✔
865
  bool    conflict = false;
92,845✔
866

867
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
92,845✔
868

869
  int32_t size = sdbGetSize(pMnode->pSdb, SDB_TRANS);
64,956✔
870
  mInfo("trans:%d, trans hash size %d", pNew->id, size);
64,956!
871

872
  while (1) {
873
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
67,512✔
874
    if (pIter == NULL) break;
67,512✔
875

876
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
2,556!
877

878
    if (pNew->conflict == TRN_CONFLICT_DB) {
2,556✔
879
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
1,076!
880
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
1,076!
UNCOV
881
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
×
UNCOV
882
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
×
883
      }
884
    }
885

886
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
2,556✔
887
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
1,480✔
888
      if (pTrans->conflict == TRN_CONFLICT_DB) {
1,480✔
889
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
229✔
890
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
229✔
891
      }
892
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
1,480✔
893
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
564✔
894
      }
895
    }
896

897
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
2,556!
898
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
×
899
      if (pTrans->conflict == TRN_CONFLICT_ARBGROUP) {
×
900
        void *pGidIter = taosHashIterate(pNew->arbGroupIds, NULL);
×
901
        while (pGidIter != NULL) {
×
902
          int32_t groupId = *(int32_t *)pGidIter;
×
903
          if (taosHashGet(pTrans->arbGroupIds, &groupId, sizeof(int32_t)) != NULL) {
×
904
            taosHashCancelIterate(pNew->arbGroupIds, pGidIter);
×
905
            mndTransLogConflict(pNew, pTrans, true, &conflict);
×
906
            break;
×
907
          } else {
908
            mndTransLogConflict(pNew, pTrans, false, &conflict);
×
909
          }
910
          pGidIter = taosHashIterate(pNew->arbGroupIds, pGidIter);
×
911
        }
912
      }
913
    }
914

915
    if (pNew->conflict == TRN_CONFLICT_TSMA) {
2,556!
916
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL || pTrans->conflict == TRN_CONFLICT_TSMA) {
×
917
        mndTransLogConflict(pNew, pTrans, true, &conflict);
×
918
      } else {
919
        mndTransLogConflict(pNew, pTrans, false, &conflict);
×
920
      }
921
    }
922

923
    sdbRelease(pMnode->pSdb, pTrans);
2,556✔
924
  }
925

926
  return conflict;
64,956✔
927
}
928

929
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
92,845✔
930
  int32_t code = 0;
92,845✔
931
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
92,845✔
932
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
57,625!
933
      code = TSDB_CODE_MND_TRANS_CONFLICT;
×
934
      mError("trans:%d, failed to check tran conflict since db not set", pTrans->id);
×
935
      TAOS_RETURN(code);
×
936
    }
937
  }
938

939
  if (mndCheckTransConflict(pMnode, pTrans)) {
92,845✔
940
    code = TSDB_CODE_MND_TRANS_CONFLICT;
250✔
941
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
250!
942
    TAOS_RETURN(code);
250✔
943
  }
944

945
  TAOS_RETURN(code);
92,595✔
946
}
947

948
int32_t mndTransCheckConflictWithCompact(SMnode *pMnode, STrans *pTrans) {
169✔
949
  int32_t      code = 0;
169✔
950
  void        *pIter = NULL;
169✔
951
  bool         conflict = false;
169✔
952
  SCompactObj *pCompact = NULL;
169✔
953

UNCOV
954
  while (1) {
×
955
    bool thisConflict = false;
169✔
956
    pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
169✔
957
    if (pIter == NULL) break;
169!
958

UNCOV
959
    if (pTrans->conflict == TRN_CONFLICT_GLOBAL) {
×
UNCOV
960
      thisConflict = true;
×
961
    }
UNCOV
962
    if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
×
UNCOV
963
      if (strcasecmp(pTrans->dbname, pCompact->dbname) == 0) thisConflict = true;
×
964
    }
965

UNCOV
966
    if (thisConflict) {
×
UNCOV
967
      mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with compact:%d db:%s", pTrans->id,
×
968
             pTrans->dbname, pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
UNCOV
969
      conflict = true;
×
970
    } else {
971
      mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with compact:%d db:%s", pTrans->id, pTrans->dbname,
×
972
            pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
973
    }
UNCOV
974
    sdbRelease(pMnode->pSdb, pCompact);
×
975
  }
976

977
  if (conflict) {
169!
UNCOV
978
    code = TSDB_CODE_MND_TRANS_CONFLICT_COMPACT;
×
UNCOV
979
    mError("trans:%d, failed to check tran conflict with compact since %s", pTrans->id, tstrerror(code));
×
UNCOV
980
    TAOS_RETURN(code);
×
981
  }
982

983
  TAOS_RETURN(code);
169✔
984
}
985

986
static bool mndTransActionsOfSameType(SArray *pActions) {
71,215✔
987
  int32_t size = taosArrayGetSize(pActions);
71,215✔
988
  ETrnAct lastActType = TRANS_ACTION_NULL;
71,215✔
989
  bool    same = true;
71,215✔
990
  for (int32_t i = 0; i < size; ++i) {
309,892✔
991
    STransAction *pAction = taosArrayGet(pActions, i);
238,677✔
992
    if (i > 0) {
238,677✔
993
      if (lastActType != pAction->actionType) {
181,841!
994
        same = false;
×
995
        break;
×
996
      }
997
    }
998
    lastActType = pAction->actionType;
238,677✔
999
  }
1000
  return same;
71,215✔
1001
}
1002

1003
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
32,212✔
1004
  int32_t code = 0;
32,212✔
1005
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
32,212✔
1006
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
31,053!
1007
      code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1008
      mError("trans:%d, types of parallel redo actions are not the same", pTrans->id);
×
1009
      TAOS_RETURN(code);
×
1010
    }
1011

1012
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
31,053✔
1013
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
7,950!
1014
        code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1015
        mError("trans:%d, types of parallel undo actions are not the same", pTrans->id);
×
1016
        TAOS_RETURN(code);
×
1017
      }
1018
    }
1019
  }
1020

1021
  TAOS_RETURN(code);
32,212✔
1022
}
1023

1024
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
32,212✔
1025
  int32_t code = 0;
32,212✔
1026
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
32,212!
1027
    code = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
×
1028
    mError("trans:%d, commit actions of non-changeless trans are empty", pTrans->id);
×
1029
    TAOS_RETURN(code);
×
1030
  }
1031
  if (mndTransActionsOfSameType(pTrans->commitActions) == false) {
32,212!
1032
    code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1033
    mError("trans:%d, types of commit actions are not the same", pTrans->id);
×
1034
    TAOS_RETURN(code);
×
1035
  }
1036

1037
  TAOS_RETURN(code);
32,212✔
1038
}
1039

1040
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
32,212✔
1041
  int32_t code = 0;
32,212✔
1042
  if (pTrans == NULL) {
32,212!
1043
    return TSDB_CODE_INVALID_PARA;
×
1044
  }
1045

1046
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
32,212!
1047

1048
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
32,212!
1049

1050
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
32,212!
1051

1052
  mInfo("trans:%d, prepare transaction", pTrans->id);
32,212!
1053
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
32,212✔
1054
    mError("trans:%d, failed to prepare since %s", pTrans->id, tstrerror(code));
12!
1055
    sdbWriteLock(pMnode->pSdb, SDB_TRANS);
12✔
1056
    tsMaxTransId = TMAX(pTrans->id, tsMaxTransId);
12✔
1057
    sdbUnLock(pMnode->pSdb, SDB_TRANS);
12✔
1058
    TAOS_RETURN(code);
12✔
1059
  }
1060
  mInfo("trans:%d, prepare finished", pTrans->id);
32,200!
1061

1062
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
32,200✔
1063
  if (pNew == NULL) {
32,200!
1064
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1065
    if (terrno != 0) code = terrno;
×
1066
    mError("trans:%d, failed to read from sdb since %s", pTrans->id, tstrerror(code));
×
1067
    TAOS_RETURN(code);
×
1068
  }
1069

1070
  pNew->pRpcArray = pTrans->pRpcArray;
32,200✔
1071
  pNew->rpcRsp = pTrans->rpcRsp;
32,200✔
1072
  pNew->rpcRspLen = pTrans->rpcRspLen;
32,200✔
1073
  pNew->mTraceId = pTrans->mTraceId;
32,200✔
1074
  pTrans->pRpcArray = NULL;
32,200✔
1075
  pTrans->rpcRsp = NULL;
32,200✔
1076
  pTrans->rpcRspLen = 0;
32,200✔
1077

1078
  mndTransExecute(pMnode, pNew);
32,200✔
1079
  mndReleaseTrans(pMnode, pNew);
32,200✔
1080
  // TDOD change to TAOS_RETURN(code);
1081
  return 0;
32,200✔
1082
}
1083

1084
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
32,196✔
1085
  int32_t code = 0;
32,196✔
1086
  mInfo("trans:%d, commit transaction", pTrans->id);
32,196!
1087
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
32,196✔
1088
    mError("trans:%d, failed to commit since %s", pTrans->id, tstrerror(code));
2!
1089
    TAOS_RETURN(code);
2✔
1090
  }
1091
  mInfo("trans:%d, commit finished", pTrans->id);
32,194!
1092
  TAOS_RETURN(code);
32,194✔
1093
}
1094

1095
static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
2✔
1096
  int32_t code = 0;
2✔
1097
  mInfo("trans:%d, rollback transaction", pTrans->id);
2!
1098
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
2!
UNCOV
1099
    mError("trans:%d, failed to rollback since %s", pTrans->id, tstrerror(code));
×
UNCOV
1100
    TAOS_RETURN(code);
×
1101
  }
1102
  mInfo("trans:%d, rollback finished", pTrans->id);
2!
1103
  TAOS_RETURN(code);
2✔
1104
}
1105

1106
static int32_t mndTransPreFinish(SMnode *pMnode, STrans *pTrans) {
2✔
1107
  int32_t code = 0;
2✔
1108
  mInfo("trans:%d, pre-finish transaction", pTrans->id);
2!
1109
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
2!
1110
    mError("trans:%d, failed to pre-finish since %s", pTrans->id, tstrerror(code));
×
1111
    TAOS_RETURN(code);
×
1112
  }
1113
  mInfo("trans:%d, pre-finish finished", pTrans->id);
2!
1114
  TAOS_RETURN(code);
2✔
1115
}
1116

1117
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
191,975✔
1118
  bool    sendRsp = false;
191,975✔
1119
  int32_t code = pTrans->code;
191,975✔
1120

1121
  if (pTrans->stage == TRN_STAGE_FINISH) {
191,975✔
1122
    sendRsp = true;
69,420✔
1123
  }
1124

1125
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
191,975✔
1126
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
40,424!
1127
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
10!
1128
      sendRsp = true;
10✔
1129
    }
1130
  } else {
1131
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
151,551✔
1132
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
99,895!
1133
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1134
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1135
      } else {
1136
        if (pTrans->failedTimes > 6) sendRsp = true;
99,895✔
1137
      }
1138
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
99,895✔
1139
    }
1140
  }
1141

1142
  if (!sendRsp) {
191,975✔
1143
    return;
122,543✔
1144
  } else {
1145
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
69,432!
1146
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1147
  }
1148

1149
  mInfo("vgId:1, trans:%d, start to lock rpc array", pTrans->id);
69,432!
1150
  taosWLockLatch(&pTrans->lockRpcArray);
69,432✔
1151
  int32_t size = taosArrayGetSize(pTrans->pRpcArray);
69,432✔
1152
  if (size <= 0) {
69,432✔
1153
    taosWUnLockLatch(&pTrans->lockRpcArray);
47,382✔
1154
    return;
47,382✔
1155
  }
1156

1157
  for (int32_t i = 0; i < size; ++i) {
44,100✔
1158
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
22,050✔
1159
    if (pInfo->handle != NULL) {
22,050✔
1160
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) {
20,756!
1161
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
1✔
1162
      }
1163
      if (code == TSDB_CODE_SYN_TIMEOUT) {
20,756!
1164
        code = TSDB_CODE_MND_TRANS_SYNC_TIMEOUT;
×
1165
      }
1166

1167
      if (i != 0 && code == 0) {
20,756!
1168
        code = TSDB_CODE_MNODE_NOT_FOUND;
×
1169
      }
1170
      mInfo("vgId:1, trans:%d, client:%d start to send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code,
20,756!
1171
            mndTransStr(pTrans->stage), pInfo->ahandle);
1172

1173
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
20,756✔
1174

1175
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
20,756✔
1176
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
3,175!
1177
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
3,175✔
1178
        if (pDb != NULL) {
3,175!
1179
          for (int32_t j = 0; j < 12; j++) {
3,700!
1180
            bool ready = mndIsDbReady(pMnode, pDb);
3,700✔
1181
            if (!ready) {
3,700✔
1182
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
525!
1183
              taosMsleep(1000);
525✔
1184
            } else {
1185
              break;
3,175✔
1186
            }
1187
          }
1188
        }
1189
        mndReleaseDb(pMnode, pDb);
3,175✔
1190
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
17,581✔
1191
        void   *pCont = NULL;
4,938✔
1192
        int32_t contLen = 0;
4,938✔
1193
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
4,938✔
1194
          mndTransSetRpcRsp(pTrans, pCont, contLen);
4,936✔
1195
        }
1196
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
12,643✔
1197
        void   *pCont = NULL;
11✔
1198
        int32_t contLen = 0;
11✔
1199
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
11!
1200
          mndTransSetRpcRsp(pTrans, pCont, contLen);
11✔
1201
        }
1202
      }
1203

1204
      if (pTrans->rpcRspLen != 0) {
20,756✔
1205
        void *rpcCont = rpcMallocCont(pTrans->rpcRspLen);
9,331✔
1206
        if (rpcCont != NULL) {
9,331!
1207
          memcpy(rpcCont, pTrans->rpcRsp, pTrans->rpcRspLen);
9,331✔
1208
          rspMsg.pCont = rpcCont;
9,331✔
1209
          rspMsg.contLen = pTrans->rpcRspLen;
9,331✔
1210
        }
1211
      }
1212

1213
      tmsgSendRsp(&rspMsg);
20,756✔
1214

1215
      mInfo("vgId:1, trans:%d, client:%d send rsp finished, code:0x%x stage:%s app:%p", pTrans->id, i, code,
20,756!
1216
            mndTransStr(pTrans->stage), pInfo->ahandle);
1217
    }
1218
  }
1219
  taosArrayClear(pTrans->pRpcArray);
22,050✔
1220
  taosWUnLockLatch(&pTrans->lockRpcArray);
22,050✔
1221
}
1222

1223
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
75,951✔
1224
  int32_t code = 0;
75,951✔
1225
  SMnode *pMnode = pRsp->info.node;
75,951✔
1226
  int64_t signature = (int64_t)(pRsp->info.ahandle);
75,951✔
1227
  int32_t transId = (int32_t)(signature >> 32);
75,951✔
1228
  int32_t action = (int32_t)((signature << 32) >> 32);
75,951✔
1229

1230
  STrans *pTrans = mndAcquireTrans(pMnode, transId);
75,951✔
1231
  if (pTrans == NULL) {
75,951!
1232
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1233
    if (terrno != 0) code = terrno;
×
1234
    mError("trans:%d, failed to get transId from vnode rsp since %s", transId, tstrerror(code));
×
1235
    goto _OVER;
×
1236
  }
1237

1238
  SArray *pArray = NULL;
75,951✔
1239
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
75,951✔
1240
    pArray = pTrans->redoActions;
75,943✔
1241
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
8!
1242
    pArray = pTrans->undoActions;
8✔
1243
  } else {
1244
    mError("trans:%d, invalid trans stage:%d while recv action rsp", pTrans->id, pTrans->stage);
×
1245
    goto _OVER;
×
1246
  }
1247

1248
  if (pArray == NULL) {
75,951!
1249
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
×
1250
    goto _OVER;
×
1251
  }
1252

1253
  int32_t actionNum = taosArrayGetSize(pArray);
75,951✔
1254
  if (action < 0 || action >= actionNum) {
75,951!
1255
    mError("trans:%d, invalid action:%d", transId, action);
×
1256
    goto _OVER;
×
1257
  }
1258

1259
  STransAction *pAction = taosArrayGet(pArray, action);
75,951✔
1260
  if (pAction != NULL) {
75,951!
1261
    pAction->msgReceived = 1;
75,951✔
1262
    pAction->errCode = pRsp->code;
75,951✔
1263
    pTrans->lastErrorNo = pRsp->code;
75,951✔
1264

1265
    mInfo("trans:%d, %s:%d response is received, received code:0x%x(%s), accept:0x%x(%s) retry:0x%x(%s)", transId,
75,951!
1266
          mndTransStr(pAction->stage), action, pRsp->code, tstrerror(pRsp->code), pAction->acceptableCode,
1267
          tstrerror(pAction->acceptableCode), pAction->retryCode, tstrerror(pAction->retryCode));
1268
  } else {
1269
    mInfo("trans:%d, invalid action, index:%d, code:0x%x", transId, action, pRsp->code);
×
1270
  }
1271

1272
  mndTransExecute(pMnode, pTrans);
75,951✔
1273

1274
_OVER:
75,951✔
1275
  mndReleaseTrans(pMnode, pTrans);
75,951✔
1276
  TAOS_RETURN(code);
75,951✔
1277
}
1278

1279
static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
4,292✔
1280
  pAction->rawWritten = 0;
4,292✔
1281
  pAction->msgSent = 0;
4,292✔
1282
  pAction->msgReceived = 0;
4,292✔
1283
  if (pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR || pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR ||
4,292!
1284
      pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
4,292✔
1285
    pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
18✔
1286
    mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
18!
1287
          pAction->id, pAction->epSet.inUse);
1288
  } else {
1289
    mInfo("trans:%d, %s:%d execute status is reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
4,274!
1290
  }
1291
  pAction->errCode = 0;
4,292✔
1292
}
4,292✔
1293

1294
static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
2✔
1295
  int32_t numOfActions = taosArrayGetSize(pArray);
2✔
1296

1297
  for (int32_t action = 0; action < numOfActions; ++action) {
10✔
1298
    STransAction *pAction = taosArrayGet(pArray, action);
8✔
1299
    if (pAction->msgSent && pAction->msgReceived &&
8!
1300
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
8!
1301
      continue;
6✔
1302
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
2!
1303

1304
    mndTransResetAction(pMnode, pTrans, pAction);
2✔
1305
  }
1306
}
2✔
1307

1308
// execute in sync context
1309
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
351,052✔
1310
  if (pAction->rawWritten) return 0;
351,052✔
1311
  if (topHalf) {
190,242!
1312
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
×
1313
  }
1314

1315
  if (pAction->pRaw->type >= SDB_MAX) {
190,242!
1316
    pAction->rawWritten = true;
×
1317
    pAction->errCode = 0;
×
1318
    mndSetTransLastAction(pTrans, pAction);
×
1319
    mInfo("skip sdb raw type:%d since it is not supported", pAction->pRaw->type);
×
1320
    TAOS_RETURN(TSDB_CODE_SUCCESS);
×
1321
  }
1322

1323
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pAction->pRaw);
190,242✔
1324
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
190,242!
1325
    pAction->rawWritten = true;
190,242✔
1326
    pAction->errCode = 0;
190,242✔
1327
    code = 0;
190,242✔
1328
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
190,242!
1329
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1330

1331
    mndSetTransLastAction(pTrans, pAction);
190,242✔
1332
  } else {
1333
    pAction->errCode = (terrno != 0) ? terrno : code;
×
1334
    mError("trans:%d, %s:%d failed to write sdb since %s, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage),
×
1335
           pAction->id, tstrerror(code), sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1336
    mndSetTransLastAction(pTrans, pAction);
×
1337
  }
1338

1339
  TAOS_RETURN(code);
190,242✔
1340
}
1341

1342
// execute in trans context
1343
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
539,472✔
1344
  if (pAction->msgSent) return 0;
539,472✔
1345
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
103,813✔
1346
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
27,857✔
1347
  }
1348

1349
  int64_t signature = pTrans->id;
75,956✔
1350
  signature = (signature << 32);
75,956✔
1351
  signature += pAction->id;
75,956✔
1352

1353
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
75,956✔
1354
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
75,956✔
1355
  if (rpcMsg.pCont == NULL) {
75,956!
1356
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1357
    return -1;
×
1358
  }
1359
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
75,956✔
1360
  rpcMsg.info.notFreeAhandle = 1;
75,956✔
1361

1362
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
75,956✔
1363

1364
  char    detail[1024] = {0};
75,956✔
1365
  int32_t len = tsnprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
75,956!
1366
                          pAction->epSet.numOfEps, pAction->epSet.inUse);
75,956✔
1367
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
153,935✔
1368
    len += tsnprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
77,979✔
1369
                     pAction->epSet.eps[i].port);
77,979✔
1370
  }
1371

1372
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
75,956✔
1373
  if (code == 0) {
75,956!
1374
    pAction->msgSent = 1;
75,956✔
1375
    // pAction->msgReceived = 0;
1376
    pAction->errCode = TSDB_CODE_ACTION_IN_PROGRESS;
75,956✔
1377
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
75,956!
1378

1379
    mndSetTransLastAction(pTrans, pAction);
75,956✔
1380
  } else {
UNCOV
1381
    pAction->msgSent = 0;
×
UNCOV
1382
    pAction->msgReceived = 0;
×
UNCOV
1383
    pAction->errCode = (terrno != 0) ? terrno : code;
×
UNCOV
1384
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
×
1385
           detail);
1386

UNCOV
1387
    mndSetTransLastAction(pTrans, pAction);
×
1388
  }
1389

1390
  TAOS_RETURN(code);
75,956✔
1391
}
1392

1393
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
×
1394
  if (!topHalf) return TSDB_CODE_MND_TRANS_CTX_SWITCH;
×
1395
  pAction->rawWritten = 0;
×
1396
  pAction->errCode = 0;
×
1397
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1398

1399
  mndSetTransLastAction(pTrans, pAction);
×
1400
  return 0;
×
1401
}
1402

1403
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
890,524✔
1404
  if (pAction->actionType == TRANS_ACTION_RAW) {
890,524✔
1405
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
351,052✔
1406
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
539,472!
1407
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf);
539,472✔
1408
  } else {
1409
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1410
  }
1411
}
1412

1413
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf) {
182,274✔
1414
  int32_t numOfActions = taosArrayGetSize(pArray);
182,274✔
1415
  int32_t code = 0;
182,274✔
1416

1417
  for (int32_t action = 0; action < numOfActions; ++action) {
1,002,835✔
1418
    STransAction *pAction = taosArrayGet(pArray, action);
844,522✔
1419
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
844,522✔
1420
    if (code != 0) {
844,522✔
1421
      mInfo("trans:%d, action:%d not executed since %s. numOfActions:%d", pTrans->id, action, tstrerror(code),
23,961!
1422
            numOfActions);
1423
      break;
23,961✔
1424
    }
1425
  }
1426

1427
  return code;
182,274✔
1428
}
1429

1430
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf) {
205,585✔
1431
  int32_t numOfActions = taosArrayGetSize(pArray);
205,585✔
1432
  int32_t code = 0;
205,585✔
1433
  if (numOfActions == 0) return 0;
205,585✔
1434

1435
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf)) != 0) {
182,274✔
1436
    return code;
23,961✔
1437
  }
1438

1439
  int32_t       numOfExecuted = 0;
158,313✔
1440
  int32_t       errCode = 0;
158,313✔
1441
  STransAction *pErrAction = NULL;
158,313✔
1442
  for (int32_t action = 0; action < numOfActions; ++action) {
978,874✔
1443
    STransAction *pAction = taosArrayGet(pArray, action);
820,561✔
1444
    if (pAction->msgReceived || pAction->rawWritten) {
820,561✔
1445
      numOfExecuted++;
573,099✔
1446
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
573,099✔
1447
        errCode = pAction->errCode;
4✔
1448
        pErrAction = pAction;
4✔
1449
      }
1450
    } else {
1451
      pErrAction = pAction;
247,462✔
1452
    }
1453
  }
1454

1455
  mndSetTransLastAction(pTrans, pErrAction);
158,313✔
1456

1457
  if (numOfExecuted == numOfActions) {
158,313✔
1458
    if (errCode == 0) {
89,134✔
1459
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
89,132!
1460
      return 0;
89,132✔
1461
    } else {
1462
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
2!
1463
      mndTransResetActions(pMnode, pTrans, pArray);
2✔
1464
      terrno = errCode;
2✔
1465
      return errCode;
2✔
1466
    }
1467
  } else {
1468
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
69,179!
1469

1470
    for (int32_t action = 0; action < numOfActions; ++action) {
494,287✔
1471
      STransAction *pAction = taosArrayGet(pArray, action);
425,108✔
1472
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
425,108✔
1473
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1474
             pAction->acceptableCode, pAction->retryCode);
1475
      if (pAction->msgSent) {
425,108!
1476
        if (pAction->msgReceived) {
425,108✔
1477
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
177,646✔
1478
            mndTransResetAction(pMnode, pTrans, pAction);
2✔
1479
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
2!
1480
          }
1481
        }
1482
      }
1483
    }
1484
    return TSDB_CODE_ACTION_IN_PROGRESS;
69,179✔
1485
  }
1486
}
1487

1488
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
136,157✔
1489
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf);
136,157✔
1490
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
136,157✔
1491
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
2!
1492
           mndStrExecutionContext(topHalf));
1493
  }
1494
  return code;
136,157✔
1495
}
1496

1497
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
12✔
1498
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions, topHalf);
12✔
1499
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
12!
UNCOV
1500
    mError("trans:%d, failed to execute undoActions since %s. in %s", pTrans->id, terrstr(),
×
1501
           mndStrExecutionContext(topHalf));
1502
  }
1503
  return code;
12✔
1504
}
1505

1506
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
69,416✔
1507
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf);
69,416✔
1508
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
69,416!
1509
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1510
           mndStrExecutionContext(topHalf));
1511
  }
1512
  return code;
69,416✔
1513
}
1514

1515
static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
18,584✔
1516
  int32_t code = 0;
18,584✔
1517
  int32_t numOfActions = taosArrayGetSize(pActions);
18,584✔
1518
  if (numOfActions == 0) return code;
18,584✔
1519

1520
  if (pTrans->actionPos >= numOfActions) {
18,582✔
1521
    return code;
1,327✔
1522
  }
1523

1524
  mInfo("trans:%d, execute %d actions serial, begin at action:%d, stage:%s", pTrans->id, numOfActions,
17,255!
1525
        pTrans->actionPos, mndTransStr(pTrans->stage));
1526

1527
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
26,625✔
1528
    STransAction *pAction = taosArrayGet(pActions, action);
25,468✔
1529

1530
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d", pTrans->id, pTrans->actionPos,
25,468!
1531
          mndTransStr(pAction->stage), pAction->actionType);
1532

1533
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
25,468✔
1534
    if (code == 0) {
25,468✔
1535
      if (pAction->msgSent) {
21,572✔
1536
        if (pAction->msgReceived) {
18,465✔
1537
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
7,906✔
1538
            code = pAction->errCode;
4,288✔
1539
            mndTransResetAction(pMnode, pTrans, pAction);
4,288✔
1540
          } else {
1541
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,618!
1542
          }
1543
        } else {
1544
          code = TSDB_CODE_ACTION_IN_PROGRESS;
10,559✔
1545
        }
1546
      } else if (pAction->rawWritten) {
3,107!
1547
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
3,107!
1548
          code = pAction->errCode;
×
1549
        } else {
1550
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,107!
1551
        }
1552
      } else {
1553
      }
1554
    }
1555

1556
    if (code == 0) {
25,468✔
1557
      pTrans->failedTimes = 0;
6,725✔
1558
    }
1559
    mndSetTransLastAction(pTrans, pAction);
25,468✔
1560

1561
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
25,468✔
1562
      pTrans->lastErrorNo = code;
5,521✔
1563
      pTrans->code = code;
5,521✔
1564
      mInfo("trans:%d, %s:%d, cannot execute next action in %s, code:%s", pTrans->id, mndTransStr(pAction->stage),
5,521!
1565
            action, mndStrExecutionContext(topHalf), tstrerror(code));
1566
      break;
5,521✔
1567
    }
1568

1569
    if (code == 0) {
19,947✔
1570
      pTrans->code = 0;
5,100✔
1571
      pTrans->actionPos++;
5,100✔
1572
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
5,100!
1573
            pAction->id);
1574
      (void)taosThreadMutexUnlock(&pTrans->mutex);
5,100✔
1575
      code = mndTransSync(pMnode, pTrans);
5,100✔
1576
      (void)taosThreadMutexLock(&pTrans->mutex);
5,100✔
1577
      if (code != 0) {
5,100!
1578
        pTrans->actionPos--;
×
1579
        pTrans->code = terrno;
×
1580
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1581
               mndTransStr(pAction->stage), pAction->id, terrstr());
1582
        break;
×
1583
      }
1584
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
14,847✔
1585
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
10,559!
1586
      break;
10,559✔
1587
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,288!
1588
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
36✔
1589
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,270!
1590
            code, tstrerror(code));
1591
      pTrans->lastErrorNo = code;
4,270✔
1592
      taosMsleep(300);
4,270✔
1593
      action--;
4,270✔
1594
      continue;
4,270✔
1595
    } else {
1596
      terrno = code;
18✔
1597
      pTrans->lastErrorNo = code;
18✔
1598
      pTrans->code = code;
18✔
1599
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and wait another schedule, failedTimes:%d", pTrans->id,
18!
1600
            mndTransStr(pAction->stage), pAction->id, code, tstrerror(code), pTrans->failedTimes);
1601
      break;
18✔
1602
    }
1603
  }
1604

1605
  return code;
17,255✔
1606
}
1607

1608
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
18,584✔
1609
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
18,584✔
1610
  (void)taosThreadMutexLock(&pTrans->mutex);
18,584✔
1611
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
18,584!
1612
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
18,584✔
1613
  }
1614
  (void)taosThreadMutexUnlock(&pTrans->mutex);
18,584✔
1615
  return code;
18,584✔
1616
}
1617

1618
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1619
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
×
1620
  (void)taosThreadMutexLock(&pTrans->mutex);
×
1621
  if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1622
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1623
  }
1624
  (void)taosThreadMutexUnlock(&pTrans->mutex);
×
1625
  return code;
×
1626
}
1627

1628
bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
37,247✔
1629
  bool    continueExec = true;
37,247✔
1630
  int32_t code = 0;
37,247✔
1631
  terrno = 0;
37,247✔
1632

1633
  int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions);
37,247✔
1634
  if (numOfActions == 0) goto _OVER;
37,247✔
1635

1636
  mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions);
13,237!
1637

1638
  for (int32_t action = 0; action < numOfActions; ++action) {
33,771✔
1639
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
20,534✔
1640
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
20,534✔
1641
    if (code != 0) {
20,534!
1642
      terrno = code;
×
1643
      mError("trans:%d, failed to execute prepare action:%d, numOfActions:%d, since %s", pTrans->id, action,
×
1644
             numOfActions, tstrerror(code));
1645
      return false;
×
1646
    }
1647
  }
1648

1649
_OVER:
13,237✔
1650
  pTrans->stage = TRN_STAGE_REDO_ACTION;
37,247✔
1651
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
37,247!
1652
  return continueExec;
37,247✔
1653
}
1654

1655
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
154,741✔
1656
  bool    continueExec = true;
154,741✔
1657
  int32_t code = 0;
154,741✔
1658
  terrno = 0;
154,741✔
1659

1660
  if (pTrans->exec == TRN_EXEC_SERIAL) {
154,741✔
1661
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
18,584✔
1662
  } else {
1663
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf);
136,157✔
1664
  }
1665

1666
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
154,741!
1667
    pTrans->lastErrorNo = code;
×
1668
    pTrans->code = code;
×
1669
    mInfo(
×
1670
        "trans:%d, failed to execute, will retry redo action stage in 100 ms , in %s, "
1671
        "continueExec:%d, code:%s",
1672
        pTrans->id, mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1673
    taosMsleep(100);
×
1674
    return true;
×
1675
  } else {
1676
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
154,741✔
1677
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
42,795!
1678
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1679
      return false;
42,795✔
1680
    }
1681
  }
1682
  terrno = code;
111,946✔
1683

1684
  if (code == 0) {
111,946✔
1685
    pTrans->code = 0;
32,196✔
1686
    pTrans->stage = TRN_STAGE_COMMIT;
32,196✔
1687
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
32,196!
1688
    continueExec = true;
32,196✔
1689
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
79,750!
1690
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
79,730!
1691
    continueExec = false;
79,730✔
1692
  } else {
1693
    pTrans->failedTimes++;
20✔
1694
    pTrans->code = terrno;
20✔
1695
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
20✔
1696
      if (pTrans->lastAction != 0) {
2!
1697
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
2✔
1698
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
2!
1699
          if (pTrans->failedTimes < 6) {
×
1700
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
×
1701
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
1702
            taosMsleep(1000);
×
1703
            continueExec = true;
×
1704
            return true;
×
1705
          }
1706
        }
1707
      }
1708

1709
      pTrans->stage = TRN_STAGE_ROLLBACK;
2✔
1710
      pTrans->actionPos = 0;
2✔
1711
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
2!
1712
             pTrans->actionPos);
1713
      continueExec = true;
2✔
1714
    } else {
1715
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
18!
1716
      continueExec = false;
18✔
1717
    }
1718
  }
1719

1720
  return continueExec;
111,946✔
1721
}
1722

1723
// execute in trans context
1724
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
32,196✔
1725
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
32,196!
1726

1727
  bool    continueExec = true;
32,196✔
1728
  int32_t code = mndTransCommit(pMnode, pTrans);
32,196✔
1729

1730
  if (code == 0) {
32,196✔
1731
    pTrans->code = 0;
32,194✔
1732
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
32,194✔
1733
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
32,194!
1734
    continueExec = true;
32,194✔
1735
  } else {
1736
    pTrans->code = terrno;
2✔
1737
    pTrans->failedTimes++;
2✔
1738
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
2!
1739
    continueExec = false;
2✔
1740
  }
1741

1742
  return continueExec;
32,196✔
1743
}
1744

1745
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
69,416✔
1746
  bool    continueExec = true;
69,416✔
1747
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans, topHalf);
69,416✔
1748

1749
  if (code == 0) {
69,416!
1750
    pTrans->code = 0;
69,416✔
1751
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
69,416✔
1752
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
69,416!
1753
    continueExec = true;
69,416✔
1754
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
×
1755
    pTrans->code = 0;
×
1756
    pTrans->stage = TRN_STAGE_COMMIT;
×
1757
    mInfo("trans:%d, back to commit stage", pTrans->id);
×
1758
    continueExec = true;
×
1759
  } else {
1760
    pTrans->code = terrno;
×
1761
    pTrans->failedTimes++;
×
1762
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1763
    continueExec = false;
×
1764
  }
1765

1766
  return continueExec;
69,416✔
1767
}
1768

1769
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
12✔
1770
  bool    continueExec = true;
12✔
1771
  int32_t code = 0;
12✔
1772

1773
  if (pTrans->exec == TRN_EXEC_SERIAL) {
12!
1774
    code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
×
1775
  } else {
1776
    code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf);
12✔
1777
  }
1778

1779
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
12✔
1780
  terrno = code;
10✔
1781

1782
  if (code == 0) {
10✔
1783
    pTrans->stage = TRN_STAGE_PRE_FINISH;
2✔
1784
    mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
2!
1785
    continueExec = true;
2✔
1786
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
8!
1787
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
8!
1788
    continueExec = false;
8✔
1789
  } else {
UNCOV
1790
    pTrans->failedTimes++;
×
UNCOV
1791
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
UNCOV
1792
    continueExec = false;
×
1793
  }
1794

1795
  return continueExec;
10✔
1796
}
1797

1798
// in trans context
1799
static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2✔
1800
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
2!
1801

1802
  bool    continueExec = true;
2✔
1803
  int32_t code = mndTransRollback(pMnode, pTrans);
2✔
1804

1805
  if (code == 0) {
2!
1806
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
2✔
1807
    continueExec = true;
2✔
1808
  } else {
UNCOV
1809
    pTrans->failedTimes++;
×
UNCOV
1810
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
UNCOV
1811
    continueExec = false;
×
1812
  }
1813

1814
  return continueExec;
2✔
1815
}
1816

1817
// excute in trans context
1818
static bool mndTransPerformPreFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2✔
1819
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
2!
1820

1821
  bool    continueExec = true;
2✔
1822
  int32_t code = mndTransPreFinish(pMnode, pTrans);
2✔
1823

1824
  if (code == 0) {
2!
1825
    pTrans->stage = TRN_STAGE_FINISH;
2✔
1826
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
2!
1827
    continueExec = true;
2✔
1828
  } else {
1829
    pTrans->failedTimes++;
×
1830
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1831
    continueExec = false;
×
1832
  }
1833

1834
  return continueExec;
2✔
1835
}
1836

1837
static bool mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
69,420✔
1838
  bool continueExec = false;
69,420✔
1839
  if (topHalf) return continueExec;
69,420✔
1840

1841
  SSdbRaw *pRaw = mndTransEncode(pTrans);
37,224✔
1842
  if (pRaw == NULL) {
37,224!
1843
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
×
1844
    return false;
×
1845
  }
1846
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED));
37,224!
1847

1848
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
37,224✔
1849
  if (code != 0) {
37,224!
1850
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
×
1851
  }
1852

1853
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d createTime:%" PRId64, pTrans->id, pTrans->code,
37,224!
1854
        pTrans->failedTimes, pTrans->createdTime);
1855
  return continueExec;
37,224✔
1856
}
1857

1858
void mndTransExecuteImp(SMnode *pMnode, STrans *pTrans, bool topHalf) {
191,975✔
1859
  bool continueExec = true;
191,975✔
1860

1861
  while (continueExec) {
517,764✔
1862
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64 "", pTrans->id,
325,789!
1863
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1864
    pTrans->lastExecTime = taosGetTimestampMs();
325,789✔
1865
    switch (pTrans->stage) {
325,789!
1866
      case TRN_STAGE_PREPARE:
×
1867
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1868
        break;
×
1869
      case TRN_STAGE_REDO_ACTION:
154,741✔
1870
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf);
154,741✔
1871
        break;
154,741✔
1872
      case TRN_STAGE_COMMIT:
32,196✔
1873
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
32,196✔
1874
        break;
32,196✔
1875
      case TRN_STAGE_COMMIT_ACTION:
69,416✔
1876
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
69,416✔
1877
        break;
69,416✔
1878
      case TRN_STAGE_ROLLBACK:
2✔
1879
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
2✔
1880
        break;
2✔
1881
      case TRN_STAGE_UNDO_ACTION:
12✔
1882
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf);
12✔
1883
        break;
12✔
1884
      case TRN_STAGE_PRE_FINISH:
2✔
1885
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
2✔
1886
        break;
2✔
1887
      case TRN_STAGE_FINISH:
69,420✔
1888
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
69,420✔
1889
        break;
69,420✔
1890
      default:
×
1891
        continueExec = false;
×
1892
        break;
×
1893
    }
1894
  }
1895

1896
  mndTransSendRpcRsp(pMnode, pTrans);
191,975✔
1897
}
191,975✔
1898

1899
// start trans, pullup, receive rsp, kill
1900
void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
111,994✔
1901
  bool topHalf = true;
111,994✔
1902
  mndTransExecuteImp(pMnode, pTrans, topHalf);
111,994✔
1903
}
111,994✔
1904

1905
// update trans
1906
void mndTransRefresh(SMnode *pMnode, STrans *pTrans) {
79,981✔
1907
  bool topHalf = false;
79,981✔
1908
  mndTransExecuteImp(pMnode, pTrans, topHalf);
79,981✔
1909
}
79,981✔
1910

1911
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
20,931✔
1912
  mTrace("start to process trans timer");
20,931✔
1913
  mndTransPullup(pReq->info.node);
20,931✔
1914
  return 0;
20,931✔
1915
}
1916

1917
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
×
1918
  SArray *pArray = NULL;
×
1919
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
×
1920
    pArray = pTrans->redoActions;
×
1921
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1922
    pArray = pTrans->undoActions;
×
1923
  } else {
1924
    TAOS_RETURN(TSDB_CODE_MND_TRANS_INVALID_STAGE);
×
1925
  }
1926

1927
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
1928
    STransAction *pAction = taosArrayGet(pArray, i);
×
1929
    mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
×
1930
          mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
1931
    pAction->msgSent = 1;
×
1932
    pAction->msgReceived = 1;
×
1933
    pAction->errCode = 0;
×
1934
  }
1935

1936
  mndTransExecute(pMnode, pTrans);
×
1937
  return 0;
×
1938
}
1939

1940
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
1✔
1941
  SMnode       *pMnode = pReq->info.node;
1✔
1942
  SKillTransReq killReq = {0};
1✔
1943
  int32_t       code = -1;
1✔
1944
  STrans       *pTrans = NULL;
1✔
1945

1946
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
1!
1947
    code = TSDB_CODE_INVALID_MSG;
×
1948
    goto _OVER;
×
1949
  }
1950

1951
  mInfo("trans:%d, start to kill", killReq.transId);
1!
1952
  if ((code = mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS)) != 0) {
1!
1953
    goto _OVER;
1✔
1954
  }
1955

1956
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
×
1957
  if (pTrans == NULL) {
×
1958
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1959
    if (terrno != 0) code = terrno;
×
1960
    goto _OVER;
×
1961
  }
1962

1963
  code = mndKillTrans(pMnode, pTrans);
×
1964

1965
_OVER:
1✔
1966
  if (code != 0) {
1!
1967
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
1!
1968
  }
1969

1970
  mndReleaseTrans(pMnode, pTrans);
1✔
1971
  TAOS_RETURN(code);
1✔
1972
}
1973

1974
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
54✔
1975

1976
void mndTransPullup(SMnode *pMnode) {
21,207✔
1977
  SSdb   *pSdb = pMnode->pSdb;
21,207✔
1978
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
21,207✔
1979
  if (pArray == NULL) return;
21,207!
1980

1981
  void *pIter = NULL;
21,207✔
1982
  while (1) {
3,843✔
1983
    STrans *pTrans = NULL;
25,050✔
1984
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
25,050✔
1985
    if (pIter == NULL) break;
25,050✔
1986
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
7,686!
1987
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
1988
    }
1989
    sdbRelease(pSdb, pTrans);
3,843✔
1990
  }
1991

1992
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
21,207✔
1993

1994
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
25,050✔
1995
    int32_t *pTransId = taosArrayGet(pArray, i);
3,843✔
1996
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
3,843✔
1997
    if (pTrans != NULL) {
3,843!
1998
      mndTransExecute(pMnode, pTrans);
3,843✔
1999
    }
2000
    mndReleaseTrans(pMnode, pTrans);
3,843✔
2001
  }
2002
  taosArrayDestroy(pArray);
21,207✔
2003
}
2004

2005
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
146✔
2006
  SMnode *pMnode = pReq->info.node;
146✔
2007
  SSdb   *pSdb = pMnode->pSdb;
146✔
2008
  int32_t numOfRows = 0;
146✔
2009
  STrans *pTrans = NULL;
146✔
2010
  int32_t cols = 0;
146✔
2011
  int32_t code = 0;
146✔
2012
  int32_t lino = 0;
146✔
2013

2014
  while (numOfRows < rows) {
176!
2015
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
176✔
2016
    if (pShow->pIter == NULL) break;
176✔
2017

2018
    cols = 0;
30✔
2019

2020
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2021
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->id, false), pTrans, &lino, _OVER);
30!
2022

2023
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2024
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false), pTrans, &lino,
30!
2025
                        _OVER);
2026

2027
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
30✔
2028
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
30✔
2029
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2030
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stage, false), pTrans, &lino, _OVER);
30!
2031

2032
    char opername[TSDB_TRANS_OPER_LEN + VARSTR_HEADER_SIZE] = {0};
30✔
2033
    STR_WITH_MAXSIZE_TO_VARSTR(opername, pTrans->opername, pShow->pMeta->pSchemas[cols].bytes);
30✔
2034
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2035
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)opername, false), pTrans, &lino, _OVER);
30!
2036

2037
    char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
30✔
2038
    STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pTrans->dbname), pShow->pMeta->pSchemas[cols].bytes);
30✔
2039
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2040
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pTrans, &lino, _OVER);
30!
2041

2042
    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
30✔
2043
    STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
30✔
2044
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2045
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbname, false), pTrans, &lino, _OVER);
30!
2046

2047
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2048
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
30!
2049
                        _OVER);
2050

2051
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2052
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false), pTrans, &lino,
30!
2053
                        _OVER);
2054

2055
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
30✔
2056
    char    detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
30✔
2057
    int32_t len = tsnprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
90✔
2058
                            pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
30✔
2059
    SEpSet  epset = pTrans->lastEpset;
30✔
2060
    if (epset.numOfEps > 0) {
30!
2061
      len += tsnprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
60!
2062
                       TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
60✔
2063
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
66✔
2064
        len += tsnprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
36✔
2065
      }
2066
    }
2067
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
30✔
2068
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
30✔
2069
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)lastInfo, false), pTrans, &lino, _OVER);
30!
2070

2071
    numOfRows++;
30✔
2072
    sdbRelease(pSdb, pTrans);
30✔
2073
  }
2074

2075
_OVER:
×
2076
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
146!
2077
  pShow->numOfRows += numOfRows;
146✔
2078
  return numOfRows;
146✔
2079
}
2080

2081
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
×
2082
  SSdb *pSdb = pMnode->pSdb;
×
2083
  sdbCancelFetchByType(pSdb, pIter, SDB_TRANS);
×
2084
}
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc