| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "path" |
| 11 | "strings" |
| 12 | "time" |
| 13 | |
| 14 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 15 | "github.com/benbjohnson/litestream" |
| 16 | ) |
| 17 | |
| 18 | // ErrLockLocation is LockState's answer for a bucket or location it will |
| 19 | // not read: empty, absolute, or not in clean form. |
| 20 | var ErrLockLocation = errors.New("lease: invalid lock location") |
| 21 | |
| 22 | // lockBodyLimit bounds the lock.json read. The leaser writes a few dozen |
| 23 | // bytes; anything near this is not a lock. |
| 24 | const lockBodyLimit = 64 << 10 |
| 25 | |
| 26 | // LockState reads the single-writer lease at s3://bucket/<location>/lock.json |
| 27 | // as the litestream leaser writes it (Acquire, lease.go), through the same |
| 28 | // client Acquire builds, so the read signs for the region and reaches the |
| 29 | // endpoint the writer used. It is §5.3 step 4's observation: it holds no |
| 30 | // lease and writes nothing. |
| 31 | // |
| 32 | // location is the claim holder's Location from a fresh claim read (the |
| 33 | // location moves.ReplicaSource derives its prefix from, ruling G3), never |
| 34 | // activator.HibernatePrefix and never the g-<generation>/ prefix: the lock |
| 35 | // sits at the location and Held.BindGeneration only fixes the data under |
| 36 | // it. One trailing slash is accepted, so ReplicaSource's root prefix reads |
| 37 | // the same lock. |
| 38 | // |
| 39 | // held is true while the lock is present and its expiry has not passed by |
| 40 | // more than RenewMargin, so a holder whose clock runs slightly behind |
| 41 | // still counts as holding. An absent lock (never acquired, or released: |
| 42 | // the leaser deletes it) is held=false with a zero owner and expiry and a |
| 43 | // nil error; only S3's explicit NoSuchKey counts as absent (lockAbsent). |
| 44 | // An expired lock is held=false with its owner and expiry, for |
| 45 | // the record. Anything else (a missing bucket, an unreadable or garbled |
| 46 | // body, a lock with no expiry) is an error, never "not held". |
| 47 | func LockState(ctx context.Context, bucket, location, region string) (held bool, owner string, expires time.Time, err error) { |
| 48 | key, err := lockKey(bucket, location) |
| 49 | if err != nil { |
| 50 | return false, "", time.Time{}, err |
| 51 | } |
| 52 | client, _, err := newLeaserClient(ctx, region) |
| 53 | if err != nil { |
| 54 | return false, "", time.Time{}, err |
| 55 | } |
| 56 | out, err := client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key}) |
| 57 | if err != nil { |
| 58 | if lockAbsent(err) { |
| 59 | return false, "", time.Time{}, nil |
| 60 | } |
| 61 | return false, "", time.Time{}, fmt.Errorf("lease: read %s: %w", key, err) |
| 62 | } |
| 63 | defer out.Body.Close() |
| 64 | body, err := io.ReadAll(io.LimitReader(out.Body, lockBodyLimit+1)) |
| 65 | if err != nil { |
| 66 | return false, "", time.Time{}, fmt.Errorf("lease: read %s: %w", key, err) |
| 67 | } |
| 68 | if len(body) > lockBodyLimit { |
| 69 | return false, "", time.Time{}, fmt.Errorf("lease: %s is larger than a lock", key) |
| 70 | } |
| 71 | var lock litestream.Lease |
| 72 | if err := json.Unmarshal(body, &lock); err != nil { |
| 73 | return false, "", time.Time{}, fmt.Errorf("lease: decode %s: %w", key, err) |
| 74 | } |
| 75 | if lock.ExpiresAt.IsZero() { |
| 76 | return false, "", time.Time{}, fmt.Errorf("lease: %s has no expiry", key) |
| 77 | } |
| 78 | return lockHeld(lock.ExpiresAt, time.Now()), lock.Owner, lock.ExpiresAt, nil |
| 79 | } |
| 80 | |
| 81 | // lockAbsent reports an explicit S3 object-absence answer: HTTP 404 with an |
| 82 | // error body the SDK parsed to the code NoSuchKey. Absence is what lets a |
| 83 | // cut call the lease free, so nothing weaker counts: not a 404 with a |
| 84 | // malformed, empty or unrecognised body (an endpoint that is not the store, |
| 85 | // a proxy), not NotFound or NoSuchVersion, not NoSuchBucket, and not |
| 86 | // NoSuchKey on any other status. mapReadErr is deliberately not used here: |
| 87 | // its looser reading (any 404 but NoSuchBucket) suits its callers, which |
| 88 | // treat absence as "nothing to copy", not as authority. |
| 89 | func lockAbsent(err error) bool { |
| 90 | code, status := apiCode(err) |
| 91 | return code == "NoSuchKey" && status == http.StatusNotFound |
| 92 | } |
| 93 | |
| 94 | // lockHeld is §5.3 step 4's rule: a lock is released only once its expiry |
| 95 | // has passed by more than the renewal margin. |
| 96 | func lockHeld(expires, now time.Time) bool { |
| 97 | return !now.After(expires.Add(RenewMargin)) |
| 98 | } |
| 99 | |
| 100 | // lockKey is the leaser's key for location (lss3.Leaser.lockKey with Path |
| 101 | // set, as AcquireTTL sets it), after refusing what Acquire would never be |
| 102 | // given: an empty bucket or location, an absolute one, or one not in clean |
| 103 | // form. |
| 104 | func lockKey(bucket, location string) (string, error) { |
| 105 | loc := strings.TrimSuffix(location, "/") |
| 106 | if bucket == "" || loc == "" || strings.HasPrefix(loc, "/") || strings.Contains(loc, `\`) || path.Clean(loc) != loc || loc == "." || strings.HasPrefix(loc, "../") || loc == ".." { |
| 107 | return "", ErrLockLocation |
| 108 | } |
| 109 | return loc + "/lock.json", nil |
| 110 | } |
| 111 | |
amadan