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

taosdata / TDengine / #3846

11 Apr 2025 06:14AM UTC coverage: 62.398% (-0.2%) from 62.585%
#3846

push

travis-ci

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

merge: from main to 3.0 branch

153660 of 315259 branches covered (48.74%)

Branch coverage included in aggregate %.

63 of 80 new or added lines in 9 files covered. (78.75%)

1385 existing lines in 115 files now uncovered.

239696 of 315138 relevant lines covered (76.06%)

18295808.9 hits per line

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

62.59
/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 "mndTrans.h"
18
#include "mndDb.h"
19
#include "mndPrivilege.h"
20
#include "mndShow.h"
21
#include "mndStb.h"
22
#include "mndSubscribe.h"
23
#include "mndSync.h"
24
#include "mndUser.h"
25
#include "mndVgroup.h"
26
#include "osTime.h"
27

28
#define TRANS_VER1_NUMBER  1
29
#define TRANS_VER2_NUMBER  2
30
#define TRANS_ARRAY_SIZE   8
31
#define TRANS_RESERVE_SIZE 42
32

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

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

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

57
static inline bool mndTransIsInSyncContext(bool topHalf) { return !topHalf; }
517,648✔
58

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

66
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
542,313✔
67

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

73
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
74
static void    mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
75
static int32_t mndRetrieveTransDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
76
static int32_t tsMaxTransId = 0;
77

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

89
  mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransTimer);
1,831✔
90
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
1,831✔
91

92
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndRetrieveTrans);
1,831✔
93
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndCancelGetNextTrans);
1,831✔
94
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANSACTION_DETAIL, mndRetrieveTransDetail);
1,831✔
95
  return sdbSetTable(pMnode->pSdb, table);
1,831✔
96
}
97

98
void mndCleanupTrans(SMnode *pMnode) {}
1,830✔
99

100
static int32_t mndTransGetActionsSize(SArray *pArray) {
630,752✔
101
  int32_t actionNum = taosArrayGetSize(pArray);
630,752✔
102
  int32_t rawDataLen = 0;
630,752✔
103

104
  for (int32_t i = 0; i < actionNum; ++i) {
1,889,620✔
105
    STransAction *pAction = taosArrayGet(pArray, i);
1,258,868✔
106
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,258,868✔
107
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
775,903✔
108
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
482,965!
109
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
482,965✔
110
    } else {
111
      // empty
112
    }
113
    rawDataLen += sizeof(int8_t);
1,258,868✔
114
  }
115

116
  return rawDataLen;
630,752✔
117
}
118

119
static int32_t mndTransEncodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionsNum) {
630,752✔
120
  int32_t code = 0;
630,752✔
121
  int32_t lino = 0;
630,752✔
122
  int32_t dataPos = *offset;
630,752✔
123
  int8_t  unused = 0;
630,752✔
124
  int32_t ret = -1;
630,752✔
125

126
  for (int32_t i = 0; i < actionsNum; ++i) {
1,889,620✔
127
    STransAction *pAction = taosArrayGet(pActions, i);
1,258,868✔
128
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
1,258,868!
129
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
1,258,868!
130
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
1,258,868!
131
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
1,258,868!
132
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
1,258,868!
133
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
1,258,868!
134
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
1,258,868!
135
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,258,868✔
136
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
775,903✔
137
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->rawWritten*/, _OVER)
775,903!
138
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
775,903!
139
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
775,903!
140
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
482,965!
141
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
482,965!
142
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
482,965!
143
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgSent*/, _OVER)
482,965!
144
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgReceived*/, _OVER)
482,965!
145
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
482,965!
146
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
482,965!
147
    } else {
148
      // nothing
149
    }
150
  }
151
  ret = 0;
630,752✔
152

153
_OVER:
630,752✔
154
  *offset = dataPos;
630,752✔
155
  return ret;
630,752✔
156
}
157

158
SSdbRaw *mndTransEncode(STrans *pTrans) {
157,688✔
159
  int32_t code = 0;
157,688✔
160
  int32_t lino = 0;
157,688✔
161
  terrno = TSDB_CODE_INVALID_MSG;
157,688✔
162
  int8_t sver = TRANS_VER2_NUMBER;
157,688✔
163

164
  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
157,688✔
165
  rawDataLen += mndTransGetActionsSize(pTrans->prepareActions);
157,688✔
166
  rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
157,688✔
167
  rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
157,688✔
168
  rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
157,688✔
169

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

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

190
  int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions);
157,688✔
191
  int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
157,688✔
192
  int32_t undoActionNum = taosArrayGetSize(pTrans->undoActions);
157,688✔
193
  int32_t commitActionNum = taosArrayGetSize(pTrans->commitActions);
157,688✔
194

195
  if (sver > TRANS_VER1_NUMBER) {
157,688!
196
    SDB_SET_INT32(pRaw, dataPos, prepareActionNum, _OVER)
157,688!
197
  }
198
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
157,688!
199
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
157,688!
200
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
157,688!
201

202
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
157,688!
203
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
157,688!
204
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
157,688!
205
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
157,688!
206

207
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
157,688!
208
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
157,688!
209
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
157,688!
210
  if (pTrans->param != NULL) {
157,688!
211
    SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER)
×
212
  }
213

214
  SDB_SET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER)
157,688!
215

216
  int32_t arbGroupNum = taosHashGetSize(pTrans->arbGroupIds);
157,688✔
217
  SDB_SET_INT32(pRaw, dataPos, arbGroupNum, _OVER)
157,688!
218
  void *pIter = NULL;
157,688✔
219
  pIter = taosHashIterate(pTrans->arbGroupIds, NULL);
157,688✔
220
  while (pIter) {
157,724✔
221
    int32_t arbGroupId = *(int32_t *)pIter;
36✔
222
    SDB_SET_INT32(pRaw, dataPos, arbGroupId, _OVER)
36!
223
    pIter = taosHashIterate(pTrans->arbGroupIds, pIter);
36✔
224
  }
225

226
  if (sver > TRANS_VER1_NUMBER) {
157,688!
227
    SDB_SET_INT8(pRaw, dataPos, pTrans->ableToBeKilled, _OVER)
157,688!
228
    SDB_SET_INT32(pRaw, dataPos, pTrans->killMode, _OVER)
157,688!
229
  }
230

231
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
157,688!
232
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
157,688!
233

234
  terrno = 0;
157,688✔
235

236
_OVER:
157,688✔
237
  if (terrno != 0) {
157,688!
238
    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
×
239
           dataPos, terrstr());
240
    sdbFreeRaw(pRaw);
×
241
    return NULL;
×
242
  }
243

244
  mTrace("trans:%d, encode to raw:%p, row:%p len:%d", pTrans->id, pRaw, pTrans, dataPos);
157,688✔
245
  return pRaw;
157,688✔
246
}
247

248
static int32_t mndTransDecodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionNum) {
1,205,492✔
249
  int32_t      code = 0;
1,205,492✔
250
  int32_t      lino = 0;
1,205,492✔
251
  STransAction action = {0};
1,205,492✔
252
  int32_t      dataPos = *offset;
1,205,492✔
253
  int8_t       unused = 0;
1,205,492✔
254
  int8_t       stage = 0;
1,205,492✔
255
  int8_t       actionType = 0;
1,205,492✔
256
  int32_t      dataLen = 0;
1,205,492✔
257
  int32_t      ret = -1;
1,205,492✔
258

259
  for (int32_t i = 0; i < actionNum; ++i) {
3,554,456✔
260
    memset(&action, 0, sizeof(action));
2,348,964✔
261
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
2,348,964!
262
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
2,348,964!
263
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
2,348,964!
264
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
2,348,964!
265
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
2,348,964!
266
    action.actionType = actionType;
2,348,964✔
267
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
2,348,964!
268
    action.stage = stage;
2,348,964✔
269
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
2,348,964!
270
    if (action.actionType == TRANS_ACTION_RAW) {
2,348,964✔
271
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.rawWritten*/, _OVER)
1,411,330!
272
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
1,411,330!
273
      action.pRaw = taosMemoryMalloc(dataLen);
1,411,330!
274
      if (action.pRaw == NULL) goto _OVER;
1,411,330!
275
      mTrace("raw:%p, is created", action.pRaw);
1,411,330✔
276
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
1,411,330!
277
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
1,411,330!
278
      action.pRaw = NULL;
1,411,330✔
279
    } else if (action.actionType == TRANS_ACTION_MSG) {
937,634!
280
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
937,634!
281
      tmsgUpdateDnodeEpSet(&action.epSet);
937,634✔
282
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
937,634!
283
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgSent*/, _OVER)
937,634!
284
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgReceived*/, _OVER)
937,634!
285
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
937,634!
286
      action.pCont = taosMemoryMalloc(action.contLen);
937,634!
287
      if (action.pCont == NULL) goto _OVER;
937,634!
288
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
937,634!
289
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
937,634!
290
      action.pCont = NULL;
937,634✔
291
    } else {
292
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
×
293
    }
294
  }
295
  ret = 0;
1,205,492✔
296

297
_OVER:
1,205,492✔
298
  *offset = dataPos;
1,205,492✔
299
  taosMemoryFreeClear(action.pCont);
1,205,492!
300
  return ret;
1,205,492✔
301
}
302

303
SSdbRow *mndTransDecode(SSdbRaw *pRaw) {
301,373✔
304
  terrno = TSDB_CODE_INVALID_MSG;
301,373✔
305
  int32_t  code = 0;
301,373✔
306
  int32_t  lino = 0;
301,373✔
307
  SSdbRow *pRow = NULL;
301,373✔
308
  STrans  *pTrans = NULL;
301,373✔
309
  char    *pData = NULL;
301,373✔
310
  int32_t  dataLen = 0;
301,373✔
311
  int8_t   sver = 0;
301,373✔
312
  int32_t  prepareActionNum = 0;
301,373✔
313
  int32_t  redoActionNum = 0;
301,373✔
314
  int32_t  undoActionNum = 0;
301,373✔
315
  int32_t  commitActionNum = 0;
301,373✔
316
  int32_t  dataPos = 0;
301,373✔
317
  int32_t  arbgroupIdNum = 0;
301,373✔
318

319
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
301,373!
320

321
  if (sver > TRANS_VER2_NUMBER) {
301,373!
322
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
323
    goto _OVER;
×
324
  }
325

326
  pRow = sdbAllocRow(sizeof(STrans));
301,373✔
327
  if (pRow == NULL) goto _OVER;
301,373!
328

329
  pTrans = sdbGetRowObj(pRow);
301,373✔
330
  if (pTrans == NULL) goto _OVER;
301,373!
331

332
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
301,373!
333

334
  int8_t stage = 0;
301,373✔
335
  int8_t policy = 0;
301,373✔
336
  int8_t conflict = 0;
301,373✔
337
  int8_t exec = 0;
301,373✔
338
  int8_t oper = 0;
301,373✔
339
  int8_t reserved = 0;
301,373✔
340
  int8_t actionType = 0;
301,373✔
341
  SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
301,373!
342
  SDB_GET_INT8(pRaw, dataPos, &policy, _OVER)
301,373!
343
  SDB_GET_INT8(pRaw, dataPos, &conflict, _OVER)
301,373!
344
  SDB_GET_INT8(pRaw, dataPos, &exec, _OVER)
301,373!
345
  SDB_GET_INT8(pRaw, dataPos, &oper, _OVER)
301,373!
346
  SDB_GET_INT8(pRaw, dataPos, &reserved, _OVER)
301,373!
347
  pTrans->stage = stage;
301,373✔
348
  pTrans->policy = policy;
301,373✔
349
  pTrans->conflict = conflict;
301,373✔
350
  pTrans->exec = exec;
301,373✔
351
  pTrans->oper = oper;
301,373✔
352
  SDB_GET_INT16(pRaw, dataPos, &pTrans->originRpcType, _OVER)
301,373!
353
  SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
301,373!
354
  SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
301,373!
355
  SDB_GET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
301,373!
356
  SDB_GET_INT32(pRaw, dataPos, &pTrans->actionPos, _OVER)
301,373!
357

358
  if (sver > TRANS_VER1_NUMBER) {
301,373!
359
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
301,373!
360
  }
361
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
301,373!
362
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
301,373!
363
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
301,373!
364

365
  pTrans->prepareActions = taosArrayInit(prepareActionNum, sizeof(STransAction));
301,373✔
366
  pTrans->redoActions = taosArrayInit(redoActionNum, sizeof(STransAction));
301,373✔
367
  pTrans->undoActions = taosArrayInit(undoActionNum, sizeof(STransAction));
301,373✔
368
  pTrans->commitActions = taosArrayInit(commitActionNum, sizeof(STransAction));
301,373✔
369

370
  if (pTrans->prepareActions == NULL) goto _OVER;
301,373!
371
  if (pTrans->redoActions == NULL) goto _OVER;
301,373!
372
  if (pTrans->undoActions == NULL) goto _OVER;
301,373!
373
  if (pTrans->commitActions == NULL) goto _OVER;
301,373!
374

375
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
301,373!
376
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
301,373!
377
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
301,373!
378
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
301,373!
379

380
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
301,373!
381
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
301,373!
382
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
301,373!
383
  if (pTrans->paramLen != 0) {
301,373!
384
    pTrans->param = taosMemoryMalloc(pTrans->paramLen);
×
385
    if (pTrans->param == NULL) goto _OVER;
×
386
    SDB_GET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER);
×
387
  }
388

389
  SDB_GET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER);
301,373!
390

391
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
301,373✔
392

393
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
301,373!
394
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
301,433✔
395
    int32_t arbGroupId = 0;
60✔
396
    SDB_GET_INT32(pRaw, dataPos, &arbGroupId, _OVER)
60!
397
    if ((terrno = taosHashPut(pTrans->arbGroupIds, &arbGroupId, sizeof(int32_t), NULL, 0)) != 0) goto _OVER;
60!
398
  }
399

400
  int8_t ableKill = 0;
301,373✔
401
  int8_t killMode = 0;
301,373✔
402
  SDB_GET_INT8(pRaw, dataPos, &ableKill, _OVER)
301,373!
403
  SDB_GET_INT8(pRaw, dataPos, &killMode, _OVER)
301,373!
404
  pTrans->ableToBeKilled = ableKill;
301,373✔
405
  pTrans->killMode = killMode;
301,373✔
406

407
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
301,373!
408

409
  terrno = 0;
301,373✔
410

411
_OVER:
301,373✔
412
  if (terrno != 0 && pTrans != NULL) {
301,373!
413
    mError("trans:%d, failed to parse from raw:%p since %s", pTrans->id, pRaw, terrstr());
×
414
    mndTransDropData(pTrans);
×
415
    taosMemoryFreeClear(pRow);
×
416
    return NULL;
×
417
  }
418

419
  if (pTrans != NULL) {
301,373!
420
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
301,373✔
421
  }
422
  return pRow;
301,373✔
423
}
424

425
static const char *mndTransStr(ETrnStage stage) {
2,362,212✔
426
  switch (stage) {
2,362,212!
427
    case TRN_STAGE_PREPARE:
152,626✔
428
      return "prepare";
152,626✔
429
    case TRN_STAGE_REDO_ACTION:
1,015,813✔
430
      return "redoAction";
1,015,813✔
431
    case TRN_STAGE_ROLLBACK:
10✔
432
      return "rollback";
10✔
433
    case TRN_STAGE_UNDO_ACTION:
46,292✔
434
      return "undoAction";
46,292✔
435
    case TRN_STAGE_COMMIT:
252,227✔
436
      return "commit";
252,227✔
437
    case TRN_STAGE_COMMIT_ACTION:
522,138✔
438
      return "commitAction";
522,138✔
439
    case TRN_STAGE_FINISH:
373,096✔
440
      return "finished";
373,096✔
441
    case TRN_STAGE_PRE_FINISH:
10✔
442
      return "pre-finish";
10✔
443
    default:
×
444
      return "invalid";
×
445
  }
446
}
447

448
static const char *mndTransTypeStr(ETrnAct actionType) {
×
449
  switch (actionType) {
×
450
    case TRANS_ACTION_MSG:
×
451
      return "msg";
×
452
    case TRANS_ACTION_RAW:
×
453
      return "sdb";
×
454
    default:
×
455
      return "invalid";
×
456
  }
457
}
458

459
static void mndSetTransLastAction(STrans *pTrans, STransAction *pAction) {
699,068✔
460
  if (pAction != NULL) {
699,068✔
461
    if (pAction->errCode != TSDB_CODE_ACTION_IN_PROGRESS) {
571,009✔
462
      pTrans->lastAction = pAction->id;
370,159✔
463
      pTrans->lastMsgType = pAction->msgType;
370,159✔
464
      pTrans->lastEpset = pAction->epSet;
370,159✔
465
      pTrans->lastErrorNo = pAction->errCode;
370,159✔
466
    }
467
  } else {
468
    pTrans->lastAction = 0;
128,059✔
469
    pTrans->lastMsgType = 0;
128,059✔
470
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
128,059✔
471
    pTrans->lastErrorNo = 0;
128,059✔
472
  }
473
}
699,068✔
474

475
static void mndTransTestStartFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
476
  mInfo("test trans start, param:%s, len:%d", (char *)param, paramLen);
×
477
}
×
478

479
static void mndTransTestStopFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
480
  mInfo("test trans stop, param:%s, len:%d", (char *)param, paramLen);
×
481
}
×
482

483
static TransCbFp mndTransGetCbFp(ETrnFunc ftype) {
2,036✔
484
  switch (ftype) {
2,036!
485
    case TRANS_START_FUNC_TEST:
×
486
      return mndTransTestStartFunc;
×
487
    case TRANS_STOP_FUNC_TEST:
×
488
      return mndTransTestStopFunc;
×
489
    case TRANS_START_FUNC_MQ_REB:
1,018✔
490
      return mndRebCntInc;
1,018✔
491
    case TRANS_STOP_FUNC_MQ_REB:
1,018✔
492
      return mndRebCntDec;
1,018✔
493
    default:
×
494
      return NULL;
×
495
  }
496
}
497

498
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
56,851✔
499
  mInfo("trans:%d, perform insert action, row:%p stage:%s, callfunc:1, startFunc:%d", pTrans->id, pTrans,
56,851!
500
        mndTransStr(pTrans->stage), pTrans->startFunc);
501

502
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
56,851✔
503

504
  if (pTrans->startFunc > 0) {
56,851✔
505
    TransCbFp fp = mndTransGetCbFp(pTrans->startFunc);
1,018✔
506
    if (fp) {
1,018!
507
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
1,018✔
508
    }
509
    // pTrans->startFunc = 0;
510
  }
511

512
  if (pTrans->stage == TRN_STAGE_COMMIT) {
56,851✔
513
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
3✔
514
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pTrans->id);
3!
515
  }
516

517
  if (pTrans->stage == TRN_STAGE_ROLLBACK) {
56,851!
518
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
×
519
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pTrans->id);
×
520
  }
521

522
  if (pTrans->stage == TRN_STAGE_PRE_FINISH) {
56,851!
523
    pTrans->stage = TRN_STAGE_FINISH;
×
524
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pTrans->id);
×
525
  }
526

527
  return 0;
56,851✔
528
}
529

530
void mndTransDropData(STrans *pTrans) {
348,102✔
531
  if (pTrans->prepareActions != NULL) {
348,102!
532
    mndTransDropActions(pTrans->prepareActions);
348,102✔
533
    pTrans->prepareActions = NULL;
348,102✔
534
  }
535
  if (pTrans->redoActions != NULL) {
348,102!
536
    mndTransDropActions(pTrans->redoActions);
348,102✔
537
    pTrans->redoActions = NULL;
348,102✔
538
  }
539
  if (pTrans->undoActions != NULL) {
348,102!
540
    mndTransDropActions(pTrans->undoActions);
348,102✔
541
    pTrans->undoActions = NULL;
348,102✔
542
  }
543
  if (pTrans->commitActions != NULL) {
348,102!
544
    mndTransDropActions(pTrans->commitActions);
348,102✔
545
    pTrans->commitActions = NULL;
348,102✔
546
  }
547
  if (pTrans->arbGroupIds != NULL) {
348,102!
548
    taosHashCleanup(pTrans->arbGroupIds);
348,102✔
549
  }
550
  if (pTrans->pRpcArray != NULL) {
348,102✔
551
    taosArrayDestroy(pTrans->pRpcArray);
46,729✔
552
    pTrans->pRpcArray = NULL;
46,729✔
553
  }
554
  if (pTrans->rpcRsp != NULL) {
348,102✔
555
    taosMemoryFree(pTrans->rpcRsp);
13,174!
556
    pTrans->rpcRsp = NULL;
13,174✔
557
    pTrans->rpcRspLen = 0;
13,174✔
558
  }
559
  if (pTrans->param != NULL) {
348,102!
560
    taosMemoryFree(pTrans->param);
×
561
    pTrans->param = NULL;
×
562
    pTrans->paramLen = 0;
×
563
  }
564
  (void)taosThreadMutexDestroy(&pTrans->mutex);
348,102✔
565
}
348,102✔
566

567
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc) {
179,091✔
568
  mInfo("trans:%d, perform delete action, row:%p stage:%s callfunc:%d, stopFunc:%d", pTrans->id, pTrans,
179,091!
569
        mndTransStr(pTrans->stage), callFunc, pTrans->stopFunc);
570

571
  if (pTrans->stopFunc > 0 && callFunc) {
179,091✔
572
    TransCbFp fp = mndTransGetCbFp(pTrans->stopFunc);
1,018✔
573
    if (fp) {
1,018!
574
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
1,018✔
575
    }
576
    // pTrans->stopFunc = 0;
577
  }
578

579
  mndTransDropData(pTrans);
179,091✔
580
  return 0;
179,091✔
581
}
582

583
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
262,516✔
584
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
807,665✔
585
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
545,149✔
586
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
545,149✔
587
    pOldAction->rawWritten = pNewAction->rawWritten;
545,149✔
588
    pOldAction->msgSent = pNewAction->msgSent;
545,149✔
589
    pOldAction->msgReceived = pNewAction->msgReceived;
545,149✔
590
    pOldAction->errCode = pNewAction->errCode;
545,149✔
591
  }
592
}
262,516✔
593

594
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
65,629✔
595
  mInfo("trans:%d, perform update action, old row:%p stage:%s create:%" PRId64 ", new row:%p stage:%s create:%" PRId64,
65,629!
596
        pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage), pNew->createdTime);
597

598
  if (pOld->createdTime != pNew->createdTime) {
65,629!
599
    mError("trans:%d, failed to perform update action since createTime not match, old row:%p stage:%s create:%" PRId64
×
600
           ", new row:%p stage:%s create:%" PRId64,
601
           pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage),
602
           pNew->createdTime);
603
    // only occured while sync timeout
604
    TAOS_RETURN(TSDB_CODE_MND_TRANS_SYNC_TIMEOUT);
×
605
  }
606

607
  mndTransUpdateActions(pOld->prepareActions, pNew->prepareActions);
65,629✔
608
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
65,629✔
609
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
65,629✔
610
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
65,629✔
611
  pOld->stage = pNew->stage;
65,629✔
612
  pOld->actionPos = pNew->actionPos;
65,629✔
613

614
  if (pOld->stage == TRN_STAGE_COMMIT) {
65,629✔
615
    pOld->stage = TRN_STAGE_COMMIT_ACTION;
56,607✔
616
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pNew->id);
56,607!
617
  }
618

619
  if (pOld->stage == TRN_STAGE_ROLLBACK) {
65,629✔
620
    pOld->stage = TRN_STAGE_UNDO_ACTION;
2✔
621
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pNew->id);
2!
622
  }
623

624
  if (pOld->stage == TRN_STAGE_PRE_FINISH) {
65,629✔
625
    pOld->stage = TRN_STAGE_FINISH;
2✔
626
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pNew->id);
2!
627
  }
628

629
  return 0;
65,629✔
630
}
631

632
STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
277,203✔
633
  STrans *pTrans = sdbAcquire(pMnode->pSdb, SDB_TRANS, &transId);
277,203✔
634
  if (pTrans == NULL) {
277,203✔
635
    terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
4,098✔
636
  }
637
  return pTrans;
277,203✔
638
}
639

640
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
273,106✔
641
  SSdb *pSdb = pMnode->pSdb;
273,106✔
642
  if (pTrans != NULL) mInfo("vgId:1, trans:%d, release transaction", pTrans->id);
273,106!
643
  sdbRelease(pSdb, pTrans);
273,106✔
644
}
273,106✔
645

646
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
46,729✔
647
                       const char *opername) {
648
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
46,729!
649
  if (pTrans == NULL) {
46,729!
650
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
651
    mError("failed to create transaction since %s", terrstr());
×
652
    return NULL;
×
653
  }
654

655
  if (opername != NULL) {
46,729!
656
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
46,729✔
657
  }
658

659
  int32_t sdbMaxId = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
46,729✔
660
  sdbReadLock(pMnode->pSdb, SDB_TRANS);
46,729✔
661
  pTrans->id = TMAX(sdbMaxId, tsMaxTransId + 1);
46,729✔
662
  sdbUnLock(pMnode->pSdb, SDB_TRANS);
46,729✔
663
  pTrans->stage = TRN_STAGE_PREPARE;
46,729✔
664
  pTrans->policy = policy;
46,729✔
665
  pTrans->conflict = conflict;
46,729✔
666
  pTrans->exec = TRN_EXEC_PARALLEL;
46,729✔
667
  pTrans->ableToBeKilled = false;
46,729✔
668
  pTrans->createdTime = taosGetTimestampMs();
46,729✔
669
  pTrans->prepareActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
46,729✔
670
  pTrans->redoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
46,729✔
671
  pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
46,729✔
672
  pTrans->commitActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
46,729✔
673
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
46,729✔
674
  pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
46,729✔
675
  pTrans->mTraceId = pReq ? TRACE_GET_ROOTID(&pReq->info.traceId) : tGenIdPI64();
46,729✔
676
  taosInitRWLatch(&pTrans->lockRpcArray);
46,729✔
677
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
46,729✔
678

679
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
46,729!
680
      pTrans->pRpcArray == NULL) {
46,729!
681
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
682
    mError("failed to create transaction since %s", terrstr());
×
683
    mndTransDrop(pTrans);
×
684
    return NULL;
×
685
  }
686

687
  if (pReq != NULL) {
46,729✔
688
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
57,892!
689
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
690
      return NULL;
×
691
    }
692
    pTrans->originRpcType = pReq->msgType;
28,946✔
693
  }
694

695
  mInfo("trans:%d, create transaction:%s, origin:%s", pTrans->id, pTrans->opername, opername);
46,729!
696

697
  mTrace("trans:%d, local object is created, data:%p", pTrans->id, pTrans);
46,729✔
698
  return pTrans;
46,729✔
699
}
700

701
static void mndTransDropActions(SArray *pArray) {
1,392,408✔
702
  int32_t size = taosArrayGetSize(pArray);
1,392,408✔
703
  for (int32_t i = 0; i < size; ++i) {
4,102,999✔
704
    STransAction *pAction = taosArrayGet(pArray, i);
2,710,591✔
705
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,710,591✔
706
      taosMemoryFreeClear(pAction->pRaw);
1,647,112!
707
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
1,063,479!
708
      taosMemoryFreeClear(pAction->pCont);
1,063,479!
709
    } else {
710
      // nothing
711
    }
712
  }
713

714
  taosArrayDestroy(pArray);
1,392,408✔
715
}
1,392,408✔
716

717
void mndTransDrop(STrans *pTrans) {
47,826✔
718
  if (pTrans != NULL) {
47,826✔
719
    mndTransDropData(pTrans);
46,729✔
720
    mTrace("trans:%d, local object is freed, data:%p", pTrans->id, pTrans);
46,729✔
721
    taosMemoryFreeClear(pTrans);
46,729!
722
  }
723
}
47,826✔
724

725
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
361,627✔
726
  pAction->id = taosArrayGetSize(pArray);
361,627✔
727

728
  void *ptr = taosArrayPush(pArray, pAction);
361,627✔
729
  if (ptr == NULL) {
361,627!
730
    TAOS_RETURN(terrno);
×
731
  }
732

733
  return 0;
361,627✔
734
}
735

736
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
1,658✔
737
  STransAction action = {
1,658✔
738
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
1,658✔
739
  return mndTransAppendAction(pTrans->redoActions, &action);
1,658✔
740
}
741

742
int32_t mndTransAppendNullLog(STrans *pTrans) {
×
743
  STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_NULL};
×
744
  return mndTransAppendAction(pTrans->redoActions, &action);
×
745
}
746

747
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
13,637✔
748
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
13,637✔
749
  return mndTransAppendAction(pTrans->undoActions, &action);
13,637✔
750
}
751

752
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
198,826✔
753
  STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
198,826✔
754
  return mndTransAppendAction(pTrans->commitActions, &action);
198,826✔
755
}
756

757
int32_t mndTransAppendPrepareLog(STrans *pTrans, SSdbRaw *pRaw) {
21,661✔
758
  STransAction action = {
21,661✔
759
      .pRaw = pRaw, .stage = TRN_STAGE_PREPARE, .actionType = TRANS_ACTION_RAW, .mTraceId = pTrans->mTraceId};
21,661✔
760
  return mndTransAppendAction(pTrans->prepareActions, &action);
21,661✔
761
}
762

763
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction) {
93,276✔
764
  pAction->stage = TRN_STAGE_REDO_ACTION;
93,276✔
765
  pAction->actionType = TRANS_ACTION_MSG;
93,276✔
766
  pAction->mTraceId = pTrans->mTraceId;
93,276✔
767
  return mndTransAppendAction(pTrans->redoActions, pAction);
93,276✔
768
}
769

770
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction) {
32,569✔
771
  pAction->stage = TRN_STAGE_UNDO_ACTION;
32,569✔
772
  pAction->actionType = TRANS_ACTION_MSG;
32,569✔
773
  return mndTransAppendAction(pTrans->undoActions, pAction);
32,569✔
774
}
775

776
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen) {
13,174✔
777
  pTrans->rpcRsp = pCont;
13,174✔
778
  pTrans->rpcRspLen = contLen;
13,174✔
779
}
13,174✔
780

781
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
1,006✔
782
  pTrans->startFunc = startFunc;
1,006✔
783
  pTrans->stopFunc = stopFunc;
1,006✔
784
  pTrans->param = param;
1,006✔
785
  pTrans->paramLen = paramLen;
1,006✔
786
}
1,006✔
787

788
int32_t mndSetRpcInfoForDbTrans(SMnode *pMnode, SRpcMsg *pMsg, EOperType oper, const char *dbname) {
×
789
  STrans *pTrans = NULL;
×
790
  void   *pIter = NULL;
×
791
  int32_t code = -1;
×
792

793
  while (1) {
794
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
×
795
    if (pIter == NULL) break;
×
796

797
    if (pTrans->oper == oper) {
×
798
      if (taosStrcasecmp(dbname, pTrans->dbname) == 0) {
×
799
        mInfo("trans:%d, db:%s oper:%d matched with input", pTrans->id, dbname, oper);
×
800
        taosWLockLatch(&pTrans->lockRpcArray);
×
801
        if (pTrans->pRpcArray == NULL) {
×
802
          pTrans->pRpcArray = taosArrayInit(4, sizeof(SRpcHandleInfo));
×
803
        }
804
        if (pTrans->pRpcArray != NULL && taosArrayPush(pTrans->pRpcArray, &pMsg->info) != NULL) {
×
805
          code = 0;
×
806
        }
807
        taosWUnLockLatch(&pTrans->lockRpcArray);
×
808

809
        sdbRelease(pMnode->pSdb, pTrans);
×
810
        break;
×
811
      }
812
    }
813

814
    sdbRelease(pMnode->pSdb, pTrans);
×
815
  }
816
  return code;
×
817
}
818

819
void mndTransSetDbName(STrans *pTrans, const char *dbname, const char *stbname) {
30,357✔
820
  if (dbname != NULL) {
30,357!
821
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
30,357✔
822
  }
823
  if (stbname != NULL) {
30,357✔
824
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
22,893✔
825
  }
826
}
30,357✔
827

828
void mndTransAddArbGroupId(STrans *pTrans, int32_t groupId) {
12✔
829
  if (taosHashPut(pTrans->arbGroupIds, &groupId, sizeof(int32_t), NULL, 0) != 0) {
12!
830
    mError("trans:%d, failed to put groupid into hash, groupId:%d", pTrans->id, groupId);
×
831
  }
832
}
12✔
833

834
void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
2,788✔
835

836
void mndTransSetBeKilled(STrans *pTrans, bool ableToBeKilled) { pTrans->ableToBeKilled = ableToBeKilled; }
×
837

838
void mndTransSetKillMode(STrans *pTrans, ETrnKillMode killMode) {
8✔
839
  pTrans->ableToBeKilled = true; 
8✔
840
  pTrans->killMode = killMode; 
8✔
841
}
8✔
842

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

845
void mndTransSetChangeless(STrans *pTrans) { pTrans->changeless = true; }
34✔
846

847
void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }
3,966✔
848

849
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
100,716✔
850
  int32_t  code = 0;
100,716✔
851
  SSdbRaw *pRaw = mndTransEncode(pTrans);
100,716✔
852
  if (pRaw == NULL) {
100,716!
853
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
854
    if (terrno != 0) code = terrno;
×
855
    mError("trans:%d, failed to encode while sync trans since %s", pTrans->id, tstrerror(code));
×
856
    TAOS_RETURN(code);
×
857
  }
858
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
100,716!
859

860
  mInfo("trans:%d, sync to other mnodes, stage:%s createTime:%" PRId64, pTrans->id, mndTransStr(pTrans->stage),
100,716!
861
        pTrans->createdTime);
862
  code = mndSyncPropose(pMnode, pRaw, pTrans->id);
100,716✔
863
  if (code != 0) {
100,716✔
864
    mError("trans:%d, failed to sync, errno:%s code:0x%x createTime:%" PRId64 " saved trans:%d", pTrans->id,
25!
865
           tstrerror(code), code, pTrans->createdTime, pMnode->syncMgmt.transId);
866
    sdbFreeRaw(pRaw);
25✔
867
    TAOS_RETURN(code);
25✔
868
  }
869

870
  sdbFreeRaw(pRaw);
100,691✔
871
  mInfo("trans:%d, sync finished, createTime:%" PRId64, pTrans->id, pTrans->createdTime);
100,691!
872
  TAOS_RETURN(code);
100,691✔
873
}
874

875
static bool mndCheckDbConflict(const char *conflict, STrans *pTrans) {
4,242✔
876
  if (conflict[0] == 0) return false;
4,242!
877
  if (taosStrcasecmp(conflict, pTrans->dbname) == 0) return true;
4,242✔
878
  return false;
3,949✔
879
}
880

881
static bool mndCheckStbConflict(const char *conflict, STrans *pTrans) {
21,352✔
882
  if (conflict[0] == 0) return false;
21,352✔
883
  if (taosStrcasecmp(conflict, pTrans->stbname) == 0) return true;
20,007✔
884
  return false;
19,974✔
885
}
886

887
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
25,597✔
888
  if (conflict) {
25,597✔
889
    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,
326!
890
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
891
    *globalConflict = true;
326✔
892
  } else {
893
    mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with trans:%d db:%s stb:%s type:%d", pNew->id, pNew->dbname,
25,271!
894
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
895
  }
896
}
25,597✔
897

898
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
133,881✔
899
  STrans *pTrans = NULL;
133,881✔
900
  void   *pIter = NULL;
133,881✔
901
  bool    conflict = false;
133,881✔
902

903
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
133,881✔
904

905
  int32_t size = sdbGetSize(pMnode->pSdb, SDB_TRANS);
89,698✔
906
  mInfo("trans:%d, trans hash size %d", pNew->id, size);
89,698!
907

908
  while (1) {
909
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
112,831✔
910
    if (pIter == NULL) break;
112,831✔
911

912
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
23,133!
913

914
    if (pNew->conflict == TRN_CONFLICT_DB) {
23,133✔
915
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
3,560✔
916
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
3,560✔
917
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
2,467✔
918
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
2,467✔
919
      }
920
    }
921

922
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
23,133✔
923
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
19,570✔
924
      if (pTrans->conflict == TRN_CONFLICT_DB) {
19,570✔
925
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
1,775✔
926
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
1,775✔
927
      }
928
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
19,570✔
929
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
17,110✔
930
      }
931
    }
932

933
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
23,133!
934
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
×
935
      if (pTrans->conflict == TRN_CONFLICT_ARBGROUP) {
×
936
        void *pGidIter = taosHashIterate(pNew->arbGroupIds, NULL);
×
937
        while (pGidIter != NULL) {
×
938
          int32_t groupId = *(int32_t *)pGidIter;
×
939
          if (taosHashGet(pTrans->arbGroupIds, &groupId, sizeof(int32_t)) != NULL) {
×
940
            taosHashCancelIterate(pNew->arbGroupIds, pGidIter);
×
941
            mndTransLogConflict(pNew, pTrans, true, &conflict);
×
942
            break;
×
943
          } else {
944
            mndTransLogConflict(pNew, pTrans, false, &conflict);
×
945
          }
946
          pGidIter = taosHashIterate(pNew->arbGroupIds, pGidIter);
×
947
        }
948
      }
949
    }
950

951
    if (pNew->conflict == TRN_CONFLICT_TSMA) {
23,133✔
952
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL || pTrans->conflict == TRN_CONFLICT_TSMA) {
3!
953
        mndTransLogConflict(pNew, pTrans, true, &conflict);
×
954
      } else {
955
        mndTransLogConflict(pNew, pTrans, false, &conflict);
3✔
956
      }
957
    }
958

959
    sdbRelease(pMnode->pSdb, pTrans);
23,133✔
960
  }
961

962
  return conflict;
89,698✔
963
}
964

965
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
133,881✔
966
  int32_t code = 0;
133,881✔
967
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
133,881✔
968
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
80,055!
969
      code = TSDB_CODE_MND_TRANS_CONFLICT;
×
970
      mError("trans:%d, failed to check tran conflict since db not set", pTrans->id);
×
971
      TAOS_RETURN(code);
×
972
    }
973
  }
974

975
  if (mndCheckTransConflict(pMnode, pTrans)) {
133,881✔
976
    code = TSDB_CODE_MND_TRANS_CONFLICT;
359✔
977
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
359!
978
    TAOS_RETURN(code);
359✔
979
  }
980

981
  TAOS_RETURN(code);
133,522✔
982
}
983

984
int32_t mndTransCheckConflictWithCompact(SMnode *pMnode, STrans *pTrans) {
300✔
985
  int32_t      code = 0;
300✔
986
  void        *pIter = NULL;
300✔
987
  bool         conflict = false;
300✔
988
  SCompactObj *pCompact = NULL;
300✔
989

990
  while (1) {
4✔
991
    bool thisConflict = false;
304✔
992
    pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
304✔
993
    if (pIter == NULL) break;
304✔
994

995
    if (pTrans->conflict == TRN_CONFLICT_GLOBAL) {
4✔
996
      thisConflict = true;
2✔
997
    }
998
    if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
4!
999
      if (taosStrcasecmp(pTrans->dbname, pCompact->dbname) == 0) thisConflict = true;
2!
1000
    }
1001

1002
    if (thisConflict) {
4!
1003
      mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with compact:%d db:%s", pTrans->id,
4!
1004
             pTrans->dbname, pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
1005
      conflict = true;
4✔
1006
    } else {
1007
      mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with compact:%d db:%s", pTrans->id, pTrans->dbname,
×
1008
            pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
1009
    }
1010
    sdbRelease(pMnode->pSdb, pCompact);
4✔
1011
  }
1012

1013
  if (conflict) {
300✔
1014
    code = TSDB_CODE_MND_TRANS_CONFLICT_COMPACT;
4✔
1015
    mError("trans:%d, failed to check tran conflict with compact since %s", pTrans->id, tstrerror(code));
4!
1016
    TAOS_RETURN(code);
4✔
1017
  }
1018

1019
  TAOS_RETURN(code);
296✔
1020
}
1021

1022
static bool mndTransActionsOfSameType(SArray *pActions) {
105,756✔
1023
  int32_t size = taosArrayGetSize(pActions);
105,756✔
1024
  ETrnAct lastActType = TRANS_ACTION_NULL;
105,756✔
1025
  bool    same = true;
105,756✔
1026
  for (int32_t i = 0; i < size; ++i) {
408,728✔
1027
    STransAction *pAction = taosArrayGet(pActions, i);
302,972✔
1028
    if (i > 0) {
302,972✔
1029
      if (lastActType != pAction->actionType) {
224,287!
1030
        same = false;
×
1031
        break;
×
1032
      }
1033
    }
1034
    lastActType = pAction->actionType;
302,972✔
1035
  }
1036
  return same;
105,756✔
1037
}
1038

1039
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
46,335✔
1040
  int32_t code = 0;
46,335✔
1041
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
46,335✔
1042
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
43,621!
1043
      code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1044
      mError("trans:%d, types of parallel redo actions are not the same", pTrans->id);
×
1045
      TAOS_RETURN(code);
×
1046
    }
1047

1048
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
43,621✔
1049
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
15,800!
1050
        code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1051
        mError("trans:%d, types of parallel undo actions are not the same", pTrans->id);
×
1052
        TAOS_RETURN(code);
×
1053
      }
1054
    }
1055
  }
1056

1057
  TAOS_RETURN(code);
46,335✔
1058
}
1059

1060
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
46,335✔
1061
  int32_t code = 0;
46,335✔
1062
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
46,335!
1063
    code = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
×
1064
    mError("trans:%d, commit actions of non-changeless trans are empty", pTrans->id);
×
1065
    TAOS_RETURN(code);
×
1066
  }
1067
  if (mndTransActionsOfSameType(pTrans->commitActions) == false) {
46,335!
1068
    code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1069
    mError("trans:%d, types of commit actions are not the same", pTrans->id);
×
1070
    TAOS_RETURN(code);
×
1071
  }
1072

1073
  TAOS_RETURN(code);
46,335✔
1074
}
1075

1076
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
46,335✔
1077
  int32_t code = 0;
46,335✔
1078
  if (pTrans == NULL) {
46,335!
1079
    return TSDB_CODE_INVALID_PARA;
×
1080
  }
1081

1082
  mInfo("trans:%d, action list:", pTrans->id);
46,335!
1083
  int32_t index = 0;
46,335✔
1084
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
67,973✔
1085
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
21,638✔
1086
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
21,638!
1087
          mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1088
  }
1089

1090
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->redoActions); ++i, ++index) {
140,902✔
1091
    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
94,567✔
1092
    mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
94,567!
1093
          mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));;
1094
  }
1095

1096
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
245,103✔
1097
    STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
198,768✔
1098
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
198,768!
1099
          mndTransStr(pAction->stage), i, sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1100
  }
1101

1102
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
92,541✔
1103
    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
46,206✔
1104
    if(pAction->actionType == TRANS_ACTION_MSG){
46,206✔
1105
      mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
32,569!
1106
            mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));;
1107
    }
1108
    else{
1109
      mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
13,637!
1110
            mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1111
    }
1112
  }
1113

1114

1115
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
46,335!
1116

1117
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
46,335!
1118

1119
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
46,335!
1120

1121
  mInfo("trans:%d, prepare transaction", pTrans->id);
46,335!
1122
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
46,335✔
1123
    mError("trans:%d, failed to prepare since %s", pTrans->id, tstrerror(code));
13!
1124
    sdbWriteLock(pMnode->pSdb, SDB_TRANS);
13✔
1125
    tsMaxTransId = TMAX(pTrans->id, tsMaxTransId);
13✔
1126
    sdbUnLock(pMnode->pSdb, SDB_TRANS);
13✔
1127
    TAOS_RETURN(code);
13✔
1128
  }
1129
  mInfo("trans:%d, prepare finished", pTrans->id);
46,322!
1130

1131
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
46,322✔
1132
  if (pNew == NULL) {
46,322!
1133
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1134
    if (terrno != 0) code = terrno;
×
1135
    mError("trans:%d, failed to read from sdb since %s", pTrans->id, tstrerror(code));
×
1136
    TAOS_RETURN(code);
×
1137
  }
1138

1139
  pNew->pRpcArray = pTrans->pRpcArray;
46,322✔
1140
  pNew->rpcRsp = pTrans->rpcRsp;
46,322✔
1141
  pNew->rpcRspLen = pTrans->rpcRspLen;
46,322✔
1142
  pNew->mTraceId = pTrans->mTraceId;
46,322✔
1143
  pTrans->pRpcArray = NULL;
46,322✔
1144
  pTrans->rpcRsp = NULL;
46,322✔
1145
  pTrans->rpcRspLen = 0;
46,322✔
1146

1147
  mInfo("trans:%d, execute transaction in prepare", pTrans->id);
46,322!
1148
  mndTransExecute(pMnode, pNew, false);
46,322✔
1149
  mndReleaseTrans(pMnode, pNew);
46,322✔
1150
  // TDOD change to TAOS_RETURN(code);
1151
  return 0;
46,322✔
1152
}
1153

1154
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
46,336✔
1155
  int32_t code = 0;
46,336✔
1156
  mInfo("trans:%d, commit transaction", pTrans->id);
46,336!
1157
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
46,336✔
1158
    mError("trans:%d, failed to commit since %s", pTrans->id, tstrerror(code));
12!
1159
    TAOS_RETURN(code);
12✔
1160
  }
1161
  mInfo("trans:%d, commit finished", pTrans->id);
46,324!
1162
  TAOS_RETURN(code);
46,324✔
1163
}
1164

1165
static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
2✔
1166
  int32_t code = 0;
2✔
1167
  mInfo("trans:%d, rollback transaction", pTrans->id);
2!
1168
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
2!
UNCOV
1169
    mError("trans:%d, failed to rollback since %s", pTrans->id, tstrerror(code));
×
1170
    TAOS_RETURN(code);
×
1171
  }
1172
  mInfo("trans:%d, rollback finished", pTrans->id);
2!
1173
  TAOS_RETURN(code);
2✔
1174
}
1175

1176
static int32_t mndTransPreFinish(SMnode *pMnode, STrans *pTrans) {
2✔
1177
  int32_t code = 0;
2✔
1178
  mInfo("trans:%d, pre-finish transaction", pTrans->id);
2!
1179
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
2!
UNCOV
1180
    mError("trans:%d, failed to pre-finish since %s", pTrans->id, tstrerror(code));
×
1181
    TAOS_RETURN(code);
×
1182
  }
1183
  mInfo("trans:%d, pre-finish finished", pTrans->id);
2!
1184
  TAOS_RETURN(code);
2✔
1185
}
1186

1187
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
271,715✔
1188
  bool    sendRsp = false;
271,715✔
1189
  int32_t code = pTrans->code;
271,715✔
1190

1191
  if (pTrans->stage == TRN_STAGE_FINISH) {
271,715✔
1192
    sendRsp = true;
102,939✔
1193
  }
1194

1195
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
271,715✔
1196
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
77,458!
1197
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
12!
1198
      sendRsp = true;
12✔
1199
    }
1200
  } else {
1201
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
194,257✔
1202
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
128,161!
1203
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
UNCOV
1204
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1205
      } else {
1206
        if (pTrans->failedTimes > 6) sendRsp = true;
128,161✔
1207
      }
1208
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
128,161✔
1209
    }
1210
  }
1211

1212
  if (!sendRsp) {
271,715✔
1213
    return;
168,762✔
1214
  } else {
1215
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
102,953!
1216
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1217
  }
1218

1219
  mInfo("vgId:1, trans:%d, start to lock rpc array", pTrans->id);
102,953!
1220
  taosWLockLatch(&pTrans->lockRpcArray);
102,953✔
1221
  int32_t size = taosArrayGetSize(pTrans->pRpcArray);
102,953✔
1222
  if (size <= 0) {
102,953✔
1223
    taosWUnLockLatch(&pTrans->lockRpcArray);
74,437✔
1224
    return;
74,437✔
1225
  }
1226

1227
  for (int32_t i = 0; i < size; ++i) {
57,032✔
1228
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
28,516✔
1229
    if (pInfo->handle != NULL) {
28,516✔
1230
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) {
27,002!
1231
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
1✔
1232
      }
1233
      if (code == TSDB_CODE_SYN_TIMEOUT) {
27,002!
UNCOV
1234
        code = TSDB_CODE_MND_TRANS_SYNC_TIMEOUT;
×
1235
      }
1236

1237
      if (i != 0 && code == 0) {
27,002!
UNCOV
1238
        code = TSDB_CODE_MNODE_NOT_FOUND;
×
1239
      }
1240
      mInfo("vgId:1, trans:%d, client:%d start to send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code,
27,002!
1241
            mndTransStr(pTrans->stage), pInfo->ahandle);
1242

1243
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
27,002✔
1244

1245
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
27,002✔
1246
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
3,941!
1247
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
3,941✔
1248
        if (pDb != NULL) {
3,941!
1249
          for (int32_t j = 0; j < 12; j++) {
4,744✔
1250
            bool ready = mndIsDbReady(pMnode, pDb);
4,742✔
1251
            if (!ready) {
4,742✔
1252
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
803!
1253
              taosMsleep(1000);
803✔
1254
            } else {
1255
              break;
3,939✔
1256
            }
1257
          }
1258
        }
1259
        mndReleaseDb(pMnode, pDb);
3,941✔
1260
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
23,061✔
1261
        void   *pCont = NULL;
7,277✔
1262
        int32_t contLen = 0;
7,277✔
1263
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
7,277✔
1264
          mndTransSetRpcRsp(pTrans, pCont, contLen);
7,275✔
1265
        }
1266
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
15,784✔
1267
        void   *pCont = NULL;
451✔
1268
        int32_t contLen = 0;
451✔
1269
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
451!
1270
          mndTransSetRpcRsp(pTrans, pCont, contLen);
451✔
1271
        }
1272
      }
1273

1274
      if (pTrans->rpcRspLen != 0) {
27,002✔
1275
        void *rpcCont = rpcMallocCont(pTrans->rpcRspLen);
13,174✔
1276
        if (rpcCont != NULL) {
13,174!
1277
          memcpy(rpcCont, pTrans->rpcRsp, pTrans->rpcRspLen);
13,174✔
1278
          rspMsg.pCont = rpcCont;
13,174✔
1279
          rspMsg.contLen = pTrans->rpcRspLen;
13,174✔
1280
        }
1281
      }
1282

1283
      tmsgSendRsp(&rspMsg);
27,002✔
1284

1285
      mInfo("vgId:1, trans:%d, client:%d send rsp finished, code:0x%x stage:%s app:%p", pTrans->id, i, code,
27,002!
1286
            mndTransStr(pTrans->stage), pInfo->ahandle);
1287
    }
1288
  }
1289
  taosArrayClear(pTrans->pRpcArray);
28,516✔
1290
  taosWUnLockLatch(&pTrans->lockRpcArray);
28,516✔
1291
}
1292

1293
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
97,903✔
1294
  int32_t code = 0;
97,903✔
1295
  SMnode *pMnode = pRsp->info.node;
97,903✔
1296
#ifndef TD_ASTRA_32
1297
  int64_t signature = (int64_t)(pRsp->info.ahandle);
97,903✔
1298
  int32_t transId = (int32_t)(signature >> 32);
97,903✔
1299
  int32_t action = (int32_t)((signature << 32) >> 32);
97,903✔
1300
#else
1301
  int32_t transId = (int32_t)(pRsp->info.ahandle);
1302
  int32_t action = (int32_t)(pRsp->info.ahandleEx);
1303
#endif
1304
  STrans *pTrans = mndAcquireTrans(pMnode, transId);
97,903✔
1305
  if (pTrans == NULL) {
97,903!
UNCOV
1306
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1307
    if (terrno != 0) code = terrno;
×
1308
    mError("trans:%d, failed to get transId from vnode rsp since %s", transId, tstrerror(code));
×
1309
    goto _OVER;
×
1310
  }
1311

1312
  SArray *pArray = NULL;
97,903✔
1313
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
97,903✔
1314
    pArray = pTrans->redoActions;
97,895✔
1315
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
8!
1316
    pArray = pTrans->undoActions;
8✔
1317
  } else {
UNCOV
1318
    mError("trans:%d, invalid trans stage:%d while recv action rsp", pTrans->id, pTrans->stage);
×
1319
    goto _OVER;
×
1320
  }
1321

1322
  if (pArray == NULL) {
97,903!
UNCOV
1323
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
×
1324
    goto _OVER;
×
1325
  }
1326

1327
  int32_t actionNum = taosArrayGetSize(pArray);
97,903✔
1328
  if (action < 0 || action >= actionNum) {
97,903!
UNCOV
1329
    mError("trans:%d, invalid action:%d", transId, action);
×
1330
    goto _OVER;
×
1331
  }
1332

1333
  STransAction *pAction = taosArrayGet(pArray, action);
97,903✔
1334
  if (pAction != NULL) {
97,903!
1335
    pAction->msgReceived = 1;
97,903✔
1336
    pAction->errCode = pRsp->code;
97,903✔
1337
    pAction->endTime = taosGetTimestampMs();
97,903✔
1338

1339
    // pTrans->lastErrorNo = pRsp->code;
1340
    mndSetTransLastAction(pTrans, pAction);
97,903✔
1341

1342
    mInfo("trans:%d, %s:%d response is received, received code:0x%x(%s), accept:0x%x(%s) retry:0x%x(%s)", transId,
97,903!
1343
          mndTransStr(pAction->stage), action, pRsp->code, tstrerror(pRsp->code), pAction->acceptableCode,
1344
          tstrerror(pAction->acceptableCode), pAction->retryCode, tstrerror(pAction->retryCode));
1345
  } else {
UNCOV
1346
    mInfo("trans:%d, invalid action, index:%d, code:0x%x", transId, action, pRsp->code);
×
1347
  }
1348

1349
  mInfo("trans:%d, execute transaction in process response", pTrans->id);
97,903!
1350
  mndTransExecute(pMnode, pTrans, true);
97,903✔
1351

1352
_OVER:
97,903✔
1353
  mndReleaseTrans(pMnode, pTrans);
97,903✔
1354
  TAOS_RETURN(code);
97,903✔
1355
}
1356

1357
static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
4,799✔
1358
  pAction->rawWritten = 0;
4,799✔
1359
  pAction->msgSent = 0;
4,799✔
1360
  pAction->msgReceived = 0;
4,799✔
1361
  if (pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR || pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR ||
4,799!
1362
      pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
4,799✔
1363
    pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
10✔
1364
    mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
10!
1365
          pAction->id, pAction->epSet.inUse);
1366
  } else {
1367
    mInfo("trans:%d, %s:%d execute status is reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
4,789!
1368
  }
1369
  pAction->errCode = 0;
4,799✔
1370
}
4,799✔
1371

1372
static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
4✔
1373
  int32_t numOfActions = taosArrayGetSize(pArray);
4✔
1374

1375
  for (int32_t action = 0; action < numOfActions; ++action) {
24✔
1376
    STransAction *pAction = taosArrayGet(pArray, action);
20✔
1377
    if (pAction->msgSent && pAction->msgReceived &&
20!
1378
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
20!
1379
      continue;
16✔
1380
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
4!
1381

1382
    mndTransResetAction(pMnode, pTrans, pAction);
4✔
1383
  }
1384
}
4✔
1385

1386
// execute in sync context
1387
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
450,621✔
1388
  if (pAction->rawWritten) return 0;
450,621✔
1389
  if (topHalf) {
250,321✔
1390
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
14✔
1391
  }
1392

1393
  if (pAction->pRaw->type >= SDB_MAX) {
250,307!
UNCOV
1394
    pAction->rawWritten = true;
×
UNCOV
1395
    pAction->errCode = 0;
×
1396
    mndSetTransLastAction(pTrans, pAction);
×
1397
    mInfo("skip sdb raw type:%d since it is not supported", pAction->pRaw->type);
×
1398
    TAOS_RETURN(TSDB_CODE_SUCCESS);
×
1399
  }
1400

1401
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pAction->pRaw);
250,307✔
1402
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
250,307!
1403
    pAction->rawWritten = true;
250,307✔
1404
    pAction->errCode = 0;
250,307✔
1405
    code = 0;
250,307✔
1406
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
250,307!
1407
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1408

1409
    mndSetTransLastAction(pTrans, pAction);
250,307✔
1410
  } else {
UNCOV
1411
    pAction->errCode = (terrno != 0) ? terrno : code;
×
UNCOV
1412
    mError("trans:%d, %s:%d failed to write sdb since %s, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage),
×
1413
           pAction->id, tstrerror(code), sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1414
    mndSetTransLastAction(pTrans, pAction);
×
1415
  }
1416

1417
  TAOS_RETURN(code);
250,307✔
1418
}
1419

1420
// execute in trans context
1421
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
675,066✔
1422
                                     bool notSend) {
1423
  if (pAction->msgSent) return 0;
675,066✔
1424
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
137,992✔
1425
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
39,814✔
1426
  }
1427

1428
  if (notSend) {
98,178✔
1429
    mInfo("trans:%d, action:%d skip to execute msg action", pTrans->id, pAction->id);
182!
1430
    return 0;
182✔
1431
  }
1432

1433
#ifndef TD_ASTRA_32
1434
  int64_t signature = pTrans->id;
97,996✔
1435
  signature = (signature << 32);
97,996✔
1436
  signature += pAction->id;
97,996✔
1437

1438
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
97,996✔
1439
#else
1440
  SRpcMsg rpcMsg = {.msgType = pAction->msgType,
1441
                    .contLen = pAction->contLen,
1442
                    .info.ahandle = (void *)pTrans->id,
1443
                    .info.ahandleEx = (void *)pAction->id};
1444
#endif
1445
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
97,996✔
1446
  if (rpcMsg.pCont == NULL) {
97,996!
UNCOV
1447
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
UNCOV
1448
    return -1;
×
1449
  }
1450
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
97,996✔
1451
  rpcMsg.info.notFreeAhandle = 1;
97,996✔
1452

1453
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
97,996✔
1454

1455
  char    detail[1024] = {0};
97,996✔
1456
  int32_t len = tsnprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
97,996!
1457
                          pAction->epSet.numOfEps, pAction->epSet.inUse);
97,996✔
1458
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
201,872✔
1459
    len += tsnprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
103,876✔
1460
                     pAction->epSet.eps[i].port);
103,876✔
1461
  }
1462

1463
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
97,996✔
1464
  if (code == 0) {
97,996!
1465
    pAction->msgSent = 1;
97,996✔
1466
    // pAction->msgReceived = 0;
1467
    pAction->errCode = TSDB_CODE_ACTION_IN_PROGRESS;
97,996✔
1468
    pAction->startTime = taosGetTimestampMs();
97,996✔
1469
    pAction->endTime = 0;
97,996✔
1470
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
97,996!
1471

1472
    mndSetTransLastAction(pTrans, pAction);
97,996✔
1473
  } else {
UNCOV
1474
    pAction->msgSent = 0;
×
UNCOV
1475
    pAction->msgReceived = 0;
×
UNCOV
1476
    pAction->errCode = (terrno != 0) ? terrno : code;
×
UNCOV
1477
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
×
1478
           detail);
1479

UNCOV
1480
    mndSetTransLastAction(pTrans, pAction);
×
1481
  }
1482

1483
  TAOS_RETURN(code);
97,996✔
1484
}
1485

UNCOV
1486
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
×
UNCOV
1487
  if (!topHalf) return TSDB_CODE_MND_TRANS_CTX_SWITCH;
×
UNCOV
1488
  pAction->rawWritten = 0;
×
UNCOV
1489
  pAction->errCode = 0;
×
UNCOV
1490
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1491

UNCOV
1492
  mndSetTransLastAction(pTrans, pAction);
×
UNCOV
1493
  return 0;
×
1494
}
1495

1496
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
1,125,687✔
1497
                                        bool notSend) {
1498
  if (pAction->actionType == TRANS_ACTION_RAW) {
1,125,687✔
1499
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
450,621✔
1500
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
675,066!
1501
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf, notSend);
675,066✔
1502
  } else {
UNCOV
1503
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1504
  }
1505
}
1506

1507
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf, bool notSend) {
249,343✔
1508
  int32_t numOfActions = taosArrayGetSize(pArray);
249,343✔
1509
  int32_t code = 0;
249,343✔
1510

1511
  for (int32_t action = 0; action < numOfActions; ++action) {
1,278,393✔
1512
    STransAction *pAction = taosArrayGet(pArray, action);
1,061,608✔
1513
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, notSend);
1,061,608✔
1514
    if (code != 0) {
1,061,608✔
1515
      mInfo("trans:%d, action:%d not executed since %s. numOfActions:%d", pTrans->id, action, tstrerror(code),
32,558!
1516
            numOfActions);
1517
      break;
32,558✔
1518
    }
1519
  }
1520

1521
  return code;
249,343✔
1522
}
1523

1524
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf, bool notSend) {
289,040✔
1525
  int32_t numOfActions = taosArrayGetSize(pArray);
289,040✔
1526
  int32_t code = 0;
289,040✔
1527
  if (numOfActions == 0) return 0;
289,040✔
1528

1529
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf, notSend)) != 0) {
249,343✔
1530
    return code;
32,558✔
1531
  }
1532

1533
  int32_t       numOfExecuted = 0;
216,785✔
1534
  int32_t       errCode = 0;
216,785✔
1535
  STransAction *pErrAction = NULL;
216,785✔
1536
  for (int32_t action = 0; action < numOfActions; ++action) {
1,245,835✔
1537
    STransAction *pAction = taosArrayGet(pArray, action);
1,029,050✔
1538
    if (pAction->msgReceived || pAction->rawWritten) {
1,029,050✔
1539
      numOfExecuted++;
724,447✔
1540
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
724,447✔
1541
        errCode = pAction->errCode;
31✔
1542
        pErrAction = pAction;
31✔
1543
      }
1544
    } else {
1545
      pErrAction = pAction;
304,603✔
1546
    }
1547
  }
1548

1549
  mndSetTransLastAction(pTrans, pErrAction);
216,785✔
1550

1551
  if (numOfExecuted == numOfActions) {
216,785✔
1552
    if (errCode == 0) {
128,063✔
1553
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
128,059!
1554
      return 0;
128,059✔
1555
    } else {
1556
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
4!
1557
      mndTransResetActions(pMnode, pTrans, pArray);
4✔
1558
      terrno = errCode;
4✔
1559
      return errCode;
4✔
1560
    }
1561
  } else {
1562
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
88,722!
1563

1564
    for (int32_t action = 0; action < numOfActions; ++action) {
612,028✔
1565
      STransAction *pAction = taosArrayGet(pArray, action);
523,306✔
1566
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
523,306✔
1567
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1568
             pAction->acceptableCode, pAction->retryCode);
1569
      if (pAction->msgSent) {
523,306✔
1570
        if (pAction->msgReceived) {
523,124✔
1571
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
218,703✔
1572
            mndTransResetAction(pMnode, pTrans, pAction);
27✔
1573
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
27!
1574
          }
1575
        }
1576
      }
1577
    }
1578
    return TSDB_CODE_ACTION_IN_PROGRESS;
88,722✔
1579
  }
1580
}
1581

1582
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
186,093✔
1583
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf, notSend);
186,093✔
1584
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
186,093✔
1585
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
4!
1586
           mndStrExecutionContext(topHalf));
1587
  }
1588
  return code;
186,093✔
1589
}
1590

1591
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
14✔
1592
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions, topHalf, notSend);
14✔
1593
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
14!
UNCOV
1594
    mError("trans:%d, failed to execute undoActions since %s. in %s", pTrans->id, terrstr(),
×
1595
           mndStrExecutionContext(topHalf));
1596
  }
1597
  return code;
14✔
1598
}
1599

1600
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
102,933✔
1601
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf, true);
102,933✔
1602
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
102,933!
1603
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1604
           mndStrExecutionContext(topHalf));
1605
  }
1606
  return code;
102,933✔
1607
}
1608

1609
static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
28,999✔
1610
  int32_t code = 0;
28,999✔
1611
  int32_t numOfActions = taosArrayGetSize(pActions);
28,999✔
1612
  if (numOfActions == 0) return code;
28,999✔
1613

1614
  if (pTrans->actionPos >= numOfActions) {
28,997✔
1615
    return code;
3,000✔
1616
  }
1617

1618
  mInfo("trans:%d, execute %d actions serial, begin at action:%d, stage:%s", pTrans->id, numOfActions,
25,997!
1619
        pTrans->actionPos, mndTransStr(pTrans->stage));
1620

1621
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
38,789✔
1622
    STransAction *pAction = taosArrayGet(pActions, action);
36,077✔
1623

1624
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d, msgSent:%d, msgReceived:%d", 
36,077!
1625
          pTrans->id, pTrans->actionPos, mndTransStr(pAction->stage), pAction->actionType, pAction->msgSent,
1626
          pAction->msgReceived);
1627

1628
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, false);
36,077✔
1629
    if (code == 0) {
36,077✔
1630
      if (pAction->msgSent) {
28,807✔
1631
        if (pAction->msgReceived) {
25,425✔
1632
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
11,276✔
1633
            code = pAction->errCode;
4,768✔
1634
            mndTransResetAction(pMnode, pTrans, pAction);
4,768✔
1635
          } else {
1636
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
6,508!
1637
          }
1638
        } else {
1639
          code = TSDB_CODE_ACTION_IN_PROGRESS;
14,149✔
1640
        }
1641
      } else if (pAction->rawWritten) {
3,382!
1642
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
3,382!
UNCOV
1643
          code = pAction->errCode;
×
1644
        } else {
1645
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,382!
1646
        }
1647
      } else {
1648
      }
1649
    }
1650

1651
    if (code == 0) {
36,077✔
1652
      pTrans->failedTimes = 0;
9,890✔
1653
    }
1654
    mndSetTransLastAction(pTrans, pAction);
36,077✔
1655

1656
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
36,077✔
1657
      pTrans->lastErrorNo = code;
9,119✔
1658
      pTrans->code = code;
9,119✔
1659
      mInfo("trans:%d, %s:%d, cannot execute next action in %s, code:%s", pTrans->id, mndTransStr(pAction->stage),
9,119!
1660
            action, mndStrExecutionContext(topHalf), tstrerror(code));
1661
      break;
9,119✔
1662
    }
1663

1664
    if (code == 0) {
26,958✔
1665
      pTrans->code = 0;
8,041✔
1666
      pTrans->actionPos++;
8,041✔
1667
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
8,041!
1668
            pAction->id);
1669
      (void)taosThreadMutexUnlock(&pTrans->mutex);
8,041✔
1670
      code = mndTransSync(pMnode, pTrans);
8,041✔
1671
      (void)taosThreadMutexLock(&pTrans->mutex);
8,041✔
1672
      if (code != 0) {
8,041!
UNCOV
1673
        pTrans->actionPos--;
×
UNCOV
1674
        pTrans->code = terrno;
×
UNCOV
1675
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1676
               mndTransStr(pAction->stage), pAction->id, terrstr());
UNCOV
1677
        break;
×
1678
      }
1679
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
18,917✔
1680
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
14,149!
1681
      break;
14,149✔
1682
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,768✔
1683
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
27✔
1684
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,751!
1685
            code, tstrerror(code));
1686
      pTrans->lastErrorNo = code;
4,751✔
1687
      taosMsleep(300);
4,751✔
1688
      action--;
4,751✔
1689
      continue;
4,751✔
1690
    } else {
1691
      terrno = code;
17✔
1692
      pTrans->lastErrorNo = code;
17✔
1693
      pTrans->code = code;
17✔
1694
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and wait another schedule, failedTimes:%d", pTrans->id,
17!
1695
            mndTransStr(pAction->stage), pAction->id, code, tstrerror(code), pTrans->failedTimes);
1696
      break;
17✔
1697
    }
1698
  }
1699

1700
  return code;
25,997✔
1701
}
1702

1703
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
28,999✔
1704
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
28,999✔
1705
  (void)taosThreadMutexLock(&pTrans->mutex);
28,999✔
1706
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
28,999!
1707
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
28,999✔
1708
  }
1709
  (void)taosThreadMutexUnlock(&pTrans->mutex);
28,999✔
1710
  return code;
28,999✔
1711
}
1712

UNCOV
1713
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
UNCOV
1714
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
×
UNCOV
1715
  (void)taosThreadMutexLock(&pTrans->mutex);
×
UNCOV
1716
  if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
UNCOV
1717
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1718
  }
UNCOV
1719
  (void)taosThreadMutexUnlock(&pTrans->mutex);
×
UNCOV
1720
  return code;
×
1721
}
1722

1723
bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
56,651✔
1724
  bool    continueExec = true;
56,651✔
1725
  int32_t code = 0;
56,651✔
1726
  terrno = 0;
56,651✔
1727

1728
  int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions);
56,651✔
1729
  if (numOfActions == 0) goto _OVER;
56,651✔
1730

1731
  mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions);
18,319!
1732

1733
  for (int32_t action = 0; action < numOfActions; ++action) {
46,321✔
1734
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
28,002✔
1735
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf, true);
28,002✔
1736
    if (code != 0) {
28,002!
UNCOV
1737
      terrno = code;
×
UNCOV
1738
      mError("trans:%d, failed to execute prepare action:%d, numOfActions:%d, since %s", pTrans->id, action,
×
1739
             numOfActions, tstrerror(code));
UNCOV
1740
      return false;
×
1741
    }
1742
  }
1743

1744
_OVER:
18,319✔
1745
  pTrans->stage = TRN_STAGE_REDO_ACTION;
56,651✔
1746
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
56,651!
1747
  return continueExec;
56,651✔
1748
}
1749

1750
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
215,092✔
1751
  bool    continueExec = true;
215,092✔
1752
  int32_t code = 0;
215,092✔
1753
  terrno = 0;
215,092✔
1754

1755
  if (pTrans->exec == TRN_EXEC_SERIAL) {
215,092✔
1756
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
28,999✔
1757
  } else {
1758
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf, notSend);
186,093✔
1759
  }
1760

1761
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
215,092!
UNCOV
1762
    pTrans->lastErrorNo = code;
×
UNCOV
1763
    pTrans->code = code;
×
UNCOV
1764
    mInfo(
×
1765
        "trans:%d, failed to execute, will retry redo action stage in 100 ms , in %s, "
1766
        "continueExec:%d, code:%s",
1767
        pTrans->id, mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
UNCOV
1768
    taosMsleep(100);
×
UNCOV
1769
    return true;
×
1770
  } else {
1771
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
215,092✔
1772
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
65,874!
1773
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1774
      return false;
65,874✔
1775
    }
1776
  }
1777
  terrno = code;
149,218✔
1778

1779
  if (code == 0) {
149,218✔
1780
    pTrans->code = 0;
46,336✔
1781
    pTrans->stage = TRN_STAGE_COMMIT;
46,336✔
1782
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
46,336!
1783
    continueExec = true;
46,336✔
1784
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
102,882!
1785
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
102,861!
1786
    continueExec = false;
102,861✔
1787
  } else {
1788
    pTrans->failedTimes++;
21✔
1789
    pTrans->code = terrno;
21✔
1790
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
21✔
1791
      if (pTrans->lastAction != 0) {
2!
1792
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
2✔
1793
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
2!
UNCOV
1794
          if (pTrans->failedTimes < 6) {
×
UNCOV
1795
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
×
1796
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
UNCOV
1797
            taosMsleep(1000);
×
UNCOV
1798
            continueExec = true;
×
UNCOV
1799
            return true;
×
1800
          }
1801
        }
1802
      }
1803

1804
      pTrans->stage = TRN_STAGE_ROLLBACK;
2✔
1805
      pTrans->actionPos = 0;
2✔
1806
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
2!
1807
             pTrans->actionPos);
1808
      continueExec = true;
2✔
1809
    } else {
1810
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
19!
1811
      continueExec = false;
19✔
1812
    }
1813
  }
1814

1815
  return continueExec;
149,218✔
1816
}
1817

1818
// execute in trans context
1819
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
46,336✔
1820
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
46,336!
1821

1822
  bool    continueExec = true;
46,336✔
1823
  int32_t code = mndTransCommit(pMnode, pTrans);
46,336✔
1824

1825
  if (code == 0) {
46,336✔
1826
    pTrans->code = 0;
46,324✔
1827
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
46,324✔
1828
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
46,324!
1829
    continueExec = true;
46,324✔
1830
  } else {
1831
    pTrans->code = terrno;
12✔
1832
    pTrans->failedTimes++;
12✔
1833
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
12!
1834
    continueExec = false;
12✔
1835
  }
1836

1837
  return continueExec;
46,336✔
1838
}
1839

1840
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
102,933✔
1841
  bool    continueExec = true;
102,933✔
1842
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans, topHalf);
102,933✔
1843

1844
  if (code == 0) {
102,933!
1845
    pTrans->code = 0;
102,933✔
1846
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
102,933✔
1847
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
102,933!
1848
    continueExec = true;
102,933✔
UNCOV
1849
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
×
UNCOV
1850
    pTrans->code = 0;
×
UNCOV
1851
    pTrans->stage = TRN_STAGE_COMMIT;
×
UNCOV
1852
    mInfo("trans:%d, back to commit stage", pTrans->id);
×
UNCOV
1853
    continueExec = true;
×
1854
  } else {
UNCOV
1855
    pTrans->code = terrno;
×
UNCOV
1856
    pTrans->failedTimes++;
×
UNCOV
1857
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1858
    continueExec = false;
×
1859
  }
1860

1861
  return continueExec;
102,933✔
1862
}
1863

1864
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
14✔
1865
  bool    continueExec = true;
14✔
1866
  int32_t code = 0;
14✔
1867

1868
  if (pTrans->exec == TRN_EXEC_SERIAL) {
14!
UNCOV
1869
    code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
×
1870
  } else {
1871
    code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf, notSend);
14✔
1872
  }
1873

1874
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
14✔
1875
  terrno = code;
12✔
1876

1877
  if (code == 0) {
12✔
1878
    pTrans->stage = TRN_STAGE_PRE_FINISH;
2✔
1879
    mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
2!
1880
    continueExec = true;
2✔
1881
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
10!
1882
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
10!
1883
    continueExec = false;
10✔
1884
  } else {
UNCOV
1885
    pTrans->failedTimes++;
×
UNCOV
1886
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
UNCOV
1887
    continueExec = false;
×
1888
  }
1889

1890
  return continueExec;
12✔
1891
}
1892

1893
// in trans context
1894
static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2✔
1895
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
2!
1896

1897
  bool    continueExec = true;
2✔
1898
  int32_t code = mndTransRollback(pMnode, pTrans);
2✔
1899

1900
  if (code == 0) {
2!
1901
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
2✔
1902
    continueExec = true;
2✔
1903
  } else {
UNCOV
1904
    pTrans->failedTimes++;
×
UNCOV
1905
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
UNCOV
1906
    continueExec = false;
×
1907
  }
1908

1909
  return continueExec;
2✔
1910
}
1911

1912
// excute in trans context
1913
static bool mndTransPerformPreFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
2✔
1914
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
2!
1915

1916
  bool    continueExec = true;
2✔
1917
  int32_t code = mndTransPreFinish(pMnode, pTrans);
2✔
1918

1919
  if (code == 0) {
2!
1920
    pTrans->stage = TRN_STAGE_FINISH;
2✔
1921
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
2!
1922
    continueExec = true;
2✔
1923
  } else {
UNCOV
1924
    pTrans->failedTimes++;
×
UNCOV
1925
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
UNCOV
1926
    continueExec = false;
×
1927
  }
1928

1929
  return continueExec;
2✔
1930
}
1931

1932
static bool mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
102,937✔
1933
  bool continueExec = false;
102,937✔
1934
  if (topHalf) return continueExec;
102,937✔
1935

1936
  SSdbRaw *pRaw = mndTransEncode(pTrans);
56,611✔
1937
  if (pRaw == NULL) {
56,611!
UNCOV
1938
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
×
UNCOV
1939
    return false;
×
1940
  }
1941
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED));
56,611!
1942

1943
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
56,611✔
1944
  if (code != 0) {
56,611!
UNCOV
1945
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
×
1946
  }
1947

1948
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d createTime:%" PRId64, pTrans->id, pTrans->code,
56,611!
1949
        pTrans->failedTimes, pTrans->createdTime);
1950
  return continueExec;
56,611✔
1951
}
1952

1953
void mndTransExecuteImp(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
271,715✔
1954
  bool continueExec = true;
271,715✔
1955

1956
  while (continueExec) {
739,031✔
1957
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64, pTrans->id,
467,316!
1958
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1959
    pTrans->lastExecTime = taosGetTimestampMs();
467,316✔
1960
    switch (pTrans->stage) {
467,316!
UNCOV
1961
      case TRN_STAGE_PREPARE:
×
UNCOV
1962
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
UNCOV
1963
        break;
×
1964
      case TRN_STAGE_REDO_ACTION:
215,092✔
1965
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf, notSend);
215,092✔
1966
        break;
215,092✔
1967
      case TRN_STAGE_COMMIT:
46,336✔
1968
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
46,336✔
1969
        break;
46,336✔
1970
      case TRN_STAGE_COMMIT_ACTION:
102,933✔
1971
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
102,933✔
1972
        break;
102,933✔
1973
      case TRN_STAGE_ROLLBACK:
2✔
1974
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
2✔
1975
        break;
2✔
1976
      case TRN_STAGE_UNDO_ACTION:
14✔
1977
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf, notSend);
14✔
1978
        break;
14✔
1979
      case TRN_STAGE_PRE_FINISH:
2✔
1980
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
2✔
1981
        break;
2✔
1982
      case TRN_STAGE_FINISH:
102,937✔
1983
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
102,937✔
1984
        break;
102,937✔
UNCOV
1985
      default:
×
UNCOV
1986
        continueExec = false;
×
UNCOV
1987
        break;
×
1988
    }
1989
  }
1990

1991
  mndTransSendRpcRsp(pMnode, pTrans);
271,715✔
1992
}
271,715✔
1993

1994
// start trans, pullup, receive rsp, kill
1995
void mndTransExecute(SMnode *pMnode, STrans *pTrans, bool notSend) {
149,433✔
1996
  bool topHalf = true;
149,433✔
1997
  mndTransExecuteImp(pMnode, pTrans, topHalf, notSend);
149,433✔
1998
}
149,433✔
1999

2000
// update trans
2001
void mndTransRefresh(SMnode *pMnode, STrans *pTrans) {
122,282✔
2002
  bool topHalf = false;
122,282✔
2003
  mndTransExecuteImp(pMnode, pTrans, topHalf, false);
122,282✔
2004
}
122,282✔
2005

2006
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
40,790✔
2007
  mTrace("start to process trans timer");
40,790✔
2008
  mndTransPullup(pReq->info.node);
40,790✔
2009
  return 0;
40,790✔
2010
}
2011

2012
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
1✔
2013
  SArray *pArray = NULL;
1✔
2014
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
1!
2015
    pArray = pTrans->redoActions;
1✔
UNCOV
2016
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
UNCOV
2017
    pArray = pTrans->undoActions;
×
2018
  } else {
UNCOV
2019
    TAOS_RETURN(TSDB_CODE_MND_TRANS_INVALID_STAGE);
×
2020
  }
2021

2022
  if(pTrans->ableToBeKilled == false){
1!
2023
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
1✔
2024
  }
2025
  
2026
  if(pTrans->killMode == TRN_KILL_MODE_SKIP){
×
UNCOV
2027
    for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
2028
      STransAction *pAction = taosArrayGet(pArray, i);
×
UNCOV
2029
      mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
×
2030
            mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
2031
      pAction->msgSent = 1;
×
2032
      pAction->msgReceived = 1;
×
UNCOV
2033
      pAction->errCode = 0;
×
2034
    }
2035
  }
2036
  else if(pTrans->killMode == TRN_KILL_MODE_INTERUPT){
×
2037
    pTrans->stage = TRN_STAGE_PRE_FINISH;
×
2038
  }
2039
  else{
2040
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2041
  }
2042

NEW
2043
  mInfo("trans:%d, execute transaction in kill trans", pTrans->id);
×
NEW
2044
  mndTransExecute(pMnode, pTrans, true);
×
2045
  return 0;
×
2046
}
2047

2048
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
1✔
2049
  SMnode       *pMnode = pReq->info.node;
1✔
2050
  SKillTransReq killReq = {0};
1✔
2051
  int32_t       code = -1;
1✔
2052
  STrans       *pTrans = NULL;
1✔
2053

2054
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
1!
2055
    code = TSDB_CODE_INVALID_MSG;
×
UNCOV
2056
    goto _OVER;
×
2057
  }
2058

2059
  mInfo("trans:%d, start to kill", killReq.transId);
1!
2060
  if ((code = mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS)) != 0) {
1!
2061
    goto _OVER;
1✔
2062
  }
2063

UNCOV
2064
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
×
2065
  if (pTrans == NULL) {
×
2066
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
UNCOV
2067
    if (terrno != 0) code = terrno;
×
UNCOV
2068
    goto _OVER;
×
2069
  }
2070

UNCOV
2071
  code = mndKillTrans(pMnode, pTrans);
×
2072

2073
_OVER:
1✔
2074
  if (code != 0) {
1!
2075
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
1!
2076
  }
2077

2078
  mndReleaseTrans(pMnode, pTrans);
1✔
2079
  TAOS_RETURN(code);
1✔
2080
}
2081

2082
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
292✔
2083

2084
void mndTransPullup(SMnode *pMnode) {
41,284✔
2085
  SSdb   *pSdb = pMnode->pSdb;
41,284✔
2086
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
41,284✔
2087
  if (pArray == NULL) return;
41,284!
2088

2089
  void *pIter = NULL;
41,284✔
2090
  while (1) {
5,208✔
2091
    STrans *pTrans = NULL;
46,492✔
2092
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
46,492✔
2093
    if (pIter == NULL) break;
46,492✔
2094
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
10,416!
UNCOV
2095
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
2096
    }
2097
    sdbRelease(pSdb, pTrans);
5,208✔
2098
  }
2099

2100
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
41,284✔
2101

2102
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
46,492✔
2103
    int32_t *pTransId = taosArrayGet(pArray, i);
5,208✔
2104
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
5,208✔
2105
    if (pTrans != NULL) {
5,208!
2106
      mInfo("trans:%d, execute transaction in trans pullup", pTrans->id);
5,208!
2107
      mndTransExecute(pMnode, pTrans, false);
5,208✔
2108
    }
2109
    mndReleaseTrans(pMnode, pTrans);
5,208✔
2110
  }
2111
  taosArrayDestroy(pArray);
41,284✔
2112
}
2113

2114
static char *formatTimestamp(char *buf, int64_t val, int precision) {
13,134✔
2115
  time_t tt;
2116
  if (precision == TSDB_TIME_PRECISION_MICRO) {
13,134!
UNCOV
2117
    tt = (time_t)(val / 1000000);
×
2118
  }
2119
  if (precision == TSDB_TIME_PRECISION_NANO) {
13,134!
UNCOV
2120
    tt = (time_t)(val / 1000000000);
×
2121
  } else {
2122
    tt = (time_t)(val / 1000);
13,134✔
2123
  }
2124

2125
  struct tm tm;
2126
  if (taosLocalTime(&tt, &tm, NULL, 0, NULL) == NULL) {
13,134!
UNCOV
2127
    mError("failed to get local time");
×
2128
    return NULL;
×
2129
  }
2130
  size_t pos = taosStrfTime(buf, 32, "%Y-%m-%d %H:%M:%S", &tm);
13,136✔
2131

2132
  if (precision == TSDB_TIME_PRECISION_MICRO) {
13,136!
UNCOV
2133
    sprintf(buf + pos, ".%06d", (int)(val % 1000000));
×
2134
  } else if (precision == TSDB_TIME_PRECISION_NANO) {
13,136!
UNCOV
2135
    sprintf(buf + pos, ".%09d", (int)(val % 1000000000));
×
2136
  } else {
2137
    sprintf(buf + pos, ".%03d", (int)(val % 1000));
13,136✔
2138
  }
2139

2140
  return buf;
13,136✔
2141
}
2142

2143
static void mndTransLogAction(STrans *pTrans) {
199✔
2144
  char    detail[512] = {0};
199✔
2145
  int32_t len = 0;
199✔
2146
  int32_t index = 0;
199✔
2147

2148
  if (pTrans->stage == TRN_STAGE_PREPARE) {
199!
UNCOV
2149
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
×
UNCOV
2150
      len = 0;
×
UNCOV
2151
      STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
×
UNCOV
2152
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
UNCOV
2153
                      mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
UNCOV
2154
                      sdbStatusName(pAction->pRaw->status));
×
UNCOV
2155
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2156
    }
2157
  }
2158

2159
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
199✔
2160
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->redoActions); ++i, ++index) {
8,284✔
2161
      len = 0;
8,087✔
2162
      STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
8,087✔
2163
      if (pAction->actionType == TRANS_ACTION_MSG) {
8,087✔
2164
        char bufStart[40] = {0};
6,567✔
2165
        (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
6,567✔
2166

2167
        char endStart[40] = {0};
6,567✔
2168
        (void)formatTimestamp(endStart, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
6,567✔
2169
        len += snprintf(detail + len, sizeof(detail) - len,
13,136!
2170
                        "action:%d, %s:%d msgType:%s,"
2171
                        "sent:%d, received:%d, startTime:%s, endTime:%s, ",
2172
                        index, mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType), pAction->msgSent,
6,568✔
2173
                        pAction->msgReceived, bufStart, endStart);
6,568✔
2174

2175
        SEpSet epset = pAction->epSet;
6,568✔
2176
        if (epset.numOfEps > 0) {
6,568!
2177
          len += snprintf(detail + len, sizeof(detail) - len, "numOfEps:%d inUse:%d ", epset.numOfEps, epset.inUse);
6,568✔
2178
          for (int32_t i = 0; i < epset.numOfEps; ++i) {
15,240✔
2179
            len +=
8,672✔
2180
                snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
8,672✔
2181
          }
2182
        }
2183

2184
        len += snprintf(detail + len, sizeof(detail) - len, ", errCode:0x%x(%s)\n", pAction->errCode & 0xFFFF,
6,568✔
2185
                        tstrerror(pAction->errCode));
2186
      } else {
2187
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s, written:%d\n",
1,520✔
2188
                        index, mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
1,520✔
2189
                        sdbStatusName(pAction->pRaw->status), pAction->rawWritten);
1,520✔
2190
      }
2191
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
8,088✔
2192
    }
2193
  }
2194

2195
  if (pTrans->stage == TRN_STAGE_COMMIT_ACTION) {
198!
UNCOV
2196
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
×
UNCOV
2197
      len = 0;
×
UNCOV
2198
      STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
×
UNCOV
2199
      len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
UNCOV
2200
                      mndTransStr(pAction->stage), i, sdbTableName(pAction->pRaw->type),
×
UNCOV
2201
                      sdbStatusName(pAction->pRaw->status));
×
UNCOV
2202
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2203
    }
2204

UNCOV
2205
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
×
UNCOV
2206
      len = 0;
×
2207
      STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
×
2208
      if (pAction->actionType == TRANS_ACTION_MSG) {
×
2209
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d msgType:%s\n", index,
×
2210
                        mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));
×
2211
        ;
2212
      } else {
2213
        len += snprintf(detail + len, sizeof(detail) - len, "action:%d, %s:%d sdbType:%s, sdbStatus:%s\n", index,
×
UNCOV
2214
                        mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
UNCOV
2215
                        sdbStatusName(pAction->pRaw->status));
×
2216
      }
2217
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2218
    }
2219
  }
2220
}
198✔
2221

2222
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
7,454✔
2223
  SMnode *pMnode = pReq->info.node;
7,454✔
2224
  SSdb   *pSdb = pMnode->pSdb;
7,454✔
2225
  int32_t numOfRows = 0;
7,454✔
2226
  STrans *pTrans = NULL;
7,454✔
2227
  int32_t cols = 0;
7,454✔
2228
  int32_t code = 0;
7,454✔
2229
  int32_t lino = 0;
7,454✔
2230

2231
  while (numOfRows < rows) {
7,653!
2232
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
7,653✔
2233
    if (pShow->pIter == NULL) break;
7,671✔
2234

2235
    cols = 0;
199✔
2236

2237
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2238
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->id, false), pTrans, &lino, _OVER);
199!
2239

2240
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
198✔
2241
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false), pTrans, &lino,
198!
2242
                        _OVER);
2243

2244
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
199✔
2245
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
199✔
2246
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2247
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stage, false), pTrans, &lino, _OVER);
198!
2248

2249
    char opername[TSDB_TRANS_OPER_LEN + VARSTR_HEADER_SIZE] = {0};
199✔
2250
    STR_WITH_MAXSIZE_TO_VARSTR(opername, pTrans->opername, pShow->pMeta->pSchemas[cols].bytes);
199✔
2251
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2252
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)opername, false), pTrans, &lino, _OVER);
199!
2253

2254
    char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
199✔
2255
    STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pTrans->dbname), pShow->pMeta->pSchemas[cols].bytes);
199✔
2256
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2257
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pTrans, &lino, _OVER);
199!
2258

2259
    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
199✔
2260
    STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
199✔
2261
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2262
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbname, false), pTrans, &lino, _OVER);
199!
2263

2264
    const char *killableStr = pTrans->ableToBeKilled ? "yes" : "no";
199!
2265
    char        killableVstr[10 + VARSTR_HEADER_SIZE] = {0};
199✔
2266
    STR_WITH_MAXSIZE_TO_VARSTR(killableVstr, killableStr, 10 + VARSTR_HEADER_SIZE);
199✔
2267
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2268
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killableVstr, false), pTrans, &lino, _OVER);
199!
2269

2270
    /*
2271
    const char *killModeStr = pTrans->killMode == TRN_KILL_MODE_SKIP ? "skip" : "interrupt";
2272
    char        killModeVstr[10 + VARSTR_HEADER_SIZE] = {0};
2273
    STR_WITH_MAXSIZE_TO_VARSTR(killModeVstr, killModeStr, 24);
2274
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
2275
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killModeVstr, false), pTrans, &lino, _OVER);
2276
    */
2277

2278
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
198✔
2279
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
198!
2280
                        _OVER);
2281

2282
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2283
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false), pTrans, &lino,
199!
2284
                        _OVER);
2285

2286
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
198✔
2287
    char    detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
198✔
2288
    int32_t len = tsnprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
595✔
2289
                            pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
198✔
2290
    SEpSet  epset = pTrans->lastEpset;
199✔
2291
    if (epset.numOfEps > 0) {
199✔
2292
      len += tsnprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
336!
2293
                       TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
336✔
2294
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
456✔
2295
        len += snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
288✔
2296
      }
2297
    }
2298
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
199✔
2299
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
199✔
2300
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)lastInfo, false), pTrans, &lino, _OVER);
199!
2301

2302
    mndTransLogAction(pTrans);
199✔
2303

2304
    numOfRows++;
198✔
2305
    sdbRelease(pSdb, pTrans);
198✔
2306
  }
2307

UNCOV
2308
_OVER:
×
2309
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
7,472!
2310
  pShow->numOfRows += numOfRows;
7,468✔
2311
  return numOfRows;
7,468✔
2312
}
2313

UNCOV
2314
static int32_t mndShowTransCommonColumns(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction,
×
2315
                                         int32_t transactionId, int32_t curActionId, int32_t numOfRows, int32_t *cols) {
UNCOV
2316
  int32_t code = 0;
×
UNCOV
2317
  int32_t lino = 0;
×
UNCOV
2318
  int32_t len = 0;
×
2319

UNCOV
2320
  SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, (*cols)++);
×
UNCOV
2321
  TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&transactionId, false), &lino, _OVER);
×
2322

UNCOV
2323
  char action[30 + 1] = {0};
×
UNCOV
2324
  if (curActionId == pAction->id) {
×
2325
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)<-last", mndTransStr(pAction->stage), pAction->id,
×
2326
                    mndTransTypeStr(pAction->actionType));
2327
  } else {
2328
    len += snprintf(action + len, sizeof(action) - len, "%s:%d(%s)", mndTransStr(pAction->stage), pAction->id,
×
2329
                    mndTransTypeStr(pAction->actionType));
2330
  }
2331
  char actionVStr[30 + VARSTR_HEADER_SIZE] = {0};
×
2332
  STR_WITH_MAXSIZE_TO_VARSTR(actionVStr, action, pShow->pMeta->pSchemas[*cols].bytes);
×
UNCOV
2333
  pColInfo = taosArrayGet(pBlock->pDataBlock, (*cols)++);
×
2334
  TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)actionVStr, false), &lino, _OVER);
×
2335
_OVER:
×
2336
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
UNCOV
2337
  return code;
×
2338
}
2339

UNCOV
2340
static void mndShowTransAction(SShowObj *pShow, SSDataBlock *pBlock, STransAction *pAction, int32_t transactionId,
×
2341
                               int32_t curActionId, int32_t rows, int32_t numOfRows) {
2342
  int32_t code = 0;
×
2343
  int32_t lino = 0;
×
2344
  int32_t len = 0;
×
2345
  int32_t cols = 0;
×
2346

2347
  cols = 0;
×
2348

UNCOV
2349
  if (mndShowTransCommonColumns(pShow, pBlock, pAction, transactionId, curActionId, numOfRows, &cols) != 0) return;
×
2350

2351
  if (pAction->actionType == TRANS_ACTION_MSG) {
×
UNCOV
2352
    int32_t len = 0;
×
2353

2354
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2355
    len += snprintf(objType + len, sizeof(objType) - len, "%s(s:%d,r:%d)", TMSG_INFO(pAction->msgType),
×
2356
                    pAction->msgSent, pAction->msgReceived);
×
UNCOV
2357
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2358
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
UNCOV
2359
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2360
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2361

2362
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2363
    len = 0;
×
UNCOV
2364
    len += snprintf(result + len, sizeof(result) - len, "errCode:0x%x(%s)", pAction->errCode & 0xFFFF,
×
2365
                    tstrerror(pAction->errCode));
2366
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2367
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
2368
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2369
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2370

2371
    char target[TSDB_TRANS_TARGET_LEN] = {0};
×
UNCOV
2372
    len = 0;
×
2373
    SEpSet epset = pAction->epSet;
×
2374
    if (epset.numOfEps > 0) {
×
2375
      for (int32_t i = 0; i < epset.numOfEps; ++i) {
×
UNCOV
2376
        len += snprintf(target + len, sizeof(target) - len, "ep:%d-%s:%u,", i, epset.eps[i].fqdn, epset.eps[i].port);
×
2377
      }
2378
      len += snprintf(target + len, sizeof(target) - len, "(%d:%d) ", epset.numOfEps, epset.inUse);
×
2379
    }
2380
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
UNCOV
2381
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2382
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2383
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2384

2385
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2386
    len = 0;
×
2387
    char bufStart[40] = {0};
×
UNCOV
2388
    if (pAction->startTime > 0) (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
×
2389
    char bufEnd[40] = {0};
×
UNCOV
2390
    if (pAction->endTime > 0) (void)formatTimestamp(bufEnd, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
×
2391
    len += snprintf(detail + len, sizeof(detail) - len, "startTime:%s, endTime:%s, ", bufStart, bufEnd);
×
2392
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
2393
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2394
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
UNCOV
2395
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2396

2397
  } else {
2398
    int32_t len = 0;
×
2399

2400
    char objType[TSDB_TRANS_OBJTYPE_LEN + 1] = {0};
×
2401
    if (pAction->pRaw->type == SDB_VGROUP) {
×
2402
      SSdbRow *pRow = mndVgroupActionDecode(pAction->pRaw);
×
2403
      SVgObj  *pVgroup = sdbGetRowObj(pRow);
×
2404
      len += snprintf(objType + len, sizeof(objType) - len, "%s(%d)", sdbTableName(pAction->pRaw->type), pVgroup->vgId);
×
2405
      taosMemoryFreeClear(pRow);
×
2406
    } else {
UNCOV
2407
      strcpy(objType, sdbTableName(pAction->pRaw->type));
×
2408
    }
2409
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
UNCOV
2410
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
2411
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2412
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)objTypeVStr, false), &lino, _OVER);
×
2413

2414
    char result[TSDB_TRANS_RESULT_LEN + 1] = {0};
×
2415
    len = 0;
×
2416
    len += snprintf(result + len, sizeof(result) - len, "rawWritten:%d", pAction->rawWritten);
×
UNCOV
2417
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2418
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
UNCOV
2419
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2420
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)resultVStr, false), &lino, _OVER);
×
2421

2422
    char target[TSDB_TRANS_TARGET_LEN] = "";
×
2423
    char targetVStr[TSDB_TRANS_TARGET_LEN + VARSTR_HEADER_SIZE] = {0};
×
UNCOV
2424
    STR_WITH_MAXSIZE_TO_VARSTR(targetVStr, target, pShow->pMeta->pSchemas[cols].bytes);
×
2425
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2426
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)targetVStr, false), &lino, _OVER);
×
2427

2428
    char detail[TSDB_TRANS_DETAIL_LEN] = {0};
×
2429
    len = 0;
×
2430
    len += snprintf(detail + len, sizeof(detail) - len, "sdbStatus:%s", sdbStatusName(pAction->pRaw->status));
×
2431
    char detailVStr[TSDB_TRANS_DETAIL_LEN + VARSTR_HEADER_SIZE] = {0};
×
UNCOV
2432
    STR_WITH_MAXSIZE_TO_VARSTR(detailVStr, detail, pShow->pMeta->pSchemas[cols].bytes);
×
2433
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
2434
    TAOS_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)detailVStr, false), &lino, _OVER);
×
2435
  }
2436

2437
_OVER:
×
UNCOV
2438
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2439
}
2440

2441
static SArray *mndTransGetAction(STrans *pTrans, ETrnStage stage) {
×
2442
  if (stage == TRN_STAGE_PREPARE) {
×
2443
    return pTrans->prepareActions;
×
2444
  }
2445
  if (stage == TRN_STAGE_REDO_ACTION) {
×
UNCOV
2446
    return pTrans->redoActions;
×
2447
  }
2448
  if (stage == TRN_STAGE_COMMIT_ACTION) {
×
2449
    return pTrans->commitActions;
×
2450
  }
UNCOV
2451
  if (stage == TRN_STAGE_UNDO_ACTION) {
×
2452
    return pTrans->undoActions;
×
2453
  }
2454
  return NULL;
×
2455
}
2456

2457
typedef struct STransDetailIter {
2458
  void     *pIter;
2459
  STrans   *pTrans;
2460
  ETrnStage stage;
2461
  int32_t   num;
2462
} STransDetailIter;
2463

UNCOV
2464
static void mndTransShowActions(SSdb *pSdb, STransDetailIter *pShowIter, SShowObj *pShow, SSDataBlock *pBlock,
×
2465
                                int32_t rows, int32_t *numOfRows, SArray *pActions, int32_t end, int32_t start) {
UNCOV
2466
  int32_t actionNum = taosArrayGetSize(pActions);
×
UNCOV
2467
  mInfo("stage:%s, Actions num:%d", mndTransStr(pShowIter->stage), actionNum);
×
2468

UNCOV
2469
  for (int32_t i = start; i < actionNum; ++i) {
×
UNCOV
2470
    STransAction *pAction = taosArrayGet(pShowIter->pTrans->redoActions, i);
×
UNCOV
2471
    mndShowTransAction(pShow, pBlock, pAction, pShowIter->pTrans->id, pShowIter->pTrans->lastAction, rows, *numOfRows);
×
UNCOV
2472
    (*numOfRows)++;
×
UNCOV
2473
    if (*numOfRows >= rows) break;
×
2474
  }
2475

UNCOV
2476
  if (*numOfRows == end) {
×
2477
    sdbRelease(pSdb, pShowIter->pTrans);
×
2478
    pShowIter->pTrans = NULL;
×
UNCOV
2479
    pShowIter->num = 0;
×
2480
  } else {
2481
    pShowIter->pTrans = pShowIter->pTrans;
×
2482
    pShowIter->stage = pShowIter->pTrans->stage;
×
2483
    pShowIter->num += (*numOfRows);
×
2484
  }
UNCOV
2485
}
×
2486

2487
static int32_t mndRetrieveTransDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
×
2488
  SMnode *pMnode = pReq->info.node;
×
2489
  SSdb   *pSdb = pMnode->pSdb;
×
2490
  int32_t numOfRows = 0;
×
2491

2492
  int32_t code = 0;
×
2493
  int32_t lino = 0;
×
2494

UNCOV
2495
  mInfo("start to mndRetrieveTransDetail, rows:%d, pShow->numOfRows:%d, pShow->pIter:%p", rows, pShow->numOfRows,
×
2496
        pShow->pIter);
2497

2498
  if (pShow->pIter == NULL) {
×
2499
    pShow->pIter = taosMemoryMalloc(sizeof(STransDetailIter));
×
2500
    if (pShow->pIter == NULL) {
×
2501
      mError("failed to malloc for pShow->pIter");
×
UNCOV
2502
      return 0;
×
2503
    }
2504
    memset(pShow->pIter, 0, sizeof(STransDetailIter));
×
2505
  }
2506

UNCOV
2507
  STransDetailIter *pShowIter = (STransDetailIter *)pShow->pIter;
×
2508

2509
  while (numOfRows < rows) {
×
2510
    if (pShowIter->pTrans == NULL) {
×
2511
      pShowIter->pIter = sdbFetch(pSdb, SDB_TRANS, pShowIter->pIter, (void **)&(pShowIter->pTrans));
×
2512
      mDebug("retrieve trans detail from fetch, pShow->pIter:%p, pTrans:%p", pShowIter->pIter, pShowIter->pTrans);
×
2513
      if (pShowIter->pIter == NULL) break;
×
UNCOV
2514
      mInfo("retrieve trans detail from fetch, id:%d, trans stage:%d, IterNum:%d", pShowIter->pTrans->id,
×
2515
            pShowIter->pTrans->stage, pShowIter->num);
2516

UNCOV
2517
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->pTrans->stage);
×
2518

UNCOV
2519
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions, taosArrayGetSize(pActions), 0);
×
2520
      break;
×
2521
    } else {
2522
      mInfo("retrieve trans detail from iter, id:%d, iterStage:%d, IterNum:%d", pShowIter->pTrans->id, pShowIter->stage,
×
2523
            pShowIter->num);
2524
      SArray *pActions = mndTransGetAction(pShowIter->pTrans, pShowIter->stage);
×
2525

UNCOV
2526
      mndTransShowActions(pSdb, pShowIter, pShow, pBlock, rows, &numOfRows, pActions,
×
UNCOV
2527
                          taosArrayGetSize(pActions) - pShowIter->num, pShowIter->num);
×
2528
      break;
×
2529
    }
2530
  }
2531

UNCOV
2532
_OVER:
×
2533
  pShow->numOfRows += numOfRows;
×
2534

2535
  if (code != 0) {
×
UNCOV
2536
    mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
×
2537
  } else {
2538
    mInfo("retrieve trans detail, numOfRows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows)
×
2539
  }
UNCOV
2540
  if (numOfRows == 0) {
×
UNCOV
2541
    taosMemoryFree(pShow->pIter);
×
UNCOV
2542
    pShow->pIter = NULL;
×
2543
  }
2544
  return numOfRows;
×
2545
}
2546

2547
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
×
UNCOV
2548
  SSdb *pSdb = pMnode->pSdb;
×
2549
  sdbCancelFetchByType(pSdb, pIter, SDB_TRANS);
×
UNCOV
2550
}
×
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