Subscriber.Start and Stop can overlap lifecycle generations

[questfever]

Version Information

ev-node: main at 5c88745ce84300cf67dcc7185bbf7c54822e3233

System Information

This is a platform-independent concurrency issue.

Reproduced on:

• macOS arm64
• Go 1.25.8

Network Configuration

No response

Development Environment

No response

Configuration Details

No response

Steps to Reproduce

Subscriber.Start currently publishes the new cancel function before
registering its background tasks with the WaitGroup:

1. Lock lifecycleMu
2. Store s.cancel
3. Unlock lifecycleMu
4. Call s.wg.Add(2)
5. Start the worker goroutines

Subscriber.Stop performs the following:

1. Lock lifecycleMu
2. Read s.cancel
3. Set s.cancel = nil
4. Unlock lifecycleMu
5. Cancel the context
6. Call s.wg.Wait()

This allows two problematic interleavings.

Interleaving 1: Stop waits before Start registers its tasks

1. Start stores s.cancel and unlocks.
2. Stop acquires the lock and clears s.cancel.
3. Stop cancels the context and calls wg.Wait().
4. The WaitGroup counter is still zero, so Wait() returns.
5. Start subsequently calls wg.Add(2) and launches the workers.

The workers receive an already-canceled context and will normally exit quickly, but they may start after Stop has returned and are not joined by that Stop call.

Interleaving 2: A new generation starts while the old generation is stopping

1. Generation 1 is running.
2. Stop clears s.cancel, cancels generation 1, and waits for it to exit.
3. Before generation 1 has exited, another Start observes s.cancel == nil.
4. Start registers and launches generation 2 using the same WaitGroup.

This allows both generations to overlap. The first Stop may then wait for generation 2, even though it did not cancel generation 2. Reusing the WaitGroup around the previous generation's zero transition may also result in:

• sync: WaitGroup misuse: Add called concurrently with Wait
• sync: WaitGroup is reused before previous Wait has returned

A deterministic reproduction can use a fake DA client whose first Subscribe call remains blocked after its context is canceled. Calling Start again while Stop is waiting causes a second Subscribe call to begin before the first one has exited.

Expected Result

• WaitGroup tasks are registered before the running state becomes observable.
• Stop does not return until all workers from the stopped generation exit.
• A new generation cannot start while the previous generation is stopping.
• Repeated or concurrent Start and Stop calls are safe and idempotent.
• A subscriber can be started again after a completed stop.

Actual Result

• Stop may return before the corresponding workers have started and exited.
• A second generation may start while the first generation is stopping.
• A Stop call may wait for a newer, uncanceled generation.
• WaitGroup reuse can violate the documented Add/Wait ordering rules.

Relevant Logs

Additional Information

Proposed Fix:

Serialize lifecycle transitions explicitly, for example with:

• stopped
• running
• stopping

Required invariants:

• Register the generation's WaitGroup tasks before publishing running.
• Keep the subscriber in stopping until all old workers have exited.
• Do not start a new generation while the state is stopping.
• Transition back to stopped only after wg.Wait() completes.
• Make concurrent and repeated Stop calls wait for the same stop completion.

An equivalent implementation that preserves these invariants would also be
acceptable.

[questfever]

I'd like to work on this issue. I have a deterministic reproduction that keeps the first subscription blocked during shutdown and confirms that the current implementation can start a second generation before the first Stop returns.