harley/hermes-relay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(android): serialize endpoint probe invalidation

Browse Source
This commit is contained in:
Bailey Dixon
2026-09-09 21:20:16 -04:00
parent c902215a15
commit c956232b96
4 changed files with 257 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this
### Fixed
- Android no longer crashes when a route probe finishes while a network change invalidates the endpoint cache.
- Bot Mode no longer crashes when different connections have bots with the same profile name. Both the conversation list and Active Now strip preserve each bot's connection, and opening progress appears only on the selected bot.
- Android feedback uses themed banners and action cards instead of platform toasts and default snackbars. Dashboard errors no longer misidentify missing resources as an outdated Relay. Developer settings includes local-only message previews.
- Missing chat attachments show their error and retry in the attachment card without repeated global popups. Global action messages occupy the top message area instead of covering the composer.
app/src/androidTest/kotlin/com/hermesandroid/relay/network/shared/EndpointResolverConcurrencyInstrumentedTest.kt
+100
View File
@@ -0,0 +1,100 @@
package com.hermesandroid.relay.network.shared
import androidx.test.ext.junit.runners.AndroidJUnit4
import com.hermesandroid.relay.data.ApiEndpoint
import com.hermesandroid.relay.data.EndpointCandidate
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import okhttp3.OkHttpClient
import okhttp3.Protocol
import okhttp3.Response
import okhttp3.ResponseBody.Companion.toResponseBody
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test
import org.junit.runner.RunWith
import java.io.InterruptedIOException
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
@RunWith(AndroidJUnit4::class)
class EndpointResolverConcurrencyInstrumentedTest {
@Test
fun probeCompletionRacingInvalidation_staysCrashFreeOnAndroidCollections() = runBlocking {
repeat(25) { iteration ->
val candidateCount = 8
val requestsStarted = CountDownLatch(candidateCount)
val releaseRequests = CountDownLatch(1)
val raceGate = CountDownLatch(1)
val requestSequence = AtomicInteger(0)
val client = OkHttpClient.Builder()
.addInterceptor { chain ->
if (requestSequence.incrementAndGet() <= candidateCount) {
requestsStarted.countDown()
releaseRequests.await(5, TimeUnit.SECONDS)
throw InterruptedIOException("instrumented invalidation race")
}
Response.Builder()
.request(chain.request())
.protocol(Protocol.HTTP_1_1)
.code(200)
.message("OK")
.body("{}".toResponseBody())
.build()
}
.build()
val resolver = EndpointResolver(client)
val candidates = (1..candidateCount).map { index ->
EndpointCandidate(
role = "instrumented-$iteration-$index",
priority = 0,
api = ApiEndpoint(host = "127.0.0.1", port = 1, tls = false),
)
}
try {
val staleResolve = async(start = CoroutineStart.UNDISPATCHED) {
resolver.resolve(candidates, EndpointSurface.Api)
}
assertTrue(requestsStarted.await(5, TimeUnit.SECONDS))
val invalidation = async(Dispatchers.Default) {
raceGate.await(5, TimeUnit.SECONDS)
resolver.clearCache()
}
val completions = async(Dispatchers.Default) {
raceGate.await(5, TimeUnit.SECONDS)
releaseRequests.countDown()
}
raceGate.countDown()
withTimeout(2_000L) {
invalidation.await()
completions.await()
staleResolve.await()
}
assertTrue(resolver.cacheSnapshot().isEmpty())
resolver.clearCache()
val freshWinner = withTimeout(2_000L) {
resolver.resolve(listOf(candidates.first()), EndpointSurface.Api)
}
assertEquals(candidates.first(), freshWinner)
assertTrue(
resolver.probeOutcomes.value.getValue(
EndpointResolver.cacheKey(candidates.first(), EndpointSurface.Api),
).reachable,
)
} finally {
raceGate.countDown()
releaseRequests.countDown()
client.dispatcher.executorService.shutdown()
}
}
}
}
app/src/main/kotlin/com/hermesandroid/relay/network/shared/EndpointResolver.kt
+11 -2
View File
@@ -149,7 +149,10 @@ class EndpointResolver(
)
private val probeCache = ConcurrentHashMap<String, CacheEntry>()
private val inFlightProbes = ConcurrentHashMap<String, Deferred<Boolean>>()
// Every access is owned by [probeStateLock]. This must not be a
// concurrently-mutated collection: clearCache() takes a stable snapshot
// while completion callbacks remove finished probes.
private val inFlightProbes = mutableMapOf<String, Deferred<Boolean>>()
private val probeScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
private val probeStateLock = Any()
private var probeGeneration = 0L
@@ -486,7 +489,13 @@ class EndpointResolver(
probe(candidate, surface, generation)
}.also { deferred ->
inFlightProbes[key] = deferred
deferred.invokeOnCompletion { inFlightProbes.remove(key, deferred) }
deferred.invokeOnCompletion {
synchronized(probeStateLock) {
// Identity-aware removal prevents an invalidated
// probe from removing its fresh replacement.
inFlightProbes.remove(key, deferred)
}
}
deferred.start()
}
}
app/src/test/kotlin/com/hermesandroid/relay/network/shared/EndpointResolverTest.kt
+145
View File
@@ -477,6 +477,151 @@ class EndpointResolverTest {
)
}
@Test
fun clearCache_handlesConcurrentProbeCompletions_withoutThrowingOrPublishingStaleState() = runTest {
val candidateCount = 24
val staleRequestsStarted = CountDownLatch(candidateCount)
val releaseStaleRequests = CountDownLatch(1)
val staleRequestsFinished = CountDownLatch(candidateCount)
val raceGate = CountDownLatch(1)
val requestSequence = AtomicInteger(0)
val blockingClient = fastClient.newBuilder()
.addInterceptor { chain ->
if (requestSequence.incrementAndGet() <= candidateCount) {
staleRequestsStarted.countDown()
try {
releaseStaleRequests.await(5, TimeUnit.SECONDS)
} finally {
staleRequestsFinished.countDown()
}
throw InterruptedIOException("concurrent invalidation test probe")
}
chain.proceed(chain.request())
}
.build()
val resolver = EndpointResolver(blockingClient, clock = { clockMillis.get() })
val candidates = (1..candidateCount).map { index ->
candidate("concurrent-clear-$index", priority = 0, server = reachableServer)
}
try {
val staleResolve = async(start = CoroutineStart.UNDISPATCHED) {
resolver.resolve(candidates, EndpointSurface.Api)
}
assertTrue(
"every physical probe must be active before the completion/invalidation race",
staleRequestsStarted.await(5, TimeUnit.SECONDS),
)
val invalidation = async(Dispatchers.Default) {
raceGate.await(5, TimeUnit.SECONDS)
resolver.clearCache()
}
val completions = async(Dispatchers.Default) {
raceGate.await(5, TimeUnit.SECONDS)
releaseStaleRequests.countDown()
}
raceGate.countDown()
withContext(Dispatchers.Default.limitedParallelism(1)) {
withTimeout(2_000L) {
invalidation.await()
completions.await()
staleResolve.await()
}
}
assertTrue(staleRequestsFinished.await(5, TimeUnit.SECONDS))
assertTrue(resolver.cacheSnapshot().isEmpty())
resolver.clearCache()
val freshWinner = withContext(Dispatchers.Default.limitedParallelism(1)) {
withTimeout(2_000L) {
resolver.resolve(listOf(candidates.first()), EndpointSurface.Api)
}
}
assertEquals(candidates.first(), freshWinner)
assertTrue(
"a completion racing invalidation must not overwrite the fresh generation",
resolver.probeOutcomes.value.getValue(
EndpointResolver.cacheKey(candidates.first(), EndpointSurface.Api),
).reachable,
)
} finally {
raceGate.countDown()
releaseStaleRequests.countDown()
}
}
@Test
fun invalidatedProbeCompletion_cannotRemoveFreshReplacement() = runTest {
val staleRequestStarted = CountDownLatch(1)
val releaseStaleRequest = CountDownLatch(1)
val staleRequestFinished = CountDownLatch(1)
val freshRequestStarted = CountDownLatch(1)
val releaseFreshRequest = CountDownLatch(1)
val requestSequence = AtomicInteger(0)
val blockingClient = fastClient.newBuilder()
.addInterceptor { chain ->
when (requestSequence.incrementAndGet()) {
1 -> {
staleRequestStarted.countDown()
try {
releaseStaleRequest.await(5, TimeUnit.SECONDS)
} finally {
staleRequestFinished.countDown()
}
throw InterruptedIOException("invalidated identity test probe")
}
2 -> {
freshRequestStarted.countDown()
releaseFreshRequest.await(5, TimeUnit.SECONDS)
chain.proceed(chain.request())
}
else -> chain.proceed(chain.request())
}
}
.build()
val resolver = EndpointResolver(blockingClient, clock = { clockMillis.get() })
val candidate = candidate("replacement-identity-test", priority = 0, server = reachableServer)
try {
val staleResolve = async(start = CoroutineStart.UNDISPATCHED) {
resolver.resolve(listOf(candidate), EndpointSurface.Api)
}
assertTrue(staleRequestStarted.await(5, TimeUnit.SECONDS))
resolver.clearCache()
val freshResolve = async(start = CoroutineStart.UNDISPATCHED) {
resolver.resolve(listOf(candidate), EndpointSurface.Api)
}
assertTrue(freshRequestStarted.await(5, TimeUnit.SECONDS))
releaseStaleRequest.countDown()
assertTrue(staleRequestFinished.await(5, TimeUnit.SECONDS))
withContext(Dispatchers.Default.limitedParallelism(1)) {
withTimeout(1_000L) { staleResolve.await() }
}
val joiningResolve = async(start = CoroutineStart.UNDISPATCHED) {
resolver.resolve(listOf(candidate), EndpointSurface.Api)
}
releaseFreshRequest.countDown()
withContext(Dispatchers.Default.limitedParallelism(1)) {
assertEquals(candidate, withTimeout(2_000L) { freshResolve.await() })
assertEquals(candidate, withTimeout(2_000L) { joiningResolve.await() })
}
assertEquals(
"the late stale completion must leave the fresh shared probe registered",
2,
requestSequence.get(),
)
} finally {
releaseStaleRequest.countDown()
releaseFreshRequest.countDown()
}
}
// ---------------------------------------------------------------
// Test 6 — cached-reachable result is re-probed after TTL
// ---------------------------------------------------------------