Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,14 @@ project follows [Semantic Versioning](https://semver.org/).
before acquiring more work. Task and cancellation authority deadlines retain
their original bounds.

## [2.4.1] - 2026-10-14

### Fixed

- Include actual lifetime history-cache counters in the worker's shutdown
diagnostic. Logs and Laravel/Symfony events preserve those counters after
retained history is cleared.

## [2.4.0] - 2026-10-14

### Added
Expand Down
2 changes: 1 addition & 1 deletion composer.json
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@
}
},
"durable-workflow": {
"product-train": "2.4.0",
"product-train": "2.4.1",
"supported-server-versions": "2.5.0",
"worker-protocol-version": "1.20",
"control-plane-version": "2",
Expand Down
4 changes: 2 additions & 2 deletions docs/quickstart-contract.json
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@
"schema_version": 2,
"package": {
"name": "durable-workflow/sdk",
"published_version": "2.4.0",
"composer_requirement": "2.4.0",
"published_version": "2.4.1",
"composer_requirement": "2.4.1",
"onboarding_requirement": "^2.4"
},
"runtime_targets": {
Expand Down
6 changes: 6 additions & 0 deletions docs/sdk-reference.md
Original file line number Diff line number Diff line change
Expand Up @@ -462,6 +462,12 @@ retry, shutdown, and handler failures then use the application's normal logging
pipeline. The optional `diagnosticListener` receives the same event names and
structured context.

The `worker.stopped` context includes `sticky_cache` counters for `hit`, `miss`,
`eviction` and `forced_cold_replay` over that worker process's lifetime. Clearing
retained history during shutdown preserves these counters. Laravel and Symfony
forward the same context through `WorkerDiagnosticEvent`. A cache hit still
replays the workflow; these counters do not measure retained live execution.

Signal methods are signature declarations for server admission and are not
invoked. The workflow reads their committed values with
`$context->signals('set-language')` during replay. Query and update methods are
Expand Down
1 change: 1 addition & 0 deletions src/Worker.php
Original file line number Diff line number Diff line change
Expand Up @@ -399,6 +399,7 @@ public function run(int $pollTimeoutSeconds = 5): void
$this->diagnostic('worker.stopped', [
'worker_id' => $this->workerId,
'task_queue' => $this->taskQueue,
'sticky_cache' => $this->stickyCacheMetrics(),
]);
}
}
Expand Down
51 changes: 50 additions & 1 deletion tests/PortableWorkerAffinityTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -1071,6 +1071,35 @@ public function testWorkerFactoryCacheConfigurationExpiresAndFetchesAuthoritativ
self::assertSame(1, $outcome['metrics']['eviction']);
}

public function testStoppedDiagnosticAndLogPreserveActualCacheCountersAfterCleanup(): void
{
$diagnostics = [];
$logger = new class extends \Psr\Log\AbstractLogger {
/** @var array<string, array<string, mixed>> */
public array $contexts = [];

public function log($level, $message, array $context = []): void
{
$this->contexts[(string) $message] = $context;
}
};
$outcome = $this->runStickyColdReplayScenario(
'diagnostics',
logger: $logger,
diagnosticListener: static function (string $name, array $context) use (&$diagnostics): void {
$diagnostics[$name] = $context;
},
afterPolling: static fn (Worker $worker) => $worker->run(0),
);

self::assertSame(['hit' => 1, 'miss' => 3, 'eviction' => 2, 'forced_cold_replay' => 1], $outcome['metrics']);
self::assertSame(1, $outcome['history_fetches']);
self::assertSame($outcome['initial_commands'], $outcome['fallback_commands']);
self::assertSame($outcome['metrics'], $diagnostics['worker.stopped']['sticky_cache'] ?? null);
self::assertSame($diagnostics['worker.stopped'], $logger->contexts['worker.stopped'] ?? null);
self::assertSame(['worker_id', 'task_queue', 'sticky_cache'], array_keys($diagnostics['worker.stopped']));
}

public function testStickyCacheNeverPromotesAnUnreplayableRuntimeSuffix(): void
{
$cache = new StickyWorkflowCache(capacity: 1, ttlSeconds: 300);
Expand Down Expand Up @@ -1806,6 +1835,9 @@ private function runStickyColdReplayScenario(
string $scenario,
bool $acceptedStart = false,
bool $useFactory = false,
?\Psr\Log\LoggerInterface $logger = null,
?\Closure $diagnosticListener = null,
?\Closure $afterPolling = null,
): array
{
$codec = new AvroPayloadCodec();
Expand Down Expand Up @@ -1846,6 +1878,12 @@ private function runStickyColdReplayScenario(
'sticky_replay_mode' => $replayMode,
];
$tasks = match ($scenario) {
'diagnostics' => [
$task('target-initial', 'target-workflow', 'target-run', $fullHistory, 'cold_replay'),
$task('target-hit', 'target-workflow', 'target-run', [], 'sticky_hit_expected'),
$task('other-initial', 'other-workflow', 'other-run', $fullHistory, 'cold_replay'),
$task('target-fallback', 'target-workflow', 'target-run', [], 'sticky_hit_expected'),
],
'eviction' => [
$task('target-initial', 'target-workflow', 'target-run', $fullHistory, 'cold_replay'),
$task('other-initial', 'other-workflow', 'other-run', $fullHistory, 'cold_replay'),
Expand All @@ -1869,8 +1907,14 @@ private function runStickyColdReplayScenario(
array $_headers,
?array $body,
) use (&$tasks, &$commands, &$historyFetches, $fullHistory, $useFactory, $factoryTtlSeconds): array {
if (str_ends_with($uri, '/api/worker/register')) {
return ['registered' => true, 'heartbeat_interval_seconds' => 10];
}
if (str_ends_with($uri, '/api/worker/workflow-tasks/poll')) {
$next = array_shift($tasks);
if ($next === null) {
return ['task' => null, 'poll_status' => 'stopped', 'reason' => 'worker_stopped'];
}

return ['poll_status' => 'leased', 'task' => $next];
}
Expand Down Expand Up @@ -1918,6 +1962,8 @@ private function runStickyColdReplayScenario(
'sticky-workers',
stickyCacheCapacity: 1,
stickyCacheTtlSeconds: $factoryTtlSeconds,
logger: $logger,
diagnosticListener: $diagnosticListener,
) : new Worker(
$client,
'sticky-workers',
Expand All @@ -1926,8 +1972,10 @@ private function runStickyColdReplayScenario(
clock: static function () use (&$now): float {
return $now;
},
stickyCacheCapacity: $scenario === 'eviction' ? 1 : 10,
stickyCacheCapacity: in_array($scenario, ['eviction', 'diagnostics'], true) ? 1 : 10,
stickyCacheTtlSeconds: $scenario === 'expiry' ? 1 : 300,
logger: $logger,
diagnosticListener: $diagnosticListener,
);
$worker->registerWorkflow(
'sticky.workflow',
Expand All @@ -1951,6 +1999,7 @@ private function runStickyColdReplayScenario(
}
}
}
$afterPolling?->__invoke($worker);

$initial = $commands['target-initial']['commands'] ?? null;
$fallback = $commands['target-fallback']['commands'] ?? null;
Expand Down
16 changes: 14 additions & 2 deletions tests/compat/history-cache.php
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,14 @@ final class ConfiguredHistoryCacheTransport implements Transport
public function send(string $method, string $uri, array $headers, ?array $body = null): array
{
$history = [['event_type' => 'WorkflowStarted', 'payload' => ['sequence' => 0]]];
if (str_ends_with($uri, '/worker/register')) {
return ['registered' => true, 'heartbeat_interval_seconds' => 10];
}
if (str_ends_with($uri, '/workflow-tasks/poll')) {
++$this->polls;
if ($this->polls > 3) {
return ['poll_status' => 'stopped', 'reason' => 'worker_stopped', 'task' => null];
}
$identity = $this->polls === 2 ? 'other' : 'original';

return ['poll_status' => 'leased', 'task' => [
Expand Down Expand Up @@ -63,18 +69,24 @@ public function send(string $method, string $uri, array $headers, ?array $body =
}
}

/** @param Closure(Transport): Worker $makeWorker */
function verifyConfiguredHistoryCache(Closure $makeWorker): void
/**
* @param Closure(Transport): Worker $makeWorker
* @return array{hit: int, miss: int, eviction: int, forced_cold_replay: int}
*/
function verifyConfiguredHistoryCache(Closure $makeWorker): array
{
$transport = new ConfiguredHistoryCacheTransport();
$worker = $makeWorker($transport);
$worker->registerWorkflow('cache.configuration', static fn (WorkflowContext $context): string => 'same-result');
for ($index = 0; $index < 12 && $transport->completions < 3; ++$index) {
$worker->tick(0);
}
$worker->run(0);
$metrics = $worker->stickyCacheMetrics();
if ($transport->historyFetches !== 1 || $transport->completions !== 3
|| $metrics['eviction'] < 1 || $metrics['forced_cold_replay'] !== 1) {
throw new RuntimeException('Configured capacity did not evict and fetch authoritative cold history.');
}

return $metrics;
}
15 changes: 13 additions & 2 deletions tests/compat/laravel.php
Original file line number Diff line number Diff line change
Expand Up @@ -258,16 +258,27 @@ function laravelAuthenticationRejectsRole(Authentication $authentication, bool $
throw new RuntimeException('Laravel provider did not preserve role-specific client credentials.');
}
$worker = $factory->make();
verifyConfiguredHistoryCache(static function (Transport $transport) use ($app, $logger): DurableWorkflow\Worker {
$cacheStoppedContext = null;
$cacheEvents = new Dispatcher($app);
$cacheEvents->listen(WorkerDiagnosticEvent::class, static function (WorkerDiagnosticEvent $event) use (&$cacheStoppedContext): void {
if ($event->name === 'worker.stopped') {
$cacheStoppedContext = $event->context;
}
});
$cacheMetrics = verifyConfiguredHistoryCache(static function (Transport $transport) use ($app, $cacheEvents): DurableWorkflow\Worker {
$configuration = $app->make(DurableWorkflow\Bridge\ServiceConfiguration::class);

return (new WorkerFactory(
$app,
$configuration,
$configuration->workerClient($transport),
$logger,
new LaravelBridgeLogger(),
$cacheEvents,
))->make();
});
if (($cacheStoppedContext['sticky_cache'] ?? null) !== $cacheMetrics) {
throw new RuntimeException('Laravel stopped event omitted the actual lifetime history-cache counters.');
}
if ($worker->contracts()['workflows'] !== ['laravel.greeting', 'laravel.compensated-trip']
|| $worker->contracts()['activities'] !== [
'laravel.greet',
Expand Down
11 changes: 10 additions & 1 deletion tests/compat/symfony.php
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,12 @@ public function fake(): WorkflowClientFake
throw new RuntimeException('Symfony extension did not preserve role-specific client credentials.');
}
$worker = $factory->make();
verifyConfiguredHistoryCache(static function (Transport $transport): DurableWorkflow\Worker {
$cacheStoppedContext = null;
$cacheEvents = new EventDispatcher();
$cacheEvents->addListener('worker.stopped', static function (WorkerDiagnosticEvent $event) use (&$cacheStoppedContext): void {
$cacheStoppedContext = $event->context;
});
$cacheMetrics = verifyConfiguredHistoryCache(static function (Transport $transport) use ($cacheEvents): DurableWorkflow\Worker {
$configuration = DurableWorkflow\Bridge\ServiceConfiguration::fromArray([
'endpoint' => 'http://localhost:8080',
'task_queue' => 'symfony-workers',
Expand All @@ -185,8 +190,12 @@ public function fake(): WorkflowClientFake
$configuration,
$configuration->workerClient($transport),
[new SymfonyGreetingHandler(new SymfonyGreetingDependency())],
events: $cacheEvents,
))->make();
});
if (($cacheStoppedContext['sticky_cache'] ?? null) !== $cacheMetrics) {
throw new RuntimeException('Symfony stopped event omitted the actual lifetime history-cache counters.');
}
if ($worker->contracts()['workflows'] !== ['symfony.greeting']
|| $worker->contracts()['activities'] !== ['symfony.greet']
) {
Expand Down
Loading