fix(core): close traversal iterators on failure

docs/storage-lifecycle.md

@@ -69,6 +69,15 @@ does not guarantee that it will return on Java thread interruption. The call rem
 registered until its read and iterator cleanup finish; cancellation does not permit
 another thread to close a native iterator that is still in use.
 
+### Server OLTP traversal batches
+
+Server OLTP traversal workers close each active batch iterator on completion,
+interruption or failure, then release their thread's graph transactions. A capacity
+stop also releases producers waiting on a full queue. Iterator cleanup errors remain
+visible when the capacity stop ends traversal normally; a business failure retains
+its cleanup errors as suppressed exceptions. Caller transactions remain with the
+request thread.
+
 ### Callback executor configuration
 
 The gRPC application callback queue is always unbounded (`Integer.MAX_VALUE`), matching

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/traversal/algorithm/OltpTraverser.java

@@ -335,22 +335,32 @@ public EdgesConsumer(Consumer<E> consumer, long capacity) {
 
         @Override
         public void accept(Iterator<T> edgeIter) {
-            // Bind worker leases before a caller-created lazy query opens a session.
-            graph().tx().readWrite();
-            Iterator<E> ids = prepare(edgeIter);
-            long counter = 0;
-            while (ids.hasNext()) {
-                if (Thread.currentThread().isInterrupted()) {
-                    LOG.warn("Consumer is Interrupted");
-                    break;
+            Iterator<?> toClose = edgeIter;
+            Throwable failure = null;
+            try {
+                // Bind worker leases before a caller-created lazy query opens a session.
+                graph().tx().readWrite();
+                Iterator<E> ids = prepare(edgeIter);
+                toClose = ids != null ? ids : edgeIter;
+                long counter = 0;
+                while (ids.hasNext()) {
+                    if (Thread.currentThread().isInterrupted()) {
+                        LOG.warn("Consumer is Interrupted");
+                        break;
+                    }
+                    counter++;
+                    this.consumer.accept(ids.next());
                 }
-                counter++;
-                this.consumer.accept(ids.next());
-            }
-            long total = edgeIterCounter.addAndGet(counter);
-            // traverse by batch & improve performance
-            if (this.capacity != NO_LIMIT && total >= this.capacity) {
-                throw new Consumers.StopExecution("reach capacity");
+                long total = edgeIterCounter.addAndGet(counter);
+                // traverse by batch & improve performance
+                if (this.capacity != NO_LIMIT && total >= this.capacity) {
+                    throw new Consumers.StopExecution("reach capacity");
+                }
+            } catch (RuntimeException | Error e) {
+                failure = e;
+                throw e;
+            } finally {
+                closeIterator(toClose, failure);
             }
         }
     }

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Consumers.java

@@ -57,6 +57,7 @@ public final class Consumers<V> {
     private final CountDownLatch latch;
     private final BlockingQueue<V> queue;
     private volatile Throwable exception = null;
+    private volatile boolean stopped;
 
     public Consumers(ExecutorService executor, Consumer<V> consumer) {
         this(executor, consumer, null);
@@ -98,6 +99,7 @@ public Consumers(ExecutorService executor,
 
     public void start(String name) {
         this.exception = null;
+        this.stopped = false;
         if (this.executor == null) {
             return;
         }
@@ -117,15 +119,15 @@ public void start(String name) {
     private Void safeRun(ContextCallable<Void> task) {
         try {
             return task.call(); // may fail before/after runAndDone()
-        } catch (Exception e) {
+        } catch (Throwable e) {
             // This exception is from ContextCallable wrapper (setContext/resetContext/delegate dispatch),
             // not from runAndDone() business logic (that one is handled inside runAndDone()).
-            if (this.exception == null) {
-                this.exception = e;
+            if (this.recordException(e)) {
                 LOG.error("Consumer worker failed in ContextCallable wrapper", e);
             } else {
                 LOG.warn("Additional worker failure in ContextCallable wrapper; first exception already recorded", e);
             }
+            this.stop();
             this.exceptionHandle(e);
         } finally {
             this.latch.countDown();
@@ -136,25 +138,33 @@ private Void safeRun(ContextCallable<Void> task) {
     private Void runAndDone() {
         try {
             this.run();
-        } catch (Exception e) {
+        } catch (Throwable e) {
             if (e instanceof StopExecution) {
-                this.queue.clear();
-                putQueueEnd();
+                Throwable[] suppressed = e.getSuppressed();
+                // Stop stays control flow, but failed cleanup must reach the caller.
+                for (Throwable failure : suppressed) {
+                    this.recordException(failure);
+                }
             } else {
-                if (this.exception == null) {
-                    this.exception = e;
+                if (this.recordException(e)) {
                     LOG.error("Unhandled exception in consumer task", e);
                 } else {
                     LOG.warn("Additional exception in consumer task; first exception already recorded", e);
                 }
             }
+            this.stop();
             this.exceptionHandle(e);
         } finally {
             this.done();
         }
         return null;
     }
 
+    private void stop() {
+        this.stopped = true;
+        this.queue.clear();
+    }
+
     private void run() {
         LOG.debug("Start to work...");
         while (this.consume()) {
@@ -167,6 +177,9 @@ private void run() {
     private boolean consume() {
         V elem = null;
         while (elem == null) {
+            if (this.stopped) {
+                return false;
+            }
             try {
                 elem = this.queue.poll(CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS);
             } catch (InterruptedException e) {
@@ -175,6 +188,9 @@ private boolean consume() {
             }
         }
 
+        if (this.stopped) {
+            return false;
+        }
         if (elem == QUEUE_END) {
             putQueueEnd();
             return false;
@@ -192,9 +208,7 @@ private void exceptionHandle(Throwable e) {
         try {
             this.exceptionHandle.accept(e);
         } catch (Throwable ex) {
-            if (this.exception == null) {
-                this.exception = ex;
-            } else {
+            if (!this.recordException(ex)) {
                 LOG.warn("Error while calling exceptionHandle()", ex);
             }
         }
@@ -208,19 +222,32 @@ private void done() {
         try {
             this.doneHandle.run();
         } catch (Throwable e) {
-            if (this.exception == null) {
-                this.exception = e;
-            } else {
+            if (!this.recordException(e)) {
                 LOG.warn("Error while calling done()", e);
             }
+            this.stop();
         }
     }
 
+    private synchronized boolean recordException(Throwable failure) {
+        if (this.exception == null) {
+            this.exception = failure;
+            return true;
+        }
+        if (this.exception != failure) {
+            for (Throwable suppressed : this.exception.getSuppressed()) {
+                if (suppressed == failure) {
+                    return false;
+                }
+            }
+            this.exception.addSuppressed(failure);
+        }
+        return false;
+    }
+
     private Throwable throwException() {
         assert this.exception != null;
-        Throwable e = this.exception;
-        this.exception = null;
-        return e;
+        return this.exception;
     }
 
     public void provide(V v) throws Throwable {
@@ -232,7 +259,17 @@ public void provide(V v) throws Throwable {
             throw this.throwException();
         } else {
             try {
-                this.queue.put(v);
+                while (!this.stopped &&
+                       !this.queue.offer(v, CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS)) {
+                    // A stopped worker must not leave a producer blocked on a full queue.
+                }
+                if (this.stopped) {
+                    this.queue.clear();
+                    if (this.exception != null) {
+                        throw this.throwException();
+                    }
+                    throw new StopExecution("Consumer stopped");
+                }
             } catch (InterruptedException e) {
                 LOG.warn("Interrupt while queuing QUEUE_END", e);
             }
@@ -242,7 +279,10 @@ public void provide(V v) throws Throwable {
     private void putQueueEnd() {
         if (this.executor != null) {
             try {
-                this.queue.put(QUEUE_END);
+                while (!this.stopped &&
+                       !this.queue.offer(QUEUE_END, CONSUMER_WAKE_PERIOD, TimeUnit.MILLISECONDS)) {
+                    // Stop wakes consumers through its state, without another END token.
+                }
             } catch (InterruptedException e) {
                 LOG.warn("Interrupted while enqueue", e);
             }
@@ -262,7 +302,8 @@ public void await() throws Throwable {
                 for (Future f : this.runningFutures) {
                     f.cancel(true);
                 }
-                this.exception = new HugeException(error, e);
+                this.recordException(new HugeException(error, e));
+                this.stop();
                 LOG.warn(error, e);
             }
         }

hugegraph-server/hugegraph-test/pom.xml

@@ -228,11 +228,6 @@
             <plugin>
                 <groupId>org.jacoco</groupId>
                 <artifactId>jacoco-maven-plugin</artifactId>
-                <configuration>
-                    <excludes>
-                        <exclude>org/apache/hugegraph/traversal/algorithm/*.class</exclude>
-                    </excludes>
-                </configuration>
                 <executions>
                     <execution>
                         <id>pre-test</id>

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java

@@ -38,6 +38,7 @@
 @RunWith(Suite.class)
 @Suite.SuiteClasses({
         BackendLeaseCleanupTest.class,
+        OltpTraverserCursorTest.class,
         OltpTraverserRocksDBTest.class,
         PropertyKeyCoreTest.class,
         VertexLabelCoreTest.class,