fix(concord): only bump session revision on structural change, not per message

Every inbound plane wrap that a session claimed — including a plain chat
message — bumped ConcordSessionManager.revision, and the always-on preload,
the channel subscription, and the open-channel history subscription each call
invalidateFilters() on every bump. So every message re-derived and re-REQ'd
every community's control + channel planes. On a cold load of hundreds of
buffered messages that is hundreds of re-subscriptions, which the relays answer
with "there is a bug in the client, no one should be making so many requests"
and close the plane subs mid-load (each needing a fresh NIP-42 AUTH). The
result: channels load only their last few messages, or none.

ingest() now reports a ConcordIngestOutcome (NOT_MINE / NON_STRUCTURAL /
STRUCTURAL). Only a STRUCTURAL wrap — a Control-Plane fold, a guestbook
membership change, or a buffered base-rekey — bumps the revision. Chat messages
are NON_STRUCTURAL: they still reach the feed via the rumor sink → LocalCache,
but no longer churn the subscriptions. The manager keeps its Boolean contract
(claimed) for DecryptAndIndexProcessor. Verified on-device: the "so many
requests" rate-limit is gone and the plane subscription stays open and drains
steadily instead of being closed and reopened per message.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

amethyst/src/main/java/com/vitorpamplona/amethyst/model/Account.kt

@@ -5022,9 +5022,9 @@ class Account(
 
         // Keep Concord channel metadata (community name/icon, membership) live across the whole
         // app — not just the hub screen — so the Messages tab renders each channel's community
-        // chip, and per-community bans apply, as soon as a Control Plane folds. The revision bumps
-        // on every ingested message, so sample() coalesces bursts into at most one full re-index
-        // per window instead of re-scanning every channel's notes per message.
+        // chip, and per-community bans apply, as soon as a Control Plane folds. The revision now
+        // bumps only on *structural* change (a fold / membership / rekey, never a plain message),
+        // so this fires rarely; sample() stays as a cheap coalescer for a burst of folds.
         scope.launch {
             @OptIn(kotlinx.coroutines.FlowPreview::class)
             concordSessions.revision.sample(500).collect {

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/concord/ConcordCommunitySession.kt

@@ -41,6 +41,29 @@ import kotlinx.coroutines.flow.StateFlow
  */
 typealias ConcordRumorSink = (communityId: HexKey, channelIdHex: HexKey, rumor: Event) -> Unit
 
+/**
+ * The result of feeding one wrap to a session's [ConcordCommunitySession.ingest]. It separates
+ * "was it ours" from "did it change structure", so only structure-changing wraps bump the session
+ * revision (and thus re-derive plane subscriptions). Landing every chat message as a revision bump
+ * re-REQs every plane per message and gets the client rate-limited off the relays.
+ */
+enum class ConcordIngestOutcome {
+    /** The wrap is not addressed to any plane this session knows. Keep routing it elsewhere. */
+    NOT_MINE,
+
+    /** Ours and applied, but nothing the subscription set / folded structure depends on changed —
+     *  a chat/reaction/reply/delete message landing, or a duplicate wrap. Must NOT bump the revision. */
+    NON_STRUCTURAL,
+
+    /** Ours and changed structure: a Control-Plane fold (metadata/channels/membership/authority), a
+     *  guestbook membership change, or a buffered base-rekey. Bumps the revision. */
+    STRUCTURAL,
+    ;
+
+    /** True when the wrap belonged to this session (whether or not it changed structure). */
+    val claimed get() = this != NOT_MINE
+}
+
 /**
  * The live read-model of one joined Concord community, driven by inbound stream
  * wraps fed via [ingest].
@@ -155,38 +178,47 @@ class ConcordCommunitySession(
     /**
      * Ingests a stream [wrap]. If it belongs to this community's Control Plane it
      * re-folds; if it belongs to a known channel plane it re-projects that
-     * channel's messages. Returns true if the wrap was recognized and applied.
+     * channel's messages. The [ConcordIngestOutcome] tells the caller both whether
+     * the wrap was ours and — crucially — whether it changed *structure* (a fold that
+     * moves the subscription set / metadata) versus just landing a chat message. Only
+     * a [ConcordIngestOutcome.STRUCTURAL] result should bump the session revision;
+     * bumping on every message re-derives every plane's REQ per message and rate-limits
+     * the relays (they close the plane subs mid-load, so channels appear empty).
      */
-    fun ingest(wrap: Event): Boolean {
+    fun ingest(wrap: Event): ConcordIngestOutcome {
         when (wrap.pubKey) {
             controlPlaneAddress -> {
                 lock.withLock {
-                    if (controlWraps.put(wrap.id, wrap) != null) return true // dup
+                    if (controlWraps.put(wrap.id, wrap) != null) return ConcordIngestOutcome.NON_STRUCTURAL // dup
                 }
                 refold()
-                return true
+                return ConcordIngestOutcome.STRUCTURAL
             }
             guestbookAddress -> {
                 lock.withLock {
-                    if (guestbookWraps.put(wrap.id, wrap) != null) return true // dup
+                    if (guestbookWraps.put(wrap.id, wrap) != null) return ConcordIngestOutcome.NON_STRUCTURAL // dup
                 }
                 refoldGuestbook()
-                return true
+                return ConcordIngestOutcome.STRUCTURAL
             }
             nextBaseRekeyAddress -> {
                 // Buffer only — decrypting a base-rotation blob needs the account signer, so the
-                // app layer drains [pendingBaseRekeyWraps] with it and authorizes the rotator.
+                // app layer drains [pendingBaseRekeyWraps] with it and authorizes the rotator. That
+                // drain runs off the revision tick, so a buffered rekey must bump (rare — a rekey,
+                // not a message).
                 lock.withLock { baseRekeyWraps[wrap.id] = wrap }
-                return true
+                return ConcordIngestOutcome.STRUCTURAL
             }
             else -> {
-                val channelRef = lock.withLock { channelKeysByAddress[wrap.pubKey] } ?: return false
+                val channelRef = lock.withLock { channelKeysByAddress[wrap.pubKey] } ?: return ConcordIngestOutcome.NOT_MINE
                 val (channelIdHex, _) = channelRef
                 lock.withLock {
                     channelWrapsById.getOrPut(channelIdHex) { LinkedHashMap() }.put(wrap.id, wrap)
                 }
                 reprojectChannel(channelIdHex)
-                return true
+                // A chat message lands in the feed via [onRumor] → LocalCache, independent of the
+                // revision; it changes no plane address, so it must NOT bump (see the storm note above).
+                return ConcordIngestOutcome.NON_STRUCTURAL
             }
         }
     }

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/concord/ConcordSessionManager.kt

@@ -123,11 +123,17 @@ class ConcordSessionManager(
         return out
     }
 
-    /** Route an inbound stream wrap; true if it was a Concord plane wrap we applied. */
+    /**
+     * Route an inbound stream wrap; true if it was a Concord plane wrap we applied. Only a wrap that
+     * changed community *structure* (a fold, a membership/rekey change — not a plain chat message)
+     * bumps the revision: bumping per message re-derives every plane subscription per message and
+     * gets the client rate-limited off the relays (which then close the plane subs mid-load). Chat
+     * messages still reach the feed via the rumor sink → LocalCache, independent of the revision.
+     */
     fun ingest(wrap: Event): Boolean {
-        val applied = registry.ingest(wrap)
-        if (applied) bumpRevision()
-        return applied
+        val outcome = registry.ingest(wrap)
+        if (outcome == ConcordIngestOutcome.STRUCTURAL) bumpRevision()
+        return outcome.claimed
     }
 
     fun sessions() = registry.sessions()

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/concord/ConcordSessionRegistry.kt

@@ -98,16 +98,17 @@ class ConcordSessionRegistry(
         }
 
     /**
-     * Routes an inbound stream [wrap] to whichever session recognizes it. Returns
-     * true if some session applied it. A wrap belongs to at most one plane, so the
-     * first accepting session wins.
+     * Routes an inbound stream [wrap] to whichever session recognizes it, returning that session's
+     * [ConcordIngestOutcome] (or [ConcordIngestOutcome.NOT_MINE] if none claim it). A wrap belongs to
+     * at most one plane, so the first accepting session wins.
      */
-    fun ingest(wrap: Event): Boolean {
+    fun ingest(wrap: Event): ConcordIngestOutcome {
         val snapshot = lock.withLock { sessions.values.toList() }
         for (session in snapshot) {
-            if (session.ingest(wrap)) return true
+            val outcome = session.ingest(wrap)
+            if (outcome != ConcordIngestOutcome.NOT_MINE) return outcome
         }
-        return false
+        return ConcordIngestOutcome.NOT_MINE
     }
 
     fun clear() = lock.withLock { sessions.clear() }

commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/concord/ConcordCommunitySessionTest.kt

@@ -54,8 +54,9 @@ class ConcordCommunitySessionTest {
             val session = ConcordCommunitySession(entry, owner.pubKey) { communityId, channelIdHex, rumor -> captured += Triple(communityId, channelIdHex, rumor) }
             assertEquals(community.controlPlane.publicKeyHex, session.controlPlaneAddress)
 
-            // Feed the genesis control wraps → state folds, channels + membership resolve.
-            community.genesisWraps.forEach { assertTrue(session.ingest(it)) }
+            // Feed the genesis control wraps → state folds, channels + membership resolve. A fold is
+            // STRUCTURAL (it moves the subscription set), so it's allowed to bump the revision.
+            community.genesisWraps.forEach { assertEquals(ConcordIngestOutcome.STRUCTURAL, session.ingest(it)) }
             val state = session.state.value
             assertEquals("Nostrichs", state?.metadata?.name)
             assertTrue(state!!.channels.containsKey(community.generalChannelIdHex))
@@ -67,7 +68,9 @@ class ConcordCommunitySessionTest {
 
             // A channel message wrap decrypts and is emitted to the sink for #general.
             val msgWrap = ConcordActions.buildChannelMessage(owner, general, community.generalChannelIdHex, community.rootEpoch, "gm all", 2L)
-            assertTrue(session.ingest(msgWrap))
+            // A chat message lands in the feed but is NON_STRUCTURAL: it must never bump the revision
+            // (per-message re-subscription is what rate-limited the plane REQs and emptied channels).
+            assertEquals(ConcordIngestOutcome.NON_STRUCTURAL, session.ingest(msgWrap))
             val general9 = captured.filter { it.second == community.generalChannelIdHex && it.third.content == "gm all" }
             assertEquals(1, general9.size)
             assertEquals(community.communityIdHex, general9[0].first)
@@ -76,7 +79,7 @@ class ConcordCommunitySessionTest {
 
             // A reaction to that message decrypts as a kind-7 bound to the channel, e-tagging the target.
             val reactionWrap = ConcordActions.buildChannelReaction(owner, general, community.generalChannelIdHex, community.rootEpoch, message, "🤙", 3L)
-            assertTrue(session.ingest(reactionWrap))
+            assertEquals(ConcordIngestOutcome.NON_STRUCTURAL, session.ingest(reactionWrap))
             val reaction = captured.map { it.third }.first { it.kind == 7 }
             assertEquals("🤙", reaction.content)
             assertEquals(message.id, reaction.tags.first { it[0] == "e" }[1])
@@ -85,7 +88,7 @@ class ConcordCommunitySessionTest {
             // root and lowercase `e` at the immediate parent (both the message here), still bound
             // to the channel so it groups into the message's thread — the shape Armada threads.
             val replyWrap = ConcordActions.buildChannelReply(owner, general, community.generalChannelIdHex, community.rootEpoch, message, "gm back", 4L)
-            assertTrue(session.ingest(replyWrap))
+            assertEquals(ConcordIngestOutcome.NON_STRUCTURAL, session.ingest(replyWrap))
             val reply = captured.map { it.third }.first { it.content == "gm back" }
             assertEquals(1111, reply.kind)
             assertEquals(message.id, reply.tags.first { it[0] == "E" }[1])
@@ -94,6 +97,6 @@ class ConcordCommunitySessionTest {
 
             // A stray wrap from a different community is ignored.
             val outsider = ConcordCommunityFactory.create(owner, "Other", createdAt = 1L, relays = listOf("wss://r.example"))
-            assertTrue(!session.ingest(outsider.genesisWraps.first()))
+            assertEquals(ConcordIngestOutcome.NOT_MINE, session.ingest(outsider.genesisWraps.first()))
         }
 }

commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/concord/ConcordSessionRegistryTest.kt

@@ -30,7 +30,6 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
 import kotlinx.coroutines.test.runTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
-import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
 import kotlin.test.assertTrue
@@ -70,8 +69,8 @@ class ConcordSessionRegistryTest {
             assertTrue(registry.subscribeAddresses().contains(alpha.controlPlane.publicKeyHex))
             assertTrue(registry.subscribeAddresses().contains(beta.controlPlane.publicKeyHex))
 
-            // A genesis control wrap routes to Alpha's session and folds it.
-            alpha.genesisWraps.forEach { assertTrue(registry.ingest(it)) }
+            // A genesis control wrap routes to Alpha's session and folds it (STRUCTURAL).
+            alpha.genesisWraps.forEach { assertEquals(ConcordIngestOutcome.STRUCTURAL, registry.ingest(it)) }
             val alphaState = registry.sessionFor(alpha.communityIdHex)!!.state.value
             assertEquals("Alpha", alphaState?.metadata?.name)
 
@@ -81,7 +80,7 @@ class ConcordSessionRegistryTest {
 
             // A channel message decrypts and is emitted to the sink for Alpha's #general.
             val msg = ConcordActions.buildChannelMessage(owner, general, alpha.generalChannelIdHex, alpha.rootEpoch, "gm", 2L)
-            assertTrue(registry.ingest(msg))
+            assertEquals(ConcordIngestOutcome.NON_STRUCTURAL, registry.ingest(msg))
             val general9 = captured.filter { it.first == alpha.communityIdHex && it.second == alpha.generalChannelIdHex && it.third.content == "gm" }
             assertEquals(1, general9.size)
 
@@ -101,6 +100,6 @@ class ConcordSessionRegistryTest {
 
             // A wrap from an unknown community is routed nowhere.
             val gamma = ConcordCommunityFactory.create(owner, "Gamma", createdAt = 1L, relays = listOf("wss://r.example"))
-            assertFalse(registry.ingest(gamma.genesisWraps.first()))
+            assertEquals(ConcordIngestOutcome.NOT_MINE, registry.ingest(gamma.genesisWraps.first()))
         }
 }