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