diff --git a/crates/taskito-java/src/backend.rs b/crates/taskito-java/src/backend.rs index 94e77a99..955927e2 100644 --- a/crates/taskito-java/src/backend.rs +++ b/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, + /// 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, +) -> Result, BindingError> { + let mut ids: Vec> = 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 { 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]); + } +} diff --git a/crates/taskito-java/src/error.rs b/crates/taskito-java/src/error.rs index 715b950f..e21872cb 100644 --- a/crates/taskito-java/src/error.rs +++ b/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 for BindingError { fn from(err: QueueError) -> Self { Self::new(err.to_string()) diff --git a/crates/taskito-java/src/ffi_c.rs b/crates/taskito-java/src/ffi_c.rs index ac2463bf..11c5cbd2 100644 --- a/crates/taskito-java/src/ffi_c.rs +++ b/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> = created.into_iter().map(|job| job.id.into_bytes()).collect(); + let ids: Vec> = ids.into_iter().map(String::into_bytes).collect(); Ok::, String>(frame(&ids)) }); finish(catch_unwind(work), out_data, out_len) diff --git a/crates/taskito-java/src/queue/mod.rs b/crates/taskito-java/src/queue/mod.rs index f162b0bc..c5fc0a70 100644 --- a/crates/taskito-java/src/queue/mod.rs +++ b/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 = created.into_iter().map(|job| job.id).collect(); + let ids = backend::enqueue_batch_dedup(&queue.storage, new_jobs)?; new_string_array(env, &ids) }) } diff --git a/sdks/java/processor/build.gradle.kts b/sdks/java/processor/build.gradle.kts index 4e53dc47..c1bbb822 100644 --- a/sdks/java/processor/build.gradle.kts +++ b/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 diff --git a/sdks/java/spring/build.gradle.kts b/sdks/java/spring/build.gradle.kts index 8e15e366..ae028ba0 100644 --- a/sdks/java/spring/build.gradle.kts +++ b/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 diff --git a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java b/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java index 8e40df08..adad46c7 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/events/Emitter.java +++ b/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>> listeners = new EnumMap<>(EventName.class); public void on(EventName name, Consumer listener) { @@ -23,8 +26,10 @@ public void emit(OutcomeEvent event) { for (Consumer 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); } } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index 3491fb33..47647f5a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -1,18 +1,26 @@ package org.byteveda.taskito.internal; import java.util.Optional; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; import org.byteveda.taskito.spi.QueueBackend; import org.byteveda.taskito.spi.WorkerBridge; import org.byteveda.taskito.spi.WorkerControl; -/** JNI-backed {@link QueueBackend} over a native queue handle. */ +/** + * JNI-backed {@link QueueBackend} over a native queue handle. Every call holds + * the read lock; {@code close()} takes the write lock, so it waits out in-flight + * calls and later calls throw {@link IllegalStateException} instead of touching + * freed native memory. + */ 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 final ReentrantReadWriteLock stateLock = new ReentrantReadWriteLock(); + private boolean closed; // guarded by stateLock private JniQueueBackend(long handle) { this.handle = handle; @@ -24,200 +32,228 @@ public static JniQueueBackend open(String optionsJson) { return new JniQueueBackend(NativeQueue.open(optionsJson)); } + private T withOpenHandle(Supplier nativeCall) { + stateLock.readLock().lock(); + try { + if (closed) { + throw new IllegalStateException("queue backend is closed"); + } + return nativeCall.get(); + } finally { + stateLock.readLock().unlock(); + } + } + @Override public String enqueue(String taskName, byte[] payload, String optionsJson) { - return transport.enqueue(taskName, payload, optionsJson); + return withOpenHandle(() -> transport.enqueue(taskName, payload, optionsJson)); } @Override public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) { - return transport.enqueueMany(taskName, payloads, optionsJson); + return withOpenHandle(() -> transport.enqueueMany(taskName, payloads, optionsJson)); } @Override public Optional getJobJson(String jobId) { - return Optional.ofNullable(NativeQueue.getJob(handle, jobId)); + return withOpenHandle(() -> Optional.ofNullable(NativeQueue.getJob(handle, jobId))); } @Override public Optional getResult(String jobId) { - return Optional.ofNullable(transport.getResult(jobId)); + return withOpenHandle(() -> Optional.ofNullable(transport.getResult(jobId))); } @Override public boolean cancel(String jobId) { - return NativeQueue.cancel(handle, jobId); + return withOpenHandle(() -> NativeQueue.cancel(handle, jobId)); } @Override public boolean requestCancel(String jobId) { - return NativeQueue.requestCancel(handle, jobId); + return withOpenHandle(() -> NativeQueue.requestCancel(handle, jobId)); } @Override public boolean isCancelRequested(String jobId) { - return NativeQueue.isCancelRequested(handle, jobId); + return withOpenHandle(() -> NativeQueue.isCancelRequested(handle, jobId)); } @Override public void setProgress(String jobId, int progress) { - NativeQueue.setProgress(handle, jobId, progress); + withOpenHandle(() -> { + NativeQueue.setProgress(handle, jobId, progress); + return null; + }); } @Override public String statsJson() { - return NativeQueue.stats(handle); + return withOpenHandle(() -> NativeQueue.stats(handle)); } @Override public String statsByQueueJson(String queue) { - return NativeQueue.statsByQueue(handle, queue); + return withOpenHandle(() -> NativeQueue.statsByQueue(handle, queue)); } @Override public String statsAllQueuesJson() { - return NativeQueue.statsAllQueues(handle); + return withOpenHandle(() -> NativeQueue.statsAllQueues(handle)); } @Override public String listJobsJson(String filterJson) { - return NativeQueue.listJobs(handle, filterJson); + return withOpenHandle(() -> NativeQueue.listJobs(handle, filterJson)); } @Override public String jobErrorsJson(String jobId) { - return NativeQueue.jobErrors(handle, jobId); + return withOpenHandle(() -> NativeQueue.jobErrors(handle, jobId)); } @Override public String metricsJson(String taskNameOrNull, long sinceMs) { - return NativeQueue.metrics(handle, taskNameOrNull, sinceMs); + return withOpenHandle(() -> NativeQueue.metrics(handle, taskNameOrNull, sinceMs)); } @Override public String listWorkersJson() { - return NativeQueue.listWorkers(handle); + return withOpenHandle(() -> NativeQueue.listWorkers(handle)); } @Override public String listDeadJson(long limit, long offset) { - return NativeQueue.listDead(handle, limit, offset); + return withOpenHandle(() -> NativeQueue.listDead(handle, limit, offset)); } @Override public String retryDead(String deadId) { - return NativeQueue.retryDead(handle, deadId); + return withOpenHandle(() -> NativeQueue.retryDead(handle, deadId)); } @Override public boolean deleteDead(String deadId) { - return NativeQueue.deleteDead(handle, deadId); + return withOpenHandle(() -> NativeQueue.deleteDead(handle, deadId)); } @Override public long purgeDead(long olderThanMs) { - return NativeQueue.purgeDead(handle, olderThanMs); + return withOpenHandle(() -> NativeQueue.purgeDead(handle, olderThanMs)); } @Override public String listDeadByTaskJson(String taskName, long limit, long offset) { - return NativeQueue.listDeadByTask(handle, taskName, limit, offset); + return withOpenHandle(() -> NativeQueue.listDeadByTask(handle, taskName, limit, offset)); } @Override public long purgeDeadByTask(String taskName) { - return NativeQueue.purgeDeadByTask(handle, taskName); + return withOpenHandle(() -> NativeQueue.purgeDeadByTask(handle, taskName)); } @Override public long purgeCompleted(long olderThanMs) { - return NativeQueue.purgeCompleted(handle, olderThanMs); + return withOpenHandle(() -> NativeQueue.purgeCompleted(handle, olderThanMs)); } @Override public void pauseQueue(String queue) { - NativeQueue.pauseQueue(handle, queue); + withOpenHandle(() -> { + NativeQueue.pauseQueue(handle, queue); + return null; + }); } @Override public void resumeQueue(String queue) { - NativeQueue.resumeQueue(handle, queue); + withOpenHandle(() -> { + NativeQueue.resumeQueue(handle, queue); + return null; + }); } @Override public String listPausedQueuesJson() { - return NativeQueue.listPausedQueues(handle); + return withOpenHandle(() -> NativeQueue.listPausedQueues(handle)); } @Override public Optional getSetting(String key) { - return Optional.ofNullable(NativeQueue.getSetting(handle, key)); + return withOpenHandle(() -> Optional.ofNullable(NativeQueue.getSetting(handle, key))); } @Override public void setSetting(String key, String value) { - NativeQueue.setSetting(handle, key, value); + withOpenHandle(() -> { + NativeQueue.setSetting(handle, key, value); + return null; + }); } @Override public boolean deleteSetting(String key) { - return NativeQueue.deleteSetting(handle, key); + return withOpenHandle(() -> NativeQueue.deleteSetting(handle, key)); } @Override public String listSettingsJson() { - return NativeQueue.listSettings(handle); + return withOpenHandle(() -> NativeQueue.listSettings(handle)); } @Override public void writeTaskLog(String jobId, String taskName, String level, String message, String extraOrNull) { - NativeQueue.writeTaskLog(handle, jobId, taskName, level, message, extraOrNull); + withOpenHandle(() -> { + NativeQueue.writeTaskLog(handle, jobId, taskName, level, message, extraOrNull); + return null; + }); } @Override public String getTaskLogsJson(String jobId) { - return NativeQueue.getTaskLogs(handle, jobId); + return withOpenHandle(() -> NativeQueue.getTaskLogs(handle, jobId)); } @Override public boolean acquireLock(String name, String ownerId, long ttlMs) { - return NativeQueue.acquireLock(handle, name, ownerId, ttlMs); + return withOpenHandle(() -> NativeQueue.acquireLock(handle, name, ownerId, ttlMs)); } @Override public boolean releaseLock(String name, String ownerId) { - return NativeQueue.releaseLock(handle, name, ownerId); + return withOpenHandle(() -> NativeQueue.releaseLock(handle, name, ownerId)); } @Override public boolean extendLock(String name, String ownerId, long ttlMs) { - return NativeQueue.extendLock(handle, name, ownerId, ttlMs); + return withOpenHandle(() -> NativeQueue.extendLock(handle, name, ownerId, ttlMs)); } @Override public Optional lockInfoJson(String name) { - return Optional.ofNullable(NativeQueue.getLockInfo(handle, name)); + return withOpenHandle(() -> Optional.ofNullable(NativeQueue.getLockInfo(handle, name))); } @Override public long registerPeriodic( String name, String taskName, String cron, byte[] args, String queue, String timezone, boolean enabled) { - return NativeQueue.registerPeriodic(handle, name, taskName, cron, args, queue, timezone, enabled); + return withOpenHandle( + () -> NativeQueue.registerPeriodic(handle, name, taskName, cron, args, queue, timezone, enabled)); } @Override public String listPeriodicJson() { - return NativeQueue.listPeriodic(handle); + return withOpenHandle(() -> NativeQueue.listPeriodic(handle)); } @Override public boolean deletePeriodic(String name) { - return NativeQueue.deletePeriodic(handle, name); + return withOpenHandle(() -> NativeQueue.deletePeriodic(handle, name)); } @Override public boolean setPeriodicEnabled(String name, boolean enabled) { - return NativeQueue.setPeriodicEnabled(handle, name, enabled); + return withOpenHandle(() -> NativeQueue.setPeriodicEnabled(handle, name, enabled)); } @Override @@ -232,7 +268,7 @@ public String submitWorkflow( String[] deferredNames, String parentRunId, String parentNodeName) { - return NativeWorkflows.submitWorkflow( + return withOpenHandle(() -> NativeWorkflows.submitWorkflow( handle, name, version, @@ -243,37 +279,41 @@ public String submitWorkflow( paramsJson, deferredNames, parentRunId, - parentNodeName); + parentNodeName)); } @Override public String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade) { - return NativeWorkflows.markWorkflowNodeResult(handle, jobId, succeeded, error, skipCascade); + return withOpenHandle( + () -> NativeWorkflows.markWorkflowNodeResult(handle, jobId, succeeded, error, skipCascade)); } @Override public Optional getWorkflowStatusJson(String runId) { - return Optional.ofNullable(NativeWorkflows.getWorkflowStatus(handle, runId)); + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowStatus(handle, runId))); } @Override public void cancelWorkflowRun(String runId) { - NativeWorkflows.cancelWorkflowRun(handle, runId); + withOpenHandle(() -> { + NativeWorkflows.cancelWorkflowRun(handle, runId); + return null; + }); } @Override public Optional getWorkflowPlanJson(String runId) { - return Optional.ofNullable(NativeWorkflows.getWorkflowPlan(handle, runId)); + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.getWorkflowPlan(handle, runId))); } @Override public Optional workflowNodeForJobJson(String jobId) { - return Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId)); + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId))); } @Override public Optional workflowNameForRun(String runId) { - return Optional.ofNullable(NativeWorkflows.workflowNameForRun(handle, runId)); + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.workflowNameForRun(handle, runId))); } @Override @@ -287,13 +327,23 @@ public String[] expandFanOut( int maxRetries, long timeoutMs, int priority) { - return NativeWorkflows.expandFanOut( - handle, runId, parentNode, childNames, childPayloads, taskName, queue, maxRetries, timeoutMs, priority); + return withOpenHandle(() -> NativeWorkflows.expandFanOut( + handle, + runId, + parentNode, + childNames, + childPayloads, + taskName, + queue, + maxRetries, + timeoutMs, + priority)); } @Override public Optional checkFanOutCompletionJson(String runId, String parentNode) { - return Optional.ofNullable(NativeWorkflows.checkFanOutCompletion(handle, runId, parentNode)); + return withOpenHandle( + () -> Optional.ofNullable(NativeWorkflows.checkFanOutCompletion(handle, runId, parentNode))); } @Override @@ -306,98 +356,144 @@ public String createDeferredJob( int maxRetries, long timeoutMs, int priority) { - return NativeWorkflows.createDeferredJob( - handle, runId, nodeName, payload, taskName, queue, maxRetries, timeoutMs, priority); + return withOpenHandle(() -> NativeWorkflows.createDeferredJob( + handle, runId, nodeName, payload, taskName, queue, maxRetries, timeoutMs, priority)); } @Override public void cascadeSkipPending(String runId) { - NativeWorkflows.cascadeSkipPending(handle, runId); + withOpenHandle(() -> { + NativeWorkflows.cascadeSkipPending(handle, runId); + return null; + }); } @Override public Optional finalizeRunIfTerminal(String runId) { - return Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(handle, runId)); + return withOpenHandle(() -> Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(handle, runId))); } @Override public void setWorkflowNodeWaitingApproval(String runId, String nodeName) { - NativeWorkflows.setWorkflowNodeWaitingApproval(handle, runId, nodeName); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowNodeWaitingApproval(handle, runId, nodeName); + return null; + }); } @Override public void resolveWorkflowGate(String runId, String nodeName, boolean approved, String error) { - NativeWorkflows.resolveWorkflowGate(handle, runId, nodeName, approved, error); + withOpenHandle(() -> { + NativeWorkflows.resolveWorkflowGate(handle, runId, nodeName, approved, error); + return null; + }); } @Override public void setWorkflowNodeRunning(String runId, String nodeName) { - NativeWorkflows.setWorkflowNodeRunning(handle, runId, nodeName); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowNodeRunning(handle, runId, nodeName); + return null; + }); } @Override public void failWorkflowNode(String runId, String nodeName, String error) { - NativeWorkflows.failWorkflowNode(handle, runId, nodeName, error); + withOpenHandle(() -> { + NativeWorkflows.failWorkflowNode(handle, runId, nodeName, error); + return null; + }); } @Override public void skipWorkflowNode(String runId, String nodeName) { - NativeWorkflows.skipWorkflowNode(handle, runId, nodeName); + withOpenHandle(() -> { + NativeWorkflows.skipWorkflowNode(handle, runId, nodeName); + return null; + }); } @Override public void setWorkflowNodeCacheHit(String runId, String nodeName) { - NativeWorkflows.setWorkflowNodeCacheHit(handle, runId, nodeName); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowNodeCacheHit(handle, runId, nodeName); + return null; + }); } @Override public void setWorkflowRunCompensating(String runId) { - NativeWorkflows.setWorkflowRunCompensating(handle, runId); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowRunCompensating(handle, runId); + return null; + }); } @Override public void setWorkflowRunCompensated(String runId, long completedAt) { - NativeWorkflows.setWorkflowRunCompensated(handle, runId, completedAt); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowRunCompensated(handle, runId, completedAt); + return null; + }); } @Override public void setWorkflowRunCompensationFailed(String runId, long completedAt, String error) { - NativeWorkflows.setWorkflowRunCompensationFailed(handle, runId, completedAt, error); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowRunCompensationFailed(handle, runId, completedAt, error); + return null; + }); } @Override public void setWorkflowRunCompletedWithFailures(String runId, long completedAt) { - NativeWorkflows.setWorkflowRunCompletedWithFailures(handle, runId, completedAt); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowRunCompletedWithFailures(handle, runId, completedAt); + return null; + }); } @Override public void setWorkflowNodeCompensationJob( String runId, String nodeName, String compensationJobId, long startedAt) { - NativeWorkflows.setWorkflowNodeCompensationJob(handle, runId, nodeName, compensationJobId, startedAt); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowNodeCompensationJob(handle, runId, nodeName, compensationJobId, startedAt); + return null; + }); } @Override public void setWorkflowNodeCompensated(String runId, String nodeName, long completedAt) { - NativeWorkflows.setWorkflowNodeCompensated(handle, runId, nodeName, completedAt); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowNodeCompensated(handle, runId, nodeName, completedAt); + return null; + }); } @Override public void setWorkflowNodeCompensationFailed(String runId, String nodeName, String error, long completedAt) { - NativeWorkflows.setWorkflowNodeCompensationFailed(handle, runId, nodeName, error, completedAt); + withOpenHandle(() -> { + NativeWorkflows.setWorkflowNodeCompensationFailed(handle, runId, nodeName, error, completedAt); + return null; + }); } @Override public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) { - return new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson)); + return withOpenHandle(() -> new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson))); } - /** Idempotent: the native handle is freed exactly once, so a second {@code close()} - * (e.g. an explicit call plus try-with-resources) can't double-free. */ + /** Idempotent: frees the native queue handle exactly once, after in-flight calls drain. */ @Override - public synchronized void close() { - if (!closed) { - closed = true; - NativeQueue.close(handle); + public void close() { + stateLock.writeLock().lock(); + try { + if (!closed) { + closed = true; + NativeQueue.close(handle); + } + } finally { + stateLock.writeLock().unlock(); } } } 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 13f540d0..3dfd8103 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,48 +1,85 @@ package org.byteveda.taskito.internal; import java.util.Optional; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; import org.byteveda.taskito.spi.WorkerControl; -/** JNI-backed {@link WorkerControl} over a native worker handle. */ +/** + * JNI-backed {@link WorkerControl} over a native worker handle. Every call holds + * the read lock; {@code close()} takes the write lock, so it waits out in-flight + * calls and later calls throw {@link IllegalStateException} instead of touching + * freed native memory. + */ public final class JniWorkerControl implements WorkerControl { private final long handle; - private volatile boolean closed; + private final ReentrantReadWriteLock stateLock = new ReentrantReadWriteLock(); + private boolean closed; // guarded by stateLock JniWorkerControl(long handle) { this.handle = handle; } + private T withOpenHandle(Supplier nativeCall) { + stateLock.readLock().lock(); + try { + if (closed) { + throw new IllegalStateException("worker control is closed"); + } + return nativeCall.get(); + } finally { + stateLock.readLock().unlock(); + } + } + @Override public void completeJob(long token, byte[] result) { - NativeWorker.completeJob(handle, token, result); + withOpenHandle(() -> { + NativeWorker.completeJob(handle, token, result); + return null; + }); } @Override public void failJob(long token, String error) { - NativeWorker.failJob(handle, token, error); + withOpenHandle(() -> { + NativeWorker.failJob(handle, token, error); + return null; + }); } @Override public void cancelJob(long token) { - NativeWorker.cancelJob(handle, token); + withOpenHandle(() -> { + NativeWorker.cancelJob(handle, token); + return null; + }); } @Override public void stop() { - NativeWorker.stop(handle); + withOpenHandle(() -> { + NativeWorker.stop(handle); + return null; + }); } @Override public Optional meshClusterInfoJson() { - return Optional.ofNullable(NativeWorker.meshClusterInfo(handle)); + return withOpenHandle(() -> Optional.ofNullable(NativeWorker.meshClusterInfo(handle))); } - /** Idempotent: frees the native worker handle exactly once. */ + /** Idempotent: frees the native worker handle exactly once, after in-flight calls drain. */ @Override - public synchronized void close() { - if (!closed) { - closed = true; - NativeWorker.close(handle); + public void close() { + stateLock.writeLock().lock(); + try { + if (!closed) { + closed = true; + NativeWorker.close(handle); + } + } finally { + stateLock.writeLock().unlock(); } } } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java new file mode 100644 index 00000000..abd0b8e0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogLevel.java @@ -0,0 +1,35 @@ +package org.byteveda.taskito.logging; + +import java.util.Locale; + +/** Severity, from most to least verbose. {@link #SILENT} disables all output. */ +public enum LogLevel { + DEBUG(10), + INFO(20), + WARN(30), + ERROR(40), + SILENT(Integer.MAX_VALUE); + + private final int severity; + + LogLevel(int severity) { + this.severity = severity; + } + + /** Whether a message at this level clears the {@code threshold}. */ + boolean passes(LogLevel threshold) { + return severity >= threshold.severity; + } + + /** Case-insensitive parse; {@code null} when {@code raw} is not a level name. */ + static LogLevel parseOrNull(String raw) { + if (raw == null) { + return null; + } + try { + return valueOf(raw.trim().toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + return null; + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java new file mode 100644 index 00000000..a18f0fe3 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/LogSink.java @@ -0,0 +1,7 @@ +package org.byteveda.taskito.logging; + +/** Receives every formatted line that clears the level threshold. */ +@FunctionalInterface +public interface LogSink { + void accept(LogLevel level, String line); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java new file mode 100644 index 00000000..2680d7d4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/TaskitoLogger.java @@ -0,0 +1,109 @@ +package org.byteveda.taskito.logging; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.time.Instant; +import java.util.Objects; +import java.util.function.Supplier; + +/** + * A namespaced leveled logger ({@code [taskito:worker]}). Obtain via + * {@link #create}; the level threshold and sink are global, so + * {@link #setLevel}/{@link #setSink} take effect everywhere immediately. + * Messages below the threshold are dropped (suppliers are never invoked). + */ +public final class TaskitoLogger { + private static volatile LogLevel level = envLevel(); + private static volatile LogSink sink = TaskitoLogger::writeToStderr; + + private final String tag; + + private TaskitoLogger(String namespace) { + this.tag = namespace == null ? "taskito" : "taskito:" + namespace; + } + + /** The root {@code [taskito]} logger. Prefer {@link #create} for a namespaced one. */ + public static TaskitoLogger root() { + return new TaskitoLogger(null); + } + + /** A namespaced logger: {@code create("worker")} tags lines {@code [taskito:worker]}. */ + public static TaskitoLogger create(String namespace) { + Objects.requireNonNull(namespace, "namespace"); + return new TaskitoLogger(namespace); + } + + /** Set the global threshold; messages below it are dropped (and never built). */ + public static void setLevel(LogLevel newLevel) { + level = Objects.requireNonNull(newLevel, "level"); + } + + /** Replace the output sink (default: stderr). Useful for capture in tests. */ + public static void setSink(LogSink newSink) { + sink = Objects.requireNonNull(newSink, "sink"); + } + + public void debug(String message) { + emit(LogLevel.DEBUG, message, null); + } + + public void info(String message) { + emit(LogLevel.INFO, message, null); + } + + public void warn(String message) { + emit(LogLevel.WARN, message, null); + } + + public void warn(String message, Throwable cause) { + emit(LogLevel.WARN, message, cause); + } + + public void error(String message) { + emit(LogLevel.ERROR, message, null); + } + + public void error(String message, Throwable cause) { + emit(LogLevel.ERROR, message, cause); + } + + /** Lazy variant: {@code message} is built only when the level passes. */ + public void debug(Supplier message) { + if (LogLevel.DEBUG.passes(level)) { + emit(LogLevel.DEBUG, message.get(), null); + } + } + + private void emit(LogLevel messageLevel, String message, Throwable cause) { + if (!messageLevel.passes(level)) { + return; + } + StringBuilder line = new StringBuilder() + .append(Instant.now()) + .append(" [") + .append(tag) + .append("] ") + .append(messageLevel) + .append(' ') + .append(message); + if (cause != null) { + line.append(System.lineSeparator()).append(stackTraceOf(cause)); + } + sink.accept(messageLevel, line.toString()); + } + + private static String stackTraceOf(Throwable cause) { + StringWriter buffer = new StringWriter(); + cause.printStackTrace(new PrintWriter(buffer)); + return buffer.toString().stripTrailing(); + } + + private static LogLevel envLevel() { + LogLevel parsed = LogLevel.parseOrNull(System.getenv("TASKITO_LOG_LEVEL")); + return parsed == null ? LogLevel.WARN : parsed; + } + + private static void writeToStderr(LogLevel ignored, String line) { + System.err.println(line); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java new file mode 100644 index 00000000..efe3b037 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/logging/package-info.java @@ -0,0 +1,6 @@ +/** + * The SDK's zero-dependency leveled logger. Writes to stderr by default (stdout + * stays clean for CLI/piped output); threshold from {@code TASKITO_LOG_LEVEL}, + * overridable via {@link org.byteveda.taskito.logging.TaskitoLogger#setLevel}. + */ +package org.byteveda.taskito.logging; 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 310fd1a3..4333c97e 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 @@ -25,6 +25,7 @@ import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.logging.TaskitoLogger; import org.byteveda.taskito.middleware.Middleware; import org.byteveda.taskito.resources.ResourceRuntime; import org.byteveda.taskito.serialization.PayloadCodec; @@ -43,6 +44,7 @@ * jobs. */ public final class Worker implements AutoCloseable { + private static final TaskitoLogger LOG = TaskitoLogger.create("worker"); private static final long SHUTDOWN_TIMEOUT_SECONDS = 30; private static final ObjectMapper MESH_JSON = new ObjectMapper(); @@ -144,13 +146,21 @@ public synchronized void close() { executor.shutdown(); // stop accepting; let running handlers finish try { if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + // shutdownNow() only *requests* interruption; give stuck handlers + // one more window before closing the handle out from under them. executor.shutdownNow(); + if (!executor.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + LOG.warn("handler threads still running after " + (2 * SHUTDOWN_TIMEOUT_SECONDS) + + "s; closing the native worker — their late completions will be rejected"); + } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); executor.shutdownNow(); } - control.close(); // now safe to free the native handle — no handler can touch it + // Safe even if a straggler survives: JniWorkerControl serializes close() + // against in-flight native calls and rejects any call made afterwards. + control.close(); if (tracker != null) { tracker.close(); // stop the gate-timeout scheduler } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java index e819dbc3..a1aa6836 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java @@ -23,6 +23,7 @@ import org.byteveda.taskito.errors.SerializationException; import org.byteveda.taskito.errors.WorkflowException; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.logging.TaskitoLogger; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; @@ -45,6 +46,10 @@ * run reaches a terminal state. */ public final class WorkflowTracker { + private static final TaskitoLogger LOG = TaskitoLogger.create("workflows"); + private static final int FAIL_RECORD_MAX_ATTEMPTS = 5; + private static final long FAIL_RECORD_RETRY_MS = 1_000; + private final QueueBackend backend; private final Serializer serializer; private final ObjectMapper json = new ObjectMapper(); @@ -154,38 +159,62 @@ private void onOutcome(String jobId, boolean succeeded, String error) { finalizeIfTerminal(ref.runId); } - /** If the just-completed node feeds fan-outs, split its result list into child jobs. */ + /** + * If the just-completed node feeds fan-outs, split its result list into child + * jobs. Failures are scoped per branch: an unusable producer result fails + * every branch, but one branch's expansion error never touches a sibling + * whose children are already running. + */ private void expandFanOutIfAny(String runId, String node, String jobId, RunPlan plan) { List fanOuts = plan.successorsMatching(node, candidate -> candidate.fanOut != null); if (fanOuts.isEmpty()) { return; } - List items = serializer.deserialize(fetchResult(jobId), List.class); - for (PlanNode fanOut : fanOuts) { - String[] names = new String[items.size()]; - byte[][] payloads = new byte[items.size()][]; - for (int i = 0; i < items.size(); i++) { - names[i] = fanOut.name + "[" + i + "]"; - payloads[i] = serializer.serialize(items.get(i)); + List items; + try { + items = serializer.deserialize(fetchResult(jobId), List.class); + } catch (RuntimeException e) { + LOG.warn("reading fan-out producer result of node '" + node + "' in run " + runId + " failed", e); + for (PlanNode fanOut : fanOuts) { + failFanOut(runId, fanOut, plan, e); } - backend.expandFanOut( - runId, - fanOut.name, - names, - payloads, - fanOut.taskName, - fanOut.queueOrDefault(), - fanOut.retries(), - fanOut.timeout(), - fanOut.priorityOrDefault()); - // An empty producer list yields no child jobs, so no child outcome will - // ever drive the fan-in — collect it now with an empty result list. - if (items.isEmpty()) { - collectFanIns(runId, fanOut.name, List.of(), plan); + return; + } + for (PlanNode fanOut : fanOuts) { + try { + expandFanOut(runId, fanOut, items, plan); + } catch (RuntimeException e) { + LOG.warn("expanding fan-out node '" + fanOut.name + "' in run " + runId + " failed", e); + failFanOut(runId, fanOut, plan, e); } } } + /** Split the producer's items into one child job per element of {@code fanOut}. */ + private void expandFanOut(String runId, PlanNode fanOut, List items, RunPlan plan) { + String[] names = new String[items.size()]; + byte[][] payloads = new byte[items.size()][]; + for (int i = 0; i < items.size(); i++) { + names[i] = fanOut.name + "[" + i + "]"; + payloads[i] = serializer.serialize(items.get(i)); + } + backend.expandFanOut( + runId, + fanOut.name, + names, + payloads, + fanOut.taskName, + fanOut.queueOrDefault(), + fanOut.retries(), + fanOut.timeout(), + fanOut.priorityOrDefault()); + // An empty producer list yields no child jobs, so no child outcome will + // ever drive the fan-in — collect it now with an empty result list. + if (items.isEmpty()) { + collectFanIns(runId, fanOut.name, List.of(), plan); + } + } + /** When every fan-out child has settled, collect their results into the fan-in job(s). */ private void handleFanOutChild(String runId, String childName, RunPlan plan) { String parent = childName.substring(0, childName.indexOf('[')); @@ -290,11 +319,69 @@ private void promoteOrSkip(String runId, PlanNode node, Map= FAIL_RECORD_MAX_ATTEMPTS) { + LOG.error( + "failed to record failure of fan-out node '" + fanOut.name + "' in run " + runId + " after " + + attempt + " attempts; run cannot finalize", + e); + return; + } + LOG.warn( + "recording failure of fan-out node '" + fanOut.name + "' in run " + runId + " failed (attempt " + + attempt + "); retrying", + e); + gateScheduler.schedule( + () -> failFanOut(runId, fanOut, plan, cause, attempt + 1), + FAIL_RECORD_RETRY_MS, + TimeUnit.MILLISECONDS); } } + private static String describe(RuntimeException cause) { + String message = cause.getMessage(); + return cause.getClass().getSimpleName() + (message == null ? "" : ": " + message); + } + /** If a prior run cached this step's result within its TTL, mark it a cache hit instead of running it. */ private boolean reuseFromCache(String runId, PlanNode node, RunPlan plan) { String key = cacheKey(runId, node.name); diff --git a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java index 410f8a3d..07242505 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java @@ -3,6 +3,7 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.nio.file.Path; @@ -96,6 +97,33 @@ void enqueueManyAndFilteredList(@TempDir Path dir) { } } + @Test + void enqueueManyDedupesActiveUniqueKeys(@TempDir Path dir) { + try (Taskito queue = open(dir)) { + String first = queue.enqueue( + SEND_EMAIL, + Collections.singletonMap("to", "a"), + EnqueueOptions.builder().uniqueKey("k1").build()); + // The batch's duplicate key must resolve to the existing job instead + // of rolling back the whole batch on the unique index. + List ids = queue.enqueueMany( + SEND_EMAIL, + Arrays.asList(Collections.singletonMap("to", "a"), Collections.singletonMap("to", "b")), + EnqueueOptions.builder().uniqueKey("k1").build()); + assertEquals(2, ids.size()); + assertEquals(first, ids.get(0)); + assertEquals(first, ids.get(1)); + assertEquals(1, queue.stats().pending); + } + } + + @Test + void callsAfterCloseThrowInsteadOfTouchingFreedHandle(@TempDir Path dir) { + Taskito queue = open(dir); + queue.close(); + assertThrows(IllegalStateException.class, queue::stats); + } + @Test void settingsRoundTrip(@TempDir Path dir) { try (Taskito queue = open(dir)) { diff --git a/sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java new file mode 100644 index 00000000..49cc34b2 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/TaskitoLoggerTest.java @@ -0,0 +1,56 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import org.byteveda.taskito.logging.LogLevel; +import org.byteveda.taskito.logging.TaskitoLogger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class TaskitoLoggerTest { + private final List lines = new ArrayList<>(); + + @AfterEach + void restoreDefaults() { + TaskitoLogger.setLevel(LogLevel.WARN); + TaskitoLogger.setSink((level, line) -> System.err.println(line)); + } + + @Test + void dropsMessagesBelowThresholdAndTagsNamespace() { + TaskitoLogger.setSink((level, line) -> lines.add(line)); + TaskitoLogger.setLevel(LogLevel.WARN); + TaskitoLogger log = TaskitoLogger.create("worker"); + + log.info("dropped"); + log.warn("kept"); + + assertEquals(1, lines.size()); + assertTrue(lines.get(0).contains("[taskito:worker] WARN kept")); + } + + @Test + void appendsStackTraceForThrowables() { + TaskitoLogger.setSink((level, line) -> lines.add(line)); + TaskitoLogger.setLevel(LogLevel.ERROR); + + TaskitoLogger.root().error("boom", new IllegalStateException("cause detail")); + + assertEquals(1, lines.size()); + assertTrue(lines.get(0).contains("[taskito] ERROR boom")); + assertTrue(lines.get(0).contains("IllegalStateException: cause detail")); + } + + @Test + void silentDisablesEverything() { + TaskitoLogger.setSink((level, line) -> lines.add(line)); + TaskitoLogger.setLevel(LogLevel.SILENT); + + TaskitoLogger.root().error("never emitted"); + + assertTrue(lines.isEmpty()); + } +} diff --git a/sdks/java/test-support/build.gradle.kts b/sdks/java/test-support/build.gradle.kts index 033124d5..3f175f43 100644 --- a/sdks/java/test-support/build.gradle.kts +++ b/sdks/java/test-support/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-test", version.toString()) + pom { + name.set("Taskito Test Support") + description.set("In-memory queue backend for testing Taskito JVM applications without the native runtime.") + 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