amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "net"
11 "net/http"
12 "net/http/httptest"
13 "os"
14 "path/filepath"
15 "strings"
16 "sync"
17 "sync/atomic"
18 "testing"
19 "time"
20
21 "github.com/aws/smithy-go"
22 lsfile "github.com/benbjohnson/litestream/file"
23 lss3 "github.com/benbjohnson/litestream/s3"
24 "github.com/superfly/ltx"
25
26 "github.com/carlosframework/platform/internal/dbdoor"
27 "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
28 "github.com/carlosframework/platform/internal/provision"
29 "github.com/carlosframework/platform/internal/release"
30 "github.com/carlosframework/platform/internal/svcrole"
31)
32
33// Every one of these is refused BEFORE any S3 call. A door that reached the
34// bucket and mapped AccessDenied would leak both latency and the fact that
35// a bucket was tried. resolveReplica takes no store at all, so it cannot.
36func TestResolveReplicaRefusesBeforeTouchingS3(t *testing.T) {
37 s := &Server{}
38 s.cfg.Region = "eu-west-1"
39 s.cfg.HibernatePrefix = "instances"
40
41 cases := []struct {
42 name string
43 rec provision.Record
44 host string
45 want error
46 }{
47 {
48 name: "another region",
49 rec: provision.Record{Host: "a.bab.oncarlos.com", Region: "us-east-1"},
50 host: "a.bab.oncarlos.com",
51 want: ErrResidency,
52 },
53 {
54 name: "the home region seen from a regional console",
55 rec: provision.Record{Host: "a.bab.oncarlos.com"},
56 host: "a.bab.oncarlos.com",
57 want: ErrResidency,
58 },
59 {
60 name: "fleet-homed",
61 rec: provision.Record{Host: "a.bab.oncarlos.com", Region: "fleet/acme"},
62 host: "a.bab.oncarlos.com",
63 want: ErrResidency,
64 },
65 {
66 name: "pooled",
67 rec: provision.Record{Host: "a.bab.oncarlos.com", Region: "eu-west-1", Pool: "shared"},
68 host: "a.bab.oncarlos.com",
69 want: ErrPooled,
70 },
71 {
72 name: "hostile host",
73 rec: provision.Record{Host: "../../etc", Region: "eu-west-1"},
74 host: "../../etc",
75 want: ErrBadHost,
76 },
77 {
78 name: "record for another host",
79 rec: provision.Record{Host: "b.bab.oncarlos.com", Region: "eu-west-1"},
80 host: "a.bab.oncarlos.com",
81 want: ErrBadHost,
82 },
83 }
84 for _, tc := range cases {
85 t.Run(tc.name, func(t *testing.T) {
86 _, err := s.resolveReplica(tc.rec, tc.host)
87 if !errors.Is(err, tc.want) {
88 t.Fatalf("resolveReplica = %v, want %v", err, tc.want)
89 }
90 })
91 }
92}
93
94// A fleet placement is refused even on a console whose own region string
95// happened to match it: the data is on the customer's box, whatever a
96// misconfigured CARLOS_REGION says.
97func TestResolveReplicaRefusesAFleetEvenWhenTheRegionMatches(t *testing.T) {
98 s := &Server{}
99 s.cfg.Region = "fleet/acme"
100 s.cfg.HibernatePrefix = "instances"
101 if _, err := s.resolveReplica(provision.Record{Host: "a.bab.oncarlos.com", Region: "fleet/acme"}, "a.bab.oncarlos.com"); !errors.Is(err, ErrResidency) {
102 t.Fatalf("resolveReplica = %v, want ErrResidency", err)
103 }
104}
105
106// The happy path is an in-region, non-pooled instance, and the prefix is
107// the activator's own layout, not a string this package invents.
108func TestResolveReplicaBuildsTheActivatorPrefix(t *testing.T) {
109 s := &Server{}
110 s.cfg.Region = "eu-west-1"
111 s.cfg.HibernatePrefix = "instances"
112 got, err := s.resolveReplica(
113 provision.Record{Host: "a.bab.oncarlos.com", Region: "eu-west-1"},
114 "a.bab.oncarlos.com")
115 if err != nil {
116 t.Fatal(err)
117 }
118 if got != "instances/a.bab.oncarlos.com" {
119 t.Fatalf("prefix = %q, want instances/a.bab.oncarlos.com", got)
120 }
121}
122
123// A single-region deployment leaves both empty, and "" == "" must be a
124// match rather than a refusal, otherwise the door never works anywhere
125// that has not set CARLOS_REGION. Home-region records keep region "" too
126// (Config.HomeRegion), so this is also the flagship's home console.
127func TestResolveReplicaTreatsEmptyRegionsAsAMatch(t *testing.T) {
128 s := &Server{}
129 s.cfg.HibernatePrefix = "instances"
130 if _, err := s.resolveReplica(
131 provision.Record{Host: "a.bab.oncarlos.com"}, "a.bab.oncarlos.com"); err != nil {
132 t.Fatalf("single-region deployment refused: %v", err)
133 }
134}
135
136// ---- the door handlers (Task 4) ----------------------------------------------
137
138const dbTestHost = "acme.go.tito.io"
139
140type dbFx struct {
141 *consoleFixture
142 member string
143 root string // the fake data bucket: one file replica per prefix
144 opens *atomic.Int32 // how many times the door reached for a replica client
145}
146
147// dbFixture is an in-region, hibernating exec instance of titogo whose
148// replica is a real litestream replica on disk, served through the door's
149// replica seam exactly as the S3 client would serve it.
150func dbFixture(t *testing.T) *dbFx {
151 t.Helper()
152 fx, member := instanceFixture(t, "code-db")
153 root := t.TempDir()
154 fx.srv.cfg.HibernateBucket = "data-test"
155 fx.srv.cfg.HibernatePrefix = "instances"
156 opens := new(atomic.Int32)
157 fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
158 opens.Add(1)
159 return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
160 }
161 putDBRecord(t, fx, provision.Record{Host: dbTestHost})
162 return &dbFx{consoleFixture: fx, member: member, root: root, opens: opens}
163}
164
165// putDBRecord writes an instance record for titogo with the fields a
166// hibernating exec instance has, overridden by whatever rec sets.
167func putDBRecord(t *testing.T, fx *consoleFixture, rec provision.Record) {
168 t.Helper()
169 rec.App, rec.Account = "titogo", release.OpsAccount
170 if rec.Hibernate == nil {
171 yes := true
172 rec.Hibernate = &yes
173 }
174 body, err := json.Marshal(rec)
175 if err != nil {
176 t.Fatal(err)
177 }
178 if _, err := fx.rel.Put(context.Background(), provision.RecordKey(rec.Account, rec.App, rec.Host), body, ""); err != nil {
179 t.Fatal(err)
180 }
181}
182
183func (fx *dbFx) replicate(t *testing.T, host string, batches int) string {
184 t.Helper()
185 return dbdoortest.Replicate(t, filepath.Join(fx.root, "instances", host), batches)
186}
187
188func (fx *dbFx) url(p string) string { return fx.ts.URL + p }
189
190// A level the activator never writes is EMPTY, not missing. A 404 here
191// stops CalcRestorePlan at the first gap and the restore fails.
192func TestAPIDBListAnswersEmptyForUnwrittenLevels(t *testing.T) {
193 fx := dbFixture(t)
194 for level := 0; level <= dbdoor.MaxLevel; level++ {
195 status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, level)), fx.member)
196 if status != http.StatusOK {
197 t.Fatalf("level %d = %d %s, want 200 with an empty list", level, status, out)
198 }
199 var res dbdoor.ListResponse
200 if err := json.Unmarshal([]byte(out), &res); err != nil || res.Files == nil || len(res.Files) != 0 {
201 t.Fatalf("level %d = %s, want {\"files\":[]}", level, out)
202 }
203 }
204}
205
206// Wire specifics 2 and 4: ascending by min_txid, and size, min_txid and
207// max_txid on every entry.
208func TestAPIDBListAscendsAndCarriesTheMandatoryFields(t *testing.T) {
209 fx := dbFixture(t)
210 fx.replicate(t, dbTestHost, 3)
211 status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
212 if status != http.StatusOK {
213 t.Fatalf("list = %d %s", status, out)
214 }
215 var res dbdoor.ListResponse
216 if err := json.Unmarshal([]byte(out), &res); err != nil {
217 t.Fatal(err)
218 }
219 if len(res.Files) < 2 {
220 t.Fatalf("list = %s, want several L0 files", out)
221 }
222 for i, f := range res.Files {
223 if f.Size <= 0 || f.MinTXID == 0 || f.MaxTXID < f.MinTXID || f.Level != 0 {
224 t.Errorf("entry %d = %+v, want level 0 with size, min_txid and max_txid", i, f)
225 }
226 if i > 0 && f.MinTXID <= res.Files[i-1].MinTXID {
227 t.Errorf("entries %d and %d are not ascending by min_txid: %s", i-1, i, out)
228 }
229 }
230 for _, field := range []string{`"size"`, `"min_txid"`, `"max_txid"`} {
231 if !strings.Contains(out, field) {
232 t.Errorf("list body lacks %s: %s", field, out)
233 }
234 }
235}
236
237// get streams the file by coordinate, byte-for-byte, and honours the two
238// Range forms litestream's client sends.
239func TestAPIDBGetServesTheFileAndHonoursRange(t *testing.T) {
240 fx := dbFixture(t)
241 fx.replicate(t, dbTestHost, 1)
242 c := lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost))
243 files := listFiles(t, c, 0)
244 f := files[0]
245 want, err := os.ReadFile(c.LTXFilePath(0, f.MinTXID, f.MaxTXID))
246 if err != nil {
247 t.Fatal(err)
248 }
249 u := fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID)))
250 status, _, out := rawReq(t, http.MethodGet, u, fx.member, "")
251 if status != http.StatusOK || out != string(want) {
252 t.Fatalf("get = %d, %d bytes; want 200 and the file's %d bytes", status, len(out), len(want))
253 }
254 for _, rc := range []struct{ offset, size int64 }{{10, 0}, {10, 20}} {
255 req, _ := http.NewRequest(http.MethodGet, u, nil)
256 req.Header.Set("Authorization", "Bearer "+fx.member)
257 req.Header.Set("Range", dbdoor.RangeHeader(rc.offset, rc.size))
258 res, err := http.DefaultClient.Do(req)
259 if err != nil {
260 t.Fatal(err)
261 }
262 got, _ := io.ReadAll(res.Body)
263 res.Body.Close()
264 end := int64(len(want))
265 if rc.size > 0 {
266 end = rc.offset + rc.size
267 }
268 if res.StatusCode != http.StatusPartialContent || string(got) != string(want[rc.offset:end]) {
269 t.Errorf("range %+v = %d, %d bytes; want 206 and bytes %d-%d", rc, res.StatusCode, len(got), rc.offset, end)
270 }
271 }
272}
273
274func listFiles(t *testing.T, c *lsfile.ReplicaClient, level int) []*ltx.FileInfo {
275 t.Helper()
276 itr, err := c.LTXFiles(context.Background(), level, 0, false)
277 if err != nil {
278 t.Fatal(err)
279 }
280 defer itr.Close()
281 var out []*ltx.FileInfo
282 for itr.Next() {
283 out = append(out, itr.Item())
284 }
285 if len(out) == 0 {
286 t.Fatal("no files at that level")
287 }
288 return out
289}
290
291// A planned file that has gone is a 404 the client maps to os.ErrNotExist
292// (wire specific 5), which is what makes the whole-Restore retry work.
293func TestAPIDBGetOfAMissingFileIsA404(t *testing.T) {
294 fx := dbFixture(t)
295 fx.replicate(t, dbTestHost, 1)
296 status, _ := getRaw(t, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 900, 901)), fx.member)
297 if status != http.StatusNotFound {
298 t.Fatalf("get of a missing coordinate = %d, want 404", status)
299 }
300}
301
302// Coordinate addressing exists to make lock.json unreachable. This is the
303// test that keeps the door from decaying into an object proxy.
304func TestAPIDBServesNoKeyOutsideItsCoordinates(t *testing.T) {
305 fx := dbFixture(t)
306 dir := filepath.Join(fx.root, "instances", dbTestHost)
307 fx.replicate(t, dbTestHost, 1)
308 if err := os.WriteFile(filepath.Join(dir, "lock.json"), []byte(`{"owner":"box-secret"}`), 0o600); err != nil {
309 t.Fatal(err)
310 }
311 base := "/api/cli/apps/bab/titogo/db/" + dbTestHost
312 for _, bad := range []string{
313 base + "/ltx/0/../../lock.json",
314 base + "/lock.json",
315 base + "/ltx/lock.json",
316 base + "/ltx/0/lock.json",
317 base + "/ltx/10",
318 base + "/ltx/-1",
319 base + "/ltx/0/1",
320 base + "/ltx/0/0-1",
321 base + "/ltx/0/2-1",
322 base + "/ltx/10/1-2",
323 "/api/cli/apps/bab/titogo/db/..%2F..%2Fetc/ltx/0",
324 "/api/cli/apps/bab/titogo/db/" + "b.go.tito.io" + "/ltx/0",
325 } {
326 status, _, out := rawReq(t, http.MethodGet, fx.url(bad), fx.member, "")
327 if status == http.StatusOK || status == http.StatusPartialContent || strings.Contains(out, "box-secret") {
328 t.Errorf("%s = %d %s, want a refusal", bad, status, out)
329 }
330 }
331}
332
333// Not a member, no such app, no such instance: the same bytes http.NotFound
334// writes, matching instanceGate, so the door is no oracle for any of them.
335func TestAPIDBStrangersMeetTheByteIdentical404(t *testing.T) {
336 fx := dbFixture(t)
337 want := canonicalNotFound(t)
338 stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-db-stranger")
339 for name, c := range map[string]struct{ tok, path string }{
340 "not a member": {stranger, dbdoor.ListPath("bab", "titogo", dbTestHost, 0)},
341 "no such app": {fx.member, dbdoor.ListPath("bab", "nosuch", dbTestHost, 0)},
342 "no such instance": {fx.member, dbdoor.ListPath("bab", "titogo", "nobody.go.tito.io", 0)},
343 "no such account": {fx.member, dbdoor.ListPath("zzzz", "titogo", dbTestHost, 0)},
344 "get, not a member": {stranger, dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)},
345 } {
346 status, h, body := rawReq(t, http.MethodGet, fx.url(c.path), c.tok, "")
347 assertNotFound(t, name, want, status, h, body)
348 }
349 if n := fx.opens.Load(); n != 0 {
350 t.Fatalf("the door reached for a replica %d times for callers it refuses", n)
351 }
352}
353
354// Every service principal, every role including admin, and the legacy app
355// token meet the closed-door refusal before any handler runs: the same bytes
356// whether or not the app and the instance exist, so it says nothing about
357// THIS app (Ruling, Task 4: the repo's zero-Set closed door, not a 404).
358func TestAPIDBRefusesEveryServicePrincipalTheSameWay(t *testing.T) {
359 fx := dbFixture(t)
360 fx.replicate(t, dbTestHost, 1)
361 appTok, err := fx.st.NewAppToken(OpsAccountID, "titogo", "alice@keymail.dev", "signup", time.Now())
362 if err != nil {
363 t.Fatal(err)
364 }
365 toks := map[string]string{"legacy app token": appTok}
366 for _, role := range []svcrole.Role{svcrole.Publish, svcrole.Instance, svcrole.Operate, svcrole.Admin} {
367 toks[string(role)] = serviceToken(t, fx.st, OpsAccountID, "svc-"+string(role), "titogo", role)
368 }
369 var first string
370 for name, tok := range toks {
371 for _, p := range []string{
372 dbdoor.ListPath("bab", "titogo", dbTestHost, 0),
373 dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1),
374 dbdoor.ListPath("bab", "nosuch", "nobody.go.tito.io", 0),
375 } {
376 status, _, body := rawReq(t, http.MethodGet, fx.url(p), tok, "")
377 if status != http.StatusForbidden || !strings.Contains(body, "closed to service credentials") {
378 t.Errorf("%s at %s = %d %s, want the closed-door refusal", name, p, status, body)
379 }
380 if first == "" {
381 first = body
382 } else if body != first {
383 t.Errorf("%s at %s: refusal varies (%q vs %q)", name, p, body, first)
384 }
385 }
386 }
387 if n := fx.opens.Load(); n != 0 {
388 t.Fatalf("the door reached for a replica %d times for service principals", n)
389 }
390}
391
392// The residency and pooled refusals are answered with no replica client
393// ever built: a 409 naming the cause, before the bucket.
394func TestAPIDBResidencyRefusalsNeverReachTheBucket(t *testing.T) {
395 fx := dbFixture(t)
396 fx.srv.cfg.Region = "ie"
397 putDBRecord(t, fx.consoleFixture, provision.Record{Host: "uk.go.tito.io", Region: "uk"})
398 putDBRecord(t, fx.consoleFixture, provision.Record{Host: "fleet.go.tito.io", Region: "fleet/acme"})
399 putDBRecord(t, fx.consoleFixture, provision.Record{Host: "pool.go.tito.io", Region: "ie", Pool: "tito"})
400 for host, want := range map[string]string{
401 "uk.go.tito.io": "uk",
402 "fleet.go.tito.io": "fleet",
403 "pool.go.tito.io": "pool",
404 dbTestHost: "",
405 } {
406 status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", host, 0)), fx.member)
407 if status != http.StatusConflict || !strings.Contains(out, want) {
408 t.Errorf("%s = %d %s, want a 409 naming %q", host, status, out, want)
409 }
410 }
411 if n := fx.opens.Load(); n != 0 {
412 t.Fatalf("the door reached for a replica %d times before refusing on residency", n)
413 }
414}
415
416// Only an activator-managed exec instance has a replica the platform
417// writes. Everything else is refused by class, before the bucket.
418func TestAPIDBRefusesRoutesTheActivatorDoesNotReplicate(t *testing.T) {
419 fx := dbFixture(t)
420 no := false
421 putDBRecord(t, fx.consoleFixture, provision.Record{Host: "svc.go.tito.io", Kind: "service"})
422 putDBRecord(t, fx.consoleFixture, provision.Record{Host: "addr.go.tito.io", Addr: "10.0.0.1:8080"})
423 putDBRecord(t, fx.consoleFixture, provision.Record{Host: "awake.go.tito.io", Hibernate: &no})
424 for _, host := range []string{"svc.go.tito.io", "addr.go.tito.io", "awake.go.tito.io"} {
425 status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", host, 0)), fx.member)
426 if status != http.StatusConflict {
427 t.Errorf("%s = %d %s, want a 409 refusal", host, status, out)
428 }
429 }
430 if n := fx.opens.Load(); n != 0 {
431 t.Fatalf("the door reached for a replica %d times for routes with none", n)
432 }
433}
434
435// No data bucket configured is a 503, never a door that fails open.
436func TestAPIDBWithoutADataBucketIs503(t *testing.T) {
437 fx := dbFixture(t)
438 fx.srv.cfg.HibernateBucket = ""
439 status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
440 if status != http.StatusServiceUnavailable {
441 t.Fatalf("no bucket = %d %s, want 503", status, out)
442 }
443 if fx.opens.Load() != 0 {
444 t.Fatal("the door reached for a replica with no bucket configured")
445 }
446}
447
448// S3 refusing the read is the residency refusal as a backstop (spec §6),
449// never a generic 502 — and the store's own error text, which carries the
450// bucket name, never reaches the caller.
451func TestAPIDBAccessDeniedIsTheResidencyRefusal(t *testing.T) {
452 fx := dbFixture(t)
453 fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) {
454 return deniedReplica{}, nil
455 }
456 for _, p := range []string{
457 dbdoor.ListPath("bab", "titogo", dbTestHost, 0),
458 dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1),
459 } {
460 status, out := getRaw(t, fx.url(p), fx.member)
461 if status != http.StatusConflict || strings.Contains(out, "data-test") || strings.Contains(out, "secret-bucket") {
462 t.Errorf("%s = %d %s, want the residency 409 with no bucket name", p, status, out)
463 }
464 }
465}
466
467type deniedReplica struct{}
468
469func (deniedReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
470 return nil, &smithy.GenericAPIError{Code: "AccessDenied", Message: "no access to secret-bucket"}
471}
472
473func (deniedReplica) OpenLTXFile(context.Context, int, ltx.TXID, ltx.TXID, int64, int64) (io.ReadCloser, error) {
474 return nil, &smithy.GenericAPIError{Code: "AccessDenied", Message: "no access to secret-bucket"}
475}
476
477// Every successful open of a file writes an audit row naming the instance
478// and the TXID span served (spec §6 "Who may open it").
479func TestAPIDBGetIsAudited(t *testing.T) {
480 fx := dbFixture(t)
481 fx.replicate(t, dbTestHost, 1)
482 f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
483 if status, _ := getRaw(t, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID))), fx.member); status != http.StatusOK {
484 t.Fatalf("get = %d", status)
485 }
486 evs, err := fx.st.Events(OpsAccountID, "titogo", 10)
487 if err != nil {
488 t.Fatal(err)
489 }
490 for _, e := range evs {
491 if e.Action == "db-read" && e.Target == dbTestHost && e.Address == "alice@keymail.dev" &&
492 strings.Contains(e.Detail, fmt.Sprint(uint64(f.MaxTXID))) {
493 return
494 }
495 }
496 t.Fatalf("no db-read audit row naming %s and txid %d: %+v", dbTestHost, f.MaxTXID, evs)
497}
498
499// One replica client per prefix: a restore's dozens of requests share one
500// transport rather than paying a handshake each. And the production reader
501// carries the configured bucket and the resolved prefix, nothing a request
502// supplied.
503func TestCachedReadReplicaIsOnePerPrefix(t *testing.T) {
504 t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
505 s := &Server{}
506 s.cfg.HibernateBucket = "data-ie"
507 s.cfg.HibernateRegion = "eu-west-1"
508 ctx := context.Background()
509 a1, err := s.cachedReadReplica(ctx, "instances/a.example.com")
510 if err != nil {
511 t.Fatal(err)
512 }
513 a2, _ := s.cachedReadReplica(ctx, "instances/a.example.com")
514 b, _ := s.cachedReadReplica(ctx, "instances/b.example.com")
515 if a1 != a2 {
516 t.Error("the same prefix built two clients")
517 }
518 if a1 == b {
519 t.Error("two prefixes shared one client")
520 }
521 c, ok := a1.(*lss3.ReplicaClient)
522 if !ok || c.Bucket != "data-ie" || c.Path != "instances/a.example.com" || c.Region != "eu-west-1" {
523 t.Fatalf("client = %#v, want the configured bucket, region and the resolved prefix", a1)
524 }
525}
526
527// The door has its own token-keyed budget, separate from the instance
528// verbs': spending it does not touch the instance budget, and it applies
529// only after membership, so a stranger cannot learn anything from a 429.
530func TestAPIDBHasItsOwnBudget(t *testing.T) {
531 fx := dbFixture(t)
532 fx.srv.cli.db = newRateLimiter(2, cliRateWindow)
533 u := fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0))
534 for i := 0; i < 2; i++ {
535 if status, out := getRaw(t, u, fx.member); status != http.StatusOK {
536 t.Fatalf("request %d = %d %s", i, status, out)
537 }
538 }
539 if status, _ := getRaw(t, u, fx.member); status != http.StatusTooManyRequests {
540 t.Fatalf("third request = %d, want 429", status)
541 }
542 if !fx.srv.cli.instances.allow(hashToken(fx.member), time.Now()) {
543 t.Fatal("the db door spent the instance verbs' budget")
544 }
545 stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-db-budget")
546 for i := 0; i < 3; i++ {
547 if status, _ := getRaw(t, u, stranger); status != http.StatusNotFound {
548 t.Fatalf("stranger request %d = %d, want the 404 before any budget", i, status)
549 }
550 }
551}
552
553// End to end: the CLI's read-only client restores a real replica through
554// the real door, with a member's token, and gets every row.
555func TestAPIDBRestoresThroughTheDoor(t *testing.T) {
556 fx := dbFixture(t)
557 fx.replicate(t, dbTestHost, 3)
558 // The tightest bound there is, and a restore still completes under it.
559 // (Small files finish server-side before the client reads them, so the
560 // one-stream-at-a-time property itself is pinned in internal/dbdoor.)
561 fx.srv.dbStreams = newDBStreamLimits(1, 1)
562 do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
563 req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
564 if err != nil {
565 return nil, err
566 }
567 req.Header = h.Clone()
568 req.Header.Set("Authorization", "Bearer "+fx.member)
569 return http.DefaultClient.Do(req)
570 }
571 got, err := dbdoor.RestoreTo(context.Background(), dbdoor.NewClient(do), t.TempDir())
572 if err != nil {
573 t.Fatal(err)
574 }
575 defer os.RemoveAll(got.Dir)
576 db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
577 if err != nil {
578 t.Fatal(err)
579 }
580 defer db.Close()
581 var n int
582 if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
583 t.Fatal(err)
584 }
585 if n != 3*dbdoortest.Rows || got.TXID == 0 || got.At.IsZero() {
586 t.Fatalf("restored %d rows, txid %d at %v; want %d rows and an as-of", n, got.TXID, got.At, 3*dbdoortest.Rows)
587 }
588}
589
590// A ranged GET is still a read of the file. Listing the coordinates and
591// fetching each one with Range: bytes=1- returns all but one byte of every
592// file, so every successful GET writes a row, and a resumed one says where
593// it started.
594func TestAPIDBRangedGetIsAudited(t *testing.T) {
595 fx := dbFixture(t)
596 fx.replicate(t, dbTestHost, 1)
597 f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
598 req, _ := http.NewRequest(http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID))), nil)
599 req.Header.Set("Authorization", "Bearer "+fx.member)
600 req.Header.Set("Range", "bytes=1-")
601 res, err := http.DefaultClient.Do(req)
602 if err != nil {
603 t.Fatal(err)
604 }
605 io.Copy(io.Discard, res.Body)
606 res.Body.Close()
607 if res.StatusCode != http.StatusPartialContent {
608 t.Fatalf("ranged get = %d", res.StatusCode)
609 }
610 evs, err := fx.st.Events(OpsAccountID, "titogo", 10)
611 if err != nil {
612 t.Fatal(err)
613 }
614 for _, e := range evs {
615 if e.Action == "db-read" && e.Target == dbTestHost && strings.Contains(e.Detail, "byte 1") {
616 return
617 }
618 }
619 t.Fatalf("a ranged read wrote no db-read row naming its start: %+v", evs)
620}
621
622// stallReplica serves listings from a file replica but opens files whose
623// reads never finish: each blocks until the request's context ends, the way
624// a stalled S3 read does, or produces zeros forever when endless is set.
625type stallReplica struct {
626 replicaReader
627 endless bool
628 opened chan struct{}
629}
630
631func (s *stallReplica) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
632 select {
633 case s.opened <- struct{}{}:
634 default:
635 }
636 if s.endless {
637 return io.NopCloser(zeros{}), nil
638 }
639 return io.NopCloser(ctxReader{ctx}), nil
640}
641
642type ctxReader struct{ ctx context.Context }
643
644func (r ctxReader) Read([]byte) (int, error) { <-r.ctx.Done(); return 0, r.ctx.Err() }
645
646type zeros struct{}
647
648func (zeros) Read(p []byte) (int, error) { clear(p); return len(p), nil }
649
650func (fx *dbFx) stall(endless bool) *stallReplica {
651 sr := &stallReplica{endless: endless, opened: make(chan struct{}, 16)}
652 fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
653 sr.replicaReader = lsfile.NewReplicaClient(filepath.Join(fx.root, prefix))
654 return sr, nil
655 }
656 return sr
657}
658
659func (fx *dbFx) memberToken(t *testing.T, addr, code string) string {
660 t.Helper()
661 if err := fx.st.AddMember(OpsAccountID, addr, RoleMember, time.Now()); err != nil {
662 t.Fatal(err)
663 }
664 return cliToken(t, fx.consoleFixture, addr, code)
665}
666
667// startGet begins a GET and returns once the door has opened the file, so
668// the stream holds its slot; cancel ends it.
669func (fx *dbFx) startGet(t *testing.T, sr *stallReplica, tok string) (stop func()) {
670 t.Helper()
671 ctx, cancel := context.WithCancel(context.Background())
672 req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
673 req.Header.Set("Authorization", "Bearer "+tok)
674 done := make(chan struct{})
675 go func() {
676 defer close(done)
677 if res, err := http.DefaultClient.Do(req); err == nil {
678 io.Copy(io.Discard, res.Body)
679 res.Body.Close()
680 }
681 }()
682 select {
683 case <-sr.opened:
684 case <-time.After(5 * time.Second):
685 t.Fatal("the stream never opened")
686 }
687 return func() { cancel(); <-done }
688}
689
690func (fx *dbFx) activeStreams() int {
691 fx.srv.dbStreams.mu.Lock()
692 defer fx.srv.dbStreams.mu.Unlock()
693 return fx.srv.dbStreams.total
694}
695
696func (fx *dbFx) waitIdle(t *testing.T, within time.Duration) {
697 t.Helper()
698 deadline := time.Now().Add(within)
699 for fx.activeStreams() != 0 {
700 if time.Now().After(deadline) {
701 t.Fatalf("%d streams still hold slots after %v", fx.activeStreams(), within)
702 }
703 time.Sleep(10 * time.Millisecond)
704 }
705}
706
707// Slow downloads must not accumulate: active file streams are bounded per
708// principal and across the console, and a refusal says when to come back.
709func TestAPIDBBoundsConcurrentStreams(t *testing.T) {
710 fx := dbFixture(t)
711 fx.replicate(t, dbTestHost, 1)
712 fx.srv.dbStreams = newDBStreamLimits(2, 1)
713 sr := fx.stall(false)
714 bob := fx.memberToken(t, "bob@keymail.dev", "code-db-bob")
715 carol := fx.memberToken(t, "carol@keymail.dev", "code-db-carol")
716 u := fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1))
717
718 stopAlice := fx.startGet(t, sr, fx.member)
719 status, h, body := rawReq(t, http.MethodGet, u, fx.member, "")
720 if status != http.StatusTooManyRequests || h.Get("Retry-After") == "" {
721 t.Fatalf("second stream for one person = %d %q %s, want 429 with Retry-After", status, h.Get("Retry-After"), body)
722 }
723 stopBob := fx.startGet(t, sr, bob)
724 status, h, body = rawReq(t, http.MethodGet, u, carol, "")
725 if status != http.StatusTooManyRequests || h.Get("Retry-After") == "" {
726 t.Fatalf("stream past the console's bound = %d %q %s, want 429 with Retry-After", status, h.Get("Retry-After"), body)
727 }
728 stopAlice()
729 stopBob()
730 fx.waitIdle(t, 5*time.Second)
731 stopAgain := fx.startGet(t, sr, fx.member)
732 stopAgain()
733}
734
735// A stalled bucket read ends after the idle deadline and frees its slot.
736func TestAPIDBStalledReadIsCut(t *testing.T) {
737 fx := dbFixture(t)
738 fx.replicate(t, dbTestHost, 1)
739 fx.srv.dbStreamIdle = 200 * time.Millisecond
740 sr := fx.stall(false)
741 start := time.Now()
742 status, _, _ := rawReq(t, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), fx.member, "")
743 <-sr.opened
744 if time.Since(start) > 5*time.Second {
745 t.Fatalf("a stalled read held the request for %v", time.Since(start))
746 }
747 _ = status
748 fx.waitIdle(t, 5*time.Second)
749}
750
751// A client that stops reading is cut by the renewable write deadline, so a
752// trickling or abandoned download cannot hold a slot forever.
753func TestAPIDBStalledClientIsCut(t *testing.T) {
754 fx := dbFixture(t)
755 fx.replicate(t, dbTestHost, 1)
756 fx.srv.dbStreamIdle = 300 * time.Millisecond
757 sr := fx.stall(true)
758 req, _ := http.NewRequest(http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
759 req.Header.Set("Authorization", "Bearer "+fx.member)
760 res, err := http.DefaultClient.Do(req)
761 if err != nil {
762 t.Fatal(err)
763 }
764 defer res.Body.Close() // never read: the client has stalled
765 <-sr.opened
766 fx.waitIdle(t, 15*time.Second)
767}
768
769// recConn records, in order, every write and every write-deadline change on
770// a server-side connection.
771type recConn struct {
772 net.Conn
773 mu *sync.Mutex
774 events *[]string
775}
776
777func (c recConn) Write(p []byte) (int, error) {
778 c.mu.Lock()
779 *c.events = append(*c.events, "write")
780 c.mu.Unlock()
781 return c.Conn.Write(p)
782}
783
784func (c recConn) SetWriteDeadline(t time.Time) error {
785 c.mu.Lock()
786 if t.IsZero() {
787 *c.events = append(*c.events, "clear")
788 } else {
789 *c.events = append(*c.events, "deadline")
790 }
791 c.mu.Unlock()
792 return c.Conn.SetWriteDeadline(t)
793}
794
795type recListener struct {
796 net.Listener
797 mu sync.Mutex
798 events []string
799}
800
801func (l *recListener) Accept() (net.Conn, error) {
802 c, err := l.Listener.Accept()
803 if err != nil {
804 return nil, err
805 }
806 return recConn{Conn: c, mu: &l.mu, events: &l.events}, nil
807}
808
809// net/http flushes what is still buffered (the last bytes, the chunked
810// terminator) in finishRequest, AFTER the handler returns, and clears the
811// write deadline itself only after that. A handler that clears it on the
812// way out leaves that final flush with no deadline, so a client that stops
813// reading near the end holds the connection forever, outside the stream
814// bound. No byte may ever be written with the deadline cleared.
815func TestAPIDBNeverWritesWithoutADeadline(t *testing.T) {
816 fx := dbFixture(t)
817 fx.replicate(t, dbTestHost, 1)
818 f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
819 ts := httptest.NewUnstartedServer(fx.srv)
820 rl := &recListener{Listener: ts.Listener}
821 ts.Listener = rl
822 ts.Start()
823 defer ts.Close()
824 req, _ := http.NewRequest(http.MethodGet, ts.URL+dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID)), nil)
825 req.Header.Set("Authorization", "Bearer "+fx.member)
826 req.Close = true
827 res, err := http.DefaultClient.Do(req)
828 if err != nil {
829 t.Fatal(err)
830 }
831 io.Copy(io.Discard, res.Body)
832 res.Body.Close()
833 if res.StatusCode != http.StatusOK {
834 t.Fatalf("get = %d", res.StatusCode)
835 }
836 deadline := time.Now().Add(5 * time.Second)
837 for {
838 rl.mu.Lock()
839 events := append([]string(nil), rl.events...)
840 rl.mu.Unlock()
841 if len(events) > 0 && events[len(events)-1] == "clear" || time.Now().After(deadline) {
842 armed := false
843 for i, e := range events {
844 switch e {
845 case "deadline":
846 armed = true
847 case "clear":
848 armed = false
849 case "write":
850 if !armed {
851 t.Fatalf("write %d happened with no write deadline: %v", i, events)
852 }
853 }
854 }
855 if !armed && events[len(events)-1] != "clear" {
856 t.Fatalf("events = %v", events)
857 }
858 return
859 }
860 time.Sleep(10 * time.Millisecond)
861 }
862}
863