@@ -70,6 +70,11 @@ type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: T
7070type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
7171type AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
7272type PruneResult = {expiredLocal: [PresenceEntry], peerServers: [Text]}
73+type PollResult = PollUnauthorized {} | PollAuthorized {events: [PollEvent], cursor: Timestamp}
74+type VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [Text]}
75+type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: Text, remoteServers: [Text], fromKey: PubkeyHex}
76+type AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
77+type LookupResult = LookupUnauthorized {} | LookupDone {foundServer: Text}
7378
7479
7580-- ============================================
@@ -412,18 +417,25 @@ boundedSmallest = \items n getKey ->
412417 let newHeap = union acc.heap [a] in
413418 let newSize = count newHeap in
414419 if newSize > acc.size
415− then {size: newSize, heap: newHeap, heapMax: if k > acc.heapMax then k else acc.heapMax}
420+ then {size: newSize, heap: newHeap, heapMax: case acc.heapMax of
421+ Nothing {} -> Just {value: k}
422+ Just {value: v} -> if k > v then Just {value: k} else acc.heapMax}
416423 else acc
417− else if k < acc.heapMax
418− then
419− let removed = fold (\r x ->
420− if not r.done && getKey x == acc.heapMax
421− then {done: True {}, items: r.items}
422− else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
423− let newHeap = union removed.items [a] in
424− let newMax = fold (\best x -> if getKey x > best then getKey x else best) (-1) newHeap in
425− {size: acc.size, heap: newHeap, heapMax: newMax}
426− else acc) {size: 0, heap: [], heapMax: -1} items).heap
424+ else case acc.heapMax of
425+ Nothing {} -> acc
426+ Just {value: maxVal} ->
427+ if k < maxVal
428+ then
429+ let removed = fold (\r x ->
430+ if not r.done && getKey x == maxVal
431+ then {done: True {}, items: r.items}
432+ else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
433+ let newHeap = union removed.items [a] in
434+ let newMax = fold (\best x -> case best of
435+ Nothing {} -> Just {value: getKey x}
436+ Just {value: v} -> if getKey x > v then Just {value: getKey x} else best) Nothing {} newHeap in
437+ {size: acc.size, heap: newHeap, heapMax: newMax}
438+ else acc) {size: 0, heap: [], heapMax: Nothing {}} items).heap
427439
428440evictOldestAckedIds = \items n ->
429441 if n <= 0 then items
@@ -774,23 +786,17 @@ serve = \req -> case req of
774786 then collectPeerServers updatedPresence (\p -> p.server)
775787 else []
776788 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
777− fork (do
778− let expiredLocal = case result.verify of
779− VerifyOk _ -> do
780− p <- result.pr.expiredLocal
781− where not (p.pubkey == pubkey)
782− yield p
783− _ -> result.pr.expiredLocal
784− case result.verify of
785− VerifyOk {peerServers} ->
786− when federationEnabled
787− (sendToEach peerServers (\server -> do
788− _ <- fetch (federationScheme ++ server) (RecvGossip {
789− authorization: "Bearer " ++ federationSecret,
790− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
791− yield {}))
792− _ -> yield {}
793− broadcastOfflineGossip t result.pr.peerServers expiredLocal)
789+ fork (case result.verify of
790+ VerifyOk {peerServers} -> do
791+ when federationEnabled
792+ (sendToEach peerServers (\server -> do
793+ _ <- fetch (federationScheme ++ server) (RecvGossip {
794+ authorization: "Bearer " ++ federationSecret,
795+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
796+ yield {}))
797+ broadcastOfflineGossip t result.pr.peerServers (do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p)
798+ VerifyFailed {} ->
799+ broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
794800 yield (case result.verify of
795801 VerifyFailed _ -> respond {token: "", expiresAt: 0}
796802 VerifyOk _ -> respond {token: token, expiresAt: expiry})
@@ -1089,11 +1095,11 @@ serve = \req -> case req of
10891095 let originServer = normalizeServerName originServer
10901096 t <- now
10911097 let authorized = verifyFederationAuth authorization
1092− let dedupedIds = dedupMessageIds messageIds
10931098 let validAck = authorized && originServer == serverHostN && not (fromServer == serverHostN)
10941099 pruneResult <- atomic do
10951100 pr <- maybePrune t
10961101 when validAck (do
1102+ let dedupedIds = dedupMessageIds messageIds
10971103 forwards <- *forwards
10981104 set *forwards = do
10991105 f <- forwards