kainxspirits / laravel-pubsub-queue Goto Github PK

Currently the driver shows unexpected behaviour when using --stop-when-empty.

It suddenly quits, while jobs are left on the queue. In my opinion this is caused by

PubSubQueue.php:139
$messages = $subscription->pull([ 'returnImmediately' => true, 'maxMessages' => 1, ]);

See:
https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions/pull
See 'returnImmediately'

I was wondering if you guys would be oke if i submit a PR to change this to false.

OR

Should it be configurable?, i couldn't imagine a reason why you would put it on ``true```
But i guess you guys might have a valid reason for it.

You might think... wtf even use --stop-on-empty ...

It is very useful when you use a laravel or lumen job as Kubernetes Cronjob
especially with this pub-sub driver, it doesn't need any side car containers.
and can run a lean job or cronjob

No support for Laravel 5.6?
Any other project with support to L5.6?

After a job has failed, I run the following:
php artisan queue:retry all

Which results in the following error message in the queue logs:

[2019-04-02 22:02:49][] Processing: 
[2019-04-02 22:02:49][] Failed:    

After further inspection, it would seem we are using base64_encode() too soon - pushRaw() expects $payload to be a json encoded string, but not base64 encoded. I will submit a pull request.

Hi
I am facing an issue using the Queue bulk() function to publish multiple jobs on PubSub.

I am using the Bus Batch() to batch multiple jobs simultaneously. However, I get the following errors when running on GCP:

The value for message_count is too small. You passed 0 in the request, but the minimum value is 1

Any idea if the bulk has been tested before?

Thank you
Bill

Currently, the subscriber name is hard-coded (as above). This results in only allowing one laravel-pubsub-queue subscriber per Google Cloud project.

Hey @kainxspirits

Thanks for the great package. I've got everything onto the Pub/Sub topic as messages but not sure what I need to really do to the "process" the queue.

I was thinking of creating an endpoint for the jobs to push too i.e example.com/jobs which basically decodes the payload and fire the same Job again internally as a sync on a Cloud Run instance of the app.

Just wondered if this is how you intended the package to be used or if you have a different approach personally?

p.s I am playing with moving my Laravel apps over to serverless, so still picking bits up here and there,

When a job is pushed onto the queue, the following error is returned:

{
  "error": {
    "code": 400,
    "message": "Invalid value at 'messages[0]' (Map), Cannot bind a list to map for field 'attributes'.",
    "status": "INVALID_ARGUMENT",
    "details": [
      {
        "@type": "type.googleapis.com/google.rpc.BadRequest",
        "fieldViolations": [
          {
            "field": "messages[0]",
            "description": "Invalid value at 'messages[0]' (Map), Cannot bind a list to map for field 'attributes'."
          }
        ]
      }
    ]
  }
}

This is likely due to a bug in the underlying library, which probably sees the empty array passed to pushRaw, and fails to convert it to a map for the request.

However, this can be worked around by leaving out the attributes if the array is empty. pushRaw would then read:

$topic = $this->getTopic($queue, true);

$this->subscribeToTopic($topic);

$publish = ['data' => $payload];

if (!empty($options)) {
    $publish['attributes'] = $options;
}

return $topic->publish($publish);

Only a single subscription is created for all the topics which cause issues with multiple queues.

The package subscribes topic to a subscription due to another queue(topic) is unable to subscribe to it.

A possible solution can be that we can create topics and subscription with that same name so they can be linked as I think Pubsub only support one topic per subscription. Please correct me if I am wrong anywhere.

As explained on this PR : #18

PubSub message documentation : https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage

Hey there. While using this package, I noticed my payload that running through google pub/sub is encoded twice.
The problem is in the PubSubQueue class in pushRaw function

$publish = ['data' => base64_encode($payload)];

On this line we are encoding our payload, and then in

$topic->publish($publish);

We are actually executing code from Google/cloud-pubsub library, which is doing exactly the same, so the payload is encoded twice at the stage of sending it to the pub/sub.

The same if for pulling messages from subscriptions, Google package doing the base64_decode by itself.

Just wondering what was the initial purpose of encoding payload inside this package ?

Support for Laravel 6 was added 3 months ago, but no new release has been tagged since. Could we get a tagged release of this asap? https://github.com/decahedronio/laravel-app-events depends on this package, and we'd like to upgrade that to Laravel 6.

Thanks for working on this package. I was testing it though and noticed there is a performance issue with it. Pull and ack together can take anywhere from 100ms to up to 1 second. This is not related to networking but by design on pubsub infrastructure. Pubsub is not intended to pull and ack 1 message at a time, the pull feature is to be used with batches of messages, so for example pull 500 and acknowledge 500 messages.

This package can be used if you dont have heavy working tasks, but whenever you need to process thousands of jobs in a minute your bottleneck for sure will be the time it takes to pull and ack to pubsub.

Additionally returnImmediately is also deprecated as is pointed out here: https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions/pull#request-body

Warning: setting this field to true is discouraged because it adversely impacts the performance of subscriptions.pull operations. We recommend that users do not set this field.

This thread is also useful to read in regard to pulling and acknowledging 1 message at a time: googleapis/google-cloud-php#939 (comment)

I would recommend against using pubsub in general with PHP if you care about scaling. Another reason is regarding publishing messages. The first time you publish to pubsub it takes anywhere from 0.5-1s. Since php always starts a new request, your web servers will always have go through first time publish, hanging your app for up to 1s just to publish a message, this is unacceptable. Using redis and laravel queue dispatch, it takes max a few milliseconds to set or get a message.

Using other languages such as go, java, nodejs, etc they have a server running in memory that starts up once and is reused for every further request. Even if first publish takes time, all further publish events will be very fast.

Pubsub seems to have some connection issues on first requests, I guess its related to authentication.

Hi @kainxspirits ,

First of all, thanks a lot for sharing this Pubsub driver for Laravel Queues. It definitely saved me a decent amount of time since I didn't have to implement it myself.

I was looking at how the delayed messages are handled. I found here that when available_at attribute is set in future for a message, we just return immediately without acknowledging it. As a result, message gets picked again by worker after the ack deadline set on the pubsub subscription is passed. In my case, I have set the ack deadline on my pubsub subscription to the max allowed value i.e. 600s. Now I want to be able to execute messages with any custom delay value within this range like 10s, 1m, 2m etc. Current implementation doesn't allow this.

However, it is possible to do this as following:

1. When available_at > time() , get the remaining seconds like remaining_time = available_at - time()
2. Modify ack deadline of this specific message to be remaining_time.
3. Now as soon as remaining_time has passed, Pubsub subscription will make this message available again, overriding the ack deadline set on the subscription itself.

I am going to submit a PR with this implementation. Please let me know if this sounds like a valid general use case and if you would like to merge it in?

I am using this component in a project which is being deployed on the Google Cloud Run. For the pub/sub service account key, I am using google Secret manager rather than putting the JSON file in my codebase which would expose my secret. The Google Cloud pub/sub client library allows the actual JSON to be submitted after decoding it to an array. However, this fails in the current implementation on line 61 of Src/Connectors/PubSubConnector.php asit expects the $item to be string whereas in case of keyFile it is an Array.
If we pass the key.json content as json string without decoding, then it fails when this string is passed over to the Google Client Library where it expects it to be an array. Is anyone looking at this repo actively or I take a fork and create another composer for this. Please help & suggest.

Hello,

How could I run the pop() method on the queue?

I call an Endpoint on my app via the push subscription whenever a new job is pushed to the queue.

This endpoint receives the job and runs it immediately.

Shall I instead run queue:work command?

php artisan queue:work --queue=pubsub --once

Thank you
Bill

It looks live the driver does not honor job reservations like the redis or database driver.

At this point when you would run multiple instances of the queue worker, for example 10 workers. And PubSub having an acknowledge deadline of 10 seconds.
It is very likely that multiple workers run the same job.
(I was also able to actually reproduce this, multiple workers working on the same job ID)

An other flaw is the acknowledge deadline, if you have job that might take a couple of hours, it will probably exceed the PubSub acknowledge deadline;

Any thoughts how this is desired to be implemented?, i am happy to help out to write something for this.

I think building a reservation system like redis and the database driver would be an over complicated solution, while you cannot fetch a reserved job back, and delete it once it's released back to the queue, or should be deleted forever.

From my opinion it would be easier to: https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions/modifyAckDeadline

The deadline should be synchronised, by the timeout setting of the queue worker if possible.

Assume that would solve both issues...

Let me know what u guys think

Thnx
Danny

At the moment it's not possible to take advantage of a builtin retry mechanism of a pubsub subscription (https://cloud.google.com/pubsub/docs/handling-failures) as messages are acknowledged as soon as they are successfully pulled (not processed) from a subscription.

Laravel retry mechanism (push back to topic) is BADLY designed imho at tleast when it comes to pubsub.

1. requires an app to have publish permissions - VERY BAD
2. as many subscribers can exist for a single topic pushing the same message again will cause other subscribers process the same message again - BAD

is there any workaround for a large message size ?
i read max message size for pubsub is 10MB i need to dispatch job with large payload

in other cloud like aws sqs there is workaround like using this library https://github.com/SimpleSoftwareIO/simple-sqs-extended-client so we can dispatch large payload to the queue, the logic behind is to put the large payload to s3

There is a newer version for Cloud PubSub Client.

google/cloud-pubsub v1.51.0 => v2.1.1 Cloud PubSub Client for PHP

As explained on #35 (comment), we can reduce the number of administrative requests by onlyu re-creating subscriptions and topics when needed.

https://cloud.google.com/pubsub/docs/troubleshooting#admin

When i sending job queue using multiple queue, it does create new topic with no subscriber and when listening queue using command php artisan queue:work --queue=queue-first the is no message pulled in to our app.

i think its because this package only can listen one subscriber that defined in queue config

this is my code to running multiple job queue

firstJob::dispatch()->onQueue('queue-first');
secondJob::dispatch()->onQueue('queue-second');
thirdJob::dispatch()->onQueue('queue-third');

it will be great if this package can handle multiple topic and subscriber too.

Thanks

I'm note sure if it's possible with google pubsub, but we could try to implement the new ClearableQueue contract : https://github.com/illuminate/contracts/blob/8.x/Queue/ClearableQueue.php.

I did the following setup in Lumen 6 application

1. composer require kainxspirits/laravel-pubsub-queue

2. $app->register(Kainxspirits\PubSubQueue\PubSubQueueServiceProvider::class);

3. Added queue.php in config folder with the following content

 <?php

use Illuminate\Support\Facades\File;

return [
'pubsub' => [
    'driver' => 'pubsub',
    'queue' => env('PUBSUB_QUEUE', 'iprocure'),
    'project_id' => env('PUBSUB_PROJECT_ID', 'iprocure-edw'),
    'retries' => 3,
    'request_timeout' => 60,
    'subscriber' => 'mytestsubscriber',
    'key_file' => File::get(storage_path('app/gcp-pubsub.json'))
]];

4. Added $app->configure('queue'); in bootstrap/app.php

Then in my controller constructor i did the following

use Kainxspirits\PubSubQueue\PubSubQueue;

class ExampleController extends Controller
{
     public function __construct(PubSubQueue $queue)
     {
         $this->queue = $queue;
     }

     public function testPubsub(){
    
     } 
 }

When i send request for testPubsub() i see the following error

(1/1) GoogleException
No project ID was provided, and we were unable to detect a default project ID.