Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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) }
}
}

Expand Down Expand Up @@ -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(
Expand Down
Original file line number Diff line number Diff line change
@@ -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<AccountRepository>()
coEvery { accountRepository.findById(hinted.publicKey) } returns account(hinted)

val transactionRepository = mockk<TransactionRepository>()
coEvery { transactionRepository.getLastSample(10L) } returns flowOf(random, hinted)
coEvery { transactionRepository.findById(hinted.hash) } returns 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 = 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 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()
}
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,7 @@ class LastDiscovererTest {
val discoverer =
LastDiscoverer(
thisNode = mockk<AttoNode>(relaxed = true),
discoveryProperties = DiscoveryProperties(),
accountRepository = accountRepository,
transactionRepository = mockk<TransactionRepository>(),
uncheckedTransactionRepository = mockk<UncheckedTransactionRepository>(),
Expand Down