Add IP routing rules and fix gRPC buffer retention

test/server/grpc-stream-memory.test.js

@@ -0,0 +1,84 @@
+import assert from 'node:assert/strict';
+import { spawnSync } from 'node:child_process';
+import path from 'node:path';
+import test from 'node:test';
+
+test('gRPC streaming releases processed buffers while the stream stays open', (t) => {
+  const result = spawnSync(process.execPath, ['--expose-gc', '--input-type=module', '--eval', `
+    import assert from 'node:assert/strict';
+    import { create } from '@bufbuild/protobuf';
+    import { createClient } from '@connectrpc/connect';
+    import { connectNodeAdapter, createGrpcTransport } from '@connectrpc/connect-node';
+    import http2 from 'node:http2';
+    import { setImmediate as tick } from 'node:timers/promises';
+    import { ConnectionEventsSchema, StartedService } from './dist/server/generated/daemon/started_service_pb.js';
+
+    const count = 512;
+    const payload = 'x'.repeat(16 * 1024);
+    const sessions = new Set();
+    const server = http2.createServer(connectNodeAdapter({
+      routes(router) {
+        router.service(StartedService, {
+          async *subscribeConnections(_request, context) {
+            for (let i = 0; i < count; i++) {
+              yield create(ConnectionEventsSchema, {
+                events: [{ id: String(i), connection: { id: String(i), rule: payload } }],
+              });
+            }
+            await new Promise((resolve) => {
+              if (context.signal.aborted) resolve();
+              else context.signal.addEventListener('abort', resolve, { once: true });
+            });
+          },
+        });
+      },
+    }));
+    server.on('session', (session) => {
+      sessions.add(session);
+      session.once('close', () => sessions.delete(session));
+    });
+    const controller = new AbortController();
+    async function collectedBuffers() {
+      for (let i = 0; i < 3; i++) {
+        await tick();
+        global.gc();
+      }
+      return process.memoryUsage().arrayBuffers;
+    }
+    try {
+      await new Promise((resolve, reject) => {
+        server.once('error', reject);
+        server.listen(0, '127.0.0.1', resolve);
+      });
+      const client = createClient(StartedService, createGrpcTransport({
+        baseUrl: 'http://127.0.0.1:' + server.address().port,
+      }));
+      const baseline = await collectedBuffers();
+      const iterator = client.subscribeConnections({ interval: 1n }, {
+        signal: controller.signal,
+      })[Symbol.asyncIterator]();
+      for (let i = 0; i < count; i++) {
+        const next = await iterator.next();
+        assert.equal(next.done, false);
+        assert.equal(next.value.events[0].id, String(i));
+        assert.equal(next.value.events[0].connection.rule, payload);
+      }
+      const retainedBytes = await collectedBuffers() - baseline;
+      console.log(JSON.stringify({ processedBytes: count * payload.length, retainedBytes }));
+      assert.equal(controller.signal.aborted, false);
+      assert.ok(retainedBytes < 2 * 1024 * 1024,
+        'processed gRPC buffers remain retained while the stream is open: ' + retainedBytes);
+    } finally {
+      controller.abort();
+      for (const session of sessions) session.destroy();
+      if (server.listening) await new Promise((resolve) => server.close(resolve));
+    }
+  `], {
+    cwd: path.resolve(import.meta.dirname, '../..'),
+    encoding: 'utf8',
+    timeout: 20_000,
+  });
+  assert.ifError(result.error);
+  assert.equal(result.status, 0, result.stderr + result.stdout);
+  t.diagnostic(result.stdout.trim());
+});