fix(npmjs): follower crashes on truncated registry responses

src/__tests__/package-sources/npmjs/npm-js-follower.lambda.test.ts

@@ -1,4 +1,4 @@
-import { gunzipSync } from 'zlib';
+import { gunzipSync, gzipSync } from 'zlib';
 import { InvokeCommand, LambdaClient } from '@aws-sdk/client-lambda';
 import {
   GetObjectCommand,
@@ -522,6 +522,116 @@ test('records interim checkpoints so an interrupted scan resumes mid-window', as
   expect(saved.lastDeepScanAt).toBeUndefined();
 });
 
+test('retries truncated gzip responses from the registry', async () => {
+  givenHead(1_010);
+  givenNoStateFile();
+  givenLegacyMarker(1_000);
+
+  nock(REPLICA)
+    .get('/registry/_changes')
+    .query({ limit: '10000', since: '1000' })
+    .reply(200, {
+      results: [
+        {
+          seq: 1_005,
+          id: 'flaky-package',
+          changes: [{ rev: '1-aaaaaa' }],
+          deleted: false,
+        },
+      ],
+      last_seq: 1_005,
+    });
+  // First packument response is a truncated gzip body (Z_BUF_ERROR), the
+  // retry succeeds.
+  const fullBody = gzipSync(
+    Buffer.from(
+      JSON.stringify(packageDoc('flaky-package', '1-aaaaaa', '1.0.0'))
+    )
+  );
+  nock(REGISTRY)
+    .get('/flaky-package')
+    .reply(200, fullBody.subarray(0, fullBody.length - 8), {
+      'content-encoding': 'gzip',
+    });
+  nock(REGISTRY)
+    .get('/flaky-package')
+    .reply(200, packageDoc('flaky-package', '1-aaaaaa', '1.0.0'));
+
+  await handler(event, context);
+
+  expect(stagedPackages()).toEqual([
+    { name: 'flaky-package', version: '1.0.0' },
+  ]);
+  expect(savedState().has(1_005)).toBe(true);
+});
+
+test('does not receipt entries whose metadata fetch failed, so a later scan retries them', async () => {
+  givenHead(1_010);
+  givenNoStateFile();
+  givenLegacyMarker(1_000);
+
+  nock(REPLICA)
+    .get('/registry/_changes')
+    .query({ limit: '10000', since: '1000' })
+    .reply(200, {
+      results: [
+        {
+          seq: 1_004,
+          id: 'broken-package',
+          changes: [{ rev: '1-aaaaaa' }],
+          deleted: false,
+        },
+        {
+          seq: 1_006,
+          id: 'good-package',
+          changes: [{ rev: '1-bbbbbb' }],
+          deleted: false,
+        },
+      ],
+      last_seq: 1_006,
+    });
+  // A client error is used so the request is not retried until the
+  // transient-error deadline.
+  nock(REGISTRY).get('/broken-package').reply(403, 'forbidden');
+  nock(REGISTRY)
+    .get('/good-package')
+    .reply(200, packageDoc('good-package', '1-bbbbbb', '1.0.0'));
+
+  await handler(event, context);
+
+  // The healthy entry was processed and receipted; the failed one was not
+  // receipted, so the next scan will retry it.
+  expect(stagedPackages()).toEqual([
+    { name: 'good-package', version: '1.0.0' },
+  ]);
+  const state = savedState();
+  expect(state.has(1_006)).toBe(true);
+  expect(state.has(1_004)).toBe(false);
+  expect(mockPutMetric).toHaveBeenCalledWith(
+    MetricName.METADATA_FETCH_FAILURES,
+    1,
+    Unit.Count
+  );
+});
+
+test('persists its state even when the scan fails part-way', async () => {
+  givenHead(1_010);
+  givenNoStateFile();
+  givenLegacyMarker(1_000);
+
+  // The feed itself rejects the request: the scan throws. (A client error
+  // is used so the request is not retried until the transient-error deadline.)
+  nock(REPLICA)
+    .get('/registry/_changes')
+    .query({ limit: '10000', since: '1000' })
+    .reply(403, 'forbidden');
+
+  await expect(handler(event, context)).rejects.toThrow(/HTTP 403/);
+
+  // The state (here: the seeded checkpoint) was saved regardless.
+  expect(savedState().newestCheckpointSeq()).toBe(1_000);
+});
+
 test('resets its position when the feed head regresses below the newest checkpoint', async () => {
   const now = Date.now();
   // The feed head (3000) is below our newest checkpoint (5000).

src/package-sources/npmjs.ts

@@ -392,6 +392,12 @@ export class NpmJs implements IPackageSource {
                 }),
                 0
               ),
+              fillMetric(
+                this.metricMetadataFetchFailures({
+                  label: 'Metadata Fetch Failures',
+                }),
+                0
+              ),
             ],
             leftYAxis: { min: 0 },
             right: [
@@ -522,6 +528,21 @@ export class NpmJs implements IPackageSource {
     });
   }
 
+  /**
+   * The number of change entries whose registry metadata could not be fetched
+   * (after transient-error retries). These entries are not receipted, so a
+   * later scan retries them.
+   */
+  public metricMetadataFetchFailures(opts?: MetricOptions): Metric {
+    return new Metric({
+      period: Duration.minutes(5),
+      statistic: Stats.SUM,
+      ...opts,
+      metricName: MetricName.METADATA_FETCH_FAILURES,
+      namespace: METRICS_NAMESPACE,
+    });
+  }
+
   /**
    * The number of packages for which the registry packument is still behind
    * the revision announced by the `_changes` feed. The versions served so far

src/package-sources/npmjs/constants.lambda-shared.ts

@@ -32,6 +32,13 @@ export const enum MetricName {
    */
   LATE_CHANGE_LAG = 'LateChangeLag',
 
+  /**
+   * Number of change entries whose registry metadata could not be fetched
+   * (after transient-error retries). These entries are not receipted, so a
+   * later scan retries them.
+   */
+  METADATA_FETCH_FAILURES = 'MetadataFetchFailures',
+
   /**
    * The number of packages for which the registry packument is still behind
    * the revision announced by the `_changes` feed ("laggy packuments"). The

src/package-sources/npmjs/couch-changes.lambda-shared.ts

@@ -120,13 +120,31 @@ export class CouchChanges extends EventEmitter {
    */
   public async attachAllMetadata(
     changes: readonly DatabaseChange[]
-  ): Promise<AttachedChange[]> {
-    const outcomes = await Promise.all(
-      changes.map((change) => this.attachMetadata(change))
-    );
-    return outcomes.filter(
-      (outcome): outcome is AttachedChange => outcome !== undefined
+  ): Promise<AttachedMetadata> {
+    const ok = new Array<AttachedChange>();
+    const failedSeqs = new Array<number>();
+    await Promise.all(
+      changes.map(async (change) => {
+        try {
+          const outcome = await this.attachMetadata(change);
+          if (outcome !== undefined) {
+            ok.push(outcome);
+          }
+        } catch (error) {
+          // One unreachable packument must not fail the whole batch. The
+          // entry is reported as failed so the caller does not record a
+          // receipt for it; a later scan retries it.
+          console.error(
+            `Failed to fetch metadata for ${change.id}, a later scan will retry it: ${error}`
+          );
+          const seq = Number(change.seq);
+          if (!isNaN(seq)) {
+            failedSeqs.push(seq);
+          }
+        }
+      })
     );
+    return { ok, failedSeqs };
   }
 
   private async attachMetadata(
@@ -301,7 +319,13 @@ function readResponseJson(
 class RetryableError extends Error {}
 
 function isRetryableError(e: Error): boolean {
-  return e instanceof RetryableError || (e as any).code === 'ECONNRESET';
+  return (
+    e instanceof RetryableError ||
+    (e as any).code === 'ECONNRESET' ||
+    // A truncated response body (the gunzip stream ends unexpectedly). Seen
+    // repeatedly from registry.npmjs.org; transient like a connection reset.
+    (e as any).code === 'Z_BUF_ERROR'
+  );
 }
 
 async function sleep(ms: number) {
@@ -352,6 +376,22 @@ export interface DatabaseChanges {
   readonly totalCount: number;
 }
 
+/**
+ * The result of attaching registry metadata to a batch of change entries.
+ */
+export interface AttachedMetadata {
+  /**
+   * The change entries whose metadata could be fetched.
+   */
+  readonly ok: AttachedChange[];
+
+  /**
+   * The sequence numbers of change entries whose metadata fetch failed. No
+   * receipt must be recorded for these, so that a later scan retries them.
+   */
+  readonly failedSeqs: number[];
+}
+
 /**
  * A change entry with its registry metadata attached.
  */