Help with retry interceptor

[yiningzhao]

Hope someone can help me figure out why my interceptor implementation is not working.

I'm trying to add a retry interceptor for our gPRC client. Here's the implementation:

export const getNewRetryInterceptor = (): RpcInterceptor => ({
  interceptUnary(
    next: NextUnaryFn,
    method: MethodInfo<any, any>,
    input: object,
    options: RpcOptions,
  ): UnaryCall {
    let retries = 0;
    const maxRetries = 5;

    let result = next(method, input, options);

    const handledResponse = (async () => {
      while (retries < maxRetries) {
        try {
          return await result.response;
        } catch (err) {
          if (
            !(
              err instanceof RpcError &&
              err.code == GrpcStatus[GrpcStatus.UNAVAILABLE]
            )
          ) {
            throw err;
          }

          console.warn(
            `gRPC connection unavailable, retrying after 1s... (Retry ${
              retries + 1
            } of ${maxRetries})`,
          );
          await sleep(1000);
          result = next(method, input, options);
          retries++;
        }
      }
    })();

    // result.response is not read only.
    // @ts-ignore
    result.response = handledResponse;
    return result;
  },
});

This is the only interceptor for our gPRC client. When I tested it, I saw the retries happening in handleResponse but the gPRC error was returned before all retries are finished:

  console.warn
    gRPC connection unavailable, retrying after 1s... (Retry 1 of 5)

      92 |           }
      93 |
    > 94 |           console.warn(
         |                   ^
      95 |             `gRPC connection unavailable, retrying after 1s... (Retry ${
      96 |               retries + 1
      97 |             } of ${maxRetries})`,

      at src/util/retry.ts:94:19
          at async Promise.all (index 1)

  console.warn
    gRPC connection unavailable, retrying after 1s... (Retry 2 of 5)

      92 |           }
      93 |
    > 94 |           console.warn(
         |                   ^
      95 |             `gRPC connection unavailable, retrying after 1s... (Retry ${
      96 |               retries + 1
      97 |             } of ${maxRetries})`,

      at src/util/retry.ts:94:19
          at async Promise.all (index 1)

 FAIL  src/tests/orchestrator.test.ts (7.099 s)

    RpcError: 14 UNAVAILABLE: test

      at Object.callback (node_modules/@protobuf-ts/grpc-transport/build/commonjs/grpc-transport.js:37:27)
      at Object.onReceiveStatus (node_modules/@grpc/grpc-js/src/client.ts:352:26)
      at Object.onReceiveStatus (node_modules/@grpc/grpc-js/src/client-interceptors.ts:462:34)
      at Object.onReceiveStatus (node_modules/@grpc/grpc-js/src/client-interceptors.ts:424:48)
      at node_modules/@grpc/grpc-js/src/call-stream.ts:330:24

Test Suites: 1 failed, 1 total
Tests:       1 failed, 1 skipped, 2 total
Snapshots:   0 total
Time:        7.194 s
Ran all test suites matching ***[redacted]
  console.warn
    gRPC connection unavailable, retrying after 1s... (Retry 3 of 5)

      92 |           }
      93 |
    > 94 |           console.warn(
         |                   ^
      95 |             `gRPC connection unavailable, retrying after 1s... (Retry ${
      96 |               retries + 1
      97 |             } of ${maxRetries})`,

      at src/util/retry.ts:94:19
          at async Promise.all (index 1)

Jest did not exit one second after the test run has completed.

This usually means that there are asynchronous operations that weren't stopped in your tests. Consider running Jest with `--detectOpenHandles` to troubleshoot this issue.
  console.warn
    gRPC connection unavailable, retrying after 1s... (Retry 4 of 5)

      92 |           }
      93 |
    > 94 |           console.warn(
         |                   ^
      95 |             `gRPC connection unavailable, retrying after 1s... (Retry ${
      96 |               retries + 1
      97 |             } of ${maxRetries})`,

      at src/util/retry.ts:94:19
          at async Promise.all (index 1)

  console.warn
    gRPC connection unavailable, retrying after 1s... (Retry 5 of 5)

      92 |           }
      93 |
    > 94 |           console.warn(
         |                   ^
      95 |             `gRPC connection unavailable, retrying after 1s... (Retry ${
      96 |               retries + 1
      97 |             } of ${maxRetries})`,

      at src/util/retry.ts:94:19
          at async Promise.all (index 1)

[yiningzhao]

@timostamm excuse me for the direct ping. I'd really appreciate it if you could give me some advice on this.

[jcready]

I believe this will require a bit more complexity as you'll need to manage the UnaryCall promises yourself:

import {
  Deferred,
  RpcError,
  RpcInterceptor,
  RpcMetadata,
  RpcStatus,
  UnaryCall,
} from '@protobuf-ts/runtime-rpc';

export const getNewRetryInterceptor = (): RpcInterceptor => ({
  interceptUnary(next, method, input, options) {
    const maxRetries = 5,
      defHeader = new Deferred<RpcMetadata>(),
      defMessage = new Deferred<object>(),
      defStatus = new Deferred<RpcStatus>(),
      defTrailer = new Deferred<RpcMetadata>();

    let retries = 0,
      result = next(method, input, options);

    void (async () => {
      while (true) {
        try {
          await result;
          defHeader.resolve(result.headers);
          defMessage.resolve(result.response);
          defStatus.resolve(result.status);
          defTrailer.resolve(result.trailers);
        } catch (err) {
          if (
            err instanceof RpcError &&
            err.code == GrpcStatus[GrpcStatus.UNAVAILABLE] &&
            ++retries <= maxRetries
          ) {
            console.warn(
              `gRPC connection unavailable, retrying after 1s... (Retry ${retries} of ${maxRetries})`
            );
            await sleep(1000);
            result = next(method, input, options);
            continue;
          }
          defHeader.rejectPending(err);
          defMessage.rejectPending(err);
          defStatus.rejectPending(err);
          defTrailer.rejectPending(err);
          break;
        }
      }
    })();

    return new UnaryCall(
      method,
      options.meta ?? {},
      input,
      defHeader.promise,
      defMessage.promise,
      defStatus.promise,
      defTrailer.promise
    );
  },
});

[daudfauzy98]

Can I change rejectPending() to just reject()?

[jcready]

You could, but why risk it? rejectPending() checks to see if the promise has already been resolved/rejected before rejecting, where as reject() does the same check, but throws if it was already resolved/rejected.

[thegenius]

I believe this will require a bit more complexity as you'll need to manage the UnaryCall promises yourself:

import {
  Deferred,
  RpcError,
  RpcInterceptor,
  RpcMetadata,
  RpcStatus,
  UnaryCall,
} from '@protobuf-ts/runtime-rpc';

export const getNewRetryInterceptor = (): RpcInterceptor => ({
  interceptUnary(next, method, input, options) {
    const maxRetries = 5,
      defHeader = new Deferred<RpcMetadata>(),
      defMessage = new Deferred<object>(),
      defStatus = new Deferred<RpcStatus>(),
      defTrailer = new Deferred<RpcMetadata>();

    let retries = 0,
      result = next(method, input, options);

    void (async () => {
      while (true) {
        try {
          await result;
          defHeader.resolve(result.headers);
          defMessage.resolve(result.response);
          defStatus.resolve(result.status);
          defTrailer.resolve(result.trailers);
        } catch (err) {
          if (
            err instanceof RpcError &&
            err.code == GrpcStatus[GrpcStatus.UNAVAILABLE] &&
            ++retries <= maxRetries
          ) {
            console.warn(
              `gRPC connection unavailable, retrying after 1s... (Retry ${retries} of ${maxRetries})`
            );
            await sleep(1000);
            result = next(method, input, options);
            continue;
          }
          defHeader.rejectPending(err);
          defMessage.rejectPending(err);
          defStatus.rejectPending(err);
          defTrailer.rejectPending(err);
          break;
        }
      }
    })();

    return new UnaryCall(
      method,
      options.meta ?? {},
      input,
      defHeader.promise,
      defMessage.promise,
      defStatus.promise,
      defTrailer.promise
    );
  },
});

This code help me a lot, Thank you for your share.

[amjadfqs]

@jcready , sorry for the mention. But how to do this for also streaming.

[jcready]

Do you mean server streaming, client streaming, or duplex? For server streaming you have a few options here.

1. The interceptor buffers everything it receives (per try) until a try is completed successfully. At which point it flushes all of the buffered messages to the consumer at once.
2. The interceptor forwards every message it receives on every try. This means the consumer could end up receiving a duplicate message per try (at least once delivery, not only once delivery).
3. The interceptor only retries if and only if the failure happens before a single message is received.

The first option can be very memory intensive as you're effectively turning a stream into a buffer. The second option is probably not very useful without some kind of identifier on each message to deduplicate. So let's focus on the third option which, while it doesn't cover retrying for every situation, it should at least cover the case where the service is initially UNAVAILABLE.

import {
  Deferred,
  RpcError,
  RpcInterceptor,
  RpcMetadata,
  RpcOutputStreamController,
  RpcStatus,
  ServerStreamingCall,
} from '@protobuf-ts/runtime-rpc';

export const getNewRetryInterceptor = (): RpcInterceptor => ({
  interceptServerStreaming(next, method, input, options) {
    const maxRetries = 5,
      defHeader = new Deferred<RpcMetadata>(),
      responseStream = new RpcOutputStreamController<object>(),
      defStatus = new Deferred<RpcStatus>(),
      defTrailer = new Deferred<RpcMetadata>();

    let retries = 0,
      result = next(method, input, options);

    void (async () => {
      let gotMessage = false;
      while (true) {
        try {
          // Forward messages to responseStream, keep track of if we got any messages
          result.responses.onMessage((m) => {
            gotMessage = true;
            responseStream.notifyMessage(m);
          });
          // Wait for the result (headers, status, trailers)
          await result;
          defHeader.resolve(result.headers);
          responseStream.notifyComplete();
          defStatus.resolve(result.status);
          defTrailer.resolve(result.trailers);
        } catch (err) {
          if (
            !gotMessage &&
            err instanceof RpcError &&
            err.code == GrpcStatus[GrpcStatus.UNAVAILABLE] &&
            ++retries <= maxRetries
          ) {
            console.warn(
              `gRPC connection unavailable, retrying after 1s... (Retry ${retries} of ${maxRetries})`
            );
            await sleep(1000);
            result = next(method, input, options);
            continue;
          }
          defHeader.rejectPending(err);
          responseStream.notifyError(err instanceof Error ? err : new Error(String(err)));
          defStatus.rejectPending(err);
          defTrailer.rejectPending(err);
          break;
        }
      }
    })();

    return new ServerStreamingCall(
      method,
      options.meta ?? {},
      input,
      defHeader.promise,
      responseStream,
      defStatus.promise,
      defTrailer.promise
    );
  },
});