tasks

package
v1.52.8 Latest Latest
Warning

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

 Go to latest Published: Jul 30, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package tasks provides the Hanzo Tasks client for Go applications.

Drop-in replacement for Base's Tasks() / Cron(): 

// Before (Base cron):
e.App.Tasks().Add("settlement", "*/30 * * * * *", func() { ... })

// After (Hanzo Tasks):
tasks.Default().Add("settlement", "30s", func() { ... })

Add() accepts both Go duration strings ("30s", "5m", "1h", "24h") and standard 5-field cron expressions ("0 3 * * *", "0 0 5 1,4,7,10 *", "*/5 * * * *"). Anything that parses as a Go duration is treated as an interval; anything else is treated as a cron expression.

If zapAddr is set, schedules and one-shot tasks run as durable Hanzo Tasks over the ZAP binary transport (retries, dead letter, audit trail). The submit path stamps opStartWorkflow; the schedule path stamps opCreateSchedule — the same opcodes the full SDK client and the server dispatch on. If zapAddr is empty, everything runs locally via a goroutine timer (dev mode, same behaviour as cron but no persistence).

Package tasks provides the Hanzo Tasks client for Go applications.

Two methods, two use cases: 

client := tasks.New(os.Getenv("TASKS_ZAP"), nil)
client.Add("settlement.process", "30s", fn)   // recurring schedule (duration)
client.Add("audit.archive", "0 3 * * *", fn)  // recurring schedule (cron)
client.Now("webhook.deliver", payload)         // fire once immediately

Transport: ZAP (binary, low-latency) or local goroutine. When TASKS_ZAP is set, tasks submit durably over the ZAP binary protocol. When it is empty, tasks run locally via goroutine timers (dev mode).

Integration with Hanzo Base: 

app.Tasks().Add("cleanup", "1h", fn)
app.Tasks().Now("email.send", payload)

Index

Constants

View Source 
const (
	// Worker → server (Call). Existing 0x00A2..0x00A5 declared in
	// pkg/sdk/client/transport.go remain authoritative.
	OpcodeSubscribeWorkflowTasks uint16 = 0x00A0
	OpcodeSubscribeActivityTasks uint16 = 0x00A1
	OpcodeUnsubscribeTasks       uint16 = 0x00A6

	// Server → worker (Send). 0x00B2 (DeliverActivityResult) is retired:
	// Phase-2a activities advance the run by replay, not a result push.
	OpcodeDeliverWorkflowTask  uint16 = 0x00B0
	OpcodeDeliverActivityTask  uint16 = 0x00B1
	OpcodeDeliverCancelRequest uint16 = 0x00B3
	OpcodeDeliverQuery         uint16 = 0x00B4

	// Worker → server (Call). Query response.
	OpcodeRespondQuery uint16 = 0x00C4
)

Variables

View Source 
var ErrNoWorkersSubscribed = fmt.Errorf("no workers subscribed to task queue")

ErrNoWorkersSubscribed is returned by QueryWorkflow when no worker is subscribed to the workflow's task queue. Callers must surface this as a 503-class condition; there is no engine-side fallback.

Functions

func SetDefault added in v1.9.4 

func SetDefault(c *Client)

SetDefault installs the process-wide task client. main() should call this once during boot. Subsequent callers use Default() to dispatch tasks.

Types

type BatchOperation added in v1.38.0 

type BatchOperation struct {
	BatchId    string `json:"batchId"`
	Namespace  string `json:"namespace"`
	Operation  string `json:"operation"` // BATCH_OPERATION_TYPE_TERMINATE|CANCEL|SIGNAL|RESET
	Reason     string `json:"reason"`
	Query      string `json:"query"` // visibility query — "WorkflowType='X'"
	State      string `json:"state"` // BATCH_OPERATION_STATE_RUNNING|COMPLETED|FAILED
	StartTime  string `json:"startTime"`
	CloseTime  string `json:"closeTime,omitempty"`
	TotalCount int64  `json:"totalOperationCount"`
	DoneCount  int64  `json:"completeOperationCount"`
}

BatchOperation — bulk terminate/cancel/signal across many executions.

type Client 

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

Client manages both one-shot tasks and recurring schedules.

func Default added in v1.9.4 

func Default() *Client

Default returns the process-wide client. If SetDefault was never called, lazily creates a client from the TASKS_ZAP env var so callers that register schedules before main() finished wiring still work. main() may replace this with a handler-bound client via SetDefault.

func New 

func New(zapAddr string, handler Handler) *Client

New creates a Client. If zapAddr is empty, everything runs locally. If zapAddr is set, tasks submit durably over ZAP.

func (*Client) Add 

func (c *Client) Add(name, spec string, fn func()) error

Add registers a recurring task.

spec is either a Go duration ("30s", "5m", "1h") or a standard 5-field cron expression ("0 3 * * *", "*/5 * * * *", "0 14 8 * *"). The fn runs on that cadence. If zapAddr is set, creates a durable Hanzo Tasks schedule so retries, dead-letter and audit are handled server-side. Otherwise runs locally. 

tasks.Default().Add("settlement.process", "30s", func() { ... })
tasks.Default().Add("audit.archive", "0 3 * * *", func() { ... })

func (*Client) Now 

func (c *Client) Now(taskType string, payload map[string]any) error

Now submits a one-shot task for durable execution over ZAP. When no server is configured, or the submit fails, the handler runs directly.

func (*Client) Stop 

func (c *Client) Stop()

Stop cancels all local schedules and closes ZAP connection. Call on shutdown.

type Deployment added in v1.38.0 

type Deployment struct {
	Name           string              `json:"name"`
	Namespace      string              `json:"namespace"`
	Description    string              `json:"description,omitempty"`
	OwnerEmail     string              `json:"ownerEmail,omitempty"`
	DefaultCompute string              `json:"defaultCompute,omitempty"`
	Versions       []DeploymentVersion `json:"versions"`
	DefaultBuildId string              `json:"defaultBuildId"`
	CreateTime     string              `json:"createTime"`
	UpdateTime     string              `json:"updateTime,omitempty"`
}

Deployment — a worker version series. Workers register a buildId; the engine routes workflow tasks based on default + ramping rules.

type DeploymentPatch added in v1.43.0 

type DeploymentPatch struct {
	Description    *string `json:"description,omitempty"`
	OwnerEmail     *string `json:"ownerEmail,omitempty"`
	DefaultCompute *string `json:"defaultCompute,omitempty"`
}

DeploymentPatch carries the optional fields of an UpdateDeployment. Empty strings are ignored; non-nil maps fully replace existing data.

type DeploymentVersion added in v1.43.0 

type DeploymentVersion struct {
	BuildId     string            `json:"buildId"`
	State       string            `json:"state"` // DEPLOYMENT_STATE_DRAFT|CURRENT|RAMPING|RETIRED
	Description string            `json:"description,omitempty"`
	Compute     string            `json:"compute,omitempty"`
	Image       string            `json:"image,omitempty"`
	Env         map[string]string `json:"env,omitempty"`
	CreateTime  string            `json:"createTime"`
	UpdateTime  string            `json:"updateTime,omitempty"`
}

type DeploymentVersionPatch added in v1.43.0 

type DeploymentVersionPatch struct {
	Description *string           `json:"description,omitempty"`
	Compute     *string           `json:"compute,omitempty"`
	Image       *string           `json:"image,omitempty"`
	Env         map[string]string `json:"env,omitempty"`
}

DeploymentVersionPatch is metadata-only — buildId and CreateTime are immutable.

type EmbedConfig added in v1.33.1 

type EmbedConfig struct {
	DataDir string // "" → "./tasks-data" (reserved; memdb today)
	// Address is where the ZAP listener binds. A filesystem path binds a
	// unix socket — the right shape for service-to-service traffic that
	// never leaves the host, and one that needs no port. Anything else is
	// a host:port TCP address; "" → ":9999".
	Address   string
	Namespace string       // "" → "default"
	Logger    *slog.Logger // nil → slog.Default()
	// MasterKey is the 32-byte root key every shard is encrypted under.
	// Each shard file gets a data-encryption key of its own, wrapped under
	// a key the shard's tenant derives from this one, so rotating the
	// master rewraps the sidecars and leaves the ciphertext untouched.
	// nil leaves shards plaintext — the zero-config dev posture. Supply it
	// from KMS in production; hold it nowhere else.
	MasterKey []byte
	// JWTValidator validates the auth_token field on every ZAP request.
	// nil = no ZAP-side validation (dev / embedded). When non-nil and
	// RequireIdentity=true, every ZAP request must carry an auth_token
	// that validates against IAM; per-request engine is scoped
	// to claims.Owner. This mirrors the HTTP middleware trust boundary.
	JWTValidator    *edge.Verifier
	RequireIdentity bool
	// Replicator wires consensus-replication for every shard. nil →
	// LocalReplicator (single-node passthrough). cmd/tasksd builds a
	// QuasarReplicator when --replicator=quasar is set.
	Replicator replication.Replicator
	// Router selects the (org, ns, taskQueue) leader. nil → solo
	// router that returns the local node for every key.
	Router routing.Router
	// NodeID is this process's stable identifier. "" → "tasks-embed".
	NodeID string
}

EmbedConfig configures the in-process Tasks server.

type Embedded added in v1.33.1 

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

Embedded is the handle to a running in-process Tasks server.

func Embed added in v1.33.1 

func Embed(ctx context.Context, cfg EmbedConfig) (*Embedded, error)

func (*Embedded) ActivitiesForOrg added in v1.47.0 

func (e *Embedded) ActivitiesForOrg(org, ns string) ([]StandaloneActivity, error)

ActivitiesForOrg lists the standalone activities in ns for org (org=="" ⇒ the unscoped/embedded shard). It is the thin, org-scoped programmatic read the in-process host (cloud's clients/fleet) uses to render a registry from engine data without a second HTTP hop into its own surface. Tenancy is the caller's to enforce: pass the validated X-Org-Id, never client input.

func (*Embedded) ActivitiesPageForOrg added in v1.51.3 

func (e *Embedded) ActivitiesPageForOrg(org, ns, cursor string, pageSize int) ([]StandaloneActivity, string, error)

ActivitiesPageForOrg is the PAGINATED org-scoped read: it exposes the cursor ActivitiesForOrg hides, so an in-process host can walk a namespace to COMPLETION instead of silently seeing only the first (hash-ordered) 100 rows — the truncation that hides live jobs and drops online workers on a busy org. cursor "" starts the walk; pageSize<=0 defaults to the engine's 100. Returns the next cursor ("" at the end). Tenancy is the caller's to enforce: pass the validated X-Org-Id.

func (*Embedded) Address added in v1.52.4 

func (e *Embedded) Address() string

Address returns where the loopback ZAP listener is bound.

func (*Embedded) CancelActivityForOrg added in v1.51.2 

func (e *Embedded) CancelActivityForOrg(org, ns, activityID, runID, reason, identity string) error

CancelActivityForOrg cancels a standalone activity in org's shard — the org-scoped programmatic MUTATOR that mirrors ActivitiesForOrg's read, so an in-process host (cloud's clients/fleet) can cancel a queued/running job it just listed without a second HTTP hop into its own surface. Tenancy is the caller's to enforce: pass the validated X-Org-Id, never client input. Rejected (error) if the activity is missing or already terminal, exactly like the HTTP path.

func (*Embedded) ClusterHandler added in v1.43.0 

func (e *Embedded) ClusterHandler() http.Handler

ClusterHandler exposes /v1/tasks/cluster, /v1/tasks/cluster/health, /v1/tasks/namespaces/{ns}/migrate. Probe routes are unauthenticated; migrate accepts the same X-Org-Id-scoped identity as everything else.

func (*Embedded) EventsHandler added in v1.40.0 

func (e *Embedded) EventsHandler() http.Handler

EventsHandler returns the SSE realtime stream of engine events.

func (*Embedded) HTTPHandler added in v1.37.2 

func (e *Embedded) HTTPHandler() http.Handler

HTTPHandler returns the browser-only JSON shim. Mirrors zapHandlers. Per-request engine is scoped to the X-Org-Id written by pkg/auth from the validated IAM JWT (Authorization: Bearer). Client-supplied identity headers are stripped before the handler runs. Empty org → legacy unscoped store (embedded/dev path only).

func (*Embedded) MCPHandler added in v1.40.0 

func (e *Embedded) MCPHandler() http.Handler

MCPHandler returns the JSON-RPC 2.0 MCP endpoint.

func (*Embedded) RegisterWorker added in v1.43.0 

func (e *Embedded) RegisterWorker(w Worker)

RegisterWorker upserts a worker into the in-memory registry. Workers self-register on first poll/heartbeat. Real wiring will move to the dispatcher Subscribe path; this is the public surface the UI reads.

func (*Embedded) ServeGated added in v1.48.0 

func (e *Embedded) ServeGated(ctx context.Context, addr string, validator *edge.Verifier) error

ServeGated starts a SECOND ZAP listener at addr that exposes the SAME engine to out-of-process callers under mandatory identity gating: every request must carry an auth_token that validates against validator (RequireIdentity), and its engine view is org-scoped to the token owner (CONTRACT §6). The loopback listener started by Embed is unchanged — in-process callers inside the host's trust boundary keep dialing it ungated. Server-push to a worker is routed to the listener that holds its subscription, so gated and loopback workers both receive delivery. A nil validator is refused: exposing durable execution ungated across the cluster is never correct. Call once, after Embed; Stop tears down every listener.

func (*Embedded) Stop added in v1.33.1 

func (e *Embedded) Stop(ctx context.Context) error

Stop shuts the server down. Idempotent.

func (*Embedded) View added in v1.50.0 

func (e *Embedded) View(p Principal) View

View returns the control-plane view scoped to p. The zero Principal is the root view.

type Event added in v1.40.0 

type Event struct {
	Kind       string `json:"kind"`             // workflow.started|workflow.canceled|workflow.terminated|workflow.signaled|schedule.created|schedule.paused|schedule.resumed|schedule.deleted|namespace.registered|batch.started
	OrgID      string `json:"org_id,omitempty"` // tenant scope; "" = unscoped (embedded/dev)
	Namespace  string `json:"namespace,omitempty"`
	WorkflowID string `json:"workflow_id,omitempty"`
	RunID      string `json:"run_id,omitempty"`
	ScheduleID string `json:"schedule_id,omitempty"`
	BatchID    string `json:"batch_id,omitempty"`
	At         string `json:"at"`
	Data       any    `json:"data,omitempty"`
}

type ExecutionRef added in v1.38.0 

type ExecutionRef struct {
	WorkflowId string `json:"workflowId"`
	RunId      string `json:"runId"`
}

type FailureStreak added in v1.52.3 

type FailureStreak struct {
	Org          string `json:"org,omitempty"` // "" = unscoped (embedded/dev)
	Namespace    string `json:"namespace"`
	WorkflowType string `json:"workflowType,omitempty"`
	// ActivityType empty ⇒ the workflow itself failed (a decider fault with
	// no activity to blame), not an activity within it.
	ActivityType string `json:"activityType,omitempty"`
	TaskQueue    string `json:"taskQueue,omitempty"`
	// ScheduleId is set when the run came from a cron schedule, which is the
	// field that turns "some JobWorkflow is broken" into "THIS nightly backup
	// has not run". Stamped by the sweeper as a search attribute.
	ScheduleId string `json:"scheduleId,omitempty"`
	// ConsecutiveFailures is the how-long-has-this-been-dead number: 1 is a
	// blip, 4489 is the incident. Survives restart, unlike the in-memory
	// throttle counter.
	ConsecutiveFailures int64 `json:"consecutiveFailures"`
	// Attempt is the per-run retry attempt of the LAST failure — a different
	// axis from ConsecutiveFailures, which spans runs.
	Attempt int `json:"attempt,omitempty"`
	// Retrying distinguishes "failed and the engine will try again" from
	// "retries are exhausted and this run is dead".
	Retrying bool `json:"retrying,omitempty"`
	// Persistent is the alertable condition: failing continuously for at
	// least failPersistentAfter.
	Persistent       bool   `json:"persistent,omitempty"`
	FirstFailureTime string `json:"firstFailureTime"`
	LastFailureTime  string `json:"lastFailureTime"`
	LastError        string `json:"lastError,omitempty"`
	// LastWorkflowId / LastRunId point at one concrete run to go read. The
	// identity itself is deliberately run-independent.
	LastWorkflowId string `json:"lastWorkflowId,omitempty"`
	LastRunId      string `json:"lastRunId,omitempty"`
}

FailureStreak is the durable aggregate for ONE execution identity that is currently failing: an activity type (ActivityType set) or a workflow type failing on its own (ActivityType empty). One row per broken thing, upserted on each failure and DELETED on the first success — so the keyspace is bounded by how much is broken right now, not by how often it failed, and a non-empty listing is exactly the list of what needs attention.

type Handler 

type Handler func(taskType string, payload map[string]any)

Handler processes a one-shot task (webhook delivery, settlement, etc.)

type HistoryEvent added in v1.43.0 

type HistoryEvent struct {
	EventId    int64          `json:"eventId"`
	EventTime  string         `json:"eventTime"`
	EventType  string         `json:"eventType"`
	Attributes map[string]any `json:"attributes,omitempty"`
}

HistoryEvent — a single durable record in a workflow execution's history. Modeled on Temporal's HistoryEvent but JSON-native and owned by Hanzo. EventId is monotonic per (namespace, workflowId, runId); EventType matches the WORKFLOW_EXECUTION_* / WORKFLOW_TASK_* / ACTIVITY_TASK_* / TIMER_* family.

type Identity added in v1.38.0 

type Identity struct {
	Email     string `json:"email"`
	Namespace string `json:"namespace"`
	Role      string `json:"role"` // owner|admin|developer|viewer
	GrantTime string `json:"grantTime"`
}

Identity — who is allowed to operate on this namespace. Sourced from IAM (X-User-Email) when wired, manually managed today.

type Namespace added in v1.38.0 

type Namespace struct {
	NamespaceInfo NamespaceInfo `json:"namespaceInfo"`
	Config        NamespaceCfg  `json:"config"`
	IsActive      bool          `json:"isActive"`
}

type NamespaceCfg added in v1.38.0 

type NamespaceCfg struct {
	WorkflowExecutionRetentionTtl string            `json:"workflowExecutionRetentionTtl"` // "720h"
	APSLimit                      int               `json:"apsLimit"`                      // actions per second
	HistoryArchivalState          string            `json:"historyArchivalState,omitempty"`
	HistoryArchivalUri            string            `json:"historyArchivalUri,omitempty"`
	VisibilityArchivalState       string            `json:"visibilityArchivalState,omitempty"`
	VisibilityArchivalUri         string            `json:"visibilityArchivalUri,omitempty"`
	CustomData                    map[string]string `json:"customData,omitempty"`
}

type NamespaceInfo added in v1.38.0 

type NamespaceInfo struct {
	Name        string `json:"name"`
	State       string `json:"state"` // NAMESPACE_STATE_REGISTERED|DEPRECATED|DELETED
	Description string `json:"description,omitempty"`
	OwnerEmail  string `json:"ownerEmail,omitempty"`
	Region      string `json:"region,omitempty"` // e.g. "do-sfo3", informational
	CreateTime  string `json:"createTime,omitempty"`
}

type NamespaceMetadataPatch added in v1.43.0 

type NamespaceMetadataPatch struct {
	Description             *string           `json:"description,omitempty"`
	OwnerEmail              *string           `json:"ownerEmail,omitempty"`
	Retention               *string           `json:"retention,omitempty"`
	HistoryArchivalState    *string           `json:"historyArchivalState,omitempty"`
	HistoryArchivalUri      *string           `json:"historyArchivalUri,omitempty"`
	VisibilityArchivalState *string           `json:"visibilityArchivalState,omitempty"`
	VisibilityArchivalUri   *string           `json:"visibilityArchivalUri,omitempty"`
	CustomData              map[string]string `json:"customData,omitempty"`
}

NamespaceMetadataPatch carries the fields a metadata update may set. Empty string fields are ignored; non-nil maps fully replace.

type NexusEndpoint added in v1.38.0 

type NexusEndpoint struct {
	Name        string `json:"name"`
	Namespace   string `json:"namespace"`
	Description string `json:"description,omitempty"`
	Target      string `json:"target"` // ns2://other-namespace/handler
	CreateTime  string `json:"createTime"`
}

NexusEndpoint — cross-namespace operation bridge.

type Principal added in v1.52.4 

type Principal = storepkg.Principal

Principal is the tenant a store view is scoped to: an org, optionally narrowed to a project and a user. The zero value is the root tenant.

func Org added in v1.52.4 

func Org(org string) Principal

Org returns the principal naming an org.

type RetryPolicy added in v1.43.0 

type RetryPolicy struct {
	InitialInterval        string   `json:"initialInterval,omitempty"`
	BackoffCoefficient     float64  `json:"backoffCoefficient,omitempty"`
	MaximumInterval        string   `json:"maximumInterval,omitempty"`
	MaximumAttempts        int      `json:"maximumAttempts,omitempty"`
	NonRetryableErrorTypes []string `json:"nonRetryableErrorTypes,omitempty"`
}

RetryPolicy — schedule retry knobs for an activity.

type Schedule added in v1.38.0 

type Schedule struct {
	ScheduleId string         `json:"scheduleId"`
	Namespace  string         `json:"namespace"`
	Spec       ScheduleSpec   `json:"spec"`
	Action     ScheduleAction `json:"action"`
	State      ScheduleState  `json:"state"`
	Info       ScheduleInfo   `json:"info"`
}

type ScheduleAction added in v1.38.0 

type ScheduleAction struct {
	WorkflowType TypeRef `json:"workflowType"`
	TaskQueue    string  `json:"taskQueue"`
	Input        any     `json:"input,omitempty"`
}

type ScheduleInfo added in v1.38.0 

type ScheduleInfo struct {
	CreateTime     string `json:"createTime"`
	UpdateTime     string `json:"updateTime,omitempty"`
	ActionCount    int64  `json:"actionCount"`
	NextActionTime string `json:"nextActionTime,omitempty"`
}

type ScheduleSpec added in v1.38.0 

type ScheduleSpec struct {
	CronString []string `json:"cronString,omitempty"`
	Interval   []struct {
		Interval string `json:"interval"`
		Phase    string `json:"phase,omitempty"`
	} `json:"interval,omitempty"`
}

type ScheduleState added in v1.38.0 

type ScheduleState struct {
	Paused bool   `json:"paused"`
	Note   string `json:"note,omitempty"`
}

type SearchAttribute added in v1.43.0 

type SearchAttribute struct {
	Name string `json:"name"`
	Type string `json:"type"` // Keyword|Text|Int|Double|Bool|Datetime|KeywordList
}

SearchAttribute is a typed search attribute registered to a namespace.

type StandaloneActivity added in v1.43.0 

type StandaloneActivity struct {
	Execution              ExecutionRef `json:"execution"`
	Type                   TypeRef      `json:"type"`
	TaskQueue              string       `json:"taskQueue,omitempty"`
	Status                 string       `json:"status"` // ACTIVITY_TASK_STATE_*
	StartTime              string       `json:"startTime,omitempty"`
	CloseTime              string       `json:"closeTime,omitempty"`
	RetryPolicy            *RetryPolicy `json:"retryPolicy,omitempty"`
	Input                  any          `json:"input,omitempty"`
	Result                 any          `json:"result,omitempty"`
	FailureCause           string       `json:"failureCause,omitempty"`
	Identity               string       `json:"identity,omitempty"`
	Attempt                int          `json:"attempt"`
	MaximumAttempts        int          `json:"maximumAttempts,omitempty"`
	ScheduleToCloseTimeout string       `json:"scheduleToCloseTimeout,omitempty"`
	ScheduleToStartTimeout string       `json:"scheduleToStartTimeout,omitempty"`
	StartToCloseTimeout    string       `json:"startToCloseTimeout,omitempty"`
	HeartbeatTimeout       string       `json:"heartbeatTimeout,omitempty"`
	LastHeartbeatTime      string       `json:"lastHeartbeatTime,omitempty"`
	// LeaseExpiry (RFC3339) is set when a worker claims the activity
	// (ClaimNextActivity), refreshed on each heartbeat, and cleared when the
	// lease is reaped. Empty ⇒ the activity is not lease-managed (e.g. a
	// presence record heartbeated but never claimed) and the reaper ignores it.
	LeaseExpiry   string `json:"leaseExpiry,omitempty"`
	HistoryLength int64  `json:"historyLength,omitempty"`
}

StandaloneActivity — a first-class activity record keyed by (ns, activityId, runId), independent of any workflow. Workers schedule, heartbeat, and complete activities that the engine tracks without an enclosing workflow execution.

type TypeRef added in v1.38.0 

type TypeRef struct {
	Name string `json:"name"`
}

type VersionValidationResult added in v1.43.0 

type VersionValidationResult struct {
	NetworkOk         bool     `json:"networkOk"`
	WorkerRegistered  bool     `json:"workerRegistered"`
	HeartbeatReceived bool     `json:"heartbeatReceived"`
	LatencyMs         int      `json:"latencyMs"`
	Errors            []string `json:"errors"`
}

VersionValidationResult is the synthetic snapshot returned by ValidateVersion. Real validation requires the worker SDK runtime to answer; for now we return a deterministic body keyed off whether the version row exists.

type View added in v1.50.0 

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

View is an org-scoped, in-process control-plane handle on the embedded engine — the door for a HOST binary (e.g. cloud's platform cron) to manage namespaces and schedules in a specific org's shard without a ZAP hop or a written identity. In-process callers share the host's trust boundary (the same stance as the ungated loopback listener), so the org they pass is authoritative — never derive it from anything client-supplied. Workflows a schedule starts dispatch to whatever worker is subscribed on the schedule's (namespace, task queue) — worker subscription is org-agnostic; results route back to the owning shard via the task's org.

func (View) CreateSchedule added in v1.50.0 

func (v View) CreateSchedule(s Schedule) error

CreateSchedule upserts a schedule (store-keyed by namespace+id).

func (View) DeleteSchedule added in v1.50.0 

func (v View) DeleteSchedule(ns, id string) error

DeleteSchedule removes a schedule.

func (View) DescribeActivity added in v1.51.4 

func (v View) DescribeActivity(ns, activityID, runID string) (*StandaloneActivity, bool, error)

DescribeActivity loads one standalone activity in ns; ok=false when absent.

func (View) DescribeSchedule added in v1.50.0 

func (v View) DescribeSchedule(ns, id string) (*Schedule, bool, error)

DescribeSchedule loads one schedule; ok=false when absent.

func (View) FailureStreaks added in v1.52.3 

func (v View) FailureStreaks(ns string) ([]FailureStreak, error)

FailureStreaks returns what is broken in ns right now, worst first: one durable row per activity/workflow identity that has failed every attempt since it last succeeded. Empty means nothing is failing.

This is the read that was missing. cloud's clients/cron fired a JobWorkflow whose activity failed on every one of 4489 fires across 11 days while the engine's only output was an SSE event nobody was subscribed to; the six nightly backups it silently skipped were found by hand-reading SQLite. A host polls this instead — each row carries org, namespace, workflow and activity type, task queue, originating scheduleId, consecutive failures, how long it has been failing, the last error, and a run to go read.

func (View) ListSchedules added in v1.50.0 

func (v View) ListSchedules(ns string) ([]Schedule, error)

ListSchedules returns every schedule in ns.

func (View) ListWorkflows added in v1.50.0 

func (v View) ListWorkflows(ns string) ([]WorkflowExecution, error)

ListWorkflows returns every workflow execution in ns (this org's shard) — how a host observes the runs its schedules produced.

func (View) RegisterNamespace added in v1.50.0 

func (v View) RegisterNamespace(ns Namespace) error

RegisterNamespace idempotently registers ns in this org's shard — required before any workflow (including a schedule fire) can start in it.

func (View) StartActivity added in v1.51.4 

func (v View) StartActivity(ns, activityID, runID string, typ TypeRef, taskQueue string, input any, retry *RetryPolicy, scheduleToClose, scheduleToStart, startToClose, heartbeat string, identity, requestID string) (*StandaloneActivity, error)

StartActivity enqueues a standalone activity in ns (this org's shard) — the in-binary seam a host subsystem uses to put work on an org queue for a fleet worker to claim (fn.run, studio.render), mirroring the HTTP activities API.

func (View) TriggerSchedule added in v1.50.0 

func (v View) TriggerSchedule(ns, id, requestID string) (*WorkflowExecution, error)

TriggerSchedule fires the schedule's action immediately (manual run).

type Worker added in v1.43.0 

type Worker struct {
	Identity      string `json:"identity"`
	Namespace     string `json:"namespace"`
	TaskQueue     string `json:"taskQueue,omitempty"`
	SDKName       string `json:"sdkName,omitempty"`
	SDKVersion    string `json:"sdkVersion,omitempty"`
	LastHeartbeat string `json:"lastHeartbeat,omitempty"`
	FirstSeen     string `json:"firstSeen,omitempty"`
}

Worker is the registered worker as exposed to the UI.

type WorkflowExecution added in v1.38.0 

type WorkflowExecution struct {
	Execution    ExecutionRef          `json:"execution"`
	Type         TypeRef               `json:"type"`
	StartTime    string                `json:"startTime,omitempty"`
	CloseTime    string                `json:"closeTime,omitempty"`
	Status       string                `json:"status"` // WORKFLOW_EXECUTION_STATUS_*
	TaskQueue    string                `json:"taskQueue,omitempty"`
	HistoryLen   int64                 `json:"historyLength,omitempty"`
	Memo         any                   `json:"memo,omitempty"`
	SearchAttrs  map[string]any        `json:"searchAttrs,omitempty"`
	Input        any                   `json:"input,omitempty"`
	Result       any                   `json:"result,omitempty"`
	UserMetadata *WorkflowUserMetadata `json:"userMetadata,omitempty"`
}

type WorkflowUserMetadata added in v1.43.0 

type WorkflowUserMetadata struct {
	Summary   string `json:"summary,omitempty"`
	Details   string `json:"details,omitempty"`
	UpdatedBy string `json:"updatedBy,omitempty"`
	UpdatedAt string `json:"updatedAt,omitempty"`
}

WorkflowUserMetadata is operator-supplied summary/details for a workflow.

Source Files

View all Source files

Directories

Path Synopsis
Package migration moves a single (org, namespace) shard between nodes.
 Package migration moves a single (org, namespace) shard between nodes.
Package replication is the consensus-replication boundary for the Tasks shard manager.
 Package replication is the consensus-replication boundary for the Tasks shard manager.
quasar
Package routing implements sticky leader selection for (org, namespace, taskQueue) triples.
 Package routing implements sticky leader selection for (org, namespace, taskQueue) triples.
Package store implements per-(org, namespace) SQLite shards with optional consensus replication.
 Package store implements per-(org, namespace) SQLite shards with optional consensus replication.

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.