Cross-Platform Development Support for Windows Environment

Cargo.toml

@@ -1,8 +1,5 @@
 [workspace]
-members = [
-	"crates/*",
-	"integration-tests"
-]
+members = ["crates/*", "integration-tests"]
 resolver = "2"
 
 [workspace.dependencies]
@@ -16,26 +13,41 @@ bincode = "1.3.3"
 build-info = "0.0.35"
 build-info-build = { version = "0.0.35" }
 bytes = "1.4.0"
-camino = {version = "1.1.4", features = ["serde1"] }
+camino = { version = "1.1.4", features = ["serde1"] }
 clap = { version = "4.2.4", features = ["derive"] }
 compact_str = { version = "0.7.0", "features" = ["serde"] }
-config = {version = "0.13.3", default-features = false, features = ["toml"] }
+config = { version = "0.13.3", default-features = false, features = ["toml"] }
 crc32fast = "1.3.2"
 enquote = "1.1.0"
 eyre = "0.6.8"
 fallible-iterator = "0.3.0"
-foca = { version = "0.16.0", features = ["std", "tracing", "bincode-codec", "serde"] }
+foca = { version = "0.16.0", features = [
+	"std",
+	"tracing",
+	"bincode-codec",
+	"serde",
+] }
 futures = "0.3.28"
 futures-util = "0.3.28"
 hex = "0.4.3"
 hostname = "0.3.1"
 http-body = "0.4.5"
-hyper = { version = "0.14.26", features = ["h2", "http1", "http2", "server", "tcp", "stream", "client"] }
+hyper = { version = "0.14.26", features = [
+	"h2",
+	"http1",
+	"http2",
+	"server",
+	"tcp",
+	"stream",
+	"client",
+] }
 hyper-rustls = { version = "0.24.0", features = ["http2"] }
 indexmap = { version = "2.1.0", features = ["serde"] }
 itertools = { version = "0.10.5" }
 metrics = "0.22.0"
-metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = ["http-listener"] }
+metrics-exporter-prometheus = { version = "0.13.0", default-features = false, features = [
+	"http-listener",
+] }
 metrics-util = { version = "0.16.0" }
 nom = "7.0"
 once_cell = "1.17.1"
@@ -52,15 +64,31 @@ rand = { version = "0.8.5", features = ["small_rng"] }
 rangemap = { version = "1.5.1", features = ["serde1"] }
 rcgen = { version = "0.11.1", features = ["x509-parser"] }
 rhai = { version = "1.15.1", features = ["sync"] }
-rusqlite = { version = "0.30.0", features = ["serde_json", "time", "bundled", "uuid", "array", "load_extension", "column_decltype", "vtab", "functions", "chrono", "series"] }
+rusqlite = { version = "0.30.0", features = [
+	"serde_json",
+	"time",
+	"bundled",
+	"uuid",
+	"array",
+	"load_extension",
+	"column_decltype",
+	"vtab",
+	"functions",
+	"chrono",
+	"series",
+] }
 rustls = { version = "0.21.0", features = ["dangerous_configuration", "quic"] }
 rustls-pemfile = "1.0.2"
 seahash = "4.1.0"
 serde = "1.0.159"
 serde_json = { version = "1.0.95", features = ["raw_value"] }
 serde_with = "2.3.2"
 smallvec = { version = "1.11.0", features = ["serde", "write", "union"] }
-speedy = { version = "0.8.7", features = ["uuid", "smallvec", "indexmap"], package = "corro-speedy" }
+speedy = { version = "0.8.7", features = [
+	"uuid",
+	"smallvec",
+	"indexmap",
+], package = "corro-speedy" }
 sqlite3-parser = "0.12.0"
 strum = { version = "0.24.1", features = ["derive"] }
 tempfile = "3.5.0"
@@ -69,13 +97,15 @@ time = { version = "0.3.15", features = ["macros", "serde-well-known"] }
 tokio = { version = "1.34", features = ["full"] }
 tokio-metrics = "0.3.0"
 tokio-serde = { version = "0.8", features = ["json"] }
-tokio-stream = { version = "0.1.12", features = ["sync"] }
+tokio-stream = { version = "0.1.12", features = ["sync", "signal"] }
 tokio-util = { version = "0.7.7", features = ["io", "codec", "net"] }
 tower = { version = "0.4.13", features = ["limit", "load-shed", "buffer"] }
 tower-http = { version = "0.4.0", features = ["trace", "auth"] }
 tracing = "0.1.37"
 tracing-filter = { version = "0.1.0-alpha.2", features = ["smallvec"] }
-tracing-opentelemetry = { version = "0.21.0", default-features = false, features = ["tracing-log"]}
+tracing-opentelemetry = { version = "0.21.0", default-features = false, features = [
+	"tracing-log",
+] }
 tracing-subscriber = { version = "0.3.16", features = ["json", "env-filter"] }
 trust-dns-resolver = "0.22.0"
 uhlc = { version = "0.7", features = ["defmt"] }
@@ -109,4 +139,4 @@ codegen-units = 16
 [profile.test.package.backtrace]
 # backtraces are super expensive to compute in development,
 # always optimize that crate
-opt-level = 2
+opt-level = 2

crates/corro-types/src/sqlite.rs

@@ -27,8 +27,8 @@ pub const CRSQL_EXT: &[u8] = include_bytes!("../crsqlite-darwin-x86_64.dylib");
 pub const CRSQL_EXT: &[u8] = include_bytes!("../crsqlite-linux-x86_64.so");
 #[cfg(all(target_arch = "aarch64", target_os = "linux"))]
 pub const CRSQL_EXT: &[u8] = include_bytes!("../crsqlite-linux-aarch64.so");
-
-// TODO: support windows
+#[cfg(all(target_arch = "x86_64", target_os = "windows"))]
+pub const CRSQL_EXT: &[u8] = include_bytes!("../crsqlite-windows-x86_64.dll");
 
 // need to keep this alive!
 static CRSQL_EXT_DIR: Lazy<TempDir> = Lazy::new(|| {

crates/tripwire/src/lib.rs

@@ -5,6 +5,7 @@
 #![deny(clippy::await_holding_lock)]
 
 mod preempt;
+#[cfg(unix)]
 mod signalstream;
 mod tripwire;
 

crates/tripwire/src/tripwire.rs

@@ -5,13 +5,17 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio::{
-    signal::unix::{signal, SignalKind},
-    sync::{mpsc, watch},
-};
+#[cfg(unix)]
+use tokio::signal::unix::{signal, SignalKind};
+#[cfg(windows)]
+use tokio::signal::windows::ctrl_c;
+use tokio::sync::{mpsc, watch};
+#[cfg(windows)]
+use tokio_stream::wrappers::CtrlCStream;
 use tokio_stream::wrappers::{ReceiverStream, WatchStream};
 use tracing::{debug, warn};
 
+#[cfg(unix)]
 use crate::signalstream::SignalStream;
 
 /// A `Future` that completes once the program is requested to shutdown. This
@@ -51,12 +55,22 @@ impl Tripwire {
     }
 
     /// Listen for SIGTERM and SIGINT
+    #[cfg(unix)]
     pub fn new_signals() -> (Self, TripwireWorker<Select<SignalStream, SignalStream>>) {
+        // For non-Windows platforms, create signal streams for SIGTERM and SIGINT
         let sigterms = SignalStream::new(signal(SignalKind::terminate()).unwrap());
         let sigints = SignalStream::new(signal(SignalKind::interrupt()).unwrap());
         Self::new(select(sigterms, sigints))
     }
 
+    #[cfg(windows)]
+    pub fn new_signals() -> (Self, TripwireWorker<Select<CtrlCStream, CtrlCStream>>) {
+        // For Windows platforms, create two Ctrl-C signal streams to meet the requirement of `select`
+        let ctrl_c1 = CtrlCStream::new(ctrl_c().unwrap());
+        let ctrl_c2 = CtrlCStream::new(ctrl_c().unwrap());
+        Self::new(select(ctrl_c1, ctrl_c2))
+    }
+
     /// Returns an Arc of the current [TripwireState]
     pub fn state(&self) -> TripwireState {
         *self.subscription_rx.borrow()

rust-toolchain.toml

@@ -1,6 +1,7 @@
 [toolchain]
 channel = "nightly-2023-12-03"
 targets = [
+    # "x86_64-pc-windows-gnu",
     "x86_64-apple-darwin",
-    "x86_64-unknown-linux-gnu"
+    "x86_64-unknown-linux-gnu",
 ]