4c8ee60 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Extract helpers and refine types to reduce code duplication and improve type safety

4c8ee60iko6 months ago

server.knot+102−151

compare with master

unifiedsplit

@@ -49,6 +49,7 @@ type BlobHex = Text where \s -> isValidHex s && not (tooLong s maxBlobLen)

49type SignatureHex = Text where \s -> isValidHex s && not (tooLong s maxSignatureLen)

50type MessageIdHex = Text where \s -> isValidHex s && not (tooLong s maxMessageIdLen)

51type 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))

53type Timestamp = Int<Ms>

54type Timestamp = Int<Ms>

@@ -60,15 +61,15 @@ type Timestamp = Int<Ms>

61*sessions : [{token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}]

62*sessions : [{token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}]

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}]

65*presence : [{pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp}]

66*presence : [{pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp}]

67*seqCounter : Timestamp

68*seqCounter : Timestamp

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}]

71−*ackedIds : [{messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: Text}]

72+*ackedIds : [{messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}]

73*lastPrunedAt : Timestamp

74*lastPrunedAt : Timestamp

@@ -278,20 +279,14 @@ normalizeServerName = \s -> toLower (stripTrailingDot s)

278

279

279serverHostN = normalizeServerName serverHost

280serverHostN = normalizeServerName serverHost

280

281

281−containsLocalhostSuffix = \s ->

282+containsSuffix = \suffix suffixLen s ->

282 if s == "" then False {}

283 if s == "" then False {}

283− else if take 10 s == ".localhost" && (drop 10 s == "" || take 1 (drop 10 s) == ":") then True {}

284+ else if take suffixLen s == suffix && (drop suffixLen s == "" || take 1 (drop suffixLen s) == ":") then True {}

284− else containsLocalhostSuffix (drop 1 s)

285+ else containsSuffix suffix suffixLen (drop 1 s)

285

286

286−containsLocalSuffix = \s ->

287+containsLocalhostSuffix = \s -> containsSuffix ".localhost" 10 s

287− if s == "" then False {}

288+containsLocalSuffix = \s -> containsSuffix ".local" 6 s

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 ->

289+containsInternalSuffix = \s -> containsSuffix ".internal" 9 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)

295

290

296isBadServerName = \s ->

291isBadServerName = \s ->

297 let sn = toLower (stripTrailingDot s) in

292 let sn = toLower (stripTrailingDot s) in

@@ -368,28 +363,39 @@ verifyMaintenanceAuth = \authorization ->

368

363

369clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl

364clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl

370

365

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+

371evictOldestAckedIds = \items n ->

396evictOldestAckedIds = \items n ->

372 if n <= 0 then items

397 if n <= 0 then items

373− else

398+ else diff items (boundedSmallest items n (\a -> a.ackedAt))

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

393

399

394safeCompareLoop = \s1 s2 n result ->

400safeCompareLoop = \s1 s2 n result ->

395 if n <= 0 && s1 == "" && s2 == ""

401 if n <= 0 && s1 == "" && s2 == ""

@@ -414,6 +420,22 @@ collectPeerServers = \items getServer ->

414 else if count (do a <- acc; where a == s; yield a) > 0 then acc

420 else if count (do a <- acc; where a == s; yield a) > 0 then acc

415 else union acc [s]) [] items

421 else union acc [s]) [] items

416

422

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+

417sendToEach : [Text] -> (Text -> (a {})) -> a {}

439sendToEach : [Text] -> (Text -> (a {})) -> a {}

418sendToEach = \servers action ->

440sendToEach = \servers action ->

419 if count servers == 0

441 if count servers == 0

@@ -617,24 +639,7 @@ pollOnce = \token cursor -> do

617 where m.receivedAt >= t - messageMaxAge

639 where m.receivedAt >= t - messageMaxAge

618 where m.receivedAt > clampedCursor

640 where m.receivedAt > clampedCursor

619 yield m

641 yield m

620− let limitedMsgs = (fold (\acc m ->

642+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)

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

638 let authorized = case session of

643 let authorized = case session of

639 Nothing {} -> False {}

644 Nothing {} -> False {}

640 _ -> True {}

645 _ -> True {}

@@ -785,13 +790,19 @@ serve = \req -> case req of

785 let encryptedBlob = normalizeHex encryptedBlob

790 let encryptedBlob = normalizeHex encryptedBlob

786 t <- now

791 t <- now

787 msgId <- genToken

792 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

789 result <- atomic do

794 result <- atomic do

790 pr <- maybePrune t

795 pr <- maybePrune t

791 sessions <- *sessions

796 sessions <- *sessions

792 messages <- *messages

797 messages <- *messages

793 presence <- *presence

798 presence <- *presence

794 let session = lookupSession sessions (extractToken authorization) t

799 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 {}

795 seqCounter <- *seqCounter

806 seqCounter <- *seqCounter

796 let seqTs = if t > seqCounter then t else seqCounter + 1

807 let seqTs = if t > seqCounter then t else seqCounter + 1

797 let recipientMsgCount = count (do

808 let recipientMsgCount = count (do

@@ -799,42 +810,32 @@ serve = \req -> case req of

799 where m.toKey == to

810 where m.toKey == to

800 where m.receivedAt >= t - messageMaxAge

811 where m.receivedAt >= t - messageMaxAge

801 yield m)

812 yield m)

802− let shouldInsert = case session of

803− Nothing {} -> False {}

804− Just {value: _} -> recipientMsgCount < maxMessagesPerRecipient

813+ let shouldInsert = authorized && recipientMsgCount < maxMessagesPerRecipient

805 set *seqCounter = if shouldInsert then seqTs else seqCounter

814 set *seqCounter = if shouldInsert then seqTs else seqCounter

806− set *messages = case session of

815+ set *messages = if shouldInsert

807− Nothing {} -> messages

808− Just {value: s} -> if recipientMsgCount < maxMessagesPerRecipient

809− then union messages

816+ then union messages

810− [{messageId: msgId, toKey: to, fromKey: s.pubkey,

817+ [{messageId: msgId, toKey: to, fromKey: fromKey,

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

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

812− else messages

819+ else messages

813− let localOnline = case session of

820+ let localOnline = authorized && count (do

814− Nothing {} -> False {}

815− Just {value: _} -> count (do

816− p <- presence

821+ p <- presence

817− where p.pubkey == to

822+ where p.pubkey == to

818− where p.server == serverHostN

823+ where p.server == serverHostN

819− where p.expiresAt > t

824+ where p.expiresAt > t

820− yield p) > 0

825+ yield p) > 0

821− let remoteServers = case session of

826+ let remoteServers = if authorized

822− Nothing {} -> []

827+ then do

823− Just {value: _} -> do

824 p <- presence

828 p <- presence

825 where p.pubkey == to

829 where p.pubkey == to

826 where not (p.server == serverHostN)

830 where not (p.server == serverHostN)

827 where p.expiresAt > t

831 where p.expiresAt > t

828 yield p.server

832 yield p.server

829− let fromKey = case session of

833+ else []

830− Nothing {} -> ""

831− Just {value: s} -> s.pubkey

832− let status = case session of

833− Nothing {} -> "unauthorized"

834+ let status = if not authorized then "unauthorized"

834− _ -> if recipientMsgCount >= maxMessagesPerRecipient then "rejected"

835+ else if not shouldInsert then "rejected"

835− else if localOnline then "delivered"

836+ else if localOnline then "delivered"

836− else if count remoteServers > 0 then "federated"

837+ else if count remoteServers > 0 then "federated"

837− else "queued"

838+ else "queued"

838 forwards <- *forwards

839 forwards <- *forwards

839 let forwardsAtCapacity = count forwards >= maxForwardsEntries

840 let forwardsAtCapacity = count forwards >= maxForwardsEntries

840 let forwardSlots = maxForwardsEntries - count forwards

841 let forwardSlots = maxForwardsEntries - count forwards

@@ -871,11 +872,7 @@ serve = \req -> case req of

871

872

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

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

873 t <- now

874 t <- now

874− let dedupedIds = fold (\acc mid ->

875+ let dedupedIds = dedupMessageIds messageIds

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

879 result <- atomic do

876 result <- atomic do

880 pr <- maybePrune t

877 pr <- maybePrune t

881 sessions <- *sessions

878 sessions <- *sessions

@@ -992,34 +989,8 @@ serve = \req -> case req of

992 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining

989 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining

993 let fromServer = normalizeServerName fromServer

990 let fromServer = normalizeServerName fromServer

994 let validRequest = authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents

991 let validRequest = authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents

995− let onlineEntries = do

992+ let onlineEntries = validGossipEvents events validRequest t "online"

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

993+ let offlineEntries = validGossipEvents events validRequest t "offline"

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}

1023 let filteredOnline = do

994 let filteredOnline = do

1024 e <- onlineEntries

995 e <- onlineEntries

1025 where count (do

996 where count (do

@@ -1029,7 +1000,7 @@ serve = \req -> case req of

1029 where o.server == e.server

1000 where o.server == e.server

1030 where o.timestamp > e.timestamp

1001 where o.timestamp > e.timestamp

1031 yield o) == 0

1002 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}

1033 let affectedKeys = union

1004 let affectedKeys = union

1034 (do

1005 (do

1035 e <- onlineEntries

1006 e <- onlineEntries

@@ -1180,37 +1151,21 @@ serve = \req -> case req of

1180 set *presence = finalPresence

1151 set *presence = finalPresence

1181 let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)

1152 let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)

1182 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}

1153 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

1154+ let filteredOffline = 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

1155+ e <- 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

1156+ where count (do

1204 o <- onlineEntries

1157 o <- onlineEntries

1205− where o.pubkey == normalizeHex e.pubkey

1158+ where o.pubkey == e.pubkey

1206− where o.server == normalizeServerName e.server

1159+ where o.server == e.server

1207 where o.timestamp > e.timestamp

1160 where o.timestamp > e.timestamp

1208− yield o) > 0)

1161+ yield o) == 0

1209 yield e

1162 yield e

1210 let effectiveEvents = do

1163 let effectiveEvents = do

1211− e <- validEvents

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})

1212− where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == normalizeServerName e.server; yield a) > 0

1167+ where count (do a <- gossipResult.appliedKeys; where a.pubkey == e.pubkey; where a.server == 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}

1168+ yield e

1214 if not gossipResult.withinCapacity

1169 if not gossipResult.withinCapacity

1215 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)

1170 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)

1216 else yield {}

1171 else yield {}

@@ -1242,7 +1197,7 @@ serve = \req -> case req of

1242 t <- now

1197 t <- now

1243 let authorized = verifyFederationAuth authorization

1198 let authorized = verifyFederationAuth authorization

1244 let validOrigin = not (originServer == serverHostN)

1199 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

1246 result <- atomic do

1201 result <- atomic do

1247 pr <- maybePrune t

1202 pr <- maybePrune t

1248 messages <- *messages

1203 messages <- *messages

@@ -1284,11 +1239,7 @@ serve = \req -> case req of

1284 let originServer = normalizeServerName originServer

1239 let originServer = normalizeServerName originServer

1285 t <- now

1240 t <- now

1286 let authorized = verifyFederationAuth authorization

1241 let authorized = verifyFederationAuth authorization

1287− let dedupedIds = fold (\acc mid ->

1242+ let dedupedIds = dedupMessageIds messageIds

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

1292 let validAckFields = authorized && originServer == serverHostN && not (fromServer == serverHostN)

1243 let validAckFields = authorized && originServer == serverHostN && not (fromServer == serverHostN)

1293 pruneResult <- atomic do

1244 pruneResult <- atomic do

1294 pr <- maybePrune t

1245 pr <- maybePrune t