From 2e3f78b3c6f6fef32091f912bdcac89665c1d769 Mon Sep 17 00:00:00 2001 From: Hampus Date: Fri, 11 Sep 2026 16:26:34 +0200 Subject: [PATCH] fix(app): stop restarting the read-state ack batch window (#2689) --- .../features/read_state/state/ReadStates.ts | 1 - .../state/ReadStatesAckBatching.test.ts | 167 ++++++++++++++++++ 2 files changed, 167 insertions(+), 1 deletion(-) create mode 100644 fluxer_app/src/features/read_state/state/ReadStatesAckBatching.test.ts diff --git a/fluxer_app/src/features/read_state/state/ReadStates.ts b/fluxer_app/src/features/read_state/state/ReadStates.ts index ed12c855a..fc4bb0f46 100644 --- a/fluxer_app/src/features/read_state/state/ReadStates.ts +++ b/fluxer_app/src/features/read_state/state/ReadStates.ts @@ -867,7 +867,6 @@ class ReadStates { state.ackedManually = false; state.clearStickyUnread(); } - this.cancelPendingAckIfCovered(state.channelId, decision.messageId); return {acked: true, messageId: decision.messageId, hadMentions: decision.hadMentions}; } diff --git a/fluxer_app/src/features/read_state/state/ReadStatesAckBatching.test.ts b/fluxer_app/src/features/read_state/state/ReadStatesAckBatching.test.ts new file mode 100644 index 000000000..8b5e90370 --- /dev/null +++ b/fluxer_app/src/features/read_state/state/ReadStatesAckBatching.test.ts @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {Endpoints} from '@app/features/app/constants/Endpoints'; +import {Channel} from '@app/features/channel/models/Channel'; +import {ACK_BATCH_DELAY_MS, type GatewayReadState} from '@app/features/read_state/state/read_states/shared'; +import {ChannelTypes, MessageTypes} from '@fluxer/constants/src/ChannelConstants'; +import type {Channel as WireChannel} from '@fluxer/schema/src/domains/channel/ChannelSchemas'; +import type {Message as WireMessage} from '@fluxer/schema/src/domains/message/MessageResponseSchemas'; +import type {UserPartial} from '@fluxer/schema/src/domains/user/UserResponseSchemas'; +import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest'; + +const channels = new Map(); +let pinnedToEnd = false; +let automaticAck = false; +const loadedMessages: Array<{id: string; author: {id: string}}> = []; + +vi.mock('@app/features/app/state/RuntimeConfig', () => ({default: {localInstanceDomain: 'fluxer.test'}})); +vi.mock('@app/features/channel/state/Channels', () => ({ + default: {getChannel: (id: string) => channels.get(id)}, +})); +vi.mock('@app/features/messaging/state/MessagingMessages', () => ({ + default: { + getMessages: () => ({ + get hasMoreBefore() { + return false; + }, + get length() { + return loadedMessages.length; + }, + jumpDestinationId: null, + hasNewestMessages: () => true, + has: (id: string) => loadedMessages.some((m) => m.id === id), + last: () => loadedMessages[loadedMessages.length - 1], + forEachBuffered: (cb: (m: unknown) => void) => { + for (const m of loadedMessages) cb(m); + }, + }), + }, +})); +vi.mock('@app/features/user/state/Users', () => ({ + default: {getCurrentUser: () => ({id: 'me'}), cacheUsers: () => {}}, +})); +vi.mock('@app/features/relationship/state/Relationships', () => ({default: {isBlocked: () => false}})); +vi.mock('@app/features/member/state/GuildMembers', () => ({default: {getMember: () => null}})); +vi.mock('@app/features/user/state/UserGuildSettings', () => ({ + default: { + isEveryoneMentionSuppressed: () => false, + isRoleMentionSuppressed: () => false, + isGuildOrChannelMuted: () => false, + }, +})); +vi.mock('@app/features/ui/state/Dimension', () => ({default: {channelPinnedToEnd: () => pinnedToEnd}})); +vi.mock('@app/features/notification/state/NotificationAutoAck', () => ({ + default: {isAutomaticAckEnabled: () => automaticAck, disableForChannel: () => {}}, +})); +vi.mock('@app/features/platform/transport/RestTransport', () => ({ + http: {post: vi.fn(async () => ({body: {read_states: []}})), get: vi.fn()}, +})); + +const {default: ReadStates} = await import('@app/features/read_state/state/ReadStates'); +const {http} = await import('@app/features/platform/transport/RestTransport'); + +const CHANNEL = '1485064866382176262'; +const MESSAGE = { + acked: '1547700000000000000', + unread: '1547743000000000000', +}; + +function user(id: string): UserPartial { + return {id, username: id, discriminator: '0', global_name: null, avatar: null, avatar_color: null, flags: 0}; +} + +function guildText(lastMessageId: string): WireChannel { + return {id: CHANNEL, type: ChannelTypes.GUILD_TEXT, guild_id: 'guild-1', last_message_id: lastMessageId}; +} + +function readState(ackMessageId: string): GatewayReadState { + return {id: CHANNEL, last_message_id: ackMessageId, mention_count: 0, version: '1'}; +} + +function ready(): void { + channels.clear(); + channels.set(CHANNEL, new Channel(guildText(MESSAGE.unread))); + ReadStates.handleGatewayReady({readState: [readState(MESSAGE.acked)], channels: [guildText(MESSAGE.unread)]}); +} + +function wireMessage(id: string, authorId: string): WireMessage { + return { + id, + channel_id: CHANNEL, + author: user(authorId), + type: MessageTypes.DEFAULT, + flags: 0, + pinned: false, + tts: false, + mention_everyone: false, + content: 'test', + timestamp: '2026-09-11T00:00:00.000Z', + mentions: [], + mention_roles: [], + } as unknown as WireMessage; +} + +function openChannel(): void { + loadedMessages.push({id: MESSAGE.unread, author: {id: 'tuna'}}); + ReadStates.handleLoadMessages({channelId: CHANNEL, isAfter: false, messages: [{id: MESSAGE.unread}] as never}); + pinnedToEnd = true; + automaticAck = true; + ReadStates.handleChannelAckWithStickyUnread({channelId: CHANNEL}); +} + +describe('ReadStates ack batching', () => { + beforeEach(() => { + vi.useFakeTimers(); + loadedMessages.length = 0; + pinnedToEnd = false; + automaticAck = false; + vi.mocked(http.post).mockClear(); + }); + + afterEach(() => { + ReadStates.clearAll(); + vi.useRealTimers(); + }); + + it('sends the ack one batch delay after a quiet channel is read', async () => { + ready(); + expect(ReadStates.hasUnread(CHANNEL)).toBe(true); + openChannel(); + expect(ReadStates.hasUnread(CHANNEL)).toBe(false); + await vi.advanceTimersByTimeAsync(ACK_BATCH_DELAY_MS - 1); + expect(http.post).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(http.post).toHaveBeenCalledWith(Endpoints.READ_STATES_ACK, { + body: {read_states: [{channel_id: CHANNEL, message_id: MESSAGE.unread}]}, + }); + }); + + it('keeps the batch window anchored to the first ack while messages keep arriving', async () => { + ready(); + openChannel(); + let messageId = 1547743900000000000n; + for (let i = 0; i < 3; i++) { + await vi.advanceTimersByTimeAsync(900); + messageId += 1000000n; + ReadStates.handleIncomingMessage({channelId: CHANNEL, message: wireMessage(messageId.toString(), 'tuna')}); + } + expect(http.post).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(ACK_BATCH_DELAY_MS - 2700); + expect(http.post).toHaveBeenCalledTimes(1); + expect(http.post).toHaveBeenCalledWith(Endpoints.READ_STATES_ACK, { + body: {read_states: [{channel_id: CHANNEL, message_id: messageId.toString()}]}, + }); + }); + + it('keeps acking a channel read for a long stretch of steady traffic', async () => { + ready(); + openChannel(); + let messageId = 1547743900000000000n; + for (let i = 0; i < 22; i++) { + await vi.advanceTimersByTimeAsync(900); + messageId += 1000000n; + ReadStates.handleIncomingMessage({channelId: CHANNEL, message: wireMessage(messageId.toString(), 'tuna')}); + } + expect(vi.mocked(http.post).mock.calls.length).toBeGreaterThanOrEqual(3); + }); +});