Skip to content

Commit

Permalink
Merge pull request #2836 from murgatroid99/grpc-js_1.12_1.11_bugfix_m…
Browse files Browse the repository at this point in the history
…erge

grpc-js: Port bugfixes from 1.11.x into 1.12.x
  • Loading branch information
murgatroid99 authored Oct 8, 2024
2 parents 8aacdfd + 65cd9b6 commit dce2272
Show file tree
Hide file tree
Showing 8 changed files with 84 additions and 78 deletions.
2 changes: 1 addition & 1 deletion packages/grpc-js/package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "@grpc/grpc-js",
"version": "1.12.0",
"version": "1.12.1",
"description": "gRPC Library for Node - pure JS implementation",
"homepage": "https://grpc.io/",
"repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",
Expand Down
22 changes: 12 additions & 10 deletions packages/grpc-js/src/internal-channel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import {
getDefaultAuthority,
mapUriDefaultScheme,
} from './resolver';
import { trace } from './logging';
import { trace, isTracerEnabled } from './logging';
import { SubchannelAddress } from './subchannel-address';
import { mapProxyName } from './http_proxy';
import { GrpcUri, parseUri, uriToString } from './uri-parser';
Expand Down Expand Up @@ -426,15 +426,17 @@ export class InternalChannel {
JSON.stringify(options, undefined, 2)
);
const error = new Error();
trace(
LogVerbosity.DEBUG,
'channel_stacktrace',
'(' +
this.channelzRef.id +
') ' +
'Channel constructed \n' +
error.stack?.substring(error.stack.indexOf('\n') + 1)
);
if (isTracerEnabled('channel_stacktrace')){
trace(
LogVerbosity.DEBUG,
'channel_stacktrace',
'(' +
this.channelzRef.id +
') ' +
'Channel constructed \n' +
error.stack?.substring(error.stack.indexOf('\n') + 1)
);
}
this.lastActivityTimestamp = new Date();
}

Expand Down
95 changes: 33 additions & 62 deletions packages/grpc-js/src/load-balancer-pick-first.ts
Original file line number Diff line number Diff line change
Expand Up @@ -214,8 +214,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {
*/
private connectionDelayTimeout: NodeJS.Timeout;

private triedAllSubchannels = false;

/**
* The LB policy enters sticky TRANSIENT_FAILURE mode when all
* subchannels have failed to connect at least once, and it stays in that
Expand All @@ -226,12 +224,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {

private reportHealthStatus: boolean;

/**
* Indicates whether we called channelControlHelper.requestReresolution since
* the last call to updateAddressList
*/
private requestedResolutionSinceLastUpdate = false;

/**
* The most recent error reported by any subchannel as it transitioned to
* TRANSIENT_FAILURE.
Expand Down Expand Up @@ -261,6 +253,10 @@ export class PickFirstLoadBalancer implements LoadBalancer {
return this.children.every(child => child.hasReportedTransientFailure);
}

private resetChildrenReportedTF() {
this.children.every(child => child.hasReportedTransientFailure = false);
}

private calculateAndReportNewState() {
if (this.currentPick) {
if (this.reportHealthStatus && !this.currentPick.isHealthy()) {
Expand Down Expand Up @@ -293,23 +289,15 @@ export class PickFirstLoadBalancer implements LoadBalancer {
}

private requestReresolution() {
this.requestedResolutionSinceLastUpdate = true;
this.channelControlHelper.requestReresolution();
}

private maybeEnterStickyTransientFailureMode() {
if (!this.allChildrenHaveReportedTF()) {
return;
}
if (!this.requestedResolutionSinceLastUpdate) {
/* Each time we get an update we reset each subchannel's
* hasReportedTransientFailure flag, so the next time we get to this
* point after that, each subchannel has reported TRANSIENT_FAILURE
* at least once since then. That is the trigger for requesting
* reresolution, whether or not the LB policy is already in sticky TF
* mode. */
this.requestReresolution();
}
this.requestReresolution();
this.resetChildrenReportedTF();
if (this.stickyTransientFailureMode) {
this.calculateAndReportNewState();
return;
Expand All @@ -323,21 +311,16 @@ export class PickFirstLoadBalancer implements LoadBalancer {

private removeCurrentPick() {
if (this.currentPick !== null) {
/* Unref can cause a state change, which can cause a change in the value
* of this.currentPick, so we hold a local reference to make sure that
* does not impact this function. */
const currentPick = this.currentPick;
this.currentPick = null;
currentPick.unref();
currentPick.removeConnectivityStateListener(this.subchannelStateListener);
this.currentPick.removeConnectivityStateListener(this.subchannelStateListener);
this.channelControlHelper.removeChannelzChild(
currentPick.getChannelzRef()
this.currentPick.getChannelzRef()
);
if (this.reportHealthStatus) {
currentPick.removeHealthStateWatcher(
this.pickedSubchannelHealthListener
);
}
this.currentPick.removeHealthStateWatcher(
this.pickedSubchannelHealthListener
);
// Unref last, to avoid triggering listeners
this.currentPick.unref();
this.currentPick = null;
}
}

Expand Down Expand Up @@ -377,9 +360,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {

private startNextSubchannelConnecting(startIndex: number) {
clearTimeout(this.connectionDelayTimeout);
if (this.triedAllSubchannels) {
return;
}
for (const [index, child] of this.children.entries()) {
if (index >= startIndex) {
const subchannelState = child.subchannel.getConnectivityState();
Expand All @@ -392,7 +372,6 @@ export class PickFirstLoadBalancer implements LoadBalancer {
}
}
}
this.triedAllSubchannels = true;
this.maybeEnterStickyTransientFailureMode();
}

Expand Down Expand Up @@ -421,20 +400,25 @@ export class PickFirstLoadBalancer implements LoadBalancer {
this.connectionDelayTimeout.unref?.();
}

/**
* Declare that the specified subchannel should be used to make requests.
* This functions the same independent of whether subchannel is a member of
* this.children and whether it is equal to this.currentPick.
* Prerequisite: subchannel.getConnectivityState() === READY.
* @param subchannel
*/
private pickSubchannel(subchannel: SubchannelInterface) {
if (this.currentPick && subchannel.realSubchannelEquals(this.currentPick)) {
return;
}
trace('Pick subchannel with address ' + subchannel.getAddress());
this.stickyTransientFailureMode = false;
this.removeCurrentPick();
this.currentPick = subchannel;
/* Ref before removeCurrentPick and resetSubchannelList to avoid the
* refcount dropping to 0 during this process. */
subchannel.ref();
if (this.reportHealthStatus) {
subchannel.addHealthStateWatcher(this.pickedSubchannelHealthListener);
}
this.channelControlHelper.addChannelzChild(subchannel.getChannelzRef());
this.removeCurrentPick();
this.resetSubchannelList();
subchannel.addConnectivityStateListener(this.subchannelStateListener);
subchannel.addHealthStateWatcher(this.pickedSubchannelHealthListener);
this.currentPick = subchannel;
clearTimeout(this.connectionDelayTimeout);
this.calculateAndReportNewState();
}
Expand All @@ -451,20 +435,11 @@ export class PickFirstLoadBalancer implements LoadBalancer {

private resetSubchannelList() {
for (const child of this.children) {
if (
!(
this.currentPick &&
child.subchannel.realSubchannelEquals(this.currentPick)
)
) {
/* The connectivity state listener is the same whether the subchannel
* is in the list of children or it is the currentPick, so if it is in
* both, removing it here would cause problems. In particular, that
* always happens immediately after the subchannel is picked. */
child.subchannel.removeConnectivityStateListener(
this.subchannelStateListener
);
}
/* Always remoev the connectivity state listener. If the subchannel is
getting picked, it will be re-added then. */
child.subchannel.removeConnectivityStateListener(
this.subchannelStateListener
);
/* Refs are counted independently for the children list and the
* currentPick, so we call unref whether or not the child is the
* currentPick. Channelz child references are also refcounted, so
Expand All @@ -476,20 +451,16 @@ export class PickFirstLoadBalancer implements LoadBalancer {
}
this.currentSubchannelIndex = 0;
this.children = [];
this.triedAllSubchannels = false;
this.requestedResolutionSinceLastUpdate = false;
}

private connectToAddressList(addressList: SubchannelAddress[]) {
trace('connectToAddressList([' + addressList.map(address => subchannelAddressToString(address)) + '])');
const newChildrenList = addressList.map(address => ({
subchannel: this.channelControlHelper.createSubchannel(address, {}, null),
hasReportedTransientFailure: false,
}));
trace('connectToAddressList([' + addressList.map(address => subchannelAddressToString(address)) + '])');
for (const { subchannel } of newChildrenList) {
if (subchannel.getConnectivityState() === ConnectivityState.READY) {
this.channelControlHelper.addChannelzChild(subchannel.getChannelzRef());
subchannel.addConnectivityStateListener(this.subchannelStateListener);
this.pickSubchannel(subchannel);
return;
}
Expand Down
7 changes: 7 additions & 0 deletions packages/grpc-js/src/load-balancer-round-robin.ts
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,13 @@ export class RoundRobinLoadBalancer implements LoadBalancer {
channelControlHelper,
{
updateState: (connectivityState, picker) => {
/* Ensure that name resolution is requested again after active
* connections are dropped. This is more aggressive than necessary to
* accomplish that, so we are counting on resolvers to have
* reasonable rate limits. */
if (this.currentState === ConnectivityState.READY && connectivityState !== ConnectivityState.READY) {
this.channelControlHelper.requestReresolution();
}
this.calculateAndUpdateState();
},
}
Expand Down
2 changes: 1 addition & 1 deletion packages/grpc-js/src/resolver-dns.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import {
registerResolver,
registerDefaultScheme,
} from './resolver';
import { promises as dns } from 'node:dns';
import { promises as dns } from 'dns';
import { extractAndSelectServiceConfig, ServiceConfig } from './service-config';
import { Status } from './constants';
import { StatusObject } from './call-interface';
Expand Down
2 changes: 1 addition & 1 deletion packages/grpc-js/src/retrying-call.ts
Original file line number Diff line number Diff line change
Expand Up @@ -398,7 +398,7 @@ export class RetryingCall implements Call, DeadlineInfoProvider {
return list.some(
value =>
value === code ||
value.toString().toLowerCase() === Status[code].toLowerCase()
value.toString().toLowerCase() === Status[code]?.toLowerCase()
);
}

Expand Down
16 changes: 13 additions & 3 deletions packages/grpc-js/src/transport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -223,9 +223,8 @@ class Http2Transport implements Transport {
);

session.once('error', error => {
/* Do nothing here. Any error should also trigger a close event, which is
* where we want to handle that. */
this.trace('connection closed with error ' + (error as Error).message);
this.handleDisconnect();
});

if (logging.isTracerEnabled(TRACER_NAME)) {
Expand Down Expand Up @@ -383,6 +382,9 @@ class Http2Transport implements Transport {
* Handle connection drops, but not GOAWAYs.
*/
private handleDisconnect() {
if (this.disconnectHandled) {
return;
}
this.clearKeepaliveTimeout();
this.reportDisconnectToOwner(false);
/* Give calls an event loop cycle to finish naturally before reporting the
Expand Down Expand Up @@ -773,6 +775,7 @@ export class Http2SubchannelConnector implements SubchannelConnector {
);
this.session = session;
let errorMessage = 'Failed to connect';
let reportedError = false;
session.unref();
session.once('connect', () => {
session.removeAllListeners();
Expand All @@ -783,12 +786,19 @@ export class Http2SubchannelConnector implements SubchannelConnector {
this.session = null;
// Leave time for error event to happen before rejecting
setImmediate(() => {
reject(`${errorMessage} (${new Date().toISOString()})`);
if (!reportedError) {
reportedError = true;
reject(`${errorMessage} (${new Date().toISOString()})`);
}
});
});
session.once('error', error => {
errorMessage = (error as Error).message;
this.trace('connection failed with error ' + errorMessage);
if (!reportedError) {
reportedError = true;
reject(`${errorMessage} (${new Date().toISOString()})`);
}
});
});
}
Expand Down
16 changes: 16 additions & 0 deletions packages/grpc-js/test/test-retry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,22 @@ describe('Retries', () => {
}
);
});

it('Should not retry on custom error code', done => {
const metadata = new grpc.Metadata();
metadata.set('succeed-on-retry-attempt', '2');
metadata.set('respond-with-status', '300');
client.echo(
{ value: 'test value', value2: 3 },
metadata,
(error: grpc.ServiceError, response: any) => {
assert(error);
assert.strictEqual(error.code, 300);
assert.strictEqual(error.details, 'Failed on retry 0');
done();
}
);
});
});

describe('Client with hedging configured', () => {
Expand Down

0 comments on commit dce2272

Please sign in to comment.