1// SPDX-License-Identifier: MIT
2
3package cachesvc
4
5import (
6 "bufio"
7 "bytes"
8 "fmt"
9 "github.com/bnema/zerowrap"
10 "io"
11 "net"
12 "net/http"
13 "net/http/httptest"
14 "os"
15 "path/filepath"
16 "strings"
17 "sync"
18 "testing"
19 "time"
20)
21
22// chunkReq sends a PATCH whose body the test feeds through the returned writer;
23// the status arrives on the channel. The body is streamed, so the server has
24// the Content-Range and a body that stays open until the test closes it.
25func (j *job) chunkReq(id int64, lo, hi int) (w *io.PipeWriter, done <-chan int) {
26 pr, pw := io.Pipe()
27 res := make(chan int, 1)
28 req, _ := http.NewRequest("PATCH", j.r.srv.URL+APIPath+fmt.Sprintf("caches/%d", id), pr)
29 req.Header.Set("Authorization", "Bearer "+j.token)
30 req.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/*", lo, hi-1))
31 go func() {
32 resp, err := http.DefaultClient.Do(req)
33 if err != nil {
34 res <- -1
35 return
36 }
37 _ = resp.Body.Close()
38 res <- resp.StatusCode
39 }()
40 return pw, res
41}
42
43// wait receives from c; the timeout only bounds a failing test.
44func wait[T any](t *testing.T, c <-chan T, what string) T {
45 t.Helper()
46 select {
47 case v := <-c:
48 return v
49 case <-time.After(30 * time.Second):
50 t.Fatalf("timeout waiting for %s", what)
51 panic("unreachable")
52 }
53}
54
55func dataFiles(t *testing.T, r *rig) int {
56 t.Helper()
57 des, err := os.ReadDir(filepath.Join(r.dir, "data"))
58 if err != nil {
59 t.Fatal(err)
60 }
61 return len(des)
62}
63
64// A6: a retry of a range that fails halfway must not leave the earlier bytes
65// counted as valid.
66func TestFailedRetryRemovesCoverage(t *testing.T) {
67 r := newRig(t, 1<<20)
68 j := r.job("j1", repoA, main_, main_)
69 _, id := j.reserve("k", v1, 4)
70 if j.chunk(id, []byte("abcd"), 0) != 204 {
71 t.Fatal("first chunk")
72 }
73 // The same range again, with a body one byte short of its Content-Range.
74 code, _, _ := j.do("PATCH", fmt.Sprintf("caches/%d", id), []byte("x"), map[string]string{"Content-Range": "bytes 0-3/*"})
75 if code != 400 {
76 t.Fatalf("short retry: %d", code)
77 }
78 if c := j.commit(id, 4); c != 400 {
79 t.Fatalf("commit after a failed retry must fail, got %d", c)
80 }
81 if j.chunk(id, []byte("wxyz"), 0) != 204 || j.commit(id, 4) != 204 {
82 t.Fatal("a good retry works")
83 }
84 if _, data, _ := j.restore(v1, "k"); string(data) != "wxyz" {
85 t.Fatalf("content %q", data)
86 }
87}
88
89// B4: with a body held open by the test (no sleeps) commit is refused, an
90// overlapping range is refused, a disjoint one is accepted, an expiry pass does
91// not drop the upload, and the upload then completes and restores.
92func TestConcurrentChunks(t *testing.T) {
93 admitted := make(chan int64, 4)
94 r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, ReservationMaxLife: 24 * time.Hour, hookChunk: func(id int64) { admitted <- id }})
95 j := r.job("j1", repoA, main_, main_)
96 _, id := j.reserve("k", v1, 8)
97
98 pw, done := j.chunkReq(id, 0, 4)
99 wait(t, admitted, "the first chunk to be admitted")
100
101 if c := j.commit(id, 8); c != 400 {
102 t.Errorf("commit with a chunk in flight: %d", c)
103 }
104 if c := j.chunk(id, []byte("cdef"), 2); c != http.StatusConflict {
105 t.Errorf("overlapping range: %d", c)
106 }
107 if c := j.chunk(id, []byte("efgh"), 4); c != 204 {
108 t.Errorf("disjoint range: %d", c)
109 }
110 wait(t, admitted, "the second chunk to be admitted")
111
112 // Far past the idle ttl (but inside the absolute life): the next reserve runs
113 // the expiry, which must keep an upload with a chunk in flight.
114 r.clk.add(2 * time.Hour)
115 if code, _ := j.reserve("other", v1, 4); code != http.StatusCreated {
116 t.Fatalf("reserve: %d", code)
117 }
118 if n := r.svc.store.pendingCount(); n != 2 {
119 t.Fatalf("pending reservations %d, want the upload and the new one", n)
120 }
121
122 if _, err := pw.Write([]byte("abcd")); err != nil {
123 t.Fatal(err)
124 }
125 _ = pw.Close()
126 if code := wait(t, done, "the first chunk"); code != 204 {
127 t.Fatalf("first chunk: %d", code)
128 }
129 if c := j.commit(id, 8); c != 204 {
130 t.Fatalf("commit: %d", c)
131 }
132 if _, data, _ := j.restore(v1, "k"); string(data) != "abcdefgh" {
133 t.Fatalf("content %q", data)
134 }
135}
136
137// A body that stalls loses its range: the progress deadline cuts it, the range
138// is free again and is not counted as received.
139func TestStalledUploadIsCut(t *testing.T) {
140 r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, ProgressTimeout: 100 * time.Millisecond})
141 j := r.job("j1", repoA, main_, main_)
142 _, id := j.reserve("k", v1, 8)
143
144 conn, err := net.Dial("tcp", strings.TrimPrefix(r.srv.URL, "http://"))
145 if err != nil {
146 t.Fatal(err)
147 }
148 defer func() { _ = conn.Close() }()
149 // Eight bytes announced, four sent, then silence.
150 _, err = fmt.Fprintf(conn, "PATCH %scaches/%d HTTP/1.1\r\nHost: x\r\nAuthorization: Bearer %s\r\nContent-Range: bytes 0-7/*\r\nContent-Length: 8\r\n\r\nabcd", APIPath, id, j.token)
151 if err != nil {
152 t.Fatal(err)
153 }
154 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
155 if err != nil {
156 t.Fatalf("no answer: %v", err)
157 }
158 _ = resp.Body.Close()
159 if resp.StatusCode != 400 {
160 t.Fatalf("stalled upload: %d", resp.StatusCode)
161 }
162 if c := j.commit(id, 8); c != 400 {
163 t.Fatalf("commit: %d", c)
164 }
165 // Not in flight any more: the same range can be sent again.
166 if j.chunk(id, []byte("abcdefgh"), 0) != 204 || j.commit(id, 8) != 204 {
167 t.Fatal("the range is free again")
168 }
169}
170
171// The write deadline of a download moves with every write.
172type deadlineWriter struct {
173 httptest.ResponseRecorder
174 deadlines []time.Time
175}
176
177func (d *deadlineWriter) SetWriteDeadline(t time.Time) error {
178 d.deadlines = append(d.deadlines, t)
179 return nil
180}
181
182func TestProgressWriterExtendsDeadline(t *testing.T) {
183 d := &deadlineWriter{ResponseRecorder: *httptest.NewRecorder()}
184 pw := &progressWriter{ResponseWriter: d, rc: http.NewResponseController(d), idle: time.Minute}
185 for i := 0; i < 3; i++ {
186 if _, err := pw.Write([]byte("x")); err != nil {
187 t.Fatal(err)
188 }
189 }
190 if len(d.deadlines) != 3 || d.deadlines[0].IsZero() {
191 t.Fatalf("deadlines %v", d.deadlines)
192 }
193}
194
195// GO-010: slots are taken by accepted connections only. Idle listeners hold
196// none, so many sandboxes cannot starve one another; a connection over a cap is
197// closed at once, and a slot comes back when a connection closes.
198func TestLimitListener(t *testing.T) {
199 const listeners, global = 70, 64
200 slots := make(chan struct{}, global)
201 var ls []*limitListener
202 for i := 0; i < listeners; i++ {
203 ln := listenLoopbackT(t)
204 l := newLimitListener(ln, 2, slots)
205 t.Cleanup(func() { _ = l.Close() })
206 ls = append(ls, l)
207 }
208 if len(slots) != 0 {
209 t.Fatalf("idle listeners hold %d global slots", len(slots))
210 }
211 dial := func(l *limitListener) net.Conn {
212 c, err := net.Dial("tcp", l.Addr().String())
213 if err != nil {
214 t.Fatal(err)
215 }
216 t.Cleanup(func() { _ = c.Close() })
217 return c
218 }
219 // Every accepted connection is closed at cleanup, whatever happens (Close is
220 // idempotent and releases its slots once).
221 accepted := func(c net.Conn) net.Conn {
222 t.Cleanup(func() { _ = c.Close() })
223 return c
224 }
225 // Each of the first 64 listeners still accepts its connection.
226 var held []net.Conn
227 for _, l := range ls[:global] {
228 dial(l)
229 c, err := l.Accept()
230 if err != nil {
231 t.Fatal(err)
232 }
233 held = append(held, accepted(c))
234 }
235 if len(slots) != global {
236 t.Fatalf("global slots %d", len(slots))
237 }
238 // The 65th is over the global cap: closed at once. Its Accept goes on to the
239 // next connection, so a later one proves the loop is alive.
240 over := dial(ls[global])
241 next := make(chan net.Conn, 1)
242 go func() {
243 if c, err := ls[global].Accept(); err == nil {
244 next <- accepted(c)
245 }
246 }()
247 _ = over.SetReadDeadline(time.Now().Add(30 * time.Second))
248 if _, err := over.Read(make([]byte, 1)); err == nil || strings.Contains(err.Error(), "timeout") {
249 t.Fatalf("over-cap connection not closed by the server: %v", err)
250 }
251 // Free one slot: the next connection is served.
252 _ = held[0].Close()
253 _ = held[0].Close() // idempotent: no slot is released twice
254 if len(slots) != global-1 {
255 t.Fatalf("after one close: %d slots", len(slots))
256 }
257 dial(ls[global])
258 c := wait(t, next, "a connection once a slot is free")
259 held = append(held[1:], c)
260
261 // The local cap: two on one listener, the third is closed.
262 for _, c := range held[:10] {
263 _ = c.Close()
264 }
265 l := ls[0]
266 var mine []net.Conn
267 for i := 0; i < 2; i++ {
268 dial(l)
269 c, err := l.Accept()
270 if err != nil {
271 t.Fatal(err)
272 }
273 mine = append(mine, accepted(c))
274 }
275 third := dial(l)
276 go func() {
277 if c, err := l.Accept(); err == nil {
278 _ = accepted(c).Close()
279 }
280 }()
281 _ = third.SetReadDeadline(time.Now().Add(30 * time.Second))
282 if _, err := third.Read(make([]byte, 1)); err == nil || strings.Contains(err.Error(), "timeout") {
283 t.Fatalf("connection over the listener's share not closed: %v", err)
284 }
285 for _, c := range append(held[10:], mine...) {
286 _ = c.Close()
287 }
288 if len(slots) != 0 || len(l.local) != 0 {
289 t.Fatalf("slots left: global %d local %d", len(slots), len(l.local))
290 }
291}
292
293func listenLoopbackT(t *testing.T) net.Listener {
294 t.Helper()
295 ln, err := net.Listen("tcp", "127.0.0.1:0")
296 if err != nil {
297 t.Fatal(err)
298 }
299 return ln
300}
301
302// A2: a reservation without cacheSize is charged for the bytes it writes, before
303// they are accepted, and keeps the charge until commit.
304func TestSizelessUploadIsCharged(t *testing.T) {
305 r := newRig(t, 100)
306 a, b := r.job("ja", repoA, main_, main_), r.job("jb", repoA, feat, feat)
307 _, ida := a.reserve("a", v1, 0)
308 _, idb := b.reserve("b", v1, 0)
309 if c := a.chunk(ida, make([]byte, 60), 0); c != 204 {
310 t.Fatalf("first: %d", c)
311 }
312 if _, _, held := r.svc.store.stats(); held != 61 { // 60 + the 1 of b
313 t.Fatalf("held %d", held)
314 }
315 if c := b.chunk(idb, make([]byte, 50), 0); c != http.StatusServiceUnavailable {
316 t.Fatalf("over the budget: %d", c)
317 }
318 // Refused before the bytes were taken: the file is empty, nothing is covered.
319 if st, _ := os.Stat(r.svc.store.tmp(idb)); st.Size() != 0 {
320 t.Fatalf("refused chunk wrote %d bytes", st.Size())
321 }
322 if c := b.chunk(idb, make([]byte, 39), 0); c != 204 {
323 t.Fatalf("what fits: %d", c)
324 }
325 // Retrying a smaller or equal range later costs nothing more.
326 if c := b.chunk(idb, make([]byte, 39), 0); c != 204 {
327 t.Fatalf("retry: %d", c)
328 }
329 if _, _, held := r.svc.store.stats(); held != 99 {
330 t.Fatalf("held %d", held)
331 }
332 if c := a.commit(ida, 60); c != 204 {
333 t.Fatalf("commit: %d", c)
334 }
335 if n, used, held := r.svc.store.stats(); n != 1 || used != 60 || held != 39 {
336 t.Fatalf("after commit: entries %d used %d held %d", n, used, held)
337 }
338 // Above the whole limit.
339 _, idc := a.reserve("c", v1, 0)
340 if c := a.chunk(idc, make([]byte, 101), 0); c != http.StatusRequestEntityTooLarge {
341 t.Fatalf("over the limit: %d", c)
342 }
343}
344
345// A size-less upload makes room by evicting the least recently used entries.
346func TestSizelessUploadEvicts(t *testing.T) {
347 r := newRig(t, 100)
348 j := r.job("j", repoA, main_, main_)
349 for _, k := range []string{"old", "new"} {
350 if c := j.save(k, v1, bytes.Repeat([]byte(k[:1]), 40)); c != 204 {
351 t.Fatalf("%s: %d", k, c)
352 }
353 r.clk.add(time.Minute)
354 }
355 _, id := j.reserve("big", v1, 0)
356 if c := j.chunk(id, make([]byte, 50), 0); c != 204 {
357 t.Fatalf("chunk: %d", c)
358 }
359 if code, _ := j.lookup(v1, "old"); code != 204 {
360 t.Fatalf("old must be evicted: %d", code)
361 }
362 if code, _ := j.lookup(v1, "new"); code != 200 {
363 t.Fatalf("new must stay: %d", code)
364 }
365}
366
367// A3/B1: uploads in progress are bounded in all and per job, before any file is
368// created; idle ones are reclaimed with the clock.
369func TestPendingReservationLimits(t *testing.T) {
370 r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, MaxPending: 3, MaxPendingPerJob: 2})
371 j1, j2, j3 := r.job("j1", repoA, main_, main_), r.job("j2", repoA, main_, main_), r.job("j3", repoA, main_, main_)
372 for _, k := range []string{"a", "b"} {
373 if code, _ := j1.reserve(k, v1, 1); code != http.StatusCreated {
374 t.Fatalf("%s: %d", k, code)
375 }
376 }
377 if code, _ := j1.reserve("c", v1, 1); code != http.StatusServiceUnavailable {
378 t.Fatalf("per job: %d", code)
379 }
380 if code, _ := j2.reserve("d", v1, 1); code != http.StatusCreated {
381 t.Fatalf("other job: %d", code)
382 }
383 if code, _ := j3.reserve("e", v1, 1); code != http.StatusServiceUnavailable {
384 t.Fatalf("global: %d", code)
385 }
386 if n := dataFiles(t, r); n != 3 {
387 t.Fatalf("%d files for 3 reservations: refused ones must create none", n)
388 }
389 // Idle reservations are reclaimed once the ttl passes.
390 r.clk.add(2 * time.Hour)
391 if code, _ := j3.reserve("e", v1, 1); code != http.StatusCreated {
392 t.Fatalf("after expiry: %d", code)
393 }
394 if n, f := r.svc.store.pendingCount(), dataFiles(t, r); n != 1 || f != 1 {
395 t.Fatalf("pending %d files %d", n, f)
396 }
397}
398
399// A3: every entry and reservation costs at least the minimum charge, so tiny
400// entries cannot pile up without bound.
401func TestMinimumCharge(t *testing.T) {
402 r := newRigWith(t, Config{Max: 10, MinCharge: 4})
403 j := r.job("j", repoA, main_, main_)
404 for _, k := range []string{"a", "b", "c"} {
405 if c := j.save(k, v1, []byte("x")); c != 204 {
406 t.Fatalf("%s: %d", k, c)
407 }
408 r.clk.add(time.Minute)
409 }
410 if n, used, _ := r.svc.store.stats(); n != 2 || used != 8 {
411 t.Fatalf("entries %d used %d: 10 bytes hold two entries of 4", n, used)
412 }
413 // Reservations are charged too, and nothing is evictable under them.
414 r2 := newRigWith(t, Config{Max: 8, MinCharge: 4})
415 j2 := r2.job("j", repoA, main_, main_)
416 for _, k := range []string{"a", "b"} {
417 if code, _ := j2.reserve(k, v1, 1); code != http.StatusCreated {
418 t.Fatalf("%s: %d", k, code)
419 }
420 }
421 if code, _ := j2.reserve("c", v1, 1); code != http.StatusServiceUnavailable {
422 t.Fatalf("third reservation: %d", code)
423 }
424}
425
426// A4: an entry being downloaded is not evicted; with nothing else to reclaim a
427// reservation fails, and succeeds when the reader is done.
428func TestPinnedEntryIsNotEvicted(t *testing.T) {
429 r := newRig(t, 30)
430 j := r.job("j", repoA, main_, main_)
431 for _, k := range []string{"a", "b", "c"} {
432 if c := j.save(k, v1, bytes.Repeat([]byte(k), 10)); c != 204 {
433 t.Fatalf("%s: %d", k, c)
434 }
435 r.clk.add(time.Minute)
436 }
437 e, _ := r.svc.store.find(repoA, []string{main_}, []string{"a"}, v1) // "a" is now the most recent
438 r.clk.add(time.Minute)
439 f, _, release, err := r.svc.store.open(e.ID, repoA)
440 if err != nil {
441 t.Fatal(err)
442 }
443 // Whatever happens: the file first, then the pin (both are idempotent).
444 t.Cleanup(func() { _ = f.Close(); release() })
445 r.clk.add(time.Minute)
446 // "b" is the least recently used unpinned entry; "a" survives although it is
447 // the one a plain LRU would pick after more use.
448 for _, k := range []string{"b", "c"} {
449 r.clk.add(time.Minute)
450 kf, _, rel, err := r.svc.store.open(idOf(t, r, k), repoA)
451 if err != nil {
452 t.Fatal(err)
453 }
454 _ = kf.Close() // files before pins
455 rel()
456 }
457 // Everything but "a" is more recent than "a" now: LRU order puts "a" first.
458 if c := j.save("d", v1, bytes.Repeat([]byte("d"), 10)); c != 204 {
459 t.Fatalf("d: %d", c)
460 }
461 if _, err := os.Stat(r.svc.store.file(e.ID)); err != nil {
462 t.Fatalf("the pinned entry was evicted: %v", err)
463 }
464 if got, _ := io.ReadAll(f); string(got) != strings.Repeat("a", 10) {
465 t.Fatalf("reader got %q", got)
466 }
467 // Reclaiming everything else is not enough for 25 bytes while "a" is pinned.
468 if code, _ := j.reserve("big", v1, 25); code != http.StatusServiceUnavailable {
469 t.Fatalf("reserve with nothing reclaimable: %d", code)
470 }
471 _ = f.Close()
472 release()
473 release() // idempotent
474 if code, _ := j.reserve("big", v1, 25); code != http.StatusCreated {
475 t.Fatalf("after the reader is done: %d", code)
476 }
477 r.svc.store.mu.Lock()
478 pins := len(r.svc.store.pins)
479 r.svc.store.mu.Unlock()
480 if pins != 0 {
481 t.Fatalf("%d pins left", pins)
482 }
483}
484
485func idOf(t *testing.T, r *rig, key string) int64 {
486 t.Helper()
487 e, ok := r.svc.store.find(repoA, []string{main_}, []string{key}, v1)
488 if !ok {
489 t.Fatalf("no entry %q", key)
490 }
491 return e.ID
492}
493
494// A downloaded archive is unpinned when the response is done.
495func TestDownloadReleasesPin(t *testing.T) {
496 r := newRig(t, 100)
497 j := r.job("j", repoA, main_, main_)
498 if c := j.save("k", v1, []byte("data")); c != 204 {
499 t.Fatal(c)
500 }
501 if _, data, ok := j.restore(v1, "k"); !ok || string(data) != "data" {
502 t.Fatalf("%v %q", ok, data)
503 }
504 r.svc.store.mu.Lock()
505 pins := len(r.svc.store.pins)
506 r.svc.store.mu.Unlock()
507 if pins != 0 {
508 t.Fatalf("%d pins after the download", pins)
509 }
510}
511
512// B2: a reserve writes the index only when it evicted something; usage times
513// wait for the housekeeping.
514func TestReserveWritesIndexOnlyOnEviction(t *testing.T) {
515 r := newRig(t, 10)
516 j := r.job("j", repoA, main_, main_)
517 if c := j.save("k", v1, bytes.Repeat([]byte("x"), 8)); c != 204 {
518 t.Fatal(c)
519 }
520 idx := filepath.Join(r.dir, indexName)
521 if err := os.Remove(idx); err != nil {
522 t.Fatal(err)
523 }
524 if code, _ := j.lookup(v1, "k"); code != 200 { // dirties the usage time
525 t.Fatal(code)
526 }
527 if code, _ := j.reserve("small", v1, 2); code != http.StatusCreated { // fits: no eviction
528 t.Fatal(code)
529 }
530 if _, err := os.Stat(idx); !os.IsNotExist(err) {
531 t.Fatalf("the index was written by a reserve that evicted nothing: %v", err)
532 }
533 if code, _ := j.reserve("big", v1, 8); code != http.StatusCreated { // evicts "k"
534 t.Fatal(code)
535 }
536 if _, err := os.Stat(idx); err != nil {
537 t.Fatalf("an eviction must be persisted: %v", err)
538 }
539 // And the housekeeping flush still writes timestamps.
540 if err := r.svc.store.flush(); err != nil {
541 t.Fatal(err)
542 }
543}
544
545// A8: the total of a Content-Range is "*" or a positive size that covers the
546// range.
547func TestParseRange(t *testing.T) {
548 for h, want := range map[string]bool{
549 "bytes 0-3/*": true, "bytes 0-3/4": true, "bytes 0-3/100": true, "bytes 5-5/*": true,
550 "bytes 0-3/": false, "bytes 0-3/0": false, "bytes 0-3/-1": false, "bytes 0-3/3": false, "bytes 0-3/x": false,
551 "bytes 0-3": false, "bytes 3-0/*": false, "bytes -1-3/*": false, "bytes a-b/*": false, "items 0-1/*": false, "": false,
552 "bytes 0-4611686018427387904/*": false,
553 } {
554 if _, _, ok := parseRange(h); ok != want {
555 t.Errorf("parseRange(%q) = %v", h, ok)
556 }
557 }
558 if lo, hi, _ := parseRange("bytes 4-7/*"); lo != 4 || hi != 8 {
559 t.Errorf("bounds %d %d", lo, hi)
560 }
561}
562
563// syncBuf is a log sink that handler goroutines and the test may use together.
564type syncBuf struct {
565 mu sync.Mutex
566 b bytes.Buffer
567}
568
569func (s *syncBuf) Write(p []byte) (int, error) { s.mu.Lock(); defer s.mu.Unlock(); return s.b.Write(p) }
570func (s *syncBuf) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
571
572// GO-012: an eviction marks the index dirty before it is saved; a failed save
573// leaves it dirty, logs, and the housekeeping flush writes it later.
574func TestFailedIndexSaveStaysDirtyAndIsLogged(t *testing.T) {
575 var logs syncBuf
576 r := newRigWith(t, Config{Max: 10, MinCharge: 1, Log: zerowrap.New(zerowrap.Config{Output: &logs, Format: "json", Level: "debug"})})
577 j := r.job("j", repoA, main_, main_)
578 if c := j.save("k", v1, bytes.Repeat([]byte("x"), 8)); c != 204 {
579 t.Fatal(c)
580 }
581 // A directory where the temporary index goes: every save fails.
582 blocker := filepath.Join(r.dir, indexName+".tmp")
583 if err := os.Mkdir(blocker, 0o700); err != nil {
584 t.Fatal(err)
585 }
586 if code, _ := j.reserve("big", v1, 8); code != http.StatusCreated { // evicts "k"
587 t.Fatalf("reserve: %d", code)
588 }
589 r.svc.store.mu.Lock()
590 dirty := r.svc.store.dirty
591 r.svc.store.mu.Unlock()
592 if !dirty {
593 t.Fatal("a failed save must leave the index dirty")
594 }
595 if !strings.Contains(logs.String(), "cache index") {
596 t.Fatalf("the failure was not logged: %q", logs.String())
597 }
598 if err := r.svc.store.flush(); err == nil {
599 t.Fatal("flush must report the failure")
600 }
601 if err := os.Remove(blocker); err != nil {
602 t.Fatal(err)
603 }
604 if err := r.svc.store.flush(); err != nil {
605 t.Fatalf("retry: %v", err)
606 }
607 r.svc.store.mu.Lock()
608 dirty = r.svc.store.dirty
609 r.svc.store.mu.Unlock()
610 if dirty {
611 t.Fatal("dirty after a successful flush")
612 }
613 b, _ := os.ReadFile(filepath.Join(r.dir, indexName))
614 if strings.Contains(string(b), `"key":"k"`) || !strings.HasPrefix(string(b), "[") {
615 t.Fatalf("the evicted entry is still in the index: %s", b)
616 }
617}
618
619// GO-011: a reservation also ends at an absolute time (the token's expiry, or an
620// hour after it was made), however recently it was touched.
621func TestReservationAbsoluteExpiry(t *testing.T) {
622 r := newRig(t, 1<<20)
623 // Touched every 31 minutes, which keeps it from idling out, but it is over
624 // one hour after its creation.
625 j := r.job("j", repoA, main_, main_)
626 _, id := j.reserve("k", v1, 8)
627 r.clk.add(31 * time.Minute)
628 if c := j.chunk(id, []byte("abcd"), 0); c != 204 {
629 t.Fatalf("chunk: %d", c)
630 }
631 r.clk.add(31 * time.Minute)
632 if c := j.chunk(id, []byte("efgh"), 4); c != http.StatusNotFound {
633 t.Fatalf("chunk past the absolute end: %d", c)
634 }
635 if code, _ := j.reserve("other", v1, 1); code != http.StatusCreated {
636 t.Fatal(code)
637 }
638 if n, f := r.svc.store.pendingCount(), dataFiles(t, r); n != 1 || f != 1 {
639 t.Fatalf("pending %d files %d: the old reservation must be reclaimed", n, f)
640 }
641
642 // A token that ends sooner ends the reservation sooner.
643 short := &job{r: r, token: r.svc.Token("js", Scope{Repository: repoA, WriteRef: main_, ReadRefs: []string{main_}}, 10*time.Minute)}
644 _, ids := short.reserve("s", v1, 4)
645 r.svc.store.mu.Lock()
646 end := r.svc.store.pending[ids].expires
647 r.svc.store.mu.Unlock()
648 if want := r.clk.now().Add(10 * time.Minute); end.Sub(want) > time.Second || want.Sub(end) > time.Second {
649 t.Fatalf("expires %v, want about %v", end, want)
650 }
651}
652
653// GO-011: a reservation with a chunk in flight when it runs out is not dropped
654// under the writer; it is reclaimed when the chunk ends, and cannot be committed.
655func TestExpiredReservationIsReclaimedAfterActiveWrite(t *testing.T) {
656 admitted := make(chan int64, 2)
657 r := newRigWith(t, Config{Max: 1 << 20, MinCharge: 1, hookChunk: func(id int64) { admitted <- id }})
658 j := r.job("j", repoA, main_, main_)
659 _, id := j.reserve("k", v1, 8)
660 pw, done := j.chunkReq(id, 0, 8)
661 wait(t, admitted, "the chunk to be admitted")
662
663 r.clk.add(2 * time.Hour)
664 if code, _ := j.reserve("other", v1, 1); code != http.StatusCreated { // runs the expiry
665 t.Fatal(code)
666 }
667 if n := r.svc.store.pendingCount(); n != 2 {
668 t.Fatalf("pending %d: the reservation with a write in flight must stay", n)
669 }
670 if _, err := pw.Write([]byte("abcdefgh")); err != nil {
671 t.Fatal(err)
672 }
673 _ = pw.Close()
674 wait(t, done, "the chunk to end")
675 if n, f := r.svc.store.pendingCount(), dataFiles(t, r); n != 1 || f != 1 {
676 t.Fatalf("pending %d files %d: reclaimed once the write is over", n, f)
677 }
678 if c := j.commit(id, 8); c != http.StatusNotFound {
679 t.Fatalf("commit of a reclaimed reservation: %d", c)
680 }
681 if _, _, held := r.svc.store.stats(); held != 1 {
682 t.Fatalf("held %d", held)
683 }
684}
685
686// GO-011: the deadline of a transfer never goes past the credential's end.
687func TestTransferDeadlineIsCappedAtExpiry(t *testing.T) {
688 near := func(d time.Time, want time.Duration) bool {
689 got := time.Until(d)
690 return got <= want && got > want-5*time.Second
691 }
692 left := 7 * time.Second
693 rem := func() time.Duration { return left }
694 if d := deadline(time.Minute, rem); !near(d, 7*time.Second) {
695 t.Errorf("capped deadline in %v", time.Until(d))
696 }
697 if d := deadline(3*time.Second, rem); !near(d, 3*time.Second) {
698 t.Errorf("progress deadline in %v", time.Until(d))
699 }
700 left = -time.Hour
701 if d := deadline(time.Minute, rem); time.Until(d) > time.Second {
702 t.Errorf("an expired credential gets %v", time.Until(d))
703 }
704 if d := deadline(time.Minute, nil); !near(d, time.Minute) {
705 t.Errorf("no credential: %v", time.Until(d))
706 }
707
708 // The writer of a download uses it on every write.
709 d := &deadlineWriter{ResponseRecorder: *httptest.NewRecorder()}
710 pw := &progressWriter{ResponseWriter: d, rc: http.NewResponseController(d), idle: time.Minute, remaining: func() time.Duration { return 2 * time.Second }}
711 if _, err := pw.Write([]byte("x")); err != nil {
712 t.Fatal(err)
713 }
714 if len(d.deadlines) != 1 || time.Until(d.deadlines[0]) > 2*time.Second {
715 t.Fatalf("deadlines %v", d.deadlines)
716 }
717}
718
719// A token expiry also bounds the transfer through the service: the remaining
720// time comes from the service clock.
721func TestServiceRemaining(t *testing.T) {
722 r := newRig(t, 1<<20)
723 exp := r.clk.now().Add(90 * time.Second).Unix()
724 rem := r.svc.remaining(exp)
725 if got := rem(); got != 90*time.Second {
726 t.Fatalf("remaining %v", got)
727 }
728 r.clk.add(time.Minute)
729 if got := rem(); got != 30*time.Second {
730 t.Fatalf("remaining %v", got)
731 }
732}