Jelajahi Sumber

fix: gRPC handler must not mutate shared Deps.Sources (data race)

Multiple gRPC goroutines sharing s.Dep were concurrently reading
s.Dep.Sources inside pipeline.Process while another goroutine was
overwriting it with a single-source map and restoring on return.
Under load (10k/s gRPC soak) this triggers Go's runtime:

  fatal error: concurrent map read and map write

which kills the ingestd process and takes down every active stream —
matching the observed symptom where one of two loadgen-grpc instances
would die while the other kept running.

Fix: take a value copy of s.Dep at the top of processAlert and only
mutate the local copy. The underlying *Deduper / *Limiter / *MaxSeen /
JetStream interface are pointer/interface-typed, so the copy shares
those instances — which is what we want, they are all safe for
concurrent use.
Luis Rosales 1 bulan lalu
induk
melakukan
d966391706
1 mengubah file dengan 5 tambahan dan 7 penghapusan
  1. 5 7
      internal/grpcserver/handler.go

+ 5 - 7
internal/grpcserver/handler.go

@@ -193,15 +193,13 @@ func (s *Server) processAlert(ctx context.Context, alert *pbv1.Alert, src *pipel
 	}
 
 	// Override the Sources map for this call to use the authenticated source.
-	// This ensures company_id/source_id from the API key match the alert body.
+	// Per-call Sources map — passed in, never mutates s.Dep.Sources. Avoids
+	// concurrent map read/write between gRPC goroutines sharing Deps.
 	sourceKey := src.CompanyID + ":" + src.SourceID
-	origSources := s.Dep.Sources
-	s.Dep.Sources = map[string]pipeline.SourceConfig{sourceKey: *src}
+	callDeps := s.Dep // value copy; underlying pointers/maps are shared safely
+	callDeps.Sources = map[string]pipeline.SourceConfig{sourceKey: *src}
 
-	res := s.Dep.Process(ctx, body, "" /* no HMAC for gRPC */)
-
-	// Restore the original Sources map.
-	s.Dep.Sources = origSources
+	res := callDeps.Process(ctx, body, "" /* no HMAC for gRPC */)
 
 	if !res.Accepted {
 		code := errorCode(res.RejectReason)