Stabilize bootstrap persistence under MySQL load

README.md

@@ -202,8 +202,12 @@ stores it as an **unchecked transaction**:
   - 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.
 
-- **Persistence batching** (`DiscoveryProcessor`)
-  - Buffers discovered transactions and periodically persists them into the unchecked table in bulk.
+- **Bounded persistence queue** (`DiscoveryQueue`, `DiscoveryPersistenceWorker`)
+  - `DiscoveryQueue` deduplicates and buffers discovered transactions.
+  - New gap, head, and send discovery pauses at the configured target capacity. Replies already in flight can use the
+    configured headroom and suspend without being discarded when the bounded queue is full.
+  - `DiscoveryPersistenceWorker` persists batches of up to 1,000 transactions and retries the same batch after a
+    database failure.
 
 The overall effect: nodes can reconnect, observe the network’s latest heads, request missing ranges, and replay into
 their ledger state.

src/main/kotlin/cash/atto/node/DuplicateDetector.kt

@@ -16,6 +16,10 @@ class DuplicateDetector<T : Any>(
 
     fun isDuplicate(t: T): Boolean = cache.putIfAbsent(t, t) != null
 
+    fun refresh(t: T) {
+        cache[t] = t
+    }
+
     fun remove(t: T) {
         cache.remove(t)
     }

src/main/kotlin/cash/atto/node/NodeConfiguration.kt

@@ -44,7 +44,7 @@ class NodeConfiguration(
 
         return AttoNode(
             network = nodeProperties.network!!,
-            protocolVersion = 0u,
+            protocolVersion = AttoNode.CURRENT_PROTOCOL_VERSION,
             algorithm = AttoAlgorithm.V1,
             publicKey = signer.publicKey,
             publicUri = URI(nodeProperties.publicUri!!),

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

@@ -4,7 +4,6 @@ import cash.atto.commons.AttoAmount
 import cash.atto.commons.AttoHash
 import cash.atto.commons.AttoPublicKey
 import cash.atto.node.CacheSupport
-import cash.atto.node.EventPublisher
 import cash.atto.node.bootstrap.TransactionDiscovered
 import cash.atto.node.transaction.Transaction
 import cash.atto.node.transaction.TransactionRejected
@@ -24,7 +23,7 @@ import java.time.Duration
 @Component
 class DependencyDiscoverer(
     private val voteWeighter: VoteWeighter,
-    private val eventPublisher: EventPublisher,
+    private val discoveryQueue: DiscoveryQueue,
 ) : CacheSupport {
     private val logger = KotlinLogging.logger {}
 
@@ -102,19 +101,20 @@ class DependencyDiscoverer(
             val weight = holder.getWeight()
             val minimalConfirmationWeight = voteWeighter.getMinimalConfirmationWeight()
 
-            if (weight >= minimalConfirmationWeight) {
-                transactionHolderMap.remove(hash)
-                logger.debug { "Discovered approved transaction that's missing some dependency $hash" }
-                eventPublisher.publish(
-                    TransactionDiscovered(
-                        holder.reason,
-                        holder.transaction,
-                        holder.votes.values.toList(),
-                    ),
-                )
+            if (weight < minimalConfirmationWeight) {
+                return
             }
 
-            return
+            logger.debug { "Discovered approved transaction that's missing some dependency $hash" }
+            discoveryQueue.queue(
+                TransactionDiscovered(
+                    holder.reason,
+                    holder.transaction,
+                    holder.votes.values.toList(),
+                ),
+                DiscoverySource.DEPENDENCY,
+            )
+            transactionHolderMap.remove(hash)
         }
 
     override fun clear() {

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

@@ -0,0 +1,118 @@
+package cash.atto.node.bootstrap.discovery
+
+import io.micrometer.core.instrument.Counter
+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.EnumMap
+
+@Component
+class DiscoveryMetrics(
+    private val meterRegistry: MeterRegistry,
+) {
+    private val admittedCounters = EnumMap<DiscoverySource, Counter>(DiscoverySource::class.java)
+    private val persistedCounters = EnumMap<DiscoverySource, Counter>(DiscoverySource::class.java)
+    private val persistenceFailureCounter: Counter
+    private val affectedRowsCounter: Counter
+    private val batchTimer: Timer
+    private val batchSizeSummary: DistributionSummary
+    private val queueWaitTimer: Timer
+
+    init {
+        DiscoverySource.entries.forEach { source ->
+            admittedCounters[source] =
+                Counter
+                    .builder("transactions.discovery.admitted")
+                    .description("Discovered transactions admitted for unchecked persistence")
+                    .tag("source", source.metricTag)
+                    .register(meterRegistry)
+            persistedCounters[source] =
+                Counter
+                    .builder("transactions.discovery.persisted")
+                    .description("Discovered transactions committed to unchecked persistence")
+                    .tag("source", source.metricTag)
+                    .register(meterRegistry)
+        }
+        persistenceFailureCounter =
+            Counter
+                .builder("transactions.discovery.persistence.failures")
+                .description("Unchecked discovery persistence failures")
+                .register(meterRegistry)
+        affectedRowsCounter =
+            Counter
+                .builder("transactions.discovery.affected")
+                .description("Rows affected by unchecked discovery persistence")
+                .register(meterRegistry)
+        batchTimer =
+            Timer
+                .builder("transactions.discovery.batch")
+                .description("Time spent attempting an unchecked discovery persistence batch")
+                .register(meterRegistry)
+        batchSizeSummary =
+            DistributionSummary
+                .builder("transactions.discovery.batch.size")
+                .description("Discovered transactions attempted per persistence batch")
+                .register(meterRegistry)
+        queueWaitTimer =
+            Timer
+                .builder("transactions.discovery.queue.wait")
+                .description("Time a discovered transaction waited before entering a persistence batch")
+                .register(meterRegistry)
+    }
+
+    internal fun bind(queue: DiscoveryQueue) {
+        Gauge
+            .builder("transactions.discovery.backlog.depth", queue) {
+                it.getBacklogDepth().toDouble()
+            }.description("Discovered transactions waiting to enter a persistence batch")
+            .register(meterRegistry)
+        Gauge
+            .builder("transactions.discovery.backlog.overshoot", queue) {
+                it.getBacklogOvershoot().toDouble()
+            }.description("Queued and suspended discoveries above the target capacity")
+            .register(meterRegistry)
+        Gauge
+            .builder("transactions.discovery.at.capacity", queue) {
+                if (it.isAtCapacity()) 1.0 else 0.0
+            }.description("Whether new discovery should pause")
+            .register(meterRegistry)
+    }
+
+    internal fun admitted(source: DiscoverySource) {
+        admittedCounters.getValue(source).increment()
+    }
+
+    internal fun persisted(batch: List<PendingDiscovery>) {
+        batch
+            .groupingBy { it.source }
+            .eachCount()
+            .forEach { (source, count) ->
+                persistedCounters.getValue(source).increment(count.toDouble())
+            }
+    }
+
+    internal fun persistenceFailed() {
+        persistenceFailureCounter.increment()
+    }
+
+    internal fun dequeued(wait: Duration) {
+        queueWaitTimer.record(if (wait.isNegative) Duration.ZERO else wait)
+    }
+
+    internal fun affectedRows(count: Long) {
+        affectedRowsCounter.increment(count.toDouble())
+    }
+
+    internal fun startBatch(size: Int): Timer.Sample {
+        batchSizeSummary.record(size.toDouble())
+        return Timer.start(meterRegistry)
+    }
+
+    internal fun stopBatch(sample: Timer.Sample): Duration =
+        Duration.ofNanos(
+            sample.stop(batchTimer),
+        )
+}

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

@@ -0,0 +1,105 @@
+package cash.atto.node.bootstrap.discovery
+
+import cash.atto.node.CacheSupport
+import cash.atto.node.bootstrap.unchecked.UncheckedTransactionService
+import io.github.oshai.kotlinlogging.KotlinLogging
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Component
+import java.time.Clock
+import java.time.Duration
+import java.time.Instant
+import java.util.concurrent.TimeUnit
+
+@Component
+class DiscoveryPersistenceWorker(
+    private val uncheckedTransactionService: UncheckedTransactionService,
+    private val queue: DiscoveryQueue,
+    private val properties: DiscoveryProperties,
+    private val metrics: DiscoveryMetrics,
+    private val clock: Clock,
+) : CacheSupport {
+    private val logger = KotlinLogging.logger {}
+    private val flushMutex = Mutex()
+
+    private var nextRetryAt = Instant.EPOCH
+    private var retryBackoffInSeconds = properties.retryInitialBackoffInSeconds
+    private var retryBatch: List<PendingDiscovery>? = null
+
+    @Scheduled(fixedRate = 1, timeUnit = TimeUnit.SECONDS)
+    suspend fun flush() {
+        if (!flushMutex.tryLock()) {
+            return
+        }
+
+        try {
+            if (clock.instant().isBefore(nextRetryAt)) {
+                return
+            }
+
+            while (true) {
+                val batch = retryBatch ?: queue.takeBatch()
+                if (batch.isEmpty()) {
+                    return
+                }
+                retryBatch = batch
+
+                val sample = metrics.startBatch(batch.size)
+                val affectedRows =
+                    try {
+                        uncheckedTransactionService.save(batch.map { it.transaction })
+                    } catch (e: CancellationException) {
+                        metrics.stopBatch(sample)
+                        throw e
+                    } catch (e: Exception) {
+                        metrics.stopBatch(sample)
+                        metrics.persistenceFailed()
+                        scheduleRetry()
+                        logger.warn(e) {
+                            "Failed to save ${batch.size} unchecked transactions. Retrying the same batch after " +
+                                "${Duration.between(clock.instant(), nextRetryAt)}"
+                        }
+                        return
+                    }
+
+                val elapsed = metrics.stopBatch(sample)
+                queue.acknowledge(batch)
+                retryBatch = null
+                metrics.persisted(batch)
+                metrics.affectedRows(affectedRows)
+                resetRetry()
+                logger.info { "Saved ${batch.size} unchecked transactions in $elapsed" }
+            }
+        } finally {
+            flushMutex.unlock()
+        }
+    }
+
+    override fun clear() {
+        runBlocking {
+            flushMutex.withLock {
+                retryBatch = null
+                queue.reset()
+                resetRetry()
+            }
+        }
+    }
+
+    private fun scheduleRetry() {
+        nextRetryAt = clock.instant().plusSeconds(retryBackoffInSeconds)
+        retryBackoffInSeconds =
+            if (retryBackoffInSeconds >= properties.retryMaxBackoffInSeconds / 2) {
+                properties.retryMaxBackoffInSeconds
+            } else {
+                minOf(retryBackoffInSeconds * 2, properties.retryMaxBackoffInSeconds)
+            }
+    }
+
+    private fun resetRetry() {
+        nextRetryAt = Instant.EPOCH
+        retryBackoffInSeconds = properties.retryInitialBackoffInSeconds
+    }
+}