diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt index b26899f072..231c327fde 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt @@ -223,6 +223,7 @@ import com.vitorpamplona.quartz.nip01Core.metadata.MetadataEvent import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.PublishResult import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAll +import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllPagesFromPool import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchAllWithHooks import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.fetchFirst import com.vitorpamplona.quartz.nip01Core.relay.client.accessories.publishAndCollectResults @@ -3032,6 +3033,44 @@ class Account( client.fetchAll(filters = byRelay, timeoutMs = 20_000L) } + /** + * COMPLETE-mode Control-Plane sync — Armada's plane-sweep discipline for the one plane that must + * never fold on a truncated edition set. + * + * The Control Plane defines the channel list, the roster and the banlist, so a *partial* fold + * silently drops channels or mis-renders membership. Two ways that happens, both closed here: + * - **Forward-cursor gap:** the live plane subscription advances a `since` cursor, so an edition + * with a `created_at` below the high-water mark that we never actually ingested — an unban + * published while we were offline, a CORD-06 compaction re-wrap under a newly-held epoch — is + * never asked for again and stays invisible. This sweep uses **no `since`**: it re-fetches the + * whole plane every run. + * - **Per-filter cap:** a relay caps a REQ's result (~100/filter on relay.dreamith.to), which can + * crop a busy Control Plane. This **pages past the cap** ([fetchAllPagesFromPool] walks `until` + * cursors until a plane is drained), so the fold sees every edition regardless of the cap. + * + * Current + every held-prior epoch's Control Plane is swept (the anti-rollback floor folds from the + * priors). Wraps ingest through the global cache connector → [concordSessions] like every other + * Concord drain; AUTH is the shared stream-key handler. Merging communities that share a relay into + * one filter is safe here precisely because we page — the cap no longer truncates. The live control + * subscription still carries brand-new editions in real time; this is the periodic completeness pass. + */ + suspend fun syncConcordControlPlanes(entries: List) { + if (entries.isEmpty()) return + val authorsByRelay = HashMap>() + for (entry in entries) { + for (sub in ConcordSubscriptionPlanner.controlPlaneSubs(listOf(entry))) { + for (relay in sub.relays) authorsByRelay.getOrPut(relay) { HashSet() }.add(sub.pubKeyHex) + } + } + if (authorsByRelay.isEmpty()) return + // No `since`, no `limit` → fetchAllPages treats each filter as unbounded and pages until a + // plane is fully drained (empty page), so the whole Control Plane lands regardless of the cap. + val byRelay = authorsByRelay.mapValues { (_, authors) -> listOf(ConcordActions.planeFilterFor(authors.toList())) } + var drained = 0 + client.fetchAllPagesFromPool(filters = byRelay) { _, _ -> drained++ } + Log.d("Concord", "syncConcordControlPlanes: paged ${authorsByRelay.size} relay(s), drained $drained control wrap(s)") + } + // ── NIP-29 relay-group actions ─────────────────────────────────────────── // All group commands are published ONLY to the group's host relay, where // relay29 authorizes them. The relay is the source of truth; the kind-10009 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelFilterAssembler.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelFilterAssembler.kt index e90d18bc91..cd9e4280c4 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelFilterAssembler.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelFilterAssembler.kt @@ -20,7 +20,6 @@ */ package com.vitorpamplona.amethyst.ui.screen.loggedIn.chats.publicChannels.concord.datasource -import com.vitorpamplona.amethyst.commons.actions.ConcordPlaneSub import com.vitorpamplona.amethyst.commons.actions.ConcordSubscriptionPlanner import com.vitorpamplona.amethyst.commons.model.chats.ChatFeedType import com.vitorpamplona.amethyst.commons.relayClient.composeSubscriptionManagers.ComposeSubscriptionManager @@ -77,27 +76,18 @@ class ConcordChannelSubAssembler( val entries = account.concordChannelList.liveCommunities.value if (entries.isEmpty()) return null - // Control planes for every joined community, plus channel planes for the - // ones whose Control Plane has already folded. Deriving the channel planes - // is the only account-dependent step; collapsing planes into per-relay - // kind-1059 filters lives in the shared planner. - val subs = ArrayList() - subs += ConcordSubscriptionPlanner.controlPlaneSubs(entries) - // The Guestbook (membership) + next-epoch base-rekey planes. Their stream keys derive from - // the entry alone, so they AUTH on the initial connection; and since the relay now - // re-authenticates on an `auth-required` CLOSED, naming them here no longer starves the - // control/channel REQ the way it did before that fix. - subs += ConcordSubscriptionPlanner.auxiliaryPlaneSubs(entries) - for (entry in entries) { - val state = + // The Control Plane rides its OWN filter, kept apart from the Guestbook / rekey / channel + // planes, so a chatty Guestbook can't crowd its channel-defining editions out of a relay's + // per-filter result cap (the "Soapbox shows 1 of 12 channels" bug). See + // [ConcordSubscriptionPlanner.controlIsolatedFilters]. + val all = + ConcordSubscriptionPlanner.controlIsolatedFilters(entries, since = since) { entry -> account.concordSessions .sessionFor(entry.id) ?.state - ?.value ?: continue - subs += ConcordSubscriptionPlanner.channelPlaneSubs(entry, state) - } - - return ConcordSubscriptionPlanner.relayBasedFilters(subs, since) + ?.value + } + return all.ifEmpty { null } } override fun id(key: ConcordChannelQueryState) = key.account diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelSubscription.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelSubscription.kt index a6db1b8ade..dad4bf4313 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelSubscription.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/chats/publicChannels/concord/datasource/ConcordChannelSubscription.kt @@ -24,11 +24,14 @@ import androidx.compose.runtime.Composable import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.remember +import androidx.compose.runtime.rememberUpdatedState import androidx.lifecycle.compose.collectAsStateWithLifecycle import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.KeyDataSourceSubscription import com.vitorpamplona.amethyst.commons.relayClient.subscriptions.LifecycleAwareKeyDataSourceSubscription import com.vitorpamplona.amethyst.ui.navigation.bottombars.BottomBarEntry import com.vitorpamplona.amethyst.ui.screen.loggedIn.AccountViewModel +import com.vitorpamplona.quartz.nip01Core.relay.normalizer.RelayUrlNormalizer +import com.vitorpamplona.quartz.utils.TimeUtils /** * Mount on any screen that lists the user's joined Concord Channels (the Messages @@ -100,6 +103,10 @@ fun ConcordChannelPreload(accountViewModel: AccountViewModel) { bootstrapPinnedCommunities(accountViewModel) + // COMPLETE-mode Control-Plane sweep: page every joined community's whole Control Plane (no `since`, + // past the relay's per-filter cap) so the channel/roster/banlist fold is never silently truncated. + ConcordControlPlaneSync(accountViewModel) + // Warm the last-message preview of every channel of every joined community (one drain per relay), // so the Messages inbox shows a preview for channels the user has never opened. ConcordChannelPreviewAccountPreload(accountViewModel) @@ -107,6 +114,74 @@ fun ConcordChannelPreload(accountViewModel: AccountViewModel) { KeyDataSourceSubscription(state, dataSource) } +/** + * Account-wide COMPLETE-mode sync of every joined community's Control Plane — the completeness pass + * that pairs with the always-on live subscription. The live plane subscription + * ([ConcordChannelFilterAssembler]) carries brand-new editions in real time, but it advances a + * `since` cursor and rides the relay's per-filter cap, so it can miss an edition below the high-water + * mark or a cropped initial page — either of which folds a partial channel list / stale roster. + * [com.vitorpamplona.amethyst.model.Account.syncConcordControlPlanes] closes both by re-fetching the + * whole plane with no `since`, paging past the cap. + * + * It fires only on the **events** that can create a gap — never on a wall-clock poll, because a + * persistent live subscription already keeps a connected relay complete: + * 1. **Load / membership / held-epoch change** — keyed on a signature of the joined set + held + * epochs (the moments a new Control-Plane address appears: join/leave, Refounding). NOT the fold + * revision — a revision-keyed sweep would loop (drain → ingest → fold → revision → drain). + * 2. **Reconnect** — a relay hosting a joined community that comes back online may have missed + * editions published while it was down; that gap is exactly what the live sub can't backfill. We + * re-sweep when the connected-relay set *gains* one of our relays, coalescing flaps with a min + * interval. + */ +@Composable +private fun ConcordControlPlaneSync(accountViewModel: AccountViewModel) { + val account = accountViewModel.account + val communities by account.concordChannelList.liveCommunities.collectAsStateWithLifecycle() + // Always-current set for the reconnect collector, whose effect is keyed on relays (which don't + // change when a same-relay community leaves) — so it must not close over a stale `communities`. + val liveCommunities by rememberUpdatedState(communities) + + val sig = + remember(communities) { + communities.joinToString(",") { entry -> + "${entry.id}@${entry.rootEpoch}:" + entry.heldRoots.joinToString("-") { it.epoch.toString() } + } + } + + // (1) Load + membership/epoch change: one complete sweep of the whole set. + LaunchedEffect(sig) { + if (communities.isNotEmpty()) account.syncConcordControlPlanes(communities) + } + + // (2) Reconnect: re-sweep when a relay of ours transitions disconnected → connected. + val ourRelays = + remember(communities) { + communities + .flatMap { it.relays } + .mapNotNullTo(HashSet()) { RelayUrlNormalizer.normalizeOrNull(it) } + } + LaunchedEffect(ourRelays) { + if (ourRelays.isEmpty()) return@LaunchedEffect + val connectedFlow = account.client.connectedRelaysFlow() + // Baseline = already-open relays; the (1) sweep covered those, so only NEW connections sweep. + var open = connectedFlow.value.intersect(ourRelays) + var lastSweep = 0L + connectedFlow.collect { nowConnected -> + val ours = nowConnected.intersect(ourRelays) + val newlyUp = ours - open + open = ours + if (newlyUp.isEmpty()) return@collect + val now = TimeUtils.nowMillis() + if (now - lastSweep < RECONNECT_RESWEEP_MIN_INTERVAL_MS) return@collect + lastSweep = now + account.syncConcordControlPlanes(liveCommunities) + } + } +} + +/** Coalesce relay flaps: at most one reconnect-driven completeness sweep per this window. */ +private const val RECONNECT_RESWEEP_MIN_INTERVAL_MS = 60_000L + /** * Fetch the private kind-13302 list of any Concord community pinned to the bottom bar whose list we * don't already have, from the relays saved on its tab (see diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlanner.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlanner.kt index d9e781eed7..9a210a8c95 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlanner.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlanner.kt @@ -207,6 +207,39 @@ object ConcordSubscriptionPlanner { return authorsByRelay.mapValues { (_, authors) -> listOf(ConcordActions.planeFilterFor(authors)) } } + /** + * The live-subscription filters for every joined community, with the **Control Plane kept in + * its own filter**, apart from the Guestbook / next-epoch rekey / channel planes. + * + * This is the anti-starvation split. The Control Plane is what folds the channel *list* (and + * roster/roles); collapsing it into one `authors=[…every plane…]` filter lets a chatty plane — + * the Guestbook can carry dozens of membership wraps — crowd its editions out of a relay's + * per-filter result cap (measured ~100/filter on relay.dreamith.to), so only a fraction of the + * channels fold (the "Soapbox shows 1 of 12 channels" bug). A dedicated filter gives the Control + * Plane an isolated per-filter budget, so the whole plane folds no matter how busy the Guestbook + * is. Mirrors Armada's plane-sweep design (one filter per plane scope). Both groups ride the same + * per-relay REQ downstream (groupByRelay), so no extra socket/subscription is opened. + * + * [stateOf] supplies each entry's folded state (null → its channels aren't known yet, so only its + * Control/aux planes are subscribed). [since] is the per-relay EOSE cursor, applied to every filter. + */ + fun controlIsolatedFilters( + entries: List, + since: SincePerRelayMap?, + stateOf: (ConcordCommunityListEntry) -> ConcordCommunityState?, + ): List { + val controlSubs = controlPlaneSubs(entries) + + val otherSubs = ArrayList() + otherSubs += auxiliaryPlaneSubs(entries) + for (entry in entries) { + val state = stateOf(entry) ?: continue + otherSubs += channelPlaneSubs(entry, state) + } + + return relayBasedFilters(controlSubs, since).orEmpty() + relayBasedFilters(otherSubs, since).orEmpty() + } + /** * Collapses [subs] into one [RelayBasedFilter] per host relay for a live * subscription: each relay gets a single `{kinds:[1059], authors:[…all plane diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlannerTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlannerTest.kt index 33f4306799..d7e02a724e 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlannerTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/actions/ConcordSubscriptionPlannerTest.kt @@ -194,4 +194,41 @@ class ConcordSubscriptionPlannerTest { // No planes resolve to a relay -> nothing to subscribe. assertNull(ConcordSubscriptionPlanner.relayBasedFilters(emptyList(), null)) } + + @Test + fun controlIsolatedFiltersKeepControlPlaneOffTheGuestbookAndChannelFilter() = + runTest { + // Regression: a Concord community whose Control Plane got collapsed into the same + // `authors=[…]` filter as the Guestbook + channel planes folded only a fraction of its + // channels — a chatty Guestbook crowded the channel-defining editions out of the relay's + // per-filter cap (Soapbox showed 1 of 12 channels). The Control Plane must ride its own filter. + val community = ConcordCommunityFactory.create(owner, "Nostrichs", createdAt = 1L, relays = listOf("wss://r.example")) + val entry = + com.vitorpamplona.quartz.concord.cord02Community.ConcordCommunityListEntry( + id = community.communityIdHex, + owner = community.ownerPubKey, + ownerSalt = community.ownerSalt.toHexKey(), + root = community.communityRoot.toHexKey(), + rootEpoch = community.rootEpoch, + relays = listOf("wss://r.example"), + name = "Nostrichs", + ) + val state = ConcordActions.foldCommunity(community.genesisWraps, community.controlPlane, community.ownerPubKey) + + val controlPk = community.controlPlane.publicKeyHex + val guestbookPk = ConcordActions.guestbookPlane(community.communityRoot, community.communityId, community.rootEpoch).publicKeyHex + val generalPk = ConcordActions.publicChannel(community.communityRoot, community.generalChannelId, community.rootEpoch).publicKeyHex + + val filters = ConcordSubscriptionPlanner.controlIsolatedFilters(listOf(entry), stateOf = { state }, since = null) + + // The filter carrying the Control Plane must NOT also carry the Guestbook or a channel plane. + val controlFilter = filters.map { it.filter }.single { controlPk in it.authors.orEmpty() } + assertTrue(guestbookPk !in controlFilter.authors.orEmpty(), "Guestbook leaked into the Control Plane filter") + assertTrue(generalPk !in controlFilter.authors.orEmpty(), "a channel plane leaked into the Control Plane filter") + + // The Guestbook and the channel plane share the OTHER filter, and it must not carry Control. + val otherFilter = filters.map { it.filter }.single { guestbookPk in it.authors.orEmpty() } + assertTrue(generalPk in otherFilter.authors.orEmpty(), "channel plane missing from the non-control filter") + assertTrue(controlPk !in otherFilter.authors.orEmpty(), "Control Plane leaked into the Guestbook/channel filter") + } }