feat(scene): complete automation runtime #11
+14
-10
@@ -1,6 +1,7 @@
|
|||||||
package com.flagship.abox.manager.automation.data
|
package com.flagship.abox.manager.automation.data
|
||||||
|
|
||||||
import android.content.Context
|
import android.content.Context
|
||||||
|
import androidx.room.withTransaction
|
||||||
import com.flagship.abox.manager.domain.AutomationType
|
import com.flagship.abox.manager.domain.AutomationType
|
||||||
import com.flagship.abox.manager.domain.ComparisonOperator
|
import com.flagship.abox.manager.domain.ComparisonOperator
|
||||||
import com.flagship.abox.manager.domain.DeviceName
|
import com.flagship.abox.manager.domain.DeviceName
|
||||||
@@ -27,8 +28,9 @@ import kotlinx.coroutines.flow.map
|
|||||||
|
|
||||||
class RoomSceneRepository(
|
class RoomSceneRepository(
|
||||||
context: Context,
|
context: Context,
|
||||||
private val dao: AutomationDao,
|
private val database: AutomationDatabase,
|
||||||
) : SceneRepository {
|
) : SceneRepository {
|
||||||
|
private val dao = database.automationDao()
|
||||||
private val seedPreferences = context.getSharedPreferences("scene_seed_state", Context.MODE_PRIVATE)
|
private val seedPreferences = context.getSharedPreferences("scene_seed_state", Context.MODE_PRIVATE)
|
||||||
|
|
||||||
override fun observeScenes(profileId: ProviderProfileId): Flow<DomainResult<List<Scene>>> = flow {
|
override fun observeScenes(profileId: ProviderProfileId): Flow<DomainResult<List<Scene>>> = flow {
|
||||||
@@ -57,15 +59,17 @@ class RoomSceneRepository(
|
|||||||
return DomainResult.Failure(DomainError.InvalidRequest(detail = "Scene profile mismatch"))
|
return DomainResult.Failure(DomainError.InvalidRequest(detail = "Scene profile mismatch"))
|
||||||
}
|
}
|
||||||
return runCatching {
|
return runCatching {
|
||||||
|
|
|||||||
val existing = dao.getScene(profileId.value, scene.id.value)
|
database.withTransaction {
|
||||||
if (requireExisting && existing == null) throw IllegalArgumentException("Scene not found")
|
val existing = dao.getScene(profileId.value, scene.id.value)
|
||||||
if (!requireExisting && existing != null) throw IllegalArgumentException("Scene already exists")
|
if (requireExisting && existing == null) throw IllegalArgumentException("Scene not found")
|
||||||
dao.upsertScene(scene.toEntity())
|
if (!requireExisting && existing != null) throw IllegalArgumentException("Scene already exists")
|
||||||
existing?.let(SceneDocumentCodec::decode)?.triggers.orEmpty()
|
dao.upsertScene(scene.toEntity())
|
||||||
.filter { previous -> scene.triggers.none { current -> current.hasSameDefinitionAs(previous) } }
|
existing?.let(SceneDocumentCodec::decode)?.triggers.orEmpty()
|
||||||
.forEach { changed ->
|
.filter { previous -> scene.triggers.none { current -> current.hasSameDefinitionAs(previous) } }
|
||||||
dao.deleteTriggerState(profileId.value, scene.id.value, changed.id.value)
|
.forEach { changed ->
|
||||||
}
|
dao.deleteTriggerState(profileId.value, scene.id.value, changed.id.value)
|
||||||
|
}
|
||||||
|
}
|
||||||
scene
|
scene
|
||||||
}.fold({ DomainResult.Success(it) }, { DomainResult.Failure(DomainError.InvalidRequest(detail = it.message)) })
|
}.fold({ DomainResult.Success(it) }, { DomainResult.Failure(DomainError.InvalidRequest(detail = it.message)) })
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -63,7 +63,7 @@ object AutomationModule {
|
|||||||
fun provideSceneRepository(
|
fun provideSceneRepository(
|
||||||
@ApplicationContext context: Context,
|
@ApplicationContext context: Context,
|
||||||
database: AutomationDatabase,
|
database: AutomationDatabase,
|
||||||
): SceneRepository = RoomSceneRepository(context, database.automationDao())
|
): SceneRepository = RoomSceneRepository(context, database)
|
||||||
|
|
||||||
@Provides
|
@Provides
|
||||||
@Singleton
|
@Singleton
|
||||||
|
|||||||
+12
-6
@@ -38,18 +38,24 @@ class AutomationBootstrapper(
|
|||||||
previousProfileId?.takeIf { it != active?.id }?.let { scheduler.cancel(it) }
|
previousProfileId?.takeIf { it != active?.id }?.let { scheduler.cancel(it) }
|
||||||
previousProfileId = active?.id
|
previousProfileId = active?.id
|
||||||
if (active != null && session?.profileId == active.id) {
|
if (active != null && session?.profileId == active.id) {
|
||||||
|
var scheduleReady = false
|
||||||
while (true) {
|
while (true) {
|
||||||
|
pr-agent-bot
commented
漏调度 初始协调成功后, **漏调度**
初始协调成功后,`scheduleReady` 会永久保持为 true,前台循环不再重新协调。若之后创建或启停场景时的单次 `reconcile` 因会话或存储瞬时失败,失败结果又被调用方忽略,后台计划将持续过期;创建首个自动场景时甚至不会产生 WorkManager 任务,直到进程或会话重启。
<!-- pr-agent-dedup: 02d0c47771c5 -->
<!-- pr-agent-key-issue-location: d3d969c486b0 -->
|
|||||||
isForeground.first { it }
|
isForeground.first { it }
|
||||||
when (val validity = sessionCoordinator.validateSession(active.id)) {
|
when (val validity = sessionCoordinator.validateSession(active.id)) {
|
||||||
|
pr-agent-bot
commented
重复调度 有效会话在前台时,每轮 60 秒轮询都会执行完整的调度协调。其下游会更新周期任务、取消全部时间触发任务并逐个重建,即使场景没有任何变化;场景或时间触发器较多时会持续产生 WorkManager 数据库写入和任务取消,并干扰临近执行时间的任务。调度协调应仅由配置、资料或会话变化触发,轮询循环只需执行引擎评估。 **重复调度**
有效会话在前台时,每轮 60 秒轮询都会执行完整的调度协调。其下游会更新周期任务、取消全部时间触发任务并逐个重建,即使场景没有任何变化;场景或时间触发器较多时会持续产生 WorkManager 数据库写入和任务取消,并干扰临近执行时间的任务。调度协调应仅由配置、资料或会话变化触发,轮询循环只需执行引擎评估。
<!-- pr-agent-dedup: 49f38f37f27d -->
<!-- pr-agent-key-issue-location: 228b2c82db3e -->
|
|||||||
is DomainResult.Success -> if (validity.value == SessionValidity.VALID) {
|
is DomainResult.Success -> if (validity.value == SessionValidity.VALID) {
|
||||||
scheduler.reconcile(active.id)
|
if (!scheduleReady) {
|
||||||
try {
|
scheduleReady = scheduler.reconcile(active.id) is DomainResult.Success
|
||||||
engine.evaluate(active.id)
|
}
|
||||||
} catch (cancelled: CancellationException) {
|
if (scheduleReady) {
|
||||||
throw cancelled
|
try {
|
||||||
} catch (_: Exception) {}
|
engine.evaluate(active.id)
|
||||||
|
} catch (cancelled: CancellationException) {
|
||||||
|
throw cancelled
|
||||||
|
} catch (_: Exception) {}
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
scheduler.cancel(active.id)
|
scheduler.cancel(active.id)
|
||||||
|
scheduleReady = false
|
||||||
}
|
}
|
||||||
is DomainResult.Failure -> Unit
|
is DomainResult.Failure -> Unit
|
||||||
}
|
}
|
||||||
|
|||||||
+12
-2
@@ -118,8 +118,18 @@ class DefaultAutomationEngine(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (!newlyMatched) continue
|
if (!newlyMatched) continue
|
||||||
if (conditionEvaluator.evaluate(scene.conditions, snapshot, now) !is ConditionsEvaluation.Satisfied) {
|
when (conditionEvaluator.evaluate(scene.conditions, snapshot, now)) {
|
||||||
|
pr-agent-bot
commented
触发延迟 时间触发的 occurrence 仅在条件满足后才被认领。如果 08:00 已完成评估但条件不满足,该 occurrence 在宽限期内仍保持未消费;条件于 08:10 变为满足时,同一个 08:00 规则会被延迟执行。非时间触发会先持久化边沿,因此两类触发器的条件语义不一致。应在条件明确不满足时消费或记录该 occurrence,并仅对条件数据不可用的情况保留重试。 **触发延迟**
时间触发的 occurrence 仅在条件满足后才被认领。如果 08:00 已完成评估但条件不满足,该 occurrence 在宽限期内仍保持未消费;条件于 08:10 变为满足时,同一个 08:00 规则会被延迟执行。非时间触发会先持久化边沿,因此两类触发器的条件语义不一致。应在条件明确不满足时消费或记录该 occurrence,并仅对条件数据不可用的情况保留重试。
<!-- pr-agent-dedup: f4cab2f239a0 -->
<!-- pr-agent-key-issue-location: e0ecaad0e2a9 -->
|
|||||||
continue
|
is ConditionsEvaluation.Unavailable -> continue
|
||||||
|
is ConditionsEvaluation.NotSatisfied -> {
|
||||||
|
for ((trigger, occurrenceKey) in matchedTimeTriggers) {
|
||||||
|
when (val claim = runtimeRepository.claimOccurrence(profileId, scene.id, trigger.id, occurrenceKey)) {
|
||||||
|
is DomainResult.Failure -> return claim
|
||||||
|
is DomainResult.Success -> Unit
|
||||||
|
}
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
ConditionsEvaluation.Satisfied -> Unit
|
||||||
}
|
}
|
||||||
var executionCount = if (newlyMatchedStateTriggers.isEmpty()) 0 else 1
|
var executionCount = if (newlyMatchedStateTriggers.isEmpty()) 0 else 1
|
||||||
if (matchedTimeTriggers.isNotEmpty()) {
|
if (matchedTimeTriggers.isNotEmpty()) {
|
||||||
|
|||||||
Reference in New Issue
Block a user
更新竞态
场景写入和旧触发状态清理是多个独立 DAO 操作,前台轮询或 Worker 可在两者之间读取新场景和旧状态。编辑一个保留相同 ID 的阈值触发器时,引擎可能基于混合状态执行并写入新边沿,随后清理又将其删除,导致下次轮询重复执行或丢失边沿。应将读取、场景更新和触发状态清理放入同一个 Room 事务。