amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "bytes"
5 "context"
6 "crypto/ed25519"
7 "crypto/rand"
8 "errors"
9 "os"
10 "path/filepath"
11 "slices"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/hostactors"
17 "github.com/carlosframework/platform/internal/hostbox"
18 "github.com/carlosframework/platform/internal/hostclaim"
19 "github.com/carlosframework/platform/internal/hostfloor"
20 "github.com/carlosframework/platform/internal/hostguard"
21 "github.com/carlosframework/platform/internal/hostops"
22 "github.com/carlosframework/platform/internal/hostrelocate"
23 "github.com/carlosframework/platform/internal/hostwrite"
24 "github.com/carlosframework/platform/internal/lease"
25 "github.com/carlosframework/platform/internal/lease/leasetest"
26 "github.com/carlosframework/platform/internal/registry"
27 "github.com/carlosframework/platform/internal/release"
28)
29
30// finishRecorder is a Finish that records each host it is handed and
31// answers err.
32type finishRecorder struct {
33 calls []string
34 err error
35}
36
37func (f *finishRecorder) finish(_ context.Context, host, op string) error {
38 f.calls = append(f.calls, host+":"+op)
39 return f.err
40}
41
42// Carry 8: the completion pass hands Finish only a claim serving on this
43// box whose completion names this box. The completion-only runtime has no
44// Box; it asks Self each pass, and a pass that cannot answer finishes
45// nothing.
46func TestCompletionPassFinishesOnlyWhatServesOnThisBox(t *testing.T) {
47 ctx := t.Context()
48 store := release.NewDirStore(t.TempDir())
49 passClaim(t, store, "mine.example", hostclaim.Op{})
50 completeOn(t, store, "mine.example", "rel-m", "dst", false)
51 // Serving on spa, its completion naming spa: spa's to finish.
52 passClaim(t, store, "theirs.example", hostclaim.Op{})
53 completeOn(t, store, "theirs.example", "rel-t", "spa", false)
54 // Serving on spa while its record names dst: not serving here, so not
55 // this box's to finish, whatever the record says.
56 passClaim(t, store, "moved.example", hostclaim.Op{})
57 o := completeOn(t, store, "moved.example", "rel-x", "dst", false)
58 if _, err := hostclaim.Update(ctx, store, o, func(c *hostclaim.Claim) error { c.Op.Exec = "spa"; return nil }); err != nil {
59 t.Fatal(err)
60 }
61 // Serving on dst while its record names spa: the completion is spa's.
62 passClaim(t, store, "foreign.example", hostclaim.Op{})
63 o = completeOn(t, store, "foreign.example", "rel-f", "spa", false)
64 if _, err := hostclaim.Update(ctx, store, o, func(c *hostclaim.Claim) error { c.Op.Exec = "dst"; return nil }); err != nil {
65 t.Fatal(err)
66 }
67 f := &finishRecorder{}
68 hosts := func() ([]string, error) {
69 return []string{"mine.example", "theirs.example", "moved.example", "foreign.example"}, nil
70 }
71 r := &RelocationRuntime{Finish: f.finish, Hosts: hosts, Self: func(context.Context) (string, error) { return "dst", nil }}
72 r.once(ctx, store)
73 if !slices.Equal(f.calls, []string{"mine.example:rel-m"}) {
74 t.Fatalf("Finish was handed %v, want only mine.example", f.calls)
75 }
76 // No answer to "which box is this": nothing is finished.
77 f.calls = nil
78 buf := captureRacyLog(t)
79 r = &RelocationRuntime{Finish: f.finish, Hosts: hosts, Self: func(context.Context) (string, error) { return "", errors.New("no verified identity") }}
80 r.once(ctx, store)
81 if len(f.calls) != 0 || !strings.Contains(buf.String(), "no verified identity") {
82 t.Fatalf("finished %v without knowing this box:\n%s", f.calls, buf.String())
83 }
84}
85
86// Carry 8 and review M4: ErrChanged from the hold release, after the
87// pass's own serving-here check passed, is never taken as finished: it is
88// said once (a completion stuck that way keeps the barrier closed) and
89// asked again next pass. Every other refusal is alerted once and retried
90// too (carry 6: a probe-fence owner Finish cannot find is ErrRefused).
91func TestCompletionPassRetriesARefusedFinishAndAChangedClaim(t *testing.T) {
92 ctx := t.Context()
93 store := release.NewDirStore(t.TempDir())
94 passClaim(t, store, "a.example", hostclaim.Op{})
95 completeOn(t, store, "a.example", "rel-a", "dst", false)
96 buf := captureRacyLog(t)
97 f := &finishRecorder{err: hostbox.ErrChanged}
98 r := &RelocationRuntime{Finish: f.finish, Box: "dst"}
99 r.once(ctx, store)
100 r.once(ctx, store)
101 if len(f.calls) != 2 || strings.Count(buf.String(), "relocate a.example: finish: "+hostbox.ErrChanged.Error()) != 1 {
102 t.Fatalf("a changed claim: calls %v, log:\n%s", f.calls, buf.String())
103 }
104 f.calls, f.err = nil, errors.Join(hostrelocate.ErrRefused, errors.New("a.example has no probe-fence owner"))
105 r.once(ctx, store)
106 r.once(ctx, store)
107 if len(f.calls) != 2 || strings.Count(buf.String(), "no probe-fence owner") != 1 {
108 t.Fatalf("a refused Finish: calls %v (want a retry each pass), log:\n%s", f.calls, buf.String())
109 }
110 if obs, err := hostclaim.Read(ctx, store, "a.example"); err != nil {
111 t.Fatal(err)
112 } else if done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim); err != nil || !ok || done.Discharged {
113 t.Fatalf("the record was marked over a refusal: %+v %v %v", done, ok, err)
114 }
115}
116
117// Carry 1: a completion record that cannot be read is alerted and never
118// read as "no completion": Finish is not called over it, and the next pass
119// reads it again.
120func TestCompletionPassAlertsAnUnreadableRecord(t *testing.T) {
121 ctx := t.Context()
122 store := release.NewDirStore(t.TempDir())
123 passClaim(t, store, "a.example", hostclaim.Op{})
124 o := completeOn(t, store, "a.example", "rel-a", "dst", false)
125 if _, err := hostclaim.Update(ctx, store, o, func(c *hostclaim.Claim) error {
126 c.Op.Journal[hostclaim.JournalCompleted] = "not a record"
127 return nil
128 }); err != nil {
129 t.Fatal(err)
130 }
131 buf := captureRacyLog(t)
132 f := &finishRecorder{}
133 r := &RelocationRuntime{Finish: f.finish, Box: "dst"}
134 r.once(ctx, store)
135 if len(f.calls) != 0 || !strings.Contains(buf.String(), "relocate a.example: finish:") {
136 t.Fatalf("an unreadable record: calls %v, log:\n%s", f.calls, buf.String())
137 }
138}
139
140// Carry 3: the sweep never runs beside a relocation journal write. A
141// write in flight (its directory made, its temporary not yet renamed)
142// holds relocationJournals shared; the pass's sweep waits for it, then
143// keeps the journal it committed. Without the lock the sweep removes the
144// directory as a write's leftover under the write.
145func TestRelocatePassSweepWaitsForAJournalWrite(t *testing.T) {
146 t.Parallel()
147 ctx := t.Context()
148 store := release.NewDirStore(t.TempDir())
149 passClaim(t, store, "a.example", relocatingOp("rel-a"))
150 data := t.TempDir()
151 dir := filepath.Join(data, "relocate", "ops", "rel-a")
152 relocationJournals.RLock()
153 locked := true
154 defer func() {
155 if locked {
156 relocationJournals.RUnlock()
157 }
158 }()
159 if err := os.MkdirAll(dir, 0o700); err != nil {
160 t.Fatal(err)
161 }
162 temp := filepath.Join(dir, ".attempts.json.123")
163 if err := os.WriteFile(temp, []byte("{"), 0o600); err != nil {
164 t.Fatal(err)
165 }
166 r := &RelocationRuntime{Box: "dst", DataDir: data}
167 done := make(chan struct{})
168 go func() { r.once(ctx, store); close(done) }()
169 select {
170 case <-done:
171 t.Fatal("the sweep ran beside a journal write")
172 case <-time.After(300 * time.Millisecond):
173 }
174 // The write lands: its journal committed, its temporary gone.
175 if err := os.WriteFile(filepath.Join(dir, "attempts.json"), []byte(`{"v":1,"host":"a.example","op":"rel-a"}`), 0o600); err != nil {
176 t.Fatal(err)
177 }
178 if err := os.Remove(temp); err != nil {
179 t.Fatal(err)
180 }
181 relocationJournals.RUnlock()
182 locked = false
183 <-done
184 if _, err := os.Stat(filepath.Join(dir, "attempts.json")); err != nil {
185 t.Fatalf("the running op's journal was swept: %v", err)
186 }
187}
188
189// Carry 3, the writers' half: every relocation journal write this process
190// makes outside the pass's own steps holds relocationJournals shared, so
191// it waits while a sweep holds it.
192func TestRelocationJournalWritesWaitForTheSweep(t *testing.T) {
193 t.Parallel()
194 data := t.TempDir()
195 fence, err := relocationProbeFencePath(data, "rel-a")
196 if err != nil {
197 t.Fatal(err)
198 }
199 positions := ReleasePositionsFile{Dir: filepath.Join(data, "relocate", "release")}
200 for name, write := range map[string]func() error{
201 "attempts": func() error {
202 return relocationAttemptJournal{V: 1, Host: "a.example", Op: "rel-a", Opened: map[string]time.Time{}, Sessions: map[string]int{}, Starts: map[string]relocationStart{}}.write(data)
203 },
204 "probe fence": func() error { _, err := relocationProbeFenceOwner(fence, "a.example", 7); return err },
205 "positions": func() error {
206 return positions.Record(t.Context(), ReleasePosition{Request: StopRequest{Host: "a.example", OperationID: "rel-a"}, PID: 1, ProcessStart: "1"})
207 },
208 } {
209 t.Run(name, func(t *testing.T) {
210 relocationJournals.Lock()
211 done := make(chan error, 1)
212 go func() { done <- write() }()
213 select {
214 case err := <-done:
215 relocationJournals.Unlock()
216 t.Fatalf("wrote beside the sweep: %v", err)
217 case <-time.After(300 * time.Millisecond):
218 }
219 relocationJournals.Unlock()
220 if err := <-done; err != nil {
221 t.Fatal(err)
222 }
223 })
224 }
225}
226
227// Carry 4: the full runtime's config carries the agent's held-stage
228// evidence: the release key it was given and the edge's pointer digest
229// ratchet (pointerRatchet(false, ...), never root's), with the box's
230// secrets key when it has one. No release key, or a secrets key that does
231// not parse, refuses.
232func TestRelocationRuntimeConfigBindsTheAgentsHeldStageEvidence(t *testing.T) {
233 pub, _, err := ed25519.GenerateKey(rand.Reader)
234 if err != nil {
235 t.Fatal(err)
236 }
237 a := &ClaimAssembly{}
238 a.Guard.DataDir = t.TempDir()
239 opt := t.TempDir()
240 env := map[string]string{}
241 getenv := func(k string) string { return env[k] }
242 s := relocationStores{Files: release.NewDirStore(t.TempDir()), Versions: leasetest.NewVersionedBucket(), Bucket: "data-ie", Region: "ie", LeaseRegion: "eu-west-1"}
243 discharge := func(context.Context, hostclaim.Observed) error { return nil }
244 build := func(in relocationRuntimeInputs) (RelocationAssemblyConfig, error) {
245 return relocationRuntimeConfig(getenv, a, ClaimProviders{Selector: hostbox.WorkSource{Box: "dst", Region: "ie"}}, s, &registry.Registry{}, &relocationProbeActivator{}, hostactors.Directory{}, in, discharge)
246 }
247 cfg, err := build(relocationRuntimeInputs{Pub: pub, OptDir: opt})
248 if err != nil {
249 t.Fatal(err)
250 }
251 if !bytes.Equal(cfg.ReleasePub, pub) || cfg.Ratchet == nil || cfg.Ratchet != pointerRatchet(false, opt, a.Guard.DataDir, getenv) || cfg.Ratchet == pointerRatchet(true, opt, a.Guard.DataDir, getenv) || cfg.SecretsKey != nil {
252 t.Fatalf("held-stage evidence: pub %v ratchet %p secrets %v", cfg.ReleasePub, cfg.Ratchet, cfg.SecretsKey)
253 }
254 if _, err := build(relocationRuntimeInputs{OptDir: opt}); !errors.Is(err, hostwrite.ErrUnavailable) {
255 t.Fatalf("no release key: %v", err)
256 }
257 env["CARLOS_SECRETS_KEY"] = "not a key"
258 if _, err := build(relocationRuntimeInputs{Pub: pub, OptDir: opt}); err == nil {
259 t.Fatal("a secrets key that does not parse was accepted")
260 }
261}
262
263// With claim mode and region HA on and the relocation owners bound into
264// the wake's dispatcher, NewRelocationRuntime builds the full pass: this
265// box's worker in its roster region, which starts nothing (its Admission
266// and Census refuse), the participant, the J3 conversion and the hold
267// release. A roster entry that is not this box's enrollment, no release
268// key, or no data bucket refuses rather than build a pass over a guess.
269func TestRelocationRuntimeBuildsTheFullPassWithTheOwnersBound(t *testing.T) {
270 ctx := t.Context()
271 w := newLandingWorld(t, true, nil)
272 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
273 if err != nil {
274 t.Fatal(err)
275 }
276 defer reg.Close()
277 data := t.TempDir()
278 a := &ClaimAssembly{Claims: w.store, Guard: hostguard.Guard{DataDir: data}, RetirementAuthority: &hostops.ProductionAuthority{Store: w.store, Directory: w.directory},
279 Wake: &ClaimWakeAssembly{Inputs: ClaimOwnerDispatch{Relocation: &ClaimKindOwners{}}}}
280 local := &productionClaimLocal{legacyParticipation: true, assembly: a}
281 env := map[string]string{"CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1", "CARLOS_HIBERNATE_BUCKET": "data-ie"}
282 getenv := func(k string) string { return env[k] }
283 pub, _, err := ed25519.GenerateKey(rand.Reader)
284 if err != nil {
285 t.Fatal(err)
286 }
287 in := relocationRuntimeInputs{Enrollment: hostfloor.Enrollment{Box: "wrk", Enrollment: w.enrollment["wrk"]}, Pub: pub, OptDir: t.TempDir(),
288 Files: func(context.Context, string) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil },
289 Versioned: func(context.Context, string, string) (lease.VersionedStore, error) {
290 return leasetest.NewVersionedBucket(), nil
291 }}
292 providers := ClaimProviders{Selector: hostbox.WorkSource{Box: "wrk", Region: "ie"}}
293 build := func(in relocationRuntimeInputs) (*RelocationRuntime, error) {
294 return NewRelocationRuntime(ctx, getenv, w.store, local, providers, reg, &relocationProbeActivator{}, in)
295 }
296 r, err := build(in)
297 if err != nil {
298 t.Fatal(err)
299 }
300 worker, ok := r.Worker.(hostrelocate.Worker)
301 if !ok || r.Participant == nil || r.Convert == nil || r.Finish == nil || r.Box != "wrk" || r.DataDir != data || r.Hosts != nil {
302 t.Fatalf("not the full pass: %+v", r)
303 }
304 // Review M7: the held-stage evidence reaches the full pass's config.
305 if !bytes.Equal(r.config.ReleasePub, pub) || r.config.Ratchet == nil || r.config.Ratchet != pointerRatchet(false, in.OptDir, data, getenv) {
306 t.Fatalf("the full pass's held-stage evidence: pub %v ratchet %p", r.config.ReleasePub, r.config.Ratchet)
307 }
308 if worker.Self.Box != "wrk" || worker.Self.Enrollment != w.enrollment["wrk"] || worker.Region != "ie" {
309 t.Fatalf("the worker is not this box in its roster region: %+v %q", worker.Self, worker.Region)
310 }
311 if _, err := worker.Admission.Register(ctx, hostclaim.Observed{}, hostrelocate.Owner{}, ""); !errors.Is(err, errRelocationNotStarted) {
312 t.Fatalf("the pass's admission registered a start: %v", err)
313 }
314 if err := worker.Census(ctx, "wrk", nil); !errors.Is(err, errRelocationNotStarted) {
315 t.Fatalf("the pass's census passed a start: %v", err)
316 }
317 for name, change := range map[string]func(*relocationRuntimeInputs){
318 "another enrollment": func(in *relocationRuntimeInputs) { in.Enrollment.Enrollment = strings.Repeat("f", 32) },
319 "not in the roster": func(in *relocationRuntimeInputs) { in.Enrollment.Box = "gone" },
320 "no release key": func(in *relocationRuntimeInputs) { in.Pub = nil },
321 } {
322 bad := in
323 change(&bad)
324 if r, err := build(bad); err == nil || r != nil {
325 t.Fatalf("%s: built %+v", name, r)
326 }
327 }
328 // Region HA off: the completion-only runtime, though the owners are
329 // bound (ruling 82; the dormant constraint).
330 env["CARLOS_REGION_HA"] = ""
331 if r, err := build(in); err != nil || r == nil || r.Worker != nil || r.Participant != nil || r.Convert != nil || r.Finish == nil {
332 t.Fatalf("region HA off: %+v %v", r, err)
333 }
334 env["CARLOS_REGION_HA"] = "1"
335 delete(env, "CARLOS_HIBERNATE_BUCKET")
336 if r, err := build(in); !errors.Is(err, hostwrite.ErrUnavailable) || r != nil {
337 t.Fatalf("no data bucket: %+v %v", r, err)
338 }
339}
340