fix(app): distrust windows loaded across a connection gap (#2574)

fluxer_app/src/features/channel/components/ChannelMessages.tsx

@@ -71,6 +71,8 @@ import type React from 'react';
 import {useCallback, useEffect, useMemo, useRef, useState} from 'react';
 
 const TAIL_PROBE_MAX_ATTEMPTS = 2;
+const TAIL_PROBE_MIN_INTERVAL_MS = 10_000;
+const tailProbeAttemptedAt = new Map<string, number>();
 
 const MESSAGE_LIST_FOR_DESCRIPTOR = msg({
 	message: 'Message list for {channelName}',
@@ -483,7 +485,12 @@ export const Messages = observer(function Messages({
 		if (selectedChannelId !== channel.id || tailProbeKeyRef.current === tailProbeKey) {
 			return;
 		}
+		const lastAttemptAt = tailProbeAttemptedAt.get(tailProbeKey);
+		if (lastAttemptAt != null && Date.now() - lastAttemptAt < TAIL_PROBE_MIN_INTERVAL_MS) {
+			return;
+		}
 		tailProbeKeyRef.current = tailProbeKey;
+		tailProbeAttemptedAt.set(tailProbeKey, Date.now());
 		void MessageCommands.fetchMessages(channel.id, null, tailProbeMessageId, MAX_MESSAGES_PER_CHANNEL, undefined, {
 			tailProbe: {
 				watermarkMessageId: tailWatermarkMessageId,

fluxer_app/src/features/gateway/events/EventRouter.ts

@@ -13,7 +13,6 @@ import {handleFavoriteMemeDelete} from '@app/features/expressions/events/Favorit
 import {handleFavoriteMemeUpdate} from '@app/features/expressions/events/FavoriteMemeUpdate';
 import {handleWebhooksUpdate} from '@app/features/expressions/events/WebhooksUpdate';
 import {handleReady} from '@app/features/gateway/events/GatewayReady';
-import {handleResumed} from '@app/features/gateway/events/GatewayResumed';
 import type {GatewaySocket} from '@app/features/gateway/transport/GatewaySocket';
 import {handleChannelMemberCountsUpdate} from '@app/features/guild/events/ChannelMemberCountsUpdate';
 import {handleGuildBanAdd, handleGuildBanRemove} from '@app/features/guild/events/GuildBan';
@@ -92,7 +91,6 @@ export type GatewayHandlerRegistry = Map<string, GatewayEventHandler>;
 export function createHandlerRegistry(): GatewayHandlerRegistry {
 	const registry: GatewayHandlerRegistry = new Map();
 	registry.set('READY', handleReady as GatewayEventHandler);
-	registry.set('RESUMED', handleResumed as GatewayEventHandler);
 	registry.set('AUTH_SESSION_CHANGE', handleAuthSessionChange as GatewayEventHandler);
 	registry.set('USER_UPDATE', handleUserUpdate as GatewayEventHandler);
 	registry.set('USER_SETTINGS_UPDATE', handleUserSettingsUpdate as GatewayEventHandler);

fluxer_app/src/features/gateway/events/GatewayResumed.ts

@@ -1,7 +0,0 @@
-// SPDX-License-Identifier: AGPL-3.0-or-later
-
-import Messages from '@app/features/messaging/state/MessagingMessages';
-
-export function handleResumed(): void {
-	Messages.handleResumed();
-}

fluxer_app/src/features/gateway/transport/GatewayConnection.ts

@@ -4,6 +4,7 @@ import GeoIP from '@app/features/app/state/GeoIP';
 import Initialization from '@app/features/app/state/Initialization';
 import RuntimeConfig from '@app/features/app/state/RuntimeConfig';
 import RuntimeCrash from '@app/features/app/state/RuntimeCrash';
+import Channels from '@app/features/channel/state/Channels';
 import FavoriteMemes from '@app/features/expressions/state/FavoriteMemes';
 import {
 	createHandlerRegistry,
@@ -23,10 +24,12 @@ import {
 	GatewayState,
 	type GatewayVoiceStateUpdateParams,
 } from '@app/features/gateway/transport/GatewaySocket';
+import {selectGuildActivationTarget} from '@app/features/gateway/transport/GuildActivationTarget';
 import GuildMatureContentAgree from '@app/features/guild/state/GuildMatureContentAgree';
 import GuildMembers from '@app/features/member/state/GuildMembers';
 import MemberSearch from '@app/features/member/state/MemberSearch';
 import Messages from '@app/features/messaging/state/MessagingMessages';
+import Navigation from '@app/features/navigation/state/Navigation';
 import SelectedGuild from '@app/features/navigation/state/SelectedGuild';
 import Permission from '@app/features/permissions/state/Permission';
 import SessionManager from '@app/features/platform/state/AuthSession';
@@ -55,6 +58,7 @@ interface DesiredSession {
 class GatewayConnection {
 	socket: GatewaySocket | null = null;
 	isConnected: boolean = false;
+	connectionEpoch: number = 0;
 	isConnecting: boolean = false;
 	isReady: boolean = false;
 	sessionId: string | null = null;
@@ -166,11 +170,11 @@ class GatewayConnection {
 		deferUntilModulesLoaded(() => {
 			reaction(
 				() => ({
-					guildId: SelectedGuild.selectedGuildId,
+					guildId: this.activationGuildId,
 					nonce: SelectedGuild.selectionNonce,
 				}),
 				({guildId}) => {
-					if (!guildId || guildId === FAVORITES_GUILD_ID) {
+					if (!guildId) {
 						this.pendingGuildSyncId = null;
 						return;
 					}
@@ -339,6 +343,9 @@ class GatewayConnection {
 				if (!isCurrent()) {
 					return;
 				}
+				if (newState === GatewayState.Connected || previousState === GatewayState.Connected) {
+					this.connectionEpoch += 1;
+				}
 				this.isConnected = newState === GatewayState.Connected;
 				this.isConnecting = newState === GatewayState.Connecting || newState === GatewayState.Reconnecting;
 				if (newState === GatewayState.Connected) {
@@ -445,8 +452,17 @@ class GatewayConnection {
 		}
 	}
 
+	private get activationGuildId(): string | null {
+		const channelId = Navigation.channelId;
+		const channel = channelId ? Channels.getChannel(channelId) : undefined;
+		return selectGuildActivationTarget({
+			selectedGuildId: SelectedGuild.selectedGuildId,
+			openChannelGuildId: channel?.guildId ?? null,
+		});
+	}
+
 	private flushPendingGuildSync(): void {
-		const guildId = this.pendingGuildSyncId ?? SelectedGuild.selectedGuildId;
+		const guildId = this.pendingGuildSyncId ?? this.activationGuildId;
 		if (!guildId || guildId === FAVORITES_GUILD_ID) {
 			return;
 		}

fluxer_app/src/features/gateway/transport/GuildActivationTarget.test.ts

@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {FAVORITES_GUILD_ID} from '@fluxer/constants/src/AppConstants';
+import {describe, expect, it} from 'vitest';
+import {selectGuildActivationTarget} from './GuildActivationTarget';
+
+describe('selectGuildActivationTarget', () => {
+	it('returns the selected guild on a guild route', () => {
+		expect(selectGuildActivationTarget({selectedGuildId: '1', openChannelGuildId: '1'})).toBe('1');
+	});
+
+	it('returns the selected guild when no channel is open', () => {
+		expect(selectGuildActivationTarget({selectedGuildId: '1', openChannelGuildId: null})).toBe('1');
+	});
+
+	it('returns the owning guild of a channel opened from the favorites route', () => {
+		expect(selectGuildActivationTarget({selectedGuildId: null, openChannelGuildId: '1'})).toBe('1');
+	});
+
+	it('ignores the favorites pseudo guild and falls through to the open channel', () => {
+		expect(selectGuildActivationTarget({selectedGuildId: FAVORITES_GUILD_ID, openChannelGuildId: '1'})).toBe('1');
+	});
+
+	it('returns null on the favorites route while no channel is open', () => {
+		expect(selectGuildActivationTarget({selectedGuildId: FAVORITES_GUILD_ID, openChannelGuildId: null})).toBeNull();
+	});
+
+	it('returns null for a private channel', () => {
+		expect(selectGuildActivationTarget({selectedGuildId: null, openChannelGuildId: null})).toBeNull();
+	});
+});

fluxer_app/src/features/gateway/transport/GuildActivationTarget.ts

@@ -0,0 +1,13 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {FAVORITES_GUILD_ID} from '@fluxer/constants/src/AppConstants';
+
+export function selectGuildActivationTarget(input: {
+	selectedGuildId: string | null;
+	openChannelGuildId: string | null;
+}): string | null {
+	if (input.selectedGuildId && input.selectedGuildId !== FAVORITES_GUILD_ID) {
+		return input.selectedGuildId;
+	}
+	return input.openChannelGuildId;
+}

fluxer_app/src/features/messaging/commands/MessageCommands.tsx

@@ -8,6 +8,7 @@ import {Endpoints} from '@app/features/app/constants/Endpoints';
 import Authentication from '@app/features/auth/state/Authentication';
 import Channels from '@app/features/channel/state/Channels';
 import DeveloperOptions from '@app/features/devtools/state/DeveloperOptions';
+import GatewayConnection from '@app/features/gateway/transport/GatewayConnection';
 import GuildMatureContentAgree from '@app/features/guild/state/GuildMatureContentAgree';
 import {DELETE_MESSAGE_DESCRIPTOR} from '@app/features/i18n/utils/CommonMessageDescriptors';
 import GuildMembers from '@app/features/member/state/GuildMembers';
@@ -15,6 +16,7 @@ import {
 	type MessageFetchCacheHit,
 	resolveMessageFetchExecutionDecision,
 	resolveMessageFetchPreflightDecision,
+	resolveMessageFetchWindowCached,
 } from '@app/features/messaging/commands/MessageFetchStateMachine';
 import {resolveMessagePageState} from '@app/features/messaging/commands/MessagePageStateMachine';
 import {MessageDeleteFailedModal} from '@app/features/messaging/components/alerts/MessageDeleteFailedModal';
@@ -258,6 +260,7 @@ function handleMessageFetchSuccess(
 	channelId: string,
 	messages: Array<WireMessage>,
 	pageState: MessagePageState,
+	cached: boolean,
 	jump?: JumpOptions,
 	tailProbe?: TailProbeContext,
 ): void {
@@ -268,7 +271,7 @@ function handleMessageFetchSuccess(
 		isAfter: pageState.isAfter,
 		hasMoreBefore: pageState.hasMoreBefore,
 		hasMoreAfter: pageState.hasMoreAfter,
-		cached: false,
+		cached,
 		jump,
 		tailProbe: tailProbe != null,
 	});
@@ -466,17 +469,25 @@ export async function fetchMessages(
 			const started = Messages.getMessages(channelId);
 			probeEpoch = {loadGeneration: started.loadGeneration, jumpTicket: started.jumpTicket};
 		}
+		const connectedAtRequest = GatewayConnection.isConnected;
+		const epochAtRequest = GatewayConnection.connectionEpoch;
 		try {
 			const timeStart = Date.now();
 			logger.debug(`Fetching messages for channel ${channelId}`);
 			const messages = await requestChannelMessages(channelId, before, after, limit, jump);
+			const cached = resolveMessageFetchWindowCached({
+				connectedAtRequest,
+				connectedAtResponse: GatewayConnection.isConnected,
+				epochAtRequest,
+				epochAtResponse: GatewayConnection.connectionEpoch,
+			});
 			if (probeEpoch != null && !isTailProbeApplicable(channelId, probeEpoch, after)) {
 				settleTailProbe(options, 'retry');
 				return [];
 			}
 			const pageState = calculateMessagePageState(channelId, before, after, limit, messages, jump);
 			logger.info(`Fetched ${messages.length} messages for channel ${channelId}, took ${Date.now() - timeStart}ms`);
-			handleMessageFetchSuccess(channelId, messages, pageState, jump, options?.tailProbe);
+			handleMessageFetchSuccess(channelId, messages, pageState, cached, jump, options?.tailProbe);
 			settleTailProbe(options, 'applied');
 			return messages;
 		} catch (error) {

fluxer_app/src/features/messaging/commands/MessageFetchStateMachine.test.ts

@@ -8,6 +8,7 @@ import {
 	type MessageFetchPreflightInput,
 	resolveMessageFetchExecutionDecision,
 	resolveMessageFetchPreflightDecision,
+	resolveMessageFetchWindowCached,
 	selectMessageFetchExecutionDecision,
 	selectMessageFetchPreflightDecision,
 	transitionMessageFetchExecutionSnapshot,
@@ -104,3 +105,31 @@ describe('messageFetchExecutionMachine', () => {
 		expect(selectMessageFetchExecutionDecision(networkSnapshot)).toEqual({type: 'requestNetwork'});
 	});
 });
+
+describe('resolveMessageFetchWindowCached', () => {
+	function trust(overrides: Partial<Parameters<typeof resolveMessageFetchWindowCached>[0]> = {}) {
+		return {
+			connectedAtRequest: true,
+			connectedAtResponse: true,
+			epochAtRequest: 7,
+			epochAtResponse: 7,
+			...overrides,
+		};
+	}
+
+	it('trusts a window loaded inside one uninterrupted connection', () => {
+		expect(resolveMessageFetchWindowCached(trust())).toBe(false);
+	});
+
+	it('distrusts a window whose request left while the socket was down', () => {
+		expect(resolveMessageFetchWindowCached(trust({connectedAtRequest: false}))).toBe(true);
+	});
+
+	it('distrusts a window whose response landed while the socket was down', () => {
+		expect(resolveMessageFetchWindowCached(trust({connectedAtResponse: false}))).toBe(true);
+	});
+
+	it('distrusts a window whose connection epoch moved under it', () => {
+		expect(resolveMessageFetchWindowCached(trust({epochAtResponse: 8}))).toBe(true);
+	});
+});

fluxer_app/src/features/messaging/commands/MessageFetchStateMachine.ts

@@ -197,3 +197,15 @@ export function selectMessageFetchExecutionDecision(
 export function resolveMessageFetchExecutionDecision(input: MessageFetchExecutionInput): MessageFetchExecutionDecision {
 	return buildExecutionDecision(input);
 }
+
+export interface MessageFetchWindowTrustInput {
+	connectedAtRequest: boolean;
+	connectedAtResponse: boolean;
+	epochAtRequest: number;
+	epochAtResponse: number;
+}
+
+export function resolveMessageFetchWindowCached(input: MessageFetchWindowTrustInput): boolean {
+	if (!input.connectedAtRequest || !input.connectedAtResponse) return true;
+	return input.epochAtRequest !== input.epochAtResponse;
+}

fluxer_app/src/features/messaging/state/ChannelMessages.ts

@@ -3,7 +3,10 @@
 import {Message} from '@app/features/messaging/models/MessagingMessage';
 import {UploadingAttachment} from '@app/features/messaging/models/UploadingAttachment';
 import {resolveChannelIncomingMessageDecision} from '@app/features/messaging/state/ChannelIncomingMessageStateMachine';
-import {resolveChannelMessagesLoadDecision} from '@app/features/messaging/state/ChannelMessagesLoadStateMachine';
+import {
+	resolveChannelMessagesLoadDecision,
+	selectChannelMessagesLoadRestoresTrust,
+} from '@app/features/messaging/state/ChannelMessagesLoadStateMachine';
 import MessageReactions from '@app/features/messaging/state/MessageReactions';
 import {mergeAscendingById} from '@app/features/messaging/utils/MessagePaginationUtils';
 import SelectedChannel from '@app/features/navigation/state/SelectedChannel';
@@ -358,6 +361,10 @@ export class ChannelMessages {
 		ChannelMessages.retainedChannelIds.delete(channelId);
 	}
 
+	static isRetained(channelId: string): boolean {
+		return ChannelMessages.retainedChannelIds.has(channelId);
+	}
+
 	static dropBuffers(channelId: string): void {
 		const instance = ChannelMessages.channelCache.get(channelId);
 		if (!instance) return;
@@ -900,6 +907,11 @@ export class ChannelMessages {
 					jumpReturnChannelId: jump?.returnToMessageId ? (jump.returnChannelId ?? this.channelId) : null,
 					jumpReturnGuildId: jump?.returnToMessageId ? (jump.returnGuildId ?? null) : null,
 				};
+		const reachesLiveEdge = selectChannelMessagesLoadRestoresTrust({
+			mode: loadDecision.mode,
+			isAfter,
+			hasMoreAfter,
+		});
 		next = next.cloneAnd({
 			ready: true,
 			loadingMore: false,
@@ -907,7 +919,7 @@ export class ChannelMessages {
 			...jumpPatch,
 			hasMoreBefore: loadDecision.preserveHasMoreBefore ? next.hasMoreBefore : hasMoreBefore,
 			hasMoreAfter: loadDecision.preserveHasMoreAfter ? next.hasMoreAfter : hasMoreAfter,
-			cached,
+			cached: reachesLiveEdge ? cached : next.cached || cached,
 			error: false,
 		});
 		return next;

fluxer_app/src/features/messaging/state/ChannelMessagesLoadStateMachine.test.ts

@@ -11,6 +11,7 @@ import {
 	resolveChannelMessagesWindowStatus,
 	selectChannelMessagesFillerVisible,
 	selectChannelMessagesLoadDecision,
+	selectChannelMessagesLoadRestoresTrust,
 	selectChannelMessagesSpacerHeight,
 	selectChannelMessagesTailGapId,
 	selectChannelMessagesTailProbeId,
@@ -518,3 +519,23 @@ describe('selectChannelMessagesTailProbeOutcome', () => {
 		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({currentJumpTicket: 3}))).toBe('discard');
 	});
 });
+
+describe('selectChannelMessagesLoadRestoresTrust', () => {
+	it('trusts a window a replacement load rebuilt from scratch', () => {
+		expect(selectChannelMessagesLoadRestoresTrust({mode: 'replace', isAfter: false, hasMoreAfter: false})).toBe(true);
+	});
+
+	it('trusts a window whose tail merge caught up to the live edge', () => {
+		expect(selectChannelMessagesLoadRestoresTrust({mode: 'mergeAfter', isAfter: true, hasMoreAfter: false})).toBe(true);
+	});
+
+	it('keeps distrusting a tail merge that stopped short of the live edge', () => {
+		expect(selectChannelMessagesLoadRestoresTrust({mode: 'mergeAfter', isAfter: true, hasMoreAfter: true})).toBe(false);
+	});
+
+	it('keeps distrusting a page of older history', () => {
+		expect(selectChannelMessagesLoadRestoresTrust({mode: 'mergeBefore', isAfter: false, hasMoreAfter: false})).toBe(
+			false,
+		);
+	});
+});

fluxer_app/src/features/messaging/state/ChannelMessagesLoadStateMachine.ts

@@ -301,6 +301,15 @@ export function selectChannelMessagesTailProbeId(input: ChannelMessagesTailInput
 	return selectChannelMessagesTailGapId(input);
 }
 
+export function selectChannelMessagesLoadRestoresTrust(input: {
+	mode: ChannelMessagesLoadMode;
+	isAfter: boolean;
+	hasMoreAfter: boolean;
+}): boolean {
+	if (input.mode === 'replace') return true;
+	return input.isAfter && !input.hasMoreAfter;
+}
+
 export interface ChannelMessagesTailProbeResultInput {
 	probeGeneration: number;
 	currentGeneration: number;

fluxer_app/src/features/messaging/state/MessagingMessages.ts

@@ -201,7 +201,7 @@ class Messages {
 	}
 
 	private hasLoadedPage(messages: ChannelMessages): boolean {
-		return messages.ready && messages.length > 0;
+		return messages.ready && messages.length > 0 && !messages.cached;
 	}
 
 	shouldPreloadLatestPage(channelId: string): boolean {
@@ -209,7 +209,9 @@ class Messages {
 			return false;
 		}
 		const messages = ChannelMessages.get(channelId);
-		return !messages || (messages.length === 0 && !messages.loadingMore && !messages.ready);
+		if (!messages) return true;
+		if (messages.loadingMore || ChannelMessages.isRetained(channelId)) return false;
+		return messages.length === 0 ? !messages.ready : messages.cached;
 	}
 
 	@action
@@ -286,15 +288,6 @@ class Messages {
 		return true;
 	}
 
-	@action
-	handleResumed(): boolean {
-		ChannelMessages.forEach((messages) => {
-			this.commitMessages(messages.withPatch({ready: true}));
-		});
-		this.notifyChange();
-		return true;
-	}
-
 	@action
 	handleGatewayReady(): boolean {
 		const selectedChannelId = SelectedChannel.currentChannelId;
@@ -386,9 +379,10 @@ class Messages {
 		if (!isNonGuildChannel && !guildExists) {
 			return false;
 		}
+		const distrustedTailId = messages.ready && messages.cached ? (messages.last()?.id ?? null) : null;
 		this.commitMessages(messages.withPatch({loadingMore: true}));
 		this.notifyChange();
-		MessageCommands.fetchMessages(channelId, null, null, MAX_MESSAGES_PER_CHANNEL);
+		MessageCommands.fetchMessages(channelId, null, distrustedTailId, MAX_MESSAGES_PER_CHANNEL);
 		return false;
 	}
 

fluxer_app/src/features/read_state/state/ReadStates.test.ts

@@ -92,14 +92,22 @@ describe('ReadStates unread invariant', () => {
 		expect(ReadStates.hasUnread(channelId)).toBe(true);
 	});
 
-	it('clears a stale unread once the server walks the last message id back', () => {
+	it('ignores a passive update that walks the last message id back', () => {
 		const {channelId} = seedReadChannel();
 		ReadStates.handlePassiveLastMessageUpdates({[channelId]: ID.newer}, 'guild-1');
 		expect(ReadStates.hasUnread(channelId)).toBe(true);
 		ReadStates.handlePassiveLastMessageUpdates({[channelId]: ID.ack}, 'guild-1');
+		expect(ReadStates.lastMessageId(channelId)).toBe(ID.newer);
+		expect(ReadStates.hasUnread(channelId)).toBe(true);
+	});
+
+	it('still lets its own probe lower a watermark a passive update raised', () => {
+		const {channelId} = seedReadChannel();
+		loadedMessages.push({id: ID.ack, author: {id: 'someone'}});
+		ReadStates.handlePassiveLastMessageUpdates({[channelId]: ID.newer}, 'guild-1');
+		ReadStates.handleLoadMessages({channelId, isAfter: true, messages: [], tailProbeWatermarkId: ID.newer});
+		expect(ReadStates.lastMessageId(channelId)).toBe(ID.ack);
 		expect(ReadStates.hasUnread(channelId)).toBe(false);
-		expect(ReadStates.getUnreadCount(channelId)).toBe(0);
-		expect(ReadStates.getVisualUnreadMessageId(channelId)).toBeNull();
 	});
 
 	it('lowers a watermark its own probe finds nothing behind', () => {

fluxer_app/src/features/read_state/state/ReadStates.ts

@@ -599,7 +599,7 @@ class ReadStates {
 				changed = state.guildId !== guildId;
 				state.storedGuildId = guildId;
 			}
-			if (lastMessageId !== state.lastMessageId) {
+			if (isNewerMessageId(lastMessageId, state.lastMessageId)) {
 				state.lastMessageId = lastMessageId;
 				changed = true;
 				this.clearUnreadStateIfRead(state);