`Router.Close()` blocks for the full `CloseTimeout` when handlers were registered but the router was never (fully) run

Steps to reproduce

No infrastructure needed - the bug is entirely in the router's handlersWg accounting, so it can be reproduced with GoChannel.

package main

import (
        "fmt"
        "time"

        "github.com/ThreeDotsLabs/watermill"
        "github.com/ThreeDotsLabs/watermill/message"
        "github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
)

func main() {
        logger := watermill.NewStdLogger(false, false)

        sub := gochannel.NewGoChannel(gochannel.Config{}, logger)

        router, err := message.NewRouter(message.RouterConfig{
                CloseTimeout: 30 * time.Second,
        }, logger)
        if err != nil {
                panic(err)
        }

        router.AddNoPublisherHandler("h", "topic", sub, func(*message.Message) error {
                return nil
        })

        // The router is never Run - e.g. the surrounding service failed later in
        // startup and calls Close() as part of its cleanup.
        start := time.Now()
        _ = router.Close()
        fmt.Println(time.Since(start)) // ~30s, then "router close timeout"
}

There are two realistic ways we could get here:

1. Constructed-but-not-run: A service builds a router, registers handlers, fails later in startup, and calls Close() during cleanup. Close hangs.

2. Partial RunHandlers failure: RunHandlers registers and starts handlers in a loop and returns early if a subscribe fails midway (message/router.go, line 461):

   go复制

   messages, err := h.subscriber.Subscribe(ctx, h.subscribeTopic)
   if err != nil {
       cancel()
       return errors.Wrapf(err, "cannot subscribe topic %s", ...)
   }

   Run returns that error. Handlers earlier in the map already have their goroutines and will call handlersWg.Done() as the context unwinds, but any handler the loop never reached still sits at Add(1) with no goroutine, and a subsequent Close() waits out the timeout for those.

Expected behavior

Close() on a router with registered but unstarted handlers returns promptly. An unstarted handler does not increment the wait group, so there is nothing for Close to wait on.

Actual behavior

Close() blocks for the full CloseTimeout (30s by default) and then returns "router close timeout".

AddHandler increments handlersWg at registration time, but the only matching handlersWg.Done() runs inside the per-handler goroutine that RunHandlers starts. On a router whose handler goroutines never started, Close() waits on a WaitGroup that can never reach zero.

In message/router.go:

• AddHandler registers and counts the handler (line 317):

  go复制

  r.handlersWg.Add(1)
  r.handlers[handlerName] = newHandler

• The only handlersWg.Done() in the package is inside the goroutine that RunHandlers launches per handler (line 480):

  go复制

  go func() {
      defer cancel()
      ...
      h.run(ctx, middlewares)
      r.handlersWg.Done()   // only Done, only reached once RunHandlers starts the handler
      ...
  }()

• The only in-package caller of RunHandlers is Run (line 404). RunHandlers is exported, so a caller can start handlers added to an already-running router itself, but nothing in the package starts them otherwise: no Run, no goroutine, no Done.

• Close → waitForHandlers waits on that group under the timeout (lines 595-608):

  go复制

  go func() { defer waitGroup.Done(); r.handlersWg.Wait() }()
  ...
  return sync_internal.WaitGroupTimeout(&waitGroup, r.config.CloseTimeout)

Environment: watermill v1.5.3. Behaviour is transport-independent.

Possible solutions

Three suggestions, each with pros & cons.

• Count at start, not registration: Move handlersWg.Add(1) from AddHandler into RunHandlers beside the go func(). Requires synchronisation with Close(). This has a complication I hit while trying it: watchAllHandlersStopped is started by Run before RunHandlers and keys off handlersWg, so with the counter no longer incremented at registration it observes a transient zero and closes the router immediately. On a router with one pre-registered handler, that fires on the first iteration of a loop that runs it 300 times. Calling watchAllHandlersStopped after RunHandlers rather than before fixes it, though the hasNoHandlersYet sampling of len(r.handlers) is racy (this is a pre-existing condition) for handlers that start and finish before the watcher arms.

• Have Close skip or settle unstarted handlers: Complication: lock ordering — Close takes closedLock then handlersLock, RunHandlers takes handlersLock alone, so racing a Close against a RunHandlers risks a double Done.

Happy to open a PR for whichever shape you'd prefer.

PS thank you for the amazing content on your website, it has been so useful for me.