fix(daemonmux): carry the attachment admission into the open
Mapping an open-stream request to its mux Open dropped the attachment admission variant and the creation name. The daemon admits an attachment by the variant the stream declares, so every attachment arrived as "no admission" and was refused: one-shot ephemeral creation, named creation, and exact attach all failed before a session was accepted. A real composition surfaced it as "broker unavailable" on the first stream open. Both fields are part of Open and already travel the wire; only this conversion lost them. Pinned end to end: the new test converts each variant, asserts the admission and name survive, and decodes the encoded message back to prove the daemon sees the same values. Verified by mutation: dropping the two fields makes all three variants arrive as zero.
- commit
- aea1a9be54b975f2b04488a9fdfa090e921869e8
- author
- brice · · 2026-09-19 12:34 UTC
- parent
- 976dd32
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +76 -0
A internal/adapters/daemonmux/admission_conversion_test.go +74 -0
| @@ -0,0 +1,74 @@ | |||
| 1 | + | package daemonmux | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "testing" | |
| 5 | + | ||
| 6 | + | "github.com/stretchr/testify/require" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | + | "github.com/bnema/vev/internal/ports" | |
| 10 | + | "github.com/bnema/vev/internal/protocol" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | // TestOpenMessageCarriesAttachmentAdmission proves the request-to-open | |
| 14 | + | // conversion preserves the attachment admission variant and its creation name, | |
| 15 | + | // all the way to the wire and back. | |
| 16 | + | // | |
| 17 | + | // The session daemon admits an attachment by the variant the stream declares, so | |
| 18 | + | // a conversion that drops it delivers every attachment as "no admission" and the | |
| 19 | + | // daemon refuses it: one-shot ephemeral creation, named creation, and exact | |
| 20 | + | // attach all fail before a session is ever accepted, which is exactly how it | |
| 21 | + | // presented from a real composition. | |
| 22 | + | func TestOpenMessageCarriesAttachmentAdmission(t *testing.T) { | |
| 23 | + | base := ports.BrokerOpenStreamRequest{ | |
| 24 | + | Epoch: 1, | |
| 25 | + | Purpose: ports.BrokerStreamAttachment, | |
| 26 | + | Local: true, | |
| 27 | + | Connection: ports.BrokerConnectionID{1}, | |
| 28 | + | Stream: ports.BrokerStreamID(1), | |
| 29 | + | Policy: logicalTestPolicy(), | |
| 30 | + | } | |
| 31 | + | target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"} | |
| 32 | + | ||
| 33 | + | tests := []struct { | |
| 34 | + | name string | |
| 35 | + | request ports.BrokerOpenStreamRequest | |
| 36 | + | admission ports.BrokerStreamAdmission | |
| 37 | + | createAs string | |
| 38 | + | }{ | |
| 39 | + | { | |
| 40 | + | name: "exact attach keeps its exact target", | |
| 41 | + | request: func() ports.BrokerOpenStreamRequest { r := base; r.Admission = ports.BrokerAdmissionExact; r.Target = target; return r }(), | |
| 42 | + | admission: ports.BrokerAdmissionExact, | |
| 43 | + | }, | |
| 44 | + | { | |
| 45 | + | name: "named creation keeps its name", | |
| 46 | + | request: func() ports.BrokerOpenStreamRequest { r := base; r.Admission = ports.BrokerAdmissionCreateNamed; r.Name = "work"; return r }(), | |
| 47 | + | admission: ports.BrokerAdmissionCreateNamed, | |
| 48 | + | createAs: "work", | |
| 49 | + | }, | |
| 50 | + | { | |
| 51 | + | name: "ephemeral creation keeps its variant", | |
| 52 | + | request: func() ports.BrokerOpenStreamRequest { r := base; r.Admission = ports.BrokerAdmissionCreateEphemeral; return r }(), | |
| 53 | + | admission: ports.BrokerAdmissionCreateEphemeral, | |
| 54 | + | }, | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | for _, tc := range tests { | |
| 58 | + | t.Run(tc.name, func(t *testing.T) { | |
| 59 | + | open := openMessage(1, tc.request) | |
| 60 | + | require.Equal(t, tc.admission, open.Admission) | |
| 61 | + | require.Equal(t, tc.createAs, open.Name) | |
| 62 | + | require.Equal(t, tc.request.Target, open.Target) | |
| 63 | + | ||
| 64 | + | // The same values must survive the ceiling-checked wire round trip | |
| 65 | + | // the daemon actually decodes. | |
| 66 | + | decoded, err := DecodeClient(mustEncodeClient(t, open), testEnvelopeCeiling, testChunkCeiling) | |
| 67 | + | require.NoError(t, err) | |
| 68 | + | admitting, ok := decoded.(Open) | |
| 69 | + | require.True(t, ok) | |
| 70 | + | require.Equal(t, tc.admission, admitting.Admission) | |
| 71 | + | require.Equal(t, tc.createAs, admitting.Name) | |
| 72 | + | }) | |
| 73 | + | } | |
| 74 | + | } | |
| @@ -0,0 +1,74 @@ | |||||
| 1 | + | package daemonmux | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "testing" | |||
| 5 | + | ||||
| 6 | + | "github.com/stretchr/testify/require" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | + | "github.com/bnema/vev/internal/ports" | |||
| 10 | + | "github.com/bnema/vev/internal/protocol" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | // TestOpenMessageCarriesAttachmentAdmission proves the request-to-open | |||
| 14 | + | // conversion preserves the attachment admission variant and its creation name, | |||
| 15 | + | // all the way to the wire and back. | |||
| 16 | + | // | |||
| 17 | + | // The session daemon admits an attachment by the variant the stream declares, so | |||
| 18 | + | // a conversion that drops it delivers every attachment as "no admission" and the | |||
| 19 | + | // daemon refuses it: one-shot ephemeral creation, named creation, and exact | |||
| 20 | + | // attach all fail before a session is ever accepted, which is exactly how it | |||
| 21 | + | // presented from a real composition. | |||
| 22 | + | func TestOpenMessageCarriesAttachmentAdmission(t *testing.T) { | |||
| 23 | + | base := ports.BrokerOpenStreamRequest{ | |||
| 24 | + | Epoch: 1, | |||
| 25 | + | Purpose: ports.BrokerStreamAttachment, | |||
| 26 | + | Local: true, | |||
| 27 | + | Connection: ports.BrokerConnectionID{1}, | |||
| 28 | + | Stream: ports.BrokerStreamID(1), | |||
| 29 | + | Policy: logicalTestPolicy(), | |||
| 30 | + | } | |||
| 31 | + | target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "alpha"} | |||
| 32 | + | ||||
| 33 | + | tests := []struct { | |||
| 34 | + | name string | |||
| 35 | + | request ports.BrokerOpenStreamRequest | |||
| 36 | + | admission ports.BrokerStreamAdmission | |||
| 37 | + | createAs string | |||
| 38 | + | }{ | |||
| 39 | + | { | |||
| 40 | + | name: "exact attach keeps its exact target", | |||
| 41 | + | request: func() ports.BrokerOpenStreamRequest { r := base; r.Admission = ports.BrokerAdmissionExact; r.Target = target; return r }(), | |||
| 42 | + | admission: ports.BrokerAdmissionExact, | |||
| 43 | + | }, | |||
| 44 | + | { | |||
| 45 | + | name: "named creation keeps its name", | |||
| 46 | + | request: func() ports.BrokerOpenStreamRequest { r := base; r.Admission = ports.BrokerAdmissionCreateNamed; r.Name = "work"; return r }(), | |||
| 47 | + | admission: ports.BrokerAdmissionCreateNamed, | |||
| 48 | + | createAs: "work", | |||
| 49 | + | }, | |||
| 50 | + | { | |||
| 51 | + | name: "ephemeral creation keeps its variant", | |||
| 52 | + | request: func() ports.BrokerOpenStreamRequest { r := base; r.Admission = ports.BrokerAdmissionCreateEphemeral; return r }(), | |||
| 53 | + | admission: ports.BrokerAdmissionCreateEphemeral, | |||
| 54 | + | }, | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | for _, tc := range tests { | |||
| 58 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 59 | + | open := openMessage(1, tc.request) | |||
| 60 | + | require.Equal(t, tc.admission, open.Admission) | |||
| 61 | + | require.Equal(t, tc.createAs, open.Name) | |||
| 62 | + | require.Equal(t, tc.request.Target, open.Target) | |||
| 63 | + | ||||
| 64 | + | // The same values must survive the ceiling-checked wire round trip | |||
| 65 | + | // the daemon actually decodes. | |||
| 66 | + | decoded, err := DecodeClient(mustEncodeClient(t, open), testEnvelopeCeiling, testChunkCeiling) | |||
| 67 | + | require.NoError(t, err) | |||
| 68 | + | admitting, ok := decoded.(Open) | |||
| 69 | + | require.True(t, ok) | |||
| 70 | + | require.Equal(t, tc.admission, admitting.Admission) | |||
| 71 | + | require.Equal(t, tc.createAs, admitting.Name) | |||
| 72 | + | }) | |||
| 73 | + | } | |||
| 74 | + | } | |||
M internal/adapters/daemonmux/logical_conn.go +2 -0
| @@ -167,6 +167,8 @@ func openMessage(physical PhysicalStreamID, request ports.BrokerOpenStreamReques | |||
| 167 | 167 | Client: request.Stream, | |
| 168 | 168 | }, | |
| 169 | 169 | Purpose: request.Purpose, | |
| 170 | + | Admission: request.Admission, | |
| 171 | + | Name: request.Name, | |
| 170 | 172 | Local: request.Local, | |
| 171 | 173 | Endpoint: request.Endpoint, | |
| 172 | 174 | Registration: request.Registration, | |
| @@ -167,6 +167,8 @@ func openMessage(physical PhysicalStreamID, request ports.BrokerOpenStreamReques | |||||
| 167 | Client: request.Stream, | 167 | Client: request.Stream, | ||
| 168 | }, | 168 | }, | ||
| 169 | Purpose: request.Purpose, | 169 | Purpose: request.Purpose, | ||
| 170 | + | Admission: request.Admission, | |||
| 171 | + | Name: request.Name, | |||
| 170 | Local: request.Local, | 172 | Local: request.Local, | ||
| 171 | Endpoint: request.Endpoint, | 173 | Endpoint: request.Endpoint, | ||
| 172 | Registration: request.Registration, | 174 | Registration: request.Registration, | ||