Kislay Queue

Native PHP distributed job queue for long-running services. Phase 1 ships a standalone queue server, producer client, worker client, retries, delayed jobs, and DLQ support.

Part of the KislayPHP ecosystem.

What It Is

kislayphp/queue now has two modes:

• Kislay\Queue\Server for the standalone queue node
• Kislay\Queue\Client for producers and operational reads
• Kislay\Queue\Worker for consumers
• Kislay\Queue\Job for ack/nack control inside a handler
• Kislay\Queue\Queue as the legacy local in-process queue for development fallback

Delivery model in 0.0.4:

• at-least-once delivery
• one leased job per worker fetch
• retries with backoff
• delayed jobs
• dead-letter queue support
• in-memory server state only

Install

pie install kislayphp/queue:0.0.4

Enable in php.ini:

extension=kislayphp_queue.so

5-Minute Quickstart

1. Start the queue server

<?php
$server = new Kislay\Queue\Server();
$server->declare('emails', [
    'visibility_timeout_ms' => 30000,
    'max_attempts' => 5,
    'retry_backoff_ms' => 1000,
    'dead_letter_queue' => 'emails.dlq',
]);
$server->listen('0.0.0.0', 9020);
$server->run();

2. Push a job

<?php
$client = new Kislay\Queue\Client('http://127.0.0.1:9020');

$jobId = $client->push('emails', [
    'to' => 'user@example.com',
    'subject' => 'Welcome',
], [
    'headers' => ['trace_id' => 'trace-1'],
    'max_attempts' => 5,
]);

var_dump($jobId);

3. Run a worker

<?php
$worker = new Kislay\Queue\Worker('http://127.0.0.1:9020');

$worker->consume('emails', function (Kislay\Queue\Job $job) {
    $payload = $job->payload();
    $headers = $job->headers();

    sendEmail($payload['to'], $payload['subject']);

    return true;
}, [
    'worker_id' => 'emails-worker-1',
    'lease_ms' => 30000,
]);

Architecture

+------------------+        push         +----------------------+
| Producer Service | ------------------> | Kislay\Queue\Server |
| Client           |                     | owns queue state     |
+------------------+                     | leases jobs          |
                                         | retries / DLQ        |
+------------------+        fetch        +----------+-----------+
| Worker Service   | <------------------------------+
| Kislay\Queue\Worker |         ack / nack          |
+------------------+                                 |
                                                     v
                                            +------------------+
                                            | queue.dlq        |
                                            +------------------+

Public API

namespace Kislay\Queue;

class Server {
    public function __construct(array $options = []);
    public function listen(string $host, int $port): bool;
    public function run(): void;
    public function stop(): bool;
    public function declare(string $queue, ?array $options = null): bool;
    public function stats(?string $queue = null): array;
}

class Client {
    public function __construct(string $baseUrl, array $options = []);
    public function push(string $queue, mixed $payload, ?array $options = null): string;
    public function pushBatch(string $queue, array $jobs): array;
    public function stats(string $queue): array;
    public function purge(string $queue): int;
}

class Worker {
    public function __construct(string $baseUrl, array $options = []);
    public function consume(string $queue, callable $handler, ?array $options = null): bool;
    public function stop(): bool;
}

class Job {
    public function id(): string;
    public function queue(): string;
    public function payload(): mixed;
    public function headers(): array;
    public function attempts(): int;
    public function maxAttempts(): int;
    public function availableAt(): int;
    public function ack(): bool;
    public function nack(?bool $requeue = true, ?int $delayMs = null): bool;
    public function release(?int $delayMs = null): bool;
}

Legacy local queue API remains available:

$queue = new Kislay\Queue\Queue();
$queue->enqueue('jobs', ['task' => 'send_email']);
$job = $queue->dequeue('jobs');

Queue Semantics

0.0.4 semantics are intentionally explicit:

• push() creates a job in ready or delayed
• Worker::consume() fetches one leased job at a time
• if the handler returns true, the worker acks the job
• if the handler returns false, the worker nacks the job
• if a handler throws, the worker nacks the job and stops with the exception still raised
• if retries are exhausted, the job moves to the configured DLQ

Operational Notes

What 0.0.4 is good for:

• background task processing between services
• queue semantics for microservice workloads
• development and early production validation
• explicit retry / DLQ flows

What 0.0.4 does not do yet:

• durable persistence across queue server restarts
• worker concurrency inside one consume() loop
• distributed replication
• exactly-once delivery

If the queue server process stops, in-memory jobs are lost. That is the correct tradeoff for this phase; do not market it as durable yet.

End-to-End Example

See example.php for a minimal runnable server / producer / worker flow.

Ecosystem Fit

Use this module for job queue semantics.

Use eventbus for fanout and realtime pub/sub.

Typical flow:

HTTP request -> queue push -> worker processes job -> eventbus emits completion event

Docs

• Full docs: https://skelves.com/kislayphp/docs/queue
• Insight articles: https://skelves.com/insights

License

Apache License 2.0