queue

Documentation

Index

• type DB
  • func NewDB(configDir string) (*DB, error)
  • func (db *DB) ClaimTask(queueID, agentID string) (*Task, error)
  • func (db *DB) Close() error
  • func (db *DB) CompleteTask(taskID string, result json.RawMessage) error
  • func (db *DB) CreateQueue(queueID, name string, maxRetries, retentionDays int) error
  • func (db *DB) Enqueue(taskID, queueID string, spec json.RawMessage, maxRetries int) error
  • func (db *DB) FailTask(taskID, reason string) error
  • func (db *DB) GetAllQueueStats() ([]QueueStats, error)
  • func (db *DB) GetQueueID(name string) (string, error)
  • func (db *DB) GetQueueStats(queueID, queueName string) (QueueStats, error)
  • func (db *DB) GetTask(taskID string) (*Task, error)
  • func (db *DB) ListTasks(queueID, status string, limit int) ([]*Task, error)
• type Manager
  • func NewManager(db *DB, queueDefs []QueueDef) (*Manager, error)
  • func (m *Manager) ClaimTask(queueName, agentID string) (*Task, error)
  • func (m *Manager) CompleteTask(taskID string, result json.RawMessage) error
  • func (m *Manager) Enqueue(queueName string, taskSpec json.RawMessage) (string, error)
  • func (m *Manager) FailTask(taskID, reason string) error
  • func (m *Manager) GetAllStats() (QueueSnapshot, error)
  • func (m *Manager) GetQueueNames() []string
  • func (m *Manager) GetStats(queueName string) (QueueStats, error)
  • func (m *Manager) GetTask(taskID string) (*Task, error)
  • func (m *Manager) ListTasks(queueName, status string, limit int) ([]*Task, error)
• type QueueDef
• type QueueSnapshot
• type QueueStats
• type Task
• type TaskRun

Types

type DB

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

DB manages SQLite queue database

func NewDB

func NewDB(configDir string) (*DB, error)

NewDB opens or creates the queue database

func (*DB) ClaimTask

func (db *DB) ClaimTask(queueID, agentID string) (*Task, error)

ClaimTask atomically claims the next pending task from a queue

func (*DB) Close

func (db *DB) Close() error

Close closes the database connection

func (*DB) CompleteTask

func (db *DB) CompleteTask(taskID string, result json.RawMessage) error

CompleteTask marks a task as completed

func (*DB) CreateQueue

func (db *DB) CreateQueue(queueID, name string, maxRetries, retentionDays int) error

CreateQueue creates a new queue definition

func (*DB) Enqueue

func (db *DB) Enqueue(taskID, queueID string, spec json.RawMessage, maxRetries int) error

Enqueue adds a task to a queue

func (*DB) FailTask

func (db *DB) FailTask(taskID, reason string) error

FailTask handles task failure with retry logic

func (*DB) GetAllQueueStats

func (db *DB) GetAllQueueStats() ([]QueueStats, error)

GetAllQueueStats returns statistics for all queues

func (*DB) GetQueueID

func (db *DB) GetQueueID(name string) (string, error)

GetQueueID returns the ID for a queue by name, or empty string if not found

func (*DB) GetQueueStats

func (db *DB) GetQueueStats(queueID, queueName string) (QueueStats, error)

GetQueueStats returns statistics for a queue

func (*DB) GetTask

func (db *DB) GetTask(taskID string) (*Task, error)

GetTask fetches a task by ID

func (*DB) ListTasks

func (db *DB) ListTasks(queueID, status string, limit int) ([]*Task, error)

ListTasks lists tasks in a queue by status

type Manager

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

Manager provides queue operations

func NewManager

func NewManager(db *DB, queueDefs []QueueDef) (*Manager, error)

NewManager creates a new queue manager

func (*Manager) ClaimTask

func (m *Manager) ClaimTask(queueName, agentID string) (*Task, error)

ClaimTask claims the next pending task from a queue

func (*Manager) CompleteTask

func (m *Manager) CompleteTask(taskID string, result json.RawMessage) error

CompleteTask marks a task as completed

func (*Manager) Enqueue

func (m *Manager) Enqueue(queueName string, taskSpec json.RawMessage) (string, error)

Enqueue adds a task to a queue

func (*Manager) FailTask

func (m *Manager) FailTask(taskID, reason string) error

FailTask handles task failure with automatic retry or dead-letter

func (*Manager) GetAllStats

func (m *Manager) GetAllStats() (QueueSnapshot, error)

GetAllStats returns stats for all queues

func (*Manager) GetQueueNames

func (m *Manager) GetQueueNames() []string

GetQueueNames returns all queue names

func (*Manager) GetStats

func (m *Manager) GetStats(queueName string) (QueueStats, error)

GetStats returns stats for a queue

func (*Manager) GetTask

func (m *Manager) GetTask(taskID string) (*Task, error)

GetTask fetches a task by ID

func (*Manager) ListTasks

func (m *Manager) ListTasks(queueName, status string, limit int) ([]*Task, error)

ListTasks lists tasks in a queue

type QueueDef

type QueueDef struct {
	Name          string `toml:"name"`
	MaxRetries    int    `toml:"max_retries"`
	RetentionDays int    `toml:"retention_days"`
}

QueueDef represents a queue definition from config

type QueueSnapshot

type QueueSnapshot struct {
	Queues     []QueueStats `json:"queues"`
	TotalTasks int          `json:"total_tasks"`
	UpdatedAt  time.Time    `json:"updated_at"`
}

QueueSnapshot represents the state of all queues at a point in time

type QueueStats

type QueueStats struct {
	Name      string `json:"name"`
	Pending   int    `json:"pending"`
	Claimed   int    `json:"claimed"`
	Completed int    `json:"completed"`
	Failed    int    `json:"failed"`
	AvgTimeMs int64  `json:"avg_time_ms"`
}

QueueStats represents statistics for a queue

type Task

type Task struct {
	ID          string          `json:"id"`
	QueueID     string          `json:"queue_id"`
	QueueName   string          `json:"queue_name"`
	Spec        json.RawMessage `json:"spec"`
	Status      string          `json:"status"` // pending, claimed, completed, failed
	ClaimedBy   string          `json:"claimed_by,omitempty"`
	ClaimedAt   *time.Time      `json:"claimed_at,omitempty"`
	CompletedAt *time.Time      `json:"completed_at,omitempty"`
	RetryCount  int             `json:"retry_count"`
	MaxRetries  int             `json:"max_retries"`
	CreatedAt   time.Time       `json:"created_at"`
	UpdatedAt   time.Time       `json:"updated_at"`
}

Task represents a queued task

type TaskRun

type TaskRun struct {
	ID          string          `json:"id"`
	TaskID      string          `json:"task_id"`
	AgentID     string          `json:"agent_id,omitempty"`
	StartedAt   *time.Time      `json:"started_at,omitempty"`
	EndedAt     *time.Time      `json:"ended_at,omitempty"`
	Result      json.RawMessage `json:"result,omitempty"`
	Status      string          `json:"status"` // running, success, failed
	ErrorReason string          `json:"error_reason,omitempty"`
}

TaskRun represents an execution of a task