From fa8d8b32090e51ea32ab11b297ab2ae46044ce12 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Tue, 21 Apr 2026 18:16:07 -0400
Subject: [PATCH 01/29] tests: integrate sample system tests that are still
 useful

---
 handwritten/pubsub/system-test/common.test.ts |  36 ++++
 handwritten/pubsub/system-test/common.ts      |  26 +++
 .../pubsub/system-test/sample-tests.ts        | 102 ++++++++++
 .../pubsub/system-test/testResources.test.ts  |  84 ++++++++
 .../pubsub/system-test/testResources.ts       | 182 ++++++++++++++++++
 5 files changed, 430 insertions(+)
 create mode 100644 handwritten/pubsub/system-test/common.test.ts
 create mode 100644 handwritten/pubsub/system-test/common.ts
 create mode 100644 handwritten/pubsub/system-test/sample-tests.ts
 create mode 100644 handwritten/pubsub/system-test/testResources.test.ts
 create mode 100644 handwritten/pubsub/system-test/testResources.ts

diff --git a/handwritten/pubsub/system-test/common.test.ts b/handwritten/pubsub/system-test/common.test.ts
new file mode 100644
index 000000000000..77ca6642066b
--- /dev/null
+++ b/handwritten/pubsub/system-test/common.test.ts
@@ -0,0 +1,36 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import {assert} from 'chai';
+import {describe, it} from 'mocha';
+import {commandFor} from './common';
+import * as path from 'path';
+
+describe('common (unit)', () => {
+  it('commandFor finds TS samples', () => {
+    const result = commandFor('createAvroSchema');
+    assert.strictEqual(
+      result,
+      `node ${path.join('build', 'createAvroSchema.js')}`
+    );
+  });
+
+  it('commandFor finds JS samples', () => {
+    const result = commandFor('createSubscription');
+    assert.strictEqual(
+      result,
+      `node ${path.join('build', 'createSubscription.js')}`
+    );
+  });
+});
diff --git a/handwritten/pubsub/system-test/common.ts b/handwritten/pubsub/system-test/common.ts
new file mode 100644
index 000000000000..5529231d30ae
--- /dev/null
+++ b/handwritten/pubsub/system-test/common.ts
@@ -0,0 +1,26 @@
+// Copyright 2021 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import * as cp from 'child_process';
+import * as path from 'path';
+
+export const execSync = (cmd: string): string =>
+  cp.execSync(cmd, {encoding: 'utf-8'});
+
+// Processed versions of TS samples go to the same build location
+// as the rest of the JS samples.
+export function commandFor(action: string): string {
+  const jsPath = path.join('build', `${action}.js`);
+  return `node ${jsPath}`;
+}
diff --git a/handwritten/pubsub/system-test/sample-tests.ts b/handwritten/pubsub/system-test/sample-tests.ts
new file mode 100644
index 000000000000..865b129fff0e
--- /dev/null
+++ b/handwritten/pubsub/system-test/sample-tests.ts
@@ -0,0 +1,102 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import {Message, PubSub} from '../src';
+import * as assert from 'assert';
+import {describe, it, after} from 'mocha';
+import {TestResources} from './testResources';
+
+describe('Combined Samples Tests', () => {
+  const pubsub = new PubSub();
+  const resources = new TestResources('pubsub_combined');
+
+  function topicName(testId: string): string {
+    return resources.generateName(testId);
+  }
+
+  function subName(testId: string): string {
+    return resources.generateName(testId);
+  }
+
+  async function cleanSubs() {
+    const [subscriptions] = await pubsub.getSubscriptions();
+    await Promise.all(
+      resources.filterForCleanup(subscriptions).map((x: any) => x.delete?.())
+    );
+  }
+
+  async function cleanTopics() {
+    const [topics] = await pubsub.getTopics();
+    await Promise.all(
+      resources.filterForCleanup(topics).map((x: any) => x.delete?.())
+    );
+  }
+
+  after(async () => {
+    await cleanSubs();
+    await cleanTopics();
+  });
+
+  it('should create a topic', async () => {
+    const name = topicName('create');
+    
+    // --- From sample (createTopic.js) ---
+    await pubsub.createTopic(name);
+    console.log(`Topic ${name} created.`);
+    
+    // --- From test (topics.test.ts) ---
+    const [topics] = await pubsub.getTopics();
+    const exists = topics.some((t: any) => t.name.endsWith(name));
+    assert.ok(exists, 'Topic was created');
+  });
+
+  it('should publish a message', async () => {
+    const tname = topicName('publish');
+    const sname = subName('publish');
+    
+    const [topic] = await pubsub.topic(tname).get({autoCreate: true});
+    const [subscription] = await topic.subscription(sname).get({autoCreate: true});
+    
+    // --- From sample (publishMessage.js) ---
+    const data = 'Hello, world!';
+    const dataBuffer = Buffer.from(data);
+    const messageId = await topic.publishMessage({data: dataBuffer});
+    console.log(`Message ${messageId} published.`);
+    
+    // --- From test (topics.test.ts) ---
+    const message = await new Promise<Message>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout')), 10000);
+      subscription.once('message', (m: Message) => {
+        clearTimeout(timeout);
+        m.ack();
+        resolve(m);
+      });
+    });
+    
+    assert.strictEqual(message.data.toString(), data);
+  });
+
+  it('should delete a topic', async () => {
+    const name = topicName('delete');
+    await pubsub.topic(name).get({autoCreate: true});
+    
+    // --- From sample (deleteTopic.js) ---
+    await pubsub.topic(name).delete();
+    console.log(`Topic ${name} deleted.`);
+    
+    // --- From test (topics.test.ts) ---
+    const [exists] = await pubsub.topic(name).exists();
+    assert.strictEqual(exists, false);
+  });
+});
diff --git a/handwritten/pubsub/system-test/testResources.test.ts b/handwritten/pubsub/system-test/testResources.test.ts
new file mode 100644
index 000000000000..0ddd84655bd0
--- /dev/null
+++ b/handwritten/pubsub/system-test/testResources.test.ts
@@ -0,0 +1,84 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import {assert} from 'chai';
+import {describe, it, beforeEach} from 'mocha';
+import {TestResources} from './testResources';
+
+describe('testResources (unit)', () => {
+  const fixedId = 'fixed';
+  const fixedTime = Date.now();
+  const fakeTokenMaker = {
+    uuid: () => fixedId,
+    timestamp: () => fixedTime,
+  };
+
+  const suiteId = 'someSuite';
+  let testResources!: TestResources;
+
+  beforeEach(() => {
+    testResources = new TestResources(suiteId, fakeTokenMaker);
+  });
+
+  it('has predictable prefixes', () => {
+    const prefix = testResources.getPrefix('testId');
+    assert.strictEqual(prefix, `${suiteId}-${fixedTime}-testId`);
+
+    const normalizedPrefix = testResources.getPrefix('test-id-dashes');
+    assert.strictEqual(
+      normalizedPrefix,
+      `${suiteId}-${fixedTime}-test_id_dashes`
+    );
+
+    const suitePrefix = testResources.getPrefix();
+    assert.strictEqual(suitePrefix, `${suiteId}-${fixedTime}`);
+  });
+
+  it('generates names', () => {
+    const prefix = testResources.getPrefix('testId');
+    const name = testResources.generateName('testId');
+    assert.strictEqual(name, `${prefix}-${fixedId}`);
+  });
+
+  it('filters for cleanup', () => {
+    const resources = [
+      {
+        // Not related
+        name: 'ooga',
+      },
+      {
+        // For current test run
+        name: `${suiteId}-${fixedTime}-bob-98719284791`,
+      },
+      {
+        // For previous test run, but not very old
+        name: `${suiteId}-${fixedTime - 100}-bob-124897912`,
+      },
+      {
+        // For previous test run, but old
+        name: `${suiteId}-${fixedTime - 3000 * 60 * 60}-bob-57823975`,
+      },
+    ];
+    const filtered = testResources.filterForCleanup(resources);
+    assert.strictEqual(filtered.length, 2);
+    assert.strictEqual(
+      1,
+      filtered.filter(r => r.name?.includes('bob-9871')).length
+    );
+    assert.strictEqual(
+      1,
+      filtered.filter(r => r.name?.includes('bob-5782')).length
+    );
+  });
+});
diff --git a/handwritten/pubsub/system-test/testResources.ts b/handwritten/pubsub/system-test/testResources.ts
new file mode 100644
index 000000000000..3f5665096201
--- /dev/null
+++ b/handwritten/pubsub/system-test/testResources.ts
@@ -0,0 +1,182 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// I don't like that these two files (this plus ".test") are duplicated
+// across the two test structures, but because of the tangle of rootDirs
+// and package.json "files", it's hard to avoid it.
+
+import * as uuid from 'uuid';
+
+// Returns a shortened UUID that can be used to identify a
+// specific run of a specific test.
+function shortUUID() {
+  return uuid.v4().split('-').shift()!;
+}
+
+export interface TokenMaker {
+  uuid(): string;
+  timestamp(): number;
+}
+
+export const defaultMaker = {
+  uuid: shortUUID,
+  timestamp: () => Date.now(),
+};
+
+export interface Resource {
+  name?: string | null | undefined;
+  delete?(): Promise<unknown>;
+}
+
+function normalizeId(id: string): string {
+  return id.replace(/-/g, '_');
+}
+
+/**
+ * Manages the names of testing resources during a test run. It's
+ * easily to accidentally leak resources, and it's easy to accidentally
+ * have conflicts with tests running concurrently, so this class helps
+ * you manage them.
+ *
+ * Used nomenclature:
+ *  Test - a single test for a single aspect of code; for example,
+ *    "create a topic in pub/sub"
+ *  Test Suite - a collection of tests that are generally run together;
+ *     for example, "test topic operations in pub/sub"
+ *  Test Run - a single run of a test suite (or single test within a suite);
+ *     for example, "run the tests for PR #1234, 5th attempt"
+ */
+export class TestResources {
+  testSuiteId: string;
+  currentTime: string;
+  tokenMaker: TokenMaker;
+
+  /**
+   * @param testSuiteId [string] A unique ID for a test suite (e.g.
+   *   pubsub-topics).
+   */
+  constructor(testSuiteId: string, tokenMaker: TokenMaker = defaultMaker) {
+    this.testSuiteId = normalizeId(testSuiteId);
+    this.currentTime = `${tokenMaker.timestamp()}`;
+    this.tokenMaker = tokenMaker;
+  }
+
+  /**
+   * Returns the resource prefix for the current run of the test suite.
+   * Optionally, testId may specify the specific ID of a test in the
+   * suite.
+   */
+  getPrefix(testId?: string): string {
+    if (testId) {
+      return [this.testSuiteId, this.currentTime, normalizeId(testId)].join(
+        '-',
+      );
+    } else {
+      return [this.testSuiteId, this.currentTime].join('-');
+    }
+  }
+
+  /**
+   * Generates a unique resource name for one run of a test within
+   * a test suite.
+   */
+  generateName(testId: string): string {
+    return [this.getPrefix(testId), this.tokenMaker.uuid()].join('-');
+  }
+
+  /**
+   * Generates a unique resource name for one run of a test within
+   * a test suite for BigQuery resources.
+   */
+  generateBigQueryName(testId: string): string {
+    return [normalizeId(this.getPrefix(testId)), this.tokenMaker.uuid()].join(
+      '_',
+    );
+  }
+
+  /**
+   * Generates a unique resource name for one run of a test within
+   * a test suite for Cloud Storage resources.
+   */
+  generateStorageName(testId: string): string {
+    return [normalizeId(this.getPrefix(testId)), this.tokenMaker.uuid()].join(
+      '_',
+    );
+  }
+
+  /**
+   * Given a list of resource names (and a test ID), this will return
+   * a list of all resources that should be deleted to clean up for
+   * the current run of that particular test.
+   */
+  filterForTest(testId: string, allResources: Resource[]): Resource[] {
+    const prefix = this.getPrefix(testId);
+    return allResources.filter(n => n.name?.includes(prefix));
+  }
+
+  /**
+   * Given a list of resource names, this will return a list of all
+   * resources that should be deleted to clean up after the current
+   * run of a test suite.
+   */
+  filterForCurrentRun(allResources: Resource[]): Resource[] {
+    const prefix = this.getPrefix();
+    return allResources.filter(n => n.name?.includes(prefix));
+  }
+
+  /**
+   * Given a list of resource names, this will return a list of all
+   * resources that should be deleted to clean up after any run
+   * of the current test suite. Note that some of the names may
+   * still be in use.
+   */
+  filterForSuite(allResources: Resource[]): Resource[] {
+    return allResources.filter(n => n.name?.includes(this.testSuiteId));
+  }
+
+  /**
+   * Given a list of resource names, this will return a list of all
+   * resources that should be deleted to generally clean up after any
+   * run of the current test suite. This is much like filterForSuite(),
+   * but it also filters by age - items that are less than 2 hours
+   * old will not be cleaned.
+   */
+  filterForCleanup(allResources: Resource[]): Resource[] {
+    const currentRunPrefix = this.getPrefix();
+    return allResources.filter(n => {
+      let name = n.name || undefined;
+      if (name === undefined) {
+        return false;
+      }
+
+      // We'll always get at least one thing.
+      name = name.split('/').pop()!;
+
+      if (name.startsWith(currentRunPrefix)) {
+        return true;
+      }
+
+      if (name.startsWith(this.testSuiteId)) {
+        const parts = name.split('-');
+        const createdAt = Number(parts[1]);
+        const timeDiff = (Date.now() - createdAt) / (1000 * 60 * 60);
+        if (timeDiff >= 2) {
+          return true;
+        }
+      }
+
+      return false;
+    });
+  }
+}

From 5a292f3bae7819daa8936b5197e64f172dfe6e27 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Tue, 21 Apr 2026 18:24:30 -0400
Subject: [PATCH 02/29] chore: remove an unnecessary `any`

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 handwritten/pubsub/system-test/sample-tests.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/handwritten/pubsub/system-test/sample-tests.ts b/handwritten/pubsub/system-test/sample-tests.ts
index 865b129fff0e..a1679b4ebefb 100644
--- a/handwritten/pubsub/system-test/sample-tests.ts
+++ b/handwritten/pubsub/system-test/sample-tests.ts
@@ -32,7 +32,7 @@ describe('Combined Samples Tests', () => {
   async function cleanSubs() {
     const [subscriptions] = await pubsub.getSubscriptions();
     await Promise.all(
-      resources.filterForCleanup(subscriptions).map((x: any) => x.delete?.())
+      resources.filterForCleanup(subscriptions).map(x => x.delete?.())
     );
   }
 

From dd90f8ecf662958ca1dc14124f32663b7cfbfcab Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 23 Apr 2026 17:41:58 -0400
Subject: [PATCH 03/29] tests: integrate more sample tests

---
 handwritten/pubsub/package.json               |   6 +
 handwritten/pubsub/system-test/avro-js.d.ts   |  10 ++
 .../pubsub/system-test/avro-samples.test.ts   | 128 ++++++++++++++++++
 .../system-test/batch-flow-samples.test.ts    | 111 +++++++++++++++
 .../system-test/fixtures/provinces.proto      |   8 ++
 .../pubsub/system-test/otel-samples.test.ts   |  66 +++++++++
 .../system-test/protobuf-samples.test.ts      |  86 ++++++++++++
 7 files changed, 415 insertions(+)
 create mode 100644 handwritten/pubsub/system-test/avro-js.d.ts
 create mode 100644 handwritten/pubsub/system-test/avro-samples.test.ts
 create mode 100644 handwritten/pubsub/system-test/batch-flow-samples.test.ts
 create mode 100644 handwritten/pubsub/system-test/fixtures/provinces.proto
 create mode 100644 handwritten/pubsub/system-test/otel-samples.test.ts
 create mode 100644 handwritten/pubsub/system-test/protobuf-samples.test.ts

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index 875998dd31a6..d44e3933eaf7 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -69,8 +69,11 @@
     "p-defer": "^3.0.0"
   },
   "devDependencies": {
+    "@google-cloud/opentelemetry-cloud-trace-exporter": "^2.4.1",
     "@grpc/proto-loader": "^0.8.0",
     "@opentelemetry/sdk-trace-base": "^1.17.0",
+    "@opentelemetry/sdk-trace-node": "^2.7.0",
+    "@types/chai": "^5.2.3",
     "@types/duplexify": "^3.6.4",
     "@types/extend": "^3.0.4",
     "@types/lodash.snakecase": "^4.1.9",
@@ -81,7 +84,10 @@
     "@types/proxyquire": "^1.3.31",
     "@types/sinon": "^21.0.0",
     "@types/tmp": "^0.2.6",
+    "@types/uuid": "^10.0.0",
+    "avro-js": "^1.12.1",
     "c8": "^10.1.3",
+    "chai": "^6.2.2",
     "codecov": "^3.8.3",
     "execa": "~5.1.0",
     "gapic-tools": "^1.0.1",
diff --git a/handwritten/pubsub/system-test/avro-js.d.ts b/handwritten/pubsub/system-test/avro-js.d.ts
new file mode 100644
index 000000000000..4a8496b76e69
--- /dev/null
+++ b/handwritten/pubsub/system-test/avro-js.d.ts
@@ -0,0 +1,10 @@
+declare module 'avro-js' {
+  function parse(def: string): Parser;
+
+  class Parser {
+    fromBuffer<T>(buf: Buffer): T;
+    fromString<T>(str: string): T;
+    toBuffer<T>(item: T): Buffer;
+    toString<T>(item: T): string;
+  }
+}
diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
new file mode 100644
index 000000000000..246dccdf1f57
--- /dev/null
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -0,0 +1,128 @@
+import {Message, PubSub, Schema} from '../src';
+import * as assert from 'assert';
+import {describe, it, after, before} from 'mocha';
+import {TestResources} from './testResources';
+import * as avro from 'avro-js';
+import * as fs from 'fs';
+
+describe('Avro Samples System Tests', () => {
+  const pubsub = new PubSub();
+  const resources = new TestResources('ps-sys-avro');
+
+  let topicName: string;
+  let subName: string;
+  let schemaId: string;
+
+  before(async () => {
+    topicName = resources.generateName('topic');
+    subName = resources.generateName('sub');
+    schemaId = resources.generateName('schema');
+
+    const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString();
+    await pubsub.createSchema(schemaId, 'AVRO', definition);
+    await pubsub.createTopic({
+      name: topicName,
+      schemaSettings: {
+        schema: await pubsub.schema(schemaId).getName(),
+        encoding: 'BINARY',
+      },
+    });
+
+    const [topic] = await pubsub.topic(topicName).get();
+    await topic.createSubscription(subName);
+  });
+
+  after(async () => {
+    const [subscriptions] = await pubsub.getSubscriptions();
+    await Promise.all(
+      resources.filterForCleanup(subscriptions).map(x => x.delete?.())
+    );
+
+    const [topics] = await pubsub.getTopics();
+    await Promise.all(
+      resources.filterForCleanup(topics).map((x: any) => x.delete?.())
+    );
+
+    const schemas: any[] = [];
+    for await (const s of pubsub.listSchemas()) {
+      schemas.push(pubsub.schema(s.name!));
+    }
+    await Promise.all(
+      resources.filterForCleanup(schemas).map(x => x.delete?.())
+    );
+  });
+
+  it('should publish and listen for avro records', async () => {
+    const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString();
+    const [topic] = await pubsub.topic(topicName).get();
+    const [subscription] = await pubsub.subscription(subName).get();
+
+    const type = avro.parse(definition);
+
+    const province = {
+      name: 'Ontario',
+      post_abbr: 'ON',
+    };
+
+    const dataBuffer = type.toBuffer(province);
+    const messageId = await topic.publish(dataBuffer);
+    assert.ok(messageId);
+
+    const message = await new Promise<Message>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout waiting for Avro record')), 15000);
+      subscription.once('message', (m: Message) => {
+        clearTimeout(timeout);
+        m.ack();
+        resolve(m);
+      });
+    });
+
+    const schemaMetadata = Schema.metadataFromMessage(message.attributes);
+    assert.strictEqual(schemaMetadata.encoding, 'BINARY');
+
+    const result = type.fromBuffer(message.data) as any;
+    assert.strictEqual(result.name, 'Ontario');
+    assert.strictEqual(result.post_abbr, 'ON');
+  });
+
+  it('should listen for avro records with revisions', async () => {
+    const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString();
+
+    const schemaClient = await pubsub.getSchemaClient();
+
+    const [topic] = await pubsub.topic(topicName).get();
+    const [subscription] = await pubsub.subscription(subName).get();
+
+    const type = avro.parse(definition);
+    const province = {
+      name: 'Ontario',
+      post_abbr: 'ON',
+    };
+
+    const dataBuffer = type.toBuffer(province);
+    await topic.publish(dataBuffer);
+
+    const message = await new Promise<Message>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout waiting for Avro revision')), 15000);
+      subscription.once('message', (m: Message) => {
+        clearTimeout(timeout);
+        m.ack();
+        resolve(m);
+      });
+    });
+
+    const schemaMetadata = Schema.metadataFromMessage(message.attributes);
+    const revision = schemaMetadata.revision!;
+    assert.ok(revision);
+
+    const [fetchedSchema] = await schemaClient.getSchema({
+      name: `${schemaMetadata.name}@${schemaMetadata.revision}`,
+    });
+
+    const reader = avro.parse(fetchedSchema.definition!);
+    const result = reader.fromBuffer(message.data) as any;
+
+    assert.strictEqual(result.name, 'Ontario');
+    assert.strictEqual(result.post_abbr, 'ON');
+  });
+});
diff --git a/handwritten/pubsub/system-test/batch-flow-samples.test.ts b/handwritten/pubsub/system-test/batch-flow-samples.test.ts
new file mode 100644
index 000000000000..f00dd80c9462
--- /dev/null
+++ b/handwritten/pubsub/system-test/batch-flow-samples.test.ts
@@ -0,0 +1,111 @@
+import {Message, PubSub, PublishOptions} from '../src';
+import * as assert from 'assert';
+import {describe, it, after, before} from 'mocha';
+import {TestResources} from './testResources';
+
+describe('Batch and Flow Control Samples System Tests', () => {
+  const pubsub = new PubSub();
+  const resources = new TestResources('ps-sys-batch');
+
+  let topicName: string;
+  let subName: string;
+
+  before(async () => {
+    topicName = resources.generateName('topic');
+    subName = resources.generateName('sub');
+  });
+
+  after(async () => {
+    const [subscriptions] = await pubsub.getSubscriptions();
+    await Promise.all(
+      resources.filterForCleanup(subscriptions).map(x => x.delete?.())
+    );
+
+    const [topics] = await pubsub.getTopics();
+    await Promise.all(
+      resources.filterForCleanup(topics).map((x: any) => x.delete?.())
+    );
+  });
+
+  it('should publish batched messages', async () => {
+    const [topic] = await pubsub.createTopic(topicName);
+    const [subscription] = await topic.createSubscription(subName);
+
+    const publishOptions: PublishOptions = {
+      batching: {
+        maxMessages: 10,
+        maxMilliseconds: 2000,
+      },
+    };
+    const batchPublisher = pubsub.topic(topicName, publishOptions);
+
+    const promises: Promise<string>[] = [];
+    for (let i = 0; i < 10; i++) {
+      promises.push(batchPublisher.publishMessage({data: Buffer.from(`message ${i}`)}));
+    }
+
+    const messageIds = await Promise.all(promises);
+    assert.strictEqual(messageIds.length, 10);
+
+    const messages: Message[] = [];
+    await new Promise<void>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout waiting for batched messages')), 15000);
+      subscription.on('message', (m: Message) => {
+        m.ack();
+        messages.push(m);
+        if (messages.length === 10) {
+          clearTimeout(timeout);
+          subscription.removeAllListeners('message');
+          resolve();
+        }
+      });
+    });
+
+    assert.strictEqual(messages.length, 10);
+  });
+
+  it('should publish with flow control', async () => {
+    const flowTopicName = resources.generateName('flow');
+    const flowSubName = resources.generateName('flowsub');
+
+    const [topic] = await pubsub.createTopic(flowTopicName);
+    const [subscription] = await topic.createSubscription(flowSubName);
+
+    const options = {
+      flowControlOptions: {
+        maxOutstandingMessages: 5,
+        maxOutstandingBytes: 1024,
+      },
+    };
+
+    const topicWithFlow = pubsub.topic(flowTopicName, options);
+    const flow = topicWithFlow.flowControlled();
+
+    const promises = [];
+    for (let i = 0; i < 10; i++) {
+      const wait = flow.publish({data: Buffer.from('flow control message')});
+      if (wait) {
+        await wait;
+      }
+    }
+
+    const messageIds = await flow.all();
+    assert.strictEqual(messageIds.length, 10);
+
+    const messages: Message[] = [];
+    await new Promise<void>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout waiting for flow control messages')), 15000);
+      subscription.on('message', (m: Message) => {
+        m.ack();
+        messages.push(m);
+        if (messages.length === 10) {
+          clearTimeout(timeout);
+          subscription.removeAllListeners('message');
+          resolve();
+        }
+      });
+    });
+
+    assert.strictEqual(messages.length, 10);
+  });
+});
diff --git a/handwritten/pubsub/system-test/fixtures/provinces.proto b/handwritten/pubsub/system-test/fixtures/provinces.proto
new file mode 100644
index 000000000000..08f05488efc0
--- /dev/null
+++ b/handwritten/pubsub/system-test/fixtures/provinces.proto
@@ -0,0 +1,8 @@
+syntax = "proto3";
+
+package utilities;
+
+message Province {
+  string name = 1;
+  string post_abbr = 2;
+}
diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
new file mode 100644
index 000000000000..2ac84b303b82
--- /dev/null
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -0,0 +1,66 @@
+import {Message, PubSub} from '../src';
+import * as assert from 'assert';
+import {describe, it, after, before} from 'mocha';
+import {TestResources} from './testResources';
+import {BasicTracerProvider, SimpleSpanProcessor, InMemorySpanExporter} from '@opentelemetry/sdk-trace-base';
+import {Resource} from '@opentelemetry/resources';
+import {SEMRESATTRS_SERVICE_NAME} from '@opentelemetry/semantic-conventions';
+
+describe('OpenTelemetry Samples System Tests', () => {
+  const pubsub = new PubSub({enableOpenTelemetryTracing: true});
+  const resources = new TestResources('ps-sys-otel');
+  const exporter = new InMemorySpanExporter();
+
+  let topicName: string;
+  let subName: string;
+  let provider: BasicTracerProvider;
+  let processor: SimpleSpanProcessor;
+
+  before(async () => {
+    topicName = resources.generateName('topic');
+    subName = resources.generateName('sub');
+
+    provider = new BasicTracerProvider();
+    processor = new SimpleSpanProcessor(exporter);
+    provider.addSpanProcessor(processor);
+    provider.register();
+  });
+
+  after(async () => {
+    const [subscriptions] = await pubsub.getSubscriptions();
+    await Promise.all(
+      resources.filterForCleanup(subscriptions).map(x => x.delete?.())
+    );
+
+    const [topics] = await pubsub.getTopics();
+    await Promise.all(
+      resources.filterForCleanup(topics).map((x: any) => x.delete?.())
+    );
+  });
+
+  it('should publish and listen with OpenTelemetry tracing', async () => {
+    const [topic] = await pubsub.createTopic(topicName);
+    const [subscription] = await topic.createSubscription(subName);
+
+    const data = 'Hello, world!';
+    const dataBuffer = Buffer.from(data);
+
+    const messageId = await topic.publishMessage({data: dataBuffer});
+    assert.ok(messageId);
+
+    const message = await new Promise<Message>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout waiting for OTel message')), 15000);
+      subscription.once('message', (m: Message) => {
+        clearTimeout(timeout);
+        m.ack();
+        resolve(m);
+      });
+    });
+
+    assert.strictEqual(message.data.toString(), data);
+
+    await processor.forceFlush();
+    const spans = exporter.getFinishedSpans();
+    assert.ok(spans.length > 0, 'Should have generated spans');
+  });
+});
diff --git a/handwritten/pubsub/system-test/protobuf-samples.test.ts b/handwritten/pubsub/system-test/protobuf-samples.test.ts
new file mode 100644
index 000000000000..b096356dd515
--- /dev/null
+++ b/handwritten/pubsub/system-test/protobuf-samples.test.ts
@@ -0,0 +1,86 @@
+import {Message, PubSub, Schema} from '../src';
+import * as assert from 'assert';
+import {describe, it, after, before} from 'mocha';
+import {TestResources} from './testResources';
+import * as protobuf from 'protobufjs';
+import * as fs from 'fs';
+
+describe('Protobuf Samples System Tests', () => {
+  const pubsub = new PubSub();
+  const resources = new TestResources('ps-sys-proto');
+
+  let topicName: string;
+  let subName: string;
+  let schemaId: string;
+
+  before(async () => {
+    topicName = resources.generateName('topic');
+    subName = resources.generateName('sub');
+    schemaId = resources.generateName('schema');
+  });
+
+  after(async () => {
+    const [subscriptions] = await pubsub.getSubscriptions();
+    await Promise.all(
+      resources.filterForCleanup(subscriptions).map(x => x.delete?.())
+    );
+
+    const [topics] = await pubsub.getTopics();
+    await Promise.all(
+      resources.filterForCleanup(topics).map((x: any) => x.delete?.())
+    );
+
+    const schemas: any[] = [];
+    for await (const s of pubsub.listSchemas()) {
+      schemas.push(pubsub.schema(s.name!));
+    }
+    await Promise.all(
+      resources.filterForCleanup(schemas).map(x => x.delete?.())
+    );
+  });
+
+  it('should publish and listen for protobuf messages', async () => {
+    const definition = fs.readFileSync('system-test/fixtures/provinces.proto').toString();
+
+    await pubsub.createSchema(schemaId, 'PROTOCOL_BUFFER', definition);
+    const [topic] = await pubsub.createTopic({
+      name: topicName,
+      schemaSettings: {
+        schema: await pubsub.schema(schemaId).getName(),
+        encoding: 'BINARY',
+      },
+    });
+
+    const [subscription] = await topic.createSubscription(subName);
+
+    const root = await protobuf.load('system-test/fixtures/provinces.proto');
+    const Province = root.lookupType('utilities.Province');
+    const province = {
+      name: 'Ontario',
+      post_abbr: 'ON',
+    };
+
+    const messageObj = Province.create(province);
+    (messageObj as any).post_abbr = 'ON';
+    const dataBuffer = Buffer.from(Province.encode(messageObj).finish());
+
+    const messageId = await topic.publishMessage({data: dataBuffer});
+    assert.ok(messageId);
+
+    const message = await new Promise<Message>((resolve, reject) => {
+      const timeout = setTimeout(() => reject(new Error('Timeout waiting for Proto message')), 15000);
+      subscription.once('message', (m: Message) => {
+        clearTimeout(timeout);
+        m.ack();
+        resolve(m);
+      });
+    });
+
+    const schemaMetadata = Schema.metadataFromMessage(message.attributes);
+    assert.strictEqual(schemaMetadata.encoding, 'BINARY');
+
+    const result = Province.decode(message.data) as any;
+    assert.strictEqual(result.name, 'Ontario');
+    assert.strictEqual(result.postAbbr || result.post_abbr, 'ON');
+  });
+});

From 4597dcc037292029d307b483cb7e109fe842411d Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Tue, 28 Apr 2026 16:18:58 -0400
Subject: [PATCH 04/29] tests: review vibes

---
 .../pubsub/system-test/avro-samples.test.ts   | 18 ++++++++--
 .../system-test/batch-flow-samples.test.ts    | 15 +++++++-
 handwritten/pubsub/system-test/common.test.ts | 36 -------------------
 handwritten/pubsub/system-test/common.ts      | 26 --------------
 .../pubsub/system-test/otel-samples.test.ts   | 16 +++++++--
 .../system-test/protobuf-samples.test.ts      | 14 ++++++++
 .../pubsub/system-test/testResources.test.ts  |  2 +-
 .../pubsub/system-test/testResources.ts       |  2 +-
 8 files changed, 60 insertions(+), 69 deletions(-)
 delete mode 100644 handwritten/pubsub/system-test/common.test.ts
 delete mode 100644 handwritten/pubsub/system-test/common.ts

diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
index 246dccdf1f57..c8fb31c1adfd 100644
--- a/handwritten/pubsub/system-test/avro-samples.test.ts
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -1,3 +1,17 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 import {Message, PubSub, Schema} from '../src';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
@@ -65,7 +79,7 @@ describe('Avro Samples System Tests', () => {
     };
 
     const dataBuffer = type.toBuffer(province);
-    const messageId = await topic.publish(dataBuffer);
+    const messageId = await topic.publishMessage({data: dataBuffer});
     assert.ok(messageId);
 
     const message = await new Promise<Message>((resolve, reject) => {
@@ -100,7 +114,7 @@ describe('Avro Samples System Tests', () => {
     };
 
     const dataBuffer = type.toBuffer(province);
-    await topic.publish(dataBuffer);
+    await topic.publishMessage({data: dataBuffer});
 
     const message = await new Promise<Message>((resolve, reject) => {
       const timeout = setTimeout(() => reject(new Error('Timeout waiting for Avro revision')), 15000);
diff --git a/handwritten/pubsub/system-test/batch-flow-samples.test.ts b/handwritten/pubsub/system-test/batch-flow-samples.test.ts
index f00dd80c9462..5b908460c916 100644
--- a/handwritten/pubsub/system-test/batch-flow-samples.test.ts
+++ b/handwritten/pubsub/system-test/batch-flow-samples.test.ts
@@ -1,3 +1,17 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 import {Message, PubSub, PublishOptions} from '../src';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
@@ -81,7 +95,6 @@ describe('Batch and Flow Control Samples System Tests', () => {
     const topicWithFlow = pubsub.topic(flowTopicName, options);
     const flow = topicWithFlow.flowControlled();
 
-    const promises = [];
     for (let i = 0; i < 10; i++) {
       const wait = flow.publish({data: Buffer.from('flow control message')});
       if (wait) {
diff --git a/handwritten/pubsub/system-test/common.test.ts b/handwritten/pubsub/system-test/common.test.ts
deleted file mode 100644
index 77ca6642066b..000000000000
--- a/handwritten/pubsub/system-test/common.test.ts
+++ /dev/null
@@ -1,36 +0,0 @@
-// Copyright 2022 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     https://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-import {assert} from 'chai';
-import {describe, it} from 'mocha';
-import {commandFor} from './common';
-import * as path from 'path';
-
-describe('common (unit)', () => {
-  it('commandFor finds TS samples', () => {
-    const result = commandFor('createAvroSchema');
-    assert.strictEqual(
-      result,
-      `node ${path.join('build', 'createAvroSchema.js')}`
-    );
-  });
-
-  it('commandFor finds JS samples', () => {
-    const result = commandFor('createSubscription');
-    assert.strictEqual(
-      result,
-      `node ${path.join('build', 'createSubscription.js')}`
-    );
-  });
-});
diff --git a/handwritten/pubsub/system-test/common.ts b/handwritten/pubsub/system-test/common.ts
deleted file mode 100644
index 5529231d30ae..000000000000
--- a/handwritten/pubsub/system-test/common.ts
+++ /dev/null
@@ -1,26 +0,0 @@
-// Copyright 2021 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-import * as cp from 'child_process';
-import * as path from 'path';
-
-export const execSync = (cmd: string): string =>
-  cp.execSync(cmd, {encoding: 'utf-8'});
-
-// Processed versions of TS samples go to the same build location
-// as the rest of the JS samples.
-export function commandFor(action: string): string {
-  const jsPath = path.join('build', `${action}.js`);
-  return `node ${jsPath}`;
-}
diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
index 2ac84b303b82..b17bee4c5fa3 100644
--- a/handwritten/pubsub/system-test/otel-samples.test.ts
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -1,10 +1,22 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 import {Message, PubSub} from '../src';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
 import {BasicTracerProvider, SimpleSpanProcessor, InMemorySpanExporter} from '@opentelemetry/sdk-trace-base';
-import {Resource} from '@opentelemetry/resources';
-import {SEMRESATTRS_SERVICE_NAME} from '@opentelemetry/semantic-conventions';
 
 describe('OpenTelemetry Samples System Tests', () => {
   const pubsub = new PubSub({enableOpenTelemetryTracing: true});
diff --git a/handwritten/pubsub/system-test/protobuf-samples.test.ts b/handwritten/pubsub/system-test/protobuf-samples.test.ts
index b096356dd515..37179f57e15b 100644
--- a/handwritten/pubsub/system-test/protobuf-samples.test.ts
+++ b/handwritten/pubsub/system-test/protobuf-samples.test.ts
@@ -1,3 +1,17 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 import {Message, PubSub, Schema} from '../src';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
diff --git a/handwritten/pubsub/system-test/testResources.test.ts b/handwritten/pubsub/system-test/testResources.test.ts
index 0ddd84655bd0..618efbede3ea 100644
--- a/handwritten/pubsub/system-test/testResources.test.ts
+++ b/handwritten/pubsub/system-test/testResources.test.ts
@@ -1,4 +1,4 @@
-// Copyright 2022 Google LLC
+// Copyright 2022-2026 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
diff --git a/handwritten/pubsub/system-test/testResources.ts b/handwritten/pubsub/system-test/testResources.ts
index 3f5665096201..6f64e971b582 100644
--- a/handwritten/pubsub/system-test/testResources.ts
+++ b/handwritten/pubsub/system-test/testResources.ts
@@ -1,4 +1,4 @@
-// Copyright 2022 Google LLC
+// Copyright 2022-2026 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

From 78e2a55150dad17c05ded4c747cb5318a4a2cc5b Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Tue, 28 Apr 2026 16:20:26 -0400
Subject: [PATCH 05/29] chore: missing header

---
 handwritten/pubsub/system-test/avro-js.d.ts | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/handwritten/pubsub/system-test/avro-js.d.ts b/handwritten/pubsub/system-test/avro-js.d.ts
index 4a8496b76e69..4e2df28c5b2d 100644
--- a/handwritten/pubsub/system-test/avro-js.d.ts
+++ b/handwritten/pubsub/system-test/avro-js.d.ts
@@ -1,3 +1,17 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 declare module 'avro-js' {
   function parse(def: string): Parser;
 

From 9cc018f719f31c1531aa004ebd9fb4c28e9ccc59 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Tue, 28 Apr 2026 16:22:15 -0400
Subject: [PATCH 06/29] tests: use tokenMaker for timestamps

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 handwritten/pubsub/system-test/testResources.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/handwritten/pubsub/system-test/testResources.ts b/handwritten/pubsub/system-test/testResources.ts
index 6f64e971b582..40293dbee5ad 100644
--- a/handwritten/pubsub/system-test/testResources.ts
+++ b/handwritten/pubsub/system-test/testResources.ts
@@ -170,7 +170,7 @@ export class TestResources {
       if (name.startsWith(this.testSuiteId)) {
         const parts = name.split('-');
         const createdAt = Number(parts[1]);
-        const timeDiff = (Date.now() - createdAt) / (1000 * 60 * 60);
+        const timeDiff = (this.tokenMaker.timestamp() - createdAt) / (1000 * 60 * 60);
         if (timeDiff >= 2) {
           return true;
         }

From eec070a3cdaeea2172038f45b9d59d908493f7a2 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Tue, 28 Apr 2026 16:30:04 -0400
Subject: [PATCH 07/29] tests: merge sample-tests into pubsub and apply Gemini
 CR suggestion

---
 handwritten/pubsub/system-test/pubsub.ts      |  35 ++++--
 .../pubsub/system-test/sample-tests.ts        | 102 ------------------
 2 files changed, 27 insertions(+), 110 deletions(-)
 delete mode 100644 handwritten/pubsub/system-test/sample-tests.ts

diff --git a/handwritten/pubsub/system-test/pubsub.ts b/handwritten/pubsub/system-test/pubsub.ts
index 44aaff741107..c1b574ee7914 100644
--- a/handwritten/pubsub/system-test/pubsub.ts
+++ b/handwritten/pubsub/system-test/pubsub.ts
@@ -217,14 +217,33 @@ describe('pubsub', () => {
     });
 
     it('should publish a message', async () => {
-      const testTopic = await generateTopic('pub-msg');
-      const topic = testTopic.topic;
-      const message = {
-        data: Buffer.from('message from me'),
-        orderingKey: 'a',
-      };
-
-      const result = await topic.publishMessage(message);
+      const tname = generateTopicName('publish');
+      const sname = generateSubName('publish');
+      
+      const [topic] = await pubsub.topic(tname).get({autoCreate: true});
+      const [subscription] = await topic.subscription(sname).get({autoCreate: true});
+      
+      // --- From sample (publishMessage.js) ---
+      const data = 'Hello, world!';
+      const dataBuffer = Buffer.from(data);
+      const messageId = await topic.publishMessage({data: dataBuffer});
+      console.log(`Message ${messageId} published.`);
+      
+      // --- From test (topics.test.ts) ---
+      const message = await new Promise<Message>((resolve, reject) => {
+        const timeout = setTimeout(() => {
+          subscription.removeListener('message', messageHandler);
+          reject(new Error('Timeout'));
+        }, 10000);
+        function messageHandler(m: Message) {
+          clearTimeout(timeout);
+          m.ack();
+          resolve(m);
+        }
+        subscription.once('message', messageHandler);
+      });
+      
+      assert.strictEqual(message.data.toString(), data);
     });
 
     it('should publish a message with attributes', async () => {
diff --git a/handwritten/pubsub/system-test/sample-tests.ts b/handwritten/pubsub/system-test/sample-tests.ts
deleted file mode 100644
index a1679b4ebefb..000000000000
--- a/handwritten/pubsub/system-test/sample-tests.ts
+++ /dev/null
@@ -1,102 +0,0 @@
-// Copyright 2026 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-import {Message, PubSub} from '../src';
-import * as assert from 'assert';
-import {describe, it, after} from 'mocha';
-import {TestResources} from './testResources';
-
-describe('Combined Samples Tests', () => {
-  const pubsub = new PubSub();
-  const resources = new TestResources('pubsub_combined');
-
-  function topicName(testId: string): string {
-    return resources.generateName(testId);
-  }
-
-  function subName(testId: string): string {
-    return resources.generateName(testId);
-  }
-
-  async function cleanSubs() {
-    const [subscriptions] = await pubsub.getSubscriptions();
-    await Promise.all(
-      resources.filterForCleanup(subscriptions).map(x => x.delete?.())
-    );
-  }
-
-  async function cleanTopics() {
-    const [topics] = await pubsub.getTopics();
-    await Promise.all(
-      resources.filterForCleanup(topics).map((x: any) => x.delete?.())
-    );
-  }
-
-  after(async () => {
-    await cleanSubs();
-    await cleanTopics();
-  });
-
-  it('should create a topic', async () => {
-    const name = topicName('create');
-    
-    // --- From sample (createTopic.js) ---
-    await pubsub.createTopic(name);
-    console.log(`Topic ${name} created.`);
-    
-    // --- From test (topics.test.ts) ---
-    const [topics] = await pubsub.getTopics();
-    const exists = topics.some((t: any) => t.name.endsWith(name));
-    assert.ok(exists, 'Topic was created');
-  });
-
-  it('should publish a message', async () => {
-    const tname = topicName('publish');
-    const sname = subName('publish');
-    
-    const [topic] = await pubsub.topic(tname).get({autoCreate: true});
-    const [subscription] = await topic.subscription(sname).get({autoCreate: true});
-    
-    // --- From sample (publishMessage.js) ---
-    const data = 'Hello, world!';
-    const dataBuffer = Buffer.from(data);
-    const messageId = await topic.publishMessage({data: dataBuffer});
-    console.log(`Message ${messageId} published.`);
-    
-    // --- From test (topics.test.ts) ---
-    const message = await new Promise<Message>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout')), 10000);
-      subscription.once('message', (m: Message) => {
-        clearTimeout(timeout);
-        m.ack();
-        resolve(m);
-      });
-    });
-    
-    assert.strictEqual(message.data.toString(), data);
-  });
-
-  it('should delete a topic', async () => {
-    const name = topicName('delete');
-    await pubsub.topic(name).get({autoCreate: true});
-    
-    // --- From sample (deleteTopic.js) ---
-    await pubsub.topic(name).delete();
-    console.log(`Topic ${name} deleted.`);
-    
-    // --- From test (topics.test.ts) ---
-    const [exists] = await pubsub.topic(name).exists();
-    assert.strictEqual(exists, false);
-  });
-});

From 884a703f63930b61e029643821d0e18f83ce5995 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 1 May 2026 18:32:31 -0400
Subject: [PATCH 08/29] chore: re-add deleted avro-js

---
 handwritten/pubsub/system-test/avro-js.d.ts | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/handwritten/pubsub/system-test/avro-js.d.ts b/handwritten/pubsub/system-test/avro-js.d.ts
index 4e2df28c5b2d..20937ee94b06 100644
--- a/handwritten/pubsub/system-test/avro-js.d.ts
+++ b/handwritten/pubsub/system-test/avro-js.d.ts
@@ -1,4 +1,4 @@
-// Copyright 2026 Google LLC
+// Copyright 2021 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// This one doesn't seem to have typings.
 declare module 'avro-js' {
   function parse(def: string): Parser;
 

From 092929a3ad4a1718c1b5369c1713bce0441bf9a8 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Wed, 3 Jun 2026 15:39:14 -0400
Subject: [PATCH 09/29] fix: remove uuid module to simplify dependencies

---
 handwritten/pubsub/package.json                 | 1 -
 handwritten/pubsub/system-test/testResources.ts | 4 +---
 handwritten/pubsub/test/message-queues.ts       | 1 -
 handwritten/pubsub/test/message-stream.ts       | 1 -
 handwritten/pubsub/test/subscriber.ts           | 1 -
 handwritten/pubsub/test/testResources.ts        | 2 --
 6 files changed, 1 insertion(+), 9 deletions(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index d44e3933eaf7..fb400bc7143b 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -84,7 +84,6 @@
     "@types/proxyquire": "^1.3.31",
     "@types/sinon": "^21.0.0",
     "@types/tmp": "^0.2.6",
-    "@types/uuid": "^10.0.0",
     "avro-js": "^1.12.1",
     "c8": "^10.1.3",
     "chai": "^6.2.2",
diff --git a/handwritten/pubsub/system-test/testResources.ts b/handwritten/pubsub/system-test/testResources.ts
index 40293dbee5ad..705f5fed0a72 100644
--- a/handwritten/pubsub/system-test/testResources.ts
+++ b/handwritten/pubsub/system-test/testResources.ts
@@ -16,12 +16,10 @@
 // across the two test structures, but because of the tangle of rootDirs
 // and package.json "files", it's hard to avoid it.
 
-import * as uuid from 'uuid';
-
 // Returns a shortened UUID that can be used to identify a
 // specific run of a specific test.
 function shortUUID() {
-  return uuid.v4().split('-').shift()!;
+  return crypto.randomUUID().split('-').shift()!;
 }
 
 export interface TokenMaker {
diff --git a/handwritten/pubsub/test/message-queues.ts b/handwritten/pubsub/test/message-queues.ts
index d44de1dfd73a..e4fba54464a8 100644
--- a/handwritten/pubsub/test/message-queues.ts
+++ b/handwritten/pubsub/test/message-queues.ts
@@ -19,7 +19,6 @@ import {describe, it, before, beforeEach, afterEach} from 'mocha';
 import {EventEmitter} from 'events';
 import {CallOptions, GoogleError, loggingUtils, Status} from 'google-gax';
 import * as sinon from 'sinon';
-import * as crypto from 'crypto';
 import defer = require('p-defer');
 
 import * as messageTypes from '../src/message-queues';
diff --git a/handwritten/pubsub/test/message-stream.ts b/handwritten/pubsub/test/message-stream.ts
index d5be750bbcb0..0a3bd5be7c87 100644
--- a/handwritten/pubsub/test/message-stream.ts
+++ b/handwritten/pubsub/test/message-stream.ts
@@ -20,7 +20,6 @@ import {grpc} from 'google-gax';
 import * as proxyquire from 'proxyquire';
 import * as sinon from 'sinon';
 import {Duplex, PassThrough} from 'stream';
-import * as crypto from 'crypto';
 import * as defer from 'p-defer';
 import {promisify} from 'util';
 
diff --git a/handwritten/pubsub/test/subscriber.ts b/handwritten/pubsub/test/subscriber.ts
index d74cd76c005f..fcad768cdb2d 100644
--- a/handwritten/pubsub/test/subscriber.ts
+++ b/handwritten/pubsub/test/subscriber.ts
@@ -22,7 +22,6 @@ import {common as protobuf} from 'protobufjs';
 import * as proxyquire from 'proxyquire';
 import * as sinon from 'sinon';
 import {PassThrough} from 'stream';
-import * as crypto from 'crypto';
 import * as opentelemetry from '@opentelemetry/api';
 import {google} from '../protos/protos';
 import * as defer from 'p-defer';
diff --git a/handwritten/pubsub/test/testResources.ts b/handwritten/pubsub/test/testResources.ts
index 4503138e51cd..ee8a89de8c03 100644
--- a/handwritten/pubsub/test/testResources.ts
+++ b/handwritten/pubsub/test/testResources.ts
@@ -16,8 +16,6 @@
 // across the two test structures, but because of the tangle of rootDirs
 // and package.json "files", it's hard to avoid it.
 
-import * as crypto from 'crypto';
-
 // Returns a shortened UUID that can be used to identify a
 // specific run of a specific test.
 function shortUUID() {

From 78d47d99f5274866728ad6febb3c41fcf4d6a05b Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 5 Jun 2026 15:23:52 -0400
Subject: [PATCH 10/29] tests: fix broken otel tests

---
 handwritten/pubsub/package.json               |  3 +--
 .../pubsub/system-test/otel-samples.test.ts   | 26 ++++++++++++-------
 2 files changed, 17 insertions(+), 12 deletions(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index fb400bc7143b..9945073950d3 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -69,10 +69,9 @@
     "p-defer": "^3.0.0"
   },
   "devDependencies": {
-    "@google-cloud/opentelemetry-cloud-trace-exporter": "^2.4.1",
     "@grpc/proto-loader": "^0.8.0",
     "@opentelemetry/sdk-trace-base": "^1.17.0",
-    "@opentelemetry/sdk-trace-node": "^2.7.0",
+    "@opentelemetry/sdk-trace-node": "^1.17.0",
     "@types/chai": "^5.2.3",
     "@types/duplexify": "^3.6.4",
     "@types/extend": "^3.0.4",
diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
index b17bee4c5fa3..5824e0177818 100644
--- a/handwritten/pubsub/system-test/otel-samples.test.ts
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -12,27 +12,36 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import {Message, PubSub} from '../src';
+import {Message, PubSub, Subscription, Topic} from '../src';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
-import {BasicTracerProvider, SimpleSpanProcessor, InMemorySpanExporter} from '@opentelemetry/sdk-trace-base';
+import {NodeTracerProvider} from '@opentelemetry/sdk-trace-node';
+import {SimpleSpanProcessor, InMemorySpanExporter} from '@opentelemetry/sdk-trace-base';
 
 describe('OpenTelemetry Samples System Tests', () => {
   const pubsub = new PubSub({enableOpenTelemetryTracing: true});
   const resources = new TestResources('ps-sys-otel');
-  const exporter = new InMemorySpanExporter();
 
   let topicName: string;
   let subName: string;
-  let provider: BasicTracerProvider;
+  let topic: Topic;
+  let subscription: Subscription;
   let processor: SimpleSpanProcessor;
+  let provider: NodeTracerProvider;
+  let exporter: InMemorySpanExporter;
 
   before(async () => {
-    topicName = resources.generateName('topic');
-    subName = resources.generateName('sub');
+    topicName = resources.generateName('ot');
+    subName = resources.generateName('ot');
+    topic = (await pubsub.createTopic(topicName))[0];
+    subscription = (await topic.createSubscription(subName))[0];
 
-    provider = new BasicTracerProvider();
+    exporter = new InMemorySpanExporter();
+
+    // Build a tracer provider and a span processor to do
+    // something with the spans we're generating.
+    provider = new NodeTracerProvider();
     processor = new SimpleSpanProcessor(exporter);
     provider.addSpanProcessor(processor);
     provider.register();
@@ -51,9 +60,6 @@ describe('OpenTelemetry Samples System Tests', () => {
   });
 
   it('should publish and listen with OpenTelemetry tracing', async () => {
-    const [topic] = await pubsub.createTopic(topicName);
-    const [subscription] = await topic.createSubscription(subName);
-
     const data = 'Hello, world!';
     const dataBuffer = Buffer.from(data);
 

From 1bc72df73189ad58fdfb947c23da3989d517f618 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 5 Jun 2026 15:24:21 -0400
Subject: [PATCH 11/29] chore: add back in crypto imports

---
 handwritten/pubsub/system-test/testResources.ts | 2 ++
 handwritten/pubsub/test/message-queues.ts       | 1 +
 handwritten/pubsub/test/message-stream.ts       | 1 +
 handwritten/pubsub/test/subscriber.ts           | 1 +
 handwritten/pubsub/test/testResources.ts        | 2 ++
 5 files changed, 7 insertions(+)

diff --git a/handwritten/pubsub/system-test/testResources.ts b/handwritten/pubsub/system-test/testResources.ts
index 705f5fed0a72..0fd11552ad60 100644
--- a/handwritten/pubsub/system-test/testResources.ts
+++ b/handwritten/pubsub/system-test/testResources.ts
@@ -16,6 +16,8 @@
 // across the two test structures, but because of the tangle of rootDirs
 // and package.json "files", it's hard to avoid it.
 
+import * as crypto from 'node:crypto';
+
 // Returns a shortened UUID that can be used to identify a
 // specific run of a specific test.
 function shortUUID() {
diff --git a/handwritten/pubsub/test/message-queues.ts b/handwritten/pubsub/test/message-queues.ts
index e4fba54464a8..a95fa7fbfd04 100644
--- a/handwritten/pubsub/test/message-queues.ts
+++ b/handwritten/pubsub/test/message-queues.ts
@@ -20,6 +20,7 @@ import {EventEmitter} from 'events';
 import {CallOptions, GoogleError, loggingUtils, Status} from 'google-gax';
 import * as sinon from 'sinon';
 import defer = require('p-defer');
+import * as crypto from 'node:crypto';
 
 import * as messageTypes from '../src/message-queues';
 import {BatchError} from '../src/message-queues';
diff --git a/handwritten/pubsub/test/message-stream.ts b/handwritten/pubsub/test/message-stream.ts
index 0a3bd5be7c87..04c59601a7e3 100644
--- a/handwritten/pubsub/test/message-stream.ts
+++ b/handwritten/pubsub/test/message-stream.ts
@@ -22,6 +22,7 @@ import * as sinon from 'sinon';
 import {Duplex, PassThrough} from 'stream';
 import * as defer from 'p-defer';
 import {promisify} from 'util';
+import * as crypto from 'node:crypto';
 
 import * as messageTypes from '../src/message-stream';
 import {Subscriber} from '../src/subscriber';
diff --git a/handwritten/pubsub/test/subscriber.ts b/handwritten/pubsub/test/subscriber.ts
index fcad768cdb2d..2d322e0a27a6 100644
--- a/handwritten/pubsub/test/subscriber.ts
+++ b/handwritten/pubsub/test/subscriber.ts
@@ -25,6 +25,7 @@ import {PassThrough} from 'stream';
 import * as opentelemetry from '@opentelemetry/api';
 import {google} from '../protos/protos';
 import * as defer from 'p-defer';
+import * as crypto from 'node:crypto';
 
 import {HistogramOptions} from '../src/histogram';
 import {FlowControlOptions, LeaseManager} from '../src/lease-manager';
diff --git a/handwritten/pubsub/test/testResources.ts b/handwritten/pubsub/test/testResources.ts
index ee8a89de8c03..93df7dbc9c6e 100644
--- a/handwritten/pubsub/test/testResources.ts
+++ b/handwritten/pubsub/test/testResources.ts
@@ -16,6 +16,8 @@
 // across the two test structures, but because of the tangle of rootDirs
 // and package.json "files", it's hard to avoid it.
 
+import * as crypto from 'node:crypto';
+
 // Returns a shortened UUID that can be used to identify a
 // specific run of a specific test.
 function shortUUID() {

From 9ea19a2efbc3c9e2884083a1ef84e53462898c1b Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 5 Jun 2026 15:37:20 -0400
Subject: [PATCH 12/29] chore: fix(?) auth typing

---
 handwritten/pubsub/src/pubsub.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/handwritten/pubsub/src/pubsub.ts b/handwritten/pubsub/src/pubsub.ts
index e9c9fc7bdf1e..4ec6002d12f3 100644
--- a/handwritten/pubsub/src/pubsub.ts
+++ b/handwritten/pubsub/src/pubsub.ts
@@ -17,7 +17,7 @@
 import {paginator} from '@google-cloud/paginator';
 import {replaceProjectIdToken} from '@google-cloud/projectify';
 import * as extend from 'extend';
-import {AuthClient, GoogleAuth, GoogleAuthOptions} from 'google-auth-library';
+import {AuthClient, GoogleAuth} from 'google-auth-library';
 import * as gax from 'google-gax';
 
 // eslint-disable-next-line @typescript-eslint/no-var-requires
@@ -335,7 +335,7 @@ export class PubSub {
     this.isEmulator = false;
     this.determineBaseUrl_();
     this.api = {};
-    this.auth = new GoogleAuth(this.options as GoogleAuthOptions<AuthClient>);
+    this.auth = new GoogleAuth(this.options as gax.GoogleAuthOptions<AuthClient>);
     this.projectId = this.options.projectId || PROJECT_ID_PLACEHOLDER;
     if (this.projectId !== PROJECT_ID_PLACEHOLDER) {
       this.name = PubSub.formatName_(this.projectId);

From e4d522003fd78cc9aa8e97a712dd039a71b2d25b Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 5 Jun 2026 16:35:32 -0400
Subject: [PATCH 13/29] fix: a bunch of pnpm breakages

---
 handwritten/pubsub/package.json                    |  4 ++--
 handwritten/pubsub/src/lease-manager.ts            | 12 ++++++------
 handwritten/pubsub/src/logs.ts                     |  5 +++--
 handwritten/pubsub/src/message-queues.ts           |  6 +++---
 handwritten/pubsub/src/message-stream.ts           |  7 +++----
 handwritten/pubsub/src/publisher/message-queues.ts |  6 +++---
 handwritten/pubsub/src/pubsub.ts                   |  4 ++--
 handwritten/pubsub/src/subscriber.ts               | 10 +++++-----
 8 files changed, 27 insertions(+), 27 deletions(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index 9945073950d3..4e41a2a9b874 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -59,8 +59,8 @@
     "@opentelemetry/semantic-conventions": "~1.39.0",
     "arrify": "^2.0.0",
     "extend": "^3.0.2",
-    "google-auth-library": "^10.5.0",
-    "google-gax": "^5.0.5",
+    "google-auth-library": "^10.7.0",
+    "google-gax": "^5.0.7",
     "google-logging-utils": "^1.1.3",
     "heap-js": "^2.6.0",
     "is-stream-ended": "^0.1.4",
diff --git a/handwritten/pubsub/src/lease-manager.ts b/handwritten/pubsub/src/lease-manager.ts
index 447c4d7e6e8c..aa10d117b352 100644
--- a/handwritten/pubsub/src/lease-manager.ts
+++ b/handwritten/pubsub/src/lease-manager.ts
@@ -20,18 +20,18 @@ import {AckError, Message, Subscriber} from './subscriber';
 import {defaultOptions} from './default-options';
 import {Duration} from './temporal';
 import {DebugMessage} from './debug';
-import {logs as baseLogs, LoggingFunction} from './logs';
+import {logs as baseLogs, Loggers} from './logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs = {
-  callbackDelivery: baseLogs.pubsub.sublog('callback-delivery') as LoggingFunction,
-  callbackExceptions: baseLogs.pubsub.sublog('callback-exceptions') as LoggingFunction,
-  expiry: baseLogs.pubsub.sublog('expiry') as LoggingFunction,
-  subscriberFlowControl: baseLogs.pubsub.sublog('subscriber-flow-control') as LoggingFunction,
+export const logs: Loggers = {
+  callbackDelivery: baseLogs.pubsub.sublog('callback-delivery'),
+  callbackExceptions: baseLogs.pubsub.sublog('callback-exceptions'),
+  expiry: baseLogs.pubsub.sublog('expiry'),
+  subscriberFlowControl: baseLogs.pubsub.sublog('subscriber-flow-control'),
 };
 
 export interface FlowControlOptions {
diff --git a/handwritten/pubsub/src/logs.ts b/handwritten/pubsub/src/logs.ts
index dadddcf60bee..e3fa5c3841e7 100644
--- a/handwritten/pubsub/src/logs.ts
+++ b/handwritten/pubsub/src/logs.ts
@@ -15,12 +15,13 @@
 import {loggingUtils} from 'google-gax';
 
 export type LoggingFunction = loggingUtils.AdhocDebugLogFunction;
+export type Loggers = Record<string, LoggingFunction>;
 
 /**
  * Base logger. Other loggers will derive from this one.
  *
  * @private
  */
-export const logs = {
-  pubsub: loggingUtils.log('pubsub') as LoggingFunction,
+export const logs: Loggers = {
+  pubsub: loggingUtils.log('pubsub'),
 };
diff --git a/handwritten/pubsub/src/message-queues.ts b/handwritten/pubsub/src/message-queues.ts
index bdc10ecb740d..6979e1543011 100644
--- a/handwritten/pubsub/src/message-queues.ts
+++ b/handwritten/pubsub/src/message-queues.ts
@@ -34,15 +34,15 @@ import {Duration} from './temporal';
 import {addToBucket} from './util';
 import {DebugMessage} from './debug';
 import * as tracing from './telemetry-tracing';
-import {logs as baseLogs, LoggingFunction} from './logs';
+import {logs as baseLogs, Loggers} from './logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs = {
-  ackBatch: baseLogs.pubsub.sublog('ack-batch') as LoggingFunction,
+export const logs: Loggers = {
+  ackBatch: baseLogs.pubsub.sublog('ack-batch'),
 };
 
 export interface ReducedMessage {
diff --git a/handwritten/pubsub/src/message-stream.ts b/handwritten/pubsub/src/message-stream.ts
index f1f2102f4d9a..abc32fcaddca 100644
--- a/handwritten/pubsub/src/message-stream.ts
+++ b/handwritten/pubsub/src/message-stream.ts
@@ -26,16 +26,15 @@ import {defaultOptions} from './default-options';
 import {Duration} from './temporal';
 import {ExponentialRetry} from './exponential-retry';
 import {DebugMessage} from './debug';
-import {randomUUID} from 'crypto';
-import {logs as baseLogs, LoggingFunction} from './logs';
+import {logs as baseLogs, Loggers} from './logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs = {
-  subscriberStreams: baseLogs.pubsub.sublog('subscriber-streams') as LoggingFunction,
+export const logs: Loggers = {
+  subscriberStreams: baseLogs.pubsub.sublog('subscriber-streams'),
 };
 
 /*!
diff --git a/handwritten/pubsub/src/publisher/message-queues.ts b/handwritten/pubsub/src/publisher/message-queues.ts
index 0227e585cecd..34c0a99d2963 100644
--- a/handwritten/pubsub/src/publisher/message-queues.ts
+++ b/handwritten/pubsub/src/publisher/message-queues.ts
@@ -24,15 +24,15 @@ import {google} from '../../protos/protos';
 import * as tracing from '../telemetry-tracing';
 import {filterMessage} from './pubsub-message';
 import {promisify} from 'util';
-import {logs as baseLogs, LoggingFunction} from '../logs';
+import {logs as baseLogs, Loggers} from '../logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs = {
-  publishBatch: baseLogs.pubsub.sublog('publish-batch') as LoggingFunction,
+export const logs: Loggers = {
+  publishBatch: baseLogs.pubsub.sublog('publish-batch'),
 };
 
 /**
diff --git a/handwritten/pubsub/src/pubsub.ts b/handwritten/pubsub/src/pubsub.ts
index 4ec6002d12f3..e9c9fc7bdf1e 100644
--- a/handwritten/pubsub/src/pubsub.ts
+++ b/handwritten/pubsub/src/pubsub.ts
@@ -17,7 +17,7 @@
 import {paginator} from '@google-cloud/paginator';
 import {replaceProjectIdToken} from '@google-cloud/projectify';
 import * as extend from 'extend';
-import {AuthClient, GoogleAuth} from 'google-auth-library';
+import {AuthClient, GoogleAuth, GoogleAuthOptions} from 'google-auth-library';
 import * as gax from 'google-gax';
 
 // eslint-disable-next-line @typescript-eslint/no-var-requires
@@ -335,7 +335,7 @@ export class PubSub {
     this.isEmulator = false;
     this.determineBaseUrl_();
     this.api = {};
-    this.auth = new GoogleAuth(this.options as gax.GoogleAuthOptions<AuthClient>);
+    this.auth = new GoogleAuth(this.options as GoogleAuthOptions<AuthClient>);
     this.projectId = this.options.projectId || PROJECT_ID_PLACEHOLDER;
     if (this.projectId !== PROJECT_ID_PLACEHOLDER) {
       this.name = PubSub.formatName_(this.projectId);
diff --git a/handwritten/pubsub/src/subscriber.ts b/handwritten/pubsub/src/subscriber.ts
index 8332e93c819d..e30285c942ac 100644
--- a/handwritten/pubsub/src/subscriber.ts
+++ b/handwritten/pubsub/src/subscriber.ts
@@ -32,7 +32,7 @@ import {Duration, atMost as durationAtMost} from './temporal';
 import {EventEmitter} from 'events';
 
 import {awaitWithTimeout} from './util';
-import {logs as baseLogs, LoggingFunction} from './logs';
+import {logs as baseLogs, Loggers} from './logs';
 
 export {StatusError} from './message-stream';
 
@@ -41,10 +41,10 @@ export {StatusError} from './message-stream';
  *
  * @private
  */
-export const logs = {
-  slowAck: baseLogs.pubsub.sublog('slow-ack') as LoggingFunction,
-  ackNack: baseLogs.pubsub.sublog('ack-nack') as LoggingFunction,
-  debug: baseLogs.pubsub.sublog('debug') as LoggingFunction,
+export const logs: Loggers = {
+  slowAck: baseLogs.pubsub.sublog('slow-ack'),
+  ackNack: baseLogs.pubsub.sublog('ack-nack'),
+  debug: baseLogs.pubsub.sublog('debug'),
 };
 
 export type PullResponse = google.pubsub.v1.IStreamingPullResponse;

From ea1b956b853ad0239874d977f139fd3a1ba36dbe Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 9 Jul 2026 16:33:46 -0400
Subject: [PATCH 14/29] tests: clean up generated avro/proto system tests

---
 .../pubsub/system-test/avro-samples.test.ts   | 39 +++++++++-------
 handwritten/pubsub/system-test/proto-js.d.ts  |  4 ++
 .../system-test/protobuf-samples.test.ts      | 44 +++++++++----------
 3 files changed, 48 insertions(+), 39 deletions(-)
 create mode 100644 handwritten/pubsub/system-test/proto-js.d.ts

diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
index c8fb31c1adfd..d471b2cf6121 100644
--- a/handwritten/pubsub/system-test/avro-samples.test.ts
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -23,27 +23,13 @@ describe('Avro Samples System Tests', () => {
   const pubsub = new PubSub();
   const resources = new TestResources('ps-sys-avro');
 
-  let topicName: string;
-  let subName: string;
   let schemaId: string;
 
   before(async () => {
-    topicName = resources.generateName('topic');
-    subName = resources.generateName('sub');
     schemaId = resources.generateName('schema');
 
     const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString();
     await pubsub.createSchema(schemaId, 'AVRO', definition);
-    await pubsub.createTopic({
-      name: topicName,
-      schemaSettings: {
-        schema: await pubsub.schema(schemaId).getName(),
-        encoding: 'BINARY',
-      },
-    });
-
-    const [topic] = await pubsub.topic(topicName).get();
-    await topic.createSubscription(subName);
   });
 
   after(async () => {
@@ -66,9 +52,18 @@ describe('Avro Samples System Tests', () => {
     );
   });
 
-  it('should publish and listen for avro records', async () => {
+  async function publishAndListen(encoding: 'BINARY' | 'JSON') {
+    const topicName = resources.generateName(`topic-${encoding}`);
+    const subName = resources.generateName(`sub-${encoding}`);
+
     const definition = fs.readFileSync('system-test/fixtures/provinces.avsc').toString();
-    const [topic] = await pubsub.topic(topicName).get();
+    const [topic] = await pubsub.createTopic({
+      name: topicName,
+      schemaSettings: {
+        schema: await pubsub.schema(schemaId).getName(),
+        encoding,
+      }
+    });
     const [subscription] = await pubsub.subscription(subName).get();
 
     const type = avro.parse(definition);
@@ -92,11 +87,19 @@ describe('Avro Samples System Tests', () => {
     });
 
     const schemaMetadata = Schema.metadataFromMessage(message.attributes);
-    assert.strictEqual(schemaMetadata.encoding, 'BINARY');
+    assert.strictEqual(schemaMetadata.encoding, encoding);
 
     const result = type.fromBuffer(message.data) as any;
     assert.strictEqual(result.name, 'Ontario');
     assert.strictEqual(result.post_abbr, 'ON');
+  }
+
+  it('should publish and listen for avro records (binary encoding)', async () => {
+    publishAndListen('BINARY');
+  });
+
+  it('should publish and listen for avro records (json encoding)', async () => {
+    publishAndListen('JSON');
   });
 
   it('should listen for avro records with revisions', async () => {
@@ -104,6 +107,8 @@ describe('Avro Samples System Tests', () => {
 
     const schemaClient = await pubsub.getSchemaClient();
 
+    const topicName = resources.generateName(`topic-rev`);
+    const subName = resources.generateName(`sub-rev`);
     const [topic] = await pubsub.topic(topicName).get();
     const [subscription] = await pubsub.subscription(subName).get();
 
diff --git a/handwritten/pubsub/system-test/proto-js.d.ts b/handwritten/pubsub/system-test/proto-js.d.ts
new file mode 100644
index 000000000000..c498b790a35a
--- /dev/null
+++ b/handwritten/pubsub/system-test/proto-js.d.ts
@@ -0,0 +1,4 @@
+interface ProvinceObject {
+  name: string;
+  postAbbr: string;
+}
diff --git a/handwritten/pubsub/system-test/protobuf-samples.test.ts b/handwritten/pubsub/system-test/protobuf-samples.test.ts
index 37179f57e15b..7f058cdd677d 100644
--- a/handwritten/pubsub/system-test/protobuf-samples.test.ts
+++ b/handwritten/pubsub/system-test/protobuf-samples.test.ts
@@ -23,16 +23,6 @@ describe('Protobuf Samples System Tests', () => {
   const pubsub = new PubSub();
   const resources = new TestResources('ps-sys-proto');
 
-  let topicName: string;
-  let subName: string;
-  let schemaId: string;
-
-  before(async () => {
-    topicName = resources.generateName('topic');
-    subName = resources.generateName('sub');
-    schemaId = resources.generateName('schema');
-  });
-
   after(async () => {
     const [subscriptions] = await pubsub.getSubscriptions();
     await Promise.all(
@@ -54,9 +44,12 @@ describe('Protobuf Samples System Tests', () => {
   });
 
   it('should publish and listen for protobuf messages', async () => {
+    const topicName = resources.generateName('topic');
+    const subName = resources.generateName('sub');
+    const schemaId = resources.generateName('schema');
     const definition = fs.readFileSync('system-test/fixtures/provinces.proto').toString();
-
     await pubsub.createSchema(schemaId, 'PROTOCOL_BUFFER', definition);
+
     const [topic] = await pubsub.createTopic({
       name: topicName,
       schemaSettings: {
@@ -67,34 +60,41 @@ describe('Protobuf Samples System Tests', () => {
 
     const [subscription] = await topic.createSubscription(subName);
 
+    // Make an encoder using the protobufjs library.
+    //
+    // Since we're providing the test message for a specific schema here, we'll
+    // also code in the path to a sample proto definition.
     const root = await protobuf.load('system-test/fixtures/provinces.proto');
     const Province = root.lookupType('utilities.Province');
-    const province = {
+    const province: ProvinceObject = {
       name: 'Ontario',
-      post_abbr: 'ON',
+      postAbbr: 'ON',
     };
 
-    const messageObj = Province.create(province);
-    (messageObj as any).post_abbr = 'ON';
-    const dataBuffer = Buffer.from(Province.encode(messageObj).finish());
+    const message = Province.create(province);
+    const dataBuffer = Buffer.from(Province.encode(message).finish());
 
     const messageId = await topic.publishMessage({data: dataBuffer});
     assert.ok(messageId);
 
-    const message = await new Promise<Message>((resolve, reject) => {
+    let received!: Message;
+    await new Promise<Message>((resolve, reject) => {
       const timeout = setTimeout(() => reject(new Error('Timeout waiting for Proto message')), 15000);
       subscription.once('message', (m: Message) => {
-        clearTimeout(timeout);
         m.ack();
+        received = m;
+        clearTimeout(timeout);
         resolve(m);
       });
     });
 
-    const schemaMetadata = Schema.metadataFromMessage(message.attributes);
+    const schemaMetadata = Schema.metadataFromMessage(received.attributes);
     assert.strictEqual(schemaMetadata.encoding, 'BINARY');
 
-    const result = Province.decode(message.data) as any;
-    assert.strictEqual(result.name, 'Ontario');
-    assert.strictEqual(result.postAbbr || result.post_abbr, 'ON');
+    for (let i = 0; i < received.length; i++) {
+      const result = Province.decode(received.data) as any;
+      assert.strictEqual(result.name, 'Ontario');
+      assert.strictEqual(result.postAbbr || result.post_abbr, 'ON');
+    }
   });
 });

From 0d50d953c82b834320639a7735fc96ffc29f2c0f Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 9 Jul 2026 16:40:39 -0400
Subject: [PATCH 15/29] chore: add copyright header

---
 handwritten/pubsub/system-test/proto-js.d.ts | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/handwritten/pubsub/system-test/proto-js.d.ts b/handwritten/pubsub/system-test/proto-js.d.ts
index c498b790a35a..c83a47f02562 100644
--- a/handwritten/pubsub/system-test/proto-js.d.ts
+++ b/handwritten/pubsub/system-test/proto-js.d.ts
@@ -1,3 +1,17 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 interface ProvinceObject {
   name: string;
   postAbbr: string;

From eb26949c86ff4a03a2a62b96c08607e6ec73809c Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 9 Jul 2026 17:26:44 -0400
Subject: [PATCH 16/29] tests: small fix from avro updates

---
 handwritten/pubsub/system-test/avro-samples.test.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
index d471b2cf6121..600e92f45d09 100644
--- a/handwritten/pubsub/system-test/avro-samples.test.ts
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -109,8 +109,8 @@ describe('Avro Samples System Tests', () => {
 
     const topicName = resources.generateName(`topic-rev`);
     const subName = resources.generateName(`sub-rev`);
-    const [topic] = await pubsub.topic(topicName).get();
-    const [subscription] = await pubsub.subscription(subName).get();
+    const [topic] = await pubsub.createTopic(topicName);
+    const [subscription] = await pubsub.createSubscription(topicName, subName);
 
     const type = avro.parse(definition);
     const province = {

From 4d9cf1267fddffb6d7b36d16ca9b9be3c658334d Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 9 Jul 2026 17:47:53 -0400
Subject: [PATCH 17/29] tests: shut down otel provider after system test

---
 handwritten/pubsub/system-test/otel-samples.test.ts | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
index 5824e0177818..0aa9b95e52b3 100644
--- a/handwritten/pubsub/system-test/otel-samples.test.ts
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -48,6 +48,9 @@ describe('OpenTelemetry Samples System Tests', () => {
   });
 
   after(async () => {
+    // Don't interfere with other tests.
+    provider.shutdown();
+
     const [subscriptions] = await pubsub.getSubscriptions();
     await Promise.all(
       resources.filterForCleanup(subscriptions).map(x => x.delete?.())

From b215ee1eb944d14d206c8287c0ee12d72be5411f Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 9 Jul 2026 18:40:40 -0400
Subject: [PATCH 18/29] fix: refactor error fix for crypto

---
 handwritten/pubsub/src/message-stream.ts | 1 +
 1 file changed, 1 insertion(+)

diff --git a/handwritten/pubsub/src/message-stream.ts b/handwritten/pubsub/src/message-stream.ts
index abc32fcaddca..c2201fe3420d 100644
--- a/handwritten/pubsub/src/message-stream.ts
+++ b/handwritten/pubsub/src/message-stream.ts
@@ -18,6 +18,7 @@ import {promisify} from '@google-cloud/promisify';
 import {ClientStub, GoogleError, grpc} from 'google-gax';
 import * as isStreamEnded from 'is-stream-ended';
 import {PassThrough} from 'stream';
+import {randomUUID} from 'crypto';
 
 import {PullRetry} from './pull-retry';
 import {Subscriber} from './subscriber';

From f3d4570ec4a814d109d4d525924c8e1b32c8c0cb Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Wed, 12 Aug 2026 16:49:28 -0400
Subject: [PATCH 19/29] chore: revert merged version changes

---
 handwritten/pubsub/package.json | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index 80dad3738111..e36cbe115f77 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -71,8 +71,9 @@
   },
   "devDependencies": {
     "@grpc/proto-loader": "^0.8.0",
-    "@opentelemetry/sdk-trace-base": "^2.8.0",
-    "@opentelemetry/sdk-trace-node": "^2.8.0",
+    "@opentelemetry/sdk-trace-base": "^@.17.0",
+    "@opentelemetry/sdk-trace-node": "^1.17.0",
+    "@types/chai": "^5.2.3",
     "@types/duplexify": "^3.6.4",
     "@types/extend": "^3.0.4",
     "@types/lodash.snakecase": "^4.1.9",

From ff93f81fc05a6ee254c63bae549d88f6e7df6f3e Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Wed, 12 Aug 2026 16:50:05 -0400
Subject: [PATCH 20/29] chore: typo

---
 handwritten/pubsub/package.json | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index e36cbe115f77..ca7a6f34bbd1 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -71,7 +71,7 @@
   },
   "devDependencies": {
     "@grpc/proto-loader": "^0.8.0",
-    "@opentelemetry/sdk-trace-base": "^@.17.0",
+    "@opentelemetry/sdk-trace-base": "^1.17.0",
     "@opentelemetry/sdk-trace-node": "^1.17.0",
     "@types/chai": "^5.2.3",
     "@types/duplexify": "^3.6.4",

From a9ed76bb35b6d8e891baca3c38d3e0f299bc45a7 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Wed, 12 Aug 2026 17:34:51 -0400
Subject: [PATCH 21/29] chore: revert the revert, do the span warp again

---
 handwritten/pubsub/package.json                     | 4 ++--
 handwritten/pubsub/system-test/otel-samples.test.ts | 5 +++--
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index ca7a6f34bbd1..5d0fc5f0936c 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -71,8 +71,8 @@
   },
   "devDependencies": {
     "@grpc/proto-loader": "^0.8.0",
-    "@opentelemetry/sdk-trace-base": "^1.17.0",
-    "@opentelemetry/sdk-trace-node": "^1.17.0",
+    "@opentelemetry/sdk-trace-base": "^2.8.0",
+    "@opentelemetry/sdk-trace-node": "^2.8.0",
     "@types/chai": "^5.2.3",
     "@types/duplexify": "^3.6.4",
     "@types/extend": "^3.0.4",
diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
index 0aa9b95e52b3..beb518a9f875 100644
--- a/handwritten/pubsub/system-test/otel-samples.test.ts
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -41,9 +41,10 @@ describe('OpenTelemetry Samples System Tests', () => {
 
     // Build a tracer provider and a span processor to do
     // something with the spans we're generating.
-    provider = new NodeTracerProvider();
     processor = new SimpleSpanProcessor(exporter);
-    provider.addSpanProcessor(processor);
+    provider = new NodeTracerProvider({
+      spanProcessors: [processor],
+    });
     provider.register();
   });
 

From b31f20f5ade93be7d3a77444e91efa38cf241d56 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 13 Aug 2026 14:41:08 -0400
Subject: [PATCH 22/29] tests: fix test failures in avro and otel samples

---
 handwritten/pubsub/src/telemetry-tracing.ts         | 4 +---
 handwritten/pubsub/system-test/avro-samples.test.ts | 8 +++++++-
 handwritten/pubsub/system-test/otel-samples.test.ts | 2 ++
 3 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/handwritten/pubsub/src/telemetry-tracing.ts b/handwritten/pubsub/src/telemetry-tracing.ts
index 54a516d1b6b4..77bc15883487 100644
--- a/handwritten/pubsub/src/telemetry-tracing.ts
+++ b/handwritten/pubsub/src/telemetry-tracing.ts
@@ -756,9 +756,7 @@ export function injectSpan(span: Span, message: MessageWithAttributes): void {
     return;
   }
 
-  if (!message.attributes) {
-    message.attributes = {};
-  }
+  message.attributes = Object.assign({}, message.attributes);
 
   if (message.attributes[modernAttributeName]) {
     console.warn(
diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
index 600e92f45d09..06bf00574a00 100644
--- a/handwritten/pubsub/system-test/avro-samples.test.ts
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -109,7 +109,13 @@ describe('Avro Samples System Tests', () => {
 
     const topicName = resources.generateName(`topic-rev`);
     const subName = resources.generateName(`sub-rev`);
-    const [topic] = await pubsub.createTopic(topicName);
+    const [topic] = await pubsub.createTopic({
+      name: topicName,
+      schemaSettings: {
+        schema: await pubsub.schema(schemaId).getName(),
+        encoding: 'BINARY',
+      },
+    });
     const [subscription] = await pubsub.createSubscription(topicName, subName);
 
     const type = avro.parse(definition);
diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
index beb518a9f875..8fe436f93171 100644
--- a/handwritten/pubsub/system-test/otel-samples.test.ts
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 import {Message, PubSub, Subscription, Topic} from '../src';
+import * as tracing from '../src/telemetry-tracing';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
@@ -50,6 +51,7 @@ describe('OpenTelemetry Samples System Tests', () => {
 
   after(async () => {
     // Don't interfere with other tests.
+    tracing.setGloballyEnabled(false);
     provider.shutdown();
 
     const [subscriptions] = await pubsub.getSubscriptions();

From 179c0b54c2cd8133ff8c8593d92a47cefebd02af Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 13 Aug 2026 15:07:55 -0400
Subject: [PATCH 23/29] fix: revert logging changes, the other ones were
 correct

---
 handwritten/pubsub/src/lease-manager.ts            | 12 ++++++------
 handwritten/pubsub/src/logs.ts                     |  5 ++---
 handwritten/pubsub/src/message-queues.ts           |  6 +++---
 handwritten/pubsub/src/message-stream.ts           |  8 +++++---
 handwritten/pubsub/src/publisher/message-queues.ts |  6 +++---
 handwritten/pubsub/src/subscriber.ts               | 10 +++++-----
 6 files changed, 24 insertions(+), 23 deletions(-)

diff --git a/handwritten/pubsub/src/lease-manager.ts b/handwritten/pubsub/src/lease-manager.ts
index aa10d117b352..447c4d7e6e8c 100644
--- a/handwritten/pubsub/src/lease-manager.ts
+++ b/handwritten/pubsub/src/lease-manager.ts
@@ -20,18 +20,18 @@ import {AckError, Message, Subscriber} from './subscriber';
 import {defaultOptions} from './default-options';
 import {Duration} from './temporal';
 import {DebugMessage} from './debug';
-import {logs as baseLogs, Loggers} from './logs';
+import {logs as baseLogs, LoggingFunction} from './logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs: Loggers = {
-  callbackDelivery: baseLogs.pubsub.sublog('callback-delivery'),
-  callbackExceptions: baseLogs.pubsub.sublog('callback-exceptions'),
-  expiry: baseLogs.pubsub.sublog('expiry'),
-  subscriberFlowControl: baseLogs.pubsub.sublog('subscriber-flow-control'),
+export const logs = {
+  callbackDelivery: baseLogs.pubsub.sublog('callback-delivery') as LoggingFunction,
+  callbackExceptions: baseLogs.pubsub.sublog('callback-exceptions') as LoggingFunction,
+  expiry: baseLogs.pubsub.sublog('expiry') as LoggingFunction,
+  subscriberFlowControl: baseLogs.pubsub.sublog('subscriber-flow-control') as LoggingFunction,
 };
 
 export interface FlowControlOptions {
diff --git a/handwritten/pubsub/src/logs.ts b/handwritten/pubsub/src/logs.ts
index e3fa5c3841e7..dadddcf60bee 100644
--- a/handwritten/pubsub/src/logs.ts
+++ b/handwritten/pubsub/src/logs.ts
@@ -15,13 +15,12 @@
 import {loggingUtils} from 'google-gax';
 
 export type LoggingFunction = loggingUtils.AdhocDebugLogFunction;
-export type Loggers = Record<string, LoggingFunction>;
 
 /**
  * Base logger. Other loggers will derive from this one.
  *
  * @private
  */
-export const logs: Loggers = {
-  pubsub: loggingUtils.log('pubsub'),
+export const logs = {
+  pubsub: loggingUtils.log('pubsub') as LoggingFunction,
 };
diff --git a/handwritten/pubsub/src/message-queues.ts b/handwritten/pubsub/src/message-queues.ts
index 6979e1543011..bdc10ecb740d 100644
--- a/handwritten/pubsub/src/message-queues.ts
+++ b/handwritten/pubsub/src/message-queues.ts
@@ -34,15 +34,15 @@ import {Duration} from './temporal';
 import {addToBucket} from './util';
 import {DebugMessage} from './debug';
 import * as tracing from './telemetry-tracing';
-import {logs as baseLogs, Loggers} from './logs';
+import {logs as baseLogs, LoggingFunction} from './logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs: Loggers = {
-  ackBatch: baseLogs.pubsub.sublog('ack-batch'),
+export const logs = {
+  ackBatch: baseLogs.pubsub.sublog('ack-batch') as LoggingFunction,
 };
 
 export interface ReducedMessage {
diff --git a/handwritten/pubsub/src/message-stream.ts b/handwritten/pubsub/src/message-stream.ts
index 432e50ab0e68..9924f91032a4 100644
--- a/handwritten/pubsub/src/message-stream.ts
+++ b/handwritten/pubsub/src/message-stream.ts
@@ -27,15 +27,17 @@ import {defaultOptions} from './default-options';
 import {Duration} from './temporal';
 import {ExponentialRetry} from './exponential-retry';
 import {DebugMessage} from './debug';
-import {logs as baseLogs, Loggers} from './logs';
+import {logs as baseLogs, LoggingFunction} from './logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs: Loggers = {
-  subscriberStreams: baseLogs.pubsub.sublog('subscriber-streams'),
+export const logs = {
+  subscriberStreams: baseLogs.pubsub.sublog(
+    'subscriber-streams',
+  ) as LoggingFunction,
 };
 
 /*!
diff --git a/handwritten/pubsub/src/publisher/message-queues.ts b/handwritten/pubsub/src/publisher/message-queues.ts
index 34c0a99d2963..0227e585cecd 100644
--- a/handwritten/pubsub/src/publisher/message-queues.ts
+++ b/handwritten/pubsub/src/publisher/message-queues.ts
@@ -24,15 +24,15 @@ import {google} from '../../protos/protos';
 import * as tracing from '../telemetry-tracing';
 import {filterMessage} from './pubsub-message';
 import {promisify} from 'util';
-import {logs as baseLogs, Loggers} from '../logs';
+import {logs as baseLogs, LoggingFunction} from '../logs';
 
 /**
  * Loggers. Exported for unit tests.
  *
  * @private
  */
-export const logs: Loggers = {
-  publishBatch: baseLogs.pubsub.sublog('publish-batch'),
+export const logs = {
+  publishBatch: baseLogs.pubsub.sublog('publish-batch') as LoggingFunction,
 };
 
 /**
diff --git a/handwritten/pubsub/src/subscriber.ts b/handwritten/pubsub/src/subscriber.ts
index ec4803812a19..780f94b7342c 100644
--- a/handwritten/pubsub/src/subscriber.ts
+++ b/handwritten/pubsub/src/subscriber.ts
@@ -31,7 +31,7 @@ import {Duration, atMost as durationAtMost} from './temporal';
 import {EventEmitter} from 'events';
 
 import {awaitWithTimeout} from './util';
-import {logs as baseLogs, Loggers} from './logs';
+import {logs as baseLogs, LoggingFunction} from './logs';
 
 export {StatusError} from './message-stream';
 
@@ -43,10 +43,10 @@ type SubscriberClient = v1.SubscriberClient;
  *
  * @private
  */
-export const logs: Loggers = {
-  slowAck: baseLogs.pubsub.sublog('slow-ack'),
-  ackNack: baseLogs.pubsub.sublog('ack-nack'),
-  debug: baseLogs.pubsub.sublog('debug'),
+export const logs = {
+  slowAck: baseLogs.pubsub.sublog('slow-ack') as LoggingFunction,
+  ackNack: baseLogs.pubsub.sublog('ack-nack') as LoggingFunction,
+  debug: baseLogs.pubsub.sublog('debug') as LoggingFunction,
 };
 
 export type PullResponse = google.pubsub.v1.IStreamingPullResponse;

From 2828510aeaeb5033727eb75021d12070f2791dd1 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 13 Aug 2026 15:20:54 -0400
Subject: [PATCH 24/29] chore: poke CI


From f8aed4589cad01542e325e9fec9db142b7b88a4c Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 13 Aug 2026 17:17:56 -0400
Subject: [PATCH 25/29] fix: remove message listener on timeout in avro-samples
 test

---
 handwritten/pubsub/system-test/avro-samples.test.ts | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
index 06bf00574a00..1a681f83823d 100644
--- a/handwritten/pubsub/system-test/avro-samples.test.ts
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -78,12 +78,16 @@ describe('Avro Samples System Tests', () => {
     assert.ok(messageId);
 
     const message = await new Promise<Message>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout waiting for Avro record')), 15000);
-      subscription.once('message', (m: Message) => {
+      const messageHandler = (m: Message) => {
         clearTimeout(timeout);
         m.ack();
         resolve(m);
-      });
+      };
+      const timeout = setTimeout(() => {
+        subscription.removeListener('message', messageHandler);
+        reject(new Error('Timeout waiting for Avro record'));
+      }, 15000);
+      subscription.once('message', messageHandler);
     });
 
     const schemaMetadata = Schema.metadataFromMessage(message.attributes);

From dc86d863e3e3265eb3210bbbb9eb721e783387ff Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 13 Aug 2026 17:24:27 -0400
Subject: [PATCH 26/29] tests: remove chai

---
 handwritten/pubsub/package.json                      | 2 --
 handwritten/pubsub/system-test/testResources.test.ts | 2 +-
 handwritten/pubsub/test/testResources.test.ts        | 2 +-
 3 files changed, 2 insertions(+), 4 deletions(-)

diff --git a/handwritten/pubsub/package.json b/handwritten/pubsub/package.json
index 5d0fc5f0936c..0ac4fc9adcc7 100644
--- a/handwritten/pubsub/package.json
+++ b/handwritten/pubsub/package.json
@@ -73,7 +73,6 @@
     "@grpc/proto-loader": "^0.8.0",
     "@opentelemetry/sdk-trace-base": "^2.8.0",
     "@opentelemetry/sdk-trace-node": "^2.8.0",
-    "@types/chai": "^5.2.3",
     "@types/duplexify": "^3.6.4",
     "@types/extend": "^3.0.4",
     "@types/lodash.snakecase": "^4.1.9",
@@ -86,7 +85,6 @@
     "@types/tmp": "^0.2.6",
     "avro-js": "^1.12.1",
     "c8": "^10.1.3",
-    "chai": "^6.2.2",
     "codecov": "^3.8.3",
     "execa": "~5.1.0",
     "gapic-tools": "^2.0.0",
diff --git a/handwritten/pubsub/system-test/testResources.test.ts b/handwritten/pubsub/system-test/testResources.test.ts
index 618efbede3ea..b63e10280508 100644
--- a/handwritten/pubsub/system-test/testResources.test.ts
+++ b/handwritten/pubsub/system-test/testResources.test.ts
@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import {assert} from 'chai';
 import {describe, it, beforeEach} from 'mocha';
 import {TestResources} from './testResources';
+import * as assert from 'node:assert';
 
 describe('testResources (unit)', () => {
   const fixedId = 'fixed';
diff --git a/handwritten/pubsub/test/testResources.test.ts b/handwritten/pubsub/test/testResources.test.ts
index b1cf3a67371d..b759389df23a 100644
--- a/handwritten/pubsub/test/testResources.test.ts
+++ b/handwritten/pubsub/test/testResources.test.ts
@@ -14,7 +14,7 @@
 
 import {describe, it, beforeEach} from 'mocha';
 import {TestResources} from './testResources';
-import * as assert from 'assert';
+import * as assert from 'node:assert';
 
 describe('testResources (unit)', () => {
   const fixedId = 'fixed';

From a1e67d8ffadd35e8446659228cc5c66f3b458d3a Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Thu, 13 Aug 2026 17:37:21 -0400
Subject: [PATCH 27/29] chore: remove unnecessary no-op changes from merging

---
 handwritten/pubsub/src/message-stream.ts | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/handwritten/pubsub/src/message-stream.ts b/handwritten/pubsub/src/message-stream.ts
index 9924f91032a4..2b71542b6ddb 100644
--- a/handwritten/pubsub/src/message-stream.ts
+++ b/handwritten/pubsub/src/message-stream.ts
@@ -18,7 +18,6 @@ import {promisify} from '@google-cloud/promisify';
 import {ClientStub, GoogleError, grpc} from 'google-gax';
 import * as isStreamEnded from 'is-stream-ended';
 import {PassThrough} from 'stream';
-import {randomUUID} from 'crypto';
 
 import {PullRetry} from './pull-retry';
 import {Subscriber} from './subscriber';
@@ -27,6 +26,7 @@ import {defaultOptions} from './default-options';
 import {Duration} from './temporal';
 import {ExponentialRetry} from './exponential-retry';
 import {DebugMessage} from './debug';
+import {randomUUID} from 'crypto';
 import {logs as baseLogs, LoggingFunction} from './logs';
 
 /**
@@ -35,9 +35,7 @@ import {logs as baseLogs, LoggingFunction} from './logs';
  * @private
  */
 export const logs = {
-  subscriberStreams: baseLogs.pubsub.sublog(
-    'subscriber-streams',
-  ) as LoggingFunction,
+  subscriberStreams: baseLogs.pubsub.sublog('subscriber-streams') as LoggingFunction,
 };
 
 /*!

From 4b011c2276de3406618aa9b206fd332ee4a2c5ba Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 28 Aug 2026 16:04:12 -0400
Subject: [PATCH 28/29] tests: factor out subscription waiting to catch and
 remove handlers

---
 .../pubsub/system-test/avro-samples.test.ts   |  25 +--
 .../system-test/batch-flow-samples.test.ts    |  33 +---
 handwritten/pubsub/system-test/common.ts      | 173 ++++++++++++++++++
 .../pubsub/system-test/otel-samples.test.ts   |  11 +-
 .../system-test/protobuf-samples.test.ts      |  13 +-
 handwritten/pubsub/system-test/pubsub.ts      | 148 ++++++---------
 6 files changed, 252 insertions(+), 151 deletions(-)
 create mode 100644 handwritten/pubsub/system-test/common.ts

diff --git a/handwritten/pubsub/system-test/avro-samples.test.ts b/handwritten/pubsub/system-test/avro-samples.test.ts
index 1a681f83823d..49cfadca3881 100644
--- a/handwritten/pubsub/system-test/avro-samples.test.ts
+++ b/handwritten/pubsub/system-test/avro-samples.test.ts
@@ -18,6 +18,7 @@ import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
 import * as avro from 'avro-js';
 import * as fs from 'fs';
+import {waitForMessage} from './common';
 
 describe('Avro Samples System Tests', () => {
   const pubsub = new PubSub();
@@ -77,17 +78,9 @@ describe('Avro Samples System Tests', () => {
     const messageId = await topic.publishMessage({data: dataBuffer});
     assert.ok(messageId);
 
-    const message = await new Promise<Message>((resolve, reject) => {
-      const messageHandler = (m: Message) => {
-        clearTimeout(timeout);
-        m.ack();
-        resolve(m);
-      };
-      const timeout = setTimeout(() => {
-        subscription.removeListener('message', messageHandler);
-        reject(new Error('Timeout waiting for Avro record'));
-      }, 15000);
-      subscription.once('message', messageHandler);
+    const message = await waitForMessage(subscription, {
+      timeoutMs: 15000,
+      timeoutErrorMessage: 'Timeout waiting for Avro record',
     });
 
     const schemaMetadata = Schema.metadataFromMessage(message.attributes);
@@ -131,13 +124,9 @@ describe('Avro Samples System Tests', () => {
     const dataBuffer = type.toBuffer(province);
     await topic.publishMessage({data: dataBuffer});
 
-    const message = await new Promise<Message>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout waiting for Avro revision')), 15000);
-      subscription.once('message', (m: Message) => {
-        clearTimeout(timeout);
-        m.ack();
-        resolve(m);
-      });
+    const message = await waitForMessage(subscription, {
+      timeoutMs: 15000,
+      timeoutErrorMessage: 'Timeout waiting for Avro revision',
     });
 
     const schemaMetadata = Schema.metadataFromMessage(message.attributes);
diff --git a/handwritten/pubsub/system-test/batch-flow-samples.test.ts b/handwritten/pubsub/system-test/batch-flow-samples.test.ts
index 5b908460c916..08b72744fb34 100644
--- a/handwritten/pubsub/system-test/batch-flow-samples.test.ts
+++ b/handwritten/pubsub/system-test/batch-flow-samples.test.ts
@@ -16,6 +16,7 @@ import {Message, PubSub, PublishOptions} from '../src';
 import * as assert from 'assert';
 import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
+import {waitForMessages} from './common';
 
 describe('Batch and Flow Control Samples System Tests', () => {
   const pubsub = new PubSub();
@@ -61,18 +62,10 @@ describe('Batch and Flow Control Samples System Tests', () => {
     const messageIds = await Promise.all(promises);
     assert.strictEqual(messageIds.length, 10);
 
-    const messages: Message[] = [];
-    await new Promise<void>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout waiting for batched messages')), 15000);
-      subscription.on('message', (m: Message) => {
-        m.ack();
-        messages.push(m);
-        if (messages.length === 10) {
-          clearTimeout(timeout);
-          subscription.removeAllListeners('message');
-          resolve();
-        }
-      });
+    const messages = await waitForMessages(subscription, {
+      count: 10,
+      timeoutMs: 15000,
+      timeoutErrorMessage: 'Timeout waiting for batched messages',
     });
 
     assert.strictEqual(messages.length, 10);
@@ -105,18 +98,10 @@ describe('Batch and Flow Control Samples System Tests', () => {
     const messageIds = await flow.all();
     assert.strictEqual(messageIds.length, 10);
 
-    const messages: Message[] = [];
-    await new Promise<void>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout waiting for flow control messages')), 15000);
-      subscription.on('message', (m: Message) => {
-        m.ack();
-        messages.push(m);
-        if (messages.length === 10) {
-          clearTimeout(timeout);
-          subscription.removeAllListeners('message');
-          resolve();
-        }
-      });
+    const messages = await waitForMessages(subscription, {
+      count: 10,
+      timeoutMs: 15000,
+      timeoutErrorMessage: 'Timeout waiting for flow control messages',
     });
 
     assert.strictEqual(messages.length, 10);
diff --git a/handwritten/pubsub/system-test/common.ts b/handwritten/pubsub/system-test/common.ts
new file mode 100644
index 000000000000..dfad1c665952
--- /dev/null
+++ b/handwritten/pubsub/system-test/common.ts
@@ -0,0 +1,173 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import {Message, Subscription} from '../src';
+
+export interface WaitForMessagesOptions {
+  /** Number of messages to collect before resolving. Defaults to 1. */
+  count?: number;
+  /** Timeout in milliseconds. Defaults to 15,000ms. */
+  timeoutMs?: number;
+  /** Whether to automatically ack received messages. Defaults to true. */
+  autoAck?: boolean;
+  /** Optional filter predicate. Only matching messages increment count and get collected. */
+  filter?: (message: Message) => boolean;
+  /** Optional per-message inspection hook called for matching messages before ack/resolve. */
+  onMessage?: (message: Message) => void | Promise<void>;
+  /** Error message prefix if timeout occurs. */
+  timeoutErrorMessage?: string;
+  /** Whether to call subscription.close() after receiving all expected messages. Defaults to false. */
+  closeWhenDone?: boolean;
+}
+
+/**
+ * Listens for messages on a subscription until the expected count is reached,
+ * a timeout occurs, or an error is emitted.
+ *
+ * Guarantees that all attached listeners ('message', 'error') and any active timeouts
+ * are removed on success, error, or timeout to avoid leaking resources.
+ */
+export async function waitForMessages(
+  subscription: Subscription,
+  options: WaitForMessagesOptions = {}
+): Promise<Message[]> {
+  const {
+    count = 1,
+    timeoutMs = 15000,
+    autoAck = true,
+    filter = () => true,
+    onMessage,
+    timeoutErrorMessage = 'Timeout waiting for messages',
+    closeWhenDone = false,
+  } = options;
+
+  return new Promise<Message[]>((resolve, reject) => {
+    const received: Message[] = [];
+    let timeoutId: NodeJS.Timeout | undefined;
+
+    let cleanedUp = false;
+    const cleanup = () => {
+      if (cleanedUp) return;
+      cleanedUp = true;
+
+      if (timeoutId) {
+        clearTimeout(timeoutId);
+        timeoutId = undefined;
+      }
+      subscription.removeListener('message', messageHandler);
+      subscription.removeListener('error', errorHandler);
+
+      if (closeWhenDone) {
+        void subscription.close();
+      }
+    };
+
+    const messageHandler = async (msg: Message) => {
+      try {
+        if (!filter(msg)) {
+          return;
+        }
+
+        if (onMessage) {
+          await onMessage(msg);
+        }
+
+        if (autoAck) {
+          msg.ack();
+        }
+
+        received.push(msg);
+
+        if (received.length >= count) {
+          cleanup();
+          resolve(received);
+        }
+      } catch (err) {
+        cleanup();
+        reject(err);
+      }
+    };
+
+    const errorHandler = (err: Error) => {
+      cleanup();
+      reject(err);
+    };
+
+    timeoutId = setTimeout(() => {
+      cleanup();
+      reject(new Error(`${timeoutErrorMessage} (${timeoutMs}ms)`));
+    }, timeoutMs);
+
+    subscription.on('error', errorHandler);
+    subscription.on('message', messageHandler);
+  });
+}
+
+/**
+ * Listens for a single message on a subscription with guaranteed cleanup.
+ */
+export async function waitForMessage(
+  subscription: Subscription,
+  options: Omit<WaitForMessagesOptions, 'count'> = {}
+): Promise<Message> {
+  const messages = await waitForMessages(subscription, {...options, count: 1});
+  return messages[0];
+}
+
+/**
+ * Tracks event listeners attached to a Subscription and removes them on cleanup.
+ */
+export class SubscriptionScope {
+  private cleanups: Array<() => void> = [];
+
+  constructor(public readonly subscription: Subscription) {}
+
+  /** Attaches an event listener and tracks it for automatic teardown */
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  on(event: string, listener: (...args: any[]) => void): this {
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    this.subscription.on(event as any, listener as any);
+    this.cleanups.push(() => {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      this.subscription.removeListener(event as any, listener as any);
+    });
+    return this;
+  }
+
+  /** Cleans up all listeners registered through this scope */
+  cleanup(): void {
+    while (this.cleanups.length > 0) {
+      const fn = this.cleanups.pop();
+      try {
+        fn?.();
+      } catch {}
+    }
+  }
+}
+
+/**
+ * Runs an asynchronous test block with a managed subscription scope, guaranteeing
+ * that all registered listeners are removed in a finally block.
+ */
+export async function withSubscriptionScope<T>(
+  subscription: Subscription,
+  fn: (scope: SubscriptionScope) => Promise<T>
+): Promise<T> {
+  const scope = new SubscriptionScope(subscription);
+  try {
+    return await fn(scope);
+  } finally {
+    scope.cleanup();
+  }
+}
diff --git a/handwritten/pubsub/system-test/otel-samples.test.ts b/handwritten/pubsub/system-test/otel-samples.test.ts
index 8fe436f93171..6bd8a05e4709 100644
--- a/handwritten/pubsub/system-test/otel-samples.test.ts
+++ b/handwritten/pubsub/system-test/otel-samples.test.ts
@@ -19,6 +19,7 @@ import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
 import {NodeTracerProvider} from '@opentelemetry/sdk-trace-node';
 import {SimpleSpanProcessor, InMemorySpanExporter} from '@opentelemetry/sdk-trace-base';
+import {waitForMessage} from './common';
 
 describe('OpenTelemetry Samples System Tests', () => {
   const pubsub = new PubSub({enableOpenTelemetryTracing: true});
@@ -72,13 +73,9 @@ describe('OpenTelemetry Samples System Tests', () => {
     const messageId = await topic.publishMessage({data: dataBuffer});
     assert.ok(messageId);
 
-    const message = await new Promise<Message>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout waiting for OTel message')), 15000);
-      subscription.once('message', (m: Message) => {
-        clearTimeout(timeout);
-        m.ack();
-        resolve(m);
-      });
+    const message = await waitForMessage(subscription, {
+      timeoutMs: 15000,
+      timeoutErrorMessage: 'Timeout waiting for OTel message',
     });
 
     assert.strictEqual(message.data.toString(), data);
diff --git a/handwritten/pubsub/system-test/protobuf-samples.test.ts b/handwritten/pubsub/system-test/protobuf-samples.test.ts
index 7f058cdd677d..9b0b46f4e230 100644
--- a/handwritten/pubsub/system-test/protobuf-samples.test.ts
+++ b/handwritten/pubsub/system-test/protobuf-samples.test.ts
@@ -18,6 +18,7 @@ import {describe, it, after, before} from 'mocha';
 import {TestResources} from './testResources';
 import * as protobuf from 'protobufjs';
 import * as fs from 'fs';
+import {waitForMessage} from './common';
 
 describe('Protobuf Samples System Tests', () => {
   const pubsub = new PubSub();
@@ -77,15 +78,9 @@ describe('Protobuf Samples System Tests', () => {
     const messageId = await topic.publishMessage({data: dataBuffer});
     assert.ok(messageId);
 
-    let received!: Message;
-    await new Promise<Message>((resolve, reject) => {
-      const timeout = setTimeout(() => reject(new Error('Timeout waiting for Proto message')), 15000);
-      subscription.once('message', (m: Message) => {
-        m.ack();
-        received = m;
-        clearTimeout(timeout);
-        resolve(m);
-      });
+    const received = await waitForMessage(subscription, {
+      timeoutMs: 15000,
+      timeoutErrorMessage: 'Timeout waiting for Proto message',
     });
 
     const schemaMetadata = Schema.metadataFromMessage(received.attributes);
diff --git a/handwritten/pubsub/system-test/pubsub.ts b/handwritten/pubsub/system-test/pubsub.ts
index c1b574ee7914..9b63cfd72ebc 100644
--- a/handwritten/pubsub/system-test/pubsub.ts
+++ b/handwritten/pubsub/system-test/pubsub.ts
@@ -36,6 +36,7 @@ import {
 import {MessageOptions} from '../src/topic';
 import {TestResources} from '../test/testResources';
 import {GoogleError} from 'google-gax';
+import {waitForMessage, waitForMessages, withSubscriptionScope} from './common';
 
 const pubsub = new PubSub();
 
@@ -122,10 +123,7 @@ describe('pubsub', () => {
     for (let i = 0; i < 6; i++) {
       await topic.topic.publishMessage(message);
     }
-    return new Promise<Message>((resolve, reject) => {
-      sub.sub.on('error', reject);
-      sub.sub.once('message', resolve);
-    });
+    return waitForMessage(sub.sub);
   }
 
   before(async () => {
@@ -230,17 +228,9 @@ describe('pubsub', () => {
       console.log(`Message ${messageId} published.`);
       
       // --- From test (topics.test.ts) ---
-      const message = await new Promise<Message>((resolve, reject) => {
-        const timeout = setTimeout(() => {
-          subscription.removeListener('message', messageHandler);
-          reject(new Error('Timeout'));
-        }, 10000);
-        function messageHandler(m: Message) {
-          clearTimeout(timeout);
-          m.ack();
-          resolve(m);
-        }
-        subscription.once('message', messageHandler);
+      const message = await waitForMessage(subscription, {
+        timeoutMs: 10000,
+        timeoutErrorMessage: 'Timeout',
       });
       
       assert.strictEqual(message.data.toString(), data);
@@ -555,107 +545,79 @@ describe('pubsub', () => {
       const testTopic = await generateTopic('dne-sub');
       const subscription = testTopic.topic.subscription(generateSubName('dne-sub'));
 
-      await new Promise((res, rej) => {
-        subscription.on('error', (err: {code: number}) => {
-          assert.strictEqual(err.code, 5);
-          subscription.close(res);
-        });
+      await withSubscriptionScope(subscription, scope => {
+        return new Promise<void>((res, rej) => {
+          scope.on('error', async (err: {code: number}) => {
+            try {
+              assert.strictEqual(err.code, 5);
+              await subscription.close();
+              res();
+            } catch (e) {
+              rej(e);
+            }
+          });
 
-        subscription.on('message', () => {
-          rej(new Error('Should not have been called.'));
+          scope.on('message', () => {
+            rej(new Error('Should not have been called.'));
+          });
         });
       });
     });
 
     it('should receive the published messages', async () => {
       const pop = await subPop('recv', 1);
-      let messageCount = 0;
       const subscription = pop.subs[0];
 
-      await new Promise((res, rej) => {
-        subscription.on('error', rej);
-
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        subscription.on('message', message => {
-          assert.deepStrictEqual(message.data, Buffer.from('hello'));
-          message.ack();
-
-          if (++messageCount === 10) {
-            subscription.close(res);
-          }
-        });
+      const messages = await waitForMessages(subscription, {
+        count: 10,
+        filter: msg => {
+          assert.deepStrictEqual(msg.data, Buffer.from('hello'));
+          return true;
+        },
+        closeWhenDone: true,
       });
+      assert.strictEqual(messages.length, 10);
     });
 
     it('should ack the message', async () => {
       const pop = await subPop('ack', 1);
       const subscription = pop.subs[0];
 
-      await new Promise((res, rej) => {
-        let finished = false;
-        subscription.on('error', () => {
-          if (!finished) {
-            finished = true;
-            subscription.close(rej);
-          }
-        });
-        subscription.on('message', ack);
-
-        function ack(message: Message) {
-          if (!finished) {
-            finished = true;
-            message.ack();
-            subscription.close(res);
-          }
-        }
+      const message = await waitForMessage(subscription, {
+        autoAck: true,
+        closeWhenDone: true,
       });
+      assert.ok(message);
     });
 
     it('should nack the message', async () => {
       const pop = await subPop('nack', 1);
       const subscription = pop.subs[0];
 
-      await new Promise((res, rej) => {
-        let finished = false;
-        subscription.on('error', () => {
-          if (!finished) {
-            finished = true;
-            subscription.close(rej);
-          }
-        });
-        subscription.on('message', nack);
-
-        function nack(message: Message) {
-          if (!finished) {
-            finished = true;
-            message.nack();
-            subscription.close(res);
-          }
-        }
+      const message = await waitForMessage(subscription, {
+        autoAck: false,
+        onMessage: msg => {
+          msg.nack();
+        },
+        closeWhenDone: true,
       });
+      assert.ok(message);
     });
 
     it('should respect flow control limits', async () => {
       const maxMessages = 3;
-      let messageCount = 0;
 
       const pop = await subPop('fcl', 1);
       const subscription = pop.topic.subscription(pop.testSubs[0].name, {
         flowControl: {maxMessages, allowExcessMessages: false},
       });
 
-      await new Promise((res, rej) => {
-        subscription.on('error', rej);
-        subscription.on('message', onMessage);
-
-        function onMessage() {
-          if (++messageCount < maxMessages) {
-            return;
-          }
-
-          subscription.close(res);
-        }
+      const messages = await waitForMessages(subscription, {
+        count: maxMessages,
+        autoAck: false,
+        closeWhenDone: true,
       });
+      assert.strictEqual(messages.length, maxMessages);
     });
 
     it('should send and receive large messages', async () => {
@@ -664,16 +626,12 @@ describe('pubsub', () => {
       const data = crypto.randomBytes(9000000); // 9mb
 
       const messageId = await pop.topic.publishMessage({data});
-      await new Promise((res, rej) => {
-        subscription.on('error', rej).on('message', (message: Message) => {
-          if (message.id !== messageId) {
-            return;
-          }
-
-          assert.deepStrictEqual(data, message.data);
-          subscription.close(res);
-        });
+      const message = await waitForMessage(subscription, {
+        filter: msg => msg.id === messageId,
+        autoAck: true,
+        closeWhenDone: true,
       });
+      assert.deepStrictEqual(data, message.data);
     });
 
     it('should detach subscriptions', async () => {
@@ -881,9 +839,13 @@ describe('pubsub', () => {
       type WorkCallback = (arg: Message, resolve: Function) => void;
       function makeMessagePromise(subscription: Subscription, workCallback: WorkCallback): Promise<void> {
         return new Promise(resolve => {
-          subscription.on('message', (arg: Message) => {
-            workCallback(arg, resolve);
-          });
+          const messageHandler = (arg: Message) => {
+            workCallback(arg, () => {
+              subscription.removeListener('message', messageHandler);
+              resolve();
+            });
+          };
+          subscription.on('message', messageHandler);
         });
       }
 

From eee9fc93fabc379417bcff4ed1682a2830f27753 Mon Sep 17 00:00:00 2001
From: feywind <57276408+feywind@users.noreply.github.com>
Date: Fri, 28 Aug 2026 16:27:51 -0400
Subject: [PATCH 29/29] chore: fix new linter errors

---
 handwritten/pubsub/test/message-queues.ts | 12 +++--------
 handwritten/pubsub/test/message-stream.ts |  2 +-
 handwritten/pubsub/test/subscriber.ts     | 26 ++++++++---------------
 3 files changed, 13 insertions(+), 27 deletions(-)

diff --git a/handwritten/pubsub/test/message-queues.ts b/handwritten/pubsub/test/message-queues.ts
index d7b63f94d5ea..1b9738023f35 100644
--- a/handwritten/pubsub/test/message-queues.ts
+++ b/handwritten/pubsub/test/message-queues.ts
@@ -17,7 +17,7 @@
 import * as assert from 'assert';
 import {describe, it, before, beforeEach, afterEach} from 'mocha';
 import {EventEmitter} from 'events';
-import {CallOptions, GoogleError, loggingUtils, Status} from 'google-gax';
+import {CallOptions, GoogleError, Status} from 'google-gax';
 import * as sinon from 'sinon';
 import defer = require('p-defer');
 import * as crypto from 'node:crypto';
@@ -418,10 +418,7 @@ describe('MessageQueues', () => {
       fakeLog.remove();
 
       assert.strictEqual(fakeLog.called, true);
-      assert.strictEqual(
-        fakeLog.fields!.severity,
-        'INFO',
-      );
+      assert.strictEqual(fakeLog.fields!.severity, 'INFO');
       assert.strictEqual(fakeLog.args![1] as string, 'logtest');
     });
 
@@ -679,10 +676,7 @@ describe('MessageQueues', () => {
       fakeLog.remove();
 
       assert.strictEqual(fakeLog.called, true);
-      assert.strictEqual(
-        fakeLog.fields!.severity,
-        'INFO',
-      );
+      assert.strictEqual(fakeLog.fields!.severity, 'INFO');
       assert.strictEqual(fakeLog.args![1] as string, 'logtest');
     });
 
diff --git a/handwritten/pubsub/test/message-stream.ts b/handwritten/pubsub/test/message-stream.ts
index a35b2b207c5e..3836b2620989 100644
--- a/handwritten/pubsub/test/message-stream.ts
+++ b/handwritten/pubsub/test/message-stream.ts
@@ -534,7 +534,7 @@ describe('MessageStream', () => {
 
         assert.strictEqual(spy.callCount, 5);
         const {args} = spy.firstCall;
-        const request = args[0] as any;
+        const request = args[0] as {protocolVersion?: string | number};
 
         assert.strictEqual(String(request.protocolVersion), '1');
 
diff --git a/handwritten/pubsub/test/subscriber.ts b/handwritten/pubsub/test/subscriber.ts
index 48d4a367d30a..3cec7ccd06af 100644
--- a/handwritten/pubsub/test/subscriber.ts
+++ b/handwritten/pubsub/test/subscriber.ts
@@ -37,7 +37,6 @@ import {SpanKind} from '@opentelemetry/api';
 import {Duration} from '../src';
 import * as tracing from '../src/telemetry-tracing';
 import {FakeLog, TestUtils} from './test-utils';
-import {loggingUtils} from 'google-gax';
 
 type PullResponse = google.pubsub.v1.IStreamingPullResponse;
 
@@ -450,10 +449,7 @@ describe('Subscriber', () => {
       await subscriber.ack(message);
 
       assert.strictEqual(fakeLog.called, true);
-      assert.strictEqual(
-        fakeLog.fields!.severity,
-        'INFO',
-      );
+      assert.strictEqual(fakeLog.fields!.severity, 'INFO');
       assert.strictEqual(fakeLog.args![1], message.id);
     });
 
@@ -468,10 +464,7 @@ describe('Subscriber', () => {
       await subscriber.ack(message);
 
       assert.strictEqual(fakeLog.called, true);
-      assert.strictEqual(
-        fakeLog.fields!.severity,
-        'INFO',
-      );
+      assert.strictEqual(fakeLog.fields!.severity, 'INFO');
       assert.strictEqual(fakeLog.args![1], message.id);
     });
 
@@ -822,9 +815,10 @@ describe('Subscriber', () => {
             timeout: Duration.from({milliseconds: 100}),
           },
         });
-        const prom = subscriber.close().then(() => {
+        const prom = (async () => {
+          await subscriber.close();
           closed = true;
-        });
+        })();
 
         // Advance time past the timeout
         clock.tick(200);
@@ -854,9 +848,10 @@ describe('Subscriber', () => {
             timeout: Duration.from({milliseconds: 100}),
           },
         });
-        const prom = subscriber.close().then(() => {
+        const prom = (async () => {
+          await subscriber.close();
           closed = true;
-        });
+        })();
 
         // Resolve drains quickly
         ackDrainDeferred.resolve();
@@ -944,10 +939,7 @@ describe('Subscriber', () => {
       await subscriber.nack(message);
 
       assert.strictEqual(fakeLog.called, true);
-      assert.strictEqual(
-        fakeLog.fields!.severity,
-        'INFO',
-      );
+      assert.strictEqual(fakeLog.fields!.severity, 'INFO');
       assert.strictEqual(fakeLog.args![1], message.id);
     });