| new file mode 100644 |
| index 000000000..1ef35717e |
| --- /dev/null |
| +++ b/internal/hostbox/reconcile_relocation_test.go |
| @@ -0,0 +1,192 @@ |
| +//go:build unix |
| + |
| +package hostbox |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "io/fs" |
| + "maps" |
| + "os" |
| + "path/filepath" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostguard" |
| + "github.com/carlosframework/platform/internal/hostwrite" |
| + "github.com/carlosframework/platform/internal/park" |
| +) |
| + |
| +// relocationReconciler is the reconciler over workFixture's guard and |
| +// store, with the real fence observer (Guard.AdmissionFenced) and a |
| +// recording hold writer. The claim relocates the host off "box". |
| +func relocationReconciler(t *testing.T) (Reconciler, RecordSelection, *[]string, hostguard.Guard) { |
| + t.Helper() |
| + source, store, obs := workFixture(t) |
| + source.Policy = Policy{Held: func(context.Context, string) (string, error) { return "", nil }, Relocation: RelocationPolicy("box")}.Check |
| + // One cold writer session, so the source has a session a stop can fence. |
| + obs, err := hostclaim.Update(context.Background(), store, obs, hostclaim.AdmitWriter(hostclaim.Admission{Box: "box", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + h := obs.Claim.Holder |
| + selected := RecordSelection{Host: obs.Host, Key: h.Key, ETag: "record-etag", Incarnation: h.Incarnation, Family: h.Family, Layout: "flat", Paths: []string{"shop.example.db"}} |
| + holds := &[]string{} |
| + r := Reconciler{Claims: store, Guard: source.Guard, Local: park.Local{Guard: source.Guard, Claims: store, DataDir: source.Guard.DataDir, Box: "box"}, Writes: hostwrite.Config{Enabled: true, Selector: source}, Box: "box", Region: "ie", Now: func() time.Time { return time.Unix(100, 0) }, |
| + CheckRecord: func(context.Context, RecordSelection) error { return nil }, |
| + CheckLocal: func(context.Context, hostclaim.Observed, RecordSelection) error { return nil }, |
| + HoldRow: func(_ context.Context, _ string, op string) error { *holds = append(*holds, op); return nil }} |
| + r.FenceClosed = func(_ context.Context, o hostclaim.Observed) (bool, error) { |
| + captured, err := hostguard.Capture(source.Guard.DataDir, o.Host) |
| + if err != nil { |
| + return false, err |
| + } |
| + return source.Guard.AdmissionFenced(o.Host, o.Claim, captured) |
| + } |
| + return r, selected, holds, source.Guard |
| +} |
| + |
| +func relocate(t *testing.T, r Reconciler, host, phase string) hostclaim.Observed { |
| + t.Helper() |
| + ctx := context.Background() |
| + obs, err := hostclaim.Read(ctx, r.Claims, host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err = hostclaim.Update(ctx, r.Claims, obs, func(c *hostclaim.Claim) error { |
| + c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: phase, Exec: "box", Participants: []string{"box", "dst"}, |
| + Intent: map[string]string{"region": "ie", "source_box": "box", "target_box": "dst"}, Journal: map[string]string{"exec_ack.box": "{}", "exec_ack.dst": "{}"}} |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return obs |
| +} |
| + |
| +// The source in releasing takes the activation path (ready) until root |
| +// begins its stop; then the hold-only path, under the flock, with the |
| +// observer read inside the existing locked operation: a bounded context |
| +// finishes both, so neither reacquires the host lock. |
| +func TestReconcileRelocationSourceReadyThenHeldWithTheRealFenceObserver(t *testing.T) { |
| + r, selected, holds, guard := relocationReconciler(t) |
| + obs := relocate(t, r, selected.Host, hostclaim.PhaseReleasing) |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + writes := 0 |
| + if err := r.Install(ctx, selected, func() error { writes++; return nil }); err != nil { |
| + t.Fatalf("the source before its fence closed: %v", err) |
| + } |
| + if writes != 1 || len(*holds) != 0 { |
| + t.Fatalf("writes %d holds %v", writes, *holds) |
| + } |
| + // Root begins the source's stop: the durable admission fence closes. |
| + // (The inventory write above moved the claim: read it fresh.) |
| + captured, err := hostguard.Capture(guard.DataDir, selected.Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err = hostclaim.Read(ctx, r.Claims, selected.Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + target := hostguard.FenceTarget{FenceID: "fence-1", Host: selected.Host, Incarnation: captured.Incarnation, Stamp: captured.Stamp, Session: obs.Claim.Holder.Session, |
| + Op: "rel-1", Phase: hostclaim.PhaseReleasing, Box: "box", Enrollment: "enrollment-1", BootID: "boot", ControllerID: "ctl", ClaimETag: obs.ETag, DirectoryRevision: "4"} |
| + if _, err := guard.BeginFence(ctx, target); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := r.Install(ctx, selected, func() error { writes++; return nil }); err != nil { |
| + t.Fatalf("the fenced source's hold: %v", err) |
| + } |
| + if writes != 1 || len(*holds) != 1 || (*holds)[0] != "rel-1" { |
| + t.Fatalf("after the fence: writes %d holds %v (the guarded path must not run)", writes, *holds) |
| + } |
| + // A box with no role mints nothing; an observer error keeps rows and |
| + // mints none. |
| + r.Box = "zzz" |
| + if err := r.Install(ctx, selected, func() error { writes++; return nil }); !errors.Is(err, ErrChanged) || writes != 1 { |
| + t.Fatalf("no role: %v writes %d", err, writes) |
| + } |
| + r.Box = "box" |
| + broken := errors.New("record unreadable") |
| + r.FenceClosed = func(context.Context, hostclaim.Observed) (bool, error) { return false, broken } |
| + if err := r.Install(ctx, selected, func() error { writes++; return nil }); !errors.Is(err, broken) || writes != 1 { |
| + t.Fatalf("an observer error: %v writes %d", err, writes) |
| + } |
| +} |
| + |
| +// With no observer (region HA off) the reconciler is today's: the serving |
| +// executor is ready; an executor of any other operation is refused by the |
| +// first check, before the guard is activated or inventory recorded, dry |
| +// run or not (round 5 finding 3). |
| +func TestReconcileWithoutAFenceObserverIsTodays(t *testing.T) { |
| + r, selected, _, guard := relocationReconciler(t) |
| + r.FenceClosed, r.HoldRow = nil, nil |
| + ctx := context.Background() |
| + writes := 0 |
| + if err := r.Install(ctx, selected, func() error { writes++; return nil }); err != nil || writes != 1 { |
| + t.Fatalf("serving: %v writes %d", err, writes) |
| + } |
| + moving := func() hostclaim.Observed { |
| + obs, err := hostclaim.Read(ctx, r.Claims, selected.Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err = hostclaim.Update(ctx, r.Claims, obs, func(c *hostclaim.Claim) error { |
| + c.Op = hostclaim.Op{Kind: hostclaim.KindMoving, ID: "mv-1", Phase: "copying", Exec: "box", Participants: []string{"box", "dst"}, Intent: map[string]string{"admission_id": "adm"}} |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return obs |
| + } |
| + for _, phase := range []string{hostclaim.PhaseStarting, hostclaim.PhaseReleasing, "moving"} { |
| + var obs hostclaim.Observed |
| + if phase == "moving" { |
| + obs = moving() // a held executor participant with HA on (RowHeld) |
| + } else { |
| + obs = relocate(t, r, selected.Host, phase) // ready with HA on (RowReady) |
| + } |
| + before := snapshotDir(t, guard.DataDir) |
| + for _, dry := range []bool{true, false} { |
| + sel := selected |
| + sel.DryRun = dry |
| + if err := r.Install(ctx, sel, func() error { writes++; return nil }); !errors.Is(err, ErrChanged) || writes != 1 { |
| + t.Fatalf("%s (dry run %v) with HA off: %v writes %d", phase, dry, err, writes) |
| + } |
| + } |
| + after, err := hostclaim.Read(ctx, r.Claims, selected.Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if after.ETag != obs.ETag { |
| + t.Fatalf("%s: the claim (inventory) was written with HA off", phase) |
| + } |
| + if got := snapshotDir(t, guard.DataDir); !maps.Equal(got, before) { |
| + t.Fatalf("%s: the guard's local state changed with HA off", phase) |
| + } |
| + } |
| +} |
| + |
| +// snapshotDir is every regular file under dir, by relative path. |
| +func snapshotDir(t *testing.T, dir string) map[string]string { |
| + t.Helper() |
| + out := map[string]string{} |
| + err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { |
| + if err != nil || !d.Type().IsRegular() { |
| + return err |
| + } |
| + raw, err := os.ReadFile(path) |
| + if err != nil { |
| + return err |
| + } |
| + out[path] = string(raw) |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return out |
| +} |
amadan