diff --git a/plans/2026-06-22-roster-catchup-subscribers.md b/plans/2026-06-22-roster-catchup-subscribers.md new file mode 100644 index 0000000000..fd1979cda0 --- /dev/null +++ b/plans/2026-06-22-roster-catchup-subscribers.md @@ -0,0 +1,83 @@ +# Roster catch-up for channel subscribers + +Continues the public-groups roster work (privileged roster, member keys, `VersionRoster` monotonic gate, task-047's `Maybe VersionRoster` on `XGrpMemRole`/`XGrpMemDel`). + +## Problem + +A channel subscriber learns the full roster only on join and on resume (`serveRoster`, Subscriber.hs:1174/1269), then tracks it via forwarded `XGrpMemRole`/`XGrpMemDel` deltas, each carrying an incrementing `VersionRoster`. The gate (`applyAtRosterVersion`, Subscriber.hs:3248) accepts any delta at `v >= cur` and advances to `v`. + +If the subscriber misses intermediate deltas and then receives one at a version more than one above its known version, it advances to `v` and applies that one change — but the roster changes carried by the skipped versions (other members' roles, removals, keys) are lost until the next resume. The subscriber can then reject a freshly-promoted member's signed action (`RGEMsgBadSignature`) for a key it never learned. + +Catch-up: on detecting a gap, the subscriber asks the relay that forwarded the delta to re-serve the full roster, which carries the complete set and keys. + +## Design + +Four pieces. No schema change, no new store function. + +### 1. Owner sends the roster before the delta (the enabler) + +Today the owner sends `XGrpMemRole`/`XGrpMemDel` first, then `broadcastRoster` (Commands.hs:2754/2885). The relay forwards the delta to subscribers *before* its own roster-blob transfer completes, so for the whole transfer its stored `roster_blob` lags its `roster_version`. A catch-up request in that window serves a stale blob the subscriber rejects (`notBelowRoster`) — which is the common case, not an edge. + +Fix: in `APIMembersRole` / `APIRemoveMembers`, reorder to **apply the change to the owner's members → `broadcastRoster v` → send the delta**. This restructures `changeRoleCurrentMems` / `deleteMemsSend` to separate "apply to owner DB" from "send delta to relays" (the roster is still built after the change, so it reflects the new roles / excludes the removed member). + +Effect, relying on FIFO order of the owner→relay connection (a guarantee the roster design already assumes): the relay applies and stores the blob at `v` (`setGroupLiveRoster`) before it processes and forwards the delta at `v`. So a relay's `roster_version` always reflects its stored blob, and any request triggered by a forwarded delta finds a current blob. The now-redundant delta hits the existing no-op suppression (`fromRole == memRole`, Subscriber.hs:3298, added in task-047) and is still forwarded to subscribers — no new relay logic, subscribers still see the delta. + +Behavioral change to core delivery (all relays/subscribers) → its own commit. Implementation must confirm FIFO holds owner→relay (the test will expose a violation). Residual: a failed roster send (rare; already `catchAllErrors`) leaves that relay's gate briefly above its blob until the next change — heals on resume. + +### 2. New event `XGrpRosterRequest VersionRoster` (Protocol.hs) + +A directed subscriber→relay control message carrying the subscriber's pre-gap version, so the relay can skip serving when it holds nothing newer. + +```haskell +XGrpRosterRequest :: VersionRoster -> ChatMsgEvent 'Json +``` + +Wire plumbing (standard single-`VersionRoster` `'Json` event): GADT constructor; `CMEventTag` `XGrpRosterRequest_`; `strEncode` `"x.grp.roster.request"`; `strP` case; `toCMEventTag`; `appJsonToCM` (`XGrpRosterRequest <$> p "version"`); `chatToAppMessage` (`o ["version" .= v]`). NOT in `isForwardedGroupMsg` (point-to-point). NOT in `requiresSignature` (subscribers/observers have no key). + +### 3. Subscriber detects the gap and requests (Subscriber.hs, `applyAtRosterVersion`) + +`applyAtRosterVersion` already reads `cur`, compares `v >= cur`, advances the gate — the one path shared by `xGrpMemRole` and `xGrpMemDel`. In its accepting branch, when the receiver is a subscriber (`not (isUserGrpFwdRelay gInfo)`) and `cur = Just c` with `v > c + 1`, send `XGrpRosterRequest c` to the relay that forwarded the delta, then run the action unchanged (the delta is still applied; the roster heals `c+1 .. v-1`). + +To target the forwarding relay, thread it as `Maybe GroupMember` through `xGrpMemRole` / `xGrpMemDel` / `applyAtRosterVersion`: `Just m` on the forwarded path (`m` in scope in `processForwardedMsg`, Subscriber.hs:3804/3806), `Nothing` on the direct path (1092/1096; the receiver there is a relay and never requests). Send via `sendGroupMessage' user gInfo [relay] (XGrpRosterRequest c)`. + +One forwarding relay only, not all relays: avoids N² requests under concurrent multi-relay gaps, and the owner-signed roster needs no cross-relay verification. The gate advances to `v` on the first gap in a batch, so later `+1` deltas aren't gaps — normally one request per batch. + +### 4. Relay serves on request (Subscriber.hs, `processEvent` + new `xGrpRosterRequest`) + +The request arrives on the direct member connection, dispatched in `processEvent` (alongside other direct group events, ~Subscriber.hs:1103): + +```haskell +XGrpRosterRequest reqVer -> Nothing <$ xGrpRosterRequest gInfo' m'' reqVer +``` + +```haskell +xGrpRosterRequest :: GroupInfo -> GroupMember -> VersionRoster -> CM () +xGrpRosterRequest gInfo m reqVer = + when (isUserGrpFwdRelay gInfo) $ do + cur <- withStore' $ \db -> getGroupRosterVersion db gInfo + when (maybe True (> reqVer) cur) $ serveRoster user gInfo m +``` + +Reuses `serveRoster` unchanged (signed header + inline blob chunks to the one requester — the join/resume path). The version check serves only when the relay holds something newer than the requester, rate-limiting same-version spam. With piece 1, the served blob is current, so the subscriber accepts it (`notBelowRoster`) and `rosterCompletion` heals the skipped set and keys. `serveRoster` is a no-op without a stored roster. + +## Files touched + +- `src/Simplex/Chat/Library/Commands.hs` — reorder owner sends in `APIMembersRole`/`APIRemoveMembers` (restructure `changeRoleCurrentMems`/`deleteMemsSend`). +- `src/Simplex/Chat/Protocol.hs` — new event, end to end. +- `src/Simplex/Chat/Library/Subscriber.hs` — gap detection + request in `applyAtRosterVersion`; thread forwarding relay through `xGrpMemRole`/`xGrpMemDel`; new `xGrpRosterRequest`; one `processEvent` dispatch line. +- No schema change, no new store fn (reuses `getGroupRosterVersion`, `getGroupRelayMembers`, `getGroupRoster`, `serveRoster`). +- `tests/ChatTests/` — one channel catch-up test. + +## Tests + +A `channels` test asserting a subscriber that observes a version gap recovers the skipped set. The hard part is staging a deterministic gap (FIFO forwarding doesn't drop deltas). Candidate: two relays where an intermediate delta reaches only one path, so the subscriber receives a later delta at a jumped version; assert it emits exactly one `XGrpRosterRequest`, the relay re-serves, and a member promoted in the skipped interval is afterwards known (its signed action accepted, no `RGEMsgBadSignature`). Confirm the gap is reliably reproducible before finalizing; fall back to asserting "a forwarded delta at a jumped version triggers one request + re-serve" if the two-relay setup is flaky. Iterate with `-m "channels"`. + +## Commits + +1. Reorder owner sends (roster before delta) — relay blob is current before it forwards deltas. +2. Protocol: add `XGrpRosterRequest`. +3. Relay: `xGrpRosterRequest` + `processEvent` dispatch. +4. Subscriber: gap detection + request to the forwarding relay. +5. Test + schema/plan regen. + +Build and run `-m "channels"` after each. diff --git a/simplex-chat.cabal b/simplex-chat.cabal index cb36e989da..e84e50bd0a 100644 --- a/simplex-chat.cabal +++ b/simplex-chat.cabal @@ -146,6 +146,7 @@ library Simplex.Chat.Store.Postgres.Migrations.M20260601_relay_sent_web_domain Simplex.Chat.Store.Postgres.Migrations.M20260602_group_roster Simplex.Chat.Store.Postgres.Migrations.M20260603_simplex_name + Simplex.Chat.Store.Postgres.Migrations.M20260629_roster_catchup else exposed-modules: Simplex.Chat.Archive @@ -309,6 +310,7 @@ library Simplex.Chat.Store.SQLite.Migrations.M20260601_relay_sent_web_domain Simplex.Chat.Store.SQLite.Migrations.M20260602_group_roster Simplex.Chat.Store.SQLite.Migrations.M20260603_simplex_name + Simplex.Chat.Store.SQLite.Migrations.M20260629_roster_catchup other-modules: Paths_simplex_chat hs-source-dirs: diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index 836f4e72fd..772e10dfb0 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/src/Simplex/Chat/Library/Commands.hs @@ -2795,34 +2795,35 @@ processChatCommand cxt nm = \case -- TODO [relays] possible optimization is to read only required members + relays g@(Group gInfo members) <- withFastStore $ \db -> getGroup db cxt user groupId when (selfSelected gInfo) $ throwCmdError "can't change role for self" - let (invitedMems, currentMems, unchangedMems, maxRole, anyAdmin, anyPending, anyPrivilegedTarget, finalPrivilegedCount) = selectMembers members + let (invitedMems, currentMems, unchangedMems, maxRole, anyAdmin, anyPending, anyPrivilegedTarget, anyRosterChange, finalPrivilegedCount) = selectMembers members when (length invitedMems + length currentMems + length unchangedMems /= length memberIds) $ throwChatError CEGroupMemberNotFound when (length memberIds > 1 && (anyAdmin || newRole >= GRAdmin)) $ throwCmdError "can't change role of multiple members when admins selected, or new role is admin" when anyPending $ throwCmdError "can't change role of members pending approval" assertUserGroupRole gInfo $ maximum ([GRAdmin, maxRole, newRole] :: [GroupMemberRole]) - -- in relay groups the roster has a single signer, so only the owner may change moderator/admin roles + -- in relay groups the roster has a single signer, so only the owner may change member/moderator/admin roles when (useRelays' gInfo && (isRosterRole newRole || anyPrivilegedTarget) && memberRole' (membership gInfo) /= GROwner) $ throwCmdError "only the group owner can change moderator and admin roles" when (useRelays' gInfo && isRosterRole newRole && finalPrivilegedCount > maxGroupRosterSize) $ throwCmdError $ "the number of members, moderators and admins would exceed the limit of " <> show maxGroupRosterSize (errs1, changed1) <- changeRoleInvitedMems user gInfo invitedMems - let doBumpRoster = useRelays' gInfo && memberRole' (membership gInfo) == GROwner && (isRosterRole newRole || anyPrivilegedTarget) - rosterVer <- if doBumpRoster then Just <$> reserveRosterVersion gInfo else pure Nothing + let doBumpRoster = useRelays' gInfo && memberRole' (membership gInfo) == GROwner && anyRosterChange + -- roster (with the change projected in) before the delta, so a relay stores the blob at this version before forwarding the delta + rosterVer <- if doBumpRoster then Just <$> broadcastRoster user gInfo (RDRoleChanged newRole currentMems) else pure Nothing (errs2, changed2, acis, msgSigned) <- changeRoleCurrentMems user g rosterVer currentMems - forM_ rosterVer $ \v -> broadcastRoster user gInfo v `catchAllErrors` eToView unless (null acis) $ toView $ CEvtNewChatItems user acis let errs = errs1 <> errs2 unless (null errs) $ toView $ CEvtChatErrors errs pure $ CRMembersRoleUser {user, groupInfo = gInfo, members = changed1 <> changed2, toRole = newRole, msgSigned} -- same order is not guaranteed where selfSelected GroupInfo {membership} = elem (groupMemberId' membership) memberIds - -- anyPrivilegedTarget: a target currently moderator/admin; finalPrivilegedCount: - -- moderators + admins after the change (targets take newRole, others keep their role). - selectMembers :: [GroupMember] -> ([GroupMember], [GroupMember], [GroupMember], GroupMemberRole, Bool, Bool, Bool, Int) - selectMembers = foldr' addMember ([], [], [], GRObserver, False, False, False, 0) + -- anyPrivilegedTarget: a target currently member/moderator/admin (gates the owner-only check); anyRosterChange: + -- a current member's role change that alters the roster blob - the only case that bumps the version, since a + -- bump with no delta reads as a gap to subscribers; finalPrivilegedCount: moderators + admins after the change. + selectMembers :: [GroupMember] -> ([GroupMember], [GroupMember], [GroupMember], GroupMemberRole, Bool, Bool, Bool, Bool, Int) + selectMembers = foldr' addMember ([], [], [], GRObserver, False, False, False, False, 0) where - addMember m@GroupMember {groupMemberId, memberStatus, memberRole} (invited, current, unchanged, maxRole, anyAdmin, anyPending, anyPrivTarget, privCount) + addMember m@GroupMember {groupMemberId, memberStatus, memberRole} (invited, current, unchanged, maxRole, anyAdmin, anyPending, anyPrivTarget, anyRosterChange, privCount) | groupMemberId `elem` memberIds = let maxRole' = max maxRole memberRole anyAdmin' = anyAdmin || memberRole >= GRAdmin @@ -2830,10 +2831,11 @@ processChatCommand cxt nm = \case anyPrivTarget' = anyPrivTarget || isRosterRole memberRole privCount' = if isRosterRole newRole then privCount + 1 else privCount in if - | memberRole == newRole -> (invited, current, m : unchanged, maxRole', anyAdmin', anyPending', anyPrivTarget', privCount') - | memberStatus == GSMemInvited -> (m : invited, current, unchanged, maxRole', anyAdmin', anyPending', anyPrivTarget', privCount') - | otherwise -> (invited, m : current, unchanged, maxRole', anyAdmin', anyPending', anyPrivTarget', privCount') - | otherwise = (invited, current, unchanged, maxRole, anyAdmin, anyPending, anyPrivTarget, if isRosterRole memberRole then privCount + 1 else privCount) + | memberRole == newRole -> (invited, current, m : unchanged, maxRole', anyAdmin', anyPending', anyPrivTarget', anyRosterChange, privCount') + | memberStatus == GSMemInvited -> (m : invited, current, unchanged, maxRole', anyAdmin', anyPending', anyPrivTarget', anyRosterChange, privCount') + -- a current member's role actually changes here; it alters the roster iff the old or new role is on it + | otherwise -> (invited, m : current, unchanged, maxRole', anyAdmin', anyPending', anyPrivTarget', anyRosterChange || isRosterRole newRole || isRosterRole memberRole, privCount') + | otherwise = (invited, current, unchanged, maxRole, anyAdmin, anyPending, anyPrivTarget, anyRosterChange, if isRosterRole memberRole then privCount + 1 else privCount) changeRoleInvitedMems :: User -> GroupInfo -> [GroupMember] -> CM ([ChatError], [GroupMember]) changeRoleInvitedMems user gInfo memsToChange = do -- not batched, as we need to send different invitations to different connections anyway @@ -2925,7 +2927,7 @@ processChatCommand cxt nm = \case withGroupLock "removeMembers" groupId $ do -- TODO [relays] possible optimization is to read only required members + relays Group gInfo members <- withFastStore $ \db -> getGroup db cxt user groupId - let (count, invitedMems, pendingApprvMems, pendingRvwMems, currentMems, maxRole, anyAdmin, anyPrivilegedRemoved) = selectMembers gmIds members + let (count, invitedMems, pendingApprvMems, pendingRvwMems, currentMems, maxRole, anyAdmin, anyPrivilegedRemoved, anyRosterRemoved) = selectMembers gmIds members gmIds = S.fromList $ L.toList groupMemberIds memCount = length groupMemberIds when (count /= memCount) $ throwChatError CEGroupMemberNotFound @@ -2935,15 +2937,15 @@ processChatCommand cxt nm = \case throwCmdError "only the group owner can remove members, moderators and admins" (errs1, deleted1) <- deleteInvitedMems user invitedMems let recipients = filter memberCurrent members - let doBumpRoster = useRelays' gInfo && memberRole' (membership gInfo) == GROwner && anyPrivilegedRemoved - rosterVer <- if doBumpRoster then Just <$> reserveRosterVersion gInfo else pure Nothing + let doBumpRoster = useRelays' gInfo && memberRole' (membership gInfo) == GROwner && anyRosterRemoved + -- roster (excluding the removed members) before the delta, so a relay stores the blob at this version before forwarding the delta + rosterVer <- if doBumpRoster then Just <$> broadcastRoster user gInfo (RDRemoved currentMems) else pure Nothing (errs2, deleted2, acis2, signed2) <- deleteMemsSend user gInfo Nothing rosterVer recipients currentMems (errs3, deleted3, acis3, signed3) <- foldM (\acc m -> deletePendingMember acc user gInfo [m] m) ([], [], [], False) pendingApprvMems let moderators = filter (\GroupMember {memberRole} -> memberRole >= GRModerator) members (errs4, deleted4, acis4, signed4) <- foldM (\acc m -> deletePendingMember acc user gInfo (m : moderators) m) ([], [], [], False) pendingRvwMems - forM_ rosterVer $ \v -> broadcastRoster user gInfo v `catchAllErrors` eToView let acis = acis2 <> acis3 <> acis4 errs = errs1 <> errs2 <> errs3 <> errs4 deleted = deleted1 <> deleted2 <> deleted3 <> deleted4 @@ -2958,20 +2960,24 @@ processChatCommand cxt nm = \case unless (null errs) $ toView $ CEvtChatErrors errs pure $ CRUserDeletedMembers user gInfo' deleted withMessages msgSigned -- same order is not guaranteed where - selectMembers :: S.Set GroupMemberId -> [GroupMember] -> (Int, [GroupMember], [GroupMember], [GroupMember], [GroupMember], GroupMemberRole, Bool, Bool) - selectMembers gmIds = foldl' addMember (0, [], [], [], [], GRObserver, False, False) + -- anyPrivilegedRemoved: any removed member is member/moderator/admin (gates the owner-only check); anyRosterRemoved: + -- a current roster member is removed - the only case that alters the blob and so bumps the version (pending/ + -- invited members aren't on the roster, and a bump with no delta reads as a gap to subscribers). + selectMembers :: S.Set GroupMemberId -> [GroupMember] -> (Int, [GroupMember], [GroupMember], [GroupMember], [GroupMember], GroupMemberRole, Bool, Bool, Bool) + selectMembers gmIds = foldl' addMember (0, [], [], [], [], GRObserver, False, False, False) where - addMember acc@(n, invited, pendingApprv, pendingRvw, current, maxRole, anyAdmin, anyPrivRemoved) m@GroupMember {groupMemberId, memberStatus, memberRole} + addMember acc@(n, invited, pendingApprv, pendingRvw, current, maxRole, anyAdmin, anyPrivRemoved, anyRosterRemoved) m@GroupMember {groupMemberId, memberStatus, memberRole} | groupMemberId `S.member` gmIds = let maxRole' = max maxRole memberRole anyAdmin' = anyAdmin || memberRole >= GRAdmin anyPrivRemoved' = anyPrivRemoved || isRosterRole memberRole n' = n + 1 in case memberStatus of - GSMemInvited -> (n', m : invited, pendingApprv, pendingRvw, current, maxRole', anyAdmin', anyPrivRemoved') - GSMemPendingApproval -> (n', invited, m : pendingApprv, pendingRvw, current, maxRole', anyAdmin', anyPrivRemoved') - GSMemPendingReview -> (n', invited, pendingApprv, m : pendingRvw, current, maxRole', anyAdmin', anyPrivRemoved') - _ -> (n', invited, pendingApprv, pendingRvw, m : current, maxRole', anyAdmin', anyPrivRemoved') + GSMemInvited -> (n', m : invited, pendingApprv, pendingRvw, current, maxRole', anyAdmin', anyPrivRemoved', anyRosterRemoved) + GSMemPendingApproval -> (n', invited, m : pendingApprv, pendingRvw, current, maxRole', anyAdmin', anyPrivRemoved', anyRosterRemoved) + GSMemPendingReview -> (n', invited, pendingApprv, m : pendingRvw, current, maxRole', anyAdmin', anyPrivRemoved', anyRosterRemoved) + -- removed current member: alters the roster blob iff it currently holds a roster role + _ -> (n', invited, pendingApprv, pendingRvw, m : current, maxRole', anyAdmin', anyPrivRemoved', anyRosterRemoved || isRosterRole memberRole) | otherwise = acc deleteInvitedMems :: User -> [GroupMember] -> CM ([ChatError], [GroupMember]) deleteInvitedMems user memsToDelete = do diff --git a/src/Simplex/Chat/Library/Internal.hs b/src/Simplex/Chat/Library/Internal.hs index d34e08401b..62ce525cbd 100644 --- a/src/Simplex/Chat/Library/Internal.hs +++ b/src/Simplex/Chat/Library/Internal.hs @@ -1187,8 +1187,8 @@ serveRoster :: User -> GroupInfo -> GroupMember -> CM () serveRoster user gInfo member = when (member `supportsVersion` groupRosterVersion) $ do cxt <- chatStoreCxt - withStore' (\db -> getGroupRoster db gInfo) >>= \case - Just (ownerGMId, brokerTs, sm@SignedMsg {signedBody}, blob_) -> + withStore' (\db -> getStoredGroupRoster db gInfo) >>= \case + Just (ownerGMId, brokerTs, sm@SignedMsg {signedBody}, blob_, storedVer_) -> case J.eitherDecodeStrict' signedBody :: Either String (ChatMessage 'Json) of Left e -> logError $ "serveRoster: cannot decode saved roster message: " <> tshow e Right chatMsg@ChatMessage {msgId} -> @@ -1198,6 +1198,10 @@ serveRoster user gInfo member = sendFwdMemberMessage member fwd (VMSigned MSSVerified sm chatMsg) forM_ ((,) <$> msgId <*> blob_) $ \(sid, blob) -> sendInlineBlobChunks user gInfo [member] sid blob + -- record the blob's own stored version as served, not roster_version (the gate): a delta can + -- advance the gate past the stored blob on a failed blob send, and recording the gate would + -- over-claim what this member was actually served, suppressing legitimate catch-up + forM_ storedVer_ $ \v -> withStore' $ \db -> setMemberRosterServedVersion db member v Left e -> logError $ "serveRoster: roster owner not found: " <> tshow e Nothing -> pure () @@ -2285,24 +2289,39 @@ sendGroupMessage' user gInfo members chatMsgEvent = ((Right msg) :| [], _) -> pure msg _ -> throwChatError $ CEInternalError "sendGroupMessage': expected 1 message" +-- The roster change being broadcast, projected onto the current roster members in broadcastRoster. This lets the +-- roster blob be built (and sent) before the change is applied to the owner's own member records, so the owner +-- never demotes/removes a member locally before the change has been propagated to relays. +data RosterDelta + = RDRoleChanged GroupMemberRole [GroupMember] -- these members now hold this role + | RDRemoved [GroupMember] -- these members are removed from the group + +applyRosterDelta :: RosterDelta -> [GroupMember] -> [GroupMember] +applyRosterDelta delta current = case delta of + RDRoleChanged role changed -> map (\m -> (m :: GroupMember) {memberRole = role}) changed <> without changed + RDRemoved removed -> without removed + where + without ms = let ids = S.fromList (map groupMemberId' ms) in filter ((`S.notMember` ids) . groupMemberId') current + -- TODO [relays] improvement: publish roster_version in link data so the owner can recover the latest version -- TODO after restoring from a stale backup (relays accept only strictly-greater versions) --- Persist the next roster version before sending the events that carry it (so a recipient never advances --- past a version the owner hasn't recorded). The matching blob is broadcast separately, by broadcastRoster, --- after the change is applied to the owner's members - so the served roster excludes demoted/removed members. -reserveRosterVersion :: GroupInfo -> CM VersionRoster -reserveRosterVersion gInfo = do +-- Reserve and persist the next roster version (committed before the events that carry it, so a recipient never +-- advances past a version the owner hasn't recorded), then broadcast the matching blob with the change projected +-- onto the served roster (so it excludes demoted/removed members). Returns the reserved version for the delta +-- that follows. The blob send is best-effort - a failed send heals on the next change or on resume. +broadcastRoster :: User -> GroupInfo -> RosterDelta -> CM VersionRoster +broadcastRoster user gInfo delta = do let rosterVer = maybe (VersionRoster 0) (\(VersionRoster n) -> VersionRoster (n + 1)) (rosterVersion gInfo) withStore' $ \db -> setGroupRosterVersion db gInfo rosterVer + sendRosterBlob rosterVer `catchAllErrors` eToView pure rosterVer - -broadcastRoster :: User -> GroupInfo -> VersionRoster -> CM () -broadcastRoster user gInfo rosterVer = do - cxt <- chatStoreCxt - (relays, rosterMems) <- withStore' $ \db -> - (,) <$> getGroupRelayMembers db cxt user gInfo <*> getGroupRosterMembers db cxt user gInfo - forM_ (L.nonEmpty relays) $ \relays' -> - sendRoster user gInfo (L.toList relays') rosterVer (buildGroupRoster rosterMems) + where + sendRosterBlob rosterVer = do + cxt <- chatStoreCxt + (relays, rosterMems) <- withStore' $ \db -> + (,) <$> getGroupRelayMembers db cxt user gInfo <*> getGroupRosterMembers db cxt user gInfo + forM_ (L.nonEmpty relays) $ \relays' -> + sendRoster user gInfo (L.toList relays') rosterVer (buildGroupRoster $ applyRosterDelta delta rosterMems) -- Send the current roster (no version bump) to a newly added relay so it can serve joiners. sendGroupRosterToRelay :: User -> GroupInfo -> GroupMember -> CM () diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index 9cc6778122..408fae8105 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -1086,11 +1086,11 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = XGrpMemIntro memInfo memRestrictions_ -> Nothing <$ xGrpMemIntro gInfo' m'' memInfo memRestrictions_ XGrpMemInv memId introInv -> Nothing <$ xGrpMemInv gInfo' m'' memId introInv XGrpMemFwd memInfo introInv -> Nothing <$ xGrpMemFwd gInfo' m'' memInfo introInv - XGrpMemRole memId memRole memberKey rosterVer -> fmap ctx <$> xGrpMemRole gInfo' m'' memId memRole memberKey rosterVer msg brokerTs + XGrpMemRole memId memRole memberKey rosterVer -> fmap ctx <$> xGrpMemRole gInfo' Nothing m'' memId memRole memberKey rosterVer msg brokerTs XGrpMemRestrict memId memRestrictions -> fmap ctx <$> xGrpMemRestrict gInfo' m'' memId memRestrictions msg brokerTs XGrpMemCon memId -> Nothing <$ xGrpMemCon gInfo' m'' memId XGrpMemDel memId withMessages rosterVer -> case encoding @e of - SJson -> fmap ctx <$> xGrpMemDel gInfo' m'' memId withMessages rosterVer verifiedMsg msg brokerTs False + SJson -> fmap ctx <$> xGrpMemDel gInfo' Nothing m'' memId withMessages rosterVer verifiedMsg msg brokerTs False SBinary -> pure Nothing XGrpLeave -> fmap ctx <$> xGrpLeave gInfo' m'' msg brokerTs XGrpDel -> Just (DeliveryTaskContext (DJSGroup {jobSpec = DJRelayRemoved}) False) <$ xGrpDel gInfo' m'' msg brokerTs @@ -1098,6 +1098,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = XGrpPrefs ps' -> fmap ctx <$> xGrpPrefs gInfo' m'' ps' msg XGrpRoster gr -> fmap ctx <$> xGrpRoster gInfo' m'' m'' gr verifiedMsg sharedMsgId_ brokerTs XGrpRosterAck ackVer ackErr -> Nothing <$ xGrpRosterAck gInfo' m'' ackVer ackErr + XGrpRosterRequest reqVer -> Nothing <$ xGrpRosterRequest gInfo' m'' reqVer -- TODO [knocking] why don't we forward these messages? XGrpDirectInv connReq mContent_ msgScope -> memberCanSend (Just m'') msgScope $ Nothing <$ xGrpDirectInv gInfo' m'' conn' connReq mContent_ msg brokerTs XGrpMsgForward fwd msg' -> Nothing <$ xGrpMsgForward gInfo' Nothing m'' fwd (ParsedMsg Nothing Nothing msg') brokerTs @@ -3239,29 +3240,57 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = -- batch), then advance it in the same transaction; a strictly lower version is a replay and is ignored. -- Only an owner sender may advance it: a non-owner signed event is rejected by the action that follows, -- but must not bump roster_version first, or every later owner roster at a lower version is dropped. - applyAtRosterVersion :: GroupInfo -> GroupMember -> Maybe VersionRoster -> CM (Maybe DeliveryJobScope) -> CM (Maybe DeliveryJobScope) - applyAtRosterVersion gInfo sender rosterVer_ action + applyAtRosterVersion :: GroupInfo -> Maybe GroupMember -> GroupMember -> Maybe VersionRoster -> CM (Maybe DeliveryJobScope) -> CM (Maybe DeliveryJobScope) + applyAtRosterVersion gInfo fwdRelay_ sender rosterVer_ action | not (useRelays' gInfo) = action | otherwise = case rosterVer_ of Nothing -> action Just _ | memberRole' sender /= GROwner -> action Just v -> do - accept <- withStore' $ \db -> do - cur <- getGroupRosterVersion db gInfo - let fresh = maybe True (v >=) cur - when fresh $ setGroupRosterVersion db gInfo v - pure fresh + (accept, prevComplete) <- withStore' $ \db -> do + gate <- getGroupRosterVersion db gInfo + prevComplete <- getCompleteRosterVersion db gInfo + let fresh = maybe True (v >=) gate + when fresh $ do + setGroupRosterVersion db gInfo v + -- advance the frontier when this delta is the next version. One version can carry several deltas + -- (a multi-member role change), delivered in order, so seeing any one of them advances the frontier + -- past that whole version. + when (v == nextCompleteVersion prevComplete) $ + setCompleteRosterVersion db gInfo v + pure (fresh, prevComplete) if accept - then action + then (requestRosterOnGap v prevComplete `catchAllErrors` eToView) >> action else messageWarning "x.grp.mem: roster version not newer than current, ignoring" $> Nothing + where + -- the next contiguous version after the complete frontier. With no frontier the baseline is the first + -- roster version (VersionRoster 0, see broadcastRoster): a subscriber seeing v0 without a prior roster is + -- current, not gapped, as it cannot have missed an earlier version - so v0 neither requests nor stays behind. + nextCompleteVersion = \case + Just (VersionRoster c) -> VersionRoster (c + 1) + Nothing -> VersionRoster 0 + -- a subscriber whose complete frontier (before this delta) lags more than one below it has missed versions: + -- ask the relay that forwarded it (it holds >= v = the new gate) to re-serve the full roster, carrying the + -- previous frontier so only a fuller snapshot is served. A stuck frontier re-asks on every following delta + -- until a roster fills it. Best-effort; relays and the direct path (fwdRelay_ = Nothing) don't ask, nor a + -- relay that predates roster support. + requestRosterOnGap v prevComplete + | isUserGrpFwdRelay gInfo = pure () + | otherwise = case fwdRelay_ of + Just relay + | gap, relay `supportsVersion` groupRosterVersion -> + void $ sendGroupMessage' user gInfo [relay] (XGrpRosterRequest prevComplete) + _ -> pure () + where + gap = v > nextCompleteVersion prevComplete - xGrpMemRole :: GroupInfo -> GroupMember -> MemberId -> GroupMemberRole -> Maybe MemberKey -> Maybe VersionRoster -> RcvMessage -> UTCTime -> CM (Maybe DeliveryJobScope) - xGrpMemRole gInfo@GroupInfo {membership} m@GroupMember {memberRole = senderRole} memId memRole memberKey_ rosterVer_ msg@RcvMessage {msgSigned} brokerTs + xGrpMemRole :: GroupInfo -> Maybe GroupMember -> GroupMember -> MemberId -> GroupMemberRole -> Maybe MemberKey -> Maybe VersionRoster -> RcvMessage -> UTCTime -> CM (Maybe DeliveryJobScope) + xGrpMemRole gInfo@GroupInfo {membership} fwdRelay_ m@GroupMember {memberRole = senderRole} memId memRole memberKey_ rosterVer_ msg@RcvMessage {msgSigned} brokerTs | membershipMemId == memId = - applyAtRosterVersion gInfo m rosterVer_ $ + applyAtRosterVersion gInfo fwdRelay_ m rosterVer_ $ let gInfo' = gInfo {membership = membership {memberRole = memRole}} in changeMemberRole gInfo' membership False (\db -> updateGroupMemberRole db user membership memRole) (RGEUserRole memRole) True - | otherwise = applyAtRosterVersion gInfo m rosterVer_ $ do + | otherwise = applyAtRosterVersion gInfo fwdRelay_ m rosterVer_ $ do defaultRole <- unknownMemberRole gInfo -- an owner-signed event with a key TOFU-creates an unknown member only for a roster role; else a plain lookup let allowCreate = useRelays' gInfo && senderRole == GROwner && isRosterRole memRole && isJust memberKey_ @@ -3437,18 +3466,22 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = `catchAllErrors` \_ -> pure (cs, as) emitRosterResults :: GroupInfo -> GroupMember -> UTCTime -> ([MemberId], [(GroupMember, GroupMemberRole, Bool)]) -> CM () - emitRosterResults gInfo author rosterBrokerTs (conflicts, applied) = do + emitRosterResults gInfo@GroupInfo {membership} author rosterBrokerTs (conflicts, applied) = do forM_ conflicts $ \mid' -> messageWarning $ "x.grp.roster: member key conflict, keeping trusted key, memberId=" <> safeDecodeUtf8 (strEncode mid') forM_ applied $ \(member, fromRole, created) -> - unless created $ createItems member fromRole + unless created $ emitRoleChange member fromRole where - createItems member fromRole = do + emitRoleChange member fromRole = do let toRole = memberRole' member - gEvent = RGEMemberRole (groupMemberId' member) (fromLocalProfile $ memberProfile member) toRole - (gInfo', author', scopeInfo) <- mkGroupChatScope gInfo author - ci <- createChatItem user (CDGroupRcv gInfo' scopeInfo author') False (CIRcvGroupEvent gEvent) Nothing (Just MSSVerified) (Just rosterBrokerTs) - toView $ CEvtNewChatItems user [ci] + (gInfo', author') <- + if sameMemberId (memberId' membership) member + then do + (gInfo', author', scopeInfo) <- mkGroupChatScope gInfo author + ci <- createChatItem user (CDGroupRcv gInfo' scopeInfo author') False (CIRcvGroupEvent $ RGEUserRole toRole) Nothing (Just MSSVerified) (Just rosterBrokerTs) + toView $ CEvtNewChatItems user [ci] + pure (gInfo', author') + else pure (gInfo, author) toView CEvtMemberRole {user, groupInfo = gInfo', byMember = author', member, fromRole, toRole, msgSigned = Just MSSVerified} sendRosterAck :: GroupInfo -> GroupMember -> VersionRoster -> Maybe Text -> CM () @@ -3474,6 +3507,19 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = messageError $ "x.grp.roster.ack: relay could not save roster, marked rejected: " <> e _ -> pure () + -- A relay re-serves the full roster to a subscriber that detected a version gap, but only when its STORED + -- blob is newer than BOTH the requester's version (Nothing = none) and the version it last served this member + -- - the latter bounds reflected amplification (a member can't re-trigger a full serve). Gating on the stored + -- blob (not roster_version, the gate) means the relay serves only a blob the requester will accept. + -- serveRoster records the served version (on all serve paths) and is a no-op without a roster. + xGrpRosterRequest :: GroupInfo -> GroupMember -> Maybe VersionRoster -> CM () + xGrpRosterRequest gInfo m reqVer_ = + when (isUserGrpFwdRelay gInfo) $ do + (stored_, served_) <- withStore' $ \db -> + (,) <$> getStoredRosterVersion db gInfo <*> getMemberRosterServedVersion db m + forM_ stored_ $ \stored -> + when (maybe True (stored >) reqVer_ && maybe True (stored >) served_) $ serveRoster user gInfo m + checkHostRole :: GroupMember -> GroupMemberRole -> CM () checkHostRole GroupMember {memberRole, localDisplayName} memRole = when (memberRole < GRAdmin || memberRole < memRole) $ throwChatError (CEGroupContactRole localDisplayName) @@ -3518,11 +3564,11 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = withStore $ \db -> setMemberVectorRelationConnected db sendingMem refMem MRSubjectConnected withStore $ \db -> setMemberVectorRelationConnected db refMem sendingMem MRReferencedConnected - xGrpMemDel :: GroupInfo -> GroupMember -> MemberId -> Bool -> Maybe VersionRoster -> VerifiedMsg 'Json -> RcvMessage -> UTCTime -> Bool -> CM (Maybe DeliveryJobScope) - xGrpMemDel gInfo@GroupInfo {membership} m@GroupMember {memberRole = senderRole} memId withMessages rosterVer_ verifiedMsg msg@RcvMessage {msgSigned} brokerTs forwarded = do + xGrpMemDel :: GroupInfo -> Maybe GroupMember -> GroupMember -> MemberId -> Bool -> Maybe VersionRoster -> VerifiedMsg 'Json -> RcvMessage -> UTCTime -> Bool -> CM (Maybe DeliveryJobScope) + xGrpMemDel gInfo@GroupInfo {membership} fwdRelay_ m@GroupMember {memberRole = senderRole} memId withMessages rosterVer_ verifiedMsg msg@RcvMessage {msgSigned} brokerTs forwarded = do let GroupMember {memberId = membershipMemId} = membership if membershipMemId == memId - then applyAtRosterVersion gInfo m rosterVer_ $ checkRole membership $ do + then applyAtRosterVersion gInfo fwdRelay_ m rosterVer_ $ checkRole membership $ do deleteGroupLinkIfExists user gInfo -- TODO [relays] possible improvement is to immediately delete rcv queues if isUserGrpFwdRelay unless (isUserGrpFwdRelay gInfo) $ deleteGroupConnections user gInfo False @@ -3534,7 +3580,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = deleteMemberItem msg gInfo RGEUserDeleted toView $ CEvtDeletedMemberUser user gInfo {membership = membership'} m withMessages msgSigned pure $ Just DJSGroup {jobSpec = DJRelayRemoved} - else applyAtRosterVersion gInfo m rosterVer_ $ + else applyAtRosterVersion gInfo fwdRelay_ m rosterVer_ $ withStore' (\db -> runExceptT $ getGroupMemberByMemberId db cxt user gInfo memId) >>= \case Left _ -> do messageError "x.grp.mem.del with unknown member ID" @@ -3800,9 +3846,9 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = XInfo p -> withAuthor XInfo_ $ \author -> void $ xInfoMember gInfo author p rcvMsg msgTs XGrpRelayNew rl -> withAuthor XGrpRelayNew_ $ \author -> void $ xGrpRelayNew gInfo author rl XGrpMemNew memInfo msgScope -> withAuthor XGrpMemNew_ $ \author -> void $ xGrpMemNew gInfo author memInfo msgScope rcvMsg msgTs - XGrpMemRole memId memRole memberKey rosterVer -> withAuthor XGrpMemRole_ $ \author -> void $ xGrpMemRole gInfo author memId memRole memberKey rosterVer rcvMsg msgTs + XGrpMemRole memId memRole memberKey rosterVer -> withAuthor XGrpMemRole_ $ \author -> void $ xGrpMemRole gInfo (Just m) author memId memRole memberKey rosterVer rcvMsg msgTs XGrpMemRestrict memId memRestrictions -> withAuthor XGrpMemRestrict_ $ \author -> void $ xGrpMemRestrict gInfo author memId memRestrictions rcvMsg msgTs - XGrpMemDel memId withMessages rosterVer -> withAuthor XGrpMemDel_ $ \author -> void $ xGrpMemDel gInfo author memId withMessages rosterVer verifiedMsg rcvMsg msgTs True + XGrpMemDel memId withMessages rosterVer -> withAuthor XGrpMemDel_ $ \author -> void $ xGrpMemDel gInfo (Just m) author memId withMessages rosterVer verifiedMsg rcvMsg msgTs True XGrpLeave -> withAuthor XGrpLeave_ $ \author -> void $ xGrpLeave gInfo author rcvMsg msgTs XGrpDel -> withAuthor XGrpDel_ $ \author -> void $ xGrpDel gInfo author rcvMsg msgTs XGrpInfo p' -> withAuthor XGrpInfo_ $ \author -> void $ xGrpInfo gInfo author p' rcvMsg msgTs diff --git a/src/Simplex/Chat/Protocol.hs b/src/Simplex/Chat/Protocol.hs index 223fe492a9..2fb8361fdb 100644 --- a/src/Simplex/Chat/Protocol.hs +++ b/src/Simplex/Chat/Protocol.hs @@ -525,6 +525,7 @@ data ChatMsgEvent (e :: MsgEncoding) where XGrpDirectInv :: ConnReqInvitation -> Maybe MsgContent -> Maybe MsgScope -> ChatMsgEvent 'Json XGrpRoster :: GroupRoster -> ChatMsgEvent 'Json XGrpRosterAck :: VersionRoster -> Maybe Text -> ChatMsgEvent 'Json + XGrpRosterRequest :: Maybe VersionRoster -> ChatMsgEvent 'Json XGrpMsgForward :: GrpMsgForward -> ChatMessage 'Json -> ChatMsgEvent 'Json XInfoProbe :: Probe -> ChatMsgEvent 'Json XInfoProbeCheck :: ProbeHash -> ChatMsgEvent 'Json @@ -1099,6 +1100,7 @@ data CMEventTag (e :: MsgEncoding) where XGrpDirectInv_ :: CMEventTag 'Json XGrpRoster_ :: CMEventTag 'Json XGrpRosterAck_ :: CMEventTag 'Json + XGrpRosterRequest_ :: CMEventTag 'Json XGrpMsgForward_ :: CMEventTag 'Json XInfoProbe_ :: CMEventTag 'Json XInfoProbeCheck_ :: CMEventTag 'Json @@ -1161,6 +1163,7 @@ instance MsgEncodingI e => StrEncoding (CMEventTag e) where XGrpDirectInv_ -> "x.grp.direct.inv" XGrpRoster_ -> "x.grp.roster" XGrpRosterAck_ -> "x.grp.roster.ack" + XGrpRosterRequest_ -> "x.grp.roster.request" XGrpMsgForward_ -> "x.grp.msg.forward" XInfoProbe_ -> "x.info.probe" XInfoProbeCheck_ -> "x.info.probe.check" @@ -1224,6 +1227,7 @@ instance StrEncoding ACMEventTag where "x.grp.direct.inv" -> XGrpDirectInv_ "x.grp.roster" -> XGrpRoster_ "x.grp.roster.ack" -> XGrpRosterAck_ + "x.grp.roster.request" -> XGrpRosterRequest_ "x.grp.msg.forward" -> XGrpMsgForward_ "x.info.probe" -> XInfoProbe_ "x.info.probe.check" -> XInfoProbeCheck_ @@ -1283,6 +1287,7 @@ toCMEventTag msg = case msg of XGrpDirectInv {} -> XGrpDirectInv_ XGrpRoster _ -> XGrpRoster_ XGrpRosterAck {} -> XGrpRosterAck_ + XGrpRosterRequest _ -> XGrpRosterRequest_ XGrpMsgForward {} -> XGrpMsgForward_ XInfoProbe _ -> XInfoProbe_ XInfoProbeCheck _ -> XInfoProbeCheck_ @@ -1444,6 +1449,7 @@ appJsonToCM AppMessageJson {v, msgId, event, params} = do XGrpDirectInv_ -> XGrpDirectInv <$> p "connReq" <*> opt "content" <*> opt "scope" XGrpRoster_ -> XGrpRoster <$> (GroupRoster <$> p "version" <*> p "fileInv") XGrpRosterAck_ -> XGrpRosterAck <$> p "version" <*> opt "error" + XGrpRosterRequest_ -> XGrpRosterRequest <$> opt "version" XGrpMsgForward_ -> do fwdSender <- opt "memberId" >>= \case Just memberId -> FwdMember memberId . fromMaybe "" <$> opt "memberName" @@ -1518,6 +1524,7 @@ chatToAppMessage chatMsg@ChatMessage {chatVRange, msgId, chatMsgEvent} = case en XGrpDirectInv connReq content scope -> o $ ("content" .=? content) $ ("scope" .=? scope) ["connReq" .= connReq] XGrpRoster GroupRoster {version, fileInv} -> o ["version" .= version, "fileInv" .= fileInv] XGrpRosterAck version err -> o $ ("error" .=? err) ["version" .= version] + XGrpRosterRequest version -> o $ ("version" .=? version) [] XGrpMsgForward GrpMsgForward {fwdSender, fwdBrokerTs} msg -> o $ encodeFwdSender fwdSender ["msg" .= msg, "msgTs" .= fwdBrokerTs] where encodeFwdSender = \case diff --git a/src/Simplex/Chat/Store/Groups.hs b/src/Simplex/Chat/Store/Groups.hs index 7ed599007c..70e64f68d7 100644 --- a/src/Simplex/Chat/Store/Groups.hs +++ b/src/Simplex/Chat/Store/Groups.hs @@ -92,7 +92,12 @@ module Simplex.Chat.Store.Groups getPublishableGroupRelays, setGroupRosterVersion, getGroupRosterVersion, - getGroupRoster, + getStoredRosterVersion, + setMemberRosterServedVersion, + getMemberRosterServedVersion, + setCompleteRosterVersion, + getCompleteRosterVersion, + getStoredGroupRoster, RcvRosterTransfer (..), createRosterTransfer, getRosterTransferVersion, @@ -1499,21 +1504,57 @@ getGroupRosterVersion db GroupInfo {groupId} = fmap join . maybeFirstRow fromOnly $ DB.query db "SELECT roster_version FROM groups WHERE group_id = ?" (Only groupId) --- The live roster header a relay re-serves to joiners, with the completed blob served alongside it --- (both are written together at completion, so the blob is present whenever the header is). -getGroupRoster :: DB.Connection -> GroupInfo -> IO (Maybe (GroupMemberId, UTCTime, SignedMsg, Maybe ByteString)) -getGroupRoster db GroupInfo {groupId} = +-- The version of the roster blob actually stored (written with the blob in setGroupLiveRoster), as opposed to +-- roster_version (the acceptance gate). The owner sends the blob before the delta, so these normally match; a +-- failed blob send leaves the gate (advanced by the delta) ahead of the stored blob until a later roster completes. +getStoredRosterVersion :: DB.Connection -> GroupInfo -> IO (Maybe VersionRoster) +getStoredRosterVersion db GroupInfo {groupId} = + fmap join . maybeFirstRow fromOnly $ + DB.query db "SELECT stored_roster_version FROM groups WHERE group_id = ?" (Only groupId) + +-- The newest roster version a relay re-served to this member on its catch-up request: bounds reflected +-- amplification, so a member can't re-trigger a full serve at a version it was already served. +setMemberRosterServedVersion :: DB.Connection -> GroupMember -> VersionRoster -> IO () +setMemberRosterServedVersion db GroupMember {groupMemberId} v = do + currentTs <- getCurrentTime + DB.execute db "UPDATE group_members SET roster_served_version = ?, updated_at = ? WHERE group_member_id = ?" (v, currentTs, groupMemberId) + +getMemberRosterServedVersion :: DB.Connection -> GroupMember -> IO (Maybe VersionRoster) +getMemberRosterServedVersion db GroupMember {groupMemberId} = + fmap join . maybeFirstRow fromOnly $ + DB.query db "SELECT roster_served_version FROM group_members WHERE group_member_id = ?" (Only groupMemberId) + +-- The highest version up to which the subscriber holds a complete, contiguous picture: advances by 1 on a +-- contiguous delta and to the roster's version on a roster apply, but stays put on a gapped delta (so a stuck +-- value re-triggers the catch-up request on every following delta until a roster fills the gap). This is the +-- subscriber's "what I have" for both gap detection and the request - as opposed to roster_version (highest seen, +-- the revert gate) and stored_roster_version (the blob a relay holds). +setCompleteRosterVersion :: DB.Connection -> GroupInfo -> VersionRoster -> IO () +setCompleteRosterVersion db GroupInfo {groupId} v = do + currentTs <- getCurrentTime + DB.execute db "UPDATE groups SET applied_complete_roster_version = ?, updated_at = ? WHERE group_id = ?" (v, currentTs, groupId) + +getCompleteRosterVersion :: DB.Connection -> GroupInfo -> IO (Maybe VersionRoster) +getCompleteRosterVersion db GroupInfo {groupId} = + fmap join . maybeFirstRow fromOnly $ + DB.query db "SELECT applied_complete_roster_version FROM groups WHERE group_id = ?" (Only groupId) + +-- The live roster header a relay re-serves to joiners, with the completed blob and its stored version +-- (all written together at completion, so the blob and version are present whenever the header is). +-- Returns the stored version, not roster_version (the gate), so callers serve/record exactly what they hold. +getStoredGroupRoster :: DB.Connection -> GroupInfo -> IO (Maybe (GroupMemberId, UTCTime, SignedMsg, Maybe ByteString, Maybe VersionRoster)) +getStoredGroupRoster db GroupInfo {groupId} = (>>= toRoster) <$> maybeFirstRow id ( DB.query db - "SELECT roster_sending_owner_gm_id, roster_broker_ts, roster_msg_chat_binding, roster_msg_signatures, roster_msg_body, roster_blob FROM groups WHERE group_id = ?" + "SELECT roster_sending_owner_gm_id, roster_broker_ts, roster_msg_chat_binding, roster_msg_signatures, roster_msg_body, roster_blob, stored_roster_version FROM groups WHERE group_id = ?" (Only groupId) ) where - toRoster (Just ownerGMId, Just brokerTs, Just cb, Just (Binary sigsBs), Just (Binary body), blob_) = - (\sigs -> (ownerGMId, brokerTs, SignedMsg cb sigs body, (\(Binary b) -> b) <$> blob_)) <$> eitherToMaybe (smpDecode sigsBs) + toRoster (Just ownerGMId, Just brokerTs, Just cb, Just (Binary sigsBs), Just (Binary body), blob_, storedVer_) = + (\sigs -> (ownerGMId, brokerTs, SignedMsg cb sigs body, (\(Binary b) -> b) <$> blob_, storedVer_)) <$> eitherToMaybe (smpDecode sigsBs) toRoster _ = Nothing -- A per-source in-flight roster transfer, keyed (group_id, from_member_id): replaces the single @@ -1598,6 +1639,10 @@ getRosterTransfer db fileId = -- Write the single live roster on groups from a completed transfer's values (header NULL on a member, -- so its live roster_msg_* stay NULL and it never re-serves; only relays re-serve). +-- Sets all three versions to the completed blob's version: the gate (roster_version - refuse anything older), +-- the stored version (stored_roster_version - the blob actually held and re-served), and the complete frontier +-- (applied_complete_roster_version - a snapshot makes the picture complete up to its version). Deltas advance the +-- gate always and the complete frontier only when contiguous, so complete <= stored <= roster_version normally. setGroupLiveRoster :: DB.Connection -> GroupInfo -> VersionRoster -> GroupMemberId -> UTCTime -> Maybe SignedMsg -> ByteString -> IO () setGroupLiveRoster db GroupInfo {groupId} v ownerGMId brokerTs sm_ blob = do currentTs <- getCurrentTime @@ -1605,13 +1650,13 @@ setGroupLiveRoster db GroupInfo {groupId} v ownerGMId brokerTs sm_ blob = do db [sql| UPDATE groups SET - roster_version = ?, roster_blob = ?, + roster_version = ?, stored_roster_version = ?, applied_complete_roster_version = ?, roster_blob = ?, roster_sending_owner_gm_id = ?, roster_broker_ts = ?, roster_msg_chat_binding = ?, roster_msg_signatures = ?, roster_msg_body = ?, updated_at = ? WHERE group_id = ? |] - ( (v, Binary blob, ownerGMId, brokerTs) + ( (v, v, v, Binary blob, ownerGMId, brokerTs) :. ((\SignedMsg {chatBinding} -> chatBinding) <$> sm_, (\SignedMsg {signatures} -> Binary (smpEncode signatures)) <$> sm_, (\SignedMsg {signedBody} -> Binary signedBody) <$> sm_, currentTs, groupId) ) diff --git a/src/Simplex/Chat/Store/Postgres/Migrations.hs b/src/Simplex/Chat/Store/Postgres/Migrations.hs index e17129cd1b..a3335640a2 100644 --- a/src/Simplex/Chat/Store/Postgres/Migrations.hs +++ b/src/Simplex/Chat/Store/Postgres/Migrations.hs @@ -39,6 +39,7 @@ import Simplex.Chat.Store.Postgres.Migrations.M20260531_member_removed_at import Simplex.Chat.Store.Postgres.Migrations.M20260601_relay_sent_web_domain import Simplex.Chat.Store.Postgres.Migrations.M20260602_group_roster import Simplex.Chat.Store.Postgres.Migrations.M20260603_simplex_name +import Simplex.Chat.Store.Postgres.Migrations.M20260629_roster_catchup import Simplex.Messaging.Agent.Store.Shared (Migration (..)) schemaMigrations :: [(String, Text, Maybe Text)] @@ -77,7 +78,8 @@ schemaMigrations = ("20260531_member_removed_at", m20260531_member_removed_at, Just down_m20260531_member_removed_at), ("20260601_relay_sent_web_domain", m20260601_relay_sent_web_domain, Just down_m20260601_relay_sent_web_domain), ("20260602_group_roster", m20260602_group_roster, Just down_m20260602_group_roster), - ("20260603_simplex_name", m20260603_simplex_name, Just down_m20260603_simplex_name) + ("20260603_simplex_name", m20260603_simplex_name, Just down_m20260603_simplex_name), + ("20260629_roster_catchup", m20260629_roster_catchup, Just down_m20260629_roster_catchup) ] -- | The list of migrations in ascending order by date diff --git a/src/Simplex/Chat/Store/Postgres/Migrations/M20260629_roster_catchup.hs b/src/Simplex/Chat/Store/Postgres/Migrations/M20260629_roster_catchup.hs new file mode 100644 index 0000000000..f5b94e7d0b --- /dev/null +++ b/src/Simplex/Chat/Store/Postgres/Migrations/M20260629_roster_catchup.hs @@ -0,0 +1,41 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Chat.Store.Postgres.Migrations.M20260629_roster_catchup where + +import Data.Text (Text) +import Text.RawString.QQ (r) + +-- Roster catch-up bookkeeping. Three monotonic per-group roster versions with distinct roles - normally equal, +-- diverging across gaps and failed transfers (applied_complete <= stored <= roster_version): +-- roster_version (added in M20260602) - the GATE: highest version seen. Advanced by every accepted owner delta +-- and by a roster apply. Revert protection: a completing roster older than this is rejected, since its +-- snapshot would undo a newer applied delta. +-- stored_roster_version - the blob HELD: the version of the roster blob actually stored (written with the blob). +-- What a relay can re-serve; a failed blob receive leaves it behind the gate (which the delta still advances) +-- until a later roster completes. Relay-side; on a member it is set but the blob is unused. +-- applied_complete_roster_version - the COMPLETE frontier: highest version up to which the picture is contiguous. +-- Advances by 1 on a contiguous delta and to the roster's version on apply, but stays put on a gapped delta. +-- The subscriber's "what I have" for gap detection and the catch-up request: a value below the gate means +-- missed versions, so each following delta re-asks the forwarding relay until a roster fills the frontier. +-- Also adds group_members.roster_served_version - the newest version a relay re-served a given member, bounding +-- reflected amplification (a member can't re-trigger a full serve at a version it was already served). +-- Backfill an existing roster's stored and complete versions from roster_version: pre-upgrade the picture is +-- contiguous up to roster_version (no gap detection existed), so a fresh NULL frontier would read every group's +-- next delta as a gap and make every subscriber request a re-serve at once. +m20260629_roster_catchup :: Text +m20260629_roster_catchup = + [r| +ALTER TABLE group_members ADD COLUMN roster_served_version BIGINT; +ALTER TABLE groups ADD COLUMN stored_roster_version BIGINT; +ALTER TABLE groups ADD COLUMN applied_complete_roster_version BIGINT; +UPDATE groups SET stored_roster_version = roster_version, applied_complete_roster_version = roster_version WHERE roster_version IS NOT NULL; +|] + +down_m20260629_roster_catchup :: Text +down_m20260629_roster_catchup = + [r| +ALTER TABLE group_members DROP COLUMN roster_served_version; +ALTER TABLE groups DROP COLUMN stored_roster_version; +ALTER TABLE groups DROP COLUMN applied_complete_roster_version; +|] diff --git a/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql b/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql index 2799a79b81..3b8e7530c8 100644 --- a/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql +++ b/src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql @@ -834,7 +834,8 @@ CREATE TABLE test_chat_schema.group_members ( member_relations_vector bytea, relay_link bytea, member_pub_key bytea, - removed_at timestamp with time zone + removed_at timestamp with time zone, + roster_served_version bigint ); @@ -994,7 +995,9 @@ CREATE TABLE test_chat_schema.groups ( roster_sending_owner_gm_id bigint, roster_broker_ts timestamp with time zone, roster_blob bytea, - group_domain_verified smallint + group_domain_verified smallint, + stored_roster_version bigint, + applied_complete_roster_version bigint ); diff --git a/src/Simplex/Chat/Store/SQLite/Migrations.hs b/src/Simplex/Chat/Store/SQLite/Migrations.hs index 0cf78db595..d135c0f742 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations.hs +++ b/src/Simplex/Chat/Store/SQLite/Migrations.hs @@ -162,6 +162,7 @@ import Simplex.Chat.Store.SQLite.Migrations.M20260531_member_removed_at import Simplex.Chat.Store.SQLite.Migrations.M20260601_relay_sent_web_domain import Simplex.Chat.Store.SQLite.Migrations.M20260602_group_roster import Simplex.Chat.Store.SQLite.Migrations.M20260603_simplex_name +import Simplex.Chat.Store.SQLite.Migrations.M20260629_roster_catchup import Simplex.Messaging.Agent.Store.Shared (Migration (..)) schemaMigrations :: [(String, Query, Maybe Query)] @@ -323,7 +324,8 @@ schemaMigrations = ("20260531_member_removed_at", m20260531_member_removed_at, Just down_m20260531_member_removed_at), ("20260601_relay_sent_web_domain", m20260601_relay_sent_web_domain, Just down_m20260601_relay_sent_web_domain), ("20260602_group_roster", m20260602_group_roster, Just down_m20260602_group_roster), - ("20260603_simplex_name", m20260603_simplex_name, Just down_m20260603_simplex_name) + ("20260603_simplex_name", m20260603_simplex_name, Just down_m20260603_simplex_name), + ("20260629_roster_catchup", m20260629_roster_catchup, Just down_m20260629_roster_catchup) ] -- | The list of migrations in ascending order by date diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/M20260629_roster_catchup.hs b/src/Simplex/Chat/Store/SQLite/Migrations/M20260629_roster_catchup.hs new file mode 100644 index 0000000000..1dacc665ea --- /dev/null +++ b/src/Simplex/Chat/Store/SQLite/Migrations/M20260629_roster_catchup.hs @@ -0,0 +1,40 @@ +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Chat.Store.SQLite.Migrations.M20260629_roster_catchup where + +import Database.SQLite.Simple (Query) +import Database.SQLite.Simple.QQ (sql) + +-- Roster catch-up bookkeeping. Three monotonic per-group roster versions with distinct roles - normally equal, +-- diverging across gaps and failed transfers (applied_complete <= stored <= roster_version): +-- roster_version (added in M20260602) - the GATE: highest version seen. Advanced by every accepted owner delta +-- and by a roster apply. Revert protection: a completing roster older than this is rejected, since its +-- snapshot would undo a newer applied delta. +-- stored_roster_version - the blob HELD: the version of the roster blob actually stored (written with the blob). +-- What a relay can re-serve; a failed blob receive leaves it behind the gate (which the delta still advances) +-- until a later roster completes. Relay-side; on a member it is set but the blob is unused. +-- applied_complete_roster_version - the COMPLETE frontier: highest version up to which the picture is contiguous. +-- Advances by 1 on a contiguous delta and to the roster's version on apply, but stays put on a gapped delta. +-- The subscriber's "what I have" for gap detection and the catch-up request: a value below the gate means +-- missed versions, so each following delta re-asks the forwarding relay until a roster fills the frontier. +-- Also adds group_members.roster_served_version - the newest version a relay re-served a given member, bounding +-- reflected amplification (a member can't re-trigger a full serve at a version it was already served). +-- Backfill an existing roster's stored and complete versions from roster_version: pre-upgrade the picture is +-- contiguous up to roster_version (no gap detection existed), so a fresh NULL frontier would read every group's +-- next delta as a gap and make every subscriber request a re-serve at once. +m20260629_roster_catchup :: Query +m20260629_roster_catchup = + [sql| +ALTER TABLE group_members ADD COLUMN roster_served_version INTEGER; +ALTER TABLE groups ADD COLUMN stored_roster_version INTEGER; +ALTER TABLE groups ADD COLUMN applied_complete_roster_version INTEGER; +UPDATE groups SET stored_roster_version = roster_version, applied_complete_roster_version = roster_version WHERE roster_version IS NOT NULL; +|] + +down_m20260629_roster_catchup :: Query +down_m20260629_roster_catchup = + [sql| +ALTER TABLE group_members DROP COLUMN roster_served_version; +ALTER TABLE groups DROP COLUMN stored_roster_version; +ALTER TABLE groups DROP COLUMN applied_complete_roster_version; +|] diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/chat_query_plans.txt b/src/Simplex/Chat/Store/SQLite/Migrations/chat_query_plans.txt index af2fb0162d..5d7b56fc8f 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations/chat_query_plans.txt +++ b/src/Simplex/Chat/Store/SQLite/Migrations/chat_query_plans.txt @@ -5370,7 +5370,7 @@ SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) Query: UPDATE groups SET - roster_version = ?, roster_blob = ?, + roster_version = ?, stored_roster_version = ?, applied_complete_roster_version = ?, roster_blob = ?, roster_sending_owner_gm_id = ?, roster_broker_ts = ?, roster_msg_chat_binding = ?, roster_msg_signatures = ?, roster_msg_body = ?, updated_at = ? @@ -6665,6 +6665,36 @@ SEARCH rcv_files USING INTEGER PRIMARY KEY (rowid=?) SEARCH snd_files USING COVERING INDEX idx_snd_files_file_id (file_id=?) SEARCH files USING COVERING INDEX idx_files_redirect_file_id (redirect_file_id=?) +Query: DELETE FROM group_members WHERE member_id = ? +Plan: +SCAN group_members +SEARCH rcv_roster_transfers USING COVERING INDEX idx_rcv_roster_transfers_from_member_id (from_member_id=?) +SEARCH group_relays USING COVERING INDEX idx_group_relays_group_member_id (group_member_id=?) +SEARCH delivery_jobs USING COVERING INDEX idx_delivery_jobs_job_scope_support_gm_id (job_scope_support_gm_id=?) +SEARCH delivery_tasks USING COVERING INDEX idx_delivery_tasks_sender_group_member_id (sender_group_member_id=?) +SEARCH delivery_tasks USING COVERING INDEX idx_delivery_tasks_job_scope_support_gm_id (job_scope_support_gm_id=?) +SEARCH received_probes USING COVERING INDEX idx_received_probes_group_member_id (group_member_id=?) +SEARCH sent_probe_hashes USING COVERING INDEX idx_sent_probe_hashes_group_member_id (group_member_id=?) +SEARCH sent_probes USING COVERING INDEX idx_sent_probes_group_member_id (group_member_id=?) +SEARCH group_snd_item_statuses USING COVERING INDEX idx_group_snd_item_statuses_group_member_id (group_member_id=?) +SEARCH chat_item_moderations USING COVERING INDEX idx_chat_item_moderations_moderator_member_id (moderator_member_id=?) +SEARCH chat_item_reactions USING COVERING INDEX idx_chat_item_reactions_group_member_id (group_member_id=?) +SEARCH chat_items USING COVERING INDEX idx_chat_items_group_scope_group_member_id (group_scope_group_member_id=?) +SEARCH chat_items USING COVERING INDEX idx_chat_items_forwarded_by_group_member_id (forwarded_by_group_member_id=?) +SEARCH chat_items USING COVERING INDEX idx_chat_items_item_deleted_by_group_member_id (item_deleted_by_group_member_id=?) +SEARCH chat_items USING COVERING INDEX idx_chat_items_group_member_id (group_member_id=?) +SEARCH pending_group_messages USING COVERING INDEX idx_pending_group_messages_group_member_id (group_member_id=?) +SEARCH messages USING COVERING INDEX idx_messages_forwarded_by_group_member_id (forwarded_by_group_member_id=?) +SEARCH messages USING COVERING INDEX idx_messages_author_group_member_id (author_group_member_id=?) +SEARCH connections USING COVERING INDEX idx_connections_group_member_id (group_member_id=?) +SEARCH rcv_files USING COVERING INDEX idx_rcv_files_group_member_id (group_member_id=?) +SEARCH snd_files USING COVERING INDEX idx_snd_files_group_member_id (group_member_id=?) +SEARCH group_member_intros USING COVERING INDEX idx_group_member_intros_to_group_member_id (to_group_member_id=?) +SEARCH group_member_intros USING COVERING INDEX idx_group_member_intros_re_group_member_id (re_group_member_id=?) +SEARCH group_members USING COVERING INDEX idx_group_members_invited_by_group_member_id (invited_by_group_member_id=?) +SEARCH contacts USING COVERING INDEX idx_contacts_grp_direct_inv_from_group_member_id (grp_direct_inv_from_group_member_id=?) +SEARCH contacts USING COVERING INDEX idx_contacts_contact_group_member_id (contact_group_member_id=?) + Query: DELETE FROM group_members WHERE user_id = ? AND group_id = ? Plan: SEARCH group_members USING COVERING INDEX idx_group_members_group_id (user_id=? AND group_id=?) @@ -7076,6 +7106,10 @@ Query: SELECT app_settings FROM app_settings Plan: SCAN app_settings +Query: SELECT applied_complete_roster_version FROM groups WHERE group_id = ? +Plan: +SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) + Query: SELECT auth_err_counter FROM connections WHERE user_id = ? AND connection_id = ? Plan: SEARCH connections USING INTEGER PRIMARY KEY (rowid=?) @@ -7285,6 +7319,10 @@ Query: SELECT max(active_order) FROM users Plan: SEARCH users +Query: SELECT member_id FROM group_members WHERE local_display_name = ? +Plan: +SCAN group_members + Query: SELECT member_id FROM group_members WHERE member_role = ? LIMIT 1 Plan: SCAN group_members @@ -7313,6 +7351,10 @@ Query: SELECT member_role, member_pub_key FROM group_members WHERE local_display Plan: SCAN group_members +Query: SELECT member_role, member_pub_key FROM group_members WHERE member_id = ? +Plan: +SCAN group_members + Query: SELECT member_status FROM group_members WHERE local_display_name = ? Plan: SCAN group_members @@ -7361,10 +7403,14 @@ Query: SELECT roster_blob FROM groups WHERE roster_blob IS NOT NULL Plan: SCAN groups -Query: SELECT roster_sending_owner_gm_id, roster_broker_ts, roster_msg_chat_binding, roster_msg_signatures, roster_msg_body, roster_blob FROM groups WHERE group_id = ? +Query: SELECT roster_sending_owner_gm_id, roster_broker_ts, roster_msg_chat_binding, roster_msg_signatures, roster_msg_body, roster_blob, stored_roster_version FROM groups WHERE group_id = ? Plan: SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) +Query: SELECT roster_served_version FROM group_members WHERE group_member_id = ? +Plan: +SEARCH group_members USING INTEGER PRIMARY KEY (rowid=?) + Query: SELECT roster_transfer_id FROM rcv_roster_transfers WHERE group_id = ? AND from_member_id = ? Plan: SEARCH rcv_roster_transfers USING COVERING INDEX idx_rcv_roster_transfers_group_id_from_member_id (group_id=? AND from_member_id=?) @@ -7389,6 +7435,10 @@ Query: SELECT should_sync FROM connections_sync WHERE connections_sync_id = 1 Plan: SEARCH connections_sync USING INTEGER PRIMARY KEY (rowid=?) +Query: SELECT stored_roster_version FROM groups WHERE group_id = ? +Plan: +SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) + Query: SELECT summary_current_members_count FROM groups WHERE group_id = ? Plan: SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) @@ -7633,6 +7683,10 @@ Query: UPDATE group_members SET member_role = ?, member_pub_key = NULL WHERE loc Plan: SCAN group_members +Query: UPDATE group_members SET roster_served_version = ?, updated_at = ? WHERE group_member_id = ? +Plan: +SEARCH group_members USING INTEGER PRIMARY KEY (rowid=?) + Query: UPDATE group_members SET support_chat_items_member_attention = ?, updated_at = ? WHERE group_member_id = ? Plan: SEARCH group_members USING INTEGER PRIMARY KEY (rowid=?) @@ -7653,6 +7707,14 @@ Query: UPDATE group_relays SET relay_status = ?, updated_at = ? WHERE group_rela Plan: SEARCH group_relays USING INTEGER PRIMARY KEY (rowid=?) +Query: UPDATE groups SET applied_complete_roster_version = ? WHERE group_id = ? +Plan: +SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) + +Query: UPDATE groups SET applied_complete_roster_version = ?, updated_at = ? WHERE group_id = ? +Plan: +SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) + Query: UPDATE groups SET business_member_id = ?, customer_member_id = ? WHERE group_id = ? Plan: SEARCH groups USING INTEGER PRIMARY KEY (rowid=?) diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql b/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql index 954660dce5..01d53ad000 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql +++ b/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql @@ -204,7 +204,9 @@ CREATE TABLE groups( roster_sending_owner_gm_id INTEGER, roster_broker_ts TEXT, roster_blob BLOB, - group_domain_verified INTEGER, -- received + group_domain_verified INTEGER, + stored_roster_version INTEGER, + applied_complete_roster_version INTEGER, -- received FOREIGN KEY(user_id, local_display_name) REFERENCES display_names(user_id, local_display_name) ON DELETE CASCADE @@ -249,6 +251,7 @@ CREATE TABLE group_members( relay_link BLOB, member_pub_key BLOB, removed_at TEXT, + roster_served_version INTEGER, FOREIGN KEY(user_id, local_display_name) REFERENCES display_names(user_id, local_display_name) ON DELETE CASCADE diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index 51fe8c54aa..18023ed12c 100644 --- a/tests/ChatTests/Groups.hs +++ b/tests/ChatTests/Groups.hs @@ -300,6 +300,8 @@ chatGroupTests = do it "concurrent fresh invitations both rejected" testRelayRejectRaceConcurrentInvitations describe "promoted members roster" $ do it "moderator action verifies via owner-signed roster" testChannelModeratorActionViaRoster + it "subscriber recovers a missed roster member after a version gap" testChannelSubscriberRosterCatchUp + it "2 relays: subscriber recovers a missed roster member after a version gap" testChannel2RelaysSubscriberRosterCatchUp it "removed moderator drops from the roster cache" testChannelRemovedModeratorRefreshesRoster it "role transitions update the roster (mod <-> admin, admin -> non-roster)" testChannelRoleTransitionsUpdateRoster it "malicious relay cannot downgrade or re-key a roster-established moderator via XGrpMemNew" testChannelRelayCannotDowngradeRosterMember @@ -9710,6 +9712,24 @@ checkMemberRow cc name expectedRole = do DB.query db "SELECT member_role FROM group_members WHERE local_display_name = ?" (Only name) :: IO [Only T.Text] map (\(Only r) -> r) roles `shouldBe` maybeToList expectedRole +-- The wire member id for a named member (look it up on a client that knows the name, e.g. the owner), used to +-- find a member by id on a subscriber that only knows it by member-id hash (e.g. after roster recovery). +getMemberIdByName :: TestCC -> T.Text -> IO ByteString +getMemberIdByName cc name = do + rows <- withCCTransaction cc $ \db -> + DB.query db "SELECT member_id FROM group_members WHERE local_display_name = ?" (Only name) :: IO [Only ByteString] + case rows of + [Only mid] -> pure mid + _ -> fail $ "expected one group_members row for " <> T.unpack name + +getMemberRoleKey :: TestCC -> ByteString -> IO (T.Text, Maybe ByteString) +getMemberRoleKey cc mid = do + rows <- withCCTransaction cc $ \db -> + DB.query db "SELECT member_role, member_pub_key FROM group_members WHERE member_id = ?" (Only mid) :: IO [(T.Text, Maybe ByteString)] + case rows of + [r] -> pure r + _ -> fail "expected one group_members row by member id" + testChannelModeratorActionViaRoster :: HasCallStack => TestParams -> IO () testChannelModeratorActionViaRoster ps = withNewTestChat ps "alice" aliceProfile $ \alice -> @@ -9780,6 +9800,99 @@ testChannelModeratorActionViaRoster ps = DB.query db "SELECT member_role FROM group_members WHERE local_display_name = ?" (Only name) :: IO [Only T.Text] map (\(Only r) -> r) roles `shouldBe` [expectedRole] +testChannelSubscriberRosterCatchUp :: HasCallStack => TestParams -> IO () +testChannelSubscriberRosterCatchUp ps = + withNewTestChat ps "alice" aliceProfile $ \alice -> + withNewTestChatOpts ps relayTestOpts "bob" bobProfile $ \bob -> + withNewTestChat ps "cath" cathProfile $ \cath -> + withNewTestChat ps "dan" danProfile $ \dan -> + withNewTestChat ps "eve" eveProfile $ \eve -> + withNewTestChat ps "frank" frankProfile $ \frank -> do + (shortLink, fullLink) <- prepareChannel1Relay "team" alice bob + forM_ [cath, dan, eve, frank] $ \member -> + memberJoinChannel "team" [bob] [alice] shortLink fullLink member + -- promote dan (roster v0) then eve (v1) into the owner-signed roster; cath learns both with their keys + threadDelay 1000000 + promoteChannelMember "team" alice bob dan [cath, eve, frank] + threadDelay 1000000 + promoteChannelMember "team" alice bob eve [cath, dan, frank] + threadDelay 1000000 + -- simulate cath having fallen behind and lost dan: capture dan's member id (from the owner, which + -- knows the name) and cath's owner-pinned key for dan, then delete dan's record and rewind cath's + -- applied frontier so the next delta arrives as a gap (v2 > applied 0 + 1) + danId <- getMemberIdByName alice "dan" + (_, danKey) <- getMemberRoleKey cath danId + withCCTransaction cath $ \db -> do + DB.execute db "DELETE FROM group_members WHERE member_id = ?" (Only danId) + DB.execute db "UPDATE groups SET applied_complete_roster_version = ? WHERE group_id = ?" (0 :: Int64, 1 :: Int64) + -- the next privileged change (frank -> v2) reaches cath at a jumped version, triggering catch-up: + -- cath requests the roster from the forwarding relay, which re-serves the current snapshot + promoteChannelMember "team" alice bob frank [cath, dan, eve] + threadDelay 2000000 -- wait for the gap request + relay re-serve to recover dan + -- cath recovered dan from the re-served roster: same member id, role, and owner-pinned key + (recRole, recKey) <- getMemberRoleKey cath danId + recRole `shouldBe` "member" + recKey `shouldBe` danKey + +-- Same recovery, but the subscriber (frank) is connected to two relays: the request goes to whichever relay +-- forwarded the gapping delta, and only an observation that catch-up works in a 2-relay channel (not the race). +testChannel2RelaysSubscriberRosterCatchUp :: HasCallStack => TestParams -> IO () +testChannel2RelaysSubscriberRosterCatchUp ps = + withNewTestChat ps "alice" aliceProfile $ \alice -> + withNewTestChatOpts ps relayTestOpts "bob" bobProfile $ \bob -> + withNewTestChatOpts ps relayTestOpts "cath" cathProfile $ \cath -> + withNewTestChat ps "dan" danProfile $ \dan -> + withNewTestChat ps "eve" eveProfile $ \eve -> + withNewTestChat ps "frank" frankProfile $ \frank -> do + (shortLink, fullLink) <- prepareChannel2Relays "team" alice bob cath + forM_ [dan, eve, frank] $ \member -> + memberJoinChannel "team" [bob, cath] [alice] shortLink fullLink member + -- promote dan (v0) then eve (v1) into the owner-signed roster, forwarded by both relays; frank + -- (the subscriber) learns both with their keys + threadDelay 1000000 + alice ##> "/mr #team dan member" + alice <## "#team: you changed the role of dan to member (signed)" + concurrentlyN_ + [ bob <## "#team: alice changed the role of dan from observer to member (signed)", + cath <## "#team: alice changed the role of dan from observer to member (signed)", + dan <## "#team: alice changed your role from observer to member (signed)", + eve <### [EndsWith "from observer to member (signed)"], + frank <### [EndsWith "from observer to member (signed)"] + ] + threadDelay 1000000 + alice ##> "/mr #team eve member" + alice <## "#team: you changed the role of eve to member (signed)" + concurrentlyN_ + [ bob <## "#team: alice changed the role of eve from observer to member (signed)", + cath <## "#team: alice changed the role of eve from observer to member (signed)", + eve <## "#team: alice changed your role from observer to member (signed)", + dan <### [EndsWith "from observer to member (signed)"], + frank <### [EndsWith "from observer to member (signed)"] + ] + threadDelay 1000000 + -- simulate frank having fallen behind and lost dan: delete dan's record and rewind frank's complete + -- frontier so the next delta (eve -> v2) arrives as a gap (2 > applied 0 + 1) + danId <- getMemberIdByName alice "dan" + (_, danKey) <- getMemberRoleKey frank danId + withCCTransaction frank $ \db -> do + DB.execute db "DELETE FROM group_members WHERE member_id = ?" (Only danId) + DB.execute db "UPDATE groups SET applied_complete_roster_version = ? WHERE group_id = ?" (0 :: Int64, 1 :: Int64) + -- eve -> moderator (v2) reaches frank at a jumped version; it requests the roster from the relay that + -- forwarded the delta, which re-serves the current snapshot (including dan), recovering dan + alice ##> "/mr #team eve moderator" + alice <## "#team: you changed the role of eve to moderator (signed)" + concurrentlyN_ + [ bob <## "#team: alice changed the role of eve from member to moderator (signed)", + cath <## "#team: alice changed the role of eve from member to moderator (signed)", + eve <## "#team: alice changed your role from member to moderator (signed)", + dan <### [EndsWith "from member to moderator (signed)"], + frank <### [EndsWith "from member to moderator (signed)"] + ] + threadDelay 2000000 -- wait for the gap request + relay re-serve to recover dan + (recRole, recKey) <- getMemberRoleKey frank danId + recRole `shouldBe` "member" + recKey `shouldBe` danKey + testChannelRemovedModeratorRefreshesRoster :: HasCallStack => TestParams -> IO () testChannelRemovedModeratorRefreshesRoster ps = withNewTestChat ps "alice" aliceProfile $ \alice -> @@ -9804,6 +9917,8 @@ testChannelRemovedModeratorRefreshesRoster ps = threadDelay 1000000 alice ##> "/rm #team cath" alice <## "#team: you removed cath from the group (signed)" + -- the relay applies the removal via the roster (revert to observer) before the delete delta + bob <## "#team: alice changed the role of cath from moderator to observer (signed)" bob <## "#team: alice removed cath from the group (signed)" cath <## "#team: alice removed you from the group (signed)" cath <## "use /d #team to delete the group" @@ -10024,6 +10139,8 @@ testChannelRemoveMemberSigned ps = threadDelay 1000000 alice ##> "/rm #team eve" alice <## "#team: you removed eve from the group (signed)" + -- the relay applies the removal via the roster (revert to observer) before the delete delta + bob <## "#team: alice changed the role of eve from member to observer (signed)" bob <## "#team: alice removed eve from the group (signed)" concurrentlyN_ [ cath <## "#team: alice removed eve from the group (signed)",