amadan

carlos / platform Public

layout: the pass, its fence discipline, and recovery that converges before release

Converge gates a move to nested on the isolation probe, defers rows that
are not ready or already fenced by another operation, fences and stops
each host, and re-reads the row under the fence before it journals,
renames through directory fds, swaps the row and reloads routing. An
unproven stop is trusted only for a quiesced host with no uncertain stop
after a successful startup sweep; every refusal lifts the fence it
installed. Recover completes each checked journal to its own end and
then converges the host to the fleet's intent under the same fence, and
holds a completed revert fenced while the intent cannot be read.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 233ee14c023275d459e33305d6c898fd2a66b621 parent 6534a48
2 files changed, +1362 −0
  • internal/layout/pass.go +605 −0
  • internal/layout/pass_linux_test.go +757 −0
diff --git a/internal/layout/pass.go b/internal/layout/pass.go
new file mode 100644
index 000000000..fae2240b2
--- /dev/null
+++ b/internal/layout/pass.go
@@ -0,0 +1,605 @@
+package layout
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Fence is what the pass needs from the activator (design §8.2 step 2):
+// stop the host and refuse wakes for it (FenceAndStop), let it wake again
+// (LiftFence, at exactly the generation this pass installed), and the three
+// reads that decide whether an unproven stop can be trusted and whether
+// somebody else already holds the host's one fence slot.
+type Fence interface {
+ FenceAndStop(ctx context.Context, host string, generation uint64) (*activator.StopEvidence, error)
+ LiftFence(host string, generation uint64) bool
+ Fenced(host string) bool
+ Quiesced(host string) bool
+ StopUncertain(host string) bool
+}
+
+// IntentUnreadable is the intent Recover is given when the agent could not
+// read the fleet's layout intent at all, as distinct from "" (no intent was
+// ever recorded). A revert completed under it is held fenced; see
+// recoverHost.
+const IntentUnreadable Mode = "unreadable"
+
+// Pass is the layout pass for one box. It runs one Converge or Recover at
+// a time; the agent's single-flight is what guarantees that.
+type Pass struct {
+ Registry *registry.Registry
+ // Fence is the activator. Nil means there is none (tests): the pass
+ // still journals and moves.
+ Fence Fence
+ DataDir string
+ SocketDir string // "" means provision.SocketDir
+ SocketsGID int // the carlos-sockets group; -1 leaves the group alone
+ // Reload asks routing to pick up a flipped row now rather than on the
+ // next 2s poll. Nil relies on the poll.
+ Reload func()
+ Now func() time.Time
+ Logf func(string, ...any)
+ Generation func() uint64 // monotonic fence generations; default time.Now().UnixNano()
+ // SweptAtStart reports that the agent's startup unit sweep succeeded,
+ // so no instance from an earlier agent can still be running. Without
+ // it an unproven stop is never trusted.
+ SweptAtStart bool
+ // NestedOK is the isolation probe's gate (design §8.5): a box that
+ // cannot establish the namespace does not start nested tenants, so it
+ // does not make any either. Nil means ungated.
+ NestedOK func() error
+ // Serialize, when set, runs the span from journal write to journal
+ // removal under the lock config sync holds, so sync never renders a
+ // host's bundle into the directory the pass is moving it out of.
+ Serialize func(func())
+
+ // abortAt is a test seam: the step name after which a move returns as
+ // if the agent had died there, leaving its journal for Recover.
+ abortAt string
+}
+
+// Result is what the pass says about one host. Outcome is one of
+// migrated, reverted, recovered, unchanged, skipped, deferred or refused.
+// A Result with an empty Host is about the whole pass.
+type Result struct {
+ Host string
+ Outcome string
+ Reason string
+}
+
+// quiesceTimeout bounds one host's stop.
+const quiesceTimeout = 60 * time.Second
+
+func (p *Pass) now() time.Time {
+ if p.Now != nil {
+ return p.Now()
+ }
+ return time.Now()
+}
+
+func (p *Pass) logf(format string, a ...any) {
+ if p.Logf != nil {
+ p.Logf(format, a...)
+ }
+}
+
+func (p *Pass) generation() uint64 {
+ if p.Generation != nil {
+ return p.Generation()
+ }
+ return uint64(time.Now().UnixNano())
+}
+
+func (p *Pass) socketDir() string {
+ if p.SocketDir != "" {
+ return p.SocketDir
+ }
+ return provision.SocketDir
+}
+
+func (p *Pass) reload() {
+ if p.Reload != nil {
+ p.Reload()
+ }
+}
+
+func (p *Pass) serialize(f func()) {
+ if p.Serialize != nil {
+ p.Serialize(f)
+ return
+ }
+ f()
+}
+
+func (p *Pass) layoutInputs(routes []registry.Route) activator.LayoutInputs {
+ return activator.LayoutInputs{
+ DataDir: p.DataDir,
+ SocketDir: p.socketDir(),
+ Journal: func(host string) bool { return JournalPresent(p.DataDir, host) },
+ Peers: func() ([]registry.Route, error) { return routes, nil },
+ }
+}
+
+// Classify is the box's one classification, shared with the status
+// snapshot so both say the same thing about a row.
+func (p *Pass) Classify(rt registry.Route, peers []registry.Route) (activator.Layout, string) {
+ return activator.Classify(rt, p.layoutInputs(peers))
+}
+
+func deadline(ctx context.Context) (context.Context, context.CancelFunc) {
+ return context.WithTimeout(ctx, quiesceTimeout)
+}
+
+// ready is the status a row must have for the pass to move it. Any other
+// status belongs to an operation in flight on the row, the residence
+// move's "moving" among them, and a row about to leave the box is not
+// worth moving.
+func ready(rt registry.Route) bool { return rt.Status == "" || rt.Status == "ready" }
+
+// Converge runs Recover, then moves every managed row that is not already
+// in mode (design §8.2 steps 1 to 9). Rows are visited in host order so a
+// log reads the same way twice. A host Recover reported is not visited
+// again: recovery already converged it, or refused it.
+func (p *Pass) Converge(ctx context.Context, mode Mode) []Result {
+ if mode != ModeNested && mode != ModeFlat {
+ return []Result{{Outcome: "refused", Reason: fmt.Sprintf("layout mode %q is not nested or flat", mode)}}
+ }
+ if mode == ModeNested && p.NestedOK != nil {
+ if err := p.NestedOK(); err != nil {
+ return []Result{{Outcome: "refused", Reason: "isolation not ready: " + err.Error()}}
+ }
+ }
+ out := p.Recover(ctx, mode)
+ seen := map[string]bool{}
+ for _, r := range out {
+ seen[r.Host] = true
+ }
+ routes, err := p.Registry.List()
+ if err != nil {
+ p.logf("layout: converge: list routes: %v", err)
+ return append(out, Result{Outcome: "refused", Reason: "list routes: " + err.Error()})
+ }
+ sort.Slice(routes, func(i, j int) bool { return routes[i].Host < routes[j].Host })
+ for _, rt := range routes {
+ if !activator.ActivatorManaged(rt) || seen[rt.Host] {
+ continue
+ }
+ if ctx.Err() != nil {
+ break
+ }
+ out = append(out, p.convergeHost(ctx, rt, routes, mode))
+ }
+ return out
+}
+
+func (p *Pass) convergeHost(ctx context.Context, rt registry.Route, routes []registry.Route, mode Mode) Result {
+ if !ready(rt) {
+ return Result{rt.Host, "deferred", "status " + rt.Status}
+ }
+ layout, reason := p.Classify(rt, routes)
+ switch layout {
+ case activator.LayoutFenced, activator.LayoutConflict:
+ return Result{rt.Host, "refused", reason}
+ case activator.LayoutForeign:
+ return Result{rt.Host, "skipped", "foreign: " + reason}
+ case activator.LayoutNested:
+ if mode == ModeNested {
+ return Result{rt.Host, "unchanged", ""}
+ }
+ case activator.LayoutFlat:
+ if mode == ModeFlat {
+ return Result{rt.Host, "unchanged", ""}
+ }
+ }
+ return p.migrateHost(ctx, rt.Host, routes, mode)
+}
+
+// migrateHost is steps 2 to 9 for one host. It checks what it can before
+// stopping anything, fences and stops the host, and releases it only after
+// the move is committed and routing has reloaded, or has been refused.
+func (p *Pass) migrateHost(ctx context.Context, host string, routes []registry.Route, mode Mode) Result {
+ plan, err := PlanFor(p.DataDir, p.socketDir(), host, mode)
+ if err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if err := Preflight(plan, routes); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ // The activator has one fence slot per host, and FenceAndStop replaces
+ // a lower generation. Fencing over another operation's fence and then
+ // lifting ours would reopen admission that operation closed.
+ if p.Fence != nil && p.Fence.Fenced(host) {
+ return Result{host, "deferred", "fenced by another operation"}
+ }
+ gen := p.generation()
+ if r, ok := p.quiesce(ctx, host, gen); !ok {
+ return r
+ }
+ defer p.release(host, gen)
+ return p.shift(host, mode)
+}
+
+// quiesce installs this pass's fence and decides whether the host is
+// proven stopped. An unproven stop is trusted only when nothing can be
+// running: no live entry, no stop on record that failed to finish, and a
+// startup sweep that stopped anything an earlier agent left. After an
+// agent restart no asleep host has a stop on record, which is why the
+// entry-less case needs this rather than a proof. Every refusal lifts the
+// fence it installed: a fence left behind would keep a host down with
+// nothing on its way to release it.
+func (p *Pass) quiesce(ctx context.Context, host string, gen uint64) (Result, bool) {
+ if p.Fence == nil {
+ return Result{}, true
+ }
+ fctx, cancel := deadline(ctx)
+ defer cancel()
+ _, err := p.Fence.FenceAndStop(fctx, host, gen)
+ if err == nil {
+ return Result{}, true
+ }
+ unproven := errors.Is(err, activator.ErrStopUnproven)
+ if unproven && p.SweptAtStart && p.Fence.Quiesced(host) && !p.Fence.StopUncertain(host) {
+ return Result{}, true
+ }
+ p.Fence.LiftFence(host, gen)
+ reason := "stop unproven"
+ if !unproven {
+ reason += ": " + err.Error()
+ }
+ p.logf("layout: %s: deferred -> %s", host, reason)
+ return Result{host, "deferred", reason}, false
+}
+
+func (p *Pass) release(host string, gen uint64) {
+ if p.Fence != nil {
+ p.Fence.LiftFence(host, gen)
+ }
+}
+
+// shift is steps 3 to 9 for a host this pass already holds fenced and
+// stopped. It re-reads the row under the fence, because the copy the pass
+// classified can be stale: a move cut can have started, or the row been
+// unflagged, re-added or migrated. Every return before the journal is
+// written leaves nothing changed.
+func (p *Pass) shift(host string, mode Mode) Result {
+ fresh, err := p.Registry.Get(host)
+ if err != nil {
+ return Result{host, "refused", "row: " + err.Error()}
+ }
+ if fresh == nil {
+ return Result{host, "refused", "row vanished under the fence"}
+ }
+ if !ready(*fresh) {
+ return Result{host, "refused", "status " + fresh.Status + " under the fence"}
+ }
+ if !activator.ActivatorManaged(*fresh) {
+ return Result{host, "refused", "no longer activator-managed under the fence"}
+ }
+ routes, err := p.Registry.List()
+ if err != nil {
+ return Result{host, "refused", "list routes: " + err.Error()}
+ }
+ want, from := activator.LayoutNested, activator.LayoutFlat
+ if mode == ModeFlat {
+ want, from = activator.LayoutFlat, activator.LayoutNested
+ }
+ switch layout, reason := p.Classify(*fresh, routes); layout {
+ case want:
+ return Result{host, "unchanged", ""}
+ case from:
+ default:
+ return Result{host, "refused", fmt.Sprintf("%s under the fence: %s", layout, reason)}
+ }
+ plan, err := PlanFor(p.DataDir, p.socketDir(), host, mode)
+ if err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if err := Preflight(plan, routes); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ // Both directions: a revert renames out of the raw directory, so it
+ // must be a real directory too, and it is cheaper to find a planted
+ // link here than after the journal has fenced the host.
+ if err := EnsureDirs(p.DataDir, p.socketDir(), host, p.SocketsGID); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ aside, err := DestinationCheck(plan, p.now)
+ if err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ present := make([]bool, len(plan.Artifacts))
+ for i, a := range plan.Artifacts {
+ // Nothing is ever renamed for the socket: the listener is gone
+ // with the stopped host, and the next wake binds the new path.
+ if a.Name == "socket" {
+ continue
+ }
+ ok, err := lstatPresent(a.Src)
+ if err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if a.Required && !ok {
+ return Result{host, "refused", fmt.Sprintf("%s is missing at %s", a.Name, a.Src)}
+ }
+ present[i] = ok
+ plan.Artifacts[i].Present = ok
+ }
+ var res Result
+ p.serialize(func() { res = p.commit(plan, plan.Journal(present), aside) })
+ if res.Outcome == "migrated" || res.Outcome == "reverted" {
+ p.reload()
+ }
+ return res
+}
+
+// commit is steps 6 to 9 up to the journal's removal. The caller reloads
+// routing afterwards, outside Serialize's lock and still under the fence.
+// Every return after the journal is written leaves it for Recover, and the
+// journal keeps the host fenced until then.
+func (p *Pass) commit(plan *Plan, j *Journal, aside []string) Result {
+ host := plan.Host
+ if err := WriteJournal(p.DataDir, j); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if p.abortAt == "after-journal" {
+ return Result{host, "refused", "aborted"}
+ }
+ if err := p.moveFiles(plan, j); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if p.abortAt == "after-files" {
+ return Result{host, "refused", "aborted"}
+ }
+ if err := UpdatePhase(p.DataDir, j, PhaseRow); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if err := p.Registry.SetInstancePaths(host, plan.From.DB, plan.From.Socket, plan.To.DB, plan.To.Socket); err != nil {
+ return Result{host, "refused", "row: " + err.Error()}
+ }
+ if p.abortAt == "after-cas" {
+ return Result{host, "refused", "aborted"}
+ }
+ if err := UpdatePhase(p.DataDir, j, PhaseDone); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ if p.abortAt == "before-remove" {
+ return Result{host, "refused", "aborted"}
+ }
+ if err := RemoveJournal(p.DataDir, host); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ outcome := "migrated"
+ if plan.Mode == ModeFlat {
+ outcome = "reverted"
+ }
+ if len(aside) > 0 {
+ p.logf("layout: %s: %s -> %s (kept aside: %s)", host, plan.From.DB, plan.To.DB, strings.Join(aside, ", "))
+ } else {
+ p.logf("layout: %s: %s -> %s", host, plan.From.DB, plan.To.DB)
+ }
+ return Result{host, outcome, ""}
+}
+
+var errAborted = errors.New("layout: aborted by test seam")
+
+// moveFiles renames every artifact the journal records present. One
+// already at its destination is done; one still at its source is renamed;
+// one at neither is lost, and one at both was not left that way by a
+// crash (the destination check cleared every destination before the
+// journal was written, and a rename leaves a file in exactly one place).
+// Either of the last two is an error that keeps the journal and so the
+// fence.
+func (p *Pass) moveFiles(plan *Plan, j *Journal) error {
+ for i, a := range plan.Artifacts {
+ if a.Name == "socket" || !j.Artifacts[i].Present {
+ continue
+ }
+ atDst, err := lstatPresent(a.Dst)
+ if err != nil {
+ return err
+ }
+ atSrc, err := lstatPresent(a.Src)
+ if err != nil {
+ return err
+ }
+ switch {
+ case atDst && atSrc:
+ return fmt.Errorf("layout: %s: %s is at both %s and %s; refusing to choose, not acting", plan.Host, a.Name, a.Src, a.Dst)
+ case atDst:
+ continue
+ case !atSrc:
+ return fmt.Errorf("layout: %s: %s is at neither %s nor %s; not acting", plan.Host, a.Name, a.Src, a.Dst)
+ }
+ if err := instancefs.RenameBetween(filepath.Dir(a.Src), filepath.Base(a.Src), filepath.Dir(a.Dst), filepath.Base(a.Dst)); err != nil {
+ return fmt.Errorf("layout: %s: move %s: %w", plan.Host, a.Name, err)
+ }
+ if p.abortAt == "mid-files" {
+ return errAborted
+ }
+ }
+ return nil
+}
+
+// Recover completes or refuses every journal in <data>/layout (design
+// §8.2, "Recovery is the journal"). It runs at agent start and at the top
+// of every Converge. intent is the fleet's recorded layout: "" when none
+// was ever recorded, IntentUnreadable when the agent could not read it.
+// Recovery is never gated on the isolation probe: it is how a host gets
+// out of the fenced state.
+func (p *Pass) Recover(ctx context.Context, intent Mode) []Result {
+ entries, err := os.ReadDir(journalDir(p.DataDir))
+ if errors.Is(err, fs.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ p.logf("layout: recover: read journals: %v", err)
+ return []Result{{Outcome: "refused", Reason: "read journals: " + err.Error()}}
+ }
+ var names []string
+ for _, e := range entries {
+ // Dot names are WriteJournal's and UpdatePhase's temp files.
+ if n := e.Name(); !strings.HasPrefix(n, ".") && strings.HasSuffix(n, ".json") {
+ names = append(names, strings.TrimSuffix(n, ".json"))
+ }
+ }
+ if len(names) == 0 {
+ return nil
+ }
+ routes, err := p.Registry.List()
+ if err != nil {
+ p.logf("layout: recover: list routes: %v", err)
+ return []Result{{Outcome: "refused", Reason: "list routes: " + err.Error()}}
+ }
+ byHost := map[string]registry.Route{}
+ for _, rt := range routes {
+ byHost[rt.Host] = rt
+ }
+ var out []Result
+ for _, host := range names {
+ if ctx.Err() != nil {
+ break
+ }
+ r := p.recoverHost(ctx, host, byHost, intent)
+ if r.Outcome == "refused" {
+ p.logf("layout: %s: recovery refused -> %s", host, r.Reason)
+ }
+ out = append(out, r)
+ }
+ return out
+}
+
+// notActing makes a recovery refusal say, in the words design §8.2 uses,
+// that the pass left the host alone and fenced.
+func notActing(err error) string {
+ if s := err.Error(); strings.HasSuffix(s, "not acting") {
+ return s
+ }
+ return err.Error() + "; not acting"
+}
+
+// recoverHost checks one journal against the row and both derived
+// endpoints, completes it to its own end under a fence, and then, still
+// under that fence, converges the host to intent. Releasing the host
+// between the two would let a forged or stale journal run it for a moment
+// in a layout the fleet did not ask for (design §3.3).
+func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]registry.Route, intent Mode) Result {
+ if !controlname.ValidHostShape(host) {
+ return Result{host, "refused", "journal file name is not a host; not acting"}
+ }
+ j, found, err := ReadJournal(p.DataDir, host)
+ if err != nil {
+ return Result{host, "refused", notActing(err)}
+ }
+ if !found {
+ return Result{host, "unchanged", ""}
+ }
+ rt, ok := byHost[host]
+ if !ok {
+ return Result{host, "refused", "journal names a host with no row; not acting"}
+ }
+ // Checked before the stop as well as after it, so a journal that
+ // cannot be honoured never costs its host a stop.
+ if _, err := Validate(j, p.DataDir, p.socketDir(), rt); err != nil {
+ return Result{host, "refused", notActing(err)}
+ }
+ plan, err := PlanFor(p.DataDir, p.socketDir(), host, j.Mode)
+ if err != nil {
+ return Result{host, "refused", notActing(err)}
+ }
+ gen := p.generation()
+ if r, ok := p.quiesce(ctx, host, gen); !ok {
+ return r
+ }
+ defer p.release(host, gen)
+ // The row was read before the host stopped; check the journal against
+ // the row as it stands now that nothing else can wake it.
+ fresh, err := p.Registry.Get(host)
+ if err != nil {
+ return Result{host, "refused", notActing(fmt.Errorf("row: %w", err))}
+ }
+ if fresh == nil {
+ return Result{host, "refused", "row vanished under the fence; not acting"}
+ }
+ atFrom, err := Validate(j, p.DataDir, p.socketDir(), *fresh)
+ if err != nil {
+ return Result{host, "refused", notActing(err)}
+ }
+ // With the intent unreadable, a completed revert stays fenced at done:
+ // a flat tenant can forge a valid revert journal for a nested host, and
+ // releasing it flat would run it unisolated until an intent is read.
+ hold := intent == IntentUnreadable && j.Mode == ModeFlat
+ var res Result
+ p.serialize(func() { res = p.complete(plan, j, atFrom, hold) })
+ if res.Outcome == "recovered" || res.Outcome == "deferred" {
+ // Both flipped the row (or found it flipped); a held host is
+ // still fenced by its journal, but routing should say where it is.
+ p.reload()
+ }
+ if res.Outcome != "recovered" || intent == "" || intent == IntentUnreadable || intent == j.Mode {
+ return res
+ }
+ next := p.shift(host, intent)
+ switch next.Outcome {
+ case "migrated", "reverted":
+ return Result{host, "recovered", fmt.Sprintf("completed to %s, then converged to %s", j.Mode, intent)}
+ case "unchanged":
+ return res
+ }
+ return Result{host, next.Outcome, fmt.Sprintf("completed to %s; converging to %s: %s", j.Mode, intent, next.Reason)}
+}
+
+// complete runs the journal's remaining phases (design §8.2's recovery
+// list). A done journal whose row is still at from is refused rather than
+// removed: a crash cannot produce it, since done is written only after the
+// row's swap committed, and removing it would release a row that names the
+// files' old place.
+func (p *Pass) complete(plan *Plan, j *Journal, atFrom, hold bool) Result {
+ host, phase := plan.Host, j.Phase
+ switch j.Phase {
+ case PhaseFiles:
+ if err := p.moveFiles(plan, j); err != nil {
+ return Result{host, "refused", notActing(err)}
+ }
+ if err := UpdatePhase(p.DataDir, j, PhaseRow); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ fallthrough
+ case PhaseRow:
+ if atFrom {
+ if err := p.Registry.SetInstancePaths(host, j.From.DB, j.From.Socket, j.To.DB, j.To.Socket); err != nil {
+ return Result{host, "refused", notActing(fmt.Errorf("row: %w", err))}
+ }
+ }
+ if err := UpdatePhase(p.DataDir, j, PhaseDone); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ case PhaseDone:
+ if atFrom {
+ return Result{host, "refused", "journal is done but the row is still at its source; not acting"}
+ }
+ default:
+ return Result{host, "refused", fmt.Sprintf("journal phase %q unknown; not acting", j.Phase)}
+ }
+ if hold {
+ p.logf("layout: %s: completed -> %s; journal kept at done until the layout intent can be read", host, j.Mode)
+ return Result{host, "deferred", "completed to flat; journal kept until the layout intent can be read"}
+ }
+ if err := RemoveJournal(p.DataDir, host); err != nil {
+ return Result{host, "refused", err.Error()}
+ }
+ p.logf("layout: %s: recovered -> %s (journal phase was %s)", host, j.Mode, phase)
+ return Result{host, "recovered", string(j.Mode)}
+}
diff --git a/internal/layout/pass_linux_test.go b/internal/layout/pass_linux_test.go
new file mode 100644
index 000000000..2f9b6dff7
--- /dev/null
+++ b/internal/layout/pass_linux_test.go
@@ -0,0 +1,757 @@
+//go:build linux
+
+package layout
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+const (
+ host = "h.example.com"
+ replica = "r.example.com"
+)
+
+// fakeFence is the activator's fence as the pass sees it. By default every
+// stop is proven; unproven, awake and uncertain make FenceAndStop answer
+// ErrStopUnproven and shape what Quiesced and StopUncertain say after it.
+// events records every fence and lift, and box's Reload adds "reload", so a
+// test can read the order the pass did things in.
+type fakeFence struct {
+ fences map[string]uint64
+ unproven map[string]bool
+ awake map[string]bool
+ uncertain map[string]bool
+ onFence func(host string)
+ events []string
+}
+
+func (f *fakeFence) FenceAndStop(_ context.Context, h string, gen uint64) (*activator.StopEvidence, error) {
+ if f.fences == nil {
+ f.fences = map[string]uint64{}
+ }
+ f.fences[h] = gen
+ f.events = append(f.events, "fence "+h)
+ if f.onFence != nil {
+ f.onFence(h)
+ }
+ if f.unproven[h] {
+ return nil, activator.ErrStopUnproven
+ }
+ return &activator.StopEvidence{}, nil
+}
+
+func (f *fakeFence) LiftFence(h string, gen uint64) bool {
+ if f.fences[h] != gen {
+ return false
+ }
+ delete(f.fences, h)
+ f.events = append(f.events, "lift "+h)
+ return true
+}
+
+func (f *fakeFence) Fenced(h string) bool { return f.fences[h] != 0 }
+func (f *fakeFence) Quiesced(h string) bool { return !f.awake[h] }
+func (f *fakeFence) StopUncertain(h string) bool { return f.uncertain[h] }
+
+func openLayoutRegistry(t *testing.T, dir string) *registry.Registry {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ return reg
+}
+
+// seedFlat writes host's flat files: the database and its wal, the admin
+// env, a litestream shadow directory with a file in it, and (for the
+// primary host only) a quarantine. The sidecar env and the rendered bundle
+// are absent, so the journal records them absent.
+func seedFlat(t *testing.T, dd, h string) {
+ t.Helper()
+ db := provision.DefaultDBPath(dd, h)
+ files := map[string]string{
+ db: "db " + h,
+ db + "-wal": "wal",
+ filepath.Join(dd, h+".env"): "CARLOS_ADMIN_TOKEN=t\n",
+ filepath.Join(dd, shadowName(h), "generation"): "g",
+ }
+ if h == host {
+ files[db+".corrupt"] = "quarantined"
+ }
+ for path, body := range files {
+ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// box is a temp data dir, a temp socket dir and a registry with two flat
+// managed rows whose files exist: an ordinary instance and a replica
+// (#162), which is its own host and migrates like any other.
+func box(t *testing.T) (*Pass, *fakeFence) {
+ t.Helper()
+ dd := t.TempDir()
+ // The socket dir is short on purpose: t.TempDir embeds the test's name,
+ // and the nested control socket must fit a unix socket address, which
+ // Preflight enforces.
+ rd, err := os.MkdirTemp("", "ly")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(rd) })
+ reg := openLayoutRegistry(t, dd)
+ for _, rt := range []registry.Route{
+ {Host: host, Kind: "instance", Hibernate: true, Status: "ready",
+ Socket: filepath.Join(rd, host+".sock"), DBPath: provision.DefaultDBPath(dd, host)},
+ {Host: replica, Kind: "instance", Hibernate: true, Replica: true,
+ Socket: filepath.Join(rd, replica+".sock"), DBPath: provision.DefaultDBPath(dd, replica)},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ seedFlat(t, dd, rt.Host)
+ }
+ f := &fakeFence{}
+ p := &Pass{Registry: reg, Fence: f, DataDir: dd, SocketDir: rd, SocketsGID: -1, Now: time.Now,
+ SweptAtStart: true,
+ Reload: func() { f.events = append(f.events, "reload") },
+ Logf: t.Logf}
+ return p, f
+}
+
+func byHost(t *testing.T, rs []Result) map[string]Result {
+ t.Helper()
+ out := map[string]Result{}
+ for _, r := range rs {
+ if _, dup := out[r.Host]; dup {
+ t.Errorf("host %q reported twice: %+v", r.Host, rs)
+ }
+ out[r.Host] = r
+ }
+ return out
+}
+
+func endpoints(t *testing.T, p *Pass, h string, mode Mode) Endpoint {
+ t.Helper()
+ plan, err := PlanFor(p.DataDir, p.SocketDir, h, mode)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return plan.To
+}
+
+func rowAt(t *testing.T, p *Pass, h string) Endpoint {
+ t.Helper()
+ rt, err := p.Registry.Get(h)
+ if err != nil || rt == nil {
+ t.Fatalf("row %s: %v %v", h, rt, err)
+ }
+ return Endpoint{DB: rt.DBPath, Socket: rt.Socket}
+}
+
+func assertFlatUntouched(t *testing.T, p *Pass, h string) {
+ t.Helper()
+ if got := rowAt(t, p, h); got != endpoints(t, p, h, ModeFlat) {
+ t.Errorf("%s row moved: %+v", h, got)
+ }
+ if b, err := os.ReadFile(provision.DefaultDBPath(p.DataDir, h)); err != nil || string(b) != "db "+h {
+ t.Errorf("%s flat db: %q %v", h, b, err)
+ }
+ if JournalPresent(p.DataDir, h) {
+ t.Errorf("%s has a journal", h)
+ }
+}
+
+func TestConvergeNestedMovesFilesFlipsTheRowAndRemovesTheJournal(t *testing.T) {
+ p, f := box(t)
+ // Step 9's order: routing reloads on the committed row with the journal
+ // gone, and only then is the host released.
+ var reloads []string
+ p.Reload = func() {
+ for _, h := range []string{host, replica} {
+ if f.fences[h] != 0 {
+ if JournalPresent(p.DataDir, h) {
+ t.Errorf("reload for %s ran with its journal still present", h)
+ }
+ reloads = append(reloads, h+" "+rowAt(t, p, h).DB)
+ }
+ }
+ }
+ res := byHost(t, p.Converge(context.Background(), ModeNested))
+ for _, h := range []string{host, replica} {
+ if res[h].Outcome != "migrated" {
+ t.Fatalf("%s: %+v", h, res[h])
+ }
+ want := endpoints(t, p, h, ModeNested)
+ if got := rowAt(t, p, h); got != want {
+ t.Fatalf("%s row = %+v, want %+v", h, got, want)
+ }
+ raw := rawInstanceDir(p.DataDir, h)
+ for _, name := range []string{h + ".db", h + ".db-wal", h + ".env", filepath.Join(shadowName(h), "generation")} {
+ if _, err := os.Stat(filepath.Join(raw, name)); err != nil {
+ t.Errorf("%s not at the raw destination: %v", name, err)
+ }
+ }
+ if _, err := os.Lstat(provision.DefaultDBPath(p.DataDir, h)); !errors.Is(err, os.ErrNotExist) {
+ t.Errorf("%s flat db still present", h)
+ }
+ if JournalPresent(p.DataDir, h) {
+ t.Errorf("%s journal left behind after a clean move", h)
+ }
+ if fi, err := os.Lstat(rawRunDir(p.SocketDir, h)); err != nil || !fi.IsDir() {
+ t.Errorf("%s run directory: %v %v", h, fi, err)
+ }
+ }
+ if _, err := os.Stat(filepath.Join(rawInstanceDir(p.DataDir, host), host+".db.corrupt")); err != nil {
+ t.Errorf("quarantine not moved: %v", err)
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("fences left installed after a clean move: %v", f.fences)
+ }
+ wantReloads := []string{host + " " + endpoints(t, p, host, ModeNested).DB, replica + " " + endpoints(t, p, replica, ModeNested).DB}
+ if !reflect.DeepEqual(reloads, wantReloads) {
+ t.Errorf("reloads under the fence = %v, want %v", reloads, wantReloads)
+ }
+ // Idempotent: a second converge does nothing.
+ for h, r := range byHost(t, p.Converge(context.Background(), ModeNested)) {
+ if r.Outcome != "unchanged" {
+ t.Fatalf("second converge, %s: %+v", h, r)
+ }
+ }
+ // Revert restores the flat layout byte for byte.
+ p.Reload = nil
+ res = byHost(t, p.Converge(context.Background(), ModeFlat))
+ for _, h := range []string{host, replica} {
+ if res[h].Outcome != "reverted" {
+ t.Fatalf("revert %s = %+v", h, res[h])
+ }
+ assertFlatUntouched(t, p, h)
+ if _, err := os.Stat(filepath.Join(p.DataDir, shadowName(h), "generation")); err != nil {
+ t.Errorf("%s shadow not back: %v", h, err)
+ }
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("fences left installed after a revert: %v", f.fences)
+ }
+}
+
+// An unproven stop is accepted only for a host with no live entry, no
+// uncertain stop on record, and an agent whose startup unit sweep
+// succeeded. Anything else defers the host, writes no journal, and lifts
+// the fence this pass installed.
+func TestConvergeTrustsAnUnprovenStopOnlyWhenNothingCanBeRunning(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ swept bool
+ awake bool
+ uncertain bool
+ want string
+ }{
+ {"swept, quiesced, certain", true, false, false, "migrated"},
+ {"no startup sweep", false, false, false, "deferred"},
+ {"an entry is live", true, true, false, "deferred"},
+ {"the last stop is uncertain", true, false, true, "deferred"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ p, f := box(t)
+ p.SweptAtStart = tc.swept
+ f.unproven = map[string]bool{host: true}
+ f.awake = map[string]bool{host: tc.awake}
+ f.uncertain = map[string]bool{host: tc.uncertain}
+ r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
+ if r.Outcome != tc.want {
+ t.Fatalf("result = %+v, want %s", r, tc.want)
+ }
+ if tc.want == "deferred" {
+ if !strings.Contains(r.Reason, "stop unproven") {
+ t.Errorf("reason = %q", r.Reason)
+ }
+ assertFlatUntouched(t, p, host)
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("a layout fence was left installed: %v", f.fences)
+ }
+ })
+ }
+}
+
+// wakeBackend is an activator.Backend with no processes, no S3 and no
+// replication: just enough for a real Service to wake a host.
+type wakeBackend struct {
+ mu sync.Mutex
+ started []registry.Route
+}
+
+type nopLease struct{}
+
+func (nopLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) {
+ return nopRep{}, nil
+}
+func (nopLease) Release(context.Context) error { return nil }
+
+type nopRep struct{}
+
+func (nopRep) Err() error { return nil }
+func (nopRep) Close(context.Context) error { return nil }
+
+type nopProc struct{}
+
+func (nopProc) Stop(context.Context) error { return nil }
+func (nopProc) Exited() bool { return false }
+
+func (b *wakeBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) {
+ return nopLease{}, nil
+}
+func (b *wakeBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
+func (b *wakeBackend) Seed(context.Context, registry.Route) error { return nil }
+func (b *wakeBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.started = append(b.started, rt)
+ return nopProc{}, "", nil
+}
+func (b *wakeBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
+func (b *wakeBackend) HasSidecar(registry.Route) bool { return false }
+func (b *wakeBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
+ return nil, errors.New("wakeBackend: no sidecar in this fixture")
+}
+func (b *wakeBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) {
+ return nil, "", errors.New("wakeBackend: no worker in this fixture")
+}
+
+func realService(t *testing.T, p *Pass) (*activator.Service, *wakeBackend) {
+ t.Helper()
+ b := &wakeBackend{}
+ svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, b, p.Registry)
+ // The wake re-reads the committed row, so the test sees which paths it
+ // actually woke on.
+ svc.Routes = p.Registry
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ svc.Shutdown(ctx)
+ })
+ p.Fence = svc
+ return svc, b
+}
+
+// After an agent restart no asleep host has a stop on record, so the real
+// activator answers ErrStopUnproven for every one of them. The pass must
+// still migrate such a host, and must leave it wakeable.
+func TestConvergeAgainstARealActivatorMigratesAnEntrylessHostAndItWakes(t *testing.T) {
+ p, _ := box(t)
+ svc, b := realService(t, p)
+ res := byHost(t, p.Converge(context.Background(), ModeNested))
+ if res[host].Outcome != "migrated" {
+ t.Fatalf("result = %+v", res[host])
+ }
+ if svc.Fenced(host) {
+ t.Fatal("the host is still fenced after the pass")
+ }
+ rt, err := p.Registry.Get(host)
+ if err != nil || rt == nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ req := httptest.NewRequest(http.MethodGet, "http://"+host+"/", nil)
+ ok := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })
+ if !svc.ServeHTTP(rec, req, *rt, ok) || rec.Code != http.StatusOK {
+ t.Fatalf("wake after the pass: code %d", rec.Code)
+ }
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ if len(b.started) != 1 || b.started[0].DBPath != endpoints(t, p, host, ModeNested).DB {
+ t.Fatalf("started = %+v", b.started)
+ }
+}
+
+func TestConvergeAgainstARealActivatorDefersWithoutAStartupSweep(t *testing.T) {
+ p, _ := box(t)
+ svc, _ := realService(t, p)
+ p.SweptAtStart = false
+ for h, r := range byHost(t, p.Converge(context.Background(), ModeNested)) {
+ if r.Outcome != "deferred" || !strings.Contains(r.Reason, "stop unproven") {
+ t.Errorf("%s: %+v", h, r)
+ }
+ if svc.Fenced(h) {
+ t.Errorf("%s left fenced after a refused quiesce", h)
+ }
+ assertFlatUntouched(t, p, h)
+ }
+}
+
+// Another operation's fence (the residence move's cut) shares the
+// activator's one fence slot per host. The pass must neither take it over
+// nor lift it.
+func TestConvergeDefersAHostSomeoneElseFenced(t *testing.T) {
+ p, _ := box(t)
+ svc, _ := realService(t, p)
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ if _, err := svc.FenceAndStop(ctx, host, 5); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
+ t.Fatalf("pre-installing the other fence: %v", err)
+ }
+ p.Generation = func() uint64 { return 1000 }
+ res := byHost(t, p.Converge(context.Background(), ModeNested))
+ if res[host].Outcome != "deferred" {
+ t.Fatalf("fenced host: %+v", res[host])
+ }
+ if res[replica].Outcome != "migrated" {
+ t.Fatalf("the other host: %+v", res[replica])
+ }
+ assertFlatUntouched(t, p, host)
+ if !svc.LiftFence(host, 5) {
+ t.Fatal("the original fence is not in place at its own generation")
+ }
+}
+
+func TestConvergeDefersARowThatIsNotReady(t *testing.T) {
+ p, f := box(t)
+ if err := p.Registry.SetStatus(host, "moving"); err != nil {
+ t.Fatal(err)
+ }
+ r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
+ if r.Outcome != "deferred" || !strings.Contains(r.Reason, "moving") {
+ t.Fatalf("result = %+v", r)
+ }
+ for _, e := range f.events {
+ if e == "fence "+host {
+ t.Fatal("a moving row was fenced")
+ }
+ }
+ assertFlatUntouched(t, p, host)
+}
+
+// The row the pass classified can change before its fence lands: a move
+// cut starting, an operator unflagging it. The pass re-reads it under the
+// fence and refuses, writing nothing.
+func TestConvergeRefusesARowThatChangedUnderTheFence(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ change func(reg *registry.Registry) error
+ }{
+ {"status", func(reg *registry.Registry) error { return reg.SetStatus(host, "moving") }},
+ {"unflagged", func(reg *registry.Registry) error { return reg.SetHibernate(host, false) }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ p, f := box(t)
+ f.onFence = func(h string) {
+ if h == host {
+ if err := tc.change(p.Registry); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
+ if r.Outcome != "refused" {
+ t.Fatalf("result = %+v", r)
+ }
+ assertFlatUntouched(t, p, host)
+ if len(f.fences) != 0 {
+ t.Errorf("fence left installed: %v", f.fences)
+ }
+ })
+ }
+}
+
+// A box whose isolation probe is not ready does not start nested tenants,
+// so it does not make any.
+func TestConvergeNestedIsGatedOnIsolation(t *testing.T) {
+ p, f := box(t)
+ calls := 0
+ p.NestedOK = func() error { calls++; return errors.New("probe: not ready") }
+ res := p.Converge(context.Background(), ModeNested)
+ if len(res) != 1 || res[0].Host != "" || res[0].Outcome != "refused" || res[0].Reason != "isolation not ready: probe: not ready" {
+ t.Fatalf("results = %+v", res)
+ }
+ if calls != 1 {
+ t.Errorf("NestedOK called %d times", calls)
+ }
+ if len(f.events) != 0 {
+ t.Errorf("the pass touched hosts: %v", f.events)
+ }
+ for _, h := range []string{host, replica} {
+ assertFlatUntouched(t, p, h)
+ }
+ // Reverting and recovering are never gated: they are how a box gets
+ // back to a layout it can serve.
+ p.NestedOK = nil
+ p.Converge(context.Background(), ModeNested)
+ p.NestedOK = func() error { return errors.New("probe: not ready") }
+ for h, r := range byHost(t, p.Converge(context.Background(), ModeFlat)) {
+ if r.Outcome != "reverted" {
+ t.Errorf("revert %s under a failing probe: %+v", h, r)
+ }
+ }
+}
+
+func TestConvergeSkipsForeignAndUnmanagedRows(t *testing.T) {
+ p, _ := box(t)
+ if err := p.Registry.Upsert(registry.Route{Host: "custom.example.com", Kind: "instance", Hibernate: true,
+ Socket: filepath.Join(p.SocketDir, "custom.example.com.sock"), DBPath: "/data/sub/store.db"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := p.Registry.Upsert(registry.Route{Host: "ext.example.com", Kind: "instance", Hibernate: false,
+ Socket: filepath.Join(p.SocketDir, "ext.example.com.sock"), DBPath: provision.DefaultDBPath(p.DataDir, "ext.example.com")}); err != nil {
+ t.Fatal(err)
+ }
+ out := byHost(t, p.Converge(context.Background(), ModeNested))
+ if out["custom.example.com"].Outcome != "skipped" || !strings.Contains(out["custom.example.com"].Reason, "foreign") {
+ t.Errorf("foreign row: %+v", out["custom.example.com"])
+ }
+ if _, listed := out["ext.example.com"]; listed {
+ t.Errorf("an unmanaged row was considered: %+v", out["ext.example.com"])
+ }
+}
+
+// A same-uid tenant can plant <data>/.instances/<iid> as a symlink before
+// the pass arrives. The pass refuses and moves nothing through it.
+func TestConvergeRefusesASymlinkedInstanceDirectory(t *testing.T) {
+ p, f := box(t)
+ elsewhere := t.TempDir()
+ raw := rawInstanceDir(p.DataDir, host)
+ if err := os.MkdirAll(filepath.Dir(raw), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(elsewhere, raw); err != nil {
+ t.Fatal(err)
+ }
+ r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
+ if r.Outcome != "refused" || !strings.Contains(r.Reason, "symlink") {
+ t.Fatalf("result = %+v", r)
+ }
+ assertFlatUntouched(t, p, host)
+ if entries, _ := os.ReadDir(elsewhere); len(entries) != 0 {
+ t.Errorf("files landed through the symlink: %v", entries)
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("fence left installed: %v", f.fences)
+ }
+}
+
+func TestSerializeWrapsJournalWriteToJournalRemoval(t *testing.T) {
+ p, _ := box(t)
+ var wrapped []string
+ p.Serialize = func(fn func()) {
+ for _, h := range []string{host, replica} {
+ if JournalPresent(p.DataDir, h) {
+ t.Errorf("%s journal present before the serialized section", h)
+ }
+ }
+ fn()
+ for _, h := range []string{host, replica} {
+ if JournalPresent(p.DataDir, h) {
+ t.Errorf("%s journal present after the serialized section", h)
+ }
+ }
+ wrapped = append(wrapped, "section")
+ }
+ p.Converge(context.Background(), ModeNested)
+ if len(wrapped) != 2 {
+ t.Fatalf("serialized sections = %d, want one per host", len(wrapped))
+ }
+}
+
+func TestRecoverCompletesEveryCrashBoundary(t *testing.T) {
+ // Drive the pass through a move with a seam that aborts after each
+ // step, then Recover, and assert the end state is the nested layout
+ // with no journal, every time.
+ for _, step := range []string{"after-journal", "mid-files", "after-files", "after-cas", "before-remove"} {
+ t.Run(step, func(t *testing.T) {
+ p, f := box(t)
+ p.abortAt = step
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ for _, h := range []string{host, replica} {
+ if !JournalPresent(p.DataDir, h) {
+ t.Fatalf("%s: no journal after aborting at %s", h, step)
+ }
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("an aborted move left its fence: %v", f.fences)
+ }
+ res := byHost(t, p.Recover(context.Background(), ""))
+ for _, h := range []string{host, replica} {
+ if res[h].Outcome != "recovered" {
+ t.Fatalf("recover %s = %+v", h, res[h])
+ }
+ if got := rowAt(t, p, h); got != endpoints(t, p, h, ModeNested) {
+ t.Errorf("%s row not nested after recovery from %s: %+v", h, step, got)
+ }
+ if b, err := os.ReadFile(filepath.Join(rawInstanceDir(p.DataDir, h), h+".db")); err != nil || string(b) != "db "+h {
+ t.Errorf("%s db not at destination after recovery from %s: %q %v", h, step, b, err)
+ }
+ if JournalPresent(p.DataDir, h) {
+ t.Errorf("%s journal remains after recovery", h)
+ }
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("recovery left a fence: %v", f.fences)
+ }
+ })
+ }
+}
+
+func TestRecoverRefusesAForgedJournalAndKeepsTheHostFenced(t *testing.T) {
+ p, _ := box(t)
+ j := &Journal{Host: host, Mode: ModeNested,
+ From: Endpoint{DB: "/data/console/console.db", Socket: filepath.Join(p.SocketDir, host+".sock")},
+ To: endpoints(t, p, host, ModeNested),
+ Phase: PhaseFiles}
+ for _, n := range artifactNames {
+ j.Artifacts = append(j.Artifacts, JournalArtifact{Name: n, Present: true})
+ }
+ if err := WriteJournal(p.DataDir, j); err != nil {
+ t.Fatal(err)
+ }
+ r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
+ if r.Outcome != "refused" || !strings.Contains(r.Reason, "not acting") {
+ t.Fatalf("forged journal: %+v", r)
+ }
+ if !JournalPresent(p.DataDir, host) {
+ t.Error("forged journal removed; the host must stay fenced for an operator")
+ }
+ if _, err := os.Stat(filepath.Join(p.DataDir, "console")); err == nil {
+ t.Error("something touched the forged source")
+ }
+ // Converge reports the host once, as the recovery refused it.
+ res := byHost(t, p.Converge(context.Background(), ModeNested))
+ if res[host].Outcome != "refused" {
+ t.Errorf("converge over a refused journal: %+v", res[host])
+ }
+}
+
+// A file at both ends of a move cannot come from a crash: the destination
+// check cleared every destination before the journal was written, and a
+// rename leaves a file at exactly one of them. Recovery refuses to choose.
+func TestRecoverRefusesAnArtifactAtBothEnds(t *testing.T) {
+ p, _ := box(t)
+ p.abortAt = "after-journal"
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ dst := filepath.Join(rawInstanceDir(p.DataDir, host), host+".db")
+ if err := os.WriteFile(dst, []byte("planted"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ r := byHost(t, p.Recover(context.Background(), ""))[host]
+ if r.Outcome != "refused" || !strings.Contains(r.Reason, "both") {
+ t.Fatalf("result = %+v", r)
+ }
+ if !JournalPresent(p.DataDir, host) {
+ t.Error("journal removed over an ambiguous artifact")
+ }
+ if b, _ := os.ReadFile(provision.DefaultDBPath(p.DataDir, host)); string(b) != "db "+host {
+ t.Error("the source database was touched")
+ }
+}
+
+// Design §3.3: a journal is completed to its own endpoint, and the host is
+// then converged to the fleet's intent before it is released. The fence
+// is held from the recovery through the convergence: nothing lifts it in
+// between, so the host never wakes at the journal's endpoint.
+func TestConvergeRecoversAJournalBeforeConvergingToANewIntent(t *testing.T) {
+ p, f := box(t)
+ p.abortAt = "after-files"
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ f.events = nil
+ // The intent changed to flat while the move was in flight: the journal
+ // completes to ITS endpoint (nested), then the host converges to flat.
+ res := byHost(t, p.Converge(context.Background(), ModeFlat))
+ for _, h := range []string{host, replica} {
+ if res[h].Outcome != "recovered" {
+ t.Errorf("%s: %+v", h, res[h])
+ }
+ assertFlatUntouched(t, p, h)
+ }
+ want := []string{
+ "fence " + host, "reload", "reload", "lift " + host,
+ "fence " + replica, "reload", "reload", "lift " + replica,
+ }
+ if !reflect.DeepEqual(f.events, want) {
+ t.Fatalf("events = %v\nwant %v", f.events, want)
+ }
+}
+
+// With no intent recorded, a journal is completed and the host released
+// where the journal ends.
+func TestRecoverWithNoIntentReleasesAtTheJournalsEndpoint(t *testing.T) {
+ p, f := box(t)
+ p.abortAt = "after-files"
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ for h, r := range byHost(t, p.Recover(context.Background(), "")) {
+ if r.Outcome != "recovered" || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) {
+ t.Errorf("%s: %+v at %+v", h, r, rowAt(t, p, h))
+ }
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("fences: %v", f.fences)
+ }
+}
+
+// When the agent starts and cannot read the fleet's intent, a revert in
+// flight is completed but kept fenced at phase done: releasing it flat
+// could let a forged revert run a nested host unisolated until an intent
+// is read. A move to nested is completed and released as usual.
+func TestRecoverHoldsACompletedRevertWhileTheIntentIsUnreadable(t *testing.T) {
+ p, f := box(t)
+ if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
+ t.Fatalf("setup: %+v", res)
+ }
+ p.abortAt = "after-cas"
+ p.Converge(context.Background(), ModeFlat)
+ p.abortAt = ""
+ res := byHost(t, p.Recover(context.Background(), IntentUnreadable))
+ for _, h := range []string{host, replica} {
+ if res[h].Outcome != "deferred" || !strings.Contains(res[h].Reason, "intent") {
+ t.Errorf("%s: %+v", h, res[h])
+ }
+ j, found, err := ReadJournal(p.DataDir, h)
+ if err != nil || !found || j.Phase != PhaseDone {
+ t.Fatalf("%s journal: %+v %v %v", h, j, found, err)
+ }
+ if rowAt(t, p, h) != endpoints(t, p, h, ModeFlat) {
+ t.Errorf("%s row not completed to flat", h)
+ }
+ }
+ if len(f.fences) != 0 {
+ t.Errorf("the journal is the fence; the activator's must not stay: %v", f.fences)
+ }
+ // Once the intent is read, the held host converges to it.
+ res = byHost(t, p.Recover(context.Background(), ModeNested))
+ for _, h := range []string{host, replica} {
+ if res[h].Outcome != "recovered" || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) || JournalPresent(p.DataDir, h) {
+ t.Errorf("%s after the intent: %+v at %+v", h, res[h], rowAt(t, p, h))
+ }
+ }
+
+ // A move toward nested is released even while the intent is unreadable.
+ q, _ := box(t)
+ q.abortAt = "after-files"
+ q.Converge(context.Background(), ModeNested)
+ q.abortAt = ""
+ for h, r := range byHost(t, q.Recover(context.Background(), IntentUnreadable)) {
+ if r.Outcome != "recovered" || JournalPresent(q.DataDir, h) {
+ t.Errorf("%s nested journal under an unreadable intent: %+v", h, r)
+ }
+ }
+}