amadan

eleven / messenger Public

server: send push notifications concurrently (bounded worker pool)

sendToAll fanned out one blocking HTTPS send at a time — O(N) latency per message. Now sends run concurrently, bounded to 32, so latency is ~O(N/32); the per-send logic moved to sendOne (sent count is atomic; 404/410 pruning unchanged, SQLite serializes the deletes). The biggest lever for scaling a large group's push delivery.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Paul Campbell 822161d7c4d5ac38f3c77f303f6f9cc4a50331da parent 37f161b
1 file changed, +56 −31
  • push.go +56 −31
diff --git a/push.go b/push.go
index 27c8446..d429075 100644
--- a/push.go
+++ b/push.go
@@ -5,6 +5,8 @@ import (
"io"
"log"
"net/http"
+ "sync"
+ "sync/atomic"
"time"
webpush "github.com/SherClockHolmes/webpush-go"
@@ -97,8 +99,14 @@ func (s *Server) sendTestPush() int {
return s.sendToAll(payload, "test", "")
}
+// maxPushConcurrency bounds simultaneous sends so a large group doesn't open
+// thousands of sockets at once. Each webpush send is a blocking HTTPS round-trip,
+// so fanning out turns O(N) sequential latency into ~O(N/concurrency).
+const maxPushConcurrency = 32
+
// sendToAll delivers a payload to every subscription (skipping any whose token
-// equals exceptToken), pruning dead ones (404/410). Returns successful sends.
+// equals exceptToken), pruning dead ones (404/410). Sends run concurrently,
+// bounded by maxPushConcurrency. Returns the number of successful sends.
func (s *Server) sendToAll(payload []byte, kind, exceptToken string) int {
subs, err := s.db.listSubscriptions()
if err != nil {
@@ -106,39 +114,56 @@ func (s *Server) sendToAll(payload []byte, kind, exceptToken string) int {
return 0
}
log.Printf("push[%s]: %d subscription(s)", kind, len(subs))
- sent := 0
+
+ sem := make(chan struct{}, maxPushConcurrency)
+ var wg sync.WaitGroup
+ var sent atomic.Int64
for _, sub := range subs {
if exceptToken != "" && sub.Token == exceptToken {
continue // don't notify the sender's own devices
}
- resp, err := webpush.SendNotification(payload, &webpush.Subscription{
- Endpoint: sub.Endpoint,
- Keys: webpush.Keys{P256dh: sub.P256dh, Auth: sub.Auth},
- }, &webpush.Options{
- // Pass the bare e-mail: webpush-go prepends "mailto:" itself. Passing
- // "mailto:…" yields "mailto:mailto:…", which Apple rejects (BadJwtToken).
- Subscriber: "admin@luan.chat",
- VAPIDPublicKey: s.vapidPub,
- VAPIDPrivateKey: s.vapidPriv,
- TTL: 60,
- Urgency: webpush.UrgencyHigh,
- })
- if err != nil {
- log.Printf("push[%s]: send error: %v", kind, err)
- continue
- }
- if resp.StatusCode >= 200 && resp.StatusCode < 300 {
- sent++
- resp.Body.Close()
- log.Printf("push[%s]: status %d (%.45s…)", kind, resp.StatusCode, sub.Endpoint)
- continue
- }
- body, _ := io.ReadAll(io.LimitReader(resp.Body, 400))
- resp.Body.Close()
- log.Printf("push[%s]: status %d body=%q (%.45s…)", kind, resp.StatusCode, string(body), sub.Endpoint)
- if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
- _ = s.db.deleteSubscription(sub.Endpoint)
- }
+ wg.Add(1)
+ sem <- struct{}{}
+ go func(sub PushSub) {
+ defer wg.Done()
+ defer func() { <-sem }()
+ if s.sendOne(payload, kind, sub) {
+ sent.Add(1)
+ }
+ }(sub)
+ }
+ wg.Wait()
+ return int(sent.Load())
+}
+
+// sendOne delivers a payload to a single subscription, pruning it if the push
+// service reports it gone (404/410). Returns true on a 2xx.
+func (s *Server) sendOne(payload []byte, kind string, sub PushSub) bool {
+ resp, err := webpush.SendNotification(payload, &webpush.Subscription{
+ Endpoint: sub.Endpoint,
+ Keys: webpush.Keys{P256dh: sub.P256dh, Auth: sub.Auth},
+ }, &webpush.Options{
+ // Pass the bare e-mail: webpush-go prepends "mailto:" itself. Passing
+ // "mailto:…" yields "mailto:mailto:…", which Apple rejects (BadJwtToken).
+ Subscriber: "admin@luan.chat",
+ VAPIDPublicKey: s.vapidPub,
+ VAPIDPrivateKey: s.vapidPriv,
+ TTL: 60,
+ Urgency: webpush.UrgencyHigh,
+ })
+ if err != nil {
+ log.Printf("push[%s]: send error: %v", kind, err)
+ return false
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode >= 200 && resp.StatusCode < 300 {
+ log.Printf("push[%s]: status %d (%.45s…)", kind, resp.StatusCode, sub.Endpoint)
+ return true
+ }
+ body, _ := io.ReadAll(io.LimitReader(resp.Body, 400))
+ log.Printf("push[%s]: status %d body=%q (%.45s…)", kind, resp.StatusCode, string(body), sub.Endpoint)
+ if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
+ _ = s.db.deleteSubscription(sub.Endpoint)
}
- return sent
+ return false
}