From 73096e012744677268a3785e61e4135853551a4d Mon Sep 17 00:00:00 2001 From: Ergologica <153913412+Ergologica@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:19:27 +0200 Subject: [PATCH 1/5] Keep wallet transactions unconfirmed at shutdown --- .../nodeView/ErgoNodeViewHolder.scala | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala b/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala index 56764d821d..d807b24935 100644 --- a/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala +++ b/src/main/scala/org/ergoplatform/nodeView/ErgoNodeViewHolder.scala @@ -29,6 +29,7 @@ import org.ergoplatform.modifiers.history.extension.Extension import scala.annotation.tailrec import scala.collection.mutable +import scala.concurrent.ExecutionContext import scala.util.{Failure, Success, Try} /** @@ -87,12 +88,35 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti Escalate } + override def preStart(): Unit = { + super.preStart() + restoreUnconfirmedWalletTransactions() + } + override def postStop(): Unit = { log.warn("Stopping ErgoNodeViewHolder") history().closeStorage() minimalState().closeStorage() } + /** + * The memory pool is not persisted, so a restart drops every transaction which was waiting in it. + * The wallet does keep its own unconfirmed transactions, so ask it for them and put them back, + * instead of waiting for peers to gossip them again - which they may never do. + */ + private def restoreUnconfirmedWalletTransactions(): Unit = { + implicit val ec: ExecutionContext = context.dispatcher + vault().unconfirmedTransactionsToRestore.onComplete { + case Success(txs) if txs.nonEmpty => + log.info(s"Putting ${txs.size} unconfirmed wallet transaction(s) back into the memory pool") + txs.foreach(tx => self ! RestoredTransaction(UnconfirmedTransaction(tx, None))) + case Success(_) => + log.debug("No unconfirmed wallet transactions to restore") + case Failure(t) => + log.warn("Could not read unconfirmed wallet transactions to restore: ", t) + } + } + /** * Update NodeView with new components and notify subscribers of changed components * @@ -658,6 +682,17 @@ abstract class ErgoNodeViewHolder[State <: ErgoState[State]](settings: ErgoSetti txModify(unconfirmedTx) case LocallyGeneratedTransaction(unconfirmedTx) => sender() ! txModify(unconfirmedTx) + case RestoredTransaction(unconfirmedTx) => + txModify(unconfirmedTx) match { + case _: ProcessingOutcome.Accepted => + log.info(s"Unconfirmed wallet transaction ${unconfirmedTx.id} is back in the memory pool") + case outcome => + // the transaction can not be brought back, e.g. it got on the blockchain while the node + // was down, or a conflicting one did. There is no point in keeping it for the next restart + log.info(s"Unconfirmed wallet transaction ${unconfirmedTx.id} was not accepted back " + + s"into the memory pool ($outcome), forgetting it") + vault().forgetUnconfirmedTransactions(Seq(unconfirmedTx.id)) + } case RecheckedTransactions(unconfirmedTxs) => val updatedPool = memoryPool().put(unconfirmedTxs) updateNodeView(updatedMempool = Some(updatedPool)) @@ -737,6 +772,12 @@ object ErgoNodeViewHolder { */ case class TransactionFromRemote(unconfirmedTx: UnconfirmedTransaction) + /** + * Wrapper for a wallet transaction which was unconfirmed when the node was stopped and is being + * put back into the memory pool now + */ + case class RestoredTransaction(unconfirmedTx: UnconfirmedTransaction) + /** * Wrapper for transactions which sit in mempool for long enough time, so `CleanWorker` is re-checking their * validity and then sending via this message to update the mempool From a78a94725b7c72b0f2e998d071690ac862f0f5cc Mon Sep 17 00:00:00 2001 From: Ergologica <153913412+Ergologica@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:20:57 +0200 Subject: [PATCH 2/5] Persist wallet transactions which are not on the blockchain yet --- .../nodeView/wallet/ErgoWallet.scala | 10 +- .../nodeView/wallet/ErgoWalletActor.scala | 108 ++++++++++++++++-- .../wallet/ErgoWalletActorMessages.scala | 15 +++ .../nodeView/wallet/ErgoWalletReader.scala | 8 ++ .../nodeView/wallet/ErgoWalletService.scala | 42 +++++++ 5 files changed, 174 insertions(+), 9 deletions(-) diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWallet.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWallet.scala index 54c0808eb0..3aeb9f2e80 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWallet.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWallet.scala @@ -9,7 +9,7 @@ import org.ergoplatform.nodeView.wallet.ErgoWalletActorMessages._ import org.ergoplatform.settings.{ErgoSettings, Parameters} import org.ergoplatform.wallet.boxes.{ReemissionData, ReplaceCompactCollectBoxSelector} import org.ergoplatform.core.VersionTag -import scorex.util.ScorexLogging +import scorex.util.{ModifierId, ScorexLogging} import scala.util.{Failure, Success, Try} @@ -46,6 +46,14 @@ class ErgoWallet(historyReader: ErgoHistoryReader, settings: ErgoSettings, param this } + /** + * Tell the wallet to stop keeping the given unconfirmed transactions across restarts, e.g. + * because the memory pool refused them when they were re-submitted. + */ + def forgetUnconfirmedTransactions(ids: Seq[ModifierId]): Unit = { + walletActor ! ForgetUnconfirmedTransactions(ids) + } + def scanPersistent(modifier: BlockSection): ErgoWallet = { modifier match { case fb: ErgoFullBlock => diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala index 78d7621a25..63d06db774 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActor.scala @@ -4,11 +4,14 @@ import akka.actor.SupervisorStrategy.{Restart, Stop} import akka.actor._ import akka.pattern.StatusReply import org.ergoplatform.ErgoBox._ +import org.ergoplatform.modifiers.ErgoFullBlock +import org.ergoplatform.modifiers.mempool.ErgoTransaction import org.ergoplatform.network.ErgoNodeViewSynchronizerMessages.{ChangedMempool, ChangedState} import org.ergoplatform.nodeView.history.ErgoHistoryReader import org.ergoplatform.nodeView.mempool.ErgoMemPoolReader import org.ergoplatform.nodeView.state.ErgoStateReader import org.ergoplatform.nodeView.wallet.ErgoWalletServiceUtils.DeriveNextKeyResult +import org.ergoplatform.nodeView.wallet.persistence.WalletStorage import org.ergoplatform.sdk.wallet.secrets.DerivationPath import org.ergoplatform.settings._ import org.ergoplatform.wallet.Constants.ScanId @@ -18,8 +21,9 @@ import org.ergoplatform._ import org.ergoplatform.core.VersionTag import org.ergoplatform.sdk.SecretString import org.ergoplatform.utils.ScorexEncoding -import scorex.util.ScorexLogging +import scorex.util.{ModifierId, ScorexLogging, bytesToId} +import scala.annotation.tailrec import scala.concurrent.duration._ import scala.util.{Failure, Success} @@ -76,7 +80,7 @@ class ErgoWalletActor(settings: ErgoSettings, val ws = settings.walletSettings // Try to read wallet from json file or test mnemonic provided in a config file val newState = ergoWalletService.readWallet(state, ws.testMnemonic.map(SecretString.create(_)), ws.testKeysQty, ws.secretStorage) - context.become(loadedWallet(newState)) + context.become(loadedWallet(ergoWalletService.restoreOffChainState(newState))) unstashAll() case _ => // stashing all messages until wallet is setup stash() @@ -223,14 +227,23 @@ class ErgoWalletActor(settings: ErgoSettings, /* SCAN COMMANDS */ //scan mempool transaction case ScanOffChain(tx) => - val dustLimit = settings.walletSettings.dustLimit - val newWalletBoxes = WalletScanLogic.extractWalletOutputs(tx, None, state.walletVars, dustLimit) - val inputs = WalletScanLogic.extractInputBoxes(tx) - val newState = state.copy(offChainRegistry = - state.offChainRegistry.updateOnTransaction(newWalletBoxes, inputs, state.walletVars.externalScans) - ) + val (newState, walletAffected) = ergoWalletService.scanOffChainUpdate(state, tx) + if (walletAffected) { + // the transaction is kept until it gets on the blockchain, so that a restart in the meantime + // does not make the wallet consider its inputs spendable again + state.storage.addUnconfirmedTransaction(tx, state.fullHeight) match { + case Success(_) => + case Failure(t) => log.error(s"Could not store unconfirmed transaction ${tx.id}: ", t) + } + } context.become(loadedWallet(newState)) + case ReadUnconfirmedTransactions => + sender() ! unconfirmedTransactionsToRestore(state) + + case ForgetUnconfirmedTransactions(ids) => + forgetUnconfirmedTransactions(state, ids) + // rescan=true means we serve a user request for rescan from arbitrary height case ScanInThePast(blockHeight, rescan) => val nextBlockHeight = state.expectedNextBlockHeight(blockHeight, settings.nodeSettings.isFullBlocksPruned) @@ -275,6 +288,7 @@ class ErgoWalletActor(settings: ErgoSettings, case Success(updatedState) => updatedState } + forgetConfirmedAndExpired(newState, newBlock) context.become(loadedWallet(newState)) } else if (nextBlockHeight < newBlock.height) { log.warn(s"Wallet: skipped blocks found starting from $nextBlockHeight, going back to scan them") @@ -489,6 +503,48 @@ class ErgoWalletActor(settings: ErgoSettings, sender() ! txsToSend } + /** + * Stored unconfirmed transactions worth putting back into the memory pool. Transactions which did + * not make it onto the blockchain for too long are dropped instead of being re-submitted forever. + */ + private def unconfirmedTransactionsToRestore(state: ErgoWalletState): Seq[ErgoTransaction] = { + val (fresh, expired) = state.storage.readUnconfirmedTransactions().partition { case (_, seenAt) => + state.fullHeight - seenAt <= WalletStorage.UnconfirmedTxLifetimeInBlocks + } + if (expired.nonEmpty) { + forgetUnconfirmedTransactions(state, expired.map(_._1.id)) + } + ErgoWalletActor.orderByDependency(fresh.map(_._1)) + } + + /** + * Drop the transactions of a just applied block from the store, and give up on the ones which + * stayed unconfirmed for longer than [[WalletStorage.UnconfirmedTxLifetimeInBlocks]] blocks. + */ + private def forgetConfirmedAndExpired(state: ErgoWalletState, block: ErgoFullBlock): Unit = { + val seenAtHeights = state.storage.unconfirmedTransactionHeights + if (seenAtHeights.nonEmpty) { + val confirmed = block.transactions.map(_.id).filter(seenAtHeights.contains) + val expired = seenAtHeights.collect { + case (id, seenAt) if block.height - seenAt > WalletStorage.UnconfirmedTxLifetimeInBlocks => id + }.toSeq + if (expired.nonEmpty) { + log.warn(s"Wallet gave up on ${expired.size} transaction(s) still unconfirmed at height ${block.height}") + } + val toForget = (confirmed ++ expired).distinct + if (toForget.nonEmpty) { + forgetUnconfirmedTransactions(state, toForget) + } + } + } + + private def forgetUnconfirmedTransactions(state: ErgoWalletState, ids: Seq[ModifierId]): Unit = { + state.storage.removeUnconfirmedTransactions(ids) match { + case Success(_) => + case Failure(t) => log.error("Could not forget unconfirmed transactions: ", t) + } + } + override def receive: Receive = emptyWallet private def wrapLegalExc[T](e: Throwable): Failure[T] = @@ -503,6 +559,42 @@ class ErgoWalletActor(settings: ErgoSettings, object ErgoWalletActor extends ScorexLogging { + /** + * Order transactions so that a transaction spending an output of another one comes after it. + * + * Unconfirmed transactions do form such chains, and both consumers of this ordering need it: the + * off-chain registry only nets a spending out if it has seen the box being spent already, and the + * memory pool refuses a transaction whose inputs it does not know yet. + * + * Transactions with no producer among `txs` keep their relative order. A cycle is impossible + * between valid transactions, but should one be given, its members are appended unordered rather + * than dropped. + */ + def orderByDependency(txs: Seq[ErgoTransaction]): Seq[ErgoTransaction] = { + val producerOf: Map[ModifierId, ModifierId] = + txs.flatMap(tx => tx.outputs.map(out => bytesToId(out.id) -> tx.id)).toMap + + @tailrec + def loop(remaining: Seq[ErgoTransaction], + ordered: Set[ModifierId], + acc: Seq[ErgoTransaction]): Seq[ErgoTransaction] = { + if (remaining.isEmpty) { + acc + } else { + val (ready, blocked) = remaining.partition { tx => + tx.inputs.forall(in => producerOf.get(bytesToId(in.boxId)).forall(ordered.contains)) + } + if (ready.isEmpty) { + acc ++ blocked + } else { + loop(blocked, ordered ++ ready.map(_.id), acc ++ ready) + } + } + } + + loop(txs, Set.empty, Seq.empty) + } + /** Start actor and register its proper closing into coordinated shutdown */ def apply(settings: ErgoSettings, parameters: Parameters, diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActorMessages.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActorMessages.scala index a5b3d470fe..9fb07e7c68 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActorMessages.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletActorMessages.scala @@ -51,6 +51,21 @@ object ErgoWalletActorMessages { */ final case class ScanOnChain(block: ErgoFullBlock) + /** + * Read wallet-related transactions which were not on the blockchain yet when the node was stopped, + * so that they can be put back into the memory pool. Answered with a `Seq[ErgoTransaction]`, + * ordered so that a transaction spending an output of another one comes after it. + */ + final case object ReadUnconfirmedTransactions + + /** + * Stop keeping the given unconfirmed transactions across restarts, e.g. because the memory pool + * refused them + * + * @param ids - identifiers of the transactions to forget + */ + final case class ForgetUnconfirmedTransactions(ids: Seq[ModifierId]) + /** * Rollback to previous version of the wallet, by throwing away effects of blocks after the version * diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletReader.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletReader.scala index c566a6fd15..4c3face3db 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletReader.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletReader.scala @@ -97,6 +97,14 @@ trait ErgoWalletReader extends NodeViewComponent { def transactionById(id: ModifierId): Future[Option[AugWalletTransaction]] = (walletActor ? GetTransaction(id)).mapTo[Option[AugWalletTransaction]] + /** + * Wallet-related transactions which were not on the blockchain yet when the node was stopped, so + * that they can be put back into the memory pool. Ordered so that a transaction spending an + * output of another one comes after it. + */ + def unconfirmedTransactionsToRestore: Future[Seq[ErgoTransaction]] = + (walletActor ? ReadUnconfirmedTransactions).mapTo[Seq[ErgoTransaction]] + def generateTransaction(requests: Seq[TransactionGenerationRequest], inputsRaw: Seq[String] = Seq.empty, dataInputsRaw: Seq[String] = Seq.empty): Future[Try[ErgoTransaction]] = diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletService.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletService.scala index 8d4fbc341a..fd89aa4c59 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletService.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/ErgoWalletService.scala @@ -220,6 +220,21 @@ trait ErgoWalletService { */ def scanBlockUpdate(state: ErgoWalletState, block: ErgoFullBlock, dustLimit: Option[Long]): Try[ErgoWalletState] + /** + * Update the off-chain part of the wallet with a transaction which is not on the blockchain yet. + * + * @return the updated state, and whether the transaction is of any interest to the wallet, that + * is whether it pays the wallet or spends one of its boxes + */ + def scanOffChainUpdate(state: ErgoWalletState, tx: ErgoTransaction): (ErgoWalletState, Boolean) + + /** + * Rebuild the off-chain part of the wallet from the transactions which were unconfirmed when the + * node was stopped. Without it the wallet forgets that it already spent their inputs and can + * build a conflicting transaction spending the same boxes a second time. + */ + def restoreOffChainState(state: ErgoWalletState): ErgoWalletState + /** * Sign a transaction */ @@ -596,6 +611,33 @@ class ErgoWalletServiceImpl(override val ergoSettings: ErgoSettings) extends Erg state.copy(registry = reg, offChainRegistry = offReg, outputsFilter = Some(updatedOutputsFilter)) } + override def scanOffChainUpdate(state: ErgoWalletState, tx: ErgoTransaction): (ErgoWalletState, Boolean) = { + val dustLimit = ergoSettings.walletSettings.dustLimit + val newWalletBoxes = WalletScanLogic.extractWalletOutputs(tx, None, state.walletVars, dustLimit) + val inputs = WalletScanLogic.extractInputBoxes(tx) + + def spendsWalletBox: Boolean = + state.offChainRegistry.offChainBoxes.exists(box => inputs.contains(box.boxId)) || + tx.inputs.exists(input => state.registry.getBox(input.boxId).isDefined) + + val newState = state.copy(offChainRegistry = + state.offChainRegistry.updateOnTransaction(newWalletBoxes, inputs, state.walletVars.externalScans) + ) + (newState, newWalletBoxes.nonEmpty || spendsWalletBox) + } + + override def restoreOffChainState(state: ErgoWalletState): ErgoWalletState = { + val stored = state.storage.readUnconfirmedTransactions().map(_._1) + if (stored.isEmpty) { + state + } else { + log.info(s"Wallet is restoring ${stored.size} unconfirmed transaction(s) kept across restart") + ErgoWalletActor.orderByDependency(stored).foldLeft(state) { case (acc, tx) => + scanOffChainUpdate(acc, tx)._1 + } + } + } + override def updateUtxoState(state: ErgoWalletState): ErgoWalletState = { (state.mempoolReaderOpt, state.stateReaderOpt) match { case (Some(mr), Some(sr)) => From 31947a79590a34be9b78876b976e4d11da8f01a9 Mon Sep 17 00:00:00 2001 From: Ergologica <153913412+Ergologica@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:21:42 +0200 Subject: [PATCH 3/5] Store unconfirmed wallet txs --- .../wallet/persistence/WalletStorage.scala | 95 ++++++++++++++++++- 1 file changed, 94 insertions(+), 1 deletion(-) diff --git a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorage.scala b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorage.scala index 09eb04bc2f..aff5d43fe9 100644 --- a/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorage.scala +++ b/src/main/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorage.scala @@ -2,6 +2,7 @@ package org.ergoplatform.nodeView.wallet.persistence import com.google.common.primitives.{Ints, Shorts} import org.ergoplatform.P2PKAddress +import org.ergoplatform.modifiers.mempool.{ErgoTransaction, ErgoTransactionSerializer} import org.ergoplatform.nodeView.state.{ErgoStateContext, ErgoStateContextSerializer} import org.ergoplatform.nodeView.wallet.scanning.{Scan, ScanRequest, ScanSerializer} import org.ergoplatform.sdk.wallet.secrets.{DerivationPath, DerivationPathSerializer, ExtendedPublicKey, ExtendedPublicKeySerializer} @@ -9,7 +10,7 @@ import org.ergoplatform.settings.{Constants, ErgoSettings, Parameters} import org.ergoplatform.wallet.Constants.{PaymentsScanId, ScanId} import scorex.crypto.hash.Blake2b256 import scorex.db.{LDBFactory, LDBKVStore} -import scorex.util.ScorexLogging +import scorex.util.{ModifierId, ScorexLogging, idToBytes} import sigma.serialization.SigmaSerializer import java.io.File @@ -24,6 +25,7 @@ import scala.util.{Failure, Success, Try} * * changed addresses * * ErgoStateContext (not version-agnostic, but state changes including rollbacks it is updated externally) * * external scans + * * wallet-related transactions which are not on the blockchain yet */ final class WalletStorage(store: LDBKVStore, settings: ErgoSettings) extends ScorexLogging { @@ -194,6 +196,76 @@ final class WalletStorage(store: LDBKVStore, settings: ErgoSettings) extends Sco .getOrElse(PaymentsScanId) } + /** + * Heights at which wallet-related unconfirmed transactions stored in the database were first seen, + * by transaction identifier. Kept in memory so that applying a block does not need a database + * lookup per transaction of the block, and so that pruning does not need to read the whole bucket. + */ + private var cachedUnconfirmedTxHeights: Option[Map[ModifierId, Int]] = None + + /** + * @return heights at which stored unconfirmed transactions were first seen, by transaction id + */ + def unconfirmedTransactionHeights: Map[ModifierId, Int] = cachedUnconfirmedTxHeights.getOrElse { + val heights = readUnconfirmedTransactions().map { case (tx, height) => tx.id -> height }.toMap + cachedUnconfirmedTxHeights = Some(heights) + heights + } + + /** + * Store a wallet-related transaction which is not on the blockchain yet, so that it survives + * a node restart. + * + * A transaction already stored is left alone rather than re-dated: it is re-scanned every time it + * is put back into the memory pool, and refreshing its height there would push its expiry back on + * every restart, so a transaction which never confirms would be kept forever. + * + * @param tx - unconfirmed transaction the wallet is interested in + * @param seenAtHeight - blockchain height at the moment the transaction was first seen + */ + def addUnconfirmedTransaction(tx: ErgoTransaction, seenAtHeight: Int): Try[Unit] = { + if (unconfirmedTransactionHeights.contains(tx.id)) { + Success(()) + } else { + store.insert(unconfirmedTxKey(tx.id), Ints.toByteArray(seenAtHeight) ++ tx.bytes).map { _ => + cachedUnconfirmedTxHeights = Some(unconfirmedTransactionHeights.updated(tx.id, seenAtHeight)) + } + } + } + + /** + * Forget stored unconfirmed transactions, e.g. once they got on the blockchain. Identifiers of + * transactions which are not stored are ignored. + */ + def removeUnconfirmedTransactions(ids: Seq[ModifierId]): Try[Unit] = { + val known = unconfirmedTransactionHeights + val toRemove = ids.filter(known.contains) + if (toRemove.isEmpty) { + Success(()) + } else { + store.remove(toRemove.map(unconfirmedTxKey).toArray).map { _ => + cachedUnconfirmedTxHeights = Some(known -- toRemove) + } + } + } + + /** + * Read unconfirmed transactions stored, along with the height each of them was seen at. + * Records which can not be parsed are skipped, a corrupted record must not prevent the wallet + * from starting. + */ + def readUnconfirmedTransactions(): Seq[(ErgoTransaction, Int)] = { + store.getRange(FirstUnconfirmedTxId, LastUnconfirmedTxId).flatMap { case (_, v) => + ErgoTransactionSerializer.parseBytesTry(v.drop(java.lang.Integer.BYTES)) match { + case Success(tx) => + Some(tx -> Ints.fromByteArray(v.take(java.lang.Integer.BYTES))) + case Failure(t) => + log.error("Corrupted data when reading an unconfirmed transaction: ", t) + None + } + } + } + /** * Close wallet storage database */ @@ -220,8 +292,15 @@ object WalletStorage { */ val PublicKeyPrefixByte: Byte = 2: Byte + /** + * Secondary prefix byte for the bucket of wallet-related transactions which are not on the + * blockchain yet + */ + val UnconfirmedTxPrefixByte: Byte = 3: Byte + val ScanPrefixArray: Array[Byte] = Array(RangedKeyPrefix, ScanPrefixByte) val PublicKeyPrefixArray: Array[Byte] = Array(RangedKeyPrefix, PublicKeyPrefixByte) + val UnconfirmedTxPrefixArray: Array[Byte] = Array(RangedKeyPrefix, UnconfirmedTxPrefixByte) // scans key space to iterate over all of them val SmallestPossibleScanId: Array[Byte] = ScanPrefixArray ++ Shorts.toByteArray(0) @@ -236,6 +315,20 @@ object WalletStorage { val FirstPublicKeyId: Array[Byte] = PublicKeyPrefixArray ++ Array.fill(33)(0: Byte) val LastPublicKeyId: Array[Byte] = PublicKeyPrefixArray ++ Array.fill(33)(-1: Byte) + def unconfirmedTxKey(txId: ModifierId): Array[Byte] = UnconfirmedTxPrefixArray ++ idToBytes(txId) + + // unconfirmed transactions space to iterate over all of them + val FirstUnconfirmedTxId: Array[Byte] = UnconfirmedTxPrefixArray ++ Array.fill(32)(0: Byte) + val LastUnconfirmedTxId: Array[Byte] = UnconfirmedTxPrefixArray ++ Array.fill(32)(-1: Byte) + + /** + * For how many blocks a wallet-related unconfirmed transaction is kept in the database before + * being given up on. A transaction which did not get on the blockchain within this many blocks + * is most likely never going to, and re-submitting it forever would only keep the wallet from + * spending its inputs. Deliberately conservative (about two days at the target block rate). + */ + val UnconfirmedTxLifetimeInBlocks: Int = 1440 + def noPrefixKey(keyString: String): Array[Byte] = Blake2b256.hash(keyString) //following keys do not start with ranged key prefix, i.e. with 8 zero bits From 4bfd6afa2265b59e44b0103166e958160c86ddf8 Mon Sep 17 00:00:00 2001 From: Ergologica <153913412+Ergologica@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:22:19 +0200 Subject: [PATCH 4/5] Test wallet side of the restart --- .../wallet/ErgoWalletServiceSpec.scala | 116 ++++++++++++++++++ .../nodeView/wallet/ErgoWalletSpec.scala | 30 +++++ 2 files changed, 146 insertions(+) diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala index cc261eebc5..c1b0b97122 100644 --- a/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletServiceSpec.scala @@ -433,4 +433,120 @@ class ErgoWalletServiceSpec } } + /** + * An on-chain, unspent box of the wallet, and a transaction spending it which stays unconfirmed. + */ + private def walletBoxAndSpendingTx(registry: WalletRegistry): (TrackedBox, ErgoTransaction) = { + val walletBox = TrackedBox( + creationTxId = modifierIdGen.sample.get, + creationOutIndex = 0, + inclusionHeightOpt = Some(100), + spendingTxIdOpt = None, + spendingHeightOpt = None, + box = testBox(1000000000L, pks.head.script, 100), + scans = Set(PaymentsScanId) + ) + registry.updateOnBlock( + ScanResults(Seq(walletBox), ArraySeq.empty, ArraySeq.empty), modifierIdGen.sample.get, blockHeight = 100).get + + val spendingTx = ErgoTransaction( + IndexedSeq(Input(walletBox.box.id, emptyProverResult)), + IndexedSeq(new ErgoBoxCandidate(walletBox.box.value, TrueTree, creationHeight = 100)) + ) + walletBox -> spendingTx + } + + property("a box spent by an unconfirmed transaction is not offered for spending again") { + withVersionedStore(2) { versionedStore => + withStore { store => + val walletService = new ErgoWalletServiceImpl(settings) + val wState = initialState(store, versionedStore) + val (walletBox, spendingTx) = walletBoxAndSpendingTx(wState.registry) + + val onChain = wState.copy(offChainRegistry = OffChainRegistry.init(wState.registry)) + onChain.walletFilter(walletBox) shouldBe true + + // the spending transaction pays a script the wallet does not track, so the only reason for + // the wallet to care about it is that it spends one of its boxes + val (offChain, walletAffected) = walletService.scanOffChainUpdate(onChain, spendingTx) + walletAffected shouldBe true + offChain.walletFilter(walletBox) shouldBe false + } + } + } + + property("unconfirmed transactions are replayed on restart, so their inputs stay spent") { + withVersionedStore(2) { versionedStore => + withStore { store => + val walletService = new ErgoWalletServiceImpl(settings) + val wState = initialState(store, versionedStore) + val (walletBox, spendingTx) = walletBoxAndSpendingTx(wState.registry) + + wState.storage.addUnconfirmedTransaction(spendingTx, seenAtHeight = 100).get + + // a restart: the off-chain registry is rebuilt from the wallet registry alone, which still + // lists the box as unspent - the spending transaction never got onto the blockchain. Left + // like this the wallet would happily spend the box a second time (issue #1154) + val restarted = initialState(store, versionedStore) + .copy(offChainRegistry = OffChainRegistry.init(wState.registry)) + restarted.walletFilter(walletBox) shouldBe true + + walletService.restoreOffChainState(restarted).walletFilter(walletBox) shouldBe false + } + } + } + + property("a restored unconfirmed transaction is dropped once it gets on the blockchain") { + withVersionedStore(2) { versionedStore => + withStore { store => + val walletService = new ErgoWalletServiceImpl(settings) + val wState = initialState(store, versionedStore) + val (_, spendingTx) = walletBoxAndSpendingTx(wState.registry) + + wState.storage.addUnconfirmedTransaction(spendingTx, seenAtHeight = 100).get + wState.storage.unconfirmedTransactionHeights shouldBe Map(spendingTx.id -> 100) + + wState.storage.removeUnconfirmedTransactions(Seq(spendingTx.id)).get + wState.storage.readUnconfirmedTransactions() shouldBe empty + + // nothing left to replay, so the state comes back untouched + val restarted = initialState(store, versionedStore) + walletService.restoreOffChainState(restarted) shouldBe restarted + } + } + } + + property("chained unconfirmed transactions are replayed parent first") { + withVersionedStore(2) { versionedStore => + withStore { store => + val walletService = new ErgoWalletServiceImpl(settings) + val wState = initialState(store, versionedStore) + val (walletBox, parentTx) = walletBoxAndSpendingTx(wState.registry) + + // a child spending the output of the parent, stored - and hence read back - in the wrong order + val childTx = ErgoTransaction( + IndexedSeq(Input(parentTx.outputs.head.id, emptyProverResult)), + IndexedSeq(new ErgoBoxCandidate(parentTx.outputs.head.value, TrueTree, creationHeight = 100)) + ) + ErgoWalletActor.orderByDependency(Seq(childTx, parentTx)).map(_.id) shouldBe Seq(parentTx.id, childTx.id) + + wState.storage.addUnconfirmedTransaction(childTx, seenAtHeight = 100).get + wState.storage.addUnconfirmedTransaction(parentTx, seenAtHeight = 100).get + + val restarted = initialState(store, versionedStore) + .copy(offChainRegistry = OffChainRegistry.init(wState.registry)) + walletService.restoreOffChainState(restarted).walletFilter(walletBox) shouldBe false + } + } + } + + property("orderByDependency keeps independent transactions in order and tolerates cycles") { + forAll(Gen.nonEmptyListOf(validErgoTransactionGen)) { generated => + // transactions generated independently spend boxes none of them creates + val txs = generated.map(_._2) + ErgoWalletActor.orderByDependency(txs).map(_.id) shouldBe txs.map(_.id) + ErgoWalletActor.orderByDependency(Seq.empty) shouldBe Seq.empty + } + } + } diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSpec.scala index 772734c811..0d0f1ad58a 100644 --- a/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/ErgoWalletSpec.scala @@ -95,6 +95,36 @@ class ErgoWalletSpec extends ErgoCorePropertyTest with WalletTestOps with Eventu } } + property("keep an off-chain transaction so it can be put back into the memory pool after a restart") { + withFixture { implicit w => + val addresses = getPublicKeys + addresses.length should be > 0 + val genesisBlock = makeGenesisBlock(addresses.head.pubkey, randomNewAsset) + applyBlock(genesisBlock) shouldBe 'success //scan by wallet happens during apply + implicit val patienceConfig: PatienceConfig = PatienceConfig(5.second, 300.millis) + + await(wallet.unconfirmedTransactionsToRestore) shouldBe empty + + val tx = eventually { + val snap = getConfirmedBalances + val req = Seq(PaymentRequest(addresses.head, snap.walletBalance / 2, Array.empty, Map.empty)) + await(wallet.generateTransaction(req)).get + } + wallet.scanOffchain(tx) + + // the memory pool is not persisted, so the wallet keeps its own unconfirmed transactions + eventually { + await(wallet.unconfirmedTransactionsToRestore).map(_.id) shouldBe Seq(tx.id) + } + + // and stops keeping them once they are on the blockchain + applyBlock(makeNextBlock(getUtxoState, Seq(tx))) shouldBe 'success + eventually { + await(wallet.unconfirmedTransactionsToRestore) shouldBe empty + } + } + } + property("Generate asset issuing transaction") { withFixture { implicit w => val address = getPublicKeys.head From 090fc7cf2694aeb47d851acb15e18e0edf750e85 Mon Sep 17 00:00:00 2001 From: Ergologica <153913412+Ergologica@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:23:00 +0200 Subject: [PATCH 5/5] Test unconfirmed tx storage --- .../persistence/WalletStorageSpec.scala | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorageSpec.scala b/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorageSpec.scala index c9909854ca..6337038286 100644 --- a/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorageSpec.scala +++ b/src/test/scala/org/ergoplatform/nodeView/wallet/persistence/WalletStorageSpec.scala @@ -18,6 +18,8 @@ class WalletStorageSpec with DBSpec { import org.ergoplatform.utils.ErgoNodeTestConstants._ import org.ergoplatform.utils.generators.ErgoNodeWalletGenerators._ + import org.ergoplatform.utils.generators.ErgoNodeTransactionGenerators.validErgoTransactionGen + import org.ergoplatform.utils.generators.CoreObjectGenerators.modifierIdGen import org.ergoplatform.wallet.utils.WalletGenerators._ it should "add and read derivation paths" in { @@ -68,6 +70,45 @@ class WalletStorageSpec } } + it should "add, read and forget unconfirmed transactions" in { + forAll(validErgoTransactionGen, extendedPubKeyListGen, externalScanReqGen) { + case ((_, tx), pubKeys, scanReq) => + withStore { store => + val storage = new WalletStorage(store, settings) + storage.readUnconfirmedTransactions() shouldBe empty + storage.unconfirmedTransactionHeights shouldBe empty + + storage.addUnconfirmedTransaction(tx, 100).get + storage.readUnconfirmedTransactions().map { case (t, h) => t.id -> h } shouldBe Seq(tx.id -> 100) + storage.unconfirmedTransactionHeights shouldBe Map(tx.id -> 100) + + // reading the database again is what happens on a node restart, and is the whole point + new WalletStorage(store, settings) + .readUnconfirmedTransactions().map { case (t, h) => t.id -> h } shouldBe Seq(tx.id -> 100) + + // storing it again keeps the height it was first seen at, so that expiry is not pushed back + storage.addUnconfirmedTransaction(tx, 200).get + storage.unconfirmedTransactionHeights shouldBe Map(tx.id -> 100) + + // the new bucket does not overlap with the ones already in this database + pubKeys.foreach(storage.addPublicKey(_).get) + storage.addScan(scanReq).get + storage.readAllKeys() should contain theSameElementsAs pubKeys.toSet + storage.allScans.length shouldBe 1 + storage.readUnconfirmedTransactions().length shouldBe 1 + + // forgetting a transaction which is not there changes nothing + storage.removeUnconfirmedTransactions(Seq(modifierIdGen.sample.get)).get + storage.readUnconfirmedTransactions().length shouldBe 1 + + storage.removeUnconfirmedTransactions(Seq(tx.id)).get + storage.readUnconfirmedTransactions() shouldBe empty + storage.unconfirmedTransactionHeights shouldBe empty + new WalletStorage(store, settings).readUnconfirmedTransactions() shouldBe empty + } + } + } + it should "always increase ids" in { forAll(externalScanReqGen) { externalScanReq => withStore { store =>