detached 1df3ba0

feat(brokeripc): retire an idle broker from another build

commit
1df3ba0f30b125f0c1ae9eb89a171044a3d73712
author
brice · · 2026-09-23 13:33 UTC
parent
bef813a
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
7 files · +238 -47
Split
UnifiedSplit
A internal/adapters/brokeripc/build.go +41 -0
@@ -0,0 +1,41 @@
1+package brokeripc
2+
3+import (
4+ "crypto/sha256"
5+ "encoding/hex"
6+ "io"
7+ "os"
8+ "runtime/debug"
9+ "sync"
10+)
11+
12+var defaultBuildIdentity = sync.OnceValue(func() string {
13+ info, ok := debug.ReadBuildInfo()
14+ version, revision, modified := "", "", false
15+ if ok {
16+ version = info.Main.Version
17+ for _, setting := range info.Settings {
18+ switch setting.Key {
19+ case "vcs.revision":
20+ revision = setting.Value
21+ case "vcs.modified":
22+ modified = setting.Value == "true"
23+ }
24+ }
25+ }
26+ if revision != "" && !modified {
27+ return version + "+" + revision
28+ }
29+ path, err := os.Executable()
30+ if err == nil {
31+ file, openErr := os.Open(path)
32+ if openErr == nil {
33+ defer file.Close()
34+ hash := sha256.New()
35+ if _, hashErr := io.Copy(hash, file); hashErr == nil {
36+ return "sha256:" + hex.EncodeToString(hash.Sum(nil)[:16])
37+ }
38+ }
39+ }
40+ return version + "+" + revision + "-dirty"
41+})
@@ -0,0 +1,41 @@
1+package brokeripc
2+
3+import (
4+ "crypto/sha256"
5+ "encoding/hex"
6+ "io"
7+ "os"
8+ "runtime/debug"
9+ "sync"
10+)
11+
12+var defaultBuildIdentity = sync.OnceValue(func() string {
13+ info, ok := debug.ReadBuildInfo()
14+ version, revision, modified := "", "", false
15+ if ok {
16+ version = info.Main.Version
17+ for _, setting := range info.Settings {
18+ switch setting.Key {
19+ case "vcs.revision":
20+ revision = setting.Value
21+ case "vcs.modified":
22+ modified = setting.Value == "true"
23+ }
24+ }
25+ }
26+ if revision != "" && !modified {
27+ return version + "+" + revision
28+ }
29+ path, err := os.Executable()
30+ if err == nil {
31+ file, openErr := os.Open(path)
32+ if openErr == nil {
33+ defer file.Close()
34+ hash := sha256.New()
35+ if _, hashErr := io.Copy(hash, file); hashErr == nil {
36+ return "sha256:" + hex.EncodeToString(hash.Sum(nil)[:16])
37+ }
38+ }
39+ }
40+ return version + "+" + revision + "-dirty"
41+})
A internal/adapters/brokeripc/build_test.go +97 -0
@@ -0,0 +1,97 @@
1+package brokeripc
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/adapters/brokerwire"
9+ "github.com/bnema/vev/internal/adapters/ipc"
10+ "github.com/bnema/vev/internal/ports"
11+ "github.com/stretchr/testify/require"
12+)
13+
14+func TestBuildRegistration(t *testing.T) {
15+ for _, tc := range []struct {
16+ name string
17+ server, client string
18+ livePeer bool
19+ retire bool
20+ stale string
21+ }{
22+ {name: "same build", server: "build-a", client: "build-a"},
23+ {name: "sole different build", server: "build-a", client: "build-b", retire: true},
24+ {name: "different build with live peer", server: "build-a", client: "build-b", livePeer: true, stale: "build-a"},
25+ } {
26+ t.Run(tc.name, func(t *testing.T) {
27+ retired := make(chan struct{}, 1)
28+ e := startEndpoint(t, Config{Build: tc.server, OnRetire: func() { retired <- struct{}{} }})
29+ if tc.livePeer {
30+ peer, err := Dial(context.Background(), e.path, Config{Build: tc.server})
31+ require.NoError(t, err)
32+ defer peer.Close()
33+ }
34+ service, err := Dial(context.Background(), e.path, Config{Build: tc.client})
35+ if tc.retire {
36+ require.ErrorIs(t, err, ErrBrokerRetired)
37+ select {
38+ case <-retired:
39+ case <-time.After(5 * time.Second):
40+ t.Fatal("retire callback not called")
41+ }
42+ _, err = Dial(context.Background(), e.path, Config{Build: tc.client})
43+ require.Error(t, err)
44+ return
45+ }
46+ require.NoError(t, err)
47+ defer service.Close()
48+ stale, ok := StaleBuild(service)
49+ require.Equal(t, tc.stale != "", ok)
50+ require.Equal(t, tc.stale, stale)
51+ select {
52+ case <-retired:
53+ t.Fatal("unexpected retirement")
54+ default:
55+ }
56+ })
57+ }
58+}
59+
60+func TestRetiringRefusesConcurrentAdmission(t *testing.T) {
61+ const epoch = ports.BrokerEpoch(0x51)
62+ path := testSocketPath(t)
63+ gated := &gatedAuthority{
64+ inner: newFakeAuthority(epoch, ports.BrokerSnapshot{}),
65+ entered: make(chan struct{}), gate: make(chan struct{}),
66+ }
67+ bound, err := listen(path, epoch, gated, Config{Build: "build-a"}, ipc.SameUserPeerVerifier())
68+ require.NoError(t, err)
69+ defer bound.Close()
70+ l := bound.(*listener)
71+ raw := rawDial(t, &endpoint{t: t, path: path, epoch: epoch}, brokerwire.DefaultCeilings())
72+ select {
73+ case <-gated.entered:
74+ case <-time.After(5 * time.Second):
75+ t.Fatal("admission not reached")
76+ }
77+ // The registration decision and the admission gate share l.mu. A pending
78+ // carriage prevents retirement; once retiring, the in-flight admission
79+ // cannot publish a new session even if its authority returns success.
80+ l.mu.Lock()
81+ require.Len(t, l.pending, 1)
82+ l.mu.Unlock()
83+ // Commit retirement under the listener lock after admission has entered
84+ // the authority, then release the authority gate.
85+ l.mu.Lock()
86+ l.retiring = true
87+ l.mu.Unlock()
88+ close(gated.gate)
89+ result := make(chan error, 1)
90+ go func() { _, err := raw.transport.RecvBounded(raw.ceilings.MaxReceiveEnvelopeBytes); result <- err }()
91+ select {
92+ case err := <-result:
93+ require.Error(t, err)
94+ case <-time.After(5 * time.Second):
95+ t.Fatal("retiring listener admitted a pending session")
96+ }
97+}
@@ -0,0 +1,97 @@
1+package brokeripc
2+
3+import (
4+ "context"
5+ "testing"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/adapters/brokerwire"
9+ "github.com/bnema/vev/internal/adapters/ipc"
10+ "github.com/bnema/vev/internal/ports"
11+ "github.com/stretchr/testify/require"
12+)
13+
14+func TestBuildRegistration(t *testing.T) {
15+ for _, tc := range []struct {
16+ name string
17+ server, client string
18+ livePeer bool
19+ retire bool
20+ stale string
21+ }{
22+ {name: "same build", server: "build-a", client: "build-a"},
23+ {name: "sole different build", server: "build-a", client: "build-b", retire: true},
24+ {name: "different build with live peer", server: "build-a", client: "build-b", livePeer: true, stale: "build-a"},
25+ } {
26+ t.Run(tc.name, func(t *testing.T) {
27+ retired := make(chan struct{}, 1)
28+ e := startEndpoint(t, Config{Build: tc.server, OnRetire: func() { retired <- struct{}{} }})
29+ if tc.livePeer {
30+ peer, err := Dial(context.Background(), e.path, Config{Build: tc.server})
31+ require.NoError(t, err)
32+ defer peer.Close()
33+ }
34+ service, err := Dial(context.Background(), e.path, Config{Build: tc.client})
35+ if tc.retire {
36+ require.ErrorIs(t, err, ErrBrokerRetired)
37+ select {
38+ case <-retired:
39+ case <-time.After(5 * time.Second):
40+ t.Fatal("retire callback not called")
41+ }
42+ _, err = Dial(context.Background(), e.path, Config{Build: tc.client})
43+ require.Error(t, err)
44+ return
45+ }
46+ require.NoError(t, err)
47+ defer service.Close()
48+ stale, ok := StaleBuild(service)
49+ require.Equal(t, tc.stale != "", ok)
50+ require.Equal(t, tc.stale, stale)
51+ select {
52+ case <-retired:
53+ t.Fatal("unexpected retirement")
54+ default:
55+ }
56+ })
57+ }
58+}
59+
60+func TestRetiringRefusesConcurrentAdmission(t *testing.T) {
61+ const epoch = ports.BrokerEpoch(0x51)
62+ path := testSocketPath(t)
63+ gated := &gatedAuthority{
64+ inner: newFakeAuthority(epoch, ports.BrokerSnapshot{}),
65+ entered: make(chan struct{}), gate: make(chan struct{}),
66+ }
67+ bound, err := listen(path, epoch, gated, Config{Build: "build-a"}, ipc.SameUserPeerVerifier())
68+ require.NoError(t, err)
69+ defer bound.Close()
70+ l := bound.(*listener)
71+ raw := rawDial(t, &endpoint{t: t, path: path, epoch: epoch}, brokerwire.DefaultCeilings())
72+ select {
73+ case <-gated.entered:
74+ case <-time.After(5 * time.Second):
75+ t.Fatal("admission not reached")
76+ }
77+ // The registration decision and the admission gate share l.mu. A pending
78+ // carriage prevents retirement; once retiring, the in-flight admission
79+ // cannot publish a new session even if its authority returns success.
80+ l.mu.Lock()
81+ require.Len(t, l.pending, 1)
82+ l.mu.Unlock()
83+ // Commit retirement under the listener lock after admission has entered
84+ // the authority, then release the authority gate.
85+ l.mu.Lock()
86+ l.retiring = true
87+ l.mu.Unlock()
88+ close(gated.gate)
89+ result := make(chan error, 1)
90+ go func() { _, err := raw.transport.RecvBounded(raw.ceilings.MaxReceiveEnvelopeBytes); result <- err }()
91+ select {
92+ case err := <-result:
93+ require.Error(t, err)
94+ case <-time.After(5 * time.Second):
95+ t.Fatal("retiring listener admitted a pending session")
96+ }
97+}
M internal/adapters/brokeripc/client.go +53 -34
@@ -90,7 +90,7 @@ func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier)
9090 _ = transport.Close()
9191 return nil, err
9292 }
93- scope, err := register(setup, transport, ceilings)
93+ scope, staleBuild, err := register(setup, transport, ceilings, cfg.Build)
9494 if err != nil {
9595 _ = transport.Close()
9696 return nil, err
@@ -114,17 +114,18 @@ func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier)
114114 }
115115 ctx, cancelSession := context.WithCancel(context.Background())
116116 c := &client{
117- transport: transport,
118- ceilings: ceilings,
119- scope: scope,
120- conn: conn,
121- cfg: cfg,
122- ctx: ctx,
123- cancel: cancelSession,
124- pending: make(map[ports.BrokerOperationID]chan operationResult),
125- kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),
126- streams: make(map[ports.BrokerStreamID]*clientStream),
127- done: make(chan struct{}),
117+ transport: transport,
118+ ceilings: ceilings,
119+ scope: scope,
120+ staleBuild: staleBuild,
121+ conn: conn,
122+ cfg: cfg,
123+ ctx: ctx,
124+ cancel: cancelSession,
125+ pending: make(map[ports.BrokerOperationID]chan operationResult),
126+ kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),
127+ streams: make(map[ports.BrokerStreamID]*clientStream),
128+ done: make(chan struct{}),
128129 }
129130 go c.run()
130131 return c, nil
@@ -138,58 +139,66 @@ func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier)
138139 // interrupted by closing the carriage, and the exchange worker is joined before
139140 // returning, so no setup goroutine is abandoned on the carriage. A Registered
140141 // observed after ctx expired is not a success.
141-func register(ctx context.Context, transport wire.BoundedTransport, ceilings brokerwire.Ceilings) (brokerwire.Scope, error) {
142+func register(ctx context.Context, transport wire.BoundedTransport, ceilings brokerwire.Ceilings, build string) (brokerwire.Scope, string, error) {
142143 type result struct {
143- scope brokerwire.Scope
144- err error
144+ scope brokerwire.Scope
145+ staleBuild string
146+ err error
145147 }
146148 done := make(chan result, 1)
147149 go func() {
148- scope, err := exchangeRegister(transport, ceilings)
149- done <- result{scope: scope, err: err}
150+ scope, staleBuild, err := exchangeRegister(transport, ceilings, build)
151+ done <- result{scope: scope, staleBuild: staleBuild, err: err}
150152 }()
151153 select {
152154 case <-ctx.Done():
153155 _ = transport.Close()
154156 <-done
155- return brokerwire.Scope{}, ctx.Err()
157+ return brokerwire.Scope{}, "", ctx.Err()
156158 case outcome := <-done:
157159 if outcome.err == nil {
158160 if err := ctx.Err(); err != nil {
159- return brokerwire.Scope{}, err
161+ return brokerwire.Scope{}, "", err
160162 }
161163 }
162- return outcome.scope, outcome.err
164+ return outcome.scope, outcome.staleBuild, outcome.err
163165 }
164166 }
165167
166168 // exchangeRegister runs one Register/Registered round trip on the carriage. It
167169 // is synchronous: the caller bounds it by closing the carriage on ctx expiry.
168-func exchangeRegister(transport wire.BoundedTransport, ceilings brokerwire.Ceilings) (brokerwire.Scope, error) {
169- payload, err := brokerwire.EncodeClient(brokerwire.Register{}, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
170+func exchangeRegister(transport wire.BoundedTransport, ceilings brokerwire.Ceilings, build string) (brokerwire.Scope, string, error) {
171+ payload, err := brokerwire.EncodeClient(brokerwire.Register{Build: build}, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
170172 if err != nil {
171- return brokerwire.Scope{}, err
173+ return brokerwire.Scope{}, "", err
172174 }
173175 if err := transport.Send(wire.Envelope{Payload: payload}); err != nil {
174- return brokerwire.Scope{}, transportFailure(err)
176+ return brokerwire.Scope{}, "", transportFailure(err)
175177 }
176178 envelope, err := transport.RecvBounded(ceilings.MaxReceiveEnvelopeBytes)
177179 if err != nil {
178- return brokerwire.Scope{}, transportFailure(err)
180+ return brokerwire.Scope{}, "", transportFailure(err)
179181 }
180182 message, err := brokerwire.DecodeServer(envelope.Payload, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
181183 if err != nil {
182- return brokerwire.Scope{}, errors.Join(ErrMalformedFrame, err)
184+ return brokerwire.Scope{}, "", errors.Join(ErrMalformedFrame, err)
183185 }
184186 registered, ok := message.(brokerwire.Registered)
185187 if !ok {
186- return brokerwire.Scope{}, errors.Join(ErrProtocol, errors.New("brokeripc: broker did not answer Register with Registered"))
188+ return brokerwire.Scope{}, "", errors.Join(ErrProtocol, errors.New("brokeripc: broker did not answer Register with Registered"))
187189 }
188190 scope := brokerwire.Scope{Epoch: registered.Epoch, Connection: registered.Connection}
189191 if err := scope.Validate(); err != nil {
190- return brokerwire.Scope{}, errors.Join(ErrProtocol, err)
192+ return brokerwire.Scope{}, "", errors.Join(ErrProtocol, err)
191193 }
192- return scope, nil
194+ if registered.Retiring {
195+ _ = transport.Close()
196+ return brokerwire.Scope{}, "", ErrBrokerRetired
197+ }
198+ if registered.Build != build {
199+ return scope, registered.Build, nil
200+ }
201+ return scope, "", nil
193202 }
194203
195204 // operationResult is one reply to a mutating operation.
@@ -202,11 +211,12 @@ type operationResult struct {
202211
203212 // client implements ports.BrokerService over one broker IPC connection.
204213 type client struct {
205- transport wire.BoundedTransport
206- ceilings brokerwire.Ceilings
207- scope brokerwire.Scope
208- conn *brokerwire.Connection
209- cfg Config
214+ transport wire.BoundedTransport
215+ ceilings brokerwire.Ceilings
216+ scope brokerwire.Scope
217+ staleBuild string
218+ conn *brokerwire.Connection
219+ cfg Config
210220
211221 ctx context.Context
212222 cancel context.CancelFunc
@@ -1042,3 +1052,12 @@ func (s *clientSubscription) notify() {
10421052 default:
10431053 }
10441054 }
1055+
1056+// StaleBuild reports the identity of a connected broker from another build.
1057+func StaleBuild(service ports.BrokerService) (string, bool) {
1058+ c, ok := service.(*client)
1059+ if !ok || c == nil || c.staleBuild == "" {
1060+ return "", false
1061+ }
1062+ return c.staleBuild, true
1063+}
@@ -90,7 +90,7 @@ func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier)
90 _ = transport.Close()90 _ = transport.Close()
91 return nil, err91 return nil, err
92 }92 }
93- scope, err := register(setup, transport, ceilings)93+ scope, staleBuild, err := register(setup, transport, ceilings, cfg.Build)
94 if err != nil {94 if err != nil {
95 _ = transport.Close()95 _ = transport.Close()
96 return nil, err96 return nil, err
@@ -114,17 +114,18 @@ func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier)
114 }114 }
115 ctx, cancelSession := context.WithCancel(context.Background())115 ctx, cancelSession := context.WithCancel(context.Background())
116 c := &client{116 c := &client{
117- transport: transport,117+ transport: transport,
118- ceilings: ceilings,118+ ceilings: ceilings,
119- scope: scope,119+ scope: scope,
120- conn: conn,120+ staleBuild: staleBuild,
121- cfg: cfg,121+ conn: conn,
122- ctx: ctx,122+ cfg: cfg,
123- cancel: cancelSession,123+ ctx: ctx,
124- pending: make(map[ports.BrokerOperationID]chan operationResult),124+ cancel: cancelSession,
125- kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),125+ pending: make(map[ports.BrokerOperationID]chan operationResult),
126- streams: make(map[ports.BrokerStreamID]*clientStream),126+ kinds: make(map[ports.BrokerOperationID]brokerwire.RegisterMutationKind),
127- done: make(chan struct{}),127+ streams: make(map[ports.BrokerStreamID]*clientStream),
128+ done: make(chan struct{}),
128 }129 }
129 go c.run()130 go c.run()
130 return c, nil131 return c, nil
@@ -138,58 +139,66 @@ func dial(ctx context.Context, path string, cfg Config, verify ipc.PeerVerifier)
138 // interrupted by closing the carriage, and the exchange worker is joined before139 // interrupted by closing the carriage, and the exchange worker is joined before
139 // returning, so no setup goroutine is abandoned on the carriage. A Registered140 // returning, so no setup goroutine is abandoned on the carriage. A Registered
140 // observed after ctx expired is not a success.141 // observed after ctx expired is not a success.
141-func register(ctx context.Context, transport wire.BoundedTransport, ceilings brokerwire.Ceilings) (brokerwire.Scope, error) {142+func register(ctx context.Context, transport wire.BoundedTransport, ceilings brokerwire.Ceilings, build string) (brokerwire.Scope, string, error) {
142 type result struct {143 type result struct {
143- scope brokerwire.Scope144+ scope brokerwire.Scope
144- err error145+ staleBuild string
146+ err error
145 }147 }
146 done := make(chan result, 1)148 done := make(chan result, 1)
147 go func() {149 go func() {
148- scope, err := exchangeRegister(transport, ceilings)150+ scope, staleBuild, err := exchangeRegister(transport, ceilings, build)
149- done <- result{scope: scope, err: err}151+ done <- result{scope: scope, staleBuild: staleBuild, err: err}
150 }()152 }()
151 select {153 select {
152 case <-ctx.Done():154 case <-ctx.Done():
153 _ = transport.Close()155 _ = transport.Close()
154 <-done156 <-done
155- return brokerwire.Scope{}, ctx.Err()157+ return brokerwire.Scope{}, "", ctx.Err()
156 case outcome := <-done:158 case outcome := <-done:
157 if outcome.err == nil {159 if outcome.err == nil {
158 if err := ctx.Err(); err != nil {160 if err := ctx.Err(); err != nil {
159- return brokerwire.Scope{}, err161+ return brokerwire.Scope{}, "", err
160 }162 }
161 }163 }
162- return outcome.scope, outcome.err164+ return outcome.scope, outcome.staleBuild, outcome.err
163 }165 }
164 }166 }
165 167
166 // exchangeRegister runs one Register/Registered round trip on the carriage. It168 // exchangeRegister runs one Register/Registered round trip on the carriage. It
167 // is synchronous: the caller bounds it by closing the carriage on ctx expiry.169 // is synchronous: the caller bounds it by closing the carriage on ctx expiry.
168-func exchangeRegister(transport wire.BoundedTransport, ceilings brokerwire.Ceilings) (brokerwire.Scope, error) {170+func exchangeRegister(transport wire.BoundedTransport, ceilings brokerwire.Ceilings, build string) (brokerwire.Scope, string, error) {
169- payload, err := brokerwire.EncodeClient(brokerwire.Register{}, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)171+ payload, err := brokerwire.EncodeClient(brokerwire.Register{Build: build}, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
170 if err != nil {172 if err != nil {
171- return brokerwire.Scope{}, err173+ return brokerwire.Scope{}, "", err
172 }174 }
173 if err := transport.Send(wire.Envelope{Payload: payload}); err != nil {175 if err := transport.Send(wire.Envelope{Payload: payload}); err != nil {
174- return brokerwire.Scope{}, transportFailure(err)176+ return brokerwire.Scope{}, "", transportFailure(err)
175 }177 }
176 envelope, err := transport.RecvBounded(ceilings.MaxReceiveEnvelopeBytes)178 envelope, err := transport.RecvBounded(ceilings.MaxReceiveEnvelopeBytes)
177 if err != nil {179 if err != nil {
178- return brokerwire.Scope{}, transportFailure(err)180+ return brokerwire.Scope{}, "", transportFailure(err)
179 }181 }
180 message, err := brokerwire.DecodeServer(envelope.Payload, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)182 message, err := brokerwire.DecodeServer(envelope.Payload, ceilings.MaxReceiveEnvelopeBytes, ceilings.StreamChunkLimit)
181 if err != nil {183 if err != nil {
182- return brokerwire.Scope{}, errors.Join(ErrMalformedFrame, err)184+ return brokerwire.Scope{}, "", errors.Join(ErrMalformedFrame, err)
183 }185 }
184 registered, ok := message.(brokerwire.Registered)186 registered, ok := message.(brokerwire.Registered)
185 if !ok {187 if !ok {
186- return brokerwire.Scope{}, errors.Join(ErrProtocol, errors.New("brokeripc: broker did not answer Register with Registered"))188+ return brokerwire.Scope{}, "", errors.Join(ErrProtocol, errors.New("brokeripc: broker did not answer Register with Registered"))
187 }189 }
188 scope := brokerwire.Scope{Epoch: registered.Epoch, Connection: registered.Connection}190 scope := brokerwire.Scope{Epoch: registered.Epoch, Connection: registered.Connection}
189 if err := scope.Validate(); err != nil {191 if err := scope.Validate(); err != nil {
190- return brokerwire.Scope{}, errors.Join(ErrProtocol, err)192+ return brokerwire.Scope{}, "", errors.Join(ErrProtocol, err)
191 }193 }
192- return scope, nil194+ if registered.Retiring {
195+ _ = transport.Close()
196+ return brokerwire.Scope{}, "", ErrBrokerRetired
197+ }
198+ if registered.Build != build {
199+ return scope, registered.Build, nil
200+ }
201+ return scope, "", nil
193 }202 }
194 203
195 // operationResult is one reply to a mutating operation.204 // operationResult is one reply to a mutating operation.
@@ -202,11 +211,12 @@ type operationResult struct {
202 211
203 // client implements ports.BrokerService over one broker IPC connection.212 // client implements ports.BrokerService over one broker IPC connection.
204 type client struct {213 type client struct {
205- transport wire.BoundedTransport214+ transport wire.BoundedTransport
206- ceilings brokerwire.Ceilings215+ ceilings brokerwire.Ceilings
207- scope brokerwire.Scope216+ scope brokerwire.Scope
208- conn *brokerwire.Connection217+ staleBuild string
209- cfg Config218+ conn *brokerwire.Connection
219+ cfg Config
210 220
211 ctx context.Context221 ctx context.Context
212 cancel context.CancelFunc222 cancel context.CancelFunc
@@ -1042,3 +1052,12 @@ func (s *clientSubscription) notify() {
1042 default:1052 default:
1043 }1053 }
1044 }1054 }
1055+
1056+// StaleBuild reports the identity of a connected broker from another build.
1057+func StaleBuild(service ports.BrokerService) (string, bool) {
1058+ c, ok := service.(*client)
1059+ if !ok || c == nil || c.staleBuild == "" {
1060+ return "", false
1061+ }
1062+ return c.staleBuild, true
1063+}
M internal/adapters/brokeripc/config.go +5 -0
@@ -50,10 +50,15 @@ type Config struct {
5050 // connection as a read-only reader: it receives publications but never
5151 // counts as observation demand, so a CLI read never triggers a probe.
5252 PassiveSubscribe bool
53+ Build string
54+ OnRetire func()
5355 }
5456
5557 // withDefaults fills every unset bound with its package default.
5658 func (c Config) withDefaults() Config {
59+ if c.Build == "" {
60+ c.Build = defaultBuildIdentity()
61+ }
5762 if c.MaxClients <= 0 {
5863 c.MaxClients = DefaultMaxClients
5964 }
@@ -50,10 +50,15 @@ type Config struct {
50 // connection as a read-only reader: it receives publications but never50 // connection as a read-only reader: it receives publications but never
51 // counts as observation demand, so a CLI read never triggers a probe.51 // counts as observation demand, so a CLI read never triggers a probe.
52 PassiveSubscribe bool52 PassiveSubscribe bool
53+ Build string
54+ OnRetire func()
53 }55 }
54 56
55 // withDefaults fills every unset bound with its package default.57 // withDefaults fills every unset bound with its package default.
56 func (c Config) withDefaults() Config {58 func (c Config) withDefaults() Config {
59+ if c.Build == "" {
60+ c.Build = defaultBuildIdentity()
61+ }
57 if c.MaxClients <= 0 {62 if c.MaxClients <= 0 {
58 c.MaxClients = DefaultMaxClients63 c.MaxClients = DefaultMaxClients
59 }64 }
M internal/adapters/brokeripc/errors.go +2 -1
@@ -15,7 +15,8 @@ import (
1515 // them.
1616 var (
1717 // ErrConfig reports an invalid listener or dialer configuration.
18- ErrConfig = errors.New("brokeripc: invalid configuration")
18+ ErrConfig = errors.New("brokeripc: invalid configuration")
19+ ErrBrokerRetired = errors.New("brokeripc: stale broker is retiring for a newer build")
1920 // ErrListenerClosed reports Accept on a closed listener.
2021 ErrListenerClosed = errors.New("brokeripc: listener is closed")
2122 // ErrSessionClosed reports session work presented after the session reached
@@ -15,7 +15,8 @@ import (
15 // them.15 // them.
16 var (16 var (
17 // ErrConfig reports an invalid listener or dialer configuration.17 // ErrConfig reports an invalid listener or dialer configuration.
18- ErrConfig = errors.New("brokeripc: invalid configuration")18+ ErrConfig = errors.New("brokeripc: invalid configuration")
19+ ErrBrokerRetired = errors.New("brokeripc: stale broker is retiring for a newer build")
19 // ErrListenerClosed reports Accept on a closed listener.20 // ErrListenerClosed reports Accept on a closed listener.
20 ErrListenerClosed = errors.New("brokeripc: listener is closed")21 ErrListenerClosed = errors.New("brokeripc: listener is closed")
21 // ErrSessionClosed reports session work presented after the session reached22 // ErrSessionClosed reports session work presented after the session reached
M internal/adapters/brokeripc/protocol_test.go +1 -1
@@ -53,7 +53,7 @@ func (r *rawCarriage) recv(t *testing.T) brokerwire.ServerMessage {
5353 // register completes the registration exchange and returns the assigned scope.
5454 func (r *rawCarriage) register(t *testing.T) brokerwire.Scope {
5555 t.Helper()
56- r.send(t, brokerwire.Register{})
56+ r.send(t, brokerwire.Register{Build: defaultBuildIdentity()})
5757 message := r.recv(t)
5858 registered, ok := message.(brokerwire.Registered)
5959 require.True(t, ok, "the first server frame after Register must be Registered")
@@ -53,7 +53,7 @@ func (r *rawCarriage) recv(t *testing.T) brokerwire.ServerMessage {
53 // register completes the registration exchange and returns the assigned scope.53 // register completes the registration exchange and returns the assigned scope.
54 func (r *rawCarriage) register(t *testing.T) brokerwire.Scope {54 func (r *rawCarriage) register(t *testing.T) brokerwire.Scope {
55 t.Helper()55 t.Helper()
56- r.send(t, brokerwire.Register{})56+ r.send(t, brokerwire.Register{Build: defaultBuildIdentity()})
57 message := r.recv(t)57 message := r.recv(t)
58 registered, ok := message.(brokerwire.Registered)58 registered, ok := message.(brokerwire.Registered)
59 require.True(t, ok, "the first server frame after Register must be Registered")59 require.True(t, ok, "the first server frame after Register must be Registered")
M internal/adapters/brokeripc/server.go +39 -11
@@ -120,6 +120,7 @@ type listener struct {
120120
121121 mu sync.Mutex
122122 closed bool
123+ retiring bool
123124 sessions map[*serverSession]struct{}
124125 pending map[wire.BoundedTransport]struct{}
125126 lastRefuse error
@@ -168,6 +169,7 @@ func (l *listener) acceptLoop() {
168169 l.failAccept(err)
169170 return
170171 }
172+ l.trackPending(transport)
171173 l.wg.Add(1)
172174 go func() {
173175 defer l.wg.Done()
@@ -214,7 +216,6 @@ func (l *listener) confirmRefusal(err error) {
214216 // and publishes the running session. Slot ownership transfers to the session on
215217 // success.
216218 func (l *listener) admit(transport wire.BoundedTransport, release func()) {
217- l.trackPending(transport)
218219 session, err := l.startSession(transport, release)
219220 l.untrackPending(transport)
220221 if err != nil {
@@ -351,8 +352,10 @@ func (l *listener) startSession(transport wire.BoundedTransport, release func())
351352 // The session deregisters itself once it reaches its terminal state, so a
352353 // long-lived listener never accumulates retired connections.
353354 session.onShutdown = func() { l.forgetSession(session) }
355+ session.retire = func() bool { return l.tryRetire(session) }
356+ session.closeListenerMux = func() { _ = l.mux.Close() }
354357 l.mu.Lock()
355- if l.closed {
358+ if l.closed || l.retiring {
356359 l.mu.Unlock()
357360 // The run loop was never started, so Close (which waits on the run
358361 // loop's done channel) would wait forever. Tear the unstarted session
@@ -431,6 +434,19 @@ func (l *listener) forgetSession(session *serverSession) {
431434 l.mu.Unlock()
432435 }
433436
437+func (l *listener) tryRetire(session *serverSession) bool {
438+ l.mu.Lock()
439+ defer l.mu.Unlock()
440+ if l.closed || l.retiring || len(l.pending) != 0 || len(l.sessions) != 1 {
441+ return false
442+ }
443+ if _, ok := l.sessions[session]; !ok {
444+ return false
445+ }
446+ l.retiring = true
447+ return true
448+}
449+
434450 func (l *listener) isClosed() bool {
435451 if l == nil {
436452 return true
@@ -447,14 +463,16 @@ func (l *listener) isClosed() bool {
447463 // state machine, the outbound frame path, one snapshot publisher, and the
448464 // bridged logical streams for exactly one client.
449465 type serverSession struct {
450- epoch ports.BrokerEpoch
451- scope brokerwire.Scope
452- conn *brokerwire.Connection
453- core ports.BrokerService
454- transport wire.BoundedTransport
455- ceilings brokerwire.Ceilings
456- cfg Config
457- release func()
466+ epoch ports.BrokerEpoch
467+ scope brokerwire.Scope
468+ conn *brokerwire.Connection
469+ core ports.BrokerService
470+ transport wire.BoundedTransport
471+ ceilings brokerwire.Ceilings
472+ cfg Config
473+ release func()
474+ retire func() bool
475+ closeListenerMux func()
458476
459477 ctx context.Context
460478 cancel context.CancelFunc
@@ -656,7 +674,17 @@ func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
656674 // the pre-Register guard before answering, so this healthy connection is
657675 // no longer bounded by the setup deadline.
658676 s.stopRegistration()
659- return s.send(brokerwire.Registered{Epoch: s.epoch, Connection: s.scope.Connection})
677+ retired := m.Build != s.cfg.Build && s.retire != nil && s.retire()
678+ err := s.send(brokerwire.Registered{Epoch: s.epoch, Connection: s.scope.Connection, Build: s.cfg.Build, Retiring: retired})
679+ if retired {
680+ go func() {
681+ s.closeListenerMux()
682+ if s.cfg.OnRetire != nil {
683+ s.cfg.OnRetire()
684+ }
685+ }()
686+ }
687+ return err
660688 case brokerwire.Subscribe:
661689 if !s.scopeMatches(m.Epoch, m.Connection) {
662690 return s.refuseScope()
@@ -120,6 +120,7 @@ type listener struct {
120 120
121 mu sync.Mutex121 mu sync.Mutex
122 closed bool122 closed bool
123+ retiring bool
123 sessions map[*serverSession]struct{}124 sessions map[*serverSession]struct{}
124 pending map[wire.BoundedTransport]struct{}125 pending map[wire.BoundedTransport]struct{}
125 lastRefuse error126 lastRefuse error
@@ -168,6 +169,7 @@ func (l *listener) acceptLoop() {
168 l.failAccept(err)169 l.failAccept(err)
169 return170 return
170 }171 }
172+ l.trackPending(transport)
171 l.wg.Add(1)173 l.wg.Add(1)
172 go func() {174 go func() {
173 defer l.wg.Done()175 defer l.wg.Done()
@@ -214,7 +216,6 @@ func (l *listener) confirmRefusal(err error) {
214 // and publishes the running session. Slot ownership transfers to the session on216 // and publishes the running session. Slot ownership transfers to the session on
215 // success.217 // success.
216 func (l *listener) admit(transport wire.BoundedTransport, release func()) {218 func (l *listener) admit(transport wire.BoundedTransport, release func()) {
217- l.trackPending(transport)
218 session, err := l.startSession(transport, release)219 session, err := l.startSession(transport, release)
219 l.untrackPending(transport)220 l.untrackPending(transport)
220 if err != nil {221 if err != nil {
@@ -351,8 +352,10 @@ func (l *listener) startSession(transport wire.BoundedTransport, release func())
351 // The session deregisters itself once it reaches its terminal state, so a352 // The session deregisters itself once it reaches its terminal state, so a
352 // long-lived listener never accumulates retired connections.353 // long-lived listener never accumulates retired connections.
353 session.onShutdown = func() { l.forgetSession(session) }354 session.onShutdown = func() { l.forgetSession(session) }
355+ session.retire = func() bool { return l.tryRetire(session) }
356+ session.closeListenerMux = func() { _ = l.mux.Close() }
354 l.mu.Lock()357 l.mu.Lock()
355- if l.closed {358+ if l.closed || l.retiring {
356 l.mu.Unlock()359 l.mu.Unlock()
357 // The run loop was never started, so Close (which waits on the run360 // The run loop was never started, so Close (which waits on the run
358 // loop's done channel) would wait forever. Tear the unstarted session361 // loop's done channel) would wait forever. Tear the unstarted session
@@ -431,6 +434,19 @@ func (l *listener) forgetSession(session *serverSession) {
431 l.mu.Unlock()434 l.mu.Unlock()
432 }435 }
433 436
437+func (l *listener) tryRetire(session *serverSession) bool {
438+ l.mu.Lock()
439+ defer l.mu.Unlock()
440+ if l.closed || l.retiring || len(l.pending) != 0 || len(l.sessions) != 1 {
441+ return false
442+ }
443+ if _, ok := l.sessions[session]; !ok {
444+ return false
445+ }
446+ l.retiring = true
447+ return true
448+}
449+
434 func (l *listener) isClosed() bool {450 func (l *listener) isClosed() bool {
435 if l == nil {451 if l == nil {
436 return true452 return true
@@ -447,14 +463,16 @@ func (l *listener) isClosed() bool {
447 // state machine, the outbound frame path, one snapshot publisher, and the463 // state machine, the outbound frame path, one snapshot publisher, and the
448 // bridged logical streams for exactly one client.464 // bridged logical streams for exactly one client.
449 type serverSession struct {465 type serverSession struct {
450- epoch ports.BrokerEpoch466+ epoch ports.BrokerEpoch
451- scope brokerwire.Scope467+ scope brokerwire.Scope
452- conn *brokerwire.Connection468+ conn *brokerwire.Connection
453- core ports.BrokerService469+ core ports.BrokerService
454- transport wire.BoundedTransport470+ transport wire.BoundedTransport
455- ceilings brokerwire.Ceilings471+ ceilings brokerwire.Ceilings
456- cfg Config472+ cfg Config
457- release func()473+ release func()
474+ retire func() bool
475+ closeListenerMux func()
458 476
459 ctx context.Context477 ctx context.Context
460 cancel context.CancelFunc478 cancel context.CancelFunc
@@ -656,7 +674,17 @@ func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
656 // the pre-Register guard before answering, so this healthy connection is674 // the pre-Register guard before answering, so this healthy connection is
657 // no longer bounded by the setup deadline.675 // no longer bounded by the setup deadline.
658 s.stopRegistration()676 s.stopRegistration()
659- return s.send(brokerwire.Registered{Epoch: s.epoch, Connection: s.scope.Connection})677+ retired := m.Build != s.cfg.Build && s.retire != nil && s.retire()
678+ err := s.send(brokerwire.Registered{Epoch: s.epoch, Connection: s.scope.Connection, Build: s.cfg.Build, Retiring: retired})
679+ if retired {
680+ go func() {
681+ s.closeListenerMux()
682+ if s.cfg.OnRetire != nil {
683+ s.cfg.OnRetire()
684+ }
685+ }()
686+ }
687+ return err
660 case brokerwire.Subscribe:688 case brokerwire.Subscribe:
661 if !s.scopeMatches(m.Epoch, m.Connection) {689 if !s.scopeMatches(m.Epoch, m.Connection) {
662 return s.refuseScope()690 return s.refuseScope()

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