diff --git a/.github/workflows/publish-wingspan-grpc-js.yml b/.github/workflows/publish-wingspan-grpc-js.yml new file mode 100644 index 000000000..9ae24fc3b --- /dev/null +++ b/.github/workflows/publish-wingspan-grpc-js.yml @@ -0,0 +1,53 @@ +name: Publish @wingspanhq/grpc-js + +on: + push: + branches: + - 'wingspan/**' + paths: + - 'packages/grpc-js/**' + - '.github/workflows/publish-wingspan-grpc-js.yml' + workflow_dispatch: + +jobs: + publish: + name: build and publish to GitHub Packages + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Init proto submodules + run: git submodule update --init packages/grpc-js-xds/deps/xds packages/grpc-js-xds/deps/protoc-gen-validate + - name: Setup node + uses: actions/setup-node@v4 + with: + node-version: '20.x' + registry-url: 'https://npm.pkg.github.com' + scope: '@wingspanhq' + # --ignore-scripts everywhere: the sibling packages' prepare scripts do + # not build in isolation, and generated proto types are checked in, so + # nothing needs regenerating. proto-loader is built explicitly because + # grpc-js consumes it as a file: devDependency (a symlink) and needs its + # build/ output for type resolution. + - name: Build proto-loader (sibling devDependency) + run: | + npm install --ignore-scripts + ./node_modules/.bin/tsc -p . + working-directory: packages/proto-loader + - name: Install grpc-js dependencies + run: npm install --ignore-scripts + working-directory: packages/grpc-js + - name: Copy ORCA protos + run: node copy-protos.js + working-directory: packages/grpc-js + - name: Compile + run: ./node_modules/.bin/tsc -p tsconfig.publish.json + working-directory: packages/grpc-js + - name: Publish + run: npm publish --ignore-scripts + working-directory: packages/grpc-js + env: + NODE_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/FORK.md b/FORK.md new file mode 100644 index 000000000..2f689f997 --- /dev/null +++ b/FORK.md @@ -0,0 +1,104 @@ +# Wingspan fork of grpc/grpc-node + +This is Wingspan's fork of [grpc/grpc-node](https://github.com/grpc/grpc-node). It exists to +publish **`@wingspanhq/grpc-js`** — a build of `@grpc/grpc-js` that carries one behavioral fix — +to GitHub Packages. No other package in this monorepo is published from the fork. + +## Why this fork exists + +A graceful HTTP/2 GOAWAY leaves grpc-js's keepalive timer running on both the client transport +and the server session. The next PING that fails or times out triggers the disconnect path, +destroying in-flight RPCs the GOAWAY had deliberately allowed to finish draining. In production +this surfaced as status 14 / `Connection dropped` on calls that were completing normally during +deploys and connection-age recycling. + +Reported upstream as [grpc/grpc-node#3068](https://github.com/grpc/grpc-node/issues/3068). +Upstream has pushed back on the premise (RFC 9113 requires a PING to be ACKed regardless of +GOAWAY), so this fix may never merge upstream in its current form. Until that conversation +resolves, this fork is the delivery vehicle. It replaces the previous mechanism — a +`patch-package` postinstall hook in `@wingspanhq/grpc` that rewrote grpc-js's compiled output in +consumers' `node_modules` — which was fragile across package managers and install layouts +(see wingspanHQ/grpc#89). + +## What is changed relative to upstream + +Branch naming: `wingspan/grpc-js-`, based on the upstream release tag +`@grpc/grpc-js@`. Each branch contains exactly two kinds of commits: + +1. **The fix** — `packages/grpc-js/src/transport.ts` and `packages/grpc-js/src/server.ts` only. + Client: mark the transport draining on `goaway`, clear the keepalive timeout, refuse new + pings, ignore in-flight ping completions. Server: begin draining before locally initiated + closes (`closeSession`, max connection age) and on peer GOAWAY. Kept free of fork-identity + noise so it can be cherry-picked into an upstream PR. +2. **Fork identity** — package rename to `@wingspanhq/grpc-js`, version `X.Y.Z-wingspan.N`, + GitHub Packages `publishConfig`, this file, and the publish workflow. + +`master` tracks upstream and carries no Wingspan changes. + +## How it is consumed + +`@wingspanhq/grpc` (the internal gRPC framework) depends on it via an npm alias: + +```json +"@grpc/grpc-js": "npm:@wingspanhq/grpc-js@1.14.4-wingspan.1" +``` + +so every `import from '@grpc/grpc-js'` — including deep imports and `@grpc/grpc-js-xds`'s peer +resolution — lands on the fork. Services do not (and should not) depend on this package +directly; it arrives transitively through `@wingspanhq/grpc`. + +Expect a yarn peer-dependency warning from `@grpc/grpc-js-xds` (`~1.14.0` does not match a +prerelease version). It is benign: the alias puts the fork at `node_modules/@grpc/grpc-js`, +which is what xds resolves. + +## Publishing a new version + +The `Publish @wingspanhq/grpc-js` workflow (`.github/workflows/publish-wingspan-grpc-js.yml`) +runs on pushes to `wingspan/**` branches that touch `packages/grpc-js/`, or manually via +workflow dispatch. It builds from checked-in generated types (no proto regeneration) and +publishes to GitHub Packages with the repo's `GITHUB_TOKEN`. + +To publish locally instead (requires a token with `write:packages` for wingspanHQ): + +```bash +git submodule update --init packages/grpc-js-xds/deps/xds packages/grpc-js-xds/deps/protoc-gen-validate +(cd packages/proto-loader && npm install --ignore-scripts && ./node_modules/.bin/tsc -p .) +cd packages/grpc-js +npm install --ignore-scripts +node copy-protos.js +./node_modules/.bin/tsc -p tsconfig.publish.json +npm publish --ignore-scripts +``` + +`--ignore-scripts` on install stops the sibling packages' `prepare` scripts, which do not build +in isolation; on publish it skips `prepare`, which would regenerate checked-in proto types with +tooling this flow deliberately avoids. proto-loader must be compiled first because grpc-js +consumes it as a `file:` devDependency (a symlink) and resolves types from its `build/` output. +`tsconfig.publish.json` compiles `src/` only (the published `build/src` tree) with node types +included explicitly; the stock `tsconfig.json` targets upstream's gulp pipeline and compiles +tests too. + +## Tracking a new upstream release + +When upstream tags `@grpc/grpc-js@X.Y.Z`: + +```bash +git fetch upstream --tags # upstream = https://github.com/grpc/grpc-node.git +git checkout -b wingspan/grpc-js-X.Y.Z "@grpc/grpc-js@X.Y.Z" +git cherry-pick # commit 1 from the previous wingspan/ branch +git cherry-pick # then bump version to X.Y.Z-wingspan.1 +``` + +Resolve conflicts in the two touched source files by re-reading the surrounding upstream code — +the fix is small and its anchors (goaway handler, `canSendPing`, ping callback, `closeSession`, +connection-age timers) are stable but not guaranteed. After building, compare the compiled +`transport.js`/`server.js` against the previous branch's output to review what upstream changed +underneath the fix. Then update the alias version and the expected SHA-256 digests in +`@wingspanhq/grpc` (`src/grpcJsPatchIntegrity.ts`) — its startup integrity check hashes the +compiled files and fails closed on drift. + +## Exit criteria + +If upstream ships an equivalent fix (or #3068 concludes with a config-level answer), point +`@wingspanhq/grpc` back at stock `@grpc/grpc-js`, delete the alias and digests bump, and archive +this fork. diff --git a/packages/grpc-js-xds/README.md b/packages/grpc-js-xds/README.md index bab5a588f..f97f66032 100644 --- a/packages/grpc-js-xds/README.md +++ b/packages/grpc-js-xds/README.md @@ -31,9 +31,11 @@ const client = new MyServiceClient('xds:///example.com:123'); - [Outlier Detection](https://github.com/grpc/proposal/blob/master/A50-xds-outlier-detection.md) - [xDS Retry Support](https://github.com/grpc/proposal/blob/master/A44-xds-retry.md) - [xDS Aggregate and Logical DNS Clusters](https://github.com/grpc/proposal/blob/master/A37-xds-aggregate-and-logical-dns-clusters.md) - - [xDS Federation](https://github.com/grpc/proposal/blob/master/A47-xds-federation.md) (Currently experimental, enabled by environment variable `GRPC_EXPERIMENTAL_XDS_FEDERATION`) + - [xDS Federation](https://github.com/grpc/proposal/blob/master/A47-xds-federation.md) - [xDS Custom Load Balancer Configuration](https://github.com/grpc/proposal/blob/master/A52-xds-custom-lb-policies.md) (Custom load balancer registration not currently supported) - [xDS Ring Hash LB Policy](https://github.com/grpc/proposal/blob/master/A42-xds-ring-hash-lb-policy.md) - [`pick_first` via xDS](https://github.com/grpc/proposal/blob/master/A62-pick-first.md#pick_first-via-xds-1) (Currently experimental, enabled by environment variable `GRPC_EXPERIMENTAL_PICKFIRST_LB_CONFIG`) - [xDS-Enabled Servers](https://github.com/grpc/proposal/blob/master/A36-xds-for-servers.md) - [xDS-Based Security for gRPC Clients and Servers](https://github.com/grpc/proposal/blob/master/A29-xds-tls-security.md) + - [xDS RBAC Support](https://github.com/grpc/proposal/blob/master/A41-xds-rbac.md) + - [`weighted_round_robin` LB policy](https://github.com/grpc/proposal/blob/master/A58-client-side-weighted-round-robin-lb-policy.md) (Inclusion in xDS registry is currently experimental, enabled by environment variable `GRPC_EXPERIMENTAL_XDS_WRR_LB`) diff --git a/packages/grpc-js-xds/package.json b/packages/grpc-js-xds/package.json index 8d937db59..7b331caac 100644 --- a/packages/grpc-js-xds/package.json +++ b/packages/grpc-js-xds/package.json @@ -1,6 +1,6 @@ { "name": "@grpc/grpc-js-xds", - "version": "1.13.0", + "version": "1.14.0", "description": "Plugin for @grpc/grpc-js. Adds the xds:// URL scheme and associated features.", "main": "build/src/index.js", "scripts": { @@ -38,7 +38,7 @@ "@types/gulp": "^4.0.6", "@types/gulp-mocha": "0.0.32", "@types/mocha": "^5.2.6", - "@types/node": ">=20.11.20", + "@types/node": "25.5.0", "@types/yargs": "^15.0.5", "grpc-health-check": "file:../grpc-health-check", "gts": "^5.0.1", @@ -55,7 +55,7 @@ "xxhash-wasm": "^1.0.2" }, "peerDependencies": { - "@grpc/grpc-js": "~1.13.0" + "@grpc/grpc-js": "~1.14.0" }, "engines": { "node": ">=10.10.0" diff --git a/packages/grpc-js-xds/src/rbac.ts b/packages/grpc-js-xds/src/rbac.ts index b1d8558f0..31f8cb0cc 100644 --- a/packages/grpc-js-xds/src/rbac.ts +++ b/packages/grpc-js-xds/src/rbac.ts @@ -225,7 +225,14 @@ export class AuthenticatedPrincipal implements PrincipalRule { } } } - return this.nameMatcher.apply(info.peerCertificate.subject.CN); + if (info.peerCertificate.subject.CN) { + if (Array.isArray(info.peerCertificate.subject.CN)) { + return info.peerCertificate.subject.CN.some(entry => this.nameMatcher!.apply(entry)); + } else { + return this.nameMatcher.apply(info.peerCertificate.subject.CN); + } + } + return false; } toString(): string { return `Authenticated(principal=${this.nameMatcher?.toString() ?? null})`; diff --git a/packages/grpc-js/package.json b/packages/grpc-js/package.json index 4bb31e9a4..4ef05e5d2 100644 --- a/packages/grpc-js/package.json +++ b/packages/grpc-js/package.json @@ -1,9 +1,9 @@ { - "name": "@grpc/grpc-js", - "version": "1.13.4", - "description": "gRPC Library for Node - pure JS implementation", + "name": "@wingspanhq/grpc-js", + "version": "1.14.4-wingspan.1", + "description": "Wingspan fork of @grpc/grpc-js carrying the GOAWAY/keepalive drain fix (grpc/grpc-node#3068)", "homepage": "https://grpc.io/", - "repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js", + "repository": "git+https://github.com/wingspanHQ/grpc-node.git", "main": "build/src/index.js", "engines": { "node": ">=12.10.0" @@ -85,5 +85,9 @@ "deps/googleapis/google/api/*.proto", "deps/googleapis/google/rpc/*.proto", "deps/protoc-gen-validate/validate/**/*.proto" - ] + ], + "publishConfig": { + "registry": "https://npm.pkg.github.com/", + "access": "restricted" + } } diff --git a/packages/grpc-js/src/compression-filter.ts b/packages/grpc-js/src/compression-filter.ts index e4428a1fb..5277d8c1c 100644 --- a/packages/grpc-js/src/compression-filter.ts +++ b/packages/grpc-js/src/compression-filter.ts @@ -119,6 +119,12 @@ class DeflateHandler extends CompressionHandler { let totalLength = 0; const messageParts: Buffer[] = []; const decompresser = zlib.createInflate(); + decompresser.on('error', (error: Error) => { + reject({ + code: Status.INTERNAL, + details: 'Failed to decompress deflate-encoded message' + }); + }); decompresser.on('data', (chunk: Buffer) => { messageParts.push(chunk); totalLength += chunk.byteLength; @@ -161,6 +167,12 @@ class GzipHandler extends CompressionHandler { let totalLength = 0; const messageParts: Buffer[] = []; const decompresser = zlib.createGunzip(); + decompresser.on('error', (error: Error) => { + reject({ + code: Status.INTERNAL, + details: 'Failed to decompress gzip-encoded message' + }); + }); decompresser.on('data', (chunk: Buffer) => { messageParts.push(chunk); totalLength += chunk.byteLength; diff --git a/packages/grpc-js/src/retrying-call.ts b/packages/grpc-js/src/retrying-call.ts index 1d49ad337..61ff58fa1 100644 --- a/packages/grpc-js/src/retrying-call.ts +++ b/packages/grpc-js/src/retrying-call.ts @@ -760,11 +760,10 @@ export class RetryingCall implements Call, DeadlineInfoProvider { this.maybeStartHedgingTimer(); } - private handleChildWriteCompleted(childIndex: number) { - const childCall = this.underlyingCalls[childIndex]; - const messageIndex = childCall.nextMessageToSend; + private handleChildWriteCompleted(childIndex: number, messageIndex: number) { this.getBufferEntry(messageIndex).callback?.(); this.clearSentMessages(); + const childCall = this.underlyingCalls[childIndex]; childCall.nextMessageToSend += 1; this.sendNextChildMessage(childIndex); } @@ -774,19 +773,33 @@ export class RetryingCall implements Call, DeadlineInfoProvider { if (childCall.state === 'COMPLETED') { return; } - if (this.getBufferEntry(childCall.nextMessageToSend)) { - const bufferEntry = this.getBufferEntry(childCall.nextMessageToSend); + const messageIndex = childCall.nextMessageToSend; + if (this.getBufferEntry(messageIndex)) { + const bufferEntry = this.getBufferEntry(messageIndex); switch (bufferEntry.entryType) { case 'MESSAGE': childCall.call.sendMessageWithContext( { callback: error => { // Ignore error - this.handleChildWriteCompleted(childIndex); + this.handleChildWriteCompleted(childIndex, messageIndex); }, }, bufferEntry.message!.message ); + // Optimization: if the next entry is HALF_CLOSE, send it immediately + // without waiting for the message callback. This is safe because the message + // has already been passed to the underlying transport. + const nextEntry = this.getBufferEntry(messageIndex + 1); + if (nextEntry.entryType === 'HALF_CLOSE') { + this.trace( + 'Sending halfClose immediately after message to child [' + + childCall.call.getCallNumber() + + '] - optimizing for unary/final message' + ); + childCall.nextMessageToSend += 1; + childCall.call.halfClose(); + } break; case 'HALF_CLOSE': childCall.nextMessageToSend += 1; @@ -813,7 +826,11 @@ export class RetryingCall implements Call, DeadlineInfoProvider { }; this.writeBuffer.push(bufferEntry); if (bufferEntry.allocated) { - context.callback?.(); + // Run this in next tick to avoid suspending the current execution context + // otherwise it might cause half closing the call before sending message + process.nextTick(() => { + context.callback?.(); + }); for (const [callIndex, call] of this.underlyingCalls.entries()) { if ( call.state === 'ACTIVE' && @@ -823,7 +840,7 @@ export class RetryingCall implements Call, DeadlineInfoProvider { { callback: error => { // Ignore error - this.handleChildWriteCompleted(callIndex); + this.handleChildWriteCompleted(callIndex, messageIndex); }, }, message @@ -843,7 +860,7 @@ export class RetryingCall implements Call, DeadlineInfoProvider { { callback: error => { // Ignore error - this.handleChildWriteCompleted(this.committedCallIndex!); + this.handleChildWriteCompleted(this.committedCallIndex!, messageIndex); }, }, message @@ -868,12 +885,21 @@ export class RetryingCall implements Call, DeadlineInfoProvider { allocated: false, }); for (const call of this.underlyingCalls) { - if ( - call?.state === 'ACTIVE' && - call.nextMessageToSend === halfCloseIndex - ) { - call.nextMessageToSend += 1; - call.call.halfClose(); + if (call?.state === 'ACTIVE') { + // Send halfClose to call when either: + // - nextMessageToSend === halfCloseIndex - 1: last message sent, callback pending (optimization) + // - nextMessageToSend === halfCloseIndex: all messages sent and acknowledged + if (call.nextMessageToSend === halfCloseIndex + || call.nextMessageToSend === halfCloseIndex - 1) { + this.trace( + 'Sending halfClose immediately to child [' + + call.call.getCallNumber() + + '] - all messages already sent' + ); + call.nextMessageToSend += 1; + call.call.halfClose(); + } + // Otherwise, halfClose will be sent by sendNextChildMessage when message callbacks complete } } } @@ -895,4 +921,4 @@ export class RetryingCall implements Call, DeadlineInfoProvider { return null; } } -} +} \ No newline at end of file diff --git a/packages/grpc-js/src/server-interceptors.ts b/packages/grpc-js/src/server-interceptors.ts index a7cddd933..9daa78785 100644 --- a/packages/grpc-js/src/server-interceptors.ts +++ b/packages/grpc-js/src/server-interceptors.ts @@ -19,7 +19,7 @@ import { PartialStatusObject } from './call-interface'; import { ServerMethodDefinition } from './make-client'; import { Metadata } from './metadata'; import { ChannelOptions } from './channel-options'; -import { Handler, ServerErrorResponse } from './server-call'; +import { Handler } from './server-call'; import { Deadline } from './deadline'; import { DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH, @@ -554,14 +554,6 @@ export class BaseServerInterceptingCall private readonly handler: Handler, options: ChannelOptions ) { - this.stream.once('error', (err: ServerErrorResponse) => { - /* We need an error handler to avoid uncaught error event exceptions, but - * there is nothing we can reasonably do here. Any error event should - * have a corresponding close event, which handles emitting the cancelled - * event. And the stream is now in a bad state, so we can't reasonably - * expect to be able to send an error over it. */ - }); - this.stream.once('close', () => { trace( 'Request to method ' + @@ -744,6 +736,12 @@ export class BaseServerInterceptingCall return new Promise((resolve, reject) => { let totalLength = 0 const messageParts: Buffer[] = []; + decompresser.on('error', (error: Error) => { + reject({ + code: Status.INTERNAL, + details: 'Failed to decompress message' + }); + }); decompresser.on('data', (chunk: Buffer) => { messageParts.push(chunk); totalLength += chunk.byteLength; diff --git a/packages/grpc-js/src/server.ts b/packages/grpc-js/src/server.ts index 6e68c695a..1c2b31f4b 100644 --- a/packages/grpc-js/src/server.ts +++ b/packages/grpc-js/src/server.ts @@ -279,6 +279,11 @@ export class Server { UntypedHandler >(); private sessions = new Map(); + private drainingSessions = new WeakSet(); + private sessionDrainHandlers = new WeakMap< + http2.ServerHttp2Session, + () => void + >(); /** * This field only exists to ensure that the start method throws an error if * it is called twice, as it did previously. @@ -1034,11 +1039,20 @@ export class Server { }); } + private beginSessionDrain(session: http2.ServerHttp2Session) { + /* A PING on a locally closing session bypasses graceful drain by + * destroying active streams, so disable keepalive before close. */ + this.drainingSessions.add(session); + const drainHandler = this.sessionDrainHandlers.get(session); + drainHandler?.(); + } + private closeSession( session: http2.ServerHttp2Session, callback?: () => void ) { this.trace('Closing session initiated by ' + session.socket?.remoteAddress); + this.beginSessionDrain(session); const sessionInfo = this.sessions.get(session); const closeCallback = () => { if (sessionInfo) { @@ -1336,6 +1350,13 @@ export class Server { stream: http2.ServerHttp2Stream, headers: http2.IncomingHttpHeaders ) { + stream.once('error', (err: ServerErrorResponse) => { + /* We need an error handler to avoid uncaught error event exceptions, but + * there is nothing we can reasonably do here. Any error event should + * have a corresponding close event, which handles emitting the cancelled + * event. And the stream is now in a bad state, so we can't reasonably + * expect to be able to send an error over it. */ + }); // for handling idle timeout this.onStreamOpened(stream); @@ -1420,6 +1441,13 @@ export class Server { stream: http2.ServerHttp2Stream, headers: http2.IncomingHttpHeaders ) { + stream.once('error', (err: ServerErrorResponse) => { + /* We need an error handler to avoid uncaught error event exceptions, but + * there is nothing we can reasonably do here. Any error event should + * have a corresponding close event, which handles emitting the cancelled + * event. And the stream is now in a bad state, so we can't reasonably + * expect to be able to send an error over it. */ + }); // for handling idle timeout this.onStreamOpened(stream); @@ -1531,6 +1559,7 @@ export class Server { connectionAgeTimer = setTimeout(() => { sessionClosedByServer = true; + this.beginSessionDrain(session); this.trace( 'Connection dropped by max connection age: ' + @@ -1569,8 +1598,14 @@ export class Server { } }; + this.sessionDrainHandlers.set(session, clearKeepaliveTimeout); + session.once('goaway', () => { + this.beginSessionDrain(session); + }); + const canSendPing = () => { return ( + !this.drainingSessions.has(session) && !session.destroyed && this.keepaliveTimeMs < KEEPALIVE_MAX_TIME_MS && this.keepaliveTimeMs > 0 @@ -1606,10 +1641,16 @@ export class Server { const pingSentSuccessfully = session.ping( (err: Error | null, duration: number, payload: Buffer) => { clearKeepaliveTimeout(); + if (this.drainingSessions.has(session)) { + this.keepaliveTrace( + 'Ignoring ping result on draining server session' + ); + return; + } if (err) { this.keepaliveTrace('Ping failed with error: ' + err.message); sessionClosedByServer = true; - session.close(); + session.destroy(); } else { this.keepaliveTrace('Received ping response'); maybeStartKeepalivePingTimer(); @@ -1631,7 +1672,7 @@ export class Server { 'Connection dropped due to ping send error: ' + pingSendError ); sessionClosedByServer = true; - session.close(); + session.destroy(); return; } @@ -1640,7 +1681,7 @@ export class Server { this.keepaliveTrace('Ping timeout passed without response'); this.trace('Connection dropped by keepalive timeout'); sessionClosedByServer = true; - session.close(); + session.destroy(); }, this.keepaliveTimeoutMs); keepaliveTimer.unref?.(); }; @@ -1663,6 +1704,7 @@ export class Server { } clearKeepaliveTimeout(); + this.sessionDrainHandlers.delete(session); if (idleTimeoutObj !== null) { clearTimeout(idleTimeoutObj.timeout); @@ -1719,6 +1761,7 @@ export class Server { connectionAgeTimer = setTimeout(() => { sessionClosedByServer = true; + this.beginSessionDrain(session); this.channelzTrace.addTrace( 'CT_INFO', 'Connection dropped by max connection age from ' + clientAddress @@ -1756,8 +1799,14 @@ export class Server { } }; + this.sessionDrainHandlers.set(session, clearKeepaliveTimeout); + session.once('goaway', () => { + this.beginSessionDrain(session); + }); + const canSendPing = () => { return ( + !this.drainingSessions.has(session) && !session.destroyed && this.keepaliveTimeMs < KEEPALIVE_MAX_TIME_MS && this.keepaliveTimeMs > 0 @@ -1793,6 +1842,12 @@ export class Server { const pingSentSuccessfully = session.ping( (err: Error | null, duration: number, payload: Buffer) => { clearKeepaliveTimeout(); + if (this.drainingSessions.has(session)) { + this.keepaliveTrace( + 'Ignoring ping result on draining server session' + ); + return; + } if (err) { this.keepaliveTrace('Ping failed with error: ' + err.message); this.channelzTrace.addTrace( @@ -1803,7 +1858,7 @@ export class Server { duration ); sessionClosedByServer = true; - session.close(); + session.destroy(); } else { this.keepaliveTrace('Received ping response'); maybeStartKeepalivePingTimer(); @@ -1826,7 +1881,7 @@ export class Server { 'Connection dropped due to ping send error: ' + pingSendError ); sessionClosedByServer = true; - session.close(); + session.destroy(); return; } @@ -1840,7 +1895,7 @@ export class Server { 'Connection dropped by keepalive timeout from ' + clientAddress ); sessionClosedByServer = true; - session.close(); + session.destroy(); }, this.keepaliveTimeoutMs); keepaliveTimeout.unref?.(); }; @@ -1867,6 +1922,7 @@ export class Server { } clearKeepaliveTimeout(); + this.sessionDrainHandlers.delete(session); if (idleTimeoutObj !== null) { clearTimeout(idleTimeoutObj.timeout); diff --git a/packages/grpc-js/src/transport.ts b/packages/grpc-js/src/transport.ts index 6fea1198c..894a42715 100644 --- a/packages/grpc-js/src/transport.ts +++ b/packages/grpc-js/src/transport.ts @@ -121,6 +121,12 @@ class Http2Transport implements Transport { */ private pendingSendKeepalivePing = false; + /** + * Indicates that the peer sent a GOAWAY, so this connection is draining and + * must not be probed with further keepalive pings. + */ + private isDraining = false; + private userAgent: string; private activeCalls: Set = new Set(); @@ -214,6 +220,8 @@ class Http2Transport implements Transport { ) { tooManyPings = true; } + this.isDraining = true; + this.clearKeepaliveTimeout(); this.trace( 'connection closed by GOAWAY with code ' + errorCode + @@ -415,6 +423,7 @@ class Http2Transport implements Transport { private canSendPing() { return ( + !this.isDraining && !this.session.destroyed && this.keepaliveTimeMs > 0 && (this.keepaliveWithoutCalls || this.activeCalls.size > 0) @@ -422,6 +431,9 @@ class Http2Transport implements Transport { } private maybeSendPing() { + if (this.isDraining) { + return; + } if (!this.canSendPing()) { this.pendingSendKeepalivePing = true; return; @@ -447,6 +459,10 @@ class Http2Transport implements Transport { const pingSentSuccessfully = this.session.ping( (err: Error | null, duration: number, payload: Buffer) => { this.clearKeepaliveTimeout(); + if (this.isDraining) { + this.keepaliveTrace('Ignoring ping result on draining transport'); + return; + } if (err) { this.keepaliveTrace('Ping failed with error ' + err.message); this.handleDisconnect(); @@ -720,7 +736,13 @@ export class Http2SubchannelConnector implements SubchannelConnector { initialWindowSize: options['grpc-node.flow_control_window'] ?? http2.getDefaultSettings?.()?.initialWindowSize ?? 65535, - } + }, + maxSendHeaderBlockLength: Number.MAX_SAFE_INTEGER, + /* By default, set a very large max session memory limit, to effectively + * disable enforcement of the limit. Some testing indicates that Node's + * behavior degrades badly when this limit is reached, so we solve that + * by disabling the check entirely. */ + maxSessionMemory: options['grpc-node.max_session_memory'] ?? Number.MAX_SAFE_INTEGER }; const session = http2.connect(`${scheme}://${targetPath}`, sessionOptions); // Prepare window size configuration for remoteSettings handler diff --git a/packages/grpc-js/test/test-end-to-end.ts b/packages/grpc-js/test/test-end-to-end.ts index c7de2d6a6..676b4cff0 100644 --- a/packages/grpc-js/test/test-end-to-end.ts +++ b/packages/grpc-js/test/test-end-to-end.ts @@ -18,7 +18,7 @@ import * as assert from 'assert'; import * as path from 'path'; import { loadProtoFile } from './common'; -import { Metadata, Server, ServerDuplexStream, ServerUnaryCall, ServiceClientConstructor, ServiceError, experimental, sendUnaryData } from '../src'; +import { Metadata, Server, ServerCredentials, ServerDuplexStream, ServerReadableStream, ServerUnaryCall, ServiceClientConstructor, ServiceError, credentials, experimental, sendUnaryData } from '../src'; import { ServiceClient } from '../src/make-client'; const protoFile = path.join(__dirname, 'fixtures', 'echo_service.proto'); @@ -36,6 +36,15 @@ const echoServiceImplementation = { call.end(); }); }, + echoClientStream(call: ServerReadableStream, callback: sendUnaryData) { + const messages: any[] = []; + call.on('data', (message: any) => { + messages.push(message); + }); + call.on('end', () => { + callback(null, { value: messages.map(m => m.value).join(','), value2: messages.length }); + }); + }, }; describe('Client should successfully communicate with server', () => { @@ -77,4 +86,20 @@ describe('Client should successfully communicate with server', () => { }); }); }).timeout(5000); + + it('Client streaming with one message should work', done => { + server = new Server(); + server.addService(EchoService.service, echoServiceImplementation); + server.bindAsync('localhost:0', ServerCredentials.createInsecure(), (error, port) => { + assert.ifError(error); + client = new EchoService(`localhost:${port}`, credentials.createInsecure()); + const call = client.echoClientStream((error: ServiceError, response: any) => { + assert.ifError(error); + assert.deepStrictEqual(response, { value: 'test value', value2: 1 }); + done(); + }); + call.write({ value: 'test value', value2: 42 }); + call.end(); + }); + }); }); diff --git a/packages/grpc-js/tsconfig.publish.json b/packages/grpc-js/tsconfig.publish.json new file mode 100644 index 000000000..350bf5e15 --- /dev/null +++ b/packages/grpc-js/tsconfig.publish.json @@ -0,0 +1,14 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "incremental": false, + "types": [ + "node" + ], + "rootDir": "src", + "outDir": "build/src" + }, + "include": [ + "src/**/*.ts" + ] +}