From 758a9c74b7478dea8adf551497930de998495a91 Mon Sep 17 00:00:00 2001 From: Rob Ribeiro Date: Sun, 1 Feb 2026 12:28:54 -0300 Subject: [PATCH] chore: add more tests to saga orchestrations --- .../domain/saga/DomainInvariantsTest.kt | 52 +++++ .../hailie/domain/saga/SagaHappyPathTest.kt | 108 +++++++++ .../hailie/domain/saga/SagaResilienceTest.kt | 214 ++++++++++++++++++ 3 files changed, 374 insertions(+) create mode 100644 core-domain/src/commonTest/kotlin/com/hailie/domain/saga/DomainInvariantsTest.kt create mode 100644 core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaHappyPathTest.kt create mode 100644 core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaResilienceTest.kt diff --git a/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/DomainInvariantsTest.kt b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/DomainInvariantsTest.kt new file mode 100644 index 0000000..2ad94db --- /dev/null +++ b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/DomainInvariantsTest.kt @@ -0,0 +1,52 @@ +package com.hailie.domain.model + +import com.hailie.domain.DeviceId +import com.hailie.domain.DomainError +import com.hailie.domain.EventOffset +import com.hailie.domain.TimestampMs +import com.hailie.domain.events.Disconnected +import com.hailie.domain.events.EventsAcked +import com.hailie.domain.events.SyncFailed +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class DomainInvariantsTest { + @Test + fun lastAckedExclusive_is_monotonic() { + val device = DeviceId("mono") + val t = TimestampMs(10_000) + val s0 = initialSyncAggregate(device) + + // Apply ack to 50 + val s1 = s0.applyEvent(EventsAcked(device, t, upTo = EventOffset(50))) + assertEquals(50, s1.lastAckedExclusive.value) + + // Apply ack to 40 (lower) → must NOT decrease + val s2 = s1.applyEvent(EventsAcked(device, t, upTo = EventOffset(40))) + assertEquals(50, s2.lastAckedExclusive.value) + + // Apply ack to 50 again (same) → no double-advance + val s3 = s2.applyEvent(EventsAcked(device, t, upTo = EventOffset(50))) + assertEquals(50, s3.lastAckedExclusive.value) + + // Apply ack to 80 → advances + val s4 = s3.applyEvent(EventsAcked(device, t, upTo = EventOffset(80))) + assertEquals(80, s4.lastAckedExclusive.value) + } + + @Test + fun permission_and_user_action_errors_are_surfacable_via_events() { + val device = DeviceId("perm") + val t = TimestampMs(11_000) + val s0 = initialSyncAggregate(device) + + // Surface a permission required error via SyncFailed + val s1 = s0.applyEvent(SyncFailed(device, t, reason = DomainError.PermissionRequired("android.permission.BLUETOOTH_CONNECT"))) + assertTrue(s1.lastError is DomainError.PermissionRequired) + + // Surface a transport-level disconnect error via Disconnected + val s2 = s1.applyEvent(Disconnected(device, t, reason = com.hailie.domain.DisconnectReason.Timeout, gattCode = 133)) + assertTrue(s2.lastError is DomainError.Transport) + } +} diff --git a/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaHappyPathTest.kt b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaHappyPathTest.kt new file mode 100644 index 0000000..7641711 --- /dev/null +++ b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaHappyPathTest.kt @@ -0,0 +1,108 @@ +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.EventRange +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.ConnectGatt +import com.hailie.domain.commands.DeliverToApp +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.events.EventsAcked +import com.hailie.domain.events.EventsDelivered +import com.hailie.domain.events.EventsRead +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 RetryGiveUp : RetryPolicy { + override fun decide( + now: TimestampMs, + attemptsForOp: Int, + reason: com.hailie.domain.RetryReason, + ): RetryDecision { + return RetryDecision.GiveUp + } +} + +private fun defaultSaga(): Saga = + DefaultSaga( + retryPolicy = RetryGiveUp(), + breakerForConnect = SimpleBreakerPolicy(failuresToOpen = 1, coolDownMs = 1), + breakerForRead = SimpleBreakerPolicy(), + breakerForDeliver = SimpleBreakerPolicy(), + breakerForAck = SimpleBreakerPolicy(), + pageSizingPolicy = AdaptivePageSizingPolicy(minPage = 20, maxPage = 100, growStep = 20, shrinkStep = 20), + ) + +class SagaHappyPathTest { + @Test + fun decision_sequence_happy_path() { + val device = DeviceId("d1") + val t = TimestampMs(1_000) + + // Start: not bonded → BondDevice + val s0 = initialSyncAggregate(deviceId = device, initialPageSize = 50) + val c0 = defaultSaga().decide(s0, lastEvent = null, now = t) + assertTrue(c0.first() is BondDevice) + + // DeviceBonded → ConnectGatt + val s1 = s0.copy(bondStatus = BondStatus.Bonded) + val c1 = defaultSaga().decide(s1, lastEvent = DeviceBonded(device, t), now = t) + assertTrue(c1.first() is ConnectGatt) + + // DeviceConnected → ReadEventCount + val s2 = s1.copy(connectionStatus = ConnectionStatus.Connected) + val c2 = defaultSaga().decide(s2, lastEvent = DeviceConnected(device, t), now = t) + assertTrue(c2.first() is ReadEventCount) + + // EventCountLoaded(total=120) → ReadEvents(offset=0, count=50) + val s3 = + s2.copy( + totalOnDevice = EventCount(120), + pageSize = PageSize(50), + lastAckedExclusive = EventOffset(0), + ) + val c3 = defaultSaga().decide(s3, lastEvent = EventCountLoaded(device, t, total = EventCount(120)), now = t) + val read1 = c3.first() as ReadEvents + assertEquals(0, read1.offset.value) + assertEquals(50, read1.count) + + // EventsRead([0,50)) → DeliverToApp([0,50)) + val range1 = EventRange(EventOffset(0), EventOffset(50)) + val c4 = defaultSaga().decide(s3, lastEvent = EventsRead(device, t, range1), now = t) + val deliver1 = c4.first() as DeliverToApp + assertEquals(range1, deliver1.range) + + // EventsDelivered([0,50)) → Acknowledge(upTo=50) + val c5 = defaultSaga().decide(s3, lastEvent = EventsDelivered(device, t, range1), now = t) + val ack1 = c5.first() as Acknowledge + assertEquals(50, ack1.upTo.value) + + // After ack to 50 and total=120 → next ReadEvents(offset=50, count=70) due to tuning (+20 growStep) + val s4 = s3.copy(lastAckedExclusive = EventOffset(50)) + val c6 = defaultSaga().decide(s4, lastEvent = EventsAcked(device, t, upTo = EventOffset(50)), now = t) + val read2 = c6.first() as ReadEvents + assertEquals(50, read2.offset.value) + assertEquals(70, read2.count) // tuned from 50 → 70 (Stable outcome, +20 growStep) + + // If we caught up (ack upTo == total), next step is ReadEventCount (re-check growth) + val sDone = s3.copy(lastAckedExclusive = EventOffset(120)) + val cDone = defaultSaga().decide(sDone, lastEvent = EventsAcked(device, t, upTo = EventOffset(120)), now = t) + assertTrue(cDone.first() is ReadEventCount) + } +} diff --git a/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaResilienceTest.kt b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaResilienceTest.kt new file mode 100644 index 0000000..20cd593 --- /dev/null +++ b/core-domain/src/commonTest/kotlin/com/hailie/domain/saga/SagaResilienceTest.kt @@ -0,0 +1,214 @@ +package com.hailie.domain.saga + +import com.hailie.domain.AttemptCounters +import com.hailie.domain.AttemptKey +import com.hailie.domain.BondStatus +import com.hailie.domain.ConnectionStatus +import com.hailie.domain.DeviceId +import com.hailie.domain.DisconnectReason +import com.hailie.domain.DomainError +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.commands.ScheduleRetry +import com.hailie.domain.events.DeviceConnected +import com.hailie.domain.events.Disconnected +import com.hailie.domain.events.EventCountLoaded +import com.hailie.domain.policy.AdaptivePageSizingPolicy +import com.hailie.domain.policy.BreakerPolicy +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 RetryScheduleFixed(private val delayMs: Long, private val cap: Int = Int.MAX_VALUE) : RetryPolicy { + override fun decide( + now: TimestampMs, + attemptsForOp: Int, + reason: com.hailie.domain.RetryReason, + ): RetryDecision { + return if (attemptsForOp >= cap) { + RetryDecision.GiveUp + } else { + RetryDecision.Schedule(TimestampMs(now.value + delayMs)) + } + } +} + +private class BreakerAlwaysAllowed : BreakerPolicy { + override fun isCallAllowed( + now: TimestampMs, + state: com.hailie.domain.BreakerState, + ): Boolean = true + + override fun onSuccess( + now: TimestampMs, + state: com.hailie.domain.BreakerState, + ): com.hailie.domain.BreakerState = state + + override fun onFailure( + now: TimestampMs, + state: com.hailie.domain.BreakerState, + error: DomainError, + ): com.hailie.domain.BreakerState = state +} + +private class BreakerAllowedAfter(private val allowAtMs: Long) : BreakerPolicy { + override fun isCallAllowed( + now: TimestampMs, + state: com.hailie.domain.BreakerState, + ): Boolean { + return now.value >= allowAtMs + } + + override fun onSuccess( + now: TimestampMs, + state: com.hailie.domain.BreakerState, + ): com.hailie.domain.BreakerState = state + + override fun onFailure( + now: TimestampMs, + state: com.hailie.domain.BreakerState, + error: DomainError, + ): com.hailie.domain.BreakerState = state +} + +private fun sagaWith( + retry: RetryPolicy, + connectBreaker: BreakerPolicy, +): Saga = + DefaultSaga( + retryPolicy = retry, + breakerForConnect = connectBreaker, + breakerForRead = SimpleBreakerPolicy(), + breakerForDeliver = SimpleBreakerPolicy(), + breakerForAck = SimpleBreakerPolicy(), + pageSizingPolicy = AdaptivePageSizingPolicy(minPage = 20, maxPage = 100, growStep = 20, shrinkStep = 20), + ) + +/** + * 1) Disconnect during paging → reconnect → resume at lastAckedExclusive + * 2) Breaker denies at t0 (schedules retry) and allows at t1 (connects) + * 3) Capped retries → stop scheduling when attempts reach cap + */ +class SagaResilienceTest { + @Test + fun disconnect_during_paging_reconnect_and_resume_from_high_water() { + val device = DeviceId("dev-paging") + val t0 = TimestampMs(10_000) + + // We were in the middle of paging previously (ack at 50, total 120, pageSize 50), but got disconnected. + val s0 = + com.hailie.domain.model.SyncAggregate( + deviceId = device, + bondStatus = BondStatus.Bonded, + connectionStatus = ConnectionStatus.Disconnected, + lastAckedExclusive = EventOffset(50), + totalOnDevice = EventCount(120), + pageSize = PageSize(50), + ) + + // On Disconnected: breaker allows → ConnectGatt + val commands0 = + sagaWith( + retry = RetryScheduleFixed(delayMs = 1_000), + connectBreaker = BreakerAlwaysAllowed(), + ).decide(s0, lastEvent = Disconnected(device, t0, reason = DisconnectReason.Timeout, gattCode = null), now = t0) + assertTrue(commands0.first() is ConnectGatt) + + // After connected: ReadEventCount + val s1 = s0.copy(connectionStatus = ConnectionStatus.Connected) + val commands1 = + sagaWith( + retry = RetryScheduleFixed(delayMs = 1_000), + connectBreaker = BreakerAlwaysAllowed(), + ).decide(s1, lastEvent = DeviceConnected(device, t0), now = t0) + assertTrue(commands1.first() is ReadEventCount) + + // Count loaded (still behind): next ReadEvents must resume at offset=50 with current pageSize (50 here) + val s2 = s1.copy(totalOnDevice = EventCount(120)) + val commands2 = + sagaWith( + retry = RetryScheduleFixed(delayMs = 1_000), + connectBreaker = BreakerAlwaysAllowed(), + ).decide(s2, lastEvent = EventCountLoaded(device, t0, total = EventCount(120)), now = t0) + + val read = commands2.first() as ReadEvents + assertEquals(50, read.offset.value) + assertEquals(50, read.count) + } + + @Test + fun breaker_half_open_probe_denied_then_allowed() { + val device = DeviceId("dev-halfopen") + val t0 = TimestampMs(5_000) + val t1 = TimestampMs(6_000) + + val s = + com.hailie.domain.model.SyncAggregate( + deviceId = device, + bondStatus = BondStatus.Bonded, + connectionStatus = ConnectionStatus.Disconnected, + lastAckedExclusive = EventOffset(0), + totalOnDevice = EventCount(0), + pageSize = PageSize(50), + ) + + // At t0 breaker denies → should schedule retry (since we injected a scheduling retry policy) + val commands0 = + sagaWith( + retry = RetryScheduleFixed(delayMs = 500), + // denies at 5_000 + connectBreaker = BreakerAllowedAfter(allowAtMs = 5_500), + ).decide(s, lastEvent = Disconnected(device, t0, reason = DisconnectReason.Timeout, gattCode = null), now = t0) + assertTrue(commands0.first() is ScheduleRetry) + + // At t1 breaker allows → should ConnectGatt + val commands1 = + sagaWith( + retry = RetryScheduleFixed(delayMs = 500), + // allows at 6_000 + connectBreaker = BreakerAllowedAfter(allowAtMs = 5_500), + ).decide(s, lastEvent = Disconnected(device, t1, reason = DisconnectReason.Timeout, gattCode = null), now = t1) + assertTrue(commands1.first() is ConnectGatt) + } + + @Test + fun capped_retries_stop_scheduling_when_attempts_reach_cap() { + val device = DeviceId("dev-cap") + val now = TimestampMs(7_000) + + // Already attempted once for ConnectGatt + val attempts = + AttemptCounters( + byKey = mapOf(AttemptKey("ConnectGatt") to 1), + ) + + val s = + com.hailie.domain.model.SyncAggregate( + deviceId = device, + bondStatus = BondStatus.Bonded, + connectionStatus = ConnectionStatus.Disconnected, + lastAckedExclusive = EventOffset(0), + totalOnDevice = EventCount(0), + pageSize = PageSize(50), + attempts = attempts, + ) + + // Retry policy cap = 1 → with attemptsForOp == 1, policy returns GiveUp → no commands should be emitted + val commands = + sagaWith( + retry = RetryScheduleFixed(delayMs = 500, cap = 1), + // keep denying path to invoke retry logic + connectBreaker = BreakerAllowedAfter(allowAtMs = 100_000), + ).decide(s, lastEvent = Disconnected(device, now, reason = DisconnectReason.Timeout, gattCode = null), now = now) + + assertTrue(commands.isEmpty()) + } +}