Build a Resilient PHP 8.3 CLI Worker for Production Workloads

Mihajlo
Aug 6, 2026
Tutorials

A queue worker is trivial until the first inconvenient crash. The happy path fetches a row and invokes a handler. Production adds competing processes, poison messages, uncertain commits, expired reservations, deployments, and termination signals arriving at precisely the wrong boundary.

This tutorial builds a native PHP 8.3 worker with PDO and MySQL 8. It uses FOR UPDATE SKIP LOCKED for concurrent claims, commits reservations before processing, recovers expired leases, applies bounded retries, dead-letters terminal failures, and shuts down through pcntl.

Its delivery guarantee is at least once. A handler can run again after a crash or lease expiration. Reliability therefore depends on protecting the business effect with a stable idempotency key where that effect is committed.

Prerequisites and architectural boundaries

You need PHP 8.3 CLI with pdo_mysql, mysqlnd, and pcntl; MySQL 8 using InnoDB; and a Linux host for the long-running process.

php --version
php -m | grep -E '^(PDO|pdo_mysql|mysqlnd|pcntl)$'
mysql --version

The worker observes three boundaries:

Claim one eligible row in a short transaction, assign a random lease token, and commit.
Run business logic without holding row locks.
In another short transaction, verify ownership, write the idempotent business effect, and mark the job complete.

A lease is a recoverable ownership claim, not a transaction held throughout execution. If a process disappears, a recovery pass returns the expired reservation to the queue or marks it dead when no attempts remain.

Project structure

report-worker/
├── database/
│   └── schema.sql
├── src/
│   └── config.php
└── bin/
    ├── enqueue.php
    └── worker.php

Create the queue and effect tables

Apply database/schema.sql with an administrative MySQL identity. This example keeps MySQL on the same host; adapt the account host only when deploying across a private network.

CREATE DATABASE report_queue
    CHARACTER SET utf8mb4
    COLLATE utf8mb4_0900_ai_ci;

USE report_queue;

CREATE TABLE jobs (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    queue_name VARCHAR(64) NOT NULL,
    job_type VARCHAR(100) NOT NULL,
    payload JSON NOT NULL,
    idempotency_key CHAR(64) NOT NULL,
    status ENUM('ready', 'reserved', 'done', 'dead')
        NOT NULL DEFAULT 'ready',
    attempts SMALLINT UNSIGNED NOT NULL DEFAULT 0,
    max_attempts SMALLINT UNSIGNED NOT NULL DEFAULT 5,
    available_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    lease_until DATETIME(6) NULL,
    lease_token BINARY(16) NULL,
    last_error VARCHAR(4000) NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
        ON UPDATE CURRENT_TIMESTAMP(6),
    PRIMARY KEY (id),
    UNIQUE KEY uq_job_idempotency (queue_name, idempotency_key),
    KEY ix_claim (queue_name, status, available_at, id),
    KEY ix_recovery (queue_name, status, lease_until, id)
) ENGINE=InnoDB;

CREATE TABLE generated_reports (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    idempotency_key CHAR(64) NOT NULL,
    tenant_id VARCHAR(100) NOT NULL,
    report_month CHAR(7) NOT NULL,
    result_json JSON NOT NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    PRIMARY KEY (id),
    UNIQUE KEY uq_report_idempotency (idempotency_key)
) ENGINE=InnoDB;

CREATE USER 'queue_worker'@'127.0.0.1'
    IDENTIFIED BY 'replace-with-a-generated-secret';

GRANT SELECT, INSERT, UPDATE
    ON report_queue.*
    TO 'queue_worker'@'127.0.0.1';

The first unique key suppresses duplicate enqueue requests. The second protects the actual database effect if processing repeats. The runtime identity has no schema-management or deletion privileges.

Bound connections, network reads, and lock waits

mysqlnd.net_read_timeout is a system-level PHP setting, so configure it before PHP starts. First locate the CLI configuration with php --ini. This Debian/Ubuntu example adds a dedicated CLI override; on another distribution, add the same directive to the loaded CLI php.ini or its scanned configuration directory.

php --ini
sudo install -d -m 0755 /etc/php/8.3/cli/conf.d
printf '%s\n' 'mysqlnd.net_read_timeout=10' \
  | sudo tee /etc/php/8.3/cli/conf.d/99-report-worker.ini >/dev/null
php -r 'echo ini_get("mysqlnd.net_read_timeout"), PHP_EOL;'

The last command must print 10. Create src/config.php. The timeouts serve different purposes and are deliberately below the 60-second lease.

<?php
declare(strict_types=1);

function database(): PDO
{
    static $pdo = null;

    if ($pdo instanceof PDO) {
        return $pdo;
    }

    if (!extension_loaded('mysqlnd')) {
        throw new RuntimeException('mysqlnd is required');
    }

    if ((int) ini_get('mysqlnd.net_read_timeout') !== 10) {
        throw new RuntimeException(
            'Set mysqlnd.net_read_timeout=10 in the CLI php.ini'
        );
    }

    $password = getenv('DB_PASS');
    if ($password === false || $password === '') {
        throw new RuntimeException('DB_PASS is required');
    }

    $host = getenv('DB_HOST') ?: '127.0.0.1';
    $port = getenv('DB_PORT') ?: '3306';
    $name = getenv('DB_NAME') ?: 'report_queue';
    $user = getenv('DB_USER') ?: 'queue_worker';

    $dsn = "mysql:host={$host};port={$port};dbname={$name};charset=utf8mb4";

    $pdo = new PDO($dsn, $user, $password, [
        PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
        PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
        PDO::ATTR_EMULATE_PREPARES => false,
        PDO::ATTR_TIMEOUT => 5,
        PDO::MYSQL_ATTR_MULTI_STATEMENTS => false,
    ]);

    $pdo->exec("SET SESSION time_zone = '+00:00'");
    $pdo->exec("SET SESSION innodb_lock_wait_timeout = 3");
    $pdo->exec("SET SESSION lock_wait_timeout = 5");

    return $pdo;
}

PDO::ATTR_TIMEOUT bounds connection establishment for PDO MySQL; it is not a query deadline. mysqlnd.net_read_timeout limits an individual network read to ten seconds. It still does not cancel server-side work, so the worker exits after an unexpected database exception and lets its supervisor create a clean connection.

innodb_lock_wait_timeout bounds InnoDB record-lock waits, while lock_wait_timeout covers metadata locks. These settings do not turn the locking SELECT ... FOR UPDATE claim into a hard query deadline. The client-side mysqlnd.net_read_timeout prevents an individual network read from waiting forever; after an unexpected database timeout, the worker exits and its supervisor creates a clean connection. Keep every claim query indexed and every reservation transaction short.

Enqueue jobs idempotently

Create bin/enqueue.php. Its key comes from the business identity of a monthly report, so repeating the command returns the existing job ID.

<?php
declare(strict_types=1);

require dirname(__DIR__) . '/src/config.php';

if ($argc !== 3) {
    fwrite(STDERR, "Usage: php bin/enqueue.php TENANT_ID YYYY-MM\n");
    exit(64);
}

$tenant = $argv[1];
$month = $argv[2];

if (!preg_match('/^[A-Za-z0-9_-]{1,100}$/D', $tenant)) {
    throw new InvalidArgumentException('Invalid tenant ID');
}

if (!preg_match('/^\d{4}-(0[1-9]|1[0-2])$/D', $month)) {
    throw new InvalidArgumentException('Month must use YYYY-MM');
}

$key = hash('sha256', "monthly-report:{$tenant}:{$month}");
$payload = json_encode([
    'tenant_id' => $tenant,
    'report_month' => $month,
], JSON_THROW_ON_ERROR);

$sql = <<<'SQL'
INSERT INTO jobs (queue_name, job_type, payload, idempotency_key)
VALUES ('reports', 'report.generate', ?, ?)
ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)
SQL;

$pdo = database();
$statement = $pdo->prepare($sql);
$statement->execute([$payload, $key]);

fwrite(STDOUT, $pdo->lastInsertId() . PHP_EOL);

Implement claims, retries, and graceful release

Create bin/worker.php. Recovery handles at most 100 rows per pass. The indexed claim uses SKIP LOCKED and a short transaction. PDO MySQL has no universal per-statement deadline, so the worker combines server lock-wait limits with a bounded client network-read timeout and supervisor restart.

<?php
declare(strict_types=1);

require dirname(__DIR__) . '/src/config.php';

const QUEUE = 'reports';
const LEASE_SECONDS = 60;

function recoverExpiredLeases(PDO $pdo): int
{
    $sql = <<<'SQL'
UPDATE jobs
SET status = CASE
        WHEN attempts >= max_attempts THEN 'dead'
        ELSE 'ready'
    END,
    available_at = CASE
        WHEN attempts >= max_attempts THEN available_at
        ELSE UTC_TIMESTAMP(6)
    END,
    lease_until = NULL,
    lease_token = NULL,
    last_error = CASE
        WHEN attempts >= max_attempts
            THEN 'Lease expired after final attempt'
        ELSE 'Lease expired; returned to queue'
    END
WHERE queue_name = ?
  AND status = 'reserved'
  AND lease_until <= UTC_TIMESTAMP(6)
ORDER BY lease_until, id
LIMIT 100
SQL;

    $statement = $pdo->prepare($sql);
    $statement->execute([QUEUE]);
    return $statement->rowCount();
}

function claim(PDO $pdo): ?array
{
    $pdo->beginTransaction();

    try {
        $select = $pdo->prepare(
            "SELECT *
             FROM jobs
             WHERE queue_name = ?
               AND status = 'ready'
               AND attempts < max_attempts
               AND available_at <= UTC_TIMESTAMP(6)
             ORDER BY available_at, id
             LIMIT 1
             FOR UPDATE SKIP LOCKED"
        );
        $select->execute([QUEUE]);
        $job = $select->fetch();

        if ($job === false) {
            $pdo->commit();
            return null;
        }

        $token = random_bytes(16);
        $update = $pdo->prepare(
            "UPDATE jobs
             SET status = 'reserved',
                 attempts = attempts + 1,
                 lease_token = ?,
                 lease_until = TIMESTAMPADD(
                     SECOND, ?, UTC_TIMESTAMP(6)
                 )
             WHERE id = ?"
        );
        $update->execute([$token, LEASE_SECONDS, $job['id']]);
        $pdo->commit();

        $job['attempts'] = (int) $job['attempts'] + 1;
        $job['lease_token'] = $token;

        $testPause = (int) (getenv('WORKER_TEST_POST_CLAIM_PAUSE_MS') ?: 0);
        if ($testPause > 0 && $testPause <= 10_000) {
            usleep($testPause * 1000);
        }

        return $job;
    } catch (Throwable $error) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        throw $error;
    }
}

function releaseClaim(PDO $pdo, array $job): bool
{
    $statement = $pdo->prepare(
        "UPDATE jobs
         SET status = 'ready',
             attempts = IF(attempts > 0, attempts - 1, 0),
             available_at = UTC_TIMESTAMP(6),
             lease_until = NULL,
             lease_token = NULL,
             last_error = 'Released during shutdown before processing'
         WHERE id = ?
           AND status = 'reserved'
           AND lease_token = ?"
    );
    $statement->execute([$job['id'], $job['lease_token']]);

    return $statement->rowCount() === 1;
}

function processJob(array $job): string
{
    if ($job['job_type'] !== 'report.generate') {
        throw new DomainException('Unsupported job type');
    }

    $payload = json_decode(
        $job['payload'],
        true,
        flags: JSON_THROW_ON_ERROR
    );

    $tenant = $payload['tenant_id'] ?? null;
    $month = $payload['report_month'] ?? null;

    if (
        !is_string($tenant)
        || !preg_match('/^[A-Za-z0-9_-]{1,100}$/D', $tenant)
        || !is_string($month)
        || !preg_match('/^\d{4}-(0[1-9]|1[0-2])$/D', $month)
    ) {
        throw new UnexpectedValueException('Malformed job payload');
    }

    if (getenv('WORKER_ENABLE_TEST_FIXTURES') === '1') {
        if (!empty($payload['test_pause'])) {
            $until = hrtime(true) + 30_000_000_000;
            while (hrtime(true) < $until) {
                sleep(1);
            }
        }

        $transientUntil = (int) (
            $payload['test_transient_until_attempt'] ?? 0
        );
        if (
            $transientUntil > 0
            && (int) $job['attempts'] <= $transientUntil
        ) {
            throw new RuntimeException(
                'Synthetic transient test failure'
            );
        }
    }

    return json_encode([
        'tenant_id' => $tenant,
        'report_month' => $month,
        'state' => 'generated',
    ], JSON_THROW_ON_ERROR);
}

function ownsLease(array|false $row, string $token): bool
{
    return $row !== false
        && (int) $row['lease_valid'] === 1
        && hash_equals($row['lease_token'], $token);
}

function complete(PDO $pdo, array $job, string $result): bool
{
    $pdo->beginTransaction();

    try {
        $lock = $pdo->prepare(
            "SELECT lease_token,
                    lease_until > UTC_TIMESTAMP(6) AS lease_valid
             FROM jobs
             WHERE id = ? AND status = 'reserved'
             FOR UPDATE"
        );
        $lock->execute([$job['id']]);
        $current = $lock->fetch();

        if (!ownsLease($current, $job['lease_token'])) {
            $pdo->rollBack();
            return false;
        }

        $payload = json_decode(
            $job['payload'],
            true,
            flags: JSON_THROW_ON_ERROR
        );

        $effect = $pdo->prepare(
            "INSERT INTO generated_reports (
                idempotency_key, tenant_id, report_month, result_json
             ) VALUES (?, ?, ?, ?)
             ON DUPLICATE KEY UPDATE idempotency_key = ?"
        );
        $effect->execute([
            $job['idempotency_key'],
            $payload['tenant_id'],
            $payload['report_month'],
            $result,
            $job['idempotency_key'],
        ]);

        $finish = $pdo->prepare(
            "UPDATE jobs
             SET status = 'done',
                 lease_until = NULL,
                 lease_token = NULL,
                 last_error = NULL
             WHERE id = ?"
        );
        $finish->execute([$job['id']]);

        $pdo->commit();
        return true;
    } catch (Throwable $error) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        throw $error;
    }
}

function errorSummary(Throwable $error): string
{
    $controlled = $error instanceof DomainException
        || $error instanceof UnexpectedValueException
        || $error instanceof JsonException;

    return $controlled
        ? substr($error::class . ': ' . $error->getMessage(), 0, 4000)
        : $error::class;
}

function failJob(PDO $pdo, array $job, Throwable $error): bool
{
    $pdo->beginTransaction();

    try {
        $lock = $pdo->prepare(
            "SELECT attempts, max_attempts, lease_token,
                    lease_until > UTC_TIMESTAMP(6) AS lease_valid
             FROM jobs
             WHERE id = ? AND status = 'reserved'
             FOR UPDATE"
        );
        $lock->execute([$job['id']]);
        $current = $lock->fetch();

        if (!ownsLease($current, $job['lease_token'])) {
            $pdo->rollBack();
            return false;
        }

        $permanent = $error instanceof DomainException
            || $error instanceof UnexpectedValueException
            || $error instanceof JsonException;
        $exhausted = $permanent
            || (int) $current['attempts']
                >= (int) $current['max_attempts'];
        $summary = errorSummary($error);

        if ($exhausted) {
            $statement = $pdo->prepare(
                "UPDATE jobs
                 SET status = 'dead',
                     lease_until = NULL,
                     lease_token = NULL,
                     last_error = ?
                 WHERE id = ?"
            );
            $statement->execute([$summary, $job['id']]);
        } else {
            $attempt = (int) $current['attempts'];
            $delay = min(
                300,
                (2 ** min($attempt, 8)) + random_int(0, 3)
            );

            $statement = $pdo->prepare(
                "UPDATE jobs
                 SET status = 'ready',
                     available_at = TIMESTAMPADD(
                         SECOND, ?, UTC_TIMESTAMP(6)
                     ),
                     lease_until = NULL,
                     lease_token = NULL,
                     last_error = ?
                 WHERE id = ?"
            );
            $statement->execute([$delay, $summary, $job['id']]);
        }

        $pdo->commit();
        return true;
    } catch (Throwable $failure) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        throw $failure;
    }
}

function logEvent(string $event, array $context = []): void
{
    fwrite(STDERR, json_encode([
        'event' => $event,
        'time' => gmdate(DATE_ATOM),
    ] + $context, JSON_THROW_ON_ERROR) . PHP_EOL);
}

$stopping = false;
pcntl_async_signals(true);

$stop = static function (int $signal) use (&$stopping): void {
    $stopping = true;
};

pcntl_signal(SIGTERM, $stop);
pcntl_signal(SIGINT, $stop);

try {
    $pdo = database();
    $nextRecoveryAt = 0.0;
    logEvent('worker_started', ['pid' => getmypid()]);

    while (!$stopping) {
        if (microtime(true) >= $nextRecoveryAt) {
            $recovered = recoverExpiredLeases($pdo);
            $nextRecoveryAt = microtime(true) + 5.0;

            if ($recovered > 0) {
                logEvent('leases_recovered', ['count' => $recovered]);
            }
        }

        if ($stopping) {
            break;
        }

        $job = claim($pdo);

        /*
         * This check must be the first action after claim().
         * No handler starts if shutdown arrived during reservation.
         */
        if ($stopping) {
            if ($job !== null) {
                try {
                    $released = releaseClaim($pdo, $job);
                    logEvent(
                        $released
                            ? 'claim_released_on_shutdown'
                            : 'claim_release_lost',
                        ['job_id' => (int) $job['id']]
                    );
                } catch (Throwable $releaseError) {
                    logEvent('claim_release_failed', [
                        'job_id' => (int) $job['id'],
                        'error' => errorSummary($releaseError),
                    ]);
                    throw $releaseError;
                }
            }
            break;
        }

        if ($job === null) {
            for ($tick = 0; $tick < 10 && !$stopping; $tick++) {
                usleep(250_000);
            }
            continue;
        }

        $startedAt = hrtime(true);
        logEvent('job_claimed', [
            'job_id' => (int) $job['id'],
            'attempt' => $job['attempts'],
        ]);

        try {
            $result = processJob($job);
        } catch (Throwable $handlerError) {
            $owned = failJob($pdo, $job, $handlerError);
            logEvent(
                $owned ? 'job_failed' : 'job_lease_lost',
                [
                    'job_id' => (int) $job['id'],
                    'error' => errorSummary($handlerError),
                    'elapsed_ms' => (int) (
                        (hrtime(true) - $startedAt) / 1_000_000
                    ),
                ]
            );
            continue;
        }

        /*
         * Do not classify finalization failures as handler failures.
         * A database exception here escapes to the outer fatal handler,
         * so systemd restarts the worker with a fresh connection.
         */
        $owned = complete($pdo, $job, $result);
        logEvent(
            $owned ? 'job_completed' : 'job_lease_lost',
            [
                'job_id' => (int) $job['id'],
                'elapsed_ms' => (int) (
                    (hrtime(true) - $startedAt) / 1_000_000
                ),
            ]
        );
    }

    logEvent('worker_stopped', ['pid' => getmypid()]);
} catch (Throwable $fatal) {
    logEvent('worker_fatal', ['error' => errorSummary($fatal)]);
    exit(1);
}

SKIP LOCKED prevents workers from waiting behind a peer’s selected row, but the transaction remains essential: the row stays locked until its lease token and deadline are stored.

The immediate stop check closes a subtle shutdown race. If SIGTERM arrives while claim() is blocked or finishing, the worker conditionally updates only the row matching both job ID and random lease token. It returns that row to ready, restores the unused attempt, logs whether release succeeded, and never calls processJob().

The business effect and transition to done share one transaction. For an external API, send the stable idempotency key when supported. MySQL cannot atomically coordinate an unrelated remote side effect; otherwise use downstream reconciliation or an outbox-based integration.

Only exceptions raised by processJob() enter failJob() and consume an attempt. A failure in complete() or in the failure-state transaction is infrastructure failure: it reaches the outer fatal handler, exits the process, and lets systemd restart with a fresh connection. The committed lease then prevents immediate duplicate ownership and remains recoverable after expiry.

Exercise concurrency and failure paths

Export credentials in each terminal, enqueue the same report twice, and start two workers:

export DB_HOST=127.0.0.1
export DB_PORT=3306
export DB_NAME=report_queue
export DB_USER=queue_worker
export DB_PASS='replace-with-a-generated-secret'

php bin/enqueue.php acme 2042-01
php bin/enqueue.php acme 2042-01
php bin/worker.php

The enqueue commands should print the same ID. Add several tenants and confirm that separate workers claim different rows. Inspect durable state with:

SELECT id, status, attempts, available_at, lease_until, last_error
FROM jobs
ORDER BY id DESC
LIMIT 20;

SELECT idempotency_key, tenant_id, report_month, result_json
FROM generated_reports
ORDER BY id DESC
LIMIT 20;

Test the post-claim shutdown path on a development instance. The disabled-by-default pause makes the timing deterministic:

php bin/enqueue.php shutdown-test 2042-02

WORKER_TEST_POST_CLAIM_PAUSE_MS=10000 php bin/worker.php &
worker_pid=$!

case "$worker_pid" in
    ''|*[!0-9]*) exit 1 ;;
esac

sleep 1
kill -TERM "$worker_pid"
wait "$worker_pid"

The log should contain claim_released_on_shutdown, followed by worker_stopped. The job should be ready, with no generated report and no consumed attempt.

Run the remaining failure fixtures only against an isolated development database. Test-only handler behavior is disabled unless WORKER_ENABLE_TEST_FIXTURES=1.

Crash and expired-lease recovery

pause_id=$(
  MYSQL_PWD="$DB_PASS" mysql --protocol=TCP     -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME" -Nse "
      INSERT INTO jobs (
        queue_name, job_type, payload, idempotency_key, max_attempts, available_at
      ) VALUES (
        'reports',
        'report.generate',
        JSON_OBJECT(
          'tenant_id', 'fixture-pause',
          'report_month', '2042-03',
          'test_pause', TRUE
        ),
        SHA2(CONCAT('fixture:pause:', UUID()), 256),
        3,
        UTC_TIMESTAMP(6)
      );
      SELECT LAST_INSERT_ID();
    "
)

WORKER_ENABLE_TEST_FIXTURES=1 php bin/worker.php   2> /tmp/report-worker-pause.log &
worker_pid=$!

for attempt in $(seq 1 50); do
  grep -q '"event":"job_claimed"' /tmp/report-worker-pause.log     && break
  sleep 0.1
done
grep -q '"event":"job_claimed"' /tmp/report-worker-pause.log
kill -KILL "$worker_pid"
wait "$worker_pid" || true

sleep 61
timeout --signal=TERM 8s php bin/worker.php   2>> /tmp/report-worker-pause.log || test "$?" -eq 124
grep -q '"event":"leases_recovered"' /tmp/report-worker-pause.log

MYSQL_PWD="$DB_PASS" mysql --protocol=TCP   -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME"   -e "SELECT id, status, attempts FROM jobs WHERE id = $pause_id"

The final query must show done. The log must contain both job_claimed and leases_recovered, proving that a second worker reclaimed the expired 60-second lease.

Permanent and transient failures

unsupported_id=$(
  MYSQL_PWD="$DB_PASS" mysql --protocol=TCP     -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME" -Nse "
      INSERT INTO jobs (
        queue_name, job_type, payload, idempotency_key, max_attempts, available_at
      ) VALUES (
        'reports',
        'unsupported.fixture',
        JSON_OBJECT(
          'tenant_id', 'fixture-unsupported',
          'report_month', '2042-04'
        ),
        SHA2(CONCAT('fixture:unsupported:', UUID()), 256),
        1,
        UTC_TIMESTAMP(6)
      );
      SELECT LAST_INSERT_ID();
    "
)
timeout --signal=TERM 5s php bin/worker.php   2> /tmp/report-worker-permanent.log || test "$?" -eq 124
MYSQL_PWD="$DB_PASS" mysql --protocol=TCP   -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME"   -e "SELECT id, status, attempts FROM jobs WHERE id = $unsupported_id"

transient_id=$(
  MYSQL_PWD="$DB_PASS" mysql --protocol=TCP     -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME" -Nse "
      INSERT INTO jobs (
        queue_name, job_type, payload, idempotency_key, max_attempts, available_at
      ) VALUES (
        'reports',
        'report.generate',
        JSON_OBJECT(
          'tenant_id', 'fixture-transient',
          'report_month', '2042-05',
          'test_transient_until_attempt', 2
        ),
        SHA2(CONCAT('fixture:transient:', UUID()), 256),
        4,
        UTC_TIMESTAMP(6)
      );
      SELECT LAST_INSERT_ID();
    "
)
timeout --signal=TERM 30s env WORKER_ENABLE_TEST_FIXTURES=1   php bin/worker.php 2> /tmp/report-worker-transient.log   || test "$?" -eq 124
MYSQL_PWD="$DB_PASS" mysql --protocol=TCP   -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME"   -e "SELECT id, status, attempts FROM jobs WHERE id = $transient_id"

grep -q '"event":"job_failed"' /tmp/report-worker-transient.log
grep -q '"event":"job_completed"' /tmp/report-worker-transient.log

The unsupported fixture must be dead after one attempt. The transient fixture must reach done on attempt three, after two synthetic failures and bounded jittered backoff.

Deploy with systemd

Install reviewed files under /opt/report-worker, owned by root and readable by a dedicated report-worker account. Store environment assignments in /etc/report-worker.env, owned by root with mode 0600. These are host-administrator operations, not container commands.

Create /etc/systemd/system/report-worker.service with sudoedit and use this exact unit:

[Unit]
Description=PHP report queue worker
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=report-worker
Group=report-worker
WorkingDirectory=/opt/report-worker
EnvironmentFile=/etc/report-worker.env
ExecStart=/usr/bin/php /opt/report-worker/bin/worker.php
Restart=on-failure
RestartSec=3
KillSignal=SIGTERM
TimeoutStopSec=75
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
UMask=0077

[Install]
WantedBy=multi-user.target

Validate and enable the exact unit file:

sudo systemd-analyze verify /etc/systemd/system/report-worker.service
sudo systemctl daemon-reload
sudo systemctl enable --now report-worker.service
sudo systemctl status report-worker.service
sudo journalctl -u report-worker.service --since today

Use a systemd template for additional instances. Scale gradually: workers consume database connections and downstream capacity even when row-lock contention is low. The 75-second stop budget exceeds the lease, while connection, network-read, and lock timeouts remain below both.

Observability, security, and performance

The worker emits structured JSON containing job IDs, attempts, elapsed time, recovery counts, lease losses, shutdown releases, and lifecycle events. Alert on dead-job growth, oldest-ready age, repeated recovery, retry rate, processing latency, release failures, and fatal database errors.

Keep credentials and sensitive payload fields out of logs and last_error. Validate payloads at consumption time even when producers are trusted. For remote MySQL, require certificate verification, restrict port 3306 to authorized application addresses, and create an account scoped to that network rather than using a public wildcard host. The worker itself needs no inbound firewall rule.

Measure claim-query plans as the table grows. Retain completed jobs only as long as operational or audit requirements demand, and archive them through a separate reviewed process. Slow network calls, report generation, and filesystem work belong outside reservation transactions and need their own deadlines below the lease.

Common production failures

Workers appear serialized: processing remains inside the claim transaction, or the claim index is missing.
Jobs overlap after expiration: processing exceeds the lease. Bound operation time, adjust the lease deliberately, or implement token-fenced renewal.
Queries still hang: only PDO::ATTR_TIMEOUT was configured. Connection, network-read, and lock-wait bounds are separate controls; PDO MySQL has no universal query timeout.
Deployments consume attempts: a post-claim stop check is missing, or release does not restore an unused attempt.
A stale worker commits: finalization fails to verify both token and unexpired deadline under a row lock.
Remote duplicates appear: idempotency exists only in MySQL rather than at the external effect boundary.

Final verification checklist

Duplicate enqueueing returns one durable job ID.
Concurrent workers claim different eligible rows without long waits.
Reservation commits before business processing begins.
Connection, network-read, and lock-wait bounds are below the lease, and claim transactions remain short and indexed.
A signal observed during claim() releases only the matching token and starts no handler.
Expired leases return to ready while attempts remain.
Exhausted and permanent failures enter dead.
An expired or stale token cannot finalize or reschedule a job.
The business-effect table rejects duplicate idempotency keys.
Retry delays and maximum attempts are finite.
Logs expose latency, retries, recovery, lease loss, release, and shutdown.
The runtime identity has no unnecessary database, filesystem, or network privileges.

A durable queue row is only the beginning of a resilient worker. The important engineering lives at the boundaries: reserve briefly, execute without locks, fence every final write, release work when shutdown wins the race, and make the real effect idempotent. Once those rules are explicit, crashes and repeated delivery stop being surprises and become ordinary states the system was built to absorb.