async

package
v0.5.0 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const EventJobType = "event"
View Source 
const MemoryReconcileJobType = "memory.reconcile"
View Source 
const ResumeContinueJobType = "resume.continue"
View Source 
const RunJobType = "run"

Variables

View Source 
var (
	ErrJobNotFound = errors.New("async: job not found")
	ErrStaleLease  = errors.New("async: stale job lease")
	ErrInvalidJob  = errors.New("async: invalid job")
)
View Source 
var (
	ErrTenantMismatch = errors.New("async: tenant mismatch")
	ErrTenantRequired = errors.New("async: tenant identity required")
)

Functions

func AuthorizeTenant added in v0.4.5 

func AuthorizeTenant(ctx context.Context, job Job) error

AuthorizeTenant enforces ownership for tenant-scoped queue access. Worker and maintenance contexts without a principal remain global unless strict mode is explicitly enabled.

func CancelAuthorized added in v0.4.5 

func CancelAuthorized(ctx context.Context, queue Queue, jobID string) error

func RequeueAuthorized added in v0.4.5 

func RequeueAuthorized(ctx context.Context, queue Queue, admin JobAdmin, jobID string) error

func StampTenant added in v0.4.5 

func StampTenant(ctx context.Context, job *Job) error

StampTenant assigns the authenticated principal tenant to a newly enqueued job. A caller cannot overwrite an explicitly supplied, different tenant.

func TenantIDFromContext added in v0.4.5 

func TenantIDFromContext(ctx context.Context) string

TenantIDFromContext returns the authenticated principal's tenant, if any.

Types

type EventPayload added in v0.1.2 

type EventPayload struct {
	Type        string             `json:"type"`
	RunID       string             `json:"run_id,omitempty"`
	Payload     json.RawMessage    `json:"payload,omitempty"`
	MaxAttempts int                `json:"max_attempts,omitempty"`
	Metadata    map[string]string  `json:"metadata,omitempty"`
	Principal   identity.Principal `json:"principal,omitempty"`
}

func (EventPayload) Event added in v0.1.2 

func (payload EventPayload) Event() eventrouter.Event

type Handler 

type Handler interface {
	HandleJob(ctx context.Context, job Job) error
}

type HandlerFunc 

type HandlerFunc func(ctx context.Context, job Job) error

func (HandlerFunc) HandleJob 

func (fn HandlerFunc) HandleJob(ctx context.Context, job Job) error

type Job 

type Job struct {
	ID             string          `json:"id"`
	Type           string          `json:"type"`
	RunID          string          `json:"run_id,omitempty"`
	TenantID       string          `json:"tenant_id,omitempty"`
	Payload        json.RawMessage `json:"payload,omitempty"`
	State          JobState        `json:"state"`
	Attempts       int             `json:"attempts"`
	MaxAttempts    int             `json:"max_attempts"`
	LastError      string          `json:"last_error,omitempty"`
	CreatedAt      time.Time       `json:"created_at"`
	UpdatedAt      time.Time       `json:"updated_at"`
	AvailableAt    time.Time       `json:"available_at"`
	LeaseWorkerID  string          `json:"lease_worker_id,omitempty"`
	LeaseExpiresAt time.Time       `json:"lease_expires_at,omitempty"`
}

func CloneJob 

func CloneJob(job Job) Job

func LoadAuthorized added in v0.4.5 

func LoadAuthorized(ctx context.Context, queue Queue, jobID string) (Job, error)

func (Job) Validate 

func (job Job) Validate() error

type JobAdmin added in v0.1.4 

type JobAdmin interface {
	ListJobs(ctx context.Context, filter JobFilter) ([]Job, error)
	Requeue(ctx context.Context, jobID string) error
}

JobAdmin extends Queue with dead-letter inspection and requeue support.

type JobFilter added in v0.1.4 

type JobFilter struct {
	State    JobState
	TenantID string
	Limit    int
}

func ScopeJobFilter added in v0.4.5 

func ScopeJobFilter(ctx context.Context, filter JobFilter) (JobFilter, error)

ScopeJobFilter binds an admin listing to the authenticated tenant. Internal maintenance callers without a principal may still request an explicit tenant or leave it empty for a global view.

type JobState 

type JobState string
const (
	JobQueued     JobState = "queued"
	JobRunning    JobState = "running"
	JobCompleted  JobState = "completed"
	JobPaused     JobState = "paused"
	JobFailed     JobState = "failed"
	JobCancelled  JobState = "cancelled"
	JobDeadLetter JobState = "dead_letter"
)

type Lease 

type Lease struct {
	JobID     string
	WorkerID  string
	Attempt   int
	ExpiresAt time.Time
}

func (Lease) Validate 

func (lease Lease) Validate() error

type LeaseReleaser added in v0.3.1 

type LeaseReleaser interface {
	Release(ctx context.Context, lease Lease) error
}

LeaseReleaser is an optional Queue capability that returns a leased job to the queued state without counting a failure, so it becomes leaseable by any worker again. Workers use it as a best-effort fallback when a leased job can neither be loaded nor failed (e.g. the queue is mid-outage); queues without it simply let the lease expire, after which the job is leaseable again anyway.

type LeaseRenewer 

type LeaseRenewer interface {
	Renew(ctx context.Context, lease Lease, ttl time.Duration) (Lease, bool, error)
}

type MemoryReconcilePayload added in v0.1.9 

type MemoryReconcilePayload struct {
	MemoryName string             `json:"memory_name"`
	Agent      string             `json:"agent"`
	RunID      string             `json:"run_id,omitempty"`
	Principal  identity.Principal `json:"principal,omitempty"`
}

func (MemoryReconcilePayload) MarshalJSONBytes added in v0.1.9 

func (payload MemoryReconcilePayload) MarshalJSONBytes() (json.RawMessage, error)

type PauseResult added in v0.3.0 

type PauseResult struct {
	RunID string `json:"run_id"`
	Token string `json:"token"`
}

PauseResult carries metadata persisted when a job pauses for HITL.

type Queue 

type Queue interface {
	Enqueue(ctx context.Context, job Job) (Job, error)
	Lease(ctx context.Context, workerID string, ttl time.Duration) (Lease, bool, error)
	Load(ctx context.Context, jobID string) (Job, error)
	Complete(ctx context.Context, lease Lease) error
	Pause(ctx context.Context, lease Lease, result PauseResult) error
	Fail(ctx context.Context, lease Lease, cause error) error
	Cancel(ctx context.Context, jobID string) error
}

type QueueMetrics added in v0.1.4 

type QueueMetrics struct {
	Queued     int
	Running    int
	DeadLetter int
}

QueueMetrics summarizes queue depth by state.

func CollectQueueMetrics added in v0.1.4 

func CollectQueueMetrics(ctx context.Context, queue Queue) (QueueMetrics, error)

CollectQueueMetrics counts jobs by state when the queue implements JobAdmin.

type ResumeContinuePayload added in v0.1.2 

type ResumeContinuePayload struct {
	Token       string             `json:"token"`
	Decision    core.Decision      `json:"decision"`
	Amendment   json.RawMessage    `json:"amendment,omitempty"`
	RunID       string             `json:"run_id,omitempty"`
	MaxAttempts int                `json:"max_attempts,omitempty"`
	Metadata    map[string]string  `json:"metadata,omitempty"`
	Principal   identity.Principal `json:"principal,omitempty"`
}

type RunPausedError added in v0.3.0 

type RunPausedError struct {
	RunID string
	Token string
}

RunPausedError indicates a run job paused awaiting human approval. Workers should mark the job as paused rather than completed or failed.

func (RunPausedError) Error added in v0.3.0 

func (e RunPausedError) Error() string

type RunPayload 

type RunPayload struct {
	RunID       string             `json:"run_id,omitempty"`
	ScenarioID  string             `json:"scenario_id,omitempty"`
	Scenario    json.RawMessage    `json:"scenario,omitempty"`
	Agent       string             `json:"agent,omitempty"`
	Prompt      string             `json:"prompt,omitempty"`
	Context     json.RawMessage    `json:"context,omitempty"`
	MaxAttempts int                `json:"max_attempts,omitempty"`
	Metadata    map[string]string  `json:"metadata,omitempty"`
	Principal   identity.Principal `json:"principal,omitempty"`
}

type Worker 

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

func NewWorker 

func NewWorker(queue Queue, handler Handler, config WorkerConfig, opts ...WorkerOption) (*Worker, error)

func (*Worker) Run 

func (worker *Worker) Run(ctx context.Context) error

type WorkerConfig 

type WorkerConfig struct {
	WorkerID      string
	Concurrency   int
	LeaseTTL      time.Duration
	RenewInterval time.Duration
	JobTimeout    time.Duration
	PollInterval  time.Duration
}

type WorkerOption added in v0.3.1 

type WorkerOption func(*Worker)

func WithDrainTimeout added in v0.3.1 

func WithDrainTimeout(d time.Duration) WorkerOption

WithDrainTimeout enables graceful shutdown draining: when the worker context is cancelled the worker stops leasing new jobs, then waits up to d for in-flight jobs to finish before falling back to the cancel-and-requeue path. Lease renewal keeps running during the drain window so drained jobs do not lose their leases. The zero value (the default) preserves the previous behaviour of cancelling in-flight jobs immediately.

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.