fix: recover sync loop after transient Sync API timeout (DX-9488)

src/api.ts

@@ -100,11 +100,33 @@ 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
+    // 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)}`
     }
@@ -149,15 +171,23 @@ 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))
+                // 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))
 
                 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 +197,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 +233,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,25 +246,50 @@ export const get = (req, RETRY = 1) => {
                 }
 
                 debug(MESSAGES.API.REQUEST_FAILED(options))
-                return reject(body)
+                return rejectOnce(body)
               }
             })
         })
 
       // 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 +314,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)
     }
   })
 }

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) => {
@@ -250,15 +255,30 @@ 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
+    // 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)
+        .catch(swallow)
     }
   }
-  return check()
+  return check().catch(swallow)
 }
 
 /**
@@ -439,4 +459,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)
+  })
+})

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
 

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