From b22184940b7811a31a7ef7b19cf32eb15d0154f7 Mon Sep 17 00:00:00 2001 From: callebtc <93376500+callebtc@users.noreply.github.com> Date: Mon, 27 Jul 2026 18:00:10 +0200 Subject: [PATCH] Harden background relay lifecycle --- .../android/mesh/UnifiedMeshService.kt | 27 +-- .../android/nostr/NostrBackgroundRuntime.kt | 56 +++--- .../android/nostr/NostrPendingEventQueue.kt | 90 +++++++++ .../android/nostr/NostrRelayManager.kt | 180 +++++++++--------- .../android/nostr/NostrSubscriptionManager.kt | 110 +++++------ .../android/service/MeshForegroundService.kt | 3 +- .../bitchat/android/ui/GeohashViewModel.kt | 8 +- .../nostr/NostrPendingEventQueueTest.kt | 81 ++++++++ .../NostrRelayManagerLifecycleSmokeTest.kt | 28 +++ 9 files changed, 401 insertions(+), 182 deletions(-) create mode 100644 app/src/main/java/com/bitchat/android/nostr/NostrPendingEventQueue.kt create mode 100644 app/src/test/kotlin/com/bitchat/android/nostr/NostrPendingEventQueueTest.kt diff --git a/app/src/main/java/com/bitchat/android/mesh/UnifiedMeshService.kt b/app/src/main/java/com/bitchat/android/mesh/UnifiedMeshService.kt index b9f84547..5a0cc697 100644 --- a/app/src/main/java/com/bitchat/android/mesh/UnifiedMeshService.kt +++ b/app/src/main/java/com/bitchat/android/mesh/UnifiedMeshService.kt @@ -6,7 +6,6 @@ import com.bitchat.android.favorites.FavoriteControlMessage import com.bitchat.android.model.BitchatFilePacket import com.bitchat.android.model.BitchatMessage import com.bitchat.android.noise.NoiseSession -import com.bitchat.android.services.AppStateStore import com.bitchat.android.wifiaware.WifiAwareController import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -14,7 +13,8 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.delay import kotlinx.coroutines.flow.collectLatest -import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.map import kotlinx.coroutines.isActive import kotlinx.coroutines.launch @@ -77,18 +77,21 @@ class UnifiedMeshService( private fun startAnnouncementScheduler() { if (announcementJob?.isActive == true) return announcementJob = serviceScope.launch { - combine(powerManager.profile, AppStateStore.directPeers) { profile, peers -> - profile.meshAnnouncementIntervalMs to peers.isNotEmpty() - }.collectLatest { (intervalMs, hasRecipients) -> - if (!hasRecipients) return@collectLatest - // Connection-specific paths already send an immediate announce. Begin the - // periodic cadence after the configured interval to avoid a transition burst. - while (isActive) { - delay(intervalMs) - if (AppStateStore.directPeers.value.isNotEmpty()) sendBroadcastAnnounce() + powerManager.profile + .map { profile -> + profile.meshAnnouncementIntervalMs to profile.hasDirectPeers + } + .distinctUntilChanged() + .collectLatest { (intervalMs, hasRecipients) -> + if (!hasRecipients) return@collectLatest + // Connection-specific paths already send an immediate announce. Begin the + // periodic cadence after the configured interval to avoid a transition burst. + while (isActive) { + delay(intervalMs) + if (powerManager.profile.value.hasDirectPeers) sendBroadcastAnnounce() + } } } - } } override fun sendMessage(content: String, mentions: List, channel: String?) { diff --git a/app/src/main/java/com/bitchat/android/nostr/NostrBackgroundRuntime.kt b/app/src/main/java/com/bitchat/android/nostr/NostrBackgroundRuntime.kt index 9e3bf785..3931ea72 100644 --- a/app/src/main/java/com/bitchat/android/nostr/NostrBackgroundRuntime.kt +++ b/app/src/main/java/com/bitchat/android/nostr/NostrBackgroundRuntime.kt @@ -61,7 +61,6 @@ object NostrBackgroundRuntime { locationChannels = LocationChannelManager.getInstance(app) subscriptions = NostrSubscriptionManager( app, - scope, owner = NostrRelayManager.OWNER_BACKGROUND ) } @@ -94,7 +93,13 @@ object NostrBackgroundRuntime { } fun ensureConversationDm(geohash: String) { - if (!initialized || geohash == activeGeohash || geohash == conversationGeohash) return + if (!initialized || geohash == conversationGeohash) return + val selectedLiveToken = activeGeohashLiveToken + val selectedChannelSubscriptionIsUsable = + geohash == activeGeohash && + (selectedLiveToken == null || + LiveLocationPrivacyGate.accepts(selectedLiveToken)) + if (selectedChannelSubscriptionIsUsable) return conversationGeohash?.let { subscriptions.unsubscribe("geo-dm-conversation-$it") } conversationGeohash = geohash subscribeGeohashDm( @@ -119,8 +124,9 @@ object NostrBackgroundRuntime { private fun observeSelectedChannel() { scope.launch { locationChannels.selectedChannel.collectLatest { channel -> - val next = (channel as? ChannelID.Location)?.channel?.geohash - val nextToken = (channel as? ChannelID.Location)?.let { + val locationChannel = channel as? ChannelID.Location + val next = locationChannel?.channel?.geohash + val nextToken = locationChannel?.let { locationChannels.liveLocationTokenForSelectedChannel(it.channel) } val previous = activeGeohash @@ -139,9 +145,8 @@ object NostrBackgroundRuntime { conversationGeohash = null } next?.let { geohash -> - val isLiveDerived = (channel as? ChannelID.Location)?.let { - locationChannels.isSelectedChannelLiveDerived(it.channel) - } == true + val isLiveDerived = + locationChannels.isSelectedChannelLiveDerived(locationChannel.channel) if (!isLiveDerived || nextToken != null) { subscribeSelectedGeohash(geohash, nextToken) } @@ -171,21 +176,28 @@ object NostrBackgroundRuntime { liveLocationToken: Long? ) { scope.launch { - if (liveLocationToken != null && - !LiveLocationPrivacyGate.accepts(liveLocationToken) - ) return@launch - val identity = NostrIdentityBridge.deriveIdentity(geohash, application) - subscriptions.subscribeGiftWraps( - pubkey = identity.publicKeyHex, - sinceMs = System.currentTimeMillis() - 172_800_000L, - id = subscriptionId, - handler = { event -> dispatch { it.geohashDm(event, geohash, identity) } }, - liveLocationToken = liveLocationToken - ) - GeohashAliasRegistry.put( - "nostr_${identity.publicKeyHex.take(16)}", - identity.publicKeyHex - ) + val subscribe = { + val identity = NostrIdentityBridge.deriveIdentity(geohash, application) + subscriptions.subscribeGiftWraps( + pubkey = identity.publicKeyHex, + sinceMs = System.currentTimeMillis() - 172_800_000L, + id = subscriptionId, + handler = { event -> + dispatch { it.geohashDm(event, geohash, identity) } + }, + liveLocationToken = liveLocationToken + ) + GeohashAliasRegistry.put( + "nostr_${identity.publicKeyHex.take(16)}", + identity.publicKeyHex + ) + } + + if (liveLocationToken == null) { + subscribe() + } else { + LiveLocationPrivacyGate.runIfAllowed(liveLocationToken, subscribe) + } } } diff --git a/app/src/main/java/com/bitchat/android/nostr/NostrPendingEventQueue.kt b/app/src/main/java/com/bitchat/android/nostr/NostrPendingEventQueue.kt new file mode 100644 index 00000000..dd9e16d1 --- /dev/null +++ b/app/src/main/java/com/bitchat/android/nostr/NostrPendingEventQueue.kt @@ -0,0 +1,90 @@ +package com.bitchat.android.nostr + +/** + * Thread-safe bounded queue of relay deliveries awaiting a usable WebSocket. + * + * Queue entries have a local ID rather than using the Nostr event ID: the same signed event may be + * intentionally published more than once with different relay sets or privacy provenance. + */ +internal class NostrPendingEventQueue( + private val capacity: Int +) { + init { + require(capacity > 0) + } + + data class Delivery( + val queueId: Long, + val event: NostrEvent, + val liveLocationToken: Long? + ) + + private data class Entry( + val queueId: Long, + val event: NostrEvent, + val pendingRelayUrls: MutableSet, + val liveLocationToken: Long? + ) + + private val lock = Any() + private val entries = ArrayDeque() + private var nextQueueId = 1L + + fun enqueue( + event: NostrEvent, + relayUrls: Collection, + liveLocationToken: Long? + ): Long? { + val pendingRelays = relayUrls.filterTo(linkedSetOf()) { it.isNotBlank() } + if (pendingRelays.isEmpty()) return null + + return synchronized(lock) { + if (entries.size >= capacity) entries.removeFirst() + val queueId = nextQueueId++ + entries.addLast( + Entry( + queueId = queueId, + event = event, + pendingRelayUrls = pendingRelays, + liveLocationToken = liveLocationToken + ) + ) + queueId + } + } + + fun pendingForRelay(relayUrl: String): List = synchronized(lock) { + entries + .asSequence() + .filter { relayUrl in it.pendingRelayUrls } + .map { Delivery(it.queueId, it.event, it.liveLocationToken) } + .toList() + } + + fun markDelivered(queueId: Long, relayUrl: String) { + synchronized(lock) { + val iterator = entries.iterator() + while (iterator.hasNext()) { + val entry = iterator.next() + if (entry.queueId != queueId) continue + entry.pendingRelayUrls.remove(relayUrl) + if (entry.pendingRelayUrls.isEmpty()) iterator.remove() + return + } + } + } + + fun removeLiveLocationEvents() { + synchronized(lock) { + entries.removeAll { it.liveLocationToken != null } + } + } + + fun clear() { + synchronized(lock) { + entries.clear() + } + } + + internal fun size(): Int = synchronized(lock) { entries.size } +} diff --git a/app/src/main/java/com/bitchat/android/nostr/NostrRelayManager.kt b/app/src/main/java/com/bitchat/android/nostr/NostrRelayManager.kt index d361ec49..3386a014 100644 --- a/app/src/main/java/com/bitchat/android/nostr/NostrRelayManager.kt +++ b/app/src/main/java/com/bitchat/android/nostr/NostrRelayManager.kt @@ -2,17 +2,17 @@ package com.bitchat.android.nostr import android.util.Log import com.bitchat.android.geohash.LiveLocationPrivacyGate -import com.google.gson.Gson import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow -import com.google.gson.JsonArray +import kotlinx.coroutines.flow.collectLatest +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.map import com.google.gson.JsonParser import kotlinx.coroutines.* import okhttp3.* import java.util.UUID import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import kotlin.math.min import kotlin.math.pow @@ -114,15 +114,8 @@ class NostrRelayManager private constructor() { // Event deduplication system private val eventDeduplicator = NostrEventDeduplicator.getInstance() - // Message queue for reliability - private data class QueuedEvent( - val event: NostrEvent, - val pendingRelayUrls: MutableSet, - val liveLocationToken: Long? = null - ) - - private val messageQueue = mutableListOf() - private val messageQueueLock = Any() + // Bounded per-relay delivery queue for reconnect reliability. + private val messageQueue = NostrPendingEventQueue(MAX_QUEUED_EVENTS) // Coroutine scope for background operations private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob()) @@ -287,14 +280,24 @@ class NostrRelayManager private constructor() { ) closeTargets.forEach { (relayUrl, relaySubscriptionIds) -> val webSocket = connections[relayUrl] ?: return@forEach - relaySubscriptionIds.forEach { subscriptionId -> + for (subscriptionId in relaySubscriptionIds) { val request = NostrRequest.Close(subscriptionId) val message = gson.toJson(request, NostrRequest::class.java) val closeQueued = runCatching { webSocket.send(message) } .getOrDefault(false) if (!closeQueued) { connections.remove(relayUrl, webSocket) + subscriptions.remove(relayUrl) webSocket.cancel() + updateRelayStatus( + relayUrl, + isConnected = false, + error = IllegalStateException("Failed to close revoked subscription") + ) + if (desiredConnected.get() && relayUrl in nonLiveRelayUrls) { + scope.launch { connectToRelay(relayUrl, liveLocationToken = null) } + } + break } } } @@ -314,9 +317,7 @@ class NostrRelayManager private constructor() { } subscriptions.replaceAll { _, ids -> ids - liveSubscriptionIds } - synchronized(messageQueueLock) { - messageQueue.removeAll { it.liveLocationToken != null } - } + messageQueue.removeLiveLocationEvents() liveGeohashTokens.keys.forEach(geohashToRelays::remove) liveGeohashTokens.clear() @@ -325,6 +326,8 @@ class NostrRelayManager private constructor() { .filterNotTo(mutableSetOf()) { it in nonLiveRelayUrls } liveOnlyRelayUrls.forEach { relayUrl -> connections.remove(relayUrl)?.cancel() + subscriptions.remove(relayUrl) + reconnectJobs.remove(relayUrl)?.cancel() } synchronized(relaysList) { relaysList.removeAll { it.url in liveOnlyRelayUrls } @@ -451,28 +454,24 @@ class NostrRelayManager private constructor() { relayUrls: List? = null, liveLocationToken: Long? = null ) { - val targetRelays = relayUrls ?: relaysList.map { it.url } + val targetRelays = (relayUrls ?: relaysList.map { it.url }) + .filter { it.isNotBlank() } + .distinct() + if (targetRelays.isEmpty()) return val queued = runNetworkAction(liveLocationToken) { - synchronized(messageQueueLock) { - if (messageQueue.size >= MAX_QUEUED_EVENTS) { - messageQueue.removeAt(0) - } - messageQueue.add( - QueuedEvent( - event = event, - pendingRelayUrls = targetRelays.toMutableSet(), - liveLocationToken = liveLocationToken - ) - ) - } + val queueId = messageQueue.enqueue( + event = event, + relayUrls = targetRelays, + liveLocationToken = liveLocationToken + ) ?: return@runNetworkAction scope.launch { if (!isNetworkActionAllowed(liveLocationToken)) return@launch targetRelays.forEach { relayUrl -> val webSocket = connections[relayUrl] if (webSocket != null) { if (sendToRelay(event, webSocket, relayUrl, liveLocationToken)) { - markQueuedRelayDelivered(event.id, relayUrl) + messageQueue.markDelivered(queueId, relayUrl) } } } @@ -673,9 +672,7 @@ class NostrRelayManager private constructor() { geohashToRelays.clear() // Clear any queued messages waiting to be sent - synchronized(messageQueueLock) { - messageQueue.clear() - } + messageQueue.clear() Log.i(TAG, "Cleared all Nostr subscriptions and routing caches") } catch (e: Exception) { @@ -767,39 +764,52 @@ class NostrRelayManager private constructor() { stopSubscriptionValidation() // Stop any existing validation subscriptionValidationJob = scope.launch { - while (isActive && desiredConnected.get()) { - val validationInterval = powerManager?.profile?.value - ?.nostr?.subscriptionValidationMs - ?: com.bitchat.android.util.AppConstants.Nostr.SUBSCRIPTION_VALIDATION_INTERVAL_MS - delay(validationInterval) - if (!desiredConnected.get()) break - - try { - val report = validateSubscriptionConsistency() - if (!report.isConsistent && report.connectedRelayCount > 0) { - Log.w(TAG, "Nostr subscription inconsistencies detected") - - // Auto-repair: re-establish subscriptions for relays with missing ones - connections.forEach { (relayUrl, webSocket) -> - val currentSubs = subscriptions[relayUrl] ?: emptySet() - val expectedSubs = activeSubscriptions.keys.filter { subId -> - val subInfo = activeSubscriptions[subId] - subInfo?.targetRelayUrls == null || subInfo.targetRelayUrls.contains(relayUrl) - }.toSet() - - val missingSubs = expectedSubs - currentSubs - if (missingSubs.isNotEmpty()) { - Log.i(TAG, "Auto-repairing ${missingSubs.size} missing subscriptions") - restoreSubscriptionsForRelay(relayUrl, webSocket) - } - } - } - } catch (e: Exception) { - Log.e(TAG, "Error during subscription validation: ${e.message}") + val manager = powerManager + if (manager == null) { + runSubscriptionValidationLoop( + com.bitchat.android.util.AppConstants.Nostr + .SUBSCRIPTION_VALIDATION_INTERVAL_MS + ) + return@launch + } + + manager.profile + .map { it.nostr.subscriptionValidationMs } + .distinctUntilChanged() + .collectLatest(::runSubscriptionValidationLoop) + } + } + + private suspend fun runSubscriptionValidationLoop(intervalMs: Long) { + while (currentCoroutineContext().isActive && desiredConnected.get()) { + delay(intervalMs) + if (!desiredConnected.get()) break + validateAndRepairSubscriptions() + } + } + + private fun validateAndRepairSubscriptions() { + try { + val report = validateSubscriptionConsistency() + if (report.isConsistent || report.connectedRelayCount == 0) return + + Log.w(TAG, "Nostr subscription inconsistencies detected") + connections.forEach { (relayUrl, webSocket) -> + val currentSubs = subscriptions[relayUrl] ?: emptySet() + val expectedSubs = activeSubscriptions.keys.filter { subId -> + val subInfo = activeSubscriptions[subId] + subInfo?.targetRelayUrls == null || + subInfo.targetRelayUrls.contains(relayUrl) + }.toSet() + + if ((expectedSubs - currentSubs).isNotEmpty()) { + Log.i(TAG, "Auto-repairing missing subscriptions") + restoreSubscriptionsForRelay(relayUrl, webSocket) } } + } catch (e: Exception) { + Log.e(TAG, "Error during subscription validation: ${e.message}") } - } /** @@ -869,8 +879,9 @@ class NostrRelayManager private constructor() { } if (success) { // Update relay stats - val relay = relaysList.find { it.url == relayUrl } - relay?.messagesSent = (relay?.messagesSent ?: 0) + 1 + relaysList.find { it.url == relayUrl }?.let { relay -> + relay.messagesSent += 1 + } updateRelaysList() true } else { @@ -883,18 +894,6 @@ class NostrRelayManager private constructor() { } } - private fun markQueuedRelayDelivered(eventId: String, relayUrl: String) { - synchronized(messageQueueLock) { - val iterator = messageQueue.iterator() - while (iterator.hasNext()) { - val queued = iterator.next() - if (queued.event.id != eventId) continue - queued.pendingRelayUrls.remove(relayUrl) - if (queued.pendingRelayUrls.isEmpty()) iterator.remove() - } - } - } - private fun handleMessage(message: String, relayUrl: String) { try { val jsonElement = JsonParser.parseString(message) @@ -908,8 +907,9 @@ class NostrRelayManager private constructor() { when (response) { is NostrResponse.Event -> { // Update relay stats - val relay = relaysList.find { it.url == relayUrl } - relay?.messagesReceived = (relay?.messagesReceived ?: 0) + 1 + relaysList.find { it.url == relayUrl }?.let { relay -> + relay.messagesReceived += 1 + } updateRelaysList() // CLIENT-SIDE FILTER ENFORCEMENT: Ensure this event matches the subscription's filter @@ -1142,17 +1142,17 @@ class NostrRelayManager private constructor() { restoreSubscriptionsForRelay(relayUrl, webSocket) // Process only events still pending for this relay, outside the queue lock. - val queuedForRelay = synchronized(messageQueueLock) { - messageQueue - .filter { - relayUrl in it.pendingRelayUrls && - isNetworkActionAllowed(it.liveLocationToken) - } - .map { it.event to it.liveLocationToken } - } - queuedForRelay.forEach { (event, token) -> - if (sendToRelay(event, webSocket, relayUrl, token)) { - markQueuedRelayDelivered(event.id, relayUrl) + val queuedForRelay = messageQueue.pendingForRelay(relayUrl) + .filter { isNetworkActionAllowed(it.liveLocationToken) } + queuedForRelay.forEach { delivery -> + if (sendToRelay( + delivery.event, + webSocket, + relayUrl, + delivery.liveLocationToken + ) + ) { + messageQueue.markDelivered(delivery.queueId, relayUrl) } } } diff --git a/app/src/main/java/com/bitchat/android/nostr/NostrSubscriptionManager.kt b/app/src/main/java/com/bitchat/android/nostr/NostrSubscriptionManager.kt index e378e31e..19ebeb9c 100644 --- a/app/src/main/java/com/bitchat/android/nostr/NostrSubscriptionManager.kt +++ b/app/src/main/java/com/bitchat/android/nostr/NostrSubscriptionManager.kt @@ -3,24 +3,32 @@ package com.bitchat.android.nostr import android.app.Application import android.util.Log import com.bitchat.android.geohash.LiveLocationPrivacyGate -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.launch /** * NostrSubscriptionManager - * - Encapsulates subscription lifecycle with NostrRelayManager + * - Encapsulates ordered subscription lifecycle with NostrRelayManager. + * + * Relay-manager operations are already non-blocking and schedule network I/O on their own scope. + * Keeping this facade synchronous prevents lifecycle cancellation from dropping unsubscribe work + * and guarantees that a channel switch closes the old subscription before opening the new one. */ class NostrSubscriptionManager( private val application: Application, - private val scope: CoroutineScope, private val owner: String = NostrRelayManager.OWNER_LEGACY ) { companion object { private const val TAG = "NostrSubscriptionManager" } private val relayManager get() = NostrRelayManager.getInstance(application) - fun connect() = scope.launch { runCatching { relayManager.connect() }.onFailure { Log.e(TAG, "connect failed: ${it.message}") } } - fun disconnect() = scope.launch { runCatching { relayManager.disconnect() }.onFailure { Log.e(TAG, "disconnect failed: ${it.message}") } } + fun connect() { + runCatching { relayManager.connect() } + .onFailure { Log.e(TAG, "connect failed: ${it.message}") } + } + + fun disconnect() { + runCatching { relayManager.disconnect() } + .onFailure { Log.e(TAG, "disconnect failed: ${it.message}") } + } fun subscribeGiftWraps( pubkey: String, @@ -29,19 +37,15 @@ class NostrSubscriptionManager( handler: (NostrEvent) -> Unit, liveLocationToken: Long? = null ) { - scope.launch { - if (liveLocationToken != null && - !LiveLocationPrivacyGate.accepts(liveLocationToken) - ) return@launch - val filter = NostrFilter.giftWrapsFor(pubkey, sinceMs) - relayManager.subscribe( - filter = filter, - id = id, - handler = handler, - owner = owner, - liveLocationToken = liveLocationToken - ) - } + if (!isAllowed(liveLocationToken)) return + val filter = NostrFilter.giftWrapsFor(pubkey, sinceMs) + relayManager.subscribe( + filter = filter, + id = id, + handler = handler, + owner = owner, + liveLocationToken = liveLocationToken + ) } /** Subscribe to geohash chat messages only (kind 20000) — low-volume, kept alive in background. */ @@ -53,22 +57,18 @@ class NostrSubscriptionManager( handler: (NostrEvent) -> Unit, liveLocationToken: Long? = null ) { - scope.launch { - if (liveLocationToken != null && - !LiveLocationPrivacyGate.accepts(liveLocationToken) - ) return@launch - val filter = NostrFilter.geohashMessages(geohash, sinceMs, limit) - relayManager.subscribeForGeohash( - geohash, - filter, - id, - handler, - includeDefaults = false, - nRelays = 5, - owner = owner, - liveLocationToken = liveLocationToken - ) - } + if (!isAllowed(liveLocationToken)) return + val filter = NostrFilter.geohashMessages(geohash, sinceMs, limit) + relayManager.subscribeForGeohash( + geohash, + filter, + id, + handler, + includeDefaults = false, + nRelays = 5, + owner = owner, + liveLocationToken = liveLocationToken + ) } /** Subscribe to geohash presence heartbeats only (kind 20001) — high-volume, paused in background. */ @@ -80,26 +80,28 @@ class NostrSubscriptionManager( handler: (NostrEvent) -> Unit, liveLocationToken: Long? = null ) { - scope.launch { - if (liveLocationToken != null && - !LiveLocationPrivacyGate.accepts(liveLocationToken) - ) return@launch - val filter = NostrFilter.geohashPresence(geohash, sinceMs, limit) - relayManager.subscribeForGeohash( - geohash, - filter, - id, - handler, - includeDefaults = false, - nRelays = 5, - owner = owner, - liveLocationToken = liveLocationToken - ) - } + if (!isAllowed(liveLocationToken)) return + val filter = NostrFilter.geohashPresence(geohash, sinceMs, limit) + relayManager.subscribeForGeohash( + geohash, + filter, + id, + handler, + includeDefaults = false, + nRelays = 5, + owner = owner, + liveLocationToken = liveLocationToken + ) } - fun unsubscribe(id: String) { scope.launch { runCatching { relayManager.unsubscribe(id) } } } - fun unsubscribeAllOwned() { - scope.launch { runCatching { relayManager.unsubscribeOwner(owner) } } + fun unsubscribe(id: String) { + runCatching { relayManager.unsubscribe(id) } } + + fun unsubscribeAllOwned() { + runCatching { relayManager.unsubscribeOwner(owner) } + } + + private fun isAllowed(liveLocationToken: Long?): Boolean = + liveLocationToken == null || LiveLocationPrivacyGate.accepts(liveLocationToken) } diff --git a/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt b/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt index 1286988f..5443808a 100644 --- a/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt +++ b/app/src/main/java/com/bitchat/android/service/MeshForegroundService.kt @@ -112,7 +112,8 @@ class MeshForegroundService : Service() { private val unifiedMeshService: com.bitchat.android.mesh.MeshService? get() = MeshServiceHolder.unifiedMeshService private val serviceJob = Job() - private val scope = CoroutineScope(Dispatchers.Default + serviceJob) + // Service lifecycle callbacks and notification state are main-thread confined. + private val scope = CoroutineScope(Dispatchers.Main.immediate + serviceJob) private var isInForeground: Boolean = false private var isShuttingDown: Boolean = false private var lastNotifiedPeerCount: Int? = null diff --git a/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt b/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt index dbc312b5..46504649 100644 --- a/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt +++ b/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt @@ -42,10 +42,9 @@ class GeohashViewModel( companion object { private const val TAG = "GeohashViewModel" } private val repo = GeohashRepository(application, state, dataManager) - private val uiSubscriptionOwner = "geohash-ui-${System.identityHashCode(this)}" + private val uiSubscriptionOwner = "geohash-ui-${UUID.randomUUID()}" private val subscriptionManager = NostrSubscriptionManager( application, - viewModelScope, owner = uiSubscriptionOwner ) private val geohashMessageHandler = GeohashMessageHandler( @@ -75,6 +74,7 @@ class GeohashViewModel( private val liveSamplingSubscriptionGeohashes = mutableSetOf() private var requestedLiveSamplingGeohashes: Set = emptySet() private var requestedUserSamplingGeohashes: Set = emptySet() + private var uiSubscriptionsShutdown = false private val liveLocationRevocationListener: () -> Unit = { val revokedLiveGeohashes = liveSamplingSubscriptionGeohashes.toSet() revokedLiveGeohashes.forEach { geohash -> @@ -424,11 +424,13 @@ class GeohashViewModel( } override fun onCleared() { - super.onCleared() shutdownUiSubscriptions() + super.onCleared() } fun shutdownUiSubscriptions() { + if (uiSubscriptionsShutdown) return + uiSubscriptionsShutdown = true subscriptionManager.unsubscribeAllOwned() geoTimer?.cancel() geoTimer = null diff --git a/app/src/test/kotlin/com/bitchat/android/nostr/NostrPendingEventQueueTest.kt b/app/src/test/kotlin/com/bitchat/android/nostr/NostrPendingEventQueueTest.kt new file mode 100644 index 00000000..8954cbe3 --- /dev/null +++ b/app/src/test/kotlin/com/bitchat/android/nostr/NostrPendingEventQueueTest.kt @@ -0,0 +1,81 @@ +package com.bitchat.android.nostr + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotEquals +import org.junit.Assert.assertNull +import org.junit.Test + +class NostrPendingEventQueueTest { + @Test + fun `empty relay set is not queued`() { + val queue = NostrPendingEventQueue(capacity = 2) + + assertNull(queue.enqueue(event("empty"), emptyList(), liveLocationToken = null)) + assertEquals(0, queue.size()) + } + + @Test + fun `capacity evicts the oldest publish`() { + val queue = NostrPendingEventQueue(capacity = 2) + queue.enqueue(event("one"), listOf("relay"), liveLocationToken = null) + queue.enqueue(event("two"), listOf("relay"), liveLocationToken = null) + queue.enqueue(event("three"), listOf("relay"), liveLocationToken = null) + + assertEquals( + listOf("two", "three"), + queue.pendingForRelay("relay").map { it.event.content } + ) + } + + @Test + fun `duplicate event publishes retain independent delivery state`() { + val queue = NostrPendingEventQueue(capacity = 4) + val signedEvent = event("same") + val firstId = requireNotNull( + queue.enqueue(signedEvent, listOf("relay-a", "relay-b"), liveLocationToken = null) + ) + val secondId = requireNotNull( + queue.enqueue(signedEvent, listOf("relay-a"), liveLocationToken = null) + ) + assertNotEquals(firstId, secondId) + + queue.markDelivered(firstId, "relay-a") + + assertEquals( + listOf(secondId), + queue.pendingForRelay("relay-a").map { it.queueId } + ) + assertEquals( + listOf(firstId), + queue.pendingForRelay("relay-b").map { it.queueId } + ) + + queue.markDelivered(firstId, "relay-b") + assertEquals(1, queue.size()) + } + + @Test + fun `privacy purge retains non-live publishes`() { + val queue = NostrPendingEventQueue(capacity = 4) + queue.enqueue(event("manual"), listOf("relay"), liveLocationToken = null) + queue.enqueue(event("live"), listOf("relay"), liveLocationToken = 42L) + + queue.removeLiveLocationEvents() + + assertEquals( + listOf("manual"), + queue.pendingForRelay("relay").map { it.event.content } + ) + } + + private fun event(content: String): NostrEvent { + val privateKey = "0".repeat(63) + "1" + return NostrEvent( + pubkey = NostrCrypto.derivePublicKey(privateKey), + createdAt = 1, + kind = NostrKind.TEXT_NOTE, + tags = emptyList(), + content = content + ).sign(privateKey) + } +} diff --git a/app/src/test/kotlin/com/bitchat/android/nostr/NostrRelayManagerLifecycleSmokeTest.kt b/app/src/test/kotlin/com/bitchat/android/nostr/NostrRelayManagerLifecycleSmokeTest.kt index 7bb4aa30..de53a579 100644 --- a/app/src/test/kotlin/com/bitchat/android/nostr/NostrRelayManagerLifecycleSmokeTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/nostr/NostrRelayManagerLifecycleSmokeTest.kt @@ -9,6 +9,34 @@ import org.robolectric.RobolectricTestRunner @RunWith(RobolectricTestRunner::class) class NostrRelayManagerLifecycleSmokeTest { + @Test + fun `owner teardown is synchronous and preserves other subscription owners`() { + val manager = NostrRelayManager.shared + manager.disconnect() + manager.clearAllSubscriptions() + val filter = NostrFilter(kinds = listOf(NostrKind.TEXT_NOTE)) + + manager.subscribe( + filter = filter, + id = "background-contract", + handler = {}, + targetRelayUrls = emptyList(), + owner = NostrRelayManager.OWNER_BACKGROUND + ) + manager.subscribe( + filter = filter, + id = "ui-contract", + handler = {}, + targetRelayUrls = emptyList(), + owner = "test-ui" + ) + + manager.unsubscribeOwner("test-ui") + + assertEquals(setOf("background-contract"), manager.getActiveSubscriptions().keys) + manager.clearAllSubscriptions() + } + @Test fun `disconnected manager maintains subscription and empty publish invariants locally`() { val manager = NostrRelayManager.shared