Fix the LifecycleWorker being uncooperative (#1396)

## Summary

This PR ensures that the `LifecycleWorker` yields at least once to the Tokio scheduler in between each batch of 100 objects.

## Problem being solved

I'm administrating a Garage cluster which has been experiencing timeouts on all endpoints while the lifecycle worker is running at midnight UTC : `Ping timeout` error messages and even requests eventually failing due to `Could not reach quorum ...`.

I have found that this happens while the lifecycle worker is working on a big bucket (containing millions of objects) with a lifecycle rule that applies to very few objects.
The `process_object()` function does not hit any `await`:
- `last_bucket` is always the same, so the `bucket_table` is not read asynchronously
- no transaction is made on the `object_table` because my lifecycle rule (almost) never applies to any object

The first commit in this PR adds an executable which reproduces the problem that I've been experiencing in a self-contained way : the lifecycle worker starves the Tokio scheduler so much that no other task is able to run (or very rarely).
To run it : `cargo run -p garage_model --bin lifecycle-starvation-test`.
This commit can be dropped post-review, as it's only useful to demonstrate the starvation.

The error messages completely stopped after adding the extra yield to the nodes of my cluster.
The duration of the lifecycle worker task does not appear to have changed at all from what I can see (looking at the timestamps produced either by the self-contained binary or by each of my nodes with the `Lifecycle worker finished` message).

## Note

An other potential fix would have been to force the `WorkerProcessor` to yield before re-enqueuing a busy task, but this would have affected all Garage workers even though it's only the `LifecycleWorker` being uncooperative.

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/1396
Reviewed-by: Alex <lx@deuxfleurs.fr>
Co-authored-by: Gauthier Zirnhelt <gauthier.zirnhelt@insimo.fr>
Co-committed-by: Gauthier Zirnhelt <gauthier.zirnhelt@insimo.fr>

src/util/background/worker.rs

@@ -103,32 +103,39 @@ impl WorkerProcessor {
 						trace!("{} (TID {}): {:?}", worker.worker.name(), worker.task_id, worker.state);
 
 						// Save worker info
-						let mut wi = self.worker_info.lock().unwrap();
+						{
-						match wi.get_mut(&worker.task_id) {
+							let mut wi = self.worker_info.lock().unwrap();
-							Some(i) => {
+							match wi.get_mut(&worker.task_id) {
-								i.state = worker.state;
+								Some(i) => {
-								i.status = worker.worker.status();
+									i.state = worker.state;
-								i.errors = worker.errors;
+									i.status = worker.worker.status();
-								i.consecutive_errors = worker.consecutive_errors;
+									i.errors = worker.errors;
-								if worker.last_error.is_some() {
+									i.consecutive_errors = worker.consecutive_errors;
-									i.last_error = worker.last_error.take();
+									if worker.last_error.is_some() {
+										i.last_error = worker.last_error.take();
+									}
+								}
+								None => {
+									wi.insert(worker.task_id, WorkerInfo {
+										name: worker.worker.name(),
+										state: worker.state,
+										status: worker.worker.status(),
+										errors: worker.errors,
+										consecutive_errors: worker.consecutive_errors,
+										last_error: worker.last_error.take(),
+									});
 								}
-							}
-							None => {
-								wi.insert(worker.task_id, WorkerInfo {
-									name: worker.worker.name(),
-									state: worker.state,
-									status: worker.worker.status(),
-									errors: worker.errors,
-									consecutive_errors: worker.consecutive_errors,
-									last_error: worker.last_error.take(),
-								});
 							}
 						}
 
 						if worker.state == WorkerState::Done {
 							info!("Worker {} (TID {}) exited", worker.worker.name(), worker.task_id);
 						} else {
+							// Yield to the Tokio scheduler between consecutive Busy steps so
+							// that a worker which never suspends on its own cannot starve other tasks.
+							if worker.state == WorkerState::Busy {
+								tokio::task::yield_now().await;
+							}
 							workers.push(async move {
 								worker.step().await;
 								worker