diff --git a/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala b/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala index 6e58782dfe..fe1e8c81d1 100644 --- a/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala +++ b/src/main/scala/org/ergoplatform/nodeView/mempool/ErgoMemPool.scala @@ -117,11 +117,11 @@ class ErgoMemPool private[mempool](private[mempool] val pool: OrderedTxPool, } val poolWithoutTx = removeTx(this, tx) - val doubleSpentTransactionIds = tx.inputs.flatMap(i => + val doubleSpentTransactionIds: Set[ModifierId] = tx.inputs.flatMap(i => poolWithoutTx.pool.inputs.get(i.boxId) ).toSet val doubleSpentTransactions = doubleSpentTransactionIds.flatMap { txId => - poolWithoutTx.pool.orderedTransactions.get(txId) + poolWithoutTx.pool.currentTransaction(txId).map(_._2) } doubleSpentTransactions.foldLeft(poolWithoutTx) { case (pool, tx) => removeTx(pool, tx.transaction) @@ -192,22 +192,32 @@ class ErgoMemPool private[mempool](private[mempool] val pool: OrderedTxPool, validationStartTime: Long): (ErgoMemPool, ProcessingOutcome) = { val tx = unconfirmedTransaction.transaction - val doubleSpendingWtxs = tx.inputs.flatMap { inp => + val doubleSpendingTxIds: Set[ModifierId] = tx.inputs.flatMap { inp => pool.inputs.get(inp.boxId) }.toSet val feeF = feeFactor(unconfirmedTransaction) - if (doubleSpendingWtxs.nonEmpty) { + val doubleSpendingEntries = doubleSpendingTxIds.toSeq.flatMap(pool.currentTransaction) + + if (doubleSpendingEntries.nonEmpty) { val ownWtx = weighted(tx, feeF) + val doubleSpendingWtxs: Set[WeightedTxId] = doubleSpendingEntries.map(_._1).toSet val doubleSpendingTotalWeight = doubleSpendingWtxs.map(_.weight).sum / doubleSpendingWtxs.size if (ownWtx.weight > doubleSpendingTotalWeight) { - val doubleSpendingTxs = doubleSpendingWtxs.map(wtx => pool.orderedTransactions(wtx)).toSeq + val doubleSpendingTxs = doubleSpendingEntries.map(_._2).distinct + // Remove the losers FIRST so the shared `inputs`/`outputs` map entries are gone before the + // winning tx writes its own. With put-then-remove, the loser's `tx.inputs.map(_.boxId)` would + // delete the winner's just-written entry for the shared box, leaving the winner orphaned + // in the index and breaking subsequent double-spend detection. val p = pool.remove(doubleSpendingTxs).put(unconfirmedTransaction, feeF) val updPool = new ErgoMemPool(p, stats, sortingOption) updPool -> new ProcessingOutcome.Accepted(unconfirmedTransaction, validationStartTime) } else { - this -> new ProcessingOutcome.DoubleSpendingLoser(doubleSpendingWtxs.map(_.id), validationStartTime) + this -> new ProcessingOutcome.DoubleSpendingLoser( + doubleSpendingWtxs.map(_.id), + validationStartTime + ) } } else { val poolSizeLimit = nodeSettings.mempoolCapacity diff --git a/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala b/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala index 2d5527ec7a..290361abc5 100644 --- a/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala +++ b/src/main/scala/org/ergoplatform/nodeView/mempool/OrderedTxPool.scala @@ -14,14 +14,18 @@ import scala.collection.immutable.TreeMap * @param orderedTransactions - collection containing transactions ordered by `tx.weight` * @param transactionsRegistry - mapping `tx.id` -> `WeightedTxId(tx.id,tx.weight)` required for getting transaction by its `id` * @param invalidatedTxIds - invalidated transaction ids in bloom filters - * @param outputs - mapping `box.id` -> `WeightedTxId(tx.id,tx.weight)` required for getting a transaction by its output box - * @param inputs - mapping `box.id` -> `WeightedTxId(tx.id,tx.weight)` required for getting a transaction by its input box id + * @param outputs - mapping `box.id` -> producing `tx.id`; current weight is resolved via `transactionsRegistry` + * @param inputs - mapping `box.id` -> spending `tx.id`; current weight is resolved via `transactionsRegistry` + * @param dataInputReaders - mapping `box.id` -> transaction ids which read it without consuming it + * @param family - explicit parent/child dependency graph between mempool transactions, used by `updateFamily` */ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTransaction], val transactionsRegistry: TreeMap[ModifierId, WeightedTxId], val invalidatedTxIds: ApproximateCacheLike[String], - val outputs: TreeMap[BoxId, WeightedTxId], - val inputs: TreeMap[BoxId, WeightedTxId]) + val outputs: TreeMap[BoxId, ModifierId], + val inputs: TreeMap[BoxId, ModifierId], + val dataInputReaders: TreeMap[BoxId, Set[ModifierId]], + val family: TxFamilyGraph) (implicit settings: ErgoSettings) extends ScorexLogging { import OrderedTxPool.weighted @@ -58,7 +62,7 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr orderedTransactions.size != transactionsRegistry.size && orderedTransactions.valuesIterator.exists(_.id == id) - private def currentTransaction(id: ModifierId): Option[(WeightedTxId, UnconfirmedTransaction)] = + private[mempool] def currentTransaction(id: ModifierId): Option[(WeightedTxId, UnconfirmedTransaction)] = transactionsRegistry.get(id) .flatMap(wtx => orderedTransactions.get(wtx).filter(_.id == id).map(wtx -> _)) .orElse { @@ -67,6 +71,39 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr } } + private def trackedTransaction(id: ModifierId): Option[(WeightedTxId, UnconfirmedTransaction)] = + transactionsRegistry.get(id) match { + case Some(_) => currentTransaction(id) + // Preserve an orphan's existing weight on re-put: its contribution may + // already be present in ancestors, so treating it as new would add it twice. + case None if hasUnregisteredTransaction(id) => currentTransaction(id) + case None => None + } + + private def addDataInputReaders(tx: ErgoTransaction): TreeMap[BoxId, Set[ModifierId]] = + tx.dataInputs.foldLeft(dataInputReaders) { (readers, dataInput) => + readers.updated( + dataInput.boxId, + readers.getOrElse(dataInput.boxId, Set.empty) + tx.id + ) + } + + private def removeDataInputReaders(tx: ErgoTransaction): TreeMap[BoxId, Set[ModifierId]] = + tx.dataInputs.foldLeft(dataInputReaders) { (readers, dataInput) => + val remaining = readers.getOrElse(dataInput.boxId, Set.empty) - tx.id + if (remaining.isEmpty) readers - dataInput.boxId + else readers.updated(dataInput.boxId, remaining) + } + + private def liveSpendChildren(tx: ErgoTransaction): Set[ModifierId] = + tx.outputs.flatMap(output => inputs.get(output.id)).filter(currentTransaction(_).isDefined).toSet + + private def liveReadChildren(tx: ErgoTransaction): Set[ModifierId] = + tx.outputs + .flatMap(output => dataInputReaders.getOrElse(output.id, Set.empty)) + .filter(currentTransaction(_).isDefined) + .toSet + def size: Int = orderedTransactions.size def get(id: ModifierId): Option[UnconfirmedTransaction] = { @@ -92,26 +129,65 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr */ def put(unconfirmedTx: UnconfirmedTransaction, feeFactor: Int): OrderedTxPool = { val tx = unconfirmedTx.transaction - - val newPool = transactionsRegistry.get(tx.id) match { - case Some(wtx) => - val currentWtx = currentTransaction(tx.id).map(_._1).getOrElse(wtx) - new OrderedTxPool( - withoutTransaction(tx.id).updated(currentWtx, unconfirmedTx), - transactionsRegistry.updated(tx.id, currentWtx), + val tracked = trackedTransaction(tx.id) + // A registry-only key may represent weight already propagated to ancestors. + // Keep it distinct from a discoverable body so current child weight can be reconciled. + val registryOnlyWtx = if (tracked.isEmpty) transactionsRegistry.get(tx.id) else None + val currentWtx = tracked.map(_._1).orElse(registryOnlyWtx) + + val newPool = currentWtx match { + case Some(existingWtx) => + val parentIds = tx.inputs.flatMap(in => outputs.get(in.boxId)).toSet + val readParentIds = tx.dataInputs.flatMap(in => outputs.get(in.boxId)).toSet + val spendChildIds = liveSpendChildren(tx) + val readChildIds = liveReadChildren(tx) + val restoredWtx = registryOnlyWtx match { + case Some(_) => + val childWeight = spendChildIds.toSeq + .flatMap(currentTransaction) + .map(_._1.weight) + .sum + existingWtx.copy(weight = existingWtx.feePerFactor + childWeight) + case None => + existingWtx + } + val updatedFamily = family + .addTx(tx.id, parentIds, readParentIds) + .addChildren(tx.id, spendChildIds, readChildIds) + val restoredPool = new OrderedTxPool( + withoutTransaction(tx.id).updated(restoredWtx, unconfirmedTx), + transactionsRegistry.updated(tx.id, restoredWtx), invalidatedTxIds, - outputs ++ tx.outputs.map(_.id -> currentWtx), - inputs ++ tx.inputs.map(_.boxId -> currentWtx) + outputs ++ tx.outputs.map(_.id -> tx.id), + inputs ++ tx.inputs.map(_.boxId -> tx.id), + addDataInputReaders(tx), + updatedFamily ) + registryOnlyWtx match { + case Some(_) => + restoredPool.reconcileFamilyWeights(parentIds, System.currentTimeMillis(), depth = 0) + case None => restoredPool + } case None => - val wtx = weighted(tx, feeFactor) + val baseWtx = weighted(tx, feeFactor) + val parentIds = tx.inputs.flatMap(in => outputs.get(in.boxId)).toSet + val readParentIds = tx.dataInputs.flatMap(in => outputs.get(in.boxId)).toSet + val spendChildIds = liveSpendChildren(tx) + val readChildIds = liveReadChildren(tx) + val childWeight = spendChildIds.toSeq.flatMap(currentTransaction).map(_._1.weight).sum + val wtx = baseWtx.copy(weight = baseWtx.weight + childWeight) + val updatedFamily = family + .addTx(tx.id, parentIds, readParentIds) + .addChildren(tx.id, spendChildIds, readChildIds) new OrderedTxPool( withoutTransaction(tx.id).updated(wtx, unconfirmedTx), transactionsRegistry.updated(wtx.id, wtx), invalidatedTxIds, - outputs ++ tx.outputs.map(_.id -> wtx), - inputs ++ tx.inputs.map(_.boxId -> wtx) - ).updateFamily(tx, wtx.weight, System.currentTimeMillis(), 0) + outputs ++ tx.outputs.map(_.id -> tx.id), + inputs ++ tx.inputs.map(_.boxId -> tx.id), + addDataInputReaders(tx), + updatedFamily + ).updateFamily(tx, parentIds, wtx.weight, System.currentTimeMillis(), 0) } if (newPool.orderedTransactions.size > mempoolCapacity) { val victim = newPool.orderedTransactions.last._2 @@ -125,45 +201,35 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr txs.foldLeft(this) { case (pool, tx) => pool.remove(tx) } } + private def removeStored(tx: ErgoTransaction, + wtx: WeightedTxId, + nextInvalidatedTxIds: ApproximateCacheLike[String]): OrderedTxPool = { + // Snapshot parents from the live graph before removeTx, so updateFamily can still walk them. + val parentIds = family.parentsOf(tx.id) + new OrderedTxPool( + withoutTransaction(tx.id), + transactionsRegistry - tx.id, + nextInvalidatedTxIds, + outputs -- tx.outputs.map(_.id), + inputs -- tx.inputs.map(_.boxId), + removeDataInputReaders(tx), + family.removeTx(tx.id) + ).updateFamily(tx, parentIds, -wtx.weight, System.currentTimeMillis(), depth = 0) + } + /** * Removes transaction from the pool * * @param tx - Transaction to remove */ def remove(tx: ErgoTransaction): OrderedTxPool = { - transactionsRegistry.get(tx.id) match { - case Some(wtx) if orderedTransactions.contains(wtx) => - new OrderedTxPool( - withoutTransaction(tx.id), - transactionsRegistry - tx.id, - invalidatedTxIds, - outputs -- tx.outputs.map(_.id), - inputs -- tx.inputs.map(_.boxId) - ).updateFamily(tx, -wtx.weight, System.currentTimeMillis(), depth = 0) - case Some(_) => - if (orderedTransactions.valuesIterator.exists(_.id == tx.id)) { - new OrderedTxPool( - withoutTransaction(tx.id), - transactionsRegistry - tx.id, - invalidatedTxIds, - outputs -- tx.outputs.map(_.id), - inputs -- tx.inputs.map(_.boxId) - ) - } else { - this - } + trackedTransaction(tx.id) match { + case Some((wtx, stored)) => + removeStored(stored.transaction, wtx, invalidatedTxIds) case None => - if (hasUnregisteredTransaction(tx.id)) { - new OrderedTxPool( - withoutTransaction(tx.id), - transactionsRegistry, - invalidatedTxIds, - outputs -- tx.outputs.map(_.id), - inputs -- tx.inputs.map(_.boxId) - ) - } else { - this - } + // A registry-only entry has no stored transaction body or trustworthy live weight. + // Keep the v6.0.4 conservative no-op instead of subtracting a guessed family weight. + this } } @@ -174,39 +240,22 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr */ def invalidate(unconfirmedTx: UnconfirmedTransaction): OrderedTxPool = { val tx = unconfirmedTx.transaction - transactionsRegistry.get(tx.id) match { - case Some(wtx) if orderedTransactions.contains(wtx) => - new OrderedTxPool( - withoutTransaction(tx.id), - transactionsRegistry - tx.id, - invalidatedTxIds.put(tx.id), - outputs -- tx.outputs.map(_.id), - inputs -- tx.inputs.map(_.boxId) - ).updateFamily(tx, -wtx.weight, System.currentTimeMillis(), depth = 0) - case Some(_) => - if (orderedTransactions.valuesIterator.exists(utx => utx.id == tx.id)) { - new OrderedTxPool( - withoutTransaction(tx.id), - transactionsRegistry - tx.id, - invalidatedTxIds.put(tx.id), - outputs -- tx.outputs.map(_.id), - inputs -- tx.inputs.map(_.boxId) - ) - } else { - new OrderedTxPool(orderedTransactions, transactionsRegistry, invalidatedTxIds.put(tx.id), outputs, inputs) - } + val nextInvalidatedTxIds = invalidatedTxIds.put(tx.id) + trackedTransaction(tx.id) match { + case Some((wtx, stored)) => + removeStored(stored.transaction, wtx, nextInvalidatedTxIds) case None => - if (hasUnregisteredTransaction(tx.id)) { - new OrderedTxPool( - withoutTransaction(tx.id), - transactionsRegistry, - invalidatedTxIds.put(tx.id), - outputs -- tx.outputs.map(_.id), - inputs -- tx.inputs.map(_.boxId) - ) - } else { - new OrderedTxPool(orderedTransactions, transactionsRegistry, invalidatedTxIds.put(tx.id), outputs, inputs) - } + // As in remove(), do not mutate indexes or family weights for an unresolved registry-only entry. + // Invalidating the supplied id is still safe and prevents immediate re-admission. + new OrderedTxPool( + orderedTransactions, + transactionsRegistry, + nextInvalidatedTxIds, + outputs, + inputs, + dataInputReaders, + family + ) } } @@ -232,6 +281,56 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr def isInvalidated(id: ModifierId): Boolean = invalidatedTxIds.mightContain(id) + /** + * Rebuild live ancestor weights from their direct live children. + * + * Registry-only transactions can return in any order. Recomputing each + * affected parent from the graph makes restoration independent of that + * order while preserving per-path weight semantics at reconvergence. + */ + private def reconcileFamilyWeights(txIds: Set[ModifierId], + startTime: Long, + depth: Int): OrderedTxPool = { + val now = System.currentTimeMillis() + val timeDiff = now - startTime + if (depth > MaxParentScanDepth || timeDiff > MaxParentScanTime) { + log.warn(s"reconcileFamilyWeights takes too long, depth: $depth, time diff: $timeDiff") + this + } else { + txIds.foldLeft(this) { case (pool, txId) => + pool.currentTransaction(txId) match { + case Some((wtx, utx)) => + val childWeight = pool.family.childrenOf(txId).toSeq + .flatMap(pool.currentTransaction) + .map(_._1.weight) + .sum + val reconciledWeight = wtx.feePerFactor + childWeight + if (reconciledWeight == wtx.weight) { + pool + } else { + val reconciledWtx = wtx.copy(weight = reconciledWeight) + val reconciledPool = new OrderedTxPool( + pool.withoutTransaction(txId).updated(reconciledWtx, utx), + pool.transactionsRegistry.updated(txId, reconciledWtx), + pool.invalidatedTxIds, + pool.outputs, + pool.inputs, + pool.dataInputReaders, + pool.family + ) + reconciledPool.reconcileFamilyWeights( + reconciledPool.family.parentsOf(txId), + startTime, + depth + 1 + ) + } + case None => + pool + } + } + } + } + /** * * Form families of transactions: take in account relations between transactions when performing ordering. @@ -245,6 +344,7 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr * @return */ private def updateFamily(tx: ErgoTransaction, + parentIds: Set[ModifierId], weight: Long, startTime: Long, depth: Int): OrderedTxPool = { @@ -255,22 +355,23 @@ class OrderedTxPool(val orderedTransactions: TreeMap[WeightedTxId, UnconfirmedTr this } else { - val uniqueTxIds: Set[WeightedTxId] = tx.inputs.flatMap(input => this.outputs.get(input.boxId)).toSet - val parentTxs = uniqueTxIds.flatMap(wtx => this.orderedTransactions.get(wtx).map(ut => wtx -> ut)) - - parentTxs.foldLeft(this) { case (pool, (snapshotWtx, _)) => - pool.currentTransaction(snapshotWtx.id) match { + parentIds.foldLeft(this) { case (pool, parentId) => + pool.currentTransaction(parentId) match { case Some((wtx, ut)) => val parent = ut.transaction val newWtx = WeightedTxId(wtx.id, wtx.weight + weight, wtx.feePerFactor, wtx.created) + // Weight propagation does not add or remove graph nodes/edges, nor change which tx produced/spent a box, + // so `family`, `outputs` and `inputs` are threaded through unchanged. Only the weight-bearing maps rebuild. val newPool = new OrderedTxPool( pool.withoutTransaction(parent.id).updated(newWtx, ut), pool.transactionsRegistry.updated(parent.id, newWtx), pool.invalidatedTxIds, - parent.outputs.foldLeft(pool.outputs)((newOutputs, box) => newOutputs.updated(box.id, newWtx)), - parent.inputs.foldLeft(pool.inputs)((newInputs, inp) => newInputs.updated(inp.boxId, newWtx)) + pool.outputs, + pool.inputs, + pool.dataInputReaders, + pool.family ) - newPool.updateFamily(parent, weight, startTime, depth + 1) + newPool.updateFamily(parent, pool.family.parentsOf(parent.id), weight, startTime, depth + 1) case None => pool } @@ -310,8 +411,11 @@ object OrderedTxPool { TreeMap.empty[WeightedTxId, UnconfirmedTransaction], TreeMap.empty[ModifierId, WeightedTxId], ExpiringApproximateCache.empty(frontCacheSize, frontCacheExpiration), - TreeMap.empty[BoxId, WeightedTxId], - TreeMap.empty[BoxId, WeightedTxId])(settings) + TreeMap.empty[BoxId, ModifierId], + TreeMap.empty[BoxId, ModifierId], + TreeMap.empty[BoxId, Set[ModifierId]], + TxFamilyGraph.empty + )(settings) } def weighted(unconfirmedTx: UnconfirmedTransaction, feeFactor: Int)(implicit ms: MonetarySettings): WeightedTxId = { diff --git a/src/main/scala/org/ergoplatform/nodeView/mempool/TxFamilyGraph.scala b/src/main/scala/org/ergoplatform/nodeView/mempool/TxFamilyGraph.scala new file mode 100644 index 0000000000..6969e34d24 --- /dev/null +++ b/src/main/scala/org/ergoplatform/nodeView/mempool/TxFamilyGraph.scala @@ -0,0 +1,191 @@ +package org.ergoplatform.nodeView.mempool + +import scorex.util.ModifierId + +import scala.annotation.tailrec + +/** + * Explicit parent/child dependency graph over mempool transactions. + * + * Nodes are identified by `ModifierId`. A spend edge `parent -> child` exists + * when `child` consumes an output produced by `parent`; a read edge exists + * when `child` references the output as a data input. Both edge kinds are + * stored eagerly in both directions. Only spend edges participate in family + * weight propagation and double-spend policy. + * + * The graph lives alongside the BoxId-keyed `outputs`/`inputs` maps in + * [[OrderedTxPool]]: the box maps stay authoritative for double-spend + * detection; this graph stays authoritative for tx-to-tx traversal. + * + * Empty adjacency sets are pruned so map keys do not accumulate on + * long-running nodes. + */ +final case class TxFamilyGraph(parents: Map[ModifierId, Set[ModifierId]], + children: Map[ModifierId, Set[ModifierId]], + readParents: Map[ModifierId, Set[ModifierId]], + readChildren: Map[ModifierId, Set[ModifierId]]) { + + /** + * Register `txId` with the given direct `parentIds`. Idempotent: a repeat + * call overwrites prior parents and reconciles the matching `children` + * back-edges (former parents that are no longer parents lose the back-edge). + */ + def addTx(txId: ModifierId, parentIds: Set[ModifierId]): TxFamilyGraph = { + val (newParents, newChildren) = reconcileEdges(txId, parentIds, parents, children) + copy(parents = newParents, children = newChildren) + } + + /** Register or reconcile both consuming and read-only parents of `txId`. */ + def addTx(txId: ModifierId, + spendParentIds: Set[ModifierId], + readParentIds: Set[ModifierId]): TxFamilyGraph = { + val (newParents, newChildren) = + reconcileEdges(txId, spendParentIds, parents, children) + val (newReadParents, newReadChildren) = + reconcileEdges(txId, readParentIds, readParents, readChildren) + + TxFamilyGraph(newParents, newChildren, newReadParents, newReadChildren) + } + + /** + * Restore outgoing edges for a producer which entered after some of its + * children, or which was reinserted after a rollback. The caller resolves + * children from the box indexes; this method only reconciles tx adjacency. + */ + def addChildren(txId: ModifierId, + spendChildIds: Set[ModifierId], + readChildIds: Set[ModifierId]): TxFamilyGraph = { + val withSpendChildren = spendChildIds.foldLeft(this) { (graph, childId) => + graph.addEdge(txId, childId, isRead = false) + } + readChildIds.foldLeft(withSpendChildren) { (graph, childId) => + graph.addEdge(txId, childId, isRead = true) + } + } + + /** + * Remove `txId` from the graph, cleaning both directions: drop the + * `parents(txId)` and `children(txId)` entries, and remove `txId` from the + * adjacency set of every former parent and former child. Empty sets are + * pruned. No-op if `txId` is not in the graph. + */ + def removeTx(txId: ModifierId): TxFamilyGraph = { + val (parentsAfter, childrenAfter) = removeEdges(txId, parents, children) + val (readParentsAfter, readChildrenAfter) = + removeEdges(txId, readParents, readChildren) + TxFamilyGraph(parentsAfter, childrenAfter, readParentsAfter, readChildrenAfter) + } + + def parentsOf(txId: ModifierId): Set[ModifierId] = parents.getOrElse(txId, Set.empty) + def childrenOf(txId: ModifierId): Set[ModifierId] = children.getOrElse(txId, Set.empty) + + def readParentsOf(txId: ModifierId): Set[ModifierId] = + readParents.getOrElse(txId, Set.empty) + + def readChildrenOf(txId: ModifierId): Set[ModifierId] = + readChildren.getOrElse(txId, Set.empty) + + def dependencyParentsOf(txId: ModifierId): Set[ModifierId] = + parentsOf(txId) ++ readParentsOf(txId) + + def dependencyChildrenOf(txId: ModifierId): Set[ModifierId] = + childrenOf(txId) ++ readChildrenOf(txId) + + /** Transitive ancestors of `txId` via BFS over `parents`. */ + def ancestorsOf(txId: ModifierId): Set[ModifierId] = + bfs(parentsOf(txId), Set(txId), parentsOf) - txId + + /** Transitive descendants of `txId` via BFS over `children`. */ + def descendantsOf(txId: ModifierId): Set[ModifierId] = + bfs(childrenOf(txId), Set(txId), childrenOf) - txId + + /** Transitive ancestors across both spend and read dependencies. */ + def dependencyAncestorsOf(txId: ModifierId): Set[ModifierId] = + bfs(dependencyParentsOf(txId), Set(txId), dependencyParentsOf) - txId + + /** Transitive descendants across both spend and read dependencies. */ + def dependencyDescendantsOf(txId: ModifierId): Set[ModifierId] = + bfs(dependencyChildrenOf(txId), Set(txId), dependencyChildrenOf) - txId + + private def addEdge(parentId: ModifierId, + childId: ModifierId, + isRead: Boolean): TxFamilyGraph = { + if (isRead) { + copy( + readParents = readParents.updated( + childId, + readParents.getOrElse(childId, Set.empty) + parentId + ), + readChildren = readChildren.updated( + parentId, + readChildren.getOrElse(parentId, Set.empty) + childId + ) + ) + } else { + copy( + parents = parents.updated(childId, parents.getOrElse(childId, Set.empty) + parentId), + children = children.updated(parentId, children.getOrElse(parentId, Set.empty) + childId) + ) + } + } + + private def reconcileEdges( + txId: ModifierId, + parentIds: Set[ModifierId], + parentIndex: Map[ModifierId, Set[ModifierId]], + childIndex: Map[ModifierId, Set[ModifierId]] + ): (Map[ModifierId, Set[ModifierId]], Map[ModifierId, Set[ModifierId]]) = { + val previousParents = parentIndex.getOrElse(txId, Set.empty) + val stale = previousParents -- parentIds + val fresh = parentIds -- previousParents + + val childrenAfterStale = stale.foldLeft(childIndex) { (acc, parentId) => + val updated = acc.getOrElse(parentId, Set.empty) - txId + if (updated.isEmpty) acc - parentId else acc.updated(parentId, updated) + } + val childrenAfterFresh = fresh.foldLeft(childrenAfterStale) { (acc, parentId) => + acc.updated(parentId, acc.getOrElse(parentId, Set.empty) + txId) + } + val parentsAfter = + if (parentIds.isEmpty) parentIndex - txId + else parentIndex.updated(txId, parentIds) + + parentsAfter -> childrenAfterFresh + } + + private def removeEdges( + txId: ModifierId, + parentIndex: Map[ModifierId, Set[ModifierId]], + childIndex: Map[ModifierId, Set[ModifierId]] + ): (Map[ModifierId, Set[ModifierId]], Map[ModifierId, Set[ModifierId]]) = { + val myParents = parentIndex.getOrElse(txId, Set.empty) + val myChildren = childIndex.getOrElse(txId, Set.empty) + + val parentsAfter = myChildren.foldLeft(parentIndex - txId) { (acc, childId) => + val updated = acc.getOrElse(childId, Set.empty) - txId + if (updated.isEmpty) acc - childId else acc.updated(childId, updated) + } + val childrenAfter = myParents.foldLeft(childIndex - txId) { (acc, parentId) => + val updated = acc.getOrElse(parentId, Set.empty) - txId + if (updated.isEmpty) acc - parentId else acc.updated(parentId, updated) + } + + parentsAfter -> childrenAfter + } + + @tailrec + private def bfs(frontier: Set[ModifierId], + visited: Set[ModifierId], + step: ModifierId => Set[ModifierId]): Set[ModifierId] = { + if (frontier.isEmpty) visited + else { + val visitedNext = visited ++ frontier + val next = frontier.flatMap(step) -- visitedNext + bfs(next, visitedNext, step) + } + } +} + +object TxFamilyGraph { + val empty: TxFamilyGraph = TxFamilyGraph(Map.empty, Map.empty, Map.empty, Map.empty) +} diff --git a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolIndexSpec.scala b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolIndexSpec.scala new file mode 100644 index 0000000000..c3f8f5a381 --- /dev/null +++ b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolIndexSpec.scala @@ -0,0 +1,264 @@ +package org.ergoplatform.nodeView.mempool + +import org.ergoplatform.{ErgoBoxCandidate, Input} +import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction} +import org.ergoplatform.nodeView.mempool.ErgoMemPoolUtils.{ProcessingOutcome, SortingOption} +import org.ergoplatform.nodeView.state.wrapped.WrappedUtxoState +import org.ergoplatform.settings.Constants.TrueTree +import org.ergoplatform.utils.ErgoTestHelpers +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks +import sigma.ast.ByteArrayConstant +import sigma.interpreter.{ContextExtension, ProverResult} + +class ErgoMemPoolIndexSpec extends AnyFlatSpec + with ErgoTestHelpers + with ScalaCheckPropertyChecks { + import org.ergoplatform.utils.ErgoCoreTestConstants._ + import org.ergoplatform.utils.ErgoNodeTestConstants._ + import org.ergoplatform.utils.generators.ErgoCoreGenerators._ + import org.ergoplatform.utils.generators.ValidBlocksGenerators._ + + it should "maintain TxFamilyGraph consistent with outputs map across put/invalidate" in { + val feeProposition = settings.chainSettings.monetary.feeProposition + + val (us, bh) = createUtxoState(settings) + val genesis = validFullBlock(None, us, bh) + val wus = WrappedUtxoState(us, bh, settings).applyModifier(genesis)(_ => ()).get + var txs = validTransactionsFromUtxoState(wus).map(tx => UnconfirmedTransaction(tx, None)) + val family_depth = 5 + val limitedPoolSettings = settings.copy( + nodeSettings = settings.nodeSettings.copy(mempoolCapacity = (family_depth + 1) * txs.size)) + var pool = ErgoMemPool.empty(limitedPoolSettings) + + def checkInvariant(): Unit = { + val p = pool.pool + val expectedParents = p.orderedTransactions.values.flatMap { utx => + val pids = utx.transaction.inputs.flatMap(in => p.outputs.get(in.boxId)).toSet + if (pids.isEmpty) None else Some(utx.transaction.id -> pids) + }.toMap + val expectedChildren = expectedParents.toSeq + .flatMap { case (child, parents) => parents.map(_ -> child) } + .groupBy(_._1) + .map { case (parent, edges) => parent -> edges.map(_._2).toSet } + val expectedReadParents = p.orderedTransactions.values.flatMap { utx => + val pids = utx.transaction.dataInputs.flatMap(in => p.outputs.get(in.boxId)).toSet + if (pids.isEmpty) None else Some(utx.transaction.id -> pids) + }.toMap + val expectedReadChildren = expectedReadParents.toSeq + .flatMap { case (child, parents) => parents.map(_ -> child) } + .groupBy(_._1) + .map { case (parent, edges) => parent -> edges.map(_._2).toSet } + val expectedDataInputReaders = p.orderedTransactions.values + .flatMap(utx => utx.transaction.dataInputs.map(_.boxId -> utx.id)) + .groupBy(_._1) + .map { case (boxId, readers) => boxId -> readers.map(_._2).toSet } + p.family.parents shouldBe expectedParents + p.family.children shouldBe expectedChildren + p.family.readParents shouldBe expectedReadParents + p.family.readChildren shouldBe expectedReadChildren + p.dataInputReaders shouldBe expectedDataInputReaders + val orderedIds = p.orderedTransactions.valuesIterator.map(_.id).toVector + orderedIds.distinct.size shouldBe orderedIds.size + p.transactionsRegistry.keySet shouldBe orderedIds.toSet + } + + txs.foreach { tx => + pool = pool.put(tx) + checkInvariant() + } + + for (_ <- 1 to family_depth) { + txs = txs.map { tx => + val spendingBox = tx.transaction.outputs.head + val sc = spendingBox.toCandidate + val out0 = new ErgoBoxCandidate(sc.value - 55000, sc.ergoTree, sc.creationHeight) + val out1 = new ErgoBoxCandidate(55000, feeProposition, sc.creationHeight) + val newTx = UnconfirmedTransaction(tx.transaction.copy( + inputs = IndexedSeq(new Input(spendingBox.id, emptyProverResult)), + outputCandidates = IndexedSeq(out0, out1)), None) + val (newPool, outcome) = pool.process(newTx, us) + outcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true + pool = newPool + checkInvariant() + newTx + } + } + + while (pool.size > 0) { + val victim = pool.getAll.head + pool = pool.invalidate(victim) + checkInvariant() + } + + pool.pool.family.parents shouldBe empty + pool.pool.family.children shouldBe empty + } + + it should "preserve the inputs index after replace-by-fee" in { + // With put-then-remove ordering, the new tx's inputs entry overwrites the loser's, + // and the subsequent loser-remove deletes the shared box id from `inputs`, + // leaving the new (winning) tx with NO inputs index entry for the shared box. + // Downstream double-spend detection would then miss further conflicts. + forAll(smallPositiveInt, smallPositiveInt) { case (n1, n2) => + whenever(n1 != n2 && n1 < n2) { + val testSettings = settings.copy( + nodeSettings = settings.nodeSettings.copy(mempoolSorting = SortingOption.FeePerByte) + ) + val (us, bh) = createUtxoState(testSettings) + val genesis = validFullBlock(None, us, bh) + val wus = WrappedUtxoState(us, bh, testSettings).applyModifier(genesis)(_ => ()).get + + val feeProp = testSettings.chainSettings.monetary.feeProposition + val inputBox = wus.takeBoxes(100).collectFirst { + case box if box.ergoTree == TrueTree => box + }.get + val feeOut = new ErgoBoxCandidate( + inputBox.value, + feeProp, + creationHeight = 0, + additionalTokens = inputBox.additionalTokens + ) + + def ctx(n: Int): ContextExtension = + ContextExtension(Map((1: Byte) -> ByteArrayConstant(Array.fill(1 + n)(0: Byte)))) + + // A larger context makes the transaction larger, so n1 < n2 gives the + // replacement transaction a higher fee-per-byte ratio at the same fee. + val txLargeLike = ErgoTransaction( + IndexedSeq(new Input(inputBox.id, new ProverResult(Array.emptyByteArray, ctx(n2)))), + IndexedSeq(feeOut)) + val txSmallLike = ErgoTransaction( + IndexedSeq(new Input(inputBox.id, new ProverResult(Array.emptyByteArray, ctx(n1)))), + IndexedSeq(feeOut)) + + val txLarge = UnconfirmedTransaction(ErgoTransaction(txLargeLike.inputs, txLargeLike.outputCandidates), None) + val txSmall = UnconfirmedTransaction(ErgoTransaction(txSmallLike.inputs, txSmallLike.outputCandidates), None) + + val pool0 = ErgoMemPool.empty(testSettings) + val (poolWithLarge, oLarge) = pool0.process(txLarge, us) + oLarge.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true + + val (poolWithSmall, oSmall) = poolWithLarge.process(txSmall, us) + oSmall.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true + poolWithSmall.size shouldBe 1 + poolWithSmall.take(1).head.transaction.id shouldBe txSmall.transaction.id + + // The crux: inputs index must point to the winner (txSmall) for the shared input box. + // With the buggy put-then-remove order this is `None`. + poolWithSmall.pool.inputs.get(inputBox.id) shouldBe Some(txSmall.transaction.id) + } + } + } + + it should "resolve an unregistered ordered conflict without dividing by zero" in { + val testSettings = settings.copy( + nodeSettings = settings.nodeSettings.copy(mempoolSorting = SortingOption.FeePerByte) + ) + val (us, bh) = createUtxoState(testSettings) + val genesis = validFullBlock(None, us, bh) + val wus = WrappedUtxoState(us, bh, testSettings).applyModifier(genesis)(_ => ()).get + val feeProp = testSettings.chainSettings.monetary.feeProposition + val inputBox = wus.takeBoxes(100).collectFirst { + case box if box.ergoTree == TrueTree => box + }.get + val feeOut = new ErgoBoxCandidate( + inputBox.value, + feeProp, + creationHeight = 0, + additionalTokens = inputBox.additionalTokens + ) + + def ctx(size: Int): ContextExtension = + ContextExtension(Map((1: Byte) -> ByteArrayConstant(Array.fill(size)(0: Byte)))) + + def spendingTx(contextSize: Int): UnconfirmedTransaction = { + val txLike = ErgoTransaction( + IndexedSeq(new Input(inputBox.id, new ProverResult(Array.emptyByteArray, ctx(contextSize)))), + IndexedSeq(feeOut) + ) + UnconfirmedTransaction(ErgoTransaction(txLike.inputs, txLike.outputCandidates), None) + } + + val loser = spendingTx(contextSize = 64) + val winner = spendingTx(contextSize = 1) + val (healthy, accepted) = ErgoMemPool.empty(testSettings).process(loser, us) + accepted.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true + + val p = healthy.pool + val orphanedPool = new OrderedTxPool( + p.orderedTransactions, + p.transactionsRegistry - loser.id, + p.invalidatedTxIds, + p.outputs, + p.inputs, + p.dataInputReaders, + p.family + )(testSettings) + val orphaned = new ErgoMemPool( + orphanedPool, + healthy.stats, + healthy.sortingOption + )(testSettings) + + val (replaced, outcome) = orphaned.process(winner, us) + outcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true + replaced.getAll.map(_.id) shouldBe Seq(winner.id) + replaced.pool.inputs(inputBox.id) shouldBe winner.id + } + + it should "ignore an unresolvable inputs-index ghost without dividing by zero" in { + val (us, bh) = createUtxoState(settings) + val genesis = validFullBlock(None, us, bh) + val wus = WrappedUtxoState(us, bh, settings).applyModifier(genesis)(_ => ()).get + val inputBox = wus.takeBoxes(100).collectFirst { + case box if box.ergoTree == TrueTree => box + }.get + val feeOut = new ErgoBoxCandidate( + inputBox.value, + feeProp, + creationHeight = 0, + additionalTokens = inputBox.additionalTokens + ) + + def spendingTx(contextSize: Int): UnconfirmedTransaction = { + val context = ContextExtension( + Map((1: Byte) -> ByteArrayConstant(Array.fill(contextSize)(0: Byte))) + ) + val txLike = ErgoTransaction( + IndexedSeq(new Input( + inputBox.id, + new ProverResult(Array.emptyByteArray, context) + )), + IndexedSeq(feeOut) + ) + UnconfirmedTransaction(ErgoTransaction(txLike.inputs, txLike.outputCandidates), None) + } + + val ghost = spendingTx(contextSize = 64) + val winner = spendingTx(contextSize = 1) + val empty = ErgoMemPool.empty(settings) + val p = empty.pool + val corruptedPool = new OrderedTxPool( + p.orderedTransactions, + p.transactionsRegistry, + p.invalidatedTxIds, + p.outputs, + p.inputs.updated(inputBox.id, ghost.id), + p.dataInputReaders, + p.family + )(settings) + val corrupted = new ErgoMemPool( + corruptedPool, + empty.stats, + empty.sortingOption + )(settings) + + corrupted.pool.currentTransaction(ghost.id) shouldBe None + val (acceptedPool, outcome) = corrupted.process(winner, us) + + outcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true + acceptedPool.getAll.map(_.id) shouldBe Seq(winner.id) + acceptedPool.pool.inputs(inputBox.id) shouldBe winner.id + } +} diff --git a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala index 1520a9f032..5fe2c1585f 100644 --- a/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/mempool/ErgoMemPoolSpec.scala @@ -56,8 +56,13 @@ class ErgoMemPoolSpec extends AnyFlatSpec val (us, bh) = createUtxoState(settings) val genesis = validFullBlock(None, us, bh) val wus = WrappedUtxoState(us, bh, settings).applyModifier(genesis)(_ => ()).get - val inputBox = wus.takeBoxes(1).head - val feeOut = new ErgoBoxCandidate(inputBox.value, feeProp, creationHeight = 0) + val inputBox = wus.takeBoxes(100).find(_.ergoTree == TrueTree).get + val feeOut = new ErgoBoxCandidate( + inputBox.value, + feeProp, + creationHeight = 0, + additionalTokens = inputBox.additionalTokens + ) val tx = ErgoTransaction( IndexedSeq(new Input(inputBox.id, ProverResult.empty)), IndexedSeq(feeOut) @@ -71,8 +76,9 @@ class ErgoMemPoolSpec extends AnyFlatSpec mempoolSorting = SortingOption.FeePerByte, )) - var poolSize = ErgoMemPool.empty(sortBySizeSettings) - poolSize = poolSize.process(UnconfirmedTransaction(tx, None), wus)._1 + val (poolSize, sizeOutcome) = + ErgoMemPool.empty(sortBySizeSettings).process(UnconfirmedTransaction(tx, None), wus) + sizeOutcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true val size = tx.size poolSize.pool.orderedTransactions.firstKey.weight shouldBe OrderedTxPool.weighted(tx, size).weight @@ -81,8 +87,9 @@ class ErgoMemPoolSpec extends AnyFlatSpec mempoolSorting = SortingOption.FeePerCycle, )) - var poolCost = ErgoMemPool.empty(sortByCostSettings) - poolCost = poolCost.process(UnconfirmedTransaction(tx, None), wus)._1 + val (poolCost, costOutcome) = + ErgoMemPool.empty(sortByCostSettings).process(UnconfirmedTransaction(tx, None), wus) + costOutcome.isInstanceOf[ProcessingOutcome.Accepted] shouldBe true val validationContext = wus.stateContext.simplifiedUpcoming() val cost = wus.validateWithCost(tx, validationContext, Int.MaxValue, None).get poolCost.pool.orderedTransactions.firstKey.weight shouldBe OrderedTxPool.weighted(tx, cost).weight @@ -791,7 +798,9 @@ class ErgoMemPoolSpec extends AnyFlatSpec TreeMap(tx.id -> wtxStale), emptyPool.invalidatedTxIds, emptyPool.outputs, - emptyPool.inputs + emptyPool.inputs, + emptyPool.dataInputReaders, + emptyPool.family )(settings) // pool.get traverses registry -> wtxStale -> orderedTransactions, diff --git a/src/test/scala/org/ergoplatform/nodeView/mempool/OrderedTxPoolSpec.scala b/src/test/scala/org/ergoplatform/nodeView/mempool/OrderedTxPoolSpec.scala index 1e40f6b8a9..dd1622986e 100644 --- a/src/test/scala/org/ergoplatform/nodeView/mempool/OrderedTxPoolSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/mempool/OrderedTxPoolSpec.scala @@ -1,6 +1,7 @@ package org.ergoplatform.nodeView.mempool import org.ergoplatform.ErgoBox.BoxId +import org.ergoplatform.DataInput import org.ergoplatform.modifiers.mempool.{ErgoTransaction, UnconfirmedTransaction} import org.ergoplatform.settings.Constants.TrueTree import org.ergoplatform.utils.ErgoTestHelpers @@ -33,8 +34,9 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { private def outputCandidates(plainCount: Int, plainValue: Long = 4000000L) = { val feeProposition = settings.chainSettings.monetary.feeProposition - IndexedSeq.fill(plainCount)(new org.ergoplatform.ErgoBoxCandidate(plainValue, TrueTree, 0)) :+ - new org.ergoplatform.ErgoBoxCandidate(1000000L, feeProposition, 0) + IndexedSeq.fill(plainCount)( + new org.ergoplatform.ErgoBoxCandidate(plainValue, TrueTree, 0) + ) :+ new org.ergoplatform.ErgoBoxCandidate(1000000L, feeProposition, 0) } private def buildReconvergentFixture(nonce: Int): ReconvergentFixture = { @@ -62,32 +64,30 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { val beforeD = Seq(a, b, c).foldLeft(ErgoMemPool.empty(settings)) { case (pool, tx) => pool.put(UnconfirmedTransaction(tx, None)) } - val uniqueParentKeys = d.inputs.flatMap(input => beforeD.pool.outputs.get(input.boxId)).toSet - val parentOrder = uniqueParentKeys - .flatMap { wtx => - beforeD.pool.orderedTransactions.get(wtx).map(unconfirmed => wtx -> unconfirmed) - } + val parentIds = d.inputs.flatMap(input => beforeD.pool.outputs.get(input.boxId)).toSet + val parentOrder = parentIds + .flatMap(id => beforeD.pool.transactionsRegistry.get(id)) .toSeq - .map(_._1.id) + .map(_.id) new ReconvergentFixture(a, b, c, d, beforeD, parentOrder) } private def fixtureWithSharedAncestorLast(): ReconvergentFixture = { - val fixture = (0 until 4096) + (0 until 4096) .iterator .map(buildReconvergentFixture) - .find { candidate => - candidate.parentOrder.lastOption.contains(candidate.a.id) && - candidate.parentOrder.take(2).toSet == Set(candidate.b.id, candidate.c.id) + .find { fixture => + fixture.parentOrder.lastOption.contains(fixture.a.id) && + fixture.parentOrder.take(2).toSet == Set(fixture.b.id, fixture.c.id) } - fixture.getOrElse(fail("No deterministic reconvergent fixture found")) + .getOrElse(fail("No deterministic reconvergent fixture found")) } private def orderedIds(pool: OrderedTxPool): Vector[ModifierId] = pool.orderedTransactions.valuesIterator.map(_.id).toVector - private def assertConsistent(pool: OrderedTxPool, expectedIds: Set[ModifierId]): Unit = { + private def assertUniqueAndRegistered(pool: OrderedTxPool, expectedIds: Set[ModifierId]): Unit = { val ids = orderedIds(pool) ids.toSet shouldBe expectedIds ids.distinct.size shouldBe ids.size @@ -97,24 +97,19 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { pool.orderedTransactions.get(key).map(_.id) shouldBe Some(id) } val transactions = pool.orderedTransactions.valuesIterator.map(_.transaction).toVector - val expectedOutputIds = transactions.flatMap(_.outputs.map(_.id)).toSet - val expectedInputIds = transactions.flatMap(_.inputs.map(_.boxId)).toSet - pool.outputs.keySet shouldBe expectedOutputIds - pool.inputs.keySet shouldBe expectedInputIds - pool.orderedTransactions.foreach { case (key, unconfirmed) => - unconfirmed.transaction.outputs.foreach { output => - pool.outputs.get(output.id) shouldBe Some(key) - } - unconfirmed.transaction.inputs.foreach { input => - pool.inputs.get(input.boxId) shouldBe Some(key) - } - } - pool.outputs.valuesIterator.foreach { key => - pool.orderedTransactions.contains(key) shouldBe true - } - pool.inputs.valuesIterator.foreach { key => - pool.orderedTransactions.contains(key) shouldBe true - } + val expectedOutputs = transactions + .flatMap(tx => tx.outputs.map(_.id -> tx.id)) + .toMap + val expectedInputs = transactions + .flatMap(tx => tx.inputs.map(_.boxId -> tx.id)) + .toMap + val expectedDataInputReaders = transactions + .flatMap(tx => tx.dataInputs.map(_.boxId -> tx.id)) + .groupBy(_._1) + .map { case (boxId, readers) => boxId -> readers.map(_._2).toSet } + pool.outputs shouldBe expectedOutputs + pool.inputs shouldBe expectedInputs + pool.dataInputReaders shouldBe expectedDataInputReaders } private def withDuplicate(pool: OrderedTxPool, tx: ErgoTransaction): OrderedTxPool = { @@ -127,7 +122,9 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { pool.transactionsRegistry, pool.invalidatedTxIds, pool.outputs, - pool.inputs + pool.inputs, + pool.dataInputReaders, + pool.family )(settings) } @@ -137,11 +134,13 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { pool.transactionsRegistry - tx.id, pool.invalidatedTxIds, pool.outputs, - pool.inputs + pool.inputs, + pool.dataInputReaders, + pool.family )(settings) } - it should "keep indexes consistent when a transaction closes a reconvergent family" in { + it should "keep one ordered entry per id when a transaction closes a reconvergent family" in { val fixture = fixtureWithSharedAncestorLast() val beforeIds = Set(fixture.a.id, fixture.b.id, fixture.c.id) val beforeWeights = beforeIds.map { id => @@ -150,44 +149,232 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { fixture.parentOrder.last shouldBe fixture.a.id fixture.parentOrder.take(2).toSet shouldBe Set(fixture.b.id, fixture.c.id) - fixture.d.inputs.map(_.boxId).distinct.size shouldBe fixture.d.inputs.size - assertConsistent(fixture.beforeD.pool, beforeIds) val afterD = fixture.beforeD.put(UnconfirmedTransaction(fixture.d, None)) val expectedIds = beforeIds + fixture.d.id - val duplicateCounts = orderedIds(afterD.pool).groupBy(identity).mapValues(_.size) + val ids = orderedIds(afterD.pool) + val duplicateCounts = ids.groupBy(identity).mapValues(_.size) val orderedKeys = afterD.pool.orderedTransactions.keysIterator .map(key => key.id -> key.weight) .toVector withClue(s"ordered keys=$orderedKeys, duplicate counts=$duplicateCounts") { - assertConsistent(afterD.pool, expectedIds) + assertUniqueAndRegistered(afterD.pool, expectedIds) val afterWeights = afterD.pool.transactionsRegistry.mapValues(_.weight) val dWeight = afterWeights(fixture.d.id) - dWeight should be > (0L) afterWeights(fixture.b.id) shouldBe beforeWeights(fixture.b.id) + dWeight afterWeights(fixture.c.id) shouldBe beforeWeights(fixture.c.id) + dWeight afterWeights(fixture.a.id) shouldBe beforeWeights(fixture.a.id) + 3L * dWeight } } + it should "track a pooled data-input producer without marking its output spent" in { + val producer = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5000), emptyProverResult)), + outputCandidates(1) + ) + val reader = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5001), emptyProverResult)), + IndexedSeq(DataInput(producer.outputs.head.id)), + outputCandidates(1) + ) - it should "heal a duplicated ancestor while propagating a new child" in { + val pool = Seq(producer, reader).foldLeft(ErgoMemPool.empty(settings)) { + case (current, tx) => current.put(UnconfirmedTransaction(tx, None)) + } + + pool.pool.family.readParentsOf(reader.id) shouldBe Set(producer.id) + pool.pool.family.readChildrenOf(producer.id) shouldBe Set(reader.id) + pool.pool.family.dependencyParentsOf(reader.id) shouldBe Set(producer.id) + pool.pool.inputs should not contain producer.outputs.head.id + } + + it should "restore a spend edge and family weight when the producer arrives after its child" in { + val producer = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5100), emptyProverResult)), + outputCandidates(1) + ) + val child = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(producer.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + + val childFirst = OrderedTxPool.empty(settings) + .put(UnconfirmedTransaction(child, None), child.size) + childFirst.family.parentsOf(child.id) shouldBe empty + + val restored = childFirst.put(UnconfirmedTransaction(producer, None), producer.size) + val producerKey = restored.transactionsRegistry(producer.id) + val childKey = restored.transactionsRegistry(child.id) + + restored.family.parentsOf(child.id) shouldBe Set(producer.id) + restored.family.childrenOf(producer.id) shouldBe Set(child.id) + producerKey.weight shouldBe producerKey.feePerFactor + childKey.weight + } + + it should "restore a read edge without changing family weight when the producer arrives late" in { + val producer = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5200), emptyProverResult)), + outputCandidates(1) + ) + val reader = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5201), emptyProverResult)), + IndexedSeq(DataInput(producer.outputs.head.id)), + outputCandidates(1) + ) + + val readerFirst = OrderedTxPool.empty(settings) + .put(UnconfirmedTransaction(reader, None), reader.size) + readerFirst.family.readParentsOf(reader.id) shouldBe empty + + val restored = readerFirst.put(UnconfirmedTransaction(producer, None), producer.size) + val producerKey = restored.transactionsRegistry(producer.id) + + restored.family.readParentsOf(reader.id) shouldBe Set(producer.id) + restored.family.readChildrenOf(producer.id) shouldBe Set(reader.id) + restored.family.parentsOf(reader.id) shouldBe empty + producerKey.weight shouldBe producerKey.feePerFactor + } + + it should "restore retained spend and read children when a producer is removed and reinserted" in { + val producer = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5300), emptyProverResult)), + outputCandidates(2) + ) + val child = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(producer.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val reader = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5301), emptyProverResult)), + IndexedSeq(DataInput(producer.outputs(1).id)), + outputCandidates(1) + ) + + val initial = Seq(producer, child, reader).foldLeft(OrderedTxPool.empty(settings)) { + case (pool, tx) => pool.put(UnconfirmedTransaction(tx, None), tx.size) + } + val removed = initial.remove(producer) + removed.transactionsRegistry.keySet shouldBe Set(child.id, reader.id) + removed.family.parentsOf(child.id) shouldBe empty + removed.family.readParentsOf(reader.id) shouldBe empty + + val restored = removed.put(UnconfirmedTransaction(producer, None), producer.size) + val producerKey = restored.transactionsRegistry(producer.id) + val childKey = restored.transactionsRegistry(child.id) + + restored.family.parentsOf(child.id) shouldBe Set(producer.id) + restored.family.readParentsOf(reader.id) shouldBe Set(producer.id) + producerKey.weight shouldBe producerKey.feePerFactor + childKey.weight + } + + it should "restore reconvergent path multiplicity when an ancestor is reinserted" in { val fixture = fixtureWithSharedAncestorLast() - val before = fixture.beforeD.pool - val ancestorWeight = before.transactionsRegistry(fixture.a.id).weight - val corrupted = withDuplicate(before, fixture.a) + val withDescendant = fixture.beforeD.pool + .put(UnconfirmedTransaction(fixture.d, None), fixture.d.size) + val expectedWeight = withDescendant.transactionsRegistry(fixture.a.id).weight + + val removed = withDescendant.remove(fixture.a) + removed.transactionsRegistry.keySet shouldBe Set(fixture.b.id, fixture.c.id, fixture.d.id) + + val restored = removed.put(UnconfirmedTransaction(fixture.a, None), fixture.a.size) + restored.family.childrenOf(fixture.a.id) shouldBe Set( + fixture.b.id, + fixture.c.id, + fixture.d.id + ) + restored.transactionsRegistry(fixture.a.id).weight shouldBe expectedWeight + } - orderedIds(corrupted).count(_ == fixture.a.id) shouldBe 2 - val healed = corrupted.put(UnconfirmedTransaction(fixture.d, None), fixture.d.size) + it should "restore a reinserted family weight through its retained ancestor" in { + val grandparent = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5350), emptyProverResult)), + outputCandidates(1) + ) + val parent = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(grandparent.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val child = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(parent.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val initial = Seq(grandparent, parent, child).foldLeft(OrderedTxPool.empty(settings)) { + case (pool, tx) => pool.put(UnconfirmedTransaction(tx, None), tx.size) + } + val expectedGrandparentWeight = initial.transactionsRegistry(grandparent.id).weight + val expectedParentWeight = initial.transactionsRegistry(parent.id).weight - assertConsistent( - healed, - Set(fixture.a.id, fixture.b.id, fixture.c.id, fixture.d.id) + val restored = initial + .remove(parent) + .put(UnconfirmedTransaction(parent, None), parent.size) + + restored.family.parentsOf(parent.id) shouldBe Set(grandparent.id) + restored.family.parentsOf(child.id) shouldBe Set(parent.id) + restored.transactionsRegistry(parent.id).weight shouldBe expectedParentWeight + restored.transactionsRegistry(grandparent.id).weight shouldBe expectedGrandparentWeight + } + + it should "keep parallel data-input readers distinct across removal" in { + val producer = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5400), emptyProverResult)), + outputCandidates(1) ) - val dWeight = healed.transactionsRegistry(fixture.d.id).weight - healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight + 3L * dWeight + def reader(nonce: Int): ErgoTransaction = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(nonce), emptyProverResult)), + IndexedSeq(DataInput(producer.outputs.head.id)), + outputCandidates(1) + ) + val reader1 = reader(5401) + val reader2 = reader(5402) + + val withReaders = Seq(reader1, reader2, producer).foldLeft(OrderedTxPool.empty(settings)) { + case (pool, tx) => pool.put(UnconfirmedTransaction(tx, None), tx.size) + } + withReaders.family.readChildrenOf(producer.id) shouldBe Set(reader1.id, reader2.id) + withReaders.dataInputReaders(producer.outputs.head.id) shouldBe Set(reader1.id, reader2.id) + + val removed = withReaders.remove(reader1) + removed.family.readChildrenOf(producer.id) shouldBe Set(reader2.id) + removed.family.readParentsOf(reader2.id) shouldBe Set(producer.id) + removed.dataInputReaders(producer.outputs.head.id) shouldBe Set(reader2.id) } + + it should "clean data-input edges and indexes when a reader is evicted" in { + val limitedSettings = settings.copy( + nodeSettings = settings.nodeSettings.copy(mempoolCapacity = 2) + ) + val producer = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5450), emptyProverResult)), + outputCandidates(1) + ) + val reader = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5451), emptyProverResult)), + IndexedSeq(DataInput(producer.outputs.head.id)), + outputCandidates(1) + ) + val unrelated = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5452), emptyProverResult)), + outputCandidates(1) + ) + + val withProducer = OrderedTxPool.empty(limitedSettings) + .put(UnconfirmedTransaction(producer, None), feeFactor = 1) + val withReader = withProducer + .put(UnconfirmedTransaction(reader, None), feeFactor = Int.MaxValue) + + withReader.family.readChildrenOf(producer.id) shouldBe Set(reader.id) + withReader.dataInputReaders(producer.outputs.head.id) shouldBe Set(reader.id) + + val afterEviction = withReader + .put(UnconfirmedTransaction(unrelated, None), feeFactor = 1) + + assertUniqueAndRegistered(afterEviction, Set(producer.id, unrelated.id)) + afterEviction.family.readChildrenOf(producer.id) shouldBe empty + afterEviction.family.readParentsOf(reader.id) shouldBe empty + afterEviction.dataInputReaders should not contain producer.outputs.head.id + } + it should "self-heal duplicate keys without propagating family weight twice" in { val fixture = fixtureWithSharedAncestorLast() val before = fixture.beforeD.pool @@ -198,7 +385,7 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { orderedIds(corrupted).count(_ == fixture.b.id) shouldBe 2 val healed = corrupted.put(UnconfirmedTransaction(fixture.b, None), fixture.b.size) - assertConsistent(healed, Set(fixture.a.id, fixture.b.id, fixture.c.id)) + assertUniqueAndRegistered(healed, Set(fixture.a.id, fixture.b.id, fixture.c.id)) val healedKeys = healed.orderedTransactions.keysIterator .filter(_.id == fixture.b.id) .toVector @@ -207,6 +394,23 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight } + it should "heal a duplicated ancestor while propagating a new child" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val ancestorWeight = before.transactionsRegistry(fixture.a.id).weight + val corrupted = withDuplicate(before, fixture.a) + + orderedIds(corrupted).count(_ == fixture.a.id) shouldBe 2 + val healed = corrupted.put(UnconfirmedTransaction(fixture.d, None), fixture.d.size) + + assertUniqueAndRegistered( + healed, + Set(fixture.a.id, fixture.b.id, fixture.c.id, fixture.d.id) + ) + val dWeight = healed.transactionsRegistry(fixture.d.id).weight + healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight + 3L * dWeight + } + it should "purge duplicate keys and subtract family weight once on removal" in { val fixture = fixtureWithSharedAncestorLast() val before = fixture.beforeD.pool @@ -218,7 +422,7 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { orderedIds(corrupted).count(_ == fixture.b.id) shouldBe 2 val removed = corrupted.remove(fixture.b) - assertConsistent(removed, Set(fixture.a.id, fixture.c.id)) + assertUniqueAndRegistered(removed, Set(fixture.a.id, fixture.c.id)) removed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight - childWeight removed.transactionsRegistry(fixture.c.id).weight shouldBe siblingWeight } @@ -234,7 +438,7 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { orderedIds(corrupted).count(_ == fixture.b.id) shouldBe 2 val invalidated = corrupted.invalidate(UnconfirmedTransaction(fixture.b, None)) - assertConsistent(invalidated, Set(fixture.a.id, fixture.c.id)) + assertUniqueAndRegistered(invalidated, Set(fixture.a.id, fixture.c.id)) invalidated.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight - childWeight invalidated.transactionsRegistry(fixture.c.id).weight shouldBe siblingWeight invalidated.isInvalidated(fixture.b.id) shouldBe true @@ -242,18 +446,243 @@ class OrderedTxPoolSpec extends AnyFlatSpec with Matchers with ErgoTestHelpers { it should "purge unregistered orphan keys on removal and invalidation" in { val fixture = fixtureWithSharedAncestorLast() - val before = fixture.beforeD.pool - val orphaned = withoutRegistry(before, fixture.b) + val orphaned = withoutRegistry(fixture.beforeD.pool, fixture.b) val expectedIds = Set(fixture.a.id, fixture.c.id) + val expectedAncestorWeight = fixture.beforeD.pool.transactionsRegistry(fixture.a.id).weight - + fixture.beforeD.pool.transactionsRegistry(fixture.b.id).weight orphaned.transactionsRegistry should not contain fixture.b.id orderedIds(orphaned).count(_ == fixture.b.id) shouldBe 1 val removed = orphaned.remove(fixture.b) - assertConsistent(removed, expectedIds) + assertUniqueAndRegistered(removed, expectedIds) + removed.transactionsRegistry(fixture.a.id).weight shouldBe expectedAncestorWeight val invalidated = orphaned.invalidate(UnconfirmedTransaction(fixture.b, None)) - assertConsistent(invalidated, expectedIds) + assertUniqueAndRegistered(invalidated, expectedIds) + invalidated.transactionsRegistry(fixture.a.id).weight shouldBe expectedAncestorWeight invalidated.isInvalidated(fixture.b.id) shouldBe true } + + it should "re-register an ordered orphan without propagating its family weight twice" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val orphanKey = before.transactionsRegistry(fixture.b.id) + val ancestorWeight = before.transactionsRegistry(fixture.a.id).weight + val orphaned = withoutRegistry(before, fixture.b) + + val healed = orphaned.put(UnconfirmedTransaction(fixture.b, None), fixture.b.size) + + assertUniqueAndRegistered(healed, Set(fixture.a.id, fixture.b.id, fixture.c.id)) + healed.transactionsRegistry(fixture.b.id) shouldBe orphanKey + healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight + } + + it should "re-register a registry-only transaction without propagating its family weight twice" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val childKey = before.transactionsRegistry(fixture.b.id) + val ancestorWeight = before.transactionsRegistry(fixture.a.id).weight + val registryOnly = new OrderedTxPool( + before.orderedTransactions - childKey, + before.transactionsRegistry, + before.invalidatedTxIds, + before.outputs, + before.inputs, + before.dataInputReaders, + before.family + )(settings) + + registryOnly.currentTransaction(fixture.b.id) shouldBe None + val healed = registryOnly.put(UnconfirmedTransaction(fixture.b, None), fixture.b.size) + + assertUniqueAndRegistered(healed, Set(fixture.a.id, fixture.b.id, fixture.c.id)) + healed.transactionsRegistry(fixture.b.id) shouldBe childKey + healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight + } + + it should "reconcile descendants added while a transaction is registry-only" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val parentKey = before.transactionsRegistry(fixture.b.id) + val ancestorWeight = before.transactionsRegistry(fixture.a.id).weight + val registryOnly = new OrderedTxPool( + before.orderedTransactions - parentKey, + before.transactionsRegistry, + before.invalidatedTxIds, + before.outputs, + before.inputs, + before.dataInputReaders, + before.family + )(settings) + val child = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(fixture.b.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val withChild = registryOnly.put(UnconfirmedTransaction(child, None), child.size) + + withChild.family.parentsOf(child.id) shouldBe Set(fixture.b.id) + withChild.currentTransaction(fixture.b.id) shouldBe None + withChild.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight + + val healed = withChild.put(UnconfirmedTransaction(fixture.b, None), fixture.b.size) + val childWeight = healed.transactionsRegistry(child.id).weight + + assertUniqueAndRegistered( + healed, + Set(fixture.a.id, fixture.b.id, fixture.c.id, child.id) + ) + healed.transactionsRegistry(fixture.b.id).weight shouldBe parentKey.weight + childWeight + healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight + childWeight + } + + it should "restore full weight through a parent reattached while its child is registry-only" in { + val grandparent = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5600), emptyProverResult)), + outputCandidates(1) + ) + val parent = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(grandparent.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val child = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(parent.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val initial = Seq(grandparent, parent, child).foldLeft(OrderedTxPool.empty(settings)) { + case (pool, tx) => pool.put(UnconfirmedTransaction(tx, None), tx.size) + } + val childKey = initial.transactionsRegistry(child.id) + val expectedParentWeight = initial.transactionsRegistry(parent.id).weight + val expectedGrandparentWeight = initial.transactionsRegistry(grandparent.id).weight + val registryOnlyChild = new OrderedTxPool( + initial.orderedTransactions - childKey, + initial.transactionsRegistry, + initial.invalidatedTxIds, + initial.outputs, + initial.inputs, + initial.dataInputReaders, + initial.family + )(settings) + + val withReinsertedParent = registryOnlyChild + .remove(parent) + .put(UnconfirmedTransaction(parent, None), parent.size) + + withReinsertedParent.family.parentsOf(child.id) shouldBe empty + withReinsertedParent.transactionsRegistry(parent.id).weight shouldBe (expectedParentWeight - childKey.weight) + withReinsertedParent.transactionsRegistry(grandparent.id).weight shouldBe (expectedGrandparentWeight - childKey.weight) + + val healed = withReinsertedParent + .put(UnconfirmedTransaction(child, None), child.size) + + assertUniqueAndRegistered(healed, Set(grandparent.id, parent.id, child.id)) + healed.transactionsRegistry(parent.id).weight shouldBe expectedParentWeight + healed.transactionsRegistry(grandparent.id).weight shouldBe expectedGrandparentWeight + } + + it should "restore family weights when registry-only transactions return parent first" in { + val grandparent = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(deterministicRootId(5700), emptyProverResult)), + outputCandidates(1) + ) + val parent = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(grandparent.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val child = ErgoTransaction( + IndexedSeq(new org.ergoplatform.Input(parent.outputs.head.id, emptyProverResult)), + outputCandidates(1) + ) + val initial = Seq(grandparent, parent, child).foldLeft(OrderedTxPool.empty(settings)) { + case (pool, tx) => pool.put(UnconfirmedTransaction(tx, None), tx.size) + } + val parentKey = initial.transactionsRegistry(parent.id) + val childKey = initial.transactionsRegistry(child.id) + val expectedGrandparentWeight = initial.transactionsRegistry(grandparent.id).weight + val expectedParentWeight = parentKey.weight + val registryOnlyChain = new OrderedTxPool( + initial.orderedTransactions -- Seq(parentKey, childKey), + initial.transactionsRegistry, + initial.invalidatedTxIds, + initial.outputs, + initial.inputs, + initial.dataInputReaders, + initial.family + )(settings) + + val healed = registryOnlyChain + .put(UnconfirmedTransaction(parent, None), parent.size) + .put(UnconfirmedTransaction(child, None), child.size) + + assertUniqueAndRegistered(healed, Set(grandparent.id, parent.id, child.id)) + healed.transactionsRegistry(parent.id).weight shouldBe expectedParentWeight + healed.transactionsRegistry(grandparent.id).weight shouldBe expectedGrandparentWeight + } + + it should "purge a transaction stored under a key different from its registry key" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val actualKey = before.transactionsRegistry(fixture.b.id) + val ancestorWeight = before.transactionsRegistry(fixture.a.id).weight + val staleKey = actualKey.copy(weight = actualKey.weight + 1L) + val broken = new OrderedTxPool( + before.orderedTransactions, + before.transactionsRegistry.updated(fixture.b.id, staleKey), + before.invalidatedTxIds, + before.outputs, + before.inputs, + before.dataInputReaders, + before.family + )(settings) + + broken.get(fixture.b.id) shouldBe None + orderedIds(broken) should contain(fixture.b.id) + + val invalidated = broken.invalidate(UnconfirmedTransaction(fixture.b, None)) + assertUniqueAndRegistered(invalidated, Set(fixture.a.id, fixture.c.id)) + invalidated.transactionsRegistry(fixture.a.id).weight shouldBe ancestorWeight - actualKey.weight + invalidated.isInvalidated(fixture.b.id) shouldBe true + } + + it should "heal a stale registry key on an ancestor during family propagation" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val actualKey = before.transactionsRegistry(fixture.a.id) + val staleKey = actualKey.copy(weight = actualKey.weight + 1L) + val broken = new OrderedTxPool( + before.orderedTransactions, + before.transactionsRegistry.updated(fixture.a.id, staleKey), + before.invalidatedTxIds, + before.outputs, + before.inputs, + before.dataInputReaders, + before.family + )(settings) + + val healed = broken.put(UnconfirmedTransaction(fixture.d, None), fixture.d.size) + val dWeight = healed.transactionsRegistry(fixture.d.id).weight + + assertUniqueAndRegistered( + healed, + Set(fixture.a.id, fixture.b.id, fixture.c.id, fixture.d.id) + ) + healed.transactionsRegistry(fixture.a.id).weight shouldBe actualKey.weight + 3L * dWeight + } + + it should "restore a missing ancestor registry entry during family propagation" in { + val fixture = fixtureWithSharedAncestorLast() + val before = fixture.beforeD.pool + val ancestorKey = before.transactionsRegistry(fixture.a.id) + val broken = withoutRegistry(before, fixture.a) + + val healed = broken.put(UnconfirmedTransaction(fixture.d, None), fixture.d.size) + val dWeight = healed.transactionsRegistry(fixture.d.id).weight + + assertUniqueAndRegistered( + healed, + Set(fixture.a.id, fixture.b.id, fixture.c.id, fixture.d.id) + ) + healed.transactionsRegistry(fixture.a.id).weight shouldBe ancestorKey.weight + 3L * dWeight + } } diff --git a/src/test/scala/org/ergoplatform/nodeView/mempool/TxFamilyGraphSpec.scala b/src/test/scala/org/ergoplatform/nodeView/mempool/TxFamilyGraphSpec.scala new file mode 100644 index 0000000000..93063f3219 --- /dev/null +++ b/src/test/scala/org/ergoplatform/nodeView/mempool/TxFamilyGraphSpec.scala @@ -0,0 +1,209 @@ +package org.ergoplatform.nodeView.mempool + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import scorex.util.ModifierId + +class TxFamilyGraphSpec extends AnyFlatSpec with Matchers { + + private def id(s: String): ModifierId = ModifierId @@ s + + private val tA = id("a") + private val tB = id("b") + private val tC = id("c") + private val tD = id("d") + + "TxFamilyGraph.empty" should "have no parents or children" in { + TxFamilyGraph.empty.parents shouldBe empty + TxFamilyGraph.empty.children shouldBe empty + TxFamilyGraph.empty.readParents shouldBe empty + TxFamilyGraph.empty.readChildren shouldBe empty + TxFamilyGraph.empty.parentsOf(tA) shouldBe empty + TxFamilyGraph.empty.childrenOf(tA) shouldBe empty + } + + "addTx" should "register a root tx with no parents without populating either map" in { + val g = TxFamilyGraph.empty.addTx(tA, Set.empty) + g.parents shouldBe empty + g.children shouldBe empty + g.parentsOf(tA) shouldBe empty + } + + it should "wire parent and back-edge for a single parent" in { + val g = TxFamilyGraph.empty.addTx(tB, Set(tA)) + g.parentsOf(tB) shouldBe Set(tA) + g.childrenOf(tA) shouldBe Set(tB) + } + + it should "wire all parents and back-edges for multiple parents (diamond)" in { + // tA, tB are roots; tC spends both; tD spends tC + val g = TxFamilyGraph.empty + .addTx(tC, Set(tA, tB)) + .addTx(tD, Set(tC)) + g.parentsOf(tC) shouldBe Set(tA, tB) + g.childrenOf(tA) shouldBe Set(tC) + g.childrenOf(tB) shouldBe Set(tC) + g.parentsOf(tD) shouldBe Set(tC) + g.childrenOf(tC) shouldBe Set(tD) + } + + it should "keep spend and read dependencies distinct" in { + val g = TxFamilyGraph.empty.addTx(tD, Set(tA, tB), Set(tC)) + + g.parentsOf(tD) shouldBe Set(tA, tB) + g.childrenOf(tA) shouldBe Set(tD) + g.childrenOf(tB) shouldBe Set(tD) + g.readParentsOf(tD) shouldBe Set(tC) + g.readChildrenOf(tC) shouldBe Set(tD) + g.dependencyParentsOf(tD) shouldBe Set(tA, tB, tC) + } + + it should "be idempotent when called twice with the same parents" in { + val g1 = TxFamilyGraph.empty.addTx(tB, Set(tA)) + val g2 = g1.addTx(tB, Set(tA)) + g2 shouldBe g1 + } + + it should "reconcile back-edges when called again with a different parent set" in { + val g1 = TxFamilyGraph.empty.addTx(tC, Set(tA, tB)) + val g2 = g1.addTx(tC, Set(tB)) // tA is no longer a parent of tC + g2.parentsOf(tC) shouldBe Set(tB) + g2.childrenOf(tA) shouldBe empty + g2.childrenOf(tB) shouldBe Set(tC) + g2.children.keySet should not contain tA // pruned + } + + it should "reconcile read back-edges independently of spend back-edges" in { + val g1 = TxFamilyGraph.empty.addTx(tD, Set(tA), Set(tB, tC)) + val g2 = g1.addTx(tD, Set(tA), Set(tC)) + + g2.parentsOf(tD) shouldBe Set(tA) + g2.childrenOf(tA) shouldBe Set(tD) + g2.readParentsOf(tD) shouldBe Set(tC) + g2.readChildrenOf(tB) shouldBe empty + g2.readChildrenOf(tC) shouldBe Set(tD) + } + + it should "preserve read dependencies when reconciling only spend parents" in { + val g1 = TxFamilyGraph.empty.addTx(tD, Set(tA), Set(tB)) + val g2 = g1.addTx(tD, Set(tC)) + + g2.parentsOf(tD) shouldBe Set(tC) + g2.childrenOf(tA) shouldBe empty + g2.childrenOf(tC) shouldBe Set(tD) + g2.readParentsOf(tD) shouldBe Set(tB) + g2.readChildrenOf(tB) shouldBe Set(tD) + } + + "removeTx" should "be a no-op for an unknown id" in { + val g = TxFamilyGraph.empty.addTx(tB, Set(tA)) + g.removeTx(tD) shouldBe g + } + + it should "remove a leaf and clean the parent's back-edge" in { + val g = TxFamilyGraph.empty.addTx(tB, Set(tA)) + val g2 = g.removeTx(tB) + g2.parentsOf(tB) shouldBe empty + g2.childrenOf(tA) shouldBe empty + g2.parents.keySet should not contain tB + g2.children.keySet should not contain tA // pruned + } + + it should "remove a middle node, dropping its parents' back-edges and trimming children's parents" in { + // chain tA -> tB -> tC + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tC, Set(tB)) + val g2 = g.removeTx(tB) + g2.parents.keySet should not contain tB + g2.children.keySet should not contain tB + g2.childrenOf(tA) shouldBe empty // tA -> tB edge gone + g2.parentsOf(tC) shouldBe empty // tB -> tC edge gone + } + + it should "remove a root and trim parents of its direct children" in { + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tC, Set(tA)) + val g2 = g.removeTx(tA) + g2.parentsOf(tB) shouldBe empty + g2.parentsOf(tC) shouldBe empty + g2.children.keySet should not contain tA + } + + it should "remove both spend and read edges" in { + val g = TxFamilyGraph.empty + .addTx(tC, Set(tA), Set(tB)) + .addTx(tD, Set.empty, Set(tC)) + val g2 = g.removeTx(tC) + + g2.childrenOf(tA) shouldBe empty + g2.readChildrenOf(tB) shouldBe empty + g2.readParentsOf(tD) shouldBe empty + } + + "ancestorsOf" should "return transitive parents and skip the start id itself" in { + // chain tA -> tB -> tC -> tD + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tC, Set(tB)) + .addTx(tD, Set(tC)) + g.ancestorsOf(tD) shouldBe Set(tA, tB, tC) + g.ancestorsOf(tA) shouldBe empty + } + + it should "handle a diamond without duplicating shared ancestors" in { + // tA -> tB, tA -> tC, tB -> tD, tC -> tD + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tC, Set(tA)) + .addTx(tD, Set(tB, tC)) + g.ancestorsOf(tD) shouldBe Set(tA, tB, tC) + } + + it should "never include the queried id when a corrupted graph contains a cycle" in { + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tA, Set(tB)) + + g.ancestorsOf(tA) shouldBe Set(tB) + g.descendantsOf(tA) shouldBe Set(tB) + g.dependencyAncestorsOf(tA) shouldBe Set(tB) + g.dependencyDescendantsOf(tA) shouldBe Set(tB) + } + + "descendantsOf" should "return transitive children and skip the start id itself" in { + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tC, Set(tB)) + .addTx(tD, Set(tC)) + g.descendantsOf(tA) shouldBe Set(tB, tC, tD) + g.descendantsOf(tD) shouldBe empty + } + + it should "handle a diamond without duplicating shared descendants" in { + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA)) + .addTx(tC, Set(tA)) + .addTx(tD, Set(tB, tC)) + g.descendantsOf(tA) shouldBe Set(tB, tC, tD) + } + + "dependencyAncestorsOf" should "traverse spend and read edges once" in { + val g = TxFamilyGraph.empty + .addTx(tB, Set(tA), Set.empty) + .addTx(tC, Set.empty, Set(tA)) + .addTx(tD, Set(tB), Set(tC)) + + g.dependencyAncestorsOf(tD) shouldBe Set(tA, tB, tC) + g.dependencyDescendantsOf(tA) shouldBe Set(tB, tC, tD) + } + + "the graph" should "stay empty after add followed by remove of the same tx" in { + val g = TxFamilyGraph.empty.addTx(tB, Set(tA)).removeTx(tB) + g.parents shouldBe empty + g.children shouldBe empty + g.readParents shouldBe empty + g.readChildren shouldBe empty + } +}