9de1b40 · skrepka · git.iko.soy

63pollTimeoutMs = 25000 -- long-poll heartbeat interval

64presenceRefreshInterval : Int<Ms>

65presenceRefreshInterval = 300000 -- 5 minutes

66+presenceRefreshThreshold : Int<Ms>

67+presenceRefreshThreshold = presenceRefreshInterval * 2 -- refresh entries expiring within this window

66federationEnabled = federationSecret != ""

68federationEnabled = federationSecret != ""

67federationAuth = "Bearer " ++ federationSecret

69federationAuth = "Bearer " ++ federationSecret

68federationUrl = \server -> federationScheme ++ server

70federationUrl = \server -> federationScheme ++ server

129type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}

131type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}

130type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

132type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

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

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

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

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

133−data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp, pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}

134data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

135data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

135

136

136-- ============================================

137-- ============================================

849 lastRefresh <- *lastPresenceRefresh

850 lastRefresh <- *lastPresenceRefresh

850 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

851 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval

851 presence <- *presence

852 presence <- *presence

852− let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p) else []

853+ let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []

853 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []

854 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []

854 when shouldRefresh (set *lastPresenceRefresh = t)

855 when shouldRefresh (set *lastPresenceRefresh = t)

855 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

856 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}

910 else if b == 14 then do s <- *messageSignal14; yield s

911 else if b == 14 then do s <- *messageSignal14; yield s

911 else do s <- *messageSignal15; yield s

912 else do s <- *messageSignal15; yield s

912

913

913−pollEventsAndDelete = \pubkey clampedCursor t -> do

914+ackPolledMessages = \pubkey cursor t -> do

914 messages <- *messages

915 messages <- *messages

915− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))

916+ let toAck = do m <- messages; where m.toKey == pubkey && m.receivedAt <= cursor; yield m

916− if count limitedMsgs > 0

917+ if count toAck > 0

917 then do

918 then do

918− set *messages = diff messages limitedMsgs

919+ set *messages = diff messages toAck

919− let polledMsgIds = do m <- limitedMsgs; yield m.messageId

920+ let ackedMsgIds = do m <- toAck; yield m.messageId

920 forwards <- *forwards

921 forwards <- *forwards

921− set *forwards = do f <- forwards; where none polledMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

922+ set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

922− let ackedFederated = federatedAckEntries limitedMsgs t

923+ let ackedFederated = federatedAckEntries toAck t

923 ackedIds <- *ackedIds

924 ackedIds <- *ackedIds

924 let combinedAckedIds = union ackedIds ackedFederated

925 let combinedAckedIds = union ackedIds ackedFederated

925− let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in

926+ let excessCount = count combinedAckedIds - maxAckedIds

926− if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)

927+ let evictedEntries = if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)

927 set *ackedIds = diff combinedAckedIds evictedEntries

928 set *ackedIds = diff combinedAckedIds evictedEntries

929+ yield {pendingAcks: ackedFederated, evictedEntries: evictedEntries}

930+ else yield {pendingAcks: [], evictedEntries: []}

931+

932+pollNewEvents = \pubkey clampedCursor t -> do

933+ messages <- *messages

934+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))

928− let evts = do

935+ let evts = do

929− m <- limitedMsgs

936+ m <- limitedMsgs

930− yield {id: m.messageId, eventType: "message",

937+ yield {id: m.messageId, eventType: "message",

931− payload: {toKey: m.toKey,

938+ payload: {toKey: m.toKey,

932− encryptedBlob: m.encryptedBlob,

939+ encryptedBlob: m.encryptedBlob,

933− timestamp: m.timestamp}}

940+ timestamp: m.timestamp}}

934− let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs

941+ let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs

935− yield (PollOk {events: evts, cursor: lastReceivedAt, pendingAcks: ackedFederated, evictedEntries: evictedEntries})

942+ yield {events: evts, cursor: lastReceivedAt}

936− else yield (PollOk {events: [], cursor: clampedCursor, pendingAcks: [], evictedEntries: []})

937

943

938pollOnce = \token cursor -> do

944pollOnce = \token cursor -> do

939 t <- now

945 t <- now

944 Authorized {pubkey} -> do

950 Authorized {pubkey} -> do

945 seqCounter <- *seqCounter

951 seqCounter <- *seqCounter

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

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

953+ ackResult <- if clampedCursor > 0

954+ then ackPolledMessages pubkey clampedCursor t

955+ else yield {pendingAcks: [], evictedEntries: []}

947 startWakeup <- *pollWakeup

956 startWakeup <- *pollWakeup

948 startSignal <- readMessageSignal pubkey

957 startSignal <- readMessageSignal pubkey

949− yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal}})

958+ yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})

950 case setup of

959 case setup of

951− Nothing {} -> yield (PollUnauthorized {})

960+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

952 Just {value: s} -> do

961 Just {value: s} -> do

953 -- Phase 1: check for existing messages (no blocking)

962 -- Phase 1: check for existing messages (no blocking)

954 initialMessages <- *messages

963 initialMessages <- *messages

955 let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t

964 let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t

956 if count initial > 0

965 if count initial > 0

966+ then do

957− then atomic do pollEventsAndDelete s.pubkey s.clampedCursor t

967+ result <- atomic do pollNewEvents s.pubkey s.clampedCursor t

968+ yield {authorized: True {}, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}

958 else do

969 else do

959 -- Phase 2: wait for signal or heartbeat (no *messages dependency)

970 -- Phase 2: wait for signal or heartbeat (no *messages dependency)

960 atomic do

971 atomic do

963 if wakeup > s.startWakeup || signal > s.startSignal

974 if wakeup > s.startWakeup || signal > s.startSignal

964 then yield {}

975 then yield {}

965 else retry

976 else retry

966− -- Phase 3: read messages after waking, delete atomically

977+ -- Phase 3: re-authenticate and read messages after waking

967 t2 <- now

978 t2 <- now

979+ result <- atomic do

980+ sessions2 <- *sessions

981+ case authenticateSession sessions2 token t2 of

982+ Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}

983+ Authorized {} -> do

968− atomic do pollEventsAndDelete s.pubkey s.clampedCursor t2

984+ r <- pollNewEvents s.pubkey s.clampedCursor t2

985+ yield {authorized: True {}, events: r.events, cursor: r.cursor}

986+ yield {authorized: result.authorized, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}

969

987

970-- ============================================

988-- ============================================

971-- Derived Relations

989-- Derived Relations

1067 t <- now

1085 t <- now

1068 result <- atomic do

1086 result <- atomic do

1069 pr <- maybePrune t

1087 pr <- maybePrune t

1088+ challenges <- *challenges

1089+ let active = do c <- challenges; where c.expiresAt > t; yield c

1090+ let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)

1091+ if activeForKey >= maxChallengesPerKey

1092+ then yield {pr: pr, accepted: False {}, rateLimited: False {}}

1093+ else if count active >= maxChallenges

1094+ then yield {pr: pr, accepted: False {}, rateLimited: False {}}

1095+ else do

1070− rateTs <- *challengeRateTs

1096+ rateTs <- *challengeRateTs

1071− rateCount <- *challengeRateCount

1097+ rateCount <- *challengeRateCount

1072− let newWindow = t - rateTs >= challengeRateWindowMs

1098+ let newWindow = t - rateTs >= challengeRateWindowMs

1073− let currentCount = if newWindow then 0 else rateCount

1099+ let currentCount = if newWindow then 0 else rateCount

1074− if currentCount >= challengeRateLimit

1100+ if currentCount >= challengeRateLimit

1075− then yield {pr: pr, accepted: False {}, rateLimited: True {}}

1101+ then yield {pr: pr, accepted: False {}, rateLimited: True {}}

1076− else do

1102+ else do

1077− set *challengeRateTs = if newWindow then t else rateTs

1103+ set *challengeRateTs = if newWindow then t else rateTs

1078− set *challengeRateCount = currentCount + 1

1104+ set *challengeRateCount = currentCount + 1

1079− challenges <- *challenges

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

1081− if count perKeyCapped >= maxChallenges

1082− then yield {pr: pr, accepted: False {}, rateLimited: False {}}

1083− else do

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

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

1085− yield {pr: pr, accepted: True {}, rateLimited: False {}}

1106+ yield {pr: pr, accepted: True {}, rateLimited: False {}}

1086 when result.rateLimited

1107 when result.rateLimited

1087 (println ("challenge: rate limited (" ++ show challengeRateLimit ++ "/s), rejecting for " ++ pubkey))

1108 (println ("challenge: rate limited (" ++ show challengeRateLimit ++ "/s), rejecting for " ++ pubkey))

1088 when (not result.accepted && not result.rateLimited)

1109 when (not result.accepted && not result.rateLimited)

1130

1151

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

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

1132 result <- case extractToken authorization of

1153 result <- case extractToken authorization of

1133− Nothing {} -> yield (PollUnauthorized {})

1154+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}

1134 Just {value: token} -> pollOnce token cursor

1155 Just {value: token} -> pollOnce token cursor

1135 tAfterPoll <- now

1156 tAfterPoll <- now

1136 maybePruneAndBroadcast tAfterPoll

1157 maybePruneAndBroadcast tAfterPoll

1137− case result of

1138− PollOk {pendingAcks, evictedEntries} ->

1139− fork (when federationEnabled (do

1158+ fork (when federationEnabled (do

1140− sendAcksAndClean pendingAcks

1159+ sendAcksAndClean result.pendingAcks

1141− retryEvictedAcks evictedEntries pendingAcks))

1160+ retryEvictedAcks result.evictedEntries result.pendingAcks))

1142− PollUnauthorized {} -> yield {}

1143− yield (respond (case result of

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

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

1161+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})

1146

1162

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

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

1148 t <- now

1164 t <- now

1153 withSessionAuth authorization t

1169 withSessionAuth authorization t

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

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

1155 (\senderKey -> do

1171 (\senderKey -> do

1156− sendRates <- *sendRates

1172+ messages <- *messages

1157− let senderEntry = headOf (do r <- sendRates; where r.pubkey == senderKey; yield r)

1173+ ackedIds <- *ackedIds

1158− let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= sendRateWindowMs

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

1159− let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count

1160− if rateCount >= maxSendRatePerKey

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

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

1162 else do

1176 else do

1177+ sendRates <- *sendRates

1178+ let senderEntry = headOf (do r <- sendRates; where r.pubkey == senderKey; yield r)

1179+ let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= sendRateWindowMs

1180+ let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count

1181+ if rateCount >= maxSendRatePerKey

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

1183+ else do

1163− let newRates = if isNewWindow

1184+ let newRates = if isNewWindow

1164− then union (do r <- sendRates; where r.pubkey != senderKey; yield r) [{pubkey: senderKey, windowStart: t, count: 1}]

1185+ then union (do r <- sendRates; where r.pubkey != senderKey; yield r) [{pubkey: senderKey, windowStart: t, count: 1}]

1165− else do r <- sendRates; yield (if r.pubkey == senderKey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)

1186+ else do r <- sendRates; yield (if r.pubkey == senderKey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)

1166− set *sendRates = if count newRates > maxSendRateEntries then (do r <- newRates; where t - r.windowStart < sendRateWindowMs; yield r) else newRates

1187+ set *sendRates = if count newRates > maxSendRateEntries then (do r <- newRates; where t - r.windowStart < sendRateWindowMs; yield r) else newRates

1167− messages <- *messages

1168− ackedIds <- *ackedIds

1169− if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)

1170− then yield {pr: pr, send: SendDuplicate {}}

1171− else if countMessagesFor messages to t >= maxMessagesPerRecipient

1188+ if countMessagesFor messages to t >= maxMessagesPerRecipient

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

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

1173− else do

1190+ else do

1174− presence <- *presence

1191+ presence <- *presence

1175− seqCounter <- *seqCounter

1192+ seqCounter <- *seqCounter

1176− let seqTs = nextSeqTs t seqCounter

1193+ let seqTs = nextSeqTs t seqCounter

1177− set *seqCounter = seqTs

1194+ set *seqCounter = seqTs

1178− set *messages = union messages

1195+ set *messages = union messages

1179− [{messageId: effectiveMsgId, toKey: to,

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

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

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

1181− signalMessageForKey to seqTs

1198+ signalMessageForKey to seqTs

1182− let toPresence = presenceForKey presence to t

1199+ let toPresence = presenceForKey presence to t

1183− let localOnline = hasLocalPresence toPresence

1200+ let localOnline = hasLocalPresence toPresence

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

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

1185− forwards <- *forwards

1202+ let delivery = if localOnline then DsDelivered {}

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

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

1203+ else if federationEnabled && count remoteServers > 0 then DsFederated {}

1188− let delivery = if localOnline then DsDelivered {}

1189− else if shouldForward then DsFederated {}

1190− else DsQueued {}

1204+ else DsQueued {}

1191− let updatedForwards = if shouldForward

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

1193− else forwards

1194− set *forwards = updatedForwards

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

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

1196 case result.send of

1206 case result.send of

1197− SendDone {delivery, remoteServers, forwardServers} ->

1207+ SendDone {delivery, remoteServers} ->

1198 fork (do

1208 fork (do

1199− when (count remoteServers > 0 && federationEnabled) (do

1209+ when (count remoteServers > 0 && federationEnabled) (

1200 sendToEach remoteServers (\server -> do

1210 sendToEach remoteServers (\server -> do

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

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

1202 toKey: to, encryptedBlob: encryptedBlob,

1212 toKey: to, encryptedBlob: encryptedBlob,

1203 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

1213 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,

1204 authorization: federationAuth})

1214 authorization: federationAuth})

1205 case fwdResult of

1215 case fwdResult of

1216+ Err _ -> do

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

1217+ println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)

1207− Ok _ -> when (any forwardServers (\s -> s == server))

1218+ atomic do

1208− (removeForward effectiveMsgId server))

1219+ forwards <- *forwards

1209− when (count forwardServers < count remoteServers)

1220+ when (count forwards < maxForwardsEntries)

1210− (println ("sendMessage: " ++ show (count remoteServers - count forwardServers) ++ " remote server(s) not tracked for " ++ effectiveMsgId ++ " (forwards at capacity " ++ show maxForwardsEntries ++ ")")))

1221+ (set *forwards = union forwards [{messageId: effectiveMsgId, toServer: server, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}])

1222+ Ok _ -> yield {}))

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

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

1212 _ -> forkBroadcastOffline t result.pr

1224 _ -> forkBroadcastOffline t result.pr

1213 yield (case result.send of

1225 yield (case result.send of