feat: static check params

Cargo.toml

@@ -41,7 +41,7 @@ matchit = { git = "https://github.com/ibraheemdev/matchit", branch = "master" }
 # Serialization / Data formats
 serde = "1.0.229"
 serde_json = "1.0.151"
-jsonschema = { version = "0.56.0", default-features = false }
+jsonschema = { version = "0.57.0", default-features = false }
 
 # Template engines
 tera = { version = "2", features = ["glob_fs", "fast"] }
@@ -54,3 +54,4 @@ url = "2.5.8"
 jsonwebtoken = { version = "11.0.0", features = ["rust_crypto"] }
 regex = "1.13.1"
 unicode-normalization = "0.1.25"
+socket2 = { version = "0.6.5", features = ["all"] }

oxapy/__init__.py

@@ -5,14 +5,13 @@
 import subprocess
 import time
 import base64
+import signal
 import typing
 import mimetypes
 import hmac
 import orjson as json
 import hashlib
 
-from functools import partial
-
 from watchdog.observers import Observer
 from watchdog.events import PatternMatchingEventHandler
 
@@ -63,92 +62,137 @@ def set_watch_dir(self, watch_dir: str):
         self.__watch_dir = watch_dir
         return self
 
-    def run(self, reload: bool = False, workers: typing.Optional[int] = None):
+    def run(
+        self,
+        reload: bool = False,
+        processes: typing.Optional[int] = None,
+        workers: typing.Optional[int] = None,
+    ):
         """
-        Starts the server or the supervisor process.
-
-        If `reload` is enabled and the current process is not flagged as a worker,
-        it launches the supervisor to watch for file changes. Otherwise, it starts
-        the actual HTTP server instance.
+        Starts the server, optionally as a pool of OS processes.
 
         Args:
-            reload (bool): Whether to enable auto-reloading on file changes. Defaults to False.
-            workers (int, optional): The number of worker processes to run. Defaults to None.
+            reload (bool): Watch for file changes and restart the pool. Defaults to False.
+            processes (int, optional): Number of OS-level worker processes to run,
+                all sharing the listening port via SO_REUSEPORT. Defaults to 1.
+            workers (int, optional): Tokio worker threads per process (passed through
+                to the Rust runtime's block_on).
         """
-        if reload and os.environ.get("OXAPY_WORKER") != "1":
-            self._run_supervisor()
-        else:
+        if os.environ.get("OXAPY_WORKER") == "1":
+            # We are a spawned worker: just run the actual server.
             return super().run(workers)
 
-    def _run_supervisor(self):
-        """
-        Manages the file watcher and the child worker process.
+        num_processes = processes if processes and processes > 0 else 1
+
+        if not reload and num_processes <= 1:
+            return super().run(workers)
 
-        Sets up a directory observer. When a watched file is modified, created,
-        or deleted, it gracefully terminates the current worker process and
-        spawns a fresh one.
+        self._run_supervisor(num_processes, reload)
+
+    def _run_supervisor(self, num_processes: int, reload: bool):
+        """
+        Manages a pool of `num_processes` worker processes.
+
+        In reload mode, any watched file change tears down and restarts the whole
+        pool; so does a worker that dies unexpectedly, since the surviving workers
+        would otherwise keep serving the pre-crash code. Outside of reload mode, a
+        worker that dies unexpectedly is respawned on its own (self-healing pool);
+        a worker that exits cleanly is left down, and once every worker has exited
+        cleanly the supervisor returns.
         """
         env = os.environ.copy()
         env["OXAPY_WORKER"] = "1"
 
-        reload_requested = threading.Event()
-        changed_file_path = ""
-
-        def on_file_changed(event):
-            """Triggers a reload sequence when a watched file is modified."""
-            nonlocal changed_file_path
-            changed_file_path = event.src_path
-            reload_requested.set()
-
-        handler = PatternMatchingEventHandler(
-            patterns=self.__patterns, ignore_directories=True
-        )
-        handler.on_modified = on_file_changed
-        handler.on_created = on_file_changed
-        handler.on_deleted = on_file_changed
-
-        observer = Observer()
-        observer.schedule(handler, self.__watch_dir, recursive=True)
-        observer.start()
-
         def spawn_worker() -> subprocess.Popen:
-            """Spawns the child server process with the worker environment flag."""
             return subprocess.Popen([sys.executable] + sys.argv, env=env)
 
         def terminate_worker(proc: subprocess.Popen):
-            """Gracefully terminates a worker process, escalating to a kill if it hangs."""
             if proc and proc.poll() is None:
                 proc.terminate()
                 try:
                     proc.wait(timeout=3)
                 except subprocess.TimeoutExpired:
                     proc.kill()
 
-        worker_process = spawn_worker()
+        def terminate_pool(pool: list[subprocess.Popen]):
+            for proc in pool:
+                terminate_worker(proc)
+
+        pool = [spawn_worker() for _ in range(num_processes)]
+
+        reload_requested = threading.Event()
+        changed_file_path = ""
+        observer = None
+
+        def restart_pool(reason: str) -> list[subprocess.Popen]:
+            """
+            Tears down every worker and spawns a fresh pool, so that no process
+            keeps serving stale code.
+
+            Args:
+                reason (str): Human readable reason, shown in the restart notice.
+
+            Returns:
+                list[subprocess.Popen]: The freshly spawned pool.
+            """
+            time.sleep(0.3)
+            reload_requested.clear()
+            terminate_pool(pool)
+            print(f"Reloading {num_processes} worker(s)... ({reason})")
+            return [spawn_worker() for _ in range(num_processes)]
+
+        # Let SIGTERM unwind the same way SIGINT/KeyboardInterrupt does.
+        def _on_sigterm(signum, frame):
+            raise KeyboardInterrupt
+
+        previous_sigterm = signal.signal(signal.SIGTERM, _on_sigterm)
+
+        if reload:
+
+            def on_file_changed(event):
+                nonlocal changed_file_path
+                changed_file_path = event.src_path
+                reload_requested.set()
+
+            handler = PatternMatchingEventHandler(
+                patterns=self.__patterns, ignore_directories=True
+            )
+            handler.on_modified = on_file_changed
+            handler.on_created = on_file_changed
+            handler.on_deleted = on_file_changed
+
+            observer = Observer()
+            observer.schedule(handler, self.__watch_dir, recursive=True)
+            observer.start()
 
         try:
             while True:
-                if reload_requested.wait(timeout=0.2):
-                    time.sleep(0.3)
-                    reload_requested.clear()
-                    terminate_worker(worker_process)
+                if reload and reload_requested.wait(timeout=0.2):
                     filename = os.path.basename(changed_file_path)
-                    print(f"Reloading... ({filename} changed)")
-                    worker_process = spawn_worker()
-                elif worker_process.poll() is not None:
-                    if worker_process.returncode != 0:
-                        reload_requested.wait()
-                        time.sleep(0.3)
-                        reload_requested.clear()
-                        worker_process = spawn_worker()
-                    else:
-                        break
+                    pool = restart_pool(f"{filename} changed")
+                    continue
+
+                time.sleep(0.2)
+
+                for i, proc in enumerate(pool):
+                    if proc.poll() is None:
+                        continue
+                    if proc.returncode != 0:
+                        if reload:
+                            pool = restart_pool(f"worker {i} exited with code {proc.returncode}")
+                            break
+                        print(f"Worker {i} exited with code {proc.returncode}, restarting...")
+                        pool[i] = spawn_worker()
+                    elif not reload and all(p.poll() is not None for p in pool):
+                        return
         except KeyboardInterrupt:
             pass
         finally:
-            observer.stop()
-            observer.join()
-            terminate_worker(worker_process)
+            signal.signal(signal.SIGTERM, previous_sigterm)
+            if observer:
+                observer.stop()
+                observer.join()
+            terminate_pool(pool)
 
 
 def _b64_encode(data: bytes) -> str:

src/into_response.rs

@@ -11,13 +11,16 @@ use crate::{
 };
 
 type Error = Box<dyn std::error::Error>;
+const TEXT_PLAIN: HeaderValue = HeaderValue::from_static("text/plain");
+const APP_JSON: HeaderValue = HeaderValue::from_static("application/json");
 
 impl TryFrom<String> for Response {
     type Error = Error;
 
+    #[inline]
     fn try_from(val: String) -> Result<Self, Self::Error> {
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("text/plain"));
+        let mut headers = HeaderMap::with_capacity(1);
+        headers.insert(CONTENT_TYPE, TEXT_PLAIN);
         Ok(Response {
             status: Status::OK,
             headers,
@@ -29,9 +32,10 @@ impl TryFrom<String> for Response {
 impl TryFrom<Bound<'_, PyAny>> for Response {
     type Error = Error;
 
+    #[inline]
     fn try_from(val: Bound<PyAny>) -> Result<Self, Self::Error> {
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        let mut headers = HeaderMap::with_capacity(1);
+        headers.insert(CONTENT_TYPE, APP_JSON);
         Ok(Response {
             status: Status::OK,
             headers,
@@ -43,9 +47,10 @@ impl TryFrom<Bound<'_, PyAny>> for Response {
 impl TryFrom<(String, Status)> for Response {
     type Error = Error;
 
+    #[inline]
     fn try_from(val: (String, Status)) -> Result<Self, Self::Error> {
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("text/plain"));
+        let mut headers = HeaderMap::with_capacity(1);
+        headers.insert(CONTENT_TYPE, TEXT_PLAIN);
         Ok(Response {
             status: val.1,
             headers,
@@ -57,9 +62,10 @@ impl TryFrom<(String, Status)> for Response {
 impl TryFrom<(Bound<'_, PyAny>, Status)> for Response {
     type Error = Error;
 
+    #[inline]
     fn try_from(val: (Bound<PyAny>, Status)) -> Result<Self, Self::Error> {
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        let mut headers = HeaderMap::with_capacity(1);
+        headers.insert(CONTENT_TYPE, APP_JSON);
         Ok(Response {
             status: val.1,
             headers,
@@ -69,9 +75,10 @@ impl TryFrom<(Bound<'_, PyAny>, Status)> for Response {
 }
 
 impl From<Status> for Response {
+    #[inline]
     fn from(val: Status) -> Self {
-        let mut headers = HeaderMap::new();
-        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        let mut headers = HeaderMap::with_capacity(1);
+        headers.insert(CONTENT_TYPE, APP_JSON);
         Response {
             status: val,
             headers,
@@ -111,6 +118,7 @@ impl From<PyErr> for Response {
 impl TryFrom<Cors> for Response {
     type Error = PyErr;
 
+    #[inline]
     fn try_from(cors: Cors) -> Result<Self, Self::Error> {
         let mut response = Response::from(Status::NO_CONTENT);
         cors.apply_headers(&mut response)?;
@@ -181,11 +189,11 @@ pub fn convert_to_response(result: Py<PyAny>, py: Python<'_>) -> PyResult<Respon
     to_response!(
         result,
         py,
-        Response,
+        String,
         Status,
+        Response,
         (String, Status),
         (Bound<PyAny>, Status),
-        String,
         Bound<PyAny>
     )
 }