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.}