[djthorpe]

A broadcaster that has stopped refuses watches with ErrServiceUnavailable until Run starts it again. When Run is started in a goroutine, a watch made straight afterwards can come before it, and be refused. go-pg's pgbroadcaster hit this when a manager runs a second time: it reported that it was listening before its broadcaster was running.

• Running() is an atomic bool, as on the other managers. It's true from the moment Run clears the stopped state until it stops, so true means watches are accepted. A caller that starts Run in a goroutine waits for Running() before reporting it's ready:

  go b.Run(ctx)
  // wait until b.Running(), then b.Watch(...) is accepted

• Never run: a broadcaster that has never run accepts watches but isn't Running().

• Run is otherwise unchanged.

• Tests: the restart in the Stopped test waits for Running(), and a new Running test checks the flag before, during and after a run. Clean under -race five times.

[Copilot]

🟢 Approval recommended

No active review issues are reported for the synchronous broadcaster startup change.

No open issues 🎉

What changed in this PR

Adds a synchronous Start() method so a stopped broadcaster can accept watches before Run begins in a goroutine, avoiding rejected watches during restart.

Changes:

• Introduces Start() and delegates to it from Run, preserving existing caller behavior.
• Replaces restart polling with Start() and adds coverage for accepting watches before Run.

File	Description
broadcaster/​broadcaster.go	Adds and documents Start(); calls it from Run.
broadcaster/​broadcaster_test.go	Updates restart coverage and tests watch acceptance after Start().

_{Review effort: Balanced}