spiral-modules / jobs Goto Github PK

View Code? Open in Web Editor NEW

60.0 14.0 12.0 608 KB

RoadRunner: Background PHP workers, Queue brokers

License: MIT License

PHP 10.71% Go 89.15% Makefile 0.13%

php7 golang spiral queue jobs async roadrunner beanstalk amazon-sqs amqp

jobs's Introduction

RoadRunner: Background PHP workers, Queue brokers

Documentation

Features

supports in memory queue, Beanstalk, AMQP, AWS SQS
can work as standalone application or as part of RoadRunner server
multiple pipelines per application
durable (prefetch control, graceful exit, reconnects)
automatic queue configuration
plug-and-play PHP library (framework agnostic)
delayed jobs
job level timeouts, retries, retry delays
PHP and Golang consumers and producers
per pipeline stop/resume
interactive stats, events, RPC
works on Windows

License:

The MIT License (MIT). Please see LICENSE for more information. Maintained by Spiral Scout.

jobs's People

Contributors

Stargazers

Watchers

Forkers

koltsov-spiralscout alex-bond vl-ru cutecube korol8484 siad007 patoui renowncoder sergey-telpuk coresh suryatmodulus

jobs's Issues

[Feature Request] Allow passing objects into jobs

Allow passing objects into jobs

$queue->push(new SomeJob([
                'user' => $user // User object
            ]));

---------------------------

class SomeJob extends AbstractJob
{
    public function do(User $user)

[BUG] Error "missing header rr-attempt" although this header exist in AMQP message

RoadRunner, PHP Application Server
Version: 2.4.18, 2020-05-23T19:45:45+0300

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

chdir(__DIR__);
require_once __DIR__ . '/vendor/autoload.php';

$id = bin2hex(random_bytes(16));
$message = new AMQPMessage(
    'test',
    [
        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
        'content_type' => 'application/octet-stream',
        'application_headers' => new AMQPTable(
            [
                'rr-id' => $id,
                'rr-attempt' => 0,
                'rr-delay' => 0,
                'rr-job' => 'local.test',
                'rr-maxAttempts' => 0,
                'rr-retryDelay' => 0,
                'rr-timeout' => 0,
            ]
        ),
    ]
);

$amqp = new AMQPStreamConnection('rabbitmq', '5672', 'guest', 'guest');
$amqp->channel()->basic_publish($message, 'local.test', 'local.test');

Screenshot from RabbitMQ admin panel:

The same error when I send a message from the RabbitMQ admin panel.

Retryable jobs support

If jobs fail, they can be restarted after the specified TTR

issue from github spiral/jobs2

Add job routing based on job handler response

Separate interface HandlerInterface to specific ones

Some kind of

Serializable interface with methods
    public function serialize(string $jobType, array $payload): string;
    public function unserialize(string $jobType, string $payload): array;

HandlerBased interface with method
    public function handle(string $jobType, string $jobID, array $payload): void;

Typecast rr-headers for AMQP driver

issue from github spiral/jobs3

Dedicate documentation section

PHP 8 / RoadRunner 2 Compatibility

I received the conflict below when attempting to add spiral/jobs to a project using PHP 8.

$ composer require -W spiral/jobs
Using version ^2.2 for spiral/jobs
./composer.json has been updated
Running composer update spiral/jobs --with-all-dependencies
Loading composer repositories with package information
Updating dependencies
Your requirements could not be resolved to an installable set of packages.

  Problem 1
    - spiral/jobs v2.2.0 requires php ^7.2 -> your php version (8.0.12) does not satisfy that requirement.
    - spiral/jobs[v2.2.1, ..., v2.2.2] require spiral/roadrunner ^1.1 -> found spiral/roadrunner[v1.1.0, ..., v1.9.2] but it conflicts with your root composer.json require (^2.5).
    - Root composer.json requires spiral/jobs ^2.2 -> satisfiable by spiral/jobs[v2.2.0, v2.2.1, v2.2.2].

You can also try re-running composer require with an explicit version constraint, e.g. "composer require spiral/jobs:*" to figure out if any version is installable, or "composer require spiral/jobs:^2.1" if you know which you need.

Installation failed, reverting ./composer.json and ./composer.lock to their original content.
ERROR: 2

[CI] Move to GitHub Actions

Eliminate usage of Travis and other CI systems. Combine all CI in the GitHub actions workflow.

[BUG] Setting maxAttempts don't work

Version v1.1.3.

maxAttempts don't work as in docs https://spiral.dev/docs/queue-jobs

    local:
        broker: amqp
        queue: my-queue
        maxAttempts: 2

This this setup works instead:

dispatch:
    jobs-remote-*:
      pipeline: beanstalk
      Attempts: 5
      retryDelay: 10
      timeout: 10000
      delay: 10

Looks like somehow tag `json:"maxAttempts,omitempty" don't make effect.

Implement better backoff strategy

Currently most of jobs brokers use static reconnect duration which equals to timeout value. Need to implement backoff strategy with exponential timeout increases after some amount of failures.

Job pipelines are neither pause not resumed.

On MacBook i7 16GB
Workers: 2
Broker: AMQP

I tried to stress run a job by command: ab -c 100 -n 5000 http://localhost:8080/

In the middle, I paused the consumption by command: ./spiral jobs:pause local

In ideal scenario, The jobs should be paused and should not be released.

But the jobs are executing without any pause.

https://storyxpress.co/video/k9d272h6sjy6fhske

[BUG] RR is losing jobs under load

I have a simple flow where I do stuff in a DB transaction scope, after which I send a job to the AMQP queue and finish the request.

I do understand that jobs are not guaranteed to reach the queue if server shuts down or client disconnects and cancels the request abruptly, but I didn't expect this to happen so often under a seemingly not very heavy load (barely a request per second on average, where request is a fairly heavy write op though).

I can expect to have about 10-20 jobs per day to be lost and not persisted into the queue. The worst part is that I can't track them down by noticing an exception or something.

AMQP consumer configuration

It would be nice to be able to fine-tune amqp consumer configuration. Options I would consider the most important:

Disabling/enabling automatic ACK
Setting prefetch size.

Fix readme, some links are dead

How can I find out that the job ended with an error

Hello.
How can I find out (or catch or log) that the job ended with an error.
The BaseTest check the same $this->assertNotEmpty($id); for both ("error job" and "success job")

I found an example in spiral/framework JobDispatcher.php:

        $consumer->serve(function (\Throwable $e = null) {
            if ($e !== null) {
                $this->handleException($e);
            }
            $this->finalizer->finalize(false);
        });

Is this the only way to catch it? Any other ways to catch it in "producers" ?

spiral/jobs[v2.2.0, ..., v2.2.1] require spiral/roadrunner ^1.1 -> found spiral/roadrunner[v1.1.0, ..., v1.9.2] but it conflicts with your root composer.json require (^2.0).

But in order to use spiral/roadrunner with php ^8.0 I need the version ^2.0 of it. So I think there is a wrong version requirement in you composer?

INFO[0158] job.PUSH jobs.checkEmailJob 0d1624d3-c8ff-4d26-a9e4-cb33feed4487
INFO[0159] job.RECV jobs.checkEmailJob 0d1624d3-c8ff-4d26-a9e4-cb33feed4487
INFO[0159] job.DONE jobs.checkEmailJob 0d1624d3-c8ff-4d26-a9e4-cb33feed4487 {707ms}

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x50 pc=0x9cac2f]

goroutine 160 [running]:
github.com/spiral/jobs/broker/sqs.unpack(0xc000341100, 0x65746e6f630a3239, 0x3a657079742d746e, 0x746163696c707061, 0x77772d782f6e6f69, 0x752d6d726f662d77, 0x65646f636e656c72)
        C:/Users/.../go/src/github.com/spiral/jobs/broker/sqs/job.go:46 +0x10f
github.com/spiral/jobs/broker/sqs.(*queue).do(0xc000267490, 0xc00000c710, 0xc000183fa0, 0xc000341100, 0x3665316566353863, 0x3837666266626338)
        C:/Users/.../go/src/github.com/spiral/jobs/broker/sqs/queue.go:154 +0x40
github.com/spiral/jobs/broker/sqs.(*queue).serve.func1(0xc000267490, 0xc00000c710, 0xc000183fa0, 0xc000341100)
        C:/Users/.../go/src/github.com/spiral/jobs/broker/sqs/queue.go:116 +0x4d
created by github.com/spiral/jobs/broker/sqs.(*queue).serve
        C:/Users/.../go/src/github.com/spiral/jobs/broker/sqs/queue.go:115 +0xe4

.rr config:

rpc:
  listen: "tcp://localhost:6002"

http:
  address: :8081
  workers.command: "php webroot/index.php rr"

jobs:
  sqs:
    key:      ...
    secret:   ...
    region:   us-east-2

  workers.command: "php webroot/index.php consumer"

  dispatch:
    jobs-*.pipeline: sqs

  pipelines:
    sqs:
      broker: sqs
      queue:  awsbounces
      declare:

  consume: ["sqs"]

^CDEBU[0462] [rpc]: stopped
DEBU[0462] [http]: stopped
INFO[0462] [sqs]: stopping {sqs}
ERRO[0462] [http]: http: Server closed
DEBU[0462] [ephemeral]: stopped
DEBU[0463] [sqs]: stopped {sqs}
DEBU[0463] [sqs]: stopped
DEBU[0463] [jobs]: stopped
panic: close of closed channel

goroutine 1 [running]:
github.com/spiral/jobs/broker/sqs.(*Broker).Stop(0xc000142680)
        C:/Users/.../go/src/github.com/spiral/jobs/broker/sqs/broker.go:99 +0x108
github.com/spiral/roadrunner/service.(*container).Stop(0xc000278420)
        C:/Users/.../go/src/github.com/spiral/roadrunner/service/container.go:205 +0xc2
github.com/spiral/jobs.(*Service).Stop(0xc000199ea0)
        C:/Users/.../go/src/github.com/spiral/jobs/service.go:141 +0x159
github.com/spiral/roadrunner/service.(*container).Stop(0xc00011b0e0)
        C:/Users/.../go/src/github.com/spiral/roadrunner/service/container.go:205 +0xc2
github.com/spiral/roadrunner/service.(*container).Serve(0xc00011b0e0, 0xb8e548, 0xc00026a6c0)
        C:/Users/.../go/src/github.com/spiral/roadrunner/service/container.go:193 +0x2b5
github.com/spiral/roadrunner/cmd/rr/cmd.serveHandler(0xc000134500, 0xc000142840, 0x0, 0x4)
        C:/Users/.../go/src/github.com/spiral/roadrunner/cmd/rr/cmd/serve.go:51 +0x74
github.com/spf13/cobra.(*Command).execute(0xc000134500, 0xc000142780, 0x4, 0x4, 0xc000134500, 0xc000142780)
        C:/Users/.../go/src/github.com/spf13/cobra/command.go:766 +0x2cc
github.com/spf13/cobra.(*Command).ExecuteC(0x10ec0c0, 0xc0001b1f20, 0x80b68a, 0xc00011b0f0)
        C:/Users/.../go/src/github.com/spf13/cobra/command.go:852 +0x2fd
github.com/spf13/cobra.(*Command).Execute(0x10ec0c0, 0x5, 0xc0001426c0)
        C:/Users/.../go/src/github.com/spf13/cobra/command.go:800 +0x2b
github.com/spiral/roadrunner/cmd/rr/cmd.Execute()
        C:/Users/.../go/src/github.com/spiral/roadrunner/cmd/rr/cmd/root.go:64 +0x31
main.main()
        D:/Projects/.../rr/main.go:39 +0x394

spiral-modules / jobs Goto Github PK

jobs's Introduction

RoadRunner: Background PHP workers, Queue brokers

Documentation

Features

License:

jobs's People

Contributors

Stargazers

Watchers

Forkers

jobs's Issues

Recommend Projects

Recommend Topics

Recommend Org