diff --git a/tests/integration/Durable/Crash/CrashBenchActivities.php b/tests/integration/Durable/Crash/CrashBenchActivities.php new file mode 100644 index 00000000..ba587f38 --- /dev/null +++ b/tests/integration/Durable/Crash/CrashBenchActivities.php @@ -0,0 +1,29 @@ +directory = sys_get_temp_dir() . '/durable-crash-bench-' . bin2hex(random_bytes(6)); + mkdir($this->directory, 0o700, true); + } + + protected function tearDown(): void + { + foreach (glob($this->directory . '/*') ?: [] as $file) { + unlink($file); + } + if (is_dir($this->directory)) { + rmdir($this->directory); + } + } + + #[Test] + public function anExecutionKilledBetweenTwoActivitiesResumesInAnotherProcessWithoutRepayingTheFirst(): void + { + $journal = $this->directory . '/journal.sqlite'; + $log = $this->directory . '/activities.log'; + $executionId = '01900000-0000-7000-8000-0000000000c1'; + + $crashed = $this->runSlice($journal, $executionId, 'start', $log, kill: true); + + // A process the kernel killed outright never gets to report an exit code of its own; what + // matters is only that it is not zero, because zero would mean it finished the workflow and + // there is nothing left to resume. + self::assertNotSame(0, $crashed['code'], 'the first process must die, not finish'); + self::assertSame( + ['bench.first'], + $this->activitiesRun($log), + 'the first process runs the first activity and dies before scheduling the second', + ); + + $resumed = $this->runSlice($journal, $executionId, 'resume', $log); + + self::assertSame(0, $resumed['code'], 'the second process must carry the execution to its end: ' . $resumed['stderr']); + self::assertSame( + ['first' => 'first:a', 'second' => 'second:b'], + json_decode(trim($resumed['stdout']), true, 512, JSON_THROW_ON_ERROR), + 'the resumed execution returns what an uninterrupted one would have returned', + ); + + // The whole bench is this one line. `bench.first` appears once and only once: it was run by + // a process that no longer exists, and the process that finished the job was served its + // result out of the journal instead of paying for it a second time. Anything the first + // process had kept in memory rather than in the journal is gone, and the execution did not + // need it. + self::assertSame( + ['bench.first', 'bench.second'], + $this->activitiesRun($log), + 'the first activity is served from the journal, never re-executed', + ); + } + + /** + * @return list + */ + private function activitiesRun(string $log): array + { + return array_values(array_filter(explode("\n", (string) @file_get_contents($log)))); + } + + /** + * @return array{code: int, stdout: string, stderr: string} + */ + private function runSlice(string $journal, string $executionId, string $phase, string $log, bool $kill = false): array + { + $environment = ['BENCH_LOG' => $log] + ($kill ? ['BENCH_KILL' => '1'] : []); + + $process = proc_open( + [PHP_BINARY, __DIR__ . '/bench_slice.php', $journal, $executionId, $phase], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], + $pipes, + null, + $environment + ['PATH' => getenv('PATH') ?: '/usr/bin:/bin'], + ); + + self::assertIsResource($process, 'the bench must be able to spawn a second PHP process'); + + $stdout = (string) stream_get_contents($pipes[1]); + $stderr = (string) stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + + return ['code' => proc_close($process), 'stdout' => $stdout, 'stderr' => $stderr]; + } +} diff --git a/tests/integration/Durable/Crash/bench_slice.php b/tests/integration/Durable/Crash/bench_slice.php new file mode 100644 index 00000000..60c7d46a --- /dev/null +++ b/tests/integration/Durable/Crash/bench_slice.php @@ -0,0 +1,116 @@ + start|resume + * + * Environment: + * BENCH_LOG file appended to, one line per activity actually executed + * BENCH_KILL set to 1 to SIGKILL the process between the two activities (optional) + * + * Exit codes: 0 finished, 3 still suspended, 4 usage. A SIGKILL leaves no exit code at all, which + * is what the caller asserts on. + * + * **The kill is between the activities, not inside one**, and that is a deliberate narrowing. A + * worker that dies *while* an activity is in flight leaves a scheduled slot with no outcome, and + * what should happen then is a redelivery — a property of the activity transport and its retry + * policy, neither of which this bare setup has. Measured on the way here: with no transport worker + * and `maxActivityRetries: 0`, resuming such an execution waits forever, which is the correct + * behaviour for a runtime that has nobody to ask. It is a real question, it is not *this* question, + * and answering both at once would leave neither answered. + */ + +use Doctrine\DBAL\DriverManager; +use Gplanchat\Bridge\Dbal\Schema\DurableSchema; +use Gplanchat\Bridge\Dbal\Store\DbalEventStore; +use Gplanchat\Durable\Exception\WorkflowSuspendedException; +use Gplanchat\Durable\ExecutionEngine; +use Gplanchat\Durable\ExecutionRuntime; +use Gplanchat\Durable\RegistryActivityExecutor; +use Gplanchat\Durable\Transport\InMemoryActivityTransport; +use Gplanchat\Durable\WorkflowEnvironment; +use integration\Durable\Crash\CrashBenchActivities; + +require __DIR__ . '/../../../../vendor/autoload.php'; + +[$journalPath, $executionId, $phase] = [$argv[1] ?? null, $argv[2] ?? null, $argv[3] ?? null]; + +if (null === $journalPath || null === $executionId || !\in_array($phase, ['start', 'resume'], true)) { + fwrite(STDERR, "usage: bench_slice.php start|resume\n"); + exit(4); +} + +$log = getenv('BENCH_LOG') ?: null; +$killBetween = '1' === getenv('BENCH_KILL'); + +/** + * What an activity does when it really runs — and the whole assertion of this bench is that the + * second process never writes the first activity's line. + */ +$record = static function (string $name) use ($log): void { + if (null !== $log) { + file_put_contents($log, $name . "\n", FILE_APPEND | LOCK_EX); + } +}; + +$connection = DriverManager::getConnection(['driver' => 'pdo_sqlite', 'path' => $journalPath]); +$eventStore = new DbalEventStore($connection, new DurableSchema($connection)); + +$executor = new RegistryActivityExecutor(); +$executor->register('bench.first', static function (array $payload) use ($record): string { + $record('bench.first'); + + return 'first:' . ($payload['tag'] ?? '?'); +}); +$executor->register('bench.second', static function (array $payload) use ($record): string { + $record('bench.second'); + + return 'second:' . ($payload['tag'] ?? '?'); +}); + +// The handler is defined here, in the file both processes run, because replay demands that the two +// processes execute the same workflow code. A closure that differed between them would fail the +// divergence guard rather than the bench — a different measurement wearing this one's name. +// +// The kill sits between the two awaits, and it is not a workflow decision: it schedules nothing, +// journals nothing and returns nothing. It is `kill -9` arriving at the one instant where the +// journal holds the first activity's outcome and knows nothing yet of the second. Reaching in from +// outside would have to race that instant; reaching in from here hits it exactly. +$handler = static function (WorkflowEnvironment $env) use ($killBetween): array { + $first = $env->await($env->activityStub(CrashBenchActivities::class)->first('a')); + + if ($killBetween) { + // SIGKILL and not exit(): no destructors, no shutdown functions, no chance for anything to + // flush a buffer that a real crash would have taken with it. A worker that is OOM-killed or + // whose container is stopped gets exactly this much warning. + posix_kill(posix_getpid(), SIGKILL); + } + + $second = $env->await($env->activityStub(CrashBenchActivities::class)->second('b')); + + return ['first' => $first, 'second' => $second]; +}; + +$runtime = new ExecutionRuntime($eventStore, new InMemoryActivityTransport(), $executor); +$engine = new ExecutionEngine($eventStore, $runtime); + +try { + $result = 'start' === $phase + ? $engine->start($executionId, $handler, 'CrashBench') + : $engine->resume($executionId, $handler, 'CrashBench'); +} catch (WorkflowSuspendedException) { + exit(3); +} + +echo json_encode($result, JSON_THROW_ON_ERROR), "\n"; +exit(0);