detached b3c9196

fix(broker): upgrade pinned host protocol versions on startup (#289)

Pinned host records stored the protocol and catalogue versions of the
build that pinned them. After a protocol bump the unchanged remote daemon
identity was refused as a conflicting policy, and the picker reported it
as an SSH connection failure.

The broker now moves stored host policies to the running build's versions
at startup, advancing the registration generation, and classifies
conflicting-policy and incompatible probe errors as incompatible.
commit
b3c91968797e62720050f4d703fbb553e360f2f7
author
bnema · · 2026-09-25 20:53 UTC
committer
GitHub · · 2026-09-25 20:53 UTC
parent
def4c6e
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
5 files · +113 -0
Unified
UnifiedSplit
A internal/app/broker_host_upgrade_test.go +67 -0
@@ -0,0 +1,67 @@
1+package app
2+
3+import (
4+ "path/filepath"
5+ "testing"
6+
7+ "github.com/stretchr/testify/require"
8+
9+ "github.com/bnema/vev/internal/adapters/brokerstore"
10+ "github.com/bnema/vev/internal/domain"
11+ "github.com/bnema/vev/internal/ports"
12+ "github.com/bnema/vev/internal/protocol"
13+ "github.com/bnema/vev/internal/protocol/catalogue"
14+)
15+
16+func TestUpgradeBrokerHostVersions(t *testing.T) {
17+ tests := []struct {
18+ name string
19+ policy func() ports.BrokerPolicy
20+ wantRevision uint64 // also the expected generation advance
21+ }{
22+ {
23+ name: "older build policy is upgraded and identity kept",
24+ policy: func() ports.BrokerPolicy {
25+ p := remoteBrokerPolicy("")
26+ p.ProtocolVersion--
27+ return p
28+ },
29+ wantRevision: 1,
30+ },
31+ {
32+ name: "current policy is left untouched",
33+ policy: func() ports.BrokerPolicy { return remoteBrokerPolicy("") },
34+ wantRevision: 0,
35+ },
36+ }
37+ for _, tt := range tests {
38+ t.Run(tt.name, func(t *testing.T) {
39+ registration, err := domain.NewRemoteRegistration("user@host", [16]byte{1})
40+ require.NoError(t, err)
41+ policy := tt.policy()
42+ route, err := ports.BrokerRouteForTransport(policy.Transport, registration.Endpoint)
43+ require.NoError(t, err)
44+ record := ports.BrokerHostRecord{Registration: registration, Pinned: true, Policy: policy, Route: route, Identity: "daemon-identity"}
45+
46+ store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(shortTempDir(t, "vevu"), "state"), InitialHosts: []ports.BrokerHostRecord{record}, InitialImportProvided: true})
47+ require.NoError(t, err)
48+ t.Cleanup(func() { require.NoError(t, store.Close()) })
49+ before, err := store.LoadHosts()
50+ require.NoError(t, err)
51+
52+ require.NoError(t, upgradeBrokerHostVersions(store))
53+
54+ after, err := store.LoadHosts()
55+ require.NoError(t, err)
56+ require.Equal(t, before.Revision+tt.wantRevision, after.Revision)
57+ require.Len(t, after.Hosts, 1)
58+ got := after.Hosts[0]
59+ require.Equal(t, protocol.Version, got.Policy.ProtocolVersion)
60+ require.Equal(t, catalogue.RemoteCatalogSchemaVersion, got.Policy.CatalogSchemaVersion)
61+ require.Equal(t, ports.BrokerDaemonIdentity("daemon-identity"), got.Identity)
62+ require.Equal(t, registration.Endpoint, got.Registration.Endpoint)
63+ require.Equal(t, registration.Incarnation, got.Registration.Incarnation)
64+ require.Equal(t, registration.Generation+domain.RemoteGeneration(tt.wantRevision), got.Registration.Generation)
65+ })
66+ }
67+}
@@ -0,0 +1,67 @@
1+package app
2+
3+import (
4+ "path/filepath"
5+ "testing"
6+
7+ "github.com/stretchr/testify/require"
8+
9+ "github.com/bnema/vev/internal/adapters/brokerstore"
10+ "github.com/bnema/vev/internal/domain"
11+ "github.com/bnema/vev/internal/ports"
12+ "github.com/bnema/vev/internal/protocol"
13+ "github.com/bnema/vev/internal/protocol/catalogue"
14+)
15+
16+func TestUpgradeBrokerHostVersions(t *testing.T) {
17+ tests := []struct {
18+ name string
19+ policy func() ports.BrokerPolicy
20+ wantRevision uint64 // also the expected generation advance
21+ }{
22+ {
23+ name: "older build policy is upgraded and identity kept",
24+ policy: func() ports.BrokerPolicy {
25+ p := remoteBrokerPolicy("")
26+ p.ProtocolVersion--
27+ return p
28+ },
29+ wantRevision: 1,
30+ },
31+ {
32+ name: "current policy is left untouched",
33+ policy: func() ports.BrokerPolicy { return remoteBrokerPolicy("") },
34+ wantRevision: 0,
35+ },
36+ }
37+ for _, tt := range tests {
38+ t.Run(tt.name, func(t *testing.T) {
39+ registration, err := domain.NewRemoteRegistration("user@host", [16]byte{1})
40+ require.NoError(t, err)
41+ policy := tt.policy()
42+ route, err := ports.BrokerRouteForTransport(policy.Transport, registration.Endpoint)
43+ require.NoError(t, err)
44+ record := ports.BrokerHostRecord{Registration: registration, Pinned: true, Policy: policy, Route: route, Identity: "daemon-identity"}
45+
46+ store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(shortTempDir(t, "vevu"), "state"), InitialHosts: []ports.BrokerHostRecord{record}, InitialImportProvided: true})
47+ require.NoError(t, err)
48+ t.Cleanup(func() { require.NoError(t, store.Close()) })
49+ before, err := store.LoadHosts()
50+ require.NoError(t, err)
51+
52+ require.NoError(t, upgradeBrokerHostVersions(store))
53+
54+ after, err := store.LoadHosts()
55+ require.NoError(t, err)
56+ require.Equal(t, before.Revision+tt.wantRevision, after.Revision)
57+ require.Len(t, after.Hosts, 1)
58+ got := after.Hosts[0]
59+ require.Equal(t, protocol.Version, got.Policy.ProtocolVersion)
60+ require.Equal(t, catalogue.RemoteCatalogSchemaVersion, got.Policy.CatalogSchemaVersion)
61+ require.Equal(t, ports.BrokerDaemonIdentity("daemon-identity"), got.Identity)
62+ require.Equal(t, registration.Endpoint, got.Registration.Endpoint)
63+ require.Equal(t, registration.Incarnation, got.Registration.Incarnation)
64+ require.Equal(t, registration.Generation+domain.RemoteGeneration(tt.wantRevision), got.Registration.Generation)
65+ })
66+ }
67+}
M internal/app/broker_offline.go +3 -0
@@ -270,6 +270,9 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
270270 return fmt.Errorf("vev: open broker sandbox store: %w", err)
271271 }
272272 defer func() { retErr = errors.Join(retErr, store.Close()) }()
273+ if err := upgradeBrokerHostVersions(store); err != nil {
274+ return err
275+ }
273276
274277 epoch, err := deps.newEpoch()
275278 if err != nil {
@@ -270,6 +270,9 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
270 return fmt.Errorf("vev: open broker sandbox store: %w", err)270 return fmt.Errorf("vev: open broker sandbox store: %w", err)
271 }271 }
272 defer func() { retErr = errors.Join(retErr, store.Close()) }()272 defer func() { retErr = errors.Join(retErr, store.Close()) }()
273+ if err := upgradeBrokerHostVersions(store); err != nil {
274+ return err
275+ }
273 276
274 epoch, err := deps.newEpoch()277 epoch, err := deps.newEpoch()
275 if err != nil {278 if err != nil {
M internal/app/broker_production.go +35 -0
@@ -8,6 +8,7 @@ import (
88
99 "github.com/bnema/vev/internal/adapters/brokerconfig"
1010 "github.com/bnema/vev/internal/adapters/ipc"
11+ "github.com/bnema/vev/internal/domain"
1112 "github.com/bnema/vev/internal/platform"
1213 "github.com/bnema/vev/internal/ports"
1314 "github.com/bnema/vev/internal/protocol"
@@ -62,6 +63,40 @@ func remoteBrokerPolicy(transport string) ports.BrokerPolicy {
6263 }
6364 }
6465
66+// upgradeBrokerHostVersions moves durable host policies to this build's
67+// protocol and catalogue versions. Those fields describe the running binary,
68+// not a user choice, so hosts pinned by an older build must follow an upgrade;
69+// otherwise the unchanged daemon identity is refused as a conflicting policy.
70+func upgradeBrokerHostVersions(store ports.BrokerHostStore) error {
71+ hosts, err := store.LoadHosts()
72+ if err != nil {
73+ return fmt.Errorf("vev: load broker hosts: %w", err)
74+ }
75+ changed := false
76+ records := ports.CloneBrokerHostRecords(hosts.Hosts)
77+ for i := range records {
78+ policy := &records[i].Policy
79+ if policy.ProtocolVersion == protocol.Version && policy.CatalogSchemaVersion == catalogue.RemoteCatalogSchemaVersion {
80+ continue
81+ }
82+ // A policy change advances the generation so stale observations are fenced.
83+ if records[i].Registration.Generation == ^domain.RemoteGeneration(0) {
84+ return ports.ErrBrokerRevisionExhausted
85+ }
86+ records[i].Registration.Generation++
87+ policy.ProtocolVersion = protocol.Version
88+ policy.CatalogSchemaVersion = catalogue.RemoteCatalogSchemaVersion
89+ changed = true
90+ }
91+ if !changed {
92+ return nil
93+ }
94+ if err := store.ReplaceHosts(hosts.Revision, records); err != nil {
95+ return fmt.Errorf("vev: upgrade broker host versions: %w", err)
96+ }
97+ return nil
98+}
99+
65100 func localDaemonPolicy() ports.BrokerPolicy {
66101 return ports.BrokerPolicy{
67102 ProtocolVersion: protocol.Version, CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
@@ -8,6 +8,7 @@ import (
8 8
9 "github.com/bnema/vev/internal/adapters/brokerconfig"9 "github.com/bnema/vev/internal/adapters/brokerconfig"
10 "github.com/bnema/vev/internal/adapters/ipc"10 "github.com/bnema/vev/internal/adapters/ipc"
11+ "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/platform"12 "github.com/bnema/vev/internal/platform"
12 "github.com/bnema/vev/internal/ports"13 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"14 "github.com/bnema/vev/internal/protocol"
@@ -62,6 +63,40 @@ func remoteBrokerPolicy(transport string) ports.BrokerPolicy {
62 }63 }
63 }64 }
64 65
66+// upgradeBrokerHostVersions moves durable host policies to this build's
67+// protocol and catalogue versions. Those fields describe the running binary,
68+// not a user choice, so hosts pinned by an older build must follow an upgrade;
69+// otherwise the unchanged daemon identity is refused as a conflicting policy.
70+func upgradeBrokerHostVersions(store ports.BrokerHostStore) error {
71+ hosts, err := store.LoadHosts()
72+ if err != nil {
73+ return fmt.Errorf("vev: load broker hosts: %w", err)
74+ }
75+ changed := false
76+ records := ports.CloneBrokerHostRecords(hosts.Hosts)
77+ for i := range records {
78+ policy := &records[i].Policy
79+ if policy.ProtocolVersion == protocol.Version && policy.CatalogSchemaVersion == catalogue.RemoteCatalogSchemaVersion {
80+ continue
81+ }
82+ // A policy change advances the generation so stale observations are fenced.
83+ if records[i].Registration.Generation == ^domain.RemoteGeneration(0) {
84+ return ports.ErrBrokerRevisionExhausted
85+ }
86+ records[i].Registration.Generation++
87+ policy.ProtocolVersion = protocol.Version
88+ policy.CatalogSchemaVersion = catalogue.RemoteCatalogSchemaVersion
89+ changed = true
90+ }
91+ if !changed {
92+ return nil
93+ }
94+ if err := store.ReplaceHosts(hosts.Revision, records); err != nil {
95+ return fmt.Errorf("vev: upgrade broker host versions: %w", err)
96+ }
97+ return nil
98+}
99+
65 func localDaemonPolicy() ports.BrokerPolicy {100 func localDaemonPolicy() ports.BrokerPolicy {
66 return ports.BrokerPolicy{101 return ports.BrokerPolicy{
67 ProtocolVersion: protocol.Version, CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,102 ProtocolVersion: protocol.Version, CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
M internal/usecase/broker/registry.go +4 -0
@@ -1128,10 +1128,14 @@ func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.Re
11281128 if result.err != nil {
11291129 kind := domain.RemoteFailureTransport
11301130 var typed domain.RemoteFailure
1131+ var brokerErr ports.BrokerError
11311132 switch {
11321133 case errors.As(result.err, &typed) && typed.Kind != domain.RemoteFailureNone:
11331134 // The dialer classified the failure (for example SSH authentication).
11341135 kind = typed.Kind
1136+ case errors.As(result.err, &brokerErr) && (brokerErr.Code == ports.BrokerErrorConflictingPolicy || brokerErr.Code == ports.BrokerErrorIncompatible):
1137+ // SSH worked; the daemon's identity or version was refused.
1138+ kind = domain.RemoteFailureIncompatible
11351139 case errors.Is(result.err, context.DeadlineExceeded):
11361140 kind = domain.RemoteFailureTimeout
11371141 }
@@ -1128,10 +1128,14 @@ func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.Re
1128 if result.err != nil {1128 if result.err != nil {
1129 kind := domain.RemoteFailureTransport1129 kind := domain.RemoteFailureTransport
1130 var typed domain.RemoteFailure1130 var typed domain.RemoteFailure
1131+ var brokerErr ports.BrokerError
1131 switch {1132 switch {
1132 case errors.As(result.err, &typed) && typed.Kind != domain.RemoteFailureNone:1133 case errors.As(result.err, &typed) && typed.Kind != domain.RemoteFailureNone:
1133 // The dialer classified the failure (for example SSH authentication).1134 // The dialer classified the failure (for example SSH authentication).
1134 kind = typed.Kind1135 kind = typed.Kind
1136+ case errors.As(result.err, &brokerErr) && (brokerErr.Code == ports.BrokerErrorConflictingPolicy || brokerErr.Code == ports.BrokerErrorIncompatible):
1137+ // SSH worked; the daemon's identity or version was refused.
1138+ kind = domain.RemoteFailureIncompatible
1135 case errors.Is(result.err, context.DeadlineExceeded):1139 case errors.Is(result.err, context.DeadlineExceeded):
1136 kind = domain.RemoteFailureTimeout1140 kind = domain.RemoteFailureTimeout
1137 }1141 }
M internal/usecase/broker/registry_outcome_test.go +4 -0
@@ -9,6 +9,7 @@ import (
99 "github.com/stretchr/testify/require"
1010
1111 "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
1213 )
1314
1415 func TestObservationOutcomeKeepsTypedDialFailure(t *testing.T) {
@@ -22,6 +23,9 @@ func TestObservationOutcomeKeepsTypedDialFailure(t *testing.T) {
2223 {name: "deadline is timeout", err: context.DeadlineExceeded, kind: domain.RemoteFailureTimeout, want: domain.RemoteAvailabilityUnreachable},
2324 {name: "wrapped authentication", err: fmt.Errorf("dial: %w", domain.RemoteFailure{Kind: domain.RemoteFailureAuthentication, Err: errors.New("exit 255")}), kind: domain.RemoteFailureAuthentication, want: domain.RemoteAvailabilityAuthFailed},
2425 {name: "wrapped trust", err: fmt.Errorf("dial: %w", domain.RemoteFailure{Kind: domain.RemoteFailureTrust}), kind: domain.RemoteFailureTrust, want: domain.RemoteAvailabilityUnreachable},
26+ {name: "conflicting policy is incompatible", err: fmt.Errorf("bind: %w", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}), kind: domain.RemoteFailureIncompatible, want: domain.RemoteAvailabilityIncompatible},
27+ {name: "incompatible broker error", err: ports.BrokerError{Code: ports.BrokerErrorIncompatible}, kind: domain.RemoteFailureIncompatible, want: domain.RemoteAvailabilityIncompatible},
28+ {name: "unavailable broker error stays transport", err: ports.BrokerError{Code: ports.BrokerErrorUnavailable}, kind: domain.RemoteFailureTransport, want: domain.RemoteAvailabilityUnreachable},
2529 }
2630 for _, tt := range tests {
2731 t.Run(tt.name, func(t *testing.T) {
@@ -9,6 +9,7 @@ import (
9 "github.com/stretchr/testify/require"9 "github.com/stretchr/testify/require"
10 10
11 "github.com/bnema/vev/internal/domain"11 "github.com/bnema/vev/internal/domain"
12+ "github.com/bnema/vev/internal/ports"
12 )13 )
13 14
14 func TestObservationOutcomeKeepsTypedDialFailure(t *testing.T) {15 func TestObservationOutcomeKeepsTypedDialFailure(t *testing.T) {
@@ -22,6 +23,9 @@ func TestObservationOutcomeKeepsTypedDialFailure(t *testing.T) {
22 {name: "deadline is timeout", err: context.DeadlineExceeded, kind: domain.RemoteFailureTimeout, want: domain.RemoteAvailabilityUnreachable},23 {name: "deadline is timeout", err: context.DeadlineExceeded, kind: domain.RemoteFailureTimeout, want: domain.RemoteAvailabilityUnreachable},
23 {name: "wrapped authentication", err: fmt.Errorf("dial: %w", domain.RemoteFailure{Kind: domain.RemoteFailureAuthentication, Err: errors.New("exit 255")}), kind: domain.RemoteFailureAuthentication, want: domain.RemoteAvailabilityAuthFailed},24 {name: "wrapped authentication", err: fmt.Errorf("dial: %w", domain.RemoteFailure{Kind: domain.RemoteFailureAuthentication, Err: errors.New("exit 255")}), kind: domain.RemoteFailureAuthentication, want: domain.RemoteAvailabilityAuthFailed},
24 {name: "wrapped trust", err: fmt.Errorf("dial: %w", domain.RemoteFailure{Kind: domain.RemoteFailureTrust}), kind: domain.RemoteFailureTrust, want: domain.RemoteAvailabilityUnreachable},25 {name: "wrapped trust", err: fmt.Errorf("dial: %w", domain.RemoteFailure{Kind: domain.RemoteFailureTrust}), kind: domain.RemoteFailureTrust, want: domain.RemoteAvailabilityUnreachable},
26+ {name: "conflicting policy is incompatible", err: fmt.Errorf("bind: %w", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}), kind: domain.RemoteFailureIncompatible, want: domain.RemoteAvailabilityIncompatible},
27+ {name: "incompatible broker error", err: ports.BrokerError{Code: ports.BrokerErrorIncompatible}, kind: domain.RemoteFailureIncompatible, want: domain.RemoteAvailabilityIncompatible},
28+ {name: "unavailable broker error stays transport", err: ports.BrokerError{Code: ports.BrokerErrorUnavailable}, kind: domain.RemoteFailureTransport, want: domain.RemoteAvailabilityUnreachable},
25 }29 }
26 for _, tt := range tests {30 for _, tt := range tests {
27 t.Run(tt.name, func(t *testing.T) {31 t.Run(tt.name, func(t *testing.T) {

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