fix: count invalidated binder delivery failure

.github/workflows/core.yml

@@ -123,7 +123,7 @@ jobs:
 
       - name: Build with Gradle
         run: |
-          ./gradlew zipAll
+          ./gradlew checkVersionName :daemon:testDebugUnitTest zipAll
 
       - name: Prepare artifact
         if: success()

build.gradle.kts

@@ -40,14 +40,16 @@ abstract class GitCommitCountValueSource : ValueSource<String, ValueSourceParame
     }
 }
 
-/** A ValueSource that executes 'git tag' to get the latest version tag. */
+/** A ValueSource that executes 'git tag' to get the latest stable version tag. */
 abstract class GitLatestTagValueSource : ValueSource<String, ValueSourceParameters.None> {
     @get:Inject abstract val execOperations: ExecOperations
 
     override fun obtain(): String {
         val output = ByteArrayOutputStream()
         val result = execOperations.exec {
-            commandLine("git", "tag", "--list", "--sort=-v:refname")
+            // Canary releases create their own tags. Letting one become VERSION_NAME puts its
+            // hyphen into the zip name, which the release workflow then parses as separate fields.
+            commandLine("git", "tag", "--list", "--sort=-v:refname", "v*")
             standardOutput = output
             isIgnoreExitValue = true
         }
@@ -218,6 +220,21 @@ val versionHashProvider =
     }
 val versionNameProvider = providers.of(GitLatestTagValueSource::class.java) {}
 
+// This repository always has a canary tag after its first successful master build. Exercising the
+// provider against the real checkout makes a regression in the v* filter fail before packaging can
+// reuse an old canary's version code and overwrite its release.
+tasks.register("checkVersionName") {
+    group = "verification"
+    description = "Checks that canary tags cannot become distribution version names."
+    inputs.property("versionName", versionNameProvider)
+    doLast {
+        val versionName = versionNameProvider.get()
+        check(!versionName.startsWith("canary-")) {
+            "Canary tag selected as the distribution version: $versionName"
+        }
+    }
+}
+
 val injectedPackageName = "com.android.shell"
 val injectedPackageUid = 2000
 val defaultManagerPackageName = "org.matrix.vector.manager"

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")
 }

daemon/src/main/kotlin/org/matrix/vector/daemon/ipc/DeliveryAttemptTracker.kt

@@ -0,0 +1,95 @@
+package org.matrix.vector.daemon.ipc
+
+internal enum class DeliveryCompletion {
+  COMMIT,
+  RECORD_FAILURE,
+  IGNORE_STALE,
+}
+
+/**
+ * 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<Int, Long>()
+  private val successfulUidGenerations = mutableMapOf<Int, Long>()
+  private val active = mutableMapOf<Int, Attempt>()
+
+  @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
+
+  /**
+   * Applies failure accounting while holding the same generation lock as lifecycle invalidation.
+   * A uidGone-invalidated failure still counts until a newer delivery succeeds. Cache clears and
+   * failures older than a successful replacement are ignored, so they cannot recreate throttling.
+   */
+  @Synchronized
+  fun complete(
+      uid: Int,
+      attempt: Attempt,
+      delivered: Boolean,
+      clearFailures: () -> Unit,
+      recordFailure: () -> Unit,
+  ): DeliveryCompletion {
+    if (cacheGeneration != attempt.cacheGeneration) {
+      return DeliveryCompletion.IGNORE_STALE
+    }
+
+    val current = active[uid] == attempt && uidGenerations[uid] == attempt.uidGeneration
+    if (delivered) {
+      if (!current) return DeliveryCompletion.IGNORE_STALE
+      successfulUidGenerations[uid] = attempt.uidGeneration
+      clearFailures()
+      return DeliveryCompletion.COMMIT
+    }
+
+    if ((successfulUidGenerations[uid] ?: Long.MIN_VALUE) > attempt.uidGeneration) {
+      return DeliveryCompletion.IGNORE_STALE
+    }
+    recordFailure()
+    return DeliveryCompletion.RECORD_FAILURE
+  }
+
+  @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()
+    successfulUidGenerations.clear()
+  }
+
+  private fun nextUidGeneration(uid: Int): Long {
+    val next = (uidGenerations[uid] ?: 0L) + 1L
+    uidGenerations[uid] = next
+    return next
+  }
+}

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<Int>()
 
-    /** The uids a send is running for right now, so the three observer callbacks agree on one. */
-    private val sending = ConcurrentHashMap.newKeySet<Int>()
+    /** 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,60 @@ 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)
-                  binderFailures.remove(uid)
-                  linkDelivery(uid, delivered)
-                } else {
-                  recordFailure(uid, module.packageName)
+                val completion =
+                    deliveryAttempts.complete(
+                        uid = uid,
+                        attempt = attempt,
+                        delivered = delivered != null,
+                        clearFailures = { binderFailures.remove(uid) },
+                        recordFailure = { recordFailure(uid, module.packageName) },
+                    )
+                if (completion == DeliveryCompletion.COMMIT) {
+                  linkDelivery(uid, checkNotNull(delivered), attempt)
                 }
               } 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 +191,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 +236,8 @@ 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. An attempt invalidated by uidGone and its replacement can
+      // both fail for the same uid; a plain get-then-put would lose one of those failures.
       binderFailures.compute(uid) { _, previous ->
         val now = SystemClock.elapsedRealtime()
         val count =
@@ -229,11 +266,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) }
       }