fix(store): unblock batch scan cancellation

docs/storage-lifecycle.md

@@ -58,6 +58,10 @@ reused executor thread is not mistaken for the cancelled reader. Half-close does
 interrupt the reader. Queued cancelled tasks still execute their cleanup/accounting
 paths, and iterator close waits for an active read to leave the iterator.
 
+Batch-scan request state is published under its manager lock, while reader notification,
+worker scheduling and iterator cleanup run outside that lock. A half-close or receipt
+waiting for a slow read therefore does not block cancellation from interrupting the reader.
+
 Unary query and count requests share the query admission, worker and cleanup owner
 registry. Count keeps partition work parallel on the service executor. Iterator cleanup
 failure prevents successful completion and remains visible to request drain.

hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/grpc/ScanBatchResponse3.java

@@ -174,44 +174,62 @@ private static class OrderManager {
         OrderDeliverer deliverer;
         boolean cancelled;
 
-        synchronized void deal(OrderWorker worker, OrderDeliverer deliverer) {
+        void deal(OrderWorker worker, OrderDeliverer deliverer) {
             if (log.isDebugEnabled()) {
                 log.debug("Receiving query request.");
             }
-            if (!this.cancelled && this.state == OrderState.NEW) {
-                this.worker = worker;
-                this.deliverer = deliverer;
-                this.worker.hereWeGo();
-                this.state = OrderState.WORKING;
+            boolean accepted;
+            synchronized (this) {
+                accepted = !this.cancelled && this.state == OrderState.NEW;
+                if (accepted) {
+                    this.worker = worker;
+                    this.deliverer = deliverer;
+                    this.state = OrderState.WORKING;
+                }
+            }
+            if (accepted) {
+                worker.hereWeGo();
             } else {
                 worker.closeIterator();
             }
         }
 
-        synchronized void receipt(int receiptTimes) {
+        void receipt(int receiptTimes) {
             if (log.isDebugEnabled()) {
                 log.debug("Receiving receipt request.");
             }
-            if (this.worker != null) {
-                this.worker.setReceipt(receiptTimes);
+            OrderWorker current;
+            synchronized (this) {
+                current = this.worker;
+            }
+            if (current != null) {
+                current.setReceipt(receiptTimes);
             }
         }
 
-        synchronized boolean inputCompleted() {
-            if (this.cancelled) {
-                return false;
-            }
-            if (this.worker == null) {
-                return true;
+        boolean inputCompleted() {
+            OrderWorker current;
+            synchronized (this) {
+                if (this.cancelled) {
+                    return false;
+                }
+                current = this.worker;
+                if (current == null) {
+                    return true;
+                }
             }
-            this.worker.inputCompleted();
+            current.inputCompleted();
             return false;
         }
 
-        synchronized void breakdown() {
-            this.cancelled = true;
-            if (this.worker != null) {
-                this.worker.breakdown();
+        void breakdown() {
+            OrderWorker current;
+            synchronized (this) {
+                this.cancelled = true;
+                current = this.worker;
+            }
+            if (current != null) {
+                current.breakdown();
             }
         }
     }

hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/service/ScanLifecycleTest.java

@@ -92,6 +92,76 @@
 
 public class ScanLifecycleTest extends ScanTestSupport {
 
+    @Test(timeout = 10000)
+    public void testBatchHalfCloseDoesNotBlockTransportCancellation() throws Exception {
+        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
+        HgStoreWrapperEx wrapper = mock(HgStoreWrapperEx.class);
+        ScanIterator iterator = mock(ScanIterator.class);
+        StreamObserver<KvPageRes> output = mock(StreamObserver.class);
+        CountDownLatch reading = new CountDownLatch(1);
+        CountDownLatch interrupted = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        when(wrapper.scanAll(anyString(), anyString(), any(byte[].class))).thenReturn(iterator);
+        when(iterator.hasNext()).thenAnswer(ignored -> {
+            reading.countDown();
+            try {
+                assertTrue(release.await(5, TimeUnit.SECONDS));
+            } catch (InterruptedException expected) {
+                interrupted.countDown();
+                Thread.currentThread().interrupt();
+            }
+            return false;
+        });
+        HgStoreStreamImpl service = scanService(executor, wrapper);
+        Context.CancellableContext context = Context.current().withCancellation();
+        StreamObserver<ScanStreamBatchReq> input = context.call(() -> service.scanBatch(output));
+        FutureTask<Void> halfClosing = new FutureTask<>(() -> {
+            input.onCompleted();
+            return null;
+        });
+        FutureTask<Void> cancelling = new FutureTask<>(() -> {
+            context.cancel(null);
+            return null;
+        });
+        Thread halfClose = new Thread(halfClosing, "batch-scan-half-close");
+        Thread cancel = new Thread(cancelling, "batch-scan-context-cancel");
+        try {
+            input.onNext(batchRequest());
+            assertTrue(reading.await(1, TimeUnit.SECONDS));
+            halfClose.start();
+            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1);
+            while (halfClose.getState() != Thread.State.BLOCKED && System.nanoTime() < deadline) {
+                Thread.yield();
+            }
+            assertEquals(Thread.State.BLOCKED, halfClose.getState());
+            assertTrue(Arrays.stream(halfClose.getStackTrace()).anyMatch(frame ->
+                    frame.getClassName().endsWith("ScanBatchResponse3$OrderWorker") &&
+                    frame.getMethodName().equals("inputCompleted")));
+
+            cancel.start();
+            assertTrue("transport cancellation must interrupt the blocked scan reader",
+                       interrupted.await(1, TimeUnit.SECONDS));
+            cancelling.get(1, TimeUnit.SECONDS);
+            halfClosing.get(1, TimeUnit.SECONDS);
+            executor.submit(() -> { }).get(1, TimeUnit.SECONDS);
+            verify(iterator).close();
+            assertTrue("the cancelled scan must release its owner", scanRegistry(service).isEmpty());
+            assertCancelled(output);
+            verify(output, never()).onNext(any(KvPageRes.class));
+            verify(output, never()).onCompleted();
+        } finally {
+            release.countDown();
+            halfClose.join(2000);
+            cancel.join(2000);
+            context.cancel(null);
+            service.shutdownScans();
+            executor.shutdownNow();
+            assertTrue(executor.awaitTermination(2, TimeUnit.SECONDS));
+            assertFalse(halfClose.isAlive());
+            assertFalse(cancel.isAlive());
+        }
+    }
+
     @Test
     public void testFailedSelectionCleanupBlocksLifecycleCompletion() throws Exception {
         ScanIterator iterator = mock(ScanIterator.class);