1package brokeripc
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "io"
8 "strings"
9 "testing"
10 "time"
11
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/vev/internal/adapters/brokerwire"
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"
17 "github.com/bnema/vev/internal/protocol"
18)
19
20// TestResubscribePublishesCurrentSnapshotPerGeneration races repeated
21// resubscriptions: a superseded publisher must never consume the wake meant for
22// its successor, so the generation that ends a burst publishes the current
23// snapshot without any further core wake, and a resync republishes it. The
24// whole burst runs with no yield between generations, so a shared wake channel
25// would coalesce a later generation's demand into a superseded publisher.
26func TestResubscribePublishesCurrentSnapshotPerGeneration(t *testing.T) {
27 const (
28 epoch = ports.BrokerEpoch(0x51)
29 generations = 32
30 bursts = 50
31 )
32 for burst := 0; burst < bursts; burst++ {
33 transport := newRecordingTransport()
34 core := newTestCore(epoch)
35 session, err := newServerSession(epoch, transport, brokerwire.DefaultCeilings(), core, Config{}, func() {}, time.Time{})
36 require.NoError(t, err)
37 require.NoError(t, session.conn.Register())
38
39 for index := 1; index <= generations; index++ {
40 generation := brokerwire.SubscriptionGeneration(index)
41 core.hub.publish(testSnapshot(epoch, ports.BrokerRevision(index)))
42 require.NoError(t, session.conn.Subscribe(generation))
43 require.NoError(t, session.retargetPublisher(generation, false))
44 }
45
46 // Each publisher owns its wake, so a later generation can never be left
47 // waiting on a demand a superseded publisher consumed.
48 session.subMu.Lock()
49 active := session.pub
50 session.subMu.Unlock()
51 require.NotNil(t, active)
52 require.Equal(t, brokerwire.SubscriptionGeneration(generations), active.generation)
53 require.True(t,
54 transport.waitForTransfer(brokerwire.SubscriptionGeneration(generations), ports.BrokerRevision(generations), 5*time.Second),
55 "the final generation must publish the current snapshot for itself")
56
57 // A resync republishes the current snapshot for the active generation.
58 before := transport.transferCount(brokerwire.SubscriptionGeneration(generations), ports.BrokerRevision(generations))
59 require.NoError(t, session.conn.Resync(brokerwire.SubscriptionGeneration(generations)))
60 session.wakePublisher()
61 require.True(t,
62 transport.waitForTransferCount(brokerwire.SubscriptionGeneration(generations), ports.BrokerRevision(generations), before+1, 5*time.Second),
63 "a resync must republish the current snapshot")
64
65 session.shutdown()
66 }
67}
68
69// TestSupersededPublisherOwnsItsWake proves each subscription generation owns
70// its own coalescing wake, so a superseded publisher cannot consume the demand
71// intended for its successor.
72func TestSupersededPublisherOwnsItsWake(t *testing.T) {
73 const epoch = ports.BrokerEpoch(0x52)
74 transport := newRecordingTransport()
75 core := newTestCore(epoch)
76 session, err := newServerSession(epoch, transport, brokerwire.DefaultCeilings(), core, Config{}, func() {}, time.Time{})
77 require.NoError(t, err)
78 require.NoError(t, session.conn.Register())
79 t.Cleanup(func() { session.shutdown() })
80
81 core.hub.publish(testSnapshot(epoch, 1))
82 require.NoError(t, session.conn.Subscribe(1))
83 require.NoError(t, session.retargetPublisher(1, false))
84 session.subMu.Lock()
85 first := session.pub
86 session.subMu.Unlock()
87
88 core.hub.publish(testSnapshot(epoch, 2))
89 require.NoError(t, session.conn.Subscribe(2))
90 require.NoError(t, session.retargetPublisher(2, false))
91 session.subMu.Lock()
92 second := session.pub
93 session.subMu.Unlock()
94
95 require.NotNil(t, first)
96 require.NotNil(t, second)
97 require.NotSame(t, first, second)
98 require.NotEqual(t, first.wake, second.wake, "each generation must own its wake channel")
99}
100
101// TestSessionCloseStreamFillsConnectionScope proves the session CloseStream
102// delegates with this connection's scope exactly like OpenStream: an absent
103// identity is filled from the session and a foreign one is refused.
104func TestSessionCloseStreamFillsConnectionScope(t *testing.T) {
105 const epoch = ports.BrokerEpoch(0x53)
106 transport := newRecordingTransport()
107 core := newTestCore(epoch)
108 session, err := newServerSession(epoch, transport, brokerwire.DefaultCeilings(), core, Config{}, func() {}, time.Time{})
109 require.NoError(t, err)
110 t.Cleanup(func() { session.shutdown() })
111
112 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
113 defer cancel()
114 opened, err := session.OpenEnvelopeStream(ctx, openRequest(1))
115 require.NoError(t, err)
116 require.NotNil(t, opened)
117
118 core.mu.Lock()
119 require.Len(t, core.opens, 1)
120 require.Equal(t, core.id, core.opens[0].Connection, "OpenStream fills the session scope")
121 require.Equal(t, ports.BrokerStreamID(1), core.opens[0].Stream)
122 core.mu.Unlock()
123
124 // A zero identity is filled from the session instead of reaching the core
125 // as a stale request.
126 require.NoError(t, session.CloseStream(ports.BrokerConnectionID{}, 1))
127 core.mu.Lock()
128 require.Equal(t, []ports.BrokerStreamID{1}, core.closedIDs)
129 core.mu.Unlock()
130
131 // A foreign identity is still refused.
132 require.ErrorIs(t, session.CloseStream(ports.BrokerConnectionID{0x99}, 2), ports.BrokerAdmissionStale)
133}
134
135// TestOpenStreamAdmissionRoundTrips proves the attachment-admission contract
136// travels the IPC wire: a create-named request reaches the core with exactly
137// its admission and validated name.
138func TestOpenStreamAdmissionRoundTrips(t *testing.T) {
139 e := startEndpoint(t, Config{})
140 client, _ := e.pair()
141 core := e.authority.last()
142 require.NotNil(t, core)
143
144 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
145 defer cancel()
146 stream, err := client.OpenStream(ctx, attachmentRequest(1, ports.BrokerAdmissionCreateNamed, "work"))
147 require.NoError(t, err)
148 require.NotNil(t, stream)
149
150 core.mu.Lock()
151 defer core.mu.Unlock()
152 require.Len(t, core.opens, 1)
153 require.Equal(t, ports.BrokerAdmissionCreateNamed, core.opens[0].Admission)
154 require.Equal(t, "work", core.opens[0].Name)
155 require.Equal(t, ports.BrokerStreamAttachment, core.opens[0].Purpose)
156 require.True(t, core.opens[0].Local)
157 require.Equal(t, ports.BrokerDaemonStartIfNeeded, core.opens[0].StartMode)
158}
159
160// TestObservationForcesExistingOnly proves the observation authorization
161// travels the IPC wire: a control stream may carry start-if-needed, while an
162// observation must carry existing-only and a spawn-capable observation is
163// refused before any frame travels.
164func TestObservationForcesExistingOnly(t *testing.T) {
165 e := startEndpoint(t, Config{})
166 client, _ := e.pair()
167 core := e.authority.last()
168 require.NotNil(t, core)
169
170 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
171 defer cancel()
172 streamID := nextStreamID(t, client)
173 observation := openRequest(streamID)
174 observation.Purpose = ports.BrokerStreamObservation
175 observation.StartMode = ports.BrokerDaemonExistingOnly
176 stream, err := client.OpenStream(ctx, observation)
177 require.NoError(t, err)
178 require.NoError(t, stream.Close())
179
180 core.mu.Lock()
181 defer core.mu.Unlock()
182 require.Len(t, core.opens, 1, "the existing-only observation reached the core")
183 require.Equal(t, ports.BrokerStreamObservation, core.opens[0].Purpose)
184 require.Equal(t, ports.BrokerDaemonExistingOnly, core.opens[0].StartMode)
185}
186
187// TestRegistrationAssignsScopeAtAccept proves the client adapter adopts exactly
188// the scope the listener assigned at accept.
189func TestRegistrationAssignsScopeAtAccept(t *testing.T) {
190 e := startEndpoint(t, Config{})
191 client, session := e.pair()
192 require.Equal(t, session.ConnectionID(), client.ConnectionID())
193 require.False(t, client.ConnectionID().IsZero())
194}
195
196// TestSubscribePublishesSnapshot proves subscription traffic: a committed
197// publication reaches the client, wakes its subscription, and round-trips host,
198// session, and tombstone state.
199func TestSubscribePublishesSnapshot(t *testing.T) {
200 e := startEndpoint(t, Config{})
201 client, _ := e.pair()
202 sub, err := client.Subscribe()
203 require.NoError(t, err)
204 t.Cleanup(sub.Close)
205
206 snapshot := e.publish(e.epoch, 7)
207 require.Eventually(t, func() bool {
208 return client.Snapshot().Revision == snapshot.Revision
209 }, 5*time.Second, 10*time.Millisecond, "committed publication must reach the client")
210
211 got := client.Snapshot()
212 require.Equal(t, e.epoch, got.Epoch)
213 require.Len(t, got.Daemons, 1)
214 require.Equal(t, snapshot.Daemons[0].Endpoint, got.Daemons[0].Endpoint)
215 require.Len(t, got.Daemons[0].Sessions, 2)
216 require.Equal(t, snapshot.Daemons[0].Sessions[0].Name, got.Daemons[0].Sessions[0].Name)
217
218 select {
219 case <-sub.Changed():
220 case <-time.After(5 * time.Second):
221 t.Fatal("subscription was not woken by a committed publication")
222 }
223}
224
225// TestSubscribeCarriesPassiveReader proves a client configured as a passive
226// reader reaches the core's passive subscription, while a default client
227// subscribes as a watching client that records observation demand.
228func TestSubscribeCarriesPassiveReader(t *testing.T) {
229 tests := []struct {
230 name string
231 passive bool
232 }{
233 {name: "watching client", passive: false},
234 {name: "passive reader", passive: true},
235 }
236 for _, tt := range tests {
237 t.Run(tt.name, func(t *testing.T) {
238 e := startEndpoint(t, Config{})
239 client := e.dialWith(Config{PassiveSubscribe: tt.passive})
240 e.accept()
241 sub, err := client.Subscribe()
242 require.NoError(t, err)
243 t.Cleanup(sub.Close)
244 core := e.authority.last()
245 require.Eventually(t, func() bool { return len(core.subscribeKinds()) == 1 }, 5*time.Second, 10*time.Millisecond)
246 require.Equal(t, []bool{tt.passive}, core.subscribeKinds())
247 })
248 }
249}
250
251// TestSubscribeGenerationAdvancesAndReplaces proves a second Subscribe advances
252// the generation series (refusing reuse) and replaces the previous
253// subscription.
254func TestSubscribeGenerationAdvancesAndReplaces(t *testing.T) {
255 e := startEndpoint(t, Config{})
256 client, _ := e.pair()
257 first, err := client.Subscribe()
258 require.NoError(t, err)
259 second, err := client.Subscribe()
260 require.NoError(t, err)
261 t.Cleanup(second.Close)
262
263 select {
264 case _, open := <-first.Changed():
265 // The previous subscription is closed: a closed channel reads
266 // immediately with open=false.
267 require.False(t, open, "a superseded subscription must be closed")
268 case <-time.After(time.Second):
269 t.Fatal("the superseded subscription must be closed")
270 }
271
272 e.publish(e.epoch, 3)
273 require.Eventually(t, func() bool { return client.Snapshot().Revision == 3 }, 5*time.Second, 10*time.Millisecond)
274}
275
276// subscribeResult is one Subscribe outcome.
277type subscribeResult struct {
278 sub ports.BrokerSubscription
279 err error
280}
281
282// subscribeGenerationOf extracts the generation of one captured Subscribe
283// frame.
284func subscribeGenerationOf(t *testing.T, message brokerwire.ClientMessage) uint64 {
285 t.Helper()
286 subscribe, ok := message.(brokerwire.Subscribe)
287 require.True(t, ok, "the captured attempt must be a Subscribe frame")
288 return subscribe.Generation
289}
290
291// awaitSubscribe waits for one Subscribe call to return.
292func awaitSubscribe(t *testing.T, results <-chan subscribeResult) subscribeResult {
293 t.Helper()
294 select {
295 case got := <-results:
296 return got
297 case <-time.After(5 * time.Second):
298 t.Fatal("Subscribe did not return")
299 return subscribeResult{}
300 }
301}
302
303// TestSubscribeSendFailureRetiresPreviousSubscription proves a Subscribe whose
304// frame never leaves has already consumed the connection's generation series:
305// the superseded subscription is retired instead of being restored, so no stale
306// subscription is presented as the live one and none stays current.
307func TestSubscribeSendFailureRetiresPreviousSubscription(t *testing.T) {
308 c, carriage := newGatedClient(t, Config{})
309
310 first := make(chan subscribeResult, 1)
311 go func() {
312 sub, err := c.Subscribe()
313 first <- subscribeResult{sub, err}
314 }()
315 require.Equal(t, uint64(1), subscribeGenerationOf(t, awaitSend(t, carriage)))
316 carriage.releaseSend(nil)
317 previous := awaitSubscribe(t, first)
318 require.NoError(t, previous.err)
319 require.NotNil(t, previous.sub)
320
321 // The second Subscribe consumes generation 2 on the connection's tracker
322 // before its carriage write fails.
323 second := make(chan subscribeResult, 1)
324 go func() {
325 sub, err := c.Subscribe()
326 second <- subscribeResult{sub, err}
327 }()
328 require.Equal(t, uint64(2), subscribeGenerationOf(t, awaitSend(t, carriage)))
329 carriage.releaseSend(errTestCarriage)
330
331 failed := awaitSubscribe(t, second)
332 require.ErrorIs(t, failed.err, errTestCarriage)
333 require.Nil(t, failed.sub, "a failed Subscribe returns no subscription")
334
335 select {
336 case _, open := <-previous.sub.Changed():
337 require.False(t, open, "the superseded subscription must be closed")
338 case <-time.After(time.Second):
339 t.Fatal("the previous subscription must be closed")
340 }
341
342 c.mu.Lock()
343 current, generation := c.sub, c.generation
344 c.mu.Unlock()
345 require.Nil(t, current, "a failed Subscribe must not leave a current subscription")
346 require.Equal(t, uint64(2), uint64(generation), "the consumed generation is not rolled back")
347}
348
349// TestOpenStreamCarriesTypedTraffic proves a logical stream carries the typed
350// session protocol in both directions through the broker.
351func TestOpenStreamCarriesTypedTraffic(t *testing.T) {
352 e := startEndpoint(t, Config{})
353 client, _ := e.pair()
354 core := e.authority.last()
355
356 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
357 defer cancel()
358 stream, err := client.OpenStream(ctx, openRequest(1))
359 require.NoError(t, err)
360 t.Cleanup(func() { _ = stream.Close() })
361
362 conn := core.logicalConn(1)
363 require.NotNil(t, conn, "the broker core must admit exactly one stream")
364
365 require.NoError(t, stream.SendClient(protocol.Ping{}))
366 select {
367 case message := <-conn.fromClient:
368 require.Equal(t, protocol.Ping{}, message, "the client message must reach the broker core")
369 case <-time.After(5 * time.Second):
370 t.Fatal("client message never reached the broker core")
371 }
372
373 conn.toClient <- protocol.Pong{}
374 done := make(chan protocol.ServerMessage, 1)
375 go func() {
376 message, err := stream.ReceiveServer()
377 if err != nil {
378 done <- nil
379 return
380 }
381 done <- message
382 }()
383 select {
384 case message := <-done:
385 require.Equal(t, protocol.Pong{}, message, "the daemon message must reach the client")
386 case <-time.After(5 * time.Second):
387 t.Fatal("daemon message never reached the client")
388 }
389}
390
391// TestReplyReachesClientBeforeOrderlyStreamClose proves a reply the broker core
392// sends immediately before an orderly stream end (a StreamClosed carrying no
393// error) always reaches the client's ReceiveServer before it observes EOF.
394// This is the exact end-to-end sequence that dropped about 70% of `vev ls`
395// calls: the daemon's handleList receives the request, does
396// SendServer(Sessions) then Close, so the broker forwards protocol.Sessions and
397// immediately sees the core end orderly. Before the fix, watchCore settled the
398// stream (closing the client-side pipe and discarding its queue) racing the
399// client's read of the already-forwarded reply; the client observed io.EOF
400// instead of the reply and reported the broker operation lost or its outcome
401// unknown. It is run repeatedly under -race because the race window is narrow,
402// not guaranteed on a single pass.
403//
404// The request is sent and its receipt is confirmed before the daemon answers,
405// exactly like every real caller (BrokerOperations always calls SendClient
406// before ReceiveServer): a reply pushed before the daemon has even received
407// the request races this fake's own SendClient/Done selection, which is a
408// harness artifact unrelated to the fix under test.
409func TestReplyReachesClientBeforeOrderlyStreamClose(t *testing.T) {
410 e := startEndpoint(t, Config{})
411 const iterations = 200
412 for i := 0; i < iterations; i++ {
413 client, _ := e.pair()
414 core := e.authority.last()
415
416 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
417 stream, err := client.OpenStream(ctx, openRequest(1))
418 require.NoError(t, err)
419
420 conn := core.logicalConn(1)
421 require.NotNil(t, conn, "iteration %d: the broker core must admit exactly one stream", i)
422
423 require.NoError(t, stream.SendClient(protocol.List{}), "iteration %d", i)
424 select {
425 case <-conn.fromClient:
426 case <-time.After(5 * time.Second):
427 t.Fatalf("iteration %d: request never reached the daemon", i)
428 }
429
430 // The daemon answers and, back to back, ends the stream orderly: exactly
431 // the sequence SendServer(Sessions) then Close produces.
432 conn.toClient <- protocol.Sessions{}
433 require.NoError(t, conn.Close())
434
435 message, err := stream.ReceiveServer()
436 require.NoError(t, err, "iteration %d: the reply must reach the client before EOF", i)
437 require.Equal(t, protocol.Sessions{}, message, "iteration %d", i)
438
439 // The stream settles orderly right after: a further receive observes a
440 // clean EOF, never an error, and never blocks.
441 _, err = stream.ReceiveServer()
442 require.ErrorIs(t, err, io.EOF, "iteration %d", i)
443
444 _ = stream.Close()
445 _ = client.Close()
446 cancel()
447 }
448}
449
450// TestOpenStreamRefusedByCoreReachesClient proves a core refusal settles only
451// that stream and reaches the caller as a typed failure.
452func TestOpenStreamRefusedByCoreReachesClient(t *testing.T) {
453 e := startEndpoint(t, Config{})
454 client, _ := e.pair()
455 core := e.authority.last()
456 core.openErr = ports.BrokerError{Code: ports.BrokerErrorIncompatible}
457
458 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
459 defer cancel()
460 stream, err := client.OpenStream(ctx, openRequest(1))
461 require.Error(t, err)
462 require.Nil(t, stream)
463 var failure ports.BrokerError
464 require.ErrorAs(t, err, &failure)
465 require.Equal(t, ports.BrokerErrorIncompatible, failure.Code)
466
467 // The connection survives a stream-local refusal: once the core admits
468 // streams again, a later operation still completes.
469 core.mu.Lock()
470 core.openErr = nil
471 core.mu.Unlock()
472 requireConnectionDelegates(t, client)
473}
474
475// TestTerminalStreamFailureFailsOnlyThatStream proves one stream's terminal
476// failure is reported to its owner and leaves a sibling stream usable.
477func TestTerminalStreamFailureFailsOnlyThatStream(t *testing.T) {
478 e := startEndpoint(t, Config{})
479 client, _ := e.pair()
480 core := e.authority.last()
481
482 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
483 defer cancel()
484 first, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
485 require.NoError(t, err)
486 second, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
487 require.NoError(t, err)
488 t.Cleanup(func() { _ = first.Close(); _ = second.Close() })
489
490 core.logicalConn(1).fail(ports.BrokerError{Code: ports.BrokerErrorAttachmentLost})
491
492 select {
493 case <-first.Done():
494 case <-time.After(5 * time.Second):
495 t.Fatal("a failed stream must publish its terminal outcome")
496 }
497 require.Error(t, first.Err())
498 select {
499 case <-second.Done():
500 t.Fatal("a failing sibling must not settle an unrelated stream")
501 default:
502 }
503
504 // The sibling still carries traffic.
505 require.NoError(t, second.SendClient(protocol.Ping{}))
506 select {
507 case message := <-core.logicalConn(2).fromClient:
508 require.Equal(t, protocol.Ping{}, message)
509 case <-time.After(5 * time.Second):
510 t.Fatal("sibling stream stopped carrying traffic")
511 }
512 requireConnectionDelegates(t, client)
513}
514
515// TestCloseStreamRetiresBothSides proves a client close reaches the broker core
516// and retires the stream identity.
517func TestCloseStreamRetiresBothSides(t *testing.T) {
518 e := startEndpoint(t, Config{})
519 client, _ := e.pair()
520 core := e.authority.last()
521
522 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
523 defer cancel()
524 stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
525 require.NoError(t, err)
526 require.NoError(t, client.CloseStream(client.ConnectionID(), 1))
527
528 require.Eventually(t, func() bool {
529 core.mu.Lock()
530 defer core.mu.Unlock()
531 for _, id := range core.closedIDs {
532 if id == 1 {
533 return true
534 }
535 }
536 return false
537 }, 5*time.Second, 10*time.Millisecond, "the broker core must retire the closed stream")
538 require.NoError(t, stream.Close(), "closing an already retired stream is idempotent")
539}
540
541// TestMembershipMutationsDelegateEndToEnd proves the whole membership contract
542// travels the real AF_UNIX carriage in both directions: AddHost returns the
543// registration the admitted core minted, UpdateHostPolicy returns the
544// generation-advanced registration for the exact authority the caller named, and
545// RemoveHost returns the core's exact bool. Each call delegates exactly once with
546// the caller's own endpoint, policy, and exact registration, so the adapter never
547// re-derives authority on the way.
548func TestMembershipMutationsDelegateEndToEnd(t *testing.T) {
549 e := startEndpoint(t, Config{})
550 client, _ := e.pair()
551 core := e.authority.last()
552
553 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
554 defer cancel()
555
556 added, err := client.AddHost(ctx, "new@host:22", testPolicy())
557 require.NoError(t, err)
558 require.Equal(t, "new@host:22", added.Endpoint)
559 require.NotZero(t, added.Generation, "an addition answer is fenced authority")
560 require.NoError(t, added.Validate())
561
562 updated, err := client.UpdateHostPolicy(ctx, added, testPolicy())
563 require.NoError(t, err)
564 require.Equal(t, added.Endpoint, updated.Endpoint)
565 require.Equal(t, added.Incarnation, updated.Incarnation, "a policy update preserves identity")
566 require.Equal(t, added.Generation+1, updated.Generation,
567 "a policy update returns a registration advanced by exactly one generation")
568
569 core.reportRemoved(true)
570 removed, err := client.RemoveHost(ctx, updated)
571 require.NoError(t, err)
572 require.True(t, removed, "a present exact registration must report removal")
573
574 core.reportRemoved(false)
575 removed, err = client.RemoveHost(ctx, updated)
576 require.NoError(t, err)
577 require.False(t, removed, "an absent exact registration is an idempotent no-op, not an error")
578
579 adds := core.addCalls()
580 require.Len(t, adds, 1, "one addition is exactly one delegation")
581 require.Equal(t, "new@host:22", adds[0].Endpoint)
582 require.Equal(t, testPolicy(), adds[0].Policy, "the caller's exact policy must be delegated unmodified")
583 require.Equal(t, added, adds[0].Result)
584
585 updates := core.updateCalls()
586 require.Len(t, updates, 1, "one policy update is exactly one delegation")
587 require.Equal(t, added, updates[0].Expected, "the exact expected registration must be delegated unmodified")
588 require.Equal(t, testPolicy(), updates[0].Policy, "the caller's exact replacement policy must be delegated")
589 require.Equal(t, updated, updates[0].Result)
590
591 removes := core.removeCalls()
592 require.Len(t, removes, 2, "each removal is exactly one delegation")
593 require.Equal(t, updated, removes[0].Expected, "the exact expected registration must be delegated")
594 require.Equal(t, updated, removes[1].Expected)
595}
596
597// TestMembershipRefusalCodesEndToEnd proves the two membership refusal codes
598// travel end to end: a stale generation removal is host conflict (11) and an
599// immutable registry is membership-immutable (12). The refusal reaches the caller
600// as a typed ports.BrokerError carrying the exact code and the matching sentinel
601// as its cause, reports no authority, and never invents a registration or a
602// removal.
603func TestMembershipRefusalCodesEndToEnd(t *testing.T) {
604 cases := []struct {
605 name string
606 configure func(core *fakeCore)
607 call func(ctx context.Context, client ports.BrokerService) (domain.RemoteRegistration, bool, error)
608 wantCode ports.BrokerErrorCode
609 }{
610 {
611 name: "stale generation removal is host conflict 11",
612 configure: func(core *fakeCore) { core.refuseRemoveHost(ports.ErrBrokerHostConflict) },
613 call: func(ctx context.Context, client ports.BrokerService) (domain.RemoteRegistration, bool, error) {
614 removed, err := client.RemoveHost(ctx, mutationRegistration("known"))
615 return domain.RemoteRegistration{}, removed, err
616 },
617 wantCode: ports.BrokerErrorHostConflict,
618 },
619 {
620 name: "immutable registry is membership immutable 12",
621 configure: func(core *fakeCore) { core.refuseAddHost(ports.ErrBrokerMembershipImmutable) },
622 call: func(ctx context.Context, client ports.BrokerService) (domain.RemoteRegistration, bool, error) {
623 registration, err := client.AddHost(ctx, "new@host:22", testPolicy())
624 return registration, false, err
625 },
626 wantCode: ports.BrokerErrorMembershipImmutable,
627 },
628 }
629 require.Equal(t, ports.BrokerErrorCode(11), ports.BrokerErrorHostConflict)
630 require.Equal(t, ports.BrokerErrorCode(12), ports.BrokerErrorMembershipImmutable)
631 for _, tc := range cases {
632 t.Run(tc.name, func(t *testing.T) {
633 e := startEndpoint(t, Config{})
634 client, _ := e.pair()
635 core := e.authority.last()
636 tc.configure(core)
637
638 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
639 defer cancel()
640 registration, removed, err := tc.call(ctx, client)
641 require.Equal(t, domain.RemoteRegistration{}, registration, "a refusal never invents authority")
642 require.False(t, removed, "a refusal never reports the host gone")
643 require.Error(t, err)
644 var failure ports.BrokerError
645 require.ErrorAs(t, err, &failure)
646 require.Equal(t, tc.wantCode, failure.Code, "the exact refusal code must reach the caller")
647 require.NotEqual(t, ports.BrokerErrorOutcomeUnknown, failure.Code,
648 "a definite refusal is never mistaken for a refresh-required outcome")
649
650 // The connection survived the refusal: a later delegation still works.
651 requireConnectionDelegates(t, client)
652 })
653 }
654}
655
656// TestMembershipWireRefusalFraming proves the refusal framing the server emits is
657// exactly the contract the wire validates: a failed result carrying its typed
658// error and no authority, under the exact code, with the operation identity the
659// caller used. It drives the raw carriage so the frames are observed as a peer
660// would receive them.
661func TestMembershipWireRefusalFraming(t *testing.T) {
662 cases := []struct {
663 name string
664 configure func(core *fakeCore)
665 send func(scope brokerwire.Scope) brokerwire.ClientMessage
666 kind brokerwire.RegisterMutationKind
667 wantCode ports.BrokerErrorCode
668 }{
669 {
670 name: "remove conflict",
671 configure: func(core *fakeCore) { core.refuseRemoveHost(ports.ErrBrokerHostConflict) },
672 send: func(scope brokerwire.Scope) brokerwire.ClientMessage {
673 return brokerwire.RemoveHost{
674 Epoch: scope.Epoch, Connection: scope.Connection,
675 Operation: ports.BrokerOperationID{0x31}, Registration: mutationRegistration("known"),
676 }
677 },
678 kind: brokerwire.MutationKindRemoveHost,
679 wantCode: ports.BrokerErrorHostConflict,
680 },
681 {
682 name: "immutable addition",
683 configure: func(core *fakeCore) { core.refuseAddHost(ports.ErrBrokerMembershipImmutable) },
684 send: func(scope brokerwire.Scope) brokerwire.ClientMessage {
685 return brokerwire.AddHost{
686 Epoch: scope.Epoch, Connection: scope.Connection,
687 Operation: ports.BrokerOperationID{0x32}, Endpoint: "new@host:22", Policy: testPolicy(),
688 }
689 },
690 kind: brokerwire.MutationKindAddHost,
691 wantCode: ports.BrokerErrorMembershipImmutable,
692 },
693 }
694 for _, tc := range cases {
695 t.Run(tc.name, func(t *testing.T) {
696 e := startEndpoint(t, Config{})
697 raw := rawDial(t, e, brokerwire.DefaultCeilings())
698 scope := raw.register(t)
699 tc.configure(e.authority.last())
700
701 request := tc.send(scope)
702 raw.send(t, request)
703 result, ok := raw.recv(t).(brokerwire.OperationResult)
704 require.True(t, ok, "a membership mutation must be answered with an OperationResult")
705 require.Equal(t, scope.Connection, result.Connection)
706 require.Equal(t, operationOf(t, request), result.Operation)
707 require.Equal(t, ports.BrokerOutcomeFailed, result.Outcome)
708 require.True(t, result.HasError, "a failure must carry its typed detail")
709 require.Equal(t, tc.wantCode, result.Error.Code, "the exact refusal code must travel")
710 require.False(t, result.Removed, "a failure carries no removal authority")
711 require.Equal(t, domain.RemoteRegistration{}, result.Registration,
712 "a failure carries no registration authority")
713 // The emitted result satisfies the same rule the peer enforces for
714 // this request kind.
715 require.NoError(t, result.ValidateForMutation(tc.kind))
716 })
717 }
718}
719
720// TestMembershipStoreOutcomeUnknownMapsToOutcomeUnknown proves an indeterminate
721// durable write is never presented as a definite refusal: both the value and the
722// pointer form of the store's outcome-unknown error become an outcome-unknown
723// result carrying the outcome-unknown code and no authority, and the caller
724// observes ports.BrokerErrorOutcomeUnknown so it refreshes rather than retries.
725func TestMembershipStoreOutcomeUnknownMapsToOutcomeUnknown(t *testing.T) {
726 cause := errors.New("torn write after the commit point")
727 cases := map[string]error{
728 "value": ports.BrokerStoreOutcomeUnknownError{Err: cause},
729 "pointer": &ports.BrokerStoreOutcomeUnknownError{Err: cause},
730 }
731 for name, stored := range cases {
732 t.Run(name, func(t *testing.T) {
733 e := startEndpoint(t, Config{})
734 raw := rawDial(t, e, brokerwire.DefaultCeilings())
735 scope := raw.register(t)
736 core := e.authority.last()
737 core.refuseAddHost(stored)
738
739 request := brokerwire.AddHost{
740 Epoch: scope.Epoch, Connection: scope.Connection,
741 Operation: ports.BrokerOperationID{0x41}, Endpoint: "new@host:22", Policy: testPolicy(),
742 }
743 raw.send(t, request)
744 result, ok := raw.recv(t).(brokerwire.OperationResult)
745 require.True(t, ok)
746 require.Equal(t, ports.BrokerOutcomeUnknown, result.Outcome,
747 "an indeterminate write is outcome-unknown, never a definite failure")
748 require.Equal(t, ports.BrokerErrorOutcomeUnknown, result.Error.Code)
749 require.False(t, result.Removed)
750 require.Equal(t, domain.RemoteRegistration{}, result.Registration)
751 require.NoError(t, result.ValidateForMutation(brokerwire.MutationKindAddHost))
752
753 // The same indeterminate write reaches the public caller as the
754 // refresh-required code, so it is never mistaken for a retryable
755 // conflict or a plain failure.
756 client, _ := e.pair()
757 fresh := e.authority.last()
758 fresh.refuseAddHost(stored)
759 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
760 defer cancel()
761 added, err := client.AddHost(ctx, "new@host:22", testPolicy())
762 require.Equal(t, domain.RemoteRegistration{}, added)
763 var failure ports.BrokerError
764 require.ErrorAs(t, err, &failure)
765 require.Equal(t, ports.BrokerErrorOutcomeUnknown, failure.Code)
766 })
767 }
768}
769
770// TestServerAbortsResultMissingRequiredAuthority proves the server never
771// publishes an addition success that cannot fence the mutation it answers: an
772// admitted core that returns no registration for a successful AddHost is a
773// protocol violation, so the session settles without answering the caller
774// instead of sending a result the peer would refuse.
775func TestServerAbortsResultMissingRequiredAuthority(t *testing.T) {
776 e := startEndpoint(t, Config{})
777 raw := rawDial(t, e, brokerwire.DefaultCeilings())
778 scope := raw.register(t)
779 session := e.accept()
780 core := e.authority.last()
781 core.zeroAddRegistration()
782
783 raw.send(t, brokerwire.AddHost{
784 Epoch: scope.Epoch, Connection: scope.Connection,
785 Operation: ports.BrokerOperationID{0x51}, Endpoint: "new@host:22", Policy: testPolicy(),
786 })
787
788 // No result travels; the connection settles as a protocol violation.
789 err := raw.awaitReadError(t, 5*time.Second)
790 require.Error(t, err, "a success without its required authority must settle the connection")
791 awaitSessionDone(t, session.(*serverSession))
792 require.ErrorIs(t, session.(*serverSession).terminalErr(), ErrProtocol)
793 require.Error(t, session.Close(), "a protocol violation is not an orderly end")
794
795 // The addition was delegated exactly once; the adapter refused to publish an
796 // unfenceable answer rather than re-delegating or inventing one.
797 require.Len(t, core.addCalls(), 1)
798}
799
800// TestDisconnectCleanupReleasesEveryOwnedResource proves a client disconnect
801// deterministically retires its streams, closes its admitted core service, and
802// releases its listener slot.
803func TestDisconnectCleanupReleasesEveryOwnedResource(t *testing.T) {
804 e := startEndpoint(t, Config{MaxClients: 2, HandshakeTimeout: time.Second})
805 client, _ := e.pair()
806 core := e.authority.last()
807
808 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
809 defer cancel()
810 _, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
811 require.NoError(t, err)
812 conn := core.logicalConn(1)
813 require.NotNil(t, conn)
814
815 require.NoError(t, client.Close())
816
817 select {
818 case <-core.done:
819 case <-time.After(5 * time.Second):
820 t.Fatal("the admitted core service must be closed on disconnect")
821 }
822 select {
823 case <-conn.done:
824 case <-time.After(5 * time.Second):
825 t.Fatal("the bridged core connection must be closed on disconnect")
826 }
827
828 require.Eventually(t, func() bool {
829 return e.listener.(*listener).liveSessions() == 0
830 }, 5*time.Second, 10*time.Millisecond, "a retired session must leave the listener's live set")
831
832 // The listener slot is reusable: a fresh client is admitted.
833 fresh := e.dial()
834 session := e.accept()
835 require.Equal(t, fresh.ConnectionID(), session.ConnectionID())
836}
837
838// TestStalledSubscriberDoesNotBlockOtherClients proves publications are
839// per-connection: a client that subscribes and never reads cannot delay another
840// client's publication or operations.
841func TestStalledSubscriberDoesNotBlockOtherClients(t *testing.T) {
842 e := startEndpoint(t, Config{MaxClients: 4})
843 stalled, _ := e.pair()
844 _, err := stalled.Subscribe()
845 require.NoError(t, err)
846
847 active, _ := e.pair()
848 sub, err := active.Subscribe()
849 require.NoError(t, err)
850 t.Cleanup(sub.Close)
851
852 for revision := ports.BrokerRevision(1); revision <= 20; revision++ {
853 e.publish(e.epoch, revision)
854 }
855 require.Eventually(t, func() bool { return active.Snapshot().Revision == 20 }, 5*time.Second, 10*time.Millisecond)
856
857 requireConnectionDelegates(t, active)
858}
859
860// TestStalledClientStreamWaitsForCredit proves a local consumer that stops
861// reading slows the broker relay instead of losing its stream: the daemon's
862// messages beyond one stream window wait at the broker, every one is delivered
863// in order once the consumer reads again, and the connection stays usable
864// meanwhile.
865func TestStalledClientStreamWaitsForCredit(t *testing.T) {
866 e := startEndpoint(t, Config{MaxClients: 2})
867 client := e.dial()
868 _ = e.accept()
869 core := e.authority.last()
870
871 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
872 defer cancel()
873 stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
874 require.NoError(t, err)
875 t.Cleanup(func() { _ = stream.Close() })
876 conn := core.logicalConn(1)
877 require.NotNil(t, conn)
878 // Complete the session preamble before the daemon speaks.
879 require.NoError(t, stream.SendClient(protocol.Ping{}))
880 select {
881 case <-conn.fromClient:
882 case <-ctx.Done():
883 t.Fatal("client message never reached the broker core")
884 }
885
886 // Well over one stream window of daemon output while nothing reads. Side
887 // effect Outputs need no view context and are never compressed.
888 const messages = 24
889 data := bytes.Repeat([]byte{'x'}, 512<<10)
890 go func() {
891 for i := range messages {
892 payload := append([]byte{byte(i)}, data...)
893 select {
894 case conn.toClient <- protocol.Output{Epoch: 1, Size: domain.Size{Cols: 80, Rows: 24}, Data: payload}:
895 case <-ctx.Done():
896 return
897 }
898 }
899 }()
900
901 // The relay fills exactly one window of the client pipe and then waits on
902 // credit instead of failing: the stream stays open and the connection keeps
903 // serving other work.
904 local, ok := stream.(*clientStream)
905 require.True(t, ok)
906 full := brokerwire.StreamWindowBytes - brokerwire.StreamChunkCredit(int(brokerwire.MaxStreamChunkBytes))
907 held := func() uint64 {
908 local.pipe.mu.Lock()
909 defer local.pipe.mu.Unlock()
910 return local.pipe.recvUsed + local.pipe.recvReturn
911 }
912 require.Eventually(t, func() bool { return held() > full }, 5*time.Second, time.Millisecond,
913 "the relay must fill one stream window")
914 // Data beyond the credit would settle the stream with ErrStreamCredit.
915 select {
916 case <-stream.Done():
917 t.Fatalf("a stalled consumer must not settle its stream: %v", stream.Err())
918 default:
919 }
920 requireConnectionDelegates(t, client)
921
922 for i := range messages {
923 message, err := stream.ReceiveServer()
924 require.NoError(t, err)
925 got, ok := message.(protocol.Output)
926 require.True(t, ok, "message %d is %T", i, message)
927 require.Len(t, got.Data, len(data)+1)
928 require.Equal(t, byte(i), got.Data[0], "messages arrive in order")
929 }
930}
931
932// TestOrderlyStreamCloseReportsNoError proves a local close is an orderly end:
933// Done closes and Err stays nil, so a caller can distinguish it from a failure.
934func TestOrderlyStreamCloseReportsNoError(t *testing.T) {
935 e := startEndpoint(t, Config{})
936 client, _ := e.pair()
937
938 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
939 defer cancel()
940 stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
941 require.NoError(t, err)
942 require.NoError(t, stream.Close())
943 select {
944 case <-stream.Done():
945 case <-time.After(5 * time.Second):
946 t.Fatal("a local close must publish its terminal outcome")
947 }
948 require.NoError(t, stream.Err())
949 require.NoError(t, stream.Close(), "closing twice is idempotent")
950}
951
952// TestScopeIdentityFencing proves the client adapter refuses a caller-supplied
953// identity that belongs to another connection and stream identities that do not
954// advance.
955func TestScopeIdentityFencing(t *testing.T) {
956 e := startEndpoint(t, Config{})
957 client, _ := e.pair()
958
959 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
960 defer cancel()
961 foreign := ports.BrokerConnectionID{0x99}
962 _, err := client.OpenStream(ctx, ports.BrokerOpenStreamRequest{
963 Purpose: ports.BrokerStreamControl, Local: true, Stream: 1, Connection: foreign, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
964 })
965 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
966
967 _, err = client.OpenStream(ctx, ports.BrokerOpenStreamRequest{
968 Purpose: ports.BrokerStreamControl, Local: true, Stream: 1, Epoch: e.epoch + 1, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
969 })
970 require.ErrorIs(t, err, ErrScopeMismatch)
971 var stale ports.BrokerError
972 require.ErrorAs(t, err, &stale)
973 require.Equal(t, ports.BrokerErrorStaleEpoch, stale.Code)
974
975 // A concurrent lower identity is admitted: an ID allocated before a higher
976 // one can legitimately arrive after it.
977 first, err := client.OpenStream(ctx, openRequest(2))
978 require.NoError(t, err)
979 t.Cleanup(func() { _ = first.Close() })
980 second, err := client.OpenStream(ctx, openRequest(1))
981 require.NoError(t, err, "a concurrent lower identity is admitted, not refused as stale")
982 t.Cleanup(func() { _ = second.Close() })
983 // Replaying a consumed identity is refused, and so is an identity evicted
984 // past the anti-replay window.
985 _, err = client.OpenStream(ctx, openRequest(1))
986 require.ErrorIs(t, err, ports.BrokerAdmissionStale, "a replayed identity is stale")
987 _, err = client.OpenStream(ctx, openRequest(ports.BrokerStreamWindowSize+3))
988 require.NoError(t, err)
989 _, err = client.OpenStream(ctx, openRequest(1))
990 require.ErrorIs(t, err, ports.BrokerAdmissionStale, "an identity evicted from the window is stale")
991 _, err = client.OpenStream(ctx, openRequest(0))
992 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid, "a zero identity is never admitted")
993}
994
995// TestRequestReconcileIsBoundedHint proves the reconcile hint is sent only for an
996// endpoint the connection already holds a registration for, and that it is
997// otherwise a no-op.
998func TestRequestReconcileIsBoundedHint(t *testing.T) {
999 e := startEndpoint(t, Config{})
1000 client, _ := e.pair()
1001 core := e.authority.last()
1002
1003 client.RequestReconcile("unknown@host:22")
1004 time.Sleep(50 * time.Millisecond)
1005 core.mu.Lock()
1006 require.Empty(t, core.reconcile)
1007 core.mu.Unlock()
1008
1009 sub, err := client.Subscribe()
1010 require.NoError(t, err)
1011 t.Cleanup(sub.Close)
1012 snapshot := e.publish(e.epoch, 4)
1013 require.Eventually(t, func() bool { return client.Snapshot().Revision == 4 }, 5*time.Second, 10*time.Millisecond)
1014
1015 client.RequestReconcile(snapshot.Daemons[0].Endpoint)
1016 require.Eventually(t, func() bool {
1017 core.mu.Lock()
1018 defer core.mu.Unlock()
1019 return len(core.reconcile) == 1 && core.reconcile[0] == snapshot.Daemons[0].Endpoint
1020 }, 5*time.Second, 10*time.Millisecond)
1021}
1022
1023// TestCoreErrorSurfacesAsTypedFailure proves a typed core failure reaches the
1024// client through the stream lifecycle.
1025func TestCoreErrorSurfacesAsTypedFailure(t *testing.T) {
1026 e := startEndpoint(t, Config{})
1027 client, _ := e.pair()
1028 core := e.authority.last()
1029
1030 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
1031 defer cancel()
1032 stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
1033 require.NoError(t, err)
1034 t.Cleanup(func() { _ = stream.Close() })
1035
1036 core.logicalConn(1).fail(ports.BrokerError{Code: ports.BrokerErrorAttachmentLost, Text: "daemon went away"})
1037 select {
1038 case <-stream.Done():
1039 case <-time.After(5 * time.Second):
1040 t.Fatal("core loss must settle the stream")
1041 }
1042 var failure ports.BrokerError
1043 require.ErrorAs(t, stream.Err(), &failure)
1044 require.Equal(t, ports.BrokerErrorAttachmentLost, failure.Code)
1045 require.Equal(t, "daemon went away", failure.Text)
1046}
1047
1048// TestSessionHandshakeFailureIsStreamLocal proves a malformed inner session
1049// stream settles only that stream.
1050func TestSessionHandshakeFailureIsStreamLocal(t *testing.T) {
1051 e := startEndpoint(t, Config{})
1052 client, _ := e.pair()
1053 core := e.authority.last()
1054
1055 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
1056 defer cancel()
1057 stream, err := client.OpenStream(ctx, openRequest(nextStreamID(t, client)))
1058 require.NoError(t, err)
1059 conn := core.logicalConn(1)
1060 require.NotNil(t, conn)
1061
1062 // The client emits a malformed inner session frame: the broker's session
1063 // carriage refuses it and resets only this stream.
1064 raw := stream.(*clientStream)
1065 _, err = raw.pipe.Write([]byte{0, 0, 0, 4, 0xff, 0xff, 0xff, 0xff})
1066 require.NoError(t, err)
1067 select {
1068 case <-stream.Done():
1069 case <-time.After(5 * time.Second):
1070 t.Fatal("a malformed inner frame must settle its stream")
1071 }
1072 requireConnectionDelegates(t, client)
1073
1074 // The broker core sees the stream retired.
1075 require.Eventually(t, func() bool {
1076 return core.logicalConn(1) == nil
1077 }, 5*time.Second, 10*time.Millisecond)
1078}
1079
1080// TestDomainFixturesAreValid guards the test fixtures against drifting from the
1081// contracts the wire enforces.
1082func TestDomainFixturesAreValid(t *testing.T) {
1083 snapshot := testSnapshot(1, 1)
1084 require.NoError(t, snapshot.Validate())
1085 for _, host := range snapshot.Daemons {
1086 require.NoError(t, ports.ValidateDurableHostProjection(host))
1087 }
1088 require.NoError(t, testPolicy().Validate())
1089 catalogueSession := testSessionAt(0, 0)
1090 require.False(t, strings.HasPrefix(catalogueSession.Name, " "))
1091 require.NotEqual(t, domain.RemoteAvailabilityUnknown, testDaemonAt(0).Availability)
1092 require.True(t, errors.Is(errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale), ports.BrokerAdmissionStale))
1093}