diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index 3b66710b..b8da4c4d 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -156,6 +156,9 @@ export type MediaType = keyof typeof MEDIA_HKDF_KEY_MAPPING export const MEDIA_KEYS = Object.keys(MEDIA_PATH_MAP) as MediaType[] +/** 120s timeout for history sync stall detection, same as WA Web's handleChunkProgress / restartPausedTimer (g = 120) */ +export const HISTORY_SYNC_PAUSED_TIMEOUT_MS = 120_000 + export const MIN_PREKEY_COUNT = 5 // Moderate prekey count (upstream uses 812, reduced to balance rate limiting and availability) diff --git a/src/Socket/chats.ts b/src/Socket/chats.ts index b14caa44..362e99fb 100644 --- a/src/Socket/chats.ts +++ b/src/Socket/chats.ts @@ -2,7 +2,7 @@ import NodeCache from '@cacheable/node-cache' import { Boom } from '@hapi/boom' import { LRUCache } from 'lru-cache' import { proto } from '../../WAProto/index.js' -import { DEFAULT_CACHE_MAX_KEYS, DEFAULT_CACHE_TTLS, PROCESSABLE_HISTORY_TYPES } from '../Defaults' +import { DEFAULT_CACHE_MAX_KEYS, DEFAULT_CACHE_TTLS, HISTORY_SYNC_PAUSED_TIMEOUT_MS, PROCESSABLE_HISTORY_TYPES } from '../Defaults' import type { BotListInfo, CacheStore, @@ -42,6 +42,8 @@ import { generateProfilePicture, getHistoryMsg, isAppStateSyncIrrecoverable, + isMissingKeyError, + MAX_SYNC_ATTEMPTS, newLTHashState, processSyncAction } from '../Utils' @@ -106,6 +108,17 @@ export const makeChatsSocket = (config: SocketConfig) => { // Timeout for AwaitingInitialSync state let awaitingSyncTimeout: NodeJS.Timeout | undefined + // In-memory history sync completion tracking (resets on reconnection) + const historySyncStatus = { + initialBootstrapComplete: false, + recentSyncComplete: false + } + let historySyncPausedTimeout: NodeJS.Timeout | undefined + + // Collections blocked on missing app state sync keys (mirrors WA Web's "Blocked" state). + // When a key arrives via APP_STATE_SYNC_KEY_SHARE, these are re-synced. + const blockedCollections = new Set() + const placeholderResendCache = config.placeholderResendCache || (new NodeCache({ @@ -695,7 +708,6 @@ export const makeChatsSocket = (config: SocketConfig) => { } catch (error: any) { attemptsMap[name] = (attemptsMap[name] || 0) + 1 - const irrecoverable = isAppStateSyncIrrecoverable(error, attemptsMap[name]) const logData = { name, attempt: attemptsMap[name], @@ -705,7 +717,23 @@ export const makeChatsSocket = (config: SocketConfig) => { error: error.stack } - if (irrecoverable) { + if (isMissingKeyError(error) && attemptsMap[name] >= MAX_SYNC_ATTEMPTS) { + // WA Web treats missing keys as "Blocked" — park the collection + // until the key arrives via APP_STATE_SYNC_KEY_SHARE. + logger.warn( + logData, + `${name} blocked on missing key from v${states[name].version}, parking after ${attemptsMap[name]} attempts` + ) + blockedCollections.add(name) + collectionsToHandle.delete(name) + } else if (isMissingKeyError(error)) { + // Retry with a snapshot which may use a different key. + logger.info( + logData, + `${name} blocked on missing key from v${states[name].version}, retrying with snapshot` + ) + forceSnapshotCollections.add(name) + } else if (isAppStateSyncIrrecoverable(error, attemptsMap[name])) { logger.warn(logData, `failed to sync ${name} from v${states[name].version}, giving up`) // reset persisted version to null so the next resyncAppState call // requests a full snapshot instead of reusing the stale version that caused the error @@ -1230,6 +1258,56 @@ export const makeChatsSocket = (config: SocketConfig) => { PROCESSABLE_HISTORY_TYPES.includes(historyMsg.syncType! as proto.HistorySync.HistorySyncType) : false + if (historyMsg && shouldProcessHistoryMsg) { + const syncType = historyMsg.syncType as proto.HistorySync.HistorySyncType + + // INITIAL_BOOTSTRAP — fire immediately, no progress check (same as WA Web K function) + if ( + syncType === proto.HistorySync.HistorySyncType.INITIAL_BOOTSTRAP && + !historySyncStatus.initialBootstrapComplete + ) { + historySyncStatus.initialBootstrapComplete = true + ev.emit('messaging-history.status', { + syncType, + status: 'complete', + explicit: true + }) + } + + // RECENT with progress === 100 — explicit completion + if ( + syncType === proto.HistorySync.HistorySyncType.RECENT && + historyMsg.progress === 100 && + !historySyncStatus.recentSyncComplete + ) { + historySyncStatus.recentSyncComplete = true + clearTimeout(historySyncPausedTimeout) + historySyncPausedTimeout = undefined + ev.emit('messaging-history.status', { + syncType, + status: 'complete', + explicit: true + }) + } + + // Reset 120s paused timeout on any RECENT chunk (like WA Web's handleChunkProgress) + if (syncType === proto.HistorySync.HistorySyncType.RECENT && !historySyncStatus.recentSyncComplete) { + clearTimeout(historySyncPausedTimeout) + historySyncPausedTimeout = setTimeout(() => { + if (!historySyncStatus.recentSyncComplete) { + historySyncStatus.recentSyncComplete = true + ev.emit('messaging-history.status', { + syncType: proto.HistorySync.HistorySyncType.RECENT, + status: 'paused', + explicit: false + }) + } + + historySyncPausedTimeout = undefined + }, HISTORY_SYNC_PAUSED_TIMEOUT_MS) + } + } + // State machine: decide on sync and flush if (historyMsg && syncState === SyncState.AwaitingInitialSync) { if (awaitingSyncTimeout) { @@ -1250,6 +1328,8 @@ export const makeChatsSocket = (config: SocketConfig) => { const doAppStateSync = async () => { if (syncState === SyncState.Syncing) { + // All collections will be synced, so clear any blocked ones + blockedCollections.clear() logger.info('Doing app state sync') await resyncAppState(ALL_WA_PATCH_NAMES, true) @@ -1330,6 +1410,9 @@ export const makeChatsSocket = (config: SocketConfig) => { // Clean up app state sync key cache on connection close if (connection === 'close') { + blockedCollections.clear() + clearTimeout(historySyncPausedTimeout) + historySyncPausedTimeout = undefined appStateSyncKeyCache.clear() logger.debug('App state sync key cache cleared on connection close') } @@ -1338,6 +1421,11 @@ export const makeChatsSocket = (config: SocketConfig) => { return } + historySyncStatus.initialBootstrapComplete = false + historySyncStatus.recentSyncComplete = false + clearTimeout(historySyncPausedTimeout) + historySyncPausedTimeout = undefined + syncState = SyncState.AwaitingInitialSync logger.info('Connection is now AwaitingInitialSync, buffering events') ev.buffer() @@ -1355,7 +1443,7 @@ export const makeChatsSocket = (config: SocketConfig) => { return } - logger.info('History sync is enabled, awaiting notification with a 20s timeout.') + logger.info('First connection, awaiting history sync notification with a 20s timeout.') if (awaitingSyncTimeout) { clearTimeout(awaitingSyncTimeout) @@ -1367,10 +1455,37 @@ export const makeChatsSocket = (config: SocketConfig) => { logger.warn('Timeout in AwaitingInitialSync, forcing state to Online and flushing buffer') syncState = SyncState.Online ev.flush() + + // Increment so subsequent reconnections skip the 20s wait. + // Late-arriving history is still processed via processMessage + // regardless of the state machine phase. + const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1 + ev.emit('creds.update', { accountSyncCounter }) } }, 20_000) }) + // When an app state sync key arrives (myAppStateKeyId is set) and there are + // collections blocked on a missing key, trigger a re-sync for just those collections. + // This mirrors WA Web's Blocked → retry-on-key-arrival behavior. + ev.on('creds.update', ({ myAppStateKeyId }) => { + if (!myAppStateKeyId || blockedCollections.size === 0) { + return + } + + // If we're in the middle of a full sync, doAppStateSync handles all collections + if (syncState === SyncState.Syncing) { + blockedCollections.clear() + return + } + + const collections = [...blockedCollections] as WAPatchName[] + blockedCollections.clear() + + logger.info({ collections }, 'app state sync key arrived, re-syncing blocked collections') + resyncAppState(collections, false).catch(error => onUnexpectedError(error, 'blocked collections resync')) + }) + ev.on('lid-mapping.update', async mappings => { try { const result = await signalRepository.lidMapping.storeLIDPNMappings(mappings) diff --git a/src/Types/Events.ts b/src/Types/Events.ts index 7df99a26..d97320c0 100644 --- a/src/Types/Events.ts +++ b/src/Types/Events.ts @@ -30,8 +30,21 @@ export type BaileysEventMap = { isLatest?: boolean progress?: number | null syncType?: proto.HistorySync.HistorySyncType | null + chunkOrder?: number | null peerDataRequestSessionId?: string | null } + /** signals history sync milestones (completion or stall) per sync type */ + 'messaging-history.status': { + /** which sync phase this status refers to */ + syncType: proto.HistorySync.HistorySyncType + /** the status of this sync phase */ + status: 'complete' | 'paused' + /** + * progress === 100 was received from the server. + * when false, completion was inferred via timeout (no more chunks arriving). + */ + explicit: boolean + } /** upsert chats */ 'chats.upsert': Chat[] /** update the given chats */ @@ -176,6 +189,7 @@ export type BufferedEventData = { isLatest: boolean progress?: number | null syncType?: proto.HistorySync.HistorySyncType + chunkOrder?: number | null peerDataRequestSessionId?: string } chatUpserts: { [jid: string]: Chat } diff --git a/src/Utils/chat-utils.ts b/src/Utils/chat-utils.ts index f0afe9ba..ba98aa0e 100644 --- a/src/Utils/chat-utils.ts +++ b/src/Utils/chat-utils.ts @@ -144,19 +144,21 @@ export const ensureLTHashStateVersion = (state: LTHashState): LTHashState => { export const MAX_SYNC_ATTEMPTS = 2 /** - * Matches WA Web's SyncdFatalError classification: - * XMPP 400/404/405/406 are fatal, TypeError indicates WASM crash. + * Check if an error is a missing app state sync key. + * WA Web treats these as "Blocked" (waits for key arrival), not fatal. + * In Baileys we retry with a snapshot which may use a different key. + */ +export const isMissingKeyError = (error: any): boolean => { + return error?.data?.isMissingKey === true +} + +/** + * Determines if an app state sync error is unrecoverable. + * TypeError indicates a WASM crash; otherwise we give up after MAX_SYNC_ATTEMPTS. + * Missing keys are NOT checked here — they are handled separately as "Blocked". */ export const isAppStateSyncIrrecoverable = (error: any, attempts: number): boolean => { - const statusCode = error?.output?.statusCode - return ( - attempts >= MAX_SYNC_ATTEMPTS || - statusCode === 400 || - statusCode === 404 || - statusCode === 405 || - statusCode === 406 || - error?.name === 'TypeError' - ) + return attempts >= MAX_SYNC_ATTEMPTS || error?.name === 'TypeError' } export const encodeSyncdPatch = async ( @@ -167,7 +169,7 @@ export const encodeSyncdPatch = async ( ) => { const key = !!myAppStateKeyId ? await getAppStateSyncKey(myAppStateKeyId) : undefined if (!key) { - throw new Boom(`myAppStateKey ("${myAppStateKeyId}") not present`, { statusCode: 404 }) + throw new Boom(`myAppStateKey ("${myAppStateKeyId}") not present`, { data: { isMissingKey: true } }) } const encKeyId = Buffer.from(myAppStateKeyId, 'base64') @@ -310,8 +312,7 @@ export const decodeSyncdMutations = async ( const keyEnc = await getAppStateSyncKey(base64Key) if (!keyEnc) { throw new Boom(`failed to find key "${base64Key}" to decode mutation`, { - statusCode: 404, - data: { msgMutations } + data: { isMissingKey: true, msgMutations } }) } @@ -341,7 +342,7 @@ export const decodeSyncdPatch = async ( const base64Key = Buffer.from(msgKeyId).toString('base64') const mainKeyObj = await getAppStateSyncKey(base64Key) if (!mainKeyObj) { - throw new Boom(`failed to find key "${base64Key}" to decode patch`, { statusCode: 404, data: { msg } }) + throw new Boom(`failed to find key "${base64Key}" to decode patch`, { data: { isMissingKey: true, msg } }) } const mainKeyData = mainKeyObj.keyData @@ -513,7 +514,7 @@ export const decodeSyncdSnapshot = async ( const base64Key = Buffer.from(snapKeyId).toString('base64') const keyEnc = await getAppStateSyncKey(base64Key) if (!keyEnc) { - throw new Boom(`failed to find key "${base64Key}" to decode mutation`, { statusCode: 404 }) + throw new Boom(`failed to find key "${base64Key}" to decode mutation`, { data: { isMissingKey: true } }) } const snapKeyData = keyEnc.keyData @@ -602,7 +603,7 @@ export const decodePatches = async ( const base64Key = Buffer.from(patchKeyId).toString('base64') const keyEnc = await getAppStateSyncKey(base64Key) if (!keyEnc) { - throw new Boom(`failed to find key "${base64Key}" to decode mutation`, { statusCode: 404 }) + throw new Boom(`failed to find key "${base64Key}" to decode mutation`, { data: { isMissingKey: true } }) } const patchKeyData = keyEnc.keyData diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index bcbd5ed2..8d893487 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -967,6 +967,7 @@ function append( data.historySets.empty = false data.historySets.syncType = eventData.syncType data.historySets.progress = eventData.progress + data.historySets.chunkOrder = eventData.chunkOrder data.historySets.peerDataRequestSessionId = eventData.peerDataRequestSessionId data.historySets.isLatest = eventData.isLatest || data.historySets.isLatest @@ -1291,6 +1292,7 @@ function consolidateEvents(data: BufferedEventData) { syncType: data.historySets.syncType, progress: data.historySets.progress, isLatest: data.historySets.isLatest, + chunkOrder: data.historySets.chunkOrder, peerDataRequestSessionId: data.historySets.peerDataRequestSessionId } } diff --git a/src/Utils/process-message.ts b/src/Utils/process-message.ts index 5763c294..33b4580a 100644 --- a/src/Utils/process-message.ts +++ b/src/Utils/process-message.ts @@ -382,6 +382,7 @@ const processMessage = async ( ev.emit('messaging-history.set', { ...data, isLatest: histNotification.syncType !== proto.HistorySync.HistorySyncType.ON_DEMAND ? isLatest : undefined, + chunkOrder: histNotification.chunkOrder, peerDataRequestSessionId: histNotification.peerDataRequestSessionId }) diff --git a/src/__tests__/Utils/app-state-sync-resilience.test.ts b/src/__tests__/Utils/app-state-sync-resilience.test.ts index ccf9aa18..38dee13e 100644 --- a/src/__tests__/Utils/app-state-sync-resilience.test.ts +++ b/src/__tests__/Utils/app-state-sync-resilience.test.ts @@ -5,6 +5,7 @@ import { decodeSyncdPatch, decodeSyncdSnapshot, isAppStateSyncIrrecoverable, + isMissingKeyError, MAX_SYNC_ATTEMPTS, newLTHashState } from '../../Utils/chat-utils' @@ -12,8 +13,8 @@ import { const missingKeyFn = async () => null describe('App State Sync', () => { - describe('missing key errors throw with statusCode 404', () => { - it('decodeSyncdPatch throws 404 on missing key', async () => { + describe('missing key errors are marked with isMissingKey (Blocked in WA Web)', () => { + it('decodeSyncdPatch throws with isMissingKey on missing key', async () => { const msg: proto.ISyncdPatch = { keyId: { id: Buffer.from('missing-key') }, mutations: [], @@ -22,12 +23,15 @@ describe('App State Sync', () => { patchMac: Buffer.alloc(32) } - await expect( - decodeSyncdPatch(msg, 'regular_low', newLTHashState(), missingKeyFn, () => {}, true) - ).rejects.toMatchObject({ output: { statusCode: 404 } }) + try { + await decodeSyncdPatch(msg, 'regular_low', newLTHashState(), missingKeyFn, () => {}, true) + fail('should have thrown') + } catch (error: any) { + expect(isMissingKeyError(error)).toBe(true) + } }) - it('decodeSyncdSnapshot throws 404 on missing snapshot key', async () => { + it('decodeSyncdSnapshot throws with isMissingKey on missing snapshot key', async () => { const snapshot: proto.ISyncdSnapshot = { version: { version: 1 as any }, records: [], @@ -35,12 +39,15 @@ describe('App State Sync', () => { mac: Buffer.alloc(32) } - await expect(decodeSyncdSnapshot('regular_low', snapshot, missingKeyFn, undefined, true)).rejects.toMatchObject({ - output: { statusCode: 404 } - }) + try { + await decodeSyncdSnapshot('regular_low', snapshot, missingKeyFn, undefined, true) + fail('should have thrown') + } catch (error: any) { + expect(isMissingKeyError(error)).toBe(true) + } }) - it('decodeSyncdMutations throws 404 on missing mutation key', async () => { + it('decodeSyncdMutations throws with isMissingKey on missing mutation key', async () => { const records: proto.ISyncdRecord[] = [ { keyId: { id: Buffer.from('missing-key') }, @@ -49,29 +56,36 @@ describe('App State Sync', () => { } ] - await expect(decodeSyncdMutations(records, newLTHashState(), missingKeyFn, () => {}, true)).rejects.toMatchObject( - { output: { statusCode: 404 } } - ) + try { + await decodeSyncdMutations(records, newLTHashState(), missingKeyFn, () => {}, true) + fail('should have thrown') + } catch (error: any) { + expect(isMissingKeyError(error)).toBe(true) + } }) - it('decodeSyncdMutations throws 404 even with validateMacs=false', async () => { - const records: proto.ISyncdRecord[] = [ - { - keyId: { id: Buffer.from('missing-key') }, - value: { blob: Buffer.alloc(64) }, - index: { blob: Buffer.alloc(32) } - } - ] - - await expect( - decodeSyncdMutations(records, newLTHashState(), missingKeyFn, () => {}, false) - ).rejects.toMatchObject({ output: { statusCode: 404 } }) + it('missing key errors are NOT irrecoverable on first attempt', async () => { + const error = new Boom('missing key', { data: { isMissingKey: true } }) + expect(isMissingKeyError(error)).toBe(true) + expect(isAppStateSyncIrrecoverable(error, 1)).toBe(false) }) }) describe('isAppStateSyncIrrecoverable', () => { - it.each([400, 404, 405, 406])('should be irrecoverable for status %d on first attempt', statusCode => { - expect(isAppStateSyncIrrecoverable(new Boom('test', { statusCode }), 1)).toBe(true) + it('should NOT be irrecoverable for status 400 (dead code path removed)', () => { + expect(isAppStateSyncIrrecoverable(new Boom('test', { statusCode: 400 }), 1)).toBe(false) + }) + + it('should NOT be irrecoverable for status 404 (missing key is Blocked, not Fatal)', () => { + expect(isAppStateSyncIrrecoverable(new Boom('test', { statusCode: 404 }), 1)).toBe(false) + }) + + it('should NOT be irrecoverable for status 405', () => { + expect(isAppStateSyncIrrecoverable(new Boom('test', { statusCode: 405 }), 1)).toBe(false) + }) + + it('should NOT be irrecoverable for status 406', () => { + expect(isAppStateSyncIrrecoverable(new Boom('test', { statusCode: 406 }), 1)).toBe(false) }) it('should be irrecoverable for TypeError', () => {