b56b682 refactor(broker): promote durable production storage · vev

M docs/architecture.md +1 -1


@@ -416,7 +416,7 @@ It is excluded from public help and changes no ordinary command, path, or
416	416		factory. The root must be absolute, cleaned, symlink-free, and outside the
417	417		production runtime and state directories; the runtime, state, and log paths are
418	418		derived only beneath it and secured with `pkg/safedir.EnsurePrivate`, and the
419		-	durable store is opened with `brokerstore.OpenOffline` over the sandbox state
	419	+	durable store is opened with `brokerstore.Open` over the sandbox state
420	420		directory with no legacy inputs.
421	421
422	422		`internal/adapters/brokerconfig` owns the strict sandbox configuration: a


@@ -416,7 +416,7 @@ It is excluded from public help and changes no ordinary command, path, or
416		factory. The root must be absolute, cleaned, symlink-free, and outside the	416		factory. The root must be absolute, cleaned, symlink-free, and outside the
417		production runtime and state directories; the runtime, state, and log paths are	417		production runtime and state directories; the runtime, state, and log paths are
418		derived only beneath it and secured with `pkg/safedir.EnsurePrivate`, and the	418		derived only beneath it and secured with `pkg/safedir.EnsurePrivate`, and the
419	-	durable store is opened with `brokerstore.OpenOffline` over the sandbox state	419	+	durable store is opened with `brokerstore.Open` over the sandbox state
420		directory with no legacy inputs.	420		directory with no legacy inputs.
421			421
422		`internal/adapters/brokerconfig` owns the strict sandbox configuration: a	422		`internal/adapters/brokerconfig` owns the strict sandbox configuration: a

M internal/adapters/brokerconfig/config.go +7 -1


@@ -288,7 +288,13 @@ type Config struct {
288	288		// route for the same pair is refused rather than silently dialed or ignored. A
289	289		// missing or malformed config is an error, never an empty default.
290	290		func Load(layout Layout) (*Config, error) {
291		-	raw, err := readConfig(filepath.Join(layout.Root, ConfigFileName))
	291	+	return LoadPath(layout, filepath.Join(layout.Root, ConfigFileName))
	292	+	}
	293	+
	294	+	// LoadPath parses the broker configuration at an explicit application-owned
	295	+	// path while retaining the route and reserved-path validation from layout.
	296	+	func LoadPath(layout Layout, path string) (*Config, error) {
	297	+	raw, err := readConfig(path)
292	298		if err != nil {
293	299		return nil, err
294	300		}

 // route for the same pair is refused rather than silently dialed or ignored. A
 // missing or malformed config is an error, never an empty default.
 func Load(layout Layout) (*Config, error) {
-	raw, err := readConfig(filepath.Join(layout.Root, ConfigFileName))
+	return LoadPath(layout, filepath.Join(layout.Root, ConfigFileName))
+}
+// LoadPath parses the broker configuration at an explicit application-owned
+// path while retaining the route and reserved-path validation from layout.
+func LoadPath(layout Layout, path string) (*Config, error) {
+	raw, err := readConfig(path)
 	if err != nil {
 		return nil, err
 	}

M internal/adapters/brokerstore/README.md +1 -1

@@ -1,6 +1,6 @@
1	1		# Offline broker store (Plan 001 P2.3)
2	2
3		-	`OpenOffline` is deliberately not wired into application composition. Supply a
	3	+	`Open` is deliberately not wired into application composition. Supply a
4	4		private destination directory, immutable copies of any legacy inputs, and an
5	5		explicit policy for each imported endpoint. An empty input path means no input;
6	6		a named missing, empty, corrupt, or unsupported file fails closed. The lifetime

@@ -1,6 +1,6 @@
1		# Offline broker store (Plan 001 P2.3)	1		# Offline broker store (Plan 001 P2.3)
2			2
3	-	`OpenOffline` is deliberately not wired into application composition. Supply a	3	+	`Open` is deliberately not wired into application composition. Supply a
4		private destination directory, immutable copies of any legacy inputs, and an	4		private destination directory, immutable copies of any legacy inputs, and an
5		explicit policy for each imported endpoint. An empty input path means no input;	5		explicit policy for each imported endpoint. An empty input path means no input;
6		a named missing, empty, corrupt, or unsupported file fails closed. The lifetime	6		a named missing, empty, corrupt, or unsupported file fails closed. The lifetime

M internal/adapters/brokerstore/legacy_test.go +4 -4


@@ -26,7 +26,7 @@ func TestCacheStrictFailures(t *testing.T) {
26	26		o := options(t)
27	27		o.LegacyCache = filepath.Join(t.TempDir(), "cache")
28	28		require.NoError(t, os.WriteFile(o.LegacyCache, []byte(data), 0600))
29		-	_, err := OpenOffline(o)
	29	+	_, err := Open(o)
30	30		require.Error(t, err)
31	31		})
32	32		}

@@ -34,7 +34,7 @@ func TestCacheStrictFailures(t *testing.T) {
34	34
35	35		func TestRecoveryCorruption(t *testing.T) {
36	36		o := options(t)
37		-	s, err := OpenOffline(o)
	37	+	s, err := Open(o)
38	38		require.NoError(t, err)
39	39		require.NoError(t, s.Close())
40	40		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))

@@ -46,13 +46,13 @@ func TestRecoveryCorruption(t *testing.T) {
46	46		// Use the production atomic writer without changing its digest metadata.
47	47		writer := Store{dir: o.Dir}
48	48		require.NoError(t, writer.write("recovery.json", r))
49		-	_, err = OpenOffline(o)
	49	+	_, err = Open(o)
50	50		require.ErrorContains(t, err, "digest mismatch")
51	51		}
52	52
53	53		func TestDefensiveCopiesAndConcurrentAccess(t *testing.T) {
54	54		o := options(t)
55		-	s, err := OpenOffline(o)
	55	+	s, err := Open(o)
56	56		require.NoError(t, err)
57	57		defer s.Close()
58	58		h, err := s.LoadHosts()


@@ -26,7 +26,7 @@ func TestCacheStrictFailures(t *testing.T) {
26		o := options(t)	26		o := options(t)
27		o.LegacyCache = filepath.Join(t.TempDir(), "cache")	27		o.LegacyCache = filepath.Join(t.TempDir(), "cache")
28		require.NoError(t, os.WriteFile(o.LegacyCache, []byte(data), 0600))	28		require.NoError(t, os.WriteFile(o.LegacyCache, []byte(data), 0600))
29	-	_, err := OpenOffline(o)	29	+	_, err := Open(o)
30		require.Error(t, err)	30		require.Error(t, err)
31		})	31		})
32		}	32		}

@@ -34,7 +34,7 @@ func TestCacheStrictFailures(t *testing.T) {
34			34
35		func TestRecoveryCorruption(t *testing.T) {	35		func TestRecoveryCorruption(t *testing.T) {
36		o := options(t)	36		o := options(t)
37	-	s, err := OpenOffline(o)	37	+	s, err := Open(o)
38		require.NoError(t, err)	38		require.NoError(t, err)
39		require.NoError(t, s.Close())	39		require.NoError(t, s.Close())
40		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))	40		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))

@@ -46,13 +46,13 @@ func TestRecoveryCorruption(t *testing.T) {
46		// Use the production atomic writer without changing its digest metadata.	46		// Use the production atomic writer without changing its digest metadata.
47		writer := Store{dir: o.Dir}	47		writer := Store{dir: o.Dir}
48		require.NoError(t, writer.write("recovery.json", r))	48		require.NoError(t, writer.write("recovery.json", r))
49	-	_, err = OpenOffline(o)	49	+	_, err = Open(o)
50		require.ErrorContains(t, err, "digest mismatch")	50		require.ErrorContains(t, err, "digest mismatch")
51		}	51		}
52			52
53		func TestDefensiveCopiesAndConcurrentAccess(t *testing.T) {	53		func TestDefensiveCopiesAndConcurrentAccess(t *testing.T) {
54		o := options(t)	54		o := options(t)
55	-	s, err := OpenOffline(o)	55	+	s, err := Open(o)
56		require.NoError(t, err)	56		require.NoError(t, err)
57		defer s.Close()	57		defer s.Close()
58		h, err := s.LoadHosts()	58		h, err := s.LoadHosts()

M internal/adapters/brokerstore/store.go +6 -5

@@ -1,5 +1,4 @@
1		-	// Package brokerstore provides offline-only broker persistence. OpenOffline is
2		-	// never composed into application wiring. Legacy sources are immutable inputs:
	1	+	// Package brokerstore provides broker persistence. Legacy sources are immutable inputs:
3	2		// the lifetime lock is the same hosts.json.lock the legacy host writer takes,
4	3		// so a live host writer is excluded, but the legacy catalog cache source has no
5	4		// lock at all and must be immutable or quiesced. No environment policy is

@@ -83,7 +82,7 @@ func (m manifest) source(label manifestSourceLabel) (manifestSource, bool) {
83	82		return manifestSource{}, false
84	83		}
85	84
86		-	// Options names a private offline destination and immutable legacy inputs.
	85	+	// Options names a private destination and immutable legacy inputs.
87	86		// Policies must explicitly cover every imported endpoint. Fault is a test seam
88	87		// called after each durable-write boundary; an error poisons this open handle.
89	88		type Options struct {

@@ -113,11 +112,13 @@ type Store struct {
113	112
114	113		var _ ports.BrokerHostStore = (*Store)(nil)
115	114
116		-	// OpenOffline returns only after validated unified state is durable. The
	115	+	// Open returns only after validated unified state is durable. The
117	116		// recovery record is written first and retained forever as the rollback copy;
118	117		// restart reuses its generated identities instead of migrating a second time.
119	118		// An existing corrupt state is an error, never silently replaced by a backup.
120		-	func OpenOffline(o Options) (*Store, error) {
	119	+	// Open acquires exclusive lifetime ownership, migrates immutable legacy input
	120	+	// when needed, and returns the durable broker store.
	121	+	func Open(o Options) (*Store, error) {
121	122		if err := safedir.EnsurePrivate(o.Dir); err != nil {
122	123		return nil, err
123	124		}

-// Package brokerstore provides offline-only broker persistence. OpenOffline is
+// Package brokerstore provides broker persistence. Legacy sources are immutable inputs:
-// never composed into application wiring. Legacy sources are immutable inputs:
 // the lifetime lock is the same hosts.json.lock the legacy host writer takes,
 // so a live host writer is excluded, but the legacy catalog cache source has no
 // lock at all and must be immutable or quiesced. No environment policy is
 	return manifestSource{}, false
 }
-// Options names a private offline destination and immutable legacy inputs.
+// Options names a private destination and immutable legacy inputs.
 // 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 {
 var _ ports.BrokerHostStore = (*Store)(nil)
-// OpenOffline returns only after validated unified state is durable. The
+// Open returns only after validated unified state is durable. The
 // recovery record is written first and retained forever as the rollback copy;
 // restart reuses its generated identities instead of migrating a second time.
 // An existing corrupt state is an error, never silently replaced by a backup.
-func OpenOffline(o Options) (*Store, error) {
+// Open acquires exclusive lifetime ownership, migrates immutable legacy input
+// when needed, and returns the durable broker store.
+func Open(o Options) (*Store, error) {
 	if err := safedir.EnsurePrivate(o.Dir); err != nil {
 		return nil, err
 	}

M internal/adapters/brokerstore/store_test.go +38 -38


@@ -49,7 +49,7 @@ func TestPriorFixtures(t *testing.T) {
49	49		require.NoError(t, err)
50	50		beforeC, err := os.ReadFile(o.LegacyCache)
51	51		require.NoError(t, err)
52		-	s, err := OpenOffline(o)
	52	+	s, err := Open(o)
53	53		require.NoError(t, err)
54	54		hosts, err := s.LoadHosts()
55	55		require.NoError(t, err)

@@ -73,7 +73,7 @@ func TestPriorFixtures(t *testing.T) {
73	73		o.LegacyHosts = "missing"
74	74		o.LegacyCache = "missing"
75	75		o.Policies = nil
76		-	s, err = OpenOffline(o)
	76	+	s, err = Open(o)
77	77		require.NoError(t, err)
78	78		defer s.Close()
79	79		again, err := s.LoadHosts()

@@ -107,7 +107,7 @@ func TestMigrationFaultRestart(t *testing.T) {
107	107		}
108	108		return nil
109	109		}
110		-	s, err := OpenOffline(o)
	110	+	s, err := Open(o)
111	111		require.ErrorIs(t, err, fault)
112	112		require.Nil(t, s)
113	113		var saved recovery

@@ -116,7 +116,7 @@ func TestMigrationFaultRestart(t *testing.T) {
116	116		require.NoError(t, strict(raw, &saved))
117	117		}
118	118		o.Fault = nil
119		-	s, err = OpenOffline(o)
	119	+	s, err = Open(o)
120	120		require.NoError(t, err)
121	121		defer s.Close()
122	122		hosts, err := s.LoadHosts()

@@ -130,7 +130,7 @@ func TestMigrationFaultRestart(t *testing.T) {
130	130		}
131	131		func TestLockLifetime(t *testing.T) {
132	132		if dir := os.Getenv("VEV_BROKERSTORE_LOCK_TEST"); dir != "" {
133		-	s, err := OpenOffline(Options{Dir: dir})
	133	+	s, err := Open(Options{Dir: dir})
134	134		if s != nil {
135	135		s.Close()
136	136		}

@@ -140,7 +140,7 @@ func TestLockLifetime(t *testing.T) {
140	140		return
141	141		}
142	142		o := options(t)
143		-	s, err := OpenOffline(o)
	143	+	s, err := Open(o)
144	144		require.NoError(t, err)
145	145		// The lifetime lock carries the legacy host writer's exact file name, so a
146	146		// store opened beside a live hosts.json excludes that writer too.

@@ -158,14 +158,14 @@ func TestLockLifetime(t *testing.T) {
158	158		wg.Add(1)
159	159		go func(i int) {
160	160		defer wg.Done()
161		-	other, e := OpenOffline(o)
	161	+	other, e := Open(o)
162	162		if other != nil {
163	163		_ = other.Close()
164		-	outcomes[i] = fmt.Errorf("concurrent OpenOffline returned a store: %w", ports.ErrBrokerStoreLocked)
	164	+	outcomes[i] = fmt.Errorf("concurrent Open returned a store: %w", ports.ErrBrokerStoreLocked)
165	165		return
166	166		}
167	167		if !errors.Is(e, ErrLocked) \|\| !errors.Is(e, ports.ErrBrokerStoreLocked) {
168		-	outcomes[i] = fmt.Errorf("concurrent OpenOffline error = %v, want ErrLocked", e)
	168	+	outcomes[i] = fmt.Errorf("concurrent Open error = %v, want ErrLocked", e)
169	169		}
170	170		}(i)
171	171		}

@@ -176,7 +176,7 @@ func TestLockLifetime(t *testing.T) {
176	176		require.NoError(t, s.Close())
177	177		_, err = s.Load()
178	178		require.Error(t, err)
179		-	s, err = OpenOffline(o)
	179	+	s, err = Open(o)
180	180		require.NoError(t, err)
181	181		require.NoError(t, s.Close())
182	182		}

@@ -185,7 +185,7 @@ func TestLockLifetime(t *testing.T) {
185	185		// writer does and proves the store refuses to own the same directory.
186	186		func TestLockExcludesLegacyHostWriter(t *testing.T) {
187	187		o := options(t)
188		-	s, err := OpenOffline(o)
	188	+	s, err := Open(o)
189	189		require.NoError(t, err)
190	190		require.NoError(t, s.Close())
191	191

@@ -194,12 +194,12 @@ func TestLockExcludesLegacyHostWriter(t *testing.T) {
194	194		defer legacyLock.Close()
195	195		require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_EX\|syscall.LOCK_NB))
196	196
197		-	_, err = OpenOffline(o)
	197	+	_, err = Open(o)
198	198		require.ErrorIs(t, err, ErrLocked)
199	199		require.ErrorIs(t, err, ports.ErrBrokerStoreLocked)
200	200
201	201		require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_UN))
202		-	s, err = OpenOffline(o)
	202	+	s, err = Open(o)
203	203		require.NoError(t, err)
204	204		require.NoError(t, s.Close())
205	205		}

@@ -213,21 +213,21 @@ func TestStrictCorruption(t *testing.T) {
213	213		o := options(t)
214	214		o.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
215	215		require.NoError(t, os.WriteFile(o.LegacyHosts, []byte(raw), 0600))
216		-	s, err := OpenOffline(o)
	216	+	s, err := Open(o)
217	217		require.Error(t, err)
218	218		require.Nil(t, s)
219	219		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
220	220		require.True(t, os.IsNotExist(err))
221	221		})
222	222		}
223		-	t.Run("no implicit policy", func(t *testing.T) { o := options(t); o.Policies = nil; _, err := OpenOffline(o); require.Error(t, err) })
	223	+	t.Run("no implicit policy", func(t *testing.T) { o := options(t); o.Policies = nil; _, err := Open(o); require.Error(t, err) })
224	224		t.Run("state never falls back", func(t *testing.T) {
225	225		o := options(t)
226		-	s, err := OpenOffline(o)
	226	+	s, err := Open(o)
227	227		require.NoError(t, err)
228	228		require.NoError(t, s.Close())
229	229		require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
230		-	_, err = OpenOffline(o)
	230	+	_, err = Open(o)
231	231		require.Error(t, err)
232	232		})
233	233		t.Run("source symlink", func(t *testing.T) {

@@ -235,13 +235,13 @@ func TestStrictCorruption(t *testing.T) {
235	235		p := filepath.Join(t.TempDir(), "link")
236	236		require.NoError(t, os.Symlink(o.LegacyHosts, p))
237	237		o.LegacyHosts = p
238		-	_, err := OpenOffline(o)
	238	+	_, err := Open(o)
239	239		require.Error(t, err)
240	240		})
241	241		}
242	242		func TestDurableStateSentinelMapping(t *testing.T) {
243	243		o := options(t)
244		-	s, err := OpenOffline(o)
	244	+	s, err := Open(o)
245	245		require.NoError(t, err)
246	246
247	247		// A conflicting authority token is a host conflict, not a generic failure.

@@ -257,7 +257,7 @@ func TestDurableStateSentinelMapping(t *testing.T) {
257	257		// Closing releases ownership without discarding the durable files.
258	258		require.Error(t, s.ReplaceHosts(1, nil))
259	259		require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
260		-	_, err = OpenOffline(o)
	260	+	_, err = Open(o)
261	261		require.ErrorIs(t, err, ErrInvalidState)
262	262		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
263	263		}

@@ -275,7 +275,7 @@ func TestNonRegularInputsAreRejected(t *testing.T) {
275	275		require.NoError(t, syscall.Mkfifo(path, 0600))
276	276		}
277	277		o.LegacyHosts = path
278		-	s, err := OpenOffline(o)
	278	+	s, err := Open(o)
279	279		require.Error(t, err)
280	280		require.Nil(t, s)
281	281		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))

@@ -291,7 +291,7 @@ func TestNonRegularInputsAreRejected(t *testing.T) {
291	291		func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
292	292		t.Run("state", func(t *testing.T) {
293	293		o := options(t)
294		-	s, err := OpenOffline(o)
	294	+	s, err := Open(o)
295	295		require.NoError(t, err)
296	296		require.NoError(t, s.Close())
297	297

@@ -303,7 +303,7 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
303	303		require.NoError(t, os.Remove(state))
304	304		require.NoError(t, os.Symlink(target, state))
305	305
306		-	_, err = OpenOffline(o)
	306	+	_, err = Open(o)
307	307		require.ErrorIs(t, err, ErrInvalidState)
308	308		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
309	309		after, err := os.ReadFile(target)

@@ -312,7 +312,7 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
312	312		})
313	313		t.Run("recovery", func(t *testing.T) {
314	314		o := options(t)
315		-	s, err := OpenOffline(o)
	315	+	s, err := Open(o)
316	316		require.NoError(t, err)
317	317		require.NoError(t, s.Close())
318	318

@@ -325,7 +325,7 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
325	325		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))
326	326		require.NoError(t, os.Symlink(target, recoveryPath))
327	327
328		-	_, err = OpenOffline(o)
	328	+	_, err = Open(o)
329	329		require.ErrorIs(t, err, ErrInvalidState)
330	330		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
331	331		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))

@@ -351,7 +351,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
351	351		absent.LegacyHosts = ""
352	352		absent.LegacyCache = ""
353	353		absent.Policies = nil
354		-	absentStore, err := OpenOffline(absent)
	354	+	absentStore, err := Open(absent)
355	355		require.NoError(t, err)
356	356		defer absentStore.Close()
357	357		hosts, err := absentStore.LoadHosts()

@@ -369,7 +369,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
369	369		empty.LegacyCache = ""
370	370		empty.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
371	371		require.NoError(t, os.WriteFile(empty.LegacyHosts, []byte(`{"version":1,"hosts":[]}`), 0600))
372		-	emptyStore, err := OpenOffline(empty)
	372	+	emptyStore, err := Open(empty)
373	373		require.NoError(t, err)
374	374		defer emptyStore.Close()
375	375		stillEmpty, err := emptyStore.LoadHosts()

@@ -401,7 +401,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
401	401		require.NoError(t, writer.write("recovery.json", r))
402	402		require.NoError(t, os.Remove(filepath.Join(empty.Dir, "state.json")))
403	403		require.NoError(t, emptyStore.Close())
404		-	_, err = OpenOffline(Options{Dir: empty.Dir})
	404	+	_, err = Open(Options{Dir: empty.Dir})
405	405		require.ErrorIs(t, err, ErrInvalidState)
406	406		require.ErrorContains(t, err, "absent membership source")
407	407		}

@@ -410,7 +410,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
410	410		// or accepting an exhausted authority token.
411	411		func TestRevisionOverflowIsRefused(t *testing.T) {
412	412		o := options(t)
413		-	s, err := OpenOffline(o)
	413	+	s, err := Open(o)
414	414		require.NoError(t, err)
415	415		defer s.Close()
416	416		h, err := s.LoadHosts()

@@ -429,7 +429,7 @@ func TestRevisionOverflowIsRefused(t *testing.T) {
429	429		// incarnation.
430	430		func TestAuthorityGenerationPolicyAndRemoveReAdd(t *testing.T) {
431	431		o := options(t)
432		-	s, err := OpenOffline(o)
	432	+	s, err := Open(o)
433	433		require.NoError(t, err)
434	434		defer s.Close()
435	435		h, err := s.LoadHosts()

@@ -495,7 +495,7 @@ func TestAuthorityGenerationPolicyAndRemoveReAdd(t *testing.T) {
495	495		// as ErrBrokerStoreInvalidState: the durable state is a victim, not the cause.
496	496		func TestReplaceHostsRejectsCallerInputBeforeCAS(t *testing.T) {
497	497		o := options(t)
498		-	s, err := OpenOffline(o)
	498	+	s, err := Open(o)
499	499		require.NoError(t, err)
500	500		defer s.Close()
501	501		h, err := s.LoadHosts()

@@ -548,7 +548,7 @@ func TestReplaceHostsRejectsCallerInputBeforeCAS(t *testing.T) {
548	548		// Removed, instead of being rejected as durable-state corruption.
549	549		func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) {
550	550		o := options(t)
551		-	s, err := OpenOffline(o)
	551	+	s, err := Open(o)
552	552		require.NoError(t, err)
553	553		h, err := s.LoadHosts()
554	554		require.NoError(t, err)

@@ -567,7 +567,7 @@ func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) {
567	567		require.Empty(t, stored.Removed)
568	568		require.NoError(t, s.Close())
569	569
570		-	s, err = OpenOffline(o)
	570	+	s, err = Open(o)
571	571		require.NoError(t, err)
572	572		defer s.Close()
573	573		reopened, err := s.Load()

@@ -579,7 +579,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
579	579		for _, point := range []string{"write", "sync", "rename", "dirsync"} {
580	580		t.Run(point, func(t *testing.T) {
581	581		o := options(t)
582		-	s, err := OpenOffline(o)
	582	+	s, err := Open(o)
583	583		require.NoError(t, err)
584	584		h, err := s.LoadHosts()
585	585		require.NoError(t, err)

@@ -606,7 +606,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
606	606		_, err = s.LoadHosts()
607	607		require.Error(t, err)
608	608		require.NoError(t, s.Close())
609		-	s, err = OpenOffline(o)
	609	+	s, err = Open(o)
610	610		require.NoError(t, err)
611	611		defer s.Close()
612	612		h, err = s.LoadHosts()

@@ -622,7 +622,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
622	622		}
623	623		t.Run("generation and restart", func(t *testing.T) {
624	624		o := options(t)
625		-	s, err := OpenOffline(o)
	625	+	s, err := Open(o)
626	626		require.NoError(t, err)
627	627		h, err := s.LoadHosts()
628	628		require.NoError(t, err)

@@ -632,7 +632,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
632	632		require.NoError(t, s.ReplaceHosts(h.Revision, h.Hosts))
633	633		require.ErrorIs(t, s.Store(snap), ErrStale)
634	634		require.NoError(t, s.Close())
635		-	s, err = OpenOffline(o)
	635	+	s, err = Open(o)
636	636		require.NoError(t, err)
637	637		defer s.Close()
638	638		got, err := s.Load()

@@ -648,7 +648,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
648	648		// daemon observation is process-local state and the store refuses it outright,
649	649		// so it can never become durable authority.
650	650		func TestLocalDaemonObservationIsNeverDurable(t *testing.T) {
651		-	s, err := OpenOffline(options(t))
	651	+	s, err := Open(options(t))
652	652		require.NoError(t, err)
653	653		defer s.Close()
654	654		err = s.Store(ports.BrokerSnapshot{Epoch: 9, Revision: 1, Daemons: []ports.BrokerDaemonObservation{{


@@ -49,7 +49,7 @@ func TestPriorFixtures(t *testing.T) {
49		require.NoError(t, err)	49		require.NoError(t, err)
50		beforeC, err := os.ReadFile(o.LegacyCache)	50		beforeC, err := os.ReadFile(o.LegacyCache)
51		require.NoError(t, err)	51		require.NoError(t, err)
52	-	s, err := OpenOffline(o)	52	+	s, err := Open(o)
53		require.NoError(t, err)	53		require.NoError(t, err)
54		hosts, err := s.LoadHosts()	54		hosts, err := s.LoadHosts()
55		require.NoError(t, err)	55		require.NoError(t, err)

@@ -73,7 +73,7 @@ func TestPriorFixtures(t *testing.T) {
73		o.LegacyHosts = "missing"	73		o.LegacyHosts = "missing"
74		o.LegacyCache = "missing"	74		o.LegacyCache = "missing"
75		o.Policies = nil	75		o.Policies = nil
76	-	s, err = OpenOffline(o)	76	+	s, err = Open(o)
77		require.NoError(t, err)	77		require.NoError(t, err)
78		defer s.Close()	78		defer s.Close()
79		again, err := s.LoadHosts()	79		again, err := s.LoadHosts()

@@ -107,7 +107,7 @@ func TestMigrationFaultRestart(t *testing.T) {
107		}	107		}
108		return nil	108		return nil
109		}	109		}
110	-	s, err := OpenOffline(o)	110	+	s, err := Open(o)
111		require.ErrorIs(t, err, fault)	111		require.ErrorIs(t, err, fault)
112		require.Nil(t, s)	112		require.Nil(t, s)
113		var saved recovery	113		var saved recovery

@@ -116,7 +116,7 @@ func TestMigrationFaultRestart(t *testing.T) {
116		require.NoError(t, strict(raw, &saved))	116		require.NoError(t, strict(raw, &saved))
117		}	117		}
118		o.Fault = nil	118		o.Fault = nil
119	-	s, err = OpenOffline(o)	119	+	s, err = Open(o)
120		require.NoError(t, err)	120		require.NoError(t, err)
121		defer s.Close()	121		defer s.Close()
122		hosts, err := s.LoadHosts()	122		hosts, err := s.LoadHosts()

@@ -130,7 +130,7 @@ func TestMigrationFaultRestart(t *testing.T) {
130		}	130		}
131		func TestLockLifetime(t *testing.T) {	131		func TestLockLifetime(t *testing.T) {
132		if dir := os.Getenv("VEV_BROKERSTORE_LOCK_TEST"); dir != "" {	132		if dir := os.Getenv("VEV_BROKERSTORE_LOCK_TEST"); dir != "" {
133	-	s, err := OpenOffline(Options{Dir: dir})	133	+	s, err := Open(Options{Dir: dir})
134		if s != nil {	134		if s != nil {
135		s.Close()	135		s.Close()
136		}	136		}

@@ -140,7 +140,7 @@ func TestLockLifetime(t *testing.T) {
140		return	140		return
141		}	141		}
142		o := options(t)	142		o := options(t)
143	-	s, err := OpenOffline(o)	143	+	s, err := Open(o)
144		require.NoError(t, err)	144		require.NoError(t, err)
145		// The lifetime lock carries the legacy host writer's exact file name, so a	145		// The lifetime lock carries the legacy host writer's exact file name, so a
146		// store opened beside a live hosts.json excludes that writer too.	146		// store opened beside a live hosts.json excludes that writer too.

@@ -158,14 +158,14 @@ func TestLockLifetime(t *testing.T) {
158		wg.Add(1)	158		wg.Add(1)
159		go func(i int) {	159		go func(i int) {
160		defer wg.Done()	160		defer wg.Done()
161	-	other, e := OpenOffline(o)	161	+	other, e := Open(o)
162		if other != nil {	162		if other != nil {
163		_ = other.Close()	163		_ = other.Close()
164	-	outcomes[i] = fmt.Errorf("concurrent OpenOffline returned a store: %w", ports.ErrBrokerStoreLocked)	164	+	outcomes[i] = fmt.Errorf("concurrent Open returned a store: %w", ports.ErrBrokerStoreLocked)
165		return	165		return
166		}	166		}
167		if !errors.Is(e, ErrLocked) \|\| !errors.Is(e, ports.ErrBrokerStoreLocked) {	167		if !errors.Is(e, ErrLocked) \|\| !errors.Is(e, ports.ErrBrokerStoreLocked) {
168	-	outcomes[i] = fmt.Errorf("concurrent OpenOffline error = %v, want ErrLocked", e)	168	+	outcomes[i] = fmt.Errorf("concurrent Open error = %v, want ErrLocked", e)
169		}	169		}
170		}(i)	170		}(i)
171		}	171		}

@@ -176,7 +176,7 @@ func TestLockLifetime(t *testing.T) {
176		require.NoError(t, s.Close())	176		require.NoError(t, s.Close())
177		_, err = s.Load()	177		_, err = s.Load()
178		require.Error(t, err)	178		require.Error(t, err)
179	-	s, err = OpenOffline(o)	179	+	s, err = Open(o)
180		require.NoError(t, err)	180		require.NoError(t, err)
181		require.NoError(t, s.Close())	181		require.NoError(t, s.Close())
182		}	182		}

@@ -185,7 +185,7 @@ func TestLockLifetime(t *testing.T) {
185		// writer does and proves the store refuses to own the same directory.	185		// writer does and proves the store refuses to own the same directory.
186		func TestLockExcludesLegacyHostWriter(t *testing.T) {	186		func TestLockExcludesLegacyHostWriter(t *testing.T) {
187		o := options(t)	187		o := options(t)
188	-	s, err := OpenOffline(o)	188	+	s, err := Open(o)
189		require.NoError(t, err)	189		require.NoError(t, err)
190		require.NoError(t, s.Close())	190		require.NoError(t, s.Close())
191			191

@@ -194,12 +194,12 @@ func TestLockExcludesLegacyHostWriter(t *testing.T) {
194		defer legacyLock.Close()	194		defer legacyLock.Close()
195		require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_EX\|syscall.LOCK_NB))	195		require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_EX\|syscall.LOCK_NB))
196			196
197	-	_, err = OpenOffline(o)	197	+	_, err = Open(o)
198		require.ErrorIs(t, err, ErrLocked)	198		require.ErrorIs(t, err, ErrLocked)
199		require.ErrorIs(t, err, ports.ErrBrokerStoreLocked)	199		require.ErrorIs(t, err, ports.ErrBrokerStoreLocked)
200			200
201		require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_UN))	201		require.NoError(t, syscall.Flock(int(legacyLock.Fd()), syscall.LOCK_UN))
202	-	s, err = OpenOffline(o)	202	+	s, err = Open(o)
203		require.NoError(t, err)	203		require.NoError(t, err)
204		require.NoError(t, s.Close())	204		require.NoError(t, s.Close())
205		}	205		}

@@ -213,21 +213,21 @@ func TestStrictCorruption(t *testing.T) {
213		o := options(t)	213		o := options(t)
214		o.LegacyHosts = filepath.Join(t.TempDir(), "hosts")	214		o.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
215		require.NoError(t, os.WriteFile(o.LegacyHosts, []byte(raw), 0600))	215		require.NoError(t, os.WriteFile(o.LegacyHosts, []byte(raw), 0600))
216	-	s, err := OpenOffline(o)	216	+	s, err := Open(o)
217		require.Error(t, err)	217		require.Error(t, err)
218		require.Nil(t, s)	218		require.Nil(t, s)
219		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))	219		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))
220		require.True(t, os.IsNotExist(err))	220		require.True(t, os.IsNotExist(err))
221		})	221		})
222		}	222		}
223	-	t.Run("no implicit policy", func(t *testing.T) { o := options(t); o.Policies = nil; _, err := OpenOffline(o); require.Error(t, err) })	223	+	t.Run("no implicit policy", func(t *testing.T) { o := options(t); o.Policies = nil; _, err := Open(o); require.Error(t, err) })
224		t.Run("state never falls back", func(t *testing.T) {	224		t.Run("state never falls back", func(t *testing.T) {
225		o := options(t)	225		o := options(t)
226	-	s, err := OpenOffline(o)	226	+	s, err := Open(o)
227		require.NoError(t, err)	227		require.NoError(t, err)
228		require.NoError(t, s.Close())	228		require.NoError(t, s.Close())
229		require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))	229		require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
230	-	_, err = OpenOffline(o)	230	+	_, err = Open(o)
231		require.Error(t, err)	231		require.Error(t, err)
232		})	232		})
233		t.Run("source symlink", func(t *testing.T) {	233		t.Run("source symlink", func(t *testing.T) {

@@ -235,13 +235,13 @@ func TestStrictCorruption(t *testing.T) {
235		p := filepath.Join(t.TempDir(), "link")	235		p := filepath.Join(t.TempDir(), "link")
236		require.NoError(t, os.Symlink(o.LegacyHosts, p))	236		require.NoError(t, os.Symlink(o.LegacyHosts, p))
237		o.LegacyHosts = p	237		o.LegacyHosts = p
238	-	_, err := OpenOffline(o)	238	+	_, err := Open(o)
239		require.Error(t, err)	239		require.Error(t, err)
240		})	240		})
241		}	241		}
242		func TestDurableStateSentinelMapping(t *testing.T) {	242		func TestDurableStateSentinelMapping(t *testing.T) {
243		o := options(t)	243		o := options(t)
244	-	s, err := OpenOffline(o)	244	+	s, err := Open(o)
245		require.NoError(t, err)	245		require.NoError(t, err)
246			246
247		// A conflicting authority token is a host conflict, not a generic failure.	247		// A conflicting authority token is a host conflict, not a generic failure.

@@ -257,7 +257,7 @@ func TestDurableStateSentinelMapping(t *testing.T) {
257		// Closing releases ownership without discarding the durable files.	257		// Closing releases ownership without discarding the durable files.
258		require.Error(t, s.ReplaceHosts(1, nil))	258		require.Error(t, s.ReplaceHosts(1, nil))
259		require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))	259		require.NoError(t, os.WriteFile(filepath.Join(o.Dir, "state.json"), []byte(`{`), 0600))
260	-	_, err = OpenOffline(o)	260	+	_, err = Open(o)
261		require.ErrorIs(t, err, ErrInvalidState)	261		require.ErrorIs(t, err, ErrInvalidState)
262		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)	262		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
263		}	263		}

@@ -275,7 +275,7 @@ func TestNonRegularInputsAreRejected(t *testing.T) {
275		require.NoError(t, syscall.Mkfifo(path, 0600))	275		require.NoError(t, syscall.Mkfifo(path, 0600))
276		}	276		}
277		o.LegacyHosts = path	277		o.LegacyHosts = path
278	-	s, err := OpenOffline(o)	278	+	s, err := Open(o)
279		require.Error(t, err)	279		require.Error(t, err)
280		require.Nil(t, s)	280		require.Nil(t, s)
281		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))	281		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))

@@ -291,7 +291,7 @@ func TestNonRegularInputsAreRejected(t *testing.T) {
291		func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {	291		func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
292		t.Run("state", func(t *testing.T) {	292		t.Run("state", func(t *testing.T) {
293		o := options(t)	293		o := options(t)
294	-	s, err := OpenOffline(o)	294	+	s, err := Open(o)
295		require.NoError(t, err)	295		require.NoError(t, err)
296		require.NoError(t, s.Close())	296		require.NoError(t, s.Close())
297			297

@@ -303,7 +303,7 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
303		require.NoError(t, os.Remove(state))	303		require.NoError(t, os.Remove(state))
304		require.NoError(t, os.Symlink(target, state))	304		require.NoError(t, os.Symlink(target, state))
305			305
306	-	_, err = OpenOffline(o)	306	+	_, err = Open(o)
307		require.ErrorIs(t, err, ErrInvalidState)	307		require.ErrorIs(t, err, ErrInvalidState)
308		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)	308		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
309		after, err := os.ReadFile(target)	309		after, err := os.ReadFile(target)

@@ -312,7 +312,7 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
312		})	312		})
313		t.Run("recovery", func(t *testing.T) {	313		t.Run("recovery", func(t *testing.T) {
314		o := options(t)	314		o := options(t)
315	-	s, err := OpenOffline(o)	315	+	s, err := Open(o)
316		require.NoError(t, err)	316		require.NoError(t, err)
317		require.NoError(t, s.Close())	317		require.NoError(t, s.Close())
318			318

@@ -325,7 +325,7 @@ func TestCommittedStateAndRecoveryAreNeverFollowed(t *testing.T) {
325		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))	325		require.NoError(t, os.Remove(filepath.Join(o.Dir, "state.json")))
326		require.NoError(t, os.Symlink(target, recoveryPath))	326		require.NoError(t, os.Symlink(target, recoveryPath))
327			327
328	-	_, err = OpenOffline(o)	328	+	_, err = Open(o)
329		require.ErrorIs(t, err, ErrInvalidState)	329		require.ErrorIs(t, err, ErrInvalidState)
330		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)	330		require.ErrorIs(t, err, ports.ErrBrokerStoreInvalidState)
331		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))	331		_, err = os.Stat(filepath.Join(o.Dir, "state.json"))

@@ -351,7 +351,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
351		absent.LegacyHosts = ""	351		absent.LegacyHosts = ""
352		absent.LegacyCache = ""	352		absent.LegacyCache = ""
353		absent.Policies = nil	353		absent.Policies = nil
354	-	absentStore, err := OpenOffline(absent)	354	+	absentStore, err := Open(absent)
355		require.NoError(t, err)	355		require.NoError(t, err)
356		defer absentStore.Close()	356		defer absentStore.Close()
357		hosts, err := absentStore.LoadHosts()	357		hosts, err := absentStore.LoadHosts()

@@ -369,7 +369,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
369		empty.LegacyCache = ""	369		empty.LegacyCache = ""
370		empty.LegacyHosts = filepath.Join(t.TempDir(), "hosts")	370		empty.LegacyHosts = filepath.Join(t.TempDir(), "hosts")
371		require.NoError(t, os.WriteFile(empty.LegacyHosts, []byte(`{"version":1,"hosts":[]}`), 0600))	371		require.NoError(t, os.WriteFile(empty.LegacyHosts, []byte(`{"version":1,"hosts":[]}`), 0600))
372	-	emptyStore, err := OpenOffline(empty)	372	+	emptyStore, err := Open(empty)
373		require.NoError(t, err)	373		require.NoError(t, err)
374		defer emptyStore.Close()	374		defer emptyStore.Close()
375		stillEmpty, err := emptyStore.LoadHosts()	375		stillEmpty, err := emptyStore.LoadHosts()

@@ -401,7 +401,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
401		require.NoError(t, writer.write("recovery.json", r))	401		require.NoError(t, writer.write("recovery.json", r))
402		require.NoError(t, os.Remove(filepath.Join(empty.Dir, "state.json")))	402		require.NoError(t, os.Remove(filepath.Join(empty.Dir, "state.json")))
403		require.NoError(t, emptyStore.Close())	403		require.NoError(t, emptyStore.Close())
404	-	_, err = OpenOffline(Options{Dir: empty.Dir})	404	+	_, err = Open(Options{Dir: empty.Dir})
405		require.ErrorIs(t, err, ErrInvalidState)	405		require.ErrorIs(t, err, ErrInvalidState)
406		require.ErrorContains(t, err, "absent membership source")	406		require.ErrorContains(t, err, "absent membership source")
407		}	407		}

@@ -410,7 +410,7 @@ func TestManifestDistinguishesAbsentFromEmptySources(t *testing.T) {
410		// or accepting an exhausted authority token.	410		// or accepting an exhausted authority token.
411		func TestRevisionOverflowIsRefused(t *testing.T) {	411		func TestRevisionOverflowIsRefused(t *testing.T) {
412		o := options(t)	412		o := options(t)
413	-	s, err := OpenOffline(o)	413	+	s, err := Open(o)
414		require.NoError(t, err)	414		require.NoError(t, err)
415		defer s.Close()	415		defer s.Close()
416		h, err := s.LoadHosts()	416		h, err := s.LoadHosts()

@@ -429,7 +429,7 @@ func TestRevisionOverflowIsRefused(t *testing.T) {
429		// incarnation.	429		// incarnation.
430		func TestAuthorityGenerationPolicyAndRemoveReAdd(t *testing.T) {	430		func TestAuthorityGenerationPolicyAndRemoveReAdd(t *testing.T) {
431		o := options(t)	431		o := options(t)
432	-	s, err := OpenOffline(o)	432	+	s, err := Open(o)
433		require.NoError(t, err)	433		require.NoError(t, err)
434		defer s.Close()	434		defer s.Close()
435		h, err := s.LoadHosts()	435		h, err := s.LoadHosts()

@@ -495,7 +495,7 @@ func TestAuthorityGenerationPolicyAndRemoveReAdd(t *testing.T) {
495		// as ErrBrokerStoreInvalidState: the durable state is a victim, not the cause.	495		// as ErrBrokerStoreInvalidState: the durable state is a victim, not the cause.
496		func TestReplaceHostsRejectsCallerInputBeforeCAS(t *testing.T) {	496		func TestReplaceHostsRejectsCallerInputBeforeCAS(t *testing.T) {
497		o := options(t)	497		o := options(t)
498	-	s, err := OpenOffline(o)	498	+	s, err := Open(o)
499		require.NoError(t, err)	499		require.NoError(t, err)
500		defer s.Close()	500		defer s.Close()
501		h, err := s.LoadHosts()	501		h, err := s.LoadHosts()

@@ -548,7 +548,7 @@ func TestReplaceHostsRejectsCallerInputBeforeCAS(t *testing.T) {
548		// Removed, instead of being rejected as durable-state corruption.	548		// Removed, instead of being rejected as durable-state corruption.
549		func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) {	549		func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) {
550		o := options(t)	550		o := options(t)
551	-	s, err := OpenOffline(o)	551	+	s, err := Open(o)
552		require.NoError(t, err)	552		require.NoError(t, err)
553		h, err := s.LoadHosts()	553		h, err := s.LoadHosts()
554		require.NoError(t, err)	554		require.NoError(t, err)

@@ -567,7 +567,7 @@ func TestDurableSnapshotDropsProcessLocalTombstones(t *testing.T) {
567		require.Empty(t, stored.Removed)	567		require.Empty(t, stored.Removed)
568		require.NoError(t, s.Close())	568		require.NoError(t, s.Close())
569			569
570	-	s, err = OpenOffline(o)	570	+	s, err = Open(o)
571		require.NoError(t, err)	571		require.NoError(t, err)
572		defer s.Close()	572		defer s.Close()
573		reopened, err := s.Load()	573		reopened, err := s.Load()

@@ -579,7 +579,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
579		for _, point := range []string{"write", "sync", "rename", "dirsync"} {	579		for _, point := range []string{"write", "sync", "rename", "dirsync"} {
580		t.Run(point, func(t *testing.T) {	580		t.Run(point, func(t *testing.T) {
581		o := options(t)	581		o := options(t)
582	-	s, err := OpenOffline(o)	582	+	s, err := Open(o)
583		require.NoError(t, err)	583		require.NoError(t, err)
584		h, err := s.LoadHosts()	584		h, err := s.LoadHosts()
585		require.NoError(t, err)	585		require.NoError(t, err)

@@ -606,7 +606,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
606		_, err = s.LoadHosts()	606		_, err = s.LoadHosts()
607		require.Error(t, err)	607		require.Error(t, err)
608		require.NoError(t, s.Close())	608		require.NoError(t, s.Close())
609	-	s, err = OpenOffline(o)	609	+	s, err = Open(o)
610		require.NoError(t, err)	610		require.NoError(t, err)
611		defer s.Close()	611		defer s.Close()
612		h, err = s.LoadHosts()	612		h, err = s.LoadHosts()

@@ -622,7 +622,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
622		}	622		}
623		t.Run("generation and restart", func(t *testing.T) {	623		t.Run("generation and restart", func(t *testing.T) {
624		o := options(t)	624		o := options(t)
625	-	s, err := OpenOffline(o)	625	+	s, err := Open(o)
626		require.NoError(t, err)	626		require.NoError(t, err)
627		h, err := s.LoadHosts()	627		h, err := s.LoadHosts()
628		require.NoError(t, err)	628		require.NoError(t, err)

@@ -632,7 +632,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
632		require.NoError(t, s.ReplaceHosts(h.Revision, h.Hosts))	632		require.NoError(t, s.ReplaceHosts(h.Revision, h.Hosts))
633		require.ErrorIs(t, s.Store(snap), ErrStale)	633		require.ErrorIs(t, s.Store(snap), ErrStale)
634		require.NoError(t, s.Close())	634		require.NoError(t, s.Close())
635	-	s, err = OpenOffline(o)	635	+	s, err = Open(o)
636		require.NoError(t, err)	636		require.NoError(t, err)
637		defer s.Close()	637		defer s.Close()
638		got, err := s.Load()	638		got, err := s.Load()

@@ -648,7 +648,7 @@ func TestFencingAndCommitFaults(t *testing.T) {
648		// daemon observation is process-local state and the store refuses it outright,	648		// daemon observation is process-local state and the store refuses it outright,
649		// so it can never become durable authority.	649		// so it can never become durable authority.
650		func TestLocalDaemonObservationIsNeverDurable(t *testing.T) {	650		func TestLocalDaemonObservationIsNeverDurable(t *testing.T) {
651	-	s, err := OpenOffline(options(t))	651	+	s, err := Open(options(t))
652		require.NoError(t, err)	652		require.NoError(t, err)
653		defer s.Close()	653		defer s.Close()
654		err = s.Store(ports.BrokerSnapshot{Epoch: 9, Revision: 1, Daemons: []ports.BrokerDaemonObservation{{	654		err = s.Store(ports.BrokerSnapshot{Epoch: 9, Revision: 1, Daemons: []ports.BrokerDaemonObservation{{

M internal/app/broker_offline.go +1 -1


@@ -247,7 +247,7 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
247	247		}
248	248		defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
249	249
250		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: layout.State})
	250	+	store, err := brokerstore.Open(brokerstore.Options{Dir: layout.State})
251	251		if err != nil {
252	252		return fmt.Errorf("vev: open broker sandbox store: %w", err)
253	253		}


@@ -247,7 +247,7 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
247		}	247		}
248		defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()	248		defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
249			249
250	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: layout.State})	250	+	store, err := brokerstore.Open(brokerstore.Options{Dir: layout.State})
251		if err != nil {	251		if err != nil {
252		return fmt.Errorf("vev: open broker sandbox store: %w", err)	252		return fmt.Errorf("vev: open broker sandbox store: %w", err)
253		}	253		}

M internal/app/broker_offline_local_test.go +1 -1


@@ -255,7 +255,7 @@ func TestBrokerLocalObservationRegistryPublishesLocalFirst(t *testing.T) {
255	255
256	256		stateDir := filepath.Join(shortTempDir(t, "vevs"), "state")
257	257		require.NoError(t, os.MkdirAll(stateDir, 0o700))
258		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: stateDir})
	258	+	store, err := brokerstore.Open(brokerstore.Options{Dir: stateDir})
259	259		require.NoError(t, err)
260	260		t.Cleanup(func() { require.NoError(t, store.Close()) })
261	261


@@ -255,7 +255,7 @@ func TestBrokerLocalObservationRegistryPublishesLocalFirst(t *testing.T) {
255			255
256		stateDir := filepath.Join(shortTempDir(t, "vevs"), "state")	256		stateDir := filepath.Join(shortTempDir(t, "vevs"), "state")
257		require.NoError(t, os.MkdirAll(stateDir, 0o700))	257		require.NoError(t, os.MkdirAll(stateDir, 0o700))
258	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: stateDir})	258	+	store, err := brokerstore.Open(brokerstore.Options{Dir: stateDir})
259		require.NoError(t, err)	259		require.NoError(t, err)
260		t.Cleanup(func() { require.NoError(t, store.Close()) })	260		t.Cleanup(func() { require.NoError(t, store.Close()) })
261			261

A internal/app/broker_production.go +28 -0

@@ -0,0 +1,28 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"path/filepath"
	5	+
	6	+	"github.com/bnema/vev/internal/adapters/brokerconfig"
	7	+	"github.com/bnema/vev/internal/adapters/ipc"
	8	+	"github.com/bnema/vev/internal/platform"
	9	+	)
	10	+
	11	+	// productionBrokerLayout is the single production broker filesystem contract.
	12	+	// Runtime ownership belongs to broker.Supervisor; state persists independently;
	13	+	// configuration remains under the user's vev configuration directory.
	14	+	func productionBrokerConfigPath() string {
	15	+	return filepath.Join(filepath.Dir(platform.ConfigPath()), "broker.json")
	16	+	}
	17	+
	18	+	func productionBrokerLayout() brokerconfig.Layout {
	19	+	root := filepath.Join(platform.StateDir(), "broker")
	20	+	runtime := filepath.Join(ipc.SocketDir(), "broker")
	21	+	return brokerconfig.Layout{
	22	+	Root: root,
	23	+	Runtime: runtime,
	24	+	State: filepath.Join(root, brokerconfig.StateDirName),
	25	+	Log: filepath.Join(root, brokerconfig.LogDirName),
	26	+	Spawn: filepath.Join(runtime, brokerconfig.SpawnDirName),
	27	+	}
	28	+	}

@@ -0,0 +1,28 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"path/filepath"
	5	+
	6	+	"github.com/bnema/vev/internal/adapters/brokerconfig"
	7	+	"github.com/bnema/vev/internal/adapters/ipc"
	8	+	"github.com/bnema/vev/internal/platform"
	9	+	)
	10	+
	11	+	// productionBrokerLayout is the single production broker filesystem contract.
	12	+	// Runtime ownership belongs to broker.Supervisor; state persists independently;
	13	+	// configuration remains under the user's vev configuration directory.
	14	+	func productionBrokerConfigPath() string {
	15	+	return filepath.Join(filepath.Dir(platform.ConfigPath()), "broker.json")
	16	+	}
	17	+
	18	+	func productionBrokerLayout() brokerconfig.Layout {
	19	+	root := filepath.Join(platform.StateDir(), "broker")
	20	+	runtime := filepath.Join(ipc.SocketDir(), "broker")
	21	+	return brokerconfig.Layout{
	22	+	Root: root,
	23	+	Runtime: runtime,
	24	+	State: filepath.Join(root, brokerconfig.StateDirName),
	25	+	Log: filepath.Join(root, brokerconfig.LogDirName),
	26	+	Spawn: filepath.Join(runtime, brokerconfig.SpawnDirName),
	27	+	}
	28	+	}

A internal/app/broker_production_test.go +22 -0

@@ -0,0 +1,22 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"path/filepath"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+	)
	9	+
	10	+	func TestProductionBrokerLayoutUsesDedicatedRuntimeStateAndConfig(t *testing.T) {
	11	+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
	12	+	t.Setenv("XDG_STATE_HOME", t.TempDir())
	13	+	t.Setenv("XDG_CONFIG_HOME", t.TempDir())
	14	+
	15	+	layout := productionBrokerLayout()
	16	+	require.Equal(t, filepath.Join(layout.Root, "state"), layout.State)
	17	+	require.Equal(t, filepath.Join(layout.Runtime, "spawn"), layout.Spawn)
	18	+	require.Equal(t, "broker", filepath.Base(layout.Root))
	19	+	require.Equal(t, "broker", filepath.Base(layout.Runtime))
	20	+	require.Equal(t, "broker.json", filepath.Base(productionBrokerConfigPath()))
	21	+	require.NotEqual(t, layout.Root, layout.Runtime)
	22	+	}

@@ -0,0 +1,22 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"path/filepath"
	5	+	"testing"
	6	+
	7	+	"github.com/stretchr/testify/require"
	8	+	)
	9	+
	10	+	func TestProductionBrokerLayoutUsesDedicatedRuntimeStateAndConfig(t *testing.T) {
	11	+	t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
	12	+	t.Setenv("XDG_STATE_HOME", t.TempDir())
	13	+	t.Setenv("XDG_CONFIG_HOME", t.TempDir())
	14	+
	15	+	layout := productionBrokerLayout()
	16	+	require.Equal(t, filepath.Join(layout.Root, "state"), layout.State)
	17	+	require.Equal(t, filepath.Join(layout.Runtime, "spawn"), layout.Spawn)
	18	+	require.Equal(t, "broker", filepath.Base(layout.Root))
	19	+	require.Equal(t, "broker", filepath.Base(layout.Runtime))
	20	+	require.Equal(t, "broker.json", filepath.Base(productionBrokerConfigPath()))
	21	+	require.NotEqual(t, layout.Root, layout.Runtime)
	22	+	}

M internal/usecase/broker/registry_durable_test.go +5 -5


@@ -32,7 +32,7 @@ func TestRegistryDurableReplaceHostsRestart(t *testing.T) {
32	32		{"add", []ports.BrokerHostRecord{first}}, {"remove", nil}, {"re-add", []ports.BrokerHostRecord{readded}}, {"restart", []ports.BrokerHostRecord{readded}},
33	33		} {
34	34		t.Run(stage.name, func(t *testing.T) {
35		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir})
	35	+	store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
36	36		require.NoError(t, err)
37	37		defer store.Close()
38	38		r, err := NewRegistry(ports.BrokerEpoch(i+1), store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)

@@ -70,7 +70,7 @@ func TestRegistryDurableReplaceHostsRestart(t *testing.T) {
70	70		func TestRegistryDurableConflictDoesNotMutateProjection(t *testing.T) {
71	71		for _, mode := range []string{"stale revision", "policy without new generation"} {
72	72		t.Run(mode, func(t *testing.T) {
73		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
	73	+	store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
74	74		require.NoError(t, err)
75	75		defer store.Close()
76	76		record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy()}

@@ -138,7 +138,7 @@ func TestRegistryReplaceHostsAfterShutdownIsTyped(t *testing.T) {
138	138		// the newest post-CAS publication must always flush; churn cannot leave the
139	139		// registry permanently publishing membership that the store will reject.
140	140		func TestRegistryDurableChurnFlushesLatestMembership(t *testing.T) {
141		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
	141	+	store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
142	142		require.NoError(t, err)
143	143		defer store.Close()
144	144		r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)

@@ -174,7 +174,7 @@ func TestRegistryDurableRoundTripObservedIdentity(t *testing.T) {
174	174		dir := filepath.Join(t.TempDir(), "broker")
175	175		record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy()}
176	176
177		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir})
	177	+	store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
178	178		require.NoError(t, err)
179	179		r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
180	180		require.NoError(t, err)

@@ -201,7 +201,7 @@ func TestRegistryDurableRoundTripObservedIdentity(t *testing.T) {
201	201		r.settle() // flush before reopening
202	202		require.NoError(t, store.Close())
203	203
204		-	reopened, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir})
	204	+	reopened, err := brokerstore.Open(brokerstore.Options{Dir: dir})
205	205		require.NoError(t, err)
206	206		defer reopened.Close()
207	207		persisted, err := reopened.Load()


@@ -32,7 +32,7 @@ func TestRegistryDurableReplaceHostsRestart(t *testing.T) {
32		{"add", []ports.BrokerHostRecord{first}}, {"remove", nil}, {"re-add", []ports.BrokerHostRecord{readded}}, {"restart", []ports.BrokerHostRecord{readded}},	32		{"add", []ports.BrokerHostRecord{first}}, {"remove", nil}, {"re-add", []ports.BrokerHostRecord{readded}}, {"restart", []ports.BrokerHostRecord{readded}},
33		} {	33		} {
34		t.Run(stage.name, func(t *testing.T) {	34		t.Run(stage.name, func(t *testing.T) {
35	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir})	35	+	store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
36		require.NoError(t, err)	36		require.NoError(t, err)
37		defer store.Close()	37		defer store.Close()
38		r, err := NewRegistry(ports.BrokerEpoch(i+1), store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)	38		r, err := NewRegistry(ports.BrokerEpoch(i+1), store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)

@@ -70,7 +70,7 @@ func TestRegistryDurableReplaceHostsRestart(t *testing.T) {
70		func TestRegistryDurableConflictDoesNotMutateProjection(t *testing.T) {	70		func TestRegistryDurableConflictDoesNotMutateProjection(t *testing.T) {
71		for _, mode := range []string{"stale revision", "policy without new generation"} {	71		for _, mode := range []string{"stale revision", "policy without new generation"} {
72		t.Run(mode, func(t *testing.T) {	72		t.Run(mode, func(t *testing.T) {
73	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})	73	+	store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
74		require.NoError(t, err)	74		require.NoError(t, err)
75		defer store.Close()	75		defer store.Close()
76		record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy()}	76		record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy()}

@@ -138,7 +138,7 @@ func TestRegistryReplaceHostsAfterShutdownIsTyped(t *testing.T) {
138		// the newest post-CAS publication must always flush; churn cannot leave the	138		// the newest post-CAS publication must always flush; churn cannot leave the
139		// registry permanently publishing membership that the store will reject.	139		// registry permanently publishing membership that the store will reject.
140		func TestRegistryDurableChurnFlushesLatestMembership(t *testing.T) {	140		func TestRegistryDurableChurnFlushesLatestMembership(t *testing.T) {
141	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})	141	+	store, err := brokerstore.Open(brokerstore.Options{Dir: filepath.Join(t.TempDir(), "broker")})
142		require.NoError(t, err)	142		require.NoError(t, err)
143		defer store.Close()	143		defer store.Close()
144		r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)	144		r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)

@@ -174,7 +174,7 @@ func TestRegistryDurableRoundTripObservedIdentity(t *testing.T) {
174		dir := filepath.Join(t.TempDir(), "broker")	174		dir := filepath.Join(t.TempDir(), "broker")
175		record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy()}	175		record := ports.BrokerHostRecord{Registration: registration(t, "host", 1), Pinned: true, Policy: poolPolicy()}
176			176
177	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir})	177	+	store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
178		require.NoError(t, err)	178		require.NoError(t, err)
179		r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)	179		r, err := NewRegistry(1, store, newTestProbe(1), newManualClock(time.Unix(100, 0)), nil)
180		require.NoError(t, err)	180		require.NoError(t, err)

@@ -201,7 +201,7 @@ func TestRegistryDurableRoundTripObservedIdentity(t *testing.T) {
201		r.settle() // flush before reopening	201		r.settle() // flush before reopening
202		require.NoError(t, store.Close())	202		require.NoError(t, store.Close())
203			203
204	-	reopened, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir})	204	+	reopened, err := brokerstore.Open(brokerstore.Options{Dir: dir})
205		require.NoError(t, err)	205		require.NoError(t, err)
206		defer reopened.Close()	206		defer reopened.Close()
207		persisted, err := reopened.Load()	207		persisted, err := reopened.Load()

M internal/usecase/broker/registry_membership_durable_test.go +2 -2


@@ -35,7 +35,7 @@ func newDurableMutableRegistry(t *testing.T, epoch ports.BrokerEpoch, store port
35	35
36	36		func openDurableStore(t testing.T, dir string, fault func(string) error) brokerstore.Store {
37	37		t.Helper()
38		-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir, Fault: fault})
	38	+	store, err := brokerstore.Open(brokerstore.Options{Dir: dir, Fault: fault})
39	39		require.NoError(t, err)
40	40		t.Cleanup(func() { require.NoError(t, store.Close()) })
41	41		return store

@@ -230,7 +230,7 @@ func TestRegistryReplaceHostsRefusesUnpublishableMembershipBeforeCAS(t *testing.
230	230		func TestRegistryMutableOutcomeUnknownRequiresReopen(t *testing.T) {
231	231		dir := filepath.Join(t.TempDir(), "broker")
232	232		fault := errors.New("power loss")
233		-	// The fault is armed only after OpenOffline so the store can be created and
	233	+	// The fault is armed only after Open so the store can be created and
234	234		// seeded normally; state.json:rename fires after the rename, so the commit
235	235		// point was already crossed when the failure is reported.
236	236		var armed atomic.Bool


@@ -35,7 +35,7 @@ func newDurableMutableRegistry(t *testing.T, epoch ports.BrokerEpoch, store port
35			35
36		func openDurableStore(t testing.T, dir string, fault func(string) error) brokerstore.Store {	36		func openDurableStore(t testing.T, dir string, fault func(string) error) brokerstore.Store {
37		t.Helper()	37		t.Helper()
38	-	store, err := brokerstore.OpenOffline(brokerstore.Options{Dir: dir, Fault: fault})	38	+	store, err := brokerstore.Open(brokerstore.Options{Dir: dir, Fault: fault})
39		require.NoError(t, err)	39		require.NoError(t, err)
40		t.Cleanup(func() { require.NoError(t, store.Close()) })	40		t.Cleanup(func() { require.NoError(t, store.Close()) })
41		return store	41		return store

@@ -230,7 +230,7 @@ func TestRegistryReplaceHostsRefusesUnpublishableMembershipBeforeCAS(t *testing.
230		func TestRegistryMutableOutcomeUnknownRequiresReopen(t *testing.T) {	230		func TestRegistryMutableOutcomeUnknownRequiresReopen(t *testing.T) {
231		dir := filepath.Join(t.TempDir(), "broker")	231		dir := filepath.Join(t.TempDir(), "broker")
232		fault := errors.New("power loss")	232		fault := errors.New("power loss")
233	-	// The fault is armed only after OpenOffline so the store can be created and	233	+	// The fault is armed only after Open so the store can be created and
234		// seeded normally; state.json:rename fires after the rename, so the commit	234		// seeded normally; state.json:rename fires after the rename, so the commit
235		// point was already crossed when the failure is reported.	235		// point was already crossed when the failure is reported.
236		var armed atomic.Bool	236		var armed atomic.Bool