core

package

Go to main page

Versions in this module

v0

v0.4.1

Apr 29, 2026

v0.4.0

Mar 26, 2026

v0.3.0

Feb 17, 2026

Changes in this version

+ var ErrJobAlreadyExists = errors.New("job already exists")

+ type DefaultPanicHandler struct

+ func (h *DefaultPanicHandler) HandlePanic(ctx context.Context, runnerName string, workerID int, panicInfo any, ...)

+ type DefaultRejectedTaskHandler struct

+ func (h *DefaultRejectedTaskHandler) HandleRejectedTask(runnerName string, reason string)

+ type DurableJobStore interface

+ CreateJob func(ctx context.Context, job *JobEntity) error

type FIFOTaskQueue

+ func (q *FIFOTaskQueue) PushWithID(t Task, traits TaskTraits) TaskID

type MemoryJobStore

+ func (s *MemoryJobStore) CreateJob(ctx context.Context, job *JobEntity) error

+ type Metrics interface

+ RecordQueueDepth func(runnerName string, depth int)

+ RecordTaskDuration func(runnerName string, priority TaskPriority, duration time.Duration)

+ RecordTaskPanic func(runnerName string, panicInfo any)

+ RecordTaskRejected func(runnerName string, reason string)

+ type NilMetrics struct

+ func (m *NilMetrics) RecordQueueDepth(runnerName string, depth int)

+ func (m *NilMetrics) RecordTaskDuration(runnerName string, priority TaskPriority, duration time.Duration)

+ func (m *NilMetrics) RecordTaskPanic(runnerName string, panicInfo any)

+ func (m *NilMetrics) RecordTaskRejected(runnerName string, reason string)

+ type PanicHandler interface

+ HandlePanic func(ctx context.Context, runnerName string, workerID int, panicInfo any, ...)

+ type ParallelTaskRunner struct

+ func NewParallelTaskRunner(threadPool ThreadPool, maxConcurrency int) *ParallelTaskRunner

+ func (r *ParallelTaskRunner) FlushAsync(callback func())

+ func (r *ParallelTaskRunner) GetThreadPool() ThreadPool

+ func (r *ParallelTaskRunner) IsClosed() bool

+ func (r *ParallelTaskRunner) MaxConcurrency() int

+ func (r *ParallelTaskRunner) Metadata() map[string]any

+ func (r *ParallelTaskRunner) Name() string

+ func (r *ParallelTaskRunner) PendingTaskCount() int

+ func (r *ParallelTaskRunner) PostDelayedTask(task Task, delay time.Duration)

+ func (r *ParallelTaskRunner) PostDelayedTaskNamed(name string, task Task, delay time.Duration)

+ func (r *ParallelTaskRunner) PostDelayedTaskWithTraits(task Task, delay time.Duration, traits TaskTraits)

+ func (r *ParallelTaskRunner) PostDelayedTaskWithTraitsNamed(name string, task Task, delay time.Duration, traits TaskTraits)

+ func (r *ParallelTaskRunner) PostRepeatingTask(task Task, interval time.Duration) RepeatingTaskHandle

+ func (r *ParallelTaskRunner) PostRepeatingTaskWithInitialDelay(task Task, initialDelay, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ func (r *ParallelTaskRunner) PostRepeatingTaskWithTraits(task Task, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ func (r *ParallelTaskRunner) PostTask(task Task)

+ func (r *ParallelTaskRunner) PostTaskAndReply(task Task, reply Task, replyRunner TaskRunner)

+ func (r *ParallelTaskRunner) PostTaskAndReplyWithTraits(task Task, taskTraits TaskTraits, reply Task, replyTraits TaskTraits, ...)

+ func (r *ParallelTaskRunner) PostTaskNamed(name string, task Task)

+ func (r *ParallelTaskRunner) PostTaskWithTraits(task Task, traits TaskTraits)

+ func (r *ParallelTaskRunner) PostTaskWithTraitsNamed(name string, task Task, traits TaskTraits)

+ func (r *ParallelTaskRunner) RecentTasks(limit int) []TaskExecutionRecord

+ func (r *ParallelTaskRunner) RunningTaskCount() int

+ func (r *ParallelTaskRunner) SetMetadata(key string, value any)

+ func (r *ParallelTaskRunner) SetName(name string)

+ func (r *ParallelTaskRunner) Shutdown()

+ func (r *ParallelTaskRunner) Stats() RunnerStats

+ func (r *ParallelTaskRunner) WaitIdle(ctx context.Context) error

+ func (r *ParallelTaskRunner) WaitShutdown(ctx context.Context) error

+ type PoolStats struct

+ Active int

+ Delayed int

+ ID string

+ Queued int

+ Running bool

+ Workers int

type PriorityTaskQueue

+ func (q *PriorityTaskQueue) PushWithID(t Task, traits TaskTraits) TaskID

+ type RejectedTaskHandler interface

+ HandleRejectedTask func(runnerName string, reason string)

+ type RunnerStats struct

+ BarrierPending bool

+ Closed bool

+ LastTaskAt time.Time

+ LastTaskName string

+ Name string

+ Pending int

+ Rejected int64

+ Running int

+ Type string

type SequencedTaskRunner

+ func (r *SequencedTaskRunner) PendingTaskCount() int

+ func (r *SequencedTaskRunner) PostDelayedTaskNamed(name string, task Task, delay time.Duration)

+ func (r *SequencedTaskRunner) PostDelayedTaskWithTraitsNamed(name string, task Task, delay time.Duration, traits TaskTraits)

+ func (r *SequencedTaskRunner) PostTaskNamed(name string, task Task)

+ func (r *SequencedTaskRunner) PostTaskWithTraitsNamed(name string, task Task, traits TaskTraits)

+ func (r *SequencedTaskRunner) RecentTasks(limit int) []TaskExecutionRecord

+ func (r *SequencedTaskRunner) RunningTaskCount() int

+ func (r *SequencedTaskRunner) Stats() RunnerStats

type SingleThreadTaskRunner

+ func (r *SingleThreadTaskRunner) PendingTaskCount() int

+ func (r *SingleThreadTaskRunner) PostDelayedTaskNamed(name string, task Task, delay time.Duration)

+ func (r *SingleThreadTaskRunner) PostDelayedTaskWithTraitsNamed(name string, task Task, delay time.Duration, traits TaskTraits)

+ func (r *SingleThreadTaskRunner) PostTaskNamed(name string, task Task)

+ func (r *SingleThreadTaskRunner) PostTaskWithTraitsNamed(name string, task Task, traits TaskTraits)

+ func (r *SingleThreadTaskRunner) RecentTasks(limit int) []TaskExecutionRecord

+ func (r *SingleThreadTaskRunner) RunningTaskCount() int

+ func (r *SingleThreadTaskRunner) Stats() RunnerStats

+ type TaskExecutionRecord struct

+ Duration time.Duration

+ FinishedAt time.Time

+ Name string

+ Panicked bool

+ Priority TaskPriority

+ RunnerName string

+ RunnerType string

+ StartedAt time.Time

+ TaskID TaskID

+ type TaskID uuid.UUID

+ func GenerateTaskID() TaskID

+ func (id TaskID) IsZero() bool

+ func (id TaskID) String() string

type TaskItem

+ ID TaskID

type TaskQueue

+ PushWithID func(t Task, traits TaskTraits) TaskID

type TaskScheduler

+ func NewFIFOTaskSchedulerWithConfig(workerCount int, config *TaskSchedulerConfig) *TaskScheduler

+ func NewPriorityTaskSchedulerWithConfig(workerCount int, config *TaskSchedulerConfig) *TaskScheduler

+ func (s *TaskScheduler) GetMetrics() Metrics

+ func (s *TaskScheduler) GetPanicHandler() PanicHandler

+ type TaskSchedulerConfig struct

+ Metrics Metrics

+ PanicHandler PanicHandler

+ RejectedTaskHandler RejectedTaskHandler

+ func DefaultTaskSchedulerConfig() *TaskSchedulerConfig

v0.2.0

Dec 26, 2025

Changes in this version

+ func RegisterHandler[T any](m *JobManager, jobType string, handler TypedHandler[T]) error

+ type DefaultLogger struct

+ func NewDefaultLogger() *DefaultLogger

+ func (l *DefaultLogger) Debug(msg string, fields ...Field)

+ func (l *DefaultLogger) Error(msg string, fields ...Field)

+ func (l *DefaultLogger) Info(msg string, fields ...Field)

+ func (l *DefaultLogger) Warn(msg string, fields ...Field)

+ type ErrorHandler func(jobID string, operation string, err error)

+ type Field struct

+ Key string

+ Value interface{}

+ func F(key string, value interface{}) Field

+ type JSONSerializer struct

+ func NewJSONSerializer() *JSONSerializer

+ func (s *JSONSerializer) Deserialize(data []byte, target any) error

+ func (s *JSONSerializer) Name() string

+ func (s *JSONSerializer) Serialize(args any) ([]byte, error)

+ type JobEntity struct

+ ArgsData []byte

+ CreatedAt time.Time

+ ID string

+ Priority int

+ Result string

+ Status JobStatus

+ Type string

+ UpdatedAt time.Time

+ type JobFilter struct

+ Limit int

+ Offset int

+ Status JobStatus

+ Type string

+ type JobManager struct

+ func NewJobManager(controlRunner TaskRunner, ioRunner TaskRunner, executionRunner TaskRunner, ...) *JobManager

+ func (m *JobManager) CancelJob(id string) error

+ func (m *JobManager) GetActiveJobCount() int

+ func (m *JobManager) GetActiveJobs() []*JobEntity

+ func (m *JobManager) GetJob(ctx context.Context, id string) (*JobEntity, error)

+ func (m *JobManager) GetRetryPolicy() RetryPolicy

+ func (m *JobManager) ListJobs(ctx context.Context, filter JobFilter) ([]*JobEntity, error)

+ func (m *JobManager) SetErrorHandler(handler ErrorHandler)

+ func (m *JobManager) SetLogger(logger Logger)

+ func (m *JobManager) SetRetryPolicy(policy RetryPolicy)

+ func (m *JobManager) Shutdown(ctx context.Context) error

+ func (m *JobManager) Start(ctx context.Context) error

+ func (m *JobManager) SubmitDelayedJob(ctx context.Context, id string, jobType string, args any, delay time.Duration, ...) error

+ func (m *JobManager) SubmitJob(ctx context.Context, id string, jobType string, args any, traits TaskTraits) error

+ type JobSerializer interface

+ Deserialize func(data []byte, target any) error

+ Name func() string

+ Serialize func(args any) ([]byte, error)

+ type JobStatus string

+ const JobStatusCanceled

+ const JobStatusCompleted

+ const JobStatusFailed

+ const JobStatusPending

+ const JobStatusRunning

+ type JobStore interface

+ DeleteJob func(ctx context.Context, id string) error

+ GetJob func(ctx context.Context, id string) (*JobEntity, error)

+ GetRecoverableJobs func(ctx context.Context) ([]*JobEntity, error)

+ ListJobs func(ctx context.Context, filter JobFilter) ([]*JobEntity, error)

+ SaveJob func(ctx context.Context, job *JobEntity) error

+ UpdateStatus func(ctx context.Context, id string, status JobStatus, result string) error

+ type Logger interface

+ Debug func(msg string, fields ...Field)

+ Error func(msg string, fields ...Field)

+ Info func(msg string, fields ...Field)

+ Warn func(msg string, fields ...Field)

+ type MemoryJobStore struct

+ func NewMemoryJobStore() *MemoryJobStore

+ func (s *MemoryJobStore) Clear()

+ func (s *MemoryJobStore) Count() int

+ func (s *MemoryJobStore) DeleteJob(ctx context.Context, id string) error

+ func (s *MemoryJobStore) GetJob(ctx context.Context, id string) (*JobEntity, error)

+ func (s *MemoryJobStore) GetRecoverableJobs(ctx context.Context) ([]*JobEntity, error)

+ func (s *MemoryJobStore) ListJobs(ctx context.Context, filter JobFilter) ([]*JobEntity, error)

+ func (s *MemoryJobStore) SaveJob(ctx context.Context, job *JobEntity) error

+ func (s *MemoryJobStore) UpdateStatus(ctx context.Context, id string, status JobStatus, result string) error

+ type NoOpLogger struct

+ func NewNoOpLogger() *NoOpLogger

+ func (l *NoOpLogger) Debug(msg string, fields ...Field)

+ func (l *NoOpLogger) Error(msg string, fields ...Field)

+ func (l *NoOpLogger) Info(msg string, fields ...Field)

+ func (l *NoOpLogger) Warn(msg string, fields ...Field)

+ type QueuePolicy int

+ const QueuePolicyDrop

+ const QueuePolicyReject

+ const QueuePolicyWait

+ type RawJobHandler func(ctx context.Context, args []byte) error

+ type RejectionCallback func(task Task, traits TaskTraits)

+ type RetryPolicy struct

+ BackoffRatio float64

+ InitialDelay time.Duration

+ MaxDelay time.Duration

+ MaxRetries int

+ func DefaultRetryPolicy() RetryPolicy

+ func NoRetry() RetryPolicy

type SingleThreadTaskRunner

+ func (r *SingleThreadTaskRunner) GetQueuePolicy() QueuePolicy

+ func (r *SingleThreadTaskRunner) RejectedCount() int64

+ func (r *SingleThreadTaskRunner) SetQueuePolicy(policy QueuePolicy)

+ func (r *SingleThreadTaskRunner) SetRejectionCallback(callback RejectionCallback)

type TaskScheduler

+ func (s *TaskScheduler) ShutdownGraceful(timeout time.Duration) error

+ type TypedHandler func(ctx context.Context, args T) error

v0.1.0

Dec 14, 2025

Changes in this version

+ func PostDelayedTaskAndReplyWithResultAndTraits[T any](targetRunner TaskRunner, task TaskWithResult[T], delay time.Duration, ...)

+ func PostDelayedTaskAndReplyWithResult[T any](targetRunner TaskRunner, task TaskWithResult[T], delay time.Duration, ...)

+ func PostTaskAndReplyWithResultAndTraits[T any](targetRunner TaskRunner, task TaskWithResult[T], taskTraits TaskTraits, ...)

+ func PostTaskAndReplyWithResult[T any](targetRunner TaskRunner, task TaskWithResult[T], reply ReplyWithResult[T], ...)

+ type DelayManager struct

+ func NewDelayManager() *DelayManager

+ func (dm *DelayManager) AddDelayedTask(task Task, delay time.Duration, traits TaskTraits, target TaskRunner)

+ func (dm *DelayManager) Stop()

+ func (dm *DelayManager) TaskCount() int

+ type DelayedTask struct

+ RunAt time.Time

+ Target TaskRunner

+ Task Task

+ Traits TaskTraits

+ type DelayedTaskHeap []*DelayedTask

+ func (h *DelayedTaskHeap) Peek() *DelayedTask

+ func (h *DelayedTaskHeap) Pop() any

+ func (h *DelayedTaskHeap) Push(x any)

+ func (h DelayedTaskHeap) Len() int

+ func (h DelayedTaskHeap) Less(i, j int) bool

+ func (h DelayedTaskHeap) Swap(i, j int)

+ type FIFOTaskQueue struct

+ func NewFIFOTaskQueue() *FIFOTaskQueue

+ func (q *FIFOTaskQueue) Clear()

+ func (q *FIFOTaskQueue) IsEmpty() bool

+ func (q *FIFOTaskQueue) Len() int

+ func (q *FIFOTaskQueue) MaybeCompact()

+ func (q *FIFOTaskQueue) PeekTraits() (TaskTraits, bool)

+ func (q *FIFOTaskQueue) Pop() (TaskItem, bool)

+ func (q *FIFOTaskQueue) PopUpTo(max int) []TaskItem

+ func (q *FIFOTaskQueue) Push(t Task, traits TaskTraits)

+ type PriorityTaskQueue struct

+ func NewPriorityTaskQueue() *PriorityTaskQueue

+ func (q *PriorityTaskQueue) Clear()

+ func (q *PriorityTaskQueue) IsEmpty() bool

+ func (q *PriorityTaskQueue) Len() int

+ func (q *PriorityTaskQueue) MaybeCompact()

+ func (q *PriorityTaskQueue) PeekTraits() (TaskTraits, bool)

+ func (q *PriorityTaskQueue) Pop() (TaskItem, bool)

+ func (q *PriorityTaskQueue) PopUpTo(max int) []TaskItem

+ func (q *PriorityTaskQueue) Push(t Task, traits TaskTraits)

+ type RepeatingTaskHandle interface

+ IsStopped func() bool

+ Stop func()

+ type ReplyWithResult func(ctx context.Context, result T, err error)

+ type SequencedTaskRunner struct

+ func NewSequencedTaskRunner(threadPool ThreadPool) *SequencedTaskRunner

+ func (r *SequencedTaskRunner) FlushAsync(callback func())

+ func (r *SequencedTaskRunner) GetThreadPool() ThreadPool

+ func (r *SequencedTaskRunner) IsClosed() bool

+ func (r *SequencedTaskRunner) Metadata() map[string]any

+ func (r *SequencedTaskRunner) Name() string

+ func (r *SequencedTaskRunner) PostDelayedTask(task Task, delay time.Duration)

+ func (r *SequencedTaskRunner) PostDelayedTaskWithTraits(task Task, delay time.Duration, traits TaskTraits)

+ func (r *SequencedTaskRunner) PostRepeatingTask(task Task, interval time.Duration) RepeatingTaskHandle

+ func (r *SequencedTaskRunner) PostRepeatingTaskWithInitialDelay(task Task, initialDelay, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ func (r *SequencedTaskRunner) PostRepeatingTaskWithTraits(task Task, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ func (r *SequencedTaskRunner) PostTask(task Task)

+ func (r *SequencedTaskRunner) PostTaskAndReply(task Task, reply Task, replyRunner TaskRunner)

+ func (r *SequencedTaskRunner) PostTaskAndReplyWithTraits(task Task, taskTraits TaskTraits, reply Task, replyTraits TaskTraits, ...)

+ func (r *SequencedTaskRunner) PostTaskWithTraits(task Task, traits TaskTraits)

+ func (r *SequencedTaskRunner) SetMetadata(key string, value interface{})

+ func (r *SequencedTaskRunner) SetName(name string)

+ func (r *SequencedTaskRunner) Shutdown()

+ func (r *SequencedTaskRunner) WaitIdle(ctx context.Context) error

+ func (r *SequencedTaskRunner) WaitShutdown(ctx context.Context) error

+ type SingleThreadTaskRunner struct

+ func NewSingleThreadTaskRunner() *SingleThreadTaskRunner

+ func (r *SingleThreadTaskRunner) FlushAsync(callback func())

+ func (r *SingleThreadTaskRunner) GetThreadPool() ThreadPool

+ func (r *SingleThreadTaskRunner) IsClosed() bool

+ func (r *SingleThreadTaskRunner) Metadata() map[string]interface{}

+ func (r *SingleThreadTaskRunner) Name() string

+ func (r *SingleThreadTaskRunner) PostDelayedTask(task Task, delay time.Duration)

+ func (r *SingleThreadTaskRunner) PostDelayedTaskWithTraits(task Task, delay time.Duration, traits TaskTraits)

+ func (r *SingleThreadTaskRunner) PostRepeatingTask(task Task, interval time.Duration) RepeatingTaskHandle

+ func (r *SingleThreadTaskRunner) PostRepeatingTaskWithInitialDelay(task Task, initialDelay, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ func (r *SingleThreadTaskRunner) PostRepeatingTaskWithTraits(task Task, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ func (r *SingleThreadTaskRunner) PostTask(task Task)

+ func (r *SingleThreadTaskRunner) PostTaskAndReply(task Task, reply Task, replyRunner TaskRunner)

+ func (r *SingleThreadTaskRunner) PostTaskAndReplyWithTraits(task Task, taskTraits TaskTraits, reply Task, replyTraits TaskTraits, ...)

+ func (r *SingleThreadTaskRunner) PostTaskWithTraits(task Task, traits TaskTraits)

+ func (r *SingleThreadTaskRunner) SetMetadata(key string, value interface{})

+ func (r *SingleThreadTaskRunner) SetName(name string)

+ func (r *SingleThreadTaskRunner) Shutdown()

+ func (r *SingleThreadTaskRunner) Stop()

+ func (r *SingleThreadTaskRunner) WaitIdle(ctx context.Context) error

+ func (r *SingleThreadTaskRunner) WaitShutdown(ctx context.Context) error

+ type Task func(ctx context.Context)

+ type TaskItem struct

+ Task Task

+ Traits TaskTraits

+ type TaskPriority int

+ const TaskPriorityBestEffort

+ const TaskPriorityUserBlocking

+ const TaskPriorityUserVisible

+ type TaskQueue interface

+ Clear func()

+ IsEmpty func() bool

+ Len func() int

+ MaybeCompact func()

+ PeekTraits func() (TaskTraits, bool)

+ Pop func() (TaskItem, bool)

+ PopUpTo func(max int) []TaskItem

+ Push func(t Task, traits TaskTraits)

+ type TaskRunner interface

+ FlushAsync func(callback func())

+ GetThreadPool func() ThreadPool

+ IsClosed func() bool

+ Metadata func() map[string]any

+ Name func() string

+ PostDelayedTask func(task Task, delay time.Duration)

+ PostDelayedTaskWithTraits func(task Task, delay time.Duration, traits TaskTraits)

+ PostRepeatingTask func(task Task, interval time.Duration) RepeatingTaskHandle

+ PostRepeatingTaskWithInitialDelay func(task Task, initialDelay, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ PostRepeatingTaskWithTraits func(task Task, interval time.Duration, traits TaskTraits) RepeatingTaskHandle

+ PostTask func(task Task)

+ PostTaskAndReply func(task Task, reply Task, replyRunner TaskRunner)

+ PostTaskAndReplyWithTraits func(task Task, taskTraits TaskTraits, reply Task, replyTraits TaskTraits, ...)

+ PostTaskWithTraits func(task Task, traits TaskTraits)

+ Shutdown func()

+ WaitIdle func(ctx context.Context) error

+ WaitShutdown func(ctx context.Context) error

+ func GetCurrentTaskRunner(ctx context.Context) TaskRunner

+ type TaskScheduler struct

+ func NewFIFOTaskScheduler(workerCount int) *TaskScheduler

+ func NewPriorityTaskScheduler(workerCount int) *TaskScheduler

+ func (s *TaskScheduler) ActiveTaskCount() int

+ func (s *TaskScheduler) DelayedTaskCount() int

+ func (s *TaskScheduler) GetWork(stopCh <-chan struct{}) (Task, bool)

+ func (s *TaskScheduler) OnTaskEnd()

+ func (s *TaskScheduler) OnTaskStart()

+ func (s *TaskScheduler) PostDelayedInternal(task Task, delay time.Duration, traits TaskTraits, target TaskRunner)

+ func (s *TaskScheduler) PostInternal(task Task, traits TaskTraits)

+ func (s *TaskScheduler) QueuedTaskCount() int

+ func (s *TaskScheduler) Shutdown()

+ func (s *TaskScheduler) WorkerCount() int

+ type TaskTraits struct

+ Category string

+ MayBlock bool

+ Priority TaskPriority

+ func DefaultTaskTraits() TaskTraits

+ func TraitsBestEffort() TaskTraits

+ func TraitsUserBlocking() TaskTraits

+ func TraitsUserVisible() TaskTraits

+ type TaskWithResult func(ctx context.Context) (T, error)

+ type ThreadPool interface

+ ActiveTaskCount func() int

+ DelayedTaskCount func() int

+ ID func() string

+ IsRunning func() bool

+ PostDelayedInternal func(task Task, delay time.Duration, traits TaskTraits, target TaskRunner)

+ PostInternal func(task Task, traits TaskTraits)

+ QueuedTaskCount func() int

+ Start func(ctx context.Context)

+ Stop func()

+ WorkerCount func() int

+ type WorkSource interface

+ GetWork func(stopCh <-chan struct{}) (Task, bool)

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