detached bef813a

feat(brokerwire): carry build identity in registration

commit
bef813ad6a9b528e751f761ccd0c11509f675f02
author
brice · · 2026-09-23 13:32 UTC
parent
fa913e1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
5 files · +110 -11
Unified
UnifiedSplit
M internal/adapters/brokerwire/codec.go +16 -4
@@ -102,7 +102,10 @@ func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.Br
102102 }
103103 switch m := message.(type) {
104104 case Register:
105- return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{}}}, nil
105+ if len(m.Build) > 128 {
106+ return nil, ErrInvalidMessage
107+ }
108+ return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: m.Build}}}, nil
106109 case *Register:
107110 if m == nil {
108111 return nil, ErrInvalidMessage
@@ -273,7 +276,10 @@ func decodeClientEnvelope(envelope *wire.BrokerClientEnvelope, maxChunkBytes uin
273276 if payload.Register == nil {
274277 return nil, ErrInvalidMessage
275278 }
276- return Register{}, nil
279+ if len(payload.Register.GetBuild()) > 128 {
280+ return nil, ErrInvalidMessage
281+ }
282+ return Register{Build: payload.Register.GetBuild()}, nil
277283 case *wire.BrokerClientEnvelope_Subscribe:
278284 return subscribeFromWire(payload.Subscribe)
279285 case *wire.BrokerClientEnvelope_Resync:
@@ -777,6 +783,9 @@ func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.Br
777783 }
778784 switch m := message.(type) {
779785 case Registered:
786+ if len(m.Build) > 128 {
787+ return nil, ErrInvalidMessage
788+ }
780789 if m.Epoch == 0 {
781790 return nil, ErrInvalidMessage
782791 }
@@ -784,7 +793,7 @@ func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.Br
784793 return nil, ErrInvalidMessage
785794 }
786795 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{
787- Scope: scopeToWire(m.Epoch, m.Connection),
796+ Scope: scopeToWire(m.Epoch, m.Connection), Build: m.Build, Retiring: m.Retiring,
788797 }}}, nil
789798 case *Registered:
790799 if m == nil {
@@ -922,7 +931,10 @@ func decodeServerEnvelope(envelope *wire.BrokerServerEnvelope, maxChunkBytes uin
922931 if err != nil {
923932 return nil, ErrInvalidMessage
924933 }
925- return Registered{Epoch: epoch, Connection: connection}, nil
934+ if len(payload.Registered.GetBuild()) > 128 {
935+ return nil, ErrInvalidMessage
936+ }
937+ return Registered{Epoch: epoch, Connection: connection, Build: payload.Registered.GetBuild(), Retiring: payload.Registered.GetRetiring()}, nil
926938 case *wire.BrokerServerEnvelope_SnapshotPart:
927939 return snapshotPartFromWire(payload.SnapshotPart)
928940 case *wire.BrokerServerEnvelope_OperationResult:
@@ -102,7 +102,10 @@ func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.Br
102 }102 }
103 switch m := message.(type) {103 switch m := message.(type) {
104 case Register:104 case Register:
105- return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{}}}, nil105+ if len(m.Build) > 128 {
106+ return nil, ErrInvalidMessage
107+ }
108+ return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: m.Build}}}, nil
106 case *Register:109 case *Register:
107 if m == nil {110 if m == nil {
108 return nil, ErrInvalidMessage111 return nil, ErrInvalidMessage
@@ -273,7 +276,10 @@ func decodeClientEnvelope(envelope *wire.BrokerClientEnvelope, maxChunkBytes uin
273 if payload.Register == nil {276 if payload.Register == nil {
274 return nil, ErrInvalidMessage277 return nil, ErrInvalidMessage
275 }278 }
276- return Register{}, nil279+ if len(payload.Register.GetBuild()) > 128 {
280+ return nil, ErrInvalidMessage
281+ }
282+ return Register{Build: payload.Register.GetBuild()}, nil
277 case *wire.BrokerClientEnvelope_Subscribe:283 case *wire.BrokerClientEnvelope_Subscribe:
278 return subscribeFromWire(payload.Subscribe)284 return subscribeFromWire(payload.Subscribe)
279 case *wire.BrokerClientEnvelope_Resync:285 case *wire.BrokerClientEnvelope_Resync:
@@ -777,6 +783,9 @@ func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.Br
777 }783 }
778 switch m := message.(type) {784 switch m := message.(type) {
779 case Registered:785 case Registered:
786+ if len(m.Build) > 128 {
787+ return nil, ErrInvalidMessage
788+ }
780 if m.Epoch == 0 {789 if m.Epoch == 0 {
781 return nil, ErrInvalidMessage790 return nil, ErrInvalidMessage
782 }791 }
@@ -784,7 +793,7 @@ func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.Br
784 return nil, ErrInvalidMessage793 return nil, ErrInvalidMessage
785 }794 }
786 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{795 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{
787- Scope: scopeToWire(m.Epoch, m.Connection),796+ Scope: scopeToWire(m.Epoch, m.Connection), Build: m.Build, Retiring: m.Retiring,
788 }}}, nil797 }}}, nil
789 case *Registered:798 case *Registered:
790 if m == nil {799 if m == nil {
@@ -922,7 +931,10 @@ func decodeServerEnvelope(envelope *wire.BrokerServerEnvelope, maxChunkBytes uin
922 if err != nil {931 if err != nil {
923 return nil, ErrInvalidMessage932 return nil, ErrInvalidMessage
924 }933 }
925- return Registered{Epoch: epoch, Connection: connection}, nil934+ if len(payload.Registered.GetBuild()) > 128 {
935+ return nil, ErrInvalidMessage
936+ }
937+ return Registered{Epoch: epoch, Connection: connection, Build: payload.Registered.GetBuild(), Retiring: payload.Registered.GetRetiring()}, nil
926 case *wire.BrokerServerEnvelope_SnapshotPart:938 case *wire.BrokerServerEnvelope_SnapshotPart:
927 return snapshotPartFromWire(payload.SnapshotPart)939 return snapshotPartFromWire(payload.SnapshotPart)
928 case *wire.BrokerServerEnvelope_OperationResult:940 case *wire.BrokerServerEnvelope_OperationResult:
M internal/adapters/brokerwire/codec_test.go +55 -0
@@ -1258,3 +1258,58 @@ func FuzzDecodeServer(f *testing.F) {
12581258 _, _ = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)
12591259 })
12601260 }
1261+
1262+func TestBuildRegistrationCodec(t *testing.T) {
1263+ for _, tc := range []struct {
1264+ name string
1265+ build string
1266+ retiring bool
1267+ }{
1268+ {"empty", "", false}, {"matching", "release+abc", false}, {"retiring", "release+def", true},
1269+ } {
1270+ t.Run(tc.name, func(t *testing.T) {
1271+ request := Register{Build: tc.build}
1272+ raw := mustEncodeClient(t, request)
1273+ decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1274+ require.NoError(t, err)
1275+ require.Equal(t, request, decoded)
1276+ want := []byte{0xaa, 0x06, 0x00}
1277+ if tc.build != "" {
1278+ want = append([]byte{0xaa, 0x06, byte(len(tc.build) + 2), 0x0a, byte(len(tc.build))}, []byte(tc.build)...)
1279+ }
1280+ require.Equal(t, want, raw)
1281+ response := Registered{Epoch: 7, Connection: testConnectionID(1), Build: tc.build, Retiring: tc.retiring}
1282+ raw = mustEncodeServer(t, response)
1283+ decodedServer, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1284+ require.NoError(t, err)
1285+ require.Equal(t, response, decodedServer)
1286+ envelope := &wire.BrokerServerEnvelope{}
1287+ require.NoError(t, proto.Unmarshal(raw, envelope))
1288+ require.Equal(t, tc.build, envelope.GetRegistered().GetBuild())
1289+ require.Equal(t, tc.retiring, envelope.GetRegistered().GetRetiring())
1290+ })
1291+ }
1292+ for _, tc := range []struct {
1293+ name string
1294+ server bool
1295+ }{{"client", false}, {"server", true}} {
1296+ t.Run("oversize-"+tc.name, func(t *testing.T) {
1297+ build := strings.Repeat("x", 129)
1298+ if tc.server {
1299+ _, err := EncodeServer(Registered{Epoch: 7, Connection: testConnectionID(1), Build: build}, testEnvelopeCeiling, testChunkCeiling)
1300+ require.ErrorIs(t, err, ErrInvalidMessage)
1301+ raw, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{Scope: scopeToWire(7, testConnectionID(1)), Build: build}}})
1302+ require.NoError(t, err)
1303+ _, err = DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1304+ require.ErrorIs(t, err, ErrInvalidMessage)
1305+ } else {
1306+ _, err := EncodeClient(Register{Build: build}, testEnvelopeCeiling, testChunkCeiling)
1307+ require.ErrorIs(t, err, ErrInvalidMessage)
1308+ raw, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: build}}})
1309+ require.NoError(t, err)
1310+ _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1311+ require.ErrorIs(t, err, ErrInvalidMessage)
1312+ }
1313+ })
1314+ }
1315+}
@@ -1258,3 +1258,58 @@ func FuzzDecodeServer(f *testing.F) {
1258 _, _ = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)1258 _, _ = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)
1259 })1259 })
1260 }1260 }
1261+
1262+func TestBuildRegistrationCodec(t *testing.T) {
1263+ for _, tc := range []struct {
1264+ name string
1265+ build string
1266+ retiring bool
1267+ }{
1268+ {"empty", "", false}, {"matching", "release+abc", false}, {"retiring", "release+def", true},
1269+ } {
1270+ t.Run(tc.name, func(t *testing.T) {
1271+ request := Register{Build: tc.build}
1272+ raw := mustEncodeClient(t, request)
1273+ decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1274+ require.NoError(t, err)
1275+ require.Equal(t, request, decoded)
1276+ want := []byte{0xaa, 0x06, 0x00}
1277+ if tc.build != "" {
1278+ want = append([]byte{0xaa, 0x06, byte(len(tc.build) + 2), 0x0a, byte(len(tc.build))}, []byte(tc.build)...)
1279+ }
1280+ require.Equal(t, want, raw)
1281+ response := Registered{Epoch: 7, Connection: testConnectionID(1), Build: tc.build, Retiring: tc.retiring}
1282+ raw = mustEncodeServer(t, response)
1283+ decodedServer, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1284+ require.NoError(t, err)
1285+ require.Equal(t, response, decodedServer)
1286+ envelope := &wire.BrokerServerEnvelope{}
1287+ require.NoError(t, proto.Unmarshal(raw, envelope))
1288+ require.Equal(t, tc.build, envelope.GetRegistered().GetBuild())
1289+ require.Equal(t, tc.retiring, envelope.GetRegistered().GetRetiring())
1290+ })
1291+ }
1292+ for _, tc := range []struct {
1293+ name string
1294+ server bool
1295+ }{{"client", false}, {"server", true}} {
1296+ t.Run("oversize-"+tc.name, func(t *testing.T) {
1297+ build := strings.Repeat("x", 129)
1298+ if tc.server {
1299+ _, err := EncodeServer(Registered{Epoch: 7, Connection: testConnectionID(1), Build: build}, testEnvelopeCeiling, testChunkCeiling)
1300+ require.ErrorIs(t, err, ErrInvalidMessage)
1301+ raw, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{Scope: scopeToWire(7, testConnectionID(1)), Build: build}}})
1302+ require.NoError(t, err)
1303+ _, err = DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1304+ require.ErrorIs(t, err, ErrInvalidMessage)
1305+ } else {
1306+ _, err := EncodeClient(Register{Build: build}, testEnvelopeCeiling, testChunkCeiling)
1307+ require.ErrorIs(t, err, ErrInvalidMessage)
1308+ raw, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: build}}})
1309+ require.NoError(t, err)
1310+ _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1311+ require.ErrorIs(t, err, ErrInvalidMessage)
1312+ }
1313+ })
1314+ }
1315+}
M internal/adapters/brokerwire/messages.go +3 -1
@@ -35,7 +35,7 @@ type ClientMessage interface{ brokerClientMessage() }
3535 type ServerMessage interface{ brokerServerMessage() }
3636
3737 // Register opens one client connection to the broker.
38-type Register struct{}
38+type Register struct{ Build string }
3939
4040 func (Register) brokerClientMessage() {}
4141
@@ -192,6 +192,8 @@ func (CancelPreview) brokerClientMessage() {}
192192 type Registered struct {
193193 Epoch ports.BrokerEpoch
194194 Connection ports.BrokerConnectionID
195+ Build string
196+ Retiring bool
195197 }
196198
197199 func (Registered) brokerServerMessage() {}
@@ -35,7 +35,7 @@ type ClientMessage interface{ brokerClientMessage() }
35 type ServerMessage interface{ brokerServerMessage() }35 type ServerMessage interface{ brokerServerMessage() }
36 36
37 // Register opens one client connection to the broker.37 // Register opens one client connection to the broker.
38-type Register struct{}38+type Register struct{ Build string }
39 39
40 func (Register) brokerClientMessage() {}40 func (Register) brokerClientMessage() {}
41 41
@@ -192,6 +192,8 @@ func (CancelPreview) brokerClientMessage() {}
192 type Registered struct {192 type Registered struct {
193 Epoch ports.BrokerEpoch193 Epoch ports.BrokerEpoch
194 Connection ports.BrokerConnectionID194 Connection ports.BrokerConnectionID
195+ Build string
196+ Retiring bool
195 }197 }
196 198
197 func (Registered) brokerServerMessage() {}199 func (Registered) brokerServerMessage() {}
M internal/protocol/wire/broker.pb.go +31 -5 generated file: collapsed
load diff
M internal/protocol/wire/schema/broker.proto +5 -1
@@ -110,7 +110,9 @@ message BrokerServerEnvelope {
110110 }
111111
112112 // Register opens one client connection to the broker.
113-message Register {}
113+message Register {
114+ string build = 1;
115+}
114116
115117 // Subscribe asks for snapshot and lifecycle publications for one scope.
116118 message Subscribe {
@@ -220,6 +222,8 @@ message CloseStream {
220222 // Registered confirms one accepted client connection and its assigned scope.
221223 message Registered {
222224 BrokerScope scope = 1;
225+ string build = 2;
226+ bool retiring = 3;
223227 }
224228
225229 // SnapshotPart is one fragment of a snapshot publication. Parts of one
@@ -110,7 +110,9 @@ message BrokerServerEnvelope {
110 }110 }
111 111
112 // Register opens one client connection to the broker.112 // Register opens one client connection to the broker.
113-message Register {}113+message Register {
114+ string build = 1;
115+}
114 116
115 // Subscribe asks for snapshot and lifecycle publications for one scope.117 // Subscribe asks for snapshot and lifecycle publications for one scope.
116 message Subscribe {118 message Subscribe {
@@ -220,6 +222,8 @@ message CloseStream {
220 // Registered confirms one accepted client connection and its assigned scope.222 // Registered confirms one accepted client connection and its assigned scope.
221 message Registered {223 message Registered {
222 BrokerScope scope = 1;224 BrokerScope scope = 1;
225+ string build = 2;
226+ bool retiring = 3;
223 }227 }
224 228
225 // SnapshotPart is one fragment of a snapshot publication. Parts of one229 // SnapshotPart is one fragment of a snapshot publication. Parts of one

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close