git.iko.soy/skrepka
on master help sign in
Batch federation ack message IDs to respect maxMessageIds limit
ac8c87fiko
server.knot+26−17
compare with master
unifiedsplit
@@ -541,7 +541,31 @@ retryPendingForwards = \pending -> do
541 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
541 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
542 yield {}
542 yield {}
543
543
544+sendBatchedAcksToServer : [{messageId: Text, toKey: Text, originServer: Text, | a}] -> Text -> IO {network} {}
545+sendBatchedAcksToServer = \serverAcks server ->
546+ if count serverAcks == 0
547+ then yield {}
548+ else do
549+ let batch = fold (\acc a ->
550+ if count acc >= maxMessageIds then acc
551+ else union acc [a]) [] serverAcks
552+ let rest = diff serverAcks batch
553+ result <- fetch (federationScheme ++ server) (FederationAck {
554+ authorization: "Bearer " ++ federationSecret,
555+ messageIds: (do a <- batch; yield a.messageId),
556+ originServer: server, fromServer: serverHostN})
557+ case result of
558+ Ok _ ->
559+ atomic do
560+ ackedIds <- *ackedIds
561+ set *ackedIds = do
562+ a <- ackedIds
563+ where not (a.originServer == server && count (do sa <- batch; where sa.messageId == a.messageId; where sa.toKey == a.toKey; yield sa) > 0)
564+ yield a
565+ Err _ -> yield {}
566+ sendBatchedAcksToServer rest server
567+
544−sendAcksAndClean : [{messageId: Text, originServer: Text, | a}] -> [Text] -> IO {network} {}
568+sendAcksAndClean : [{messageId: Text, toKey: Text, originServer: Text, | a}] -> [Text] -> IO {network} {}
545sendAcksAndClean = \pendingAcks servers ->
569sendAcksAndClean = \pendingAcks servers ->
546 if count servers == 0
570 if count servers == 0
547 then yield {}
571 then yield {}
@@ -551,22 +575,7 @@ sendAcksAndClean = \pendingAcks servers ->
551 if server == ""
575 if server == ""
552 then yield {}
576 then yield {}
553 else do
577 else do
554− fork (do
555− let serverAcks = do a <- pendingAcks; where a.originServer == server; yield a
578+ fork (sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
556− result <- fetch (federationScheme ++ server) (FederationAck {
557− authorization: "Bearer " ++ federationSecret,
558− messageIds: (do a <- serverAcks; yield a.messageId),
559− originServer: server, fromServer: serverHostN})
560− case result of
561− Ok _ ->
562− atomic do
563− ackedIds <- *ackedIds
564− set *ackedIds = do
565− a <- ackedIds
566− where not (a.originServer == server && count (do sa <- serverAcks; where sa.messageId == a.messageId; where sa.toKey == a.toKey; yield sa) > 0)
567− yield a
568− Err _ -> yield {}
569− )
570 sendAcksAndClean pendingAcks rest
579 sendAcksAndClean pendingAcks rest
571
580
572pollOnce : Text -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}
581pollOnce : Text -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}