wip

simplex-chat.cabal

@@ -144,7 +144,7 @@ library
           Simplex.Chat.Store.Postgres.Migrations.M20260531_member_removed_at
           Simplex.Chat.Store.Postgres.Migrations.M20260601_relay_sent_web_domain
           Simplex.Chat.Store.Postgres.Migrations.M20260602_group_roster
-          Simplex.Chat.Store.Postgres.Migrations.M20260629_member_roster_served
+          Simplex.Chat.Store.Postgres.Migrations.M20260629_roster_catchup
   else
       exposed-modules:
           Simplex.Chat.Archive
@@ -307,7 +307,7 @@ library
           Simplex.Chat.Store.SQLite.Migrations.M20260531_member_removed_at
           Simplex.Chat.Store.SQLite.Migrations.M20260601_relay_sent_web_domain
           Simplex.Chat.Store.SQLite.Migrations.M20260602_group_roster
-          Simplex.Chat.Store.SQLite.Migrations.M20260629_member_roster_served
+          Simplex.Chat.Store.SQLite.Migrations.M20260629_roster_catchup
   other-modules:
       Paths_simplex_chat
   hs-source-dirs:

src/Simplex/Chat/Library/Subscriber.hs

@@ -3252,30 +3252,34 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage =
           Nothing -> action
           Just _ | memberRole' sender /= GROwner -> action
           Just v -> do
-            (accept, cur) <- withStore' $ \db -> do
-              cur <- getGroupRosterVersion db gInfo
-              let fresh = maybe True (v >=) cur
-              when fresh $ setGroupRosterVersion db gInfo v
-              pure (fresh, cur)
+            (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 complete frontier only when this delta is the next version (no gap)
+                when (maybe False (\(VersionRoster c) -> v == VersionRoster (c + 1)) prevComplete) $
+                  setCompleteRosterVersion db gInfo v
+              pure (fresh, prevComplete)
             if accept
-              then (requestRosterOnGap gInfo fwdRelay_ cur v `catchAllErrors` eToView) >> action
+              then (requestRosterOnGap v prevComplete `catchAllErrors` eToView) >> action
               else messageWarning "x.grp.mem: roster version not newer than current, ignoring" $> Nothing
-
-    -- A subscriber that skipped versions (no roster yet, or v above cur+1) asks the relay that forwarded this
-    -- delta to re-serve the full roster, recovering the privileged set and keys carried by the missed versions.
-    -- The request carries the subscriber's current version (Nothing if none); the relay serves only what it holds
-    -- newer. Best-effort: a failed request must not block applying the delta. Relays and the direct path don't
-    -- request (fwdRelay_ is Nothing), and a relay that predates roster support is skipped.
-    requestRosterOnGap :: GroupInfo -> Maybe GroupMember -> Maybe VersionRoster -> VersionRoster -> CM ()
-    requestRosterOnGap gInfo fwdRelay_ cur_ v
-      | isUserGrpFwdRelay gInfo = pure ()
-      | otherwise = case fwdRelay_ of
-          Just relay
-            | gap, relay `supportsVersion` groupRosterVersion ->
-                void $ sendGroupMessage' user gInfo [relay] (XGrpRosterRequest cur_)
-          _ -> pure ()
       where
-        gap = maybe True (\(VersionRoster c) -> v > VersionRoster (c + 1)) cur_
+        -- 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 = maybe True (\(VersionRoster c) -> v > VersionRoster (c + 1)) prevComplete
 
     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

src/Simplex/Chat/Store/Groups.hs

@@ -93,6 +93,8 @@ module Simplex.Chat.Store.Groups
     getStoredRosterVersion,
     setMemberRosterServedVersion,
     getMemberRosterServedVersion,
+    setCompleteRosterVersion,
+    getCompleteRosterVersion,
     getStoredGroupRoster,
     RcvRosterTransfer (..),
     createRosterTransfer,
@@ -1501,6 +1503,21 @@ 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.
@@ -1601,10 +1618,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 BOTH versions: a completed blob advances the gate (roster_version - refuse anything older, the downgrade
--- protection for the no-delta join/new-relay/re-serve paths, where the blob is the only thing that sets the gate)
--- and records the stored version (stored_roster_version - what is actually held and re-served). Deltas advance
--- only the gate, so stored_roster_version <= roster_version always.
+-- 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
@@ -1612,13 +1629,13 @@ setGroupLiveRoster db GroupInfo {groupId} v ownerGMId brokerTs sm_ blob = do
     db
     [sql|
       UPDATE groups SET
-        roster_version = ?, stored_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, 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)
     )
 

src/Simplex/Chat/Store/Postgres/Migrations.hs

@@ -38,7 +38,7 @@ import Simplex.Chat.Store.Postgres.Migrations.M20260530_client_services
 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.M20260629_member_roster_served
+import Simplex.Chat.Store.Postgres.Migrations.M20260629_roster_catchup
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Text, Maybe Text)]
@@ -77,7 +77,7 @@ 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),
-    ("20260629_member_roster_served", m20260629_member_roster_served, Just down_m20260629_member_roster_served)
+    ("20260629_roster_catchup", m20260629_roster_catchup, Just down_m20260629_roster_catchup)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Chat/Store/Postgres/Migrations/M20260629_member_roster_served.hs

@@ -1,21 +0,0 @@
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE QuasiQuotes #-}
-
-module Simplex.Chat.Store.Postgres.Migrations.M20260629_member_roster_served where
-
-import Data.Text (Text)
-import Text.RawString.QQ (r)
-
-m20260629_member_roster_served :: Text
-m20260629_member_roster_served =
-  [r|
-ALTER TABLE group_members ADD COLUMN roster_served_version BIGINT;
-ALTER TABLE groups ADD COLUMN stored_roster_version BIGINT;
-|]
-
-down_m20260629_member_roster_served :: Text
-down_m20260629_member_roster_served =
-  [r|
-ALTER TABLE group_members DROP COLUMN roster_served_version;
-ALTER TABLE groups DROP COLUMN stored_roster_version;
-|]

src/Simplex/Chat/Store/Postgres/Migrations/M20260629_roster_catchup.hs

@@ -0,0 +1,37 @@
+{-# 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).
+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;
+|]
+
+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;
+|]

src/Simplex/Chat/Store/SQLite/Migrations.hs

@@ -161,7 +161,7 @@ import Simplex.Chat.Store.SQLite.Migrations.M20260530_client_services
 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.M20260629_member_roster_served
+import Simplex.Chat.Store.SQLite.Migrations.M20260629_roster_catchup
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Query, Maybe Query)]
@@ -323,7 +323,7 @@ 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),
-    ("20260629_member_roster_served", m20260629_member_roster_served, Just down_m20260629_member_roster_served)
+    ("20260629_roster_catchup", m20260629_roster_catchup, Just down_m20260629_roster_catchup)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Chat/Store/SQLite/Migrations/M20260629_member_roster_served.hs

@@ -1,20 +0,0 @@
-{-# LANGUAGE QuasiQuotes #-}
-
-module Simplex.Chat.Store.SQLite.Migrations.M20260629_member_roster_served where
-
-import Database.SQLite.Simple (Query)
-import Database.SQLite.Simple.QQ (sql)
-
-m20260629_member_roster_served :: Query
-m20260629_member_roster_served =
-  [sql|
-ALTER TABLE group_members ADD COLUMN roster_served_version INTEGER;
-ALTER TABLE groups ADD COLUMN stored_roster_version INTEGER;
-|]
-
-down_m20260629_member_roster_served :: Query
-down_m20260629_member_roster_served =
-  [sql|
-ALTER TABLE group_members DROP COLUMN roster_served_version;
-ALTER TABLE groups DROP COLUMN stored_roster_version;
-|]

src/Simplex/Chat/Store/SQLite/Migrations/M20260629_roster_catchup.hs

@@ -0,0 +1,36 @@
+{-# 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).
+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;
+|]
+
+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;
+|]

tests/ChatTests/Groups.hs

@@ -9800,12 +9800,12 @@ testChannelSubscriberRosterCatchUp ps =
               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
-              -- roster_version so the next delta arrives as a gap (v2 > 0+1)
+              -- applied frontier so the next delta arrives as a gap (v2 > applied 0 + 1)
               danId <- memberId alice "dan"
               (_, danKey) <- roleKeyById cath danId
               withCCTransaction cath $ \db -> do
                 DB.execute db "DELETE FROM group_members WHERE member_id = ?" (Only danId)
-                DB.execute db "UPDATE groups SET roster_version = ? WHERE group_id = ?" (0 :: Int64, 1 :: Int64)
+                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]