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

taosdata / TDengine / #3847

11 Apr 2025 06:14AM UTC coverage: 62.612% (+0.2%) from 62.398%
#3847

push

travis-ci

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

merge: from main to 3.0 branch

154571 of 315259 branches covered (49.03%)

Branch coverage included in aggregate %.

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

946 existing lines in 106 files now uncovered.

240135 of 315138 relevant lines covered (76.2%)

19768383.08 hits per line

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

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

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

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

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

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

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

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

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

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

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

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

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

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

116
  return rawDataLen;
603,884✔
117
}
118

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

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

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

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

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

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

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

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

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

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

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

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

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

234
  terrno = 0;
150,971✔
235

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

409
  terrno = 0;
285,932✔
410

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

425
static const char *mndTransStr(ETrnStage stage) {
2,274,028✔
426
  switch (stage) {
2,274,028!
427
    case TRN_STAGE_PREPARE:
143,487✔
428
      return "prepare";
143,487✔
429
    case TRN_STAGE_REDO_ACTION:
991,671✔
430
      return "redoAction";
991,671✔
431
    case TRN_STAGE_ROLLBACK:
10✔
432
      return "rollback";
10✔
433
    case TRN_STAGE_UNDO_ACTION:
44,512✔
434
      return "undoAction";
44,512✔
435
    case TRN_STAGE_COMMIT:
240,379✔
436
      return "commit";
240,379✔
437
    case TRN_STAGE_COMMIT_ACTION:
499,650✔
438
      return "commitAction";
499,650✔
439
    case TRN_STAGE_FINISH:
354,309✔
440
      return "finished";
354,309✔
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) {
672,134✔
460
  if (pAction != NULL) {
672,134✔
461
    if (pAction->errCode != TSDB_CODE_ACTION_IN_PROGRESS) {
550,310✔
462
      pTrans->lastAction = pAction->id;
355,367✔
463
      pTrans->lastMsgType = pAction->msgType;
355,367✔
464
      pTrans->lastEpset = pAction->epSet;
355,367✔
465
      pTrans->lastErrorNo = pAction->errCode;
355,367✔
466
    }
467
  } else {
468
    pTrans->lastAction = 0;
121,824✔
469
    pTrans->lastMsgType = 0;
121,824✔
470
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
121,824✔
471
    pTrans->lastErrorNo = 0;
121,824✔
472
  }
473
}
672,134✔
474

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

733
  return 0;
347,696✔
734
}
735

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

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

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

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

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

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

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

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

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

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

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

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

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

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

819
void mndTransSetDbName(STrans *pTrans, const char *dbname, const char *stbname) {
29,153✔
820
  if (dbname != NULL) {
29,153!
821
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
29,153✔
822
  }
823
  if (stbname != NULL) {
29,153✔
824
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
22,018✔
825
  }
826
}
29,153✔
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,856✔
835

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

912
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
22,465!
913

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

922
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
22,465✔
923
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
19,059✔
924
      if (pTrans->conflict == TRN_CONFLICT_DB) {
19,059✔
925
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
1,467✔
926
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
1,467✔
927
      }
928
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
19,059✔
929
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
16,913✔
930
      }
931
    }
932

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

962
  return conflict;
84,380✔
963
}
964

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

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

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

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

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

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

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

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

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

1096
  for (int32_t i = 0; i < taosArrayGetSize(pTrans->commitActions); ++i, ++index) {
235,137✔
1097
    STransAction *pAction = taosArrayGet(pTrans->commitActions, i);
190,655✔
1098
    mInfo("trans:%d, action:%d, %s:%d sdbType:%s, sdbStatus:%s", pTrans->id, index,
190,655!
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,908✔
1103
    STransAction *pAction = taosArrayGet(pTrans->undoActions, i);
44,426✔
1104
    if(pAction->actionType == TRANS_ACTION_MSG){
44,426✔
1105
      mInfo("trans:%d, action:%d, %s:%d msgType:%s", pTrans->id, index,
31,471!
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,955!
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,482!
1116

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

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

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

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

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

1154
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
44,478✔
1155
  int32_t code = 0;
44,478✔
1156
  mInfo("trans:%d, commit transaction", pTrans->id);
44,478!
1157
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
44,478✔
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,476!
1162
  TAOS_RETURN(code);
44,476✔
1163
}
1164

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1293
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
95,080✔
1294
  int32_t code = 0;
95,080✔
1295
  SMnode *pMnode = pRsp->info.node;
95,080✔
1296
#ifndef TD_ASTRA_32
1297
  int64_t signature = (int64_t)(pRsp->info.ahandle);
95,080✔
1298
  int32_t transId = (int32_t)(signature >> 32);
95,080✔
1299
  int32_t action = (int32_t)((signature << 32) >> 32);
95,080✔
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,080✔
1305
  if (pTrans == NULL) {
95,080!
UNCOV
1306
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1307
    if (terrno != 0) code = terrno;
×
1308
    mError("trans:%d, failed to get transId from vnode rsp since %s", transId, tstrerror(code));
×
1309
    goto _OVER;
×
1310
  }
1311

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

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

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

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

1339
    // pTrans->lastErrorNo = pRsp->code;
1340
    mndSetTransLastAction(pTrans, pAction);
95,080✔
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,080!
1343
          mndTransStr(pAction->stage), action, pRsp->code, tstrerror(pRsp->code), pAction->acceptableCode,
1344
          tstrerror(pAction->acceptableCode), pAction->retryCode, tstrerror(pAction->retryCode));
1345
  } else {
UNCOV
1346
    mInfo("trans:%d, invalid action, index:%d, code:0x%x", transId, action, pRsp->code);
×
1347
  }
1348

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

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

1357
static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
4,681✔
1358
  pAction->rawWritten = 0;
4,681✔
1359
  pAction->msgSent = 0;
4,681✔
1360
  pAction->msgReceived = 0;
4,681✔
1361
  if (pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR || pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR ||
4,681!
1362
      pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
4,681✔
1363
    pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
14✔
1364
    mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
14!
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,667!
1368
  }
1369
  pAction->errCode = 0;
4,681✔
1370
}
4,681✔
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,392✔
1388
  if (pAction->rawWritten) return 0;
430,392✔
1389
  if (topHalf) {
238,197✔
1390
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
13✔
1391
  }
1392

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

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

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

1417
  TAOS_RETURN(code);
238,184✔
1418
}
1419

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

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

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

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

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

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

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

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

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

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

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

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

1496
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf,
1,091,527✔
1497
                                        bool notSend) {
1498
  if (pAction->actionType == TRANS_ACTION_RAW) {
1,091,527✔
1499
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
430,392✔
1500
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
661,135!
1501
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf, notSend);
661,135✔
1502
  } else {
UNCOV
1503
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1504
  }
1505
}
1506

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

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

1521
  return code;
237,461✔
1522
}
1523

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

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

1533
  int32_t       numOfExecuted = 0;
207,469✔
1534
  int32_t       errCode = 0;
207,469✔
1535
  STransAction *pErrAction = NULL;
207,469✔
1536
  for (int32_t action = 0; action < numOfActions; ++action) {
1,207,491✔
1537
    STransAction *pAction = taosArrayGet(pArray, action);
1,000,022✔
1538
    if (pAction->msgReceived || pAction->rawWritten) {
1,000,022✔
1539
      numOfExecuted++;
701,294✔
1540
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
701,294✔
1541
        errCode = pAction->errCode;
30✔
1542
        pErrAction = pAction;
30✔
1543
      }
1544
    } else {
1545
      pErrAction = pAction;
298,728✔
1546
    }
1547
  }
1548

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

1551
  if (numOfExecuted == numOfActions) {
207,469✔
1552
    if (errCode == 0) {
121,826✔
1553
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
121,824!
1554
      return 0;
121,824✔
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,643!
1563

1564
    for (int32_t action = 0; action < numOfActions; ++action) {
600,218✔
1565
      STransAction *pAction = taosArrayGet(pArray, action);
514,575✔
1566
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
514,575✔
1567
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1568
             pAction->acceptableCode, pAction->retryCode);
1569
      if (pAction->msgSent) {
514,575✔
1570
        if (pAction->msgReceived) {
514,386✔
1571
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
215,847✔
1572
            mndTransResetAction(pMnode, pTrans, pAction);
28✔
1573
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
28!
1574
          }
1575
        }
1576
      }
1577
    }
1578
    return TSDB_CODE_ACTION_IN_PROGRESS;
85,643✔
1579
  }
1580
}
1581

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

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

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

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

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

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

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

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

1651
    if (code == 0) {
36,243✔
1652
      pTrans->failedTimes = 0;
10,025✔
1653
    }
1654
    mndSetTransLastAction(pTrans, pAction);
36,243✔
1655

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

1664
    if (code == 0) {
26,991✔
1665
      pTrans->code = 0;
8,178✔
1666
      pTrans->actionPos++;
8,178✔
1667
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
8,178!
1668
            pAction->id);
1669
      (void)taosThreadMutexUnlock(&pTrans->mutex);
8,178✔
1670
      code = mndTransSync(pMnode, pTrans);
8,178✔
1671
      (void)taosThreadMutexLock(&pTrans->mutex);
8,178✔
1672
      if (code != 0) {
8,178!
UNCOV
1673
        pTrans->actionPos--;
×
UNCOV
1674
        pTrans->code = terrno;
×
UNCOV
1675
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1676
               mndTransStr(pAction->stage), pAction->id, terrstr());
UNCOV
1677
        break;
×
1678
      }
1679
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
18,813✔
1680
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
14,162!
1681
      break;
14,162✔
1682
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
4,651✔
1683
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
31✔
1684
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
4,634!
1685
            code, tstrerror(code));
1686
      pTrans->lastErrorNo = code;
4,634✔
1687
      taosMsleep(300);
4,634✔
1688
      action--;
4,634✔
1689
      continue;
4,634✔
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,211✔
1701
}
1702

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

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

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

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

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

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

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

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

1755
  if (pTrans->exec == TRN_EXEC_SERIAL) {
207,149✔
1756
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
29,281✔
1757
  } else {
1758
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf, notSend);
177,868✔
1759
  }
1760

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

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

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

1815
  return continueExec;
144,293✔
1816
}
1817

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

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

1825
  if (code == 0) {
44,478✔
1826
    pTrans->code = 0;
44,476✔
1827
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
44,476✔
1828
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
44,476!
1829
    continueExec = true;
44,476✔
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,478✔
1838
}
1839

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

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

1861
  return continueExec;
97,948✔
1862
}
1863

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

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

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

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

1890
  return continueExec;
12✔
1891
}
1892

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

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

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

1909
  return continueExec;
2✔
1910
}
1911

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

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

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

1929
  return continueExec;
2✔
1930
}
1931

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

2140
  return buf;
14,492✔
2141
}
2142

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

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

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

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

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

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

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

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

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

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

2235
    cols = 0;
215✔
2236

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

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

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

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

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

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

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

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

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

2302
    mndTransLogAction(pTrans);
215✔
2303

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

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

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

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

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

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

2347
  cols = 0;
×
2348

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

© 2026 Coveralls, Inc