#4829

Committed 30 Oct 2025 09:25AM UTC coverage: 49.734% (-11.3%) from 61.071%

Build # #4829

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #33435 from taosdata/3.0

merge 3.0

Run Details

123072 of 323930 branches covered (37.99%)

Branch coverage included in aggregate %.

7 of 25 new or added lines in 3 files covered. (28.0%)

35232 existing lines in 327 files now uncovered.

172062 of 269495 relevant lines covered (63.85%)

70709785.06 hits per line

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

68.52

/source/libs/sync/src/syncCommit.c

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

#define _DEFAULT_SOURCE
#include "syncCommit.h"
#include "syncIndexMgr.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncUtil.h"

// \* Leader i advances its commitIndex.
// \* This is done as a separate step from handling AppendEntries responses,
// \* in part to minimize atomic regions, and in part so that leaders of
// \* single-server clusters are able to mark entries committed.
// AdvanceCommitIndex(i) ==
//     /\ state[i] = Leader
//     /\ LET \* The set of servers that agree up through index.
//            Agree(index) == {i} \cup {k \in Server :
//                                          matchIndex[i][k] >= index}
//            \* The maximum indexes for which a quorum agrees
//            agreeIndexes == {index \in 1..Len(log[i]) :
//                                 Agree(index) \in Quorum}
//            \* New value for commitIndex'[i]
//            newCommitIndex ==
//               IF /\ agreeIndexes /= {}
//                  /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
//               THEN
//                   Max(agreeIndexes)
//               ELSE
//                   commitIndex[i]
//        IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
//     /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
//

static inline int64_t syncNodeAbs64(int64_t a, int64_t b) {
  int64_t c = a > b ? a - b : b - a;
  return c;
}

int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode) { return pSyncNode->quorum; }

bool syncNodeAgreedUpon(SSyncNode* pNode, SyncIndex index) {
  int            count = 0;
  SSyncIndexMgr* pMatches = pNode->pMatchIndex;
  if (pNode->replicaNum != pMatches->replicaNum) {
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return false;
  };

  for (int i = 0; i < pNode->totalReplicaNum; i++) {
    if(pNode->raftCfg.cfg.nodeInfo[i].nodeRole == TAOS_SYNC_ROLE_VOTER){
      SyncIndex matchIndex = pMatches->index[i];
      if (matchIndex >= index) {
        count++;
      }
    }
  }

  return count >= pNode->quorum;
}

int64_t syncNodeUpdateCommitIndex(SSyncNode* ths, SyncIndex commitIndex) {
  int32_t   code = 0;
  SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);
  commitIndex = TMAX(commitIndex, ths->commitIndex);
  ths->commitIndex = TMIN(commitIndex, lastVer);
  // TODO add return when error
  (void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->commitIndex);
  return ths->commitIndex;
}

int64_t syncNodeCheckCommitIndex(SSyncNode* ths, SyncIndex indexLikely, const STraceId *trace) {
  int32_t code = 0;
  if (indexLikely > ths->commitIndex && syncNodeAgreedUpon(ths, indexLikely)) {
    SyncIndex commitIndex = indexLikely;
    SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, commitIndex);
    sGDebug(trace, "vgId:%d, index:%" PRId64 ", agreed upon, role:%d term:%" PRId64 " return index:%" PRId64, ths->vgId,
            commitIndex, ths->state, raftStoreGetTerm(ths), returnIndex);
  }
  return ths->commitIndex;
}

int64_t syncNodeUpdateAssignedCommitIndex(SSyncNode* ths, SyncIndex assignedCommitIndex) {
  int32_t   code = 0;
  SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);
  assignedCommitIndex = TMAX(assignedCommitIndex, ths->assignedCommitIndex);
  ths->assignedCommitIndex = TMIN(assignedCommitIndex, lastVer);
  // TODO add return when error
  (void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->assignedCommitIndex);
  return ths->commitIndex;
}

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 "syncCommit.h"
18	#include "syncIndexMgr.h"
19	#include "syncRaftLog.h"
20	#include "syncRaftStore.h"
21	#include "syncUtil.h"
22
23	// \* Leader i advances its commitIndex.
24	// \* This is done as a separate step from handling AppendEntries responses,
25	// \* in part to minimize atomic regions, and in part so that leaders of
26	// \* single-server clusters are able to mark entries committed.
27	// AdvanceCommitIndex(i) ==
28	// /\ state[i] = Leader
29	// /\ LET \* The set of servers that agree up through index.
30	// Agree(index) == {i} \cup {k \in Server :
31	// matchIndex[i][k] >= index}
32	// \* The maximum indexes for which a quorum agrees
33	// agreeIndexes == {index \in 1..Len(log[i]) :
34	// Agree(index) \in Quorum}
35	// \* New value for commitIndex'[i]
36	// newCommitIndex ==
37	// IF /\ agreeIndexes /= {}
38	// /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
39	// THEN
40	// Max(agreeIndexes)
41	// ELSE
42	// commitIndex[i]
43	// IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
44	// /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
45	//
46
47	static inline int64_t syncNodeAbs64(int64_t a, int64_t b) {
48	int64_t c = a > b ? a - b : b - a;
49	return c;
50	}
51
52	int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode) { return pSyncNode->quorum; }	3,547,562✔
53
54	bool syncNodeAgreedUpon(SSyncNode* pNode, SyncIndex index) {	3,337,249✔
55	int count = 0;	3,337,249✔
56	SSyncIndexMgr* pMatches = pNode->pMatchIndex;	3,337,249✔
57	if (pNode->replicaNum != pMatches->replicaNum) {	3,337,249!
58	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	×
59	return false;	×
60	};
61
62	for (int i = 0; i < pNode->totalReplicaNum; i++) {	13,343,510✔
63	if(pNode->raftCfg.cfg.nodeInfo[i].nodeRole == TAOS_SYNC_ROLE_VOTER){	10,006,438✔
64	SyncIndex matchIndex = pMatches->index[i];	9,995,136✔
65	if (matchIndex >= index) {	9,994,959✔
66	count++;	6,675,625✔
67	}
68	}
69	}
70
71	return count >= pNode->quorum;	3,337,072✔
72	}
73
74	int64_t syncNodeUpdateCommitIndex(SSyncNode* ths, SyncIndex commitIndex) {	177,652,477✔
75	int32_t code = 0;	177,652,477✔
76	SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);	177,652,477✔
77	commitIndex = TMAX(commitIndex, ths->commitIndex);	177,659,084✔
78	ths->commitIndex = TMIN(commitIndex, lastVer);	177,658,798✔
79	// TODO add return when error
80	(void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->commitIndex);	177,659,154✔
81	return ths->commitIndex;	177,652,044✔
82	}
83
84	int64_t syncNodeCheckCommitIndex(SSyncNode* ths, SyncIndex indexLikely, const STraceId *trace) {	7,041,260✔
85	int32_t code = 0;	7,041,260✔
86	if (indexLikely > ths->commitIndex && syncNodeAgreedUpon(ths, indexLikely)) {	7,041,260✔
87	SyncIndex commitIndex = indexLikely;	3,332,202✔
88	SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, commitIndex);	3,332,202✔
89	sGDebug(trace, "vgId:%d, index:%" PRId64 ", agreed upon, role:%d term:%" PRId64 " return index:%" PRId64, ths->vgId,	3,332,379!
90	commitIndex, ths->state, raftStoreGetTerm(ths), returnIndex);
91	}
92	return ths->commitIndex;	7,041,260✔
93	}
94
UNCOV 95	int64_t syncNodeUpdateAssignedCommitIndex(SSyncNode* ths, SyncIndex assignedCommitIndex) {	×
UNCOV 96	int32_t code = 0;	×
UNCOV 97	SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);	×
UNCOV 98	assignedCommitIndex = TMAX(assignedCommitIndex, ths->assignedCommitIndex);	×
UNCOV 99	ths->assignedCommitIndex = TMIN(assignedCommitIndex, lastVer);	×
100	// TODO add return when error
UNCOV 101	(void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->assignedCommitIndex);	×
UNCOV 102	return ths->commitIndex;	×
103	}

taosdata / TDengine / #4829

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

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