Documentation
¶
Overview ¶
Package internalprocessors provides internal task processors for queue management including cleaning and healing operations.
Index ¶
- type CleanerTaskStorage
- type HealerTaskStorage
- type QueueCleaner
- func (q *QueueCleaner) CleanTasksQueue(ctx context.Context, taskType entity.TaskType) ([]*entity.Task, error)
- func (p QueueCleaner) Run(ctx context.Context)
- func (p QueueCleaner) SetProcessPeriod(processPeriod xvalue.Value[time.Duration])
- func (p QueueCleaner) SetProcessTimeout(timeout xvalue.Value[time.Duration])
- func (q *QueueCleaner) SetUpdatedAtTimeAgo(updatedAtTimeAgo xvalue.Value[time.Duration])
- func (p QueueCleaner) Stop()
- type QueueHealer
- func (q *QueueHealer) CureTasks(ctx context.Context, taskType entity.TaskType) ([]*entity.Task, error)
- func (p QueueHealer) Run(ctx context.Context)
- func (p QueueHealer) SetProcessPeriod(processPeriod xvalue.Value[time.Duration])
- func (p QueueHealer) SetProcessTimeout(timeout xvalue.Value[time.Duration])
- func (q *QueueHealer) SetUpdatedAtTimeAgo(updatedAtTimeAgo xvalue.Value[time.Duration])
- func (p QueueHealer) Stop()
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CleanerTaskStorage ¶
type CleanerTaskStorage interface {
DeleteTasks(ctx context.Context, taskType entity.TaskType, statuses []entity.TaskStatus, updatedAtTimeAgo time.Duration) ([]*entity.Task, error)
}
CleanerTaskStorage defines the storage interface required for the cleaner processor to delete old tasks.
type HealerTaskStorage ¶
type HealerTaskStorage interface {
CureTasks(ctx context.Context, taskType entity.TaskType, status entity.TaskStatus, updatedAtTimeAgo time.Duration, comment string) ([]*entity.Task, error)
}
HealerTaskStorage defines the storage interface required for the healer processor to cure stuck tasks.
type QueueCleaner ¶
type QueueCleaner struct {
// contains filtered or unexported fields
}
QueueCleaner removes old completed, canceled, or failed tasks from the queue.
func NewQueueCleaner ¶
func NewQueueCleaner(taskStorage CleanerTaskStorage, taskType entity.TaskType) *QueueCleaner
NewQueueCleaner creates a new queue cleaner with the specified storage and options.
func (*QueueCleaner) CleanTasksQueue ¶
func (q *QueueCleaner) CleanTasksQueue(ctx context.Context, taskType entity.TaskType) ([]*entity.Task, error)
CleanTasksQueue removes old tasks with done, canceled, or attempts_left status from the queue.
func (QueueCleaner) Run ¶
Run starts the processor, fetching and processing tasks until the context is canceled.
func (QueueCleaner) SetProcessPeriod ¶
SetProcessPeriod sets the interval between runs. The value is re-read on every tick, so a dynamic value takes effect after the current period elapses. Must be called before Run.
func (QueueCleaner) SetProcessTimeout ¶
SetProcessTimeout sets the timeout of a single run; it is re-read on every run. Must be called before Run.
func (*QueueCleaner) SetUpdatedAtTimeAgo ¶
func (q *QueueCleaner) SetUpdatedAtTimeAgo(updatedAtTimeAgo xvalue.Value[time.Duration])
SetUpdatedAtTimeAgo sets the time threshold for considering tasks as old enough to be cleaned. The value is re-read on every run. Must be called before Run.
type QueueHealer ¶
type QueueHealer struct {
// contains filtered or unexported fields
}
QueueHealer identifies and fixes tasks that have been stuck in pending or processing status for too long.
func NewQueueHealer ¶
func NewQueueHealer(taskStorage HealerTaskStorage, taskType entity.TaskType, processingTimeout func(ctx context.Context) time.Duration) *QueueHealer
NewQueueHealer creates a new queue healer. processingTimeout returns the current processing timeout of the task type and is read on every run.
func (*QueueHealer) CureTasks ¶
func (q *QueueHealer) CureTasks(ctx context.Context, taskType entity.TaskType) ([]*entity.Task, error)
CureTasks marks tasks stuck in pending or processing status as errored based on one threshold: the configured one, raised to the processing floor.
func (QueueHealer) Run ¶
Run starts the processor, fetching and processing tasks until the context is canceled.
func (QueueHealer) SetProcessPeriod ¶
SetProcessPeriod sets the interval between runs. The value is re-read on every tick, so a dynamic value takes effect after the current period elapses. Must be called before Run.
func (QueueHealer) SetProcessTimeout ¶
SetProcessTimeout sets the timeout of a single run; it is re-read on every run. Must be called before Run.
func (*QueueHealer) SetUpdatedAtTimeAgo ¶
func (q *QueueHealer) SetUpdatedAtTimeAgo(updatedAtTimeAgo xvalue.Value[time.Duration])
SetUpdatedAtTimeAgo sets the time threshold for considering a task as stuck. The value is re-read on every run. Must be called before Run.
Source Files