1200f6a · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Backend improvements: three-phase poll optimization, challenge expiry, forward ack cleanup, stats endpoint

1200f6aiko6 months ago

server.knot+57−22

compare with master

unifiedsplit

@@ -114,7 +114,6 @@ type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: T

114type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}

115data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}

116data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]} | SendDuplicate {}

117−data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}

118data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}

117data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}

119data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}

118data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}

120data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

119data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

@@ -153,7 +152,7 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

153route AuthApi where

152route AuthApi where

154 /auth

153 /auth

155 POST {pubkey: PubkeyHex} /challenge

154 POST {pubkey: PubkeyHex} /challenge

156− -> {challenge: Text, ok: Bool}

155+ -> {challenge: Text, ok: Bool, expiresAt: Timestamp}

157 = ReqChallenge

156 = ReqChallenge

158 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex} /verify

157 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex} /verify

159 -> {token: Text, expiresAt: Timestamp}

158 -> {token: Text, expiresAt: Timestamp}

@@ -194,7 +193,12 @@ route MaintenanceApi where

194route HealthApi where

193route HealthApi where

195 GET /health -> {ok: Bool} = Health

194 GET /health -> {ok: Bool} = Health

196

195

196+route StatsApi where

197+ GET /stats headers {authorization: Text}

198+ -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ok: Bool}

199+ = Stats

200+

197−route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi | HealthApi

198

202

199-- ============================================

203-- ============================================

200-- Helpers

204-- Helpers

@@ -789,6 +793,20 @@ readMessageSignal = \pubkey ->

789 else if b == 2 then do s <- *messageSignal2; yield s

793 else if b == 2 then do s <- *messageSignal2; yield s

790 else do s <- *messageSignal3; yield s

794 else do s <- *messageSignal3; yield s

791

795

796+pollEvents = \allMessages pubkey clampedCursor t ->

797+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t)) in

798+ if count limitedMsgs > 0

799+ then

800+ let evts = do

801+ m <- limitedMsgs

802+ yield {id: m.messageId, eventType: "message",

803+ payload: {toKey: m.toKey,

804+ encryptedBlob: m.encryptedBlob,

805+ timestamp: m.timestamp}}

806+ in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs

807+ in PollOk {events: evts, cursor: lastReceivedAt}

808+ else PollOk {events: [], cursor: clampedCursor}

809+

792pollOnce = \token cursor -> do

810pollOnce = \token cursor -> do

793 t <- now

811 t <- now

794 sessions <- *sessions

812 sessions <- *sessions

@@ -799,24 +817,23 @@ pollOnce = \token cursor -> do

799 seqCounter <- *seqCounter

817 seqCounter <- *seqCounter

800 let clampedCursor = clamp 0 (max seqCounter t) cursor

818 let clampedCursor = clamp 0 (max seqCounter t) cursor

801 startWakeup <- *pollWakeup

819 startWakeup <- *pollWakeup

802− atomic do

820+ startSignal <- readMessageSignal pubkey

803− wakeup <- *pollWakeup

821+ -- Phase 1: check for existing messages (no blocking)

804− messageSignal <- readMessageSignal pubkey

822+ initialMessages <- *messages

805− allMessages <- *messages

806− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t))

823+ let initial = pollableMessages initialMessages pubkey clampedCursor t

807− if count limitedMsgs > 0

824+ if count initial > 0

808− then

825+ then yield (pollEvents initialMessages pubkey clampedCursor t)

809− let evts = do

826+ else do

810− m <- limitedMsgs

827+ -- Phase 2: wait for signal or heartbeat (no *messages dependency)

811− yield {id: m.messageId, eventType: "message",

828+ atomic do

812− payload: {toKey: m.toKey,

829+ wakeup <- *pollWakeup

813− encryptedBlob: m.encryptedBlob,

830+ signal <- readMessageSignal pubkey

814− timestamp: m.timestamp}}

815− in let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs

816− in yield (PollOk {events: evts, cursor: lastReceivedAt})

817− else if wakeup > startWakeup

831+ if wakeup > startWakeup || signal > startSignal

818− then yield (PollOk {events: [], cursor: clampedCursor})

832+ then yield {}

819− else retry

833+ else retry

834+ -- Phase 3: read messages after waking

835+ newMessages <- *messages

836+ yield (pollEvents newMessages pubkey clampedCursor t)

820

837

821-- ============================================

838-- ============================================

822-- Derived Relations

839-- Derived Relations

@@ -935,7 +952,7 @@ serve = \req -> case req of

935 when (not result.accepted)

952 when (not result.accepted)

936 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

953 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))

937 forkBroadcastOffline t result.pr

954 forkBroadcastOffline t result.pr

938− yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted})

955+ yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})

939

956

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

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

941 t <- now

958 t <- now

@@ -1057,6 +1074,9 @@ serve = \req -> case req of

1057 let ackedMessages = matchingAckMessages messages pubkey dedupedIds

1074 let ackedMessages = matchingAckMessages messages pubkey dedupedIds

1058 let remainingMessages = diff messages ackedMessages

1075 let remainingMessages = diff messages ackedMessages

1059 set *messages = remainingMessages

1076 set *messages = remainingMessages

1077+ forwards <- *forwards

1078+ let ackedMsgIds = do m <- ackedMessages; yield m.messageId

1079+ set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

1060 let ackedFederated = federatedAckEntries ackedMessages t

1080 let ackedFederated = federatedAckEntries ackedMessages t

1061 ackedIds <- *ackedIds

1081 ackedIds <- *ackedIds

1062 let combinedAckedIds = union ackedIds ackedFederated

1082 let combinedAckedIds = union ackedIds ackedFederated

@@ -1166,6 +1186,21 @@ serve = \req -> case req of

1166

1186

1167 Health {respond} -> yield (respond {ok: True {}})

1187 Health {respond} -> yield (respond {ok: True {}})

1168

1188

1189+ Stats {authorization, respond} -> do

1190+ let authorized = verifyMaintenanceAuth authorization

1191+ if authorized

1192+ then do

1193+ result <- atomic do

1194+ s <- *sessions

1195+ m <- *messages

1196+ p <- *presence

1197+ f <- *forwards

1198+ c <- *challenges

1199+ a <- *ackedIds

1200+ yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a}

1201+ yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ok: True {}})

1202+ else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ok: False {}})

1203+

1169-- ============================================

1204-- ============================================

1170-- Entry Point

1205-- Entry Point

1171-- ============================================

1206-- ============================================