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 "crypto/ecdsa"
6 "crypto/elliptic"
7 "crypto/rand"
8 "encoding/json"
9 "errors"
10 "io"
11 "net/http"
12 "net/http/httptest"
13 "os"
14 "strings"
15 "testing"
16 "time"
17
18 "github.com/carlosframework/platform/internal/dbdoor"
19 "github.com/carlosframework/platform/internal/hostclaim"
20 "github.com/carlosframework/platform/internal/lease"
21 "github.com/carlosframework/platform/internal/lease/leasetest"
22 "github.com/carlosframework/platform/internal/provision"
23 "github.com/carlosframework/platform/internal/release"
24)
25
26type doorVersionHook struct {
27 lease.VersionedStore
28 after func()
29}
30
31func (v doorVersionHook) ListVersions(ctx context.Context, b, p string) ([]lease.ObjectVersion, error) {
32 out, err := v.VersionedStore.ListVersions(ctx, b, p)
33 if v.after != nil {
34 v.after()
35 }
36 return out, err
37}
38
39type doorReadFault struct {
40 release.Store
41 err error
42}
43
44func (s *doorReadFault) Get(ctx context.Context, key string) ([]byte, string, error) {
45 if s.err != nil {
46 return nil, "", s.err
47 }
48 return s.Store.Get(ctx, key)
49}
50
51func doorPlanFixture(t *testing.T, legacy bool) (*Server, dbRequest, provision.Record, *leasetest.VersionedBucket) {
52 t.Helper()
53 rec := provision.Record{Host: "shop.example", Account: "bab", App: "jam"}
54 inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
55 if err != nil {
56 t.Fatal(err)
57 }
58 loc := "instances/" + rec.Host
59 if !legacy {
60 loc += "/" + inc
61 }
62 claim := hostclaim.Claim{V: 2, Host: rec.Host, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey(rec.Account, rec.App, rec.Host), Account: rec.Account, App: rec.App, Incarnation: inc, Location: loc}, Op: hostclaim.Op{Kind: hostclaim.KindServing}}
63 store := release.NewDirStore(t.TempDir())
64 data, _ := json.Marshal(claim)
65 key, _ := hostclaim.Key(rec.Host)
66 if _, err := store.Put(t.Context(), key, data, ""); err != nil {
67 t.Fatal(err)
68 }
69 model := leasetest.NewVersionedBucket()
70 model.AddBucket("data")
71 s := &Server{cfg: Config{HomeRegion: "ie", HostClaim: true, Releases: store, HibernateBucket: "data", DBVersions: model}, dbHandleKey: []byte(strings.Repeat("s", 32))}
72 g := dbRequest{acct: &Account{Sqid: "bab"}, app: "jam", host: rec.Host}
73 g.claim, err = s.dbClaim(t.Context(), g, rec)
74 if err != nil {
75 t.Fatal(err)
76 }
77 return s, g, rec, model
78}
79
80func writeDoorClaim(t *testing.T, s *Server, g dbRequest, c *hostclaim.Claim) {
81 t.Helper()
82 key, _ := hostclaim.Key(g.host)
83 _, etag, err := s.cfg.Releases.Get(t.Context(), key)
84 if err != nil {
85 t.Fatal(err)
86 }
87 b, err := json.Marshal(c)
88 if err != nil {
89 t.Fatal(err)
90 }
91 if _, err = s.cfg.Releases.Put(t.Context(), key, b, etag); err != nil {
92 t.Fatal(err)
93 }
94}
95func cloneDoorClaim(c *hostclaim.Claim) *hostclaim.Claim {
96 b, _ := json.Marshal(c)
97 var out hostclaim.Claim
98 _ = json.Unmarshal(b, &out)
99 return &out
100}
101
102func TestDoorPlanHandles(t *testing.T) {
103 key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
104 if err != nil {
105 t.Fatal(err)
106 }
107 secret := dbHandleSecret(key)
108 h := dbFileHandle{V: 1, Host: "shop.example", Location: "instances/shop.example/inc", Key: "instances/shop.example/inc/0000/a", VersionID: "v+/?", Incarnation: "inc", Expiry: 200, Account: "bab", App: "jam", Size: 6}
109 token, err := signDBHandle(secret, h)
110 if err != nil {
111 t.Fatal(err)
112 }
113 got, err := verifyDBHandle(dbHandleSecret(key), token, time.Unix(199, 0))
114 if err != nil || got != h {
115 t.Fatal(got, err)
116 }
117 for _, bad := range []string{"", token + "x", "x" + token, strings.Repeat("x", 8193)} {
118 if _, err := verifyDBHandle(secret, bad, time.Unix(199, 0)); !errors.Is(err, errDBHandle) {
119 t.Fatal(bad, err)
120 }
121 }
122 if _, err := verifyDBHandle([]byte(strings.Repeat("x", 32)), token, time.Unix(199, 0)); !errors.Is(err, errDBHandle) {
123 t.Fatal(err)
124 }
125 if _, err := verifyDBHandle(secret, token, time.Unix(200, 0)); !errors.Is(err, errDBHandleExpired) {
126 t.Fatal(err)
127 }
128}
129
130func TestDoorPlanListingAndPinnedOpen(t *testing.T) {
131 for _, legacy := range []bool{false, true} {
132 t.Run(map[bool]string{true: "legacy", false: "scoped"}[legacy], func(t *testing.T) {
133 s, g, _, m := doorPlanFixture(t, legacy)
134 loc := g.claim.Claim.Holder.Location
135 key := dbVersionKey(loc, 0, 1, 1)
136 m.Put("data", key, []byte("older"))
137 id := m.Put("data", key, append(doorLTXHeader(t, 1, 1), []byte("original")...))
138 deleted := dbVersionKey(loc, 0, 2, 2)
139 m.Put("data", deleted, []byte("hidden"))
140 m.BareDelete("data", deleted)
141 m.Put("data", loc+"/lock.json", []byte("secret"))
142 m.Put("data", dbVersionKey(loc+"/other-incarnation", 0, 3, 3), []byte("other"))
143 m.Put("data", dbVersionKey(loc+"-sibling", 0, 4, 4), []byte("sibling"))
144 m.Put("data", dbVersionKey(loc+"/g-old", 0, 5, 5), []byte("generation"))
145 m.PageSize = 1
146 out, err := s.listDBVersions(t.Context(), g, 0)
147 if err != nil || out.Protocol != 2 || len(out.Files) != 1 {
148 t.Fatalf("%+v %v", out, err)
149 }
150 token := out.Files[0].Handle
151 h, err := verifyDBHandle(s.dbHandleKey, token, time.Now())
152 if err != nil || h.VersionID != id || h.Location != loc {
153 t.Fatal(h, err)
154 }
155 req := httptest.NewRequest(http.MethodGet, "/", nil)
156 req.Header.Set(dbdoor.HandleHeader, token)
157 rd, err := s.pinnedDBReader(req, g, 0, 1, 1)
158 if err != nil {
159 t.Fatal(err)
160 }
161 // Stall after admission, then overwrite the identical legacy key and
162 // advance the holder to a new incarnation before the physical open.
163 m.Put("data", key, []byte("replacement"))
164 next := cloneDoorClaim(g.claim.Claim)
165 next.Gen++
166 next.Holder.CreatedAt = time.Unix(1, 0)
167 next.Holder.Incarnation, _ = hostclaim.Incarnation("bab", "jam", g.host, next.Holder.CreatedAt)
168 next.Holder.Location = "instances/" + g.host + "/" + next.Holder.Incarnation
169 writeDoorClaim(t, s, g, next)
170 rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 102, 3)
171 if err != nil {
172 t.Fatal(err)
173 }
174 body, err := io.ReadAll(rc)
175 rc.Close()
176 if err != nil || string(body) != "igi" {
177 t.Fatalf("%q %v", body, err)
178 }
179 g2 := g
180 g2.claim.Claim = next
181 if _, err := s.pinnedDBReader(req, g2, 0, 1, 1); err == nil {
182 t.Fatal("new incarnation accepted old handle")
183 }
184 bad := g
185 bad.host = "other.example"
186 if _, err := s.pinnedDBReader(req, bad, 0, 1, 1); err == nil {
187 t.Fatal("cross host")
188 }
189 if _, err := s.pinnedDBReader(req, g, 0, 2, 2); err == nil {
190 t.Fatal("cross coordinate")
191 }
192 if err := m.DeleteVersion(t.Context(), "data", key, id); err != nil {
193 t.Fatal(err)
194 }
195 if _, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 0, 0); !errors.Is(err, os.ErrNotExist) {
196 t.Fatal(err)
197 }
198 })
199 }
200}
201
202func TestDoorPlanBracket(t *testing.T) {
203 for _, change := range []string{"holder", "op", "same-gen-op", "chore", "absent", "malformed", "read-error"} {
204 t.Run(change, func(t *testing.T) {
205 s, g, _, m := doorPlanFixture(t, false)
206 m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1), doorLTXHeader(t, 1, 1))
207 fault := &doorReadFault{Store: s.cfg.Releases}
208 s.cfg.Releases = fault
209 s.cfg.DBVersions = doorVersionHook{VersionedStore: m, after: func() {
210 next := cloneDoorClaim(g.claim.Claim)
211 next.Gen++
212 switch change {
213 case "holder":
214 next.Holder.Stamp = next.Gen
215 case "op":
216 next.Op = hostclaim.Op{Kind: hostclaim.KindDeleting, ID: "delete"}
217 case "same-gen-op":
218 next.Gen--
219 next.Op.Exec = "another-box"
220 case "chore":
221 next.Touched.Complete = true
222 case "absent":
223 fault.err = os.ErrNotExist
224 return
225 case "read-error":
226 fault.err = errors.New("offline")
227 return
228 case "malformed":
229 next.V = 999
230 }
231 writeDoorClaim(t, s, g, next)
232 }}
233 out, err := s.listDBVersions(t.Context(), g, 0)
234 if change == "chore" {
235 if err != nil || len(out.Files) != 1 {
236 t.Fatal(out, err)
237 }
238 } else if err == nil || len(out.Files) != 0 {
239 t.Fatalf("partial/success: %+v %v", out, err)
240 }
241 })
242 }
243}
244
245func TestDoorPlanClaimRefusals(t *testing.T) {
246 for _, change := range []string{"absent", "free", "deleting", "parking", "purging", "owner", "incarnation", "location", "malformed", "read-error"} {
247 t.Run(change, func(t *testing.T) {
248 s, g, rec, m := doorPlanFixture(t, true)
249 m.FailOn = func(op, key string) error { t.Fatalf("refusal reached data bucket: %s %s", op, key); return nil }
250 c := cloneDoorClaim(g.claim.Claim)
251 want := errDBClaimClosed
252 switch change {
253 case "absent":
254 s.cfg.Releases = release.NewDirStore(t.TempDir())
255 want = errDBNotAdopted
256 case "free":
257 c.Holder = nil
258 c.Op = hostclaim.Op{}
259 case "deleting", "parking", "purging":
260 c.Op = hostclaim.Op{Kind: change, ID: "x"}
261 case "owner":
262 g.app = "other"
263 want = os.ErrNotExist
264 case "incarnation":
265 rec.CreatedAt = time.Unix(1, 0)
266 want = os.ErrNotExist
267 case "location":
268 c.Holder.Location = ""
269 want = hostclaim.ErrMalformed
270 case "malformed":
271 c.V = 99
272 want = hostclaim.ErrMalformed
273 case "read-error":
274 want = errors.New("offline")
275 s.cfg.Releases = &doorReadFault{Store: s.cfg.Releases, err: want}
276 }
277 if change != "absent" && change != "read-error" {
278 writeDoorClaim(t, s, dbRequest{host: rec.Host}, c)
279 }
280 if _, err := s.dbClaim(t.Context(), g, rec); !errors.Is(err, want) {
281 t.Fatalf("%v want %v", err, want)
282 }
283 })
284 }
285}
286
287func TestDoorPlanListingFailureAndRanges(t *testing.T) {
288 s, g, _, m := doorPlanFixture(t, true)
289 for _, lo := range []uint64{1, 2, 3} {
290 m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, lo, lo), []byte("sixsix"))
291 }
292 m.PageSize = 1
293 m.FailListAfterPages = 1
294 if out, err := s.listDBVersions(t.Context(), g, 0); err == nil || len(out.Files) != 0 {
295 t.Fatal(out, err)
296 }
297 for _, tc := range []struct {
298 offset, size, want int64
299 ok bool
300 }{{0, 0, 6, true}, {2, 0, 4, true}, {2, 2, 2, true}, {2, 100, 4, true}, {6, 0, 0, false}, {-1, 0, 0, false}, {0, -1, 0, false}} {
301 n, err := dbVersionRange(6, tc.offset, tc.size)
302 if (err == nil) != tc.ok || n != tc.want {
303 t.Fatal(tc, n, err)
304 }
305 }
306 w := httptest.NewRecorder()
307 r := httptest.NewRequest("GET", "/", nil)
308 r.Header.Set("Range", "bytes=2-3")
309 r.Header.Set(dbdoor.HandleHeader, "signed")
310 if err := dbVersionHeaders(w, r, 6, 2, 2); err != nil {
311 t.Fatal(err)
312 }
313 if w.Header().Get("Content-Length") != "2" || w.Header().Get("Content-Range") != "bytes 2-3/6" || w.Header().Get(dbdoor.HandleHeader) != "signed" {
314 t.Fatal(w.Header())
315 }
316}
317
318type doorOpenHook struct {
319 lease.VersionedStore
320 open func(context.Context, string, string, string) (io.ReadCloser, error)
321}
322
323func (v doorOpenHook) OpenVersion(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
324 return v.open(ctx, b, k, id)
325}
326
327type doorTrackedReader struct {
328 io.Reader
329 closed bool
330}
331
332func (r *doorTrackedReader) Close() error { r.closed = true; return nil }
333
334type doorCancelledReader struct{ ctx context.Context }
335
336func (r doorCancelledReader) Read([]byte) (int, error) { return 0, r.ctx.Err() }
337
338func TestDoorPlanRangeReaderCloses(t *testing.T) {
339 for _, mode := range []string{"complete", "short-skip", "cancelled-skip"} {
340 t.Run(mode, func(t *testing.T) {
341 s, g, _, m := doorPlanFixture(t, true)
342 ctx, cancel := context.WithCancel(t.Context())
343 defer cancel()
344 tracked := &doorTrackedReader{Reader: strings.NewReader("abcdef")}
345 if mode == "short-skip" {
346 tracked.Reader = strings.NewReader("a")
347 }
348 if mode == "cancelled-skip" {
349 cancel()
350 tracked.Reader = doorCancelledReader{ctx: ctx}
351 }
352 key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
353 s.cfg.DBVersions = doorOpenHook{VersionedStore: m, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
354 if b != "data" || k != key || id != "selected" {
355 t.Fatal(b, k, id)
356 }
357 return tracked, nil
358 }}
359 rd := &dbVersionReader{store: s.cfg.DBVersions, bucket: "data", handle: dbFileHandle{Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 6}}
360 rc, err := rd.OpenLTXFile(ctx, 0, 1, 1, 2, 2)
361 if mode != "complete" {
362 if err == nil || !tracked.closed {
363 t.Fatal(err, tracked.closed)
364 }
365 if mode == "cancelled-skip" && !errors.Is(err, context.Canceled) {
366 t.Fatal(err)
367 }
368 return
369 }
370 if err != nil {
371 t.Fatal(err)
372 }
373 b, err := io.ReadAll(rc)
374 if err != nil || string(b) != "cd" {
375 t.Fatal(string(b), err)
376 }
377 if tracked.closed {
378 t.Fatal("closed before caller finished")
379 }
380 rc.Close()
381 if !tracked.closed {
382 t.Fatal("reader leaked")
383 }
384 })
385 }
386}
387
388// TestDBDoorRefusesDuringMoves (moves spec §19.1, Review Focus 4, K14): the
389// db door refuses on every move kind, in every phase, including
390// landing/committed and the restoring and reverting opened phases; it
391// reopens at the close, when the claim is serving again. The refusal is a
392// 409 with Retry-After: 1, and the data bucket is never read.
393func TestDBDoorRefusesDuringMoves(t *testing.T) {
394 const move = "mv_abcdefghijklmnopqrstuvwxyz"
395 reloc := func(phase, exec string) hostclaim.Op {
396 return hostclaim.Op{Kind: hostclaim.KindRelocating, ID: move, Phase: phase, Exec: exec,
397 Intent: map[string]string{"region": "uk", "source_box": "ie-box", "target_box": "uk-box"}}
398 }
399 cases := []hostclaim.Op{
400 {Kind: hostclaim.KindMoving, ID: move, Phase: hostclaim.PhaseMoving, Exec: "ie-box"},
401 {Kind: hostclaim.KindMoving, ID: move, Phase: hostclaim.PhaseSealed},
402 {Kind: hostclaim.KindMoving, ID: move, Phase: hostclaim.PhaseAborting, Exec: "ie-box"},
403 {Kind: hostclaim.KindLanding, ID: move, Phase: "checking", Exec: "uk-box"},
404 {Kind: hostclaim.KindLanding, ID: move, Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box"},
405 {Kind: hostclaim.KindLanding, ID: move, Phase: hostclaim.PhaseLandingProbed, Exec: "uk-box"},
406 {Kind: hostclaim.KindLanding, ID: move, Phase: hostclaim.PhaseLandingCommitted, Exec: "uk-box"},
407 {Kind: hostclaim.KindReverting, ID: move, Exec: "ie-box", Journal: map[string]string{hostclaim.JournalRevPhase: hostclaim.RevPhaseEvidence}},
408 {Kind: hostclaim.KindReverting, ID: move, Exec: "ie-box", Journal: map[string]string{hostclaim.JournalRevPhase: hostclaim.RevPhaseVerify}},
409 {Kind: hostclaim.KindReverting, ID: move, Phase: hostclaim.PhaseProbing, Exec: "ie-box"},
410 {Kind: hostclaim.KindReverting, ID: move, Phase: hostclaim.PhaseOpened, Exec: "ie-box"},
411 {Kind: hostclaim.KindRestoring, ID: move, Phase: hostclaim.PhaseReleasing, Exec: "uk-box"},
412 {Kind: hostclaim.KindRestoring, ID: move, Phase: hostclaim.PhaseReleased},
413 {Kind: hostclaim.KindRestoring, ID: move, Phase: hostclaim.PhaseClaimed, Exec: "ie-box"},
414 {Kind: hostclaim.KindRestoring, ID: move, Phase: hostclaim.PhaseProbing, Exec: "ie-box"},
415 {Kind: hostclaim.KindRestoring, ID: move, Phase: hostclaim.PhaseProbed, Exec: "ie-box"},
416 {Kind: hostclaim.KindRestoring, ID: move, Phase: hostclaim.PhaseOpened, Exec: "ie-box"},
417 reloc("starting", "ie-box"),
418 reloc("releasing", "none"),
419 reloc("pinning", "none"),
420 reloc("installing", "uk-box"),
421 reloc("installed", "uk-box"),
422 reloc("probing", "uk-box"),
423 reloc("probe-failed", "uk-box"),
424 reloc("serving", "uk-box"),
425 }
426 kinds := map[string]bool{}
427 for _, op := range cases {
428 kinds[op.Kind] = true
429 name := op.Kind + "/" + op.Phase
430 if rp := op.Journal[hostclaim.JournalRevPhase]; rp != "" {
431 name += "rev_phase=" + rp
432 }
433 t.Run(name, func(t *testing.T) {
434 s, g, rec, m := doorPlanFixture(t, false)
435 m.FailOn = func(op, key string) error { t.Fatalf("a refusal reached the data bucket: %s %s", op, key); return nil }
436 c := cloneDoorClaim(g.claim.Claim)
437 c.Op = op
438 writeDoorClaim(t, s, dbRequest{host: rec.Host}, c)
439 _, err := s.dbClaim(t.Context(), g, rec)
440 if !errors.Is(err, errDBClaimClosed) {
441 t.Fatalf("%v, want the door closed", err)
442 }
443 w := httptest.NewRecorder()
444 s.dbClaimFailure(w, httptest.NewRequest(http.MethodGet, "/", nil), g, err)
445 if w.Code != http.StatusConflict || w.Header().Get("Retry-After") != "1" {
446 t.Fatalf("%d Retry-After %q, want 409 and 1", w.Code, w.Header().Get("Retry-After"))
447 }
448 })
449 }
450 for _, k := range []string{hostclaim.KindMoving, hostclaim.KindLanding, hostclaim.KindReverting, hostclaim.KindRestoring, hostclaim.KindRelocating} {
451 if !kinds[k] {
452 t.Errorf("the table has no %s row", k)
453 }
454 }
455 // The close: serving again, the door reopens.
456 t.Run("serving", func(t *testing.T) {
457 s, g, rec, _ := doorPlanFixture(t, false)
458 c := cloneDoorClaim(g.claim.Claim)
459 c.Op = hostclaim.Op{Kind: hostclaim.KindServing}
460 writeDoorClaim(t, s, dbRequest{host: rec.Host}, c)
461 if _, err := s.dbClaim(t.Context(), g, rec); err != nil {
462 t.Fatalf("serving: %v", err)
463 }
464 })
465}
466