fix(marmot): terminalize a disband on selection, and bump the MDK pin to 0.9.21

**The disband was terminalizing on application.** `group-lifecycle-v1.md`
("Convergence and realization") says a valid disband Commit is never
terminalized through ordinary linear advancement: admitting one moves the
lifecycle to `Recovering` EVEN WITH NO DIVERGENT EDGE, and only a SELECTED
disband Commit moves it to `Disbanded`. We went straight to `Disbanded` in
`recordApplied`, so a disband that lost a branch race had already destroyed the
group locally — and `Disbanded` is absorbing, so that client stops processing
group traffic and can never learn the branch it lost was the one everyone else
kept.

Most of the machinery for this was already written and never wired:
`ConvergencePass.markDisbandCandidateAdmitted`, `isRecovery`, and
`LocalOutboundGate.DISBANDING` all existed with no callers outside one unit
test.

- `recordApplied` now admits the disband for selection instead of
  terminalizing; `settle` stays the only path to `Disbanded`.
- The pass opens WITHOUT `markForkDetected` — the spec is explicit that the
  forced transition does not assert a fork.
- `settleUncontested` resolves a pass with no divergent material. Every pass
  used to be opened BY a divergent commit, so `freezeInputs` could assume one
  existed; with none it returns null and `settle` bailed WITHOUT clearing the
  pass, leaving it open forever and spinning every caller polling for
  settlement. A no-fork disband is exactly that shape.

**The request is now durable.** The `Disbanding` gate goes up first and is
persisted (gate storage added to the obligation store as default methods, so
existing stores keep compiling), because it has to outlive a publish no relay
acknowledged, a crash, a restart and a losing branch. An unacknowledged publish
no longer throws the intent away, and `requireOutboundAllowed` honours the gate,
so a group with a pending disband refuses new messages instead of carrying on as
if nothing had been asked.

**Regeneration is bounded to one attempt per epoch.** `resolveDisbandRequest`
runs at settlement and regenerates against the selected state when an active
branch won — but regenerating opens a fresh pass, and settling that pass calls
back in, so without the bound the two spin against each other forever. (MDK
bounds the same loop with `DisbandRequest.last_prepared_epoch`.) Waiting for a
new epoch is also right on the merits: a commit authenticating against the same
parent that just lost would lose again.

**MDK pin → 0.9.21 (`fdd398a8`).** The two shipping apps have diverged —
android is on 0.9.21, ios still on 0.9.20 — so the comment claiming they agree
was false. The rule is now written down: take the newer, because that is where
new validation lands and a client satisfying it satisfies the older one.

Also renames three shared-source test functions that contained a comma.
Kotlin/Native rejects those outright, which is why `test-quartz-linux-native`
and `test-quartz-ios` failed while every JVM run passed — the pre-push hook runs
JVM tasks only, with the native ones disabled on this host.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016kCuA6tc4JQzHPCDd39GHq
This commit is contained in:
Claude
2026-09-11 01:02:37 +00:00
parent 94a51efa69
commit 809983219d
7 changed files with 476 additions and 42 deletions
+14 -5
View File
@@ -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"
@@ -1043,10 +1043,16 @@ class MarmotManager(
private suspend fun commitAndPublish(
nostrGroupId: HexKey,
relays: List<NormalizedRelayUrl>,
/**
* 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<HexKey, Long>()
/** 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.
*
@@ -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.
@@ -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<IllegalStateException> {
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<IllegalStateException> {
bob.manager.buildTextMessage(nostrGroupId, "still here?")
@@ -198,15 +217,117 @@ class MarmotDisbandTest {
val f = Fixture(publisher = MarmotPublisher { _, _ -> false })
f.createCurrentProfile()
val thrown = assertFailsWith<IllegalStateException> { 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<IllegalStateException> {
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<IllegalStateException> {
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
}
}
@@ -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
@@ -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<ByteArray>
/**
* 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<HexKey, String> = emptyMap()
}
/** Non-durable default. A client that uses this loses publish-before-apply across restart. */
class InMemoryPublishObligationStore : MarmotPublishObligationStore {
private val entries = LinkedHashMap<HexKey, ByteArray>()
private val gateEntries = LinkedHashMap<HexKey, String>()
override suspend fun save(
obligationId: HexKey,
@@ -115,6 +138,19 @@ class InMemoryPublishObligationStore : MarmotPublishObligationStore {
}
override suspend fun loadAll(): List<ByteArray> = 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<HexKey, String> = gateEntries.toMap()
}
/** Why a publish attempt ended. */
@@ -182,9 +218,16 @@ class MarmotPublishGate(
private val gates = mutableMapOf<HexKey, LocalOutboundGate>()
private val lifecycles = mutableMapOf<HexKey, GroupLifecycleState>()
/** 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<MarmotPublishObligation> =
@@ -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 =