| 1 | package ops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "slices" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/activator" |
| 14 | "github.com/carlosframework/platform/internal/dbverify" |
| 15 | "github.com/carlosframework/platform/internal/lease" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | ) |
| 18 | |
| 19 | func at(s string) time.Time { |
| 20 | t, err := time.Parse(time.RFC3339, s) |
| 21 | if err != nil { |
| 22 | panic(err) |
| 23 | } |
| 24 | return t |
| 25 | } |
| 26 | |
| 27 | // The two passes must partition the fleet: every database is owned by exactly |
| 28 | // one of them. This test is the mirror of TestConfigGenSkipsHibernatingRoutes |
| 29 | // — what that one proves is EXCLUDED here must be INCLUDED, or it is drilled |
| 30 | // by nobody. |
| 31 | func TestHibernatingHostsIsTheComplementOfFleetDBs(t *testing.T) { |
| 32 | dir := t.TempDir() |
| 33 | reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| 34 | if err != nil { |
| 35 | t.Fatal(err) |
| 36 | } |
| 37 | unitDir := t.TempDir() |
| 38 | |
| 39 | awake := filepath.Join(dir, "a.example.com.db") |
| 40 | asleep := filepath.Join(dir, "b.example.com.db") |
| 41 | unitTenant := filepath.Join(unitDir, "state.db") |
| 42 | for _, p := range []string{awake, asleep, unitTenant} { |
| 43 | if err := os.WriteFile(p, []byte("x"), 0o644); err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | } |
| 47 | if err := reg.Upsert(registry.Route{Host: "a.example.com", DBPath: awake}); err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | if err := reg.Upsert(registry.Route{Host: "b.example.com", DBPath: asleep, Hibernate: true}); err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | if err := reg.Upsert(registry.Route{Host: "c.example.com", DBPath: unitTenant, |
| 54 | Hibernate: true, Backing: registry.BackingUnit}); err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | |
| 58 | hosts := hostsOf(mustHibernating(t, reg)) |
| 59 | if !slices.Contains(hosts, "b.example.com") { |
| 60 | t.Fatalf("the hibernating exec route is drilled by nobody: %v", hosts) |
| 61 | } |
| 62 | if slices.Contains(hosts, "a.example.com") { |
| 63 | t.Fatalf("an awake route is claimed by both passes: %v", hosts) |
| 64 | } |
| 65 | if slices.Contains(hosts, "c.example.com") { |
| 66 | t.Fatalf("a unit tenant (box-wide replicator's job) is claimed by both passes: %v", hosts) |
| 67 | } |
| 68 | |
| 69 | // And nothing this pass claims may appear in the box-wide config. |
| 70 | dbs, err := FleetDBs(reg, dir) |
| 71 | if err != nil { |
| 72 | t.Fatal(err) |
| 73 | } |
| 74 | cfg := LitestreamConfig(dbs, "b", "p", "eu-west-1") |
| 75 | for _, h := range hosts { |
| 76 | if strings.Contains(cfg, h+".db") { |
| 77 | t.Fatalf("%s is replicated by BOTH passes — two replicators on one WAL:\n%s", h, cfg) |
| 78 | } |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | // routes builds the drill's input from bare host names. The DBPath is left |
| 83 | // empty on purpose: these tests drive the RESTORE outcomes, and an absent |
| 84 | // source is reported as its own verdict, which TestVerifyHibernatingStale |
| 85 | // covers explicitly. |
| 86 | func routes(hosts ...string) []HibernatingRoute { |
| 87 | out := make([]HibernatingRoute, 0, len(hosts)) |
| 88 | for _, h := range hosts { |
| 89 | out = append(out, HibernatingRoute{Host: h}) |
| 90 | } |
| 91 | return out |
| 92 | } |
| 93 | |
| 94 | func hostsOf(rs []HibernatingRoute) []string { |
| 95 | out := make([]string, 0, len(rs)) |
| 96 | for _, r := range rs { |
| 97 | out = append(out, r.Host) |
| 98 | } |
| 99 | return out |
| 100 | } |
| 101 | |
| 102 | func mustHibernating(t *testing.T, reg *registry.Registry) []HibernatingRoute { |
| 103 | t.Helper() |
| 104 | got, err := HibernatingHosts(reg) |
| 105 | if err != nil { |
| 106 | t.Fatal(err) |
| 107 | } |
| 108 | return got |
| 109 | } |
| 110 | |
| 111 | // swapVerifyOne points the restore seam at a stub. |
| 112 | func swapVerifyOne(t *testing.T, fn func(ctx context.Context, bucket, prefix, region, dir string) (uint32, error)) { |
| 113 | t.Helper() |
| 114 | prev := verifyOne |
| 115 | verifyOne = fn |
| 116 | t.Cleanup(func() { verifyOne = prev }) |
| 117 | } |
| 118 | |
| 119 | // The three verdicts, and — most importantly — that an S3 failure is |
| 120 | // "unverified", not a corruption alarm, and a never-woken instance is neither. |
| 121 | func TestVerifyHibernatingVerdicts(t *testing.T) { |
| 122 | now := at("2026-08-15T09:00:00Z") |
| 123 | swapVerifyOne(t, func(_ context.Context, _, prefix, _, _ string) (uint32, error) { |
| 124 | switch { |
| 125 | case strings.Contains(prefix, "good"): |
| 126 | return 1, nil |
| 127 | case strings.Contains(prefix, "damaged"): |
| 128 | return 0, errors.New("integrity_check: page 4 corrupt") |
| 129 | case strings.Contains(prefix, "new"): |
| 130 | return 0, lease.ErrNoReplica |
| 131 | } |
| 132 | return 0, errors.New("unexpected host") |
| 133 | }) |
| 134 | |
| 135 | cfg := &activator.Config{Bucket: "instances", Prefix: "hib", Region: "eu-west-1"} |
| 136 | got := VerifyHibernating(context.Background(), cfg, |
| 137 | routes("good.example.com", "damaged.example.com", "new.example.com"), 0, now, t.TempDir()) |
| 138 | |
| 139 | if v := got["good.example.com"]; !v.OK || v.Unverified { |
| 140 | t.Fatalf("clean replica verdict = %+v", v) |
| 141 | } |
| 142 | if v := got["damaged.example.com"]; v.OK || v.Unverified { |
| 143 | t.Fatalf("damaged replica must be a failure, got %+v", v) |
| 144 | } |
| 145 | if v := got["new.example.com"]; v.OK || !v.Unverified { |
| 146 | t.Fatalf("never-woken instance must be unverified (nothing was proven), got %+v", v) |
| 147 | } |
| 148 | |
| 149 | rec := dbverify.Merge(nil, "box-a", now, |
| 150 | []string{"good.example.com", "damaged.example.com", "new.example.com"}, got) |
| 151 | if rec.Failures() != 1 { |
| 152 | t.Fatalf("Failures() = %d, want 1 — only the damaged replica", rec.Failures()) |
| 153 | } |
| 154 | if rec.Unverified() != 1 { |
| 155 | t.Fatalf("Unverified() = %d, want 1", rec.Unverified()) |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | // The per-host prefix must be the one the activator actually writes to, or |
| 160 | // every drill reports "no replica" over perfectly good backups. |
| 161 | func TestVerifyHibernatingUsesTheActivatorsPrefix(t *testing.T) { |
| 162 | var seen string |
| 163 | swapVerifyOne(t, func(_ context.Context, _, prefix, _, _ string) (uint32, error) { |
| 164 | seen = prefix |
| 165 | return 1, nil |
| 166 | }) |
| 167 | cfg := &activator.Config{Bucket: "instances", Prefix: "hib"} |
| 168 | VerifyHibernating(context.Background(), cfg, routes("x.example.com"), 0, time.Now(), t.TempDir()) |
| 169 | |
| 170 | want := activator.HibernatePrefix("hib", "x.example.com") |
| 171 | if seen != want { |
| 172 | t.Fatalf("drilled prefix %q, activator writes %q", seen, want) |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | // A capped run visits the cap and no more — the rest keep their old verdict |
| 177 | // via Merge, which is what stops a cap from reading as full coverage. |
| 178 | func TestVerifyHibernatingHonoursTheCap(t *testing.T) { |
| 179 | calls := 0 |
| 180 | swapVerifyOne(t, func(context.Context, string, string, string, string) (uint32, error) { |
| 181 | calls++ |
| 182 | return 1, nil |
| 183 | }) |
| 184 | cfg := &activator.Config{Bucket: "instances"} |
| 185 | got := VerifyHibernating(context.Background(), cfg, |
| 186 | routes("a.example.com", "b.example.com", "c.example.com"), 2, time.Now(), t.TempDir()) |
| 187 | |
| 188 | if calls != 2 { |
| 189 | t.Fatalf("restored %d replicas under --max 2", calls) |
| 190 | } |
| 191 | if len(got) != 2 { |
| 192 | t.Fatalf("returned %d verdicts under --max 2: %v", len(got), got) |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | // Oldest-first is what makes a capped drill cover the whole fleet over time |
| 197 | // instead of re-drilling the same alphabetical head every night. |
| 198 | func TestOldestFirstPutsNeverVerifiedThenStalest(t *testing.T) { |
| 199 | prior := &dbverify.Record{Hosts: map[string]dbverify.Verdict{ |
| 200 | "recent.example.com": {OK: true, VerifiedAt: at("2026-08-14T00:00:00Z")}, |
| 201 | "stale.example.com": {OK: true, VerifiedAt: at("2026-08-01T00:00:00Z")}, |
| 202 | }} |
| 203 | got := hostsOf(OldestFirst(routes("recent.example.com", "stale.example.com", "never.example.com"), prior)) |
| 204 | want := []string{"never.example.com", "stale.example.com", "recent.example.com"} |
| 205 | if !slices.Equal(got, want) { |
| 206 | t.Fatalf("OldestFirst = %v, want %v", got, want) |
| 207 | } |
| 208 | |
| 209 | // With no prior record at all, the order must still be deterministic. |
| 210 | if got := hostsOf(OldestFirst(routes("b.example.com", "a.example.com"), nil)); !slices.Equal(got, []string{"a.example.com", "b.example.com"}) { |
| 211 | t.Fatalf("OldestFirst with no prior record = %v", got) |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | // A hibernating instance's replica has a live owner — the activator's embedded |
| 216 | // replicator, under the lease — and an owner that has stopped writing leaves a |
| 217 | // replica that still restores perfectly. Same blind spot as the box-wide |
| 218 | // drill, same signal: the replica's change counter stopped moving while the |
| 219 | // database it mirrors did not. |
| 220 | func TestVerifyHibernatingCatchesAStaleReplica(t *testing.T) { |
| 221 | dir := t.TempDir() |
| 222 | live := filepath.Join(dir, "live.db") |
| 223 | writeCounterDB(t, live, 30) |
| 224 | // Quiet for longer than the grace window: unambiguously "replication |
| 225 | // stopped", not "still catching up". |
| 226 | old := time.Now().Add(-2 * time.Hour) |
| 227 | if err := os.Chtimes(live, old, old); err != nil { |
| 228 | t.Fatal(err) |
| 229 | } |
| 230 | |
| 231 | // The replica restores clean, but at a change counter from long ago. |
| 232 | swapVerifyOne(t, func(context.Context, string, string, string, string) (uint32, error) { |
| 233 | return 2, nil |
| 234 | }) |
| 235 | |
| 236 | now := at("2026-08-15T09:00:00Z") |
| 237 | got := VerifyHibernating(context.Background(), &activator.Config{Bucket: "instances"}, |
| 238 | []HibernatingRoute{{Host: "frozen.example.com", DBPath: live}}, 0, now, t.TempDir()) |
| 239 | |
| 240 | v := got["frozen.example.com"] |
| 241 | if v.OK { |
| 242 | t.Fatalf("a replica frozen while its instance kept writing was reported OK: %+v", v) |
| 243 | } |
| 244 | if v.Unverified { |
| 245 | t.Fatalf("a stale replica must be a FAILURE, not merely unverified: %+v", v) |
| 246 | } |
| 247 | if !strings.Contains(v.Detail, "replication has stopped") { |
| 248 | t.Fatalf("verdict does not say what is wrong: %q", v.Detail) |
| 249 | } |
| 250 | |
| 251 | // And it must reach the published record as a failure, so the operator |
| 252 | // sees it without a session on the box. |
| 253 | rec := dbverify.Merge(nil, "box-a", now, []string{"frozen.example.com"}, got) |
| 254 | if rec.Failures() != 1 { |
| 255 | t.Fatalf("Failures() = %d, want 1", rec.Failures()) |
| 256 | } |
| 257 | } |
| 258 | |
amadan