diff --git a/packages/grpc-js/src/server.ts b/packages/grpc-js/src/server.ts index feb511b41..33db41004 100644 --- a/packages/grpc-js/src/server.ts +++ b/packages/grpc-js/src/server.ts @@ -435,6 +435,14 @@ export class Server { ); } + private keepaliveTrace(text: string): void { + logging.trace( + LogVerbosity.DEBUG, + 'keepalive', + '(' + this.channelzRef.id + ') ' + text + ); + } + addProtoService(): never { throw new Error('Not implemented. Use addService() instead'); } @@ -1376,8 +1384,7 @@ export class Server { let connectionAgeTimer: NodeJS.Timeout | null = null; let connectionAgeGraceTimer: NodeJS.Timeout | null = null; - let keeapliveTimeTimer: NodeJS.Timeout | null = null; - let keepaliveTimeoutTimer: NodeJS.Timeout | null = null; + let keepaliveTimer: NodeJS.Timeout | null = null; let sessionClosedByServer = false; const idleTimeoutObj = this.enableIdleTimeout(session); @@ -1420,41 +1427,90 @@ export class Server { connectionAgeTimer.unref?.(); } - if (this.keepaliveTimeMs < KEEPALIVE_MAX_TIME_MS) { - keeapliveTimeTimer = setInterval(() => { - keepaliveTimeoutTimer = setTimeout(() => { - sessionClosedByServer = true; - session.close(); - }, this.keepaliveTimeoutMs); - keepaliveTimeoutTimer.unref?.(); + const clearKeepaliveTimeout = () => { + if (keepaliveTimer) { + clearTimeout(keepaliveTimer); + keepaliveTimer = null; + } + }; - try { - session.ping( - (err: Error | null, duration: number, payload: Buffer) => { - if (keepaliveTimeoutTimer) { - clearTimeout(keepaliveTimeoutTimer); - } - - if (err) { - sessionClosedByServer = true; - this.trace( - 'Connection dropped due to error of a ping frame ' + - err.message + - ' return in ' + - duration - ); - session.close(); - } + const canSendPing = () => { + return ( + !session.destroyed && + this.keepaliveTimeMs < KEEPALIVE_MAX_TIME_MS && + this.keepaliveTimeMs > 0 + ); + }; + + /* eslint-disable-next-line prefer-const */ + let sendPing: () => void; // hoisted for use in maybeStartKeepalivePingTimer + + const maybeStartKeepalivePingTimer = () => { + if (!canSendPing()) { + return; + } + this.keepaliveTrace( + 'Starting keepalive timer for ' + this.keepaliveTimeMs + 'ms' + ); + keepaliveTimer = setTimeout(() => { + clearKeepaliveTimeout(); + sendPing(); + }, this.keepaliveTimeMs); + keepaliveTimer.unref?.(); + }; + + sendPing = () => { + if (!canSendPing()) { + return; + } + this.keepaliveTrace( + 'Sending ping with timeout ' + this.keepaliveTimeoutMs + 'ms' + ); + let pingSendError = ''; + try { + const pingSentSuccessfully = session.ping( + (err: Error | null, duration: number, payload: Buffer) => { + clearKeepaliveTimeout(); + if (err) { + this.keepaliveTrace('Ping failed with error: ' + err.message); + sessionClosedByServer = true; + session.close(); + } else { + this.keepaliveTrace('Received ping response'); + maybeStartKeepalivePingTimer(); } - ); - } catch (e) { - clearTimeout(keepaliveTimeoutTimer); - // The ping can't be sent because the session is already closed - session.destroy(); + } + ); + if (!pingSentSuccessfully) { + pingSendError = 'Ping returned false'; } - }, this.keepaliveTimeMs); - keeapliveTimeTimer.unref?.(); - } + } catch (e) { + // grpc/grpc-node#2139 + pingSendError = + (e instanceof Error ? e.message : '') || 'Unknown error'; + } + + if (pingSendError) { + this.keepaliveTrace('Ping send failed: ' + pingSendError); + this.trace( + 'Connection dropped due to ping send error: ' + pingSendError + ); + sessionClosedByServer = true; + session.close(); + return; + } + + keepaliveTimer = setTimeout(() => { + clearKeepaliveTimeout(); + this.keepaliveTrace('Ping timeout passed without response'); + this.trace('Connection dropped by keepalive timeout'); + sessionClosedByServer = true; + session.close(); + }, this.keepaliveTimeoutMs); + keepaliveTimer.unref?.(); + }; + + maybeStartKeepalivePingTimer(); session.on('close', () => { if (!sessionClosedByServer) { @@ -1471,12 +1527,7 @@ export class Server { clearTimeout(connectionAgeGraceTimer); } - if (keeapliveTimeTimer) { - clearInterval(keeapliveTimeTimer); - if (keepaliveTimeoutTimer) { - clearTimeout(keepaliveTimeoutTimer); - } - } + clearKeepaliveTimeout(); if (idleTimeoutObj !== null) { clearTimeout(idleTimeoutObj.timeout); @@ -1521,8 +1572,7 @@ export class Server { let connectionAgeTimer: NodeJS.Timeout | null = null; let connectionAgeGraceTimer: NodeJS.Timeout | null = null; - let keeapliveTimeTimer: NodeJS.Timeout | null = null; - let keepaliveTimeoutTimer: NodeJS.Timeout | null = null; + let keepaliveTimeout: NodeJS.Timeout | null = null; let sessionClosedByServer = false; const idleTimeoutObj = this.enableIdleTimeout(session); @@ -1564,49 +1614,103 @@ export class Server { connectionAgeTimer.unref?.(); } - if (this.keepaliveTimeMs < KEEPALIVE_MAX_TIME_MS) { - keeapliveTimeTimer = setInterval(() => { - keepaliveTimeoutTimer = setTimeout(() => { - sessionClosedByServer = true; - this.channelzTrace.addTrace( - 'CT_INFO', - 'Connection dropped by keepalive timeout from ' + clientAddress - ); + const clearKeepaliveTimeout = () => { + if (keepaliveTimeout) { + clearTimeout(keepaliveTimeout); + keepaliveTimeout = null; + } + }; + + const canSendPing = () => { + return ( + !session.destroyed && + this.keepaliveTimeMs < KEEPALIVE_MAX_TIME_MS && + this.keepaliveTimeMs > 0 + ); + }; - session.close(); - }, this.keepaliveTimeoutMs); - keepaliveTimeoutTimer.unref?.(); + /* eslint-disable-next-line prefer-const */ + let sendPing: () => void; // hoisted for use in maybeStartKeepalivePingTimer - try { - session.ping( - (err: Error | null, duration: number, payload: Buffer) => { - if (keepaliveTimeoutTimer) { - clearTimeout(keepaliveTimeoutTimer); - } - - if (err) { - sessionClosedByServer = true; - this.channelzTrace.addTrace( - 'CT_INFO', - 'Connection dropped due to error of a ping frame ' + - err.message + - ' return in ' + - duration - ); - - session.close(); - } + const maybeStartKeepalivePingTimer = () => { + if (!canSendPing()) { + return; + } + this.keepaliveTrace( + 'Starting keepalive timer for ' + this.keepaliveTimeMs + 'ms' + ); + keepaliveTimeout = setTimeout(() => { + clearKeepaliveTimeout(); + sendPing(); + }, this.keepaliveTimeMs); + keepaliveTimeout.unref?.(); + }; + + sendPing = () => { + if (!canSendPing()) { + return; + } + this.keepaliveTrace( + 'Sending ping with timeout ' + this.keepaliveTimeoutMs + 'ms' + ); + let pingSendError = ''; + try { + const pingSentSuccessfully = session.ping( + (err: Error | null, duration: number, payload: Buffer) => { + clearKeepaliveTimeout(); + if (err) { + this.keepaliveTrace('Ping failed with error: ' + err.message); + this.channelzTrace.addTrace( + 'CT_INFO', + 'Connection dropped due to error of a ping frame ' + + err.message + + ' return in ' + + duration + ); + sessionClosedByServer = true; + session.close(); + } else { + this.keepaliveTrace('Received ping response'); + maybeStartKeepalivePingTimer(); } - ); - channelzSessionInfo.keepAlivesSent += 1; - } catch (e) { - clearTimeout(keepaliveTimeoutTimer); - // The ping can't be sent because the session is already closed - session.destroy(); + } + ); + if (!pingSentSuccessfully) { + pingSendError = 'Ping returned false'; } - }, this.keepaliveTimeMs); - keeapliveTimeTimer.unref?.(); - } + } catch (e) { + // grpc/grpc-node#2139 + pingSendError = + (e instanceof Error ? e.message : '') || 'Unknown error'; + } + + if (pingSendError) { + this.keepaliveTrace('Ping send failed: ' + pingSendError); + this.channelzTrace.addTrace( + 'CT_INFO', + 'Connection dropped due to ping send error: ' + pingSendError + ); + sessionClosedByServer = true; + session.close(); + return; + } + + channelzSessionInfo.keepAlivesSent += 1; + + keepaliveTimeout = setTimeout(() => { + clearKeepaliveTimeout(); + this.keepaliveTrace('Ping timeout passed without response'); + this.channelzTrace.addTrace( + 'CT_INFO', + 'Connection dropped by keepalive timeout from ' + clientAddress + ); + sessionClosedByServer = true; + session.close(); + }, this.keepaliveTimeoutMs); + keepaliveTimeout.unref?.(); + }; + + maybeStartKeepalivePingTimer(); session.on('close', () => { if (!sessionClosedByServer) { @@ -1627,12 +1731,7 @@ export class Server { clearTimeout(connectionAgeGraceTimer); } - if (keeapliveTimeTimer) { - clearInterval(keeapliveTimeTimer); - if (keepaliveTimeoutTimer) { - clearTimeout(keepaliveTimeoutTimer); - } - } + clearKeepaliveTimeout(); if (idleTimeoutObj !== null) { clearTimeout(idleTimeoutObj.timeout); diff --git a/packages/grpc-js/src/transport.ts b/packages/grpc-js/src/transport.ts index 934b62111..1acbab40e 100644 --- a/packages/grpc-js/src/transport.ts +++ b/packages/grpc-js/src/transport.ts @@ -102,28 +102,24 @@ class Http2Transport implements Transport { /** * The amount of time in between sending pings */ - private keepaliveTimeMs = -1; + private readonly keepaliveTimeMs: number; /** * The amount of time to wait for an acknowledgement after sending a ping */ - private keepaliveTimeoutMs: number = KEEPALIVE_TIMEOUT_MS; + private readonly keepaliveTimeoutMs: number; /** - * Timer reference for timeout that indicates when to send the next ping + * Indicates whether keepalive pings should be sent without any active calls + */ + private readonly keepaliveWithoutCalls: boolean; + /** + * Timer reference indicating when to send the next ping or when the most recent ping will be considered lost. */ - private keepaliveTimerId: NodeJS.Timeout | null = null; + private keepaliveTimer: NodeJS.Timeout | null = null; /** * Indicates that the keepalive timer ran out while there were no active * calls, and a ping should be sent the next time a call starts. */ private pendingSendKeepalivePing = false; - /** - * Timer reference tracking when the most recent ping will be considered lost - */ - private keepaliveTimeoutId: NodeJS.Timeout | null = null; - /** - * Indicates whether keepalive pings should be sent without any active calls - */ - private keepaliveWithoutCalls = false; private userAgent: string; @@ -183,9 +179,13 @@ class Http2Transport implements Transport { if ('grpc.keepalive_time_ms' in options) { this.keepaliveTimeMs = options['grpc.keepalive_time_ms']!; + } else { + this.keepaliveTimeMs = -1; } if ('grpc.keepalive_timeout_ms' in options) { this.keepaliveTimeoutMs = options['grpc.keepalive_timeout_ms']!; + } else { + this.keepaliveTimeoutMs = KEEPALIVE_TIMEOUT_MS; } if ('grpc.keepalive_permit_without_calls' in options) { this.keepaliveWithoutCalls = @@ -196,7 +196,6 @@ class Http2Transport implements Transport { session.once('close', () => { this.trace('session closed'); - this.stopKeepalivePings(); this.handleDisconnect(); }); @@ -384,6 +383,7 @@ class Http2Transport implements Transport { * Handle connection drops, but not GOAWAYs. */ private handleDisconnect() { + this.clearKeepaliveTimeout(); this.reportDisconnectToOwner(false); /* Give calls an event loop cycle to finish naturally before reporting the * disconnnection to them. */ @@ -391,6 +391,7 @@ class Http2Transport implements Transport { for (const call of this.activeCalls) { call.onDisconnect(); } + this.session.destroy(); }); } @@ -398,63 +399,58 @@ class Http2Transport implements Transport { this.disconnectListeners.push(listener); } - private clearKeepaliveTimer() { - if (!this.keepaliveTimerId) { - return; - } - clearTimeout(this.keepaliveTimerId); - this.keepaliveTimerId = null; - } - - private clearKeepaliveTimeout() { - if (!this.keepaliveTimeoutId) { - return; - } - clearTimeout(this.keepaliveTimeoutId); - this.keepaliveTimeoutId = null; - } - private canSendPing() { return ( + !this.session.destroyed && this.keepaliveTimeMs > 0 && (this.keepaliveWithoutCalls || this.activeCalls.size > 0) ); } private maybeSendPing() { - this.clearKeepaliveTimer(); if (!this.canSendPing()) { this.pendingSendKeepalivePing = true; return; } + if (this.keepaliveTimer) { + console.error('keepaliveTimeout is not null'); + return; + } if (this.channelzEnabled) { this.keepalivesSent += 1; } this.keepaliveTrace( 'Sending ping with timeout ' + this.keepaliveTimeoutMs + 'ms' ); - if (!this.keepaliveTimeoutId) { - this.keepaliveTimeoutId = setTimeout(() => { - this.keepaliveTrace('Ping timeout passed without response'); - this.handleDisconnect(); - }, this.keepaliveTimeoutMs); - this.keepaliveTimeoutId.unref?.(); - } + this.keepaliveTimer = setTimeout(() => { + this.keepaliveTimer = null; + this.keepaliveTrace('Ping timeout passed without response'); + this.handleDisconnect(); + }, this.keepaliveTimeoutMs); + this.keepaliveTimer.unref?.(); + let pingSendError = ''; try { - this.session!.ping( + const pingSentSuccessfully = this.session.ping( (err: Error | null, duration: number, payload: Buffer) => { + this.clearKeepaliveTimeout(); if (err) { this.keepaliveTrace('Ping failed with error ' + err.message); this.handleDisconnect(); + } else { + this.keepaliveTrace('Received ping response'); + this.maybeStartKeepalivePingTimer(); } - this.keepaliveTrace('Received ping response'); - this.clearKeepaliveTimeout(); - this.maybeStartKeepalivePingTimer(); } ); + if (!pingSentSuccessfully) { + pingSendError = 'Ping returned false'; + } } catch (e) { - /* If we fail to send a ping, the connection is no longer functional, so - * we should discard it. */ + // grpc/grpc-node#2139 + pingSendError = (e instanceof Error ? e.message : '') || 'Unknown error'; + } + if (pingSendError) { + this.keepaliveTrace('Ping send failed: ' + pingSendError); this.handleDisconnect(); } } @@ -472,25 +468,28 @@ class Http2Transport implements Transport { if (this.pendingSendKeepalivePing) { this.pendingSendKeepalivePing = false; this.maybeSendPing(); - } else if (!this.keepaliveTimerId && !this.keepaliveTimeoutId) { + } else if (!this.keepaliveTimer) { this.keepaliveTrace( 'Starting keepalive timer for ' + this.keepaliveTimeMs + 'ms' ); - this.keepaliveTimerId = setTimeout(() => { + this.keepaliveTimer = setTimeout(() => { + this.keepaliveTimer = null; this.maybeSendPing(); }, this.keepaliveTimeMs); - this.keepaliveTimerId.unref?.(); + this.keepaliveTimer.unref?.(); } /* Otherwise, there is already either a keepalive timer or a ping pending, * wait for those to resolve. */ } - private stopKeepalivePings() { - if (this.keepaliveTimerId) { - clearTimeout(this.keepaliveTimerId); - this.keepaliveTimerId = null; + /** + * Clears whichever keepalive timeout is currently active, if any. + */ + private clearKeepaliveTimeout() { + if (this.keepaliveTimer) { + clearTimeout(this.keepaliveTimer); + this.keepaliveTimer = null; } - this.clearKeepaliveTimeout(); } private removeActiveCall(call: Http2SubchannelCall) { @@ -534,7 +533,7 @@ class Http2Transport implements Transport { * error here. */ try { - http2Stream = this.session!.request(headers); + http2Stream = this.session.request(headers); } catch (e) { this.handleDisconnect(); throw e;