242926b · skrepka · git.iko.soy

git.iko.soy/skrepka

Fix 3 server.knot bugs: poll cursor regression, missing offline gossip broadcasts, and superseded event re-propagation

242926biko7 months ago

server.knot+39−18

compare with master

unifiedsplit

@@ -37,6 +37,8 @@ type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int,

38*presence : [{pubkey: Text, server: Text, expiresAt: Int, timestamp: Int}]

40+*seqCounter : Int

41+

40-- ============================================

42-- ============================================

41-- Routes

43-- Routes

42-- ============================================

44-- ============================================

@@ -200,12 +202,14 @@ serve = \req -> case req of

200 else do

202 else do

201 challenge <- genToken

203 challenge <- genToken

202 t <- now

204 t <- now

203− atomic do

205+ expiredLocal <- atomic do

204− pruneExpired t

206+ el <- pruneExpired t

205 challenges <- *challenges

207 challenges <- *challenges

206 set *challenges = union challenges

208 set *challenges = union challenges

207 [{pubkey: pubkey, challenge: challenge, expiresAt: t + 60000}]

209 [{pubkey: pubkey, challenge: challenge, expiresAt: t + 60000}]

210+ yield el

208 respond {challenge: challenge}

211 respond {challenge: challenge}

212+ broadcastOfflineGossip t expiredLocal

209

213

210 VerifyAuth {pubkey, challenge, signature, respond} -> do

214 VerifyAuth {pubkey, challenge, signature, respond} -> do

211 t <- now

215 t <- now

@@ -283,8 +287,8 @@ serve = \req -> case req of

283 expiredLocal <- pruneExpired t

287 expiredLocal <- pruneExpired t

284 sessions <- *sessions

288 sessions <- *sessions

285 messages <- *messages

289 messages <- *messages

290+ seqCounter <- *seqCounter

286 let session = lookupSession sessions token t

291 let session = lookupSession sessions token t

287− let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages

288 let evts = case session of

292 let evts = case session of

289 Nothing {} -> []

293 Nothing {} -> []

290 Just {value: s} -> do

294 Just {value: s} -> do

@@ -298,7 +302,7 @@ serve = \req -> case req of

298 timestamp: m.timestamp}}

302 timestamp: m.timestamp}}

299 let newCursor = case session of

303 let newCursor = case session of

300 Nothing {} -> 0

304 Nothing {} -> 0

301− _ -> if maxReceived > cursor then maxReceived else cursor

305+ _ -> if seqCounter > cursor then seqCounter else cursor

302 yield {events: evts, cursor: newCursor, expiredLocal: expiredLocal}

306 yield {events: evts, cursor: newCursor, expiredLocal: expiredLocal}

303 respond {events: result.events, cursor: result.cursor}

307 respond {events: result.events, cursor: result.cursor}

304 broadcastOfflineGossip t result.expiredLocal

308 broadcastOfflineGossip t result.expiredLocal

@@ -314,8 +318,12 @@ serve = \req -> case req of

314 messages <- *messages

318 messages <- *messages

315 presence <- *presence

319 presence <- *presence

316 let session = lookupSession sessions (extractToken authorization) t

320 let session = lookupSession sessions (extractToken authorization) t

317− let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages

321+ seqCounter <- *seqCounter

318− let seqTs = if t > maxReceived then t else maxReceived + 1

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

319 set *messages = case session of

327 set *messages = case session of

320 Nothing {} -> messages

328 Nothing {} -> messages

321 Just {value: s} -> if validInput

329 Just {value: s} -> if validInput

@@ -408,7 +416,7 @@ serve = \req -> case req of

408 LookupPresence {pubkey, authorization, respond} -> do

416 LookupPresence {pubkey, authorization, respond} -> do

409 t <- now

417 t <- now

410 result <- atomic do

418 result <- atomic do

411− pruneExpired t

419+ expiredLocal <- pruneExpired t

412 sessions <- *sessions

420 sessions <- *sessions

413 presence <- *presence

421 presence <- *presence

414 let session = lookupSession sessions (extractToken authorization) t

422 let session = lookupSession sessions (extractToken authorization) t

@@ -431,8 +439,9 @@ serve = \req -> case req of

431 yield p) of

439 yield p) of

432 Just {value: p} -> p.server

440 Just {value: p} -> p.server

433 Nothing {} -> fold (\_ p -> p.server) "" valid

441 Nothing {} -> fold (\_ p -> p.server) "" valid

434− yield {foundServer: foundServer, authorized: authorized}

442+ yield {foundServer: foundServer, authorized: authorized, expiredLocal: expiredLocal}

435 respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}

443 respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}

444+ broadcastOfflineGossip t result.expiredLocal

436

445

437 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do

446 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do

438 t <- now

447 t <- now

@@ -493,8 +502,8 @@ serve = \req -> case req of

493 where not (a.pubkey == e.pubkey && a.server == e.server)

502 where not (a.pubkey == e.pubkey && a.server == e.server)

494 yield a)

503 yield a)

495 [e]) [] filteredOnline

504 [e]) [] filteredOnline

496− appliedKeys <- atomic do

505+ gossipResult <- atomic do

497− pruneExpired t

506+ expiredLocal <- pruneExpired t

498 presence <- *presence

507 presence <- *presence

499 let appliedOnline = do

508 let appliedOnline = do

500 e <- deduped

509 e <- deduped

@@ -552,7 +561,7 @@ serve = \req -> case req of

552 set *presence = if validRequest

561 set *presence = if validRequest

553 then afterOffline

562 then afterOffline

554 else presence

563 else presence

555− yield (union appliedOnline appliedOffline)

564+ yield {appliedKeys: (union appliedOnline appliedOffline), expiredLocal: expiredLocal}

556 let validEvents = do

565 let validEvents = do

557 e <- events

566 e <- events

558 where validRequest

567 where validRequest

@@ -563,10 +572,16 @@ serve = \req -> case req of

563 where not (e.server == serverHost)

572 where not (e.server == serverHost)

564 where e.timestamp <= t

573 where e.timestamp <= t

565 where e.timestamp + (clampTtl e.ttl) * 1000 > t

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

566 yield e

581 yield e

567 let effectiveEvents = do

582 let effectiveEvents = do

568 e <- validEvents

583 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

570 yield e

585 yield e

571 if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0

586 if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0

572 then do

587 then do

@@ -589,6 +604,7 @@ serve = \req -> case req of

589 {events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHost})

604 {events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHost})

590 else yield {}

605 else yield {}

591 respond {ok: validRequest}

606 respond {ok: validRequest}

607+ broadcastOfflineGossip t gossipResult.expiredLocal

592

608

593 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do

609 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do

594 t <- now

610 t <- now

@@ -596,23 +612,26 @@ serve = \req -> case req of

596 let validOrigin = not (originServer == "") && not (originServer == serverHost)

612 let validOrigin = not (originServer == "") && not (originServer == serverHost)

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

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

598 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t else timestamp

614 let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t else timestamp

599− atomic do

615+ expiredLocal <- atomic do

600− pruneExpired t

616+ el <- pruneExpired t

601 messages <- *messages

617 messages <- *messages

602− let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages

618+ seqCounter <- *seqCounter

603− let seqTs = if t > maxReceived then t else maxReceived + 1

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

604 let exists = count (do

620 let exists = count (do

605 m <- messages

621 m <- messages

606 where m.messageId == messageId

622 where m.messageId == messageId

607 where m.timestamp >= t - messageMaxAge

623 where m.timestamp >= t - messageMaxAge

608 yield m) > 0

624 yield m) > 0

609 let shouldInsert = authorized && validOrigin && validFields && not exists

625 let shouldInsert = authorized && validOrigin && validFields && not exists

626+ set *seqCounter = if shouldInsert then seqTs else seqCounter

610 set *messages = if shouldInsert

627 set *messages = if shouldInsert

611 then union messages

628 then union messages

612 [{messageId: messageId, toKey: toKey, fromKey: fromKey,

629 [{messageId: messageId, toKey: toKey, fromKey: fromKey,

613 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

630 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

614 else messages

631 else messages

632+ yield el

615 respond {ok: authorized && validOrigin && validFields}

633 respond {ok: authorized && validOrigin && validFields}

634+ broadcastOfflineGossip t expiredLocal

616

635

617 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

636 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

618 t <- now

637 t <- now

@@ -623,8 +642,8 @@ serve = \req -> case req of

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

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

624 else union acc [mid]) [] messageIds

643 else union acc [mid]) [] messageIds

625 let validAck = authorized && originServer == serverHost && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxPubkeyLen) && count messageIds <= maxMessageIds

644 let validAck = authorized && originServer == serverHost && not (fromServer == "") && not (fromServer == serverHost) && not (tooLong fromServer maxPubkeyLen) && count messageIds <= maxMessageIds

626− atomic do

645+ expiredLocal <- atomic do

627− pruneExpired t

646+ el <- pruneExpired t

628 messages <- *messages

647 messages <- *messages

629 presence <- *presence

648 presence <- *presence

630 set *messages = if validAck

649 set *messages = if validAck

@@ -640,7 +659,9 @@ serve = \req -> case req of

640 yield p) == 0

659 yield p) == 0

641 yield m)

660 yield m)

642 else messages

661 else messages

662+ yield el

643 respond {ok: validAck}

663 respond {ok: validAck}

664+ broadcastOfflineGossip t expiredLocal

644

665

645-- ============================================

666-- ============================================

646-- Entry Point

667-- Entry Point