api

package module
v0.7.4 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	ErrCodeDeadlineExceeded = "DEADLINE_EXCEEDED"
	ErrCodeCancelled        = "CANCELLED"
	ErrCodeGateDropped      = "GATE_DROPPED"
	ErrCodeGateError        = "GATE_ERROR"
	ErrCodeInferenceError   = "INFERENCE_ERROR"
	ErrCodeInvalidRequest   = "INVALID_REQUEST"
)

Error codes for non-HTTP failures surfaced in ResultMessage.ErrorCode. These are result-level codes describing why a request could not be completed.

View Source 
const LabelClassification = "classification"

Variables

This section is empty.

Functions

func RequestActiveTokenKey 

func RequestActiveTokenKey(requestID string) string

RequestActiveTokenKey returns the Redis key tracking the currently active request generation.

func RequestCancellationKey 

func RequestCancellationKey(requestID string) string

RequestCancellationKey returns the Redis key used for per-request cancellation markers.

Types

type CancellationChecker 

type CancellationChecker interface {
	IsCancelled(ctx context.Context, requestID, requestToken string) (bool, error)
}

CancellationChecker reports whether a specific request generation has been cancelled.

type ClientError 

type ClientError struct {
	ErrorCategory ErrorCategory
	Message       string
	RawError      error         // original error if available
	RetryAfter    time.Duration // server-specified retry delay from Retry-After header (0 means not set)
	StatusCode    int           // HTTP status code; 0 means no HTTP response was received
}

ClientError represents an inference client error with category and context.

func (*ClientError) Category 

func (e *ClientError) Category() ErrorCategory

func (*ClientError) Error 

func (e *ClientError) Error() string

func (*ClientError) Unwrap 

func (e *ClientError) Unwrap() error

type ErrorCategory 

type ErrorCategory string

ErrorCategory defines the category of an inference client error. 

const (
	ErrCategoryRateLimit  ErrorCategory = "RATE_LIMIT"   // retryable
	ErrCategoryServer     ErrorCategory = "SERVER_ERROR" // retryable
	ErrCategoryInvalidReq ErrorCategory = "INVALID_REQ"  // not retryable
	ErrCategoryAuth       ErrorCategory = "AUTH_ERROR"   // not retryable
	ErrCategoryParse      ErrorCategory = "PARSE_ERROR"  // not retryable
	ErrCategoryUnknown    ErrorCategory = "UNKNOWN"      // not retryable
)

func (ErrorCategory) Fatal 

func (c ErrorCategory) Fatal() bool

Fatal returns true if errors in this category should not be retried.

func (ErrorCategory) Sheddable 

func (c ErrorCategory) Sheddable() bool

Sheddable returns true if errors in this category represent rate limiting or load shedding.

type InferenceClient 

type InferenceClient interface {
	// SendRequest sends an inference request to the specified URL with the given headers and payload.
	//
	// On success (nil error), the returned *InferenceResponse is always non-nil.
	// On error, *InferenceResponse may still be non-nil if the upstream sent an HTTP
	// response (e.g. 4xx/5xx); callers should check resp.StatusCode > 0 to determine
	// whether an HTTP response was received.
	//
	// Errors should implement InferenceError to provide an ErrorCategory via Category().
	// ErrorCategory determines retry and shedding behavior through its Fatal() and Sheddable() methods:
	//   - ErrCategoryRateLimit:  retryable, sheddable (e.g. 429)
	//   - ErrCategoryServer:     retryable (e.g. 5xx)
	//   - ErrCategoryInvalidReq: not retryable (e.g. 4xx)
	//   - ErrCategoryAuth:       not retryable
	//   - ErrCategoryParse:      not retryable
	//   - ErrCategoryUnknown:    not retryable
	SendRequest(ctx context.Context, url string, headers map[string]string, payload []byte) (*InferenceResponse, error)
}

InferenceClient defines the interface for sending inference requests. This interface allows for pluggable implementations beyond the default HTTP client.

type InferenceError 

type InferenceError interface {
	error
	// Category returns the error category, which determines retry and shedding behavior.
	Category() ErrorCategory
}

InferenceError represents an error that occurred during inference request processing.

type InferenceResponse 

type InferenceResponse struct {
	StatusCode int
	Body       []byte
}

InferenceResponse holds the HTTP response from an upstream inference request. StatusCode is non-zero whenever an HTTP response was received (including 4xx/5xx). A zero StatusCode means no HTTP response was obtained (e.g. transport/network failure). Body contains the response body; it may be partial if a read error occurred.

type InternalRequest 

type InternalRequest struct {
	InternalRouting `json:"-"`
	PublicRequest   Request
	// IngestionTime records when the message was pulled from the broker into
	// the in-process pipeline. It is in-process only: the custom JSON
	// (un)marshaling does not persist it, so a re-enqueued (retried) message is
	// re-stamped on its next delivery. Zero when not set (e.g. drained messages).
	IngestionTime time.Time `json:"-"`
}

InternalRequest is the internal envelope: routing data plus a concrete Request. It is used on channels and for persistence; JSON uses a tagged envelope.

func NewInternalRequest 

func NewInternalRequest(routing InternalRouting, typedReq Request) *InternalRequest

NewInternalRequest returns an InternalRequest with a non-nil PublicRequest. routing fields may be zero; PublicRequest must be non-nil.

func (*InternalRequest) MarshalJSON 

func (r *InternalRequest) MarshalJSON() ([]byte, error)

MarshalJSON encodes InternalRequest as a tagged envelope so the concrete Request type round-trips.

func (*InternalRequest) UnmarshalJSON 

func (r *InternalRequest) UnmarshalJSON(b []byte) error

UnmarshalJSON decodes the tagged envelope.

type InternalRouting 

type InternalRouting struct {
	RetryCount             int    `json:"retry_count,omitempty"`
	QueueID                string `json:"queue_id,omitempty"`
	RequestToken           string `json:"request_token,omitempty"`
	RequestQueueName       string `json:"request_queue_name,omitempty"`
	ResultQueueName        string `json:"result_queue_name,omitempty"`
	TransportCorrelationID string `json:"transport_correlation_id,omitempty"`
	// Labels is the framework's per-message label set. Seeded by the
	// Flow at pull time from the originating channel's effective
	// policy read and mutate this map in place. Producer-controlled
	// per-message correlation data rides on body.Metadata, not Labels.
	Labels map[string]string `json:"labels,omitempty"`
}

InternalRouting holds the resolved, authoritative routing fields used by infrastructure (producers, workers, retry logic). These are not part of the caller-facing contract and should not be set by callers directly. Producers merge caller-supplied per-message overrides (e.g. RedisRequest.RequestQueueName) with producer defaults and write the final values here. All internal pipeline code reads routing exclusively from this struct rather than reaching back into the typed request.

func (*InternalRouting) GetClassification 

func (ir *InternalRouting) GetClassification() QuotaClassification

GetClassification retrieves the quota classification from the Labels map.

func (*InternalRouting) SetClassification 

func (ir *InternalRouting) SetClassification(c QuotaClassification)

SetClassification sets the quota classification inside the Labels map.

type PriorityTier 

type PriorityTier string
const (
	TierInteractive PriorityTier = "interactive"
	TierAsync       PriorityTier = "async"
	TierBatch       PriorityTier = "batch"
)

type PubSubRequest 

type PubSubRequest struct {
	RequestMessage
	PubSubID string `json:"pubsub_id,omitempty"`
}

PubSubRequest is the concrete Request implementation for GCP Pub/Sub flows. Optional PubSubID is merged into InternalRouting.TransportCorrelationID in producers.

type QuotaClassification 

type QuotaClassification string
const (
	ClassificationNone     QuotaClassification = ""
	ClassificationReserved QuotaClassification = "reserved"
	ClassificationOverflow QuotaClassification = "overflow"
)

type RedisRequest 

type RedisRequest struct {
	RequestMessage
	RequestQueueName string `json:"request_queue_name,omitempty"`
	ResultQueueName  string `json:"result_queue_name,omitempty"`
}

RedisRequest is the concrete Request implementation for Redis-based flows. Per-message queue fields here override producer defaults; producers merge them into InternalRouting on InternalRequest before enqueue.

type Request 

type Request interface {
	ReqID() string
	ReqCreated() int64
	ReqDeadline() int64
	ReqPayload() map[string]any
	ReqMetadata() map[string]string
	ReqHeaders() map[string]string
	ReqEndpoint() string
}

Request is the public interface for submitting requests to the async queue. It exposes only the caller-visible fields. Concrete types like RequestMessage, RedisRequest, and PubSubRequest satisfy this interface.

type RequestMessage 

type RequestMessage struct {
	ID       string            `json:"id"`
	Created  int64             `json:"created"`  // Unix seconds
	Deadline int64             `json:"deadline"` // Unix seconds
	Payload  map[string]any    `json:"payload"`
	Metadata map[string]string `json:"metadata,omitempty"`
	Headers  map[string]string `json:"headers,omitempty"`
	Endpoint string            `json:"endpoint,omitempty"`
}

RequestMessage contains the caller-visible fields of a request. Metadata is reserved for opaque, caller-supplied pass-through data (e.g. tracing IDs, user labels). The system does not read or write Metadata for its own routing or correlation. Request interface accessors use the Req prefix (e.g. ReqPayload) to avoid colliding with the struct's exported field names used for JSON serialization.

func (*RequestMessage) ReqCreated 

func (r *RequestMessage) ReqCreated() int64

func (*RequestMessage) ReqDeadline 

func (r *RequestMessage) ReqDeadline() int64

func (*RequestMessage) ReqEndpoint 

func (r *RequestMessage) ReqEndpoint() string

func (*RequestMessage) ReqHeaders 

func (r *RequestMessage) ReqHeaders() map[string]string

func (*RequestMessage) ReqID 

func (r *RequestMessage) ReqID() string

func (*RequestMessage) ReqMetadata 

func (r *RequestMessage) ReqMetadata() map[string]string

func (*RequestMessage) ReqPayload 

func (r *RequestMessage) ReqPayload() map[string]any

type ResultMessage 

type ResultMessage struct {
	ID           string            `json:"id"`
	StatusCode   int               `json:"status_code,omitempty"`
	Payload      string            `json:"payload"`
	ErrorCode    string            `json:"error_code,omitempty"`
	ErrorMessage string            `json:"error_message,omitempty"`
	Routing      InternalRouting   `json:"-"`
	Metadata     map[string]string `json:"-"`
}

ResultMessage is the async inference result returned to callers.

Wire-format semantics:

  • StatusCode > 0: an HTTP response was received. Payload contains the response body.
  • StatusCode == 0: no HTTP response. ErrorCode/ErrorMessage describe the failure.

Routing and Metadata are infrastructure pass-through (json:"-").

func NewCancelledResult 

func NewCancelledResult(req Request, routing InternalRouting) ResultMessage

NewCancelledResult builds a ResultMessage for a cancelled request.

func NewDeadlineExceededResult 

func NewDeadlineExceededResult(req Request, routing InternalRouting) ResultMessage

NewDeadlineExceededResult builds a ResultMessage for deadline expiry.

func NewErrorResult 

func NewErrorResult(req Request, routing InternalRouting, errorCode, errMsg string) ResultMessage

NewErrorResult builds a non-HTTP error ResultMessage. errorCode must be one of the ErrCode* constants; errMsg is a human-readable description. Payload is populated with a JSON error object for backward compatibility with consumers that only read Payload.

func NewGateDroppedResult 

func NewGateDroppedResult(req Request, routing InternalRouting) ResultMessage

NewGateDroppedResult builds a ResultMessage for a gate-dropped request.

func NewHTTPResult 

func NewHTTPResult(req Request, routing InternalRouting, statusCode int, responseBody []byte) ResultMessage

NewHTTPResult builds a ResultMessage for any HTTP response (success or error). statusCode is the actual HTTP status code; responseBody is the raw body.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.