feat(java): add Panama FFM fast-path for hot byte ops

.github/workflows/publish-java.yml

@@ -129,7 +129,9 @@ jobs:
         uses: actions/setup-java@v5
         with:
           distribution: temurin
-          java-version: "21"
+          # 22+ so the jar task compiles the FFM overlay into META-INF/versions/22
+          # (see build.gradle.kts ffmCapable); a 17/21 build ships JNI-only.
+          java-version: "25"
 
       - name: Build dashboard SPA
         uses: ./.github/actions/dashboard-build

crates/taskito-java/src/ffi_c.rs

@@ -0,0 +1,270 @@
+//! C-ABI fast path for the hot byte ops, parallel to the jni-rs surface.
+//!
+//! These `extern "C"` exports let the Java SDK call `enqueue`, `enqueueMany`, and
+//! `getResult` over Project Panama (FFM) on JDK 22+, avoiding the JNI string and
+//! `byte[]` marshalling. They are perf-only: the JNI entry points in [`crate::queue`]
+//! remain the control plane and the fallback on JDKs without FFM. Both surfaces
+//! share one cdylib and operate on the same opaque `QueueHandle` pointer.
+//!
+//! ## Contract
+//! Each function returns an `i32` status — [`OK`], [`ERR`], or (results only)
+//! [`ABSENT`] — and, when it produces bytes, writes a freshly heap-allocated
+//! buffer's pointer and length through the `out_data`/`out_len` out-params. The
+//! caller copies the bytes and must release the buffer with [`taskito_ffi_free`].
+//! On [`ERR`] the buffer holds a UTF-8 error message. Inputs are borrowed for the
+//! duration of the call and never freed here. Every body is panic-guarded so a
+//! Rust panic can never unwind across the FFI boundary.
+
+use std::panic::{catch_unwind, AssertUnwindSafe};
+use std::slice;
+
+use taskito_core::Storage;
+
+use crate::backend::QueueHandle;
+use crate::convert::{build_new_job, EnqueueOptions};
+
+/// The call succeeded; `out` holds the result bytes (job id, ids frame, or result).
+pub const OK: i32 = 0;
+/// The call failed; `out` holds a UTF-8 error message.
+pub const ERR: i32 = 1;
+/// `getResult` only: the job has no result yet; `out` is null/empty.
+pub const ABSENT: i32 = 2;
+
+/// Borrow `len` bytes at `ptr` as an owned `Vec`; a null or empty input is empty.
+///
+/// # Safety
+/// `ptr` must be valid for `len` bytes for the duration of the call, or null.
+unsafe fn read_bytes(ptr: *const u8, len: usize) -> Vec<u8> {
+    if ptr.is_null() || len == 0 {
+        Vec::new()
+    } else {
+        slice::from_raw_parts(ptr, len).to_vec()
+    }
+}
+
+/// Borrow `len` bytes at `ptr` as a UTF-8 `String`.
+///
+/// # Safety
+/// See [`read_bytes`].
+unsafe fn read_string(ptr: *const u8, len: usize) -> Result<String, String> {
+    String::from_utf8(read_bytes(ptr, len)).map_err(|e| format!("invalid UTF-8 argument: {e}"))
+}
+
+/// Hand `bytes` to the caller: write its pointer and length to the out-params and
+/// leak the allocation, to be reclaimed by [`taskito_ffi_free`].
+///
+/// # Safety
+/// `out_data` and `out_len` must be valid, writable pointers.
+unsafe fn emit(bytes: Vec<u8>, out_data: *mut *mut u8, out_len: *mut usize) {
+    let mut boxed = bytes.into_boxed_slice();
+    *out_len = boxed.len();
+    *out_data = boxed.as_mut_ptr();
+    std::mem::forget(boxed);
+}
+
+/// Resolve a producer result into a status code, emitting bytes or the error text.
+///
+/// # Safety
+/// `out_data`/`out_len` must be valid, writable pointers.
+unsafe fn finish(
+    result: std::thread::Result<Result<Vec<u8>, String>>,
+    out_data: *mut *mut u8,
+    out_len: *mut usize,
+) -> i32 {
+    match result {
+        Ok(Ok(bytes)) => {
+            emit(bytes, out_data, out_len);
+            OK
+        }
+        Ok(Err(message)) => {
+            emit(message.into_bytes(), out_data, out_len);
+            ERR
+        }
+        Err(_) => {
+            emit(b"taskito native panic".to_vec(), out_data, out_len);
+            ERR
+        }
+    }
+}
+
+/// Length-prefixed frame: `[count u32][len u32][bytes]...`, all little-endian.
+/// Matches the Java side's framing so batch payloads and ids cross as one buffer.
+fn frame(items: &[Vec<u8>]) -> Vec<u8> {
+    let total = 4 + items.iter().map(|item| 4 + item.len()).sum::<usize>();
+    let mut out = Vec::with_capacity(total);
+    out.extend_from_slice(&(items.len() as u32).to_le_bytes());
+    for item in items {
+        out.extend_from_slice(&(item.len() as u32).to_le_bytes());
+        out.extend_from_slice(item);
+    }
+    out
+}
+
+/// Decode a [`frame`]d buffer back into its elements.
+fn unframe(buf: &[u8]) -> Result<Vec<Vec<u8>>, String> {
+    let mut pos = 0usize;
+    let take_u32 = |buf: &[u8], pos: &mut usize| -> Result<usize, String> {
+        let end = pos.checked_add(4).ok_or("frame length overflow")?;
+        let slice = buf.get(*pos..end).ok_or("truncated frame header")?;
+        *pos = end;
+        Ok(u32::from_le_bytes(slice.try_into().unwrap()) as usize)
+    };
+    let count = take_u32(buf, &mut pos)?;
+    let mut items = Vec::with_capacity(count);
+    for _ in 0..count {
+        let len = take_u32(buf, &mut pos)?;
+        let end = pos.checked_add(len).ok_or("frame element overflow")?;
+        let element = buf.get(pos..end).ok_or("truncated frame element")?;
+        items.push(element.to_vec());
+        pos = end;
+    }
+    Ok(items)
+}
+
+/// `enqueue` — mirrors `NativeQueue.enqueue`. On success `out` is the job id.
+///
+/// # Safety
+/// `handle` must be a live `QueueHandle` pointer; the input pointers must be valid
+/// for their lengths; `out_data`/`out_len` must be writable.
+#[no_mangle]
+pub unsafe extern "C" fn taskito_ffi_enqueue(
+    handle: i64,
+    task_ptr: *const u8,
+    task_len: usize,
+    payload_ptr: *const u8,
+    payload_len: usize,
+    options_ptr: *const u8,
+    options_len: usize,
+    out_data: *mut *mut u8,
+    out_len: *mut usize,
+) -> i32 {
+    let work = AssertUnwindSafe(|| {
+        let queue = handle_ref(handle);
+        let task = read_string(task_ptr, task_len)?;
+        let payload = read_bytes(payload_ptr, payload_len);
+        let raw_options = read_string(options_ptr, options_len)?;
+        let options: EnqueueOptions = parse_options(&raw_options, "enqueue options")?;
+        let unique = options.unique_key.is_some();
+        let new_job = build_new_job(task, payload, options, queue.namespace.as_deref());
+        let job = if unique {
+            queue.storage.enqueue_unique(new_job)
+        } else {
+            queue.storage.enqueue(new_job)
+        }
+        .map_err(|e| e.to_string())?;
+        Ok::<Vec<u8>, String>(job.id.into_bytes())
+    });
+    finish(catch_unwind(work), out_data, out_len)
+}
+
+/// `enqueueMany` — mirrors `NativeQueue.enqueueMany`. `payloads` is a [`frame`]d
+/// buffer; `options` is a JSON array. On success `out` is a frame of job ids.
+///
+/// # Safety
+/// See [`taskito_ffi_enqueue`].
+#[no_mangle]
+pub unsafe extern "C" fn taskito_ffi_enqueue_many(
+    handle: i64,
+    task_ptr: *const u8,
+    task_len: usize,
+    payloads_ptr: *const u8,
+    payloads_len: usize,
+    options_ptr: *const u8,
+    options_len: usize,
+    out_data: *mut *mut u8,
+    out_len: *mut usize,
+) -> i32 {
+    let work = AssertUnwindSafe(|| {
+        let queue = handle_ref(handle);
+        let task = read_string(task_ptr, task_len)?;
+        let payloads = unframe(&read_bytes(payloads_ptr, payloads_len))?;
+        let raw_options = read_string(options_ptr, options_len)?;
+        let option_list: Vec<EnqueueOptions> =
+            parse_options(&raw_options, "enqueue batch options")?;
+        if option_list.len() != payloads.len() {
+            return Err("payloads and options must have the same length".to_string());
+        }
+        let namespace = queue.namespace.as_deref();
+        let new_jobs = payloads
+            .into_iter()
+            .zip(option_list)
+            .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace))
+            .collect();
+        let created = queue
+            .storage
+            .enqueue_batch(new_jobs)
+            .map_err(|e| e.to_string())?;
+        let ids: Vec<Vec<u8>> = created.into_iter().map(|job| job.id.into_bytes()).collect();
+        Ok::<Vec<u8>, String>(frame(&ids))
+    });
+    finish(catch_unwind(work), out_data, out_len)
+}
+
+/// `getResult` — mirrors `NativeQueue.getResult`. Returns [`ABSENT`] (with a null
+/// `out`) when the job has no result, [`OK`] with the result bytes, or [`ERR`].
+///
+/// # Safety
+/// See [`taskito_ffi_enqueue`].
+#[no_mangle]
+pub unsafe extern "C" fn taskito_ffi_get_result(
+    handle: i64,
+    job_ptr: *const u8,
+    job_len: usize,
+    out_data: *mut *mut u8,
+    out_len: *mut usize,
+) -> i32 {
+    let work = AssertUnwindSafe(|| {
+        let queue = handle_ref(handle);
+        let id = read_string(job_ptr, job_len)?;
+        let result = queue
+            .storage
+            .get_job(&id)
+            .map_err(|e| e.to_string())?
+            .and_then(|job| job.result);
+        Ok::<Option<Vec<u8>>, String>(result)
+    });
+    match catch_unwind(work) {
+        Ok(Ok(Some(bytes))) => {
+            emit(bytes, out_data, out_len);
+            OK
+        }
+        Ok(Ok(None)) => {
+            *out_data = std::ptr::null_mut();
+            *out_len = 0;
+            ABSENT
+        }
+        Ok(Err(message)) => {
+            emit(message.into_bytes(), out_data, out_len);
+            ERR
+        }
+        Err(_) => {
+            emit(b"taskito native panic".to_vec(), out_data, out_len);
+            ERR
+        }
+    }
+}
+
+/// Reclaim a buffer previously handed out by one of the producer functions.
+///
+/// # Safety
+/// `ptr`/`len` must be a pair produced by [`emit`] (i.e. returned through an
+/// `out_data`/`out_len` pair), and freed at most once.
+#[no_mangle]
+pub unsafe extern "C" fn taskito_ffi_free(ptr: *mut u8, len: usize) {
+    if !ptr.is_null() && len != 0 {
+        drop(Box::from_raw(std::ptr::slice_from_raw_parts_mut(ptr, len)));
+    }
+}
+
+/// Borrow the queue behind a handle.
+///
+/// # Safety
+/// `handle` must be a live `QueueHandle` pointer (see [`crate::handle`]).
+unsafe fn handle_ref<'a>(handle: i64) -> &'a QueueHandle {
+    crate::handle::borrow::<QueueHandle>(handle)
+}
+
+/// Parse JSON options without the JNI-flavoured `BindingError` (no env to throw on).
+fn parse_options<'a, T: serde::Deserialize<'a>>(raw: &'a str, what: &str) -> Result<T, String> {
+    serde_json::from_str(raw).map_err(|e| format!("invalid {what} JSON: {e}"))
+}

crates/taskito-java/src/lib.rs

@@ -13,6 +13,7 @@ mod convert;
 mod dispatcher;
 mod error;
 mod ffi;
+mod ffi_c;
 mod handle;
 mod jvm;
 mod queue;

sdks/java/README.md

@@ -379,3 +379,10 @@ development against a freshly built library, point the loader at it directly:
 ```bash
 -Dtaskito.native.lib=/abs/path/to/target/release/libtaskito_java.so
 ```
+
+On JDK 22+ the hot byte ops use a Project Panama (FFM) fast path; older JDKs (or
+a jar built without the overlay) transparently fall back to JNI. FFM calls a
+restricted native method, so a future JDK will deny it by default. The jar's
+`Enable-Native-Access: ALL-UNNAMED` manifest attribute only covers `java -jar`;
+apps that use the SDK as a classpath dependency should launch with
+`--enable-native-access=ALL-UNNAMED` to grant access and silence the warning.

sdks/java/build.gradle.kts

@@ -136,6 +136,49 @@ tasks.named("processResources") { dependsOn(copyNative) }
 // dependency lazily — sourcesJar is registered after this script evaluates.
 tasks.withType<Jar>().configureEach { dependsOn(copyNative) }
 
+// --- FFM fast-path overlay (Multi-Release JAR) ----------------------------
+// Base classes target 17 (JNI transport + the fallback). On a build JDK >= 22 we
+// also compile the Project Panama (FFM) transport at --release 22 and package it
+// under META-INF/versions/22; the runtime selects it on JDK 22+ (see
+// NativeTransport.create), else stays on JNI. Older build JDKs simply omit the
+// overlay — same public API, faster impl where available (not feature divergence).
+val ffmCapable = JavaVersion.current() >= JavaVersion.VERSION_22
+
+if (ffmCapable) {
+    val java22 by sourceSets.creating {
+        java.srcDir("src/main/java22")
+        compileClasspath += sourceSets["main"].output
+        runtimeClasspath += sourceSets["main"].output
+    }
+
+    tasks.named<JavaCompile>("compileJava22Java") {
+        options.release.set(22)
+    }
+
+    tasks.named<Jar>("jar") {
+        manifest {
+            attributes(
+                "Multi-Release" to "true",
+                // Only takes effect when the jar is run directly (java -jar); it does
+                // NOT cover consumers that depend on the SDK on their classpath — they
+                // must pass --enable-native-access=ALL-UNNAMED themselves (see README).
+                // Restricted FFM methods only warn today but a future JDK denies them.
+                "Enable-Native-Access" to "ALL-UNNAMED",
+            )
+        }
+        into("META-INF/versions/22") { from(java22.output) }
+    }
+
+    // Exercise the FFM transport in the test suite on this JDK 22+ build. Set on
+    // the test task directly: mutating the source set's runtimeClasspath here is
+    // too late (the java plugin has already captured the test task's classpath).
+    tasks.named<Test>("test") {
+        classpath += java22.output
+        // Silence (and forward-proof against) the restricted-native-access warning.
+        jvmArgs("--enable-native-access=ALL-UNNAMED")
+    }
+}
+
 tasks.test {
     useJUnitPlatform()
 }

sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java

@@ -8,10 +8,15 @@
 /** JNI-backed {@link QueueBackend} over a native queue handle. */
 public final class JniQueueBackend implements QueueBackend {
     private final long handle;
+
+    /** Hot byte ops (enqueue/enqueueMany/getResult) route through this; FFM when available, else JNI. */
+    private final NativeTransport transport;
+
     private volatile boolean closed;
 
     private JniQueueBackend(long handle) {
         this.handle = handle;
+        this.transport = NativeTransport.create(handle);
     }
 
     /** Open a native backend from its JSON options. */
@@ -21,12 +26,12 @@ public static JniQueueBackend open(String optionsJson) {
 
     @Override
     public String enqueue(String taskName, byte[] payload, String optionsJson) {
-        return NativeQueue.enqueue(handle, taskName, payload, optionsJson);
+        return transport.enqueue(taskName, payload, optionsJson);
     }
 
     @Override
     public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) {
-        return NativeQueue.enqueueMany(handle, taskName, payloads, optionsJson);
+        return transport.enqueueMany(taskName, payloads, optionsJson);
     }
 
     @Override
@@ -36,7 +41,7 @@ public Optional<String> getJobJson(String jobId) {
 
     @Override
     public Optional<byte[]> getResult(String jobId) {
-        return Optional.ofNullable(NativeQueue.getResult(handle, jobId));
+        return Optional.ofNullable(transport.getResult(jobId));
     }
 
     @Override