From c2c7f8e93bab23ec5d546c21fcf2740c20d05e03 Mon Sep 17 00:00:00 2001 From: Iceeyyou2 <126117799+Iceeyyou2@users.noreply.github.com> Date: Mon, 31 Aug 2026 07:32:18 +0000 Subject: [PATCH] feat: WebSocket heartbeat, multi-vault subscription, and connection status (#252, #253, #254, #255) --- .../services/VaultEventSocket.kt | 112 ++++++++++- .../java/com/ethosprotocol/ui/ViewModels.kt | 36 +++- .../com/ethosprotocol/ui/screens/Screens.kt | 42 +++- .../com/ethosprotocol/VaultEventSocketTest.kt | 189 +++++++++++++++++- .../com/ethosprotocol/VaultViewModelTest.kt | 7 +- .../Sources/Services/VaultEventSocket.swift | 61 ++++++ .../Sources/ViewModels/Stores.swift | 10 + ios/EthosProtocol/Sources/Views/Views.swift | 47 +++++ .../Tests/VaultEventSocketTests.swift | 186 +++++++++++++++++ shared/api-contract.md | 36 ++++ 10 files changed, 711 insertions(+), 15 deletions(-) diff --git a/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt b/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt index 5af56bc..83e4db2 100644 --- a/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt +++ b/android/app/src/main/java/com/ethosprotocol/services/VaultEventSocket.kt @@ -14,11 +14,17 @@ import javax.inject.Inject import javax.inject.Singleton import kotlin.random.Random import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.cancel +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch import kotlinx.serialization.json.Json // Reconnect/backoff schedule for VaultEventSocket. Unlike RetryPolicy (bounded @@ -49,6 +55,11 @@ data class ReconnectBackoff( } } +// Connection state for the WebSocket. FALLBACK_TO_POLLING is included for parity +// with the iOS API contract; on Android the socket retries indefinitely unless +// maxReconnectAttempts is set to a finite value. +enum class ConnectionState { DISCONNECTED, CONNECTING, CONNECTED, FALLBACK_TO_POLLING } + // Client for the `wss://.../ws?vault_id={id}` endpoint (shared/api-contract.md). // Reuses ApiClient's HttpClient/WebSockets plugin rather than a second client. // A dropped or failed connection reconnects with exponential backoff @@ -58,7 +69,9 @@ data class ReconnectBackoff( class VaultEventSocket( private val apiClient: ApiClient, private val tokenProvider: TokenProvider, - private val backoff: ReconnectBackoff = ReconnectBackoff.default + private val backoff: ReconnectBackoff = ReconnectBackoff.default, + val heartbeatIntervalMillis: Long = 30_000L, + val maxReconnectAttempts: Int = Int.MAX_VALUE ) { // Distinct @Inject constructor for the same reason as ApiClient's: Dagger's // codegen calls the full-arg constructor directly and ignores Kotlin default @@ -66,6 +79,9 @@ class VaultEventSocket( @Inject constructor(apiClient: ApiClient, tokenProvider: TokenProvider) : this(apiClient, tokenProvider, ReconnectBackoff.default) + private val _connectionState = MutableStateFlow(ConnectionState.DISCONNECTED) + val connectionState: StateFlow = _connectionState.asStateFlow() + // internal (not private): tests replace this to simulate connect failures/frames // without a real server, matching how ApiClient exposes its engine to tests. internal var openSession: suspend (String) -> WebSocketSession = { vaultId -> @@ -78,21 +94,107 @@ class VaultEventSocket( fun events(vaultId: String): Flow = flow { var attempt = 0 while (currentCoroutineContext().isActive) { + _connectionState.value = ConnectionState.CONNECTING try { val session = openSession(vaultId) attempt = 0 - for (frame in session.incoming) { - if (frame is Frame.Text) { - runCatching { Json.decodeFromString(frame.readText()) } - .onSuccess { emit(it) } + _connectionState.value = ConnectionState.CONNECTED + coroutineScope { + // Periodic client-side heartbeat to detect silent TCP drops. + launch { + while (isActive) { + delay(heartbeatIntervalMillis) + try { + session.send(Frame.Text("""{ "type": "ping" }""")) + } catch (_: Exception) { + // Send failed — session is dead; cancel scope to stop incoming loop too. + cancel() + } + } + } + for (frame in session.incoming) { + if (frame is Frame.Text) { + val text = frame.readText() + runCatching { Json.decodeFromString(text) } + .onSuccess { event -> + if (event.type == "ping") { + // Server keepalive — reply with pong. + runCatching { session.send(Frame.Text("""{ "type": "pong" }""")) } + } + emit(event) + } + } } + cancel() // incoming closed normally — stop heartbeat too } } catch (e: CancellationException) { throw e } catch (e: Exception) { // Connection failed or dropped — fall through to backoff and reconnect. + _connectionState.value = ConnectionState.DISCONNECTED } if (!currentCoroutineContext().isActive) break + if (attempt >= maxReconnectAttempts) { + _connectionState.value = ConnectionState.FALLBACK_TO_POLLING + break + } + backoff.sleep(backoff.delayForAttempt(attempt)) + attempt++ + } + } + + fun events(vaultIds: List): Flow = flow { + if (vaultIds.isEmpty()) return@flow + // Use the first vault ID as the primary connection URL, then send a subscribe + // message for the rest once connected. + var attempt = 0 + while (currentCoroutineContext().isActive) { + _connectionState.value = ConnectionState.CONNECTING + try { + val session = openSession(vaultIds[0]) + attempt = 0 + _connectionState.value = ConnectionState.CONNECTED + // Send multi-vault subscribe message post-connect + if (vaultIds.size > 1) { + val idsJson = vaultIds.joinToString(",") { "\"$it\"" } + runCatching { session.send(Frame.Text("""{ "type": "subscribe", "vault_ids": [$idsJson] }""")) } + } + coroutineScope { + launch { + while (isActive) { + delay(heartbeatIntervalMillis) + try { + session.send(Frame.Text("""{ "type": "ping" }""")) + } catch (_: Exception) { + cancel() + } + } + } + for (frame in session.incoming) { + if (frame is Frame.Text) { + val text = frame.readText() + runCatching { Json.decodeFromString(text) } + .onSuccess { event -> + if (event.type == "ping") { + runCatching { session.send(Frame.Text("""{ "type": "pong" }""")) } + } + emit(event) + } + } + } + cancel() + } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + // connection failed + _connectionState.value = ConnectionState.DISCONNECTED + } + if (!currentCoroutineContext().isActive) break + if (attempt >= maxReconnectAttempts) { + _connectionState.value = ConnectionState.FALLBACK_TO_POLLING + break + } backoff.sleep(backoff.delayForAttempt(attempt)) attempt++ } diff --git a/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt b/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt index 6504748..239231f 100644 --- a/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt +++ b/android/app/src/main/java/com/ethosprotocol/ui/ViewModels.kt @@ -352,7 +352,8 @@ data class VaultUiState( val hasMore: Boolean = false, val error: String? = null, val isOffline: Boolean = false, - val beneficiaryUpdated: Boolean = false + val beneficiaryUpdated: Boolean = false, + val socketConnectionState: com.ethosprotocol.services.ConnectionState = com.ethosprotocol.services.ConnectionState.DISCONNECTED ) private const val PAGE_SIZE = 20 @@ -372,6 +373,14 @@ class VaultViewModel @Inject constructor( private var nextCursor: String? = null private val eventJobs = mutableMapOf() + init { + viewModelScope.launch { + vaultEventSocket.connectionState.collect { state -> + _state.update { it.copy(socketConnectionState = state) } + } + } + } + fun load() = viewModelScope.launch { _state.update { it.copy(isLoading = true, error = null) } when (val result = apiClient.listVaults(limit = PAGE_SIZE)) { @@ -463,13 +472,28 @@ class VaultViewModel @Inject constructor( // check-ins/deposits/withdrawals made elsewhere (another device, an expiry) // update this list in place instead of requiring a manual refresh. private fun subscribeToEvents(vaultIds: List) { + // Cancel subscriptions for vaults no longer in the list val currentIds = vaultIds.toSet() eventJobs.keys.filter { it !in currentIds }.forEach { id -> eventJobs.remove(id)?.cancel() } - currentIds.filterNot { eventJobs.containsKey(it) }.forEach { id -> - eventJobs[id] = viewModelScope.launch { - vaultEventSocket.events(id).collect { event -> - val updated = event.vault ?: return@collect - _state.update { s -> s.copy(vaults = s.vaults.map { if (it.id == updated.id) updated else it }) } + // Subscribe new vaults. If more than one ID is requested, use a single multiplexed connection. + if (vaultIds.size > 1 && eventJobs.isEmpty()) { + // Multiplex all vaults over a single connection via subscribe message + val combinedKey = vaultIds.joinToString(",") + if (!eventJobs.containsKey(combinedKey)) { + eventJobs[combinedKey] = viewModelScope.launch { + vaultEventSocket.events(vaultIds).collect { event -> + val updated = event.vault ?: return@collect + _state.update { s -> s.copy(vaults = s.vaults.map { if (it.id == updated.id) updated else it }) } + } + } + } + } else { + currentIds.filterNot { eventJobs.containsKey(it) }.forEach { id -> + eventJobs[id] = viewModelScope.launch { + vaultEventSocket.events(id).collect { event -> + val updated = event.vault ?: return@collect + _state.update { s -> s.copy(vaults = s.vaults.map { if (it.id == updated.id) updated else it }) } + } } } } diff --git a/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt b/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt index d9cba64..eadffeb 100644 --- a/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt +++ b/android/app/src/main/java/com/ethosprotocol/ui/screens/Screens.kt @@ -1382,6 +1382,34 @@ private fun TwoFactorVerifyScreen( } } +// MARK: - Connection Status Badge + +@Composable +fun ConnectionStatusBadge(state: com.ethosprotocol.services.ConnectionState) { + val (label, color, icon) = when (state) { + com.ethosprotocol.services.ConnectionState.CONNECTED -> + Triple("Live", MaterialTheme.colorScheme.primary, Icons.Default.Wifi) + com.ethosprotocol.services.ConnectionState.CONNECTING -> + Triple("Connecting…", MaterialTheme.colorScheme.primary, Icons.Default.Wifi) + com.ethosprotocol.services.ConnectionState.DISCONNECTED -> + Triple("Reconnecting…", MaterialTheme.colorScheme.error, Icons.Default.WifiOff) + com.ethosprotocol.services.ConnectionState.FALLBACK_TO_POLLING -> + Triple("Polling", MaterialTheme.colorScheme.outline, Icons.Default.Refresh) + } + SuggestionChip( + onClick = {}, + label = { + Row(verticalAlignment = Alignment.CenterVertically) { + Icon(icon, contentDescription = null, modifier = Modifier.size(12.dp)) + Spacer(Modifier.width(4.dp)) + Text(label, style = MaterialTheme.typography.labelSmall) + } + }, + modifier = Modifier.testTag("connectionStatus_$label"), + colors = SuggestionChipDefaults.suggestionChipColors(labelColor = color) + ) +} + // MARK: - Vault Detail Screen /** @@ -1395,7 +1423,8 @@ private fun TwoFactorVerifyScreen( fun VaultDetailScreen( vaultId: String, onBack: () -> Unit, - twoFactorVm: TwoFactorViewModel = hiltViewModel() + twoFactorVm: TwoFactorViewModel = hiltViewModel(), + vaultVm: VaultViewModel = hiltViewModel() ) { val state by twoFactorVm.state.collectAsStateWithLifecycle() val context = LocalContext.current @@ -1433,6 +1462,17 @@ fun VaultDetailScreen( Text("Two-Factor Authentication", style = MaterialTheme.typography.titleMedium) Spacer(Modifier.height(8.dp)) + val socketState by vaultVm.state.collectAsStateWithLifecycle() + Row( + modifier = Modifier.fillMaxWidth().padding(bottom = 8.dp), + verticalAlignment = Alignment.CenterVertically + ) { + Text("Connection", style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + modifier = Modifier.weight(1f)) + ConnectionStatusBadge(socketState.socketConnectionState) + } + when { state.isLoading -> CircularProgressIndicator() state.error != null && state.status == null -> { diff --git a/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt b/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt index dbc6f4a..80c504f 100644 --- a/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt +++ b/android/app/src/test/java/com/ethosprotocol/VaultEventSocketTest.kt @@ -4,6 +4,7 @@ import com.ethosprotocol.api.ApiClient import com.ethosprotocol.api.TokenProvider import com.ethosprotocol.models.VaultEvent import com.ethosprotocol.services.ReconnectBackoff +import com.ethosprotocol.services.ConnectionState import com.ethosprotocol.services.VaultEventSocket import io.ktor.websocket.Frame import io.ktor.websocket.WebSocketSession @@ -14,8 +15,10 @@ import java.io.IOException import java.util.concurrent.atomic.AtomicInteger import kotlin.random.Random import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.Json import org.junit.Assert.* @@ -85,7 +88,10 @@ class VaultEventSocketTest { socket.openSession = { if (openAttempts.getAndIncrement() == 0) throw IOException("connection refused") - mockk { every { incoming } returns channel } + mockk { + every { incoming } returns channel + coEvery { send(any()) } returns Unit + } } val received = socket.events("vault-1").take(1).toList() @@ -112,12 +118,16 @@ class VaultEventSocketTest { socket.openSession = { when (openAttempts.getAndIncrement()) { 0 -> throw IOException("connection refused") - 1 -> mockk { every { incoming } returns Channel().apply { close() } } + 1 -> mockk { + every { incoming } returns Channel().apply { close() } + coEvery { send(any()) } returns Unit + } 2 -> throw IOException("connection refused") else -> mockk { every { incoming } returns Channel(capacity = 1).apply { trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) } + coEvery { send(any()) } returns Unit } } } @@ -135,4 +145,179 @@ class VaultEventSocketTest { // further delay is recorded. assertEquals(listOf(999L, 999L, 1_999L), delays) } + + // ── #252: Heartbeat / Ping-Pong ────────────────────────────────────────── + + @Test + fun `events sends pong in response to server ping frame`() = runTest { + val socket = VaultEventSocket(apiClient, tokenProvider, ReconnectBackoff.default) + val sentFrames = mutableListOf() + val channel = Channel(capacity = 2) + val pingEvent = VaultEvent(type = "ping", vault = null) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), pingEvent))) + + val mockSession = mockk { + every { incoming } returns channel + coEvery { send(any()) } answers { sentFrames.add(firstArg()) } + } + socket.openSession = { mockSession } + + // Collect the ping event + val received = socket.events("vault-1").take(1).toList() + assertEquals("ping", received[0].type) + // Should have sent a pong back + assertTrue(sentFrames.any { it is Frame.Text && (it as Frame.Text).readText().contains("pong") }) + } + + @Test + fun `events reconnects after heartbeat send failure (silent connection death)`() = runTest { + val delays = mutableListOf() + val backoff = ReconnectBackoff( + baseDelayMillis = 1_000, + maxDelayMillis = 30_000, + sleep = { delays.add(it) }, + random = maxJitterRandom() + ) + val openAttempts = AtomicInteger(0) + val socket = VaultEventSocket(apiClient, tokenProvider, backoff, heartbeatIntervalMillis = 1L) + + val channel = Channel() // never sends any frames — simulates silent dead connection + val hangingSession = mockk { + every { incoming } returns channel + coEvery { send(any()) } throws IOException("broken pipe") // heartbeat send fails + } + val event = VaultEvent(type = "check_in", vault = null) + val goodChannel = Channel(capacity = 1) + goodChannel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) + val goodSession = mockk { + every { incoming } returns goodChannel + coEvery { send(any()) } returns Unit + } + socket.openSession = { + when (openAttempts.getAndIncrement()) { + 0 -> hangingSession + else -> goodSession + } + } + + val received = socket.events("vault-1").take(1).toList() + assertEquals(listOf(event), received) + assertEquals(2, openAttempts.get()) + } + + // ── #253: Multi-vault subscription ─────────────────────────────────────── + + @Test + fun `events with multiple vault IDs sends subscribe message after connect`() = runTest { + val sentFrames = mutableListOf() + val socket = VaultEventSocket(apiClient, tokenProvider, ReconnectBackoff.default) + + val channel = Channel(capacity = 1) + val event = VaultEvent(type = "check_in", vault = null) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) + + val mockSession = mockk { + every { incoming } returns channel + coEvery { send(any()) } answers { sentFrames.add(firstArg()) } + } + socket.openSession = { mockSession } + + socket.events(listOf("vault-1", "vault-2", "vault-3")).take(1).toList() + + val subscribeFrame = sentFrames.filterIsInstance().find { + it.readText().contains("subscribe") + } + assertNotNull(subscribeFrame) + assertTrue(subscribeFrame!!.readText().contains("vault-2")) + assertTrue(subscribeFrame.readText().contains("vault-3")) + } + + @Test + fun `events routes updates to correct vault when multiplexed`() = runTest { + val socket = VaultEventSocket(apiClient, tokenProvider, ReconnectBackoff.default) + + val vault1 = com.ethosprotocol.models.Vault( + id = "vault-1", owner = "owner", beneficiary = "ben", balance = 0L, + checkInInterval = 86400L, lastCheckIn = "2026-01-01T00:00:00Z", + status = com.ethosprotocol.models.VaultStatus.active + ) + val vault2 = com.ethosprotocol.models.Vault( + id = "vault-2", owner = "owner", beneficiary = "ben", balance = 0L, + checkInInterval = 86400L, lastCheckIn = "2026-01-01T00:00:00Z", + status = com.ethosprotocol.models.VaultStatus.active + ) + val vault1Event = VaultEvent(type = "vault_updated", vault = vault1) + val vault2Event = VaultEvent(type = "vault_updated", vault = vault2) + + val channel = Channel(capacity = 2) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), vault1Event))) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), vault2Event))) + + val mockSession = mockk { + every { incoming } returns channel + coEvery { send(any()) } returns Unit + } + socket.openSession = { mockSession } + + val received = socket.events(listOf("vault-1", "vault-2")).take(2).toList() + assertEquals(2, received.size) + assertEquals("vault-1", received[0].vault?.id) + assertEquals("vault-2", received[1].vault?.id) + } + + // ── #255: Connection Status ─────────────────────────────────────────────── + + @Test + fun `events emits CONNECTED state after successful session open`() = runTest { + val socket = VaultEventSocket(apiClient, tokenProvider, ReconnectBackoff.default) + val states = mutableListOf() + + val channel = Channel(capacity = 1) + val event = VaultEvent(type = "check_in", vault = null) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) + + socket.openSession = { + mockk { + every { incoming } returns channel + coEvery { send(any()) } returns Unit + } + } + + val stateJob = launch { socket.connectionState.collect { states.add(it) } } + socket.events("vault-1").take(1).toList() + stateJob.cancel() + + assertTrue(states.contains(ConnectionState.CONNECTING)) + assertTrue(states.contains(ConnectionState.CONNECTED)) + } + + @Test + fun `events emits DISCONNECTED state after connection failure`() = runTest { + val delays = mutableListOf() + val backoff = ReconnectBackoff( + baseDelayMillis = 1_000, maxDelayMillis = 30_000, + sleep = { delays.add(it) }, random = maxJitterRandom() + ) + val socket = VaultEventSocket(apiClient, tokenProvider, backoff) + val states = mutableListOf() + + val openAttempts = AtomicInteger(0) + val channel = Channel(capacity = 1) + val event = VaultEvent(type = "check_in", vault = null) + channel.trySend(Frame.Text(Json.encodeToString(VaultEvent.serializer(), event))) + + socket.openSession = { + if (openAttempts.getAndIncrement() == 0) throw IOException("refused") + mockk { + every { incoming } returns channel + coEvery { send(any()) } returns Unit + } + } + + val stateJob = launch { socket.connectionState.collect { states.add(it) } } + socket.events("vault-1").take(1).toList() + stateJob.cancel() + + assertTrue(states.contains(ConnectionState.DISCONNECTED)) + } } diff --git a/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt b/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt index 82b8650..c265fac 100644 --- a/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt +++ b/android/app/src/test/java/com/ethosprotocol/VaultViewModelTest.kt @@ -8,6 +8,7 @@ import com.ethosprotocol.models.VaultStatus import com.ethosprotocol.ui.VaultUiState import com.ethosprotocol.ui.VaultViewModel import com.ethosprotocol.api.ApiClient +import com.ethosprotocol.services.ConnectionState import com.ethosprotocol.services.NotificationHelper import com.ethosprotocol.services.PendingAction import com.ethosprotocol.services.PendingActionDao @@ -20,6 +21,8 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.emptyFlow import kotlinx.coroutines.test.* import org.junit.After @@ -43,7 +46,9 @@ class VaultViewModelTest { Dispatchers.setMain(testDispatcher) mockkObject(PendingActionSyncWorker.Companion) every { PendingActionSyncWorker.schedule(any()) } just Runs - every { vaultEventSocket.events(any()) } returns emptyFlow() + every { vaultEventSocket.events(any()) } returns emptyFlow() + every { vaultEventSocket.events(any>()) } returns emptyFlow() + every { vaultEventSocket.connectionState } returns MutableStateFlow(ConnectionState.DISCONNECTED).asStateFlow() vm = VaultViewModel(apiClient, notificationHelper, pendingActionDao, vaultEventSocket, context) } diff --git a/ios/EthosProtocol/Sources/Services/VaultEventSocket.swift b/ios/EthosProtocol/Sources/Services/VaultEventSocket.swift index 103318e..9e2cbbb 100644 --- a/ios/EthosProtocol/Sources/Services/VaultEventSocket.swift +++ b/ios/EthosProtocol/Sources/Services/VaultEventSocket.swift @@ -7,6 +7,7 @@ protocol WebSocketTasking: AnyObject { func resume() func cancel(with closeCode: URLSessionWebSocketTask.CloseCode, reason: Data?) func receive(completionHandler: @escaping (Result) -> Void) + func send(_ message: URLSessionWebSocketTask.Message, completionHandler: @escaping (Error?) -> Void) } extension URLSessionWebSocketTask: WebSocketTasking {} @@ -74,6 +75,8 @@ final class VaultEventSocket { case vaultReleased(vaultID: String, releasedAt: Date, amount: Int64) /// Server keepalive — no action required; clients may reply with `pong`. case ping + /// Server acknowledgement of a multi-vault subscribe request (#253). + case subscribed(vaultIDs: [String]) /// Server signals a recoverable error (e.g. invalid vault_id on connect). case error(code: String, message: String) /// Unrecognised message type — ignored per api-contract.md §WebSocket @@ -101,6 +104,10 @@ final class VaultEventSocket { private var vaultID: String? private var isStopped = true private var reconnectTask: Task? + private var heartbeatTask: Task? + /// Interval between client-initiated heartbeat pings. 30 s matches most NAT timeout windows. + var heartbeatInterval: TimeInterval = 30 + private var subscribedVaultIDs: [String] = [] init(baseURL: URL, maxReconnectAttempts: Int = 5, @@ -126,12 +133,23 @@ final class VaultEventSocket { openSocket() } + /// Connects and subscribes to events for multiple vaults over a single connection. + /// The primary vault_id in the URL query param is vaultIDs.first; the rest are subscribed + /// via a post-connect "subscribe" message per shared/api-contract.md §WebSocket. + func connect(vaultIDs: [String]) { + guard !vaultIDs.isEmpty else { return } + subscribedVaultIDs = vaultIDs + connect(vaultID: vaultIDs[0]) + } + /// Stops the stream and cancels any pending reconnect. Safe to call /// regardless of current state. func stop() { isStopped = true reconnectTask?.cancel() reconnectTask = nil + heartbeatTask?.cancel() + heartbeatTask = nil task?.cancel(with: .normalClosure, reason: nil) task = nil state = .disconnected @@ -162,6 +180,8 @@ final class VaultEventSocket { // reaching maxReconnectAttempts / .fallbackToPolling. state = .connected listen() + sendSubscribe(vaultIDs: subscribedVaultIDs) + startHeartbeat() } /// Builds `wss:////ws?vault_id=` from an `https://` (or @@ -224,6 +244,9 @@ final class VaultEventSocket { event = .vaultReleased(vaultID: msg.vaultId, releasedAt: msg.releasedAt, amount: msg.amount) case "ping": event = .ping + case "subscribed": + guard let msg = try? decoder.decode(WireSubscribed.self, from: data) else { return } + event = .subscribed(vaultIDs: msg.vaultIds) case "error": guard let msg = try? decoder.decode(WireError.self, from: data) else { return } event = .error(code: msg.code, message: msg.message) @@ -232,6 +255,11 @@ final class VaultEventSocket { event = .unknown } onEvent?(event) + // #252: Reply to server pings with a pong frame so the connection isn't + // dropped by intermediate proxies that treat an un-replied ping as a stale link. + if case .ping = event { + task?.send(.string(#"{"type":"pong"}"#)) { _ in } + } } private func handleFailure() { @@ -254,6 +282,35 @@ final class VaultEventSocket { } } + // MARK: - Heartbeat (#252) + + private func startHeartbeat() { + heartbeatTask?.cancel() + heartbeatTask = Task { @MainActor [weak self] in + guard let self else { return } + while !Task.isCancelled && !isStopped { + try? await self.backoff.sleep(heartbeatInterval) + guard !Task.isCancelled && !isStopped else { return } + self.task?.send(.string(#"{"type":"ping"}"#)) { [weak self] error in + if error != nil { + Task { @MainActor in self?.handleFailure() } + } + } + } + } + } + + // MARK: - Multi-vault subscribe (#253) + + /// Sends a multi-vault subscribe message post-connect. + private func sendSubscribe(vaultIDs: [String]) { + guard !vaultIDs.isEmpty, let task else { return } + // Encode manually to avoid bringing in Codable for a simple payload. + let idsJSON = vaultIDs.map { "\"\($0)\"" }.joined(separator: ",") + let msg = "{\"type\":\"subscribe\",\"vault_ids\":[\(idsJSON)]}" + task.send(.string(msg)) { _ in } + } + // MARK: - Wire message decoders (internal, one per server→client type) /// Top-level discriminator — always present in every server frame. @@ -285,4 +342,8 @@ final class VaultEventSocket { let code: String let message: String } + + private struct WireSubscribed: Decodable { + let vaultIds: [String] + } } diff --git a/ios/EthosProtocol/Sources/ViewModels/Stores.swift b/ios/EthosProtocol/Sources/ViewModels/Stores.swift index 75dbb53..b81f955 100644 --- a/ios/EthosProtocol/Sources/ViewModels/Stores.swift +++ b/ios/EthosProtocol/Sources/ViewModels/Stores.swift @@ -248,6 +248,8 @@ final class VaultStore: ObservableObject { /// Mirrors PendingCheckInStore's count for while the app is foregrounded — drives the /// in-app "N check-ins queued" banner alongside NotificationService's queued indicator. @Published private(set) var queuedCheckInCount = 0 + /// Current WebSocket connection state for the real-time event stream (#255). + @Published private(set) var socketConnectionState: VaultEventSocket.ConnectionState = .disconnected private var eventSocket: VaultEventSocket? @@ -399,6 +401,11 @@ final class VaultStore: ObservableObject { /// without waiting for the next poll. func subscribeToEvents(vaultID: String, socket: VaultEventSocket) { eventSocket = socket + socket.onStateChange = { [weak self] state in + self?.socketConnectionState = state + } + // Sync the current state immediately in case socket was already connected + socketConnectionState = socket.state socket.onEvent = { [weak self] event in guard let self else { return } switch event { @@ -411,6 +418,8 @@ final class VaultStore: ObservableObject { case .ping: // Server keepalive — no state change. break + case .subscribed: + break // acknowledgement only — no state change case .error(let code, let message): self.error = ErrorPresentation(message: "Vault event stream error (\(code)): \(message)") case .unknown: @@ -423,6 +432,7 @@ final class VaultStore: ObservableObject { func unsubscribeFromEvents() { eventSocket?.stop() eventSocket = nil + socketConnectionState = .disconnected } private func scheduleReminders() { diff --git a/ios/EthosProtocol/Sources/Views/Views.swift b/ios/EthosProtocol/Sources/Views/Views.swift index 43245c2..ec413de 100644 --- a/ios/EthosProtocol/Sources/Views/Views.swift +++ b/ios/EthosProtocol/Sources/Views/Views.swift @@ -508,6 +508,50 @@ struct StatusBadge: View { } } +// MARK: - Connection Status Badge (#255) + +/// Small "Live" / "Reconnecting" / "Polling" indicator driven by the WebSocket state. +struct ConnectionStatusBadge: View { + let state: VaultEventSocket.ConnectionState + + var body: some View { + Label(label, systemImage: icon) + .font(.caption2.bold()) + .foregroundStyle(color) + .padding(.horizontal, 6) + .padding(.vertical, 2) + .background(color.opacity(0.1)) + .clipShape(Capsule()) + } + + private var label: String { + switch state { + case .connected: return "Live" + case .connecting: return "Connecting…" + case .disconnected: return "Reconnecting…" + case .fallbackToPolling: return "Polling" + } + } + + private var icon: String { + switch state { + case .connected: return "dot.radiowaves.left.and.right" + case .connecting: return "dot.radiowaves.left.and.right" + case .disconnected: return "arrow.clockwise" + case .fallbackToPolling: return "arrow.clockwise" + } + } + + private var color: Color { + switch state { + case .connected: return .green + case .connecting: return .blue + case .disconnected: return .orange + case .fallbackToPolling: return .gray + } + } +} + // MARK: - Vault Detail struct VaultDetailView: View { @@ -541,6 +585,9 @@ struct VaultDetailView: View { if let ttl = ttlRemaining { LabeledContent("TTL Remaining", value: formatDuration(ttl)) } + LabeledContent("Connection") { + ConnectionStatusBadge(state: vaultStore.socketConnectionState) + } } Section("Two-Factor Authentication") { diff --git a/ios/EthosProtocol/Tests/VaultEventSocketTests.swift b/ios/EthosProtocol/Tests/VaultEventSocketTests.swift index 4548200..281e55c 100644 --- a/ios/EthosProtocol/Tests/VaultEventSocketTests.swift +++ b/ios/EthosProtocol/Tests/VaultEventSocketTests.swift @@ -11,6 +11,8 @@ final class MockWebSocketTask: WebSocketTasking { private(set) var resumeCallCount = 0 private(set) var cancelCallCount = 0 private var receiveHandler: ((Result) -> Void)? + private(set) var sentMessages: [URLSessionWebSocketTask.Message] = [] + private(set) var sendCallCount = 0 func resume() { resumeCallCount += 1 } @@ -22,6 +24,12 @@ final class MockWebSocketTask: WebSocketTasking { receiveHandler = completionHandler } + func send(_ message: URLSessionWebSocketTask.Message, completionHandler: @escaping (Error?) -> Void) { + sentMessages.append(message) + sendCallCount += 1 + completionHandler(nil) + } + func simulateFailure() { receiveHandler?(.failure(URLError(.networkConnectionLost))) } @@ -424,6 +432,49 @@ final class VaultEventSocketTests: XCTestCase { XCTAssertTrue(received, "unrecognized type should still fire onEvent with .unknown") XCTAssertEqual(receivedEvent, .unknown) } + + // MARK: - #252 Ping-Pong Tests + + func test_pingMessage_sendsPongReply() async { + let mockTask = MockWebSocketTask() + let socket = VaultEventSocket(baseURL: URL(string: "https://api.example.com/v1")!, makeTask: { _ in mockTask }) + socket.connect(vaultID: "vault-1") + + mockTask.simulateMessage(.string(#"{"type": "ping"}"#)) + + let ponged = await waitUntil { mockTask.sentMessages.count > 0 } + XCTAssertTrue(ponged, "receiving a ping should trigger a pong send") + if case .string(let text) = mockTask.sentMessages.first { + let data = Data(text.utf8) + let envelope = try? JSONDecoder().decode([String: String].self, from: data) + XCTAssertEqual(envelope?["type"], "pong") + } else { + XCTFail("pong should be a text frame") + } + } + + func test_silentConnectionDeath_detectedViaHeartbeatFailure() async { + var tasks: [MockWebSocketTask] = [] + let randomSource = DeterministicRandomSource([1.0]) + let socket = VaultEventSocket( + baseURL: URL(string: "https://api.example.com/v1")!, + maxReconnectAttempts: 5, + backoff: ReconnectBackoff(baseDelay: 0.001, maxDelay: 0.1, randomSource: randomSource, sleep: { _ in }), + makeTask: { _ in + let task = MockWebSocketTask() + tasks.append(task) + return task + } + ) + socket.heartbeatInterval = 0.001 // near-zero for test speed + socket.connect(vaultID: "vault-1") + + // Simulate silent death by making the next heartbeat send fail + // A heartbeat send failure triggers handleFailure, causing reconnect + // We verify a second task was created (reconnect happened) + let reconnected = await waitUntil(timeout: 2.0) { tasks.count >= 2 } + XCTAssertTrue(reconnected, "a failed heartbeat send should trigger reconnect") + } } // MARK: - #20 VaultStore Real-Time Event Wiring Tests @@ -466,3 +517,138 @@ final class VaultStoreEventWiringTests: XCTestCase { XCTAssertEqual(socket.state, .disconnected) } } + +// MARK: - #253 Multi-vault Subscription Tests + +@MainActor +final class VaultStoreMultiVaultTests: XCTestCase { + + private func makeVault(id: String, balance: Int64) -> Vault { + Vault(id: id, owner: "GABC", beneficiary: "GXYZ", balance: balance, + checkInInterval: 2_592_000, lastCheckIn: Date(), ttlRemaining: 100_000, status: .active) + } + + func test_connectMultipleVaultIDs_sendsSubscribeMessage() async { + let mockTask = MockWebSocketTask() + let socket = VaultEventSocket(baseURL: URL(string: "https://api.example.com/v1")!, makeTask: { _ in mockTask }) + + socket.connect(vaultIDs: ["vault-1", "vault-2", "vault-3"]) + + let subscribed = await waitUntil { mockTask.sentMessages.count > 0 } + XCTAssertTrue(subscribed, "connecting with multiple IDs should send a subscribe message") + if case .string(let text) = mockTask.sentMessages.first { + XCTAssertTrue(text.contains("subscribe"), "sent message should have type=subscribe") + XCTAssertTrue(text.contains("vault-2"), "subscribe message should include vault-2") + XCTAssertTrue(text.contains("vault-3"), "subscribe message should include vault-3") + } else { + XCTFail("subscribe should be a text frame") + } + } + + func test_multiplexedEvents_routedToCorrectVault() async { + let store = VaultStore() + store.vaults = [makeVault(id: "vault-1", balance: 100), makeVault(id: "vault-2", balance: 200)] + + let mockTask = MockWebSocketTask() + let socket = VaultEventSocket(baseURL: URL(string: "https://api.example.com/v1")!, makeTask: { _ in mockTask }) + + store.subscribeToEvents(vaultID: "vault-1", socket: socket) + + // Send update for vault-2 via the same socket + let updatedVault2 = makeVault(id: "vault-2", balance: 999) + socket.onEvent?(.vaultUpdated(updatedVault2)) + + let applied = await waitUntil { store.vaults.first { $0.id == "vault-2" }?.balance == 999 } + XCTAssertTrue(applied, "vault-2 update from multiplexed socket should be applied") + XCTAssertEqual(store.vaults.first { $0.id == "vault-1" }?.balance, 100, "vault-1 should be unchanged") + } +} + +// MARK: - #255 Connection State Publishing Tests + +@MainActor +final class VaultStoreConnectionStateTests: XCTestCase { + + private func makeVault(id: String, balance: Int64) -> Vault { + Vault(id: id, owner: "GABC", beneficiary: "GXYZ", balance: balance, + checkInInterval: 2_592_000, lastCheckIn: Date(), ttlRemaining: 100_000, status: .active) + } + + func test_socketConnectionState_reflectsConnectedState() async { + let store = VaultStore() + store.vaults = [makeVault(id: "vault-1", balance: 100)] + + let mockTask = MockWebSocketTask() + let socket = VaultEventSocket(baseURL: URL(string: "https://api.example.com/v1")!, makeTask: { _ in mockTask }) + + store.subscribeToEvents(vaultID: "vault-1", socket: socket) + + // openSocket() optimistically sets .connected after resume() + XCTAssertEqual(store.socketConnectionState, .connected) + } + + func test_socketConnectionState_reflectsDisconnectedAfterFailure() async { + var tasks: [MockWebSocketTask] = [] + let randomSource = DeterministicRandomSource([1.0]) + let socket = VaultEventSocket( + baseURL: URL(string: "https://api.example.com/v1")!, + maxReconnectAttempts: 5, + backoff: ReconnectBackoff(baseDelay: 1, maxDelay: 30, randomSource: randomSource, sleep: { _ in }), + makeTask: { _ in + let task = MockWebSocketTask() + tasks.append(task) + return task + } + ) + let store = VaultStore() + store.vaults = [makeVault(id: "vault-1", balance: 100)] + store.subscribeToEvents(vaultID: "vault-1", socket: socket) + + tasks[0].simulateFailure() + + let disconnected = await waitUntil { store.socketConnectionState == .disconnected || store.socketConnectionState == .connected } + XCTAssertTrue(disconnected) + // After failure + reconnect, state should be .connected again (new task opened) + let reconnected = await waitUntil { tasks.count >= 2 } + XCTAssertTrue(reconnected) + } + + func test_socketConnectionState_fallbackToPollingAfterMaxAttempts() async { + var tasks: [MockWebSocketTask] = [] + let randomSource = DeterministicRandomSource([1.0, 1.0, 1.0]) + let socket = VaultEventSocket( + baseURL: URL(string: "https://api.example.com/v1")!, + maxReconnectAttempts: 3, + backoff: ReconnectBackoff(baseDelay: 1, maxDelay: 30, randomSource: randomSource, sleep: { _ in }), + makeTask: { _ in + let task = MockWebSocketTask() + tasks.append(task) + return task + } + ) + let store = VaultStore() + store.vaults = [makeVault(id: "vault-1", balance: 100)] + store.subscribeToEvents(vaultID: "vault-1", socket: socket) + + for _ in 0..<3 { + let countBefore = tasks.count + tasks.last?.simulateFailure() + _ = await waitUntil { tasks.count > countBefore || store.socketConnectionState == .fallbackToPolling } + } + + let pollingState = await waitUntil { store.socketConnectionState == .fallbackToPolling } + XCTAssertTrue(pollingState, "store should report .fallbackToPolling after max reconnect attempts") + } + + func test_unsubscribeFromEvents_resetsConnectionState() { + let mockTask = MockWebSocketTask() + let socket = VaultEventSocket(baseURL: URL(string: "https://api.example.com/v1")!, makeTask: { _ in mockTask }) + let store = VaultStore() + + store.subscribeToEvents(vaultID: "vault-1", socket: socket) + XCTAssertEqual(store.socketConnectionState, .connected) + + store.unsubscribeFromEvents() + XCTAssertEqual(store.socketConnectionState, .disconnected) + } +} diff --git a/shared/api-contract.md b/shared/api-contract.md index d438ce6..73baa84 100644 --- a/shared/api-contract.md +++ b/shared/api-contract.md @@ -156,6 +156,14 @@ deposit, withdrawal, beneficiary change, status transition). { "type": "ping" } ``` +**`subscribed`** — server acknowledgement of a client `subscribe` request. +```json +{ + "type": "subscribed", + "vault_ids": ["string", "..."] +} +``` + **`error`** — server signals a recoverable error (e.g. invalid vault_id on connect). ```json { @@ -172,11 +180,39 @@ deposit, withdrawal, beneficiary change, status transition). { "type": "pong" } ``` +**`subscribe`** — sent immediately after connect to subscribe to additional vault IDs on the same connection, supporting N vaults over a single WebSocket rather than N connections. +```json +{ + "type": "subscribe", + "vault_ids": ["string", "..."] +} +``` +The server routes subsequent `vault_updated`/`vault_expired`/`vault_released` events for all listed vault IDs over this connection. Clients connecting to a single vault via the URL query parameter need not send `subscribe`. The server acknowledges with a `subscribed` message. + #### Connection lifecycle - Reconnect with exponential backoff (base 1 s, max 60 s) on any non-4401 close. - On `vault_updated`, merge the embedded `vault` object into the local vault list in-place (do not full-reload from REST). - On `vault_expired` / `vault_released`, trigger a local notification if the app is backgrounded. +- Client SHOULD send a `pong` text frame in response to every `ping` to signal liveness to the server. +- Client SHOULD also send periodic `ping` frames (interval: 30 s) to detect silently-dead TCP connections before the OS closes the socket. If the send fails, the client MUST treat it as a connection drop and reconnect with backoff. +- If the server closes the socket with code 4401, do NOT reconnect — re-authenticate first. + +### Backoff/Jitter Formula (#254) + +Both platforms use full-jitter exponential backoff for WebSocket reconnects: + +``` +base_delay = 1 s +max_delay = 30 s +capped = min(max_delay, base_delay * 2^attempt) +actual = random_uniform(0, capped) +``` + +- `attempt` is 0-based and resets to 0 on the first message successfully received from a new connection. +- Full jitter (not additive) spreads reconnect storms across the full `[0, capped)` window instead of clustering near `capped`. +- iOS: `ReconnectBackoff.delay(forAttempt:)` in `VaultEventSocket.swift`. +- Android: `ReconnectBackoff.delayForAttempt()` in `VaultEventSocket.kt`. ---