watcher: deliver events to consumers in order, without drops

[benoit-nexthop]

Fixes the event delivery guarantees of the database watcher. Addresses the root cause of #847.

The watcher currently fans out each event to every consumer in a fresh goroutine (go c.Send(payload)), so two events emitted back to back can be delivered in reverse order. Instance deletion emits update(status=deleted) followed by delete. When a consumer observes them in the wrong order, the late update re-inserts an already deleted instance into event-driven state. Since the status is terminal, nothing ever evicts it, it'll linger forever. In our production deployment this leaked hundreds of phantom instances over 9 days (dashboard reported 895 instances vs 77 real) and starved scale-up, with queued jobs waiting up to 17 minutes for a runner on an otherwise idle system.

Consumer channels are also buffered at 1 with a 1 second send timeout that silently drops events during bursts, logged only at debug level.

This PR replaces the per-event goroutine with a synchronous enqueue into an ordered per-consumer queue, drained by a dedicated dispatch goroutine:

• Events are delivered to each consumer in emission order.
• Send never blocks on a slow consumer and never drops events; a warning is logged if a consumer's queue keeps growing.
• The dispatch loop owns closing the messages channel, so a send on a closed channel cannot happen.

Includes regression tests for in-order delivery and for no drops with a slow consumer under burst.

This is only the first step of fixing #847: it removes the source of the corruption, but event-driven state should also self-heal if an event is ever missed (e.g. periodic reconciliation of the instance cache and scale set worker state against the database), and consumers should evict on update(status=deleted) as defense in depth. I'll send follow up changes once this one lands.

[gabriel-samfira]

I was actually looking at this bit of code. You can use a WaitGroup{} here to send notifications to all consumers in parallel. Events would still be sent synchronously, but if we have multiple slow consumers, the wait time would not compound. We would wait at most 1 second, regardless of how many slow consumers we have.

Something like:

wg := sync.WaitGroup{}
for _, c := range w.consumers {
    wg.Go(func(){
        c.Send(payload)
    })
}
wg.Wait()

[gabriel-samfira]

Thanks for this PR. I think the fix for the issue you've been seeing will need fixes in multiple places. I think the pain we've been having with syncing state and acting on runner transitions (from pending_create to creating to running to deleting, etc) is due to a more fundamental flaw in how we track state.

There is a desired state and an observed state that I sadly mashed together. The desired state is what the user expresses and what the code tries to converge to. The observed state is the state of the resource we need (the runner in our case) as it's being created/updated/removed by a provider. I need to think of how we can separate the two. That way, when the provider acts on a snapshot of the desired state (a runner was placed in pending create), we can still issue a force_delete on the runner without interfering with the provider. The provider doesn't need to lock the whole instance as it's creating it. Once its done and it sees that the new desired state is to remove it, it can just act again.

This change is still good, and I will have a closer look at it once I get back from vacation, but I also need to have a closer look at how we handle state, so we don't just keep plugging holes here and there.