Documentation
¶
Index ¶
- Constants
- type HandlerFunc
- type PersistedSemaphore
- type ProcessedItem
- type ProcessedStatus
- type QueueItem
- type QueueStatus
- type Registry
- type Scheduler
- type SchedulerConfig
- type SchedulerMeta
- type Semaphore
- type Store
- func (s *Store) ClaimBatch(ctx context.Context, limit int) ([]*QueueItem, error)
- func (s *Store) DB() *gorm.DB
- func (s *Store) Enqueue(ctx context.Context, taskType, payload string, priority, maxRuns, thread int) (*QueueItem, error)
- func (s *Store) GetSemaphore(ctx context.Context) bool
- func (s *Store) PendingCount(ctx context.Context) (int64, error)
- func (s *Store) RecentProcessed(ctx context.Context, n int) ([]*ProcessedItem, error)
- func (s *Store) RecordResult(ctx context.Context, item *QueueItem, startedAt time.Time, execErr error) error
- func (s *Store) RecoverStuck(ctx context.Context) (int64, error)
- func (s *Store) SetSemaphore(ctx context.Context, green bool) error
Constants ¶
const SemaphoreKey = "semaphore"
SemaphoreKey is the key used in scheduler_meta for the semaphore state.
const SemaphoreValueGreen = "green"
SemaphoreValueGreen is the stored value for a green semaphore.
const SemaphoreValueRed = "red"
SemaphoreValueRed is the stored value for a red semaphore.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type HandlerFunc ¶
HandlerFunc processes a single queue item. Return a non-nil error to mark the item as processed with error status.
type PersistedSemaphore ¶
type PersistedSemaphore struct {
// contains filtered or unexported fields
}
PersistedSemaphore implements Semaphore and persists every state change to the database via the Store, so the state survives application restarts.
func NewPersistedSemaphore ¶
func NewPersistedSemaphore(store *Store, startGreen bool) *PersistedSemaphore
NewPersistedSemaphore creates a PersistedSemaphore with an initial state.
func (*PersistedSemaphore) IsGreen ¶
func (ps *PersistedSemaphore) IsGreen() bool
IsGreen reports whether execution is currently allowed.
func (*PersistedSemaphore) SetGreen ¶
func (ps *PersistedSemaphore) SetGreen()
SetGreen allows task execution and persists the state.
func (*PersistedSemaphore) SetRed ¶
func (ps *PersistedSemaphore) SetRed()
SetRed blocks task execution and persists the state.
type ProcessedItem ¶
type ProcessedItem struct {
ID uint `gorm:"primaryKey;autoIncrement"`
QueueItemID uint `gorm:"not null;index"` // FK → queue_items.id (soft ref)
TaskType string `gorm:"not null;index"`
Payload string `gorm:"type:text"`
Status ProcessedStatus `gorm:"not null;index"`
ErrorMsg string `gorm:"type:text"` // empty when Status == "ok"
StartedAt time.Time `gorm:"not null"`
FinishedAt time.Time `gorm:"not null"`
DurationMs int64 `gorm:"not null"` // FinishedAt - StartedAt in ms
CreatedAt time.Time
}
ProcessedItem is an immutable record of a completed execution. Rows are never updated — only inserted.
func (ProcessedItem) TableName ¶
func (ProcessedItem) TableName() string
TableName overrides the GORM default.
type ProcessedStatus ¶
type ProcessedStatus string
ProcessedStatus is the final outcome recorded in processed_items.
const ( ProcessedOK ProcessedStatus = "ok" ProcessedError ProcessedStatus = "error" )
type QueueItem ¶
type QueueItem struct {
ID uint `gorm:"primaryKey;autoIncrement"`
TaskType string `gorm:"not null;index"` // e.g. "send_email", "sync_records"
Payload string `gorm:"type:text"` // JSON or plain string, interpreted by the handler
Status QueueStatus `gorm:"not null;default:'pending';index"`
Priority int `gorm:"not null;default:0;index"` // higher = runs first within a tick
MaxRuns int `gorm:"not null;default:1"` // 1 = run once, -1 = repeat indefinitely, N = run N times
Thread int `gorm:"not null;default:0"` // 0 = sequential, 1 = concurrent per type
CreatedAt time.Time
UpdatedAt time.Time
}
QueueItem is a task waiting to be executed. Any external program can insert a row into this table to enqueue work.
INSERT INTO queue_items (task_type, payload, priority, max_runs, thread) VALUES (?, ?, ?, ?, ?);
type QueueStatus ¶
type QueueStatus string
QueueStatus represents the lifecycle state of a queued item.
const ( StatusPending QueueStatus = "pending" StatusProcessing QueueStatus = "processing" // claimed by a scheduler tick )
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry maps task types to their handlers. Register all handlers before calling Scheduler.Start.
func (*Registry) Register ¶
func (r *Registry) Register(taskType string, h HandlerFunc)
Register associates a task type with a handler. Panics on duplicate registration to catch misconfiguration at startup.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler polls the database every interval and executes pending tasks.
func NewScheduler ¶
func NewScheduler(cfg SchedulerConfig, sem Semaphore, st *Store, reg *Registry) *Scheduler
NewScheduler creates a Scheduler.
type SchedulerConfig ¶
SchedulerConfig holds constructor options.
type SchedulerMeta ¶
type SchedulerMeta struct {
Key string `gorm:"primaryKey;size:64"`
Value string `gorm:"size:16;not null"`
}
SchedulerMeta stores key-value metadata for the scheduler, such as the semaphore state. This table persists across restarts so the scheduler can recover its previous state.
func (SchedulerMeta) TableName ¶
func (SchedulerMeta) TableName() string
TableName overrides the GORM default.
type Semaphore ¶
type Semaphore interface {
SetGreen()
SetRed()
IsGreen() bool
}
Semaphore is the interface for the scheduler kill switch. Green (true) means tasks can run; Red (false) means tasks are blocked.
func NewSemaphore ¶
NewSemaphore creates a Semaphore. Pass true to start green, false to start red.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store handles all database operations for the scheduler.
func (*Store) ClaimBatch ¶
ClaimBatch atomically claims up to `limit` pending tasks and marks them as "processing" so that concurrent scheduler instances never double-execute.
func (*Store) Enqueue ¶
func (s *Store) Enqueue(ctx context.Context, taskType, payload string, priority, maxRuns, thread int) (*QueueItem, error)
Enqueue inserts a new pending task. maxRuns controls how many times the task executes: 1 = once (default), -1 = repeat indefinitely. thread controls concurrency: 0 = sequential per type, 1 = concurrent per type. Any value of maxRuns < -1 is clamped to 1.
func (*Store) GetSemaphore ¶
GetSemaphore reads the persisted semaphore state from the database. Returns true if green, false if red or not found.
func (*Store) PendingCount ¶
PendingCount returns the number of tasks currently in pending or processing state.
func (*Store) RecentProcessed ¶
RecentProcessed returns the last `n` processed items ordered by newest first.
func (*Store) RecordResult ¶
func (s *Store) RecordResult(ctx context.Context, item *QueueItem, startedAt time.Time, execErr error) error
RecordResult writes a ProcessedItem, then either deletes the QueueItem (if MaxRuns is exhausted) or decrements MaxRuns and resets the status to pending for re-execution on the next tick. A maxRuns of -1 means repeat indefinitely.
func (*Store) RecoverStuck ¶
RecoverStuck resets any items stuck in "processing" state (e.g. after a crash) back to "pending" so they can be re-claimed on the next tick. Call this once at startup.
Source Files