@@ -49,6 +49,7 @@ type BlobHex = Text where \s -> isValidHex s && not (tooLong s maxBlobLen)
4949type SignatureHex = Text where \s -> isValidHex s && not (tooLong s maxSignatureLen)
5050type MessageIdHex = Text where \s -> isValidHex s && not (tooLong s maxMessageIdLen)
5151type ServerName = Text where \s -> not (tooLong s maxServerNameLen) && not (isBadServerName s)
52+type OptionalServerName = Text where \s -> s == "" || (not (tooLong s maxServerNameLen) && not (isBadServerName s))
5253
5354type Timestamp = Int<Ms>
5455
@@ -60,15 +61,15 @@ type Timestamp = Int<Ms>
6061
6162*sessions : [{token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}]
6263
63−*messages : [{messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}]
64+*messages : [{messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: OptionalServerName, receivedAt: Timestamp}]
6465
6566*presence : [{pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp}]
6667
6768*seqCounter : Timestamp
6869
69−*forwards : [{messageId: MessageIdHex, toServer: Text, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}]
70+*forwards : [{messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}]
7071
71−*ackedIds : [{messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: Text}]
72+*ackedIds : [{messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}]
7273
7374*lastPrunedAt : Timestamp
7475
@@ -278,20 +279,14 @@ normalizeServerName = \s -> toLower (stripTrailingDot s)
278279
279280serverHostN = normalizeServerName serverHost
280281
281−containsLocalhostSuffix = \s ->
282+containsSuffix = \suffix suffixLen s ->
282283 if s == "" then False {}
283− else if take 10 s == ".localhost" && (drop 10 s == "" || take 1 (drop 10 s) == ":") then True {}
284− else containsLocalhostSuffix (drop 1 s)
284+ else if take suffixLen s == suffix && (drop suffixLen s == "" || take 1 (drop suffixLen s) == ":") then True {}
285+ else containsSuffix suffix suffixLen (drop 1 s)
285286
286−containsLocalSuffix = \s ->
287− if s == "" then False {}
288− else if take 6 s == ".local" && (drop 6 s == "" || take 1 (drop 6 s) == ":") then True {}
289− else containsLocalSuffix (drop 1 s)
290−
291−containsInternalSuffix = \s ->
292− if s == "" then False {}
293− else if take 9 s == ".internal" && (drop 9 s == "" || take 1 (drop 9 s) == ":") then True {}
294− else containsInternalSuffix (drop 1 s)
287+containsLocalhostSuffix = \s -> containsSuffix ".localhost" 10 s
288+containsLocalSuffix = \s -> containsSuffix ".local" 6 s
289+containsInternalSuffix = \s -> containsSuffix ".internal" 9 s
295290
296291isBadServerName = \s ->
297292 let sn = toLower (stripTrailingDot s) in
@@ -368,28 +363,39 @@ verifyMaintenanceAuth = \authorization ->
368363
369364clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
370365
366+clampTimestamp = \timestamp t -> if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
367+
368+dedupMessageIds = \messageIds ->
369+ fold (\acc mid ->
370+ let nmid = normalizeHex mid in
371+ if count acc >= maxMessageIds then acc
372+ else if count (do a <- acc; where a == nmid; yield a) > 0 then acc
373+ else union acc [nmid]) [] messageIds
374+
375+boundedSmallest = \items n getKey ->
376+ (fold (\acc a ->
377+ let k = getKey a in
378+ if acc.size < n
379+ then
380+ let newHeap = union acc.heap [a] in
381+ let newSize = count newHeap in
382+ if newSize > acc.size
383+ then {size: newSize, heap: newHeap, heapMax: if k > acc.heapMax then k else acc.heapMax}
384+ else acc
385+ else if k < acc.heapMax
386+ then
387+ let removed = fold (\r x ->
388+ if not r.done && getKey x == acc.heapMax
389+ then {done: True {}, items: r.items}
390+ else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
391+ let newHeap = union removed.items [a] in
392+ let newMax = fold (\best x -> if getKey x > best then getKey x else best) (-1) newHeap in
393+ {size: acc.size, heap: newHeap, heapMax: newMax}
394+ else acc) {size: 0, heap: [], heapMax: -1} items).heap
395+
371396evictOldestAckedIds = \items n ->
372397 if n <= 0 then items
373− else
374− let toEvict = (fold (\acc a ->
375− if acc.size < n
376− then
377− let newHeap = union acc.heap [a] in
378− let newSize = count newHeap in
379− if newSize > acc.size
380− then {size: newSize, heap: newHeap, heapMax: if a.ackedAt > acc.heapMax then a.ackedAt else acc.heapMax}
381− else acc
382− else if a.ackedAt < acc.heapMax
383− then
384− let removed = fold (\r x ->
385− if not r.done && x.ackedAt == acc.heapMax
386− then {done: True {}, items: r.items}
387− else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
388− let newHeap = union removed.items [a] in
389− let newMax = fold (\best x -> if x.ackedAt > best then x.ackedAt else best) (-1) newHeap in
390− {size: acc.size, heap: newHeap, heapMax: newMax}
391− else acc) {size: 0, heap: [], heapMax: -1} items).heap
392− in diff items toEvict
398+ else diff items (boundedSmallest items n (\a -> a.ackedAt))
393399
394400safeCompareLoop = \s1 s2 n result ->
395401 if n <= 0 && s1 == "" && s2 == ""
@@ -414,6 +420,22 @@ collectPeerServers = \items getServer ->
414420 else if count (do a <- acc; where a == s; yield a) > 0 then acc
415421 else union acc [s]) [] items
416422
423+validGossipEvents = \events validRequest t eventType ->
424+ do
425+ e <- events
426+ where validRequest
427+ where e.eventType == eventType
428+ where not (e.server == "")
429+ where not (e.pubkey == "")
430+ where not (tooLong e.pubkey maxPubkeyLen)
431+ where not (tooLong e.server maxServerNameLen)
432+ where not (isBadServerName e.server)
433+ where isValidHex e.pubkey
434+ where not (normalizeServerName e.server == serverHostN)
435+ where e.timestamp <= t + clockSkewTolerance
436+ where e.timestamp + (clampTtl e.ttl) * 1000 > t
437+ yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, expiresAt: e.timestamp + (clampTtl e.ttl) * 1000, timestamp: e.timestamp, ttl: clampTtl e.ttl}
438+
417439sendToEach : [Text] -> (Text -> (a {})) -> a {}
418440sendToEach = \servers action ->
419441 if count servers == 0
@@ -617,24 +639,7 @@ pollOnce = \token cursor -> do
617639 where m.receivedAt >= t - messageMaxAge
618640 where m.receivedAt > clampedCursor
619641 yield m
620− let limitedMsgs = (fold (\acc m ->
621− if acc.size < maxPollEvents
622− then
623− let newHeap = union acc.heap [m] in
624− let newSize = count newHeap in
625− if newSize > acc.size
626− then {size: newSize, heap: newHeap, heapMax: if m.receivedAt > acc.heapMax then m.receivedAt else acc.heapMax}
627− else acc
628− else if m.receivedAt < acc.heapMax
629− then
630− let removed = fold (\r x ->
631− if not r.done && x.receivedAt == acc.heapMax
632− then {done: True {}, items: r.items}
633− else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
634− let newHeap = union removed.items [m] in
635− let newMax = fold (\best x -> if x.receivedAt > best then x.receivedAt else best) (-1) newHeap in
636− {size: acc.size, heap: newHeap, heapMax: newMax}
637− else acc) {size: 0, heap: [], heapMax: -1} matchingMsgs).heap
642+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
638643 let authorized = case session of
639644 Nothing {} -> False {}
640645 _ -> True {}
@@ -785,13 +790,19 @@ serve = \req -> case req of
785790 let encryptedBlob = normalizeHex encryptedBlob
786791 t <- now
787792 msgId <- genToken
788− let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
793+ let clampedTs = clampTimestamp timestamp t
789794 result <- atomic do
790795 pr <- maybePrune t
791796 sessions <- *sessions
792797 messages <- *messages
793798 presence <- *presence
794799 let session = lookupSession sessions (extractToken authorization) t
800+ let fromKey = case session of
801+ Nothing {} -> ""
802+ Just {value: s} -> s.pubkey
803+ let authorized = case session of
804+ Nothing {} -> False {}
805+ _ -> True {}
795806 seqCounter <- *seqCounter
796807 let seqTs = if t > seqCounter then t else seqCounter + 1
797808 let recipientMsgCount = count (do
@@ -799,42 +810,32 @@ serve = \req -> case req of
799810 where m.toKey == to
800811 where m.receivedAt >= t - messageMaxAge
801812 yield m)
802− let shouldInsert = case session of
803− Nothing {} -> False {}
804− Just {value: _} -> recipientMsgCount < maxMessagesPerRecipient
813+ let shouldInsert = authorized && recipientMsgCount < maxMessagesPerRecipient
805814 set *seqCounter = if shouldInsert then seqTs else seqCounter
806− set *messages = case session of
807− Nothing {} -> messages
808− Just {value: s} -> if recipientMsgCount < maxMessagesPerRecipient
809− then union messages
810− [{messageId: msgId, toKey: to, fromKey: s.pubkey,
811− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
812− else messages
813− let localOnline = case session of
814− Nothing {} -> False {}
815− Just {value: _} -> count (do
816− p <- presence
817− where p.pubkey == to
818− where p.server == serverHostN
819− where p.expiresAt > t
820− yield p) > 0
821− let remoteServers = case session of
822− Nothing {} -> []
823− Just {value: _} -> do
815+ set *messages = if shouldInsert
816+ then union messages
817+ [{messageId: msgId, toKey: to, fromKey: fromKey,
818+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
819+ else messages
820+ let localOnline = authorized && count (do
821+ p <- presence
822+ where p.pubkey == to
823+ where p.server == serverHostN
824+ where p.expiresAt > t
825+ yield p) > 0
826+ let remoteServers = if authorized
827+ then do
824828 p <- presence
825829 where p.pubkey == to
826830 where not (p.server == serverHostN)
827831 where p.expiresAt > t
828832 yield p.server
829− let fromKey = case session of
830− Nothing {} -> ""
831− Just {value: s} -> s.pubkey
832− let status = case session of
833− Nothing {} -> "unauthorized"
834− _ -> if recipientMsgCount >= maxMessagesPerRecipient then "rejected"
835− else if localOnline then "delivered"
836− else if count remoteServers > 0 then "federated"
837− else "queued"
833+ else []
834+ let status = if not authorized then "unauthorized"
835+ else if not shouldInsert then "rejected"
836+ else if localOnline then "delivered"
837+ else if count remoteServers > 0 then "federated"
838+ else "queued"
838839 forwards <- *forwards
839840 let forwardsAtCapacity = count forwards >= maxForwardsEntries
840841 let forwardSlots = maxForwardsEntries - count forwards
@@ -871,11 +872,7 @@ serve = \req -> case req of
871872
872873 AckMessages {messageIds, authorization, respond} -> do
873874 t <- now
874− let dedupedIds = fold (\acc mid ->
875− let nmid = normalizeHex mid in
876− if count acc >= maxMessageIds then acc
877− else if count (do a <- acc; where a == nmid; yield a) > 0 then acc
878− else union acc [nmid]) [] messageIds
875+ let dedupedIds = dedupMessageIds messageIds
879876 result <- atomic do
880877 pr <- maybePrune t
881878 sessions <- *sessions
@@ -992,34 +989,8 @@ serve = \req -> case req of
992989 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
993990 let fromServer = normalizeServerName fromServer
994991 let validRequest = authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents
995− let onlineEntries = do
996− e <- events
997− where validRequest
998− where e.eventType == "online"
999− where not (e.server == "")
1000− where not (e.pubkey == "")
1001− where not (tooLong e.pubkey maxPubkeyLen)
1002− where not (tooLong e.server maxServerNameLen)
1003− where not (isBadServerName e.server)
1004− where isValidHex e.pubkey
1005− where not (normalizeServerName e.server == serverHostN)
1006− where e.timestamp <= t + clockSkewTolerance
1007− where e.timestamp + (clampTtl e.ttl) * 1000 > t
1008− yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, expiresAt: e.timestamp + (clampTtl e.ttl) * 1000, timestamp: e.timestamp}
1009− let offlineEntries = do
1010− e <- events
1011− where validRequest
1012− where e.eventType == "offline"
1013− where not (e.server == "")
1014− where not (e.pubkey == "")
1015− where not (tooLong e.pubkey maxPubkeyLen)
1016− where not (tooLong e.server maxServerNameLen)
1017− where not (isBadServerName e.server)
1018− where isValidHex e.pubkey
1019− where not (normalizeServerName e.server == serverHostN)
1020− where e.timestamp <= t + clockSkewTolerance
1021− where t - e.timestamp < (clampTtl e.ttl) * 1000
1022− yield {pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp}
992+ let onlineEntries = validGossipEvents events validRequest t "online"
993+ let offlineEntries = validGossipEvents events validRequest t "offline"
1023994 let filteredOnline = do
1024995 e <- onlineEntries
1025996 where count (do
@@ -1029,7 +1000,7 @@ serve = \req -> case req of
10291000 where o.server == e.server
10301001 where o.timestamp > e.timestamp
10311002 yield o) == 0
1032− yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp}
1003+ yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp, ttl: e.ttl}
10331004 let affectedKeys = union
10341005 (do
10351006 e <- onlineEntries
@@ -1180,37 +1151,21 @@ serve = \req -> case req of
11801151 set *presence = finalPresence
11811152 let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
11821153 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1183− let validEvents = do
1184− e <- events
1185− where validRequest
1186− where e.eventType == "online" || e.eventType == "offline"
1187− where not (e.server == "")
1188− where not (e.pubkey == "")
1189− where not (tooLong e.pubkey maxPubkeyLen)
1190− where not (tooLong e.server maxServerNameLen)
1191− where not (isBadServerName e.server)
1192− where isValidHex e.pubkey
1193− where not (normalizeServerName e.server == serverHostN)
1194− where e.timestamp <= t + clockSkewTolerance
1195− where e.timestamp + (clampTtl e.ttl) * 1000 > t
1196− where not (e.eventType == "online" && count (do
1197− o <- offlineEntries
1198− where o.server == fromServer
1199− where o.pubkey == normalizeHex e.pubkey
1200− where o.server == normalizeServerName e.server
1201− where o.timestamp > e.timestamp
1202− yield o) > 0)
1203− where not (e.eventType == "offline" && count (do
1154+ let filteredOffline = do
1155+ e <- offlineEntries
1156+ where count (do
12041157 o <- onlineEntries
1205− where o.pubkey == normalizeHex e.pubkey
1206− where o.server == normalizeServerName e.server
1158+ where o.pubkey == e.pubkey
1159+ where o.server == e.server
12071160 where o.timestamp > e.timestamp
1208− yield o) > 0)
1161+ yield o) == 0
12091162 yield e
12101163 let effectiveEvents = do
1211− e <- validEvents
1212− where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == normalizeServerName e.server; yield a) > 0
1213− yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp, ttl: clampTtl e.ttl}
1164+ e <- union
1165+ (do e <- filteredOnline; yield {eventType: "online", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1166+ (do e <- filteredOffline; yield {eventType: "offline", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1167+ where count (do a <- gossipResult.appliedKeys; where a.pubkey == e.pubkey; where a.server == e.server; yield a) > 0
1168+ yield e
12141169 if not gossipResult.withinCapacity
12151170 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
12161171 else yield {}
@@ -1242,7 +1197,7 @@ serve = \req -> case req of
12421197 t <- now
12431198 let authorized = verifyFederationAuth authorization
12441199 let validOrigin = not (originServer == serverHostN)
1245− let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
1200+ let clampedTs = clampTimestamp timestamp t
12461201 result <- atomic do
12471202 pr <- maybePrune t
12481203 messages <- *messages
@@ -1284,11 +1239,7 @@ serve = \req -> case req of
12841239 let originServer = normalizeServerName originServer
12851240 t <- now
12861241 let authorized = verifyFederationAuth authorization
1287− let dedupedIds = fold (\acc mid ->
1288− let nmid = normalizeHex mid in
1289− if count acc >= maxMessageIds then acc
1290− else if count (do a <- acc; where a == nmid; yield a) > 0 then acc
1291− else union acc [nmid]) [] messageIds
1242+ let dedupedIds = dedupMessageIds messageIds
12921243 let validAckFields = authorized && originServer == serverHostN && not (fromServer == serverHostN)
12931244 pruneResult <- atomic do
12941245 pr <- maybePrune t