20416384001

Committed 21 Dec 2025 09:57PM UTC coverage: 51.395% (-0.003%) from 51.398%

Build # 20416384001

Build Type

push

github

Committed by

web-flow

Commit Message

fix(task): move task execution inside spawn to prevent blocking (#328)

Moves the invocation of the task logic inside the spawn block. This
ensures that any synchronous setup done by the task does not block the
caller of `TaskHandler::spawn`.

Also renames `Task::start` to `Task::run` to better reflect its
behavior.

Signed-off-by: Jean Mertz <git@jeanmertz.com>

Coverage Stats

0 of 2 new or added lines in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

8806 of 17134 relevant lines covered (51.39%)

132.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/crates/jp_task/src/handler.rs

use std::{error::Error, time::Duration};

use jp_workspace::Workspace;
use tokio::task::{JoinError, JoinSet};
use tokio_util::sync::CancellationToken;
use tracing::{debug, trace, warn};

use crate::Task;

#[derive(Debug, Default)]
pub struct TaskHandler {
    tasks: JoinSet<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>>,
    cancel_token: CancellationToken,
}

impl TaskHandler {
    pub fn spawn(&mut self, task: impl Task) {
        let name = task.name();
        debug!(name, "Spawning task.");
        let token = self.cancel_token.child_token();
        self.tasks.spawn(async move {
            let mut task = Box::new(task).run(token);
            let now = tokio::time::Instant::now();
            loop {
                jp_macro::select!(
                    biased,
                    tokio::time::sleep(Duration::from_millis(500)),
                    |_wake| {
                        trace!(name, elapsed_ms = %now.elapsed().as_millis(), "Task running...");
                    },
                    &mut task,
                    |v| {
                        debug!(name, elapsed_ms = %now.elapsed().as_millis(), "Task completed.");
                        return v;
                    }
                );
            }
        });
    }

    pub async fn sync(
        &mut self,
        workspace: &mut Workspace,
        timeout: Duration,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        if self.tasks.is_empty() {
            return Ok(());
        }

        let mut tasks: Vec<Box<dyn Task>> = Vec::new();
        self.wait_for_tasks(timeout, &mut tasks, false).await;

        // Force long-running tasks to stop if they aren't responding to the
        // cancellation signal.
        self.wait_for_tasks(Duration::from_secs(2), &mut tasks, true)
            .await;

        for task in tasks {
            if let Err(error) = task.sync(workspace).await {
                tracing::error!(%error, "Error syncing background task.");
            }
        }

        Ok(())
    }

    async fn wait_for_tasks(
        &mut self,
        timeout: Duration,
        tasks: &mut Vec<Box<dyn Task>>,
        shutdown: bool,
    ) {
        let timeout = tokio::time::sleep(timeout);
        tokio::pin!(timeout);

        loop {
            jp_macro::select!(
                biased,
                // Ask long-running tasks to stop.
                &mut timeout,
                |_wake| {
                    if shutdown {
                        warn!("Tasks did not respond to cancellation signal. Forcing shutdown.");
                        self.tasks.shutdown().await;
                    } else {
                        warn!(
                            "Task finalization timed out. Signalling cancellation to remaining \
                             tasks."
                        );
                        self.cancel_token.cancel();
                    }
                    break;
                },
                self.tasks.join_next(),
                |task| {
                    match task {
                        Some(task) => handle_task_completion(task, tasks),
                        None => break,
                    }
                },
            );
        }
    }
}

#[expect(clippy::type_complexity)]
fn handle_task_completion(
    result: Result<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>, JoinError>,
    tasks: &mut Vec<Box<dyn Task>>,
) {
    match result {
        Ok(Ok(task)) => tasks.push(task),
        Ok(Err(error)) => tracing::error!(%error, "Background task failed."),
        Err(error) => tracing::error!(%error, "Error waiting for background task to complete."),
    }
}

1	use std::{error::Error, time::Duration};
2
3	use jp_workspace::Workspace;
4	use tokio::task::{JoinError, JoinSet};
5	use tokio_util::sync::CancellationToken;
6	use tracing::{debug, trace, warn};
7
8	use crate::Task;
9
10	#[derive(Debug, Default)]
11	pub struct TaskHandler {
12	tasks: JoinSet<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>>,
13	cancel_token: CancellationToken,
14	}
15
16	impl TaskHandler {
17	pub fn spawn(&mut self, task: impl Task) {	×
18	let name = task.name();	×
19	debug!(name, "Spawning task.");	×
NEW 20	let token = self.cancel_token.child_token();	×
21	self.tasks.spawn(async move {	×
NEW 22	let mut task = Box::new(task).run(token);	×
UNCOV 23	let now = tokio::time::Instant::now();	×
24	loop {
25	jp_macro::select!(	×
26	biased,
27	tokio::time::sleep(Duration::from_millis(500)),	×
28	\|_wake\| {
29	trace!(name, elapsed_ms = %now.elapsed().as_millis(), "Task running...");	×
30	},
31	&mut task,	×
32	\|v\| {
33	debug!(name, elapsed_ms = %now.elapsed().as_millis(), "Task completed.");	×
34	return v;	×
35	}
36	);
37	}
38	});	×
39	}	×
40
41	pub async fn sync(	×
42	&mut self,	×
43	workspace: &mut Workspace,	×
44	timeout: Duration,	×
45	) -> Result<(), Box<dyn Error + Send + Sync>> {	×
46	if self.tasks.is_empty() {	×
47	return Ok(());	×
48	}	×
49
50	let mut tasks: Vec<Box<dyn Task>> = Vec::new();	×
51	self.wait_for_tasks(timeout, &mut tasks, false).await;	×
52
53	// Force long-running tasks to stop if they aren't responding to the
54	// cancellation signal.
55	self.wait_for_tasks(Duration::from_secs(2), &mut tasks, true)	×
56	.await;	×
57
58	for task in tasks {	×
59	if let Err(error) = task.sync(workspace).await {	×
60	tracing::error!(%error, "Error syncing background task.");	×
61	}	×
62	}
63
64	Ok(())	×
65	}	×
66
67	async fn wait_for_tasks(	×
68	&mut self,	×
69	timeout: Duration,	×
70	tasks: &mut Vec<Box<dyn Task>>,	×
71	shutdown: bool,	×
72	) {	×
73	let timeout = tokio::time::sleep(timeout);	×
74	tokio::pin!(timeout);	×
75
76	loop {
77	jp_macro::select!(	×
78	biased,
79	// Ask long-running tasks to stop.
80	&mut timeout,	×
81	\|_wake\| {
82	if shutdown {	×
83	warn!("Tasks did not respond to cancellation signal. Forcing shutdown.");	×
84	self.tasks.shutdown().await;	×
85	} else {
86	warn!(	×
87	"Task finalization timed out. Signalling cancellation to remaining \	×
88	tasks."	×
89	);
90	self.cancel_token.cancel();	×
91	}
92	break;	×
93	},
94	self.tasks.join_next(),	×
95	\|task\| {
96	match task {	×
97	Some(task) => handle_task_completion(task, tasks),	×
98	None => break,	×
99	}
100	},
101	);
102	}
103	}	×
104	}
105
106	#[expect(clippy::type_complexity)]
107	fn handle_task_completion(	×
108	result: Result<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>, JoinError>,	×
109	tasks: &mut Vec<Box<dyn Task>>,	×
110	) {	×
111	match result {	×
112	Ok(Ok(task)) => tasks.push(task),	×
113	Ok(Err(error)) => tracing::error!(%error, "Background task failed."),	×
114	Err(error) => tracing::error!(%error, "Error waiting for background task to complete."),	×
115	}
116	}	×

dcdpr / jp / 20416384001

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous