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 "encoding/json"
8 "errors"
9 "fmt"
10 "os"
11 "os/exec"
12 "path/filepath"
13 "sort"
14 "strings"
15 "testing"
16 "time"
17
18 "github.com/carlosframework/platform/internal/activator"
19 "github.com/carlosframework/platform/internal/appconfig"
20 "github.com/carlosframework/platform/internal/genselect"
21 "github.com/carlosframework/platform/internal/hostbox"
22 "github.com/carlosframework/platform/internal/hostclaim"
23 "github.com/carlosframework/platform/internal/hostguard"
24 "github.com/carlosframework/platform/internal/hostwrite"
25 "github.com/carlosframework/platform/internal/lease"
26 "github.com/carlosframework/platform/internal/moves"
27 "github.com/carlosframework/platform/internal/park"
28 "github.com/carlosframework/platform/internal/provision"
29 "github.com/carlosframework/platform/internal/registry"
30 "github.com/carlosframework/platform/internal/release"
31)
32
33// The target's staging park after the commit and the close (moves spec §7.9
34// steps 4 and 5; Plan 5 Task 3). The fixture is the source's finalization
35// harness (newFinalHarness: a landing driven to restored, the source box ie
36// with its real registry row and artifacts), given a probe so that the
37// TARGET box (uk) probes, commits and opens through Plan 3's committed arm,
38// not by hand. Both boxes run the real park engine over the fake versioned
39// buckets with one fake lease table. Nothing in production runs the
40// source region's acknowledgement yet (C4's regional runner), so
41// regionalAck stands in for it.
42
43func newCloseHarness(t *testing.T, opts ...p1Option) *finalHarness {
44 t.Helper()
45 h := newFinalHarness(t, opts...)
46 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
47 rec.Move.CutAttempts = 2
48 return nil
49 }); err != nil {
50 t.Fatal(err)
51 }
52 p := &probeLog{t: t, h: h.landingHarness}
53 h.probe, h.probeLive = p.wake, p.liveProbe
54 h.ownAuthorize = true
55 return h
56}
57
58// chore finds the op's park-replica in region: the staging lane when lane
59// is true, the location otherwise.
60func (h *finalHarness) chore(t *testing.T, region string, lane bool) *hostclaim.Obligation {
61 t.Helper()
62 for _, o := range h.claim(t).Obligations {
63 if o.Op == p1ID && o.Kind == "park-replica" && o.Region == region && o.Park != nil && isStagingLocation(o.Park.Location) == lane {
64 return &o
65 }
66 }
67 return nil
68}
69
70// regionalAck is C4's regional runner for the source region: the park
71// engine over data-ie, run once at the harness's clock.
72func (h *finalHarness) regionalAck(t *testing.T, o *hostclaim.Obligation) error {
73 t.Helper()
74 src, dst := park.PlacementRegion(h.src), park.PlacementRegion(h.dst)
75 e := park.Engine{Store: h.vs, Claims: h.s, Bucket: h.srcBucket, Region: src, Leases: h.leases, Now: h.clock.Now,
76 NewID: obligationID, View: sourceTestViewFor(src, dst), Authorize: sourceTestAuthorizeFor(h.srcBox, src)}
77 return e.Run(context.Background(), park.Work{Host: p1Host, ID: o.ID, Op: p1ID})
78}
79
80// clockTo moves the harness's clock to at, never back.
81func (h *finalHarness) clockTo(t *testing.T, at time.Time) {
82 t.Helper()
83 if d := at.Sub(h.clock.now); d > 0 {
84 h.clock.Advance(d)
85 }
86 if h.clock.now.After(at.Add(time.Second)) {
87 t.Fatalf("fixture: the clock is already at %v, past %v", h.clock.now, at)
88 }
89}
90
91// clockPast moves the harness's clock past at, if it is not there already.
92func (h *finalHarness) clockPast(at time.Time) {
93 if d := at.Sub(h.clock.now); d >= 0 {
94 h.clock.Advance(d + time.Second)
95 }
96}
97
98// committedAndOpen asserts the target committed and opened the landing.
99func (h *finalHarness) committedAndOpen(t *testing.T) {
100 t.Helper()
101 m := h.read(t, p1Host).Move
102 if m == nil || m.State != provision.MoveCommitted || m.OpenedAt == nil {
103 t.Fatalf("the target did not commit and open: %+v", m)
104 }
105 if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
106 t.Fatalf("the claim: %+v", c.Op)
107 }
108}
109
110// notClosed asserts the move is still open on both halves.
111func (h *finalHarness) notClosed(t *testing.T, why string) {
112 t.Helper()
113 if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveCommitted {
114 t.Fatalf("%s: the record closed: %+v", why, m)
115 }
116 if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
117 t.Fatalf("%s: the claim was released: %+v", why, c.Op)
118 }
119}
120
121// done asserts the close: the claim serving under the target with this
122// move recorded as released and nothing owed, the record at rest in uk with
123// LastMove done.
124func (h *finalHarness) done(t *testing.T) {
125 t.Helper()
126 rec := h.read(t, p1Host)
127 if rec.Move != nil || rec.Region != h.dst || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "done" {
128 t.Fatalf("the record after the close: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
129 }
130 c := h.claim(t)
131 if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != h.dstBox || hostclaim.RequireReleasedMove(p1ID)(c) != nil || len(c.Obligations) != 0 {
132 t.Fatalf("the claim after the close: %+v obligations %+v", c.Op, c.Obligations)
133 }
134}
135
136// TestStagingParkAcknowledgesAtSealBoundPlusH (Review Focus 1, witness
137// wStagingScopeDeadline; §7.9 step 4, K6): the commit queues the lane's
138// park scoped to the seal's bound T, and the target runs it through
139// laneParkTarget. At T + 16 min the run is ErrPending and nothing is
140// acknowledged; a restarted worker at T + 17 min + 1 s acknowledges, and
141// the scope it acknowledged is still T's (quiet_after = T + H).
142func TestStagingParkAcknowledgesAtSealBoundPlusH(t *testing.T) {
143 t.Parallel()
144 h := newCloseHarness(t)
145 h.land(t)
146 h.committedAndOpen(t)
147 seal, err := journalledSeal(h.claim(t))
148 if err != nil {
149 t.Fatal(err)
150 }
151 bound := seal.Bound()
152 o := h.chore(t, "uk", true)
153 if o == nil {
154 t.Fatal("the commit queued no staging park")
155 }
156 if want := moves.StagingScope(p1Host, p1ID, h.inc, bound); *o.Park != want || o.Owner != "region:uk" {
157 t.Fatalf("the staging park %+v %+v, want %+v", o, o.Park, want)
158 }
159 // The first pass archived the lane; nothing is acknowledged yet.
160 lane := moves.ScopeLocation(p1Host, p1ID) + "/"
161 for k := range h.f.buckets["data-uk"] {
162 if strings.HasPrefix(k, lane) {
163 t.Fatalf("the lane still holds %s after the target's pass", k)
164 }
165 }
166 // T + 16 min: the engine answers ErrPending, and the pass acknowledges
167 // nothing.
168 h.clockTo(t, bound.Add(16*time.Minute))
169 e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now,
170 NewID: obligationID, View: standInView("uk", roleTarget, h.read(t, p1Host).Move), Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
171 if err := e.Run(context.Background(), park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); !errors.Is(err, park.ErrPending) {
172 t.Fatalf("at T + 16 min the run answered %v, want ErrPending", err)
173 }
174 h.land(t)
175 if again := h.chore(t, "uk", true); again == nil || !again.Park.EvidenceAt.Equal(bound) {
176 t.Fatalf("at T + 16 min: %+v", again)
177 }
178 // A restarted target at T + 17 min + 1 s acknowledges.
179 h.freshTarget()
180 h.clockTo(t, bound.Add(park.SignatureAge+moves.ParkRequestLimit+time.Second))
181 h.land(t)
182 if again := h.chore(t, "uk", true); again != nil {
183 t.Fatalf("the staging park is still owed past T + H: %+v", again)
184 }
185 if q := h.claim(t).QuietAfter["uk"]; !q.Equal(bound.Add(park.SignatureAge + moves.ParkRequestLimit)) {
186 t.Fatalf("quiet_after %v, want T + H = %v", q, bound.Add(park.SignatureAge+moves.ParkRequestLimit))
187 }
188 if n := len(h.vs.callsOn("copy-version data-uk/parked/" + h.inc + "/" + p1Host + "/" + o.Retirement + "/moves/")); n == 0 {
189 t.Fatal("nothing was archived under the lane's retirement")
190 }
191}
192
193// TestCloseWaitsForBothObligations (§7.9 step 5): with the target open and
194// the source parked, the close still waits while the source region's
195// location park is owed; once both parks are acknowledged the claim is
196// serving under uk-box and the record at rest in uk.
197func TestCloseWaitsForBothObligations(t *testing.T) {
198 t.Parallel()
199 h := newCloseHarness(t)
200 h.land(t)
201 h.committedAndOpen(t)
202 h.sourcePass(t)
203 if m := h.read(t, p1Host).Move; m.SourceParkedAt == nil {
204 t.Fatalf("the source did not finalize: %+v", m)
205 }
206 h.land(t)
207 h.notClosed(t, "both parks owed")
208 // Past the staging lane's horizon the target acknowledges it; the
209 // source's location park is still owed (no regional runner ran it).
210 lane := h.chore(t, "uk", true)
211 h.clockPast(lane.Park.EvidenceAt.Add(park.SignatureAge + lane.Park.RequestLimit))
212 h.land(t)
213 if h.chore(t, "uk", true) != nil {
214 t.Fatal("the staging park was not acknowledged")
215 }
216 src := h.chore(t, "ie", false)
217 if src == nil {
218 t.Fatal("fixture: the source's location park was acknowledged by the finalization")
219 }
220 h.land(t)
221 h.notClosed(t, "the source's location park owed")
222 // The source region's runner acknowledges after its horizon; then the
223 // target's next pass closes.
224 h.clockPast(src.Park.EvidenceAt.Add(park.SignatureAge + src.Park.RequestLimit))
225 if err := h.regionalAck(t, src); err != nil {
226 t.Fatalf("the source region's acknowledgement: %v", err)
227 }
228 if h.chore(t, "ie", false) != nil {
229 t.Fatal("the source's location park was not acknowledged")
230 }
231 h.land(t)
232 h.done(t)
233 // A further pass of either box changes nothing.
234 before := h.claim(t).Gen
235 h.land(t)
236 h.sourcePass(t)
237 if after := h.claim(t).Gen; after != before {
238 t.Fatalf("a pass after the close wrote the claim (%d to %d)", before, after)
239 }
240}
241
242// TestCloseIsRecordFirst (§7.5, §19 K8 as Paul's ruling M4 amends them):
243// the close clears the record before it releases the claim. A crash between
244// leaves the record at rest with LastMove done and the claim
245// landing/committed with nothing owed; no new move can start (the claim is
246// not serving), and no box pass writes the claim again: the release is the
247// console's recovery pass (TestBoxClosesAreRecordFirst; the console's
248// TestRecordFirstCloseCrashIsFinishedByThePass). Without a crash the same
249// pass releases.
250func TestCloseIsRecordFirst(t *testing.T) {
251 t.Parallel()
252 h := newCloseHarness(t)
253 h.land(t)
254 h.sourcePass(t)
255 h.clock.Advance(24 * time.Hour)
256 if err := h.regionalAck(t, h.chore(t, "ie", false)); err != nil {
257 t.Fatal(err)
258 }
259 h.landAt = "after-close-record"
260 h.crashLand(t)
261 rec := h.read(t, p1Host)
262 if rec.Move != nil || rec.LastMove == nil || rec.LastMove.Outcome != "done" {
263 t.Fatalf("the record did not close first: %+v %+v", rec.Move, rec.LastMove)
264 }
265 c := h.claim(t)
266 if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != hostclaim.PhaseCommitted || c.Op.ID != p1ID {
267 t.Fatalf("the claim at the crash point: %+v", c.Op)
268 }
269 if err := hostclaim.StartGuard(c, []string{"uk"}, []string{"uk-box"}, ""); err == nil {
270 t.Fatal("a new operation could start over the unreleased claim")
271 }
272 gen := c.Gen
273 h.freshTarget()
274 h.land(t)
275 if after := h.claim(t).Gen; after != gen {
276 t.Fatalf("a box pass after the record's close wrote the claim (%d to %d)", gen, after)
277 }
278
279 // Without the crash, the pass that closes the record releases.
280 h2 := newCloseHarness(t)
281 h2.land(t)
282 h2.sourcePass(t)
283 h2.clock.Advance(24 * time.Hour)
284 if err := h2.regionalAck(t, h2.chore(t, "ie", false)); err != nil {
285 t.Fatal(err)
286 }
287 h2.land(t)
288 h2.done(t)
289}
290
291// Any participant that sees both stamps and nothing owed closes (§7.9 step
292// 5): the source's pass that stamps source_parked_at closes too when the
293// target has opened and both parks are already acknowledged.
294func TestTheSourceClosesWhenItFinishesLast(t *testing.T) {
295 t.Parallel()
296 h := newCloseHarness(t)
297 h.landFault = func(point string) error {
298 if point == "after-close-record" {
299 t.Fatal("the target closed before the source parked")
300 }
301 return nil
302 }
303 h.land(t)
304 h.committedAndOpen(t)
305 h.clock.Advance(24 * time.Hour)
306 h.land(t)
307 if h.chore(t, "uk", true) != nil {
308 t.Fatal("the staging park was not acknowledged")
309 }
310 h.notClosed(t, "the source has not parked")
311 // The source's location park is acknowledged in the finalization's own
312 // run, past its horizon; the same pass stamps and closes.
313 h.sourcePass(t)
314 h.done(t)
315}
316
317// K9: the console's commit (Task 7a) builds its scopes from the record and
318// the claim alone, through moves.CommitScopes with FlippedAt, so whichever
319// of the box and the console commits first writes the same obligations,
320// byte for byte, and the other's replay is no change.
321func TestConsoleCommitMatchesTheBoxs(t *testing.T) {
322 t.Parallel()
323 h, _ := probedLanding(t)
324 var pre *hostclaim.Claim
325 h.landFault = func(point string) error {
326 switch point {
327 case "before-commit":
328 pre = h.claim(t)
329 case "after-claim-committed":
330 return errCrash
331 }
332 return nil
333 }
334 h.land(t)
335 if pre == nil || pre.Op.Phase != hostclaim.PhaseLandingProbed {
336 t.Fatalf("fixture: the claim before the commit: %+v", pre)
337 }
338 box := h.claim(t)
339 if box.Op.Phase != hostclaim.PhaseCommitted {
340 t.Fatalf("the box did not commit the claim: %+v", box.Op)
341 }
342 rec := h.read(t, p1Host)
343 source, staging, err := moves.CommitScopes(pre, rec.Move.FlippedAt)
344 if err != nil {
345 t.Fatal(err)
346 }
347 var minted []string
348 for _, o := range box.Obligations[len(pre.Obligations):] {
349 minted = append(minted, o.ID, o.Retirement)
350 }
351 newID := func() string {
352 if len(minted) == 0 {
353 t.Fatal("the console's commit minted more ids than the box's")
354 }
355 id := minted[0]
356 minted = minted[1:]
357 return id
358 }
359 console := pre
360 if err := hostclaim.Commit(p1ID, rec.Move.From, pre.Op.Intent["source"], rec.Move.To, source, staging, newID)(console); err != nil {
361 t.Fatal(err)
362 }
363 got, _ := json.Marshal(console.Obligations)
364 want, _ := json.Marshal(box.Obligations)
365 if !bytes.Equal(got, want) || len(minted) != 0 {
366 t.Fatalf("the console's commit queues\n%s\nthe box's\n%s", got, want)
367 }
368 // Without the flip's time there is no source scope (H7).
369 if _, _, err := moves.CommitScopes(pre, nil); err == nil {
370 t.Fatal("scopes without the flip")
371 }
372}
373
374// Carried follow-up (d) (Plan 3 Task 8; Task 3 rulings): the finalization's
375// moves-owned rename runs inside the host guard with the claim and the
376// record read fresh there, so authority that changes before it renames
377// nothing at the live paths and journals no plan.
378func TestSourceFinalizationRenameIsGuarded(t *testing.T) {
379 t.Parallel()
380 for name, change := range map[string]func(t *testing.T, h *finalHarness){
381 "the claim's executor is this box again": func(t *testing.T, h *finalHarness) {
382 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Exec = "ie-box" })
383 },
384 "the claim has released the move": func(t *testing.T, h *finalHarness) {
385 rewriteClaimOp(t, h.s, func(op *hostclaim.Op) {
386 *op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "uk-box", Journal: map[string]string{}}
387 })
388 },
389 "the inventory claims a path": func(t *testing.T, h *finalHarness) {
390 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
391 if c.Inventory == nil {
392 c.Inventory = map[string]hostclaim.InventoryEntry{}
393 }
394 c.Inventory["ie-box"] = hostclaim.InventoryEntry{Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 9}}
395 })
396 },
397 "the record is no longer this commit": func(t *testing.T, h *finalHarness) {
398 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
399 rec.Move.ID = p1ID2
400 return nil
401 }); err != nil {
402 t.Fatal(err)
403 }
404 },
405 } {
406 t.Run(name, func(t *testing.T) {
407 h := newFinalHarness(t)
408 h.commit(t)
409 changed := false
410 inner := h.cfg.finalFault
411 h.cfg.finalFault = func(point string) error {
412 if point == "after-"+finalStepInstallation && !changed {
413 changed = true
414 change(t, h)
415 }
416 return inner(point)
417 }
418 h.sourcePass(t)
419 if !changed {
420 t.Fatalf("the rename's guard point was never reached: %v", h.finalHits)
421 }
422 want := []string{p1Host + ".db", p1Host + ".db.gen", p1Host + ".db.corrupt", p1Host, p1Host + ".env", p1Host + ".config.env", p1Host + "-sidecar.env", filepath.Join(".moves", p1Host, p1ID)}
423 if got := h.live(t); strings.Join(got, " ") != strings.Join(want, " ") {
424 t.Fatalf("live paths moved: %v, want %v", got, want)
425 }
426 if got := h.parked(t); len(got) != 0 {
427 t.Fatalf("parked: %v", got)
428 }
429 if j := h.sourceJournal(t); j.Done(finalStepLocal+"-plan") || j.Done(finalStepLocal) {
430 t.Fatalf("a plan was journalled: %v", j.Steps)
431 }
432 })
433 }
434}
435
436// The close harness's production probe mode (home part 2, Task 2). The
437// fake (probeLog.wake) admits with a literal box and region and so never
438// meets the production owners, their admission evidence or the C8 files
439// publication, which is where a home box's roster name meets the move's
440// @home. This mode wakes the target's probe through the production
441// movesProbeOwners over the claim runtime (claim Plan 9 Group D's
442// ClaimWakeAssembly, as TestMovesProbeOwnersThroughTheClaimRuntime drives
443// it), with the target box's identity from the roster: Box the target box,
444// Region its claim engine region (its roster name, ie for a home box),
445// HomeRegion and FleetHomeRegion the fleet's designation. Admit, Prepare,
446// the probe's open, the attempt's reservation and Materialize are the
447// claim's and the owners' own. What stands in is what main cannot run here:
448// the launch (every claim start is refused on main until the isolated root
449// launcher lands), the held works and render (the claim's move read access
450// over this box's registry), and the activator's live probe entry. A
451// Materialize that passes is the attempt passing.
452
453// p1Work is a held action's selected work in the production probe mode:
454// it runs fn, and records the identities it ran under.
455type p1Work struct {
456 order *[]string
457 name string
458}
459
460func (w p1Work) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
461 *w.order = append(*w.order, w.name+":"+id.Config+"|"+id.Release)
462 return fn()
463}
464
465// p1Markers is the claim's marker store over the harness's fake versioned
466// buckets: a conditional put on the current version's etag.
467type p1Markers struct{ vs *fakeVersioned }
468
469func (m p1Markers) PutMarker(ctx context.Context, bucket, key string, body []byte, expected string, check func(context.Context) error) (string, error) {
470 if err := check(ctx); err != nil {
471 return "", err
472 }
473 current := ""
474 if o, ok := m.vs.f.buckets[bucket][key]; ok {
475 current = o.ETag
476 }
477 if current != expected {
478 return "", hostbox.ErrChanged
479 }
480 ver := m.vs.f.nextVersion()
481 m.vs.f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
482 m.vs.f.setVersion(bucket, key, ver)
483 return ver, nil
484}
485
486// p1FilesStore is the target's data bucket as a release.Store for the C8
487// provider (movesFiles.Files): current reads with the version as the etag,
488// create-only and compare-and-swap writes.
489type p1FilesStore struct {
490 vs *fakeVersioned
491 bucket string
492}
493
494func (s p1FilesStore) Get(_ context.Context, key string) ([]byte, string, error) {
495 o, ok := s.vs.f.buckets[s.bucket][key]
496 if !ok {
497 return nil, "", os.ErrNotExist
498 }
499 return []byte(o.body), s.vs.versionID(s.bucket, key), nil
500}
501
502func (s p1FilesStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
503 if ifMatch == "" {
504 ver, err := s.vs.PutIfAbsent(ctx, s.bucket, key, body)
505 if errors.Is(err, lease.ErrExists) {
506 return "", release.ErrCASConflict
507 }
508 return ver, err
509 }
510 if _, ok := s.vs.f.buckets[s.bucket][key]; !ok || s.vs.versionID(s.bucket, key) != ifMatch {
511 return "", release.ErrCASConflict
512 }
513 ver := s.vs.f.nextVersion()
514 s.vs.f.store(s.bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: int64(len(body))}, string(body)})
515 s.vs.f.setVersion(s.bucket, key, ver)
516 return ver, nil
517}
518
519func (s p1FilesStore) List(_ context.Context, prefix string) ([]string, error) {
520 var out []string
521 for k := range s.vs.f.buckets[s.bucket] {
522 if strings.HasPrefix(k, prefix) {
523 out = append(out, k)
524 }
525 }
526 sort.Strings(out)
527 return out, nil
528}
529
530func (p1FilesStore) ListDirs(context.Context, string) ([]string, error) {
531 return nil, errors.New("test: not implemented")
532}
533
534func (p1FilesStore) Delete(context.Context, string) error {
535 return errors.New("test: not implemented")
536}
537
538// prodProbe is the production mode's state on one box: the claim runtime
539// (claim Plan 9 Group D's ClaimWakeAssembly, the recovery fixture's
540// controller) with the moves owners bound, on the box the side names.
541type prodProbe struct {
542 t *testing.T
543 h *phase1Harness
544 side prodSide
545 ctl *recoveryController
546 owners *movesProbeOwners
547 // order is the held works, render and release, in the order they ran.
548 order []string
549 // evidence counts the writer's admission evidence calls, and
550 // rejectEvidence, when set, refuses them before the owners are asked
551 // (the rejected-evidence witness).
552 evidence int
553 rejectEvidence error
554 // writerEvidence is the writer's AdmissionEvidence as bound.
555 writerEvidence func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error
556 // selected is the probe's launch inputs as the owners select them for
557 // the attempt, and attempt the reservation the runtime journalled for
558 // it: root's launch check compares the reservation's LaunchInputs with
559 // the hash of the launch it is asked to start.
560 selected ClaimLaunchInputs
561 attempt hostclaim.WakeRetryAttempt
562 // admissions are the admissions the runtime made, one per attempt, and
563 // admitted the claim each left, the last attempt's last; observed is
564 // the claim each start read before its admission.
565 admissions []hostclaim.AdmissionRecord
566 admitted hostclaim.Observed
567 observed []hostclaim.Observed
568 // stopped counts the predecessor writer sessions this box's own wakes
569 // stopped through the production Stop.
570 stopped int
571 // materialized is the claim and the launch inputs the runtime last
572 // materialized: the admitted inputs, which for a fresh generation are
573 // not the selected ones (each Select mints its own generation).
574 materialized hostclaim.Observed
575 materializedIn ClaimLaunchInputs
576 // afterMaterialize, when set, runs in the n-th wake (from 1) after
577 // Materialize and before the launch, while the attempt is live; an error
578 // fails the attempt there (the runtime's Failed), as an attempt that
579 // failed before readiness, or an agent that died there, leaves it (home
580 // part 2, Tasks 3 and 4).
581 afterMaterialize func(n int) error
582 // beforeAdmit, when set, sees each start's claim and the owners'
583 // selection before the runtime's admission.
584 beforeAdmit func(o hostclaim.Observed, in ClaimLaunchInputs)
585 wakes int
586 last activator.ProbeResult
587 live bool
588 seq int
589}
590
591// prodSide is the box a production mode runs on: its roster box, its
592// placement code (its composer's region), its data directory, data bucket
593// and the bucket's AWS region, its registry row, and the inputs its held
594// render renders (the move's pins for a landing, the recorded recovery
595// inputs for a recovery).
596type prodSide struct {
597 box, code, dataDir, bucket, aws string
598 route func() registry.Route
599 inputs func() *provision.MoveInputs
600}
601
602// productionProbe switches the harness's probe to the production mode: the
603// target's probe wakes through movesProbeOwners over the claim runtime, with
604// the target box's roster identity.
605func (h *landingHarness) productionProbe(t *testing.T) *prodProbe {
606 t.Helper()
607 p := newProdWake(t, h.phase1Harness, prodSide{box: h.dstBox, code: h.dst, dataDir: h.tdir, bucket: h.dstBucket, aws: h.dstAWS,
608 route: func() registry.Route { return h.route },
609 inputs: func() *provision.MoveInputs { return h.read(t, p1Host).Move.Inputs }})
610 h.probe, h.probeLive = p.wake, p.liveProbe
611 return p
612}
613
614// newProdWake builds the production mode on side's box. The owners are the
615// production movesProbeOwners (which dispatch a recovery op to their
616// recovery owners) with the real C8 provider, bound into the claim runtime
617// with the owners' admission evidence, the moves attempt journal and the
618// production artifact retirement (claimArtifactRetirement over park.Local,
619// as claimconstruction.go wires RetireArtifacts and RecoverArtifacts).
620// What stands in is what main cannot run here: the launch (every claim
621// start is refused on main until the isolated root launcher lands), its
622// readiness, the held works and render (the claim's move read access over
623// this box's registry), root's stop transport (the recovery fixture's
624// coordinator), the cross-box cold handover (AfterAdmission) and the
625// activator's live-probe entry. A Materialize that passes is the attempt
626// passing.
627func newProdWake(t *testing.T, h *phase1Harness, side prodSide) *prodProbe {
628 t.Helper()
629 if !h.realEnv {
630 // The probe's launch is the box's own composition; a source that
631 // pinned the h.env fixture would never match it.
632 t.Fatal("fixture: the production probe mode needs realLaunchEnv")
633 }
634 // Claim Plan 15: admissions through the real CAS record their origin
635 // under both switches; the real C8's proof reads that record.
636 t.Setenv("CARLOS_HOST_CLAIM", "1")
637 t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
638 actor := h.roster[side.box]
639 journal, err := hostbox.OpenLaunchJournal(filepath.Join(t.TempDir(), "launches.db"))
640 if err != nil {
641 t.Fatal(err)
642 }
643 t.Cleanup(func() { journal.Close() })
644 p := &prodProbe{t: t, h: h, side: side}
645 guard := hostguard.Guard{DataDir: side.dataDir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}}
646 f := &recoveryFixture{t: t, claims: h.s, evidence: release.NewDirStore(t.TempDir()), signer: ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize)),
647 guard: guard, journal: journal, kernel: &recoveryKernel{journal: journal}, now: h.clock.Now(), recordPrelaunch: true,
648 data: h.vs, bucket: side.bucket, box: actor.Box, enrollment: actor.Enrollment, region: actor.Region, leaseRegion: side.aws}
649 key, err := moves.ReportKeyFromSecrets(h.cfg.SecretsKey)
650 if err != nil {
651 t.Fatal(err)
652 }
653 filesDir := func(host string) (string, string) {
654 return filepath.Join(side.dataDir, host), filepath.Join(side.dataDir, host+".env")
655 }
656 owners := &movesProbeOwners{HomeRegion: actor.HomeRegion, FleetHomeRegion: actor.FleetHomeRegion, Claims: h.s, Records: MovesRecordsFrom(h.s),
657 ReadyPath: movesReadyPath(h.s), Data: h.vs, Bucket: side.bucket, Region: actor.Region, Box: actor.Box, DataDir: side.dataDir, FilesDir: filesDir,
658 Markers: p1Markers{h.vs},
659 Route: func(host string) (*registry.Route, error) { rt := side.route(); return &rt, nil },
660 // The held row's launch: what this box would launch, composed by
661 // the production composer over its data directory with its own
662 // region, as claimMoveLaunchSource composes it
663 // (ExecBackend.LaunchLayers, CARLOS_MOVE_PROBE dropped), the live
664 // release, and the env report made from that same composition, whose
665 // digest is the launch's move-portable pin.
666 Launch: func(_ context.Context, rt registry.Route) (ClaimServingLaunch, moves.EnvReport, error) {
667 composed, supplied, _, err := (&activator.ExecBackend{DataDir: side.dataDir, Region: side.code}).LaunchLayers(rt)
668 if err != nil {
669 return ClaimServingLaunch{}, moves.EnvReport{}, err
670 }
671 delete(composed, "CARLOS_MOVE_PROBE")
672 env := make([]string, 0, len(composed))
673 for k, v := range composed {
674 env = append(env, k+"="+v)
675 }
676 sort.Strings(env)
677 return ClaimServingLaunch{Executable: "/opt/carlos/live/" + p1Host, ExecutableSHA256: strings.Repeat("a", 64), Env: env, Version: "v7"}, moves.ReportLaunch(key, composed, supplied), nil
678 },
679 EnvKey: key,
680 Admitted: MovesRecordAdmittedAttempts,
681 Works: func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
682 p.order = append(p.order, "select:"+action)
683 return p1Work{order: &p.order, name: action}, nil
684 },
685 Render: func(ctx context.Context, _ registry.Route, _ string, work func(appconfig.HeldRender) (hostwrite.Work, error)) (appconfig.HeldRender, error) {
686 p.order = append(p.order, "render")
687 in := side.inputs()
688 hr := appconfig.HeldRender{BundleETags: in.BundleETags, OverlayETag: in.OverlayETag}
689 if _, err := work(hr); err != nil {
690 return appconfig.HeldRender{}, err
691 }
692 return hr, nil
693 },
694 Release: func(ctx context.Context, _ registry.Route, version string, work hostwrite.Work) error {
695 p.order = append(p.order, "release:"+version)
696 return work.With(ctx, hostwrite.Identity{Release: version}, func() error { return nil })
697 },
698 NewID: func() string { p.seq++; return fmt.Sprintf("fresh%d", p.seq) },
699 Now: h.clock.Now}
700 // The C8 provider is the real one over the box's bucket, as the claim
701 // assembly binds it, with Plan 15's proof over the claim store.
702 owners.Files = &movesFiles{Files: p1FilesStore{vs: h.vs, bucket: side.bucket}, Data: h.vs, Bucket: side.bucket, HomeRegion: actor.FleetHomeRegion, Region: actor.Region, Box: actor.Box,
703 DataDir: side.dataDir, EnvFile: envFileOf(filesDir), Proof: ClaimOriginProof{Claims: h.s, HomeRegion: actor.FleetHomeRegion, Region: actor.Region}}
704 if !owners.valid() {
705 t.Fatal("fixture: the production probe owners are not valid")
706 }
707 attempts := &movesProbeAttempts{Claims: h.s, Records: MovesRecordsFrom(h.s), DataDir: side.dataDir, NewID: owners.NewID, Now: h.clock.Now, Admitted: MovesRecordAdmittedAttempts}
708 bound := owners.MovesOwners(attempts)
709 dispatch := ClaimOwnerDispatch{Moves: &bound.ClaimKindOwners}
710 // The writer's admission evidence is the owner dispatch's, as the claim
711 // assembly binds it (claimassembly.go), bound before the runtime is
712 // built: Build captures the writer, so a later swap would leave the
713 // controller's allow-all evidence in force.
714 f.homeRegion, f.fleetHomeRegion = actor.HomeRegion, actor.FleetHomeRegion
715 f.admissionEvidence = func(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, plan *genselect.Plan) error {
716 p.evidence++
717 if p.rejectEvidence != nil {
718 return p.rejectEvidence
719 }
720 return dispatch.AdmissionEvidence(ctx, o, in, plan)
721 }
722 p.writerEvidence = f.admissionEvidence
723 c := f.controller("boot-1", "controller-1", 4101, "start-1")
724 c.wake.Inputs = p1RecordingInputs{ClaimOwnerDispatch: dispatch, p: p}
725 c.wake.Continue = hostclaim.SessionContinues
726 c.wake.Operations = bound.Operations
727 c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
728 // The production artifact retirement: a restore over a database of
729 // another generation retires that installation through park.Local's
730 // RetireSuperseded, which re-tags its inventory paths with the admitted
731 // installation's stamp.
732 retirement := claimArtifactRetirement{Local: park.Local{Guard: guard, Claims: h.s, DataDir: side.dataDir, Box: actor.Box}, Continue: hostclaim.SessionContinues}
733 c.wake.RetireArtifacts, c.wake.RecoverArtifacts = retirement.Retire, retirement.Recover
734 p.ctl, p.owners = c, owners
735 return p
736}
737
738// p1RecordingInputs is the owner dispatch as the runtime's inputs, every
739// method its own, with Materialize recording what the runtime admitted.
740type p1RecordingInputs struct {
741 ClaimOwnerDispatch
742 p *prodProbe
743}
744
745func (r p1RecordingInputs) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
746 r.p.materialized, r.p.materializedIn = o, in
747 return r.ClaimOwnerDispatch.Materialize(ctx, o, in, work)
748}
749
750// wake is one probe attempt through the claim runtime: the activator's
751// ProbeWake with its start policy, as far as the claim runs on main.
752func (p *prodProbe) wake(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) {
753 policy := activator.StartPolicy{MaxStarts: 5, Window: time.Duration(moves.ProbeSeconds * float64(time.Second)), Probe: true,
754 Deadline: cfg.Deadline, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
755 return p.start(ctx, p.side.route(), policy)
756}
757
758// recoveryWake is the activator's recovery probe entry
759// (ClaimRecoveryProbeWake) on this box: one claim wake under the
760// adapter's policy.
761func (p *prodProbe) recoveryWake(ctx context.Context, rt registry.Route, policy activator.StartPolicy) error {
762 _, err := p.start(ctx, rt, policy)
763 return err
764}
765
766// start is one claim wake's first start, in the activator's order
767// (claimwake.go): Prerequisites' stop of this box's own predecessor writer
768// session (the production Stop, through the fixture's root coordinator),
769// Admit, Prepare, Restore (the production cold restore of the selected
770// baseline, retiring a superseded installation), then the attempt's
771// Reserve and Materialize; a failure from then on is the attempt's
772// (Failed). The launch, its readiness and Healthy stand in: a Materialize
773// that passes is the attempt passing.
774func (p *prodProbe) start(ctx context.Context, rt registry.Route, policy activator.StartPolicy) (activator.ProbeResult, error) {
775 p.live = false
776 h := p.h
777 o, err := hostclaim.Read(ctx, h.s, p1Host)
778 if err != nil {
779 return activator.ProbeResult{}, err
780 }
781 if prior := o.Claim.Holder; prior != nil && prior.WriterBox == p.side.box && prior.Session != 0 {
782 if err := p.ctl.wake.Stop(ctx, o); err != nil {
783 return activator.ProbeResult{}, fmt.Errorf("stop the predecessor session %d: %w", prior.Session, err)
784 }
785 p.stopped++
786 if o, err = hostclaim.Read(ctx, h.s, p1Host); err != nil {
787 return activator.ProbeResult{}, err
788 }
789 }
790 live, cancel := context.WithCancel(context.Background())
791 p.t.Cleanup(cancel)
792 done := make(chan struct{})
793 go func() { <-live.Done(); close(done) }()
794 p.ctl.wake.mu.Lock()
795 p.ctl.wake.pending[p1Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
796 p.ctl.wake.mu.Unlock()
797 p.observed = append(p.observed, o)
798 if p.selected, err = p.owners.Select(ctx, rt, o, policy); err != nil {
799 return activator.ProbeResult{}, err
800 }
801 if p.beforeAdmit != nil {
802 p.beforeAdmit(o, p.selected)
803 }
804 admitted, err := p.ctl.runtime.Admit(ctx, rt, o, policy)
805 if err != nil {
806 return activator.ProbeResult{}, err
807 }
808 ah := admitted.Claim.Holder
809 p.admissions = append(p.admissions, ah.Admissions[len(ah.Admissions)-1])
810 p.admitted = admitted
811 p.wakes++
812 s, err := p.ctl.runtime.Prepare(ctx, rt, admitted, policy)
813 if err != nil {
814 return activator.ProbeResult{}, err
815 }
816 if err := s.Restore(ctx); err != nil {
817 return activator.ProbeResult{}, fmt.Errorf("restore: %w", err)
818 }
819 if err := s.Reserve(ctx, 0); err != nil {
820 return activator.ProbeResult{}, err
821 }
822 p.attempt = s.CurrentAttempt()
823 if err := s.Materialize(ctx); err != nil {
824 return activator.ProbeResult{}, errors.Join(err, s.Failed(ctx, 0))
825 }
826 if s.OpenProbe != nil {
827 // At the exec (the start's beforeStart hook).
828 if _, err := s.OpenProbe(ctx, h.clock.Now()); err != nil {
829 return activator.ProbeResult{}, err
830 }
831 }
832 if p.afterMaterialize != nil {
833 if err := p.afterMaterialize(p.wakes); err != nil {
834 return activator.ProbeResult{}, errors.Join(err, s.Failed(ctx, 0))
835 }
836 }
837 p.last = activator.ProbeResult{Session: ah.Session, Stamp: ah.Stamp, Generation: ah.Generation}
838 p.live = true
839 return p.last, nil
840}
841
842// liveProbe is the activator's LiveProbe over the mode's last passing call.
843func (p *prodProbe) liveProbe(host string) (string, activator.ProbeResult, bool) {
844 if host != p1Host || !p.live {
845 return "", activator.ProbeResult{}, false
846 }
847 return p1ID, p.last, true
848}
849