carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostbox
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "testing"
8 "time"
9)
10
11type renewableFake struct {
12 lost chan struct{}
13 released int
14}
15
16func (h *renewableFake) Err() error {
17 select {
18 case <-h.lost:
19 return ErrChanged
20 default:
21 return nil
22 }
23}
24func (h *renewableFake) Lost() <-chan struct{} { return h.lost }
25func (h *renewableFake) Release(context.Context) error { h.released++; return nil }
26func TestRenewingLeaseCancelsRequestsAndReleasesOnce(t *testing.T) {
27 h := &renewableFake{lost: make(chan struct{})}
28 l := RenewingLeases{Region: "eu-west-1", Open: func(context.Context, string, string, string) (Renewable, error) { return h, nil }}
29 held, err := l.Acquire(t.Context(), "bucket", "instances/x.example/inc")
30 if err != nil {
31 t.Fatal(err)
32 }
33 ctx := held.(interface {
34 Context(context.Context) context.Context
35 }).Context(t.Context())
36 close(h.lost)
37 select {
38 case <-ctx.Done():
39 case <-time.After(time.Second):
40 t.Fatal("lease loss did not cancel request")
41 }
42 if !errors.Is(held.Check(ctx), context.Canceled) {
43 t.Fatal("lost handle accepted work")
44 }
45 held.Close()
46 held.Close()
47 if h.released != 1 {
48 t.Fatalf("released %d times", h.released)
49 }
50}
51func TestAggregateContextCancelsOnAnyConstituent(t *testing.T) {
52 a, b := &renewableFake{lost: make(chan struct{})}, &renewableFake{lost: make(chan struct{})}
53 leases := RenewingLeases{Region: "r", Open: func(_ context.Context, _, prefix, _ string) (Renewable, error) {
54 if prefix == "a" {
55 return a, nil
56 }
57 return b, nil
58 }}
59 held, err := (AggregateLeases{Leases: leases, Resolve: func(context.Context, string, string) ([]string, error) { return []string{"b", "a"}, nil }}).Acquire(t.Context(), "bucket", "a")
60 if err != nil {
61 t.Fatal(err)
62 }
63 defer held.Close()
64 ctx := held.(interface {
65 Context(context.Context) context.Context
66 }).Context(t.Context())
67 close(b.lost)
68 select {
69 case <-ctx.Done():
70 case <-time.After(time.Second):
71 t.Fatal("aggregate loss not propagated")
72 }
73}
74
75type releaseContextLease struct {
76 renewableFake
77 check func(context.Context) error
78}
79
80func (h *releaseContextLease) Release(ctx context.Context) error { h.released++; return h.check(ctx) }
81func TestRenewingReleaseSurvivesItsOwnCancellation(t *testing.T) {
82 for _, cancelled := range []bool{false, true} {
83 t.Run(fmt.Sprint(cancelled), func(t *testing.T) {
84 h := &releaseContextLease{renewableFake: renewableFake{lost: make(chan struct{})}}
85 l := RenewingLeases{Region: "eu-west-1", Open: func(context.Context, string, string, string) (Renewable, error) { return h, nil }}
86 held, err := l.Acquire(t.Context(), "bucket", "instances/x.example/inc")
87 if err != nil {
88 t.Fatal(err)
89 }
90 ctx, cancel := context.WithCancel(held.(interface {
91 Context(context.Context) context.Context
92 }).Context(t.Context()))
93 defer cancel()
94 if cancelled {
95 cancel()
96 }
97 h.check = func(cleanup context.Context) error {
98 if err := cleanup.Err(); err != nil {
99 return err
100 }
101 deadline, ok := cleanup.Deadline()
102 if !ok || time.Until(deadline) > RequestLimit || time.Until(deadline) <= 0 {
103 t.Fatal("release is not request bounded")
104 }
105 select {
106 case <-ctx.Done():
107 case <-time.After(time.Second):
108 t.Fatal("work context still active during release")
109 }
110 return cleanup.Err()
111 }
112 if err := held.Release(ctx); err != nil {
113 t.Fatal("release cancelled itself", err)
114 }
115 held.Close()
116 if h.released != 1 {
117 t.Fatalf("released %d times", h.released)
118 }
119 if err := held.Check(t.Context()); err == nil {
120 t.Fatal("released lease still grants work")
121 }
122 })
123 }
124}
125