Help us shape the future of Go! Take the 2026 Go Developer Survey

server

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package server is the gRPC adapter for jobrun.v1.JobService and jobrun.v1.ScheduleService: it validates request shape, converts proto <-> domain and maps domain errors to status codes. Business rules live in internal/job and the store.

The ports below are consumer-defined and satisfied by job.Store; tests pass fakes.

Index

Constants

View Source 
const (
	// RemoteOwnerPrefix prefixes the lease_owner of every remote worker, so
	// `jobrun list` tells in-process and remote leases apart.
	RemoteOwnerPrefix = "remote:"
	// MaxWorkerIDLen bounds ClaimRequest.worker_id.
	MaxWorkerIDLen = 128
	// MaxCreditsPerMessage bounds ClaimRequest.credits.
	MaxCreditsPerMessage = 1024
	// MaxLeasesPerHeartbeat bounds HeartbeatRequest.leases.
	MaxLeasesPerHeartbeat = 4096
	// MaxClaimQueues bounds ClaimRequest.queues.
	MaxClaimQueues = 32
)

Limits of the worker protocol.

Variables

This section is empty.

Functions

This section is empty.

Types

type DepthSource 

type DepthSource interface {
	Depth() ([]job.DepthRow, time.Time)
}

DepthSource is a cached depth table (admin.DepthCache satisfies it). When one is installed with WithDepthCache, Stats serves it instead of calling Jobs.Depth per request, as docs/PLAN.md requires.

type EventPublisher 

type EventPublisher interface {
	Publish(job.Event)
}

EventPublisher receives the transitions WorkerServer drives, exactly like the in-process pool publishes its own (admin.Broker satisfies it).

type JobOption 

type JobOption func(*JobServer)

JobOption configures NewJobServer.

func WithDepthCache 

func WithDepthCache(d DepthSource) JobOption

WithDepthCache makes Stats read the cached depth table instead of hitting the store on every call. Pass the admin.DepthCache.

type JobServer 

type JobServer struct {
	jobrunv1.UnimplementedJobServiceServer
	// contains filtered or unexported fields
}

JobServer implements jobrunv1.JobServiceServer on top of Jobs.

func NewJobServer 

func NewJobServer(jobs Jobs, log *slog.Logger, opts ...JobOption) *JobServer

NewJobServer builds the adapter. A nil logger discards.

func (*JobServer) Cancel 

func (s *JobServer) Cancel(ctx context.Context, req *jobrunv1.CancelRequest) (*jobrunv1.CancelResponse, error)

Cancel stops a queued job at once or requests a cooperative cancel of a leased one.

func (*JobServer) Enqueue 

func (s *JobServer) Enqueue(ctx context.Context, req *jobrunv1.EnqueueRequest) (resp *jobrunv1.EnqueueResponse, err error)

Enqueue inserts one job. A replay of (queue, idempotency_key) returns the stored job with created=false.

It opens the jobrun.enqueue producer span (a child of the caller's trace when the gRPC stats handler extracted one) and stores its W3C traceparent with the job, so the run span - in-process or on a remote worker - joins the same trace. The trace context is captured at write time or the trace breaks.

func (*JobServer) Get 

func (s *JobServer) Get(ctx context.Context, req *jobrunv1.GetRequest) (*jobrunv1.GetResponse, error)

Get returns one job, with its history when include_events is set.

func (*JobServer) List 

func (s *JobServer) List(ctx context.Context, req *jobrunv1.ListRequest) (*jobrunv1.ListResponse, error)

List pages jobs newest first with an opaque keyset page_token.

func (*JobServer) Purge 

func (s *JobServer) Purge(ctx context.Context, req *jobrunv1.PurgeRequest) (*jobrunv1.PurgeResponse, error)

Purge deletes finished jobs by queue, status and age.

func (*JobServer) Register 

func (s *JobServer) Register(reg grpc.ServiceRegistrar)

Register attaches the service to a gRPC server (or any ServiceRegistrar).

func (*JobServer) Retry 

func (s *JobServer) Retry(ctx context.Context, req *jobrunv1.RetryRequest) (*jobrunv1.RetryResponse, error)

Retry moves a dead or cancelled job back to queued with attempts reset.

func (*JobServer) Stats 

func (s *JobServer) Stats(ctx context.Context, req *jobrunv1.StatsRequest) (*jobrunv1.StatsResponse, error)

Stats returns the queue depth by queue and status, from the depth cache when one is installed.

type Jobs 

type Jobs interface {
	Enqueue(ctx context.Context, j job.Job) (job.Job, error)
	Get(ctx context.Context, id job.ID) (job.Job, error)
	Events(ctx context.Context, id job.ID) ([]job.Event, error)
	List(ctx context.Context, q job.ListQuery) ([]job.Job, job.Cursor, error)
	Cancel(ctx context.Context, id job.ID) (job.Job, error)
	Retry(ctx context.Context, id job.ID) (job.Job, error)
	Purge(ctx context.Context, q job.PurgeQuery) (int64, error)
	Depth(ctx context.Context) ([]job.DepthRow, error)
}

Jobs is the slice of job.Store that JobServer drives.

type Leases 

type Leases interface {
	Claim(ctx context.Context, q job.ClaimQuery) ([]job.Job, error)
	Heartbeat(ctx context.Context, owner string, leases []job.Lease, until time.Time) (job.Renewal, error)
	Complete(ctx context.Context, l job.Lease, owner string, fn job.Commit) error
	Fail(ctx context.Context, l job.Lease, owner, cause string, perm bool, retryAt time.Time) (job.Status, error)
	Get(ctx context.Context, id job.ID) (job.Job, error)
}

Leases is the slice of job.Store that WorkerServer drives.

type ScheduleServer 

type ScheduleServer struct {
	jobrunv1.UnimplementedScheduleServiceServer
	// contains filtered or unexported fields
}

ScheduleServer implements jobrunv1.ScheduleServiceServer on top of Schedules.

func NewScheduleServer 

func NewScheduleServer(svc Schedules, log *slog.Logger) *ScheduleServer

NewScheduleServer builds the adapter. A nil logger discards.

func (*ScheduleServer) CreateSchedule 

func (s *ScheduleServer) CreateSchedule(ctx context.Context, req *jobrunv1.CreateScheduleRequest) (*jobrunv1.CreateScheduleResponse, error)

CreateSchedule registers a new schedule; enabled defaults to true.

func (*ScheduleServer) DeleteSchedule 

func (s *ScheduleServer) DeleteSchedule(ctx context.Context, req *jobrunv1.DeleteScheduleRequest) (*jobrunv1.DeleteScheduleResponse, error)

DeleteSchedule removes a schedule; materialised jobs are untouched.

func (*ScheduleServer) GetSchedule 

func (s *ScheduleServer) GetSchedule(ctx context.Context, req *jobrunv1.GetScheduleRequest) (*jobrunv1.GetScheduleResponse, error)

GetSchedule returns one schedule by id.

func (*ScheduleServer) ListSchedules 

func (s *ScheduleServer) ListSchedules(ctx context.Context, req *jobrunv1.ListSchedulesRequest) (*jobrunv1.ListSchedulesResponse, error)

ListSchedules pages by name ascending; the token is the last name seen.

func (*ScheduleServer) Register 

func (s *ScheduleServer) Register(reg grpc.ServiceRegistrar)

Register attaches the service to a gRPC server (or any ServiceRegistrar).

func (*ScheduleServer) UpdateSchedule 

func (s *ScheduleServer) UpdateSchedule(ctx context.Context, req *jobrunv1.UpdateScheduleRequest) (*jobrunv1.UpdateScheduleResponse, error)

UpdateSchedule changes cron, timezone, template or enabled; unset fields keep their value. next_run is recomputed from now.

type Schedules 

type Schedules interface {
	CreateSchedule(ctx context.Context, s job.Schedule) (job.Schedule, error)
	UpdateSchedule(ctx context.Context, s job.Schedule) (job.Schedule, error)
	DeleteSchedule(ctx context.Context, id job.ID) error
	GetSchedule(ctx context.Context, id job.ID) (job.Schedule, error)
	ListSchedules(ctx context.Context, afterName string, limit int32) ([]job.Schedule, error)
}

Schedules is the slice of job.Schedules that ScheduleServer drives.

type WorkerConfig 

type WorkerConfig struct {
	// LeaseTTL is the lease granted per claim and per heartbeat (30 s).
	LeaseTTL time.Duration
	// PollFloor is the claim loop's tick when the dispatcher is silent (200 ms).
	PollFloor time.Duration
	// BatchSize is the most jobs one Claim round trip asks for (32).
	BatchSize int
	// MaxOutstanding caps the credits a stream may hold at once (1024).
	MaxOutstanding int
}

WorkerConfig tunes WorkerServer; zero fields take the defaults.

type WorkerOption 

type WorkerOption func(*WorkerServer)

WorkerOption configures NewWorkerServer.

func WithDispatcher 

func WithDispatcher(d job.Dispatcher) WorkerOption

WithDispatcher wakes Claim streams on new work (postgres.Notifier); without one they poll at PollFloor.

func WithPublisher 

func WithPublisher(p EventPublisher) WorkerOption

WithPublisher sends remote transitions to the admin broker.

type WorkerServer 

type WorkerServer struct {
	jobrunv1.UnimplementedWorkerServiceServer
	// contains filtered or unexported fields
}

WorkerServer implements jobrunv1.WorkerServiceServer: it leases jobs to out-of-process workers. Leases are fenced by (owner, token) exactly like the in-process pool's, so the reaper and a remote worker can never both finish a job. Effects of remote handlers are at-least-once: nothing the worker does can commit in the completion transaction.

func NewWorkerServer 

func NewWorkerServer(l Leases, log *slog.Logger, cfg WorkerConfig, opts ...WorkerOption) *WorkerServer

NewWorkerServer builds the adapter. A nil logger discards.

func (*WorkerServer) Claim 

func (s *WorkerServer) Claim(stream grpc.BidiStreamingServer[jobrunv1.ClaimRequest, jobrunv1.ClaimResponse]) error

Claim is the lease stream; see the proto for the protocol.

func (*WorkerServer) Complete 

func (s *WorkerServer) Complete(ctx context.Context, req *jobrunv1.CompleteRequest) (resp *jobrunv1.CompleteResponse, err error)

Complete marks a remote job succeeded (or cancelled when a cancel was requested meanwhile). There is no Commit: remote effects are at-least-once.

func (*WorkerServer) Drain 

func (s *WorkerServer) Drain()

Drain ends every open Claim stream (and refuses new ones) so that grpc.Server.GracefulStop does not wait on long-lived streams. Remote workers keep heartbeating and reporting the jobs they already hold until the server is gone. Safe to call more than once.

func (*WorkerServer) Fail 

func (s *WorkerServer) Fail(ctx context.Context, req *jobrunv1.FailRequest) (resp *jobrunv1.FailResponse, err error)

Fail records a remote attempt failure with the job's own backoff policy.

func (*WorkerServer) Heartbeat 

func (s *WorkerServer) Heartbeat(ctx context.Context, req *jobrunv1.HeartbeatRequest) (*jobrunv1.HeartbeatResponse, error)

Heartbeat extends a remote worker's leases and reports kept, cancelled and lost.

func (*WorkerServer) Register 

func (s *WorkerServer) Register(reg grpc.ServiceRegistrar)

Register attaches the service to a gRPC server (or any ServiceRegistrar).

func (*WorkerServer) Stats 

func (s *WorkerServer) Stats() WorkerStats

Stats returns a snapshot of the counters.

type WorkerStats 

type WorkerStats struct {
	Streams   int64 // Claim streams open right now
	Delivered int64 // jobs pushed down a Claim stream
	Completed int64 // Complete calls that committed
	Failed    int64 // Fail calls that committed
	// Fenced counts leases a Heartbeat reported lost to a remote worker (it
	// must cancel those handlers and never finish them).
	Fenced int64
	// LeaseLost counts Complete/Fail calls refused because the lease was gone.
	LeaseLost int64
	// Undelivered counts jobs claimed for a stream that broke before the send;
	// they are failed straight back to the queue.
	Undelivered int64
}

WorkerStats are WorkerServer's cumulative counters.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.