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// Remote helper argv vocabulary. Durable SSH routes store this argv verbatim,
94// so these spellings are part of the persisted route contract; the CLI parses
95// the same constants.
96const (
97 BrokerMuxStdioCommand = "_broker-mux-stdio"
98 BrokerMuxQUICBootstrapCommand = "_broker-mux-quic-bootstrap"
99 BrokerRouteScopeArg = "--production"
100)
101
102// BrokerRouteForTransport maps the closed policy transport vocabulary to the
103// canonical remote route.
104func BrokerRouteForTransport(transport, endpoint string) (BrokerRouteSpec, error) {
105 switch transport {
106 case "ssh-stdio", "stdio":
107 return BrokerRouteSpec{Kind: BrokerRouteSSHStdio, Target: endpoint, Argv: []string{"vev", BrokerMuxStdioCommand, BrokerRouteScopeArg}}, nil
108 case "ssh-quic", "quic":
109 return BrokerRouteSpec{Kind: BrokerRouteSSHQUIC, Target: endpoint, Argv: []string{"vev", BrokerMuxQUICBootstrapCommand, BrokerRouteScopeArg}}, nil
110 default:
111 return BrokerRouteSpec{}, fmt.Errorf("ports: unsupported broker transport %q", transport)
112 }
113}
114
115func CloneBrokerHostRecords(hosts []BrokerHostRecord) []BrokerHostRecord {
116 out := append([]BrokerHostRecord(nil), hosts...)
117 for i := range out {
118 out[i].Route = out[i].Route.Clone()
119 }
120 return out
121}
122
123// UpgradeBrokerHostRoutes upgrades records written before routes became
124// explicit. It is intentionally closed over the approved remote vocabulary.
125func UpgradeBrokerHostRoutes(hosts []BrokerHostRecord) ([]BrokerHostRecord, error) {
126 out := CloneBrokerHostRecords(hosts)
127 for i := range out {
128 if out[i].Route.Kind != "" {
129 continue
130 }
131 route, err := BrokerRouteForTransport(out[i].Policy.Transport, out[i].Registration.Endpoint)
132 if err != nil {
133 return nil, fmt.Errorf("ports: upgrade broker route for %q: %w", out[i].Registration.Endpoint, err)
134 }
135 out[i].Route = route
136 }
137 return out, nil
138}
139
140// BrokerHostRecord is durable registration authority. Membership and policy
141// are never inferred from an observation or from the broker environment.
142type BrokerHostRecord struct {
143 Registration domain.RemoteRegistration
144 Pinned bool
145 Learned bool
146 Policy BrokerPolicy
147 Route BrokerRouteSpec
148 // Identity is the durable authenticated daemon binding. Empty means the
149 // registration has not completed first contact yet.
150 Identity BrokerDaemonIdentity
151}
152
153// BrokerHosts is an optimistic-concurrency token and the ordered authority.
154// Revision is store-local and independent of broker publication epochs.
155type BrokerHosts struct {
156 Revision uint64
157 Hosts []BrokerHostRecord
158}
159
160func (h BrokerHosts) Validate() error {
161 if h.Revision == 0 {
162 return errors.New("ports: invalid broker host state")
163 }
164 return ValidateBrokerHostRecords(h.Hosts)
165}
166
167// ValidateBrokerHostRecords reports whether caller-supplied membership is
168// internally valid, independently of any revision token: every registration
169// and policy is valid, every record is anchored as pinned or learned, no
170// endpoint repeats, and the set is within the host bound. A durable store
171// applies this rule to caller input before its CAS, so malformed membership is
172// a caller error rather than durable-state corruption.
173func ValidateBrokerHostRecords(hosts []BrokerHostRecord) error {
174 if len(hosts) > BrokerMaxHosts {
175 return errors.New("ports: invalid broker host state")
176 }
177 seen := make(map[string]bool, len(hosts))
178 for _, host := range hosts {
179 if err := host.Registration.Validate(); err != nil {
180 return err
181 }
182 if err := host.Policy.Validate(); err != nil {
183 return err
184 }
185 if err := host.Route.Validate(); err != nil {
186 return err
187 }
188 if host.Identity != "" {
189 if err := host.Identity.Validate(); err != nil {
190 return err
191 }
192 }
193 if (!host.Pinned && !host.Learned) || seen[host.Registration.Endpoint] {
194 return errors.New("ports: invalid broker membership")
195 }
196 seen[host.Registration.Endpoint] = true
197 }
198 return nil
199}
200
201// BrokerHostStore owns durable membership separately from advisory snapshots.
202// ReplaceHosts compares expected with the current host revision, increments it,
203// and atomically discards observations for removed/replaced registrations or
204// changed policies. Callers supply fresh incarnations on remove/re-add.
205// ReplaceHosts validates the supplied membership (ValidateBrokerHostRecords)
206// before its compare-and-swap, so malformed caller input is reported as a
207// caller error and never as invalid durable state. Close releases exclusive
208// ownership; no operation is valid after Close.
209// BrokerHostAuthorityReader exposes committed durable routing authority.
210// Returned records must be defensive deep copies.
211type BrokerHostAuthorityReader interface {
212 LookupHost(ctx context.Context, endpoint string) (BrokerHostRecord, bool, error)
213}
214
215type BrokerHostStore interface {
216 BrokerSnapshotStore
217 LoadHosts() (BrokerHosts, error)
218 ReplaceHosts(expected uint64, hosts []BrokerHostRecord) error
219 Close() error
220}
221
222// BrokerHostStore is a BrokerSnapshotStore: the exclusive owner of durable
223// membership also owns the advisory snapshot, so the registry writes through
224// one seam. This guard keeps that composition compile-checked.
225var _ BrokerSnapshotStore = (BrokerHostStore)(nil)
226
227// ValidateDurableHostProjection reports whether one daemon observation
228// satisfies the rules the offline store and wire accept: a local observation
229// is never durable, availability and failure kinds stay in their closed
230// ranges, and a present session inventory is a valid catalogue cache entry
231// against the exact registration incarnation. The offline store and the
232// registry both apply this rule, so a projection the registry publishes is
233// never rejected (and therefore silently never persisted) by the store.
234// Transient state (Checking and the live failure cause) is deliberately not
235// part of the durable rule.
236func ValidateDurableHostProjection(obs BrokerDaemonObservation) error {
237 if obs.Local {
238 return errors.New("ports: local daemon observation is never durable")
239 }
240 if obs.Availability < domain.RemoteAvailabilityUnknown || obs.Availability > domain.RemoteAvailabilityNoDaemon {
241 return errors.New("ports: broker host availability is out of range")
242 }
243 if obs.LastFailure.Kind > domain.RemoteFailureInvalidResponse {
244 return errors.New("ports: broker host failure kind is out of range")
245 }
246 // Observed identity is part of the durable rule, not only of the live one: a
247 // restored record with partial identity or without the protocol version its
248 // identity was authenticated with would otherwise reach a publication the
249 // wire refuses, taking down every client connection instead of failing
250 // closed here.
251 if err := validateObservedDaemonState(obs); err != nil {
252 return err
253 }
254 if len(obs.Sessions) == 0 {
255 return nil
256 }
257 return catalogue.ValidateRemoteCatalogCacheEntries([]catalogue.RemoteCatalogCacheEntry{{
258 Host: obs.Endpoint,
259 Incarnation: obs.Registration.Incarnation,
260 FetchedAt: obs.LastSuccess,
261 Sessions: obs.Sessions,
262 }})
263}
264
265// ValidateDurableObservation checks the shared remote-only durable observation shape.
266func ValidateDurableObservation(daemon BrokerDaemonObservation) error {
267 if daemon.Local {
268 return errors.New("local daemon observation is never durable")
269 }
270 if err := domain.ValidateRemoteHostTarget(daemon.Endpoint); err != nil {
271 return err
272 }
273 if err := daemon.Registration.Validate(); err != nil {
274 return err
275 }
276 if daemon.Endpoint != daemon.Registration.Endpoint {
277 return errors.New("observation endpoint does not match registration")
278 }
279 if len(daemon.Sessions) > 0 && !daemon.InventoryKnown {
280 return errors.New("observation carries sessions without known inventory")
281 }
282 return ValidateDurableHostProjection(daemon)
283}
284
285// DurableProjection returns the persistable copy of a published snapshot. It
286// is the single rule for what is transient, shared by the registry writer and
287// every store:
288//
289// - Removed tombstones are process-local fencing; a restart never adopts them.
290// - Checking and the live failure cause are in-flight observation state.
291// - Policy is membership authority, not observation: the loader re-stamps it
292// from the matching BrokerHostRecord, so persisting it would create a
293// second, stale authority.
294// - Tab Attention is a transient bell that must not resurrect on reload.
295//
296// Local observations are kept so ValidateDurableSnapshot still refuses them;
297// a producer that may hold one drops it first (see WithoutLocal).
298func (s BrokerSnapshot) DurableProjection() BrokerSnapshot {
299 out := s.Clone()
300 out.Removed = nil
301 for i := range out.Daemons {
302 daemon := &out.Daemons[i]
303 daemon.Checking = false
304 daemon.LastFailure.Err = nil
305 daemon.Policy = BrokerPolicy{}
306 for j := range daemon.Sessions {
307 for k := range daemon.Sessions[j].Tabs {
308 daemon.Sessions[j].Tabs[k].Attention = false
309 }
310 }
311 }
312 return out
313}
314
315// WithoutLocal returns s without the process-local daemon observation. The
316// daemon list is new, but nested data stays shared with s; chain
317// DurableProjection (or Clone) before mutating the result.
318func (s BrokerSnapshot) WithoutLocal() BrokerSnapshot {
319 out := s
320 daemons := make([]BrokerDaemonObservation, 0, len(s.Daemons))
321 for _, daemon := range s.Daemons {
322 if !daemon.Local {
323 daemons = append(daemons, daemon)
324 }
325 }
326 out.Daemons = daemons
327 return out
328}
329
330// ValidateDurableSnapshot checks the shared durable snapshot shape.
331func ValidateDurableSnapshot(snapshot BrokerSnapshot) error {
332 if snapshot.Epoch == 0 {
333 if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
334 return errors.New("invalid empty snapshot")
335 }
336 return nil
337 }
338 if snapshot.Revision == 0 {
339 return errors.New("snapshot has no revision")
340 }
341 if len(snapshot.Daemons) > BrokerMaxDaemonsPerSnapshot {
342 return errors.New("snapshot has too many daemons")
343 }
344 seen := make(map[string]struct{}, len(snapshot.Daemons))
345 for _, daemon := range snapshot.Daemons {
346 if err := ValidateDurableObservation(daemon); err != nil {
347 return err
348 }
349 if _, duplicate := seen[daemon.Endpoint]; duplicate {
350 return errors.New("snapshot has duplicate host")
351 }
352 seen[daemon.Endpoint] = struct{}{}
353 }
354 if len(snapshot.Removed) > BrokerMaxTombstones {
355 return errors.New("snapshot has too many tombstones")
356 }
357 for _, tombstone := range snapshot.Removed {
358 if err := tombstone.Validate(); err != nil {
359 return err
360 }
361 if _, live := seen[tombstone.Endpoint]; live {
362 return errors.New("snapshot carries a live host as tombstone")
363 }
364 }
365 return nil
366}