middleware

package
v0.64.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 11, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Overview

idempotency.go — the DURABLE, exactly-once request-idempotency path (WS-043 / F048), the transactional upgrade of the best-effort in-memory DeduplicateUnary (F023). Where the memory store caches the response AFTER the handler returns (per-pod, TTL-only, non-atomic), the durable path claims an idempotency record INSIDE the handler's transaction and completes it in the same commit — so a committed effect always has a retrievable response, a retry that lands on another pod or after a restart replays the ORIGINAL response verbatim (including server-generated ids/etag), and a concurrent duplicate is a conflict rather than a second execution.

The store persists opaque response BYTES + the proto message name; this file owns the proto marshal/unmarshal and the registry lookup, so the persistence adapter (persistence/gormtx) stays a pure row store. The interceptor opens the outer persistence.TxRunner.Atomically; the generated CRUD handler's repository write NEST-JOINS it (GormTxRunner reuses an on-ctx tx), so claim + effect + completion commit as one unit.

Index

Constants

View Source
const (
	// StatusInProgress re-exports persistence.StatusInProgress.
	StatusInProgress = persistence.StatusInProgress
	// StatusCompleted re-exports persistence.StatusCompleted.
	StatusCompleted = persistence.StatusCompleted
)
View Source
const DefaultCellID = "default"

DefaultCellID is the cell identifier returned in outgoing headers when no specific cell is resolved from the tenant.

View Source
const DefaultIdempotencyTTL = 24 * time.Hour

DefaultIdempotencyTTL is the retention applied to a durable idempotency record when DurableDedup.TTL is zero. ~24h matches OCI/Stripe/AWS client-token retention.

View Source
const MaxRequestIDLen = 255

MaxRequestIDLen bounds a client-supplied request_id, matching the idempotency_keys.request_id column width. An over-length id is rejected before it reaches the store.

Variables

View Source
var (
	// ErrIdempotencyInProgress is returned when a duplicate arrives while the
	// original is still in flight — AlreadyExists (HTTP 409), never a re-execution.
	ErrIdempotencyInProgress = status.Error(codes.AlreadyExists, "a request with this request_id is already in progress")
	// ErrIdempotencyFingerprintMismatch is returned when a key is reused with a
	// different request body — InvalidArgument (HTTP 400).
	ErrIdempotencyFingerprintMismatch = status.Error(codes.InvalidArgument, "request_id reused with different request parameters")
	// ErrIdempotencyNonProtoResponse is returned when the durable path cannot
	// serialize the handler's response for replay — a loud Internal error rather
	// than silently degrading to non-durable behavior.
	ErrIdempotencyNonProtoResponse = status.Error(codes.Internal, "durable idempotency requires a protobuf response")
	// ErrIdempotencyRequestIDTooLong is returned when the client's request_id
	// exceeds MaxRequestIDLen — rejected up front as InvalidArgument rather than
	// letting an over-length value hit the store and surface a raw driver error.
	ErrIdempotencyRequestIDTooLong = status.Errorf(codes.InvalidArgument, "request_id exceeds %d characters", MaxRequestIDLen)
	// ErrIdempotencyResponseTooLarge is returned when a response exceeds
	// DurableDedup.MaxResponseBytes — a loud Internal error so the operator either
	// raises the cap or excludes the method.
	ErrIdempotencyResponseTooLarge = status.Error(codes.Internal, "idempotency response exceeds configured MaxResponseBytes")
)

Idempotency sentinel errors, mapped to the gRPC/HTTP codes AIP-155 / Stripe / OCI use for these cases.

Functions

func Apply

func Apply(msg proto.Message, paths []string)

Apply zeroes out fields of msg that are not listed in paths (AIP-157). An empty paths slice is a no-op — all fields are retained. Paths may use either proto field names (snake_case) or JSON names (camelCase). Unknown paths are silently ignored. Only top-level fields are considered; the function does not recurse into nested message fields.

func DeduplicateUnary

func DeduplicateUnary(store DeduplicationStore) grpc.UnaryServerInterceptor

DeduplicateUnary returns a gRPC unary interceptor that deduplicates requests by request_id, SCOPED to the verified tenant and the method (see [idempotencyCacheKey]). Requests without a request_id or with validate_only=true pass through without touching the cache. Handler errors are not cached.

This is the best-effort in-memory path: the cache is per-process and the store runs AFTER the handler returns, so it does not survive a crash/restart or a retry that lands on another pod, and it does not coalesce concurrent duplicates. For durable, exactly-once idempotency (a claim/complete inside the handler's transaction), use DurableDeduplicateUnary.

func DurableDeduplicateUnary added in v0.62.0

func DurableDeduplicateUnary(cfg DurableDedup) grpc.UnaryServerInterceptor

DurableDeduplicateUnary returns a gRPC unary interceptor providing durable, exactly-once idempotency keyed by (verified tenant, method, request_id).

Fast path (no transaction): a completed record replays the stored response and SKIPS the handler entirely. Slow path: open Atomically, claim the key, run the handler (its repository write nest-joins this transaction), then complete the record in the same commit — so the claim, the effect, and the stored response are one atomic unit. A handler error rolls the claim back with the effect (errors are never cached, as in F023). Requests without a request_id or with validate_only=true pass through untouched.

CONCURRENCY: a genuine concurrent duplicate does NOT get an immediate 409 — its claim INSERT blocks on the winner's uncommitted row and then, once the winner commits, replays the winner's response (coalesced, exactly-once). The ErrIdempotencyInProgress (409) result is reserved for observing an already COMMITTED in_progress record — the reserve→remote→complete (saga) pattern — which the single-transaction handler path never leaves behind.

The key is scoped to the VERIFIED tenant only, not the caller's subject: any caller in the tenant that presents a completed request_id replays its response, so request_ids MUST be high-entropy (AIP-155 UUIDv4) — a guessable id lets a tenant peer replay a response and bypass per-resource checks the handler runs internally (method-level authz still applies). With no verified principal the tenant is empty and all such callers share one scope; use this behind an Authenticator.

func DurableReserveUnary added in v0.63.0

func DurableReserveUnary(cfg DurableDedup) grpc.UnaryServerInterceptor

DurableReserveUnary returns a gRPC unary interceptor for the reserve→remote→ complete SAGA path — the durable idempotency variant for a handler whose effect is a REMOTE call rather than a local DB write. Unlike DurableDeduplicateUnary (which runs the handler INSIDE the claim transaction), it holds NO transaction across the handler:

  1. Reserve — claim + COMMIT an in_progress record in its own short transaction, then release it. 2. Remote effect — run the handler OUTSIDE any transaction; the handler performs the side effect and MUST pass the same request_id to the remote system so the remote is idempotent. 3. Complete — transition the record to completed with the stored response in a second short transaction.

It is selected via DurableDedup.Mode == DurableModeReserve. Fast-path replay, fingerprinting, TTL, the request_id cap, tenant scoping, and pass-through gates all behave exactly as the transactional path.

RETRY / FAILURE SEMANTICS (see spec 048 Increment 2, DB-4):

  • A duplicate that observes the committed in_progress reservation gets ErrIdempotencyInProgress (409) — the reservation is committed, so (unlike the transactional path) the conflict is immediate, not coalesced. A completed duplicate replays verbatim.
  • Handler (remote) error: the reservation is RELEASED (DurableIdempotencyStore.Abandon, best-effort) so an immediate retry re-executes — safe because the remote is idempotent by request_id. A release failure leaves the record to expire by TTL.
  • Handler succeeded but Complete failed (the "remote succeeded, record lost" gap): the reservation is LEFT in_progress and the error propagates. A duplicate within TTL gets 409; after TTL a retry re-executes and the remote dedups. Releasing here is deliberately avoided — it would invite a needless re-run of a succeeded remote effect and drop the 409 guard.

TTL: set it SHORTER than the transactional default if you want fast recovery from the Complete gap, but keep it LONGER than the maximum handler/remote latency. Abandon (the handler-error release) matches on status=in_progress, not on the claim instance, so a TTL shorter than the remote call lets a reservation expire mid-flight, be reclaimed by a retry, and then be deleted by the original's late error — forcing a harmless (remote-idempotent) re-drive. Keeping TTL > max latency prevents that.

func ErrorMapperUnary

func ErrorMapperUnary() grpc.UnaryServerInterceptor

ErrorMapperUnary returns a gRPC unary interceptor that maps well-known persistence errors to canonical gRPC status codes with AIP-193 detail messages.

func FieldMaskUnary

func FieldMaskUnary(verbMap map[string]string) grpc.UnaryServerInterceptor

FieldMaskUnary returns a gRPC unary interceptor that validates UpdateMask on update-verb methods. verbMap maps FullMethod → verb string (e.g. "update"). For any method whose verb is "update", the request must implement GetUpdateMask() []string and the mask must be non-empty; otherwise codes.InvalidArgument is returned.

func FieldMaskUnarySource added in v0.18.0

func FieldMaskUnarySource(src func() map[string]string) grpc.UnaryServerInterceptor

FieldMaskUnarySource is FieldMaskUnary over a lazily-resolved verb map, so the interceptor sees verbs for rules contributed after construction (e.g. the server's accumulated AddRules set). src is consulted on each request.

func IdempotencyMethodFromContext added in v0.63.0

func IdempotencyMethodFromContext(ctx context.Context) (string, bool)

IdempotencyMethodFromContext returns the durable-idempotency method a durable interceptor stashed on ctx (and true), or ("", false). A routing TxRunner (e.g. servicekit's per-module host holder) reads it to bind the transaction to the same backend the store call targets — independent of the transport, so store routing (by the key method) and tx routing never diverge.

func InboundBearerFromContext added in v0.56.0

func InboundBearerFromContext(ctx context.Context) (string, bool)

InboundBearerFromContext returns the caller's raw inbound bearer and true, or "" and false when none is present (an unauthenticated/public path, or a stage that did not stash it). A delegation TokenSource fails closed on false: there is no verified identity to act on behalf of.

func IsSystemContext added in v0.55.0

func IsSystemContext(ctx context.Context) bool

IsSystemContext reports whether ctx was marked by WithSystemContext.

func LoggingUnary added in v0.24.0

func LoggingUnary(logger *slog.Logger) grpc.UnaryServerInterceptor

LoggingUnary returns a gRPC unary server interceptor that emits one structured slog record per RPC, trace-correlated and redaction-on-by-default.

At Info it logs a single summary line per call: method, grpc.code, duration_ms, request_id (from RequestIDUnary), and tenant (account-id, from TenantIDUnary). When a span is active it also attaches trace_id and span_id pulled from the OTel API span context (trace.SpanContextFromContext) — so logs correlate with the spans the otelgrpc/otelhttp stats handlers produce. The OTel API is in core; with no SDK installed the span context is simply invalid and the trace fields are omitted (no overhead, no behavioral change).

At Debug it additionally logs the request and (on success) the response, each first run through redact.Message so (infoblox.field.v1.opts).secret = true fields are replaced with "[REDACTED]" before they reach any log sink. Payload logging is therefore off at Info (summaries only) and redacted wherever it is on — redaction is on by default, never opt-in.

A nil logger falls back to slog.Default(). Place this interceptor after RequestIDUnary/TenantIDUnary (so request_id and tenant are in context) and before the authz interceptor (so the record captures the final code, including PermissionDenied).

func PrincipalFromContext added in v0.32.0

func PrincipalFromContext(ctx context.Context) (authz.Principal, bool)

PrincipalFromContext returns the authorized principal stashed on ctx and true, or the zero principal and false when none is present (an unauthenticated or public path).

func ReadMaskUnary

func ReadMaskUnary() grpc.UnaryServerInterceptor

ReadMaskUnary returns a gRPC unary interceptor that applies a read_mask to proto responses (AIP-157). If the request implements GetReadMask() *fieldmaskpb.FieldMask and the mask has non-empty paths, the interceptor calls Apply on the response after the handler returns. Requests without GetReadMask(), nil masks, or empty path lists pass through unchanged. Handler errors are passed through without modification.

func RequestIDFromContext

func RequestIDFromContext(ctx context.Context) string

RequestIDFromContext returns the request-ID stored in ctx, or "" if absent.

func RequestIDUnary

func RequestIDUnary() grpc.UnaryServerInterceptor

RequestIDUnary returns a gRPC unary interceptor that propagates or generates a request-ID. It reads the "x-request-id" key from incoming metadata; if present and non-empty that value is used, otherwise a new UUID is generated. The ID is stored in context and echoed as an outgoing header.

func TenantIDFromContext

func TenantIDFromContext(ctx context.Context) string

TenantIDFromContext returns the tenant that scopes the current request. The VERIFIED PRINCIPAL is the authority: when an authenticated principal is on the context (put there by the authn/authz stage via WithPrincipal), its Tenant is returned. The "account-id" header (see TenantIDUnary) is a routing hint only and is used solely as a fallback on paths that never establish a principal (e.g. the event consumer, which injects the tenant explicitly via WithTenantID). Returns "" when no tenant is established — callers on a tenant-scoped resource must fail closed.

NOT AN AUTHORIZATION / CONFIDENTIALITY BOUNDARY ON ITS OWN (SEC-042-01). The header fallback is a client-SETTABLE value: when NO authn/authz stage runs to establish a verified principal, this returns whatever "account-id" the caller sent. That is correct for its purpose — routing/cell selection and injecting a trusted tenant on non-gRPC paths (the event consumer) — but it means TenantIDFromContext MUST NOT be the sole basis of a tenant/confidentiality fence unless the chain includes an Authenticator (or otherwise guarantees a verified principal). A generated repository relies on the authz interceptor running ahead of it; a bespoke consumer that fences tenant data without an Authenticator is exposed to a spoofed header and should use VerifiedTenantID for the confidentiality decision instead.

func TenantIDUnary

func TenantIDUnary() grpc.UnaryServerInterceptor

TenantIDUnary returns a gRPC unary interceptor that extracts the "account-id" key from incoming metadata and stores it in context as a ROUTING/CELLS hint. It also sets a "cell-id: default" outgoing header. The stashed value does NOT override the verified principal: TenantIDFromContext returns the principal's Tenant when one is present and only falls back to this header otherwise, so a spoofed account-id cannot widen a request's tenant scope.

func ValidateOnlyFromContext

func ValidateOnlyFromContext(ctx context.Context) bool

ValidateOnlyFromContext returns the validate-only flag stored in ctx by ValidateOnlyUnary, or false if absent.

func ValidateOnlyUnary

func ValidateOnlyUnary() grpc.UnaryServerInterceptor

ValidateOnlyUnary returns a gRPC unary interceptor that stores true in context when the request implements GetValidateOnly() bool and returns true. The handler is always called.

func VerifiedTenantID added in v0.61.1

func VerifiedTenantID(ctx context.Context) (string, bool)

VerifiedTenantID returns the tenant of the VERIFIED principal on ctx and true, or "" and false when no verified principal is present. Unlike TenantIDFromContext it NEVER falls back to the client-settable "account-id" header, so it is the safe basis for a tenant/confidentiality decision: it is authoritative only when an authn/authz stage established the principal (via WithPrincipal). A caller enforcing a tenant fence WITHOUT relying on the generated repository's authz-gated scoping (e.g. a bespoke query path, a horizontal ility consumer, an export endpoint) should gate on this — a false return means "no verified tenant; fail closed" and must NOT be widened by the header. A principal present but carrying an empty Tenant returns "" and true: the identity is verified, the tenant is genuinely unset (fail closed on a tenant-scoped resource) — distinct from the false "no principal at all" case.

func WithInboundBearer added in v0.56.0

func WithInboundBearer(ctx context.Context, raw string) context.Context

WithInboundBearer returns a copy of ctx carrying the caller's raw inbound bearer token. The authentication interceptor calls it next to WithPrincipal after a successful verify, so a downstream TokenSource can act on behalf of the caller when calling another service.

func WithPrincipal added in v0.32.0

func WithPrincipal(ctx context.Context, p authz.Principal) context.Context

WithPrincipal returns a copy of ctx carrying the authorized principal. Called by the authz interceptor after a successful decision.

func WithResource added in v0.32.0

func WithResource(ctx context.Context, r ResourceRef) context.Context

WithResource returns a copy of ctx carrying the resource/verb the current operation targets. Called by the authz interceptor.

func WithSystemContext added in v0.55.0

func WithSystemContext(ctx context.Context) context.Context

WithSystemContext marks ctx as a trusted cross-tenant/system operation that BYPASSES the tenant fence (admin, migration, background jobs). It is the ONLY sanctioned way to run a tenant-scoped query/mutation without an established tenant — an absent tenant otherwise fails closed. NEVER derive it from client input.

func WithTenantID

func WithTenantID(ctx context.Context, tenantID string) context.Context

WithTenantID injects tenantID directly into ctx. Intended for tests and non-gRPC call paths that cannot go through TenantIDUnary (e.g. the event consumer). It sets the header-fallback value only; a verified principal on the context still takes precedence in TenantIDFromContext.

Types

type DeduplicationStore

type DeduplicationStore interface {
	Load(requestID string) (any, bool)
	Store(requestID string, response any)
}

DeduplicationStore is the backing store for DeduplicateUnary.

type DurableDedup added in v0.62.0

type DurableDedup struct {
	Store DurableIdempotencyStore
	Tx    persistence.TxRunner
	// TTL is the record retention; defaults to [DefaultIdempotencyTTL] (~24h).
	TTL time.Duration
	// DisableFingerprint turns OFF the param-fingerprint guard, which is ON by
	// default: with it on, reusing a request_id with a DIFFERENT request body is
	// rejected [ErrIdempotencyFingerprintMismatch] (Stripe-style). Turn it off only
	// when a stable per-request fingerprint is not desired.
	DisableFingerprint bool
	// MaxResponseBytes, when > 0, rejects a response whose marshaled size exceeds it
	// with [ErrIdempotencyResponseTooLarge] (fail loud) — a per-key storage bound.
	// Zero means unlimited (the default).
	MaxResponseBytes int
	// Mode selects the interceptor: DurableModeTransactional (default — claim/handler/
	// complete in one tx, for a LOCAL DB effect) or DurableModeReserve (reserve→remote→
	// complete, for a REMOTE effect). server.New reads it to pick the interceptor.
	Mode DurableDedupMode
}

DurableDedup groups the wiring for the durable idempotency path. Set it on server.Config to select DurableDeduplicateUnary over the best-effort memory path. Store and Tx are required.

PRECONDITIONS (exactly-once holds only when all are met):

  • The handler's domain write MUST participate in the transaction this interceptor opens — i.e. it writes through an SDK repository / TxRunner over the SAME backend as Store and Tx (the generated GORM/ent repositories do this by resolving their connection from persistence.TxFromContext). A handler that writes via a different *gorm.DB, a raw *sql.DB, or a Tx pointed at a different database commits its effect independently — then a rolled-back claim double-applies, or a committed claim replays a phantom success. This is NOT enforced at runtime; it is the caller's contract.
  • Use it for LOCAL, fast, DB-effect handlers. Because the handler runs inside the transaction, a handler that makes a slow remote call holds a DB connection and the claim row lock for its whole duration, and the remote effect is NOT covered by the rollback. Remote-effect handlers want the reserve→remote→ complete (committed in_progress) saga pattern instead.
  • The store's conflict handling assumes READ COMMITTED isolation (the default).

type DurableDedupMode added in v0.63.0

type DurableDedupMode int

DurableDedupMode selects which durable interceptor a DurableDedup configures.

const (
	// DurableModeTransactional (the default) claims, runs the handler, and completes
	// the record inside ONE transaction ([DurableDeduplicateUnary]) — exactly-once for
	// a LOCAL DB effect that nest-joins the same transaction.
	DurableModeTransactional DurableDedupMode = iota
	// DurableModeReserve reserves (claims + commits) an in_progress record, runs the
	// handler OUTSIDE any transaction (the REMOTE effect), then completes the record in
	// a second short transaction ([DurableReserveUnary]) — so no DB connection or claim
	// row lock is held across the remote call.
	DurableModeReserve
)

type DurableIdempotencyStore added in v0.62.0

type DurableIdempotencyStore interface {
	// Lookup returns the live (non-expired) record for key without a transaction.
	// ok is false when no live record exists.
	Lookup(ctx context.Context, key IdempotencyKey) (rec IdempotencyRecord, ok bool, err error)
	// Claim inserts an in_progress record for key inside the ctx transaction,
	// setting the given fingerprint and an expiry of now+ttl. On a fresh claim it
	// returns claimed=true. If a LIVE record already exists it returns that record
	// with claimed=false (the caller decides replay vs. conflict). An EXPIRED
	// conflicting record is reclaimed as a fresh in_progress claim.
	Claim(ctx context.Context, key IdempotencyKey, fingerprint string, ttl time.Duration) (existing IdempotencyRecord, claimed bool, err error)
	// Complete transitions key's in_progress record to completed with the response
	// bytes + proto type name, inside the ctx transaction.
	Complete(ctx context.Context, key IdempotencyKey, responseType string, response []byte) error
	// Abandon deletes key's in_progress reservation inside the ctx transaction,
	// releasing it so a retry re-executes. It is the release-on-error step of the
	// reserve→remote→complete saga path ([DurableReserveUnary]); the transactional
	// path never calls it (it rolls the claim back with the effect instead). It MUST
	// be guarded to status = in_progress so it can never erase a completed (durable)
	// response. It returns whether a row was deleted (false when the record was
	// already completed or gone).
	Abandon(ctx context.Context, key IdempotencyKey) (bool, error)
	// GC deletes records whose expiry is at or before now, returning the count
	// removed. Intended for a periodic sweep; correctness does not depend on it
	// (Lookup/Claim already treat expired records as absent).
	GC(ctx context.Context, now time.Time) (int64, error)
}

DurableIdempotencyStore is the durable backing store for DurableDeduplicateUnary. Claim and Complete MUST bind to the transaction on ctx (persistence.TxFromContext) so they commit atomically with the handler's effect; Lookup is a non-transactional fast-path read. Implementations live in a persistence adapter (e.g. persistence/gormtx.GormDurableDedupStore).

type IdempotencyKey added in v0.62.0

type IdempotencyKey = persistence.IdempotencyKey

IdempotencyKey aliases persistence.IdempotencyKey.

type IdempotencyRecord added in v0.62.0

type IdempotencyRecord = persistence.IdempotencyRecord

IdempotencyRecord aliases persistence.IdempotencyRecord.

type IdempotencyStatus added in v0.62.0

type IdempotencyStatus = persistence.IdempotencyStatus

IdempotencyStatus aliases persistence.IdempotencyStatus.

type MemoryDeduplicationStore

type MemoryDeduplicationStore struct {
	// contains filtered or unexported fields
}

MemoryDeduplicationStore is a thread-safe in-memory DeduplicationStore with per-entry TTL.

func NewMemoryDeduplicationStore

func NewMemoryDeduplicationStore(ttl time.Duration) *MemoryDeduplicationStore

NewMemoryDeduplicationStore returns a MemoryDeduplicationStore that expires entries after ttl.

func (*MemoryDeduplicationStore) Load

func (s *MemoryDeduplicationStore) Load(requestID string) (any, bool)

func (*MemoryDeduplicationStore) Store

func (s *MemoryDeduplicationStore) Store(requestID string, response any)

type ResourceRef added in v0.32.0

type ResourceRef struct {
	Type string
	Name string
	Verb string
}

ResourceRef identifies the resource an operation targets: its type, its concrete name/id, and the verb being performed.

Name is BEST-EFFORT: the authz stage resolves the resource TYPE and the VERB from the method's rule, but it does not (today) parse the concrete id out of the request, so Name is typically empty on the value the interceptor stashes. The authoritative resource identity for a change record comes from the entity itself (see events.ChangeEmitting), not from this ref.

func ResourceFromContext added in v0.32.0

func ResourceFromContext(ctx context.Context) (ResourceRef, bool)

ResourceFromContext returns the resource/verb the current operation targets and true, or the zero ref and false when none is present.

Directories

Path Synopsis
Package etag provides gRPC middleware for HTTP ETag / conditional-request semantics: it reads the If-Match precondition from incoming metadata and writes the ETag for the response to the outgoing trailer.
Package etag provides gRPC middleware for HTTP ETag / conditional-request semantics: it reads the If-Match precondition from incoming metadata and writes the ETag for the response to the outgoing trailer.
Package redact provides proto-reflection-based helpers that replace write-only field values with "[REDACTED]" (strings) or the zero value (other kinds) before they are logged or returned.
Package redact provides proto-reflection-based helpers that replace write-only field values with "[REDACTED]" (strings) or the zero value (other kinds) before they are logged or returned.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL