fix(brokeripc): deliver replies queued before an orderly stream close

An orderly close discarded queued chunks and a normal core end settled the stream before the relay forwarded its reply, so one-shot operations like vev ls failed with 'reply lost' on most calls.

commit: aa991bfe6503fa5b5a11c26388ea17f94b1ac372
author: brice · 30d ago · 2026-09-22 20:09 UTC
parent: 11ffa7d
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +211 -6

M internal/adapters/brokeripc/helpers_test.go +10 -0


@@ -309,6 +309,16 @@ func (c *fakeLogicalConn) SendClient(message protocol.ClientMessage) error {
309	309		}
310	310
311	311		func (c *fakeLogicalConn) ReceiveServer() (protocol.ServerMessage, error) {
	312	+	// A message already queued when Close ran must still be observed first: a
	313	+	// real carriage never reports the connection done before a message the peer
	314	+	// sent ahead of an orderly close has been delivered. Draining non-blocking
	315	+	// before the select keeps that ordering deterministic instead of racing a
	316	+	// buffered send against an already-closed done channel.
	317	+	select {
	318	+	case message := <-c.toClient:
	319	+	return message, nil
	320	+	default:
	321	+	}
312	322		select {
313	323		case message := <-c.toClient:
314	324		return message, nil


@@ -309,6 +309,16 @@ func (c *fakeLogicalConn) SendClient(message protocol.ClientMessage) error {
309	}	309		}
310		310
311	func (c *fakeLogicalConn) ReceiveServer() (protocol.ServerMessage, error) {	311		func (c *fakeLogicalConn) ReceiveServer() (protocol.ServerMessage, error) {
		312	+	// A message already queued when Close ran must still be observed first: a
		313	+	// real carriage never reports the connection done before a message the peer
		314	+	// sent ahead of an orderly close has been delivered. Draining non-blocking
		315	+	// before the select keeps that ordering deterministic instead of racing a
		316	+	// buffered send against an already-closed done channel.
		317	+	select {
		318	+	case message := <-c.toClient:
		319	+	return message, nil
		320	+	default:
		321	+	}
312	select {	322		select {
313	case message := <-c.toClient:	323		case message := <-c.toClient:
314	return message, nil	324		return message, nil

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


@@ -168,9 +168,21 @@ func TestStreamPipeBoundsAndClose(t *testing.T) {
168	168		require.Equal(t, "one", string(buf[:n]))
169	169		require.NoError(t, pipe.deliver([]byte("three")))
170	170
	171	+	// An orderly Close keeps what is already queued ("two" and "three"): Read
	172	+	// drains it before observing io.EOF, so a chunk delivered just ahead of an
	173	+	// orderly close is never discarded underneath its reader (see
	174	+	// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose for the focused
	175	+	// coverage of this behavior).
171	176		require.NoError(t, pipe.Close())
172	177		require.ErrorIs(t, pipe.deliver([]byte("four")), ErrStreamGone)
173		-	_, err = pipe.Read(buf)
	178	+	drained := make([]byte, 8)
	179	+	n, err = pipe.Read(drained)
	180	+	require.NoError(t, err)
	181	+	require.Equal(t, "two", string(drained[:n]))
	182	+	n, err = pipe.Read(drained)
	183	+	require.NoError(t, err)
	184	+	require.Equal(t, "three", string(drained[:n]))
	185	+	_, err = pipe.Read(drained)
174	186		require.ErrorIs(t, err, io.EOF)
175	187
176	188		// A reader parked on an empty pipe is released by closeWith with its cause.


@@ -168,9 +168,21 @@ func TestStreamPipeBoundsAndClose(t *testing.T) {
168		require.Equal(t, "one", string(buf[:n]))	168		require.Equal(t, "one", string(buf[:n]))
169		require.NoError(t, pipe.deliver([]byte("three")))	169		require.NoError(t, pipe.deliver([]byte("three")))
170			170
			171	+	// An orderly Close keeps what is already queued ("two" and "three"): Read
			172	+	// drains it before observing io.EOF, so a chunk delivered just ahead of an
			173	+	// orderly close is never discarded underneath its reader (see
			174	+	// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose for the focused
			175	+	// coverage of this behavior).
171		require.NoError(t, pipe.Close())	176		require.NoError(t, pipe.Close())
172		require.ErrorIs(t, pipe.deliver([]byte("four")), ErrStreamGone)	177		require.ErrorIs(t, pipe.deliver([]byte("four")), ErrStreamGone)
173	-	_, err = pipe.Read(buf)	178	+	drained := make([]byte, 8)
			179	+	n, err = pipe.Read(drained)
			180	+	require.NoError(t, err)
			181	+	require.Equal(t, "two", string(drained[:n]))
			182	+	n, err = pipe.Read(drained)
			183	+	require.NoError(t, err)
			184	+	require.Equal(t, "three", string(drained[:n]))
			185	+	_, err = pipe.Read(drained)
174		require.ErrorIs(t, err, io.EOF)	186		require.ErrorIs(t, err, io.EOF)
175			187
176		// A reader parked on an empty pipe is released by closeWith with its cause.	188		// A reader parked on an empty pipe is released by closeWith with its cause.

M internal/adapters/brokeripc/pipe.go +10 -3


@@ -128,7 +128,12 @@ func (p *streamPipe) deliver(chunk []byte) error {
128	128		}
129	129
130	130		// closeWith settles the pipe exactly once with cause. A nil cause is an orderly
131		-	// close; every blocked Read observes it promptly.
	131	+	// close; every blocked Read observes it promptly. On a nil cause, any chunks
	132	+	// already queued are kept: Read serves them first and only returns io.EOF once
	133	+	// the queue is drained, so a reply that arrived just before an orderly peer
	134	+	// close is never discarded underneath its reader. An error cause is a failure:
	135	+	// the queue is discarded, since its content can no longer be trusted to reach
	136	+	// a consumer that is being torn down for cause.
132	137		func (p *streamPipe) closeWith(cause error) {
133	138		p.mu.Lock()
134	139		if p.closed {

@@ -137,8 +142,10 @@ func (p *streamPipe) closeWith(cause error) {
137	142		}
138	143		p.closed = true
139	144		p.err = cause
140		-	p.queue = nil
141		-	p.queued = 0
	145	+	if cause != nil {
	146	+	p.queue = nil
	147	+	p.queued = 0
	148	+	}
142	149		p.mu.Unlock()
143	150		close(p.closedCh)
144	151		select {

 }
 // closeWith settles the pipe exactly once with cause. A nil cause is an orderly
-// close; every blocked Read observes it promptly.
+// close; every blocked Read observes it promptly. On a nil cause, any chunks
+// already queued are kept: Read serves them first and only returns io.EOF once
+// the queue is drained, so a reply that arrived just before an orderly peer
+// close is never discarded underneath its reader. An error cause is a failure:
+// the queue is discarded, since its content can no longer be trusted to reach
+// a consumer that is being torn down for cause.
 func (p *streamPipe) closeWith(cause error) {
 	p.mu.Lock()
 	if p.closed {
 	}
 	p.closed = true
 	p.err = cause
-	p.queue = nil
+	if cause != nil {
-	p.queued = 0
+		p.queue = nil
+		p.queued = 0
+	}
 	p.mu.Unlock()
 	close(p.closedCh)
 	select {

A internal/adapters/brokeripc/pipe_test.go +108 -0

@@ -0,0 +1,108 @@
	1	+	package brokeripc
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"errors"
	6	+	"io"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+	)
	11	+
	12	+	// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose proves closeWith
	13	+	// keeps chunks that are already queued when its cause is nil, an orderly
	14	+	// close: Read drains every queued byte before it returns io.EOF. This is
	15	+	// exactly what a reply queued just ahead of an orderly StreamClosed needs -
	16	+	// the reply chunk must survive the close instead of being thrown away
	17	+	// underneath its reader, which is root cause 1 of the broker reply loss
	18	+	// ("vev ls" reporting a lost/unknown outcome on a successful call). An error
	19	+	// cause is the opposite: the queued content can no longer be trusted once the
	20	+	// pipe is torn down for cause, so it is discarded and Read observes the error
	21	+	// immediately instead of replaying stale data.
	22	+	func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
	23	+	errCause := errors.New("stream pipe test: cause")
	24	+
	25	+	tests := []struct {
	26	+	name string
	27	+	chunks [][]byte
	28	+	cause error
	29	+	readSize int
	30	+	wantData []byte
	31	+	wantErr error
	32	+	}{
	33	+	{
	34	+	name: "orderly close keeps one queued chunk",
	35	+	chunks: [][]byte{[]byte("hello")},
	36	+	cause: nil,
	37	+	readSize: 64,
	38	+	wantData: []byte("hello"),
	39	+	wantErr: io.EOF,
	40	+	},
	41	+	{
	42	+	name: "orderly close keeps three chunks split across a frame boundary",
	43	+	chunks: [][]byte{[]byte("AB"), []byte("CD"), []byte("EF")},
	44	+	cause: nil,
	45	+	// Smaller than the combined payload, so one logical frame's worth of
	46	+	// queued data is drained across several Read calls, exactly like
	47	+	// io.ReadFull reading a header then a body out of the same queue.
	48	+	readSize: 3,
	49	+	wantData: []byte("ABCDEF"),
	50	+	wantErr: io.EOF,
	51	+	},
	52	+	{
	53	+	name: "error cause discards the queue and Read returns the error",
	54	+	chunks: [][]byte{[]byte("hello")},
	55	+	cause: errCause,
	56	+	readSize: 64,
	57	+	wantData: nil,
	58	+	wantErr: errCause,
	59	+	},
	60	+	{
	61	+	name: "orderly close with no queued data is immediate EOF",
	62	+	chunks: nil,
	63	+	cause: nil,
	64	+	readSize: 64,
	65	+	wantData: nil,
	66	+	wantErr: io.EOF,
	67	+	},
	68	+	}
	69	+
	70	+	for _, tt := range tests {
	71	+	t.Run(tt.name, func(t *testing.T) {
	72	+	p := newStreamPipe(0, 0, func([]byte) error { return nil })
	73	+	for _, chunk := range tt.chunks {
	74	+	require.NoError(t, p.deliver(chunk))
	75	+	}
	76	+	p.closeWith(tt.cause)
	77	+
	78	+	var got bytes.Buffer
	79	+	buf := make([]byte, tt.readSize)
	80	+	var readErr error
	81	+	for {
	82	+	n, err := p.Read(buf)
	83	+	got.Write(buf[:n])
	84	+	if err != nil {
	85	+	readErr = err
	86	+	break
	87	+	}
	88	+	}
	89	+	require.Equal(t, tt.wantData, got.Bytes())
	90	+	if errors.Is(tt.wantErr, io.EOF) {
	91	+	require.ErrorIs(t, readErr, io.EOF)
	92	+	} else {
	93	+	require.Equal(t, tt.wantErr, readErr)
	94	+	}
	95	+
	96	+	// closeWith is idempotent: a second call with a different cause never
	97	+	// resurrects a discarded queue or changes the already-observed outcome.
	98	+	p.closeWith(errors.New("stream pipe test: second cause"))
	99	+	n, err := p.Read(buf)
	100	+	require.Zero(t, n)
	101	+	if errors.Is(tt.wantErr, io.EOF) {
	102	+	require.ErrorIs(t, err, io.EOF)
	103	+	} else {
	104	+	require.Equal(t, tt.wantErr, err)
	105	+	}
	106	+	})
	107	+	}
	108	+	}

@@ -0,0 +1,108 @@
	1	+	package brokeripc
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"errors"
	6	+	"io"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+	)
	11	+
	12	+	// TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose proves closeWith
	13	+	// keeps chunks that are already queued when its cause is nil, an orderly
	14	+	// close: Read drains every queued byte before it returns io.EOF. This is
	15	+	// exactly what a reply queued just ahead of an orderly StreamClosed needs -
	16	+	// the reply chunk must survive the close instead of being thrown away
	17	+	// underneath its reader, which is root cause 1 of the broker reply loss
	18	+	// ("vev ls" reporting a lost/unknown outcome on a successful call). An error
	19	+	// cause is the opposite: the queued content can no longer be trusted once the
	20	+	// pipe is torn down for cause, so it is discarded and Read observes the error
	21	+	// immediately instead of replaying stale data.
	22	+	func TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose(t *testing.T) {
	23	+	errCause := errors.New("stream pipe test: cause")
	24	+
	25	+	tests := []struct {
	26	+	name string
	27	+	chunks [][]byte
	28	+	cause error
	29	+	readSize int
	30	+	wantData []byte
	31	+	wantErr error
	32	+	}{
	33	+	{
	34	+	name: "orderly close keeps one queued chunk",
	35	+	chunks: [][]byte{[]byte("hello")},
	36	+	cause: nil,
	37	+	readSize: 64,
	38	+	wantData: []byte("hello"),
	39	+	wantErr: io.EOF,
	40	+	},
	41	+	{
	42	+	name: "orderly close keeps three chunks split across a frame boundary",
	43	+	chunks: [][]byte{[]byte("AB"), []byte("CD"), []byte("EF")},
	44	+	cause: nil,
	45	+	// Smaller than the combined payload, so one logical frame's worth of
	46	+	// queued data is drained across several Read calls, exactly like
	47	+	// io.ReadFull reading a header then a body out of the same queue.
	48	+	readSize: 3,
	49	+	wantData: []byte("ABCDEF"),
	50	+	wantErr: io.EOF,
	51	+	},
	52	+	{
	53	+	name: "error cause discards the queue and Read returns the error",
	54	+	chunks: [][]byte{[]byte("hello")},
	55	+	cause: errCause,
	56	+	readSize: 64,
	57	+	wantData: nil,
	58	+	wantErr: errCause,
	59	+	},
	60	+	{
	61	+	name: "orderly close with no queued data is immediate EOF",
	62	+	chunks: nil,
	63	+	cause: nil,
	64	+	readSize: 64,
	65	+	wantData: nil,
	66	+	wantErr: io.EOF,
	67	+	},
	68	+	}
	69	+
	70	+	for _, tt := range tests {
	71	+	t.Run(tt.name, func(t *testing.T) {
	72	+	p := newStreamPipe(0, 0, func([]byte) error { return nil })
	73	+	for _, chunk := range tt.chunks {
	74	+	require.NoError(t, p.deliver(chunk))
	75	+	}
	76	+	p.closeWith(tt.cause)
	77	+
	78	+	var got bytes.Buffer
	79	+	buf := make([]byte, tt.readSize)
	80	+	var readErr error
	81	+	for {
	82	+	n, err := p.Read(buf)
	83	+	got.Write(buf[:n])
	84	+	if err != nil {
	85	+	readErr = err
	86	+	break
	87	+	}
	88	+	}
	89	+	require.Equal(t, tt.wantData, got.Bytes())
	90	+	if errors.Is(tt.wantErr, io.EOF) {
	91	+	require.ErrorIs(t, readErr, io.EOF)
	92	+	} else {
	93	+	require.Equal(t, tt.wantErr, readErr)
	94	+	}
	95	+
	96	+	// closeWith is idempotent: a second call with a different cause never
	97	+	// resurrects a discarded queue or changes the already-observed outcome.
	98	+	p.closeWith(errors.New("stream pipe test: second cause"))
	99	+	n, err := p.Read(buf)
	100	+	require.Zero(t, n)
	101	+	if errors.Is(tt.wantErr, io.EOF) {
	102	+	require.ErrorIs(t, err, io.EOF)
	103	+	} else {
	104	+	require.Equal(t, tt.wantErr, err)
	105	+	}
	106	+	})
	107	+	}
	108	+	}

M internal/adapters/brokeripc/session_test.go +60 -0


@@ -3,6 +3,7 @@ package brokeripc
3	3		import (
4	4		"context"
5	5		"errors"
	6	+	"io"
6	7		"strings"
7	8		"testing"
8	9		"time"

@@ -368,6 +369,65 @@ func TestOpenStreamCarriesTypedTraffic(t *testing.T) {
368	369		}
369	370		}
370	371
	372	+	// TestReplyReachesClientBeforeOrderlyStreamClose proves a reply the broker core
	373	+	// sends immediately before an orderly stream end (a StreamClosed carrying no
	374	+	// error) always reaches the client's ReceiveServer before it observes EOF.
	375	+	// This is the exact end-to-end sequence that dropped about 70% of `vev ls`
	376	+	// calls: the daemon's handleList receives the request, does
	377	+	// SendServer(Sessions) then Close, so the broker forwards protocol.Sessions and
	378	+	// immediately sees the core end orderly. Before the fix, watchCore settled the
	379	+	// stream (closing the client-side pipe and discarding its queue) racing the
	380	+	// client's read of the already-forwarded reply; the client observed io.EOF
	381	+	// instead of the reply and reported the broker operation lost or its outcome
	382	+	// unknown. It is run repeatedly under -race because the race window is narrow,
	383	+	// not guaranteed on a single pass.
	384	+	//
	385	+	// The request is sent and its receipt is confirmed before the daemon answers,
	386	+	// exactly like every real caller (BrokerOperations always calls SendClient
	387	+	// before ReceiveServer): a reply pushed before the daemon has even received
	388	+	// the request races this fake's own SendClient/Done selection, which is a
	389	+	// harness artifact unrelated to the fix under test.
	390	+	func TestReplyReachesClientBeforeOrderlyStreamClose(t *testing.T) {
	391	+	e := startEndpoint(t, Config{})
	392	+	const iterations = 200
	393	+	for i := 0; i < iterations; i++ {
	394	+	client, _ := e.pair()
	395	+	core := e.authority.last()
	396	+
	397	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	398	+	stream, err := client.OpenStream(ctx, openRequest(1))
	399	+	require.NoError(t, err)
	400	+
	401	+	conn := core.logicalConn(1)
	402	+	require.NotNil(t, conn, "iteration %d: the broker core must admit exactly one stream", i)
	403	+
	404	+	require.NoError(t, stream.SendClient(protocol.List{}), "iteration %d", i)
	405	+	select {
	406	+	case <-conn.fromClient:
	407	+	case <-time.After(5 * time.Second):
	408	+	t.Fatalf("iteration %d: request never reached the daemon", i)
	409	+	}
	410	+
	411	+	// The daemon answers and, back to back, ends the stream orderly: exactly
	412	+	// the sequence SendServer(Sessions) then Close produces.
	413	+	conn.toClient <- protocol.Sessions{}
	414	+	require.NoError(t, conn.Close())
	415	+
	416	+	message, err := stream.ReceiveServer()
	417	+	require.NoError(t, err, "iteration %d: the reply must reach the client before EOF", i)
	418	+	require.Equal(t, protocol.Sessions{}, message, "iteration %d", i)
	419	+
	420	+	// The stream settles orderly right after: a further receive observes a
	421	+	// clean EOF, never an error, and never blocks.
	422	+	_, err = stream.ReceiveServer()
	423	+	require.ErrorIs(t, err, io.EOF, "iteration %d", i)
	424	+
	425	+	_ = stream.Close()
	426	+	_ = client.Close()
	427	+	cancel()
	428	+	}
	429	+	}
	430	+
371	431		// TestOpenStreamRefusedByCoreReachesClient proves a core refusal settles only
372	432		// that stream and reaches the caller as a typed failure.
373	433		func TestOpenStreamRefusedByCoreReachesClient(t *testing.T) {


@@ -3,6 +3,7 @@ package brokeripc
3	import (	3		import (
4	"context"	4		"context"
5	"errors"	5		"errors"
		6	+	"io"
6	"strings"	7		"strings"
7	"testing"	8		"testing"
8	"time"	9		"time"

@@ -368,6 +369,65 @@ func TestOpenStreamCarriesTypedTraffic(t *testing.T) {
368	}	369		}
369	}	370		}
370		371
		372	+	// TestReplyReachesClientBeforeOrderlyStreamClose proves a reply the broker core
		373	+	// sends immediately before an orderly stream end (a StreamClosed carrying no
		374	+	// error) always reaches the client's ReceiveServer before it observes EOF.
		375	+	// This is the exact end-to-end sequence that dropped about 70% of `vev ls`
		376	+	// calls: the daemon's handleList receives the request, does
		377	+	// SendServer(Sessions) then Close, so the broker forwards protocol.Sessions and
		378	+	// immediately sees the core end orderly. Before the fix, watchCore settled the
		379	+	// stream (closing the client-side pipe and discarding its queue) racing the
		380	+	// client's read of the already-forwarded reply; the client observed io.EOF
		381	+	// instead of the reply and reported the broker operation lost or its outcome
		382	+	// unknown. It is run repeatedly under -race because the race window is narrow,
		383	+	// not guaranteed on a single pass.
		384	+	//
		385	+	// The request is sent and its receipt is confirmed before the daemon answers,
		386	+	// exactly like every real caller (BrokerOperations always calls SendClient
		387	+	// before ReceiveServer): a reply pushed before the daemon has even received
		388	+	// the request races this fake's own SendClient/Done selection, which is a
		389	+	// harness artifact unrelated to the fix under test.
		390	+	func TestReplyReachesClientBeforeOrderlyStreamClose(t *testing.T) {
		391	+	e := startEndpoint(t, Config{})
		392	+	const iterations = 200
		393	+	for i := 0; i < iterations; i++ {
		394	+	client, _ := e.pair()
		395	+	core := e.authority.last()
		396	+
		397	+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		398	+	stream, err := client.OpenStream(ctx, openRequest(1))
		399	+	require.NoError(t, err)
		400	+
		401	+	conn := core.logicalConn(1)
		402	+	require.NotNil(t, conn, "iteration %d: the broker core must admit exactly one stream", i)
		403	+
		404	+	require.NoError(t, stream.SendClient(protocol.List{}), "iteration %d", i)
		405	+	select {
		406	+	case <-conn.fromClient:
		407	+	case <-time.After(5 * time.Second):
		408	+	t.Fatalf("iteration %d: request never reached the daemon", i)
		409	+	}
		410	+
		411	+	// The daemon answers and, back to back, ends the stream orderly: exactly
		412	+	// the sequence SendServer(Sessions) then Close produces.
		413	+	conn.toClient <- protocol.Sessions{}
		414	+	require.NoError(t, conn.Close())
		415	+
		416	+	message, err := stream.ReceiveServer()
		417	+	require.NoError(t, err, "iteration %d: the reply must reach the client before EOF", i)
		418	+	require.Equal(t, protocol.Sessions{}, message, "iteration %d", i)
		419	+
		420	+	// The stream settles orderly right after: a further receive observes a
		421	+	// clean EOF, never an error, and never blocks.
		422	+	_, err = stream.ReceiveServer()
		423	+	require.ErrorIs(t, err, io.EOF, "iteration %d", i)
		424	+
		425	+	_ = stream.Close()
		426	+	_ = client.Close()
		427	+	cancel()
		428	+	}
		429	+	}
		430	+
371	// TestOpenStreamRefusedByCoreReachesClient proves a core refusal settles only	431		// TestOpenStreamRefusedByCoreReachesClient proves a core refusal settles only
372	// that stream and reaches the caller as a typed failure.	432		// that stream and reaches the caller as a typed failure.
373	func TestOpenStreamRefusedByCoreReachesClient(t *testing.T) {	433		func TestOpenStreamRefusedByCoreReachesClient(t *testing.T) {

M internal/adapters/brokeripc/stream.go +10 -2


@@ -120,12 +120,20 @@ func (st *serverStream) relayServer() {
120	120		}
121	121
122	122		// watchCore publishes the core connection's terminal outcome even when no read
123		-	// is in flight.
	123	+	// is in flight. It settles the stream itself only for a failure: an orderly
	124	+	// core end (Err()==nil) is left to relayServer, whose own core.ReceiveServer()
	125	+	// call observes the same terminal outcome right after forwarding any reply
	126	+	// that was already in flight, and settles the stream from there. Settling here
	127	+	// on every Done would race relayServer's in-flight wire.SendServer and could
	128	+	// close the carriage while a received reply is still being written out to the
	129	+	// client, discarding it.
124	130		func (st *serverStream) watchCore() {
125	131		defer st.session.wg.Done()
126	132		select {
127	133		case <-st.core.Done():
128		-	st.fail(coreFailure(st.core, nil))
	134	+	if err := st.core.Err(); err != nil {
	135	+	st.fail(coreFailure(st.core, nil))
	136	+	}
129	137		case <-st.done:
130	138		}
131	139		}

 }
 // watchCore publishes the core connection's terminal outcome even when no read
-// is in flight.
+// is in flight. It settles the stream itself only for a failure: an orderly
+// core end (Err()==nil) is left to relayServer, whose own core.ReceiveServer()
+// call observes the same terminal outcome right after forwarding any reply
+// that was already in flight, and settles the stream from there. Settling here
+// on every Done would race relayServer's in-flight wire.SendServer and could
+// close the carriage while a received reply is still being written out to the
+// client, discarding it.
 func (st *serverStream) watchCore() {
 	defer st.session.wg.Done()
 	select {
 	case <-st.core.Done():
-		st.fail(coreFailure(st.core, nil))
+		if err := st.core.Err(); err != nil {
+			st.fail(coreFailure(st.core, nil))
+		}
 	case <-st.done:
 	}
 }