workerclient

Documentation

Overview

Package workerclient is the Go SDK for the codeq worker streaming gRPC API. The high-level entry point is Client.Run, which opens a long-lived bidirectional stream, authenticates once, and dispatches claimed tasks to a user-provided Handler running on a configurable number of concurrent slots. Phase 2 of the throughput refactor relies on this client to bypass the per-call HTTP middleware tax.

Index

• type Client
  • func New(cfg Config) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Run(ctx context.Context, h Handler) error
• type Config
• type Handler
• type Result
  • func Abandon() Result
  • func Completed(body map[string]any) Result
  • func Failed(err string) Result
  • func Nack(delaySeconds int, reason string) Result
• type Task

Types

type Client

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

Client is the long-lived worker stream client. Use New to construct one, Run to start dispatching tasks, and Close to release the underlying connection.

func New

func New(cfg Config) (*Client, error)

New dials the server and returns a Client ready for Run. The connection stays open until Close is called.

func (*Client) Close

func (c *Client) Close() error

Close releases the underlying gRPC connection. Safe to call multiple times.

func (*Client) Run

func (c *Client) Run(ctx context.Context, h Handler) error

Run opens a stream, completes the Hello handshake, and dispatches tasks to h until ctx is cancelled or a fatal stream error occurs. The returned error is the cause of termination — nil only when the server cleanly closed the stream.

type Config

type Config struct {
	// Addr is the gRPC dial target of the codeq worker stream server
	// (e.g. "localhost:9091"). Required.
	Addr string

	// Token is the bearer token presented in Hello. Required.
	Token string

	// WorkerID identifies this worker for lease ownership. If empty the
	// server uses the JWT subject from Token.
	WorkerID string

	// Commands restricts what this worker pulls. nil/empty means "use
	// whatever the token's eventTypes claim allows".
	Commands []string

	// Concurrency is the number of in-flight tasks the client maintains.
	// Defaults to 1. Each slot holds one Ready→Task(s)→Result(s) cycle.
	Concurrency int

	// LeaseSeconds is sent on each Ready. 0 means "server default".
	LeaseSeconds int

	// BatchSize controls how many tasks each slot tries to claim per
	// Ready, and how many Results coalesce into one ResultBatch. 0/1
	// keeps the legacy single-task path (one Ready → one Task → one
	// Result per cycle). >1 enables Phase 6 / Q2 batching: each cycle
	// pulls up to BatchSize Tasks via Ready{count=BatchSize} → server
	// replies with TaskBatch, and the resulting Results are sent back
	// as one ResultBatch. Amortises gRPC framing + Pebble commit cost
	// across the batch.
	BatchSize int

	// IdleBackoff is how long a slot waits before re-sending Ready when
	// the previous Ready didn't yield a task within ReadyTimeout. Defaults
	// to 50ms.
	IdleBackoff time.Duration

	// DialOptions are forwarded to grpc.NewClient. If empty, the client
	// uses insecure transport credentials — set this for TLS/mTLS.
	DialOptions []grpc.DialOption

	// Logger receives structured info/warn/error events. Defaults to
	// slog.Default().
	Logger *slog.Logger
}

Config controls Client behavior. Addr and Token are required; the rest have sensible defaults documented per-field.

type Handler

type Handler func(ctx context.Context, t Task) Result

Handler processes one task and returns the disposition. Implementations must be safe for concurrent invocation — Client.Run will fan out up to Config.Concurrency calls in parallel.

type Result

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

Result is what a Handler returns. Construct one with Completed, Failed, Nack, or Abandon — the zero value is invalid and the client will reject it.

func Abandon

func Abandon() Result

Abandon releases the lease without nacking. The task goes straight back to pending and another worker can claim it immediately. Used by workers that are shutting down cleanly mid-task.

func Completed

func Completed(body map[string]any) Result

Completed marks a task done with an optional result payload. The payload is JSON-encoded by the server; pass nil for "no payload".

func Failed

func Failed(err string) Result

Failed marks a task as permanently failed with a human-readable error. The scheduler will respect MaxAttempts; failures past the limit go to the DLQ.

func Nack

func Nack(delaySeconds int, reason string) Result

Nack returns the task to the queue after delaySeconds. reason is recorded for observability. delaySeconds<0 is clamped to 0.

type Task

type Task struct {
	ID          string
	Command     string
	Payload     []byte
	Priority    int
	Attempts    int
	MaxAttempts int
	TenantID    string
	Webhook     string
	LeaseUntil  string
}

Task is the unit of work the handler receives. It mirrors the server-side domain.Task but exposes only the fields a handler usually needs.