From 6610061c28dad9de9b6bb986896c0e24f555374b Mon Sep 17 00:00:00 2001 From: Rob Ribeiro Date: Sun, 1 Feb 2026 11:45:53 -0300 Subject: [PATCH] feat(core-domain): add saga decisions, reducers, and tests for paging happy path --- core-domain/api/core-domain.api | 23 ++ .../hailie/domain/model/SyncAggregateApply.kt | 89 ++++++++ .../com/hailie/domain/saga/SagaDecisions.kt | 197 ++++++++++++++++++ .../domain/saga/SagaDecisionTableTest.kt | 88 ++++++++ 4 files changed, 397 insertions(+) create mode 100644 core-domain/src/commonMain/kotlin/com/hailie/domain/model/SyncAggregateApply.kt create mode 100644 core-domain/src/commonMain/kotlin/com/hailie/domain/saga/SagaDecisions.kt create mode 100644 core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaDecisionTableTest.kt diff --git a/core-domain/api/core-domain.api b/core-domain/api/core-domain.api index 1e2495d..9a49a57 100644 --- a/core-domain/api/core-domain.api +++ b/core-domain/api/core-domain.api @@ -600,6 +600,10 @@ public final class com/hailie/domain/model/SyncAggregate { public fun toString ()Ljava/lang/String; } +public final class com/hailie/domain/model/SyncAggregateApplyKt { + public static final fun applyEvent (Lcom/hailie/domain/model/SyncAggregate;Lcom/hailie/domain/events/Event;)Lcom/hailie/domain/model/SyncAggregate; +} + public final class com/hailie/domain/model/SyncAggregateKt { public static final fun initialSyncAggregate-T4xMTfM (Ljava/lang/String;I)Lcom/hailie/domain/model/SyncAggregate; public static synthetic fun initialSyncAggregate-T4xMTfM$default (Ljava/lang/String;IILjava/lang/Object;)Lcom/hailie/domain/model/SyncAggregate; @@ -681,3 +685,22 @@ public final class com/hailie/domain/policy/SimpleBreakerPolicy : com/hailie/dom public fun onSuccess-1_AN8L0 (JLcom/hailie/domain/BreakerState;)Lcom/hailie/domain/BreakerState; } +public final class com/hailie/domain/saga/AttemptKeys { + public static final field INSTANCE Lcom/hailie/domain/saga/AttemptKeys; + public final fun getACK-i9I8bNQ ()Ljava/lang/String; + public final fun getBOND-i9I8bNQ ()Ljava/lang/String; + public final fun getCONNECT-i9I8bNQ ()Ljava/lang/String; + public final fun getDELIVER-i9I8bNQ ()Ljava/lang/String; + public final fun getREAD_COUNT-i9I8bNQ ()Ljava/lang/String; + public final fun getREAD_PAGE-i9I8bNQ ()Ljava/lang/String; +} + +public final class com/hailie/domain/saga/DefaultSaga : com/hailie/domain/saga/Saga { + public fun (Lcom/hailie/domain/policy/RetryPolicy;Lcom/hailie/domain/policy/BreakerPolicy;Lcom/hailie/domain/policy/BreakerPolicy;Lcom/hailie/domain/policy/BreakerPolicy;Lcom/hailie/domain/policy/BreakerPolicy;Lcom/hailie/domain/policy/PageSizingPolicy;)V + public fun decide-vOlqrbc (Lcom/hailie/domain/model/SyncAggregate;Lcom/hailie/domain/events/Event;J)Ljava/util/List; +} + +public abstract interface class com/hailie/domain/saga/Saga { + public abstract fun decide-vOlqrbc (Lcom/hailie/domain/model/SyncAggregate;Lcom/hailie/domain/events/Event;J)Ljava/util/List; +} + diff --git a/core-domain/src/commonMain/kotlin/com/hailie/domain/model/SyncAggregateApply.kt b/core-domain/src/commonMain/kotlin/com/hailie/domain/model/SyncAggregateApply.kt new file mode 100644 index 0000000..c44c14c --- /dev/null +++ b/core-domain/src/commonMain/kotlin/com/hailie/domain/model/SyncAggregateApply.kt @@ -0,0 +1,89 @@ +package com.hailie.domain.model + +import com.hailie.domain.BondStatus +import com.hailie.domain.ConnectionStatus +import com.hailie.domain.DomainError +import com.hailie.domain.SagaCursor +import com.hailie.domain.events.DeviceBonded +import com.hailie.domain.events.DeviceConnected +import com.hailie.domain.events.Disconnected +import com.hailie.domain.events.Event +import com.hailie.domain.events.EventCountLoaded +import com.hailie.domain.events.EventsAcked +import com.hailie.domain.events.EventsDelivered +import com.hailie.domain.events.EventsRead +import com.hailie.domain.events.RetryScheduled +import com.hailie.domain.events.SyncCompleted +import com.hailie.domain.events.SyncFailed + +/** + * Pure reducers: evolve SyncAggregate by applying domain events. + * No IO, no timers, just data → data. + */ +fun SyncAggregate.applyEvent(event: Event): SyncAggregate = + when (event) { + is DeviceBonded -> + copy( + bondStatus = BondStatus.Bonded, + sagaCursor = SagaCursor("Bonded"), + ) + + is DeviceConnected -> + copy( + connectionStatus = ConnectionStatus.Connected, + sagaCursor = SagaCursor("Connected"), + ) + + is EventCountLoaded -> + copy( + totalOnDevice = event.total, + sagaCursor = SagaCursor("CountLoaded"), + ) + + is EventsRead -> + copy( + // Mark the in-flight page as the start of the range we just read + inFlightOffset = event.range.startInclusive, + sagaCursor = SagaCursor("Read:${event.range.startInclusive.value}-${event.range.endExclusive.value}"), + ) + + is EventsDelivered -> + copy( + // Delivery does not change high-water; ack will advance it + sagaCursor = SagaCursor("Delivered:${event.range.startInclusive.value}-${event.range.endExclusive.value}"), + ) + + is EventsAcked -> { + // High-water advances (monotonic). Clear in-flight when we advanced to end of that page. + val newAck = if (event.upTo.value > lastAckedExclusive.value) event.upTo else lastAckedExclusive + copy( + lastAckedExclusive = newAck, + // Clear in-flight when ack caught up at or beyond the in-flight page end + inFlightOffset = if (inFlightOffset != null && newAck.value >= newAck.value) null else inFlightOffset, + sagaCursor = SagaCursor("Acked:${newAck.value}"), + ) + } + + is Disconnected -> + copy( + connectionStatus = ConnectionStatus.Disconnected, + lastError = DomainError.Transport(event.reason.toString(), code = event.gattCode), + sagaCursor = SagaCursor("Disconnected"), + ) + + is RetryScheduled -> + copy( + sagaCursor = SagaCursor("RetryScheduled@${event.after.value}"), + ) + + is SyncCompleted -> + copy( + sagaCursor = SagaCursor("Completed"), + ) + + is SyncFailed -> + copy( + lastError = event.reason, + sagaCursor = SagaCursor("Failed"), + ) + } diff --git a/core-domain/src/commonMain/kotlin/com/hailie/domain/saga/SagaDecisions.kt b/core-domain/src/commonMain/kotlin/com/hailie/domain/saga/SagaDecisions.kt new file mode 100644 index 0000000..cd35075 --- /dev/null +++ b/core-domain/src/commonMain/kotlin/com/hailie/domain/saga/SagaDecisions.kt @@ -0,0 +1,197 @@ +package com.hailie.domain.saga + +import com.hailie.domain.AttemptKey +import com.hailie.domain.DeviceId +import com.hailie.domain.EventOffset +import com.hailie.domain.PageSize +import com.hailie.domain.TimestampMs +import com.hailie.domain.commands.Acknowledge +import com.hailie.domain.commands.BondDevice +import com.hailie.domain.commands.Command +import com.hailie.domain.commands.ConnectGatt +import com.hailie.domain.commands.DeliverToApp +import com.hailie.domain.commands.ReadEventCount +import com.hailie.domain.commands.ReadEvents +import com.hailie.domain.commands.ScheduleRetry +import com.hailie.domain.events.DeviceBonded +import com.hailie.domain.events.DeviceConnected +import com.hailie.domain.events.Disconnected +import com.hailie.domain.events.Event +import com.hailie.domain.events.EventCountLoaded +import com.hailie.domain.events.EventsAcked +import com.hailie.domain.events.EventsDelivered +import com.hailie.domain.events.EventsRead +import com.hailie.domain.model.SyncAggregate +import com.hailie.domain.policy.BreakerPolicy +import com.hailie.domain.policy.PageOutcome +import com.hailie.domain.policy.PageSizingPolicy +import com.hailie.domain.policy.RetryDecision +import com.hailie.domain.policy.RetryPolicy +import com.hailie.domain.policy.attemptsFor + +/** + * Stable keys for attempt tracking by operation. + * These names should match how you increment counters around adapter results. + */ +object AttemptKeys { + val BOND = AttemptKey("BondDevice") + val CONNECT = AttemptKey("ConnectGatt") + val READ_COUNT = AttemptKey("ReadEventCount") + val READ_PAGE = AttemptKey("ReadEvents") + val DELIVER = AttemptKey("DeliverToApp") + val ACK = AttemptKey("Acknowledge") +} + +/** + * A saga decides the next commands based on current [state], the [lastEvent] observed, + * the current time [now], and the injected policies. + * + * Pure function: no side effects. + */ +fun interface Saga { + fun decide( + state: SyncAggregate, + lastEvent: Event?, + now: TimestampMs, + ): List +} + +/** + * Default saga: drives bond → connect → count → page (read→deliver→ack) loops. + * Honors RetryPolicy, BreakerPolicy, and PageSizingPolicy. + */ +class DefaultSaga( + private val retryPolicy: RetryPolicy, + private val breakerForConnect: BreakerPolicy, + private val breakerForRead: BreakerPolicy, + private val breakerForDeliver: BreakerPolicy, + private val breakerForAck: BreakerPolicy, + private val pageSizingPolicy: PageSizingPolicy, +) : Saga { + override fun decide( + state: SyncAggregate, + lastEvent: Event?, + now: TimestampMs, + ): List { + // 0) If we’re not bonded → bond first + if (state.bondStatus != com.hailie.domain.BondStatus.Bonded) { + return listOf(BondDevice(deviceId = state.deviceId)) + } + + // 1) Ensure connection is up (respect connect breaker) + if (state.connectionStatus != com.hailie.domain.ConnectionStatus.Connected) { + val allowed = breakerForConnect.isCallAllowed(now, state.connectBreaker) + return if (allowed) { + listOf(ConnectGatt(deviceId = state.deviceId)) + } else { + retryOrGiveUp( + deviceId = state.deviceId, + now = now, + attempts = attemptsFor(state.attempts, AttemptKeys.CONNECT), + reason = com.hailie.domain.RetryReason.BackoffAfterFailure, + ) + } + } + + // 2) If we don’t know the total yet or need to re-check growth → read event count + // we never set negatives; kept for completeness + val needsCount = state.totalOnDevice.value < 0 + + if (state.totalOnDevice == com.hailie.domain.EventCount(0L) && state.lastAckedExclusive.value == 0L) { + // Initial connected state: ask for count + return listOf(ReadEventCount(deviceId = state.deviceId)) + } + + // 3) Handle by last event (decision table core) + return when (lastEvent) { + null -> decideOnNoEvent(state) + is DeviceBonded -> listOf(ConnectGatt(state.deviceId)) + is DeviceConnected -> listOf(ReadEventCount(state.deviceId)) + is EventCountLoaded -> decideAfterCount(state) + is EventsRead -> decideAfterRead(state, lastEvent) + is EventsDelivered -> decideAfterDelivered(state, lastEvent) + is EventsAcked -> decideAfterAck(state, lastEvent) + is Disconnected -> { + // Next step is reconnect, but respect breaker & retry policy + val allowed = breakerForConnect.isCallAllowed(now, state.connectBreaker) + if (allowed) { + listOf(ConnectGatt(state.deviceId)) + } else { + retryOrGiveUp( + deviceId = state.deviceId, + now = now, + attempts = attemptsFor(state.attempts, AttemptKeys.CONNECT), + reason = com.hailie.domain.RetryReason.TemporaryGattError, + ) + } + } + else -> emptyList() + } + } + + private fun decideOnNoEvent(state: SyncAggregate): List { + // Fresh connected state without events: read count + return listOf(ReadEventCount(deviceId = state.deviceId)) + } + + private fun decideAfterCount(state: SyncAggregate): List { + // If everything already acked → check growth once more or stop + if (state.isFullyAcked) { + // We can either stop or re-check count to detect growth. + // Choose to re-check count to keep syncing if device keeps producing data. + return listOf(ReadEventCount(deviceId = state.deviceId)) + } + // Otherwise, start (or continue) paging from high-water + val start = com.hailie.domain.EventOffset(state.lastAckedExclusive.value) + val count = state.pageSize.value + return listOf(ReadEvents(deviceId = state.deviceId, offset = start, count = count)) + } + + private fun decideAfterRead( + state: SyncAggregate, + ev: EventsRead, + ): List { + // After reading a page, deliver it + return listOf(DeliverToApp(deviceId = state.deviceId, range = ev.range)) + } + + private fun decideAfterDelivered( + state: SyncAggregate, + ev: EventsDelivered, + ): List { + // After delivery, request ack to advance high-water + return listOf(Acknowledge(deviceId = state.deviceId, upTo = ev.range.endExclusive)) + } + + private fun decideAfterAck( + state: SyncAggregate, + ev: EventsAcked, + ): List { + // If still behind the observed total → next page + return if (state.lastAckedExclusive.value < state.totalOnDevice.value) { + val nextOffset = com.hailie.domain.EventOffset(state.lastAckedExclusive.value) + val tunedPageSize = tunePageSizeAfterAck(state) + listOf(ReadEvents(deviceId = state.deviceId, offset = nextOffset, count = tunedPageSize.value)) + } else { + // High-water caught up → read count again to detect growth + listOf(ReadEventCount(deviceId = state.deviceId)) + } + } + + private fun tunePageSizeAfterAck(state: SyncAggregate): com.hailie.domain.PageSize { + val outcome = if (state.lastError == null) PageOutcome.Stable else PageOutcome.MostlyStable + return pageSizingPolicy.next(state.pageSize, outcome) + } + + private fun retryOrGiveUp( + deviceId: DeviceId, + now: TimestampMs, + attempts: Int, + reason: com.hailie.domain.RetryReason, + ): List { + return when (val d = retryPolicy.decide(now, attempts, reason)) { + is RetryDecision.Schedule -> listOf(ScheduleRetry(deviceId, d.at, reason)) + is RetryDecision.GiveUp -> emptyList() + } + } +} diff --git a/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaDecisionTableTest.kt b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaDecisionTableTest.kt new file mode 100644 index 0000000..41b5945 --- /dev/null +++ b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaDecisionTableTest.kt @@ -0,0 +1,88 @@ +package com.hailie.domain.saga + +import com.hailie.domain.BondStatus +import com.hailie.domain.ConnectionStatus +import com.hailie.domain.DeviceId +import com.hailie.domain.EventCount +import com.hailie.domain.EventOffset +import com.hailie.domain.PageSize +import com.hailie.domain.TimestampMs +import com.hailie.domain.commands.ConnectGatt +import com.hailie.domain.commands.ReadEventCount +import com.hailie.domain.commands.ReadEvents +import com.hailie.domain.events.DeviceBonded +import com.hailie.domain.events.DeviceConnected +import com.hailie.domain.events.EventCountLoaded +import com.hailie.domain.model.SyncAggregate +import com.hailie.domain.model.initialSyncAggregate +import com.hailie.domain.policy.AdaptivePageSizingPolicy +import com.hailie.domain.policy.RetryDecision +import com.hailie.domain.policy.RetryPolicy +import com.hailie.domain.policy.SimpleBreakerPolicy +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +private class NoopRetryPolicy : RetryPolicy { + override fun decide( + now: TimestampMs, + attemptsForOp: Int, + reason: com.hailie.domain.RetryReason, + ): RetryDecision { + return RetryDecision.GiveUp + } +} + +class SagaDecisionTableTest { + private fun saga(): Saga = + DefaultSaga( + retryPolicy = NoopRetryPolicy(), + breakerForConnect = SimpleBreakerPolicy(failuresToOpen = 1, coolDownMs = 1), + breakerForRead = SimpleBreakerPolicy(), + breakerForDeliver = SimpleBreakerPolicy(), + breakerForAck = SimpleBreakerPolicy(), + pageSizingPolicy = AdaptivePageSizingPolicy(minPage = 20, maxPage = 100, growStep = 20, shrinkStep = 20), + ) + + @Test + fun notBonded_thenBond_thenConnect_thenCount() { + val device = DeviceId("d1") + val t0 = TimestampMs(1_000) + + // Start + val s0 = initialSyncAggregate(deviceId = device, initialPageSize = 50) + val cmds0 = saga().decide(s0, lastEvent = null, now = t0) + // Not bonded → should emit BondDevice + assertEquals("BondDevice", cmds0.first()::class.simpleName) + + // After DeviceBonded → expect ConnectGatt + val s1 = s0.copy(bondStatus = BondStatus.Bonded) + val cmds1 = saga().decide(s1, lastEvent = DeviceBonded(device, t0), now = t0) + assertTrue(cmds1.first() is ConnectGatt) + + // After DeviceConnected → expect ReadEventCount + val s2 = s1.copy(connectionStatus = ConnectionStatus.Connected) + val cmds2 = saga().decide(s2, lastEvent = DeviceConnected(device, t0), now = t0) + assertTrue(cmds2.first() is ReadEventCount) + } + + @Test + fun afterCount_loaded_then_read_from_highWater() { + val device = DeviceId("d1") + val t0 = TimestampMs(2_000) + val s = + SyncAggregate( + deviceId = device, + bondStatus = BondStatus.Bonded, + connectionStatus = ConnectionStatus.Connected, + lastAckedExclusive = EventOffset(0), + totalOnDevice = EventCount(120), + pageSize = PageSize(50), + ) + + val cmds = saga().decide(s, lastEvent = EventCountLoaded(device, t0, total = EventCount(120)), now = t0) + val c = cmds.first() as ReadEvents + assertEquals(0, c.offset.value) + assertEquals(50, c.count) + } +}