amadan
Sidecar isolation Task 8 (1/4): the layout pass moves worker rows (D8, D9, D11)
The pass's eligibility widens from ActivatorManaged to ActivatorManaged || WorkerManaged, the latter only while the agent-start workers readiness holds (Pass.WorkersReady, nil = false), at all three sites: Converge, RestoreHolds and the re-read under the fence. A worker journal is still recovered without it: completing a journal is exact. PlanFor learns the worker shape from the host's suffix (provision.SidecarSuffix), never the row's Kind: both endpoints carry Socket "" and the socket artifact is recorded absent on both sides. A row whose Kind disagrees with the suffix (WorkerRowAgrees) is refused before any stop, in Converge and in recovery, like any journal mismatch. D8 rev 2's missing-database rule (DatabaseAbsent): the db stops being Required when it and every companion (db-wal, db-shm, litestream-shadow, corrupt, <db>-litestream) are absent at both endpoints; the journal then records it absent and the rest moves. A companion without its database refuses with the names. Validate no longer refuses a journal whose db is recorded absent; recovery's existing refusal of anything recorded absent that exists at either end is what keeps a present database from being stranded, now pinned. D9's running-replicator check: boxReplicatorHolds (replicator.go) decides from /etc/litestream.yml and carlos-litestream's ActiveState, SubState, ExecMainStartTimestamp, MainPID and cgroup through injected observations; BoxReplicatorHolds is the production pair (systemctl show --timestamp=unix, read-only, and cgroup.events under the unit's ControlGroup). The pass asks it for EVERY worker row (R4-1) before the stop and again under the fence inside the serialized section, right before the journal and the renames; a hold defers the move with the spec's reason (a placeholder constant in layout_copy.go for Task 12). Nil holds.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
8 files changed,
+1574
−36
internal/layout/journal.go+9 −7internal/layout/journal_test.go+13 −11internal/layout/layout_copy.go+9 −0internal/layout/pass.go+105 −17internal/layout/pass_worker_test.go+815 −0internal/layout/plan.go+121 −1internal/layout/replicator.go+254 −0internal/layout/replicator_test.go+248 −0
diff --git a/internal/layout/journal.go b/internal/layout/journal.go| index a8a66255f..ae8457fcc 100644 |
| --- a/internal/layout/journal.go |
| +++ b/internal/layout/journal.go |
| @@ -337,13 +337,15 @@ func Validate(j *Journal, dataDir, socketDir string, row registry.Route) (atFrom |
| return false, fmt.Errorf("layout: %s: journal artifact %d is %q, expected %q", j.Host, i, a.Name, artifactNames[i]) |
| } |
| } |
| - // The db is the one artifact recovery cannot proceed without: every |
| - // other artifact missing is a smaller, optional file the pass would |
| - // simply skip, but a journal claiming its db was never even there |
| - // describes a move that cannot have happened. |
| - if !j.Artifacts[0].Present { |
| - return false, fmt.Errorf("layout: %s: journal's db artifact is not marked present; not acting", j.Host) |
| - } |
| + // A journal may record its db absent (sidecar isolation D8 rev 2): a |
| + // row whose database was never on this box moves its config and extras |
| + // without one, and the next wake restores from the bucket. The pass |
| + // only writes that when the database and every companion were absent |
| + // at both endpoints (DatabaseAbsent), and recovery holds a journal to |
| + // it: complete refuses, not acting, when anything recorded absent |
| + // exists at either end, so a forged journal that claims a present |
| + // database was never there cannot strand it. |
| + // |
| // The extra names pass the same rule a scan applies, and an extra is |
| // never absent: the plan carries one because it was seen. |
| if err := p.AddExtras(j.extraNames()); err != nil { |
diff --git a/internal/layout/journal_test.go b/internal/layout/journal_test.go| index 00e5f72cc..fd712ba71 100644 |
| --- a/internal/layout/journal_test.go |
| +++ b/internal/layout/journal_test.go |
| @@ -198,20 +198,22 @@ func TestValidateRefusesAWrongArtifactCount(t *testing.T) { |
| } |
| } |
| -// TestValidateRequiresTheDBArtifactPresent pins that a forged journal |
| -// cannot claim its db was never there and still be accepted: everything |
| -// else in an optional artifact list can be absent and the pass simply |
| -// skips it, but a journal with no db describes a move that cannot have |
| -// happened, whatever its endpoints say. |
| -func TestValidateRequiresTheDBArtifactPresent(t *testing.T) { |
| +// TestValidateAcceptsADBRecordedAbsent pins sidecar isolation D8 rev 2: a |
| +// journal may record its db absent (a standby row that never held the |
| +// database moves its config and extras without one), so Validate no longer |
| +// refuses it on shape. The data safety moved to recovery: complete refuses |
| +// any artifact recorded absent that exists at either end (pinned by |
| +// TestRecoverRefusesAForgedJournalClaimingThePresentDBAbsent), so a forged |
| +// journal cannot strand a database that is there. |
| +func TestValidateAcceptsADBRecordedAbsent(t *testing.T) { |
| dd := t.TempDir() |
| j := journalFor(t, dd, Mode("nested")) |
| row := registry.Route{Host: "h.example.com", DBPath: j.From.DB, Socket: j.From.Socket} |
| - forged := *j |
| - forged.Artifacts = append([]JournalArtifact(nil), j.Artifacts...) |
| - forged.Artifacts[0].Present = false |
| - if _, err := Validate(&forged, dd, "/run/carlos", row); err == nil { |
| - t.Error("a journal whose db artifact is not present was accepted") |
| + standby := *j |
| + standby.Artifacts = append([]JournalArtifact(nil), j.Artifacts...) |
| + standby.Artifacts[0].Present = false |
| + if atFrom, err := Validate(&standby, dd, "/run/carlos", row); err != nil || !atFrom { |
| + t.Errorf("a journal recording its db absent: atFrom %v, err %v", atFrom, err) |
| } |
| } |
diff --git a/internal/layout/layout_copy.go b/internal/layout/layout_copy.go| new file mode 100644 |
| index 000000000..72293670b |
| --- /dev/null |
| +++ b/internal/layout/layout_copy.go |
| @@ -0,0 +1,9 @@ |
| +package layout |
| + |
| +// User-facing strings of the layout pass that the sidecar isolation spec |
| +// marks "copy to review". Placeholders until Task 12 collects, reviews and |
| +// fills them; the words here are the spec's own. |
| + |
| +// replicatorHoldsCopy is the reason a worker's move is deferred while the |
| +// box-wide replicator may still hold its database (spec §3.9, D9). |
| +const replicatorHoldsCopy = "the box-wide replicator may still hold this database" // copy: placeholder, Task 12 |
diff --git a/internal/layout/pass.go b/internal/layout/pass.go| index 26a3870cd..2630b8ed0 100644 |
| --- a/internal/layout/pass.go |
| +++ b/internal/layout/pass.go |
| @@ -62,6 +62,25 @@ type Pass struct { |
| // cannot establish the namespace does not start nested tenants, so it |
| // does not make any either. Nil means ungated. |
| NestedOK func() error |
| + // WorkersReady is the box's workers readiness as the agent found it at |
| + // its start (sidecar isolation D11: CARLOS_NESTED_WORKERS=on and both |
| + // the running build and carlos.prev at or above the worker layout |
| + // floor). While it does not hold, worker rows are not the pass's to |
| + // move in either direction; they are still recovered from a journal, |
| + // since completing a journal is exact. Nil means not ready. It is not |
| + // the whole gate: a move to nested still needs the fleet's intent |
| + // (Converge's mode) and NestedOK, as for instance rows. |
| + WorkersReady func() bool |
| + // Replicator reports whether the box-wide litestream may still hold a |
| + // worker's database open (sidecar isolation D9, R3-3): its config |
| + // names the file, or the daemon is in a state the pass cannot place |
| + // against that config. Asked for EVERY worker row's move (R4-1), |
| + // before the host is stopped and again under the fence immediately |
| + // before the first rename; a hold defers the move with the reason, |
| + // retried on every pass. Nil holds every worker move: a box with no |
| + // observation wired cannot vouch for the daemon. Never asked for an |
| + // instance row. Production wires BoxReplicatorHolds. |
| + Replicator func(dbPath string) (holds bool, reason string) |
| // 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. |
| @@ -229,6 +248,36 @@ func deadline(ctx context.Context) (context.Context, context.CancelFunc) { |
| // worth moving. |
| func ready(rt registry.Route) bool { return rt.Status == "" || rt.Status == "ready" } |
| +// workersReady is WorkersReady with nil read as not ready. |
| +func (p *Pass) workersReady() bool { return p.WorkersReady != nil && p.WorkersReady() } |
| + |
| +// managed is the pass's eligibility (design §8.2 step 1, widened by |
| +// sidecar isolation D8): an activator-managed instance row always, a |
| +// native worker row only while the box's workers readiness holds. |
| +func (p *Pass) managed(rt registry.Route) bool { |
| + return activator.ActivatorManaged(rt) || (p.workersReady() && activator.WorkerManaged(rt)) |
| +} |
| + |
| +// replicatorHolds asks Replicator about a worker plan's source database. |
| +// An instance plan never holds: the box-wide replicator never names an |
| +// activator-managed instance's database (ops.ActivatorReplicated). |
| +func (p *Pass) replicatorHolds(plan *Plan) (bool, string) { |
| + if !plan.Worker { |
| + return false, "" |
| + } |
| + if p.Replicator == nil { |
| + return true, "no replicator observation is wired on this box" |
| + } |
| + return p.Replicator(plan.From.DB) |
| +} |
| + |
| +// replicatorDeferred is the result for a worker move the box-wide |
| +// replicator's state defers (D9): not a refusal and not a hold, since no |
| +// tenant brought it about; the next pass asks again. |
| +func replicatorDeferred(host, why string) Result { |
| + return Result{host, "deferred", replicatorHoldsCopy + ": " + why} |
| +} |
| + |
| // 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 |
| @@ -261,7 +310,7 @@ func (p *Pass) Converge(ctx context.Context, mode Mode) []Result { |
| sort.Slice(routes, func(i, j int) bool { return routes[i].Host < routes[j].Host }) |
| migrated := false |
| for _, rt := range routes { |
| - if !activator.ActivatorManaged(rt) || seen[rt.Host] { |
| + if !p.managed(rt) || seen[rt.Host] { |
| continue |
| } |
| if ctx.Err() != nil { |
| @@ -287,6 +336,9 @@ func (p *Pass) convergeHost(ctx context.Context, rt registry.Route, routes []reg |
| if !ready(rt) { |
| return Result{rt.Host, "deferred", "status " + rt.Status} |
| } |
| + if err := WorkerRowAgrees(rt.Host, rt); err != nil { |
| + return Result{rt.Host, "refused", err.Error()} |
| + } |
| layout, reason := p.Classify(rt, routes) |
| switch layout { |
| case activator.LayoutFenced, activator.LayoutConflict: |
| @@ -345,14 +397,17 @@ func (p *Pass) migrateHost(ctx context.Context, host string, routes []registry.R |
| if err := Preflight(plan, routes); err != nil { |
| return Result{host, "refused", err.Error()} |
| } |
| - // A host whose database is not on this box cannot move. Finding that |
| - // out must not cost it a stop, and the pass directories and a |
| - // destination check, on every pass. The same check under the fence |
| - // stays the authority. |
| - if db := plan.Artifacts[0]; db.Required { |
| - if ok, err := lstatPresent(db.Src); err != nil || !ok { |
| - return Result{host, "refused", fmt.Sprintf("db is missing at %s", db.Src)} |
| - } |
| + // A host whose database's history is here without the database cannot |
| + // move (DatabaseAbsent). Finding that out must not cost it a stop, and |
| + // the pass directories and a destination check, on every pass. The same |
| + // check under the fence (presence) stays the authority. |
| + if _, err := DatabaseAbsent(plan); err != nil { |
| + return Result{host, "refused", err.Error()} |
| + } |
| + // Nor must a worker be stopped for a move the box-wide replicator's |
| + // state defers (D9). The read under the fence is the one that counts. |
| + if holds, why := p.replicatorHolds(plan); holds { |
| + return replicatorDeferred(host, why) |
| } |
| r, fence, gen := p.quiesce(ctx, host, p.generation()) |
| if fence != fenceOurs { |
| @@ -526,10 +581,14 @@ func (p *Pass) restoreHold(ctx context.Context, host string) Result { |
| if err != nil { |
| return Result{host, "refused", "list routes: " + err.Error()} |
| } |
| - if rt == nil || !activator.ActivatorManaged(*rt) { |
| + if rt == nil || !p.managed(*rt) { |
| removeHoldMarker(p.DataDir, host) |
| return Result{host, "released", "hold marker dropped: the row is gone or no longer managed"} |
| } |
| + if err := WorkerRowAgrees(host, *rt); err != nil { |
| + removeHoldMarker(p.DataDir, host) |
| + return Result{host, "released", "hold marker dropped: " + err.Error()} |
| + } |
| if layout, _ := p.Classify(*rt, routes); layout != activator.LayoutFlat { |
| removeHoldMarker(p.DataDir, host) |
| return Result{host, "released", "hold marker dropped: the row is " + string(layout)} |
| @@ -744,8 +803,11 @@ func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, h hand |
| if !ready(*fresh) { |
| return Result{host, "refused", "status " + fresh.Status + " under the fence"}, handoverNone, false |
| } |
| - if !activator.ActivatorManaged(*fresh) { |
| - return Result{host, "refused", "no longer activator-managed under the fence"}, handoverNone, false |
| + if !p.managed(*fresh) { |
| + return Result{host, "refused", "no longer managed under the fence"}, handoverNone, false |
| + } |
| + if err := WorkerRowAgrees(host, *fresh); err != nil { |
| + return Result{host, "refused", err.Error()}, handoverNone, false |
| } |
| routes, err := p.Registry.List() |
| if err != nil { |
| @@ -794,6 +856,14 @@ func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, h hand |
| res, h, hold = Result{host, "refused", err.Error()}, handoverNone, true |
| return |
| } |
| + // The running replicator, not the file, decides a worker's move |
| + // (D9): read here, under the fence and inside the serialized |
| + // section, so nothing of the host's is renamed between this |
| + // observation and the renames the journal below commits to. |
| + if holds, why := p.replicatorHolds(plan); holds { |
| + res, h, hold = replicatorDeferred(host, why), handoverNone, false |
| + return |
| + } |
| res, h = p.commit(plan, plan.Journal(present), aside, prior) |
| }) |
| if res.Outcome == "migrated" || res.Outcome == "reverted" { |
| @@ -803,11 +873,20 @@ func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, h hand |
| } |
| // presence records which artifacts are at their sources, in artifactNames |
| -// order, and refuses a plan whose database is not. It then adds to the |
| -// plan every entry the app made for the host (ScanExtras; hosts is the |
| -// box's rows, which decide ownership), so the journal written from the |
| -// plan names them and the move carries them. |
| +// order, and refuses a plan whose database is not, unless the database and |
| +// every companion are absent at both endpoints (DatabaseAbsent, D8 rev 2), |
| +// in which case the database is recorded absent and the rest moves. It |
| +// then adds to the plan every entry the app made for the host |
| +// (ScanExtras; hosts is the box's rows, which decide ownership), so the |
| +// journal written from the plan names them and the move carries them. |
| func presence(plan *Plan, hosts []string) ([]bool, error) { |
| + absent, err := DatabaseAbsent(plan) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if absent { |
| + plan.Artifacts[0].Required = false |
| + } |
| present := make([]bool, len(plan.Artifacts)) |
| for i, a := range plan.Artifacts { |
| // Nothing is ever renamed for the socket: the listener is gone |
| @@ -1102,7 +1181,13 @@ func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]r |
| 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. |
| + // cannot be honoured never costs its host a stop. The row's kind must |
| + // agree with the host's shape first: the plan a worker journal is |
| + // checked against is derived from the host's suffix (D8), and a row |
| + // whose Kind says otherwise is a row that was edited. |
| + if err := WorkerRowAgrees(host, rt); err != nil { |
| + return Result{host, "refused", err.Error()} |
| + } |
| if _, err := Validate(j, p.DataDir, p.socketDir(), rt); err != nil { |
| return Result{host, "refused", notActing(err)} |
| } |
| @@ -1144,6 +1229,9 @@ func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]r |
| if fresh == nil { |
| return Result{host, "refused", "row vanished under the fence; not acting"} |
| } |
| + if err := WorkerRowAgrees(host, *fresh); err != nil { |
| + return Result{host, "refused", err.Error()} |
| + } |
| atFrom, err := Validate(j, p.DataDir, p.socketDir(), *fresh) |
| if err != nil { |
| return Result{host, "refused", notActing(err)} |
diff --git a/internal/layout/pass_worker_test.go b/internal/layout/pass_worker_test.go| new file mode 100644 |
| index 000000000..0d689228d |
| --- /dev/null |
| +++ b/internal/layout/pass_worker_test.go |
| @@ -0,0 +1,815 @@ |
| +//go:build linux |
| + |
| +package layout |
| + |
| +// The layout pass over native worker rows (sidecar isolation spec §3.8 D8, |
| +// §3.9 D9's running-replicator check, §3.11 D11's readiness gate). |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "sort" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +const ( |
| + // hibWorker hibernates: it moves asleep. onWorker is always-on: it is |
| + // awake whenever the agent is, so it is deferred and then stopped once. |
| + hibWorker = "bot.shop.abc123" + provision.SidecarSuffix |
| + onWorker = "cron.shop.abc123" + provision.SidecarSuffix |
| +) |
| + |
| +// fakeReplicator is the pass's Replicator seam: it records every database |
| +// it was asked about and answers from answers in order, the last answer |
| +// repeating. No answers means "does not hold". |
| +type fakeReplicator struct { |
| + mu sync.Mutex |
| + asked []string |
| + answers []bool |
| +} |
| + |
| +func (r *fakeReplicator) holds(db string) (bool, string) { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + n := len(r.asked) |
| + r.asked = append(r.asked, db) |
| + if len(r.answers) == 0 { |
| + return false, "" |
| + } |
| + if n >= len(r.answers) { |
| + n = len(r.answers) - 1 |
| + } |
| + if r.answers[n] { |
| + return true, "fixture: carlos-litestream.service started before the config" |
| + } |
| + return false, "" |
| +} |
| + |
| +func workerRow(dd, h string, hibernate bool) registry.Route { |
| + return registry.Route{Host: h, Kind: "sidecar", Hibernate: hibernate, Status: "ready", |
| + Args: `["work"]`, App: "shop", Account: "acct", DBPath: provision.DefaultDBPath(dd, h)} |
| +} |
| + |
| +// seedWorkerFlat is seedFlat plus the rendered bundle config delivery |
| +// writes for every sidecar on every box. |
| +func seedWorkerFlat(t *testing.T, dd, h string) { |
| + t.Helper() |
| + seedFlat(t, dd, h) |
| + if err := os.WriteFile(filepath.Join(dd, h+".config.env"), []byte("LOG_LEVEL=debug\n"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// workerBox is box with two flat worker rows whose files exist, a ready |
| +// box (WorkersReady true) and a replicator that does not hold. |
| +func workerBox(t *testing.T) (*Pass, *fakeFence, *fakeReplicator) { |
| + t.Helper() |
| + p, f := box(t) |
| + for h, hib := range map[string]bool{hibWorker: true, onWorker: false} { |
| + if err := p.Registry.Upsert(workerRow(p.DataDir, h, hib)); err != nil { |
| + t.Fatal(err) |
| + } |
| + seedWorkerFlat(t, p.DataDir, h) |
| + } |
| + p.WorkersReady = func() bool { return true } |
| + rep := &fakeReplicator{} |
| + p.Replicator = rep.holds |
| + return p, f, rep |
| +} |
| + |
| +// workerEndpoint is a worker's row at a layout: the database path alone, |
| +// no socket in either layout (D2). |
| +func workerEndpoint(dd, h string, mode Mode) Endpoint { |
| + if mode == ModeNested { |
| + return Endpoint{DB: provision.NestedDBPath(dd, h)} |
| + } |
| + return Endpoint{DB: provision.DefaultDBPath(dd, h)} |
| +} |
| + |
| +func fenceEvents(f *fakeFence, h string) (fences, lifts int) { |
| + for _, e := range f.events { |
| + switch e { |
| + case "fence " + h: |
| + fences++ |
| + case "lift " + h: |
| + lifts++ |
| + } |
| + } |
| + return fences, lifts |
| +} |
| + |
| +func assertWorkerFlatUntouched(t *testing.T, p *Pass, h string) { |
| + t.Helper() |
| + if got := rowAt(t, p, h); got != workerEndpoint(p.DataDir, 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) |
| + } |
| + // EnsureDirs may have made the raw directory before a check under the |
| + // fence refused; nothing of the host's may be in it. |
| + if entries, err := os.ReadDir(rawInstanceDir(p.DataDir, h)); err == nil && len(entries) != 0 { |
| + names := make([]string, 0, len(entries)) |
| + for _, e := range entries { |
| + names = append(names, e.Name()) |
| + } |
| + t.Errorf("%s has files in its instance directory: %v", h, names) |
| + } |
| +} |
| + |
| +func TestPassMovesHibernatingWorkerAsleep(t *testing.T) { |
| + p, f, rep := workerBox(t) |
| + res := byHost(t, p.Converge(context.Background(), ModeNested)) |
| + if r := res[hibWorker]; r.Outcome != "migrated" { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if got, want := rowAt(t, p, hibWorker), workerEndpoint(p.DataDir, hibWorker, ModeNested); got != want { |
| + t.Fatalf("row = %+v, want %+v (no socket in either layout)", got, want) |
| + } |
| + raw := rawInstanceDir(p.DataDir, hibWorker) |
| + for _, name := range []string{hibWorker + ".db", hibWorker + ".db-wal", hibWorker + ".env", hibWorker + ".config.env", shadowName(hibWorker) + "/generation"} { |
| + if _, err := os.Lstat(filepath.Join(raw, name)); err != nil { |
| + t.Errorf("%s not at the nested path: %v", name, err) |
| + } |
| + } |
| + for _, name := range []string{hibWorker + ".db", hibWorker + ".db-wal", hibWorker + ".env", hibWorker + ".config.env"} { |
| + if _, err := os.Lstat(filepath.Join(p.DataDir, name)); !errors.Is(err, os.ErrNotExist) { |
| + t.Errorf("%s still at the flat path: %v", name, err) |
| + } |
| + } |
| + if JournalPresent(p.DataDir, hibWorker) { |
| + t.Error("journal left behind") |
| + } |
| + // A nested worker row classifies nested with the same Classify a wake |
| + // would use, and the socket artifact was never a file to move. |
| + rt, _ := p.Registry.Get(hibWorker) |
| + routes, _ := p.Registry.List() |
| + if l, why := p.Classify(*rt, routes); l != activator.LayoutNested { |
| + t.Errorf("moved row classifies %s (%s)", l, why) |
| + } |
| + if fences, lifts := fenceEvents(f, hibWorker); fences != 1 || lifts != 1 { |
| + t.Errorf("fence events for the worker: %d fences, %d lifts; all: %v", fences, lifts, f.events) |
| + } |
| + // The replicator was asked about the worker's flat database (twice: |
| + // before the stop and under the fence) and never about an instance. |
| + for _, db := range rep.asked { |
| + if !WorkerHost(strings.TrimSuffix(filepath.Base(db), ".db")) { |
| + t.Errorf("replicator asked about a non-worker database %s", db) |
| + } |
| + } |
| + n := 0 |
| + for _, db := range rep.asked { |
| + if db == provision.DefaultDBPath(p.DataDir, hibWorker) { |
| + n++ |
| + } |
| + } |
| + if n != 2 { |
| + t.Errorf("replicator asked %d times about %s, want 2 (before the stop, under the fence): %v", n, hibWorker, rep.asked) |
| + } |
| +} |
| + |
| +// Review Focus 5 and D11: without the agent-start readiness, worker rows |
| +// are not the pass's. Nil and false alike; instance rows move as before. |
| +func TestWorkersNotEligibleWithoutReadiness(t *testing.T) { |
| + for _, c := range []struct { |
| + name string |
| + ready func() bool |
| + }{ |
| + {"nil", nil}, |
| + {"false", func() bool { return false }}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + p, f, rep := workerBox(t) |
| + p.WorkersReady = c.ready |
| + res := byHost(t, p.Converge(context.Background(), ModeNested)) |
| + for _, h := range []string{hibWorker, onWorker} { |
| + if r, listed := res[h]; listed { |
| + t.Errorf("%s was considered: %+v", h, r) |
| + } |
| + assertWorkerFlatUntouched(t, p, h) |
| + if fences, _ := fenceEvents(f, h); fences != 0 { |
| + t.Errorf("%s was fenced: %v", h, f.events) |
| + } |
| + } |
| + if res[host].Outcome != "migrated" || res[replica].Outcome != "migrated" { |
| + t.Errorf("instance rows: %+v %+v", res[host], res[replica]) |
| + } |
| + if len(rep.asked) != 0 { |
| + t.Errorf("replicator asked while workers were not eligible: %v", rep.asked) |
| + } |
| + // The revert direction is not the pass's either: the box's |
| + // workers are off, so the pass leaves worker rows where they |
| + // stand. |
| + p.WorkersReady = func() bool { return true } |
| + p.Converge(context.Background(), ModeNested) |
| + p.WorkersReady = c.ready |
| + out := byHost(t, p.Converge(context.Background(), ModeFlat)) |
| + if r, listed := out[hibWorker]; listed { |
| + t.Errorf("a nested worker was considered for revert without readiness: %+v", r) |
| + } |
| + if got := rowAt(t, p, hibWorker); got != workerEndpoint(p.DataDir, hibWorker, ModeNested) { |
| + t.Errorf("row = %+v, want still nested", got) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// D8: an always-on worker is awake whenever the agent is. It is deferred |
| +// for the awake grace like any awake host and then stopped once for the |
| +// move; it wakes nested on the sweep's next always-on start. |
| +func TestPassDefersAwakeAlwaysOnWorkerThenStopsOnce(t *testing.T) { |
| + p, f, _ := workerBox(t) |
| + p.DeferAwake = true |
| + p.AwakeGrace = time.Hour |
| + clock := time.Unix(1700000000, 0) |
| + p.Now = func() time.Time { return clock } |
| + f.awake = map[string]bool{onWorker: true} |
| + |
| + res := byHost(t, p.Converge(context.Background(), ModeNested)) |
| + if r := res[onWorker]; r.Outcome != "deferred" || !strings.Contains(r.Reason, "awake") { |
| + t.Fatalf("first pass: %+v", r) |
| + } |
| + if r := res[hibWorker]; r.Outcome != "migrated" { |
| + t.Fatalf("the asleep worker on the same pass: %+v", r) |
| + } |
| + clock = clock.Add(30 * time.Minute) |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[onWorker]; r.Outcome != "deferred" { |
| + t.Fatalf("half way: %+v", r) |
| + } |
| + if fences, _ := fenceEvents(f, onWorker); fences != 0 { |
| + t.Fatalf("the awake worker was fenced inside its grace: %v", f.events) |
| + } |
| + clock = clock.Add(30 * time.Minute) |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[onWorker]; r.Outcome != "migrated" { |
| + t.Fatalf("grace spent: %+v", r) |
| + } |
| + if got := rowAt(t, p, onWorker); got != workerEndpoint(p.DataDir, onWorker, ModeNested) { |
| + t.Fatalf("row = %+v", got) |
| + } |
| + if fences, lifts := fenceEvents(f, onWorker); fences != 1 || lifts != 1 { |
| + t.Fatalf("stopped %d times, released %d times, want once each: %v", fences, lifts, f.events) |
| + } |
| + if len(p.awakeSince) != 0 { |
| + t.Errorf("awake clocks still held: %v", p.awakeSince) |
| + } |
| +} |
| + |
| +// R4-1: the running-replicator check applies to every worker row, a |
| +// hibernating one included. A worker that was always-on when the daemon |
| +// started and has since been switched to hibernating may still be open in |
| +// it, so the row's flag proves nothing. Held before the stop, the worker |
| +// is not stopped; held under the fence, it is released and nothing moves. |
| +func TestFormerlyAlwaysOnWorkerMoveHeldWhileReplicatorNamesFile(t *testing.T) { |
| + t.Run("before the stop", func(t *testing.T) { |
| + p, f, rep := workerBox(t) |
| + rep.answers = []bool{true} |
| + res := byHost(t, p.Converge(context.Background(), ModeNested)) |
| + r := res[hibWorker] |
| + if r.Outcome != "deferred" || !strings.Contains(r.Reason, replicatorHoldsCopy) || !strings.Contains(r.Reason, "fixture") { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if fences, _ := fenceEvents(f, hibWorker); fences != 0 { |
| + t.Errorf("a worker the replicator holds was stopped for nothing: %v", f.events) |
| + } |
| + assertWorkerFlatUntouched(t, p, hibWorker) |
| + if res[host].Outcome != "migrated" { |
| + t.Errorf("the instance row on the same pass: %+v", res[host]) |
| + } |
| + if _, held := p.Held(hibWorker); held { |
| + t.Error("a replicator hold is not a tenant's doing and must not hold the host down") |
| + } |
| + }) |
| + t.Run("under the fence", func(t *testing.T) { |
| + p, f, rep := workerBox(t) |
| + // The pre-stop read says clear; the daemon restarts on an old |
| + // config between it and the read under the fence. |
| + rep.answers = []bool{false, true} |
| + res := byHost(t, p.Converge(context.Background(), ModeNested)) |
| + r := res[hibWorker] |
| + if r.Outcome != "deferred" || !strings.Contains(r.Reason, replicatorHoldsCopy) { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if fences, lifts := fenceEvents(f, hibWorker); fences != 1 || lifts != 1 { |
| + t.Errorf("fence events: %d fences, %d lifts; want the stop and its release: %v", fences, lifts, f.events) |
| + } |
| + assertWorkerFlatUntouched(t, p, hibWorker) |
| + if _, held := p.Held(hibWorker); held { |
| + t.Error("held after a replicator deferral") |
| + } |
| + // The next pass, with the daemon placed, moves it. |
| + rep.answers = []bool{false} |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[hibWorker]; r.Outcome != "migrated" { |
| + t.Fatalf("next pass: %+v", r) |
| + } |
| + }) |
| + t.Run("no replicator wired holds", func(t *testing.T) { |
| + p, f, _ := workerBox(t) |
| + p.Replicator = nil |
| + r := byHost(t, p.Converge(context.Background(), ModeNested))[hibWorker] |
| + if r.Outcome != "deferred" || !strings.Contains(r.Reason, replicatorHoldsCopy) { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if fences, _ := fenceEvents(f, hibWorker); fences != 0 { |
| + t.Errorf("fenced: %v", f.events) |
| + } |
| + assertWorkerFlatUntouched(t, p, hibWorker) |
| + }) |
| +} |
| + |
| +// writeWorkerJournal writes a nested journal for h recording each |
| +// artifact present exactly when it is at its flat source, except the names |
| +// in absent. |
| +func writeWorkerJournal(t *testing.T, p *Pass, h string, mode Mode, absent ...string) *Journal { |
| + t.Helper() |
| + plan, err := PlanFor(p.DataDir, p.SocketDir, h, mode) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + present := make([]bool, len(plan.Artifacts)) |
| + for i, a := range plan.Artifacts { |
| + if a.Name == "socket" || slicesContains(absent, a.Name) { |
| + continue |
| + } |
| + present[i], _ = lstatPresent(a.Src) |
| + } |
| + j := plan.Journal(present) |
| + if err := WriteJournal(p.DataDir, j); err != nil { |
| + t.Fatal(err) |
| + } |
| + return j |
| +} |
| + |
| +// D8: the worker test is the host's suffix, never the row's Kind alone. A |
| +// journal for a worker host whose row says instance, and a sidecar row on |
| +// a host without the suffix, are both refused without a stop. |
| +func TestWorkerJournalKindSuffixMismatchRefused(t *testing.T) { |
| + t.Run("worker host, instance row, journal", func(t *testing.T) { |
| + p, f, _ := workerBox(t) |
| + writeWorkerJournal(t, p, hibWorker, ModeNested) |
| + edited := workerRow(p.DataDir, hibWorker, true) |
| + edited.Kind = "instance" |
| + if err := p.Registry.Upsert(edited); err != nil { |
| + t.Fatal(err) |
| + } |
| + r := byHost(t, p.Recover(context.Background(), ModeNested))[hibWorker] |
| + if r.Outcome != "refused" || !strings.Contains(r.Reason, `row kind "instance"`) || !strings.Contains(r.Reason, "not acting") { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if !JournalPresent(p.DataDir, hibWorker) { |
| + t.Error("the journal was removed; the host must stay fenced by it") |
| + } |
| + if fences, _ := fenceEvents(f, hibWorker); fences != 0 { |
| + t.Errorf("stopped for a journal that cannot be honoured: %v", f.events) |
| + } |
| + if b, err := os.ReadFile(provision.DefaultDBPath(p.DataDir, hibWorker)); err != nil || string(b) != "db "+hibWorker { |
| + t.Errorf("flat db: %q %v", b, err) |
| + } |
| + // Converge refuses the same row before any fence, and nothing |
| + // moves. |
| + res := byHost(t, p.Converge(context.Background(), ModeNested)) |
| + if r := res[hibWorker]; r.Outcome != "refused" { |
| + t.Errorf("converge: %+v", r) |
| + } |
| + }) |
| + t.Run("sidecar row on a host without the suffix", func(t *testing.T) { |
| + p, f, _ := workerBox(t) |
| + odd := "worker.example.com" |
| + rt := workerRow(p.DataDir, odd, true) |
| + if err := p.Registry.Upsert(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + seedWorkerFlat(t, p.DataDir, odd) |
| + r := byHost(t, p.Converge(context.Background(), ModeNested))[odd] |
| + if r.Outcome != "refused" || !strings.Contains(r.Reason, provision.SidecarSuffix) { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if fences, _ := fenceEvents(f, odd); fences != 0 { |
| + t.Errorf("fenced: %v", f.events) |
| + } |
| + assertWorkerFlatUntouched(t, p, odd) |
| + }) |
| +} |
| + |
| +// A forged journal that records a present database absent would, if |
| +// honoured, move the rest and strand the database flat. Recovery refuses |
| +// it, not acting, and the journal keeps the host fenced for an operator. |
| +func TestRecoverRefusesAForgedJournalClaimingThePresentDBAbsent(t *testing.T) { |
| + p, _, _ := workerBox(t) |
| + writeWorkerJournal(t, p, hibWorker, ModeNested, "db") |
| + db := provision.DefaultDBPath(p.DataDir, hibWorker) |
| + r := byHost(t, p.Recover(context.Background(), ModeNested))[hibWorker] |
| + if r.Outcome != "refused" || !strings.Contains(r.Reason, "records db absent") || !strings.Contains(r.Reason, db) { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if !JournalPresent(p.DataDir, hibWorker) { |
| + t.Error("the journal was removed") |
| + } |
| + if b, err := os.ReadFile(db); err != nil || string(b) != "db "+hibWorker { |
| + t.Errorf("flat db: %q %v", b, err) |
| + } |
| + if _, err := os.Lstat(filepath.Join(p.DataDir, hibWorker+".env")); err != nil { |
| + t.Errorf("the env was moved without the database: %v", err) |
| + } |
| +} |
| + |
| +// R1-4: a worker's move takes the full extras inventory, exactly as an |
| +// instance's does. The box-wide replicator's meta directory beside a |
| +// present database is one of them. |
| +func TestWorkerExtrasMove(t *testing.T) { |
| + p, _, _ := workerBox(t) |
| + blobs := filepath.Join(p.DataDir, hibWorker+"-blobs") |
| + if err := os.MkdirAll(blobs, 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(blobs, "x"), []byte("blob"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + meta := provision.DefaultDBPath(p.DataDir, hibWorker) + "-litestream" |
| + if err := os.MkdirAll(meta, 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(meta, "generation"), []byte("1"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[hibWorker]; r.Outcome != "migrated" { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + raw := rawInstanceDir(p.DataDir, hibWorker) |
| + for _, rel := range []string{hibWorker + "-blobs/x", hibWorker + ".db-litestream/generation"} { |
| + if _, err := os.Lstat(filepath.Join(raw, rel)); err != nil { |
| + t.Errorf("%s not at the nested path: %v", rel, err) |
| + } |
| + } |
| + for _, path := range []string{blobs, meta} { |
| + if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) { |
| + t.Errorf("%s still flat: %v", path, err) |
| + } |
| + } |
| + if note := p.Note(hibWorker); !strings.Contains(note, hibWorker+"-blobs") || !strings.Contains(note, hibWorker+".db-litestream") { |
| + t.Errorf("note = %q, want the extras named", note) |
| + } |
| +} |
| + |
| +// D8 rev 2: a standby row, with config and no database (nor any companion) |
| +// at either endpoint, moves with the database journaled absent. The next |
| +// wake restores from the bucket into the nested path. |
| +func TestStandbyRowMovesConfigOnly(t *testing.T) { |
| + p, f, _ := workerBox(t) |
| + standby := "idle.shop.abc123" + provision.SidecarSuffix |
| + if err := p.Registry.Upsert(workerRow(p.DataDir, standby, true)); err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, body := range map[string]string{standby + ".config.env": "LOG_LEVEL=debug\n", standby + ".env": "CARLOS_ADMIN_TOKEN=t\n"} { |
| + if err := os.WriteFile(filepath.Join(p.DataDir, name), []byte(body), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // Stop after the journal, so the test reads what the pass recorded. |
| + p.abortAt = "after-journal" |
| + p.Converge(context.Background(), ModeNested) |
| + p.abortAt = "" |
| + j, found, err := ReadJournal(p.DataDir, standby) |
| + if err != nil || !found { |
| + t.Fatalf("journal: %v %v", found, err) |
| + } |
| + byName := map[string]bool{} |
| + for _, a := range j.Artifacts { |
| + byName[a.Name] = a.Present |
| + } |
| + if byName["db"] || byName["db-wal"] || byName["litestream-shadow"] || byName["socket"] { |
| + t.Errorf("journal records db %v wal %v shadow %v socket %v present, want all absent", byName["db"], byName["db-wal"], byName["litestream-shadow"], byName["socket"]) |
| + } |
| + if !byName["config-env"] || !byName["env"] { |
| + t.Errorf("journal records config-env %v env %v, want both present", byName["config-env"], byName["env"]) |
| + } |
| + if j.From.Socket != "" || j.To.Socket != "" { |
| + t.Errorf("journal endpoints carry sockets: %+v -> %+v", j.From, j.To) |
| + } |
| + // Recovery honours it, and the row ends nested with no database |
| + // anywhere. |
| + r := byHost(t, p.Recover(context.Background(), ModeNested))[standby] |
| + if r.Outcome != "recovered" { |
| + t.Fatalf("recover: %+v", r) |
| + } |
| + if got := rowAt(t, p, standby); got != workerEndpoint(p.DataDir, standby, ModeNested) { |
| + t.Fatalf("row = %+v", got) |
| + } |
| + raw := rawInstanceDir(p.DataDir, standby) |
| + for _, name := range []string{standby + ".config.env", standby + ".env"} { |
| + if _, err := os.Lstat(filepath.Join(raw, name)); err != nil { |
| + t.Errorf("%s not at the nested path: %v", name, err) |
| + } |
| + if _, err := os.Lstat(filepath.Join(p.DataDir, name)); !errors.Is(err, os.ErrNotExist) { |
| + t.Errorf("%s still flat: %v", name, err) |
| + } |
| + } |
| + for _, path := range []string{provision.DefaultDBPath(p.DataDir, standby), filepath.Join(raw, standby+".db")} { |
| + if _, err := os.Lstat(path); !errors.Is(err, os.ErrNotExist) { |
| + t.Errorf("a database appeared at %s: %v", path, err) |
| + } |
| + } |
| + if JournalPresent(p.DataDir, standby) { |
| + t.Error("journal left behind") |
| + } |
| + if fences, lifts := fenceEvents(f, standby); fences == 0 || fences != lifts { |
| + t.Errorf("fence events: %d fences, %d lifts: %v", fences, lifts, f.events) |
| + } |
| + // A plain pass does the same in one step for a second standby. |
| + second := "idle2.shop.abc123" + provision.SidecarSuffix |
| + if err := p.Registry.Upsert(workerRow(p.DataDir, second, true)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(p.DataDir, second+".config.env"), []byte("A=1\n"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[second]; r.Outcome != "migrated" { |
| + t.Fatalf("second standby: %+v", r) |
| + } |
| + if _, err := os.Lstat(filepath.Join(rawInstanceDir(p.DataDir, second), second+".config.env")); err != nil { |
| + t.Errorf("second standby's config: %v", err) |
| + } |
| +} |
| + |
| +// D8 rev 2's other half: a companion without its database is local history |
| +// that went missing. The move is refused with the names, before any stop, |
| +// at either endpoint. Instance rows get the same rule. |
| +func TestCompanionWithoutDatabaseRefused(t *testing.T) { |
| + cases := []struct { |
| + name string |
| + plant func(t *testing.T, p *Pass, h string) string |
| + }{ |
| + {"wal at the source", func(t *testing.T, p *Pass, h string) string { |
| + return provision.DefaultDBPath(p.DataDir, h) + "-wal" |
| + }}, |
| + {"shadow at the source", func(t *testing.T, p *Pass, h string) string { |
| + return filepath.Join(p.DataDir, shadowName(h), "generation") |
| + }}, |
| + {"meta directory at the source", func(t *testing.T, p *Pass, h string) string { |
| + return provision.DefaultDBPath(p.DataDir, h) + "-litestream/generation" |
| + }}, |
| + {"shm at the destination", func(t *testing.T, p *Pass, h string) string { |
| + return filepath.Join(rawInstanceDir(p.DataDir, h), h+".db-shm") |
| + }}, |
| + {"database at the destination only", func(t *testing.T, p *Pass, h string) string { |
| + return filepath.Join(rawInstanceDir(p.DataDir, h), h+".db") |
| + }}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + p, f, _ := workerBox(t) |
| + h := "lone.shop.abc123" + provision.SidecarSuffix |
| + if err := p.Registry.Upsert(workerRow(p.DataDir, h, true)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(p.DataDir, h+".config.env"), []byte("A=1\n"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + path := c.plant(t, p, h) |
| + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(path, []byte("history"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + r := byHost(t, p.Converge(context.Background(), ModeNested))[h] |
| + named := path |
| + if strings.HasSuffix(path, "/generation") { |
| + named = filepath.Dir(path) |
| + } |
| + if r.Outcome != "refused" || !strings.Contains(r.Reason, "db is missing at") || !strings.Contains(r.Reason, named) { |
| + t.Fatalf("result = %+v, want a refusal naming %s", r, named) |
| + } |
| + if fences, _ := fenceEvents(f, h); fences != 0 { |
| + t.Errorf("stopped before the refusal: %v", f.events) |
| + } |
| + if JournalPresent(p.DataDir, h) { |
| + t.Error("a journal was written") |
| + } |
| + if _, err := os.Lstat(filepath.Join(p.DataDir, h+".config.env")); err != nil { |
| + t.Errorf("the config moved without the database: %v", err) |
| + } |
| + if _, err := os.Lstat(path); err != nil { |
| + t.Errorf("the planted companion moved: %v", err) |
| + } |
| + }) |
| + } |
| + t.Run("instance row too", func(t *testing.T) { |
| + p, f := box(t) |
| + if err := os.Remove(provision.DefaultDBPath(p.DataDir, host)); err != nil { |
| + t.Fatal(err) |
| + } |
| + r := byHost(t, p.Converge(context.Background(), ModeNested))[host] |
| + if r.Outcome != "refused" || !strings.Contains(r.Reason, provision.DefaultDBPath(p.DataDir, host)+"-wal") { |
| + t.Fatalf("result = %+v", r) |
| + } |
| + if fences, _ := fenceEvents(f, host); fences != 0 { |
| + t.Errorf("fenced: %v", f.events) |
| + } |
| + }) |
| +} |
| + |
| +// workerWakeBackend is wakeBackend with a worker start and a restore that |
| +// writes a database at the row's path, the way restorePlain fills an empty |
| +// path from the bucket on a cold box. |
| +type workerWakeBackend struct { |
| + wakeBackend |
| + svc *activator.Service |
| + restored []string |
| +} |
| + |
| +func (b *workerWakeBackend) Restore(_ context.Context, rt registry.Route) (bool, error) { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + if err := os.MkdirAll(filepath.Dir(rt.DBPath), 0o700); err != nil { |
| + return false, err |
| + } |
| + if err := os.WriteFile(rt.DBPath, []byte("db from the bucket"), 0o600); err != nil { |
| + return false, err |
| + } |
| + b.restored = append(b.restored, rt.DBPath) |
| + return false, nil |
| +} |
| + |
| +func (b *workerWakeBackend) StartWorker(_ context.Context, rt registry.Route, _ map[string]string) (activator.Proc, string, error) { |
| + b.mu.Lock() |
| + b.started = append(b.started, rt) |
| + b.mu.Unlock() |
| + // The child reports ready over the agent socket, which ends the |
| + // settle window; off this goroutine, as the real call arrives. |
| + go b.svc.NoteWorkerReady(rt.Host, "") |
| + return nopProc{}, "", nil |
| +} |
| + |
| +// R1-5: two boxes, where only one has ever held the database. The other's |
| +// row moves with config only, and the host then fails over to it nested, |
| +// restoring into the nested path. |
| +func TestTwoBoxFailoverAfterConfigOnlyMove(t *testing.T) { |
| + h := hibWorker |
| + pA, _, _ := workerBox(t) // holds the database |
| + pB, _, _ := workerBox(t) |
| + // Box B never woke the worker: config only. |
| + for _, name := range []string{h + ".db", h + ".db-wal", h + ".env"} { |
| + if err := os.Remove(filepath.Join(pB.DataDir, name)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := os.RemoveAll(filepath.Join(pB.DataDir, shadowName(h))); err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, p := range map[string]*Pass{"A": pA, "B": pB} { |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[h]; r.Outcome != "migrated" { |
| + t.Fatalf("box %s: %+v", name, r) |
| + } |
| + if got := rowAt(t, p, h); got != workerEndpoint(p.DataDir, h, ModeNested) { |
| + t.Fatalf("box %s row = %+v", name, got) |
| + } |
| + } |
| + if _, err := os.Lstat(filepath.Join(rawInstanceDir(pA.DataDir, h), h+".db")); err != nil { |
| + t.Fatalf("box A's database: %v", err) |
| + } |
| + if _, err := os.Lstat(filepath.Join(rawInstanceDir(pB.DataDir, h), h+".db")); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("box B has a database it never held: %v", err) |
| + } |
| + if _, err := os.Lstat(filepath.Join(rawInstanceDir(pB.DataDir, h), h+".config.env")); err != nil { |
| + t.Fatalf("box B's config did not move: %v", err) |
| + } |
| + |
| + // Failover: the host wakes on B through a real activator, against the |
| + // row the pass flipped, and restores into the nested path. |
| + b := &workerWakeBackend{} |
| + svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, b, pB.Registry) |
| + svc.Routes = pB.Registry |
| + b.svc = svc |
| + t.Cleanup(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + defer cancel() |
| + svc.Shutdown(ctx) |
| + }) |
| + rt, err := pB.Registry.Get(h) |
| + if err != nil || rt == nil { |
| + t.Fatal(err) |
| + } |
| + if err := svc.WakeWorker(*rt); err != nil { |
| + t.Fatalf("WakeWorker on B after the move: %v", err) |
| + } |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + want := provision.NestedDBPath(pB.DataDir, h) |
| + if len(b.restored) != 1 || b.restored[0] != want { |
| + t.Fatalf("restored = %v, want [%s]", b.restored, want) |
| + } |
| + if len(b.started) != 1 || b.started[0].DBPath != want || b.started[0].Socket != "" { |
| + t.Fatalf("started = %+v, want the nested row", b.started) |
| + } |
| + if _, err := os.Lstat(provision.DefaultDBPath(pB.DataDir, h)); !errors.Is(err, os.ErrNotExist) { |
| + t.Errorf("a flat database appeared on B: %v", err) |
| + } |
| +} |
| + |
| +// snapshotFlat is every entry of the data directory named for h, with the |
| +// contents of every regular file under it. |
| +func snapshotFlat(t *testing.T, dd, h string) map[string]string { |
| + t.Helper() |
| + out := map[string]string{} |
| + entries, err := os.ReadDir(dd) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, e := range entries { |
| + if !belongsTo(e.Name(), h) { |
| + continue |
| + } |
| + root := filepath.Join(dd, e.Name()) |
| + err := filepath.Walk(root, func(path string, fi os.FileInfo, err error) error { |
| + if err != nil { |
| + return err |
| + } |
| + rel, _ := filepath.Rel(dd, path) |
| + if fi.IsDir() { |
| + out[rel+"/"] = "" |
| + return nil |
| + } |
| + b, err := os.ReadFile(path) |
| + if err != nil { |
| + return err |
| + } |
| + out[rel] = string(b) |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func sortedKeys(m map[string]string) []string { |
| + keys := make([]string, 0, len(m)) |
| + for k := range m { |
| + keys = append(keys, k) |
| + } |
| + sort.Strings(keys) |
| + return keys |
| +} |
| + |
| +// D11: carlos system layout revert brings a worker back byte for byte, |
| +// extras included, with the row at the flat endpoints and no socket. |
| +func TestRevertWorkerByteForByte(t *testing.T) { |
| + p, _, _ := workerBox(t) |
| + blobs := filepath.Join(p.DataDir, hibWorker+"-blobs") |
| + if err := os.MkdirAll(blobs, 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(blobs, "x"), []byte("blob"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := snapshotFlat(t, p.DataDir, hibWorker) |
| + if len(before) < 6 { |
| + t.Fatalf("snapshot too small to prove anything: %v", sortedKeys(before)) |
| + } |
| + if r := byHost(t, p.Converge(context.Background(), ModeNested))[hibWorker]; r.Outcome != "migrated" { |
| + t.Fatalf("to nested: %+v", r) |
| + } |
| + if left := snapshotFlat(t, p.DataDir, hibWorker); len(left) != 0 { |
| + t.Fatalf("left flat after the move: %v", sortedKeys(left)) |
| + } |
| + if r := byHost(t, p.Converge(context.Background(), ModeFlat))[hibWorker]; r.Outcome != "reverted" { |
| + t.Fatalf("to flat: %+v", r) |
| + } |
| + after := snapshotFlat(t, p.DataDir, hibWorker) |
| + if len(after) != len(before) { |
| + t.Fatalf("entries before %v, after %v", sortedKeys(before), sortedKeys(after)) |
| + } |
| + for k, v := range before { |
| + if after[k] != v { |
| + t.Errorf("%s: before %q, after %q", k, v, after[k]) |
| + } |
| + } |
| + if got := rowAt(t, p, hibWorker); got != workerEndpoint(p.DataDir, hibWorker, ModeFlat) { |
| + t.Fatalf("row = %+v", got) |
| + } |
| + if JournalPresent(p.DataDir, hibWorker) { |
| + t.Error("journal left behind") |
| + } |
| + // The raw instance directory may remain (empty); nothing of the |
| + // host's is in it. |
| + if entries, err := os.ReadDir(rawInstanceDir(p.DataDir, hibWorker)); err == nil && len(entries) != 0 { |
| + names := make([]string, 0, len(entries)) |
| + for _, e := range entries { |
| + names = append(names, e.Name()) |
| + } |
| + t.Errorf("left nested after the revert: %v", names) |
| + } |
| +} |
diff --git a/internal/layout/plan.go b/internal/layout/plan.go| index 2fc6a9120..c89388fc6 100644 |
| --- a/internal/layout/plan.go |
| +++ b/internal/layout/plan.go |
| @@ -10,6 +10,7 @@ package layout |
| import ( |
| "fmt" |
| "path/filepath" |
| + "slices" |
| "strconv" |
| "strings" |
| @@ -18,6 +19,7 @@ import ( |
| "github.com/carlosframework/platform/internal/instancectl" |
| "github.com/carlosframework/platform/internal/instancekey" |
| "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| @@ -74,6 +76,12 @@ type Plan struct { |
| Mode Mode |
| From, To Endpoint |
| Artifacts []Artifact |
| + // Worker marks a plan for a native worker host (sidecar isolation D8): |
| + // decided by the host's suffix alone (WorkerHost), never by the row's |
| + // Kind, because recovery recomputes the plan from the host name and a |
| + // same-uid flat tenant can edit the row. Both endpoints carry Socket "" |
| + // and the socket artifact is recorded absent on both sides. |
| + Worker bool |
| // DataDir is the box's data directory, kept so the checks that run on |
| // the plan can find the pass's own directory under it (the aside |
| // directory, outside any instance tree) without being handed it again. |
| @@ -231,6 +239,30 @@ func IsNestedPlatformName(host, name string) bool { |
| return err == nil && instancectl.GenLaunchFileName(gen) == name |
| } |
| +// WorkerHost reports whether host is a native worker's (sidecar isolation |
| +// D2): every sidecar host the platform mints ends in |
| +// provision.SidecarSuffix, which never resolves. This is the pass's one |
| +// worker test; the row's Kind must agree with it (WorkerRowAgrees) or the |
| +// row is refused like any journal mismatch. |
| +func WorkerHost(host string) bool { |
| + return strings.HasSuffix(host, provision.SidecarSuffix) |
| +} |
| + |
| +// WorkerRowAgrees checks that a row's Kind says what its host's shape says: |
| +// a sidecar row on a worker host, or a non-sidecar row on any other. A |
| +// disagreement is a row someone edited, or a journal forged for the wrong |
| +// kind of host, and the pass refuses to act on it rather than derive a |
| +// plan the row cannot have come from. |
| +func WorkerRowAgrees(host string, rt registry.Route) error { |
| + if WorkerHost(host) == rt.IsSidecar() { |
| + return nil |
| + } |
| + if rt.IsSidecar() { |
| + return fmt.Errorf("layout: %s: row kind %q on a host without the %s suffix; not acting", host, rt.Kind, provision.SidecarSuffix) |
| + } |
| + return fmt.Errorf("layout: %s: row kind %q on a worker host (%s suffix); not acting", host, rt.Kind, provision.SidecarSuffix) |
| +} |
| + |
| // PlanFor derives the plan for host in the given direction from the host |
| // name alone (design §8.2 step 3). The endpoints are the row values the |
| // pass will compare against and swap; the artifacts are the files (and one |
| @@ -241,6 +273,11 @@ func IsNestedPlatformName(host, name string) bool { |
| // any directory a caller passes, which is invisible in production |
| // (socketDir is always provision.SocketDir there) but would silently break |
| // every test that wants an isolated socket tree. |
| +// |
| +// A worker host (WorkerHost) has no socket in either layout (sidecar |
| +// isolation D2, D8): both endpoints carry Socket "", and the socket |
| +// artifact, which the pass never renames anyway, is recorded absent on |
| +// both sides. The eight other artifacts keep their order and their paths. |
| func PlanFor(dataDir, socketDir, host string, mode Mode) (*Plan, error) { |
| if mode != ModeNested && mode != ModeFlat { |
| return nil, fmt.Errorf("layout: mode %q is not nested or flat", mode) |
| @@ -249,6 +286,7 @@ func PlanFor(dataDir, socketDir, host string, mode Mode) (*Plan, error) { |
| return nil, fmt.Errorf("layout: %q is not a hostname", host) |
| } |
| iid := provision.InstanceID(host) |
| + worker := WorkerHost(host) |
| flat := Endpoint{ |
| DB: provision.DefaultDBPath(dataDir, host), |
| Socket: filepath.Join(socketDir, host+".sock"), |
| @@ -258,7 +296,15 @@ func PlanFor(dataDir, socketDir, host string, mode Mode) (*Plan, error) { |
| Socket: filepath.Join(socketDir, "instances", iid, "app.sock"), |
| } |
| src, dst := flatFiles(dataDir, socketDir, host), nestedFiles(dataDir, socketDir, host) |
| - p := &Plan{Host: host, IID: iid, Mode: mode, From: flat, To: nested, DataDir: dataDir} |
| + if worker { |
| + flat.Socket, nested.Socket = "", "" |
| + for i, name := range artifactNames { |
| + if name == "socket" { |
| + src[i], dst[i] = "", "" |
| + } |
| + } |
| + } |
| + p := &Plan{Host: host, IID: iid, Mode: mode, From: flat, To: nested, DataDir: dataDir, Worker: worker} |
| if mode == ModeFlat { |
| p.From, p.To = nested, flat |
| src, dst = dst, src |
| @@ -275,6 +321,80 @@ func PlanFor(dataDir, socketDir, host string, mode Mode) (*Plan, error) { |
| return p, nil |
| } |
| +// databaseCompanions are the artifacts that are evidence of a database's |
| +// local history: files that exist only beside a database that was here. |
| +var databaseCompanions = []string{"db-wal", "db-shm", "litestream-shadow", "corrupt"} |
| + |
| +// DatabaseAbsent decides the database's Required flag for one move |
| +// (sidecar isolation D8, revision 2). The pass used to refuse any host |
| +// whose database was not at its source, which kept a row minted on a box |
| +// that never woke it flat for good: config delivery writes a sidecar's |
| +// .config.env on every box whether or not it has run, so a standby has |
| +// config and no database. |
| +// |
| +// It reports absent (and the caller drops Required, so the journal records |
| +// the database absent and the rest moves) only when the database and every |
| +// one of its companions are absent at BOTH endpoints: db, db-wal, db-shm, |
| +// the activator's litestream shadow, the quarantine, and the box-wide |
| +// replicator's <db>-litestream directory. The next wake restores from the |
| +// bucket into whichever path the row then names, which is what every cold |
| +// box does already. A companion without its database (a WAL with no |
| +// database, a shadow directory, a meta directory) is local history that |
| +// went missing, and the move is refused with the names, as before; an |
| +// operator decides. A database that is at its source is simply present, |
| +// and the answer is false with no error. |
| +// |
| +// Filesystem absence is enough here (spec §3.8, the part of R1-4 declined): |
| +// the platform's own rule is already that an absent database means |
| +// "restore from the bucket", and a flat tenant that can rename a stopped |
| +// sibling's database can delete it outright with no pass involved. |
| +func DatabaseAbsent(plan *Plan) (absent bool, err error) { |
| + db := plan.Artifacts[0] |
| + if db.Name != "db" { |
| + return false, fmt.Errorf("layout: %s: plan's first artifact is %q, not db", plan.Host, db.Name) |
| + } |
| + atSrc, err := lstatPresent(db.Src) |
| + if err != nil { |
| + return false, err |
| + } |
| + if atSrc { |
| + return false, nil |
| + } |
| + var found []string |
| + note := func(path string) error { |
| + ok, err := lstatPresent(path) |
| + if err != nil { |
| + return err |
| + } |
| + if ok { |
| + found = append(found, path) |
| + } |
| + return nil |
| + } |
| + if err := note(db.Dst); err != nil { |
| + return false, err |
| + } |
| + for _, a := range plan.Artifacts { |
| + if !slices.Contains(databaseCompanions, a.Name) { |
| + continue |
| + } |
| + for _, path := range []string{a.Src, a.Dst} { |
| + if err := note(path); err != nil { |
| + return false, err |
| + } |
| + } |
| + } |
| + for _, path := range []string{db.Src + "-litestream", db.Dst + "-litestream"} { |
| + if err := note(path); err != nil { |
| + return false, err |
| + } |
| + } |
| + if len(found) > 0 { |
| + return false, fmt.Errorf("db is missing at %s, but its history is not: %s exist(s); refusing to move a database's files without it", db.Src, strings.Join(found, ", ")) |
| + } |
| + return true, nil |
| +} |
| + |
| // Journal builds the journal for this plan with the given presence |
| // vector (one bool per fixed artifact, artifactNames order), plus the |
| // plan's extra entries by name. |
diff --git a/internal/layout/replicator.go b/internal/layout/replicator.go| new file mode 100644 |
| index 000000000..81f417d97 |
| --- /dev/null |
| +++ b/internal/layout/replicator.go |
| @@ -0,0 +1,254 @@ |
| +package layout |
| + |
| +// The box-wide replicator's hold on a worker's database (sidecar isolation |
| +// spec §3.9, D9; R2-3, R3-3, R4-1). |
| +// |
| +// Every worker's lease replication is its only replication, and the |
| +// box-wide litestream's generated config leaves every worker out |
| +// (ops.ActivatorReplicated). But /etc/litestream.yml is regenerated by an |
| +// operator (carlos ops litestream) and litestream reads it only at start, |
| +// so a daemon that started on an older config may still hold a worker's |
| +// database open, checkpointing its WAL, long after the file stopped naming |
| +// it. Renaming the database under such a daemon is the two-writer case |
| +// the file itself warns about. The running daemon, not the file, decides: |
| +// the pass moves a worker row only when all of these hold, read under the |
| +// fence right before the first rename, and not only for always-on workers |
| +// (a worker switched from always-on to hibernating since the daemon |
| +// started may still be open in it, so the row's flag proves nothing): |
| +// |
| +// - the config does not exist, or exists and does not name the database, |
| +// so any start from now on cannot open it; |
| +// - carlos-litestream is exactly active/running and its main process |
| +// started AFTER the config's mtime (so it read the config that excludes |
| +// the file), or exactly inactive or failed with no main PID and an |
| +// empty cgroup (so nothing of it runs); |
| +// - no observation failed. |
| +// |
| +// Anything else (activating, deactivating, reloading, an unreadable |
| +// property, a config naming the file, a running daemon with no config to |
| +// place it against) holds, with the reason, and the pass defers the move |
| +// and asks again on its next pass. Because the config already excludes the |
| +// file, a restart or a pending start after the check opens a config |
| +// without it. |
| + |
| +import ( |
| + "bufio" |
| + "context" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "os/exec" |
| + "path/filepath" |
| + "strconv" |
| + "strings" |
| + "time" |
| +) |
| + |
| +const ( |
| + // litestreamUnit is the box-wide replicator's unit. |
| + litestreamUnit = "carlos-litestream.service" |
| + // litestreamConfigPath is the config carlos ops litestream generates |
| + // and the unit's ExecStart reads (internal/host/units). |
| + litestreamConfigPath = "/etc/litestream.yml" |
| + // systemctlTimeout bounds each systemctl show the production |
| + // observation runs; an answer that does not come is a failed |
| + // observation, which holds. |
| + systemctlTimeout = 5 * time.Second |
| + // cgroupRoot is where the unit's cgroup.events is read from, under the |
| + // ControlGroup path systemd reports. |
| + cgroupRoot = "/sys/fs/cgroup" |
| +) |
| + |
| +// boxReplicatorHolds is the decision above with its two observations |
| +// injected: show answers one systemctl show property of a unit (the string |
| +// the --value form prints), cgroupEmpty reports whether the unit's cgroup |
| +// has no process in it (an absent cgroup is empty). It returns true when |
| +// the move must wait, with the reason an operator reads. |
| +func boxReplicatorHolds(dbPath string, show func(unit, prop string) (string, error), cgroupEmpty func(unit string) (bool, error), configPath string) (bool, string) { |
| + if show == nil || cgroupEmpty == nil { |
| + return true, "no observation of " + litestreamUnit + " is wired" |
| + } |
| + configExists, configMtime, err := configNames(configPath, dbPath) |
| + if err != nil { |
| + return true, err.Error() |
| + } |
| + props := map[string]string{} |
| + for _, prop := range []string{"ActiveState", "SubState", "ExecMainStartTimestamp", "MainPID"} { |
| + v, err := show(litestreamUnit, prop) |
| + if err != nil { |
| + return true, fmt.Sprintf("%s %s unreadable: %v", litestreamUnit, prop, err) |
| + } |
| + props[prop] = strings.TrimSpace(v) |
| + } |
| + state, sub := props["ActiveState"], props["SubState"] |
| + switch state { |
| + case "active": |
| + if sub != "running" { |
| + return true, fmt.Sprintf("%s is active/%s, not running", litestreamUnit, sub) |
| + } |
| + if !configExists { |
| + return true, fmt.Sprintf("%s is running but %s does not exist, so what it read at start cannot be placed", litestreamUnit, configPath) |
| + } |
| + started, err := parseSystemdTimestamp(props["ExecMainStartTimestamp"]) |
| + if err != nil { |
| + return true, fmt.Sprintf("%s ExecMainStartTimestamp %q: %v", litestreamUnit, props["ExecMainStartTimestamp"], err) |
| + } |
| + if !started.After(configMtime) { |
| + return true, fmt.Sprintf("%s started at %s, before %s was written at %s, so it may have read a config naming this database", litestreamUnit, started.UTC().Format(time.RFC3339), configPath, configMtime.UTC().Format(time.RFC3339)) |
| + } |
| + return false, "" |
| + case "inactive", "failed": |
| + if props["MainPID"] != "0" { |
| + return true, fmt.Sprintf("%s is %s but reports MainPID %s", litestreamUnit, state, props["MainPID"]) |
| + } |
| + empty, err := cgroupEmpty(litestreamUnit) |
| + if err != nil { |
| + return true, fmt.Sprintf("%s cgroup unreadable: %v", litestreamUnit, err) |
| + } |
| + if !empty { |
| + return true, fmt.Sprintf("%s is %s but its cgroup is not empty", litestreamUnit, state) |
| + } |
| + return false, "" |
| + } |
| + return true, fmt.Sprintf("%s is %s/%s, neither running nor stopped", litestreamUnit, state, sub) |
| +} |
| + |
| +// configNames reads the replicator's config and reports whether it exists, |
| +// its mtime, and an error when it cannot be read or when it names dbPath. |
| +// The generated file writes each database as " - path: <path>"; a |
| +// hand-written one may quote the value or omit the space, so every |
| +// whitespace-separated token is compared, quotes and a "path:" prefix |
| +// stripped, as a cleaned path. A match anywhere is a hold: the generous |
| +// reading is the safe one. |
| +func configNames(configPath, dbPath string) (exists bool, mtime time.Time, err error) { |
| + fi, err := os.Stat(configPath) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return false, time.Time{}, nil |
| + } |
| + if err != nil { |
| + return false, time.Time{}, fmt.Errorf("%s unreadable: %w", configPath, err) |
| + } |
| + f, err := os.Open(configPath) |
| + if err != nil { |
| + return true, fi.ModTime(), fmt.Errorf("%s unreadable: %w", configPath, err) |
| + } |
| + defer f.Close() |
| + want := filepath.Clean(dbPath) |
| + sc := bufio.NewScanner(f) |
| + sc.Buffer(make([]byte, 0, 64*1024), 1<<20) |
| + for sc.Scan() { |
| + line := sc.Text() |
| + if i := strings.IndexByte(line, '#'); i >= 0 { |
| + line = line[:i] |
| + } |
| + for _, tok := range strings.Fields(line) { |
| + tok = strings.TrimPrefix(tok, "path:") |
| + tok = strings.Trim(tok, `"'`) |
| + if tok != "" && filepath.Clean(tok) == want { |
| + return true, fi.ModTime(), fmt.Errorf("%s names %s", configPath, dbPath) |
| + } |
| + } |
| + } |
| + if err := sc.Err(); err != nil { |
| + return true, fi.ModTime(), fmt.Errorf("%s unreadable: %w", configPath, err) |
| + } |
| + return true, fi.ModTime(), nil |
| +} |
| + |
| +// parseSystemdTimestamp reads a timestamp as systemctl show prints it: |
| +// "@<seconds>" under --timestamp=unix (what the production observation |
| +// asks for), or the default "Fri 2026-10-24 10:00:00 UTC". The textual |
| +// form is accepted only in UTC or GMT: Go reads an unknown zone |
| +// abbreviation as offset zero, which would place a daemon started in |
| +// summer time an hour wrong, and an hour is exactly the kind of margin |
| +// this comparison must not be wrong by. Anything else is an error, and an |
| +// error holds. |
| +func parseSystemdTimestamp(s string) (time.Time, error) { |
| + s = strings.TrimSpace(s) |
| + if s == "" || s == "n/a" { |
| + return time.Time{}, errors.New("no timestamp") |
| + } |
| + if rest, ok := strings.CutPrefix(s, "@"); ok { |
| + secs, frac, _ := strings.Cut(rest, ".") |
| + n, err := strconv.ParseInt(secs, 10, 64) |
| + if err != nil { |
| + return time.Time{}, fmt.Errorf("not a unix timestamp: %w", err) |
| + } |
| + var nanos int64 |
| + if frac != "" { |
| + // Up to microseconds, as systemd prints them. |
| + for len(frac) < 9 { |
| + frac += "0" |
| + } |
| + nanos, err = strconv.ParseInt(frac[:9], 10, 64) |
| + if err != nil { |
| + return time.Time{}, fmt.Errorf("not a unix timestamp: %w", err) |
| + } |
| + } |
| + return time.Unix(n, nanos), nil |
| + } |
| + t, err := time.Parse("Mon 2006-01-02 15:04:05 MST", s) |
| + if err != nil { |
| + return time.Time{}, err |
| + } |
| + if name, _ := t.Zone(); name != "UTC" && name != "GMT" { |
| + return time.Time{}, fmt.Errorf("zone %q is not UTC; ask systemctl for --timestamp=unix", name) |
| + } |
| + return t, nil |
| +} |
| + |
| +// BoxReplicatorHolds is the production observation of the box-wide |
| +// replicator for the pass's Replicator seam: systemctl show, run as the |
| +// agent user and read-only, and the unit's cgroup.events under the |
| +// ControlGroup systemd reports for it. An inactive unit has no control |
| +// group, which is an empty one. |
| +func BoxReplicatorHolds(dbPath string) (bool, string) { |
| + return boxReplicatorHolds(dbPath, systemctlShow, unitCgroupEmpty, litestreamConfigPath) |
| +} |
| + |
| +// systemctlShow prints one property of a unit. --timestamp=unix makes |
| +// every timestamp property "@<seconds>", which parseSystemdTimestamp reads |
| +// without a zone. |
| +func systemctlShow(unit, prop string) (string, error) { |
| + ctx, cancel := context.WithTimeout(context.Background(), systemctlTimeout) |
| + defer cancel() |
| + out, err := exec.CommandContext(ctx, "systemctl", "show", "--timestamp=unix", "-p", prop, "--value", unit).Output() |
| + if err != nil { |
| + var ee *exec.ExitError |
| + if errors.As(err, &ee) && len(ee.Stderr) > 0 { |
| + return "", fmt.Errorf("systemctl show %s: %w: %s", unit, err, strings.TrimSpace(string(ee.Stderr))) |
| + } |
| + return "", fmt.Errorf("systemctl show %s: %w", unit, err) |
| + } |
| + return strings.TrimSpace(string(out)), nil |
| +} |
| + |
| +// unitCgroupEmpty reads the unit's cgroup.events. A unit systemd reports |
| +// no ControlGroup for has no cgroup, so nothing runs in it; a cgroup |
| +// directory that is gone is empty the same way; otherwise cgroup.events |
| +// must say "populated 0". |
| +func unitCgroupEmpty(unit string) (bool, error) { |
| + group, err := systemctlShow(unit, "ControlGroup") |
| + if err != nil { |
| + return false, err |
| + } |
| + if group == "" { |
| + return true, nil |
| + } |
| + if !strings.HasPrefix(group, "/") || strings.Contains(group, "..") { |
| + return false, fmt.Errorf("ControlGroup %q is not a cgroup path", group) |
| + } |
| + b, err := os.ReadFile(filepath.Join(cgroupRoot, group, "cgroup.events")) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return true, nil |
| + } |
| + if err != nil { |
| + return false, err |
| + } |
| + for _, line := range strings.Split(string(b), "\n") { |
| + if f := strings.Fields(line); len(f) == 2 && f[0] == "populated" { |
| + return f[1] == "0", nil |
| + } |
| + } |
| + return false, fmt.Errorf("cgroup.events of %s has no populated line", group) |
| +} |
diff --git a/internal/layout/replicator_test.go b/internal/layout/replicator_test.go| new file mode 100644 |
| index 000000000..5a1fa7584 |
| --- /dev/null |
| +++ b/internal/layout/replicator_test.go |
| @@ -0,0 +1,248 @@ |
| +package layout |
| + |
| +import ( |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// replicatorFixture is one observation of carlos-litestream: the config |
| +// file (written, or absent when config is ""), its mtime, the four |
| +// properties systemctl show would print, and the cgroup's answer. |
| +type replicatorFixture struct { |
| + config string |
| + configMtime time.Time |
| + props map[string]string |
| + propErr map[string]error |
| + cgroupEmpty bool |
| + cgroupErr error |
| +} |
| + |
| +const fixtureDB = "/data/bot.shop.abc123.sidecar.internal.db" |
| + |
| +var ( |
| + configAt = time.Date(2026, 10, 24, 10, 0, 0, 0, time.UTC) |
| + before = configAt.Add(-time.Minute) |
| + after = configAt.Add(time.Minute) |
| +) |
| + |
| +func unixStamp(t time.Time) string { return "@" + itoa(t.Unix()) } |
| + |
| +func itoa(n int64) string { |
| + b, neg := []byte{}, n < 0 |
| + if neg { |
| + n = -n |
| + } |
| + for n >= 10 { |
| + b = append([]byte{byte('0' + n%10)}, b...) |
| + n /= 10 |
| + } |
| + b = append([]byte{byte('0' + n)}, b...) |
| + if neg { |
| + b = append([]byte{'-'}, b...) |
| + } |
| + return string(b) |
| +} |
| + |
| +// configWithout is a generated config naming another database only. |
| +const configWithout = "snapshot:\n interval: 1h\ndbs:\n - path: /data/registry.db\n replicas:\n - url: s3://b/p/data/registry.db\n" |
| + |
| +// configWith names the worker's database. |
| +const configWith = configWithout + " - path: " + fixtureDB + "\n replicas:\n - url: s3://b/p" + fixtureDB + "\n" |
| + |
| +func running(started time.Time) map[string]string { |
| + return map[string]string{"ActiveState": "active", "SubState": "running", "ExecMainStartTimestamp": unixStamp(started), "MainPID": "4242"} |
| +} |
| + |
| +func stopped(state string) map[string]string { |
| + return map[string]string{"ActiveState": state, "SubState": "dead", "ExecMainStartTimestamp": "n/a", "MainPID": "0"} |
| +} |
| + |
| +func (f replicatorFixture) holds(t *testing.T) (bool, string) { |
| + t.Helper() |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "litestream.yml") |
| + if f.config != "" { |
| + if err := os.WriteFile(path, []byte(f.config), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.Chtimes(path, f.configMtime, f.configMtime); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + show := func(unit, prop string) (string, error) { |
| + if unit != litestreamUnit { |
| + t.Errorf("asked about %q, want %s", unit, litestreamUnit) |
| + } |
| + if err := f.propErr[prop]; err != nil { |
| + return "", err |
| + } |
| + v, ok := f.props[prop] |
| + if !ok { |
| + t.Errorf("asked for %q, which the fixture does not define", prop) |
| + } |
| + return v, nil |
| + } |
| + cgroup := func(unit string) (bool, error) { return f.cgroupEmpty, f.cgroupErr } |
| + return boxReplicatorHolds(fixtureDB, show, cgroup, path) |
| +} |
| + |
| +func wantHold(t *testing.T, f replicatorFixture, reasonHas string) { |
| + t.Helper() |
| + holds, why := f.holds(t) |
| + if !holds { |
| + t.Fatalf("holds = false, want a hold mentioning %q", reasonHas) |
| + } |
| + if !strings.Contains(why, reasonHas) { |
| + t.Fatalf("reason = %q, want it to mention %q", why, reasonHas) |
| + } |
| +} |
| + |
| +func wantProceed(t *testing.T, f replicatorFixture) { |
| + t.Helper() |
| + if holds, why := f.holds(t); holds { |
| + t.Fatalf("holds = true (%s), want the move to proceed", why) |
| + } |
| +} |
| + |
| +// The config names the file: a hold whatever the daemon is doing, running |
| +// or stopped, because the next start would open it. |
| +func TestAlwaysOnMoveHeldWhileReplicatorNamesFile(t *testing.T) { |
| + t.Run("running after the config", func(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWith, configMtime: configAt, props: running(after)}, "names "+fixtureDB) |
| + }) |
| + t.Run("inactive and empty", func(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWith, configMtime: configAt, props: stopped("inactive"), cgroupEmpty: true}, "names "+fixtureDB) |
| + }) |
| + t.Run("quoted hand-written path", func(t *testing.T) { |
| + cfg := "dbs:\n - path: \"" + fixtureDB + "\"\n" |
| + wantHold(t, replicatorFixture{config: cfg, configMtime: configAt, props: running(after)}, "names "+fixtureDB) |
| + }) |
| + t.Run("a longer path is not this file", func(t *testing.T) { |
| + cfg := "dbs:\n - path: " + fixtureDB + "-litestream/x.db\n" |
| + wantProceed(t, replicatorFixture{config: cfg, configMtime: configAt, props: running(after)}) |
| + }) |
| +} |
| + |
| +// A running daemon that started before the config was written may have |
| +// read a config that named the file. |
| +func TestAlwaysOnMoveHeldWhileReplicatorStartedBeforeConfig(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(before)}, "before") |
| + t.Run("started at the same instant", func(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(configAt)}, "before") |
| + }) |
| + t.Run("running with no config at all", func(t *testing.T) { |
| + wantHold(t, replicatorFixture{props: running(after)}, "does not exist") |
| + }) |
| +} |
| + |
| +func TestAlwaysOnMoveHeldWhileDeactivating(t *testing.T) { |
| + props := map[string]string{"ActiveState": "deactivating", "SubState": "stop-sigterm", "ExecMainStartTimestamp": unixStamp(after), "MainPID": "4242"} |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props, cgroupEmpty: true}, "deactivating") |
| +} |
| + |
| +func TestAlwaysOnMoveHeldWhileActivating(t *testing.T) { |
| + props := map[string]string{"ActiveState": "activating", "SubState": "start", "ExecMainStartTimestamp": unixStamp(after), "MainPID": "4242"} |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}, "activating") |
| + t.Run("reloading", func(t *testing.T) { |
| + props := map[string]string{"ActiveState": "reloading", "SubState": "reload", "ExecMainStartTimestamp": unixStamp(after), "MainPID": "4242"} |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}, "reloading") |
| + }) |
| + t.Run("active but not running", func(t *testing.T) { |
| + props := running(after) |
| + props["SubState"] = "exited" |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}, "not running") |
| + }) |
| +} |
| + |
| +func TestAlwaysOnMoveHeldOnUnreadableProperty(t *testing.T) { |
| + boom := errors.New("systemctl: connection refused") |
| + for _, prop := range []string{"ActiveState", "SubState", "ExecMainStartTimestamp", "MainPID"} { |
| + t.Run(prop, func(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after), propErr: map[string]error{prop: boom}}, prop+" unreadable") |
| + }) |
| + } |
| + t.Run("timestamp in an unknown zone", func(t *testing.T) { |
| + props := running(after) |
| + props["ExecMainStartTimestamp"] = "Fri 2026-10-24 12:00:00 CEST" |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}, "not UTC") |
| + }) |
| + t.Run("timestamp unparseable", func(t *testing.T) { |
| + props := running(after) |
| + props["ExecMainStartTimestamp"] = "n/a" |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}, "ExecMainStartTimestamp") |
| + }) |
| + t.Run("cgroup unreadable", func(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: stopped("inactive"), cgroupErr: errors.New("permission denied")}, "cgroup unreadable") |
| + }) |
| + t.Run("inactive with a main pid", func(t *testing.T) { |
| + props := stopped("inactive") |
| + props["MainPID"] = "77" |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props, cgroupEmpty: true}, "MainPID 77") |
| + }) |
| + t.Run("inactive with a populated cgroup", func(t *testing.T) { |
| + wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: stopped("failed"), cgroupEmpty: false}, "cgroup is not empty") |
| + }) |
| + t.Run("config unreadable", func(t *testing.T) { |
| + dir := t.TempDir() |
| + // A directory where the file should be: Stat succeeds, Open |
| + // succeeds, the read fails. |
| + path := filepath.Join(dir, "litestream.yml") |
| + if err := os.Mkdir(path, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + show := func(unit, prop string) (string, error) { return running(after)[prop], nil } |
| + holds, why := boxReplicatorHolds(fixtureDB, show, func(string) (bool, error) { return true, nil }, path) |
| + if !holds || !strings.Contains(why, "unreadable") { |
| + t.Fatalf("holds = %v (%s), want a hold for an unreadable config", holds, why) |
| + } |
| + }) |
| + t.Run("no observation wired", func(t *testing.T) { |
| + if holds, _ := boxReplicatorHolds(fixtureDB, nil, nil, "/nonexistent"); !holds { |
| + t.Fatal("nil observations must hold") |
| + } |
| + }) |
| +} |
| + |
| +func TestAlwaysOnMoveProceedsWhenInactiveEmptyAndConfigExcludes(t *testing.T) { |
| + wantProceed(t, replicatorFixture{config: configWithout, configMtime: configAt, props: stopped("inactive"), cgroupEmpty: true}) |
| + t.Run("failed and empty", func(t *testing.T) { |
| + wantProceed(t, replicatorFixture{config: configWithout, configMtime: configAt, props: stopped("failed"), cgroupEmpty: true}) |
| + }) |
| + t.Run("no config at all", func(t *testing.T) { |
| + wantProceed(t, replicatorFixture{props: stopped("inactive"), cgroupEmpty: true}) |
| + }) |
| +} |
| + |
| +func TestAlwaysOnMoveProceedsWhenRunningAfterConfigWithout(t *testing.T) { |
| + wantProceed(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after)}) |
| + t.Run("textual UTC timestamp", func(t *testing.T) { |
| + props := running(after) |
| + props["ExecMainStartTimestamp"] = after.Format("Mon 2006-01-02 15:04:05 MST") |
| + wantProceed(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}) |
| + }) |
| + t.Run("unix timestamp with microseconds", func(t *testing.T) { |
| + props := running(after) |
| + props["ExecMainStartTimestamp"] = unixStamp(after) + ".250000" |
| + wantProceed(t, replicatorFixture{config: configWithout, configMtime: configAt, props: props}) |
| + }) |
| +} |
| + |
| +func TestParseSystemdTimestamp(t *testing.T) { |
| + got, err := parseSystemdTimestamp("@1761300000") |
| + if err != nil || !got.Equal(time.Unix(1761300000, 0)) { |
| + t.Fatalf("unix form: %v %v", got, err) |
| + } |
| + got, err = parseSystemdTimestamp("Fri 2026-10-24 10:00:00 UTC") |
| + if err != nil || !got.Equal(configAt) { |
| + t.Fatalf("textual UTC form: %v %v", got, err) |
| + } |
| + for _, bad := range []string{"", "n/a", "@x", "Fri 2026-10-24 10:00:00 BST", "2026-10-24"} { |
| + if _, err := parseSystemdTimestamp(bad); err == nil { |
| + t.Errorf("%q parsed, want an error", bad) |
| + } |
| + } |
| +} |