@@ -113,7 +113,7 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
113113type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
114114type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
115115data 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 {}
117117data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
118118data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
119119data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}
@@ -159,7 +159,7 @@ route ClientApi where
159159 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
160160 = Poll
161161 /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}
163163 -> {status: MessageStatus, messageId: OptionalMessageId}
164164 = SendMessage
165165 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}
@@ -396,7 +396,7 @@ isBadServerName = \sn ->
396396
397397authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
398398authenticateSession = \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
400400 Nothing {} -> Unauthorized {}
401401 Just {value: s} -> Authorized {pubkey: s.pubkey}
402402
@@ -603,10 +603,12 @@ forEachBatch = \items batchSize process ->
603603sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
604604sendBatchedGossipToServer = \server allEvents hops ->
605605 forEachBatch allEvents maxGossipEvents (\batch -> do
606− _ <- fetch (federationUrl server) (RecvGossip {
606+ result <- fetch (federationUrl server) (RecvGossip {
607607 authorization: federationAuth,
608608 events: batch, hopsRemaining: hops, fromServer: serverHostN})
609− yield {})
609+ case result of
610+ Err _ -> println ("federation: gossip to " ++ server ++ " failed")
611+ Ok _ -> yield {})
610612
611613sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
612614sendBatchedGossip = \peerServers allEvents hops ->
@@ -673,7 +675,9 @@ retryForwardsForServer = \entries remaining ->
673675 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
674676 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
675677 authorization: federationAuth})
676− whenOk result (removeForward f.messageId f.toServer)
678+ case result of
679+ Err _ -> println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)
680+ Ok _ -> removeForward f.messageId f.toServer
677681 retryForwardsForServer rest (remaining - 1)
678682
679683retryPendingForwards : [ForwardEntry] -> IO {network} {}
@@ -709,10 +713,12 @@ sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originSer
709713sendBatchedAcksToServer = \serverAcks server ->
710714 forEachBatch serverAcks maxMessageIds (\batch -> do
711715 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
712− whenOk result (atomic do
713− ackedIds <- *ackedIds
714− let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
715− set *ackedIds = updatedAckedIds))
716+ case result of
717+ Err _ -> println ("federation: ack to " ++ server ++ " failed")
718+ Ok _ -> atomic do
719+ ackedIds <- *ackedIds
720+ let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
721+ set *ackedIds = updatedAckedIds)
716722
717723sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
718724sendAcksAndClean = \pendingAcks ->
@@ -753,7 +759,7 @@ pruneAndRetry = \t -> do
753759 retryPendingForwards result.pendingForwards
754760 sendAcksAndClean result.pendingAcks)
755761
756−pollOnce = \token cursor startWakeup -> do
762+pollOnce = \token cursor -> do
757763 t <- now
758764 sessions <- *sessions
759765 let auth = authenticateSession sessions token t
@@ -762,6 +768,7 @@ pollOnce = \token cursor startWakeup -> do
762768 Authorized {pubkey} -> do
763769 seqCounter <- *seqCounter
764770 let clampedCursor = clamp 0 (max seqCounter t) cursor
771+ startWakeup <- *pollWakeup
765772 atomic do
766773 wakeup <- *pollWakeup
767774 allMessages <- *messages
@@ -885,17 +892,19 @@ serve = \req -> case req of
885892 ReqChallenge {pubkey, respond} -> do
886893 challenge <- genToken
887894 t <- now
888− pruneResult <- atomic do
895+ result <- atomic do
889896 pr <- maybePrune t
890897 challenges <- *challenges
891− let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges
892− let updatedChallenges = union
893− capped
894− [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
895− set *challenges = updatedChallenges
896− yield pr
897− forkBroadcastOffline t pruneResult
898− yield (respond {challenge: challenge})
898+ let perKeyCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
899+ if count perKeyCapped >= maxChallenges
900+ then yield {pr: pr, accepted: False {}}
901+ else do
902+ set *challenges = union perKeyCapped [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
903+ yield {pr: pr, accepted: True {}}
904+ when (not result.accepted)
905+ (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
906+ forkBroadcastOffline t result.pr
907+ yield (respond {challenge: if result.accepted then challenge else ""})
899908
900909 VerifyAuth {pubkey, challenge, signature, respond} -> do
901910 t <- now
@@ -937,19 +946,18 @@ serve = \req -> case req of
937946 yield (respond {token: token, expiresAt: expiry})
938947
939948 Poll {cursor, authorization, respond} -> do
940− startWakeup <- *pollWakeup
941949 result <- case extractToken authorization of
942950 Nothing {} -> yield (PollUnauthorized {})
943− Just {value: token} -> pollOnce token cursor startWakeup
951+ Just {value: token} -> pollOnce token cursor
944952 tAfterPoll <- now
945953 maybePruneAndBroadcast tAfterPoll
946954 yield (respond (case result of
947955 PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}
948956 PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))
949957
950− SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
958+ SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do
951959 t <- now
952− msgId <- genToken
960+ effectiveMsgId <- if messageId != "" then yield messageId else genToken
953961 let clampedTs = clampTimestamp timestamp t
954962 result <- atomic do
955963 pr <- maybePrune t
@@ -957,30 +965,33 @@ serve = \req -> case req of
957965 (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})
958966 (\_ -> do
959967 messages <- *messages
960− if countMessagesFor messages to t >= maxMessagesPerRecipient
961− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
962− else do
963− presence <- *presence
964− seqCounter <- *seqCounter
965− let seqTs = nextSeqTs t seqCounter
966− set *seqCounter = seqTs
967− set *messages = union messages
968− [{messageId: msgId, toKey: to,
969− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
970− let toPresence = presenceForKey presence to t
971− let localOnline = hasLocalPresence toPresence
972− let remoteServers = collectPeerServers toPresence (\p -> p.server)
973− forwards <- *forwards
974− let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
975− let shouldForward = federationEnabled && count forwardServers > 0
976− let delivery = if localOnline then DsDelivered {}
977− else if shouldForward then DsFederated {}
978− else DsQueued {}
979− let updatedForwards = if shouldForward
980− then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})
981− else forwards
982− set *forwards = updatedForwards
983− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})
968+ ackedIds <- *ackedIds
969+ if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)
970+ then yield {pr: pr, send: SendDuplicate {}}
971+ else if countMessagesFor messages to t >= maxMessagesPerRecipient
972+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
973+ else do
974+ presence <- *presence
975+ seqCounter <- *seqCounter
976+ let seqTs = nextSeqTs t seqCounter
977+ set *seqCounter = seqTs
978+ set *messages = union messages
979+ [{messageId: effectiveMsgId, toKey: to,
980+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
981+ let toPresence = presenceForKey presence to t
982+ let localOnline = hasLocalPresence toPresence
983+ let remoteServers = collectPeerServers toPresence (\p -> p.server)
984+ forwards <- *forwards
985+ let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
986+ let shouldForward = federationEnabled && count forwardServers > 0
987+ let delivery = if localOnline then DsDelivered {}
988+ else if shouldForward then DsFederated {}
989+ else DsQueued {}
990+ let updatedForwards = if shouldForward
991+ then union forwards (do s <- forwardServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})
992+ else forwards
993+ set *forwards = updatedForwards
994+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})
984995 case result.send of
985996 SendDone {delivery, remoteServers, forwardServers} ->
986997 fork (do
@@ -988,17 +999,20 @@ serve = \req -> case req of
988999 sendToEach remoteServers (\server -> do
9891000 fwdResult <- fetch (federationUrl server) (ForwardMessage {
9901001 toKey: to, encryptedBlob: encryptedBlob,
991− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
1002+ timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
9921003 authorization: federationAuth})
993− when (any forwardServers (\s -> s == server))
994− (whenOk fwdResult (removeForward msgId server)))
1004+ case fwdResult of
1005+ Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1006+ Ok _ -> when (any forwardServers (\s -> s == server))
1007+ (removeForward effectiveMsgId server))
9951008 when (isQueued delivery)
9961009 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))
9971010 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
9981011 _ -> forkBroadcastOffline t result.pr
9991012 yield (case result.send of
10001013 SendFailed {status} -> respond {status: status, messageId: ""}
1001− SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: msgId})
1014+ SendDuplicate {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId}
1015+ SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
10021016
10031017 AckMessages {messageIds, authorization, respond} -> do
10041018 t <- now