README

Пример PHP-приложения для фоновой обработки сообщений через RoadRunner Jobs, Kafka, RabbitMQ и встроенный Memory-драйвер.

Главная идея проекта: PHP-код не подключается к Kafka или RabbitMQ напрямую. Подключениями, получением сообщений, ACK/NACK и worker-процессами управляет RoadRunner. Приложение получает готовую задачу через Goridge, преобразует её в JobEnvelope и выбирает handler по broker + destination + messageType, а для нескольких независимых consumers — дополнительно по consumer.

Логи HTTP-side кода и jobs workers имеют единый плоский JSONL-формат, совместимый с OpenTelemetry Collector. В репозитории есть пример Collector для будущей отправки логов в Grafana Loki.

В пакет также входит framework-neutral RoadRunner Dashboard: история jobs, реальные worker metrics через Informer RPC и кооперативная отмена задач. SQLite работает по умолчанию, Redis доступен для нескольких instances. Полное руководство: docs/dashboard.md.

После установки package проект можно запускать двумя способами: нативно через rr serve либо в изолированном контейнере одной командой vendor/bin/rr-worker up. Docker-режим опционален: он запускает только RoadRunner и PHP workers, а Kafka/RabbitMQ остаются частью внешней инфраструктуры.

Почему выбран такой подход

Подход строится вокруг разделения ответственности: RoadRunner отвечает за инфраструктуру доставки и жизненный цикл процессов, а PHP-приложение — за контракт сообщения, маршрутизацию и прикладную логику.

Основные преимущества:

PHP-коду не нужны отдельные Kafka- и AMQP-клиенты, управление соединениями и собственный consumer loop для каждого брокера;
Kafka, RabbitMQ и Memory используют одинаковые JobEnvelope, JobHandler и lifecycle обработки;
прикладной handler явно объявляет physical Kafka topic или RabbitMQ queue, но не зависит от credentials и broker client;
#[Subscribe] делает связь consumer + broker + destination + messageType → handler явной и доступной для проверки при старте worker;
долгоживущие RoadRunner workers уменьшают расходы на повторный bootstrap PHP для каждого сообщения;
worker pool, получение сообщений и ACK/NACK управляются централизованно;
один trace_id связывает отправку сообщения и его фоновую обработку;
единый JSONL-формат упрощает поиск и parsing логов HTTP-side кода, producers и consumers;
приложение не зависит от Loki: Collector читает файл отдельно, поэтому временная недоступность observability backend не должна останавливать jobs;
локальный Docker Compose воспроизводит полный путь через Kafka и RabbitMQ без установки брокеров на машине разработчика.

Решение и практический выигрыш

Решение	Практический выигрыш
Работа с брокерами внутри RoadRunner	Меньше broker-specific кода и PHP-зависимостей в приложении
Единый `JobEnvelope`	Один версионируемый контракт для разных transport drivers
Routing по `consumer + broker + destination + messageType`	Physical source и logical contract проверяются независимо; один message вызывает ровно один handler
Декларативный `#[Subscribe]`	Связь destination/type с handler видна рядом с кодом, а дубли обнаруживаются при старте
Явный `HandlerRegistry`	Состав доступных handlers предсказуем и не зависит от runtime scanning или cache
Долгоживущий worker pool	Не требуется запускать новый PHP-процесс и заново собирать приложение для каждого job
ACK только после успешного handler	Сообщение подтверждается после завершения прикладной операции
NACK при исключении	Broker/RoadRunner получает явный сигнал о неуспешной обработке
Общий trace context	Producer, HTTP-side код и consumer можно искать по одному `trace_id`
Плоский JSONL	Записи легко читать, валидировать и передавать в Collector без разбора смешанного stdout
Collector между приложением и Loki	Формат экспорта и observability backend можно менять без изменения business handlers
Раздельные application и infrastructure logs	RoadRunner/Kafka/RabbitMQ diagnostics не смешиваются с прикладными событиями
Kafka и RabbitMQ в Compose	Integration-сценарий одинаково воспроизводится локально и в CI

Выигрыш для команды

новый handler добавляется небольшим классом, одним #[Subscribe] и регистрацией в worker;
разработчик тестирует прикладную логику без прямого подключения к брокеру;
DevOps меняет адреса, credentials, pipelines и параметры consumer group без изменения handler-классов;
единые lifecycle events позволяют строить общие dashboards и alerts для всех брокеров;
явные границы упрощают расследование: отдельно проверяются producer, RoadRunner pipeline, broker, routing и handler.

Компромиссы

У подхода есть цена, которую необходимо учитывать:

RoadRunner становится обязательной частью runtime и требует отдельной конфигурации, обновления и мониторинга;
новый handler необходимо зарегистрировать вручную — атрибут не выполняет автоматический поиск классов;
нужно различать RoadRunner pipeline, физический destination брокера и логический JobEnvelope.type;
долгоживущие PHP workers требуют аккуратно очищать request/job state и контролировать утечки памяти;
retry, backoff, dead-letter queue и идемпотентность не появляются автоматически — их нужно проектировать под конкретный broker и бизнес-процесс;
файловые JSONL-логи требуют volume, rotation и retention; для нескольких replicas нужно учитывать семантику file locking выбранного storage.

Этот подход особенно полезен, когда сервису нужны фоновые handlers, единая модель обработки для нескольких брокеров и централизованная observability. Для маленького процесса с одним простым consumer прямой broker client иногда может оказаться проще.

1. Руководство разработчика

Что это за проект

Проект показывает базовую архитектуру асинхронного PHP worker-приложения:

RoadRunner запускает и контролирует долгоживущие PHP workers;
RoadRunner самостоятельно работает с Kafka, RabbitMQ или Memory queue;
producer отправляет задачу в RoadRunner через Jobs RPC;
worker принимает задачу через Goridge;
приложение определяет брокер по RoadRunner driver;
HandlerRegistry выбирает handler по атрибуту #[Subscribe];
результат подтверждается через ACK, ошибка — через NACK;
весь жизненный цикл записывается в плоские JSONL-логи с trace context.

Это не framework и не готовый универсальный message bus. Это небольшой каркас, на котором можно строить собственные consumers и producers.

Технологии

Компонент	Назначение
PHP 8.4	Runtime приложения и Docker reference environment
RoadRunner 2025.1.15	Process manager, RPC и Jobs plugin
`spiral/roadrunner-jobs`	PHP API для producer и consumer
Kafka 4.2.1	Тестовый Kafka broker
RabbitMQ 4.2.9	Тестовый AMQP broker
Monolog 3	Логирование приложения
Symfony Console и Process	Безопасный CLI для управления Docker runtime
OpenTelemetry Collector Contrib	Чтение JSONL и преобразование в OTLP LogRecord
PHPUnit 12	Unit-тесты

Версии Kafka и RabbitMQ выше относятся к compose.test.yaml. Внешние production-брокеры могут использовать другие совместимые версии.

Как устроен проект

flowchart LR
    P["PHP producer / HTTP-side код"]
    RPC["RoadRunner Jobs RPC"]
    PL["RoadRunner pipeline"]
    B["Kafka / RabbitMQ / Memory"]
    C["RoadRunner consumer"]
    W["jobs-worker.php"]
    E["JobEnvelope"]
    R["HandlerRegistry"]
    H["JobHandler"]
    A["ACK / NACK"]

    P --> RPC --> PL --> B --> C --> W --> E --> R --> H --> A

Основные компоненты:

Файл/класс	Ответственность
`bin/jobs-worker.php`	Собирает зависимости, запускает consumer loop, выполняет ACK/NACK
`JobEnvelope`	Контракт сообщения и JSON-сериализация
`MessageBroker`	Преобразует RoadRunner driver в `kafka`, `rabbitmq` или `memory`
`Subscribe`	Связывает consumer alias, broker, physical destination и optional logical message type с handler-классом
`HandlerRegistry`	Проверяет handlers и строит routing table
`JobRunner`	Находит handler и вызывает `handle()`
`JobProcessor`	Управляет trace context и lifecycle-логами задачи
`JobHandler`	Интерфейс прикладного handler
`LoggerFactory`	Создаёт одинаковый logger для HTTP-side и worker-кода
`TraceContext`	Создаёт и распространяет `trace_id`, `span_id`, `trace_flags`

Маршрутизация сообщений

В проекте есть три разных понятия, которые не следует смешивать.

1. RoadRunner pipeline

Pipeline — именованная конфигурация RoadRunner Jobs. Примеры:

local — Memory driver в .rr.yaml;
kafka-test — Kafka driver в .rr.test.yaml;
rabbitmq-test — AMQP driver в .rr.test.yaml.

Producer подключается именно к pipeline:

$queue = $jobs->connect('kafka-test', $options);

2. Физический destination брокера

Это реальный Kafka topic или RabbitMQ queue/exchange:

Kafka topic: rr-demo;
RabbitMQ queue/exchange/routing key: rr-demo.

Фактическую подписку на эти значения создаёт .rr.yaml; producer выбирает destination через свои options. Handler повторяет ожидаемый destination в #[Subscribe], чтобы сообщение из другой очереди не попало в него случайно.

3. Логический message type приложения

Logical message type находится в JobEnvelope.type, например demo.kafka или orders.created. Он задаётся в messageType:

#[Subscribe(
    destination: 'rr-demo',
    messageType: 'demo.kafka',
    broker: MessageBroker::Kafka,
)]
final readonly class DemoKafkaHandler implements JobHandler
{
    public function handle(JobEnvelope $job, JobExecutionContext $context): void
    {
        // Прикладная обработка.
    }
}

Итоговый точный ключ маршрутизации без consumer aliases:

MessageBroker + ReceivedTask.getQueue() + JobEnvelope.type

Если один физический topic независимо обрабатывают несколько consumer groups, vendor worker преобразует RoadRunner pipeline в стабильный consumer alias:

RoadRunner pipeline → consumer alias
ReceivedTask.getQueue() → physical destination
consumer alias + MessageBroker + destination + JobEnvelope.type → handler

Один и тот же logical message type разрешено использовать для разных брокеров. Например, orders.created может иметь отдельный Kafka handler и отдельный RabbitMQ handler. Одинаковая комбинация broker + destination + messageType также разрешена для разных consumer aliases. Дублирование полного ключа внутри одного consumer запрещено.

Правила `#[Subscribe]`

Атрибут применяется только к классу.
У каждого зарегистрированного handler должен быть ровно один #[Subscribe].
destination обязателен и не может быть пустой строкой.
messageType необязателен; если он задан, пустая строка запрещена.
consumer необязателен, но при наличии не может быть пустой строкой.
Одна комбинация consumer + broker + destination + messageType может принадлежать только одному handler.
Registry сначала ищет exact handler с совпавшим messageType, затем fallback handler того же consumer + broker + destination без messageType.
Если exact и fallback отсутствуют, задача получает routing error и NACK.
Fallback не пересекает consumer aliases, brokers или destinations.
Атрибут не выполняет автоматический поиск классов.
Новый handler необходимо вручную добавить в HandlerRegistry в bin/jobs-worker.php.

Registry создаётся при старте worker. После deployment RoadRunner перезапустит worker-процессы, поэтому отдельный runtime cache или механизм hot discovery не нужен.

Миграция с `Subscribe(topic: ...)`

Новый контракт является breaking change следующей major-версии. Старый параметр и property topic удалены без deprecated alias:

// Раньше: physical destination невозможно было отличить от logical type.
#[Subscribe(topic: 'demo.message.v1', broker: MessageBroker::Kafka)]

// Теперь: оба значения объявлены независимо.
#[Subscribe(
    destination: 'laravel.test.v1',
    messageType: 'demo.message.v1',
    broker: MessageBroker::Kafka,
)]

Если handler должен принимать любой JobEnvelope.type из одного physical destination, не передавайте messageType:

#[Subscribe(destination: 'laravel.test.v1', broker: MessageBroker::Kafka)]

Это fallback handler. Exact handler с совпавшим messageType всегда имеет приоритет. Producer API KafkaOptions(topic: ...) не меняется: там topic действительно означает physical Kafka topic.

Формат JobEnvelope

Каждая задача передаётся как JSON-объект следующего вида:

{
  "id": "2c8e058e1d5210949f07683fbaf18b6a",
  "type": "demo.kafka",
  "version": 1,
  "created_at": "2026-08-02T12:00:00+00:00",
  "payload": {
    "message": "Hello"
  },
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736"
}

Поле	Тип	Обязательно	Назначение
`id`	string	да	Уникальный ID задачи
`type`	string	да	Logical message type и ключ exact-выбора handler
`version`	integer	да	Версия контракта, должна быть не меньше `1`
`created_at`	string	да	Время создания сообщения
`payload`	object	да	Прикладные данные
`trace_id`	string	нет	32 шестнадцатеричных символа, не может состоять из нулей

JobEnvelope проверяет только общий envelope. Структуру payload проверяет конкретный handler. Например, DemoMessageHandler требует строковое поле payload.message, а DemoSleepHandler — message и seconds от 0 до 30.

Старые сообщения без trace_id поддерживаются. Для них worker создаёт новый trace.

Добавление handler

Шаг 1. Создать handler

<?php

declare(strict_types=1);

namespace Bellissimopizza\RoadRunnerWorker\Handler;

use Bellissimopizza\RoadRunnerWorker\Job\JobEnvelope;
use Bellissimopizza\RoadRunnerWorker\Job\JobExecutionContext;
use Bellissimopizza\RoadRunnerWorker\Job\JobHandler;
use Bellissimopizza\RoadRunnerWorker\Job\MessageBroker;
use Bellissimopizza\RoadRunnerWorker\Job\Subscribe;

#[Subscribe(
    destination: 'company.events',
    messageType: 'orders.created',
    broker: MessageBroker::Kafka,
)]
final readonly class OrderCreatedHandler implements JobHandler
{
    public function handle(JobEnvelope $job, JobExecutionContext $context): void
    {
        $orderId = $job->payload['order_id'] ?? null;

        if (!is_string($orderId) || $orderId === '') {
            throw new \InvalidArgumentException(
                'payload.order_id must be a non-empty string.',
            );
        }

        // Выполнить прикладную операцию.
    }
}

Шаг 2. Зарегистрировать handler

Добавить экземпляр в bin/jobs-worker.php:

$registry = new HandlerRegistry([
    new DemoSleepHandler($logger),
    new DemoKafkaHandler($logger),
    new DemoRabbitMqHandler($logger),
    new OrderCreatedHandler(),
]);

Для production-приложения ручную сборку массива можно заменить контейнером зависимостей, но сам HandlerRegistry ожидает готовый iterable<JobHandler>.

Шаг 3. Настроить pipeline

Если handler использует уже существующий pipeline и физический destination, RoadRunner менять не нужно. Для нового destination добавьте или измените pipeline в .rr.yaml/deployment-конфигурации RoadRunner.

Шаг 4. Добавить тесты

Минимально рекомендуется проверить:

корректный payload;
невалидный payload;
exact/fallback routing по broker, destination, message type и consumer;
lifecycle job.received → job.completed;
lifecycle job.received → job.failed при исключении.

Публикация сообщений

Producer подключается не к Kafka/RabbitMQ, а к RoadRunner RPC:

use Spiral\Goridge\RPC\RPC;
use Spiral\RoadRunner\Jobs\Jobs;

$jobs = new Jobs(RPC::create('tcp://127.0.0.1:6001'));
$queue = $jobs->connect('local');
$task = $queue->create(
    name: $job->type,
    payload: $job->toJson(),
);
$dispatched = $queue->dispatch($task);

Для Kafka физический topic задаётся через KafkaOptions:

use Spiral\RoadRunner\Jobs\KafkaOptions;

$queue = $jobs->connect(
    'kafka-test',
    new KafkaOptions(topic: 'rr-demo'),
);

Полный integration producer находится в bin/publish-test-messages.php.

bin/http-worker.php демонстрирует HTTP-side сценарий: создание trace, формирование JobEnvelope, публикацию в Memory pipeline и логирование job.dispatching/job.dispatched. Несмотря на имя файла, сейчас это CLI-пример producer, а не RoadRunner HTTP worker с request loop.

ACK, NACK и ошибки

bin/jobs-worker.php ожидает задачи в бесконечном consumer loop:

JSON преобразуется в JobEnvelope.
RoadRunner driver преобразуется в MessageBroker.
JobProcessor запускает trace и пишет job.received.
JobRunner вызывает подходящий handler.
При успехе пишется job.completed, затем вызывается $task->ack().
При исключении пишется job.failed, затем вызывается $task->nack($exception).

Ошибка может произойти на трёх этапах:

Этап	Поведение
`decode`	Payload не является корректным `JobEnvelope`; логируется `job.failed`
`routing`	RoadRunner driver не поддерживается до запуска processor; логируется `job.failed` с `failure_stage=routing`
обработка	Handler не найден или выбросил исключение; `JobProcessor` логирует `job.failed` с duration

Поведение повторной доставки, retry и dead-letter queue определяется конфигурацией RoadRunner и брокера. В текущем примере отдельная retry/DLQ политика не настроена.

Trace context

TraceContext хранит контекст только на время одной операции и обязательно очищается в finally, что важно для долгоживущих workers.

producer создаёт trace_id и кладёт его в JobEnvelope;
consumer продолжает тот же trace_id и создаёт новый span_id;
TraceProcessor добавляет trace-поля во все записи текущей операции;
после обработки контекст очищается, поэтому следующий job не наследует trace;
TraceMiddleware умеет читать W3C traceparent версии 00;
для обратной совместимости поддерживается заголовок X-Trace-Id.

Если входной trace отсутствует или невалиден, создаётся новый. Поле trace_flags ограничивается младшим битом sampled-флага.

TraceMiddleware готов для подключения к HTTP request lifecycle, но полноценный HTTP server/request handler в текущей версии проекта ещё не собран.

Логирование

Общий принцип

HTTP-side код, producer и jobs worker создают logger через одну фабрику:

$logger = LoggerFactory::create(
    config: $loggingConfig,
    traceContext: $traceContext,
    component: 'jobs-worker',
);

Каждая запись — один плоский JSON-объект и символ перевода строки:

{"timestamp":"2026-08-02T08:14:28.819256Z","level":"INFO","event":"job.completed","service":"roadrunner-worker","service_namespace":"bellissimo","service_version":"1.0.0","environment":"integration","job_id":"2c8e058e1d5210949f07683fbaf18b6a","job_type":"demo.kafka","job_version":1,"broker":"kafka","topic":"rr-demo","pipeline":"kafka-test","duration_ms":0.433,"component":"jobs-worker","process_pid":18,"trace_id":"4bf92f3577b34da6a3ce929d0e0e4736","span_id":"00f067aa0ba902b7","trace_flags":1}

В JSONL нет вложенных resource, scope или attributes. Это облегчает чтение, поиск и дальнейший parsing в Collector/Grafana.

Основные поля

Поле	Назначение
`timestamp`	UTC, формат RFC 3339 с микросекундами
`level`	Уровень Monolog: `INFO`, `ERROR` и т. д.
`event`	Стабильное имя события
`service`	Имя сервиса
`service_namespace`	Namespace сервиса
`service_version`	Версия сервиса
`environment`	Среда deployment
`component`	Источник: `jobs-worker`, `http-worker`, `integration-producer`
`process_pid`	PID PHP-процесса
`broker`	`kafka`, `rabbitmq` или `memory`
`topic`	Physical destination (`messaging.destination.name`): Kafka topic, RabbitMQ queue или Memory queue
`pipeline`	RoadRunner pipeline producer или consumer task
`consumer`	Стабильный consumer alias для pipeline-aware routing
`job_id`, `job_type`, `job_version`	Данные задачи
`message_id`	ID, возвращённый RoadRunner после dispatch
`duration_ms`	Время обработки задачи
`trace_id`, `span_id`, `trace_flags`	Trace correlation
`error_type`, `error_message`, `error_stacktrace`	Данные исключения
`failure_stage`	`decode` или `routing` для ранней ошибки worker

Дополнительные resource attributes преобразуются из dotted notation в snake_case: например, cloud.region превращается в cloud_region. Массивы сериализуются в JSON-строку. Коллизия полей после flattening считается ошибкой: нельзя одновременно передавать, например, job.id и job_id.

StreamHandler использует file locking, поэтому несколько workers могут безопасно дописывать целые строки в один локальный файл.

Lifecycle events

Event	Кто пишет	Значение
`job.dispatching`	producer	Начало отправки
`job.dispatched`	producer	RoadRunner принял задачу
`job.dispatch_failed`	producer	Отправка завершилась ошибкой
`job.received`	consumer	Worker начал обработку
`job.completed`	consumer	Handler успешно завершён
`job.failed`	consumer	Decode, routing или handler завершился ошибкой
`demo.message.received`	demo handler	Получено demo-сообщение
`demo.sleep.started`	demo handler	Началась sleep-задача
`demo.sleep.finished`	demo handler	Sleep-задача завершилась

Локальный запуск

Требования

PHP 8.4 CLI;
extension sockets;
Composer 2;
совместимый бинарник RoadRunner;
Docker с Compose plugin — для полного integration-окружения.

Установка PHP-зависимостей

composer install

RoadRunner CLI доступен как vendor/bin/rr. Команда get-binary может скачать бинарник для локальной платформы, но способ фиксации версии должен определяться политикой проекта. Dockerfile уже содержит RoadRunner 2025.1.15.

Memory queue

.rr.yaml запускает два jobs workers и pipeline local на Memory driver:

rr serve -c .rr.yaml

Во втором терминале:

php bin/http-worker.php

Producer отправит demo.sleep, worker вызовет DemoSleepHandler, а после успешной обработки подтвердит задачу.

Memory queue существует только внутри процесса RoadRunner и не сохраняет сообщения после перезапуска.

Нативный и изолированный запуск

Нативный режим остаётся основным простым способом запуска. Если бинарник RoadRunner установлен глобально:

rr serve -c .rr.yaml -w .

Если используется Composer binary:

vendor/bin/rr get-binary
./rr serve -c .rr.yaml -w .

Изолированный server runtime запускает те же RoadRunner и PHP workers в Docker, не добавляя Kafka или RabbitMQ в Compose:

vendor/bin/rr-worker up

Оба режима читают .rr.yaml из проекта пользователя. Для параметров, которые различаются между host и container, можно использовать подстановку окружения:

version: "3"

rpc:
  listen: ${RR_RPC_LISTEN:-tcp://127.0.0.1:6001}

При нативном старте сработает безопасный loopback default. В контейнере CLI передаст RR_RPC_LISTEN=tcp://0.0.0.0:6001, а наружу RPC всё равно будет опубликован только на заданном loopback-адресе host.

Kafka и RabbitMQ через Docker

docker compose -f compose.test.yaml up --build

Compose запускает Kafka, RabbitMQ, RoadRunner-приложение и одноразовый producer. Producer автоматически отправляет demo.kafka и demo.rabbitmq.

Проверить состояния:

docker compose -f compose.test.yaml ps -a

Ожидаемый результат: app, kafka, rabbitmq healthy, producer завершён с кодом 0.

Прочитать application logs:

docker compose -f compose.test.yaml exec app \
  tail -f /var/log/app/application.jsonl

Остановить окружение:

docker compose -f compose.test.yaml down

Удалить также volume с тестовыми логами:

docker compose -f compose.test.yaml down -v

RabbitMQ Management UI: http://localhost:15672, пользователь и пароль — roadrunner.

Тестирование

Запуск всего unit test suite:

php vendor/bin/phpunit

Текущие тесты покрывают:

сериализацию и валидацию JobEnvelope;
mapping RoadRunner driver → MessageBroker;
правила и ошибки HandlerRegistry;
успешный и ошибочный lifecycle JobProcessor;
работу demo handlers;
parsing traceparent и очистку trace context;
плоский JSONL formatter;
logger factory для HTTP-side компонента;
конфигурацию логирования;
runtime config, validation, Compose orchestration и все CLI-команды;
структуру Kafka/RabbitMQ authentication examples.

Integration-проверка брокеров выполняется через compose.test.yaml и bin/publish-test-messages.php.

2. Руководство DevOps

RoadRunner

RoadRunner выполняет две роли:

управляет пулом долгоживущих PHP workers;
владеет подключениями к брокерам и доставкой задач через Jobs plugin.

PHP worker не содержит Kafka/AMQP client и не хранит broker credentials. Credentials и адреса брокеров находятся в конфигурации RoadRunner или во внешней системе секретов, используемой deployment-платформой.

`.rr.yaml`

Минимальное локальное окружение:

RPC слушает 127.0.0.1:6001;
worker command: php bin/jobs-worker.php;
relay: pipes;
pipeline local использует Memory driver;
RoadRunner запускает два workers.

`.rr.test.yaml`

Integration-окружение:

RPC слушает 0.0.0.0:6001 внутри контейнера;
Kafka broker: kafka:9092;
AMQP address: rabbitmq:5672;
pipelines: kafka-test, rabbitmq-test;
consume list содержит оба pipeline;
pool содержит два PHP workers;
RoadRunner technical log level: debug.

В production не следует публиковать RPC-порт наружу без сетевых ограничений. Producer должен обращаться к нему через private network/service discovery.

Docker

Образ приложения

Dockerfile использует multi-stage build:

берёт бинарник RoadRunner 2025.1.15;
берёт Composer 2;
собирает PHP 8.4 CLI runtime с extension sockets;
устанавливает production Composer dependencies с authoritative classmap;
копирует vendor, RoadRunner и приложение в финальный образ.

Финальная команда образа использует .rr.test.yaml, поэтому для production нужно передать свою команду/config либо создать отдельный deployment image.

Сервисы `compose.test.yaml`

Сервис	Роль	Healthcheck/завершение
`kafka`	Kafka broker в KRaft single-node режиме	`kafka-topics --list`
`rabbitmq`	RabbitMQ + Management UI	`rabbitmq-diagnostics ping`
`app`	RoadRunner и jobs workers	TCP-проверка RPC `6001`
`producer`	Одноразовая отправка двух сообщений	Ожидается exit code `0`

Compose создаёт named volume application_logs, общий для app и producer. Оба процесса пишут в /var/log/app/application.jsonl с file locking.

Открытые порты тестового окружения:

Порт	Назначение
`6001`	RoadRunner RPC
`15672`	RabbitMQ Management UI

Kafka и AMQP доступны только внутри Compose network.

Изолированный server runtime

Package содержит CLI vendor/bin/rr-worker, который собирает immutable image из текущего проекта и управляет одним сервисом: RoadRunner с PHP workers. Kafka, RabbitMQ, Collector, Loki и другие инфраструктурные компоненты этот Compose намеренно не запускает.

После подключения package через настроенный Composer repository:

composer require bellissimopizza/road-runner-worker
vendor/bin/rr-worker validate
vendor/bin/rr-worker up

CLI и Docker templates берутся из установленной версии package, а application context, composer.json, .dockerignore и .rr.yaml — из проекта пользователя.

Команды

Команда	Назначение
`vendor/bin/rr-worker validate`	Проверить проект, `.rr.yaml`, Docker и итоговый Compose
`vendor/bin/rr-worker up`	Собрать immutable image и запустить runtime в фоне
`vendor/bin/rr-worker down`	Остановить и удалить runtime-контейнер и network, сохранив volume логов
`vendor/bin/rr-worker restart`	Перезапустить уже запущенный runtime
`vendor/bin/rr-worker status`	Показать состояние и вернуть ненулевой exit code, если runtime не работает
`vendor/bin/rr-worker logs`	Показать последние 200 строк и продолжить чтение (`follow`)

Команды можно выполнять из корня проекта или его подкаталога. Корень определяется по ближайшему composer.json. Для up сначала выполняются те же проверки, что и для validate.

validate проверяет структуру и YAML-конфигурацию, Docker Engine, Docker Compose, secret/certificate mounts и итоговый Compose. Он не подключается к Kafka или RabbitMQ: broker connectivity должна проверяться отдельной readiness или deployment-проверкой в инфраструктуре.

Immutable image

Default Dockerfile находится внутри package. Во время up он:

копирует текущий проект в build stage;
выполняет composer install --no-dev --classmap-authoritative;
переносит приложение и production dependencies в финальный image;
запускает rr serve с конфигурацией проекта.

Исходный код не bind-mountится в контейнер, поэтому запущенный runtime не меняется вслед за файлами на host. Любое изменение приложения применяется новой сборкой через vendor/bin/rr-worker up.

Если проекту нужен собственный PHP image или extensions, задайте RR_WORKER_DOCKERFILE. Dockerfile должен принимать build arguments RR_WORKER_PHP_VERSION и RR_WORKER_ROADRUNNER_VERSION, копировать application context и запускать RoadRunner. Default Dockerfile полезен как reference implementation.

Корневой .dockerignore обязателен и должен как минимум исключать:

.rr-worker.env
.rr-worker

Рекомендуется также исключить .env, .env.*, .git, .idea, локальный vendor и бинарник rr. Это уменьшает build context и не позволяет случайно запечь runtime secrets в image.

Сеть и RPC

По умолчанию RPC публикуется как 127.0.0.1:6001. Формат RR_WORKER_RPC_BIND — host:port, например:

RR_WORKER_RPC_BIND=127.0.0.1:6101

Для подключения контейнера к заранее созданной Docker network задайте RR_WORKER_NETWORK. Network считается external и автоматически не создаётся. В конфигурации .rr.yaml broker address должен быть доступен из этой network. Если брокер находится вне Docker host, используйте его DNS/IP, а не container loopback.

Аутентификация Kafka и RabbitMQ

Поддержка определяется RoadRunner Jobs plugin и конфигурацией пользователя. Package не хранит credentials и не добавляет PHP broker clients.

Broker	Варианты конфигурации
Kafka	без аутентификации, TLS, mTLS, SASL/PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
RabbitMQ	login/password в DSN, TLS, mTLS

Готовые фрагменты находятся в resources/examples/auth/:

kafka-no-auth.yaml;
kafka-sasl.yaml;
kafka-tls.yaml;
rabbitmq-password.yaml;
rabbitmq-tls.yaml.

Скопируйте нужный broker section в собственную .rr.yaml и замените имена переменных окружения согласно вашей системе секретов.

Environment, secrets и certificates

Несекретные параметры можно хранить в .rr-worker.env либо передавать через окружение процесса. Значения из окружения имеют приоритет. Файл не обязателен и не копируется в image.

Для secret-файлов поддерживается соглашение NAME_FILE. Например:

KAFKA_USERNAME_FILE=/run/rr-worker/secrets/kafka-user
KAFKA_PASSWORD_FILE=/run/rr-worker/secrets/kafka-password

Host-каталог задаётся через RR_WORKER_SECRETS_DIR и монтируется read-only в /run/rr-worker/secrets. Entrypoint читает файл, экспортирует NAME, удаляет NAME_FILE из окружения и не печатает значение. Одновременное определение NAME и NAME_FILE считается ошибкой.

TLS-файлы монтируются отдельно:

RR_WORKER_CERTS_DIR=/srv/orders/certs

В .rr.yaml используются container paths:

kafka:
  brokers:
    - ${KAFKA_BROKER}
  tls:
    root_ca: /run/rr-worker/certs/ca.pem
    cert: /run/rr-worker/certs/client.pem
    key: /run/rr-worker/certs/client-key.pem

При нативном запуске контейнерных mounts нет: передайте host paths через переменные в .rr.yaml или используйте отдельные значения окружения для host. Пример всех переменных находится в resources/examples/.rr-worker.env.example.

Переменные окружения

Application logging

Переменная	Default	Назначение
`LOG_CHANNEL`	`app`	Канал Monolog
`LOG_LEVEL`	`info`	Минимальный уровень логов
`LOG_STREAM`	`php://stderr`	Stream или путь JSONL-файла
`OTEL_SERVICE_NAME`	`roadrunner-worker`	Имя сервиса
`OTEL_SERVICE_NAMESPACE`	`bellissimo`	Namespace сервиса
`OTEL_SERVICE_VERSION`	`1.0.0`	Версия deployment
`OTEL_DEPLOYMENT_ENVIRONMENT`	`development`	`development`, `integration`, `production` и т. п.
`OTEL_RESOURCE_ATTRIBUTES`	пусто	Дополнительные `key=value`, разделённые запятыми

Пример:

export LOG_STREAM=/var/log/app/application.jsonl
export LOG_LEVEL=info
export OTEL_SERVICE_NAME=orders-worker
export OTEL_SERVICE_NAMESPACE=bellissimo
export OTEL_SERVICE_VERSION=2026.08.02
export OTEL_DEPLOYMENT_ENVIRONMENT=production
export OTEL_RESOURCE_ATTRIBUTES='cloud.region=uz-tas-1,service.instance.id=worker-01'

Значения OTEL_RESOURCE_ATTRIBUTES не поддерживают escaping запятых или знака =. Для сложных значений парсер необходимо расширить.

Integration producer

Переменная	Default	Назначение
`RR_RPC_ADDRESS`	`tcp://127.0.0.1:6001`	Адрес RoadRunner RPC

Server runtime CLI

Переменная	Default	Назначение
`RR_WORKER_CONFIG`	`.rr.yaml`	RoadRunner config относительно корня проекта
`RR_WORKER_ENV_FILE`	`.rr-worker.env`	Optional env-файл относительно корня проекта
`RR_WORKER_DOCKERFILE`	Dockerfile package	Пользовательский Dockerfile
`RR_WORKER_CERTS_DIR`	не задан	Host-каталог TLS certificates для read-only mount
`RR_WORKER_SECRETS_DIR`	не задан	Host-каталог secret-файлов для read-only mount
`RR_WORKER_NETWORK`	не задан	Существующая external Docker network
`RR_WORKER_RPC_BIND`	`127.0.0.1:6001`	Публикуемый host address и port RPC
`RR_WORKER_PROJECT_NAME`	вычисляется	Имя Docker Compose project
`RR_WORKER_IMAGE`	`<project>:latest`	Имя собираемого immutable image
`RR_WORKER_PHP_VERSION`	`8.4`	Build argument версии PHP
`RR_WORKER_ROADRUNNER_VERSION`	`2025.1.15`	Build argument версии RoadRunner
`RR_WORKER_STOP_GRACE_PERIOD`	`30s`	Grace period перед остановкой контейнера

Пути RR_WORKER_CONFIG, RR_WORKER_DOCKERFILE, RR_WORKER_CERTS_DIR и RR_WORKER_SECRETS_DIR могут быть абсолютными или относительными к корню проекта. Точный состав broker variables зависит от .rr.yaml пользователя.

Collector

Переменная	Пример	Назначение
`APPLICATION_LOG_PATH`	`/var/log/app/*.jsonl`	Файлы application logs
`LOKI_OTLP_ENDPOINT`	`http://loki:3100/otlp`	OTLP HTTP endpoint Loki

Application и infrastructure logs

Логи намеренно разделены:

PHP application logs → /var/log/app/application.jsonl;
RoadRunner, Kafka и RabbitMQ technical logs → container stdout/stderr.

Причина разделения: Collector должен читать только гарантированно валидные JSONL-записи приложения. Technical logs имеют другой формат и не смешиваются с прикладными событиями.

RoadRunner использует worker STDOUT для Goridge protocol, поэтому PHP-код не должен писать application logs в stdout. Default php://stderr безопасен для локальной разработки. В файловом варианте используется /var/log/app.

Application logs:

docker compose -f compose.test.yaml exec app \
  tail -f /var/log/app/application.jsonl

Infrastructure logs:

docker compose -f compose.test.yaml logs -f app kafka rabbitmq

Named volume сохраняется после обычного docker compose down. Это позволяет Collector продолжить чтение после пересоздания контейнера, но требует политики rotation и retention.

OpenTelemetry Collector и Grafana Loki

Пример конфигурации находится в deploy/otel-collector/config.yaml. Приложение не подключается к Collector и не отправляет OTLP напрямую.

Поток логов:

flowchart LR
    A["PHP app / workers"]
    F["application.jsonl"]
    C["OTel Collector Contrib file_log"]
    T["Transform processor"]
    L["Loki OTLP endpoint"]
    G["Grafana Explore"]

    A --> F --> C --> T --> L --> G

Collector выполняет следующие действия:

file_log/application читает JSONL;
json_parser переносит поля в LogRecord attributes;
timestamp, level, trace_id, span_id, trace_flags превращаются в стандартные поля OpenTelemetry LogRecord;
event становится body записи;
service-поля переносятся в resource attributes;
batch группирует записи;
debug показывает результат проверки;
otlphttp/loki отправляет записи в Loki.

Нужна Contrib-сборка Collector, потому что core-сборка не содержит file_log.

Пример запуска Collector вне Docker Compose:

export APPLICATION_LOG_PATH='/var/log/app/*.jsonl'
export LOKI_OTLP_ENDPOINT='http://loki:3100/otlp'

otelcol-contrib --config deploy/otel-collector/config.yaml

Путь LOKI_OTLP_ENDPOINT должен завершаться на /otlp. Экспортёр самостоятельно добавляет /v1/logs.

После проверки pipeline exporter debug можно убрать. В Loki 3.x structured metadata обычно включены по умолчанию. Для старых установок может потребоваться:

limits_config:
  allow_structured_metadata: true

Примеры LogQL в Grafana Explore:

{service_name="roadrunner-worker"}

{service_name="roadrunner-worker"} | broker="kafka"

{service_name="roadrunner-worker"} | trace_id="4bf92f3577b34da6a3ce929d0e0e4736"

{service_name="roadrunner-worker"} |= "job.failed"

Фактический синтаксис фильтрации structured metadata зависит от версии Loki и способа индексации. Перед production rollout проверьте запросы на вашей версии.

Диагностика

Симптом	Возможная причина	Проверка/решение
`No handler subscribed...`	Не совпадают consumer alias, broker, physical destination или `JobEnvelope.type`	Проверить `consumer_pipelines`, `#[Subscribe]`, driver, queue/topic и type
`must declare exactly one #[Subscribe]`	Атрибута нет или их несколько	Оставить ровно один атрибут
`Duplicate subscription...`	Два handler используют один consumer, broker, destination и message type	Изменить ключ маршрутизации или убрать дубль
`Unsupported message broker...`	RoadRunner вернул неподдерживаемый driver	Добавить case в `MessageBroker` и handler strategy
`Job payload must be an object`	Payload не JSON object	Проверить producer serialization
Ошибка `trace_id`	ID имеет неверный формат или состоит из нулей	Передавать 32 hex-символа
Producer не подключается	Неверный `RR_RPC_ADDRESS` или RPC недоступен	Проверить port/network и RoadRunner logs
`rr-worker validate` не проходит	Некорректны `.rr.yaml`, Dockerfile, `.dockerignore`, mount или Compose	Исправить все errors; warnings не блокируют запуск
Broker недоступен после `up`	Container не видит DNS/network брокера	Задать `RR_WORKER_NETWORK` и проверить broker address из этой network
`_FILE` secret отклонён	Заданы и `NAME`, и `NAME_FILE`, либо путь вне secret mount	Оставить один источник и путь `/run/rr-worker/secrets/...`
TLS-файл не найден	Не задан `RR_WORKER_CERTS_DIR` или container path неверен	Проверить read-only mount `/run/rr-worker/certs` и host-файл
`rr-worker status` возвращает `1`	Runtime отсутствует, остановлен или unhealthy	Посмотреть `vendor/bin/rr-worker logs` и Docker health status
Kafka job не приходит	Не совпал pipeline, physical topic или consumer group	Проверить `.rr.test.yaml` и `KafkaOptions`
RabbitMQ job не приходит	Не совпали queue/exchange/routing key	Проверить AMQP pipeline и Management UI
Нет application JSONL	Неверный `LOG_STREAM` или volume не смонтирован	Проверить env, каталог и права записи
Collector не видит старые записи	Receiver использует `start_at: end`	Это ожидаемо: читаются новые записи после старта
Collector не знает `file_log`	Запущена core-сборка	Использовать OpenTelemetry Collector Contrib
Loki exporter возвращает 404	Неверный endpoint	Использовать base endpoint, заканчивающийся `/otlp`
В stdout нет application events	Логи пишутся в JSONL-файл	Это ожидаемое разделение логов

3. Справочник

Структура каталогов

.
├── bin/
│   ├── jobs-worker.php              # Consumer loop
│   ├── http-worker.php              # CLI-пример HTTP-side producer
│   ├── publish-test-messages.php    # Kafka/RabbitMQ integration producer
│   ├── rr-kafka-worker.php          # Laravel consumer из Composer vendor/bin
│   └── rr-worker                    # CLI изолированного server runtime
├── config/
│   └── logging.php                  # Переменные логирования
├── deploy/
│   ├── brokers/README.md            # Integration-окружение брокеров
│   └── otel-collector/
│       ├── config.yaml              # Пример Collector
│       └── README.md                 # Подключение Collector/Loki
├── src/
│   ├── Console/                     # Команды up/down/restart/status/logs/validate
│   ├── Handler/                     # Demo handlers
│   ├── Http/                        # Trace middleware и HTTP-заготовки
│   ├── Job/                         # Envelope, routing и lifecycle
│   ├── Logging/                     # JSONL logger и trace context
│   └── Runtime/                     # Config, validation и Docker Compose orchestration
├── resources/
│   ├── compose/server.yaml          # Compose только для RoadRunner + PHP workers
│   ├── docker/server/               # Immutable server image и entrypoint
│   └── examples/auth/               # Kafka/RabbitMQ auth examples
├── tests/                            # PHPUnit tests
├── .rr.yaml                         # Локальный Memory pipeline
├── .rr.test.yaml                    # Kafka/RabbitMQ integration config
├── compose.test.yaml                # Полное тестовое окружение
└── Dockerfile                       # Reference application image

Поддерживаемые брокеры

`MessageBroker`	Значение в логах	RoadRunner driver
`MessageBroker::Kafka`	`kafka`	`Driver::Kafka`
`MessageBroker::RabbitMq`	`rabbitmq`	`Driver::AMQP`
`MessageBroker::Memory`	`memory`	`Driver::Memory`

Добавление нового broker требует не только нового enum case, но и поддержки соответствующего RoadRunner driver/plugin, конфигурации pipeline и integration тестов.

Полезные команды

# Unit-тесты
php vendor/bin/phpunit

# Проверка изолированного server runtime
vendor/bin/rr-worker validate

# Сборка immutable image и запуск RoadRunner + PHP workers
vendor/bin/rr-worker up

# Состояние и streaming логов server runtime
vendor/bin/rr-worker status
vendor/bin/rr-worker logs

# Перезапуск и остановка server runtime
vendor/bin/rr-worker restart
vendor/bin/rr-worker down

# Проверка Compose-конфигурации
docker compose -f compose.test.yaml config

# Сборка и запуск integration stack
docker compose -f compose.test.yaml up -d --build

# Состояние всех контейнеров, включая producer
docker compose -f compose.test.yaml ps -a

# Application logs
docker compose -f compose.test.yaml exec app \
  tail -f /var/log/app/application.jsonl

# Infrastructure logs
docker compose -f compose.test.yaml logs -f app kafka rabbitmq

# Остановка с сохранением application log volume
docker compose -f compose.test.yaml down

# Полная очистка тестовых контейнеров и volume
docker compose -f compose.test.yaml down -v

Текущие ограничения

Нет автоматического discovery handlers: регистрация выполняется вручную.
Один handler может иметь только один #[Subscribe].
Core demo worker собирает handlers вручную; Laravel vendor worker создаёт явно перечисленные roadrunner.handlers через Laravel Service Container. Для producer-only runtime отсутствующий или пустой список разрешён.
Нет отдельной retry/DLQ политики в репозитории.
Нет полноценного RoadRunner HTTP request loop; http-worker.php — producer example, а RequestHandler пока является заготовкой.
Приложение формирует OpenTelemetry-совместимые JSONL-поля, но не использует OpenTelemetry SDK и не отправляет OTLP напрямую.
Collector и Loki не входят в compose.test.yaml; предоставлен только пример конфигурации для будущего подключения.
Изолированный server runtime предназначен для immutable deployment. Hot reload и bind mounts для local development в него не входят; нативный запуск при этом полностью поддерживается.
Тестовые Kafka/RabbitMQ настройки не предназначены для production.

bellissimopizza / road-runner-worker

Maintainers

Package info

Statistics

Security