| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/binary" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | |
| 13 | "github.com/benbjohnson/litestream" |
| 14 | "github.com/superfly/ltx" |
| 15 | _ "modernc.org/sqlite" |
| 16 | ) |
| 17 | |
| 18 | // restoreForVerify is Restore behind a package var so the drill's three |
| 19 | // outcomes (clean, damaged, never-replicated) are testable without S3. |
| 20 | var restoreForVerify = Restore |
| 21 | |
| 22 | // ErrNoReplica reports that a prefix holds no backup at all — a brand-new |
| 23 | // instance that has never replicated, not a damaged one. Verification callers |
| 24 | // must keep the two apart: "nothing to restore yet" is not a failure, and |
| 25 | // reporting it as one would train operators to ignore the alarm. |
| 26 | var ErrNoReplica = errors.New("no replica in prefix") |
| 27 | |
| 28 | // CheckDB opens path read-only and proves it is a sane SQLite database: |
| 29 | // integrity_check == ok and at least one table present. It is the one check a |
| 30 | // disaster recovery leans on, so it lives here — beside Restore — where both |
| 31 | // the wake path (internal/activator) and the verification pass (internal/ops) |
| 32 | // can reach it without either importing the other. |
| 33 | // |
| 34 | // The schema check is not pedantry: a restore that yields a valid but empty |
| 35 | // database passes integrity_check and would be useless in the recovery it was |
| 36 | // meant for. |
| 37 | func CheckDB(path string) error { |
| 38 | return CheckDBContext(context.Background(), path) |
| 39 | } |
| 40 | |
| 41 | // CheckDBContext is CheckDB that stops when ctx does: the integrity scan |
| 42 | // reads every page, and a caller that has been asked to stop (carlos db, |
| 43 | // on Ctrl-C) should not finish a full scan first. |
| 44 | func CheckDBContext(ctx context.Context, path string) error { |
| 45 | db, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| 46 | if err != nil { |
| 47 | return err |
| 48 | } |
| 49 | defer db.Close() |
| 50 | db.SetMaxOpenConns(1) |
| 51 | // busy_timeout first, always (Global Constraints) — and on a garbage file |
| 52 | // this first statement is what fails loudly instead of a later query. |
| 53 | if _, err := db.ExecContext(ctx, "PRAGMA busy_timeout=5000"); err != nil { |
| 54 | return fmt.Errorf("open: %w", err) |
| 55 | } |
| 56 | var res string |
| 57 | if err := db.QueryRowContext(ctx, "PRAGMA integrity_check").Scan(&res); err != nil { |
| 58 | return fmt.Errorf("integrity_check: %w", err) |
| 59 | } |
| 60 | if res != "ok" { |
| 61 | return fmt.Errorf("integrity_check: %s", res) |
| 62 | } |
| 63 | var tables int |
| 64 | if err := db.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE type='table'").Scan(&tables); err != nil { |
| 65 | return fmt.Errorf("schema count: %w", err) |
| 66 | } |
| 67 | if tables == 0 { |
| 68 | return errors.New("restored database has no tables") |
| 69 | } |
| 70 | return nil |
| 71 | } |
| 72 | |
| 73 | // VerifyReplica restores s3://bucket/prefix into a throwaway file under dir |
| 74 | // and proves it opens clean, then deletes it. It returns ErrNoReplica when |
| 75 | // the prefix has never been replicated to, and a describing error when the |
| 76 | // restore or the check fails. |
| 77 | // |
| 78 | // IT DELIBERATELY DOES NOT TAKE THE LEASE, and that is safe in a way Restore |
| 79 | // itself is not. Restore's contract requires the lease because it materializes |
| 80 | // the LIVE database: an unleased restore there can resurrect a stale copy that |
| 81 | // later replicates over the true holder's newer state. Here the output path is |
| 82 | // a temp file that is never replicated from and is removed before return, so |
| 83 | // there is nothing that could shadow anything. What the lease forbids is a |
| 84 | // second WRITER; a reader costs the holder nothing — the box-wide |
| 85 | // `carlos ops restore-verify` has always restored databases that its own |
| 86 | // replicator was actively streaming, and litestream restores a committed |
| 87 | // position rather than a live tail. |
| 88 | func VerifyReplica(ctx context.Context, bucket, prefix, region, dir string) (uint32, error) { |
| 89 | out := filepath.Join(dir, "verify-"+filepath.Base(prefix)+".db") |
| 90 | // A retry must not trip over a previous attempt's file: litestream refuses |
| 91 | // to restore over an existing output path. |
| 92 | _ = os.Remove(out) |
| 93 | // Sweep on EVERY path, including a failed check. The sidecars are swept |
| 94 | // too: CheckDB's "mode=ro" does not stop the driver attaching a WAL, and a |
| 95 | // drill that left a copy of every tenant's database on the box would be |
| 96 | // both a disk bug and a data-exposure one. |
| 97 | defer func() { |
| 98 | _ = os.Remove(out) |
| 99 | _ = os.Remove(out + "-wal") |
| 100 | _ = os.Remove(out + "-shm") |
| 101 | }() |
| 102 | |
| 103 | restored, err := restoreForVerify(ctx, out, bucket, prefix, region) |
| 104 | if err != nil { |
| 105 | return 0, fmt.Errorf("restore: %w", err) |
| 106 | } |
| 107 | if !restored { |
| 108 | return 0, ErrNoReplica |
| 109 | } |
| 110 | if err := CheckDB(out); err != nil { |
| 111 | return 0, err |
| 112 | } |
| 113 | // The counter is read BEFORE the deferred sweep removes the copy: it is |
| 114 | // what lets the caller ask the second question, whether this replica is |
| 115 | // still tracking its source (see ops.replicaFreshness). |
| 116 | return ChangeCounter(out) |
| 117 | } |
| 118 | |
| 119 | // ChangeCounter reads the SQLite file-change counter: bytes 24-27 of the |
| 120 | // database header, big-endian. SQLite increments it on every write transaction |
| 121 | // that modifies the main database file, which makes it the cheapest honest |
| 122 | // answer to "is this replica still tracking its source?" — compare the number |
| 123 | // on the live database with the number on a restored copy. |
| 124 | // |
| 125 | // It lives here, beside CheckDB and Restore, because both drills need it: the |
| 126 | // box-wide one (internal/ops) and the hibernation one, which restores through |
| 127 | // VerifyReplica. |
| 128 | // |
| 129 | // In WAL mode the counter in the main file only advances at a checkpoint, so a |
| 130 | // source with a hot WAL reads LOWER than reality. That skew is deliberately |
| 131 | // harmless: it can only make a stale replica look current (a missed alarm), |
| 132 | // never a current replica look stale (a false one). |
| 133 | func ChangeCounter(path string) (uint32, error) { |
| 134 | f, err := os.Open(path) |
| 135 | if err != nil { |
| 136 | return 0, err |
| 137 | } |
| 138 | defer f.Close() |
| 139 | |
| 140 | var header [28]byte |
| 141 | if _, err := io.ReadFull(f, header[:]); err != nil { |
| 142 | return 0, fmt.Errorf("read header: %w", err) |
| 143 | } |
| 144 | if string(header[:15]) != "SQLite format 3" { |
| 145 | return 0, fmt.Errorf("not a SQLite database") |
| 146 | } |
| 147 | return binary.BigEndian.Uint32(header[24:28]), nil |
| 148 | } |
| 149 | |
| 150 | // HasReplica reports whether s3://bucket/prefix holds any LTX file at any |
| 151 | // level: the seed fallback's "has this host a replica of its own" (seeded |
| 152 | // instances spec §3.4 step 1). A listing error is returned, never read as |
| 153 | // "no replica". |
| 154 | func HasReplica(ctx context.Context, bucket, prefix, region string) (bool, error) { |
| 155 | region, err := resolveRegion(ctx, region) |
| 156 | if err != nil { |
| 157 | return false, err |
| 158 | } |
| 159 | client, err := newReplicaClient(bucket, prefix, region) |
| 160 | if err != nil { |
| 161 | return false, err |
| 162 | } |
| 163 | return hasReplicaIn(ctx, client) |
| 164 | } |
| 165 | |
| 166 | // ltxLister is the one replica-client method HasReplica needs. |
| 167 | type ltxLister interface { |
| 168 | LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) |
| 169 | } |
| 170 | |
| 171 | func hasReplicaIn(ctx context.Context, client ltxLister) (bool, error) { |
| 172 | for level := 0; level <= litestream.SnapshotLevel; level++ { |
| 173 | itr, err := client.LTXFiles(ctx, level, 0, false) |
| 174 | if err != nil { |
| 175 | return false, err |
| 176 | } |
| 177 | found := itr.Next() |
| 178 | ierr := itr.Err() |
| 179 | cerr := itr.Close() |
| 180 | if ierr != nil { |
| 181 | return false, ierr |
| 182 | } |
| 183 | if cerr != nil { |
| 184 | return false, cerr |
| 185 | } |
| 186 | if found { |
| 187 | return true, nil |
| 188 | } |
| 189 | } |
| 190 | return false, nil |
| 191 | } |
| 192 | |
amadan