From 20244039ea2135aa9af9e8c3835e644cb3e5b5e6 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Sun, 9 Aug 2026 14:40:10 +0800 Subject: [PATCH 1/7] fix: prevent stale module binder deliveries Invalidate asynchronous delivery work when module UIDs disappear or the module cache is reset, and scope death recipients to the Binder they watch. Bound the delivery workers and add regression coverage for duplicate, stale, and reset attempts. Signed-off-by: LIghtJUNction --- .github/workflows/core.yml | 2 +- daemon/build.gradle.kts | 1 + .../daemon/ipc/DeliveryAttemptTracker.kt | 55 +++++++++++ .../vector/daemon/ipc/ModuleAppService.kt | 92 +++++++++++++------ .../daemon/ipc/DeliveryAttemptTrackerTest.kt | 49 ++++++++++ 5 files changed, 169 insertions(+), 30 deletions(-) create mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt create mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt diff --git a/.github/workflows/core.yml b/.github/workflows/core.yml index 89b65713d..ab5ee5a0f 100644 --- a/.github/workflows/core.yml +++ b/.github/workflows/core.yml @@ -123,7 +123,7 @@ jobs: - name: Build with Gradle run: | - ./gradlew zipAll + ./gradlew :daemon:testDebugUnitTest zipAll - name: Prepare artifact if: success() diff --git a/daemon/build.gradle.kts b/daemon/build.gradle.kts index 499e80978..52fe1002d 100644 --- a/daemon/build.gradle.kts +++ b/daemon/build.gradle.kts @@ -150,4 +150,5 @@ dependencies { implementation(projects.services.managerService) compileOnly(libs.androidx.annotation) compileOnly(projects.hiddenapi.stubs) + testImplementation("junit:junit:4.13.2") } diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt new file mode 100644 index 000000000..9dec596fd --- /dev/null +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt @@ -0,0 +1,55 @@ +package org.matrix.vector.daemon.ipc + +/** + * Serializes one Binder delivery attempt per uid and invalidates work after lifecycle changes. + * + * The observer callbacks and delivery workers run on different threads. Keeping the attempt + * ownership and generations behind one synchronized boundary prevents a duplicate callback from + * invalidating the worker that already owns the uid, and prevents a stale worker from removing a + * replacement attempt when it finishes. + */ +internal class DeliveryAttemptTracker { + + internal data class Attempt(val cacheGeneration: Long, val uidGeneration: Long) + + private var cacheGeneration = 0L + private val uidGenerations = mutableMapOf() + private val active = mutableMapOf() + + @Synchronized + fun begin(uid: Int): Attempt? { + if (active.containsKey(uid)) return null + val attempt = Attempt(cacheGeneration, nextUidGeneration(uid)) + active[uid] = attempt + return attempt + } + + @Synchronized + fun isCurrent(uid: Int, attempt: Attempt): Boolean = + cacheGeneration == attempt.cacheGeneration && active[uid] == attempt && + uidGenerations[uid] == attempt.uidGeneration + + @Synchronized + fun finish(uid: Int, attempt: Attempt) { + if (active[uid] == attempt) active.remove(uid) + } + + @Synchronized + fun invalidate(uid: Int) { + nextUidGeneration(uid) + active.remove(uid) + } + + @Synchronized + fun clear() { + cacheGeneration++ + active.clear() + uidGenerations.clear() + } + + private fun nextUidGeneration(uid: Int): Long { + val next = (uidGenerations[uid] ?: 0L) + 1L + uidGenerations[uid] = next + return next + } +} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt index 916ac2cb4..371881718 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt @@ -76,8 +76,8 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. */ private val uidSet = ConcurrentHashMap.newKeySet() - /** The uids a send is running for right now, so the three observer callbacks agree on one. */ - private val sending = ConcurrentHashMap.newKeySet() + /** Coordinates active delivery attempts and invalidates work after a uid/cache reset. */ + private val deliveryAttempts = DeliveryAttemptTracker() /** * What tells [uidSet] that a delivery is over: the provider binder we spoke to, and the @@ -125,46 +125,55 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. // AMS gives up on it, and it runs from an IUidObserver callback - one binder thread, serving // every uid transition on the device. A module app that never publishes therefore stalls the // delivery of every *other* module's binder behind it: eight and a half seconds, measured, on - // a device where one module app was crash-looping. One thread per module keeps that local. + // a device where one module app was crash-looping. A small fixed pool keeps this local without + // allowing an event storm to create an unbounded number of blocked threads. private val binderExecutor = - Executors.newCachedThreadPool { r -> Thread(r, "vector-module-binder") } + Executors.newFixedThreadPool(4) { r -> Thread(r, "vector-module-binder") } fun uidClear() { + deliveryAttempts.clear() uidSet.clear() + binderFailures.clear() + deliveries.forEach { (uid, delivery) -> + if (deliveries.remove(uid, delivery)) { + runCatching { delivery.first.unlinkToDeath(delivery.second, 0) } + } + } } fun uidStarts(uid: Int) { - if (uid in uidSet || !sending.add(uid)) return + if (uid in uidSet) return + val attempt = deliveryAttempts.begin(uid) ?: return val module = ConfigCache.getModuleByUid(uid) if (module?.code?.legacy != false) { - sending.remove(uid) + deliveryAttempts.finish(uid, attempt) return } if (isThrottled(uid)) { - sending.remove(uid) + deliveryAttempts.finish(uid, attempt) return } val service = serviceMap.getOrPut(module) { ModuleAppService(module) } - // Off the observer thread, and never inline: see [binderExecutor]. Caught, because a uid - // left in [sending] by a rejected submission is one this never looks at again. + // Off the observer thread, and never inline: see [binderExecutor]. Caught, because an + // attempt left in [deliveryAttempts] by a rejected submission is one this never looks at + // again. runCatching { binderExecutor.execute { try { val delivered = service.sendBinder(uid) - if (delivered != null) { - uidSet.add(uid) + if (deliveryAttempts.isCurrent(uid, attempt) && delivered != null) { binderFailures.remove(uid) - linkDelivery(uid, delivered) - } else { + linkDelivery(uid, delivered, attempt) + } else if (deliveryAttempts.isCurrent(uid, attempt)) { recordFailure(uid, module.packageName) } } finally { - sending.remove(uid) + deliveryAttempts.finish(uid, attempt) } } } .onFailure { - sending.remove(uid) + deliveryAttempts.finish(uid, attempt) Log.w(TAG, "Could not schedule the binder delivery for ${module.packageName}", it) } } @@ -177,17 +186,40 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. * recipient on a proxy is not a client of anything, so unlike the provider reference it puts * no floor under the process's priority. */ - private fun linkDelivery(uid: Int, provider: IBinder) { - val recipient = IBinder.DeathRecipient { uidSet.remove(uid) } + private fun linkDelivery( + uid: Int, + provider: IBinder, + attempt: DeliveryAttemptTracker.Attempt, + ) { + if (!deliveryAttempts.isCurrent(uid, attempt)) return + + lateinit var recipient: IBinder.DeathRecipient + recipient = IBinder.DeathRecipient { + val current = deliveries[uid] + if (current?.first === provider && current.second === recipient && + deliveries.remove(uid, current)) { + uidSet.remove(uid) + } + } + val delivery = provider to recipient + val previous = deliveries.put(uid, delivery) + previous?.let { (old, oldRecipient) -> + runCatching { old.unlinkToDeath(oldRecipient, 0) } + } + uidSet.add(uid) runCatching { - provider.linkToDeath(recipient, 0) - deliveries.put(uid, provider to recipient)?.let { (old, previous) -> - runCatching { old.unlinkToDeath(previous, 0) } - } - } - // Already dead, which is an answer in itself: whatever took the binder is gone, so the - // uid must not stay marked as served. - .onFailure { uidSet.remove(uid) } + provider.linkToDeath(recipient, 0) + if (!deliveryAttempts.isCurrent(uid, attempt) || deliveries[uid] !== delivery) { + if (deliveries.remove(uid, delivery)) uidSet.remove(uid) + runCatching { provider.unlinkToDeath(recipient, 0) } + } + } + // Already dead, which is an answer in itself: whatever took the binder is gone, so the + // uid must not stay marked as served. + .onFailure { + if (deliveries.remove(uid, delivery)) uidSet.remove(uid) + runCatching { provider.unlinkToDeath(recipient, 0) } + } } /** True while a uid has spent its attempts and its cooldown has not elapsed. */ @@ -199,8 +231,9 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. private fun recordFailure(uid: Int, modulePkg: String) { var crossed = false - // Read-modify-write in one step. Two threads cannot be here for one uid while [sending] - // holds, but that is an invariant of another field and not one to build arithmetic on. + // Read-modify-write in one step. Two threads cannot be here for one uid while + // [deliveryAttempts] holds the attempt, but that is an invariant of another field and not + // one to build arithmetic on. binderFailures.compute(uid) { _, previous -> val now = SystemClock.elapsedRealtime() val count = @@ -229,11 +262,12 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. } fun uidGone(uid: Int) { + deliveryAttempts.invalidate(uid) uidSet.remove(uid) // A send that never returns — `provider.call` runs the module's own onServiceBind, with no // deadline — would otherwise leave the uid here for the life of the daemon, and every later - // delivery for it refused at the top of uidStarts. - sending.remove(uid) + // delivery for it refused at the top of uidStarts. The generation invalidation above makes + // a late return inert and releases the uid for a replacement attempt. deliveries.remove(uid)?.let { (binder, recipient) -> runCatching { binder.unlinkToDeath(recipient, 0) } } diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt new file mode 100644 index 000000000..84fc93116 --- /dev/null +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt @@ -0,0 +1,49 @@ +package org.matrix.vector.daemon.ipc + +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertNotEquals +import org.junit.Assert.assertTrue +import org.junit.Test + +class DeliveryAttemptTrackerTest { + + @Test + fun duplicateStartsDoNotInvalidateTheActiveAttempt() { + val tracker = DeliveryAttemptTracker() + val first = tracker.begin(42) ?: error("first attempt was not created") + + assertNull(tracker.begin(42)) + assertTrue(tracker.isCurrent(42, first)) + } + + @Test + fun uidGoneInvalidatesOldWorkAndKeepsReplacementOwnership() { + val tracker = DeliveryAttemptTracker() + val first = tracker.begin(42) ?: error("first attempt was not created") + + tracker.invalidate(42) + val replacement = tracker.begin(42) ?: error("replacement attempt was not created") + + assertNotEquals(first, replacement) + assertFalse(tracker.isCurrent(42, first)) + assertTrue(tracker.isCurrent(42, replacement)) + + tracker.finish(42, first) + assertTrue(tracker.isCurrent(42, replacement)) + } + + @Test + fun clearInvalidatesEveryOldAttempt() { + val tracker = DeliveryAttemptTracker() + val first = tracker.begin(1) ?: error("first attempt was not created") + val second = tracker.begin(2) ?: error("second attempt was not created") + + tracker.clear() + + assertFalse(tracker.isCurrent(1, first)) + assertFalse(tracker.isCurrent(2, second)) + val replacement = tracker.begin(1) ?: error("replacement attempt was not created") + assertTrue(tracker.isCurrent(1, replacement)) + } +} From c3ac0edca94039aa7d1891034812bfc66a22b057 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Sun, 9 Aug 2026 17:24:13 +0800 Subject: [PATCH 2/7] fix: short-circuit stale binder deliveries Signed-off-by: LIghtJUNction --- .../matrix/vector/daemon/data/ConfigCache.kt | 15 +++++++++++++++ .../vector/daemon/ipc/ModuleAppService.kt | 18 ++++++++++++------ .../daemon/ipc/DeliveryAttemptTrackerTest.kt | 13 +++++++++++++ 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt index e81bf8310..d64792ce5 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt @@ -411,6 +411,15 @@ object ConfigCache { } } + // A new LoadedModule object means the service endpoint belongs to a new APK generation. A + // scope-only rebuild keeps the same objects and does not need to interrupt an already + // delivered module service. + val moduleGenerationChanged = + oldState.modules.size != newModules.size || + oldState.modules.any { (packageName, oldModule) -> + newModules[packageName] !== oldModule + } + // --- ATOMIC STATE SWAP --- // // Against the *current* state, not against the copy taken at the top of this function. A @@ -430,6 +439,12 @@ object ConfigCache { staticScopes = newStaticScopes } + if (moduleGenerationChanged) { + // The state swap starts a new cache generation. Any provider binder or queued delivery built + // from the previous module map must not repopulate delivery state after this point. + ModuleAppService.uidClear() + } + Log.d(TAG, "Cache Update Complete. Map Swap successful.") // Targets are removed only after the module set has been published. diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt index 371881718..64b06f050 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt @@ -160,12 +160,18 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. runCatching { binderExecutor.execute { try { - val delivered = service.sendBinder(uid) - if (deliveryAttempts.isCurrent(uid, attempt) && delivered != null) { - binderFailures.remove(uid) - linkDelivery(uid, delivered, attempt) - } else if (deliveryAttempts.isCurrent(uid, attempt)) { - recordFailure(uid, module.packageName) + // A fixed executor can queue work behind a blocked provider lookup. Do not start + // an obsolete lookup after uidGone() or a cache generation reset has already made + // this attempt inert; the post-send check below still handles invalidation while + // the lookup is in flight. + if (deliveryAttempts.isCurrent(uid, attempt)) { + val delivered = service.sendBinder(uid) + if (deliveryAttempts.isCurrent(uid, attempt) && delivered != null) { + binderFailures.remove(uid) + linkDelivery(uid, delivered, attempt) + } else if (deliveryAttempts.isCurrent(uid, attempt)) { + recordFailure(uid, module.packageName) + } } } finally { deliveryAttempts.finish(uid, attempt) diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt index 84fc93116..66a3f11d3 100644 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt @@ -46,4 +46,17 @@ class DeliveryAttemptTrackerTest { val replacement = tracker.begin(1) ?: error("replacement attempt was not created") assertTrue(tracker.isCurrent(1, replacement)) } + + @Test + fun staleCompletionAfterClearCannotReleaseReplacementOwnership() { + val tracker = DeliveryAttemptTracker() + val first = tracker.begin(7) ?: error("first attempt was not created") + + tracker.clear() + val replacement = tracker.begin(7) ?: error("replacement attempt was not created") + + tracker.finish(7, first) + + assertTrue(tracker.isCurrent(7, replacement)) + } } From 469f0e4de609426ad557cb3d188b3940e156aeb4 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Sun, 9 Aug 2026 19:50:14 +0800 Subject: [PATCH 3/7] fix: address stale module binder delivery review Scope cache invalidation to changed module generations, preserve failure throttles, and keep delivery attempts from blocking unrelated modules. Add deterministic coverage for failure churn, binder death identity, scoped invalidation, and generation detection. Signed-off-by: LIghtJUNction --- .../matrix/vector/daemon/data/ConfigCache.kt | 47 +++++-- .../vector/daemon/ipc/BinderFailureTracker.kt | 50 +++++++ .../daemon/ipc/DeliveryAttemptTracker.kt | 11 ++ .../vector/daemon/ipc/DeliveryRegistry.kt | 52 +++++++ .../vector/daemon/ipc/ModuleAppService.kt | 131 ++++++++---------- .../vector/daemon/data/ConfigCacheTest.kt | 32 +++++ .../daemon/ipc/BinderFailureTrackerTest.kt | 55 ++++++++ .../daemon/ipc/DeliveryAttemptTrackerTest.kt | 24 ++++ .../vector/daemon/ipc/DeliveryRegistryTest.kt | 40 ++++++ 9 files changed, 353 insertions(+), 89 deletions(-) create mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt create mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt create mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt create mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt create mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt index d64792ce5..024dfcd97 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt @@ -26,6 +26,25 @@ import org.matrix.vector.daemon.utils.getRealUsers private const val TAG = "VectorConfigCache" +/** Returns app ids whose [LoadedModule] object changed between two published generations. */ +internal fun moduleGenerationAppIds( + oldModules: Map, + newModules: Map, +): Set { + val changed = mutableSetOf() + oldModules.forEach { (packageName, oldModule) -> + if (newModules[packageName] !== oldModule && oldModule.appId >= 0) { + changed += oldModule.appId + } + } + newModules.forEach { (packageName, newModule) -> + if (oldModules[packageName] !== newModule && newModule.appId >= 0) { + changed += newModule.appId + } + } + return changed +} + object ConfigCache { // Module preference operations are delegated to PreferenceStore // Writable operations of modules are delegated to ModuleDatabase @@ -411,15 +430,6 @@ object ConfigCache { } } - // A new LoadedModule object means the service endpoint belongs to a new APK generation. A - // scope-only rebuild keeps the same objects and does not need to interrupt an already - // delivered module service. - val moduleGenerationChanged = - oldState.modules.size != newModules.size || - oldState.modules.any { (packageName, oldModule) -> - newModules[packageName] !== oldModule - } - // --- ATOMIC STATE SWAP --- // // Against the *current* state, not against the copy taken at the top of this function. A @@ -431,24 +441,31 @@ object ConfigCache { // // `oldState` is still the right thing to *read* from above: reusing an already-parsed module // is a decision about what was loaded when the rebuild started. - synchronized(this) { + val (publishedModules, changedModuleAppIds) = synchronized(this) { + // Compare with the state being replaced, not the snapshot from the start of this rebuild. + // A concurrent writer may have updated the state while package and APK queries ran above. + val currentModules = state.modules + val changedAppIds = moduleGenerationAppIds(currentModules, newModules) state = state.copy(modules = newModules, scopes = newScopes, unloadable = unloadable) // Swapped here rather than sixty lines earlier, so that the claims and the modules they // belong to become visible together. Between the two assignments a reader could see the new // static scopes against the old module set. staticScopes = newStaticScopes + currentModules to changedAppIds } - if (moduleGenerationChanged) { - // The state swap starts a new cache generation. Any provider binder or queued delivery built - // from the previous module map must not repopulate delivery state after this point. - ModuleAppService.uidClear() + if (changedModuleAppIds.isNotEmpty()) { + // Only the module generations that changed need to release their provider binders. Keeping + // other modules' deliveries and failure runs avoids re-feeding an unrelated crash loop. + // A module that is already running may not emit another uid transition, so immediately + // offer the affected uids a delivery for the newly published generation. + ModuleAppService.uidClear(changedModuleAppIds).forEach { ModuleAppService.uidStarts(it) } } Log.d(TAG, "Cache Update Complete. Map Swap successful.") // Targets are removed only after the module set has been published. - (oldState.modules.keys - newModules.keys).forEach { + (publishedModules.keys - newModules.keys).forEach { FrameworkService.forgetHotReloadTargets(it) } FrameworkService.backfillLoadedVersions() diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt new file mode 100644 index 000000000..4006b386f --- /dev/null +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt @@ -0,0 +1,50 @@ +package org.matrix.vector.daemon.ipc + +import android.os.SystemClock + +/** + * Keeps the retry throttle for failed module-provider handshakes. + * + * The clock and limits are injectable so the lifecycle rules can be tested without waiting for + * real time. A uid's failure run deliberately survives uidGone() and cache invalidation: those + * events describe a process or module generation change, not a successful handshake. + */ +internal class BinderFailureTracker( + private val now: () -> Long = { SystemClock.elapsedRealtime() }, + private val maxConsecutiveFailures: Int = 3, + private val retryCooldownMs: Long = 60_000L, + private val failureRunMs: Long = 10 * retryCooldownMs, +) { + + private data class FailureRun(val count: Int, val atElapsed: Long) + + private val failures = mutableMapOf() + + @Synchronized + fun isThrottled(uid: Int): Boolean { + val run = failures[uid] ?: return false + if (run.count < maxConsecutiveFailures) return false + return now() - run.atElapsed < retryCooldownMs + } + + /** Records one failed send and reports whether this call crossed the throttle threshold. */ + @Synchronized + fun recordFailure(uid: Int): Boolean { + val timestamp = now() + val previous = failures[uid] + val count = + when { + previous == null || timestamp - previous.atElapsed >= failureRunMs -> 1 + else -> minOf(previous.count + 1, maxConsecutiveFailures) + } + failures[uid] = FailureRun(count, timestamp) + return count == maxConsecutiveFailures && (previous?.count ?: 0) < count + } + + @Synchronized + fun clear(uid: Int) { + failures.remove(uid) + } + + internal fun count(uid: Int): Int? = synchronized(this) { failures[uid]?.count } +} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt index 9dec596fd..fd8a2789b 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt @@ -40,6 +40,17 @@ internal class DeliveryAttemptTracker { active.remove(uid) } + /** Invalidates only the active attempts selected by [predicate]. */ + @Synchronized + fun invalidateMatching(predicate: (Int) -> Boolean): Set { + val invalidated = active.keys.filter(predicate).toSet() + invalidated.forEach { uid -> + nextUidGeneration(uid) + active.remove(uid) + } + return invalidated + } + @Synchronized fun clear() { cacheGeneration++ diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt new file mode 100644 index 000000000..43f68a0e6 --- /dev/null +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt @@ -0,0 +1,52 @@ +package org.matrix.vector.daemon.ipc + +/** + * Stores the exact provider/recipient pair associated with a uid. + * + * Binder proxies and death recipients are identity-bearing objects. In particular, a late death + * callback from an old provider must not remove a replacement entry for the same uid. Keeping the + * identity check here makes that rule explicit and gives it a small, deterministic unit-test + * surface. + */ +internal class DeliveryRegistry { + + internal data class Entry( + val provider: Provider, + val recipient: Recipient, + ) + + private val entries = mutableMapOf>() + + @Synchronized + fun put(uid: Int, provider: Provider, recipient: Recipient): Entry? = + entries.put(uid, Entry(provider, recipient)) + + @Synchronized + fun isCurrent(uid: Int, provider: Provider, recipient: Recipient): Boolean = + entries[uid]?.let { it.provider === provider && it.recipient === recipient } == true + + @Synchronized + fun removeIfCurrent(uid: Int, provider: Provider, recipient: Recipient): Boolean { + val current = entries[uid] ?: return false + if (current.provider !== provider || current.recipient !== recipient) return false + entries.remove(uid) + return true + } + + @Synchronized + fun remove(uid: Int): Entry? = entries.remove(uid) + + @Synchronized + fun removeMatching(predicate: (Int) -> Boolean): List>> { + val removed = mutableListOf>>() + entries.entries.removeIf { (uid, entry) -> + if (!predicate(uid)) { + false + } else { + removed += uid to entry + true + } + } + return removed + } +} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt index 64b06f050..59f6b3ef2 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt @@ -7,7 +7,6 @@ import android.os.Bundle import android.os.IBinder import android.os.ParcelFileDescriptor import android.os.RemoteException -import android.os.SystemClock import android.util.Log import io.github.libxposed.service.HookedProcess import io.github.libxposed.service.IHotReloadCallback @@ -84,7 +83,7 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. * recipient watching it. Held because a `DeathRecipient` nothing references is one the runtime * may collect before it ever fires. */ - private val deliveries = ConcurrentHashMap>() + private val deliveries = DeliveryRegistry() private val serviceMap = Collections.synchronizedMap(WeakHashMap()) @@ -104,41 +103,46 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. * crash-looping copy in a work profile throttle the healthy copy in user 0, and let either * one's success wipe the other's run. * - * Once [MAX_CONSECUTIVE_BINDER_FAILURES] have piled up the retries are throttled to one per - * [BINDER_RETRY_COOLDOWN_MS] — the count is held at the ceiling rather than reset by the - * attempt that the cooldown lets through, or the ceiling would simply be re-climbed and three - * more attempts allowed every minute for ever. A run is forgotten after - * [BINDER_FAILURE_RUN_MS] without a failure, so an occasional one never accumulates. Throttled - * rather than abandoned, and cleared by the first success, because the app may simply have been - * mid-update or out of memory; a module written off for good on three failures would be a worse - * bug than the one this is fixing. + * Once three failures have piled up the retries are throttled to one per minute — the count is + * held at the ceiling rather than reset by the attempt that the cooldown lets through, or the + * ceiling would simply be re-climbed and three more attempts allowed every minute for ever. A + * run is forgotten after ten minutes without a failure, so an occasional one never accumulates. + * Throttled rather than abandoned, and cleared by the first success, because the app may simply + * have been mid-update or out of memory; a module written off for good on three failures would + * be a worse bug than the one this is fixing. */ - private val binderFailures = ConcurrentHashMap() - - private class FailureRun(val count: Int, val atElapsed: Long) - - private const val MAX_CONSECUTIVE_BINDER_FAILURES = 3 - private const val BINDER_RETRY_COOLDOWN_MS = 60_000L - private const val BINDER_FAILURE_RUN_MS = 10 * BINDER_RETRY_COOLDOWN_MS + private val binderFailures = BinderFailureTracker() // The delivery blocks in getContentProviderExternal until the app publishes its provider or // AMS gives up on it, and it runs from an IUidObserver callback - one binder thread, serving // every uid transition on the device. A module app that never publishes therefore stalls the // delivery of every *other* module's binder behind it: eight and a half seconds, measured, on - // a device where one module app was crash-looping. A small fixed pool keeps this local without - // allowing an event storm to create an unbounded number of blocked threads. + // a device where one module app was crash-looping. Keep one worker per blocked lookup instead + // of queueing all modules behind a fixed global pool: [deliveryAttempts] deduplicates repeated + // callbacks for a uid, and uidGone() invalidation intentionally lets a replacement proceed + // without waiting for the stale lookup to return. private val binderExecutor = - Executors.newFixedThreadPool(4) { r -> Thread(r, "vector-module-binder") } - - fun uidClear() { - deliveryAttempts.clear() - uidSet.clear() - binderFailures.clear() - deliveries.forEach { (uid, delivery) -> - if (deliveries.remove(uid, delivery)) { - runCatching { delivery.first.unlinkToDeath(delivery.second, 0) } - } + Executors.newCachedThreadPool { r -> Thread(r, "vector-module-binder") } + + /** + * Invalidates deliveries for the module generations identified by [moduleAppIds]. Other + * modules keep both their live binder and their retry throttle. + */ + fun uidClear(moduleAppIds: Set): Set { + if (moduleAppIds.isEmpty()) return emptySet() + val belongsToChangedModule = { uid: Int -> uid % PER_USER_RANGE in moduleAppIds } + val invalidatedUids = mutableSetOf() + + invalidatedUids += deliveryAttempts.invalidateMatching(belongsToChangedModule) + uidSet.toList().forEach { uid -> + if (belongsToChangedModule(uid) && uidSet.remove(uid)) invalidatedUids += uid + } + deliveries.removeMatching(belongsToChangedModule).forEach { (uid, delivery) -> + invalidatedUids += uid + uidSet.remove(uid) + runCatching { delivery.provider.unlinkToDeath(delivery.recipient, 0) } } + return invalidatedUids } fun uidStarts(uid: Int) { @@ -160,17 +164,20 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. runCatching { binderExecutor.execute { try { - // A fixed executor can queue work behind a blocked provider lookup. Do not start - // an obsolete lookup after uidGone() or a cache generation reset has already made - // this attempt inert; the post-send check below still handles invalidation while - // the lookup is in flight. + // Do not start an obsolete lookup after uidGone() or a cache generation reset has + // already made this attempt inert; the post-send check below still handles + // invalidation while the lookup is in flight. if (deliveryAttempts.isCurrent(uid, attempt)) { val delivered = service.sendBinder(uid) - if (deliveryAttempts.isCurrent(uid, attempt) && delivered != null) { - binderFailures.remove(uid) - linkDelivery(uid, delivered, attempt) - } else if (deliveryAttempts.isCurrent(uid, attempt)) { + if (delivered == null) { + // A uid can disappear while AMS is waiting for its provider. That makes this + // attempt stale, but it is still a failed launch and must feed the retry + // throttle; otherwise a crash-looping app can evade the three-failure limit + // by dying at exactly this point. recordFailure(uid, module.packageName) + } else if (deliveryAttempts.isCurrent(uid, attempt)) { + binderFailures.clear(uid) + linkDelivery(uid, delivered, attempt) } } } finally { @@ -201,69 +208,45 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. lateinit var recipient: IBinder.DeathRecipient recipient = IBinder.DeathRecipient { - val current = deliveries[uid] - if (current?.first === provider && current.second === recipient && - deliveries.remove(uid, current)) { + if (deliveries.removeIfCurrent(uid, provider, recipient)) { uidSet.remove(uid) } } - val delivery = provider to recipient - val previous = deliveries.put(uid, delivery) - previous?.let { (old, oldRecipient) -> - runCatching { old.unlinkToDeath(oldRecipient, 0) } + deliveries.put(uid, provider, recipient)?.let { previous -> + runCatching { previous.provider.unlinkToDeath(previous.recipient, 0) } } uidSet.add(uid) runCatching { provider.linkToDeath(recipient, 0) - if (!deliveryAttempts.isCurrent(uid, attempt) || deliveries[uid] !== delivery) { - if (deliveries.remove(uid, delivery)) uidSet.remove(uid) + if (!deliveryAttempts.isCurrent(uid, attempt) || + !deliveries.isCurrent(uid, provider, recipient)) { + if (deliveries.removeIfCurrent(uid, provider, recipient)) uidSet.remove(uid) runCatching { provider.unlinkToDeath(recipient, 0) } } } // Already dead, which is an answer in itself: whatever took the binder is gone, so the // uid must not stay marked as served. .onFailure { - if (deliveries.remove(uid, delivery)) uidSet.remove(uid) + if (deliveries.removeIfCurrent(uid, provider, recipient)) uidSet.remove(uid) runCatching { provider.unlinkToDeath(recipient, 0) } } } /** True while a uid has spent its attempts and its cooldown has not elapsed. */ private fun isThrottled(uid: Int): Boolean { - val run = binderFailures[uid] ?: return false - if (run.count < MAX_CONSECUTIVE_BINDER_FAILURES) return false - return SystemClock.elapsedRealtime() - run.atElapsed < BINDER_RETRY_COOLDOWN_MS + return binderFailures.isThrottled(uid) } private fun recordFailure(uid: Int, modulePkg: String) { - var crossed = false - // Read-modify-write in one step. Two threads cannot be here for one uid while - // [deliveryAttempts] holds the attempt, but that is an invariant of another field and not - // one to build arithmetic on. - binderFailures.compute(uid) { _, previous -> - val now = SystemClock.elapsedRealtime() - val count = - when { - // A run is forgotten only after a long quiet spell, not after one cooldown. Forgetting - // it at the cooldown meant the attempt the cooldown let through reset the count, so - // the ceiling was re-climbed and three more attempts allowed every minute, for ever. - previous == null || now - previous.atElapsed >= BINDER_FAILURE_RUN_MS -> 1 - // Held at the ceiling rather than growing without bound: what the number decides is - // only whether we are throttled, and pinning it here is what makes the cooldown mean - // one attempt rather than another three. - else -> minOf(previous.count + 1, MAX_CONSECUTIVE_BINDER_FAILURES) - } - crossed = count == MAX_CONSECUTIVE_BINDER_FAILURES && (previous?.count ?: 0) < count - FailureRun(count, now) - } + val crossed = binderFailures.recordFailure(uid) // Once, on the way past the ceiling. The failures themselves are already logged one by one // in sendBinder; what is worth saying here is that we have stopped trying, which is the part // a reader chasing a module that never receives its service cannot otherwise see. if (crossed) { Log.w( TAG, - "$modulePkg/$uid failed to take its binder $MAX_CONSECUTIVE_BINDER_FAILURES times in" + - " a row; retrying at most once every ${BINDER_RETRY_COOLDOWN_MS / 1000}s") + "$modulePkg/$uid failed to take its binder three times in a row; retrying at most once" + + " every 60s") } } @@ -274,8 +257,8 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. // deadline — would otherwise leave the uid here for the life of the daemon, and every later // delivery for it refused at the top of uidStarts. The generation invalidation above makes // a late return inert and releases the uid for a replacement attempt. - deliveries.remove(uid)?.let { (binder, recipient) -> - runCatching { binder.unlinkToDeath(recipient, 0) } + deliveries.remove(uid)?.let { delivery -> + runCatching { delivery.provider.unlinkToDeath(delivery.recipient, 0) } } } diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt new file mode 100644 index 000000000..6a5a062cd --- /dev/null +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt @@ -0,0 +1,32 @@ +package org.matrix.vector.daemon.data + +import org.junit.Assert.assertEquals +import org.junit.Test +import org.matrix.vector.ipc.LoadedModule + +class ConfigCacheTest { + + @Test + fun onlyReplacedModuleAppIdsAreInvalidated() { + val unchanged = LoadedModule().apply { appId = 10001 } + val oldChanged = LoadedModule().apply { appId = 10002 } + val newChanged = LoadedModule().apply { appId = 10002 } + val added = LoadedModule().apply { appId = 10003 } + + val oldModules = mapOf("unchanged" to unchanged, "changed" to oldChanged) + val newModules = + mapOf("unchanged" to unchanged, "changed" to newChanged, "added" to added) + + assertEquals(setOf(10002, 10003), moduleGenerationAppIds(oldModules, newModules)) + } + + @Test + fun removedModuleAppIdIsInvalidated() { + val removed = LoadedModule().apply { appId = 10004 } + + assertEquals( + setOf(10004), + moduleGenerationAppIds(mapOf("removed" to removed), emptyMap()), + ) + } +} diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt new file mode 100644 index 000000000..c453b4db4 --- /dev/null +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt @@ -0,0 +1,55 @@ +package org.matrix.vector.daemon.ipc + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test + +class BinderFailureTrackerTest { + + @Test + fun failureDuringUidChurnStillReachesThrottle() { + var now = 0L + val failures = + BinderFailureTracker( + now = { now }, + maxConsecutiveFailures = 3, + retryCooldownMs = 100, + failureRunMs = 1_000, + ) + val attempts = DeliveryAttemptTracker() + val attempt = attempts.begin(42) ?: error("attempt was not created") + + assertFalse(failures.recordFailure(42)) + attempts.invalidate(42) + now = 1 + assertFalse(failures.recordFailure(42)) + now = 2 + assertTrue(failures.recordFailure(42)) + + assertEquals(3, failures.count(42)) + assertFalse(attempts.isCurrent(42, attempt)) + assertTrue(failures.isThrottled(42)) + } + + @Test + fun cooldownAttemptDoesNotResetTheFailureRun() { + var now = 0L + val failures = + BinderFailureTracker( + now = { now }, + maxConsecutiveFailures = 3, + retryCooldownMs = 100, + failureRunMs = 1_000, + ) + + repeat(3) { failures.recordFailure(7) } + now = 100 + assertFalse(failures.isThrottled(7)) + failures.recordFailure(7) + + assertEquals(3, failures.count(7)) + now = 101 + assertTrue(failures.isThrottled(7)) + } +} diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt index 66a3f11d3..395dc3236 100644 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt @@ -59,4 +59,28 @@ class DeliveryAttemptTrackerTest { assertTrue(tracker.isCurrent(7, replacement)) } + + @Test + fun scopedInvalidationLeavesOtherModuleAttemptCurrent() { + val tracker = DeliveryAttemptTracker() + val changedModule = tracker.begin(10042) ?: error("changed-module attempt was not created") + val otherModule = tracker.begin(20043) ?: error("other-module attempt was not created") + + tracker.invalidateMatching { it == 10042 } + + assertFalse(tracker.isCurrent(10042, changedModule)) + assertTrue(tracker.isCurrent(20043, otherModule)) + } + + @Test + fun queuedObsoleteAttemptDoesNotStartProviderLookup() { + val tracker = DeliveryAttemptTracker() + val attempt = tracker.begin(42) ?: error("attempt was not created") + tracker.invalidate(42) + + var lookupStarted = false + if (tracker.isCurrent(42, attempt)) lookupStarted = true + + assertFalse(lookupStarted) + } } diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt new file mode 100644 index 000000000..f6b38e3b7 --- /dev/null +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt @@ -0,0 +1,40 @@ +package org.matrix.vector.daemon.ipc + +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertTrue +import org.junit.Test + +class DeliveryRegistryTest { + + @Test + fun lateDeathOfOldProviderCannotRemoveReplacement() { + val registry = DeliveryRegistry() + val oldProvider = Any() + val oldRecipient = Any() + val newProvider = Any() + val newRecipient = Any() + + registry.put(42, oldProvider, oldRecipient) + registry.put(42, newProvider, newRecipient) + + assertFalse(registry.removeIfCurrent(42, oldProvider, oldRecipient)) + assertTrue(registry.isCurrent(42, newProvider, newRecipient)) + assertNotNull(registry.remove(42)) + } + + @Test + fun scopedRemovalDoesNotTouchOtherUids() { + val registry = DeliveryRegistry() + val provider = Any() + val recipient = Any() + registry.put(10042, provider, recipient) + registry.put(20043, provider, recipient) + + val removed = registry.removeMatching { it == 10042 } + + assertTrue(removed.any { it.first == 10042 }) + assertFalse(registry.isCurrent(10042, provider, recipient)) + assertTrue(registry.isCurrent(20043, provider, recipient)) + } +} From 2335baebf34d98f0b470776f1c112e8ae349ccbc Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Sun, 9 Aug 2026 22:20:38 +0800 Subject: [PATCH 4/7] refactor: centralize module binder delivery state Signed-off-by: LIghtJUNction --- .../vector/daemon/ipc/BinderFailureTracker.kt | 50 --- .../daemon/ipc/DeliveryAttemptTracker.kt | 66 ---- .../vector/daemon/ipc/DeliveryRegistry.kt | 52 --- .../matrix/vector/daemon/ipc/DeliveryState.kt | 298 ++++++++++++++++++ .../vector/daemon/ipc/ModuleAppService.kt | 137 ++------ .../daemon/ipc/BinderFailureTrackerTest.kt | 55 ---- .../daemon/ipc/DeliveryAttemptTrackerTest.kt | 86 ----- .../vector/daemon/ipc/DeliveryRegistryTest.kt | 40 --- .../vector/daemon/ipc/DeliveryStateTest.kt | 94 ++++++ 9 files changed, 426 insertions(+), 452 deletions(-) delete mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt delete mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt delete mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt create mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt delete mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt delete mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt delete mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt create mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt deleted file mode 100644 index 4006b386f..000000000 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTracker.kt +++ /dev/null @@ -1,50 +0,0 @@ -package org.matrix.vector.daemon.ipc - -import android.os.SystemClock - -/** - * Keeps the retry throttle for failed module-provider handshakes. - * - * The clock and limits are injectable so the lifecycle rules can be tested without waiting for - * real time. A uid's failure run deliberately survives uidGone() and cache invalidation: those - * events describe a process or module generation change, not a successful handshake. - */ -internal class BinderFailureTracker( - private val now: () -> Long = { SystemClock.elapsedRealtime() }, - private val maxConsecutiveFailures: Int = 3, - private val retryCooldownMs: Long = 60_000L, - private val failureRunMs: Long = 10 * retryCooldownMs, -) { - - private data class FailureRun(val count: Int, val atElapsed: Long) - - private val failures = mutableMapOf() - - @Synchronized - fun isThrottled(uid: Int): Boolean { - val run = failures[uid] ?: return false - if (run.count < maxConsecutiveFailures) return false - return now() - run.atElapsed < retryCooldownMs - } - - /** Records one failed send and reports whether this call crossed the throttle threshold. */ - @Synchronized - fun recordFailure(uid: Int): Boolean { - val timestamp = now() - val previous = failures[uid] - val count = - when { - previous == null || timestamp - previous.atElapsed >= failureRunMs -> 1 - else -> minOf(previous.count + 1, maxConsecutiveFailures) - } - failures[uid] = FailureRun(count, timestamp) - return count == maxConsecutiveFailures && (previous?.count ?: 0) < count - } - - @Synchronized - fun clear(uid: Int) { - failures.remove(uid) - } - - internal fun count(uid: Int): Int? = synchronized(this) { failures[uid]?.count } -} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt deleted file mode 100644 index fd8a2789b..000000000 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt +++ /dev/null @@ -1,66 +0,0 @@ -package org.matrix.vector.daemon.ipc - -/** - * Serializes one Binder delivery attempt per uid and invalidates work after lifecycle changes. - * - * The observer callbacks and delivery workers run on different threads. Keeping the attempt - * ownership and generations behind one synchronized boundary prevents a duplicate callback from - * invalidating the worker that already owns the uid, and prevents a stale worker from removing a - * replacement attempt when it finishes. - */ -internal class DeliveryAttemptTracker { - - internal data class Attempt(val cacheGeneration: Long, val uidGeneration: Long) - - private var cacheGeneration = 0L - private val uidGenerations = mutableMapOf() - private val active = mutableMapOf() - - @Synchronized - fun begin(uid: Int): Attempt? { - if (active.containsKey(uid)) return null - val attempt = Attempt(cacheGeneration, nextUidGeneration(uid)) - active[uid] = attempt - return attempt - } - - @Synchronized - fun isCurrent(uid: Int, attempt: Attempt): Boolean = - cacheGeneration == attempt.cacheGeneration && active[uid] == attempt && - uidGenerations[uid] == attempt.uidGeneration - - @Synchronized - fun finish(uid: Int, attempt: Attempt) { - if (active[uid] == attempt) active.remove(uid) - } - - @Synchronized - fun invalidate(uid: Int) { - nextUidGeneration(uid) - active.remove(uid) - } - - /** Invalidates only the active attempts selected by [predicate]. */ - @Synchronized - fun invalidateMatching(predicate: (Int) -> Boolean): Set { - val invalidated = active.keys.filter(predicate).toSet() - invalidated.forEach { uid -> - nextUidGeneration(uid) - active.remove(uid) - } - return invalidated - } - - @Synchronized - fun clear() { - cacheGeneration++ - active.clear() - uidGenerations.clear() - } - - private fun nextUidGeneration(uid: Int): Long { - val next = (uidGenerations[uid] ?: 0L) + 1L - uidGenerations[uid] = next - return next - } -} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt deleted file mode 100644 index 43f68a0e6..000000000 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistry.kt +++ /dev/null @@ -1,52 +0,0 @@ -package org.matrix.vector.daemon.ipc - -/** - * Stores the exact provider/recipient pair associated with a uid. - * - * Binder proxies and death recipients are identity-bearing objects. In particular, a late death - * callback from an old provider must not remove a replacement entry for the same uid. Keeping the - * identity check here makes that rule explicit and gives it a small, deterministic unit-test - * surface. - */ -internal class DeliveryRegistry { - - internal data class Entry( - val provider: Provider, - val recipient: Recipient, - ) - - private val entries = mutableMapOf>() - - @Synchronized - fun put(uid: Int, provider: Provider, recipient: Recipient): Entry? = - entries.put(uid, Entry(provider, recipient)) - - @Synchronized - fun isCurrent(uid: Int, provider: Provider, recipient: Recipient): Boolean = - entries[uid]?.let { it.provider === provider && it.recipient === recipient } == true - - @Synchronized - fun removeIfCurrent(uid: Int, provider: Provider, recipient: Recipient): Boolean { - val current = entries[uid] ?: return false - if (current.provider !== provider || current.recipient !== recipient) return false - entries.remove(uid) - return true - } - - @Synchronized - fun remove(uid: Int): Entry? = entries.remove(uid) - - @Synchronized - fun removeMatching(predicate: (Int) -> Boolean): List>> { - val removed = mutableListOf>>() - entries.entries.removeIf { (uid, entry) -> - if (!predicate(uid)) { - false - } else { - removed += uid to entry - true - } - } - return removed - } -} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt new file mode 100644 index 000000000..479f0ecbb --- /dev/null +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt @@ -0,0 +1,298 @@ +package org.matrix.vector.daemon.ipc + +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicLong + +/** + * The complete lifecycle of one module-app UID. + * + * Failure accounting and the active bit live in every state rather than in side maps. The + * [DeliveryStateStore] changes one UID only through [ConcurrentHashMap.compute], so a worker result + * cannot pass an ownership check and then be invalidated before its state is committed. + */ +internal sealed interface DeliveryState { + val active: Boolean + val lastAttemptId: Long + + data class Idle( + override val active: Boolean, + override val lastAttemptId: Long = 0L, + val failureCount: Int = 0, + val lastFailureAt: Long = 0L, + ) : DeliveryState + + data class Sending( + override val active: Boolean, + val attemptId: Long, + val failureCount: Int = 0, + val lastFailureAt: Long = 0L, + ) : DeliveryState { + override val lastAttemptId: Long + get() = attemptId + } + + data class Delivered( + override val active: Boolean, + val attemptId: Long, + val provider: Provider, + val recipient: Recipient, + ) : DeliveryState { + override val lastAttemptId: Long + get() = attemptId + } + + data class Throttled( + override val active: Boolean, + override val lastAttemptId: Long, + val count: Int, + val lastFailureAt: Long, + val cooldownUntil: Long, + ) : DeliveryState +} + +/** + * Atomic state machine for module-app binder delivery. + * + * There is deliberately one map and no companion attempt/registry/failure maps. A tombstoned + * [Sending] state is retained after [uidGone] until its worker completes, which lets a failure be + * counted while still preventing that old worker from publishing a replacement. + */ +internal class DeliveryStateStore( + private val now: () -> Long = { android.os.SystemClock.elapsedRealtime() }, + private val maxConsecutiveFailures: Int = 3, + private val retryCooldownMs: Long = 60_000L, + private val failureRunMs: Long = 10 * retryCooldownMs, +) { + + internal data class RemovedDelivery( + val uid: Int, + val provider: Provider, + val recipient: Recipient, + ) + + internal data class Invalidation( + val redeliveryUids: Set, + val removedDeliveries: List>, + ) + + private val states = ConcurrentHashMap>() + private val nextAttemptId = AtomicLong() + + /** Marks the UID active and atomically claims a new attempt when eligible. */ + fun begin(uid: Int): Long? { + var claimed: Long? = null + states.compute(uid) { _, current -> + when (current) { + null -> + DeliveryState.Sending( + active = true, + attemptId = nextAttemptId.incrementAndGet(), + ) + .also { claimed = it.attemptId } + is DeliveryState.Idle -> + DeliveryState.Sending( + active = true, + attemptId = nextAttemptId.incrementAndGet(), + failureCount = current.failureCount, + lastFailureAt = current.lastFailureAt, + ) + .also { claimed = it.attemptId } + is DeliveryState.Sending -> + if (!current.active) { + DeliveryState.Sending( + active = true, + attemptId = nextAttemptId.incrementAndGet(), + failureCount = current.failureCount, + lastFailureAt = current.lastFailureAt, + ) + .also { claimed = it.attemptId } + } else { + current + } + is DeliveryState.Delivered -> current.copy(active = true) + is DeliveryState.Throttled -> + if (now() >= current.cooldownUntil) { + DeliveryState.Sending( + active = true, + attemptId = nextAttemptId.incrementAndGet(), + failureCount = current.count, + lastFailureAt = current.lastFailureAt, + ) + .also { claimed = it.attemptId } + } else { + current.copy(active = true) + } + } + } + return claimed + } + + fun isCurrentSending(uid: Int, attemptId: Long): Boolean = + (states[uid] as? DeliveryState.Sending)?.let { it.active && it.attemptId == attemptId } == true + + /** Completes a non-delivery path without touching a newer state. */ + fun finish(uid: Int, attemptId: Long) { + states.computeIfPresent(uid) { _, current -> + val sending = current as? DeliveryState.Sending ?: return@computeIfPresent current + if (sending.attemptId != attemptId) return@computeIfPresent current + if (!sending.active && sending.failureCount == 0) return@computeIfPresent null + DeliveryState.Idle( + active = sending.active, + lastAttemptId = attemptId, + failureCount = sending.failureCount, + lastFailureAt = sending.lastFailureAt, + ) + } + } + + /** Publishes a successful provider only if this worker still owns the UID. */ + fun commitSuccess( + uid: Int, + attemptId: Long, + provider: Provider, + recipient: Recipient, + ): Boolean { + var accepted = false + states.computeIfPresent(uid) { _, current -> + val sending = current as? DeliveryState.Sending ?: return@computeIfPresent current + if (!sending.active || sending.attemptId != attemptId) return@computeIfPresent current + accepted = true + DeliveryState.Delivered( + active = true, + attemptId = attemptId, + provider = provider, + recipient = recipient, + ) + } + return accepted + } + + /** Records a result only for this attempt or an invalidated tombstone for this same attempt. */ + fun recordFailure(uid: Int, attemptId: Long): Boolean { + var crossed = false + val timestamp = now() + states.computeIfPresent(uid) { _, current -> + val previousCount: Int + val previousAt: Long + val active: Boolean + when (current) { + is DeliveryState.Sending -> { + if (current.attemptId != attemptId) return@computeIfPresent current + previousCount = current.failureCount + previousAt = current.lastFailureAt + active = current.active + } + is DeliveryState.Idle -> { + if (current.lastAttemptId != attemptId) return@computeIfPresent current + previousCount = current.failureCount + previousAt = current.lastFailureAt + active = current.active + } + else -> return@computeIfPresent current + } + + val count = + if (previousCount == 0 || timestamp - previousAt >= failureRunMs) { + 1 + } else { + minOf(previousCount + 1, maxConsecutiveFailures) + } + crossed = count == maxConsecutiveFailures && previousCount < count + if (count >= maxConsecutiveFailures) { + DeliveryState.Throttled( + active = active, + lastAttemptId = attemptId, + count = count, + lastFailureAt = timestamp, + cooldownUntil = timestamp + retryCooldownMs, + ) + } else { + DeliveryState.Idle( + active = active, + lastAttemptId = attemptId, + failureCount = count, + lastFailureAt = timestamp, + ) + } + } + return crossed + } + + fun isCurrentDelivery(uid: Int, provider: Provider, recipient: Recipient): Boolean = + (states[uid] as? DeliveryState.Delivered)?.let { + it.provider === provider && it.recipient === recipient + } == true + + internal fun failureCount(uid: Int): Int = + when (val state = states[uid]) { + is DeliveryState.Idle -> state.failureCount + is DeliveryState.Sending -> state.failureCount + is DeliveryState.Throttled -> state.count + is DeliveryState.Delivered, null -> 0 + } + + internal fun isThrottled(uid: Int): Boolean = + (states[uid] as? DeliveryState.Throttled)?.let { now() < it.cooldownUntil } == true + + /** Removes a delivery only when its provider and recipient are still the current pair. */ + fun removeIfCurrentDelivery(uid: Int, provider: Provider, recipient: Recipient): Boolean { + var removed = false + states.computeIfPresent(uid) { _, current -> + val delivered = current as? DeliveryState.Delivered ?: return@computeIfPresent current + if (delivered.provider !== provider || delivered.recipient !== recipient) { + return@computeIfPresent current + } + removed = true + DeliveryState.Idle(active = delivered.active, lastAttemptId = delivered.attemptId) + } + return removed + } + + /** + * Invalidates one module generation while preserving active UID observations and failure runs. + * The returned provider pairs are unlinked by the caller outside the map operation. + */ + fun invalidateMatching(predicate: (Int) -> Boolean): Invalidation { + val redelivery = mutableSetOf() + val removed = mutableListOf>() + states.keys.filter(predicate).forEach { uid -> + states.computeIfPresent(uid) { _, current -> + if (current.active) redelivery += uid + when (current) { + is DeliveryState.Delivered -> { + removed += RemovedDelivery(uid, current.provider, current.recipient) + DeliveryState.Idle(active = true, lastAttemptId = current.attemptId) + } + is DeliveryState.Sending -> + DeliveryState.Idle( + active = current.active, + lastAttemptId = current.attemptId, + failureCount = current.failureCount, + lastFailureAt = current.lastFailureAt, + ) + is DeliveryState.Idle -> current + is DeliveryState.Throttled -> current + } + } + } + return Invalidation(redelivery, removed) + } + + /** Marks a UID gone. A running worker gets a tombstone; an idle UID is removed immediately. */ + fun invalidateGone(uid: Int): RemovedDelivery? { + var removed: RemovedDelivery? = null + states.computeIfPresent(uid) { _, current -> + when (current) { + is DeliveryState.Sending -> current.copy(active = false) + is DeliveryState.Delivered -> { + removed = RemovedDelivery(uid, current.provider, current.recipient) + null + } + is DeliveryState.Idle -> + if (current.failureCount > 0) current.copy(active = false) else null + is DeliveryState.Throttled -> current.copy(active = false) + } + } + return removed + } +} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt index 59f6b3ef2..c015db617 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt @@ -15,7 +15,6 @@ import io.github.libxposed.service.IXposedService import java.io.Serializable import java.util.Collections import java.util.WeakHashMap -import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit @@ -59,66 +58,21 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. private const val RELOAD_TIMEOUT_SECONDS = 30L /** - * The uids whose module app is holding a binder we handed it. - * - * A binder belongs to the *process* that received it, but a uid can outlive any one of its - * processes: an app with a `:remote` or crash-handler process, or a shared user id, keeps its - * uid alive when the process we served is reaped, so no [uidGone] arrives and the replacement - * process would be refused here forever. That was unreachable while the reference below pinned - * every module app at foreground priority and nothing ever reaped it. Giving the reference back - * makes it the ordinary case, so entries are also dropped by [linkDelivery] when the process - * that took the binder dies. - * - * Recorded on a *successful* send rather than on the attempt: a failed send leaves nothing on - * the other side, and treating it as delivered meant the one module that most needed another - * attempt never got one. - */ - private val uidSet = ConcurrentHashMap.newKeySet() - - /** Coordinates active delivery attempts and invalidates work after a uid/cache reset. */ - private val deliveryAttempts = DeliveryAttemptTracker() - - /** - * What tells [uidSet] that a delivery is over: the provider binder we spoke to, and the - * recipient watching it. Held because a `DeathRecipient` nothing references is one the runtime - * may collect before it ever fires. + * Coordinates active UIDs, in-flight attempts, and provider/death-recipient ownership under + * one lock. A failed send remains an active UID so a later module-generation swap can retry it + * even though it has no successful delivery entry. */ - private val deliveries = DeliveryRegistry() + private val deliveryState = DeliveryStateStore() private val serviceMap = Collections.synchronizedMap(WeakHashMap()) - /** - * Consecutive failed sends per uid, and when the last one was. - * - * A module app that dies before it can publish its provider is not a transient failure to be - * retried at the speed of the uid observer. It happens — an app that crashes on start, or one - * another module deliberately kills, as in #889 where a module in a third module's scope took - * its host down on every launch — and the delivery below *starts the process*, so retrying is - * not a passive act: it feeds the very loop it is failing on. Fourteen starts in seventy-six - * seconds were observed that way, six of them ours. - * - * Per uid and not per package, because `getModuleByUid` matches on the app id: one module - * installed for two users is one `LoadedModule` under two uids, and keying by name would let a - * crash-looping copy in a work profile throttle the healthy copy in user 0, and let either - * one's success wipe the other's run. - * - * Once three failures have piled up the retries are throttled to one per minute — the count is - * held at the ceiling rather than reset by the attempt that the cooldown lets through, or the - * ceiling would simply be re-climbed and three more attempts allowed every minute for ever. A - * run is forgotten after ten minutes without a failure, so an occasional one never accumulates. - * Throttled rather than abandoned, and cleared by the first success, because the app may simply - * have been mid-update or out of memory; a module written off for good on three failures would - * be a worse bug than the one this is fixing. - */ - private val binderFailures = BinderFailureTracker() - // The delivery blocks in getContentProviderExternal until the app publishes its provider or // AMS gives up on it, and it runs from an IUidObserver callback - one binder thread, serving // every uid transition on the device. A module app that never publishes therefore stalls the // delivery of every *other* module's binder behind it: eight and a half seconds, measured, on // a device where one module app was crash-looping. Keep one worker per blocked lookup instead - // of queueing all modules behind a fixed global pool: [deliveryAttempts] deduplicates repeated + // of queueing all modules behind a fixed global pool: [deliveryState] deduplicates repeated // callbacks for a uid, and uidGone() invalidation intentionally lets a replacement proceed // without waiting for the stale lookup to return. private val binderExecutor = @@ -131,35 +85,23 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. fun uidClear(moduleAppIds: Set): Set { if (moduleAppIds.isEmpty()) return emptySet() val belongsToChangedModule = { uid: Int -> uid % PER_USER_RANGE in moduleAppIds } - val invalidatedUids = mutableSetOf() - - invalidatedUids += deliveryAttempts.invalidateMatching(belongsToChangedModule) - uidSet.toList().forEach { uid -> - if (belongsToChangedModule(uid) && uidSet.remove(uid)) invalidatedUids += uid - } - deliveries.removeMatching(belongsToChangedModule).forEach { (uid, delivery) -> - invalidatedUids += uid - uidSet.remove(uid) + val invalidation = deliveryState.invalidateMatching(belongsToChangedModule) + invalidation.removedDeliveries.forEach { delivery -> runCatching { delivery.provider.unlinkToDeath(delivery.recipient, 0) } } - return invalidatedUids + return invalidation.redeliveryUids } fun uidStarts(uid: Int) { - if (uid in uidSet) return - val attempt = deliveryAttempts.begin(uid) ?: return + val attempt = deliveryState.begin(uid) ?: return val module = ConfigCache.getModuleByUid(uid) if (module?.code?.legacy != false) { - deliveryAttempts.finish(uid, attempt) - return - } - if (isThrottled(uid)) { - deliveryAttempts.finish(uid, attempt) + deliveryState.finish(uid, attempt) return } val service = serviceMap.getOrPut(module) { ModuleAppService(module) } // Off the observer thread, and never inline: see [binderExecutor]. Caught, because an - // attempt left in [deliveryAttempts] by a rejected submission is one this never looks at + // attempt left in [deliveryState] by a rejected submission is one this never looks at // again. runCatching { binderExecutor.execute { @@ -167,32 +109,31 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. // Do not start an obsolete lookup after uidGone() or a cache generation reset has // already made this attempt inert; the post-send check below still handles // invalidation while the lookup is in flight. - if (deliveryAttempts.isCurrent(uid, attempt)) { + if (deliveryState.isCurrentSending(uid, attempt)) { val delivered = service.sendBinder(uid) if (delivered == null) { // A uid can disappear while AMS is waiting for its provider. That makes this // attempt stale, but it is still a failed launch and must feed the retry // throttle; otherwise a crash-looping app can evade the three-failure limit // by dying at exactly this point. - recordFailure(uid, module.packageName) - } else if (deliveryAttempts.isCurrent(uid, attempt)) { - binderFailures.clear(uid) + recordFailure(uid, module.packageName, attempt) + } else if (deliveryState.isCurrentSending(uid, attempt)) { linkDelivery(uid, delivered, attempt) } } } finally { - deliveryAttempts.finish(uid, attempt) + deliveryState.finish(uid, attempt) } } } .onFailure { - deliveryAttempts.finish(uid, attempt) + deliveryState.finish(uid, attempt) Log.w(TAG, "Could not schedule the binder delivery for ${module.packageName}", it) } } /** - * Watches the process that took the binder, so [uidSet] forgets the uid when it dies. + * Watches the process that took the binder, so the delivery state forgets the uid when it dies. * * [uidGone] is not enough on its own — it only fires when the *uid* has no processes left — * and this is what makes a second delivery to a restarted module app possible. A death @@ -202,43 +143,35 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. private fun linkDelivery( uid: Int, provider: IBinder, - attempt: DeliveryAttemptTracker.Attempt, + attemptId: Long, ) { - if (!deliveryAttempts.isCurrent(uid, attempt)) return - lateinit var recipient: IBinder.DeathRecipient recipient = IBinder.DeathRecipient { - if (deliveries.removeIfCurrent(uid, provider, recipient)) { - uidSet.remove(uid) - } + deliveryState.removeIfCurrentDelivery(uid, provider, recipient) } - deliveries.put(uid, provider, recipient)?.let { previous -> - runCatching { previous.provider.unlinkToDeath(previous.recipient, 0) } - } - uidSet.add(uid) + if (!deliveryState.commitSuccess(uid, attemptId, provider, recipient)) return runCatching { provider.linkToDeath(recipient, 0) - if (!deliveryAttempts.isCurrent(uid, attempt) || - !deliveries.isCurrent(uid, provider, recipient)) { - if (deliveries.removeIfCurrent(uid, provider, recipient)) uidSet.remove(uid) - runCatching { provider.unlinkToDeath(recipient, 0) } + if (!deliveryState.isCurrentDelivery(uid, provider, recipient)) { + if (deliveryState.removeIfCurrentDelivery(uid, provider, recipient)) { + runCatching { provider.unlinkToDeath(recipient, 0) } + } } } // Already dead, which is an answer in itself: whatever took the binder is gone, so the // uid must not stay marked as served. .onFailure { - if (deliveries.removeIfCurrent(uid, provider, recipient)) uidSet.remove(uid) + deliveryState.removeIfCurrentDelivery(uid, provider, recipient) runCatching { provider.unlinkToDeath(recipient, 0) } } } - /** True while a uid has spent its attempts and its cooldown has not elapsed. */ - private fun isThrottled(uid: Int): Boolean { - return binderFailures.isThrottled(uid) - } - - private fun recordFailure(uid: Int, modulePkg: String) { - val crossed = binderFailures.recordFailure(uid) + private fun recordFailure( + uid: Int, + modulePkg: String, + attemptId: Long, + ) { + val crossed = deliveryState.recordFailure(uid, attemptId) // Once, on the way past the ceiling. The failures themselves are already logged one by one // in sendBinder; what is worth saying here is that we have stopped trying, which is the part // a reader chasing a module that never receives its service cannot otherwise see. @@ -251,13 +184,11 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. } fun uidGone(uid: Int) { - deliveryAttempts.invalidate(uid) - uidSet.remove(uid) // A send that never returns — `provider.call` runs the module's own onServiceBind, with no // deadline — would otherwise leave the uid here for the life of the daemon, and every later - // delivery for it refused at the top of uidStarts. The generation invalidation above makes - // a late return inert and releases the uid for a replacement attempt. - deliveries.remove(uid)?.let { delivery -> + // delivery for it refused at the top of uidStarts. The lifecycle invalidation makes a late + // return inert and releases the uid for a replacement attempt. + deliveryState.invalidateGone(uid)?.let { delivery -> runCatching { delivery.provider.unlinkToDeath(delivery.recipient, 0) } } } diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt deleted file mode 100644 index c453b4db4..000000000 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/BinderFailureTrackerTest.kt +++ /dev/null @@ -1,55 +0,0 @@ -package org.matrix.vector.daemon.ipc - -import org.junit.Assert.assertEquals -import org.junit.Assert.assertFalse -import org.junit.Assert.assertTrue -import org.junit.Test - -class BinderFailureTrackerTest { - - @Test - fun failureDuringUidChurnStillReachesThrottle() { - var now = 0L - val failures = - BinderFailureTracker( - now = { now }, - maxConsecutiveFailures = 3, - retryCooldownMs = 100, - failureRunMs = 1_000, - ) - val attempts = DeliveryAttemptTracker() - val attempt = attempts.begin(42) ?: error("attempt was not created") - - assertFalse(failures.recordFailure(42)) - attempts.invalidate(42) - now = 1 - assertFalse(failures.recordFailure(42)) - now = 2 - assertTrue(failures.recordFailure(42)) - - assertEquals(3, failures.count(42)) - assertFalse(attempts.isCurrent(42, attempt)) - assertTrue(failures.isThrottled(42)) - } - - @Test - fun cooldownAttemptDoesNotResetTheFailureRun() { - var now = 0L - val failures = - BinderFailureTracker( - now = { now }, - maxConsecutiveFailures = 3, - retryCooldownMs = 100, - failureRunMs = 1_000, - ) - - repeat(3) { failures.recordFailure(7) } - now = 100 - assertFalse(failures.isThrottled(7)) - failures.recordFailure(7) - - assertEquals(3, failures.count(7)) - now = 101 - assertTrue(failures.isThrottled(7)) - } -} diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt deleted file mode 100644 index 395dc3236..000000000 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTrackerTest.kt +++ /dev/null @@ -1,86 +0,0 @@ -package org.matrix.vector.daemon.ipc - -import org.junit.Assert.assertFalse -import org.junit.Assert.assertNull -import org.junit.Assert.assertNotEquals -import org.junit.Assert.assertTrue -import org.junit.Test - -class DeliveryAttemptTrackerTest { - - @Test - fun duplicateStartsDoNotInvalidateTheActiveAttempt() { - val tracker = DeliveryAttemptTracker() - val first = tracker.begin(42) ?: error("first attempt was not created") - - assertNull(tracker.begin(42)) - assertTrue(tracker.isCurrent(42, first)) - } - - @Test - fun uidGoneInvalidatesOldWorkAndKeepsReplacementOwnership() { - val tracker = DeliveryAttemptTracker() - val first = tracker.begin(42) ?: error("first attempt was not created") - - tracker.invalidate(42) - val replacement = tracker.begin(42) ?: error("replacement attempt was not created") - - assertNotEquals(first, replacement) - assertFalse(tracker.isCurrent(42, first)) - assertTrue(tracker.isCurrent(42, replacement)) - - tracker.finish(42, first) - assertTrue(tracker.isCurrent(42, replacement)) - } - - @Test - fun clearInvalidatesEveryOldAttempt() { - val tracker = DeliveryAttemptTracker() - val first = tracker.begin(1) ?: error("first attempt was not created") - val second = tracker.begin(2) ?: error("second attempt was not created") - - tracker.clear() - - assertFalse(tracker.isCurrent(1, first)) - assertFalse(tracker.isCurrent(2, second)) - val replacement = tracker.begin(1) ?: error("replacement attempt was not created") - assertTrue(tracker.isCurrent(1, replacement)) - } - - @Test - fun staleCompletionAfterClearCannotReleaseReplacementOwnership() { - val tracker = DeliveryAttemptTracker() - val first = tracker.begin(7) ?: error("first attempt was not created") - - tracker.clear() - val replacement = tracker.begin(7) ?: error("replacement attempt was not created") - - tracker.finish(7, first) - - assertTrue(tracker.isCurrent(7, replacement)) - } - - @Test - fun scopedInvalidationLeavesOtherModuleAttemptCurrent() { - val tracker = DeliveryAttemptTracker() - val changedModule = tracker.begin(10042) ?: error("changed-module attempt was not created") - val otherModule = tracker.begin(20043) ?: error("other-module attempt was not created") - - tracker.invalidateMatching { it == 10042 } - - assertFalse(tracker.isCurrent(10042, changedModule)) - assertTrue(tracker.isCurrent(20043, otherModule)) - } - - @Test - fun queuedObsoleteAttemptDoesNotStartProviderLookup() { - val tracker = DeliveryAttemptTracker() - val attempt = tracker.begin(42) ?: error("attempt was not created") - tracker.invalidate(42) - - var lookupStarted = false - if (tracker.isCurrent(42, attempt)) lookupStarted = true - - assertFalse(lookupStarted) - } -} diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt deleted file mode 100644 index f6b38e3b7..000000000 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryRegistryTest.kt +++ /dev/null @@ -1,40 +0,0 @@ -package org.matrix.vector.daemon.ipc - -import org.junit.Assert.assertFalse -import org.junit.Assert.assertNotNull -import org.junit.Assert.assertTrue -import org.junit.Test - -class DeliveryRegistryTest { - - @Test - fun lateDeathOfOldProviderCannotRemoveReplacement() { - val registry = DeliveryRegistry() - val oldProvider = Any() - val oldRecipient = Any() - val newProvider = Any() - val newRecipient = Any() - - registry.put(42, oldProvider, oldRecipient) - registry.put(42, newProvider, newRecipient) - - assertFalse(registry.removeIfCurrent(42, oldProvider, oldRecipient)) - assertTrue(registry.isCurrent(42, newProvider, newRecipient)) - assertNotNull(registry.remove(42)) - } - - @Test - fun scopedRemovalDoesNotTouchOtherUids() { - val registry = DeliveryRegistry() - val provider = Any() - val recipient = Any() - registry.put(10042, provider, recipient) - registry.put(20043, provider, recipient) - - val removed = registry.removeMatching { it == 10042 } - - assertTrue(removed.any { it.first == 10042 }) - assertFalse(registry.isCurrent(10042, provider, recipient)) - assertTrue(registry.isCurrent(20043, provider, recipient)) - } -} diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt new file mode 100644 index 000000000..21c52e6e9 --- /dev/null +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt @@ -0,0 +1,94 @@ +package org.matrix.vector.daemon.ipc + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertTrue +import org.junit.Test + +class DeliveryStateTest { + + @Test + fun staleSuccessCannotPublishAfterUidGoneAndReplacement() { + val states = DeliveryStateStore(now = { 0L }) + val first = states.begin(42) ?: error("first attempt was not created") + + states.invalidateGone(42) + val replacement = states.begin(42) ?: error("replacement attempt was not created") + + assertFalse(states.commitSuccess(42, first, Any(), Any())) + assertTrue(states.isCurrentSending(42, replacement)) + } + + @Test + fun staleFailureCannotOverwriteReplacementSuccess() { + val states = DeliveryStateStore(now = { 0L }) + val first = states.begin(7) ?: error("first attempt was not created") + states.invalidateMatching { it == 7 } + val replacement = states.begin(7) ?: error("replacement attempt was not created") + + val provider = Any() + val recipient = Any() + assertTrue(states.commitSuccess(7, replacement, provider, recipient)) + assertFalse(states.recordFailure(7, first)) + assertTrue(states.isCurrentDelivery(7, provider, recipient)) + } + + @Test + fun failedActiveUidIsRedeliveredOnGenerationChange() { + val states = DeliveryStateStore(now = { 0L }) + val attempt = states.begin(10042) ?: error("attempt was not created") + states.recordFailure(10042, attempt) + states.finish(10042, attempt) + + val invalidation = states.invalidateMatching { it == 10042 } + + assertEquals(setOf(10042), invalidation.redeliveryUids) + val replacement = states.begin(10042) ?: error("replacement attempt was not created") + assertNotNull(replacement) + } + + @Test + fun oldDeathRecipientCannotRemoveReplacementDelivery() { + val states = DeliveryStateStore(now = { 0L }) + val firstAttempt = states.begin(11) ?: error("first attempt was not created") + val oldProvider = Any() + val oldRecipient = Any() + assertTrue(states.commitSuccess(11, firstAttempt, oldProvider, oldRecipient)) + + states.invalidateMatching { it == 11 } + val replacementAttempt = states.begin(11) ?: error("replacement attempt was not created") + val newProvider = Any() + val newRecipient = Any() + assertTrue(states.commitSuccess(11, replacementAttempt, newProvider, newRecipient)) + + assertFalse(states.removeIfCurrentDelivery(11, oldProvider, oldRecipient)) + assertTrue(states.isCurrentDelivery(11, newProvider, newRecipient)) + } + + @Test + fun threeFailuresEnterCooldownAndCooldownAttemptKeepsRun() { + var now = 0L + val states = + DeliveryStateStore( + now = { now }, + retryCooldownMs = 100, + failureRunMs = 1_000, + ) + + repeat(3) { + val attempt = states.begin(9) ?: error("attempt was not created") + states.recordFailure(9, attempt) + states.finish(9, attempt) + } + + now = 100 + val cooldownAttempt = states.begin(9) ?: error("cooldown attempt was not created") + states.recordFailure(9, cooldownAttempt) + states.finish(9, cooldownAttempt) + now = 101 + + assertEquals(3, states.failureCount(9)) + assertTrue(states.isThrottled(9)) + } +} From dccaba6a2c4ea40695370f480beca375ec53e3fa Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Sun, 9 Aug 2026 22:23:23 +0800 Subject: [PATCH 5/7] test: cover stale failure after uid disappearance Signed-off-by: LIghtJUNction --- .../matrix/vector/daemon/ipc/DeliveryStateTest.kt | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt index 21c52e6e9..8031df754 100644 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt +++ b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt @@ -20,6 +20,18 @@ class DeliveryStateTest { assertTrue(states.isCurrentSending(42, replacement)) } + @Test + fun failureAfterUidGoneIsCountedWithoutPublishing() { + val states = DeliveryStateStore(now = { 0L }) + val attempt = states.begin(43) ?: error("attempt was not created") + + states.invalidateGone(43) + + assertFalse(states.commitSuccess(43, attempt, Any(), Any())) + assertFalse(states.recordFailure(43, attempt)) + assertEquals(1, states.failureCount(43)) + } + @Test fun staleFailureCannotOverwriteReplacementSuccess() { val states = DeliveryStateStore(now = { 0L }) From d68a0c4d8454ac3d9b86216249ea2888195286c4 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Sun, 9 Aug 2026 22:24:56 +0800 Subject: [PATCH 6/7] docs: clarify atomic delivery state ownership Signed-off-by: LIghtJUNction --- .../kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt index c015db617..5e1a0f855 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt @@ -58,9 +58,9 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. private const val RELOAD_TIMEOUT_SECONDS = 30L /** - * Coordinates active UIDs, in-flight attempts, and provider/death-recipient ownership under - * one lock. A failed send remains an active UID so a later module-generation swap can retry it - * even though it has no successful delivery entry. + * Coordinates active UIDs, in-flight attempts, and provider/death-recipient ownership through + * one per-UID atomic state transition. A failed send remains an active UID so a later + * module-generation swap can retry it even though it has no successful delivery entry. */ private val deliveryState = DeliveryStateStore() From f6d83fddc89576668f6fd7d81d1fcf7f1500d86a Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Mon, 10 Aug 2026 15:43:23 +0800 Subject: [PATCH 7/7] fix: minimize module binder delivery ownership Signed-off-by: LIghtJUNction --- .github/workflows/core.yml | 2 +- daemon/build.gradle.kts | 1 - .../matrix/vector/daemon/data/ConfigCache.kt | 36 +-- .../matrix/vector/daemon/ipc/DeliveryState.kt | 298 ------------------ .../vector/daemon/ipc/ModuleAppService.kt | 258 ++++++++++----- .../vector/daemon/data/ConfigCacheTest.kt | 32 -- .../vector/daemon/ipc/DeliveryStateTest.kt | 106 ------- 7 files changed, 184 insertions(+), 549 deletions(-) delete mode 100644 daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt delete mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt delete mode 100644 daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt diff --git a/.github/workflows/core.yml b/.github/workflows/core.yml index ab5ee5a0f..89b65713d 100644 --- a/.github/workflows/core.yml +++ b/.github/workflows/core.yml @@ -123,7 +123,7 @@ jobs: - name: Build with Gradle run: | - ./gradlew :daemon:testDebugUnitTest zipAll + ./gradlew zipAll - name: Prepare artifact if: success() diff --git a/daemon/build.gradle.kts b/daemon/build.gradle.kts index 52fe1002d..499e80978 100644 --- a/daemon/build.gradle.kts +++ b/daemon/build.gradle.kts @@ -150,5 +150,4 @@ dependencies { implementation(projects.services.managerService) compileOnly(libs.androidx.annotation) compileOnly(projects.hiddenapi.stubs) - testImplementation("junit:junit:4.13.2") } diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt index 024dfcd97..e81bf8310 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/data/ConfigCache.kt @@ -26,25 +26,6 @@ import org.matrix.vector.daemon.utils.getRealUsers private const val TAG = "VectorConfigCache" -/** Returns app ids whose [LoadedModule] object changed between two published generations. */ -internal fun moduleGenerationAppIds( - oldModules: Map, - newModules: Map, -): Set { - val changed = mutableSetOf() - oldModules.forEach { (packageName, oldModule) -> - if (newModules[packageName] !== oldModule && oldModule.appId >= 0) { - changed += oldModule.appId - } - } - newModules.forEach { (packageName, newModule) -> - if (oldModules[packageName] !== newModule && newModule.appId >= 0) { - changed += newModule.appId - } - } - return changed -} - object ConfigCache { // Module preference operations are delegated to PreferenceStore // Writable operations of modules are delegated to ModuleDatabase @@ -441,31 +422,18 @@ object ConfigCache { // // `oldState` is still the right thing to *read* from above: reusing an already-parsed module // is a decision about what was loaded when the rebuild started. - val (publishedModules, changedModuleAppIds) = synchronized(this) { - // Compare with the state being replaced, not the snapshot from the start of this rebuild. - // A concurrent writer may have updated the state while package and APK queries ran above. - val currentModules = state.modules - val changedAppIds = moduleGenerationAppIds(currentModules, newModules) + synchronized(this) { state = state.copy(modules = newModules, scopes = newScopes, unloadable = unloadable) // Swapped here rather than sixty lines earlier, so that the claims and the modules they // belong to become visible together. Between the two assignments a reader could see the new // static scopes against the old module set. staticScopes = newStaticScopes - currentModules to changedAppIds - } - - if (changedModuleAppIds.isNotEmpty()) { - // Only the module generations that changed need to release their provider binders. Keeping - // other modules' deliveries and failure runs avoids re-feeding an unrelated crash loop. - // A module that is already running may not emit another uid transition, so immediately - // offer the affected uids a delivery for the newly published generation. - ModuleAppService.uidClear(changedModuleAppIds).forEach { ModuleAppService.uidStarts(it) } } Log.d(TAG, "Cache Update Complete. Map Swap successful.") // Targets are removed only after the module set has been published. - (publishedModules.keys - newModules.keys).forEach { + (oldState.modules.keys - newModules.keys).forEach { FrameworkService.forgetHotReloadTargets(it) } FrameworkService.backfillLoadedVersions() diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt deleted file mode 100644 index 479f0ecbb..000000000 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryState.kt +++ /dev/null @@ -1,298 +0,0 @@ -package org.matrix.vector.daemon.ipc - -import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.atomic.AtomicLong - -/** - * The complete lifecycle of one module-app UID. - * - * Failure accounting and the active bit live in every state rather than in side maps. The - * [DeliveryStateStore] changes one UID only through [ConcurrentHashMap.compute], so a worker result - * cannot pass an ownership check and then be invalidated before its state is committed. - */ -internal sealed interface DeliveryState { - val active: Boolean - val lastAttemptId: Long - - data class Idle( - override val active: Boolean, - override val lastAttemptId: Long = 0L, - val failureCount: Int = 0, - val lastFailureAt: Long = 0L, - ) : DeliveryState - - data class Sending( - override val active: Boolean, - val attemptId: Long, - val failureCount: Int = 0, - val lastFailureAt: Long = 0L, - ) : DeliveryState { - override val lastAttemptId: Long - get() = attemptId - } - - data class Delivered( - override val active: Boolean, - val attemptId: Long, - val provider: Provider, - val recipient: Recipient, - ) : DeliveryState { - override val lastAttemptId: Long - get() = attemptId - } - - data class Throttled( - override val active: Boolean, - override val lastAttemptId: Long, - val count: Int, - val lastFailureAt: Long, - val cooldownUntil: Long, - ) : DeliveryState -} - -/** - * Atomic state machine for module-app binder delivery. - * - * There is deliberately one map and no companion attempt/registry/failure maps. A tombstoned - * [Sending] state is retained after [uidGone] until its worker completes, which lets a failure be - * counted while still preventing that old worker from publishing a replacement. - */ -internal class DeliveryStateStore( - private val now: () -> Long = { android.os.SystemClock.elapsedRealtime() }, - private val maxConsecutiveFailures: Int = 3, - private val retryCooldownMs: Long = 60_000L, - private val failureRunMs: Long = 10 * retryCooldownMs, -) { - - internal data class RemovedDelivery( - val uid: Int, - val provider: Provider, - val recipient: Recipient, - ) - - internal data class Invalidation( - val redeliveryUids: Set, - val removedDeliveries: List>, - ) - - private val states = ConcurrentHashMap>() - private val nextAttemptId = AtomicLong() - - /** Marks the UID active and atomically claims a new attempt when eligible. */ - fun begin(uid: Int): Long? { - var claimed: Long? = null - states.compute(uid) { _, current -> - when (current) { - null -> - DeliveryState.Sending( - active = true, - attemptId = nextAttemptId.incrementAndGet(), - ) - .also { claimed = it.attemptId } - is DeliveryState.Idle -> - DeliveryState.Sending( - active = true, - attemptId = nextAttemptId.incrementAndGet(), - failureCount = current.failureCount, - lastFailureAt = current.lastFailureAt, - ) - .also { claimed = it.attemptId } - is DeliveryState.Sending -> - if (!current.active) { - DeliveryState.Sending( - active = true, - attemptId = nextAttemptId.incrementAndGet(), - failureCount = current.failureCount, - lastFailureAt = current.lastFailureAt, - ) - .also { claimed = it.attemptId } - } else { - current - } - is DeliveryState.Delivered -> current.copy(active = true) - is DeliveryState.Throttled -> - if (now() >= current.cooldownUntil) { - DeliveryState.Sending( - active = true, - attemptId = nextAttemptId.incrementAndGet(), - failureCount = current.count, - lastFailureAt = current.lastFailureAt, - ) - .also { claimed = it.attemptId } - } else { - current.copy(active = true) - } - } - } - return claimed - } - - fun isCurrentSending(uid: Int, attemptId: Long): Boolean = - (states[uid] as? DeliveryState.Sending)?.let { it.active && it.attemptId == attemptId } == true - - /** Completes a non-delivery path without touching a newer state. */ - fun finish(uid: Int, attemptId: Long) { - states.computeIfPresent(uid) { _, current -> - val sending = current as? DeliveryState.Sending ?: return@computeIfPresent current - if (sending.attemptId != attemptId) return@computeIfPresent current - if (!sending.active && sending.failureCount == 0) return@computeIfPresent null - DeliveryState.Idle( - active = sending.active, - lastAttemptId = attemptId, - failureCount = sending.failureCount, - lastFailureAt = sending.lastFailureAt, - ) - } - } - - /** Publishes a successful provider only if this worker still owns the UID. */ - fun commitSuccess( - uid: Int, - attemptId: Long, - provider: Provider, - recipient: Recipient, - ): Boolean { - var accepted = false - states.computeIfPresent(uid) { _, current -> - val sending = current as? DeliveryState.Sending ?: return@computeIfPresent current - if (!sending.active || sending.attemptId != attemptId) return@computeIfPresent current - accepted = true - DeliveryState.Delivered( - active = true, - attemptId = attemptId, - provider = provider, - recipient = recipient, - ) - } - return accepted - } - - /** Records a result only for this attempt or an invalidated tombstone for this same attempt. */ - fun recordFailure(uid: Int, attemptId: Long): Boolean { - var crossed = false - val timestamp = now() - states.computeIfPresent(uid) { _, current -> - val previousCount: Int - val previousAt: Long - val active: Boolean - when (current) { - is DeliveryState.Sending -> { - if (current.attemptId != attemptId) return@computeIfPresent current - previousCount = current.failureCount - previousAt = current.lastFailureAt - active = current.active - } - is DeliveryState.Idle -> { - if (current.lastAttemptId != attemptId) return@computeIfPresent current - previousCount = current.failureCount - previousAt = current.lastFailureAt - active = current.active - } - else -> return@computeIfPresent current - } - - val count = - if (previousCount == 0 || timestamp - previousAt >= failureRunMs) { - 1 - } else { - minOf(previousCount + 1, maxConsecutiveFailures) - } - crossed = count == maxConsecutiveFailures && previousCount < count - if (count >= maxConsecutiveFailures) { - DeliveryState.Throttled( - active = active, - lastAttemptId = attemptId, - count = count, - lastFailureAt = timestamp, - cooldownUntil = timestamp + retryCooldownMs, - ) - } else { - DeliveryState.Idle( - active = active, - lastAttemptId = attemptId, - failureCount = count, - lastFailureAt = timestamp, - ) - } - } - return crossed - } - - fun isCurrentDelivery(uid: Int, provider: Provider, recipient: Recipient): Boolean = - (states[uid] as? DeliveryState.Delivered)?.let { - it.provider === provider && it.recipient === recipient - } == true - - internal fun failureCount(uid: Int): Int = - when (val state = states[uid]) { - is DeliveryState.Idle -> state.failureCount - is DeliveryState.Sending -> state.failureCount - is DeliveryState.Throttled -> state.count - is DeliveryState.Delivered, null -> 0 - } - - internal fun isThrottled(uid: Int): Boolean = - (states[uid] as? DeliveryState.Throttled)?.let { now() < it.cooldownUntil } == true - - /** Removes a delivery only when its provider and recipient are still the current pair. */ - fun removeIfCurrentDelivery(uid: Int, provider: Provider, recipient: Recipient): Boolean { - var removed = false - states.computeIfPresent(uid) { _, current -> - val delivered = current as? DeliveryState.Delivered ?: return@computeIfPresent current - if (delivered.provider !== provider || delivered.recipient !== recipient) { - return@computeIfPresent current - } - removed = true - DeliveryState.Idle(active = delivered.active, lastAttemptId = delivered.attemptId) - } - return removed - } - - /** - * Invalidates one module generation while preserving active UID observations and failure runs. - * The returned provider pairs are unlinked by the caller outside the map operation. - */ - fun invalidateMatching(predicate: (Int) -> Boolean): Invalidation { - val redelivery = mutableSetOf() - val removed = mutableListOf>() - states.keys.filter(predicate).forEach { uid -> - states.computeIfPresent(uid) { _, current -> - if (current.active) redelivery += uid - when (current) { - is DeliveryState.Delivered -> { - removed += RemovedDelivery(uid, current.provider, current.recipient) - DeliveryState.Idle(active = true, lastAttemptId = current.attemptId) - } - is DeliveryState.Sending -> - DeliveryState.Idle( - active = current.active, - lastAttemptId = current.attemptId, - failureCount = current.failureCount, - lastFailureAt = current.lastFailureAt, - ) - is DeliveryState.Idle -> current - is DeliveryState.Throttled -> current - } - } - } - return Invalidation(redelivery, removed) - } - - /** Marks a UID gone. A running worker gets a tombstone; an idle UID is removed immediately. */ - fun invalidateGone(uid: Int): RemovedDelivery? { - var removed: RemovedDelivery? = null - states.computeIfPresent(uid) { _, current -> - when (current) { - is DeliveryState.Sending -> current.copy(active = false) - is DeliveryState.Delivered -> { - removed = RemovedDelivery(uid, current.provider, current.recipient) - null - } - is DeliveryState.Idle -> - if (current.failureCount > 0) current.copy(active = false) else null - is DeliveryState.Throttled -> current.copy(active = false) - } - } - return removed - } -} diff --git a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt index 5e1a0f855..592c5d597 100644 --- a/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt +++ b/daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/ModuleAppService.kt @@ -7,6 +7,7 @@ import android.os.Bundle import android.os.IBinder import android.os.ParcelFileDescriptor import android.os.RemoteException +import android.os.SystemClock import android.util.Log import io.github.libxposed.service.HookedProcess import io.github.libxposed.service.IHotReloadCallback @@ -15,6 +16,7 @@ import io.github.libxposed.service.IXposedService import java.io.Serializable import java.util.Collections import java.util.WeakHashMap +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit @@ -58,138 +60,240 @@ class ModuleAppService(private val loadedModule: LoadedModule) : IXposedService. private const val RELOAD_TIMEOUT_SECONDS = 30L /** - * Coordinates active UIDs, in-flight attempts, and provider/death-recipient ownership through - * one per-UID atomic state transition. A failed send remains an active UID so a later - * module-generation swap can retry it even though it has no successful delivery entry. + * The uids whose module app is holding a binder we handed it. + * + * A binder belongs to the *process* that received it, but a uid can outlive any one of its + * processes: an app with a `:remote` or crash-handler process, or a shared user id, keeps its + * uid alive when the process we served is reaped, so no [uidGone] arrives and the replacement + * process would be refused here forever. That was unreachable while the reference below pinned + * every module app at foreground priority and nothing ever reaped it. Giving the reference back + * makes it the ordinary case, so entries are also dropped by [linkDelivery] when the process + * that took the binder dies. + * + * Recorded on a *successful* send rather than on the attempt: a failed send leaves nothing on + * the other side, and treating it as delivered meant the one module that most needed another + * attempt never got one. + */ + private val uidSet = ConcurrentHashMap.newKeySet() + + /** + * Which send is running for a uid right now, so the three observer callbacks agree on one. + * + * The uid alone was not enough to say that. [uidGone] deliberately drops the marker rather + * than wait for a send that may never return, so from that moment a replacement process can + * start a second send while the first is still blocked — and the first, on its way out, would + * remove the marker the second is holding, letting a *third* start behind it, and would then + * commit its own dead process's result over the second's. The value is the attempt that owns + * the uid: a send touches nothing unless the token it was handed is still the one here. + */ + private val sending = ConcurrentHashMap() + + /** + * What tells [uidSet] that a delivery is over: the provider binder we spoke to, and the + * recipient watching it. Held because a `DeathRecipient` nothing references is one the runtime + * may collect before it ever fires. + */ + private val deliveries = ConcurrentHashMap>() + + /** + * Held wherever [uidSet] and [deliveries] change hands: the three places that hand a uid over + * are a send committing its result, [uidGone], and the death of the process that took the + * binder, and each of them reads one of the two fields to decide what to do to the other. + * Both are individually atomic, which is what made the gap between them easy to miss — a send + * that tested its ownership and was then overtaken by [uidGone] before it added the uid left + * the uid marked as served with nothing serving it, and the replacement process was refused at + * the top of [uidStarts] with no later uid edge to come. + * + * Nothing that blocks runs under it: the send itself is outside, and `linkToDeath` is a local + * call. */ - private val deliveryState = DeliveryStateStore() + private val deliveryLock = Any() private val serviceMap = Collections.synchronizedMap(WeakHashMap()) + /** + * Consecutive failed sends per uid, and when the last one was. + * + * A module app that dies before it can publish its provider is not a transient failure to be + * retried at the speed of the uid observer. It happens — an app that crashes on start, or one + * another module deliberately kills, as in #889 where a module in a third module's scope took + * its host down on every launch — and the delivery below *starts the process*, so retrying is + * not a passive act: it feeds the very loop it is failing on. Fourteen starts in seventy-six + * seconds were observed that way, six of them ours. + * + * Per uid and not per package, because `getModuleByUid` matches on the app id: one module + * installed for two users is one `LoadedModule` under two uids, and keying by name would let a + * crash-looping copy in a work profile throttle the healthy copy in user 0, and let either + * one's success wipe the other's run. + * + * Once [MAX_CONSECUTIVE_BINDER_FAILURES] have piled up the retries are throttled to one per + * [BINDER_RETRY_COOLDOWN_MS] — the count is held at the ceiling rather than reset by the + * attempt that the cooldown lets through, or the ceiling would simply be re-climbed and three + * more attempts allowed every minute for ever. A run is forgotten after + * [BINDER_FAILURE_RUN_MS] without a failure, so an occasional one never accumulates. Throttled + * rather than abandoned, and cleared by the first success, because the app may simply have been + * mid-update or out of memory; a module written off for good on three failures would be a worse + * bug than the one this is fixing. + */ + private val binderFailures = ConcurrentHashMap() + + private class FailureRun(val count: Int, val atElapsed: Long) + + private const val MAX_CONSECUTIVE_BINDER_FAILURES = 3 + private const val BINDER_RETRY_COOLDOWN_MS = 60_000L + private const val BINDER_FAILURE_RUN_MS = 10 * BINDER_RETRY_COOLDOWN_MS + // The delivery blocks in getContentProviderExternal until the app publishes its provider or // AMS gives up on it, and it runs from an IUidObserver callback - one binder thread, serving // every uid transition on the device. A module app that never publishes therefore stalls the // delivery of every *other* module's binder behind it: eight and a half seconds, measured, on - // a device where one module app was crash-looping. Keep one worker per blocked lookup instead - // of queueing all modules behind a fixed global pool: [deliveryState] deduplicates repeated - // callbacks for a uid, and uidGone() invalidation intentionally lets a replacement proceed - // without waiting for the stale lookup to return. + // a device where one module app was crash-looping. One thread per module keeps that local. private val binderExecutor = Executors.newCachedThreadPool { r -> Thread(r, "vector-module-binder") } - /** - * Invalidates deliveries for the module generations identified by [moduleAppIds]. Other - * modules keep both their live binder and their retry throttle. - */ - fun uidClear(moduleAppIds: Set): Set { - if (moduleAppIds.isEmpty()) return emptySet() - val belongsToChangedModule = { uid: Int -> uid % PER_USER_RANGE in moduleAppIds } - val invalidation = deliveryState.invalidateMatching(belongsToChangedModule) - invalidation.removedDeliveries.forEach { delivery -> - runCatching { delivery.provider.unlinkToDeath(delivery.recipient, 0) } - } - return invalidation.redeliveryUids + fun uidClear() { + uidSet.clear() } fun uidStarts(uid: Int) { - val attempt = deliveryState.begin(uid) ?: return + if (uid in uidSet) return + // What identifies this attempt for as long as it runs, and what every later step of it is + // tested against: see [sending]. + val attempt = Any() + if (sending.putIfAbsent(uid, attempt) != null) return val module = ConfigCache.getModuleByUid(uid) if (module?.code?.legacy != false) { - deliveryState.finish(uid, attempt) + sending.remove(uid, attempt) + return + } + if (isThrottled(uid)) { + sending.remove(uid, attempt) return } val service = serviceMap.getOrPut(module) { ModuleAppService(module) } - // Off the observer thread, and never inline: see [binderExecutor]. Caught, because an - // attempt left in [deliveryState] by a rejected submission is one this never looks at - // again. + // Off the observer thread, and never inline: see [binderExecutor]. Caught, because a uid + // left in [sending] by a rejected submission is one this never looks at again. runCatching { binderExecutor.execute { try { - // Do not start an obsolete lookup after uidGone() or a cache generation reset has - // already made this attempt inert; the post-send check below still handles - // invalidation while the lookup is in flight. - if (deliveryState.isCurrentSending(uid, attempt)) { - val delivered = service.sendBinder(uid) - if (delivered == null) { - // A uid can disappear while AMS is waiting for its provider. That makes this - // attempt stale, but it is still a failed launch and must feed the retry - // throttle; otherwise a crash-looping app can evade the three-failure limit - // by dying at exactly this point. - recordFailure(uid, module.packageName, attempt) - } else if (deliveryState.isCurrentSending(uid, attempt)) { - linkDelivery(uid, delivered, attempt) + val delivered = service.sendBinder(uid) + if (delivered != null) { + // Only the attempt that still owns the uid may say the module has its service. + // An abandoned one spoke to a process the uid has already outlived, and marking + // the uid served on its word is what refuses the process that replaced it. + synchronized(deliveryLock) { + if (sending[uid] === attempt) { + uidSet.add(uid) + binderFailures.remove(uid) + linkDelivery(uid, delivered) + } } + } else { + // Counted whether or not this attempt still owns the uid, unlike the success + // above: a module app that dies while the platform waits for its provider is + // exactly what the throttle is for, and it is also exactly what takes the + // ownership away. A failure dropped for being stale is one the restart loop + // never has to pay for. + recordFailure(uid, module.packageName) } } finally { - deliveryState.finish(uid, attempt) + sending.remove(uid, attempt) } } } .onFailure { - deliveryState.finish(uid, attempt) + sending.remove(uid, attempt) Log.w(TAG, "Could not schedule the binder delivery for ${module.packageName}", it) } } /** - * Watches the process that took the binder, so the delivery state forgets the uid when it dies. + * Watches the process that took the binder, so [uidSet] forgets the uid when it dies. * * [uidGone] is not enough on its own — it only fires when the *uid* has no processes left — * and this is what makes a second delivery to a restarted module app possible. A death * recipient on a proxy is not a client of anything, so unlike the provider reference it puts * no floor under the process's priority. + * + * Called under [deliveryLock], by the attempt that owns the uid. */ - private fun linkDelivery( - uid: Int, - provider: IBinder, - attemptId: Long, - ) { - lateinit var recipient: IBinder.DeathRecipient - recipient = IBinder.DeathRecipient { - deliveryState.removeIfCurrentDelivery(uid, provider, recipient) - } - if (!deliveryState.commitSuccess(uid, attemptId, provider, recipient)) return + private fun linkDelivery(uid: Int, provider: IBinder) { + val recipient = + object : IBinder.DeathRecipient { + override fun binderDied() { + // This delivery's own entry, not merely this uid's. A death notification for the + // process we served can arrive after a replacement process has taken a binder of its + // own — the notification is queued when the process dies, not when we get to it — + // and forgetting the uid then sends the module a second copy of a service it already + // holds, starting a process to do it. + synchronized(deliveryLock) { + if (deliveries.remove(uid, provider to this)) uidSet.remove(uid) + } + } + } runCatching { - provider.linkToDeath(recipient, 0) - if (!deliveryState.isCurrentDelivery(uid, provider, recipient)) { - if (deliveryState.removeIfCurrentDelivery(uid, provider, recipient)) { - runCatching { provider.unlinkToDeath(recipient, 0) } + provider.linkToDeath(recipient, 0) + deliveries.put(uid, provider to recipient)?.let { (old, previous) -> + runCatching { old.unlinkToDeath(previous, 0) } + } } - } - } - // Already dead, which is an answer in itself: whatever took the binder is gone, so the - // uid must not stay marked as served. - .onFailure { - deliveryState.removeIfCurrentDelivery(uid, provider, recipient) - runCatching { provider.unlinkToDeath(recipient, 0) } - } + // Already dead, which is an answer in itself: whatever took the binder is gone, so the + // uid must not stay marked as served. + .onFailure { uidSet.remove(uid) } } - private fun recordFailure( - uid: Int, - modulePkg: String, - attemptId: Long, - ) { - val crossed = deliveryState.recordFailure(uid, attemptId) + /** True while a uid has spent its attempts and its cooldown has not elapsed. */ + private fun isThrottled(uid: Int): Boolean { + val run = binderFailures[uid] ?: return false + if (run.count < MAX_CONSECUTIVE_BINDER_FAILURES) return false + return SystemClock.elapsedRealtime() - run.atElapsed < BINDER_RETRY_COOLDOWN_MS + } + + private fun recordFailure(uid: Int, modulePkg: String) { + var crossed = false + // Read-modify-write in one step, and not merely as a precaution: an attempt abandoned by + // [uidGone] and the replacement that took the uid from it can both be counting here for the + // same uid, which is the case a plain get-then-put would lose. + binderFailures.compute(uid) { _, previous -> + val now = SystemClock.elapsedRealtime() + val count = + when { + // A run is forgotten only after a long quiet spell, not after one cooldown. Forgetting + // it at the cooldown meant the attempt the cooldown let through reset the count, so + // the ceiling was re-climbed and three more attempts allowed every minute, for ever. + previous == null || now - previous.atElapsed >= BINDER_FAILURE_RUN_MS -> 1 + // Held at the ceiling rather than growing without bound: what the number decides is + // only whether we are throttled, and pinning it here is what makes the cooldown mean + // one attempt rather than another three. + else -> minOf(previous.count + 1, MAX_CONSECUTIVE_BINDER_FAILURES) + } + crossed = count == MAX_CONSECUTIVE_BINDER_FAILURES && (previous?.count ?: 0) < count + FailureRun(count, now) + } // Once, on the way past the ceiling. The failures themselves are already logged one by one // in sendBinder; what is worth saying here is that we have stopped trying, which is the part // a reader chasing a module that never receives its service cannot otherwise see. if (crossed) { Log.w( TAG, - "$modulePkg/$uid failed to take its binder three times in a row; retrying at most once" + - " every 60s") + "$modulePkg/$uid failed to take its binder $MAX_CONSECUTIVE_BINDER_FAILURES times in" + + " a row; retrying at most once every ${BINDER_RETRY_COOLDOWN_MS / 1000}s") } } fun uidGone(uid: Int) { - // A send that never returns — `provider.call` runs the module's own onServiceBind, with no - // deadline — would otherwise leave the uid here for the life of the daemon, and every later - // delivery for it refused at the top of uidStarts. The lifecycle invalidation makes a late - // return inert and releases the uid for a replacement attempt. - deliveryState.invalidateGone(uid)?.let { delivery -> - runCatching { delivery.provider.unlinkToDeath(delivery.recipient, 0) } + synchronized(deliveryLock) { + uidSet.remove(uid) + // A send that never returns — `provider.call` runs the module's own onServiceBind, with no + // deadline — would otherwise leave the uid here for the life of the daemon, and every later + // delivery for it refused at the top of uidStarts. Giving the uid up rather than waiting is + // what lets the process that replaces this one be served at once; the attempt token is what + // stops the send we walked away from committing over it. + sending.remove(uid) + deliveries.remove(uid)?.let { (binder, recipient) -> + runCatching { binder.unlinkToDeath(recipient, 0) } + } } } diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt deleted file mode 100644 index 6a5a062cd..000000000 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/data/ConfigCacheTest.kt +++ /dev/null @@ -1,32 +0,0 @@ -package org.matrix.vector.daemon.data - -import org.junit.Assert.assertEquals -import org.junit.Test -import org.matrix.vector.ipc.LoadedModule - -class ConfigCacheTest { - - @Test - fun onlyReplacedModuleAppIdsAreInvalidated() { - val unchanged = LoadedModule().apply { appId = 10001 } - val oldChanged = LoadedModule().apply { appId = 10002 } - val newChanged = LoadedModule().apply { appId = 10002 } - val added = LoadedModule().apply { appId = 10003 } - - val oldModules = mapOf("unchanged" to unchanged, "changed" to oldChanged) - val newModules = - mapOf("unchanged" to unchanged, "changed" to newChanged, "added" to added) - - assertEquals(setOf(10002, 10003), moduleGenerationAppIds(oldModules, newModules)) - } - - @Test - fun removedModuleAppIdIsInvalidated() { - val removed = LoadedModule().apply { appId = 10004 } - - assertEquals( - setOf(10004), - moduleGenerationAppIds(mapOf("removed" to removed), emptyMap()), - ) - } -} diff --git a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt b/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt deleted file mode 100644 index 8031df754..000000000 --- a/daemon/src/test/kotlin/org/matrix/vector/daemon/ipc/DeliveryStateTest.kt +++ /dev/null @@ -1,106 +0,0 @@ -package org.matrix.vector.daemon.ipc - -import org.junit.Assert.assertEquals -import org.junit.Assert.assertFalse -import org.junit.Assert.assertNotNull -import org.junit.Assert.assertTrue -import org.junit.Test - -class DeliveryStateTest { - - @Test - fun staleSuccessCannotPublishAfterUidGoneAndReplacement() { - val states = DeliveryStateStore(now = { 0L }) - val first = states.begin(42) ?: error("first attempt was not created") - - states.invalidateGone(42) - val replacement = states.begin(42) ?: error("replacement attempt was not created") - - assertFalse(states.commitSuccess(42, first, Any(), Any())) - assertTrue(states.isCurrentSending(42, replacement)) - } - - @Test - fun failureAfterUidGoneIsCountedWithoutPublishing() { - val states = DeliveryStateStore(now = { 0L }) - val attempt = states.begin(43) ?: error("attempt was not created") - - states.invalidateGone(43) - - assertFalse(states.commitSuccess(43, attempt, Any(), Any())) - assertFalse(states.recordFailure(43, attempt)) - assertEquals(1, states.failureCount(43)) - } - - @Test - fun staleFailureCannotOverwriteReplacementSuccess() { - val states = DeliveryStateStore(now = { 0L }) - val first = states.begin(7) ?: error("first attempt was not created") - states.invalidateMatching { it == 7 } - val replacement = states.begin(7) ?: error("replacement attempt was not created") - - val provider = Any() - val recipient = Any() - assertTrue(states.commitSuccess(7, replacement, provider, recipient)) - assertFalse(states.recordFailure(7, first)) - assertTrue(states.isCurrentDelivery(7, provider, recipient)) - } - - @Test - fun failedActiveUidIsRedeliveredOnGenerationChange() { - val states = DeliveryStateStore(now = { 0L }) - val attempt = states.begin(10042) ?: error("attempt was not created") - states.recordFailure(10042, attempt) - states.finish(10042, attempt) - - val invalidation = states.invalidateMatching { it == 10042 } - - assertEquals(setOf(10042), invalidation.redeliveryUids) - val replacement = states.begin(10042) ?: error("replacement attempt was not created") - assertNotNull(replacement) - } - - @Test - fun oldDeathRecipientCannotRemoveReplacementDelivery() { - val states = DeliveryStateStore(now = { 0L }) - val firstAttempt = states.begin(11) ?: error("first attempt was not created") - val oldProvider = Any() - val oldRecipient = Any() - assertTrue(states.commitSuccess(11, firstAttempt, oldProvider, oldRecipient)) - - states.invalidateMatching { it == 11 } - val replacementAttempt = states.begin(11) ?: error("replacement attempt was not created") - val newProvider = Any() - val newRecipient = Any() - assertTrue(states.commitSuccess(11, replacementAttempt, newProvider, newRecipient)) - - assertFalse(states.removeIfCurrentDelivery(11, oldProvider, oldRecipient)) - assertTrue(states.isCurrentDelivery(11, newProvider, newRecipient)) - } - - @Test - fun threeFailuresEnterCooldownAndCooldownAttemptKeepsRun() { - var now = 0L - val states = - DeliveryStateStore( - now = { now }, - retryCooldownMs = 100, - failureRunMs = 1_000, - ) - - repeat(3) { - val attempt = states.begin(9) ?: error("attempt was not created") - states.recordFailure(9, attempt) - states.finish(9, attempt) - } - - now = 100 - val cooldownAttempt = states.begin(9) ?: error("cooldown attempt was not created") - states.recordFailure(9, cooldownAttempt) - states.finish(9, cooldownAttempt) - now = 101 - - assertEquals(3, states.failureCount(9)) - assertTrue(states.isThrottled(9)) - } -}