refactor(brokerstore): drop legacy host and cache import

commit: 2f0d2b4ab1080e6888e61f917e5baad3d174f271
author: brice · 30d ago · 2026-09-23 12:14 UTC
parent: 2d72cbf
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +148 -826

A internal/adapters/brokerstore/json.go +91 -0

@@ -0,0 +1,91 @@
	1	+	package brokerstore
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"encoding/json"
	6	+	"errors"
	7	+	"io"
	8	+	"os"
	9	+	"syscall"
	10	+	"unicode/utf8"
	11	+	)
	12	+
	13	+	func readBounded(path string) ([]byte, error) {
	14	+	fd, err := syscall.Open(path, syscall.O_RDONLY\|syscall.O_CLOEXEC\|syscall.O_NOFOLLOW\|syscall.O_NONBLOCK, 0)
	15	+	if err != nil {
	16	+	return nil, err
	17	+	}
	18	+	f := os.NewFile(uintptr(fd), path)
	19	+	defer f.Close()
	20	+	st, err := f.Stat()
	21	+	if err != nil {
	22	+	return nil, err
	23	+	}
	24	+	if !st.Mode().IsRegular() \|\| st.Size() > MaxFileBytes {
	25	+	return nil, errors.New("brokerstore: invalid file type or size")
	26	+	}
	27	+	b, err := io.ReadAll(io.LimitReader(f, MaxFileBytes+1))
	28	+	if len(b) > MaxFileBytes {
	29	+	return nil, errors.New("brokerstore: oversized input")
	30	+	}
	31	+	return b, err
	32	+	}
	33	+
	34	+	// strict rejects duplicate keys as well as unknown fields, trailing JSON,
	35	+	// invalid UTF-8 and excessive nesting. Byte bounds apply before decoding.
	36	+	func strict(raw []byte, dst any) error {
	37	+	if len(raw) > MaxFileBytes \|\| !utf8.Valid(raw) {
	38	+	return errors.New("brokerstore: invalid JSON size or encoding")
	39	+	}
	40	+	scan := json.NewDecoder(bytes.NewReader(raw))
	41	+	if err := scanValue(scan, 0); err != nil {
	42	+	return err
	43	+	}
	44	+	if _, err := scan.Token(); err != io.EOF {
	45	+	return errors.New("brokerstore: trailing JSON")
	46	+	}
	47	+	dec := json.NewDecoder(bytes.NewReader(raw))
	48	+	dec.DisallowUnknownFields()
	49	+	return dec.Decode(dst)
	50	+	}
	51	+	func scanValue(d *json.Decoder, depth int) error {
	52	+	if depth > 64 {
	53	+	return errors.New("brokerstore: JSON nesting limit")
	54	+	}
	55	+	t, err := d.Token()
	56	+	if err != nil {
	57	+	return err
	58	+	}
	59	+	delim, ok := t.(json.Delim)
	60	+	if !ok {
	61	+	return nil
	62	+	}
	63	+	switch delim {
	64	+	case '{':
	65	+	seen := map[string]bool{}
	66	+	for d.More() {
	67	+	key, err := d.Token()
	68	+	if err != nil {
	69	+	return err
	70	+	}
	71	+	k, ok := key.(string)
	72	+	if !ok \|\| seen[k] {
	73	+	return errors.New("brokerstore: duplicate JSON key")
	74	+	}
	75	+	seen[k] = true
	76	+	if err := scanValue(d, depth+1); err != nil {
	77	+	return err
	78	+	}
	79	+	}
	80	+	case '[':
	81	+	for d.More() {
	82	+	if err := scanValue(d, depth+1); err != nil {
	83	+	return err
	84	+	}
	85	+	}
	86	+	default:
	87	+	return errors.New("brokerstore: unexpected delimiter")
	88	+	}
	89	+	_, err = d.Token()
	90	+	return err
	91	+	}

@@ -0,0 +1,91 @@
	1	+	package brokerstore
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"encoding/json"
	6	+	"errors"
	7	+	"io"
	8	+	"os"
	9	+	"syscall"
	10	+	"unicode/utf8"
	11	+	)
	12	+
	13	+	func readBounded(path string) ([]byte, error) {
	14	+	fd, err := syscall.Open(path, syscall.O_RDONLY\|syscall.O_CLOEXEC\|syscall.O_NOFOLLOW\|syscall.O_NONBLOCK, 0)
	15	+	if err != nil {
	16	+	return nil, err
	17	+	}
	18	+	f := os.NewFile(uintptr(fd), path)
	19	+	defer f.Close()
	20	+	st, err := f.Stat()
	21	+	if err != nil {
	22	+	return nil, err
	23	+	}
	24	+	if !st.Mode().IsRegular() \|\| st.Size() > MaxFileBytes {
	25	+	return nil, errors.New("brokerstore: invalid file type or size")
	26	+	}
	27	+	b, err := io.ReadAll(io.LimitReader(f, MaxFileBytes+1))
	28	+	if len(b) > MaxFileBytes {
	29	+	return nil, errors.New("brokerstore: oversized input")
	30	+	}
	31	+	return b, err
	32	+	}
	33	+
	34	+	// strict rejects duplicate keys as well as unknown fields, trailing JSON,
	35	+	// invalid UTF-8 and excessive nesting. Byte bounds apply before decoding.
	36	+	func strict(raw []byte, dst any) error {
	37	+	if len(raw) > MaxFileBytes \|\| !utf8.Valid(raw) {
	38	+	return errors.New("brokerstore: invalid JSON size or encoding")
	39	+	}
	40	+	scan := json.NewDecoder(bytes.NewReader(raw))
	41	+	if err := scanValue(scan, 0); err != nil {
	42	+	return err
	43	+	}
	44	+	if _, err := scan.Token(); err != io.EOF {
	45	+	return errors.New("brokerstore: trailing JSON")
	46	+	}
	47	+	dec := json.NewDecoder(bytes.NewReader(raw))
	48	+	dec.DisallowUnknownFields()
	49	+	return dec.Decode(dst)
	50	+	}
	51	+	func scanValue(d *json.Decoder, depth int) error {
	52	+	if depth > 64 {
	53	+	return errors.New("brokerstore: JSON nesting limit")
	54	+	}
	55	+	t, err := d.Token()
	56	+	if err != nil {
	57	+	return err
	58	+	}
	59	+	delim, ok := t.(json.Delim)
	60	+	if !ok {
	61	+	return nil
	62	+	}
	63	+	switch delim {
	64	+	case '{':
	65	+	seen := map[string]bool{}
	66	+	for d.More() {
	67	+	key, err := d.Token()
	68	+	if err != nil {
	69	+	return err
	70	+	}
	71	+	k, ok := key.(string)
	72	+	if !ok \|\| seen[k] {
	73	+	return errors.New("brokerstore: duplicate JSON key")
	74	+	}
	75	+	seen[k] = true
	76	+	if err := scanValue(d, depth+1); err != nil {
	77	+	return err
	78	+	}
	79	+	}
	80	+	case '[':
	81	+	for d.More() {
	82	+	if err := scanValue(d, depth+1); err != nil {
	83	+	return err
	84	+	}
	85	+	}
	86	+	default:
	87	+	return errors.New("brokerstore: unexpected delimiter")
	88	+	}
	89	+	_, err = d.Token()
	90	+	return err
	91	+	}

D internal/adapters/brokerstore/legacy.go +0 -289

@@ -1,289 +0,0 @@
1	-	package brokerstore
2	-
3	-	import (
4	-	"bytes"
5	-	"crypto/rand"
6	-	"encoding/hex"
7	-	"encoding/json"
8	-	"errors"
9	-	"fmt"
10	-	"io"
11	-	"os"
12	-	"syscall"
13	-	"unicode/utf8"
14	-
15	-	"github.com/bnema/vev/internal/domain"
16	-	"github.com/bnema/vev/internal/ports"
17	-	)
18	-
19	-	func readBounded(path string) ([]byte, error) {
20	-	fd, err := syscall.Open(path, syscall.O_RDONLY\|syscall.O_CLOEXEC\|syscall.O_NOFOLLOW\|syscall.O_NONBLOCK, 0)
21	-	if err != nil {
22	-	return nil, err
23	-	}
24	-	f := os.NewFile(uintptr(fd), path)
25	-	defer f.Close()
26	-	st, err := f.Stat()
27	-	if err != nil {
28	-	return nil, err
29	-	}
30	-	if !st.Mode().IsRegular() \|\| st.Size() > MaxFileBytes {
31	-	return nil, errors.New("brokerstore: invalid file type or size")
32	-	}
33	-	b, err := io.ReadAll(io.LimitReader(f, MaxFileBytes+1))
34	-	if len(b) > MaxFileBytes {
35	-	return nil, errors.New("brokerstore: oversized input")
36	-	}
37	-	return b, err
38	-	}
39	-
40	-	// strict rejects duplicate keys as well as unknown fields, trailing JSON,
41	-	// invalid UTF-8 and excessive nesting. Byte bounds apply before decoding.
42	-	func strict(raw []byte, dst any) error {
43	-	if len(raw) > MaxFileBytes \|\| !utf8.Valid(raw) {
44	-	return errors.New("brokerstore: invalid JSON size or encoding")
45	-	}
46	-	scan := json.NewDecoder(bytes.NewReader(raw))
47	-	if err := scanValue(scan, 0); err != nil {
48	-	return err
49	-	}
50	-	if _, err := scan.Token(); err != io.EOF {
51	-	return errors.New("brokerstore: trailing JSON")
52	-	}
53	-	dec := json.NewDecoder(bytes.NewReader(raw))
54	-	dec.DisallowUnknownFields()
55	-	return dec.Decode(dst)
56	-	}
57	-	func scanValue(d *json.Decoder, depth int) error {
58	-	if depth > 64 {
59	-	return errors.New("brokerstore: JSON nesting limit")
60	-	}
61	-	t, err := d.Token()
62	-	if err != nil {
63	-	return err
64	-	}
65	-	delim, ok := t.(json.Delim)
66	-	if !ok {
67	-	return nil
68	-	}
69	-	switch delim {
70	-	case '{':
71	-	seen := map[string]bool{}
72	-	for d.More() {
73	-	key, err := d.Token()
74	-	if err != nil {
75	-	return err
76	-	}
77	-	k, ok := key.(string)
78	-	if !ok \|\| seen[k] {
79	-	return errors.New("brokerstore: duplicate JSON key")
80	-	}
81	-	seen[k] = true
82	-	if err := scanValue(d, depth+1); err != nil {
83	-	return err
84	-	}
85	-	}
86	-	case '[':
87	-	for d.More() {
88	-	if err := scanValue(d, depth+1); err != nil {
89	-	return err
90	-	}
91	-	}
92	-	default:
93	-	return errors.New("brokerstore: unexpected delimiter")
94	-	}
95	-	_, err = d.Token()
96	-	return err
97	-	}
98	-	func source(path string) ([]byte, error) {
99	-	if path == "" {
100	-	return nil, nil
101	-	}
102	-	return readBounded(path)
103	-	}
104	-	func migrate(o Options) (recovery, error) {
105	-	var r recovery
106	-	var err error
107	-	if r.Hosts, err = source(o.LegacyHosts); err != nil {
108	-	return r, err
109	-	}
110	-	if r.Cache, err = source(o.LegacyCache); err != nil {
111	-	return r, err
112	-	}
113	-	r.State.Manifest = manifest{Version: 1, ImportVersion: 1, ImportCompleted: true, Sources: []manifestSource{
114	-	{Label: manifestMembership, Present: o.LegacyHosts != "", SHA256: digest(r.Hosts)},
115	-	{Label: manifestObservations, Present: o.LegacyCache != "", SHA256: digest(r.Cache)},
116	-	}}
117	-	r.State.Hosts = ports.BrokerHosts{Revision: 1}
118	-	if o.InitialImportProvided {
119	-	r.State.Hosts.Hosts = ports.CloneBrokerHostRecords(o.InitialHosts)
120	-	if err := ports.ValidateBrokerHostRecords(r.State.Hosts.Hosts); err != nil && len(r.State.Hosts.Hosts) != 0 {
121	-	return r, fmt.Errorf("brokerstore: initial import: %w", err)
122	-	}
123	-	} else if o.LegacyHosts != "" {
124	-	r.State.Hosts.Hosts, err = decodeHosts(r.Hosts, o.Policies)
125	-	if err != nil {
126	-	return r, err
127	-	}
128	-	}
129	-	if o.LegacyCache != "" {
130	-	entries, err := decodeCache(r.Cache)
131	-	if err != nil {
132	-	return r, err
133	-	}
134	-	for _, e := range entries {
135	-	for _, h := range r.State.Hosts.Hosts {
136	-	// Unbound v2/v3 cache data is validated and backed up, never promoted
137	-	// to authority. v4 can seed only its exact surviving incarnation.
138	-	if e.Host == h.Registration.Endpoint && e.Incarnation == h.Registration.Incarnation {
139	-	// A legacy cache entry carries only observed sessions: daemon
140	-	// identity, incarnation, protocol version, and capabilities stay
141	-	// zero (never invented), and policy stays absent because the
142	-	// loader re-stamps it from membership.
143	-	r.State.Snapshot.Daemons = append(r.State.Snapshot.Daemons, ports.BrokerDaemonObservation{
144	-	Endpoint: e.Host,
145	-	DisplayOrigin: domain.RemoteDisplayOrigin(e.Host),
146	-	Registration: h.Registration,
147	-	Availability: domain.RemoteAvailabilityUnknown,
148	-	InventoryKnown: true,
149	-	LastSuccess: e.FetchedAt,
150	-	Sessions: e.Sessions,
151	-	})
152	-	}
153	-	}
154	-	}
155	-	}
156	-	if len(r.State.Snapshot.Daemons) > 0 {
157	-	var epoch [8]byte
158	-	if _, err = rand.Read(epoch[:]); err != nil {
159	-	return r, err
160	-	}
161	-	var n uint64
162	-	for _, b := range epoch {
163	-	n = n<<8 \| uint64(b)
164	-	}
165	-	if n == 0 {
166	-	n = 1
167	-	}
168	-	r.State.Snapshot.Epoch = ports.BrokerEpoch(n)
169	-	r.State.Snapshot.Revision = 1
170	-	}
171	-	return r, validate(r.State)
172	-	}
173	-
174	-	type legacyRecord struct {
175	-	Endpoint string `json:"endpoint"`
176	-	Incarnation string `json:"incarnation"`
177	-	Generation uint64 `json:"generation"`
178	-	}
179	-
180	-	func decodeHosts(raw []byte, policies map[string]ports.BrokerPolicy) ([]ports.BrokerHostRecord, error) {
181	-	var header struct {
182	-	Version int `json:"version"`
183	-	}
184	-	if err := json.Unmarshal(raw, &header); err != nil {
185	-	return nil, err
186	-	}
187	-	var pinned, learned []legacyRecord
188	-	switch header.Version {
189	-	case 1:
190	-	var f struct {
191	-	Version int `json:"version"`
192	-	Hosts []string `json:"hosts"`
193	-	}
194	-	if err := strict(raw, &f); err != nil {
195	-	return nil, err
196	-	}
197	-	if f.Hosts == nil {
198	-	return nil, errors.New("brokerstore: missing hosts")
199	-	}
200	-	for _, e := range f.Hosts {
201	-	learned = append(learned, legacyRecord{Endpoint: e})
202	-	}
203	-	case 2:
204	-	var f struct {
205	-	Version int `json:"version"`
206	-	Pinned []string `json:"pinned"`
207	-	Learned []string `json:"learned"`
208	-	}
209	-	if err := strict(raw, &f); err != nil {
210	-	return nil, err
211	-	}
212	-	if f.Pinned == nil \|\| f.Learned == nil {
213	-	return nil, errors.New("brokerstore: missing memberships")
214	-	}
215	-	for _, e := range f.Pinned {
216	-	pinned = append(pinned, legacyRecord{Endpoint: e})
217	-	}
218	-	for _, e := range f.Learned {
219	-	learned = append(learned, legacyRecord{Endpoint: e})
220	-	}
221	-	case 3:
222	-	var f struct {
223	-	Version int `json:"version"`
224	-	Pinned []legacyRecord `json:"pinned"`
225	-	Learned []legacyRecord `json:"learned"`
226	-	}
227	-	if err := strict(raw, &f); err != nil {
228	-	return nil, err
229	-	}
230	-	if f.Pinned == nil \|\| f.Learned == nil {
231	-	return nil, errors.New("brokerstore: missing memberships")
232	-	}
233	-	pinned, learned = f.Pinned, f.Learned
234	-	default:
235	-	return nil, fmt.Errorf("brokerstore: unsupported hosts version %d", header.Version)
236	-	}
237	-	if len(pinned) > ports.BrokerMaxHosts \|\| len(learned) > ports.BrokerMaxHosts {
238	-	return nil, errors.New("brokerstore: too many hosts")
239	-	}
240	-	var out []ports.BrokerHostRecord
241	-	indices := map[string]int{}
242	-	for group, records := range [][]legacyRecord{pinned, learned} {
243	-	seen := map[string]bool{}
244	-	for _, record := range records {
245	-	if seen[record.Endpoint] {
246	-	return nil, errors.New("brokerstore: duplicate membership")
247	-	}
248	-	seen[record.Endpoint] = true
249	-	reg := domain.RemoteRegistration{Endpoint: record.Endpoint, Generation: domain.RemoteGeneration(record.Generation)}
250	-	if header.Version == 3 {
251	-	b, err := hex.DecodeString(record.Incarnation)
252	-	if err != nil \|\| len(b) != 16 {
253	-	return nil, errors.New("brokerstore: invalid incarnation")
254	-	}
255	-	copy(reg.Incarnation[:], b)
256	-	}
257	-	if index, ok := indices[record.Endpoint]; ok {
258	-	if header.Version == 3 && !reg.Equal(out[index].Registration) {
259	-	return nil, errors.New("brokerstore: conflicting registration")
260	-	}
261	-	out[index].Learned = true
262	-	continue
263	-	}
264	-	if header.Version != 3 {
265	-	if _, err := rand.Read(reg.Incarnation[:]); err != nil {
266	-	return nil, err
267	-	}
268	-	reg.Generation = 1
269	-	}
270	-	if err := reg.Validate(); err != nil {
271	-	return nil, err
272	-	}
273	-	policy, ok := policies[record.Endpoint]
274	-	if !ok {
275	-	return nil, fmt.Errorf("brokerstore: explicit policy required for %q", record.Endpoint)
276	-	}
277	-	if err := policy.Validate(); err != nil {
278	-	return nil, err
279	-	}
280	-	indices[record.Endpoint] = len(out)
281	-	route, err := ports.BrokerRouteForTransport(policy.Transport, reg.Endpoint)
282	-	if err != nil {
283	-	return nil, fmt.Errorf("brokerstore: upgrade route for %q: %w", record.Endpoint, err)
284	-	}
285	-	out = append(out, ports.BrokerHostRecord{Registration: reg, Pinned: group == 0, Learned: group == 1, Policy: policy, Route: route})
286	-	}
287	-	}
288	-	return out, nil
289	-	}

@@ -1,289 +0,0 @@
1	-	package brokerstore
2	-
3	-	import (
4	-	"bytes"
5	-	"crypto/rand"
6	-	"encoding/hex"
7	-	"encoding/json"
8	-	"errors"
9	-	"fmt"
10	-	"io"
11	-	"os"
12	-	"syscall"
13	-	"unicode/utf8"
14	-
15	-	"github.com/bnema/vev/internal/domain"
16	-	"github.com/bnema/vev/internal/ports"
17	-	)
18	-
19	-	func readBounded(path string) ([]byte, error) {
20	-	fd, err := syscall.Open(path, syscall.O_RDONLY\|syscall.O_CLOEXEC\|syscall.O_NOFOLLOW\|syscall.O_NONBLOCK, 0)
21	-	if err != nil {
22	-	return nil, err
23	-	}
24	-	f := os.NewFile(uintptr(fd), path)
25	-	defer f.Close()
26	-	st, err := f.Stat()
27	-	if err != nil {
28	-	return nil, err
29	-	}
30	-	if !st.Mode().IsRegular() \|\| st.Size() > MaxFileBytes {
31	-	return nil, errors.New("brokerstore: invalid file type or size")
32	-	}
33	-	b, err := io.ReadAll(io.LimitReader(f, MaxFileBytes+1))
34	-	if len(b) > MaxFileBytes {
35	-	return nil, errors.New("brokerstore: oversized input")
36	-	}
37	-	return b, err
38	-	}
39	-
40	-	// strict rejects duplicate keys as well as unknown fields, trailing JSON,
41	-	// invalid UTF-8 and excessive nesting. Byte bounds apply before decoding.
42	-	func strict(raw []byte, dst any) error {
43	-	if len(raw) > MaxFileBytes \|\| !utf8.Valid(raw) {
44	-	return errors.New("brokerstore: invalid JSON size or encoding")
45	-	}
46	-	scan := json.NewDecoder(bytes.NewReader(raw))
47	-	if err := scanValue(scan, 0); err != nil {
48	-	return err
49	-	}
50	-	if _, err := scan.Token(); err != io.EOF {
51	-	return errors.New("brokerstore: trailing JSON")
52	-	}
53	-	dec := json.NewDecoder(bytes.NewReader(raw))
54	-	dec.DisallowUnknownFields()
55	-	return dec.Decode(dst)
56	-	}
57	-	func scanValue(d *json.Decoder, depth int) error {
58	-	if depth > 64 {
59	-	return errors.New("brokerstore: JSON nesting limit")
60	-	}
61	-	t, err := d.Token()
62	-	if err != nil {
63	-	return err
64	-	}
65	-	delim, ok := t.(json.Delim)
66	-	if !ok {
67	-	return nil
68	-	}
69	-	switch delim {
70	-	case '{':
71	-	seen := map[string]bool{}
72	-	for d.More() {
73	-	key, err := d.Token()
74	-	if err != nil {
75	-	return err
76	-	}
77	-	k, ok := key.(string)
78	-	if !ok \|\| seen[k] {
79	-	return errors.New("brokerstore: duplicate JSON key")
80	-	}
81	-	seen[k] = true
82	-	if err := scanValue(d, depth+1); err != nil {
83	-	return err
84	-	}
85	-	}
86	-	case '[':
87	-	for d.More() {
88	-	if err := scanValue(d, depth+1); err != nil {
89	-	return err
90	-	}
91	-	}
92	-	default:
93	-	return errors.New("brokerstore: unexpected delimiter")
94	-	}
95	-	_, err = d.Token()
96	-	return err
97	-	}
98	-	func source(path string) ([]byte, error) {
99	-	if path == "" {
100	-	return nil, nil
101	-	}
102	-	return readBounded(path)
103	-	}
104	-	func migrate(o Options) (recovery, error) {
105	-	var r recovery
106	-	var err error
107	-	if r.Hosts, err = source(o.LegacyHosts); err != nil {
108	-	return r, err
109	-	}
110	-	if r.Cache, err = source(o.LegacyCache); err != nil {
111	-	return r, err
112	-	}
113	-	r.State.Manifest = manifest{Version: 1, ImportVersion: 1, ImportCompleted: true, Sources: []manifestSource{
114	-	{Label: manifestMembership, Present: o.LegacyHosts != "", SHA256: digest(r.Hosts)},
115	-	{Label: manifestObservations, Present: o.LegacyCache != "", SHA256: digest(r.Cache)},
116	-	}}
117	-	r.State.Hosts = ports.BrokerHosts{Revision: 1}
118	-	if o.InitialImportProvided {
119	-	r.State.Hosts.Hosts = ports.CloneBrokerHostRecords(o.InitialHosts)
120	-	if err := ports.ValidateBrokerHostRecords(r.State.Hosts.Hosts); err != nil && len(r.State.Hosts.Hosts) != 0 {
121	-	return r, fmt.Errorf("brokerstore: initial import: %w", err)
122	-	}
123	-	} else if o.LegacyHosts != "" {
124	-	r.State.Hosts.Hosts, err = decodeHosts(r.Hosts, o.Policies)
125	-	if err != nil {
126	-	return r, err
127	-	}
128	-	}
129	-	if o.LegacyCache != "" {
130	-	entries, err := decodeCache(r.Cache)
131	-	if err != nil {
132	-	return r, err
133	-	}
134	-	for _, e := range entries {
135	-	for _, h := range r.State.Hosts.Hosts {
136	-	// Unbound v2/v3 cache data is validated and backed up, never promoted
137	-	// to authority. v4 can seed only its exact surviving incarnation.
138	-	if e.Host == h.Registration.Endpoint && e.Incarnation == h.Registration.Incarnation {
139	-	// A legacy cache entry carries only observed sessions: daemon
140	-	// identity, incarnation, protocol version, and capabilities stay
141	-	// zero (never invented), and policy stays absent because the
142	-	// loader re-stamps it from membership.
143	-	r.State.Snapshot.Daemons = append(r.State.Snapshot.Daemons, ports.BrokerDaemonObservation{
144	-	Endpoint: e.Host,
145	-	DisplayOrigin: domain.RemoteDisplayOrigin(e.Host),
146	-	Registration: h.Registration,
147	-	Availability: domain.RemoteAvailabilityUnknown,
148	-	InventoryKnown: true,
149	-	LastSuccess: e.FetchedAt,
150	-	Sessions: e.Sessions,
151	-	})
152	-	}
153	-	}
154	-	}
155	-	}
156	-	if len(r.State.Snapshot.Daemons) > 0 {
157	-	var epoch [8]byte
158	-	if _, err = rand.Read(epoch[:]); err != nil {
159	-	return r, err
160	-	}
161	-	var n uint64
162	-	for _, b := range epoch {
163	-	n = n<<8 \| uint64(b)
164	-	}
165	-	if n == 0 {
166	-	n = 1
167	-	}
168	-	r.State.Snapshot.Epoch = ports.BrokerEpoch(n)
169	-	r.State.Snapshot.Revision = 1
170	-	}
171	-	return r, validate(r.State)
172	-	}
173	-
174	-	type legacyRecord struct {
175	-	Endpoint string `json:"endpoint"`
176	-	Incarnation string `json:"incarnation"`
177	-	Generation uint64 `json:"generation"`
178	-	}
179	-
180	-	func decodeHosts(raw []byte, policies map[string]ports.BrokerPolicy) ([]ports.BrokerHostRecord, error) {
181	-	var header struct {
182	-	Version int `json:"version"`
183	-	}
184	-	if err := json.Unmarshal(raw, &header); err != nil {
185	-	return nil, err
186	-	}
187	-	var pinned, learned []legacyRecord
188	-	switch header.Version {
189	-	case 1:
190	-	var f struct {
191	-	Version int `json:"version"`
192	-	Hosts []string `json:"hosts"`
193	-	}
194	-	if err := strict(raw, &f); err != nil {
195	-	return nil, err
196	-	}
197	-	if f.Hosts == nil {
198	-	return nil, errors.New("brokerstore: missing hosts")
199	-	}
200	-	for _, e := range f.Hosts {
201	-	learned = append(learned, legacyRecord{Endpoint: e})
202	-	}
203	-	case 2:
204	-	var f struct {
205	-	Version int `json:"version"`
206	-	Pinned []string `json:"pinned"`
207	-	Learned []string `json:"learned"`
208	-	}
209	-	if err := strict(raw, &f); err != nil {
210	-	return nil, err
211	-	}
212	-	if f.Pinned == nil \|\| f.Learned == nil {
213	-	return nil, errors.New("brokerstore: missing memberships")
214	-	}
215	-	for _, e := range f.Pinned {
216	-	pinned = append(pinned, legacyRecord{Endpoint: e})
217	-	}
218	-	for _, e := range f.Learned {
219	-	learned = append(learned, legacyRecord{Endpoint: e})
220	-	}
221	-	case 3:
222	-	var f struct {
223	-	Version int `json:"version"`
224	-	Pinned []legacyRecord `json:"pinned"`
225	-	Learned []legacyRecord `json:"learned"`
226	-	}
227	-	if err := strict(raw, &f); err != nil {
228	-	return nil, err
229	-	}
230	-	if f.Pinned == nil \|\| f.Learned == nil {
231	-	return nil, errors.New("brokerstore: missing memberships")
232	-	}
233	-	pinned, learned = f.Pinned, f.Learned
234	-	default:
235	-	return nil, fmt.Errorf("brokerstore: unsupported hosts version %d", header.Version)
236	-	}
237	-	if len(pinned) > ports.BrokerMaxHosts \|\| len(learned) > ports.BrokerMaxHosts {
238	-	return nil, errors.New("brokerstore: too many hosts")
239	-	}
240	-	var out []ports.BrokerHostRecord
241	-	indices := map[string]int{}
242	-	for group, records := range [][]legacyRecord{pinned, learned} {
243	-	seen := map[string]bool{}
244	-	for _, record := range records {
245	-	if seen[record.Endpoint] {
246	-	return nil, errors.New("brokerstore: duplicate membership")
247	-	}
248	-	seen[record.Endpoint] = true
249	-	reg := domain.RemoteRegistration{Endpoint: record.Endpoint, Generation: domain.RemoteGeneration(record.Generation)}
250	-	if header.Version == 3 {
251	-	b, err := hex.DecodeString(record.Incarnation)
252	-	if err != nil \|\| len(b) != 16 {
253	-	return nil, errors.New("brokerstore: invalid incarnation")
254	-	}
255	-	copy(reg.Incarnation[:], b)
256	-	}
257	-	if index, ok := indices[record.Endpoint]; ok {
258	-	if header.Version == 3 && !reg.Equal(out[index].Registration) {
259	-	return nil, errors.New("brokerstore: conflicting registration")
260	-	}
261	-	out[index].Learned = true
262	-	continue
263	-	}
264	-	if header.Version != 3 {
265	-	if _, err := rand.Read(reg.Incarnation[:]); err != nil {
266	-	return nil, err
267	-	}
268	-	reg.Generation = 1
269	-	}
270	-	if err := reg.Validate(); err != nil {
271	-	return nil, err
272	-	}
273	-	policy, ok := policies[record.Endpoint]
274	-	if !ok {
275	-	return nil, fmt.Errorf("brokerstore: explicit policy required for %q", record.Endpoint)
276	-	}
277	-	if err := policy.Validate(); err != nil {
278	-	return nil, err
279	-	}
280	-	indices[record.Endpoint] = len(out)
281	-	route, err := ports.BrokerRouteForTransport(policy.Transport, reg.Endpoint)
282	-	if err != nil {
283	-	return nil, fmt.Errorf("brokerstore: upgrade route for %q: %w", record.Endpoint, err)
284	-	}
285	-	out = append(out, ports.BrokerHostRecord{Registration: reg, Pinned: group == 0, Learned: group == 1, Policy: policy, Route: route})
286	-	}
287	-	}
288	-	return out, nil
289	-	}

D internal/adapters/brokerstore/legacy_cache.go +0 -258

@@ -1,258 +0,0 @@
1	-	package brokerstore
2	-
3	-	import (
4	-	"encoding/hex"
5	-	"encoding/json"
6	-	"fmt"
7	-	"sort"
8	-	"time"
9	-	"unicode/utf8"
10	-
11	-	"github.com/bnema/vev/internal/domain"
12	-	"github.com/bnema/vev/internal/protocol/catalogue"
13	-	)
14	-
15	-	const (
16	-	catalogCacheFileVersion = 4
17	-	catalogCacheLegacyFileVersion = 3
18	-	catalogCacheExactFileVersion = 2
19	-	)
20	-
21	-	type catalogCacheFile struct {
22	-	Version *int `json:"version"`
23	-	Hosts *[]catalogCacheHost `json:"hosts"`
24	-	}
25	-
26	-	type catalogCacheVersion struct {
27	-	Version *int `json:"version"`
28	-	}
29	-
30	-	type catalogCacheHost struct {
31	-	Target *string `json:"target"`
32	-	FetchedAtUnixNano *int64 `json:"fetched_at_unix_nano"`
33	-	Incarnation *string `json:"incarnation,omitempty"`
34	-	Sessions *[]catalogCacheSession `json:"sessions"`
35	-	}
36	-
37	-	type catalogCacheSession struct {
38	-	LifecycleID *domain.SessionLifecycleID `json:"lifecycle_id"`
39	-	Name *string `json:"name"`
40	-	State *catalogue.RemoteCatalogSessionState `json:"state"`
41	-	Ephemeral *bool `json:"ephemeral"`
42	-	LastUsedSeq *uint64 `json:"last_used_seq,omitempty"`
43	-	Tabs *[]catalogue.RemoteCatalogTab `json:"tabs"`
44	-	ActiveTabID *string `json:"active_tab_id,omitempty"`
45	-	Attached *bool `json:"attached"`
46	-	}
47	-
48	-	type catalogCacheFileV2 struct {
49	-	Version *int `json:"version"`
50	-	Hosts *[]catalogCacheHostV2 `json:"hosts"`
51	-	}
52	-
53	-	type catalogCacheHostV2 struct {
54	-	Target *string `json:"target"`
55	-	FetchedAtUnixNano *int64 `json:"fetched_at_unix_nano"`
56	-	Sessions *[]catalogCacheSessionV2 `json:"sessions"`
57	-	}
58	-
59	-	type catalogCacheSessionV2 struct {
60	-	LifecycleID *domain.SessionLifecycleID `json:"lifecycle_id,omitempty"`
61	-	Name *string `json:"name"`
62	-	State *catalogue.RemoteCatalogSessionState `json:"state"`
63	-	Ephemeral *bool `json:"ephemeral"`
64	-	LastUsedSeq *uint64 `json:"last_used_seq,omitempty"`
65	-	Tabs *uint16 `json:"tabs,omitempty"`
66	-	TabList *[]catalogue.RemoteCatalogTab `json:"tab_list,omitempty"`
67	-	ActiveTabID *string `json:"active_tab_id,omitempty"`
68	-	Attached *bool `json:"attached"`
69	-	}
70	-
71	-	func decodeCache(raw []byte) ([]catalogue.RemoteCatalogCacheEntry, error) {
72	-	var version catalogCacheVersion
73	-	if err := json.Unmarshal(raw, &version); err != nil {
74	-	return nil, err
75	-	}
76	-	if version.Version == nil {
77	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing version")
78	-	}
79	-
80	-	var file catalogCacheFile
81	-	requireIncarnation := false
82	-	switch *version.Version {
83	-	case catalogCacheFileVersion:
84	-	requireIncarnation = true
85	-	if err := strict(raw, &file); err != nil {
86	-	return nil, err
87	-	}
88	-	case catalogCacheLegacyFileVersion:
89	-	if err := strict(raw, &file); err != nil {
90	-	return nil, err
91	-	}
92	-	case catalogCacheExactFileVersion:
93	-	var legacy catalogCacheFileV2
94	-	if err := strict(raw, &legacy); err != nil {
95	-	return nil, err
96	-	}
97	-	migrated, err := migrateExactCatalogCacheV2(legacy)
98	-	if err != nil {
99	-	return nil, err
100	-	}
101	-	file = migrated
102	-	default:
103	-	return nil, fmt.Errorf("remote catalog cache: unsupported cache file version %d", *version.Version)
104	-	}
105	-	if file.Hosts == nil {
106	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing hosts")
107	-	}
108	-
109	-	entries := make([]catalogue.RemoteCatalogCacheEntry, 0, len(*file.Hosts))
110	-	for _, host := range *file.Hosts {
111	-	if host.Target == nil \|\| host.FetchedAtUnixNano == nil \|\| host.Sessions == nil {
112	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing host fields")
113	-	}
114	-	if !utf8.ValidString(*host.Target) {
115	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: invalid UTF-8")
116	-	}
117	-	if *host.FetchedAtUnixNano <= 0 {
118	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: non-positive fetched time")
119	-	}
120	-	entry := catalogue.RemoteCatalogCacheEntry{
121	-	Host: *host.Target,
122	-	FetchedAt: time.Unix(0, *host.FetchedAtUnixNano),
123	-	Sessions: make([]catalogue.RemoteCatalogSession, 0, len(*host.Sessions)),
124	-	}
125	-	if host.Incarnation != nil {
126	-	raw, err := hex.DecodeString(*host.Incarnation)
127	-	if err != nil \|\| len(raw) != 16 {
128	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: invalid incarnation for %q", *host.Target)
129	-	}
130	-	copy(entry.Incarnation[:], raw)
131	-	} else if requireIncarnation {
132	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing incarnation for %q", *host.Target)
133	-	}
134	-	// A zero incarnation is unbound advisory data: it loads but must
135	-	// never seed a registration.
136	-	for _, session := range *host.Sessions {
137	-	if session.LifecycleID == nil \|\| session.Name == nil \|\| session.State == nil \|\| session.Ephemeral == nil \|\| session.Attached == nil \|\| session.Tabs == nil {
138	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing session fields")
139	-	}
140	-	if !utf8.ValidString(session.Name) \|\| !utf8.ValidString(string(session.State)) {
141	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: invalid UTF-8")
142	-	}
143	-	tabs := make([]catalogue.RemoteCatalogTab, len(*session.Tabs))
144	-	copy(tabs, *session.Tabs)
145	-	decoded := catalogue.RemoteCatalogSession{
146	-	LifecycleID: *session.LifecycleID,
147	-	Name: *session.Name,
148	-	State: *session.State,
149	-	Ephemeral: *session.Ephemeral,
150	-	Tabs: tabs,
151	-	Attached: *session.Attached,
152	-	}
153	-	if session.LastUsedSeq != nil {
154	-	decoded.LastUsedSeq = *session.LastUsedSeq
155	-	}
156	-	if session.ActiveTabID != nil {
157	-	decoded.ActiveTabID = *session.ActiveTabID
158	-	}
159	-	entry.Sessions = append(entry.Sessions, decoded)
160	-	}
161	-	entries = append(entries, entry)
162	-	}
163	-	normalized, err := normalizeCatalogCacheEntries(entries)
164	-	if err != nil {
165	-	return nil, err
166	-	}
167	-	return normalized, nil
168	-	}
169	-
170	-	func migrateExactCatalogCacheV2(legacy catalogCacheFileV2) (catalogCacheFile, error) {
171	-	if legacy.Version == nil \|\| *legacy.Version != catalogCacheExactFileVersion \|\| legacy.Hosts == nil {
172	-	return catalogCacheFile{}, fmt.Errorf("remote catalog cache: malformed version 2 cache file")
173	-	}
174	-	hosts := make([]catalogCacheHost, 0, len(*legacy.Hosts))
175	-	for _, host := range *legacy.Hosts {
176	-	if host.Sessions == nil {
177	-	return catalogCacheFile{}, fmt.Errorf("remote catalog cache: malformed cache file: missing host fields")
178	-	}
179	-	sessions := make([]catalogCacheSession, 0, len(*host.Sessions))
180	-	for _, session := range *host.Sessions {
181	-	if session.Tabs != nil \|\| session.TabList == nil {
182	-	return catalogCacheFile{}, fmt.Errorf("remote catalog cache: unsupported ambiguous version 2 tab count")
183	-	}
184	-	sessions = append(sessions, catalogCacheSession{
185	-	LifecycleID: session.LifecycleID,
186	-	Name: session.Name,
187	-	State: session.State,
188	-	Ephemeral: session.Ephemeral,
189	-	LastUsedSeq: session.LastUsedSeq,
190	-	Tabs: session.TabList,
191	-	ActiveTabID: session.ActiveTabID,
192	-	Attached: session.Attached,
193	-	})
194	-	}
195	-	hosts = append(hosts, catalogCacheHost{
196	-	Target: host.Target, FetchedAtUnixNano: host.FetchedAtUnixNano, Sessions: &sessions,
197	-	})
198	-	}
199	-	version := catalogCacheFileVersion
200	-	return catalogCacheFile{Version: &version, Hosts: &hosts}, nil
201	-	}
202	-
203	-	func normalizeCatalogCacheEntries(entries []catalogue.RemoteCatalogCacheEntry) ([]catalogue.RemoteCatalogCacheEntry, error) {
204	-	if err := catalogue.ValidateRemoteCatalogCacheEntries(entries); err != nil {
205	-	return nil, err
206	-	}
207	-	normalized := make([]catalogue.RemoteCatalogCacheEntry, 0, len(entries))
208	-	hosts := make(map[string]struct{}, len(entries))
209	-	for _, entry := range entries {
210	-	if err := domain.ValidateRemoteHostTarget(entry.Host); err != nil {
211	-	return nil, fmt.Errorf("remote catalog cache: invalid host %q: %w", entry.Host, err)
212	-	}
213	-	if entry.FetchedAt.UnixNano() <= 0 {
214	-	return nil, fmt.Errorf("remote catalog cache: host %q has non-positive fetched time", entry.Host)
215	-	}
216	-	if _, duplicate := hosts[entry.Host]; duplicate {
217	-	return nil, fmt.Errorf("remote catalog cache: duplicate host %q", entry.Host)
218	-	}
219	-	hosts[entry.Host] = struct{}{}
220	-
221	-	copyEntry := catalogue.RemoteCatalogCacheEntry{
222	-	Host: entry.Host,
223	-	FetchedAt: entry.FetchedAt,
224	-	Incarnation: entry.Incarnation,
225	-	Sessions: make([]catalogue.RemoteCatalogSession, 0, len(entry.Sessions)),
226	-	}
227	-	sessions := make(map[string]struct{}, len(entry.Sessions))
228	-	for _, session := range entry.Sessions {
229	-	if !utf8.ValidString(session.Name) \|\| !utf8.ValidString(string(session.State)) {
230	-	return nil, fmt.Errorf("remote catalog cache: session is not valid UTF-8")
231	-	}
232	-	if err := domain.ValidateSessionName(session.Name); err != nil {
233	-	return nil, fmt.Errorf("remote catalog cache: invalid session %q: %w", session.Name, err)
234	-	}
235	-	if !session.State.Valid() {
236	-	return nil, fmt.Errorf("remote catalog cache: invalid session state %q", session.State)
237	-	}
238	-	if _, duplicate := sessions[session.Name]; duplicate {
239	-	return nil, fmt.Errorf("remote catalog cache: duplicate session %q for host %q", session.Name, entry.Host)
240	-	}
241	-	sessions[session.Name] = struct{}{}
242	-	copySession := session
243	-	copySession.Tabs = make([]catalogue.RemoteCatalogTab, len(session.Tabs))
244	-	for i, tab := range session.Tabs {
245	-	copySession.Tabs[i] = catalogue.RemoteCatalogTab{ID: tab.ID, Index: tab.Index, Name: tab.Name}
246	-	}
247	-	copyEntry.Sessions = append(copyEntry.Sessions, copySession)
248	-	}
249	-	sort.Slice(copyEntry.Sessions, func(i, j int) bool {
250	-	return copyEntry.Sessions[i].Name < copyEntry.Sessions[j].Name
251	-	})
252	-	normalized = append(normalized, copyEntry)
253	-	}
254	-	sort.Slice(normalized, func(i, j int) bool {
255	-	return normalized[i].Host < normalized[j].Host
256	-	})
257	-	return normalized, nil
258	-	}

@@ -1,258 +0,0 @@
1	-	package brokerstore
2	-
3	-	import (
4	-	"encoding/hex"
5	-	"encoding/json"
6	-	"fmt"
7	-	"sort"
8	-	"time"
9	-	"unicode/utf8"
10	-
11	-	"github.com/bnema/vev/internal/domain"
12	-	"github.com/bnema/vev/internal/protocol/catalogue"
13	-	)
14	-
15	-	const (
16	-	catalogCacheFileVersion = 4
17	-	catalogCacheLegacyFileVersion = 3
18	-	catalogCacheExactFileVersion = 2
19	-	)
20	-
21	-	type catalogCacheFile struct {
22	-	Version *int `json:"version"`
23	-	Hosts *[]catalogCacheHost `json:"hosts"`
24	-	}
25	-
26	-	type catalogCacheVersion struct {
27	-	Version *int `json:"version"`
28	-	}
29	-
30	-	type catalogCacheHost struct {
31	-	Target *string `json:"target"`
32	-	FetchedAtUnixNano *int64 `json:"fetched_at_unix_nano"`
33	-	Incarnation *string `json:"incarnation,omitempty"`
34	-	Sessions *[]catalogCacheSession `json:"sessions"`
35	-	}
36	-
37	-	type catalogCacheSession struct {
38	-	LifecycleID *domain.SessionLifecycleID `json:"lifecycle_id"`
39	-	Name *string `json:"name"`
40	-	State *catalogue.RemoteCatalogSessionState `json:"state"`
41	-	Ephemeral *bool `json:"ephemeral"`
42	-	LastUsedSeq *uint64 `json:"last_used_seq,omitempty"`
43	-	Tabs *[]catalogue.RemoteCatalogTab `json:"tabs"`
44	-	ActiveTabID *string `json:"active_tab_id,omitempty"`
45	-	Attached *bool `json:"attached"`
46	-	}
47	-
48	-	type catalogCacheFileV2 struct {
49	-	Version *int `json:"version"`
50	-	Hosts *[]catalogCacheHostV2 `json:"hosts"`
51	-	}
52	-
53	-	type catalogCacheHostV2 struct {
54	-	Target *string `json:"target"`
55	-	FetchedAtUnixNano *int64 `json:"fetched_at_unix_nano"`
56	-	Sessions *[]catalogCacheSessionV2 `json:"sessions"`
57	-	}
58	-
59	-	type catalogCacheSessionV2 struct {
60	-	LifecycleID *domain.SessionLifecycleID `json:"lifecycle_id,omitempty"`
61	-	Name *string `json:"name"`
62	-	State *catalogue.RemoteCatalogSessionState `json:"state"`
63	-	Ephemeral *bool `json:"ephemeral"`
64	-	LastUsedSeq *uint64 `json:"last_used_seq,omitempty"`
65	-	Tabs *uint16 `json:"tabs,omitempty"`
66	-	TabList *[]catalogue.RemoteCatalogTab `json:"tab_list,omitempty"`
67	-	ActiveTabID *string `json:"active_tab_id,omitempty"`
68	-	Attached *bool `json:"attached"`
69	-	}
70	-
71	-	func decodeCache(raw []byte) ([]catalogue.RemoteCatalogCacheEntry, error) {
72	-	var version catalogCacheVersion
73	-	if err := json.Unmarshal(raw, &version); err != nil {
74	-	return nil, err
75	-	}
76	-	if version.Version == nil {
77	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing version")
78	-	}
79	-
80	-	var file catalogCacheFile
81	-	requireIncarnation := false
82	-	switch *version.Version {
83	-	case catalogCacheFileVersion:
84	-	requireIncarnation = true
85	-	if err := strict(raw, &file); err != nil {
86	-	return nil, err
87	-	}
88	-	case catalogCacheLegacyFileVersion:
89	-	if err := strict(raw, &file); err != nil {
90	-	return nil, err
91	-	}
92	-	case catalogCacheExactFileVersion:
93	-	var legacy catalogCacheFileV2
94	-	if err := strict(raw, &legacy); err != nil {
95	-	return nil, err
96	-	}
97	-	migrated, err := migrateExactCatalogCacheV2(legacy)
98	-	if err != nil {
99	-	return nil, err
100	-	}
101	-	file = migrated
102	-	default:
103	-	return nil, fmt.Errorf("remote catalog cache: unsupported cache file version %d", *version.Version)
104	-	}
105	-	if file.Hosts == nil {
106	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing hosts")
107	-	}
108	-
109	-	entries := make([]catalogue.RemoteCatalogCacheEntry, 0, len(*file.Hosts))
110	-	for _, host := range *file.Hosts {
111	-	if host.Target == nil \|\| host.FetchedAtUnixNano == nil \|\| host.Sessions == nil {
112	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing host fields")
113	-	}
114	-	if !utf8.ValidString(*host.Target) {
115	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: invalid UTF-8")
116	-	}
117	-	if *host.FetchedAtUnixNano <= 0 {
118	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: non-positive fetched time")
119	-	}
120	-	entry := catalogue.RemoteCatalogCacheEntry{
121	-	Host: *host.Target,
122	-	FetchedAt: time.Unix(0, *host.FetchedAtUnixNano),
123	-	Sessions: make([]catalogue.RemoteCatalogSession, 0, len(*host.Sessions)),
124	-	}
125	-	if host.Incarnation != nil {
126	-	raw, err := hex.DecodeString(*host.Incarnation)
127	-	if err != nil \|\| len(raw) != 16 {
128	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: invalid incarnation for %q", *host.Target)
129	-	}
130	-	copy(entry.Incarnation[:], raw)
131	-	} else if requireIncarnation {
132	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing incarnation for %q", *host.Target)
133	-	}
134	-	// A zero incarnation is unbound advisory data: it loads but must
135	-	// never seed a registration.
136	-	for _, session := range *host.Sessions {
137	-	if session.LifecycleID == nil \|\| session.Name == nil \|\| session.State == nil \|\| session.Ephemeral == nil \|\| session.Attached == nil \|\| session.Tabs == nil {
138	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: missing session fields")
139	-	}
140	-	if !utf8.ValidString(session.Name) \|\| !utf8.ValidString(string(session.State)) {
141	-	return nil, fmt.Errorf("remote catalog cache: malformed cache file: invalid UTF-8")
142	-	}
143	-	tabs := make([]catalogue.RemoteCatalogTab, len(*session.Tabs))
144	-	copy(tabs, *session.Tabs)
145	-	decoded := catalogue.RemoteCatalogSession{
146	-	LifecycleID: *session.LifecycleID,
147	-	Name: *session.Name,
148	-	State: *session.State,
149	-	Ephemeral: *session.Ephemeral,
150	-	Tabs: tabs,
151	-	Attached: *session.Attached,
152	-	}
153	-	if session.LastUsedSeq != nil {
154	-	decoded.LastUsedSeq = *session.LastUsedSeq
155	-	}
156	-	if session.ActiveTabID != nil {
157	-	decoded.ActiveTabID = *session.ActiveTabID
158	-	}
159	-	entry.Sessions = append(entry.Sessions, decoded)
160	-	}
161	-	entries = append(entries, entry)
162	-	}
163	-	normalized, err := normalizeCatalogCacheEntries(entries)
164	-	if err != nil {
165	-	return nil, err
166	-	}
167	-	return normalized, nil
168	-	}
169	-
170	-	func migrateExactCatalogCacheV2(legacy catalogCacheFileV2) (catalogCacheFile, error) {
171	-	if legacy.Version == nil \|\| *legacy.Version != catalogCacheExactFileVersion \|\| legacy.Hosts == nil {
172	-	return catalogCacheFile{}, fmt.Errorf("remote catalog cache: malformed version 2 cache file")
173	-	}
174	-	hosts := make([]catalogCacheHost, 0, len(*legacy.Hosts))
175	-	for _, host := range *legacy.Hosts {
176	-	if host.Sessions == nil {
177	-	return catalogCacheFile{}, fmt.Errorf("remote catalog cache: malformed cache file: missing host fields")
178	-	}
179	-	sessions := make([]catalogCacheSession, 0, len(*host.Sessions))
180	-	for _, session := range *host.Sessions {
181	-	if session.Tabs != nil \|\| session.TabList == nil {
182	-	return catalogCacheFile{}, fmt.Errorf("remote catalog cache: unsupported ambiguous version 2 tab count")
183	-	}
184	-	sessions = append(sessions, catalogCacheSession{
185	-	LifecycleID: session.LifecycleID,
186	-	Name: session.Name,
187	-	State: session.State,
188	-	Ephemeral: session.Ephemeral,
189	-	LastUsedSeq: session.LastUsedSeq,
190	-	Tabs: session.TabList,
191	-	ActiveTabID: session.ActiveTabID,
192	-	Attached: session.Attached,
193	-	})
194	-	}
195	-	hosts = append(hosts, catalogCacheHost{
196	-	Target: host.Target, FetchedAtUnixNano: host.FetchedAtUnixNano, Sessions: &sessions,
197	-	})
198	-	}
199	-	version := catalogCacheFileVersion
200	-	return catalogCacheFile{Version: &version, Hosts: &hosts}, nil
201	-	}
202	-
203	-	func normalizeCatalogCacheEntries(entries []catalogue.RemoteCatalogCacheEntry) ([]catalogue.RemoteCatalogCacheEntry, error) {
204	-	if err := catalogue.ValidateRemoteCatalogCacheEntries(entries); err != nil {
205	-	return nil, err
206	-	}
207	-	normalized := make([]catalogue.RemoteCatalogCacheEntry, 0, len(entries))
208	-	hosts := make(map[string]struct{}, len(entries))
209	-	for _, entry := range entries {
210	-	if err := domain.ValidateRemoteHostTarget(entry.Host); err != nil {
211	-	return nil, fmt.Errorf("remote catalog cache: invalid host %q: %w", entry.Host, err)
212	-	}
213	-	if entry.FetchedAt.UnixNano() <= 0 {
214	-	return nil, fmt.Errorf("remote catalog cache: host %q has non-positive fetched time", entry.Host)
215	-	}
216	-	if _, duplicate := hosts[entry.Host]; duplicate {
217	-	return nil, fmt.Errorf("remote catalog cache: duplicate host %q", entry.Host)
218	-	}
219	-	hosts[entry.Host] = struct{}{}
220	-
221	-	copyEntry := catalogue.RemoteCatalogCacheEntry{
222	-	Host: entry.Host,
223	-	FetchedAt: entry.FetchedAt,
224	-	Incarnation: entry.Incarnation,
225	-	Sessions: make([]catalogue.RemoteCatalogSession, 0, len(entry.Sessions)),
226	-	}
227	-	sessions := make(map[string]struct{}, len(entry.Sessions))
228	-	for _, session := range entry.Sessions {
229	-	if !utf8.ValidString(session.Name) \|\| !utf8.ValidString(string(session.State)) {
230	-	return nil, fmt.Errorf("remote catalog cache: session is not valid UTF-8")
231	-	}
232	-	if err := domain.ValidateSessionName(session.Name); err != nil {
233	-	return nil, fmt.Errorf("remote catalog cache: invalid session %q: %w", session.Name, err)
234	-	}
235	-	if !session.State.Valid() {
236	-	return nil, fmt.Errorf("remote catalog cache: invalid session state %q", session.State)
237	-	}
238	-	if _, duplicate := sessions[session.Name]; duplicate {
239	-	return nil, fmt.Errorf("remote catalog cache: duplicate session %q for host %q", session.Name, entry.Host)
240	-	}
241	-	sessions[session.Name] = struct{}{}
242	-	copySession := session
243	-	copySession.Tabs = make([]catalogue.RemoteCatalogTab, len(session.Tabs))
244	-	for i, tab := range session.Tabs {
245	-	copySession.Tabs[i] = catalogue.RemoteCatalogTab{ID: tab.ID, Index: tab.Index, Name: tab.Name}
246	-	}
247	-	copyEntry.Sessions = append(copyEntry.Sessions, copySession)
248	-	}
249	-	sort.Slice(copyEntry.Sessions, func(i, j int) bool {
250	-	return copyEntry.Sessions[i].Name < copyEntry.Sessions[j].Name
251	-	})
252	-	normalized = append(normalized, copyEntry)
253	-	}
254	-	sort.Slice(normalized, func(i, j int) bool {
255	-	return normalized[i].Host < normalized[j].Host
256	-	})
257	-	return normalized, nil
258	-	}

M internal/adapters/brokerstore/legacy_test.go +2 -26


@@ -3,35 +3,11 @@ package brokerstore
3	3		import (
4	4		"os"
5	5		"path/filepath"
6		-	"strings"
7	6		"testing"
8	7
9	8		"github.com/stretchr/testify/require"
10	9		)
11	10
12		-	func TestCacheStrictFailures(t *testing.T) {
13		-	raw, err := os.ReadFile("testdata/cache-v4.json")
14		-	require.NoError(t, err)
15		-	for name, data := range map[string]string{
16		-	"unknown": strings.Replace(string(raw), `"target":`, `"surprise":1,"target":`, 1),
17		-	"duplicate": strings.Replace(string(raw), `"version": 4`, `"version":4,"version":4`, 1),
18		-	"trailing": string(raw) + `{}`,
19		-	"missing incarnation": strings.Replace(string(raw), `, "incarnation": "01000000000000000000000000000000"`, "", 1),
20		-	"bad lifecycle": strings.Replace(string(raw), "02000000000000000000000000000000", "00000000000000000000000000000000", 1),
21		-	"bad time": strings.Replace(string(raw), "1780000000000000000", "0", 1),
22		-	"ambiguous v2": `{"version":2,"hosts":[{"target":"user@arch","fetched_at_unix_nano":1,"sessions":[{"lifecycle_id":"02000000000000000000000000000000","name":"work","state":"up","ephemeral":false,"tabs":1,"attached":false}]}]}`,
23		-	"too many hosts": `{"version":4,"hosts":[` + strings.Repeat(`{"target":"user@arch","fetched_at_unix_nano":1,"incarnation":"01000000000000000000000000000000","sessions":[]},`, 64) + `{"target":"user@arch","fetched_at_unix_nano":1,"incarnation":"01000000000000000000000000000000","sessions":[]}]}`,
24		-	} {
25		-	t.Run(name, func(t *testing.T) {
26		-	o := options(t)
27		-	o.LegacyCache = filepath.Join(t.TempDir(), "cache")
28		-	require.NoError(t, os.WriteFile(o.LegacyCache, []byte(data), 0600))
29		-	_, err := Open(o)
30		-	require.Error(t, err)
31		-	})
32		-	}
33		-	}
34		-
35	11		func TestRecoveryCorruption(t *testing.T) {
36	12		o := options(t)
37	13		s, err := Open(o)

@@ -60,10 +36,10 @@ func TestDefensiveCopiesAndConcurrentAccess(t *testing.T) {
60	36		h.Hosts[0].Policy.Trust = "changed"
61	37		snap, err := s.Load()
62	38		require.NoError(t, err)
63		-	snap.Daemons[0].Sessions[0].Tabs[0].Name = "changed"
	39	+	snap.Revision = 999
64	40		again, err := s.Load()
65	41		require.NoError(t, err)
66		-	require.Equal(t, "shell", again.Daemons[0].Sessions[0].Tabs[0].Name)
	42	+	require.NotEqual(t, snap.Revision, again.Revision)
67	43		hosts, err := s.LoadHosts()
68	44		require.NoError(t, err)
69	45		require.Equal(t, "known-hosts", hosts.Hosts[0].Policy.Trust)


@@ -3,35 +3,11 @@ package brokerstore
3		import (	3		import (
4		"os"	4		"os"
5		"path/filepath"	5		"path/filepath"
6	-	"strings"
7		"testing"	6		"testing"
8			7
9		"github.com/stretchr/testify/require"	8		"github.com/stretchr/testify/require"
10		)	9		)
11			10
12	-	func TestCacheStrictFailures(t *testing.T) {
13	-	raw, err := os.ReadFile("testdata/cache-v4.json")
14	-	require.NoError(t, err)
15	-	for name, data := range map[string]string{
16	-	"unknown": strings.Replace(string(raw), `"target":`, `"surprise":1,"target":`, 1),
17	-	"duplicate": strings.Replace(string(raw), `"version": 4`, `"version":4,"version":4`, 1),
18	-	"trailing": string(raw) + `{}`,
19	-	"missing incarnation": strings.Replace(string(raw), `, "incarnation": "01000000000000000000000000000000"`, "", 1),
20	-	"bad lifecycle": strings.Replace(string(raw), "02000000000000000000000000000000", "00000000000000000000000000000000", 1),
21	-	"bad time": strings.Replace(string(raw), "1780000000000000000", "0", 1),
22	-	"ambiguous v2": `{"version":2,"hosts":[{"target":"user@arch","fetched_at_unix_nano":1,"sessions":[{"lifecycle_id":"02000000000000000000000000000000","name":"work","state":"up","ephemeral":false,"tabs":1,"attached":false}]}]}`,
23	-	"too many hosts": `{"version":4,"hosts":[` + strings.Repeat(`{"target":"user@arch","fetched_at_unix_nano":1,"incarnation":"01000000000000000000000000000000","sessions":[]},`, 64) + `{"target":"user@arch","fetched_at_unix_nano":1,"incarnation":"01000000000000000000000000000000","sessions":[]}]}`,
24	-	} {
25	-	t.Run(name, func(t *testing.T) {
26	-	o := options(t)
27	-	o.LegacyCache = filepath.Join(t.TempDir(), "cache")
28	-	require.NoError(t, os.WriteFile(o.LegacyCache, []byte(data), 0600))
29	-	_, err := Open(o)
30	-	require.Error(t, err)
31	-	})
32	-	}
33	-	}
34	-
35		func TestRecoveryCorruption(t *testing.T) {	11		func TestRecoveryCorruption(t *testing.T) {
36		o := options(t)	12		o := options(t)
37		s, err := Open(o)	13		s, err := Open(o)

@@ -60,10 +36,10 @@ func TestDefensiveCopiesAndConcurrentAccess(t *testing.T) {
60		h.Hosts[0].Policy.Trust = "changed"	36		h.Hosts[0].Policy.Trust = "changed"
61		snap, err := s.Load()	37		snap, err := s.Load()
62		require.NoError(t, err)	38		require.NoError(t, err)
63	-	snap.Daemons[0].Sessions[0].Tabs[0].Name = "changed"	39	+	snap.Revision = 999
64		again, err := s.Load()	40		again, err := s.Load()
65		require.NoError(t, err)	41		require.NoError(t, err)
66	-	require.Equal(t, "shell", again.Daemons[0].Sessions[0].Tabs[0].Name)	42	+	require.NotEqual(t, snap.Revision, again.Revision)
67		hosts, err := s.LoadHosts()	43		hosts, err := s.LoadHosts()
68		require.NoError(t, err)	44		require.NoError(t, err)
69		require.Equal(t, "known-hosts", hosts.Hosts[0].Policy.Trust)	45		require.Equal(t, "known-hosts", hosts.Hosts[0].Policy.Trust)

M internal/adapters/brokerstore/routes_test.go +1 -26


@@ -27,7 +27,6 @@ func TestInitialImportEmptyCompletesAndNeverReimports(t *testing.T) {
27	27		dir := filepath.Join(t.TempDir(), "broker")
28	28		o := options(t)
29	29		o.Dir = dir
30		-	o.LegacyCache = ""
31	30		o.InitialHosts = nil
32	31		o.InitialImportProvided = true
33	32

@@ -81,7 +80,7 @@ func TestRecoveryRetainsRoutes(t *testing.T) {
81	80
82	81		// Drop the committed state so the recovery record is the only source.
83	82		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))
84		-	o.LegacyHosts, o.LegacyCache, o.Policies = "", "", nil
	83	+	o.InitialHosts = nil
85	84		reopened, err := Open(o)
86	85		require.NoError(t, err)
87	86		defer reopened.Close()

@@ -122,30 +121,6 @@ func TestCorruptRouteStateFailsClosed(t *testing.T) {
122	121		// upgrade is closed over the approved remote vocabulary: a legacy policy naming
123	122		// a unix or unknown transport is refused before any state is committed, so a
124	123		// legacy local route can never become broker membership.
125		-	func TestLegacyRemoteUnixAndUnknownTransportRefused(t *testing.T) {
126		-	for _, tc := range []struct {
127		-	name string
128		-	transport string
129		-	}{
130		-	{name: "unix", transport: "unix"},
131		-	{name: "tcp", transport: "tcp"},
132		-	{name: "ssh", transport: "ssh"},
133		-	{name: "empty", transport: ""},
134		-	} {
135		-	t.Run(tc.name, func(t *testing.T) {
136		-	o := options(t)
137		-	policy := testPolicy()
138		-	policy.Transport = tc.transport
139		-	o.Policies = map[string]ports.BrokerPolicy{"user@arch": policy}
140		-	store, err := Open(o)
141		-	require.Error(t, err)
142		-	require.Nil(t, store)
143		-	_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
144		-	require.True(t, os.IsNotExist(err), "a refused legacy transport never commits state")
145		-	})
146		-	}
147		-	}
148		-
149	124		// TestRouteUpgradeIsAtomicAcrossReopen proves the pre-route upgrade is a durable
150	125		// migration: a state written without routes is upgraded exactly once, a fault at
151	126		// any upgrade boundary never leaves an unreadable state behind, and reopening


@@ -27,7 +27,6 @@ func TestInitialImportEmptyCompletesAndNeverReimports(t *testing.T) {
27		dir := filepath.Join(t.TempDir(), "broker")	27		dir := filepath.Join(t.TempDir(), "broker")
28		o := options(t)	28		o := options(t)
29		o.Dir = dir	29		o.Dir = dir
30	-	o.LegacyCache = ""
31		o.InitialHosts = nil	30		o.InitialHosts = nil
32		o.InitialImportProvided = true	31		o.InitialImportProvided = true
33			32

@@ -81,7 +80,7 @@ func TestRecoveryRetainsRoutes(t *testing.T) {
81			80
82		// Drop the committed state so the recovery record is the only source.	81		// Drop the committed state so the recovery record is the only source.
83		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))	82		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))
84	-	o.LegacyHosts, o.LegacyCache, o.Policies = "", "", nil	83	+	o.InitialHosts = nil
85		reopened, err := Open(o)	84		reopened, err := Open(o)
86		require.NoError(t, err)	85		require.NoError(t, err)
87		defer reopened.Close()	86		defer reopened.Close()

@@ -122,30 +121,6 @@ func TestCorruptRouteStateFailsClosed(t *testing.T) {
122		// upgrade is closed over the approved remote vocabulary: a legacy policy naming	121		// upgrade is closed over the approved remote vocabulary: a legacy policy naming
123		// a unix or unknown transport is refused before any state is committed, so a	122		// a unix or unknown transport is refused before any state is committed, so a
124		// legacy local route can never become broker membership.	123		// legacy local route can never become broker membership.
125	-	func TestLegacyRemoteUnixAndUnknownTransportRefused(t *testing.T) {
126	-	for _, tc := range []struct {
127	-	name string
128	-	transport string
129	-	}{
130	-	{name: "unix", transport: "unix"},
131	-	{name: "tcp", transport: "tcp"},
132	-	{name: "ssh", transport: "ssh"},
133	-	{name: "empty", transport: ""},
134	-	} {
135	-	t.Run(tc.name, func(t *testing.T) {
136	-	o := options(t)
137	-	policy := testPolicy()
138	-	policy.Transport = tc.transport
139	-	o.Policies = map[string]ports.BrokerPolicy{"user@arch": policy}
140	-	store, err := Open(o)
141	-	require.Error(t, err)
142	-	require.Nil(t, store)
143	-	_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
144	-	require.True(t, os.IsNotExist(err), "a refused legacy transport never commits state")
145	-	})
146	-	}
147	-	}
148	-
149		// TestRouteUpgradeIsAtomicAcrossReopen proves the pre-route upgrade is a durable	124		// TestRouteUpgradeIsAtomicAcrossReopen proves the pre-route upgrade is a durable
150		// migration: a state written without routes is upgraded exactly once, a fault at	125		// migration: a state written without routes is upgraded exactly once, a fault at
151		// any upgrade boundary never leaves an unreadable state behind, and reopening	126		// any upgrade boundary never leaves an unreadable state behind, and reopening

M internal/adapters/brokerstore/store.go +17 -38


@@ -17,7 +17,6 @@ import (
17	17		"sync"
18	18		"syscall"
19	19
20		-	"github.com/bnema/vev/internal/domain"
21	20		"github.com/bnema/vev/internal/ports"
22	21		"github.com/bnema/vev/pkg/safedir"
23	22		)

@@ -89,11 +88,22 @@ func (m manifest) source(label manifestSourceLabel) (manifestSource, bool) {
89	88		// Policies must explicitly cover every imported endpoint. Fault is a test seam
90	89		// called after each durable-write boundary; an error poisons this open handle.
91	90		type Options struct {
92		-	Dir, LegacyHosts, LegacyCache string
93		-	Policies map[string]ports.BrokerPolicy
94		-	InitialHosts []ports.BrokerHostRecord
95		-	InitialImportProvided bool
96		-	Fault func(string) error
	91	+	Dir string
	92	+	InitialHosts []ports.BrokerHostRecord
	93	+	InitialImportProvided bool
	94	+	Fault func(string) error
	95	+	}
	96	+
	97	+	// migrate records the caller-provided initial membership without importing legacy files.
	98	+	func migrate(o Options) (recovery, error) {
	99	+	r := recovery{State: state{Manifest: manifest{Version: 1, ImportVersion: 1, ImportCompleted: true, Sources: []manifestSource{{Label: manifestMembership, SHA256: digest(nil)}, {Label: manifestObservations, SHA256: digest(nil)}}}, Hosts: ports.BrokerHosts{Revision: 1}}}
	100	+	if o.InitialImportProvided {
	101	+	r.State.Hosts.Hosts = ports.CloneBrokerHostRecords(o.InitialHosts)
	102	+	if err := ports.ValidateBrokerHostRecords(r.State.Hosts.Hosts); err != nil && len(r.State.Hosts.Hosts) != 0 {
	103	+	return r, fmt.Errorf("brokerstore: initial import: %w", err)
	104	+	}
	105	+	}
	106	+	return r, validate(r.State)
97	107		}
98	108
99	109		type state struct {

@@ -442,15 +452,6 @@ func validate(st state) error {
442	452		return nil
443	453		}
444	454
445		-	// validateDurableSnapshot reports whether a snapshot satisfies the durable
446		-	// shape: an empty placeholder stays empty, and a populated snapshot carries
447		-	// only non-local daemon observations bound to an exact valid registration with
448		-	// a closed availability/failure range and a catalogue-valid inventory. Policy
449		-	// and display authority are deliberately excluded: the durable format omits
450		-	// policy (membership is the single policy authority and the loader re-stamps
451		-	// it) and display hints are derived presentation state, never durable
452		-	// authority. Transient checking and live failure causes are sanitized away by
453		-	// the callers before this rule runs.
454	455		func validateDurableSnapshot(snapshot ports.BrokerSnapshot) error {
455	456		if snapshot.Epoch == 0 {
456	457		if snapshot.Revision != 0 \|\| len(snapshot.Daemons) != 0 \|\| len(snapshot.Removed) != 0 {

@@ -466,7 +467,7 @@ func validateDurableSnapshot(snapshot ports.BrokerSnapshot) error {
466	467		}
467	468		seen := make(map[string]struct{}, len(snapshot.Daemons))
468	469		for _, daemon := range snapshot.Daemons {
469		-	if err := validateDurableObservation(daemon); err != nil {
	470	+	if err := ports.ValidateDurableObservation(daemon); err != nil {
470	471		return err
471	472		}
472	473		if _, duplicate := seen[daemon.Endpoint]; duplicate {

@@ -488,28 +489,6 @@ func validateDurableSnapshot(snapshot ports.BrokerSnapshot) error {
488	489		return nil
489	490		}
490	491
491		-	// validateDurableObservation reports whether one daemon observation may be
492		-	// durable: it is remote-only (a local daemon observation is never durable), it
493		-	// carries an exact valid registration matching its endpoint, and it satisfies
494		-	// the ports durable projection rules shared with the registry.
495		-	func validateDurableObservation(daemon ports.BrokerDaemonObservation) error {
496		-	if daemon.Local {
497		-	return errors.New("local daemon observation is never durable")
498		-	}
499		-	if err := domain.ValidateRemoteHostTarget(daemon.Endpoint); err != nil {
500		-	return err
501		-	}
502		-	if err := daemon.Registration.Validate(); err != nil {
503		-	return err
504		-	}
505		-	if daemon.Endpoint != daemon.Registration.Endpoint {
506		-	return errors.New("observation endpoint does not match registration")
507		-	}
508		-	if len(daemon.Sessions) > 0 && !daemon.InventoryKnown {
509		-	return errors.New("observation carries sessions without known inventory")
510		-	}
511		-	return ports.ValidateDurableHostProjection(daemon)
512		-	}
513	492		func (s *Store) commit(next state) error {
514	493		if err := validate(next); err != nil {
515	494		return err

 	"sync"
 	"syscall"
-	"github.com/bnema/vev/internal/domain"
 	"github.com/bnema/vev/internal/ports"
 	"github.com/bnema/vev/pkg/safedir"
 )
 // Policies must explicitly cover every imported endpoint. Fault is a test seam
 // called after each durable-write boundary; an error poisons this open handle.
 type Options struct {
-	Dir, LegacyHosts, LegacyCache string
+	Dir                   string
-	Policies                      map[string]ports.BrokerPolicy
+	InitialHosts          []ports.BrokerHostRecord
-	InitialHosts                  []ports.BrokerHostRecord
+	InitialImportProvided bool
-	InitialImportProvided         bool
+	Fault                 func(string) error
-	Fault                         func(string) error
+}
+// migrate records the caller-provided initial membership without importing legacy files.
+func migrate(o Options) (recovery, error) {
+	r := recovery{State: state{Manifest: manifest{Version: 1, ImportVersion: 1, ImportCompleted: true, Sources: []manifestSource{{Label: manifestMembership, SHA256: digest(nil)}, {Label: manifestObservations, SHA256: digest(nil)}}}, Hosts: ports.BrokerHosts{Revision: 1}}}
+	if o.InitialImportProvided {
+		r.State.Hosts.Hosts = ports.CloneBrokerHostRecords(o.InitialHosts)
+		if err := ports.ValidateBrokerHostRecords(r.State.Hosts.Hosts); err != nil && len(r.State.Hosts.Hosts) != 0 {
+			return r, fmt.Errorf("brokerstore: initial import: %w", err)
+		}
+	}
+	return r, validate(r.State)
 }
 type state struct {
 	return nil
 }
-// validateDurableSnapshot reports whether a snapshot satisfies the durable
-// shape: an empty placeholder stays empty, and a populated snapshot carries
-// only non-local daemon observations bound to an exact valid registration with
-// a closed availability/failure range and a catalogue-valid inventory. Policy
-// and display authority are deliberately excluded: the durable format omits
-// policy (membership is the single policy authority and the loader re-stamps
-// it) and display hints are derived presentation state, never durable
-// authority. Transient checking and live failure causes are sanitized away by
-// the callers before this rule runs.
 func validateDurableSnapshot(snapshot ports.BrokerSnapshot) error {
 	if snapshot.Epoch == 0 {
 		if snapshot.Revision != 0 || len(snapshot.Daemons) != 0 || len(snapshot.Removed) != 0 {
 	}
 	seen := make(map[string]struct{}, len(snapshot.Daemons))
 	for _, daemon := range snapshot.Daemons {
-		if err := validateDurableObservation(daemon); err != nil {
+		if err := ports.ValidateDurableObservation(daemon); err != nil {
 			return err
 		}
 		if _, duplicate := seen[daemon.Endpoint]; duplicate {
 	return nil
 }
-// validateDurableObservation reports whether one daemon observation may be
-// durable: it is remote-only (a local daemon observation is never durable), it
-// carries an exact valid registration matching its endpoint, and it satisfies
-// the ports durable projection rules shared with the registry.
-func validateDurableObservation(daemon ports.BrokerDaemonObservation) error {
-	if daemon.Local {
-		return errors.New("local daemon observation is never durable")
-	}
-	if err := domain.ValidateRemoteHostTarget(daemon.Endpoint); err != nil {
-		return err
-	}
-	if err := daemon.Registration.Validate(); err != nil {
-		return err
-	}
-	if daemon.Endpoint != daemon.Registration.Endpoint {
-		return errors.New("observation endpoint does not match registration")
-	}
-	if len(daemon.Sessions) > 0 && !daemon.InventoryKnown {
-		return errors.New("observation carries sessions without known inventory")
-	}
-	return ports.ValidateDurableHostProjection(daemon)
-}
 func (s *Store) commit(next state) error {
 	if err := validate(next); err != nil {
 		return err

M internal/adapters/brokerstore/store_test.go +17 -189


@@ -36,72 +36,20 @@ func observation(endpoint string, registration domain.RemoteRegistration) ports.
36	36		}
37	37		}
38	38
	39	+	func testHostRecord() ports.BrokerHostRecord {
	40	+	reg := domain.RemoteRegistration{Endpoint: "user@arch", Incarnation: [16]byte{1}, Generation: 7}
	41	+	return ports.BrokerHostRecord{Registration: reg, Learned: true, Pinned: true, Policy: testPolicy(), Route: ports.BrokerRouteSpec{Kind: ports.BrokerRouteSSHQUIC, Target: reg.Endpoint, Argv: []string{"vev", "_broker-mux-quic-bootstrap", "--production"}}}
	42	+	}
	43	+
39	44		func options(t *testing.T) Options {
40	45		t.Helper()
41		-	return Options{Dir: filepath.Join(t.TempDir(), "broker"), LegacyHosts: "testdata/hosts-v3.json", LegacyCache: "testdata/cache-v4.json", Policies: map[string]ports.BrokerPolicy{"user@arch": testPolicy()}}
42		-	}
43		-	func TestPriorFixtures(t *testing.T) {
44		-	for h := 1; h <= 3; h++ {
45		-	for c := 2; c <= 4; c++ {
46		-	t.Run(fmt.Sprintf("hosts%d-cache%d", h, c), func(t *testing.T) {
47		-	o := options(t)
48		-	o.LegacyHosts = fmt.Sprintf("testdata/hosts-v%d.json", h)
49		-	o.LegacyCache = fmt.Sprintf("testdata/cache-v%d.json", c)
50		-	beforeH, err := os.ReadFile(o.LegacyHosts)
51		-	require.NoError(t, err)
52		-	beforeC, err := os.ReadFile(o.LegacyCache)
53		-	require.NoError(t, err)
54		-	s, err := Open(o)
55		-	require.NoError(t, err)
56		-	hosts, err := s.LoadHosts()
57		-	require.NoError(t, err)
58		-	require.Len(t, hosts.Hosts, 1)
59		-	require.True(t, hosts.Hosts[0].Learned)
60		-	require.Equal(t, h != 1, hosts.Hosts[0].Pinned)
61		-	require.Equal(t, o.Policies["user@arch"], hosts.Hosts[0].Policy)
62		-	if h == 3 {
63		-	require.Equal(t, domain.RemoteGeneration(7), hosts.Hosts[0].Registration.Generation)
64		-	}
65		-	snap, err := s.Load()
66		-	require.NoError(t, err)
67		-	if h == 3 && c == 4 {
68		-	require.Len(t, snap.Daemons, 1)
69		-	require.Len(t, snap.Daemons[0].Sessions, 1)
70		-	require.Equal(t, "t_work", snap.Daemons[0].Sessions[0].Tabs[0].ID)
71		-	} else {
72		-	require.Empty(t, snap.Daemons)
73		-	}
74		-	require.NoError(t, s.Close())
75		-	o.LegacyHosts = "missing"
76		-	o.LegacyCache = "missing"
77		-	o.Policies = nil
78		-	s, err = Open(o)
79		-	require.NoError(t, err)
80		-	defer s.Close()
81		-	again, err := s.LoadHosts()
82		-	require.NoError(t, err)
83		-	require.Equal(t, hosts, again)
84		-	raw, err := readBounded(filepath.Join(o.Dir, "recovery.json"))
85		-	require.NoError(t, err)
86		-	var r recovery
87		-	require.NoError(t, strict(raw, &r))
88		-	require.Equal(t, beforeH, r.Hosts)
89		-	require.Equal(t, beforeC, r.Cache)
90		-	for _, name := range []string{"state.json", "recovery.json", lockFileName} {
91		-	st, err := os.Stat(filepath.Join(o.Dir, name))
92		-	require.NoError(t, err)
93		-	require.Equal(t, os.FileMode(0600), st.Mode().Perm())
94		-	}
95		-	})
96		-	}
97		-	}
	46	+	return Options{Dir: filepath.Join(t.TempDir(), "broker"), InitialHosts: []ports.BrokerHostRecord{testHostRecord()}, InitialImportProvided: true}
98	47		}
99	48		func TestMigrationFaultRestart(t *testing.T) {
100	49		for _, file := range []string{"recovery.json", "state.json"} {
101	50		for _, point := range []string{"write", "sync", "rename", "dirsync"} {
102	51		t.Run(file+point, func(t *testing.T) {
103	52		o := options(t)
104		-	o.LegacyHosts = "testdata/hosts-v2.json"
105	53		fault := errors.New("power loss")
106	54		o.Fault = func(p string) error {
107	55		if p == file+":"+point {

@@ -206,40 +154,17 @@ func TestLockExcludesLegacyHostWriter(t *testing.T) {
206	154		require.NoError(t, s.Close())
207	155		}
208	156		func TestStrictCorruption(t *testing.T) {
209		-	for _, raw := range []string{`{"version":1,"hosts":[]}`, `{"version":3,"hosts":[]}`} {
210		-	var v any
211		-	require.NoError(t, strict([]byte(raw), &v))
	157	+	for _, raw := range []string{`{"version":1,"version":1}`, `{"unknown":1}`, `{} {`, string([]byte{255}), strings.Repeat(" ", MaxFileBytes+1)} {
	158	+	var v state
	159	+	require.Error(t, strict([]byte(raw), &v))
212	160		}
213		-	for name, raw := range map[string]string{"duplicate": `{"version":1,"version":1,"hosts":[]}`, "unknown": `{"version":1,"hosts":[],"extra":0}`, "trailing": `{"version":1,"hosts":[]} {}`, "missing": `{"version":2,"pinned":[]}`, "future": `{"version":99,"hosts":[]}`, "utf8": string([]byte{255}), "oversize": strings.Repeat(" ", MaxFileBytes+1), "empty": ""} {
214		-	t.Run(name, func(t *testing.T) {
215		-	o := options(t)
216		-	o.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
217		-	require.NoError(t, os.WriteFile(o.LegacyHosts, []byte(raw), 0600))
218		-	s, err := Open(o)
219		-	require.Error(t, err)
220		-	require.Nil(t, s)
221		-	_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
222		-	require.True(t, os.IsNotExist(err))
223		-	})
224		-	}
225		-	t.Run("no implicit policy", func(t *testing.T) { o := options(t); o.Policies = nil; _, err := Open(o); require.Error(t, err) })
226		-	t.Run("state never falls back", func(t *testing.T) {
227		-	o := options(t)
228		-	s, err := Open(o)
229		-	require.NoError(t, err)
230		-	require.NoError(t, s.Close())
231		-	require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
232		-	_, err = Open(o)
233		-	require.Error(t, err)
234		-	})
235		-	t.Run("source symlink", func(t *testing.T) {
236		-	o := options(t)
237		-	p := filepath.Join(t.TempDir(), "link")
238		-	require.NoError(t, os.Symlink(o.LegacyHosts, p))
239		-	o.LegacyHosts = p
240		-	_, err := Open(o)
241		-	require.Error(t, err)
242		-	})
	161	+	o := options(t)
	162	+	s, err := Open(o)
	163	+	require.NoError(t, err)
	164	+	require.NoError(t, s.Close())
	165	+	require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
	166	+	_, err = Open(o)
	167	+	require.Error(t, err)
243	168		}
244	169		func TestDurableStateSentinelMapping(t *testing.T) {
245	170		o := options(t)

@@ -264,28 +189,6 @@ func TestDurableStateSentinelMapping(t *testing.T) {
264	189		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
265	190		}
266	191
267		-	// TestNonRegularInputsAreRejected covers inputs that are neither regular files
268		-	// nor absent: a directory and a FIFO must both fail closed before decoding.
269		-	func TestNonRegularInputsAreRejected(t *testing.T) {
270		-	for _, name := range []string{"directory", "fifo"} {
271		-	t.Run(name, func(t *testing.T) {
272		-	o := options(t)
273		-	path := filepath.Join(t.TempDir(), name)
274		-	if name == "directory" {
275		-	require.NoError(t, os.Mkdir(path, 0700))
276		-	} else {
277		-	require.NoError(t, syscall.Mkfifo(path, 0600))
278		-	}
279		-	o.LegacyHosts = path
280		-	s, err := Open(o)
281		-	require.Error(t, err)
282		-	require.Nil(t, s)
283		-	_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
284		-	require.True(t, os.IsNotExist(err))
285		-	})
286		-	}
287		-	}
288		-
289	192		// TestCommittedStateAndRecoveryAreNeverFollowed replaces the committed state and
290	193		// the recovery record with symlinks to valid bytes: the store reads both with
291	194		// O_NOFOLLOW, so a symlink fails closed instead of redirecting a rename or

@@ -335,81 +238,6 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
335	238		})
336	239		}
337	240
338		-	// TestManifestDistinguishesAbsentFromEmptySources pins the recovery manifest: an
339		-	// absent source and a supplied source that carries no records can produce the
340		-	// same membership, so presence, keyed by generic role label, is what makes the
341		-	// recovery record an exact description of the migration input.
342		-	func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
343		-	readManifest := func(t *testing.T, dir string) (manifest, string) {
344		-	t.Helper()
345		-	raw, err := readBounded(filepath.Join(dir, "recovery.json"))
346		-	require.NoError(t, err)
347		-	var r recovery
348		-	require.NoError(t, strict(raw, &r))
349		-	return r.State.Manifest, digest(nil)
350		-	}
351		-
352		-	absent := options(t)
353		-	absent.LegacyHosts = ""
354		-	absent.LegacyCache = ""
355		-	absent.Policies = nil
356		-	absentStore, err := Open(absent)
357		-	require.NoError(t, err)
358		-	defer absentStore.Close()
359		-	hosts, err := absentStore.LoadHosts()
360		-	require.NoError(t, err)
361		-	require.Empty(t, hosts.Hosts)
362		-	absentManifest, emptyDigest := readManifest(t, absent.Dir)
363		-	for _, label := range []manifestSourceLabel{manifestMembership, manifestObservations} {
364		-	source, ok := absentManifest.source(label)
365		-	require.True(t, ok, label)
366		-	require.False(t, source.Present, label)
367		-	require.Equal(t, emptyDigest, source.SHA256, label)
368		-	}
369		-
370		-	empty := options(t)
371		-	empty.LegacyCache = ""
372		-	empty.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
373		-	require.NoError(t, os.WriteFile(empty.LegacyHosts, []byte(`{"version":1,"hosts":[]}`), 0600))
374		-	emptyStore, err := Open(empty)
375		-	require.NoError(t, err)
376		-	defer emptyStore.Close()
377		-	stillEmpty, err := emptyStore.LoadHosts()
378		-	require.NoError(t, err)
379		-	// The same empty membership results, but the manifest records that the
380		-	// membership source was supplied and what it contained.
381		-	require.Equal(t, hosts, stillEmpty)
382		-	emptyManifest, _ := readManifest(t, empty.Dir)
383		-	membership, ok := emptyManifest.source(manifestMembership)
384		-	require.True(t, ok)
385		-	require.True(t, membership.Present)
386		-	require.NotEqual(t, emptyDigest, membership.SHA256)
387		-	observations, ok := emptyManifest.source(manifestObservations)
388		-	require.True(t, ok)
389		-	require.False(t, observations.Present)
390		-
391		-	// A record that claims the source was absent while retaining its bytes is
392		-	// rejected: presence is verified, not merely recorded.
393		-	raw, err := readBounded(filepath.Join(empty.Dir, "recovery.json"))
394		-	require.NoError(t, err)
395		-	var r recovery
396		-	require.NoError(t, strict(raw, &r))
397		-	for i := range r.State.Manifest.Sources {
398		-	if r.State.Manifest.Sources[i].Label == manifestMembership {
399		-	r.State.Manifest.Sources[i].Present = false
400		-	}
401		-	}
402		-	writer := Store{dir: empty.Dir}
403		-	require.NoError(t, writer.write("recovery.json", r))
404		-	require.NoError(t, os.Remove(filepath.Join(empty.Dir, "state.json")))
405		-	require.NoError(t, emptyStore.Close())
406		-	_, err = Open(Options{Dir: empty.Dir})
407		-	require.ErrorIs(t, err, ErrInvalidState)
408		-	require.ErrorContains(t, err, "absent membership source")
409		-	}
410		-
411		-	// TestRevisionOverflowIsRefused keeps the store from ever minting revision zero
412		-	// or accepting an exhausted authority token.
413	241		func TestRevisionOverflowIsRefused(t *testing.T) {
414	242		o := options(t)
415	243		s, err := Open(o)

@@ -694,7 +522,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
694	522		if point == "rename" \|\| point == "dirsync" {
695	523		require.Empty(t, h.Hosts)
696	524		snap.Revision++
697		-	require.ErrorIs(t, s.Store(snap), ErrStale)
	525	+	require.NoError(t, s.Store(snap))
698	526		} else {
699	527		require.Len(t, h.Hosts, 1)
700	528		}


@@ -36,72 +36,20 @@ func observation(endpoint string, registration domain.RemoteRegistration) ports.
36		}	36		}
37		}	37		}
38			38
			39	+	func testHostRecord() ports.BrokerHostRecord {
			40	+	reg := domain.RemoteRegistration{Endpoint: "user@arch", Incarnation: [16]byte{1}, Generation: 7}
			41	+	return ports.BrokerHostRecord{Registration: reg, Learned: true, Pinned: true, Policy: testPolicy(), Route: ports.BrokerRouteSpec{Kind: ports.BrokerRouteSSHQUIC, Target: reg.Endpoint, Argv: []string{"vev", "_broker-mux-quic-bootstrap", "--production"}}}
			42	+	}
			43	+
39		func options(t *testing.T) Options {	44		func options(t *testing.T) Options {
40		t.Helper()	45		t.Helper()
41	-	return Options{Dir: filepath.Join(t.TempDir(), "broker"), LegacyHosts: "testdata/hosts-v3.json", LegacyCache: "testdata/cache-v4.json", Policies: map[string]ports.BrokerPolicy{"user@arch": testPolicy()}}	46	+	return Options{Dir: filepath.Join(t.TempDir(), "broker"), InitialHosts: []ports.BrokerHostRecord{testHostRecord()}, InitialImportProvided: true}
42	-	}
43	-	func TestPriorFixtures(t *testing.T) {
44	-	for h := 1; h <= 3; h++ {
45	-	for c := 2; c <= 4; c++ {
46	-	t.Run(fmt.Sprintf("hosts%d-cache%d", h, c), func(t *testing.T) {
47	-	o := options(t)
48	-	o.LegacyHosts = fmt.Sprintf("testdata/hosts-v%d.json", h)
49	-	o.LegacyCache = fmt.Sprintf("testdata/cache-v%d.json", c)
50	-	beforeH, err := os.ReadFile(o.LegacyHosts)
51	-	require.NoError(t, err)
52	-	beforeC, err := os.ReadFile(o.LegacyCache)
53	-	require.NoError(t, err)
54	-	s, err := Open(o)
55	-	require.NoError(t, err)
56	-	hosts, err := s.LoadHosts()
57	-	require.NoError(t, err)
58	-	require.Len(t, hosts.Hosts, 1)
59	-	require.True(t, hosts.Hosts[0].Learned)
60	-	require.Equal(t, h != 1, hosts.Hosts[0].Pinned)
61	-	require.Equal(t, o.Policies["user@arch"], hosts.Hosts[0].Policy)
62	-	if h == 3 {
63	-	require.Equal(t, domain.RemoteGeneration(7), hosts.Hosts[0].Registration.Generation)
64	-	}
65	-	snap, err := s.Load()
66	-	require.NoError(t, err)
67	-	if h == 3 && c == 4 {
68	-	require.Len(t, snap.Daemons, 1)
69	-	require.Len(t, snap.Daemons[0].Sessions, 1)
70	-	require.Equal(t, "t_work", snap.Daemons[0].Sessions[0].Tabs[0].ID)
71	-	} else {
72	-	require.Empty(t, snap.Daemons)
73	-	}
74	-	require.NoError(t, s.Close())
75	-	o.LegacyHosts = "missing"
76	-	o.LegacyCache = "missing"
77	-	o.Policies = nil
78	-	s, err = Open(o)
79	-	require.NoError(t, err)
80	-	defer s.Close()
81	-	again, err := s.LoadHosts()
82	-	require.NoError(t, err)
83	-	require.Equal(t, hosts, again)
84	-	raw, err := readBounded(filepath.Join(o.Dir, "recovery.json"))
85	-	require.NoError(t, err)
86	-	var r recovery
87	-	require.NoError(t, strict(raw, &r))
88	-	require.Equal(t, beforeH, r.Hosts)
89	-	require.Equal(t, beforeC, r.Cache)
90	-	for _, name := range []string{"state.json", "recovery.json", lockFileName} {
91	-	st, err := os.Stat(filepath.Join(o.Dir, name))
92	-	require.NoError(t, err)
93	-	require.Equal(t, os.FileMode(0600), st.Mode().Perm())
94	-	}
95	-	})
96	-	}
97	-	}
98		}	47		}
99		func TestMigrationFaultRestart(t *testing.T) {	48		func TestMigrationFaultRestart(t *testing.T) {
100		for _, file := range []string{"recovery.json", "state.json"} {	49		for _, file := range []string{"recovery.json", "state.json"} {
101		for _, point := range []string{"write", "sync", "rename", "dirsync"} {	50		for _, point := range []string{"write", "sync", "rename", "dirsync"} {
102		t.Run(file+point, func(t *testing.T) {	51		t.Run(file+point, func(t *testing.T) {
103		o := options(t)	52		o := options(t)
104	-	o.LegacyHosts = "testdata/hosts-v2.json"
105		fault := errors.New("power loss")	53		fault := errors.New("power loss")
106		o.Fault = func(p string) error {	54		o.Fault = func(p string) error {
107		if p == file+":"+point {	55		if p == file+":"+point {

@@ -206,40 +154,17 @@ func TestLockExcludesLegacyHostWriter(t *testing.T) {
206		require.NoError(t, s.Close())	154		require.NoError(t, s.Close())
207		}	155		}
208		func TestStrictCorruption(t *testing.T) {	156		func TestStrictCorruption(t *testing.T) {
209	-	for _, raw := range []string{`{"version":1,"hosts":[]}`, `{"version":3,"hosts":[]}`} {	157	+	for _, raw := range []string{`{"version":1,"version":1}`, `{"unknown":1}`, `{} {`, string([]byte{255}), strings.Repeat(" ", MaxFileBytes+1)} {
210	-	var v any	158	+	var v state
211	-	require.NoError(t, strict([]byte(raw), &v))	159	+	require.Error(t, strict([]byte(raw), &v))
212		}	160		}
213	-	for name, raw := range map[string]string{"duplicate": `{"version":1,"version":1,"hosts":[]}`, "unknown": `{"version":1,"hosts":[],"extra":0}`, "trailing": `{"version":1,"hosts":[]} {}`, "missing": `{"version":2,"pinned":[]}`, "future": `{"version":99,"hosts":[]}`, "utf8": string([]byte{255}), "oversize": strings.Repeat(" ", MaxFileBytes+1), "empty": ""} {	161	+	o := options(t)
214	-	t.Run(name, func(t *testing.T) {	162	+	s, err := Open(o)
215	-	o := options(t)	163	+	require.NoError(t, err)
216	-	o.LegacyHosts = filepath.Join(t.TempDir(), "hosts")	164	+	require.NoError(t, s.Close())
217	-	require.NoError(t, os.WriteFile(o.LegacyHosts, []byte(raw), 0600))	165	+	require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
218	-	s, err := Open(o)	166	+	_, err = Open(o)
219	-	require.Error(t, err)	167	+	require.Error(t, err)
220	-	require.Nil(t, s)
221	-	_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
222	-	require.True(t, os.IsNotExist(err))
223	-	})
224	-	}
225	-	t.Run("no implicit policy", func(t *testing.T) { o := options(t); o.Policies = nil; _, err := Open(o); require.Error(t, err) })
226	-	t.Run("state never falls back", func(t *testing.T) {
227	-	o := options(t)
228	-	s, err := Open(o)
229	-	require.NoError(t, err)
230	-	require.NoError(t, s.Close())
231	-	require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
232	-	_, err = Open(o)
233	-	require.Error(t, err)
234	-	})
235	-	t.Run("source symlink", func(t *testing.T) {
236	-	o := options(t)
237	-	p := filepath.Join(t.TempDir(), "link")
238	-	require.NoError(t, os.Symlink(o.LegacyHosts, p))
239	-	o.LegacyHosts = p
240	-	_, err := Open(o)
241	-	require.Error(t, err)
242	-	})
243		}	168		}
244		func TestDurableStateSentinelMapping(t *testing.T) {	169		func TestDurableStateSentinelMapping(t *testing.T) {
245		o := options(t)	170		o := options(t)

@@ -264,28 +189,6 @@ func TestDurableStateSentinelMapping(t *testing.T) {
264		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)	189		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
265		}	190		}
266			191
267	-	// TestNonRegularInputsAreRejected covers inputs that are neither regular files
268	-	// nor absent: a directory and a FIFO must both fail closed before decoding.
269	-	func TestNonRegularInputsAreRejected(t *testing.T) {
270	-	for _, name := range []string{"directory", "fifo"} {
271	-	t.Run(name, func(t *testing.T) {
272	-	o := options(t)
273	-	path := filepath.Join(t.TempDir(), name)
274	-	if name == "directory" {
275	-	require.NoError(t, os.Mkdir(path, 0700))
276	-	} else {
277	-	require.NoError(t, syscall.Mkfifo(path, 0600))
278	-	}
279	-	o.LegacyHosts = path
280	-	s, err := Open(o)
281	-	require.Error(t, err)
282	-	require.Nil(t, s)
283	-	_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
284	-	require.True(t, os.IsNotExist(err))
285	-	})
286	-	}
287	-	}
288	-
289		// TestCommittedStateAndRecoveryAreNeverFollowed replaces the committed state and	192		// TestCommittedStateAndRecoveryAreNeverFollowed replaces the committed state and
290		// the recovery record with symlinks to valid bytes: the store reads both with	193		// the recovery record with symlinks to valid bytes: the store reads both with
291		// O_NOFOLLOW, so a symlink fails closed instead of redirecting a rename or	194		// O_NOFOLLOW, so a symlink fails closed instead of redirecting a rename or

@@ -335,81 +238,6 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
335		})	238		})
336		}	239		}
337			240
338	-	// TestManifestDistinguishesAbsentFromEmptySources pins the recovery manifest: an
339	-	// absent source and a supplied source that carries no records can produce the
340	-	// same membership, so presence, keyed by generic role label, is what makes the
341	-	// recovery record an exact description of the migration input.
342	-	func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
343	-	readManifest := func(t *testing.T, dir string) (manifest, string) {
344	-	t.Helper()
345	-	raw, err := readBounded(filepath.Join(dir, "recovery.json"))
346	-	require.NoError(t, err)
347	-	var r recovery
348	-	require.NoError(t, strict(raw, &r))
349	-	return r.State.Manifest, digest(nil)
350	-	}
351	-
352	-	absent := options(t)
353	-	absent.LegacyHosts = ""
354	-	absent.LegacyCache = ""
355	-	absent.Policies = nil
356	-	absentStore, err := Open(absent)
357	-	require.NoError(t, err)
358	-	defer absentStore.Close()
359	-	hosts, err := absentStore.LoadHosts()
360	-	require.NoError(t, err)
361	-	require.Empty(t, hosts.Hosts)
362	-	absentManifest, emptyDigest := readManifest(t, absent.Dir)
363	-	for _, label := range []manifestSourceLabel{manifestMembership, manifestObservations} {
364	-	source, ok := absentManifest.source(label)
365	-	require.True(t, ok, label)
366	-	require.False(t, source.Present, label)
367	-	require.Equal(t, emptyDigest, source.SHA256, label)
368	-	}
369	-
370	-	empty := options(t)
371	-	empty.LegacyCache = ""
372	-	empty.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
373	-	require.NoError(t, os.WriteFile(empty.LegacyHosts, []byte(`{"version":1,"hosts":[]}`), 0600))
374	-	emptyStore, err := Open(empty)
375	-	require.NoError(t, err)
376	-	defer emptyStore.Close()
377	-	stillEmpty, err := emptyStore.LoadHosts()
378	-	require.NoError(t, err)
379	-	// The same empty membership results, but the manifest records that the
380	-	// membership source was supplied and what it contained.
381	-	require.Equal(t, hosts, stillEmpty)
382	-	emptyManifest, _ := readManifest(t, empty.Dir)
383	-	membership, ok := emptyManifest.source(manifestMembership)
384	-	require.True(t, ok)
385	-	require.True(t, membership.Present)
386	-	require.NotEqual(t, emptyDigest, membership.SHA256)
387	-	observations, ok := emptyManifest.source(manifestObservations)
388	-	require.True(t, ok)
389	-	require.False(t, observations.Present)
390	-
391	-	// A record that claims the source was absent while retaining its bytes is
392	-	// rejected: presence is verified, not merely recorded.
393	-	raw, err := readBounded(filepath.Join(empty.Dir, "recovery.json"))
394	-	require.NoError(t, err)
395	-	var r recovery
396	-	require.NoError(t, strict(raw, &r))
397	-	for i := range r.State.Manifest.Sources {
398	-	if r.State.Manifest.Sources[i].Label == manifestMembership {
399	-	r.State.Manifest.Sources[i].Present = false
400	-	}
401	-	}
402	-	writer := Store{dir: empty.Dir}
403	-	require.NoError(t, writer.write("recovery.json", r))
404	-	require.NoError(t, os.Remove(filepath.Join(empty.Dir, "state.json")))
405	-	require.NoError(t, emptyStore.Close())
406	-	_, err = Open(Options{Dir: empty.Dir})
407	-	require.ErrorIs(t, err, ErrInvalidState)
408	-	require.ErrorContains(t, err, "absent membership source")
409	-	}
410	-
411	-	// TestRevisionOverflowIsRefused keeps the store from ever minting revision zero
412	-	// or accepting an exhausted authority token.
413		func TestRevisionOverflowIsRefused(t *testing.T) {	241		func TestRevisionOverflowIsRefused(t *testing.T) {
414		o := options(t)	242		o := options(t)
415		s, err := Open(o)	243		s, err := Open(o)

@@ -694,7 +522,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
694		if point == "rename" \|\| point == "dirsync" {	522		if point == "rename" \|\| point == "dirsync" {
695		require.Empty(t, h.Hosts)	523		require.Empty(t, h.Hosts)
696		snap.Revision++	524		snap.Revision++
697	-	require.ErrorIs(t, s.Store(snap), ErrStale)	525	+	require.NoError(t, s.Store(snap))
698		} else {	526		} else {
699		require.Len(t, h.Hosts, 1)	527		require.Len(t, h.Hosts, 1)
700		}	528		}

M internal/ports/broker_store.go +20 -0


@@ -252,3 +252,23 @@ func ValidateDurableHostProjection(obs BrokerDaemonObservation) error {
252	252		Sessions: obs.Sessions,
253	253		}})
254	254		}
	255	+
	256	+	// ValidateDurableObservation checks the shared remote-only durable observation shape.
	257	+	func 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	+	}


@@ -252,3 +252,23 @@ func ValidateDurableHostProjection(obs BrokerDaemonObservation) error {
252	Sessions: obs.Sessions,	252		Sessions: obs.Sessions,
253	}})	253		}})
254	}	254		}
		255	+
		256	+	// ValidateDurableObservation checks the shared remote-only durable observation shape.
		257	+	func 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	+	}