Documentation
¶
Index ¶
- Variables
- func BatchSubmit(q Queue, urls []string, batchSize int) error
- func ExportQueue(q *InMemoryQueue) ([]byte, error)
- func ImportQueue(q *InMemoryQueue, data []byte) error
- func QueueStats(q Queue) map[string]any
- type Assignment
- type ClusterStatus
- type InMemoryQueue
- func (q *InMemoryQueue) Ack(ctx context.Context, task *Task) error
- func (q *InMemoryQueue) Close() error
- func (q *InMemoryQueue) Len() int
- func (q *InMemoryQueue) Nack(ctx context.Context, task *Task) error
- func (q *InMemoryQueue) Pop(ctx context.Context) (*Task, error)
- func (q *InMemoryQueue) Push(ctx context.Context, task *Task) error
- type Master
- func (m *Master) AssignTasks() []Assignment
- func (m *Master) CompleteTask(taskID string, result json.RawMessage)
- func (m *Master) GetClusterStatus() ClusterStatus
- func (m *Master) Heartbeat(nodeID string, stats NodeStats)
- func (m *Master) MonitorNodes(ctx context.Context, timeout time.Duration)
- func (m *Master) RegisterNode(node *Node)
- func (m *Master) SubmitTask(task *Task)
- func (m *Master) UnregisterNode(nodeID string)
- type MasterAPI
- type Node
- type NodeStats
- type NodeStatus
- type Queue
- type RedisQueue
- func (q *RedisQueue) Ack(ctx context.Context, task *Task) error
- func (q *RedisQueue) Close() error
- func (q *RedisQueue) InFlight() int
- func (q *RedisQueue) Len() int
- func (q *RedisQueue) Nack(ctx context.Context, task *Task) error
- func (q *RedisQueue) Pop(ctx context.Context) (*Task, error)
- func (q *RedisQueue) Push(ctx context.Context, task *Task) error
- type RedisQueueConfig
- type Role
- type Task
- type Worker
- type WorkerConfig
Constants ¶
This section is empty.
Variables ¶
var ErrQueueClosed = errors.New("queue closed")
ErrQueueClosed is returned by a closed queue.
Functions ¶
func BatchSubmit ¶
BatchSubmit submits multiple URLs as a single task to the queue.
func ExportQueue ¶
func ExportQueue(q *InMemoryQueue) ([]byte, error)
ExportQueue serializes all pending tasks as JSON.
func ImportQueue ¶
func ImportQueue(q *InMemoryQueue, data []byte) error
ImportQueue loads tasks from JSON into the queue.
func QueueStats ¶
QueueStats returns statistics about the queue.
Types ¶
type Assignment ¶
type Assignment struct {
TaskID string `json:"task_id"`
NodeID string `json:"node_id"`
Task *Task `json:"task"`
}
Assignment pairs a task with a worker node.
type ClusterStatus ¶
type ClusterStatus struct {
Nodes []*Node `json:"nodes"`
TotalNodes int `json:"total_nodes"`
PendingTasks int `json:"pending_tasks"`
RunningTasks int `json:"running_tasks"`
DoneTasks int `json:"done_tasks"`
}
ClusterStatus holds the overall cluster state.
type InMemoryQueue ¶
type InMemoryQueue struct {
// contains filtered or unexported fields
}
InMemoryQueue is an in-memory implementation of the distributed queue. Used for single-node operation or testing.
func NewInMemoryQueue ¶
func NewInMemoryQueue(logger *slog.Logger) *InMemoryQueue
NewInMemoryQueue creates a new in-memory queue.
func (*InMemoryQueue) Ack ¶
func (q *InMemoryQueue) Ack(ctx context.Context, task *Task) error
Ack is a no-op for the in-memory queue: there is no separate in-flight state to clear, and a process crash takes the queue with it either way.
func (*InMemoryQueue) Nack ¶
func (q *InMemoryQueue) Nack(ctx context.Context, task *Task) error
Nack returns the task to the queue.
type Master ¶
type Master struct {
// contains filtered or unexported fields
}
Master coordinates work distribution across worker nodes.
func (*Master) AssignTasks ¶
func (m *Master) AssignTasks() []Assignment
AssignTasks distributes pending tasks to available workers.
func (*Master) CompleteTask ¶
func (m *Master) CompleteTask(taskID string, result json.RawMessage)
CompleteTask marks a task as done.
func (*Master) GetClusterStatus ¶
func (m *Master) GetClusterStatus() ClusterStatus
GetClusterStatus returns the status of all nodes and tasks.
func (*Master) MonitorNodes ¶
MonitorNodes checks for offline nodes.
func (*Master) RegisterNode ¶
RegisterNode adds a worker node to the cluster.
func (*Master) SubmitTask ¶
SubmitTask adds a task to the queue.
func (*Master) UnregisterNode ¶
UnregisterNode removes a worker node.
type MasterAPI ¶
type MasterAPI struct {
// contains filtered or unexported fields
}
MasterAPI exposes the master coordinator as an HTTP API for workers.
func NewMasterAPI ¶
NewMasterAPI creates a new master HTTP API wrapper.
type Node ¶
type Node struct {
ID string `json:"id"`
Address string `json:"address"`
Role Role `json:"role"`
Status NodeStatus `json:"status"`
Capacity int `json:"capacity"` // Max concurrent workers
ActiveJobs int `json:"active_jobs"`
LastSeen time.Time `json:"last_seen"`
Stats NodeStats `json:"stats"`
}
Node represents a worker node in the distributed cluster.
type NodeStats ¶
type NodeStats struct {
RequestsSent int64 `json:"requests_sent"`
RequestsFailed int64 `json:"requests_failed"`
ItemsScraped int64 `json:"items_scraped"`
BytesDownloaded int64 `json:"bytes_downloaded"`
Uptime time.Duration `json:"uptime"`
}
NodeStats holds per-node statistics.
type NodeStatus ¶
type NodeStatus string
NodeStatus tracks a node's health.
const ( NodeReady NodeStatus = "ready" NodeBusy NodeStatus = "busy" NodeOffline NodeStatus = "offline" )
type Queue ¶
type Queue interface {
// Push adds a task to the queue.
Push(ctx context.Context, task *Task) error
// Pop removes and returns the next task. Blocks until available or context cancelled.
Pop(ctx context.Context) (*Task, error)
// Ack marks a claimed task as complete.
//
// Implementations that can lose a task on worker death need this to
// distinguish "finished" from "still in flight"; the in-memory queue
// implements it as a no-op because a crash takes the queue with it.
Ack(ctx context.Context, task *Task) error
// Nack returns a claimed task to the queue.
Nack(ctx context.Context, task *Task) error
// Len returns the number of tasks waiting to be claimed.
Len() int
// Close closes the queue.
Close() error
}
Queue is an interface for distributed task queues. Implementations can use Redis, in-memory, or any other backend.
func NewQueue ¶
func NewQueue(ctx context.Context, cfg *config.DistributedConfig, logger *slog.Logger) (Queue, error)
NewQueue builds the queue described by cfg: Redis-backed when a redis_addr is configured, in-memory otherwise.
It returns an error when Redis is configured but unreachable, rather than quietly running in memory. The previous behaviour — log a line about Redis and use an in-memory queue anyway — meant a typo in redis_addr produced a cluster where every worker had its own private queue, nothing was shared, and nothing survived a restart. That is a failure you find in production, at the worst possible time.
type RedisQueue ¶
type RedisQueue struct {
// contains filtered or unexported fields
}
RedisQueue is a Redis-backed distributed task queue.
This replaces a placeholder that logged "Redis queue initialized" and delegated every operation to an in-memory queue — so a multi-process deployment silently had no shared queue at all, and nothing survived a restart. A queue named for a backend it does not use is worse than no queue.
Delivery is at-least-once. A pop atomically moves the task from the pending list to a per-consumer processing list, so a worker that dies mid-task leaves the task recoverable rather than losing it. The reaper returns tasks whose visibility deadline has passed. Callers must Ack a task they finish, or Nack one they cannot; an unacked task comes back.
At-least-once means a task can be delivered twice — once to a worker that stalled past its deadline and once to its replacement. Crawl tasks are idempotent (the engine's deduplicator absorbs the repeat), which is why at-least-once is the right trade here rather than the considerably more expensive exactly-once.
func NewRedisQueue ¶
func NewRedisQueue(ctx context.Context, cfg *RedisQueueConfig, logger *slog.Logger) (*RedisQueue, error)
NewRedisQueue connects to Redis and starts the reaper.
It returns an error rather than falling back to an in-memory queue. Silently degrading to a non-shared, non-durable queue is how the previous implementation turned a configuration mistake into data loss that only showed up in production.
func (*RedisQueue) Ack ¶
func (q *RedisQueue) Ack(ctx context.Context, task *Task) error
Ack marks a task complete and removes it from the processing list.
func (*RedisQueue) Close ¶
func (q *RedisQueue) Close() error
Close stops the reaper and releases the connection. Tasks left in flight stay in Redis and are recovered by whichever process reaps them next — which is the point of a shared queue.
func (*RedisQueue) InFlight ¶
func (q *RedisQueue) InFlight() int
InFlight returns the number of claimed but unacked tasks.
func (*RedisQueue) Len ¶
func (q *RedisQueue) Len() int
Len returns the number of tasks waiting to be claimed. In-flight tasks are not counted; use InFlight for those.
func (*RedisQueue) Nack ¶
func (q *RedisQueue) Nack(ctx context.Context, task *Task) error
Nack returns a task to the pending queue immediately, rather than waiting out its visibility deadline. Used when a worker knows it cannot complete the task.
func (*RedisQueue) Pop ¶
func (q *RedisQueue) Pop(ctx context.Context) (*Task, error)
Pop claims the next task, blocking until one is available or ctx is done.
BLMOVE is what makes this recoverable: the task leaves the pending list and enters the processing list in one atomic step, so there is no window in which a crash loses it.
func (*RedisQueue) Push ¶
func (q *RedisQueue) Push(ctx context.Context, task *Task) error
Push adds a task to the queue.
Higher-priority tasks go to the head. Redis lists have no priority ordering, so this is a two-bucket approximation rather than the in-memory queue's full sort — enough to keep retries behind fresh work without a sorted set on the hot path.
type RedisQueueConfig ¶
type RedisQueueConfig struct {
Addr string
Password string
DB int
// Key is the namespace prefix for this queue's Redis keys.
Key string
// Visibility is how long a claimed task may be in flight before it is
// considered abandoned. Zero uses the default.
Visibility time.Duration
// ReapInterval is how often to sweep for abandoned tasks. Zero uses the
// default.
ReapInterval time.Duration
}
RedisQueueConfig configures a Redis-backed queue.
type Task ¶
type Task struct {
ID string `json:"id"`
Type string `json:"type"` // crawl, parse, store
URLs []string `json:"urls"`
Config map[string]any `json:"config"`
Priority int `json:"priority"`
Created time.Time `json:"created"`
Assigned string `json:"assigned_to,omitempty"`
Status string `json:"status"` // pending, running, done, failed
Result json.RawMessage `json:"result,omitempty"`
}
Task represents a unit of work assigned to a worker.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker represents a distributed crawl worker that pulls tasks from a master and executes crawl jobs independently.
Architecture:
Master Node
↓
Redis Queue
↓ ↓ ↓
Worker Worker Worker
func NewWorker ¶
func NewWorker(cfg *WorkerConfig, logger *slog.Logger) *Worker
NewWorker creates a new distributed worker.
func (*Worker) SetCrawlFunc ¶
SetCrawlFunc sets the function that executes crawl tasks.
func (*Worker) Start ¶
Start starts the worker: registers with master, begins heartbeat and task polling.
type WorkerConfig ¶
WorkerConfig configures a distributed worker.
func DefaultWorkerConfig ¶
func DefaultWorkerConfig() *WorkerConfig
DefaultWorkerConfig returns sensible defaults.
Source Files