Line data Source code
1 : // Copyright (c) 2018-2021 The Dash Core developers
2 : // Copyright (c) 2022 The PIVX Core developers
3 : // Distributed under the MIT/X11 software license, see the accompanying
4 : // file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 :
6 : #include "llmq/quorums_dkgsessionhandler.h"
7 :
8 : #include "activemasternode.h"
9 : #include "chainparams.h"
10 : #include "llmq/quorums_blockprocessor.h"
11 : #include "llmq/quorums_connections.h"
12 : #include "llmq/quorums_debug.h"
13 : #include "net_processing.h"
14 : #include "shutdown.h"
15 : #include "util/threadnames.h"
16 : #include "validation.h"
17 :
18 : namespace llmq
19 : {
20 :
21 2528 : CDKGPendingMessages::CDKGPendingMessages(size_t _maxMessagesPerNode) :
22 2528 : maxMessagesPerNode(_maxMessagesPerNode)
23 : {
24 2528 : }
25 :
26 456 : void CDKGPendingMessages::PushPendingMessage(NodeId from, CDataStream& vRecv, int invType)
27 : {
28 : // this will also consume the data, even if we bail out early
29 912 : auto pm = std::make_shared<CDataStream>(std::move(vRecv));
30 :
31 456 : {
32 456 : LOCK(cs);
33 :
34 456 : if (messagesPerNode[from] >= maxMessagesPerNode) {
35 : // TODO ban?
36 0 : LogPrint(BCLog::NET, "CDKGPendingMessages::%s -- too many messages, peer=%d\n", __func__, from);
37 0 : return;
38 : }
39 456 : messagesPerNode[from]++;
40 : }
41 :
42 456 : CHashWriter hw(SER_GETHASH, 0);
43 456 : hw.write(pm->data(), pm->size());
44 456 : uint256 hash = hw.GetHash();
45 :
46 1368 : LOCK2(cs_main, cs);
47 :
48 456 : g_connman->RemoveAskFor(hash, invType);
49 :
50 456 : if (!seenMessages.emplace(hash).second) {
51 0 : LogPrint(BCLog::NET, "CDKGPendingMessages::%s -- already seen %s, peer=%d\n", __func__, hash.ToString(), from);
52 0 : return;
53 : }
54 :
55 912 : pendingMessages.emplace_back(std::make_pair(from, std::move(pm)));
56 : }
57 :
58 11792 : std::list<CDKGPendingMessages::BinaryMessage> CDKGPendingMessages::PopPendingMessages(size_t maxCount)
59 : {
60 11792 : LOCK(cs);
61 :
62 11792 : std::list<BinaryMessage> ret;
63 450 : while (!pendingMessages.empty() && ret.size() < maxCount) {
64 450 : ret.emplace_back(std::move(pendingMessages.front()));
65 12242 : pendingMessages.pop_front();
66 : }
67 :
68 23584 : return ret;
69 : }
70 :
71 3690 : bool CDKGPendingMessages::HasSeen(const uint256& hash) const
72 : {
73 3690 : LOCK(cs);
74 7380 : return seenMessages.count(hash) != 0;
75 : }
76 :
77 1388 : void CDKGPendingMessages::Clear()
78 : {
79 1388 : LOCK(cs);
80 1388 : pendingMessages.clear();
81 1388 : messagesPerNode.clear();
82 1388 : seenMessages.clear();
83 1388 : }
84 :
85 : //////
86 :
87 632 : CDKGSessionHandler::CDKGSessionHandler(const Consensus::LLMQParams& _params, CBLSWorker& _blsWorker, CDKGSessionManager& _dkgManager) :
88 : params(_params),
89 : blsWorker(_blsWorker),
90 : dkgManager(_dkgManager),
91 : curSession(std::make_shared<CDKGSession>(_params, _blsWorker, _dkgManager)),
92 632 : pendingContributions((size_t)_params.size * 2), // we allow size*2 messages as we need to make sure we see bad behavior (double messages)
93 632 : pendingComplaints((size_t)_params.size * 2),
94 632 : pendingJustifications((size_t)_params.size * 2),
95 632 : pendingPrematureCommitments((size_t)_params.size * 2)
96 : {
97 632 : if (params.type == Consensus::LLMQ_NONE) {
98 0 : throw std::runtime_error("Can't initialize CDKGSessionHandler with LLMQ_NONE type.");
99 : }
100 632 : }
101 :
102 632 : CDKGSessionHandler::~CDKGSessionHandler()
103 : {
104 632 : }
105 :
106 9183 : void CDKGSessionHandler::UpdatedBlockTip(const CBlockIndex* pindexNew)
107 : {
108 9183 : AssertLockHeld(cs_main);
109 9183 : LOCK(cs);
110 :
111 9183 : int quorumStageInt = pindexNew->nHeight % params.dkgInterval;
112 9183 : const CBlockIndex* pindexQuorum = chainActive[pindexNew->nHeight - quorumStageInt];
113 :
114 9183 : currentHeight = pindexNew->nHeight;
115 9183 : quorumHeight = pindexQuorum->nHeight;
116 9183 : quorumHash = pindexQuorum->GetBlockHash();
117 :
118 9183 : bool fNewPhase = (quorumStageInt % params.dkgPhaseBlocks) == 0;
119 9183 : int phaseInt = quorumStageInt / params.dkgPhaseBlocks + 1;
120 9183 : QuorumPhase oldPhase = phase;
121 :
122 9183 : if (fNewPhase && phaseInt >= QuorumPhase_Initialized && phaseInt <= QuorumPhase_Idle) {
123 3213 : phase = static_cast<QuorumPhase>(phaseInt);
124 : }
125 :
126 9183 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - currentHeight=%d, quorumHeight=%d, oldPhase=%d, newPhase=%d\n", __func__,
127 : params.name, currentHeight, quorumHeight, oldPhase, phase);
128 9183 : }
129 :
130 292 : void CDKGSessionHandler::ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv)
131 : {
132 : // We don't handle messages in the calling thread as deserialization/processing of these would block everything
133 292 : if (strCommand == NetMsgType::QCONTRIB) {
134 138 : pendingContributions.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_CONTRIB);
135 154 : } else if (strCommand == NetMsgType::QCOMPLAINT) {
136 28 : pendingComplaints.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_COMPLAINT);
137 126 : } else if (strCommand == NetMsgType::QJUSTIFICATION) {
138 8 : pendingJustifications.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_JUSTIFICATION);
139 118 : } else if (strCommand == NetMsgType::QPCOMMITMENT) {
140 118 : pendingPrematureCommitments.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_PREMATURE_COMMITMENT);
141 : }
142 292 : }
143 :
144 347 : void CDKGSessionHandler::StartThread()
145 : {
146 347 : if (phaseHandlerThread.joinable()) {
147 0 : throw std::runtime_error("Tried to start an already started CDKGSessionHandler thread.");
148 : }
149 :
150 347 : std::string threadName = strprintf("llmq-%d", (uint8_t)params.type);
151 347 : phaseHandlerThread = std::thread(&TraceThread<std::function<void()> >, threadName, std::function<void()>(std::bind(&CDKGSessionHandler::PhaseHandlerThread, this)));
152 347 : }
153 :
154 357 : void CDKGSessionHandler::StopThread()
155 : {
156 357 : stopRequested = true;
157 357 : if (phaseHandlerThread.joinable()) {
158 347 : phaseHandlerThread.join();
159 : }
160 357 : }
161 :
162 347 : bool CDKGSessionHandler::InitNewQuorum(const CBlockIndex* pindexQuorum)
163 : {
164 694 : curSession = std::make_shared<CDKGSession>(params, blsWorker, dkgManager);
165 :
166 347 : if (!deterministicMNManager->IsDIP3Enforced(pindexQuorum->nHeight) ||
167 347 : !activeMasternodeManager) {
168 : return false;
169 : }
170 :
171 350 : auto mns = deterministicMNManager->GetAllQuorumMembers(params.type, pindexQuorum);
172 :
173 175 : if (!curSession->Init(pindexQuorum, mns, activeMasternodeManager->GetProTx())) {
174 11 : LogPrintf("CDKGSessionHandler::%s -- quorum initialization failed for %s\n", __func__, curSession->params.name);
175 : return false;
176 : }
177 :
178 : return true;
179 : }
180 :
181 147789 : CDKGSessionHandler::QuorumPhaseAndHash CDKGSessionHandler::GetPhaseAndQuorumHash() const
182 : {
183 147789 : LOCK(cs);
184 147789 : return {phase, quorumHash};
185 : }
186 :
187 535 : class AbortPhaseException : public std::exception {
188 : };
189 :
190 1380 : void CDKGSessionHandler::WaitForNextPhase(QuorumPhase curPhase,
191 : QuorumPhase nextPhase,
192 : const uint256& expectedQuorumHash,
193 : const WhileWaitFunc& runWhileWaiting)
194 : {
195 1380 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
196 :
197 215578 : while (true) {
198 108479 : if (stopRequested || ShutdownRequested()) {
199 322 : throw AbortPhaseException();
200 : }
201 108157 : auto currState = GetPhaseAndQuorumHash();
202 3059020 : if (!expectedQuorumHash.IsNull() && currState.quorumHash != expectedQuorumHash) {
203 6 : throw AbortPhaseException();
204 : }
205 108151 : if (currState.phase == nextPhase) {
206 : break;
207 : }
208 107123 : if (curPhase != QuorumPhase_None && currState.phase != curPhase) {
209 24 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - aborting due unexpected phase change\n", __func__, params.name);
210 24 : throw AbortPhaseException();
211 : }
212 107099 : if (!runWhileWaiting()) {
213 106727 : MilliSleep(100);
214 : }
215 107099 : }
216 :
217 1028 : if (nextPhase == QuorumPhase_Initialized) {
218 347 : quorumDKGDebugManager->ResetLocalSessionStatus(params.type);
219 : } else {
220 1362 : quorumDKGDebugManager->UpdateLocalSessionStatus(params.type, [&](CDKGDebugSessionStatus& status) {
221 348 : bool changed = status.phase != (uint8_t) nextPhase;
222 348 : status.phase = (uint8_t) nextPhase;
223 348 : return changed;
224 : });
225 : }
226 :
227 1028 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
228 :
229 1028 : }
230 :
231 183 : void CDKGSessionHandler::WaitForNewQuorum(const uint256& oldQuorumHash)
232 : {
233 183 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting\n", __func__, params.name);
234 :
235 79131 : while (true) {
236 39657 : if (stopRequested || ShutdownRequested()) {
237 25 : throw AbortPhaseException();
238 : }
239 39632 : auto currState = GetPhaseAndQuorumHash();
240 39632 : if (currState.quorumHash != oldQuorumHash) {
241 : break;
242 : }
243 39474 : MilliSleep(100);
244 39474 : }
245 :
246 158 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done\n", __func__, params.name);
247 158 : }
248 :
249 : // Sleep some time to not fully overload the whole network
250 548 : void CDKGSessionHandler::SleepBeforePhase(QuorumPhase curPhase,
251 : const uint256& expectedQuorumHash,
252 : double randomSleepFactor,
253 : const WhileWaitFunc& runWhileWaiting)
254 : {
255 548 : if (Params().IsRegTestNet()) {
256 : // On regtest, blocks can be mined on demand without any significant time passing between these.
257 : // We shouldn't wait before phases in this case.
258 548 : return;
259 : }
260 :
261 0 : if (!curSession->AreWeMember()) {
262 : // Non-members do not participate and do not create any network load, no need to sleep.
263 : return;
264 : }
265 :
266 : // Two blocks can come very close to each other, this happens pretty regularly. We don't want to be
267 : // left behind and marked as a bad member. This means that we should not count the last block of the
268 : // phase as a safe one to keep sleeping, that's why we calculate the phase sleep time as a time of
269 : // the full phase minus one block here.
270 0 : const int nTargetSpacing = Params().GetConsensus().nTargetSpacing;
271 0 : double phaseSleepTime = (params.dkgPhaseBlocks - 1) * nTargetSpacing * 1000;
272 : // Expected phase sleep time per member
273 0 : double phaseSleepTimePerMember = phaseSleepTime / params.size;
274 : // Don't expect perfect block times and thus reduce the phase time to be on the secure side (caller chooses factor)
275 0 : double adjustedPhaseSleepTimePerMember = phaseSleepTimePerMember * randomSleepFactor;
276 :
277 0 : int64_t sleepTime = (int64_t)(adjustedPhaseSleepTimePerMember * curSession->GetMyMemberIndex());
278 0 : int64_t endTime = GetTimeMillis() + sleepTime;
279 0 : int heightTmp{-1};
280 0 : int heightStart{-1};
281 0 : {
282 0 : LOCK(cs);
283 0 : heightTmp = heightStart = currentHeight;
284 : }
285 :
286 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting sleep for %d ms, curPhase=%d\n", __func__, params.name, sleepTime, curPhase);
287 :
288 0 : while (GetTimeMillis() < endTime) {
289 0 : if (stopRequested) {
290 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - aborting due to stop/shutdown requested\n", __func__, params.name);
291 0 : throw AbortPhaseException();
292 : }
293 0 : {
294 0 : LOCK(cs);
295 0 : if (currentHeight > heightTmp) {
296 : // New block(s) just came in
297 0 : int64_t expectedBlockTime = (currentHeight - heightStart) * nTargetSpacing * 1000;
298 0 : if (expectedBlockTime > sleepTime) {
299 : // Blocks came faster than we expected, jump into the phase func asap
300 : break;
301 : }
302 : heightTmp = currentHeight;
303 : }
304 0 : if (phase != curPhase || quorumHash != expectedQuorumHash) {
305 : // Something went wrong and/or we missed quite a few blocks and it's just too late now
306 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - aborting due unexpected phase/expectedQuorumHash change\n", __func__, params.name);
307 0 : throw AbortPhaseException();
308 : }
309 : }
310 0 : if (!runWhileWaiting()) {
311 0 : MilliSleep(100);
312 : }
313 : }
314 :
315 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done, curPhase=%d\n", __func__, params.name, curPhase);
316 : }
317 :
318 548 : void CDKGSessionHandler::HandlePhase(QuorumPhase curPhase,
319 : QuorumPhase nextPhase,
320 : const uint256& expectedQuorumHash,
321 : double randomSleepFactor,
322 : const StartPhaseFunc& startPhaseFunc,
323 : const WhileWaitFunc& runWhileWaiting)
324 : {
325 548 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
326 :
327 548 : SleepBeforePhase(curPhase, expectedQuorumHash, randomSleepFactor, runWhileWaiting);
328 548 : startPhaseFunc();
329 548 : WaitForNextPhase(curPhase, nextPhase, expectedQuorumHash, runWhileWaiting);
330 :
331 535 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
332 535 : }
333 :
334 : // returns a set of NodeIds which sent invalid messages
335 : template<typename Message>
336 372 : std::set<NodeId> BatchVerifyMessageSigs(CDKGSession& session, const std::vector<std::pair<NodeId, std::shared_ptr<Message>>>& messages)
337 : {
338 372 : if (messages.empty()) {
339 0 : return {};
340 : }
341 :
342 372 : std::set<NodeId> ret;
343 372 : bool revertToSingleVerification = false;
344 :
345 372 : CBLSSignature aggSig;
346 372 : std::vector<CBLSPublicKey> pubKeys;
347 744 : std::vector<uint256> messageHashes;
348 744 : std::set<uint256> messageHashesSet;
349 372 : pubKeys.reserve(messages.size());
350 372 : messageHashes.reserve(messages.size());
351 372 : bool first = true;
352 786 : for (const auto& p : messages ) {
353 450 : const auto& msg = *p.second;
354 :
355 450 : auto member = session.GetMember(msg.proTxHash);
356 450 : if (!member) {
357 : // should not happen as it was verified before
358 0 : ret.emplace(p.first);
359 0 : continue;
360 : }
361 :
362 450 : if (first) {
363 372 : aggSig = msg.sig;
364 : } else {
365 78 : aggSig.AggregateInsecure(msg.sig);
366 : }
367 450 : first = false;
368 :
369 450 : auto msgHash = msg.GetSignHash();
370 450 : if (!messageHashesSet.emplace(msgHash).second) {
371 : // can only happen in 2 cases:
372 : // 1. Someone sent us the same message twice but with differing signature, meaning that at least one of them
373 : // must be invalid. In this case, we'd have to revert to single message verification nevertheless
374 : // 2. Someone managed to find a way to create two different binary representations of a message that deserializes
375 : // to the same object representation. This would be some form of malleability. However, this shouldn't be
376 : // possible as only deterministic/unique BLS signatures and very simple data types are involved
377 36 : revertToSingleVerification = true;
378 36 : break;
379 : }
380 :
381 414 : pubKeys.emplace_back(member->dmn->pdmnState->pubKeyOperator.Get());
382 414 : messageHashes.emplace_back(msgHash);
383 : }
384 : if (!revertToSingleVerification) {
385 336 : bool valid = aggSig.VerifyInsecureAggregated(pubKeys, messageHashes);
386 336 : if (valid) {
387 : // all good
388 336 : return ret;
389 : }
390 :
391 : // are all messages from the same node?
392 : NodeId firstNodeId;
393 1 : first = true;
394 1 : bool nodeIdsAllSame = true;
395 2 : for (auto it = messages.begin(); it != messages.end(); ++it) {
396 : if (first) {
397 1 : firstNodeId = it->first;
398 : } else {
399 : first = false;
400 : if (it->first != firstNodeId) {
401 : nodeIdsAllSame = false;
402 : break;
403 : }
404 : }
405 : }
406 : // if yes, take a short path and return a set with only him
407 : if (nodeIdsAllSame) {
408 1 : ret.emplace(firstNodeId);
409 336 : return ret;
410 : }
411 : // different nodes, let's figure out who are the bad ones
412 : }
413 :
414 108 : for (const auto& p : messages) {
415 72 : if (ret.count(p.first)) {
416 0 : continue;
417 : }
418 :
419 72 : const auto& msg = *p.second;
420 72 : auto member = session.GetMember(msg.proTxHash);
421 72 : bool valid = msg.sig.VerifyInsecure(member->dmn->pdmnState->pubKeyOperator.Get(), msg.GetSignHash());
422 72 : if (!valid) {
423 72 : ret.emplace(p.first);
424 : }
425 : }
426 372 : return ret;
427 : }
428 :
429 : template<typename Message>
430 11792 : static bool ProcessPendingMessageBatch(CDKGSession& session, CDKGPendingMessages& pendingMessages, size_t maxCount)
431 : {
432 11792 : auto msgs = pendingMessages.PopAndDeserializeMessages<Message>(maxCount);
433 11792 : if (msgs.empty()) {
434 : return false;
435 : }
436 :
437 11792 : std::vector<uint256> hashes;
438 744 : std::vector<std::pair<NodeId, std::shared_ptr<Message>>> preverifiedMessages;
439 372 : hashes.reserve(msgs.size());
440 372 : preverifiedMessages.reserve(msgs.size());
441 :
442 822 : for (const auto& p : msgs) {
443 450 : if (!p.second) {
444 0 : LogPrint(BCLog::NET, "%s -- failed to deserialize message, peer=%d\n", __func__, p.first);
445 : {
446 0 : LOCK(cs_main);
447 0 : Misbehaving(p.first, 100);
448 : }
449 0 : continue;
450 : }
451 450 : const auto& msg = *p.second;
452 :
453 450 : bool ban = false;
454 450 : if (!session.PreVerifyMessage(msg, ban)) {
455 0 : if (ban) {
456 0 : LogPrint(BCLog::NET, "%s -- banning node due to failed preverification, peer=%d\n", __func__, p.first);
457 : {
458 0 : LOCK(cs_main);
459 0 : Misbehaving(p.first, 100);
460 : }
461 : }
462 0 : LogPrint(BCLog::NET, "%s -- skipping message due to failed preverification, peer=%d\n", __func__, p.first);
463 0 : continue;
464 : }
465 450 : hashes.emplace_back(::SerializeHash(msg));
466 450 : preverifiedMessages.emplace_back(p);
467 : }
468 372 : if (preverifiedMessages.empty()) {
469 : return true;
470 : }
471 :
472 744 : auto badNodes = BatchVerifyMessageSigs(session, preverifiedMessages);
473 372 : if (!badNodes.empty()) {
474 2 : LOCK(cs_main);
475 2 : for (auto nodeId : badNodes) {
476 1 : LogPrint(BCLog::NET, "%s -- failed to verify signature, peer=%d\n", __func__, nodeId);
477 2 : Misbehaving(nodeId, 100);
478 : }
479 : }
480 :
481 822 : for (size_t i = 0; i < preverifiedMessages.size(); i++) {
482 450 : NodeId nodeId = preverifiedMessages[i].first;
483 449 : if (badNodes.count(nodeId)) {
484 1 : continue;
485 : }
486 449 : const auto& msg = *preverifiedMessages[i].second;
487 449 : bool ban = false;
488 449 : session.ReceiveMessage(hashes[i], msg, ban);
489 449 : if (ban) {
490 0 : LogPrint(BCLog::NET, "%s -- banning node after ReceiveMessage failed, peer=%d\n", __func__, nodeId);
491 0 : LOCK(cs_main);
492 0 : Misbehaving(nodeId, 100);
493 0 : badNodes.emplace(nodeId);
494 : }
495 : }
496 :
497 372 : return true;
498 : }
499 :
500 668 : void CDKGSessionHandler::HandleDKGRound()
501 : {
502 668 : uint256 curQuorumHash;
503 :
504 989 : WaitForNextPhase(QuorumPhase_None, QuorumPhase_Initialized, UINT256_ZERO, []{return false;});
505 :
506 347 : {
507 347 : LOCK(cs);
508 347 : pendingContributions.Clear();
509 347 : pendingComplaints.Clear();
510 347 : pendingJustifications.Clear();
511 347 : pendingPrematureCommitments.Clear();
512 347 : curQuorumHash = quorumHash;
513 : }
514 :
515 694 : const CBlockIndex* pindexQuorum = WITH_LOCK(cs_main, return LookupBlockIndex(curQuorumHash));
516 347 : if (!pindexQuorum) {
517 : // should never happen
518 0 : LogPrintf("%s: ERROR: Unable to find block %s\n", __func__, curQuorumHash.ToString());
519 0 : return;
520 : }
521 :
522 347 : if (!InitNewQuorum(pindexQuorum)) {
523 : // should actually never happen
524 183 : WaitForNewQuorum(curQuorumHash);
525 158 : throw AbortPhaseException();
526 : }
527 :
528 164 : quorumDKGDebugManager->UpdateLocalSessionStatus(params.type, [&](CDKGDebugSessionStatus& status) {
529 83 : bool changed = status.phase != (uint8_t) QuorumPhase_Initialized;
530 83 : status.phase = (uint8_t) QuorumPhase_Initialized;
531 83 : return changed;
532 : });
533 :
534 164 : EnsureQuorumConnections(params.type, pindexQuorum, curSession->myProTxHash);
535 328 : if (curSession->AreWeMember()) {
536 83 : AddQuorumProbeConnections(params.type, pindexQuorum, curSession->myProTxHash);
537 : }
538 :
539 182 : WaitForNextPhase(QuorumPhase_Initialized, QuorumPhase_Contribute, curQuorumHash, []{return false;});
540 :
541 : // Contribute
542 292 : auto fContributeStart = [this]() {
543 146 : curSession->Contribute(pendingContributions);
544 146 : };
545 3439 : auto fContributeWait = [this] {
546 3293 : return ProcessPendingMessageBatch<CDKGContribution>(*curSession, pendingContributions, 8);
547 146 : };
548 302 : HandlePhase(QuorumPhase_Contribute, QuorumPhase_Complain, curQuorumHash, 0.05, fContributeStart, fContributeWait);
549 :
550 : // Complain
551 272 : auto fComplainStart = [this]() {
552 136 : curSession->VerifyAndComplain(pendingComplaints);
553 136 : };
554 3161 : auto fComplainWait = [this] {
555 3025 : return ProcessPendingMessageBatch<CDKGComplaint>(*curSession, pendingComplaints, 8);
556 136 : };
557 275 : HandlePhase(QuorumPhase_Complain, QuorumPhase_Justify, curQuorumHash, 0.05, fComplainStart, fComplainWait);
558 :
559 : // Justify
560 266 : auto fJustifyStart = [this]() {
561 133 : curSession->VerifyAndJustify(pendingJustifications);
562 133 : };
563 2844 : auto fJustifyWait = [this] {
564 2711 : return ProcessPendingMessageBatch<CDKGJustification>(*curSession, pendingJustifications, 8);
565 133 : };
566 266 : HandlePhase(QuorumPhase_Justify, QuorumPhase_Commit, curQuorumHash, 0.05, fJustifyStart, fJustifyWait);
567 :
568 : // Commit
569 266 : auto fCommitStart = [this]() {
570 133 : curSession->VerifyAndCommit(pendingPrematureCommitments);
571 133 : };
572 2896 : auto fCommitWait = [this] {
573 2763 : return ProcessPendingMessageBatch<CDKGPrematureCommitment>(*curSession, pendingPrematureCommitments, 8);
574 133 : };
575 266 : HandlePhase(QuorumPhase_Commit, QuorumPhase_Finalize, curQuorumHash, 0.1, fCommitStart, fCommitWait);
576 :
577 266 : auto finalCommitments = curSession->FinalizeCommitments();
578 198 : for (const auto& fqc : finalCommitments) {
579 65 : quorumBlockProcessor->AddAndRelayMinableCommitment(fqc);
580 : }
581 : }
582 :
583 347 : void CDKGSessionHandler::PhaseHandlerThread()
584 : {
585 1015 : while (!stopRequested && !ShutdownRequested()) {
586 668 : try {
587 668 : HandleDKGRound();
588 1070 : } catch (AbortPhaseException& e) {
589 535 : quorumDKGDebugManager->UpdateLocalSessionStatus(params.type, [&](CDKGDebugSessionStatus& status) {
590 32 : status.aborted = true;
591 32 : return true;
592 : });
593 535 : LogPrintf("CDKGSessionHandler::%s -- aborted current DKG session for llmq=%s\n", __func__, params.name);
594 : }
595 : }
596 347 : }
597 :
598 : }
|
