perf(brokerwire): decode stream data without a second copy

proto.Unmarshal already copies bytes fields out of the payload, so the
decoded chunk is handed to the stream pipe as is. A test pins that the
decoded data never aliases the payload.

commit: ff9bba96523af7339a62f00398c27ef595a0acb7
author: brice · 14d ago · 2026-10-08 15:05 UTC
committer: brice · 14d ago · 2026-10-08 15:05 UTC
parent: aa93a57
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +24 -4

M internal/adapters/brokeripc/limits_test.go +1 -1


@@ -173,7 +173,7 @@ func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
173	173		require.ErrorIs(t, parked.deliver([]byte("late")), ErrStreamGone)
174	174		}
175	175
176		-	// TestStreamPipeOutboundError proves a failing outbound hook surfaces instead of
	176	+	// TestStreamPipeOutboundError proves a failing sendFrame hook surfaces instead of
177	177		// silently dropping bytes.
178	178		func TestStreamPipeOutboundError(t *testing.T) {
179	179		failure := errors.New("outbound failed")


@@ -173,7 +173,7 @@ func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
173		require.ErrorIs(t, parked.deliver([]byte("late")), ErrStreamGone)	173		require.ErrorIs(t, parked.deliver([]byte("late")), ErrStreamGone)
174		}	174		}
175			175
176	-	// TestStreamPipeOutboundError proves a failing outbound hook surfaces instead of	176	+	// TestStreamPipeOutboundError proves a failing sendFrame hook surfaces instead of
177		// silently dropping bytes.	177		// silently dropping bytes.
178		func TestStreamPipeOutboundError(t *testing.T) {	178		func TestStreamPipeOutboundError(t *testing.T) {
179		failure := errors.New("outbound failed")	179		failure := errors.New("outbound failed")

M internal/adapters/brokeripc/session_test.go +1 -1


@@ -911,7 +911,7 @@ func TestStalledClientStreamWaitsForCredit(t *testing.T) {
911	911		}
912	912		require.Eventually(t, func() bool { return held() > full }, 5*time.Second, time.Millisecond,
913	913		"the relay must fill one stream window")
914		-	require.LessOrEqual(t, held(), brokerwire.StreamWindowBytes, "the relay never exceeds its credit")
	914	+	// Data beyond the credit would settle the stream with ErrStreamCredit.
915	915		select {
916	916		case <-stream.Done():
917	917		t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())


@@ -911,7 +911,7 @@ func TestStalledClientStreamWaitsForCredit(t *testing.T) {
911		}	911		}
912		require.Eventually(t, func() bool { return held() > full }, 5*time.Second, time.Millisecond,	912		require.Eventually(t, func() bool { return held() > full }, 5*time.Second, time.Millisecond,
913		"the relay must fill one stream window")	913		"the relay must fill one stream window")
914	-	require.LessOrEqual(t, held(), brokerwire.StreamWindowBytes, "the relay never exceeds its credit")	914	+	// Data beyond the credit would settle the stream with ErrStreamCredit.
915		select {	915		select {
916		case <-stream.Done():	916		case <-stream.Done():
917		t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())	917		t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())

M internal/adapters/brokerwire/codec.go +6 -2


@@ -791,7 +791,9 @@ func clientStreamDataFromWire(message *wire.ClientStreamData, maxChunkBytes uint
791	791		if len(data) == 0 {
792	792		return ClientStreamData{}, ErrInvalidMessage
793	793		}
794		-	return ClientStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: append([]byte(nil), data...)}, nil
	794	+	// proto.Unmarshal copies bytes fields out of the payload, so the caller
	795	+	// owns data without another copy.
	796	+	return ClientStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: data}, nil
795	797		}
796	798
797	799		func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.BrokerServerEnvelope, error) {

@@ -1545,7 +1547,9 @@ func serverStreamDataFromWire(message *wire.ServerStreamData, maxChunkBytes uint
1545	1547		if len(data) == 0 {
1546	1548		return ServerStreamData{}, ErrInvalidMessage
1547	1549		}
1548		-	return ServerStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: append([]byte(nil), data...)}, nil
	1550	+	// proto.Unmarshal copies bytes fields out of the payload, so the caller
	1551	+	// owns data without another copy.
	1552	+	return ServerStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: data}, nil
1549	1553		}
1550	1554
1551	1555		func streamClosedToWire(m StreamClosed) (*wire.StreamClosed, error) {

 	if len(data) == 0 {
 		return ClientStreamData{}, ErrInvalidMessage
 	}
-	return ClientStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: append([]byte(nil), data...)}, nil
+	// proto.Unmarshal copies bytes fields out of the payload, so the caller
+	// owns data without another copy.
+	return ClientStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: data}, nil
 }
 func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.BrokerServerEnvelope, error) {
 	if len(data) == 0 {
 		return ServerStreamData{}, ErrInvalidMessage
 	}
-	return ServerStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: append([]byte(nil), data...)}, nil
+	// proto.Unmarshal copies bytes fields out of the payload, so the caller
+	// owns data without another copy.
+	return ServerStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: data}, nil
 }
 func streamClosedToWire(m StreamClosed) (*wire.StreamClosed, error) {

M internal/adapters/brokerwire/codec_test.go +16 -0


@@ -377,6 +377,22 @@ func TestBrokerClientVariants(t *testing.T) {
377	377		}
378	378		}
379	379
	380	+	// TestBrokerStreamDataOwnsItsBytes proves decoded stream data never aliases the
	381	+	// payload: receivers reuse their buffers, and brokeripc queues Data as is.
	382	+	func TestBrokerStreamDataOwnsItsBytes(t *testing.T) {
	383	+	connection := testConnectionID(0x11)
	384	+	client := mustEncodeClient(t, ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")})
	385	+	server := mustEncodeServer(t, ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")})
	386	+	decodedClient, err := DecodeClient(client, testEnvelopeCeiling, testChunkCeiling)
	387	+	require.NoError(t, err)
	388	+	decodedServer, err := DecodeServer(server, testEnvelopeCeiling, testChunkCeiling)
	389	+	require.NoError(t, err)
	390	+	clear(client)
	391	+	clear(server)
	392	+	require.Equal(t, []byte("frame"), decodedClient.(ClientStreamData).Data)
	393	+	require.Equal(t, []byte("frame"), decodedServer.(ServerStreamData).Data)
	394	+	}
	395	+
380	396		// TestBrokerOpenStreamAdmissionVariants pins the attachment-admission
381	397		// contract: exact, create-named, and create-ephemeral round-trip, control
382	398		// carries none, and wrong admission/purpose/name/target combinations are


@@ -377,6 +377,22 @@ func TestBrokerClientVariants(t *testing.T) {
377	}	377		}
378	}	378		}
379		379
		380	+	// TestBrokerStreamDataOwnsItsBytes proves decoded stream data never aliases the
		381	+	// payload: receivers reuse their buffers, and brokeripc queues Data as is.
		382	+	func TestBrokerStreamDataOwnsItsBytes(t *testing.T) {
		383	+	connection := testConnectionID(0x11)
		384	+	client := mustEncodeClient(t, ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")})
		385	+	server := mustEncodeServer(t, ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")})
		386	+	decodedClient, err := DecodeClient(client, testEnvelopeCeiling, testChunkCeiling)
		387	+	require.NoError(t, err)
		388	+	decodedServer, err := DecodeServer(server, testEnvelopeCeiling, testChunkCeiling)
		389	+	require.NoError(t, err)
		390	+	clear(client)
		391	+	clear(server)
		392	+	require.Equal(t, []byte("frame"), decodedClient.(ClientStreamData).Data)
		393	+	require.Equal(t, []byte("frame"), decodedServer.(ServerStreamData).Data)
		394	+	}
		395	+
380	// TestBrokerOpenStreamAdmissionVariants pins the attachment-admission	396		// TestBrokerOpenStreamAdmissionVariants pins the attachment-admission
381	// contract: exact, create-named, and create-ephemeral round-trip, control	397		// contract: exact, create-named, and create-ephemeral round-trip, control
382	// carries none, and wrong admission/purpose/name/target combinations are	398		// carries none, and wrong admission/purpose/name/target combinations are