| 1 | // litestream — off-box replication of the fleet's SQLite files, and the daily |
| 2 | // proof that the replicas restore. Two verbs, both run ON the box (titogo's |
| 3 | // operator-side SSH orchestration deliberately did not come along — units and |
| 4 | // credentials are provisioned by bootstrap; this is only the box-side half): |
| 5 | // |
| 6 | // carlos ops litestream (re)generate /etc/litestream.yml from the |
| 7 | // registry + data dir; prints "changed" or |
| 8 | // "unchanged" so the calling timer/unit knows |
| 9 | // whether a replicator restart is due |
| 10 | // carlos ops restore-verify restore every configured database from S3 and |
| 11 | // integrity-check it |
| 12 | // |
| 13 | // The config is always generated, never hand-maintained: `carlos add`/`remove` |
| 14 | // change the database set, and re-running the verb is the one way the |
| 15 | // replication set follows. Credentials live only in the unit's EnvironmentFile |
| 16 | // (0600) — this code never sees or prints them. |
| 17 | // |
| 18 | // Ported from titogo internal/ops/litestream.go (config generator, round-trip |
| 19 | // parser, restore-verify), plus Eleven deploy/fleet-backup.sh's hibernation |
| 20 | // exclusion. |
| 21 | package ops |
| 22 | |
| 23 | import ( |
| 24 | "bytes" |
| 25 | "errors" |
| 26 | "fmt" |
| 27 | "os" |
| 28 | "os/exec" |
| 29 | "path/filepath" |
| 30 | "sort" |
| 31 | "strconv" |
| 32 | "strings" |
| 33 | "time" |
| 34 | |
| 35 | "github.com/carlosframework/platform/internal/lease" |
| 36 | "github.com/carlosframework/platform/internal/provision" |
| 37 | "github.com/carlosframework/platform/internal/registry" |
| 38 | ) |
| 39 | |
| 40 | // litestreamVersion is the pinned release provisioned on the box when the |
| 41 | // binary is missing. Bump deliberately; the restore path is the thing a |
| 42 | // version change can silently break, and restore-verify is what catches it. |
| 43 | const litestreamVersion = "0.5.15" |
| 44 | |
| 45 | // retryDelay is the pause before restore-verify's one retry; a var so tests |
| 46 | // can shrink it. |
| 47 | var retryDelay = 10 * time.Second |
| 48 | |
| 49 | // SnapshotInterval and SnapshotRetention pin the fleet's point-in-time |
| 50 | // recovery window instead of inheriting litestream's 24h default. They are |
| 51 | // lease's constants, which hibernating instances' replicas use too, so the |
| 52 | // box-wide window and the per-instance one are one decision; lease.go |
| 53 | // records why a week. |
| 54 | const ( |
| 55 | SnapshotInterval = lease.SnapshotInterval |
| 56 | SnapshotRetention = lease.SnapshotRetention |
| 57 | ) |
| 58 | |
| 59 | // SnapshotPolicySupported reports whether `litestream version` output names a |
| 60 | // binary that HONOURS the snapshot stanza LitestreamConfig writes. The policy |
| 61 | // arrived with litestream 0.5; 0.3 and 0.4 parse the stanza and drop it on the |
| 62 | // floor. |
| 63 | // |
| 64 | // This exists because parsing is not evidence. Verified on the fleet |
| 65 | // 2026-08-15: the flagship's litestream v0.3.13 accepts the generated config, |
| 66 | // exits 0, lists the databases — and ignores the retention entirely. So does |
| 67 | // 0.5.15 when handed a misspelled key. A config that STATES a recovery window |
| 68 | // nothing enforces is worse than one that states none, because it reads as a |
| 69 | // decision someone made and checked. |
| 70 | // |
| 71 | // Unparseable output is UNSUPPORTED, never assumed-fine: the whole point is to |
| 72 | // refuse to claim a window we cannot show is real. |
| 73 | func SnapshotPolicySupported(versionOutput string) bool { |
| 74 | f := strings.Fields(strings.TrimSpace(versionOutput)) |
| 75 | if len(f) == 0 { |
| 76 | return false |
| 77 | } |
| 78 | v := strings.TrimPrefix(f[len(f)-1], "v") // "litestream v0.5.15" or "0.5.15" |
| 79 | major, rest, ok := strings.Cut(v, ".") |
| 80 | if !ok { |
| 81 | return false |
| 82 | } |
| 83 | minor, _, _ := strings.Cut(rest, ".") |
| 84 | maj, err := strconv.Atoi(major) |
| 85 | if err != nil { |
| 86 | return false |
| 87 | } |
| 88 | min, err := strconv.Atoi(minor) |
| 89 | if err != nil { |
| 90 | return false |
| 91 | } |
| 92 | if maj > 0 { |
| 93 | return true // a future 1.x must not regress to "unsupported" |
| 94 | } |
| 95 | return min >= 5 |
| 96 | } |
| 97 | |
| 98 | // litestreamVersionOutput runs `<bin> version`. A binary that is missing or |
| 99 | // will not answer yields "", which SnapshotPolicySupported treats as |
| 100 | // unsupported. |
| 101 | func litestreamVersionOutput(bin string) string { |
| 102 | out, err := exec.Command(bin, "version").CombinedOutput() |
| 103 | if err != nil { |
| 104 | return "" |
| 105 | } |
| 106 | return string(out) |
| 107 | } |
| 108 | |
| 109 | // LitestreamConfig renders the generated /etc/litestream.yml. Deliberately |
| 110 | // minimal — path + replica URL + region, plus the snapshot policy above — so |
| 111 | // it can't rot against litestream's config surface; other tuning belongs in |
| 112 | // flags on the unit, not here. The snapshot stanza is the considered |
| 113 | // exception to that minimalism: a recovery window is a decision, and leaving |
| 114 | // it to a library default meant nobody had made it. |
| 115 | // Paths are sorted so re-runs with an unchanged fleet produce a byte-identical |
| 116 | // file (the `litestream` verb's changed/unchanged report depends on it). |
| 117 | func LitestreamConfig(dbPaths []string, bucket, prefix, region string) string { |
| 118 | sorted := append([]string(nil), dbPaths...) |
| 119 | sort.Strings(sorted) |
| 120 | var b strings.Builder |
| 121 | b.WriteString("# GENERATED by `carlos ops litestream` — do not hand-edit; re-run the verb\n") |
| 122 | b.WriteString("# after `carlos add`/`remove` so the replication set follows the fleet.\n") |
| 123 | // Global snapshot policy: applies to every database below. |
| 124 | b.WriteString("snapshot:\n") |
| 125 | fmt.Fprintf(&b, " interval: %s\n", SnapshotInterval) |
| 126 | fmt.Fprintf(&b, " retention: %s\n", SnapshotRetention) |
| 127 | b.WriteString("dbs:\n") |
| 128 | for _, p := range sorted { |
| 129 | fmt.Fprintf(&b, " - path: %s\n", p) |
| 130 | b.WriteString(" replicas:\n") |
| 131 | fmt.Fprintf(&b, " - url: s3://%s/%s/%s\n", bucket, prefix, replicaKey(p)) |
| 132 | if region != "" { |
| 133 | fmt.Fprintf(&b, " region: %s\n", region) |
| 134 | } |
| 135 | } |
| 136 | return b.String() |
| 137 | } |
| 138 | |
| 139 | // replicaKey is a database's place in the replica bucket: its full path with |
| 140 | // the leading separator dropped, so /data/registry.db becomes |
| 141 | // data/registry.db. |
| 142 | // |
| 143 | // It is the full path and not the basename because the basename is NOT |
| 144 | // unique, and the consequence of pretending otherwise is severe. Every |
| 145 | // SQLite-backed tenant on this fleet calls its database registry.db or |
| 146 | // data.db, and the box's own control DB is /data/registry.db; a basename key |
| 147 | // collapses all of them onto one S3 prefix, where two litestream replicators |
| 148 | // interleave their WAL into a replica that restores as neither database. |
| 149 | // Tree discovery (treeDBs) turns that from a curiosity into the normal case — |
| 150 | // every tenant tree contains a registry.db. |
| 151 | // |
| 152 | // Keeping the path readable rather than hashing it is deliberate: the person |
| 153 | // reading this bucket is doing a restore, probably at speed, and needs to see |
| 154 | // which database each key is. |
| 155 | // |
| 156 | // Changing this scheme moves every database to a new replica lineage, so it |
| 157 | // is a one-time freedom: no box has ever run a config generated by this verb |
| 158 | // (the flagship's is hand-written, tito's comes from the legacy `titogo ops |
| 159 | // litestream`), so there is no live lineage to break. There will be after the |
| 160 | // first cutover — treat it as fixed then. |
| 161 | func replicaKey(dbPath string) string { |
| 162 | return strings.TrimPrefix(filepath.Clean(dbPath), "/") |
| 163 | } |
| 164 | |
| 165 | // ParseLitestreamDBs recovers the database paths from a generated config. |
| 166 | // It reads only the exact shape LitestreamConfig writes (the round-trip is |
| 167 | // pinned by TestConfigRoundTrip) — it is not a YAML parser. |
| 168 | func ParseLitestreamDBs(cfg string) []string { |
| 169 | var dbs []string |
| 170 | for _, line := range strings.Split(cfg, "\n") { |
| 171 | if p, ok := strings.CutPrefix(line, " - path: "); ok { |
| 172 | dbs = append(dbs, strings.TrimSpace(p)) |
| 173 | } |
| 174 | } |
| 175 | return dbs |
| 176 | } |
| 177 | |
| 178 | // CoverageShortfall names every database the EXISTING config replicates that |
| 179 | // the proposed set does not — the databases a write would stop replicating. |
| 180 | // |
| 181 | // This is the guard for the one genuinely dangerous moment in this verb's |
| 182 | // life: cutting a box over from a hand-maintained config to a generated one. |
| 183 | // The generated list comes from the registry, so anything the registry does |
| 184 | // not know about is simply absent from it — and nothing downstream would say |
| 185 | // so. Replication of the dropped database stops, and the drill keeps passing |
| 186 | // because the drill only checks what the config names. The failure is |
| 187 | // invisible in both directions at once. |
| 188 | // |
| 189 | // Growing the set is always fine and needs no ceremony; shrinking it is a |
| 190 | // decision someone has to make on purpose. |
| 191 | func CoverageShortfall(existingConfig string, proposed []string) []string { |
| 192 | have := map[string]bool{} |
| 193 | for _, p := range proposed { |
| 194 | have[filepath.Clean(p)] = true |
| 195 | } |
| 196 | var missing []string |
| 197 | for _, p := range ParseLitestreamDBs(existingConfig) { |
| 198 | if !have[filepath.Clean(p)] { |
| 199 | missing = append(missing, p) |
| 200 | } |
| 201 | } |
| 202 | sort.Strings(missing) |
| 203 | return missing |
| 204 | } |
| 205 | |
| 206 | // FleetDBs lists the non-empty *.db files this box-wide replicator owns: |
| 207 | // everything found by globbing dataDir (the agent's own exec-child |
| 208 | // databases, plus the registry itself), PLUS — separately, not via the glob |
| 209 | // — every BackingUnit route's DBPath when the operator set one. |
| 210 | // |
| 211 | // WHAT IS COVERED: a unit tenant (carlos-app@<host>: DynamicUser, |
| 212 | // StateDirectory=/var/lib/carlos-app/<host>) whose route was added with an |
| 213 | // explicit `-db <path>`, hibernating or not — its wake/sleep is systemctl |
| 214 | // and the agent can't even read a DynamicUser's 0700 state dir, so this |
| 215 | // box-wide replicator (running as root) is the only thing that can stream |
| 216 | // its writes off-box. |
| 217 | // |
| 218 | // WHAT IS NOT COVERED, and why this function does not try to guess it: a |
| 219 | // unit tenant's database filename inside its state dir is the APP's own |
| 220 | // choice — the platform enforces no naming convention there the way it does |
| 221 | // the dataDir/<host>.db convention for exec children — so there is no |
| 222 | // honest default path to assume. A unit route added without -db is |
| 223 | // silently invisible here. cli.go's runAdd requires -db for |
| 224 | // `-backing unit -hibernate` routes for exactly this reason (going |
| 225 | // unreplicated there is a genuine data-loss risk, not just a missed |
| 226 | // nicety); a non-hibernating unit tenant without -db is still the |
| 227 | // operator's to hand-configure. See internal/activator/backend_unit.go's |
| 228 | // package comment for the fuller operator-facing story. |
| 229 | // |
| 230 | // Hibernating EXEC-backed routes' databases (matched by DBPath, found via |
| 231 | // the dataDir glob) are excluded from the results below: the activator's |
| 232 | // embedded Litestream owns their replication (to the instances bucket, |
| 233 | // under the lease) — a second replicator on the same WAL would fight it for |
| 234 | // checkpointing and corrupt the replica. (Eleven's fleet-backup.sh lesson, |
| 235 | // carried verbatim.) That exclusion never applies to BackingUnit routes — |
| 236 | // their replication has no other owner at all, so they are added here |
| 237 | // rather than filtered. |
| 238 | // treeDBWalkDepth bounds the descent. It is a loop guard, not a policy: os.Stat |
| 239 | // follows symlinks (which it must — systemd's StateDirectory is a symlink into |
| 240 | // /var/lib/private), and a symlink pointing at an ancestor would otherwise |
| 241 | // recurse forever. No real tenant layout is anywhere near this deep. |
| 242 | const treeDBWalkDepth = 8 |
| 243 | |
| 244 | // treeDBs lists the *.db files beneath root. |
| 245 | // |
| 246 | // It exists because a unit tenant's "database" is sometimes a TREE. kass on |
| 247 | // the flagship keeps registry.db plus an instances/ directory that grows as the |
| 248 | // tenant creates instances; a single -db path cannot say that, which is why |
| 249 | // that box's config was still hand-maintained — and why it had silently |
| 250 | // drifted, missing an instance database created after the list was written. |
| 251 | // |
| 252 | // Two things are deliberately skipped: |
| 253 | // |
| 254 | // - dot-prefixed directories, which is where litestream keeps its own shadow |
| 255 | // WAL (".<db>-litestream"). Replicating a replicator's bookkeeping would be |
| 256 | // absurd, and worse, some of those files are named *.db. |
| 257 | // - anything that is not a non-empty *.db. Sidecars (-wal, -shm) travel with |
| 258 | // their database and are not replicated separately; an empty file is a |
| 259 | // placeholder, not a database. |
| 260 | // |
| 261 | // Discovery requires the .db convention, unlike a single-file DBPath which is |
| 262 | // taken verbatim: when the operator names one file we believe them, but when |
| 263 | // we go looking we only claim what we can recognise. |
| 264 | func treeDBs(root string) []string { |
| 265 | var out []string |
| 266 | var walk func(dir string, depth int) |
| 267 | walk = func(dir string, depth int) { |
| 268 | if depth > treeDBWalkDepth { |
| 269 | return |
| 270 | } |
| 271 | entries, err := os.ReadDir(dir) |
| 272 | if err != nil { |
| 273 | return // unreadable subtree: nothing to claim, and not this pass's business to fail |
| 274 | } |
| 275 | for _, e := range entries { |
| 276 | name := e.Name() |
| 277 | if strings.HasPrefix(name, ".") { |
| 278 | continue |
| 279 | } |
| 280 | p := filepath.Join(dir, name) |
| 281 | // Stat, not e.Info(): entries may be symlinks and the target is |
| 282 | // what matters. |
| 283 | fi, err := os.Stat(p) |
| 284 | if err != nil { |
| 285 | continue |
| 286 | } |
| 287 | if fi.IsDir() { |
| 288 | walk(p, depth+1) |
| 289 | continue |
| 290 | } |
| 291 | if !strings.HasSuffix(name, ".db") || fi.Size() == 0 { |
| 292 | continue |
| 293 | } |
| 294 | out = append(out, p) |
| 295 | } |
| 296 | } |
| 297 | walk(root, 0) |
| 298 | return out |
| 299 | } |
| 300 | |
| 301 | // consoleDBPath resolves where this box's console keeps its database: |
| 302 | // CONSOLE_DB when set (the same env var cmd/console reads, so a relocated |
| 303 | // console cannot silently drop out of replication), else the production |
| 304 | // layout <data>/console/console.db — expressed against dataDir rather than |
| 305 | // hardcoded to /data so a test box's console is found under its own root and |
| 306 | // never the real one. |
| 307 | func consoleDBPath(dataDir string) string { |
| 308 | if p := strings.TrimSpace(os.Getenv("CONSOLE_DB")); p != "" { |
| 309 | return filepath.Clean(p) |
| 310 | } |
| 311 | return filepath.Join(dataDir, "console", "console.db") |
| 312 | } |
| 313 | |
| 314 | func FleetDBs(reg *registry.Registry, dataDir string) ([]string, error) { |
| 315 | dbs, _, err := FleetDBsWithSkipped(reg, dataDir, nil) |
| 316 | return dbs, err |
| 317 | } |
| 318 | |
| 319 | // LayoutJournalPresent is the layout pass's "held" predicate for dataDir: a |
| 320 | // host has a migration journal unless its journal path is definitely absent. |
| 321 | // A journal that cannot be read is not proof there is none. It is the same |
| 322 | // test internal/layout's JournalPresent makes, spelled out here so this |
| 323 | // package does not depend on the pass. |
| 324 | func LayoutJournalPresent(dataDir string) func(host string) bool { |
| 325 | return func(host string) bool { |
| 326 | _, err := os.Lstat(provision.LayoutJournalPath(dataDir, host)) |
| 327 | return !errors.Is(err, os.ErrNotExist) |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | // FleetDBsWithSkipped is FleetDBs, plus the files under dataDir it declined to |
| 332 | // claim — see the "glob proposes, registry disposes" note inside. The verb |
| 333 | // prints them; nothing here is dropped without being named. |
| 334 | // |
| 335 | // fenced, when non-nil, names hosts the layout pass holds mid-move. Such a |
| 336 | // host's row does not claim its database, so the file is named among the |
| 337 | // skipped instead of being replicated from a directory it is leaving. |
| 338 | func FleetDBsWithSkipped(reg *registry.Registry, dataDir string, fenced func(host string) bool) (dbs []string, skipped []string, err error) { |
| 339 | routes, err := reg.List() |
| 340 | if err != nil { |
| 341 | return nil, nil, err |
| 342 | } |
| 343 | hibernating := map[string]bool{} |
| 344 | unitPaths := map[string]bool{} |
| 345 | for _, r := range routes { |
| 346 | if r.Backing == registry.BackingUnit { |
| 347 | // Unit tenants' data lives outside dataDir (their systemd |
| 348 | // StateDirectory), so the glob below can never find them on its |
| 349 | // own — DBPath, when the operator set it, is the only way this |
| 350 | // replicator learns where they are. |
| 351 | if r.DBPath != "" { |
| 352 | unitPaths[filepath.Clean(r.DBPath)] = true |
| 353 | } |
| 354 | continue |
| 355 | } |
| 356 | if r.Hibernate && r.DBPath != "" { |
| 357 | hibernating[filepath.Clean(r.DBPath)] = true |
| 358 | } |
| 359 | } |
| 360 | |
| 361 | matches, err := filepath.Glob(filepath.Join(dataDir, "*.db")) |
| 362 | if err != nil { |
| 363 | return nil, nil, err |
| 364 | } |
| 365 | // A nested row's DB lives one level down, under |
| 366 | // <dataDir>/instances/<id>/<host>.db (isolated-exec design §7): the flat |
| 367 | // glob above never walks into that directory, so its matches are folded |
| 368 | // in here before the claimed-files gate below, which already knows how |
| 369 | // to accept or skip each one. A nested row's DBPath was added to |
| 370 | // claimed by the very same loop that adds a flat row's. |
| 371 | nested, err := filepath.Glob(filepath.Join(dataDir, "instances", "*", "*.db")) |
| 372 | if err != nil { |
| 373 | return nil, nil, err |
| 374 | } |
| 375 | matches = append(matches, nested...) |
| 376 | // THE GLOB PROPOSES, THE REGISTRY DISPOSES. A *.db under dataDir is |
| 377 | // replicated when a route claims it, or when it is the box's own registry |
| 378 | // or console. Anything else is somebody's leftover, and streaming leftovers |
| 379 | // off-box costs storage forever: a box with a hand-written config can |
| 380 | // accumulate a large, root-owned, unrelated *.db file in /data that |
| 381 | // matches this glob exactly and would otherwise escape replication |
| 382 | // silently. |
| 383 | // |
| 384 | // Skipped files are NAMED to the caller, never dropped in silence — a |
| 385 | // database that someone did want, quietly left out, is the precise failure |
| 386 | // this area keeps producing. The shrink guard (CoverageShortfall) is the |
| 387 | // second net: if a skipped file is one the live config replicates today, |
| 388 | // the write is refused outright. |
| 389 | claimed := map[string]bool{ |
| 390 | filepath.Clean(filepath.Join(dataDir, "registry.db")): true, |
| 391 | filepath.Clean(consoleDBPath(dataDir)): true, |
| 392 | } |
| 393 | for _, r := range routes { |
| 394 | if r.DBPath == "" || (fenced != nil && fenced(r.Host)) { |
| 395 | continue |
| 396 | } |
| 397 | claimed[filepath.Clean(r.DBPath)] = true |
| 398 | } |
| 399 | var globbed []string |
| 400 | for _, p := range matches { |
| 401 | if claimed[filepath.Clean(p)] { |
| 402 | globbed = append(globbed, p) |
| 403 | continue |
| 404 | } |
| 405 | skipped = append(skipped, filepath.Clean(p)) |
| 406 | } |
| 407 | matches = globbed |
| 408 | |
| 409 | // Unit-tenant DBPaths are added on top of the glob: they typically live |
| 410 | // entirely outside dataDir, at their StateDirectory path. A DBPath naming |
| 411 | // a DIRECTORY is a tree — see treeDBs. These are claimed by construction: |
| 412 | // a route named them. |
| 413 | for p := range unitPaths { |
| 414 | if fi, err := os.Stat(p); err == nil && fi.IsDir() { |
| 415 | matches = append(matches, treeDBs(p)...) |
| 416 | continue |
| 417 | } |
| 418 | matches = append(matches, p) |
| 419 | } |
| 420 | // And the console's own database, for the same reason in miniature: it |
| 421 | // sits at <data>/console/console.db — ONE DIRECTORY BELOW the flat glob, |
| 422 | // which is exactly why it went unreplicated. It holds no platform state |
| 423 | // (routes and channels are read live) but it holds every account, |
| 424 | // ownership row, session and the deployment's Keymail credentials: losing |
| 425 | // it loses who everyone is. Boxes without a console simply have no file |
| 426 | // here and the loop below drops it. |
| 427 | matches = append(matches, consoleDBPath(dataDir)) |
| 428 | |
| 429 | seen := map[string]bool{} |
| 430 | for _, p := range matches { |
| 431 | p = filepath.Clean(p) |
| 432 | if seen[p] { |
| 433 | continue // a unit DBPath that happens to also sit under dataDir |
| 434 | } |
| 435 | fi, err := os.Stat(p) |
| 436 | if err != nil || !fi.Mode().IsRegular() || fi.Size() == 0 { |
| 437 | continue // gone, odd, or an empty placeholder — nothing to replicate |
| 438 | } |
| 439 | if hibernating[p] { |
| 440 | continue |
| 441 | } |
| 442 | seen[p] = true |
| 443 | dbs = append(dbs, p) |
| 444 | } |
| 445 | sort.Strings(skipped) |
| 446 | return dbs, skipped, nil |
| 447 | } |
| 448 | |
| 449 | // restoreAndCheckFn is the per-database restore behind a package var, so the |
| 450 | // one-at-a-time property below is testable without S3 or a litestream binary. |
| 451 | var restoreAndCheckFn = restoreAndCheck |
| 452 | |
| 453 | // RestoreVerify restores every database in the generated config from S3 into |
| 454 | // tmpDir and proves each opens clean. Returns an error on any failure so the |
| 455 | // caller (a oneshot unit) can exit non-zero where the fleet sweeps see it. |
| 456 | // |
| 457 | // Exactly ONE restored copy exists at a time: each is deleted as soon as it |
| 458 | // has been checked. This is a memory constraint, not housekeeping — |
| 459 | // carlos-restore-verify.service runs PrivateTmp=yes, so tmpDir is a tmpfs and |
| 460 | // every byte written there is charged to the unit's own cgroup against |
| 461 | // MemoryMax. Keeping all of them meant the drill's peak was the SUM of the |
| 462 | // fleet's databases: measured on carlos-tito-ie 2026-08-15, 54 databases |
| 463 | // totalling 247M under MemoryMax=256M, OOM-killed at a "256.2M memory peak" |
| 464 | // after ~40 clean restores — a drill that grew its own failure as the fleet |
| 465 | // grew. The peak is now the largest single database. |
| 466 | func RestoreVerify(litestreamBin, cfgPath, tmpDir string) error { |
| 467 | raw, err := os.ReadFile(cfgPath) |
| 468 | if err != nil { |
| 469 | return fmt.Errorf("restore-verify: %v", err) |
| 470 | } |
| 471 | dbs := ParseLitestreamDBs(string(raw)) |
| 472 | if len(dbs) == 0 { |
| 473 | return fmt.Errorf("restore-verify: no databases in %s — regenerate it: carlos ops litestream", cfgPath) |
| 474 | } |
| 475 | |
| 476 | failures, stale := 0, 0 |
| 477 | for _, db := range dbs { |
| 478 | // Sweep this database's copy however the check ends — a failure must |
| 479 | // not leave the wreckage behind for the next restore to be charged for. |
| 480 | out := filepath.Join(tmpDir, filepath.Base(db)) |
| 481 | err := restoreAndCheckFn(litestreamBin, cfgPath, db, tmpDir) |
| 482 | if err != nil { |
| 483 | removeRestored(out) |
| 484 | // One retry after a pause: a database created moments ago may not |
| 485 | // have finished its first replication pass yet. |
| 486 | time.Sleep(retryDelay) |
| 487 | err = restoreAndCheckFn(litestreamBin, cfgPath, db, tmpDir) |
| 488 | } |
| 489 | // Freshness is judged BEFORE the copy is swept, and only once the |
| 490 | // restore itself passed: "restores clean but stopped tracking its |
| 491 | // source" is a different finding from "does not restore", and |
| 492 | // conflating them would hide whichever came second. |
| 493 | var fresh freshness |
| 494 | var freshDetail string |
| 495 | if err == nil { |
| 496 | fresh, freshDetail = replicaFreshness(db, out, sourceModTime(db)) |
| 497 | } |
| 498 | removeRestored(out) // every path, success included |
| 499 | if err != nil { |
| 500 | failures++ |
| 501 | fmt.Printf("FAIL %s: %v\n", db, err) |
| 502 | continue |
| 503 | } |
| 504 | switch fresh { |
| 505 | case freshnessStale: |
| 506 | stale++ |
| 507 | fmt.Printf("STALE %s: %s\n", db, freshDetail) |
| 508 | case freshnessNoSource: |
| 509 | stale++ |
| 510 | fmt.Printf("STALE %s: %s\n", db, freshDetail) |
| 511 | case freshnessUnknown: |
| 512 | // Never a pass and never an alarm — see the freshness doc. |
| 513 | fmt.Printf("ok %s (freshness unknown: %s)\n", db, freshDetail) |
| 514 | default: |
| 515 | fmt.Printf("ok %s\n", db) |
| 516 | } |
| 517 | } |
| 518 | switch { |
| 519 | case failures > 0 && stale > 0: |
| 520 | return fmt.Errorf("restore-verify: %d of %d replicas failed to restore clean, and %d more restore but have stopped tracking their source", |
| 521 | failures, len(dbs), stale) |
| 522 | case failures > 0: |
| 523 | return fmt.Errorf("restore-verify: %d of %d replicas failed to restore clean", failures, len(dbs)) |
| 524 | case stale > 0: |
| 525 | // A stale replica restores perfectly and is still a durability |
| 526 | // failure: what comes back is the database as it was when replication |
| 527 | // stopped. It fails the run for the same reason an unrestorable one |
| 528 | // does. |
| 529 | return fmt.Errorf("restore-verify: %d of %d replicas restore clean but have STOPPED TRACKING their source — replication is broken for them", |
| 530 | stale, len(dbs)) |
| 531 | } |
| 532 | fmt.Printf("restore-verify: all %d replicas restore clean and are current\n", len(dbs)) |
| 533 | return nil |
| 534 | } |
| 535 | |
| 536 | // removeRestored deletes one restored copy and the sidecars an open may have |
| 537 | // attached beside it (CheckRestoredDB's read-only DSN does not stop the driver |
| 538 | // creating a -wal). Errors are dropped: this is reclamation, and a file that |
| 539 | // is already gone is the outcome wanted. |
| 540 | func removeRestored(path string) { |
| 541 | for _, p := range []string{path, path + "-wal", path + "-shm"} { |
| 542 | _ = os.Remove(p) |
| 543 | } |
| 544 | } |
| 545 | |
| 546 | // restoreAndCheck pulls one database back from its replica and runs the same |
| 547 | // checks a disaster recovery would lean on: the file opens, integrity_check |
| 548 | // says ok, and the schema is non-empty. |
| 549 | func restoreAndCheck(lsBin, cfgPath, dbPath, tmpDir string) error { |
| 550 | out := filepath.Join(tmpDir, filepath.Base(dbPath)) |
| 551 | _ = os.Remove(out) // a retry must not trip over the failed attempt's file |
| 552 | cmd := exec.Command(lsBin, "restore", "-config", cfgPath, "-o", out, dbPath) |
| 553 | if b, err := cmd.CombinedOutput(); err != nil { |
| 554 | return fmt.Errorf("litestream restore: %v: %s", err, strings.TrimSpace(string(b))) |
| 555 | } |
| 556 | return CheckRestoredDB(out) |
| 557 | } |
| 558 | |
| 559 | // CheckRestoredDB opens a restored file read-only and proves it is a sane |
| 560 | // SQLite database: integrity_check == ok and at least one table present. |
| 561 | // |
| 562 | // The implementation moved to internal/lease so the WAKE path can run the |
| 563 | // same check without importing this package — one definition of "restores |
| 564 | // clean", used by both the daily sweep and the moment an instance is about to |
| 565 | // serve a restored file. The name stays here because the ops verb and its |
| 566 | // tests are its documented door. |
| 567 | func CheckRestoredDB(path string) error { return lease.CheckDB(path) } |
| 568 | |
| 569 | // writeIfChanged writes content to path only when it differs, and reports |
| 570 | // whether it wrote. The byte-identical no-op is what lets a timer re-run the |
| 571 | // generator freely without a restart-worthy mtime/content churn. |
| 572 | func writeIfChanged(path, content string) (bool, error) { |
| 573 | if old, err := os.ReadFile(path); err == nil && bytes.Equal(old, []byte(content)) { |
| 574 | return false, nil |
| 575 | } |
| 576 | if err := os.WriteFile(path, []byte(content), 0o644); err != nil { |
| 577 | return false, err |
| 578 | } |
| 579 | return true, nil |
| 580 | } |
| 581 | |
amadan