Skip to content

Keepalive bugfixes and unify timers strategies between client and server #2760

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 12 commits into from
Jun 18, 2024
Merged

Keepalive bugfixes and unify timers strategies between client and server #2760

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
ad598ec
Serverside keepalive error detection and cleanups
davidfiala May 28, 2024
334f0dc
remove comment
davidfiala May 28, 2024
d799a7a
unify server and client keepalive matching comments and discussion on…
davidfiala May 29, 2024
577b4b4
add keepalive server trace back in to match channelz vs non-channelz …
davidfiala May 29, 2024
7883164
return imports back to original order
davidfiala May 29, 2024
19cdc12
another missing trace message for parity
davidfiala May 29, 2024
bed5e85
resolve hoisting
davidfiala May 29, 2024
d325b5f
hoist in second location
davidfiala May 29, 2024
a77d94f
Based on grpc/grpc-node#2139 I wrapped http2session.ping in a try-cat…
davidfiala May 29, 2024
c2da436
remove keepaliveDisabled from server.ts. rename keepaliveTimer.
davidfiala May 29, 2024
3c5ab22
per discussion, avoid tracking keepalive disabled state and instead d…
davidfiala Jun 6, 2024
98cd87f
ensure that client keepalive timers are always cleared when they trig…
davidfiala Jun 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
277 changes: 190 additions & 87 deletions packages/grpc-js/src/server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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');
}
Expand Down Expand Up @@ -1376,8 +1384,8 @@ 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 keepaliveDisabled = false;
let sessionClosedByServer = false;

const idleTimeoutObj = this.enableIdleTimeout(session);
Expand Down Expand Up @@ -1420,41 +1428,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 (keepaliveTimeout) {
clearTimeout(keepaliveTimeout);
keepaliveTimeout = 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 (
!keepaliveDisabled &&
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'
);
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);
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;
}

keepaliveTimeout = setTimeout(() => {
clearKeepaliveTimeout();
this.keepaliveTrace('Ping timeout passed without response');
this.trace('Connection dropped by keepalive timeout');
sessionClosedByServer = true;
session.close();
}, this.keepaliveTimeoutMs);
keepaliveTimeout.unref?.();
};

maybeStartKeepalivePingTimer();

session.on('close', () => {
if (!sessionClosedByServer) {
Expand All @@ -1471,12 +1528,8 @@ export class Server {
clearTimeout(connectionAgeGraceTimer);
}

if (keeapliveTimeTimer) {
clearInterval(keeapliveTimeTimer);
if (keepaliveTimeoutTimer) {
clearTimeout(keepaliveTimeoutTimer);
}
}
keepaliveDisabled = true;
clearKeepaliveTimeout();

if (idleTimeoutObj !== null) {
clearTimeout(idleTimeoutObj.timeout);
Expand Down Expand Up @@ -1521,8 +1574,8 @@ 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 keepaliveDisabled = false;
let sessionClosedByServer = false;

const idleTimeoutObj = this.enableIdleTimeout(session);
Expand Down Expand Up @@ -1564,49 +1617,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 (
!keepaliveDisabled &&
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) {
Expand All @@ -1627,12 +1734,8 @@ export class Server {
clearTimeout(connectionAgeGraceTimer);
}

if (keeapliveTimeTimer) {
clearInterval(keeapliveTimeTimer);
if (keepaliveTimeoutTimer) {
clearTimeout(keepaliveTimeoutTimer);
}
}
keepaliveDisabled = true;
clearKeepaliveTimeout();

if (idleTimeoutObj !== null) {
clearTimeout(idleTimeoutObj.timeout);
Expand Down
Loading