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