estib: napify long running process

estib-vega · estib-vega · commit fb7f075a4958 · 2026-03-12T16:28:18.000+01:00
Building a POC on top of byron’s POC:
- Export a function from Rust to Node that takes in a duration in ms and
a callback.
- The function returns immediately, doesn’t block.
- The callback get’s event payloads from the Rust side.
- A normal abort signal can be used to abort the process at anytime.
diff --git a/crates/but-api/src/lib.rs b/crates/but-api/src/lib.rs
@@ -51,6 +51,30 @@ pub mod poc {
         time::{Duration, Instant},
     };
 
+    #[cfg(feature = "napi")]
+    use std::{
+        collections::HashMap,
+        sync::{Mutex, OnceLock, atomic::AtomicU32},
+    };
+
+    #[cfg(feature = "napi")]
+    use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode};
+
+    #[cfg(feature = "napi")]
+    static NEXT_TASK_ID: AtomicU32 = AtomicU32::new(1);
+    #[cfg(feature = "napi")]
+    static TASK_INTERRUPTS: OnceLock<Mutex<HashMap<u32, Arc<AtomicBool>>>> = OnceLock::new();
+
+    #[cfg(feature = "napi")]
+    fn task_interrupts() -> &'static Mutex<HashMap<u32, Arc<AtomicBool>>> {
+        TASK_INTERRUPTS.get_or_init(|| Mutex::new(HashMap::new()))
+    }
+
+    #[cfg(feature = "napi")]
+    fn emit_event(callback: &ThreadsafeFunction<LongRunningEvent>, event: LongRunningEvent) {
+        let _ = callback.call(Ok(event), ThreadsafeFunctionCallMode::NonBlocking);
+    }
+
     /// Either the actual data that is more and more complete, or increments that can be merged
     /// into the actual data by the receiver.
     /// Sending all data whenever it changes is probably better.
@@ -105,6 +129,143 @@ pub mod poc {
         rx
     }
 
+    /// Kinds of events emitted to JavaScript while a long-running task executes.
+    #[cfg(feature = "napi")]
+    #[napi_derive::napi(string_enum)]
+    pub enum LongRunningEventKind {
+        /// The task produced an intermediate progress update.
+        Progress,
+        /// The task finished successfully.
+        Done,
+        /// The task stopped because interruption was requested.
+        Cancelled,
+        /// The task failed with an error.
+        Error,
+    }
+
+    /// Event payload for Node callbacks.
+    #[cfg(feature = "napi")]
+    #[napi_derive::napi(object)]
+    pub struct LongRunningEvent {
+        /// The id of the task that emitted this event.
+        pub task_id: u32,
+        /// The event category.
+        pub kind: LongRunningEventKind,
+        /// The latest completed step if available.
+        pub step: Option<u32>,
+        /// An optional error message for failed tasks.
+        pub message: Option<String>,
+    }
+
+    /// Start a long-running task and stream progress to `callback` via a ThreadsafeFunction.
+    ///
+    /// Returns the task id, which can be used to interrupt processing with
+    /// [`long_running_cancel_tsfn()`].
+    #[cfg(feature = "napi")]
+    #[napi_derive::napi(js_name = "longRunningStartTsfn")]
+    pub fn long_running_start_tsfn(
+        duration_ms: u32,
+        callback: ThreadsafeFunction<LongRunningEvent>,
+    ) -> napi::Result<u32> {
+        let task_id = NEXT_TASK_ID.fetch_add(1, Ordering::Relaxed);
+        let should_interrupt = Arc::new(AtomicBool::new(false));
+
+        let mut tasks = task_interrupts().lock().map_err(|_| {
+            napi::Error::new(
+                napi::Status::GenericFailure,
+                "task registry is poisoned".to_string(),
+            )
+        })?;
+        tasks.insert(task_id, should_interrupt.clone());
+        drop(tasks);
+
+        let rx = long_running_non_blocking_thread(
+            Duration::from_millis(u64::from(duration_ms)),
+            gix::progress::Discard,
+            should_interrupt.clone(),
+        );
+
+        thread::spawn(move || {
+            let mut last_step = None;
+            let mut failed = false;
+
+            for result in rx {
+                match result {
+                    Ok(data) => {
+                        let step = u32::try_from(data.0).unwrap_or(u32::MAX);
+                        last_step = Some(step);
+                        emit_event(
+                            &callback,
+                            LongRunningEvent {
+                                task_id,
+                                kind: LongRunningEventKind::Progress,
+                                step: Some(step),
+                                message: None,
+                            },
+                        );
+                    }
+                    Err(err) => {
+                        failed = true;
+                        emit_event(
+                            &callback,
+                            LongRunningEvent {
+                                task_id,
+                                kind: LongRunningEventKind::Error,
+                                step: last_step,
+                                message: Some(format!("{err:#}")),
+                            },
+                        );
+                        break;
+                    }
+                }
+            }
+
+            if !failed {
+                let kind = if should_interrupt.load(Ordering::Relaxed) {
+                    LongRunningEventKind::Cancelled
+                } else {
+                    LongRunningEventKind::Done
+                };
+                emit_event(
+                    &callback,
+                    LongRunningEvent {
+                        task_id,
+                        kind,
+                        step: last_step,
+                        message: None,
+                    },
+                );
+            }
+
+            if let Ok(mut tasks) = task_interrupts().lock() {
+                tasks.remove(&task_id);
+            }
+        });
+
+        Ok(task_id)
+    }
+
+    /// Interrupt a task started with [`long_running_start_tsfn()`].
+    ///
+    /// Returns `true` if interruption was requested successfully.
+    #[cfg(feature = "napi")]
+    #[napi_derive::napi(js_name = "longRunningCancelTsfn")]
+    pub fn long_running_cancel_tsfn(task_id: u32) -> napi::Result<bool> {
+        let tasks = task_interrupts().lock().map_err(|_| {
+            napi::Error::new(
+                napi::Status::GenericFailure,
+                "task registry is poisoned".to_string(),
+            )
+        })?;
+
+        if let Some(flag) = tasks.get(&task_id) {
+            flag.store(true, Ordering::Relaxed);
+            return Ok(true);
+        }
+
+        Ok(false)
+    }
+
     /// Like [`long_running_non_blocking_scoped_thread()`], but uses a regular thread and an owned
     /// cancellation flag so the task can outlive the current stack frame.
     pub fn long_running_non_blocking_thread(
diff --git a/packages/but-sdk/src/generated/index.d.ts b/packages/but-sdk/src/generated/index.d.ts
@@ -81,6 +81,45 @@ export declare function listBranchesNapi(projectId: string, filter: BranchListin
 
 export declare function listProjectsNapi(openedProjects: Array<ProjectHandleOrLegacyProjectId>): Promise<Array<ProjectForFrontend>>
 
+/**
+ * Interrupt a task started with [`long_running_start_tsfn()`].
+ *
+ * Returns `true` if interruption was requested successfully.
+ */
+export declare function longRunningCancelTsfn(taskId: number): boolean
+
+/** Event payload for Node callbacks. */
+export interface LongRunningEvent {
+  /** The id of the task that emitted this event. */
+  taskId: number
+  /** The event category. */
+  kind: LongRunningEventKind
+  /** The latest completed step if available. */
+  step?: number
+  /** An optional error message for failed tasks. */
+  message?: string
+}
+
+/** Kinds of events emitted to JavaScript while a long-running task executes. */
+export declare const enum LongRunningEventKind {
+  /** The task produced an intermediate progress update. */
+  Progress = 'Progress',
+  /** The task finished successfully. */
+  Done = 'Done',
+  /** The task stopped because interruption was requested. */
+  Cancelled = 'Cancelled',
+  /** The task failed with an error. */
+  Error = 'Error'
+}
+
+/**
+ * Start a long-running task and stream progress to `callback` via a ThreadsafeFunction.
+ *
+ * Returns the task id, which can be used to interrupt processing with
+ * [`long_running_cancel_tsfn()`].
+ */
+export declare function longRunningStartTsfn(durationMs: number, callback: ((err: Error | null, arg: LongRunningEvent) => any)): number
+
 /**
  * Provide a unified diff for `change`, but fail if `change` is a [type-change](but_core::ModeFlags::TypeChange)
  * or if it involves a change to a [submodule](gix::object::Kind::Commit).
diff --git a/packages/but-sdk/src/generated/index.js b/packages/but-sdk/src/generated/index.js
@@ -579,7 +579,7 @@ if (!nativeBinding) {
   throw new Error(`Failed to load native binding`)
 }
 
-const { applyNapi, assignHunkNapi, branchDetailsNapi, branchDiffNapi, changesInWorktreeNapi, commitAmendNapi, commitCreateNapi, commitDetailsWithLineStatsNapi, commitInsertBlankNapi, commitMoveChangesBetweenNapi, commitMoveNapi, commitMoveToBranchNapi, commitRewordNapi, commitUncommitChangesNapi, headInfoNapi, listBranchesNapi, listProjectsNapi, treeChangeDiffsNapi, unapplyStackNapi } = nativeBinding
+const { applyNapi, assignHunkNapi, branchDetailsNapi, branchDiffNapi, changesInWorktreeNapi, commitAmendNapi, commitCreateNapi, commitDetailsWithLineStatsNapi, commitInsertBlankNapi, commitMoveChangesBetweenNapi, commitMoveNapi, commitMoveToBranchNapi, commitRewordNapi, commitUncommitChangesNapi, headInfoNapi, listBranchesNapi, listProjectsNapi, longRunningCancelTsfn, LongRunningEventKind, longRunningStartTsfn, treeChangeDiffsNapi, unapplyStackNapi } = nativeBinding
 export { applyNapi }
 export { assignHunkNapi }
 export { branchDetailsNapi }
@@ -597,5 +597,8 @@ export { commitUncommitChangesNapi }
 export { headInfoNapi }
 export { listBranchesNapi }
 export { listProjectsNapi }
+export { longRunningCancelTsfn }
+export { LongRunningEventKind }
+export { longRunningStartTsfn }
 export { treeChangeDiffsNapi }
 export { unapplyStackNapi }
diff --git a/packages/but-sdk/src/test.ts b/packages/but-sdk/src/test.ts
@@ -1,23 +1,102 @@
 /* eslint-disable no-console */
-import { listProjectsNapi, stackDetailsNapi, stacksNapi } from "./generated/index.js";
+import {
+	longRunningCancelTsfn,
+	LongRunningEventKind,
+	longRunningStartTsfn,
+} from "./generated/index.js";
 
+function isNumber(something: unknown): something is number {
+	return typeof something === "number";
+}
+
+/**
+ * So what's the deal here?
+ *
+ * We're testing running a long-running process being triggered by JS but not blocking it.
+ *
+ * The process emits back events that we can read and react to.
+ *
+ * It can be stopped and handled gracefully.
+ */
+async function runLongRunning({
+	durationMs,
+	signal,
+	onProgress,
+}: {
+	durationMs: number;
+	signal?: AbortSignal;
+	onProgress?: (step: number) => void;
+}) {
+	return await new Promise<{ lastStep: number }>((resolve, reject) => {
+		let taskId = 0;
+		let lastStep = 0;
+
+		taskId = longRunningStartTsfn(durationMs, (err, event) => {
+			if (err) {
+				reject(err);
+				return;
+			}
+
+			if (isNumber(event.step)) {
+				lastStep = event.step;
+			}
+
+			if (event.kind === LongRunningEventKind.Progress && isNumber(event.step)) {
+				onProgress?.(event.step);
+				return;
+			}
+
+			if (event.kind === LongRunningEventKind.Done) {
+				resolve({ lastStep });
+				return;
+			}
+
+			if (event.kind === LongRunningEventKind.Cancelled) {
+				reject(new Error("yep. interrupted"));
+				return;
+			}
+
+			if (event.kind === LongRunningEventKind.Error) {
+				reject(new Error(event.message ?? "unknown error"));
+			}
+		});
+
+		console.log("start long running process, but don't block");
+
+		if (signal) {
+			console.log("there is a signal");
+			if (signal.aborted) {
+				console.log("signal is aborted");
+				longRunningCancelTsfn(taskId);
+			} else {
+				signal.addEventListener(
+					"abort",
+					() => {
+						console.log("signal abort event triggered");
+						longRunningCancelTsfn(taskId);
+					},
+					{ once: true },
+				);
+			}
+		}
+	});
+}
 async function main() {
-	const projects = await listProjectsNapi([]);
-	console.log(projects);
-
-	if (projects.length === 0) {
-		console.log("No projects found");
-	}
-
-	const project = projects.at(0);
-	if (!project) throw new Error("The world is wrong");
-
-	const stacks = await stacksNapi(project.id, null);
-	for (const stack of stacks) {
-		const details = await stackDetailsNapi(project.id, stack.id);
-		console.log("This are the details for stack with id: " + stack.id);
-		console.log(details);
-	}
+	const abortController = new AbortController();
+	setTimeout(() => {
+		// Abort after a second
+		console.log("after waiting a second, interrupt.");
+		abortController.abort();
+	}, 1000);
+
+	const result = await runLongRunning({
+		durationMs: 5000,
+		signal: abortController.signal,
+		onProgress: (step: number) => console.log(`step ${step}`),
+	}).catch((e) => `probably interrupt error: ${e}`);
+
+	console.log("\nresult");
+	console.log(result);
 }
 
 main();
diff --git a/packages/but-sdk/tsconfig.json b/packages/but-sdk/tsconfig.json
@@ -15,7 +15,7 @@
 		"declaration": true,
 		"declarationMap": true,
 		"verbatimModuleSyntax": false,
-		"isolatedModules": true,
+		"isolatedModules": false,
 		"outDir": "dist",
 		"rootDir": "src"
 	},
