From cd3ced5a1cd41283586e386b3e71c4512271ff96 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 6 Oct 2026 21:52:19 -0700 Subject: [PATCH] fix(webpush): isolate slow browser delivery endpoints Serialize each endpoint within the shared worker pool and pending budget. Keep accepted alerts, per-send authority checks, and normal success bookkeeping intact. Co-authored-by: Sebastien Tardif --- CHANGELOG.md | 2 + apps/api/internal/httpapi/webpush.go | 88 ++++++ apps/api/internal/httpapi/webpush_test.go | 331 +++++++++++++++++++++- docs/features/push-notifications.md | 6 +- 4 files changed, 417 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bbbb3127..91a925a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Prevent a slow browser push endpoint from occupying every delivery worker, while keeping one shared limit of 1,024 pending alerts and normal success bookkeeping. Thanks @SebTardif. + - Keep startup available during default OpenClaw ID discovery server errors while custom issuers still fail closed. Thanks @SebTardif. ## 0.7.0 - 2026-09-30 diff --git a/apps/api/internal/httpapi/webpush.go b/apps/api/internal/httpapi/webpush.go index 7c04287d..d49155e6 100644 --- a/apps/api/internal/httpapi/webpush.go +++ b/apps/api/internal/httpapi/webpush.go @@ -18,6 +18,10 @@ const ( // webPushWorkers bounds how many pushes are in flight at once. Delivery is // off the request path, so this is the only backpressure the relay sees. webPushWorkers = 4 + // webPushInFlightPerSubscription bounds how many of those workers one + // endpoint may hold. Further pushes for that endpoint wait behind the + // one already in flight, so a slow relay cannot occupy the whole pool. + webPushInFlightPerSubscription = 1 // webPushQueueSize bounds the memory a wedged push service can cost. Chat // notifications are not durable mail: an overflow is dropped and counted. webPushQueueSize = 1024 @@ -90,6 +94,15 @@ type WebPushNotifier struct { droppedLogged time.Time stopPruning context.CancelFunc pruning sync.WaitGroup + // flightMu guards inflight, pending, and pendingCount. inflight counts + // Sends in progress for an endpoint. pending holds later pushes for an + // endpoint that is already sending; the worker that owns the slot runs + // them. pendingCount is the shared channel plus every waiting list, and + // it stops at webPushQueueSize. + flightMu sync.Mutex + inflight map[string]int + pending map[string][]webPushDelivery + pendingCount int } // NewWebPushNotifier builds a notifier that sends through the same outbound @@ -122,6 +135,8 @@ func newWebPushNotifier(sender webPushSender, subscriptions WebPushSubscriptionS keyID: keyID, queue: make(chan webPushDelivery, webPushQueueSize), stopPruning: stopPruning, + inflight: map[string]int{}, + pending: map[string][]webPushDelivery{}, } notifier.workers.Add(webPushWorkers) for range webPushWorkers { @@ -153,9 +168,14 @@ func (n *WebPushNotifier) Notify(_ context.Context, notification PushNotificatio messageID: notification.MessageID, message: message, } + if !n.reservePending() { + dropped++ + continue + } select { case n.queue <- delivery: default: + n.releasePending() dropped++ } } @@ -228,8 +248,76 @@ func (n *WebPushNotifier) prune(parent context.Context) { func (n *WebPushNotifier) work() { defer n.workers.Done() for delivery := range n.queue { + n.deliverSubscription(delivery) + } +} + +// deliverSubscription runs delivery, then the pushes waiting for that +// endpoint. Another worker does not block inside Send for an endpoint that +// is already in flight. +func (n *WebPushNotifier) deliverSubscription(delivery webPushDelivery) { + if !n.claimSubscription(delivery) { + return + } + for { + n.releasePending() n.deliver(delivery) + next, ok := n.takeNextSubscription(delivery.endpoint) + if !ok { + return + } + delivery = next + } +} + +func (n *WebPushNotifier) claimSubscription(delivery webPushDelivery) bool { + n.flightMu.Lock() + defer n.flightMu.Unlock() + if n.inflight[delivery.endpoint] >= webPushInFlightPerSubscription { + n.pending[delivery.endpoint] = append(n.pending[delivery.endpoint], delivery) + return false + } + n.inflight[delivery.endpoint]++ + return true +} + +func (n *WebPushNotifier) reservePending() bool { + n.flightMu.Lock() + defer n.flightMu.Unlock() + if n.pendingCount >= webPushQueueSize { + return false + } + n.pendingCount++ + return true +} + +func (n *WebPushNotifier) releasePending() { + n.flightMu.Lock() + defer n.flightMu.Unlock() + if n.pendingCount > 0 { + n.pendingCount-- + } +} + +func (n *WebPushNotifier) takeNextSubscription(endpoint string) (webPushDelivery, bool) { + n.flightMu.Lock() + defer n.flightMu.Unlock() + pending := n.pending[endpoint] + if len(pending) == 0 { + n.inflight[endpoint]-- + if n.inflight[endpoint] == 0 { + delete(n.inflight, endpoint) + } + return webPushDelivery{}, false + } + next := pending[0] + pending[0] = webPushDelivery{} + if len(pending) == 1 { + delete(n.pending, endpoint) + } else { + n.pending[endpoint] = pending[1:] } + return next, true } // Detached workers recover panics so a notification failure cannot crash the server. diff --git a/apps/api/internal/httpapi/webpush_test.go b/apps/api/internal/httpapi/webpush_test.go index 10c080e7..b814aa4e 100644 --- a/apps/api/internal/httpapi/webpush_test.go +++ b/apps/api/internal/httpapi/webpush_test.go @@ -134,6 +134,7 @@ type fakeSender struct { err error sent int hold chan struct{} + delay time.Duration panc bool subscriptions []webpush.Subscription } @@ -141,6 +142,7 @@ type fakeSender struct { func (f *fakeSender) Send(_ context.Context, subscription webpush.Subscription, _ webpush.Message) error { f.mu.Lock() hold := f.hold + delay := f.delay err := f.err shouldPanic := f.panc f.sent++ @@ -149,6 +151,9 @@ func (f *fakeSender) Send(_ context.Context, subscription webpush.Subscription, if hold != nil { <-hold } + if delay > 0 { + time.Sleep(delay) + } if shouldPanic { panic("push exploded") } @@ -226,20 +231,23 @@ func TestWebPushNotifierToleratesBookkeepingFailures(t *testing.T) { } // TestWebPushNotifierSurvivesAPanickingDelivery guards the detached pool: an -// unrecovered panic in a worker would take the whole server down. +// unrecovered panic in a worker would take the whole server down. One +// endpoint keeps a single in-flight send. Later pushes for it wait, and +// nothing accepted may disappear. func TestWebPushNotifierSurvivesAPanickingDelivery(t *testing.T) { t.Parallel() sender := &fakeSender{panc: true} subscriptions := &fakeSubscriptionStore{} notifier := newWebPushNotifier(sender, subscriptions, "") - for range 8 { + const burst = 8 + for range burst { if err := notifier.Notify(context.Background(), pushNotificationFor("usr_1", "https://push.example.com/send/device")); err != nil { t.Fatal(err) } } notifier.Close() - if sender.count() != 8 { - t.Fatalf("every delivery must be attempted, got %d", sender.count()) + if attempts := sender.count(); attempts != burst { + t.Fatalf("every accepted delivery must be attempted after a panic, got %d", attempts) } if len(subscriptions.recorded()) != 0 { t.Fatalf("a panicking delivery records nothing: %#v", subscriptions.recorded()) @@ -252,17 +260,37 @@ func TestWebPushNotifierDropsWhenTheQueueIsFull(t *testing.T) { sender := &fakeSender{hold: hold} notifier := newWebPushNotifier(sender, &fakeSubscriptionStore{}, "") notification := pushNotificationFor("usr_1", "https://push.example.com/send/device") - for range webPushQueueSize + webPushWorkers + 50 { + for range webPushQueueSize*2 + webPushWorkers + 50 { if err := notifier.Notify(context.Background(), notification); err != nil { t.Fatal(err) } } - notifier.dropMu.Lock() - dropped := notifier.dropped - notifier.dropMu.Unlock() + deadline := time.Now().Add(2 * time.Second) + var dropped int64 + for time.Now().Before(deadline) { + notifier.dropMu.Lock() + dropped = notifier.dropped + notifier.dropMu.Unlock() + if dropped > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } if dropped == 0 { t.Fatal("expected the overflow to be counted") } + notifier.flightMu.Lock() + waiters := 0 + for _, list := range notifier.pending { + waiters += len(list) + } + pending := notifier.pendingCount + notifier.flightMu.Unlock() + queued := len(notifier.queue) + t.Logf("pending=%d waiters=%d channel=%d dropped=%d", pending, waiters, queued, dropped) + if pending > webPushQueueSize || waiters+queued > webPushQueueSize { + t.Fatalf("pending budget exceeded: pending=%d waiters=%d channel=%d", pending, waiters, queued) + } close(hold) notifier.Close() } @@ -414,6 +442,293 @@ func TestWebPushNotifierBacksOffATimedOutRelay(t *testing.T) { } } +// TestWebPushNotifierCapsInFlightDeliveriesPerSubscription holds one endpoint +// inside Send and queues more pushes for it. Only one worker may stay there. +// A different endpoint must still get a worker from the rest of the pool. +func TestWebPushNotifierCapsInFlightDeliveriesPerSubscription(t *testing.T) { + t.Parallel() + sender := &concurrencySender{started: make(chan string, 16), hold: make(chan struct{})} + notifier := newWebPushNotifier(sender, &fakeSubscriptionStore{}, "") + t.Cleanup(notifier.Close) + t.Cleanup(sender.release) + + slow := "https://push.example.com/send/slow" + for range 8 { + if err := notifier.Notify(context.Background(), pushNotificationFor("usr_1", slow)); err != nil { + t.Fatal(err) + } + } + select { + case got := <-sender.started: + if got != slow { + t.Fatalf("first send was %s", got) + } + case <-time.After(5 * time.Second): + t.Fatal("no delivery started") + } + time.Sleep(50 * time.Millisecond) + if max := sender.maxConcurrent(slow); max != 1 { + t.Fatalf("subscription had %d deliveries in flight, want 1", max) + } + + other := "https://push.example.com/send/other" + if err := notifier.Notify(context.Background(), pushNotificationFor("usr_2", other)); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + for sender.maxConcurrent(other) < 1 { + if time.Now().After(deadline) { + t.Fatal("a different subscription never started while the slow one was in flight") + } + time.Sleep(10 * time.Millisecond) + } + if max := sender.maxConcurrent(slow); max != 1 { + t.Fatalf("subscription had %d deliveries in flight after another device started, want 1", max) + } +} + +// TestWebPushNotifierIsolatesEndpointsOverHTTP holds one real encrypted send +// and lets a second endpoint proceed. Extra pushes for the held endpoint do +// not take more workers. +func TestWebPushNotifierIsolatesEndpointsOverHTTP(t *testing.T) { + ctx := context.Background() + st, owner, channel, _ := newWebPushTestStore(t) + message, _, err := st.CreateMessage(ctx, store.CreateMessageInput{ChannelID: channel.ID, AuthorID: owner.ID, Body: "burst"}) + if err != nil { + t.Fatal(err) + } + var mu sync.Mutex + slowInFlight, slowMax, slowRequests, otherStarted := 0, 0, 0, 0 + hold := make(chan struct{}) + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + if strings.HasSuffix(r.URL.Path, "/slow") { + mu.Lock() + slowInFlight++ + slowRequests++ + if slowInFlight > slowMax { + slowMax = slowInFlight + } + mu.Unlock() + <-hold + mu.Lock() + slowInFlight-- + mu.Unlock() + } else { + mu.Lock() + otherStarted++ + mu.Unlock() + } + w.WriteHeader(http.StatusCreated) + })) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(hold) }) } + t.Cleanup(func() { + release() + relay.Close() + }) + slow := relay.URL + "/slow" + other := relay.URL + "/other" + registerKeyedDevice(t, st, owner.ID, slow, "") + registerKeyedDevice(t, st, owner.ID, other, "") + publicKey, privateKey, err := webpush.GenerateKeys() + if err != nil { + t.Fatal(err) + } + notifier := newWebPushNotifier(&webpush.Sender{ + PublicKey: publicKey, + PrivateKey: privateKey, + Subject: "https://chat.example.com", + Client: relay.Client(), + }, st, "") + t.Cleanup(notifier.Close) + t.Cleanup(release) + notify := func(endpoint string) { + t.Helper() + if err := notifier.Notify(ctx, PushNotification{ + UserID: owner.ID, + MessageID: message.ID, + Subscriptions: []store.PushSubscriptionTarget{{Endpoint: endpoint}}, + }); err != nil { + t.Fatal(err) + } + } + notify(slow) + deadline := time.Now().Add(5 * time.Second) + for { + mu.Lock() + started := slowInFlight == 1 + mu.Unlock() + if started { + break + } + if time.Now().After(deadline) { + t.Fatal("the slow endpoint never accepted a request") + } + time.Sleep(10 * time.Millisecond) + } + notify(slow) + notify(slow) + notify(slow) + notify(other) + deadline = time.Now().Add(5 * time.Second) + for { + mu.Lock() + ready := otherStarted == 1 && slowMax == 1 + mu.Unlock() + if ready { + break + } + if time.Now().After(deadline) { + mu.Lock() + other, max := otherStarted, slowMax + mu.Unlock() + t.Fatalf("other=%d slow_max=%d", other, max) + } + time.Sleep(10 * time.Millisecond) + } + release() + deadline = time.Now().Add(5 * time.Second) + for { + mu.Lock() + done := slowRequests == 4 && slowInFlight == 0 + mu.Unlock() + if done { + break + } + if time.Now().After(deadline) { + mu.Lock() + requests, inFlight := slowRequests, slowInFlight + mu.Unlock() + t.Fatalf("slow_requests=%d in_flight=%d", requests, inFlight) + } + time.Sleep(10 * time.Millisecond) + } + notifier.Close() + mu.Lock() + t.Logf("slow_max=%d slow_requests=%d other_while_held=%d", slowMax, slowRequests, otherStarted) + mu.Unlock() +} + +type concurrencySender struct { + mu sync.Mutex + current map[string]int + max map[string]int + started chan string + hold chan struct{} + once sync.Once +} + +func (c *concurrencySender) Send(_ context.Context, subscription webpush.Subscription, _ webpush.Message) error { + endpoint := subscription.Endpoint + c.mu.Lock() + if c.current == nil { + c.current = map[string]int{} + c.max = map[string]int{} + } + c.current[endpoint]++ + if c.current[endpoint] > c.max[endpoint] { + c.max[endpoint] = c.current[endpoint] + } + c.mu.Unlock() + c.started <- endpoint + <-c.hold + c.mu.Lock() + c.current[endpoint]-- + c.mu.Unlock() + return nil +} + +func (c *concurrencySender) maxConcurrent(endpoint string) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.max[endpoint] +} + +func (c *concurrencySender) release() { + c.once.Do(func() { close(c.hold) }) +} + +// TestWebPushNotifierRecordsASlowSuccess records a delayed 201 as a normal success. It does not start a cooldown. +func TestWebPushNotifierRecordsASlowSuccess(t *testing.T) { + previousTimeout := webPushSendTimeout + webPushSendTimeout = 2 * time.Second + defer func() { webPushSendTimeout = previousTimeout }() + + subscriptions := &fakeSubscriptionStore{} + notifier := newWebPushNotifier(&fakeSender{delay: 180 * time.Millisecond}, subscriptions, "") + if err := notifier.Notify(context.Background(), pushNotificationFor("usr_1", "https://push.example.com/send/device")); err != nil { + t.Fatal(err) + } + notifier.Close() + recorded := subscriptions.recorded() + if len(recorded) != 1 || recorded[0].kind != "success" { + t.Fatalf("slow success bookkeeping = %#v", recorded) + } + if recorded[0].userID != "usr_1" || recorded[0].endpoint != "https://push.example.com/send/device" { + t.Fatalf("unexpected bookkeeping target %#v", recorded[0]) + } +} + +// TestSlowSuccessfulPushLeavesFailureHistoryClear sends through webpush.Sender +// to a local relay that waits and then returns 201. The sqlite row records +// success without a cooldown or a failure run. +func TestSlowSuccessfulPushLeavesFailureHistoryClear(t *testing.T) { + previousTimeout := webPushSendTimeout + webPushSendTimeout = 2 * time.Second + defer func() { webPushSendTimeout = previousTimeout }() + + ctx := context.Background() + st, owner, channel, databasePath := newWebPushTestStore(t) + message, _, err := st.CreateMessage(ctx, store.CreateMessageInput{ChannelID: channel.ID, AuthorID: owner.ID, Body: "slow but delivered"}) + if err != nil { + t.Fatal(err) + } + relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(180 * time.Millisecond) + w.WriteHeader(http.StatusCreated) + })) + t.Cleanup(relay.Close) + endpoint := relay.URL + "/send/device" + registerKeyedDevice(t, st, owner.ID, endpoint, "") + publicKey, privateKey, err := webpush.GenerateKeys() + if err != nil { + t.Fatal(err) + } + notifier := newWebPushNotifier(&webpush.Sender{ + PublicKey: publicKey, + PrivateKey: privateKey, + Subject: "https://chat.example.com", + Client: relay.Client(), + }, st, "") + if err := notifier.Notify(ctx, PushNotification{ + UserID: owner.ID, + MessageID: message.ID, + Subscriptions: []store.PushSubscriptionTarget{{Endpoint: endpoint}}, + }); err != nil { + t.Fatal(err) + } + notifier.Close() + + db, err := sql.Open("sqlite", "file:"+databasePath+"?_pragma=busy_timeout(5000)") + if err != nil { + t.Fatal(err) + } + defer db.Close() + var failures int64 + var failingSince, lastSuccess, nextAttempt sql.NullString + if err := db.QueryRow( + `SELECT failure_count, failing_since, last_success_at, next_attempt_at FROM user_push_subscriptions WHERE endpoint = ?`, + endpoint, + ).Scan(&failures, &failingSince, &lastSuccess, &nextAttempt); err != nil { + t.Fatal(err) + } + t.Logf("failure_count=%d failing_since_set=%t last_success_set=%t next_attempt_set=%t", failures, failingSince.Valid, lastSuccess.Valid, nextAttempt.Valid) + if failures != 0 || failingSince.Valid || !lastSuccess.Valid || nextAttempt.Valid { + t.Fatalf("row failure_count=%d failing_since=%q last_success=%q next_attempt=%q", failures, failingSince.String, lastSuccess.String, nextAttempt.String) + } +} + // TestWebPushNotifierStoresTheRelaysRetryAfter follows a refusal from the push // service to the backoff on the stored row. Retry-After may be a number of // seconds or an HTTP-date, and either way the device is not tried again before diff --git a/docs/features/push-notifications.md b/docs/features/push-notifications.md index 6240ee35..fc8cdd14 100644 --- a/docs/features/push-notifications.md +++ b/docs/features/push-notifications.md @@ -109,8 +109,10 @@ or not the app is open: still starting can arrive before the app is listening. Delivery happens off the request path in a small worker pool, so posting a -message never waits on a push service. The queue in front of that pool holds -1,024 pending pushes and lives in memory. When a push service stalls or a +message never waits on a push service. Each device endpoint can occupy only +one worker; its later alerts wait without blocking workers for other devices. +The shared queue and endpoint waiting lists together hold at most 1,024 +pending pushes in memory. A delayed successful delivery does not add a cooldown. When a push service stalls or a burst of messages outgrows it, further pushes are dropped and counted, with one log line a minute giving the count; the messages themselves are unaffected and are read in the app as usual. Phone alerts are best effort by