carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package release
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io/fs"
8 "reflect"
9 "sort"
10 "sync"
11 "testing"
12 "time"
13)
14
15// condCounter wraps a DirStore and records the etag every GetIfChanged call
16// carried, so a test can see that the second poll was conditional.
17type condCounter struct {
18 *DirStore
19 mu sync.Mutex
20 etags []string
21 getErr error // when set, GetIfChanged fails with it
22}
23
24func (c *condCounter) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
25 c.mu.Lock()
26 c.etags = append(c.etags, etag)
27 err := c.getErr
28 c.mu.Unlock()
29 if err != nil {
30 return nil, "", false, err
31 }
32 return c.DirStore.GetIfChanged(ctx, key, etag)
33}
34
35func (c *condCounter) setErr(err error) {
36 c.mu.Lock()
37 c.getErr = err
38 c.mu.Unlock()
39}
40
41func (c *condCounter) seen() []string {
42 c.mu.Lock()
43 defer c.mu.Unlock()
44 return append([]string(nil), c.etags...)
45}
46
47func writeScoped(t *testing.T, s Store, keys ...string) {
48 t.Helper()
49 for _, k := range keys {
50 if _, err := s.Put(context.Background(), k, []byte("x"), ""); err != nil {
51 t.Fatal(err)
52 }
53 }
54}
55
56func TestChangeWatchFirstPollReportsAll(t *testing.T) {
57 t.Parallel()
58 d := indexedDirStore(t)
59 writeScoped(t, d, "apps/bab/shop/instances/h.json", "accounts/acme/members.json")
60 w := NewChangeWatch(d)
61 if w.Usable() {
62 t.Fatal("a watch that has not polled is usable")
63 }
64 if _, ok := w.Gen("apps/bab/shop"); ok {
65 t.Fatal("Gen ok before the first poll")
66 }
67 moved, err := w.Poll(context.Background())
68 if err != nil {
69 t.Fatal(err)
70 }
71 want := []string{"accounts/acme", "apps/bab/shop"}
72 if !reflect.DeepEqual(moved, want) {
73 t.Fatalf("moved = %v, want %v", moved, want)
74 }
75 if !w.Usable() {
76 t.Fatal("not usable after a good poll")
77 }
78 if g, ok := w.Gen("apps/bab/shop"); !ok || g != 1 {
79 t.Fatalf("Gen = %d, %v; want 1, true", g, ok)
80 }
81 if g, ok := w.Gen("apps/bab/other"); !ok || g != 0 {
82 t.Fatalf("Gen of an unlisted scope = %d, %v; want 0, true", g, ok)
83 }
84}
85
86func TestChangeWatchUnchangedIs304(t *testing.T) {
87 t.Parallel()
88 c := &condCounter{DirStore: indexedDirStore(t)}
89 writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json")
90 w := NewChangeWatch(c)
91 if _, err := w.Poll(context.Background()); err != nil {
92 t.Fatal(err)
93 }
94 moved, err := w.Poll(context.Background())
95 if err != nil || moved != nil {
96 t.Fatalf("second poll = %v, %v; want nil, nil", moved, err)
97 }
98 seen := c.seen()
99 if len(seen) != 2 || seen[0] != "" || seen[1] == "" {
100 t.Fatalf("etags sent = %q; want an empty one, then the index's", seen)
101 }
102}
103
104func TestChangeWatchReportsOnlyMoved(t *testing.T) {
105 t.Parallel()
106 d := indexedDirStore(t)
107 writeScoped(t, d, "apps/bab/shop/instances/h.json", "accounts/acme/members.json")
108 w := NewChangeWatch(d)
109 if _, err := w.Poll(context.Background()); err != nil {
110 t.Fatal(err)
111 }
112 writeScoped(t, d, "apps/bab/shop/instances/i.json", "control/certs/example.com.json")
113 moved, err := w.Poll(context.Background())
114 if err != nil {
115 t.Fatal(err)
116 }
117 want := []string{"apps/bab/shop", "control/certs"}
118 if !reflect.DeepEqual(moved, want) {
119 t.Fatalf("moved = %v, want %v", moved, want)
120 }
121 if g, _ := w.Gen("apps/bab/shop"); g != 2 {
122 t.Fatalf("Gen = %d, want 2", g)
123 }
124}
125
126func TestChangeWatchAbsentIndexIsUsableEmpty(t *testing.T) {
127 t.Parallel()
128 d := indexedDirStore(t)
129 w := NewChangeWatch(d)
130 moved, err := w.Poll(context.Background())
131 if err != nil || len(moved) != 0 {
132 t.Fatalf("poll = %v, %v; want nothing", moved, err)
133 }
134 if !w.Usable() {
135 t.Fatal("an absent index is the pre-first-bump state and must be usable")
136 }
137 if obs, ok := w.Observed(); !ok || len(obs) != 0 {
138 t.Fatalf("Observed = %v, %v; want empty, true", obs, ok)
139 }
140 // The first write then shows up as an appearance.
141 writeScoped(t, d, "apps/bab/shop/instances/h.json")
142 moved, err = w.Poll(context.Background())
143 if err != nil || !reflect.DeepEqual(moved, []string{"apps/bab/shop"}) {
144 t.Fatalf("poll after first write = %v, %v", moved, err)
145 }
146}
147
148func TestChangeWatchDeniedIsUnusable(t *testing.T) {
149 t.Parallel()
150 c := &condCounter{DirStore: indexedDirStore(t)}
151 writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json")
152 c.setErr(fmt.Errorf("get %s: %w", ChangeIndexKey, fs.ErrPermission))
153 w := NewChangeWatch(c)
154 moved, err := w.Poll(context.Background())
155 if !errors.Is(err, fs.ErrPermission) || moved != nil {
156 t.Fatalf("poll = %v, %v; want the access error", moved, err)
157 }
158 if w.Usable() {
159 t.Fatal("usable after a denied read")
160 }
161 if _, ok := w.Gen("apps/bab/shop"); ok {
162 t.Fatal("Gen ok while unusable")
163 }
164 if _, ok := w.Observed(); ok {
165 t.Fatal("Observed ok while unusable")
166 }
167}
168
169func TestChangeWatchUsableThenErrorThenRecovers(t *testing.T) {
170 t.Parallel()
171 c := &condCounter{DirStore: indexedDirStore(t)}
172 writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json", "accounts/acme/members.json")
173 w := NewChangeWatch(c)
174 if _, err := w.Poll(context.Background()); err != nil {
175 t.Fatal(err)
176 }
177 c.setErr(errors.New("network down"))
178 writeScoped(t, c.DirStore, "apps/bab/shop/instances/i.json")
179 if _, err := w.Poll(context.Background()); err == nil {
180 t.Fatal("poll succeeded through an error")
181 }
182 if w.Usable() {
183 t.Fatal("still usable after an error")
184 }
185 c.setErr(nil)
186 moved, err := w.Poll(context.Background())
187 if err != nil {
188 t.Fatal(err)
189 }
190 if !reflect.DeepEqual(moved, []string{"apps/bab/shop"}) {
191 t.Fatalf("moved = %v; want the scope that rose while unusable", moved)
192 }
193 if !w.Usable() {
194 t.Fatal("not usable after recovering")
195 }
196 // An unchanged index after an error recovers without a spurious move.
197 c.setErr(errors.New("blip"))
198 _, _ = w.Poll(context.Background())
199 c.setErr(nil)
200 moved, err = w.Poll(context.Background())
201 if err != nil || moved != nil || !w.Usable() {
202 t.Fatalf("recovery on an unchanged index = %v, %v, usable %v", moved, err, w.Usable())
203 }
204}
205
206func TestChangeWatchUnknownSchemaIsUnusable(t *testing.T) {
207 t.Parallel()
208 d := NewDirStore(t.TempDir())
209 if _, err := d.Put(context.Background(), ChangeIndexKey, []byte(`{"schema":2,"entries":{"apps/a/b":1}}`), ""); err != nil {
210 t.Fatal(err)
211 }
212 w := NewChangeWatch(d)
213 _, err := w.Poll(context.Background())
214 if !errors.Is(err, ErrChangeIndexSchema) {
215 t.Fatalf("err = %v, want ErrChangeIndexSchema", err)
216 }
217 if w.Usable() {
218 t.Fatal("usable on an unknown schema")
219 }
220 // A malformed body is unknown too.
221 _, etag, _ := d.Get(context.Background(), ChangeIndexKey)
222 if _, err := d.Put(context.Background(), ChangeIndexKey, []byte(`not json`), etag); err != nil {
223 t.Fatal(err)
224 }
225 if _, err := w.Poll(context.Background()); err == nil || w.Usable() {
226 t.Fatalf("malformed index: err %v, usable %v", err, w.Usable())
227 }
228}
229
230func TestChangeWatchWithoutExtensionIsUnusable(t *testing.T) {
231 t.Parallel()
232 w := NewChangeWatch(bareStore{NewDirStore(t.TempDir())})
233 _, err := w.Poll(context.Background())
234 if !errors.Is(err, errors.ErrUnsupported) {
235 t.Fatalf("err = %v, want ErrUnsupported", err)
236 }
237 if w.Usable() {
238 t.Fatal("usable without ConditionalGetter")
239 }
240 // The prefix wrapper over such a store answers ErrUnsupported itself.
241 w = NewChangeWatch(NewPrefixStore(bareStore{NewDirStore(t.TempDir())}, "base/"))
242 if _, err := w.Poll(context.Background()); err == nil || w.Usable() {
243 t.Fatalf("prefixed bare store: err %v, usable %v", err, w.Usable())
244 }
245}
246
247func TestChangeWatchObservedIsACopy(t *testing.T) {
248 t.Parallel()
249 d := indexedDirStore(t)
250 writeScoped(t, d, "apps/bab/shop/instances/h.json")
251 w := NewChangeWatch(d)
252 if _, err := w.Poll(context.Background()); err != nil {
253 t.Fatal(err)
254 }
255 obs, _ := w.Observed()
256 obs["apps/bab/shop"] = 99
257 delete(obs, "nothing")
258 if g, _ := w.Gen("apps/bab/shop"); g != 1 {
259 t.Fatalf("mutating Observed changed the watch: Gen = %d", g)
260 }
261}
262
263func TestChangeWatchConcurrentPolls(t *testing.T) {
264 t.Parallel()
265 d := indexedDirStore(t)
266 writeScoped(t, d, "apps/bab/shop/instances/h.json")
267 w := NewChangeWatch(d)
268 var wg sync.WaitGroup
269 var mu sync.Mutex
270 var all []string
271 for range 8 {
272 wg.Add(1)
273 go func() {
274 defer wg.Done()
275 moved, err := w.Poll(context.Background())
276 if err != nil {
277 t.Error(err)
278 }
279 w.Gen("apps/bab/shop")
280 w.Observed()
281 mu.Lock()
282 all = append(all, moved...)
283 mu.Unlock()
284 }()
285 }
286 wg.Wait()
287 sort.Strings(all)
288 if !reflect.DeepEqual(all, []string{"apps/bab/shop"}) {
289 t.Fatalf("scope reported %v across racing polls; want once", all)
290 }
291}
292
293func TestReadGateDue(t *testing.T) {
294 t.Parallel()
295 const maxAge = 30 * time.Minute
296 t0 := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC)
297 const scope = "apps/bab/shop"
298 ctx := context.Background()
299
300 usableWatch := func(t *testing.T) (*ChangeWatch, *DirStore) {
301 d := indexedDirStore(t)
302 writeScoped(t, d, "apps/bab/shop/instances/h.json")
303 w := NewChangeWatch(d)
304 if _, err := w.Poll(ctx); err != nil {
305 t.Fatal(err)
306 }
307 return w, d
308 }
309
310 tests := []struct {
311 name string
312 run func(t *testing.T) bool // returns Due's answer
313 want bool
314 }{
315 {"nil watch is due", func(t *testing.T) bool {
316 _, due := NewReadGate(nil, maxAge).Due(scope, t0)
317 return due
318 }, true},
319 {"unusable watch is due", func(t *testing.T) bool {
320 g := NewReadGate(NewChangeWatch(NewDirStore(t.TempDir())), maxAge)
321 _, due := g.Due(scope, t0)
322 return due
323 }, true},
324 {"never marked is due", func(t *testing.T) bool {
325 w, _ := usableWatch(t)
326 _, due := NewReadGate(w, maxAge).Due(scope, t0)
327 return due
328 }, true},
329 {"marked ok, same gen, within age is not due", func(t *testing.T) bool {
330 w, _ := usableWatch(t)
331 g := NewReadGate(w, maxAge)
332 gen, _ := g.Due(scope, t0)
333 g.Mark(scope, gen, t0, true)
334 _, due := g.Due(scope, t0.Add(maxAge-time.Second))
335 return due
336 }, false},
337 {"generation moved is due", func(t *testing.T) bool {
338 w, d := usableWatch(t)
339 g := NewReadGate(w, maxAge)
340 gen, _ := g.Due(scope, t0)
341 g.Mark(scope, gen, t0, true)
342 writeScoped(t, d, "apps/bab/shop/instances/i.json")
343 if _, err := w.Poll(ctx); err != nil {
344 t.Fatal(err)
345 }
346 _, due := g.Due(scope, t0.Add(time.Second))
347 return due
348 }, true},
349 {"other scope moving is not due", func(t *testing.T) bool {
350 w, d := usableWatch(t)
351 g := NewReadGate(w, maxAge)
352 gen, _ := g.Due(scope, t0)
353 g.Mark(scope, gen, t0, true)
354 writeScoped(t, d, "accounts/acme/members.json")
355 if _, err := w.Poll(ctx); err != nil {
356 t.Fatal(err)
357 }
358 _, due := g.Due(scope, t0.Add(time.Second))
359 return due
360 }, false},
361 {"age at maxAge is due", func(t *testing.T) bool {
362 w, _ := usableWatch(t)
363 g := NewReadGate(w, maxAge)
364 gen, _ := g.Due(scope, t0)
365 g.Mark(scope, gen, t0, true)
366 _, due := g.Due(scope, t0.Add(maxAge))
367 return due
368 }, true},
369 {"last mark failed is due", func(t *testing.T) bool {
370 w, _ := usableWatch(t)
371 g := NewReadGate(w, maxAge)
372 gen, _ := g.Due(scope, t0)
373 g.Mark(scope, gen, t0, false)
374 _, due := g.Due(scope, t0.Add(time.Second))
375 return due
376 }, true},
377 {"a failed mark after a good one is due", func(t *testing.T) bool {
378 w, _ := usableWatch(t)
379 g := NewReadGate(w, maxAge)
380 gen, _ := g.Due(scope, t0)
381 g.Mark(scope, gen, t0, true)
382 g.Mark(scope, gen, t0.Add(time.Minute), false)
383 _, due := g.Due(scope, t0.Add(2*time.Minute))
384 return due
385 }, true},
386 {"watch going unusable makes a marked scope due", func(t *testing.T) bool {
387 c := &condCounter{DirStore: indexedDirStore(t)}
388 writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json")
389 w := NewChangeWatch(c)
390 if _, err := w.Poll(ctx); err != nil {
391 t.Fatal(err)
392 }
393 g := NewReadGate(w, maxAge)
394 gen, _ := g.Due(scope, t0)
395 g.Mark(scope, gen, t0, true)
396 c.setErr(errors.New("down"))
397 _, _ = w.Poll(ctx)
398 _, due := g.Due(scope, t0.Add(time.Second))
399 return due
400 }, true},
401 }
402 for _, tc := range tests {
403 t.Run(tc.name, func(t *testing.T) {
404 t.Parallel()
405 if got := tc.run(t); got != tc.want {
406 t.Fatalf("due = %v, want %v", got, tc.want)
407 }
408 })
409 }
410}
411
412func TestReadGateCapturedGenSurvivesRaceWithPoll(t *testing.T) {
413 t.Parallel()
414 ctx := context.Background()
415 d := indexedDirStore(t)
416 writeScoped(t, d, "apps/bab/shop/instances/h.json")
417 w := NewChangeWatch(d)
418 if _, err := w.Poll(ctx); err != nil {
419 t.Fatal(err)
420 }
421 g := NewReadGate(w, time.Hour)
422 t0 := time.Now()
423 gen, due := g.Due("apps/bab/shop", t0)
424 if !due {
425 t.Fatal("never-marked scope not due")
426 }
427 // A write lands and is polled while the read is in flight.
428 writeScoped(t, d, "apps/bab/shop/instances/i.json")
429 if _, err := w.Poll(ctx); err != nil {
430 t.Fatal(err)
431 }
432 g.Mark("apps/bab/shop", gen, t0, true) // the generation captured BEFORE the read
433 if _, due := g.Due("apps/bab/shop", t0); !due {
434 t.Fatal("a mark of the pre-read generation hid a write that landed during the read")
435 }
436}
437
438// blockingCond is a DirStore whose GetIfChanged, once armed, parks until
439// released: a hung index GET holding the watch's poll lock.
440type blockingCond struct {
441 *DirStore
442 mu sync.Mutex
443 block chan struct{} // nil = no blocking
444 entered chan struct{}
445}
446
447func (b *blockingCond) arm() (entered, release chan struct{}) {
448 b.mu.Lock()
449 defer b.mu.Unlock()
450 b.block, b.entered = make(chan struct{}), make(chan struct{}, 1)
451 return b.entered, b.block
452}
453
454func (b *blockingCond) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
455 b.mu.Lock()
456 block, entered := b.block, b.entered
457 b.mu.Unlock()
458 if block != nil {
459 entered <- struct{}{}
460 <-block
461 }
462 return b.DirStore.GetIfChanged(ctx, key, etag)
463}
464
465// A Poll waiting behind a stuck one gives up when its context ends. The
466// give-up is a failed poll like any other: the watch is unknown until the
467// next good one (a frozen observation must never vouch for a gated read),
468// but it keeps its ETag and entries, so that next poll is a 304 that
469// reports nothing moved.
470func TestChangeWatchPollLockWaitHonoursContext(t *testing.T) {
471 t.Parallel()
472 d := indexedDirStore(t)
473 writeScoped(t, d, "apps/bab/shop/instances/h.json")
474 b := &blockingCond{DirStore: d}
475 w := NewChangeWatch(b)
476 if _, err := w.Poll(context.Background()); err != nil {
477 t.Fatal(err)
478 }
479 gen, _ := w.Gen("apps/bab/shop")
480
481 entered, release := b.arm()
482 stuck := make(chan error, 1)
483 go func() {
484 _, err := w.Poll(context.Background())
485 stuck <- err
486 }()
487 <-entered
488
489 ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
490 defer cancel()
491 done := make(chan error, 1)
492 go func() {
493 _, err := w.Poll(ctx)
494 done <- err
495 }()
496 select {
497 case err := <-done:
498 if !errors.Is(err, context.DeadlineExceeded) {
499 t.Fatalf("waiting Poll returned %v, want the context's error", err)
500 }
501 case <-time.After(5 * time.Second):
502 close(release)
503 t.Fatal("a Poll waiting behind a stuck one ignored its context")
504 }
505 if w.Usable() {
506 t.Fatal("a give-up on the lock left the watch usable: its observation is frozen")
507 }
508 if _, ok := w.Gen("apps/bab/shop"); ok {
509 t.Fatal("Gen vouched for a scope after a give-up")
510 }
511 if _, ok := w.Observed(); ok {
512 t.Fatal("Observed vouched after a give-up")
513 }
514 close(release)
515 if err := <-stuck; err != nil {
516 t.Fatalf("the stuck Poll, released: %v", err)
517 }
518 if g, ok := w.Gen("apps/bab/shop"); !ok || g != gen {
519 t.Fatalf("Gen after the stuck poll answered = %d, %v; want %d, true", g, ok, gen)
520 }
521}
522
523// A Poll whose context has already ended reads nothing, and is a failed
524// poll: the watch is unknown until the next good one, which is a 304
525// because the watch kept its ETag.
526func TestChangeWatchCancelledPollIsUnknownThenA304(t *testing.T) {
527 t.Parallel()
528 d := indexedDirStore(t)
529 writeScoped(t, d, "apps/bab/shop/instances/h.json")
530 c := &condCounter{DirStore: d}
531 w := NewChangeWatch(c)
532 if _, err := w.Poll(context.Background()); err != nil {
533 t.Fatal(err)
534 }
535 ctx, cancel := context.WithCancel(context.Background())
536 cancel()
537 if _, err := w.Poll(ctx); !errors.Is(err, context.Canceled) {
538 t.Fatalf("a cancelled Poll returned %v", err)
539 }
540 if w.Usable() {
541 t.Fatal("a cancelled Poll left the watch usable")
542 }
543 moved, err := w.Poll(context.Background())
544 if err != nil || moved != nil || !w.Usable() {
545 t.Fatalf("recovery = %v, %v, usable %v; want nothing moved and usable", moved, err, w.Usable())
546 }
547 seen := c.seen()
548 if last := seen[len(seen)-1]; last == "" {
549 t.Fatalf("the recovery poll was unconditional (etags %q): the give-up dropped the ETag", seen)
550 }
551}
552