feat(java): static-DAG workflows + package-by-feature reorg

.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
+        run: cargo build -p taskito-java --release --features postgres,redis,workflows
 
       - name: Stage native binary under its classifier
         shell: bash

crates/taskito-java/Cargo.toml

@@ -11,11 +11,13 @@ crate-type = ["cdylib"]
 
 [features]
 default = []
-postgres = ["taskito-core/postgres"]
-redis = ["taskito-core/redis"]
+postgres = ["taskito-core/postgres", "taskito-workflows?/postgres"]
+redis = ["taskito-core/redis", "taskito-workflows?/redis"]
+workflows = ["dep:taskito-workflows"]
 
 [dependencies]
 taskito-core = { path = "../taskito-core" }
+taskito-workflows = { path = "../taskito-workflows", optional = true }
 jni = "0.21"
 serde = { workspace = true }
 serde_json = { workspace = true }

crates/taskito-java/src/backend.rs

@@ -16,6 +16,48 @@ const DEFAULT_POSTGRES_SCHEMA: &str = "taskito";
 pub struct QueueHandle {
     pub storage: StorageBackend,
     pub namespace: Option<String>,
+    /// Workflow storage, built lazily on first workflow call (its constructor
+    /// runs the workflow-table migrations). Shares this queue's connection pool.
+    #[cfg(feature = "workflows")]
+    pub workflow_storage: std::sync::OnceLock<taskito_workflows::WorkflowStorageBackend>,
+}
+
+#[cfg(feature = "workflows")]
+impl QueueHandle {
+    /// Return the workflow storage, initializing it (and its migrations) once.
+    pub fn workflow_store(
+        &self,
+    ) -> Result<taskito_workflows::WorkflowStorageBackend, BindingError> {
+        if let Some(wf) = self.workflow_storage.get() {
+            return Ok(wf.clone());
+        }
+        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)
+    }
+}
+
+/// Construct workflow storage matching this queue's core backend.
+#[cfg(feature = "workflows")]
+fn build_workflow_storage(
+    storage: &StorageBackend,
+) -> Result<taskito_workflows::WorkflowStorageBackend, BindingError> {
+    use taskito_workflows::{WorkflowSqliteStorage, WorkflowStorageBackend};
+    let wf = match storage {
+        StorageBackend::Sqlite(s) => {
+            WorkflowStorageBackend::Sqlite(WorkflowSqliteStorage::new(s.clone())?)
+        }
+        #[cfg(feature = "postgres")]
+        StorageBackend::Postgres(s) => WorkflowStorageBackend::Postgres(
+            taskito_workflows::WorkflowPostgresStorage::new(s.clone())?,
+        ),
+        #[cfg(feature = "redis")]
+        StorageBackend::Redis(s) => {
+            WorkflowStorageBackend::Redis(taskito_workflows::WorkflowRedisStorage::new(s.clone())?)
+        }
+    };
+    Ok(wf)
 }
 
 /// Error for a backend that is unknown or whose cargo feature is not compiled in.
@@ -66,5 +108,7 @@ pub fn open(options: OpenOptions) -> Result<QueueHandle, BindingError> {
     Ok(QueueHandle {
         storage,
         namespace: options.namespace,
+        #[cfg(feature = "workflows")]
+        workflow_storage: std::sync::OnceLock::new(),
     })
 }

crates/taskito-java/src/error.rs

@@ -34,3 +34,11 @@ impl From<QueueError> for BindingError {
         Self::new(err.to_string())
     }
 }
+
+/// Workflow errors (e.g. DAG validation) reach the boundary as `TaskitoException`.
+#[cfg(feature = "workflows")]
+impl From<taskito_workflows::WorkflowError> for BindingError {
+    fn from(err: taskito_workflows::WorkflowError) -> Self {
+        Self::new(err.to_string())
+    }
+}

crates/taskito-java/src/ffi.rs

@@ -78,6 +78,28 @@ pub fn read_bytes_array(
     Ok(out)
 }
 
+/// Read a Java `String[]` argument into owned strings; a null array is empty.
+#[cfg_attr(not(feature = "workflows"), allow(dead_code))]
+pub fn read_string_array(
+    env: &mut JNIEnv,
+    value: &JObjectArray,
+) -> Result<Vec<String>, BindingError> {
+    if value.is_null() {
+        return Ok(Vec::new());
+    }
+    let len = env
+        .get_array_length(value)
+        .map_err(|e| BindingError::new(format!("invalid String[] argument: {e}")))?;
+    let mut out = Vec::with_capacity(len as usize);
+    for index in 0..len {
+        let element = env
+            .get_object_array_element(value, index)
+            .map_err(|e| BindingError::new(format!("invalid String[] element: {e}")))?;
+        out.push(read_string(env, &JString::from(element))?);
+    }
+    Ok(out)
+}
+
 /// Build a Java `String` return value from an owned Rust `String`.
 pub fn new_string(env: &mut JNIEnv, value: String) -> Result<jstring, BindingError> {
     env.new_string(value)

crates/taskito-java/src/lib.rs

@@ -17,6 +17,8 @@ mod handle;
 mod jvm;
 mod queue;
 mod worker;
+#[cfg(feature = "workflows")]
+mod workflows;
 
 use std::ffi::c_void;