README

Опциональный AMPHP runtime и Symfony Console CLI для romanfedorskij/message-bus-async. Пакет получает durable jobs из PostgreSQL, открывает application scope на job, выполняет точный queued binding и сохраняет результат с проверкой lease.

Линия 0.1.x требует совместимый core ^0.1.

Когда Пакет Нужен

durable messages из core нужно выполнять отдельными процессами;
job должен иметь heartbeat, retry, cancellation и stale recovery;
обработчики должны получать новый service scope для каждой job;
workers нужно ставить на pause, resume, drain, stop или restart;
runtime должен работать кооперативно в AMPHP event loop без скрытого fork pool.

Возможности

Задача	Тип или команда	Документ
Claim и guarded transitions PostgreSQL job	`PostgresMessageConsumer`	Consumer
Кооперативный consumer loop	`QueueWorkerRunner`	Runner
Выполнение точного queued binding	`MessageBusQueueWorker`	Runner
Scope на одну job	`WorkerModule` или application scope factory	CLI
Durable control и status	`PostgresWorkerControlGateway`	Control plane
Операционный вход	`vendor/bin/message-bus-worker`	CLI

Core отвечает за dispatch, envelope codecs, enqueue/status/cancel и общую queue schema. Зависимость направлена только от worker к core.

Требования

PHP 8.4.12 или новее в пределах PHP 8;
romanfedorskij/message-bus-async ^0.1;
AMPHP 3 и amphp/postgres 2;
ext-pgsql или ext-pq;
Symfony Console 7.4 или 8.1.

ext-pcntl и Thesis DIC опциональны. Без PCNTL shutdown запрашивается через worker control или другой host adapter.

Установка

До регистрации пакетов в Packagist добавьте оба GitHub VCS repository:

composer config repositories.message-bus-async vcs \
  https://github.com/wolfcharaa/message-bus-async.git
composer config repositories.message-bus-async-worker vcs \
  https://github.com/wolfcharaa/message-bus-async-worker.git
composer require romanfedorskij/message-bus-async-worker:^0.1

Executable появится как vendor/bin/message-bus-worker.

Первый Worker

Host application предоставляет PHP bootstrap, который возвращает WorkerRuntime или рекомендуемый WorkerRuntimeProviderInterface. Provider удерживает pools, clients и container scopes на протяжении всей CLI operation и закрывает их после неё.

vendor/bin/message-bus-worker worker:run \
  --bootstrap=config/message-bus-worker.php \
  --transport=postgres \
  --queue=addresses

Пакет не ищет application container и не создаёт schema при старте. Сгенерируйте worker-control DDL командой worker:schema:postgres и примените его через versioned migration приложения. Полные варианты bootstrap и команды управления приведены в руководстве по CLI.

Как Выполняется Job

Consumer claims одну доступную строку и фиксирует lease.
Runner открывает новый job scope и разрешает queue worker.
Параллельный monitor обновляет heartbeat и проверяет cancellation/control.
Точный binding из producer выполняется локально и не ставится в очередь второй раз.
Consumer делает ack, retry, reject или cancel с проверкой lease owner, затем scope закрывается.

Delivery — at-least-once. После crash или stale lease сообщение может выполниться повторно, поэтому state-changing handlers должны быть идемпотентными. CPU-bound handler обязан иметь cooperative suspension points или выполняться через отдельный process adapter.

Граница Процесса

Один CLI process запускает один cooperative worker. Количество процессов, restart по exit code 75, hard kill и resource limits настраиваются в systemd, Supervisor, Kubernetes, Docker или другом внешнем supervisor. Пакет не создаёт скрытый fork pool.

Документация

Worker CLI — bootstrap, run, status, control, schema и exit codes.
PostgreSQL consumer — claim, lease, transitions, retry и stale recovery.
Worker runner — scopes, heartbeat, cancellation и outcome mapping.
Worker control — desired state, targeting, acknowledgements и storage ownership.
Core queue contract — producer-side schema и delivery guarantees.

Разработка

Source workspace ожидает core checkout в ../MessageBusAsync; Composer подключает его как 0.1.0 через path repository.

composer validate --strict
composer analyse
composer test

Optional integration suite требует disposable PostgreSQL database:

MESSAGE_BUS_TEST_POSTGRES_CONFIG='host=127.0.0.1 port=5432 user=test password=test db=test' \
  composer test:integration

Лицензия

MIT

romanfedorskij / message-bus-async-worker

Package info

Security

Statistics