From 276f6e1fcf7b7fa746c7cfa4aff411ee230b421c Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 29 Jun 2026 16:44:17 +0530 Subject: [PATCH 1/6] feat(java): mesh scheduling JNI bridge (mesh feature) --- Cargo.lock | 1 + crates/taskito-java/Cargo.toml | 2 + crates/taskito-java/src/convert.rs | 5 ++ crates/taskito-java/src/lib.rs | 2 + crates/taskito-java/src/mesh.rs | 111 +++++++++++++++++++++++++++++ crates/taskito-java/src/worker.rs | 64 ++++++++++++++++- 6 files changed, 184 insertions(+), 1 deletion(-) create mode 100644 crates/taskito-java/src/mesh.rs diff --git a/Cargo.lock b/Cargo.lock index b21c4422..9b4fb5d5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1516,6 +1516,7 @@ dependencies = [ "serde", "serde_json", "taskito-core", + "taskito-mesh", "taskito-workflows", "tokio", "uuid", diff --git a/crates/taskito-java/Cargo.toml b/crates/taskito-java/Cargo.toml index 29465f38..2e15b9a6 100644 --- a/crates/taskito-java/Cargo.toml +++ b/crates/taskito-java/Cargo.toml @@ -14,10 +14,12 @@ default = [] postgres = ["taskito-core/postgres", "taskito-workflows?/postgres"] redis = ["taskito-core/redis", "taskito-workflows?/redis"] workflows = ["dep:taskito-workflows"] +mesh = ["dep:taskito-mesh"] [dependencies] taskito-core = { path = "../taskito-core" } taskito-workflows = { path = "../taskito-workflows", optional = true } +taskito-mesh = { path = "../taskito-mesh", optional = true } jni = "0.21" serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index 53e7de8d..8019808a 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -165,6 +165,11 @@ pub struct WorkerOptions { /// Per-task retry-backoff policies, registered with the scheduler at start. /// The core owns the retry engine; this only feeds it the backoff curve. pub task_configs: Option>, + /// Raw `taskito_mesh::MeshConfig` JSON. When present (and the `mesh` feature + /// is built), the worker interposes the mesh bridge between scheduler and + /// dispatcher; otherwise this is ignored. + #[cfg_attr(not(feature = "mesh"), allow(dead_code))] + pub mesh_config: Option, } /// A task's retry-backoff curve. Fields left unset fall back to the core's diff --git a/crates/taskito-java/src/lib.rs b/crates/taskito-java/src/lib.rs index 759044e2..f65d42bd 100644 --- a/crates/taskito-java/src/lib.rs +++ b/crates/taskito-java/src/lib.rs @@ -16,6 +16,8 @@ mod ffi; mod ffi_c; mod handle; mod jvm; +#[cfg(feature = "mesh")] +mod mesh; mod queue; mod worker; #[cfg(feature = "workflows")] diff --git a/crates/taskito-java/src/mesh.rs b/crates/taskito-java/src/mesh.rs new file mode 100644 index 00000000..af001502 --- /dev/null +++ b/crates/taskito-java/src/mesh.rs @@ -0,0 +1,111 @@ +//! Mesh scheduling bridge (feature `mesh`). +//! +//! When a worker is started with a `meshConfig`, [`spawn_mesh`] interposes a +//! [`taskito_mesh::MeshNode`] between the scheduler and the dispatcher: the +//! scheduler emits ready jobs into an intermediate channel, the bridge sorts +//! them into the node's affinity-aware local deque (and steals from busy peers), +//! then drains the deque to the real dispatcher channel. The DB stays the source +//! of truth, so the scheduler and dispatcher remain mesh-unaware — identical to +//! the Python and Node shells. + +use std::sync::Arc; + +use taskito_core::job::Job; +use taskito_core::scheduler::Scheduler; +use taskito_mesh::{MeshConfig, MeshNode}; +use tokio::sync::mpsc::Sender; + +/// Build the mesh node, start its gossip + steal-server tasks and the bridge +/// loop on `runtime`, and return the node so the worker can request shutdown and +/// read cluster info. `threads` is advertised to peers as this worker's capacity. +pub fn spawn_mesh( + runtime: &tokio::runtime::Runtime, + scheduler: Arc, + job_tx: Sender, + config_json: &str, + worker_id: String, + queues: Vec, + threads: u16, +) -> Arc { + let config: MeshConfig = serde_json::from_str(config_json).unwrap_or_default(); + log::info!( + "[taskito-java] mesh scheduling enabled (gossip_port={}, local_buffer={})", + config.gossip_port, + config.local_buffer_capacity, + ); + let node = Arc::new(MeshNode::new(worker_id, config)); + + // `spawn_gossip`/`spawn_steal_server` call the free `tokio::spawn` internally, + // so they must run inside the runtime context (we are on the JNI thread here). + let (gossip, steal_server) = { + let _guard = runtime.enter(); + ( + node.spawn_gossip(queues, threads), + node.spawn_steal_server(), + ) + }; + + let bridge_node = node.clone(); + runtime.spawn(async move { + run_mesh_bridge(scheduler, bridge_node, job_tx).await; + }); + // Keep the gossip + steal-server tasks alive for the worker's lifetime; they + // stop when the node's shutdown is signalled (see `request_shutdown`). + runtime.spawn(async move { + let _ = tokio::join!(gossip, steal_server); + }); + + node +} + +/// Pump jobs from the scheduler through the mesh local deque to the dispatcher. +async fn run_mesh_bridge(scheduler: Arc, mesh_node: Arc, job_tx: Sender) { + let (mesh_tx, mut mesh_rx) = tokio::sync::mpsc::channel::(64); + + let sched = scheduler.clone(); + let sched_task = tokio::spawn(async move { + sched.run(mesh_tx).await; + }); + + loop { + // Drain the local deque to the dispatcher first. + while let Some(job) = mesh_node.pop_local() { + if job_tx.send(job).await.is_err() { + let _ = sched_task.await; + return; + } + } + + // Steal from a busier peer when our deque is low. + if mesh_node.should_steal() { + mesh_node.try_steal().await; + while let Some(job) = mesh_node.pop_local() { + if job_tx.send(job).await.is_err() { + let _ = sched_task.await; + return; + } + } + } + + // Wait for the scheduler to produce jobs, then prefetch the batch. + match mesh_rx.recv().await { + Some(job) => { + let mut batch = vec![job]; + while let Ok(extra) = mesh_rx.try_recv() { + batch.push(extra); + } + mesh_node.prefetch(batch); + } + None => break, + } + } + + // Scheduler done: drain whatever remains in the deque. + while let Some(job) = mesh_node.pop_local() { + if job_tx.send(job).await.is_err() { + break; + } + } + + let _ = sched_task.await; +} diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 14008aa9..61d55dd3 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -38,6 +38,10 @@ pub struct WorkerHandle { registry: Arc, shutdown: Arc, heartbeat_stop: Arc, + /// The mesh node, when mesh scheduling is enabled. Held so `stop` can signal + /// its gossip + steal-server tasks and `meshClusterInfo` can read its state. + #[cfg(feature = "mesh")] + mesh: Option>, } /// Build and start a worker over `storage`, calling back into the Java @@ -56,6 +60,9 @@ fn start_worker( let queues = options .queues .unwrap_or_else(|| vec![DEFAULT_QUEUE.to_string()]); + // Mesh gossip advertises the served queues; capture before `queues` moves. + #[cfg(feature = "mesh")] + let mesh_queues = queues.clone(); let capacity = options .channel_capacity .map(|c| (c as usize).max(1)) @@ -70,6 +77,9 @@ fn start_worker( let lifecycle_storage = storage.clone(); let queues_csv = queues.join(","); let worker_id = format!("java-{}", uuid::Uuid::now_v7()); + // The mesh node id is this worker's id; capture before `worker_id` moves. + #[cfg(feature = "mesh")] + let mesh_worker_id = worker_id.clone(); let mut scheduler = Scheduler::new(storage, queues, config, namespace); // Claim execution under this worker's id so dead-worker recovery can @@ -87,8 +97,30 @@ fn start_worker( // so the result queue can't grow without bound. let (result_tx, result_rx) = crossbeam_channel::unbounded(); - // Scheduler loop: poll storage and dispatch ready jobs. + // Scheduler loop: poll storage and dispatch ready jobs. With mesh enabled and + // a config supplied, route the scheduler's output through the mesh bridge + // (affinity-sorted local deque + work-stealing) instead of straight to the + // dispatcher; the DB stays the source of truth, so the plain path is identical. let scheduler_run = scheduler.clone(); + #[cfg(feature = "mesh")] + let mesh = match options.mesh_config.as_deref() { + Some(config_json) => Some(crate::mesh::spawn_mesh( + &runtime, + scheduler_run, + job_tx, + config_json, + mesh_worker_id, + mesh_queues, + capacity as u16, + )), + None => { + runtime.spawn(async move { + scheduler_run.run(job_tx).await; + }); + None + } + }; + #[cfg(not(feature = "mesh"))] runtime.spawn(async move { scheduler_run.run(job_tx).await; }); @@ -121,6 +153,8 @@ fn start_worker( registry, shutdown, heartbeat_stop, + #[cfg(feature = "mesh")] + mesh, }) } @@ -412,10 +446,38 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_stop( let worker = unsafe { borrow_worker(handle) }; worker.shutdown.notify_one(); worker.heartbeat_stop.notify_one(); + // Signal the mesh node so its gossip + steal-server tasks exit instead of + // lingering until the runtime drops. + #[cfg(feature = "mesh")] + if let Some(mesh) = &worker.mesh { + mesh.request_shutdown(); + } Ok(()) }) } +/// `String meshClusterInfo(long workerHandle)` — a JSON `ClusterInfo` snapshot, +/// or `null` when this worker is not mesh-enabled. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorker_meshClusterInfo<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jni::sys::jstring { + guard(&mut env, std::ptr::null_mut(), |_env| { + let worker = unsafe { borrow_worker(handle) }; + #[cfg(feature = "mesh")] + if let Some(mesh) = &worker.mesh { + let json = serde_json::to_string(&mesh.cluster_info()).map_err(|e| { + crate::error::BindingError::new(format!("failed to encode cluster info: {e}")) + })?; + return crate::ffi::new_string(_env, json); + } + let _ = worker; + Ok(std::ptr::null_mut()) + }) +} + /// `void close(long workerHandle)` — stop the runtime and reclaim the handle. /// The Java side drains its handler executor before calling this, so no /// `completeJob`/`failJob` can still borrow the handle being freed. From d8b674a50f1340e62c202d9ae5d876ee941b3b63 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 29 Jun 2026 16:44:25 +0530 Subject: [PATCH 2/6] feat(java): Worker.mesh + MeshOptions/cluster info --- sdks/java/build.gradle.kts | 2 +- .../taskito/internal/JniWorkerControl.java | 6 + .../taskito/internal/NativeWorker.java | 3 + .../byteveda/taskito/spi/WorkerControl.java | 7 + .../taskito/worker/MeshClusterInfo.java | 23 +++ .../byteveda/taskito/worker/MeshOptions.java | 190 ++++++++++++++++++ .../org/byteveda/taskito/worker/Worker.java | 30 +++ 7 files changed, 260 insertions(+), 1 deletion(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/worker/MeshClusterInfo.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts index 96562437..f22c5762 100644 --- a/sdks/java/build.gradle.kts +++ b/sdks/java/build.gradle.kts @@ -117,7 +117,7 @@ val nativeStaging = layout.buildDirectory.dir("native") // Build the native library for the local platform. val cargoBuild = tasks.register("cargoBuild") { workingDir = crateDir.asFile - commandLine("cargo", "build", "--release", "--features", "postgres,redis,workflows") + commandLine("cargo", "build", "--release", "--features", "postgres,redis,workflows,mesh") } // Stage the built library under its platform-classifier resource path. diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java index b0f602fc..13f540d0 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniWorkerControl.java @@ -1,5 +1,6 @@ package org.byteveda.taskito.internal; +import java.util.Optional; import org.byteveda.taskito.spi.WorkerControl; /** JNI-backed {@link WorkerControl} over a native worker handle. */ @@ -31,6 +32,11 @@ public void stop() { NativeWorker.stop(handle); } + @Override + public Optional meshClusterInfoJson() { + return Optional.ofNullable(NativeWorker.meshClusterInfo(handle)); + } + /** Idempotent: frees the native worker handle exactly once. */ @Override public synchronized void close() { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java index af8edd68..6586a9f0 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorker.java @@ -23,4 +23,7 @@ private NativeWorker() {} public static native void stop(long handle); public static native void close(long handle); + + /** A JSON {@code ClusterInfo} snapshot, or {@code null} when not mesh-enabled. */ + public static native String meshClusterInfo(long handle); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java index 41ddb4f7..dc6b7f53 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/WorkerControl.java @@ -1,5 +1,7 @@ package org.byteveda.taskito.spi; +import java.util.Optional; + /** Controls a running worker and completes its in-flight jobs. */ public interface WorkerControl extends AutoCloseable { void completeJob(long token, byte[] result); @@ -11,6 +13,11 @@ public interface WorkerControl extends AutoCloseable { /** Stop the scheduler and heartbeat loops; in-flight jobs drain. */ void stop(); + /** A JSON {@code ClusterInfo} snapshot when mesh is enabled, else empty. */ + default Optional meshClusterInfoJson() { + return Optional.empty(); + } + @Override void close(); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/MeshClusterInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/MeshClusterInfo.java new file mode 100644 index 00000000..073a37b0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/MeshClusterInfo.java @@ -0,0 +1,23 @@ +package org.byteveda.taskito.worker; + +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * A snapshot of a mesh worker's view of its cluster, from + * {@link Worker#meshClusterInfo()}. Capacities and loads are summed across alive + * peers (excluding this node); buffered counts are job-deque lengths. + * + * @param peerCount alive peers discovered via gossip (excludes self) + * @param totalCapacity summed advertised capacity of those peers + * @param totalLoad summed in-flight jobs across those peers + * @param totalBuffered summed local-deque length across those peers + * @param localBufferLen this worker's local-deque length + * @param adaptivePrefetch this worker's current prefetch budget + */ +public record MeshClusterInfo( + @JsonProperty("peer_count") int peerCount, + @JsonProperty("total_capacity") int totalCapacity, + @JsonProperty("total_load") int totalLoad, + @JsonProperty("total_buffered") int totalBuffered, + @JsonProperty("local_buffer_len") int localBufferLen, + @JsonProperty("adaptive_prefetch") int adaptivePrefetch) {} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java new file mode 100644 index 00000000..2a607656 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/MeshOptions.java @@ -0,0 +1,190 @@ +package org.byteveda.taskito.worker; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.byteveda.taskito.errors.SerializationException; + +/** + * Mesh scheduling options for {@link Worker.Builder#mesh(MeshOptions)}. Workers + * sharing a cluster gossip over SWIM (UDP) for peer discovery and steal work + * from busy peers (TCP); the database stays the source of truth, so mesh is a + * pure throughput optimization layered under the normal worker. + * + *

Build with {@link #builder()}. The only field most clusters set is the + * seed list (and a distinct {@link Builder#port(int)} per host when co-located): + * + * 

{@code
+ * Worker w = taskito.worker()
+ *     .handle(task, handler)
+ *     .mesh(MeshOptions.builder().port(7946).seed("10.0.0.2:7946").build())
+ *     .start();
+ * }
+ */ +public final class MeshOptions { + private static final ObjectMapper JSON = new ObjectMapper(); + + private final int gossipPort; + private final List seeds; + private final String bindAddr; + private final String advertiseAddr; + private final boolean enableStealing; + private final double affinityWeight; + private final int localBufferCapacity; + private final int maxStealBatch; + private final int stealThreshold; + private final int virtualNodes; + private final int stealRateLimit; + private final String encryptionKey; + + private MeshOptions(Builder b) { + this.gossipPort = b.gossipPort; + this.seeds = List.copyOf(b.seeds); + this.bindAddr = b.bindAddr; + this.advertiseAddr = b.advertiseAddr; + this.enableStealing = b.enableStealing; + this.affinityWeight = b.affinityWeight; + this.localBufferCapacity = b.localBufferCapacity; + this.maxStealBatch = b.maxStealBatch; + this.stealThreshold = b.stealThreshold; + this.virtualNodes = b.virtualNodes; + this.stealRateLimit = b.stealRateLimit; + this.encryptionKey = b.encryptionKey; + } + + public static Builder builder() { + return new Builder(); + } + + /** + * The {@code MeshConfig} JSON the native worker reads. Every non-optional + * field is emitted (the core config has no serde defaults), with the SWIM + * protocol timings left at their core defaults. + */ + String toConfigJson() { + Map config = new LinkedHashMap<>(); + config.put("gossip_port", gossipPort); + config.put("steal_port", gossipPort + 1); + config.put("bind_addr", bindAddr); + config.put("seeds", seeds); + config.put("protocol_period_ms", 500); + config.put("indirect_ping_count", 3); + config.put("suspicion_multiplier", 4); + config.put("virtual_nodes", virtualNodes); + config.put("local_buffer_capacity", localBufferCapacity); + config.put("max_steal_batch", maxStealBatch); + config.put("steal_threshold", stealThreshold); + config.put("affinity_weight", affinityWeight); + config.put("enable_stealing", enableStealing); + config.put("steal_rate_limit", stealRateLimit); + if (advertiseAddr != null) { + config.put("advertise_addr", advertiseAddr); + } + if (encryptionKey != null) { + config.put("encryption_key", encryptionKey); + } + try { + return JSON.writeValueAsString(config); + } catch (Exception e) { + throw new SerializationException("failed to encode mesh config", e); + } + } + + /** Mirrors {@code taskito_mesh::MeshConfig} defaults; only the seed list is usually set. */ + public static final class Builder { + private int gossipPort = 7946; + private final List seeds = new ArrayList<>(); + private String bindAddr = "0.0.0.0"; + private String advertiseAddr; + private boolean enableStealing = true; + private double affinityWeight = 0.7; + private int localBufferCapacity = 64; + private int maxStealBatch = 4; + private int stealThreshold = 2; + private int virtualNodes = 150; + private int stealRateLimit = 10; + private String encryptionKey; + + /** Gossip (UDP) port; the work-stealing (TCP) port is {@code port + 1}. */ + public Builder port(int port) { + if (port < 1 || port > 65534) { + throw new IllegalArgumentException("mesh port must be in 1..=65534 (steal port is port+1)"); + } + this.gossipPort = port; + return this; + } + + /** Add a seed peer ({@code host:port}) used for initial cluster join. */ + public Builder seed(String hostPort) { + this.seeds.add(hostPort); + return this; + } + + public Builder seeds(List seeds) { + this.seeds.addAll(seeds); + return this; + } + + /** Listen address for gossip and steal (default {@code 0.0.0.0}). */ + public Builder bindAddr(String bindAddr) { + this.bindAddr = bindAddr; + return this; + } + + /** Address advertised to peers; required when {@code bindAddr} is {@code 0.0.0.0} across hosts. */ + public Builder advertiseAddr(String advertiseAddr) { + this.advertiseAddr = advertiseAddr; + return this; + } + + public Builder enableStealing(boolean enableStealing) { + this.enableStealing = enableStealing; + return this; + } + + /** 0.0 ignores affinity, 1.0 is strict affinity (default 0.7). */ + public Builder affinityWeight(double affinityWeight) { + this.affinityWeight = affinityWeight; + return this; + } + + /** Max jobs prefetched into the local deque (default 64). */ + public Builder localBufferCapacity(int localBufferCapacity) { + this.localBufferCapacity = localBufferCapacity; + return this; + } + + public Builder maxStealBatch(int maxStealBatch) { + this.maxStealBatch = maxStealBatch; + return this; + } + + public Builder stealThreshold(int stealThreshold) { + this.stealThreshold = stealThreshold; + return this; + } + + public Builder virtualNodes(int virtualNodes) { + this.virtualNodes = virtualNodes; + return this; + } + + /** Max steal requests served per peer per second; 0 is unlimited (default 10). */ + public Builder stealRateLimit(int stealRateLimit) { + this.stealRateLimit = stealRateLimit; + return this; + } + + /** Base64 (32-byte) key XOR-applied to gossip datagrams; deters casual sniffing only. */ + public Builder encryptionKey(String encryptionKey) { + this.encryptionKey = encryptionKey; + return this; + } + + public MeshOptions build() { + return new MeshOptions(this); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index 6d47f33e..310fd1a3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -10,6 +10,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -43,6 +44,7 @@ */ public final class Worker implements AutoCloseable { private static final long SHUTDOWN_TIMEOUT_SECONDS = 30; + private static final ObjectMapper MESH_JSON = new ObjectMapper(); private final WorkerControl control; private final ExecutorService executor; @@ -83,6 +85,20 @@ public void stop() { control.stop(); } + /** + * A snapshot of this worker's mesh cluster view (peer count, capacity, load), + * or empty when the worker was not started with {@link Builder#mesh}. + */ + public Optional meshClusterInfo() { + return control.meshClusterInfoJson().map(json -> { + try { + return MESH_JSON.readValue(json, MeshClusterInfo.class); + } catch (Exception e) { + throw new SerializationException("failed to decode mesh cluster info", e); + } + }); + } + /** * Approve a parked workflow gate so its successors run. Requires this worker * to be tracking workflows ({@code trackWorkflows()} on the builder). @@ -160,6 +176,7 @@ public static final class Builder { private Integer batchSize; private WorkflowTracker tracker; private AutoscaleOptions autoscale; + private MeshOptions mesh; Builder( QueueBackend backend, @@ -248,6 +265,16 @@ public Builder autoscale(AutoscaleOptions options) { return this; } + /** + * Join a scheduling mesh: discover peers via gossip and steal work from + * busy ones. The DB stays the source of truth — mesh only changes how + * ready jobs are buffered and balanced, so it is safe to add to any worker. + */ + public Builder mesh(MeshOptions options) { + this.mesh = options; + return this; + } + public Builder on(EventName name, Consumer listener) { listeners.computeIfAbsent(name, key -> new ArrayList<>()).add(listener); return this; @@ -370,6 +397,9 @@ private String encodeOptions() { if (!taskPolicies.isEmpty()) { options.put("taskConfigs", encodeTaskConfigs()); } + if (mesh != null) { + options.put("meshConfig", mesh.toConfigJson()); + } try { return JSON.writeValueAsString(options); } catch (Exception e) { From b0183714f16577e83c008cec20b755330929aef9 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 29 Jun 2026 16:44:25 +0530 Subject: [PATCH 3/6] test(java): cover mesh worker + cluster info --- .../org/byteveda/taskito/MeshWorkerTest.java | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/MeshWorkerTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/MeshWorkerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/MeshWorkerTest.java new file mode 100644 index 00000000..1f62f7c2 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/MeshWorkerTest.java @@ -0,0 +1,59 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.MeshClusterInfo; +import org.byteveda.taskito.worker.MeshOptions; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * A mesh-enabled worker runs jobs end-to-end through the mesh bridge (solo node, + * no peers) and exposes cluster info; a non-mesh worker reports no cluster. + */ +class MeshWorkerTest { + + private static final Task ECHO = Task.of("mesh.echo", String.class); + + @Test + @Timeout(60) + void meshWorkerProcessesJobs(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("mesh.db").toString()).open()) { + try (Worker worker = queue.worker() + .handle(ECHO, payload -> payload) + .mesh(MeshOptions.builder().port(17946).build()) + .start()) { + String id = queue.enqueue(ECHO, "via-mesh"); + queue.awaitJob(id, Duration.ofSeconds(30)); + assertEquals("via-mesh", queue.getResult(id, String.class).orElseThrow()); + + MeshClusterInfo info = worker.meshClusterInfo().orElseThrow(); + assertEquals(0, info.peerCount(), "a solo node has no peers"); + } + } + } + + @Test + @Timeout(60) + void nonMeshWorkerHasNoClusterInfo(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("plain.db").toString()).open()) { + try (Worker worker = queue.worker().handle(ECHO, payload -> payload).start()) { + assertTrue(worker.meshClusterInfo().isEmpty()); + } + } + } + + @Test + void rejectsOutOfRangePort() { + assertThrows(IllegalArgumentException.class, () -> MeshOptions.builder().port(70000)); + } +} From a897df6a4be5fefa02728c49c80d123acc60fc35 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 2 Jul 2026 23:10:40 +0530 Subject: [PATCH 4/6] ci: build Java native with mesh feature --- .github/workflows/ci-java.yml | 2 +- .github/workflows/publish-java.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci-java.yml b/.github/workflows/ci-java.yml index 372ad4b7..3ed11543 100644 --- a/.github/workflows/ci-java.yml +++ b/.github/workflows/ci-java.yml @@ -26,7 +26,7 @@ jobs: uses: ./.github/actions/setup-rust - name: Build the JNI cdylib - run: cargo build --release --features postgres,redis,workflows -p taskito-java + run: cargo build --release --features postgres,redis,workflows,mesh -p taskito-java - name: Upload the cdylib uses: actions/upload-artifact@v7 diff --git a/.github/workflows/publish-java.yml b/.github/workflows/publish-java.yml index 1061971c..621117e0 100644 --- a/.github/workflows/publish-java.yml +++ b/.github/workflows/publish-java.yml @@ -95,7 +95,7 @@ jobs: uses: ./.github/actions/setup-rust - name: Build cdylib - run: cargo build -p taskito-java --release --features postgres,redis,workflows + run: cargo build -p taskito-java --release --features postgres,redis,workflows,mesh - name: Stage native binary under its classifier shell: bash From d33f1d6bfa1d6d23bd8f4a413f1ea9eb9abb6a27 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 2 Jul 2026 23:20:11 +0530 Subject: [PATCH 5/6] fix(java): log mesh config parse and task failures --- crates/taskito-java/src/mesh.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/crates/taskito-java/src/mesh.rs b/crates/taskito-java/src/mesh.rs index af001502..8f87c214 100644 --- a/crates/taskito-java/src/mesh.rs +++ b/crates/taskito-java/src/mesh.rs @@ -27,7 +27,10 @@ pub fn spawn_mesh( queues: Vec, threads: u16, ) -> Arc { - let config: MeshConfig = serde_json::from_str(config_json).unwrap_or_default(); + let config: MeshConfig = serde_json::from_str(config_json).unwrap_or_else(|e| { + log::error!("[taskito-java] invalid mesh config, falling back to defaults: {e}"); + MeshConfig::default() + }); log::info!( "[taskito-java] mesh scheduling enabled (gossip_port={}, local_buffer={})", config.gossip_port, @@ -52,7 +55,13 @@ pub fn spawn_mesh( // Keep the gossip + steal-server tasks alive for the worker's lifetime; they // stop when the node's shutdown is signalled (see `request_shutdown`). runtime.spawn(async move { - let _ = tokio::join!(gossip, steal_server); + let (gossip_res, steal_res) = tokio::join!(gossip, steal_server); + if let Err(e) = gossip_res { + log::error!("[taskito-java] mesh gossip task failed: {e}"); + } + if let Err(e) = steal_res { + log::error!("[taskito-java] mesh steal-server task failed: {e}"); + } }); node From c1c688c57562107b8af16f8b039b25ea309a134d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 2 Jul 2026 23:20:24 +0530 Subject: [PATCH 6/6] fix(java): saturate mesh capacity cast to u16 --- crates/taskito-java/src/worker.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/taskito-java/src/worker.rs b/crates/taskito-java/src/worker.rs index 61d55dd3..51bfa6f2 100644 --- a/crates/taskito-java/src/worker.rs +++ b/crates/taskito-java/src/worker.rs @@ -111,7 +111,7 @@ fn start_worker( config_json, mesh_worker_id, mesh_queues, - capacity as u16, + capacity.min(u16::MAX as usize) as u16, )), None => { runtime.spawn(async move {