pipeline

package
v1.9.1 Latest Latest
Warning

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

Go to latest
Published: Oct 28, 2026 License: MIT Imports: 5 Imported by: 0

Documentation

Overview

Package pipeline defines the composable middleware pipeline that processes every request through an ordered chain of middleware.

Index

Constants

View Source
const (
	CodeUnauthenticated = "unauthenticated" // 401
	CodeForbidden       = "forbidden"       // 403
	CodeInvalidRequest  = "invalid_request" // 400
	CodeRateLimited     = "rate_limited"    // 429
	CodeQuotaExceeded   = "quota_exceeded"  // 429
	CodeBudgetExceeded  = "budget_exceeded" // 429
	CodeUnavailable     = "unavailable"     // 503
	CodeContentBlocked  = "content_blocked" // 400, a guard block
	CodeModelUnpriced   = "model_unpriced"  // 400, the unpriced-model policy
)

Refusal codes. Each maps to one HTTP status at the edge.

View Source
const (
	// StateProviderName is the name of the provider the terminal stage
	// called, set before the call so a failed call is still attributed.
	StateProviderName = "provider_name"
	// StateCacheHit is true when the response came from the response cache
	// or the stream cache.
	StateCacheHit = "cache_hit"
	// StateOriginalModel is the model name the caller asked for, set by the
	// alias stage when it rewrote the request to the alias's target.
	StateOriginalModel = "original_model"
	// StateAliasProvider is the provider the alias stage's chosen target
	// names, empty when the target names none. The provider call routes to
	// it when the request names no provider of its own.
	StateAliasProvider = "alias_target_provider"
	// StateAliasModel is the model the alias stage rewrote the request to.
	StateAliasModel = "alias_target_model"
)

Request.State keys more than one stage reads.

Variables

View Source
var (
	// ErrNoTerminal reports a pipeline with nothing that calls a provider.
	ErrNoTerminal = errors.New("nexus: pipeline has no terminal stage")
	// ErrManyTerminals reports a pipeline with more than one terminal stage.
	ErrManyTerminals = errors.New("nexus: pipeline has more than one terminal stage")
)

Functions

func CacheHit

func CacheHit(ctx context.Context) bool

CacheHit returns whether the response was served from cache.

func HTTPStatus

func HTTPStatus(err error) (status int, code string)

HTTPStatus maps err to the status and code an HTTP edge should answer with: a refusal's own, or 500 internal_error for anything else.

func IsPermanent

func IsPermanent(err error) bool

IsPermanent reports whether err, or anything it wraps, was marked Permanent.

func KeyID

func KeyID(ctx context.Context) string

KeyID returns the key ID from context.

func Permanent

func Permanent(err error) error

Permanent marks err as one that a retry cannot fix: no provider is registered, routing found no candidate. The retry stage gives up on it.

func ProviderName

func ProviderName(ctx context.Context) string

ProviderName returns the provider name from context.

func RequestID

func RequestID(ctx context.Context) string

RequestID returns the request ID from context.

func RetryAfter

func RetryAfter(err error) time.Duration

RetryAfter is how long the refusal in err says to wait, or 0.

func Scopes

func Scopes(ctx context.Context) ([]string, bool)

Scopes returns the authenticated key's scopes, and false when no edge set them (an in-process caller).

func StartTime

func StartTime(ctx context.Context) time.Time

StartTime returns the request start time from context.

func TenantID

func TenantID(ctx context.Context) string

TenantID returns the tenant ID from context.

func WithCacheHit

func WithCacheHit(ctx context.Context, hit bool) context.Context

WithCacheHit sets the cache hit flag in context.

func WithKeyID

func WithKeyID(ctx context.Context, id string) context.Context

WithKeyID sets the key ID in context.

func WithProviderName

func WithProviderName(ctx context.Context, name string) context.Context

WithProviderName sets the provider name in context.

func WithRequestID

func WithRequestID(ctx context.Context, id string) context.Context

WithRequestID sets the request ID in context.

func WithScopes

func WithScopes(ctx context.Context, scopes []string) context.Context

WithScopes records the scopes of the key the edge authenticated.

func WithStartTime

func WithStartTime(ctx context.Context, t time.Time) context.Context

WithStartTime sets the request start time in context.

func WithTenantID

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

WithTenantID sets the tenant ID in context.

Types

type Builder

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

Builder creates a pipeline from ordered middleware.

func NewBuilder

func NewBuilder() *Builder

NewBuilder creates a new pipeline builder.

func (*Builder) Build

func (b *Builder) Build() (Service, error)

Build creates a pipeline Service. Non-terminal middleware runs in priority order (lower = earlier, ties keep the order they were added); the one Terminal runs last.

func (*Builder) Use

func (b *Builder) Use(m ...Middleware) *Builder

Use adds middleware to the pipeline.

type Inspector

type Inspector interface {
	Stages() []Stage
}

Inspector is implemented by a pipeline that can list its stages in the order they run.

type Middleware

type Middleware interface {
	// Name returns a unique identifier for this middleware.
	Name() string

	// Priority returns execution order among non-terminal middleware
	// (lower = earlier = further out). The terminal always runs last.
	// Built-in bands:
	//   0-19:    request id, tracing, usage (they wrap everything)
	//   20-99:   timeout, identity, access, quota, stream lifecycle
	//   100-199: guardrails
	//   200-299: transform, alias, cache
	//   300-399: retry; custom middleware here runs once per attempt
	Priority() int

	// Process handles the request. Call next(ctx) to continue.
	Process(ctx context.Context, req *Request, next NextFunc) (*Response, error)
}

Middleware processes requests in the pipeline. Call next(ctx) to continue the chain, or return early to short-circuit.

type NextFunc

type NextFunc func(ctx context.Context) (*Response, error)

NextFunc calls the next middleware in the chain.

type Refusal

type Refusal interface {
	error
	RefusalCode() string
	StatusCode() int
}

Refusal is an error a stage returns when it refuses a request before any provider is called: authentication, scopes, quotas, budgets. The usage stage records it as refused, at exactly $0, under RefusalCode. The one exception is middlewares.QuotaError, which cuts a stream already served in part: it comes from the stream's Next, is recorded as an error under its code and status, and is priced from the tokens used.

type RefusalError

type RefusalError struct {
	Code         string
	Status       int
	Message      string
	Limit        string
	RetryAfter   time.Duration
	Unattributed bool
	Cause        error
}

RefusalError is the refusal every Nexus stage returns. Limit names the limit that refused the request ("60" requests a minute, "10.00" USD), when there is one. RetryAfter is how long the caller should wait before the limit can let it through, zero when waiting will not help. Unattributed asks the usage stage to record the request under no tenant and no key: for identities that disagree, and for a tenant that does not exist.

func (*RefusalError) Error

func (e *RefusalError) Error() string

func (*RefusalError) RefusalCode

func (e *RefusalError) RefusalCode() string

func (*RefusalError) StatusCode

func (e *RefusalError) StatusCode() int

func (*RefusalError) Unwrap

func (e *RefusalError) Unwrap() error

type Request

type Request struct {
	Completion *provider.CompletionRequest
	Embedding  *provider.EmbeddingRequest
	Type       RequestType // "completion", "stream", "embedding"

	// Mutable state middleware can read/write.
	State map[string]any
}

Request wraps the unified request with pipeline metadata.

type RequestType

type RequestType string

RequestType identifies the type of request being processed.

const (
	RequestCompletion RequestType = "completion"
	RequestStream     RequestType = "stream"
	RequestEmbedding  RequestType = "embedding"
)

type Response

type Response struct {
	Completion *provider.CompletionResponse
	Stream     provider.Stream
	Embedding  *provider.EmbeddingResponse
}

Response wraps the unified response.

type Service

type Service interface {
	// Execute processes a completion request through the full pipeline.
	Execute(ctx context.Context, req *provider.CompletionRequest) (*provider.CompletionResponse, error)

	// ExecuteStream processes a streaming request through the pipeline.
	ExecuteStream(ctx context.Context, req *provider.CompletionRequest) (provider.Stream, error)

	// ExecuteEmbedding processes an embedding request.
	ExecuteEmbedding(ctx context.Context, req *provider.EmbeddingRequest) (*provider.EmbeddingResponse, error)
}

Service processes requests through an ordered middleware chain.

type Stage

type Stage struct {
	Name     string `json:"name"`
	Priority int    `json:"priority"`
	Terminal bool   `json:"terminal"`
}

Stage describes one stage of a built pipeline.

type Terminal

type Terminal interface {
	Terminal()
}

Terminal marks the middleware that ends the chain by calling a provider. A pipeline has exactly one, and it always runs last whatever its priority, so a stage that wraps the call (usage, tracing, retry, custom middleware) can never be sorted behind it and silently skipped.

Directories

Path Synopsis
Package middlewares provides built-in pipeline middleware for Nexus.
Package middlewares provides built-in pipeline middleware for Nexus.

Jump to

Keyboard shortcuts

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