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 "context"
5 "encoding/json"
6 "errors"
7 "net/http/httptest"
8 "os"
9 "path/filepath"
10 "sort"
11 "strings"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/activator"
16 "github.com/carlosframework/platform/internal/hostclaim"
17 "github.com/carlosframework/platform/internal/lease"
18 "github.com/carlosframework/platform/internal/lease/leasetest"
19 "github.com/carlosframework/platform/internal/moves"
20 "github.com/carlosframework/platform/internal/moves/archive"
21 "github.com/carlosframework/platform/internal/provision"
22 "github.com/carlosframework/platform/internal/registry"
23 "github.com/carlosframework/platform/internal/release"
24)
25
26// The cut harness: the Phase 1 harness driven to ready (manual cut) and
27// cut as the door does, with the cut's hooks faked. fakeEvidence answers a
28// configurable report; the final checkpoint writes the stopped database's
29// last transaction (TXID 3) into the source's replica; cutFault is the
30// crash seam, crashing once at the point named.
31
32var errCrash = errors.New("injected crash")
33
34type fakeEvidence struct {
35 calls int
36 stopped bool
37 err error
38 onCall func()
39}
40
41func (e *fakeEvidence) report() moves.StoppedReport {
42 r := moves.StoppedReport{V: 1, Host: p1Host, Steps: map[string]moves.Check{}, CheckedAt: time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)}
43 for _, id := range []string{moves.StepTree, moves.StepStray, moves.StepFDs, moves.StepLease} {
44 r.Steps[id] = moves.Check{OK: true}
45 }
46 if !e.stopped {
47 r.Steps[moves.StepFDs] = moves.Check{Reason: "1 process(es) hold the database open"}
48 }
49 return r
50}
51
52type cutHarness struct {
53 *phase1Harness
54 ev *fakeEvidence
55 holds []string
56 fences int
57 drains int
58 checkpoints int
59 locations []string
60 bundle map[string]string
61 localPos uint64
62 timers []registry.Timer
63 crashAt string
64 crashed bool
65 hits []string
66 // scoped keeps the Phase 1 harness's scoped holder
67 // (instances/<host>/<incarnation>), which the cut refuses (ruling
68 // T5-lease). By default ready moves the claim to the root location the
69 // activator's lease lives at, instances/<host>, generation cur.
70 scoped bool
71}
72
73// cutRoot is the activator's lease prefix for the harness's host, the one
74// location Phase 1 cuts (ruling T5-lease); cutLineage is the claim
75// generation's replica under it, the prefix the lane mirrors.
76var (
77 cutRoot = "instances/" + p1Host
78 cutLineage = cutRoot + "/g-cur/"
79)
80
81// PutFileIfAbsent makes the fake a lease.FilePutIfAbsent.
82func (v *fakeVersioned) PutFileIfAbsent(ctx context.Context, bucket, key, path string) (string, error) {
83 body, err := os.ReadFile(path)
84 if err != nil {
85 return "", err
86 }
87 return v.PutIfAbsent(ctx, bucket, key, body)
88}
89
90var _ lease.FilePutIfAbsent = (*fakeVersioned)(nil)
91
92const lastTXKey = "0000/0000000000000003-0000000000000003.ltx"
93
94func newCutHarness(t *testing.T, opts ...p1Option) *cutHarness {
95 t.Helper()
96 h := &cutHarness{phase1Harness: newPhase1Harness(t, opts...), ev: &fakeEvidence{stopped: true}, localPos: 3,
97 bundle: map[string]string{"default": "e1", "environment": ""}}
98 h.timers = []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}}
99 last := ltxForTest(t, 4096, 4, 3, 3)
100 if err := os.WriteFile(h.route.DBPath, []byte("the stopped database"), 0o600); err != nil {
101 t.Fatal(err)
102 }
103 h.cfg.HoldMoving = func(host, op string) error { h.holds = append(h.holds, host+"="+op); return nil }
104 h.cfg.Fence = func(context.Context, string, uint64) error { h.fences++; return nil }
105 h.cfg.TimersDrained = func(context.Context, string) error { h.drains++; return nil }
106 h.cfg.Evidence = func(_ context.Context, host, location string, _ []string) (moves.StoppedReport, error) {
107 h.ev.calls++
108 h.locations = append(h.locations, location)
109 if h.ev.onCall != nil {
110 h.ev.onCall()
111 }
112 return h.ev.report(), h.ev.err
113 }
114 h.cfg.FinalCheckpoint = func(ctx context.Context, host, _ string, holder hostclaim.Holder, check func(context.Context) error) error {
115 if err := check(ctx); err != nil {
116 return err
117 }
118 h.checkpoints++
119 prefix, _ := moves.ReplicaSource(holder)
120 if host != p1Host || prefix != cutLineage {
121 t.Errorf("checkpoint of %s at %s, want %s at the claim's lineage %s", host, prefix, p1Host, cutLineage)
122 }
123 h.f.store(h.srcBucket, prefix+lastTXKey, fakeMoveObject{lease.MirrorObject{ETag: "cur-c", Size: int64(len(last))}, string(last)})
124 return nil
125 }
126 h.cfg.LocalPosition = func(string) (uint64, error) { return h.localPos, nil }
127 h.cfg.Timers = func(string) ([]registry.Timer, error) { return h.timers, nil }
128 h.cfg.InputVersions = func(context.Context, release.Store, provision.Record) (map[string]string, string, error) {
129 out := map[string]string{}
130 for k, v := range h.bundle {
131 out[k] = v
132 }
133 return out, "", nil
134 }
135 h.cfg.cutFault = func(point string) error {
136 h.hits = append(h.hits, point)
137 if point == h.crashAt && !h.crashed {
138 h.crashed = true
139 return errCrash
140 }
141 return nil
142 }
143 // The cut guard: the source can recover a failed cut, as every box
144 // will once the wake-source gate opens (armStandIns).
145 armStandIns(&h.cfg, h.s, roleSource)
146 return h
147}
148
149// targetCfg composes with the harness's current bundle versions.
150func (h *cutHarness) target() moveCfg {
151 c := h.targetCfg()
152 c.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) {
153 env, instance, err := h.heldEnv(ctx, s, rec, carried)
154 if err != nil {
155 return nil, nil, nil, provision.MoveInputs{}, err
156 }
157 b := map[string]string{}
158 for k, v := range h.bundle {
159 b[k] = v
160 }
161 return env, instance, nil, provision.MoveInputs{BundleETags: b}, nil
162 }
163 return c
164}
165
166// ready drives the move to a persisted ready with its inputs pinned. The
167// claim holder is at the root location unless the harness is scoped.
168func (h *cutHarness) ready(t *testing.T) {
169 t.Helper()
170 h.start(t, "manual")
171 if !h.scoped {
172 rootHolder(t, h.phase1Harness)
173 }
174 h.pass(t)
175 h.targetPass(t, h.target())
176 h.pass(t)
177 if got := h.read(t, p1Host).Move; got.State != provision.MoveReady || got.Inputs == nil {
178 t.Fatalf("the fixture did not reach ready: %+v", got)
179 }
180}
181
182// rootHolder moves a started Phase 1 harness to the root location: the
183// Phase 1 harness keeps the replica under the scoped location, the cut
184// accepts only the root one (ruling T5-lease), so the same objects move
185// there and the claim follows.
186func rootHolder(t *testing.T, h *phase1Harness) {
187 t.Helper()
188 scoped := "instances/" + p1Host + "/" + h.inc + "/"
189 for k, o := range h.f.buckets[h.srcBucket] {
190 if rel, ok := strings.CutPrefix(k, scoped); ok {
191 delete(h.f.buckets[h.srcBucket], k)
192 h.f.store(h.srcBucket, cutRoot+"/"+rel, o)
193 }
194 }
195 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Location = cutRoot })
196}
197
198// cut is the console's cut door: ready to cutting/requested.
199func (h *cutHarness) cut(t *testing.T) {
200 t.Helper()
201 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
202 if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); err != nil {
203 t.Fatal(err)
204 }
205}
206
207func (h *cutHarness) move(t *testing.T) *provision.MoveState {
208 t.Helper()
209 return h.read(t, p1Host).Move
210}
211
212// crashPass runs one pass that must crash at h.crashAt.
213func (h *cutHarness) crashPass(t *testing.T) {
214 t.Helper()
215 h.pass(t)
216 if !h.crashed {
217 t.Fatalf("the pass never reached %s (points reached: %v)", h.crashAt, h.hits)
218 }
219}
220
221// freshWorker drops the source worker: the next pass starts with no
222// in-memory state, as a restarted agent does.
223func (h *cutHarness) freshWorker() { h.worker1 = nil }
224
225func (h *cutHarness) stoppedKey() string {
226 return moves.StoppedKey(moves.ReportPrefix("bab", "app", p1Host, p1ID), "source")
227}
228
229func (h *cutHarness) seal(t *testing.T) moves.Seal {
230 t.Helper()
231 s, err := journalledSeal(h.claim(t))
232 if err != nil {
233 t.Fatal(err)
234 }
235 return s
236}
237
238func (h *cutHarness) journalArchive(t *testing.T) cutArchiveStep {
239 t.Helper()
240 j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
241 if err != nil {
242 t.Fatal(err)
243 }
244 var step cutArchiveStep
245 if err := json.Unmarshal([]byte(j.Steps[cutStepArchive]), &step); err != nil {
246 t.Fatalf("no archive identity journalled: %v", err)
247 }
248 step.dir = archiveDir(h.cfg.DataDir, p1Host, p1ID)
249 return step
250}
251
252// Admission needs this round's own passing source report, a fresh passing
253// target report, and the pin to match (G1, G13). With all of them the
254// record is admitted with cut_at, the budget and the deadline; a stale
255// target report or a failing source sends it back to ready with the pin
256// cleared, and a passing round pins again.
257func TestCutAdmitsOnlyFreshPassingReports(t *testing.T) {
258 t.Parallel()
259 h := newCutHarness(t)
260 h.ready(t)
261 h.ev.stopped = false // stop at stopping, to look at the admission
262 h.cut(t)
263 h.pass(t)
264 m := h.move(t)
265 if m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping || m.CutAt == nil || m.CutBudgetS != int(moves.CutWindowDefault) || m.CutDeadline == nil {
266 t.Fatalf("after the admission: %+v", m)
267 }
268 if want := moves.CutDeadline(*m.CutAt, moves.CutWindowDefault); !m.CutDeadline.Equal(want) {
269 t.Fatalf("deadline %v, want %v", m.CutDeadline, want)
270 }
271
272 for name, spoil := range map[string]func(h *cutHarness){
273 "the target's report is stale": func(h *cutHarness) { h.clock.Advance(moves.ReportFreshness + time.Minute) },
274 "the source does not answer": func(h *cutHarness) { h.probe = moves.Check{Reason: s1NoAnswer(p1Host)} },
275 "a second box in the target region": func(h *cutHarness) {
276 if _, err := h.s.Put(context.Background(), "control/edges/uk-box2.json",
277 []byte(`{"v":1,"box":"uk-box2","eip":"1.2.3.9","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil {
278 panic(err)
279 }
280 },
281 } {
282 t.Run(name, func(t *testing.T) {
283 h := newCutHarness(t)
284 h.ready(t)
285 h.cut(t)
286 spoil(h)
287 h.pass(t)
288 m := h.move(t)
289 if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.CutAt != nil {
290 t.Fatalf("not back to ready: %+v", m)
291 }
292 if h.ev.calls != 0 || len(h.holds) != 0 {
293 t.Fatalf("something stopped: %d evidence calls, holds %v", h.ev.calls, h.holds)
294 }
295 })
296 }
297}
298
299// The inputs pinned at readiness are compared at requested and again at
300// admitted (§7.4). A drift goes back to ready; the target reports the new
301// versions, a passing round pins them (MoveGuardedRepin), and the next cut
302// admits: no loop.
303func TestCutGoesBackToReadyWhenInputsMove(t *testing.T) {
304 t.Parallel()
305 ctx := context.Background()
306 h := newCutHarness(t)
307 h.ready(t)
308 h.ev.stopped = false
309 h.bundle["environment"] = "e2"
310 h.cut(t)
311 h.pass(t)
312 if m := h.move(t); m.State != provision.MoveReady || m.Inputs != nil {
313 t.Fatalf("a drift at requested did not go back to ready: %+v", m)
314 }
315 // Ready without a pin is never cut: the door refuses.
316 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
317 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); !errors.Is(err, provision.ErrMoveConflict) {
318 t.Fatalf("cut without a pin: %v", err)
319 }
320 h.targetPass(t, h.target())
321 h.pass(t)
322 m := h.move(t)
323 if m.State != provision.MoveReady || m.Inputs == nil || m.Inputs.BundleETags["environment"] != "e2" {
324 t.Fatalf("not pinned again: %+v", m)
325 }
326 h.cut(t)
327 h.pass(t)
328 if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping {
329 t.Fatalf("the repinned cut did not admit: %+v", m)
330 }
331
332 // At admitted: a record admitted before the drift goes back to ready
333 // on the replay of admitted.
334 h2 := newCutHarness(t)
335 h2.ready(t)
336 h2.cut(t)
337 now := h2.clock.Now()
338 if _, err := provision.SetRecordMove(ctx, h2.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
339 t.Fatal(err)
340 }
341 h2.bundle["default"] = "e9"
342 h2.pass(t)
343 if m := h2.move(t); m.State != provision.MoveReady || m.Inputs != nil || h2.ev.calls != 0 {
344 t.Fatalf("a drift at admitted: %+v, %d evidence calls", m, h2.ev.calls)
345 }
346}
347
348// Review Focus 1: a fresh worker on the same store, with the record at
349// stopping and nothing in memory, holds, fences, drains, asks for the
350// evidence and advances; the cut then runs to the flip in one pass.
351func TestCutResumesStoppingFromEvidenceAlone(t *testing.T) {
352 h := newCutHarness(t)
353 h.ready(t)
354 h.ev.stopped = false
355 h.cut(t)
356 h.pass(t)
357 h.pass(t)
358 if m := h.move(t); m.CutPhase != provision.CutStopping || h.ev.calls != 2 {
359 t.Fatalf("an unproven stop advanced: %+v after %d calls", m, h.ev.calls)
360 }
361 if h.claim(t).Op.Exec != "ie-box" {
362 t.Fatal("execution moved before the evidence")
363 }
364 h.freshWorker()
365 h.ev.stopped = true
366 h.pass(t)
367 if m := h.move(t); m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped {
368 t.Fatalf("the resumed cut: %+v", m)
369 }
370 if h.ev.calls != 3 || h.fences != 3 || h.drains != 3 || len(h.holds) != 3 || h.holds[0] != p1Host+"="+p1ID {
371 t.Fatalf("evidence %d, fences %d, drains %d, holds %v", h.ev.calls, h.fences, h.drains, h.holds)
372 }
373 for _, loc := range h.locations {
374 if loc != cutRoot {
375 t.Fatalf("the lease step read %s, not the activator's lease location %s", loc, cutRoot)
376 }
377 }
378}
379
380// Ruling T5-lease: Phase 1 cuts only a holder whose location is the
381// activator's own lease prefix, activator.HibernatePrefix(prefix, host),
382// because the final checkpoint and the stop evidence take that very lease
383// object. A scoped holder (instances/<host>/<incarnation>) goes back to
384// ready at requested and at admitted, with the reason recorded, and
385// nothing stops. The refusal stands: a passing round pins again and the
386// next cut is refused the same way.
387func TestCutRefusesAScopedHolder(t *testing.T) {
388 t.Parallel()
389 ctx := context.Background()
390 check := func(t *testing.T, h *cutHarness, when string) {
391 t.Helper()
392 m := h.move(t)
393 if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.Reason != provision.RefusedReplicaLayout {
394 t.Fatalf("%s: not refused back to ready with the reason: %+v", when, m)
395 }
396 if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 || h.checkpoints != 0 {
397 t.Fatalf("%s: something stopped: evidence %d, holds %v, fences %d, checkpoints %d", when, h.ev.calls, h.holds, h.fences, h.checkpoints)
398 }
399 if c := h.claim(t); c.Op.Exec != "ie-box" || c.Op.Kind != hostclaim.KindMoving {
400 t.Fatalf("%s: the claim moved: %+v", when, c.Op)
401 }
402 }
403 h := newCutHarness(t)
404 h.scoped = true
405 h.ready(t)
406 if loc := h.claim(t).Holder.Location; loc != "instances/"+p1Host+"/"+h.inc {
407 t.Fatalf("fixture: holder at %s", loc)
408 }
409 h.cut(t)
410 h.pass(t)
411 check(t, h, "at requested")
412 // Fix round 2: the refusal stands and nothing loops. decideReady never
413 // repins a scoped holder, so later passes write nothing, and the door
414 // refuses a cut with no pin.
415 refused := h.move(t)
416 h.pass(t)
417 h.pass(t)
418 if m := h.move(t); m.State != provision.MoveReady || m.Inputs != nil || m.Reason != provision.RefusedReplicaLayout || m.UpdatedAt != refused.UpdatedAt {
419 t.Fatalf("passes after the refusal wrote the record: %+v", m)
420 }
421 g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
422 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); !errors.Is(err, provision.ErrMoveConflict) {
423 t.Fatalf("a cut of the refused move: %v", err)
424 }
425
426 h2 := newCutHarness(t)
427 h2.scoped = true
428 h2.ready(t)
429 h2.cut(t)
430 now := h2.clock.Now()
431 if _, err := provision.SetRecordMove(ctx, h2.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
432 t.Fatal(err)
433 }
434 h2.pass(t)
435 check(t, h2, "at admitted")
436}
437
438// Ruling T5-lease: the cut's final checkpoint takes the activator's own
439// lease object, with the real lease package against a fake S3. While the
440// activator's writer holds it the cut cannot checkpoint: execution stays
441// with the source, no report is published and nothing but the lock is
442// written. The production hook refuses a holder at any other location
443// before it asks for a lease at all.
444func TestCutCheckpointWaitsForTheActivatorsLease(t *testing.T) {
445 ctx := context.Background()
446 fake := leasetest.NewVersionedBucket()
447 fake.AddBucket("data-ie")
448 handler, requests := leasetest.NewS3Server(fake, "data-ie")
449 srv := httptest.NewServer(handler) // 127.0.0.1 only
450 defer srv.Close()
451 t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
452 t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
453 t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
454 t.Setenv("AWS_SESSION_TOKEN", "")
455 t.Setenv("AWS_PROFILE", "")
456 t.Setenv("AWS_REGION", "")
457
458 h := newCutHarness(t)
459 h.cfg.FinalCheckpoint = finalCheckpoint("data-ie", "instances", "eu-west-1")
460 h.ready(t)
461 // The activator's writer, as ExecBackend.Acquire takes it.
462 writer, err := lease.Acquire(ctx, "data-ie", activator.HibernatePrefix("instances", p1Host), "eu-west-1")
463 if err != nil {
464 t.Fatal(err)
465 }
466 defer writer.Release(ctx)
467 h.cut(t)
468 h.pass(t)
469 m := h.move(t)
470 if m.State != provision.MoveCutting || m.CutPhase != provision.CutStopped {
471 t.Fatalf("with the writer's lease held: %+v", m)
472 }
473 if exec := h.claim(t).Op.Exec; exec != "ie-box" {
474 t.Fatalf("execution left the source: %q", exec)
475 }
476 if _, _, err := h.s.Get(ctx, h.stoppedKey()); !errors.Is(err, os.ErrNotExist) {
477 t.Fatalf("a stopped report was published: %v", err)
478 }
479 vers, err := fake.ListVersions(ctx, "data-ie", "")
480 if err != nil {
481 t.Fatal(err)
482 }
483 for _, v := range vers {
484 if v.Key != activator.HibernatePrefix("instances", p1Host)+"/lock.json" {
485 t.Fatalf("the cut wrote %s while the writer held the lease", v.Key)
486 }
487 }
488 meta := filepath.Join(filepath.Dir(h.route.DBPath), ".activator-"+filepath.Base(h.route.DBPath)+"-litestream")
489 if _, err := os.Stat(meta); !errors.Is(err, os.ErrNotExist) {
490 t.Fatalf("a checkpoint ran: %s exists (%v)", meta, err)
491 }
492 holder := *h.claim(t).Holder
493 var he *lease.HeldError
494 if err := h.cfg.FinalCheckpoint(ctx, p1Host, h.route.DBPath, holder, func(context.Context) error { return nil }); !errors.As(err, &he) {
495 t.Fatalf("the hook against the held lease: %v", err)
496 }
497 // A holder away from the activator's prefix: refused with no request.
498 before := requests()
499 foreign := holder
500 foreign.Location = "instances/" + p1Host + "/" + h.inc
501 err = h.cfg.FinalCheckpoint(ctx, p1Host, h.route.DBPath, foreign, func(context.Context) error { return nil })
502 if !errors.Is(err, errHolderLocation) || requests() != before {
503 t.Fatalf("a scoped holder: %v (%d requests)", err, requests()-before)
504 }
505}
506
507// K3 and K13: the stopped report exists, complete with the replica step
508// and N, before the claim's execution is none. A crash after the
509// checkpoint leaves execution with the source and no report; a crash after
510// the report leaves one report, which the replay reuses as it is.
511func TestStoppedReportIsCompleteBeforeExecToNone(t *testing.T) {
512 t.Parallel()
513 ctx := context.Background()
514 for _, point := range []string{"after-checkpoint", "after-report", "after-record-source", "after-exec-none"} {
515 t.Run(point, func(t *testing.T) {
516 h := newCutHarness(t)
517 h.ready(t)
518 h.crashAt = point
519 h.cut(t)
520 h.crashPass(t)
521 if m := h.move(t); m.CutPhase != provision.CutStopped {
522 t.Fatalf("after the crash: %+v", m)
523 }
524 body, _, err := h.s.Get(ctx, h.stoppedKey())
525 exec := h.claim(t).Op.Exec
526 switch point {
527 case "after-checkpoint":
528 if !errors.Is(err, os.ErrNotExist) || exec != "ie-box" {
529 t.Fatalf("report %v, exec %q", err, exec)
530 }
531 default:
532 var r moves.StoppedReport
533 if err != nil || json.Unmarshal(body, &r) != nil || !r.Complete() || r.FinalTXID != 3 || r.DBSHA256 == "" || r.TimersSHA256 == "" || r.FilesPlain == "" {
534 t.Fatalf("report %v: %+v", err, r)
535 }
536 if want := point == "after-exec-none"; (exec == "") != want {
537 t.Fatalf("exec %q at %s", exec, point)
538 }
539 }
540 h.freshWorker()
541 h.pass(t)
542 if m := h.move(t); m.State != provision.MoveLanding {
543 t.Fatalf("the replay did not finish: %+v", m)
544 }
545 after, _, err := h.s.Get(ctx, h.stoppedKey())
546 if err != nil {
547 t.Fatal(err)
548 }
549 if point != "after-checkpoint" && string(after) != string(body) {
550 t.Fatalf("the replay published a second report:\n%s\n%s", body, after)
551 }
552 keys, _ := h.s.List(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID))
553 n := 0
554 for _, k := range keys {
555 if strings.Contains(k, "stopped-") {
556 n++
557 }
558 }
559 if n != 1 {
560 t.Fatalf("%d stopped reports: %v", n, keys)
561 }
562 src := h.move(t).Source
563 if src == nil || src.FinalTXID != 3 || src.Evidence != h.stoppedKey() || src.Files == nil || src.Files.Plain == "" || src.Files.Sealed == "" {
564 t.Fatalf("source baselines: %+v", src)
565 }
566 c := h.claim(t)
567 if c.Op.Journal[hostclaim.JournalEvidenceSource] != h.stoppedKey() || c.Op.Journal[hostclaim.JournalCutSession] == "" {
568 t.Fatalf("claim journal: %v", c.Op.Journal)
569 }
570 })
571 }
572}
573
574// Review Focus 2: three final rounds that defer revert the cut, and the
575// claim follows once execution is none: reverting under the source, the
576// lane's park queued with the writers' horizon.
577func TestCutRevertsAfterThreeDeferredRounds(t *testing.T) {
578 h := newCutHarness(t)
579 h.ready(t)
580 n := 0
581 h.f.beforeCopy = func(srcKey string) {
582 if strings.HasSuffix(srcKey, lastTXKey) {
583 // The source changes under the copy: the conditional copy is
584 // refused and the object deferred.
585 h.f.bmu.Lock()
586 n++
587 o := h.f.buckets["data-ie"][srcKey]
588 o.ETag = "changed-" + string(rune('a'+n))
589 h.f.buckets["data-ie"][srcKey] = o
590 h.f.bmu.Unlock()
591 }
592 }
593 h.cut(t)
594 for i := 1; i <= 2; i++ {
595 h.pass(t)
596 if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopped {
597 t.Fatalf("after deferred round %d: %+v", i, m)
598 }
599 if got := h.ledger(t).DeferredRounds; got != i {
600 t.Fatalf("ledger counts %d deferred rounds, want %d", got, i)
601 }
602 }
603 h.pass(t)
604 m := h.move(t)
605 if m.State != provision.MoveReverting || m.Reason != provision.RevertDeferredRounds {
606 t.Fatalf("after three deferred rounds: %+v", m)
607 }
608 c := h.claim(t)
609 if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalRevPhase] != "evidence" {
610 t.Fatalf("claim: %+v", c.Op)
611 }
612 var park *hostclaim.ParkScope
613 for _, o := range c.Obligations {
614 if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" {
615 park = o.Park
616 }
617 }
618 if park == nil || park.Location != moves.ScopeLocation(p1Host, p1ID) || park.Incarnation != h.inc || park.RequestLimit != cutRequestLimit {
619 t.Fatalf("staging park: %+v", park)
620 }
621 if !park.EvidenceAt.After(h.clock.now.Add(moves.SignatureLife)) {
622 t.Fatalf("evidence_at %v is not past every signature's life", park.EvidenceAt)
623 }
624 // A replay changes nothing: the claim is reverting already.
625 h.pass(t)
626 if c2 := h.claim(t); c2.Gen != c.Gen {
627 t.Fatalf("a replay wrote the claim again (%d to %d)", c.Gen, c2.Gen)
628 }
629}
630
631// §7.3 seal by version: the seal names N and the exact version of every
632// current replica object, of the timers and of both archive objects. The
633// witness (wSealByVersion): a stale copy that lands after the seal makes a
634// version the seal does not name, and the seal still names the earlier one.
635func TestSealNamesEveryCurrentVersionAndN(t *testing.T) {
636 h := newCutHarness(t)
637 h.ready(t)
638 h.cut(t)
639 h.pass(t)
640 if m := h.move(t); m.State != provision.MoveLanding {
641 t.Fatalf("cut did not finish: %+v", m)
642 }
643 seal := h.seal(t)
644 if seal.N != 3 || !seal.ReachesN() || !seal.FromDB || !seal.SrcCovered || len(seal.Timers) != 1 || len(seal.Files) != 2 {
645 t.Fatalf("seal: %+v", seal)
646 }
647 lane := moves.ReplicaPrefix(p1Host, p1ID)
648 want := []string{lane + "0000/0000000000000001-0000000000000002.ltx", lane + "0001/0000000000000001-0000000000000002.ltx", lane + lastTXKey}
649 sort.Strings(want)
650 var got []string
651 for _, o := range seal.Replica {
652 got = append(got, o.Key)
653 if cur := h.f.versionOf("data-uk", o.Key); cur != o.VersionID {
654 t.Fatalf("%s: sealed %s, current %s", o.Key, o.VersionID, cur)
655 }
656 }
657 sort.Strings(got)
658 if strings.Join(got, ",") != strings.Join(want, ",") {
659 t.Fatalf("sealed %v, want %v", got, want)
660 }
661 for _, o := range append(append([]moves.SealedObject{}, seal.Timers...), seal.Files...) {
662 if cur := h.f.versionOf("data-uk", o.Key); cur != o.VersionID || !lease.UsableVersionID(o.VersionID) {
663 t.Fatalf("%s: sealed %q, current %q", o.Key, o.VersionID, cur)
664 }
665 }
666 if !seal.Bound().After(seal.SealedAt) {
667 t.Fatalf("bound %v is not after the seal %v", seal.Bound(), seal.SealedAt)
668 }
669 // The witness.
670 stale := seal.Replica[0]
671 if _, _, err := h.f.CopyVersionedAt(context.Background(), h.clock.Now(), "data-ie",
672 cutLineage+"0000/0000000000000001-0000000000000002.ltx", "", "data-uk", stale.Key); err != nil {
673 t.Fatal(err)
674 }
675 if cur := h.f.versionOf("data-uk", stale.Key); cur == stale.VersionID {
676 t.Fatal("the stale copy did not make a new version")
677 }
678 if again := h.seal(t); again.Replica[0].VersionID != stale.VersionID {
679 t.Fatal("the seal moved with the stale copy")
680 }
681 // Every staging write of the cut went through the guard.
682 for _, op := range h.f.ops {
683 if strings.Contains(op, "data-uk/") && !strings.HasPrefix(op, "copy ") {
684 t.Fatalf("an unguarded write reached the target bucket: %s", op)
685 }
686 }
687 for _, c := range h.f.stagingCalls() {
688 if c.SignAt.IsZero() || c.Bucket != "data-uk" {
689 t.Fatalf("unguarded or misdirected request: %+v", c)
690 }
691 }
692}
693
694// writeOnStop makes the instance write a file just before it is stopped:
695// the stopped files differ from the archive the target verified, so the
696// cut builds its own.
697func (h *cutHarness) writeOnStop(t *testing.T) {
698 h.ev.onCall = func() {
699 if h.ev.calls == 1 {
700 if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "late.txt"), []byte("written before the stop"), 0o644); err != nil {
701 t.Error(err)
702 }
703 }
704 }
705}
706
707func snapshotSealed(h *cutHarness) []string {
708 var out []string
709 for k := range h.f.buckets[h.srcBucket] {
710 if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) && strings.HasSuffix(k, ".sealed") {
711 out = append(out, k)
712 }
713 }
714 sort.Strings(out)
715 return out
716}
717
718// codex 19, with the real randomized sealer: the archive's identity is
719// journalled before any PUT, and every replay publishes that same key,
720// whatever it crashed after: the journal entry, the first snapshot PUT,
721// the second, the first copy. Once journalled the identity is final
722// (fix round 1, ruling 4): a replay with the sealed file gone finishes it
723// from what was published (the manifest from the journalled bytes if only
724// the sealed object landed), and one that finds nothing published cannot,
725// so the cut reverts rather than reseal under a new name.
726func TestCutArchiveIsCreateOnlyAndReused(t *testing.T) {
727 t.Parallel()
728 for _, point := range []string{"archive-journaled", "archive-sealed-snapshot", "archive-own-put", "archive-first-copy"} {
729 t.Run("crash "+point, func(t *testing.T) {
730 h := newCutHarness(t)
731 h.ready(t)
732 h.writeOnStop(t)
733 h.crashAt = point
734 h.cut(t)
735 h.crashPass(t)
736 if m := h.move(t); m.CutPhase != provision.CutStaged {
737 t.Fatalf("after the crash: %+v", m)
738 }
739 id := h.journalArchive(t)
740 if id.Reuse {
741 t.Fatal("the stopped files changed, yet the preparing archive was reused")
742 }
743 h.freshWorker()
744 h.pass(t)
745 m := h.move(t)
746 if m.State != provision.MoveLanding || m.Source.Files.Sealed != id.Sealed {
747 t.Fatalf("the replay: %+v, files %+v, journalled %s", m, m.Source.Files, id.Sealed)
748 }
749 if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], id.Name+".sealed") {
750 t.Fatalf("snapshots %v, want only %s", got, id.Name)
751 }
752 if !sealNamesArchive(h.seal(t), id.Sealed) {
753 t.Fatalf("the seal names %+v, not %s", h.seal(t).Files, id.Name)
754 }
755 })
756 }
757 t.Run("file gone, nothing published", func(t *testing.T) {
758 h := newCutHarness(t)
759 h.ready(t)
760 h.writeOnStop(t)
761 h.crashAt = "archive-journaled"
762 h.cut(t)
763 h.crashPass(t)
764 first := h.journalArchive(t)
765 if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
766 t.Fatal(err)
767 }
768 h.pass(t)
769 m := h.move(t)
770 if m.State != provision.MoveReverting || m.Reason != provision.RevertArchiveRefused {
771 t.Fatalf("an identity nobody can finish did not revert: %+v", m)
772 }
773 if again := h.journalArchive(t); again.Sealed != first.Sealed {
774 t.Fatalf("the journalled identity changed from %s to %s", first.Name, again.Name)
775 }
776 if got := snapshotSealed(h); len(got) != 0 {
777 t.Fatalf("snapshots %v, want none", got)
778 }
779 if h.seal(t).HasFiles() {
780 t.Fatal("the seal names files")
781 }
782 })
783 t.Run("file gone, sealed snapshot only", func(t *testing.T) {
784 h := newCutHarness(t)
785 h.ready(t)
786 h.writeOnStop(t)
787 h.crashAt = "archive-sealed-snapshot"
788 h.cut(t)
789 h.crashPass(t)
790 first := h.journalArchive(t)
791 snap := moves.SnapshotsPrefix(p1Host, p1ID)
792 if _, ok := h.f.buckets["data-ie"][snap+first.Name+".sealed"]; !ok {
793 t.Fatal("fixture: the sealed object did not land")
794 }
795 if _, ok := h.f.buckets["data-ie"][snap+first.Name+".json"]; ok {
796 t.Fatal("fixture: the manifest landed too")
797 }
798 if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
799 t.Fatal(err)
800 }
801 h.freshWorker()
802 h.pass(t)
803 m := h.move(t)
804 if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
805 t.Fatalf("the identity was not finished: %+v files %+v", m, m.Source.Files)
806 }
807 manifest, ok := h.f.buckets["data-ie"][snap+first.Name+".json"]
808 if !ok {
809 t.Fatal("the manifest was not written from the journal")
810 }
811 var man archive.Manifest
812 if err := json.Unmarshal([]byte(manifest.body), &man); err != nil || man.Sealed != first.Sealed || man.Plain != first.Plain {
813 t.Fatalf("the written manifest: %v %+v", err, man)
814 }
815 copied := 0
816 for _, c := range h.f.stagingCalls() {
817 if c.Op == "copy" && strings.HasPrefix(c.Key, moves.FilesPrefix(p1Host, p1ID)+first.Name) {
818 copied++
819 }
820 }
821 if copied != 2 || len(snapshotSealed(h)) != 1 {
822 t.Fatalf("%d copies from the snapshot, %d sealed snapshots", copied, len(snapshotSealed(h)))
823 }
824 })
825 t.Run("file gone, snapshot published", func(t *testing.T) {
826 h := newCutHarness(t)
827 h.ready(t)
828 h.writeOnStop(t)
829 h.crashAt = "archive-own-put"
830 h.cut(t)
831 h.crashPass(t)
832 first := h.journalArchive(t)
833 if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
834 t.Fatal(err)
835 }
836 h.pass(t)
837 m := h.move(t)
838 if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
839 t.Fatalf("not reused: %+v files %+v", m, m.Source.Files)
840 }
841 // Staged by a server-side copy from the snapshot.
842 copied := 0
843 for _, c := range h.f.stagingCalls() {
844 if c.Op == "copy" && strings.HasPrefix(c.Key, moves.FilesPrefix(p1Host, p1ID)+first.Name) {
845 copied++
846 }
847 }
848 if copied != 2 {
849 t.Fatalf("%d copies from the snapshot, want 2", copied)
850 }
851 })
852 t.Run("unchanged files reuse the verified archive", func(t *testing.T) {
853 h := newCutHarness(t)
854 h.ready(t)
855 staged := h.ledger(t)
856 h.cut(t)
857 h.pass(t)
858 m := h.move(t)
859 if m.State != provision.MoveLanding || m.Source.Files.Sealed != staged.ArchiveSealed {
860 t.Fatalf("the verified archive was not reused: %+v (ledger %s)", m.Source.Files, staged.ArchiveSealed)
861 }
862 seal := h.seal(t)
863 for _, o := range seal.Files {
864 if o.VersionID != staged.ArchiveVersions.SealedVersion && o.VersionID != staged.ArchiveVersions.ManifestVersion {
865 t.Fatalf("the seal names %s@%s, not the verified versions", o.Key, o.VersionID)
866 }
867 }
868 if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], archiveName(staged.ArchiveSealed)+".sealed") {
869 t.Fatalf("snapshots %v", got)
870 }
871 })
872}
873
874// G8: archived is written only after the seal names the archive. A crash
875// after each of the archive's PUTs leaves the record staged; the replay
876// completes and the seal names both files/ versions before archived. A
877// record hand-edited to archived whose seal has no files does not flip.
878func TestArchivedNeverPrecedesTheSealedArchive(t *testing.T) {
879 t.Parallel()
880 for _, point := range []string{"archive-own-put", "archive-first-copy", "archive-second-copy", "after-archive-seal"} {
881 t.Run(point, func(t *testing.T) {
882 h := newCutHarness(t)
883 h.ready(t)
884 h.writeOnStop(t)
885 h.crashAt = point
886 h.cut(t)
887 h.crashPass(t)
888 if m := h.move(t); m.CutPhase != provision.CutStaged {
889 t.Fatalf("archived before the seal named the archive: %+v", m)
890 }
891 if point != "after-archive-seal" && h.seal(t).HasFiles() {
892 t.Fatal("the seal names files before the copies finished")
893 }
894 h.pass(t)
895 m := h.move(t)
896 seal := h.seal(t)
897 if m.State != provision.MoveLanding || !sealNamesArchive(seal, m.Source.Files.Sealed) {
898 t.Fatalf("after the replay: %+v, seal files %+v", m, seal.Files)
899 }
900 for _, o := range seal.Files {
901 if !strings.HasPrefix(o.Key, moves.FilesPrefix(p1Host, p1ID)) || h.f.versionOf("data-uk", o.Key) != o.VersionID {
902 t.Fatalf("seal names %s@%s", o.Key, o.VersionID)
903 }
904 }
905 })
906 }
907 t.Run("hand-edited to archived", func(t *testing.T) {
908 ctx := context.Background()
909 h := newCutHarness(t)
910 h.ready(t)
911 h.crashAt = "archive-journaled"
912 h.cut(t)
913 h.crashPass(t)
914 if h.seal(t).HasFiles() {
915 t.Fatal("fixture: the seal already names files")
916 }
917 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
918 rec.Move.CutPhase = provision.CutArchived
919 return nil
920 }); err != nil {
921 t.Fatal(err)
922 }
923 h.pass(t)
924 h.pass(t)
925 m := h.move(t)
926 if m.State != provision.MoveCutting || m.CutPhase != provision.CutArchived || h.read(t, p1Host).Region != "ie" {
927 t.Fatalf("a hand-edited archived flipped: %+v", m)
928 }
929 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving {
930 t.Fatalf("the claim flipped: %+v", c.Op)
931 }
932 })
933}
934
935// §7.5: the record flips first (region, landing, the landing deadline,
936// checking), then the claim follows: landing/checking with the target
937// executing, the source's stamp and generation journalled, the holder's
938// generation untouched. A crash between the two leaves the record ahead
939// and the claim moving/sealed with nobody executing, which anyone who sees
940// it may finish with the same Flip.
941func TestFlipWritesRecordThenClaim(t *testing.T) {
942 t.Parallel()
943 ctx := context.Background()
944 h := newCutHarness(t)
945 h.ready(t)
946 h.cut(t)
947 h.pass(t)
948 rec := h.read(t, p1Host)
949 m := rec.Move
950 if rec.Region != "uk" || m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped || m.LandingPhase != provision.LandChecking || m.FlippedAt == nil || m.LandingDeadline == nil {
951 t.Fatalf("record: region %s %+v", rec.Region, m)
952 }
953 if d := m.LandingDeadline.Sub(*m.FlippedAt); d != cutLandingWindow {
954 t.Fatalf("landing window %v", d)
955 }
956 c := h.claim(t)
957 if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != "checking" || c.Op.Exec != "uk-box" ||
958 c.Op.Journal[hostclaim.JournalSourceStamp] != "1" || c.Op.Journal[hostclaim.JournalSrcGen] != "cur" || c.Holder.Generation != "cur" {
959 t.Fatalf("claim: %+v holder %+v", c.Op, c.Holder)
960 }
961
962 h2 := newCutHarness(t)
963 h2.ready(t)
964 h2.crashAt = "after-record-flip"
965 h2.cut(t)
966 h2.crashPass(t)
967 if r := h2.read(t, p1Host); r.Region != "uk" || r.Move.State != provision.MoveLanding {
968 t.Fatalf("record not flipped: %+v", r.Move)
969 }
970 c2 := h2.claim(t)
971 if c2.Op.Kind != hostclaim.KindMoving || c2.Op.Phase != hostclaim.PhaseSealed || c2.Op.Exec != "" {
972 t.Fatalf("claim after the crash: %+v", c2.Op)
973 }
974 obs, err := hostclaim.Read(ctx, h2.s, p1Host)
975 if err != nil {
976 t.Fatal(err)
977 }
978 if _, err := hostclaim.Update(ctx, h2.s, obs, hostclaim.Flip(p1ID, "uk-box", obs.Claim.Holder.Stamp, obs.Claim.Holder.Generation)); err != nil {
979 t.Fatalf("the recovery's flip: %v", err)
980 }
981 if c3 := h2.claim(t); c3.Op.Kind != hostclaim.KindLanding || c3.Op.Exec != "uk-box" {
982 t.Fatalf("claim after the recovery: %+v", c3.Op)
983 }
984}
985
986// The deadline (§7.4): at admitted it goes back to ready; at stopping it
987// reverts with "stop evidence budget" and the claim stays with the source,
988// which still executes; after execution was given up it reverts with the
989// deadline's reason and the claim follows to reverting.
990func TestCutDeadlinePassedReverts(t *testing.T) {
991 t.Parallel()
992 ctx := context.Background()
993 g := provision.MoveGuard{ID: p1ID, Incarnation: h0inc(t)}
994
995 t.Run("admitted", func(t *testing.T) {
996 h := newCutHarness(t)
997 h.ready(t)
998 h.cut(t)
999 now := h.clock.Now()
1000 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
1001 t.Fatal(err)
1002 }
1003 h.clock.Advance(2 * time.Minute)
1004 h.pass(t)
1005 if m := h.move(t); m.State != provision.MoveReady || h.ev.calls != 0 {
1006 t.Fatalf("admitted past the deadline: %+v", m)
1007 }
1008 })
1009 t.Run("stopping", func(t *testing.T) {
1010 h := newCutHarness(t)
1011 h.ready(t)
1012 h.ev.stopped = false
1013 h.cut(t)
1014 h.pass(t)
1015 h.clock.Advance(2 * time.Minute)
1016 h.pass(t)
1017 m := h.move(t)
1018 if m.State != provision.MoveReverting || m.Reason != provision.RevertStopEvidenceBudget {
1019 t.Fatalf("stopping past the deadline: %+v", m)
1020 }
1021 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
1022 t.Fatalf("the claim moved without the source's evidence: %+v", c.Op)
1023 }
1024 })
1025 t.Run("stopped, execution given up", func(t *testing.T) {
1026 h := newCutHarness(t)
1027 h.ready(t)
1028 h.crashAt = "after-exec-none"
1029 h.cut(t)
1030 h.crashPass(t)
1031 h.clock.Advance(2 * time.Minute)
1032 h.pass(t)
1033 m := h.move(t)
1034 if m.State != provision.MoveReverting || m.Reason != provision.RevertCutDeadline {
1035 t.Fatalf("stopped past the deadline: %+v", m)
1036 }
1037 c := h.claim(t)
1038 if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" {
1039 t.Fatalf("claim: %+v", c.Op)
1040 }
1041 j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
1042 if err != nil || j.Steps[cutStepRevertBound] == "" {
1043 t.Fatalf("the revert's bound is not journalled: %v %+v", err, j)
1044 }
1045 })
1046 t.Run("the claim half after a crash", func(t *testing.T) {
1047 h := newCutHarness(t)
1048 h.ready(t)
1049 h.crashAt = "after-exec-none"
1050 h.cut(t)
1051 h.crashPass(t)
1052 // The record decides the revert; the claim half is lost.
1053 if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveRevert(g, provision.RevertCutDeadline)); err != nil {
1054 t.Fatal(err)
1055 }
1056 if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving {
1057 t.Fatalf("fixture: %+v", c.Op)
1058 }
1059 h.pass(t)
1060 if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting {
1061 t.Fatalf("the reverting arm did not write the claim half: %+v", c.Op)
1062 }
1063 })
1064}
1065
1066// With the switch off, a cutting record is never touched.
1067func TestCutIsInertWithTheSwitchOff(t *testing.T) {
1068 h := newCutHarness(t)
1069 h.ready(t)
1070 h.cut(t)
1071 before := h.read(t, p1Host)
1072 claimBefore := h.claim(t)
1073 h.cfg.Enabled = false
1074 h.pass(t)
1075 h.pass(t)
1076 if after := h.read(t, p1Host); after.Move.CutPhase != before.Move.CutPhase || after.Move.UpdatedAt != before.Move.UpdatedAt {
1077 t.Fatalf("the record moved: %+v", after.Move)
1078 }
1079 if c := h.claim(t); c.Gen != claimBefore.Gen {
1080 t.Fatal("the claim was written")
1081 }
1082 if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 {
1083 t.Fatal("the cut ran with the switch off")
1084 }
1085}
1086
1087func h0inc(t *testing.T) string {
1088 t.Helper()
1089 inc, err := hostclaim.Incarnation("bab", "app", p1Host, p1Created)
1090 if err != nil {
1091 t.Fatal(err)
1092 }
1093 return inc
1094}
1095
1096// nullVersions is a staging writer whose copies of matching keys answer
1097// "null", as a versioning-suspended bucket does.
1098type nullVersions struct {
1099 lease.StagingWriter
1100 match func(key string) bool
1101}
1102
1103func (n nullVersions) CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) {
1104 etag, ver, err := n.StagingWriter.CopyVersionedAt(ctx, signAt, srcBucket, srcKey, srcETag, dstBucket, dstKey)
1105 if err == nil && n.match(dstKey) {
1106 ver = "null"
1107 }
1108 return etag, ver, err
1109}
1110
1111// G27: the seal needs a usable version for every current replica key. A
1112// final round that copied one without (a "null") seals nothing and runs
1113// again; that is not a deferred round.
1114func TestSealRefusesAnObjectWithoutAUsableVersion(t *testing.T) {
1115 h := newCutHarness(t)
1116 h.ready(t)
1117 suspended := true
1118 h.cfg.Staging = func(context.Context, string) (lease.StagingWriter, error) {
1119 return nullVersions{StagingWriter: h.f, match: func(key string) bool { return suspended && strings.HasSuffix(key, lastTXKey) }}, nil
1120 }
1121 h.freshWorker() // the worker memoizes its staging writer
1122 h.cut(t)
1123 h.pass(t)
1124 if m := h.move(t); m.CutPhase != provision.CutStopped {
1125 t.Fatalf("sealed over a null version: %+v", m)
1126 }
1127 if _, ok := h.claim(t).Op.Journal[hostclaim.JournalSeal]; ok {
1128 t.Fatal("a seal was journalled")
1129 }
1130 if n := h.ledger(t).DeferredRounds; n != 0 {
1131 t.Fatalf("counted as %d deferred rounds", n)
1132 }
1133 suspended = false
1134 h.pass(t)
1135 if m := h.move(t); m.State != provision.MoveLanding {
1136 t.Fatalf("the next round did not seal: %+v", m)
1137 }
1138 for _, o := range h.seal(t).Replica {
1139 if !lease.UsableVersionID(o.VersionID) {
1140 t.Fatalf("sealed %s@%q", o.Key, o.VersionID)
1141 }
1142 }
1143}
1144
1145// holdMoving holds the row moving under the op, once, with the
1146// stage-requested marker touched for a write that landed; a row already so
1147// held is left alone, and a row another op holds is refused.
1148func TestHoldMovingIsConditional(t *testing.T) {
1149 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
1150 if err != nil {
1151 t.Fatal(err)
1152 }
1153 defer reg.Close()
1154 if err := reg.Upsert(registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true}); err != nil {
1155 t.Fatal(err)
1156 }
1157 var written []string
1158 hold := holdMoving(reg, func(host string) { written = append(written, host) })
1159 if err := hold(p1Host, p1ID); err != nil {
1160 t.Fatal(err)
1161 }
1162 if err := hold(p1Host, p1ID); err != nil {
1163 t.Fatalf("a second hold under the same op: %v", err)
1164 }
1165 rt, err := reg.Get(p1Host)
1166 if err != nil || rt.Status != "moving" || rt.HoldOp != p1ID || len(written) != 1 {
1167 t.Fatalf("row %+v, %v; marker touched %d times", rt, err, len(written))
1168 }
1169 if err := hold(p1Host, p1ID2); err == nil {
1170 t.Fatal("another op's hold was taken over")
1171 }
1172 if rt, _ := reg.Get(p1Host); rt.HoldOp != p1ID {
1173 t.Fatalf("the hold moved to %s", rt.HoldOp)
1174 }
1175 if err := hold("absent.example.com", p1ID); err == nil {
1176 t.Fatal("a missing row was held")
1177 }
1178}
1179
1180// localLTXPosition reads the newest level-0 file of the activator's meta
1181// directory (lease.Held's SetMetaPath); no directory is position 0.
1182func TestLocalLTXPosition(t *testing.T) {
1183 dir := t.TempDir()
1184 db := filepath.Join(dir, p1Host+".db")
1185 if n, err := localLTXPosition(db); err != nil || n != 0 {
1186 t.Fatalf("no meta directory: %d, %v", n, err)
1187 }
1188 l0 := filepath.Join(dir, ".activator-"+p1Host+".db-litestream", "ltx", "0")
1189 if err := os.MkdirAll(l0, 0o755); err != nil {
1190 t.Fatal(err)
1191 }
1192 for _, name := range []string{"0000000000000001-0000000000000005.ltx", "0000000000000006-000000000000000a.ltx", "junk", "0000000000000003-0000000000000004.ltx"} {
1193 if err := os.WriteFile(filepath.Join(l0, name), nil, 0o600); err != nil {
1194 t.Fatal(err)
1195 }
1196 }
1197 if n, err := localLTXPosition(db); err != nil || n != 10 {
1198 t.Fatalf("position %d, %v; want 10", n, err)
1199 }
1200}
1201
1202// Ruling 3: coverage is never assumed. A database that is gone before the
1203// final checkpoint leaves the seal FromDB false and SrcCovered false, since
1204// nothing proved the replica holds its last write; one that was there
1205// through the checkpoint, with its position journalled equal to N, keeps
1206// SrcCovered true if the disk is lost afterwards; one still present but
1207// behind its replica is not covered.
1208func TestMissingDatabaseIsNotCovered(t *testing.T) {
1209 t.Parallel()
1210 t.Run("lost before the checkpoint", func(t *testing.T) {
1211 h := newCutHarness(t)
1212 h.ready(t)
1213 h.crashAt = "after-evidence"
1214 h.cut(t)
1215 h.crashPass(t)
1216 if m := h.move(t); m.CutPhase != provision.CutStopping {
1217 t.Fatalf("after the crash: %+v", m)
1218 }
1219 if err := os.Remove(h.route.DBPath); err != nil {
1220 t.Fatal(err)
1221 }
1222 h.freshWorker()
1223 h.pass(t)
1224 m := h.move(t)
1225 if m.State != provision.MoveLanding || h.checkpoints != 0 {
1226 t.Fatalf("the cut without a database: %+v (%d checkpoints)", m, h.checkpoints)
1227 }
1228 if m.Source.FinalTXID != 2 || m.Source.DBSHA256 != "" {
1229 t.Fatalf("source baselines: %+v", m.Source)
1230 }
1231 seal := h.seal(t)
1232 if seal.FromDB || seal.SrcCovered || seal.N != 2 {
1233 t.Fatalf("seal claims coverage with no database: %+v", seal)
1234 }
1235 })
1236 t.Run("lost after the checkpoint journalled coverage", func(t *testing.T) {
1237 h := newCutHarness(t)
1238 h.ready(t)
1239 h.crashAt = "after-exec-none"
1240 h.cut(t)
1241 h.crashPass(t)
1242 if err := os.Remove(h.route.DBPath); err != nil {
1243 t.Fatal(err)
1244 }
1245 h.freshWorker()
1246 h.pass(t)
1247 if m := h.move(t); m.State != provision.MoveLanding {
1248 t.Fatalf("the replay: %+v", m)
1249 }
1250 seal := h.seal(t)
1251 if seal.FromDB || !seal.SrcCovered || seal.N != 3 {
1252 t.Fatalf("seal: %+v", seal)
1253 }
1254 })
1255 t.Run("present but behind its replica", func(t *testing.T) {
1256 h := newCutHarness(t)
1257 h.ready(t)
1258 h.localPos = 2
1259 h.cut(t)
1260 h.pass(t)
1261 if m := h.move(t); m.State != provision.MoveLanding {
1262 t.Fatalf("the cut: %+v", m)
1263 }
1264 if seal := h.seal(t); !seal.FromDB || seal.SrcCovered {
1265 t.Fatalf("seal: %+v", seal)
1266 }
1267 })
1268}
1269
1270// regularFile: a regular file is one; an absent path is not and no error;
1271// anything else (a path under a file, a directory where a file should be)
1272// is an error, never absence.
1273func TestRegularFileDistinguishesAbsence(t *testing.T) {
1274 dir := t.TempDir()
1275 file := filepath.Join(dir, "a.db")
1276 if err := os.WriteFile(file, []byte("x"), 0o600); err != nil {
1277 t.Fatal(err)
1278 }
1279 if ok, err := regularFile(file); !ok || err != nil {
1280 t.Fatalf("a file: %v %v", ok, err)
1281 }
1282 if ok, err := regularFile(filepath.Join(dir, "absent.db")); ok || err != nil {
1283 t.Fatalf("absent: %v %v", ok, err)
1284 }
1285 if ok, err := regularFile(filepath.Join(file, "under-a-file.db")); ok || err == nil {
1286 t.Fatalf("a path under a file: %v %v", ok, err)
1287 }
1288 if ok, err := regularFile(dir); ok || err == nil {
1289 t.Fatalf("a directory: %v %v", ok, err)
1290 }
1291}
1292
1293// Ruling 4: a listing error is an error, never absence. With the sealed
1294// file gone, a replay whose ListVersions fails changes nothing: the record
1295// stays staged, the journalled identity stands, nothing is rebuilt and
1296// nothing reverts. The next replay, listing again, finishes the identity
1297// when something was published, and reverts when nothing was.
1298func TestArchiveListingFailureIsNotAbsence(t *testing.T) {
1299 t.Parallel()
1300 for _, tc := range []struct {
1301 name, crash string
1302 then string
1303 }{
1304 {"the sealed snapshot landed", "archive-sealed-snapshot", provision.MoveLanding},
1305 {"nothing landed", "archive-journaled", provision.MoveReverting},
1306 } {
1307 t.Run(tc.name, func(t *testing.T) {
1308 h := newCutHarness(t)
1309 h.ready(t)
1310 h.writeOnStop(t)
1311 h.crashAt = tc.crash
1312 h.cut(t)
1313 h.crashPass(t)
1314 first := h.journalArchive(t)
1315 if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
1316 t.Fatal(err)
1317 }
1318 failed := 0
1319 h.vs.deny = func(op, _, key string) error {
1320 if op == "list-versions" && strings.Contains(key, moves.SnapshotsPrefix(p1Host, p1ID)) {
1321 failed++
1322 return errors.New("injected listing failure")
1323 }
1324 return nil
1325 }
1326 h.freshWorker()
1327 h.pass(t)
1328 if failed == 0 {
1329 t.Fatal("the replay never listed the snapshots")
1330 }
1331 m := h.move(t)
1332 if m.State != provision.MoveCutting || m.CutPhase != provision.CutStaged {
1333 t.Fatalf("after the listing failure: %+v", m)
1334 }
1335 if again := h.journalArchive(t); again.Sealed != first.Sealed {
1336 t.Fatalf("the listing failure changed the identity from %s to %s", first.Name, again.Name)
1337 }
1338 if h.seal(t).HasFiles() {
1339 t.Fatal("the seal names files")
1340 }
1341 h.vs.deny = nil
1342 h.freshWorker()
1343 h.pass(t)
1344 m = h.move(t)
1345 if m.State != tc.then {
1346 t.Fatalf("after the listing succeeded: %+v", m)
1347 }
1348 if tc.then == provision.MoveLanding && (m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed)) {
1349 t.Fatalf("not the journalled identity: %+v", m.Source.Files)
1350 }
1351 })
1352 }
1353}
1354
1355// Ruling 5: the archive seal's bound goes into the ledger as well as the
1356// seal, and a ledger that cannot be read or written stops the phase. An
1357// injected ledger-read failure after the archive is sealed leaves the
1358// record staged and the ledger's bound as the stopped seal wrote it; the
1359// next pass finishes and the ledger carries the archive seal's bound.
1360func TestArchiveSealStopsOnALedgerFailure(t *testing.T) {
1361 h := newCutHarness(t)
1362 h.ready(t)
1363 ledgerKey := moves.LedgerKey(p1Host, p1ID)
1364 armed := false
1365 h.cfg.cutFault = func(point string) error {
1366 if point == "after-archive-seal" && !armed {
1367 armed = true
1368 h.f.failGet = func(_, key string) error {
1369 if key == ledgerKey {
1370 return errors.New("injected ledger read failure")
1371 }
1372 return nil
1373 }
1374 }
1375 return nil
1376 }
1377 h.cut(t)
1378 h.pass(t)
1379 if !armed {
1380 t.Fatal("the pass never sealed the archive")
1381 }
1382 m := h.move(t)
1383 if m.State != provision.MoveCutting || m.CutPhase != provision.CutStaged {
1384 t.Fatalf("the phase advanced over a ledger failure: %+v", m)
1385 }
1386 sealBound := h.seal(t).LeaseUntil
1387 h.f.failGet = nil
1388 // The ledger still carries the stopped seal's bound, written before the
1389 // archive: never later than the archive seal's.
1390 before := h.ledger(t)
1391 if before.LeaseUntil.IsZero() || before.LeaseUntil.After(sealBound) {
1392 t.Fatalf("ledger bound %v against the archive seal's %v", before.LeaseUntil, sealBound)
1393 }
1394 h.freshWorker()
1395 h.pass(t)
1396 if m := h.move(t); m.State != provision.MoveLanding {
1397 t.Fatalf("the replay: %+v", m)
1398 }
1399 after := h.ledger(t)
1400 if seal := h.seal(t); !after.LeaseUntil.Equal(seal.LeaseUntil) || after.LeaseUntil.Before(before.LeaseUntil) {
1401 t.Fatalf("ledger bound %v, seal bound %v (was %v)", after.LeaseUntil, seal.LeaseUntil, before.LeaseUntil)
1402 }
1403}
1404
1405// Ruling 6: the crash matrix for the side effects the review found
1406// unexercised. Each point crashes once; the durable state a fresh worker
1407// would find is asserted; the fresh worker replays to the flip, with one
1408// report, a seal naming the archive at the record's N, and the ledger
1409// carrying the seal's bound.
1410func TestCutCrashMatrix(t *testing.T) {
1411 t.Parallel()
1412 ctx := context.Background()
1413 lane := moves.ReplicaPrefix(p1Host, p1ID)
1414 for _, point := range []string{"after-evidence", "after-final-round", "after-seal", "archive-sealed-snapshot"} {
1415 t.Run(point, func(t *testing.T) {
1416 h := newCutHarness(t)
1417 h.ready(t)
1418 h.writeOnStop(t) // so the archive is the cut's own build
1419 h.crashAt = point
1420 h.cut(t)
1421 h.crashPass(t)
1422 m := h.move(t)
1423 c := h.claim(t)
1424 j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
1425 if err != nil {
1426 t.Fatal(err)
1427 }
1428 _, _, reportErr := h.s.Get(ctx, h.stoppedKey())
1429 _, sealed := c.Op.Journal[hostclaim.JournalSeal]
1430 switch point {
1431 case "after-evidence":
1432 // stopping: the evidence journalled, execution still the
1433 // source's, no report, no hold given up.
1434 if m.CutPhase != provision.CutStopping || !j.Done(cutStepEvidence) || c.Op.Exec != "ie-box" || !errors.Is(reportErr, os.ErrNotExist) {
1435 t.Fatalf("durable state: %+v, evidence %v, exec %q, report %v", m, j.Done(cutStepEvidence), c.Op.Exec, reportErr)
1436 }
1437 case "after-final-round":
1438 // stopped: execution given up, the report published, the
1439 // lane holding the final round's copy, and no seal yet.
1440 if m.CutPhase != provision.CutStopped || c.Op.Exec != "" || c.Op.Phase != hostclaim.PhaseMoving || reportErr != nil || sealed {
1441 t.Fatalf("durable state: %+v, claim %+v, report %v", m, c.Op, reportErr)
1442 }
1443 if _, ok := h.f.buckets["data-uk"][lane+lastTXKey]; !ok {
1444 t.Fatal("the final round's copy is not in the lane")
1445 }
1446 case "after-seal":
1447 // stopped: the claim sealed with a seal naming the replica and
1448 // the timers but no files; the ledger without a bound yet.
1449 if m.CutPhase != provision.CutStopped || c.Op.Phase != hostclaim.PhaseSealed || !sealed {
1450 t.Fatalf("durable state: %+v, claim %+v", m, c.Op)
1451 }
1452 if seal := h.seal(t); seal.HasFiles() || seal.N != 3 || len(seal.Timers) != 1 {
1453 t.Fatalf("seal: %+v", seal)
1454 }
1455 if l := h.ledger(t); !l.LeaseUntil.IsZero() {
1456 t.Fatalf("the ledger carries a bound (%v) before the seal's pass finished", l.LeaseUntil)
1457 }
1458 case "archive-sealed-snapshot":
1459 // staged: the identity journalled, the sealed object in
1460 // move-snapshots/ and the manifest not, the seal without files.
1461 id := h.journalArchive(t)
1462 snap := moves.SnapshotsPrefix(p1Host, p1ID)
1463 _, sealedThere := h.f.buckets["data-ie"][snap+id.Name+".sealed"]
1464 _, manifestThere := h.f.buckets["data-ie"][snap+id.Name+".json"]
1465 if m.CutPhase != provision.CutStaged || !sealedThere || manifestThere || h.seal(t).HasFiles() {
1466 t.Fatalf("durable state: %+v, sealed %v, manifest %v", m, sealedThere, manifestThere)
1467 }
1468 }
1469 h.freshWorker()
1470 h.pass(t)
1471 m = h.move(t)
1472 if m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped {
1473 t.Fatalf("the replay: %+v", m)
1474 }
1475 keys, _ := h.s.List(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID))
1476 n := 0
1477 for _, k := range keys {
1478 if strings.Contains(k, "stopped-") {
1479 n++
1480 }
1481 }
1482 seal := h.seal(t)
1483 if n != 1 || !sealNamesArchive(seal, m.Source.Files.Sealed) || seal.N != m.Source.FinalTXID || !seal.ReachesN() {
1484 t.Fatalf("%d reports; seal %+v; source %+v", n, seal, m.Source)
1485 }
1486 for _, o := range append(append(append([]moves.SealedObject{}, seal.Replica...), seal.Timers...), seal.Files...) {
1487 if h.f.versionOf("data-uk", o.Key) != o.VersionID {
1488 t.Fatalf("the seal names %s@%s, current %s", o.Key, o.VersionID, h.f.versionOf("data-uk", o.Key))
1489 }
1490 }
1491 if l := h.ledger(t); !l.LeaseUntil.Equal(seal.LeaseUntil) {
1492 t.Fatalf("ledger bound %v, seal bound %v", l.LeaseUntil, seal.LeaseUntil)
1493 }
1494 if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Exec != "uk-box" {
1495 t.Fatalf("claim: %+v", c.Op)
1496 }
1497 })
1498 }
1499}
1500
1501// Fix round 2, item 1: a loss is declared only when it is established. The
1502// sealed file alone, with the journalled manifest, is enough to publish
1503// the identity: losing only the local manifest before the first PUT does
1504// not revert. A read of the sealed file that fails is an error, never a
1505// loss: the record stays staged, the identity stands, and once the file
1506// reads again the replay finishes it.
1507func TestArchiveLossIsDeclaredOnlyWhenEstablished(t *testing.T) {
1508 t.Parallel()
1509 t.Run("only the manifest is gone", func(t *testing.T) {
1510 h := newCutHarness(t)
1511 h.ready(t)
1512 h.writeOnStop(t)
1513 h.crashAt = "archive-journaled"
1514 h.cut(t)
1515 h.crashPass(t)
1516 first := h.journalArchive(t)
1517 if err := os.Remove(first.manifestPath()); err != nil {
1518 t.Fatal(err)
1519 }
1520 h.freshWorker()
1521 h.pass(t)
1522 m := h.move(t)
1523 if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
1524 t.Fatalf("a kept sealed file with its manifest journalled: %+v files %+v", m, m.Source.Files)
1525 }
1526 snap := moves.SnapshotsPrefix(p1Host, p1ID)
1527 if _, ok := h.f.buckets["data-ie"][snap+first.Name+".json"]; !ok {
1528 t.Fatal("the manifest was not published from the journal")
1529 }
1530 })
1531 t.Run("the sealed file cannot be read", func(t *testing.T) {
1532 h := newCutHarness(t)
1533 h.ready(t)
1534 h.writeOnStop(t)
1535 h.crashAt = "archive-journaled"
1536 h.cut(t)
1537 h.crashPass(t)
1538 first := h.journalArchive(t)
1539 // The sealed path answers a read error (EISDIR), not absence.
1540 kept, err := os.ReadFile(first.sealedPath())
1541 if err != nil {
1542 t.Fatal(err)
1543 }
1544 if err := os.Remove(first.sealedPath()); err != nil {
1545 t.Fatal(err)
1546 }
1547 if err := os.Mkdir(first.sealedPath(), 0o700); err != nil {
1548 t.Fatal(err)
1549 }
1550 h.freshWorker()
1551 h.pass(t)
1552 m := h.move(t)
1553 if m.State != provision.MoveCutting || m.CutPhase != provision.CutStaged {
1554 t.Fatalf("a read error decided the archive: %+v", m)
1555 }
1556 if again := h.journalArchive(t); again.Sealed != first.Sealed {
1557 t.Fatalf("the identity changed from %s to %s", first.Name, again.Name)
1558 }
1559 if got := snapshotSealed(h); len(got) != 0 {
1560 t.Fatalf("snapshots %v, want none", got)
1561 }
1562 if err := os.Remove(first.sealedPath()); err != nil {
1563 t.Fatal(err)
1564 }
1565 if err := os.WriteFile(first.sealedPath(), kept, 0o600); err != nil {
1566 t.Fatal(err)
1567 }
1568 h.freshWorker()
1569 h.pass(t)
1570 m = h.move(t)
1571 if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
1572 t.Fatalf("once the file read again: %+v files %+v", m, m.Source.Files)
1573 }
1574 })
1575}
1576
1577// Fix round 2, item 2: an automatic move on a scoped holder does not loop.
1578// decideReady never repins or requests a cut for a holder whose location is
1579// not the activator's prefix, so repeated passes write nothing: the record's
1580// stamp and the claim's generation stand.
1581func TestScopedAutomaticMoveDoesNotLoop(t *testing.T) {
1582 h := newCutHarness(t)
1583 h.scoped = true
1584 h.start(t, "auto")
1585 h.pass(t)
1586 h.targetPass(t, h.target())
1587 h.pass(t)
1588 rec := h.read(t, p1Host)
1589 if rec.Move.State == provision.MoveCutting {
1590 t.Fatalf("a cut was requested for a scoped holder: %+v", rec.Move)
1591 }
1592 claimGen := h.claim(t).Gen
1593 for i := 0; i < 3; i++ {
1594 h.pass(t)
1595 again := h.read(t, p1Host)
1596 if again.Move.State == provision.MoveCutting || again.Move.UpdatedAt != rec.Move.UpdatedAt || again.Move.State != rec.Move.State {
1597 t.Fatalf("pass %d wrote the record: %+v (was %+v)", i, again.Move, rec.Move)
1598 }
1599 if h.claim(t).Gen != claimGen {
1600 t.Fatalf("pass %d wrote the claim", i)
1601 }
1602 }
1603 if h.ev.calls != 0 || len(h.holds) != 0 {
1604 t.Fatal("something stopped")
1605 }
1606}
1607