amqp long running jobs - graceful shutdown

[dr-diesel]

Hello,
I'm testing RR consuming queue from RabbitMQ and I face problem with long running tasks shutdown scenario.

When I shutdown RR with SIGINT/SIGTERM , the pipeline closes and RR tries to close workers. My test PHP worker just simulates 20s sleep as long running task and then tries to ack() the message. Unfortunately the amqp plugin is already closed and the Message can't be acked.

ERROR jobs acknowledge failed, job might be lost {"ID": "0bbc5456-f0fc-4b8c-84a2-cb849cf95ff9", "error": "jobs_handle_response: failed to acknowledge the JOB, the pipeline is probably stopped", "error": "failed to acknowledge the JOB, the pipeline is probably stopped"}

The jobs.num_pollers is fixed number of threads in RR config, so I tested 10 pollers with 2 workers - to be evt. able to increase number of workers to 10 if the queue rises and reduce e.g. back to two workers if it's empty - this works, tested. But when pollers>workers, the poller tries to deliver the message to worker even if the shutdown is in process. In case there is no available worker it tries to nack() the message and this is also reported as closed pipeline.

2024-06-07T15:41:58+0000        ERROR   server          no free workers in the pool, wait timeout exceed        {"reason": "no free workers", "internal_event_name": "EventNoFreeWorkers", "error": "worker_watcher_get_free_worker: NoFreeWorkers:\n\tcontext deadline exceeded"}
2024-06-07T15:41:58+0000        ERROR   jobs            job processed with errors       {"error": "static_pool_exec: NoFreeWorkers:\n\tstatic_pool_exec:\n\tworker_watcher_get_free_worker:\n\tcontext deadline exceeded", "ID": "bf4b7d6c-7379-4cd1-840e-0fcbccf3f570", "start": "2024-06-07T15:40:58+0000", "elapsed": 60009}
2024-06-07T15:41:58+0000        ERROR   jobs            negatively acknowledge failed   {"ID": "bf4b7d6c-7379-4cd1-840e-0fcbccf3f570", "error": "failed to acknowledge the JOB, the pipeline is probably stopped"}
2024-06-07T15:41:58+0000        DEBUG   jobs            ------> job poller was stopped <------

When I set jobs.num_pollers to 1 or 2 , I can't rise number of parallel jobs rising the workers via RPC api. The num_pollers goroutines are waiting for workers, next workers are only waiting.

How am I able to gracefully shutdown RR with long running parallel tasks from amqp - with ability of PHP worker to ack() last message and receive message to shutdown ? (+ in case the num_pollers > workers, it receives further few waiting messages before shutdown message).
It does not necessarily mean to be SIGINT/SIGTERM . I wondered if in-memory queue with higher priority could receive message to shutdown the workers, but that would not shutdown RR.

Thank you for interesting project. I'm former PHPdev, but tried to inspect Go src in VS code with amqp+jobs plugins and RR sdk, built and tried to debug shutdown procedure.

Regards,
Jan

[rustatian]

Hey and welcome to RR @dr-diesel 👋

num_pollers option is just the number of goroutines which in parallel trying to obtain a job from the internal priorities queue (in the jobs plugin) and push them to the workers. If your workers are superfast, it makes sense to increase the number of pollers.

You also correctly found the problem with the graceful shutdown. RR doesn't know about JOBS which correctly are in the workers. And the case is, that while it in the worker, you may stop the pipeline. Due to the plugin architecture of RR, the JOBS plugin doesn't know about the drivers (aka amqp, kafka, etc). It only knows, that it should call some method to acknowledge this JOB. But the pipeline, as you know, was already stopped. Thus you see this error. Nothing bad is going to happen with the message (lost, for example), only an error message from RR to point your attention.

EDIT: Here you may locate a corresponding issue: #1382

[dr-diesel]

Thank you for very quick reply @rustatian 🙏🏻

num_pollers is relevant for long running tasks, when I start 10 workers and let num_pollers=2, only 2 workers run, others stay ready because only 2 goroutines wait for response from workers - tested. When num_pollers=10 and I added workers dynamically to 10 still 10 workers run in parallel, but 11th stays ready.

I don't close amqp via RPC. It is closed from stop signal. The setup is simple - amqp driver, pipeline consuming queue, php script with 20s sleep. Ctrl C the RR and the script can't ack() the message and waiting pollers can't nack() their waiting jobs.

When I put breakpoint on amqp.Driver.Stop , it's called from here

func (p *Plugin) Stop(ctx context.Context) error {
	// Broadcast stop signal to all pollers
	close(p.stopCh)

	sema := semaphore.NewWeighted(int64(p.cfg.CfgOptions.Parallelism))
	// range over all consumers and call stop
	p.consumers.Range(func(key, value any) bool {
		// acquire semaphore, but if RR canceled the context, we should stop
		errA := sema.Acquire(ctx, 1)
		if errA != nil {
			return false
		}

		go func() {
			consumer := value.(jobsApi.Driver)
			ctxT, cancel := context.WithTimeout(ctx, time.Second*time.Duration(p.cfg.Timeout))
>>>			err := consumer.Stop(ctxT)

The driver is put into consumers in jobs.Plugin.Declare()

So during jobs.Plugin.Stop() the driver is stopped and the ack(), nack() become unavailable. Couldn't the amqp driver stop only consuming in this stage and shutdown later after pollers and workers are done ?

[rustatian]

num_pollers is relevant for long running tasks

Yes, because while consuming the items from the priority queue, goroutine will wait for the response from the PHP worker.

I don't close amqp via RPC

I didn't say about RPC. But this doesn't matter actually, since the mechanism is the same.

waiting pollers can't nack() their waiting jobs.

That's actually doesn't matter from the broker POV, since the message would be simply re-delivered.

Couldn't the amqp driver stop only consuming in this stage and shutdown later after pollers and workers are done ?

Actually can, during the grace timeout period, but this is not implemented.

[rustatian]

@dr-diesel If you have additional questions, feel free to continue this discussion or join our discord server for real-time communication.