Method Claim

internal/idempotency/store.go:155–238

Claim atomically reserves (userID, key) for the caller. The outcome determines what the HTTP layer should do next: - OutcomeAcquired → run the side effect, then Complete on success or Release on caller-side abort. - OutcomeReplay → write Cached to the response, do NOT re-run the side effect. - Ou

(ctx context.Context, userID, key, path, bodyHash string)

// lock and the loser sees a non-stale in_progress on its follow-up
// read.
func (s *Store) Claim(ctx context.Context, userID, key, path, bodyHash string) (ClaimResult, error) {
	if userID == "" {
		return ClaimResult{}, errors.New("idempotency: userID required")
	}
	if key == "" {
		return ClaimResult{}, errors.New("idempotency: key required")
	}

	staleSecs := int(StaleClaimWindow.Seconds())

	// Atomic claim path. Returns a row iff we own the slot — either as
	// a fresh INSERT, or as a stale-takeover where the DO UPDATE's
	// WHERE clause is true. Returns no rows when an existing row
	// blocks us (completed, or in_progress but not yet stale), in
	// which case we read the existing row to classify the outcome.
	var owned int
	err := s.pool.QueryRow(ctx,
		`INSERT INTO idempotency_keys (
		     user_id, key, request_path, request_body_hash,
		     response_status, response_content_type, response_body,
		     status, created_at, completed_at
		 )
		 VALUES ($1, $2, $3, $4, 0, '', ''::bytea, 'in_progress', now(), NULL)
		 ON CONFLICT (user_id, key) DO UPDATE
		    SET request_path          = EXCLUDED.request_path,
		        request_body_hash     = EXCLUDED.request_body_hash,
		        response_status       = 0,
		        response_content_type = '',
		        response_body         = ''::bytea,
		        status                = 'in_progress',
		        created_at            = now(),
		        completed_at          = NULL
		  WHERE idempotency_keys.status = 'in_progress'
		    AND idempotency_keys.created_at < now() - make_interval(secs => $5)
		 RETURNING 1`,
		userID, key, path, bodyHash, staleSecs,
	).Scan(&owned)
	if err == nil {
		return ClaimResult{Outcome: OutcomeAcquired}, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return ClaimResult{}, err
	}

	// Lost the race. Read the existing row to classify.
	var (
		gotStatus string
		gotHash   string
		gotCode   int
		gotCT     string
		gotBody   []byte
	)
	err = s.pool.QueryRow(ctx,
		`SELECT status, request_body_hash, response_status,
		        response_content_type, response_body
		   FROM idempotency_keys
		  WHERE user_id = $1 AND key = $2`,
		userID, key,