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

taosdata / TDengine / #3559

18 Dec 2024 12:59AM UTC coverage: 59.805% (+0.03%) from 59.778%
#3559

push

travis-ci

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

merge: main to 3.0 branch

132705 of 287544 branches covered (46.15%)

Branch coverage included in aggregate %.

87 of 95 new or added lines in 19 files covered. (91.58%)

1132 existing lines in 133 files now uncovered.

209591 of 284807 relevant lines covered (73.59%)

8125235.78 hits per line

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

70.41
/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; }
407,973✔
56

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

64
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
408,343✔
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,518✔
77
  SSdbTable table = {
1,518✔
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,518✔
88
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
1,518✔
89

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

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

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

101
  for (int32_t i = 0; i < actionNum; ++i) {
1,511,780✔
102
    STransAction *pAction = taosArrayGet(pArray, i);
1,047,792✔
103
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,047,792✔
104
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
642,807✔
105
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
404,985!
106
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
404,985✔
107
    } else {
108
      // empty
109
    }
110
    rawDataLen += sizeof(int8_t);
1,047,792✔
111
  }
112

113
  return rawDataLen;
463,988✔
114
}
115

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

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

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

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

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

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

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

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

192
  if (sver > TRANS_VER1_NUMBER) {
115,997✔
193
    SDB_SET_INT32(pRaw, dataPos, prepareActionNum, _OVER)
35,824!
194
  }
195
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
115,997!
196
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
115,997!
197
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
115,997!
198

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

204
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
115,997!
205
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
115,997!
206
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
115,997!
207
  if (pTrans->param != NULL) {
115,997!
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)
115,997!
212

213
  int32_t arbGroupNum = taosHashGetSize(pTrans->arbGroupIds);
115,997✔
214
  SDB_SET_INT32(pRaw, dataPos, arbGroupNum, _OVER)
115,997!
215
  void *pIter = NULL;
115,997✔
216
  pIter = taosHashIterate(pTrans->arbGroupIds, NULL);
115,997✔
217
  while (pIter) {
116,015✔
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)
115,997!
224
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
115,997!
225

226
  terrno = 0;
115,997✔
227

228
_OVER:
115,997✔
229
  if (terrno != 0) {
115,997!
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);
115,997✔
237
  return pRaw;
115,997✔
238
}
239

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

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

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

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

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

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

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

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

324
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
223,973!
325

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

350
  if (sver > TRANS_VER1_NUMBER) {
223,973✔
351
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
77,602!
352
  }
353
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
223,973!
354
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
223,973!
355
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
223,973!
356

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

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

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

372
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
223,973!
373
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
223,973!
374
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
223,973!
375
  if (pTrans->paramLen != 0) {
223,973!
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);
223,973!
382

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

385
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
223,973!
386
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
224,003✔
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)
223,973!
393

394
  terrno = 0;
223,973✔
395

396
_OVER:
223,973✔
397
  if (terrno != 0 && pTrans != NULL) {
223,973!
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) {
223,973!
405
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
223,973✔
406
  }
407
  return pRow;
223,973✔
408
}
409

410
static const char *mndTransStr(ETrnStage stage) {
1,570,153✔
411
  switch (stage) {
1,570,153!
412
    case TRN_STAGE_PREPARE:
98,583✔
413
      return "prepare";
98,583✔
414
    case TRN_STAGE_REDO_ACTION:
743,842✔
415
      return "redoAction";
743,842✔
416
    case TRN_STAGE_ROLLBACK:
48✔
417
      return "rollback";
48✔
418
    case TRN_STAGE_UNDO_ACTION:
322✔
419
      return "undoAction";
322✔
420
    case TRN_STAGE_COMMIT:
186,706✔
421
      return "commit";
186,706✔
422
    case TRN_STAGE_COMMIT_ACTION:
259,274✔
423
      return "commitAction";
259,274✔
424
    case TRN_STAGE_FINISH:
281,332✔
425
      return "finished";
281,332✔
426
    case TRN_STAGE_PRE_FINISH:
46✔
427
      return "pre-finish";
46✔
428
    default:
×
429
      return "invalid";
×
430
  }
431
}
432

433
static void mndSetTransLastAction(STrans *pTrans, STransAction *pAction) {
484,878✔
434
  if (pAction != NULL) {
484,878✔
435
    pTrans->lastAction = pAction->id;
387,765✔
436
    pTrans->lastMsgType = pAction->msgType;
387,765✔
437
    pTrans->lastEpset = pAction->epSet;
387,765✔
438
    pTrans->lastErrorNo = pAction->errCode;
387,765✔
439
  } else {
440
    pTrans->lastAction = 0;
97,113✔
441
    pTrans->lastMsgType = 0;
97,113✔
442
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
97,113✔
443
    pTrans->lastErrorNo = 0;
97,113✔
444
  }
445
}
484,878✔
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,986✔
456
  switch (ftype) {
1,986!
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:
993✔
462
      return mndRebCntInc;
993✔
463
    case TRANS_STOP_FUNC_MQ_REB:
993✔
464
      return mndRebCntDec;
993✔
465
    default:
×
466
      return NULL;
×
467
  }
468
}
469

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

474
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
42,411✔
475

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

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

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

494
  if (pTrans->stage == TRN_STAGE_PRE_FINISH) {
42,411!
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;
42,411✔
500
}
501

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

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

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

551
  mndTransDropData(pTrans);
133,175✔
552
  return 0;
133,175✔
553
}
554

555
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
193,916✔
556
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
655,318✔
557
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
461,402✔
558
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
461,402✔
559
    pOldAction->rawWritten = pNewAction->rawWritten;
461,402✔
560
    pOldAction->msgSent = pNewAction->msgSent;
461,402✔
561
    pOldAction->msgReceived = pNewAction->msgReceived;
461,402✔
562
    pOldAction->errCode = pNewAction->errCode;
461,402✔
563
  }
564
}
193,916✔
565

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

570
  if (pOld->createdTime != pNew->createdTime) {
48,480✔
571
    mError("trans:%d, failed to perform update action since createTime not match, old row:%p stage:%s create:%" PRId64
1!
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
576
    TAOS_RETURN(TSDB_CODE_MND_TRANS_SYNC_TIMEOUT);
1✔
577
  }
578

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

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

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

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

601
  return 0;
48,479✔
602
}
603

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

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

618
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
34,409✔
619
                       const char *opername) {
620
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
34,409!
621
  if (pTrans == NULL) {
34,409!
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) {
34,409!
628
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
34,409✔
629
  }
630

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

650
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
34,409!
651
      pTrans->pRpcArray == NULL) {
34,409!
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) {
34,409✔
659
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
47,570!
660
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
661
      return NULL;
×
662
    }
663
    pTrans->originRpcType = pReq->msgType;
23,785✔
664
  }
665

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

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

672
static void mndTransDropActions(SArray *pArray) {
1,033,528✔
673
  int32_t size = taosArrayGetSize(pArray);
1,033,528✔
674
  for (int32_t i = 0; i < size; ++i) {
3,289,828✔
675
    STransAction *pAction = taosArrayGet(pArray, i);
2,256,300✔
676
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,256,300✔
677
      taosMemoryFreeClear(pAction->pRaw);
1,361,649!
678
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
894,651!
679
      taosMemoryFreeClear(pAction->pCont);
894,651!
680
    } else {
681
      // nothing
682
    }
683
  }
684

685
  taosArrayDestroy(pArray);
1,033,528✔
686
}
1,033,528✔
687

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

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

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

704
  return 0;
298,549✔
705
}
706

707
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
1,686✔
708
  STransAction action = {
1,686✔
709
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
1,686✔
710
  return mndTransAppendAction(pTrans->redoActions, &action);
1,686✔
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) {
11,220✔
719
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
11,220✔
720
  return mndTransAppendAction(pTrans->undoActions, &action);
11,220✔
721
}
722

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

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

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

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

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

752
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
983✔
753
  pTrans->startFunc = startFunc;
983✔
754
  pTrans->stopFunc = stopFunc;
983✔
755
  pTrans->param = param;
983✔
756
  pTrans->paramLen = paramLen;
983✔
757
}
983✔
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) {
24,292✔
791
  if (dbname != NULL) {
24,292!
792
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
24,292✔
793
  }
794
  if (stbname != NULL) {
24,292✔
795
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
18,062✔
796
  }
797
}
24,292✔
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,285✔
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,214✔
812

813
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
73,529✔
814
  int32_t  code = 0;
73,529✔
815
  SSdbRaw *pRaw = mndTransEncode(pTrans);
73,529✔
816
  if (pRaw == NULL) {
73,529!
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));
73,529!
823

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

834
  sdbFreeRaw(pRaw);
73,510✔
835
  mInfo("trans:%d, sync finished, createTime:%" PRId64, pTrans->id, pTrans->createdTime);
73,510!
836
  TAOS_RETURN(code);
73,510✔
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) {
16,990✔
846
  if (conflict[0] == 0) return false;
16,990✔
847
  if (strcasecmp(conflict, pTrans->stbname) == 0) return true;
16,988✔
848
  return false;
16,944✔
849
}
850

851
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
17,219✔
852
  if (conflict) {
17,219✔
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,
267!
854
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
855
    *globalConflict = true;
267✔
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,
16,952!
858
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
859
  }
860
}
17,219✔
861

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

867
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
101,114✔
868

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

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

876
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
18,698✔
877

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

886
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
18,698✔
887
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
17,641✔
888
      if (pTrans->conflict == TRN_CONFLICT_DB) {
17,641✔
889
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
227✔
890
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
227✔
891
      }
892
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
17,641✔
893
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
16,761✔
894
      }
895
    }
896

897
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
18,698!
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) {
18,698!
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);
18,698✔
924
  }
925

926
  return conflict;
72,667✔
927
}
928

929
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
101,114✔
930
  int32_t code = 0;
101,114✔
931
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
101,114✔
932
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
64,416!
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)) {
101,114✔
940
    code = TSDB_CODE_MND_TRANS_CONFLICT;
293✔
941
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
293!
942
    TAOS_RETURN(code);
293✔
943
  }
944

945
  TAOS_RETURN(code);
100,821✔
946
}
947

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

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

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

966
    if (thisConflict) {
4!
967
      mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with compact:%d db:%s", pTrans->id,
4!
968
             pTrans->dbname, pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
969
      conflict = true;
4✔
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
    }
974
    sdbRelease(pMnode->pSdb, pCompact);
4✔
975
  }
976

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

983
  TAOS_RETURN(code);
167✔
984
}
985

986
static bool mndTransActionsOfSameType(SArray *pActions) {
76,111✔
987
  int32_t size = taosArrayGetSize(pActions);
76,111✔
988
  ETrnAct lastActType = TRANS_ACTION_NULL;
76,111✔
989
  bool    same = true;
76,111✔
990
  for (int32_t i = 0; i < size; ++i) {
328,155✔
991
    STransAction *pAction = taosArrayGet(pActions, i);
252,044✔
992
    if (i > 0) {
252,044✔
993
      if (lastActType != pAction->actionType) {
191,433!
994
        same = false;
×
995
        break;
×
996
      }
997
    }
998
    lastActType = pAction->actionType;
252,044✔
999
  }
1000
  return same;
76,111✔
1001
}
1002

1003
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
34,073✔
1004
  int32_t code = 0;
34,073✔
1005
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
34,073✔
1006
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
32,857!
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) {
32,857✔
1013
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
9,181!
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);
34,073✔
1022
}
1023

1024
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
34,073✔
1025
  int32_t code = 0;
34,073✔
1026
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
34,073!
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) {
34,073!
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);
34,073✔
1038
}
1039

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

1046
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
34,073!
1047

1048
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
34,073!
1049

1050
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
34,073!
1051

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

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

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

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

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

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

1106
static int32_t mndTransPreFinish(SMnode *pMnode, STrans *pTrans) {
6✔
1107
  int32_t code = 0;
6✔
1108
  mInfo("trans:%d, pre-finish transaction", pTrans->id);
6!
1109
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
6!
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);
6!
1114
  TAOS_RETURN(code);
6✔
1115
}
1116

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

1121
  if (pTrans->stage == TRN_STAGE_FINISH) {
209,094✔
1122
    sendRsp = true;
76,341✔
1123
  }
1124

1125
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
209,094✔
1126
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
51,764✔
1127
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
54✔
1128
      sendRsp = true;
54✔
1129
    }
1130
  } else {
1131
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
157,330✔
1132
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
103,279!
1133
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1134
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1135
      } else {
1136
        if (pTrans->failedTimes > 6) sendRsp = true;
103,279✔
1137
      }
1138
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
103,279✔
1139
    }
1140
  }
1141

1142
  if (!sendRsp) {
209,094✔
1143
    return;
132,697✔
1144
  } else {
1145
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
76,397!
1146
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1147
  }
1148

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

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

1167
      if (i != 0 && code == 0) {
22,047!
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,
22,047!
1171
            mndTransStr(pTrans->stage), pInfo->ahandle);
1172

1173
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
22,047✔
1174

1175
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
22,047✔
1176
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
3,214!
1177
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
3,214✔
1178
        if (pDb != NULL) {
3,214!
1179
          for (int32_t j = 0; j < 12; j++) {
3,800!
1180
            bool ready = mndIsDbReady(pMnode, pDb);
3,800✔
1181
            if (!ready) {
3,800✔
1182
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
586!
1183
              taosMsleep(1000);
586✔
1184
            } else {
1185
              break;
3,214✔
1186
            }
1187
          }
1188
        }
1189
        mndReleaseDb(pMnode, pDb);
3,214✔
1190
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
18,833✔
1191
        void   *pCont = NULL;
5,770✔
1192
        int32_t contLen = 0;
5,770✔
1193
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
5,770✔
1194
          mndTransSetRpcRsp(pTrans, pCont, contLen);
5,765✔
1195
        }
1196
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
13,063✔
1197
        void   *pCont = NULL;
10✔
1198
        int32_t contLen = 0;
10✔
1199
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
10!
1200
          mndTransSetRpcRsp(pTrans, pCont, contLen);
10✔
1201
        }
1202
      }
1203

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

1213
      tmsgSendRsp(&rspMsg);
22,047✔
1214

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

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

1230
  STrans *pTrans = mndAcquireTrans(pMnode, transId);
80,061✔
1231
  if (pTrans == NULL) {
80,061!
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;
80,061✔
1239
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
80,061✔
1240
    pArray = pTrans->redoActions;
80,036✔
1241
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
25!
1242
    pArray = pTrans->undoActions;
25✔
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) {
80,061!
1249
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
×
1250
    goto _OVER;
×
1251
  }
1252

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

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

1265
    mInfo("trans:%d, %s:%d response is received, received code:0x%x(%s), accept:0x%x(%s) retry:0x%x(%s)", transId,
80,061!
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);
80,061✔
1273

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

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

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

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

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

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

1315
  if (pAction->pRaw->type >= SDB_MAX) {
206,977!
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);
206,977✔
1324
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
206,977!
1325
    pAction->rawWritten = true;
206,977✔
1326
    pAction->errCode = 0;
206,977✔
1327
    code = 0;
206,977✔
1328
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
206,977!
1329
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1330

1331
    mndSetTransLastAction(pTrans, pAction);
206,977✔
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);
206,977✔
1340
}
1341

1342
// execute in trans context
1343
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
564,128✔
1344
  if (pAction->msgSent) return 0;
564,128✔
1345
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
111,889✔
1346
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
31,824✔
1347
  }
1348

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

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

1362
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
80,065✔
1363

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

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

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

1387
    mndSetTransLastAction(pTrans, pAction);
1✔
1388
  }
1389

1390
  TAOS_RETURN(code);
80,065✔
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) {
938,801✔
1404
  if (pAction->actionType == TRANS_ACTION_RAW) {
938,801✔
1405
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
374,673✔
1406
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
564,128!
1407
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf);
564,128✔
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) {
197,053✔
1414
  int32_t numOfActions = taosArrayGetSize(pArray);
197,053✔
1415
  int32_t code = 0;
197,053✔
1416

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

1427
  return code;
197,053✔
1428
}
1429

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

1435
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf)) != 0) {
197,053✔
1436
    return code;
27,435✔
1437
  }
1438

1439
  int32_t       numOfExecuted = 0;
169,618✔
1440
  int32_t       errCode = 0;
169,618✔
1441
  STransAction *pErrAction = NULL;
169,618✔
1442
  for (int32_t action = 0; action < numOfActions; ++action) {
1,030,577✔
1443
    STransAction *pAction = taosArrayGet(pArray, action);
860,959✔
1444
    if (pAction->msgReceived || pAction->rawWritten) {
860,959✔
1445
      numOfExecuted++;
603,998✔
1446
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
603,998✔
1447
        errCode = pAction->errCode;
8✔
1448
        pErrAction = pAction;
8✔
1449
      }
1450
    } else {
1451
      pErrAction = pAction;
256,961✔
1452
    }
1453
  }
1454

1455
  mndSetTransLastAction(pTrans, pErrAction);
169,618✔
1456

1457
  if (numOfExecuted == numOfActions) {
169,618✔
1458
    if (errCode == 0) {
97,119✔
1459
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
97,113!
1460
      return 0;
97,113✔
1461
    } else {
1462
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
6!
1463
      mndTransResetActions(pMnode, pTrans, pArray);
6✔
1464
      terrno = errCode;
6✔
1465
      return errCode;
6✔
1466
    }
1467
  } else {
1468
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
72,499!
1469

1470
    for (int32_t action = 0; action < numOfActions; ++action) {
513,114✔
1471
      STransAction *pAction = taosArrayGet(pArray, action);
440,615✔
1472
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
440,615✔
1473
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1474
             pAction->acceptableCode, pAction->retryCode);
1475
      if (pAction->msgSent) {
440,615!
1476
        if (pAction->msgReceived) {
440,615✔
1477
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
183,654✔
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;
72,499✔
1485
  }
1486
}
1487

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

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

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

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

1520
  if (pTrans->actionPos >= numOfActions) {
20,554✔
1521
    return code;
1,479✔
1522
  }
1523

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

1527
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
29,432✔
1528
    STransAction *pAction = taosArrayGet(pActions, action);
28,218✔
1529

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

1533
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
28,218✔
1534
    if (code == 0) {
28,218✔
1535
      if (pAction->msgSent) {
23,828✔
1536
        if (pAction->msgReceived) {
20,438✔
1537
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
8,806✔
1538
            code = pAction->errCode;
4,985✔
1539
            mndTransResetAction(pMnode, pTrans, pAction);
4,985✔
1540
          } else {
1541
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,821!
1542
          }
1543
        } else {
1544
          code = TSDB_CODE_ACTION_IN_PROGRESS;
11,632✔
1545
        }
1546
      } else if (pAction->rawWritten) {
3,390!
1547
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
3,390!
1548
          code = pAction->errCode;
×
1549
        } else {
1550
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,390!
1551
        }
1552
      } else {
1553
      }
1554
    }
1555

1556
    if (code == 0) {
28,218✔
1557
      pTrans->failedTimes = 0;
7,211✔
1558
    }
1559
    mndSetTransLastAction(pTrans, pAction);
28,218✔
1560

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

1569
    if (code == 0) {
22,007✔
1570
      pTrans->code = 0;
5,390✔
1571
      pTrans->actionPos++;
5,390✔
1572
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
5,390!
1573
            pAction->id);
1574
      (void)taosThreadMutexUnlock(&pTrans->mutex);
5,390✔
1575
      code = mndTransSync(pMnode, pTrans);
5,390✔
1576
      (void)taosThreadMutexLock(&pTrans->mutex);
5,390✔
1577
      if (code != 0) {
5,390!
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) {
16,617✔
1585
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
11,632!
1586
      break;
11,632✔
1587
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,985!
1588
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
40✔
1589
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,967!
1590
            code, tstrerror(code));
1591
      pTrans->lastErrorNo = code;
4,967✔
1592
      taosMsleep(300);
4,967✔
1593
      action--;
4,967✔
1594
      continue;
4,967✔
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;
19,075✔
1606
}
1607

1608
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
20,556✔
1609
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
20,556✔
1610
  (void)taosThreadMutexLock(&pTrans->mutex);
20,556✔
1611
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
20,556!
1612
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
20,556✔
1613
  }
1614
  (void)taosThreadMutexUnlock(&pTrans->mutex);
20,556✔
1615
  return code;
20,556✔
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) {
42,318✔
1629
  bool    continueExec = true;
42,318✔
1630
  int32_t code = 0;
42,318✔
1631
  terrno = 0;
42,318✔
1632

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

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

1638
  for (int32_t action = 0; action < numOfActions; ++action) {
36,791✔
1639
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
22,189✔
1640
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
22,189✔
1641
    if (code != 0) {
22,189!
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:
14,602✔
1650
  pTrans->stage = TRN_STAGE_REDO_ACTION;
42,318✔
1651
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
42,318!
1652
  return continueExec;
42,318✔
1653
}
1654

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

1660
  if (pTrans->exec == TRN_EXEC_SERIAL) {
166,756✔
1661
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
20,556✔
1662
  } else {
1663
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf);
146,200✔
1664
  }
1665

1666
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
166,756!
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)) {
166,756✔
1677
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
48,582!
1678
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1679
      return false;
48,582✔
1680
    }
1681
  }
1682
  terrno = code;
118,174✔
1683

1684
  if (code == 0) {
118,174✔
1685
    pTrans->code = 0;
34,054✔
1686
    pTrans->stage = TRN_STAGE_COMMIT;
34,054✔
1687
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
34,054!
1688
    continueExec = true;
34,054✔
1689
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
84,120!
1690
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
84,096!
1691
    continueExec = false;
84,096✔
1692
  } else {
1693
    pTrans->failedTimes++;
24✔
1694
    pTrans->code = terrno;
24✔
1695
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
24✔
1696
      if (pTrans->lastAction != 0) {
6✔
1697
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
3✔
1698
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
3!
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;
6✔
1710
      pTrans->actionPos = 0;
6✔
1711
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
6!
1712
             pTrans->actionPos);
1713
      continueExec = true;
6✔
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;
118,174✔
1721
}
1722

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

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

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

1742
  return continueExec;
34,054✔
1743
}
1744

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

1749
  if (code == 0) {
76,321!
1750
    pTrans->code = 0;
76,321✔
1751
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
76,321✔
1752
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
76,321!
1753
    continueExec = true;
76,321✔
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;
76,321✔
1767
}
1768

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

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

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

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

1795
  return continueExec;
42✔
1796
}
1797

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

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

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

1814
  return continueExec;
6✔
1815
}
1816

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

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

1824
  if (code == 0) {
6!
1825
    pTrans->stage = TRN_STAGE_FINISH;
6✔
1826
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
6!
1827
    continueExec = true;
6✔
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;
6✔
1835
}
1836

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

1841
  SSdbRaw *pRaw = mndTransEncode(pTrans);
42,284✔
1842
  if (pRaw == NULL) {
42,284!
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));
42,284!
1847

1848
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
42,284✔
1849
  if (code != 0) {
42,284!
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,
42,284!
1854
        pTrans->failedTimes, pTrans->createdTime);
1855
  return continueExec;
42,284✔
1856
}
1857

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

1861
  while (continueExec) {
562,637✔
1862
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64 "", pTrans->id,
353,543!
1863
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1864
    pTrans->lastExecTime = taosGetTimestampMs();
353,543✔
1865
    switch (pTrans->stage) {
353,543!
1866
      case TRN_STAGE_PREPARE:
×
1867
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1868
        break;
×
1869
      case TRN_STAGE_REDO_ACTION:
166,756✔
1870
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf);
166,756✔
1871
        break;
166,756✔
1872
      case TRN_STAGE_COMMIT:
34,054✔
1873
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
34,054✔
1874
        break;
34,054✔
1875
      case TRN_STAGE_COMMIT_ACTION:
76,321✔
1876
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
76,321✔
1877
        break;
76,321✔
1878
      case TRN_STAGE_ROLLBACK:
6✔
1879
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
6✔
1880
        break;
6✔
1881
      case TRN_STAGE_UNDO_ACTION:
59✔
1882
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf);
59✔
1883
        break;
59✔
1884
      case TRN_STAGE_PRE_FINISH:
6✔
1885
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
6✔
1886
        break;
6✔
1887
      case TRN_STAGE_FINISH:
76,341✔
1888
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
76,341✔
1889
        break;
76,341✔
1890
      default:
×
1891
        continueExec = false;
×
1892
        break;
×
1893
    }
1894
  }
1895

1896
  mndTransSendRpcRsp(pMnode, pTrans);
209,094✔
1897
}
209,094✔
1898

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

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

1911
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
21,733✔
1912
  mTrace("start to process trans timer");
21,733✔
1913
  mndTransPullup(pReq->info.node);
21,733✔
1914
  return 0;
21,733✔
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; }
92✔
1975

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

1981
  void *pIter = NULL;
22,118✔
1982
  while (1) {
4,178✔
1983
    STrans *pTrans = NULL;
26,296✔
1984
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
26,296✔
1985
    if (pIter == NULL) break;
26,296✔
1986
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
8,356!
1987
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
1988
    }
1989
    sdbRelease(pSdb, pTrans);
4,178✔
1990
  }
1991

1992
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
22,118✔
1993

1994
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
26,296✔
1995
    int32_t *pTransId = taosArrayGet(pArray, i);
4,178✔
1996
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
4,178✔
1997
    if (pTrans != NULL) {
4,178!
1998
      mndTransExecute(pMnode, pTrans);
4,178✔
1999
    }
2000
    mndReleaseTrans(pMnode, pTrans);
4,178✔
2001
  }
2002
  taosArrayDestroy(pArray);
22,118✔
2003
}
2004

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

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

2018
    cols = 0;
165✔
2019

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

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

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

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

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

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

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

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

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

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

UNCOV
2075
_OVER:
×
2076
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
281!
2077
  pShow->numOfRows += numOfRows;
281✔
2078
  return numOfRows;
281✔
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