test(alert): drain notifications before teardown
CI / Test (rest) (pull_request) Successful in 39s
CI / Lint (pull_request) Successful in 11s
CI / Test (store) (pull_request) Successful in 43s
CI / Build (windows/amd64) (pull_request) Successful in 7s
CI / Build (linux/amd64) (pull_request) Successful in 8s
CI / Build (linux/arm64) (pull_request) Successful in 8s
CI / Test (server-http) (pull_request) Successful in 1m34s
e2e / Playwright vs docker-compose (pull_request) Successful in 1m27s

This commit is contained in:
2026-08-22 11:31:56 +01:00
parent 320be34970
commit 39a0875d58
3 changed files with 16 additions and 4 deletions
+11
View File
@@ -61,9 +61,20 @@ type Engine struct {
stuckThresholds map[string]time.Duration stuckThresholds map[string]time.Duration
closeOnce sync.Once closeOnce sync.Once
notifyWG sync.WaitGroup
done chan struct{} done chan struct{}
} }
func (e *Engine) dispatchNotification(ctx context.Context, payload notification.Payload) {
e.notifyWG.Add(1)
go func() {
defer e.notifyWG.Done()
e.hub.Dispatch(ctx, payload)
}()
}
func (e *Engine) waitNotifications() { e.notifyWG.Wait() }
// NewEngine builds the engine. agentOfflineFloor + tickPeriod default // NewEngine builds the engine. agentOfflineFloor + tickPeriod default
// to 15min and 60s respectively when zero. // to 15min and 60s respectively when zero.
func NewEngine(st *store.Store, hub *notification.Hub) *Engine { func NewEngine(st *store.Store, hub *notification.Hub) *Engine {
+4 -4
View File
@@ -60,7 +60,7 @@ func (e *Engine) raiseAndNotify(ctx context.Context, hostID, kind, dedupKey, sev
if err == nil { if err == nil {
hostName = host.Name hostName = host.Name
} }
go e.hub.Dispatch(ctx, notification.Payload{ e.dispatchNotification(ctx, notification.Payload{
Event: notification.EventRaised, Event: notification.EventRaised,
AlertID: id, AlertID: id,
Severity: severity, Severity: severity,
@@ -85,7 +85,7 @@ func (e *Engine) Acknowledge(ctx context.Context, alertID, userID string, when t
return nil //nolint:nilerr return nil //nolint:nilerr
} }
p := alertPayload(ctx, e.store, notification.EventAcknowledged, a) p := alertPayload(ctx, e.store, notification.EventAcknowledged, a)
go e.hub.Dispatch(context.WithoutCancel(ctx), p) e.dispatchNotification(context.WithoutCancel(ctx), p)
return nil return nil
} }
@@ -99,7 +99,7 @@ func (e *Engine) Resolve(ctx context.Context, alertID string, when time.Time) er
return nil return nil
} }
p := alertPayload(ctx, e.store, notification.EventResolved, a) p := alertPayload(ctx, e.store, notification.EventResolved, a)
go e.hub.Dispatch(context.WithoutCancel(ctx), p) e.dispatchNotification(context.WithoutCancel(ctx), p)
return nil return nil
} }
@@ -164,7 +164,7 @@ func (e *Engine) resolveAndNotify(ctx context.Context, hostID, kind, dedupKey st
if a.Kind != kind || a.DedupKey != dedupKey { if a.Kind != kind || a.DedupKey != dedupKey {
continue continue
} }
go e.hub.Dispatch(ctx, notification.Payload{ e.dispatchNotification(ctx, notification.Payload{
Event: notification.EventResolved, Event: notification.EventResolved,
AlertID: a.ID, AlertID: a.ID,
Severity: a.Severity, Severity: a.Severity,
+1
View File
@@ -25,6 +25,7 @@ func setupEngine(t *testing.T) (*Engine, *store.Store, string) {
aead, _ := crypto.NewAEAD(key) aead, _ := crypto.NewAEAD(key)
hub := notification.NewHub(st, aead, "https://rm.example") hub := notification.NewHub(st, aead, "https://rm.example")
eng := NewEngine(st, hub) eng := NewEngine(st, hub)
t.Cleanup(eng.waitNotifications)
hostID := ulid.Make().String() hostID := ulid.Make().String()
if err := st.CreateHost(context.Background(), store.Host{ if err := st.CreateHost(context.Background(), store.Host{
ID: hostID, Name: "alfa-01", OS: "linux", Arch: "amd64", ID: hostID, Name: "alfa-01", OS: "linux", Arch: "amd64",