Rethink graceful shutdown

[kirillsalykin]

Currently graceful shutdown communicated to all workers via NOTIFY, which works, but it feels not reliable, if worker misses the message - it wont shutdown gracefully.

I believe it is more reliable to communicate shutdown to every worker/job directly, i was thinking about this:

 let job = Job::builder()
        .step(|_ctx, _input| async move {
            let sleep_duration = std::time::Duration::from_secs(5);

            tracing::info!(?sleep_duration, "Hello from a long-running task");

            // Artificial delay to simulate a long-running job.
            tokio::time::sleep(sleep_duration).await;

            To::done()
        })
        .name(QUEUE_NAME)
        .pool(pool.clone())
        **.with_gracefull_shutdown(shutdown_signal()), <timeout>)**
        .build()
        .await?;

(Inspired by https://github.com/tokio-rs/axum/blob/280d16a61059f57230819a79b15aa12a263e8cca/examples/graceful-shutdown/src/main.rs#L50)

timeout will force worker to stop if it didnt gracefully stopped after signal received.
Worker wont accept any new tasks after signal recieved.

This might be a little more boilerplate compared to the current solution, cause you need to set this on every job, but it gives you possibility to configure each job independently.

What do you think?

Thanks.

[maxcountryman]

I don't think following axum is a great fit: one obvious reason is because a common use case will be to use underway with axum. To support this use case, what you really want is to be able to register handlers (graceful_shutdown for example) for whatever signals you're intercepting and then pass that to axum (see the graceful shutdown example). It's pretty awkward to have to register a shutdown handler for each job and it wouldn't play nice with other frameworks (you'd much prefer to have axum trigger the shutdown signal handler since that's also how you'd signal other systems to shutdown).

That said, I think it probably makes sense to add some kind of stop method to the job struct. This would work by flipping a bit on the job and propagating that through to the worker and scheduler. So if you really wanted, you could call stop as a failsafe if you don't trust that the Postgres notification will be received.

[kirillsalykin]

It would be nice to be able to re-use the same signal in both axum and underway, shutdown_signal()

(you'd much prefer to have axum trigger the shutdown signal handler since that's also how you'd signal other systems to shutdown)

I wouldnt agree here (but please correct me if I misunderstand how things work), when I start my system i have axum subsystem and underway subsystem, once the system received shutdown signal - it should be propagated to both axum and underway, so I dont expect axum to propogate anything.

It's pretty awkward to have to register a shutdown handler for each job

I would agree here, but there is no worker pool which creates workers and can be single point of configuration.

[maxcountryman]

Axum takes some method that it'll manage to handle the graceful shutdown.

Since Axum already does this it doesn't make sense to spawn this twice in my view.

In cases where you aren't using Axum, you can spawn it directly yourself like we do in the example.

[kirillsalykin]

My bad, I was thinking of using tokio::sync::broadcast to update both axum and underway about shutting down.

Anyway, having stop on each job would probably work for me.

[maxcountryman]

Let me think more about this.

As a first step, I can look at implementing stop because I think that'll involve a refactor that uses Notify (tokio, not Postgres). With that, the shutdown notification could come from different sources (not just a Postgres NOTIFY).

[kirillsalykin]

thank you!