Documentation
¶
Overview ¶
Package hostinproc is the in-process operator host: it implements pkg/operator.Host inside the dispatcher process over internal/services/operatorsvc, the same session logic the OperatorService handlers serve to operators that run outside the engine. A session opened here gets assigned actions by direct call, applies action deltas synchronously, reports events and opens durable invocations on the tenant-scoped in-engine paths, and is kept alive by one host-wide bulk heartbeat per tick.
The host is engine-internal: it runs only where a dispatcher runs and is wired from cmd/hatchet-engine/engine. Contract operators are hosted here or over gRPC as a deployment choice; the DAG operator is hosted here only, for lifecycle.
Index ¶
- type AdminService
- type HeartbeatStore
- type Host
- type Opt
- func WithAdminService(a AdminService) Opt
- func WithHeartbeatInterval(d time.Duration) Opt
- func WithHeartbeatStore(s HeartbeatStore) Opt
- func WithLogger(l *zerolog.Logger) Opt
- func WithService(svc *operatorsvc.Service) Opt
- func WithTenantStore(s TenantStore) Opt
- func WithWorkflowStore(s WorkflowStore) Opt
- type TenantStore
- type WorkflowStore
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AdminService ¶
type AdminService interface {
PutWorkflow(ctx context.Context, req *v1.CreateWorkflowVersionRequest) (*v1.CreateWorkflowVersionResponse, error)
}
AdminService puts a workflow for the tenant on the context, the way the engine's admin service does for a gRPC caller; internal/services/admin/v1.AdminService satisfies it.
type HeartbeatStore ¶
type HeartbeatStore interface {
UpdateWorkerHeartbeats(ctx context.Context, workerIds []uuid.UUID, lastHeartbeatAt time.Time) error
}
HeartbeatStore is the bulk heartbeat write.
type Host ¶
type Host struct {
// contains filtered or unexported fields
}
Host opens in-process sessions and heartbeats their workers until they close.
func New ¶
New builds a host and starts its heartbeat ticker; Close stops it. Sessions are closed by whoever opened them, before the host.
func (*Host) Close ¶
func (h *Host) Close()
Close stops the heartbeat ticker. Open sessions keep working but are no longer kept alive, so they are closed first.
func (*Host) Open ¶
func (h *Host) Open(ctx context.Context, id operator.Identity, o operator.OpenOpts) (operator.Session, error)
Open implements operator.Host. It registers the operator (an existing row by id, or an upsert by name, kind and leasing manager) and its worker, opens a handler-backed engine session under a fresh session id that is both the dispatcher's key and the worker row's listener fence, links the initial action set, and adds the worker to the heartbeat set. A failure after the session was opened closes it again, so the caller never inherits a half-open worker.
Every worker this host creates is exempt from the tenant's worker and slot limits: an operator hosted inside the engine is infrastructure that runs whether or not the tenant runs workers of its own, whichever kind or leasing manager its row has.
func (*Host) SessionCount ¶
SessionCount is the number of open sessions.
type Opt ¶
type Opt func(*opts)
func WithAdminService ¶
func WithAdminService(a AdminService) Opt
WithAdminService lets sessions put workflows. Optional: without it Session.PutWorkflow reports ErrNotSupported; with it, WithWorkflowStore is required too.
func WithHeartbeatInterval ¶
WithHeartbeatInterval sets how often the bulk heartbeat runs.
func WithHeartbeatStore ¶
func WithHeartbeatStore(s HeartbeatStore) Opt
WithHeartbeatStore sets the bulk heartbeat write. Required.
func WithLogger ¶
func WithService ¶
func WithService(svc *operatorsvc.Service) Opt
WithService sets the engine session logic the host opens sessions over. Required.
func WithTenantStore ¶
func WithTenantStore(s TenantStore) Opt
WithTenantStore sets where the tenant a session belongs to is read from. Required.
func WithWorkflowStore ¶
func WithWorkflowStore(s WorkflowStore) Opt
WithWorkflowStore sets where the steps of a put workflow are read from.
type TenantStore ¶
type TenantStore interface {
GetTenantByID(ctx context.Context, tenantId uuid.UUID) (*sqlcv1.Tenant, error)
}
TenantStore reads the tenant row a session belongs to; the tenant is put on every context the host hands the engine, the way the gRPC auth middleware does for a token.
type WorkflowStore ¶
type WorkflowStore interface {
ListStepsByWorkflowVersionId(ctx context.Context, tenantId uuid.UUID, workflowVersionId uuid.UUID) ([]*sqlcv1.ListStepsByWorkflowVersionIdsRow, error)
}
WorkflowStore lists the steps the engine stored for a workflow version, which is where the action ids a put workflow derives are read from: the stored ids are the normalized ones. repository.WorkflowRepository satisfies it.
Source Files