From 926abb63eb3f4e0e9da2f830e828819a7b089026 Mon Sep 17 00:00:00 2001 From: Adhik Joshi Date: Thu, 13 Aug 2026 19:02:53 +0530 Subject: [PATCH] perf: stop reading every task_result payload to prune keys Redis already expires pruneOldTaskResults() SCANned the whole keyspace every PRUNE_CHECK_INTERVAL and GET the JSON of every task_result key just to compare one timestamp. It could never delete anything: task_result keys are written with a TTL (setex 3600 in storeResult(), TASK_RESULT_RETENTION in the webhook paths), while the prune only deletes when now - finished_at exceeds TASK_RESULT_RETENTION. Redis expiry always wins that race. Measured on a production shard over 26 days: 196.7M SCAN + 228.8M GET to issue 2 DELs -- ~2.2 hours of blocked event loop and 13.5TB of network output. Across the fleet that scan was roughly 85% of all Redis CPU time. Sampling 1,200 live task_result keys found zero without a TTL. Drop the call from the startWorkers() loop, since Redis TTL is the mechanism. Keep the method as a manual repair entry point (now public) for the one case Redis cannot handle by itself: a key whose TTL went missing. That version pipelines TTL, an 8-byte reply, and reads a value only for keys already known to be broken, so a healthy keyspace costs zero payload transfers. Deletions use UNLINK so multi-MB frees land off the main thread. Bulk-reading with MGET would be worse, not better: task_result payloads reach 7MB, so a batched read builds one huge client output buffer, which is what pushes RSS past the container limit and gets redis-server OOM-killed. Adds integration coverage asserting the control case (50 healthy keys cost 0 value reads), that exactly one read happens for one leaked key, that a recent orphan is bounded rather than deleted, and that the worker loop no longer calls the scan. --- .gitignore | 1 + src/ModelQ.php | 138 +++++++++++++++++---- tests/Integration/PruneTaskResultsTest.php | 130 +++++++++++++++++++ 3 files changed, 242 insertions(+), 27 deletions(-) create mode 100644 tests/Integration/PruneTaskResultsTest.php diff --git a/.gitignore b/.gitignore index 448a594..9f9a48e 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ phpstan.neon .vscode/ *.log modelq_errors.log +.phpunit.cache/ diff --git a/src/ModelQ.php b/src/ModelQ.php index 581fe70..f31ac15 100644 --- a/src/ModelQ.php +++ b/src/ModelQ.php @@ -29,6 +29,7 @@ class ModelQ public const TASK_HISTORY_RETENTION = 86400; // 24 hours (configurable) public const TASK_TTL = 86400; // 24 hours TTL for all tasks public const DEFAULT_STREAM_TIMEOUT = 300; // 5 minutes default stream timeout + public const SCAN_BATCH = 500; // keys per SCAN round-trip private Redis $redis; private string $serverId; @@ -247,7 +248,11 @@ public function startWorkers(int $workers = 1): void if ($now - $lastPrune >= self::PRUNE_CHECK_INTERVAL) { $this->pruneInactiveServers(); $this->requeueStuckProcessingTasks(); - $this->pruneOldTaskResults(); + // pruneOldTaskResults() is deliberately NOT called here. Every + // task_result key is written with a TTL, so Redis expires it on + // its own; running the scan every PRUNE_CHECK_INTERVAL only + // re-read the whole keyspace. Measured on a production shard + // over 26 days: 196.7M SCAN + 228.8M GET to issue 2 DELs. $lastPrune = $now; } @@ -957,36 +962,40 @@ private function requeueStuckProcessingTasks(?float $threshold = null): void } /** - * Prune old task results. + * Delete `task_result:*` keys (and their `task:*` twin) that lost their TTL + * and are older than $olderThanSeconds. Returns the number pruned. + * + * Expiry is Redis's job. Every task_result key is written with a TTL (see + * storeResult() and the webhook paths), so a key that still has one needs + * nothing from us. This only has to catch keys whose TTL went missing -- a + * write path that forgot the EX, a RENAME/RESTORE that dropped it -- which + * would otherwise live forever. + * + * The cost is in deciding *which* keys to read, not in the delete. TTL is an + * 8-byte reply and is pipelined, so a healthy keyspace is walked without + * transferring a single payload; only keys already known to be broken get + * read. The previous version GET the JSON of every key to compare one + * timestamp. Measured on a production shard over 26 days that was 196.7M + * SCAN + 228.8M GET -- roughly 2.2 hours of blocked event loop and 13.5TB of + * network output -- to issue 2 deletes. + * + * Bulk-reading with MGET would make this worse, not better: task_result + * payloads reach 7MB, so a batched read builds one huge client output + * buffer, and that is what pushes RSS past the container memory limit and + * gets redis-server OOM-killed. */ - private function pruneOldTaskResults(?int $olderThanSeconds = null): void + public function pruneOldTaskResults(?int $olderThanSeconds = null): int { $olderThanSeconds = $olderThanSeconds ?? self::TASK_RESULT_RETENTION; $now = microtime(true); - $keysDeleted = 0; + $pruned = 0; $iterator = null; - while ($keys = $this->redis->scan($iterator, 'task_result:*', 100)) { - foreach ($keys as $key) { - try { - $taskJson = $this->redis->get($key); - if (!$taskJson) { - continue; - } - - $taskData = json_decode($taskJson, true); - $timestamp = $taskData['finished_at'] ?? $taskData['started_at'] ?? null; - - if ($timestamp && ($now - $timestamp) > $olderThanSeconds) { - $this->redis->del($key); - $taskId = str_replace('task_result:', '', $key); - $this->redis->del("task:{$taskId}"); - $keysDeleted++; - $this->logger->info("Deleted old keys: {$key} and task:{$taskId}"); - } - } catch (Throwable $e) { - $this->logger->error("Error processing key {$key}: " . $e->getMessage()); - } + while ($keys = $this->redis->scan($iterator, 'task_result:*', self::SCAN_BATCH)) { + try { + $pruned += $this->pruneUntrackedKeys($keys, $now, $olderThanSeconds); + } catch (Throwable $e) { + $this->logger->error('Error pruning task_result batch: ' . $e->getMessage()); } if ($iterator === 0) { @@ -994,9 +1003,84 @@ private function pruneOldTaskResults(?int $olderThanSeconds = null): void } } - if ($keysDeleted > 0) { - $this->logger->info("Pruned {$keysDeleted} task(s) older than {$olderThanSeconds} seconds."); + if ($pruned > 0) { + $this->logger->info("Pruned {$pruned} task(s) older than {$olderThanSeconds} seconds."); } + + return $pruned; + } + + /** + * Prune the keys in one SCAN batch that have no TTL. Returns the number deleted. + * + * @param array $keys + */ + private function pruneUntrackedKeys(array $keys, float $now, int $olderThanSeconds): int + { + $pipe = $this->redis->multi(Redis::PIPELINE); + foreach ($keys as $key) { + $pipe->ttl($key); + } + $ttls = $pipe->exec(); + + // -1 == exists with no expiry (leaked). -2 == already gone. + // Anything with a TTL is Redis's problem, not ours. + $orphans = []; + foreach ($keys as $i => $key) { + if (($ttls[$i] ?? null) === -1) { + $orphans[] = $key; + } + } + + if ($orphans === []) { + return 0; + } + + // Only now do we read values, and only for the broken keys. + $pipe = $this->redis->multi(Redis::PIPELINE); + foreach ($orphans as $key) { + $pipe->get($key); + } + $blobs = $pipe->exec(); + + $expired = []; + $keep = []; + foreach ($orphans as $i => $key) { + $blob = $blobs[$i] ?? null; + $taskData = is_string($blob) ? json_decode($blob, true) : null; + $timestamp = is_array($taskData) + ? ($taskData['finished_at'] ?? $taskData['started_at'] ?? null) + : null; + + if ($timestamp && ($now - $timestamp) > $olderThanSeconds) { + $expired[] = $key; + } else { + $keep[] = $key; + } + } + + $pipe = $this->redis->multi(Redis::PIPELINE); + foreach ($expired as $key) { + $taskId = str_replace('task_result:', '', $key); + // UNLINK, not DEL: frees multi-MB payloads off the main thread. + $pipe->unlink($key); + $pipe->unlink("task:{$taskId}"); + } + foreach ($keep as $key) { + // Not old enough to drop, but it must not live forever. + $pipe->expire($key, $olderThanSeconds); + } + $pipe->exec(); + + foreach ($expired as $key) { + $taskId = str_replace('task_result:', '', $key); + $this->logger->info("Pruned untracked {$key} and task:{$taskId}"); + } + foreach ($keep as $key) { + $this->logger->warning("task_result key had no TTL, set to {$olderThanSeconds}s: {$key}"); + } + + return count($expired); } /** diff --git a/tests/Integration/PruneTaskResultsTest.php b/tests/Integration/PruneTaskResultsTest.php new file mode 100644 index 0000000..abeab4c --- /dev/null +++ b/tests/Integration/PruneTaskResultsTest.php @@ -0,0 +1,130 @@ +getCalls++; + + return parent::get($key); + } +} + +class PruneTaskResultsTest extends TestCase +{ + private CountingRedis $redis; + + private ModelQ $modelq; + + protected function setUp(): void + { + $this->redis = new CountingRedis(); + $this->redis->connect('127.0.0.1', 6379); + $this->redis->flushDb(); + $this->modelq = new ModelQ(redisClient: $this->redis); + $this->redis->getCalls = 0; + } + + protected function tearDown(): void + { + $this->redis->flushDb(); + $this->redis->close(); + } + + private function payload(float $finishedAt): string + { + return (string) json_encode(['status' => 'completed', 'finished_at' => $finishedAt]); + } + + /** + * The control case: a keyspace of healthy keys costs zero value reads. + * + * This is the whole point of the rewrite. The old implementation GET every + * key on every pass; production showed 228.8M GETs to issue 2 deletes. + */ + public function testHealthyKeysAreNeverRead(): void + { + for ($i = 0; $i < 50; $i++) { + $this->redis->setex("task_result:h{$i}", 3600, $this->payload(microtime(true))); + } + $this->redis->getCalls = 0; + + $pruned = $this->modelq->pruneOldTaskResults(86400); + + $this->assertSame(0, $pruned); + $this->assertSame(0, $this->redis->getCalls, 'healthy keys must never have their value read'); + + for ($i = 0; $i < 50; $i++) { + $this->assertNotFalse($this->redis->get("task_result:h{$i}")); + $this->assertGreaterThan(0, $this->redis->ttl("task_result:h{$i}")); + } + } + + /** + * One broken key among many healthy ones costs exactly one read. + */ + public function testOnlyTheKeyThatLostItsTtlIsRead(): void + { + for ($i = 0; $i < 50; $i++) { + $this->redis->setex("task_result:h{$i}", 3600, $this->payload(microtime(true))); + } + // No TTL, and old enough to prune. + $this->redis->set('task_result:leaked', $this->payload(microtime(true) - 90000)); + $this->redis->set('task:leaked', $this->payload(microtime(true) - 90000)); + $this->redis->getCalls = 0; + + $pruned = $this->modelq->pruneOldTaskResults(86400); + + $this->assertSame(1, $pruned); + $this->assertSame(1, $this->redis->getCalls, 'only the TTL-less key should be read'); + $this->assertFalse($this->redis->get('task_result:leaked')); + $this->assertFalse($this->redis->get('task:leaked'), 'the task: twin must go too'); + } + + /** + * A key with no TTL but not yet old is kept, and stops living forever. + */ + public function testRecentOrphanIsBoundedNotDeleted(): void + { + $this->redis->set('task_result:fresh', $this->payload(microtime(true))); + $this->assertSame(-1, $this->redis->ttl('task_result:fresh')); + + $pruned = $this->modelq->pruneOldTaskResults(86400); + + $this->assertSame(0, $pruned); + $this->assertNotFalse($this->redis->get('task_result:fresh')); + $this->assertGreaterThan(0, $this->redis->ttl('task_result:fresh')); + } + + /** + * The 60s worker loop must not call the scan; Redis TTLs handle expiry. + */ + public function testWorkerLoopDoesNotScanTaskResults(): void + { + $method = new ReflectionMethod(ModelQ::class, 'startWorkers'); + $lines = (array) file((string) $method->getFileName()); + $body = implode('', array_slice( + $lines, + $method->getStartLine() - 1, + $method->getEndLine() - $method->getStartLine() + 1 + )); + $body = (string) preg_replace('#^\s*//.*$#m', '', $body); + + $this->assertStringNotContainsString('pruneOldTaskResults', $body); + } +}