feat: advertise configured account heads during discovery

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<AttoAddress> = emptySet()
 
     @PostConstruct
     fun validate() {

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(

src/main/kotlin/cash/atto/node/transaction/TransactionRepository.kt

@@ -39,6 +39,15 @@ interface TransactionRepository :
     )
     suspend fun getLastSample(limit: Long): Flow<Transaction>
 
+    @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<AttoPublicKey>): Flow<Transaction>
+
     @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,

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<TransactionRepository>()
+            coEvery { transactionRepository.getLastSample(10L) } returns flowOf(random, hinted)
+            coEvery { transactionRepository.findLastByPublicKeys(listOf(hinted.publicKey)) } returns flowOf(hinted)
+
+            val uncheckedRepository = mockk<UncheckedTransactionRepository>()
+            coEvery { uncheckedRepository.count() } returns 0
+
+            val sent = mutableListOf<BroadcastNetworkMessage<*>>()
+            val connectionManager = mockk<NodeConnectionManager>()
+            coEvery { connectionManager.send(any<BroadcastNetworkMessage<*>>()) } answers {
+                sent += firstArg<BroadcastNetworkMessage<*>>()
+            }
+
+            val thisNode = mockk<AttoNode>()
+            every { thisNode.isNotHistorical() } returns false
+
+            val discoverer =
+                LastDiscoverer(
+                    thisNode = thisNode,
+                    discoveryProperties = properties,
+                    accountRepository = mockk<AccountRepository>(),
+                    transactionRepository = transactionRepository,
+                    uncheckedTransactionRepository = uncheckedRepository,
+                    nodeConnectionManager = connectionManager,
+                    networkMessagePublisher = mockk<NetworkMessagePublisher>(relaxed = true),
+                    eventPublisher = mockk<EventPublisher>(relaxed = true),
+                    discoveryQueue = mockk<DiscoveryQueue>(relaxed = true),
+                    voteConverter = mockk<VoteConverter>(relaxed = true),
+                    voteWeighter = mockk<VoteWeighter>(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()
+}

src/test/kotlin/cash/atto/node/bootstrap/discovery/LastDiscovererTest.kt

@@ -153,6 +153,7 @@ class LastDiscovererTest {
         val discoverer =
             LastDiscoverer(
                 thisNode = mockk<AttoNode>(relaxed = true),
+                discoveryProperties = DiscoveryProperties(),
                 accountRepository = accountRepository,
                 transactionRepository = mockk<TransactionRepository>(),
                 uncheckedTransactionRepository = mockk<UncheckedTransactionRepository>(),