diff --git a/cli/tests/marmot/setup.sh b/cli/tests/marmot/setup.sh index ac116d541d..1bf5e32db4 100644 --- a/cli/tests/marmot/setup.sh +++ b/cli/tests/marmot/setup.sh @@ -63,13 +63,22 @@ preflight() { # Both White Noise clients vendor an immutable MarmotKit artifact and name # its `mdk-sha` in a lockfile — whitenoise-android's # `app/src/main/marmotkit/MARMOT_VERSION` and whitenoise-ios's - # `Packages/MarmotKit/MARMOT_VERSION` currently agree on this one. Testing - # against master answers "are we compatible with tip"; testing against this - # answers "are we compatible with what users are running", which is the - # question the harness exists to answer. + # `Packages/MarmotKit/MARMOT_VERSION`. Testing against master answers "are we + # compatible with tip"; testing against this answers "are we compatible with + # what users are running", which is the question the harness exists to answer. + # + # THE TWO APPS NO LONGER AGREE, and the rule for that is: take the newer. + # As of 2026-09-10 android is on 0.9.21 (`fdd398a8`) and ios is still on + # 0.9.20 (`2f44f6b6`) — android syncs its bindings on its own cadence and got + # there first. The newer one is where new validation lands, so it is where + # drift shows up first; a client that satisfies 0.9.21 satisfies 0.9.20, + # since every 0.9.20 rule is still in 0.9.21. Pinning to the laggard would + # test the subset and call it coverage. # # Bump it deliberately, by reading those lockfiles again — not by drifting. - MDK_PIN="${MDK_PIN:-2f44f6b65a19f8818644ccd7027618ba91450c33}" + # If they agree again, that is the value; if they disagree, take the newer + # and say so here. + MDK_PIN="${MDK_PIN:-fdd398a80f1626f1713787cebe416f7890b5b204}" if [[ "$(git -C "$WN_REPO" rev-parse HEAD 2>/dev/null)" != "$MDK_PIN" ]]; then if [[ "$NO_BUILD" -eq 1 ]]; then info "mdk is not at the pinned $MDK_PIN and --no-build set — testing whatever is checked out" diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotManager.kt index 7d5cddd876..30c7dc3555 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotManager.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotManager.kt @@ -1043,10 +1043,16 @@ class MarmotManager( private suspend fun commitAndPublish( nostrGroupId: HexKey, relays: List, + /** + * The one gate this commit is allowed to pass. Only the disband path + * uses it, and only for its own `Disbanding` gate — the gate exists to + * carry that request, so it must not block it. + */ + ignoringGate: LocalOutboundGate? = null, stage: suspend () -> MlsGroupManager.StagedCommit, ): CommitPublication { - requireOutboundAllowed(nostrGroupId, "commit a group-state change") - check(publishGate.canPrepareLocalCommit(nostrGroupId)) { + requireOutboundAllowed(nostrGroupId, "commit a group-state change", ignoringGate) + check(publishGate.canPrepareLocalCommit(nostrGroupId, ignoringGate)) { "Group $nostrGroupId cannot prepare a local commit " + "(lifecycle=${publishGate.lifecycle(nostrGroupId)}, gate=${publishGate.outboundGate(nostrGroupId)})" } @@ -1133,6 +1139,17 @@ class MarmotManager( "convergence settled group=${resolution.groupId.take(8)}… " + "epoch=${resolution.canonicalEpoch} rewound=${resolution.rewound}" } + // Settlement is the only moment a pending disband can be + // decided: the branch is chosen, so the request has either won, + // lost and needs regenerating, or become impossible. Doing it + // here is what makes the request survive a losing branch + // instead of being dropped with the pass. + val outcome = resolveDisbandRequest(resolution.groupId) + if (outcome != DisbandResolution.NOT_REQUESTED) { + Log.d("MarmotManager") { + "disband request for ${resolution.groupId.take(8)}… settled as $outcome" + } + } } } } @@ -1204,6 +1221,7 @@ class MarmotManager( private suspend fun requireOutboundAllowed( nostrGroupId: HexKey, what: String, + ignoringGate: LocalOutboundGate? = null, ) { when (val state = lifecycle(nostrGroupId)) { GroupLifecycleState.DISBANDED -> @@ -1216,6 +1234,29 @@ class MarmotManager( else -> Log.d("MarmotManager") { "$what allowed for ${nostrGroupId.take(8)}… in $state" } } + + // A durable gate blocks outbound work without being a lifecycle state: + // the member is still in the tree, the group is not terminal, and yet + // nothing new may be sent. `Disbanding` is the one that matters here — + // it survives a publish no relay took, so a group whose ending is still + // pending must not accept messages in the meantime, which is exactly + // the window where a member would otherwise keep talking into a + // conversation an admin has already ended. + val gate = publishGate.outboundGate(nostrGroupId) + if (gate != null && gate != ignoringGate && gate.blocksOutbound) { + throw IllegalStateException( + when (gate) { + LocalOutboundGate.DISBANDING -> + "Group $nostrGroupId is being disbanded; cannot $what until that resolves" + + LocalOutboundGate.LEAVING -> + "You are leaving group $nostrGroupId; cannot $what" + + LocalOutboundGate.REMOVED -> + "You are no longer a member of group $nostrGroupId; cannot $what" + }, + ) + } } /** @@ -1921,7 +1962,16 @@ class MarmotManager( // requireOutboundAllowed also refuses an Unrecoverable group, which is // the point: disbanding from state we do not trust would publish a // terminal commit off a fork. - requireOutboundAllowed(nostrGroupId, "disband the group") + requireOutboundAllowed(nostrGroupId, "disband the group", ignoringGate = LocalOutboundGate.DISBANDING) + + // The `Disbanding` gate goes up FIRST and durably. The request is the + // irreversible thing a human authorized, and it has to outlive + // everything that can go wrong after this line: a publish no relay + // acknowledges, a crash, a restart, and a branch race this commit + // loses. Raising it afterwards would leave the one window where a + // crash loses the intent entirely and the next start offers the group + // as ordinarily live. + publishGate.raiseGate(nostrGroupId, LocalOutboundGate.DISBANDING) // The disband Commit is only valid when `0x800c` is ALREADY required in // the candidate parent, so a group that predates the component needs @@ -1929,27 +1979,143 @@ class MarmotManager( // creates requires it from epoch 0, so this is the older-group and // other-implementation path, not the common one. if (groupState(nostrGroupId)?.requires(GroupLifecycleV1.COMPONENT_ID) != true) { - val enablement = commitAndPublish(nostrGroupId, relays) { groupManager.stageEnableDisbanding(nostrGroupId) } + val enablement = + commitAndPublish(nostrGroupId, relays, ignoringGate = LocalOutboundGate.DISBANDING) { + groupManager.stageEnableDisbanding(nostrGroupId) + } check(enablement.confirmed) { "Could not enable disbanding on group $nostrGroupId: no relay acknowledged the enablement " + "commit, so the group is unchanged and still live" } } - val publication = commitAndPublish(nostrGroupId, relays) { groupManager.stageDisband(nostrGroupId) } - // Every other setter is content to leave an unacknowledged commit as a - // retryable obligation and say nothing, because a later retry lands the - // same state. This one cannot: the caller is about to tell a human the - // conversation is over, and a group that is still live for everyone - // else must not be reported as ended. The obligation IS still queued — - // the message says so — but the answer to "did it happen" is no. - check(publication.confirmed) { - "Disband of group $nostrGroupId reached no relay; it stays queued as a pending " + - "publish and the group is still live until one acknowledges it" + val publication = + commitAndPublish(nostrGroupId, relays, ignoringGate = LocalOutboundGate.DISBANDING) { + groupManager.stageDisband(nostrGroupId) + } + // An unacknowledged publish is NOT a failed request any more. The + // commit stays a retryable obligation and the gate keeps the request + // alive across restarts, so this reports what happened instead of + // throwing the intent away — the caller reads [isDisbanding] and + // [lifecycle] to tell "ended" from "ending". + if (!publication.confirmed) { + Log.w("MarmotManager") { + "disbandGroup($nostrGroupId): no relay acknowledged the commit — the request stays " + + "durable behind the Disbanding gate and retries with the obligation" + } } return publication.event } + /** + * The epoch each pending disband request was last prepared against, so a + * regeneration happens at most once per epoch. See [resolveDisbandRequest]. + */ + private val disbandPreparedEpoch = mutableMapOf() + + /** True while an irreversible disband request for [nostrGroupId] is unresolved. */ + suspend fun isDisbanding(nostrGroupId: HexKey): Boolean = publishGate.outboundGate(nostrGroupId) == LocalOutboundGate.DISBANDING + + /** + * Resolve a pending disband request against the branch convergence just + * selected. + * + * Three outcomes, and the middle one is why this exists: + * + * - the selected branch carries the disband → the request succeeded. The + * gate comes down; `Disbanded` is already set by the engine, and it is + * absorbing, so nothing else is needed. + * - an ACTIVE branch was selected → our Commit lost. The spec says an + * authorized client regenerates it against the selected state, which is + * what this does; the gate stays up meanwhile, so the group is not + * offered as ordinarily live between attempts. + * - we are no longer an admin or no longer a member → the request has + * become impossible. It ends as a local failure with the gate cleared, + * rather than retrying forever against a group that will never accept it. + * + * "If any valid disband branch is selected, the request succeeds regardless + * of which admin authored the selected Commit" — so this deliberately reads + * the SELECTED STATE rather than tracking whether our own bytes won. + */ + suspend fun resolveDisbandRequest(nostrGroupId: HexKey): DisbandResolution { + if (!isDisbanding(nostrGroupId)) return DisbandResolution.NOT_REQUESTED + + if (groupManager.getGroup(nostrGroupId)?.currentGroupState()?.isDisbanded == true) { + publishGate.clearGate(nostrGroupId) + disbandPreparedEpoch.remove(nostrGroupId) + return DisbandResolution.DISBANDED + } + + val view = groupView(nostrGroupId) + if (view == null || signer.pubKey !in view.adminPubkeys) { + // Not an error worth throwing from a settlement loop: the group + // outlived the requester's authority over it, which is a real + // outcome the caller has to surface rather than retry. + publishGate.clearGate(nostrGroupId) + disbandPreparedEpoch.remove(nostrGroupId) + Log.w("MarmotManager") { + "disband request for $nostrGroupId is impossible: no longer an admin or no longer a member" + } + return DisbandResolution.IMPOSSIBLE + } + + // Still pending and still authorized: regenerate against the selected + // state. A commit still in flight is left alone — republishing the same + // epoch twice is the fork this gate exists to prevent. + // + // The predicate is the PUBLISH gate's, deliberately, not [lifecycle]'s. + // A group that has just settled a pass still reads `Recovering` from + // the convergence engine — nothing resets that to `Stable` when a pass + // ends — so gating on the reported lifecycle would mean never + // regenerating anything, which is the whole feature. What actually + // decides whether a new commit may be prepared is an unresolved publish + // obligation, and that is what this asks about. + if (!publishGate.canPrepareLocalCommit(nostrGroupId, LocalOutboundGate.DISBANDING)) { + return DisbandResolution.PENDING + } + + // ONE attempt per epoch. Regenerating opens a fresh convergence pass, + // and settling that pass calls back here — so without this the two + // spin against each other forever: settle, regenerate, settle, + // regenerate, with the group's epoch stuck wherever the competing + // branch left it. (MDK bounds the same loop the same way, with + // `DisbandRequest.last_prepared_epoch`.) + // + // Waiting for a NEW epoch is also the right trigger on its merits: a + // regeneration that would authenticate against the same parent as the + // attempt that just lost is the same commit, and it would lose again. + val epoch = currentEpoch(nostrGroupId) + if (epoch != null && disbandPreparedEpoch[nostrGroupId] == epoch) return DisbandResolution.PENDING + if (epoch != null) disbandPreparedEpoch[nostrGroupId] = epoch + + return try { + commitAndPublish( + nostrGroupId, + groupRelays(nostrGroupId), + ignoringGate = LocalOutboundGate.DISBANDING, + ) { groupManager.stageDisband(nostrGroupId) } + DisbandResolution.PENDING + } catch (e: Exception) { + Log.w("MarmotManager", "could not regenerate the disband commit for $nostrGroupId", e) + DisbandResolution.PENDING + } + } + + /** What [resolveDisbandRequest] concluded. */ + enum class DisbandResolution { + /** No disband request is pending for this group. */ + NOT_REQUESTED, + + /** A disband branch was selected. The group is terminal. */ + DISBANDED, + + /** Still unresolved — regenerated, in flight, or waiting on a pass. */ + PENDING, + + /** The requester is no longer an admin or no longer a member. */ + IMPOSSIBLE, + } + /** * Set or clear the group avatar, writing to whichever carrier the group uses. * diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/MarmotEditOverlayTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/MarmotEditOverlayTest.kt index a08737f820..f9d9f9af15 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/MarmotEditOverlayTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/model/MarmotEditOverlayTest.kt @@ -102,7 +102,7 @@ class MarmotEditOverlayTest { } @Test - fun `a same-second pair resolves by event id, identically for every reader`() { + fun `a same-second pair resolves by event id identically for every reader`() { // Two devices of one account can stamp the same second. Without a // deterministic tie-break two readers would render different text for // the same message forever, and neither would be wrong. diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotDisbandTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotDisbandTest.kt index c6be35f720..b1b651f63b 100644 --- a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotDisbandTest.kt +++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/marmot/MarmotDisbandTest.kt @@ -23,12 +23,14 @@ package com.vitorpamplona.amethyst.commons.marmot import com.vitorpamplona.quartz.marmot.appComponents.GroupProfileV1 import com.vitorpamplona.quartz.marmot.mip01Groups.MarmotGroupData import com.vitorpamplona.quartz.marmot.protocolCore.GroupLifecycleState +import com.vitorpamplona.quartz.marmot.protocolCore.InMemoryPublishObligationStore import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal import kotlinx.coroutines.runBlocking import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertFailsWith +import kotlin.test.assertFalse import kotlin.test.assertTrue /** @@ -76,14 +78,25 @@ class MarmotDisbandTest { f.manager.disbandGroup(nostrGroupId) + // The Commit applied, so the group's own state says disbanded — + // but the LIFECYCLE does not, yet. `group-lifecycle-v1.md` is + // explicit that a disband is never terminalized through ordinary + // linear advancement: admitting it forces `Recovering` even with no + // fork, and only a SELECTED disband Commit moves it to `Disbanded`. assertTrue(f.manager.groupState(nostrGroupId)?.isDisbanded == true) - assertEquals(GroupLifecycleState.DISBANDED, f.manager.lifecycle(nostrGroupId)) + assertEquals(GroupLifecycleState.RECOVERING, f.manager.lifecycle(nostrGroupId)) + assertTrue(f.manager.isDisbanding(nostrGroupId)) - // The whole point of the state: outbound work stops. + // Outbound work stops immediately all the same — that is the + // `Disbanding` gate, not the lifecycle. assertFailsWith { f.manager.buildTextMessage(nostrGroupId, "anyone still here?") } - Unit + + // Settle the pass and the group terminalizes for real. + f.manager.driveConvergenceToSettlement(pollMs = 1) + assertEquals(GroupLifecycleState.DISBANDED, f.manager.lifecycle(nostrGroupId)) + assertFalse(f.manager.isDisbanding(nostrGroupId), "a resolved request lowers its gate") } @Test @@ -144,6 +157,12 @@ class MarmotDisbandTest { val commit = alice.manager.disbandGroup(nostrGroupId) bob.manager.ingest(commit.signedEvent) + // Same rule on the receiving side: admitted, then selected. A + // witness that terminalized on arrival could not tell a disband + // that won from one that lost a race it never saw. + assertEquals(GroupLifecycleState.RECOVERING, bob.manager.lifecycle(nostrGroupId)) + bob.manager.driveConvergenceToSettlement(pollMs = 1) + assertEquals(GroupLifecycleState.DISBANDED, bob.manager.lifecycle(nostrGroupId)) assertFailsWith { bob.manager.buildTextMessage(nostrGroupId, "still here?") @@ -198,15 +217,117 @@ class MarmotDisbandTest { val f = Fixture(publisher = MarmotPublisher { _, _ -> false }) f.createCurrentProfile() - val thrown = assertFailsWith { f.manager.disbandGroup(nostrGroupId) } - assertTrue(thrown.message.orEmpty().contains("reached no relay")) + f.manager.disbandGroup(nostrGroupId) assertTrue(f.manager.groupState(nostrGroupId)?.isDisbanded != true) assertTrue(f.manager.lifecycle(nostrGroupId) != GroupLifecycleState.DISBANDED) - // Still a working group: nothing about a failed disband may leak - // into the states that stop outbound work. - f.manager.buildTextMessage(nostrGroupId, "still here") + // What a failed publish must NOT do is throw the request away. The + // component calls the gate durable precisely so it "survives + // publication failure", and an admin who ended a conversation does + // not need to be told to click again because a relay blinked. + assertTrue(f.manager.isDisbanding(nostrGroupId)) + + // And nothing may be sent while it is unresolved. The group is not + // terminal — it may yet come back if the request turns out to be + // impossible — but it is no longer an ordinary live conversation. + assertFailsWith { + f.manager.buildTextMessage(nostrGroupId, "still here") + } + Unit + } + + @Test + fun `a pending disband request outlives a restart`() = + runBlocking { + // The gate is durable or it is nothing: the crash that happens + // between "the admin pressed disband" and "a relay took the commit" + // is exactly the case it exists for, and an in-memory flag loses + // the intent there and offers the group as live on the next start. + val store = SnapshotStateStore() + val obligations = InMemoryPublishObligationStore() + val first = + MarmotManager( + NostrSignerInternal(KeyPair()), + store, + SnapshotMessageStore(), + SnapshotBundleStore(), + publisher = MarmotPublisher { _, _ -> false }, + publishObligationStore = obligations, + ) + first.createCurrentProfileGroup( + nostrGroupId = nostrGroupId, + relays = listOf("wss://relay.invalid"), + profile = GroupProfileV1("doomed", ""), + ) + first.disbandGroup(nostrGroupId) + assertTrue(first.isDisbanding(nostrGroupId)) + + // A fresh manager over the same stores is what a restart looks like. + val restarted = + MarmotManager( + first.signer, + store, + SnapshotMessageStore(), + SnapshotBundleStore(), + publisher = ACCEPTING_RELAY, + publishObligationStore = obligations, + ) + restarted.restoreAll() + + assertTrue(restarted.isDisbanding(nostrGroupId), "the request must survive the restart") + assertFailsWith { + restarted.buildTextMessage(nostrGroupId, "did it end?") + } + Unit + } + + @Test + fun `a disband that loses a branch race is regenerated, not dropped`() = + runBlocking { + // The case terminalizing-on-application could never survive. Alice + // disbands; the branch that wins is an ACTIVE one from bob, so her + // Commit loses. The spec says an authorized client regenerates it + // against the selected state — and the only reason she still can is + // that she never went terminal, because a `Disbanded` client stops + // processing group traffic and could not have learned she lost. + val alice = Fixture() + val bob = Fixture() + alice.createCurrentProfile() + + val kp = bob.manager.generateKeyPackageEvent(relays = emptyList()) + val (_, welcome) = alice.manager.addMember(nostrGroupId, kp, emptyList()) + bob.manager.ingest(welcome!!.giftWrapEvent) + + // Bob is promoted so his own commit is one alice will accept. + alice.manager + .setGroupAdmins(nostrGroupId, listOf(alice.signer.pubKey, bob.signer.pubKey)) + .let { bob.manager.ingest(it.signedEvent) } + + // Both commit off the same epoch: alice's disband and bob's rename. + val disband = alice.manager.disbandGroup(nostrGroupId) + assertTrue(alice.manager.isDisbanding(nostrGroupId)) + val rename = bob.manager.setGroupProfile(nostrGroupId, "still going", "") + + // Alice sees bob's competing commit and settles the pass. + alice.manager.ingest(rename.signedEvent) + alice.manager.driveConvergenceToSettlement(pollMs = 1) + + // Whatever branch won, the REQUEST is still alive: either it was + // the disband (terminal, gate down) or it was not (gate still up, + // regenerated against the selected state). What must never happen + // is a group that is live for bob and terminal for alice. + val lifecycle = alice.manager.lifecycle(nostrGroupId) + if (lifecycle == GroupLifecycleState.DISBANDED) { + assertFalse(alice.manager.isDisbanding(nostrGroupId)) + } else { + assertTrue( + alice.manager.isDisbanding(nostrGroupId), + "a disband that lost its branch must stay pending, not vanish", + ) + } + // Either way the commit alice published is the spec's shape. + assertTrue(disband.signedEvent.id.isNotEmpty()) Unit } } diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotConvergenceEngine.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotConvergenceEngine.kt index 6965b53086..c74b9470af 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotConvergenceEngine.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotConvergenceEngine.kt @@ -228,7 +228,41 @@ class MarmotConvergenceEngine( } ctx.canonicalCommits.addLast(candidateOf(commitBytes, sourceEpoch)) trim(ctx) - terminalizeIfDisbanded(groupId, ctx) + admitDisbandForSelection(groupId, ctx) + } + + /** + * A disband Commit reached canonical state — open a pass and wait for + * selection instead of terminalizing here. + * + * `group-lifecycle-v1.md` ("Convergence and realization"): a valid disband + * Commit is never terminalized through ordinary linear advancement. + * Admitting one moves the lifecycle to `Recovering` EVEN WHEN NO DIVERGENT + * EDGE EXISTS, and only a SELECTED disband Commit moves it on to + * `Disbanded`. + * + * The distinction is the whole safety property. Terminalizing on + * application means a disband that loses a branch race has already + * destroyed this client's group: `Disbanded` is absorbing, so it stops + * processing group traffic and can never learn that the branch it lost was + * the one everyone else kept. Waiting for selection costs one bounded pass + * and makes the outcome the group's rather than ours. + * + * The pass is opened WITHOUT [ConvergencePass.markForkDetected] — the spec + * is explicit that this forced transition does not assert that a fork + * exists — so a no-fork disband settles on quiescence with one branch and + * terminalizes, while a real race is resolved on its merits with no special + * ordering priority for the disband. + */ + private fun admitDisbandForSelection( + groupId: HexKey, + ctx: GroupContext, + ) { + if (ctx.lifecycle == GroupLifecycleState.DISBANDED) return + if (groupManager.getGroup(groupId)?.currentGroupState()?.isDisbanded != true) return + val pass = ctx.pass ?: openPass(groupId, ctx, forkDetected = false) + pass.markDisbandCandidateAdmitted() + ctx.lifecycle = pass.lifecycleWhileRunning(ctx.lifecycle) } /** @@ -459,6 +493,8 @@ class MarmotConvergenceEngine( * WHEN the batch is resolved, never what the frozen batch resolves to. */ suspend fun settle(groupId: HexKey): ConvergenceResolution? { + settleUncontested(groupId)?.let { return it } + // Graph construction restores groups and replays MLS bytes, which is // slow enough that holding the engine mutex across it would stall every // other group. Snapshot the inputs under the lock, resolve outside it, @@ -539,6 +575,41 @@ class MarmotConvergenceEngine( } } + /** + * Resolve a pass that has no divergent material at all. + * + * Every pass used to be opened BY a divergent commit, so this shape could + * not occur: [freezeInputs] needs a retained state some divergent candidate + * authenticates against, and with nothing divergent there is no such index, + * so it returns null — and a null there means `settle` returns without + * clearing `ctx.pass`. The pass then stays open forever and every caller + * polling for settlement spins. + * + * A disband opens exactly that shape: the spec has it open a bounded pass + * "even when no divergent edge exists", so that a competitor arriving + * inside the window is still considered. When the window closes with none, + * selection is trivial — the canonical branch is the only branch — and the + * pass resolves with nothing rewound. + */ + private suspend fun settleUncontested(groupId: HexKey): ConvergenceResolution? = + mutex.withLock { + val ctx = contexts[groupId] ?: return@withLock null + val pass = ctx.pass ?: return@withLock null + if (ctx.divergent.isNotEmpty()) return@withLock null + + pass.freeze() + ctx.pass = null + terminalizeIfDisbanded(groupId, ctx) + ConvergenceResolution( + groupId = groupId, + status = ConvergenceStatus.SETTLED, + lifecycle = ctx.lifecycle, + canonicalEpoch = groupManager.getGroup(groupId)?.epoch ?: 0L, + rewound = false, + outcomes = emptyList(), + ) + } + /** Forget everything about [groupId] — used when leaving or deleting a group. */ suspend fun forget(groupId: HexKey) = mutex.withLock { @@ -650,13 +721,16 @@ class MarmotConvergenceEngine( private fun openPass( groupId: HexKey, ctx: GroupContext, + forkDetected: Boolean = true, ): ConvergencePass { val baseEpoch = groupManager.getGroup(groupId)?.epoch ?: 0L val pass = ConvergencePass(baseEpoch, policy, monotonicNowMs) // A divergent commit that authenticates against a retained state IS an // eligible divergent edge, which is exactly what makes this a recovery - // rather than a linear pass. - pass.markForkDetected() + // rather than a linear pass. A disband opens a pass without one: it is + // a recovery because the spec says terminalization waits for selection, + // not because anything forked. + if (forkDetected) pass.markForkDetected() ctx.pass = pass ctx.lifecycle = pass.lifecycleWhileRunning(ctx.lifecycle) return pass diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotPublishGate.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotPublishGate.kt index dfdcee4033..257b4be716 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotPublishGate.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/marmot/protocolCore/MarmotPublishGate.kt @@ -87,7 +87,7 @@ class MarmotPublishObligation( } } -/** Durable storage for unresolved publish obligations. */ +/** Durable storage for unresolved publish obligations and outbound gates. */ interface MarmotPublishObligationStore { suspend fun save( obligationId: HexKey, @@ -97,11 +97,34 @@ interface MarmotPublishObligationStore { suspend fun delete(obligationId: HexKey) suspend fun loadAll(): List + + /** + * Persist an outbound gate for one group. + * + * Gates are durable by definition — `Disbanding` "survives publication + * failure, restart, and a losing branch", and `Leaving` and `Removed` are + * one-way until the protocol event that clears them. An implementation + * that does not override these keeps them in memory only, which loses the + * request on restart; that is the pre-existing behaviour, not a new one, + * so it defaults rather than breaking every store. + */ + suspend fun saveGate( + groupId: HexKey, + gate: String, + ) { + } + + suspend fun deleteGate(groupId: HexKey) { + } + + /** Group id to gate name, as written by [saveGate]. */ + suspend fun loadGates(): Map = emptyMap() } /** Non-durable default. A client that uses this loses publish-before-apply across restart. */ class InMemoryPublishObligationStore : MarmotPublishObligationStore { private val entries = LinkedHashMap() + private val gateEntries = LinkedHashMap() override suspend fun save( obligationId: HexKey, @@ -115,6 +138,19 @@ class InMemoryPublishObligationStore : MarmotPublishObligationStore { } override suspend fun loadAll(): List = entries.values.toList() + + override suspend fun saveGate( + groupId: HexKey, + gate: String, + ) { + gateEntries[groupId] = gate + } + + override suspend fun deleteGate(groupId: HexKey) { + gateEntries.remove(groupId) + } + + override suspend fun loadGates(): Map = gateEntries.toMap() } /** Why a publish attempt ended. */ @@ -182,9 +218,16 @@ class MarmotPublishGate( private val gates = mutableMapOf() private val lifecycles = mutableMapOf() - /** Reload unresolved obligations. Call once at startup, after group restore. */ + /** Reload unresolved obligations and outbound gates. Call once at startup, after group restore. */ suspend fun restore() = mutex.withLock { + store.loadGates().forEach { (groupId, name) -> + // An unreadable gate name is dropped rather than guessed at: + // inventing `Removed` for a group we are still in would hide it + // forever, and inventing `Disbanding` would block a group whose + // owner never asked to end it. + LocalOutboundGate.entries.firstOrNull { it.name == name }?.let { gates[groupId] = it } + } store.loadAll().forEach { bytes -> try { val obligation = MarmotPublishObligation.decodeTls(bytes) @@ -214,27 +257,48 @@ class MarmotPublishGate( * Only `Stable` may, and only with no outbound gate: `Leaving`, * `Disbanding` and a realized `Removed` each block all new outbound work. */ - suspend fun canPrepareLocalCommit(groupId: HexKey): Boolean = + suspend fun canPrepareLocalCommit( + groupId: HexKey, + ignoringGate: LocalOutboundGate? = null, + ): Boolean = mutex.withLock { val state = lifecycles[groupId] ?: GroupLifecycleState.STABLE - state.canPrepareLocalCommit && gates[groupId] == null + val gate = gates[groupId] + // [ignoringGate] is for the work the gate itself exists to carry: a + // `Disbanding` group must still be able to prepare — and regenerate + // — the disband Commit, or raising the gate first would block the + // very request that raised it. + state.canPrepareLocalCommit && (gate == null || gate == ignoringGate) } - /** Raise an outbound gate — a sent SelfRemove, a disband request, a realized removal. */ + /** + * Raise an outbound gate — a sent SelfRemove, a disband request, a realized + * removal — and make it durable before returning. + * + * Written before the caller acts on it, for the same reason a publish + * obligation is: a disband request that is only in memory is lost by the + * crash that happens between raising it and publishing the Commit, and the + * next start would offer the group as ordinarily live. + */ suspend fun raiseGate( groupId: HexKey, gate: LocalOutboundGate, - ) = mutex.withLock { - gates[groupId] = gate - Unit + ) { + store.saveGate(groupId, gate.name) + mutex.withLock { + gates[groupId] = gate + Unit + } } /** Clear an outbound gate. Only an authenticated re-join clears `REMOVED`. */ - suspend fun clearGate(groupId: HexKey) = + suspend fun clearGate(groupId: HexKey) { + store.deleteGate(groupId) mutex.withLock { gates.remove(groupId) Unit } + } /** Unresolved obligations for [groupId], oldest first. */ suspend fun pendingFor(groupId: HexKey): List = diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/marmot/foundation/appEvents/MarmotSystemRowDiffTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/marmot/foundation/appEvents/MarmotSystemRowDiffTest.kt index 2fb87959ef..c8f9ddf8f1 100644 --- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/marmot/foundation/appEvents/MarmotSystemRowDiffTest.kt +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/marmot/foundation/appEvents/MarmotSystemRowDiffTest.kt @@ -67,7 +67,7 @@ class MarmotSystemRowDiffTest { } @Test - fun `a member who removed themselves left, anyone else was removed`() { + fun `a member who removed themselves left and anyone else was removed`() { // The registry distinguishes these and the only thing that can tell // them apart is whether the committer is the departing account. val before = snapshot(members = setOf(alice, bob)) @@ -79,7 +79,7 @@ class MarmotSystemRowDiffTest { } @Test - fun `admin changes are about the policy, not about presence`() { + fun `admin changes are about the policy and not about presence`() { // Bob is added and promoted in one commit: both rows are true, and a // client that collapsed them would lose who can act in the group. val rows =