perf: stop reading every task_result payload to prune keys Redis already expires

.gitignore

@@ -6,3 +6,4 @@ phpstan.neon
 .vscode/
 *.log
 modelq_errors.log
+.phpunit.cache/

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,46 +962,125 @@ 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) {
  break;
  }
  }
 
- 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<int, string> $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);
  }
 
  /**

tests/Integration/PruneTaskResultsTest.php

@@ -0,0 +1,130 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ModelsLab\ModelQ\Tests\Integration;
+
+use ModelsLab\ModelQ\ModelQ;
+use PHPUnit\Framework\TestCase;
+use Redis;
+use ReflectionMethod;
+
+/**
+ * Counts value reads. phpredis `multi(Redis::PIPELINE)` returns the same object,
+ * so this catches pipelined GETs as well as direct ones.
+ */
+class CountingRedis extends Redis
+{
+ public int $getCalls = 0;
+
+ public function get($key): mixed
+ {
+ $this->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);
+ }
+}