Documentation
¶
Overview ¶
Package cluster provides distributed worker coordination, consensus-based leader election, worker registration, and work stealing / rebalancing.
When running multiple Dispatch instances, the cluster package coordinates which instance is the leader (responsible for cron firing, stale-job recovery, and work redistribution) and which are followers.
Worker Entity ¶
Each running Dispatch instance registers itself as a Worker with:
- a unique id.WorkerID
- its hostname
- the list of queues it polls
- its concurrency limit
- a state: WorkerActive, WorkerDraining, or WorkerDead
Workers send periodic heartbeats. If a heartbeat is not received within the configured threshold, the worker is considered dead and its in-flight jobs are eligible for reassignment (work stealing).
Leader Election ¶
One worker at a time holds leadership. The leader:
- fires cron entries
- reclaims stale jobs from dead workers
- rebalances work across the cluster
Leadership is managed by Store.AcquireLeadership using optimistic locking. If leadership is lost mid-operation, [dispatch.ErrLeadershipLost] is returned.
Kubernetes Consensus ¶
For K8s deployments use the cluster/k8s sub-package which uses Kubernetes leader election via client-go for consensus.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Store ¶
type Store interface {
// RegisterWorker adds a new worker to the cluster registry.
RegisterWorker(ctx context.Context, w *Worker) error
// DeregisterWorker removes a worker from the cluster registry.
DeregisterWorker(ctx context.Context, workerID id.WorkerID) error
// HeartbeatWorker updates the last-seen timestamp for a worker,
// indicating it is still alive. It returns dispatch.ErrWorkerNotFound
// when the row is gone, which the engine's heartbeat takes as the
// signal to register the worker again.
HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error
// GetWorker returns one registered worker, or dispatch.ErrWorkerNotFound
// when no row exists for workerID.
GetWorker(ctx context.Context, workerID id.WorkerID) (*Worker, error)
// ListWorkers returns all registered workers.
ListWorkers(ctx context.Context) ([]*Worker, error)
// ReapDeadWorkers returns workers whose last-seen timestamp is older than
// the given threshold, indicating they may have crashed.
ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*Worker, error)
// DeleteStaleWorkers removes worker rows whose last-seen timestamp is
// older than the given threshold. Returns the number of rows deleted.
// Used at startup to clear out rows from prior instances that died
// without deregistering — without this, a stale leader row blocks the
// partial-unique leader index and prevents the new instance from ever
// claiming leadership.
DeleteStaleWorkers(ctx context.Context, threshold time.Duration) (int64, error)
// AcquireLeadership attempts to become the cluster leader.
// Returns true if this worker is now leader. The leadership
// expires after ttl if not renewed.
AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)
// RenewLeadership extends the leader's hold. Must be called
// before the TTL expires.
RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)
// GetLeader returns the current cluster leader, or nil if there
// is no leader.
GetLeader(ctx context.Context) (*Worker, error)
}
Store defines the persistence contract for cluster worker management.
type Worker ¶
type Worker struct {
ID id.WorkerID `json:"id"`
Hostname string `json:"hostname"`
Queues []string `json:"queues"`
Concurrency int `json:"concurrency"`
State WorkerState `json:"state"`
// Capacity is what this worker can run at once. It is advisory:
// empty means unknown, which disables the enqueue-time unschedulable
// check rather than rejecting everything.
Capacity resource.Set `json:"capacity,omitempty"`
IsLeader bool `json:"is_leader"`
LeaderUntil *time.Time `json:"leader_until,omitempty"`
LastSeen time.Time `json:"last_seen"`
Metadata map[string]string `json:"metadata,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
Worker represents a Dispatch worker instance in a distributed cluster.
type WorkerState ¶
type WorkerState string
WorkerState represents the lifecycle state of a worker.
const ( // WorkerActive means the worker is healthy and processing jobs. WorkerActive WorkerState = "active" // WorkerDraining means the worker is finishing in-flight jobs // but not accepting new ones (graceful shutdown). WorkerDraining WorkerState = "draining" // WorkerDead means the worker has stopped responding and should // have its jobs reassigned. WorkerDead WorkerState = "dead" )
Source Files
Directories