diff --git a/crates/cgka-conformance-simulator/src/client.rs b/crates/cgka-conformance-simulator/src/client.rs index 4dc05662..7421fc10 100644 --- a/crates/cgka-conformance-simulator/src/client.rs +++ b/crates/cgka-conformance-simulator/src/client.rs @@ -862,6 +862,9 @@ impl HarnessClient { ..msg.clone() }), PeeledContent::Welcome { .. } => Err("group peeler returned a welcome".into()), + PeeledContent::RemovalNotice { .. } => { + Err("group peeler returned a removal notice".into()) + } } } } diff --git a/crates/cgka-conformance-simulator/tests/openmls_replay_probe.rs b/crates/cgka-conformance-simulator/tests/openmls_replay_probe.rs index f1772661..53687d5d 100644 --- a/crates/cgka-conformance-simulator/tests/openmls_replay_probe.rs +++ b/crates/cgka-conformance-simulator/tests/openmls_replay_probe.rs @@ -1426,6 +1426,8 @@ fn dummy_group(group_id: GroupId) -> Group { name: "probe".to_owned(), description: String::new(), epoch: EpochId(1), + participation: Default::default(), + membership_intervals: Vec::new(), members: vec![Member { id: MemberId::new(vec![1]), credential: vec![1], diff --git a/crates/cgka-engine/src/audit_helpers.rs b/crates/cgka-engine/src/audit_helpers.rs index dbf4176c..9cfef786 100644 --- a/crates/cgka-engine/src/audit_helpers.rs +++ b/crates/cgka-engine/src/audit_helpers.rs @@ -77,6 +77,9 @@ pub(crate) fn stale_reason_str(reason: &StaleReason) -> &'static str { StaleReason::UnknownGroup => "unknown_group", StaleReason::OwnEcho => "own_echo", StaleReason::PeelFailed => "peel_failed", + StaleReason::Evicted => "evicted", + StaleReason::Quarantined => "quarantined", + StaleReason::PreMembership => "pre_membership", } } diff --git a/crates/cgka-engine/src/distributed_convergence.rs b/crates/cgka-engine/src/distributed_convergence.rs index 207cfdd6..111aeeca 100644 --- a/crates/cgka-engine/src/distributed_convergence.rs +++ b/crates/cgka-engine/src/distributed_convergence.rs @@ -428,6 +428,7 @@ impl Engine { selected_tip, origin_commit_id, origin_commit_actor, + &observations, )?; } self.emit_application_replay_events(group_id, &observations); @@ -500,6 +501,7 @@ impl Engine { selected_tip: EpochId, origin_commit_id: Option, origin_commit_actor: Option, + observations: &[OpenMlsReplayObservation], ) -> Result<(), OpenMlsProjectionError> { if previous_tip != selected_tip { self.events_buf.push_back(GroupEvent::EpochChanged { @@ -593,18 +595,57 @@ impl Engine { ); } } + // SelfRemove senders consumed by the replayed commits, so departures + // attribute to the leaver here exactly like the direct-ingest seam + // (`MemberLeft` vs `MemberRemoved` is no longer path-dependent). + let self_removed: HashSet = observations + .iter() + .flat_map(|observation| match observation { + OpenMlsReplayObservation::CommitStaged { + self_remove_senders, + .. + } => self_remove_senders.as_slice(), + _ => &[], + }) + .map(|identity| MemberId::new(identity.clone())) + .collect(); for member_id in previous_ids.difference(¤t_ids) { if member_id == self.identity.self_id() { self.clear_leave_request_state(group_id) .map_err(|e| OpenMlsProjectionError::Storage(format!("{e:?}")))?; + // Applied removal via the convergence path: the same + // authoritative participation transition as the direct seam + // (spec group-state.md, "Reaching a non-member state"). + let participation = if self_removed.contains(self.identity.self_id()) { + cgka_traits::GroupParticipation::Left + } else { + cgka_traits::GroupParticipation::Evicted + }; + self.set_group_participation(group_id, participation) + .map_err(|e| OpenMlsProjectionError::Storage(format!("{e:?}")))?; } + let (change, actor) = if self_removed.contains(member_id) { + // A leave is attributed to the leaver, not the member that + // sequenced the auto-commit. + ( + GroupStateChange::MemberLeft { + member: member_id.clone(), + }, + Some(member_id.clone()), + ) + } else { + ( + GroupStateChange::MemberRemoved { + member: member_id.clone(), + }, + None, + ) + }; self.push_group_state_change( group_id, selected_tip, - None, - GroupStateChange::MemberRemoved { - member: member_id.clone(), - }, + actor, + change, origin_commit_id.clone(), ); } diff --git a/crates/cgka-engine/src/engine.rs b/crates/cgka-engine/src/engine.rs index 3c6f2108..9fd8f159 100644 --- a/crates/cgka-engine/src/engine.rs +++ b/crates/cgka-engine/src/engine.rs @@ -136,6 +136,12 @@ pub struct Engine { /// Fast runtime gate for groups where the local member must not produce /// further outbound group traffic while a leave request is outstanding. pub(crate) leaving_groups: HashSet, + /// Groups mid-`resolve_group_quarantine`: the explicit recovery pass + /// re-runs withheld input through ordinary ingest, so the quarantine + /// withhold guard stands down for exactly these groups for exactly that + /// pass — the mechanism behind "no silent re-activation" (spec + /// group-state.md, "Quarantine"). + pub(crate) quarantine_resolutions: HashSet, /// Delayed SelfRemove auto-commit attempts keyed by the standalone /// proposal's content-derived message id. These are re-checked against live @@ -339,6 +345,7 @@ impl EngineBuilder { sent_message_ids: BoundedIdSet::with_capacity(DEDUP_CACHE_CAPACITY), leave_requests: HashMap::new(), leaving_groups: HashSet::new(), + quarantine_resolutions: HashSet::new(), scheduled_self_remove_auto_commits: HashMap::new(), pending_convergence_groups: HashSet::new(), convergence_policy: crate::canonicalization::CanonicalizationPolicy::default(), @@ -1001,6 +1008,76 @@ impl Engine { Ok(()) } + /// Transition the local identity's durable participation record for a + /// group and surface the change as `GroupEvent::ParticipationChanged`. + /// Participation lives on the durable Marmot group record + /// (`spec/protocol-core/group-state.md`, "Participation"), so it survives + /// live-OpenMLS-state teardown and rolls back together with the record + /// when fork recovery restores a pre-commit snapshot. Idempotent: writing + /// the already-current state emits nothing. + pub(crate) fn set_group_participation( + &mut self, + group_id: &GroupId, + participation: cgka_traits::GroupParticipation, + ) -> Result<(), EngineError> { + let mut group = match self.storage.get_group(group_id) { + Ok(group) => group, + // No durable record: nothing to transition. Participation exists + // only for groups the engine holds a record of; "no such group" + // stays distinguishable at the accessor. + Err(cgka_traits::StorageError::NotFound) => return Ok(()), + Err(e) => return Err(EngineError::Backend(format!("get_group: {e:?}"))), + }; + if group.participation == participation { + return Ok(()); + } + group.participation = participation; + // Membership intervals ride the same transition (errors.md, + // `PreMembership`): ending membership closes the open interval at the + // epoch the removing commit reached; a restored `Member` (rejoin or + // quarantine resolution) opens a new one. Quarantine holds change no + // interval — they assert nothing about membership. + match participation { + cgka_traits::GroupParticipation::Left | cgka_traits::GroupParticipation::Evicted => { + if let Some(open) = group + .membership_intervals + .iter_mut() + .find(|interval| interval.ended_at.is_none()) + { + // The removing commit advanced the record to `epoch`; the + // last epoch this identity held keys for is the commit's + // SOURCE epoch (`epoch - 1`) — post-removal epochs were + // never readable, so they stay outside the interval. + open.ended_at = Some(EpochId(group.epoch.0.saturating_sub(1))); + } + } + cgka_traits::GroupParticipation::Member => { + if !group + .membership_intervals + .iter() + .any(|interval| interval.ended_at.is_none()) + { + group + .membership_intervals + .push(cgka_traits::MembershipInterval { + joined_at: group.epoch, + ended_at: None, + }); + } + } + cgka_traits::GroupParticipation::Quarantined { .. } => {} + } + self.storage + .put_group(&group) + .map_err(|e| EngineError::Backend(format!("put_group: {e:?}")))?; + self.events_buf + .push_back(cgka_traits::engine::GroupEvent::ParticipationChanged { + group_id: group_id.clone(), + participation, + }); + Ok(()) + } + fn sent_self_remove_leaving_gate_should_restore( &self, group_id: &GroupId, @@ -1522,11 +1599,114 @@ impl CgkaEngine for Engine { &mut self, group_id: &GroupId, ) -> Result, EngineError> { + // A withheld group is excluded from live convergence (spec + // group-state.md, "Quarantine"); only the explicit + // `resolve_group_quarantine` transition may process its stored input. + match self.storage.get_group(group_id).map(|g| g.participation) { + Ok(cgka_traits::GroupParticipation::Quarantined { .. }) => { + return Ok(Vec::new()); + } + Ok(_) | Err(cgka_traits::StorageError::NotFound) => {} + Err(e) => return Err(EngineError::Backend(format!("get_group: {e:?}"))), + } let now_ms = self.convergence_now_ms(); self.converge_and_drain_queued_outbound_intents(group_id, now_ms) .await } + async fn quarantine_group( + &mut self, + group_id: &GroupId, + reason: cgka_traits::QuarantineReason, + ) -> Result<(), EngineError> { + let group = match self.storage.get_group(group_id) { + Ok(group) => group, + Err(cgka_traits::StorageError::NotFound) => { + return Err(EngineError::UnknownGroup(group_id.clone())); + } + Err(e) => return Err(EngineError::Backend(format!("get_group: {e:?}"))), + }; + // Quarantine is a hold, not a verdict: never downgrade an + // authoritative non-member state to "withheld". + if matches!( + group.participation, + cgka_traits::GroupParticipation::Left | cgka_traits::GroupParticipation::Evicted + ) { + return Ok(()); + } + self.set_group_participation( + group_id, + cgka_traits::GroupParticipation::Quarantined { reason }, + ) + } + + async fn resolve_group_quarantine( + &mut self, + group_id: &GroupId, + ) -> Result { + let current = self + .participation(group_id)? + .ok_or_else(|| EngineError::UnknownGroup(group_id.clone()))?; + if !matches!(current, cgka_traits::GroupParticipation::Quarantined { .. }) { + return Ok(current); + } + // One deliberate convergence pass over the withheld/stored input. The + // ordinary apply seams decide the outcome: a stored removal commit + // resolves participation to Left/Evicted through the same code path + // as live traffic. The withhold guard stands down for this group for + // exactly this pass, and the pass runs past the settlement quiescence + // window: resolution is an explicit decision over a closed input set, + // not live traffic waiting for stragglers. + self.quarantine_resolutions.insert(group_id.clone()); + let quiescence_ms = self + .convergence_policy_for_group(group_id) + .map(|policy| policy.settlement_quiescence_ms) + .unwrap_or_default(); + // The pass re-ingests withheld rows, stamping fresh receive times + // *after* this snapshot; a bare `window + 1` margin would leave them + // inside quiescence forever. One minute of slack makes the closed + // input set decisively quiescent without touching wall-clock state. + let now_ms = self + .convergence_now_ms() + .saturating_add(quiescence_ms.saturating_add(60_000)); + let pass = self + .advance_convergence_inputs_until_settled(group_id, now_ms) + .await; + self.quarantine_resolutions.remove(group_id); + let _ = pass?; + let after = self.participation(group_id)?.unwrap_or(current); + if !matches!(after, cgka_traits::GroupParticipation::Quarantined { .. }) { + return Ok(after); + } + // No authoritative outcome from the pass. Restore Member only when the + // live MLS state is active and the canonical roster still contains the + // local identity — otherwise the hold stands. + let provider = crate::provider::EngineOpenMlsProvider::::new( + &self.crypto, + self.storage.mls_storage(), + ); + let mls_gid = openmls::group::GroupId::from_slice(group_id.as_slice()); + let storage = as openmls_traits::OpenMlsProvider>::storage( + &provider, + ); + let live_group = openmls::group::MlsGroup::load(storage, &mls_gid) + .map_err(|e| EngineError::Backend(format!("load live group: {e:?}")))?; + let self_in_roster = self + .storage + .get_group(group_id) + .map_err(|e| EngineError::Backend(format!("get_group: {e:?}")))? + .members + .iter() + .any(|member| &member.id == self.identity.self_id()); + let live_and_member = + live_group.is_some_and(|mls_group| mls_group.is_active()) && self_in_roster; + if live_and_member { + self.set_group_participation(group_id, cgka_traits::GroupParticipation::Member)?; + return Ok(cgka_traits::GroupParticipation::Member); + } + Ok(after) + } + async fn confirm_published( &mut self, pending: PendingStateRef, @@ -1731,6 +1911,28 @@ impl CgkaEngine for Engine { self.do_members(group_id) } + async fn wrap_removal_notice( + &self, + commit: &TransportMessage, + recipient: &MemberId, + ) -> Result, EngineError> { + self.peeler + .wrap_removal_notice(commit, recipient) + .await + .map_err(EngineError::Peeler) + } + + fn participation( + &self, + group_id: &GroupId, + ) -> Result, EngineError> { + match self.storage.get_group(group_id) { + Ok(group) => Ok(Some(group.participation)), + Err(cgka_traits::StorageError::NotFound) => Ok(None), + Err(e) => Err(EngineError::Backend(format!("get_group: {e:?}"))), + } + } + fn epoch(&self, group_id: &GroupId) -> Result { self.epoch_manager .epoch(group_id) diff --git a/crates/cgka-engine/src/group_lifecycle.rs b/crates/cgka-engine/src/group_lifecycle.rs index d382a463..1c62ceee 100644 --- a/crates/cgka-engine/src/group_lifecycle.rs +++ b/crates/cgka-engine/src/group_lifecycle.rs @@ -23,7 +23,7 @@ use cgka_traits::app_components::{AppComponentSet, default_group_components}; use cgka_traits::capabilities::{GroupCapabilities, TransportKind}; use cgka_traits::engine::{CreateGroupRequest, KeyPackage, SendResult, WelcomeMetadata}; use cgka_traits::error::EngineError; -use cgka_traits::group::{Group, Member}; +use cgka_traits::group::{Group, GroupParticipation, Member}; use cgka_traits::message::{MessageRecord, MessageState, StoredMessagePayload}; use cgka_traits::storage::StorageProvider; use cgka_traits::transport::{EncryptedPayload, TransportEnvelope, TransportMessage}; @@ -221,6 +221,11 @@ impl Engine { epoch: EpochId(mls_group.epoch().as_u64()), members: projected_members, required_capabilities: required_caps, + participation: GroupParticipation::Member, + membership_intervals: vec![cgka_traits::MembershipInterval { + joined_at: EpochId(mls_group.epoch().as_u64()), + ended_at: None, + }], }; self.storage.put_group(&group_record)?; // #740: index this group's transport routing id for O(1) inbound @@ -529,6 +534,29 @@ impl Engine { crate::app_components::require_admin(&mls_group, &group_id, &welcome_sender_id)?; // 6. Persist Marmot group record from signed group-context data. + // + // A verified rejoin is the reinstatement path in + // spec/protocol-core/group-state.md ("Participation"): a fresh + // membership grant returns the identity to `Member`. Capture the prior + // participation of any retained record first so the transition is + // surfaced, not silently overwritten. + let prior_record = self.storage.get_group(&group_id).ok(); + let prior_participation = prior_record.as_ref().map(|group| group.participation); + // Retained membership history survives a rejoin: prior intervals stay + // closed and the fresh grant opens a new one at the welcome's epoch + // (errors.md: membership is a set of intervals, not a boundary). + let mut membership_intervals = prior_record + .map(|group| group.membership_intervals) + .unwrap_or_default(); + if !membership_intervals + .iter() + .any(|interval| interval.ended_at.is_none()) + { + membership_intervals.push(cgka_traits::MembershipInterval { + joined_at: EpochId(mls_group.epoch().as_u64()), + ended_at: None, + }); + } let mut group_record = Group { id: group_id.clone(), name: String::new(), @@ -538,6 +566,8 @@ impl Engine { required_capabilities: crate::capability_manager::required_capabilities_from_group( &mls_group, ), + participation: GroupParticipation::Member, + membership_intervals, }; mirror_app_components_into_record(&mls_group, &mut group_record); self.storage.put_group(&group_record)?; @@ -549,6 +579,15 @@ impl Engine { self.transport_group_id_index .insert(transport_group_id, group_id.clone()); } + if let Some(prior) = prior_participation + && prior != GroupParticipation::Member + { + self.events_buf + .push_back(cgka_traits::engine::GroupEvent::ParticipationChanged { + group_id: group_id.clone(), + participation: GroupParticipation::Member, + }); + } // Cache self's capabilities. Other members' capabilities arrive as // we ingest commits that touched their leaves; join-via-welcome diff --git a/crates/cgka-engine/src/message_processor/ingest.rs b/crates/cgka-engine/src/message_processor/ingest.rs index 8ed63c3f..ebc2042e 100644 --- a/crates/cgka-engine/src/message_processor/ingest.rs +++ b/crates/cgka-engine/src/message_processor/ingest.rs @@ -64,6 +64,29 @@ impl Engine { }); } + // Inbox gift wraps carry either a welcome rumor or a removal notice. + // Peel once to discriminate: a removal notice is only a carrier — its + // embedded, transport-validated group message re-enters the ordinary + // inbound pipeline, where dedup, validation, and (if it really is our + // removal commit) the participation transition all apply as if the + // message had arrived on the group stream (spec member-departure.md, + // "Removal notices" — the notice itself has no authority). A peel + // failure falls through to the welcome path so its established error + // mapping stays intact. + if let Ok(peeled) = self.peeler.peel_welcome(msg).await + && let PeeledContent::RemovalNotice { embedded } = peeled.content + { + if !matches!( + embedded.envelope, + cgka_traits::transport::TransportEnvelope::GroupMessage { .. } + ) { + return Ok(IngestOutcome::Stale { + reason: StaleReason::PeelFailed, + }); + } + return Box::pin(self.do_ingest(embedded)).await; + } + // Reuse the existing join_welcome machinery. Map its error shapes // to typed stale reasons where applicable. match self.do_join_welcome(msg.clone()).await { @@ -81,6 +104,31 @@ impl Engine { } } + /// Invariant repair for the inactive-group ingest arms: an inactive MLS + /// group implies a merged removal already ended our membership, so a + /// durable record still claiming `Member` lost its participation + /// transition. Hold it as `Quarantined(IntegrityHold)` — the Left/Evicted + /// reason is only readable from the removal commit, and this arm does not + /// have it (spec group-state.md, "Reaching a non-member state"). Records + /// already non-member (or quarantined) are left untouched. + fn repair_participation_if_member_claims_inactive_group( + &mut self, + group_id: &GroupId, + ) -> Result<(), EngineError> { + let Ok(group) = self.storage.get_group(group_id) else { + return Ok(()); + }; + if group.participation == cgka_traits::GroupParticipation::Member { + self.set_group_participation( + group_id, + cgka_traits::GroupParticipation::Quarantined { + reason: cgka_traits::QuarantineReason::IntegrityHold, + }, + )?; + } + Ok(()) + } + pub(crate) async fn ingest_group_message( &mut self, msg: &TransportMessage, @@ -99,6 +147,31 @@ impl Engine { let provider = EngineOpenMlsProvider::::new(&self.crypto, self.storage.mls_storage()); let mls_gid = openmls::group::GroupId::from_slice(group_id.as_slice()); + // Withheld group (spec group-state.md, "Quarantine"): excluded + // from live inbound processing. The input is retained (Retryable) + // at the group's held epoch so the explicit resolution pass finds + // it inside the convergence window — ordinary inbound MUST NOT + // silently re-activate a quarantined group. + if !self.quarantine_resolutions.contains(&group_id) + && let Ok(group) = self.storage.get_group(&group_id) + && matches!( + group.participation, + cgka_traits::GroupParticipation::Quarantined { .. } + ) + { + // PeelDeferred: the resolution pass replays exactly this state + // through `retry_deferred_peels`, re-entering ordinary ingest + // with the guard stood down. + self.persist_transport_message_for_existing_group( + msg, + &group_id, + group.epoch, + MessageState::PeelDeferred, + )?; + return Ok(IngestOutcome::Stale { + reason: StaleReason::Quarantined, + }); + } let mut mls_group = match MlsGroup::load( as openmls_traits::OpenMlsProvider>::storage( &provider, @@ -107,6 +180,26 @@ impl Engine { ) { Ok(Some(g)) => g, Ok(None) => { + // No live MLS state. If the durable record says we are no + // longer a member, this is post-departure traffic for a + // tombstoned group — `evicted` classification, terminal — + // not an unknown group (spec/foundation/errors.md scopes + // `unknown_group` to "no state at all"). + if matches!( + self.storage.get_group(&group_id).map(|g| g.participation), + Ok(cgka_traits::GroupParticipation::Left + | cgka_traits::GroupParticipation::Evicted) + ) { + self.persist_transport_message_for_existing_group( + msg, + &group_id, + EpochId(0), + MessageState::Failed, + )?; + return Ok(IngestOutcome::Stale { + reason: StaleReason::Evicted, + }); + } self.persist_transport_message_for_existing_group( msg, &group_id, @@ -127,6 +220,14 @@ impl Engine { // group returns to Stable. let current_epoch = EpochId(mls_group.epoch().as_u64()); if !mls_group.is_active() { + // An inactive MLS group means a merged removal already ended + // our membership, so participation must already be non-member. + // A durable record still claiming `Member` here lost the + // transition (legacy record or storage inconsistency): hold it + // as Quarantined(IntegrityHold) rather than fabricating a + // Left/Evicted reason we did not read from an applied commit + // (spec group-state.md, "Reaching a non-member state"). + self.repair_participation_if_member_claims_inactive_group(&group_id)?; self.persist_transport_message( msg, &group_id, @@ -134,7 +235,7 @@ impl Engine { MessageState::Failed, )?; return Ok(IngestOutcome::Stale { - reason: StaleReason::PeelFailed, + reason: StaleReason::Evicted, }); } if !self.epoch_manager.can_ingest(&group_id) { @@ -287,7 +388,9 @@ impl Engine { }; let mls_bytes = match peeled.content { PeeledContent::MlsMessage { bytes } => bytes, - PeeledContent::Welcome { .. } => { + // Neither shape belongs on the group-message peel path: a + // welcome or removal notice arriving here is malformed input. + PeeledContent::Welcome { .. } | PeeledContent::RemovalNotice { .. } => { self.persist_transport_message( msg, &group_id, @@ -430,6 +533,22 @@ impl Engine { &raw_msg_id, "too_distant_in_the_past", )?; + // The source epoch is readable even though the content is + // not. Outside every retained membership interval it is + // PreMembership — terminal by design: this client was not + // a member when it was sent and can never hold the keys + // (errors.md). Empty history fails open to PeelFailed. + if let Ok(group) = self.storage.get_group(&group_id) + && !group.membership_intervals.is_empty() + && !cgka_traits::membership_intervals_contain( + &group.membership_intervals, + msg_epoch, + ) + { + return Ok(IngestOutcome::Stale { + reason: StaleReason::PreMembership, + }); + } return Ok(IngestOutcome::Stale { reason: StaleReason::PeelFailed, }); @@ -553,14 +672,27 @@ impl Engine { } self.update_stored_message_state(&msg.id, MessageState::Failed)?; + // Deliberately NOT PreMembership even when msg_epoch + // predates every membership interval: a commit older than + // our join is already represented by the state we joined + // with (welcome-before-commit), and AlreadyAtEpoch is its + // established classification. PreMembership is reserved + // for CONTENT this client could never read — the + // too-distant-in-the-past arm above. return Ok(IngestOutcome::Stale { reason: StaleReason::AlreadyAtEpoch { current, msg_epoch }, }); } Err(ProcessMessageError::GroupStateError(MlsGroupStateError::UseAfterEviction)) => { + // Path-1 aftermath, not a fresh eviction discovery: this + // guard only fires once a merged removal already made the + // group inactive (spec group-state.md). Participation must + // already be non-member; repair a record that lost the + // transition instead of silently swallowing the signal. + self.repair_participation_if_member_claims_inactive_group(&group_id)?; self.update_stored_message_state(&msg.id, MessageState::Failed)?; return Ok(IngestOutcome::Stale { - reason: StaleReason::PeelFailed, + reason: StaleReason::Evicted, }); } Err(e) => { @@ -822,6 +954,17 @@ impl Engine { .any(|member| member == self.identity.self_id()) { self.clear_leave_request_state(&group_id)?; + // Applied removal: the only authoritative participation + // transition (spec group-state.md, "Reaching a + // non-member state"). The reason comes from the commit + // itself: our own consumed SelfRemove resolves to + // `Left`, a peer's removal to `Evicted`. + let participation = if self_removed.contains(self.identity.self_id()) { + cgka_traits::GroupParticipation::Left + } else { + cgka_traits::GroupParticipation::Evicted + }; + self.set_group_participation(&group_id, participation)?; } else if after_ids.contains(self.identity.self_id()) { if self.load_leave_request_state(&group_id)?.is_some() { // A SelfRemove proposal is valid only in its diff --git a/crates/cgka-engine/src/openmls_projection.rs b/crates/cgka-engine/src/openmls_projection.rs index 93f81b39..d23fbd50 100644 --- a/crates/cgka-engine/src/openmls_projection.rs +++ b/crates/cgka-engine/src/openmls_projection.rs @@ -229,6 +229,12 @@ pub enum OpenMlsReplayObservation { priority: CommitOrderingPriority, committer: Vec, consumed_proposal_refs: Vec, + /// Credential identities of the members whose SelfRemove proposals + /// this commit consumed. Lets the convergence emission path attribute + /// a departure to the leaver (`MemberLeft` / participation `Left`) + /// instead of collapsing every removal into `MemberRemoved`, matching + /// the direct-ingest seam. + self_remove_senders: Vec>, }, ApplicationProcessed { message_id: String, @@ -395,6 +401,7 @@ fn materialize_openmls_candidate_paths_budgeted( priority, committer, consumed_proposal_refs: commit_consumed_proposal_refs, + self_remove_senders: _, } = observation else { continue; @@ -1606,6 +1613,17 @@ fn process_openmls_messages_inner( .map(|proposal| tls_hex(proposal.proposal_reference_ref())) .collect::, _>>()?; consumed_proposal_refs.sort(); + // Resolved against the pre-merge group, like the direct-ingest + // seam: the leaving leaves disappear once the merge lands. + let mut self_remove_senders: Vec> = staged + .queued_proposals() + .filter(|queued| matches!(queued.proposal(), Proposal::SelfRemove)) + .filter_map(|queued| { + crate::identity::member_id_of_sender(queued.sender(), &mls_group) + .map(|id| id.as_slice().to_vec()) + }) + .collect(); + self_remove_senders.sort(); observations.push(OpenMlsReplayObservation::CommitStaged { message_id, source_epoch, @@ -1613,6 +1631,7 @@ fn process_openmls_messages_inner( priority, committer, consumed_proposal_refs, + self_remove_senders, }); mls_group .merge_staged_commit(&provider, *staged) diff --git a/crates/cgka-engine/tests/hydration_quarantine.rs b/crates/cgka-engine/tests/hydration_quarantine.rs index 47c63820..c7b481d5 100644 --- a/crates/cgka-engine/tests/hydration_quarantine.rs +++ b/crates/cgka-engine/tests/hydration_quarantine.rs @@ -176,6 +176,8 @@ fn insert_marmot_group_without_openmls_state( members: Vec::new(), epoch: EpochId(epoch), required_capabilities: GroupCapabilities::default(), + participation: Default::default(), + membership_intervals: Vec::new(), }) .expect("insert marmot group record without openmls state"); } diff --git a/crates/cgka-engine/tests/ingest.rs b/crates/cgka-engine/tests/ingest.rs index 516ff5c8..9c2be3b5 100644 --- a/crates/cgka-engine/tests/ingest.rs +++ b/crates/cgka-engine/tests/ingest.rs @@ -1000,3 +1000,127 @@ async fn distinct_mls_messages_are_not_collapsed_by_content_dedup() { "both distinct messages must be delivered; got {delivered:?}" ); } + +/// Peeler stub whose `peel_welcome` returns a removal notice embedding a +/// group message, exercising the engine's carrier re-injection seam without +/// NIP-59 machinery (the Nostr wire shape is covered in +/// transport-nostr-peeler). +struct RemovalNoticePeeler { + embedded: TransportMessage, +} + +#[async_trait] +impl TransportPeeler for RemovalNoticePeeler { + async fn peel_group_message( + &self, + _msg: &TransportMessage, + _ctx: &GroupContextSnapshot, + ) -> Result { + Err(PeelerError::DecryptFailed) + } + + async fn peel_welcome(&self, msg: &TransportMessage) -> Result { + Ok(PeeledMessage { + id: msg.id.clone(), + group_id: None, + sender: None, + content: PeeledContent::RemovalNotice { + embedded: self.embedded.clone(), + }, + origin: msg.clone(), + }) + } + + async fn wrap_group_message( + &self, + _payload: &EncryptedPayload, + _ctx: &GroupContextSnapshot, + ) -> Result { + Err(PeelerError::WrapFailed("not used".into())) + } + + async fn wrap_welcome( + &self, + _payload: &EncryptedPayload, + _recipient: &MemberId, + ) -> Result { + Err(PeelerError::WrapFailed("not used".into())) + } +} + +#[tokio::test] +async fn removal_notice_reinjects_embedded_group_message_into_ordinary_ingest() { + // The embedded message targets a group this client has no state for, so a + // re-injected classification of UnknownGroup proves the embedded message + // went through the ordinary group-message pipeline (a rejected notice + // would surface PeelFailed instead) while the notice itself granted no + // authority. + let embedded = TransportMessage { + id: MessageId::new(vec![7; 4]), + payload: vec![9, 9, 9], + timestamp: Timestamp(0), + causal_deps: vec![], + source: TransportSource("test".into()), + envelope: TransportEnvelope::GroupMessage { + transport_group_id: vec![0xEE; 32], + }, + }; + let mut engine = build_client_with_peeler( + b"me", + Box::new(RemovalNoticePeeler { + embedded: embedded.clone(), + }), + ); + let notice = TransportMessage { + id: MessageId::new(vec![8; 4]), + payload: vec![], + timestamp: Timestamp(0), + causal_deps: vec![], + source: TransportSource("test".into()), + envelope: TransportEnvelope::Welcome { + recipient: engine.self_id(), + }, + }; + + let outcome = engine.ingest(notice).await.unwrap(); + assert!( + matches!( + outcome, + IngestOutcome::Stale { + reason: StaleReason::UnknownGroup + } + ), + "embedded message should re-enter ordinary ingest; got {outcome:?}" + ); + + // A notice whose embedded message is not a group message is rejected + // before re-injection. + let bogus = TransportMessage { + envelope: TransportEnvelope::Welcome { + recipient: engine.self_id(), + }, + id: MessageId::new(vec![9; 4]), + payload: vec![], + timestamp: Timestamp(0), + causal_deps: vec![], + source: TransportSource("test".into()), + }; + let mut engine = build_client_with_peeler( + b"me", + Box::new(RemovalNoticePeeler { + embedded: TransportMessage { + envelope: TransportEnvelope::Welcome { + recipient: MemberId::new(vec![1; 32]), + }, + ..embedded + }, + }), + ); + let outcome = engine.ingest(bogus).await.unwrap(); + assert!(matches!( + outcome, + IngestOutcome::Stale { + reason: StaleReason::PeelFailed + } + )); +} diff --git a/crates/cgka-engine/tests/invite_leave.rs b/crates/cgka-engine/tests/invite_leave.rs index 9dbf7b95..ec77f555 100644 --- a/crates/cgka-engine/tests/invite_leave.rs +++ b/crates/cgka-engine/tests/invite_leave.rs @@ -42,10 +42,12 @@ async fn advance_selfremove_auto_commit(engine: &mut E, group_id: } /// True if `events` contains a `GroupStateChanged` departure (removed or left) -/// for `member`. Accepts either variant because the leave/removed distinction -/// is path-dependent: the direct inbound seam classifies a SelfRemove as -/// `MemberLeft`, while a convergence reorg surfaces it as an unattributed -/// `MemberRemoved`. +/// for `member`. Loose matcher for tests that only care that a departure was +/// observed. The leave/removed distinction itself is no longer path-dependent: +/// both the direct inbound seam and the convergence replay path resolve a +/// consumed SelfRemove to `MemberLeft` (attributed to the leaver) and +/// everything else to `MemberRemoved`; participation assertions pin the strict +/// classification where it matters. fn emits_departure_of(events: &[cgka_traits::engine::GroupEvent], member: &MemberId) -> bool { events.iter().any(|event| { matches!( @@ -759,6 +761,24 @@ async fn readd_after_remove_produces_fresh_welcome_join() { emits_departure_of(&bob_remove_events, &bob.self_id()), "bob should observe his own removal; got {bob_remove_events:?}" ); + // Applied removal is the authoritative participation transition (spec + // group-state.md, "Reaching a non-member state"): a peer's removal + // resolves to `Evicted`, surfaced as `ParticipationChanged`. + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Evicted), + "bob's participation should be Evicted after applying the removal" + ); + assert!( + bob_remove_events.iter().any(|event| matches!( + event, + cgka_traits::engine::GroupEvent::ParticipationChanged { + group_id: g, + participation: cgka_traits::GroupParticipation::Evicted, + } if g == &group_id + )), + "bob should emit ParticipationChanged(Evicted); got {bob_remove_events:?}" + ); // After being removed, bob retains a tombstoned local record of the group // (the engine does NOT eagerly destroy local state on removal — retaining // it preserves the convergence artifacts a late winning branch needs to @@ -774,6 +794,27 @@ async fn readd_after_remove_produces_fresh_welcome_join() { "bob should no longer be a member of his retained record; got {bob_after_remove:?}" ); + // While bob is out, alice sends an app message at the post-removal epoch — + // an epoch bob never held keys for. Kept for the PreMembership assertion + // after the rejoin below. + let gap_message = match alice + .send(SendIntent::AppMessage { + group_id: group_id.clone(), + payload: app_payload_for(&alice, b"sent while bob was out"), + }) + .await + .unwrap() + { + SendResult::ApplicationMessage { msg } => msg, + other => panic!("expected ApplicationMessage, got {other:?}"), + }; + let routed_gap_message = TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..gap_message + }; + // 3. Alice re-adds bob with a brand-new KeyPackage (never reuse the first). let bob_kp_2 = bob.fresh_key_package().await.unwrap(); let readd = alice @@ -783,16 +824,37 @@ async fn readd_after_remove_produces_fresh_welcome_join() { }) .await .unwrap(); - let (readd_pending, re_welcome) = match readd { + let (readd_pending, re_welcome, readd_commit) = match readd { SendResult::GroupEvolution { + msg, mut welcomes, pending, - .. - } => (pending, welcomes.remove(0)), + } => (pending, welcomes.remove(0), msg), other => panic!("expected GroupEvolution, got {other:?}"), }; alice.confirm_published(readd_pending).await.unwrap(); + // Post-eviction group traffic classifies as `evicted` (stale, terminal): + // an evicted client was removed from the ratchet, cannot apply a later + // commit for the group, and MUST NOT try (spec group-state.md). + // Reinstatement arrives only through the fresh Welcome below. + let routed_readd_commit = TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..readd_commit + }; + let evicted_outcome = bob.ingest(routed_readd_commit).await.unwrap(); + assert!( + matches!( + evicted_outcome, + IngestOutcome::Stale { + reason: cgka_traits::ingest::StaleReason::Evicted + } + ), + "post-eviction commit should classify as Stale(Evicted); got {evicted_outcome:?}" + ); + // 4. Bob ingests the NEW Welcome and must successfully re-join: emit // GroupJoined, the group is visible again, and bob is a member. Before // the fix this was a silent no-op / error on stale leftover state. @@ -805,6 +867,24 @@ async fn readd_after_remove_produces_fresh_welcome_join() { )), "bob should emit GroupJoined on the re-add Welcome; got {rejoin_events:?}" ); + // A verified rejoin is the reinstatement path: participation returns to + // `Member` through the fresh membership grant, and the transition is + // surfaced rather than silently overwritten. + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Member), + "bob's participation should be Member again after the re-add welcome" + ); + assert!( + rejoin_events.iter().any(|event| matches!( + event, + cgka_traits::engine::GroupEvent::ParticipationChanged { + group_id: g, + participation: cgka_traits::GroupParticipation::Member, + } if g == &group_id + )), + "bob should emit ParticipationChanged(Member) on rejoin; got {rejoin_events:?}" + ); let bob_members = bob.members(&group_id).unwrap(); assert!( bob_members.iter().any(|member| member.id == bob.self_id()), @@ -821,6 +901,31 @@ async fn readd_after_remove_produces_fresh_welcome_join() { bob_members.len(), "alice and bob should agree on the re-added group's membership" ); + + // Membership intervals (errors.md, `PreMembership`): the first interval + // closed at the last epoch bob held keys for (the removal commit's source + // epoch), and the rejoin opened a fresh one at the welcome's epoch. + let bob_record = bob.participation(&group_id).unwrap(); + assert_eq!(bob_record, Some(cgka_traits::GroupParticipation::Member)); + let bob_group = { + // Read via the engine surface: members()+epoch() prove liveness; the + // interval shape is asserted through a gap-epoch classification below. + bob.epoch(&group_id).unwrap() + }; + let _ = bob_group; + + // A message from inside bob's removed gap: alice sends an app message + // while bob is out (epoch 2 for alice, bob's readable epochs are 1 and 3+). + let gap_outcome = bob.ingest(routed_gap_message).await.unwrap(); + assert!( + matches!( + gap_outcome, + IngestOutcome::Stale { + reason: cgka_traits::ingest::StaleReason::PreMembership + } + ), + "a gap-epoch message is PreMembership — terminal, never retried; got {gap_outcome:?}" + ); } #[tokio::test] @@ -1284,6 +1389,31 @@ async fn selfremove_full_flow_with_auto_commit() { emits_departure_of(&bob_events, &bob.self_id()), "bob should emit a departure for himself; got {bob_events:?}" ); + // The commit consumed bob's own SelfRemove, so the participation reason + // resolves to `Left` (not `Evicted`) — read from the applied commit, on + // the convergence apply path as well as the direct seam (spec + // group-state.md, "Reaching a non-member state"). + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Left), + "leaver's participation should resolve to Left from his consumed SelfRemove" + ); + assert!( + bob_events.iter().any(|event| matches!( + event, + cgka_traits::engine::GroupEvent::ParticipationChanged { + group_id: g, + participation: cgka_traits::GroupParticipation::Left, + } if g == &group_id + )), + "bob should emit ParticipationChanged(Left); got {bob_events:?}" + ); + // The remaining member's own participation is untouched by bob's leave. + assert_eq!( + alice.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Member), + "alice remains a Member after bob's departure" + ); } #[tokio::test] @@ -1787,3 +1917,183 @@ fn walk(dir: &std::path::Path) -> Vec { } out } + +#[tokio::test] +async fn quarantine_withholds_inbound_and_resolve_restores_member() { + let mut alice = build_client(b"alice"); + let mut bob = build_client(b"bob"); + let mut carol = build_client(b"carol"); + let bob_kp = bob.fresh_key_package().await.unwrap(); + let carol_kp = carol.fresh_key_package().await.unwrap(); + + let (group_id, create) = alice + .create_group(CreateGroupRequest { + name: "quarantine".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let welcome_for_bob = match create { + SendResult::GroupCreated { + pending, + mut welcomes, + } => { + alice.confirm_published(pending).await.unwrap(); + welcomes.remove(0) + } + _ => unreachable!(), + }; + bob.join_welcome(welcome_for_bob).await.unwrap(); + bob.drain_events(); + + // Hold bob's copy of the group. + bob.quarantine_group(&group_id, cgka_traits::QuarantineReason::PendingMembership) + .await + .unwrap(); + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Quarantined { + reason: cgka_traits::QuarantineReason::PendingMembership + }) + ); + let bob_events = bob.drain_events(); + assert!( + bob_events.iter().any(|event| matches!( + event, + cgka_traits::engine::GroupEvent::ParticipationChanged { + participation: cgka_traits::GroupParticipation::Quarantined { .. }, + .. + } + )), + "quarantine should surface as ParticipationChanged; got {bob_events:?}" + ); + + // Live inbound is withheld (retained, not processed): alice invites carol + // and bob's copy of that commit classifies Quarantined without advancing + // his epoch. Convergence is also excluded for the held group. + let invite = alice + .send(SendIntent::Invite { + group_id: group_id.clone(), + key_packages: vec![carol_kp], + }) + .await + .unwrap(); + let (commit, invite_pending) = match invite { + SendResult::GroupEvolution { msg, pending, .. } => (msg, pending), + other => panic!("expected GroupEvolution, got {other:?}"), + }; + alice.confirm_published(invite_pending).await.unwrap(); + let routed = TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..commit + }; + let outcome = bob.ingest(routed).await.unwrap(); + assert!( + matches!( + outcome, + IngestOutcome::Stale { + reason: cgka_traits::ingest::StaleReason::Quarantined + } + ), + "inbound for a withheld group is retained, not processed; got {outcome:?}" + ); + assert_eq!(bob.epoch(&group_id).unwrap().0, 1, "epoch must not advance"); + assert!( + bob.advance_convergence(&group_id).await.unwrap().is_empty(), + "live convergence skips a withheld group" + ); + assert_eq!(bob.epoch(&group_id).unwrap().0, 1); + + // Explicit recovery: the resolution pass consumes the retained commit + // through the ordinary apply seams and restores Member. + let resolved = bob.resolve_group_quarantine(&group_id).await.unwrap(); + assert_eq!(resolved, cgka_traits::GroupParticipation::Member); + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Member) + ); + assert_eq!( + bob.epoch(&group_id).unwrap().0, + 2, + "the withheld commit applies during resolution" + ); + assert_eq!(bob.members(&group_id).unwrap().len(), 3); +} + +#[tokio::test] +async fn quarantine_never_downgrades_an_authoritative_non_member_state() { + let mut alice = build_client(b"alice"); + let mut bob = build_client(b"bob"); + let bob_kp = bob.fresh_key_package().await.unwrap(); + + let (group_id, create) = alice + .create_group(CreateGroupRequest { + name: "no-downgrade".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let welcome_for_bob = match create { + SendResult::GroupCreated { + pending, + mut welcomes, + } => { + alice.confirm_published(pending).await.unwrap(); + welcomes.remove(0) + } + _ => unreachable!(), + }; + bob.join_welcome(welcome_for_bob).await.unwrap(); + + let remove = alice + .send(SendIntent::RemoveMembers { + group_id: group_id.clone(), + members: vec![bob.self_id()], + }) + .await + .unwrap(); + let (remove_commit, remove_pending) = match remove { + SendResult::GroupEvolution { msg, pending, .. } => (msg, pending), + other => panic!("expected GroupEvolution, got {other:?}"), + }; + alice.confirm_published(remove_pending).await.unwrap(); + let routed = TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: group_id.as_slice().to_vec(), + }, + ..remove_commit + }; + let outcome = bob.ingest(routed).await.unwrap(); + assert!(matches!(outcome, IngestOutcome::Buffered { .. })); + converge_buffered_commit(&mut bob, &group_id); + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Evicted) + ); + + // Quarantine is a hold, not a verdict: it must not mask Evicted. + bob.quarantine_group(&group_id, cgka_traits::QuarantineReason::IntegrityHold) + .await + .unwrap(); + assert_eq!( + bob.participation(&group_id).unwrap(), + Some(cgka_traits::GroupParticipation::Evicted), + "an authoritative non-member state is never downgraded to a hold" + ); + // Resolving a non-quarantined group is a no-op that reports the current + // authoritative state. + assert_eq!( + bob.resolve_group_quarantine(&group_id).await.unwrap(), + cgka_traits::GroupParticipation::Evicted + ); +} diff --git a/crates/cgka-session/src/lib.rs b/crates/cgka-session/src/lib.rs index 5c3cd511..1e13ec81 100644 --- a/crates/cgka-session/src/lib.rs +++ b/crates/cgka-session/src/lib.rs @@ -569,6 +569,36 @@ impl AccountDeviceSession { self.engine.self_id() } + /// Wrap a removal notice for `recipient` embedding the already-published + /// commit (spec/protocol-core/member-departure.md, "Removal notices"). + /// `Ok(None)` means the active transport binding does not carry removal + /// notices. + pub async fn wrap_removal_notice( + &self, + commit: &TransportMessage, + recipient: &MemberId, + ) -> Result, EngineError> { + self.engine.wrap_removal_notice(commit, recipient).await + } + + /// Withhold a group pending an explicit recovery transition + /// (spec/protocol-core/group-state.md, "Quarantine"). + pub async fn quarantine_group( + &mut self, + group_id: &GroupId, + reason: cgka_traits::QuarantineReason, + ) -> Result<(), EngineError> { + self.engine.quarantine_group(group_id, reason).await + } + + /// The explicit recovery transition out of `Quarantined`. + pub async fn resolve_group_quarantine( + &mut self, + group_id: &GroupId, + ) -> Result { + self.engine.resolve_group_quarantine(group_id).await + } + pub fn set_convergence_policy(&mut self, policy: CanonicalizationPolicy) { tracing::debug!( target: TRACE_TARGET, diff --git a/crates/marmot-account/src/runtime.rs b/crates/marmot-account/src/runtime.rs index a435868e..100127f1 100644 --- a/crates/marmot-account/src/runtime.rs +++ b/crates/marmot-account/src/runtime.rs @@ -17,7 +17,7 @@ use cgka_traits::ingest::IngestOutcome; use cgka_traits::transport::{TransportEnvelope, TransportMessage}; use cgka_traits::{ EpochId, GroupId, Timestamp, TransportAccountActivation, TransportAdapter, TransportDelivery, - TransportGroupSync, TransportPublishReport, TransportPublishRequest, + TransportGroupBackfill, TransportGroupSync, TransportPublishReport, TransportPublishRequest, }; use marmot_forensics::{ AuditEventContext, AuditEventKind, AuditTransportWire, MessageArtifactKind, PublishRelayFailure, @@ -182,6 +182,60 @@ where Ok(()) } + /// Withhold a group pending an explicit recovery transition + /// (spec/protocol-core/group-state.md, "Quarantine"). The resulting + /// `ParticipationChanged` event surfaces on the next drain. + pub async fn quarantine_group( + &mut self, + group_id: &GroupId, + reason: cgka_traits::QuarantineReason, + ) -> AccountResult<()> { + self.session.quarantine_group(group_id, reason).await?; + Ok(()) + } + + /// The explicit recovery transition out of `Quarantined`. + pub async fn resolve_group_quarantine( + &mut self, + group_id: &GroupId, + ) -> AccountResult { + Ok(self.session.resolve_group_quarantine(group_id).await?) + } + + /// Membership recovery probe: re-issue one group's subscription with a + /// widened `since` so a missed group-evolution commit (most importantly a + /// missed removal commit) is recovered from relay history + /// (spec/protocol-core/group-state.md, "Reaching a non-member state"). + /// A group with no live route (already withheld/quarantined) is a no-op. + pub async fn backfill_transport_group( + &self, + group_id: &GroupId, + since: Option, + ) -> AccountResult<()> { + let Some(group_subscription) = self + .routing + .group_subscriptions() + .into_iter() + .find(|subscription| &subscription.group_id == group_id) + else { + return Ok(()); + }; + tracing::debug!( + target: TRACE_TARGET, + method = "backfill_transport_group", + has_since = since.is_some(), + "issuing membership backfill probe" + ); + self.adapter + .backfill_account_group(TransportGroupBackfill { + account_id: self.session.self_id(), + group_subscription, + since, + }) + .await?; + Ok(()) + } + pub async fn publish_fresh_key_package(&mut self) -> AccountResult { tracing::debug!( target: TRACE_TARGET, @@ -342,6 +396,10 @@ where let mut queue = VecDeque::new(); output.absorb_session_effects(effects, &mut queue); + // Commits published in this drain, kept so the removal-notice pass + // below can embed the exact published wire event + // (spec/protocol-core/member-departure.md, "Removal notices"). + let mut published_commits: Vec = Vec::new(); while let Some(work) = queue.pop_front() { match work { PublishWork::ApplicationMessage { msg } | PublishWork::Proposal { msg } => { @@ -362,6 +420,7 @@ where welcomes, pending, } => { + published_commits.push(msg.clone()); self.publish_group_evolution( msg, welcomes, @@ -373,6 +432,7 @@ where .await?; } PublishWork::AutoPublish { msg, pending } => { + published_commits.push(msg.clone()); self.publish_pending( vec![msg], pending, @@ -384,10 +444,99 @@ where } } } + self.send_removal_notices(&published_commits, &output.events) + .await; Ok(output) } + /// Committer-side removal notices (spec/protocol-core/member-departure.md, + /// "Removal notices"): after a commit this device published and confirmed + /// removes members, gift-deliver each removed member the published commit + /// through its account inbox. The correlation is the confirmed events' + /// `origin_commit_id` against the commits published in this drain — only + /// the committer holds the published wire event, so only the committer + /// sends (the SHOULD in the spec). Notices are a delivery aid: every + /// failure here is logged and swallowed — a missing inbox route or a + /// binding without notice support MUST NOT fail the commit flow. + async fn send_removal_notices( + &self, + published_commits: &[TransportMessage], + events: &[GroupEvent], + ) { + if published_commits.is_empty() { + return; + } + let self_id = self.session.self_id(); + for event in events { + let GroupEvent::GroupStateChanged { + change: + cgka_traits::engine::GroupStateChange::MemberRemoved { member } + | cgka_traits::engine::GroupStateChange::MemberLeft { member }, + origin_commit_id: Some(origin_commit_id), + .. + } = event + else { + continue; + }; + if member == &self_id { + continue; + } + let Some(commit) = published_commits + .iter() + .find(|commit| &commit.id == origin_commit_id) + else { + continue; + }; + let notice = match self.session.wrap_removal_notice(commit, member).await { + // The active binding does not carry removal notices. + Ok(None) => continue, + Ok(Some(notice)) => notice, + Err(_) => { + tracing::warn!( + target: TRACE_TARGET, + method = "send_removal_notices", + error_code = "wrap_failed", + error_summary = "redacted", + "failed to wrap removal notice" + ); + continue; + } + }; + let target = match self.routing.publish_target(¬ice) { + Ok(target) => target, + Err(_) => { + // No cached inbox route for the removed member. The notice + // is best-effort; the recovery probe and ordinary delivery + // remain the other discovery paths. + tracing::debug!( + target: TRACE_TARGET, + method = "send_removal_notices", + error_code = "no_inbox_route", + error_summary = "redacted", + "skipping removal notice without an inbox route" + ); + continue; + } + }; + let request = TransportPublishRequest { + account_id: self_id.clone(), + message: notice, + target, + required_acks: 1, + }; + if self.adapter.publish(request).await.is_err() { + tracing::warn!( + target: TRACE_TARGET, + method = "send_removal_notices", + error_code = "publish_failed", + error_summary = "redacted", + "failed to publish removal notice" + ); + } + } + } + /// Confirm a published commit, retrying on transient backend contention. /// /// `confirm_published` is the apply half of publish-before-apply: by the diff --git a/crates/marmot-account/tests/runtime.rs b/crates/marmot-account/tests/runtime.rs index bd65927d..d78695ac 100644 --- a/crates/marmot-account/tests/runtime.rs +++ b/crates/marmot-account/tests/runtime.rs @@ -126,6 +126,29 @@ impl TransportPeeler for MockPeeler { }) } + async fn wrap_removal_notice( + &self, + commit: &TransportMessage, + recipient: &MemberId, + ) -> Result, PeelerError> { + // Synthetic notice: inbox-addressed carrier for the published commit, + // recognizable by carrying the commit's payload. The Nostr wire shape + // is covered by transport-nostr-peeler's removal_notice tests; this + // exercises the runtime's committer-side correlation + publish seam. + let mut id_material = commit.id.as_slice().to_vec(); + id_material.extend_from_slice(recipient.as_slice()); + Ok(Some(TransportMessage { + id: hash_id(&id_material), + payload: commit.payload.clone(), + timestamp: commit.timestamp, + causal_deps: vec![], + source: TransportSource("marmot-account-test".into()), + envelope: TransportEnvelope::Welcome { + recipient: recipient.clone(), + }, + })) + } + async fn wrap_welcome( &self, payload: &EncryptedPayload, @@ -198,6 +221,7 @@ struct RecordingAdapter { struct RecordingAdapterInner { activations: Mutex>, syncs: Mutex>, + backfills: Mutex>, publishes: Mutex>, accepted_counts: Mutex>, } @@ -215,6 +239,10 @@ impl RecordingAdapter { .push_back(accepted_count); } + fn backfills(&self) -> Vec { + self.inner.backfills.lock().unwrap().clone() + } + fn activations(&self) -> Vec { self.inner.activations.lock().unwrap().clone() } @@ -242,6 +270,14 @@ impl TransportAdapter for RecordingAdapter { Ok(()) } + async fn backfill_account_group( + &self, + backfill: cgka_traits::TransportGroupBackfill, + ) -> Result<(), TransportAdapterError> { + self.inner.backfills.lock().unwrap().push(backfill); + Ok(()) + } + async fn deactivate_account( &self, _account_id: &MemberId, @@ -405,6 +441,50 @@ async fn activate_transport_uses_session_identity_and_policy() { assert_eq!(activations[0].since, Some(Timestamp(10))); } +#[tokio::test] +async fn backfill_transport_group_reissues_only_that_groups_subscription() { + let dir = tempfile::tempdir().unwrap(); + let key = SqlCipherKey::new("marmot account backfill key").unwrap(); + let session = session(dir.path().join("alice.sqlite"), &key, b"alice"); + let adapter = RecordingAdapter::default(); + let group_id = cgka_traits::GroupId::new(vec![0xAB; 8]); + let other_group = cgka_traits::GroupId::new(vec![0xCD; 8]); + let policy = StaticTransportRouting::new(vec![TransportEndpoint("wss://inbox.example".into())]) + .with_group_route( + group_id.clone(), + group_id.as_slice().to_vec(), + vec![TransportEndpoint("wss://group.example".into())], + ); + let runtime = AccountDeviceRuntime::new( + session, + adapter.clone(), + policy, + RecordingKeyPackages::default(), + ); + + // The membership recovery probe forwards the group's routed subscription + // and the widened anchor to the adapter... + runtime + .backfill_transport_group(&group_id, Some(Timestamp(1_700_000_000))) + .await + .unwrap(); + // ...and a group with no live route is a no-op, not an error. + runtime + .backfill_transport_group(&other_group, Some(Timestamp(1_700_000_000))) + .await + .unwrap(); + + let backfills = adapter.backfills(); + assert_eq!(backfills.len(), 1); + assert_eq!(backfills[0].account_id, runtime.session().self_id()); + assert_eq!(backfills[0].group_subscription.group_id, group_id); + assert_eq!( + backfills[0].group_subscription.endpoints, + vec![TransportEndpoint("wss://group.example".into())] + ); + assert_eq!(backfills[0].since, Some(Timestamp(1_700_000_000))); +} + #[tokio::test] async fn publish_fresh_key_package_uses_directory_boundary() { let dir = tempfile::tempdir().unwrap(); @@ -1006,6 +1086,8 @@ async fn drain_surfaces_hydration_quarantine_without_inbound_delivery() { members: Vec::new(), epoch: EpochId(9), required_capabilities: GroupCapabilities::default(), + participation: Default::default(), + membership_intervals: Vec::new(), }) .unwrap(); } @@ -1185,3 +1267,87 @@ async fn auto_publish_confirms_pending_when_commit_was_partially_exposed() { assert_eq!(runtime.session().epoch(&group_id).unwrap().0, 2); assert_eq!(runtime.session().members(&group_id).unwrap().len(), 1); } + +#[tokio::test] +async fn committer_publishes_removal_notice_to_removed_members_inbox() { + let dir = tempfile::tempdir().unwrap(); + let key = SqlCipherKey::new("marmot removal notice key").unwrap(); + let mut alice_session = session(dir.path().join("alice.sqlite"), &key, b"alice"); + let mut bob_session = session(dir.path().join("bob.sqlite"), &key, b"bob"); + let bob_kp = bob_session.fresh_key_package().await.unwrap(); + let bob_id = bob_session.self_id(); + + let created = alice_session + .create_group(CreateGroupRequest { + name: "removal notice".into(), + description: "".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![], + initial_admins: vec![], + }) + .await + .unwrap(); + let create_pending = match &created.effects.publish[0] { + PublishWork::GroupCreated { pending, .. } => *pending, + other => panic!("expected GroupCreated publish work, got {other:?}"), + }; + alice_session + .confirm_published(create_pending) + .await + .unwrap(); + + let adapter = RecordingAdapter::default(); + adapter.accept_next(1); // the removal commit + adapter.accept_next(1); // the removal notice + let policy = + StaticTransportRouting::new(vec![TransportEndpoint("wss://alice-inbox.example".into())]) + .with_group_route( + created.group_id.clone(), + created.group_id.as_slice().to_vec(), + vec![TransportEndpoint("wss://group.example".into())], + ) + .with_inbox_route( + bob_id.clone(), + vec![TransportEndpoint("wss://bob-inbox.example".into())], + ); + let mut runtime = AccountDeviceRuntime::new( + alice_session, + adapter.clone(), + policy, + RecordingKeyPackages::default(), + ); + + runtime + .send(SendIntent::RemoveMembers { + group_id: created.group_id.clone(), + members: vec![bob_id.clone()], + }) + .await + .unwrap(); + + // Publish 1: the removal commit to the group's relays. Publish 2: the + // committer-side removal notice, inbox-addressed to the removed member and + // carrying the exact published commit (spec member-departure.md, "Removal + // notices"). + let publishes = adapter.publishes(); + assert_eq!(publishes.len(), 2, "commit + removal notice"); + assert!(matches!( + publishes[0].message.envelope, + TransportEnvelope::GroupMessage { .. } + )); + assert!(matches!( + &publishes[1].message.envelope, + TransportEnvelope::Welcome { recipient } if recipient == &bob_id + )); + assert!(matches!( + &publishes[1].target, + cgka_traits::TransportPublishTarget::Inbox { recipient, endpoints } + if recipient == &bob_id + && endpoints == &vec![TransportEndpoint("wss://bob-inbox.example".into())] + )); + assert_eq!( + publishes[1].message.payload, publishes[0].message.payload, + "the notice embeds the published commit" + ); +} diff --git a/crates/marmot-app/src/client/mod.rs b/crates/marmot-app/src/client/mod.rs index 89861e46..7eabc3b2 100644 --- a/crates/marmot-app/src/client/mod.rs +++ b/crates/marmot-app/src/client/mod.rs @@ -66,6 +66,11 @@ pub struct AppClient { /// broadcasts `ProjectionUpdated` so live timeline subscriptions refresh. pub(crate) pending_projection_updates: Vec, pub(crate) pending_convergence_groups: HashSet, + /// Per-group membership recovery probe bookkeeping (attempt count + + /// last-attempt time), in-memory: probes are cheap re-subscriptions, so a + /// restart simply re-arms them on the next undecryptable delivery. + pub(crate) membership_probe_state: + std::collections::HashMap, } /// A point-in-time copy of the live session's read-only group projections @@ -329,6 +334,7 @@ impl AppClient { .iter() .copied() .collect(), + participation: crate::groups::group_participation_tag(&group.participation).to_owned(), }) } @@ -1953,7 +1959,20 @@ impl AppClient { /// actually change rather than only on a membership-count change. pub(crate) fn refresh_group_routes(&mut self) -> Result { let mut changed = false; + // Groups the local account left or was removed from keep their + // projection rows (the chat list still renders them) but MUST NOT + // rejoin the live routing/subscription set: re-adding a dead group's + // route here would silently resubscribe it on the next transport sync + // (spec/protocol-core/group-state.md, "Participation"). + let removed_membership: std::collections::HashSet = self + .app + .account_group_ids_with_removed_membership(&self.state.label)? + .into_iter() + .collect(); for group in &self.state.groups { + if removed_membership.contains(&group.group_id_hex) { + continue; + } let group_id = GroupId::new(hex::decode(&group.group_id_hex)?); if self .routing diff --git a/crates/marmot-app/src/client/sync.rs b/crates/marmot-app/src/client/sync.rs index 3d15cc55..8a0530d9 100644 --- a/crates/marmot-app/src/client/sync.rs +++ b/crates/marmot-app/src/client/sync.rs @@ -265,11 +265,27 @@ impl AppClient { ) -> Result<(), AppError> { let source_message_id_hex = hex::encode(delivery.message.id.as_slice()); let source_recorded_at = delivery.message.timestamp.0; + let group_hint = delivery.group_id_hint.clone(); let effects = self.runtime.ingest_delivery(delivery).await?; fail_if_publish_failed(&effects.effects)?; self.remember_buffered_convergence_outcome(&effects.outcome); self.remember_pending_convergence_effects(&effects.effects); - self.remember_transport_cursor(source_recorded_at); + if outcome_advances_transport_cursor(&effects.outcome) { + self.remember_transport_cursor(source_recorded_at); + if let Some(group_id) = &group_hint { + // Decryption resumed (or the input was consumed): the group is + // not starving on a missed commit, so re-arm its probe budget. + self.membership_probe_state + .remove(&hex::encode(group_id.as_slice())); + } + } else if let Some(group_id) = &group_hint { + // An undecryptable delivery is evidence that something EARLIER is + // missing — possibly the commit that removed us (#722). The cursor + // stays anchored at the last consumed input (spec group-state.md: + // the recovery window is anchored there, not at last received), + // and the group's membership recovery probe fires with backoff. + self.maybe_backfill_group_membership(group_id).await?; + } self.observe_account_device_effects( &effects.effects, display_names, @@ -534,6 +550,58 @@ impl AppClient { SelfMembership::Member, )?; } + // Participation is the authoritative live-group gate + // (spec/protocol-core/group-state.md): a group the local identity + // left, was evicted from, or that is quarantined leaves the + // routing table and the transport subscription set NOW — + // deterministically, not only when the in-memory group count + // happens to change (the `len() != before` gate above misses a + // convergence-applied removal, leaving the dead group subscribed + // forever). A restored `Member` re-enters the live set the same + // way. + if let cgka_traits::engine::GroupEvent::ParticipationChanged { + group_id, + participation, + } = event + { + let group_id_hex = hex::encode(group_id.as_slice()); + match participation { + cgka_traits::GroupParticipation::Left => { + self.app.set_group_self_membership( + &self.state.label, + &group_id_hex, + SelfMembership::Left, + )?; + self.routing.remove_group(group_id); + self.sync_runtime_groups().await?; + } + cgka_traits::GroupParticipation::Evicted => { + self.app.set_group_self_membership( + &self.state.label, + &group_id_hex, + SelfMembership::Removed, + )?; + self.routing.remove_group(group_id); + self.sync_runtime_groups().await?; + } + cgka_traits::GroupParticipation::Quarantined { .. } => { + // Withheld, not asserted non-member: no membership + // flag write, but the group leaves live routing until + // an explicit recovery transition. + self.routing.remove_group(group_id); + self.sync_runtime_groups().await?; + } + cgka_traits::GroupParticipation::Member => { + self.app.set_group_self_membership( + &self.state.label, + &group_id_hex, + SelfMembership::Member, + )?; + self.refresh_group_routes()?; + self.sync_runtime_groups().await?; + } + } + } } // #760: strip all collected push-gossip messages in one pass. if !gossip_message_ids.is_empty() { @@ -575,8 +643,133 @@ impl AppClient { TRANSPORT_CURSOR_MAX_FUTURE_SKEW.as_secs(), )); } + + /// Membership recovery probe driver (spec/protocol-core/group-state.md, + /// "Reaching a non-member state"): on undecryptable traffic for a group, + /// re-fetch its message stream from a bounded window anchored at the last + /// input this client successfully consumed, with exponential backoff so a + /// relay withholding the commit costs bounded bandwidth, not a hot loop. + async fn maybe_backfill_group_membership( + &mut self, + group_id: &cgka_traits::GroupId, + ) -> Result<(), AppError> { + let group_id_hex = hex::encode(group_id.as_slice()); + // Only probe groups this account actually holds a record of. + if self.state_group_record(group_id).is_none() { + return Ok(()); + } + let now = unix_now_seconds(); + let due = self + .membership_probe_state + .get(&group_id_hex) + .is_none_or(|probe| probe.next_due(now)); + if !due { + return Ok(()); + } + let entry = self + .membership_probe_state + .entry(group_id_hex.clone()) + .or_default(); + entry.attempts = entry.attempts.saturating_add(1); + entry.last_attempt_at = now; + // Probes stayed dry past the policy bound: hold the group as + // Quarantined(PendingMembership) instead of probing forever (spec + // group-state.md, "Reaching a non-member state", bounded hold). The + // kind-451 inbox notice remains a live discovery path (it does not + // need the group route), and explicit resolution re-runs convergence + // over everything retained meanwhile. + if entry.attempts > MEMBERSHIP_PROBE_QUARANTINE_ATTEMPTS { + self.runtime + .quarantine_group(group_id, cgka_traits::QuarantineReason::PendingMembership) + .await?; + let drained = self.drain_pending_session_events().await?; + let _ = drained; + return Ok(()); + } + // Anchor: the newest app event this client consumed for the group — + // derived from decrypted input, so it is "the last input successfully + // consumed" — widened by a slack. No consumed input yet means a full + // re-fetch of the group's stream (bounded by relay retention; dedup + // collapses the overlap). + let anchor = self + .app + .messages_with_query( + &self.state.label, + crate::AppMessageQuery { + group_id_hex: Some(group_id_hex), + limit: Some(1), + }, + )? + .into_iter() + .next_back() + .map(|message| { + clamped_transport_cursor( + None, + message.recorded_at, + now, + TRANSPORT_CURSOR_MAX_FUTURE_SKEW.as_secs(), + ) + }); + let since = anchor + .map(|at| cgka_traits::Timestamp(at.saturating_sub(MEMBERSHIP_BACKFILL_SLACK_SECS))); + self.runtime + .backfill_transport_group(group_id, since) + .await?; + Ok(()) + } +} + +/// Whether an ingest outcome may advance the persisted transport cursor. +/// +/// The cursor becomes the relay `since` filter, so advancing it past input we +/// could not decrypt permanently skips the window that likely contains the +/// missing group-evolution commit — the #722 silent-eviction mechanism: a +/// post-removal message advances the cursor beyond the removal commit's +/// timestamp and no future subscription ever fetches it. Undecryptable input +/// therefore freezes the cursor at the last consumed input; every other +/// outcome (consumed, buffered-for-convergence, terminal-stale) is durably +/// accounted for and safe to advance past. +pub(crate) fn outcome_advances_transport_cursor( + outcome: &cgka_traits::ingest::IngestOutcome, +) -> bool { + !matches!( + outcome, + cgka_traits::ingest::IngestOutcome::Stale { + reason: cgka_traits::ingest::StaleReason::PeelFailed + } + ) +} + +/// Backoff bookkeeping for one group's membership recovery probe. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct MembershipProbeState { + pub(crate) attempts: u32, + pub(crate) last_attempt_at: u64, +} + +impl MembershipProbeState { + /// Exponential cooldown: 60s doubling per attempt, capped at one hour. + pub(crate) fn next_due(&self, now: u64) -> bool { + if self.attempts == 0 { + return true; + } + let shift = (self.attempts - 1).min(6); + let cooldown = (MEMBERSHIP_BACKFILL_BASE_COOLDOWN_SECS << shift) + .min(MEMBERSHIP_BACKFILL_MAX_COOLDOWN_SECS); + now.saturating_sub(self.last_attempt_at) >= cooldown + } } +/// Widen the probe window this far before the last consumed input, absorbing +/// relay timestamp skew between the missed commit and the anchor message. +const MEMBERSHIP_BACKFILL_SLACK_SECS: u64 = 300; +const MEMBERSHIP_BACKFILL_BASE_COOLDOWN_SECS: u64 = 60; +const MEMBERSHIP_BACKFILL_MAX_COOLDOWN_SECS: u64 = 3600; +/// Dry probes tolerated before the group is held as +/// `Quarantined(PendingMembership)` — with the backoff schedule above this is +/// several hours of failed recovery attempts. +const MEMBERSHIP_PROBE_QUARANTINE_ATTEMPTS: u32 = 8; + pub(crate) fn is_own_relay_echo( delivery: &cgka_traits::TransportDelivery, local_account_id_hex: &str, @@ -751,3 +944,73 @@ mod transport_cursor_tests { ); } } + +#[cfg(test)] +mod membership_probe_tests { + use super::{MembershipProbeState, outcome_advances_transport_cursor}; + use cgka_traits::ingest::{IngestOutcome, StaleReason}; + + #[test] + fn undecryptable_input_freezes_the_cursor_and_everything_else_advances_it() { + // The #722 mechanism: advancing the persisted cursor past input we + // could not decrypt permanently skips the relay window that likely + // contains the missed removal commit. + assert!(!outcome_advances_transport_cursor(&IngestOutcome::Stale { + reason: StaleReason::PeelFailed + })); + + assert!(outcome_advances_transport_cursor(&IngestOutcome::Processed)); + assert!(outcome_advances_transport_cursor( + &IngestOutcome::Buffered { + group_id: cgka_traits::GroupId::new(vec![1; 4]), + epoch: cgka_traits::EpochId(1), + } + )); + for reason in [ + StaleReason::AlreadySeen, + StaleReason::NotForThisClient, + StaleReason::UnknownGroup, + StaleReason::OwnEcho, + StaleReason::Evicted, + ] { + let outcome = IngestOutcome::Stale { + reason: reason.clone(), + }; + assert!( + outcome_advances_transport_cursor(&outcome), + "durably-accounted-for outcome should advance the cursor: {reason:?}" + ); + } + } + + #[test] + fn probe_backoff_doubles_per_attempt_and_caps_at_an_hour() { + let now = 1_800_000_000; + // Fresh group: due immediately. + assert!(MembershipProbeState::default().next_due(now)); + + // First retry waits the 60s base cooldown. + let one = MembershipProbeState { + attempts: 1, + last_attempt_at: now, + }; + assert!(!one.next_due(now + 59)); + assert!(one.next_due(now + 60)); + + // Fourth retry waits 60 << 3 = 480s. + let four = MembershipProbeState { + attempts: 4, + last_attempt_at: now, + }; + assert!(!four.next_due(now + 479)); + assert!(four.next_due(now + 480)); + + // Deep attempt counts cap at one hour, not overflow. + let deep = MembershipProbeState { + attempts: 40, + last_attempt_at: now, + }; + assert!(!deep.next_due(now + 3_599)); + assert!(deep.next_due(now + 3_600)); + } +} diff --git a/crates/marmot-app/src/groups.rs b/crates/marmot-app/src/groups.rs index d5c01b04..ab597109 100644 --- a/crates/marmot-app/src/groups.rs +++ b/crates/marmot-app/src/groups.rs @@ -75,6 +75,35 @@ pub struct AppGroupMlsState { pub epoch: u64, pub member_count: usize, pub required_app_components: Vec, + /// The local identity's participation in the group, as a stable + /// low-cardinality tag (see [`group_participation_tag`]). Clients gate + /// the composer and group list on it: only `"member"` allows sending + /// (spec/protocol-core/group-state.md, "Participation and public + /// surfaces"). Defaults to `"member"` for snapshots serialized before the + /// field existed. + #[serde(default = "default_participation_member")] + pub participation: String, +} + +fn default_participation_member() -> String { + group_participation_tag(&cgka_traits::GroupParticipation::Member).to_owned() +} + +/// Stable, low-cardinality tag for a [`cgka_traits::GroupParticipation`]. +/// String-tagged so app and FFI surfaces stay additive when new participation +/// states or quarantine reasons appear. +pub fn group_participation_tag(participation: &cgka_traits::GroupParticipation) -> &'static str { + match participation { + cgka_traits::GroupParticipation::Member => "member", + cgka_traits::GroupParticipation::Left => "left", + cgka_traits::GroupParticipation::Evicted => "evicted", + cgka_traits::GroupParticipation::Quarantined { + reason: cgka_traits::QuarantineReason::PendingMembership, + } => "quarantined_pending_membership", + cgka_traits::GroupParticipation::Quarantined { + reason: cgka_traits::QuarantineReason::IntegrityHold, + } => "quarantined_integrity_hold", + } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -983,7 +1012,8 @@ pub(crate) fn event_group_id(event: &GroupEvent) -> Option<&GroupId> { | GroupEvent::GroupUnrecoverable { group_id, .. } | GroupEvent::PendingCommitRecovered { group_id, .. } | GroupEvent::GroupHydrationQuarantined { group_id, .. } - | GroupEvent::GroupHydrationRecovered { group_id, .. } => Some(group_id), + | GroupEvent::GroupHydrationRecovered { group_id, .. } + | GroupEvent::ParticipationChanged { group_id, .. } => Some(group_id), } } @@ -1436,3 +1466,47 @@ mod fail_if_publish_failed_tests { } } } + +#[cfg(test)] +mod participation_tag_tests { + use super::{AppGroupMlsState, group_participation_tag}; + + #[test] + fn participation_tags_are_stable_and_snapshots_default_to_member() { + // The tag vocabulary crosses the FFI boundary; renames are breaking. + assert_eq!( + group_participation_tag(&cgka_traits::GroupParticipation::Member), + "member" + ); + assert_eq!( + group_participation_tag(&cgka_traits::GroupParticipation::Left), + "left" + ); + assert_eq!( + group_participation_tag(&cgka_traits::GroupParticipation::Evicted), + "evicted" + ); + assert_eq!( + group_participation_tag(&cgka_traits::GroupParticipation::Quarantined { + reason: cgka_traits::QuarantineReason::PendingMembership + }), + "quarantined_pending_membership" + ); + assert_eq!( + group_participation_tag(&cgka_traits::GroupParticipation::Quarantined { + reason: cgka_traits::QuarantineReason::IntegrityHold + }), + "quarantined_integrity_hold" + ); + + // Pre-field serialized snapshots decode as live members. + let legacy = serde_json::json!({ + "group_id_hex": "aa", + "epoch": 3, + "member_count": 2, + "required_app_components": [], + }); + let decoded: AppGroupMlsState = serde_json::from_value(legacy).unwrap(); + assert_eq!(decoded.participation, "member"); + } +} diff --git a/crates/marmot-app/src/lib.rs b/crates/marmot-app/src/lib.rs index ccfc80a1..04cf223b 100644 --- a/crates/marmot-app/src/lib.rs +++ b/crates/marmot-app/src/lib.rs @@ -131,7 +131,8 @@ pub use groups::{ AppGroupAvatarUrlComponent, AppGroupEncryptedMediaComponent, AppGroupHydrationQuarantineReason, AppGroupImageComponent, AppGroupMemberRecord, AppGroupMessageRetentionComponent, AppGroupMlsState, AppGroupNostrRoutingComponent, AppGroupProfileComponent, AppGroupRecord, - AppGroupSystemEvent, AppQuarantinedGroup, group_system_event_from_message, + AppGroupSystemEvent, AppQuarantinedGroup, group_participation_tag, + group_system_event_from_message, }; pub use ids::{ account_id_hex_from_ref, nprofile_for_account_id, npub_for_account_id, validate_relay_urls, @@ -1022,6 +1023,7 @@ impl MarmotApp { state: open.state, pending_projection_updates: Vec::new(), pending_convergence_groups: std::collections::HashSet::new(), + membership_probe_state: std::collections::HashMap::new(), }; // One-time upgrade backfill: derive `self_membership` for pre-0018 rows // from current engine state so groups the local account already left / @@ -1703,6 +1705,20 @@ impl MarmotApp { Ok(()) } + /// `group_id_hex` of every `account_groups` row whose local account + /// membership ended (`self_membership = 'removed'`). The routing refresh + /// filters these out of the live subscription set. + pub(crate) fn account_group_ids_with_removed_membership( + &self, + account_ref: &str, + ) -> Result, AppError> { + let account = self.account_home().account(account_ref)?; + self.ensure_account_state(&account.label)?; + Ok(self + .account_storage(&account.label)? + .account_group_ids_with_removed_membership()?) + } + /// `group_id_hex` of every `account_groups` row still carrying the migration /// default `self_membership = 'member'`. The one-time open/upgrade backfill /// uses this to derive membership for legacy rows from current engine state. @@ -2941,6 +2957,18 @@ impl AppTransportRouting { true } + /// Drop a group's route so publishes fail fast and the next transport + /// sync unsubscribes it. Participation-driven: a group the local identity + /// left, was evicted from, or that is quarantined must leave the live + /// routing set deterministically, not only when the in-memory group count + /// happens to change (spec/protocol-core/group-state.md). + fn remove_group(&self, group_id: &GroupId) { + let mut state = self.write(); + state + .group_routes + .retain(|route| &route.group_id != group_id); + } + fn snapshot(&self) -> AppRoutingState { self.read().clone() } diff --git a/crates/marmot-app/src/relay_plane/mod.rs b/crates/marmot-app/src/relay_plane/mod.rs index a053190a..bd0b49c7 100644 --- a/crates/marmot-app/src/relay_plane/mod.rs +++ b/crates/marmot-app/src/relay_plane/mod.rs @@ -763,6 +763,27 @@ impl TransportAdapter for MarmotRelayPlaneAccountAdapter { .await } + async fn backfill_account_group( + &self, + backfill: cgka_traits::TransportGroupBackfill, + ) -> Result<(), TransportAdapterError> { + if backfill.account_id != self.account_id { + return Err(TransportAdapterError::AccountNotActive(backfill.account_id)); + } + let backfill = self + .relay_plane + .inner + .relay_safety + .sanitize_group_backfill(backfill) + .map_err(TransportAdapterError::Subscription)?; + self.relay_plane + .inner + .transport + .adapter + .backfill_account_group(backfill) + .await + } + async fn deactivate_account(&self, account_id: &MemberId) -> Result<(), TransportAdapterError> { if account_id != &self.account_id { return Err(TransportAdapterError::AccountNotActive(account_id.clone())); diff --git a/crates/marmot-app/src/relay_plane/safety.rs b/crates/marmot-app/src/relay_plane/safety.rs index d25522eb..93fcc954 100644 --- a/crates/marmot-app/src/relay_plane/safety.rs +++ b/crates/marmot-app/src/relay_plane/safety.rs @@ -42,6 +42,17 @@ impl RelaySafetyPolicy { Ok(sync) } + pub(crate) fn sanitize_group_backfill( + &self, + mut backfill: cgka_traits::TransportGroupBackfill, + ) -> Result { + backfill.group_subscription.endpoints = self.sanitize_endpoints( + backfill.group_subscription.endpoints.clone(), + "group backfill", + )?; + Ok(backfill) + } + pub(crate) fn sanitize_publish_request( &self, mut request: TransportPublishRequest, diff --git a/crates/marmot-app/src/runtime/event_routing.rs b/crates/marmot-app/src/runtime/event_routing.rs index e6aa6d0d..aa7ffda1 100644 --- a/crates/marmot-app/src/runtime/event_routing.rs +++ b/crates/marmot-app/src/runtime/event_routing.rs @@ -123,9 +123,8 @@ pub(crate) fn chat_list_trigger_from_event(event: &MarmotAppEvent) -> ChatListUp | GroupEvent::GroupUnrecoverable { .. } | GroupEvent::PendingCommitRecovered { .. } | GroupEvent::GroupHydrationQuarantined { .. } - | GroupEvent::GroupHydrationRecovered { .. } => { - ChatListUpdateTrigger::MembershipChanged - } + | GroupEvent::GroupHydrationRecovered { .. } + | GroupEvent::ParticipationChanged { .. } => ChatListUpdateTrigger::MembershipChanged, GroupEvent::MessageReceived { .. } | GroupEvent::AppMessageInvalidated { .. } => { ChatListUpdateTrigger::SnapshotRefresh } @@ -150,6 +149,7 @@ fn group_id_from_event(event: &GroupEvent) -> &GroupId { | GroupEvent::GroupUnrecoverable { group_id, .. } | GroupEvent::PendingCommitRecovered { group_id, .. } | GroupEvent::GroupHydrationQuarantined { group_id, .. } - | GroupEvent::GroupHydrationRecovered { group_id, .. } => group_id, + | GroupEvent::GroupHydrationRecovered { group_id, .. } + | GroupEvent::ParticipationChanged { group_id, .. } => group_id, } } diff --git a/crates/marmot-app/src/tests.rs b/crates/marmot-app/src/tests.rs index eca3d852..4b273d18 100644 --- a/crates/marmot-app/src/tests.rs +++ b/crates/marmot-app/src/tests.rs @@ -1315,6 +1315,37 @@ fn app_transport_routing_recovers_from_poisoned_lock() { assert_eq!(routing.snapshot().required_acks, 2); } +#[test] +fn app_transport_routing_remove_group_drops_only_that_route() { + let routing = AppTransportRouting::new(AppRoutingState { + local_inbox_endpoints: Vec::new(), + key_package_endpoints: Vec::new(), + inbox_routes: HashMap::new(), + group_routes: Vec::new(), + required_acks: 1, + }); + let keep = cgka_traits::GroupId::new(vec![0xAA; 4]); + let drop = cgka_traits::GroupId::new(vec![0xBB; 4]); + for (group_id, transport_id) in [(&keep, vec![0x01; 4]), (&drop, vec![0x02; 4])] { + routing.add_group(cgka_traits::transport_adapter::TransportGroupSubscription { + group_id: group_id.clone(), + transport_group_id: transport_id, + endpoints: vec![cgka_traits::transport_adapter::TransportEndpoint( + "wss://relay.example".into(), + )], + }); + } + + routing.remove_group(&drop); + + let routes = routing.snapshot().group_routes; + assert_eq!(routes.len(), 1, "only the removed group's route is dropped"); + assert_eq!(routes[0].group_id, keep); + // Idempotent: removing an absent group is a no-op. + routing.remove_group(&drop); + assert_eq!(routing.snapshot().group_routes.len(), 1); +} + #[test] fn relay_plane_rebuild_uses_persisted_cursor_with_bounded_overlap() { let relay_plane = MarmotRelayPlane::with_subscription_rebuild_lookback(Duration::from_secs(30)); diff --git a/crates/marmot-uniffi/src/conversions/event.rs b/crates/marmot-uniffi/src/conversions/event.rs index e29298ba..9fe9435d 100644 --- a/crates/marmot-uniffi/src/conversions/event.rs +++ b/crates/marmot-uniffi/src/conversions/event.rs @@ -23,7 +23,8 @@ fn group_id_from_event(event: &GroupEvent) -> &GroupId { | GroupEvent::GroupUnrecoverable { group_id, .. } | GroupEvent::PendingCommitRecovered { group_id, .. } | GroupEvent::GroupHydrationQuarantined { group_id, .. } - | GroupEvent::GroupHydrationRecovered { group_id, .. } => group_id, + | GroupEvent::GroupHydrationRecovered { group_id, .. } + | GroupEvent::ParticipationChanged { group_id, .. } => group_id, } } @@ -124,8 +125,16 @@ pub enum GroupEventKindFfi { GroupHydrationRecovered { recovered_epoch: u64, }, + ParticipationChanged { + /// Stable, low-cardinality participation tag (see + /// [`group_participation_tag`]); clients gate the composer and group + /// list on it. + participation: String, + }, } +pub(crate) use marmot_app::group_participation_tag; + /// Stable, low-cardinality tag for a [`GroupStateChange`] — surfaced to FFI in /// place of re-modeling the member-id-bearing variants. The subject/actor ids /// are intentionally not duplicated here; clients that need them should consume @@ -231,6 +240,9 @@ impl From for GroupEventKindFfi { } => Self::GroupHydrationRecovered { recovered_epoch: recovered_epoch.0, }, + GroupEvent::ParticipationChanged { participation, .. } => Self::ParticipationChanged { + participation: group_participation_tag(&participation).to_string(), + }, } } } diff --git a/crates/marmot-uniffi/src/conversions/group.rs b/crates/marmot-uniffi/src/conversions/group.rs index 22d092e4..e3de458e 100644 --- a/crates/marmot-uniffi/src/conversions/group.rs +++ b/crates/marmot-uniffi/src/conversions/group.rs @@ -326,6 +326,11 @@ pub struct AppGroupMlsStateFfi { pub epoch: u64, pub member_count: u32, pub required_app_components: Vec, + /// Stable participation tag ("member" / "left" / "evicted" / + /// "quarantined_pending_membership" / "quarantined_integrity_hold"). + /// Clients gate the composer and group list on it: only "member" allows + /// sending. + pub participation: String, } impl From for AppGroupMlsStateFfi { @@ -335,6 +340,7 @@ impl From for AppGroupMlsStateFfi { epoch: value.epoch, member_count: value.member_count as u32, required_app_components: value.required_app_components, + participation: value.participation, } } } diff --git a/crates/storage-sqlite/src/account_projection.rs b/crates/storage-sqlite/src/account_projection.rs index 5e8e3a8e..1d88a094 100644 --- a/crates/storage-sqlite/src/account_projection.rs +++ b/crates/storage-sqlite/src/account_projection.rs @@ -549,6 +549,29 @@ impl SqliteAccountStorage { Ok(()) } + /// `group_id_hex` of every `account_groups` row whose `self_membership` + /// is `'removed'` — groups the local account left or was removed from. + /// The routing refresh uses this to keep dead groups out of the live + /// subscription set (spec/protocol-core/group-state.md: a non-member group + /// is excluded from live processing). + pub fn account_group_ids_with_removed_membership(&self) -> StorageResult> { + let conn = self.lock()?; + let mut statement = conn + .prepare( + "SELECT group_id_hex + FROM account_groups + WHERE self_membership = 'removed' + ORDER BY group_id_hex", + ) + .storage()?; + let ids = statement + .query_map([], |row| row.get::<_, String>(0)) + .storage()? + .collect::, _>>() + .storage()?; + Ok(ids) + } + /// `group_id_hex` of every `account_groups` row whose `self_membership` is /// still the migration default `'member'`. Used by the one-time /// open/upgrade backfill to decide which legacy rows need their membership diff --git a/crates/storage-sqlite/src/account_projection/tests.rs b/crates/storage-sqlite/src/account_projection/tests.rs index 3b99fabc..13ca04df 100644 --- a/crates/storage-sqlite/src/account_projection/tests.rs +++ b/crates/storage-sqlite/src/account_projection/tests.rs @@ -1456,3 +1456,43 @@ fn app_messages_replay_order_matches_cursor_comparator() { assert_eq!(dups[0].group_id_hex, "aa"); assert_eq!(dups[1].group_id_hex, "bb"); } + +#[test] +fn removed_membership_group_ids_reflect_self_membership_flag() { + let store = SqliteAccountStorage::in_memory().unwrap(); + let state = StoredAccountState { + label: "alice".to_owned(), + seen_events: Vec::new(), + last_transport_timestamp: None, + groups: vec![group("aa", "alpha"), group("bb", "beta")], + }; + store.save_account_projection_state(&state, 16).unwrap(); + + assert!( + store + .account_group_ids_with_removed_membership() + .unwrap() + .is_empty(), + "fresh rows default to member" + ); + + store + .set_group_self_membership("aa", SelfMembership::Removed) + .unwrap(); + assert_eq!( + store.account_group_ids_with_removed_membership().unwrap(), + vec!["aa".to_owned()], + "only the flipped row reports removed membership" + ); + + // A rejoin flips it back and the group leaves the removed set. + store + .set_group_self_membership("aa", SelfMembership::Member) + .unwrap(); + assert!( + store + .account_group_ids_with_removed_membership() + .unwrap() + .is_empty() + ); +} diff --git a/crates/storage-sqlite/src/storage/test_support.rs b/crates/storage-sqlite/src/storage/test_support.rs index d08bad5e..20def93d 100644 --- a/crates/storage-sqlite/src/storage/test_support.rs +++ b/crates/storage-sqlite/src/storage/test_support.rs @@ -38,6 +38,8 @@ pub(crate) fn sample_group(id: GroupId, epoch: u64, members: usize) -> Group { }) .collect(), required_capabilities: GroupCapabilities::default(), + participation: Default::default(), + membership_intervals: Vec::new(), } } diff --git a/crates/traits/src/engine.rs b/crates/traits/src/engine.rs index 0a0401d5..17b51f69 100644 --- a/crates/traits/src/engine.rs +++ b/crates/traits/src/engine.rs @@ -21,7 +21,7 @@ use crate::app_components::{AppComponentData, AppComponentId}; use crate::capabilities::{Feature, FeatureStatus, GroupCapabilities}; use crate::engine_state::PendingStateRef; use crate::error::EngineError; -use crate::group::Member; +use crate::group::{GroupParticipation, Member, QuarantineReason}; use crate::group_context::{GroupContext, SecretBytes}; use crate::ingest::IngestOutcome; use crate::transport::TransportMessage; @@ -372,6 +372,17 @@ pub enum GroupEvent { group_id: GroupId, reason: GroupHydrationQuarantineReason, }, + /// The local identity's participation in the group changed (see + /// `spec/protocol-core/group-state.md`, "Participation"): an applied + /// removal commit resolved to `Left`/`Evicted`, a quarantine hold was + /// placed, or a verified rejoin restored `Member`. Distinct from + /// [`GroupEvent::GroupStateChanged`]: this is local-identity lifecycle + /// state the application gates surfaces on (composer, group list), not a + /// group system row. + ParticipationChanged { + group_id: GroupId, + participation: GroupParticipation, + }, EpochChanged { group_id: GroupId, from: EpochId, @@ -674,6 +685,51 @@ pub trait CgkaEngine: Send + Sync { fn members(&self, group_id: &GroupId) -> Result, EngineError>; + /// Wrap a removal notice for `recipient` embedding the already-published + /// commit (spec/protocol-core/member-departure.md, "Removal notices"). + /// Delegates to the active transport peeler; `Ok(None)` means the binding + /// does not carry removal notices (the default). + async fn wrap_removal_notice( + &self, + commit: &TransportMessage, + recipient: &MemberId, + ) -> Result, EngineError> { + let _ = (commit, recipient); + Ok(None) + } + + /// The local identity's participation in the group. `Ok(None)` means the + /// engine has no durable record of the group at all ("no such group") — + /// distinct from every participation state, per + /// `spec/protocol-core/group-state.md`, "Participation and public + /// surfaces". + fn participation(&self, group_id: &GroupId) -> Result, EngineError>; + + /// Withhold the group (`spec/protocol-core/group-state.md`, "Quarantine"): + /// participation moves to `Quarantined { reason }`, live inbound is + /// retained-but-unprocessed, and live convergence skips the group until + /// [`CgkaEngine::resolve_group_quarantine`] runs. A group already + /// authoritatively non-member (`Left`/`Evicted`) is never downgraded to a + /// hold; quarantining it is a no-op. + async fn quarantine_group( + &mut self, + group_id: &GroupId, + reason: QuarantineReason, + ) -> Result<(), EngineError>; + + /// The explicit recovery transition out of `Quarantined`: run one + /// deliberate convergence pass over the withheld/stored input, let the + /// ordinary apply seams decide the outcome (an applied removal commit + /// resolves to `Left`/`Evicted`), and restore `Member` only when the live + /// MLS state is active and the canonical roster still contains the local + /// identity. Returns the resulting participation; a group that cannot be + /// safely resolved stays `Quarantined`. Ordinary inbound never triggers + /// this — quarantine cannot be silently re-activated. + async fn resolve_group_quarantine( + &mut self, + group_id: &GroupId, + ) -> Result; + fn epoch(&self, group_id: &GroupId) -> Result; /// Stable identity of the local client across every group. diff --git a/crates/traits/src/group.rs b/crates/traits/src/group.rs index 1e86320a..93ba6ca5 100644 --- a/crates/traits/src/group.rs +++ b/crates/traits/src/group.rs @@ -19,6 +19,21 @@ pub struct Group { pub epoch: EpochId, pub members: Vec, pub required_capabilities: GroupCapabilities, + /// The local identity's participation in this group (see + /// `spec/protocol-core/group-state.md`, "Participation"). Lives on the + /// durable record — not the MLS tree — so it survives live-OpenMLS-state + /// teardown after a removal, and so fork-recovery snapshot rollback + /// restores it together with the rest of the group state (a rolled-back + /// removal commit must also roll back the non-member transition). + /// Defaults to `Member` for records written before this field existed. + #[serde(default)] + pub participation: GroupParticipation, + /// Epoch intervals during which the local identity was a member (see + /// [`MembershipInterval`]). Maintained by the participation transitions; + /// empty means "no retained history" for legacy records, which fails open + /// in [`membership_intervals_contain`]. + #[serde(default)] + pub membership_intervals: Vec, } /// One member of a group, as storage sees it. @@ -30,3 +45,122 @@ pub struct Member { pub id: MemberId, pub credential: Vec, } + +/// One epoch interval during which the local identity was a member of a +/// group. Because a group may be left/removed and later rejoined, membership +/// is a set of intervals, not a single boundary +/// (spec/foundation/errors.md, `PreMembership`). `ended_at` is `None` for the +/// currently-open interval and holds the last epoch in which the local +/// identity held membership keys: the source epoch immediately before the +/// removing commit's resulting epoch. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct MembershipInterval { + pub joined_at: EpochId, + pub ended_at: Option, +} + +/// Whether `epoch` falls inside any retained membership interval. +/// +/// An EMPTY interval set means "no retained membership history" (legacy +/// records) and fails open: without history a client MUST NOT classify input +/// as `PreMembership` (spec/foundation/errors.md scopes the outcome to groups +/// the client has membership history for). +pub fn membership_intervals_contain(intervals: &[MembershipInterval], epoch: EpochId) -> bool { + if intervals.is_empty() { + return true; + } + intervals.iter().any(|interval| { + epoch >= interval.joined_at && interval.ended_at.is_none_or(|ended| epoch <= ended) + }) +} + +/// The local identity's participation in a group — a dimension orthogonal to +/// the convergence lifecycle (`Stable`/`Recovering`/`Unrecoverable`/…). +/// +/// This is the shared vocabulary for the group participation states defined in +/// `spec/protocol-core/group-state.md`. Ingest, convergence, and public group +/// accessors map to it so a caller can tell a live group from one this identity +/// has been evicted from, or one being withheld pending recovery. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub enum GroupParticipation { + /// The local identity is present in the group's canonical roster; the only + /// state in which local commits or delivered app payloads are allowed. + #[default] + Member, + /// The local identity voluntarily departed (its SelfRemove was committed). + /// Non-member; the group is inactive for this identity. Kept distinct from + /// [`GroupParticipation::Evicted`] so a surface can tell "you left" from + /// "you were removed". + Left, + /// The local identity was removed by another member. Non-member; the group + /// is inactive for this identity. Reached only by applying the removal + /// commit — delivered in order, recovered through the transport's + /// missed-input recovery, or carried by a removal notice (see + /// `spec/protocol-core/group-state.md`, "Reaching a non-member state") — + /// never asserted from an undecryptable message or an MLS error. + Evicted, + /// The group is excluded from the live group set pending an explicit + /// recovery transition; neither trusted as a live member group nor asserted + /// non-member. Carries why it is withheld, because the two holds have + /// opposite expected exits (see [`QuarantineReason`]). + Quarantined { reason: QuarantineReason }, +} + +/// Why a group is held in [`GroupParticipation::Quarantined`]. The reason +/// determines the expected exit: `PendingMembership` resolves through the +/// removal commit (or recovered group-evolution input), `IntegrityHold` +/// through a verified repair path. Mirrors the quarantine reasons in +/// `spec/protocol-core/group-state.md`, "Quarantine". +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum QuarantineReason { + /// Undecryptable group traffic suggests the local identity may have been + /// removed, and the discovery probes (transport missed-input recovery, + /// removal notice) have not yet recovered the removal commit. + PendingMembership, + /// Stored group material failed to load or validate, or a durable + /// invariant check failed. + IntegrityHold, +} + +#[cfg(test)] +mod membership_interval_tests { + use super::{MembershipInterval, membership_intervals_contain}; + use crate::types::EpochId; + + #[test] + fn empty_history_fails_open() { + // No retained history: PreMembership MUST NOT be classified + // (errors.md scopes it to groups with membership history). + assert!(membership_intervals_contain(&[], EpochId(0))); + assert!(membership_intervals_contain(&[], EpochId(999))); + } + + #[test] + fn intervals_cover_joins_gaps_and_rejoins() { + // Member epochs 1..=1, removed, rejoined at 3 (still open). + let intervals = [ + MembershipInterval { + joined_at: EpochId(1), + ended_at: Some(EpochId(1)), + }, + MembershipInterval { + joined_at: EpochId(3), + ended_at: None, + }, + ]; + assert!( + !membership_intervals_contain(&intervals, EpochId(0)), + "pre-join epochs are outside" + ); + assert!(membership_intervals_contain(&intervals, EpochId(1))); + assert!( + !membership_intervals_contain(&intervals, EpochId(2)), + "the removed gap is outside" + ); + assert!(membership_intervals_contain(&intervals, EpochId(3))); + assert!( + membership_intervals_contain(&intervals, EpochId(50)), + "an open interval extends forward" + ); + } +} diff --git a/crates/traits/src/ingest.rs b/crates/traits/src/ingest.rs index 40b7e001..90f91327 100644 --- a/crates/traits/src/ingest.rs +++ b/crates/traits/src/ingest.rs @@ -46,6 +46,25 @@ pub enum StaleReason { /// retryable depending on whether the engine has evidence that another /// epoch context could later peel it. PeelFailed, + /// The local identity is no longer a member of this group (participation + /// `Left` or `Evicted`, or the live MLS state is inactive from a merged + /// removal): further inbound can no longer affect the group and is stale + /// with the `evicted` category (spec/foundation/errors.md). Reaching the + /// non-member state itself is a participation transition, never derived + /// from this classification. + Evicted, + /// The input's source epoch falls outside every retained interval during + /// which the local identity was a member — terminal by design: this + /// client can never hold the keys (spec/foundation/errors.md, + /// `PreMembership` -> `pre_membership`). Never used for groups without + /// retained membership history, and never retried. + PreMembership, + /// The group is withheld (`GroupParticipation::Quarantined`): excluded + /// from live inbound processing pending an explicit recovery transition + /// (spec/protocol-core/group-state.md, "Quarantine"). The input is + /// retained (`Retryable`) so the resolution pass can still consume it — + /// withheld, not discarded. + Quarantined, } /// Decrypted inbound message ready for engine processing. @@ -70,4 +89,11 @@ pub enum PeeledContent { MlsMessage { bytes: Vec }, /// Welcome payload (MLS welcome bytes). Welcome { bytes: Vec }, + /// A removal notice: an inbox-delivered carrier for a group message the + /// removed member may have missed — most importantly its own removal + /// commit (spec/protocol-core/member-departure.md, "Removal notices"). + /// The notice has no authority of its own: the engine re-injects the + /// embedded, transport-validated group message into the ordinary inbound + /// pipeline, and only an applied removal commit changes participation. + RemovalNotice { embedded: TransportMessage }, } diff --git a/crates/traits/src/lib.rs b/crates/traits/src/lib.rs index 4d9dd9c3..5b7053c3 100644 --- a/crates/traits/src/lib.rs +++ b/crates/traits/src/lib.rs @@ -74,7 +74,10 @@ pub use engine_state::{ Recovering, StagedCommitHandle, WelcomeState, }; pub use error::{EngineError, PeelerError}; -pub use group::{Group, Member}; +pub use group::{ + Group, GroupParticipation, Member, MembershipInterval, QuarantineReason, + membership_intervals_contain, +}; pub use group_context::{GroupContext, GroupContextSnapshot, SecretBytes}; pub use ingest::{IngestOutcome, PeeledContent, PeeledMessage, StaleReason}; pub use message::{MessageRecord, MessageState, StoredMessagePayload}; @@ -90,8 +93,9 @@ pub use transport::{ pub use transport_adapter::{ TransportAccountActivation, TransportAdapter, TransportAdapterError, TransportDelivery, TransportDeliveryPlane, TransportDeliverySource, TransportEndpoint, TransportEndpointFailure, - TransportEndpointReceipt, TransportGroupSubscription, TransportGroupSync, - TransportPublishReport, TransportPublishRequest, TransportPublishTarget, TransportWireMetadata, + TransportEndpointReceipt, TransportGroupBackfill, TransportGroupSubscription, + TransportGroupSync, TransportPublishReport, TransportPublishRequest, TransportPublishTarget, + TransportWireMetadata, }; pub use types::{Backend, EpochId, GroupId, MemberId, MessageId}; pub use welcome::PendingWelcome; diff --git a/crates/traits/src/peeler.rs b/crates/traits/src/peeler.rs index 01bc7c54..9087f55b 100644 --- a/crates/traits/src/peeler.rs +++ b/crates/traits/src/peeler.rs @@ -151,4 +151,18 @@ pub trait TransportPeeler: Send + Sync { ) -> Result { self.wrap_welcome(payload, recipient).await } + + /// Wrap a removal notice for `recipient` embedding the already-published + /// group message `commit` (spec/protocol-core/member-departure.md, + /// "Removal notices"; the transport binding defines the shape). + /// `Ok(None)` means this binding does not carry removal notices — the + /// default, so bindings (and test peelers) opt in explicitly. + async fn wrap_removal_notice( + &self, + commit: &TransportMessage, + recipient: &MemberId, + ) -> Result, PeelerError> { + let _ = (commit, recipient); + Ok(None) + } } diff --git a/crates/traits/src/transport_adapter.rs b/crates/traits/src/transport_adapter.rs index 53a1654b..8677c68e 100644 --- a/crates/traits/src/transport_adapter.rs +++ b/crates/traits/src/transport_adapter.rs @@ -84,6 +84,21 @@ pub struct TransportGroupSync { pub since: Option, } +/// A bounded re-fetch of one group's message stream — the transport +/// realization of the missed-input recovery probe required by +/// `spec/protocol-core/group-state.md` ("Reaching a non-member state"). +/// Re-issues the group's subscription with `since` widened to the caller's +/// anchor (last input successfully consumed, minus a local slack) so a missed +/// group-evolution commit — most importantly a missed removal commit — is +/// recovered from replayable history. Recovered events flow through ordinary +/// validation and deduplication; the probe never chooses group state. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct TransportGroupBackfill { + pub account_id: MemberId, + pub group_subscription: TransportGroupSubscription, + pub since: Option, +} + /// Publish target for an already-wrapped transport message. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum TransportPublishTarget { @@ -286,6 +301,16 @@ pub trait TransportAdapter: Send + Sync { sync: TransportGroupSync, ) -> Result<(), TransportAdapterError>; + /// Re-issue one group's subscription with a widened `since` window — the + /// membership recovery probe (`TransportGroupBackfill`). Implementations + /// reuse the group's existing subscription identity so the probe replaces + /// (never duplicates) the live subscription and the relay replays stored + /// events from the widened window. + async fn backfill_account_group( + &self, + backfill: TransportGroupBackfill, + ) -> Result<(), TransportAdapterError>; + /// Deactivate every subscription owned by an account. async fn deactivate_account(&self, account_id: &MemberId) -> Result<(), TransportAdapterError>; diff --git a/crates/traits/tests/snapshots.rs b/crates/traits/tests/snapshots.rs index ffffd1ff..274f2984 100644 --- a/crates/traits/tests/snapshots.rs +++ b/crates/traits/tests/snapshots.rs @@ -18,7 +18,7 @@ use cgka_traits::engine::{ SendResult, }; use cgka_traits::engine_state::PendingStateRef; -use cgka_traits::group::{Group, Member}; +use cgka_traits::group::{Group, GroupParticipation, Member, QuarantineReason}; use cgka_traits::ingest::{IngestOutcome, PeeledContent, PeeledMessage, StaleReason}; use cgka_traits::message::StoredMessagePayload; use cgka_traits::transport::{ @@ -347,6 +347,24 @@ fn snapshot_ingest_outcomes() { reason: StaleReason::PeelFailed } ); + insta::assert_json_snapshot!( + "stale_evicted", + IngestOutcome::Stale { + reason: StaleReason::Evicted + } + ); + insta::assert_json_snapshot!( + "stale_quarantined", + IngestOutcome::Stale { + reason: StaleReason::Quarantined + } + ); + insta::assert_json_snapshot!( + "stale_pre_membership", + IngestOutcome::Stale { + reason: StaleReason::PreMembership + } + ); } #[test] @@ -609,6 +627,42 @@ fn snapshot_group_and_member() { credential: vec![], }], required_capabilities: GroupCapabilities::default(), + participation: GroupParticipation::Member, + membership_intervals: vec![], + } + ); + // Records persisted before the participation field existed must load as + // `Member` (spec: participation defaults to live membership for legacy + // records; the serialized form is self-describing JSON in storage). + let legacy = serde_json::json!({ + "id": gid(), + "name": "ops", + "description": "for ops talk", + "epoch": 3, + "members": [], + "required_capabilities": GroupCapabilities::default(), + }); + let decoded: Group = serde_json::from_value(legacy).expect("legacy record decodes"); + assert_eq!(decoded.participation, GroupParticipation::Member); +} + +#[test] +fn snapshot_group_participation() { + // Locks the serialized variant casing: this enum crosses the FFI boundary, + // so its wire shape must not drift silently. + insta::assert_json_snapshot!("participation_member", GroupParticipation::Member); + insta::assert_json_snapshot!("participation_left", GroupParticipation::Left); + insta::assert_json_snapshot!("participation_evicted", GroupParticipation::Evicted); + insta::assert_json_snapshot!( + "participation_quarantined_pending_membership", + GroupParticipation::Quarantined { + reason: QuarantineReason::PendingMembership + } + ); + insta::assert_json_snapshot!( + "participation_quarantined_integrity_hold", + GroupParticipation::Quarantined { + reason: QuarantineReason::IntegrityHold } ); } diff --git a/crates/traits/tests/snapshots/snapshots__group.snap b/crates/traits/tests/snapshots/snapshots__group.snap index f49c63f2..8b70a49e 100644 --- a/crates/traits/tests/snapshots/snapshots__group.snap +++ b/crates/traits/tests/snapshots/snapshots__group.snap @@ -1,7 +1,6 @@ --- source: crates/traits/tests/snapshots.rs -assertion_line: 510 -expression: "Group\n{\n id: gid(), name: \"ops\".into(), description: \"for ops talk\".into(), epoch:\n EpochId(3), members: vec![Member { id: mem_id(), credential: vec![], }],\n required_capabilities: GroupCapabilities::default(),\n}" +expression: "Group\n{\n id: gid(), name: \"ops\".into(), description: \"for ops talk\".into(), epoch:\n EpochId(3), members: vec![Member { id: mem_id(), credential: vec![], }],\n required_capabilities: GroupCapabilities::default(), participation:\n GroupParticipation::Member, membership_intervals: vec![],\n}" --- { "id": [ @@ -31,5 +30,7 @@ expression: "Group\n{\n id: gid(), name: \"ops\".into(), description: \"for o "app_components": { "ids": [] } - } + }, + "participation": "Member", + "membership_intervals": [] } diff --git a/crates/traits/tests/snapshots/snapshots__participation_evicted.snap b/crates/traits/tests/snapshots/snapshots__participation_evicted.snap new file mode 100644 index 00000000..4f3db998 --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__participation_evicted.snap @@ -0,0 +1,5 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "GroupParticipation::Evicted" +--- +"Evicted" diff --git a/crates/traits/tests/snapshots/snapshots__participation_left.snap b/crates/traits/tests/snapshots/snapshots__participation_left.snap new file mode 100644 index 00000000..94ae20d0 --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__participation_left.snap @@ -0,0 +1,5 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "GroupParticipation::Left" +--- +"Left" diff --git a/crates/traits/tests/snapshots/snapshots__participation_member.snap b/crates/traits/tests/snapshots/snapshots__participation_member.snap new file mode 100644 index 00000000..19647885 --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__participation_member.snap @@ -0,0 +1,5 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "GroupParticipation::Member" +--- +"Member" diff --git a/crates/traits/tests/snapshots/snapshots__participation_quarantined_integrity_hold.snap b/crates/traits/tests/snapshots/snapshots__participation_quarantined_integrity_hold.snap new file mode 100644 index 00000000..a250b1b5 --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__participation_quarantined_integrity_hold.snap @@ -0,0 +1,9 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "GroupParticipation::Quarantined { reason: QuarantineReason::IntegrityHold }" +--- +{ + "Quarantined": { + "reason": "IntegrityHold" + } +} diff --git a/crates/traits/tests/snapshots/snapshots__participation_quarantined_pending_membership.snap b/crates/traits/tests/snapshots/snapshots__participation_quarantined_pending_membership.snap new file mode 100644 index 00000000..448b621a --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__participation_quarantined_pending_membership.snap @@ -0,0 +1,9 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "GroupParticipation::Quarantined\n{ reason: QuarantineReason::PendingMembership }" +--- +{ + "Quarantined": { + "reason": "PendingMembership" + } +} diff --git a/crates/traits/tests/snapshots/snapshots__stale_evicted.snap b/crates/traits/tests/snapshots/snapshots__stale_evicted.snap new file mode 100644 index 00000000..f6e50e52 --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__stale_evicted.snap @@ -0,0 +1,9 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "IngestOutcome::Stale { reason: StaleReason::Evicted }" +--- +{ + "Stale": { + "reason": "Evicted" + } +} diff --git a/crates/traits/tests/snapshots/snapshots__stale_pre_membership.snap b/crates/traits/tests/snapshots/snapshots__stale_pre_membership.snap new file mode 100644 index 00000000..2762c6d5 --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__stale_pre_membership.snap @@ -0,0 +1,9 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "IngestOutcome::Stale { reason: StaleReason::PreMembership }" +--- +{ + "Stale": { + "reason": "PreMembership" + } +} diff --git a/crates/traits/tests/snapshots/snapshots__stale_quarantined.snap b/crates/traits/tests/snapshots/snapshots__stale_quarantined.snap new file mode 100644 index 00000000..9922798b --- /dev/null +++ b/crates/traits/tests/snapshots/snapshots__stale_quarantined.snap @@ -0,0 +1,9 @@ +--- +source: crates/traits/tests/snapshots.rs +expression: "IngestOutcome::Stale { reason: StaleReason::Quarantined }" +--- +{ + "Stale": { + "reason": "Quarantined" + } +} diff --git a/crates/transport-nostr-adapter/src/lib.rs b/crates/transport-nostr-adapter/src/lib.rs index 30e7d1c8..b05576af 100644 --- a/crates/transport-nostr-adapter/src/lib.rs +++ b/crates/transport-nostr-adapter/src/lib.rs @@ -19,8 +19,9 @@ use cgka_traits::transport::{Timestamp, TransportEnvelope, TransportMessage, Tra use cgka_traits::{ GroupId, MemberId, TransportAccountActivation, TransportAdapter, TransportAdapterError, TransportDelivery, TransportDeliveryPlane, TransportDeliverySource, TransportEndpoint, - TransportEndpointFailure, TransportEndpointReceipt, TransportGroupSubscription, - TransportGroupSync, TransportPublishReport, TransportPublishRequest, TransportWireMetadata, + TransportEndpointFailure, TransportEndpointReceipt, TransportGroupBackfill, + TransportGroupSubscription, TransportGroupSync, TransportPublishReport, + TransportPublishRequest, TransportWireMetadata, }; use nostr::RelayUrl; use sha2::{Digest, Sha256}; @@ -568,6 +569,41 @@ impl TransportAdapter for NostrTransportAdapter { Ok(()) } + async fn backfill_account_group( + &self, + backfill: TransportGroupBackfill, + ) -> Result<(), TransportAdapterError> { + tracing::debug!( + target: "transport_nostr_adapter::adapter", + method = "backfill_account_group", + has_since = backfill.since.is_some(), + "re-issuing group subscription for membership backfill probe" + ); + // Resolve the account's CURRENT route for this group and re-issue that + // subscription with the widened `since`. Using the stored route (not + // the caller-supplied endpoints) keeps the probe pinned to the signed + // routing state, and reusing the subscription identity means the relay + // replaces the live subscription and replays stored events from the + // widened window — no duplicate subscription to leak or diff away. + let subscription = { + let state = self.state.read().await; + let routes = state.accounts.get(&backfill.account_id).ok_or_else(|| { + TransportAdapterError::AccountNotActive(backfill.account_id.clone()) + })?; + let group = routes + .groups + .iter() + .find(|group| group.group_id == backfill.group_subscription.group_id) + .ok_or_else(|| { + TransportAdapterError::Subscription( + "backfill target group is not subscribed for this account".to_string(), + ) + })?; + group_subscription(&backfill.account_id, group, backfill.since) + }; + self.relay_client.subscribe(subscription).await + } + async fn deactivate_account(&self, account_id: &MemberId) -> Result<(), TransportAdapterError> { let removed_count = { let state = self.state.read().await; diff --git a/crates/transport-nostr-adapter/tests/inbound_routing.rs b/crates/transport-nostr-adapter/tests/inbound_routing.rs index 91e20098..8fe27de0 100644 --- a/crates/transport-nostr-adapter/tests/inbound_routing.rs +++ b/crates/transport-nostr-adapter/tests/inbound_routing.rs @@ -1082,3 +1082,86 @@ fn group_event(id_byte: &str, transport_group_id: &[u8]) -> NostrTransportEvent sig: None, } } + +#[tokio::test] +async fn backfill_reissues_the_stored_route_with_the_probe_window() { + let relay = Arc::new(FakeRelayClient::default()); + let adapter = NostrTransportAdapter::new(relay.clone()); + let alice = MemberId::new(vec![0xA1; 32]); + let group_id = cgka_traits::GroupId::new(vec![0xC3; 32]); + let subscription = TransportGroupSubscription { + group_id: group_id.clone(), + transport_group_id: vec![0xD4; 32], + endpoints: vec![TransportEndpoint("wss://group.example".into())], + }; + adapter + .activate_account(TransportAccountActivation { + account_id: alice.clone(), + inbox_endpoints: vec![TransportEndpoint("wss://alice-inbox.example".into())], + group_subscriptions: vec![subscription.clone()], + since: None, + }) + .await + .expect("activation succeeds"); + let live_count = relay.subscriptions.lock().unwrap().len(); + + // The probe re-issues the group's subscription with the widened window. + // Caller-supplied endpoints are ignored in favor of the STORED route, so + // a stale caller cannot point the probe at rogue relays. + adapter + .backfill_account_group(cgka_traits::TransportGroupBackfill { + account_id: alice.clone(), + group_subscription: TransportGroupSubscription { + endpoints: vec![TransportEndpoint("wss://rogue.example".into())], + ..subscription.clone() + }, + since: Some(cgka_traits::Timestamp(1_600_000_000)), + }) + .await + .expect("backfill succeeds"); + + { + let subs = relay.subscriptions.lock().unwrap(); + let probe = subs.last().expect("probe subscription issued"); + assert_eq!(subs.len(), live_count + 1); + match probe { + transport_nostr_adapter::NostrSubscription::Group { + endpoints, since, .. + } => { + assert_eq!( + endpoints, + &vec![TransportEndpoint("wss://group.example".into())], + "probe must use the stored route, not caller endpoints" + ); + assert_eq!(*since, Some(cgka_traits::Timestamp(1_600_000_000))); + } + other => panic!("expected a group subscription, got {other:?}"), + } + // Same subscription identity as the live subscription: the relay + // replaces the live filter and replays, rather than stacking a + // duplicate. + assert_eq!( + probe.subscription_id(), + subs[live_count - 1].subscription_id(), + "probe reuses the live subscription id" + ); + } + + // Unknown group: typed rejection, no subscription issued. + let err = adapter + .backfill_account_group(cgka_traits::TransportGroupBackfill { + account_id: alice, + group_subscription: TransportGroupSubscription { + group_id: cgka_traits::GroupId::new(vec![0xEE; 32]), + transport_group_id: vec![0xEE; 32], + endpoints: vec![TransportEndpoint("wss://group.example".into())], + }, + since: None, + }) + .await + .unwrap_err(); + assert!(matches!( + err, + cgka_traits::TransportAdapterError::Subscription(_) + )); +} diff --git a/crates/transport-nostr-peeler/src/lib.rs b/crates/transport-nostr-peeler/src/lib.rs index 6815f87e..93aabf95 100644 --- a/crates/transport-nostr-peeler/src/lib.rs +++ b/crates/transport-nostr-peeler/src/lib.rs @@ -29,6 +29,10 @@ pub const KIND_NIP59_GIFT_WRAP: u64 = 1059; /// Marmot welcome rumor kind inside the NIP-59 seal. pub const KIND_MARMOT_WELCOME_RUMOR: u16 = 444; +/// Marmot removal notice rumor kind inside a NIP-59 gift wrap +/// (spec/transports/nostr.md, "Removal notice delivery"). +pub const KIND_MARMOT_REMOVAL_NOTICE_RUMOR: u16 = 451; + /// Source label carried by [`cgka_traits::transport::TransportMessage`] values /// produced here. pub const NOSTR_SOURCE: &str = "nostr"; diff --git a/crates/transport-nostr-peeler/src/peeler.rs b/crates/transport-nostr-peeler/src/peeler.rs index d3716298..0b121622 100644 --- a/crates/transport-nostr-peeler/src/peeler.rs +++ b/crates/transport-nostr-peeler/src/peeler.rs @@ -1,9 +1,9 @@ use crate::error::to_peeler_error; use crate::event::{decode_hex, decode_hex_exact}; use crate::{ - DEFAULT_EXPORTER_LABEL, GROUP_TAG, KIND_MARMOT_GROUP_MESSAGE, KIND_MARMOT_WELCOME_RUMOR, - KIND_NIP59_GIFT_WRAP, NOSTR_GROUP_CONTENT_MIN_LEN, NOSTR_GROUP_KEY_LEN, NostrTransportEvent, - RECIPIENT_TAG, + DEFAULT_EXPORTER_LABEL, GROUP_TAG, KIND_MARMOT_GROUP_MESSAGE, KIND_MARMOT_REMOVAL_NOTICE_RUMOR, + KIND_MARMOT_WELCOME_RUMOR, KIND_NIP59_GIFT_WRAP, NOSTR_GROUP_CONTENT_MIN_LEN, + NOSTR_GROUP_KEY_LEN, NostrTransportEvent, RECIPIENT_TAG, }; use async_trait::async_trait; use cgka_traits::engine::WelcomeMetadata; @@ -26,6 +26,7 @@ const WELCOME_SIGNER_CONTEXT: &str = "nostr_welcome_signer"; const KEY_PACKAGE_EVENT_TAG: &str = "e"; const EXPIRATION_TAG: &str = "expiration"; const WELCOME_RELAYS_TAG: &str = "relays"; +const REMOVAL_NOTICE_EVENT_TAG: &str = "e"; /// Empty AAD for the outer kind-445 ChaCha20-Poly1305 sealing /// (`spec/transports/nostr.md`: `aad = ""`). @@ -251,6 +252,9 @@ impl TransportPeeler for NostrMlsPeeler { .await .map_err(map_nip59_error)?; + if unwrapped.rumor.kind == Kind::Custom(KIND_MARMOT_REMOVAL_NOTICE_RUMOR) { + return peel_removal_notice_rumor(msg, &unwrapped); + } if unwrapped.rumor.kind != Kind::Custom(KIND_MARMOT_WELCOME_RUMOR) { return Err(PeelerError::Malformed(format!( "expected Marmot welcome rumor kind {KIND_MARMOT_WELCOME_RUMOR}, got {}", @@ -370,6 +374,137 @@ impl TransportPeeler for NostrMlsPeeler { let event = NostrTransportEvent::from_nostr_event(&gift_wrap).map_err(to_peeler_error)?; event.to_transport_message().map_err(to_peeler_error) } + + async fn wrap_removal_notice( + &self, + commit: &TransportMessage, + recipient: &MemberId, + ) -> Result, PeelerError> { + // spec/transports/nostr.md, "Removal notice delivery": an unsigned + // kind-451 rumor embedding the published kind-445 removal-commit event + // (stringified-event convention), gift-wrapped to the removed member. + // Embedding removes the fetch round-trip and the relay-retention + // dependency; the receiver validates the embedded event exactly like a + // fetched one, so the notice never carries authority of its own. + let event = NostrTransportEvent::from_transport_message(commit).map_err(to_peeler_error)?; + if event.kind != KIND_MARMOT_GROUP_MESSAGE { + return Err(PeelerError::WrapFailed(format!( + "removal notice must embed a kind {KIND_MARMOT_GROUP_MESSAGE} group message, got {}", + event.kind + ))); + } + if event.sig.is_none() { + return Err(PeelerError::WrapFailed( + "removal notice must embed the signed, published group message event".into(), + )); + } + // Reuse the 445 envelope validation for the h value. + let transport_group_id = match event.to_transport_message().map_err(to_peeler_error)? { + TransportMessage { + envelope: TransportEnvelope::GroupMessage { transport_group_id }, + .. + } => transport_group_id, + _ => { + return Err(PeelerError::WrapFailed( + "embedded event did not map to a group message envelope".into(), + )); + } + }; + let signer = self.welcome_signer()?; + let sender_pubkey = signer + .get_public_key() + .await + .map_err(|e| PeelerError::WrapFailed(format!("signer public key: {e}")))?; + let recipient_pubkey = Self::recipient_pubkey(recipient)?; + let content = serde_json::to_string(&event) + .map_err(|e| PeelerError::WrapFailed(format!("embedded event JSON: {e}")))?; + let rumor: UnsignedEvent = + EventBuilder::new(Kind::Custom(KIND_MARMOT_REMOVAL_NOTICE_RUMOR), content) + .tags([ + Tag::custom( + nostr::TagKind::custom(GROUP_TAG), + [hex::encode(&transport_group_id)], + ), + Tag::custom( + nostr::TagKind::custom(REMOVAL_NOTICE_EVENT_TAG), + [event.id.clone()], + ), + ]) + .build(sender_pubkey); + let gift_wrap = EventBuilder::gift_wrap(signer, &recipient_pubkey, rumor, []) + .await + .map_err(|e| PeelerError::WrapFailed(format!("NIP-59 gift wrap: {e}")))?; + let wrapped = NostrTransportEvent::from_nostr_event(&gift_wrap).map_err(to_peeler_error)?; + Ok(Some( + wrapped.to_transport_message().map_err(to_peeler_error)?, + )) + } +} + +/// Peel an unwrapped kind-451 removal notice rumor +/// (spec/transports/nostr.md, "Removal notice delivery"). The embedded +/// kind-445 event MUST pass the same validation as a fetched one; the caller +/// re-injects the returned transport message into the ordinary inbound +/// pipeline, so the notice itself carries no authority. +fn peel_removal_notice_rumor( + msg: &TransportMessage, + unwrapped: &nostr::nips::nip59::UnwrappedGift, +) -> Result { + let rumor_h = rumor_tag_value(&unwrapped.rumor, GROUP_TAG) + .ok_or_else(|| PeelerError::Malformed("removal notice rumor is missing h tag".into()))? + .to_owned(); + let rumor_event_id = rumor_tag_value(&unwrapped.rumor, REMOVAL_NOTICE_EVENT_TAG) + .ok_or_else(|| PeelerError::Malformed("removal notice rumor is missing e tag".into()))? + .to_owned(); + decode_hex_exact("removal notice e tag", &rumor_event_id, 32).map_err(to_peeler_error)?; + + // Embedded event: parse, verify id + signature, and run the kind-445 + // envelope validation (32-byte id, exactly one h tag) via the shared DTO. + let embedded_event = + ::from_json(unwrapped.rumor.content.as_bytes()) + .map_err(|e| PeelerError::Malformed(format!("embedded event parse: {e}")))?; + embedded_event + .verify() + .map_err(|e| PeelerError::Malformed(format!("embedded event verification: {e}")))?; + let dto = NostrTransportEvent::from_nostr_event(&embedded_event).map_err(to_peeler_error)?; + if dto.kind != KIND_MARMOT_GROUP_MESSAGE { + return Err(PeelerError::Malformed(format!( + "removal notice must embed a kind {KIND_MARMOT_GROUP_MESSAGE} event, got {}", + dto.kind + ))); + } + if dto.id != rumor_event_id { + return Err(PeelerError::Malformed( + "removal notice e tag does not match the embedded event id".into(), + )); + } + let decoded_len = BASE64_STANDARD + .decode(dto.content.as_bytes()) + .map_err(|e| PeelerError::Malformed(format!("embedded event content base64: {e}")))? + .len(); + if decoded_len < NOSTR_GROUP_CONTENT_MIN_LEN { + return Err(PeelerError::Malformed( + "embedded event content shorter than nonce + AEAD tag".into(), + )); + } + let embedded = dto.to_transport_message().map_err(to_peeler_error)?; + match &embedded.envelope { + TransportEnvelope::GroupMessage { transport_group_id } + if hex::encode(transport_group_id) == rumor_h => {} + _ => { + return Err(PeelerError::Malformed( + "removal notice h tag does not match the embedded event's group".into(), + )); + } + } + + Ok(PeeledMessage { + id: msg.id.clone(), + group_id: None, + sender: Some(MemberId::new(unwrapped.sender.to_bytes().to_vec())), + content: PeeledContent::RemovalNotice { embedded }, + origin: msg.clone(), + }) } /// First value of a tag on an unwrapped NIP-59 rumor (`tag[0] == name` → @@ -1113,3 +1248,155 @@ mod tests { .unwrap() } } + +#[cfg(test)] +mod removal_notice_tests { + use super::*; + use crate::KIND_MARMOT_REMOVAL_NOTICE_RUMOR; + use cgka_traits::group_context::GroupContextSnapshot; + use cgka_traits::ingest::PeeledContent; + use cgka_traits::types::EpochId; + use std::collections::HashMap; + + fn sender_keys() -> Keys { + Keys::generate() + } + + fn receiver_keys() -> Keys { + Keys::generate() + } + + async fn wrapped_commit(group_id: &[u8]) -> TransportMessage { + // A real signed kind-445 event, exactly what the committer published. + let ctx = GroupContextSnapshot::new( + EpochId(4), + HashMap::from([( + DEFAULT_EXPORTER_LABEL.to_string(), + vec![0x42; NOSTR_GROUP_KEY_LEN], + )]), + Some(group_id.to_vec()), + ); + NostrMlsPeeler::default() + .wrap_group_message( + &EncryptedPayload { + ciphertext: b"removal commit mls bytes".to_vec(), + aad: vec![], + }, + &ctx, + ) + .await + .expect("wrap commit succeeds") + } + + #[tokio::test] + async fn removal_notice_wrap_and_peel_round_trips_the_published_commit() { + let sender = sender_keys(); + let receiver = receiver_keys(); + let recipient = MemberId::new(receiver.public_key().to_bytes().to_vec()); + let sender_peeler = NostrMlsPeeler::new().with_welcome_signer(sender.clone()); + let receiver_peeler = NostrMlsPeeler::new().with_welcome_signer(receiver.clone()); + let group_id = vec![0x5c; 32]; + let commit = wrapped_commit(&group_id).await; + + let notice = sender_peeler + .wrap_removal_notice(&commit, &recipient) + .await + .expect("wrap succeeds") + .expect("nostr binding carries removal notices"); + + // Outer shape: a NIP-59 gift wrap addressed to the removed member; + // relays never see the group id or the embedded commit. + assert!(matches!( + notice.envelope, + TransportEnvelope::Welcome { ref recipient } + if recipient.as_slice() == receiver.public_key().as_bytes() + )); + let outer = NostrTransportEvent::from_transport_message(¬ice).unwrap(); + assert_eq!(outer.kind, KIND_NIP59_GIFT_WRAP); + assert!( + !outer + .tags + .iter() + .any(|tag| tag.first().map(String::as_str) == Some("h")), + "gift wrap must not leak the group id to relays" + ); + + let peeled = receiver_peeler + .peel_welcome(¬ice) + .await + .expect("peel succeeds"); + assert_eq!( + peeled.sender, + Some(MemberId::new(sender.public_key().to_bytes().to_vec())) + ); + let PeeledContent::RemovalNotice { embedded } = peeled.content else { + panic!("expected a removal notice, got {:?}", peeled.content); + }; + // The embedded message is byte-identical to the published commit, so + // it re-enters the inbound pipeline exactly as if fetched from the + // group stream (dedup collapses any overlap). + assert_eq!(embedded.id, commit.id); + assert_eq!(embedded.envelope, commit.envelope); + assert_eq!(embedded.payload, commit.payload); + } + + #[tokio::test] + async fn removal_notice_rejects_non_group_message_embeds() { + let sender = sender_keys(); + let receiver = receiver_keys(); + let recipient = MemberId::new(receiver.public_key().to_bytes().to_vec()); + let sender_peeler = NostrMlsPeeler::new().with_welcome_signer(sender.clone()); + + // A gift wrap (kind 1059) is not an embeddable group message. + let commit = wrapped_commit(&[0x5c; 32]).await; + let notice = sender_peeler + .wrap_removal_notice(&commit, &recipient) + .await + .unwrap() + .unwrap(); + let err = sender_peeler + .wrap_removal_notice(¬ice, &recipient) + .await + .unwrap_err(); + assert!(matches!(err, PeelerError::WrapFailed(_))); + } + + #[tokio::test] + async fn removal_notice_peel_rejects_a_tampered_embedded_event() { + let sender = sender_keys(); + let receiver = receiver_keys(); + let recipient = MemberId::new(receiver.public_key().to_bytes().to_vec()); + let receiver_peeler = NostrMlsPeeler::new().with_welcome_signer(receiver.clone()); + let group_id = vec![0x5c; 32]; + let commit = wrapped_commit(&group_id).await; + + // Hand-build a notice whose embedded event content was tampered after + // signing: verification of the embedded event must fail, so a forged + // notice can never inject bytes the committer did not publish. + let mut embedded = NostrTransportEvent::from_transport_message(&commit).unwrap(); + embedded.content = BASE64_STANDARD.encode(vec![0xEE; NOSTR_GROUP_CONTENT_MIN_LEN + 4]); + let rumor: UnsignedEvent = EventBuilder::new( + Kind::Custom(KIND_MARMOT_REMOVAL_NOTICE_RUMOR), + serde_json::to_string(&embedded).unwrap(), + ) + .tags([ + Tag::custom(nostr::TagKind::custom(GROUP_TAG), [hex::encode(&group_id)]), + Tag::custom( + nostr::TagKind::custom(REMOVAL_NOTICE_EVENT_TAG), + [embedded.id.clone()], + ), + ]) + .build(sender.public_key()); + let gift_wrap = EventBuilder::gift_wrap(&sender, &receiver.public_key(), rumor, []) + .await + .unwrap(); + let msg = NostrTransportEvent::from_nostr_event(&gift_wrap) + .unwrap() + .to_transport_message() + .unwrap(); + let _ = recipient; + + let err = receiver_peeler.peel_welcome(&msg).await.unwrap_err(); + assert!(matches!(err, PeelerError::Malformed(_))); + } +} diff --git a/spec/foundation/errors.md b/spec/foundation/errors.md index 16945c8d..3c364e99 100644 --- a/spec/foundation/errors.md +++ b/spec/foundation/errors.md @@ -21,6 +21,10 @@ An input that does not produce application content SHOULD map to one of these ca - `unsupported_required_feature`: the group requires a feature the client does not understand. - `authorization_failed`: the sender or committer is not allowed to make the change. - `missing_history`: the client would need retained state it no longer has. +- `evicted`: the input is for a group this identity is no longer a member of. +- `pre_membership`: the client has retained membership history for the group, and the input falls outside every interval + during which this identity was a member, so it can never be decrypted. A group the client has no state for at all is + `unknown_group`, not `pre_membership`. - `transport_rejected`: publication or delivery failed at the transport layer. Protocol-core docs can split these into more detailed outcomes when needed. @@ -61,12 +65,27 @@ Protocol-core documents name some outcomes in `PascalCase`. Each maps to one dis | ----------------------- | ----------- | ----------------- | ----------------------------------------------------------- | | `BeyondAnchor` | `stale` | `stale_epoch` | [retained-history.md](../protocol-core/retained-history.md) | | `MissingRetainedAnchor` | `deferred` | `missing_history` | [retained-history.md](../protocol-core/retained-history.md) | +| `PreMembership` | `stale` | `pre_membership` | [inbound-processing.md](../protocol-core/inbound-processing.md) | `BeyondAnchor` is window exclusion by design: the source epoch is older than the retained anchor, and the input will never be processed. `MissingRetainedAnchor` is storage loss: required retained state inside the rollback horizon is gone, canonical group state does not change, and the group moves to `Unrecoverable` (a group lifecycle state, not a disposition) until a verified repair path exists; the input stays deferred rather than terminal. +Non-membership (`Left` / `Evicted`) is a participation state, not a convergence disposition — it is reached only by +applying the removal commit, whether delivered in order, recovered through the transport's missed-input recovery, or +carried by a removal notice (per [group-state.md](../protocol-core/group-state.md)); it is never asserted from an +undecryptable message or an unverified notice. The `evicted` category is only for classifying an inbound message that arrives for a group this +identity is no longer a member of; such input is `stale`. + +`PreMembership` is terminal by design: the input falls outside every interval during which this identity was a member of +the group, so this client can never hold the keys to decrypt it. Because a group may be left/removed and later rejoined, +membership is a set of epoch intervals, not a single boundary; a client classifies an undecryptable historical message +against those retained intervals. Input inside a prior valid interval may still be recoverable from retained state and is +not `PreMembership`. It is scoped to groups the client has membership history for: with no retained state for the group +at all the input is `unknown_group`, not `PreMembership`. Unlike a deferred `MissingRetainedAnchor`, `PreMembership` MUST +NOT be deferred or retried. + ## Protocol and local errors Protocol rejections are part of interop. Local failures are not. diff --git a/spec/foundation/registries.md b/spec/foundation/registries.md index cf5bbb66..9a6b9862 100644 --- a/spec/foundation/registries.md +++ b/spec/foundation/registries.md @@ -92,6 +92,7 @@ list) — are not Marmot-owned and are defined in [../transports/nostr.md](../tr | `448` | Push token list response | Marmot app payload | [push-notifications.md](../features/push-notifications.md) | | `449` | Push token removal | Marmot app payload | [push-notifications.md](../features/push-notifications.md) | | `450` | Multi-device identity proof event | Local signing template, not relayed | [multi-device.md](../features/multi-device.md) | +| `451` | Marmot removal notice rumor | Nostr account transport | [nostr.md](../transports/nostr.md) | | `1009` | Message edit | Marmot app payload | [application-messages.md](application-messages.md) | | `1200` | Agent text stream start | Marmot app payload | [agent-text-streams-quic.md](../features/agent-text-streams-quic.md) | | `1201` | Agent activity | Marmot app payload | [agent-text-streams-quic.md](../features/agent-text-streams-quic.md) | diff --git a/spec/protocol-core/group-state.md b/spec/protocol-core/group-state.md index 19bf7b9b..0402a61b 100644 --- a/spec/protocol-core/group-state.md +++ b/spec/protocol-core/group-state.md @@ -128,3 +128,108 @@ outbound work. Queued group-state changes are regenerated after convergence status reaches `Settled` and the lifecycle state allows outbound work. A staged commit created before branch selection MUST NOT be reused after convergence changes the canonical state. + +## Participation + +The lifecycle states and convergence status above describe how a client converges on the group's canonical MLS state. +They do not describe whether the local identity is still a live member of that group. That is a separate, orthogonal +dimension: a group can be `Stable` and `Settled` and yet no longer include the local identity. + +Participation has four states: + +- `Member`: the local identity is present in the group's canonical roster. This is the only participation state in + which a client MAY prepare local group-state commits or emit delivered app payloads for the group. +- `Left`: the local identity voluntarily departed — its SelfRemove was committed (see + [member-departure.md](./member-departure.md)). Non-member; the group is inactive for this identity. +- `Evicted`: the local identity was removed by another member. Non-member; the group is inactive for this identity. +- `Quarantined`: the group is excluded from live processing and from the live group set pending an explicit recovery + transition. A quarantined group is neither trusted as a live member group nor asserted non-member; it is withheld. + +`Left` and `Evicted` are kept distinct — mirroring the `MemberLeft` vs `MemberRemoved` distinction elsewhere — so a +surface can tell "you left" from "you were removed" without labeling one as the other. A client that does not need the +distinction MAY treat both as a single non-member state, but the protocol MUST preserve the reason. + +Participation is orthogonal to the lifecycle state, but two couplings hold. `Left`, `Evicted`, and `Quarantined` are +terminal for normal processing the same way `Unrecoverable` is: a client MUST NOT apply group-state changes or release +outbound work while in them. Unlike `Unrecoverable` — a convergence failure the client MAY repair from retained +material — `Left` and `Evicted` reflect the canonical group's membership. They clear only through a verified rejoin or +reinstatement path — normally a new Welcome to a later epoch, or another explicit protocol-defined reinstatement. A +non-member client does not return to `Member` by resuming normal in-group processing: it was removed from the ratchet, +so it cannot apply a later commit for that group, and it MUST NOT try. Reinstatement returns the identity to `Member` +through a fresh membership grant, not through the group's own inbound stream. + +### Reaching a non-member state + +Removal authority lives in exactly one artifact: the commit that removes the identity. A client transitions to +`Left`/`Evicted` only by applying that commit — its own SelfRemove committed resolves to `Left`, a peer's removal to +`Evicted`; the roster diff after merging shows self in the removed set. No other input is authoritative: not an +undecryptable message, not a transport claim, not an out-of-band notice. Everything else in this section is a +discovery mechanism whose only job is to get that one commit delivered and applied. The removed identity can always +still process it: the removal commit is protected under the last epoch it was a member of, so it remains readable to +the removed member no matter how late it arrives. + +The removal commit is not guaranteed to arrive in order. Transport timing or ordering can deliver later, post-removal +traffic first, and MLS gives **no** signal in that case: with the removal commit unmerged the group is still active, +and later traffic merely fails to decrypt — indistinguishable at the MLS layer from future-epoch or corrupt input. +(The MLS `UseAfterEviction` guard is not this signal: it fires only after a merged removal has already made the group +inactive — aftermath of the applied-removal path, not a fresh discovery.) A client MUST therefore treat undecryptable +group traffic as a possible missed-removal symptom and actively pursue delivery of the missing commit: + +1. **Recovery probe.** Undecryptable traffic for a group MUST trigger the active transport's missed-input recovery + mechanism, bounded around the last input the client successfully consumed, looking for group-evolution input its + retained candidate keys can open. Every transport binding states its recovery mechanism — or the delivery + guarantees under which group-evolution input cannot be missed, in which case this trigger never fires (see + [../transports/README.md](../transports/README.md), "Transport document checklist"). If the missing removal commit + is recovered, it is applied through the ordinary inbound flow and the applied-removal transition above fires — + late, but identically. +2. **Removal notice.** On a transport binding with a recipient inbox address, the committer of a removal SHOULD also + send each removed member a removal notice through the member's account inbox, carrying or referencing the removal + commit (see [member-departure.md](./member-departure.md), "Removal notices"; the binding defines the shape). A + notice has no authority of its own: the receiver resolves it by validating and applying the carried or fetched + commit through the ordinary inbound flow, and a client MUST NOT change participation on an unverified notice. A + notice that does not resolve to a valid commit removing the local identity is ignored. Because a forged notice can + at most cause a validation attempt, an adversary gains nothing a real removal would not already grant. +3. **Bounded hold.** When undecryptable traffic persists and the probes above have stayed dry past a local policy + bound, the client MUST move the group to `Quarantined` with the `pending_membership` reason (see "Quarantine" + below): withheld, still probing, asserting neither `Member` nor a non-member state. It leaves that hold only when + the removal commit (or other group-evolution input that restores decryption) arrives — never by guessing. + +The `Left` vs `Evicted` reason comes only from the applied removal commit, so every route preserves it. A client MUST +NOT fabricate a reason it has not read from an applied commit. + +This design accepts an irreducible limit: a removed client that receives nothing at all — no later traffic, no notice — +is indistinguishable from a member of a quiet group, and no mechanism at any layer can distinguish them. The guarantee +is therefore eventual, not immediate: participation resolves to `Left`/`Evicted` once the removal commit is delivered +by any route, and a client keeps the discovery mechanisms above active rather than leaving a dead group readable as +`Member` indefinitely. + +### Quarantine + +A client places a group in `Quarantined` when it cannot safely treat the group as live but holds no applied removal +commit. Quarantine is a hold, not a verdict about membership, and it carries a reason so surfaces and recovery flows +can tell the holds apart — the two reasons have opposite expected exits: + +- `pending_membership`: undecryptable traffic suggests the local identity may have been removed, and the discovery + probes in "Reaching a non-member state" have not yet recovered the removal commit. The expected exit is resolution: + the removal commit arrives and the group transitions to `Left`/`Evicted`, or recovered group-evolution input + restores decryption and the group returns to `Member`. +- `integrity_hold`: stored group material fails to load or validate, or a durable invariant check fails. The expected + exit is repair: a verified repair path returns the group to live processing, or resolves it to a non-member state. + +While a group is `Quarantined`, under either reason: + +- it MUST be excluded from the live group set and from live inbound and convergence processing; +- every group accessor MUST agree that the group is withheld: a client MUST NOT expose a durable roster through one + accessor while another accessor reports the group as unknown. Either all live accessors reflect the quarantine, or the + group is exposed only through an explicit quarantine accessor; +- the group MUST NOT return to live processing except through an explicit recovery transition. Ordinary inbound or + convergence input MUST NOT silently re-activate a quarantined group. + +### Participation and public surfaces + +Public group APIs MUST let a caller distinguish a live member group, a non-member group, `Quarantined`, and "no such +group" from one another. Because `Left`/`Evicted` are reached only by applying the removal commit, a non-member state +always carries its reason; a group whose membership is merely in doubt is reported as `Quarantined` with its quarantine +reason (`pending_membership` vs `integrity_hold`) rather than being assigned a participation it cannot prove. +Collapsing a non-member or `Quarantined` group into either "active member" or "unknown group" is a defect: the first +keeps a dead group usable; the second loses the fact that the group existed and why it is no longer live. diff --git a/spec/protocol-core/inbound-processing.md b/spec/protocol-core/inbound-processing.md index 05993b03..a1c2475f 100644 --- a/spec/protocol-core/inbound-processing.md +++ b/spec/protocol-core/inbound-processing.md @@ -78,6 +78,24 @@ Input that cannot affect the group MUST receive a stale disposition. This includ - commits that fork from outside the rollback horizon: these are ineligible for branch selection (see [convergence.md](./convergence.md), "Eligibility") and, when their source epoch is also older than the retained anchor, are reported as `BeyondAnchor`. +- messages that fall outside every interval during which the local identity was a member of the group + (`PreMembership` -> `pre_membership`). Because a group may be left or removed and later rejoined, membership is a set + of epoch intervals, not a single boundary: a message inside a prior valid interval may still be recoverable from + retained state, while one outside all intervals is terminal. This is scoped to groups the client has membership + history for; input for a group the client has no state for at all is `unknown_group`, not `PreMembership`. Unlike a + deferred `MissingRetainedAnchor`, a `PreMembership` message MUST NOT be retried; +- messages for a group the local identity is no longer a member of (`evicted`): once participation is `Left` or + `Evicted` (see [group-state.md](./group-state.md)), further inbound for that group can no longer affect it and is + stale. + +Reaching `Left`/`Evicted` is a participation transition, not a disposition: it happens only when the removal commit is +applied (see [group-state.md](./group-state.md), "Reaching a non-member state"), never read off an inbound message's +processing error. An undecryptable post-removal message when the removal commit was never applied is an ordinary +wrong-epoch failure at this layer — `deferred` while the missing commit may still be recovered — but it MUST trigger +the discovery mechanisms in [group-state.md](./group-state.md): the active transport's missed-input recovery, bounded +around the last input the client successfully consumed, and resolution of any removal notice. When those probes stay +dry past the local policy bound, the group moves to `Quarantined` (`pending_membership`) and the input is withheld +with it, rather than given a terminal disposition it has not earned. The `snake_case` names in parentheses are the shared categories in [../foundation/errors.md](../foundation/errors.md); `BeyondAnchor` is a named convergence outcome that maps to the `stale` disposition and the `stale_epoch` category. diff --git a/spec/protocol-core/member-departure.md b/spec/protocol-core/member-departure.md index a392128c..1ed2cf96 100644 --- a/spec/protocol-core/member-departure.md +++ b/spec/protocol-core/member-departure.md @@ -73,6 +73,25 @@ is consumed MUST bound storage and commit eligibility to one retained proposal. under [inbound-processing.md](./inbound-processing.md), and non-identical redundant proposals are stale unless a future protocol version defines a distinct retry identity. +## Removal notices + +A commit that removes a member severs that member from the group's key schedule: everything after it is undecryptable +to them, so the removal commit itself is the last group input the removed member can process — and the only artifact +that can tell them they are out (see [group-state.md](./group-state.md), "Reaching a non-member state"). Delivery of +that one commit is therefore worth reinforcing beyond the ordinary group stream, which the removed member may fetch +late, partially, or not at all. + +After a commit that removes one or more members is published and applied, and when the active transport binding has a +recipient inbox address, the committer SHOULD send each removed member a removal notice through that member's account +inbox. Any other remaining member MAY also send one; duplicate notices collapse through ordinary deduplication of the +carried commit. The transport binding defines the notice shape; it MUST carry or reference the removal commit so the +receiver can validate and apply it. + +A removal notice is a delivery aid, not an authority: the receiver's participation changes only when the carried or +fetched commit validates and applies through the ordinary inbound flow (see [group-state.md](./group-state.md)). The +notice covers both departure paths — the committer of a SelfRemove-only commit notifies the leaver, confirming the +departure as `Left`, and the committer of an admin removal notifies the removed member, resolving to `Evicted`. + ## Validation A SelfRemove flow is invalid if: diff --git a/spec/transports/README.md b/spec/transports/README.md index 582aef16..768d284c 100644 --- a/spec/transports/README.md +++ b/spec/transports/README.md @@ -25,6 +25,13 @@ Each transport document MUST define: - envelope bytes for MLS Welcome delivery; - publish targets and acknowledgement rules; - receive filters or fetch rules; +- missed-input recovery: how a client re-obtains group-evolution input it did not receive — replayable history with a + recovery fetch rule, or delivery guarantees under which group-evolution input cannot be missed. Protocol-core + participation transitions depend on this ([../protocol-core/group-state.md](../protocol-core/group-state.md), + "Reaching a non-member state"); +- envelope bytes for removal notice delivery when the transport has a recipient inbox address + ([../protocol-core/member-departure.md](../protocol-core/member-departure.md), "Removal notices"), or an explicit + statement that the binding does not carry removal notices; - duplicate ids and replay handling inputs; - stale-input hints, if the envelope carries any; - validation that runs before MLS peeling; diff --git a/spec/transports/nostr.md b/spec/transports/nostr.md index 2f939d55..a3b5bc39 100644 --- a/spec/transports/nostr.md +++ b/spec/transports/nostr.md @@ -181,6 +181,36 @@ A receiver MUST reject a welcome that is not addressed to its own account identi A receiver MUST reject a kind `444` rumor whose content is not valid base64, whose `e` tag is missing or not a 32-byte hex Nostr event id, or whose `relays` tag is missing or empty. +## Removal notice delivery + +Nostr removal notices reinforce delivery of a removal commit to the member it removes +([../protocol-core/member-departure.md](../protocol-core/member-departure.md), "Removal notices"). They use NIP-59 +gift wraps with the same layering as welcomes: the outer relay event is kind `1059`, containing a kind `13` seal, +containing an unsigned kind `451` Marmot removal notice rumor. + +The gift-wrap recipient is the removed member's Nostr public key, and the notice is published to that account's kind +`10050` inbox relay set ("Account inbox relays" above). + +The inner kind `451` rumor MUST include: + +- `content`: the JSON-serialized kind `445` event that carries the removal commit, exactly as published to the group's + relays (the stringified-event convention NIP-18 reposts use); +- `h` tag: the lowercase hex `nostr_group_id` of the group, mirroring the kind `445` `h` tag; +- `e` tag: the Nostr event id of that kind `445` event. + +It MAY include a `relays` tag with relay URLs, using the relay URL profile above, where that kind `445` event is +fetchable. The inner rumor MUST NOT have a `sig` field. + +A receiver MUST reject a notice that is not addressed to its own account identity, and MUST ignore a notice for a +group it has no state for (`unknown_group`). The embedded kind `445` event MUST pass the same validation as a fetched +one — valid event id and Nostr signature, exactly one `h` tag whose value matches the rumor's `h` tag, base64 content +of at least 28 decoded bytes — and is then handed to the ordinary inbound pipeline: outer decryption with retained +candidate keys, deduplication on the recovered MLS bytes, and protocol-core convergence. A notice carries no authority +of its own: participation changes only if the recovered commit validates and applies as a removal of the local +identity ([../protocol-core/group-state.md](../protocol-core/group-state.md), "Reaching a non-member state"). A notice +whose embedded event is missing, invalid, or does not resolve to such a commit is ignored; before discarding, the +receiver MAY fall back to fetching the `e`-referenced event from the `relays` hints or the group's relay list. + ## KeyPackage publication Nostr KeyPackages use kind `30443`. @@ -250,6 +280,21 @@ A Nostr transport client subscribes to: Clients SHOULD use a `since` value when resubscribing if they have a retained transport timestamp. The timestamp is a fetch hint only. +### Membership backfill probe + +The membership backfill probe is this binding's missed-input recovery mechanism (the checklist item in +[README.md](./README.md); the recovery obligation is +[../protocol-core/group-state.md](../protocol-core/group-state.md), "Reaching a non-member state"). When a group holds +undecryptable kind `445` events — content no retained candidate key authenticates ("Outer decryption and epoch +selection" above) — the client MUST re-fetch kind +`445` events for the group's `nostr_group_id`, and any prior routing id the rotation rules still require, from the +full relay list in `marmot.transport.nostr.routing.v1`, with `since` set a local slack before the transport timestamp +of the last kind `445` event the client successfully consumed for that group. The window bounds the query; the +recovered events flow through ordinary validation, deduplication, and convergence, so the probe never chooses group +state — it only recovers candidate bytes that may include the missed removal commit. The probe MAY be repeated with +backoff; when it stays dry past the local policy bound, the group is held under the quarantine rules in +[../protocol-core/group-state.md](../protocol-core/group-state.md) rather than probed forever. + ## Publish targets and acknowledgements Group messages are published to the relay list in `marmot.transport.nostr.routing.v1`, after applying any local safety @@ -273,6 +318,8 @@ A Nostr transport client MUST validate the outer event enough to classify it bef MUST have base64 content whose decoded length is at least 28 bytes; - kind `1059` welcomes MUST be signed Nostr events and MUST have a `p` tag; - kind `444` welcome rumors MUST have `e` and `relays` tags after NIP-59 unwrapping; +- kind `451` removal notice rumors MUST have `h` and `e` tags after NIP-59 unwrapping, and their embedded kind `445` + event MUST pass kind `445` validation before it is handed to the peeler ("Removal notice delivery" above); - kind `30443` KeyPackage event content MUST be base64-encoded `MLSMessage` bytes whose wire format is `mls_key_package`; - fields that claim to be hex or base64 MUST decode successfully; @@ -304,6 +351,9 @@ Relays see only transport-envelope metadata, never plaintext or MLS secrets: - welcomes are NIP-59 gift wraps addressed to the invitee's account public key; the inbox address is the deliberate account-addressing exception ([../foundation/identity.md](../foundation/identity.md)). The gift wrap and seal hide the sender and the inner `kind 444` rumor. +- removal notices are NIP-59 gift wraps addressed to the removed member's account public key, the same + account-addressing exception as welcomes. The group id, embedded commit event, and sender stay inside the seal; + relays see only the kind `1059` envelope. - kind `30443` KeyPackage events are authored by the account identity, because their purpose is to let others find that account's packages. diff --git a/spec/transports/quic.md b/spec/transports/quic.md index 17b8d54f..e2f48d25 100644 --- a/spec/transports/quic.md +++ b/spec/transports/quic.md @@ -13,6 +13,10 @@ chooses group state, and a preview record never substitutes for the authoritativ binding is not required to display agent output: a `receive` member can ignore QUIC candidates and render the final kind `9` MLS message. +Because this binding carries no group-evolution input and has no recipient inbox address, the missed-input recovery and +removal notice checklist items ([README.md](./README.md)) do not apply here: both are owned by the active group +transport. + ## Scope This binding owns: