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..ed93d2d5 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,21 @@ class LastDiscoverer( return } - val transactions = transactionRepository.getLastSample(10) + val transactions = transactionRepository.getLastSample(RANDOM_SAMPLE_SIZE).toList().toMutableList() + for (address in discoveryProperties.hintedAddresses) { + val account = accountRepository.findById(address.publicKey) + if (account == null || account.algorithm != address.algorithm) { + continue + } + + transactionRepository.findById(account.lastTransactionHash)?.let(transactions::add) + } transactions + .distinctBy { it.hash } .map { AttoBootstrapTransactionPush(it.toAttoTransaction()) } .map { BroadcastNetworkMessage(BroadcastStrategy.EVERYONE, setOf(), it) } - .collect { nodeConnectionManager.send(it) } + .forEach { nodeConnectionManager.send(it) } } } @@ -221,6 +231,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/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..aa9b0b62 --- /dev/null +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererBroadcastTest.kt @@ -0,0 +1,128 @@ +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.Account +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 +import java.time.Instant + +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 accountRepository = mockk() + coEvery { accountRepository.findById(hinted.publicKey) } returns account(hinted) + + val transactionRepository = mockk() + coEvery { transactionRepository.getLastSample(10L) } returns flowOf(random, hinted) + coEvery { transactionRepository.findById(hinted.hash) } returns 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 = accountRepository, + 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 account(transaction: Transaction): Account = + Account( + publicKey = transaction.publicKey, + network = transaction.block.network, + version = 0U.toAttoVersion(), + algorithm = transaction.algorithm, + height = transaction.height.value.toLong(), + balance = transaction.block.balance, + lastTransactionTimestamp = Instant.now(), + lastTransactionHash = transaction.hash, + representativeAlgorithm = transaction.algorithm, + representativePublicKey = transaction.publicKey, + ) + + 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(),