Concurrent requests to `SUNSUBCRIBE` causes a race condition bug (`ClientClosedError`)

[BrentLayne]

Description

Bug description

Making concurrent requests to SUNSUBSCRIBE causes one or more of the requests to throw the ClientClosedError

Why is this happening?

This bug is caused by a race condition due to how the executeShardedUnsubscribeCommand function works:

• The function awaits multiple async calls
• THEN the function calls client.disconnect()

Every time the executeShardedUnsubscribeCommand function encounters an await, function execution is paused and the javascript thread begins to process the other concurrent requests to SUNBSUBSCRIBE. As a result, the requests to SUNSUBSCRIBE are now racing against each other.

The request that 'wins' this race will call disconnect() first and disconnect the socket. The subsequent requests will try to disconnect the socket again, resulting in the ClientClosedError being thrown

What is the expected behaviour?

I can make concurrent requests to SUNSUBSCRIBE and all the requests resolve with no errors thrown

Steps to reproduce

Run the below script to reproduce the race condition bug:

import { createCluster } from 'redis'


const client = createCluster({ rootNodes });

await client.connect();

const onMessage = () => {};

await client.sSubscribe('test', onMessage);
await client.sSubscribe('test2', onMessage);

try {
  // concurrent requests to sUnsubscribe are racing against each other to disconnect the socket
  await Promise.all([
    client.sUnsubscribe('test', onMessage),
    client.sUnsubscribe('test2', onMessage),
  ]);
} catch (error) {
  // this log statement logs the ClientClosedError
  console.log('*** error ***', error);
}

Node.js Version

18.18.2

Redis Server Version

7.0.7

Node Redis Version

4.6.7

Platform

macOS

Logs

No response

[BrentLayne]

I think the fix might be as simple as adding a check for client.isOpen after the async calls, right before invoking client.disconnect() in executeShardedUnsubscribeCommand

i.e. the function definition should be changed to be the following:

async executeShardedUnsubscribeCommand(
        channel: string,
        unsubscribe: (client: RedisClientType<M, F, S>) => Promise<void>
    ): Promise<void> {
        const { master } = this.slots[calculateSlot(channel)];
        if (!master.pubSubClient) return Promise.resolve();

        const client = await master.pubSubClient;
        await unsubscribe(client);
        
        // NEW: Confirm that the client.isOpen before disconnecting
        if (!client.isPubSubActive && client.isOpen) {
            await client.disconnect();
            master.pubSubClient = undefined;
        }
    }

With this change, I no longer see an error

[BrentLayne]

Also worth noting that cluster UNSUBSCRIBE and PUNSUBSCRIBE likely has the exact same problem because the code for executeUnsubscribeCommand is structured the exact same way

[leibale]

@BrentLayne Good catch! Want to open a PR (preferably with a test)? Or you prefer I'll take it from here?

[BrentLayne]

@leibale I should have a PR up shortly!

[BrentLayne]

@leibale for your consideration: #2687

Thanks!

[leibale]

fixed in #2687