worker

type HealthTracker ¶

type HealthTracker struct {
	// contains filtered or unexported fields
}

HealthTracker tracks the health status of all workers. It is safe for concurrent use.

func NewHealthTracker() *HealthTracker

func (h *HealthTracker) GetStatus() map[string]interface{}

func (h *HealthTracker) IsHealthy() bool

func (h *HealthTracker) MarkDegraded(name string, since time.Time, reason string)

func (h *HealthTracker) MarkFailed(name string)

func (h *HealthTracker) MarkFailedWithReason(name string, since time.Time, reason string)

func (h *HealthTracker) MarkHealthy(name string)

type Limits ¶ added in v1.6.0

type Limits struct {
	// MaxAttempts is the number of restarts. The worker escalates when the
	// run after the last allowed restart fails (0: on the first failure).
	MaxAttempts int
	// MaxDuration is the time since the first failed run of the episode.
	MaxDuration time.Duration
}

Limits bounds a failure episode, which starts at the first failed run and ends when a run stays up for the stable window. Whichever limit is hit first escalates the worker to failed. A negative value disables a limit.

type RegisterOption ¶ added in v1.6.0

type RegisterOption func(*registration)

RegisterOption configures how the supervisor runs a registered worker.

func WithRestartPolicy(p RestartPolicy) RegisterOption

type RestartPolicy ¶ added in v1.6.0

type RestartPolicy struct {
	Startup  Limits
	Recovery Limits
}

RestartPolicy makes the supervisor restart a worker whose Run returns an error, instead of marking it failed for good.

A worker is in the startup phase until a run has stayed up for the stable window (30s) once, counted from when the run calls Ready; after that it is in the recovery phase. Each phase has its own Limits. While within the limits the worker is degraded (/healthz 200); once either limit is hit it is failed (/healthz 503). The supervisor keeps restarting it either way, so it can come back to healthy.

Backoff between runs: 1s doubling, capped at 60s, ±20% jitter; 60s once failed.

type SupervisorOption ¶

type SupervisorOption func(*WorkerSupervisor)

SupervisorOption configures a WorkerSupervisor.

func WithShutdownTimeout(timeout time.Duration) SupervisorOption

type Worker ¶

type Worker interface {
	// Name returns a unique identifier for this worker (e.g., "http-server", "retrymq-consumer")
	Name() string

	// Run executes the worker and blocks until context is cancelled or error occurs.
	// Returns nil or context.Canceled for graceful shutdown.
	// Returns error for fatal failures.
	Run(ctx context.Context) error
}

Worker represents a long-running background process. Each worker runs in its own goroutine and can be monitored for health.

Workers should: - Block in Run() until context is cancelled or a fatal error occurs - Return nil or context.Canceled for graceful shutdown - Return non-nil error only for fatal failures

type WorkerHealth ¶

type WorkerHealth struct {
	Status string     `json:"status"`
	Since  *time.Time `json:"since,omitempty"`
	Reason string     `json:"reason,omitempty"`
}

WorkerHealth represents the health status of a single worker. Error details are NOT exposed for security reasons: /healthz is unauthenticated and broker errors carry hosts and credentials.

type WorkerSupervisor ¶

type WorkerSupervisor struct {
	// contains filtered or unexported fields
}

WorkerSupervisor manages and supervises multiple workers. It tracks their health and handles graceful shutdown.

func NewWorkerSupervisor(logger *logging.Logger, opts ...SupervisorOption) *WorkerSupervisor

func (r *WorkerSupervisor) GetHealthTracker() *HealthTracker

func (r *WorkerSupervisor) Register(w Worker, opts ...RegisterOption)

func (r *WorkerSupervisor) Run(ctx context.Context) error