diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryProperties.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryProperties.kt index c7f9f7d7..6bbd3ce9 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryProperties.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryProperties.kt @@ -1,5 +1,6 @@ package cash.atto.node.bootstrap.discovery +import cash.atto.commons.AttoAddress import jakarta.annotation.PostConstruct import kotlinx.coroutines.channels.Channel import org.springframework.boot.context.properties.ConfigurationProperties @@ -12,6 +13,7 @@ class DiscoveryProperties { var headroom: Int = 2_000 var batchSize: Int = 1_000 var persistenceTargetPerSecond: Long = 1_000 + var hintedAddresses: Set = emptySet() @PostConstruct fun validate() { diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/LastDiscoverer.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/LastDiscoverer.kt index eee22c60..d6e87f22 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/discovery/LastDiscoverer.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/LastDiscoverer.kt @@ -34,7 +34,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.cancel -import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -48,6 +48,7 @@ import java.util.concurrent.TimeUnit @Component class LastDiscoverer( private val thisNode: AttoNode, + private val discoveryProperties: DiscoveryProperties, private val accountRepository: AccountRepository, private val transactionRepository: TransactionRepository, private val uncheckedTransactionRepository: UncheckedTransactionRepository, @@ -102,12 +103,20 @@ class LastDiscoverer( return } - val transactions = transactionRepository.getLastSample(10) + val hintedPublicKeys = discoveryProperties.hintedAddresses.map { it.publicKey } + val hintedTransactions = + if (hintedPublicKeys.isEmpty()) { + emptyList() + } else { + transactionRepository.findLastByPublicKeys(hintedPublicKeys).toList() + } + val transactions = transactionRepository.getLastSample(RANDOM_SAMPLE_SIZE).toList() + hintedTransactions transactions + .distinctBy { it.hash } .map { AttoBootstrapTransactionPush(it.toAttoTransaction()) } .map { BroadcastNetworkMessage(BroadcastStrategy.EVERYONE, setOf(), it) } - .collect { nodeConnectionManager.send(it) } + .forEach { nodeConnectionManager.send(it) } } } @@ -221,6 +230,10 @@ class LastDiscoverer( private fun startElection(transaction: Transaction) { eventPublisher.publish(TransactionReceived(transaction)) } + + private companion object { + const val RANDOM_SAMPLE_SIZE = 10L + } } private data class PendingHeadAdmission( diff --git a/src/main/kotlin/cash/atto/node/transaction/TransactionRepository.kt b/src/main/kotlin/cash/atto/node/transaction/TransactionRepository.kt index 84d29c46..5669cf09 100644 --- a/src/main/kotlin/cash/atto/node/transaction/TransactionRepository.kt +++ b/src/main/kotlin/cash/atto/node/transaction/TransactionRepository.kt @@ -39,6 +39,15 @@ interface TransactionRepository : ) suspend fun getLastSample(limit: Long): Flow + @Query( + """ + SELECT t.* FROM transaction t + JOIN account a ON t.hash = a.last_transaction_hash + WHERE a.public_key IN (:publicKeys) + """, + ) + suspend fun findLastByPublicKeys(publicKeys: Collection): Flow + @Query("SELECT * FROM transaction t WHERE t.public_key = :publicKey AND t.height BETWEEN :fromHeight and :toHeight ORDER BY height ASC") suspend fun findAsc( publicKey: AttoPublicKey, diff --git a/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererBroadcastTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererBroadcastTest.kt new file mode 100644 index 00000000..322a7d37 --- /dev/null +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererBroadcastTest.kt @@ -0,0 +1,109 @@ +package cash.atto.node.bootstrap.discovery + +import cash.atto.commons.AttoAddress +import cash.atto.commons.AttoAlgorithm +import cash.atto.commons.AttoAmount +import cash.atto.commons.AttoHash +import cash.atto.commons.AttoInstant +import cash.atto.commons.AttoNetwork +import cash.atto.commons.AttoPublicKey +import cash.atto.commons.AttoReceiveBlock +import cash.atto.commons.AttoSignature +import cash.atto.commons.AttoTransaction +import cash.atto.commons.AttoWork +import cash.atto.commons.toAttoHeight +import cash.atto.commons.toAttoVersion +import cash.atto.node.EventPublisher +import cash.atto.node.account.AccountRepository +import cash.atto.node.bootstrap.unchecked.UncheckedTransactionRepository +import cash.atto.node.network.BroadcastNetworkMessage +import cash.atto.node.network.NetworkMessagePublisher +import cash.atto.node.network.NodeConnectionManager +import cash.atto.node.transaction.Transaction +import cash.atto.node.transaction.TransactionRepository +import cash.atto.node.transaction.toTransaction +import cash.atto.node.vote.convertion.VoteConverter +import cash.atto.node.vote.weight.VoteWeighter +import cash.atto.protocol.AttoBootstrapTransactionPush +import cash.atto.protocol.AttoNode +import io.mockk.coEvery +import io.mockk.every +import io.mockk.firstArg +import io.mockk.mockk +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class LastDiscovererBroadcastTest { + @Test + fun `configured account head is included once alongside random sample`() = + runTest { + val random = transaction(1) + val hinted = transaction(2) + val hintedAddress = AttoAddress(hinted.algorithm, hinted.publicKey) + val properties = DiscoveryProperties().apply { hintedAddresses = setOf(hintedAddress) } + + val transactionRepository = mockk() + coEvery { transactionRepository.getLastSample(10L) } returns flowOf(random, hinted) + coEvery { transactionRepository.findLastByPublicKeys(listOf(hinted.publicKey)) } returns flowOf(hinted) + + val uncheckedRepository = mockk() + coEvery { uncheckedRepository.count() } returns 0 + + val sent = mutableListOf>() + val connectionManager = mockk() + coEvery { connectionManager.send(any>()) } answers { + sent += firstArg>() + } + + val thisNode = mockk() + every { thisNode.isNotHistorical() } returns false + + val discoverer = + LastDiscoverer( + thisNode = thisNode, + discoveryProperties = properties, + accountRepository = mockk(), + transactionRepository = transactionRepository, + uncheckedTransactionRepository = uncheckedRepository, + nodeConnectionManager = connectionManager, + networkMessagePublisher = mockk(relaxed = true), + eventPublisher = mockk(relaxed = true), + discoveryQueue = mockk(relaxed = true), + voteConverter = mockk(relaxed = true), + voteWeighter = mockk(relaxed = true), + ) + + try { + discoverer.broadcastSample() + + val hashes = + sent.map { + (it.payload as AttoBootstrapTransactionPush).transaction.hash + } + assertEquals(listOf(random.hash, hinted.hash), hashes) + } finally { + discoverer.close() + } + } + + private fun transaction(marker: Byte): Transaction = + AttoTransaction( + block = + AttoReceiveBlock( + version = 0U.toAttoVersion(), + network = AttoNetwork.LOCAL, + algorithm = AttoAlgorithm.V1, + publicKey = AttoPublicKey(ByteArray(32) { marker }), + height = 2U.toAttoHeight(), + balance = AttoAmount.MAX, + timestamp = AttoInstant.now(), + previous = AttoHash(ByteArray(32) { (marker + 1).toByte() }), + sendHashAlgorithm = AttoAlgorithm.V1, + sendHash = AttoHash(ByteArray(32) { (marker + 2).toByte() }), + ), + signature = AttoSignature(ByteArray(64) { (marker + 3).toByte() }), + work = AttoWork(ByteArray(8) { (marker + 4).toByte() }), + ).toTransaction() +} diff --git a/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererTest.kt index 5514496b..6b5f2287 100644 --- a/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererTest.kt +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererTest.kt @@ -153,6 +153,7 @@ class LastDiscovererTest { val discoverer = LastDiscoverer( thisNode = mockk(relaxed = true), + discoveryProperties = DiscoveryProperties(), accountRepository = accountRepository, transactionRepository = mockk(), uncheckedTransactionRepository = mockk(),