Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 9e3af35b54 | |||
| a631c8c3c3 |
+23
-149
@@ -54,7 +54,7 @@ import {
|
|||||||
resolveLidToPn
|
resolveLidToPn
|
||||||
} from '../Utils'
|
} from '../Utils'
|
||||||
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
|
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
|
||||||
import processMessage, { getChatId } from '../Utils/process-message'
|
import processMessage from '../Utils/process-message'
|
||||||
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
||||||
import {
|
import {
|
||||||
type BinaryNode,
|
type BinaryNode,
|
||||||
@@ -130,18 +130,6 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
/** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */
|
/** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */
|
||||||
const notificationMutex = makeKeyedMutex()
|
const notificationMutex = makeKeyedMutex()
|
||||||
|
|
||||||
/**
|
|
||||||
* Per-chat mutex dedicated to post-upsert work (history app-state sync +
|
|
||||||
* processMessage side effects). Kept separate from `messageMutex` because
|
|
||||||
* the inbound caller already holds `messageMutex(chatId)` while running
|
|
||||||
* decrypt + upsertMessage; sharing the same mutex would let a concurrently-
|
|
||||||
* arrived message N+1 enqueue *between* msg N's outer callback and msg N's
|
|
||||||
* post-upsert task, so msg N+1's processMessage could run before msg N's
|
|
||||||
* (breaking per-chat ordering of side effects). With a separate mutex,
|
|
||||||
* post-upsert tasks enqueue strictly in upsertMessage call order.
|
|
||||||
*/
|
|
||||||
const postUpsertMutex = makeKeyedMutex()
|
|
||||||
|
|
||||||
// Timeout for AwaitingInitialSync state
|
// Timeout for AwaitingInitialSync state
|
||||||
let awaitingSyncTimeout: NodeJS.Timeout | undefined
|
let awaitingSyncTimeout: NodeJS.Timeout | undefined
|
||||||
|
|
||||||
@@ -1411,19 +1399,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
blockedCollections.clear()
|
blockedCollections.clear()
|
||||||
|
|
||||||
logger.info('Doing app state sync')
|
logger.info('Doing app state sync')
|
||||||
try {
|
await resyncAppState(ALL_WA_PATCH_NAMES, true)
|
||||||
await resyncAppState(ALL_WA_PATCH_NAMES, true)
|
|
||||||
} catch (err) {
|
|
||||||
// Failure recovery: without this, syncState would stay at Syncing
|
|
||||||
// and ev.flush() would never run, leaving the event buffer pinned
|
|
||||||
// until the buffer's own safety timeout expires. Force the state
|
|
||||||
// machine forward so live inbound events can flow even if the
|
|
||||||
// app-state resync failed (collections are already cleared, so
|
|
||||||
// blocked patches will be retried on the next creds.update tick).
|
|
||||||
syncState = SyncState.Online
|
|
||||||
ev.flush()
|
|
||||||
throw err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Sync is complete, go online and flush everything
|
// Sync is complete, go online and flush everything
|
||||||
syncState = SyncState.Online
|
syncState = SyncState.Online
|
||||||
@@ -1435,131 +1411,29 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Post-upsert work: history app-state sync + processMessage side effects.
|
await Promise.all([
|
||||||
// Awaiting here keeps `messages.upsert` pinned in the event buffer
|
(async () => {
|
||||||
// (createBufferedFunction only schedules flush after work() resolves), so
|
if (shouldProcessHistoryMsg) {
|
||||||
// the hot path detaches this work to release the emit on the next debounce
|
|
||||||
// tick.
|
|
||||||
//
|
|
||||||
// Use Promise.allSettled so the combined promise only settles after BOTH
|
|
||||||
// tasks finish. With a plain Promise.all, an early rejection from one task
|
|
||||||
// would release the keyed mutex while the other task is still mutating
|
|
||||||
// chat state — letting the next message of the same chat overtake it and
|
|
||||||
// break per-chat ordering.
|
|
||||||
//
|
|
||||||
// Returns the per-task settle status so the keyShare branch can know
|
|
||||||
// whether processMessage actually persisted the new app-state-sync key
|
|
||||||
// before triggering doAppStateSync (otherwise the sync would hit
|
|
||||||
// isMissingKeyError and park collections in blockedCollections).
|
|
||||||
const postUpsertTasks = async (): Promise<{ processMessageOk: boolean }> => {
|
|
||||||
const [historyResult, processResult] = await Promise.allSettled([
|
|
||||||
shouldProcessHistoryMsg ? doAppStateSync() : Promise.resolve(),
|
|
||||||
processMessage(msg, {
|
|
||||||
signalRepository,
|
|
||||||
shouldProcessHistoryMsg,
|
|
||||||
placeholderResendCache,
|
|
||||||
ev,
|
|
||||||
creds: authState.creds,
|
|
||||||
keyStore: authState.keys,
|
|
||||||
logger,
|
|
||||||
options: config.options,
|
|
||||||
getMessage
|
|
||||||
})
|
|
||||||
])
|
|
||||||
|
|
||||||
if (historyResult.status === 'rejected') {
|
|
||||||
logger?.warn(
|
|
||||||
{ err: historyResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
|
||||||
'history doAppStateSync failed'
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
if (processResult.status === 'rejected') {
|
|
||||||
logger?.warn(
|
|
||||||
{ err: processResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
|
||||||
'processMessage failed'
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
return { processMessageOk: processResult.status === 'fulfilled' }
|
|
||||||
}
|
|
||||||
|
|
||||||
// Use getChatId + jidNormalizedUser so the mutex key matches the chat-id
|
|
||||||
// scheme processMessage uses for chat updates (broadcasts target the
|
|
||||||
// participant). When getChatId/jidNormalizedUser yields nothing usable
|
|
||||||
// (missing or malformed JID), prefer a message-derived fallback over a
|
|
||||||
// single global 'unknown' bucket — that bucket would head-of-line block
|
|
||||||
// every malformed message behind a shared queue. msg.key.id is unique
|
|
||||||
// per message so unrelated malformed inputs no longer serialize together;
|
|
||||||
// for valid messages we still hit the normalized chat-id path, so the
|
|
||||||
// per-chat ordering guarantee is unchanged where it matters.
|
|
||||||
const rawChatId = getChatId(msg.key)
|
|
||||||
const normalizedChatId = rawChatId ? jidNormalizedUser(rawChatId) : ''
|
|
||||||
const postUpsertChatId = normalizedChatId || msg.key?.id || 'unknown'
|
|
||||||
|
|
||||||
// Wrap in `postUpsertMutex(chatId)` (a SEPARATE keyed mutex from the outer
|
|
||||||
// `messageMutex` held by the inbound caller) so per-chat ordering of
|
|
||||||
// processMessage side effects (chat.unreadCount, LID/PN mapping,
|
|
||||||
// messages.update, history downloads) is preserved across messages of the
|
|
||||||
// same chat.
|
|
||||||
//
|
|
||||||
// Why a separate mutex: if we re-used messageMutex, a concurrently-arrived
|
|
||||||
// message N+1 could enqueue on the outer mutex BEFORE msg N's post-upsert
|
|
||||||
// task gets enqueued (because N's outer callback yields on `await decrypt()`
|
|
||||||
// before reaching the inner enqueue site). The queue would then be
|
|
||||||
// [OuterN+1, InnerN, ...], so InnerN+1 would beat InnerN to processMessage.
|
|
||||||
// With its own mutex, post-upsert tasks enqueue strictly in upsertMessage
|
|
||||||
// call order (which IS message arrival order because the outer
|
|
||||||
// messageMutex serializes the upserts per-chat).
|
|
||||||
const postUpsertWork = postUpsertMutex.mutex(postUpsertChatId, postUpsertTasks)
|
|
||||||
|
|
||||||
const isKeyShareDuringSync =
|
|
||||||
!!msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing
|
|
||||||
|
|
||||||
if (isKeyShareDuringSync) {
|
|
||||||
// appStateSyncKeyShare path: processMessage persists the new app-state-sync
|
|
||||||
// key in its APP_STATE_SYNC_KEY_SHARE handler (via keyStore.transaction).
|
|
||||||
// The follow-up doAppStateSync() needs that key to decrypt patches, so
|
|
||||||
// we MUST wait for processMessage to actually succeed before kicking off
|
|
||||||
// the sync — otherwise it would hit isMissingKeyError and park
|
|
||||||
// collections in blockedCollections, regressing the very issue this
|
|
||||||
// branch was added to fix.
|
|
||||||
//
|
|
||||||
// No deadlock with the inbound caller's messageMutex because
|
|
||||||
// postUpsertMutex is a different mutex instance.
|
|
||||||
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
|
|
||||||
const { processMessageOk } = await postUpsertWork
|
|
||||||
|
|
||||||
if (!processMessageOk) {
|
|
||||||
logger?.warn(
|
|
||||||
{ messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
|
||||||
'processMessage failed during key-share — skipping doAppStateSync to avoid isMissingKeyError'
|
|
||||||
)
|
|
||||||
} else {
|
|
||||||
try {
|
|
||||||
await doAppStateSync()
|
await doAppStateSync()
|
||||||
} catch (err) {
|
|
||||||
logger?.warn(
|
|
||||||
{ err, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
|
||||||
'doAppStateSync failed after key-share persistence'
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
}
|
})(),
|
||||||
} else {
|
processMessage(msg, {
|
||||||
// `postUpsertWork` is not expected to reject — `Promise.allSettled`
|
signalRepository,
|
||||||
// inside `postUpsertTasks` never rejects, and per-task failures are
|
shouldProcessHistoryMsg,
|
||||||
// already logged inline. The defensive catch routes any truly
|
placeholderResendCache,
|
||||||
// unexpected rejection (e.g. `postUpsertMutex` internal corruption,
|
ev,
|
||||||
// future synchronous throws inside processMessage) through
|
creds: authState.creds,
|
||||||
// `onUnexpectedError` instead of letting it surface as an
|
keyStore: authState.keys,
|
||||||
// UnhandledPromiseRejection — which on Node ≥15 can terminate the
|
logger,
|
||||||
// long-running socket process.
|
options: config.options,
|
||||||
postUpsertWork.catch(err =>
|
getMessage
|
||||||
onUnexpectedError(
|
})
|
||||||
err,
|
])
|
||||||
`processing post-upsert work for message ${msg.key?.id || 'unknown'} on ${msg.key?.remoteJid || 'unknown chat'}`
|
|
||||||
)
|
// If the app state key arrives and we are waiting to sync, trigger the sync now.
|
||||||
)
|
if (msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing) {
|
||||||
|
logger.info('App state sync key arrived, triggering app state sync')
|
||||||
|
await doAppStateSync()
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
+1
-12
@@ -341,18 +341,7 @@ export const addTransactionCapability = (
|
|||||||
|
|
||||||
return result
|
return result
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
// SessionError is part of the normal Bad MAC recovery flow
|
logger.error({ error }, 'transaction failed, rolling back')
|
||||||
// (retry receipt → sender resends as pkmsg → new session within ~1.3s).
|
|
||||||
// Logging it as ERROR creates 2 noise lines per recoverable Bad MAC cycle.
|
|
||||||
// Downgrade to debug for SessionError; keep ERROR for everything else.
|
|
||||||
// The error is still re-thrown — recovery behavior is unchanged.
|
|
||||||
const errName = (error as { name?: string })?.name
|
|
||||||
if (errName === 'SessionError') {
|
|
||||||
logger.debug({ error }, 'transaction failed (SessionError — recoverable via retry receipt)')
|
|
||||||
} else {
|
|
||||||
logger.error({ error }, 'transaction failed, rolling back')
|
|
||||||
}
|
|
||||||
|
|
||||||
throw error
|
throw error
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -58,11 +58,7 @@ export const BAD_MAC_ERROR_TEXT = 'Bad MAC'
|
|||||||
export const DECRYPTION_RETRY_CONFIG = {
|
export const DECRYPTION_RETRY_CONFIG = {
|
||||||
maxRetries: 3,
|
maxRetries: 3,
|
||||||
baseDelayMs: 100,
|
baseDelayMs: 100,
|
||||||
// 'No matching sessions found' is the libsignal error when decryptWithSessions exhausts
|
sessionRecordErrors: ['No session record', 'SessionError: No session record'],
|
||||||
// all stored sessions for a JID. Same recovery flow (retry receipt → pkmsg → new session)
|
|
||||||
// — categorise it as session-record so the caller logs DEBUG on retry, ERROR only when
|
|
||||||
// retries are exhausted (instead of dumping the full stack as an unknown error).
|
|
||||||
sessionRecordErrors: ['No session record', 'SessionError: No session record', 'No matching sessions found'],
|
|
||||||
corruptedSessionErrors: ['Bad MAC', 'MessageCounterError', MISSING_KEYS_ERROR_TEXT]
|
corruptedSessionErrors: ['Bad MAC', 'MessageCounterError', MISSING_KEYS_ERROR_TEXT]
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -425,26 +421,9 @@ export const decryptMessageNode = (
|
|||||||
const isCorrupted = isCorruptedSessionError(originalError)
|
const isCorrupted = isCorruptedSessionError(originalError)
|
||||||
const isSessionRecord = isSessionRecordError(originalError)
|
const isSessionRecord = isSessionRecordError(originalError)
|
||||||
|
|
||||||
// Slim error projection — keep name/message/type for diagnosis,
|
|
||||||
// drop `stack` which adds 4-5 lines of node_modules paths per log
|
|
||||||
// for known-recoverable libsignal errors.
|
|
||||||
//
|
|
||||||
// CRITICAL: only slim for KNOWN-RECOVERABLE categories (corrupted /
|
|
||||||
// session-record). The unknown-error branch keeps the full Error so
|
|
||||||
// protobuf/parsing/runtime bugs still emit a stack trace where it
|
|
||||||
// matters most. Catches Copilot/Codex P2 review on PR #391.
|
|
||||||
const slimErr = originalError
|
|
||||||
? {
|
|
||||||
name: (originalError as { name?: string }).name,
|
|
||||||
message: (originalError as { message?: string }).message,
|
|
||||||
type: (originalError as { type?: string }).type
|
|
||||||
}
|
|
||||||
: undefined
|
|
||||||
const isRecoverableCategory = isCorrupted || isSessionRecord
|
|
||||||
|
|
||||||
const errorContext = {
|
const errorContext = {
|
||||||
key: fullMessage.key,
|
key: fullMessage.key,
|
||||||
err: isRecoverableCategory ? slimErr : originalError,
|
err: originalError,
|
||||||
messageType: tag === 'plaintext' ? 'plaintext' : attrs.type,
|
messageType: tag === 'plaintext' ? 'plaintext' : attrs.type,
|
||||||
sender,
|
sender,
|
||||||
author,
|
author,
|
||||||
|
|||||||
+2
-11
@@ -28,16 +28,7 @@ console.info = function (...args: unknown[]) {
|
|||||||
|
|
||||||
// Track errors by type + JID to avoid duplicates (using Map for better performance)
|
// Track errors by type + JID to avoid duplicates (using Map for better performance)
|
||||||
const _errorTimestamps = new Map<string, number>()
|
const _errorTimestamps = new Map<string, number>()
|
||||||
// Dedup window for repeated decrypt-error console lines (Bad MAC / Counter / etc).
|
const DEDUP_WINDOW_MS = 150
|
||||||
// Was 150ms, but retry attempts of the SAME message are typically ~300-1000ms apart,
|
|
||||||
// so the second attempt fell outside the window and double-printed.
|
|
||||||
//
|
|
||||||
// TRADE-OFF: dedup key is `errorType + JID` (no message-id). With 5s, a burst of
|
|
||||||
// errors for the SAME JID — even of slightly different categories or different
|
|
||||||
// messages — collapses to one log line every 5s. This is intentional for a noisy
|
|
||||||
// production stream; if you need per-message visibility, set BAILEYS_LOG_LEVEL=debug
|
|
||||||
// to bypass this console-side dedup and see the structured pino logs in full.
|
|
||||||
const DEDUP_WINDOW_MS = 5000
|
|
||||||
|
|
||||||
console.error = function (...args: unknown[]) {
|
console.error = function (...args: unknown[]) {
|
||||||
if (args.length > 0 && typeof args[0] === 'string') {
|
if (args.length > 0 && typeof args[0] === 'string') {
|
||||||
@@ -79,7 +70,7 @@ console.error = function (...args: unknown[]) {
|
|||||||
const lastTime = _errorTimestamps.get(dedupeKey)
|
const lastTime = _errorTimestamps.get(dedupeKey)
|
||||||
|
|
||||||
if (lastTime && now - lastTime < DEDUP_WINDOW_MS) {
|
if (lastTime && now - lastTime < DEDUP_WINDOW_MS) {
|
||||||
return // Skip duplicate within DEDUP_WINDOW_MS window
|
return // Skip duplicate within 150ms window
|
||||||
}
|
}
|
||||||
|
|
||||||
_errorTimestamps.set(dedupeKey, now)
|
_errorTimestamps.set(dedupeKey, now)
|
||||||
|
|||||||
Reference in New Issue
Block a user