wu-sheng commented on code in PR #141:
URL: https://github.com/apache/skywalking-nodejs/pull/141#discussion_r3532611856
##########
src/agent/core/remote/TraceSegmentServiceClient.ts:
##########
@@ -165,44 +166,89 @@ export default class TraceSegmentServiceClient implements
BootService, GRPCChann
return;
}
+ let snapshots: Segment[] = [];
+ let settled = false;
let stream: ReturnType<TraceSegmentReportServiceClient['collect']> |
undefined;
- try {
- stream = this.reporterClient.collect(
- new grpc.Metadata(),
- { deadline: Date.now() + config.traceTimeout },
- (error) => {
- if (error) {
- logReportError('Failed to report trace data', error);
- this.reportGrpcError(error);
- }
- resolve();
- },
- );
+ let writesStarted = false;
- for (const segment of this.buffer) {
- if (segment) {
- if (logger._isDebugEnabled) {
- logger.debug('Sending segment ', { segment });
- }
- stream.write(segment.transform());
+ const settle = (error?: unknown): void => {
+ if (settled) {
+ return;
+ }
+ settled = true;
+ if (error != null) {
+ logReportError('Failed to report trace data', error);
+ this.reportGrpcError(error);
+ if (!this.closed && !writesStarted) {
+ this.requeueSegments(snapshots);
}
}
- } catch (error) {
- logReportError('Failed to report trace data', error);
- this.reportGrpcError(error);
resolve();
- } finally {
- this.buffer.length = 0;
+ };
+
+ void (async () => {
try {
- stream?.end();
+ snapshots = this.buffer.splice(0, this.buffer.length);
+
+ stream = this.reporterClient!.collect(
+ new grpc.Metadata(),
+ { deadline: grpcUpstreamDeadlineMs() },
+ (error, commands) => {
+ if (error) {
+ settle(error);
+ } else {
+
ServiceManager.INSTANCE.findService(CommandService)?.receiveCommand(commands);
+ settle();
+ }
+ },
+ );
+
+ for (const segment of snapshots) {
+ if (this.closed) {
+ break;
+ }
+ if (segment) {
+ if (logger._isDebugEnabled) {
+ logger.debug(
+ 'Sending segment traceId=%s segmentId=%s spanCount=%d',
+ segment.relatedTraces[0]?.toString?.() ?? 'unknown',
+ segment.segmentId.toString(),
+ segment.spans.length,
+ );
+ }
+ let writeAccepted = stream.write(segment.transform());
+ writesStarted = true;
+ while (writeAccepted === false && stream && !this.closed) {
+ await new Promise<void>((resolveDrain) =>
stream!.once('drain', resolveDrain));
Review Comment:
Thanks for the fast turnaround — the other 12 are correctly addressed, and
the new tests on the `unhandledRejection` / `Infinity`-deadline /
shutdown-during-preload paths are great.
One heads-up on **this** one: I don't think the fix actually unblocks the
await, because grpc-js never emits any of the three raced events on a
*client-streaming* writable.
- `collect` is client-streaming, and `client.js` `makeClientStreamRequest`
delivers the terminal error via the **unary callback** and emits only
`'metadata'` and `'status'` on the writable — it never emits `'error'` or
`'close'` on the stream itself.
- On terminal status, `retrying-call.js` `reportStatus` does
`this.writeBuffer = []` **without** invoking the pending `_write` callbacks
(they're only fired from `handleChildWriteCompleted` on a *successful* child
send). So the backpressured write never completes and `'drain'` never fires
either.
So in the exact hang scenario — `stream.write()` returns `false` under
HTTP/2 flow-control backpressure, then the call deadlines/fails — none of
`drain`/`error`/`close` fire and the `Promise.race` still waits forever;
`shutdown()` can't wake it either (the `if (this.closed) break` after the race
is unreachable while the race is pending).
A bounded escape is the reliable fix. The call already carries the upstream
deadline, so e.g.:
```ts
const drainDeadlineMs = config.grpcUpstreamTimeout ?
config.grpcUpstreamTimeout * 1000 : 30000;
let drained = false;
await Promise.race([
new Promise<void>((r) => stream!.once('drain', () => { drained = true;
r(); })),
new Promise<void>((r) => { const t = setTimeout(r, drainDeadlineMs);
t.unref(); }),
]);
if (this.closed || !drained) break; // give up this tick on
timeout/shutdown; stream.end() still runs after the loop
```
Or simplest, mirror `MeterSender`: on `writeAccepted === false`, log +
`break` and drop the remainder of this tick's segments.
(Low severity — needs sustained send-side backpressure plus a mid-stream
failure — but the current fix subscribes to events that don't fire for this
call type, so the behavior is unchanged from before.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
