Merge pull request #21 from bluesky-social/jetstream_limiters

Add per-IP rate limiters

Author: jazware

pkg/server/server.go

@@ -17,15 +17,17 @@ import (
 	"github.com/labstack/echo/v4"
 	"go.opentelemetry.io/otel"
 	"golang.org/x/sync/semaphore"
+	"golang.org/x/time/rate"
 )
 
 type Server struct {
-	Subscribers map[int64]*Subscriber
-	lk          sync.RWMutex
-	nextSub     int64
-	Consumer    *consumer.Consumer
-	maxSubRate  float64
-	seq         int64
+	Subscribers   map[int64]*Subscriber
+	lk            sync.RWMutex
+	nextSub       int64
+	Consumer      *consumer.Consumer
+	maxSubRate    float64
+	seq           int64
+	perIPLimiters map[string]*rate.Limiter
 }
 
 var upgrader = websocket.Upgrader{
@@ -40,8 +42,9 @@ var tracer = otel.Tracer("jetstream-server")
 
 func NewServer(maxSubRate float64) (*Server, error) {
 	s := Server{
-		Subscribers: make(map[int64]*Subscriber),
-		maxSubRate:  maxSubRate,
+		Subscribers:   make(map[int64]*Subscriber),
+		maxSubRate:    maxSubRate,
+		perIPLimiters: make(map[string]*rate.Limiter),
 	}
 
 	return &s, nil

pkg/server/subscriber.go

@@ -204,6 +204,12 @@ func (s *Server) AddSubscriber(ws *websocket.Conn, realIP string, opts *Subscrib
 	s.lk.Lock()
 	defer s.lk.Unlock()
 
+	lim := s.perIPLimiters[realIP]
+	if lim == nil {
+		lim = rate.NewLimiter(rate.Limit(s.maxSubRate), int(s.maxSubRate))
+		s.perIPLimiters[realIP] = lim
+	}
+
 	sub := Subscriber{
 		ws:                ws,
 		realIP:            realIP,
@@ -216,7 +222,7 @@ func (s *Server) AddSubscriber(ws *websocket.Conn, realIP string, opts *Subscrib
 		compress:          opts.Compress,
 		deliveredCounter:  eventsDelivered.WithLabelValues(realIP),
 		bytesCounter:      bytesDelivered.WithLabelValues(realIP),
-		rl:                rate.NewLimiter(rate.Limit(s.maxSubRate), int(s.maxSubRate)),
+		rl:                lim,
 	}
 
 	s.Subscribers[s.nextSub] = &sub