Add terminal sync cancellation

## Summary

POS terminal cleanup needs a way to stop an in-flight v1 remote-replica sync without dropping the future and racing local replica deletion against libsql cleanup. Add a cooperative cancellation path that returns through libsql's normal rollback and settlement flow.

## Approach

- Track the active foreground sync with a per-sync cancellation token stored outside the replicator mutex.
- Thread cancellation through the replication state machine, remote waits, snapshot streaming, and SQLite injection.
- Interrupt and then join blocking SQLite injection work before returning cancellation.
- Expose Database::cancel_current_sync_for_shutdown() for terminal shutdown callers and add coverage for cancellation rollback and the no-active-sync API case.

Co-authored-by: Claude <noreply@anthropic.com>
Orchestrated-by: ae <noreply@shopify.com>

Author: tszymczyszyn-shopify

libsql-replication/src/injector/error.rs

@@ -9,4 +9,6 @@ pub enum Error {
     Sqlite(#[from] rusqlite::Error),
     #[error("A fatal error occured injecting frames: {0}")]
     FatalInjectError(BoxError),
+    #[error("sync cancelled for terminal shutdown")]
+    SyncCancelledForShutdown,
 }

libsql-replication/src/injector/mod.rs

@@ -2,6 +2,7 @@ use std::future::Future;
 
 use super::rpc::replication::Frame as RpcFrame;
 pub use sqlite_injector::SqliteInjector;
+use tokio_util::sync::CancellationToken;
 
 use crate::frame::FrameNo;
 
@@ -18,6 +19,15 @@ pub trait Injector {
         frame: RpcFrame,
     ) -> impl Future<Output = Result<Option<FrameNo>>> + Send;
 
+    /// Inject a singular frame, cooperatively observing terminal sync cancellation.
+    fn inject_frame_with_cancellation(
+        &mut self,
+        frame: RpcFrame,
+        _token: &CancellationToken,
+    ) -> impl Future<Output = Result<Option<FrameNo>>> + Send {
+        self.inject_frame(frame)
+    }
+
     /// Discard any uncommintted frames.
     fn rollback(&mut self) -> impl Future<Output = ()> + Send;
 

libsql-replication/src/injector/sqlite_injector/mod.rs

@@ -5,6 +5,7 @@ use std::{collections::VecDeque, path::PathBuf};
 use parking_lot::Mutex;
 use rusqlite::OpenFlags;
 use tokio::task::spawn_blocking;
+use tokio_util::sync::CancellationToken;
 
 use crate::frame::{Frame, FrameNo};
 use crate::rpc::replication::Frame as RpcFrame;
@@ -23,6 +24,24 @@ pub type FrameBuffer = Arc<Mutex<VecDeque<Frame>>>;
 
 pub struct SqliteInjector {
     pub(in super::super) inner: Arc<Mutex<SqliteInjectorInner>>,
+    interrupt_handle: InjectorInterruptHandle,
+}
+
+#[derive(Clone, Default)]
+struct InjectorInterruptHandle {
+    current: Arc<Mutex<Option<rusqlite::InterruptHandle>>>,
+}
+
+impl InjectorInterruptHandle {
+    fn set(&self, handle: rusqlite::InterruptHandle) {
+        *self.current.lock() = Some(handle);
+    }
+
+    fn interrupt(&self) {
+        if let Some(handle) = self.current.lock().as_ref() {
+            handle.interrupt();
+        }
+    }
 }
 
 impl Injector for SqliteInjector {
@@ -35,6 +54,32 @@ impl Injector for SqliteInjector {
             .unwrap()
     }
 
+    async fn inject_frame_with_cancellation(
+        &mut self,
+        frame: RpcFrame,
+        token: &CancellationToken,
+    ) -> Result<Option<FrameNo>> {
+        let inner = self.inner.clone();
+        let interrupt_handle = self.interrupt_handle.clone();
+        let frame =
+            Frame::try_from(&frame.data[..]).map_err(|e| Error::FatalInjectError(e.into()))?;
+        let mut join = spawn_blocking(move || inner.lock().inject_frame(frame));
+
+        tokio::select! {
+            biased;
+
+            result = &mut join => result.unwrap(),
+            _ = token.cancelled() => {
+                interrupt_handle.interrupt();
+                match join.await {
+                    Ok(Ok(result)) => Ok(result),
+                    Ok(Err(_)) => Err(Error::SyncCancelledForShutdown),
+                    Err(e) => Err(Error::FatalInjectError(e.into())),
+                }
+            }
+        }
+    }
+
     async fn rollback(&mut self) {
         let inner = self.inner.clone();
         spawn_blocking(move || inner.lock().rollback())
@@ -58,14 +103,23 @@ impl SqliteInjector {
         auto_checkpoint: u32,
         encryption_config: Option<libsql_sys::EncryptionConfig>,
     ) -> super::Result<Self> {
+        let interrupt_handle = InjectorInterruptHandle::default();
+        let inner_interrupt_handle = interrupt_handle.clone();
         let inner = spawn_blocking(move || {
-            SqliteInjectorInner::new(path, capacity, auto_checkpoint, encryption_config)
+            SqliteInjectorInner::new(
+                path,
+                capacity,
+                auto_checkpoint,
+                encryption_config,
+                inner_interrupt_handle,
+            )
         })
         .await
         .unwrap()?;
 
         Ok(Self {
             inner: Arc::new(Mutex::new(inner)),
+            interrupt_handle,
         })
     }
 }
@@ -86,6 +140,7 @@ pub(in super::super) struct SqliteInjectorInner {
     path: PathBuf,
     encryption_config: Option<libsql_sys::EncryptionConfig>,
     auto_checkpoint: u32,
+    interrupt_handle: InjectorInterruptHandle,
 }
 
 /// Methods from this trait are called before and after performing a frame injection.
@@ -98,6 +153,7 @@ impl SqliteInjectorInner {
         capacity: usize,
         auto_checkpoint: u32,
         encryption_config: Option<libsql_sys::EncryptionConfig>,
+        interrupt_handle: InjectorInterruptHandle,
     ) -> Result<Self, Error> {
         let path = path.as_ref().to_path_buf();
 
@@ -113,6 +169,7 @@ impl SqliteInjectorInner {
             auto_checkpoint,
             encryption_config.clone(),
         )?;
+        interrupt_handle.set(connection.get_interrupt_handle());
 
         Ok(Self {
             is_txn: false,
@@ -124,6 +181,7 @@ impl SqliteInjectorInner {
             path,
             encryption_config,
             auto_checkpoint,
+            interrupt_handle,
         })
     }
 
@@ -235,6 +293,7 @@ impl SqliteInjectorInner {
                 self.auto_checkpoint,
                 self.encryption_config.clone(),
             )?;
+            self.interrupt_handle.set(new_conn.get_interrupt_handle());
 
             let _ = std::mem::replace(&mut *conn, new_conn);
         }
@@ -281,8 +340,14 @@ mod test {
     fn test_simple_inject_frames() {
         let temp = tempfile::tempdir().unwrap();
 
-        let mut injector =
-            SqliteInjectorInner::new(temp.path().join("data"), 10, 10000, None).unwrap();
+        let mut injector = SqliteInjectorInner::new(
+            temp.path().join("data"),
+            10,
+            10000,
+            None,
+            InjectorInterruptHandle::default(),
+        )
+        .unwrap();
         let log = wal_log();
         for frame in log {
             injector.inject_frame(frame).unwrap();
@@ -302,8 +367,14 @@ mod test {
         let temp = tempfile::tempdir().unwrap();
 
         // inject one frame at a time
-        let mut injector =
-            SqliteInjectorInner::new(temp.path().join("data"), 1, 10000, None).unwrap();
+        let mut injector = SqliteInjectorInner::new(
+            temp.path().join("data"),
+            1,
+            10000,
+            None,
+            InjectorInterruptHandle::default(),
+        )
+        .unwrap();
         let log = wal_log();
         for frame in log {
             injector.inject_frame(frame).unwrap();
@@ -323,8 +394,14 @@ mod test {
         let temp = tempfile::tempdir().unwrap();
 
         // inject one frame at a time
-        let mut injector =
-            SqliteInjectorInner::new(temp.path().join("data"), 10, 1000, None).unwrap();
+        let mut injector = SqliteInjectorInner::new(
+            temp.path().join("data"),
+            10,
+            1000,
+            None,
+            InjectorInterruptHandle::default(),
+        )
+        .unwrap();
         let mut frames = wal_log();
 
         assert!(injector