Refactor node component ownership and recovery

README.md

@@ -160,6 +160,9 @@ All streaming endpoints use **NDJSON** (`application/x-ndjson`) and combine:
 Address-specific account streams subscribe to live updates before reading account snapshots. Live flow emissions
 wait for that subscriber to finish the snapshot, then continue through the same subscription.
 Each collection forwards only increasing heights for each requested address.
+Transaction and account-entry hash/range streams use the same subscription-before-query handoff. Hash streams
+complete after one result; range streams order and deduplicate results by height before completing the requested range.
+Slow snapshot reads or clients backpressure the shared live feed; disconnecting cancels that subscriber and its query.
 
 Receivable streams subscribe to live updates before reading the database. They scan the database before draining
 a bounded live-update queue and continuing on the same subscription. Only hashes from the initial
@@ -224,8 +227,9 @@ stores it as an **unchecked transaction**:
     `AttoTransactionStreamRequest/Response`.
   - Handles out-of-order deliveries with an in-memory buffer.
 
-- **Head/last discovery** (`LastDiscoverer`)
-  - Historical nodes periodically broadcast a sample of recent “head” transactions (`AttoBootstrapTransactionPush`).
+- **Head/last discovery** (`BootstrapHeadBroadcaster`, `LastDiscoverer`)
+  - `BootstrapHeadBroadcaster` periodically broadcasts a sample of recent “head” transactions
+    (`AttoBootstrapTransactionPush`) from historical nodes with no unchecked backlog.
   - If a receiving node is behind, it starts a lightweight vote stream for the head to confirm it’s real, then turns it
     into a `TransactionDiscovered` event once the votes reach quorum.
   - Partial vote collections expire without starting a live election. A later head advertisement can retry discovery.

build.gradle.kts

@@ -139,6 +139,10 @@ tasks.withType<Test> {
     maxHeapSize = "4g"
 }
 
+tasks.named("check") {
+    dependsOn("compileBenchmarkKotlin")
+}
+
 graalvmNative {
     binaries {
         named("main") {

src/benchmark/kotlin/cash/atto/node/election/ElectionVoterBenchmark.kt

@@ -18,11 +18,12 @@ import cash.atto.node.account.Account
 import cash.atto.node.account.AccountRepository
 import cash.atto.node.account.AccountUpdated
 import cash.atto.node.network.NetworkMessagePublisher
+import cash.atto.node.network.guardian.ExpectedResponses
 import cash.atto.node.transaction.Transaction
 import cash.atto.node.transaction.TransactionSource
-import cash.atto.node.vote.VoteRepository
 import cash.atto.node.vote.weight.VoteWeightProperties
 import cash.atto.node.vote.weight.VoteWeighter
+import cash.atto.node.vote.weight.Weight
 import cash.atto.node.vote.weight.WeightRepository
 import cash.atto.node.vote.weight.WeightService
 import cash.atto.protocol.AttoNode
@@ -35,15 +36,20 @@ import kotlinx.benchmark.TearDown
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.runBlocking
 import org.openjdk.jmh.annotations.Level
 import org.openjdk.jmh.annotations.Threads
 import org.springframework.context.ApplicationEvent
 import org.springframework.context.ApplicationEventPublisher
+import org.springframework.transaction.ReactiveTransaction
+import org.springframework.transaction.ReactiveTransactionManager
+import org.springframework.transaction.TransactionDefinition
+import reactor.core.publisher.Mono
 import java.lang.reflect.Proxy
 import java.net.URI
+import java.time.Clock
 import java.time.Instant
-import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.time.Duration.Companion.milliseconds
 
@@ -86,12 +92,13 @@ class ElectionVoterBenchmark {
                 signer = signer,
                 voteWeighter = voteWeighter,
                 eventPublisher = EventPublisher(springPublisher),
-                messagePublisher = NetworkMessagePublisher(springPublisher),
+                messagePublisher = NetworkMessagePublisher(springPublisher, ExpectedResponses(Clock.systemUTC())),
                 accountRepository = accountRepository,
             )
         account = createAccount(node)
     }
 
+    // Measures event handling and scheduling; signing completes asynchronously.
     @Benchmark
     @Threads(100)
     fun processElectionFlow() =
@@ -213,47 +220,36 @@ private fun createVoteWeighter(node: AttoNode): VoteWeighter {
             minimalRebroadcastWeight = "1"
             samplePeriodInDays = 1
         }
-    val voteWeighter =
-        VoteWeighter(
-            thisNode = node,
-            properties = properties,
-            weightService = WeightService(interfaceProxy<WeightRepository>()),
-            voteRepository = interfaceProxy<VoteRepository>(),
-            genesisTransaction = createTransaction(node, 1U),
-        )
-
-    val weightMapField = VoteWeighter::class.java.getDeclaredField("weightMap")
-    weightMapField.isAccessible = true
-    @Suppress("UNCHECKED_CAST")
-    val weightMap = weightMapField.get(voteWeighter) as ConcurrentHashMap<AttoPublicKey, AttoAmount>
-    weightMap[node.publicKey] = AttoAmount.MAX
-
-    return voteWeighter
+    val weight = Weight(node.publicKey, AttoAmount.MAX, Instant.now())
+    return VoteWeighter(
+        thisNode = node,
+        properties = properties,
+        weightService = WeightService(weightRepository(weight), BenchmarkTransactionManager()),
+        genesisTransaction = createTransaction(node, 1U),
+    ).also { it.init() }
 }
 
-private inline fun <reified T> interfaceProxy(): T =
+private fun weightRepository(weight: Weight): WeightRepository =
     Proxy
         .newProxyInstance(
-            T::class.java.classLoader,
-            arrayOf(T::class.java),
+            WeightRepository::class.java.classLoader,
+            arrayOf(WeightRepository::class.java),
         ) { proxy, method, args ->
             when (method.name) {
-                "toString" -> "${T::class.java.simpleName}Proxy"
+                "findCalculatedWeights" -> flowOf(weight)
+                "upsert", "deleteAllExcept" -> Unit
+                "toString" -> "BenchmarkWeightRepository"
                 "hashCode" -> System.identityHashCode(proxy)
                 "equals" -> proxy === args?.singleOrNull()
-                else -> defaultValue(method.returnType)
+                else -> error("Unexpected benchmark repository call: ${method.name}")
             }
-        } as T
+        } as WeightRepository
 
-private fun defaultValue(returnType: Class<*>): Any? =
-    when (returnType) {
-        java.lang.Boolean.TYPE -> false
-        java.lang.Byte.TYPE -> 0.toByte()
-        java.lang.Short.TYPE -> 0.toShort()
-        java.lang.Integer.TYPE -> 0
-        java.lang.Long.TYPE -> 0L
-        java.lang.Float.TYPE -> 0f
-        java.lang.Double.TYPE -> 0.0
-        java.lang.Character.TYPE -> 0.toChar()
-        else -> null
-    }
+private class BenchmarkTransactionManager : ReactiveTransactionManager {
+    override fun getReactiveTransaction(definition: TransactionDefinition?): Mono<ReactiveTransaction> =
+        Mono.just(object : ReactiveTransaction {})
+
+    override fun commit(transaction: ReactiveTransaction): Mono<Void> = Mono.empty()
+
+    override fun rollback(transaction: ReactiveTransaction): Mono<Void> = error("Unexpected benchmark initialization rollback")
+}

src/main/kotlin/cash/atto/node/account/entry/AccountEntryController.kt

@@ -19,12 +19,13 @@ import io.swagger.v3.oas.annotations.responses.ApiResponse
 import io.swagger.v3.oas.annotations.tags.Tag
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onCompletion
 import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.flow.onSubscription
 import kotlinx.coroutines.flow.take
 import org.springframework.context.event.EventListener
 import org.springframework.http.HttpStatus
@@ -94,24 +95,14 @@ class AccountEntryController(
     )
     suspend fun stream(
         @PathVariable hash: AttoHash,
-    ): Flow<AttoAccountEntry> {
-        val entryDatabaseFlow: Flow<AttoAccountEntry> =
-            flow {
-                val transaction = repository.findById(hash)
-                if (transaction != null) {
-                    emit(transaction.toAtto())
-                }
-            }
-
-        val entryFlow =
-            entryFlow
-                .filter { it.hash == hash }
-
-        return merge(entryFlow, entryDatabaseFlow)
+    ): Flow<AttoAccountEntry> =
+        entryFlow
+            .onSubscription {
+                repository.findById(hash)?.let { emit(it.toAtto()) }
+            }.filter { it.hash == hash }
             .onStart { logger.trace { "Started streaming $hash account entry" } }
             .onCompletion { logger.trace { "Stopped streaming $hash account entry" } }
             .take(1)
-    }
 
     @PostMapping("/accounts/entries/stream", produces = [MediaType.APPLICATION_NDJSON_VALUE])
     @Operation(
@@ -149,17 +140,14 @@ class AccountEntryController(
                         if (it.value > Int.MAX_VALUE.toUInt()) Int.MAX_VALUE else it.value.toInt()
                     }
 
-                val dbFlow =
-                    repository
-                        .findAsc(publicKey, range.fromHeight.value.toBigInteger(), range.toHeight.value.toBigInteger())
-                        .map { it.toAtto() }
-
-                val liveFlow =
-                    entryFlow
-                        .filter { it.publicKey == publicKey }
-                        .filter { it.height in fromHeight..toHeight }
-
-                merge(liveFlow, dbFlow)
+                entryFlow
+                    .onSubscription {
+                        emitAll(
+                            repository
+                                .findAsc(publicKey, fromHeight.value.toBigInteger(), toHeight.value.toBigInteger())
+                                .map { it.toAtto() },
+                        )
+                    }.filter { it.publicKey == publicKey && it.height in fromHeight..toHeight }
                     .sortByHeight(fromHeight)
                     .take(expectedCount)
             }

src/main/kotlin/cash/atto/node/bootstrap/BootstrapLoadMonitor.kt

@@ -1,38 +1,29 @@
 package cash.atto.node.bootstrap
 
-import cash.atto.node.bootstrap.discovery.DiscoveryMetrics
 import cash.atto.node.bootstrap.discovery.DiscoveryProperties
-import cash.atto.node.bootstrap.unchecked.UncheckedTransactionInserter
 import io.micrometer.core.instrument.DistributionSummary
 import io.micrometer.core.instrument.Gauge
 import io.micrometer.core.instrument.MeterRegistry
-import io.micrometer.core.instrument.Timer
 import org.springframework.stereotype.Component
 import java.time.Duration
-import java.util.concurrent.TimeUnit
-import kotlin.math.roundToLong
 
 @Component
 class BootstrapLoadMonitor(
     private val properties: DiscoveryProperties,
-    private val meterRegistry: MeterRegistry,
+    meterRegistry: MeterRegistry,
 ) {
     private val targetRatioSummary =
         DistributionSummary
             .builder(TARGET_RATIO_METRIC_NAME)
             .description("Unchecked persistence elapsed time divided by its proportional target")
             .register(meterRegistry)
 
-    private val insertTimer: Timer by lazy {
-        meterRegistry.get(UncheckedTransactionInserter.METRIC_NAME).timer()
-    }
-
     @Volatile
     private var targetCapacity = properties.batchSize
-    private var observedInsertCount = 0L
-    private var observedInsertNanos = 0L
-    private var observedPersistedRows = 0L
-    private var observedFailureCount = 0L
+    private var pendingInsertCount = 0L
+    private var pendingInsertNanos = 0L
+    private var pendingPersistedRows = 0L
+    private var pendingFailureCount = 0L
 
     init {
         Gauge
@@ -47,31 +38,31 @@ class BootstrapLoadMonitor(
     internal fun targetCapacity(): Int = targetCapacity
 
     @Synchronized
-    internal fun poll() {
-        val insertCount = insertTimer.count()
-        val insertNanos = insertTimer.totalTime(TimeUnit.NANOSECONDS).roundToLong()
-        val persistedRows =
-            meterRegistry
-                .find(DiscoveryMetrics.PERSISTED_METRIC_NAME)
-                .counters()
-                .sumOf { it.count() }
-                .roundToLong()
-        val failureCount =
-            meterRegistry
-                .get(DiscoveryMetrics.PERSISTENCE_FAILURE_METRIC_NAME)
-                .counter()
-                .count()
-                .roundToLong()
+    internal fun recordInsert(elapsedNanos: Long) {
+        pendingInsertCount++
+        pendingInsertNanos += elapsedNanos
+    }
 
-        val completedInsertDelta = insertCount - observedInsertCount
-        val elapsedNanosDelta = maxOf(0L, insertNanos - observedInsertNanos)
-        val persistedRowsDelta = persistedRows - observedPersistedRows
-        val failureCountDelta = failureCount - observedFailureCount
+    @Synchronized
+    internal fun recordPersisted(rowCount: Int) {
+        pendingPersistedRows += rowCount
+    }
 
-        observedInsertCount = insertCount
-        observedInsertNanos = insertNanos
-        observedPersistedRows = persistedRows
-        observedFailureCount = failureCount
+    @Synchronized
+    internal fun recordFailure() {
+        pendingFailureCount++
+    }
+
+    @Synchronized
+    internal fun poll() {
+        val completedInsertDelta = pendingInsertCount
+        val elapsedNanosDelta = maxOf(0L, pendingInsertNanos)
+        val persistedRowsDelta = pendingPersistedRows
+        val failureCountDelta = pendingFailureCount
+        pendingInsertCount = 0
+        pendingInsertNanos = 0
+        pendingPersistedRows = 0
+        pendingFailureCount = 0
 
         if (failureCountDelta > 0) {
             targetCapacity = properties.batchSize
@@ -81,7 +72,7 @@ class BootstrapLoadMonitor(
             return
         }
         check(completedInsertDelta > 0) {
-            "Persisted $persistedRowsDelta unchecked transactions without an insert timer measurement"
+            "Persisted $persistedRowsDelta unchecked transactions without an insert measurement"
         }
 
         val elapsed = Duration.ofNanos(elapsedNanosDelta)

src/main/kotlin/cash/atto/node/bootstrap/discovery/BootstrapHeadBroadcaster.kt

@@ -0,0 +1,44 @@
+package cash.atto.node.bootstrap.discovery
+
+import cash.atto.node.bootstrap.unchecked.UncheckedTransactionRepository
+import cash.atto.node.network.BroadcastNetworkMessage
+import cash.atto.node.network.BroadcastStrategy
+import cash.atto.node.network.NodeConnectionManager
+import cash.atto.node.transaction.TransactionRepository
+import cash.atto.protocol.AttoBootstrapTransactionPush
+import cash.atto.protocol.AttoNode
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import java.util.concurrent.TimeUnit
+
+@Component
+class BootstrapHeadBroadcaster(
+    private val thisNode: AttoNode,
+    private val transactionRepository: TransactionRepository,
+    private val uncheckedTransactionRepository: UncheckedTransactionRepository,
+    private val nodeConnectionManager: NodeConnectionManager,
+) {
+    private val mutex = Mutex()
+
+    @Scheduled(fixedRate = 1, timeUnit = TimeUnit.MINUTES)
+    suspend fun broadcastSample() {
+        if (thisNode.isNotHistorical()) {
+            return
+        }
+
+        mutex.withLock {
+            if (uncheckedTransactionRepository.count() > 0) {
+                return
+            }
+
+            transactionRepository
+                .getLastSample(10)
+                .map { AttoBootstrapTransactionPush(it.toAttoTransaction()) }
+                .map { BroadcastNetworkMessage(BroadcastStrategy.EVERYONE, setOf(), it) }
+                .collect { nodeConnectionManager.send(it) }
+        }
+    }
+}