Dealing with server restart and streamed responses

[Brammm]

Hi all

Implemented nice-grpc a while ago in our Electron app instead of @grpc/grpc-js. Had some difficulty comprehending how to set up our streamed responses but it ended up working out. One thing we've noticed (with our previous implementation as well) is that when the server restarts, we lose our subscriptions and have to call the setup code again.

This function currently looks like this. It gets called when the Electron app boots and then again when we notice the server has restarted (forgive the abundance of try/catch blocks. We're chasing some weird Electron issues and seeing if these fix anything).

export const setupProgressRecon = (
  reconService: ReconstructionServiceClient,
  webContents: Electron.WebContents,
  monitorWebContents: Electron.WebContents,
  streamName: string,
) => {  
  try {
    (async () => {
      for await (const data of reconService.subscribe({})) {
        log.debug('⚡️⚡️⚡️⚡️', streamName, data);
        try {
          webContents.send(streamName, data);
          monitorWebContents.send(streamName, data);
        } catch (e) {
          log.error(e);
        }
      }
    })();
  } catch (e) {
    log.error(e);
  }
};

The weird thing is, if we don't call this function after restart, the streamed data stops coming in and we see no log output. If we call the function again after restart, we now get double logs, making me assume we're double processing messages.

What would be the correct way to deal with a server restart and setting up the subscription again?

[aikoven]

Hey @Brammm!

When the server restarts, the streaming call interrupts and the client will eventually receive a gRPC error. It is up to the client to correctly handle it and establish a new streaming call.

There's no single retry strategy for streaming calls, and this is the reason why our retry middleware only handles unary calls. The retry strategy depends on the particular stream semantics. That's why you have to implement it yourself on the client.

One way to do it is using while and try/catch:

const retryableStatuses: Status[] = [
  Status.UNKNOWN,
  Status.INTERNAL,
  Status.UNAVAILABLE,
  Status.CANCELLED,
];

while (true) {
  try {
    for await (const response of client.streamingMethod(/* ... */)) {
      // ...
    }
  } catch (err) {
    if (err instanceof ClientError && retryableStatuses.includes(error.code)) {
      continue; 
    }

    // TODO: add an exponential backoff here

    throw err;
  }
}

This example uses primitive strategy that doesn't do any stitching of two streams together. But it can be adjusted for the particular case.

Regarding your code sample: you will not catch the mentioned gRPC error in your outer try/catch block, because of the dangling promise:

    (async () => {
       // ...
    })();

Any error thrown from the inside will result in an unhandled promise rejection. To fix it, you can use await before that IIFE, or just unwrap it.