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

taosdata / TDengine / #3567

26 Dec 2024 02:00AM UTC coverage: 62.733% (+0.3%) from 62.422%
#3567

push

travis-ci

web-flow
Merge pull request #29290 from taosdata/enh/TD-33262-3.0

enh: test coverage of tfs

138620 of 284159 branches covered (48.78%)

Branch coverage included in aggregate %.

20 of 25 new or added lines in 3 files covered. (80.0%)

606 existing lines in 128 files now uncovered.

216140 of 281347 relevant lines covered (76.82%)

18804943.25 hits per line

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

69.56
/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; }
482,582✔
56

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

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

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

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

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

101
  for (int32_t i = 0; i < actionNum; ++i) {
1,752,404✔
102
    STransAction *pAction = taosArrayGet(pArray, i);
1,195,584✔
103
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,195,584✔
104
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
736,875✔
105
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
458,709!
106
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
458,709✔
107
    } else {
108
      // empty
109
    }
110
    rawDataLen += sizeof(int8_t);
1,195,584✔
111
  }
112

113
  return rawDataLen;
556,820✔
114
}
115

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

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

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

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

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

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

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

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

192
  if (sver > TRANS_VER1_NUMBER) {
139,205✔
193
    SDB_SET_INT32(pRaw, dataPos, prepareActionNum, _OVER)
44,444!
194
  }
195
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
139,205!
196
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
139,205!
197
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
139,205!
198

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

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

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

226
  terrno = 0;
139,205✔
227

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

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

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

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

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

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

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

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

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

324
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
263,711!
325

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

350
  if (sver > TRANS_VER1_NUMBER) {
263,711✔
351
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
90,738!
352
  }
353
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
263,711!
354
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
263,711!
355
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
263,711!
356

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

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

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

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

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

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

394
  terrno = 0;
263,711✔
395

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

410
static const char *mndTransStr(ETrnStage stage) {
1,817,143✔
411
  switch (stage) {
1,817,143!
412
    case TRN_STAGE_PREPARE:
114,581✔
413
      return "prepare";
114,581✔
414
    case TRN_STAGE_REDO_ACTION:
856,568✔
415
      return "redoAction";
856,568✔
416
    case TRN_STAGE_ROLLBACK:
10✔
417
      return "rollback";
10✔
418
    case TRN_STAGE_UNDO_ACTION:
74✔
419
      return "undoAction";
74✔
420
    case TRN_STAGE_COMMIT:
219,858✔
421
      return "commit";
219,858✔
422
    case TRN_STAGE_COMMIT_ACTION:
297,809✔
423
      return "commitAction";
297,809✔
424
    case TRN_STAGE_FINISH:
328,233✔
425
      return "finished";
328,233✔
426
    case TRN_STAGE_PRE_FINISH:
10✔
427
      return "pre-finish";
10✔
428
    default:
×
429
      return "invalid";
×
430
  }
431
}
432

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

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

474
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
49,094✔
475

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

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

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

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

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

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

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

551
  mndTransDropData(pTrans);
156,376✔
552
  return 0;
156,376✔
553
}
554

555
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
233,648✔
556
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
758,353✔
557
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
524,705✔
558
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
524,705✔
559
    pOldAction->rawWritten = pNewAction->rawWritten;
524,705✔
560
    pOldAction->msgSent = pNewAction->msgSent;
524,705✔
561
    pOldAction->msgReceived = pNewAction->msgReceived;
524,705✔
562
    pOldAction->errCode = pNewAction->errCode;
524,705✔
563
  }
564
}
233,648✔
565

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

570
  if (pOld->createdTime != pNew->createdTime) {
58,413✔
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);
58,412✔
580
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
58,412✔
581
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
58,412✔
582
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
58,412✔
583
  pOld->stage = pNew->stage;
58,412✔
584
  pOld->actionPos = pNew->actionPos;
58,412✔
585

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

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

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

601
  return 0;
58,412✔
602
}
603

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

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

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

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

650
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
41,102!
651
      pTrans->pRpcArray == NULL) {
41,102!
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) {
41,102✔
659
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
55,124!
660
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
661
      return NULL;
×
662
    }
663
    pTrans->originRpcType = pReq->msgType;
27,562✔
664
  }
665

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

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

672
static void mndTransDropActions(SArray *pArray) {
1,219,252✔
673
  int32_t size = taosArrayGetSize(pArray);
1,219,252✔
674
  for (int32_t i = 0; i < size; ++i) {
3,784,337✔
675
    STransAction *pAction = taosArrayGet(pArray, i);
2,565,085✔
676
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,565,085✔
677
      taosMemoryFreeClear(pAction->pRaw);
1,559,562!
678
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
1,005,523!
679
      taosMemoryFreeClear(pAction->pCont);
1,005,523!
680
    } else {
681
      // nothing
682
    }
683
  }
684

685
  taosArrayDestroy(pArray);
1,219,252✔
686
}
1,219,252✔
687

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

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

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

704
  return 0;
340,386✔
705
}
706

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

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

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

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

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

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

752
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
1,004✔
753
  pTrans->startFunc = startFunc;
1,004✔
754
  pTrans->stopFunc = stopFunc;
1,004✔
755
  pTrans->param = param;
1,004✔
756
  pTrans->paramLen = paramLen;
1,004✔
757
}
1,004✔
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) {
29,197✔
791
  if (dbname != NULL) {
29,197!
792
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
29,197✔
793
  }
794
  if (stbname != NULL) {
29,197✔
795
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
22,125✔
796
  }
797
}
29,197✔
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; }
2,865✔
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,659✔
812

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

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

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

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

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

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

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

867
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
119,378✔
868

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

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

876
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
21,081✔
877

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

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

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

926
  return conflict;
83,069✔
927
}
928

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

945
  TAOS_RETURN(code);
119,053✔
946
}
947

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

954
  while (1) {
4✔
955
    bool thisConflict = false;
304✔
956
    pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
304✔
957
    if (pIter == NULL) break;
304✔
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) {
300✔
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);
296✔
984
}
985

986
static bool mndTransActionsOfSameType(SArray *pActions) {
90,204✔
987
  int32_t size = taosArrayGetSize(pActions);
90,204✔
988
  ETrnAct lastActType = TRANS_ACTION_NULL;
90,204✔
989
  bool    same = true;
90,204✔
990
  for (int32_t i = 0; i < size; ++i) {
375,128✔
991
    STransAction *pAction = taosArrayGet(pActions, i);
284,924✔
992
    if (i > 0) {
284,924✔
993
      if (lastActType != pAction->actionType) {
213,395!
994
        same = false;
×
995
        break;
×
996
      }
997
    }
998
    lastActType = pAction->actionType;
284,924✔
999
  }
1000
  return same;
90,204✔
1001
}
1002

1003
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
40,742✔
1004
  int32_t code = 0;
40,742✔
1005
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
40,742✔
1006
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
37,949!
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) {
37,949✔
1013
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
11,513!
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);
40,742✔
1022
}
1023

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

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

1046
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
40,742!
1047

1048
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
40,742!
1049

1050
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
40,742!
1051

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

1062
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
40,694✔
1063
  if (pNew == NULL) {
40,694!
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;
40,694✔
1071
  pNew->rpcRsp = pTrans->rpcRsp;
40,694✔
1072
  pNew->rpcRspLen = pTrans->rpcRspLen;
40,694✔
1073
  pNew->mTraceId = pTrans->mTraceId;
40,694✔
1074
  pTrans->pRpcArray = NULL;
40,694✔
1075
  pTrans->rpcRsp = NULL;
40,694✔
1076
  pTrans->rpcRspLen = 0;
40,694✔
1077

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

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

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

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

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

1121
  if (pTrans->stage == TRN_STAGE_FINISH) {
244,262✔
1122
    sendRsp = true;
89,566✔
1123
  }
1124

1125
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
244,262✔
1126
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
61,639!
1127
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
10!
1128
      sendRsp = true;
10✔
1129
    }
1130
  } else {
1131
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
182,623✔
1132
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
120,365!
1133
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1134
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1135
      } else {
1136
        if (pTrans->failedTimes > 6) sendRsp = true;
120,365✔
1137
      }
1138
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
120,365✔
1139
    }
1140
  }
1141

1142
  if (!sendRsp) {
244,262✔
1143
    return;
154,684✔
1144
  } else {
1145
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
89,578!
1146
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1147
  }
1148

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

1157
  for (int32_t i = 0; i < size; ++i) {
54,318✔
1158
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
27,159✔
1159
    if (pInfo->handle != NULL) {
27,159✔
1160
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) {
25,686!
1161
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
1✔
1162
      }
1163
      if (code == TSDB_CODE_SYN_TIMEOUT) {
25,686!
1164
        code = TSDB_CODE_MND_TRANS_SYNC_TIMEOUT;
×
1165
      }
1166

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

1173
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
25,686✔
1174

1175
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
25,686✔
1176
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
3,628!
1177
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
3,628✔
1178
        if (pDb != NULL) {
3,628!
1179
          for (int32_t j = 0; j < 12; j++) {
4,411✔
1180
            bool ready = mndIsDbReady(pMnode, pDb);
4,408✔
1181
            if (!ready) {
4,408✔
1182
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
783!
1183
              taosMsleep(1000);
783✔
1184
            } else {
1185
              break;
3,625✔
1186
            }
1187
          }
1188
        }
1189
        mndReleaseDb(pMnode, pDb);
3,628✔
1190
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
22,058✔
1191
        void   *pCont = NULL;
6,832✔
1192
        int32_t contLen = 0;
6,832✔
1193
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
6,832✔
1194
          mndTransSetRpcRsp(pTrans, pCont, contLen);
6,830✔
1195
        }
1196
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
15,226✔
1197
        void   *pCont = NULL;
479✔
1198
        int32_t contLen = 0;
479✔
1199
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
479!
1200
          mndTransSetRpcRsp(pTrans, pCont, contLen);
479✔
1201
        }
1202
      }
1203

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

1213
      tmsgSendRsp(&rspMsg);
25,686✔
1214

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

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

1230
  STrans *pTrans = mndAcquireTrans(pMnode, transId);
91,727✔
1231
  if (pTrans == NULL) {
91,727!
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;
91,727✔
1239
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
91,727✔
1240
    pArray = pTrans->redoActions;
91,719✔
1241
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
8!
1242
    pArray = pTrans->undoActions;
8✔
1243
  } else {
1244
    mError("trans:%d, invalid trans stage:%d while recv action rsp", pTrans->id, pTrans->stage);
×
1245
    goto _OVER;
×
1246
  }
1247

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

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

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

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

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

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

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

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

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

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

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

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

1342
// execute in trans context
1343
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
618,959✔
1344
  if (pAction->msgSent) return 0;
618,959✔
1345
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
129,592✔
1346
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
37,780✔
1347
  }
1348

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

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

1362
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
91,812✔
1363

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

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

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

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

1390
  TAOS_RETURN(code);
91,812✔
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) {
1,044,142✔
1404
  if (pAction->actionType == TRANS_ACTION_RAW) {
1,044,142✔
1405
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
425,183✔
1406
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
618,959!
1407
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf);
618,959✔
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) {
224,868✔
1414
  int32_t numOfActions = taosArrayGetSize(pArray);
224,868✔
1415
  int32_t code = 0;
224,868✔
1416

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

1427
  return code;
224,868✔
1428
}
1429

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

1435
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf)) != 0) {
224,868✔
1436
    return code;
30,130✔
1437
  }
1438

1439
  int32_t       numOfExecuted = 0;
194,738✔
1440
  int32_t       errCode = 0;
194,738✔
1441
  STransAction *pErrAction = NULL;
194,738✔
1442
  for (int32_t action = 0; action < numOfActions; ++action) {
1,145,811✔
1443
    STransAction *pAction = taosArrayGet(pArray, action);
951,073✔
1444
    if (pAction->msgReceived || pAction->rawWritten) {
951,073✔
1445
      numOfExecuted++;
672,760✔
1446
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
672,760✔
1447
        errCode = pAction->errCode;
3✔
1448
        pErrAction = pAction;
3✔
1449
      }
1450
    } else {
1451
      pErrAction = pAction;
278,313✔
1452
    }
1453
  }
1454

1455
  mndSetTransLastAction(pTrans, pErrAction);
194,738✔
1456

1457
  if (numOfExecuted == numOfActions) {
194,738✔
1458
    if (errCode == 0) {
113,532✔
1459
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
113,530!
1460
      return 0;
113,530✔
1461
    } else {
1462
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
2!
1463
      mndTransResetActions(pMnode, pTrans, pArray);
2✔
1464
      terrno = errCode;
2✔
1465
      return errCode;
2✔
1466
    }
1467
  } else {
1468
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
81,206!
1469

1470
    for (int32_t action = 0; action < numOfActions; ++action) {
555,939✔
1471
      STransAction *pAction = taosArrayGet(pArray, action);
474,733✔
1472
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
474,733✔
1473
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1474
             pAction->acceptableCode, pAction->retryCode);
1475
      if (pAction->msgSent) {
474,733!
1476
        if (pAction->msgReceived) {
474,733✔
1477
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
196,420✔
1478
            mndTransResetAction(pMnode, pTrans, pAction);
1✔
1479
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
1!
1480
          }
1481
        }
1482
      }
1483
    }
1484
    return TSDB_CODE_ACTION_IN_PROGRESS;
81,206✔
1485
  }
1486
}
1487

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

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

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

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

1520
  if (pTrans->actionPos >= numOfActions) {
30,329✔
1521
    return code;
3,121✔
1522
  }
1523

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

1527
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
40,802✔
1528
    STransAction *pAction = taosArrayGet(pActions, action);
38,011✔
1529

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

1533
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
38,011✔
1534
    if (code == 0) {
38,011✔
1535
      if (pAction->msgSent) {
30,348✔
1536
        if (pAction->msgReceived) {
26,801✔
1537
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
11,993✔
1538
            code = pAction->errCode;
5,025✔
1539
            mndTransResetAction(pMnode, pTrans, pAction);
5,025✔
1540
          } else {
1541
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
6,968!
1542
          }
1543
        } else {
1544
          code = TSDB_CODE_ACTION_IN_PROGRESS;
14,808✔
1545
        }
1546
      } else if (pAction->rawWritten) {
3,547!
1547
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
3,547!
1548
          code = pAction->errCode;
×
1549
        } else {
1550
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,547!
1551
        }
1552
      } else {
1553
      }
1554
    }
1555

1556
    if (code == 0) {
38,011✔
1557
      pTrans->failedTimes = 0;
10,515✔
1558
    }
1559
    mndSetTransLastAction(pTrans, pAction);
38,011✔
1560

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

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

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

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

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

1638
  for (int32_t action = 0; action < numOfActions; ++action) {
40,891✔
1639
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
24,917✔
1640
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
24,917✔
1641
    if (code != 0) {
24,917!
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:
15,974✔
1650
  pTrans->stage = TRN_STAGE_REDO_ACTION;
48,919✔
1651
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
48,919!
1652
  return continueExec;
48,919✔
1653
}
1654

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

1660
  if (pTrans->exec == TRN_EXEC_SERIAL) {
195,378✔
1661
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
30,331✔
1662
  } else {
1663
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf);
165,047✔
1664
  }
1665

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

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

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

1720
  return continueExec;
136,742✔
1721
}
1722

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

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

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

1742
  return continueExec;
40,708✔
1743
}
1744

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

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

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

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

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

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

1795
  return continueExec;
10✔
1796
}
1797

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

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

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

1814
  return continueExec;
2✔
1815
}
1816

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

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

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

1834
  return continueExec;
2✔
1835
}
1836

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

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

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

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

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

1896
  mndTransSendRpcRsp(pMnode, pTrans);
244,262✔
1897
}
244,262✔
1898

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

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

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

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

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

1992
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
31,919✔
1993

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

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

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

2018
    cols = 0;
172✔
2019

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

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

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

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

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

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

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

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

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

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

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