fix: npm follower occasionally drops incoming updates (#1831)

The npm follower grabs information from multiple sources that can be out
of sync. This subsequently causes some new package updates to not be
identified and sent to ingestion.

<img width="433" height="612" alt="Screenshot 2026-01-07 at 6 03 43 PM"
src="https://github.com/user-attachments/assets/99c4a8d5-cd90-43d0-9d7a-f8cf1e56683f"
/>

Basically, with the new NPM API, our NPM follower ingests information
from multiple sources. Occasionally there is a delay between picking up
a new change from the NPM changes stream and the change actually showing
up in the specific package metadata.

Previously, we would just drop this change altogether as unactionable.
Subsequent updates to the package could trigger another metadata lookup
that would grab the change, but this results in an indeterminate delay.

Now, we identify that the metadata is out of sync by comparing the
sequential part of the `rev` property. If the metadata `rev` is below
the changes `rev`, we know it is lagging behind and exponentially retry
until the metadata is updated or until 30 seconds (where we then proceed
with the out of sync data).

The NPM follower function does some basic math for when to grab the next
batch and when to exit and wait for a subsequent lambda execution.
Because this change increases the batch time by up to 30 seconds, I have
increased the expected worst-case-scenario time by 30 seconds as well.

----

*By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache-2.0 license*

---------

Co-authored-by: Rico Huijbers <[email protected]>

Author: kaizencc

src/__tests__/__snapshots__/construct-hub.test.ts.snap

@@ -16,6 +16,8 @@ const DEFAULT_BATCH_SIZE = 100;
 
 const MAX_CONNS_PER_HOST = 100;
 
+const MAX_PACKAGE_SERVER_LAG_MS = 30_000; // 30 seconds
+
 /**
  * A utility class that helps with traversing CouchDB database changes streams
  * in a promise-based, page-by-page manner.
@@ -82,29 +84,60 @@ export class CouchChanges extends EventEmitter {
     };
   }
 
+  /**
+   * Fetch the metadata associated with a change. The change comes associated with a revision number,
+   * which can be compared to the revision number of the metadata to determine if the replica is
+   * lagging behind the changes stream. If so, we retry until the replica is up-to-date or until
+   * 30 seconds elapsed after which we return the potentially stale metadata.
+   */
   private async fetchAndFilterMetadata(change: DatabaseChange) {
     // Filter out deleted packages or null ids
     if (change.deleted || !change.id) {
       console.log(`Skipping ${change.id}: deleted or null id`);
       return;
     }
 
+    const latestChangesRev = getMaxSequentialRevision(change);
     const metadataUrl = new URL(change.id, NPM_REGISTRY_URL);
     console.log(`Fetching metadata for ${change.id}: ${metadataUrl}`);
 
-    try {
-      const meta = await this.https('get', metadataUrl);
-      change.doc = meta; // add metadata to the change object
-      return change;
-    } catch (e: any) {
-      if (e.message?.includes('HTTP 404')) {
-        console.log(
-          `Skipping ${change.id} because of HTTP 404 (Not Found) error`
-        );
-        return;
+    // Retry configuration
+    const baseDelay = 1_000; // 1 second
+    const maxDelay = 8_000; // 8 seconds max
+    let attempt = 0;
+    const startTime = Date.now();
+
+    while(Date.now() - startTime < MAX_PACKAGE_SERVER_LAG_MS) {
+      try {
+        const meta = await this.https('get', metadataUrl);
+        const latestReplicaRev = parseSequentialRevision(meta._rev as string);
+
+        change.doc = meta; // add metadata to the change object
+
+        // Happy path: replica is up-to-date
+        if (latestReplicaRev >= latestChangesRev) {
+          return change;
+        }
+
+        // Unhappy path: replica is behind. Calculate delay and retry
+        const delay = Math.floor(Math.random() * Math.min(baseDelay * Math.pow(2, attempt), maxDelay));
+        console.log(`${change.id}: package _rev ${latestReplicaRev} < expected replication rev ${latestChangesRev}, retrying in ${delay} ms`);
+        await new Promise(resolve => setTimeout(resolve, delay));
+        attempt++;
+      } catch (e: any) {
+        if (e.message?.includes('HTTP 404')) {
+          console.log(
+            `Skipping ${change.id} because of HTTP 404 (Not Found) error`
+          );
+          return;
+        }
+        throw e;
       }
-      throw e;
     }
+
+    // Timeout reached, proceed with stale data
+    console.log(`Timeout reached for ${change.id}, replica may be stale`);
+    return change;
   }
 
   private async fetchAndFilterAllMetadata(
@@ -245,6 +278,17 @@ async function sleep(ms: number) {
   return new Promise((ok) => setTimeout(ok, ms));
 }
 
+function parseSequentialRevision(rev: string): number {
+  return parseInt(rev.split('-')[0]);
+}
+
+function getMaxSequentialRevision(change: DatabaseChange): number {
+  return Math.max(...change.changes
+    .map(change => parseSequentialRevision(change.rev))
+    .filter(num => !isNaN(num))
+  );
+}
+
 export interface DatabaseChanges {
   /**
    * The last sequence ID from this change set. This is the value that should be

src/__tests__/devapp/__snapshots__/snapshot.test.ts.snap

@@ -97,9 +97,9 @@ export async function handler(event: ScheduledEvent, context: Context) {
   // The last written marker seq id.
   let updatedMarker = initialMarker;
 
-  // The slowest batch processing time so far (starts at 30 seconds). This is how much time should
+  // The slowest batch processing time so far (starts at 60 seconds). This is how much time should
   // be left before timeout if a new batch is to be fetched.
-  let maxBatchProcessingTime = 30_000;
+  let maxBatchProcessingTime = 60_000;
   // Whether we should continue reading more items or not... This is set to false when the current
   // latest change is reached (i.e: next page of changes is empty).
   let shouldContinue = true;