1package brokeripc
2
3import (
4 "net"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/require"
9
10 "github.com/bnema/vev/internal/adapters/brokerwire"
11 "github.com/bnema/vev/internal/adapters/ipc"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol/wire"
14)
15
16// newRunningSession starts one registered server session over a real net.Pipe
17// carriage so a test can drive its terminal outcome from the peer side.
18func newRunningSession(t *testing.T, epoch ports.BrokerEpoch) (*serverSession, net.Conn) {
19 t.Helper()
20 serverConn, peerConn := net.Pipe()
21 t.Cleanup(func() { _ = peerConn.Close() })
22 transport, ok := ipc.NewTransport(serverConn).(wire.BoundedTransport)
23 require.True(t, ok, "the IPC carriage must be a bounded transport")
24 session, err := newServerSession(epoch, transport, brokerwire.DefaultCeilings(), newTestCore(epoch), Config{}, func() {}, time.Time{})
25 require.NoError(t, err)
26 require.NoError(t, session.conn.Register())
27 go session.run()
28 return session, peerConn
29}
30
31// awaitSessionDone waits for one session to reach its terminal state.
32func awaitSessionDone(t *testing.T, session *serverSession) {
33 t.Helper()
34 select {
35 case <-session.Done():
36 case <-time.After(5 * time.Second):
37 t.Fatal("session did not become terminal")
38 }
39}
40
41// TestSessionErrReportsOrderlyEndAsNil pins the server session Err contract: a
42// peer disconnect and a local Close are orderly ends and report nil, while a
43// protocol failure reports the typed cause that settled the session. Done is
44// closed and Err stable in every case.
45func TestSessionErrReportsOrderlyEndAsNil(t *testing.T) {
46 t.Run("peer EOF is orderly", func(t *testing.T) {
47 session, peer := newRunningSession(t, 0x61)
48 require.NoError(t, peer.Close())
49 awaitSessionDone(t, session)
50
51 // The underlying cause is the peer's EOF, but Err presents it as an
52 // orderly end so a watcher never mistakes a departure for a failure.
53 require.Error(t, session.terminalErr(), "the peer EOF remains the recorded internal cause")
54 require.True(t, orderlyDisconnect(session.terminalErr()))
55 require.NoError(t, session.Err())
56 require.NoError(t, session.Err(), "Err is stable after Done")
57 require.NoError(t, session.Close())
58 })
59
60 t.Run("local Close is orderly", func(t *testing.T) {
61 session, _ := newRunningSession(t, 0x62)
62 require.NoError(t, session.Close())
63 awaitSessionDone(t, session)
64 require.NoError(t, session.Err())
65 })
66
67 t.Run("protocol failure is typed", func(t *testing.T) {
68 session, peer := newRunningSession(t, 0x63)
69 raw := &rawPeer{conn: peer, ceilings: brokerwire.DefaultCeilings()}
70 // A second Register is a protocol-order violation and settles the
71 // connection.
72 raw.send(t, brokerwire.Register{})
73 awaitSessionDone(t, session)
74
75 require.Error(t, session.Err())
76 require.ErrorIs(t, session.Err(), ErrProtocol)
77 require.Error(t, session.Close(), "a protocol failure is not an orderly end")
78 })
79}