fix(client): make ReconnectingOutbox KMP-native-safe (@Volatile→AtomicReference, Math.pow→kotlin.math.pow)
CI red on tag 10 release #1983: client:compileCommonMainKotlinMetadata and :compileKotlinLinuxArm64 both failed with: e: ReconnectingOutbox.kt:154 Unresolved reference 'Volatile' e: ReconnectingOutbox.kt:227 Unresolved reference 'Math' @Volatile is JVM-only annotation; java.lang.Math is JVM-only API. On linuxArm64/macosArm64 they don't resolve. Fix: - @Volatile private var lastSeen: Instant? → AtomicReference<Instant?> (kotlin.concurrent.atomics, same module as the AtomicBoolean already used for ). .load() / .store() / @OptIn(ExperimentalAtomicApi::class). - Math.pow(m, e) → m.pow(e) via kotlin.math.pow import. commonMain stays KMP-clean; jvmTest green (95 tasks); linuxX64 / linuxArm64 / mingwX64 / macosX64 / macosArm64 compile green.
This commit is contained in:
@@ -13,8 +13,10 @@ import kotlinx.coroutines.launch
|
|||||||
import pw.binom.agentik.outbox.CommonEvent
|
import pw.binom.agentik.outbox.CommonEvent
|
||||||
import pw.binom.agentik.outbox.OutboxStore
|
import pw.binom.agentik.outbox.OutboxStore
|
||||||
import kotlin.concurrent.atomics.AtomicBoolean
|
import kotlin.concurrent.atomics.AtomicBoolean
|
||||||
|
import kotlin.concurrent.atomics.AtomicReference
|
||||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||||
import kotlin.math.min
|
import kotlin.math.min
|
||||||
|
import kotlin.math.pow
|
||||||
import kotlin.random.Random
|
import kotlin.random.Random
|
||||||
import kotlin.time.Duration
|
import kotlin.time.Duration
|
||||||
import kotlin.time.Duration.Companion.seconds
|
import kotlin.time.Duration.Companion.seconds
|
||||||
@@ -151,8 +153,8 @@ class ReconnectingOutbox(
|
|||||||
private val started = AtomicBoolean(false)
|
private val started = AtomicBoolean(false)
|
||||||
private var job: Job? = null
|
private var job: Job? = null
|
||||||
|
|
||||||
@Volatile
|
@OptIn(ExperimentalAtomicApi::class)
|
||||||
private var lastSeen: Instant? = null
|
private val lastSeen: AtomicReference<Instant?> = AtomicReference(null)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Live-события из [outbox] с авто-reconnect. [after] — начальный курсор;
|
* Live-события из [outbox] с авто-reconnect. [after] — начальный курсор;
|
||||||
@@ -183,10 +185,11 @@ class ReconnectingOutbox(
|
|||||||
@OptIn(ExperimentalAtomicApi::class)
|
@OptIn(ExperimentalAtomicApi::class)
|
||||||
private fun ensureStarted(initialCursor: Instant?) {
|
private fun ensureStarted(initialCursor: Instant?) {
|
||||||
if (!started.compareAndSet(false, true)) return
|
if (!started.compareAndSet(false, true)) return
|
||||||
lastSeen = initialCursor
|
lastSeen.store(initialCursor)
|
||||||
job = scope.launch { runLoop() }
|
job = scope.launch { runLoop() }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@OptIn(ExperimentalAtomicApi::class)
|
||||||
private suspend fun runLoop() {
|
private suspend fun runLoop() {
|
||||||
var attempt = 0
|
var attempt = 0
|
||||||
var connected = false
|
var connected = false
|
||||||
@@ -194,12 +197,12 @@ class ReconnectingOutbox(
|
|||||||
attempt++
|
attempt++
|
||||||
_status.emit(ConnectionStatus.Connecting(attempt))
|
_status.emit(ConnectionStatus.Connecting(attempt))
|
||||||
val error: Throwable? = try {
|
val error: Throwable? = try {
|
||||||
outbox.events(after = lastSeen).collect { event ->
|
outbox.events(after = lastSeen.load()).collect { event ->
|
||||||
lastSeen = event.date
|
lastSeen.store(event.date)
|
||||||
_events.emit(event)
|
_events.emit(event)
|
||||||
if (!connected) {
|
if (!connected) {
|
||||||
connected = true
|
connected = true
|
||||||
_status.emit(ConnectionStatus.Connected(lastSeen!!))
|
_status.emit(ConnectionStatus.Connected(event.date))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
null
|
null
|
||||||
@@ -224,7 +227,7 @@ class ReconnectingOutbox(
|
|||||||
private fun computeBackoff(attempt: Int): Duration {
|
private fun computeBackoff(attempt: Int): Duration {
|
||||||
// attempt 1 → initial, 2 → initial * m, 3 → initial * m^2, ...
|
// attempt 1 → initial, 2 → initial * m, 3 → initial * m^2, ...
|
||||||
val base = (policy.initial.inWholeMilliseconds.toDouble() *
|
val base = (policy.initial.inWholeMilliseconds.toDouble() *
|
||||||
Math.pow(policy.multiplier, (attempt - 1).toDouble()))
|
policy.multiplier.pow((attempt - 1).toDouble()))
|
||||||
.toLong()
|
.toLong()
|
||||||
val capped = min(base, policy.max.inWholeMilliseconds)
|
val capped = min(base, policy.max.inWholeMilliseconds)
|
||||||
val jitterMs = (capped * policy.jitter * random.nextDouble()).toLong()
|
val jitterMs = (capped * policy.jitter * random.nextDouble()).toLong()
|
||||||
|
|||||||
Reference in New Issue
Block a user