Skip to content

Commit fafca76

Browse files
blackheavenblackheaven
committed
WPB-28483: remove serial per-user round trips from MLS commit-bundle path
POST /v17/mls/commit-bundles spent ~800ms on sequential I/O for a 7-user internal_add join. Remove the avoidable round trips without changing protocol semantics: - getCommitData fetches all pending proposals once; commit proposal refs are dereferenced from that list (derefOrCheckProposalFrom) instead of one CQL point-read per ref, and checkReferences reuses the same list instead of re-scanning the partition. - incrementEpochNoRead skips the full conversation re-read on the internal-commit path, where the result is discarded. - brig client-data lookups in checkClients fan out concurrently (getClientData is total; validation stays serial so Error-effect failures still abort the whole commit exactly as before). - addMLSClients logged batches run concurrently (disjoint rows, children kept Error-free so failures propagate as IO exceptions). - Local welcome pushes are fire-and-forget via pushNotificationAsync; delivery is already asynchronous downstream of gundeck.
1 parent 2c2fedb commit fafca76

9 files changed

Lines changed: 102 additions & 54 deletions

File tree

‎changelog.d/5-internal/WPB-28483‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
MLS commit-bundles are processed with less sequential I/O: proposal references are resolved from a single store read, client and client-store updates fan out concurrently, and welcome pushes no longer block the response.

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/Federation.hs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ import Galley.Types.Error
4141
import Imports
4242
import Network.Wai.Utilities.Exception
4343
import Polysemy
44+
import Polysemy.Async (Async)
4445
import Polysemy.Error
4546
import Polysemy.Input
4647
import Polysemy.Internal.Kind (Append)
@@ -603,7 +604,8 @@ sendMLSCommitBundle ::
603604
Member TeamCollaboratorsSubsystem r,
604605
Member E.MLSCommitLockStore r,
605606
Member FeaturesConfigSubsystem r,
606-
Member (Input ConversationSubsystemConfig) r
607+
Member (Input ConversationSubsystemConfig) r,
608+
Member Async r
607609
) =>
608610
Domain ->
609611
MLSMessageSendRequest ->
@@ -891,8 +893,6 @@ onMLSMessageSent domain rmm =
891893
mlsSendWelcome ::
892894
( Member (Error InternalError) r,
893895
Member NotificationSubsystem r,
894-
Member ExternalAccess r,
895-
Member P.TinyLog r,
896896
Member (Input (Maybe (MLSKeysByPurpose MLSPrivateKeys))) r,
897897
Member (Input (Local ())) r,
898898
Member Now r

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/Interpreter.hs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import Data.Qualified
2929
import Imports
3030
import Network.Wai.Utilities.JSONResponse (JSONResponse)
3131
import Polysemy
32+
import Polysemy.Async (Async)
3233
import Polysemy.Error
3334
import Polysemy.Input
3435
import Polysemy.Resource (Resource)
@@ -123,7 +124,8 @@ interpretConversationSubsystem ::
123124
Member (Input (Maybe (MLSKeysByPurpose MLSPrivateKeys))) r,
124125
Member UserClientIndexStore r,
125126
Member (Input FanoutLimit) r,
126-
Member TinyLog r
127+
Member TinyLog r,
128+
Member Async r
127129
) =>
128130
InterpreterFor ConversationSubsystem r
129131
interpretConversationSubsystem = interpret $ \case

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/CheckClients.hs‎

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@ import Data.Set qualified as Set
3131
import Data.Tuple.Extra
3232
import Imports
3333
import Polysemy
34+
import Polysemy.Async (Async)
35+
import Polysemy.Async qualified as P
3436
import Polysemy.Error
3537
import Wire.API.Error
3638
import Wire.API.Error.Galley
@@ -50,7 +52,8 @@ checkClients ::
5052
Member (FederationAPIAccess FederatorClient) r,
5153
Member (ErrorS MLSClientMismatch) r,
5254
Member (ErrorS MLSIdentityMismatch) r,
53-
Member (Error MLSProtocolError) r
55+
Member (Error MLSProtocolError) r,
56+
Member Async r
5457
) =>
5558
Local ConvOrSubConv ->
5659
CipherSuiteTag ->
@@ -59,9 +62,20 @@ checkClients ::
5962
checkClients lConvOrSub ciphersuite newCM = do
6063
let convOrSub = tUnqualified lConvOrSub
6164
cm = convOrSub.members
62-
fmap catMaybes . forM (Map.assocs (unClientMap newCM)) $
63-
\(qtarget, newclients) -> do
64-
mClientData <- getClientData lConvOrSub ciphersuite qtarget
65+
assocs = Map.assocs (unClientMap newCM)
66+
-- Fetch client data from brig concurrently. getClientData is total: it
67+
-- reports failures via Maybe, never via an Error effect, so parallel
68+
-- results cannot be silently swallowed by sequenceConcurrently.
69+
-- Validation below runs serially so that Error-effect throws abort the
70+
-- whole commit exactly as in the fully serial implementation.
71+
-- getClientData is Error-free, so the outer 'Maybe' that
72+
-- sequenceConcurrently attaches to every child result is always 'Just'.
73+
mClientDatas <-
74+
fmap (map (fromMaybe Nothing)) . P.sequenceConcurrently $
75+
flip fmap assocs $ \(qtarget, _) ->
76+
getClientData lConvOrSub ciphersuite qtarget
77+
fmap catMaybes . forM (zip assocs mClientDatas) $
78+
\((qtarget, newclients), mClientData) -> do
6579
unreachable <- case (mClientData, cmLookup qtarget cm) of
6680
-- user is already present, skip check in this case
6781
(_, Just existingClients) -> do
@@ -103,9 +117,9 @@ checkClients lConvOrSub ciphersuite newCM = do
103117
pure False
104118

105119
-- Check that new leaf nodes are using the registered signature keys.
106-
for_ mClientData $ \clientData ->
120+
for_ mClientData $ \cd ->
107121
for_ (Map.assocs newclients) $ \(cid, (_, mKp)) ->
108-
checkSignatureKey (fmap (.leafNode) mKp) (Map.lookup cid clientData.infoMap)
122+
checkSignatureKey (fmap (.leafNode) mKp) (Map.lookup cid cd.infoMap)
109123

110124
pure $ guard unreachable $> qtarget
111125

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/Core.hs‎

Lines changed: 33 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
module Wire.ConversationSubsystem.MLS.Commit.Core
1919
( getCommitData,
2020
incrementEpoch,
21+
incrementEpochNoRead,
2122
getClientInfo,
2223
getSingleClientInfo,
2324
checkSignatureKey,
@@ -70,7 +71,7 @@ import Wire.ExternalAccess
7071
import Wire.FederationAPIAccess
7172
import Wire.LegalHoldStore (LegalHoldStore)
7273
import Wire.NotificationSubsystem
73-
import Wire.ProposalStore (ProposalStore)
74+
import Wire.ProposalStore (StoredProposal, ProposalStore, getAllPendingProposals)
7475
import Wire.Sem.Now (Now)
7576
import Wire.Sem.Random (Random)
7677
import Wire.TeamCollaboratorsSubsystem
@@ -114,22 +115,27 @@ getCommitData ::
114115
Epoch ->
115116
CipherSuiteTag ->
116117
IncomingBundle ->
117-
Sem r (IndexMap, ProposalAction)
118+
Sem r (IndexMap, ProposalAction, [StoredProposal])
118119
getCommitData senderIdentity lConvOrSub epoch ciphersuite bundle = do
119120
let convOrSub = tUnqualified lConvOrSub
120121
groupId = cnvmlsGroupId convOrSub.mlsMeta
121122

122-
runState convOrSub.indexMap $ do
123-
creatorAction <-
124-
if epoch == Epoch 0
125-
then addProposedClient (Left . RegularClient $ senderIdentity.client)
126-
else mempty
127-
proposals <-
128-
traverse
129-
(derefOrCheckProposal epoch ciphersuite groupId)
130-
bundle.commit.value.proposals
131-
action <- applyProposals ciphersuite proposals
132-
pure (creatorAction <> action)
123+
-- Fetch all pending proposals once: used both for dereferencing commit
124+
-- proposal refs and by checkReferences downstream.
125+
storedProposals <- getAllPendingProposals groupId epoch
126+
(creatorAction, action) <-
127+
runState convOrSub.indexMap $ do
128+
creatorAction <-
129+
if epoch == Epoch 0
130+
then addProposedClient (Left . RegularClient $ senderIdentity.client)
131+
else mempty
132+
proposals <-
133+
traverse
134+
(derefOrCheckProposalFrom storedProposals epoch ciphersuite groupId)
135+
bundle.commit.value.proposals
136+
action <- applyProposals ciphersuite proposals
137+
pure (creatorAction <> action)
138+
pure (creatorAction, action, storedProposals)
133139

134140
incrementEpoch ::
135141
( Member ConversationStore r,
@@ -149,6 +155,20 @@ incrementEpoch (SubConv c s) = do
149155
getSubConversation (mcId c) (scSubConvId s) >>= noteS @'ConvNotFound
150156
pure (SubConv c subconv)
151157

158+
-- | Bump the MLS epoch without re-reading the conversation afterwards.
159+
-- Use when the caller discards the result; avoids 2-3 CQL round trips.
160+
incrementEpochNoRead ::
161+
( Member ConversationStore r
162+
) =>
163+
ConvOrSubConv ->
164+
Sem r ()
165+
incrementEpochNoRead =
166+
\case
167+
Conv c ->
168+
setConversationEpoch (mcId c) (succ (cnvmlsEpoch (mcMLSData c)))
169+
SubConv _c s ->
170+
setSubConversationEpoch (scParentConvId s) (scSubConvId s) (succ (cnvmlsEpoch (scMLSData s)))
171+
152172
getClientInfo ::
153173
( Member BrigAPIAccess r,
154174
Member (FederationAPIAccess FederatorClient) r,

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Commit/InternalCommit.hs‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,8 @@ import Galley.Types.Error
3333
import Imports
3434
import Polysemy
3535
import Polysemy.Error
36+
import Polysemy.Async qualified as P
37+
import Polysemy.Async (Async)
3638
import Polysemy.Input (Input)
3739
import Polysemy.Resource (Resource)
3840
import Wire.API.Conversation hiding (Member)
@@ -78,6 +80,7 @@ processInternalCommit ::
7880
Member (ErrorS 'MissingLegalholdConsent) r,
7981
Member (ErrorS 'GroupIdVersionNotSupported) r,
8082
Member Resource r,
83+
Member Async r,
8184
Member Random r,
8285
Member (ErrorS MLSInvalidLeafNodeSignature) r,
8386
Member MLSCommitLockStore r,
@@ -93,14 +96,14 @@ processInternalCommit ::
9396
Epoch ->
9497
ProposalAction ->
9598
Commit ->
99+
[StoredProposal] ->
96100
Codensity (Sem r) [LocalConversationUpdate]
97-
processInternalCommit senderIdentity con lConvOrSub ciphersuite ciphersuiteUpdate epoch action commit = do
101+
processInternalCommit senderIdentity con lConvOrSub ciphersuite ciphersuiteUpdate epoch action commit storedProposals = do
98102
let convOrSub = tUnqualified lConvOrSub
99103
qusr = cidQualifiedUser senderIdentity.client
100104
cm = convOrSub.members
101105
newUserClients = Map.assocs (unClientMap (paAdd action))
102-
103-
lift $ checkReferences convOrSub epoch commit
106+
lift $ checkReferences storedProposals commit
104107

105108
-- check update path
106109
lift $ traverse_ (checkUpdatePath lConvOrSub senderIdentity ciphersuite) commit.path
@@ -244,9 +247,14 @@ processInternalCommit senderIdentity con lConvOrSub ciphersuite ciphersuiteUpdat
244247
removeMLSClients gid qtarget (Map.keysSet clients)
245248

246249
-- add clients to the conversation state
247-
for_ newUserClients $ \(qtarget, newClients) -> do
248-
addMLSClients gid qtarget $
249-
Set.fromList [(cid, idx) | (cid, (idx, _)) <- Map.assocs newClients]
250+
-- Note: safe to run concurrently because the children only perform store
251+
-- writes on disjoint rows; their failures surface as IO exceptions, which
252+
-- sequenceConcurrently propagates. If an Error-effect throw is ever added
253+
-- here, it would be swallowed — keep children Error-free.
254+
void . P.sequenceConcurrently $
255+
flip fmap newUserClients $ \(qtarget, newClients) ->
256+
addMLSClients gid qtarget $
257+
Set.fromList [(cid, idx) | (cid, (idx, _)) <- Map.assocs newClients]
250258

251259
for_ action.paHistoryClientAdd $ uncurry (addHistoryClient gid)
252260

@@ -256,9 +264,8 @@ processInternalCommit senderIdentity con lConvOrSub ciphersuite ciphersuiteUpdat
256264
when ciphersuiteUpdate $ case convOrSub.id of
257265
Conv cid -> setConversationCipherSuite cid ciphersuite
258266
SubConv cid sub -> setSubConversationCipherSuite cid sub ciphersuite
259-
260267
-- increment epoch number
261-
for_ lConvOrSub incrementEpoch
268+
for_ lConvOrSub incrementEpochNoRead
262269

263270
pure events
264271

@@ -330,12 +337,10 @@ existingMembers :: Local StoredConversation -> Set (Qualified UserId)
330337
existingMembers lconv = existingLocalMembers lconv <> existingRemoteMembers lconv
331338

332339
checkReferences ::
333-
( Member ProposalStore r,
334-
Member (ErrorS MLSCommitMissingReferences) r
340+
( Member (ErrorS MLSCommitMissingReferences) r
335341
) =>
336-
ConvOrSubConv -> Epoch -> Commit -> Sem r ()
337-
checkReferences convOrSub epoch commit = do
338-
allPendingProposals <- getAllPendingProposals (cnvmlsGroupId convOrSub.mlsMeta) epoch
342+
[StoredProposal] -> Commit -> Sem r ()
343+
checkReferences allPendingProposals commit = do
339344
let referencedProposals = Set.fromList $ mapMaybe (\x -> preview _Ref x) commit.proposals
340345
let (includedProposals, missingProposals) =
341346
partition

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Message.hs‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ import Data.Tuple.Extra
4343
import Galley.Types.Error
4444
import Imports
4545
import Polysemy
46+
import Polysemy.Async (Async)
4647
import Polysemy.Error
4748
import Polysemy.Input
4849
import Polysemy.Output
@@ -183,7 +184,8 @@ postMLSCommitBundle ::
183184
Member FederationSubsystem r,
184185
Member TeamSubsystem r,
185186
Member (Input ConversationSubsystemConfig) r,
186-
Member FeaturesConfigSubsystem r
187+
Member FeaturesConfigSubsystem r,
188+
Member Async r
187189
) =>
188190
Local x ->
189191
Qualified UserId ->
@@ -220,7 +222,8 @@ postMLSCommitBundleFromLocalUser ::
220222
Member FederationSubsystem r,
221223
Member TeamSubsystem r,
222224
Member (Input ConversationSubsystemConfig) r,
223-
Member FeaturesConfigSubsystem r
225+
Member FeaturesConfigSubsystem r,
226+
Member Async r
224227
) =>
225228
Version ->
226229
Local UserId ->
@@ -257,7 +260,8 @@ postMLSCommitBundleToLocalConv ::
257260
Member FederationSubsystem r,
258261
Member TeamSubsystem r,
259262
Member (Input ConversationSubsystemConfig) r,
260-
Member FeaturesConfigSubsystem r
263+
Member FeaturesConfigSubsystem r,
264+
Member Async r
261265
) =>
262266
Qualified UserId ->
263267
ClientId ->
@@ -325,7 +329,7 @@ postMLSCommitBundleToLocalConv qusr c conn bundle ctype lConvOrSubId = do
325329
(events, newClients) <- case senderIdentity.index of
326330
Just _ -> do
327331
-- extract added/removed clients from bundle
328-
(newIndexMap, action) <-
332+
(newIndexMap, action, storedProposals) <-
329333
lift $
330334
getCommitData senderIdentity lConvOrSub bundle.epoch ciphersuite bundle
331335

@@ -358,6 +362,7 @@ postMLSCommitBundleToLocalConv qusr c conn bundle ctype lConvOrSubId = do
358362
bundle.epoch
359363
action
360364
bundle.commit.value
365+
storedProposals
361366
-- the sender client is included in the Add action on the first commit,
362367
-- but it doesn't need to get a welcome message, so we filter it out here
363368
let newClients = cmRemoveClient senderIdentity.client (paAdd action)

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Proposal.hs‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
module Wire.ConversationSubsystem.MLS.Proposal
1919
( -- * Proposal processing
20-
derefOrCheckProposal,
20+
derefOrCheckProposalFrom,
2121
checkProposal,
2222
processProposal,
2323
proposalProcessingStage,
@@ -145,24 +145,26 @@ type HasProposalEffects r =
145145
Member TeamCollaboratorsSubsystem r
146146
)
147147

148-
derefOrCheckProposal ::
148+
-- | Dereference a commit proposal, looking up refs in a prefetched list of
149+
-- pending proposals instead of issuing one point-read per ref.
150+
derefOrCheckProposalFrom ::
149151
( Member (Error MLSProtocolError) r,
150152
Member (ErrorS 'MLSInvalidLeafNodeIndex) r,
151153
Member (ErrorS 'MLSUnsupportedProposal) r,
152-
Member ProposalStore r,
153154
Member (State IndexMap) r,
154155
Member (ErrorS 'MLSProposalNotFound) r,
155156
Member (ErrorS 'MLSInvalidLeafNodeSignature) r
156157
) =>
158+
[StoredProposal] ->
157159
Epoch ->
158160
CipherSuiteTag ->
159161
GroupId ->
160162
ProposalOrRef ->
161163
Sem r Proposal
162-
derefOrCheckProposal epoch _ciphersuite groupId (Ref ref) = do
163-
p <- getProposal groupId epoch ref >>= noteS @'MLSProposalNotFound
164-
pure p.value
165-
derefOrCheckProposal _epoch ciphersuite _ (Inline p) = do
164+
derefOrCheckProposalFrom stored _epoch _ciphersuite _groupId (Ref ref) =
165+
noteS @'MLSProposalNotFound $
166+
(.proposal.value) <$> find ((== ref) . (.ref)) stored
167+
derefOrCheckProposalFrom _stored _epoch ciphersuite _ (Inline p) = do
166168
im <- get
167169
checkProposal ciphersuite im p
168170
pure p

‎libs/wire-subsystems/src/Wire/ConversationSubsystem/MLS/Welcome.hs‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -48,15 +48,13 @@ import Wire.API.MLS.SubConversation
4848
import Wire.API.MLS.Welcome
4949
import Wire.API.Message
5050
import Wire.API.Push.V2 (RecipientClients (..))
51-
import Wire.ExternalAccess
5251
import Wire.FederationAPIAccess
5352
import Wire.NotificationSubsystem
5453
import Wire.Sem.Now (Now)
5554
import Wire.Sem.Now qualified as Now
5655

5756
sendWelcomes ::
5857
( Member (FederationAPIAccess FederatorClient) r,
59-
Member ExternalAccess r,
6058
Member P.TinyLog r,
6159
Member Now r,
6260
Member NotificationSubsystem r
@@ -79,9 +77,7 @@ sendWelcomes loc qusr con cids welcome = do
7977
convFrom (SubConv c _) = c
8078

8179
sendLocalWelcomes ::
82-
( Member P.TinyLog r,
83-
Member ExternalAccess r,
84-
Member NotificationSubsystem r
80+
( Member NotificationSubsystem r
8581
) =>
8682
Qualified ConvId ->
8783
Qualified UserId ->
@@ -100,8 +96,11 @@ sendLocalWelcomes qcnv qusr con now welcome lclients = do
10096
mempty
10197
$ tUnqualified lclients
10298
let e = Event qcnv Nothing (EventFromUser qusr) now Nothing $ EdMLSWelcome welcome.raw
103-
runMessagePush lclients (Just qcnv) $
104-
newMessagePush mempty con defMessageMetadata rcpts e
99+
-- Fire-and-forget: delivery is asynchronous downstream of gundeck anyway;
100+
-- blocking the commit-bundle response on the fan-out is wasted latency.
101+
void $
102+
pushNotificationAsync
103+
(toPush (newMessagePush mempty con defMessageMetadata rcpts e))
105104

106105
sendRemoteWelcomes ::
107106
( Member (FederationAPIAccess FederatorClient) r,

0 commit comments

Comments
 (0)