fix(process-engine): survive optimistic-lock races on task, process, and context rows

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/helpers/TaskExecutorService.java

@@ -112,29 +112,41 @@ public void execute(@NotNull Runnable command) {
             final Future<Boolean> future = delegate.submit(wrapper);
             tasks.put(fk, future);
             if (timeout != 0L) {
-                tasks.put(fk2, delegate.submit(() -> {
-                    try {
-                        future.get(timeout, TimeUnit.SECONDS);
-                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
-                        log.info("Task {} execution was interrupted by timeout", taskId);
-                        future.cancel(true);
-                        ProcessInstanceImpl processInstance = ProcessOrchestrator.getInstance().getProcessInstance(taskInstance.getProcessID());
-                        processInstance.setState(TaskState.FAILED);
-                        processInstance.save();
-                        TaskInstanceImpl task = ProcessOrchestrator.getInstance().getTaskInstanceRepository().getTaskInstance(taskId);
-                        task.setState(TaskState.FAILED);
-                        task.save();
-                    }
-
-                    return Boolean.TRUE;
-                }));
+                tasks.put(fk2, delegate.submit(() -> watchTask(future, timeout, taskId, taskInstance.getProcessID())));
             }
 
 
         }
     }
 
+    // Watchdog for tasks with a sync timeout. The completion callback in execute()
+    // cancels the watchdog with cancel(true) once the task finishes, so an
+    // interrupt means "stop watching", not "the task timed out" — only a real
+    // timeout or a failed worker future may mark the task and process FAILED.
+    // Package-private so tests can drive it directly.
+    Boolean watchTask(Future<Boolean> future, long timeout, String taskId, String processId) {
+        try {
+            future.get(timeout, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } catch (ExecutionException | TimeoutException e) {
+            log.info("Task {} execution was interrupted by timeout", taskId);
+            future.cancel(true);
+            ProcessInstanceImpl processInstance = ProcessOrchestrator.getInstance().getProcessInstance(processId);
+            processInstance.setState(TaskState.FAILED);
+            processInstance.saveResolvingConflict(pi -> pi.setState(TaskState.FAILED));
+            TaskInstanceImpl task = ProcessOrchestrator.getInstance().getTaskInstanceRepository().getTaskInstance(taskId);
+            task.setState(TaskState.FAILED);
+            task.saveResolvingConflict(t -> t.setState(TaskState.FAILED));
+        }
+        return Boolean.TRUE;
+    }
+
 
+    // Reflection into db-scheduler's lambda internals is the only way to reach the
+    // Execution from the submitted Runnable; on any failure getId degrades to null
+    // and the task runs without the wrapper bookkeeping.
+    @SuppressWarnings("java:S3011")
     @Nullable
     private Execution getId(Runnable task) {
         try {
@@ -159,23 +171,27 @@ private Execution getId(Runnable task) {
         }
     }
 
-    public Future<?> terminate(String key) {
+    public Future<Void> terminate(String key) {
         List<Future<Boolean>> fs = tasks
                 .entrySet()
                 .stream()
-                .filter(e -> e.getKey().equals(key))
+                .filter(e -> e.getKey().getTaskId().equals(key))
                 .map(f -> woreService.submit(new TerminateRunnable(f.getValue(), key)))
                 .toList();
         return woreService.submit(() -> {
                     if (!fs.isEmpty())
                         fs.forEach(f -> {
                             try {
                                 f.get();
-                            } catch (InterruptedException | ExecutionException e) {
-                                throw new RuntimeException(e);
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                throw new IllegalStateException(e);
+                            } catch (ExecutionException e) {
+                                throw new IllegalStateException(e);
                             }
                         });
-                }
+                },
+                null
         );
     }
 }

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/DataContext.java

@@ -6,6 +6,7 @@
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.netcracker.core.scheduler.po.repository.ContextRepository;
+import com.netcracker.core.scheduler.po.repository.VersionMismatchException;
 import com.netcracker.core.scheduler.po.serializers.DataContextDeserializer;
 import com.netcracker.core.scheduler.po.serializers.DataContextSerializer;
 import lombok.Getter;
@@ -17,6 +18,9 @@
 @Getter
 @JsonSerialize(using = DataContextSerializer.class)
 @JsonDeserialize(using = DataContextDeserializer.class)
+// Content equality inherited from HashMap is intentional: id, version, and the
+// dirty flag are persistence bookkeeping, not part of the context's identity.
+@SuppressWarnings("java:S2160")
 public class DataContext extends HashMap<String, Object> {
 
     @Setter
@@ -71,8 +75,31 @@ public void save() {
         repository.putContext(this);
     }
 
+    /**
+     * Applies the mutation and saves. On a version conflict the fresh context is
+     * reloaded, the same mutation is re-applied to it, and the fresh copy is
+     * saved — so a fixed-value update (task state description, start/end time)
+     * survives a concurrent writer instead of aborting the whole execution.
+     * The mutation must be idempotent.
+     */
     public void apply(Consumer<DataContext> function) {
         function.accept(this);
-        save();
+        try {
+            save();
+        } catch (VersionMismatchException e) {
+            DataContext fresh = repository.getContext(getId());
+            fresh.setRepository(repository);
+            function.accept(fresh);
+            fresh.save();
+            // Sync this instance with the persisted state: contents, version, and
+            // a clean dirty flag — otherwise the caller keeps a stale copy and the
+            // very next save() fails again despite apply() having succeeded.
+            // super-level access bypasses the dirty-marking overrides.
+            super.clear();
+            for (Entry<String, Object> entry : fresh.entrySet()) {
+                super.put(entry.getKey(), entry.getValue());
+            }
+            setVersion(fresh.getVersion());
+        }
     }
 }

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/FutureKey.java

@@ -19,14 +19,9 @@ public FutureKey(String taskId) {
 
     @Override
     public boolean equals(Object obj) {
-        if (obj instanceof String) {
-            return taskId.equals(obj);
-        } else if (obj instanceof UUID) {
-            return uuid.equals(obj);
-        } else if (obj instanceof FutureKey futureKey) {
-            return taskId.equals(futureKey.taskId) && uuid.equals(futureKey.uuid);
-        }
-        return false;
+        return obj instanceof FutureKey futureKey
+                && taskId.equals(futureKey.taskId)
+                && uuid.equals(futureKey.uuid);
     }
 
     @Override

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/Process.java

@@ -1,3 +1,3 @@
 package com.netcracker.core.scheduler.po;
 
 import com.github.kagkarlsson.scheduler.task.CompletionHandler;
@@ -56,7 +56,7 @@
                     logger.info("Process {} are failed", taskInstance.getId());
                     taskInstance.getData().getProcess().setState(TaskState.FAILED);
                     executionOperations.remove();
-                    taskInstance.getData().getProcess().save();
+                    taskInstance.getData().getProcess().saveResolvingConflict(pi -> pi.setState(TaskState.FAILED));
                     return;
                 }
                 completed = completed && task.getState() == TaskState.COMPLETED;
@@ -65,9 +65,13 @@
                 executionOperations.reschedule(executionComplete, Instant.now().plusSeconds(2), taskInstance.getData());
             else {
                 logger.info("Process {} are completed", taskInstance.getId());
-                taskInstance.getData().getProcess().setEndTime(Calendar.getInstance().getTime());
+                java.util.Date endTime = Calendar.getInstance().getTime();
+                taskInstance.getData().getProcess().setEndTime(endTime);
                 taskInstance.getData().getProcess().setState(TaskState.COMPLETED);
-                taskInstance.getData().getProcess().save();
+                taskInstance.getData().getProcess().saveResolvingConflict(pi -> {
+                    pi.setEndTime(endTime);
+                    pi.setState(TaskState.COMPLETED);
+                });
                 executionOperations.remove();
             }
             taskInstance.getData().getProcess().save();

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/model/pojo/ProcessInstanceImpl.java

@@ -1,13 +1,15 @@
 package com.netcracker.core.scheduler.po.model.pojo;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreType;
 import com.netcracker.core.scheduler.po.DataContext;
 import com.netcracker.core.scheduler.po.ProcessOrchestrator;
+import com.netcracker.core.scheduler.po.repository.VersionMismatchException;
 import com.netcracker.core.scheduler.po.repository.TaskInstanceRepository;
 import com.netcracker.core.scheduler.po.task.TaskState;
 import lombok.Getter;
 
 import java.util.Date;
+import java.util.function.Consumer;
 import java.util.List;
 import java.util.Objects;
 
@@ -87,6 +89,28 @@
         return Objects.hash(getName(), getId(), getProcessDefinitionID(), getStartTime(), getEndTime(), getState(), getVersion());
     }
 
+    /**
+     * Saves, and on a version conflict reloads the row, re-applies the mutation
+     * to the fresh copy, saves it, and syncs this instance with the persisted
+     * state — so callers (and any later raw save()) keep working with a clean,
+     * current object. Meant for must-win writes such as the terminal
+     * COMPLETED/FAILED transitions; the mutation must be idempotent.
+     */
+    public ProcessInstanceImpl saveResolvingConflict(Consumer<ProcessInstanceImpl> reapply) {
+        try {
+            save();
+        } catch (VersionMismatchException e) {
+            ProcessInstanceImpl fresh = reload();
+            reapply.accept(fresh);
+            fresh.save();
+            setState(fresh.getState());
+            setStartTime(fresh.getStartTime());
+            setEndTime(fresh.getEndTime());
+            setVersion(fresh.getVersion());
+        }
+        return this;
+    }
+
     public void save() {
         if (isDirty)
             ProcessOrchestrator.getInstance().getProcessInstanceRepository().putProcessInstance(this);

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/model/pojo/TaskInstanceImpl.java

@@ -1,14 +1,16 @@
 package com.netcracker.core.scheduler.po.model.pojo;
 
 import com.netcracker.core.scheduler.po.DataContext;
 import com.netcracker.core.scheduler.po.ProcessOrchestrator;
+import com.netcracker.core.scheduler.po.repository.VersionMismatchException;
 import com.netcracker.core.scheduler.po.task.NamedTask;
 import com.netcracker.core.scheduler.po.task.TaskState;
 import lombok.Getter;
 
 import java.util.Date;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Consumer;
 
 
 public class TaskInstanceImpl {
@@ -108,6 +110,33 @@
         return ProcessOrchestrator.getInstance().getTaskInstanceRepository().getTaskInstance(id);
     }
 
+    /**
+     * Saves, and on a version conflict reloads the row, re-applies the mutation
+     * to the fresh copy, and saves that copy. Meant for must-win writes such as
+     * the terminal COMPLETED/FAILED transitions: several actors (the process
+     * tick, the executing task, the failure handler, the timeout watchdog) hold
+     * independent in-memory copies of the same row, and a completion that loses
+     * the version race by milliseconds must not abort — an aborted completion
+     * leaves the db-scheduler execution stuck picked until dead-execution
+     * revival. The mutation must be idempotent. This instance is synced with
+     * the persisted state afterwards, so callers keep a clean, current object.
+     */
+    public TaskInstanceImpl saveResolvingConflict(Consumer<TaskInstanceImpl> reapply) {
+        try {
+            save();
+        } catch (VersionMismatchException e) {
+            TaskInstanceImpl fresh = reload();
+            reapply.accept(fresh);
+            fresh.save();
+            setState(fresh.getState());
+            setName(fresh.getName());
+            setType(fresh.getType());
+            setDependsOn(fresh.getDependsOn());
+            setVersion(fresh.getVersion());
+        }
+        return this;
+    }
+
     /**/
 
     public DataContext getContext() {

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/repository/impl/ContextRepositoryImpl.java

@@ -53,24 +53,31 @@ public void putContext(DataContext context) {
         if (version == null) {
             jdbcRunner.execute(insertQuery, (PreparedStatement p) -> assignParameters(context, p));
             context.setDirty(false);
-        } else if (Objects.equals(context.getVersion(), version)) {
-            context.setVersion(version + 1);
-            jdbcRunner.execute(
+        } else {
+            // The version check must live in the UPDATE itself: a check-then-act
+            // compare leaves a window where a concurrent writer slips between the
+            // SELECT above and the UPDATE, silently losing one of the writes. The
+            // in-memory version is bumped only after the guarded UPDATE succeeded.
+            int expectedVersion = context.getVersion();
+            int updated = jdbcRunner.execute(
                     "update " +
                             tableName +
                             " set " +
                             "version = ?, " +
-                            "context_data = ? where id=?"
+                            "context_data = ? where id=? and version=?"
                     , (PreparedStatement p) -> {
-                        p.setString(3, context.getId());
+                        p.setInt(1, expectedVersion + 1);
                         p.setObject(2, serializer.serialize(context));
-                        p.setInt(1, context.getVersion());
+                        p.setString(3, context.getId());
+                        p.setInt(4, expectedVersion);
                     }
             );
+            if (updated == 0) {
+                throw new VersionMismatchException("Current version are less than saved");
+            }
+            context.setVersion(expectedVersion + 1);
             context.setDirty(false);
-        } else throw new
-
-                VersionMismatchException("Current version are less than saved");
+        }
     }
 
     @Override

core-process-orchestrator/src/main/java/com/netcracker/core/scheduler/po/repository/impl/ProcessInstanceRepositoryImpl.java

@@ -37,25 +37,28 @@ public void putProcessInstance(ProcessInstanceImpl processInstance) {
         if (version == null) {
             insertInstance(processInstance);
         } else {
-            if (version.equals(processInstance.getVersion())) {
-                processInstance.setVersion(version + 1);
-
-                updateInstance(processInstance);
-            } else throw new VersionMismatchException("Current version are less than saved");
-
+            // The version check must live in the UPDATE itself: a check-then-act
+            // compare leaves a window where a concurrent writer slips between the
+            // SELECT above and the UPDATE, silently losing one of the writes. The
+            // in-memory version is bumped only after the guarded UPDATE succeeded.
+            int expectedVersion = processInstance.getVersion();
+            if (updateInstance(processInstance, expectedVersion) == 0) {
+                throw new VersionMismatchException("Current version are less than saved");
+            }
+            processInstance.setVersion(expectedVersion + 1);
         }
     }
 
-    private void updateInstance(ProcessInstanceImpl processInstance) {
-        jdbcRunner.execute(
+    private int updateInstance(ProcessInstanceImpl processInstance, int expectedVersion) {
+        return jdbcRunner.execute(
                 "update " +
                 tableName +
                 " set " +
                 "   state=?," +
                 "   start_time=?," +
                 "   end_time=?," +
                 "   version=?" +
-                " where pi_id=?"
+                " where pi_id=? and version=?"
                 ,
                 (PreparedStatement p) -> {
                     p.setString(1, processInstance.getState().toString());
@@ -67,9 +70,10 @@ private void updateInstance(ProcessInstanceImpl processInstance) {
                     if (processInstance.getEndTime() != null) {
                         p.setLong(3, processInstance.getEndTime().getTime());
                     } else p.setLong(3, 0L);
-                    p.setInt(4, processInstance.getVersion());
+                    p.setInt(4, expectedVersion + 1);
 
                     p.setString(5, processInstance.getId());
+                    p.setInt(6, expectedVersion);
                 }
         );
     }