fix(java): pre-release hardening for 0.19.0

crates/taskito-java/src/backend.rs

@@ -1,6 +1,6 @@
 //! Construct a [`QueueHandle`] from caller options (mirrors the Node shell).
 
-use taskito_core::{SqliteStorage, StorageBackend};
+use taskito_core::{NewJob, SqliteStorage, Storage, StorageBackend};
 
 use crate::convert::OpenOptions;
 use crate::error::BindingError;
@@ -20,6 +20,11 @@ pub struct QueueHandle {
     /// runs the workflow-table migrations). Shares this queue's connection pool.
     #[cfg(feature = "workflows")]
     pub workflow_storage: std::sync::OnceLock<taskito_workflows::WorkflowStorageBackend>,
+    /// Serializes the lazy init above: the constructor runs DDL migrations, and
+    /// concurrent `CREATE TABLE/INDEX IF NOT EXISTS` from separate sessions can
+    /// race in Postgres's catalog, failing one thread's first workflow call.
+    #[cfg(feature = "workflows")]
+    workflow_init: std::sync::Mutex<()>,
 }
 
 #[cfg(feature = "workflows")]
@@ -31,8 +36,16 @@ impl QueueHandle {
         if let Some(wf) = self.workflow_storage.get() {
             return Ok(wf.clone());
         }
+        // A poisoned lock only means another thread panicked mid-init; the
+        // OnceLock still tells the truth, so recover and retry the init.
+        let _init = self
+            .workflow_init
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if let Some(wf) = self.workflow_storage.get() {
+            return Ok(wf.clone()); // another thread won the race while we waited
+        }
         let wf = build_workflow_storage(&self.storage)?;
-        // A racing thread's handle wraps the same pool, so either is fine.
         let _ = self.workflow_storage.set(wf.clone());
         Ok(wf)
     }
@@ -60,6 +73,40 @@ fn build_workflow_storage(
     Ok(wf)
 }
 
+/// Enqueue a batch, routing any job with a `unique_key` through the dedup path
+/// (matching single `enqueue`) instead of the raw batch insert. The raw insert
+/// would hit the partial unique index on an active duplicate and roll back the
+/// whole batch; here a duplicate resolves to the existing job's id, and only
+/// keyless jobs share the one-transaction fast path. Ids come back in input order.
+pub fn enqueue_batch_dedup(
+    storage: &StorageBackend,
+    jobs: Vec<NewJob>,
+) -> Result<Vec<String>, BindingError> {
+    let mut ids: Vec<Option<String>> = std::iter::repeat_with(|| None).take(jobs.len()).collect();
+    let mut plain_jobs = Vec::new();
+    let mut plain_slots = Vec::new();
+    for (index, job) in jobs.into_iter().enumerate() {
+        if job.unique_key.is_some() {
+            ids[index] = Some(storage.enqueue_unique(job)?.id);
+        } else {
+            plain_jobs.push(job);
+            plain_slots.push(index);
+        }
+    }
+    if !plain_jobs.is_empty() {
+        let created = storage.enqueue_batch(plain_jobs)?;
+        if created.len() != plain_slots.len() {
+            return Err(BindingError::new(
+                "storage returned a different number of jobs than were batch-enqueued",
+            ));
+        }
+        for (slot, job) in plain_slots.into_iter().zip(created) {
+            ids[slot] = Some(job.id);
+        }
+    }
+    Ok(ids.into_iter().flatten().collect())
+}
+
 /// Error for a backend that is unknown or whose cargo feature is not compiled in.
 fn unknown_backend(name: &str) -> BindingError {
     BindingError::new(format!(
@@ -110,5 +157,53 @@ pub fn open(options: OpenOptions) -> Result<QueueHandle, BindingError> {
         namespace: options.namespace,
         #[cfg(feature = "workflows")]
         workflow_storage: std::sync::OnceLock::new(),
+        #[cfg(feature = "workflows")]
+        workflow_init: std::sync::Mutex::new(()),
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn new_job(unique_key: Option<&str>) -> NewJob {
+        NewJob {
+            queue: "default".into(),
+            task_name: "task".into(),
+            payload: Vec::new(),
+            priority: 0,
+            scheduled_at: 0,
+            max_retries: 0,
+            timeout_ms: 0,
+            unique_key: unique_key.map(str::to_owned),
+            metadata: None,
+            notes: None,
+            depends_on: Vec::new(),
+            expires_at: None,
+            result_ttl_ms: None,
+            namespace: None,
+        }
+    }
+
+    /// A batch containing an active duplicate `unique_key` must dedup that job
+    /// to the existing id instead of failing the whole batch on the unique index.
+    #[test]
+    fn batch_dedup_resolves_duplicate_unique_keys() {
+        let storage = StorageBackend::Sqlite(SqliteStorage::in_memory().expect("storage"));
+        let existing = storage
+            .enqueue_unique(new_job(Some("k1")))
+            .expect("seed enqueue");
+        let ids = enqueue_batch_dedup(
+            &storage,
+            vec![new_job(Some("k1")), new_job(Some("k2")), new_job(None)],
+        )
+        .expect("batch enqueue");
+        assert_eq!(ids.len(), 3);
+        assert_eq!(
+            ids[0], existing.id,
+            "duplicate key resolves to existing job"
+        );
+        assert_ne!(ids[1], ids[0]);
+        assert_ne!(ids[2], ids[0]);
+    }
+}

crates/taskito-java/src/error.rs

@@ -8,6 +8,7 @@ pub const TASKITO_EXCEPTION: &str = "org/byteveda/taskito/TaskitoException";
 
 /// A deferred Java exception. The binding builds one of these on failure and
 /// throws it at the FFI boundary instead of unwinding a Rust panic across it.
+#[derive(Debug)]
 pub struct BindingError {
     class: &'static str,
     message: String,
@@ -29,6 +30,14 @@ impl BindingError {
     }
 }
 
+/// The message alone: the C-ABI surface (`ffi_c`) reports errors as strings,
+/// with no exception class to carry.
+impl std::fmt::Display for BindingError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.message)
+    }
+}
+
 impl From<QueueError> for BindingError {
     fn from(err: QueueError) -> Self {
         Self::new(err.to_string())

crates/taskito-java/src/ffi_c.rs

@@ -190,11 +190,9 @@ pub unsafe extern "C" fn taskito_ffi_enqueue_many(
             .zip(option_list)
             .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace))
             .collect();
-        let created = queue
-            .storage
-            .enqueue_batch(new_jobs)
+        let ids = crate::backend::enqueue_batch_dedup(&queue.storage, new_jobs)
             .map_err(|e| e.to_string())?;
-        let ids: Vec<Vec<u8>> = created.into_iter().map(|job| job.id.into_bytes()).collect();
+        let ids: Vec<Vec<u8>> = ids.into_iter().map(String::into_bytes).collect();
         Ok::<Vec<u8>, String>(frame(&ids))
     });
     finish(catch_unwind(work), out_data, out_len)

crates/taskito-java/src/queue/mod.rs

@@ -93,8 +93,11 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueue<'l
 }
 
 /// `String[] enqueueMany(long handle, String taskName, byte[][] payloads, String optionsJson)`
-/// — enqueue a batch in one storage transaction. `optionsJson` is a JSON array
-/// of per-job options, the same length as `payloads`. Returns the new job ids.
+/// — enqueue a batch. `optionsJson` is a JSON array of per-job options, the same
+/// length as `payloads`. Keyless jobs insert in one storage transaction; a job
+/// with a `uniqueKey` takes the dedup path instead (an active duplicate resolves
+/// to the existing job's id, matching single `enqueue`). Returns the job ids in
+/// input order.
 #[no_mangle]
 pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueueMany<'local>(
     mut env: JNIEnv<'local>,
@@ -121,8 +124,7 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_enqueueMan
             .zip(option_list)
             .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace))
             .collect();
-        let created = queue.storage.enqueue_batch(new_jobs)?;
-        let ids: Vec<String> = created.into_iter().map(|job| job.id).collect();
+        let ids = backend::enqueue_batch_dedup(&queue.storage, new_jobs)?;
         new_string_array(env, &ids)
     })
 }

sdks/java/processor/build.gradle.kts

@@ -5,11 +5,40 @@ plugins {
     `java-library`
     checkstyle
     id("com.diffplug.spotless") version "7.2.1"
+    id("com.vanniktech.maven.publish") version "0.37.0"
 }
 
 group = "org.byteveda"
 version = "0.18.0"
 
+mavenPublishing {
+    publishToMavenCentral()
+    signAllPublications()
+    coordinates(group.toString(), "taskito-processor", version.toString())
+    pom {
+        name.set("Taskito Processor")
+        description.set("Compile-time annotation processor generating task-handler bindings for the Taskito JVM SDK.")
+        url.set("https://github.com/ByteVeda/taskito")
+        licenses {
+            license {
+                name.set("MIT")
+                url.set("https://opensource.org/licenses/MIT")
+            }
+        }
+        developers {
+            developer {
+                id.set("byteveda")
+                name.set("ByteVeda")
+            }
+        }
+        scm {
+            url.set("https://github.com/ByteVeda/taskito")
+            connection.set("scm:git:https://github.com/ByteVeda/taskito.git")
+            developerConnection.set("scm:git:ssh://git@github.com/ByteVeda/taskito.git")
+        }
+    }
+}
+
 java {
     sourceCompatibility = JavaVersion.VERSION_17
     targetCompatibility = JavaVersion.VERSION_17

sdks/java/spring/build.gradle.kts

@@ -4,12 +4,41 @@ plugins {
     `java-library`
     checkstyle
     id("com.diffplug.spotless") version "7.2.1"
+    id("com.vanniktech.maven.publish") version "0.37.0"
 }
 
 group = "org.byteveda"
 
 version = "0.18.0"
 
+mavenPublishing {
+    publishToMavenCentral()
+    signAllPublications()
+    coordinates(group.toString(), "taskito-spring", version.toString())
+    pom {
+        name.set("Taskito Spring")
+        description.set("Spring Boot 3 starter that auto-configures a Taskito bean.")
+        url.set("https://github.com/ByteVeda/taskito")
+        licenses {
+            license {
+                name.set("MIT")
+                url.set("https://opensource.org/licenses/MIT")
+            }
+        }
+        developers {
+            developer {
+                id.set("byteveda")
+                name.set("ByteVeda")
+            }
+        }
+        scm {
+            url.set("https://github.com/ByteVeda/taskito")
+            connection.set("scm:git:https://github.com/ByteVeda/taskito.git")
+            developerConnection.set("scm:git:ssh://git@github.com/ByteVeda/taskito.git")
+        }
+    }
+}
+
 java {
     sourceCompatibility = JavaVersion.VERSION_17
     targetCompatibility = JavaVersion.VERSION_17

sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java

@@ -5,9 +5,12 @@
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
+import org.byteveda.taskito.logging.TaskitoLogger;
 
 /** Dispatches {@link OutcomeEvent}s to registered listeners. Thread-safe. */
 public final class Emitter {
+    private static final TaskitoLogger LOG = TaskitoLogger.create("events");
+
     private final Map<EventName, List<Consumer<OutcomeEvent>>> listeners = new EnumMap<>(EventName.class);
 
     public void on(EventName name, Consumer<OutcomeEvent> listener) {
@@ -23,8 +26,10 @@ public void emit(OutcomeEvent event) {
         for (Consumer<OutcomeEvent> listener : bound) {
             try {
                 listener.accept(event);
-            } catch (RuntimeException ignored) {
-                // A listener fault must not break event dispatch.
+            } catch (RuntimeException e) {
+                // A listener fault must not break dispatch — but log it: this is
+                // the only place a workflow-tracker failure would surface.
+                LOG.warn("listener for " + event.name + " (job " + event.jobId + ") threw", e);
             }
         }
     }