210 lines
10 KiB
Go
210 lines
10 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"errors"
|
|
"net/http"
|
|
"time"
|
|
|
|
"textmachine/platform/internal/books"
|
|
"textmachine/platform/internal/pgstore"
|
|
)
|
|
|
|
// idempotency.go: `Idempotency-Key` on the three writes that create something (canon
|
|
// §IdempotencyKey).
|
|
//
|
|
// The semantics are entirely the server's duty and are written out here rather than inferred:
|
|
// scoped to (principal, method, path) and the same key on another operation is ANOTHER key · the
|
|
// same key with the same request replays the ORIGINAL response and does no new work · the same key
|
|
// with a DIFFERENT request is `409 key_reused` · a repeat while the first is in flight is
|
|
// `409 key_in_flight` with `Retry-After` · the record is kept at least 24 hours · a key over 255
|
|
// characters is `400` · omitting the header is legal and means no retry protection.
|
|
|
|
// IdempotencyKeys is the storage this needs. An interface so the HTTP layer keeps knowing nothing
|
|
// about SQL, and so an instance without a database simply has none.
|
|
type IdempotencyKeys interface {
|
|
ClaimIdempotency(ctx context.Context, k pgstore.IdempotencyKey, now time.Time) (*pgstore.IdempotentResponse, pgstore.ClaimToken, error)
|
|
CompleteIdempotency(ctx context.Context, k pgstore.IdempotencyKey, token pgstore.ClaimToken, resp pgstore.IdempotentResponse, now time.Time) error
|
|
ReleaseIdempotency(ctx context.Context, k pgstore.IdempotencyKey, token pgstore.ClaimToken) error
|
|
}
|
|
|
|
// maxKeyLength is the contract's own bound; over it the request is refused rather than truncated.
|
|
const maxKeyLength = 255
|
|
|
|
// keyInFlightRetry is what a caller is told to wait when the first attempt under its key is still
|
|
// running. Deliberately short: the wait is for one request to finish, not for a translation.
|
|
const keyInFlightRetry = 2 * time.Second
|
|
|
|
// bodyIdentity says whether a request's identity depends on bytes its fingerprint cannot carry.
|
|
//
|
|
// A JSON write is compared whole, so the fingerprint IS the request. A multipart upload declares its
|
|
// parts and not its file, so two DIFFERENT books arrive under one key with the same declaration —
|
|
// and a replay decided on the declaration alone answers the second with the first one's book.
|
|
type bodyIdentity bool
|
|
|
|
const (
|
|
fingerprintIsTheRequest bodyIdentity = false
|
|
bodyDecidesIdentity bodyIdentity = true
|
|
)
|
|
|
|
// idempotent is one claimed key, or nothing at all when the caller sent no header. `active` is
|
|
// cleared by whichever of complete/release runs first, so a handler can DEFER the release and every
|
|
// exit settles the key — a panic into the recovery middleware included.
|
|
type idempotent struct {
|
|
h *v0
|
|
key pgstore.IdempotencyKey
|
|
// token is this ATTEMPT's claim. Presented on both writes, so an attempt whose claim was taken
|
|
// over can neither record its own receipt nor delete its successor's row.
|
|
token pgstore.ClaimToken
|
|
active bool
|
|
// replay is a stored answer this request may be given ONCE it has shown it is the same request:
|
|
// set only where the body decides identity, and then the caller owes the comparison.
|
|
replay *pgstore.IdempotentResponse
|
|
}
|
|
|
|
// beginIdempotent claims the key a write was presented with.
|
|
//
|
|
// The fingerprint is what makes "the same request" decidable, and what goes into it is the caller's
|
|
// choice because it differs by route: a JSON body is compared whole, while a multipart upload is
|
|
// compared over its DECLARED parts and never over the bytes — a server does not hold a book in
|
|
// memory to compare it, and a retry of an interrupted upload re-sends the same file.
|
|
//
|
|
// ok is false when the answer has already been written: a conflict, or a replay of the first
|
|
// attempt's own response.
|
|
func (h *v0) beginIdempotent(w http.ResponseWriter, r *http.Request, user string, fingerprint []byte,
|
|
identity bodyIdentity) (*idempotent, bool) {
|
|
raw := r.Header.Get("Idempotency-Key")
|
|
if raw == "" || h.keys == nil {
|
|
return &idempotent{}, true // legal, and it means no retry protection
|
|
}
|
|
if len(raw) > maxKeyLength {
|
|
Invalid(w, r, Item{Pointer: "/Idempotency-Key", Code: ItemTooLong})
|
|
return nil, false
|
|
}
|
|
// The REQUEST's path and not the route's pattern: the scope is (principal, method, path), and two
|
|
// books' runs live at two paths. Keyed by the pattern, one key presented for two books would read
|
|
// as the same key for a different request — a conflict where the client did nothing wrong.
|
|
sum := sha256.Sum256(fingerprint)
|
|
key := pgstore.IdempotencyKey{
|
|
UserID: user, Method: r.Method, Path: r.URL.Path, Key: raw, Fingerprint: sum[:],
|
|
}
|
|
replay, token, err := h.keys.ClaimIdempotency(r.Context(), key, time.Now())
|
|
switch {
|
|
case errors.Is(err, pgstore.ErrKeyReused):
|
|
FailCause(w, r, CodeIdempotencyConflict, CauseKeyReused)
|
|
return nil, false
|
|
case errors.Is(err, pgstore.ErrKeyContended):
|
|
// The claim loop gave up: this key was taken and handed back under us as fast as the racers
|
|
// could fail. The ANSWER is the same as an ordinary in-flight repeat, because the caller's
|
|
// remedy is the same one — wait and present it again — and it must not be a 500: nothing was
|
|
// wrong with the request and there is nothing its client could fix (PD-369). Ordered BEFORE
|
|
// the case below because it wraps it.
|
|
//
|
|
// Loud on OUR side and quiet on the wire, which is the whole mitigation for "the fix hides a
|
|
// storm": repeated contention on one key is an operator's signal, never a caller's fault.
|
|
h.log.WarnContext(r.Context(), "an idempotency key is under repeated contention", "err", err)
|
|
h.keyInFlight(w, r)
|
|
return nil, false
|
|
case errors.Is(err, pgstore.ErrKeyInFlight):
|
|
h.keyInFlight(w, r)
|
|
return nil, false
|
|
case err != nil:
|
|
h.log.ErrorContext(r.Context(), "the idempotency key could not be claimed", "err", err)
|
|
Fail(w, r, CodeInternalError)
|
|
return nil, false
|
|
}
|
|
if replay != nil {
|
|
if identity == bodyDecidesIdentity && len(replay.ContentSHA256) > 0 {
|
|
// NOT replayed here: what was compared is the DECLARATION, and this route's identity is the
|
|
// file. The caller reads the body and `replayIfIdentical` decides from the digest — the one
|
|
// answer that must not be given is the stored one to a request nobody has verified.
|
|
return &idempotent{h: h, key: key, replay: replay}, true
|
|
}
|
|
h.replay(w, r, replay)
|
|
return nil, false
|
|
}
|
|
return &idempotent{h: h, key: key, token: token, active: true}, true
|
|
}
|
|
|
|
// replay writes the first attempt's own answer, verbatim: same status, same body, same Location.
|
|
// Doing the work again is exactly what the header exists to prevent.
|
|
func (h *v0) replay(w http.ResponseWriter, r *http.Request, stored *pgstore.IdempotentResponse) {
|
|
if stored.Location != "" {
|
|
w.Header().Set("Location", stored.Location)
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(stored.Status)
|
|
if _, err := w.Write(stored.Body); err != nil {
|
|
h.log.DebugContext(r.Context(), "replayed body not delivered", "err", err)
|
|
}
|
|
}
|
|
|
|
// keyInFlight is the contract's answer to "this key is busy, come back": ONE writer, because the two
|
|
// paths that reach it — a first attempt still running, and a claim that lost the race too often —
|
|
// must not drift into two different words for one remedy.
|
|
func (h *v0) keyInFlight(w http.ResponseWriter, r *http.Request) {
|
|
WriteProblem(w, r, Problem{Code: CodeIdempotencyConflict,
|
|
Cause: &Cause{Code: CauseKeyInFlight}, RetryAfter: keyInFlightRetry})
|
|
}
|
|
|
|
// replayIfIdentical answers a repeat whose identity is its BODY: the stored answer when the bytes
|
|
// digest to what the first attempt accepted, and `key_reused` when they do not — which is what the
|
|
// canon calls a different request under the same key.
|
|
//
|
|
// The body is read either way, and that is the cost of an identity the declaration cannot carry. It
|
|
// buys the thing the declaration could not: a second, different book is never answered with the
|
|
// first one's address.
|
|
func (i *idempotent) replayIfIdentical(w http.ResponseWriter, r *http.Request, digest []byte) {
|
|
if !bytes.Equal(digest, i.replay.ContentSHA256) {
|
|
FailCause(w, r, CodeIdempotencyConflict, CauseKeyReused)
|
|
return
|
|
}
|
|
i.h.replay(w, r, i.replay)
|
|
}
|
|
|
|
// settleCtx detaches a key's own write from the request: the client that will retry is precisely
|
|
// the one that hung up, and on `r.Context()` the key then stays in flight for `ClaimStale` and the
|
|
// retry re-does the work. Same rule as the intake's writeCtx, applied to the receipt.
|
|
//
|
|
// The budget is books.ReceiptBudget and not a number of this package's own, because it is a TERM of
|
|
// what the boot compares against the windows an upload must finish inside (books.UploadSettle): the
|
|
// receipt is the last thing an upload does, and a copy of its budget here is a copy that can drift
|
|
// from the sum that has to contain it.
|
|
func settleCtx(ctx context.Context) (context.Context, context.CancelFunc) {
|
|
return context.WithTimeout(context.WithoutCancel(ctx), books.ReceiptBudget)
|
|
}
|
|
|
|
// complete records what this attempt answered, so a repeat is given the same thing. `content` is the
|
|
// digest of the body it accepted, for a route whose identity the fingerprint cannot settle; nil
|
|
// where it can.
|
|
func (i *idempotent) complete(ctx context.Context, status int, location string, body, content []byte) {
|
|
if !i.active {
|
|
return
|
|
}
|
|
i.active = false
|
|
ctx, cancel := settleCtx(ctx)
|
|
defer cancel()
|
|
resp := pgstore.IdempotentResponse{Status: status, Location: location, Body: body, ContentSHA256: content}
|
|
if err := i.h.keys.CompleteIdempotency(ctx, i.key, i.token, resp, time.Now()); err != nil {
|
|
// The work IS done and the caller has its answer; what is lost is the ability to replay it, so
|
|
// a retry does the work twice — or, on ErrClaimLost, a later attempt already did. Loud, and
|
|
// not a failure of the request.
|
|
i.h.log.ErrorContext(ctx, "the idempotency key could not be recorded", "err", err)
|
|
}
|
|
}
|
|
|
|
// release gives the key back after an attempt that did not complete: a failed attempt is not a
|
|
// completed one, and the same key may be presented again. Safe to defer — see idempotent.
|
|
func (i *idempotent) release(ctx context.Context) {
|
|
if !i.active {
|
|
return
|
|
}
|
|
i.active = false
|
|
ctx, cancel := settleCtx(ctx)
|
|
defer cancel()
|
|
if err := i.h.keys.ReleaseIdempotency(ctx, i.key, i.token); err != nil {
|
|
i.h.log.ErrorContext(ctx, "the idempotency key could not be released", "err", err)
|
|
}
|
|
}
|