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

taosdata / TDengine / #3845

11 Apr 2025 06:14AM UTC coverage: 62.585% (+18.7%) from 43.852%
#3845

push

travis-ci

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

merge: from main to 3.0 branch

154461 of 315259 branches covered (48.99%)

Branch coverage included in aggregate %.

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

3223 existing lines in 85 files now uncovered.

240073 of 315138 relevant lines covered (76.18%)

18471065.7 hits per line

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

62.43
/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; }
503,445✔
58

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

66
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
519,714✔
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,764✔
79
  SSdbTable table = {
1,764✔
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,764✔
90
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
1,764✔
91

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

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

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

104
  for (int32_t i = 0; i < actionNum; ++i) {
1,814,795✔
105
    STransAction *pAction = taosArrayGet(pArray, i);
1,211,767✔
106
    if (pAction->actionType == TRANS_ACTION_RAW) {
1,211,767✔
107
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
742,324✔
108
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
469,443!
109
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
469,443✔
110
    } else {
111
      // empty
112
    }
113
    rawDataLen += sizeof(int8_t);
1,211,767✔
114
  }
115

116
  return rawDataLen;
603,028✔
117
}
118

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

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

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

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

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

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

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

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

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

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

207
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
150,757!
208
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
150,757!
209
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
150,757!
210
  if (pTrans->param != NULL) {
150,757!
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)
150,757!
215

216
  int32_t arbGroupNum = taosHashGetSize(pTrans->arbGroupIds);
150,757✔
217
  SDB_SET_INT32(pRaw, dataPos, arbGroupNum, _OVER)
150,757!
218
  void *pIter = NULL;
150,757✔
219
  pIter = taosHashIterate(pTrans->arbGroupIds, NULL);
150,757✔
220
  while (pIter) {
150,793✔
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) {
150,757!
227
    SDB_SET_INT8(pRaw, dataPos, pTrans->ableToBeKilled, _OVER)
150,757!
228
    SDB_SET_INT32(pRaw, dataPos, pTrans->killMode, _OVER)
150,757!
229
  }
230

231
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
150,757!
232
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
150,757!
233

234
  terrno = 0;
150,757✔
235

236
_OVER:
150,757✔
237
  if (terrno != 0) {
150,757!
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);
150,757✔
245
  return pRaw;
150,757✔
246
}
247

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

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

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

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

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

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

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

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

332
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
285,962!
333

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

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

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

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

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

380
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
285,962!
381
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
285,962!
382
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
285,962!
383
  if (pTrans->paramLen != 0) {
285,962!
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);
285,962!
390

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

393
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
285,962!
394
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
286,022✔
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;
285,962✔
401
  int8_t killMode = 0;
285,962✔
402
  SDB_GET_INT8(pRaw, dataPos, &ableKill, _OVER)
285,962!
403
  SDB_GET_INT8(pRaw, dataPos, &killMode, _OVER)
285,962!
404
  pTrans->ableToBeKilled = ableKill;
285,962✔
405
  pTrans->killMode = killMode;
285,962✔
406

407
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
285,962!
408

409
  terrno = 0;
285,962✔
410

411
_OVER:
285,962✔
412
  if (terrno != 0 && pTrans != NULL) {
285,962!
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) {
285,962!
420
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
285,962✔
421
  }
422
  return pRow;
285,962✔
423
}
424

425
static const char *mndTransStr(ETrnStage stage) {
2,276,917✔
426
  switch (stage) {
2,276,917!
427
    case TRN_STAGE_PREPARE:
143,604✔
428
      return "prepare";
143,604✔
429
    case TRN_STAGE_REDO_ACTION:
995,169✔
430
      return "redoAction";
995,169✔
431
    case TRN_STAGE_ROLLBACK:
10✔
432
      return "rollback";
10✔
433
    case TRN_STAGE_UNDO_ACTION:
44,531✔
434
      return "undoAction";
44,531✔
435
    case TRN_STAGE_COMMIT:
240,199✔
436
      return "commit";
240,199✔
437
    case TRN_STAGE_COMMIT_ACTION:
499,140✔
438
      return "commitAction";
499,140✔
439
    case TRN_STAGE_FINISH:
354,254✔
440
      return "finished";
354,254✔
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) {
673,252✔
460
  if (pAction != NULL) {
673,252✔
461
    if (pAction->errCode != TSDB_CODE_ACTION_IN_PROGRESS) {
551,505✔
462
      pTrans->lastAction = pAction->id;
355,799✔
463
      pTrans->lastMsgType = pAction->msgType;
355,799✔
464
      pTrans->lastEpset = pAction->epSet;
355,799✔
465
      pTrans->lastErrorNo = pAction->errCode;
355,799✔
466
    }
467
  } else {
468
    pTrans->lastAction = 0;
121,747✔
469
    pTrans->lastMsgType = 0;
121,747✔
470
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
121,747✔
471
    pTrans->lastErrorNo = 0;
121,747✔
472
  }
473
}
673,252✔
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,028✔
484
  switch (ftype) {
2,028!
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,014✔
490
      return mndRebCntInc;
1,014✔
491
    case TRANS_STOP_FUNC_MQ_REB:
1,014✔
492
      return mndRebCntDec;
1,014✔
493
    default:
×
494
      return NULL;
×
495
  }
496
}
497

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

502
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
53,750✔
503

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

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

517
  if (pTrans->stage == TRN_STAGE_ROLLBACK) {
53,750!
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) {
53,750!
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;
53,750✔
528
}
529

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

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

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

579
  mndTransDropData(pTrans);
169,833✔
580
  return 0;
169,833✔
581
}
582

583
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
250,332✔
584
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
776,814✔
585
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
526,482✔
586
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
526,482✔
587
    pOldAction->rawWritten = pNewAction->rawWritten;
526,482✔
588
    pOldAction->msgSent = pNewAction->msgSent;
526,482✔
589
    pOldAction->msgReceived = pNewAction->msgReceived;
526,482✔
590
    pOldAction->errCode = pNewAction->errCode;
526,482✔
591
  }
592
}
250,332✔
593

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

598
  if (pOld->createdTime != pNew->createdTime) {
62,583!
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);
62,583✔
608
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
62,583✔
609
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
62,583✔
610
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
62,583✔
611
  pOld->stage = pNew->stage;
62,583✔
612
  pOld->actionPos = pNew->actionPos;
62,583✔
613

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

619
  if (pOld->stage == TRN_STAGE_ROLLBACK) {
62,583✔
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) {
62,583✔
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;
62,583✔
630
}
631

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

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

646
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
44,818✔
647
                       const char *opername) {
648
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
44,818!
649
  if (pTrans == NULL) {
44,818!
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) {
44,818!
656
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
44,818✔
657
  }
658

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

679
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
44,818!
680
      pTrans->pRpcArray == NULL) {
44,818!
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) {
44,818✔
688
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
55,348!
689
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
690
      return NULL;
×
691
    }
692
    pTrans->originRpcType = pReq->msgType;
27,674✔
693
  }
694

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

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

701
static void mndTransDropActions(SArray *pArray) {
1,323,120✔
702
  int32_t size = taosArrayGetSize(pArray);
1,323,120✔
703
  for (int32_t i = 0; i < size; ++i) {
3,925,521✔
704
    STransAction *pAction = taosArrayGet(pArray, i);
2,602,401✔
705
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,602,401✔
706
      taosMemoryFreeClear(pAction->pRaw);
1,572,873!
707
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
1,029,528!
708
      taosMemoryFreeClear(pAction->pCont);
1,029,528!
709
    } else {
710
      // nothing
711
    }
712
  }
713

714
  taosArrayDestroy(pArray);
1,323,120✔
715
}
1,323,120✔
716

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

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

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

733
  return 0;
347,405✔
734
}
735

736
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
1,652✔
737
  STransAction action = {
1,652✔
738
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
1,652✔
739
  return mndTransAppendAction(pTrans->redoActions, &action);
1,652✔
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) {
12,951✔
748
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
12,951✔
749
  return mndTransAppendAction(pTrans->undoActions, &action);
12,951✔
750
}
751

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

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

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

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

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

781
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
1,002✔
782
  pTrans->startFunc = startFunc;
1,002✔
783
  pTrans->stopFunc = stopFunc;
1,002✔
784
  pTrans->param = param;
1,002✔
785
  pTrans->paramLen = paramLen;
1,002✔
786
}
1,002✔
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) {
29,143✔
820
  if (dbname != NULL) {
29,143!
821
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
29,143✔
822
  }
823
  if (stbname != NULL) {
29,143✔
824
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
21,974✔
825
  }
826
}
29,143✔
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,830✔
835

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

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

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

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

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

849
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
96,917✔
850
  int32_t  code = 0;
96,917✔
851
  SSdbRaw *pRaw = mndTransEncode(pTrans);
96,917✔
852
  if (pRaw == NULL) {
96,917!
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));
96,917!
859

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

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

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

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

887
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
25,710✔
888
  if (conflict) {
25,710✔
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,
348!
890
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
891
    *globalConflict = true;
348✔
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,362!
894
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
895
  }
896
}
25,710✔
897

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

903
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
127,625✔
904

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

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

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

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

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

933
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
23,314!
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,314✔
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,314✔
960
  }
961

962
  return conflict;
84,546✔
963
}
964

965
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
127,625✔
966
  int32_t code = 0;
127,625✔
967
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
127,625✔
968
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
75,175!
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)) {
127,625✔
976
    code = TSDB_CODE_MND_TRANS_CONFLICT;
381✔
977
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
381!
978
    TAOS_RETURN(code);
381✔
979
  }
980

981
  TAOS_RETURN(code);
127,244✔
982
}
983

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

990
  while (1) {
4✔
991
    bool thisConflict = false;
295✔
992
    pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
295✔
993
    if (pIter == NULL) break;
295✔
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) {
291✔
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);
287✔
1020
}
1021

1022
static bool mndTransActionsOfSameType(SArray *pActions) {
101,238✔
1023
  int32_t size = taosArrayGetSize(pActions);
101,238✔
1024
  ETrnAct lastActType = TRANS_ACTION_NULL;
101,238✔
1025
  bool    same = true;
101,238✔
1026
  for (int32_t i = 0; i < size; ++i) {
392,474✔
1027
    STransAction *pAction = taosArrayGet(pActions, i);
291,236✔
1028
    if (i > 0) {
291,236✔
1029
      if (lastActType != pAction->actionType) {
216,048!
1030
        same = false;
×
1031
        break;
×
1032
      }
1033
    }
1034
    lastActType = pAction->actionType;
291,236✔
1035
  }
1036
  return same;
101,238✔
1037
}
1038

1039
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
44,404✔
1040
  int32_t code = 0;
44,404✔
1041
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
44,404✔
1042
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
41,648!
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) {
41,648✔
1049
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
15,186!
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);
44,404✔
1058
}
1059

1060
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
44,404✔
1061
  int32_t code = 0;
44,404✔
1062
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
44,404!
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) {
44,404!
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);
44,404✔
1074
}
1075

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

1082
  mInfo("trans:%d, action list:", pTrans->id);
44,404!
1083
  int32_t index = 0;
44,404✔
1084
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->prepareActions); ++i, ++index) {
64,660✔
1085
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, i);
20,256✔
1086
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
20,256!
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) {
136,299✔
1091
    STransAction *pAction = taosArrayGet(pTrans->redoActions, i);
91,895✔
1092
    mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
91,895!
1093
          mndTransStr(pAction->stage), pAction->id, TMSG_INFO(pAction->msgType));;
1094
  }
1095

1096
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
234,777✔
1097
    STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
190,373✔
1098
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
190,373!
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) {
88,849✔
1103
    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
44,445✔
1104
    if(pAction->actionType == TRANS_ACTION_MSG){
44,445✔
1105
      mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
31,494!
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,
12,951!
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));
44,404!
1116

1117
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
44,404!
1118

1119
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
44,404!
1120

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

1131
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
44,395✔
1132
  if (pNew == NULL) {
44,395!
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;
44,395✔
1140
  pNew->rpcRsp = pTrans->rpcRsp;
44,395✔
1141
  pNew->rpcRspLen = pTrans->rpcRspLen;
44,395✔
1142
  pNew->mTraceId = pTrans->mTraceId;
44,395✔
1143
  pTrans->pRpcArray = NULL;
44,395✔
1144
  pTrans->rpcRsp = NULL;
44,395✔
1145
  pTrans->rpcRspLen = 0;
44,395✔
1146

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

1154
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
44,401✔
1155
  int32_t code = 0;
44,401✔
1156
  mInfo("trans:%d, commit transaction", pTrans->id);
44,401!
1157
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
44,401✔
1158
    mError("trans:%d, failed to commit since %s", pTrans->id, tstrerror(code));
2!
1159
    TAOS_RETURN(code);
2✔
1160
  }
1161
  mInfo("trans:%d, commit finished", pTrans->id);
44,399!
1162
  TAOS_RETURN(code);
44,399✔
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!
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!
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) {
260,973✔
1188
  bool    sendRsp = false;
260,973✔
1189
  int32_t code = pTrans->code;
260,973✔
1190

1191
  if (pTrans->stage == TRN_STAGE_FINISH) {
260,973✔
1192
    sendRsp = true;
97,901✔
1193
  }
1194

1195
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
260,973✔
1196
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
75,062!
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) {
185,911✔
1202
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
123,671!
1203
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1204
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1205
      } else {
1206
        if (pTrans->failedTimes > 6) sendRsp = true;
123,671✔
1207
      }
1208
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
123,671✔
1209
    }
1210
  }
1211

1212
  if (!sendRsp) {
260,973✔
1213
    return;
163,058✔
1214
  } else {
1215
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
97,915!
1216
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1217
  }
1218

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

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

1237
      if (i != 0 && code == 0) {
25,729!
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,
25,729!
1241
            mndTransStr(pTrans->stage), pInfo->ahandle);
1242

1243
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
25,729✔
1244

1245
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
25,729✔
1246
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
3,716!
1247
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
3,716✔
1248
        if (pDb != NULL) {
3,716!
1249
          for (int32_t j = 0; j < 12; j++) {
4,516✔
1250
            bool ready = mndIsDbReady(pMnode, pDb);
4,512✔
1251
            if (!ready) {
4,512✔
1252
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
800!
1253
              taosMsleep(1000);
800✔
1254
            } else {
1255
              break;
3,712✔
1256
            }
1257
          }
1258
        }
1259
        mndReleaseDb(pMnode, pDb);
3,716✔
1260
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
22,013✔
1261
        void   *pCont = NULL;
6,959✔
1262
        int32_t contLen = 0;
6,959✔
1263
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
6,959✔
1264
          mndTransSetRpcRsp(pTrans, pCont, contLen);
6,957✔
1265
        }
1266
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
15,054✔
1267
        void   *pCont = NULL;
473✔
1268
        int32_t contLen = 0;
473✔
1269
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
473!
1270
          mndTransSetRpcRsp(pTrans, pCont, contLen);
473✔
1271
        }
1272
      }
1273

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

1283
      tmsgSendRsp(&rspMsg);
25,729✔
1284

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

1293
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
95,417✔
1294
  int32_t code = 0;
95,417✔
1295
  SMnode *pMnode = pRsp->info.node;
95,417✔
1296
#ifndef TD_ASTRA_32
1297
  int64_t signature = (int64_t)(pRsp->info.ahandle);
95,417✔
1298
  int32_t transId = (int32_t)(signature >> 32);
95,417✔
1299
  int32_t action = (int32_t)((signature << 32) >> 32);
95,417✔
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);
95,417✔
1305
  if (pTrans == NULL) {
95,417!
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;
95,417✔
1313
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
95,417✔
1314
    pArray = pTrans->redoActions;
95,409✔
1315
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
8!
1316
    pArray = pTrans->undoActions;
8✔
1317
  } else {
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) {
95,417!
1323
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
×
1324
    goto _OVER;
×
1325
  }
1326

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

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

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

1342
    mInfo("trans:%d, %s:%d response is received, received code:0x%x(%s), accept:0x%x(%s) retry:0x%x(%s)", transId,
95,417!
1343
          mndTransStr(pAction->stage), action, pRsp->code, tstrerror(pRsp->code), pAction->acceptableCode,
1344
          tstrerror(pAction->acceptableCode), pAction->retryCode, tstrerror(pAction->retryCode));
1345
  } else {
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);
95,417!
1350
  mndTransExecute(pMnode, pTrans, true);
95,417✔
1351

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

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

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

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

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

1386
// execute in sync context
1387
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
430,047✔
1388
  if (pAction->rawWritten) return 0;
430,047✔
1389
  if (topHalf) {
238,137✔
1390
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
7✔
1391
  }
1392

1393
  if (pAction->pRaw->type >= SDB_MAX) {
238,130!
1394
    pAction->rawWritten = true;
×
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);
238,130✔
1402
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
238,130!
1403
    pAction->rawWritten = true;
238,130✔
1404
    pAction->errCode = 0;
238,130✔
1405
    code = 0;
238,130✔
1406
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
238,130!
1407
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1408

1409
    mndSetTransLastAction(pTrans, pAction);
238,130✔
1410
  } else {
1411
    pAction->errCode = (terrno != 0) ? terrno : code;
×
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);
238,130✔
1418
}
1419

1420
// execute in trans context
1421
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
664,075✔
1422
                                     bool notSend) {
1423
  if (pAction->msgSent) return 0;
664,075✔
1424
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
133,118✔
1425
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
37,416✔
1426
  }
1427

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

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

1438
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
95,541✔
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);
95,541✔
1446
  if (rpcMsg.pCont == NULL) {
95,541!
1447
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1448
    return -1;
×
1449
  }
1450
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
95,541✔
1451
  rpcMsg.info.notFreeAhandle = 1;
95,541✔
1452

1453
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
95,541✔
1454

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

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

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

1480
    mndSetTransLastAction(pTrans, pAction);
19✔
1481
  }
1482

1483
  TAOS_RETURN(code);
95,541✔
1484
}
1485

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

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

1496
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
1,094,122✔
1497
                                        bool notSend) {
1498
  if (pAction->actionType == TRANS_ACTION_RAW) {
1,094,122✔
1499
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
430,047✔
1500
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
664,075!
1501
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf, notSend);
664,075✔
1502
  } else {
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) {
237,687✔
1508
  int32_t numOfActions = taosArrayGetSize(pArray);
237,687✔
1509
  int32_t code = 0;
237,687✔
1510

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

1521
  return code;
237,687✔
1522
}
1523

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

1529
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf, notSend)) != 0) {
237,687✔
1530
    return code;
30,102✔
1531
  }
1532

1533
  int32_t       numOfExecuted = 0;
207,585✔
1534
  int32_t       errCode = 0;
207,585✔
1535
  STransAction *pErrAction = NULL;
207,585✔
1536
  for (int32_t action = 0; action < numOfActions; ++action) {
1,209,628✔
1537
    STransAction *pAction = taosArrayGet(pArray, action);
1,002,043✔
1538
    if (pAction->msgReceived || pAction->rawWritten) {
1,002,043✔
1539
      numOfExecuted++;
702,116✔
1540
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
702,116✔
1541
        errCode = pAction->errCode;
27✔
1542
        pErrAction = pAction;
27✔
1543
      }
1544
    } else {
1545
      pErrAction = pAction;
299,927✔
1546
    }
1547
  }
1548

1549
  mndSetTransLastAction(pTrans, pErrAction);
207,585✔
1550

1551
  if (numOfExecuted == numOfActions) {
207,585✔
1552
    if (errCode == 0) {
121,749✔
1553
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
121,747!
1554
      return 0;
121,747✔
1555
    } else {
1556
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
2!
1557
      mndTransResetActions(pMnode, pTrans, pArray);
2✔
1558
      terrno = errCode;
2✔
1559
      return errCode;
2✔
1560
    }
1561
  } else {
1562
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
85,836!
1563

1564
    for (int32_t action = 0; action < numOfActions; ++action) {
602,857✔
1565
      STransAction *pAction = taosArrayGet(pArray, action);
517,021✔
1566
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
517,021✔
1567
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1568
             pAction->acceptableCode, pAction->retryCode);
1569
      if (pAction->msgSent) {
517,021✔
1570
        if (pAction->msgReceived) {
516,860✔
1571
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
217,094✔
1572
            mndTransResetAction(pMnode, pTrans, pAction);
25✔
1573
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
25!
1574
          }
1575
        }
1576
      }
1577
    }
1578
    return TSDB_CODE_ACTION_IN_PROGRESS;
85,836✔
1579
  }
1580
}
1581

1582
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf, bool notSend) {
178,102✔
1583
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf, notSend);
178,102✔
1584
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
178,102✔
1585
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
21!
1586
           mndStrExecutionContext(topHalf));
1587
  }
1588
  return code;
178,102✔
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!
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) {
97,897✔
1601
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf, true);
97,897✔
1602
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
97,897!
1603
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1604
           mndStrExecutionContext(topHalf));
1605
  }
1606
  return code;
97,897✔
1607
}
1608

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

1614
  if (pTrans->actionPos >= numOfActions) {
29,355✔
1615
    return code;
3,040✔
1616
  }
1617

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

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

1624
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d, msgSent:%d, msgReceived:%d", 
36,579!
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,579✔
1629
    if (code == 0) {
36,579✔
1630
      if (pAction->msgSent) {
29,239✔
1631
        if (pAction->msgReceived) {
25,870✔
1632
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
11,507✔
1633
            code = pAction->errCode;
4,926✔
1634
            mndTransResetAction(pMnode, pTrans, pAction);
4,926✔
1635
          } else {
1636
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
6,581!
1637
          }
1638
        } else {
1639
          code = TSDB_CODE_ACTION_IN_PROGRESS;
14,363✔
1640
        }
1641
      } else if (pAction->rawWritten) {
3,369!
1642
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
3,369!
1643
          code = pAction->errCode;
×
1644
        } else {
1645
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
3,369!
1646
        }
1647
      } else {
1648
      }
1649
    }
1650

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

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

1664
    if (code == 0) {
27,397✔
1665
      pTrans->code = 0;
8,108✔
1666
      pTrans->actionPos++;
8,108✔
1667
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
8,108!
1668
            pAction->id);
1669
      (void)taosThreadMutexUnlock(&pTrans->mutex);
8,108✔
1670
      code = mndTransSync(pMnode, pTrans);
8,108✔
1671
      (void)taosThreadMutexLock(&pTrans->mutex);
8,108✔
1672
      if (code != 0) {
8,108!
1673
        pTrans->actionPos--;
×
1674
        pTrans->code = terrno;
×
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());
1677
        break;
×
1678
      }
1679
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
19,289✔
1680
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
14,363!
1681
      break;
14,363✔
1682
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,926✔
1683
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
28✔
1684
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,909!
1685
            code, tstrerror(code));
1686
      pTrans->lastErrorNo = code;
4,909✔
1687
      taosMsleep(300);
4,909✔
1688
      action--;
4,909✔
1689
      continue;
4,909✔
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;
26,315✔
1701
}
1702

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

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

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

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

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

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

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

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

1755
  if (pTrans->exec == TRN_EXEC_SERIAL) {
207,461✔
1756
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
29,359✔
1757
  } else {
1758
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf, notSend);
178,102✔
1759
  }
1760

1761
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
207,461!
1762
    pTrans->lastErrorNo = code;
×
1763
    pTrans->code = code;
×
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));
1768
    taosMsleep(100);
×
1769
    return true;
×
1770
  } else {
1771
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
207,461✔
1772
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
62,833!
1773
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1774
      return false;
62,833✔
1775
    }
1776
  }
1777
  terrno = code;
144,628✔
1778

1779
  if (code == 0) {
144,628✔
1780
    pTrans->code = 0;
44,401✔
1781
    pTrans->stage = TRN_STAGE_COMMIT;
44,401✔
1782
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
44,401!
1783
    continueExec = true;
44,401✔
1784
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
100,227!
1785
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
100,189!
1786
    continueExec = false;
100,189✔
1787
  } else {
1788
    pTrans->failedTimes++;
38✔
1789
    pTrans->code = terrno;
38✔
1790
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
38✔
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!
1794
          if (pTrans->failedTimes < 6) {
×
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);
1797
            taosMsleep(1000);
×
1798
            continueExec = true;
×
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);
36!
1811
      continueExec = false;
36✔
1812
    }
1813
  }
1814

1815
  return continueExec;
144,628✔
1816
}
1817

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

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

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

1837
  return continueExec;
44,401✔
1838
}
1839

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

1844
  if (code == 0) {
97,897!
1845
    pTrans->code = 0;
97,897✔
1846
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
97,897✔
1847
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
97,897!
1848
    continueExec = true;
97,897✔
1849
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
×
1850
    pTrans->code = 0;
×
1851
    pTrans->stage = TRN_STAGE_COMMIT;
×
1852
    mInfo("trans:%d, back to commit stage", pTrans->id);
×
1853
    continueExec = true;
×
1854
  } else {
1855
    pTrans->code = terrno;
×
1856
    pTrans->failedTimes++;
×
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;
97,897✔
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!
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 {
1885
    pTrans->failedTimes++;
×
1886
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
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 {
1904
    pTrans->failedTimes++;
×
1905
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
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 {
1924
    pTrans->failedTimes++;
×
1925
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1926
    continueExec = false;
×
1927
  }
1928

1929
  return continueExec;
2✔
1930
}
1931

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

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

1943
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
53,500✔
1944
  if (code != 0) {
53,500!
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,
53,500!
1949
        pTrans->failedTimes, pTrans->createdTime);
1950
  return continueExec;
53,500✔
1951
}
1952

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

1956
  while (continueExec) {
708,651✔
1957
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64, pTrans->id,
447,678!
1958
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1959
    pTrans->lastExecTime = taosGetTimestampMs();
447,678✔
1960
    switch (pTrans->stage) {
447,678!
1961
      case TRN_STAGE_PREPARE:
×
1962
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1963
        break;
×
1964
      case TRN_STAGE_REDO_ACTION:
207,461✔
1965
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf, notSend);
207,461✔
1966
        break;
207,461✔
1967
      case TRN_STAGE_COMMIT:
44,401✔
1968
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
44,401✔
1969
        break;
44,401✔
1970
      case TRN_STAGE_COMMIT_ACTION:
97,897✔
1971
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
97,897✔
1972
        break;
97,897✔
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:
97,901✔
1983
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
97,901✔
1984
        break;
97,901✔
1985
      default:
×
1986
        continueExec = false;
×
1987
        break;
×
1988
    }
1989
  }
1990

1991
  mndTransSendRpcRsp(pMnode, pTrans);
260,973✔
1992
}
260,973✔
1993

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

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

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

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

2022
  if(pTrans->ableToBeKilled == false){
×
2023
    return TSDB_CODE_MND_TRANS_NOT_ABLE_TO_kILLED;
×
2024
  }
2025
  
2026
  if(pTrans->killMode == TRN_KILL_MODE_SKIP){
×
2027
    for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
2028
      STransAction *pAction = taosArrayGet(pArray, i);
×
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;
×
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);
×
UNCOV
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;
×
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

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

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; }
357✔
2083

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

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

2100
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
39,090✔
2101

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

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

2125
  struct tm tm;
2126
  if (taosLocalTime(&tt, &tm, NULL, 0, NULL) == NULL) {
13,090!
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,090✔
2131

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

2140
  return buf;
13,090✔
2141
}
2142

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

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

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

2167
        char endStart[40] = {0};
6,545✔
2168
        (void)formatTimestamp(endStart, pAction->endTime, TSDB_TIME_PRECISION_MILLI);
6,545✔
2169
        len += snprintf(detail + len, sizeof(detail) - len,
13,090!
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,545✔
2173
                        pAction->msgReceived, bufStart, endStart);
6,545✔
2174

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

2184
        len += snprintf(detail + len, sizeof(detail) - len, ", errCode:0x%x(%s)\n", pAction->errCode & 0xFFFF,
6,545✔
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,545✔
2188
                        index, mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
1,545✔
2189
                        sdbStatusName(pAction->pRaw->status), pAction->rawWritten);
1,545✔
2190
      }
2191
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
8,090✔
2192
    }
2193
  }
2194

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

2205
    for (int32_t i = 0; i < taosArrayGetSize(pTrans->undoActions); ++i, ++index) {
×
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,
×
2214
                        mndTransStr(pAction->stage), pAction->id, sdbTableName(pAction->pRaw->type),
×
2215
                        sdbStatusName(pAction->pRaw->status));
×
2216
      }
2217
      mDebug("trans:%d, show tran action, detail:%s", pTrans->id, detail);
×
2218
    }
2219
  }
2220
}
176✔
2221

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

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

2235
    cols = 0;
176✔
2236

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

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

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

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

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

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

2264
    const char *killableStr = pTrans->ableToBeKilled ? "yes" : "no";
176!
2265
    char        killableVstr[10 + VARSTR_HEADER_SIZE] = {0};
176✔
2266
    STR_WITH_MAXSIZE_TO_VARSTR(killableVstr, killableStr, 10 + VARSTR_HEADER_SIZE);
176✔
2267
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
176✔
2268
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)killableVstr, false), pTrans, &lino, _OVER);
176!
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++);
176✔
2279
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
176!
2280
                        _OVER);
2281

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

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

2302
    mndTransLogAction(pTrans);
176✔
2303

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

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

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

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

2323
  char action[30 + 1] = {0};
×
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);
×
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));
×
2337
  return code;
×
2338
}
2339

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

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

2351
  if (pAction->actionType == TRANS_ACTION_MSG) {
×
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);
×
2357
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
2358
    STR_WITH_MAXSIZE_TO_VARSTR(objTypeVStr, objType, pShow->pMeta->pSchemas[cols].bytes);
×
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;
×
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};
×
2372
    len = 0;
×
2373
    SEpSet epset = pAction->epSet;
×
2374
    if (epset.numOfEps > 0) {
×
2375
      for (int32_t i = 0; i < epset.numOfEps; ++i) {
×
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};
×
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};
×
2388
    if (pAction->startTime > 0) (void)formatTimestamp(bufStart, pAction->startTime, TSDB_TIME_PRECISION_MILLI);
×
2389
    char bufEnd[40] = {0};
×
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++);
×
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 {
2407
      strcpy(objType, sdbTableName(pAction->pRaw->type));
×
2408
    }
2409
    char objTypeVStr[TSDB_TRANS_OBJTYPE_LEN + VARSTR_HEADER_SIZE] = {0};
×
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);
×
2417
    char resultVStr[TSDB_TRANS_RESULT_LEN + VARSTR_HEADER_SIZE] = {0};
×
2418
    STR_WITH_MAXSIZE_TO_VARSTR(resultVStr, result, pShow->pMeta->pSchemas[cols].bytes);
×
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};
×
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};
×
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:
×
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) {
×
2446
    return pTrans->redoActions;
×
2447
  }
2448
  if (stage == TRN_STAGE_COMMIT_ACTION) {
×
2449
    return pTrans->commitActions;
×
2450
  }
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

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) {
2466
  int32_t actionNum = taosArrayGetSize(pActions);
×
2467
  mInfo("stage:%s, Actions num:%d", mndTransStr(pShowIter->stage), actionNum);
×
2468

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

2476
  if (*numOfRows == end) {
×
2477
    sdbRelease(pSdb, pShowIter->pTrans);
×
2478
    pShowIter->pTrans = NULL;
×
2479
    pShowIter->num = 0;
×
2480
  } else {
2481
    pShowIter->pTrans = pShowIter->pTrans;
×
2482
    pShowIter->stage = pShowIter->pTrans->stage;
×
2483
    pShowIter->num += (*numOfRows);
×
2484
  }
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

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");
×
2502
      return 0;
×
2503
    }
2504
    memset(pShow->pIter, 0, sizeof(STransDetailIter));
×
2505
  }
2506

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;
×
2514
      mInfo("retrieve trans detail from fetch, id:%d, trans stage:%d, IterNum:%d", pShowIter->pTrans->id,
×
2515
            pShowIter->pTrans->stage, pShowIter->num);
2516

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

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

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

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

2535
  if (code != 0) {
×
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
  }
2540
  if (numOfRows == 0) {
×
2541
    taosMemoryFree(pShow->pIter);
×
2542
    pShow->pIter = NULL;
×
2543
  }
2544
  return numOfRows;
×
2545
}
2546

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