| 1 | package lease |
| 2 | |
| 3 | // The tombstone lease: the offline operator move's linearization point. |
| 4 | // |
| 5 | // A move that has stopped a host on its source box writes a lease at the |
| 6 | // host's own lock.json under the owner "moved:<id>" and a long expiry. From |
| 7 | // then on lease.Acquire on that location fails with a *HeldError naming the |
| 8 | // move, so no box can wake the host and write to the replica behind the |
| 9 | // move's back. Three operations manage it, each a conditional write over the |
| 10 | // client AcquireTTL builds (IfNoneMatch "*" on create, IfMatch on replace |
| 11 | // or delete) and none of them starts a renewer: the move's sealer re-asserts |
| 12 | // the tombstone itself on every pass. |
| 13 | // |
| 14 | // - Tombstone writes it. It never replaces a live holder's lease. |
| 15 | // - Reassert pushes the expiry out again, only on a lease the same move |
| 16 | // owns, with the ETag it just read. |
| 17 | // - LiftTombstone lifts it, only when the owner is a moved:<id> lease the |
| 18 | // caller claims as its own, with the ETag it just read. It lifts by a |
| 19 | // conditional PUT that rewrites the lease as already expired, never by a |
| 20 | // DELETE: Ceph RGW and MinIO are not known to enforce If-Match on a |
| 21 | // DELETE, and the conditional PUT is the primitive the rest of this |
| 22 | // package relies on. IsLiftedTombstone reads that state back. |
| 23 | // |
| 24 | // "Free" is exactly what Acquire means by it (leaseFree): the lease's expiry |
| 25 | // has passed. There is no RenewMargin here. A tombstone that is free to a |
| 26 | // wake is free to the next Tombstone, and the two cannot drift apart. |
| 27 | |
| 28 | import ( |
| 29 | "bytes" |
| 30 | "context" |
| 31 | "encoding/json" |
| 32 | "errors" |
| 33 | "fmt" |
| 34 | "io" |
| 35 | "net/http" |
| 36 | "strings" |
| 37 | "time" |
| 38 | |
| 39 | "github.com/aws/aws-sdk-go-v2/aws" |
| 40 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 41 | "github.com/benbjohnson/litestream" |
| 42 | lss3 "github.com/benbjohnson/litestream/s3" |
| 43 | ) |
| 44 | |
| 45 | // TombstonePrefix starts the owner of every tombstone lease. The three |
| 46 | // functions here refuse any other owner. |
| 47 | const TombstonePrefix = "moved:" |
| 48 | |
| 49 | // MaxTombstoneTTL is the longest a tombstone is ever written for. The moves |
| 50 | // package's TombstoneTTL equals it; this package cannot import that one. |
| 51 | const MaxTombstoneTTL = 7 * 24 * time.Hour |
| 52 | |
| 53 | var ( |
| 54 | // ErrTombstoneOwner refuses an owner that is not moved:<id>. |
| 55 | ErrTombstoneOwner = errors.New("lease: a tombstone owner is moved:<id>") |
| 56 | // ErrTombstoneTTL refuses a TTL that is not positive or exceeds |
| 57 | // MaxTombstoneTTL. |
| 58 | ErrTombstoneTTL = errors.New("lease: a tombstone lives for at most 7 days") |
| 59 | // ErrTombstoneChanged reports that the lease was rewritten or removed |
| 60 | // between the read and the conditional write. Nothing was overwritten; |
| 61 | // the caller's next pass reads it afresh. |
| 62 | ErrTombstoneChanged = errors.New("lease: the lock changed under the tombstone write") |
| 63 | // ErrTombstoneLost reports that the tombstone Reassert was asked to |
| 64 | // keep is absent or expired: the stop was lost. Nothing was written. |
| 65 | ErrTombstoneLost = errors.New("lease: the tombstone is absent or expired; the stop was lost") |
| 66 | ) |
| 67 | |
| 68 | // leaseFree is the predicate litestream's AcquireLease applies to the lease |
| 69 | // it reads (Lease.IsExpired: now is after the expiry), which is what decides |
| 70 | // whether lease.Acquire takes a lock over. Tombstone, Reassert and |
| 71 | // IsLiftedTombstone use it so a lease is free to them exactly when it is free |
| 72 | // to a wake. TestTombstoneAndAcquireAgreeOnFree pins the two together. |
| 73 | func leaseFree(expires, now time.Time) bool { return now.After(expires) } |
| 74 | |
| 75 | // liftedExpiry is the expiry LiftTombstone writes: long past, so every |
| 76 | // reading (leaseFree, and LockState's RenewMargin) agrees the lease is free. |
| 77 | var liftedExpiry = time.Unix(0, 0).UTC() |
| 78 | |
| 79 | // IsLiftedTombstone reports whether a lock read as owner and expires (from |
| 80 | // LockState) is a lifted tombstone: a moved:<id> lease that has expired. That |
| 81 | // is what LiftTombstone leaves behind, and it is free to every Acquire (which |
| 82 | // already treats any expired lease as free) and to the next Tombstone. It is |
| 83 | // false for an ordinary process's lease, expired or not, and for a tombstone |
| 84 | // that is still unexpired. An absent lock is not "lifted" here: callers read |
| 85 | // absence from LockState's zero owner and expiry and treat it the same way. |
| 86 | func IsLiftedTombstone(owner string, expires time.Time) bool { |
| 87 | return tombstoneOwnerOK(owner) && !expires.IsZero() && leaseFree(expires, time.Now()) |
| 88 | } |
| 89 | |
| 90 | func tombstoneOwnerOK(owner string) bool { |
| 91 | return strings.HasPrefix(owner, TombstonePrefix) && len(owner) > len(TombstonePrefix) |
| 92 | } |
| 93 | |
| 94 | func tombstoneTTLOK(ttl time.Duration) bool { return ttl > 0 && ttl <= MaxTombstoneTTL } |
| 95 | |
| 96 | // tombstoneLeaser is the leaser a tombstone call writes with, built over the |
| 97 | // client AcquireTTL builds, and the lock's key. ttl is zero for calls that |
| 98 | // write no expiry (LiftTombstone). |
| 99 | func tombstoneLeaser(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) (*lss3.Leaser, string, error) { |
| 100 | key, err := lockKey(bucket, location) |
| 101 | if err != nil { |
| 102 | return nil, "", err |
| 103 | } |
| 104 | client, _, err := newLeaserClient(ctx, region) |
| 105 | if err != nil { |
| 106 | return nil, "", err |
| 107 | } |
| 108 | leaser := lss3.NewLeaser() |
| 109 | leaser.SetClient(client) |
| 110 | leaser.Bucket = bucket |
| 111 | leaser.Path = strings.TrimSuffix(location, "/") |
| 112 | leaser.TTL = ttl |
| 113 | leaser.Owner = owner |
| 114 | return leaser, key, nil |
| 115 | } |
| 116 | |
| 117 | // readLock reads the lock with the ETag it was read at. Only S3's explicit |
| 118 | // NoSuchKey is absence (lockAbsent); a garbled body, an oversized one or a |
| 119 | // lock with no expiry is an error, never "free". |
| 120 | func readLock(ctx context.Context, leaser *lss3.Leaser, key string) (lock *litestream.Lease, absent bool, err error) { |
| 121 | out, err := leaser.Client().GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(leaser.Bucket), Key: aws.String(key)}) |
| 122 | if err != nil { |
| 123 | if lockAbsent(err) { |
| 124 | return nil, true, nil |
| 125 | } |
| 126 | return nil, false, fmt.Errorf("lease: read %s: %w", key, err) |
| 127 | } |
| 128 | defer out.Body.Close() |
| 129 | body, err := io.ReadAll(io.LimitReader(out.Body, lockBodyLimit+1)) |
| 130 | if err != nil { |
| 131 | return nil, false, fmt.Errorf("lease: read %s: %w", key, err) |
| 132 | } |
| 133 | if len(body) > lockBodyLimit { |
| 134 | return nil, false, fmt.Errorf("lease: %s is larger than a lock", key) |
| 135 | } |
| 136 | var l litestream.Lease |
| 137 | if err := json.Unmarshal(body, &l); err != nil { |
| 138 | return nil, false, fmt.Errorf("lease: decode %s: %w", key, err) |
| 139 | } |
| 140 | if l.ExpiresAt.IsZero() { |
| 141 | return nil, false, fmt.Errorf("lease: %s has no expiry", key) |
| 142 | } |
| 143 | if out.ETag == nil || *out.ETag == "" { |
| 144 | return nil, false, fmt.Errorf("lease: %s came back without an ETag", key) |
| 145 | } |
| 146 | l.ETag = *out.ETag |
| 147 | return &l, false, nil |
| 148 | } |
| 149 | |
| 150 | func heldErr(l *litestream.Lease) error { |
| 151 | return fmt.Errorf("acquire lease: %w", &HeldError{Owner: l.Owner, ExpiresAt: l.ExpiresAt}) |
| 152 | } |
| 153 | |
| 154 | // Tombstone writes the lease at s3://bucket/<location>/lock.json owned by |
| 155 | // owner (moved:<id>), expiring ttl from now (at most MaxTombstoneTTL). |
| 156 | // |
| 157 | // - absent, or expired (as Acquire reads it): created (IfNoneMatch "*") or |
| 158 | // replaced (IfMatch on the expired lease's ETag); |
| 159 | // - live and held by owner: nil and no write; the caller re-asserts with |
| 160 | // Reassert; |
| 161 | // - live and held by anyone else: a *HeldError, nothing written; |
| 162 | // - the lock changed under the write and cannot be read back: |
| 163 | // ErrTombstoneChanged. |
| 164 | // |
| 165 | // A live holder is never replaced: the replace is conditional on the ETag of |
| 166 | // the lease that was read as expired, so a holder that renews in between |
| 167 | // wins and Tombstone reports it as held. |
| 168 | func Tombstone(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) error { |
| 169 | if !tombstoneOwnerOK(owner) { |
| 170 | return ErrTombstoneOwner |
| 171 | } |
| 172 | if !tombstoneTTLOK(ttl) { |
| 173 | return ErrTombstoneTTL |
| 174 | } |
| 175 | leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, owner, ttl) |
| 176 | if err != nil { |
| 177 | return err |
| 178 | } |
| 179 | cur, absent, err := readLock(ctx, leaser, key) |
| 180 | if err != nil { |
| 181 | return err |
| 182 | } |
| 183 | if !absent && !leaseFree(cur.ExpiresAt, time.Now()) { |
| 184 | if cur.Owner == owner { |
| 185 | return nil |
| 186 | } |
| 187 | return heldErr(cur) |
| 188 | } |
| 189 | // Absent or expired. AcquireLease reads again and writes conditionally: |
| 190 | // IfNoneMatch "*" when absent, IfMatch on the ETag it reads otherwise. |
| 191 | // If a holder appeared since our read it answers LeaseExistsError. |
| 192 | if _, err := leaser.AcquireLease(ctx); err != nil { |
| 193 | var lexists *litestream.LeaseExistsError |
| 194 | if errors.As(err, &lexists) { |
| 195 | switch lexists.Owner { |
| 196 | case "": |
| 197 | // The write lost its condition and AcquireLease could not |
| 198 | // read the winner back (it was removed again, or the read |
| 199 | // failed). There is no holder to name. |
| 200 | return fmt.Errorf("tombstone lease: %w", ErrTombstoneChanged) |
| 201 | case owner: |
| 202 | return nil |
| 203 | } |
| 204 | return fmt.Errorf("acquire lease: %w", &HeldError{Owner: lexists.Owner, ExpiresAt: lexists.ExpiresAt}) |
| 205 | } |
| 206 | return fmt.Errorf("tombstone lease: %w", err) |
| 207 | } |
| 208 | return nil |
| 209 | } |
| 210 | |
| 211 | // Reassert moves the expiry of a tombstone owner already holds to now+ttl: |
| 212 | // |
| 213 | // - live and held by owner: rewritten with IfMatch on the ETag just read. |
| 214 | // If the lock changed since, ErrTombstoneChanged and nothing is written; |
| 215 | // - absent, or expired (including lifted), whoever wrote it: |
| 216 | // ErrTombstoneLost, and nothing is written. The stop was lost: in |
| 217 | // between a wake may have acquired, written and released the lease, so |
| 218 | // a tombstone is never taken again here; |
| 219 | // - live and held by anyone else: a *HeldError, nothing written. |
| 220 | // |
| 221 | // Callers read LockState first and treat absent or expired as a lost stop; |
| 222 | // ErrTombstoneLost is the same verdict for a lock that went between that |
| 223 | // read and this one. Never Reassert after Lift. |
| 224 | func Reassert(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) error { |
| 225 | if !tombstoneOwnerOK(owner) { |
| 226 | return ErrTombstoneOwner |
| 227 | } |
| 228 | if !tombstoneTTLOK(ttl) { |
| 229 | return ErrTombstoneTTL |
| 230 | } |
| 231 | leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, owner, ttl) |
| 232 | if err != nil { |
| 233 | return err |
| 234 | } |
| 235 | cur, absent, err := readLock(ctx, leaser, key) |
| 236 | if err != nil { |
| 237 | return err |
| 238 | } |
| 239 | if absent { |
| 240 | return fmt.Errorf("reassert tombstone: %w", ErrTombstoneLost) |
| 241 | } |
| 242 | if leaseFree(cur.ExpiresAt, time.Now()) { |
| 243 | return fmt.Errorf("reassert tombstone: %w (%s, expired %s)", ErrTombstoneLost, cur.Owner, cur.ExpiresAt.UTC().Format(time.RFC3339)) |
| 244 | } |
| 245 | if cur.Owner != owner { |
| 246 | return heldErr(cur) |
| 247 | } |
| 248 | if _, err := leaser.RenewLease(ctx, cur); err != nil { |
| 249 | if errors.Is(err, litestream.ErrLeaseNotHeld) { |
| 250 | return fmt.Errorf("reassert tombstone: %w", ErrTombstoneChanged) |
| 251 | } |
| 252 | return fmt.Errorf("reassert tombstone: %w", err) |
| 253 | } |
| 254 | return nil |
| 255 | } |
| 256 | |
| 257 | // LiftTombstone lifts the tombstone at s3://bucket/<location>/lock.json by |
| 258 | // rewriting it as an expired lease of the same owner and generation, with |
| 259 | // IfMatch on the ETag just read. It never deletes the object. |
| 260 | // |
| 261 | // - absent: (false, nil), and mine is not consulted; |
| 262 | // - owner is not a moved:<id> that mine accepts: (false, *HeldError), |
| 263 | // nothing written. This holds for an expired lease too: only the move |
| 264 | // that wrote a tombstone lifts it; |
| 265 | // - mine and already lifted (expired, IsLiftedTombstone): (false, nil), |
| 266 | // nothing written; |
| 267 | // - mine and live: rewritten, (true, nil). Acquire takes the lease from |
| 268 | // here, as it takes any expired lease, and so does Tombstone; |
| 269 | // - the lock was rewritten since the read: (false, ErrTombstoneChanged), |
| 270 | // and the rewrite stands. One that was removed since is (false, nil). |
| 271 | // |
| 272 | // After a lift, LockState reports held=false with the move's owner and a |
| 273 | // past expiry; IsLiftedTombstone recognises that. Never Reassert after it. |
| 274 | func LiftTombstone(ctx context.Context, bucket, location, region string, mine func(owner string) bool) (lifted bool, err error) { |
| 275 | leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0) |
| 276 | if err != nil { |
| 277 | return false, err |
| 278 | } |
| 279 | cur, absent, err := readLock(ctx, leaser, key) |
| 280 | if err != nil { |
| 281 | return false, err |
| 282 | } |
| 283 | if absent { |
| 284 | return false, nil |
| 285 | } |
| 286 | if !tombstoneOwnerOK(cur.Owner) || mine == nil || !mine(cur.Owner) { |
| 287 | return false, heldErr(cur) |
| 288 | } |
| 289 | if IsLiftedTombstone(cur.Owner, cur.ExpiresAt) { |
| 290 | return false, nil |
| 291 | } |
| 292 | body, err := json.Marshal(&litestream.Lease{Generation: cur.Generation, ExpiresAt: liftedExpiry, Owner: cur.Owner}) |
| 293 | if err != nil { |
| 294 | return false, fmt.Errorf("lift tombstone: %w", err) |
| 295 | } |
| 296 | _, err = leaser.Client().PutObject(ctx, &awss3.PutObjectInput{ |
| 297 | Bucket: aws.String(leaser.Bucket), |
| 298 | Key: aws.String(key), |
| 299 | Body: bytes.NewReader(body), |
| 300 | ContentType: aws.String("application/json"), |
| 301 | IfMatch: aws.String(cur.ETag), |
| 302 | }) |
| 303 | if err == nil { |
| 304 | return true, nil |
| 305 | } |
| 306 | if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed || code == "NoSuchKey" || status == http.StatusNotFound { |
| 307 | // Rewritten, or gone (AWS answers a conditional PUT on a missing key |
| 308 | // 404, some stores 412). Read it back to tell which. |
| 309 | if _, gone, rerr := readLock(ctx, leaser, key); rerr == nil && gone { |
| 310 | return false, nil |
| 311 | } |
| 312 | return false, fmt.Errorf("lift tombstone: %w", ErrTombstoneChanged) |
| 313 | } |
| 314 | return false, fmt.Errorf("lift tombstone: %w", err) |
| 315 | } |
| 316 | |
amadan