@@ -37,6 +37,8 @@ type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int,
3737
3838*presence : [{pubkey: Text, server: Text, expiresAt: Int, timestamp: Int}]
3939
40+*seqCounter : Int
41+
4042-- ============================================
4143-- Routes
4244-- ============================================
@@ -200,12 +202,14 @@ serve = \req -> case req of
200202 else do
201203 challenge <- genToken
202204 t <- now
203− atomic do
204− pruneExpired t
205+ expiredLocal <- atomic do
206+ el <- pruneExpired t
205207 challenges <- *challenges
206208 set *challenges = union challenges
207209 [{pubkey: pubkey, challenge: challenge, expiresAt: t + 60000}]
210+ yield el
208211 respond {challenge: challenge}
212+ broadcastOfflineGossip t expiredLocal
209213
210214 VerifyAuth {pubkey, challenge, signature, respond} -> do
211215 t <- now
@@ -283,8 +287,8 @@ serve = \req -> case req of
283287 expiredLocal <- pruneExpired t
284288 sessions <- *sessions
285289 messages <- *messages
290+ seqCounter <- *seqCounter
286291 let session = lookupSession sessions token t
287− let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
288292 let evts = case session of
289293 Nothing {} -> []
290294 Just {value: s} -> do
@@ -298,7 +302,7 @@ serve = \req -> case req of
298302 timestamp: m.timestamp}}
299303 let newCursor = case session of
300304 Nothing {} -> 0
301− _ -> if maxReceived > cursor then maxReceived else cursor
305+ _ -> if seqCounter > cursor then seqCounter else cursor
302306 yield {events: evts, cursor: newCursor, expiredLocal: expiredLocal}
303307 respond {events: result.events, cursor: result.cursor}
304308 broadcastOfflineGossip t result.expiredLocal
@@ -314,8 +318,12 @@ serve = \req -> case req of
314318 messages <- *messages
315319 presence <- *presence
316320 let session = lookupSession sessions (extractToken authorization) t
317− let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
318− let seqTs = if t > maxReceived then t else maxReceived + 1
321+ seqCounter <- *seqCounter
322+ let seqTs = if t > seqCounter then t else seqCounter + 1
323+ let shouldInsert = case session of
324+ Nothing {} -> False {}
325+ Just {value: _} -> validInput
326+ set *seqCounter = if shouldInsert then seqTs else seqCounter
319327 set *messages = case session of
320328 Nothing {} -> messages
321329 Just {value: s} -> if validInput
@@ -408,7 +416,7 @@ serve = \req -> case req of
408416 LookupPresence {pubkey, authorization, respond} -> do
409417 t <- now
410418 result <- atomic do
411− pruneExpired t
419+ expiredLocal <- pruneExpired t
412420 sessions <- *sessions
413421 presence <- *presence
414422 let session = lookupSession sessions (extractToken authorization) t
@@ -431,8 +439,9 @@ serve = \req -> case req of
431439 yield p) of
432440 Just {value: p} -> p.server
433441 Nothing {} -> fold (\_ p -> p.server) "" valid
434− yield {foundServer: foundServer, authorized: authorized}
442+ yield {foundServer: foundServer, authorized: authorized, expiredLocal: expiredLocal}
435443 respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}
444+ broadcastOfflineGossip t result.expiredLocal
436445
437446 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
438447 t <- now
@@ -493,8 +502,8 @@ serve = \req -> case req of
493502 where not (a.pubkey == e.pubkey && a.server == e.server)
494503 yield a)
495504 [e]) [] filteredOnline
496− appliedKeys <- atomic do
497− pruneExpired t
505+ gossipResult <- atomic do
506+ expiredLocal <- pruneExpired t
498507 presence <- *presence
499508 let appliedOnline = do
500509 e <- deduped
@@ -552,7 +561,7 @@ serve = \req -> case req of
552561 set *presence = if validRequest
553562 then afterOffline
554563 else presence
555− yield (union appliedOnline appliedOffline)
564+ yield {appliedKeys: (union appliedOnline appliedOffline), expiredLocal: expiredLocal}
556565 let validEvents = do
557566 e <- events
558567 where validRequest
@@ -563,10 +572,16 @@ serve = \req -> case req of
563572 where not (e.server == serverHost)
564573 where e.timestamp <= t
565574 where e.timestamp + (clampTtl e.ttl) * 1000 > t
575+ where not (e.eventType == "online" && count (do
576+ o <- offlineEntries
577+ where o.pubkey == e.pubkey
578+ where o.server == e.server
579+ where o.timestamp >= e.timestamp
580+ yield o) > 0)
566581 yield e
567582 let effectiveEvents = do
568583 e <- validEvents
569− where count (do a <- appliedKeys; where a.pubkey == e.pubkey; where a.server == e.server; yield a) > 0
584+ where count (do a <- gossipResult.appliedKeys; where a.pubkey == e.pubkey; where a.server == e.server; yield a) > 0
570585 yield e
571586 if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0
572587 then do
@@ -589,6 +604,7 @@ serve = \req -> case req of
589604 {events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHost})
590605 else yield {}
591606 respond {ok: validRequest}
607+ broadcastOfflineGossip t gossipResult.expiredLocal
592608
593609 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
594610 t <- now
@@ -596,23 +612,26 @@ serve = \req -> case req of
596612 let validOrigin = not (originServer == "") && not (originServer == serverHost)
597613 let validFields = not (toKey == "") && not (fromKey == "") && not (encryptedBlob == "") && not (messageId == "") && not (tooLong toKey maxPubkeyLen) && not (tooLong fromKey maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && not (tooLong messageId maxPubkeyLen)
598614 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t else timestamp
599− atomic do
600− pruneExpired t
615+ expiredLocal <- atomic do
616+ el <- pruneExpired t
601617 messages <- *messages
602− let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
603− let seqTs = if t > maxReceived then t else maxReceived + 1
618+ seqCounter <- *seqCounter
619+ let seqTs = if t > seqCounter then t else seqCounter + 1
604620 let exists = count (do
605621 m <- messages
606622 where m.messageId == messageId
607623 where m.timestamp >= t - messageMaxAge
608624 yield m) > 0
609625 let shouldInsert = authorized && validOrigin && validFields && not exists
626+ set *seqCounter = if shouldInsert then seqTs else seqCounter
610627 set *messages = if shouldInsert
611628 then union messages
612629 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
613630 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
614631 else messages
632+ yield el
615633 respond {ok: authorized && validOrigin && validFields}
634+ broadcastOfflineGossip t expiredLocal
616635
617636 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
618637 t <- now
@@ -623,8 +642,8 @@ serve = \req -> case req of
623642 else if count (do a <- acc; where a == mid; yield a) > 0 then acc
624643 else union acc [mid]) [] messageIds
625644 let validAck = authorized && originServer == serverHost && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxPubkeyLen) && count messageIds <= maxMessageIds
626− atomic do
627− pruneExpired t
645+ expiredLocal <- atomic do
646+ el <- pruneExpired t
628647 messages <- *messages
629648 presence <- *presence
630649 set *messages = if validAck
@@ -640,7 +659,9 @@ serve = \req -> case req of
640659 yield p) == 0
641660 yield m)
642661 else messages
662+ yield el
643663 respond {ok: validAck}
664+ broadcastOfflineGossip t expiredLocal
644665
645666-- ============================================
646667-- Entry Point