From 0151968f1c497a918efb5cca2b78a5bc20b87d5a Mon Sep 17 00:00:00 2001 From: cafebedouin Date: Thu, 17 Sep 2026 10:45:07 -0500 Subject: [PATCH 1/4] Block announcements: derive difficulty only from a parent bound to the best chain; request an unknown parent from the sender only Input block announcements (processInputBlock) and ordering block announcements (processOrderingBlockAnnouncement) are validated against the difficulty derived from their parent header, and only when that parent is known, is in the best header chain, and is one block below the announced header (at genesis height, the configured initial difficulty). The announced header's own nBits is never used. Policy for an announcement whose parent is unknown or not bound this way: it is not processed, stored or relayed, and there is no penalty. An unknown parent header is requested from the announcing peer only, once, with a tracked request that expires after the delivery timeout instead of running a delivery check (requestHeaderFromSenderOnly / SenderOnlyRequestExpired): no other peer is asked for a header only the sender claimed to know, and no peer is penalized for not delivering it. The expiration carries the timer the delivery tracker stores for the request, so an expiration queued for an earlier attempt does not clear a newer request for the same header. Tests: InputBlockParentBindingSpec and OrderingBlockAnnouncementParentCheckSpec, with real Autolykos proof-of-work; the three regression cases from the review of #2552 are included as posted. Supersedes #2552 and #2553. Co-Authored-By: Claude Fable 5.1 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_0158ircj1NFMWMqGsKmhmaVw --- .../network/ErgoNodeViewSynchronizer.scala | 81 ++- .../ErgoNodeViewSynchronizerMessages.scala | 14 + .../network/InputBlockParentBindingSpec.scala | 462 ++++++++++++++++++ ...ringBlockAnnouncementParentCheckSpec.scala | 428 ++++++++++++++++ 4 files changed, 977 insertions(+), 8 deletions(-) create mode 100644 src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala create mode 100644 src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala index 2cc3fffd27..540dc9f5bb 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala @@ -616,6 +616,22 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, } } + /** + * Request a header from `peer` only, once. The request is tracked, so the header is accepted if `peer` delivers + * it before the delivery timeout. After the timeout the request is forgotten instead of checked: no other peer is + * asked for a header that only `peer` claimed to know, and no peer is penalized for not delivering it. + */ + protected def requestHeaderFromSenderOnly(headerId: ModifierId, peer: ConnectedPeer): Unit = { + val hid = Header.modifierTypeId + log.debug(s"Requesting header $headerId from $peer only") + networkControllerRef ! SendToNetwork(Message(RequestModifierSpec, Right(InvData(hid, Seq(headerId))), None), SendToPeer(peer)) + deliveryTracker.setRequested(hid, headerId, peer) { _ => + val expiration = new SenderOnlyRequestExpired(hid, headerId) + expiration.timer = context.system.scheduler.scheduleOnce(deliveryTimeout, self, expiration) + expiration.timer + } + } + /* * Private helper methods to request UTXO set snapshots metadata and related data (manifests, chunks) from peers */ @@ -1487,10 +1503,30 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, if (subBlockHeader.height == hr.fullBlockHeight + 1) { val powScheme = settings.chainSettings.powScheme val parentHeaderOpt = hr.modifierById(subBlockHeader.parentId).collect { case h: Header => h } - val expectedNBits: Option[Long] = parentHeaderOpt.map { parent => - val expectedDiff = hr.requiredDifficultyAfter(parent) - import org.ergoplatform.mining.difficulty.DifficultySerializer - DifficultySerializer.encodeCompactBits(expectedDiff) + // Expected difficulty comes only from a known parent in the best chain, one block below the announced header + // (or from the configured initial difficulty at genesis height), never from the announced header itself + val expectedNBits: Option[Long] = if (subBlockHeader.isGenesis) { + if (hr.bestFullBlockIdOpt.isEmpty) Some(settings.chainSettings.initialNBits) else None + } else { + parentHeaderOpt + .filter(parent => subBlockHeader.height == parent.height + 1 && hr.isInBestChain(parent)) + .map { parent => + val expectedDiff = hr.requiredDifficultyAfter(parent) + import org.ergoplatform.mining.difficulty.DifficultySerializer + DifficultySerializer.encodeCompactBits(expectedDiff) + } + } + if (expectedNBits.isEmpty) { + // Policy point: input block whose parent is unknown, or known but not bound as above. + // Default: do not process it; if the parent header is unknown, request it from the sender only, once. + // Another policy (e.g. keeping the input block until its parent arrives) can replace this branch. + if (parentHeaderOpt.isEmpty && !subBlockHeader.isGenesis) { + if (deliveryTracker.status(subBlockHeader.parentId, Header.modifierTypeId, Seq(hr)) == ModifiersStatus.Unknown) { + requestHeaderFromSenderOnly(subBlockHeader.parentId, remote) + } + } + log.info(s"Not processing input block $subBlockId: parent ${subBlockHeader.parentId} is not bound to the best chain") + return } val valid = usrOpt .map(_.stateContext.currentParameters) @@ -1812,10 +1848,31 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, if (!hr.contains(oba.header.id)) { val parentHeaderOpt = hr.modifierById(oba.header.parentId).collect { case h: Header => h } - val expectedNBits: Option[Long] = parentHeaderOpt.map { parent => - val expectedDiff = hr.requiredDifficultyAfter(parent) - import org.ergoplatform.mining.difficulty.DifficultySerializer - DifficultySerializer.encodeCompactBits(expectedDiff) + // Expected difficulty comes only from a known parent in the best chain, one block below the announced header + // (or from the configured initial difficulty at genesis height), never from the announced header itself + val expectedNBits: Option[Long] = if (oba.header.isGenesis) { + if (hr.bestHeaderOpt.isEmpty) Some(settings.chainSettings.initialNBits) else None + } else { + parentHeaderOpt + .filter(parent => oba.header.height == parent.height + 1 && hr.isInBestChain(parent)) + .map { parent => + val expectedDiff = hr.requiredDifficultyAfter(parent) + import org.ergoplatform.mining.difficulty.DifficultySerializer + DifficultySerializer.encodeCompactBits(expectedDiff) + } + } + + if (expectedNBits.isEmpty) { + // Policy point: announcement whose parent is unknown, or known but not bound as above. + // Default: do not store, relay or process it; if the parent header is unknown, request it from the sender only, once. + // Another policy (e.g. keeping the announcement until its parent arrives) can replace this branch. + if (parentHeaderOpt.isEmpty && !oba.header.isGenesis) { + if (deliveryTracker.status(oba.header.parentId, Header.modifierTypeId, Seq(hr)) == ModifiersStatus.Unknown) { + requestHeaderFromSenderOnly(oba.header.parentId, remote) + } + } + log.info(s"Not processing ordering block announcement ${oba.header.id}: parent ${oba.header.parentId} is not bound to the best chain") + return } if (!oba.valid(settings.chainSettings.powScheme, expectedNBits)) { @@ -1897,6 +1954,14 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, * re-request modifier from a different random peer, if our node does not know a peer who have it */ protected def checkDelivery(hr: ErgoHistory): Receive = { + case expiration: SenderOnlyRequestExpired => + // clear the request only if this expiration belongs to the current request attempt + val current = deliveryTracker.getRequestedInfo(expiration.modifierTypeId, expiration.modifierId) + if (current.exists(_.cancellable eq expiration.timer)) { + log.info(s"Peer ${current.get.peer} has not delivered ${expiration.modifierTypeId} : ${expiration.modifierId} on time, forgetting the request") + deliveryTracker.clearStatusForModifier(expiration.modifierId, expiration.modifierTypeId, ModifiersStatus.Requested) + } + case CheckDelivery(peer, modifierTypeId, modifierId) => if (deliveryTracker.status(modifierId, modifierTypeId, Seq.empty) == ModifiersStatus.Requested) { // If transaction not delivered on time, we just forget about it. diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala index 044db7a7b4..c738218633 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala @@ -8,6 +8,7 @@ import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader import org.ergoplatform.nodeView.state.{ErgoStateReader, UtxoStateReader} import org.ergoplatform.nodeView.wallet.ErgoWalletReader import scorex.core.network.ConnectedPeer +import akka.actor.Cancellable import scorex.util.ModifierId import org.ergoplatform.ErgoLikeContext.Height import org.ergoplatform.modifiers.history.popow.NipopowProof @@ -33,6 +34,19 @@ object ErgoNodeViewSynchronizerMessages { modifierTypeId: NetworkObjectTypeId.Value, modifierId: ModifierId) + /** + * Expiration of a header request sent to one peer only (see `requestHeaderFromSenderOnly`): after the + * delivery timeout the request is forgotten, without asking another peer and without penalizing anyone. + * + * `timer` is the scheduled expiration itself, the `Cancellable` the delivery tracker stores for the request, + * so an expiration is matched to the request attempt that scheduled it: an expiration queued for an earlier + * attempt does not clear a newer request for the same header. + */ + final class SenderOnlyRequestExpired(val modifierTypeId: NetworkObjectTypeId.Value, val modifierId: ModifierId) { + /** set right after scheduling, before the expiration can be handled; unset, it matches no request */ + @volatile var timer: Cancellable = _ + } + trait PeerManagerEvent case class HandshakedPeer(remote: ConnectedPeer) extends PeerManagerEvent diff --git a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala new file mode 100644 index 0000000000..6354d8ab90 --- /dev/null +++ b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala @@ -0,0 +1,462 @@ +package org.ergoplatform.network + +import akka.actor.{ActorRef, Props} +import akka.testkit.{TestActorRef, TestProbe} +import org.ergoplatform.AutolykosSolution +import org.ergoplatform.consensus.Older +import org.ergoplatform.mining.{AutolykosPowScheme, InputBlockFields} +import org.ergoplatform.mining.difficulty.DifficultySerializer +import org.ergoplatform.modifiers.ErgoFullBlock +import org.ergoplatform.modifiers.history.header.{Header, HeaderSerializer} +import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._ +import org.ergoplatform.network.message.{InvData, Message, ModifiersData, ModifiersSpec, RequestModifierSpec} +import org.ergoplatform.network.peer.PeerInfo +import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.ModifiersFromRemote +import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryUtils, ErgoSyncInfoMessageSpec} +import org.ergoplatform.nodeView.mempool.ErgoMemPool +import org.ergoplatform.nodeView.state.StateType +import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState +import org.ergoplatform.settings.ErgoSettings +import org.ergoplatform.subblocks.InputBlockAnnouncement +import org.ergoplatform.wallet.utils.FileUtils +import org.scalatest.matchers.should.Matchers +import org.scalatest.propspec.AnyPropSpec +import scorex.core.network.NetworkController.ReceivableMessages.{PenalizePeer, SendToNetwork} +import scorex.core.network.{ConnectedPeer, DeliveryTracker, ModifiersStatus, SendToPeer} +import scorex.testkit.utils.AkkaFixture +import scorex.util.{ModifierId, bytesToId} + +import scala.concurrent.duration._ +import scala.concurrent.{Await, ExecutionContextExecutor} + +/** + * An input block announcement is processed only if its parent header is known, is in the best chain, and is + * exactly one block below the announced header (at genesis height, with no full blocks yet, the configured + * initial difficulty is used). The expected difficulty is derived from that parent. Otherwise the announcement is + * not processed, and an unknown parent header is requested from the sender only, once: the request expires after + * the delivery timeout without asking another peer or penalizing anyone. + * + * The synchronizer under test validates proof-of-work with the real Autolykos scheme (the default test + * configuration uses a fake scheme that accepts any header), while the local history is built with the + * test configuration as usual. + */ +class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUtils { + import org.ergoplatform.utils.ErgoCoreTestConstants._ + import org.ergoplatform.utils.ErgoNodeTestConstants._ + import org.ergoplatform.utils.generators.ChainGenerator._ + import org.ergoplatform.utils.generators.ConnectedPeerGenerators._ + import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._ + + class SynchronizerUnderTest(networkControllerRef: ActorRef, + viewHolderRef: ActorRef, + settings: ErgoSettings, + syncTracker: ErgoSyncTracker, + deliveryTracker: DeliveryTracker)(implicit ec: ExecutionContextExecutor) + extends ErgoNodeViewSynchronizer(networkControllerRef, viewHolderRef, ErgoSyncInfoMessageSpec, + settings, syncTracker, deliveryTracker)(ec) + + /** + * @param initialDifficultyHex overrides the test chain's difficulty (the local chain keeps it) + * @param applyLocalChain whether the generated chain is applied to the local history + */ + class Fixture(initialDifficultyHex: Option[String] = None, + applyLocalChain: Boolean = true, + requestTimeout: FiniteDuration = 2.seconds) extends AkkaFixture { + implicit val ec: ExecutionContextExecutor = system.dispatcher + + val historySettings: ErgoSettings = { + val s = settings.copy(directory = createTempDir.getAbsolutePath) + initialDifficultyHex.fold(s)(hex => s.copy(chainSettings = s.chainSettings.copy(initialDifficultyHex = hex))) + } + + private val cs = historySettings.chainSettings + val realPowScheme = new AutolykosPowScheme(cs.powScheme.k, cs.powScheme.n) + // Expiration tests use a short timeout; replay tests keep unrelated deadlines outside their window. + val synchronizerSettings: ErgoSettings = + historySettings.copy(chainSettings = cs.copy(powScheme = realPowScheme), + scorexSettings = historySettings.scorexSettings.copy( + network = historySettings.scorexSettings.network.copy(deliveryTimeout = requestTimeout))) + + val ncProbe = TestProbe("NetworkControllerProbe") + val viewHolderProbe = TestProbe("ViewHolderProbe") + val pchProbe = TestProbe("PeerHandlerProbe") + + val syncTracker = ErgoSyncTracker(synchronizerSettings.scorexSettings.network) + val deliveryTracker: DeliveryTracker = DeliveryTracker.empty(synchronizerSettings) + + val synchronizer: TestActorRef[SynchronizerUnderTest] = TestActorRef(Props( + new SynchronizerUnderTest(ncProbe.ref, viewHolderProbe.ref, synchronizerSettings, syncTracker, deliveryTracker))) + + val peer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, pchProbe.ref, + Some(PeerInfo(defaultPeerSpec, System.currentTimeMillis()))) + + val hist: ErgoHistory = ErgoHistory.readOrGenerate(historySettings)(null) + val chain: Seq[ErgoFullBlock] = genChain(3, hist, nBits = historySettings.chainSettings.initialNBits) + if (applyLocalChain) applyChain(hist, chain) + + val state: WrappedUtxoState = + boxesHolderGen.map(WrappedUtxoState(_, createTempDir, parameters, historySettings)).sample.get + val mempool: ErgoMemPool = ErgoMemPool.empty(historySettings) + + def process(ib: InputBlockAnnouncement): Unit = + synchronizer.underlyingActor.processInputBlock(ib, hist, mempool, peer, Some(state)) + + /** Makes the synchronizer handle scheduled and network messages, with another peer that headers could be asked from. */ + def initialize(): ConnectedPeer = { + val otherPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("OtherPeer").ref, + Some(PeerInfo(defaultPeerSpec.copy(features = Seq(ModePeerFeature(StateType.Utxo, verifyingTransactions = true, None, -1))), + System.currentTimeMillis()))) + syncTracker.updateStatus(otherPeer, Older, Some(hist.fullBlockHeight + 10)) + synchronizer ! ChangedState(state) + synchronizer ! ChangedHistory(hist) + synchronizer ! ChangedMempool(mempool) + Thread.sleep(300) + viewHolderProbe.receiveWhile(max = 300.millis, idle = 100.millis) { case m => m } + ncProbe.receiveWhile(max = 300.millis, idle = 100.millis) { case m => m } + otherPeer + } + + def deliverHeader(h: Header): Unit = synchronizer ! Message(ModifiersSpec, + Left(ModifiersSpec.toBytes(ModifiersData(Header.modifierTypeId, Map(h.id -> HeaderSerializer.toBytes(h))))), Some(peer)) + } + + private def withFixture(test: Fixture => Any): Unit = withFixture(new Fixture)(test) + + private def withFixture(f: Fixture)(test: Fixture => Any): Unit = + try test(f) finally Await.result(f.system.terminate(), Duration.Inf) + + // difficulty 2^80: a zero nonce does not meet the target + private val highDifficultyHex = "01" + "00" * 10 + + /** + * Builds an input block announcement at `height` naming `parentId`, with a version 2 header declaring + * `nBits` and a Merkle proof consistent with the header's extension root. The nonce is left at zero. + */ + private def announcement(f: Fixture, parentId: ModifierId, height: Int, nBits: Long): InputBlockAnnouncement = { + import org.ergoplatform.modifiers.history.extension.ExtensionCandidate + import scorex.crypto.hash.Digest32 + + val txDigest = Digest32 @@ Array.fill(32)(0.toByte) + val prevTxDigest = Digest32 @@ Array.fill(32)(0.toByte) + val ibExtension = InputBlockFields.toExtensionFields(None, txDigest, prevTxDigest) + val block = nextBlock(f.hist.bestFullBlockOpt, f.chain.head.blockTransactions.txs, defaultExtension ++ ibExtension) + val fields = block.extension.fields + val proof = ExtensionCandidate(fields).proofForInputBlockData.get + + val sol = block.header.powSolution + val header = block.header.copy( + version = Header.Interpreter60Version, + parentId = parentId, + height = height, + nBits = nBits, + powSolution = new AutolykosSolution(sol.pk, sol.w, Array.fill(8)(0: Byte), sol.d) + ) + InputBlockAnnouncement(InputBlockAnnouncement.initialMessageVersion, header, + new InputBlockFields(None, txDigest, prevTxDigest, proof), None) + } + + private def requiredNBitsAfter(f: Fixture, parent: Header): Long = + DifficultySerializer.encodeCompactBits(f.hist.requiredDifficultyAfter(parent)) + + private def viewHolderGotInputBlock(f: Fixture): Boolean = + f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } + .exists(_.isInstanceOf[ProcessInputBlock]) + + private def networkMessages(f: Fixture): Seq[Any] = + f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } + + private def headerRequests(msgs: Seq[Any]): Seq[(Seq[ModifierId], Any)] = msgs.collect { + case SendToNetwork(msg, strategy) if msg.spec.messageCode == RequestModifierSpec.messageCode && + msg.data.get.asInstanceOf[InvData].typeId == Header.modifierTypeId => + msg.data.get.asInstanceOf[InvData].ids -> strategy + } + + private def penalized(msgs: Seq[Any]): Boolean = msgs.exists(_.isInstanceOf[PenalizePeer]) + + property("input block with unknown parent header is dropped, and the parent header is requested from the sender") { + withFixture { f => + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val tip = f.hist.bestFullBlockOpt.get.header + val ib = announcement(f, unknownParent, tip.height + 1, DifficultySerializer.encodeCompactBits(1)) + + // the announced header on its own passes the proof-of-work and Merkle checks + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, None) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + val msgs = networkMessages(f) + penalized(msgs) shouldBe false + headerRequests(msgs) shouldBe Seq(Seq(unknownParent) -> SendToPeer(f.peer)) + msgs.collect { case s: SendToNetwork => s }.size shouldBe 1 + // the request is tracked, so the header is accepted when it arrives + f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested + } + } + + property("unknown parent header: the header delivered by the sender is accepted") { + withFixture { f => + f.initialize() + val tip = f.hist.bestFullBlockOpt.get + // a real header that is not in local history: a sibling of the best full block + val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header + f.process(announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1))) + headerRequests(networkMessages(f)) shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) + + f.deliverHeader(sibling) + + penalized(networkMessages(f)) shouldBe false + f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { + case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) + case _ => false + } shouldBe true + } + } + + property("input block whose known parent is an old header is dropped without requests") { + withFixture { f => + val oldParent = f.chain.head.header + val tip = f.hist.bestFullBlockOpt.get.header + oldParent.id should not be tip.id + val ib = announcement(f, oldParent.id, tip.height + 1, requiredNBitsAfter(f, oldParent)) + + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(requiredNBitsAfter(f, oldParent))) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + networkMessages(f) shouldBe empty + } + } + + property("an input announcement is accepted on replay after its parent joins the best header chain") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + val originalTip = f.hist.bestFullBlockOpt.get + val parent = nextBlock(Some(f.chain(1)), originalTip.blockTransactions.txs, defaultExtension) + parent.height shouldBe originalTip.height + f.hist.contains(parent.header) shouldBe false + val ib = announcement(f, parent.id, originalTip.height + 1, requiredNBitsAfter(f, parent.header)) + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, + Some(requiredNBitsAfter(f, parent.header))) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + val initialMessages = networkMessages(f) + headerRequests(initialMessages) shouldBe Seq(Seq(parent.id) -> SendToPeer(f.peer)) + penalized(initialMessages) shouldBe false + + f.hist.append(parent.header).get + f.hist.isInBestChain(parent.header) shouldBe false + val nextHeader = nextBlock(Some(parent), originalTip.blockTransactions.txs, defaultExtension).header + f.hist.append(nextHeader).get + f.hist.isInBestChain(parent.header) shouldBe true + f.hist.bestFullBlockIdOpt shouldBe Some(originalTip.id) + ib.header.height shouldBe f.hist.fullBlockHeight + 1 + + f.process(ib) + + f.viewHolderProbe.expectMsg(ProcessInputBlock(ib, f.peer)) + val replayMessages = networkMessages(f) + penalized(replayMessages) shouldBe false + headerRequests(replayMessages) shouldBe empty + } + } + + property("an input announcement is accepted on replay after its parent becomes the best full block in history") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + val originalTip = f.hist.bestFullBlockOpt.get + val parent = nextBlock(Some(originalTip), originalTip.blockTransactions.txs, defaultExtension) + f.hist.contains(parent.header) shouldBe false + val ib = announcement(f, parent.id, parent.height + 1, requiredNBitsAfter(f, parent.header)) + ib.header.height shouldBe f.hist.fullBlockHeight + 2 + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, + Some(requiredNBitsAfter(f, parent.header))) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + val initialMessages = networkMessages(f) + headerRequests(initialMessages) shouldBe Seq(Seq(parent.id) -> SendToPeer(f.peer)) + penalized(initialMessages) shouldBe false + + applyChain(f.hist, Seq(parent)) + f.hist.bestFullBlockIdOpt shouldBe Some(parent.id) + f.hist.isInBestChain(parent.header) shouldBe true + ib.header.height shouldBe f.hist.fullBlockHeight + 1 + + f.process(ib) + + f.viewHolderProbe.expectMsg(ProcessInputBlock(ib, f.peer)) + val replayMessages = networkMessages(f) + penalized(replayMessages) shouldBe false + headerRequests(replayMessages) shouldBe empty + } + } + + property("unknown-parent discovery expires without requesting the header from another peer") { + withFixture { f => + val otherPeer = f.initialize() + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val tip = f.hist.bestFullBlockOpt.get.header + val ib = announcement(f, unknownParent, tip.height + 1, DifficultySerializer.encodeCompactBits(1)) + + f.process(ib) + + // Observe two delivery deadlines, including a potential retry against the other peer. + val messages = f.ncProbe.receiveWhile(max = 5.seconds, idle = 5.seconds) { case m => m } + val requests = headerRequests(messages).filter(_._1.contains(unknownParent)) + requests.map(_._2 == SendToPeer(f.peer)) shouldBe Seq(true) + messages.exists { + case p: PenalizePeer => p.address == otherPeer.connectionId.remoteAddress + case _ => false + } shouldBe false + f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, unknownParent) shouldBe None + f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Unknown + viewHolderGotInputBlock(f) shouldBe false + } + } + + property("an expiration from an earlier request attempt does not clear the current request for the header") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + f.initialize() + val tip = f.hist.bestFullBlockOpt.get + val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header + val ib = announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1)) + f.process(ib) + headerRequests(networkMessages(f)) shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) + val first = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get + + // the first request is cleared (as when its expiration is already queued) and the header is requested again + f.deliveryTracker.setUnknown(sibling.id, Header.modifierTypeId) + f.process(ib) + headerRequests(networkMessages(f)) shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) + val current = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get + current should not be theSameInstanceAs(first) + + // the first attempt's expiration arrives now + val stale = new SenderOnlyRequestExpired(Header.modifierTypeId, sibling.id) + stale.timer = first.cancellable + f.synchronizer ! stale + + networkMessages(f) shouldBe empty + f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get should be theSameInstanceAs current + f.deliveryTracker.status(sibling.id, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested + + f.deliverHeader(sibling) + + penalized(networkMessages(f)) shouldBe false + f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { + case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) + case _ => false + } shouldBe true + } + } + + property("input block whose parent is a known header outside the best chain is dropped") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get + val forkBlock = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension) + f.hist.append(forkBlock.header).get + forkBlock.header.height shouldBe tip.header.height + forkBlock.header.id should not be tip.id + f.hist.bestFullBlockIdOpt shouldBe Some(tip.id) + f.hist.isInBestChain(forkBlock.header) shouldBe false + + val ib = announcement(f, forkBlock.header.id, tip.header.height + 1, requiredNBitsAfter(f, forkBlock.header)) + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(requiredNBitsAfter(f, forkBlock.header))) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + networkMessages(f) shouldBe empty + } + } + + property("input block extending the best chain with the required difficulty is still processed") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get.header + val ib = announcement(f, tip.id, tip.height + 1, requiredNBitsAfter(f, tip)) + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe true + penalized(networkMessages(f)) shouldBe false + } + } + + property("input block extending the best chain with a different difficulty is rejected and penalized") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get.header + val required = f.hist.requiredDifficultyAfter(tip) + val ib = announcement(f, tip.id, tip.height + 1, DifficultySerializer.encodeCompactBits(required * 2)) + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + penalized(networkMessages(f)) shouldBe true + } + } + + property("input block extending the best chain with the required difficulty but insufficient work is rejected and penalized") { + withFixture(new Fixture(Some(highDifficultyHex))) { f => + val tip = f.hist.bestFullBlockOpt.get.header + val required = requiredNBitsAfter(f, tip) + required shouldBe tip.nBits + val ib = announcement(f, tip.id, tip.height + 1, required) + val params = f.state.stateContext.currentParameters + f.realPowScheme.checkInputBlockPoW(ib.header, params) shouldBe false + ib.valid(f.realPowScheme, params, Some(required)) shouldBe false + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + val msgs = networkMessages(f) + penalized(msgs) shouldBe true + msgs.collect { case s: SendToNetwork => s } shouldBe empty + } + } + + property("input block whose known parent is an old header is dropped without a penalty, even with insufficient work") { + withFixture(new Fixture(Some(highDifficultyHex))) { f => + val oldParent = f.chain.head.header + val tip = f.hist.bestFullBlockOpt.get.header + val ib = announcement(f, oldParent.id, tip.height + 1, requiredNBitsAfter(f, oldParent)) + f.realPowScheme.checkInputBlockPoW(ib.header, f.state.stateContext.currentParameters) shouldBe false + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + networkMessages(f) shouldBe empty + } + } + + property("input block at genesis height with no full blocks is checked against the configured initial difficulty") { + withFixture(new Fixture(applyLocalChain = false)) { f => + f.hist.bestFullBlockIdOpt shouldBe None + val initial = f.historySettings.chainSettings.initialNBits + val ib = announcement(f, Header.GenesisParentId, ErgoHistoryUtils.GenesisHeight, initial) + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe true + penalized(networkMessages(f)) shouldBe false + } + } + + property("input block at genesis height with a different difficulty is rejected and penalized") { + withFixture(new Fixture(Some(highDifficultyHex), applyLocalChain = false)) { f => + // configured initial difficulty 2^80; the announcement declares difficulty 1 + val other = DifficultySerializer.encodeCompactBits(1) + other should not be f.historySettings.chainSettings.initialNBits + val ib = announcement(f, Header.GenesisParentId, ErgoHistoryUtils.GenesisHeight, other) + // on its own, the header meets the difficulty it declares + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, None) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + penalized(networkMessages(f)) shouldBe true + } + } +} diff --git a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala new file mode 100644 index 0000000000..00b9d21ff5 --- /dev/null +++ b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala @@ -0,0 +1,428 @@ +package org.ergoplatform.network + +import akka.actor.{ActorRef, Props} +import akka.testkit.{TestActorRef, TestProbe} +import org.ergoplatform.AutolykosSolution +import org.ergoplatform.consensus.{Equal, Older} +import org.ergoplatform.mining.{AutolykosPowScheme, InputBlockFields} +import org.ergoplatform.mining.difficulty.DifficultySerializer +import org.ergoplatform.modifiers.{ErgoFullBlock, OrderingBlockAnnouncementTypeId} +import org.ergoplatform.modifiers.history.header.{Header, HeaderSerializer} +import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._ +import org.ergoplatform.network.message.inputblocks.{OrderingBlockAnnouncement, OrderingBlockAnnouncementMessageSpec} +import org.ergoplatform.network.message.{InvData, InvSpec, Message, ModifiersData, ModifiersSpec, RequestModifierSpec} +import org.ergoplatform.network.peer.PeerInfo +import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.ModifiersFromRemote +import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryUtils, ErgoSyncInfoMessageSpec} +import org.ergoplatform.nodeView.mempool.ErgoMemPool +import org.ergoplatform.nodeView.state.StateType +import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState +import org.ergoplatform.settings.ErgoSettings +import org.ergoplatform.wallet.utils.FileUtils +import org.scalatest.matchers.should.Matchers +import org.scalatest.propspec.AnyPropSpec +import scorex.core.network.NetworkController.ReceivableMessages.{PenalizePeer, SendToNetwork} +import scorex.core.network.{ConnectedPeer, DeliveryTracker, ModifiersStatus, SendToPeer} +import scorex.testkit.utils.AkkaFixture +import scorex.util.{ModifierId, bytesToId} + +import scala.concurrent.duration._ +import scala.concurrent.{Await, ExecutionContextExecutor} + +/** + * An ordering block announcement is stored, relayed and handed to the node view holder only if its parent + * header is known, is in the best header chain, and is exactly one block below the announced header; the + * expected difficulty is derived from that parent (at genesis height, with an empty header chain, the configured + * initial difficulty is used). If the parent header is unknown, it is requested from the sender only, once: the + * request expires after the delivery timeout without asking another peer or penalizing anyone. + * + * The synchronizer under test validates proof-of-work with the real Autolykos scheme (the default test + * configuration uses a fake scheme that accepts any header), while the local history is built with the + * test configuration as usual. + */ +class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers with FileUtils { + import org.ergoplatform.utils.ErgoCoreTestConstants._ + import org.ergoplatform.utils.ErgoNodeTestConstants._ + import org.ergoplatform.utils.generators.ChainGenerator._ + import org.ergoplatform.utils.generators.ConnectedPeerGenerators._ + import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators._ + + class SynchronizerUnderTest(networkControllerRef: ActorRef, + viewHolderRef: ActorRef, + settings: ErgoSettings, + syncTracker: ErgoSyncTracker, + deliveryTracker: DeliveryTracker)(implicit ec: ExecutionContextExecutor) + extends ErgoNodeViewSynchronizer(networkControllerRef, viewHolderRef, ErgoSyncInfoMessageSpec, + settings, syncTracker, deliveryTracker)(ec) + + /** + * @param initialDifficultyHex overrides the test chain's difficulty (the local chain keeps it) + * @param applyLocalChain whether the generated chain is applied to the local history + * @param requestTimeout delivery timeout of the synchronizer under test + */ + class Fixture(initialDifficultyHex: Option[String] = None, + applyLocalChain: Boolean = true, + requestTimeout: FiniteDuration = 2.seconds) extends AkkaFixture { + implicit val ec: ExecutionContextExecutor = system.dispatcher + + val historySettings: ErgoSettings = { + val s = settings.copy(directory = createTempDir.getAbsolutePath) + initialDifficultyHex.fold(s)(hex => s.copy(chainSettings = s.chainSettings.copy(initialDifficultyHex = hex))) + } + + private val cs = historySettings.chainSettings + val realPowScheme = new AutolykosPowScheme(cs.powScheme.k, cs.powScheme.n) + // a short delivery timeout by default, so that what happens after it can be observed + val synchronizerSettings: ErgoSettings = + historySettings.copy(chainSettings = cs.copy(powScheme = realPowScheme), + scorexSettings = historySettings.scorexSettings.copy( + network = historySettings.scorexSettings.network.copy(deliveryTimeout = requestTimeout))) + + val ncProbe = TestProbe("NetworkControllerProbe") + val viewHolderProbe = TestProbe("ViewHolderProbe") + val pchProbe = TestProbe("PeerHandlerProbe") + val syncTracker = ErgoSyncTracker(synchronizerSettings.scorexSettings.network) + val deliveryTracker: DeliveryTracker = DeliveryTracker.empty(synchronizerSettings) + + val synchronizer: TestActorRef[SynchronizerUnderTest] = TestActorRef(Props( + new SynchronizerUnderTest(ncProbe.ref, viewHolderProbe.ref, synchronizerSettings, syncTracker, deliveryTracker))) + + val peer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, pchProbe.ref, + Some(PeerInfo(defaultPeerSpec, System.currentTimeMillis()))) + + val hist: ErgoHistory = ErgoHistory.readOrGenerate(historySettings)(null) + val chain: Seq[ErgoFullBlock] = genChain(3, hist, nBits = historySettings.chainSettings.initialNBits) + if (applyLocalChain) applyChain(hist, chain) + + // a peer supporting sub-blocks, within the relay height window + val subBlockPeer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("SubBlockPeer").ref, + Some(PeerInfo(PeerSpec(synchronizerSettings.scorexSettings.network.agentName, Version.SubblocksVersion, + synchronizerSettings.scorexSettings.network.nodeName, None, + Seq(ModePeerFeature(StateType.Utxo, verifyingTransactions = true, None, -1))), System.currentTimeMillis()))) + syncTracker.updateStatus(subBlockPeer, Equal, Some(hist.fullBlockHeight + 1)) + + val state: WrappedUtxoState = + boxesHolderGen.map(WrappedUtxoState(_, createTempDir, parameters, historySettings)).sample.get + synchronizer ! ChangedState(state) + synchronizer ! ChangedHistory(hist) + synchronizer ! ChangedMempool(ErgoMemPool.empty(historySettings)) + Thread.sleep(300) + viewHolderProbe.receiveWhile(max = 300.millis, idle = 100.millis) { case m => m } + ncProbe.receiveWhile(max = 300.millis, idle = 100.millis) { case m => m } + + def send(oba: OrderingBlockAnnouncement): Unit = + synchronizer ! Message(OrderingBlockAnnouncementMessageSpec, + Left(OrderingBlockAnnouncementMessageSpec.toBytes(oba)), Some(peer)) + + /** Adds another peer that headers could be asked from. */ + def addOlderPeer(): ConnectedPeer = { + val otherPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("OtherPeer").ref, + Some(PeerInfo(defaultPeerSpec.copy(features = Seq(ModePeerFeature(StateType.Utxo, verifyingTransactions = true, None, -1))), + System.currentTimeMillis()))) + syncTracker.updateStatus(otherPeer, Older, Some(hist.fullBlockHeight + 10)) + otherPeer + } + + def deliverHeader(h: Header): Unit = synchronizer ! Message(ModifiersSpec, + Left(ModifiersSpec.toBytes(ModifiersData(Header.modifierTypeId, Map(h.id -> HeaderSerializer.toBytes(h))))), Some(peer)) + } + + private def withFixture(test: Fixture => Any): Unit = withFixture(new Fixture)(test) + + private def withFixture(f: Fixture)(test: Fixture => Any): Unit = + try test(f) finally Await.result(f.system.terminate(), Duration.Inf) + + // difficulty 2^80: a zero nonce does not meet the target + private val highDifficultyHex = "01" + "00" * 10 + + /** + * Builds an ordering block announcement at `height` naming `parentId`, with a version 2 header declaring + * `nBits` and extension fields consistent with the header's extension root. The nonce is left at zero. + */ + private def announcement(f: Fixture, parentId: ModifierId, height: Int, nBits: Long): OrderingBlockAnnouncement = { + import scorex.crypto.hash.Digest32 + + val digest = Digest32 @@ Array.fill(32)(0.toByte) + val ibExtension = InputBlockFields.toExtensionFields(None, digest, digest) + val block = nextBlock(f.hist.bestFullBlockOpt, f.chain.head.blockTransactions.txs, defaultExtension ++ ibExtension) + val sol = block.header.powSolution + val header = block.header.copy( + version = Header.Interpreter60Version, + parentId = parentId, + height = height, + nBits = nBits, + powSolution = new AutolykosSolution(sol.pk, sol.w, Array.fill(8)(0: Byte), sol.d) + ) + OrderingBlockAnnouncement(OrderingBlockAnnouncement.CurrentVersion, header, Seq.empty, Seq.empty, + block.extension.fields) + } + + private def requiredNBitsAfter(f: Fixture, parent: Header): Long = + DifficultySerializer.encodeCompactBits(f.hist.requiredDifficultyAfter(parent)) + + private case class Outcome(stored: Boolean, relayed: Boolean, handedOff: Boolean, penalized: Boolean, + headerRequests: Seq[(Seq[ModifierId], Any)]) + + private def outcome(f: Fixture, oba: OrderingBlockAnnouncement): Outcome = { + val net = f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } + val vh = f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } + def inv(s: SendToNetwork) = s.message.data.get.asInstanceOf[InvData] + Outcome( + stored = f.hist.getOrderingBlockAnnouncement(oba.header.id).isDefined, + relayed = net.exists { + case s: SendToNetwork => s.message.spec.messageCode == InvSpec.messageCode && + inv(s).typeId == OrderingBlockAnnouncementTypeId.value && inv(s).ids.contains(oba.header.id) + case _ => false + }, + handedOff = vh.exists { + case ProcessOrderingBlock(o) => o.header.id == oba.header.id + case _ => false + }, + penalized = net.exists(_.isInstanceOf[PenalizePeer]), + headerRequests = net.collect { + case s: SendToNetwork if s.message.spec.messageCode == RequestModifierSpec.messageCode && + inv(s).typeId == Header.modifierTypeId => inv(s).ids -> s.sendingStrategy + } + ) + } + + property("ordering block announcement with unknown parent header is not stored, relayed or processed; parent header is requested from the sender") { + withFixture { f => + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val oba = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, DifficultySerializer.encodeCompactBits(1)) + + // the announced header on its own passes the proof-of-work and extension checks + oba.valid(f.realPowScheme, None) shouldBe true + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq(Seq(unknownParent) -> SendToPeer(f.peer))) + // the request is tracked, so the header is accepted when it arrives + f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested + } + } + + property("unknown parent header: the header delivered by the sender is accepted") { + withFixture { f => + f.addOlderPeer() + val tip = f.hist.bestFullBlockOpt.get + // a real header that is not in local history: a sibling of the best full block + val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header + val oba = announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1)) + f.send(oba) + outcome(f, oba).headerRequests shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) + + f.deliverHeader(sibling) + + f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } + .exists(_.isInstanceOf[PenalizePeer]) shouldBe false + f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { + case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) + case _ => false + } shouldBe true + } + } + + property("unknown-parent request expires without requesting the header from another peer or penalizing anyone") { + withFixture { f => + val otherPeer = f.addOlderPeer() + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val oba = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, DifficultySerializer.encodeCompactBits(1)) + + f.send(oba) + + // observe two delivery deadlines, including a potential retry against the other peer + val messages = f.ncProbe.receiveWhile(max = 5.seconds, idle = 5.seconds) { case m => m } + val requests = messages.collect { + case s: SendToNetwork if s.message.spec.messageCode == RequestModifierSpec.messageCode && + s.message.data.get.asInstanceOf[InvData].ids.contains(unknownParent) => s.sendingStrategy + } + requests shouldBe Seq(SendToPeer(f.peer)) + messages.exists { + case p: PenalizePeer => p.address == otherPeer.connectionId.remoteAddress + case _ => false + } shouldBe false + messages.exists(_.isInstanceOf[PenalizePeer]) shouldBe false + f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, unknownParent) shouldBe None + f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Unknown + f.hist.getOrderingBlockAnnouncement(oba.header.id) shouldBe None + } + } + + property("an expiration from an earlier request attempt does not clear the current request for the header") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + f.addOlderPeer() + val tip = f.hist.bestFullBlockOpt.get + val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header + val oba = announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1)) + f.send(oba) + outcome(f, oba).headerRequests shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) + val first = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get + + // the first request is cleared (as when its expiration is already queued) and the header is requested again + f.deliveryTracker.setUnknown(sibling.id, Header.modifierTypeId) + f.send(oba) + outcome(f, oba).headerRequests shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) + val current = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get + current should not be theSameInstanceAs(first) + + // the first attempt's expiration arrives now + val stale = new SenderOnlyRequestExpired(Header.modifierTypeId, sibling.id) + stale.timer = first.cancellable + f.synchronizer ! stale + + outcome(f, oba).headerRequests shouldBe empty + f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get should be theSameInstanceAs current + f.deliveryTracker.status(sibling.id, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested + + f.deliverHeader(sibling) + + f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } + .exists(_.isInstanceOf[PenalizePeer]) shouldBe false + f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { + case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) + case _ => false + } shouldBe true + } + } + + property("ordering block announcement whose known parent is an old header is not stored, relayed or processed") { + withFixture { f => + val oldParent = f.chain.head.header + val oba = announcement(f, oldParent.id, f.hist.fullBlockHeight + 1, requiredNBitsAfter(f, oldParent)) + oba.valid(f.realPowScheme, Some(requiredNBitsAfter(f, oldParent))) shouldBe true + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement whose parent is a known header outside the best chain is not stored, relayed or processed") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get + val forkBlock = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension) + f.hist.append(forkBlock.header).get + forkBlock.header.height shouldBe tip.header.height + f.hist.isInBestChain(forkBlock.header) shouldBe false + + val oba = announcement(f, forkBlock.header.id, tip.header.height + 1, requiredNBitsAfter(f, forkBlock.header)) + oba.valid(f.realPowScheme, Some(requiredNBitsAfter(f, forkBlock.header))) shouldBe true + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement whose height is not parent height + 1 is not stored, relayed or processed") { + withFixture { f => + val parent = f.chain(1).header + f.hist.isInBestChain(parent) shouldBe true + val oba = announcement(f, parent.id, parent.height + 2, requiredNBitsAfter(f, parent)) + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement extending the best chain with the required difficulty is stored, relayed and processed") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get.header + val oba = announcement(f, tip.id, tip.height + 1, requiredNBitsAfter(f, tip)) + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = true, relayed = true, handedOff = true, penalized = false, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement competing with the best full block (parent in best chain) is still accepted") { + withFixture { f => + val parent = f.chain(1).header + val oba = announcement(f, parent.id, parent.height + 1, requiredNBitsAfter(f, parent)) + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = true, relayed = true, handedOff = true, penalized = false, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement extending the best chain with a different difficulty is rejected and penalized") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get.header + val oba = announcement(f, tip.id, tip.height + 1, + DifficultySerializer.encodeCompactBits(f.hist.requiredDifficultyAfter(tip) * 2)) + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = true, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement extending the best chain with the required difficulty but insufficient work is rejected and penalized") { + withFixture(new Fixture(Some(highDifficultyHex))) { f => + val tip = f.hist.bestFullBlockOpt.get.header + val required = requiredNBitsAfter(f, tip) + required shouldBe tip.nBits + val oba = announcement(f, tip.id, tip.height + 1, required) + f.realPowScheme.validate(oba.header).isSuccess shouldBe false + oba.valid(f.realPowScheme, Some(required)) shouldBe false + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = true, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement whose known parent is an old header is not penalized, even with insufficient work") { + withFixture(new Fixture(Some(highDifficultyHex))) { f => + val oldParent = f.chain.head.header + val oba = announcement(f, oldParent.id, f.hist.fullBlockHeight + 1, requiredNBitsAfter(f, oldParent)) + f.realPowScheme.validate(oba.header).isSuccess shouldBe false + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + } + } + + property("ordering block announcement at genesis height with an empty header chain is checked against the configured initial difficulty") { + withFixture(new Fixture(applyLocalChain = false)) { f => + f.hist.bestHeaderOpt shouldBe None + val initial = f.historySettings.chainSettings.initialNBits + val oba = announcement(f, Header.GenesisParentId, ErgoHistoryUtils.GenesisHeight, initial) + + f.send(oba) + + val o = outcome(f, oba) + (o.stored, o.relayed, o.penalized) shouldBe ((true, true, false)) + } + } + + property("ordering block announcement at genesis height with a different difficulty is rejected and penalized") { + withFixture(new Fixture(Some(highDifficultyHex), applyLocalChain = false)) { f => + // configured initial difficulty 2^80; the announcement declares difficulty 1 + val other = DifficultySerializer.encodeCompactBits(1) + other should not be f.historySettings.chainSettings.initialNBits + val oba = announcement(f, Header.GenesisParentId, ErgoHistoryUtils.GenesisHeight, other) + // on its own, the header meets the difficulty it declares + oba.valid(f.realPowScheme, None) shouldBe true + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = true, + headerRequests = Seq.empty) + } + } +} From 59e47f3f2074a2c9c881c13cb3a94386aa4c7234 Mon Sep 17 00:00:00 2001 From: cafebedouin Date: Fri, 18 Sep 2026 02:55:22 -0500 Subject: [PATCH 2/4] Bind announcement difficulty to the known parent one block below; drop the sender-only request Addresses review of the parent-check PR. Changes since the reviewed commit: - Difficulty binding no longer requires the parent to be on the best chain. The expected difficulty is derived from the known parent one block below via requiredDifficultyAfter (which handles an off-best-chain parent), never from the announced header's own nBits. This restores weak-blocks behaviour for a known fork parent; restricting processing to best-chain parents is a separate policy, not applied here. - Genesis: both paths guard on bestHeaderOpt.isEmpty and bind parentId == GenesisParentId (the input path previously used bestFullBlockIdOpt.isEmpty, which is always true where it is reached). - Unbindable announcements (unknown parent, parent not exactly one block below, or a genesis-height announcement once headers exist or with a non-genesis parent) are dropped; a real parent arrives via normal header sync. The sender-only header request and its expiration message/handler are removed and will return as a follow-up once request-attempt identity is available. - Logging on the drop path lowered to debug. Tests: off-best-chain parents are now processed; unknown parents drop without a request; the sender-request and expiration tests are removed; the two "accepted on replay" cases are retained (one adapted to assert no request on first delivery); genesis-parent cases added. Both parent-check specs pass (24 properties); the genesis guard is mutation-checked. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01F9aGNTUgEQRrSKKBLTCqcY --- .../network/ErgoNodeViewSynchronizer.scala | 76 +++---- .../ErgoNodeViewSynchronizerMessages.scala | 14 -- .../network/InputBlockParentBindingSpec.scala | 196 +++++------------- ...ringBlockAnnouncementParentCheckSpec.scala | 140 +++---------- 4 files changed, 110 insertions(+), 316 deletions(-) diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala index 540dc9f5bb..d87b19cccb 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala @@ -616,22 +616,6 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, } } - /** - * Request a header from `peer` only, once. The request is tracked, so the header is accepted if `peer` delivers - * it before the delivery timeout. After the timeout the request is forgotten instead of checked: no other peer is - * asked for a header that only `peer` claimed to know, and no peer is penalized for not delivering it. - */ - protected def requestHeaderFromSenderOnly(headerId: ModifierId, peer: ConnectedPeer): Unit = { - val hid = Header.modifierTypeId - log.debug(s"Requesting header $headerId from $peer only") - networkControllerRef ! SendToNetwork(Message(RequestModifierSpec, Right(InvData(hid, Seq(headerId))), None), SendToPeer(peer)) - deliveryTracker.setRequested(hid, headerId, peer) { _ => - val expiration = new SenderOnlyRequestExpired(hid, headerId) - expiration.timer = context.system.scheduler.scheduleOnce(deliveryTimeout, self, expiration) - expiration.timer - } - } - /* * Private helper methods to request UTXO set snapshots metadata and related data (manifests, chunks) from peers */ @@ -1503,13 +1487,19 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, if (subBlockHeader.height == hr.fullBlockHeight + 1) { val powScheme = settings.chainSettings.powScheme val parentHeaderOpt = hr.modifierById(subBlockHeader.parentId).collect { case h: Header => h } - // Expected difficulty comes only from a known parent in the best chain, one block below the announced header - // (or from the configured initial difficulty at genesis height), never from the announced header itself + // Expected difficulty is derived from the known parent one block below the announced header + // (or the configured initial difficulty at genesis), never from the announced header's own nBits. + // A known parent that is not on the best chain still yields a real difficulty via + // requiredDifficultyAfter (which falls back to headerChainBack), so it is bound here as well; + // whether to additionally restrict processing to best-chain parents is a separate policy, not + // applied here. val expectedNBits: Option[Long] = if (subBlockHeader.isGenesis) { - if (hr.bestFullBlockIdOpt.isEmpty) Some(settings.chainSettings.initialNBits) else None + if (hr.bestHeaderOpt.isEmpty && subBlockHeader.parentId == Header.GenesisParentId) { + Some(settings.chainSettings.initialNBits) + } else None } else { parentHeaderOpt - .filter(parent => subBlockHeader.height == parent.height + 1 && hr.isInBestChain(parent)) + .filter(parent => subBlockHeader.height == parent.height + 1) .map { parent => val expectedDiff = hr.requiredDifficultyAfter(parent) import org.ergoplatform.mining.difficulty.DifficultySerializer @@ -1517,15 +1507,9 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, } } if (expectedNBits.isEmpty) { - // Policy point: input block whose parent is unknown, or known but not bound as above. - // Default: do not process it; if the parent header is unknown, request it from the sender only, once. - // Another policy (e.g. keeping the input block until its parent arrives) can replace this branch. - if (parentHeaderOpt.isEmpty && !subBlockHeader.isGenesis) { - if (deliveryTracker.status(subBlockHeader.parentId, Header.modifierTypeId, Seq(hr)) == ModifiersStatus.Unknown) { - requestHeaderFromSenderOnly(subBlockHeader.parentId, remote) - } - } - log.info(s"Not processing input block $subBlockId: parent ${subBlockHeader.parentId} is not bound to the best chain") + // Unbindable: the parent is unknown, is not exactly one block below, or this is a + // genesis-height announcement past genesis. Drop it; a real parent arrives via header sync. + log.debug(s"Not processing input block $subBlockId: parent ${subBlockHeader.parentId} does not bind an expected difficulty") return } val valid = usrOpt @@ -1848,13 +1832,19 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, if (!hr.contains(oba.header.id)) { val parentHeaderOpt = hr.modifierById(oba.header.parentId).collect { case h: Header => h } - // Expected difficulty comes only from a known parent in the best chain, one block below the announced header - // (or from the configured initial difficulty at genesis height), never from the announced header itself + // Expected difficulty is derived from the known parent one block below the announced header + // (or the configured initial difficulty at genesis), never from the announced header's own nBits. + // A known parent that is not on the best chain still yields a real difficulty via + // requiredDifficultyAfter (which falls back to headerChainBack), so it is bound here as well; + // whether to additionally restrict processing to best-chain parents is a separate policy, not + // applied here. val expectedNBits: Option[Long] = if (oba.header.isGenesis) { - if (hr.bestHeaderOpt.isEmpty) Some(settings.chainSettings.initialNBits) else None + if (hr.bestHeaderOpt.isEmpty && oba.header.parentId == Header.GenesisParentId) { + Some(settings.chainSettings.initialNBits) + } else None } else { parentHeaderOpt - .filter(parent => oba.header.height == parent.height + 1 && hr.isInBestChain(parent)) + .filter(parent => oba.header.height == parent.height + 1) .map { parent => val expectedDiff = hr.requiredDifficultyAfter(parent) import org.ergoplatform.mining.difficulty.DifficultySerializer @@ -1863,15 +1853,9 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, } if (expectedNBits.isEmpty) { - // Policy point: announcement whose parent is unknown, or known but not bound as above. - // Default: do not store, relay or process it; if the parent header is unknown, request it from the sender only, once. - // Another policy (e.g. keeping the announcement until its parent arrives) can replace this branch. - if (parentHeaderOpt.isEmpty && !oba.header.isGenesis) { - if (deliveryTracker.status(oba.header.parentId, Header.modifierTypeId, Seq(hr)) == ModifiersStatus.Unknown) { - requestHeaderFromSenderOnly(oba.header.parentId, remote) - } - } - log.info(s"Not processing ordering block announcement ${oba.header.id}: parent ${oba.header.parentId} is not bound to the best chain") + // Unbindable: the parent is unknown, is not exactly one block below, or this is a + // genesis-height announcement past genesis. Drop it; a real parent arrives via header sync. + log.debug(s"Not processing ordering block announcement ${oba.header.id}: parent ${oba.header.parentId} does not bind an expected difficulty") return } @@ -1954,14 +1938,6 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, * re-request modifier from a different random peer, if our node does not know a peer who have it */ protected def checkDelivery(hr: ErgoHistory): Receive = { - case expiration: SenderOnlyRequestExpired => - // clear the request only if this expiration belongs to the current request attempt - val current = deliveryTracker.getRequestedInfo(expiration.modifierTypeId, expiration.modifierId) - if (current.exists(_.cancellable eq expiration.timer)) { - log.info(s"Peer ${current.get.peer} has not delivered ${expiration.modifierTypeId} : ${expiration.modifierId} on time, forgetting the request") - deliveryTracker.clearStatusForModifier(expiration.modifierId, expiration.modifierTypeId, ModifiersStatus.Requested) - } - case CheckDelivery(peer, modifierTypeId, modifierId) => if (deliveryTracker.status(modifierId, modifierTypeId, Seq.empty) == ModifiersStatus.Requested) { // If transaction not delivered on time, we just forget about it. diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala index c738218633..044db7a7b4 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizerMessages.scala @@ -8,7 +8,6 @@ import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader import org.ergoplatform.nodeView.state.{ErgoStateReader, UtxoStateReader} import org.ergoplatform.nodeView.wallet.ErgoWalletReader import scorex.core.network.ConnectedPeer -import akka.actor.Cancellable import scorex.util.ModifierId import org.ergoplatform.ErgoLikeContext.Height import org.ergoplatform.modifiers.history.popow.NipopowProof @@ -34,19 +33,6 @@ object ErgoNodeViewSynchronizerMessages { modifierTypeId: NetworkObjectTypeId.Value, modifierId: ModifierId) - /** - * Expiration of a header request sent to one peer only (see `requestHeaderFromSenderOnly`): after the - * delivery timeout the request is forgotten, without asking another peer and without penalizing anyone. - * - * `timer` is the scheduled expiration itself, the `Cancellable` the delivery tracker stores for the request, - * so an expiration is matched to the request attempt that scheduled it: an expiration queued for an earlier - * attempt does not clear a newer request for the same header. - */ - final class SenderOnlyRequestExpired(val modifierTypeId: NetworkObjectTypeId.Value, val modifierId: ModifierId) { - /** set right after scheduling, before the expiration can be handled; unset, it matches no request */ - @volatile var timer: Cancellable = _ - } - trait PeerManagerEvent case class HandshakedPeer(remote: ConnectedPeer) extends PeerManagerEvent diff --git a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala index 6354d8ab90..795fdd8b5c 100644 --- a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala +++ b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala @@ -3,18 +3,15 @@ package org.ergoplatform.network import akka.actor.{ActorRef, Props} import akka.testkit.{TestActorRef, TestProbe} import org.ergoplatform.AutolykosSolution -import org.ergoplatform.consensus.Older import org.ergoplatform.mining.{AutolykosPowScheme, InputBlockFields} import org.ergoplatform.mining.difficulty.DifficultySerializer import org.ergoplatform.modifiers.ErgoFullBlock -import org.ergoplatform.modifiers.history.header.{Header, HeaderSerializer} +import org.ergoplatform.modifiers.history.header.{Header} import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._ -import org.ergoplatform.network.message.{InvData, Message, ModifiersData, ModifiersSpec, RequestModifierSpec} +import org.ergoplatform.network.message.{InvData, RequestModifierSpec} import org.ergoplatform.network.peer.PeerInfo -import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.ModifiersFromRemote import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryUtils, ErgoSyncInfoMessageSpec} import org.ergoplatform.nodeView.mempool.ErgoMemPool -import org.ergoplatform.nodeView.state.StateType import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState import org.ergoplatform.settings.ErgoSettings import org.ergoplatform.subblocks.InputBlockAnnouncement @@ -30,11 +27,11 @@ import scala.concurrent.duration._ import scala.concurrent.{Await, ExecutionContextExecutor} /** - * An input block announcement is processed only if its parent header is known, is in the best chain, and is - * exactly one block below the announced header (at genesis height, with no full blocks yet, the configured - * initial difficulty is used). The expected difficulty is derived from that parent. Otherwise the announcement is - * not processed, and an unknown parent header is requested from the sender only, once: the request expires after - * the delivery timeout without asking another peer or penalizing anyone. + * An input block announcement is processed only if its parent header is known and is exactly one block below + * the announced header (at genesis height, with no headers yet and the genesis parent, the configured initial + * difficulty is used). The expected difficulty is derived from that parent via requiredDifficultyAfter, so an + * off-best-chain parent is bound as well. Any announcement that cannot be bound — unknown parent, wrong height, + * or a spurious genesis-height announcement — is dropped without a request. * * The synchronizer under test validates proof-of-work with the real Autolykos scheme (the default test * configuration uses a fake scheme that accepts any header), while the local history is built with the @@ -101,23 +98,6 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti def process(ib: InputBlockAnnouncement): Unit = synchronizer.underlyingActor.processInputBlock(ib, hist, mempool, peer, Some(state)) - /** Makes the synchronizer handle scheduled and network messages, with another peer that headers could be asked from. */ - def initialize(): ConnectedPeer = { - val otherPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("OtherPeer").ref, - Some(PeerInfo(defaultPeerSpec.copy(features = Seq(ModePeerFeature(StateType.Utxo, verifyingTransactions = true, None, -1))), - System.currentTimeMillis()))) - syncTracker.updateStatus(otherPeer, Older, Some(hist.fullBlockHeight + 10)) - synchronizer ! ChangedState(state) - synchronizer ! ChangedHistory(hist) - synchronizer ! ChangedMempool(mempool) - Thread.sleep(300) - viewHolderProbe.receiveWhile(max = 300.millis, idle = 100.millis) { case m => m } - ncProbe.receiveWhile(max = 300.millis, idle = 100.millis) { case m => m } - otherPeer - } - - def deliverHeader(h: Header): Unit = synchronizer ! Message(ModifiersSpec, - Left(ModifiersSpec.toBytes(ModifiersData(Header.modifierTypeId, Map(h.id -> HeaderSerializer.toBytes(h))))), Some(peer)) } private def withFixture(test: Fixture => Any): Unit = withFixture(new Fixture)(test) @@ -173,13 +153,13 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti private def penalized(msgs: Seq[Any]): Boolean = msgs.exists(_.isInstanceOf[PenalizePeer]) - property("input block with unknown parent header is dropped, and the parent header is requested from the sender") { + property("input block with unknown parent header is dropped without requesting the header") { withFixture { f => val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) val tip = f.hist.bestFullBlockOpt.get.header val ib = announcement(f, unknownParent, tip.height + 1, DifficultySerializer.encodeCompactBits(1)) - // the announced header on its own passes the proof-of-work and Merkle checks + // the announced header on its own passes proof-of-work and Merkle checks; only the parent binding drops it ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, None) shouldBe true f.process(ib) @@ -187,29 +167,8 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti viewHolderGotInputBlock(f) shouldBe false val msgs = networkMessages(f) penalized(msgs) shouldBe false - headerRequests(msgs) shouldBe Seq(Seq(unknownParent) -> SendToPeer(f.peer)) - msgs.collect { case s: SendToNetwork => s }.size shouldBe 1 - // the request is tracked, so the header is accepted when it arrives - f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested - } - } - - property("unknown parent header: the header delivered by the sender is accepted") { - withFixture { f => - f.initialize() - val tip = f.hist.bestFullBlockOpt.get - // a real header that is not in local history: a sibling of the best full block - val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header - f.process(announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1))) - headerRequests(networkMessages(f)) shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) - - f.deliverHeader(sibling) - - penalized(networkMessages(f)) shouldBe false - f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { - case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) - case _ => false - } shouldBe true + headerRequests(msgs) shouldBe empty + f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Unknown } } @@ -229,8 +188,30 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti } } - property("an input announcement is accepted on replay after its parent joins the best header chain") { - withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + property("input block whose parent is a known header outside the best chain is still processed") { + withFixture { f => + val tip = f.hist.bestFullBlockOpt.get + val forkBlock = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension) + f.hist.append(forkBlock.header).get + forkBlock.header.height shouldBe tip.header.height + forkBlock.header.id should not be tip.id + f.hist.bestFullBlockIdOpt shouldBe Some(tip.id) + f.hist.isInBestChain(forkBlock.header) shouldBe false + + // a known off-best-chain parent still yields a real difficulty via requiredDifficultyAfter, + // so the announcement binds and is processed; restricting to best-chain parents is left to the maintainer + val ib = announcement(f, forkBlock.header.id, tip.header.height + 1, requiredNBitsAfter(f, forkBlock.header)) + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(requiredNBitsAfter(f, forkBlock.header))) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe true + penalized(networkMessages(f)) shouldBe false + } + } + + property("an input announcement is accepted on replay once its parent header is known") { + withFixture { f => val originalTip = f.hist.bestFullBlockOpt.get val parent = nextBlock(Some(f.chain(1)), originalTip.blockTransactions.txs, defaultExtension) parent.height shouldBe originalTip.height @@ -239,23 +220,19 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(requiredNBitsAfter(f, parent.header))) shouldBe true + // first delivery: the parent is unknown, so the announcement is dropped without a request f.process(ib) - viewHolderGotInputBlock(f) shouldBe false val initialMessages = networkMessages(f) - headerRequests(initialMessages) shouldBe Seq(Seq(parent.id) -> SendToPeer(f.peer)) + headerRequests(initialMessages) shouldBe empty penalized(initialMessages) shouldBe false + // once the parent header is known (even off the best chain), a replay is accepted f.hist.append(parent.header).get f.hist.isInBestChain(parent.header) shouldBe false - val nextHeader = nextBlock(Some(parent), originalTip.blockTransactions.txs, defaultExtension).header - f.hist.append(nextHeader).get - f.hist.isInBestChain(parent.header) shouldBe true - f.hist.bestFullBlockIdOpt shouldBe Some(originalTip.id) ib.header.height shouldBe f.hist.fullBlockHeight + 1 f.process(ib) - f.viewHolderProbe.expectMsg(ProcessInputBlock(ib, f.peer)) val replayMessages = networkMessages(f) penalized(replayMessages) shouldBe false @@ -273,8 +250,8 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(requiredNBitsAfter(f, parent.header))) shouldBe true + // at +2 the announcement goes through the pre-existing request branch (unchanged by this PR) f.process(ib) - viewHolderGotInputBlock(f) shouldBe false val initialMessages = networkMessages(f) headerRequests(initialMessages) shouldBe Seq(Seq(parent.id) -> SendToPeer(f.peer)) @@ -286,7 +263,6 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti ib.header.height shouldBe f.hist.fullBlockHeight + 1 f.process(ib) - f.viewHolderProbe.expectMsg(ProcessInputBlock(ib, f.peer)) val replayMessages = networkMessages(f) penalized(replayMessages) shouldBe false @@ -294,85 +270,6 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti } } - property("unknown-parent discovery expires without requesting the header from another peer") { - withFixture { f => - val otherPeer = f.initialize() - val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) - val tip = f.hist.bestFullBlockOpt.get.header - val ib = announcement(f, unknownParent, tip.height + 1, DifficultySerializer.encodeCompactBits(1)) - - f.process(ib) - - // Observe two delivery deadlines, including a potential retry against the other peer. - val messages = f.ncProbe.receiveWhile(max = 5.seconds, idle = 5.seconds) { case m => m } - val requests = headerRequests(messages).filter(_._1.contains(unknownParent)) - requests.map(_._2 == SendToPeer(f.peer)) shouldBe Seq(true) - messages.exists { - case p: PenalizePeer => p.address == otherPeer.connectionId.remoteAddress - case _ => false - } shouldBe false - f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, unknownParent) shouldBe None - f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Unknown - viewHolderGotInputBlock(f) shouldBe false - } - } - - property("an expiration from an earlier request attempt does not clear the current request for the header") { - withFixture(new Fixture(requestTimeout = 30.seconds)) { f => - f.initialize() - val tip = f.hist.bestFullBlockOpt.get - val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header - val ib = announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1)) - f.process(ib) - headerRequests(networkMessages(f)) shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) - val first = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get - - // the first request is cleared (as when its expiration is already queued) and the header is requested again - f.deliveryTracker.setUnknown(sibling.id, Header.modifierTypeId) - f.process(ib) - headerRequests(networkMessages(f)) shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) - val current = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get - current should not be theSameInstanceAs(first) - - // the first attempt's expiration arrives now - val stale = new SenderOnlyRequestExpired(Header.modifierTypeId, sibling.id) - stale.timer = first.cancellable - f.synchronizer ! stale - - networkMessages(f) shouldBe empty - f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get should be theSameInstanceAs current - f.deliveryTracker.status(sibling.id, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested - - f.deliverHeader(sibling) - - penalized(networkMessages(f)) shouldBe false - f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { - case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) - case _ => false - } shouldBe true - } - } - - property("input block whose parent is a known header outside the best chain is dropped") { - withFixture { f => - val tip = f.hist.bestFullBlockOpt.get - val forkBlock = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension) - f.hist.append(forkBlock.header).get - forkBlock.header.height shouldBe tip.header.height - forkBlock.header.id should not be tip.id - f.hist.bestFullBlockIdOpt shouldBe Some(tip.id) - f.hist.isInBestChain(forkBlock.header) shouldBe false - - val ib = announcement(f, forkBlock.header.id, tip.header.height + 1, requiredNBitsAfter(f, forkBlock.header)) - ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(requiredNBitsAfter(f, forkBlock.header))) shouldBe true - - f.process(ib) - - viewHolderGotInputBlock(f) shouldBe false - networkMessages(f) shouldBe empty - } - } - property("input block extending the best chain with the required difficulty is still processed") { withFixture { f => val tip = f.hist.bestFullBlockOpt.get.header @@ -444,6 +341,21 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti } } + property("input block at genesis height whose parent is not the genesis parent is dropped") { + withFixture(new Fixture(applyLocalChain = false)) { f => + f.hist.bestFullBlockIdOpt shouldBe None + val nonGenesisParent = bytesToId(Array.fill(32)(0x5a.toByte)) + nonGenesisParent should not be Header.GenesisParentId + val ib = announcement(f, nonGenesisParent, ErgoHistoryUtils.GenesisHeight, + f.historySettings.chainSettings.initialNBits) + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + penalized(networkMessages(f)) shouldBe false + } + } + property("input block at genesis height with a different difficulty is rejected and penalized") { withFixture(new Fixture(Some(highDifficultyHex), applyLocalChain = false)) { f => // configured initial difficulty 2^80; the announcement declares difficulty 1 diff --git a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala index 00b9d21ff5..c1f3f35043 100644 --- a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala +++ b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala @@ -3,16 +3,15 @@ package org.ergoplatform.network import akka.actor.{ActorRef, Props} import akka.testkit.{TestActorRef, TestProbe} import org.ergoplatform.AutolykosSolution -import org.ergoplatform.consensus.{Equal, Older} +import org.ergoplatform.consensus.{Equal} import org.ergoplatform.mining.{AutolykosPowScheme, InputBlockFields} import org.ergoplatform.mining.difficulty.DifficultySerializer import org.ergoplatform.modifiers.{ErgoFullBlock, OrderingBlockAnnouncementTypeId} -import org.ergoplatform.modifiers.history.header.{Header, HeaderSerializer} +import org.ergoplatform.modifiers.history.header.{Header} import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._ import org.ergoplatform.network.message.inputblocks.{OrderingBlockAnnouncement, OrderingBlockAnnouncementMessageSpec} -import org.ergoplatform.network.message.{InvData, InvSpec, Message, ModifiersData, ModifiersSpec, RequestModifierSpec} +import org.ergoplatform.network.message.{InvData, InvSpec, Message, RequestModifierSpec} import org.ergoplatform.network.peer.PeerInfo -import org.ergoplatform.nodeView.ErgoNodeViewHolder.ReceivableMessages.ModifiersFromRemote import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryUtils, ErgoSyncInfoMessageSpec} import org.ergoplatform.nodeView.mempool.ErgoMemPool import org.ergoplatform.nodeView.state.StateType @@ -22,7 +21,7 @@ import org.ergoplatform.wallet.utils.FileUtils import org.scalatest.matchers.should.Matchers import org.scalatest.propspec.AnyPropSpec import scorex.core.network.NetworkController.ReceivableMessages.{PenalizePeer, SendToNetwork} -import scorex.core.network.{ConnectedPeer, DeliveryTracker, ModifiersStatus, SendToPeer} +import scorex.core.network.{ConnectedPeer, DeliveryTracker, ModifiersStatus} import scorex.testkit.utils.AkkaFixture import scorex.util.{ModifierId, bytesToId} @@ -31,10 +30,10 @@ import scala.concurrent.{Await, ExecutionContextExecutor} /** * An ordering block announcement is stored, relayed and handed to the node view holder only if its parent - * header is known, is in the best header chain, and is exactly one block below the announced header; the - * expected difficulty is derived from that parent (at genesis height, with an empty header chain, the configured - * initial difficulty is used). If the parent header is unknown, it is requested from the sender only, once: the - * request expires after the delivery timeout without asking another peer or penalizing anyone. + * header is known and is exactly one block below the announced header; the expected difficulty is derived from + * that parent via requiredDifficultyAfter, so an off-best-chain parent is bound as well (at genesis height, with + * an empty header chain and the genesis parent, the configured initial difficulty is used). Any announcement + * that cannot be bound — unknown parent, wrong height, or a spurious genesis-height announcement — is dropped. * * The synchronizer under test validates proof-of-work with the real Autolykos scheme (the default test * configuration uses a fake scheme that accepts any header), while the local history is built with the @@ -114,17 +113,6 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers synchronizer ! Message(OrderingBlockAnnouncementMessageSpec, Left(OrderingBlockAnnouncementMessageSpec.toBytes(oba)), Some(peer)) - /** Adds another peer that headers could be asked from. */ - def addOlderPeer(): ConnectedPeer = { - val otherPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("OtherPeer").ref, - Some(PeerInfo(defaultPeerSpec.copy(features = Seq(ModePeerFeature(StateType.Utxo, verifyingTransactions = true, None, -1))), - System.currentTimeMillis()))) - syncTracker.updateStatus(otherPeer, Older, Some(hist.fullBlockHeight + 10)) - otherPeer - } - - def deliverHeader(h: Header): Unit = synchronizer ! Message(ModifiersSpec, - Left(ModifiersSpec.toBytes(ModifiersData(Header.modifierTypeId, Map(h.id -> HeaderSerializer.toBytes(h))))), Some(peer)) } private def withFixture(test: Fixture => Any): Unit = withFixture(new Fixture)(test) @@ -186,104 +174,19 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers ) } - property("ordering block announcement with unknown parent header is not stored, relayed or processed; parent header is requested from the sender") { + property("ordering block announcement with unknown parent header is dropped without requesting the header") { withFixture { f => val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) val oba = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, DifficultySerializer.encodeCompactBits(1)) - // the announced header on its own passes the proof-of-work and extension checks + // the announced header on its own passes proof-of-work and extension checks; only the parent binding drops it oba.valid(f.realPowScheme, None) shouldBe true f.send(oba) outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, - headerRequests = Seq(Seq(unknownParent) -> SendToPeer(f.peer))) - // the request is tracked, so the header is accepted when it arrives - f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested - } - } - - property("unknown parent header: the header delivered by the sender is accepted") { - withFixture { f => - f.addOlderPeer() - val tip = f.hist.bestFullBlockOpt.get - // a real header that is not in local history: a sibling of the best full block - val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header - val oba = announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1)) - f.send(oba) - outcome(f, oba).headerRequests shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) - - f.deliverHeader(sibling) - - f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } - .exists(_.isInstanceOf[PenalizePeer]) shouldBe false - f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { - case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) - case _ => false - } shouldBe true - } - } - - property("unknown-parent request expires without requesting the header from another peer or penalizing anyone") { - withFixture { f => - val otherPeer = f.addOlderPeer() - val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) - val oba = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, DifficultySerializer.encodeCompactBits(1)) - - f.send(oba) - - // observe two delivery deadlines, including a potential retry against the other peer - val messages = f.ncProbe.receiveWhile(max = 5.seconds, idle = 5.seconds) { case m => m } - val requests = messages.collect { - case s: SendToNetwork if s.message.spec.messageCode == RequestModifierSpec.messageCode && - s.message.data.get.asInstanceOf[InvData].ids.contains(unknownParent) => s.sendingStrategy - } - requests shouldBe Seq(SendToPeer(f.peer)) - messages.exists { - case p: PenalizePeer => p.address == otherPeer.connectionId.remoteAddress - case _ => false - } shouldBe false - messages.exists(_.isInstanceOf[PenalizePeer]) shouldBe false - f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, unknownParent) shouldBe None + headerRequests = Seq.empty) f.deliveryTracker.status(unknownParent, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Unknown - f.hist.getOrderingBlockAnnouncement(oba.header.id) shouldBe None - } - } - - property("an expiration from an earlier request attempt does not clear the current request for the header") { - withFixture(new Fixture(requestTimeout = 30.seconds)) { f => - f.addOlderPeer() - val tip = f.hist.bestFullBlockOpt.get - val sibling = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension).header - val oba = announcement(f, sibling.id, tip.header.height + 1, DifficultySerializer.encodeCompactBits(1)) - f.send(oba) - outcome(f, oba).headerRequests shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) - val first = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get - - // the first request is cleared (as when its expiration is already queued) and the header is requested again - f.deliveryTracker.setUnknown(sibling.id, Header.modifierTypeId) - f.send(oba) - outcome(f, oba).headerRequests shouldBe Seq(Seq(sibling.id) -> SendToPeer(f.peer)) - val current = f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get - current should not be theSameInstanceAs(first) - - // the first attempt's expiration arrives now - val stale = new SenderOnlyRequestExpired(Header.modifierTypeId, sibling.id) - stale.timer = first.cancellable - f.synchronizer ! stale - - outcome(f, oba).headerRequests shouldBe empty - f.deliveryTracker.getRequestedInfo(Header.modifierTypeId, sibling.id).get should be theSameInstanceAs current - f.deliveryTracker.status(sibling.id, Header.modifierTypeId, Seq.empty) shouldBe ModifiersStatus.Requested - - f.deliverHeader(sibling) - - f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m } - .exists(_.isInstanceOf[PenalizePeer]) shouldBe false - f.viewHolderProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.exists { - case ModifiersFromRemote(mods) => mods.exists(_.id == sibling.id) - case _ => false - } shouldBe true } } @@ -300,7 +203,7 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers } } - property("ordering block announcement whose parent is a known header outside the best chain is not stored, relayed or processed") { + property("ordering block announcement whose parent is a known header outside the best chain is still processed") { withFixture { f => val tip = f.hist.bestFullBlockOpt.get val forkBlock = nextBlock(Some(f.chain(1)), tip.blockTransactions.txs, defaultExtension) @@ -308,12 +211,14 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers forkBlock.header.height shouldBe tip.header.height f.hist.isInBestChain(forkBlock.header) shouldBe false + // a known off-best-chain parent still yields a real difficulty via requiredDifficultyAfter, + // so the announcement binds and is processed; restricting to best-chain parents is left to the maintainer val oba = announcement(f, forkBlock.header.id, tip.header.height + 1, requiredNBitsAfter(f, forkBlock.header)) oba.valid(f.realPowScheme, Some(requiredNBitsAfter(f, forkBlock.header))) shouldBe true f.send(oba) - outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + outcome(f, oba) shouldBe Outcome(stored = true, relayed = true, handedOff = true, penalized = false, headerRequests = Seq.empty) } } @@ -410,6 +315,21 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers } } + property("ordering block announcement at genesis height whose parent is not the genesis parent is dropped") { + withFixture(new Fixture(applyLocalChain = false)) { f => + f.hist.bestHeaderOpt shouldBe None + val nonGenesisParent = bytesToId(Array.fill(32)(0x5a.toByte)) + nonGenesisParent should not be Header.GenesisParentId + val oba = announcement(f, nonGenesisParent, ErgoHistoryUtils.GenesisHeight, + f.historySettings.chainSettings.initialNBits) + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + } + } + property("ordering block announcement at genesis height with a different difficulty is rejected and penalized") { withFixture(new Fixture(Some(highDifficultyHex), applyLocalChain = false)) { f => // configured initial difficulty 2^80; the announcement declares difficulty 1 From c31be8bc01235fedb1071d520dca9e30965ba28f Mon Sep 17 00:00:00 2001 From: cafebedouin Date: Sat, 19 Sep 2026 13:06:58 -0500 Subject: [PATCH 3/4] Clear a requested announcement's pending request on the benign unbindable-parent drop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the difficulty-binding change, addressing the receive-lifecycle regression @a-shannon found and @jozanek verified (review of 2026-09-19). Both announcement handlers return on the unbindable-parent drop (`expectedNBits.isEmpty`) *before* `setReceivedIfRequested`. So a requested-then-delivered announcement whose parent does not bind a difficulty stays `Requested` with its delivery timer live: a subsequent `CheckDelivery` then emits `NonDeliveryPenalty` against the peer that actually delivered it, and the stranded entry blocks requesting or replaying the announcement later. Add `clearRequestedIfFromSupplier` (mirrors `setReceivedIfRequested`, and the `getRequestedInfo(..) if ri.peer == remote` guard already used on the snapshot download paths) and call it on both unbindable-drop exits. It clears the entry to `Unknown` (re-requestable, not `Received` which would suppress later requests), and only when this peer is the one we requested from — an unsolicited response from another peer must not erase the real supplier's pending request. Tests (from @a-shannon's review patch): the two "requested with an unknown parent is delivered without a non-delivery penalty" cases now assert (status, timer-cancelled, penalty) = (Unknown, true, false); previously (Requested, false, true). Also includes @a-shannon's two headers-present / no-full-block genesis cases covering the genesis guard. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01QLpJKG92ySATpx4Gg5kYCJ --- .../network/ErgoNodeViewSynchronizer.scala | 21 +++++++ .../network/InputBlockParentBindingSpec.scala | 58 ++++++++++++++++++- ...ringBlockAnnouncementParentCheckSpec.scala | 47 +++++++++++++++ 3 files changed, 124 insertions(+), 2 deletions(-) diff --git a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala index d87b19cccb..0ca45c317b 100644 --- a/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala +++ b/src/main/scala/org/ergoplatform/network/ErgoNodeViewSynchronizer.scala @@ -1395,6 +1395,25 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, } } + /** + * A requested announcement that is then validly dropped (e.g. its parent does not bind an + * expected difficulty) must not be left `Requested`: a subsequent `CheckDelivery` would + * penalize the peer that actually delivered it, and the pending entry would block requesting + * or replaying the announcement later. Clear it back to `Unknown` (re-requestable) on such a + * benign-drop exit, but only when this peer is the one we requested it from -- an unsolicited + * response from another peer must not erase the real supplier's pending request. Mirrors the + * `getRequestedInfo(..) if ri.peer == remote` guard used on the snapshot download paths. + */ + private def clearRequestedIfFromSupplier(modifierId: ModifierId, + modifierTypeId: NetworkObjectTypeId.Value, + remote: ConnectedPeer): Unit = { + deliveryTracker.getRequestedInfo(modifierTypeId, modifierId) match { + case Some(info) if info.peer == remote => + deliveryTracker.setUnknown(modifierId, modifierTypeId) + case _ => () + } + } + /** * Request an input block from a peer by its ID. * @@ -1510,6 +1529,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, // Unbindable: the parent is unknown, is not exactly one block below, or this is a // genesis-height announcement past genesis. Drop it; a real parent arrives via header sync. log.debug(s"Not processing input block $subBlockId: parent ${subBlockHeader.parentId} does not bind an expected difficulty") + clearRequestedIfFromSupplier(subBlockId, InputBlockTypeId.value, remote) return } val valid = usrOpt @@ -1856,6 +1876,7 @@ class ErgoNodeViewSynchronizer(networkControllerRef: ActorRef, // Unbindable: the parent is unknown, is not exactly one block below, or this is a // genesis-height announcement past genesis. Drop it; a real parent arrives via header sync. log.debug(s"Not processing ordering block announcement ${oba.header.id}: parent ${oba.header.parentId} does not bind an expected difficulty") + clearRequestedIfFromSupplier(oba.header.id, OrderingBlockAnnouncementTypeId.value, remote) return } diff --git a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala index 795fdd8b5c..48a0297a03 100644 --- a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala +++ b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala @@ -5,10 +5,11 @@ import akka.testkit.{TestActorRef, TestProbe} import org.ergoplatform.AutolykosSolution import org.ergoplatform.mining.{AutolykosPowScheme, InputBlockFields} import org.ergoplatform.mining.difficulty.DifficultySerializer -import org.ergoplatform.modifiers.ErgoFullBlock +import org.ergoplatform.modifiers.{ErgoFullBlock, InputBlockTypeId} import org.ergoplatform.modifiers.history.header.{Header} import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages._ -import org.ergoplatform.network.message.{InvData, RequestModifierSpec} +import org.ergoplatform.network.message.{InvData, InvSpec, Message, RequestModifierSpec} +import org.ergoplatform.network.message.inputblocks.InputBlockMessageSpec import org.ergoplatform.network.peer.PeerInfo import org.ergoplatform.nodeView.history.{ErgoHistory, ErgoHistoryUtils, ErgoSyncInfoMessageSpec} import org.ergoplatform.nodeView.mempool.ErgoMemPool @@ -153,6 +154,39 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti private def penalized(msgs: Seq[Any]): Boolean = msgs.exists(_.isInstanceOf[PenalizePeer]) + property("requested input block with an unknown parent is delivered without a non-delivery penalty") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + f.synchronizer ! ChangedState(f.state) + f.synchronizer ! ChangedHistory(f.hist) + f.synchronizer ! ChangedMempool(f.mempool) + networkMessages(f) + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val ib = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, + DifficultySerializer.encodeCompactBits(1)) + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, None) shouldBe true + val inv = InvData(InputBlockTypeId.value, Seq(ib.id)) + f.synchronizer ! Message(InvSpec, Left(InvSpec.toBytes(inv)), Some(f.peer)) + val requests = networkMessages(f).collect { + case SendToNetwork(msg, _) if msg.spec.messageCode == RequestModifierSpec.messageCode => + msg.data.get.asInstanceOf[InvData] + } + requests should contain(inv) + f.deliveryTracker.status(ib.id, InputBlockTypeId.value, Seq.empty) shouldBe ModifiersStatus.Requested + val attempt = f.deliveryTracker.getRequestedInfo(InputBlockTypeId.value, ib.id).get + + f.synchronizer ! Message(InputBlockMessageSpec, Left(InputBlockMessageSpec.toBytes(ib)), Some(f.peer)) + viewHolderGotInputBlock(f) shouldBe false + networkMessages(f) shouldBe empty + val statusAfterDelivery = f.deliveryTracker.status(ib.id, InputBlockTypeId.value, Seq.empty) + val timerCancelledAfterDelivery = attempt.cancellable.isCancelled + f.synchronizer ! CheckDelivery(f.peer, InputBlockTypeId.value, ib.id) + val penaltyAfterDelivery = penalized(networkMessages(f)) + + (statusAfterDelivery, timerCancelledAfterDelivery, penaltyAfterDelivery) shouldBe + ((ModifiersStatus.Unknown, true, false)) + } + } + property("input block with unknown parent header is dropped without requesting the header") { withFixture { f => val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) @@ -341,6 +375,26 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti } } + property("input block at genesis height is dropped when headers exist but no full block has been applied") { + withFixture(new Fixture(applyLocalChain = false)) { f => + f.hist.append(f.chain.head.header).get + f.hist.bestHeaderOpt.isDefined shouldBe true + f.hist.bestFullBlockIdOpt shouldBe None + f.hist.fullBlockHeight shouldBe 0 + val initial = f.historySettings.chainSettings.initialNBits + val ib = announcement(f, Header.GenesisParentId, ErgoHistoryUtils.GenesisHeight, initial) + f.hist.contains(ib.header) shouldBe false + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, Some(initial)) shouldBe true + + f.process(ib) + + viewHolderGotInputBlock(f) shouldBe false + val msgs = networkMessages(f) + penalized(msgs) shouldBe false + headerRequests(msgs) shouldBe empty + } + } + property("input block at genesis height whose parent is not the genesis parent is dropped") { withFixture(new Fixture(applyLocalChain = false)) { f => f.hist.bestFullBlockIdOpt shouldBe None diff --git a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala index c1f3f35043..d13b7b2a0f 100644 --- a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala +++ b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala @@ -174,6 +174,35 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers ) } + property("requested ordering announcement with an unknown parent is delivered without a non-delivery penalty") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val oba = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, + DifficultySerializer.encodeCompactBits(1)) + oba.valid(f.realPowScheme, None) shouldBe true + val inv = InvData(OrderingBlockAnnouncementTypeId.value, Seq(oba.header.id)) + f.synchronizer ! Message(InvSpec, Left(InvSpec.toBytes(inv)), Some(f.peer)) + val requests = f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.collect { + case SendToNetwork(msg, _) if msg.spec.messageCode == RequestModifierSpec.messageCode => + msg.data.get.asInstanceOf[InvData] + } + requests should contain(inv) + f.deliveryTracker.status(oba.header.id, OrderingBlockAnnouncementTypeId.value, Seq.empty) shouldBe ModifiersStatus.Requested + val attempt = f.deliveryTracker.getRequestedInfo(OrderingBlockAnnouncementTypeId.value, oba.header.id).get + + f.send(oba) + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + val statusAfterDelivery = f.deliveryTracker.status(oba.header.id, OrderingBlockAnnouncementTypeId.value, Seq.empty) + val timerCancelledAfterDelivery = attempt.cancellable.isCancelled + f.synchronizer ! CheckDelivery(f.peer, OrderingBlockAnnouncementTypeId.value, oba.header.id) + val penaltyAfterDelivery = outcome(f, oba).penalized + + (statusAfterDelivery, timerCancelledAfterDelivery, penaltyAfterDelivery) shouldBe + ((ModifiersStatus.Unknown, true, false)) + } + } + property("ordering block announcement with unknown parent header is dropped without requesting the header") { withFixture { f => val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) @@ -315,6 +344,24 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers } } + property("ordering block announcement at genesis height is dropped when headers exist but no full block has been applied") { + withFixture(new Fixture(applyLocalChain = false)) { f => + f.hist.append(f.chain.head.header).get + f.hist.bestHeaderOpt.isDefined shouldBe true + f.hist.bestFullBlockIdOpt shouldBe None + f.hist.fullBlockHeight shouldBe 0 + val initial = f.historySettings.chainSettings.initialNBits + val oba = announcement(f, Header.GenesisParentId, ErgoHistoryUtils.GenesisHeight, initial) + f.hist.contains(oba.header) shouldBe false + oba.valid(f.realPowScheme, Some(initial)) shouldBe true + + f.send(oba) + + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + } + } + property("ordering block announcement at genesis height whose parent is not the genesis parent is dropped") { withFixture(new Fixture(applyLocalChain = false)) { f => f.hist.bestHeaderOpt shouldBe None From cb032c8826d54a0899d49990c8edf0ee5103da8a Mon Sep 17 00:00:00 2001 From: cafebedouin Date: Mon, 21 Sep 2026 04:06:48 -0500 Subject: [PATCH 4/4] Cover the supplier guard on the unbindable-parent drop: delivery from another peer Test-only follow-up to the receipt clear, for the gap @a-shannon and @jozanek noted (review of 2026-09-20): both specs delivered only from the peer the announcement was requested from, so the `info.peer == remote` guard in `clearRequestedIfFromSupplier` had no coverage. Add one case per announcement path: request the announcement from peer A via `Inv -> RequestModifier`, deliver the unbindable announcement from peer B, and assert that the entry is still `Requested`, that A's `RequestedInfo` is unchanged, and that its delivery timer is not cancelled. Each fixture gains an `otherPeer` that nothing is requested from. With the guard removed (`case Some(_) =>`) exactly these two cases fail with "Unknown was not equal to Requested"; the other 28 pass. With it in place the two specs pass 30/30. Co-Authored-By: Claude Fable 5.1 --- .../network/InputBlockParentBindingSpec.scala | 36 +++++++++++++++++++ ...ringBlockAnnouncementParentCheckSpec.scala | 33 +++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala index 48a0297a03..17aef4d244 100644 --- a/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala +++ b/src/test/scala/org/ergoplatform/network/InputBlockParentBindingSpec.scala @@ -88,6 +88,10 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti val peer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, pchProbe.ref, Some(PeerInfo(defaultPeerSpec, System.currentTimeMillis()))) + // a second peer, which nothing is ever requested from + val otherPeer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("OtherPeerHandlerProbe").ref, + Some(PeerInfo(defaultPeerSpec, System.currentTimeMillis()))) + val hist: ErgoHistory = ErgoHistory.readOrGenerate(historySettings)(null) val chain: Seq[ErgoFullBlock] = genChain(3, hist, nBits = historySettings.chainSettings.initialNBits) if (applyLocalChain) applyChain(hist, chain) @@ -187,6 +191,38 @@ class InputBlockParentBindingSpec extends AnyPropSpec with Matchers with FileUti } } + property("requested input block with an unknown parent delivered by another peer leaves the supplier's request pending") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + f.synchronizer ! ChangedState(f.state) + f.synchronizer ! ChangedHistory(f.hist) + f.synchronizer ! ChangedMempool(f.mempool) + networkMessages(f) + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val ib = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, + DifficultySerializer.encodeCompactBits(1)) + ib.valid(f.realPowScheme, f.state.stateContext.currentParameters, None) shouldBe true + val inv = InvData(InputBlockTypeId.value, Seq(ib.id)) + f.synchronizer ! Message(InvSpec, Left(InvSpec.toBytes(inv)), Some(f.peer)) + val requests = networkMessages(f).collect { + case SendToNetwork(msg, _) if msg.spec.messageCode == RequestModifierSpec.messageCode => + msg.data.get.asInstanceOf[InvData] + } + requests should contain(inv) + val attempt = f.deliveryTracker.getRequestedInfo(InputBlockTypeId.value, ib.id).get + attempt.peer shouldBe f.peer + f.otherPeer should not be f.peer + + // the announcement arrives from a peer it was not requested from + f.synchronizer ! Message(InputBlockMessageSpec, Left(InputBlockMessageSpec.toBytes(ib)), Some(f.otherPeer)) + viewHolderGotInputBlock(f) shouldBe false + penalized(networkMessages(f)) shouldBe false + + f.deliveryTracker.status(ib.id, InputBlockTypeId.value, Seq.empty) shouldBe ModifiersStatus.Requested + f.deliveryTracker.getRequestedInfo(InputBlockTypeId.value, ib.id) shouldBe Some(attempt) + attempt.cancellable.isCancelled shouldBe false + } + } + property("input block with unknown parent header is dropped without requesting the header") { withFixture { f => val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) diff --git a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala index d13b7b2a0f..6cb347e77a 100644 --- a/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala +++ b/src/test/scala/org/ergoplatform/network/OrderingBlockAnnouncementParentCheckSpec.scala @@ -89,6 +89,10 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers val peer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, pchProbe.ref, Some(PeerInfo(defaultPeerSpec, System.currentTimeMillis()))) + // a second peer, which nothing is ever requested from + val otherPeer: ConnectedPeer = ConnectedPeer(connectionIdGen.sample.get, TestProbe("OtherPeerHandlerProbe").ref, + Some(PeerInfo(defaultPeerSpec, System.currentTimeMillis()))) + val hist: ErgoHistory = ErgoHistory.readOrGenerate(historySettings)(null) val chain: Seq[ErgoFullBlock] = genChain(3, hist, nBits = historySettings.chainSettings.initialNBits) if (applyLocalChain) applyChain(hist, chain) @@ -203,6 +207,35 @@ class OrderingBlockAnnouncementParentCheckSpec extends AnyPropSpec with Matchers } } + property("requested ordering announcement with an unknown parent delivered by another peer leaves the supplier's request pending") { + withFixture(new Fixture(requestTimeout = 30.seconds)) { f => + val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte)) + val oba = announcement(f, unknownParent, f.hist.fullBlockHeight + 1, + DifficultySerializer.encodeCompactBits(1)) + oba.valid(f.realPowScheme, None) shouldBe true + val inv = InvData(OrderingBlockAnnouncementTypeId.value, Seq(oba.header.id)) + f.synchronizer ! Message(InvSpec, Left(InvSpec.toBytes(inv)), Some(f.peer)) + val requests = f.ncProbe.receiveWhile(max = 1.second, idle = 300.millis) { case m => m }.collect { + case SendToNetwork(msg, _) if msg.spec.messageCode == RequestModifierSpec.messageCode => + msg.data.get.asInstanceOf[InvData] + } + requests should contain(inv) + val attempt = f.deliveryTracker.getRequestedInfo(OrderingBlockAnnouncementTypeId.value, oba.header.id).get + attempt.peer shouldBe f.peer + f.otherPeer should not be f.peer + + // the announcement arrives from a peer it was not requested from + f.synchronizer ! Message(OrderingBlockAnnouncementMessageSpec, + Left(OrderingBlockAnnouncementMessageSpec.toBytes(oba)), Some(f.otherPeer)) + outcome(f, oba) shouldBe Outcome(stored = false, relayed = false, handedOff = false, penalized = false, + headerRequests = Seq.empty) + + f.deliveryTracker.status(oba.header.id, OrderingBlockAnnouncementTypeId.value, Seq.empty) shouldBe ModifiersStatus.Requested + f.deliveryTracker.getRequestedInfo(OrderingBlockAnnouncementTypeId.value, oba.header.id) shouldBe Some(attempt) + attempt.cancellable.isCancelled shouldBe false + } + } + property("ordering block announcement with unknown parent header is dropped without requesting the header") { withFixture { f => val unknownParent = bytesToId(Array.fill(32)(0x5a.toByte))