1package brokeripc
2
3import (
4 "encoding/binary"
5 "io"
6 "net"
7 "testing"
8 "time"
9
10 "github.com/stretchr/testify/require"
11
12 "github.com/bnema/vev/internal/adapters/brokerwire"
13 "github.com/bnema/vev/internal/adapters/ipc"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol"
16 "github.com/bnema/vev/internal/protocol/wire"
17)
18
19// Session shutdown ordering (GO-007). A bridged stream writes through its own
20// carriage, whose writer is parked inside the outer carriage write while the
21// peer is not reading. Settling a stream joins that writer, so shutdown must
22// close the outer carriage first: otherwise the join waits on the peer and the
23// session never releases its slot or done.
24
25// rawPeer drives one broker peer over a bare net carriage, framing by hand
26// instead of composing a transport. Consuming a length prefix and deliberately
27// leaving the body unread is what parks the server's write on this non-reading
28// peer deterministically.
29type rawPeer struct {
30 conn net.Conn
31 ceilings brokerwire.Ceilings
32}
33
34// send writes one encoded client frame.
35func (p *rawPeer) send(t *testing.T, message brokerwire.ClientMessage) {
36 t.Helper()
37 payload, err := brokerwire.EncodeClient(message, p.ceilings.MaxReceiveEnvelopeBytes, p.ceilings.StreamChunkLimit)
38 require.NoError(t, err)
39 frame := make([]byte, 4+len(payload))
40 binary.BigEndian.PutUint32(frame[:4], uint32(len(payload)))
41 copy(frame[4:], payload)
42 _, err = p.conn.Write(frame)
43 require.NoError(t, err)
44}
45
46// recv reads and decodes one whole server frame.
47func (p *rawPeer) recv(t *testing.T) brokerwire.ServerMessage {
48 t.Helper()
49 var header [4]byte
50 _, err := io.ReadFull(p.conn, header[:])
51 require.NoError(t, err)
52 payload := make([]byte, binary.BigEndian.Uint32(header[:]))
53 _, err = io.ReadFull(p.conn, payload)
54 require.NoError(t, err)
55 message, err := brokerwire.DecodeServer(payload, p.ceilings.MaxReceiveEnvelopeBytes, p.ceilings.StreamChunkLimit)
56 require.NoError(t, err)
57 return message
58}
59
60// recvHeader consumes only the length prefix of the next server frame, leaving
61// its body unread: the server's frame write is then provably still in flight on
62// this peer, so the stream writer that issued it is parked inside it.
63func (p *rawPeer) recvHeader(t *testing.T) uint32 {
64 t.Helper()
65 var header [4]byte
66 _, err := io.ReadFull(p.conn, header[:])
67 require.NoError(t, err)
68 length := binary.BigEndian.Uint32(header[:])
69 require.NotZero(t, length, "the parked frame must carry a payload")
70 return length
71}
72
73// TestProtocolViolationWithParkedStreamWriterReleasesSession proves a protocol
74// violation settles a session whose bridged stream writer is parked on a
75// non-reading peer: the violation alone closes done, closes the admitted core
76// service, and releases the listener's client slot, with no external session or
77// listener Close.
78func TestProtocolViolationWithParkedStreamWriterReleasesSession(t *testing.T) {
79 const epoch = ports.BrokerEpoch(0x55)
80 serverConn, peerConn := net.Pipe()
81 t.Cleanup(func() { _ = peerConn.Close() })
82
83 transport, ok := ipc.NewTransport(serverConn).(wire.BoundedTransport)
84 require.True(t, ok, "the IPC carriage must be a bounded transport")
85 core := newTestCore(epoch)
86 released := make(chan struct{}, 1)
87 session, err := newServerSession(epoch, transport, brokerwire.DefaultCeilings(), core, Config{}, func() { released <- struct{}{} }, time.Time{})
88 require.NoError(t, err)
89
90 // The peer's Register already happened; the duplicate Register below is the
91 // violation.
92 require.NoError(t, session.conn.Register())
93 go session.run()
94 peer := &rawPeer{conn: peerConn, ceilings: brokerwire.DefaultCeilings()}
95
96 session.startStream(brokerwire.OpenStream{
97 Epoch: epoch, Connection: session.scope.Connection, Stream: 1,
98 Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(),
99 })
100 opened, isOpened := peer.recv(t).(brokerwire.StreamOpened)
101 require.True(t, isOpened, "the stream must be established before its writer can be parked")
102 require.Equal(t, ports.BrokerStreamID(1), opened.Stream)
103
104 // One daemon message reaches the stream's relay, and the peer consumes only
105 // the length prefix of the frame it produces: the broker's carriage write can
106 // no longer complete, so the stream writer is parked inside it.
107 conn := core.logicalConn(1)
108 require.NotNil(t, conn, "the broker core must have admitted the stream")
109 conn.toClient <- protocol.Pong{}
110 peer.recvHeader(t)
111
112 peer.send(t, brokerwire.Register{})
113
114 select {
115 case <-session.done:
116 case <-time.After(5 * time.Second):
117 t.Fatal("a protocol violation must settle a session with a parked stream writer")
118 }
119 select {
120 case <-released:
121 case <-time.After(5 * time.Second):
122 t.Fatal("a settled session must release its client slot")
123 }
124 require.ErrorIs(t, session.terminalErr(), ErrProtocol)
125 select {
126 case <-core.done:
127 default:
128 t.Fatal("a settled session must close its admitted core service")
129 }
130}