Documentation
¶
Overview ¶
Package queue provides task queue implementations for the Vigolium HTTP server. It supports both disk-based (LevelDB) and Redis-based queues with a unified interface.
Index ¶
- Variables
- type Config
- type DiskQueue
- func (q *DiskQueue) Ack(taskID string) error
- func (q *DiskQueue) Close() error
- func (q *DiskQueue) Dequeue(ctx context.Context) (*ScanTask, error)
- func (q *DiskQueue) Enqueue(ctx context.Context, task *ScanTask) error
- func (q *DiskQueue) EnqueueBatch(ctx context.Context, tasks []*ScanTask) error
- func (q *DiskQueue) Metrics() *QueueMetrics
- func (q *DiskQueue) SaveMetadata() error
- type DiskQueueConfig
- type HybridQueue
- type HybridQueueConfig
- type MemoryQueue
- type Queue
- type QueueInputSource
- type QueueMetrics
- type QueueType
- type RedisQueue
- func (q *RedisQueue) Ack(taskID string) error
- func (q *RedisQueue) AckWithMessageID(ctx context.Context, messageID string) error
- func (q *RedisQueue) ClaimPendingTasks(ctx context.Context, minIdleTime time.Duration) ([]*ScanTask, error)
- func (q *RedisQueue) Close() error
- func (q *RedisQueue) Dequeue(ctx context.Context) (*ScanTask, error)
- func (q *RedisQueue) Enqueue(ctx context.Context, task *ScanTask) error
- func (q *RedisQueue) Metrics() *QueueMetrics
- type RedisQueueConfig
- type ScanTask
- type Segment
- func (s *Segment) AckTask(taskID string) error
- func (s *Segment) CachedDiskSize() int64
- func (s *Segment) CanDelete() bool
- func (s *Segment) Close() error
- func (s *Segment) CompletedTasks() int64
- func (s *Segment) Delete() error
- func (s *Segment) DiskSize() int64
- func (s *Segment) GetNextPending() (*ScanTask, error)
- func (s *Segment) ID() uint64
- func (s *Segment) IsClosed() bool
- func (s *Segment) IsSealed() bool
- func (s *Segment) Iterator() iterator.Iterator
- func (s *Segment) Path() string
- func (s *Segment) PendingTasks() int64
- func (s *Segment) Seal()
- func (s *Segment) TotalTasks() int64
- func (s *Segment) WriteTask(task *ScanTask) error
- func (s *Segment) WriteTasks(tasks []*ScanTask) error
- type SegmentConfig
- type TaskStatus
Constants ¶
This section is empty.
Variables ¶
var ( // ErrQueueClosed is returned when attempting to enqueue to a closed queue. ErrQueueClosed = errors.New("queue is closed") // ErrQueueFull is returned when the queue capacity is reached (for bounded queues). ErrQueueFull = errors.New("queue is full") // ErrInvalidTask is returned when a task is missing required fields. ErrInvalidTask = errors.New("invalid task: must have either URL or RawRequest") // ErrTaskNotFound is returned when a task ID is not found. ErrTaskNotFound = errors.New("task not found") // ErrSegmentSealed is returned when attempting to write to a sealed segment. ErrSegmentSealed = errors.New("segment is sealed") // ErrNoActiveSegment is returned when there is no active segment for writing. ErrNoActiveSegment = errors.New("no active segment") )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Type is the queue type: "disk", "redis", or "hybrid".
Type QueueType
// Disk queue options
DiskDir string
MaxPerSegment int
// Hybrid queue options
// MemBufferSize is the in-memory channel capacity for hybrid mode.
// Tasks beyond this spill to the disk/redis backend. Default: 10000.
MemBufferSize int
// Redis queue options
RedisAddr string
RedisPassword string
RedisDB int
RedisStreamName string
RedisGroupName string
RedisConsumer string
RedisBlockTimeout time.Duration
}
Config holds configuration for creating a queue.
type DiskQueue ¶
type DiskQueue struct {
// contains filtered or unexported fields
}
DiskQueue implements Queue using LevelDB with segment rotation. Tasks are persisted to disk and survive server restarts.
func NewDiskQueue ¶
func NewDiskQueue(cfg DiskQueueConfig) (*DiskQueue, error)
NewDiskQueue creates a new disk-based queue.
func (*DiskQueue) Ack ¶
Ack marks a task as completed. It first tries the segment recorded for this task at dequeue (O(1)); if that mapping is missing or stale, it falls back to scanning every segment.
func (*DiskQueue) Dequeue ¶
Dequeue retrieves the next pending task. Blocks until a task is available or context is cancelled. Uses event-driven wakeup from Enqueue with a fallback ticker for recovery (e.g., pre-existing tasks on disk after restart).
func (*DiskQueue) EnqueueBatch ¶
EnqueueBatch enqueues multiple tasks efficiently using batch writes.
func (*DiskQueue) Metrics ¶
func (q *DiskQueue) Metrics() *QueueMetrics
Metrics returns current queue statistics.
func (*DiskQueue) SaveMetadata ¶
SaveMetadata saves queue metadata to disk for recovery.
type DiskQueueConfig ¶
type DiskQueueConfig struct {
// BaseDir is the root directory for queue storage.
// Default: os.TempDir()/vigolium-queue
BaseDir string
// MaxRecordsPerSegment is the maximum tasks per segment before rotation.
// Default: 10000
MaxRecordsPerSegment int
}
DiskQueueConfig holds configuration for DiskQueue.
func DefaultDiskQueueConfig ¶
func DefaultDiskQueueConfig() DiskQueueConfig
DefaultDiskQueueConfig returns sensible defaults.
type HybridQueue ¶
type HybridQueue struct {
// contains filtered or unexported fields
}
HybridQueue combines an in-memory buffered channel with a persistent backend (disk/redis). Enqueue writes to the channel first (zero disk I/O); when the channel is full, tasks spill to the underlying queue. A background goroutine drains spillover back into the channel so consumers always read from one place.
func NewHybridQueue ¶
func NewHybridQueue(cfg HybridQueueConfig) (*HybridQueue, error)
NewHybridQueue creates a hybrid queue wrapping the given backend.
func (*HybridQueue) Ack ¶
func (q *HybridQueue) Ack(taskID string) error
Ack marks a task as completed. Tries the backend first (for spillover tasks), then treats as in-memory no-op.
func (*HybridQueue) Close ¶
func (q *HybridQueue) Close() error
Close releases resources. Stops the drainer, closes the backend, and drains the memory channel.
func (*HybridQueue) Dequeue ¶
func (q *HybridQueue) Dequeue(ctx context.Context) (*ScanTask, error)
Dequeue retrieves the next pending task. Reads from the unified memory channel, which is fed by both direct enqueue and the background drainer (for spillover tasks).
func (*HybridQueue) Enqueue ¶
func (q *HybridQueue) Enqueue(ctx context.Context, task *ScanTask) error
Enqueue adds a task to the queue. Fast path: writes to the in-memory channel with no disk I/O. Slow path: spills to the backend when the channel is full.
func (*HybridQueue) Metrics ¶
func (q *HybridQueue) Metrics() *QueueMetrics
Metrics returns current queue statistics.
type HybridQueueConfig ¶
type HybridQueueConfig struct {
// MemBufferSize is the capacity of the in-memory channel.
// Tasks beyond this count spill to the backend queue.
// Default: 10000
MemBufferSize int
// Backend is the persistent queue used for spillover.
// Typically a *DiskQueue.
Backend Queue
}
HybridQueueConfig holds configuration for HybridQueue.
type MemoryQueue ¶
type MemoryQueue struct {
// contains filtered or unexported fields
}
MemoryQueue is a pure in-memory FIFO queue with no disk or network backend.
It exists so a server that does not need durable task spillover (the common case: any server not running scan-on-receive) can still satisfy the Queue interface for diagnostics/metrics without opening a LevelDB directory, spawning drainer/cleanup goroutines, or taking the on-disk exclusive lock that would otherwise block a second server instance on the same host.
func NewMemoryQueue ¶
func NewMemoryQueue(buffer int) *MemoryQueue
NewMemoryQueue creates an empty in-memory queue. buffer <= 0 uses the default.
func (*MemoryQueue) Ack ¶
func (q *MemoryQueue) Ack(_ string) error
Ack marks a task completed. In-memory delivery is at-most-once, so this only updates metrics.
func (*MemoryQueue) Close ¶
func (q *MemoryQueue) Close() error
Close prevents new Enqueue calls; Dequeue drains remaining items then returns io.EOF. Safe to call multiple times.
func (*MemoryQueue) Dequeue ¶
func (q *MemoryQueue) Dequeue(ctx context.Context) (*ScanTask, error)
Dequeue returns the next task, blocking until one is available, the context is cancelled, or the queue is closed and drained (io.EOF).
func (*MemoryQueue) Enqueue ¶
func (q *MemoryQueue) Enqueue(ctx context.Context, task *ScanTask) error
Enqueue adds a task, blocking while the buffer is full until space frees up, the context is cancelled, or the queue is closed.
func (*MemoryQueue) Metrics ¶
func (q *MemoryQueue) Metrics() *QueueMetrics
Metrics reports the in-memory queue counters. Disk fields stay zero.
type Queue ¶
type Queue interface {
// Enqueue adds a task to the queue.
// Returns ctx.Err() if context is cancelled while waiting.
Enqueue(ctx context.Context, task *ScanTask) error
// Dequeue retrieves the next pending task from the queue.
// Blocks until a task is available or context is cancelled.
// Returns (nil, io.EOF) when queue is closed and drained.
// Returns (nil, ctx.Err()) if context is cancelled.
Dequeue(ctx context.Context) (*ScanTask, error)
// Ack marks a task as completed.
// For disk-based queue, this triggers segment cleanup check.
// For Redis queue, this is typically a no-op (ack on dequeue).
Ack(taskID string) error
// Close releases resources and prevents new Enqueue calls.
// Dequeue continues until queue is drained, then returns io.EOF.
Close() error
// Metrics returns current queue statistics.
// May return nil if metrics are not available.
Metrics() *QueueMetrics
}
Queue is the abstraction for task queue operations. Implementations must be safe for concurrent access from multiple goroutines.
type QueueInputSource ¶
type QueueInputSource struct {
// contains filtered or unexported fields
}
QueueInputSource adapts Queue to the InputSource interface. This allows the existing Executor to consume tasks from queue without modification.
func NewQueueInputSource ¶
func NewQueueInputSource(queue Queue) *QueueInputSource
NewQueueInputSource creates a new QueueInputSource.
func (*QueueInputSource) Next ¶
Next returns the next WorkItem from the queue with ack callback. Implements source.InputSource interface.
func (*QueueInputSource) Queue ¶
func (q *QueueInputSource) Queue() Queue
Queue returns the underlying queue.
type QueueMetrics ¶
type QueueMetrics struct {
// Depth is the current number of pending tasks.
// May be -1 if not available (e.g., for Redis without XPENDING call).
Depth int64
// TotalEnqueued is the total number of tasks added since start.
TotalEnqueued int64
// TotalDequeued is the total number of tasks retrieved since start.
TotalDequeued int64
// TotalCompleted is the total number of tasks acknowledged since start.
TotalCompleted int64
// EnqueueErrors is the number of failed enqueue operations.
EnqueueErrors int64
// DequeueErrors is the number of failed dequeue operations.
DequeueErrors int64
// SegmentCount is the number of active disk segments.
// Only set for disk-based and hybrid queues.
SegmentCount int
// DiskUsageBytes is the approximate total disk usage across all segments.
// Only set for disk-based and hybrid queues; 0 for Redis or in-memory.
DiskUsageBytes int64
}
QueueMetrics holds statistics about queue operations.
type QueueType ¶
type QueueType string
QueueType represents the type of queue implementation.
const ( // QueueTypeDisk is a disk-based queue using LevelDB with segment rotation. QueueTypeDisk QueueType = "disk" // QueueTypeRedis is a Redis-based queue using Redis Streams. QueueTypeRedis QueueType = "redis" // QueueTypeHybrid is an in-memory buffered queue that spills over to a disk/redis backend. // Fast path avoids disk I/O on enqueue; a background goroutine drains spillover back to memory. QueueTypeHybrid QueueType = "hybrid" // QueueTypeMemory is a pure in-memory queue with no disk/network backend and no // background goroutines — used where durable spillover isn't needed so startup // avoids a LevelDB directory and its exclusive on-disk lock. QueueTypeMemory QueueType = "memory" )
type RedisQueue ¶
type RedisQueue struct {
// contains filtered or unexported fields
}
RedisQueue implements Queue using Redis Streams.
func NewRedisQueue ¶
func NewRedisQueue(cfg RedisQueueConfig) (*RedisQueue, error)
NewRedisQueue creates a new Redis-based queue.
func (*RedisQueue) Ack ¶
func (q *RedisQueue) Ack(taskID string) error
Ack marks a task as completed and XACKs its Redis Stream message so it leaves the consumer group's pending-entries list. The message ID is resolved from the taskID→messageID map populated at Dequeue.
func (*RedisQueue) AckWithMessageID ¶
func (q *RedisQueue) AckWithMessageID(ctx context.Context, messageID string) error
AckWithMessageID acknowledges a specific message by its Redis Stream ID.
func (*RedisQueue) ClaimPendingTasks ¶
func (q *RedisQueue) ClaimPendingTasks(ctx context.Context, minIdleTime time.Duration) ([]*ScanTask, error)
ClaimPendingTasks claims and returns tasks that have been pending for too long. This is useful for handling crashed workers.
func (*RedisQueue) Dequeue ¶
func (q *RedisQueue) Dequeue(ctx context.Context) (*ScanTask, error)
Dequeue retrieves the next pending task from the Redis Stream.
func (*RedisQueue) Enqueue ¶
func (q *RedisQueue) Enqueue(ctx context.Context, task *ScanTask) error
Enqueue adds a task to the Redis Stream.
func (*RedisQueue) Metrics ¶
func (q *RedisQueue) Metrics() *QueueMetrics
Metrics returns current queue statistics.
type RedisQueueConfig ¶
type RedisQueueConfig struct {
// Addr is the Redis server address (e.g., "localhost:6379").
Addr string
// Password is the Redis password (empty for no auth).
Password string
// DB is the Redis database number.
DB int
// StreamName is the Redis Stream name for tasks.
// Default: "vigolium:tasks"
StreamName string
// ConsumerGroup is the consumer group name.
// Default: "vigolium-workers"
ConsumerGroup string
// ConsumerName is the unique name for this consumer.
// Default: auto-generated UUID
ConsumerName string
// BlockTimeout is how long to wait for new messages.
// Default: 5s
BlockTimeout time.Duration
// MaxRetries is the number of times to retry failed tasks.
// Default: 3
MaxRetries int
}
RedisQueueConfig holds configuration for the Redis-based queue.
type ScanTask ¶
type ScanTask struct {
// ID is the unique identifier for this task.
ID string `json:"id"`
// URL is the target URL to scan.
// Required if RawRequest is empty.
URL string `json:"url,omitempty"`
// RawRequest is the raw HTTP request string.
// Optional - if empty, a GET request will be generated from URL.
RawRequest string `json:"raw_request,omitempty"`
// EnableModules is the list of module IDs to run.
// Empty means use all default modules.
EnableModules []string `json:"enable_modules,omitempty"`
// WebhookURL is the URL to send results to.
// Optional - if empty, results go to default output only.
WebhookURL string `json:"webhook_url,omitempty"`
// Metadata contains arbitrary key-value pairs for tracking.
Metadata map[string]string `json:"metadata,omitempty"`
// Status is the current state of the task.
Status TaskStatus `json:"status"`
// CreatedAt is when the task was enqueued.
CreatedAt time.Time `json:"created_at"`
// UpdatedAt is when the task was last updated.
UpdatedAt time.Time `json:"updated_at,omitempty"`
}
ScanTask represents a scan request in the queue.
func NewScanTask ¶
NewScanTask creates a new ScanTask with a generated UUID and pending status.
func NewScanTaskFromRequest ¶
func NewScanTaskFromRequest(url, rawRequest, webhookURL string, enableModules []string, metadata map[string]string) *ScanTask
NewScanTaskFromRequest creates a ScanTask from an API request.
func (*ScanTask) MarkCompleted ¶
func (t *ScanTask) MarkCompleted()
MarkCompleted updates the task status to completed.
func (*ScanTask) MarkProcessing ¶
func (t *ScanTask) MarkProcessing()
MarkProcessing updates the task status to processing.
func (*ScanTask) MarshalJSON ¶
MarshalJSON implements json.Marshaler.
func (*ScanTask) UnmarshalJSON ¶
UnmarshalJSON implements json.Unmarshaler.
type Segment ¶
type Segment struct {
// contains filtered or unexported fields
}
func NewSegment ¶
func NewSegment(cfg SegmentConfig) (*Segment, error)
NewSegment creates a new segment with a LevelDB database.
func OpenSegment ¶
OpenSegment opens an existing segment from disk and recovers state.
func (*Segment) AckTask ¶
AckTask marks a task as completed by deleting its record.
The previous implementation rewrote the task with status=completed, leaving the (potentially large, RawRequest-bearing) record on disk until the entire 10k-task segment became deletable — so a segment with one stuck task pinned thousands of completed records. A completed record is never read again (GetNextPending consults only the pending index, cleared at dequeue; the Iterator method is unused), so we delete it outright and let LevelDB reclaim the space incrementally. recoverState rebuilds totalTasks from the surviving records, so deletion stays consistent with CanDelete (completed >= total) and PendingTasks (total - completed) across restarts.
func (*Segment) CachedDiskSize ¶
CachedDiskSize returns the approximate disk usage without filesystem access.
func (*Segment) CanDelete ¶
CanDelete returns true if all tasks are completed and segment can be deleted.
func (*Segment) CompletedTasks ¶
CompletedTasks returns the number of completed tasks.
func (*Segment) DiskSize ¶
DiskSize returns the approximate disk usage of this segment in bytes. Walks the segment directory and sums file sizes.
func (*Segment) GetNextPending ¶
GetNextPending retrieves and marks the next pending task as processing. Returns nil if no pending tasks exist.
func (*Segment) PendingTasks ¶
PendingTasks returns the number of pending tasks.
func (*Segment) Seal ¶
func (s *Segment) Seal()
Seal marks the segment as read-only and persists that state so it survives a restart (see keySealed).
func (*Segment) TotalTasks ¶
TotalTasks returns the total number of tasks in this segment.
func (*Segment) WriteTasks ¶
WriteTasks writes multiple tasks in a single LevelDB batch, amortizing the WAL flush across all tasks. This is significantly faster than individual WriteTask calls for bulk enqueue operations.
type SegmentConfig ¶
type SegmentConfig struct {
// BaseDir is the root directory for all segments.
BaseDir string
// ID is the unique segment identifier.
ID uint64
}
SegmentConfig holds configuration for segment creation.
type TaskStatus ¶
type TaskStatus string
TaskStatus represents the current state of a scan task.
const ( // TaskStatusPending indicates the task is waiting to be processed. TaskStatusPending TaskStatus = "pending" // TaskStatusProcessing indicates the task is currently being processed. TaskStatusProcessing TaskStatus = "processing" // TaskStatusCompleted indicates the task has been completed. TaskStatusCompleted TaskStatus = "completed" )
Source Files