From 051727ffa052adb10d1593a49acec6f87a4c526d Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Tue, 30 Jun 2026 16:49:54 +0400 Subject: [PATCH] wip --- simplex-chat.cabal | 4 +- src/Simplex/Chat/Library/Subscriber.hs | 46 ++++++++++--------- src/Simplex/Chat/Store/Groups.hs | 29 +++++++++--- src/Simplex/Chat/Store/Postgres/Migrations.hs | 4 +- .../M20260629_member_roster_served.hs | 21 --------- .../Migrations/M20260629_roster_catchup.hs | 37 +++++++++++++++ src/Simplex/Chat/Store/SQLite/Migrations.hs | 4 +- .../M20260629_member_roster_served.hs | 20 -------- .../Migrations/M20260629_roster_catchup.hs | 36 +++++++++++++++ tests/ChatTests/Groups.hs | 4 +- 10 files changed, 129 insertions(+), 76 deletions(-) delete mode 100644 src/Simplex/Chat/Store/Postgres/Migrations/M20260629_member_roster_served.hs create mode 100644 src/Simplex/Chat/Store/Postgres/Migrations/M20260629_roster_catchup.hs delete mode 100644 src/Simplex/Chat/Store/SQLite/Migrations/M20260629_member_roster_served.hs create mode 100644 src/Simplex/Chat/Store/SQLite/Migrations/M20260629_roster_catchup.hs diff --git a/simplex-chat.cabal b/simplex-chat.cabal index 2577c86deb..8927856f50 100644 --- a/simplex-chat.cabal +++ b/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: diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index 09f8539fc2..23f918dcab 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/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 diff --git a/src/Simplex/Chat/Store/Groups.hs b/src/Simplex/Chat/Store/Groups.hs index 661a0a4967..1f97612955 100644 --- a/src/Simplex/Chat/Store/Groups.hs +++ b/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) ) diff --git a/src/Simplex/Chat/Store/Postgres/Migrations.hs b/src/Simplex/Chat/Store/Postgres/Migrations.hs index 842a390d5a..b736f08731 100644 --- a/src/Simplex/Chat/Store/Postgres/Migrations.hs +++ b/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 diff --git a/src/Simplex/Chat/Store/Postgres/Migrations/M20260629_member_roster_served.hs b/src/Simplex/Chat/Store/Postgres/Migrations/M20260629_member_roster_served.hs deleted file mode 100644 index 45b3c73bc5..0000000000 --- a/src/Simplex/Chat/Store/Postgres/Migrations/M20260629_member_roster_served.hs +++ /dev/null @@ -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; -|] 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..18bd2e5fd3 --- /dev/null +++ b/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; +|] diff --git a/src/Simplex/Chat/Store/SQLite/Migrations.hs b/src/Simplex/Chat/Store/SQLite/Migrations.hs index 64ec77d320..a0f40d3554 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations.hs +++ b/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 diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/M20260629_member_roster_served.hs b/src/Simplex/Chat/Store/SQLite/Migrations/M20260629_member_roster_served.hs deleted file mode 100644 index 49caffcdcb..0000000000 --- a/src/Simplex/Chat/Store/SQLite/Migrations/M20260629_member_roster_served.hs +++ /dev/null @@ -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; -|] 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..35182cf746 --- /dev/null +++ b/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; +|] diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index 95bf8b2640..7ec4bd9659 100644 --- a/tests/ChatTests/Groups.hs +++ b/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]