05d753c · skrepka · git.iko.soy

433 set *lastPrunedAt = t

434 pruneExpired t

435

436−sendBatchedGossip = \peerServers allEvents ->

436+sendBatchedGossipToServer = \server allEvents ->

437 if count allEvents == 0

438 then yield {}

439 else do

441 if count acc >= maxGossipEvents then acc

442 else union acc [e]) [] allEvents

443 let rest = diff allEvents batch

444− sendToEach peerServers (\server ->

445− fetch (federationScheme ++ server) (RecvGossip {

444+ fetch (federationScheme ++ server) (RecvGossip {

446− authorization: "Bearer " ++ federationSecret,

445+ authorization: "Bearer " ++ federationSecret,

447− events: batch, hopsRemaining: 2, fromServer: serverHostN}))

446+ events: batch, hopsRemaining: 2, fromServer: serverHostN})

447+ sendBatchedGossipToServer server rest

448+

448− sendBatchedGossip peerServers rest

449+sendBatchedGossip = \peerServers allEvents ->

450+ if count allEvents == 0

451+ then yield {}

452+ else sendToEach peerServers (\server ->

453+ sendBatchedGossipToServer server allEvents)

449

454

450broadcastOfflineGossip = \t peerServers expiredLocal -> do

455broadcastOfflineGossip = \t peerServers expiredLocal -> do

451 let expiredEntries = fold (\acc p ->

456 let expiredEntries = fold (\acc p ->

452− if count (do a <- acc; where a.pubkey == p.pubkey; yield a) > 0 then acc

457+ if count (do a <- acc; where a.pubkey == p.pubkey; where a.expiresAt >= p.expiresAt; yield a) > 0 then acc

458+ else union

459+ (do a <- acc; where not (a.pubkey == p.pubkey); yield a)

453− else union acc [{pubkey: p.pubkey, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}]) [] expiredLocal

460+ [{pubkey: p.pubkey, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000), expiresAt: p.expiresAt}]) [] expiredLocal

454 if not (federationSecret == "") && count expiredEntries > 0 && count peerServers > 0

461 if not (federationSecret == "") && count expiredEntries > 0 && count peerServers > 0

455 then do

462 then do

456 let gossipEvents = do

463 let gossipEvents = do

457 e <- expiredEntries

464 e <- expiredEntries

458− yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: t, ttl: e.ttl}

465+ yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}

459 sendBatchedGossip peerServers gossipEvents

466 sendBatchedGossip peerServers gossipEvents

460 else yield {}

467 else yield {}

461

468

520 where m.receivedAt >= t - messageMaxAge

527 where m.receivedAt >= t - messageMaxAge

521 where m.receivedAt > clampedCursor

528 where m.receivedAt > clampedCursor

522 yield m

529 yield m

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

530+ let limitedMsgs = (fold (\acc m ->

524− if count acc >= maxPollEvents then acc

531+ if acc.size < maxPollEvents

525− else union acc [m]) [] matchingMsgs

532+ then

533+ let newHeap = union acc.heap [m]

534+ {size: count newHeap, heap: newHeap, heapMax: if m.receivedAt > acc.heapMax then m.receivedAt else acc.heapMax}

535+ else if m.receivedAt < acc.heapMax

536+ then

537+ let removed = fold (\r x ->

538+ if not r.done && x.receivedAt == acc.heapMax

539+ then {done: True {}, items: r.items}

540+ else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap

541+ let newHeap = union removed.items [m]

542+ let newMax = fold (\best x -> if x.receivedAt > best then x.receivedAt else best) -1 newHeap

543+ {size: acc.size, heap: newHeap, heapMax: newMax}

544+ else acc) {size: 0, heap: [], heapMax: -1} matchingMsgs).heap

526 let authorized = case session of

545 let authorized = case session of

527 Nothing {} -> False {}

546 Nothing {} -> False {}

528 _ -> True {}

547 _ -> True {}

566 pr <- maybePrune t

585 pr <- maybePrune t

567 challenges <- *challenges

586 challenges <- *challenges

568 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; where c.expiresAt > t; yield c)

587 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; where c.expiresAt > t; yield c)

569− let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges

588+ let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && c2.expiresAt > t && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges

570 let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0

589 let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0

571 then fold (\acc c ->

590 then fold (\acc c ->

572 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry

591 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry

573 then {removed: True {}, items: acc.items}

592 then {removed: True {}, items: acc.items}

574 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges

593 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges

575 else {removed: False {}, items: challenges}

594 else {removed: False {}, items: challenges}

595+ let withoutExpired = if count cappedChallenges.items >= maxChallenges

596+ then do c <- cappedChallenges.items; where c.expiresAt > t; yield c

597+ else cappedChallenges.items

576− let globalOldestExpiry = if count cappedChallenges.items >= maxChallenges

598+ let globalOldestExpiry = if count withoutExpired >= maxChallenges

577− then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) -1 cappedChallenges.items

599+ then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) -1 withoutExpired

578 else -1

600 else -1

579 let globalCapped = if globalOldestExpiry >= 0

601 let globalCapped = if globalOldestExpiry >= 0

580 then (fold (\acc c ->

602 then (fold (\acc c ->

581 if not acc.removed && c.expiresAt == globalOldestExpiry

603 if not acc.removed && c.expiresAt == globalOldestExpiry

582 then {removed: True {}, items: acc.items}

604 then {removed: True {}, items: acc.items}

583− else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} cappedChallenges.items).items

605+ else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} withoutExpired).items

584− else cappedChallenges.items

606+ else withoutExpired

585 set *challenges = union

607 set *challenges = union

586 globalCapped

608 globalCapped

587 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

609 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

852 originServer: server, fromServer: serverHostN})

874 originServer: server, fromServer: serverHostN})

853 yield {}) (\err -> do

875 yield {}) (\err -> do

854 println ("federation: error sending evicted ack to " ++ server ++ ": " ++ show err)

876 println ("federation: error sending evicted ack to " ++ server ++ ": " ++ show err)

855− atomic do

877+ evicted <- atomic do

856 ackedIds <- *ackedIds

878 ackedIds <- *ackedIds

857 let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

879 let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

858− let available = maxAckedIds - count ackedIds

880+ let combined = union ackedIds reinsert

859− let toReinsert = if available <= 0 then []

881+ let excess = count combined - maxAckedIds

860− else fold (\acc a -> if count acc >= available then acc else union acc [a]) [] reinsert

882+ let afterEviction = if excess <= 0 then combined

861− set *ackedIds = union ackedIds toReinsert))

883+ else evictOldestAckedIds combined excess

884+ set *ackedIds = afterEviction

885+ yield (count combined - count afterEviction)

886+ if evicted > 0

887+ then println ("federation: evicted " ++ show evicted ++ " old ack entries to reinsert failed acks for " ++ server)

888+ else yield {}))

862 else yield {}

889 else yield {}

863 broadcastOfflineGossip t result.peerServers result.expiredLocal

890 broadcastOfflineGossip t result.peerServers result.expiredLocal

864

891

874 let authorized = case session of

901 let authorized = case session of

875 Nothing {} -> False {}

902 Nothing {} -> False {}

876 _ -> True {}

903 _ -> True {}

877− let valid = do

904+ let valid = if not validPubkey then [] else do

878 p <- presence

905 p <- presence

879 where p.pubkey == pubkey

906 where p.pubkey == pubkey

880 where p.expiresAt > t

907 where p.expiresAt > t

1151 messages <- *messages

1178 messages <- *messages

1152 seqCounter <- *seqCounter

1179 seqCounter <- *seqCounter

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

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

1154− let recipientMsgCount = count (do

1181+ let recipientMsgCount = if not (authorized && validOrigin && validFields) then 0

1182+ else count (do

1155− m <- messages

1183+ m <- messages

1156− where m.toKey == toKey

1184+ where m.toKey == toKey

1157− where m.receivedAt >= t - messageMaxAge

1185+ where m.receivedAt >= t - messageMaxAge

1158− yield m)

1186+ yield m)

1159− let existsInMessages = count (do

1187+ let existsInMessages = if not (authorized && validOrigin && validFields) then False {}

1188+ else count (do

1160− m <- messages

1189+ m <- messages

1161− where m.messageId == messageId

1190+ where m.messageId == messageId

1162− where m.toKey == toKey

1191+ where m.toKey == toKey

1163− yield m) > 0

1192+ yield m) > 0

1164 ackedIds <- *ackedIds

1193 ackedIds <- *ackedIds

1194+ let existsInAcked = if not (authorized && validOrigin && validFields) then False {}

1165− let existsInAcked = count (do

1195+ else count (do

1166− a <- ackedIds

1196+ a <- ackedIds

1167− where a.messageId == messageId

1197+ where a.messageId == messageId

1168− where a.toKey == toKey

1198+ where a.toKey == toKey

1169− yield a) > 0

1199+ yield a) > 0

1170 let exists = existsInMessages || existsInAcked

1200 let exists = existsInMessages || existsInAcked

1171 let shouldInsert = authorized && validOrigin && validFields && not exists && recipientMsgCount < maxMessagesPerRecipient

1201 let shouldInsert = authorized && validOrigin && validFields && not exists && recipientMsgCount < maxMessagesPerRecipient

1172 set *seqCounter = if shouldInsert then seqTs else seqCounter

1202 set *seqCounter = if shouldInsert then seqTs else seqCounter