Build the visitor per request instead of sharing it

Requests from one IP (anonymous and untiered users) shared a single visitor
object with a mutable user, so overlapping requests could be authorized as
each other's user. The visitor is now built per request and never changed;
the limit service, keyed by visitor key, still shares the buckets and counts.
The Firebase penalty becomes a limiter kind, open subscriptions move to a
node-local gauge per key, and tier users' stats are enqueued per admitted
unit.

Fixes #1051

server/server.go

@@ -64,9 +64,9 @@ type Server struct {
 	smtpServerBackend *smtpBackend
 	mailer            mail.Sender
 	topics            map[string]*topic
-	visitors          map[string]*visitor // ip:<ip> or user:<user>
-	limiter           *limit.Service      // Decides on and counts the visitors' usage, keyed by visitor key
-	ban               *ban.Service        // Abuse ban-feed; nil when the feature is disabled (no ban file)
+	limiter           *limit.Service     // Decides on and counts the visitors' usage, keyed by visitor key
+	subscriptions     *subscriptionGauge // Open subscriptions per visitor key on this node
+	ban               *ban.Service       // Abuse ban-feed; nil when the feature is disabled (no ban file)
 	firebaseClient    *firebaseClient
 	twilio            *twilio.Client
 	messages          int64                               // Total number of messages (persisted if messageCache enabled)
@@ -301,7 +301,7 @@ func New(conf *Config) (*Server, error) {
 		messages:        messages,
 		messagesHistory: []int64{messages},
 		messagesFlushed: messages, // The loaded total is already persisted; only new publishes are deltas
-		visitors:        make(map[string]*visitor),
+		subscriptions:   newSubscriptionGauge(),
 		limiter:         limit.New(&limit.Config{ResetTime: conf.VisitorStatsResetTime}),
 		stripe:          stripe,
 		streams:         newStreamRegistry(),
@@ -1116,10 +1116,6 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess
 			return nil, err
 		}
 	}
-	u := v.User()
-	if s.userManager != nil && u != nil && u.Tier != nil {
-		go v.EnqueueUserStatsDelta()
-	}
 	s.mu.Lock()
 	s.messages++
 	s.mu.Unlock()
@@ -2164,9 +2160,6 @@ func (s *Server) resetStats() {
 	s.mu.Lock()
 	defer s.mu.Unlock() // Includes the database query to avoid races with other processes
 	s.limiter.ResetUsage()
-	for _, v := range s.visitors {
-		v.ResetStats()
-	}
 	if s.userManager != nil {
 		if err := s.userManager.ResetStats(); err != nil {
 			log.Tag(tagResetter).Warn("Failed to write to database: %s", err.Error())
@@ -2178,7 +2171,7 @@ func (s *Server) runFirebaseKeepaliver() {
 	if s.firebaseClient == nil {
 		return
 	}
-	v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0
+	v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0
 	for {
 		select {
 		case <-time.After(s.config.FirebaseKeepaliveInterval):
@@ -2357,18 +2350,10 @@ func (s *Server) transformMatrixJSON(next handleFunc) handleFunc {
 	}
 }
 
+// visitor returns the visitor for this request. It is built per request, so concurrent requests from
+// the same IP (which share a limiter key) cannot see each other's user (#1051).
 func (s *Server) visitor(ip netip.Addr, user *user.User) *visitor {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	id := visitorID(ip, user, s.config)
-	v, exists := s.visitors[id]
-	if !exists {
-		s.visitors[id] = newVisitor(s.config, s.limiter, s.messageCache, s.userManager, ip, user)
-		return s.visitors[id]
-	}
-	v.Keepalive()
-	v.SetUser(user) // Always update with the latest user, may be nil!
-	return v
+	return newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, ip, user)
 }
 
 func (s *Server) writeJSON(w http.ResponseWriter, v any) error {