Do not send header twice via malfeasance stream

spacemeshos · Feb 26, 2025 · 0166f4e · 0166f4e
1 parent fdd243d
commit 0166f4e
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 22 deletions.
diff --git a/api/grpcserver/v2alpha1/malfeasance.go b/api/grpcserver/v2alpha1/malfeasance.go
@@ -145,6 +145,20 @@ func (s *MalfeasanceStreamService) Stream(
 	request *spacemeshv2alpha1.MalfeasanceStreamRequest,
 	stream spacemeshv2alpha1.MalfeasanceStreamService_StreamServer,
 ) error {
+	var sub *events.BufferedSubscription[events.EventMalfeasance]
+	if request.Watch {
+		matcher := malfeasanceMatcher{request}
+		var err error
+		sub, err = events.SubscribeMatched(matcher.match)
+		if err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+		defer sub.Close()
+		if err := stream.SendHeader(metadata.MD{}); err != nil {
+			return err
+		}
+	}
+
 	legacyProofs, err := fetchLegacyFromDB(
 		stream.Context(),
 		s.db,
@@ -187,20 +201,9 @@ func (s *MalfeasanceStreamService) Stream(
 		return nil
 	}
 
-	matcher := malfeasanceMatcher{request}
-	sub, err := events.SubscribeMatched(matcher.match)
-	if err != nil {
-		return status.Error(codes.Internal, err.Error())
-	}
-	defer sub.Close()
 	eventsOut := sub.Out()
 	eventsFull := sub.Full()
 
-	if err := stream.SendHeader(metadata.MD{}); err != nil {
-		ctxzap.Debug(stream.Context(), "failed to send stream header",
-			zap.Error(err),
-		)
-	}
 	for {
 		select {
 		// process pending events first

diff --git a/api/grpcserver/v2beta1/malfeasance.go b/api/grpcserver/v2beta1/malfeasance.go
@@ -145,6 +145,20 @@ func (s *MalfeasanceStreamService) Stream(
 	request *spacemeshv2beta1.MalfeasanceStreamRequest,
 	stream spacemeshv2beta1.MalfeasanceStreamService_StreamServer,
 ) error {
+	var sub *events.BufferedSubscription[events.EventMalfeasance]
+	if request.Watch {
+		matcher := malfeasanceMatcher{request}
+		var err error
+		sub, err = events.SubscribeMatched(matcher.match)
+		if err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+		defer sub.Close()
+		if err := stream.SendHeader(metadata.MD{}); err != nil {
+			return err
+		}
+	}
+
 	legacyProofs, err := fetchLegacyFromDB(
 		stream.Context(),
 		s.db,
@@ -187,20 +201,9 @@ func (s *MalfeasanceStreamService) Stream(
 		return nil
 	}
 
-	matcher := malfeasanceMatcher{request}
-	sub, err := events.SubscribeMatched(matcher.match)
-	if err != nil {
-		return status.Error(codes.Internal, err.Error())
-	}
-	defer sub.Close()
 	eventsOut := sub.Out()
 	eventsFull := sub.Full()
 
-	if err := stream.SendHeader(metadata.MD{}); err != nil {
-		ctxzap.Debug(stream.Context(), "failed to send stream header",
-			zap.Error(err),
-		)
-	}
 	for {
 		select {
 		// process pending events first