TxManager

nodecore-grpc/src/main/java/nodecore/api/grpc/utilities/extensions/ByteString.kt

@@ -1,9 +1,11 @@
 package nodecore.api.grpc.utilities.extensions
 
 import com.google.protobuf.ByteString
+import org.veriblock.core.crypto.AnyVbkHash
 import org.veriblock.core.crypto.PreviousBlockVbkHash
 import org.veriblock.core.crypto.PreviousKeystoneVbkHash
 import org.veriblock.core.crypto.VbkHash
+import org.veriblock.core.crypto.asAnyVbkHash
 import org.veriblock.core.crypto.asVbkHash
 import org.veriblock.core.crypto.asVbkPreviousBlockHash
 import org.veriblock.core.crypto.asVbkPreviousKeystoneHash
@@ -46,3 +48,4 @@ fun ByteArray.toByteString(): ByteString =
 fun ByteString.asVbkHash(): VbkHash = toByteArray().asVbkHash()
 fun ByteString.asVbkPreviousBlockHash(): PreviousBlockVbkHash = toByteArray().asVbkPreviousBlockHash()
 fun ByteString.asVbkPreviousKeystoneHash(): PreviousKeystoneVbkHash = toByteArray().asVbkPreviousKeystoneHash()
+fun ByteString.asAnyVbkHash(): AnyVbkHash = toByteArray().asAnyVbkHash()

nodecore-spv/src/main/java/org/veriblock/spv/SpvContext.kt

@@ -21,6 +21,7 @@ import org.veriblock.spv.model.LedgerValue
 import org.veriblock.spv.model.TransactionPool
 import org.veriblock.spv.net.*
 import org.veriblock.spv.service.*
+import org.veriblock.spv.service.tx.TransactionManager
 import org.veriblock.spv.util.AddressStateChangeEvent
 import org.veriblock.spv.util.SpvEventBus.addressStateUpdatedEvent
 import org.veriblock.spv.wallet.PendingTransactionDownloadedListener
@@ -51,7 +52,7 @@ class SpvContext(
     val p2PService: P2PService
     val addressManager: AddressManager
     val transactionService: TransactionService
-    val pendingTransactionContainer: PendingTransactionContainer
+    val transactionManager: TransactionManager
     val pendingTransactionDownloadedListener: PendingTransactionDownloadedListener
 
     private val addressState: ConcurrentHashMap<Address, LedgerContext> = ConcurrentHashMap()
@@ -100,17 +101,17 @@ class SpvContext(
             blockStore = BlockStore(networkParameters, directory)
             transactionPool = TransactionPool()
             blockchain = Blockchain(blockStore)
-            pendingTransactionContainer = PendingTransactionContainer()
-            p2PService = P2PService(pendingTransactionContainer, networkParameters)
+            transactionManager = TransactionManager(blockchain)
+            p2PService = P2PService(transactionManager, networkParameters)
             addressManager = AddressManager()
             val walletFile = File(directory, filePrefix + FILE_EXTENSION)
             addressManager.load(walletFile)
             pendingTransactionDownloadedListener = PendingTransactionDownloadedListener(this)
-            peerTable = SpvPeerTable(this, p2PService, peerDiscovery, pendingTransactionContainer)
+            peerTable = SpvPeerTable(this, p2PService, peerDiscovery)
             transactionService = TransactionService(addressManager, networkParameters)
             spvService = SpvService(
                 this, peerTable, transactionService, addressManager,
-                pendingTransactionContainer, blockchain
+                transactionManager, blockchain
             )
 
             Runtime.getRuntime().addShutdownHook(Thread({ shutdown() }, "ShutdownHook nodecore-spv"))

nodecore-spv/src/main/java/org/veriblock/spv/model/PopTransactionLight.kt

@@ -59,7 +59,7 @@ class PopTransactionLight(
     @Throws(IOException::class)
     private fun serializeToStream(stream: OutputStream) {
         stream.write(transactionTypeIdentifier.id.toInt())
-        inputAddress!!.serializeToStream(stream)
+        inputAddress.serializeToStream(stream)
         SerializeDeserializeService.serialize(endorsedBlock, stream)
         SerializeDeserializeService.serialize(bitcoinTx, stream)
         SerializeDeserializeService.serialize(bitcoinMerklePath, stream)

nodecore-spv/src/main/java/org/veriblock/spv/model/StandardTransaction.kt

@@ -105,7 +105,7 @@ open class StandardTransaction : Transaction {
             builder.type = VeriBlockMessages.Transaction.Type.MULTISIG
         }
         builder.sourceAmount = inputAmount!!.atomicUnits
-        builder.sourceAddress = ByteString.copyFrom(inputAddress!!.toByteArray())
+        builder.sourceAddress = ByteString.copyFrom(inputAddress.toByteArray())
         builder.data = ByteString.copyFrom(data)
         builder.size = toByteArray(networkParameters).size
         for (output in getOutputs()) {
@@ -127,7 +127,7 @@ open class StandardTransaction : Transaction {
         stream.write(transactionTypeIdentifier.id.toInt())
 
         // Write source address
-        inputAddress!!.serializeToStream(stream)
+        inputAddress.serializeToStream(stream)
 
         // Write source amount
         SerializeDeserializeService.serialize(inputAmount!!, stream)

nodecore-spv/src/main/java/org/veriblock/spv/model/Transaction.kt

@@ -14,7 +14,7 @@ import org.veriblock.core.params.NetworkParameters
 
 abstract class Transaction {
     lateinit var txId: VbkTxId
-    var inputAddress: AddressLight? = null
+    lateinit var inputAddress: AddressLight
     var transactionMeta: TransactionMeta? = null
     var signature: ByteArray? = null
     var publicKey: ByteArray? = null

nodecore-spv/src/main/java/org/veriblock/spv/net/P2PService.kt

@@ -3,17 +3,16 @@ package org.veriblock.spv.net
 import com.google.protobuf.ByteString
 import nodecore.api.grpc.VeriBlockMessages
 import org.veriblock.core.utilities.createLogger
-import org.veriblock.core.crypto.Sha256Hash
 import org.veriblock.core.crypto.VbkTxId
 import org.veriblock.core.params.NetworkParameters
 import org.veriblock.spv.model.TransactionTypeIdentifier
-import org.veriblock.spv.service.PendingTransactionContainer
+import org.veriblock.spv.service.tx.TransactionManager
 import org.veriblock.spv.util.nextMessageId
 
 private val logger = createLogger {}
 
 class P2PService(
-    private val pendingTransactionContainer: PendingTransactionContainer,
+    private val pendingTransactionContainer: TransactionManager,
     private val networkParameters: NetworkParameters
 ) {
     fun onTransactionRequest(txIds: List<VbkTxId>, sender: SpvPeer) {

nodecore-spv/src/main/java/org/veriblock/spv/net/SpvPeerTable.kt

@@ -19,19 +19,17 @@ import nodecore.api.grpc.VeriBlockMessages
 import nodecore.api.grpc.VeriBlockMessages.Event.ResultsCase
 import nodecore.api.grpc.VeriBlockMessages.TransactionAnnounce
 import nodecore.api.grpc.utilities.ByteStringUtility
-import nodecore.api.grpc.utilities.extensions.toByteString
-import nodecore.api.grpc.utilities.extensions.toHex
 import org.veriblock.core.crypto.BloomFilter
 import org.veriblock.core.utilities.createLogger
 import org.veriblock.core.crypto.asVbkTxId
 import org.veriblock.core.utilities.debugError
-import org.veriblock.core.utilities.debugWarn
 import org.veriblock.sdk.models.VeriBlockBlock
 import org.veriblock.spv.SpvContext
 import org.veriblock.spv.model.*
 import org.veriblock.spv.serialization.MessageSerializer
 import org.veriblock.spv.serialization.MessageSerializer.deserializeNormalTransaction
 import org.veriblock.spv.service.*
+import org.veriblock.spv.service.tx.TransactionManager
 import org.veriblock.spv.util.SpvEventBus
 import org.veriblock.spv.util.Threading
 import org.veriblock.spv.util.Threading.PEER_TABLE_DISPATCHER
@@ -58,8 +56,7 @@ const val AMOUNT_OF_BLOCKS_WHEN_WE_CAN_START_WORKING = 4//50
 class SpvPeerTable(
     private val spvContext: SpvContext,
     private val p2pService: P2PService,
-    peerDiscovery: PeerDiscovery,
-    pendingTransactionContainer: PendingTransactionContainer
+    peerDiscovery: PeerDiscovery
 ) {
     private val lock = ReentrantLock()
     private val running = AtomicBoolean(false)
@@ -68,7 +65,6 @@ class SpvPeerTable(
     var maximumPeers = DEFAULT_CONNECTIONS
     var downloadPeer: SpvPeer? = null
     val bloomFilter: BloomFilter
-    private val pendingTransactionContainer: PendingTransactionContainer
 
     private val peers = ConcurrentHashMap<NetworkAddress, SpvPeer>()
     private val pendingPeers = ConcurrentHashMap<NetworkAddress, SpvPeer>()
@@ -84,12 +80,10 @@ class SpvPeerTable(
         bloomFilter = createBloomFilter()
         blockchain = spvContext.blockchain
         discovery = peerDiscovery
-        this.pendingTransactionContainer = pendingTransactionContainer
 
-        SpvEventBus.pendingTransactionDownloadedEvent.register(
-            spvContext.pendingTransactionDownloadedListener,
-            spvContext.pendingTransactionDownloadedListener::onPendingTransactionDownloaded
-        )
+        SpvEventBus.pendingTransactionDownloadedEvent.register(this) {
+            logger.debug {"Tx=${it.txId} downloaded..."}
+        }
     }
 
     fun start() {

nodecore-spv/src/main/java/org/veriblock/spv/service/Model.kt

@@ -1,7 +1,9 @@
 package org.veriblock.spv.service
 
+import org.veriblock.core.crypto.AnyVbkHash
 import org.veriblock.core.crypto.Sha256Hash
 import org.veriblock.core.crypto.VbkTxId
+import org.veriblock.core.crypto.VbkHash
 import org.veriblock.sdk.models.Address
 import org.veriblock.sdk.models.Coin
 import org.veriblock.spv.model.AddressLight
@@ -89,7 +91,7 @@ data class TransactionInfo(
     val bitcoinBlockHash: String,
     val bitcoinTxId: String,
     val bitcoinConfirmations: Int,
-    val blockHash: String,
+    val blockHash: AnyVbkHash,
     val merklePath: String
 )
 

nodecore-spv/src/main/java/org/veriblock/spv/service/SpvService.kt

@@ -26,7 +26,6 @@ import org.veriblock.core.utilities.extensions.toHex
 import org.veriblock.core.wallet.AddressManager
 import org.veriblock.core.wallet.AddressPubKey
 import org.veriblock.sdk.models.Address
-import org.veriblock.sdk.models.Coin
 import org.veriblock.sdk.models.asCoin
 import org.veriblock.sdk.services.SerializeDeserializeService
 import org.veriblock.spv.SpvContext
@@ -42,6 +41,8 @@ import org.veriblock.spv.model.Transaction
 import org.veriblock.spv.model.asLightAddress
 import org.veriblock.spv.net.SpvPeerTable
 import org.veriblock.spv.service.TransactionService.Companion.predictAltChainEndorsementTransactionSize
+import org.veriblock.spv.service.tx.TransactionManager
+import org.veriblock.spv.service.tx.TxStatusChangedEvent
 import org.veriblock.spv.util.buildMessage
 import java.io.File
 import java.io.IOException
@@ -54,7 +55,7 @@ class SpvService(
     private val peerTable: SpvPeerTable,
     private val transactionService: TransactionService,
     private val addressManager: AddressManager,
-    private val pendingTransactionContainer: PendingTransactionContainer,
+    private val pendingTransactionContainer: TransactionManager,
     private val blockchain: Blockchain
 ) {
     fun getStateInfo(): StateInfo {
@@ -123,7 +124,14 @@ class SpvService(
             addressCoinsIndexList, outputs
         )
         return transactions.asFlow().onEach {
-            pendingTransactionContainer.addTransaction(it)
+            val event = pendingTransactionContainer.addTransaction(it)
+            event.register(this) { e ->
+                when (e) {
+                    is TxStatusChangedEvent.Confirmation -> logger.info { "Tx=${e.id} has confirmation=${e.confirmations} in a block=${e.block}" }
+                    is TxStatusChangedEvent.GotRequiredConfirmationsN -> logger.info { "Tx=${e.id} is confirmed!" }
+                    is TxStatusChangedEvent.Invalid -> logger.info { "Tx=${e.id} is invalid" }
+                }
+            }
             peerTable.advertise(it)
         }.map {
             it.txId
@@ -348,7 +356,7 @@ class SpvService(
 
     fun getLastBitcoinBlock(): Sha256Hash = spvContext.networkParameters.bitcoinOriginBlock.hash    //Mock todo SPV-111
 
-    fun getTransactions(ids: List<Sha256Hash>) = ids.mapNotNull {
+    fun getTransactions(ids: List<Sha256Hash>): List<TransactionInfo> = ids.mapNotNull {
         pendingTransactionContainer.getTransactionInfo(it)
     }
 

nodecore-spv/src/main/java/org/veriblock/spv/service/TransactionService.kt

@@ -233,7 +233,7 @@ class TransactionService(
             builder.txId = ByteString.copyFrom(tx.txId.bytes)
             builder.type = VeriBlockMessages.Transaction.Type.STANDARD
             builder.sourceAmount = tx.inputAmount!!.atomicUnits
-            builder.sourceAddress = ByteString.copyFrom(tx.inputAddress!!.toByteArray())
+            builder.sourceAddress = ByteString.copyFrom(tx.inputAddress.toByteArray())
             builder.data = ByteString.copyFrom(tx.data)
             //        builder.setTimestamp(getTimeStamp());
             //        builder.setSize(tx.getSize());

nodecore-spv/src/main/java/org/veriblock/spv/service/task/PendingTransactionsUpdateTask.kt

@@ -1,7 +1,9 @@
 package org.veriblock.spv.net
 
-
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.map
 import nodecore.api.grpc.VeriBlockMessages
+import nodecore.api.grpc.utilities.extensions.asAnyVbkHash
 import nodecore.api.grpc.utilities.extensions.toByteString
 import nodecore.api.grpc.utilities.extensions.toHex
 import org.veriblock.core.crypto.asVbkTxId
@@ -29,23 +31,36 @@ fun SpvContext.startPendingTransactionsUpdateTask() {
 }
 
 suspend fun SpvContext.requestPendingTransactions() {
-    val pendingTransactionIds = pendingTransactionContainer.getPendingTransactionIds()
+    val pendingTransactionIds = transactionManager.getPendingTransactionIds()
     try {
         for (txId in pendingTransactionIds) {
             val request = buildMessage {
                 transactionRequest = VeriBlockMessages.GetTransactionRequest.newBuilder()
                     .setId(txId.bytes.toByteString())
                     .build()
             }
-            val response = peerTable.requestMessage(request)
-            if (response.transactionReply.success) {
-                pendingTransactionContainer.updateTransactionInfo(response.transactionReply.transaction.toModel())
-            } else {
-                val transaction = pendingTransactionContainer.getTransaction(txId)
-                if (transaction != null) {
-                    peerTable.advertise(transaction)
+            peerTable.requestAllMessages(request)
+                .map { it.transactionReply }
+                .collect {
+                    if (it.success) {
+                        // tx exists
+                        val info = it.transaction.toModel()
+                        logger.info { "Tx=${txId} found: VBK:${info.blockNumber}:${info.blockHash}"}
+
+                        transactionManager.onTransactionInfo(info)
+                    } else {
+                        // remote peer does not know about this tx.
+                        logger.info { "Tx=${txId} NOT found"}
+
+                        val transaction = transactionManager.getTransaction(txId)
+                        if (transaction != null) {
+                            // rebroadcast it
+                            peerTable.advertise(transaction)
+                        }
+
+                        transactionManager.onMissingTransactionInfo(txId)
+                    }
                 }
-            }
         }
     } catch (e: Exception) {
         logger.debugWarn(e) { "Unable to request pending transactions" }
@@ -61,7 +76,7 @@ private fun VeriBlockMessages.TransactionInfo.toModel() = TransactionInfo(
     bitcoinBlockHash = bitcoinBlockHash.toHex(),
     bitcoinTxId = bitcoinTxId.toHex(),
     bitcoinConfirmations = bitcoinConfirmations,
-    blockHash = blockHash.toHex(),
+    blockHash = blockHash.asAnyVbkHash(),
     merklePath = merklePath
 )