pub async fn worker(
	config: Arc<Config>,
	errors: mpsc::Sender<RuntimeError>,
	events: priority::Receiver<Event, Priority>,
) -> Result<(), CriticalError> {
	let mut jobtasks = LateJoinSet::default();
	let mut jobs = HashMap::<Id, Job>::new();

	while let Some(mut set) = throttle_collect(
		config.clone(),
		events.clone(),
		errors.clone(),
		Instant::now(),
	)
	.await?
	{
		let events: Arc<[Event]> = Arc::from(take(&mut set).into_boxed_slice());

		trace!("preparing action handler");
		let action = Handler::new(events.clone(), jobs.clone());

		debug!("running action handler");
		let action = match config.action_handler.call(action) {
			ActionReturn::Sync(action) => action,
			ActionReturn::Async(action) => Box::into_pin(action).await,
		};

		debug!("take control of new tasks");
		for (id, (job, task)) in action.new {
			trace!(?id, "taking control of new task");
			jobtasks.insert(task);
			jobs.insert(id, job);
		}

		if let Some(manner) = action.quit {
			debug!(?manner, "quitting worker");
			match manner {
				QuitManner::Abort => break,
				QuitManner::Graceful { signal, grace } => {
					debug!(?signal, ?grace, "quitting worker gracefully");
					let mut tasks = LateJoinSet::default();
					for (id, job) in jobs.drain() {
						trace!(?id, "quitting job");
						tasks.spawn(async move {
							job.stop_with_signal(signal, grace);
							job.delete().await;
						});
					}
					// TODO: spawn to process actions, and allow events to come in while
					//       waiting for graceful shutdown, e.g. a second Ctrl-C to hasten
					debug!("waiting for graceful shutdown tasks");
					tasks.join_all().await;
					debug!("waiting for job tasks to end");
					jobtasks.join_all().await;
					break;
				}
			}
		}

		let gc: Vec<Id> = jobs
			.iter()
			.filter_map(|(id, job)| {
				if job.is_dead() {
					trace!(?id, "job is dead, gc'ing");
					Some(*id)
				} else {
					None
				}
			})
			.collect();
		if !gc.is_empty() {
			debug!("garbage collect old tasks");
			for id in gc {
				jobs.remove(&id);
			}
		}

		debug!("action handler finished");
	}

	debug!("action worker finished");
	Ok(())
}
The main worker of a Watchexec process.

This is the main loop of the process. It receives events from the event channel, filters them,
debounces them, obtains the desired outcome of an actioned event, calls the appropriate handlers
and schedules processes as needed.
crates/lib/src/action/worker.rs