1package ports
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "path/filepath"
8 "strings"
9 "unicode"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/protocol/catalogue"
13)
14
15// Durable broker store sentinels. Adapters wrap these so a caller can classify
16// a store failure without importing the adapter: conflicting authority means
17// refresh the token and retry, a locked store means another owner is live, and
18// invalid state is resolved offline from the recovery record. None of them is
19// ever retried blindly inside the store.
20var (
21 // ErrBrokerMembershipImmutable reports that a registry was not explicitly
22 // configured to accept membership mutations.
23 ErrBrokerMembershipImmutable = errors.New("ports: broker membership is immutable")
24 // ErrBrokerHostConflict reports that durable host authority changed under
25 // the caller's expected revision, or that the supplied membership would
26 // regress or re-trust an existing registration without a fresh identity.
27 ErrBrokerHostConflict = errors.New("ports: broker durable host authority conflict")
28 // ErrBrokerStoreLocked reports that another owner already holds the durable
29 // store's lifetime lock.
30 ErrBrokerStoreLocked = errors.New("ports: broker durable store is owned elsewhere")
31 // ErrBrokerStoreInvalidState reports durable state that fails validation.
32 // Such state is never silently replaced or rolled back by the store.
33 ErrBrokerStoreInvalidState = errors.New("ports: broker durable store state is invalid")
34)
35
36// BrokerStoreOutcomeUnknownError reports a write failure after the durable
37// commit point. Callers must reopen and reload authority rather than retrying.
38type BrokerStoreOutcomeUnknownError struct{ Err error }
39
40func (e BrokerStoreOutcomeUnknownError) Error() string {
41 if e.Err == nil {
42 return "ports: broker store outcome unknown"
43 }
44 return "ports: broker store outcome unknown: " + e.Err.Error()
45}
46func (e BrokerStoreOutcomeUnknownError) Unwrap() error { return e.Err }
47
48type BrokerRouteKind string
49
50const (
51 BrokerRouteUnix BrokerRouteKind = "unix"
52 BrokerRouteSSHStdio BrokerRouteKind = "ssh-stdio"
53 BrokerRouteSSHQUIC BrokerRouteKind = "ssh-quic"
54 BrokerMaxRouteArgv = 32
55 BrokerMaxRouteArgBytes = 4 << 10
56)
57
58// BrokerRouteSpec is durable routing authority. Adapters derive all runtime
59// trust, timeout, address and launch details from policy and this minimal route.
60type BrokerRouteSpec struct {
61 Kind BrokerRouteKind
62 Path string
63 Target string
64 Argv []string
65}
66
67func (r BrokerRouteSpec) Validate() error {
68 switch r.Kind {
69 case BrokerRouteUnix:
70 if r.Path == "" || !filepath.IsAbs(r.Path) || filepath.Clean(r.Path) != r.Path || r.Target != "" || len(r.Argv) != 0 {
71 return errors.New("ports: invalid unix broker route")
72 }
73 case BrokerRouteSSHStdio, BrokerRouteSSHQUIC:
74 if r.Path != "" || domain.ValidateRemoteHostTarget(r.Target) != nil || len(r.Argv) == 0 || len(r.Argv) > BrokerMaxRouteArgv {
75 return errors.New("ports: invalid ssh broker route")
76 }
77 for _, arg := range r.Argv {
78 if arg == "" || len(arg) > BrokerMaxRouteArgBytes || strings.IndexFunc(arg, func(v rune) bool { return v == 0 || unicode.IsControl(v) }) >= 0 {
79 return fmt.Errorf("ports: invalid ssh broker route argv")
80 }
81 }
82 default:
83 return errors.New("ports: unknown broker route kind")
84 }
85 return nil
86}
87
88func (r BrokerRouteSpec) Clone() BrokerRouteSpec {
89 r.Argv = append([]string(nil), r.Argv...)
90 return r
91}
92
93// BrokerRouteForTransport maps the closed legacy policy transport vocabulary
94// to canonical remote routes.
95func BrokerRouteForTransport(transport, endpoint string) (BrokerRouteSpec, error) {
96 switch transport {
97 case "ssh-stdio", "stdio":
98 return BrokerRouteSpec{Kind: BrokerRouteSSHStdio, Target: endpoint, Argv: []string{"vev", "_broker-mux-stdio", "--production"}}, nil
99 case "ssh-quic", "quic":
100 return BrokerRouteSpec{Kind: BrokerRouteSSHQUIC, Target: endpoint, Argv: []string{"vev", "_broker-mux-quic-bootstrap", "--production"}}, nil
101 default:
102 return BrokerRouteSpec{}, fmt.Errorf("ports: unsupported broker transport %q", transport)
103 }
104}
105
106func CloneBrokerHostRecords(hosts []BrokerHostRecord) []BrokerHostRecord {
107 out := append([]BrokerHostRecord(nil), hosts...)
108 for i := range out {
109 out[i].Route = out[i].Route.Clone()
110 }
111 return out
112}
113
114// UpgradeBrokerHostRoutes upgrades records written before routes became
115// explicit. It is intentionally closed over the approved remote vocabulary.
116func UpgradeBrokerHostRoutes(hosts []BrokerHostRecord) ([]BrokerHostRecord, error) {
117 out := CloneBrokerHostRecords(hosts)
118 for i := range out {
119 if out[i].Route.Kind != "" {
120 continue
121 }
122 route, err := BrokerRouteForTransport(out[i].Policy.Transport, out[i].Registration.Endpoint)
123 if err != nil {
124 return nil, fmt.Errorf("ports: upgrade broker route for %q: %w", out[i].Registration.Endpoint, err)
125 }
126 out[i].Route = route
127 }
128 return out, nil
129}
130
131// BrokerHostRecord is durable registration authority. Membership and policy
132// are never inferred from an observation or from the broker environment.
133type BrokerHostRecord struct {
134 Registration domain.RemoteRegistration
135 Pinned bool
136 Learned bool
137 Policy BrokerPolicy
138 Route BrokerRouteSpec
139 // Identity is the durable authenticated daemon binding. Empty means the
140 // registration has not completed first contact yet.
141 Identity BrokerDaemonIdentity
142}
143
144// BrokerHosts is an optimistic-concurrency token and the ordered authority.
145// Revision is store-local and independent of broker publication epochs.
146type BrokerHosts struct {
147 Revision uint64
148 Hosts []BrokerHostRecord
149}
150
151func (h BrokerHosts) Validate() error {
152 if h.Revision == 0 {
153 return errors.New("ports: invalid broker host state")
154 }
155 return ValidateBrokerHostRecords(h.Hosts)
156}
157
158// ValidateBrokerHostRecords reports whether caller-supplied membership is
159// internally valid, independently of any revision token: every registration
160// and policy is valid, every record is anchored as pinned or learned, no
161// endpoint repeats, and the set is within the host bound. A durable store
162// applies this rule to caller input before its CAS, so malformed membership is
163// a caller error rather than durable-state corruption.
164func ValidateBrokerHostRecords(hosts []BrokerHostRecord) error {
165 if len(hosts) > BrokerMaxHosts {
166 return errors.New("ports: invalid broker host state")
167 }
168 seen := make(map[string]bool, len(hosts))
169 for _, host := range hosts {
170 if err := host.Registration.Validate(); err != nil {
171 return err
172 }
173 if err := host.Policy.Validate(); err != nil {
174 return err
175 }
176 if err := host.Route.Validate(); err != nil {
177 return err
178 }
179 if host.Identity != "" {
180 if err := host.Identity.Validate(); err != nil {
181 return err
182 }
183 }
184 if (!host.Pinned && !host.Learned) || seen[host.Registration.Endpoint] {
185 return errors.New("ports: invalid broker membership")
186 }
187 seen[host.Registration.Endpoint] = true
188 }
189 return nil
190}
191
192// BrokerHostStore owns durable membership separately from advisory snapshots.
193// ReplaceHosts compares expected with the current host revision, increments it,
194// and atomically discards observations for removed/replaced registrations or
195// changed policies. Callers supply fresh incarnations on remove/re-add.
196// ReplaceHosts validates the supplied membership (ValidateBrokerHostRecords)
197// before its compare-and-swap, so malformed caller input is reported as a
198// caller error and never as invalid durable state. Close releases exclusive
199// ownership; no operation is valid after Close.
200// BrokerHostAuthorityReader exposes committed durable routing authority.
201// Returned records must be defensive deep copies.
202type BrokerHostAuthorityReader interface {
203 LookupHost(ctx context.Context, endpoint string) (BrokerHostRecord, bool, error)
204}
205
206type BrokerHostStore interface {
207 BrokerSnapshotStore
208 LoadHosts() (BrokerHosts, error)
209 ReplaceHosts(expected uint64, hosts []BrokerHostRecord) error
210 Close() error
211}
212
213// BrokerHostStore is a BrokerSnapshotStore: the exclusive owner of durable
214// membership also owns the advisory snapshot, so the registry writes through
215// one seam. This guard keeps that composition compile-checked.
216var _ BrokerSnapshotStore = (BrokerHostStore)(nil)
217
218// ValidateDurableHostProjection reports whether one daemon observation
219// satisfies the rules the offline store and wire accept: a local observation
220// is never durable, availability and failure kinds stay in their closed
221// ranges, and a present session inventory is a valid catalogue cache entry
222// against the exact registration incarnation. The offline store and the
223// registry both apply this rule, so a projection the registry publishes is
224// never rejected (and therefore silently never persisted) by the store.
225// Transient state (Checking and the live failure cause) is deliberately not
226// part of the durable rule.
227func ValidateDurableHostProjection(obs BrokerDaemonObservation) error {
228 if obs.Local {
229 return errors.New("ports: local daemon observation is never durable")
230 }
231 if obs.Availability < domain.RemoteAvailabilityUnknown || obs.Availability > domain.RemoteAvailabilityNoDaemon {
232 return errors.New("ports: broker host availability is out of range")
233 }
234 if obs.LastFailure.Kind > domain.RemoteFailureInvalidResponse {
235 return errors.New("ports: broker host failure kind is out of range")
236 }
237 // Observed identity is part of the durable rule, not only of the live one: a
238 // restored record with partial identity or without the protocol version its
239 // identity was authenticated with would otherwise reach a publication the
240 // wire refuses, taking down every client connection instead of failing
241 // closed here.
242 if err := validateObservedDaemonState(obs); err != nil {
243 return err
244 }
245 if len(obs.Sessions) == 0 {
246 return nil
247 }
248 return catalogue.ValidateRemoteCatalogCacheEntries([]catalogue.RemoteCatalogCacheEntry{{
249 Host: obs.Endpoint,
250 Incarnation: obs.Registration.Incarnation,
251 FetchedAt: obs.LastSuccess,
252 Sessions: obs.Sessions,
253 }})
254}
255
256// ValidateDurableObservation checks the shared remote-only durable observation shape.
257func ValidateDurableObservation(daemon BrokerDaemonObservation) error {
258 if daemon.Local {
259 return errors.New("local daemon observation is never durable")
260 }
261 if err := domain.ValidateRemoteHostTarget(daemon.Endpoint); err != nil {
262 return err
263 }
264 if err := daemon.Registration.Validate(); err != nil {
265 return err
266 }
267 if daemon.Endpoint != daemon.Registration.Endpoint {
268 return errors.New("observation endpoint does not match registration")
269 }
270 if len(daemon.Sessions) > 0 && !daemon.InventoryKnown {
271 return errors.New("observation carries sessions without known inventory")
272 }
273 return ValidateDurableHostProjection(daemon)
274}
275
276// DurableProjection returns the persistable copy of a published snapshot. It
277// is the single rule for what is transient, shared by the registry writer and
278// every store:
279//
280// - Removed tombstones are process-local fencing; a restart never adopts them.
281// - Checking and the live failure cause are in-flight observation state.
282// - Policy is membership authority, not observation: the loader re-stamps it
283// from the matching BrokerHostRecord, so persisting it would create a
284// second, stale authority.
285// - Tab Attention is a transient bell that must not resurrect on reload.
286//
287// Local observations are kept so ValidateDurableSnapshot still refuses them;
288// a producer that may hold one drops it first (see WithoutLocal).
289func (s BrokerSnapshot) DurableProjection() BrokerSnapshot {
290 out := s.Clone()
291 out.Removed = nil
292 for i := range out.Daemons {
293 daemon := &out.Daemons[i]
294 daemon.Checking = false
295 daemon.LastFailure.Err = nil
296 daemon.Policy = BrokerPolicy{}
297 for j := range daemon.Sessions {
298 for k := range daemon.Sessions[j].Tabs {
299 daemon.Sessions[j].Tabs[k].Attention = false
300 }
301 }
302 }
303 return out
304}
305
306// WithoutLocal returns s without the process-local daemon observation. The
307// daemon list is new, but nested data stays shared with s; chain
308// DurableProjection (or Clone) before mutating the result.
309func (s BrokerSnapshot) WithoutLocal() BrokerSnapshot {
310 out := s
311 daemons := make([]BrokerDaemonObservation, 0, len(s.Daemons))
312 for _, daemon := range s.Daemons {
313 if !daemon.Local {
314 daemons = append(daemons, daemon)
315 }
316 }
317 out.Daemons = daemons
318 return out
319}
320
321// ValidateDurableSnapshot checks the shared durable snapshot shape.
322func ValidateDurableSnapshot(snapshot BrokerSnapshot) error {
323 if snapshot.Epoch == 0 {
324 if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
325 return errors.New("invalid empty snapshot")
326 }
327 return nil
328 }
329 if snapshot.Revision == 0 {
330 return errors.New("snapshot has no revision")
331 }
332 if len(snapshot.Daemons) > BrokerMaxDaemonsPerSnapshot {
333 return errors.New("snapshot has too many daemons")
334 }
335 seen := make(map[string]struct{}, len(snapshot.Daemons))
336 for _, daemon := range snapshot.Daemons {
337 if err := ValidateDurableObservation(daemon); err != nil {
338 return err
339 }
340 if _, duplicate := seen[daemon.Endpoint]; duplicate {
341 return errors.New("snapshot has duplicate host")
342 }
343 seen[daemon.Endpoint] = struct{}{}
344 }
345 if len(snapshot.Removed) > BrokerMaxTombstones {
346 return errors.New("snapshot has too many tombstones")
347 }
348 for _, tombstone := range snapshot.Removed {
349 if err := tombstone.Validate(); err != nil {
350 return err
351 }
352 if _, live := seen[tombstone.Endpoint]; live {
353 return errors.New("snapshot carries a live host as tombstone")
354 }
355 }
356 return nil
357}