amadan

carlos / platform Public

layout: hand a held host to its next journal in one rename, and keep the fence if that fails

The converge leg after recovery used to remove the held journal and then
create the new one, which left a gap, reachable by a same-uid tenant
that makes <data>/layout unwritable, in which nothing fenced the host
and the pass released it at the completed endpoint. ReplaceJournal now
checks the file is the held journal and renames a synced temp over it,
then fsyncs the directory. Any failure on the handover keeps the pass's
in-memory fence and reports refused. The presence scan moves inside the
serialized section with the journal write, so a bundle rendered just
before the lock is recorded and moved.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 91a73ba1af55f1310f735008c0b9b4528b7aca57 parent 03b2ea6
4 files changed, +332 −80
  • internal/layout/journal.go +67 −35
  • internal/layout/journal_test.go +63 −0
  • internal/layout/pass.go +83 −45
  • internal/layout/pass_linux_test.go +119 −0

CI passed — run details

diff --git a/internal/layout/journal.go b/internal/layout/journal.go
index fdb4a5344..2dee2e1d9 100644
--- a/internal/layout/journal.go
+++ b/internal/layout/journal.go
@@ -7,6 +7,7 @@ import (
"io"
"os"
"path/filepath"
+ "reflect"
"time"
"github.com/carlosframework/platform/internal/provision"
@@ -94,30 +95,10 @@ func WriteJournal(dataDir string, j *Journal) error {
if j.WrittenAt == "" {
j.WrittenAt = time.Now().UTC().Format(time.RFC3339)
}
- raw, err := json.MarshalIndent(j, "", " ")
+ tmpName, err := writeTemp(dir, j)
if err != nil {
return err
}
- tmp, err := os.CreateTemp(dir, "."+j.Host+".tmp-*")
- if err != nil {
- return err
- }
- tmpName := tmp.Name()
- if _, err := tmp.Write(append(raw, '\n')); err != nil {
- tmp.Close()
- os.Remove(tmpName)
- return err
- }
- if err := tmp.Sync(); err != nil {
- tmp.Close()
- os.Remove(tmpName)
- return err
- }
- if err := tmp.Close(); err != nil {
- os.Remove(tmpName)
- return err
- }
-
linkErr := os.Link(tmpName, provision.LayoutJournalPath(dataDir, j.Host))
// The temp file's only job was to become the final name; whether that
// succeeded or not, its own name must not linger (nothing reads a
@@ -180,42 +161,93 @@ func ReadJournal(dataDir, host string) (*Journal, bool, error) {
func UpdatePhase(dataDir string, j *Journal, p Phase) error {
next := *j
next.Phase = p
- raw, err := json.MarshalIndent(&next, "", " ")
+ dir := journalDir(dataDir)
+ name, err := writeTemp(dir, &next)
if err != nil {
return err
}
- dir := journalDir(dataDir)
+ if err := os.Rename(name, provision.LayoutJournalPath(dataDir, j.Host)); err != nil {
+ os.Remove(name)
+ return err
+ }
+ // Only now, with the rename itself having succeeded, does the phase
+ // this call promised become true of both the in-memory journal and
+ // the one on disk.
+ j.Phase = p
+ return syncDir(dir)
+}
+
+// writeTemp writes j to a fresh temp file in dir and syncs it, returning
+// its name. The sync comes before the caller links or renames the file to
+// the journal's canonical name: that makes these bytes visible there, so
+// they must already be durable, or a crash just after could leave the name
+// pointing at content that never landed.
+func writeTemp(dir string, j *Journal) (string, error) {
+ raw, err := json.MarshalIndent(j, "", " ")
+ if err != nil {
+ return "", err
+ }
tmp, err := os.CreateTemp(dir, "."+j.Host+".tmp-*")
if err != nil {
- return err
+ return "", err
}
name := tmp.Name()
if _, err := tmp.Write(append(raw, '\n')); err != nil {
tmp.Close()
os.Remove(name)
- return err
+ return "", err
}
- // Synced before the rename, not after: a rename makes these bytes
- // visible at the journal's canonical name, so they must already be
- // durable, or a crash between the rename and a Sync afterward could
- // leave that name pointing at content that never landed.
if err := tmp.Sync(); err != nil {
tmp.Close()
os.Remove(name)
- return err
+ return "", err
}
if err := tmp.Close(); err != nil {
os.Remove(name)
+ return "", err
+ }
+ return name, nil
+}
+
+// ReplaceJournal hands a host over from the journal the caller holds to a
+// new one in a single rename, so the host is never without a journal: a
+// crash, or a failure anywhere here, leaves either the whole old journal
+// or the whole new one at the canonical name. Removing the old one and
+// then creating the new one would leave a gap in which nothing fences the
+// host, and a same-uid tenant that can make <data>/layout unwritable at
+// the right moment could hold that gap open.
+//
+// The file replaced must be exactly held, as ReadJournal reads it back;
+// anything else there is refused and left alone. No exclusive create is
+// needed on the final name: the caller holds both the host's fence and
+// the section config sync is locked out of, and it has just checked what
+// it is replacing.
+func ReplaceJournal(dataDir string, held, next *Journal) error {
+ if next.Host != held.Host {
+ return fmt.Errorf("layout: replacing %s's journal with one for %s", held.Host, next.Host)
+ }
+ cur, found, err := ReadJournal(dataDir, held.Host)
+ if err != nil {
return err
}
- if err := os.Rename(name, provision.LayoutJournalPath(dataDir, j.Host)); err != nil {
+ if !found {
+ return fmt.Errorf("layout: %s: no journal to replace", held.Host)
+ }
+ if !reflect.DeepEqual(cur, held) {
+ return fmt.Errorf("layout: %s: the journal on disk is not the one this pass holds; not replacing it", held.Host)
+ }
+ if next.WrittenAt == "" {
+ next.WrittenAt = time.Now().UTC().Format(time.RFC3339)
+ }
+ dir := journalDir(dataDir)
+ name, err := writeTemp(dir, next)
+ if err != nil {
+ return err
+ }
+ if err := os.Rename(name, provision.LayoutJournalPath(dataDir, next.Host)); err != nil {
os.Remove(name)
return err
}
- // Only now, with the rename itself having succeeded, does the phase
- // this call promised become true of both the in-memory journal and
- // the one on disk.
- j.Phase = p
return syncDir(dir)
}
diff --git a/internal/layout/journal_test.go b/internal/layout/journal_test.go
index 01dea7a32..e14980f93 100644
--- a/internal/layout/journal_test.go
+++ b/internal/layout/journal_test.go
@@ -251,3 +251,66 @@ func TestJournalPresent(t *testing.T) {
t.Error("journal removed, JournalPresent = true")
}
}
+
+// ReplaceJournal hands a host over from one journal to the next in one
+// rename: at every instant the final name holds one whole journal, so the
+// host is never without one.
+func TestReplaceJournalSwapsContentsInOneStep(t *testing.T) {
+ dd := t.TempDir()
+ old := journalFor(t, dd, ModeFlat)
+ if err := WriteJournal(dd, old); err != nil {
+ t.Fatal(err)
+ }
+ if err := UpdatePhase(dd, old, PhaseDone); err != nil {
+ t.Fatal(err)
+ }
+ next := journalFor(t, dd, ModeNested)
+ if err := ReplaceJournal(dd, old, next); err != nil {
+ t.Fatalf("ReplaceJournal: %v", err)
+ }
+ got, found, err := ReadJournal(dd, "h.example.com")
+ if err != nil || !found || got.Mode != ModeNested || got.Phase != PhaseFiles {
+ t.Fatalf("after replace: %+v %v %v", got, found, err)
+ }
+ if left := tempFilesIn(t, journalDir(dd)); len(left) != 0 {
+ t.Errorf("temp files left: %v", left)
+ }
+ if entries, _ := os.ReadDir(journalDir(dd)); len(entries) != 1 {
+ t.Errorf("journal dir holds %d entries, want exactly one", len(entries))
+ }
+}
+
+// The file replaced must be the journal the caller holds: anything else
+// there (a newer phase, another direction, nothing at all) is refused and
+// left exactly as it is.
+func TestReplaceJournalRefusesAnythingButTheHeldJournal(t *testing.T) {
+ dd := t.TempDir()
+ held := journalFor(t, dd, ModeFlat)
+ next := journalFor(t, dd, ModeNested)
+ if err := ReplaceJournal(dd, held, next); err == nil {
+ t.Fatal("replaced a journal that is not there")
+ }
+ if JournalPresent(dd, "h.example.com") {
+ t.Fatal("a journal was created where none was held")
+ }
+ if err := WriteJournal(dd, held); err != nil {
+ t.Fatal(err)
+ }
+ stale := *held
+ stale.Phase = PhaseDone
+ if err := ReplaceJournal(dd, &stale, next); err == nil {
+ t.Fatal("replaced a journal whose contents differ from the held one")
+ }
+ other := journalFor(t, dd, ModeNested)
+ other.Host = "o.example.com"
+ if err := ReplaceJournal(dd, held, other); err == nil {
+ t.Fatal("replaced one host's journal with another host's")
+ }
+ got, _, _ := ReadJournal(dd, "h.example.com")
+ if got.Mode != ModeFlat || got.Phase != PhaseFiles {
+ t.Fatalf("the held journal changed: %+v", got)
+ }
+ if left := tempFilesIn(t, journalDir(dd)); len(left) != 0 {
+ t.Errorf("temp files left: %v", left)
+ }
+}
diff --git a/internal/layout/pass.go b/internal/layout/pass.go
index b12e49087..c8ec0fbe4 100644
--- a/internal/layout/pass.go
+++ b/internal/layout/pass.go
@@ -308,28 +308,27 @@ func (p *Pass) release(host string, gen uint64) {
//
// prior is a completed journal recovery is still holding the host with,
// or nil. That journal is this pass's own, so it does not count as the
-// host being fenced here, and it is removed only immediately before the
-// new journal's exclusive write: until then it keeps the host fenced, so
-// a refusal anywhere in between leaves the host held, not released.
-// replaced reports whether prior was removed, and so whether the host's
-// fence is now the new journal (or nothing, after a completed move).
-func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, replaced bool) {
+// host being fenced here. It is replaced by the new journal in one rename
+// (ReplaceJournal), so the host holds one journal or the other at every
+// instant, and a refusal before the handover leaves it held by prior. The
+// handover result says which of those happened.
+func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, h handover) {
fresh, err := p.Registry.Get(host)
if err != nil {
- return Result{host, "refused", "row: " + err.Error()}, false
+ return Result{host, "refused", "row: " + err.Error()}, handoverNone
}
if fresh == nil {
- return Result{host, "refused", "row vanished under the fence"}, false
+ return Result{host, "refused", "row vanished under the fence"}, handoverNone
}
if !ready(*fresh) {
- return Result{host, "refused", "status " + fresh.Status + " under the fence"}, false
+ return Result{host, "refused", "status " + fresh.Status + " under the fence"}, handoverNone
}
if !activator.ActivatorManaged(*fresh) {
- return Result{host, "refused", "no longer activator-managed under the fence"}, false
+ return Result{host, "refused", "no longer activator-managed under the fence"}, handoverNone
}
routes, err := p.Registry.List()
if err != nil {
- return Result{host, "refused", "list routes: " + err.Error()}, false
+ return Result{host, "refused", "list routes: " + err.Error()}, handoverNone
}
want, from := activator.LayoutNested, activator.LayoutFlat
if mode == ModeFlat {
@@ -341,28 +340,48 @@ func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, replac
}
switch layout, reason := activator.Classify(*fresh, in); layout {
case want:
- return Result{host, "unchanged", ""}, false
+ return Result{host, "unchanged", ""}, handoverNone
case from:
default:
- return Result{host, "refused", fmt.Sprintf("%s under the fence: %s", layout, reason)}, false
+ return Result{host, "refused", fmt.Sprintf("%s under the fence: %s", layout, reason)}, handoverNone
}
plan, err := PlanFor(p.DataDir, p.socketDir(), host, mode)
if err != nil {
- return Result{host, "refused", err.Error()}, false
+ return Result{host, "refused", err.Error()}, handoverNone
}
if err := Preflight(plan, routes); err != nil {
- return Result{host, "refused", err.Error()}, false
+ return Result{host, "refused", err.Error()}, handoverNone
}
// 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()}, false
+ return Result{host, "refused", err.Error()}, handoverNone
}
aside, err := DestinationCheck(plan, p.now)
if err != nil {
- return Result{host, "refused", err.Error()}, false
+ return Result{host, "refused", err.Error()}, handoverNone
}
+ // The presence scan runs inside the section config sync is locked out
+ // of, with the journal write: a bundle rendered between a scan outside
+ // it and the journal would be recorded absent and left at the old path.
+ p.serialize(func() {
+ present, err := presence(plan)
+ if err != nil {
+ res, h = Result{host, "refused", err.Error()}, handoverNone
+ return
+ }
+ res, h = p.commit(plan, plan.Journal(present), aside, prior)
+ })
+ if res.Outcome == "migrated" || res.Outcome == "reverted" {
+ p.reload()
+ }
+ return res, h
+}
+
+// presence records which artifacts are at their sources, in artifactNames
+// order, and refuses a plan whose database is not.
+func presence(plan *Plan) ([]bool, error) {
present := make([]bool, len(plan.Artifacts))
for i, a := range plan.Artifacts {
// Nothing is ever renamed for the socket: the listener is gone
@@ -372,49 +391,53 @@ func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, replac
}
ok, err := lstatPresent(a.Src)
if err != nil {
- return Result{host, "refused", err.Error()}, false
+ return nil, err
}
if a.Required && !ok {
- return Result{host, "refused", fmt.Sprintf("%s is missing at %s", a.Name, a.Src)}, false
+ return nil, fmt.Errorf("%s is missing at %s", a.Name, a.Src)
}
present[i] = ok
plan.Artifacts[i].Present = ok
}
- p.serialize(func() { res, replaced = p.commit(plan, plan.Journal(present), aside, prior) })
- if res.Outcome == "migrated" || res.Outcome == "reverted" {
- p.reload()
- }
- return res, replaced
+ return present, nil
}
// 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, prior *Journal) (Result, bool) {
+func (p *Pass) commit(plan *Plan, j *Journal, aside []string, prior *Journal) (Result, handover) {
host := plan.Host
- replaced := false
- if prior != nil {
- if err := RemoveJournal(p.DataDir, host); err != nil {
- return Result{host, "refused", err.Error()}, false
+ if prior == nil {
+ if err := WriteJournal(p.DataDir, j); err != nil {
+ return Result{host, "refused", err.Error()}, handoverNone
}
- replaced = true
+ return p.commitWritten(plan, j, aside), handoverNone
}
- if err := WriteJournal(p.DataDir, j); err != nil {
- // Put the held journal back: without it, or the new one, nothing
- // would keep the host from being released at an endpoint that is
- // not the intent.
- if replaced {
- if rerr := WriteJournal(p.DataDir, prior); rerr != nil {
- return Result{host, "refused", fmt.Sprintf("%v; restoring the held journal: %v", err, rerr)}, true
- }
- }
- return Result{host, "refused", err.Error()}, false
+ if err := ReplaceJournal(p.DataDir, prior, j); err != nil {
+ // Whatever is on disk now, the pass cannot vouch that it fences
+ // the host, so the caller keeps its own fence instead.
+ return Result{host, "refused", err.Error()}, handoverFailed
}
- r := p.commitWritten(plan, j, aside)
- return r, replaced
+ return p.commitWritten(plan, j, aside), handoverDone
}
+// handover is what became of the journal recovery held a host with when
+// the converge leg ran.
+type handover int
+
+const (
+ // handoverNone: the leg stopped before the handover; the held journal
+ // is untouched and still fences the host.
+ handoverNone handover = iota
+ // handoverDone: the new journal replaced it; the host is fenced by the
+ // new journal until that move completes.
+ handoverDone
+ // handoverFailed: the replacement itself failed, and what is on disk
+ // is not known to fence the host. The in-memory fence must stay.
+ handoverFailed
+)
+
// commitWritten is commit after the journal is on disk.
func (p *Pass) commitWritten(plan *Plan, j *Journal, aside []string) Result {
host := plan.Host
@@ -624,6 +647,7 @@ func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]r
return Result{host, "refused", notActing(err)}
}
gen := p.generation()
+ keepFence := false
r, fence := p.quiesce(ctx, host, gen)
switch fence {
case fenceNone:
@@ -633,7 +657,11 @@ func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]r
return Result{host, "deferred", "fenced by another operation and not yet stopped"}
}
case fenceOurs:
- defer p.release(host, gen)
+ defer func() {
+ if !keepFence {
+ 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.
@@ -685,11 +713,21 @@ func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]r
return Result{host, "recovered", fmt.Sprintf("completed to %s; not converging to nested: isolation not ready: %v", j.Mode, gerr)}
}
}
- next, replaced := p.shift(host, intent, j)
+ next, h := p.shift(host, intent, j)
switch {
case next.Outcome == "migrated" || next.Outcome == "reverted":
return Result{host, "recovered", fmt.Sprintf("completed to %s, then converged to %s", j.Mode, intent)}
- case !replaced:
+ case h == handoverFailed:
+ // Nothing on disk is known to fence the host, so the pass's own
+ // fence stays and the host stays down until an operator or an
+ // agent restart acts. Releasing it here would run it at the
+ // completed endpoint, which after a forged revert is a nested
+ // host running flat.
+ keepFence = true
+ reason := fmt.Sprintf("completed to %s; handing over to the move to %s failed: %s; fence kept", j.Mode, intent, next.Reason)
+ p.logf("layout: %s: refused -> %s", host, reason)
+ return Result{host, "refused", reason}
+ case h == handoverNone:
reason := fmt.Sprintf("completed to %s; journal kept at done: converging to %s: %s %s", j.Mode, intent, next.Outcome, next.Reason)
p.logf("layout: %s: held -> %s", host, reason)
return Result{host, "deferred", reason}
diff --git a/internal/layout/pass_linux_test.go b/internal/layout/pass_linux_test.go
index 908b90735..1978b138c 100644
--- a/internal/layout/pass_linux_test.go
+++ b/internal/layout/pass_linux_test.go
@@ -1118,3 +1118,122 @@ func TestConvergeRefusesAHostWithNoLocalDatabaseWithoutFencingIt(t *testing.T) {
t.Errorf("the other host: %+v", res[replica])
}
}
+
+// crashedRevert leaves both hosts nested-then-mid-revert: each has a flat
+// journal at phase files with its files already at the flat destination.
+func crashedRevert(t *testing.T, p *Pass) {
+ t.Helper()
+ if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
+ t.Fatalf("setup: %+v", res)
+ }
+ p.abortAt = "after-files"
+ p.Converge(context.Background(), ModeFlat)
+ p.abortAt = ""
+}
+
+// The handover from the completed journal to the converge leg's journal
+// is one rename, and anything going wrong on it leaves the host down: the
+// held journal stays whole, the pass's fence stays installed, and nothing
+// is released. A same-uid tenant can make <data>/layout unwritable at the
+// right moment, so this is reachable, not only a disk fault.
+func TestRecoverKeepsTheFenceWhenTheHandoverFails(t *testing.T) {
+ if os.Geteuid() == 0 {
+ t.Skip("root ignores the directory mode this test relies on")
+ }
+ p, f := box(t)
+ crashedRevert(t, p)
+ dir := journalDir(p.DataDir)
+ calls := 0
+ p.Serialize = func(fn func()) {
+ calls++
+ // The second section on the first host is the handover: make the
+ // temp file impossible to create there, as a racing tenant could.
+ if calls == 2 {
+ if err := os.Chmod(dir, 0o500); err != nil {
+ t.Fatal(err)
+ }
+ defer os.Chmod(dir, 0o700)
+ }
+ fn()
+ }
+ f.events = nil
+ r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
+ if r.Outcome != "refused" || !strings.Contains(r.Reason, "fence kept") {
+ t.Fatalf("result = %+v", r)
+ }
+ j, found, err := ReadJournal(p.DataDir, host)
+ if err != nil || !found || j.Mode != ModeFlat || j.Phase != PhaseDone {
+ t.Fatalf("held journal not intact: %+v %v %v", j, found, err)
+ }
+ if f.fences[host] == 0 {
+ t.Error("the pass's fence was lifted after a failed handover")
+ }
+ for _, e := range f.events {
+ if e == "lift "+host {
+ t.Errorf("released after a failed handover: %v", f.events)
+ }
+ }
+ if left := tempFilesIn(t, dir); len(left) != 0 {
+ t.Errorf("temp files left: %v", left)
+ }
+}
+
+// A successful handover leaves exactly one journal, the converge leg's
+// own, and no temp file.
+func TestRecoverHandsOverToTheNewJournalInOneStep(t *testing.T) {
+ p, _ := box(t)
+ crashedRevert(t, p)
+ p.abortAt = "after-journal"
+ p.Recover(context.Background(), ModeNested)
+ p.abortAt = ""
+ j, found, err := ReadJournal(p.DataDir, host)
+ if err != nil || !found || j.Mode != ModeNested || j.Phase != PhaseFiles {
+ t.Fatalf("journal after the handover: %+v %v %v", j, found, err)
+ }
+ entries, err := os.ReadDir(journalDir(p.DataDir))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var names []string
+ for _, e := range entries {
+ names = append(names, e.Name())
+ }
+ want := []string{host + ".json", replica + ".json"}
+ if !reflect.DeepEqual(names, want) {
+ t.Fatalf("journal dir = %v, want %v", names, want)
+ }
+}
+
+// The presence scan runs inside the section config sync is locked out of,
+// so a bundle rendered just before the pass took the lock is recorded and
+// moved, never stranded at the flat path behind a journal that calls it
+// absent.
+func TestThePresenceScanRunsInsideTheSerializedSection(t *testing.T) {
+ p, _ := box(t)
+ rendered := false
+ p.Serialize = func(fn func()) {
+ if !rendered {
+ // Config sync's last render, landing while it still held the
+ // lock the pass is now taking.
+ for _, h := range []string{host, replica} {
+ if err := os.WriteFile(filepath.Join(p.DataDir, h+".config.env"), []byte("A=1\n"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ rendered = true
+ }
+ fn()
+ }
+ 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])
+ }
+ if _, err := os.Lstat(filepath.Join(p.DataDir, h+".config.env")); !errors.Is(err, os.ErrNotExist) {
+ t.Errorf("%s bundle stranded at the flat path", h)
+ }
+ if _, err := os.Stat(filepath.Join(rawInstanceDir(p.DataDir, h), h+".config.env")); err != nil {
+ t.Errorf("%s bundle not moved: %v", h, err)
+ }
+ }
+}