From ab826f635fffa09fd71181ea972c7973133d4004 Mon Sep 17 00:00:00 2001
From: Colinka <colinkanodes@gmail.com>
Date: Thu, 28 May 2026 11:27:26 +0000
Subject: [PATCH] fix: drain bidi stream response side in publisher to prevent
 deadlock

The publisher CLIs (cmd/multi-publish, cmd/single in publish mode) call
client.ListenCommands(ctx) and only invoke stream.Send() in a loop. They
never call stream.Recv(). Because ListenCommands is a bidirectional gRPC
stream and the server emits trace-event Response messages for every
published Request, the per-stream HTTP/2 flow-control window fills up,
the server's response handler blocks, and stream.Send() on the client
eventually blocks too. Result: a publisher that completes 7-8 messages
at payloads >=100KB and then hangs indefinitely.

Fix: spawn a goroutine that drains stream.Recv() until the stream
closes. This unblocks the response side without changing the proto or
the response handling on the client (the trace events are discarded,
which matches existing behaviour at the user-visible level).

Verified on getoptimum/p2pnode:v0.0.1-rc16 with the docker-compose-optimum
stack:

  before fix: -count=40 -datasize=102400 hung after 7 publications.
  after fix:  -count=40 -datasize=102400 completes in ~8s.
              -count=20 -datasize=1048576 completes in ~8s.
              -count=10 -datasize=2516582 completes in ~6s.
---
 grpc_p2p_client/cmd/multi-publish/main.go | 13 +++++++++++++
 grpc_p2p_client/cmd/single/main.go        | 12 ++++++++++++
 2 files changed, 25 insertions(+)

diff --git a/grpc_p2p_client/cmd/multi-publish/main.go b/grpc_p2p_client/cmd/multi-publish/main.go
index f59e048..c54f958 100644
--- a/grpc_p2p_client/cmd/multi-publish/main.go
+++ b/grpc_p2p_client/cmd/multi-publish/main.go
@@ -130,6 +130,19 @@ func sendMessages(ctx context.Context, ip string, datasize int, write bool, data
 		return fmt.Errorf("[%s] ListenCommands failed: %w", ip, err)
 	}
 
+	// Drain the response side of the bidi stream so the per-stream HTTP/2
+	// flow-control window does not fill up and block stream.Send. Without
+	// this drain the publisher deadlocks after ~7-8 publications at
+	// payload sizes >=100KB, because each publish triggers several
+	// trace-event Response messages from the server.
+	go func() {
+		for {
+			if _, err := stream.Recv(); err != nil {
+				return
+			}
+		}
+	}()
+
 	println(fmt.Sprintf("Connected to node at: %s…", ip))
 
 	for i := 0; i < *count; i++ {
diff --git a/grpc_p2p_client/cmd/single/main.go b/grpc_p2p_client/cmd/single/main.go
index 1b26ac1..d524b83 100644
--- a/grpc_p2p_client/cmd/single/main.go
+++ b/grpc_p2p_client/cmd/single/main.go
@@ -115,6 +115,18 @@ func publish(ctx context.Context, stream protobuf.CommandStream_ListenCommandsCl
 		log.Fatal("-msg is required in publish mode")
 	}
 
+	// Drain the response side of the bidi stream so the per-stream HTTP/2
+	// flow-control window does not fill up and block stream.Send. Without
+	// this drain a multi-message publish (-count >> 1 with large -msg)
+	// deadlocks after ~7-8 publications at payload sizes >=100KB.
+	go func() {
+		for {
+			if _, err := stream.Recv(); err != nil {
+				return
+			}
+		}
+	}()
+
 	for i := 0; i < count; i++ {
 		start := time.Now()
 		var data []byte