operatorsvc

package
v0.109.2 Latest Latest
Warning

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

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

Documentation

Overview

Package operatorsvc holds the engine-side logic of an operator session: registering the operator and its worker, opening and closing the session that makes the worker live, applying action set deltas, authorizing an operator and its workers, and opening durable invocations.

It is shared by every operator host. internal/services/grpcoperator serves it over OperatorService to operators that run outside the engine, and internal/operator/hostinproc calls the same functions for operators hosted inside the dispatcher, so the two hosts differ only in how assigned actions reach the operator and in how the caller is authenticated.

The API takes no OperatorService protocol messages: callers pass plain values and get repository rows back. Errors carry connect codes because they are returned to RPC callers unchanged; in-process callers can ignore the codes.

Index

Constants

View Source 
const (

	// HandshakeHoldLimit bounds the invocation responses held while the register-worker ack is
	// awaited. A session that exceeds it is refused rather than buffering without end.
	HandshakeHoldLimit = 256

	// RetainedResponseLimit bounds what the receive loop retains for Recv over the life of the
	// invocation: responses queued for delivery plus entry completions held for an ack or an
	// expectation. The handshake bound covers a burst before the operator can read; this one
	// allows for an operator that reads but is momentarily behind a fan-out of completions, and
	// for entries the engine completes that the operator never registers. A channel past it
	// fails with ErrChannelClosed rather than buffer without end inside the engine process.
	RetainedResponseLimit = 4 * HandshakeHoldLimit
)
View Source 
const (
	// DefaultMaxListenStreamsPerOperator caps the Listen streams one operator may hold open on
	// this engine replica at once. Every stream is a live worker with its own session and
	// action set, so the cap bounds what one tenant token can allocate here.
	DefaultMaxListenStreamsPerOperator = 100

	// DefaultMaxActionsPerOperator caps the action links held across all workers of one
	// operator. Deltas that would exceed it are refused with ResourceExhausted. The cap is
	// checked by the repository inside the delta's transaction, so it holds across sessions
	// and across engine replicas.
	DefaultMaxActionsPerOperator = 1_000_000

	// MaxActionsPerDelta caps the ids (adds plus removes) one delta may carry.
	MaxActionsPerDelta = 1000
)

Variables

View Source 
var (
	ErrRequestInFlight = operator.ErrRequestInFlight
	ErrChannelClosed   = operator.ErrChannelClosed
	ErrSessionEnded    = operator.ErrSessionEnded
)

The channel's errors are the contract's, so an operator sees the same sentinels whichever host opened its channel.

View Source 
var ErrNoStream = errors.New("operatorsvc: session has no stream to send on")

ErrNoStream is returned by Send on a session whose delivery is a handler: there is no stream to write protocol messages onto.

Functions

func WithTenant 

func WithTenant(ctx context.Context, tenant *sqlcv1.Tenant) context.Context

WithTenant puts the tenant on the context the way the gRPC auth middleware does, so the dispatcher's and the admin service's handlers see the same value whichever host called them.

Types

type ActionHandler 

type ActionHandler = operator.ActionHandler

ActionHandler receives assigned actions by direct call. It is the in-process delivery: the call runs on the dispatcher's delivery goroutine and its error requeues the task. It is the contract's handler type, so an operator written against pkg/operator is hosted here unchanged.

type CloseOpt 

type CloseOpt func(*closeOpts)

func WithoutPause 

func WithoutPause() CloseOpt

WithoutPause closes the session without pausing its worker. It is what a session whose operator pauses for itself uses: a gRPC operator pauses on its stream before it hangs up, and a stream that ends unexpectedly must leave the worker assignable so the operator's next connection resumes a worker that can be given work.

type DispatcherBackend 

type DispatcherBackend interface {
	AddOperatorStreamSession(ctx context.Context, workerId uuid.UUID, sessionId uuid.UUID, stream *rpcstream.Sender[v1contracts.OperatorListenResponse]) StreamSession
	AddOperatorSession(workerId uuid.UUID, sessionId uuid.UUID, handler ActionHandler) HandlerSession
	NotifyNewWorker(ctx context.Context, tenant *sqlcv1.Tenant, workerId uuid.UUID)
	SendStepActionEvent(ctx context.Context, req *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)
	RegisterDurableTask(ctx context.Context, externalId uuid.UUID) (chan<- *v1contracts.DurableTaskRequest, <-chan *v1contracts.DurableTaskResponse, error)
}

DispatcherBackend is what the service needs from the local dispatcher: session registration for either delivery, scheduler notification, task events, and the channel-backed durable session.

func NewDispatcherBackend 

func NewDispatcherBackend(d *dispatcher.DispatcherImpl) DispatcherBackend

NewDispatcherBackend adapts the local dispatcher to what an operator session needs from it.

type DurableChannel 

type DurableChannel = operator.DurableChannel

DurableChannel is the contract's channel; the in-process implementation is what OpenDurable returns.

type HandlerSession 

type HandlerSession interface {
	SetPaused(paused bool)
	Release()
}

HandlerSession is the dispatcher session handle for a handler-backed session; see dispatcher.OperatorHandlerSession. SetPaused is the same as on StreamSession.

type OpenOpts 

type OpenOpts struct {
	// Stream is the guarded sender of the Listen stream the dispatcher fans actions out on.
	Stream *rpcstream.Sender[v1contracts.OperatorListenResponse]

	// Handler receives assigned actions by direct call.
	Handler ActionHandler

	// Worker is the worker row, when the caller already read it (the gRPC path reads it to
	// authorize the session). It is loaded here when nil.
	Worker *sqlcv1.GetWorkerForEngineRow
}

OpenOpts describes how a session delivers assigned actions to its operator. Exactly one of Stream and Handler is set: Stream encodes actions onto a gRPC stream, Handler receives them by direct call in the engine process.

type OperatorStore 

type OperatorStore interface {
	GetOperatorById(ctx context.Context, operatorId uuid.UUID) (*sqlcv1.V1Operator, error)
	UpsertOperator(ctx context.Context, tenantId uuid.UUID, opts repository.UpsertOperatorOpts) (*sqlcv1.V1Operator, error)
	UpdateOperator(ctx context.Context, tenantId, operatorId uuid.UUID, opts repository.UpdateOperatorOpts) (*sqlcv1.V1Operator, error)
}

OperatorStore is the subset of repository.OperatorRepository the service uses. It is narrow so tests can substitute a double without stubbing the whole repository tree.

type Opt 

type Opt func(*opts)

func WithAnalytics 

func WithAnalytics(a analytics.Analytics) Opt

func WithDispatcherBackend 

func WithDispatcherBackend(d DispatcherBackend) Opt

WithDispatcherBackend sets the local dispatcher sessions are registered with. Required.

func WithDispatcherId 

func WithDispatcherId(id uuid.UUID) Opt

WithDispatcherId sets the id of the local dispatcher; every operator worker whose session is opened here is pinned to it, since the delivery path lives here. Required.

func WithLogger 

func WithLogger(l *zerolog.Logger) Opt

func WithMaxActionsPerOperator 

func WithMaxActionsPerOperator(n int64) Opt

WithMaxActionsPerOperator caps the action links held across all workers of one operator; zero disables the cap.

func WithMaxListenStreamsPerOperator 

func WithMaxListenStreamsPerOperator(n int) Opt

WithMaxListenStreamsPerOperator caps the stream-backed sessions one operator may hold open on this replica; zero disables the cap.

func WithNotifyInterval 

func WithNotifyInterval(d time.Duration) Opt

WithNotifyInterval sets the window scheduler notifications are folded into.

func WithOperatorCacheTTL 

func WithOperatorCacheTTL(d time.Duration) Opt

WithOperatorCacheTTL sets how long an authorized operator row is reused.

func WithOperatorStore 

func WithOperatorStore(s OperatorStore) Opt

WithOperatorStore sets the operator rows the service registers and authorizes. Required.

func WithValidator 

func WithValidator(v validator.Validator) Opt

func WithWorkerStore 

func WithWorkerStore(s WorkerStore) Opt

WithWorkerStore sets the worker rows the service creates, activates and links actions to. Required.

type RegisterOpts 

type RegisterOpts struct {
	// OperatorId is an existing operator row, as claimed by the in-process claimer. Name, Kind
	// and LeasingManager are taken from the row and no upsert happens; the row's worker_id is pointed
	// at the session's worker, which is how ClaimOperators recognises the assignment on later
	// polls.
	OperatorId *uuid.UUID

	// Name is the operator name, unique per (tenant, kind). Ignored when OperatorId is set.
	Name string

	// Kind is what the operator is. Only GRPC rows, contract operators, are upserted by name;
	// the DAG operator's rows are the engine's own and are registered by OperatorId. Ignored
	// when OperatorId is set.
	Kind sqlcv1.V1OperatorKind

	// LeasingManager is who keeps the operator alive, and is what the row is set to whether the
	// upsert creates or finds it: SELF for a registration that holds its own stream or leaser,
	// DISPATCHER for a row a dispatcher should claim through the claimer. Ignored when
	// OperatorId is set.
	LeasingManager sqlcv1.V1OperatorLeasingManager

	// IsExemptFromLimits leaves the worker out of the tenant's worker and slot limits. It is a
	// hosting fact: the in-process host sets it for every worker it creates, the wire never
	// does.
	IsExemptFromLimits bool

	// WorkerName names the worker row. It defaults to the operator name, which is what one
	// worker per connection looks like in the dashboard.
	WorkerName string

	// SlotConfig maps slot type to max units, defaulting to {"default": 100}.
	SlotConfig map[string]int32

	Labels      map[string]*contracts.WorkerLabels
	RuntimeInfo *contracts.RuntimeInfo

	// ResumeWorkerId names a worker of this operator to reuse instead of creating one. A
	// worker that no longer exists, or that belongs to another operator, is replaced by a new
	// one rather than refused, since a caller only ever learns about its own workers.
	ResumeWorkerId *uuid.UUID
}

RegisterOpts describes the operator to upsert, or the existing row to register under, and the worker to back this session.

type Registration 

type Registration struct {
	Operator *sqlcv1.V1Operator

	TenantId   uuid.UUID
	OperatorId uuid.UUID
	WorkerId   uuid.UUID

	// Resumed reports whether WorkerId is the worker ResumeWorkerId named.
	Resumed bool
}

Registration is the identity the engine assigned to the caller.

type Service 

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

func New 

func New(fs ...Opt) (*Service, error)

func (*Service) AuthorizeOperator 

func (s *Service) AuthorizeOperator(ctx context.Context, tenant *sqlcv1.Tenant, operatorId uuid.UUID) (*sqlcv1.V1Operator, error)

AuthorizeOperator resolves operatorId and checks that it is a self-leased GRPC operator owned by tenant. Lookups are cached for the cache's TTL; misses and errors are never cached, so a freshly created operator is visible on the next call. Only such operators are reachable from outside the engine, which is the only caller that authorizes today: a DAG row is the engine's own, and a row the engine leases is driven by the claimer, never over the wire.

func (*Service) AuthorizeWorker 

func (s *Service) AuthorizeWorker(ctx context.Context, tenant *sqlcv1.Tenant, operatorId uuid.UUID, workerId uuid.UUID) (*sqlcv1.GetWorkerForEngineRow, error)

AuthorizeWorker checks that workerId names a worker owned by the operator. It is the guard on every call that carries a worker id, so an operator cannot act on another operator's worker. A worker that does not exist is reported the same way as one owned by someone else.

func (*Service) Cleanup 

func (s *Service) Cleanup() error

Cleanup stops the operator cache's expiry goroutine.

func (*Service) DispatcherId 

func (s *Service) DispatcherId() uuid.UUID

DispatcherId is the dispatcher every session opened here pins its worker to.

func (*Service) ListenStreamCount 

func (s *Service) ListenStreamCount(operatorId uuid.UUID) int

ListenStreamCount reports the open stream-backed sessions of the operator on this replica.

func (*Service) OpenSession 

func (s *Service) OpenSession(ctx context.Context, tenant *sqlcv1.Tenant, op *sqlcv1.V1Operator, workerId uuid.UUID, opts OpenOpts) (*Session, error)

OpenSession makes the worker live: it re-pins the worker to this dispatcher, refreshes its action hash when a previous session left it pending, activates it under a fresh session id that doubles as the listener fence, registers the delivery with the dispatcher and notifies the scheduler. Stream-backed sessions are also counted against the per-operator stream cap; in-process sessions hold no stream, so only the action budget applies to them.

func (*Service) PauseWorker 

func (s *Service) PauseWorker(ctx context.Context, tenant *sqlcv1.Tenant, workerId uuid.UUID, paused bool) error

PauseWorker stops the scheduler assigning to the worker, or lets it be assigned to again. It returns once the change is committed. Register clears the pause when it resumes a worker, so an operator that crashed while paused comes back assignable. A live session pauses through Session.Pause, which also stops the session delivering.

func (*Service) Register 

func (s *Service) Register(ctx context.Context, tenant *sqlcv1.Tenant, opts RegisterOpts) (Registration, error)

Register upserts the operator by (tenant, name, kind), or loads the row OperatorId names, and creates the worker for this session, or resumes the worker named by ResumeWorkerId when it still belongs to the operator. The worker starts with no actions: the caller links them on its session.

func (*Service) SendStepActionEvent 

func (s *Service) SendStepActionEvent(ctx context.Context, tenant *sqlcv1.Tenant, op *sqlcv1.V1Operator, ev *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)

SendStepActionEvent reports task progress for an action delivered to one of the operator's workers. The event's worker must belong to the operator; the dispatcher then handles it exactly like an SDK worker's event.

type Session 

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

Session is one live worker: its dispatcher routing entry, its listener fence on the worker row, its action budget and its scheduler notifications. It lives from OpenSession until Close.

The methods that drive the session (Heartbeat, ApplyDelta, Send) are meant to be called from the one goroutine that owns the session's protocol loop, which is how the throttles read as one sequence; Pause and Close are safe to call from anywhere, and Close runs once.

func (*Session) ApplyDelta 

func (ss *Session) ApplyDelta(ctx context.Context, add, remove []string) (bool, error)

ApplyDelta validates and applies one change to the worker's action set and asks the scheduler to reload it. It reports whether the set changed; a delta that only repeats what the worker already has needs no notification, and the caller can still acknowledge it.

func (*Session) Close 

func (ss *Session) Close(ctx context.Context, fs ...CloseOpt) error

Close ends the session: pause, then drain, then deactivate. Pausing stops the scheduler assigning new work, releasing the dispatcher session stops anything further being delivered, and the deactivation, fenced on this session's id, marks the worker inactive. A host that waits for its operator's in-flight work does so between Pause and Close. A hash refresh the notifier still owed is done here, so the row never keeps a pending hash past its session.

The pause, the refresh and the deactivation run detached from ctx because the common exit is the operator being gone, at which point ctx is already cancelled. Close runs once; later calls return the first result.

func (*Session) Fin 

func (ss *Session) Fin() <-chan bool

Fin fires when the dispatcher wants a stream-backed session hung up. A handler-backed session returns nil, which never selects: the dispatcher has no stream to reclaim.

func (*Session) Heartbeat 

func (ss *Session) Heartbeat(ctx context.Context, at time.Time) error

Heartbeat records that the operator is alive. Writes are throttled to one per second, so a host that heartbeats faster costs nothing; a throttled call reports no error.

func (*Session) OpenDurable 

func (ss *Session) OpenDurable(ctx context.Context, taskExternalId uuid.UUID, invocation int32) (DurableChannel, error)

OpenDurable opens one durable invocation's pipe for the session's worker and runs the register-worker handshake: the first request names the worker running the invocation, and the engine's ack is consumed here so Recv only ever returns invocation traffic. The engine routes responses to the task as soon as it is registered, before the ack, so invocation traffic that arrives during the handshake (a completion restored for a resumed invocation, for instance) is held for the channel, under the same ack-before-entry ordering, up to HandshakeHoldLimit responses. ctx bounds the handshake; the invocation itself is detached from it and ends on Close.

func (*Session) Operator 

func (ss *Session) Operator() *sqlcv1.V1Operator

Operator is the operator row the session belongs to.

func (*Session) Pause 

func (ss *Session) Pause(ctx context.Context, paused bool) error

Pause stops the scheduler assigning to the session's worker, or lets it be assigned to again. It returns once the change is committed, so a host that pauses before draining knows no further work will arrive: the dispatcher session stops delivering before the pause is written, and an action the scheduler assigned in the meantime goes back to the queue rather than to the operator. Resuming lifts the pause in the opposite order, so nothing is refused once the scheduler may assign again. The write is fenced on the session id like the deactivation: a session superseded by a newer one on the same worker leaves the worker's scheduling state to it.

func (*Session) Send 

func (ss *Session) Send(ctx context.Context, msg *v1contracts.OperatorListenResponse) error

Send writes a protocol message on a stream-backed session's stream, serialised with the dispatcher's own action sends.

func (*Session) SendStepActionEvent 

func (ss *Session) SendStepActionEvent(ctx context.Context, ev *contracts.StepActionEvent) error

SendStepActionEvent reports task progress for an action delivered to the session's worker, on the tenant-scoped path in-engine operators use: the tenant is put on the context and the dispatcher handles the event exactly like an SDK worker's. An empty WorkerId is filled from the session; another worker's id is refused, since the session only ever speaks for its own worker.

func (*Session) SessionId 

func (ss *Session) SessionId() uuid.UUID

SessionId is the id the worker's listener fence and the dispatcher session are keyed on.

func (*Session) Tenant 

func (ss *Session) Tenant() *sqlcv1.Tenant

Tenant is the tenant the session belongs to.

func (*Session) WorkerId 

func (ss *Session) WorkerId() uuid.UUID

WorkerId is the worker this session makes live.

type StreamSession 

type StreamSession interface {
	Fin() <-chan bool
	Send(ctx context.Context, msg *v1contracts.OperatorListenResponse) error
	SetPaused(paused bool)
	Release()
}

StreamSession is the dispatcher session handle for a stream-backed session; see dispatcher.OperatorStreamSession. SetPaused(true) makes the dispatcher return every action it is asked to deliver to the queue instead, so a pause the scheduler has not observed yet still delivers nothing.

type WorkerStore 

type WorkerStore interface {
	CreateNewWorker(ctx context.Context, tenantId uuid.UUID, opts *repository.CreateWorkerOpts) (*sqlcv1.Worker, error)
	GetWorkerForEngine(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID) (*sqlcv1.GetWorkerForEngineRow, error)
	UpdateWorker(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID, opts *repository.UpdateWorkerOpts) (*sqlcv1.Worker, error)
	ActivateWorkerListener(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID, sessionId uuid.UUID) (*sqlcv1.Worker, error)
	DeactivateWorkerListener(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID, sessionId uuid.UUID) (*sqlcv1.Worker, error)
	UpdateWorkerHeartbeat(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID, lastHeartbeatAt time.Time) error
	UpsertWorkerLabels(ctx context.Context, workerId uuid.UUID, opts []repository.UpsertWorkerLabelOpts) ([]*sqlcv1.WorkerLabel, error)
	PauseWorkerForListener(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID, sessionId uuid.UUID, paused bool) error
	ApplyWorkerActionsDelta(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID, add, remove []string, maxOperatorLinks int64) (added, removed int, err error)
	RefreshWorkerActionHash(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID) error
	CountOperatorWorkerActions(ctx context.Context, tenantId uuid.UUID, operatorId uuid.UUID) (int64, error)
}

WorkerStore is the subset of repository.WorkerRepository the service uses.

Source Files

View all Source files

Directories

Path Synopsis
Package operatorsvctest holds in-memory doubles for the stores and the dispatcher an operator session talks to.
 Package operatorsvctest holds in-memory doubles for the stores and the dispatcher an operator session talks to.

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.