amadan

carlos / platform Public

Sidecar isolation Task 8 fix round 1: recovery asks the replicator, the config is decoded strictly and read as one version, an absent-db journal may name no companion (D8, D9)

1. complete asks the replicator seam at the files phase before any
   rename, under the fence and inside the serialized section, exactly as
   a fresh move does; a hold leaves the journal as it is (the host stays
   fenced by it) and the next recovery asks again. A journal past the
   files phase has nothing left to rename.
2. litestreamDBPaths decodes the one shape the generator writes (block
   mapping, dbs block sequence, path plain or simply quoted) and holds on
   anything else: flow collections, escaped scalars, anchors, tags, block
   scalars, tabs, a second document, no or two dbs keys, a path with
   environment expansion, any unrecognised line. No YAML library is in
   go.mod and none is added (GOPROXY off; go.mod needs a root envelope).
3. readConfig takes one stable version (open, fstat, read, fstat, stat:
   same file, size and mtime, whole file read) and configFile.unchanged
   checks the path again after the daemon is observed; a config rewritten
   in place, replaced, removed or appearing in the window holds.
4. Validate refuses a journal recording the db absent with any companion
   present (db-wal, db-shm, litestream-shadow, corrupt) or naming the
   replicator's meta directory as an extra, so recovery cannot move an
   orphan and flip the row; pinned at both endpoints in both directions.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 0bdbe7822494f3727c40e921d69d950d78a8c882 parent 0fbb62d
6 files changed, +586 −46
  • internal/layout/journal.go +21 −5
  • internal/layout/pass.go +9 −0
  • internal/layout/pass_worker_test.go +180 −1
  • internal/layout/plan.go +7 −1
  • internal/layout/replicator.go +234 −38
  • internal/layout/replicator_test.go +135 −1
diff --git a/internal/layout/journal.go b/internal/layout/journal.go
index ae8457fcc..416c6e003 100644
--- a/internal/layout/journal.go
+++ b/internal/layout/journal.go
@@ -8,6 +8,7 @@ import (
"os"
"path/filepath"
"reflect"
+ "slices"
"strings"
"time"
@@ -341,11 +342,26 @@ func Validate(j *Journal, dataDir, socketDir string, row registry.Route) (atFrom
// 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.
- //
+ // at both endpoints (DatabaseAbsent), and a journal must say the same:
+ // one that records the db absent and a companion present describes a
+ // move the pass refuses to make (a WAL, a shadow or the replicator's
+ // meta directory without their database is history gone missing), so
+ // it is refused here whatever is on disk. Recovery then holds the
+ // journal to what it records: 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.
+ if !j.Artifacts[0].Present {
+ for _, a := range j.Artifacts {
+ if a.Present && slices.Contains(databaseCompanions, a.Name) {
+ return false, fmt.Errorf("layout: %s: journal records db absent but %s present; not acting", j.Host, a.Name)
+ }
+ }
+ for _, e := range j.Extra {
+ if e.Name == litestreamMetaName(j.Host) {
+ return false, fmt.Errorf("layout: %s: journal records db absent but names its replicator meta directory %s; not acting", j.Host, e.Name)
+ }
+ }
+ }
// 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/pass.go b/internal/layout/pass.go
index 2630b8ed0..f63b658f7 100644
--- a/internal/layout/pass.go
+++ b/internal/layout/pass.go
@@ -1332,6 +1332,15 @@ func (p *Pass) complete(plan *Plan, j *Journal, atFrom, remove bool) Result {
}
switch j.Phase {
case PhaseFiles:
+ // Renames are still ahead, so a worker journal asks the box-wide
+ // replicator exactly as a fresh move does (D9), under the fence
+ // and inside the serialized section. A hold leaves the journal as
+ // it is, which keeps the host fenced, and the next recovery asks
+ // again. A journal past this phase has nothing left to rename.
+ if holds, why := p.replicatorHolds(plan); holds {
+ p.logf("layout: %s: recovery deferred -> %s: %s", host, replicatorHoldsCopy, why)
+ return replicatorDeferred(host, why)
+ }
if err := p.moveFiles(plan); err != nil {
return Result{host, "refused", notActing(err)}
}
diff --git a/internal/layout/pass_worker_test.go b/internal/layout/pass_worker_test.go
index 0d689228d..03261ec3a 100644
--- a/internal/layout/pass_worker_test.go
+++ b/internal/layout/pass_worker_test.go
@@ -10,6 +10,7 @@ import (
"errors"
"os"
"path/filepath"
+ "slices"
"sort"
"strings"
"sync"
@@ -405,9 +406,12 @@ func TestWorkerJournalKindSuffixMismatchRefused(t *testing.T) {
// 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.
+// The journal records the companions absent too, so Validate's own
+// companion rule (TestRecoverRefusesAnAbsentDBJournalWithOrphanCompanions)
+// is not what refuses it; complete's check of the files is.
func TestRecoverRefusesAForgedJournalClaimingThePresentDBAbsent(t *testing.T) {
p, _, _ := workerBox(t)
- writeWorkerJournal(t, p, hibWorker, ModeNested, "db")
+ writeWorkerJournal(t, p, hibWorker, ModeNested, "db", "db-wal", "db-shm", "litestream-shadow", "corrupt")
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) {
@@ -813,3 +817,178 @@ func TestRevertWorkerByteForByte(t *testing.T) {
t.Errorf("left nested after the revert: %v", names)
}
}
+
+// Fix round 1, finding 1: recovery renames under the same D9 check a fresh
+// move does. A worker journal at the files phase is deferred while the
+// replicator holds, left exactly as it was (so the host stays fenced by
+// it), and completed once the hold clears. A journal past the files phase
+// has nothing left to rename and completes regardless.
+func TestRecoverDefersAWorkerJournalWhileTheReplicatorHolds(t *testing.T) {
+ t.Run("files phase waits", func(t *testing.T) {
+ p, f, rep := workerBox(t)
+ p.abortAt = "after-journal"
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ j, found, _ := ReadJournal(p.DataDir, hibWorker)
+ if !found || j.Phase != PhaseFiles {
+ t.Fatalf("setup: %+v %v", j, found)
+ }
+ rep.answers = []bool{true}
+ r := byHost(t, p.Recover(context.Background(), ModeNested))[hibWorker]
+ if r.Outcome != "deferred" || !strings.Contains(r.Reason, replicatorHoldsCopy) {
+ t.Fatalf("recover under a hold = %+v", r)
+ }
+ again, found, _ := ReadJournal(p.DataDir, hibWorker)
+ if !found || again.Phase != PhaseFiles || len(again.Artifacts) != len(j.Artifacts) {
+ t.Fatalf("journal after the deferral: %+v %v, want the files-phase journal untouched", again, found)
+ }
+ if b, err := os.ReadFile(provision.DefaultDBPath(p.DataDir, hibWorker)); err != nil || string(b) != "db "+hibWorker {
+ t.Fatalf("the database was renamed under a hold: %q %v", b, err)
+ }
+ if got := rowAt(t, p, hibWorker); got != workerEndpoint(p.DataDir, hibWorker, ModeFlat) {
+ t.Fatalf("row = %+v, want still flat", got)
+ }
+ if f.fences[hibWorker] != 0 {
+ t.Fatal("the pass kept its in-memory fence; the journal is what fences the host")
+ }
+ if _, held := p.Held(hibWorker); held {
+ t.Fatal("held after a replicator deferral in recovery")
+ }
+ // The next recovery, with the daemon placed, completes it.
+ rep.answers = []bool{false}
+ if r := byHost(t, p.Recover(context.Background(), ModeNested))[hibWorker]; r.Outcome != "recovered" {
+ t.Fatalf("recover after the hold cleared = %+v", r)
+ }
+ if _, err := os.Lstat(filepath.Join(rawInstanceDir(p.DataDir, hibWorker), hibWorker+".db")); err != nil {
+ t.Fatalf("database not at the nested path after recovery: %v", err)
+ }
+ if JournalPresent(p.DataDir, hibWorker) {
+ t.Fatal("journal left behind")
+ }
+ })
+ t.Run("row phase has nothing to rename", func(t *testing.T) {
+ p, _, rep := workerBox(t)
+ p.abortAt = "after-cas"
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ if j, found, _ := ReadJournal(p.DataDir, hibWorker); !found || j.Phase != PhaseRow {
+ t.Fatalf("setup: %+v %v", j, found)
+ }
+ rep.answers = []bool{true}
+ if r := byHost(t, p.Recover(context.Background(), ModeNested))[hibWorker]; r.Outcome != "recovered" {
+ t.Fatalf("recover at the row phase = %+v", r)
+ }
+ })
+ t.Run("instance journals never ask", func(t *testing.T) {
+ p, _, rep := workerBox(t)
+ p.WorkersReady = func() bool { return false }
+ p.abortAt = "after-journal"
+ p.Converge(context.Background(), ModeNested)
+ p.abortAt = ""
+ rep.answers = []bool{true}
+ if r := byHost(t, p.Recover(context.Background(), ModeNested))[host]; r.Outcome != "recovered" {
+ t.Fatalf("instance recovery = %+v", r)
+ }
+ if len(rep.asked) != 0 {
+ t.Fatalf("replicator asked about %v during instance recovery", rep.asked)
+ }
+ })
+}
+
+// Fix round 1, finding 4: a journal that records the database absent and a
+// companion present describes a move the pass never makes (DatabaseAbsent
+// refuses a companion without its database), so Validate refuses it on
+// shape and recovery moves nothing, at either endpoint and in either
+// direction. The replicator's meta directory is a companion too, carried
+// as an extra.
+func TestRecoverRefusesAnAbsentDBJournalWithOrphanCompanions(t *testing.T) {
+ for _, mode := range []Mode{ModeNested, ModeFlat} {
+ for _, companion := range append(append([]string(nil), databaseCompanions...), "meta") {
+ t.Run(string(mode)+"/"+companion, func(t *testing.T) {
+ p, f, _ := workerBox(t)
+ if mode == ModeFlat {
+ // Start nested, so a revert journal's source is the
+ // nested directory.
+ if r := byHost(t, p.Converge(context.Background(), ModeNested))[hibWorker]; r.Outcome != "migrated" {
+ t.Fatalf("setup: %+v", r)
+ }
+ }
+ plan, err := PlanFor(p.DataDir, p.SocketDir, hibWorker, mode)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Only the orphan remains at the source: the database and
+ // the other companions are gone.
+ srcDir := filepath.Dir(plan.Artifacts[0].Src)
+ for _, a := range plan.Artifacts {
+ if a.Name == "socket" || a.Name == companion {
+ continue
+ }
+ if a.Name == "db" || slices.Contains(databaseCompanions, a.Name) {
+ os.RemoveAll(a.Src)
+ }
+ }
+ var orphan string
+ present := make([]bool, len(plan.Artifacts))
+ j := plan.Journal(present)
+ for i, a := range plan.Artifacts {
+ if a.Name == companion {
+ orphan = a.Src
+ if err := os.MkdirAll(filepath.Dir(orphan), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if a.Dir {
+ os.MkdirAll(orphan, 0o700)
+ os.WriteFile(filepath.Join(orphan, "generation"), []byte("g"), 0o600)
+ } else {
+ os.WriteFile(orphan, []byte("orphan"), 0o600)
+ }
+ j.Artifacts[i].Present = true
+ } else if a.Name != "socket" && a.Name != "db" && !slices.Contains(databaseCompanions, a.Name) {
+ ok, _ := lstatPresent(a.Src)
+ j.Artifacts[i].Present = ok
+ }
+ }
+ if companion == "meta" {
+ orphan = filepath.Join(srcDir, litestreamMetaName(hibWorker))
+ if err := os.MkdirAll(orphan, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ j.Extra = append(j.Extra, JournalArtifact{Name: litestreamMetaName(hibWorker), Present: true})
+ }
+ if err := WriteJournal(p.DataDir, j); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := Validate(j, p.DataDir, p.SocketDir, workerRowAt(t, p, hibWorker)); err == nil || !strings.Contains(err.Error(), "records db absent but") {
+ t.Fatalf("Validate = %v, want the companion refusal", err)
+ }
+ r := byHost(t, p.Recover(context.Background(), mode))[hibWorker]
+ if r.Outcome != "refused" || !strings.Contains(r.Reason, "records db absent but") || !strings.Contains(r.Reason, "not acting") {
+ t.Fatalf("recover = %+v", r)
+ }
+ if _, err := os.Lstat(orphan); err != nil {
+ t.Fatalf("the orphan %s was moved: %v", orphan, err)
+ }
+ if !JournalPresent(p.DataDir, hibWorker) {
+ t.Fatal("the journal was removed")
+ }
+ if fences, _ := fenceEvents(f, hibWorker); mode == ModeNested && fences != 0 {
+ t.Fatalf("stopped for a journal that cannot be honoured: %v", f.events)
+ }
+ // The config env, recorded present, did not move either.
+ if _, err := os.Lstat(filepath.Join(srcDir, hibWorker+".config.env")); err != nil {
+ t.Fatalf("the config moved: %v", err)
+ }
+ })
+ }
+ }
+}
+
+func workerRowAt(t *testing.T, p *Pass, h string) registry.Route {
+ t.Helper()
+ rt, err := p.Registry.Get(h)
+ if err != nil || rt == nil {
+ t.Fatalf("row %s: %v %v", h, rt, err)
+ }
+ return *rt
+}
diff --git a/internal/layout/plan.go b/internal/layout/plan.go
index c89388fc6..ca43a7e88 100644
--- a/internal/layout/plan.go
+++ b/internal/layout/plan.go
@@ -325,6 +325,12 @@ func PlanFor(dataDir, socketDir, host string, mode Mode) (*Plan, error) {
// local history: files that exist only beside a database that was here.
var databaseCompanions = []string{"db-wal", "db-shm", "litestream-shadow", "corrupt"}
+// litestreamMetaName is the box-wide replicator's meta directory beside a
+// host's database, <host>.db-litestream: a companion too, but an extra to
+// the pass (it is not one of the nine artifacts), so it is named here for
+// DatabaseAbsent and Validate.
+func litestreamMetaName(host string) string { return host + ".db-litestream" }
+
// 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
@@ -384,7 +390,7 @@ func DatabaseAbsent(plan *Plan) (absent bool, err error) {
}
}
}
- for _, path := range []string{db.Src + "-litestream", db.Dst + "-litestream"} {
+ for _, path := range []string{filepath.Join(filepath.Dir(db.Src), litestreamMetaName(plan.Host)), filepath.Join(filepath.Dir(db.Dst), litestreamMetaName(plan.Host))} {
if err := note(path); err != nil {
return false, err
}
diff --git a/internal/layout/replicator.go b/internal/layout/replicator.go
index 81f417d97..b0aa0a8e2 100644
--- a/internal/layout/replicator.go
+++ b/internal/layout/replicator.go
@@ -32,13 +32,14 @@ package layout
// without it.
import (
- "bufio"
"context"
"errors"
"fmt"
+ "io"
"os"
"os/exec"
"path/filepath"
+ "slices"
"strconv"
"strings"
"time"
@@ -64,14 +65,24 @@ const (
// 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.
+//
+// The config is read as one stable version (readConfig) and checked again
+// after the daemon has been observed (configFile.unchanged): the generator
+// writes the file in place, so a regeneration between the read and the
+// daemon's observation, or between the two stats of the read, would let
+// new contents be placed against an old mtime, or an old daemon against a
+// config it never saw. Either disagreement holds.
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)
+ cfg, err := readConfig(configPath)
if err != nil {
return true, err.Error()
}
+ if slices.Contains(cfg.paths, filepath.Clean(dbPath)) {
+ return true, fmt.Sprintf("%s names %s", configPath, dbPath)
+ }
props := map[string]string{}
for _, prop := range []string{"ActiveState", "SubState", "ExecMainStartTimestamp", "MainPID"} {
v, err := show(litestreamUnit, prop)
@@ -86,17 +97,16 @@ func boxReplicatorHolds(dbPath string, show func(unit, prop string) (string, err
if sub != "running" {
return true, fmt.Sprintf("%s is active/%s, not running", litestreamUnit, sub)
}
- if !configExists {
+ if !cfg.exists {
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))
+ if !started.After(cfg.mtime) {
+ 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, cfg.mtime.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"])
@@ -108,51 +118,237 @@ func boxReplicatorHolds(dbPath string, show func(unit, prop string) (string, err
if !empty {
return true, fmt.Sprintf("%s is %s but its cgroup is not empty", litestreamUnit, state)
}
- return false, ""
+ default:
+ return true, fmt.Sprintf("%s is %s/%s, neither running nor stopped", litestreamUnit, state, sub)
+ }
+ // The daemon was placed against the config read above; the config must
+ // still be that version now, or the placement is of a file that is
+ // gone.
+ if err := cfg.unchanged(configPath); err != nil {
+ return true, err.Error()
}
- return true, fmt.Sprintf("%s is %s/%s, neither running nor stopped", litestreamUnit, state, sub)
+ return false, ""
}
-// 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)
+// configFile is one stable version of the replicator's config: whether it
+// exists, the mtime the daemon's start is placed against, the FileInfo the
+// version is recognised by again (os.SameFile), and every database path it
+// names.
+type configFile struct {
+ exists bool
+ mtime time.Time
+ fi os.FileInfo
+ paths []string
+}
+
+// maxConfigSize bounds the read: the generated file is a few lines per
+// database.
+const maxConfigSize = 1 << 20
+
+// readConfig reads one version of the config: open, fstat, read, fstat
+// again, then stat the path again. Every FileInfo must agree (same file,
+// size and mtime) and the bytes read must be the whole file, or the file
+// was being written or replaced during the read and the version is not
+// known. An absent file is a version too (exists false).
+func readConfig(path string) (configFile, error) {
+ f, err := os.Open(path)
if errors.Is(err, os.ErrNotExist) {
- return false, time.Time{}, nil
+ return configFile{}, nil
+ }
+ if err != nil {
+ return configFile{}, fmt.Errorf("%s unreadable: %w", path, err)
+ }
+ defer f.Close()
+ before, err := f.Stat()
+ if err != nil {
+ return configFile{}, fmt.Errorf("%s unreadable: %w", path, err)
}
+ if !before.Mode().IsRegular() {
+ return configFile{}, fmt.Errorf("%s is not a regular file", path)
+ }
+ if before.Size() > maxConfigSize {
+ return configFile{}, fmt.Errorf("%s is %d bytes, over the %d this reader accepts", path, before.Size(), maxConfigSize)
+ }
+ data, err := io.ReadAll(io.LimitReader(f, maxConfigSize+1))
if err != nil {
- return false, time.Time{}, fmt.Errorf("%s unreadable: %w", configPath, err)
+ return configFile{}, fmt.Errorf("%s unreadable: %w", path, err)
}
- f, err := os.Open(configPath)
+ after, err := f.Stat()
if err != nil {
- return true, fi.ModTime(), fmt.Errorf("%s unreadable: %w", configPath, err)
+ return configFile{}, fmt.Errorf("%s unreadable: %w", path, 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 !sameVersion(before, after) || int64(len(data)) != before.Size() {
+ return configFile{}, fmt.Errorf("%s changed while it was being read", path)
+ }
+ now, err := os.Stat(path)
+ if err != nil {
+ return configFile{}, fmt.Errorf("%s replaced while it was being read: %w", path, err)
+ }
+ if !sameVersion(before, now) {
+ return configFile{}, fmt.Errorf("%s replaced while it was being read", path)
+ }
+ paths, err := litestreamDBPaths(data)
+ if err != nil {
+ return configFile{}, fmt.Errorf("%s: %w", path, err)
+ }
+ return configFile{exists: true, mtime: before.ModTime(), fi: before, paths: paths}, nil
+}
+
+// sameVersion reports whether two FileInfos describe the same file at the
+// same size and mtime.
+func sameVersion(a, b os.FileInfo) bool {
+ return os.SameFile(a, b) && a.Size() == b.Size() && a.ModTime().Equal(b.ModTime())
+}
+
+// unchanged reports an error when path no longer holds the version c was
+// read as: appeared, gone, replaced, or written since.
+func (c configFile) unchanged(path string) error {
+ fi, err := os.Stat(path)
+ if !c.exists {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ return fmt.Errorf("%s unreadable after the observation: %w", path, err)
+ }
+ return fmt.Errorf("%s appeared during the observation", path)
+ }
+ if err != nil {
+ return fmt.Errorf("%s changed during the observation: %w", path, err)
+ }
+ if !sameVersion(c.fi, fi) {
+ return fmt.Errorf("%s changed during the observation", path)
+ }
+ return nil
+}
+
+// litestreamDBPaths decodes the database paths of a litestream config in
+// the one shape this platform writes (ops.LitestreamConfig): a block
+// mapping at the top level with a dbs key whose value is a block sequence
+// of block mappings, each with a path key whose value is a plain or simply
+// quoted scalar. That is the whole of the YAML this reader understands,
+// and anything outside it is an error, which holds the move: a flow
+// collection ([...] or {...}), an anchor, alias, tag or merge key, a block
+// scalar, a quoted scalar with an escape, a tab, a second document, a line
+// in no recognised form, or a config with no dbs key. A path at any depth
+// under any key named path counts (a file replica's path too): the
+// generous reading is the safe one. Decoding the general language without
+// a YAML library, or adding one, is what this refuses to do; the generator
+// is the one writer this config has (carlos ops litestream), and a hand-
+// written config holds every worker move until it is regenerated.
+func litestreamDBPaths(data []byte) ([]string, error) {
+ var paths []string
+ sawDBs := false
+ for n, raw := range strings.Split(string(data), "\n") {
+ line := strings.TrimRight(raw, "\r")
+ if strings.ContainsRune(line, '\t') {
+ return nil, fmt.Errorf("line %d: tab; not the generated shape", n+1)
+ }
+ line = stripComment(line)
+ if strings.TrimSpace(line) == "" {
+ continue
+ }
+ if line == "---" || line == "..." {
+ return nil, fmt.Errorf("line %d: document marker; not the generated shape", n+1)
+ }
+ indent := len(line) - len(strings.TrimLeft(line, " "))
+ rest := line[indent:]
+ if strings.HasPrefix(rest, "- ") || rest == "-" {
+ rest = strings.TrimSpace(rest[1:])
+ if rest == "" {
+ continue
+ }
+ }
+ key, value, ok := strings.Cut(rest, ":")
+ if !ok || !plainKey(key) {
+ return nil, fmt.Errorf("line %d: not a key; not the generated shape", n+1)
+ }
+ if value != "" && !strings.HasPrefix(value, " ") {
+ return nil, fmt.Errorf("line %d: not a key; not the generated shape", n+1)
+ }
+ value = strings.TrimSpace(value)
+ if key == "dbs" {
+ if indent != 0 || value != "" || sawDBs {
+ return nil, fmt.Errorf("line %d: dbs must be one top-level block key; not the generated shape", n+1)
+ }
+ sawDBs = true
+ continue
+ }
+ if value == "" {
+ if key == "path" {
+ return nil, fmt.Errorf("line %d: path with no scalar value; not the generated shape", n+1)
+ }
+ continue
+ }
+ scalar, err := plainOrQuotedScalar(value)
+ if err != nil {
+ return nil, fmt.Errorf("line %d: %w; not the generated shape", n+1, err)
+ }
+ if key == "path" {
+ if scalar == "" || strings.Contains(scalar, "$") {
+ return nil, fmt.Errorf("line %d: path %q is not a literal path; not the generated shape", n+1, scalar)
}
+ paths = append(paths, filepath.Clean(scalar))
+ }
+ }
+ if !sawDBs {
+ return nil, errors.New("no dbs key; not the generated shape")
+ }
+ return paths, nil
+}
+
+// stripComment removes a YAML comment: a # at the start of the line or
+// preceded by a space, outside quotes.
+func stripComment(line string) string {
+ inSingle, inDouble := false, false
+ for i, r := range line {
+ switch {
+ case r == '\'' && !inDouble:
+ inSingle = !inSingle
+ case r == '"' && !inSingle:
+ inDouble = !inDouble
+ case r == '#' && !inSingle && !inDouble && (i == 0 || line[i-1] == ' '):
+ return line[:i]
+ }
+ }
+ return line
+}
+
+// plainKey is a YAML plain key of the shape the generator writes.
+func plainKey(key string) bool {
+ if key == "" {
+ return false
+ }
+ for _, r := range key {
+ if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '_' || r == '-' || r == '.') {
+ return false
+ }
+ }
+ return true
+}
+
+// plainOrQuotedScalar decodes a scalar value in the three forms the
+// generator's grammar admits: plain, 'single-quoted' with no doubled
+// quote, "double-quoted" with no backslash. Anything else is an error.
+func plainOrQuotedScalar(v string) (string, error) {
+ switch {
+ case strings.HasPrefix(v, "\"") && strings.HasSuffix(v, "\"") && len(v) >= 2:
+ inner := v[1 : len(v)-1]
+ if strings.ContainsAny(inner, "\\\"") {
+ return "", fmt.Errorf("quoted scalar %s has an escape", v)
+ }
+ return inner, nil
+ case strings.HasPrefix(v, "'") && strings.HasSuffix(v, "'") && len(v) >= 2:
+ inner := v[1 : len(v)-1]
+ if strings.Contains(inner, "'") {
+ return "", fmt.Errorf("quoted scalar %s has an escape", v)
}
+ return inner, nil
}
- if err := sc.Err(); err != nil {
- return true, fi.ModTime(), fmt.Errorf("%s unreadable: %w", configPath, err)
+ if strings.ContainsAny(v[:1], "[{&*!|>%@`\"'") || strings.Contains(v, ": ") || strings.HasSuffix(v, ":") || strings.ContainsAny(v, "[]{},") {
+ return "", fmt.Errorf("scalar %q is not plain", v)
}
- return true, fi.ModTime(), nil
+ return v, nil
}
// parseSystemdTimestamp reads a timestamp as systemctl show prints it:
diff --git a/internal/layout/replicator_test.go b/internal/layout/replicator_test.go
index 5a1fa7584..ec37fd787 100644
--- a/internal/layout/replicator_test.go
+++ b/internal/layout/replicator_test.go
@@ -19,6 +19,9 @@ type replicatorFixture struct {
propErr map[string]error
cgroupEmpty bool
cgroupErr error
+ // onShow runs once, at the first show call, with the config's path:
+ // the window between the config read and the daemon's observation.
+ onShow func(t *testing.T, path string)
}
const fixtureDB = "/data/bot.shop.abc123.sidecar.internal.db"
@@ -73,10 +76,15 @@ func (f replicatorFixture) holds(t *testing.T) (bool, string) {
t.Fatal(err)
}
}
+ shown := false
show := func(unit, prop string) (string, error) {
if unit != litestreamUnit {
t.Errorf("asked about %q, want %s", unit, litestreamUnit)
}
+ if f.onShow != nil && !shown {
+ shown = true
+ f.onShow(t, path)
+ }
if err := f.propErr[prop]; err != nil {
return "", err
}
@@ -196,7 +204,7 @@ func TestAlwaysOnMoveHeldOnUnreadableProperty(t *testing.T) {
}
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") {
+ if !holds || !strings.Contains(why, "not a regular file") {
t.Fatalf("holds = %v (%s), want a hold for an unreadable config", holds, why)
}
})
@@ -246,3 +254,129 @@ func TestParseSystemdTimestamp(t *testing.T) {
}
}
}
+
+// Fix round 1, finding 2: the config is decoded in the one shape the
+// generator writes, and every other shape holds the move. Valid YAML the
+// token scan used to misread (flow style, escaped scalars) now holds
+// rather than proceeding.
+func TestAlwaysOnMoveHeldOnConfigShapeOutsideTheGeneratedForm(t *testing.T) {
+ hold := func(name, cfg string) {
+ t.Run(name, func(t *testing.T) {
+ wantHold(t, replicatorFixture{config: cfg, configMtime: configAt, props: running(after)}, "not the generated shape")
+ })
+ }
+ hold("flow sequence naming the file", "dbs: [{path: "+fixtureDB+", replicas: []}]\n")
+ hold("flow sequence naming another file", "dbs: [{path: /data/other.db}]\n")
+ hold("flow mapping item", "dbs:\n - {path: "+fixtureDB+"}\n")
+ hold("escaped double-quoted scalar", "dbs:\n - path: \"/data/bot.shop.abc123.sidecar.internal\\x2edb\"\n")
+ hold("escaped single-quoted scalar", "dbs:\n - path: '/data/it''s.db'\n")
+ hold("anchor and alias", "x: &p "+fixtureDB+"\ndbs:\n - path: *p\n")
+ hold("block scalar", "dbs:\n - path: |\n "+fixtureDB+"\n")
+ hold("tag", "dbs:\n - path: !!str "+fixtureDB+"\n")
+ hold("tab indentation", "dbs:\n\t- path: /data/other.db\n")
+ hold("second document", "dbs:\n - path: /data/other.db\n---\ndbs:\n - path: "+fixtureDB+"\n")
+ hold("no dbs key", "snapshot:\n interval: 1h\n")
+ hold("dbs twice", "dbs:\n - path: /data/other.db\ndbs:\n - path: "+fixtureDB+"\n")
+ hold("dbs nested", "x:\n dbs:\n - path: /data/other.db\n")
+ hold("environment expansion in a path", "dbs:\n - path: ${DATA}/other.db\n")
+ hold("a line in no recognised form", "dbs:\n - path: /data/other.db\n garbage line here\n")
+ hold("empty path", "dbs:\n - path:\n")
+ t.Run("the generated form with comments and quotes decodes", func(t *testing.T) {
+ cfg := "# GENERATED\nsnapshot:\n interval: 1h # hourly\n retention: 24h\ndbs:\n - path: /data/registry.db\n replicas:\n - url: s3://b/p/data/registry.db\n region: eu-west-1\n - path: '/data/quoted.db'\n replicas:\n - url: \"s3://b/p/data/quoted.db\"\n"
+ paths, err := litestreamDBPaths([]byte(cfg))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(paths) != 2 || paths[0] != "/data/registry.db" || paths[1] != "/data/quoted.db" {
+ t.Fatalf("paths = %v", paths)
+ }
+ wantProceed(t, replicatorFixture{config: cfg, configMtime: configAt, props: running(after)})
+ })
+ t.Run("the generator's own output decodes to its inputs", func(t *testing.T) {
+ // The exact text ops.LitestreamConfig writes for two databases,
+ // pinned here so the decoder and the generator cannot drift apart
+ // unnoticed.
+ cfg := "# GENERATED by `carlos ops litestream` — do not hand-edit; re-run the verb\n# after `carlos add`/`remove` so the replication set follows the fleet.\nsnapshot:\n interval: 1h\n retention: 24h\ndbs:\n - path: /data/a.example.com.db\n replicas:\n - url: s3://bucket/prefix/data/a.example.com.db\n region: eu-west-2\n - path: /data/registry.db\n replicas:\n - url: s3://bucket/prefix/data/registry.db\n region: eu-west-2\n"
+ paths, err := litestreamDBPaths([]byte(cfg))
+ if err != nil || len(paths) != 2 || paths[0] != "/data/a.example.com.db" || paths[1] != "/data/registry.db" {
+ t.Fatalf("paths = %v, err %v", paths, err)
+ }
+ })
+ t.Run("an empty dbs block names nothing", func(t *testing.T) {
+ wantProceed(t, replicatorFixture{config: "snapshot:\n interval: 1h\ndbs:\n", configMtime: configAt, props: running(after)})
+ })
+ t.Run("a path under replicas counts too", func(t *testing.T) {
+ wantHold(t, replicatorFixture{config: "dbs:\n - path: /data/other.db\n replicas:\n - path: " + fixtureDB + "\n", configMtime: configAt, props: running(after)}, "names "+fixtureDB)
+ })
+}
+
+// Fix round 1, finding 3: the config's contents and mtime must be one
+// version, and that version must still be the file after the daemon has
+// been observed. A regeneration in the window (in place, as the generator
+// writes; by rename; or a file appearing) holds.
+func TestAlwaysOnMoveHeldWhenConfigChangesDuringObservation(t *testing.T) {
+ t.Run("rewritten in place", func(t *testing.T) {
+ wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after), onShow: func(t *testing.T, path string) {
+ if err := os.WriteFile(path, []byte(configWith), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chtimes(path, after.Add(time.Hour), after.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ }}, "changed during the observation")
+ })
+ t.Run("rewritten in place with the mtime kept", func(t *testing.T) {
+ // Same mtime, different size: still a different version.
+ wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after), onShow: func(t *testing.T, path string) {
+ if err := os.WriteFile(path, []byte(configWith), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chtimes(path, configAt, configAt); err != nil {
+ t.Fatal(err)
+ }
+ }}, "changed during the observation")
+ })
+ t.Run("replaced by rename", func(t *testing.T) {
+ wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after), onShow: func(t *testing.T, path string) {
+ tmp := path + ".tmp"
+ if err := os.WriteFile(tmp, []byte(configWithout), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chtimes(tmp, configAt, configAt); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Rename(tmp, path); err != nil {
+ t.Fatal(err)
+ }
+ }}, "changed during the observation")
+ })
+ t.Run("removed", func(t *testing.T) {
+ wantHold(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after), onShow: func(t *testing.T, path string) {
+ if err := os.Remove(path); err != nil {
+ t.Fatal(err)
+ }
+ }}, "changed during the observation")
+ })
+ t.Run("appeared", func(t *testing.T) {
+ wantHold(t, replicatorFixture{props: stopped("inactive"), cgroupEmpty: true, onShow: func(t *testing.T, path string) {
+ if err := os.WriteFile(path, []byte(configWith), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }}, "appeared during the observation")
+ })
+ t.Run("unchanged proceeds", func(t *testing.T) {
+ wantProceed(t, replicatorFixture{config: configWithout, configMtime: configAt, props: running(after), onShow: func(*testing.T, string) {}})
+ })
+ t.Run("a directory at the path", func(t *testing.T) {
+ dir := t.TempDir()
+ 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, "not a regular file") {
+ t.Fatalf("holds = %v (%s)", holds, why)
+ }
+ })
+}