amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package host
4
5import (
6 "bytes"
7 "context"
8 "encoding/base64"
9 "encoding/json"
10 "errors"
11 "fmt"
12 "io"
13 "net/http"
14 "net/http/httptest"
15 "os"
16 "path/filepath"
17 "reflect"
18 "slices"
19 "sort"
20 "strings"
21 "sync"
22 "testing"
23 "time"
24
25 "github.com/carlosframework/platform/internal/activator"
26 "github.com/carlosframework/platform/internal/appconfig"
27 "github.com/carlosframework/platform/internal/console"
28 "github.com/carlosframework/platform/internal/domains"
29 "github.com/carlosframework/platform/internal/hostops"
30 "github.com/carlosframework/platform/internal/instancekey"
31 "github.com/carlosframework/platform/internal/layout"
32 "github.com/carlosframework/platform/internal/lease"
33 "github.com/carlosframework/platform/internal/lease/leasetest"
34 "github.com/carlosframework/platform/internal/moves"
35 "github.com/carlosframework/platform/internal/provision"
36 "github.com/carlosframework/platform/internal/registry"
37 "github.com/carlosframework/platform/internal/release"
38 "github.com/carlosframework/platform/internal/schedule"
39 "github.com/carlosframework/platform/internal/status"
40)
41
42// The offline operator move end to end (Task 8 of docs/superpowers/plans/
43// 2026-10-20-offline-operator-move.md): the console's real doors, in
44// process, driven in the operator's order (start, seal, flip, open, finish),
45// and two boxes in two regions, each ticked between the door calls in the
46// agent's order (reconcileSerializedWith, the edges-file write, movesPass).
47//
48// Production wiring throughout: console.NewServer over the deployment store
49// with an operator's CLI token; per box newMovesWorker armed by armOffline,
50// a real registry and data directory, a real activator.Service, the
51// production edges-file writer fed the identity edgesFileLoop builds from
52// the box's own environment, the production build-state carry
53// (moveCarry(newBuildStates)), a real litestream replica in data-ie written
54// through lease.ReplicateWithLease, leasetest's S3 server for data-ie and
55// data-uk, and the exec backend's real launch composition, restore and
56// lease take.
57//
58// The fakes, and why:
59// - the activator's wake backend (fakeWakeBackend), which only the fence
60// needs, as the plan allows; the real ExecBackend does every lease take
61// and the restore;
62// - the operator page (sealWorld.arm's Page records instead of writing
63// the outbox), so the test can say no page fired;
64// - the box status snapshots, written once before the move as each box's
65// status loop would (status.WriteBox), because the console's fleet read
66// joins them and no status loop runs in a test;
67// - the reconcile Layout intent (offBox.reconcile names flat or nested)
68// instead of mintLayoutIntentWith, as T3 to T5's worlds do;
69// - the source's config.env is rendered once by RenderHeld before the
70// seal, standing in for config sync's render at rest;
71// - the agent process environment: both boxes live in one test process,
72// and the launch composer (childEnv) reads os.Environ(), so each box's
73// LaunchLayers runs with that box's own agent variables swapped into
74// the process environment for the call (withAgentEnv). That is what a
75// separate carlos-edge process per box gives production.
76// - for the nested source only, T5's raw-directory stand-in for the bind
77// mount (the fixture's mount is a symlink the nested reader refuses).
78
79const e2eAddr = "ops@example.com"
80
81// The agent environments of the two boxes: the variables measured on dev
82// and flagship that differ from box to box (97f541f2). Each box's launch
83// composer sees its own.
84var (
85 e2eIEAgentEnv = map[string]string{
86 "SYSTEMD_EXEC_PID": "1101",
87 "MEMORY_PRESSURE_WATCH": "/sys/fs/cgroup/system.slice/carlos-edge.service/memory.pressure",
88 "INVOCATION_ID": "0f0e0d0c0b0a09080706050403020100",
89 "AWS_REGION": sealIERegion,
90 "STATUS_BOX_REGION": sealIERegion,
91 "PLATFORM_EDGE_IPS": "192.0.2.10",
92 }
93 e2eUKAgentEnv = map[string]string{
94 "SYSTEMD_EXEC_PID": "2202",
95 "MEMORY_PRESSURE_WATCH": "/sys/fs/cgroup/system.slice/carlos-edge.service/memory.pressure.uk",
96 "INVOCATION_ID": "00010203040506070809",
97 "AWS_REGION": sealUKRegion,
98 "STATUS_BOX_REGION": sealUKRegion,
99 "PLATFORM_EDGE_IPS": "198.51.100.20,198.51.100.21",
100 }
101)
102
103// envSwapMu serializes the agent-environment swap: the composer is called
104// from one goroutine at a time in this test, and the lock keeps it so.
105var envSwapMu sync.Mutex
106
107// withAgentEnv runs compose with env as the process's agent environment,
108// restored after the call.
109func withAgentEnv(env map[string]string, compose func(registry.Route) (map[string]string, map[string]bool, []string, error)) func(registry.Route) (map[string]string, map[string]bool, []string, error) {
110 return func(rt registry.Route) (map[string]string, map[string]bool, []string, error) {
111 envSwapMu.Lock()
112 defer envSwapMu.Unlock()
113 type prev struct {
114 v string
115 ok bool
116 }
117 old := map[string]prev{}
118 for k, v := range env {
119 pv, ok := os.LookupEnv(k)
120 old[k] = prev{pv, ok}
121 os.Setenv(k, v)
122 }
123 defer func() {
124 for k, p := range old {
125 if p.ok {
126 os.Setenv(k, p.v)
127 } else {
128 os.Unsetenv(k)
129 }
130 }
131 }()
132 return compose(rt)
133 }
134}
135
136// recordingStore is the deployment store every party uses (the console and
137// both boxes), logging each write in order.
138type recordingStore struct {
139 *release.DirStore
140 mu sync.Mutex
141 writes []string // "put <key>" or "delete <key>"
142}
143
144func (r *recordingStore) note(op, key string) {
145 r.mu.Lock()
146 r.writes = append(r.writes, op+" "+key)
147 r.mu.Unlock()
148}
149
150func (r *recordingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
151 r.note("put", key)
152 return r.DirStore.Put(ctx, key, body, ifMatch)
153}
154
155func (r *recordingStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
156 r.note("put", key)
157 return r.DirStore.PutFrom(ctx, key, body, size, ifMatch)
158}
159
160func (r *recordingStore) Delete(ctx context.Context, key string) error {
161 r.note("delete", key)
162 return r.DirStore.Delete(ctx, key)
163}
164
165func (r *recordingStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
166 r.note("delete", key)
167 return r.DirStore.DeleteIfMatch(ctx, key, etag)
168}
169
170func (r *recordingStore) mark() int {
171 r.mu.Lock()
172 defer r.mu.Unlock()
173 return len(r.writes)
174}
175
176func (r *recordingStore) since(m int) []string {
177 r.mu.Lock()
178 defer r.mu.Unlock()
179 return append([]string(nil), r.writes[m:]...)
180}
181
182// e2eView is the part of the console's status view the test drives by.
183type e2eView struct {
184 State string `json:"state"`
185 ID string `json:"id"`
186 Next string `json:"next"`
187 Waiting string `json:"waiting"`
188 Failed bool `json:"failed"`
189 Last *provision.MoveOutcome `json:"last"`
190}
191
192// e2eWorld is shop.example.com at home on ie-box (data-ie) with uk-box
193// (data-uk) in uk, the console's doors over the same store.
194type e2eWorld struct {
195 *landWorld
196 t *testing.T
197 rs *recordingStore
198 base string // the doors' base URL
199 token string
200 ieEB *activator.ExecBackend // ie-box's real lease take and restore
201
202 id string
203 recBefore map[string]any
204 storeMark int
205 bucketsAt map[string]map[string]bool // bucket -> key@version seen before the move
206 // The source's files tree and env/config bytes, before the move.
207 srcTree map[string]string
208 srcEnvRaw, srcCfgRaw []byte
209 srcFiles, srcEnv, cfg string
210 srcTimers map[string]registry.Timer
211 rows int
212}
213
214func newE2EWorld(t *testing.T, opts landOpts) *e2eWorld {
215 t.Helper()
216 ctx := context.Background()
217 t.Setenv("CARLOS_HOST_CLAIM", "")
218 // The test process's own values of the agent variables, so each box's
219 // swap is visibly its own and the process value is restored after.
220 for k := range e2eIEAgentEnv {
221 t.Setenv(k, "test-process")
222 }
223
224 // The data buckets over the S3 wire protocol, on loopback.
225 fake := leasetest.NewVersionedBucket()
226 fake.AddBucket(sealIE)
227 fake.AddBucket(sealUK)
228 h, _ := leasetest.NewS3Server(fake)
229 s3srv := httptest.NewServer(h)
230 t.Cleanup(s3srv.Close)
231 t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
232 t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
233 t.Setenv("AWS_SESSION_TOKEN", "")
234 t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
235 t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "absent"))
236 t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(t.TempDir(), "absent"))
237 t.Setenv("CARLOS_S3_ENDPOINT", s3srv.URL)
238
239 // One deployment store for everyone, recording every write.
240 rs := &recordingStore{DirStore: release.NewDirStore(t.TempDir())}
241
242 // The keys: the deployment's secrets key (the console's public half,
243 // each box's private half) and the release key.
244 keyB64, _, err := appconfig.GenerateKey()
245 if err != nil {
246 t.Fatal(err)
247 }
248 priv, err := appconfig.ParsePrivateKey(keyB64, "test key")
249 if err != nil {
250 t.Fatal(err)
251 }
252 seed, pubB64, err := release.GenerateKey()
253 if err != nil {
254 t.Fatal(err)
255 }
256 relPriv, err := release.ParseSigningKey(seed, "test")
257 if err != nil {
258 t.Fatal(err)
259 }
260 relPub, err := release.ParsePublicKey(pubB64, "test")
261 if err != nil {
262 t.Fatal(err)
263 }
264
265 // The console: an operator, its CLI token, the ops account owning app.
266 dir := t.TempDir()
267 cst, err := console.Open(filepath.Join(dir, "console.db"))
268 if err != nil {
269 t.Fatal(err)
270 }
271 t.Cleanup(func() { cst.Close() })
272 var srv *console.Server
273 ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { srv.ServeHTTP(w, r) }))
274 t.Cleanup(ts.Close)
275 if srv, err = console.NewServer(console.Config{Origin: ts.URL, Keymail: "http://keymail.invalid",
276 Registry: filepath.Join(dir, "registry.db"), Releases: rs, ReleasePub: relPub, Key: relPriv, AppScheme: "http",
277 HomeRegion: "ie", SecretsPub: base64.StdEncoding.EncodeToString(priv.PublicKey().Bytes())}, cst); err != nil {
278 t.Fatal(err)
279 }
280 now := time.Now()
281 if err := cst.AddMember(console.OpsAccountID, e2eAddr, console.RoleOwner, now); err != nil {
282 t.Fatal(err)
283 }
284 if err := cst.SetOperator(e2eAddr, true); err != nil {
285 t.Fatal(err)
286 }
287 if _, err := cst.CreateAppIn(console.OpsAccountID, "app", e2eAddr, now); err != nil {
288 t.Fatal(err)
289 }
290 userCode, deviceCode, err := cst.NewCLIAuth("ops@laptop", "127.0.0.1", now)
291 if err != nil {
292 t.Fatal(err)
293 }
294 if err := cst.ApproveCLIAuth(userCode, e2eAddr, now); err != nil {
295 t.Fatal(err)
296 }
297 _, token, _, err := cst.PollCLIAuth(deviceCode, now)
298 if err != nil || token == "" {
299 t.Fatalf("no CLI token: %v", err)
300 }
301
302 // The deployment: the ops account, app's manifest and the instance's
303 // record at home.
304 wd := &offWorld{t: t, s: rs, acct: release.OpsAccount, now: now}
305 if _, _, err := rs.Get(ctx, domains.AccountManifestKey(wd.acct)); errors.Is(err, os.ErrNotExist) {
306 manifest, _ := json.Marshal(map[string]any{"id": console.OpsAccountID, "sqid": wd.acct, "name": "ops", "created_at": now})
307 if _, err := rs.Put(ctx, domains.AccountManifestKey(wd.acct), manifest, ""); err != nil {
308 t.Fatal(err)
309 }
310 }
311 if err := provision.PutManifest(ctx, rs, wd.acct, "app", provision.Manifest{Schema: 1, App: "app", Domain: "example.com"}); err != nil {
312 t.Fatal(err)
313 }
314 wd.create(offHost)
315
316 // The boxes. ie-box is the home's one box; uk-box is uk's.
317 sw := &sealWorld{offWorld: wd, fake: fake, keyB64: keyB64, priv: priv}
318 sw.ie = sw.sourceBox("ie-box")
319 sw.ie.env["CARLOS_HIBERNATE_REGION"] = sealIERegion
320 sw.uk = sw.box("uk-box", "uk", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealUK, "CARLOS_SECRETS_KEY": keyB64,
321 "CARLOS_HIBERNATE_REGION": sealUKRegion, "CARLOS_RELEASE_PUBKEY": pubB64})
322 lw := &landWorld{t: t, sw: sw, wd: wd, s3: &landS3{f: fake}, ie: sw.ie, uk: sw.uk, opts: opts, relPriv: relPriv}
323 shipVersion(t, rs, wd.acct, "app", sealVersion)
324 shipVersion(t, rs, wd.acct, "app", "v2")
325 lw.promote(sealVersion)
326 lw.bundle(landStore)
327 lw.ukEB = &activator.ExecBackend{DataDir: lw.uk.dd, OptDir: lw.uk.opt, Region: "uk"}
328 lw.arm(lw.uk)
329 w := &e2eWorld{landWorld: lw, t: t, rs: rs, base: ts.URL + "/api/cli/apps/" + wd.acct + "/app/instances/" + offHost + "/move/offline", token: token}
330
331 // The source box's production build-state carry, and each box's launch
332 // composer under its own agent environment.
333 ie := lw.ie
334 bs, bsErr := newBuildStates(&activator.Config{Bucket: sealIE, Region: sealIERegion, Prefix: sealPrefix}, ie.reg, ie.opt)
335 ie.w.cfg.CarryRequest = moveCarry(bs, bsErr)
336 ie.w.cfg.LaunchLayers = withAgentEnv(e2eIEAgentEnv, ie.w.cfg.LaunchLayers)
337 lw.uk.w.cfg.LaunchLayers = withAgentEnv(e2eUKAgentEnv, lw.ukEB.LaunchLayers)
338 w.ieEB = &activator.ExecBackend{Cfg: activator.Config{Bucket: sealIE, Region: sealIERegion, Prefix: sealPrefix}, DataDir: ie.dd, OptDir: ie.opt}
339
340 // The source at rest, as a box that has served it: rows replicated in
341 // two sessions, then a tail the stop never checkpointed.
342 sw.instance(ie) // rows 0..19, replicated
343 db := sw.dbPath(ie)
344 sw.writeRows(ie, db, 20, 25, true)
345 sw.writeRows(ie, db, 25, 30, false)
346 w.rows = 30
347 w.srcFiles, w.srcEnv, w.cfg = filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env"), filepath.Join(ie.dd, offHost+".config.env")
348 if opts.nested {
349 raw := sw.nest(ie)
350 w.srcFiles, w.srcEnv, w.cfg = filepath.Join(raw, offHost), filepath.Join(raw, offHost+".env"), filepath.Join(raw, offHost+".config.env")
351 for _, name := range []string{"launch.run.json", "launch.seed.json", "launch.gen.7.json"} {
352 writeFile(t, filepath.Join(raw, name), "{}")
353 }
354 // T5's stand-in for the bind mount: the same two files, composed by
355 // the same code, from the raw instance directory.
356 ie.w.cfg.LaunchLayers = withAgentEnv(e2eIEAgentEnv, func(rt registry.Route) (map[string]string, map[string]bool, []string, error) {
357 rt.DBPath, rt.Socket = provision.DefaultDBPath(raw, rt.Host), filepath.Join(provision.SocketDir, rt.Host+".sock")
358 return (&activator.ExecBackend{DataDir: raw}).LaunchLayers(rt)
359 })
360 }
361 if opts.noFiles {
362 if err := os.RemoveAll(w.srcFiles); err != nil {
363 t.Fatal(err)
364 }
365 }
366 lw.srcFiles, lw.srcEnv, lw.srcConfig = w.srcFiles, w.srcEnv, w.cfg
367
368 // Both boxes published as their loops publish them, and their status
369 // snapshots.
370 w.tickAll()
371 for _, b := range []*offBox{ie, lw.uk} {
372 if err := status.WriteBox(ctx, rs, status.Box{Label: b.name, GeneratedAt: time.Now(), Metrics: &status.BoxMetrics{}}); err != nil {
373 t.Fatal(err)
374 }
375 }
376 if !ie.lists(offHost) || lw.uk.lists(offHost) {
377 t.Fatal("fixture: ie-box must serve the host and uk-box must not")
378 }
379 return w
380}
381
382// publish is one pass of edgesFileLoop on b: the identity it builds at arm
383// from the box's own environment, the capability from the box's armed
384// worker, through the production writer.
385func (w *e2eWorld) publish(b *offBox) {
386 w.t.Helper()
387 sharedPool, inbound := edgesFileEligibility(b.getenv)
388 aws := sealIERegion
389 if b.region == "uk" {
390 aws = sealUKRegion
391 }
392 ident := edgesFile{
393 Box: b.name, BoxID: b.name, Region: aws, Placement: b.region,
394 DataBucket: strings.TrimSpace(b.getenv("CARLOS_HIBERNATE_BUCKET")),
395 DataBucketRegion: strings.TrimSpace(b.getenv("CARLOS_HIBERNATE_REGION")),
396 HibernationConfigured: strings.TrimSpace(b.getenv("CARLOS_HIBERNATE_BUCKET")) != "",
397 ServesSharedPool: sharedPool,
398 AcceptsInbound: inbound,
399 SecretsKeyFP: edgesFileSecretsFP(b.getenv),
400 Capabilities: withOfflineCapability(edgesFileCapabilities(b.getenv), b.w.cfg.OfflineCapable && b.w.cfg.Offline != nil),
401 }
402 if err := edgesFileWrite(context.Background(), w.wd.s, b.reg, ident, "192.0.2.1", &b.last, false); err != nil {
403 w.t.Fatal(err)
404 }
405}
406
407// tick is the agent's order on one box: reconcile, the edges file, the
408// moves pass. The boxes' clock is the wall clock, which the doors judge
409// report freshness by.
410func (w *e2eWorld) tick(b *offBox) {
411 w.t.Helper()
412 w.tickAt(b, time.Now())
413}
414
415// tickAt is tick with the boxes' clock reading at.
416func (w *e2eWorld) tickAt(b *offBox, at time.Time) {
417 w.t.Helper()
418 w.setBoxClock(at)
419 b.reconcile()
420 w.publish(b)
421 b.pass()
422}
423
424func (w *e2eWorld) setBoxClock(at time.Time) {
425 w.wd.mu.Lock()
426 w.wd.now = at
427 w.wd.mu.Unlock()
428}
429
430func (w *e2eWorld) tickAll() {
431 w.t.Helper()
432 w.tick(w.ie)
433 w.tick(w.uk)
434}
435
436// door posts to one of the doors (verb "" is start).
437func (w *e2eWorld) door(verb, body string) (int, string) {
438 w.t.Helper()
439 u := w.base
440 if verb != "" {
441 u += "/" + verb
442 }
443 return w.do(http.MethodPost, u, body)
444}
445
446func (w *e2eWorld) do(method, u, body string) (int, string) {
447 w.t.Helper()
448 req, err := http.NewRequest(method, u, strings.NewReader(body))
449 if err != nil {
450 w.t.Fatal(err)
451 }
452 req.Header.Set("Authorization", "Bearer "+w.token)
453 req.Header.Set("Content-Type", "application/json")
454 res, err := http.DefaultClient.Do(req)
455 if err != nil {
456 w.t.Fatal(err)
457 }
458 defer res.Body.Close()
459 raw, _ := io.ReadAll(res.Body)
460 return res.StatusCode, string(raw)
461}
462
463func (w *e2eWorld) decode(out string) e2eView {
464 w.t.Helper()
465 var v e2eView
466 if err := json.Unmarshal([]byte(out), &v); err != nil {
467 w.t.Fatalf("%v in %s", err, out)
468 }
469 return v
470}
471
472// accept posts a verb and wants the door's 202 with the state.
473func (w *e2eWorld) accept(verb, body, wantState string) e2eView {
474 w.t.Helper()
475 code, out := w.door(verb, body)
476 if code != http.StatusAccepted {
477 w.t.Fatalf("%s = %d %s", verbName(verb), code, out)
478 }
479 v := w.decode(out)
480 if v.State != wantState {
481 w.t.Fatalf("%s: state %q, want %q (%s)", verbName(verb), v.State, wantState, out)
482 }
483 return v
484}
485
486func verbName(v string) string {
487 if v == "" {
488 return "start"
489 }
490 return v
491}
492
493func (w *e2eWorld) status() e2eView {
494 w.t.Helper()
495 code, out := w.do(http.MethodGet, w.base, "")
496 if code != http.StatusOK {
497 w.t.Fatalf("status = %d %s", code, out)
498 }
499 return w.decode(out)
500}
501
502// until ticks both boxes, then reads the status view, until ok says the
503// step has reported.
504func (w *e2eWorld) until(what string, ok func(e2eView) bool) e2eView {
505 w.t.Helper()
506 var v e2eView
507 for range 6 {
508 w.tickAll()
509 if v = w.status(); ok(v) {
510 return v
511 }
512 }
513 w.t.Fatalf("%s: never reported; the view is %+v", what, v)
514 return v
515}
516
517func nextIs(verb string) func(e2eView) bool {
518 return func(v e2eView) bool { return v.Next == verb }
519}
520
521func (w *e2eWorld) recordRaw() []byte {
522 w.t.Helper()
523 raw, _, err := w.rs.Get(context.Background(), provision.RecordKey(w.wd.acct, "app", offHost))
524 if err != nil {
525 w.t.Fatal(err)
526 }
527 return raw
528}
529
530func asJSONMap(t *testing.T, raw []byte) map[string]any {
531 t.Helper()
532 out := map[string]any{}
533 if err := json.Unmarshal(raw, &out); err != nil {
534 t.Fatal(err)
535 }
536 return out
537}
538
539// requireRecordDiff: the record differs from the one read before start in
540// region, move, last_move and record_seq only (move.updated_at lives in
541// move).
542func (w *e2eWorld) requireRecordDiff(when string) map[string]any {
543 w.t.Helper()
544 after := asJSONMap(w.t, w.recordRaw())
545 allowed := map[string]bool{"region": true, "move": true, "last_move": true, "record_seq": true}
546 keys := map[string]bool{}
547 for k := range w.recBefore {
548 keys[k] = true
549 }
550 for k := range after {
551 keys[k] = true
552 }
553 for k := range keys {
554 if !allowed[k] && !reflect.DeepEqual(w.recBefore[k], after[k]) {
555 w.t.Errorf("%s: record field %q changed: %v -> %v", when, k, w.recBefore[k], after[k])
556 }
557 }
558 return after
559}
560
561// bucketVersions is every key@version in bucket.
562func (w *e2eWorld) bucketVersions(bucket string) map[string]bool {
563 w.t.Helper()
564 vers, err := w.sw.fake.ListVersions(context.Background(), bucket, "")
565 if err != nil {
566 w.t.Fatal(err)
567 }
568 out := map[string]bool{}
569 for _, v := range vers {
570 out[v.Key+"@"+v.VersionID] = true
571 }
572 return out
573}
574
575// newBucketKeys is the keys written or deleted in bucket since snap.
576func (w *e2eWorld) newBucketKeys(bucket string, snap map[string]bool) []string {
577 w.t.Helper()
578 var out []string
579 seen := map[string]bool{}
580 for kv := range w.bucketVersions(bucket) {
581 if snap[kv] {
582 continue
583 }
584 k := kv[:strings.LastIndex(kv, "@")]
585 if !seen[k] {
586 seen[k] = true
587 out = append(out, k)
588 }
589 }
590 sort.Strings(out)
591 return out
592}
593
594func (w *e2eWorld) snapBuckets() {
595 w.bucketsAt = map[string]map[string]bool{sealIE: w.bucketVersions(sealIE), sealUK: w.bucketVersions(sealUK)}
596}
597
598// requireClosedWriteSet: since start, the deployment store saw writes only
599// in the plan's closed write set, and the data buckets only under the
600// move's own data keys. control/hosts/, control/host-census/,
601// control/host-actors.json, the app's admission and the domains claim are
602// named, so a write there reads as what it is.
603func (w *e2eWorld) requireClosedWriteSet(when string) {
604 t := w.t
605 t.Helper()
606 recKey := provision.RecordKey(w.wd.acct, "app", offHost)
607 admission, err := hostops.AdmissionKey(w.wd.acct, "app")
608 if err != nil {
609 t.Fatal(err)
610 }
611 forbidden := []string{"control/hosts/", "control/host-census/", "control/host-actors.json", admission, domains.ClaimsPrefix, domains.ClaimKey(offHost)}
612 allowed := func(key string) bool {
613 switch {
614 case key == recKey, key == release.SerialKey,
615 strings.HasPrefix(key, "control/offline-moves/"),
616 strings.HasPrefix(key, "control/config-ack/"),
617 strings.HasPrefix(key, edgesFilePrefix):
618 return true
619 }
620 return false
621 }
622 ops := w.rs.since(w.storeMark)
623 if !slices.ContainsFunc(ops, func(op string) bool { return strings.HasPrefix(op, "put control/offline-moves/") }) {
624 t.Fatalf("%s: the recording saw no report written: %v", when, ops)
625 }
626 for _, op := range ops {
627 key := op[strings.IndexByte(op, ' ')+1:]
628 for _, f := range forbidden {
629 if strings.HasPrefix(key, f) {
630 t.Errorf("%s: the move wrote the claim's key %s (%s)", when, key, op)
631 }
632 }
633 if !allowed(key) {
634 t.Errorf("%s: a write outside the closed write set: %s", when, op)
635 }
636 }
637 loc := activator.HibernatePrefix(sealPrefix, offHost) + "/"
638 for _, k := range w.newBucketKeys(sealIE, w.bucketsAt[sealIE]) {
639 if !strings.HasPrefix(k, loc) && k != "move-receipts/"+offHost+".json" {
640 t.Errorf("%s: data-ie written outside the move's keys: %s", when, k)
641 }
642 }
643 for _, k := range w.newBucketKeys(sealUK, w.bucketsAt[sealUK]) {
644 if !strings.HasPrefix(k, activator.MovesPrefix(offHost)) && !strings.HasPrefix(k, moves.OfflineArchivePrefix(offHost, w.id)) {
645 t.Errorf("%s: data-uk written outside the move's keys: %s", when, k)
646 }
647 }
648}
649
650// begin captures the state the move is judged against, then starts it
651// through the door and ticks until the seal door would accept.
652func (w *e2eWorld) begin() {
653 t := w.t
654 t.Helper()
655 w.recBefore = asJSONMap(t, w.recordRaw())
656 timers, err := w.ie.reg.TimersFor(offHost)
657 if err != nil {
658 t.Fatal(err)
659 }
660 w.srcTimers = map[string]registry.Timer{}
661 for _, tm := range timers {
662 w.srcTimers[tm.Name] = tm
663 }
664 w.storeMark = w.rs.mark()
665 w.snapBuckets()
666
667 v := w.accept("", `{"region":"uk"}`, provision.OfflineStopping)
668 w.id = v.ID
669 if !strings.HasPrefix(w.id, "mv_") {
670 t.Fatalf("move id %q", w.id)
671 }
672 w.until("the source's stop", nextIs("seal"))
673 w.requireRecordDiff("after start")
674 if w.ie.lists(offHost) {
675 t.Fatal("ie-box still lists the host after its stop")
676 }
677
678 w.renderSourceConfig()
679}
680
681// renderSourceConfig stands in for config sync's render at rest: the
682// source's config.env from the bundle (store binding and secret), and the
683// source's bytes the move is judged against.
684func (w *e2eWorld) renderSourceConfig() {
685 t := w.t
686 t.Helper()
687 hr, err := w.ie.w.cfg.RenderHeld(context.Background(), w.wd.s, *w.ie.row(offHost), w.id, nil, func(string) error { return nil })
688 if err != nil {
689 t.Fatal(err)
690 }
691 if hr.Path != w.cfg {
692 // The nested fixture's sockets are not production's, so the render
693 // took the flat path (T5's note): production renders into the
694 // instance directory.
695 if err := os.Rename(hr.Path, w.cfg); err != nil {
696 t.Fatal(err)
697 }
698 }
699 if isDir(w.srcFiles) {
700 w.srcTree = tree(t, w.srcFiles)
701 }
702 w.srcEnvRaw, w.srcCfgRaw = mustRead(t, w.srcEnv), mustRead(t, w.cfg)
703 if !bytes.Contains(w.srcCfgRaw, []byte("STORE_BUCKET")) {
704 t.Fatalf("fixture: the source's config carries no store binding: %q", w.srcCfgRaw)
705 }
706}
707
708func mustRead(t *testing.T, p string) []byte {
709 t.Helper()
710 b, err := os.ReadFile(p)
711 if err != nil {
712 t.Fatal(err)
713 }
714 return b
715}
716
717// sealAndFlip drives seal, the sealer's passes and flip, and asserts the
718// stop between them: the tombstone holds data-ie against every wake.
719func (w *e2eWorld) seal() {
720 t := w.t
721 t.Helper()
722 w.accept("seal", "", provision.OfflineSealing)
723 w.until("the seal", nextIs("flip"))
724 if st := w.wd.record().Move.State; st != provision.OfflineSealed {
725 t.Fatalf("state %s after the seal", st)
726 }
727 w.requireStopped("after the seal")
728}
729
730// requireStopped: a wake's lease take on data-ie meets this move's
731// tombstone, through lease.Acquire and through ie-box's ExecBackend (the one
732// lease take for every wake source), and the row stays held. Until open,
733// the activator also refuses a request for the held row at its fence (the
734// source lifts the fence once it has parked, at open).
735func (w *e2eWorld) requireStopped(when string) {
736 t := w.t
737 t.Helper()
738 rt := w.requireTombstoned(when)
739 if !w.ie.svc.Fenced(offHost) {
740 t.Fatalf("%s: ie-box's activator is not fenced", when)
741 }
742 proxied := false
743 rec := httptest.NewRecorder()
744 w.ie.svc.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+offHost+"/", nil), rt,
745 http.HandlerFunc(func(http.ResponseWriter, *http.Request) { proxied = true }))
746 if proxied || rec.Code < 500 {
747 t.Fatalf("%s: a request on ie-box was answered %d (proxied %v)", when, rec.Code, proxied)
748 }
749}
750
751// requireTombstoned is requireStopped's lease half, and the held row.
752func (w *e2eWorld) requireTombstoned(when string) registry.Route {
753 t := w.t
754 t.Helper()
755 ctx := context.Background()
756 _, err := acquireAsAWake(ctx)
757 var he *lease.HeldError
758 if !errors.As(err, &he) || he.Owner != moves.TombstoneOwner(w.id) {
759 t.Fatalf("%s: lease.Acquire on data-ie = %v, want a HeldError naming %s", when, err, moves.TombstoneOwner(w.id))
760 }
761 rt := w.ie.row(offHost)
762 if rt == nil || rt.HoldOp != w.id || rt.Status == "ready" {
763 t.Fatalf("%s: ie-box's row %+v is not held by the move", when, rt)
764 }
765 if l, err := w.ieEB.Acquire(ctx, *rt); err == nil || !strings.Contains(err.Error(), moves.TombstoneOwner(w.id)) {
766 if l != nil {
767 l.Release(ctx)
768 }
769 t.Fatalf("%s: ie-box's exec backend took the lease: %v", when, err)
770 }
771 return *rt
772}
773
774func (w *e2eWorld) flipAndLand() {
775 w.t.Helper()
776 w.accept("flip", "", provision.OfflineLanding)
777 w.requireRecordDiff("after flip")
778 w.until("the landing", nextIs("open"))
779}
780
781// open drives open and ticks until the console would take finish: the
782// target opened and every source box parked.
783func (w *e2eWorld) open() {
784 w.t.Helper()
785 w.accept("open", "", provision.OfflineOpen)
786 w.until("the open", nextIs("finish"))
787 w.requireRecordDiff("after open")
788}
789
790// finish drives both finish calls and the passes between.
791func (w *e2eWorld) finish(wantOutcome string) {
792 t := w.t
793 t.Helper()
794 code, out := w.door("finish", "")
795 if code != http.StatusAccepted {
796 t.Fatalf("finish = %d %s", code, out)
797 }
798 if v := w.decode(out); v.State != "none" {
799 if v.State != provision.OfflineFinishing {
800 t.Fatalf("finish: state %q", v.State)
801 }
802 w.until("the finishing", nextIs("finish"))
803 w.accept("finish", "", "none")
804 }
805 v := w.status()
806 if v.State != "none" || v.Last == nil || v.Last.ID != w.id || v.Last.Outcome != wantOutcome {
807 t.Fatalf("after finish: %+v (last %+v)", v, v.Last)
808 }
809}
810
811// requireTombstoneLifted: data-ie's lock is absent or a lifted tombstone.
812func (w *e2eWorld) requireTombstoneLifted() {
813 w.t.Helper()
814 _, owner, expires, err := lease.LockState(context.Background(), sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
815 if err != nil || !(owner == "" || lease.IsLiftedTombstone(owner, expires)) {
816 w.t.Fatalf("data-ie's lock is %q until %v (%v), want it lifted", owner, expires, err)
817 }
818}
819
820// requireTarget: what uk-box holds after open, against the source as it
821// was before the move.
822func (w *e2eWorld) requireTarget() {
823 t := w.t
824 t.Helper()
825 uk := w.uk
826 if w.opts.noFiles {
827 if got := tree(t, filepath.Join(uk.dd, offHost)); len(got) != 1 {
828 t.Fatalf("a source with no files directory landed %v", got)
829 }
830 } else if got := tree(t, filepath.Join(uk.dd, offHost)); !reflect.DeepEqual(got, w.srcTree) {
831 t.Fatalf("files differ:\nsource %v\ntarget %v", w.srcTree, got)
832 }
833 if got := mustRead(t, filepath.Join(uk.dd, offHost+".env")); !bytes.Equal(got, w.srcEnvRaw) {
834 t.Fatalf("<host>.env differs:\nsource %q\ntarget %q", w.srcEnvRaw, got)
835 }
836 ukCfg := mustRead(t, filepath.Join(uk.dd, offHost+".config.env"))
837 if !bytes.Equal(ukCfg, w.srcCfgRaw) {
838 t.Fatalf("<host>.config.env differs:\nsource %q\ntarget %q", w.srcCfgRaw, ukCfg)
839 }
840 // The instance key the app is handed: byte for byte the source's.
841 srcKey, _ := instancekey.Effective(instancekey.ParseEnv(w.srcCfgRaw), instancekey.ParseEnv(w.srcEnvRaw))
842 dstKey, _ := instancekey.Effective(instancekey.ParseEnv(ukCfg), instancekey.ParseEnv(mustRead(t, filepath.Join(uk.dd, offHost+".env"))))
843 if srcKey == "" || dstKey != srcKey {
844 t.Fatalf("instance key: source %q, target %q", srcKey, dstKey)
845 }
846 rt := uk.row(offHost)
847 if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
848 t.Fatalf("uk-box's row after open: %+v", rt)
849 }
850 // The launch the target composes, under its own agent environment: the
851 // store binding, its secret, the key; and the agent's own variables
852 // are uk-box's, not ie-box's (the pin matched across them).
853 env, _, _, err := uk.w.cfg.LaunchLayers(*rt)
854 if err != nil {
855 t.Fatal(err)
856 }
857 for k, want := range map[string]string{"STORE_BUCKET": landStore, "STORE_SECRET_KEY": landSecret, "CARLOS_INSTANCE_KEY": srcKey,
858 "SYSTEMD_EXEC_PID": e2eUKAgentEnv["SYSTEMD_EXEC_PID"], "PLATFORM_EDGE_IPS": e2eUKAgentEnv["PLATFORM_EDGE_IPS"]} {
859 if env[k] != want {
860 t.Fatalf("the target launches with %s=%q, want %q", k, env[k], want)
861 }
862 }
863 // Timers, NextDue and replication arrived.
864 rows, err := uk.reg.TimersFor(offHost)
865 if err != nil || len(rows) != 2 || moves.TimersDigest(rows) != w.sealDigest() {
866 t.Fatalf("timers on the target %+v %v", rows, err)
867 }
868 for _, tm := range rows {
869 want, ok := w.srcTimers[tm.Name]
870 if !ok || tm.Kind != want.Kind || tm.Spec != want.Spec || tm.Path != want.Path || tm.NextAt != want.NextAt {
871 t.Fatalf("timer %+v on the target, the source had %+v", tm, want)
872 }
873 }
874 if rt.NextDue != 1_800_000_000 || rt.Replication != registry.ReplicationBatched {
875 t.Fatalf("row columns %+v", rt)
876 }
877 // The edges loop's pass on the registry change the release made.
878 w.publish(uk)
879 w.publish(w.ie)
880 if !uk.lists(offHost) {
881 t.Fatal("uk-box's edges file does not list the opened host")
882 }
883 if w.ie.lists(offHost) {
884 t.Fatal("ie-box's edges file lists the moved host")
885 }
886}
887
888func (w *e2eWorld) sealDigest() string {
889 w.t.Helper()
890 raw, _, err := w.wd.s.Get(context.Background(), moves.OfflineSealKey(offHost, w.id))
891 if err != nil {
892 w.t.Fatal(err)
893 }
894 var r moves.OfflineReport
895 if err := json.Unmarshal(raw, &r); err != nil {
896 w.t.Fatal(err)
897 }
898 return r.TimersDigest
899}
900
901// firstWake is the target's first wake: the real exec backend's restore
902// promotes moves/<host>/ into data-uk's own prefix, and the database passes
903// the integrity check holding every row written before the stop.
904func (w *e2eWorld) firstWake() {
905 t := w.t
906 t.Helper()
907 ctx := context.Background()
908 m, err := lease.NewS3Mirror(ctx, sealUKRegion, sealUKRegion)
909 if err != nil {
910 t.Fatal(err)
911 }
912 rt := w.uk.row(offHost)
913 eb := &activator.ExecBackend{Cfg: activator.Config{Bucket: sealUK, Region: sealUKRegion, Prefix: sealPrefix}, DataDir: w.uk.dd, OptDir: w.uk.opt, Mirror: m}
914 l, err := eb.Acquire(ctx, *rt)
915 if err != nil {
916 t.Fatalf("the first wake's lease on data-uk: %v", err)
917 }
918 defer l.Release(ctx)
919 fresh, err := eb.Restore(ctx, *rt)
920 if err != nil || fresh {
921 t.Fatalf("restore: fresh %v, %v", fresh, err)
922 }
923 if err := lease.CheckDB(rt.DBPath); err != nil {
924 t.Fatal(err)
925 }
926 if n := countRows(t, rt.DBPath); n != w.rows {
927 t.Fatalf("the target restored %d rows, want %d", n, w.rows)
928 }
929 if got := w.s3.current(t, sealUK, activator.MovesPrefix(offHost)); len(got) != 0 {
930 t.Fatalf("staging left after the promotion: %v", keysOf(got))
931 }
932}
933
934// requireRetired: ie-box's parked report lists every source-retire path
935// as parked, absent or left with a reason; what it parked is in the park
936// directory and not in place; nothing it had is gone.
937func (w *e2eWorld) requireRetired() moves.OfflineReport {
938 t := w.t
939 t.Helper()
940 ie := w.ie
941 r := w.wd.reportFor(w.id, moves.OfflineRoleSource, "ie-box")
942 if r.Phase != moves.OfflinePhaseParked {
943 t.Fatalf("ie-box reports %s, want parked", r.Phase)
944 }
945 for _, rp := range r.Retired {
946 switch rp.Action {
947 case moves.RetiredParked:
948 if lexists(rp.Path) {
949 t.Errorf("%s is reported parked but is still in place", rp.Path)
950 }
951 case moves.RetiredAbsent:
952 if lexists(rp.Path) && !w.opts.nested {
953 t.Errorf("%s is reported absent but exists", rp.Path)
954 }
955 case moves.RetiredLeft:
956 if rp.Reason == "" {
957 t.Errorf("%s left with no reason", rp.Path)
958 }
959 default:
960 t.Errorf("%s: action %q", rp.Path, rp.Action)
961 }
962 }
963 if !w.opts.nested {
964 // Every path of the manifest's source-retire row, each with the
965 // action this source must get ("" is parked or absent, as the
966 // database's companions are whatever litestream left).
967 db := provision.DefaultDBPath(ie.dd, offHost)
968 files := moves.RetiredParked
969 if w.opts.noFiles {
970 files = moves.RetiredAbsent
971 }
972 want := map[string]string{
973 db: moves.RetiredParked, db + "-wal": "", db + "-shm": "",
974 db + "-litestream": moves.RetiredAbsent, activatorMetaDir(db): moves.RetiredParked,
975 filepath.Join(ie.dd, offHost): files,
976 filepath.Join(ie.dd, offHost+".env"): moves.RetiredParked,
977 filepath.Join(ie.dd, offHost+".config.env"): moves.RetiredParked,
978 filepath.Join(ie.dd, "programs", offHost+".json"): moves.RetiredAbsent,
979 filepath.Join(ie.opt, "live", offHost): moves.RetiredLeft,
980 }
981 got := map[string]string{}
982 for _, rp := range r.Retired {
983 got[rp.Path] = rp.Action
984 }
985 for p, action := range want {
986 a, ok := got[p]
987 if !ok || (action != "" && a != action) || (action == "" && a != moves.RetiredParked && a != moves.RetiredAbsent) {
988 t.Errorf("source-retire %s: %q (listed %v), want %q", p, a, ok, action)
989 }
990 }
991 if len(got) != len(want) {
992 t.Errorf("the inventory lists %d paths, want %d: %+v", len(got), len(want), r.Retired)
993 }
994 } else {
995 // Nested: the raw directory's contents, each listed, the platform's
996 // launch descriptors included; the directory and its view are left.
997 raw := layout.RawInstanceDir(ie.dd, offHost)
998 listed := map[string]string{}
999 for _, rp := range r.Retired {
1000 listed[rp.Path] = rp.Action
1001 }
1002 for _, name := range []string{offHost, offHost + ".env", offHost + ".config.env", offHost + ".db", "launch.run.json", "launch.gen.7.json"} {
1003 if listed[filepath.Join(raw, name)] != moves.RetiredParked {
1004 t.Errorf("the nested source's %s: %q, want parked", name, listed[filepath.Join(raw, name)])
1005 }
1006 }
1007 if listed[raw] != moves.RetiredLeft {
1008 t.Errorf("the raw instance directory: %q, want left", listed[raw])
1009 }
1010 }
1011 w.requireParkedBytes()
1012 return r
1013}
1014
1015// requireParkedBytes: nothing the source had is gone. The app's files, env
1016// and config are in the park directory byte for byte, and the database with
1017// every row.
1018func (w *e2eWorld) requireParkedBytes() {
1019 t := w.t
1020 t.Helper()
1021 park := moves.OfflineParkDir(w.ie.dd, offHost, w.id, moves.OfflineRoleSource)
1022 if w.opts.nested {
1023 park = filepath.Join(park, "instance")
1024 for _, name := range []string{"launch.run.json", "launch.gen.7.json"} {
1025 if !lexists(filepath.Join(park, name)) {
1026 t.Errorf("the nested source's %s was not parked", name)
1027 }
1028 }
1029 }
1030 if !w.opts.noFiles {
1031 if got := tree(t, filepath.Join(park, offHost)); !reflect.DeepEqual(got, w.srcTree) {
1032 t.Fatalf("the parked files differ:\nsource %v\nparked %v", w.srcTree, got)
1033 }
1034 }
1035 if got := mustRead(t, filepath.Join(park, offHost+".env")); !bytes.Equal(got, w.srcEnvRaw) {
1036 t.Fatal("the parked env differs")
1037 }
1038 if got := mustRead(t, filepath.Join(park, offHost+".config.env")); !bytes.Equal(got, w.srcCfgRaw) {
1039 t.Fatal("the parked config differs")
1040 }
1041 if n := countRows(t, filepath.Join(park, offHost+".db")); n != w.rows {
1042 t.Fatalf("the parked database holds %d rows, want %d", n, w.rows)
1043 }
1044}
1045
1046// reportFor reads one box's report for op.
1047func (wd *offWorld) reportFor(op, role, box string) moves.OfflineReport {
1048 wd.t.Helper()
1049 body, _, err := wd.s.Get(context.Background(), moves.OfflineReportKey(offHost, op, role, box))
1050 if err != nil {
1051 wd.t.Fatalf("no %s report from %s: %v", role, box, err)
1052 }
1053 var r moves.OfflineReport
1054 if err := json.Unmarshal(body, &r); err != nil {
1055 wd.t.Fatal(err)
1056 }
1057 if err := moves.ValidateOfflineReport(r); err != nil {
1058 wd.t.Fatal(err)
1059 }
1060 return r
1061}
1062
1063// TestOfflineMoveE2E is the two-box witness: the whole offline move through
1064// the console's doors and both boxes' production passes.
1065func TestOfflineMoveE2E(t *testing.T) {
1066 t.Run("done", func(t *testing.T) { offlineMoveDone(t, landOpts{}) })
1067 t.Run("nested source with launch files", func(t *testing.T) { offlineMoveDone(t, landOpts{nested: true}) })
1068 t.Run("no files directory", func(t *testing.T) { offlineMoveDone(t, landOpts{noFiles: true}) })
1069 t.Run("rollback after a tampered archive", offlineMoveRollback)
1070 // Harness Run 9, finding 7: the doors driven with nothing between them
1071 // but what production runs (each door's serial bump, the boxes' own
1072 // nudges and the offline re-assert loop), first back to back, then with
1073 // the operator three minutes behind at every step, past the doors' two
1074 // minutes of belief in a report.
1075 t.Run("doors back to back", func(t *testing.T) { offlineMoveDoorsApart(t, 0) })
1076 t.Run("doors three minutes apart", func(t *testing.T) { offlineMoveDoorsApart(t, 3*time.Minute) })
1077}
1078
1079// e2eLoops are what acts on the two boxes between door calls in production,
1080// and nothing else: each box's fast-converge tick (fastTick, the real serial
1081// rule: a moved serial runs the pass set once per clamp, and a nudge made
1082// during the pass runs it again), whose pass set is the agent's order
1083// (reconcile, the edges file, the moves pass); and each box's offline
1084// re-assert tick (offlineReassert). Nothing else ticks a box.
1085type e2eLoops struct {
1086 w *e2eWorld
1087 fast map[*offBox]*fastDeps
1088 tickAt time.Time // fastTick's own clock, stepped past the clamp each round
1089 boxAt time.Time // what the boxes' clock reads during a serial pass
1090 passes int // serial passes run
1091 again int // re-assert ticks run
1092}
1093
1094func (w *e2eWorld) loops() *e2eLoops {
1095 l := &e2eLoops{w: w, fast: map[*offBox]*fastDeps{}, tickAt: time.Now()}
1096 for _, b := range []*offBox{w.ie, w.uk} {
1097 // The sweep's registry is empty: the sweep stages releases and is
1098 // not what this test is about. The serial leg is.
1099 l.fast[b] = &fastDeps{reg: restartTestRegistry(w.t), store: w.rs, fullPass: func(context.Context) {
1100 l.passes++
1101 w.tickAt(b, l.boxAt)
1102 }}
1103 }
1104 l.round() // the boot read seeds each box's serial
1105 return l
1106}
1107
1108// round is one fastTick on each box; it reports whether either ran its pass.
1109func (l *e2eLoops) round() bool {
1110 before := l.passes
1111 l.tickAt = l.tickAt.Add(fastFullClamp + time.Second)
1112 for _, b := range []*offBox{l.w.ie, l.w.uk} {
1113 fastTick(context.Background(), l.fast[b], l.tickAt)
1114 }
1115 return l.passes > before
1116}
1117
1118// after is what the boxes do once a door has bumped the serial, when the
1119// operator runs the next door gap later: the serial passes at the time of
1120// the bump, then the re-assert loop's ticks through the gap, up to now.
1121func (l *e2eLoops) after(gap time.Duration) {
1122 t := l.w.t
1123 t.Helper()
1124 l.boxAt = time.Now().Add(-gap)
1125 rounds := 0
1126 for l.round() {
1127 if rounds++; rounds > 4 {
1128 t.Fatal("the serial kept moving after four rounds of passes")
1129 }
1130 }
1131 if gap > 0 {
1132 // Without the re-assert, a door would read the reports as the bump's
1133 // passes left them: older than it believes. The status view says
1134 // so, and never as a process (MV-O31).
1135 if v := l.w.status(); v.Next != "" || !strings.Contains(v.Waiting, "is not current") {
1136 t.Fatalf("%s behind the bump, the view read %+v; want a report that is not current", gap, v)
1137 }
1138 }
1139 for at := l.boxAt.Add(offlineReassertEvery); !at.After(time.Now()); at = at.Add(offlineReassertEvery) {
1140 l.reassert(at)
1141 }
1142}
1143
1144func (l *e2eLoops) reassert(at time.Time) {
1145 l.w.setBoxClock(at)
1146 for _, b := range []*offBox{l.w.ie, l.w.uk} {
1147 b.w.offlineReassert(context.Background(), l.w.rs)
1148 }
1149 l.again++
1150}
1151
1152// offlineMoveDoorsApart drives start to open with the operator gap behind
1153// each door's bump, and wants every door accepted on its first or second try,
1154// the second after one more re-assert tick. It logs the time each door took
1155// to its answer and from start to open.
1156func offlineMoveDoorsApart(t *testing.T, gap time.Duration) {
1157 w := newE2EWorld(t, landOpts{})
1158 l := w.loops()
1159 tries := map[string]int{}
1160 door := func(verb, body, want string) e2eView {
1161 t.Helper()
1162 for try := 1; ; try++ {
1163 code, out := w.door(verb, body)
1164 if code == http.StatusAccepted {
1165 tries[verbName(verb)] = try
1166 v := w.decode(out)
1167 if v.State != want {
1168 t.Fatalf("%s: state %q, want %q", verbName(verb), v.State, want)
1169 }
1170 return v
1171 }
1172 if try == 2 {
1173 t.Fatalf("%s refused twice, %s behind the previous step: %d %s", verbName(verb), gap, code, out)
1174 }
1175 l.reassert(time.Now())
1176 }
1177 }
1178 next := func(step, verb string) {
1179 t.Helper()
1180 if v := w.status(); v.Next != verb {
1181 t.Fatalf("after %s, %s behind: next %q, want %q (waiting %q)", step, gap, v.Next, verb, v.Waiting)
1182 }
1183 }
1184 timers, err := w.ie.reg.TimersFor(offHost)
1185 if err != nil {
1186 t.Fatal(err)
1187 }
1188 w.srcTimers = map[string]registry.Timer{}
1189 for _, tm := range timers {
1190 w.srcTimers[tm.Name] = tm
1191 }
1192 began := time.Now()
1193 var took []string
1194 timed := func(name string, f func()) {
1195 at := time.Now()
1196 f()
1197 took = append(took, fmt.Sprintf("%s %s", name, time.Since(at).Round(time.Millisecond)))
1198 }
1199
1200 timed("start", func() {
1201 w.id = door("", `{"region":"uk"}`, provision.OfflineStopping).ID
1202 l.after(gap)
1203 })
1204 next("start", "seal")
1205 w.renderSourceConfig()
1206 timed("seal", func() {
1207 door("seal", "", provision.OfflineSealing)
1208 l.after(gap)
1209 })
1210 if st := w.wd.record().Move.State; st != provision.OfflineSealed {
1211 t.Fatalf("state %s after the seal's passes", st)
1212 }
1213 next("seal", "flip")
1214 timed("flip", func() {
1215 door("flip", "", provision.OfflineLanding)
1216 l.after(gap)
1217 })
1218 if st := w.wd.record().Move.State; st != provision.OfflineLanded {
1219 t.Fatalf("state %s after the flip's passes", st)
1220 }
1221 next("flip", "open")
1222 timed("open", func() {
1223 door("open", "", provision.OfflineOpen)
1224 l.after(0)
1225 })
1226 next("open", "finish")
1227 total := time.Since(began).Round(time.Millisecond)
1228 t.Logf("operator gap %s: start to open in %s of test time (%s); %d serial passes, %d re-assert ticks; tries %v",
1229 gap, total, strings.Join(took, ", "), l.passes, l.again, tries)
1230 // The move is the real one: the target holds what the source had.
1231 w.requireTarget()
1232}
1233
1234func offlineMoveDone(t *testing.T, opts landOpts) {
1235 w := newE2EWorld(t, opts)
1236 ctx := context.Background()
1237
1238 // The env pin's premise: the two boxes compose under different agent
1239 // environments, each its own.
1240 for _, c := range []struct {
1241 b *offBox
1242 env map[string]string
1243 }{{w.ie, e2eIEAgentEnv}, {w.uk, e2eUKAgentEnv}} {
1244 rt := registry.Route{Host: offHost, Kind: "instance", DBPath: provision.DefaultDBPath(c.b.dd, offHost), Socket: filepath.Join(provision.SocketDir, offHost+".sock")}
1245 env, _, _, err := c.b.w.cfg.LaunchLayers(rt)
1246 if err != nil {
1247 t.Fatal(err)
1248 }
1249 for k, v := range c.env {
1250 if env[k] != v {
1251 t.Fatalf("%s composes %s=%q, want its own %q", c.b.name, k, env[k], v)
1252 }
1253 }
1254 }
1255 if os.Getenv("SYSTEMD_EXEC_PID") != "test-process" {
1256 t.Fatal("the agent environment swap leaked into the process")
1257 }
1258
1259 w.begin()
1260 w.seal()
1261 w.flipAndLand()
1262 if r := w.wd.reportFor(w.id, moves.OfflineRoleTarget, "uk-box"); !r.Items["env-pin"].OK {
1263 t.Fatalf("env-pin across the two agent environments: %+v", r.Items["env-pin"])
1264 }
1265 w.requireStopped("after the landing")
1266 w.open()
1267
1268 w.requireTarget()
1269 retired := w.requireRetired()
1270 if rt := w.ie.row(offHost); rt == nil {
1271 t.Fatal("ie-box's row went before finishing")
1272 }
1273 w.requireTombstoned("after open")
1274 w.requireSourceSchedulerQuiet()
1275 // The writes so far, before the first wake (which is not the tool's).
1276 w.requireClosedWriteSet("through open")
1277 w.firstWake()
1278 w.storeMark = w.rs.mark()
1279 w.snapBuckets()
1280
1281 w.finish("done")
1282 w.requireClosedWriteSet("from open to the end")
1283 w.requireTombstoneLifted()
1284 if rt := w.ie.row(offHost); rt != nil {
1285 t.Fatalf("ie-box still has a row after finish: %+v", rt)
1286 }
1287 w.tickAll()
1288 if w.ie.lists(offHost) || !w.uk.lists(offHost) {
1289 t.Fatalf("after finish: ie lists %v, uk lists %v", w.ie.lists(offHost), w.uk.lists(offHost))
1290 }
1291 w.requireRetiredAfterFinish(retired)
1292
1293 after := w.requireRecordDiff("after finish")
1294 if after["region"] != "uk" {
1295 t.Fatalf("region %v", after["region"])
1296 }
1297 if _, ok := after["move"]; ok {
1298 t.Fatalf("a move is left on the record: %v", after["move"])
1299 }
1300 rec := w.wd.record()
1301 if rec.LastMove == nil || rec.LastMove.Outcome != "done" || rec.LastMove.ID != w.id {
1302 t.Fatalf("last_move %+v", rec.LastMove)
1303 }
1304 if len(w.sw.pages) != 0 {
1305 t.Fatalf("pages fired: %v", w.sw.pages)
1306 }
1307 // The source's lease is free once the move has finished.
1308 held, err := acquireAsAWake(ctx)
1309 if err != nil {
1310 t.Fatalf("data-ie after finish: %v", err)
1311 }
1312 held.Release(ctx)
1313}
1314
1315// requireSourceSchedulerQuiet is harness Run 9's defect 8. After open the
1316// source keeps its row, held, until finishing, its timers with it, and open
1317// has lifted the fence. The source's own schedule runner, built as the agent
1318// builds it (newScheduleDeps over ie-box's registry, the deployment store,
1319// its activator and data directory), ticks with every timer due: it writes
1320// nothing for the host. No env file at either layout's path (the tick used
1321// to mint one, a fresh admin token, in the parked instance directory), no
1322// timer row moves, and no store write.
1323func (w *e2eWorld) requireSourceSchedulerQuiet() {
1324 t := w.t
1325 t.Helper()
1326 ie := w.ie
1327 if ie.svc.Fenced(offHost) {
1328 t.Fatal("fixture: ie-box is still fenced after open")
1329 }
1330 // Run 9's app declared a one-minute schedule; the row is the source's
1331 // as the converger writes it, due now. (The fixture's own timers are
1332 // the seal's, kept for the digest, and are not all ones the runner can
1333 // fire.)
1334 if err := ie.reg.UpsertTimer(registry.Timer{Host: offHost, Name: "every-minute", Kind: schedule.KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: 1}); err != nil {
1335 t.Fatal(err)
1336 }
1337 before, err := ie.reg.TimersFor(offHost)
1338 if err != nil || len(before) == 0 {
1339 t.Fatalf("fixture: ie-box keeps no timers for the host after open (%v)", err)
1340 }
1341 mark := w.rs.mark()
1342 sd := newScheduleDeps(ie.reg, w.rs, ie.svc, nil, ie.dd)
1343 sd.runner.Tick(context.Background())
1344 // A delivery the tick started holds the host's claim until it ends.
1345 for deadline := time.Now().Add(10 * time.Second); sd.runner.Busy(offHost); {
1346 if time.Now().After(deadline) {
1347 t.Fatal("the source's tick did not finish")
1348 }
1349 time.Sleep(5 * time.Millisecond)
1350 }
1351 for _, p := range []string{
1352 filepath.Join(ie.dd, offHost+".env"),
1353 filepath.Join(provision.InstanceDir(ie.dd, offHost), offHost+".env"),
1354 filepath.Join(layout.RawInstanceDir(ie.dd, offHost), offHost+".env"),
1355 } {
1356 if lexists(p) {
1357 t.Fatalf("the source's scheduler wrote %s after open", p)
1358 }
1359 }
1360 after, err := ie.reg.TimersFor(offHost)
1361 // The runner's daily sweep of finished once-rows is box-wide and is not
1362 // a tick of the host (the timers digest leaves that history out, joint
1363 // N14): only the live rows are compared.
1364 live := func(ts []registry.Timer) []registry.Timer {
1365 return slices.DeleteFunc(slices.Clone(ts), func(tm registry.Timer) bool { return tm.Kind == "once" && tm.NextAt == 0 })
1366 }
1367 before, after = live(before), live(after)
1368 if err != nil || !reflect.DeepEqual(after, before) {
1369 t.Fatalf("the source's tick moved the host's timers (%v):\nbefore %+v\nafter %+v", err, before, after)
1370 }
1371 if ops := w.rs.since(mark); len(ops) != 0 {
1372 t.Fatalf("the source's tick wrote to the store: %v", ops)
1373 }
1374}
1375
1376// requireRetiredAfterFinish: finishing removed the row and nothing parked
1377// came back or went.
1378func (w *e2eWorld) requireRetiredAfterFinish(at moves.OfflineReport) {
1379 t := w.t
1380 t.Helper()
1381 park := moves.OfflineParkDir(w.ie.dd, offHost, w.id, moves.OfflineRoleSource)
1382 if !isDir(park) {
1383 t.Fatal("the park directory is gone after finish")
1384 }
1385 for _, rp := range at.Retired {
1386 if rp.Action == moves.RetiredParked && lexists(rp.Path) {
1387 t.Fatalf("%s came back after finish", rp.Path)
1388 }
1389 }
1390 r := w.wd.reportFor(w.id, moves.OfflineRoleSource, "ie-box")
1391 if r.Phase != moves.OfflinePhaseGone {
1392 t.Fatalf("ie-box reports %s after finish, want gone", r.Phase)
1393 }
1394 w.requireParkedBytes()
1395}
1396
1397// A tampered archive fails the landing; open refuses; abort lifts the stop
1398// and unfences; the source serves again with every row.
1399func offlineMoveRollback(t *testing.T) {
1400 w := newE2EWorld(t, landOpts{})
1401 ctx := context.Background()
1402 w.begin()
1403 w.seal()
1404
1405 // Tamper with the sealed archive in data-uk.
1406 key := moves.OfflineArchivePrefix(offHost, w.id) + offlineArchiveSealed
1407 body, ok := w.sw.fake.Current(sealUK, key)
1408 if !ok || len(body) < 64 {
1409 t.Fatal("fixture: no sealed archive")
1410 }
1411 tampered := bytes.Clone(body)
1412 tampered[len(tampered)/2] ^= 0xff
1413 w.sw.fake.Put(sealUK, key, tampered)
1414
1415 w.accept("flip", "", provision.OfflineLanding)
1416 var v e2eView
1417 for range 4 {
1418 w.tickAll()
1419 if v = w.status(); v.Failed {
1420 break
1421 }
1422 }
1423 r := w.wd.reportFor(w.id, moves.OfflineRoleTarget, "uk-box")
1424 if !v.Failed || r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK {
1425 t.Fatalf("the landing of a tampered archive: view %+v, report %s %+v", v, r.Phase, r.Items["files"])
1426 }
1427 if code, out := w.door("open", ""); code != http.StatusConflict {
1428 t.Fatalf("open on a failed landing = %d %s", code, out)
1429 }
1430 if rt := w.uk.row(offHost); rt == nil || rt.HoldOp != w.id {
1431 t.Fatalf("uk-box's row %+v, want held by the move", rt)
1432 }
1433
1434 w.accept("abort", "", provision.OfflineReverting)
1435 w.until("the revert", nextIs("finish"))
1436 w.finish("aborted")
1437 w.requireClosedWriteSet("through the abort")
1438 after := w.requireRecordDiff("after the abort")
1439 if r, ok := after["region"]; ok && r != "" {
1440 t.Fatalf("region %v after the abort, want home", r)
1441 }
1442 if _, ok := after["move"]; ok {
1443 t.Fatalf("a move is left on the record: %v", after["move"])
1444 }
1445
1446 // The source: the row ready and listed, unfenced, the tombstone gone.
1447 w.tickAll()
1448 rt := w.ie.row(offHost)
1449 if rt == nil || rt.Status != "ready" || rt.HoldOp != "" {
1450 t.Fatalf("ie-box's row after the abort: %+v", rt)
1451 }
1452 if !w.ie.lists(offHost) {
1453 t.Fatal("ie-box does not list the host after the abort")
1454 }
1455 if w.ie.svc.Fenced(offHost) {
1456 t.Fatal("ie-box is still fenced after the abort")
1457 }
1458 w.requireTombstoneLifted()
1459
1460 // The target: no row; what it extracted is parked, not deleted.
1461 if rt := w.uk.row(offHost); rt != nil {
1462 t.Fatalf("uk-box has a row after the abort: %+v", rt)
1463 }
1464 for _, name := range []string{offHost, offHost + ".env", offHost + ".config.env"} {
1465 if lexists(filepath.Join(w.uk.dd, name)) {
1466 t.Errorf("uk-box keeps %s in place after the abort", name)
1467 }
1468 }
1469 if parked := w.parkedTarget(); !parked[offHost+".config.env"] {
1470 t.Errorf("uk-box parked %v, want what it rendered", parked)
1471 }
1472 if w.uk.lists(offHost) {
1473 t.Fatal("uk-box lists the host after the abort")
1474 }
1475 // Both staging prefixes are empty, and data-uk's own prefix untouched.
1476 for _, p := range []string{activator.MovesPrefix(offHost), moves.OfflineArchivePrefix(offHost, w.id), activator.HibernatePrefix(sealPrefix, offHost) + "/"} {
1477 if got := w.s3.current(t, sealUK, p); len(got) != 0 {
1478 t.Fatalf("data-uk %s holds %v after the abort", p, keysOf(got))
1479 }
1480 }
1481
1482 // The source serves again with no write lost: the wake takes the
1483 // lease, the warm cache is the database, and it holds every row; so
1484 // does the replica.
1485 l, err := w.ieEB.Acquire(ctx, *rt)
1486 if err != nil {
1487 t.Fatalf("a wake on ie-box after the abort: %v", err)
1488 }
1489 defer l.Release(ctx)
1490 if fresh, err := w.ieEB.Restore(ctx, *rt); err != nil || fresh {
1491 t.Fatalf("restore on ie-box: fresh %v, %v", fresh, err)
1492 }
1493 if err := lease.CheckDB(rt.DBPath); err != nil {
1494 t.Fatal(err)
1495 }
1496 if n := countRows(t, rt.DBPath); n != w.rows {
1497 t.Fatalf("ie-box's database holds %d rows, want %d", n, w.rows)
1498 }
1499 scratch := filepath.Join(t.TempDir(), "replica.db")
1500 if ok, err := lease.Restore(ctx, scratch, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion); err != nil || !ok {
1501 t.Fatalf("restore data-ie's replica: %v %v", ok, err)
1502 }
1503 if n := countRows(t, scratch); n != w.rows {
1504 t.Fatalf("data-ie's replica holds %d rows, want %d", n, w.rows)
1505 }
1506 if len(w.sw.pages) != 0 {
1507 t.Fatalf("pages fired: %v", w.sw.pages)
1508 }
1509}
1510
1511// parkedTarget is the base names uk-box parked for the move.
1512func (w *e2eWorld) parkedTarget() map[string]bool {
1513 out := map[string]bool{}
1514 for _, role := range []string{moves.OfflineRoleTarget, moves.OfflineParkTargetBefore} {
1515 ents, _ := os.ReadDir(moves.OfflineParkDir(w.uk.dd, offHost, w.id, role))
1516 for _, e := range ents {
1517 out[e.Name()] = true
1518 }
1519 }
1520 return out
1521}
1522