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, DefaultStreamInboundChunks, cfg.StreamInboundChunks)
151 require.Equal(t, uint64(DefaultStreamInboundBytes), cfg.StreamInboundBytes)
152 require.Equal(t, brokerwire.MaxPendingOperations, cfg.MaxPendingOperations)
153
154 require.ErrorIs(t, (Config{StreamInboundBytes: 1}).withDefaults().validate(), ErrConfig)
155 require.ErrorIs(t, (Config{MaxPendingOperations: brokerwire.MaxPendingOperations + 1}).withDefaults().validate(), ErrConfig)
156 // A negative explicit bound is replaced by its default rather than refused.
157 require.NoError(t, (Config{MaxClients: -1, StreamInboundChunks: -1}).withDefaults().validate())
158}
159
160// TestStreamPipeBoundsAndClose proves the per-stream byte channel is bounded,
161// reports backpressure instead of growing, and releases a parked reader on
162// close.
163func TestStreamPipeBoundsAndClose(t *testing.T) {
164 pipe := newStreamPipe(2, 4096, func([]byte) error { return nil })
165 require.NoError(t, pipe.deliver([]byte("one")))
166 require.NoError(t, pipe.deliver([]byte("two")))
167 require.ErrorIs(t, pipe.deliver([]byte("three")), ErrStreamBackpressure)
168
169 buf := make([]byte, 3)
170 n, err := pipe.Read(buf)
171 require.NoError(t, err)
172 require.Equal(t, "one", string(buf[:n]))
173 require.NoError(t, pipe.deliver([]byte("three")))
174
175 // An orderly Close keeps what is already queued ("two" and "three"): Read
176 // drains it before observing io.EOF, so a chunk delivered just ahead of an
177 // orderly close is never discarded underneath its reader (see
178 // TestStreamPipeCloseWithPreservesQueuedDataOnOrderlyClose for the focused
179 // coverage of this behavior).
180 require.NoError(t, pipe.Close())
181 require.ErrorIs(t, pipe.deliver([]byte("four")), ErrStreamGone)
182 drained := make([]byte, 8)
183 n, err = pipe.Read(drained)
184 require.NoError(t, err)
185 require.Equal(t, "two", string(drained[:n]))
186 n, err = pipe.Read(drained)
187 require.NoError(t, err)
188 require.Equal(t, "three", string(drained[:n]))
189 _, err = pipe.Read(drained)
190 require.ErrorIs(t, err, io.EOF)
191
192 // A reader parked on an empty pipe is released by closeWith with its cause.
193 parked := newStreamPipe(1, 4096, func([]byte) error { return nil })
194 done := make(chan error, 1)
195 go func() {
196 _, err := parked.Read(buf)
197 done <- err
198 }()
199 time.Sleep(20 * time.Millisecond)
200 parked.closeWith(ErrStreamBackpressure)
201 select {
202 case err := <-done:
203 require.ErrorIs(t, err, ErrStreamBackpressure)
204 case <-time.After(5 * time.Second):
205 t.Fatal("closeWith must release a parked reader")
206 }
207}
208
209// TestStreamPipeOutboundError proves a failing outbound hook surfaces instead of
210// silently dropping bytes.
211func TestStreamPipeOutboundError(t *testing.T) {
212 pipe := newStreamPipe(1, 4096, func([]byte) error { return ErrStreamBackpressure })
213 _, err := pipe.Write([]byte("payload"))
214 require.ErrorIs(t, err, ErrStreamBackpressure)
215 require.NoError(t, pipe.Close())
216 _, err = pipe.Write([]byte("payload"))
217 require.ErrorIs(t, err, ErrStreamGone)
218}
219
220// TestErrorDetailMapping proves the wire detail and the local typed failure stay
221// mutually consistent, including admission and context outcomes.
222func TestErrorDetailMapping(t *testing.T) {
223 cases := []struct {
224 name string
225 err error
226 code ports.BrokerErrorCode
227 admission uint32
228 }{
229 {"typed", ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "no"}, ports.BrokerErrorIncompatible, 0},
230 {"no daemon", ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Text: "remote daemon is absent"}, ports.BrokerErrorNoDaemon, 0},
231 {"limit", ports.BrokerAdmissionLimit, ports.BrokerErrorUnavailable, 1},
232 {"closed", ports.BrokerAdmissionClosed, ports.BrokerErrorUnavailable, 2},
233 {"stale", ports.BrokerAdmissionStale, ports.BrokerErrorUnavailable, 3},
234 {"invalid", ports.BrokerAdmissionInvalid, ports.BrokerErrorUnavailable, 4},
235 {"deadline", context.DeadlineExceeded, ports.BrokerErrorTimeout, 0},
236 {"cancelled", context.Canceled, ports.BrokerErrorCancelled, 0},
237 // An indeterminate store outcome is classified before the context and
238 // transport rules: a failure after the durable commit point must never be
239 // presented as a retryable timeout, however its cause would classify
240 // alone. Both the value and the pointer form are legal errors because the
241 // ports type carries value-receiver methods.
242 {"outcome unknown value", ports.BrokerStoreOutcomeUnknownError{Err: errors.New("power loss")}, ports.BrokerErrorOutcomeUnknown, 0},
243 {"outcome unknown wrapping deadline", ports.BrokerStoreOutcomeUnknownError{Err: context.DeadlineExceeded}, ports.BrokerErrorOutcomeUnknown, 0},
244 {"outcome unknown pointer", &ports.BrokerStoreOutcomeUnknownError{Err: context.DeadlineExceeded}, ports.BrokerErrorOutcomeUnknown, 0},
245 {"unknown", errors.New("boom"), ports.BrokerErrorUnavailable, 0},
246 }
247 for _, tc := range cases {
248 t.Run(tc.name, func(t *testing.T) {
249 detail := errorDetail(tc.err)
250 require.Equal(t, tc.code, detail.Code)
251 require.Equal(t, tc.admission, detail.AdmissionCode)
252 // The wire encoder validates the detail exactly as a peer would.
253 _, err := brokerwire.EncodeServer(brokerwire.StreamClosed{
254 Epoch: 1, Connection: ports.BrokerConnectionID{1}, Stream: 1,
255 Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
256 }, brokerwire.MaxBrokerEnvelopeBytes, brokerwire.MaxStreamChunkBytes)
257 require.NoError(t, err, "a produced detail must always be wire-valid")
258 })
259 }
260 require.Equal(t, brokerwire.ErrorDetail{}, errorDetail(nil))
261 require.NoError(t, failureFromDetail(brokerwire.ErrorDetail{}), "absent detail is an orderly outcome")
262
263 scoped := failureFromDetail(errorDetail(ports.BrokerAdmissionLimit))
264 require.ErrorIs(t, scoped, ports.BrokerAdmissionLimit)
265}
266
267// TestErrorDetailKeepsStreamLossCause proves a physical stream loss reaches
268// the client with its failure kind and a bounded, display-safe cause in the
269// display text instead of a bare attachment_lost.
270func TestErrorDetailKeepsStreamLossCause(t *testing.T) {
271 lost := func(err error) ports.BrokerStreamLost {
272 return ports.BrokerStreamLost{Connection: ports.BrokerConnectionID{1}, Stream: 1, Epoch: 1, Cause: domain.RemoteFailureTimeout, Err: err}
273 }
274 cases := []struct {
275 name string
276 err error
277 want string
278 }{
279 {"with cause", lost(errors.New("quic: idle timeout")), "timeout: quic: idle timeout"},
280 {"without cause", lost(nil), "timeout"},
281 {"wrapped", fmt.Errorf("attach: %w", lost(errors.New("reset"))), "timeout: reset"},
282 {"control characters removed", lost(errors.New("bad\x1b[31m\nline")), "timeout: bad[31mline"},
283 }
284 for _, tc := range cases {
285 t.Run(tc.name, func(t *testing.T) {
286 detail := errorDetail(tc.err)
287 require.Equal(t, ports.BrokerErrorAttachmentLost, detail.Code)
288 require.Equal(t, tc.want, detail.Text)
289 _, err := brokerwire.EncodeServer(brokerwire.StreamClosed{
290 Epoch: 1, Connection: ports.BrokerConnectionID{1}, Stream: 1, Error: detail, HasError: true,
291 }, brokerwire.MaxBrokerEnvelopeBytes, brokerwire.MaxStreamChunkBytes)
292 require.NoError(t, err)
293
294 var typed ports.BrokerError
295 require.ErrorAs(t, failureFromDetail(detail), &typed)
296 require.Equal(t, tc.want, typed.Text)
297 })
298 }
299}
300
301// TestAdmissionErrorMapping proves connection-tracker refusals map onto the
302// closed ports admission taxonomy.
303func TestAdmissionErrorMapping(t *testing.T) {
304 require.NoError(t, admissionError(nil))
305 require.ErrorIs(t, admissionError(brokerwire.ErrTooManyStreams), ports.BrokerAdmissionLimit)
306 require.ErrorIs(t, admissionError(brokerwire.ErrTooManyPendingOperations), ports.BrokerAdmissionLimit)
307 require.ErrorIs(t, admissionError(brokerwire.ErrConnectionClosed), ports.BrokerAdmissionClosed)
308 require.ErrorIs(t, admissionError(brokerwire.ErrStreamIDReused), ports.BrokerAdmissionStale)
309 require.ErrorIs(t, admissionError(brokerwire.ErrFutureStream), ports.BrokerAdmissionInvalid)
310}
311
312// TestScopeRequestFillsIdentity proves an absent epoch/connection identity is
313// filled from the connection, a foreign one is refused as stale, and a request
314// without a stream identity is refused as invalid rather than allocated here.
315func TestScopeRequestFillsIdentity(t *testing.T) {
316 c := &client{
317 scope: brokerwire.Scope{Epoch: 4, Connection: ports.BrokerConnectionID{4}},
318 ceilings: brokerwire.DefaultCeilings(),
319 pending: map[ports.BrokerOperationID]chan operationResult{},
320 streams: map[ports.BrokerStreamID]*clientStream{},
321 done: make(chan struct{}),
322 cfg: Config{}.withDefaults(),
323 transport: nil,
324 }
325 scoped, err := c.scopeRequest(openRequest(1))
326 require.NoError(t, err)
327 require.Equal(t, ports.BrokerEpoch(4), scoped.Epoch)
328 require.Equal(t, ports.BrokerConnectionID{4}, scoped.Connection)
329 require.Equal(t, ports.BrokerStreamID(1), scoped.Stream)
330
331 _, err = c.scopeRequest(ports.BrokerOpenStreamRequest{
332 Purpose: ports.BrokerStreamControl, Local: true, Stream: 1, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
333 Connection: ports.BrokerConnectionID{0x11},
334 })
335 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
336
337 _, err = c.scopeRequest(ports.BrokerOpenStreamRequest{
338 Purpose: ports.BrokerStreamControl, Local: true, Stream: 1, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded, Epoch: 99,
339 })
340 require.ErrorIs(t, err, ErrScopeMismatch)
341
342 scoped, err = c.scopeRequest(openRequest(5))
343 require.NoError(t, err)
344 require.Equal(t, ports.BrokerStreamID(5), scoped.Stream)
345
346 // The adapter never invents a stream identity: an incomplete request is
347 // refused, and the calling service is the only allocator.
348 _, err = c.scopeRequest(ports.BrokerOpenStreamRequest{
349 Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
350 })
351 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
352}
353
354// TestSocketPaths proves the endpoint name is distinct from the daemon socket
355// and lives in the per-user runtime directory.
356func TestSocketPaths(t *testing.T) {
357 require.Equal(t, "/run/x/vev/broker.sock", SocketPath("/run/x/vev"))
358 require.Equal(t, "broker.sock", SocketFileName)
359 require.NotEqual(t, "daemon.sock", SocketFileName)
360 require.Contains(t, DefaultSocketPath(), SocketFileName)
361}
362
363// TestPreambleCarriageIgnoresUnboundedTransport proves a carriage that cannot be
364// bounded is refused rather than read without a bound.
365func TestPreambleCarriageIgnoresUnboundedTransport(t *testing.T) {
366 _, err := recvPreamble(unboundedTransport{})
367 require.ErrorIs(t, err, ErrConfig)
368}
369
370// unboundedTransport implements only wire.Transport.
371type unboundedTransport struct{}
372
373func (unboundedTransport) Send(wire.Envelope) error { return nil }
374func (unboundedTransport) Recv() (wire.Envelope, error) {
375 return wire.Envelope{}, io.EOF
376}
377func (unboundedTransport) Close() error { return nil }
378
379// TestServerSessionRejectsInvalidConstruction proves construction refuses a nil
380// carriage, a nil core, and a zero epoch instead of building a session that can
381// never serve.
382func TestServerSessionRejectsInvalidConstruction(t *testing.T) {
383 core := &fakeCore{id: ports.BrokerConnectionID{1}, hub: newSnapshotHub(ports.BrokerSnapshot{}), streams: map[ports.BrokerStreamID]*fakeLogicalConn{}, done: make(chan struct{})}
384 _, err := newServerSession(0, nil, brokerwire.DefaultCeilings(), core, Config{}, nil, time.Time{})
385 require.ErrorIs(t, err, ErrConfig)
386 _, err = newServerSession(1, nil, brokerwire.DefaultCeilings(), nil, Config{}, nil, time.Time{})
387 require.ErrorIs(t, err, ErrConfig)
388 require.NoError(t, brokerwire.DefaultCeilings().Validate())
389 require.Error(t, brokerwire.Ceilings{}.Validate())
390}
391
392// TestOrderlyDisconnectClassification proves a peer disconnect, a local close,
393// and a peer registration timeout are orderly ends while a protocol violation
394// is not.
395func TestOrderlyDisconnectClassification(t *testing.T) {
396 require.True(t, orderlyDisconnect(nil))
397 require.True(t, orderlyDisconnect(io.EOF))
398 require.True(t, orderlyDisconnect(ErrSessionClosed))
399 require.True(t, orderlyDisconnect(ErrConnectionClosed))
400 require.True(t, orderlyDisconnect(errors.Join(ErrRegistrationTimeout, context.DeadlineExceeded)))
401 require.True(t, orderlyDisconnect(&net.OpError{Op: "read", Net: "unix", Err: syscall.ECONNRESET}))
402 require.True(t, orderlyDisconnect(&net.OpError{Op: "write", Net: "unix", Err: os.NewSyscallError("write", syscall.EPIPE)}))
403 require.True(t, orderlyDisconnect(net.ErrClosed))
404 require.True(t, orderlyDisconnect(fs.ErrClosed))
405 require.True(t, orderlyDisconnect(fmt.Errorf("write: %w", streamframe.ErrClosed)))
406 require.False(t, orderlyDisconnect(ErrMalformedFrame))
407 require.False(t, orderlyDisconnect(ErrProtocol))
408}
409
410// TestStreamBridgeRequiresConfig proves the bridge refuses missing participants
411// rather than building a half-wired stream.
412func TestStreamBridgeRequiresConfig(t *testing.T) {
413 _, err := newServerStream(nil, 1, nil)
414 require.ErrorIs(t, err, ErrConfig)
415 _, err = newClientStream(nil, 1)
416 require.ErrorIs(t, err, ErrConfig)
417}
418
419// TestClientSubscriptionCoalesces proves the local subscription wakes once per
420// publication and closes exactly once.
421func TestClientSubscriptionCoalesces(t *testing.T) {
422 sub := newClientSubscription()
423 sub.notify()
424 sub.notify()
425 <-sub.Changed()
426 select {
427 case <-sub.Changed():
428 t.Fatal("notifications must coalesce")
429 default:
430 }
431 var wg sync.WaitGroup
432 for i := 0; i < 4; i++ {
433 wg.Add(1)
434 go func() { defer wg.Done(); sub.Close() }()
435 }
436 wg.Wait()
437 _, open := <-sub.Changed()
438 require.False(t, open)
439}