diff --git a/toju-app/src/app/domains/chat/README.md b/toju-app/src/app/domains/chat/README.md index b7c65a1..955f3bd 100644 --- a/toju-app/src/app/domains/chat/README.md +++ b/toju-app/src/app/domains/chat/README.md @@ -17,6 +17,7 @@ chat/ │ ├── message-integrity.rules.ts headHash, inventory refresh, revision merge predicates │ ├── message-revision.builder.rules.ts buildMessageRevision, materializeMessageFromRevision │ ├── message-sync.rules.ts Inventory-based sync: chunkArray, findMissingIds, limits +│ ├── message-sync-round.rules.ts Inventory round evidence: createInventoryRound, recordInventoryReply, isInventoryRoundClean │ └── auto-scroll.rules.ts resolveAutoScrollBehavior (instant on channel switch, smooth for live msgs) + isStuckToBottom predicate │ ├── feature/ @@ -128,6 +129,16 @@ sequenceDiagram `findMissingIds` compares each remote item's timestamp and reaction/attachment counts against the local map. Any item that is missing, newer, or has different counts is requested. +### Polling cadence: only evidence buys the slow poll + +`store/messages/messages-sync.effects.ts` polls every connected peer for its inventory and alternates between `SYNC_POLL_FAST_MS` (10 s) and `SYNC_POLL_SLOW_MS` (15 min). Which one it picks is decided by the round model in `message-sync-round.rules.ts`, not by elapsed time: + +- The poll records who it actually reached. `sendToPeer` returns whether the payload entered an open data channel, so a peer listed as connected whose channel is closed counts as **undelivered**, not asked. +- Each reply dispatches `peerInventoryCompared` with the number of ids that comparison showed we lack. +- A round is **clean** only when every asked peer replied and no reply reported missing ids. A timed-out round, a partially answered round, and a round with an undelivered request are all dirty, and dirty keeps the fast poll. +- Every scored round re-arms the poll timer, so the delay always comes from the verdict of the round that just closed - not from the previous round's verdict, which is how a client could end up waiting out 15 minutes right after discovering it was behind. +- A reply that lands after its round closed (a peer answering the reconnect or room-activation kickoff) re-arms the fast cadence when it reports missing ids, so convergence never waits out a slow poll. + ## GIF integration `KlipyService` checks availability on the active server, then proxies search requests through the server API. Rendered remote images now attempt a direct load first and only fall back to the image proxy after the browser reports a load failure, which is the practical approximation of a CORS or mixed-content fallback path in the renderer. diff --git a/toju-app/src/app/domains/chat/domain/rules/message-sync-round.rules.spec.ts b/toju-app/src/app/domains/chat/domain/rules/message-sync-round.rules.spec.ts new file mode 100644 index 0000000..cf2e68f --- /dev/null +++ b/toju-app/src/app/domains/chat/domain/rules/message-sync-round.rules.spec.ts @@ -0,0 +1,109 @@ +import { + describe, + it, + expect +} from 'vitest'; +import { + createInventoryRound, + isInventoryRoundClean, + recordInventoryReply +} from './message-sync-round.rules'; + +describe('message-sync-round.rules', () => { + it('is clean when every asked peer replied with nothing missing', () => { + let round = createInventoryRound({ + roomId: 'room-1', + askedPeerIds: ['peer-a', 'peer-b'] + }); + + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 }); + + expect(isInventoryRoundClean(round)).toBe(false); + + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-b', missingIdCount: 0 }); + + expect(isInventoryRoundClean(round)).toBe(true); + }); + + it('is dirty while a peer we asked has not replied', () => { + // A timed-out round is not a clean round: silence is not evidence of + // convergence, and treating it as clean drops the poll to the slow + // cadence while the peer still holds messages we never received. + const round = createInventoryRound({ + roomId: 'room-1', + askedPeerIds: ['peer-a', 'peer-b'] + }); + + expect(isInventoryRoundClean( + recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 }) + )).toBe(false); + }); + + it('is dirty when a request could not be delivered', () => { + const round = createInventoryRound({ + roomId: 'room-1', + askedPeerIds: ['peer-a'], + undeliveredPeerIds: ['peer-b'] + }); + + expect(isInventoryRoundClean( + recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 }) + )).toBe(false); + }); + + it('is dirty when any reply reported missing ids', () => { + let round = createInventoryRound({ + roomId: 'room-1', + askedPeerIds: ['peer-a', 'peer-b'] + }); + + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 3 }); + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-b', missingIdCount: 0 }); + + expect(isInventoryRoundClean(round)).toBe(false); + }); + + it('is dirty when nobody was asked', () => { + expect(isInventoryRoundClean(createInventoryRound({ roomId: 'room-1', askedPeerIds: [] }))).toBe(false); + expect(isInventoryRoundClean(null)).toBe(false); + }); + + it('counts a reply from a peer outside the round, but not toward completeness', () => { + // A peer answering a room-activation kickoff still proves we are behind. + let round = createInventoryRound({ + roomId: 'room-1', + askedPeerIds: ['peer-a'] + }); + + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-c', missingIdCount: 2 }); + + expect(isInventoryRoundClean(round)).toBe(false); + expect(round?.repliedPeerIds).toEqual([]); + + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 }); + + expect(isInventoryRoundClean(round)).toBe(false); + }); + + it('ignores replies for another room and never double-counts a peer', () => { + let round = createInventoryRound({ + roomId: 'room-1', + askedPeerIds: ['peer-a'] + }); + + round = recordInventoryReply(round, { roomId: 'room-2', peerId: 'peer-a', missingIdCount: 5 }); + + expect(round?.repliedPeerIds).toEqual([]); + expect(round?.missingIdCount).toBe(0); + + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 }); + round = recordInventoryReply(round, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 }); + + expect(round?.repliedPeerIds).toEqual(['peer-a']); + expect(isInventoryRoundClean(round)).toBe(true); + }); + + it('leaves a closed round closed', () => { + expect(recordInventoryReply(null, { roomId: 'room-1', peerId: 'peer-a', missingIdCount: 0 })).toBeNull(); + }); +}); diff --git a/toju-app/src/app/domains/chat/domain/rules/message-sync-round.rules.ts b/toju-app/src/app/domains/chat/domain/rules/message-sync-round.rules.ts new file mode 100644 index 0000000..01790e5 --- /dev/null +++ b/toju-app/src/app/domains/chat/domain/rules/message-sync-round.rules.ts @@ -0,0 +1,75 @@ +/** + * Evidence model for one message-inventory reconciliation round. + * + * The sync poll asks every reachable peer for its room inventory and then + * decides how soon to ask again. That decision must rest on replies, not on + * elapsed time: a round that timed out, only partly answered, or could not be + * delivered proves nothing about convergence, so it must not buy the slow + * polling cadence. + */ + +/** Peers asked, peers heard from, and what they reported, for one room. */ +export interface InventoryRound { + readonly roomId: string; + /** Peers the inventory request was actually handed to the transport for. */ + readonly askedPeerIds: readonly string[]; + /** Subset of `askedPeerIds` that answered with an inventory. */ + readonly repliedPeerIds: readonly string[]; + /** Peers whose data channel refused the request. */ + readonly undeliveredPeerIds: readonly string[]; + /** Total ids any reply showed we are missing or hold at a stale revision. */ + readonly missingIdCount: number; +} + +/** Starts a round from the peers a poll reached and the peers it could not. */ +export function createInventoryRound(input: { + roomId: string; + askedPeerIds: readonly string[]; + undeliveredPeerIds?: readonly string[]; +}): InventoryRound { + return { + roomId: input.roomId, + askedPeerIds: [...input.askedPeerIds], + repliedPeerIds: [], + undeliveredPeerIds: [...(input.undeliveredPeerIds ?? [])], + missingIdCount: 0 + }; +} + +/** + * Records one peer's inventory comparison. + * + * Replies for another room are ignored. A reply from a peer outside the round + * (it answered an earlier kickoff) still counts its missing ids - we are + * demonstrably behind - but cannot complete the round on another peer's behalf. + */ +export function recordInventoryReply( + round: InventoryRound | null, + reply: { roomId: string; peerId: string; missingIdCount: number } +): InventoryRound | null { + if (!round || round.roomId !== reply.roomId) + return round; + + const isAsked = round.askedPeerIds.includes(reply.peerId); + const alreadyReplied = round.repliedPeerIds.includes(reply.peerId); + + return { + ...round, + repliedPeerIds: + isAsked && !alreadyReplied + ? [...round.repliedPeerIds, reply.peerId] + : round.repliedPeerIds, + missingIdCount: round.missingIdCount + Math.max(0, reply.missingIdCount) + }; +} + +/** A round is clean only when every asked peer replied and nothing was missing. */ +export function isInventoryRoundClean(round: InventoryRound | null): boolean { + if (!round || round.askedPeerIds.length === 0) + return false; + + if (round.undeliveredPeerIds.length > 0 || round.missingIdCount > 0) + return false; + + return round.askedPeerIds.every((peerId) => round.repliedPeerIds.includes(peerId)); +} diff --git a/toju-app/src/app/store/messages/messages-incoming.handlers.spec.ts b/toju-app/src/app/store/messages/messages-incoming.handlers.spec.ts index 583e1fd..d6a5e11 100644 --- a/toju-app/src/app/store/messages/messages-incoming.handlers.spec.ts +++ b/toju-app/src/app/store/messages/messages-incoming.handlers.spec.ts @@ -2,6 +2,7 @@ import { defaultIfEmpty, firstValueFrom } from 'rxjs'; import { type Message } from '../../shared-kernel'; import { dispatchIncomingMessage } from './messages-incoming.handlers'; +import { MessagesActions } from './messages.actions'; function createMessage(overrides: Partial = {}): Message { return { @@ -17,6 +18,17 @@ function createMessage(overrides: Partial = {}): Message { }; } +function createInventoryItem(id: string, ts: number) { + return { + id, + ts, + rc: 0, + ac: 0, + revision: 1, + headHash: `hash-${id}` + }; +} + function createContext(overrides: Record = {}) { return { db: { @@ -197,6 +209,73 @@ describe('dispatchIncomingMessage room-scoped sync', () => { }); }); + it('reports how many ids a peer inventory showed we are missing', async () => { + // The sync round decides its polling cadence from this report: a reply we + // never hear about is indistinguishable from a peer that went silent. + const sendToPeer = vi.fn(); + const context = createContext({ + attachments: { getForMessage: vi.fn(() => []) }, + currentRoom: { id: 'room-a' }, + db: { getMessages: vi.fn(async () => []) }, + savedRooms: [{ id: 'room-a' }], + webrtc: { sendToPeer } + }); + const remoteItems = [createInventoryItem('remote-1', 1), createInventoryItem('remote-2', 2)]; + const action = await firstValueFrom( + dispatchIncomingMessage( + { + type: 'chat-inventory', + roomId: 'room-a', + fromPeerId: 'peer-1', + items: remoteItems + } as never, + context as never + ).pipe(defaultIfEmpty(null)) + ); + + expect(action).toEqual(MessagesActions.peerInventoryCompared({ + missingIdCount: 2, + peerId: 'peer-1', + roomId: 'room-a' + })); + + expect(sendToPeer).toHaveBeenCalledWith('peer-1', { + type: 'chat-sync-request-ids', + roomId: 'room-a', + ids: ['remote-1', 'remote-2'] + }); + }); + + it('reports a converged comparison when a peer inventory holds nothing new', async () => { + const sendToPeer = vi.fn(); + const context = createContext({ + attachments: { getForMessage: vi.fn(() => []) }, + currentRoom: { id: 'room-a' }, + db: { getMessages: vi.fn(async () => []) }, + savedRooms: [{ id: 'room-a' }], + webrtc: { sendToPeer } + }); + const action = await firstValueFrom( + dispatchIncomingMessage( + { + type: 'chat-inventory', + roomId: 'room-a', + fromPeerId: 'peer-1', + items: [] + } as never, + context as never + ).pipe(defaultIfEmpty(null)) + ); + + expect(action).toEqual(MessagesActions.peerInventoryCompared({ + missingIdCount: 0, + peerId: 'peer-1', + roomId: 'room-a' + })); + + expect(sendToPeer).not.toHaveBeenCalled(); + }); + it('ignores chat messages for rooms that are not saved or currently viewed', async () => { const saveMessage = vi.fn(async () => undefined); const rememberMessageRoom = vi.fn(); diff --git a/toju-app/src/app/store/messages/messages-incoming.handlers.ts b/toju-app/src/app/store/messages/messages-incoming.handlers.ts index c7f7651..0d48d90 100644 --- a/toju-app/src/app/store/messages/messages-incoming.handlers.ts +++ b/toju-app/src/app/store/messages/messages-incoming.handlers.ts @@ -15,7 +15,7 @@ import { from, EMPTY } from 'rxjs'; -import { mergeMap } from 'rxjs/operators'; +import { map, mergeMap } from 'rxjs/operators'; import { Action } from '@ngrx/store'; import { DELETED_MESSAGE_CONTENT, @@ -159,8 +159,9 @@ function handleInventoryRequest( } /** - * Compares a peer's inventory against local state - * and requests any missing or stale messages. + * Compares a peer's inventory against local state, requests any missing or + * stale messages, and reports the comparison so the sync round can tell a + * converged reply apart from silence. */ function handleInventory( event: IncomingMessageEvent, @@ -197,8 +198,18 @@ function handleInventory( webrtc.sendToPeer(fromPeerId, syncRequestIdsEvent); } + + return missing.length; })() - ).pipe(mergeMap(() => EMPTY)); + ).pipe( + map((missingIdCount) => + MessagesActions.peerInventoryCompared({ + missingIdCount, + peerId: fromPeerId, + roomId + }) + ) + ); } /** diff --git a/toju-app/src/app/store/messages/messages-sync.effects.spec.ts b/toju-app/src/app/store/messages/messages-sync.effects.spec.ts new file mode 100644 index 0000000..65ced19 --- /dev/null +++ b/toju-app/src/app/store/messages/messages-sync.effects.spec.ts @@ -0,0 +1,260 @@ +import { Injector, runInInjectionContext } from '@angular/core'; +import { Actions } from '@ngrx/effects'; +import { Store, type Action } from '@ngrx/store'; +import { + BehaviorSubject, + Observable, + Subject +} from 'rxjs'; +import type { ChatEvent, Room } from '../../shared-kernel'; +import { RealtimeSessionFacade } from '../../core/realtime'; +import { DatabaseService } from '../../infrastructure/persistence'; +import { DebuggingService } from '../../core/services/debugging.service'; +import { + SYNC_POLL_FAST_MS, + SYNC_POLL_SLOW_MS, + SYNC_TIMEOUT_MS +} from './messages.helpers'; +import { MessagesActions } from './messages.actions'; +import { MessagesSyncEffects } from './messages-sync.effects'; +import { selectCurrentRoom } from '../rooms/rooms.selectors'; + +const ROOM_ID = 'room-1'; + +describe('MessagesSyncEffects sync cadence', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('keeps polling fast after a round nobody answered', async () => { + // A timed-out inventory round proves nothing: the peers may still hold + // messages we never received. Backing off to the slow poll here is what + // left a client 15 minutes behind. + const harness = createHarness(); + + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); + + it('backs off only once every asked peer replied with nothing missing', async () => { + const harness = createHarness(); + + await harness.advance(SYNC_POLL_FAST_MS); + harness.reply('peer-a', 0); + harness.reply('peer-b', 0); + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual([]); + + await harness.advance(SYNC_POLL_SLOW_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); + + it('keeps polling fast when only some asked peers replied', async () => { + const harness = createHarness(); + + await harness.advance(SYNC_POLL_FAST_MS); + harness.reply('peer-a', 0); + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); + + it('keeps polling fast when a request never reached a peer', async () => { + const harness = createHarness(); + + harness.setUndeliverable('peer-b'); + + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a']); + + harness.reply('peer-a', 0); + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + harness.setDeliverable('peer-b'); + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); + + it('keeps polling fast when a reply reported missing messages', async () => { + const harness = createHarness(); + + await harness.advance(SYNC_POLL_FAST_MS); + harness.reply('peer-a', 4); + harness.reply('peer-b', 0); + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); + + it('re-arms the fast poll when a peer connects during the slow cadence', async () => { + const harness = createHarness(); + + await harness.advance(SYNC_POLL_FAST_MS); + harness.reply('peer-a', 0); + harness.reply('peer-b', 0); + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + + harness.connectPeer('peer-c'); + + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); + + it('re-arms the fast poll when a reconnected peer reports missing messages', async () => { + // Convergence must not wait out the slow poll: a peer answering the + // reconnect kickoff with ids we lack re-arms the aggressive cadence. + const harness = createHarness(); + + await harness.advance(SYNC_POLL_FAST_MS); + harness.reply('peer-a', 0); + harness.reply('peer-b', 0); + await harness.advance(SYNC_TIMEOUT_MS); + harness.clearSent(); + + harness.reply('peer-a', 7); + + await harness.advance(SYNC_POLL_FAST_MS); + + expect(harness.inventoryRequests()).toEqual(['peer-a', 'peer-b']); + + harness.destroy(); + }); +}); + +interface SentEvent { + peerId: string; + event: ChatEvent; +} + +function createHarness() { + const actions$ = new Subject(); + const currentRoom$ = new BehaviorSubject({ id: ROOM_ID } as Room); + const peerConnected$ = new Subject(); + const undeliverablePeerIds = new Set(); + const sent: SentEvent[] = []; + const webrtc = { + getConnectedPeers: () => ['peer-a', 'peer-b'], + onPeerConnected: peerConnected$.asObservable(), + onPeerDisconnected: new Subject().asObservable(), + sendToPeer: vi.fn((peerId: string, event: ChatEvent) => { + if (undeliverablePeerIds.has(peerId)) + return false; + + sent.push({ event, peerId }); + + return true; + }) + }; + const injector = Injector.create({ + providers: [ + { + provide: Actions, + useValue: new Actions(actions$) + }, + { + provide: Store, + useValue: { + dispatch: vi.fn(), + select: (selector: unknown): Observable => { + if (selector === selectCurrentRoom) + return currentRoom$; + + throw new Error('Unexpected selector requested by MessagesSyncEffects test.'); + } + } + }, + { + provide: DatabaseService, + useValue: { + getRoomMessageStats: vi.fn(async () => ({ count: 0, lastUpdated: 0 })) + } + }, + { + provide: DebuggingService, + useValue: { warn: vi.fn() } + }, + { + provide: RealtimeSessionFacade, + useValue: webrtc + } + ] + }); + const effects = runInInjectionContext(injector, () => new MessagesSyncEffects()); + // The store loop feeds dispatched actions back into `actions$`; the sync + // cadence depends on that round trip (poll -> startSync -> timeout). + const subscriptions = [ + effects.periodicSyncPoll$.subscribe((action) => actions$.next(action)), + effects.syncTimeout$.subscribe((action) => actions$.next(action)), + effects.trackInventoryReplies$.subscribe(), + effects.restoreFastPollOnPeerConnect$.subscribe() + ]; + + return { + advance: async (ms: number): Promise => { + await vi.advanceTimersByTimeAsync(ms); + }, + clearSent: (): void => { + sent.length = 0; + }, + connectPeer: (peerId: string): void => { + peerConnected$.next(peerId); + }, + destroy: (): void => { + for (const subscription of subscriptions) { + subscription.unsubscribe(); + } + }, + inventoryRequests: (): string[] => + sent + .filter(({ event }) => event.type === 'chat-inventory-request') + .map(({ peerId }) => peerId), + reply: (peerId: string, missingIdCount: number): void => { + actions$.next(MessagesActions.peerInventoryCompared({ + missingIdCount, + peerId, + roomId: ROOM_ID + })); + }, + setDeliverable: (peerId: string): void => { + undeliverablePeerIds.delete(peerId); + }, + setUndeliverable: (peerId: string): void => { + undeliverablePeerIds.add(peerId); + } + }; +} diff --git a/toju-app/src/app/store/messages/messages-sync.effects.ts b/toju-app/src/app/store/messages/messages-sync.effects.ts index 75741fc..64c6e8d 100644 --- a/toju-app/src/app/store/messages/messages-sync.effects.ts +++ b/toju-app/src/app/store/messages/messages-sync.effects.ts @@ -8,7 +8,7 @@ * Extracted from the monolithic MessagesEffects to keep each * class focused on a single concern. */ -/* eslint-disable @typescript-eslint/member-ordering */ + import { Injectable, inject } from '@angular/core'; import { Actions, @@ -36,7 +36,6 @@ import { } from 'rxjs/operators'; import { MessagesActions } from './messages.actions'; import { RoomsActions } from '../rooms/rooms.actions'; -import { selectMessagesSyncing } from './messages.selectors'; import { selectCurrentRoom } from '../rooms/rooms.selectors'; import { RealtimeSessionFacade } from '../../core/realtime'; import { DatabaseService } from '../../infrastructure/persistence'; @@ -46,6 +45,12 @@ import { SYNC_POLL_SLOW_MS, SYNC_TIMEOUT_MS } from './messages.helpers'; +import { + createInventoryRound, + isInventoryRoundClean, + recordInventoryReply, + type InventoryRound +} from '../../domains/chat/domain/rules/message-sync-round.rules'; @Injectable() export class MessagesSyncEffects { @@ -55,9 +60,12 @@ export class MessagesSyncEffects { private readonly debugging = inject(DebuggingService); private readonly webrtc = inject(RealtimeSessionFacade); - /** Tracks whether the last sync cycle found no new messages. */ + /** Tracks whether the last sync cycle proved there was nothing left to fetch. */ private lastSyncClean = false; + /** Evidence collected for the inventory round the current poll is waiting on. */ + private currentRound: InventoryRound | null = null; + /** Subject to reset the periodic sync timer. */ private readonly syncReset$ = new Subject(); @@ -157,6 +165,7 @@ export class MessagesSyncEffects { ofType(RoomsActions.joinRoomSuccess, RoomsActions.viewServerSuccess), tap(() => { this.lastSyncClean = false; + this.currentRound = null; this.syncReset$.next(); }) ), @@ -166,18 +175,18 @@ export class MessagesSyncEffects { /** * Alternates between fast (10 s) and slow (15 min) sync intervals. * Sends inventory requests to all connected peers for the active room. + * + * Each scored round re-arms this timer (see `syncTimeout$`), so the delay is + * always chosen from the verdict of the round that just closed. The `repeat` + * only keeps the loop alive for polls that never open a round at all - no + * active room, or no connected peers. */ periodicSyncPoll$ = createEffect(() => this.syncReset$.pipe( startWith(undefined), switchMap(() => - timer(SYNC_POLL_FAST_MS).pipe( - repeat({ - delay: () => - timer( - this.lastSyncClean ? SYNC_POLL_SLOW_MS : SYNC_POLL_FAST_MS - ) - }), + timer(this.nextPollDelayMs()).pipe( + repeat({ delay: () => timer(this.nextPollDelayMs()) }), withLatestFrom(this.store.select(selectCurrentRoom)), filter( ([, room]) => @@ -187,24 +196,28 @@ export class MessagesSyncEffects { const peers = this.webrtc.getConnectedPeers(); if (!room || peers.length === 0) { + this.currentRound = null; + return of(MessagesActions.syncComplete()); } + const askedPeerIds: string[] = []; + const undeliveredPeerIds: string[] = []; + for (const pid of peers) { - try { - this.webrtc.sendToPeer(pid, { - type: 'chat-inventory-request', - roomId: room.id - }); - } catch (error) { - this.debugging.warn('messages', 'Failed to request peer inventory during sync poll', { - error, - peerId: pid, - roomId: room.id - }); + if (this.requestPeerInventory(pid, room.id)) { + askedPeerIds.push(pid); + } else { + undeliveredPeerIds.push(pid); } } + this.currentRound = createInventoryRound({ + askedPeerIds, + roomId: room.id, + undeliveredPeerIds + }); + return of(MessagesActions.startSync()); }) ) @@ -213,35 +226,93 @@ export class MessagesSyncEffects { ); /** - * Auto-completes a sync cycle after a timeout if no messages arrive. - * Switches to slow polling when the cycle is clean. + * Closes a sync cycle once the round's reply window elapses, and scores the + * round: only a fully answered round with nothing missing buys the slow + * cadence. Silence, a partial answer, or an undelivered request all keep the + * fast poll, because none of them is evidence that we are caught up. */ syncTimeout$ = createEffect(() => this.actions$.pipe( ofType(MessagesActions.startSync), - switchMap(() => from( - new Promise((resolve) => setTimeout(resolve, SYNC_TIMEOUT_MS)) - )), - withLatestFrom(this.store.select(selectMessagesSyncing)), - filter(([, syncing]) => syncing), + switchMap(() => timer(SYNC_TIMEOUT_MS)), map(() => { - this.lastSyncClean = true; + this.lastSyncClean = isInventoryRoundClean(this.currentRound); + this.currentRound = null; + this.syncReset$.next(); + return MessagesActions.syncComplete(); }) ) ); /** - * When a peer (re)connects, revert to aggressive polling in case - * we missed messages while disconnected. + * Records each peer's inventory comparison against the open round. + * + * A reply that arrives after the round closed - typically a peer answering + * the reconnect or room-activation kickoff - still proves we are behind, so + * it re-arms the fast cadence instead of leaving us parked on a slow poll + * that was scheduled while we looked converged. */ - syncReceivedMessages$ = createEffect( + trackInventoryReplies$ = createEffect( () => - this.webrtc.onPeerConnected.pipe( - tap(() => { - this.lastSyncClean = false; + this.actions$.pipe( + ofType(MessagesActions.peerInventoryCompared), + tap(({ missingIdCount, peerId, roomId }) => { + if (this.currentRound) { + this.currentRound = recordInventoryReply(this.currentRound, { + missingIdCount, + peerId, + roomId + }); + + return; + } + + if (missingIdCount > 0) { + this.lastSyncClean = false; + this.syncReset$.next(); + } }) ), { dispatch: false } ); + + /** + * When a peer (re)connects, revert to aggressive polling in case we missed + * messages while disconnected. The pending delay was chosen for the old peer + * set, so re-arm it as well instead of only flipping the flag. + */ + restoreFastPollOnPeerConnect$ = createEffect( + () => + this.webrtc.onPeerConnected.pipe( + tap(() => { + this.lastSyncClean = false; + this.syncReset$.next(); + }) + ), + { dispatch: false } + ); + + /** Delay before the next inventory poll, from the last round's verdict. */ + private nextPollDelayMs(): number { + return this.lastSyncClean ? SYNC_POLL_SLOW_MS : SYNC_POLL_FAST_MS; + } + + /** Sends one inventory request and reports whether the peer received it. */ + private requestPeerInventory(peerId: string, roomId: string): boolean { + try { + return this.webrtc.sendToPeer(peerId, { + type: 'chat-inventory-request', + roomId + }); + } catch (error) { + this.debugging.warn('messages', 'Failed to request peer inventory during sync poll', { + error, + peerId, + roomId + }); + + return false; + } + } } diff --git a/toju-app/src/app/store/messages/messages.actions.ts b/toju-app/src/app/store/messages/messages.actions.ts index 9ba6af6..80c2cee 100644 --- a/toju-app/src/app/store/messages/messages.actions.ts +++ b/toju-app/src/app/store/messages/messages.actions.ts @@ -86,6 +86,16 @@ export const MessagesActions = createActionGroup({ 'Start Sync': emptyProps(), /** Marks the end of a message sync cycle. */ 'Sync Complete': emptyProps(), + /** + * One peer answered an inventory request and we compared it against local + * state. `missingIdCount` is how many of that peer's ids we lack or hold at + * a stale revision - the only evidence that a sync round converged. + */ + 'Peer Inventory Compared': props<{ + peerId: string; + roomId: string; + missingIdCount: number; + }>(), /** Attaches fetched link metadata to a message. */ 'Update Link Metadata': props<{ messageId: string; linkMetadata: LinkMetadata[] }>(),