Merge pull request #3459 from vitorpamplona/claude/dispatchers-thread-caps-s8yp4c

Add lock-free concurrent collections and fix UDP socket threading

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/search/EventDeduplicator.kt

@@ -20,18 +20,23 @@
  */
 package com.vitorpamplona.amethyst.commons.search
 
-import com.vitorpamplona.amethyst.commons.util.KmpLock
-import com.vitorpamplona.amethyst.commons.util.withLock
+import com.vitorpamplona.amethyst.commons.util.ConcurrentSet
 
+/**
+ * Tracks event ids already seen so duplicate deliveries can be dropped.
+ *
+ * Fed from relay subscription callbacks, which can arrive on multiple threads
+ * concurrently, so this uses a [ConcurrentSet] (lock-striped on JVM/Android)
+ * rather than a single lock that would serialize every delivery.
+ */
 class EventDeduplicator {
-    private val lock = KmpLock()
-    private val seenIds = mutableSetOf<String>()
+    private val seenIds = ConcurrentSet<String>()
 
-    fun tryAdd(id: String): Boolean = lock.withLock { seenIds.add(id) }
+    fun tryAdd(id: String): Boolean = seenIds.add(id)
 
-    fun contains(id: String): Boolean = lock.withLock { id in seenIds }
+    fun contains(id: String): Boolean = seenIds.contains(id)
 
-    fun clear() = lock.withLock { seenIds.clear() }
+    fun clear() = seenIds.clear()
 
-    val size: Int get() = lock.withLock { seenIds.size }
+    val size: Int get() = seenIds.size
 }

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.kt

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.util
+
+/**
+ * KMP-friendly thread-safe set of non-null elements.
+ *
+ * On JVM/Android this is backed by `ConcurrentHashMap.newKeySet()`, so
+ * concurrent [add] / [contains] / [remove] from many threads do NOT all
+ * serialize on one monitor — writes are lock-striped and reads are lock-free.
+ * That is the difference from wrapping a plain set in a single [KmpLock], where
+ * every operation from every thread contends the same lock.
+ *
+ * iOS has no lock-free set in the standard library, so its actual guards a plain
+ * set with a reentrant lock — same behaviour as before, no regression. The
+ * win is on the JVM/Android app, which is where the high-throughput event
+ * paths run.
+ *
+ * Use this only when the access pattern is genuinely multi-threaded and hot
+ * (e.g. per-event dedup fed by multiple relay callback threads). A set touched
+ * only from one thread does not need it.
+ */
+expect class ConcurrentSet<E : Any>() {
+    /** Adds [element]; returns true if it was not already present. */
+    fun add(element: E): Boolean
+
+    fun contains(element: E): Boolean
+
+    /** Removes [element]; returns true if it was present. */
+    fun remove(element: E): Boolean
+
+    fun clear()
+
+    val size: Int
+}

commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetTest.kt

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.util
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class ConcurrentSetTest {
+    @Test
+    fun `add returns true only the first time`() {
+        val set = ConcurrentSet<String>()
+        assertTrue(set.add("a"))
+        assertFalse(set.add("a"))
+        assertEquals(1, set.size)
+    }
+
+    @Test
+    fun `contains reflects membership`() {
+        val set = ConcurrentSet<String>()
+        assertFalse(set.contains("a"))
+        set.add("a")
+        assertTrue(set.contains("a"))
+    }
+
+    @Test
+    fun `remove returns true only when present`() {
+        val set = ConcurrentSet<String>()
+        set.add("a")
+        assertTrue(set.remove("a"))
+        assertFalse(set.remove("a"))
+        assertFalse(set.contains("a"))
+    }
+
+    @Test
+    fun `clear empties the set`() {
+        val set = ConcurrentSet<String>()
+        set.add("a")
+        set.add("b")
+        set.clear()
+        assertEquals(0, set.size)
+        assertFalse(set.contains("a"))
+    }
+}

commons/src/iosMain/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.ios.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.util
+
+actual class ConcurrentSet<E : Any> {
+    // No lock-free set in the K/N standard library — guard a plain set with the
+    // same reentrant lock KmpLock uses. Same behaviour as the previous
+    // KmpLock-wrapped sets; the lock-striped win only exists on JVM/Android.
+    private val lock = KmpLock()
+    private val set = HashSet<E>()
+
+    actual fun add(element: E): Boolean = lock.withLock { set.add(element) }
+
+    actual fun contains(element: E): Boolean = lock.withLock { set.contains(element) }
+
+    actual fun remove(element: E): Boolean = lock.withLock { set.remove(element) }
+
+    actual fun clear() = lock.withLock { set.clear() }
+
+    actual val size: Int get() = lock.withLock { set.size }
+}

commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSet.jvmAndroid.kt

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.util
+
+import java.util.concurrent.ConcurrentHashMap
+
+actual class ConcurrentSet<E : Any> {
+    // Lock-striped writes, lock-free reads — no single monitor across threads.
+    private val set: MutableSet<E> = ConcurrentHashMap.newKeySet()
+
+    actual fun add(element: E): Boolean = set.add(element)
+
+    actual fun contains(element: E): Boolean = set.contains(element)
+
+    actual fun remove(element: E): Boolean = set.remove(element)
+
+    actual fun clear() = set.clear()
+
+    actual val size: Int get() = set.size
+}

commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/util/ConcurrentSetConcurrencyTest.kt

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.commons.util
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class ConcurrentSetConcurrencyTest {
+    @Test
+    fun `exactly one add per key wins across threads`() {
+        val set = ConcurrentSet<Int>()
+        val keys = 10_000
+        val threads = 8
+        val pool = Executors.newFixedThreadPool(threads)
+        val start = CountDownLatch(1)
+        val done = CountDownLatch(threads)
+        // Every thread races to add the SAME keys; exactly one add per key must
+        // return true. With a single lock this would still hold — the point is
+        // that it also holds under the lock-striped ConcurrentHashMap backing.
+        val wonAdds = AtomicInteger(0)
+
+        repeat(threads) {
+            pool.execute {
+                start.await()
+                var local = 0
+                for (k in 0 until keys) {
+                    if (set.add(k)) local++
+                }
+                wonAdds.addAndGet(local)
+                done.countDown()
+            }
+        }
+        start.countDown()
+        assertTrue(done.await(30, TimeUnit.SECONDS), "workers did not finish in time")
+        pool.shutdown()
+
+        assertEquals(keys, wonAdds.get(), "each key must be added exactly once")
+        assertEquals(keys, set.size)
+    }
+}

desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/service/DesktopCachedRichTextParser.kt

@@ -23,25 +23,23 @@ package com.vitorpamplona.amethyst.desktop.service
 import com.vitorpamplona.amethyst.commons.model.ImmutableListOfLists
 import com.vitorpamplona.amethyst.commons.richtext.RichTextParser
 import com.vitorpamplona.amethyst.commons.richtext.RichTextViewerState
+import com.vitorpamplona.quartz.utils.cache.ConcurrentLruCache
 
 object DesktopCachedRichTextParser {
     private const val MAX_CACHE_SIZE = 50
 
-    private val cache =
-        java.util.Collections.synchronizedMap(
-            object : LinkedHashMap<String, RichTextViewerState>(64, 0.75f, true) {
-                override fun removeEldestEntry(eldest: Map.Entry<String, RichTextViewerState>) = size > MAX_CACHE_SIZE
-            },
-        )
+    // Lock-free get on the feed rich-text render path; the previous access-order
+    // synchronizedMap took a monitor even on reads.
+    private val cache = ConcurrentLruCache<String, RichTextViewerState>(MAX_CACHE_SIZE)
 
     fun parseText(
         content: String,
         tags: ImmutableListOfLists<String>,
         callbackUri: String? = null,
     ): RichTextViewerState {
-        cache[content]?.let { return it }
+        cache.get(content)?.let { return it }
         val state = RichTextParser().parseText(content, tags, callbackUri)
-        cache[content] = state
+        cache.put(content, state)
         return state
     }
 

quartz/src/appleMain/kotlin/com/vitorpamplona/quartz/utils/cache/LargeCache.apple.kt

@@ -22,7 +22,6 @@ package com.vitorpamplona.quartz.utils.cache
 
 import io.github.charlietap.cachemap.CacheMap
 import io.github.charlietap.cachemap.cacheMapOf
-import kotlinx.coroutines.runBlocking
 
 // An implementation of a Threadsafe map, using CacheMap.
 // Investigating a Swift-based alternative(for now)
@@ -70,17 +69,16 @@ actual class LargeCache<K, V> : ICacheOperations<K, V> {
     actual fun createIfAbsent(
         key: K,
         builder: (key: K) -> V,
-    ): Boolean =
-        runBlocking {
-            val value = concurrentMap.get(key)
-            if (value != null) {
-                false
-            } else {
-                val newObject = builder(key)
-                concurrentMap.put(key, newObject)
-                concurrentMap[key] != null
-            }
+    ): Boolean {
+        val value = concurrentMap.get(key)
+        return if (value != null) {
+            false
+        } else {
+            val newObject = builder(key)
+            concurrentMap.put(key, newObject)
+            concurrentMap[key] != null
         }
+    }
 
     actual override fun size(): Int = concurrentMap.size
 

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip57Zaps/validate/LnurlEndpointCache.kt

@@ -20,6 +20,8 @@
  */
 package com.vitorpamplona.quartz.nip57Zaps.validate
 
+import com.vitorpamplona.quartz.utils.cache.ConcurrentLruCache
+
 /**
  * Process-wide cache of LNURL-pay endpoint metadata, keyed by the canonical
  * `/.well-known/lnurlp/<user>` URL the recipient resolves to.
@@ -37,37 +39,23 @@ package com.vitorpamplona.quartz.nip57Zaps.validate
 object LnurlEndpointCache {
     private const val MAX_ENTRIES = 1000
 
-    // Insertion-ordered map so we can evict the oldest entry once we hit the cap.
-    // Synchronized externally — every mutating call holds the monitor.
-    private val cache: LinkedHashMap<String, LnurlEndpointInfo> = LinkedHashMap()
+    // Bounded cache with a lock-free get — hot on the zap-validation read path.
+    // Eviction is least-recently-put (a get does not refresh recency), matching
+    // the previous LinkedHashMap-based behaviour where only put reordered.
+    private val cache = ConcurrentLruCache<String, LnurlEndpointInfo>(MAX_ENTRIES)
 
-    @Synchronized
-    fun get(url: String): LnurlEndpointInfo? = cache[LnurlForm.normalizeUrl(url)]
+    fun get(url: String): LnurlEndpointInfo? = cache.get(LnurlForm.normalizeUrl(url))
 
-    @Synchronized
     fun put(
         url: String,
         info: LnurlEndpointInfo,
     ) {
-        val key = LnurlForm.normalizeUrl(url)
-        // Re-insert so the entry becomes "youngest" in iteration order.
-        cache.remove(key)
-        cache[key] = info
-        if (cache.size > MAX_ENTRIES) {
-            val oldest =
-                cache.entries
-                    .iterator()
-                    .next()
-                    .key
-            cache.remove(oldest)
-        }
+        cache.put(LnurlForm.normalizeUrl(url), info)
     }
 
-    @Synchronized
     fun clear() {
         cache.clear()
     }
 
-    @Synchronized
-    internal fun size(): Int = cache.size
+    internal fun size(): Int = cache.size()
 }

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCache.kt

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.utils.cache
+
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * A bounded, thread-safe cache with a **lock-free [get]**.
+ *
+ * The common alternative — a `Collections.synchronizedMap` wrapping an
+ * access-order `LinkedHashMap`, or an `@Synchronized`-guarded `LinkedHashMap` —
+ * takes a single monitor on *every* operation, including reads (an access-order
+ * map structurally mutates on `get`, so it cannot be read without the lock).
+ * On hot read paths (zap-receipt validation, feed rich-text rendering) that one
+ * monitor serializes every reader across all dispatcher threads.
+ *
+ * Here storage is a [ConcurrentHashMap], so [get] never takes a lock. [put] and
+ * [clear] hold a small monitor that keeps the map and the recency order
+ * consistent; that lock is off the read path entirely. Eviction is
+ * **least-recently-put** order (a `get` does not refresh recency); re-putting a
+ * key moves it to the youngest position. This matches the semantics the previous
+ * monitor-based caches relied on and keeps the read path contention-free.
+ *
+ * Because writes and eviction happen atomically under the monitor while readers
+ * see the [ConcurrentHashMap] directly, an external [size] can transiently
+ * observe at most `maxSize + 1` (the instant a new entry is inserted before the
+ * over-cap entry is evicted, within a single locked section). It settles to
+ * `<= maxSize` once writers quiesce.
+ *
+ * Keys and values must be non-null (a [ConcurrentHashMap] constraint).
+ */
+class ConcurrentLruCache<K : Any, V : Any>(
+    private val maxSize: Int,
+) {
+    init {
+        require(maxSize > 0) { "maxSize must be > 0, was $maxSize" }
+    }
+
+    private val map = ConcurrentHashMap<K, V>()
+
+    // Guards writes + eviction so the map and the recency order stay consistent.
+    // Reads never touch it. Writes are the cold path here, so serializing them
+    // is fine; the point is a lock-free [get].
+    private val writeLock = Any()
+    private val order = ArrayDeque<K>()
+
+    fun get(key: K): V? = map[key]
+
+    fun put(
+        key: K,
+        value: V,
+    ) {
+        synchronized(writeLock) {
+            // Re-inserting an existing key makes it the youngest again.
+            val existed = map.put(key, value) != null
+            if (existed) order.remove(key)
+            order.addLast(key)
+            while (order.size > maxSize) {
+                val oldest = order.removeFirst()
+                map.remove(oldest)
+            }
+        }
+    }
+
+    fun clear() {
+        synchronized(writeLock) {
+            order.clear()
+            map.clear()
+        }
+    }
+
+    fun size(): Int = map.size
+}

quartz/src/jvmAndroidTest/kotlin/com/vitorpamplona/quartz/utils/cache/ConcurrentLruCacheTest.kt

@@ -0,0 +1,136 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.utils.cache
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class ConcurrentLruCacheTest {
+    @Test
+    fun `get returns put value`() {
+        val cache = ConcurrentLruCache<String, Int>(4)
+        cache.put("a", 1)
+        assertEquals(1, cache.get("a"))
+        assertNull(cache.get("missing"))
+    }
+
+    @Test
+    fun `re-put overwrites value`() {
+        val cache = ConcurrentLruCache<String, Int>(4)
+        cache.put("a", 1)
+        cache.put("a", 2)
+        assertEquals(2, cache.get("a"))
+        assertEquals(1, cache.size())
+    }
+
+    @Test
+    fun `evicts least-recently-put when over capacity`() {
+        val cache = ConcurrentLruCache<String, Int>(3)
+        cache.put("a", 1)
+        cache.put("b", 2)
+        cache.put("c", 3)
+        cache.put("d", 4) // pushes out "a"
+
+        assertNull(cache.get("a"))
+        assertEquals(2, cache.get("b"))
+        assertEquals(3, cache.get("c"))
+        assertEquals(4, cache.get("d"))
+        assertEquals(3, cache.size())
+    }
+
+    @Test
+    fun `re-putting a key refreshes its recency`() {
+        val cache = ConcurrentLruCache<String, Int>(3)
+        cache.put("a", 1)
+        cache.put("b", 2)
+        cache.put("c", 3)
+        // Touch "a" via put so it is no longer the oldest.
+        cache.put("a", 11)
+        cache.put("d", 4) // oldest is now "b", not "a"
+
+        assertNull(cache.get("b"))
+        assertEquals(11, cache.get("a"))
+        assertEquals(3, cache.get("c"))
+        assertEquals(4, cache.get("d"))
+    }
+
+    @Test
+    fun `get does not refresh recency`() {
+        val cache = ConcurrentLruCache<String, Int>(3)
+        cache.put("a", 1)
+        cache.put("b", 2)
+        cache.put("c", 3)
+        // A read must NOT save "a" from eviction (least-recently-put semantics).
+        assertEquals(1, cache.get("a"))
+        cache.put("d", 4)
+        assertNull(cache.get("a"))
+    }
+
+    @Test
+    fun `clear empties the cache`() {
+        val cache = ConcurrentLruCache<String, Int>(4)
+        cache.put("a", 1)
+        cache.put("b", 2)
+        cache.clear()
+        assertEquals(0, cache.size())
+        assertNull(cache.get("a"))
+    }
+
+    @Test
+    fun `size never exceeds capacity under concurrent puts`() {
+        val cap = 100
+        val cache = ConcurrentLruCache<Int, Int>(cap)
+        val threads = 8
+        val perThread = 5_000
+        val pool = Executors.newFixedThreadPool(threads)
+        val start = CountDownLatch(1)
+        val done = CountDownLatch(threads)
+        // Writes+eviction are atomic under one lock, so an external reader can see
+        // at most one over-cap entry (new inserted before old evicted).
+        val overBound = AtomicInteger(0)
+
+        repeat(threads) { t ->
+            pool.execute {
+                start.await()
+                for (i in 0 until perThread) {
+                    cache.put(t * perThread + i, i)
+                    // Interleave reads; they must never throw or take a lock.
+                    cache.get((t * perThread + i) - 1)
+                    if (cache.size() > cap + 1) overBound.incrementAndGet()
+                }
+                done.countDown()
+            }
+        }
+        start.countDown()
+        assertTrue(done.await(30, TimeUnit.SECONDS), "workers did not finish in time")
+        pool.shutdown()
+
+        assertEquals(0, overBound.get(), "cache size exceeded cap+1 during concurrent puts")
+        // Once writers quiesce it must settle to <= cap.
+        assertTrue(cache.size() <= cap, "final size ${cache.size()} exceeds cap $cap")
+    }
+}

quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt

@@ -34,12 +34,18 @@ import javax.crypto.spec.SecretKeySpec
  * (which is much cheaper than `getInstance`) plus the AEAD math itself.
  *
  * Single-thread per direction: one PacketProtection feeds either the read
- * loop OR the send loop, never both. The class still synchronizes on a
- * private monitor as a defence-in-depth: the lock-split refactor in
- * `QuicConnectionDriver` keeps each side single-threaded by design, but
- * a future caller (test harness, key-update path) sharing the instance
- * across coroutines would otherwise corrupt the cached `Cipher` state
- * silently. The JCA `Cipher` itself is documented as not thread-safe.
+ * loop OR the send loop, never both. The class still synchronizes as
+ * defence-in-depth: the lock-split refactor in `QuicConnectionDriver`
+ * keeps each side single-threaded by design, but a future caller (test
+ * harness, key-update path) sharing the instance across coroutines would
+ * otherwise corrupt the cached `Cipher` state silently. The JCA `Cipher`
+ * itself is documented as not thread-safe.
+ *
+ * Two independent monitors — [encryptLock] guards the encrypt-side state
+ * ([encryptCipher] + [recentEncryptNonces]); [decryptLock] guards
+ * [decryptCipher]. The two state groups are disjoint, so seal (send loop)
+ * and open (read loop) — which genuinely run on different coroutines —
+ * no longer serialize against each other per packet.
  */
 class JcaAesGcmAead(
     key: ByteArray,
@@ -67,6 +73,11 @@ class JcaAesGcmAead(
     private val encryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding")
     private val decryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding")
 
+    // Disjoint monitors so encrypt and decrypt don't serialize against each
+    // other. seal-family holds encryptLock; open-family holds decryptLock.
+    private val encryptLock = Any()
+    private val decryptLock = Any()
+
     /**
      * Last nonce successfully consumed by [seal]. We use a fresh
      * [Cipher.getInstance] when the caller asks us to seal under the
@@ -89,7 +100,7 @@ class JcaAesGcmAead(
         aad: ByteArray,
         plaintext: ByteArray,
     ): ByteArray =
-        synchronized(this) {
+        synchronized(encryptLock) {
             val reuse = recentEncryptNonces.any { it.contentEquals(nonce) }
             if (reuse) {
                 val fresh = Cipher.getInstance("AES/GCM/NoPadding")
@@ -132,7 +143,7 @@ class JcaAesGcmAead(
         aad: ByteArray,
         ciphertext: ByteArray,
     ): ByteArray? =
-        synchronized(this) {
+        synchronized(decryptLock) {
             try {
                 decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
                 decryptCipher.updateAAD(aad)
@@ -161,7 +172,7 @@ class JcaAesGcmAead(
         ciphertextOffset: Int,
         ciphertextLength: Int,
     ): ByteArray? =
-        synchronized(this) {
+        synchronized(decryptLock) {
             try {
                 decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
                 decryptCipher.updateAAD(aad, aadOffset, aadLength)
@@ -187,7 +198,7 @@ class JcaAesGcmAead(
         plaintextOffset: Int,
         plaintextLength: Int,
     ): ByteArray =
-        synchronized(this) {
+        synchronized(encryptLock) {
             val reuse = recentEncryptNonces.any { it.contentEquals(nonce) }
             if (reuse) {
                 val fresh = Cipher.getInstance("AES/GCM/NoPadding")
@@ -230,7 +241,7 @@ class JcaAesGcmAead(
         output: ByteArray,
         outputOffset: Int,
     ): Int =
-        synchronized(this) {
+        synchronized(encryptLock) {
             val reuse = recentEncryptNonces.any { it.contentEquals(nonce) }
             val cipher: Cipher
             if (reuse) {

quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaChaCha20Poly1305Aead.kt

@@ -35,10 +35,14 @@ import javax.crypto.spec.SecretKeySpec
  * allocations the pure-Kotlin [ChaCha20Poly1305Aead] requires.
  *
  * Single-thread per direction (one PacketProtection per side, one
- * direction per side). Synchronization on a private monitor is
- * defence-in-depth — a future caller (test harness, key-update path)
- * sharing the instance across coroutines would otherwise corrupt
- * the cached `Cipher` state silently.
+ * direction per side). Synchronization is defence-in-depth — a future
+ * caller (test harness, key-update path) sharing the instance across
+ * coroutines would otherwise corrupt the cached `Cipher` state silently.
+ *
+ * Two independent monitors — [encryptLock] guards the encrypt-side state
+ * ([encryptCipher] + [recentEncryptNonces]); [decryptLock] guards
+ * [decryptCipher]. Disjoint state, so seal and open don't serialize
+ * against each other per packet.
  */
 class JcaChaCha20Poly1305Aead(
     key: ByteArray,
@@ -55,6 +59,11 @@ class JcaChaCha20Poly1305Aead(
     private val encryptCipher: Cipher = Cipher.getInstance("ChaCha20-Poly1305")
     private val decryptCipher: Cipher = Cipher.getInstance("ChaCha20-Poly1305")
 
+    // Disjoint monitors so encrypt and decrypt don't serialize against each
+    // other. seal-family holds encryptLock; open-family holds decryptLock.
+    private val encryptLock = Any()
+    private val decryptLock = Any()
+
     /**
      * Last-N nonces successfully consumed by [seal] / [sealRange] /
      * [sealInto]. JCA's ChaCha20-Poly1305 (like AES-GCM) refuses to
@@ -72,7 +81,7 @@ class JcaChaCha20Poly1305Aead(
         aad: ByteArray,
         plaintext: ByteArray,
     ): ByteArray =
-        synchronized(this) {
+        synchronized(encryptLock) {
             sealCommon(nonce, aad, 0, aad.size, plaintext, 0, plaintext.size, output = null, outputOffset = 0)
                 .first
         }
@@ -87,7 +96,7 @@ class JcaChaCha20Poly1305Aead(
         plaintextOffset: Int,
         plaintextLength: Int,
     ): ByteArray =
-        synchronized(this) {
+        synchronized(encryptLock) {
             sealCommon(nonce, aad, aadOffset, aadLength, plaintext, plaintextOffset, plaintextLength, output = null, outputOffset = 0)
                 .first
         }
@@ -104,7 +113,7 @@ class JcaChaCha20Poly1305Aead(
         output: ByteArray,
         outputOffset: Int,
     ): Int =
-        synchronized(this) {
+        synchronized(encryptLock) {
             sealCommon(nonce, aad, aadOffset, aadLength, plaintext, plaintextOffset, plaintextLength, output, outputOffset)
                 .second
         }
@@ -177,7 +186,7 @@ class JcaChaCha20Poly1305Aead(
         aad: ByteArray,
         ciphertext: ByteArray,
     ): ByteArray? =
-        synchronized(this) {
+        synchronized(decryptLock) {
             try {
                 decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, IvParameterSpec(nonce))
                 decryptCipher.updateAAD(aad)
@@ -197,7 +206,7 @@ class JcaChaCha20Poly1305Aead(
         ciphertextOffset: Int,
         ciphertextLength: Int,
     ): ByteArray? =
-        synchronized(this) {
+        synchronized(decryptLock) {
             try {
                 decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, IvParameterSpec(nonce))
                 decryptCipher.updateAAD(aad, aadOffset, aadLength)

quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt

@@ -21,6 +21,8 @@
 package com.vitorpamplona.quic.transport
 
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExecutorCoroutineDispatcher
+import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.withContext
 import java.net.InetAddress
 import java.net.InetSocketAddress
@@ -28,17 +30,31 @@ import java.net.StandardSocketOptions
 import java.nio.ByteBuffer
 import java.nio.channels.ClosedChannelException
 import java.nio.channels.DatagramChannel
+import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicLong
 
 /**
- * JVM/Android UDP socket using blocking [DatagramChannel] dispatched onto
- * [Dispatchers.IO]. We don't use NIO selectors because each QUIC connection
- * has exactly one socket and one receive loop — Selector doesn't pay for
- * itself at this scale.
+ * JVM/Android UDP socket using a blocking [DatagramChannel]. We don't use NIO
+ * selectors because each QUIC connection has exactly one socket and one receive
+ * loop — a Selector doesn't pay for itself at this scale.
  *
- * The receive buffer is sized to 64 KiB (max IPv4/IPv6 datagram); QUIC packets
- * cap at MTU (~1500 in practice).
+ * Threading: the blocking `recvfrom` parks its thread for the *entire* life of
+ * the connection (it only returns when a datagram arrives or the socket
+ * closes). If that ran on the shared [Dispatchers.IO] pool it would pin one
+ * pool thread per connection, and past ~64 concurrent connections it would
+ * starve *all* other `Dispatchers.IO` work in the process — this module's and
+ * the host app's alike. So each socket owns two dedicated daemon threads:
+ * [recvDispatcher] for the perpetually-blocked receive, and [sendDispatcher]
+ * for the (rarely-blocking, but still-blocking) send. The receive can't share a
+ * thread with send — it would monopolise it — hence two. Both are shut down in
+ * [close]. QUIC's blocking socket I/O therefore never touches the shared pool.
+ *
+ * [connect] still resolves DNS + binds on [Dispatchers.IO]: that's a one-shot
+ * setup cost, not a lifetime parker, so it doesn't need isolation.
+ *
+ * The receive buffer is sized to typical Ethernet MTU; QUIC packets cap at MTU
+ * (~1500 in practice).
  */
 actual class UdpSocket private constructor(
     private val channel: DatagramChannel,
@@ -46,6 +62,19 @@ actual class UdpSocket private constructor(
 ) {
     private val closed = AtomicBoolean(false)
 
+    // Dedicated single-thread executors so the blocking socket calls never
+    // occupy the shared Dispatchers.IO pool. Daemon threads so a leaked socket
+    // can't keep the JVM alive. Separate recv/send threads because the receive
+    // parks continuously and would otherwise block sends behind it. We keep the
+    // ExecutorService handles (not just the dispatchers) so close() can call
+    // shutdownNow() — an interrupt that breaks the parked recvfrom immediately
+    // (ClosedByInterruptException) rather than the graceful shutdown() that
+    // dispatcher.close() would do.
+    private val recvExecutor = Executors.newSingleThreadExecutor { r -> Thread(r, "quic-udp-recv").apply { isDaemon = true } }
+    private val sendExecutor = Executors.newSingleThreadExecutor { r -> Thread(r, "quic-udp-send").apply { isDaemon = true } }
+    private val recvDispatcher: ExecutorCoroutineDispatcher = recvExecutor.asCoroutineDispatcher()
+    private val sendDispatcher: ExecutorCoroutineDispatcher = sendExecutor.asCoroutineDispatcher()
+
     // Sized to typical Ethernet MTU + a bit; QUIC tops out at ~1500 in practice
     // and any larger inbound frame is dropped as malformed anyway. The previous
     // 64 KiB buffer was wasteful per connection.
@@ -76,15 +105,19 @@ actual class UdpSocket private constructor(
     actual val receiveBufferSizeBytes: Int
         get() = channel.getOption(StandardSocketOptions.SO_RCVBUF)
 
-    actual suspend fun send(payload: ByteArray): Int =
-        withContext(Dispatchers.IO) {
+    actual suspend fun send(payload: ByteArray): Int {
+        // Fail fast without dispatching onto a possibly shut-down executor.
+        if (closed.get()) throw ClosedChannelException()
+        return withContext(sendDispatcher) {
             if (closed.get()) throw ClosedChannelException()
             val buf = ByteBuffer.wrap(payload)
             channel.send(buf, remote)
         }
+    }
 
-    actual suspend fun receive(): ByteArray? =
-        withContext(Dispatchers.IO) {
+    actual suspend fun receive(): ByteArray? {
+        if (closed.get()) return null
+        return withContext(recvDispatcher) {
             if (closed.get()) return@withContext null
             try {
                 // No synchronized — only the read loop touches readBuf, by
@@ -101,6 +134,7 @@ actual class UdpSocket private constructor(
                 null
             }
         }
+    }
 
     actual fun close() {
         if (closed.compareAndSet(false, true)) {
@@ -109,6 +143,15 @@ actual class UdpSocket private constructor(
             } catch (_: Throwable) {
                 // already closed
             }
+            // shutdownNow() interrupts the dedicated workers: a thread parked in
+            // a blocking recvfrom throws ClosedByInterruptException (a
+            // ClosedChannelException, caught below), so receive() returns null
+            // and both threads exit promptly instead of leaking per closed
+            // connection. channel.close() above would also unblock it
+            // (AsynchronousCloseException), but the interrupt is immediate and
+            // guarantees the executor terminates.
+            recvExecutor.shutdownNow()
+            sendExecutor.shutdownNow()
         }
     }
 

quic/src/jvmTest/kotlin/com/vitorpamplona/quic/transport/UdpSocketTest.kt

@@ -0,0 +1,146 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quic.transport
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import java.net.DatagramPacket
+import java.net.DatagramSocket
+import java.net.InetSocketAddress
+import java.nio.channels.ClosedChannelException
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class UdpSocketTest {
+    // A plain UDP peer on loopback that echoes one datagram back.
+    private val peer = DatagramSocket(InetSocketAddress("127.0.0.1", 0))
+
+    @AfterTest
+    fun tearDown() {
+        runCatching { peer.close() }
+    }
+
+    private fun threadNames(): List<String> = Thread.getAllStackTraces().keys.map { it.name }
+
+    private fun hasThreadPrefixed(prefix: String) = threadNames().any { it.startsWith(prefix) }
+
+    @Test
+    fun `round trips a datagram over loopback`() {
+        runBlocking {
+            val socket = UdpSocket.connect("127.0.0.1", peer.localPort)
+            try {
+                // Peer thread: receive one packet and echo it back to the sender.
+                val echo =
+                    Thread {
+                        val buf = ByteArray(2048)
+                        val incoming = DatagramPacket(buf, buf.size)
+                        peer.receive(incoming)
+                        peer.send(DatagramPacket(incoming.data, incoming.length, incoming.socketAddress))
+                    }.apply {
+                        isDaemon = true
+                        start()
+                    }
+
+                socket.send(byteArrayOf(1, 2, 3, 4))
+                val reply = withTimeoutOrNull(3_000) { socket.receive() }
+                echo.join(1_000)
+
+                assertTrue(reply != null && reply.contentEquals(byteArrayOf(1, 2, 3, 4)), "should echo the datagram back")
+                assertEquals(1, socket.receivedDatagramCount)
+            } finally {
+                socket.close()
+            }
+        }
+    }
+
+    @Test
+    fun `blocking receive runs on a dedicated thread, not Dispatchers-IO`() {
+        runBlocking {
+            val socket = UdpSocket.connect("127.0.0.1", peer.localPort)
+            try {
+                // Park a receive with no incoming datagram — it blocks in recvfrom
+                // on the socket's dedicated recv thread, not a Dispatchers.IO worker.
+                val pending = async(Dispatchers.IO) { socket.receive() }
+                // Give the receive time to reach the blocking call on its own thread.
+                delay(200)
+
+                assertTrue(hasThreadPrefixed("quic-udp-recv"), "a dedicated recv thread must carry the blocking receive")
+
+                // Closing unblocks the parked receive (returns null), proving the
+                // blocking call was on the dedicated thread and is released on close.
+                socket.close()
+                val result = withTimeoutOrNull(2_000) { pending.await() }
+                assertNull(result, "receive() must return null once the socket closes")
+            } finally {
+                socket.close()
+            }
+        }
+    }
+
+    @Test
+    fun `close shuts down the dedicated threads`() {
+        runBlocking {
+            val socket = UdpSocket.connect("127.0.0.1", peer.localPort)
+            try {
+                // The single-thread executors spawn their thread lazily on first
+                // use, so touch BOTH directions to bring both threads up.
+                socket.send(byteArrayOf(0))
+                val recv = launch(Dispatchers.IO) { socket.receive() }
+                delay(200)
+                assertTrue(
+                    hasThreadPrefixed("quic-udp-recv") && hasThreadPrefixed("quic-udp-send"),
+                    "both dedicated threads must be up while open",
+                )
+
+                socket.close()
+                recv.join()
+
+                // The executors shut down on close; their threads must exit promptly.
+                val gone =
+                    withTimeoutOrNull(2_000) {
+                        while (hasThreadPrefixed("quic-udp-recv") || hasThreadPrefixed("quic-udp-send")) delay(25)
+                        true
+                    }
+                assertTrue(gone == true, "dedicated recv/send threads must be gone after close() — else they leak per connection")
+            } finally {
+                socket.close()
+            }
+        }
+    }
+
+    @Test
+    fun `after close receive returns null and send throws`() {
+        runBlocking {
+            val socket = UdpSocket.connect("127.0.0.1", peer.localPort)
+            socket.close()
+            assertNull(socket.receive(), "receive() returns null after close")
+            assertFailsWith<ClosedChannelException> { socket.send(byteArrayOf(9)) }
+        }
+    }
+}