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 2276 : CDKGPendingMessages::CDKGPendingMessages(size_t _maxMessagesPerNode) :
22 2276 : maxMessagesPerNode(_maxMessagesPerNode)
23 : {
24 2276 : }
25 :
26 0 : void CDKGPendingMessages::PushPendingMessage(NodeId from, CDataStream& vRecv, int invType)
27 : {
28 : // this will also consume the data, even if we bail out early
29 0 : auto pm = std::make_shared<CDataStream>(std::move(vRecv));
30 :
31 0 : {
32 0 : LOCK(cs);
33 :
34 0 : 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 0 : messagesPerNode[from]++;
40 : }
41 :
42 0 : CHashWriter hw(SER_GETHASH, 0);
43 0 : hw.write(pm->data(), pm->size());
44 0 : uint256 hash = hw.GetHash();
45 :
46 0 : LOCK2(cs_main, cs);
47 :
48 0 : g_connman->RemoveAskFor(hash, invType);
49 :
50 0 : 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 0 : pendingMessages.emplace_back(std::make_pair(from, std::move(pm)));
56 : }
57 :
58 0 : std::list<CDKGPendingMessages::BinaryMessage> CDKGPendingMessages::PopPendingMessages(size_t maxCount)
59 : {
60 0 : LOCK(cs);
61 :
62 0 : std::list<BinaryMessage> ret;
63 0 : while (!pendingMessages.empty() && ret.size() < maxCount) {
64 0 : ret.emplace_back(std::move(pendingMessages.front()));
65 0 : pendingMessages.pop_front();
66 : }
67 :
68 0 : return ret;
69 : }
70 :
71 0 : bool CDKGPendingMessages::HasSeen(const uint256& hash) const
72 : {
73 0 : LOCK(cs);
74 0 : return seenMessages.count(hash) != 0;
75 : }
76 :
77 40 : void CDKGPendingMessages::Clear()
78 : {
79 40 : LOCK(cs);
80 40 : pendingMessages.clear();
81 40 : messagesPerNode.clear();
82 40 : seenMessages.clear();
83 40 : }
84 :
85 : //////
86 :
87 569 : 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 569 : pendingContributions((size_t)_params.size * 2), // we allow size*2 messages as we need to make sure we see bad behavior (double messages)
93 569 : pendingComplaints((size_t)_params.size * 2),
94 569 : pendingJustifications((size_t)_params.size * 2),
95 569 : pendingPrematureCommitments((size_t)_params.size * 2)
96 : {
97 569 : if (params.type == Consensus::LLMQ_NONE) {
98 0 : throw std::runtime_error("Can't initialize CDKGSessionHandler with LLMQ_NONE type.");
99 : }
100 569 : }
101 :
102 569 : CDKGSessionHandler::~CDKGSessionHandler()
103 : {
104 569 : }
105 :
106 795 : void CDKGSessionHandler::UpdatedBlockTip(const CBlockIndex* pindexNew)
107 : {
108 795 : AssertLockHeld(cs_main);
109 795 : LOCK(cs);
110 :
111 795 : int quorumStageInt = pindexNew->nHeight % params.dkgInterval;
112 795 : const CBlockIndex* pindexQuorum = chainActive[pindexNew->nHeight - quorumStageInt];
113 :
114 795 : currentHeight = pindexNew->nHeight;
115 795 : quorumHeight = pindexQuorum->nHeight;
116 795 : quorumHash = pindexQuorum->GetBlockHash();
117 :
118 795 : bool fNewPhase = (quorumStageInt % params.dkgPhaseBlocks) == 0;
119 795 : int phaseInt = quorumStageInt / params.dkgPhaseBlocks + 1;
120 795 : QuorumPhase oldPhase = phase;
121 :
122 795 : if (fNewPhase && phaseInt >= QuorumPhase_Initialized && phaseInt <= QuorumPhase_Idle) {
123 278 : phase = static_cast<QuorumPhase>(phaseInt);
124 : }
125 :
126 795 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - currentHeight=%d, quorumHeight=%d, oldPhase=%d, newPhase=%d\n", __func__,
127 : params.name, currentHeight, quorumHeight, oldPhase, phase);
128 795 : }
129 :
130 0 : 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 0 : if (strCommand == NetMsgType::QCONTRIB) {
134 0 : pendingContributions.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_CONTRIB);
135 0 : } else if (strCommand == NetMsgType::QCOMPLAINT) {
136 0 : pendingComplaints.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_COMPLAINT);
137 0 : } else if (strCommand == NetMsgType::QJUSTIFICATION) {
138 0 : pendingJustifications.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_JUSTIFICATION);
139 0 : } else if (strCommand == NetMsgType::QPCOMMITMENT) {
140 0 : pendingPrematureCommitments.PushPendingMessage(pfrom->GetId(), vRecv, MSG_QUORUM_PREMATURE_COMMITMENT);
141 : }
142 0 : }
143 :
144 285 : void CDKGSessionHandler::StartThread()
145 : {
146 285 : if (phaseHandlerThread.joinable()) {
147 0 : throw std::runtime_error("Tried to start an already started CDKGSessionHandler thread.");
148 : }
149 :
150 285 : std::string threadName = strprintf("llmq-%d", (uint8_t)params.type);
151 285 : phaseHandlerThread = std::thread(&TraceThread<std::function<void()> >, threadName, std::function<void()>(std::bind(&CDKGSessionHandler::PhaseHandlerThread, this)));
152 285 : }
153 :
154 295 : void CDKGSessionHandler::StopThread()
155 : {
156 295 : stopRequested = true;
157 295 : if (phaseHandlerThread.joinable()) {
158 285 : phaseHandlerThread.join();
159 : }
160 295 : }
161 :
162 10 : bool CDKGSessionHandler::InitNewQuorum(const CBlockIndex* pindexQuorum)
163 : {
164 20 : curSession = std::make_shared<CDKGSession>(params, blsWorker, dkgManager);
165 :
166 10 : if (!deterministicMNManager->IsDIP3Enforced(pindexQuorum->nHeight) ||
167 10 : !activeMasternodeManager) {
168 : return false;
169 : }
170 :
171 0 : auto mns = deterministicMNManager->GetAllQuorumMembers(params.type, pindexQuorum);
172 :
173 0 : if (!curSession->Init(pindexQuorum, mns, activeMasternodeManager->GetProTx())) {
174 0 : LogPrintf("CDKGSessionHandler::%s -- quorum initialization failed for %s\n", __func__, curSession->params.name);
175 : return false;
176 : }
177 :
178 : return true;
179 : }
180 :
181 88879 : CDKGSessionHandler::QuorumPhaseAndHash CDKGSessionHandler::GetPhaseAndQuorumHash() const
182 : {
183 88879 : LOCK(cs);
184 88879 : return {phase, quorumHash};
185 : }
186 :
187 288 : class AbortPhaseException : public std::exception {
188 : };
189 :
190 288 : void CDKGSessionHandler::WaitForNextPhase(QuorumPhase curPhase,
191 : QuorumPhase nextPhase,
192 : const uint256& expectedQuorumHash,
193 : const WhileWaitFunc& runWhileWaiting)
194 : {
195 288 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
196 :
197 159312 : while (true) {
198 79800 : if (stopRequested || ShutdownRequested()) {
199 278 : throw AbortPhaseException();
200 : }
201 79522 : auto currState = GetPhaseAndQuorumHash();
202 2624230 : if (!expectedQuorumHash.IsNull() && currState.quorumHash != expectedQuorumHash) {
203 0 : throw AbortPhaseException();
204 : }
205 79522 : if (currState.phase == nextPhase) {
206 : break;
207 : }
208 79512 : if (curPhase != QuorumPhase_None && currState.phase != curPhase) {
209 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - aborting due unexpected phase change\n", __func__, params.name);
210 0 : throw AbortPhaseException();
211 : }
212 79512 : if (!runWhileWaiting()) {
213 79512 : MilliSleep(100);
214 : }
215 79512 : }
216 :
217 10 : if (nextPhase == QuorumPhase_Initialized) {
218 10 : quorumDKGDebugManager->ResetLocalSessionStatus(params.type);
219 : } else {
220 0 : quorumDKGDebugManager->UpdateLocalSessionStatus(params.type, [&](CDKGDebugSessionStatus& status) {
221 0 : bool changed = status.phase != (uint8_t) nextPhase;
222 0 : status.phase = (uint8_t) nextPhase;
223 0 : return changed;
224 : });
225 : }
226 :
227 10 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
228 :
229 10 : }
230 :
231 10 : void CDKGSessionHandler::WaitForNewQuorum(const uint256& oldQuorumHash)
232 : {
233 10 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting\n", __func__, params.name);
234 :
235 18718 : while (true) {
236 9364 : if (stopRequested || ShutdownRequested()) {
237 7 : throw AbortPhaseException();
238 : }
239 9357 : auto currState = GetPhaseAndQuorumHash();
240 9357 : if (currState.quorumHash != oldQuorumHash) {
241 : break;
242 : }
243 9354 : MilliSleep(100);
244 9354 : }
245 :
246 3 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done\n", __func__, params.name);
247 3 : }
248 :
249 : // Sleep some time to not fully overload the whole network
250 0 : void CDKGSessionHandler::SleepBeforePhase(QuorumPhase curPhase,
251 : const uint256& expectedQuorumHash,
252 : double randomSleepFactor,
253 : const WhileWaitFunc& runWhileWaiting)
254 : {
255 0 : 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 0 : 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 0 : 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 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - starting, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
326 :
327 0 : SleepBeforePhase(curPhase, expectedQuorumHash, randomSleepFactor, runWhileWaiting);
328 0 : startPhaseFunc();
329 0 : WaitForNextPhase(curPhase, nextPhase, expectedQuorumHash, runWhileWaiting);
330 :
331 0 : LogPrint(BCLog::DKG, "CDKGSessionHandler::%s -- %s - done, curPhase=%d, nextPhase=%d\n", __func__, params.name, curPhase, nextPhase);
332 0 : }
333 :
334 : // returns a set of NodeIds which sent invalid messages
335 : template<typename Message>
336 0 : std::set<NodeId> BatchVerifyMessageSigs(CDKGSession& session, const std::vector<std::pair<NodeId, std::shared_ptr<Message>>>& messages)
337 : {
338 0 : if (messages.empty()) {
339 0 : return {};
340 : }
341 :
342 0 : std::set<NodeId> ret;
343 0 : bool revertToSingleVerification = false;
344 :
345 0 : CBLSSignature aggSig;
346 0 : std::vector<CBLSPublicKey> pubKeys;
347 0 : std::vector<uint256> messageHashes;
348 0 : std::set<uint256> messageHashesSet;
349 0 : pubKeys.reserve(messages.size());
350 0 : messageHashes.reserve(messages.size());
351 0 : bool first = true;
352 0 : for (const auto& p : messages ) {
353 0 : const auto& msg = *p.second;
354 :
355 0 : auto member = session.GetMember(msg.proTxHash);
356 0 : if (!member) {
357 : // should not happen as it was verified before
358 0 : ret.emplace(p.first);
359 0 : continue;
360 : }
361 :
362 0 : if (first) {
363 0 : aggSig = msg.sig;
364 : } else {
365 0 : aggSig.AggregateInsecure(msg.sig);
366 : }
367 0 : first = false;
368 :
369 0 : auto msgHash = msg.GetSignHash();
370 0 : 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 0 : revertToSingleVerification = true;
378 0 : break;
379 : }
380 :
381 0 : pubKeys.emplace_back(member->dmn->pdmnState->pubKeyOperator.Get());
382 0 : messageHashes.emplace_back(msgHash);
383 : }
384 : if (!revertToSingleVerification) {
385 0 : bool valid = aggSig.VerifyInsecureAggregated(pubKeys, messageHashes);
386 0 : if (valid) {
387 : // all good
388 0 : return ret;
389 : }
390 :
391 : // are all messages from the same node?
392 : NodeId firstNodeId;
393 0 : first = true;
394 0 : bool nodeIdsAllSame = true;
395 0 : for (auto it = messages.begin(); it != messages.end(); ++it) {
396 : if (first) {
397 0 : 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 0 : ret.emplace(firstNodeId);
409 0 : return ret;
410 : }
411 : // different nodes, let's figure out who are the bad ones
412 : }
413 :
414 0 : for (const auto& p : messages) {
415 0 : if (ret.count(p.first)) {
416 0 : continue;
417 : }
418 :
419 0 : const auto& msg = *p.second;
420 0 : auto member = session.GetMember(msg.proTxHash);
421 0 : bool valid = msg.sig.VerifyInsecure(member->dmn->pdmnState->pubKeyOperator.Get(), msg.GetSignHash());
422 0 : if (!valid) {
423 0 : ret.emplace(p.first);
424 : }
425 : }
426 0 : return ret;
427 : }
428 :
429 : template<typename Message>
430 0 : static bool ProcessPendingMessageBatch(CDKGSession& session, CDKGPendingMessages& pendingMessages, size_t maxCount)
431 : {
432 0 : auto msgs = pendingMessages.PopAndDeserializeMessages<Message>(maxCount);
433 0 : if (msgs.empty()) {
434 : return false;
435 : }
436 :
437 0 : std::vector<uint256> hashes;
438 0 : std::vector<std::pair<NodeId, std::shared_ptr<Message>>> preverifiedMessages;
439 0 : hashes.reserve(msgs.size());
440 0 : preverifiedMessages.reserve(msgs.size());
441 :
442 0 : for (const auto& p : msgs) {
443 0 : 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 0 : const auto& msg = *p.second;
452 :
453 0 : bool ban = false;
454 0 : 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 0 : hashes.emplace_back(::SerializeHash(msg));
466 0 : preverifiedMessages.emplace_back(p);
467 : }
468 0 : if (preverifiedMessages.empty()) {
469 : return true;
470 : }
471 :
472 0 : auto badNodes = BatchVerifyMessageSigs(session, preverifiedMessages);
473 0 : if (!badNodes.empty()) {
474 0 : LOCK(cs_main);
475 0 : for (auto nodeId : badNodes) {
476 0 : LogPrint(BCLog::NET, "%s -- failed to verify signature, peer=%d\n", __func__, nodeId);
477 0 : Misbehaving(nodeId, 100);
478 : }
479 : }
480 :
481 0 : for (size_t i = 0; i < preverifiedMessages.size(); i++) {
482 0 : NodeId nodeId = preverifiedMessages[i].first;
483 0 : if (badNodes.count(nodeId)) {
484 0 : continue;
485 : }
486 0 : const auto& msg = *preverifiedMessages[i].second;
487 0 : bool ban = false;
488 0 : session.ReceiveMessage(hashes[i], msg, ban);
489 0 : 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 0 : return true;
498 : }
499 :
500 288 : void CDKGSessionHandler::HandleDKGRound()
501 : {
502 288 : uint256 curQuorumHash;
503 :
504 566 : WaitForNextPhase(QuorumPhase_None, QuorumPhase_Initialized, UINT256_ZERO, []{return false;});
505 :
506 10 : {
507 10 : LOCK(cs);
508 10 : pendingContributions.Clear();
509 10 : pendingComplaints.Clear();
510 10 : pendingJustifications.Clear();
511 10 : pendingPrematureCommitments.Clear();
512 10 : curQuorumHash = quorumHash;
513 : }
514 :
515 20 : const CBlockIndex* pindexQuorum = WITH_LOCK(cs_main, return LookupBlockIndex(curQuorumHash));
516 10 : 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 10 : if (!InitNewQuorum(pindexQuorum)) {
523 : // should actually never happen
524 10 : WaitForNewQuorum(curQuorumHash);
525 3 : throw AbortPhaseException();
526 : }
527 :
528 0 : quorumDKGDebugManager->UpdateLocalSessionStatus(params.type, [&](CDKGDebugSessionStatus& status) {
529 0 : bool changed = status.phase != (uint8_t) QuorumPhase_Initialized;
530 0 : status.phase = (uint8_t) QuorumPhase_Initialized;
531 0 : return changed;
532 : });
533 :
534 0 : EnsureQuorumConnections(params.type, pindexQuorum, curSession->myProTxHash);
535 0 : if (curSession->AreWeMember()) {
536 0 : AddQuorumProbeConnections(params.type, pindexQuorum, curSession->myProTxHash);
537 : }
538 :
539 0 : WaitForNextPhase(QuorumPhase_Initialized, QuorumPhase_Contribute, curQuorumHash, []{return false;});
540 :
541 : // Contribute
542 0 : auto fContributeStart = [this]() {
543 0 : curSession->Contribute(pendingContributions);
544 0 : };
545 0 : auto fContributeWait = [this] {
546 0 : return ProcessPendingMessageBatch<CDKGContribution>(*curSession, pendingContributions, 8);
547 0 : };
548 0 : HandlePhase(QuorumPhase_Contribute, QuorumPhase_Complain, curQuorumHash, 0.05, fContributeStart, fContributeWait);
549 :
550 : // Complain
551 0 : auto fComplainStart = [this]() {
552 0 : curSession->VerifyAndComplain(pendingComplaints);
553 0 : };
554 0 : auto fComplainWait = [this] {
555 0 : return ProcessPendingMessageBatch<CDKGComplaint>(*curSession, pendingComplaints, 8);
556 0 : };
557 0 : HandlePhase(QuorumPhase_Complain, QuorumPhase_Justify, curQuorumHash, 0.05, fComplainStart, fComplainWait);
558 :
559 : // Justify
560 0 : auto fJustifyStart = [this]() {
561 0 : curSession->VerifyAndJustify(pendingJustifications);
562 0 : };
563 0 : auto fJustifyWait = [this] {
564 0 : return ProcessPendingMessageBatch<CDKGJustification>(*curSession, pendingJustifications, 8);
565 0 : };
566 0 : HandlePhase(QuorumPhase_Justify, QuorumPhase_Commit, curQuorumHash, 0.05, fJustifyStart, fJustifyWait);
567 :
568 : // Commit
569 0 : auto fCommitStart = [this]() {
570 0 : curSession->VerifyAndCommit(pendingPrematureCommitments);
571 0 : };
572 0 : auto fCommitWait = [this] {
573 0 : return ProcessPendingMessageBatch<CDKGPrematureCommitment>(*curSession, pendingPrematureCommitments, 8);
574 0 : };
575 0 : HandlePhase(QuorumPhase_Commit, QuorumPhase_Finalize, curQuorumHash, 0.1, fCommitStart, fCommitWait);
576 :
577 0 : auto finalCommitments = curSession->FinalizeCommitments();
578 0 : for (const auto& fqc : finalCommitments) {
579 0 : quorumBlockProcessor->AddAndRelayMinableCommitment(fqc);
580 : }
581 : }
582 :
583 285 : void CDKGSessionHandler::PhaseHandlerThread()
584 : {
585 573 : while (!stopRequested && !ShutdownRequested()) {
586 288 : try {
587 288 : HandleDKGRound();
588 576 : } catch (AbortPhaseException& e) {
589 288 : quorumDKGDebugManager->UpdateLocalSessionStatus(params.type, [&](CDKGDebugSessionStatus& status) {
590 0 : status.aborted = true;
591 0 : return true;
592 : });
593 288 : LogPrintf("CDKGSessionHandler::%s -- aborted current DKG session for llmq=%s\n", __func__, params.name);
594 : }
595 : }
596 285 : }
597 :
598 : }
|
