perf(command-bus)!: store pending commands in a single Redis hash

[osbre]

Follow-up on #2247 (redis implementation). RedisCommandRepository kept one key per command, so every read ran SCAN ... MATCH command:pending:* - which walks every key in Redis. command:monitor does that twice a second, and kv-store shares one Redis with the cache, so most of the work was stepping over unrelated cache keys.

Pending and failed commands now live in one hash each, read with HSCAN. The public CommandRepository interface is unchanged.

Operation	Before	After
store()	SET command:pending:{uuid}	HSET
getPendingCommands()	SCAN + one GET per command	HSCAN loop (COUNT 500)
findPendingCommand()	GET command:pending:{uuid}	HGET
markAsDone()	UNLINK command:pending:{uuid}	HDEL
markAsFailed()	EXISTS + RENAME	Lua script (HGET + HSET/HDEL)

Two fixes came along with it:

• markAsFailed() sent EXISTS then RENAME as two calls, and threw if a markAsDone() landed in between. It is now one script.
• unserialize() warns and returns false on a corrupted payload rather than throwing, so the old catch (Throwable) never caught anything. Handled in unserializeCommand(), with failOnWarning="true" added to phpunit.xml.dist.

Benchmarks

Pending	Unrelated keys	Before	After	Speedup
100	0	55.09 ms	0.71 ms	78×
1,000	0	542.08 ms	3.38 ms	160×
10,000	0	5,344.95 ms	35.24 ms	152×
1,000	100,000	1,083.15 ms	3.11 ms	348×

Rows 2 and 4 are the point: same backlog, but 100,000 unrelated cache keys make the old version twice as slow.

Redis 7 in Docker on macOS, median of 15 runs, ~0.5 ms round trip.

redis-command-repository-bench.php

Requires ext-redis and a throwaway Redis server on 127.0.0.1:6399 (calls FLUSHALL):

docker run --rm -p 6399:6379 redis:7-alpine
php redis-command-repository-bench.php

<?php

declare(strict_types=1);

final class DummyCommand
{
    public function __construct(
        public string $id,
        public array $payload,
    ) {}
}

const HASH_KEY = 'command:pending';

function connect(): Redis
{
    $r = new Redis();
    $r->connect('127.0.0.1', 6399, 3);

    return $r;
}

function seed(Redis $r, int $pending, int $noise): void
{
    $r->flushAll();

    $pipe = $r->multi(Redis::PIPELINE);

    for ($i = 0; $i < $pending; $i++) {
        $uuid = sprintf('%08x-0000-4000-8000-%012x', $i, $i);
        $value = serialize(new DummyCommand($uuid, ['attempt' => 1, 'data' => str_repeat('x', 64)]));

        $pipe->set('command:pending:' . $uuid, $value);
        $pipe->hSet(HASH_KEY, $uuid, $value);
    }

    for ($i = 0; $i < $noise; $i++) {
        $pipe->set('cache:some:other:key:' . $i, 'noise');
    }

    $pipe->exec();
}

// Old implementation: SCAN over the keyspace + one GET per match.
function readCurrent(Redis $r): int
{
    $commands = [];
    $cursor = null;

    while (($keys = $r->scan($cursor, 'command:pending:*', 100)) !== false) {
        foreach ($keys as $key) {
            $value = $r->get($key);

            if (! is_string($value)) {
                continue;
            }

            $command = @unserialize($value);

            if (! is_object($command)) {
                continue;
            }

            $commands[substr($key, strlen('command:pending:'))] = $command;
        }

        if ((int) $cursor === 0) {
            break;
        }
    }

    return count($commands);
}

// Shipped implementation: incremental HSCAN over a single hash.
function readHscan(Redis $r): int
{
    $commands = [];
    $cursor = null;

    while (($fields = $r->hScan(HASH_KEY, $cursor, null, 500)) !== false) {
        foreach ($fields as $uuid => $value) {
            $command = @unserialize($value);

            if (! is_object($command)) {
                continue;
            }

            $commands[$uuid] = $command;
        }

        if ((int) $cursor === 0) {
            break;
        }
    }

    return count($commands);
}

// Rejected variant, kept to show what the incremental read costs.
function readHgetall(Redis $r): int
{
    $commands = [];

    foreach ($r->hGetAll(HASH_KEY) as $uuid => $value) {
        $command = @unserialize($value);

        if (! is_object($command)) {
            continue;
        }

        $commands[$uuid] = $command;
    }

    return count($commands);
}

function bench(callable $fn, Redis $r, int $runs): float
{
    $fn($r); // warm up

    $times = [];

    for ($i = 0; $i < $runs; $i++) {
        $start = hrtime(true);
        $fn($r);
        $times[] = (hrtime(true) - $start) / 1e6;
    }

    sort($times);

    return $times[intdiv(count($times), 2)]; // median, in ms
}

$redis = connect();
$runs = 15;

$scenarios = [[100, 0], [1_000, 0], [10_000, 0], [1_000, 100_000]];

printf("%-10s %-10s %12s %12s %12s %10s %8s\n", 'pending', 'noise', 'scan (ms)', 'hgetall (ms)', 'hscan (ms)', 'speedup', 'found');

foreach ($scenarios as [$pending, $noise]) {
    seed($redis, $pending, $noise);

    $found = readHscan($redis);
    $current = bench('readCurrent', $redis, $runs);
    $hgetall = bench('readHgetall', $redis, $runs);
    $hscan = bench('readHscan', $redis, $runs);

    printf(
        "%-10s %-10s %12.2f %12.2f %12.2f %9.1fx %8d\n",
        number_format($pending),
        number_format($noise),
        $current,
        $hgetall,
        $hscan,
        $current / $hscan,
        $found,
    );
}

$redis->flushAll();

Migration

Needed only if the queue isn't empty when you deploy:

# `command:pending:{uuid}` string keys become fields of the `command:pending` hash
migrate() {
  redis-cli --scan --pattern "$1:*" | while read -r key; do
    redis-cli EVAL '
      local value = redis.call("GET", KEYS[1])
      if not value then return 0 end
      redis.call("HSET", KEYS[2], ARGV[1], value)
      redis.call("UNLINK", KEYS[1])
      return 1
    ' 2 "$key" "$1" "${key#"$1":}"
  done
}

migrate command:pending
migrate command:failed

Scanning is client-side so each move stays atomic without a script that blocks the server, and
also because Redis before 7 refuses to write after a SCAN.

Notes

• Went with HSCAN instead HGETALL. HGETALL is ~2× faster, but it reads the whole hash in one command, and Redis is single threaded - a big backlog makes every other client wait. HSCAN reads in batches of 500 that others can be served between.
• No per-command TTLs. Hash fields can't expire individually before Redis 7.4. The old keys had no TTL either.
• Still no way to claim a command, so a killed monitor leaves a command pending forever. Equally true before this PR; fixing it means a breaking interface change, so it needs its own RFC.

Tests

Integration tests alongside the three existing ones, which now share a pinned Redis database, hoisted fixtures, and suppressed event handling: the single-hash layout, the atomic move to the failed hash, marking an unknown uuid as failed, corrupted payloads on both read paths, and a 1,200-command backlog that forces HSCAN through more than one batch. Skipped when Redis isn't reachable.

[brendt]

The only thing that's inconvenient is the manual migration.

Can we make it so that this command is run once automatically when command:monitor start and when it hasn't migrated yet?

[osbre]

@brendt Thanks, added. Is the instanceof RedisCommandRepository check good enough or does it need a separate interface? Also I marked everything migration-related as @deprecated for removal in 4.x, if that matches your vision.