carlos / platform Public

Download this file

1	//go:build linux
2
3	package activator
4
5	import (
6	"context"
7	"errors"
8	"path/filepath"
9	"slices"
10	"sync"
11	"testing"
12	"time"
13
14	"github.com/carlosframework/platform/internal/hostclaim"
15	"github.com/carlosframework/platform/internal/hostwrite"
16	"github.com/carlosframework/platform/internal/registry"
17	)
18
19	// The held wake check on the claimed path (CARLOS_HOST_CLAIM; settings plan
20	// 4 Task 7). Its launcher execs the claim inputs' file, so a retained copy
21	// cannot be injected: the wake asks root, waits for the install, and the
22	// launch's before hook refuses a hashed build that is not the target.
23
24	type heldClaimed struct {
25	a *Service
26	b *planGenBackend
27	f *rootLaunchFixture
28	log *execLog
29	rt registry.Route
30	mu sync.Mutex
31	reserved int
32	failed int
33	}
34
35	func (c *heldClaimed) counts() (reserved, failed, launches int) {
36	c.mu.Lock()
37	reserved, failed = c.reserved, c.failed
38	c.mu.Unlock()
39	c.f.root.mu.Lock()
40	defer c.f.root.mu.Unlock()
41	return reserved, failed, len(c.f.root.launches)
42	}
43
44	// newHeldClaimed is a claims-on Service whose wakes go through doClaimWake
45	// and launch through the root fixture. hashed is the build the launch
46	// snapshot hashes and names; live is what live/<host> names at the start.
47	func newHeldClaimed(t *testing.T, live, hashed string) *heldClaimed {
48	t.Helper()
49	f := newRootLaunchFixture(t)
50	f.root.grant = launchedChild(t)
51	host := f.launch.Target.Host
52	obs := claimedObs(host)
53	obs.Claim.Holder.WriterBox = "box"
54	admitted := probeAdmitted(obs)
55	work, _ := (hostwrite.Config{}).Select(t.Context(), host, "spawn")
56	c := &heldClaimed{f: f, b: newPlanGenBackend()}
57	c.b.fakeBackend.setLive(live)
58	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
59	if err != nil {
60	t.Fatal(err)
61	}
62	var mu sync.Mutex
63	reads := 0
64	c.a = New(Config{Idle: time.Hour, HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Bucket: "data",
65	Read: func(context.Context, string) (hostclaim.Observed, error) {
66	mu.Lock()
67	defer mu.Unlock()
68	if reads++; reads > 1 {
69	return admitted, nil
70	}
71	return obs, nil
72	},
73	StopForeign: func(context.Context, string) error { return nil },
74	Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil },
75	Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
76	return admitted, nil
77	},
78	FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil },
79	Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
80	s := rootLaunched(t, f, claimedStartFor(admitted, work, "v1"))
81	s.Staged, s.Version = claimedBuildOf(t, f, hashed), hashed
82	s.Replicate = func(context.Context, string) (Rep, error) { return &fakeRep{}, nil }
83	s.Reserve = func(context.Context, int) error { c.mu.Lock(); c.reserved++; c.mu.Unlock(); return nil }
84	s.Failed = func(context.Context, int) error { c.mu.Lock(); c.failed++; c.mu.Unlock(); return nil }
85	return s, nil
86	}}}, c.b, reg)
87	c.log = &execLog{spawns: func() int { f.root.mu.Lock(); defer f.root.mu.Unlock(); return len(f.root.launches) }}
88	c.a.BeforeExec = c.log.hook
89	c.rt = registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: filepath.Join(f.runDir, "app.sock"), DBPath: filepath.Join(f.launchDir, "app.db")}
90	return c
91	}
92
93	// A claimed held wake on a stale live link asks root for the target and
94	// waits for the install; the launch then snapshots the target and runs it,
95	// recorded first.
96	func TestAClaimedHeldWakeWaitsForRootsInstall(t *testing.T) {
97	t.Parallel()
98	c := newHeldClaimed(t, "v3", "v1")
99	c.a.HeldTarget = heldOn("v1")
100	r := &asks{then: func(string, string) { c.b.fakeBackend.setLive("v1") }}
101	c.a.RequestStage = r.ask
102	done, err := c.a.beginRequest(t.Context(), c.rt)
103	if err != nil {
104	t.Fatalf("the claimed held wake: %v", err)
105	}
106	done()
107	if got := r.seen(); !slices.Equal(got, []string{c.rt.Host + "@v1"}) {
108	t.Fatalf("asked root %v, want v1", got)
109	}
110	calls, at, _ := c.log.seen()
111	if _, _, launches := c.counts(); !slices.Equal(calls, []string{c.rt.Host + "@v1"}) || at[0] != 0 || launches != 1 {
112	t.Fatalf("records %v at exec counts %v, %d launches; want v1 recorded, then launched once", calls, at, launches)
113	}
114	}
115
116	// The race half: root installed something other than the target between
117	// the wait and the launch snapshot. The before hook refuses it before the
118	// record: nothing records or execs, no failed attempt is written, and the
119	// start loop does not retry it (ruling R-EXEC-CLAIMRETRY's shape).
120	func TestAClaimedHeldWakeRefusesAnIdentityThatIsNotItsTarget(t *testing.T) {
121	c := newHeldClaimed(t, "v1", "v2")
122	c.a.HeldTarget = heldOn("v1")
123	c.a.RequestStage = func(string, string) { t.Error("asked root, with the target already installed") }
124	began := time.Now()
125	_, err := c.a.beginRequest(t.Context(), c.rt)
126	if !errors.Is(err, errStaleLink) {
127	t.Fatalf("err = %v, want errStaleLink", err)
128	}
129	waitFor(t, "the claimed wake to end", func() bool { return c.a.entryGone(c.rt.Host) })
130	if calls, _, _ := c.log.seen(); len(calls) != 0 {
131	t.Fatalf("records %v: a build that is not the target was recorded", calls)
132	}
133	if reserved, failed, launches := c.counts(); launches != 0 || failed != 0 || reserved != 1 {
134	t.Fatalf("reserved=%d failed=%d launches=%d, want one attempt, nothing run, no failed attempt", reserved, failed, launches)
135	}
136	if took := time.Since(began); took >= 5*time.Second {
137	t.Fatalf("took %v: the refusal waited out a retry backoff", took)
138	}
139	if got := memberSafeWakeReason(err); got != memberSafeWakeReason(ErrFenced) {
140	t.Fatalf("member reads %q, want the fence's wording", got)
141	}
142	}
143