carlos / platform Public

Download this file

1	package host
2
3	import (
4	"context"
5	"encoding/json"
6	"errors"
7	"fmt"
8	"slices"
9	"strings"
10	"sync"
11	"testing"
12	"time"
13
14	"github.com/carlosframework/platform/internal/activator"
15	"github.com/carlosframework/platform/internal/genselect"
16	"github.com/carlosframework/platform/internal/hostbox"
17	"github.com/carlosframework/platform/internal/hostclaim"
18	"github.com/carlosframework/platform/internal/hostguard"
19	"github.com/carlosframework/platform/internal/hostwrite"
20	"github.com/carlosframework/platform/internal/registry"
21	"github.com/superfly/ltx"
22	)
23
24	// plan9D5Sync stands in for the session's replicator. run sees each call's
25	// ordinal and does what litestream's sync would have uploaded by then.
26	type plan9D5Sync struct {
27	mu sync.Mutex
28	calls int
29	run func(n int) error
30	}
31
32	func (s *plan9D5Sync) SyncAndWait(context.Context) error {
33	s.mu.Lock()
34	s.calls++
35	n, run := s.calls, s.run
36	s.mu.Unlock()
37	if run != nil {
38	return run(n)
39	}
40	return nil
41	}
42
43	func (s *plan9D5Sync) count() int {
44	s.mu.Lock()
45	defer s.mu.Unlock()
46	return s.calls
47	}
48
49	// plan9D5Pub is one admitted attempt's publisher: the real assembly
50	// publish, the claim's serving owners and a real producer registry.
51	type plan9D5Pub struct {
52	f *plan9D5Serving
53	w *ClaimWakeAssembly
54	a *claimWakeAttempt
55	producers *ClaimProducers
56	target hostguard.FenceTarget
57	syncer *plan9D5Sync
58	work plan9D5Work
59	mu sync.Mutex
60	current hostclaim.Observed
61	published []hostclaim.Observed
62	}
63
64	func plan9D5Publisher(t *testing.T, f *plan9D5Serving, admitted hostclaim.Observed) *plan9D5Pub {
65	t.Helper()
66	h := admitted.Claim.Holder
67	target := hostguard.FenceTarget{Host: admitted.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "box-1", Region: "ie", Enrollment: "enrollment-1", BootID: "boot-1", ControllerID: "controller-1", ClaimETag: admitted.ETag}
68	producers, err := NewClaimProducers(target, "start-1", func(context.Context, hostguard.FenceTarget, string) error { return nil })
69	if err != nil {
70	t.Fatal(err)
71	}
72	live, cancel := context.WithCancel(context.Background())
73	t.Cleanup(cancel)
74	p := &plan9D5Pub{f: f, producers: producers, target: target, syncer: &plan9D5Sync{}, work: plan9D5Work{identities: &[]hostwrite.Identity{}}, current: admitted}
75	p.w = &ClaimWakeAssembly{Publication: ClaimOwnerDispatch{Serving: f.owners.owners()}, Producers: func(context.Context, hostclaim.Observed) (*ClaimProducers, error) { return producers, nil }}
76	p.a = &claimWakeAttempt{observed: admitted, held: recoveryHeld{}, ctx: live, cancel: cancel, done: make(chan struct{}), launch: &activator.RoleLaunch{Target: target}, syncer: p.syncer}
77	return p
78	}
79
80	func (p *plan9D5Pub) read(context.Context, string) (hostclaim.Observed, error) {
81	p.mu.Lock()
82	defer p.mu.Unlock()
83	return p.current, nil
84	}
85
86	func (p *plan9D5Pub) markPublished(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) {
87	p.mu.Lock()
88	defer p.mu.Unlock()
89	p.published = append(p.published, o)
90	return o, nil
91	}
92
93	func (p *plan9D5Pub) publishedCount() int {
94	p.mu.Lock()
95	defer p.mu.Unlock()
96	return len(p.published)
97	}
98
99	func (p *plan9D5Pub) publish(ctx context.Context, in ClaimLaunchInputs) (hostclaim.Observed, error) {
100	selected := claimLeaseWork{Work: p.work, Alive: p.a.ctx, Check: p.a.leaseErr}
101	admitted, _ := p.read(ctx, "")
102	return p.w.publish(ctx, p.a, admitted, admitted, in, selected, p.read, p.markPublished)
103	}
104
105	// drain is root's fence of the attempt's launch tuple.
106	func (p *plan9D5Pub) drain(ctx context.Context) (DrainAcknowledgement, error) {
107	t := p.target
108	return p.producers.DrainClaim(ctx, StopRequest{Host: t.Host, Incarnation: t.Incarnation, InstallationStamp: t.Stamp, WriterSession: t.Session, Box: t.Box, Enrollment: t.Enrollment, BootID: t.BootID, ControllerID: t.ControllerID}, "fence-1")
109	}
110
111	func (p *plan9D5Pub) outcome() (*ClaimPublicationReceipt, error) {
112	p.a.mu.Lock()
113	defer p.a.mu.Unlock()
114	return p.a.receipt, p.a.pubErr
115	}
116
117	// plan9D5RootBody is a level-0 [1, 1] file, the first upload's shape.
118	func plan9D5RootBody(t *testing.T) []byte {
119	t.Helper()
120	h := ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, MinTXID: 1, MaxTXID: 1}
121	body, err := h.MarshalBinary()
122	if err != nil {
123	t.Fatal(err)
124	}
125	return body
126	}
127
128	func plan9D5RootKey(gen string) string {
129	return fmt.Sprintf("instances/%s/g-%s/0000/%s", recoveryHost, gen, ltx.FormatFilename(1, 1))
130	}
131
132	const plan9D5MarkerKey = "instances/" + recoveryHost + "/generation.json"
133
134	// plan9D5Fresh is a serving session admitted cold into g1 from g0's baseline.
135	func plan9D5Fresh(t *testing.T) (*plan9D5Serving, hostclaim.Observed, ClaimLaunchInputs) {
136	t.Helper()
137	f := plan9D5Fixture(t)
138	f.file(t, "g0", 9, 1, 11)
139	f.marker(t, "g0")
140	admitted := plan9D5Observed(hostclaim.Holder{Stamp: 7, Session: 7, Generation: "g1", WriterBox: "box-1",
141	Admissions: []hostclaim.AdmissionRecord{{Session: 7, Box: "box-1", Generation: "g1", Predecessor: "g0", Region: "ie", PredecessorRegion: "ie", Mode: "cold"}}})
142	return f, admitted, ClaimLaunchInputs{Identity: f.launch.Identity, Admission: hostclaim.Admission{NewGeneration: "g1"}}
143	}
144
145	// The review's P1-3: replication opens before the session's first upload
146	// lands. Publication syncs the current replicator first and names only the
147	// root that sync acknowledged.
148	func TestPlan9D5PublisherSyncsTheSessionRootBeforeTheMarker(t *testing.T) {
149	f, admitted, in := plan9D5Fresh(t)
150	ctx := t.Context()
151	root := plan9D5RootBody(t)
152	// Without the sync the generation has no root yet: the marker is refused.
153	if _, err := f.owners.PublishFenced(ctx, admitted, in, ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}); !errors.Is(err, hostbox.ErrChanged) {
154	t.Fatalf("a marker before the session's first upload: %v", err)
155	}
156	p := plan9D5Publisher(t, f, admitted)
157	p.syncer.run = func(n int) error {
158	if n == 1 {
159	f.bucket.Put("data", plan9D5RootKey("g1"), root)
160	}
161	return nil
162	}
163	if _, err := p.publish(ctx, in); err != nil {
164	t.Fatal(err)
165	}
166	receipt, err := p.outcome()
167	if err != nil || receipt == nil || receipt.Root.Key != plan9D5RootKey("g1") || receipt.Root.VersionID != f.bucket.LatestID("data", plan9D5RootKey("g1")) || receipt.MarkerVersion != f.bucket.LatestID("data", plan9D5MarkerKey) {
168	t.Fatalf("receipt %+v %v", receipt, err)
169	}
170	if p.syncer.count() != 1 || p.publishedCount() != 1 {
171	t.Fatalf("syncs=%d published=%d", p.syncer.count(), p.publishedCount())
172	}
173	body, _ := f.bucket.Current("data", plan9D5MarkerKey)
174	var m genselect.Marker
175	if json.Unmarshal(body, &m) != nil || m.Op != "g1" || m.Box != "box-1" || m.SnapshotTXID != 0 {
176	t.Fatalf("marker %s", body)
177	}
178	if err := f.owners.publicationEvidence(ctx, admitted); err != nil {
179	t.Fatal(err)
180	}
181	select {
182	case <-p.a.publisher:
183	default:
184	t.Fatal("the publisher did not join")
185	}
186	// The publisher is a producer of the launch: root's drain names it.
187	ack, err := p.drain(ctx)
188	if err != nil || !slices.Contains(ack.ProducerIDs, "publisher") {
189	t.Fatalf("drain %+v %v", ack, err)
190	}
191	// The evidence is bound to the exact root version the marker named: a
192	// late duplicate of the upload does not replace it, a removed root does.
193	f.bucket.Put("data", plan9D5RootKey("g1"), root)
194	if err := f.owners.publicationEvidence(ctx, admitted); err != nil {
195	t.Fatalf("a late duplicate root: %v", err)
196	}
197	f.bucket.BareDelete("data", plan9D5RootKey("g1"))
198	if err := f.owners.publicationEvidence(ctx, admitted); err == nil {
199	t.Fatal("evidence for a root that is gone")
200	}
201	}
202
203	func TestPlan9D5PublisherSyncFailureWritesNoMarker(t *testing.T) {
204	f, admitted, in := plan9D5Fresh(t)
205	ctx := t.Context()
206	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
207	p := plan9D5Publisher(t, f, admitted)
208	failure := errors.New("replica sync failed")
209	p.syncer.run = func(int) error { return failure }
210	before := f.bucket.LatestID("data", plan9D5MarkerKey)
211	if _, err := p.publish(ctx, in); !errors.Is(err, failure) {
212	t.Fatalf("publish after a failed sync: %v", err)
213	}
214	if f.bucket.LatestID("data", plan9D5MarkerKey) != before || p.publishedCount() != 0 {
215	t.Fatal("a marker or publication without an acknowledged sync")
216	}
217	if receipt, err := p.outcome(); receipt != nil || !errors.Is(err, failure) {
218	t.Fatalf("outcome %+v %v", receipt, err)
219	}
220	// One publisher per attempt: a second call never re-registers.
221	if _, err := p.publish(ctx, in); !errors.Is(err, hostbox.ErrChanged) {
222	t.Fatalf("second publisher: %v", err)
223	}
224	}
225
226	// plan9D5HeldMarkers holds a marker request in flight until released.
227	type plan9D5HeldMarkers struct {
228	plan9D5Markers
229	entered, release chan struct{}
230	}
231
232	func (m plan9D5HeldMarkers) PutMarker(ctx context.Context, bucket, key string, body []byte, expected string, check func(context.Context) error) (string, error) {
233	close(m.entered)
234	<-m.release
235	return m.plan9D5Markers.PutMarker(ctx, bucket, key, body, expected, check)
236	}
237
238	// The review's P2-1 (spec model 29): a fenced session whose publication
239	// listed the marker before its successor wrote one must not overwrite it.
240	// Root's drain cancels the publisher, waits for its request to return, and
241	// the store's check refuses the write.
242	func TestPlan9D5PublisherCancelledByRootFenceWritesNoStaleMarker(t *testing.T) {
243	t.Parallel()
244	f, admitted, in := plan9D5Fresh(t)
245	ctx := t.Context()
246	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
247	held := plan9D5HeldMarkers{plan9D5Markers{f.bucket}, make(chan struct{}), make(chan struct{})}
248	f.owners.Sources.Markers = held
249	p := plan9D5Publisher(t, f, admitted)
250	published := make(chan error, 1)
251	go func() {
252	_, err := p.publish(ctx, in)
253	published <- err
254	}()
255	<-held.entered
256	drained := make(chan error, 1)
257	var ack DrainAcknowledgement
258	go func() {
259	var err error
260	ack, err = p.drain(ctx)
261	drained <- err
262	}()
263	select {
264	case err := <-drained:
265	t.Fatalf("the drain acknowledged a publisher with a request in flight: %v", err)
266	case <-time.After(100 * time.Millisecond):
267	}
268	// The successor's marker lands after the fence.
269	f.marker(t, "g2")
270	successor := f.bucket.LatestID("data", plan9D5MarkerKey)
271	close(held.release)
272	if err := <-published; !errors.Is(err, context.Canceled) {
273	t.Fatalf("fenced publication: %v", err)
274	}
275	if err := <-drained; err != nil || !slices.Contains(ack.ProducerIDs, "publisher") {
276	t.Fatalf("drain %+v %v", ack, err)
277	}
278	if f.bucket.LatestID("data", plan9D5MarkerKey) != successor || p.publishedCount() != 0 {
279	t.Fatal("the fenced publisher overwrote its successor's marker")
280	}
281	if receipt, _ := p.outcome(); receipt != nil {
282	t.Fatalf("receipt for a fenced publication: %+v", receipt)
283	}
284	if err := f.owners.publicationEvidence(ctx, admitted); !errors.Is(err, hostbox.ErrChanged) {
285	t.Fatalf("evidence for a fenced publication: %v", err)
286	}
287	}
288
289	// A drained producer set refuses a publisher that has not started yet.
290	func TestPlan9D5PublisherRefusedAfterTheFence(t *testing.T) {
291	f, admitted, in := plan9D5Fresh(t)
292	ctx := t.Context()
293	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
294	p := plan9D5Publisher(t, f, admitted)
295	if _, err := p.drain(ctx); err != nil {
296	t.Fatal(err)
297	}
298	before := f.bucket.LatestID("data", plan9D5MarkerKey)
299	if _, err := p.publish(ctx, in); !errors.Is(err, ErrClaimStopIdentity) || p.syncer.count() != 0 {
300	t.Fatalf("publisher after the fence: %v syncs=%d", err, p.syncer.count())
301	}
302	if f.bucket.LatestID("data", plan9D5MarkerKey) != before {
303	t.Fatal("a marker after the fence")
304	}
305	}
306
307	// The selection check runs immediately before the marker request: a claim
308	// that no longer names the session writes nothing.
309	func TestPlan9D5PublisherChecksTheSelectionBeforeTheMarker(t *testing.T) {
310	f, admitted, in := plan9D5Fresh(t)
311	ctx := t.Context()
312	f.bucket.Put("data", plan9D5RootKey("g1"), plan9D5RootBody(t))
313	p := plan9D5Publisher(t, f, admitted)
314	p.work.err = hostbox.ErrChanged
315	before := f.bucket.LatestID("data", plan9D5MarkerKey)
316	if _, err := p.publish(ctx, in); !errors.Is(err, hostbox.ErrChanged) {
317	t.Fatalf("publication under a changed selection: %v", err)
318	}
319	if f.bucket.LatestID("data", plan9D5MarkerKey) != before || len(*p.work.identities) == 0 {
320	t.Fatal("the marker moved, or the selection was never checked")
321	}
322	}
323
324	func plan9D5ZeroPosition(t *testing.T) (*plan9D5Serving, hostclaim.Observed, ClaimLaunchInputs) {
325	t.Helper()
326	f := plan9D5Fixture(t)
327	admitted := plan9D5Observed(hostclaim.Holder{Stamp: 5, Session: 5, Generation: "fresh-1", WriterBox: "box-1",
328	Admissions: []hostclaim.AdmissionRecord{{Session: 5, Box: "box-1", Generation: "fresh-1", Region: "ie", Mode: "cold"}}})
329	return f, admitted, ClaimLaunchInputs{Identity: f.launch.Identity, Empty: true, Admission: hostclaim.Admission{NewGeneration: "fresh-1"}}
330	}
331
332	// R8-3: an empty bootstrap's app may not have committed anything when it is
333	// ready. No marker names a generation without a root; the publisher waits
334	// in the background and publishes after the first durable root.
335	func TestPlan9D5ZeroPositionBootstrapPublishesAfterTheFirstRoot(t *testing.T) {
336	t.Parallel()
337	f, admitted, in := plan9D5ZeroPosition(t)
338	ctx := t.Context()
339	root := plan9D5RootBody(t)
340	var mu sync.Mutex
341	committed := false
342	p := plan9D5Publisher(t, f, admitted)
343	p.syncer.run = func(int) error {
344	mu.Lock()
345	defer mu.Unlock()
346	if committed && f.bucket.LatestID("data", plan9D5RootKey("fresh-1")) == "" {
347	f.bucket.Put("data", plan9D5RootKey("fresh-1"), root)
348	}
349	return nil
350	}
351	if _, err := p.publish(ctx, in); err != nil {
352	t.Fatal(err)
353	}
354	if _, ok := f.bucket.Current("data", plan9D5MarkerKey); ok || p.publishedCount() != 0 {
355	t.Fatal("a marker or publication before the generation's root")
356	}
357	if receipt, _ := p.outcome(); receipt != nil {
358	t.Fatal("a receipt before the generation's root")
359	}
360	// A non-empty admission never defers: its restored root must exist.
361	if _, err := f.owners.PublishFenced(ctx, admitted, ClaimLaunchInputs{Admission: in.Admission}, f.publication()); !errors.Is(err, genselect.ErrEmpty) {
362	t.Fatalf("a non-empty admission without a root: %v", err)
363	}
364	mu.Lock()
365	committed = true
366	mu.Unlock()
367	select {
368	case <-p.a.publisher:
369	case <-time.After(20 * time.Second):
370	t.Fatal("the publisher never published the first root")
371	}
372	receipt, err := p.outcome()
373	if err != nil || receipt == nil || receipt.Root.Key != plan9D5RootKey("fresh-1") || p.publishedCount() != 1 {
374	t.Fatalf("receipt %+v %v published=%d", receipt, err, p.publishedCount())
375	}
376	body, _ := f.bucket.Current("data", plan9D5MarkerKey)
377	var m genselect.Marker
378	if json.Unmarshal(body, &m) != nil || m.Op != "fresh-1" || m.SnapshotTXID != 0 || !strings.Contains(string(body), `"snapshot_txid":0`) {
379	t.Fatalf("marker %s", body)
380	}
381	if err := f.owners.publicationEvidence(ctx, admitted); err != nil {
382	t.Fatal(err)
383	}
384	}
385
386	func TestPlan9D5ZeroPositionPublisherEndsAtTheFence(t *testing.T) {
387	f, admitted, in := plan9D5ZeroPosition(t)
388	ctx := t.Context()
389	p := plan9D5Publisher(t, f, admitted)
390	if _, err := p.publish(ctx, in); err != nil {
391	t.Fatal(err)
392	}
393	ack, err := p.drain(ctx)
394	if err != nil || !slices.Contains(ack.ProducerIDs, "publisher") {
395	t.Fatalf("drain %+v %v", ack, err)
396	}
397	select {
398	case <-p.a.publisher:
399	default:
400	t.Fatal("the drain acknowledged before the waiting publisher joined")
401	}
402	if receipt, err := p.outcome(); receipt != nil || !errors.Is(err, context.Canceled) {
403	t.Fatalf("outcome %+v %v", receipt, err)
404	}
405	if _, ok := f.bucket.Current("data", plan9D5MarkerKey); ok || p.publishedCount() != 0 {
406	t.Fatal("a fenced zero-position publisher published")
407	}
408	}
409
410	// The review's P1-4: no C8 provider, no serving wake. Select refuses, typed,
411	// before a generation is minted, any lease is taken or anything launches.
412	func TestPlan9D5ServingWithoutFilesProviderRefusesBeforeLaunch(t *testing.T) {
413	f := plan9D5Fixture(t)
414	f.owners.Sources.Files = nil
415	created := plan9D5Observed(hostclaim.Holder{Stamp: 3})
416	var refusal *ClaimDependencyError
417	_, err := f.owners.Select(t.Context(), plan9D5Route, created, activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second})
418	if !errors.As(err, &refusal) || !errors.Is(err, hostwrite.ErrUnavailable) || len(refusal.Contracts) != 1 || !strings.Contains(refusal.Contracts[0], "C8") || f.seq != 0 {
419	t.Fatalf("select without a files provider: %v seq=%d", err, f.seq)
420	}
421	}
422
423	// The C8 provider gets the start's captured work, after the identity proof.
424	func TestPlan9D5FilesProviderRunsUnderTheCapturedWork(t *testing.T) {
425	f := plan9D5Fixture(t)
426	admitted := plan9D5Observed(hostclaim.Holder{Stamp: 7, Session: 7, Generation: "g1", WriterBox: "box-1"})
427	in := ClaimLaunchInputs{Identity: f.launch.Identity, Admission: hostclaim.Admission{NewGeneration: "g1"}}
428	var seen []hostwrite.Identity
429	if err := f.owners.Materialize(t.Context(), admitted, in, plan9D5Work{identities: &seen}); err != nil {
430	t.Fatal(err)
431	}
432	if len(seen) != 2 || len(f.files) != 1 {
433	t.Fatalf("work uses %v, files %v", seen, f.files)
434	}
435	}
436
437	// A moves owner set without its own fenced publisher keeps a typed refusal,
438	// never the unfenced Publish.
439	func TestPlan9D5FencedPublicationRoutesByKind(t *testing.T) {
440	var calls []string
441	moves := plan9D5Owner{"moves", &calls}.owners()
442	d := ClaimOwnerDispatch{Moves: &moves}
443	var refusal *ClaimDependencyError
444	_, err := d.PublishFenced(t.Context(), plan9D5Kind(hostclaim.KindLanding), ClaimLaunchInputs{}, ClaimPublication{})
445	if !errors.As(err, &refusal) || len(calls) != 0 {
446	t.Fatalf("moves without a fenced publisher: %v %v", err, calls)
447	}
448	if _, err := d.PublishFenced(t.Context(), plan9D5Kind(hostclaim.KindServing), ClaimLaunchInputs{}, ClaimPublication{}); !errors.Is(err, hostwrite.ErrUnavailable) {
449	t.Fatalf("serving without owners: %v", err)
450	}
451	}
452
453	// Probe readiness uses the exact attempt the start passed to Start.
454	func TestPlan9D5ProbeStartCapturesItsAttempt(t *testing.T) {
455	probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
456	_, c, ops, _, admitted := plan9D2Landing(t, hostclaim.PhaseLandingProbing, probe, true)
457	s, err := c.runtime.Prepare(t.Context(), registry.Route{Host: recoveryHost}, admitted, probe)
458	if err != nil {
459	t.Fatal(err)
460	}
461	a := c.wake.get(recoveryHost)
462	a.mu.Lock()
463	before := a.started
464	a.mu.Unlock()
465	if before != nil {
466	t.Fatal("an attempt captured before Start")
467	}
468	attempt := s.CurrentAttempt()
469	a.mu.Lock()
470	started := a.started
471	a.mu.Unlock()
472	if started == nil || *started != attempt || attempt.ID != "attempt" || slices.Index(ops.calls, "current") < 0 {
473	t.Fatalf("captured %+v for %+v", started, attempt)
474	}
475	}
476