feat: mesh scheduling — SWIM gossip, work-stealing, hash ring

.github/workflows/ci.yml

@@ -131,6 +131,9 @@ jobs:
       - name: Check build with native-async features
         run: cargo check --workspace --features native-async
 
+      - name: Check build with mesh feature
+        run: cargo check --workspace --features mesh
+
   rust-test-postgres:
     name: Rust Tests (PostgreSQL)
     needs: changes
@@ -215,6 +218,35 @@ jobs:
           LD_LIBRARY_PATH: ${{ env.pythonLocation }}/lib
           TASKITO_REDIS_TEST_URL: redis://localhost:6379/15
 
+  rust-test-mesh:
+    name: Rust Tests (Mesh)
+    needs: changes
+    if: needs.changes.outputs.rust == 'true' || github.event_name == 'push' || github.event_name == 'workflow_dispatch'
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v6
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Restore Cargo cache
+        uses: Swatinem/rust-cache@v2
+        with:
+          save-if: false
+          cache-bin: false
+          prefix-key: v1-rust-bin-fix
+
+      - name: Run mesh crate tests
+        run: cargo test -p taskito-mesh
+        env:
+          LD_LIBRARY_PATH: ${{ env.pythonLocation }}/lib
+
   test:
     name: Python Tests (${{ matrix.os }} / Python ${{ matrix.python-version }})
     needs: [lint, changes]
@@ -272,7 +304,7 @@ jobs:
         uses: PyO3/maturin-action@v1.51.0
         with:
           command: develop
-          args: --release --features extension-module,postgres,redis,native-async,workflows
+          args: --release --features extension-module,postgres,redis,native-async,workflows,mesh
           maturin-version: v1.13.3
 
       - name: Run Python test suite
@@ -284,7 +316,7 @@ jobs:
   ci-status:
     name: CI status
     if: always()
-    needs: [lint, rust-test, rust-test-postgres, rust-test-redis, test]
+    needs: [lint, rust-test, rust-test-postgres, rust-test-redis, rust-test-mesh, test]
     runs-on: ubuntu-latest
     steps:
       - name: Check that no required job failed

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["crates/taskito-core", "crates/taskito-python", "crates/taskito-async", "crates/taskito-workflows"]
+members = ["crates/taskito-core", "crates/taskito-python", "crates/taskito-async", "crates/taskito-workflows", "crates/taskito-mesh"]
 resolver = "2"
 
 [workspace.dependencies]

crates/taskito-mesh/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "taskito-mesh"
+version = "0.15.2"
+edition = "2021"
+
+[dependencies]
+taskito-core = { path = "../taskito-core" }
+tokio = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+bincode = "1"
+log = { workspace = true }
+rand = { workspace = true }
+xxhash-rust = { version = "0.8", features = ["xxh3"] }
+base64 = "0.22"

crates/taskito-mesh/src/config.rs

@@ -0,0 +1,86 @@
+use std::net::{IpAddr, Ipv4Addr};
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MeshConfig {
+    /// UDP port for SWIM gossip protocol.
+    pub gossip_port: u16,
+    /// TCP port for work-stealing connections (default: gossip_port + 1).
+    pub steal_port: u16,
+    /// Bind address for gossip and steal listeners.
+    pub bind_addr: String,
+    /// Seed nodes for initial cluster join (e.g., `["host1:7946"]`).
+    pub seeds: Vec<String>,
+    /// SWIM protocol period in milliseconds.
+    pub protocol_period_ms: u64,
+    /// Indirect ping targets for failure detection.
+    pub indirect_ping_count: usize,
+    /// Suspicion timeout multiplier (applied to `log(N+1) * protocol_period`).
+    pub suspicion_multiplier: u32,
+    /// Virtual nodes per worker on the consistent hash ring.
+    pub virtual_nodes: usize,
+    /// Max jobs in the local deque before refusing to prefetch.
+    pub local_buffer_capacity: usize,
+    /// Max jobs to steal per request.
+    pub max_steal_batch: usize,
+    /// Steal when own deque length is at or below this threshold.
+    pub steal_threshold: usize,
+    /// Affinity weight: 0.0 = ignore affinity, 1.0 = strict affinity.
+    pub affinity_weight: f64,
+    /// Whether work-stealing is enabled.
+    pub enable_stealing: bool,
+    /// IP address to advertise to peers for gossip and steal connections.
+    /// Required when `bind_addr` is `0.0.0.0` and peers run on other hosts.
+    /// Falls back to `bind_addr` when unset.
+    pub advertise_addr: Option<String>,
+    /// Shared encryption key for gossip messages (base64-encoded, 32 bytes).
+    /// When set, gossip datagrams are XOR-encrypted with this key.
+    /// Not cryptographically strong — prevents casual sniffing only.
+    pub encryption_key: Option<String>,
+    /// Max steal requests per peer per second. 0 = unlimited.
+    pub steal_rate_limit: u32,
+}
+
+impl Default for MeshConfig {
+    fn default() -> Self {
+        Self {
+            gossip_port: 7946,
+            steal_port: 7947,
+            bind_addr: "0.0.0.0".to_string(),
+            seeds: Vec::new(),
+            protocol_period_ms: 500,
+            indirect_ping_count: 3,
+            suspicion_multiplier: 4,
+            virtual_nodes: 150,
+            local_buffer_capacity: 64,
+            max_steal_batch: 4,
+            steal_threshold: 2,
+            affinity_weight: 0.7,
+            enable_stealing: true,
+            advertise_addr: None,
+            encryption_key: None,
+            steal_rate_limit: 10,
+        }
+    }
+}
+
+impl MeshConfig {
+    /// Resolve the IP to advertise to peers.
+    /// Prefers `advertise_addr`, falls back to `bind_addr`.
+    pub fn advertise_ip(&self) -> IpAddr {
+        self.advertise_addr
+            .as_deref()
+            .or(Some(self.bind_addr.as_str()))
+            .and_then(|s| s.parse::<IpAddr>().ok())
+            .unwrap_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
+    }
+
+    /// Decode the encryption key from base64. Returns None if unset or invalid.
+    pub fn decoded_encryption_key(&self) -> Option<Vec<u8>> {
+        self.encryption_key.as_ref().and_then(|k| {
+            use base64::Engine;
+            base64::engine::general_purpose::STANDARD.decode(k).ok()
+        })
+    }
+}