Documentation
¶
Overview ¶
Package queue provides a coalescing, rate-limited, per-key work queue with the same semantics as Kubernetes client-go's workqueue: a key added while pending is deduplicated, a key added while being processed is re-delivered exactly once after the current run finishes, and no two workers ever process the same key concurrently. This is the serialization backbone of the orchestrator's single-writer reconcile model (RFC #129 Phase 3): correctness against concurrent spec writers comes from the store's UpdateFunc CAS, while this queue guarantees reconciles of one service never interleave.
Index ¶
- type Queue
- func (q *Queue) Add(key string)
- func (q *Queue) AddAfter(key string, d time.Duration)
- func (q *Queue) AddRateLimited(key string)
- func (q *Queue) Done(key string)
- func (q *Queue) Forget(key string)
- func (q *Queue) Get() (key string, shutdown bool)
- func (q *Queue) Len() int
- func (q *Queue) ShutDown()
- func (q *Queue) Stats() Stats
- func (q *Queue) Work(ctx context.Context, n int, ...)
- type RateLimiter
- type Stats
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is a coalescing per-key work queue. All methods are safe for concurrent use.
Key lifecycle (client-go workqueue semantics):
Add(k) : k ∉ dirty → dirty + pending(order); k ∈ dirty → no-op (coalesced);
k ∈ processing → dirty only (re-delivered after Done)
Get() : pops the oldest pending key → processing (removed from dirty)
Done(k) : k leaves processing; if re-Added meanwhile (k ∈ dirty) → pending again
func New ¶
func New(name string, rl RateLimiter) *Queue
New creates a Queue. name is used for identification in logs/stats only. A nil rl gets the DefaultRateLimiter.
func (*Queue) Add ¶
Add marks key as needing a run. Duplicate adds of a pending key coalesce; adding a key that is currently being processed schedules exactly one re-run after the in-flight run completes. Add on a shut-down queue is a no-op.
func (*Queue) AddAfter ¶
AddAfter adds key after delay d (immediately when d <= 0). Implemented with one timer per call rather than a shared heap: Add's dirty-set dedup makes redundant timers harmless, and at orchestrator scale (tens of services) the simplicity wins. A timer firing after ShutDown lands on the no-op Add path.
func (*Queue) AddRateLimited ¶
AddRateLimited adds key after the rate limiter's backoff for it.
func (*Queue) Done ¶
Done releases key after processing. If the key was re-Added while being processed, it is requeued for exactly one further run.
func (*Queue) Get ¶
Get blocks until a key is available or the queue shuts down. The returned key is exclusively held by the caller (no other Get returns it) until the caller invokes Done.
func (*Queue) ShutDown ¶
func (q *Queue) ShutDown()
ShutDown stops the queue: pending keys already in the queue are still delivered, then Get returns shutdown=true to every waiter. Idempotent.
func (*Queue) Work ¶
func (q *Queue) Work(ctx context.Context, n int, handler func(ctx context.Context, key string) error)
Work runs n workers calling handler for each key until ctx is cancelled or the queue is shut down, then blocks until all workers have exited. A handler error requeues the key with backoff; success resets its backoff. Per-key exclusivity is inherited from Get/Done: handler never runs concurrently for the same key.
type RateLimiter ¶
type RateLimiter interface {
// When returns the delay to apply before re-adding key, growing with each
// consecutive call for the same key.
When(key string) time.Duration
// Forget resets key's failure history (called after a successful sync).
Forget(key string)
// NumRequeues reports how many times key has been rate-limited since the
// last Forget.
NumRequeues(key string) int
}
RateLimiter computes per-item retry delays for AddRateLimited. Implementations must be safe for concurrent use.
func DefaultRateLimiter ¶
func DefaultRateLimiter() RateLimiter
DefaultRateLimiter matches the client-go default: 5ms base doubling up to 1000s. The reconciler's periodic resync is the safety net for anything that backs off this far.
func NewItemExponentialRateLimiter ¶
func NewItemExponentialRateLimiter(base, max time.Duration) RateLimiter
NewItemExponentialRateLimiter returns a per-key exponential-backoff limiter.
type Stats ¶
type Stats struct {
Adds uint64 // Add calls that enqueued or dirtied a key
Coalesced uint64 // Add calls absorbed by an already-dirty key
Requeues uint64 // AddRateLimited calls (failed syncs)
Processed uint64 // handler invocations completed (via Work)
Depth int // current pending keys
MaxDepth int // high-water mark of Depth
WorkDurationTotal time.Duration // cumulative handler runtime (via Work)
}
Stats is a point-in-time snapshot of queue counters, for observability.
Source Files