eabbcfe fix(broker): report a missing session codec and drop the typed view cache · vev

M internal/adapters/daemonmux/aggregate_test.go +1 -1


@@ -225,7 +225,7 @@ func TestAggregateCloseDrainsQueuedConnections(t *testing.T) {
225	225
226	226		// Every connection the aggregate held is released, including the one its
227	227		// forwarder was holding while backpressured; the child itself is closed.
228		-	for i, broker := range []*LogicalConnection{first, second, third} {
	228	+	for i, broker := range []typedLogical{first, second, third} {
229	229		require.Eventually(t, func() bool { return channelClosed(broker.Done()) }, listenerTestDeadline, time.Millisecond,
230	230		"queued connection %d was not cleaned by Close", i+1)
231	231		}


@@ -225,7 +225,7 @@ func TestAggregateCloseDrainsQueuedConnections(t *testing.T) {
225			225
226		// Every connection the aggregate held is released, including the one its	226		// Every connection the aggregate held is released, including the one its
227		// forwarder was holding while backpressured; the child itself is closed.	227		// forwarder was holding while backpressured; the child itself is closed.
228	-	for i, broker := range []*LogicalConnection{first, second, third} {	228	+	for i, broker := range []typedLogical{first, second, third} {
229		require.Eventually(t, func() bool { return channelClosed(broker.Done()) }, listenerTestDeadline, time.Millisecond,	229		require.Eventually(t, func() bool { return channelClosed(broker.Done()) }, listenerTestDeadline, time.Millisecond,
230		"queued connection %d was not cleaned by Close", i+1)	230		"queued connection %d was not cleaned by Close", i+1)
231		}	231		}

M internal/adapters/daemonmux/listener_test.go +7 -8


@@ -201,11 +201,11 @@ func (h listenerHarness) open(stream int) (LogicalConnection, error) {
201	201		return h.connector.Open(ctx, controlRequest(stream))
202	202		}
203	203
204		-	func (h listenerHarness) mustOpen(stream int) LogicalConnection {
	204	+	func (h *listenerHarness) mustOpen(stream int) typedLogical {
205	205		h.t.Helper()
206	206		connection, err := h.open(stream)
207	207		require.NoError(h.t, err, "open stream %d", stream)
208		-	return connection
	208	+	return newTypedLogical(connection)
209	209		}
210	210
211	211		// pending reports the admitted streams waiting in the accept queue.

@@ -285,9 +285,8 @@ func servePongs(conn ports.ServerConnection) error {
285	285		}
286	286
287	287		// listenerRoundTrip proves one typed exchange on a logical connection.
288		-	func listenerRoundTrip(t *testing.T, stream ports.BrokerEnvelopeStream) {
	288	+	func listenerRoundTrip(t *testing.T, conn ports.ClientConnection) {
289	289		t.Helper()
290		-	conn := asTyped(stream)
291	290		require.NoError(t, conn.SendClient(protocol.Ping{}))
292	291		message, err := conn.ReceiveServer()
293	292		require.NoError(t, err)

@@ -482,7 +481,7 @@ func TestListenerHundredTypedAdmissions(t *testing.T) {
482	481		}
483	482
484	483		for _, connection := range connections {
485		-	listenerRoundTrip(t, connection)
	484	+	listenerRoundTrip(t, newTypedLogical(connection))
486	485		}
487	486
488	487		for i, connection := range connections {

@@ -592,7 +591,7 @@ func TestListenerQueueFullRefusesOnlyOffendingStream(t *testing.T) {
592	591		go func() { _ = servePongs(first) }()
593	592		select {
594	593		case connection := <-opened:
595		-	listenerRoundTrip(t, connection)
	594	+	listenerRoundTrip(t, newTypedLogical(connection))
596	595		case err := <-openErrs:
597	596		t.Fatalf("queued stream did not open: %v", err)
598	597		case <-time.After(listenerTestDeadline):

@@ -709,7 +708,7 @@ func TestListenerDeadlineConsumesQueueDelay(t *testing.T) {
709	708		select {
710	709		case connection := <-opened:
711	710		go func() { _ = servePongs(daemonConnection) }()
712		-	listenerRoundTrip(t, connection)
	711	+	listenerRoundTrip(t, newTypedLogical(connection))
713	712		case <-time.After(listenerTestDeadline):
714	713		t.Fatal("queued stream did not open")
715	714		}

@@ -767,7 +766,7 @@ func TestListenerMalformedStreamDoesNotStopAccept(t *testing.T) {
767	766		func TestListenerPhysicalLossUnblocksAll(t *testing.T) {
768	767		h := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
769	768
770		-	connections := make([]*LogicalConnection, 0, 2)
	769	+	connections := make([]typedLogical, 0, 2)
771	770		daemonConnections := make([]*ListenerConnection, 0, 2)
772	771		for i := 1; i <= 2; i++ {
773	772		probe := h.acceptAsync()

 	return h.connector.Open(ctx, controlRequest(stream))
 }
-func (h *listenerHarness) mustOpen(stream int) *LogicalConnection {
+func (h *listenerHarness) mustOpen(stream int) typedLogical {
 	h.t.Helper()
 	connection, err := h.open(stream)
 	require.NoError(h.t, err, "open stream %d", stream)
-	return connection
+	return newTypedLogical(connection)
 }
 // pending reports the admitted streams waiting in the accept queue.
 }
 // listenerRoundTrip proves one typed exchange on a logical connection.
-func listenerRoundTrip(t *testing.T, stream ports.BrokerEnvelopeStream) {
+func listenerRoundTrip(t *testing.T, conn ports.ClientConnection) {
 	t.Helper()
-	conn := asTyped(stream)
 	require.NoError(t, conn.SendClient(protocol.Ping{}))
 	message, err := conn.ReceiveServer()
 	require.NoError(t, err)
 	}
 	for _, connection := range connections {
-		listenerRoundTrip(t, connection)
+		listenerRoundTrip(t, newTypedLogical(connection))
 	}
 	for i, connection := range connections {
 	go func() { _ = servePongs(first) }()
 	select {
 	case connection := <-opened:
-		listenerRoundTrip(t, connection)
+		listenerRoundTrip(t, newTypedLogical(connection))
 	case err := <-openErrs:
 		t.Fatalf("queued stream did not open: %v", err)
 	case <-time.After(listenerTestDeadline):
 	select {
 	case connection := <-opened:
 		go func() { _ = servePongs(daemonConnection) }()
-		listenerRoundTrip(t, connection)
+		listenerRoundTrip(t, newTypedLogical(connection))
 	case <-time.After(listenerTestDeadline):
 		t.Fatal("queued stream did not open")
 	}
 func TestListenerPhysicalLossUnblocksAll(t *testing.T) {
 	h := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
-	connections := make([]*LogicalConnection, 0, 2)
+	connections := make([]typedLogical, 0, 2)
 	daemonConnections := make([]*ListenerConnection, 0, 2)
 	for i := 1; i <= 2; i++ {
 		probe := h.acceptAsync()

M internal/adapters/daemonmux/logical_conn_test.go +1 -1


@@ -324,7 +324,7 @@ func openWithDeadline(t testing.T, connector LogicalConnector, stream int) typ
324	324		defer cancel()
325	325		connection, err := connector.Open(ctx, controlRequest(stream))
326	326		require.NoError(t, err)
327		-	return typedLogical{typedStream: asTyped(connection), LogicalConnection: connection}
	327	+	return newTypedLogical(connection)
328	328		}
329	329
330	330		// watchRegistered reports whether the pump still holds an inbound watch entry


@@ -324,7 +324,7 @@ func openWithDeadline(t testing.T, connector LogicalConnector, stream int) typ
324		defer cancel()	324		defer cancel()
325		connection, err := connector.Open(ctx, controlRequest(stream))	325		connection, err := connector.Open(ctx, controlRequest(stream))
326		require.NoError(t, err)	326		require.NoError(t, err)
327	-	return typedLogical{typedStream: asTyped(connection), LogicalConnection: connection}	327	+	return newTypedLogical(connection)
328		}	328		}
329			329
330		// watchRegistered reports whether the pump still holds an inbound watch entry	330		// watchRegistered reports whether the pump still holds an inbound watch entry

M internal/adapters/daemonmux/typed_stream_test.go +7 -10


@@ -2,7 +2,6 @@ package daemonmux
2	2
3	3		import (
4	4		"context"
5		-	"sync"
6	5
7	6		"github.com/bnema/vev/internal/adapters/sessionwire"
8	7		"github.com/bnema/vev/internal/ports"

@@ -19,16 +18,10 @@ var _ ports.BrokerLogicalConnection = typedStream{}
19	18
20	19		func (s typedStream) Close() error { return s.ClientConnection.Close() }
21	20
22		-	// typedViews keeps one session codec per raw stream: the codec owns the
23		-	// preamble, so a second codec on the same stream would restart it.
24		-	var typedViews sync.Map // ports.BrokerEnvelopeStream -> typedStream
25		-
	21	+	// asTyped builds the one typed view of stream. The codec owns the session
	22	+	// preamble, so each raw stream must get exactly one view.
26	23		func asTyped(stream ports.BrokerEnvelopeStream) typedStream {
27		-	if view, ok := typedViews.Load(stream); ok {
28		-	return view.(typedStream)
29		-	}
30		-	view, _ := typedViews.LoadOrStore(stream, typedStream{ClientConnection: sessionwire.BrokerCodec{}.Client(stream), BrokerEnvelopeStream: stream})
31		-	return view.(typedStream)
	24	+	return typedStream{ClientConnection: sessionwire.BrokerCodec{}.Client(stream), BrokerEnvelopeStream: stream}
32	25		}
33	26
34	27		// typedLogical is a typed view of one concrete logical connection.

@@ -39,6 +32,10 @@ type typedLogical struct {
39	32
40	33		func (c typedLogical) Close() error { return c.typedStream.Close() }
41	34
	35	+	func newTypedLogical(connection *LogicalConnection) typedLogical {
	36	+	return typedLogical{typedStream: asTyped(connection), LogicalConnection: connection}
	37	+	}
	38	+
42	39		type envelopeOpener interface {
43	40		OpenStream(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error)
44	41		}

 import (
 	"context"
-	"sync"
 	"github.com/bnema/vev/internal/adapters/sessionwire"
 	"github.com/bnema/vev/internal/ports"
 func (s typedStream) Close() error { return s.ClientConnection.Close() }
-// typedViews keeps one session codec per raw stream: the codec owns the
+// asTyped builds the one typed view of stream. The codec owns the session
-// preamble, so a second codec on the same stream would restart it.
+// preamble, so each raw stream must get exactly one view.
-var typedViews sync.Map // ports.BrokerEnvelopeStream -> typedStream
 func asTyped(stream ports.BrokerEnvelopeStream) typedStream {
-	if view, ok := typedViews.Load(stream); ok {
+	return typedStream{ClientConnection: sessionwire.BrokerCodec{}.Client(stream), BrokerEnvelopeStream: stream}
-		return view.(typedStream)
-	}
-	view, _ := typedViews.LoadOrStore(stream, typedStream{ClientConnection: sessionwire.BrokerCodec{}.Client(stream), BrokerEnvelopeStream: stream})
-	return view.(typedStream)
 }
 // typedLogical is a typed view of one concrete logical connection.
 func (c typedLogical) Close() error { return c.typedStream.Close() }
+func newTypedLogical(connection *LogicalConnection) typedLogical {
+	return typedLogical{typedStream: asTyped(connection), LogicalConnection: connection}
+}
 type envelopeOpener interface {
 	OpenStream(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error)
 }

M internal/usecase/broker/service.go +4 -1


@@ -65,9 +65,12 @@ func NewAuthority(epoch ports.BrokerEpoch, registry Registry, pool Pool, super
65	65		if len(clocks) > 0 {
66	66		clk = clocks[0]
67	67		}
68		-	if nilDependency(clk) \|\| nilDependency(codec) {
	68	+	if nilDependency(clk) {
69	69		return nil, errors.New("broker: invalid authority clock")
70	70		}
	71	+	if nilDependency(codec) {
	72	+	return nil, errors.New("broker: authority requires a session codec")
	73	+	}
71	74		return &Authority{epoch: epoch, registry: registry, pool: pool, supervisor: supervisor, clock: clk, codec: codec}, nil
72	75		}
73	76

 	if len(clocks) > 0 {
 		clk = clocks[0]
 	}
-	if nilDependency(clk) || nilDependency(codec) {
+	if nilDependency(clk) {
 		return nil, errors.New("broker: invalid authority clock")
 	}
+	if nilDependency(codec) {
+		return nil, errors.New("broker: authority requires a session codec")
+	}
 	return &Authority{epoch: epoch, registry: registry, pool: pool, supervisor: supervisor, clock: clk, codec: codec}, nil
 }