From 309dce3627ed414bad258f5d0e9bb5f1a680811e Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Fri, 9 Oct 2026 18:26:03 +0800 Subject: [PATCH] fix(android): reconcile rejected runtime state and sync admission #377 --- .../agent/persistence/PurchaseTaskStore.kt | 6 ++ .../agent/service/AgentForegroundService.kt | 10 +++- .../goauto/agent/service/AgentStateStore.kt | 9 ++- .../service/PurchaseRuntimeReconciliation.kt | 16 +++++ .../PurchaseRuntimeReconciliationTest.kt | 59 +++++++++++++++++++ 5 files changed, 98 insertions(+), 2 deletions(-) create mode 100644 android/app/src/main/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliation.kt create mode 100644 android/app/src/test/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliationTest.kt diff --git a/android/app/src/main/java/cn/ilapage/goauto/agent/persistence/PurchaseTaskStore.kt b/android/app/src/main/java/cn/ilapage/goauto/agent/persistence/PurchaseTaskStore.kt index 8f99ae3..4d25f1b 100644 --- a/android/app/src/main/java/cn/ilapage/goauto/agent/persistence/PurchaseTaskStore.kt +++ b/android/app/src/main/java/cn/ilapage/goauto/agent/persistence/PurchaseTaskStore.kt @@ -278,6 +278,12 @@ class PurchaseTaskStore(context: Context) : SQLiteOpenHelper(context, DATABASE_N "SELECT COUNT(*) FROM purchase_outbox WHERE upload_status='rejected'", null, ).use { it.moveToFirst(); it.getInt(0) } + @Synchronized + fun isCurrentAttemptRejected(taskId: Long, attemptId: String): Boolean = readableDatabase.rawQuery( + "SELECT 1 FROM purchase_task WHERE task_id=? AND attempt_id=? AND upload_status='rejected'", + arrayOf(taskId.toString(), attemptId), + ).use { it.moveToFirst() } + @Synchronized fun acknowledgeRejection(outboxId: Long) { writableDatabase.execSQL(PurchaseRejectionSql.acknowledge, arrayOf(System.currentTimeMillis(), outboxId)) diff --git a/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentForegroundService.kt b/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentForegroundService.kt index 2c180c2..e7b5f3a 100644 --- a/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentForegroundService.kt +++ b/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentForegroundService.kt @@ -204,6 +204,7 @@ class AgentForegroundService : Service() { } override fun onDestroy() { + stateStore.stop() probeHandoff.invalidate() if (diagnosticInstance === this) diagnosticInstance = null repurchaseClosed.set(true) @@ -621,7 +622,7 @@ class AgentForegroundService : Service() { publishCurrentPageResult(0L, "采集间隔中，还需 $remaining 秒。", replacementOriginType, replacementOriginTaskId) return } - if (!taskMutex.tryAcquire(CURRENT_PAGE_RESERVATION_ID)) { + if (!tryAcquireCurrentPage(taskMutex, working, CURRENT_PAGE_RESERVATION_ID)) { publishCurrentPageResult(0L, "设备正在执行任务，请稍后再试。", replacementOriginType, replacementOriginTaskId) return } @@ -1085,6 +1086,13 @@ class AgentForegroundService : Service() { markRejected = { item, code -> probeHandoff.invalidate() purchaseStore.markRejected(item, code) + synchronized(taskMutex) { + val runtime = stateStore.read() + if (shouldClearRejectedPurchase(item.taskId, runtime.currentTaskId, runtime.currentTaskType, + taskMutex.currentTaskId(), purchaseStore.isCurrentAttemptRejected(item.taskId, item.attemptId))) { + stateStore.clearActiveTask(item.taskId) + } + } sendBroadcast(Intent(ACTION_PURCHASE_REJECTIONS_CHANGED).setPackage(packageName)) }, ).flush() diff --git a/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentStateStore.kt b/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentStateStore.kt index c425462..38b1bc7 100644 --- a/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentStateStore.kt +++ b/android/app/src/main/java/cn/ilapage/goauto/agent/service/AgentStateStore.kt @@ -191,8 +191,9 @@ class AgentSettingsStore(context: Context) { class AgentStateStore(context: Context) { private val preferences = context.getSharedPreferences(PREFERENCES, Context.MODE_PRIVATE) + private val publicationGate = AgentStatePublicationGate() - fun update(code: String, message: String, deviceId: Long? = null, tokenStored: Boolean? = null, heartbeat: Boolean = false, connection: Boolean = false) { + fun update(code: String, message: String, deviceId: Long? = null, tokenStored: Boolean? = null, heartbeat: Boolean = false, connection: Boolean = false) = publicationGate.publish { val editor = preferences.edit() .putString(STATE_CODE, code) .putString(STATE_MESSAGE, message) @@ -203,6 +204,12 @@ class AgentStateStore(context: Context) { editor.apply() } + /** This service instance cannot publish a late heartbeat over STOPPED or a restarted service. */ + fun stop() = publicationGate.stop { + preferences.edit().putString(STATE_CODE, "STOPPED").putString(CONNECTION_CODE, "STOPPED") + .putString(STATE_MESSAGE, "前台服务已停止").apply() + } + fun connectionCode(): String = preferences.getString(CONNECTION_CODE, "STOPPED") ?: "STOPPED" fun read(): AgentState = AgentState( diff --git a/android/app/src/main/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliation.kt b/android/app/src/main/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliation.kt new file mode 100644 index 0000000..4e58c83 --- /dev/null +++ b/android/app/src/main/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliation.kt @@ -0,0 +1,16 @@ +package cn.ilapage.goauto.agent.service + +internal fun shouldClearRejectedPurchase( + rejectedTaskId: Long, runtimeTaskId: Long?, runtimeTaskType: String?, + executingTaskId: Long?, currentAttemptRejected: Boolean, +): Boolean = executingTaskId == null && runtimeTaskId == rejectedTaskId && + runtimeTaskType == "purchase" && currentAttemptRejected + +internal class AgentStatePublicationGate { + private var stopped = false + @Synchronized fun publish(write: () -> Unit) { if (!stopped) write() } + @Synchronized fun stop(write: () -> Unit) { stopped = true; write() } +} + +internal fun tryAcquireCurrentPage(mutex: TaskExecutionMutex, syncing: java.util.concurrent.atomic.AtomicBoolean, reservation: Long): Boolean = + synchronized(mutex) { !syncing.get() && mutex.tryAcquire(reservation) } diff --git a/android/app/src/test/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliationTest.kt b/android/app/src/test/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliationTest.kt new file mode 100644 index 0000000..f7f35b4 --- /dev/null +++ b/android/app/src/test/java/cn/ilapage/goauto/agent/service/PurchaseRuntimeReconciliationTest.kt @@ -0,0 +1,59 @@ +package cn.ilapage.goauto.agent.service + +import org.junit.Assert.* +import org.junit.Test + +class PurchaseRuntimeReconciliationTest { + @Test fun `manual current page cannot enter while sync owns recovery and flush`() { + val mutex = TaskExecutionMutex() + val syncing = java.util.concurrent.atomic.AtomicBoolean(false) + val syncEntered = java.util.concurrent.CountDownLatch(1) + val finishSync = java.util.concurrent.CountDownLatch(1) + val worker = java.util.concurrent.Executors.newSingleThreadExecutor() + val future = worker.submit { + synchronized(mutex) { assertTrue(syncing.compareAndSet(false, true)) } + syncEntered.countDown() + check(finishSync.await(3, java.util.concurrent.TimeUnit.SECONDS)) + syncing.set(false) + } + try { + assertTrue(syncEntered.await(3, java.util.concurrent.TimeUnit.SECONDS)) + assertFalse(tryAcquireCurrentPage(mutex, syncing, Long.MAX_VALUE)) + assertNull(mutex.currentTaskId()) + finishSync.countDown() + future.get(3, java.util.concurrent.TimeUnit.SECONDS) + assertTrue(tryAcquireCurrentPage(mutex, syncing, Long.MAX_VALUE)) + } finally { finishSync.countDown(); worker.shutdownNow() } + } + + @Test fun `reboot restored rejected task clears runtime so repurchase becomes idle`() { + var runtimeTaskId: Long? = 42 + val currentDatabaseAttemptRejected = true + if (shouldClearRejectedPurchase(42, runtimeTaskId, "purchase", null, currentDatabaseAttemptRejected)) runtimeTaskId = null + assertNull(runtimeTaskId) + assertTrue(runtimeTaskId == null) // final runtime predicate in repurchaseLocalIdle + } + + @Test fun `another task collection active executor or newer attempt cannot be cleared`() { + assertFalse(shouldClearRejectedPurchase(42, 43, "purchase", null, true)) + assertFalse(shouldClearRejectedPurchase(42, 42, "collection", null, true)) + assertFalse(shouldClearRejectedPurchase(42, 42, "purchase", 42, true)) + assertFalse(shouldClearRejectedPurchase(42, 42, "purchase", 43, true)) + assertFalse(shouldClearRejectedPurchase(42, 42, "purchase", null, false)) + } + + @Test fun `stopped service remains stopped after an in flight heartbeat completes`() { + val gate = AgentStatePublicationGate() + var connection = "CONNECTING" + gate.publish { connection = "ONLINE" } + assertEquals("ONLINE", connection) + gate.stop { connection = "STOPPED" } + gate.publish { connection = "ONLINE" } + assertEquals("STOPPED", connection) + val restartedService = AgentStatePublicationGate() + restartedService.publish { connection = "CONNECTING" } + assertEquals("CONNECTING", connection) + gate.publish { connection = "ONLINE" } + assertEquals("CONNECTING", connection) + } +}