amadan
appconfig, ops: config sync and the litestream generator leave a held host alone
SyncExec takes variadic options, so its call sites keep their shape; SkipFenced makes it write, remove and bounce nothing for a host the layout pass holds, and say so. FleetDBsWithSkipped takes a fenced hook, and a held host's row claims no database, so its file is named among the skipped. The litestream verb passes LayoutJournalPresent, which is the pass's own Lstat test on the journal path.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
5 files changed,
+198
−7
internal/appconfig/syncexec.go+35 −1internal/appconfig/syncexec_test.go+58 −0internal/ops/litestream.go+23 −4internal/ops/ops.go+1 −1internal/ops/ops_test.go+81 −1
diff --git a/internal/appconfig/syncexec.go b/internal/appconfig/syncexec.go| index 834a10f46..04d791b76 100644 |
| --- a/internal/appconfig/syncexec.go |
| +++ b/internal/appconfig/syncexec.go |
| @@ -92,7 +92,13 @@ func ConfigEnvPath(dataDir string, rt registry.Route) string { |
| // route whose file is REMOVED (manifest absent) gets no ack: there is |
| // nothing for a route to have adopted once its config is gone, and a |
| // removal is not one of spec §5's three delivery shapes. |
| -func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, dataDir string, bounce func(host string) error, appsDomain string, ack func(ConfigAck)) error { |
| +// |
| +// opts adjust one pass; see SkipFenced. |
| +func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv *ecdh.PrivateKey, dataDir string, bounce func(host string) error, appsDomain string, ack func(ConfigAck), opts ...SyncOption) error { |
| + var o syncOptions |
| + for _, opt := range opts { |
| + opt(&o) |
| + } |
| routes, err := reg.List() |
| if err != nil { |
| return err |
| @@ -408,6 +414,15 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv |
| if !in { |
| continue |
| } |
| + // A host the layout pass holds has its files between two |
| + // directories, and either path this pass could pick would be the |
| + // wrong one by the time the pass finishes. Nothing is written or |
| + // removed for it; the next pass after the move renders it where the |
| + // row then says. |
| + if o.fenced != nil && o.fenced(rt.Host) { |
| + fmt.Printf(" %s: layout migration in flight; bundle not written this pass\n", rt.Host) |
| + continue |
| + } |
| k := groupKey{release.RouteAccount(rt), rt.App, rt.Environment} |
| // Insist the group was really resolved rather than taking the map's |
| // zero value: a zero resolution reads as "absent", which on this |
| @@ -626,6 +641,25 @@ func SyncExec(ctx context.Context, reg *registry.Registry, s release.Store, priv |
| return nil |
| } |
| +// SyncOption adjusts one SyncExec pass. The options are variadic so the |
| +// ordinary call keeps its shape. |
| +type SyncOption func(*syncOptions) |
| + |
| +type syncOptions struct { |
| + fenced func(host string) bool |
| +} |
| + |
| +// SkipFenced makes SyncExec leave alone every in-scope route for which |
| +// fenced reports true: no bundle is written, removed or bounced for it, |
| +// and the pass says so. The agent passes the layout pass's journal |
| +// predicate. The pass also holds the agent's config-sync lock while it |
| +// moves a host, so this is the second net: it covers a journal that |
| +// outlives one pass, such as a completed revert held until the intent can |
| +// be read. |
| +func SkipFenced(fenced func(host string) bool) SyncOption { |
| + return func(o *syncOptions) { o.fenced = fenced } |
| +} |
| + |
| // execRouteBase is the lifecycle-managed-instance shape every exec route |
| // kind shares: the unit path's syncable() exclusions, narrowed to |
| // non-unit backing (unit routes belong to the root adopt path). Neither |
diff --git a/internal/appconfig/syncexec_test.go b/internal/appconfig/syncexec_test.go| index 6769eceee..333df7ba5 100644 |
| --- a/internal/appconfig/syncexec_test.go |
| +++ b/internal/appconfig/syncexec_test.go |
| @@ -176,6 +176,64 @@ func TestSyncExecWritesANestedRouteIntoItsInstanceDir(t *testing.T) { |
| } |
| } |
| +// TestSyncExecSkipsAHostTheLayoutPassHolds pins the second net under the |
| +// layout pass: a host with a migration journal is between two directories, |
| +// so SyncExec writes no bundle for it at either path, bounces nothing, and |
| +// says why, while a sibling on the same pass converges as usual. |
| +func TestSyncExecSkipsAHostTheLayoutPassHolds(t *testing.T) { |
| + ctx, s, reg, dd, priv, _ := execKit(t) |
| + pk, err := ParsePrivateKey(priv, "test") |
| + if err != nil { |
| + t.Fatalf("ParsePrivateKey: %v", err) |
| + } |
| + if err := SetEnv(ctx, s, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil { |
| + t.Fatalf("SetEnv: %v", err) |
| + } |
| + held, sibling := "inst.example.com", "other.example.com" |
| + if err := os.MkdirAll(provision.InstanceDir(dd, held), 0o755); err != nil { |
| + t.Fatalf("mkdir instance dir: %v", err) |
| + } |
| + for _, rt := range []registry.Route{ |
| + {Host: held, Kind: "instance", App: "eleven", Account: "bab", |
| + Status: "ready", DBPath: provision.NestedDBPath(dd, held), Socket: provision.NestedSocket(held)}, |
| + {Host: sibling, Kind: "instance", App: "eleven", Account: "bab", |
| + Status: "ready", Socket: "/run/carlos/" + sibling + ".sock"}, |
| + } { |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatalf("Upsert: %v", err) |
| + } |
| + } |
| + var bounced []string |
| + bounce := func(h string) error { bounced = append(bounced, h); return nil } |
| + var asked []string |
| + fenced := func(h string) bool { asked = append(asked, h); return h == held } |
| + out := captureStdout(t, func() { |
| + if err := SyncExec(ctx, reg, s, pk, dd, bounce, "", nil, SkipFenced(fenced)); err != nil { |
| + t.Fatalf("SyncExec: %v", err) |
| + } |
| + }) |
| + for _, path := range []string{ |
| + filepath.Join(provision.InstanceDir(dd, held), held+".config.env"), |
| + filepath.Join(dd, held+".config.env"), |
| + } { |
| + if _, err := os.Lstat(path); !os.IsNotExist(err) { |
| + t.Errorf("a held host's bundle was written at %s (err %v)", path, err) |
| + } |
| + } |
| + if got, err := os.ReadFile(filepath.Join(dd, sibling+".config.env")); err != nil || string(got) != "LOG_LEVEL=debug\n" { |
| + t.Errorf("sibling bundle = %q, %v", got, err) |
| + } |
| + if len(bounced) != 1 || bounced[0] != sibling { |
| + t.Errorf("bounced = %v, want exactly [%s]", bounced, sibling) |
| + } |
| + if !strings.Contains(out, held+": layout migration in flight; bundle not written this pass") { |
| + t.Errorf("the skip was not reported:\n%s", out) |
| + } |
| + if !slices.Contains(asked, held) || !slices.Contains(asked, sibling) { |
| + t.Errorf("fenced was asked about %v", asked) |
| + } |
| +} |
| + |
| // TestSyncExecFIFOAtANestedBundleDoesNotHangThePass pins the review fix: a |
| // nested row's existing bundle is read through instancefs, so a FIFO a |
| // tenant planted at that name (isolated-exec design §3.3, no tenant-planted |
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go| index 21396ef01..312aaddcd 100644 |
| --- a/internal/ops/litestream.go |
| +++ b/internal/ops/litestream.go |
| @@ -22,6 +22,7 @@ package ops |
| import ( |
| "bytes" |
| + "errors" |
| "fmt" |
| "os" |
| "os/exec" |
| @@ -32,6 +33,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/registry" |
| ) |
| @@ -321,14 +323,30 @@ func consoleDBPath(dataDir string) string { |
| } |
| func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) { |
| - dbs, _, err := FleetDBsWithSkipped(reg, dataDir) |
| + dbs, _, err := FleetDBsWithSkipped(reg, dataDir, nil) |
| return dbs, err |
| } |
| +// LayoutJournalPresent is the layout pass's "held" predicate for dataDir: a |
| +// host has a migration journal unless its journal path is definitely absent. |
| +// A journal that cannot be read is not proof there is none. It is the same |
| +// test internal/layout's JournalPresent makes, spelled out here so this |
| +// package does not depend on the pass. |
| +func LayoutJournalPresent(dataDir string) func(host string) bool { |
| + return func(host string) bool { |
| + _, err := os.Lstat(provision.LayoutJournalPath(dataDir, host)) |
| + return !errors.Is(err, os.ErrNotExist) |
| + } |
| +} |
| + |
| // FleetDBsWithSkipped is FleetDBs, plus the files under dataDir it declined to |
| // claim — see the "glob proposes, registry disposes" note inside. The verb |
| // prints them; nothing here is dropped without being named. |
| -func FleetDBsWithSkipped(reg *registry.Registry, dataDir string) (dbs []string, skipped []string, err error) { |
| +// |
| +// fenced, when non-nil, names hosts the layout pass holds mid-move. Such a |
| +// host's row does not claim its database, so the file is named among the |
| +// skipped instead of being replicated from a directory it is leaving. |
| +func FleetDBsWithSkipped(reg *registry.Registry, dataDir string, fenced func(host string) bool) (dbs []string, skipped []string, err error) { |
| routes, err := reg.List() |
| if err != nil { |
| return nil, nil, err |
| @@ -384,9 +402,10 @@ func FleetDBsWithSkipped(reg *registry.Registry, dataDir string) (dbs []string, |
| filepath.Clean(consoleDBPath(dataDir)): true, |
| } |
| for _, r := range routes { |
| - if r.DBPath != "" { |
| - claimed[filepath.Clean(r.DBPath)] = true |
| + if r.DBPath == "" || (fenced != nil && fenced(r.Host)) { |
| + continue |
| } |
| + claimed[filepath.Clean(r.DBPath)] = true |
| } |
| var globbed []string |
| for _, p := range matches { |
diff --git a/internal/ops/ops.go b/internal/ops/ops.go| index 6ff589464..dc24bb278 100644 |
| --- a/internal/ops/ops.go |
| +++ b/internal/ops/ops.go |
| @@ -114,7 +114,7 @@ func runLitestream(args []string) { |
| if err != nil { |
| log.Fatalf("litestream: open registry %s: %v", rp, err) |
| } |
| - dbs, skipped, err := FleetDBsWithSkipped(reg, dir) |
| + dbs, skipped, err := FleetDBsWithSkipped(reg, dir, LayoutJournalPresent(dir)) |
| if err != nil { |
| log.Fatalf("litestream: %v", err) |
| } |
diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go| index edfcb71ae..c16d4c5aa 100644 |
| --- a/internal/ops/ops_test.go |
| +++ b/internal/ops/ops_test.go |
| @@ -773,7 +773,7 @@ func TestFleetDBsSkipsUnclaimedFilesAndSaysSo(t *testing.T) { |
| t.Fatal(err) |
| } |
| - dbs, skipped, err := FleetDBsWithSkipped(reg, dir) |
| + dbs, skipped, err := FleetDBsWithSkipped(reg, dir, nil) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -790,3 +790,83 @@ func TestFleetDBsSkipsUnclaimedFilesAndSaysSo(t *testing.T) { |
| t.Fatalf("the unclaimed file was dropped SILENTLY — skipped = %v", skipped) |
| } |
| } |
| + |
| +// A host with a layout migration journal is between two directories, so |
| +// the generator claims neither of its database paths: its file is named |
| +// among the skipped rather than replicated from a place it is leaving. |
| +func TestFleetDBsSkipsAHostTheLayoutPassHolds(t *testing.T) { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + held := filepath.Join(dir, "held.example.com.db") |
| + kept := filepath.Join(dir, "kept.example.com.db") |
| + for _, p := range []string{held, kept} { |
| + if err := os.WriteFile(p, []byte("x"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + for _, rt := range []registry.Route{ |
| + {Host: "held.example.com", DBPath: held}, |
| + {Host: "kept.example.com", DBPath: kept}, |
| + } { |
| + if err := reg.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + journal := provision.LayoutJournalPath(dir, "held.example.com") |
| + if err := os.MkdirAll(filepath.Dir(journal), 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(journal, []byte("{}"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + dbs, skipped, err := FleetDBsWithSkipped(reg, dir, LayoutJournalPresent(dir)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if slices.Contains(dbs, held) { |
| + t.Fatalf("a held host's database joined the replication set: %v", dbs) |
| + } |
| + if !slices.Contains(skipped, held) { |
| + t.Fatalf("a held host's database was dropped silently: skipped = %v", skipped) |
| + } |
| + if !slices.Contains(dbs, kept) { |
| + t.Fatalf("a host with no journal was skipped: %v", dbs) |
| + } |
| + |
| + // No hook is the old behaviour: every route's database is claimed. |
| + dbs, _, err = FleetDBsWithSkipped(reg, dir, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Contains(dbs, held) { |
| + t.Fatalf("with no hook the held host was skipped: %v", dbs) |
| + } |
| +} |
| + |
| +// LayoutJournalPresent answers true for anything at the journal path that |
| +// is not definitely absent, a dangling symlink included: a journal that |
| +// cannot be read is not proof there is none. |
| +func TestLayoutJournalPresentIsLstatNotAbsent(t *testing.T) { |
| + dir := t.TempDir() |
| + present := LayoutJournalPresent(dir) |
| + if present("a.example.com") { |
| + t.Fatal("no journal directory, but a journal was reported") |
| + } |
| + link := provision.LayoutJournalPath(dir, "a.example.com") |
| + if err := os.MkdirAll(filepath.Dir(link), 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.Symlink(filepath.Join(dir, "nowhere"), link); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !present("a.example.com") { |
| + t.Fatal("a dangling symlink at the journal path was read as no journal") |
| + } |
| + if present("b.example.com") { |
| + t.Fatal("another host was reported held") |
| + } |
| +} |