Documentation
¶
Overview ¶
Package dispatcher provides a workqueue dispatcher that dequeues keys and invokes a callback for each one.
The dispatcher handles orphaned in-progress keys, concurrency limits, and batch sizing. Use Handle for synchronous dispatch or HandleAsync for non-blocking dispatch with a Future to await results.
Error Handling ¶
When a callback returns an error, the dispatcher requeues, dead-letters, or drops the key depending on the error type and retry budget. To emit these errors as CloudEvents, pass WithErrorIngressURI:
handler := dispatcher.Handler(wq, 10, 5, callback, 3,
dispatcher.WithErrorIngressURI(ctx, ingressURI, "my-wq"),
)
When the ingress URL is empty the option is a no-op, making the feature entirely opt-in.
Failure-Retry Backoff ¶
By default a failed-and-requeued key reuses the workqueue's existing backoff. To install a custom failure-retry backoff (for example decorrelated exponential jitter), pass WithBackoff:
handler := dispatcher.Handler(wq, 10, 5, callback, 3,
dispatcher.WithBackoff(func(attempts int) time.Duration {
return backoff(attempts)
}),
)
The hook is called with the key's current attempt count on each requeued failure. A positive return value defers the retry by that duration while preserving the attempt count, so the dead-letter cutoff stays reachable. A nil hook or a non-positive return falls back to the default bare requeue, making the option entirely opt-in.
Index ¶
- Constants
- func Handle(ctx context.Context, wq workqueue.Interface, concurrency, batchSize int, ...) error
- func Handler(wq workqueue.Interface, concurrency, batchSize int, f Callback, maxRetry int, ...) http.Handler
- type Callback
- type DispatcherErrorEvent
- type ErrorAction
- type ErrorContext
- type Future
- type Option
Examples ¶
Constants ¶
const (
// ErrorEventType is the CloudEvent type for dispatcher error events.
ErrorEventType = "dev.chainguard.workqueue.error.v1"
)
Variables ¶
This section is empty.
Functions ¶
func Handle ¶
func Handle(ctx context.Context, wq workqueue.Interface, concurrency, batchSize int, f Callback, opts ...Option) error
Handle is a synchronous form of HandleAsync.
Example ¶
ExampleHandle demonstrates dispatching a single round of work from a workqueue using Handle.
package main
import (
"context"
"fmt"
"chainguard.dev/driftlessaf/workqueue"
"chainguard.dev/driftlessaf/workqueue/dispatcher"
"chainguard.dev/driftlessaf/workqueue/inmem"
)
func main() {
wq := inmem.NewWorkQueue(5)
ctx := context.Background()
if err := wq.Queue(ctx, "example-key", workqueue.Options{}); err != nil {
panic(err)
}
processed := false
err := dispatcher.Handle(ctx, wq, 5, 0, func(_ context.Context, key string, _ workqueue.Options) error {
fmt.Printf("Processing key: %s\n", key)
processed = true
return nil
})
if err != nil {
panic(err)
}
fmt.Printf("Processed: %v\n", processed)
}
Output: Processing key: example-key Processed: true
Types ¶
type Callback ¶
Callback is the function that Handle calls to process a particular key.
func ServiceCallback ¶
func ServiceCallback(client workqueue.WorkqueueServiceClient) Callback
ServiceCallback returns a Callback that invokes the given service.
Example ¶
ExampleServiceCallback demonstrates creating a Callback that delegates to a WorkqueueServiceClient.
package main
import (
"fmt"
"chainguard.dev/driftlessaf/workqueue"
"chainguard.dev/driftlessaf/workqueue/dispatcher"
)
func main() {
// ServiceCallback wraps a gRPC WorkqueueServiceClient as a dispatcher Callback.
// In production, pass a real client obtained from a gRPC connection.
var client workqueue.WorkqueueServiceClient
cb := dispatcher.ServiceCallback(client)
_ = cb
fmt.Println("ServiceCallback created")
}
Output: ServiceCallback created
type DispatcherErrorEvent ¶ added in v0.4.0
type DispatcherErrorEvent struct {
// Key is the workqueue key that failed.
Key string `json:"key"`
// Error is the error message.
Error string `json:"error"`
// Attempts is the number of times the key has been attempted.
Attempts int `json:"attempts"`
// Action describes the disposition (requeued, dead-lettered, dropped).
Action string `json:"action"`
// NonRetriableReason is set when Action is "dropped", providing the
// reason the error was marked non-retriable.
NonRetriableReason string `json:"nonRetriableReason,omitempty"`
// OccurAt is the time the error occurred. This field is used as the
// BigQuery partition field.
OccurAt time.Time `json:"occurAt"`
// ReconcilerName is the name of the reconciler / workqueue that produced
// this error. It allows distinguishing errors from different reconcilers
// that share the same key space.
ReconcilerName string `json:"reconcilerName,omitempty"`
}
DispatcherErrorEvent is the CloudEvent payload for dispatcher errors.
type ErrorAction ¶ added in v0.4.0
type ErrorAction int
ErrorAction describes the disposition of a key after a dispatch error.
const ( // ErrorRequeued indicates the key was returned to the queue for retry. ErrorRequeued ErrorAction = iota // ErrorDeadLettered indicates the key exhausted its retry budget and was // moved to the dead-letter queue. ErrorDeadLettered // ErrorDropped indicates the error was marked non-retriable and the key // was completed without further processing. ErrorDropped )
func (ErrorAction) String ¶ added in v0.4.0
func (a ErrorAction) String() string
type ErrorContext ¶ added in v0.4.0
type ErrorContext struct {
// Key is the workqueue key that failed.
Key string
// Err is the error returned by the callback.
Err error
// Attempts is the number of times the key has been attempted (including
// the current attempt).
Attempts int
// Action is what the dispatcher did with the key after the error.
Action ErrorAction
// NonRetriableReason is set when Action is ErrorDropped, providing
// the reason the error was marked non-retriable.
NonRetriableReason string
}
ErrorContext carries information about a dispatch error.
type Future ¶
type Future func() error
Future is a function that can be used to block on the result of a round of dispatching work.
func HandleAsync ¶
func HandleAsync(ctx context.Context, wq workqueue.Interface, concurrency, batchSize int, f Callback, maxRetry int, opts ...Option) Future
HandleAsync initiates a single iteration of the dispatcher, possibly invoking the callback for several different keys. It returns a future that can be used to block on the result.
type Option ¶ added in v0.4.0
type Option func(*config)
Option configures optional dispatcher behavior.
func WithBackoff ¶ added in v0.7.14
WithBackoff sets the failure-retry backoff for the dispatcher. On each callback failure that is requeued (not dead-lettered, not dropped as non-retriable), the dispatcher calls fn with the key's current attempt count and, when fn returns a positive duration, requeues the key with that not-before delay while preserving the attempt count (so the dead-letter cutoff stays reachable).
When fn is nil or returns a non-positive duration, the dispatcher falls back to a bare requeue, identical to the behavior with this option unset. This makes the option entirely opt-in: an unconfigured dispatcher keeps the existing requeue behavior bit-for-bit. Use it to install decorrelated exponential jitter or any other attempt-driven backoff curve.
Example ¶
ExampleWithBackoff demonstrates installing a failure-retry backoff. The hook receives the key's attempt count and returns the delay before the next retry; here it grows linearly with attempts. Returning a non-positive duration (or leaving the option unset) keeps the default bare-requeue behavior.
package main
import (
"context"
"fmt"
"time"
"chainguard.dev/driftlessaf/workqueue"
"chainguard.dev/driftlessaf/workqueue/dispatcher"
"chainguard.dev/driftlessaf/workqueue/inmem"
)
func main() {
wq := inmem.NewWorkQueue(5)
ctx := context.Background()
if err := wq.Queue(ctx, "example-key", workqueue.Options{}); err != nil {
panic(err)
}
// Back off two seconds per attempt before the next retry.
backoff := func(attempts int) time.Duration {
return time.Duration(attempts) * 2 * time.Second
}
err := dispatcher.Handle(ctx, wq, 5, 0, func(_ context.Context, _ string, _ workqueue.Options) error {
return fmt.Errorf("something went wrong")
}, dispatcher.WithBackoff(backoff))
if err != nil {
panic(err)
}
fmt.Println("dispatched with failure-retry backoff")
}
Output: dispatched with failure-retry backoff
func WithDispatchPeriod ¶ added in v0.9.4
WithDispatchPeriod sets the minimum interval between dispatch passes for a Handler: at most one pass is admitted per period, and triggers beyond that are acknowledged and dropped. The default is one second. It has no effect on Handle or HandleAsync, which initiate unconditionally.
func WithErrorIngressURI ¶ added in v0.4.1
WithErrorIngressURI configures the dispatcher to emit errors as CloudEvents to the given broker URL. workqueueName is used as the CloudEvent source. When ingressURI is empty, error events are disabled (no-op).
The CloudEvents client is constructed eagerly so it is shared across all dispatch rounds. If construction fails the option degrades gracefully to a no-op with a warning log.
Example ¶
ExampleWithErrorIngressURI demonstrates enabling error events. When the broker URL is empty, error events are silently disabled.
package main
import (
"context"
"fmt"
"chainguard.dev/driftlessaf/workqueue"
"chainguard.dev/driftlessaf/workqueue/dispatcher"
"chainguard.dev/driftlessaf/workqueue/inmem"
)
func main() {
wq := inmem.NewWorkQueue(5)
ctx := context.Background()
if err := wq.Queue(ctx, "example-key", workqueue.Options{}); err != nil {
panic(err)
}
// Empty URL disables error events (no-op).
err := dispatcher.Handle(ctx, wq, 5, 0, func(_ context.Context, _ string, _ workqueue.Options) error {
return fmt.Errorf("something went wrong")
}, dispatcher.WithErrorIngressURI(ctx, "", "my-workqueue"))
if err != nil {
panic(err)
}
fmt.Println("dispatched with error events disabled")
}
Output: dispatched with error events disabled
Source Files