1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "io/fs"
9 "net"
10 "os"
11 "sync"
12 "syscall"
13 "testing"
14 "time"
15
16 "github.com/stretchr/testify/require"
17
18 "github.com/bnema/vev/internal/adapters/brokerwire"
19 "github.com/bnema/vev/internal/adapters/streamframe"
20 "github.com/bnema/vev/internal/domain"
21 "github.com/bnema/vev/internal/ports"
22 "github.com/bnema/vev/internal/protocol/catalogue"
23 "github.com/bnema/vev/internal/protocol/wire"
24)
25
26// TestSnapshotPartsLayoutRoundTrips proves the emitter produces exactly the
27// multipart layout the P3.1 assembler commits, including tombstones and
28// session-less hosts.
29func TestSnapshotPartsLayoutRoundTrips(t *testing.T) {
30 snapshot := testSnapshot(9, 12)
31 snapshot.Daemons = append(snapshot.Daemons, testDaemonAt(1))
32 snapshot.Removed = []ports.BrokerHostTombstone{{
33 Endpoint: "retired@old:22",
34 Registration: testDaemonAt(0).Registration,
35 RetiredRevision: 3,
36 }}
37 snapshot.Removed[0].Registration.Endpoint = "retired@old:22"
38 snapshot.Removed[0].Endpoint = "retired@old:22"
39 require.NoError(t, snapshot.Validate())
40
41 parts, err := snapshotParts(snapshot, 9, ports.BrokerConnectionID{9, 9}, 3)
42 require.NoError(t, err)
43 require.Len(t, parts, 1+2+2+1+1)
44
45 begin, ok := parts[0].Part.(brokerwire.SnapshotBegin)
46 require.True(t, ok)
47 require.Equal(t, uint32(2), begin.HostCount)
48 require.Equal(t, uint32(2), begin.SessionCount)
49 require.Equal(t, uint32(1), begin.TombstoneCount)
50 _, ok = parts[len(parts)-1].Part.(brokerwire.SnapshotEnd)
51 require.True(t, ok)
52
53 assembler := brokerwire.NewSnapshotAssembler(9, ports.BrokerConnectionID{9, 9}, brokerwire.WithGeneration(3))
54 var committed ports.BrokerSnapshot
55 found := false
56 for index, part := range parts {
57 require.Equal(t, uint32(index), part.Index)
58 require.Equal(t, brokerwire.SubscriptionGeneration(3), part.Generation)
59 done, assembled, err := assembler.Add(part)
60 require.NoError(t, err, "part %d", index)
61 if done {
62 committed = assembled
63 found = true
64 }
65 }
66 require.True(t, found, "the transfer must commit")
67 require.Equal(t, snapshot.Revision, committed.Revision)
68 require.Len(t, committed.Daemons, 2)
69 require.Len(t, committed.Daemons[0].Sessions, 2)
70 for _, host := range committed.Daemons {
71 for _, session := range host.Sessions {
72 require.Equal(t, []catalogue.RemoteCatalogTab{}, session.Tabs)
73 }
74 }
75 require.Len(t, committed.Removed, 1)
76 require.Equal(t, "retired@old:22", committed.Removed[0].Endpoint)
77}
78
79// TestSnapshotPartsLocalDaemonLayoutRoundTrips proves the emitter and the
80// assembler agree on a local-first transfer: Begin advertises local_present,
81// the local daemon is index 0 with its session bound locally, and the committed
82// snapshot reproduces the local-first publication.
83func TestSnapshotPartsLocalDaemonLayoutRoundTrips(t *testing.T) {
84 snapshot := testLocalSnapshot(4, 2)
85 require.NoError(t, snapshot.Validate())
86
87 parts, err := snapshotParts(snapshot, 4, ports.BrokerConnectionID{4, 4}, 2)
88 require.NoError(t, err)
89 require.Len(t, parts, 1+2+2+1)
90
91 begin, ok := parts[0].Part.(brokerwire.SnapshotBegin)
92 require.True(t, ok)
93 require.True(t, begin.LocalPresent)
94 require.Equal(t, uint32(2), begin.HostCount)
95 require.Equal(t, uint32(2), begin.SessionCount)
96
97 assembler := brokerwire.NewSnapshotAssembler(4, ports.BrokerConnectionID{4, 4}, brokerwire.WithGeneration(2))
98 var committed ports.BrokerSnapshot
99 found := false
100 for index, part := range parts {
101 require.Equal(t, uint32(index), part.Index)
102 done, assembled, err := assembler.Add(part)
103 require.NoError(t, err, "part %d", index)
104 if done {
105 committed = assembled
106 found = true
107 }
108 }
109 require.True(t, found, "the transfer must commit")
110 require.Equal(t, snapshot.Revision, committed.Revision)
111 require.Len(t, committed.Daemons, 2)
112 require.True(t, committed.Daemons[0].Local)
113 require.Empty(t, committed.Daemons[0].Endpoint)
114 require.Len(t, committed.Daemons[0].Sessions, 1)
115 require.Equal(t, snapshot.Daemons[0].Sessions[0].Name, committed.Daemons[0].Sessions[0].Name)
116 require.Equal(t, "user0@host0:22", committed.Daemons[1].Endpoint)
117 require.Len(t, committed.Daemons[1].Sessions, 1)
118}
119
120// TestSnapshotPartsRefusesOverBoundInventory proves the emitter fails closed on
121// an inventory the peer's assembler would refuse, instead of silently sending a
122// transfer that can never commit.
123func TestSnapshotPartsRefusesOverBoundInventory(t *testing.T) {
124 snapshot := testSnapshot(1, 1)
125 over := make([]catalogue.RemoteCatalogSession, ports.BrokerMaxSessionsPerHost+1)
126 snapshot.Daemons[0].Sessions = over
127 _, err := snapshotParts(snapshot, 1, ports.BrokerConnectionID{1}, 1)
128 require.Error(t, err)
129
130 snapshot = testSnapshot(0, 1)
131 _, err = snapshotParts(snapshot, 0, ports.BrokerConnectionID{1}, 1)
132 require.ErrorIs(t, err, ErrScopeMismatch)
133}
134
135// TestSnapshotPartsRequiresRevision proves a snapshot without a publication
136// revision is refused rather than published as revision zero.
137func TestSnapshotPartsRequiresRevision(t *testing.T) {
138 snapshot := testSnapshot(1, 1)
139 snapshot.Revision = 0
140 _, err := snapshotParts(snapshot, 1, ports.BrokerConnectionID{1}, 1)
141 require.Error(t, err)
142}
143
144// TestConfigDefaultsAndValidation proves the zero configuration is valid and an
145// unusable explicit bound is refused.
146func TestConfigDefaultsAndValidation(t *testing.T) {
147 cfg := Config{}.withDefaults()
148 require.NoError(t, cfg.validate())
149 require.Equal(t, DefaultMaxClients, cfg.MaxClients)
150 require.Equal(t, brokerwire.MaxPendingOperations, cfg.MaxPendingOperations)
151
152 require.ErrorIs(t, (Config{MaxPendingOperations: brokerwire.MaxPendingOperations + 1}).withDefaults().validate(), ErrConfig)
153 // A negative explicit bound is replaced by its default rather than refused.
154 require.NoError(t, (Config{MaxClients: -1}).withDefaults().validate())
155}
156
157// TestStreamPipeCloseReleasesParkedReader proves a reader parked on an empty
158// pipe is released by closeWith with its cause.
159func TestStreamPipeCloseReleasesParkedReader(t *testing.T) {
160 parked := newTestPipe()
161 done := make(chan error, 1)
162 go func() {
163 _, err := parked.Read(make([]byte, 3))
164 done <- err
165 }()
166 select {
167 case err := <-done:
168 t.Fatalf("Read returned %v on an empty pipe", err)
169 case <-time.After(20 * time.Millisecond):
170 }
171 parked.closeWith(ErrStreamCredit)
172 select {
173 case err := <-done:
174 require.ErrorIs(t, err, ErrStreamCredit)
175 case <-time.After(5 * time.Second):
176 t.Fatal("closeWith must release a parked reader")
177 }
178 require.ErrorIs(t, parked.deliver([]byte("late")), ErrStreamGone)
179}
180
181// TestStreamPipeOutboundError proves a failing sendFrame hook surfaces instead of
182// silently dropping bytes.
183func TestStreamPipeOutboundError(t *testing.T) {
184 failure := errors.New("outbound failed")
185 pipe := newStreamPipe(int(brokerwire.MaxStreamChunkBytes), func([]byte) error { return failure }, func(uint64) error { return nil })
186 _, err := pipe.Write([]byte("payload"))
187 require.ErrorIs(t, err, failure)
188 require.NoError(t, pipe.Close())
189 _, err = pipe.Write([]byte("payload"))
190 require.ErrorIs(t, err, ErrStreamGone)
191}
192
193// TestErrorDetailMapping proves the wire detail and the local typed failure stay
194// mutually consistent, including admission and context outcomes.
195func TestErrorDetailMapping(t *testing.T) {
196 cases := []struct {
197 name string
198 err error
199 code ports.BrokerErrorCode
200 admission uint32
201 }{
202 {"typed", ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "no"}, ports.BrokerErrorIncompatible, 0},
203 {"no daemon", ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Text: "remote daemon is absent"}, ports.BrokerErrorNoDaemon, 0},
204 {"limit", ports.BrokerAdmissionLimit, ports.BrokerErrorUnavailable, 1},
205 {"closed", ports.BrokerAdmissionClosed, ports.BrokerErrorUnavailable, 2},
206 {"stale", ports.BrokerAdmissionStale, ports.BrokerErrorUnavailable, 3},
207 {"invalid", ports.BrokerAdmissionInvalid, ports.BrokerErrorUnavailable, 4},
208 {"deadline", context.DeadlineExceeded, ports.BrokerErrorTimeout, 0},
209 {"cancelled", context.Canceled, ports.BrokerErrorCancelled, 0},
210 // An indeterminate store outcome is classified before the context and
211 // transport rules: a failure after the durable commit point must never be
212 // presented as a retryable timeout, however its cause would classify
213 // alone. Both the value and the pointer form are legal errors because the
214 // ports type carries value-receiver methods.
215 {"outcome unknown value", ports.BrokerStoreOutcomeUnknownError{Err: errors.New("power loss")}, ports.BrokerErrorOutcomeUnknown, 0},
216 {"outcome unknown wrapping deadline", ports.BrokerStoreOutcomeUnknownError{Err: context.DeadlineExceeded}, ports.BrokerErrorOutcomeUnknown, 0},
217 {"outcome unknown pointer", &ports.BrokerStoreOutcomeUnknownError{Err: context.DeadlineExceeded}, ports.BrokerErrorOutcomeUnknown, 0},
218 {"unknown", errors.New("boom"), ports.BrokerErrorUnavailable, 0},
219 }
220 for _, tc := range cases {
221 t.Run(tc.name, func(t *testing.T) {
222 detail := errorDetail(tc.err)
223 require.Equal(t, tc.code, detail.Code)
224 require.Equal(t, tc.admission, detail.AdmissionCode)
225 // The wire encoder validates the detail exactly as a peer would.
226 _, err := brokerwire.EncodeServer(brokerwire.StreamClosed{
227 Epoch: 1, Connection: ports.BrokerConnectionID{1}, Stream: 1,
228 Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
229 }, brokerwire.MaxBrokerEnvelopeBytes, brokerwire.MaxStreamChunkBytes)
230 require.NoError(t, err, "a produced detail must always be wire-valid")
231 })
232 }
233 require.Equal(t, brokerwire.ErrorDetail{}, errorDetail(nil))
234 require.NoError(t, failureFromDetail(brokerwire.ErrorDetail{}), "absent detail is an orderly outcome")
235
236 scoped := failureFromDetail(errorDetail(ports.BrokerAdmissionLimit))
237 require.ErrorIs(t, scoped, ports.BrokerAdmissionLimit)
238}
239
240// TestErrorDetailKeepsStreamLossCause proves a physical stream loss reaches
241// the client with its failure kind and a bounded, display-safe cause in the
242// display text instead of a bare attachment_lost.
243func TestErrorDetailKeepsStreamLossCause(t *testing.T) {
244 lost := func(err error) ports.BrokerStreamLost {
245 return ports.BrokerStreamLost{Connection: ports.BrokerConnectionID{1}, Stream: 1, Epoch: 1, Cause: domain.RemoteFailureTimeout, Err: err}
246 }
247 cases := []struct {
248 name string
249 err error
250 want string
251 }{
252 {"with cause", lost(errors.New("quic: idle timeout")), "timeout: quic: idle timeout"},
253 {"without cause", lost(nil), "timeout"},
254 {"wrapped", fmt.Errorf("attach: %w", lost(errors.New("reset"))), "timeout: reset"},
255 {"control characters removed", lost(errors.New("bad\x1b[31m\nline")), "timeout: bad[31mline"},
256 }
257 for _, tc := range cases {
258 t.Run(tc.name, func(t *testing.T) {
259 detail := errorDetail(tc.err)
260 require.Equal(t, ports.BrokerErrorAttachmentLost, detail.Code)
261 require.Equal(t, tc.want, detail.Text)
262 _, err := brokerwire.EncodeServer(brokerwire.StreamClosed{
263 Epoch: 1, Connection: ports.BrokerConnectionID{1}, Stream: 1, Error: detail, HasError: true,
264 }, brokerwire.MaxBrokerEnvelopeBytes, brokerwire.MaxStreamChunkBytes)
265 require.NoError(t, err)
266
267 var typed ports.BrokerError
268 require.ErrorAs(t, failureFromDetail(detail), &typed)
269 require.Equal(t, tc.want, typed.Text)
270 })
271 }
272}
273
274// TestAdmissionErrorMapping proves connection-tracker refusals map onto the
275// closed ports admission taxonomy.
276func TestAdmissionErrorMapping(t *testing.T) {
277 require.NoError(t, admissionError(nil))
278 require.ErrorIs(t, admissionError(brokerwire.ErrTooManyStreams), ports.BrokerAdmissionLimit)
279 require.ErrorIs(t, admissionError(brokerwire.ErrTooManyPendingOperations), ports.BrokerAdmissionLimit)
280 require.ErrorIs(t, admissionError(brokerwire.ErrConnectionClosed), ports.BrokerAdmissionClosed)
281 require.ErrorIs(t, admissionError(brokerwire.ErrStreamIDReused), ports.BrokerAdmissionStale)
282 require.ErrorIs(t, admissionError(brokerwire.ErrFutureStream), ports.BrokerAdmissionInvalid)
283}
284
285// TestScopeRequestFillsIdentity proves an absent epoch/connection identity is
286// filled from the connection, a foreign one is refused as stale, and a request
287// without a stream identity is refused as invalid rather than allocated here.
288func TestScopeRequestFillsIdentity(t *testing.T) {
289 c := &client{
290 scope: brokerwire.Scope{Epoch: 4, Connection: ports.BrokerConnectionID{4}},
291 ceilings: brokerwire.DefaultCeilings(),
292 pending: map[ports.BrokerOperationID]chan operationResult{},
293 streams: map[ports.BrokerStreamID]*clientStream{},
294 done: make(chan struct{}),
295 cfg: Config{}.withDefaults(),
296 transport: nil,
297 }
298 scoped, err := c.scopeRequest(openRequest(1))
299 require.NoError(t, err)
300 require.Equal(t, ports.BrokerEpoch(4), scoped.Epoch)
301 require.Equal(t, ports.BrokerConnectionID{4}, scoped.Connection)
302 require.Equal(t, ports.BrokerStreamID(1), scoped.Stream)
303
304 _, err = c.scopeRequest(ports.BrokerOpenStreamRequest{
305 Purpose: ports.BrokerStreamControl, Local: true, Stream: 1, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
306 Connection: ports.BrokerConnectionID{0x11},
307 })
308 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
309
310 _, err = c.scopeRequest(ports.BrokerOpenStreamRequest{
311 Purpose: ports.BrokerStreamControl, Local: true, Stream: 1, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded, Epoch: 99,
312 })
313 require.ErrorIs(t, err, ErrScopeMismatch)
314
315 scoped, err = c.scopeRequest(openRequest(5))
316 require.NoError(t, err)
317 require.Equal(t, ports.BrokerStreamID(5), scoped.Stream)
318
319 // The adapter never invents a stream identity: an incomplete request is
320 // refused, and the calling service is the only allocator.
321 _, err = c.scopeRequest(ports.BrokerOpenStreamRequest{
322 Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
323 })
324 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
325}
326
327// TestSocketPaths proves the endpoint name is distinct from the daemon socket
328// and lives in the per-user runtime directory.
329func TestSocketPaths(t *testing.T) {
330 require.Equal(t, "/run/x/vev/broker.sock", SocketPath("/run/x/vev"))
331 require.Equal(t, "broker.sock", SocketFileName)
332 require.NotEqual(t, "daemon.sock", SocketFileName)
333 require.Contains(t, DefaultSocketPath(), SocketFileName)
334}
335
336// TestPreambleCarriageIgnoresUnboundedTransport proves a carriage that cannot be
337// bounded is refused rather than read without a bound.
338func TestPreambleCarriageIgnoresUnboundedTransport(t *testing.T) {
339 _, err := recvPreamble(unboundedTransport{})
340 require.ErrorIs(t, err, ErrConfig)
341}
342
343// unboundedTransport implements only wire.Transport.
344type unboundedTransport struct{}
345
346func (unboundedTransport) Send(wire.Envelope) error { return nil }
347func (unboundedTransport) Recv() (wire.Envelope, error) {
348 return wire.Envelope{}, io.EOF
349}
350func (unboundedTransport) Close() error { return nil }
351
352// TestServerSessionRejectsInvalidConstruction proves construction refuses a nil
353// carriage, a nil core, and a zero epoch instead of building a session that can
354// never serve.
355func TestServerSessionRejectsInvalidConstruction(t *testing.T) {
356 core := &fakeCore{id: ports.BrokerConnectionID{1}, hub: newSnapshotHub(ports.BrokerSnapshot{}), streams: map[ports.BrokerStreamID]*fakeLogicalConn{}, done: make(chan struct{})}
357 _, err := newServerSession(0, nil, brokerwire.DefaultCeilings(), core, Config{}, nil, time.Time{})
358 require.ErrorIs(t, err, ErrConfig)
359 _, err = newServerSession(1, nil, brokerwire.DefaultCeilings(), nil, Config{}, nil, time.Time{})
360 require.ErrorIs(t, err, ErrConfig)
361 require.NoError(t, brokerwire.DefaultCeilings().Validate())
362 require.Error(t, brokerwire.Ceilings{}.Validate())
363}
364
365// TestOrderlyDisconnectClassification proves a peer disconnect, a local close,
366// and a peer registration timeout are orderly ends while a protocol violation
367// is not.
368func TestOrderlyDisconnectClassification(t *testing.T) {
369 require.True(t, orderlyDisconnect(nil))
370 require.True(t, orderlyDisconnect(io.EOF))
371 require.True(t, orderlyDisconnect(ErrSessionClosed))
372 require.True(t, orderlyDisconnect(ErrConnectionClosed))
373 require.True(t, orderlyDisconnect(errors.Join(ErrRegistrationTimeout, context.DeadlineExceeded)))
374 require.True(t, orderlyDisconnect(&net.OpError{Op: "read", Net: "unix", Err: syscall.ECONNRESET}))
375 require.True(t, orderlyDisconnect(&net.OpError{Op: "write", Net: "unix", Err: os.NewSyscallError("write", syscall.EPIPE)}))
376 require.True(t, orderlyDisconnect(net.ErrClosed))
377 require.True(t, orderlyDisconnect(fs.ErrClosed))
378 require.True(t, orderlyDisconnect(fmt.Errorf("write: %w", streamframe.ErrClosed)))
379 require.False(t, orderlyDisconnect(ErrMalformedFrame))
380 require.False(t, orderlyDisconnect(ErrProtocol))
381}
382
383// TestStreamBridgeRequiresConfig proves the bridge refuses missing participants
384// rather than building a half-wired stream.
385func TestStreamBridgeRequiresConfig(t *testing.T) {
386 _, err := newServerStream(nil, 1, nil)
387 require.ErrorIs(t, err, ErrConfig)
388 _, err = newClientStream(nil, 1)
389 require.ErrorIs(t, err, ErrConfig)
390}
391
392// TestClientSubscriptionCoalesces proves the local subscription wakes once per
393// publication and closes exactly once.
394func TestClientSubscriptionCoalesces(t *testing.T) {
395 sub := newClientSubscription()
396 sub.notify()
397 sub.notify()
398 <-sub.Changed()
399 select {
400 case <-sub.Changed():
401 t.Fatal("notifications must coalesce")
402 default:
403 }
404 var wg sync.WaitGroup
405 for i := 0; i < 4; i++ {
406 wg.Add(1)
407 go func() { defer wg.Done(); sub.Close() }()
408 }
409 wg.Wait()
410 _, open := <-sub.Changed()
411 require.False(t, open)
412}