| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/aws/aws-sdk-go-v2/aws" |
| 14 | "github.com/aws/aws-sdk-go-v2/credentials" |
| 15 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 16 | "github.com/benbjohnson/litestream" |
| 17 | lss3 "github.com/benbjohnson/litestream/s3" |
| 18 | ) |
| 19 | |
| 20 | // deleteStub answers the lease's DELETE with each status in turn and |
| 21 | // records the If-Match header it was sent. |
| 22 | type deleteStub struct { |
| 23 | mu sync.Mutex |
| 24 | statuses []int |
| 25 | ifMatch []string |
| 26 | } |
| 27 | |
| 28 | func (s *deleteStub) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 29 | if r.Method != http.MethodDelete { |
| 30 | http.Error(w, "unexpected "+r.Method, http.StatusMethodNotAllowed) |
| 31 | return |
| 32 | } |
| 33 | s.mu.Lock() |
| 34 | s.ifMatch = append(s.ifMatch, r.Header.Get("If-Match")) |
| 35 | status := http.StatusNoContent |
| 36 | if len(s.statuses) > 0 { |
| 37 | status, s.statuses = s.statuses[0], s.statuses[1:] |
| 38 | } |
| 39 | s.mu.Unlock() |
| 40 | switch status { |
| 41 | case http.StatusNoContent: |
| 42 | w.WriteHeader(status) |
| 43 | case http.StatusPreconditionFailed: |
| 44 | w.Header().Set("Content-Type", "application/xml") |
| 45 | w.WriteHeader(status) |
| 46 | _, _ = w.Write([]byte("<Error><Code>PreconditionFailed</Code><Message>no</Message></Error>")) |
| 47 | default: |
| 48 | w.Header().Set("Content-Type", "application/xml") |
| 49 | w.WriteHeader(status) |
| 50 | _, _ = w.Write([]byte("<Error><Code>AccessDenied</Code><Message>no</Message></Error>")) |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | func (s *deleteStub) sent() []string { |
| 55 | s.mu.Lock() |
| 56 | defer s.mu.Unlock() |
| 57 | return append([]string(nil), s.ifMatch...) |
| 58 | } |
| 59 | |
| 60 | // heldAgainst builds a Held whose lease record has ETag etag, with its |
| 61 | // leaser pointed at srv. The TTL is long, so the renewer never renews |
| 62 | // inside the test. |
| 63 | func heldAgainst(t *testing.T, srv *httptest.Server, etag string) *Held { |
| 64 | t.Helper() |
| 65 | client := awss3.New(awss3.Options{ |
| 66 | Region: "us-east-1", |
| 67 | BaseEndpoint: aws.String(srv.URL), |
| 68 | UsePathStyle: true, |
| 69 | Credentials: credentials.NewStaticCredentialsProvider("k", "s", ""), |
| 70 | RetryMaxAttempts: 1, |
| 71 | }) |
| 72 | leaser := lss3.NewLeaser() |
| 73 | leaser.SetClient(client) |
| 74 | leaser.Bucket = "b" |
| 75 | leaser.Path = "p" |
| 76 | leaser.TTL = time.Hour |
| 77 | rec := &litestream.Lease{Generation: 1, ETag: etag, Owner: "test", ExpiresAt: time.Now().Add(time.Hour)} |
| 78 | return &Held{leaser: leaser, rn: startRenewer(leaser, rec, time.Now(), marginForTTL(time.Hour))} |
| 79 | } |
| 80 | |
| 81 | // A failed release leaves a conditional obligation that RetryRelease |
| 82 | // discharges with the same ETag (#292): Release spends the handle, so |
| 83 | // without it the lease stays held for its whole TTL. |
| 84 | func TestRetryReleaseDischargesAFailedRelease(t *testing.T) { |
| 85 | stub := &deleteStub{statuses: []int{http.StatusForbidden, http.StatusNoContent}} |
| 86 | srv := httptest.NewServer(stub) |
| 87 | defer srv.Close() |
| 88 | h := heldAgainst(t, srv, `"e1"`) |
| 89 | ctx := context.Background() |
| 90 | |
| 91 | if err := h.Release(ctx); err == nil { |
| 92 | t.Fatal("Release succeeded against a refusing store") |
| 93 | } |
| 94 | if !h.ReleaseOwed() { |
| 95 | t.Fatal("a failed release left no obligation") |
| 96 | } |
| 97 | if err := h.RetryRelease(ctx); err != nil { |
| 98 | t.Fatalf("RetryRelease: %v", err) |
| 99 | } |
| 100 | if h.ReleaseOwed() { |
| 101 | t.Fatal("the obligation outlived a successful retry") |
| 102 | } |
| 103 | if err := h.RetryRelease(ctx); err != nil { |
| 104 | t.Fatalf("RetryRelease with nothing owed: %v", err) |
| 105 | } |
| 106 | got := stub.sent() |
| 107 | if len(got) != 2 || got[0] != `"e1"` || got[1] != `"e1"` { |
| 108 | t.Fatalf("If-Match sent = %q, want the lease's ETag on both deletes, and no third", got) |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | // A retry whose condition fails (the lease expired and someone else holds |
| 113 | // it now) has nothing of ours left to release: the obligation is resolved, |
| 114 | // and nothing else's lease is deleted. |
| 115 | func TestRetryReleaseResolvesWhenTheLeaseIsNoLongerOurs(t *testing.T) { |
| 116 | stub := &deleteStub{statuses: []int{http.StatusForbidden, http.StatusPreconditionFailed}} |
| 117 | srv := httptest.NewServer(stub) |
| 118 | defer srv.Close() |
| 119 | h := heldAgainst(t, srv, `"e1"`) |
| 120 | ctx := context.Background() |
| 121 | _ = h.Release(ctx) |
| 122 | if err := h.RetryRelease(ctx); err != nil { |
| 123 | t.Fatalf("RetryRelease after a failed condition: %v", err) |
| 124 | } |
| 125 | if h.ReleaseOwed() { |
| 126 | t.Fatal("a lease that is no longer ours is still owed") |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | // A successful release owes nothing. |
| 131 | func TestASuccessfulReleaseOwesNothing(t *testing.T) { |
| 132 | stub := &deleteStub{} |
| 133 | srv := httptest.NewServer(stub) |
| 134 | defer srv.Close() |
| 135 | h := heldAgainst(t, srv, `"e1"`) |
| 136 | if err := h.Release(context.Background()); err != nil { |
| 137 | t.Fatalf("Release: %v", err) |
| 138 | } |
| 139 | if h.ReleaseOwed() { |
| 140 | t.Fatal("a successful release left an obligation") |
| 141 | } |
| 142 | } |
| 143 | |
| 144 | // Round 2 P2: under a persistent DELETE failure with acquisition still |
| 145 | // allowed, obligations must not pile up forever. Each failed release stays |
| 146 | // owed until its own record's expiry; past it, a retry that still fails |
| 147 | // retires the obligation (ErrReleaseLapsed) rather than retrying forever. |
| 148 | // Two acquisitions in a row, both failing to release, across that expiry. |
| 149 | func TestAnOwedReleaseIsRetiredOnceItsRecordHasExpired(t *testing.T) { |
| 150 | stub := &deleteStub{statuses: []int{ |
| 151 | http.StatusForbidden, http.StatusForbidden, http.StatusForbidden, http.StatusForbidden, |
| 152 | http.StatusForbidden, http.StatusForbidden, http.StatusForbidden, http.StatusForbidden, |
| 153 | }} |
| 154 | srv := httptest.NewServer(stub) |
| 155 | defer srv.Close() |
| 156 | ctx := context.Background() |
| 157 | expires := time.Now().Add(300 * time.Millisecond) |
| 158 | var held []*Held |
| 159 | for i := 0; i < 2; i++ { |
| 160 | h := heldExpiring(t, srv, fmt.Sprintf(`"e%d"`, i), expires) |
| 161 | if err := h.Release(ctx); err == nil { |
| 162 | t.Fatal("Release succeeded against a refusing store") |
| 163 | } |
| 164 | held = append(held, h) |
| 165 | } |
| 166 | for _, h := range held { |
| 167 | if err := h.RetryRelease(ctx); err == nil || errors.Is(err, ErrReleaseLapsed) { |
| 168 | t.Fatalf("a retry before the record's expiry = %v, want the store's refusal", err) |
| 169 | } |
| 170 | if !h.ReleaseOwed() { |
| 171 | t.Fatal("an unexpired record stopped being owed") |
| 172 | } |
| 173 | } |
| 174 | time.Sleep(time.Until(expires) + 50*time.Millisecond) |
| 175 | for _, h := range held { |
| 176 | if err := h.RetryRelease(ctx); !errors.Is(err, ErrReleaseLapsed) { |
| 177 | t.Fatalf("a failed retry past the record's expiry = %v, want ErrReleaseLapsed", err) |
| 178 | } |
| 179 | if h.ReleaseOwed() { |
| 180 | t.Fatal("an expired record is still owed") |
| 181 | } |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | // heldExpiring is heldAgainst with the record's own expiry set. |
| 186 | func heldExpiring(t *testing.T, srv *httptest.Server, etag string, expires time.Time) *Held { |
| 187 | t.Helper() |
| 188 | h := heldAgainst(t, srv, etag) |
| 189 | h.rn.mu.Lock() |
| 190 | h.rn.lease.ExpiresAt = expires |
| 191 | h.rn.mu.Unlock() |
| 192 | return h |
| 193 | } |
| 194 | |
amadan