| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "io/fs" |
| 11 | "net/http" |
| 12 | "os" |
| 13 | "path/filepath" |
| 14 | "strconv" |
| 15 | "strings" |
| 16 | "sync/atomic" |
| 17 | "testing" |
| 18 | "time" |
| 19 | |
| 20 | lsfile "github.com/benbjohnson/litestream/file" |
| 21 | |
| 22 | "github.com/carlosframework/platform/internal/dbdoor" |
| 23 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 24 | "github.com/carlosframework/platform/internal/genselect" |
| 25 | "github.com/carlosframework/platform/internal/hostclaim" |
| 26 | "github.com/carlosframework/platform/internal/lease/leasetest" |
| 27 | "github.com/carlosframework/platform/internal/provision" |
| 28 | "github.com/carlosframework/platform/internal/svcrole" |
| 29 | ) |
| 30 | |
| 31 | func enableDoorPlan(t *testing.T, fx *dbFx) *leasetest.VersionedBucket { |
| 32 | t.Helper() |
| 33 | rec, err := provision.GetRecord(t.Context(), fx.rel, "bab", "titogo", dbTestHost) |
| 34 | if err != nil { |
| 35 | t.Fatal(err) |
| 36 | } |
| 37 | inc, err := hostclaim.Incarnation("bab", "titogo", dbTestHost, rec.CreatedAt) |
| 38 | if err != nil { |
| 39 | t.Fatal(err) |
| 40 | } |
| 41 | c := hostclaim.Claim{V: 2, Host: dbTestHost, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "titogo", dbTestHost), Account: "bab", App: "titogo", CreatedAt: rec.CreatedAt, Incarnation: inc, Location: "instances/" + dbTestHost}, Op: hostclaim.Op{Kind: hostclaim.KindServing}} |
| 42 | raw, _ := json.Marshal(c) |
| 43 | key, _ := hostclaim.Key(dbTestHost) |
| 44 | if _, err := fx.rel.Put(t.Context(), key, raw, ""); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | m := leasetest.NewVersionedBucket() |
| 48 | m.AddBucket("data-test") |
| 49 | fx.srv.cfg.HomeRegion = "ie" |
| 50 | fx.srv.cfg.HostClaim = true |
| 51 | fx.srv.cfg.DBVersions = m |
| 52 | secret, err := fx.st.SigningKey() |
| 53 | if err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | fx.srv.dbHandleKey = dbHandleSecret(secret) |
| 57 | return m |
| 58 | } |
| 59 | |
| 60 | func TestDoorPlanHTTPCompatibilityAndRestore(t *testing.T) { |
| 61 | for _, mode := range []string{"legacy", "origin", "published", "unfinished"} { |
| 62 | t.Run(mode, func(t *testing.T) { |
| 63 | enabled := mode != "legacy" |
| 64 | fx := dbFixture(t) |
| 65 | fx.replicate(t, dbTestHost, 3) |
| 66 | if enabled { |
| 67 | model := enableDoorPlan(t, fx) |
| 68 | location := "instances/" + dbTestHost |
| 69 | if mode == "published" || mode == "unfinished" { |
| 70 | obs, err := hostclaim.Read(t.Context(), fx.rel, dbTestHost) |
| 71 | if err != nil { |
| 72 | t.Fatal(err) |
| 73 | } |
| 74 | c := cloneDoorClaim(obs.Claim) |
| 75 | c.Holder.Generation = "live" |
| 76 | marker := genselect.Marker{Op: "live", Host: dbTestHost, Incarnation: c.Holder.Incarnation, Box: "box", InstalledAt: time.Unix(1, 0).UTC()} |
| 77 | raw, err := json.Marshal(marker) |
| 78 | if err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | model.Put("data-test", location+"/generation.json", raw) |
| 82 | if mode == "unfinished" { |
| 83 | c.Holder.Generation = "next" |
| 84 | c.Holder.Admissions = []hostclaim.AdmissionRecord{{Generation: "next", Predecessor: "live", Region: "ie", PredecessorRegion: "ie", Mode: "cold"}} |
| 85 | } |
| 86 | writeDoorClaim(t, fx.srv, dbRequest{host: dbTestHost}, c) |
| 87 | location += "/g-live" |
| 88 | } |
| 89 | root := filepath.Join(fx.root, "instances", dbTestHost, "ltx") |
| 90 | err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { |
| 91 | if err != nil { |
| 92 | return err |
| 93 | } |
| 94 | if d.IsDir() { |
| 95 | return nil |
| 96 | } |
| 97 | rel, err := filepath.Rel(root, p) |
| 98 | if err != nil { |
| 99 | return err |
| 100 | } |
| 101 | parts := strings.Split(filepath.ToSlash(rel), "/") |
| 102 | if len(parts) != 2 { |
| 103 | return fmt.Errorf("unexpected file: %s", rel) |
| 104 | } |
| 105 | level, err := strconv.Atoi(parts[0]) |
| 106 | if err != nil { |
| 107 | return err |
| 108 | } |
| 109 | b, err := os.ReadFile(p) |
| 110 | if err != nil { |
| 111 | return err |
| 112 | } |
| 113 | model.Put("data-test", dbLevelPrefix(location, level)+parts[1], b) |
| 114 | return nil |
| 115 | }) |
| 116 | if err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | } |
| 120 | do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 121 | r, err := http.NewRequestWithContext(ctx, "GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil) |
| 122 | if err != nil { |
| 123 | return nil, err |
| 124 | } |
| 125 | r.Header = h.Clone() |
| 126 | r.Header.Set("Authorization", "Bearer "+fx.member) |
| 127 | return http.DefaultClient.Do(r) |
| 128 | } |
| 129 | got, err := dbdoor.RestoreTo(t.Context(), dbdoor.NewClient(do), t.TempDir()) |
| 130 | if err != nil { |
| 131 | t.Fatal(err) |
| 132 | } |
| 133 | defer os.RemoveAll(got.Dir) |
| 134 | db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro") |
| 135 | if err != nil { |
| 136 | t.Fatal(err) |
| 137 | } |
| 138 | defer db.Close() |
| 139 | var rows int |
| 140 | if err := db.QueryRow("SELECT count(*) FROM t").Scan(&rows); err != nil { |
| 141 | t.Fatal(err) |
| 142 | } |
| 143 | if rows != 3*dbdoortest.Rows { |
| 144 | t.Fatal(rows) |
| 145 | } |
| 146 | // The old CLI ignores additive fields and sends no handle. |
| 147 | res, err := do(t.Context(), dbdoor.ListTail(0), http.Header{}) |
| 148 | if err != nil { |
| 149 | t.Fatal(err) |
| 150 | } |
| 151 | var list dbdoor.ListResponse |
| 152 | if err := json.NewDecoder(res.Body).Decode(&list); err != nil { |
| 153 | t.Fatal(err) |
| 154 | } |
| 155 | res.Body.Close() |
| 156 | if len(list.Files) == 0 { |
| 157 | t.Fatal("empty L0") |
| 158 | } |
| 159 | f := list.Files[0] |
| 160 | res, err = do(t.Context(), dbdoor.GetTail(f.Level, f.MinTXID, f.MaxTXID), http.Header{}) |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | io.Copy(io.Discard, res.Body) |
| 165 | res.Body.Close() |
| 166 | want := http.StatusOK |
| 167 | if enabled { |
| 168 | want = http.StatusBadRequest |
| 169 | } |
| 170 | if res.StatusCode != want { |
| 171 | t.Fatalf("old CLI: %d want %d", res.StatusCode, want) |
| 172 | } |
| 173 | if !enabled && list.Protocol != 0 { |
| 174 | t.Fatal("flag-off wire changed") |
| 175 | } |
| 176 | if enabled && fx.opens.Load() != 0 { |
| 177 | t.Fatal("pinned door opened unversioned replica") |
| 178 | } |
| 179 | }) |
| 180 | } |
| 181 | } |
| 182 | |
| 183 | func TestDoorPlanHTTPAuthorizationAndMissingClaim(t *testing.T) { |
| 184 | fx := dbFixture(t) |
| 185 | model := enableDoorPlan(t, fx) |
| 186 | model.FailOn = func(op, key string) error { t.Fatalf("unauthorized data access %s %s", op, key); return nil } |
| 187 | stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "door-plan-stranger") |
| 188 | status, h, body := rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), stranger, "") |
| 189 | assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body) |
| 190 | admin := serviceToken(t, fx.st, OpsAccountID, "door-plan-admin", "titogo", svcrole.Admin) |
| 191 | status, _, _ = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), admin, "") |
| 192 | if status != http.StatusForbidden { |
| 193 | t.Fatal(status) |
| 194 | } |
| 195 | // A second existing record with no claim must not auto-adopt here. |
| 196 | other := "other.go.tito.io" |
| 197 | putDBRecord(t, fx.consoleFixture, provision.Record{Host: other}) |
| 198 | status, h, body = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", other, 0)), fx.member, "") |
| 199 | if status != 409 || h.Get("Retry-After") != "1" || !strings.Contains(body, "host not adopted") { |
| 200 | t.Fatal(status, h, body) |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) { |
| 205 | fx := dbFixture(t) |
| 206 | model := enableDoorPlan(t, fx) |
| 207 | key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1) |
| 208 | payload := append(doorLTXHeader(t, 1, 1), []byte("original")...) |
| 209 | model.Put("data-test", key, payload) |
| 210 | var opens atomic.Int32 |
| 211 | fx.srv.cfg.DBVersions = doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) { |
| 212 | rc, err := model.OpenVersion(ctx, b, k, id) |
| 213 | if err != nil { |
| 214 | return nil, err |
| 215 | } |
| 216 | if opens.Add(1) == 2 { |
| 217 | return &dbRangeReader{Reader: io.LimitReader(rc, 3), Closer: rc}, nil |
| 218 | } |
| 219 | return rc, nil |
| 220 | }} |
| 221 | do := func(tail string, h http.Header) *http.Response { |
| 222 | t.Helper() |
| 223 | r, err := http.NewRequest("GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil) |
| 224 | if err != nil { |
| 225 | t.Fatal(err) |
| 226 | } |
| 227 | r.Header = h.Clone() |
| 228 | r.Header.Set("Authorization", "Bearer "+fx.member) |
| 229 | res, err := http.DefaultClient.Do(r) |
| 230 | if err != nil { |
| 231 | t.Fatal(err) |
| 232 | } |
| 233 | return res |
| 234 | } |
| 235 | res := do(dbdoor.ListTail(0), http.Header{}) |
| 236 | var list dbdoor.ListResponse |
| 237 | if err := json.NewDecoder(res.Body).Decode(&list); err != nil { |
| 238 | t.Fatal(err) |
| 239 | } |
| 240 | res.Body.Close() |
| 241 | if len(list.Files) != 1 { |
| 242 | t.Fatal(list) |
| 243 | } |
| 244 | f := list.Files[0] |
| 245 | h := http.Header{} |
| 246 | h.Set(dbdoor.HandleHeader, f.Handle) |
| 247 | res = do(dbdoor.GetTail(0, 1, 1), h) |
| 248 | first, err := io.ReadAll(res.Body) |
| 249 | res.Body.Close() |
| 250 | if !errors.Is(err, io.ErrUnexpectedEOF) || string(first) != string(payload[:3]) || res.Header.Get(dbdoor.HandleHeader) != f.Handle { |
| 251 | t.Fatal(string(first), err, res.Header) |
| 252 | } |
| 253 | model.Put("data-test", key, []byte("replacement")) |
| 254 | h.Set("Range", "bytes=3-") |
| 255 | res = do(dbdoor.GetTail(0, 1, 1), h) |
| 256 | rest, err := io.ReadAll(res.Body) |
| 257 | res.Body.Close() |
| 258 | if err != nil || res.StatusCode != 206 || string(append(first, rest...)) != string(payload) { |
| 259 | t.Fatal(string(rest), err, res.StatusCode) |
| 260 | } |
| 261 | events, err := fx.st.Events(OpsAccountID, "titogo", 10) |
| 262 | if err != nil { |
| 263 | t.Fatal(err) |
| 264 | } |
| 265 | ranged := false |
| 266 | for _, e := range events { |
| 267 | if e.Action == dbAuditAction && e.Target == dbTestHost && strings.Contains(e.Detail, "byte 3") { |
| 268 | ranged = true |
| 269 | } |
| 270 | } |
| 271 | if !ranged { |
| 272 | t.Fatal("missing pinned range audit", events) |
| 273 | } |
| 274 | // A flag-off console ignores the handle header exactly as today (R1): |
| 275 | // it serves the coordinate and never echoes the handle, so a v2 client |
| 276 | // rejects the response (Task 1's acknowledgement check). |
| 277 | // Keep startup configuration immutable while either server is running. |
| 278 | legacy := dbFixture(t) |
| 279 | legacy.replicate(t, dbTestHost, 1) |
| 280 | lc := lsfile.NewReplicaClient(filepath.Join(legacy.root, "instances", dbTestHost)) |
| 281 | lf := listFiles(t, lc, 0)[0] |
| 282 | req, err := http.NewRequest("GET", legacy.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(lf.MinTXID), uint64(lf.MaxTXID))), nil) |
| 283 | if err != nil { |
| 284 | t.Fatal(err) |
| 285 | } |
| 286 | req.Header = h.Clone() |
| 287 | req.Header.Del("Range") // a whole-file GET, so success is 200, not 206 |
| 288 | req.Header.Set("Authorization", "Bearer "+legacy.member) |
| 289 | res, err = http.DefaultClient.Do(req) |
| 290 | if err != nil { |
| 291 | t.Fatal(err) |
| 292 | } |
| 293 | io.Copy(io.Discard, res.Body) |
| 294 | res.Body.Close() |
| 295 | if res.StatusCode != 200 { |
| 296 | t.Fatal("flag-off changed its answer to a request carrying a handle header", res.StatusCode) |
| 297 | } |
| 298 | if res.Header.Get(dbdoor.HandleHeader) != "" { |
| 299 | t.Fatal("flag-off console echoed a handle it cannot enforce") |
| 300 | } |
| 301 | } |
| 302 | |
amadan