fix(app-state) - missing key Blocked state, history sync status & 120s timeout

fix(app-state) - missing key Blocked state, history sync status & 120s timeout

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)

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<WAPatchName>()
+
 	const placeholderResendCache =
 		config.placeholderResendCache ||
 		(new NodeCache<number>({
@@ -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)

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 }

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

src/Utils/event-buffer.ts

@@ -967,6 +967,7 @@ function append<E extends BufferableEvent>(
 			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
 		}
 	}

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
 					})
 

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', () => {