db2d141 · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement backend improvements: idempotent sends, challenge capacity limits, faster session auth, poll race fix, and federation error logging

db2d141iko6 months ago

client/ApiClient.fs+1−1

compare with master

unifiedsplit

@@ -94,7 +94,7 @@ module ApiClient =

94 }

96 let sendMessage (serverUrl: string) (token: string) (toHex: string) (blobHex: string) (timestamp: int64) =

97− let body = JsonSerializer.Serialize({| ``to`` = toHex; encryptedBlob = blobHex; timestamp = timestamp |})

97+ let body = JsonSerializer.Serialize({| ``to`` = toHex; encryptedBlob = blobHex; timestamp = timestamp; messageId = "" |})

98 async {

99 use! doc = sendRequest client $"{serverUrl}/messages" body (Some token)

100 match doc.RootElement.GetProperty("status").GetString() with

server.knot+66−52

@@ -113,7 +113,7 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times

113type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}

114type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}

115data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}

116−data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]}

116+data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]} | SendDuplicate {}

117data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}

118data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}

119data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}

@@ -159,7 +159,7 @@ route ClientApi where

159 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}

160 = Poll

161 /messages

162− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} / headers {authorization: Text}

162+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}

163 -> {status: MessageStatus, messageId: OptionalMessageId}

164 = SendMessage

165 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}

@@ -396,7 +396,7 @@ isBadServerName = \sn ->

396

397authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

398authenticateSession = \sessions token t ->

399− case headOf (do s <- sessions; where s.expiresAt > t && safeCompare s.token token; yield s) of

399+ case headOf (do s <- sessions; where s.expiresAt > t && s.token == token; yield s) of

400 Nothing {} -> Unauthorized {}

401 Just {value: s} -> Authorized {pubkey: s.pubkey}

402

@@ -603,10 +603,12 @@ forEachBatch = \items batchSize process ->

603sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}

604sendBatchedGossipToServer = \server allEvents hops ->

605 forEachBatch allEvents maxGossipEvents (\batch -> do

606− _ <- fetch (federationUrl server) (RecvGossip {

606+ result <- fetch (federationUrl server) (RecvGossip {

607 authorization: federationAuth,

608 events: batch, hopsRemaining: hops, fromServer: serverHostN})

609+ case result of

610+ Err _ -> println ("federation: gossip to " ++ server ++ " failed")

609− yield {})

611+ Ok _ -> yield {})

610

612

611sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}

613sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}

612sendBatchedGossip = \peerServers allEvents hops ->

614sendBatchedGossip = \peerServers allEvents hops ->

@@ -673,7 +675,9 @@ retryForwardsForServer = \entries remaining ->

673 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

675 toKey: f.toKey, encryptedBlob: f.encryptedBlob,

674 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,

676 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,

675 authorization: federationAuth})

677 authorization: federationAuth})

678+ case result of

679+ Err _ -> println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)

676− whenOk result (removeForward f.messageId f.toServer)

680+ Ok _ -> removeForward f.messageId f.toServer

677 retryForwardsForServer rest (remaining - 1)

681 retryForwardsForServer rest (remaining - 1)

678

682

679retryPendingForwards : [ForwardEntry] -> IO {network} {}

683retryPendingForwards : [ForwardEntry] -> IO {network} {}

@@ -709,10 +713,12 @@ sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originSer

709sendBatchedAcksToServer = \serverAcks server ->

713sendBatchedAcksToServer = \serverAcks server ->

710 forEachBatch serverAcks maxMessageIds (\batch -> do

714 forEachBatch serverAcks maxMessageIds (\batch -> do

711 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

715 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

716+ case result of

717+ Err _ -> println ("federation: ack to " ++ server ++ " failed")

712− whenOk result (atomic do

718+ Ok _ -> atomic do

713− ackedIds <- *ackedIds

719+ ackedIds <- *ackedIds

714− let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch

720+ let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch

715− set *ackedIds = updatedAckedIds))

721+ set *ackedIds = updatedAckedIds)

716

722

717sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}

723sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}

718sendAcksAndClean = \pendingAcks ->

724sendAcksAndClean = \pendingAcks ->

@@ -753,7 +759,7 @@ pruneAndRetry = \t -> do

753 retryPendingForwards result.pendingForwards

759 retryPendingForwards result.pendingForwards

754 sendAcksAndClean result.pendingAcks)

760 sendAcksAndClean result.pendingAcks)

755

761

756−pollOnce = \token cursor startWakeup -> do

762+pollOnce = \token cursor -> do

757 t <- now

763 t <- now

758 sessions <- *sessions

764 sessions <- *sessions

759 let auth = authenticateSession sessions token t

765 let auth = authenticateSession sessions token t

@@ -762,6 +768,7 @@ pollOnce = \token cursor startWakeup -> do

762 Authorized {pubkey} -> do

768 Authorized {pubkey} -> do

763 seqCounter <- *seqCounter

769 seqCounter <- *seqCounter

764 let clampedCursor = clamp 0 (max seqCounter t) cursor

770 let clampedCursor = clamp 0 (max seqCounter t) cursor

771+ startWakeup <- *pollWakeup

765 atomic do

772 atomic do

766 wakeup <- *pollWakeup

773 wakeup <- *pollWakeup

767 allMessages <- *messages

774 allMessages <- *messages

@@ -885,17 +892,19 @@ serve = \req -> case req of

885 ReqChallenge {pubkey, respond} -> do

892 ReqChallenge {pubkey, respond} -> do

886 challenge <- genToken

893 challenge <- genToken

887 t <- now

894 t <- now

888− pruneResult <- atomic do

895+ result <- atomic do

889 pr <- maybePrune t

896 pr <- maybePrune t

890 challenges <- *challenges

897 challenges <- *challenges

891− let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges

898+ let perKeyCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t

892− let updatedChallenges = union

899+ if count perKeyCapped >= maxChallenges

893− capped

900+ then yield {pr: pr, accepted: False {}}

901+ else do

894− [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

902+ set *challenges = union perKeyCapped [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

895− set *challenges = updatedChallenges

903+ yield {pr: pr, accepted: True {}}

896− yield pr

904+ when (not result.accepted)

905+ (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

897− forkBroadcastOffline t pruneResult

906+ forkBroadcastOffline t result.pr

898− yield (respond {challenge: challenge})

907+ yield (respond {challenge: if result.accepted then challenge else ""})

899

908

900 VerifyAuth {pubkey, challenge, signature, respond} -> do

909 VerifyAuth {pubkey, challenge, signature, respond} -> do

901 t <- now

910 t <- now

@@ -937,19 +946,18 @@ serve = \req -> case req of

937 yield (respond {token: token, expiresAt: expiry})

946 yield (respond {token: token, expiresAt: expiry})

938

947

939 Poll {cursor, authorization, respond} -> do

948 Poll {cursor, authorization, respond} -> do

940− startWakeup <- *pollWakeup

941 result <- case extractToken authorization of

949 result <- case extractToken authorization of

942 Nothing {} -> yield (PollUnauthorized {})

950 Nothing {} -> yield (PollUnauthorized {})

943− Just {value: token} -> pollOnce token cursor startWakeup

951+ Just {value: token} -> pollOnce token cursor

944 tAfterPoll <- now

952 tAfterPoll <- now

945 maybePruneAndBroadcast tAfterPoll

953 maybePruneAndBroadcast tAfterPoll

946 yield (respond (case result of

954 yield (respond (case result of

947 PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}

955 PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}

948 PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))

956 PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))

949

957

950− SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do

958+ SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do

951 t <- now

959 t <- now

952− msgId <- genToken

960+ effectiveMsgId <- if messageId != "" then yield messageId else genToken

953 let clampedTs = clampTimestamp timestamp t

961 let clampedTs = clampTimestamp timestamp t

954 result <- atomic do

962 result <- atomic do

955 pr <- maybePrune t

963 pr <- maybePrune t

@@ -957,30 +965,33 @@ serve = \req -> case req of

957 (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})

965 (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})

958 (\_ -> do

966 (\_ -> do

959 messages <- *messages

967 messages <- *messages

968+ ackedIds <- *ackedIds

969+ if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)

970+ then yield {pr: pr, send: SendDuplicate {}}

960− if countMessagesFor messages to t >= maxMessagesPerRecipient

971+ else if countMessagesFor messages to t >= maxMessagesPerRecipient

961− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}

972+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}}

962− else do

973+ else do

963− presence <- *presence

974+ presence <- *presence

964− seqCounter <- *seqCounter

975+ seqCounter <- *seqCounter

965− let seqTs = nextSeqTs t seqCounter

976+ let seqTs = nextSeqTs t seqCounter

966− set *seqCounter = seqTs

977+ set *seqCounter = seqTs

967− set *messages = union messages

978+ set *messages = union messages

968− [{messageId: msgId, toKey: to,

979+ [{messageId: effectiveMsgId, toKey: to,

969− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]

980+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]

970− let toPresence = presenceForKey presence to t

981+ let toPresence = presenceForKey presence to t

971− let localOnline = hasLocalPresence toPresence

982+ let localOnline = hasLocalPresence toPresence

972− let remoteServers = collectPeerServers toPresence (\p -> p.server)

983+ let remoteServers = collectPeerServers toPresence (\p -> p.server)

973− forwards <- *forwards

984+ forwards <- *forwards

974− let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers

985+ let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers

975− let shouldForward = federationEnabled && count forwardServers > 0

986+ let shouldForward = federationEnabled && count forwardServers > 0

976− let delivery = if localOnline then DsDelivered {}

987+ let delivery = if localOnline then DsDelivered {}

977− else if shouldForward then DsFederated {}

988+ else if shouldForward then DsFederated {}

978− else DsQueued {}

989+ else DsQueued {}

979− let updatedForwards = if shouldForward

990+ let updatedForwards = if shouldForward

980− then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})

991+ then union forwards (do s <- forwardServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})

981− else forwards

992+ else forwards

982− set *forwards = updatedForwards

993+ set *forwards = updatedForwards

983− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})

994+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})

984 case result.send of

995 case result.send of

985 SendDone {delivery, remoteServers, forwardServers} ->

996 SendDone {delivery, remoteServers, forwardServers} ->

986 fork (do

997 fork (do

@@ -988,17 +999,20 @@ serve = \req -> case req of

988 sendToEach remoteServers (\server -> do

999 sendToEach remoteServers (\server -> do

989 fwdResult <- fetch (federationUrl server) (ForwardMessage {

1000 fwdResult <- fetch (federationUrl server) (ForwardMessage {

990 toKey: to, encryptedBlob: encryptedBlob,

1001 toKey: to, encryptedBlob: encryptedBlob,

991− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

1002+ timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

992 authorization: federationAuth})

1003 authorization: federationAuth})

1004+ case fwdResult of

1005+ Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)

993− when (any forwardServers (\s -> s == server))

1006+ Ok _ -> when (any forwardServers (\s -> s == server))

994− (whenOk fwdResult (removeForward msgId server)))

1007+ (removeForward effectiveMsgId server))

995 when (isQueued delivery)

1008 when (isQueued delivery)

996 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))

1009 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))

997 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

1010 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

998 _ -> forkBroadcastOffline t result.pr

1011 _ -> forkBroadcastOffline t result.pr

999 yield (case result.send of

1012 yield (case result.send of

1000 SendFailed {status} -> respond {status: status, messageId: ""}

1013 SendFailed {status} -> respond {status: status, messageId: ""}

1014+ SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}

1001− SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: msgId})

1015+ SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})

1002

1016

1003 AckMessages {messageIds, authorization, respond} -> do

1017 AckMessages {messageIds, authorization, respond} -> do

1004 t <- now

1018 t <- now