Skip to content

Commit 200ae7d

Browse files
oleonovoleonov
committed
Expire stuck never-broadcast pending transactions and free their inputs
Extend PendingTransactionReconciler to remove NEW (never-broadcast) outgoing transactions whose funds never reached the network: once Blockchair confirms the transaction is absent and it is older than one hour, delete it and free the funding UTXOs (delete path, no failedToSpend). Only transactions that never entered the send pipeline (no SentTransaction) are NEW-expired, and recording a broadcast attempt is an atomic conditional insert, so an expiry-delete can never race an in-flight broadcast. RELAYED handling is unchanged.
1 parent 8e06f97 commit 200ae7d

10 files changed

Lines changed: 710 additions & 65 deletions

File tree

‎bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/BitcoinCoreBuilder.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -485,9 +485,9 @@ class BitcoinCoreBuilder {
485485
storage = storage,
486486
statusProvider = BlockchairPendingTransactionStatusProvider.create(blockchairApi, network.blockchairChainId),
487487
dataListener = dataProvider,
488-
invalidateOutgoing = invalidator::invalidate,
488+
outgoingInvalidator = invalidator,
489489
logTag = network.logTag,
490-
coroutineDispatcher = coroutineDispatcher
490+
coroutineDispatcher = coroutineDispatcher,
491491
)
492492

493493
val isShared = sharedPeerGroupHolder != null

‎bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/core/Interfaces.kt‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,15 @@ interface IStorage {
131131
fun getRelayedPendingTransactions(status: Int): List<Transaction>
132132
fun getIncomingPendingTxHashes(): List<ByteArray>
133133
fun incomingPendingTransactionsExist(): Boolean
134-
fun deleteRelayedPendingTransactions(transactions: List<Transaction>): List<Transaction>
134+
fun deleteRelayedPendingTransactions(
135+
transactions: List<Transaction>,
136+
expectedStatus: Int = Transaction.Status.RELAYED
137+
): List<Transaction>
138+
139+
// Deletes only NEW transactions that have no SentTransaction record, i.e. transactions whose
140+
// bytes were never handed to the network. Re-checked atomically per transaction inside the
141+
// delete: a concurrent send that records a SentTransaction blocks the delete.
142+
fun deleteNewExpiredTransactions(transactions: List<Transaction>): List<Transaction>
135143

136144
// InvalidTransaction
137145

@@ -192,6 +200,13 @@ interface IStorage {
192200
fun updateSentTransaction(transaction: SentTransaction)
193201
fun deleteSentTransaction(transaction: SentTransaction)
194202

203+
// Atomically records a broadcast attempt only if the transaction is still pending (exists,
204+
// unmined, status == NEW): the pending check and the SentTransaction insert happen inside one
205+
// DB transaction, so this method and a concurrent deleteNewExpiredTransactions can never both
206+
// "win" - whichever commits first decides the outcome for the other. Returns false, writing
207+
// nothing, when the transaction was deleted or mined concurrently.
208+
fun recordBroadcastAttemptIfPending(transaction: SentTransaction): Boolean
209+
195210
fun getChainWork(block: Block): BigInteger
196211
}
197212

‎bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/network/peer/task/SendTransactionTask.kt‎

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.horizontalsystems.bitcoincore.network.peer.task
22

3+
import io.horizontalsystems.bitcoincore.core.IStorage
34
import io.horizontalsystems.bitcoincore.extensions.toReversedHex
45
import io.horizontalsystems.bitcoincore.models.InventoryItem
56
import io.horizontalsystems.bitcoincore.network.messages.GetDataMessage
@@ -9,7 +10,13 @@ import io.horizontalsystems.bitcoincore.network.messages.TransactionMessage
910
import io.horizontalsystems.bitcoincore.storage.FullTransaction
1011
import java.util.concurrent.TimeUnit
1112

12-
class SendTransactionTask(val transaction: FullTransaction) : PeerTask() {
13+
// storage/external are nullable/default so existing tests that construct this task directly for
14+
// unrelated protocol behavior keep compiling; production always supplies both (see TransactionSender).
15+
class SendTransactionTask(
16+
val transaction: FullTransaction,
17+
private val storage: IStorage? = null,
18+
private val external: Boolean = false,
19+
) : PeerTask() {
1320

1421
enum class CompletionReason {
1522
REQUESTED_BY_PEER,
@@ -36,18 +43,40 @@ class SendTransactionTask(val transaction: FullTransaction) : PeerTask() {
3643
message is GetDataMessage &&
3744
message.inventory.any { it.type == InventoryItem.MSG_TX && it.hash.contentEquals(transaction.header.hash) }
3845

39-
if (transactionRequested) {
46+
if (!transactionRequested) {
47+
return false
48+
}
49+
50+
if (isStillBroadcastable()) {
4051
completionReason = CompletionReason.REQUESTED_BY_PEER
4152
requester?.send(TransactionMessage(transaction, 0))
42-
listener?.onTaskCompleted(this)
53+
} else {
54+
// Own transaction was deleted from storage (expired without ever broadcasting) or is
55+
// already mined: let the peer's request go unanswered instead of sending bytes for a
56+
// transaction we no longer own.
57+
completionReason = CompletionReason.TIMEOUT
4358
}
59+
listener?.onTaskCompleted(this)
4460

45-
return transactionRequested
61+
return true
4662
}
4763

4864
override fun handleTimeout() {
4965
completionReason = CompletionReason.TIMEOUT
5066
listener?.onTaskCompleted(this)
5167
}
5268

69+
// Plain existence/pending re-check: every peer's getdata for a still-pending transaction is
70+
// served (no ownership gate), only a deleted (expired) or mined transaction is refused. No lock
71+
// is needed here - PendingTransactionReconciler's NEW-expiry delete is blocked structurally by
72+
// the SentTransaction record TransactionSender writes before this task is even started.
73+
private fun isStillBroadcastable(): Boolean {
74+
val trackedStorage = storage ?: return true
75+
if (external) {
76+
return true
77+
}
78+
val stored = trackedStorage.getTransaction(transaction.header.hash) ?: return false
79+
return stored.blockHash == null
80+
}
81+
5382
}

‎bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/storage/Storage.kt‎

Lines changed: 60 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -438,31 +438,65 @@ open class Storage(protected open val store: CoreDatabase) : IStorage {
438438
return store.transaction.getIncomingPendingTxCount() > 0
439439
}
440440

441-
override fun deleteRelayedPendingTransactions(transactions: List<Transaction>): List<Transaction> {
441+
override fun deleteRelayedPendingTransactions(
442+
transactions: List<Transaction>,
443+
expectedStatus: Int
444+
): List<Transaction> {
442445
val deletedTransactions = mutableListOf<Transaction>()
443446

444447
store.runInTransaction {
445448
transactions.forEach { transaction ->
446-
// The transaction may be confirmed while the reconciler is doing network lookup.
449+
// The transaction's status may have changed (confirmed or relayed) while the
450+
// reconciler was doing its network lookup, so re-check it here.
447451
val pendingTransaction = store.transaction.getByHash(transaction.hash)
448-
?.takeIf { it.blockHash == null && it.status == Transaction.Status.RELAYED }
452+
?.takeIf { it.blockHash == null && it.status == expectedStatus }
449453
?: return@forEach
450454

451-
store.sentTransaction.getTransaction(pendingTransaction.hash)?.let {
452-
store.sentTransaction.delete(it)
455+
deletePendingTransaction(pendingTransaction)
456+
deletedTransactions += pendingTransaction
457+
}
458+
}
459+
460+
return deletedTransactions
461+
}
462+
463+
// A NEW transaction is only deleted while it has no SentTransaction record, i.e. its bytes were
464+
// never handed to the network. The SentTransaction check happens inside the same DB transaction
465+
// as the delete, atomically with the re-read of the transaction's current status: a sender that
466+
// records a SentTransaction concurrently (see TransactionSender.recordBroadcastAttempt, called
467+
// before bytes reach a peer or the API) always wins the race and blocks the delete.
468+
override fun deleteNewExpiredTransactions(transactions: List<Transaction>): List<Transaction> {
469+
val deletedTransactions = mutableListOf<Transaction>()
470+
471+
store.runInTransaction {
472+
transactions.forEach { transaction ->
473+
val pendingTransaction = store.transaction.getByHash(transaction.hash)
474+
?.takeIf { it.blockHash == null && it.status == Transaction.Status.NEW }
475+
?: return@forEach
476+
477+
if (store.sentTransaction.getTransaction(pendingTransaction.hash) != null) {
478+
return@forEach
453479
}
454480

455-
store.input.deleteAll(getTransactionInputs(pendingTransaction))
456-
store.output.deleteAll(getTransactionOutputs(pendingTransaction))
457-
store.transactionMetadata.delete(pendingTransaction.hash)
458-
store.transaction.delete(pendingTransaction)
481+
deletePendingTransaction(pendingTransaction)
459482
deletedTransactions += pendingTransaction
460483
}
461484
}
462485

463486
return deletedTransactions
464487
}
465488

489+
private fun deletePendingTransaction(pendingTransaction: Transaction) {
490+
store.sentTransaction.getTransaction(pendingTransaction.hash)?.let {
491+
store.sentTransaction.delete(it)
492+
}
493+
494+
store.input.deleteAll(getTransactionInputs(pendingTransaction))
495+
store.output.deleteAll(getTransactionOutputs(pendingTransaction))
496+
store.transactionMetadata.delete(pendingTransaction.hash)
497+
store.transaction.delete(pendingTransaction)
498+
}
499+
466500
private fun convertToFullTransaction(transaction: Transaction): FullTransaction {
467501
val transactionSerializer = requireNotNull(transactionSerializer)
468502
return FullTransaction(
@@ -667,6 +701,23 @@ open class Storage(protected open val store: CoreDatabase) : IStorage {
667701
store.sentTransaction.delete(transaction)
668702
}
669703

704+
override fun recordBroadcastAttemptIfPending(transaction: SentTransaction): Boolean {
705+
var recorded = false
706+
707+
store.runInTransaction {
708+
val isPending = store.transaction.getByHash(transaction.hash)
709+
?.let { it.blockHash == null && it.status == Transaction.Status.NEW }
710+
?: false
711+
712+
if (isPending) {
713+
store.sentTransaction.insert(transaction)
714+
recorded = true
715+
}
716+
}
717+
718+
return recorded
719+
}
720+
670721
override fun getChainWork(block: Block): BigInteger {
671722
var totalWork = BigInteger.ZERO
672723
var currentBlock: Block? = block

‎bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/transactions/PendingTransactionReconciler.kt‎

Lines changed: 36 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -127,10 +127,11 @@ class PendingTransactionReconciler(
127127
private val storage: IStorage,
128128
private val statusProvider: PendingTransactionStatusProvider?,
129129
private val dataListener: IBlockchainDataListener,
130-
private val invalidateOutgoing: (Transaction) -> Unit,
130+
private val outgoingInvalidator: OutgoingTransactionInvalidator,
131131
private val logTag: String,
132132
private val coroutineDispatcher: CoroutineDispatcher,
133-
private val minimumPendingAgeSeconds: Long = DEFAULT_MINIMUM_PENDING_AGE_SECONDS
133+
private val minimumPendingAgeSeconds: Long = DEFAULT_MINIMUM_PENDING_AGE_SECONDS,
134+
private val minimumNewPendingAgeSeconds: Long = DEFAULT_MINIMUM_NEW_PENDING_AGE_SECONDS,
134135
) {
135136
private var coroutineScope = createCoroutineScope()
136137
private val running = AtomicBoolean(false)
@@ -163,7 +164,8 @@ class PendingTransactionReconciler(
163164

164165
suspend fun reconcile(nowSeconds: Long = System.currentTimeMillis() / 1000) {
165166
val statusProvider = statusProvider ?: return
166-
val pendingTransactions = storage.getRelayedPendingTransactions(Transaction.Status.RELAYED)
167+
val pendingTransactions = storage.getRelayedPendingTransactions(Transaction.Status.RELAYED) +
168+
storage.getRelayedPendingTransactions(Transaction.Status.NEW)
167169

168170
if (pendingTransactions.isEmpty()) {
169171
return
@@ -197,8 +199,10 @@ class PendingTransactionReconciler(
197199
}
198200
}
199201

202+
// Restricted to RELAYED: a malformed NEW transaction (never broadcast) instead ages out through
203+
// the normal NEW-expiry path below, which is short enough (minimumNewPendingAgeSeconds) on its own.
200204
private fun isMalformedOutgoingTransaction(transaction: Transaction): Boolean {
201-
if (!transaction.isOutgoing) {
205+
if (!transaction.isOutgoing || transaction.status != Transaction.Status.RELAYED) {
202206
return false
203207
}
204208

@@ -218,7 +222,12 @@ class PendingTransactionReconciler(
218222
}
219223

220224
private fun Transaction.isStale(nowSeconds: Long): Boolean {
221-
return nowSeconds - timestamp >= minimumPendingAgeSeconds
225+
val ageThreshold = if (status == Transaction.Status.NEW) {
226+
minimumNewPendingAgeSeconds
227+
} else {
228+
minimumPendingAgeSeconds
229+
}
230+
return nowSeconds - timestamp >= ageThreshold
222231
}
223232

224233
private fun handleDroppedTransactions(transactions: List<Transaction>) {
@@ -227,8 +236,13 @@ class PendingTransactionReconciler(
227236
}
228237

229238
val (outgoing, incoming) = transactions.partition { it.isOutgoing }
230-
outgoing.forEach(invalidateOutgoing)
231-
deleteIncomingTransactions(incoming)
239+
val (newOutgoing, relayedOutgoing) = outgoing.partition { it.status == Transaction.Status.NEW }
240+
241+
// NEW outgoing transactions never reached the network, so their inputs are freed by
242+
// deleting them outright instead of invalidating (which would mark inputs failedToSpend).
243+
deletePendingTransactions(newOutgoing)
244+
relayedOutgoing.forEach(outgoingInvalidator::invalidate)
245+
deletePendingTransactions(incoming)
232246
}
233247

234248
private fun deleteMalformedTransactions(transactions: List<Transaction>) {
@@ -240,17 +254,26 @@ class PendingTransactionReconciler(
240254
"Deleting malformed relayed outgoing transactions without inputs: " +
241255
transactions.joinToString { it.hash.toReversedHex() }
242256
)
243-
val deletedTransactions = storage.deleteRelayedPendingTransactions(transactions)
244-
notifyDeletedTransactions(deletedTransactions)
257+
deletePendingTransactions(transactions)
245258
}
246259

247-
private fun deleteIncomingTransactions(transactions: List<Transaction>) {
260+
private fun deletePendingTransactions(transactions: List<Transaction>) {
248261
if (transactions.isEmpty()) {
249262
return
250263
}
251264

252-
val deletedTransactions = storage.deleteRelayedPendingTransactions(transactions)
253-
notifyDeletedTransactions(deletedTransactions)
265+
transactions.groupBy { it.status }.forEach { (status, group) ->
266+
val deletedTransactions = if (status == Transaction.Status.NEW) {
267+
// A transaction currently having its bytes handed to the network (API broadcast or
268+
// P2P getdata serve) already has a SentTransaction record (see
269+
// TransactionSender.recordBroadcastAttempt), so storage skips deleting it even
270+
// though it is still NEW - deleting it out from under a real send would clobber it.
271+
storage.deleteNewExpiredTransactions(group)
272+
} else {
273+
storage.deleteRelayedPendingTransactions(group, status)
274+
}
275+
notifyDeletedTransactions(deletedTransactions)
276+
}
254277
}
255278

256279
private fun notifyDeletedTransactions(deletedTransactions: List<Transaction>) {
@@ -282,5 +305,6 @@ class PendingTransactionReconciler(
282305

283306
companion object {
284307
const val DEFAULT_MINIMUM_PENDING_AGE_SECONDS = 24 * 60 * 60L
308+
const val DEFAULT_MINIMUM_NEW_PENDING_AGE_SECONDS = 60 * 60L
285309
}
286310
}

‎bitcoincore/src/main/kotlin/io/horizontalsystems/bitcoincore/transactions/TransactionInvalidator.kt‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,17 @@ import io.horizontalsystems.bitcoincore.models.InvalidTransaction
77
import io.horizontalsystems.bitcoincore.models.Transaction
88
import io.horizontalsystems.bitcoincore.storage.FullTransactionInfo
99

10+
fun interface OutgoingTransactionInvalidator {
11+
fun invalidate(transaction: Transaction)
12+
}
13+
1014
class TransactionInvalidator(
1115
private val storage: IStorage,
1216
private val transactionInfoConverter: ITransactionInfoConverter,
1317
private val listener: IBlockchainDataListener
14-
) {
18+
) : OutgoingTransactionInvalidator {
1519

16-
fun invalidate(transaction: Transaction) {
20+
override fun invalidate(transaction: Transaction) {
1721
val currentTransaction = storage.getTransaction(transaction.hash)
1822
?.takeIf { it.blockHash == null }
1923
?: return

0 commit comments

Comments
 (0)