fix: set sentinel after subscribe to avoid race condition

[ken8203]

Summary

Fixes a race condition where consumers could miss the producer's request channel subscription.

Problem

Previously, the sentinel key was set before subscribing to the request channel:

sequenceDiagram
    participant P as Producer
    participant R as Redis
    participant C as Consumer

    P->>R: SET sentinel = "1"
    Note over P: Haven't subscribed yet...
    C->>R: GET sentinel
    R-->>C: "1" (exists)
    C->>R: SUBSCRIBE chunk:listenerId
    C->>R: PUBLISH request:streamId
    Note over R: ❌ No subscriber, message lost!
    P->>R: SUBSCRIBE request:streamId
    Note over P: Too late...
    C--xC: Timeout (1s)

Loading

Solution

Move sentinel set to after the subscribe completes:

sequenceDiagram
    participant P as Producer
    participant R as Redis
    participant C as Consumer

    P->>R: SUBSCRIBE request:streamId
    Note over P: ✅ Subscribe first
    P->>R: SET sentinel = "1"
    Note over P: Then set sentinel
    C->>R: GET sentinel
    R-->>C: "1" (exists)
    C->>R: SUBSCRIBE chunk:listenerId
    C->>R: PUBLISH request:streamId
    R-->>P: Received listenerId
    P->>R: PUBLISH chunk:listenerId
    R-->>C: Received data
    Note over C: ✅ Stream works!

Loading

Changes

• Moved ctx.publisher.set(sentinel) from createNewResumableStream wrapper to inside the actual function, after ctx.subscriber.subscribe() completes
• Updated comments to explain the ordering requirement

Testing

All existing tests pass.

[mdnanocom]

+1

[mdnanocom]

@cramforce are you available for a review here? still having issues in production because of this

[cramforce]

Thanks for the mention. Any chance you can add a test?

[mdnanocom]

@ken8203

[ken8203]

sure!

[ken8203]

Hi @cramforce,

I've added a test for the race condition. Please take a look!

[cramforce]

Please re-push with signed git

[mdnanocom]

@ken8203 can you sign commits?

[ken8203]

@mdnanocom @cramforce

Hi, I've signed the commits. Please check again!

[mdnanocom]

@cramforce are we good here?

[mdnanocom]

@cramforce ?