probo/contrib/claude/go-worker.md

Go Worker

Background workers use go.gearno.de/kit/worker. The kit handles the polling loop, semaphore-based concurrency, graceful shutdown, non-cancellable contexts, Prometheus metrics, and OpenTelemetry tracing. You only implement a handler.

Handler interface

Implement worker.Handler[T] with Claim and Process. Optionally implement worker.StaleRecoverer for stale row recovery.

type fooHandler struct {
	pg         *pg.Client
	logger     *log.Logger
	staleAfter time.Duration
}

func (h *fooHandler) Claim(ctx context.Context) (coredata.FooItem, error) {
	var item coredata.FooItem

	if err := h.pg.WithTx(
		ctx,
		func(ctx context.Context, tx pg.Tx) error {
			if err := item.LoadNextPendingForUpdateSkipLocked(ctx, tx); err != nil {
				return err
			}

			now := time.Now()
			item.Status = coredata.FooStatusProcessing
			item.UpdatedAt = now
			return item.Update(ctx, tx, coredata.NewNoScope())
		},
	); err != nil {
		if errors.Is(err, coredata.ErrResourceNotFound) {
			return coredata.FooItem{}, worker.ErrNoTask
		}
		return coredata.FooItem{}, err
	}

	return item, nil
}

func (h *fooHandler) Process(ctx context.Context, item coredata.FooItem) error {
	if err := h.handle(ctx, &item); err != nil {
		if failErr := h.fail(ctx, &item, err); failErr != nil {
			h.logger.ErrorCtx(ctx, "cannot fail item", log.Error(failErr))
		}
		return err
	}
	return nil
}

func (h *fooHandler) RecoverStale(ctx context.Context) error {
	return h.pg.WithConn(
		ctx,
		func(ctx context.Context, conn pg.Querier) error {
			return coredata.ResetStaleFooItems(ctx, conn, h.staleAfter)
		},
	)
}

Constructor

The constructor builds the handler and returns *worker.Worker[T]. Use worker.WithInterval and worker.WithMaxConcurrency for tuning. Keep domain-specific options (e.g. timeouts, staleAfter) on the handler struct.

func NewFooWorker(
	pgClient *pg.Client,
	logger *log.Logger,
	opts ...worker.Option,
) *worker.Worker[coredata.FooItem] {
	h := &fooHandler{
		pg:         pgClient,
		logger:     logger,
		staleAfter: 5 * time.Minute,
	}

	return worker.New(
		"foo-worker",
		h,
		logger,
		opts...,
	)
}

Key principles

• Claim with FOR UPDATE SKIP LOCKED — prevents multiple workers from picking the same row
• Return worker.ErrNoTask from Claim when no work is available (not the coredata sentinel)
• Context is non-cancellable — the kit provides context.WithoutCancel to both Claim and Process
• Process handles its own failures — update DB status on error, the kit only logs and records metrics
• Stale recovery is optional — implement worker.StaleRecoverer if the worker marks rows as "processing"
• Kit provides observability — Prometheus metrics (worker_tasks_total, worker_task_duration_seconds, etc.) and OTel traces are automatic