From 007cd61febe6b6ff5f639eec9a2bc4347abb6ed1 Mon Sep 17 00:00:00 2001 From: harshitha-cstk Date: Mon, 13 Jul 2026 17:16:45 +0530 Subject: [PATCH 1/4] fix: recover sync loop after transient Sync API timeout --- src/api.ts | 77 +++++++++++++++++++++++++++++++++++---------- src/core/index.ts | 29 ++++++++++++++--- src/core/inet.ts | 6 +++- src/core/process.ts | 5 ++- 4 files changed, 93 insertions(+), 24 deletions(-) diff --git a/src/api.ts b/src/api.ts index 515cbfe..3c780b0 100644 --- a/src/api.ts +++ b/src/api.ts @@ -100,8 +100,26 @@ export const init = (contentstack) => { */ export const get = (req, RETRY = 1) => { return new Promise((resolve, reject) => { + // Ensure this request settles exactly once. Previously a socket timeout both + // reject()ed AND (via destroy() -> 'error') scheduled a detached retry whose + // result landed on an already-settled promise and was silently discarded. + let settled = false + let retryScheduled = false + const resolveOnce = (value) => { + if (!settled) { + settled = true + resolve(value) + } + } + const rejectOnce = (err) => { + if (!settled) { + settled = true + reject(err) + } + } + if (RETRY > MAX_RETRY_LIMIT) { - return reject(new Error('Max retry limit exceeded!')) + return rejectOnce(new Error('Max retry limit exceeded!')) } req.method = Contentstack.verbs.get req.path = req.path || Contentstack.apis.sync @@ -149,15 +167,15 @@ export const get = (req, RETRY = 1) => { .on('end', () => { debug(MESSAGES.API.STATUS(response.statusCode)) if (response.statusCode >= 200 && response.statusCode <= 399) { - return resolve(JSON.parse(body)) + return resolveOnce(JSON.parse(body)) } else if (response.statusCode === 429) { timeDelay = Math.pow(Math.SQRT2, RETRY) * RETRY_DELAY_BASE debug(MESSAGES.API.RATE_LIMIT(options.path, timeDelay)) return setTimeout(() => { return get(req, RETRY) - .then(resolve) - .catch(reject) + .then(resolveOnce) + .catch(rejectOnce) }, timeDelay) } else if (response.statusCode >= 500) { // retry, with delay @@ -167,8 +185,8 @@ export const get = (req, RETRY = 1) => { return setTimeout(() => { return get(req, RETRY) - .then(resolve) - .catch(reject) + .then(resolveOnce) + .catch(rejectOnce) }, timeDelay) } else { // Enhanced error handling for Error 141 (Invalid sync_token) @@ -203,8 +221,8 @@ export const get = (req, RETRY = 1) => { return setTimeout(() => { return get(req, RETRY) - .then(resolve) - .catch(reject) + .then(resolveOnce) + .catch(rejectOnce) }, timeDelay) } else { debug('Error 141 recovery already attempted, failing to prevent infinite loop') @@ -216,7 +234,7 @@ export const get = (req, RETRY = 1) => { } debug(MESSAGES.API.REQUEST_FAILED(options)) - return reject(body) + return rejectOnce(body) } }) }) @@ -224,17 +242,42 @@ export const get = (req, RETRY = 1) => { // Set socket timeout to handle socket hang ups httpRequest.setTimeout(options.timeout, () => { debug(MESSAGES.API.REQUEST_TIMEOUT(options.path)) - httpRequest.destroy() + if (settled) { + return + } const timeoutError = Object.assign(new Error('Request timeout'), { code: 'ETIMEDOUT', }) as Error & { code: string } - reject(timeoutError) + // Retry the timed-out request in place so a subsequent success actually + // settles THIS promise (and the sync_token/checkpoint advances). Mark + // retryScheduled so the destroy()-triggered 'error' does not double-handle. + if (RETRY < MAX_RETRY_LIMIT) { + retryScheduled = true + timeDelay = Math.pow(Math.SQRT2, RETRY) * RETRY_DELAY_BASE + RETRY++ + debug(`Request timeout: waiting ${timeDelay}ms before retry ${RETRY}/${MAX_RETRY_LIMIT}`) + httpRequest.destroy() + + return setTimeout(() => { + return get(req, RETRY) + .then(resolveOnce) + .catch(rejectOnce) + }, timeDelay) + } + httpRequest.destroy() + return rejectOnce(timeoutError) }) // Enhanced error handling for network and connection errors httpRequest.on('error', (error: any) => { debug(MESSAGES.API.REQUEST_ERROR(options.path, error?.message, error?.code)) - + + // Ignore errors once the promise has settled or a retry is already scheduled + // (e.g. the 'error' emitted by our own destroy() inside the timeout handler). + if (settled || retryScheduled) { + return + } + // List of retryable network error codes const retryableErrors = [ 'ECONNRESET', // Connection reset by peer @@ -259,17 +302,17 @@ export const get = (req, RETRY = 1) => { return setTimeout(() => { return get(req, RETRY) - .then(resolve) - .catch(reject) + .then(resolveOnce) + .catch(rejectOnce) }, timeDelay) } - - return reject(error) + + return rejectOnce(error) }) httpRequest.end() } catch (error) { - return reject(error) + return rejectOnce(error) } }) } diff --git a/src/core/index.ts b/src/core/index.ts index b0c4bef..5df60bd 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -197,6 +197,11 @@ const check = async () => { } catch (error) { logger.error(error) debug(MESSAGES.SYNC_CORE.CHECK_ERROR, error); + // Invariant: always release the sync gate on ANY failure (network or not), + // so the next notify/poke can restart syncing. Several failure paths (e.g. a + // non-network content-type schema error, filterItems/plugin errors, or a + // failed checkpoint save) otherwise left SQ=true and wedged the loop shut. + flag.SQ = false check().then(() => { debug(MESSAGES.SYNC_CORE.CHECK_RECOVERED); }).catch((error) => { @@ -251,11 +256,18 @@ export const unlock = (refire?: boolean) => { debug(MESSAGES.SYNC_CORE.SYNC_UNLOCKED, refire) flag.lockdown = false if (typeof refire === 'boolean' && refire) { + // Fully re-arm the sync gate. Clearing lockdown alone is not enough: the + // failed sync left SQ=true / WQ=false, so check()'s (!SQ && WQ) gate would + // stay shut and sync would never resume. + flag.SQ = false flag.WQ = true - if (flag.requestCache && Object.keys(flag.requestCache)) { - return fire(flag.requestCache.params) - .then(flag.requestCache.resolve) - .catch(flag.requestCache.reject) + if (flag.requestCache && Object.keys(flag.requestCache).length) { + const cached = flag.requestCache + // Clear before replaying so a later unlock() cannot re-fire a stale request + flag.requestCache = undefined + return fire(cached.params) + .then(cached.resolve) + .catch(cached.reject) } } return check() @@ -439,4 +451,11 @@ const postProcess = (req, resp) => { }) } -emitter.on('check', check) +// check() rejects on a failed sync; when invoked as an event callback that +// rejection would be unhandled and trip the process-level lockdown. Catch it +// here — check()'s own catch already released the gate, so the next poke resumes. +emitter.on('check', () => { + check().catch((error) => { + debug(MESSAGES.SYNC_CORE.CHECK_FAILED, error) + }) +}) diff --git a/src/core/inet.ts b/src/core/inet.ts index 4573491..ec661a0 100644 --- a/src/core/inet.ts +++ b/src/core/inet.ts @@ -69,7 +69,11 @@ export const checkNetConnectivity = () => { emitter.emit('disconnected', currentTimeout += sm.inet.retryIncrement) }) } else if (disconnected) { - poke() + // poke() rejects if the resumed sync fails; without a catch that rejection + // would be unhandled and trip the process-level lockdown. + poke().catch((error) => { + debug('poke after reconnect failed:', error) + }) } disconnected = false diff --git a/src/core/process.ts b/src/core/process.ts index fe17813..98f7459 100644 --- a/src/core/process.ts +++ b/src/core/process.ts @@ -38,7 +38,10 @@ const unhandledErrors = (error) => { logger.error(error) lock() setTimeout(() => { - unlock() + // Pass refire=true so recovery re-arms the sync gate (resets SQ, sets WQ) + // and replays/continues syncing. unlock() with no arg only cleared the + // lockdown flag and left the gate closed, permanently stopping sync. + unlock(true) }, 10000) } From 6c21a0d48d6c931262648fda530cbab29741d700 Mon Sep 17 00:00:00 2001 From: harshitha-cstk Date: Mon, 13 Jul 2026 18:36:48 +0530 Subject: [PATCH 2/4] fix: strip stale query string before rebuilding retry URL --- src/api.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/api.ts b/src/api.ts index 3c780b0..14a2397 100644 --- a/src/api.ts +++ b/src/api.ts @@ -122,7 +122,11 @@ export const get = (req, RETRY = 1) => { return rejectOnce(new Error('Max retry limit exceeded!')) } req.method = Contentstack.verbs.get - req.path = req.path || Contentstack.apis.sync + // Strip any previously-appended query string. On a retry, req.path already + // holds the fully-built path (base + '?' + qs) from the prior attempt; without + // this, the query string would be appended again, producing a malformed URL + // like '/v3/stacks/sync?...&sync_token=x?...&sync_token=x' that the API rejects. + req.path = (req.path || Contentstack.apis.sync).split('?')[0] if (req.qs) { req.path += `?${stringify(req.qs)}` } From 9060d24eadc5265dde6ab6ade1918299ef1129dd Mon Sep 17 00:00:00 2001 From: harshitha-cstk Date: Mon, 13 Jul 2026 19:05:42 +0530 Subject: [PATCH 3/4] fix: prevent unlock() rejection from re-triggering lockdown unlock() returned check()/fire(), which reject on a sync failure. Callers (process.ts, q.ts) invoke unlock() fire-and-forget, so that rejection could surface as an unhandled rejection and re-trip the process-level lockdown, undoing the recovery this PR adds. Attach a .catch to both the requestCache replay and the check() call so unlock() handles/logs its own errors and is a safe, self-contained gate toggle. --- src/core/index.ts | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/core/index.ts b/src/core/index.ts index 5df60bd..3516197 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -255,6 +255,13 @@ export const lock = () => { export const unlock = (refire?: boolean) => { debug(MESSAGES.SYNC_CORE.SYNC_UNLOCKED, refire) flag.lockdown = false + // Callers (process.ts, q.ts) invoke unlock() fire-and-forget. check()/fire() + // reject on a sync failure and would otherwise surface as an UNHANDLED rejection + // that re-trips the process-level lockdown, undoing the recovery this enables. + // Swallow+log here so unlock() is a safe, self-contained gate toggle for callers. + const swallow = (error) => { + debug(MESSAGES.SYNC_CORE.CHECK_FAILED, error) + } if (typeof refire === 'boolean' && refire) { // Fully re-arm the sync gate. Clearing lockdown alone is not enough: the // failed sync left SQ=true / WQ=false, so check()'s (!SQ && WQ) gate would @@ -268,9 +275,10 @@ export const unlock = (refire?: boolean) => { return fire(cached.params) .then(cached.resolve) .catch(cached.reject) + .catch(swallow) } } - return check() + return check().catch(swallow) } /** From 97968915e8a2d318baf3cac81da8f4e74e902fe8 Mon Sep 17 00:00:00 2001 From: harshitha-cstk Date: Tue, 14 Jul 2026 14:57:38 +0530 Subject: [PATCH 4/4] fix: guard success-path JSON.parse against invalid response body --- src/api.ts | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/api.ts b/src/api.ts index 14a2397..94d1da4 100644 --- a/src/api.ts +++ b/src/api.ts @@ -171,7 +171,15 @@ export const get = (req, RETRY = 1) => { .on('end', () => { debug(MESSAGES.API.STATUS(response.statusCode)) if (response.statusCode >= 200 && response.statusCode <= 399) { - return resolveOnce(JSON.parse(body)) + // JSON.parse can throw on an empty/invalid body despite a 2xx/3xx. + // This runs in the 'end' handler, so an uncaught throw would bypass + // rejectOnce and crash the process / re-trigger the global lockdown. + try { + return resolveOnce(JSON.parse(body)) + } catch (parseError) { + debug('Failed to parse success response body:', parseError) + return rejectOnce(parseError) + } } else if (response.statusCode === 429) { timeDelay = Math.pow(Math.SQRT2, RETRY) * RETRY_DELAY_BASE debug(MESSAGES.API.RATE_LIMIT(options.path, timeDelay))