fix: harden Android capture, memory and background operations

app/src/androidTest/java/life/andre/message487/StorageDeliveryDeviceTest.kt

@@ -1,5 +1,6 @@
 package life.andre.message487
 
+import kotlinx.coroutines.async
 import android.content.Context
 import androidx.core.content.FileProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
@@ -81,6 +82,41 @@ class StorageDeliveryDeviceTest {
         }
     }
 
+    @Test fun allocationFailureFallsBackToTextAndNextCaptureStillRuns() {
+        val delivered = java.util.concurrent.CountDownLatch(2)
+        val errors = java.util.concurrent.atomic.AtomicInteger()
+        NotificationImageQueue().use { queue ->
+            queue.submit("failed", { withImageMemory<List<NotificationImage>> { throw OutOfMemoryError("injected") } },
+                { assertTrue(it.isEmpty()); delivered.countDown() },
+                { assertTrue(it is ImageMemoryException); errors.incrementAndGet() })
+            queue.submit("next", { withImageMemory { emptyList() } }, { delivered.countDown() }, { throw it })
+            assertTrue(delivered.await(3, TimeUnit.SECONDS))
+            assertEquals(1, errors.get())
+        }
+    }
+
+    @Test fun androidHttpDeadlineAndCoroutineCancellationReleaseConnections() {
+        MockWebServer().use { server ->
+            server.start()
+            server.enqueue(MockResponse().setBody("{\"status\":\"accepted\",\"event_id\":\"a\"}")
+                .throttleBody(1, 50, TimeUnit.MILLISECONDS))
+            val result = WebhookClient(2_000, 200).sendJson(server.url("/").toString(), "a", "{}", true, token)
+            assertEquals(DeliveryStatus.TIMEOUT, result.status)
+            assertNotNull(server.takeRequest(2, TimeUnit.SECONDS))
+            server.enqueue(MockResponse().setSocketPolicy(okhttp3.mockwebserver.SocketPolicy.NO_RESPONSE))
+            kotlinx.coroutines.runBlocking {
+                val task = async(kotlinx.coroutines.Dispatchers.IO) {
+                    WebhookClient().sendCancellable(QueuedRequest(server.url("/").toString(), true, "{}", token), "a")
+                }
+                assertNotNull(server.takeRequest(2, TimeUnit.SECONDS))
+                val start = System.nanoTime()
+                task.cancel(); task.join()
+                assertTrue(task.isCancelled)
+                assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) < 2_000)
+            }
+        }
+    }
+
     @Test fun androidHttpTransportRequiresMatchingAckAndDoesNotFollowRedirects() {
         MockWebServer().use { server ->
             server.start()

app/src/androidTest/java/life/andre/message487/UiThreadDeviceTest.kt

@@ -48,6 +48,49 @@ class UiThreadDeviceTest {
         }
     }
 
+    @Test fun initializationAndUpdateFailuresKeepIoAndDiagnosticsOffMainThread() {
+        val reads = java.util.concurrent.atomic.AtomicInteger()
+        val formatted = java.util.concurrent.CountDownLatch(1)
+        val formattedOnMain = java.util.concurrent.atomic.AtomicBoolean()
+        val failure = object : java.io.IOException("injected storage failure") {
+            override fun getStackTrace(): Array<StackTraceElement> {
+                formattedOnMain.set(android.os.Looper.myLooper() == android.os.Looper.getMainLooper())
+                formatted.countDown()
+                return super.getStackTrace()
+            }
+        }
+        compose.runOnUiThread {
+            previousPolicy = StrictMode.getThreadPolicy()
+            StrictMode.setThreadPolicy(StrictMode.ThreadPolicy.Builder()
+                .detectDiskReads().detectDiskWrites().detectNetwork().penaltyDeath().build())
+            model = ConnectionViewModel(app) {
+                check(android.os.Looper.myLooper() != android.os.Looper.getMainLooper())
+                // Real disk I/O on both the failing read and its successful retry.
+                java.io.File(app.cacheDir, "settings-read-probe").writeText("probe")
+                if (reads.incrementAndGet() == 1) throw failure
+                app.graph.settings.load()
+            }
+            app.reportLocalFailure(failure)
+        }
+        compose.setContent { MessageTheme { MessageScreen(model) } }
+        compose.waitUntil(10_000) { model.state.value.initializationFailed && !model.state.value.busy }
+        node(R.string.retry).performClick()
+        compose.waitUntil(10_000) { !model.state.value.initializationFailed && !model.state.value.busy }
+        assertEquals(2, reads.get())
+        assertTrue(formatted.await(3, java.util.concurrent.TimeUnit.SECONDS))
+        assertFalse(formattedOnMain.get())
+        kotlinx.coroutines.runBlocking {
+            kotlinx.coroutines.withContext(kotlinx.coroutines.Dispatchers.Main) {
+                var failed = false
+                try {
+                    // Creates/cleans a real temporary APK, then fails before any network request.
+                    AppUpdates(app).download(AppUpdate(UpdateSource.GITHUB, "99.0.0"))
+                } catch (_: IllegalArgumentException) { failed = true }
+                assertTrue(failed)
+            }
+        }
+    }
+
     @Test fun navigateEditAndRefreshWithoutMainThreadIo() {
         show()
         node(R.string.connection_nav).performClick()
@@ -56,8 +99,15 @@ class UiThreadDeviceTest {
         compose.onNode(hasText(app.getString(R.string.sources_tab)) and hasClickAction()).performClick()
         compose.onNode(hasScrollToNodeAction()).performScrollToNode(hasText(app.getString(R.string.image_max_edge)))
         node(R.string.image_max_edge).performScrollTo().performTextReplacement("2")
+        compose.waitUntil(10_000) {
+            compose.waitForIdle()
+            model.settings.value.imageMaxEdge == 2 && !model.state.value.busy
+        }
         node(R.string.image_max_edge).performTextInput("048")
-        compose.waitUntil(10_000) { model.settings.value.imageMaxEdge == 2048 }
+        compose.waitUntil(10_000) {
+            compose.waitForIdle()
+            model.settings.value.imageMaxEdge == 2048 && !model.state.value.busy
+        }
         node(R.string.image_max_edge).assertIsEnabled()
         icon(R.string.back).performClick()
         node(R.string.journal).performClick()
@@ -69,7 +119,12 @@ class UiThreadDeviceTest {
         node(R.string.update_github).performScrollTo().performClick()
         compose.waitUntil(10_000) { AppUpdates(app).source() == UpdateSource.GITHUB }
         node(R.string.update_automatic).performScrollTo()
-        compose.onNodeWithContentDescription(app.getString(R.string.update_automatic)).performClick()
+        // The preference write can finish before the ViewModel clears busy on the main thread.
+        val automatic = hasContentDescription(app.getString(R.string.update_automatic))
+        compose.waitUntil(10_000) {
+            compose.onAllNodes(automatic and isEnabled()).fetchSemanticsNodes().isNotEmpty()
+        }
+        compose.onNode(automatic).assertIsOn().performClick()
         compose.waitUntil(10_000) { !UpdatePreferences(app).automatic }
         UpdatePreferences(app).detected(UpdateSource.GITHUB, AppUpdate(UpdateSource.GITHUB, "99.0.0"))
         // App settings, permission checks and the update dialog refresh on ON_RESUME.

app/src/main/java/life/andre/message487/AppUpdates.kt

@@ -110,7 +110,7 @@ internal class AppUpdates(private val context: Context) {
 
     @Suppress("DEPRECATION")
     fun source(): UpdateSource? {
-        val saved = preferences.getString("source", null)
+        val saved = preferences.readString("source", null)
         UpdateSource.entries.firstOrNull { it.name == saved }?.let { return it }
         val installer = try {
             if (Build.VERSION.SDK_INT >= 30) manager.getInstallSourceInfo(context.packageName).installingPackageName
@@ -130,53 +130,60 @@ internal class AppUpdates(private val context: Context) {
     }
 
     fun select(source: UpdateSource) {
-        preferences.edit().putString("source", source.name).commit()
+        preferences.edit().putString("source", source.name).save()
     }
 
     suspend fun check(source: UpdateSource): AppUpdate? = withContext(Dispatchers.IO) {
-        val url = if (source == UpdateSource.FDROID) FDROID_UPDATE_API else GITHUB_RELEASES_API
-        val json = request(url) { connection ->
-            requireUpdateMetadataStatus(connection.responseCode)
-            connection.inputStream.use { stream ->
-                String(readUpdateMetadata(stream), Charsets.UTF_8)
+        cancellableHttp { cancellation ->
+            val url = if (source == UpdateSource.FDROID) FDROID_UPDATE_API else GITHUB_RELEASES_API
+            val json = request(url, cancellation) { connection ->
+                requireUpdateMetadataStatus(connection.responseCode)
+                connection.inputStream.use { stream ->
+                    String(readUpdateMetadata(stream), Charsets.UTF_8)
+                }
             }
+            if (source == UpdateSource.FDROID) parseFdroidUpdate(json, installedCode())
+            else parseGithubUpdate(json, BuildConfig.VERSION_NAME)
         }
-        if (source == UpdateSource.FDROID) parseFdroidUpdate(json, installedCode())
-        else parseGithubUpdate(json, BuildConfig.VERSION_NAME)
     }
 
     suspend fun download(update: AppUpdate): File = withContext(Dispatchers.IO) {
-        require(update.source == UpdateSource.GITHUB)
-        val directory = File(context.cacheDir, "updates").apply { mkdirs() }
-        val partial = File.createTempFile("update-", ".part", directory)
-        val apk = File(directory, partial.nameWithoutExtension + ".apk")
-        try {
-            val digest = MessageDigest.getInstance("SHA-256")
-            request(requireNotNull(update.apkUrl)) { connection ->
-                require(connection.responseCode == 200)
-                connection.inputStream.use { input ->
-                    partial.outputStream().use { output ->
-                        val buffer = ByteArray(64 * 1024)
-                        var total = 0L
-                        while (true) {
-                            coroutineContext.ensureActive()
-                            val count = input.read(buffer)
-                            if (count < 0) break
-                            total += count
-                            require(total <= update.size && total <= MAX_UPDATE_APK_BYTES)
-                            digest.update(buffer, 0, count)
-                            output.write(buffer, 0, count)
+        cancellableHttp { cancellation ->
+            require(update.source == UpdateSource.GITHUB)
+            val directory = File(context.cacheDir, "updates").apply { mkdirs() }
+            cleanupUpdateFiles(directory)
+            val partial = File.createTempFile("update-", ".part", directory)
+            val apk = File(directory, partial.nameWithoutExtension + ".apk")
+            try {
+                val digest = MessageDigest.getInstance("SHA-256")
+                request(requireNotNull(update.apkUrl), cancellation) { connection ->
+                    require(connection.responseCode == 200)
+                    connection.inputStream.use { input ->
+                        partial.outputStream().use { output ->
+                            val buffer = ByteArray(64 * 1024)
+                            var total = 0L
+                            while (true) {
+                                coroutineContext.ensureActive()
+                                if (cancellation.isCanceled) throw kotlinx.coroutines.CancellationException("Download cancelled")
+                                val count = input.read(buffer)
+                                if (count < 0) break
+                                total += count
+                                require(total <= update.size && total <= MAX_UPDATE_APK_BYTES)
+                                digest.update(buffer, 0, count)
+                                output.write(buffer, 0, count)
+                            }
+                            require(total == update.size)
                         }
-                        require(total == update.size)
                     }
                 }
+                require(digest.digest().hex() == update.sha256)
+                verifyApk(partial, update.version)
+                check(partial.renameTo(apk))
+                cleanupUpdateFiles(directory, keep = apk)
+                apk
+            } finally {
+                partial.delete()
             }
-            require(digest.digest().hex() == update.sha256)
-            verifyApk(partial, update.version)
-            check(partial.renameTo(apk))
-            apk
-        } finally {
-            partial.delete()
         }
     }
 
@@ -202,24 +209,31 @@ internal class AppUpdates(private val context: Context) {
 private fun ByteArray.hex(): String = joinToString("") { "%02x".format(it) }
 
 // GitHub release assets redirect to its CDN. Only HTTPS redirects on these exact hosts are allowed.
-private fun <T> request(url: String, read: (HttpURLConnection) -> T): T {
-    var current = url
-    repeat(6) {
-        val uri = URI(current)
-        require(uri.scheme == "https" && uri.userInfo == null && (uri.port == -1 || uri.port == 443))
-        require(uri.host in setOf("api.github.com", "github.com", "release-assets.githubusercontent.com", "f-droid.org"))
-        val connection = uri.toURL().openConnection() as HttpURLConnection
-        try {
-            connection.connectTimeout = 15_000
-            connection.readTimeout = 30_000
-            connection.instanceFollowRedirects = false
-            connection.setRequestProperty("User-Agent", "Message487-Updates")
-            if (connection.responseCode in listOf(301, 302, 303, 307, 308)) {
-                current = uri.resolve(requireNotNull(connection.getHeaderField("Location"))).toString()
-            } else return read(connection)
-        } finally { connection.disconnect() }
-    }
-    error("Too many redirects")
+private fun <T> request(url: String, cancellation: android.os.CancellationSignal, read: (HttpURLConnection) -> T): T =
+    HttpRequestControl(cancellation).use { control ->
+        var current = url
+        repeat(6) {
+            control.checkActive()
+            val uri = URI(current)
+            require(uri.scheme == "https" && uri.userInfo == null && (uri.port == -1 || uri.port == 443))
+            require(uri.host in setOf("api.github.com", "github.com", "release-assets.githubusercontent.com", "f-droid.org"))
+            val connection = uri.toURL().openConnection() as HttpURLConnection
+            try {
+                control.attach(connection)
+                connection.connectTimeout = 15_000
+                connection.readTimeout = 30_000
+                connection.instanceFollowRedirects = false
+                connection.setRequestProperty("User-Agent", "Message487-Updates")
+                if (connection.responseCode in listOf(301, 302, 303, 307, 308)) {
+                    current = uri.resolve(requireNotNull(connection.getHeaderField("Location"))).toString()
+                } else {
+                    val result = read(connection)
+                    control.checkActive()
+                    return result
+                }
+            } finally { connection.disconnect() }
+        }
+        error("Too many redirects")
 }
 
 internal fun readUpdateMetadata(input: InputStream): ByteArray {
@@ -232,3 +246,12 @@ internal fun readUpdateMetadata(input: InputStream): ByteArray {
         output.write(buffer, 0, count)
     }
 }
+
+internal fun cleanupUpdateFiles(directory: File, keep: File? = null, now: Long = System.currentTimeMillis()) {
+    val files = directory.listFiles().orEmpty().filter { it.name.startsWith("update-") && it.isFile }
+    // Keep two APKs so a previously granted installer URI can still be read.
+    files.filter { it.extension == "apk" && it != keep }.sortedByDescending(File::lastModified)
+        .drop(if (keep == null) 2 else 1).forEach { if (!it.delete()) throw java.io.IOException("Could not remove old update") }
+    files.filter { it.extension == "part" && now - it.lastModified() > java.util.concurrent.TimeUnit.DAYS.toMillis(1) }
+        .forEach { if (!it.delete()) throw java.io.IOException("Could not remove abandoned update") }
+}

app/src/main/java/life/andre/message487/BackgroundOperations.kt

@@ -0,0 +1,48 @@
+package life.andre.message487
+
+import android.content.BroadcastReceiver
+import android.content.Context
+import java.util.concurrent.ArrayBlockingQueue
+import java.util.concurrent.Executor
+import java.util.concurrent.RejectedExecutionException
+import java.util.concurrent.ScheduledThreadPoolExecutor
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+import life.andre.message487.diagnostics.DiagnosticEvent
+
+internal fun boundedExecutor(name: String, capacity: Int) = ThreadPoolExecutor(
+    1, 1, 0, TimeUnit.MILLISECONDS, ArrayBlockingQueue<Runnable>(capacity),
+    { task -> Thread(task, name).apply { isDaemon = true } }, ThreadPoolExecutor.AbortPolicy(),
+)
+
+internal fun Context.reportLocalFailure(error: Exception) {
+    try { (applicationContext as? MessageApplication)?.diagnostics?.record(DiagnosticEvent.LOCAL_OPERATION_FAILED, error = error) }
+    catch (_: Exception) { /* Reporting must not turn a recoverable failure into a process crash. */ }
+}
+
+internal fun Executor.submitSafely(onFailure: (Exception) -> Unit, block: () -> Unit): Boolean = try {
+    execute { try { block() } catch (error: Exception) { onFailure(error) } }
+    true
+} catch (error: RejectedExecutionException) { onFailure(error); false }
+
+private val broadcastDeadlines = ScheduledThreadPoolExecutor(1) { task ->
+    Thread(task, "message487-broadcast-deadlines").apply { isDaemon = true }
+}.apply { removeOnCancelPolicy = true }
+
+internal fun runBroadcast(context: Context, pending: BroadcastReceiver.PendingResult?, executor: Executor, onFailure: (Exception) -> Unit = context::reportLocalFailure, block: () -> Unit) {
+    val finished = AtomicBoolean()
+    fun finish(): Boolean {
+        if (!finished.compareAndSet(false, true)) return false
+        try { pending?.finish() } catch (error: Exception) { context.reportLocalFailure(error) }
+        return true
+    }
+    val deadline = broadcastDeadlines.schedule({
+        if (finish()) onFailure(java.io.IOException("Broadcast processing exceeded its deadline"))
+    }, 8, TimeUnit.SECONDS)
+    val accepted = executor.submitSafely(onFailure) {
+        try { if (!finished.get()) block() }
+        finally { deadline.cancel(false); finish() }
+    }
+    if (!accepted) { deadline.cancel(false); finish() }
+}