amadan
Merge pull request #215 from carlosframework/fleet-claims-only
litestream: the glob proposes, the registry disposes
3 files changed,
+102
−6
internal/ops/litestream.go+46 −5internal/ops/ops.go+8 −1internal/ops/ops_test.go+48 −0
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go| index e246f3e1f..d6f706e15 100644 |
| --- a/internal/ops/litestream.go |
| +++ b/internal/ops/litestream.go |
| @@ -321,9 +321,17 @@ func consoleDBPath(dataDir string) string { |
| } |
| func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) { |
| + dbs, _, err := FleetDBsWithSkipped(reg, dataDir) |
| + return dbs, err |
| +} |
| + |
| +// FleetDBsWithSkipped is FleetDBs, plus the files under dataDir it declined to |
| +// claim — see the "glob proposes, registry disposes" note inside. The verb |
| +// prints them; nothing here is dropped without being named. |
| +func FleetDBsWithSkipped(reg *registry.Registry, dataDir string) (dbs []string, skipped []string, err error) { |
| routes, err := reg.List() |
| if err != nil { |
| - return nil, err |
| + return nil, nil, err |
| } |
| hibernating := map[string]bool{} |
| unitPaths := map[string]bool{} |
| @@ -345,11 +353,44 @@ func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) { |
| matches, err := filepath.Glob(filepath.Join(dataDir, "*.db")) |
| if err != nil { |
| - return nil, err |
| + return nil, nil, err |
| + } |
| + // THE GLOB PROPOSES, THE REGISTRY DISPOSES. A *.db under dataDir is |
| + // replicated when a route claims it, or when it is the box's own registry |
| + // or console. Anything else is somebody's leftover, and streaming leftovers |
| + // off-box costs storage forever: the flagship has a 410M |
| + // seapointish-purge-backup.db in /data, root-owned, matching this glob |
| + // exactly, which today escapes replication only because that box's config |
| + // is still hand-written. |
| + // |
| + // Skipped files are NAMED to the caller, never dropped in silence — a |
| + // database that someone did want, quietly left out, is the precise failure |
| + // this area keeps producing. The shrink guard (CoverageShortfall) is the |
| + // second net: if a skipped file is one the live config replicates today, |
| + // the write is refused outright. |
| + claimed := map[string]bool{ |
| + filepath.Clean(filepath.Join(dataDir, "registry.db")): true, |
| + filepath.Clean(consoleDBPath(dataDir)): true, |
| + } |
| + for _, r := range routes { |
| + if r.DBPath != "" { |
| + claimed[filepath.Clean(r.DBPath)] = true |
| + } |
| } |
| + var globbed []string |
| + for _, p := range matches { |
| + if claimed[filepath.Clean(p)] { |
| + globbed = append(globbed, p) |
| + continue |
| + } |
| + skipped = append(skipped, filepath.Clean(p)) |
| + } |
| + matches = globbed |
| + |
| // Unit-tenant DBPaths are added on top of the glob: they typically live |
| // entirely outside dataDir, at their StateDirectory path. A DBPath naming |
| - // a DIRECTORY is a tree — see treeDBs. |
| + // a DIRECTORY is a tree — see treeDBs. These are claimed by construction: |
| + // a route named them. |
| for p := range unitPaths { |
| if fi, err := os.Stat(p); err == nil && fi.IsDir() { |
| matches = append(matches, treeDBs(p)...) |
| @@ -367,7 +408,6 @@ func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) { |
| matches = append(matches, consoleDBPath(dataDir)) |
| seen := map[string]bool{} |
| - var dbs []string |
| for _, p := range matches { |
| p = filepath.Clean(p) |
| if seen[p] { |
| @@ -383,7 +423,8 @@ func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) { |
| seen[p] = true |
| dbs = append(dbs, p) |
| } |
| - return dbs, nil |
| + sort.Strings(skipped) |
| + return dbs, skipped, nil |
| } |
| // restoreAndCheckFn is the per-database restore behind a package var, so the |
diff --git a/internal/ops/ops.go b/internal/ops/ops.go| index 678679cd9..3e52cf61f 100644 |
| --- a/internal/ops/ops.go |
| +++ b/internal/ops/ops.go |
| @@ -111,13 +111,20 @@ func runLitestream(args []string) { |
| if err != nil { |
| log.Fatalf("litestream: open registry %s: %v", rp, err) |
| } |
| - dbs, err := FleetDBs(reg, dir) |
| + dbs, skipped, err := FleetDBsWithSkipped(reg, dir) |
| if err != nil { |
| log.Fatalf("litestream: %v", err) |
| } |
| if len(dbs) == 0 { |
| log.Fatalf("litestream: no non-empty *.db files under %s — wrong box?", dir) |
| } |
| + if len(skipped) > 0 { |
| + // Named, never silent: these are files under the data dir that no route |
| + // claims. Usually leftovers (a purge backup, a restore scratch); if one |
| + // is a database someone still wants, the fix is a route that says so. |
| + fmt.Fprintf(os.Stderr, "litestream: %d file(s) under %s are claimed by no route and will NOT be replicated:\n %s\n", |
| + len(skipped), dir, strings.Join(skipped, "\n ")) |
| + } |
| // Never quietly replicate less than the file being replaced — see |
| // CoverageShortfall. This is what makes a cutover from a hand-maintained |
diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go| index 97cea384b..28be7f860 100644 |
| --- a/internal/ops/ops_test.go |
| +++ b/internal/ops/ops_test.go |
| @@ -706,3 +706,51 @@ func TestReplicaURLsAreUniquePerPath(t *testing.T) { |
| t.Fatalf("replica keys lost the path that identifies them:\n%s", cfg) |
| } |
| } |
| + |
| +// The dataDir glob takes every *.db it finds, which means the box-wide |
| +// replicator will happily stream anything that lands there — including things |
| +// that are not tenant databases at all. The flagship has a 410M |
| +// seapointish-purge-backup.db sitting in /data: a leftover from a purge, owned |
| +// by root, matching the glob exactly. Today it is not replicated only because |
| +// that box's config is hand-written; the moment the generator takes over it |
| +// would join the replication set and start costing storage forever. |
| +// |
| +// So the glob proposes and the REGISTRY disposes: a file under dataDir is |
| +// replicated when some route claims it (or it is the box's own registry or |
| +// console). Anything else is skipped and NAMED — never silently, because a |
| +// skipped database that someone did want is exactly the failure this whole |
| +// area keeps producing. |
| +func TestFleetDBsSkipsUnclaimedFilesAndSaysSo(t *testing.T) { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + claimed := filepath.Join(dir, "app.example.com.db") |
| + stray := filepath.Join(dir, "seapointish-purge-backup.db") |
| + for _, p := range []string{claimed, stray} { |
| + if err := os.WriteFile(p, []byte("x"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := reg.Upsert(registry.Route{Host: "app.example.com", DBPath: claimed}); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + dbs, skipped, err := FleetDBsWithSkipped(reg, dir) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !slices.Contains(dbs, claimed) { |
| + t.Fatalf("a route's own database was skipped: %v", dbs) |
| + } |
| + if !slices.Contains(dbs, filepath.Join(dir, "registry.db")) { |
| + t.Fatalf("the box registry was skipped: %v", dbs) |
| + } |
| + if slices.Contains(dbs, stray) { |
| + t.Fatalf("an unclaimed file joined the replication set: %v", dbs) |
| + } |
| + if !slices.Contains(skipped, stray) { |
| + t.Fatalf("the unclaimed file was dropped SILENTLY — skipped = %v", skipped) |
| + } |
| +} |