fix(quartz): make IngestQueue writer-start KMP-safe via AtomicBoolean
Replaces JVM-only @Volatile + synchronized double-checked locking with kotlin.concurrent.atomics.AtomicBoolean.compareAndSet so the file builds on all commonMain targets, mirroring the pattern in BasicRelayClient. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
+5
-7
@@ -33,6 +33,8 @@ import kotlinx.coroutines.channels.Channel
|
|||||||
import kotlinx.coroutines.channels.ClosedReceiveChannelException
|
import kotlinx.coroutines.channels.ClosedReceiveChannelException
|
||||||
import kotlinx.coroutines.coroutineScope
|
import kotlinx.coroutines.coroutineScope
|
||||||
import kotlinx.coroutines.launch
|
import kotlinx.coroutines.launch
|
||||||
|
import kotlin.concurrent.atomics.AtomicBoolean
|
||||||
|
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||||
import kotlin.coroutines.CoroutineContext
|
import kotlin.coroutines.CoroutineContext
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -73,6 +75,7 @@ import kotlin.coroutines.CoroutineContext
|
|||||||
* propagates back through the WebSocket pump so a slow disk
|
* propagates back through the WebSocket pump so a slow disk
|
||||||
* eventually slows the publisher rather than ballooning JVM memory.
|
* eventually slows the publisher rather than ballooning JVM memory.
|
||||||
*/
|
*/
|
||||||
|
@OptIn(ExperimentalAtomicApi::class)
|
||||||
class IngestQueue(
|
class IngestQueue(
|
||||||
private val store: IEventStore,
|
private val store: IEventStore,
|
||||||
parentContext: CoroutineContext,
|
parentContext: CoroutineContext,
|
||||||
@@ -120,9 +123,7 @@ class IngestQueue(
|
|||||||
* keeps relays that never see an EVENT (read-only sessions,
|
* keeps relays that never see an EVENT (read-only sessions,
|
||||||
* negentropy-only) from paying for the writer at all.
|
* negentropy-only) from paying for the writer at all.
|
||||||
*/
|
*/
|
||||||
@Volatile
|
private val writerStarted = AtomicBoolean(false)
|
||||||
private var writerStarted = false
|
|
||||||
private val startLock = Any()
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Hand off [event] for insertion. [onComplete] is invoked once
|
* Hand off [event] for insertion. [onComplete] is invoked once
|
||||||
@@ -144,11 +145,8 @@ class IngestQueue(
|
|||||||
}
|
}
|
||||||
|
|
||||||
private fun ensureWriterStarted() {
|
private fun ensureWriterStarted() {
|
||||||
if (writerStarted) return
|
if (writerStarted.compareAndSet(expectedValue = false, newValue = true)) {
|
||||||
synchronized(startLock) {
|
|
||||||
if (writerStarted) return
|
|
||||||
scope.launch { drainLoop() }
|
scope.launch { drainLoop() }
|
||||||
writerStarted = true
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user