api

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Jul 10, 2026 License: Apache-2.0 Imports: 32 Imported by: 0

Documentation

Overview

Package api implements the ConnectRPC services of conveyord: the task enqueue API, the worker session protocol, and the bearer-token authentication shared by every service.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func NewAuthInterceptor 

func NewAuthInterceptor(tokens []string) connect.Interceptor

NewAuthInterceptor builds the bearer-token interceptor from the accepted tokens. Token comparison is constant-time per candidate.

func NewReadOnlyInterceptor 

func NewReadOnlyInterceptor() connect.Interceptor

NewReadOnlyInterceptor builds an interceptor that denies admin mutations.

Types

type AdminService 

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

AdminService serves the inspection and operations API.

func NewAdminService 

func NewAdminService(engine *actors.Engine, taskLog broker.Broker, timeSource clock.Clock, sessions SessionLister, allowInsecureWebhooks bool) *AdminService

NewAdminService assembles the admin API service.

func (*AdminService) ArchiveTask 

func (s *AdminService) ArchiveTask(ctx context.Context, request *connect.Request[conveyorv1.ArchiveTaskRequest]) (*connect.Response[conveyorv1.ArchiveTaskResponse], error)

ArchiveTask dead-letters a waiting (scheduled, pending, or retry) task.

func (*AdminService) BatchArchiveTasks 

func (s *AdminService) BatchArchiveTasks(ctx context.Context, request *connect.Request[conveyorv1.BatchTasksRequest]) (*connect.Response[conveyorv1.BatchTasksResponse], error)

BatchArchiveTasks dead-letters each listed task.

func (*AdminService) BatchCancelTasks 

func (s *AdminService) BatchCancelTasks(ctx context.Context, request *connect.Request[conveyorv1.BatchTasksRequest]) (*connect.Response[conveyorv1.BatchTasksResponse], error)

BatchCancelTasks cancels each listed task.

func (*AdminService) BatchDeleteTasks 

func (s *AdminService) BatchDeleteTasks(ctx context.Context, request *connect.Request[conveyorv1.BatchTasksRequest]) (*connect.Response[conveyorv1.BatchTasksResponse], error)

BatchDeleteTasks deletes each listed task, reporting per-id outcomes.

func (*AdminService) BatchRunTasks 

func (s *AdminService) BatchRunTasks(ctx context.Context, request *connect.Request[conveyorv1.BatchTasksRequest]) (*connect.Response[conveyorv1.BatchTasksResponse], error)

BatchRunTasks makes each listed task due immediately.

func (*AdminService) BrokerInfo 

func (s *AdminService) BrokerInfo(ctx context.Context, _ *connect.Request[conveyorv1.BrokerInfoRequest]) (*connect.Response[conveyorv1.BrokerInfoResponse], error)

BrokerInfo reports the storage engine's driver and runtime statistics.

func (*AdminService) CancelTask 

func (s *AdminService) CancelTask(ctx context.Context, request *connect.Request[conveyorv1.CancelTaskRequest]) (*connect.Response[conveyorv1.CancelTaskResponse], error)

CancelTask cancels a scheduled, pending, or retry task durably. An active task instead receives a best-effort Cancel frame through its executing worker session: the handler's context is canceled and the aborted attempt is archived instead of retried; only a genuine success reported before the cancel lands still completes the task.

func (*AdminService) ClusterInfo 

func (s *AdminService) ClusterInfo(ctx context.Context, _ *connect.Request[conveyorv1.ClusterInfoRequest]) (*connect.Response[conveyorv1.ClusterInfoResponse], error)

ClusterInfo reports this node and its best-effort view of cluster peers.

func (*AdminService) DeleteCron 

func (s *AdminService) DeleteCron(ctx context.Context, request *connect.Request[conveyorv1.DeleteCronRequest]) (*connect.Response[conveyorv1.DeleteCronResponse], error)

DeleteCron removes a cron entry; deleting an absent id succeeds.

func (*AdminService) DeleteGroupConfig added in v0.3.0 

func (s *AdminService) DeleteGroupConfig(ctx context.Context, request *connect.Request[conveyorv1.DeleteGroupConfigRequest]) (*connect.Response[conveyorv1.DeleteGroupConfigResponse], error)

DeleteGroupConfig clears a group's override, reverting it to the queue-wide or global default. The sweeper picks up the change on its next tick.

func (*AdminService) DeleteQueueConcurrencyLimit added in v0.2.0 

func (s *AdminService) DeleteQueueConcurrencyLimit(ctx context.Context, request *connect.Request[conveyorv1.DeleteQueueConcurrencyLimitRequest]) (*connect.Response[conveyorv1.DeleteQueueConcurrencyLimitResponse], error)

DeleteQueueConcurrencyLimit clears a queue's concurrency limit and reverts the live queue grain to unbounded keys (a zero max-active in the change message).

func (*AdminService) DeleteQueueRateLimit 

func (s *AdminService) DeleteQueueRateLimit(ctx context.Context, request *connect.Request[conveyorv1.DeleteQueueRateLimitRequest]) (*connect.Response[conveyorv1.DeleteQueueRateLimitResponse], error)

DeleteQueueRateLimit clears a queue's override and reverts the live queue grain to the server's global default (a zero rate in the change message).

func (*AdminService) DeleteTask 

func (s *AdminService) DeleteTask(ctx context.Context, request *connect.Request[conveyorv1.DeleteTaskRequest]) (*connect.Response[conveyorv1.DeleteTaskResponse], error)

DeleteTask removes a task in any state except active.

func (*AdminService) DeleteWebhookWorker added in v0.4.0 

func (s *AdminService) DeleteWebhookWorker(ctx context.Context, request *connect.Request[conveyorv1.DeleteWebhookWorkerRequest]) (*connect.Response[conveyorv1.DeleteWebhookWorkerResponse], error)

DeleteWebhookWorker removes one registration; deleting an absent name succeeds.

func (*AdminService) ListConcurrencyLimits added in v0.2.0 

func (s *AdminService) ListConcurrencyLimits(ctx context.Context, _ *connect.Request[conveyorv1.ListConcurrencyLimitsRequest]) (*connect.Response[conveyorv1.ListConcurrencyLimitsResponse], error)

ListConcurrencyLimits returns every per-queue concurrency limit, ordered by queue name.

func (*AdminService) ListCron 

func (s *AdminService) ListCron(ctx context.Context, _ *connect.Request[conveyorv1.ListCronRequest]) (*connect.Response[conveyorv1.ListCronResponse], error)

ListCron returns all persisted cron entries ordered by id.

func (*AdminService) ListGroupConfigs added in v0.3.0 

func (s *AdminService) ListGroupConfigs(ctx context.Context, _ *connect.Request[conveyorv1.ListGroupConfigsRequest]) (*connect.Response[conveyorv1.ListGroupConfigsResponse], error)

ListGroupConfigs returns every per-group aggregation override, ordered by queue then group.

func (*AdminService) ListQueues 

func (s *AdminService) ListQueues(ctx context.Context, _ *connect.Request[conveyorv1.ListQueuesRequest]) (*connect.Response[conveyorv1.ListQueuesResponse], error)

ListQueues reports every known queue with its pause flag and per-state task counts.

func (*AdminService) ListRateLimits 

func (s *AdminService) ListRateLimits(ctx context.Context, _ *connect.Request[conveyorv1.ListRateLimitsRequest]) (*connect.Response[conveyorv1.ListRateLimitsResponse], error)

ListRateLimits returns every per-queue dispatch-rate override, ordered by queue name.

func (*AdminService) ListTasks 

func (s *AdminService) ListTasks(ctx context.Context, request *connect.Request[conveyorv1.ListTasksRequest]) (*connect.Response[conveyorv1.ListTasksResponse], error)

ListTasks pages through tasks newest first, optionally filtered by queue and state.

func (*AdminService) ListWebhookWorkers added in v0.4.0 

func (s *AdminService) ListWebhookWorkers(ctx context.Context, _ *connect.Request[conveyorv1.ListWebhookWorkersRequest]) (*connect.Response[conveyorv1.ListWebhookWorkersResponse], error)

ListWebhookWorkers returns every registration with its secrets redacted: they are write-only credentials, and a dashboard listing must never leak them.

func (*AdminService) ListWorkerSessions 

func (s *AdminService) ListWorkerSessions(_ context.Context, _ *connect.Request[conveyorv1.ListWorkerSessionsRequest]) (*connect.Response[conveyorv1.ListWorkerSessionsResponse], error)

ListWorkerSessions reports the worker sessions connected to this node.

func (*AdminService) PauseCron 

func (s *AdminService) PauseCron(ctx context.Context, request *connect.Request[conveyorv1.PauseCronRequest]) (*connect.Response[conveyorv1.PauseCronResponse], error)

PauseCron suspends materialization of one cron entry.

func (*AdminService) PauseQueue 

func (s *AdminService) PauseQueue(ctx context.Context, request *connect.Request[conveyorv1.PauseQueueRequest]) (*connect.Response[conveyorv1.PauseQueueResponse], error)

PauseQueue persists the pause flag and drains the live queue grain so dispatching stops immediately; queued work stays in the broker.

func (*AdminService) PauseWebhookWorker added in v0.4.0 

func (s *AdminService) PauseWebhookWorker(ctx context.Context, request *connect.Request[conveyorv1.PauseWebhookWorkerRequest]) (*connect.Response[conveyorv1.PauseWebhookWorkerResponse], error)

PauseWebhookWorker suspends delivery to one registration.

func (*AdminService) RescheduleTask added in v0.3.0 

func (s *AdminService) RescheduleTask(ctx context.Context, request *connect.Request[conveyorv1.RescheduleTaskRequest]) (*connect.Response[conveyorv1.RescheduleTaskResponse], error)

RescheduleTask moves a waiting (scheduled, pending, or retry) task's due time to a new instant. A future time leaves the task scheduled; a past or present time makes it due immediately and wakes its queue grain.

func (*AdminService) ResumeCron 

func (s *AdminService) ResumeCron(ctx context.Context, request *connect.Request[conveyorv1.ResumeCronRequest]) (*connect.Response[conveyorv1.ResumeCronResponse], error)

ResumeCron resumes materialization of one cron entry.

func (*AdminService) ResumeQueue 

func (s *AdminService) ResumeQueue(ctx context.Context, request *connect.Request[conveyorv1.ResumeQueueRequest]) (*connect.Response[conveyorv1.ResumeQueueResponse], error)

ResumeQueue clears the pause flag and triggers an immediate lease cycle on the queue grain.

func (*AdminService) ResumeWebhookWorker added in v0.4.0 

func (s *AdminService) ResumeWebhookWorker(ctx context.Context, request *connect.Request[conveyorv1.ResumeWebhookWorkerRequest]) (*connect.Response[conveyorv1.ResumeWebhookWorkerResponse], error)

ResumeWebhookWorker resumes delivery to one paused registration.

func (*AdminService) RunTask 

func (s *AdminService) RunTask(ctx context.Context, request *connect.Request[conveyorv1.RunTaskRequest]) (*connect.Response[conveyorv1.RunTaskResponse], error)

RunTask makes a scheduled, pending, retry, or archived task due immediately and wakes its queue grain.

func (*AdminService) SetGroupConfig added in v0.3.0 

func (s *AdminService) SetGroupConfig(ctx context.Context, request *connect.Request[conveyorv1.SetGroupConfigRequest]) (*connect.Response[conveyorv1.SetGroupConfigResponse], error)

SetGroupConfig persists a group's aggregation override. The group sweeper reads the new value on its next tick (firing is poll-based), so no live actor wake is needed, unlike the queue-grain rate-limit path. An empty group sets the queue-wide default applied to every group on the queue without its own override.

func (*AdminService) SetQueueConcurrencyLimit added in v0.2.0 

func (s *AdminService) SetQueueConcurrencyLimit(ctx context.Context, request *connect.Request[conveyorv1.SetQueueConcurrencyLimitRequest]) (*connect.Response[conveyorv1.SetQueueConcurrencyLimitResponse], error)

SetQueueConcurrencyLimit persists a queue's per-key concurrency limit and pushes the new value to the live queue grain so it takes effect immediately.

func (*AdminService) SetQueueRateLimit 

func (s *AdminService) SetQueueRateLimit(ctx context.Context, request *connect.Request[conveyorv1.SetQueueRateLimitRequest]) (*connect.Response[conveyorv1.SetQueueRateLimitResponse], error)

SetQueueRateLimit persists a queue's dispatch-rate override and pushes the new values to the live queue grain so the limit takes effect immediately.

func (*AdminService) UpsertCron 

func (s *AdminService) UpsertCron(ctx context.Context, request *connect.Request[conveyorv1.UpsertCronRequest]) (*connect.Response[conveyorv1.UpsertCronResponse], error)

UpsertCron validates and persists a cron entry by id.

func (*AdminService) UpsertWebhookWorker added in v0.4.0 

func (s *AdminService) UpsertWebhookWorker(ctx context.Context, request *connect.Request[conveyorv1.UpsertWebhookWorkerRequest]) (*connect.Response[conveyorv1.UpsertWebhookWorkerResponse], error)

UpsertWebhookWorker creates or replaces one registration and nudges the webhook manager so the change applies immediately.

func (*AdminService) WatchEvents added in v0.2.0 

func (s *AdminService) WatchEvents(ctx context.Context, request *connect.Request[conveyorv1.WatchEventsRequest], stream *connect.ServerStream[conveyorv1.TaskEvent]) error

WatchEvents streams task lifecycle transitions to the caller until the client disconnects. It subscribes to the node-local event bus with the request's queue and event-type filters; a watcher too slow to keep up has events dropped (counted by the events.dropped metric), never stalling dispatch.

type SessionLister 

type SessionLister interface {
	// Sessions returns a snapshot of the live worker sessions.
	Sessions() []SessionSnapshot
}

SessionLister reports the worker sessions connected to this node. The WorkerService implements it; AdminService surfaces it for the dashboard.

type SessionSnapshot 

type SessionSnapshot struct {
	// ID is the server-assigned session id.
	ID string
	// Queues are the queue names the worker serves, sorted.
	Queues []string
	// Concurrency is the worker's declared concurrency.
	Concurrency int32
	// SDKVersion is the worker's reported SDK version.
	SDKVersion string
	// ConnectedAt is when the session's Hello was accepted.
	ConnectedAt time.Time
}

SessionSnapshot is the externally visible description of one connected worker session, surfaced by the AdminService worker-topology view.

type TaskService 

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

TaskService serves the enqueue-side API.

func NewTaskService 

func NewTaskService(engine *actors.Engine, taskLog broker.Broker, timeSource clock.Clock, defaultMaxRetry int32) *TaskService

NewTaskService assembles the enqueue-side API service.

func (*TaskService) Enqueue 

func (s *TaskService) Enqueue(ctx context.Context, request *connect.Request[conveyorv1.EnqueueRequest]) (*connect.Response[conveyorv1.EnqueueResponse], error)

Enqueue durably commits one task and reports its initial state.

func (*TaskService) EnqueueBatch 

func (s *TaskService) EnqueueBatch(ctx context.Context, request *connect.Request[conveyorv1.EnqueueBatchRequest]) (*connect.Response[conveyorv1.EnqueueBatchResponse], error)

EnqueueBatch commits many tasks in one call. Items fail independently: each result carries either the committed task or that item's error.

func (*TaskService) EnqueueTx added in v0.4.0 

func (s *TaskService) EnqueueTx(ctx context.Context, request *connect.Request[conveyorv1.EnqueueTxRequest]) (*connect.Response[conveyorv1.EnqueueTxResponse], error)

EnqueueTx commits every task atomically: either all are enqueued or none are. A validation failure on any task rejects the whole request before the broker is touched; a broker failure rolls the whole set back. On success it returns the committed tasks in request order.

func (*TaskService) GetTask 

func (s *TaskService) GetTask(ctx context.Context, request *connect.Request[conveyorv1.GetTaskRequest]) (*connect.Response[conveyorv1.GetTaskResponse], error)

GetTask returns the current state of one task.

type WebhookService added in v0.4.0 

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

WebhookService serves the asynchronous-completion callbacks of webhook workers. It is authenticated by per-delivery lease tokens, never by API bearer tokens, so it mounts without the bearer interceptor: an endpoint holding a token may heartbeat and resolve exactly one delivery.

func NewWebhookService added in v0.4.0 

func NewWebhookService(engine *actors.Engine, taskLog broker.Broker) *WebhookService

NewWebhookService assembles the webhook callback service.

func (*WebhookService) Heartbeat added in v0.4.0 

func (s *WebhookService) Heartbeat(ctx context.Context, request *connect.Request[conveyorv1.WebhookHeartbeatRequest]) (*connect.Response[conveyorv1.WebhookHeartbeatResponse], error)

Heartbeat extends one asynchronously completing delivery's lease.

func (*WebhookService) ReportResult added in v0.4.0 

func (s *WebhookService) ReportResult(ctx context.Context, request *connect.Request[conveyorv1.WebhookReportResultRequest]) (*connect.Response[conveyorv1.WebhookReportResultResponse], error)

ReportResult resolves one asynchronously completing delivery.

type WorkerService 

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

WorkerService serves the worker session protocol: one bidirectional stream per worker process, bridged to a per-session gateway actor.

func NewWorkerService 

func NewWorkerService(engine *actors.Engine, logger *slog.Logger, timeSource clock.Clock) *WorkerService

NewWorkerService assembles the worker session service.

func (*WorkerService) ActiveSessions 

func (s *WorkerService) ActiveSessions() int64

ActiveSessions returns the number of live worker sessions.

func (*WorkerService) DrainSessions 

func (s *WorkerService) DrainSessions(ctx context.Context) error

DrainSessions ends every live worker session and rejects new ones, then waits until each closed session has stopped its gateway — releasing all in-flight tasks for immediate redelivery — or the context lapses. The server runs it before stopping the engine: once the actor system begins stopping it rejects all user messages, and gateways can no longer process drain requests.

func (*WorkerService) Session 

func (s *WorkerService) Session(ctx context.Context, stream *connect.BidiStream[conveyorv1.WorkerMessage, conveyorv1.ServerMessage]) error

Session implements the worker stream protocol: the first frame must be Hello, the server answers Welcome, and every later frame is validated and forwarded to the session's gateway actor. Any protocol violation ends the stream; the deferred gateway stop releases all in-flight tasks for immediate redelivery, whatever the reason the stream ended.

func (*WorkerService) Sessions 

func (s *WorkerService) Sessions() []SessionSnapshot

Sessions returns a snapshot of the live worker sessions on this node, sorted by session id for a stable ordering.

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.