Re: [PR] fix(core): preserve task and graph lifecycle [hugegraph]

["imbajin (via GitHub)" <gi...@apache.org>]

imbajin commented on code in PR #3308:
URL: https://github.com/apache/hugegraph/pull/3308#discussion_r4247598006


##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java:
##########
@@ -469,17 +550,200 @@ private void saveTaskWithoutResult(HugeTask<?> task) {
         });
     }
 
+    private boolean cancelLocal(PendingTask pending, Consumer<Boolean> afterCancel) {
+        return this.cancelLocal(pending, null, afterCancel);
+    }
+
+    private boolean cancelLocal(PendingTask pending, Runnable beforeCancel,
+                                Consumer<Boolean> afterCancel) {
+        return this.cancelLocal(pending, beforeCancel, afterCancel, false);
+    }
+
+    private boolean cancelLocal(PendingTask pending, Runnable beforeCancel,
+                                Consumer<Boolean> afterCancel, boolean terminalOnly) {
+        if (SCHEDULER_WORKER.get() == this.schedulerExecutor) {
+            return this.cancelOwned(pending, beforeCancel, afterCancel, terminalOnly);
+        }
+        // Keep callbacks and final persistence off the request/close thread and
+        // the DB worker, because callbacks may synchronously save to that worker.
+        return this.call(() -> {
+            SCHEDULER_WORKER.set(this.schedulerExecutor);
+            try {
+                return this.cancelOwned(pending, beforeCancel, afterCancel, terminalOnly);
+            } finally {
+                SCHEDULER_WORKER.remove();
+                // An interrupted request may stop waiting before this owner finishes.
+                this.finishCancelledQueued(pending);
+            }
+        }, this.schedulerExecutor, true);
+    }
+
+    private boolean cancelOwned(PendingTask pending, Runnable beforeCancel,
+                                Consumer<Boolean> afterCancel, boolean terminalOnly) {
+        LockResult acquired = null;
+        boolean acquire;
+        synchronized (pending) {
+            if (this.runningTasks.get(pending.task.id()) != pending) {
+                return false;
+            }
+            acquire = !pending.task.ephemeralTask() && pending.taskLock == null;
+            pending.cancellationDepth++;
+        }
+        boolean entered = false;
+        boolean coordinated = false;
+        try {
+            if (acquire) {
+                // Don't block publication/release of this runner's token during the lock RPC.
+                acquired = this.tryLockTask(pending.task.id().asString());
+                if (!acquired.lockSuccess()) {
+                    return false;
+                }
+                synchronized (pending) {
+                    pending.taskLock = acquired;
+                }
+            }
+            entered = true;
+            if (!pending.task.ephemeralTask()) {
+                HugeTask<?> stored;
+                try {
+                    stored = this.taskWithoutResult(pending.task.id());
+                } catch (NotFoundException e) {
+                    pending.discardQueued = true;
+                    coordinated = true;
+                    return false;
+                }
+                if (beforeCancel == null) {
+                    if (TaskStatus.COMPLETED_STATUSES.contains(stored.status())) {
+                        pending.discardQueued = true;
+                        coordinated = true;
+                        return false;
+                    }
+                    if (terminalOnly) {
+                        if (stored.status() == TaskStatus.DELETING) {
+                            pending.discardQueued = true;
+                            coordinated = true;
+                        } else if (pending.task.isCancelled() &&
+                                   pending.task.status() == TaskStatus.CANCELLED &&
+                                   stored.status() == TaskStatus.CANCELLING) {

Review Comment:
   [P1] Retry cancellation persistence for an already-running owner
   
   This only repairs the queued-reservation case, whose record has already been changed to CANCELLING. For a persistent task holding pending.taskLock, cancel() skips requestCancellation and cancelOwned() skips the acquired != null transition, so a transient failure of done/cancelled/final save can leave the record RUNNING while the in-memory task is CANCELLED and coordinationPending stays true. Once the worker exits, close() retries through terminalOnly, but this condition never persists the terminal state or clears the ticket. Because close() already cancelled cron, later close retries remain blocked even after the store recovers.
   
   Please cover a running LockHoldingJob failure/recovery case and allow the owned RUNNING case to finish safely without replaying callbacks or overwriting a remote terminal/DELETING state. The new checkCancellationPersistence test gates execution before the owner acquires its lock, so it covers the queued case only.
   
   Evidence boundary: independently traced the source and reproduced this with extracted cancel/cancelOwned/terminal-cleanup methods and real JDK FutureTask in a bounded cloud model. After three recovered retries, the running case retained stored RUNNING/ticket=1; the queued control reached CANCELLED/ticket=0, with callbacks remaining exactly once. Storage/PD locks/dispatch were stubbed; this is not a live PD/HStore validation.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org