fix(grpc): recreate stuck clients before retries and fix reflection client recreation - #190
Open
SeqviriouM wants to merge 2 commits into
Open
fix(grpc): recreate stuck clients before retries and fix reflection client recreation#190SeqviriouM wants to merge 2 commits into
SeqviriouM wants to merge 2 commits into
Conversation
Reviewer's GuideAdjusts gRPC client recreation and retry behavior to avoid reusing stuck channels, ensures reflection clients are recreated and closed correctly, and introduces connectivity-aware error handling and caching utilities while bumping @grpc/grpc-js. File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
Fixed security issues:
-
In the
getServiceretry error branch you callclearReflectionClientCache(actionEndpoint), but in the dynamic-endpoint caseactionEndpointis not defined in this scope; consider usinggetActionEndpoint(args)(wrapped in a try/catch as you did earlier) or reusingclearServiceCacheto avoid a potentialReferenceErrorand keep the behavior consistent. -
clearReflectionClientCachecloses reflection clients immediately, while other client shutdown paths usecloseServiceClientwith a grace period; consider reusingcloseServiceClient(and/orCLIENT_CLOSE_GRACE_MS) here to avoid abruptly terminating any in-flight reflection calls.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In the `getService` retry error branch you call `clearReflectionClientCache(actionEndpoint)`, but in the dynamic-endpoint case `actionEndpoint` is not defined in this scope; consider using `getActionEndpoint(args)` (wrapped in a try/catch as you did earlier) or reusing `clearServiceCache` to avoid a potential `ReferenceError` and keep the behavior consistent.
- `clearReflectionClientCache` closes reflection clients immediately, while other client shutdown paths use `closeServiceClient` with a grace period; consider reusing `closeServiceClient` (and/or `CLIENT_CLOSE_GRACE_MS`) here to avoid abruptly terminating any in-flight reflection calls.
## Individual Comments
### Comment 1
<location path="lib/components/grpc.ts" line_range="408" />
<code_context>
+ * the cache already holds another client (e.g. a concurrent request has
+ * re-created it earlier).
+ */
function clearInstancesCache<Context extends GatewayContext>(
service: ServiceClient,
instancesMap: typeof serviceInstancesMap | typeof reflectionServiceInstancesMap,
</code_context>
<issue_to_address>
**issue (complexity):** Consider splitting cache clearing into separate reflection and non-reflection helpers and moving reflection-specific logic into grpc-reflection to simplify control flow and remove symbol-based coupling.
You can recover a lot of readability by de‑genericizing the cache clearing and pulling the reflection‑specific lifecycle into its own helpers, without losing any of the new behavior.
### 1. Split `clearInstancesCache` by cache type
Right now `clearInstancesCache` must understand both:
- `serviceInstancesMap: Record<string, Record<string, ServiceClient>>`
- `reflectionServiceInstancesMap: Record<string, Record<string, Promise<ServiceClient>>>`
and uses `serviceCachePromiseSymbol` + `markCachedServicePromise` to reconcile them. You can keep the semantics but make each function operate on a single, known shape and drop the symbol tagging entirely.
For example:
```ts
// non-reflection: cache stores concrete clients
function clearServiceInstance<Context extends GatewayContext>(
service: ServiceClient,
protoKey: string,
actionEndpoint: string,
closeTimeout: number,
ctx: Context,
): boolean {
const cachePath: [string, string] = [protoKey, actionEndpoint];
const cached = _.get(serviceInstancesMap, cachePath);
if (cached !== service) {
ctx.log(`Service client not matched cached service for cachePath '${cachePath}'`);
return false;
}
_.unset(serviceInstancesMap, cachePath);
closeServiceClient(service, closeTimeout, (error) => {
ctx.logError('Failed to close connection during clearing instances cache', error, {
cachePath,
});
});
return true;
}
```
```ts
// reflection: cache stores Promise<ServiceClient>
async function clearReflectionServiceInstance<Context extends GatewayContext>(
service: ServiceClient,
protoKey: string,
actionEndpoint: string,
closeTimeout: number,
ctx: Context,
): Promise<boolean> {
const cachePath: [string, string] = [protoKey, actionEndpoint];
const cachedPromise = _.get(reflectionServiceInstancesMap, cachePath);
if (!cachedPromise) {
ctx.log(`No reflection client in cache for cachePath '${cachePath}'`);
return false;
}
let cachedClient: ServiceClient | undefined;
try {
cachedClient = await Promise.race([
cachedPromise,
// cheap safeguard: don't hang forever on a never‑resolving promise
new Promise<ServiceClient>((_, reject) =>
setTimeout(() => reject(new Error('Reflection client promise timeout')), 0),
),
]);
} catch {
// if it never resolved or failed, we still want to drop the cache entry
}
if (cachedClient && cachedClient !== service) {
ctx.log(
`Reflection service client not matched cached promise for cachePath '${cachePath}'`,
);
return false;
}
_.unset(reflectionServiceInstancesMap, cachePath);
closeServiceClient(service, closeTimeout, (error) => {
ctx.logError(
'Failed to close connection during clearing reflection instances cache',
error,
{cachePath},
);
});
return true;
}
```
This removes:
- `serviceCachePromiseSymbol`
- `ServiceClientWithCacheRef`
- `markCachedServicePromise`
- The “sometimes a client, sometimes a promise” semantics in `clearInstancesCache`.
### 2. Inline `clearServiceCache` into explicit recreate paths
With specialized clear functions, `clearServiceCache` becomes thin indirection. Making recreate behavior explicit per branch is easier to follow and keeps the reflection‑specific behavior local.
For example:
```ts
if (typeof config.endpoint === 'function') {
// ...
recreateService = async (service, closeTimeout, ctx, args) => {
const actionEndpoint = getActionEndpoint(args);
if ('reflection' in config) {
const cleared = await clearReflectionServiceInstance(
service,
config.protoKey,
actionEndpoint,
closeTimeout,
ctx,
);
if (cleared) {
clearReflectionClientCache(actionEndpoint);
}
} else {
clearServiceInstance(service, config.protoKey, actionEndpoint, closeTimeout, ctx);
}
};
} else if (endpoints) {
// ...
recreateService = async (service, closeTimeout, ctx) => {
if ('reflection' in config) {
const cleared = await clearReflectionServiceInstance(
service,
config.protoKey,
actionEndpoint,
closeTimeout,
ctx,
);
if (cleared) {
clearReflectionClientCache(actionEndpoint);
}
} else {
clearServiceInstance(service, config.protoKey, actionEndpoint, closeTimeout, ctx);
}
};
}
```
This keeps:
- Synchronous cache invalidation for the non‑reflection clients.
- Reflection client cache tied to service client recreation.
- The “close with grace period” behavior via `closeServiceClient`.
### 3. Move reflection‑specific helpers out of `grpc.ts`
Given you already have `grpc-reflection.ts`, the reflection lifecycle is a good candidate to encapsulate there:
- `getServiceInstanceReflectCached`
- `clearReflectionServiceInstance` (from above)
- `clearReflectionClientCache`
Then `grpc.ts` only depends on a small, reflection‑aware interface instead of knowing about promise caches and reflection client coupling:
```ts
// grpc.ts
import {
getServiceInstanceReflectCached,
clearReflectionServiceInstance,
clearReflectionClientCache,
} from '../utils/grpc-reflection';
```
This addresses the current cross‑cutting concerns: all reflection‑only details (promise cache, coupling between reflection client and service client) are in one place.
---
These changes keep all the new behaviors (no leaked connections, synchronous cache clearing, reflection cache tied to client recreation, channel‑state‑aware retries) but substantially reduce the layers of indirection and remove the symbol‑based coupling between creation, storage, and clearing.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| * the cache already holds another client (e.g. a concurrent request has | ||
| * re-created it earlier). | ||
| */ | ||
| function clearInstancesCache<Context extends GatewayContext>( |
There was a problem hiding this comment.
issue (complexity): Consider splitting cache clearing into separate reflection and non-reflection helpers and moving reflection-specific logic into grpc-reflection to simplify control flow and remove symbol-based coupling.
You can recover a lot of readability by de‑genericizing the cache clearing and pulling the reflection‑specific lifecycle into its own helpers, without losing any of the new behavior.
1. Split clearInstancesCache by cache type
Right now clearInstancesCache must understand both:
serviceInstancesMap: Record<string, Record<string, ServiceClient>>reflectionServiceInstancesMap: Record<string, Record<string, Promise<ServiceClient>>>
and uses serviceCachePromiseSymbol + markCachedServicePromise to reconcile them. You can keep the semantics but make each function operate on a single, known shape and drop the symbol tagging entirely.
For example:
// non-reflection: cache stores concrete clients
function clearServiceInstance<Context extends GatewayContext>(
service: ServiceClient,
protoKey: string,
actionEndpoint: string,
closeTimeout: number,
ctx: Context,
): boolean {
const cachePath: [string, string] = [protoKey, actionEndpoint];
const cached = _.get(serviceInstancesMap, cachePath);
if (cached !== service) {
ctx.log(`Service client not matched cached service for cachePath '${cachePath}'`);
return false;
}
_.unset(serviceInstancesMap, cachePath);
closeServiceClient(service, closeTimeout, (error) => {
ctx.logError('Failed to close connection during clearing instances cache', error, {
cachePath,
});
});
return true;
}// reflection: cache stores Promise<ServiceClient>
async function clearReflectionServiceInstance<Context extends GatewayContext>(
service: ServiceClient,
protoKey: string,
actionEndpoint: string,
closeTimeout: number,
ctx: Context,
): Promise<boolean> {
const cachePath: [string, string] = [protoKey, actionEndpoint];
const cachedPromise = _.get(reflectionServiceInstancesMap, cachePath);
if (!cachedPromise) {
ctx.log(`No reflection client in cache for cachePath '${cachePath}'`);
return false;
}
let cachedClient: ServiceClient | undefined;
try {
cachedClient = await Promise.race([
cachedPromise,
// cheap safeguard: don't hang forever on a never‑resolving promise
new Promise<ServiceClient>((_, reject) =>
setTimeout(() => reject(new Error('Reflection client promise timeout')), 0),
),
]);
} catch {
// if it never resolved or failed, we still want to drop the cache entry
}
if (cachedClient && cachedClient !== service) {
ctx.log(
`Reflection service client not matched cached promise for cachePath '${cachePath}'`,
);
return false;
}
_.unset(reflectionServiceInstancesMap, cachePath);
closeServiceClient(service, closeTimeout, (error) => {
ctx.logError(
'Failed to close connection during clearing reflection instances cache',
error,
{cachePath},
);
});
return true;
}This removes:
serviceCachePromiseSymbolServiceClientWithCacheRefmarkCachedServicePromise- The “sometimes a client, sometimes a promise” semantics in
clearInstancesCache.
2. Inline clearServiceCache into explicit recreate paths
With specialized clear functions, clearServiceCache becomes thin indirection. Making recreate behavior explicit per branch is easier to follow and keeps the reflection‑specific behavior local.
For example:
if (typeof config.endpoint === 'function') {
// ...
recreateService = async (service, closeTimeout, ctx, args) => {
const actionEndpoint = getActionEndpoint(args);
if ('reflection' in config) {
const cleared = await clearReflectionServiceInstance(
service,
config.protoKey,
actionEndpoint,
closeTimeout,
ctx,
);
if (cleared) {
clearReflectionClientCache(actionEndpoint);
}
} else {
clearServiceInstance(service, config.protoKey, actionEndpoint, closeTimeout, ctx);
}
};
} else if (endpoints) {
// ...
recreateService = async (service, closeTimeout, ctx) => {
if ('reflection' in config) {
const cleared = await clearReflectionServiceInstance(
service,
config.protoKey,
actionEndpoint,
closeTimeout,
ctx,
);
if (cleared) {
clearReflectionClientCache(actionEndpoint);
}
} else {
clearServiceInstance(service, config.protoKey, actionEndpoint, closeTimeout, ctx);
}
};
}This keeps:
- Synchronous cache invalidation for the non‑reflection clients.
- Reflection client cache tied to service client recreation.
- The “close with grace period” behavior via
closeServiceClient.
3. Move reflection‑specific helpers out of grpc.ts
Given you already have grpc-reflection.ts, the reflection lifecycle is a good candidate to encapsulate there:
getServiceInstanceReflectCachedclearReflectionServiceInstance(from above)clearReflectionClientCache
Then grpc.ts only depends on a small, reflection‑aware interface instead of knowing about promise caches and reflection client coupling:
// grpc.ts
import {
getServiceInstanceReflectCached,
clearReflectionServiceInstance,
clearReflectionClientCache,
} from '../utils/grpc-reflection';This addresses the current cross‑cutting concerns: all reflection‑only details (promise cache, coupling between reflection client and service client) are in one place.
These changes keep all the new behaviors (no leaked connections, synchronous cache clearing, reflection cache tied to client recreation, channel‑state‑aware retries) but substantially reduce the layers of indirection and remove the symbol‑based coupling between creation, storage, and clearing.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Ensure gRPC retries use fresh clients when channels are stuck and keep reflection-client caches synchronized.
Bug Fixes:
Enhancements:
Build: