251 lines
10 KiB
Go
251 lines
10 KiB
Go
package pgstore
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// idempotency.go: the storage half of `Idempotency-Key`.
|
|
|
|
// ErrKeyReused is the same key presented for a DIFFERENT request. Told apart from a replay because
|
|
// the remedies differ: one is "your retry worked", the other is "your client has a bug".
|
|
var ErrKeyReused = errors.New("pgstore: this idempotency key was used for another request")
|
|
|
|
// ErrKeyInFlight is a repeat that arrived while the first attempt is still running. Waiting is the
|
|
// remedy, which is why the answer carries Retry-After.
|
|
var ErrKeyInFlight = errors.New("pgstore: the first request under this idempotency key is still running")
|
|
|
|
// IdempotencyKey identifies one claim: (principal, method, path), as the canon ratified — the same
|
|
// key on another operation is ANOTHER key, not a conflict.
|
|
//
|
|
// The path is KEYED by its digest and stored beside it: an address is unbounded, and an over-long
|
|
// one overflowed the index tuple, so the claim failed and the caller was answered 500 where the
|
|
// contract promises 409.
|
|
type IdempotencyKey struct {
|
|
UserID string
|
|
Method string
|
|
Path string
|
|
Key string
|
|
Fingerprint []byte
|
|
}
|
|
|
|
// pathHash is what the index is keyed by. Derived rather than carried, so it cannot disagree with
|
|
// the path beside it.
|
|
func (k IdempotencyKey) pathHash() []byte {
|
|
sum := sha256.Sum256([]byte(k.Path))
|
|
return sum[:]
|
|
}
|
|
|
|
// IdempotentResponse is a completed first attempt, replayed verbatim.
|
|
type IdempotentResponse struct {
|
|
Status int
|
|
Location string
|
|
Body []byte
|
|
// ContentSHA256 is the digest of the body that attempt accepted, for a request whose identity its
|
|
// FINGERPRINT cannot settle: a multipart upload declares its parts and not its bytes, so two
|
|
// different books can arrive under one key with the same declaration. A caller holding one of
|
|
// these may not replay until it has read the repeat's body and found the same digest.
|
|
//
|
|
// Absent where the fingerprint IS the whole request — a JSON write is compared whole.
|
|
ContentSHA256 []byte
|
|
}
|
|
|
|
// ClaimStale is how long a claim may be outstanding before another attempt may take it over; without
|
|
// it a dead process locks the key until the retention sweep.
|
|
//
|
|
// ⚠ Must exceed the longest write here — an upload, bounded by its route's read deadline. Enforced
|
|
// at boot (config.Load): under a longer deadline a retry takes the claim from a LIVE attempt and the
|
|
// user gets the second book the header exists to prevent.
|
|
const ClaimStale = 30 * time.Minute
|
|
|
|
// ClaimToken says WHICH attempt holds a claim. The key identifies the request; this identifies the
|
|
// try, and without it every writer addressed the row by identity alone — so a late attempt released
|
|
// its successor's claim, and a superseded one wrote its own receipt over the row.
|
|
type ClaimToken string
|
|
|
|
// NewClaimToken mints one. It is an owner marker rather than a secret: what matters is that two
|
|
// attempts never share one.
|
|
func NewClaimToken() ClaimToken { return ClaimToken(newID("clm")) }
|
|
|
|
// errKeyVanished is the row disappearing between an attempt's insert and its re-read. Not an answer
|
|
// to anybody: the key is simply free again, and the claim is taken from the top.
|
|
var errKeyVanished = errors.New("pgstore: the idempotency key was released mid-claim")
|
|
|
|
// claimRounds bounds that retry. Two attempts of one key can hand it back and forth at most as fast
|
|
// as they can fail, and a caller that loses three rounds is meeting something other than a race.
|
|
const claimRounds = 3
|
|
|
|
// ClaimIdempotency takes the right to perform a request once.
|
|
//
|
|
// It answers one of four things, and the four are the whole semantics: a token to go ahead with, a
|
|
// stored response to replay, ErrKeyInFlight while the first attempt is still running, ErrKeyReused
|
|
// when the key was used for a different request. The token is what the caller must present to
|
|
// complete or release: it is the proof that the claim is still this attempt's.
|
|
func (s *Store) ClaimIdempotency(ctx context.Context, k IdempotencyKey, now time.Time) (*IdempotentResponse, ClaimToken, error) {
|
|
// ⚠ Retried, because losing a race is not one of the four answers. Two attempts arrive together;
|
|
// one wins the insert and then FAILS FAST — a 4xx gives the key straight back — and the loser's
|
|
// re-read, on a fresh READ COMMITTED snapshot, finds nothing at all. Answered rather than retried
|
|
// that was a 500 on precisely the case this header exists for, which is the same 500 the
|
|
// `do nothing` below was written to close. Reproduced on a live database.
|
|
for range claimRounds {
|
|
out, token, err := s.claim(ctx, k, now)
|
|
if errors.Is(err, errKeyVanished) {
|
|
continue
|
|
}
|
|
return out, token, err
|
|
}
|
|
return nil, "", fmt.Errorf("pgstore: the idempotency key was claimed and released under us %d times", claimRounds)
|
|
}
|
|
|
|
func (s *Store) claim(ctx context.Context, k IdempotencyKey, now time.Time) (*IdempotentResponse, ClaimToken, error) {
|
|
var out *IdempotentResponse
|
|
token := NewClaimToken()
|
|
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
|
// The generated layer bound to the CALLER'S transaction. It must not open one of its own:
|
|
// the read below takes `for update` and the write that follows has to be under that lock.
|
|
q := s.q.WithTx(tx)
|
|
id := ReadIdempotencyKeyParams{UserID: k.UserID, Method: k.Method, PathSha256: k.pathHash(), Key: k.Key}
|
|
row, err := q.ReadIdempotencyKey(ctx, id)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
// ⚠ `for update` on a missing row locks nothing, so two simultaneous first attempts both
|
|
// land here. `do nothing` plus a re-read decides between them; a bare insert made the loser
|
|
// raise a unique violation — a 500 on the one case this header exists for.
|
|
inserted, err := q.InsertIdempotencyClaim(ctx, InsertIdempotencyClaimParams{
|
|
UserID: k.UserID,
|
|
Method: k.Method,
|
|
Path: k.Path,
|
|
PathSha256: k.pathHash(),
|
|
Key: k.Key,
|
|
RequestSha256: k.Fingerprint,
|
|
ClaimedAt: now,
|
|
ClaimToken: string(token),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: claim idempotency key: %w", err)
|
|
}
|
|
if inserted == 1 {
|
|
return nil
|
|
}
|
|
if row, err = q.ReadIdempotencyKey(ctx, id); err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return errKeyVanished // the winner released it; the key is free and this attempt starts over
|
|
}
|
|
return fmt.Errorf("pgstore: re-read idempotency key: %w", err)
|
|
}
|
|
} else if err != nil {
|
|
return fmt.Errorf("pgstore: read idempotency key: %w", err)
|
|
}
|
|
if !bytes.Equal(row.RequestSha256, k.Fingerprint) {
|
|
return ErrKeyReused
|
|
}
|
|
if row.FinishedAt != nil && row.Status != nil {
|
|
out = &IdempotentResponse{
|
|
Status: int(*row.Status),
|
|
Location: row.Location,
|
|
Body: row.Body,
|
|
ContentSHA256: row.ContentSha256,
|
|
}
|
|
return nil
|
|
}
|
|
if now.Sub(row.ClaimedAt) < ClaimStale {
|
|
return ErrKeyInFlight
|
|
}
|
|
// The first attempt is gone: its process died, or it answered an error and released nothing.
|
|
// Taking the claim over is what keeps a retry from meeting `key_in_flight` forever, and the NEW
|
|
// token is what stops the old attempt — which may still be alive and merely slow — from
|
|
// completing or releasing what is no longer its claim.
|
|
if err := q.RetakeIdempotencyClaim(ctx, RetakeIdempotencyClaimParams{
|
|
UserID: k.UserID,
|
|
Method: k.Method,
|
|
PathSha256: k.pathHash(),
|
|
Key: k.Key,
|
|
ClaimedAt: now,
|
|
RequestSha256: k.Fingerprint,
|
|
ClaimToken: string(token),
|
|
}); err != nil {
|
|
return fmt.Errorf("pgstore: re-claim idempotency key: %w", err)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
if out != nil {
|
|
return out, "", nil // a replay holds no claim
|
|
}
|
|
return nil, token, nil
|
|
}
|
|
|
|
// ErrClaimLost is a completion whose row is gone or already finished. The work IS done, but nothing
|
|
// will replay it — a fact worth reporting rather than losing in a nil.
|
|
var ErrClaimLost = errors.New("pgstore: this attempt no longer holds the idempotency key")
|
|
|
|
// CompleteIdempotency records what the first attempt answered, so a repeat can be given the same
|
|
// thing without doing the work again.
|
|
//
|
|
// Only the attempt that still HOLDS the claim may write the receipt. An attempt whose claim was
|
|
// taken over has had its work done by somebody else; letting it answer would replay a `Location` for
|
|
// a book that was never created.
|
|
func (s *Store) CompleteIdempotency(ctx context.Context, k IdempotencyKey, token ClaimToken, resp IdempotentResponse, now time.Time) error {
|
|
status := int32(resp.Status)
|
|
n, err := s.q.CompleteIdempotencyKey(ctx, CompleteIdempotencyKeyParams{
|
|
UserID: k.UserID,
|
|
Method: k.Method,
|
|
PathSha256: k.pathHash(),
|
|
Key: k.Key,
|
|
Status: &status,
|
|
Location: resp.Location,
|
|
Body: resp.Body,
|
|
FinishedAt: &now,
|
|
ContentSha256: resp.ContentSHA256,
|
|
ClaimToken: string(token),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: complete idempotency key: %w", err)
|
|
}
|
|
if n == 0 {
|
|
return ErrClaimLost
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ReleaseIdempotency gives a key back after an attempt that did NOT complete.
|
|
//
|
|
// A failed attempt is not a completed one, and the contract names the case that matters: a `408` is
|
|
// not a completed attempt, so the same key may be presented again and the retry is a repeat of the
|
|
// first call rather than a second book.
|
|
//
|
|
// On the token for the same reason as the completion, and this is the half that cost a duplicate
|
|
// book: a slow attempt that finally failed deleted the row of the successor which had taken the key
|
|
// over and was already doing the work.
|
|
func (s *Store) ReleaseIdempotency(ctx context.Context, k IdempotencyKey, token ClaimToken) error {
|
|
n, err := s.q.ReleaseIdempotencyKey(ctx, ReleaseIdempotencyKeyParams{
|
|
UserID: k.UserID,
|
|
Method: k.Method,
|
|
PathSha256: k.pathHash(),
|
|
Key: k.Key,
|
|
ClaimToken: string(token),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: release idempotency key: %w", err)
|
|
}
|
|
if n == 0 {
|
|
return ErrClaimLost
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// SweepIdempotency forgets records past the window the contract promises ("at least 24 hours").
|
|
func (s *Store) SweepIdempotency(ctx context.Context, before time.Time) (int64, error) {
|
|
n, err := s.q.SweepIdempotency(ctx, before)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("pgstore: sweep idempotency keys: %w", err)
|
|
}
|
|
return n, nil
|
|
}
|