fix(da): serialize subscriber lifecycle transitions

[questfever]

Overview

Closes #3429

Subscriber.Start previously exposed its cancel function before registering the worker goroutines with the WaitGroup. Subscriber.Stop also cleared that function before the current generation had fully exited.

Those orderings allowed Stop to miss newly registered workers and allowed a new subscriber generation to start while the previous generation was still stopping.

This PR serializes the subscriber lifecycle and keeps the stopping state visible until all workers from the current generation have exited.

Changes

• Register worker tasks before publishing the running state.
• Prevent a new generation from starting while shutdown is in progress.
• Keep the subscriber in the stopping state until all workers exit.
• Make repeated and concurrent Start/Stop calls safe.
• Preserve restart support after a completed stop.
• Add deterministic lifecycle concurrency tests.

Testing

• go test ./block/internal/da
• go test -race ./block/internal/da
• Repeated concurrent lifecycle stress test

Summary by CodeRabbit

• Bug Fixes

  • Improved subscriber start and stop reliability during concurrent operations.
  • Prevented overlapping subscription sessions and ensured repeated stop requests complete safely.
  • Ensured shutdown waits for active subscription work to finish.

• Tests

  • Added coverage for concurrent lifecycle transitions and repeated start/stop scenarios.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 588382fa-f695-4d5a-b90f-ea345d2831b8

📥 Commits

Reviewing files that changed from the base of the PR and between 5c88745 and 2817e55.

📒 Files selected for processing (2)

• block/internal/da/subscriber.go
• block/internal/da/subscriber_test.go

---

📝 Walkthrough

Walkthrough

The subscriber now uses explicit stopped, running, and stopping states. Start and Stop operations serialize lifecycle transitions, wait for worker shutdown, support restart after completion, and handle repeated calls safely. Tests cover concurrent lifecycle operations across subscription generations.

Changes

Subscriber lifecycle serialization

Layer / File(s)	Summary
Lifecycle state machine block/internal/da/subscriber.go	Adds explicit lifecycle states and stop-completion signaling. Start publishes a running generation only after registration. Stop serializes shutdown, waits for concurrent callers, joins workers, and resets the subscriber to stopped.
Lifecycle concurrency validation block/internal/da/subscriber_test.go	Adds synchronized test helpers and verifies concurrent starts and stops, blocked shutdown, generation isolation, restart after stopping, and repeated stops.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Mergeability Score: ⚪ Minimal · up to 2817e

The change serializes subscriber start and stop transitions and adds focused concurrency coverage; no actionable merge-blocking risk remains beyond normal checks and review.

Possibly related PRs

• evstack/ev-node#3418: Both changes add lifecycle state handling and concurrency-safe Start/Stop behavior.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the main change: serializing subscriber lifecycle transitions.
Description check	✅ Passed	The description includes the required overview, issue reference, rationale, changes, and testing information.
Linked Issues check	✅ Passed	The implementation and tests address all linked issue requirements for safe, serialized, restartable subscriber lifecycles.
Out of Scope Changes check	✅ Passed	The changes are limited to subscriber lifecycle synchronization and directly related concurrency tests.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[questfever]

@tac0turtle Hi, Could you please review this PR at your convenience? Thank you very much.

[tac0turtle]

[suggestion] The concurrent-Stop assertion is a single default poll immediately after launching the goroutine, with no handshake that the second Stop has entered <-stopDone. If Stop treated subscriberStopping as a no-op (returned immediately), this check can still take the default branch if the goroutine has not been scheduled yet; after releaseGeneration(0) the later waitForLifecycleSignal on concurrentStopDone only proves the call eventually returned, not that it waited with the first Stop. Contrast with the first Stop, which is actually proven: the test waits for canceled[0] (so cancel() has run and Wait is blocked on the unreleased worker) before asserting stopDone is still open.

Suggestion: Handshake that the second Stop is inside the wait (for example a short timeout loop that fails only if concurrentStopDone closes before release, or observe state == subscriberStopping / a test hook after the waiter has copied stopDone), then release generation 0. Keep the existing canceled[0] handshake for the first Stop; that one is already deterministic.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 62.52%. Comparing base (5c88745) to head (2817e55).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3430      +/-   ##
==========================================
+ Coverage   62.46%   62.52%   +0.05%     
==========================================
  Files         121      121              
  Lines       13469    13485      +16     
==========================================
+ Hits         8413     8431      +18     
- Misses       4114     4115       +1     
+ Partials      942      939       -3     

Flag	Coverage Δ
combined	62.52% <100.00%> (+0.05%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.