From 790a07ee7be7bc530a3a518b47bb28948ecb893d Mon Sep 17 00:00:00 2001 From: rotilho Date: Wed, 5 Aug 2026 21:10:46 +0200 Subject: [PATCH] Adapt bootstrap work to disk pressure Throttle unchecked maintenance, persistence, and gap discovery with Linux I/O PSI while preserving the configured physical queue and lossless admission. Centralize bootstrap priorities, keep exact persistence retries, and merge unchecked orchestration into the processor with explicit transaction scope. Cover PSI spikes and recovery, adaptive capacity, forced drains, retry ordering, scheduler overlap, cleanup progress, and transaction boundaries. --- build.gradle.kts | 2 +- .../node/bootstrap/BootstrapController.kt | 225 ++++++++++++ .../bootstrap/discovery/DiscoveryMetrics.kt | 14 +- .../discovery/DiscoveryPersistenceWorker.kt | 103 ++++-- .../discovery/DiscoveryPressureMonitor.kt | 179 +++++++++ .../bootstrap/discovery/DiscoveryQueue.kt | 49 ++- .../node/bootstrap/discovery/GapDiscoverer.kt | 31 +- .../UncheckedTransactionProcessor.kt | 224 ++++++++++-- .../UncheckedTransactionProcessorStarter.kt | 161 -------- .../unchecked/UncheckedTransactionService.kt | 11 +- .../unchecked/UncheckedWorkTracker.kt | 8 - .../node/bootstrap/BootstrapControllerTest.kt | 283 ++++++++++++++ .../discovery/DiscoveryPressureMonitorTest.kt | 184 ++++++++++ .../bootstrap/discovery/DiscoveryQueueTest.kt | 257 ++++++++++++- .../discovery/DiscoveryTestController.kt | 29 +- .../bootstrap/discovery/GapDiscovererTest.kt | 53 +-- ...ncheckedTransactionProcessorStarterTest.kt | 113 ------ .../UncheckedTransactionProcessorTest.kt | 346 ++++++++++++++++++ .../UncheckedTransactionServiceTest.kt | 4 +- .../UncheckedTransactionController.kt | 6 +- 20 files changed, 1835 insertions(+), 447 deletions(-) create mode 100644 src/main/kotlin/cash/atto/node/bootstrap/BootstrapController.kt create mode 100644 src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitor.kt delete mode 100644 src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarter.kt create mode 100644 src/test/kotlin/cash/atto/node/bootstrap/BootstrapControllerTest.kt create mode 100644 src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitorTest.kt delete mode 100644 src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarterTest.kt create mode 100644 src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorTest.kt diff --git a/build.gradle.kts b/build.gradle.kts index 1cc41c71..e9861cdd 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -144,7 +144,7 @@ graalvmNative { binaries { named("main") { buildArgs.add("-march=compatibility") - buildArgs.add("-R:MinHeapSize=400m") + buildArgs.add("-R:MinHeapSize=200m") } } } diff --git a/src/main/kotlin/cash/atto/node/bootstrap/BootstrapController.kt b/src/main/kotlin/cash/atto/node/bootstrap/BootstrapController.kt new file mode 100644 index 00000000..2f496eed --- /dev/null +++ b/src/main/kotlin/cash/atto/node/bootstrap/BootstrapController.kt @@ -0,0 +1,225 @@ +package cash.atto.node.bootstrap + +import cash.atto.node.bootstrap.discovery.DiscoveryPersistenceResult +import cash.atto.node.bootstrap.discovery.DiscoveryPersistenceWorker +import cash.atto.node.bootstrap.discovery.DiscoveryPressureMonitor +import cash.atto.node.bootstrap.discovery.DiscoveryQueue +import cash.atto.node.bootstrap.discovery.GapDiscoverer +import cash.atto.node.bootstrap.discovery.GapDiscoveryResult +import cash.atto.node.bootstrap.unchecked.UncheckedProcessingResult +import cash.atto.node.bootstrap.unchecked.UncheckedTransactionProcessor +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Gauge +import io.micrometer.core.instrument.MeterRegistry +import kotlinx.coroutines.sync.Mutex +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Component +import java.util.concurrent.TimeUnit + +@Component +class BootstrapController( + private val pressureMonitor: DiscoveryPressureMonitor, + private val discoveryQueue: DiscoveryQueue, + private val persistenceWorker: DiscoveryPersistenceWorker, + private val uncheckedTransactionProcessor: UncheckedTransactionProcessor, + private val gapDiscoverer: GapDiscoverer, + meterRegistry: MeterRegistry, +) { + private val runMutex = Mutex() + + @Volatile + private var diskCredit = 0.0 + + private val decisionCounters = + BootstrapDecision.entries.associateWith { decision -> + Counter + .builder("transactions.bootstrap.controller.decisions") + .description("Bootstrap controller decisions") + .tag("decision", decision.tag) + .register(meterRegistry) + } + + init { + Gauge + .builder("transactions.bootstrap.controller.disk.credit", this) { + it.diskCredit + }.description("Fractional PSI-derived credit available for bootstrap disk work") + .register(meterRegistry) + } + + // A suspending fixed-delay task blocks Spring's shared fixed-delay scheduler until it completes. + @Scheduled(fixedRate = 1, timeUnit = TimeUnit.SECONDS) + suspend fun runOnce() { + if (!runMutex.tryLock()) { + return + } + + try { + accrueDiskCredit() + + if (persistenceWorker.isRetryWaiting()) { + finishWithoutAction(BootstrapDecision.RETRY_WAIT) + return + } + + if (discoveryQueue.isPhysicalBufferFull()) { + persist(forced = true) + return + } + + if (diskCredit < REQUIRED_DISK_CREDIT) { + finishWithoutAction(BootstrapDecision.PRESSURE_WAIT) + return + } + + if (persistenceWorker.hasRetryBatch()) { + persist(forced = false) + return + } + + if (processUnchecked()) { + return + } + + if (persist(forced = false)) { + return + } + + discoverGaps() + } finally { + runMutex.unlock() + } + } + + private fun accrueDiskCredit() { + val availableShare = pressureMonitor.availableShare() + diskCredit = + if (availableShare == 0.0) { + 0.0 + } else { + diskCredit + availableShare + } + } + + private suspend fun processUnchecked(): Boolean { + val result = + try { + uncheckedTransactionProcessor.processIfDue() + } catch (exception: Exception) { + consumeDiskCredit() + record(BootstrapDecision.MAINTENANCE) + throw exception + } + + return when (result) { + UncheckedProcessingResult.SkippedIdle -> false + UncheckedProcessingResult.SkippedBusy -> { + finishWithoutAction(BootstrapDecision.IDLE) + true + } + + is UncheckedProcessingResult.Completed -> { + consumeDiskCredit() + record(BootstrapDecision.MAINTENANCE) + true + } + } + } + + private suspend fun persist(forced: Boolean): Boolean { + val result = + try { + persistenceWorker.persistIfReady() + } catch (exception: Exception) { + finishPersistence(forced) + throw exception + } + + when (result) { + DiscoveryPersistenceResult.Idle -> { + if (forced) { + finishWithoutAction(BootstrapDecision.IDLE) + } + return forced + } + + DiscoveryPersistenceResult.Busy -> { + finishWithoutAction(BootstrapDecision.IDLE) + } + + DiscoveryPersistenceResult.RetryWaiting -> { + finishWithoutAction(BootstrapDecision.RETRY_WAIT) + } + + DiscoveryPersistenceResult.Failed, + DiscoveryPersistenceResult.Persisted, + -> { + finishPersistence(forced) + } + } + return true + } + + private suspend fun discoverGaps() { + val result = + try { + gapDiscoverer.discoverIfDue() + } catch (exception: Exception) { + consumeDiskCredit() + record(BootstrapDecision.GAP) + throw exception + } + + when (result) { + GapDiscoveryResult.Idle, + GapDiscoveryResult.Busy, + -> { + finishWithoutAction(BootstrapDecision.IDLE) + } + + GapDiscoveryResult.Queried -> { + consumeDiskCredit() + record(BootstrapDecision.GAP) + } + } + } + + private fun finishPersistence(forced: Boolean) { + if (forced) { + diskCredit = 0.0 + record(BootstrapDecision.FORCED_DRAIN) + } else { + consumeDiskCredit() + record(BootstrapDecision.PERSISTENCE) + } + } + + private fun finishWithoutAction(decision: BootstrapDecision) { + diskCredit = minOf(REQUIRED_DISK_CREDIT, diskCredit) + record(decision) + } + + private fun consumeDiskCredit() { + diskCredit = maxOf(0.0, diskCredit - REQUIRED_DISK_CREDIT) + } + + private fun record(decision: BootstrapDecision) { + decisionCounters.getValue(decision).increment() + } + + private companion object { + const val REQUIRED_DISK_CREDIT = 1.0 + } +} + +private enum class BootstrapDecision( + val tag: String, +) { + MAINTENANCE("maintenance"), + PERSISTENCE("persistence"), + GAP("gap"), + FORCED_DRAIN("forced-drain"), + PRESSURE_WAIT("pressure-wait"), + RETRY_WAIT("retry-wait"), + IDLE("idle"), +} diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryMetrics.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryMetrics.kt index dbb10647..af604248 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryMetrics.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryMetrics.kt @@ -67,12 +67,22 @@ class DiscoveryMetrics( Gauge .builder("transactions.discovery.backlog.depth", queue) { it.getBacklogDepth().toDouble() - }.description("Discovered transactions waiting to enter a persistence batch") + }.description("Discovered transactions not yet committed to unchecked persistence") + .register(meterRegistry) + Gauge + .builder("transactions.discovery.in.flight", queue) { + it.getInFlightCount().toDouble() + }.description("Discovered transactions in the active or retrying persistence batch") + .register(meterRegistry) + Gauge + .builder("transactions.discovery.capacity.target", queue) { + it.getTargetCapacity().toDouble() + }.description("Current disk-pressure-adjusted discovery admission target") .register(meterRegistry) Gauge .builder("transactions.discovery.backlog.overshoot", queue) { it.getBacklogOvershoot().toDouble() - }.description("Queued and suspended discoveries above the target capacity") + }.description("Outstanding discoveries above the target capacity") .register(meterRegistry) Gauge .builder("transactions.discovery.at.capacity", queue) { diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPersistenceWorker.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPersistenceWorker.kt index 7ac5380c..ab45abcc 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPersistenceWorker.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPersistenceWorker.kt @@ -2,17 +2,16 @@ package cash.atto.node.bootstrap.discovery import cash.atto.node.CacheSupport import cash.atto.node.bootstrap.unchecked.UncheckedTransactionService +import cash.atto.node.bootstrap.unchecked.UncheckedWorkTracker 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( @@ -21,58 +20,39 @@ class DiscoveryPersistenceWorker( private val properties: DiscoveryProperties, private val metrics: DiscoveryMetrics, private val clock: Clock, + private val workTracker: UncheckedWorkTracker, ) : CacheSupport { private val logger = KotlinLogging.logger {} private val flushMutex = Mutex() + @Volatile private var nextRetryAt = Instant.EPOCH + private var retryBackoffInSeconds = properties.retryInitialBackoffInSeconds + + @Volatile private var retryBatch: List? = null - @Scheduled(fixedRate = 1, timeUnit = TimeUnit.SECONDS) - suspend fun flush() { + internal fun isRetryWaiting(): Boolean = retryBatch != null && clock.instant().isBefore(nextRetryAt) + + internal fun hasRetryBatch(): Boolean = retryBatch != null + + suspend fun persistIfReady(): DiscoveryPersistenceResult { if (!flushMutex.tryLock()) { - return + return DiscoveryPersistenceResult.Busy } try { if (clock.instant().isBefore(nextRetryAt)) { - return + return DiscoveryPersistenceResult.RetryWaiting } - 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" } + val batch = retryBatch ?: queue.takeBatch() + if (batch.isEmpty()) { + return DiscoveryPersistenceResult.Idle } + retryBatch = batch + return persist(batch) } finally { flushMutex.unlock() } @@ -81,13 +61,46 @@ class DiscoveryPersistenceWorker( override fun clear() { runBlocking { flushMutex.withLock { + val discoveries = retryBatch.orEmpty() retryBatch = null - queue.reset() + queue.reset(discoveries) resetRetry() } } } + private suspend fun persist(discoveries: List): DiscoveryPersistenceResult { + val sample = metrics.startBatch(discoveries.size) + val affectedRows = + try { + uncheckedTransactionService.save(discoveries.map { it.transaction }) + } catch (exception: CancellationException) { + metrics.stopBatch(sample) + throw exception + } catch (exception: Exception) { + metrics.stopBatch(sample) + metrics.persistenceFailed() + scheduleRetry() + logger.warn(exception) { + "Failed to save ${discoveries.size} unchecked transactions. Retrying the same batch after " + + "${Duration.between(clock.instant(), nextRetryAt)}" + } + return DiscoveryPersistenceResult.Failed + } + + val elapsed = metrics.stopBatch(sample) + queue.acknowledge(discoveries) + retryBatch = null + metrics.persisted(discoveries) + metrics.affectedRows(affectedRows) + if (affectedRows > 0) { + workTracker.markChanged() + } + resetRetry() + logger.info { "Saved ${discoveries.size} unchecked transactions in $elapsed" } + return DiscoveryPersistenceResult.Persisted + } + private fun scheduleRetry() { nextRetryAt = clock.instant().plusSeconds(retryBackoffInSeconds) retryBackoffInSeconds = @@ -103,3 +116,15 @@ class DiscoveryPersistenceWorker( retryBackoffInSeconds = properties.retryInitialBackoffInSeconds } } + +sealed interface DiscoveryPersistenceResult { + data object Idle : DiscoveryPersistenceResult + + data object Busy : DiscoveryPersistenceResult + + data object RetryWaiting : DiscoveryPersistenceResult + + data object Failed : DiscoveryPersistenceResult + + data object Persisted : DiscoveryPersistenceResult +} diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitor.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitor.kt new file mode 100644 index 00000000..0ecf6285 --- /dev/null +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitor.kt @@ -0,0 +1,179 @@ +package cash.atto.node.bootstrap.discovery + +import io.micrometer.core.instrument.Gauge +import io.micrometer.core.instrument.MeterRegistry +import jakarta.annotation.PostConstruct +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Component +import java.nio.file.Files +import java.nio.file.Path +import java.util.concurrent.TimeUnit + +data class LinuxIoPressureReading( + val totalStallMicros: Long, + val avg10Percent: Double, +) + +fun interface LinuxIoPressureSource { + fun read(): LinuxIoPressureReading? +} + +@Component +class ProcLinuxIoPressureSource : LinuxIoPressureSource { + override fun read(): LinuxIoPressureReading? = + try { + parseIoPressure(Files.readString(IO_PRESSURE_PATH)) + } catch (_: Exception) { + null + } + + private companion object { + val IO_PRESSURE_PATH: Path = Path.of("/proc/pressure/io") + } +} + +@Component +class DiscoveryPressureMonitor( + private val source: LinuxIoPressureSource, + meterRegistry: MeterRegistry, +) { + @Volatile + private var snapshot = DiscoveryPressureSnapshot() + + private var previous: ObservedPressure? = null + + init { + Gauge + .builder("transactions.bootstrap.disk.pressure", this) { + it.currentSnapshot().effective + }.description("Effective Linux PSI I/O stall share controlling discovery admission") + .register(meterRegistry) + Gauge + .builder("transactions.bootstrap.disk.pressure.recent", this) { + it.currentSnapshot().recent + }.description("I/O stall share measured from the latest cumulative PSI interval") + .register(meterRegistry) + Gauge + .builder("transactions.bootstrap.disk.pressure.avg10", this) { + it.currentSnapshot().avg10 + }.description("Linux PSI I/O some avg10 stall share") + .register(meterRegistry) + Gauge + .builder("transactions.bootstrap.disk.pressure.available", this) { + if (it.currentSnapshot().available) 1.0 else 0.0 + }.description("Whether Linux PSI I/O pressure is available to discovery admission") + .register(meterRegistry) + } + + @PostConstruct + fun start() { + refresh() + } + + @Scheduled(fixedRate = 1, timeUnit = TimeUnit.SECONDS) + fun refresh() { + refresh(System.nanoTime()) + } + + fun targetCapacity(maximum: Int): Int { + require(maximum > 0) { "Maximum discovery capacity must be positive" } + + return (maximum * availableShare()) + .toInt() + .coerceIn(0, maximum) + } + + internal fun availableShare(): Double = 1.0 - snapshot.effective + + @Synchronized + internal fun refresh(observedAtNanos: Long) { + val reading = source.read() + if ( + reading == null || + reading.totalStallMicros < 0 || + !reading.avg10Percent.isFinite() || + reading.avg10Percent < 0 + ) { + previous = null + snapshot = DiscoveryPressureSnapshot() + return + } + + val avg10 = (reading.avg10Percent / 100.0).coerceIn(0.0, 1.0) + val recent = calculateRecentPressure(previous, reading, observedAtNanos) + previous = ObservedPressure(observedAtNanos, reading.totalStallMicros) + snapshot = + DiscoveryPressureSnapshot( + available = true, + recent = recent, + avg10 = avg10, + effective = maxOf(recent, avg10), + ) + } + + internal fun currentSnapshot(): DiscoveryPressureSnapshot = snapshot +} + +internal data class DiscoveryPressureSnapshot( + val available: Boolean = false, + val recent: Double = 0.0, + val avg10: Double = 0.0, + val effective: Double = 0.0, +) + +private data class ObservedPressure( + val observedAtNanos: Long, + val totalStallMicros: Long, +) + +private fun calculateRecentPressure( + previous: ObservedPressure?, + current: LinuxIoPressureReading, + observedAtNanos: Long, +): Double { + if ( + previous == null || + current.totalStallMicros < previous.totalStallMicros || + observedAtNanos <= previous.observedAtNanos + ) { + return 0.0 + } + + val elapsedNanos = observedAtNanos - previous.observedAtNanos + val stalledMicros = current.totalStallMicros - previous.totalStallMicros + return (stalledMicros.toDouble() * NANOS_PER_MICROSECOND / elapsedNanos) + .coerceIn(0.0, 1.0) +} + +internal fun parseIoPressure(content: String): LinuxIoPressureReading? { + val some = + content + .lineSequence() + .map(String::trim) + .firstOrNull { it.startsWith("some ") } + ?: return null + val fields = + some + .splitToSequence(' ') + .filter(String::isNotEmpty) + .mapNotNull { field -> + val separator = field.indexOf('=') + if (separator < 1 || separator == field.lastIndex) { + null + } else { + field.substring(0, separator) to field.substring(separator + 1) + } + }.toMap() + val total = fields["total"]?.toLongOrNull() ?: return null + val avg10 = fields["avg10"]?.toDoubleOrNull() ?: return null + if (total < 0 || !avg10.isFinite() || avg10 < 0) { + return null + } + + return LinuxIoPressureReading( + totalStallMicros = total, + avg10Percent = avg10, + ) +} + +private const val NANOS_PER_MICROSECOND = 1_000.0 diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueue.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueue.kt index aef91c50..ae42085a 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueue.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueue.kt @@ -20,10 +20,12 @@ class DiscoveryQueue( private val properties: DiscoveryProperties, private val metrics: DiscoveryMetrics, private val clock: Clock, + private val pressureMonitor: DiscoveryPressureMonitor, ) { private val logger = KotlinLogging.logger {} private val duplicateDetector = DuplicateDetector(10.minutes) - private val queued = AtomicInteger() + private val outstanding = AtomicInteger() + private val inFlight = AtomicInteger() private val buffer = Channel( capacity = properties.capacity + properties.headroom, @@ -57,7 +59,7 @@ class DiscoveryQueue( return false } - queued.incrementAndGet() + outstanding.incrementAndGet() buffer.send(discovery) metrics.admitted(source) @@ -71,17 +73,15 @@ class DiscoveryQueue( return true } - fun isAtCapacity(): Boolean = queued.get() >= properties.capacity + fun isAtCapacity(): Boolean = outstanding.get() >= getTargetCapacity() - fun remainingTargetCapacity(): Int = maxOf(0, properties.capacity - queued.get()) + fun remainingTargetCapacity(): Int = maxOf(0, getTargetCapacity() - outstanding.get()) internal fun takeBatch(): List = buildList(properties.batchSize) { while (size < properties.batchSize) { val discovery = buffer.tryReceive().getOrNull() ?: break - check(queued.decrementAndGet() >= 0) { - "Discovery queue depth became negative" - } + inFlight.incrementAndGet() metrics.dequeued(Duration.between(discovery.enqueuedAt, clock.instant())) add(discovery) } @@ -91,22 +91,47 @@ class DiscoveryQueue( batch.forEach { discovery -> duplicateDetector.refresh(discovery.transaction.hash) } + releaseInFlight(batch.size) } - internal fun reset() { + internal fun reset(retryBatch: List) { + releaseInFlight(retryBatch.size) while (true) { buffer.tryReceive().getOrNull() ?: break - queued.decrementAndGet() + check(outstanding.decrementAndGet() >= 0) { + "Discovery outstanding count became negative" + } } duplicateDetector.clear() } - internal fun getBacklogDepth(): Int = queued.get() + internal fun getBacklogDepth(): Int = outstanding.get() + + internal fun getInFlightCount(): Int = inFlight.get() + + internal fun getTargetCapacity(): Int = pressureMonitor.targetCapacity(properties.capacity) - internal fun getBacklogOvershoot(): Int = maxOf(0, queued.get() - properties.capacity) + internal fun getBacklogOvershoot(): Int = maxOf(0, outstanding.get() - getTargetCapacity()) + + internal fun isPhysicalBufferFull(): Boolean = outstanding.get() - inFlight.get() >= properties.capacity + properties.headroom private fun discard(discovery: PendingDiscovery) { - queued.decrementAndGet() + check(outstanding.decrementAndGet() >= 0) { + "Discovery outstanding count became negative" + } duplicateDetector.remove(discovery.transaction.hash) } + + private fun releaseInFlight(count: Int) { + if (count == 0) { + return + } + + check(inFlight.addAndGet(-count) >= 0) { + "Discovery in-flight count became negative" + } + check(outstanding.addAndGet(-count) >= 0) { + "Discovery outstanding count became negative" + } + } } diff --git a/src/main/kotlin/cash/atto/node/bootstrap/discovery/GapDiscoverer.kt b/src/main/kotlin/cash/atto/node/bootstrap/discovery/GapDiscoverer.kt index de9c9de7..d3efb0cf 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/discovery/GapDiscoverer.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/discovery/GapDiscoverer.kt @@ -20,7 +20,6 @@ import io.micrometer.core.instrument.MeterRegistry import io.micrometer.core.instrument.Timer import kotlinx.coroutines.flow.toList import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock import org.springframework.context.event.EventListener import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Component @@ -89,18 +88,29 @@ class GapDiscoverer( .forEach { activeSessions.remove(it.publicKey, it) } } + // A retry may wait for queue space, so this must not occupy Spring's shared fixed-delay scheduler. @Scheduled(fixedRate = 1, timeUnit = TimeUnit.SECONDS) - suspend fun resolve() { - if (resolveMutex.isLocked) { + suspend fun maintainSessions() { + if (!resolveMutex.tryLock()) { return } - resolveMutex.withLock { + try { retryActiveSessions() + } finally { + resolveMutex.unlock() + } + } + + suspend fun discoverIfDue(): GapDiscoveryResult { + if (!resolveMutex.tryLock()) { + return GapDiscoveryResult.Busy + } + try { val slots = availableSessionSlots() if (slots == 0 || shouldSkipIdleScan()) { - return + return GapDiscoveryResult.Idle } val startingGeneration = workTracker.currentGeneration() @@ -124,6 +134,9 @@ class GapDiscoverer( } updateIdleSuppression(gaps.isNotEmpty(), startingGeneration) + return GapDiscoveryResult.Queried + } finally { + resolveMutex.unlock() } } @@ -268,6 +281,14 @@ class GapDiscoverer( } } +sealed interface GapDiscoveryResult { + data object Idle : GapDiscoveryResult + + data object Busy : GapDiscoveryResult + + data object Queried : GapDiscoveryResult +} + private fun GapView.startHeight(requestBudget: ULong): AttoHeight { val count = endHeight - startHeight + 1U if (count.value > requestBudget) { diff --git a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessor.kt b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessor.kt index 13d02d9b..b5eb307e 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessor.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessor.kt @@ -8,15 +8,27 @@ import cash.atto.node.account.AccountService import cash.atto.node.account.getByAlgorithmAndPublicKey import cash.atto.node.bootstrap.TransactionResolved import cash.atto.node.bootstrap.TransactionStuck +import cash.atto.node.bootstrap.discovery.DiscoveryProperties import cash.atto.node.transaction.Transaction import cash.atto.node.transaction.TransactionSource import cash.atto.node.transaction.validation.TransactionValidationManager import io.github.oshai.kotlinlogging.KotlinLogging +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.MeterRegistry +import io.micrometer.core.instrument.Timer +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock import org.springframework.stereotype.Component -import org.springframework.transaction.annotation.Isolation -import org.springframework.transaction.annotation.Transactional +import org.springframework.transaction.ReactiveTransactionManager +import org.springframework.transaction.TransactionDefinition +import org.springframework.transaction.reactive.TransactionalOperator +import org.springframework.transaction.reactive.executeAndAwait +import org.springframework.transaction.support.DefaultTransactionDefinition +import java.time.Clock +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.time.measureTime private val logger = KotlinLogging.logger {} @@ -26,57 +38,193 @@ class UncheckedTransactionProcessor( private val transactionValidationManager: TransactionValidationManager, private val accountService: AccountService, private val eventPublisher: EventPublisher, + private val uncheckedTransactionRepository: UncheckedTransactionRepository, + private val uncheckedTransactionService: UncheckedTransactionService, + private val workTracker: UncheckedWorkTracker, + private val discoveryProperties: DiscoveryProperties, + meterRegistry: MeterRegistry, + transactionManager: ReactiveTransactionManager, + private val clock: Clock, ) { private val mutex = Mutex() + private val batchSize = 1_000L + private val resolutionTransaction = + TransactionalOperator.create( + transactionManager, + DefaultTransactionDefinition().apply { + isolationLevel = TransactionDefinition.ISOLATION_READ_COMMITTED + }, + ) - @Transactional(isolation = Isolation.READ_COMMITTED) - suspend fun process(candidateTransactions: Collection): Int { + private val selectionTimer = + Timer + .builder("transactions.unchecked.resolution.query") + .description("Time selecting unchecked transactions eligible for resolution") + .register(meterRegistry) + private val processingTimer = + Timer + .builder("transactions.unchecked.resolution.process") + .description("Time validating and resolving unchecked transactions") + .register(meterRegistry) + private val cleanupTimer = + Timer + .builder("transactions.unchecked.cleanup") + .description("Time deleting bounded batches of stale unchecked transactions") + .register(meterRegistry) + private val resolvedTransactions = + Counter + .builder("transactions.unchecked.resolved") + .description("Unchecked transactions successfully resolved") + .register(meterRegistry) + private val deletedTransactions = + Counter + .builder("transactions.unchecked.cleanup.deleted") + .description("Stale unchecked transactions deleted") + .register(meterRegistry) + + @Volatile + private var idleGeneration = Long.MIN_VALUE + + @Volatile + private var nextMaintenanceAt = Instant.EPOCH + + suspend fun processIfDue(): UncheckedProcessingResult { + if (shouldSkipIdleScan()) { + return UncheckedProcessingResult.SkippedIdle + } + if (!mutex.tryLock()) { + return UncheckedProcessingResult.SkippedBusy + } + + try { + if (shouldSkipIdleScan()) { + return UncheckedProcessingResult.SkippedIdle + } + + val startingGeneration = workTracker.currentGeneration() + val candidateTransactions = + selectionTimer.recordSuspending { + uncheckedTransactionRepository + .findTopOldest(batchSize) + .map { it.toTransaction() } + .toList() + } + val resolved = resolve(candidateTransactions) + val deleted = + cleanupTimer.recordSuspending { + uncheckedTransactionService.cleanUp(batchSize) + } + if (deleted > 0) { + deletedTransactions.increment(deleted.toDouble()) + } + + val endingGeneration = workTracker.currentGeneration() + if (resolved == 0 && deleted == 0 && startingGeneration == endingGeneration) { + idleGeneration = endingGeneration + nextMaintenanceAt = clock.instant().plusSeconds(discoveryProperties.idleQueryFallbackInSeconds) + } else { + idleGeneration = Long.MIN_VALUE + nextMaintenanceAt = Instant.EPOCH + } + + return UncheckedProcessingResult.Completed( + selected = candidateTransactions.size, + resolved = resolved, + deleted = deleted, + ) + } finally { + mutex.unlock() + } + } + + private suspend fun resolve(candidateTransactions: List): Int { if (candidateTransactions.isEmpty()) { return 0 } + var resolved = 0 + val elapsed = + measureTime { + resolved = + processingTimer.recordSuspending { + resolutionTransaction.executeAndAwait { + resolveBatch(candidateTransactions) + } + } + } + if (resolved > 0) { + resolvedTransactions.increment(resolved.toDouble()) + logger.info { "Resolved $resolved unchecked transactions in $elapsed" } + } + return resolved + } + + private suspend fun resolveBatch(candidateTransactions: List): Int { logger.debug { "Starting resolution of ${candidateTransactions.size} unchecked transaction..." } - mutex.withLock { - val accountMap = HashMap() - val violations = HashSet() - var resolvedCounter = 0 + val accountMap = HashMap() + val violations = HashSet() + var resolvedCounter = 0 - candidateTransactions - .forEach { transaction -> - if (violations.contains(transaction.publicKey)) { - logger.debug { "Skipping $transaction because previous transaction for the same public key already failed" } - return@forEach - } + candidateTransactions.forEach { transaction -> + if (violations.contains(transaction.publicKey)) { + logger.debug { "Skipping $transaction because previous transaction for the same public key already failed" } + return@forEach + } - logger.debug { "Unchecked solving $transaction" } - val account = - accountMap[transaction.publicKey] ?: accountRepository.getByAlgorithmAndPublicKey( - transaction.algorithm, - transaction.publicKey, - transaction.block.network, - ) - - logger.debug { "Start validation $transaction from $account" } - - val violation = transactionValidationManager.validate(account, transaction) - if (violation != null) { - eventPublisher.publish(TransactionStuck(violation.reason, transaction)) - violations.add(transaction.publicKey) - return@forEach - } + logger.debug { "Unchecked solving $transaction" } + val account = + accountMap[transaction.publicKey] ?: accountRepository.getByAlgorithmAndPublicKey( + transaction.algorithm, + transaction.publicKey, + transaction.block.network, + ) - logger.debug { "No violation found for $transaction" } + logger.debug { "Start validation $transaction from $account" } - accountMap[transaction.publicKey] = accountService.add(TransactionSource.BOOTSTRAP, listOf(transaction)).first() + val violation = transactionValidationManager.validate(account, transaction) + if (violation != null) { + eventPublisher.publish(TransactionStuck(violation.reason, transaction)) + violations.add(transaction.publicKey) + return@forEach + } - logger.debug { "Resolved $transaction" } - eventPublisher.publishAfterCommit(TransactionResolved(transaction)) + logger.debug { "No violation found for $transaction" } - resolvedCounter++ - } + accountMap[transaction.publicKey] = + accountService.add(TransactionSource.BOOTSTRAP, listOf(transaction)).first() + + logger.debug { "Resolved $transaction" } + eventPublisher.publishAfterCommit(TransactionResolved(transaction)) - return resolvedCounter + resolvedCounter++ } + + return resolvedCounter + } + + private fun shouldSkipIdleScan(): Boolean = + workTracker.currentGeneration() == idleGeneration && + clock.instant().isBefore(nextMaintenanceAt) +} + +sealed interface UncheckedProcessingResult { + data object SkippedIdle : UncheckedProcessingResult + + data object SkippedBusy : UncheckedProcessingResult + + data class Completed( + val selected: Int, + val resolved: Int, + val deleted: Int, + ) : UncheckedProcessingResult +} + +private suspend fun Timer.recordSuspending(block: suspend () -> T): T { + val startedAt = System.nanoTime() + try { + return block() + } finally { + record(System.nanoTime() - startedAt, TimeUnit.NANOSECONDS) } } diff --git a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarter.kt b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarter.kt deleted file mode 100644 index c0a52c04..00000000 --- a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarter.kt +++ /dev/null @@ -1,161 +0,0 @@ -package cash.atto.node.bootstrap.unchecked - -import cash.atto.node.bootstrap.discovery.DiscoveryProperties -import cash.atto.node.transaction.Transaction -import io.github.oshai.kotlinlogging.KotlinLogging -import io.micrometer.core.instrument.Counter -import io.micrometer.core.instrument.MeterRegistry -import io.micrometer.core.instrument.Timer -import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.toList -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.Instant -import java.util.concurrent.TimeUnit -import kotlin.time.measureTime - -@Component -class UncheckedTransactionProcessorStarter( - private val uncheckedTransactionRepository: UncheckedTransactionRepository, - private val processor: UncheckedTransactionProcessor, - private val uncheckedTransactionService: UncheckedTransactionService, - private val workTracker: UncheckedWorkTracker, - private val discoveryProperties: DiscoveryProperties, - private val meterRegistry: MeterRegistry, - private val clock: Clock, -) { - private val logger = KotlinLogging.logger {} - private val mutex = Mutex() - private val batchSize = 1_000L - private val maxResolutionPassNanos = TimeUnit.MILLISECONDS.toNanos(500) - private val maxCleanupPassNanos = TimeUnit.MILLISECONDS.toNanos(500) - private val maxCleanupBatches = 10 - - private val selectionTimer = - Timer - .builder("transactions.unchecked.resolution.query") - .description("Time selecting unchecked transactions eligible for resolution") - .register(meterRegistry) - private val processingTimer = - Timer - .builder("transactions.unchecked.resolution.process") - .description("Time validating and resolving unchecked transactions") - .register(meterRegistry) - private val cleanupTimer = - Timer - .builder("transactions.unchecked.cleanup") - .description("Time deleting bounded batches of stale unchecked transactions") - .register(meterRegistry) - private val resolvedTransactions = - Counter - .builder("transactions.unchecked.resolved") - .description("Unchecked transactions successfully resolved") - .register(meterRegistry) - private val deletedTransactions = - Counter - .builder("transactions.unchecked.cleanup.deleted") - .description("Stale unchecked transactions deleted") - .register(meterRegistry) - - @Volatile - private var idleGeneration = Long.MIN_VALUE - - @Volatile - private var nextMaintenanceAt = Instant.EPOCH - - @Scheduled(fixedRate = 1, timeUnit = TimeUnit.SECONDS) - suspend fun process() { - if (mutex.isLocked || shouldSkipIdleScan()) { - return - } - mutex.withLock { - if (shouldSkipIdleScan()) { - return - } - - val startingGeneration = workTracker.currentGeneration() - val passStarted = System.nanoTime() - var madeProgress = false - var candidateTransactions: List - var resolvedCounter = 0 - - do { - candidateTransactions = - selectionTimer.recordSuspending { - uncheckedTransactionRepository - .findTopOldest(batchSize) - .map { it.toTransaction() } - .toList() - } - - val elapsed = - measureTime { - resolvedCounter = 0 - resolvedCounter = - processingTimer.recordSuspending { - processor.process(candidateTransactions) - } - } - if (resolvedCounter > 0) { - madeProgress = true - resolvedTransactions.increment(resolvedCounter.toDouble()) - logger.info { "Resolved $resolvedCounter unchecked transactions in $elapsed" } - } - - val deleted = cleanUp() - if (deleted > 0) { - madeProgress = true - deletedTransactions.increment(deleted.toDouble()) - } - } while ( - candidateTransactions.isNotEmpty() && - resolvedCounter == candidateTransactions.size && - System.nanoTime() - passStarted < maxResolutionPassNanos - ) - - val endingGeneration = workTracker.currentGeneration() - if (!madeProgress && startingGeneration == endingGeneration) { - idleGeneration = endingGeneration - nextMaintenanceAt = clock.instant().plusSeconds(discoveryProperties.idleQueryFallbackInSeconds) - } else { - idleGeneration = Long.MIN_VALUE - nextMaintenanceAt = Instant.EPOCH - } - } - } - - private suspend fun cleanUp(): Int { - val startedAt = System.nanoTime() - var deletedTotal = 0 - - repeat(maxCleanupBatches) { - val deleted = - cleanupTimer.recordSuspending { - uncheckedTransactionService.cleanUp(batchSize) - } - deletedTotal += deleted - - if (deleted < batchSize || System.nanoTime() - startedAt >= maxCleanupPassNanos) { - return deletedTotal - } - } - - return deletedTotal - } - - private fun shouldSkipIdleScan(): Boolean = - workTracker.currentGeneration() == idleGeneration && - clock.instant().isBefore(nextMaintenanceAt) -} - -private suspend fun Timer.recordSuspending(block: suspend () -> T): T { - val startedAt = System.nanoTime() - try { - return block() - } finally { - record(System.nanoTime() - startedAt, TimeUnit.NANOSECONDS) - } -} diff --git a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionService.kt b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionService.kt index f8ea45b2..606055b1 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionService.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionService.kt @@ -1,8 +1,5 @@ package cash.atto.node.bootstrap.unchecked -import cash.atto.node.EventPublisher -import cash.atto.node.bootstrap.UncheckedTransactionSaved -import cash.atto.node.executeAfterCommit import io.github.oshai.kotlinlogging.KotlinLogging import io.micrometer.core.annotation.Timed import org.springframework.stereotype.Service @@ -13,7 +10,6 @@ import org.springframework.transaction.annotation.Transactional class UncheckedTransactionService( private val uncheckedTransactionRepository: UncheckedTransactionRepository, private val uncheckedTransactionInserter: UncheckedTransactionInserter, - private val eventPublisher: EventPublisher, ) { private val logger = KotlinLogging.logger {} @@ -21,12 +17,7 @@ class UncheckedTransactionService( @Timed("unchecked_transactions_save", description = "Time taken to save an unchecked transaction") suspend fun save(uncheckedTransactions: Collection): Long { val affectedRows = uncheckedTransactionInserter.insert(uncheckedTransactions) - executeAfterCommit { - uncheckedTransactions.forEach { - logger.debug { "Saved $it" } - eventPublisher.publish(UncheckedTransactionSaved(it)) - } - } + logger.debug { "Saved ${uncheckedTransactions.size} unchecked transactions" } return affectedRows } diff --git a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedWorkTracker.kt b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedWorkTracker.kt index ac350d07..7d8225bd 100644 --- a/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedWorkTracker.kt +++ b/src/main/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedWorkTracker.kt @@ -1,7 +1,6 @@ package cash.atto.node.bootstrap.unchecked import cash.atto.node.account.AccountUpdated -import cash.atto.node.bootstrap.UncheckedTransactionSaved import org.springframework.context.event.EventListener import org.springframework.stereotype.Component import java.util.concurrent.atomic.AtomicLong @@ -22,11 +21,4 @@ class UncheckedWorkTracker { ) { markChanged() } - - @EventListener - fun process( - @Suppress("UNUSED_PARAMETER") event: UncheckedTransactionSaved, - ) { - markChanged() - } } diff --git a/src/test/kotlin/cash/atto/node/bootstrap/BootstrapControllerTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/BootstrapControllerTest.kt new file mode 100644 index 00000000..7981cf39 --- /dev/null +++ b/src/test/kotlin/cash/atto/node/bootstrap/BootstrapControllerTest.kt @@ -0,0 +1,283 @@ +package cash.atto.node.bootstrap + +import cash.atto.node.bootstrap.discovery.DiscoveryPersistenceResult +import cash.atto.node.bootstrap.discovery.DiscoveryPersistenceWorker +import cash.atto.node.bootstrap.discovery.DiscoveryPressureMonitor +import cash.atto.node.bootstrap.discovery.DiscoveryQueue +import cash.atto.node.bootstrap.discovery.GapDiscoverer +import cash.atto.node.bootstrap.discovery.GapDiscoveryResult +import cash.atto.node.bootstrap.unchecked.UncheckedProcessingResult +import cash.atto.node.bootstrap.unchecked.UncheckedTransactionProcessor +import io.micrometer.core.instrument.simple.SimpleMeterRegistry +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class BootstrapControllerTest { + @Test + fun `maintenance progress keeps priority on the following tick`() = + runTest { + // Given + val fixture = fixture() + coEvery { fixture.processor.processIfDue() } returns + UncheckedProcessingResult.Completed(selected = 10, resolved = 5, deleted = 5) + + // When + fixture.controller.runOnce() + fixture.controller.runOnce() + + // Then + coVerify(exactly = 2) { fixture.processor.processIfDue() } + coVerify(exactly = 0) { fixture.worker.persistIfReady() } + coVerify(exactly = 0) { fixture.gapDiscoverer.discoverIfDue() } + assertEquals(2.0, fixture.decisions("maintenance")) + } + + @Test + fun `gap discovery runs on the tick after maintenance reports no progress`() = + runTest { + // Given + val fixture = fixture() + coEvery { fixture.processor.processIfDue() } returnsMany + listOf( + UncheckedProcessingResult.Completed(0, 0, 0), + UncheckedProcessingResult.SkippedIdle, + ) + coEvery { fixture.gapDiscoverer.discoverIfDue() } returns GapDiscoveryResult.Queried + + // When + fixture.controller.runOnce() + fixture.controller.runOnce() + + // Then + coVerify(exactly = 2) { fixture.processor.processIfDue() } + coVerify(exactly = 1) { fixture.worker.persistIfReady() } + coVerify(exactly = 1) { fixture.gapDiscoverer.discoverIfDue() } + assertEquals(1.0, fixture.decisions("maintenance")) + assertEquals(1.0, fixture.decisions("gap")) + } + + @Test + fun `queued persistence runs before gap discovery when maintenance is idle`() = + runTest { + // Given + val fixture = fixture() + coEvery { fixture.worker.persistIfReady() } returns DiscoveryPersistenceResult.Persisted + + // When + fixture.controller.runOnce() + + // Then + coVerify(exactly = 1) { fixture.processor.processIfDue() } + coVerify(exactly = 1) { fixture.worker.persistIfReady() } + coVerify(exactly = 0) { fixture.gapDiscoverer.discoverIfDue() } + assertEquals(1.0, fixture.decisions("persistence")) + } + + @Test + fun `fractional pressure credit skips ticks without shrinking batch work`() = + runTest { + // Given + val fixture = fixture() + every { fixture.pressureMonitor.availableShare() } returns 0.5 + coEvery { fixture.processor.processIfDue() } returns + UncheckedProcessingResult.Completed(1, 1, 0) + + // When + fixture.controller.runOnce() + + // Then + coVerify(exactly = 0) { fixture.processor.processIfDue() } + assertEquals(0.5, fixture.diskCredit()) + assertEquals(1.0, fixture.decisions("pressure-wait")) + + // When + fixture.controller.runOnce() + + // Then + coVerify(exactly = 1) { fixture.processor.processIfDue() } + assertEquals(0.0, fixture.diskCredit()) + assertEquals(1.0, fixture.decisions("maintenance")) + } + + @Test + fun `full pressure pauses normal work but a full physical buffer forces one drain`() = + runTest { + // Given + val fixture = fixture() + every { fixture.pressureMonitor.availableShare() } returns 0.0 + + // When + fixture.controller.runOnce() + + // Then + coVerify(exactly = 0) { fixture.worker.persistIfReady() } + assertEquals(1.0, fixture.decisions("pressure-wait")) + + // Given + every { fixture.queue.isPhysicalBufferFull() } returns true + coEvery { fixture.worker.persistIfReady() } returns DiscoveryPersistenceResult.Persisted + + // When + fixture.controller.runOnce() + + // Then + coVerify(exactly = 1) { fixture.worker.persistIfReady() } + coVerify(exactly = 0) { fixture.processor.processIfDue() } + coVerify(exactly = 0) { fixture.gapDiscoverer.discoverIfDue() } + assertEquals(0.0, fixture.diskCredit()) + assertEquals(1.0, fixture.decisions("forced-drain")) + } + + @Test + fun `retry backoff blocks every bootstrap database action`() = + runTest { + // Given + val fixture = fixture() + every { fixture.worker.isRetryWaiting() } returns true + + // When + fixture.controller.runOnce() + + // Then + verify(exactly = 1) { fixture.worker.isRetryWaiting() } + coVerify(exactly = 0) { fixture.worker.persistIfReady() } + coVerify(exactly = 0) { fixture.processor.processIfDue() } + coVerify(exactly = 0) { fixture.gapDiscoverer.discoverIfDue() } + assertEquals(1.0, fixture.decisions("retry-wait")) + } + + @Test + fun `due retry runs before unchecked maintenance`() = + runTest { + // Given + val fixture = fixture() + every { fixture.worker.hasRetryBatch() } returns true + coEvery { fixture.worker.persistIfReady() } returns DiscoveryPersistenceResult.Persisted + + // When + fixture.controller.runOnce() + + // Then + coVerify(exactly = 1) { fixture.worker.persistIfReady() } + coVerify(exactly = 0) { fixture.processor.processIfDue() } + coVerify(exactly = 0) { fixture.gapDiscoverer.discoverIfDue() } + assertEquals(1.0, fixture.decisions("persistence")) + } + + @Test + fun `overlapping ticks do not start another bootstrap action`() = + runTest { + // Given + val fixture = fixture() + val maintenanceStarted = CompletableDeferred() + val releaseMaintenance = CompletableDeferred() + coEvery { fixture.processor.processIfDue() } coAnswers { + maintenanceStarted.complete(Unit) + releaseMaintenance.await() + UncheckedProcessingResult.Completed(1, 1, 0) + } + + // When + val activeTick = async { fixture.controller.runOnce() } + maintenanceStarted.await() + fixture.controller.runOnce() + + // Then + coVerify(exactly = 1) { fixture.processor.processIfDue() } + coVerify(exactly = 0) { fixture.worker.persistIfReady() } + coVerify(exactly = 0) { fixture.gapDiscoverer.discoverIfDue() } + + // When + releaseMaintenance.complete(Unit) + activeTick.await() + + // Then + assertEquals(1.0, fixture.decisions("maintenance")) + } + + @Test + fun `unavailable pressure preserves one action per completed tick`() = + runTest { + // Given + val fixture = fixture() + every { fixture.pressureMonitor.availableShare() } returns 1.0 + coEvery { fixture.processor.processIfDue() } returns + UncheckedProcessingResult.Completed(1, 1, 0) + + // When + fixture.controller.runOnce() + fixture.controller.runOnce() + + // Then + coVerify(exactly = 2) { fixture.processor.processIfDue() } + assertEquals(2.0, fixture.decisions("maintenance")) + assertEquals(0.0, fixture.diskCredit()) + } + + private fun fixture(): Fixture { + val registry = SimpleMeterRegistry() + val pressureMonitor = mockk() + val queue = mockk() + val worker = mockk() + val processor = mockk() + val gapDiscoverer = mockk() + + every { pressureMonitor.availableShare() } returns 1.0 + every { queue.isPhysicalBufferFull() } returns false + every { worker.isRetryWaiting() } returns false + every { worker.hasRetryBatch() } returns false + coEvery { worker.persistIfReady() } returns DiscoveryPersistenceResult.Idle + coEvery { processor.processIfDue() } returns UncheckedProcessingResult.SkippedIdle + coEvery { gapDiscoverer.discoverIfDue() } returns GapDiscoveryResult.Idle + + val controller = + BootstrapController( + pressureMonitor = pressureMonitor, + discoveryQueue = queue, + persistenceWorker = worker, + uncheckedTransactionProcessor = processor, + gapDiscoverer = gapDiscoverer, + meterRegistry = registry, + ) + return Fixture( + controller = controller, + pressureMonitor = pressureMonitor, + queue = queue, + worker = worker, + processor = processor, + gapDiscoverer = gapDiscoverer, + registry = registry, + ) + } + + private data class Fixture( + val controller: BootstrapController, + val pressureMonitor: DiscoveryPressureMonitor, + val queue: DiscoveryQueue, + val worker: DiscoveryPersistenceWorker, + val processor: UncheckedTransactionProcessor, + val gapDiscoverer: GapDiscoverer, + val registry: SimpleMeterRegistry, + ) { + fun decisions(decision: String): Double = + registry + .get("transactions.bootstrap.controller.decisions") + .tag("decision", decision) + .counter() + .count() + + fun diskCredit(): Double = + registry + .get("transactions.bootstrap.controller.disk.credit") + .gauge() + .value() + } +} diff --git a/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitorTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitorTest.kt new file mode 100644 index 00000000..051b5a98 --- /dev/null +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryPressureMonitorTest.kt @@ -0,0 +1,184 @@ +package cash.atto.node.bootstrap.discovery + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.time.Duration + +class DiscoveryPressureMonitorTest { + @Test + fun `recent spike dominates a lower avg10`() { + // Given + val fixture = PressureFixture(LinuxIoPressureReading(1_000, 10.0)) + fixture.refresh() + + // When: 600 ms of the next second was I/O stalled. + fixture.reading = LinuxIoPressureReading(601_000, 10.0) + fixture.advanceAndRefresh(Duration.ofSeconds(1)) + + // Then + assertEquals( + DiscoveryPressureSnapshot( + available = true, + recent = 0.6, + avg10 = 0.1, + effective = 0.6, + ), + fixture.monitor.currentSnapshot(), + ) + assertEquals(100, fixture.monitor.targetCapacity(250)) + } + + @Test + fun `avg10 controls recovery when the latest interval is quieter`() { + // Given + val fixture = PressureFixture(LinuxIoPressureReading(1_000, 60.0)) + fixture.refresh() + + // When + fixture.reading = LinuxIoPressureReading(101_000, 55.0) + fixture.advanceAndRefresh(Duration.ofSeconds(1)) + + // Then + assertEquals(0.1, fixture.monitor.currentSnapshot().recent, 0.000_001) + assertEquals(0.55, fixture.monitor.currentSnapshot().avg10, 0.000_001) + assertEquals(0.55, fixture.monitor.currentSnapshot().effective, 0.000_001) + assertEquals(112, fixture.monitor.targetCapacity(250)) + } + + @Test + fun `first reading and counter reset use avg10 without inventing recent pressure`() { + // Given + val fixture = PressureFixture(LinuxIoPressureReading(1_000, 20.0)) + + // When + fixture.refresh() + + // Then + assertEquals(0.0, fixture.monitor.currentSnapshot().recent) + assertEquals(0.2, fixture.monitor.currentSnapshot().effective) + + // When + fixture.reading = LinuxIoPressureReading(500, 30.0) + fixture.advanceAndRefresh(Duration.ofSeconds(1)) + + // Then + assertEquals(0.0, fixture.monitor.currentSnapshot().recent) + assertEquals(0.3, fixture.monitor.currentSnapshot().effective) + } + + @Test + fun `nonpositive elapsed time keeps avg10 and resets the recent baseline`() { + // Given + val fixture = PressureFixture(LinuxIoPressureReading(1_000, 20.0)) + fixture.refresh() + + // When + fixture.reading = LinuxIoPressureReading(501_000, 30.0) + fixture.refresh() + + // Then + assertEquals(0.0, fixture.monitor.currentSnapshot().recent) + assertEquals(0.3, fixture.monitor.currentSnapshot().effective) + } + + @Test + fun `unavailable source fails open immediately and establishes a fresh baseline`() { + // Given + val fixture = PressureFixture(LinuxIoPressureReading(1_000, 100.0)) + fixture.refresh() + assertEquals(0, fixture.monitor.targetCapacity(250)) + + // When + fixture.reading = null + fixture.advanceAndRefresh(Duration.ofSeconds(1)) + + // Then + assertFalse(fixture.monitor.currentSnapshot().available) + assertEquals(250, fixture.monitor.targetCapacity(250)) + + // When + fixture.reading = LinuxIoPressureReading(901_000, 10.0) + fixture.advanceAndRefresh(Duration.ofSeconds(1)) + + // Then + assertTrue(fixture.monitor.currentSnapshot().available) + assertEquals(0.0, fixture.monitor.currentSnapshot().recent) + assertEquals(0.1, fixture.monitor.currentSnapshot().effective) + } + + @Test + fun `capacity and metrics read the cached snapshot without resampling`() { + // Given + val registry = SimpleMeterRegistry() + var reads = 0 + val monitor = + DiscoveryPressureMonitor( + source = { + reads++ + LinuxIoPressureReading(1_000, 60.0) + }, + meterRegistry = registry, + ) + monitor.refresh(0) + + // When + assertEquals(100, monitor.targetCapacity(250)) + registry.get("transactions.bootstrap.disk.pressure").gauge().value() + registry.get("transactions.bootstrap.disk.pressure.recent").gauge().value() + registry.get("transactions.bootstrap.disk.pressure.avg10").gauge().value() + registry.get("transactions.bootstrap.disk.pressure.available").gauge().value() + + // Then + assertEquals(1, reads) + } + + @Test + fun `parses avg10 and cumulative total from the same some line`() { + val content = + """ + some avg10=1.45 avg60=3.89 avg300=2.13 total=612783157 + full avg10=1.41 avg60=3.65 avg300=1.99 total=569653160 + """.trimIndent() + + assertEquals( + LinuxIoPressureReading( + totalStallMicros = 612_783_157, + avg10Percent = 1.45, + ), + parseIoPressure(content), + ) + } + + @Test + fun `rejects unavailable or malformed some readings`() { + assertNull(parseIoPressure("full avg10=1.0 total=42")) + assertNull(parseIoPressure("some avg10=1.0 total=invalid")) + assertNull(parseIoPressure("some avg10=invalid total=42")) + assertNull(parseIoPressure("some avg10=NaN total=42")) + assertNull(parseIoPressure("some avg10=1.0")) + } + + private class PressureFixture( + var reading: LinuxIoPressureReading?, + ) { + private var now = 0L + val monitor = + DiscoveryPressureMonitor( + source = { reading }, + meterRegistry = SimpleMeterRegistry(), + ) + + fun refresh() { + monitor.refresh(now) + } + + fun advanceAndRefresh(elapsed: Duration) { + now += elapsed.toNanos() + refresh() + } + } +} diff --git a/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueueTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueueTest.kt index 04fc4be2..fe5b2bec 100644 --- a/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueueTest.kt +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryQueueTest.kt @@ -15,12 +15,16 @@ import cash.atto.node.EventPublisher import cash.atto.node.bootstrap.TransactionDiscovered import cash.atto.node.bootstrap.unchecked.UncheckedTransaction import cash.atto.node.bootstrap.unchecked.UncheckedTransactionService +import cash.atto.node.bootstrap.unchecked.UncheckedWorkTracker import cash.atto.node.transaction.Transaction import io.micrometer.core.instrument.simple.SimpleMeterRegistry import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every import io.mockk.mockk import io.mockk.verify import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.async import kotlinx.coroutines.cancelAndJoin @@ -37,6 +41,7 @@ import java.time.Duration import java.time.Instant import java.time.ZoneId import java.time.ZoneOffset +import java.util.concurrent.atomic.AtomicInteger @OptIn(ExperimentalCoroutinesApi::class) class DiscoveryQueueTest { @@ -62,12 +67,52 @@ class DiscoveryQueueTest { // When suspended.cancelAndJoin() - fixture.worker.flush() + fixture.worker.persistIfReady() + + // Then + assertEquals(1, fixture.queue.remainingTargetCapacity()) + + // When + fixture.worker.persistIfReady() // Then assertEquals(2, fixture.queue.remainingTargetCapacity()) } + @Test + fun `adaptive target pauses clients without resizing the physical channel`() = + runTest { + // Given + val fixture = + fixture( + service = mockk(relaxed = true), + properties = properties(capacity = 3, headroom = 2, batchSize = 1), + ) + fixture.discoveryCapacity.set(1) + + // When + fixture.queue.queue(event(24), DiscoverySource.GAP) + + // Then + assertTrue(fixture.queue.isAtCapacity()) + assertEquals(0, fixture.queue.remainingTargetCapacity()) + assertEquals(1.0, fixture.gauge("transactions.discovery.capacity.target")) + + // When: an already in-flight response still enters the physical headroom. + fixture.queue.queue(event(25), DiscoverySource.GAP) + + // Then + assertEquals(1.0, fixture.gauge("transactions.discovery.backlog.overshoot")) + + // When + fixture.discoveryCapacity.set(3) + + // Then + assertFalse(fixture.queue.isAtCapacity()) + assertEquals(1, fixture.queue.remainingTargetCapacity()) + assertEquals(3.0, fixture.gauge("transactions.discovery.capacity.target")) + } + @Test fun `capacity suspends the next admission without blocking unrelated work and preserves FIFO order`() = runTest { @@ -102,9 +147,10 @@ class DiscoveryQueueTest { assertEquals(1.0, fixture.gauge("transactions.discovery.backlog.overshoot")) // When - fixture.worker.flush() + fixture.worker.persistIfReady() assertTrue(thirdAdmission.await()) - fixture.worker.flush() + fixture.worker.persistIfReady() + fixture.worker.persistIfReady() // Then assertEquals(events.map { it.transaction.hash }, savedHashes) @@ -145,9 +191,10 @@ class DiscoveryQueueTest { assertFalse(fifth.isCompleted) // When - fixture.worker.flush() + fixture.worker.persistIfReady() fifth.await() - fixture.worker.flush() + fixture.worker.persistIfReady() + fixture.worker.persistIfReady() // Then assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) @@ -179,18 +226,26 @@ class DiscoveryQueueTest { fixture.queue.queue(events[1], DiscoverySource.DEPENDENCY) // When - fixture.worker.flush() + fixture.worker.persistIfReady() fixture.queue.queue(events[2], DiscoverySource.DEPENDENCY) - fixture.worker.flush() + fixture.worker.persistIfReady() // Then assertEquals(1, attempts.size) - assertFalse(fixture.queue.isAtCapacity()) - assertEquals(1.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertTrue(fixture.queue.isAtCapacity()) + assertEquals(3.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(2.0, fixture.gauge("transactions.discovery.in.flight")) // When clock.advance(Duration.ofSeconds(1)) - fixture.worker.flush() + fixture.worker.persistIfReady() + + // Then + assertEquals(1.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(0.0, fixture.gauge("transactions.discovery.in.flight")) + + // When + fixture.worker.persistIfReady() // Then assertEquals( @@ -227,17 +282,57 @@ class DiscoveryQueueTest { // When try { - fixture.worker.flush() + fixture.worker.persistIfReady() fail("Expected cancellation") } catch (_: CancellationException) { // Expected. The worker must retain the batch. } - fixture.worker.flush() + assertEquals(2.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(2.0, fixture.gauge("transactions.discovery.in.flight")) + fixture.worker.persistIfReady() // Then val expected = events.map { it.transaction.hash } assertEquals(listOf(expected, expected), attempts) assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(0.0, fixture.gauge("transactions.discovery.in.flight")) + } + + @Test + fun `clear releases a retry batch and buffered rows for rediscovery`() = + runTest { + // Given + var failNext = true + val service = mockk() + coEvery { service.save(any()) } answers { + if (failNext) { + failNext = false + error("simulated MySQL failure") + } + firstArg>().size.toLong() + } + val fixture = fixture(service, properties(capacity = 3, batchSize = 2)) + val events = listOf(event(40), event(41), event(42)) + events.take(2).forEach { fixture.queue.queue(it, DiscoverySource.GAP) } + fixture.worker.persistIfReady() + fixture.queue.queue(events.last(), DiscoverySource.GAP) + assertEquals(3.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(2.0, fixture.gauge("transactions.discovery.in.flight")) + + // When + fixture.worker.clear() + + // Then + assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(0.0, fixture.gauge("transactions.discovery.in.flight")) + assertEquals(3, fixture.queue.remainingTargetCapacity()) + + // When + assertTrue(fixture.queue.queue(events.first(), DiscoverySource.GAP)) + fixture.worker.persistIfReady() + + // Then + assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) } @Test @@ -278,7 +373,7 @@ class DiscoveryQueueTest { // When val first = fixture.queue.queue(event, DiscoverySource.SEND) val pendingDuplicate = fixture.queue.queue(event, DiscoverySource.SEND) - fixture.worker.flush() + fixture.worker.persistIfReady() val committedDuplicate = fixture.queue.queue(event, DiscoverySource.SEND) // Then @@ -322,9 +417,9 @@ class DiscoveryQueueTest { assertEquals(1.0, fixture.gauge("transactions.discovery.at.capacity")) // When - fixture.worker.flush() + fixture.worker.persistIfReady() suspended.await() - fixture.worker.flush() + fixture.worker.persistIfReady() // Then assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) @@ -334,6 +429,113 @@ class DiscoveryQueueTest { assertTrue(queueWait.max(java.util.concurrent.TimeUnit.SECONDS) >= 5.0) } + @Test + fun `empty flush does not attempt persistence`() = + runTest { + // Given + val service = mockk(relaxed = true) + val fixture = fixture(service) + + // When + fixture.worker.persistIfReady() + + // Then + coVerify(exactly = 0) { service.save(any()) } + } + + @Test + fun `outstanding capacity includes the batch until its commit is acknowledged`() = + runTest { + // Given + val saveStarted = CompletableDeferred() + val releaseSave = CompletableDeferred() + val service = mockk() + coEvery { service.save(any()) } coAnswers { + saveStarted.complete(Unit) + releaseSave.await() + firstArg>().size.toLong() + } + val fixture = fixture(service, properties(capacity = 2, batchSize = 2)) + fixture.queue.queue(event(15), DiscoverySource.GAP) + fixture.queue.queue(event(16), DiscoverySource.GAP) + + // When + val flush = async { fixture.worker.persistIfReady() } + saveStarted.await() + + // Then + assertTrue(fixture.queue.isAtCapacity()) + assertEquals(0, fixture.queue.remainingTargetCapacity()) + assertEquals(2.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(2.0, fixture.gauge("transactions.discovery.in.flight")) + + // When + releaseSave.complete(Unit) + flush.await() + + // Then + assertFalse(fixture.queue.isAtCapacity()) + assertEquals(2, fixture.queue.remainingTargetCapacity()) + assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) + assertEquals(0.0, fixture.gauge("transactions.discovery.in.flight")) + } + + @Test + fun `configured batches preserve FIFO order across scheduled passes`() = + runTest { + // Given + val attempts = mutableListOf>() + val service = mockk() + coEvery { service.save(any()) } answers { + firstArg>() + .also { batch -> attempts += batch.map { it.hash } } + .size + .toLong() + } + val fixture = fixture(service, properties(capacity = 5, batchSize = 2)) + val events = (17..21).map { event(it.toByte()) } + events.forEach { fixture.queue.queue(it, DiscoverySource.GAP) } + + // When + fixture.worker.persistIfReady() + fixture.worker.persistIfReady() + fixture.worker.persistIfReady() + + // Then + assertEquals( + listOf( + events.take(2).map { it.transaction.hash }, + events.drop(2).take(2).map { it.transaction.hash }, + listOf(events.last().transaction.hash), + ), + attempts, + ) + assertEquals(0.0, fixture.gauge("transactions.discovery.backlog.depth")) + } + + @Test + fun `successful persistence marks work changed once only when rows were affected`() = + runTest { + // Given + val service = mockk() + coEvery { service.save(any()) } returnsMany listOf(1L, 0L) + val fixture = fixture(service, properties(capacity = 2, batchSize = 1)) + fixture.queue.queue(event(22), DiscoverySource.SEND) + fixture.queue.queue(event(23), DiscoverySource.SEND) + + // When + fixture.worker.persistIfReady() + + // Then + assertEquals(1L, fixture.workTracker.currentGeneration()) + + // When + fixture.worker.persistIfReady() + + // Then + assertEquals(1L, fixture.workTracker.currentGeneration()) + } + private fun fixture( service: UncheckedTransactionService, properties: DiscoveryProperties = properties(), @@ -343,9 +545,12 @@ class DiscoveryQueueTest { ): Fixture { properties.validate() val metrics = DiscoveryMetrics(registry) + val discoveryCapacity = AtomicInteger(properties.capacity) + val pressureMonitor = pressureMonitor(discoveryCapacity) val queue = - DiscoveryQueue(eventPublisher, properties, metrics, clock) + DiscoveryQueue(eventPublisher, properties, metrics, clock, pressureMonitor) .also { it.start() } + val workTracker = UncheckedWorkTracker() val worker = DiscoveryPersistenceWorker( uncheckedTransactionService = service, @@ -353,10 +558,26 @@ class DiscoveryQueueTest { properties = properties, metrics = metrics, clock = clock, + workTracker = workTracker, ) - return Fixture(queue, worker, registry) + return Fixture( + queue, + worker, + registry, + discoveryCapacity, + workTracker, + ) } + private fun pressureMonitor(discoveryCapacity: AtomicInteger): DiscoveryPressureMonitor = + mockk { + every { + targetCapacity(any()) + } answers { + minOf(discoveryCapacity.get(), firstArg()) + } + } + private fun properties( capacity: Int = 10, headroom: Int = 0, @@ -396,6 +617,8 @@ class DiscoveryQueueTest { val queue: DiscoveryQueue, val worker: DiscoveryPersistenceWorker, val registry: SimpleMeterRegistry, + val discoveryCapacity: AtomicInteger, + val workTracker: UncheckedWorkTracker, ) { fun gauge(name: String): Double = registry.get(name).gauge().value() diff --git a/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryTestController.kt b/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryTestController.kt index 55702b37..c318c986 100644 --- a/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryTestController.kt +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/DiscoveryTestController.kt @@ -1,6 +1,6 @@ package cash.atto.node.bootstrap.discovery -import cash.atto.node.bootstrap.unchecked.UncheckedTransactionProcessorStarter +import cash.atto.node.bootstrap.unchecked.UncheckedTransactionProcessor import io.swagger.v3.oas.annotations.Operation import org.springframework.web.bind.annotation.PostMapping import org.springframework.web.bind.annotation.RequestMapping @@ -12,12 +12,12 @@ class DiscoveryTestController( val discoveryPersistenceWorker: DiscoveryPersistenceWorker, val gapDiscoverer: GapDiscoverer, val lastDiscoverer: LastDiscoverer, - val uncheckedTransactionProcessorStarter: UncheckedTransactionProcessorStarter, + val uncheckedTransactionProcessor: UncheckedTransactionProcessor, ) { @PostMapping("/gap") @Operation(description = "Start processing of gap transaction") suspend fun gap() { - gapDiscoverer.resolve() + discoverGaps() } @PostMapping("/last") @@ -29,16 +29,27 @@ class DiscoveryTestController( @PostMapping("/flush") @Operation(description = "Flush discovered transactions") suspend fun flush() { - discoveryPersistenceWorker.flush() + flushAll() } @PostMapping("/settle") @Operation(description = "Flush discoveries, resolve gaps, and process unchecked transactions") suspend fun settle() { - discoveryPersistenceWorker.flush() - uncheckedTransactionProcessorStarter.process() - gapDiscoverer.resolve() - discoveryPersistenceWorker.flush() - uncheckedTransactionProcessorStarter.process() + flushAll() + uncheckedTransactionProcessor.processIfDue() + discoverGaps() + flushAll() + uncheckedTransactionProcessor.processIfDue() + } + + private suspend fun discoverGaps() { + gapDiscoverer.maintainSessions() + gapDiscoverer.discoverIfDue() + } + + private suspend fun flushAll() { + while (discoveryPersistenceWorker.persistIfReady() == DiscoveryPersistenceResult.Persisted) { + // Test-only endpoint: preserve the original deterministic full-drain behavior. + } } } diff --git a/src/test/kotlin/cash/atto/node/bootstrap/discovery/GapDiscovererTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/discovery/GapDiscovererTest.kt index c463c1ff..3b368ee7 100644 --- a/src/test/kotlin/cash/atto/node/bootstrap/discovery/GapDiscovererTest.kt +++ b/src/test/kotlin/cash/atto/node/bootstrap/discovery/GapDiscovererTest.kt @@ -57,7 +57,7 @@ class GapDiscovererTest { ) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then coVerify(exactly = 0) { fixture.repository.findGaps(any()) } @@ -65,18 +65,18 @@ class GapDiscovererTest { } @Test - fun `insufficient full request margin prevents the gap query`() = + fun `adaptive target below the 250 transaction request budget prevents the gap query`() = runTest { // given val fixture = fixture( gaps = listOf(gap(publicKey(1), 1U, 1U, hash(1))), - capacity = 100, - remainingCapacity = 99, + capacity = 250, + remainingCapacity = 249, ) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then coVerify(exactly = 0) { fixture.repository.findGaps(any()) } @@ -91,8 +91,8 @@ class GapDiscovererTest { val fixture = fixture(gaps = gaps, capacity = 3_000) // when - fixture.discoverer.resolve() - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() + fixture.discoverer.discoverIfDue() // then assertEquals(3, fixture.requests.size) @@ -113,7 +113,7 @@ class GapDiscovererTest { ) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then assertEquals(1, fixture.requests.size) @@ -130,7 +130,7 @@ class GapDiscovererTest { val fixture = fixture(gaps = gaps, capacity = 3_000, nodes = nodes) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then assertEquals(3, fixture.requests.size) @@ -151,8 +151,8 @@ class GapDiscovererTest { ) // when - fixture.discoverer.resolve() - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() + fixture.discoverer.discoverIfDue() // then assertEquals( @@ -173,7 +173,7 @@ class GapDiscovererTest { ) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then val request = fixture.requests.single() @@ -192,7 +192,7 @@ class GapDiscovererTest { ) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then val request = fixture.requests.single() @@ -213,11 +213,11 @@ class GapDiscovererTest { capacity = 1_000, queueResult = false, ) - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // when fixture.discoverer.process(fixture.response(transaction)) - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then assertEquals(2, fixture.requests.size) @@ -231,7 +231,7 @@ class GapDiscovererTest { // given val gaps = (1..3).map { gap(publicKey(it.toByte()), 1U, 2U, hash(it.toByte())) } val fixture = fixture(gaps = gaps, capacity = 3_000) - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // when fixture.disconnect() @@ -241,7 +241,7 @@ class GapDiscovererTest { // when fixture.connect() - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then assertEquals(6, fixture.requests.size) @@ -255,11 +255,12 @@ class GapDiscovererTest { val clock = MutableClock() val gaps = (1..2).map { gap(publicKey(it.toByte()), 1U, 2U, hash(it.toByte())) } val fixture = fixture(gaps = gaps, capacity = 2_000, clock = clock) - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // when clock.advance(Duration.ofSeconds(61)) - fixture.discoverer.resolve() + fixture.discoverer.maintainSessions() + fixture.discoverer.discoverIfDue() // then assertEquals(4, fixture.requests.size) @@ -280,7 +281,7 @@ class GapDiscovererTest { // when try { - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() fail("Expected request publication to fail") } catch (_: IllegalStateException) { // expected @@ -290,7 +291,7 @@ class GapDiscovererTest { assertEquals(0, fixture.discoverer.activeSessionCount()) // when - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then assertEquals(1, fixture.requests.size) @@ -303,7 +304,7 @@ class GapDiscovererTest { // given val gaps = (1..3).map { gap(publicKey(it.toByte()), 1U, 1U, hash(it.toByte())) } val fixture = fixture(gaps = gaps, capacity = 3_000) - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // when fixture.discoverer.clear() @@ -320,22 +321,22 @@ class GapDiscovererTest { val fixture = fixture(gaps = emptyList(), clock = clock) // when - fixture.discoverer.resolve() - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() + fixture.discoverer.discoverIfDue() // then coVerify(exactly = 1) { fixture.repository.findGaps(any()) } // when fixture.workTracker.markChanged() - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then coVerify(exactly = 2) { fixture.repository.findGaps(any()) } // when clock.advance(Duration.ofSeconds(31)) - fixture.discoverer.resolve() + fixture.discoverer.discoverIfDue() // then coVerify(exactly = 3) { fixture.repository.findGaps(any()) } diff --git a/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarterTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarterTest.kt deleted file mode 100644 index 632f4bf3..00000000 --- a/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorStarterTest.kt +++ /dev/null @@ -1,113 +0,0 @@ -package cash.atto.node.bootstrap.unchecked - -import cash.atto.node.bootstrap.discovery.DiscoveryProperties -import io.micrometer.core.instrument.simple.SimpleMeterRegistry -import io.mockk.coEvery -import io.mockk.coVerify -import io.mockk.mockk -import kotlinx.coroutines.flow.emptyFlow -import kotlinx.coroutines.test.runTest -import org.junit.jupiter.api.Test -import java.time.Clock -import java.time.Duration -import java.time.Instant -import java.time.ZoneId -import java.time.ZoneOffset - -class UncheckedTransactionProcessorStarterTest { - @Test - fun `skips repeated scans until work changes or maintenance is due`() = - runTest { - // given - val repository = mockk() - val processor = mockk(relaxed = true) - val service = mockk() - val workTracker = UncheckedWorkTracker() - val clock = MutableClock() - val starter = newStarter(repository, processor, service, workTracker, clock) - - coEvery { repository.findTopOldest(1_000L) } returns emptyFlow() - coEvery { service.cleanUp(1_000L) } returns 0 - - // when - starter.process() - starter.process() - - // then - coVerify(exactly = 1) { repository.findTopOldest(1_000L) } - coVerify(exactly = 1) { service.cleanUp(1_000L) } - - // when - workTracker.markChanged() - starter.process() - - // then - coVerify(exactly = 2) { repository.findTopOldest(1_000L) } - - // when - starter.process() - clock.advance(Duration.ofSeconds(30)) - starter.process() - - // then - coVerify(exactly = 3) { repository.findTopOldest(1_000L) } - coVerify(exactly = 3) { service.cleanUp(1_000L) } - } - - @Test - fun `bounds cleanup work per scheduled pass`() = - runTest { - // given - val repository = mockk() - val processor = mockk(relaxed = true) - val service = mockk() - val starter = - newStarter( - repository, - processor, - service, - UncheckedWorkTracker(), - MutableClock(), - ) - - coEvery { repository.findTopOldest(1_000L) } returns emptyFlow() - coEvery { service.cleanUp(1_000L) } returns 1_000 - - // when - starter.process() - - // then - coVerify(exactly = 10) { service.cleanUp(1_000L) } - } - - private fun newStarter( - repository: UncheckedTransactionRepository, - processor: UncheckedTransactionProcessor, - service: UncheckedTransactionService, - workTracker: UncheckedWorkTracker, - clock: Clock, - ): UncheckedTransactionProcessorStarter = - UncheckedTransactionProcessorStarter( - uncheckedTransactionRepository = repository, - processor = processor, - uncheckedTransactionService = service, - workTracker = workTracker, - discoveryProperties = DiscoveryProperties(), - meterRegistry = SimpleMeterRegistry(), - clock = clock, - ) - - private class MutableClock( - private var current: Instant = Instant.parse("2026-08-03T00:00:00Z"), - ) : Clock() { - override fun getZone(): ZoneId = ZoneOffset.UTC - - override fun withZone(zone: ZoneId): Clock = this - - override fun instant(): Instant = current - - fun advance(duration: Duration) { - current = current.plus(duration) - } - } -} diff --git a/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorTest.kt new file mode 100644 index 00000000..954ead93 --- /dev/null +++ b/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionProcessorTest.kt @@ -0,0 +1,346 @@ +package cash.atto.node.bootstrap.unchecked + +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.AttoWork +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.account.AccountService +import cash.atto.node.bootstrap.TransactionResolved +import cash.atto.node.bootstrap.discovery.DiscoveryProperties +import cash.atto.node.transaction.Transaction +import cash.atto.node.transaction.TransactionSource +import cash.atto.node.transaction.validation.TransactionValidationManager +import io.micrometer.core.instrument.simple.SimpleMeterRegistry +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.fail +import org.junit.jupiter.api.Test +import org.springframework.transaction.ReactiveTransaction +import org.springframework.transaction.ReactiveTransactionManager +import org.springframework.transaction.TransactionDefinition +import reactor.core.publisher.Mono +import java.time.Clock +import java.time.Duration +import java.time.Instant +import java.time.ZoneId +import java.time.ZoneOffset + +class UncheckedTransactionProcessorTest { + @Test + fun `skips repeated scans until work changes or maintenance is due`() = + runTest { + // Given + val fixture = fixture() + coEvery { fixture.repository.findTopOldest(1_000L) } returns emptyFlow() + coEvery { fixture.service.cleanUp(1_000L) } returns 0 + + // When + val first = fixture.processor.processIfDue() + val suppressed = fixture.processor.processIfDue() + + // Then + assertEquals(UncheckedProcessingResult.Completed(0, 0, 0), first) + assertEquals(UncheckedProcessingResult.SkippedIdle, suppressed) + coVerify(exactly = 1) { fixture.repository.findTopOldest(1_000L) } + coVerify(exactly = 1) { fixture.service.cleanUp(1_000L) } + + // When + fixture.workTracker.markChanged() + fixture.processor.processIfDue() + fixture.processor.processIfDue() + fixture.clock.advance(Duration.ofSeconds(30)) + fixture.processor.processIfDue() + + // Then + coVerify(exactly = 3) { fixture.repository.findTopOldest(1_000L) } + coVerify(exactly = 3) { fixture.service.cleanUp(1_000L) } + } + + @Test + fun `cleanup progress keeps maintenance due for the following pass`() = + runTest { + // Given + val fixture = fixture() + coEvery { fixture.repository.findTopOldest(1_000L) } returns emptyFlow() + coEvery { fixture.service.cleanUp(1_000L) } returnsMany listOf(1_000, 0) + + // When + val cleanupProgress = fixture.processor.processIfDue() + val completedCleanup = fixture.processor.processIfDue() + val suppressed = fixture.processor.processIfDue() + + // Then + assertEquals(UncheckedProcessingResult.Completed(0, 0, 1_000), cleanupProgress) + assertEquals(UncheckedProcessingResult.Completed(0, 0, 0), completedCleanup) + assertEquals(UncheckedProcessingResult.SkippedIdle, suppressed) + coVerify(exactly = 2) { fixture.service.cleanUp(1_000L) } + assertEquals( + 1_000.0, + fixture.registry + .get("transactions.unchecked.cleanup.deleted") + .counter() + .count(), + ) + } + + @Test + fun `selects before the resolution transaction and cleans up after commit`() = + runTest { + // Given + val timeline = mutableListOf() + val fixture = fixture(timeline) + val unchecked = mockk() + val transaction = Transaction.sample() + val account = mockk() + + every { unchecked.toTransaction() } returns transaction + every { account.algorithm } returns transaction.algorithm + coEvery { fixture.repository.findTopOldest(1_000L) } returns + flow { + timeline += "select" + emit(unchecked) + } + coEvery { fixture.accountRepository.findById(transaction.publicKey) } returns account + coEvery { fixture.validationManager.validate(account, transaction) } returns null + coEvery { + fixture.accountService.add(TransactionSource.BOOTSTRAP, listOf(transaction)) + } returns listOf(account) + coEvery { fixture.service.cleanUp(1_000L) } answers { + timeline += "cleanup" + 1 + } + + // When + val result = fixture.processor.processIfDue() + + // Then + assertEquals(UncheckedProcessingResult.Completed(1, 1, 1), result) + assertEquals(listOf("select", "begin", "commit", "cleanup"), timeline) + assertEquals(TransactionDefinition.ISOLATION_READ_COMMITTED, fixture.transactionManager.isolationLevel) + assertEquals(1, fixture.transactionManager.commits) + assertEquals(0, fixture.transactionManager.rollbacks) + coVerify(exactly = 1) { + fixture.eventPublisher.publishAfterCommit( + match { it.transaction == transaction }, + ) + } + } + + @Test + fun `resolution failure rolls back skips cleanup and releases the pass mutex`() = + runTest { + // Given + val fixture = fixture() + val unchecked = mockk() + val transaction = Transaction.sample() + val account = mockk() + var validationAttempts = 0 + + every { unchecked.toTransaction() } returns transaction + every { account.algorithm } returns transaction.algorithm + coEvery { fixture.repository.findTopOldest(1_000L) } returns flowOf(unchecked) + coEvery { fixture.accountRepository.findById(transaction.publicKey) } returns account + coEvery { fixture.validationManager.validate(account, transaction) } coAnswers { + validationAttempts++ + if (validationAttempts == 1) { + error("simulated resolution failure") + } + null + } + coEvery { + fixture.accountService.add(TransactionSource.BOOTSTRAP, listOf(transaction)) + } returns listOf(account) + coEvery { fixture.service.cleanUp(1_000L) } returns 0 + + // When + try { + fixture.processor.processIfDue() + fail("Expected resolution failure") + } catch (_: IllegalStateException) { + // Expected. + } + + // Then + assertEquals(0, fixture.transactionManager.commits) + assertEquals(1, fixture.transactionManager.rollbacks) + coVerify(exactly = 0) { fixture.service.cleanUp(any()) } + + // When + val retry = fixture.processor.processIfDue() + + // Then + assertEquals(UncheckedProcessingResult.Completed(1, 1, 0), retry) + assertEquals(1, fixture.transactionManager.commits) + assertEquals(1, fixture.transactionManager.rollbacks) + coVerify(exactly = 1) { fixture.service.cleanUp(1_000L) } + } + + @Test + fun `overlapping invocation skips instead of queuing another pass`() = + runTest { + // Given + val fixture = fixture() + val selectionStarted = CompletableDeferred() + val releaseSelection = CompletableDeferred() + coEvery { fixture.repository.findTopOldest(1_000L) } returns + flow { + selectionStarted.complete(Unit) + releaseSelection.await() + } + coEvery { fixture.service.cleanUp(1_000L) } returns 0 + + // When + val firstPass = async { fixture.processor.processIfDue() } + selectionStarted.await() + val overlap = fixture.processor.processIfDue() + + // Then + assertEquals(UncheckedProcessingResult.SkippedBusy, overlap) + coVerify(exactly = 1) { fixture.repository.findTopOldest(1_000L) } + coVerify(exactly = 0) { fixture.service.cleanUp(any()) } + + // When + releaseSelection.complete(Unit) + val completed = firstPass.await() + + // Then + assertEquals(UncheckedProcessingResult.Completed(0, 0, 0), completed) + coVerify(exactly = 1) { fixture.service.cleanUp(1_000L) } + } + + private fun fixture(timeline: MutableList = mutableListOf()): Fixture { + val repository = mockk() + val service = mockk() + val workTracker = UncheckedWorkTracker() + val clock = MutableClock() + val accountRepository = mockk() + val validationManager = mockk() + val accountService = mockk() + val eventPublisher = mockk(relaxed = true) + val transactionManager = RecordingReactiveTransactionManager(timeline) + val registry = SimpleMeterRegistry() + val processor = + UncheckedTransactionProcessor( + accountRepository = accountRepository, + transactionValidationManager = validationManager, + accountService = accountService, + eventPublisher = eventPublisher, + uncheckedTransactionRepository = repository, + uncheckedTransactionService = service, + workTracker = workTracker, + discoveryProperties = DiscoveryProperties(), + meterRegistry = registry, + transactionManager = transactionManager, + clock = clock, + ) + return Fixture( + processor = processor, + repository = repository, + service = service, + workTracker = workTracker, + clock = clock, + accountRepository = accountRepository, + validationManager = validationManager, + accountService = accountService, + eventPublisher = eventPublisher, + transactionManager = transactionManager, + registry = registry, + ) + } + + private data class Fixture( + val processor: UncheckedTransactionProcessor, + val repository: UncheckedTransactionRepository, + val service: UncheckedTransactionService, + val workTracker: UncheckedWorkTracker, + val clock: MutableClock, + val accountRepository: AccountRepository, + val validationManager: TransactionValidationManager, + val accountService: AccountService, + val eventPublisher: EventPublisher, + val transactionManager: RecordingReactiveTransactionManager, + val registry: SimpleMeterRegistry, + ) + + private class RecordingReactiveTransactionManager( + private val timeline: MutableList, + ) : ReactiveTransactionManager { + var commits = 0 + private set + var rollbacks = 0 + private set + var isolationLevel = TransactionDefinition.ISOLATION_DEFAULT + private set + + override fun getReactiveTransaction(definition: TransactionDefinition?): Mono { + isolationLevel = definition?.isolationLevel ?: TransactionDefinition.ISOLATION_DEFAULT + timeline += "begin" + return Mono.just(SimpleReactiveTransaction) + } + + override fun commit(transaction: ReactiveTransaction): Mono = + Mono.fromRunnable { + commits++ + timeline += "commit" + } + + override fun rollback(transaction: ReactiveTransaction): Mono = + Mono.fromRunnable { + rollbacks++ + timeline += "rollback" + } + } + + private object SimpleReactiveTransaction : ReactiveTransaction + + private class MutableClock( + private var current: Instant = Instant.parse("2026-08-03T00:00:00Z"), + ) : Clock() { + override fun getZone(): ZoneId = ZoneOffset.UTC + + override fun withZone(zone: ZoneId): Clock = this + + override fun instant(): Instant = current + + fun advance(duration: Duration) { + current = current.plus(duration) + } + } + + private fun Transaction.Companion.sample(): Transaction = + Transaction( + block = + AttoReceiveBlock( + version = 0U.toAttoVersion(), + network = AttoNetwork.LOCAL, + algorithm = AttoAlgorithm.V1, + publicKey = AttoPublicKey(ByteArray(32) { 1 }), + height = cash.atto.commons.AttoHeight(2UL), + balance = AttoAmount.MAX, + timestamp = AttoInstant.now(), + previous = AttoHash(ByteArray(32) { 2 }), + sendHashAlgorithm = AttoAlgorithm.V1, + sendHash = AttoHash(ByteArray(32) { 3 }), + ), + signature = AttoSignature(ByteArray(64) { 4 }), + work = AttoWork(ByteArray(8) { 5 }), + ) +} diff --git a/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionServiceTest.kt b/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionServiceTest.kt index fc533719..d99d435a 100644 --- a/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionServiceTest.kt +++ b/src/test/kotlin/cash/atto/node/bootstrap/unchecked/UncheckedTransactionServiceTest.kt @@ -1,6 +1,5 @@ package cash.atto.node.bootstrap.unchecked -import cash.atto.node.EventPublisher import io.mockk.coEvery import io.mockk.coVerify import io.mockk.mockk @@ -15,8 +14,7 @@ class UncheckedTransactionServiceTest { // given val repository = mockk() val inserter = mockk() - val eventPublisher = mockk() - val service = UncheckedTransactionService(repository, inserter, eventPublisher) + val service = UncheckedTransactionService(repository, inserter) coEvery { repository.deleteExistingInTransaction(1_000L) } returns 37 // when diff --git a/src/test/kotlin/cash/atto/node/unchecked/UncheckedTransactionController.kt b/src/test/kotlin/cash/atto/node/unchecked/UncheckedTransactionController.kt index 62ccc82c..23046ce3 100644 --- a/src/test/kotlin/cash/atto/node/unchecked/UncheckedTransactionController.kt +++ b/src/test/kotlin/cash/atto/node/unchecked/UncheckedTransactionController.kt @@ -1,7 +1,7 @@ package cash.atto.node.unchecked import cash.atto.commons.AttoTransaction -import cash.atto.node.bootstrap.unchecked.UncheckedTransactionProcessorStarter +import cash.atto.node.bootstrap.unchecked.UncheckedTransactionProcessor import cash.atto.node.bootstrap.unchecked.UncheckedTransactionRepository import io.swagger.v3.oas.annotations.Operation import kotlinx.coroutines.flow.Flow @@ -16,13 +16,13 @@ import org.springframework.web.bind.annotation.RestController @RequestMapping("/unchecked-transactions") @Profile("default") class UncheckedTransactionController( - val uncheckedTransactionProcessorStarter: UncheckedTransactionProcessorStarter, + val uncheckedTransactionProcessor: UncheckedTransactionProcessor, val uncheckedTransactionRepository: UncheckedTransactionRepository, ) { @PostMapping @Operation(description = "Process unchecked transactions") suspend fun process() { - uncheckedTransactionProcessorStarter.process() + uncheckedTransactionProcessor.processIfDue() } @GetMapping