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
7 changes: 7 additions & 0 deletions scripts/ci/run-replay-regression-fixture.php
Original file line number Diff line number Diff line change
Expand Up @@ -1005,6 +1005,13 @@ private static function workflow(string $workflowType): callable

return ['failure' => $message ?? 'missing', 'failed_time' => $failedTime, 'time' => $time];
},
'golden.business-replay-clock-group-condition' => static function (WorkflowContext $context): array {
$values = $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition'), $context->deferTimer(2)]);
try { $time = $context->replayNow()->format(self::CLOCK_FORMAT); }
catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { $time = null; }

return ['values' => $values, 'time' => $time];
},
'golden.business-replay-clock-group-child' => static function (WorkflowContext $context): array {
$values = $context->all([$context->deferChildWorkflow('clock-child'), $context->deferTimer(2)]);
try { $time = $context->replayNow()->format(self::CLOCK_FORMAT); }
Expand Down
10 changes: 6 additions & 4 deletions src/Worker/BusinessReplayClock.php
Original file line number Diff line number Diff line change
Expand Up @@ -145,14 +145,16 @@ private function groupMemberTime(array $step, array $path, Closure $pathMatches)
$family = $step['shape'] ?? null;
$index = $step['resolution_order'] ?? null;
$sequence = $step['sequence'] ?? null;
if (!in_array($family, ['activity', 'timer', self::CHILD_SHAPE], true) || !is_int($index) || !is_int($sequence)
if (!in_array($family, ['activity', 'timer', self::CHILD_SHAPE, 'condition_wait'], true) || !is_int($index) || !is_int($sequence)
|| !isset($this->history[$index]) || $path === []) {
return null;
}
$rowMatches = fn (array $body, string $kind): bool => $this->groupRow($body, $kind, $path, $pathMatches);
$matches = $family === self::CHILD_SHAPE
? ChildReplayClockEvidence::matches($this->history, $this->bySequence[$sequence] ?? [], $index, $sequence, $rowMatches)
: $this->scalarBoundary($index, $family, $sequence, $rowMatches);
$matches = match ($family) {
self::CHILD_SHAPE => ChildReplayClockEvidence::matches($this->history, $this->bySequence[$sequence] ?? [], $index, $sequence, $rowMatches),
'condition_wait' => ConditionReplayClockEvidence::matches($this->history, $index, $sequence, $rowMatches),
default => $this->scalarBoundary($index, $family, $sequence, $rowMatches),
};
if (!$matches) {
return null;
}
Expand Down
20 changes: 15 additions & 5 deletions src/Worker/ConditionReplayClockEvidence.php
Original file line number Diff line number Diff line change
Expand Up @@ -4,21 +4,25 @@

namespace DurableWorkflow\Worker;

use Closure;

/** @internal Proves one consumed physical condition wait, including its timeout pair. */
final class ConditionReplayClockEvidence
{
private const TERMINALS = ['ConditionWaitSatisfied', 'ConditionWaitTimedOut'];

/** @param list<array<string, mixed>> $history */
public static function matches(array $history, int $index, int $sequence): bool
/** @param list<array<string, mixed>> $history
* @param (Closure(array<string, mixed>, string): bool)|null $rowMatches
*/
public static function matches(array $history, int $index, int $sequence, ?Closure $rowMatches = null): bool
{
$event = $history[$index];
$payload = $event['payload'] ?? null;
$waitId = is_array($payload) ? ($payload['condition_wait_id'] ?? null) : null;
if (!is_array($payload) || !ReplayClockHistory::identity($waitId) || ReplayClockHistory::sequence($payload) !== $sequence) {
return false;
}
$rows = self::linkedRows($history, $sequence, $waitId);
$rows = self::linkedRows($history, $sequence, $waitId, $rowMatches ?? ReplayClockHistory::scalarRow(...));

return $rows !== null && self::lifecycle($history, $rows, $index);
}
Expand All @@ -41,9 +45,10 @@ private static function lifecycle(array $history, array $rows, int $index): bool
}

/** @param list<array<string, mixed>> $history
* @param Closure(array<string, mixed>, string): bool $rowMatches
* @return array<string, list<int>>|null
*/
private static function linkedRows(array $history, int $sequence, string $waitId): ?array
private static function linkedRows(array $history, int $sequence, string $waitId, Closure $rowMatches): ?array
{
$positions = [];
$allowed = ['ConditionWaitOpened', ...self::TERMINALS, 'TimerScheduled', 'TimerFired', 'TimerCancelled'];
Expand All @@ -55,8 +60,13 @@ private static function linkedRows(array $history, int $sequence, string $waitId
|| (ReplayClockHistory::legacySequenceClaim($body) !== $sequence && ($body['condition_wait_id'] ?? null) !== $waitId)) {
continue;
}
// Core's condition-timeout timers carry the wait identity, without
// the owning condition's parallel path. The complete timer pair
// must remain scalar and correlate with that validated wait.
$rowQualified = str_starts_with((string) $kind, 'Timer')
? ReplayClockHistory::scalarRow($body, 'condition_wait') : $rowMatches($body, 'condition_wait');
if (!in_array($kind, $allowed, true) || ReplayClockHistory::sequence($body) !== $sequence
|| ($body['condition_wait_id'] ?? null) !== $waitId || !ReplayClockHistory::scalarRow($body, 'condition_wait')
|| ($body['condition_wait_id'] ?? null) !== $waitId || !$rowQualified
|| ($body['cancellation_scope'] ?? null) !== null) {
return null;
}
Expand Down
186 changes: 186 additions & 0 deletions tests/ConditionGroupBusinessReplayClockTest.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,186 @@
<?php

declare(strict_types=1);

namespace DurableWorkflow\Tests;

use DurableWorkflow\Codec\AvroPayloadCodec;
use DurableWorkflow\Exception\NonDeterministicWorkflow;
use DurableWorkflow\Exception\ReplayClockUnavailable;
use DurableWorkflow\Worker\Replayer;
use DurableWorkflow\Worker\WorkflowClaimAborted;
use DurableWorkflow\Worker\WorkflowContext;
use PHPUnit\Framework\Attributes\DataProvider;
use PHPUnit\Framework\TestCase;

final class ConditionGroupBusinessReplayClockTest extends TestCase
{
private const FORMAT = 'Y-m-d\TH:i:s.uP';
private const FIRST = '2026-11-06T00:00:00.123400+00:00';
private const SECOND = '2026-11-06T00:00:00.123499+00:00';

#[DataProvider('conditionOutcomes')]
public function testPersistedConditionGroupPreservesResultsAndExactAuthoredTie(string $terminal, bool $backed): void
{
$handler = self::handler();
$history = self::history($handler, $terminal, $backed);
$expected = [[$terminal === 'ConditionWaitSatisfied', null], self::FIRST];

self::assertSame($expected, $this->replay($handler, $history));
self::assertSame($expected, $this->replay($handler, $history));
}

public static function conditionOutcomes(): array
{
return [['ConditionWaitSatisfied', false], ['ConditionWaitSatisfied', true], ['ConditionWaitTimedOut', true]];
}

public function testConditionAndSiblingUseTheirOwnConsumedTimes(): void
{
$handler = self::handler();
$history = self::history($handler, times: [self::FIRST, '2026-11-06T00:00:00.124500Z']);
// A later, unconsumed condition is not part of the first group's fold.
$history[] = ['event_type' => 'ConditionWaitSatisfied', 'recorded_at' => '2099-01-01T00:00:00Z',
'payload' => ['sequence' => 3, 'condition_wait_id' => 'future']];

self::assertSame([[true, null], '2026-11-06T00:00:00.124500+00:00'], $this->replay($handler, $history));
}

public function testTwoConditionsCannotBorrowOnePhysicalWait(): void
{
$handler = self::handler(true);
$history = self::history($handler);
foreach ($history as &$event) {
if (($event['payload']['sequence'] ?? null) === 2) {
$event['payload']['condition_wait_id'] = 'wait-1';
}
}
unset($event);

self::assertSame([[true, true], null], $this->replay($handler, $history));
}

public function testNestedConditionGroupsUseTheAuthoredLeafPaths(): void
{
$handler = static function (WorkflowContext $context): array {
$values = $context->all([static fn (): array => $context->all([
$context->deferCondition(static fn (): bool => true, 'clock-condition', 5), $context->deferTimer(2),
]), $context->deferTimer(2)]);

return [$values, self::read($context)];
};

self::assertSame([[[true, null], null], self::FIRST], $this->replay($handler,
self::history($handler, times: [self::FIRST, self::SECOND, self::SECOND])));
}

#[DataProvider('unprovedRows')]
public function testUnprovedConditionLifecycleCannotSupplyTheGroupClock(string $mutation): void
{
$handler = self::handler();
$history = self::history($handler, backed: true);
switch ($mutation) {
case 'terminal-path': $history[3]['payload']['parallel_group_id'] = 'foreign'; break;
case 'terminal-scope': $history[3]['payload']['cancellation_scope_id'] = 'foreign'; break;
case 'terminal-local': $history[3]['payload']['execution_mode'] = 'local'; break;
case 'foreign-wait': $history[3]['payload']['condition_wait_id'] = 'foreign'; break;
case 'foreign-definition': $history[3]['payload']['condition_definition_fingerprint'] = 'foreign'; break;
case 'foreign-occurrence': $history[3]['payload']['condition_wait_occurrence_id'] = 'foreign'; break;
case 'timer-resource': $history[2]['payload']['timer_id'] = 'foreign'; break;
case 'timer-kind': $history[2]['payload']['timer_kind'] = 'signal_timeout'; break;
case 'timer-path': $history[1]['payload']['parallel_group_id'] = 'foreign'; break;
case 'duplicate-timer': array_splice($history, 2, 0, [$history[1]]); break;
case 'duplicate-terminal': $history[] = $history[3]; break;
case 'coerced-duplicate':
$duplicate = $history[3]; $duplicate['payload']['sequence'] = '1'; $history[] = $duplicate; break;
case 'bad-time': $history[3]['recorded_at'] = '2026-02-30T00:00:00Z'; break;
}

if (in_array($mutation, ['terminal-path', 'terminal-scope'], true)) {
$this->expectException($mutation === 'terminal-path' ? NonDeterministicWorkflow::class : WorkflowClaimAborted::class);
}
self::assertSame([[true, null], null], $this->replay($handler, $history));
}

public static function unprovedRows(): array
{
return array_map(static fn (string $mutation): array => [$mutation], ['terminal-path', 'terminal-scope',
'terminal-local', 'foreign-wait', 'foreign-definition', 'foreign-occurrence', 'timer-resource',
'timer-kind', 'timer-path', 'duplicate-timer', 'duplicate-terminal', 'coerced-duplicate', 'bad-time']);
}

public function testACompletedGroupCannotLeaveItsClockAvailableOutsideExecution(): void
{
$retained = null;
$handler = static function (WorkflowContext $context) use (&$retained): ?string {
$retained = $context;
$context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition', 5), $context->deferTimer(2)]);

return self::read($context);
};
self::assertSame(self::FIRST, $this->replay($handler, self::history(self::handler())));
$this->expectException(ReplayClockUnavailable::class);
$retained->replayNow();
}

private static function handler(bool $conditionsOnly = false): callable
{
return static function (WorkflowContext $context) use ($conditionsOnly): array {
$values = $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition', 5),
$conditionsOnly ? $context->deferCondition(static fn (): bool => true, 'clock-condition', 5) : $context->deferTimer(2)]);

return [$values, self::read($context)];
};
}

private static function read(WorkflowContext $context): ?string
{
try { return $context->replayNow()->format(self::FORMAT); }
catch (ReplayClockUnavailable) { return null; }
}

/** @param list<string> $times
* @return list<array<string, mixed>>
*/
private static function history(callable $handler, string $terminal = 'ConditionWaitSatisfied', bool $backed = false,
array $times = [self::FIRST, self::SECOND]): array
{
$commands = (new Replayer(new AvroPayloadCodec()))->replay($handler, [], [], 'php-workers')->commands;
$history = [];
foreach ($commands as $offset => $command) {
$sequence = $offset + 1;
$metadata = array_intersect_key($command, array_flip(['parallel_group_id', 'parallel_group_kind', 'parallel_group_mode',
'parallel_group_base_sequence', 'parallel_group_size', 'parallel_group_index', 'parallel_group_path']));
$body = ['sequence' => $sequence];
if ($command['type'] === 'open_condition_wait') {
$body += ['condition_wait_id' => 'wait-'.$sequence, 'condition_key' => $command['condition_key'],
'condition_definition_fingerprint' => $command['condition_definition_fingerprint'], 'timeout_seconds' => 5];
$history[] = ['event_type' => 'ConditionWaitOpened', 'payload' => $body + $metadata];
if ($backed) {
// Core's timeout writer has wait identity but no parallel metadata.
$timer = $body + ['timer_id' => 'timer-'.$sequence, 'timer_kind' => 'condition_timeout', 'delay_seconds' => 5];
$history[] = ['event_type' => 'TimerScheduled', 'payload' => $timer];
$history[] = ['event_type' => $terminal === 'ConditionWaitTimedOut' ? 'TimerFired' : 'TimerCancelled',
'recorded_at' => $times[$offset], 'payload' => $timer];
$body['timer_id'] = $timer['timer_id'];
}
$history[] = ['event_type' => $terminal, 'recorded_at' => $times[$offset], 'payload' => $body + $metadata];
} else {
$body += ['timer_id' => 'timer-'.$sequence, 'delay_seconds' => 2];
$history[] = ['event_type' => 'TimerScheduled', 'payload' => $body + $metadata];
$history[] = ['event_type' => 'TimerFired', 'recorded_at' => $times[$offset], 'payload' => $body + $metadata];
}
}

return $history;
}

/** @param list<array<string, mixed>> $history */
private function replay(callable $handler, array $history): mixed
{
$codec = new AvroPayloadCodec();
$result = (new Replayer($codec))->replay($handler, $history, [], 'php-workers', ['workflow_id' => 'parent', 'run_id' => 'run']);

return $codec->decodeEnvelope($result->commands[array_key_last($result->commands)]['result']);
}
}
7 changes: 7 additions & 0 deletions tests/Support/ReplayRegressionFixture.php
Original file line number Diff line number Diff line change
Expand Up @@ -690,6 +690,13 @@ private static function workflow(string $workflowType): callable

return ['failure' => $message ?? 'missing', 'failed_time' => $failedTime, 'time' => $time];
},
'golden.business-replay-clock-group-condition' => static function (WorkflowContext $context): array {
$values = $context->all([$context->deferCondition(static fn (): bool => true, 'clock-condition'), $context->deferTimer(2)]);
try { $time = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); }
catch (\DurableWorkflow\Exception\ReplayClockUnavailable) { $time = null; }

return ['values' => $values, 'time' => $time];
},
'golden.business-replay-clock-group-child' => static function (WorkflowContext $context): array {
$values = $context->all([$context->deferChildWorkflow('clock-child'), $context->deferTimer(2)]);
try { $time = $context->replayNow()->format('Y-m-d\TH:i:s.uP'); }
Expand Down
Loading
Loading