Durable Workflow PHP SDK — API Reference

Guides

Worker
in package

FinalYes

Managed synchronous remote worker for workflow, activity, query, and update tasks.

Table of Contents

Constants

DEFAULT_HEARTBEAT_INTERVAL_SECONDS  : mixed = 30
INITIAL_TRANSIENT_RETRY_DELAY_SECONDS  : mixed = 0.1
MAX_HEARTBEAT_INTERVAL_SECONDS  : mixed = 3600
MAX_TRANSIENT_RETRY_DELAY_SECONDS  : mixed = 5.0
TRANSIENT_RETRY_SLEEP_SLICE_SECONDS  : mixed = 0.1

Properties

$taskQueue  : string
$activities  : array<string, HandlerDefinition>
$buildId  : string|null
$client  : Client
$clock  : callable(): float
$diagnosticListener  : callable(string, array<string, mixed>): void|null
$handlerDiscovery  : HandlerDiscovery
$heartbeatIntervalSeconds  : int
$lastHeartbeatAt  : float
$logger  : LoggerInterface
$queries  : array<string, array<string, HandlerDefinition>>
$registered  : bool
$replayer  : Replayer
$shutdownRequested  : bool
$signals  : array<string, array<string, callable(mixed ...$arguments): mixed>>
$sleeper  : callable(int): void
$transientPollRetryObserver  : callable(string, int, float, ServerException): void|null
$updates  : array<string, array<string, HandlerDefinition>>
$workerId  : string
$workflows  : array<string, HandlerDefinition>

Methods

__construct()  : mixed
contracts()  : array{workflows: list, activities: list, workflow_commands: array}
Return the complete local definition sent during worker registration.
create()  : self
Construct the preferred class-oriented worker surface.
declareSignal()  : self
Declare a replay-consumed signal and its argument signature.
register()  : self
Discover and register one or more attribute-based handler services.
registerActivity()  : self
registerQuery()  : self
registerUpdate()  : self
registerWorkflow()  : self
requestShutdown()  : void
run()  : void
tick()  : bool
Execute at most one task of each kind; useful for custom supervisors and tests.
validate()  : void
Validate every registered command contract without contacting the server.
acknowledgeTaskFailure()  : void
applyHeartbeatInterval()  : void
assertDiscoveriesCanRegister()  : void
assertHandlerContext()  : void
assertUnique()  : void
assertValidDeclarationName()  : void
commandHandlerContracts()  : array<int, array{name: string, parameters: list}>
completeHistory()  : array<int, array<string, mixed>>
decodeArguments()  : array<int, mixed>
diagnostic()  : void
elapsedSinceHeartbeat()  : float
executeActivityTask()  : void
executeQueryTask()  : void
executeUpdate()  : array<string, mixed>
executeWorkflowTask()  : void
handlerFailure()  : void
heartbeat()  : void
heartbeatIfDue()  : void
historyFromTask()  : array<int, array<string, mixed>>
installSignalHandlers()  : void
isTerminalTaskConflict()  : bool
isTransientRegistrationFailure()  : bool
isTransientWorkflowTaskLeaseRefusal()  : bool
matchesWorkflowTaskLeaseFence()  : bool
now()  : float
pollWithRetry()  : array<string, mixed>|null
preparePoll()  : int
registerActivityDefinition()  : self
registerQueryDefinition()  : self
registerUpdateDefinition()  : self
registerWithRetry()  : array<string, mixed>|null
registerWorkflowDefinition()  : self
renewWorkflowTaskLease()  : bool
stopForTerminalPoll()  : bool
taskFromPoll()  : array<string, mixed>|null
transientRetryDelay()  : float
validHeartbeatInterval()  : int|null
waitForTransientRetry()  : void
workflowCommandContracts()  : array<string, array{queries: list, query_contracts: list}>, signals: list, signal_contracts: list}>, updates: list, update_validators: list, update_contracts: list}>}>
workflowTaskLeaseResponseFailure()  : ServerException

Constants

DEFAULT_HEARTBEAT_INTERVAL_SECONDS

private mixed DEFAULT_HEARTBEAT_INTERVAL_SECONDS = 30

INITIAL_TRANSIENT_RETRY_DELAY_SECONDS

private mixed INITIAL_TRANSIENT_RETRY_DELAY_SECONDS = 0.1

MAX_HEARTBEAT_INTERVAL_SECONDS

private mixed MAX_HEARTBEAT_INTERVAL_SECONDS = 3600

MAX_TRANSIENT_RETRY_DELAY_SECONDS

private mixed MAX_TRANSIENT_RETRY_DELAY_SECONDS = 5.0

TRANSIENT_RETRY_SLEEP_SLICE_SECONDS

private mixed TRANSIENT_RETRY_SLEEP_SLICE_SECONDS = 0.1

Properties

$taskQueue read-only

public string $taskQueue

$activities

private array<string, HandlerDefinition> $activities = []

$buildId read-only

private string|null $buildId = null

$clock read-only

private callable(): float $clock

$diagnosticListener read-only

private callable(string, array<string, mixed>): void|null $diagnosticListener

$handlerDiscovery read-only

private HandlerDiscovery $handlerDiscovery

$heartbeatIntervalSeconds

private int $heartbeatIntervalSeconds

$lastHeartbeatAt

private float $lastHeartbeatAt = 0.0

$logger read-only

private LoggerInterface $logger

$queries

private array<string, array<string, HandlerDefinition>> $queries = []

$registered

private bool $registered = false

$shutdownRequested

private bool $shutdownRequested = false

$signals

private array<string, array<string, callable(mixed ...$arguments): mixed>> $signals = []

$sleeper read-only

private callable(int): void $sleeper

$transientPollRetryObserver read-only

private callable(string, int, float, ServerException): void|null $transientPollRetryObserver = null

$updates

private array<string, array<string, HandlerDefinition>> $updates = []

$workerId read-only

private string $workerId

$workflows

private array<string, HandlerDefinition> $workflows = []

Methods

__construct()

public __construct(Client $client, string $taskQueue[, string|null $workerId = null ][, int $heartbeatIntervalSeconds = self::DEFAULT_HEARTBEAT_INTERVAL_SECONDS ][, string|null $buildId = null ][, Closure|null $clock = null ][, Closure|null $sleeper = null ][, Closure|null $transientPollRetryObserver = null ][, ContainerInterface|null $container = null ][, LoggerInterface|null $logger = null ][, callable|null $diagnosticListener = null ]) : mixed
Parameters
$client : Client
$taskQueue : string
$workerId : string|null = null
$heartbeatIntervalSeconds : int = self::DEFAULT_HEARTBEAT_INTERVAL_SECONDS
$buildId : string|null = null
$clock : Closure|null = null
$sleeper : Closure|null = null
$transientPollRetryObserver : Closure|null = null
$container : ContainerInterface|null = null
$logger : LoggerInterface|null = null
$diagnosticListener : callable|null = null

contracts()

Return the complete local definition sent during worker registration.

public contracts() : array{workflows: list, activities: list, workflow_commands: array}
Return values
array{workflows: list, activities: list, workflow_commands: array}

create()

Construct the preferred class-oriented worker surface.

public static create(Client $client, string $taskQueue[, ContainerInterface|null $container = null ][, LoggerInterface|null $logger = null ][, callable(string, array<string, mixed>): void|null $diagnosticListener = null ]) : self
Parameters
$client : Client
$taskQueue : string
$container : ContainerInterface|null = null
$logger : LoggerInterface|null = null
$diagnosticListener : callable(string, array<string, mixed>): void|null = null
Return values
self

declareSignal()

Declare a replay-consumed signal and its argument signature.

public declareSignal(string $workflowType, string $signalName[, callable(mixed ...$arguments): mixed|null $signature = null ]) : self

The optional signature is reflected for registration metadata only and is never invoked. Workflows continue to consume signals deterministically through WorkflowContext::signals().

Parameters
$workflowType : string
$signalName : string
$signature : callable(mixed ...$arguments): mixed|null = null
Return values
self

register()

Discover and register one or more attribute-based handler services.

public register(class-string|object ...$services) : self
Parameters
$services : class-string|object
Return values
self

registerActivity()

public registerActivity(string $activityType, callable(ActivityContext, mixed ...$arguments): mixed $handler) : self
Parameters
$activityType : string
$handler : callable(ActivityContext, mixed ...$arguments): mixed
Return values
self

registerQuery()

public registerQuery(string $workflowType, string $queryName, callable(QueryContext, mixed ...$arguments): mixed $handler) : self
Parameters
$workflowType : string
$queryName : string
$handler : callable(QueryContext, mixed ...$arguments): mixed
Return values
self

registerUpdate()

public registerUpdate(string $workflowType, string $updateName, callable(QueryContext, mixed ...$arguments): mixed $handler) : self
Parameters
$workflowType : string
$updateName : string
$handler : callable(QueryContext, mixed ...$arguments): mixed
Return values
self

registerWorkflow()

public registerWorkflow(string $workflowType, callable(WorkflowContext, mixed ...$arguments): mixed $handler) : self
Parameters
$workflowType : string
$handler : callable(WorkflowContext, mixed ...$arguments): mixed
Return values
self

requestShutdown()

public requestShutdown() : void

run()

public run([int $pollTimeoutSeconds = 5 ]) : void
Parameters
$pollTimeoutSeconds : int = 5

tick()

Execute at most one task of each kind; useful for custom supervisors and tests.

public tick([int $pollTimeoutSeconds = 1 ]) : bool
Parameters
$pollTimeoutSeconds : int = 1
Return values
bool

validate()

Validate every registered command contract without contacting the server.

public validate() : void

acknowledgeTaskFailure()

private acknowledgeTaskFailure(string $taskKind, string $taskId, Throwable $taskFailure, callable(Throwable): void $failureAcknowledgement) : void
Parameters
$taskKind : string
$taskId : string
$taskFailure : Throwable
$failureAcknowledgement : callable(Throwable): void

applyHeartbeatInterval()

private applyHeartbeatInterval(array<string, mixed> $response) : void
Parameters
$response : array<string, mixed>

assertDiscoveriesCanRegister()

private assertDiscoveriesCanRegister(array<int, DiscoveredHandlers> $discoveries) : void
Parameters
$discoveries : array<int, DiscoveredHandlers>

assertHandlerContext()

private assertHandlerContext(callable $handler, string $contextClass, string $contract) : void
Parameters
$handler : callable
$contextClass : string
$contract : string

assertUnique()

private assertUnique(array<string, mixed> $registry, string $name, string $kind) : void
Parameters
$registry : array<string, mixed>
$name : string
$kind : string

assertValidDeclarationName()

private assertValidDeclarationName(string $name, string $kind[, bool $signalDeclaration = false ]) : void
Parameters
$name : string
$kind : string
$signalDeclaration : bool = false

commandHandlerContracts()

private commandHandlerContracts(array<string, HandlerDefinition|callable> $handlers[, class-string|null $contextClass = null ]) : array<int, array{name: string, parameters: list}>
Parameters
$handlers : array<string, HandlerDefinition|callable>
$contextClass : class-string|null = null
Return values
array<int, array{name: string, parameters: list}>

completeHistory()

private completeHistory(array<string, mixed> $task, string $leaseOwner, int $attempt) : array<int, array<string, mixed>>
Parameters
$task : array<string, mixed>
$leaseOwner : string
$attempt : int
Return values
array<int, array<string, mixed>>

decodeArguments()

private decodeArguments(mixed $raw) : array<int, mixed>
Parameters
$raw : mixed
Return values
array<int, mixed>

diagnostic()

private diagnostic(string $event[, array<string, mixed> $context = [] ][, string $level = 'info' ]) : void
Parameters
$event : string
$context : array<string, mixed> = []
$level : string = 'info'

elapsedSinceHeartbeat()

private elapsedSinceHeartbeat() : float
Return values
float

executeActivityTask()

private executeActivityTask(array<string, mixed> $task) : void
Parameters
$task : array<string, mixed>

executeQueryTask()

private executeQueryTask(array<string, mixed> $task) : void
Parameters
$task : array<string, mixed>

executeUpdate()

private executeUpdate(string $workflowType, string $updateId, array<int, array<string, mixed>> $history, array<string, mixed> $task) : array<string, mixed>
Parameters
$workflowType : string
$updateId : string
$history : array<int, array<string, mixed>>
$task : array<string, mixed>
Return values
array<string, mixed>

executeWorkflowTask()

private executeWorkflowTask(array<string, mixed> $task) : void
Parameters
$task : array<string, mixed>

handlerFailure()

private handlerFailure(string $kind, string $identity, Throwable $exception) : void
Parameters
$kind : string
$identity : string
$exception : Throwable

heartbeat()

private heartbeat() : void

heartbeatIfDue()

private heartbeatIfDue() : void

historyFromTask()

private historyFromTask(array<string, mixed> $task) : array<int, array<string, mixed>>
Parameters
$task : array<string, mixed>
Return values
array<int, array<string, mixed>>

installSignalHandlers()

private installSignalHandlers() : void

isTerminalTaskConflict()

private isTerminalTaskConflict(string $taskKind, string $taskId, Throwable $exception) : bool
Parameters
$taskKind : string
$taskId : string
$exception : Throwable
Return values
bool

isTransientWorkflowTaskLeaseRefusal()

private isTransientWorkflowTaskLeaseRefusal(array<string, mixed> $response) : bool
Parameters
$response : array<string, mixed>
Return values
bool

matchesWorkflowTaskLeaseFence()

private matchesWorkflowTaskLeaseFence(array<string, mixed> $response, string $taskId, string $leaseOwner, int $taskAttempt) : bool
Parameters
$response : array<string, mixed>
$taskId : string
$leaseOwner : string
$taskAttempt : int
Return values
bool

now()

private now() : float
Return values
float

pollWithRetry()

private pollWithRetry(string $taskKind, callable(): array<string, mixed> $poll) : array<string, mixed>|null
Parameters
$taskKind : string
$poll : callable(): array<string, mixed>
Return values
array<string, mixed>|null

preparePoll()

private preparePoll(int $requestedTimeoutSeconds) : int
Parameters
$requestedTimeoutSeconds : int
Return values
int

registerActivityDefinition()

private registerActivityDefinition(string $activityType, HandlerDefinition $handler) : self
Parameters
$activityType : string
$handler : HandlerDefinition
Return values
self

registerQueryDefinition()

private registerQueryDefinition(string $workflowType, string $queryName, HandlerDefinition $handler) : self
Parameters
$workflowType : string
$queryName : string
$handler : HandlerDefinition
Return values
self

registerUpdateDefinition()

private registerUpdateDefinition(string $workflowType, string $updateName, HandlerDefinition $handler) : self
Parameters
$workflowType : string
$updateName : string
$handler : HandlerDefinition
Return values
self

registerWithRetry()

private registerWithRetry() : array<string, mixed>|null
Return values
array<string, mixed>|null

registerWorkflowDefinition()

private registerWorkflowDefinition(string $workflowType, HandlerDefinition $handler) : self
Parameters
$workflowType : string
$handler : HandlerDefinition
Return values
self

renewWorkflowTaskLease()

private renewWorkflowTaskLease(string $taskId, string $leaseOwner, int $taskAttempt) : bool
Parameters
$taskId : string
$leaseOwner : string
$taskAttempt : int
Return values
bool

stopForTerminalPoll()

private stopForTerminalPoll(array<string, mixed> $response) : bool
Parameters
$response : array<string, mixed>
Return values
bool

taskFromPoll()

private taskFromPoll(array<string, mixed> $response) : array<string, mixed>|null
Parameters
$response : array<string, mixed>
Return values
array<string, mixed>|null

transientRetryDelay()

private transientRetryDelay(int $attempt, mixed $retryAfterSeconds) : float
Parameters
$attempt : int
$retryAfterSeconds : mixed
Return values
float

validHeartbeatInterval()

private validHeartbeatInterval(mixed $interval) : int|null
Parameters
$interval : mixed
Return values
int|null

waitForTransientRetry()

private waitForTransientRetry(float $delaySeconds) : void
Parameters
$delaySeconds : float

workflowCommandContracts()

private workflowCommandContracts() : array<string, array{queries: list, query_contracts: list}>, signals: list, signal_contracts: list}>, updates: list, update_validators: list, update_contracts: list}>}>
Return values
array<string, array{queries: list, query_contracts: list}>, signals: list, signal_contracts: list}>, updates: list, update_validators: list, update_contracts: list}>}>

workflowTaskLeaseResponseFailure()

private workflowTaskLeaseResponseFailure(string $fallbackMessage, array<string, mixed> $response) : ServerException
Parameters
$fallbackMessage : string
$response : array<string, mixed>
Return values
ServerException
On this page

Search results