Documentation
¶
Index ¶
- Constants
- type QueueManager
- func (qm *QueueManager) ClearQueue(ctx context.Context, queueKey string) error
- func (qm *QueueManager) DequeueJobByID(ctx context.Context, jobID string) (*common.Job, error)
- func (qm *QueueManager) DequeueNextJob(ctx context.Context) (*common.Job, error)
- func (qm *QueueManager) EnqueuePendingJob(ctx context.Context, job *common.Job) error
- func (qm *QueueManager) GetFailedJobCount(ctx context.Context) (int64, error)
- func (qm *QueueManager) GetPendingJobCount(ctx context.Context) (int64, error)
- func (qm *QueueManager) GetPendingJobs(ctx context.Context, limit int) ([]*common.Job, error)
- func (qm *QueueManager) GetRunningJobCount(ctx context.Context) (int64, error)
- func (qm *QueueManager) MoveToFailed(ctx context.Context, job *common.Job, reason string) error
- func (qm *QueueManager) MoveToRunning(ctx context.Context, job *common.Job) error
- func (qm *QueueManager) RequeueFailed(ctx context.Context, job *common.Job) error
Constants ¶
const ( QueuePendingKey = "ares:queue:pending" QueueRunningKey = "ares:queue:running" QueueFailedKey = "ares:queue:failed" )
Queue names (keys in Redis)
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type QueueManager ¶
type QueueManager struct {
// contains filtered or unexported fields
}
QueueManager: Manages job queues with priority support Uses Redis sorted sets for efficient priority queue operations Complexity: O(log n) for enqueue, O(log n) for dequeue
func NewQueueManager ¶
func NewQueueManager(redisClient *redis.RedisClient) *QueueManager
NewQueueManager: Create a new queue manager
func (*QueueManager) ClearQueue ¶
func (qm *QueueManager) ClearQueue(ctx context.Context, queueKey string) error
ClearQueue: Clear all jobs from a queue (dangerous - use carefully)
func (*QueueManager) DequeueJobByID ¶
DequeueJobByID: Get a specific job by ID and remove it from pending queue Useful for: canceling a job, moving to running queue, etc.
func (*QueueManager) DequeueNextJob ¶
DequeueNextJob: Get the next job to schedule (highest priority, oldest within priority) Returns: (job, error) Atomic operation: removes from queue and returns job CORRECTED: ZPopMin returns map[string]float64, not slice
func (*QueueManager) EnqueuePendingJob ¶
EnqueuePendingJob: Add a job to the pending queue Score = -priority + timestamp/1e10 (sorts by priority desc, then by time asc) This ensures: high priority jobs first, then FIFO within same priority
func (*QueueManager) GetFailedJobCount ¶
func (qm *QueueManager) GetFailedJobCount(ctx context.Context) (int64, error)
GetFailedJobCount: Get number of jobs in failed queue
func (*QueueManager) GetPendingJobCount ¶
func (qm *QueueManager) GetPendingJobCount(ctx context.Context) (int64, error)
GetPendingJobCount: Get number of jobs in pending queue
func (*QueueManager) GetPendingJobs ¶
GetPendingJobs: Get top N pending jobs (for debugging)
func (*QueueManager) GetRunningJobCount ¶
func (qm *QueueManager) GetRunningJobCount(ctx context.Context) (int64, error)
GetRunningJobCount: Get number of jobs in running queue
func (*QueueManager) MoveToFailed ¶
MoveToFailed: Move a job from running to failed queue Call this when job execution fails
func (*QueueManager) MoveToRunning ¶
MoveToRunning: Move a job from pending to running queue Call this when job is scheduled and pod is created
func (*QueueManager) RequeueFailed ¶
RequeueFailed: Re-queue a failed job with exponential backoff Implements: delay = min(300s, 2^attempt * 1s)
Source Files