fix(rocksdb): retain Topling OLAP handles

docs/storage-lifecycle.md

@@ -652,6 +652,14 @@ At request/task completion, explicitly committed writes are preserved; unfinishe
 are rolled back and cached backend leases are released. Shared schema and element caches
 retain their invalidation listeners until the graph closes.
 
+For standalone Topling storage, graph truncate and OLAP property clear retain column
+families and their native handles. OLAP clear removes only the selected property table
+in its owning database; pending writes on the clearing thread are discarded before the
+clear is committed. Other threads retain their pending batches. If those threads commit
+after the clear, their writes can populate the cleared table again. Callers that require
+the table to remain empty must coordinate concurrent writers. Graph truncate resets
+pending batches across all sessions instead.
+
 Stop incoming work and use the component's normal stop script. PD drains its scheduled
 metadata work, joins Raft and closes the metadata database and native options. Do not
 use a forced kill as evidence of normal resource cleanup.

hugegraph-server/hugegraph-dist/src/assembly/static/bin/arthas.sh

@@ -123,18 +123,26 @@ stop_agent() {
     local session attempt agent_pid="" line endpoint rows=0
     connect_ip=$(cat "$state_dir/connect-ip")
     endpoint=$(format_endpoint)
-    if ! session=$(bounded_client session 2>&1); then
+    for attempt in 1 2 3; do
+        if session=$(bounded_client session 2>&1); then
+            break
+        fi
         if [[ "$session" == *"Connection refused"* ]]; then
             echo "Arthas listener closed at $endpoint" >&2
             return 0
         fi
         if [[ "$session" == *"not permitted"* ]]; then
             echo "Error: Arthas credentials rejected at $endpoint; agent left running" >&2
-        else
+            return 1
+        fi
+        if [ "$attempt" -eq 3 ]; then
             echo "Error: could not verify Arthas cleanup at $endpoint; agent left running" >&2
+            return 1
         fi
-        return 1
-    fi
+        # An explicit stop can close the connection during this probe. Retry
+        # until the listener refuses connections or its PID can be checked.
+        sleep 0.2
+    done
     # AuthCommand echoes the username. Trust only one complete session table row.
     while IFS= read -r line; do
         if [[ "$line" =~ ^[[:space:]]*JAVA_PID[[:space:]]+([0-9]+)[[:space:]]*$ ]]; then

hugegraph-server/hugegraph-dist/src/assembly/travis/test-arthas-cli.sh

@@ -97,8 +97,8 @@ with tempfile.TemporaryDirectory(prefix="arthas-test-") as work:
     script = ["bash", str(root / "bin/arthas.sh")]
     client = script + ["--telnet-port", str(safe_port)]
 
-    def run(arguments, success=True):
-        result = subprocess.run(arguments, env=client_env, text=True,
+    def run(arguments, success=True, env=None):
+        result = subprocess.run(arguments, env=client_env if env is None else env, text=True,
                                 stdout=subprocess.PIPE, stderr=subprocess.STDOUT, input="", timeout=45)
         assert (result.returncode == 0) == success, result.stdout
         return result.stdout
@@ -320,7 +320,32 @@ with tempfile.TemporaryDirectory(prefix="arthas-test-") as work:
                 endpoint = f"[{connect_host}]:{safe_port}" if ":" in connect_host else f"{connect_host}:{safe_port}"
                 assert output.count(f"Arthas listener closed at {endpoint}") == 1, output
                 wait_port(False, connect_host)
-            run(client + ["--command", "stop"])
+            # Reproduce the cleanup race after an explicit stop: the first
+            # session probe loses its connection while the agent shuts down.
+            # All attachment, stop and subsequent probes still use the real JVM.
+            wrapper_home = root / "java-wrapper"
+            (wrapper_home / "bin").mkdir(parents=True)
+            marker = root / "cleanup-race"
+            injected = root / "cleanup-race-injected"
+            wrapper = wrapper_home / "bin/java"
+            wrapper.write_text(f"#!{sys.executable}\n" +
+                               "import pathlib, subprocess, sys\n" +
+                               f"marker = pathlib.Path({str(marker)!r})\n" +
+                               "args = sys.argv[1:]\n" +
+                               "if '-f' in args and marker.exists():\n" +
+                               "    batch = pathlib.Path(args[args.index('-f') + 1]).read_text()\n" +
+                               "    if batch.strip() == 'session':\n" +
+                               f"        marker.rename({str(injected)!r})\n" +
+                               "        print('Connection reset during shutdown', file=sys.stderr)\n" +
+                               "        sys.exit(1)\n" +
+                               f"status = subprocess.call([{java!r}] + args)\n" +
+                               "if '--command' in args and args[args.index('--command') + 1] == 'stop':\n" +
+                               "    marker.touch()\n" +
+                               "sys.exit(status)\n")
+            wrapper.chmod(0o755)
+            output = run(client + ["--command", "stop"], env=dict(client_env, JAVA_HOME=str(wrapper_home)))
+            assert injected.is_file() and not marker.exists(), "Shutdown race was not exercised"
+            assert output.count(f"Arthas listener closed at 127.0.0.1:{safe_port}") == 1, output
             wait_port(False)
             # Disable native local authentication bypass for the following
             # attachments; earlier signal cases exercise native no-auth defaults.

hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java

@@ -1135,14 +1135,32 @@ public void createOlapTable(Id id) {
 
         @Override
         public void clearOlapTable(Id id) {
-            String name = this.olapTableName(id);
-            RocksDBTable table = this.table(name);
-            RocksDBSessions db = this.db(HugeType.OLAP);
-            if (!db.existsTable(table.table())) {
-                throw new HugeException("Not exist table '%s''", name);
+            Lock writeLock = this.storeLock().writeLock();
+            writeLock.lock();
+            try {
+                String name = this.olapTableName(id);
+                RocksDBTable table = this.table(name);
+                RocksDBSessions db = this.db(HugeType.OLAP);
+                if (!db.existsTable(table.table())) {
+                    throw new HugeException("Not exist table '%s''", name);
+                }
+                if (super.toplingProvider) {
+                    // Keep the CF owner alive. Use only the clearing thread's
+                    // lease so pending writes cannot be committed by clearTables.
+                    RocksDBSessions.Session session = (RocksDBSessions.Session) db.useSession();
+                    try {
+                        session.reset();
+                        db.clearTables(Collections.singletonList(table.table()));
+                    } finally {
+                        db.close();
+                    }
+                } else {
+                    this.dropTable(db, table.table());
+                    this.createTable(db, table.table());
+                }
+            } finally {
+                writeLock.unlock();
             }
-            this.dropTable(db, table.table());
-            this.createTable(db, table.table());
         }
 
         @Override

hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java

@@ -33,6 +33,8 @@
 import org.apache.hugegraph.backend.store.rocksdb.RocksDBSessions;
 import org.apache.hugegraph.backend.store.rocksdb.RocksDBStdSessions;
 import org.apache.hugegraph.exception.BackendException;
+import org.apache.hugegraph.id.Id;
+import org.apache.hugegraph.id.IdGenerator;
 import org.apache.hugegraph.backend.store.rocksdb.RocksDBStore;
 import org.apache.hugegraph.backend.store.rocksdb.RocksDBStoreProvider;
 import org.apache.hugegraph.backend.store.rocksdbsst.RocksDBSstSessions;
@@ -308,6 +310,198 @@ protected Map<String, RocksDBSessions> tableDBMapping() {
         }
     }
 
+    @Test
+    public void testToplingOlapClearKeepsHandleAndOtherTables() throws Exception {
+        this.assertOlapClear(true, false);
+    }
+
+    @Test
+    public void testToplingOlapClearRoutesIndependentDatabase() throws Exception {
+        this.assertOlapClear(true, true);
+    }
+
+    @Test
+    public void testStandardOlapClearRecreatesTable() throws Exception {
+        this.assertOlapClear(false, false);
+    }
+
+    @Test
+    public void testToplingOlapClearRollsBackFailedCommit() throws Exception {
+        Id id = IdGenerator.of(123L);
+        String table = "store+ap_123";
+        RocksDBStore store = this.adapterStore(true, ImmutableList.of(TABLE));
+        store.createOlapTable(id);
+        RocksDBSessions.Session session = this.rocks.session();
+        session.put(table, getBytes("m"), getBytes("old"));
+        session.put(table, getBytes("n"), getBytes("old"));
+        session.commit();
+        session.put(table, getBytes("z"), getBytes("pending"));
+        AtomicReference<?> owner = Whitebox.getInternalState(this.rocks, "rocksdb");
+        Object opened = owner.get();
+        Map<String, ?> handles = Whitebox.getInternalState(opened, "cfHandles");
+        Object handle = handles.get(table);
+        Object references = Whitebox.getInternalState(session, "refs");
+        RocksDB realDB = Whitebox.getInternalState(opened, "rocksdb");
+        RocksDB faultDB = Mockito.mock(RocksDB.class, AdditionalAnswers.delegatesTo(realDB));
+        RocksDBException writeFailure = new RocksDBException("injected clear commit failure");
+        Mockito.doThrow(writeFailure).when(faultDB).write(Mockito.any(WriteOptions.class),
+                                                        Mockito.any(WriteBatch.class));
+        Whitebox.setInternalState(opened, "rocksdb", faultDB);
+        try {
+            Throwable failure = Assert.assertThrows(BackendException.class, () -> store.clearOlapTable(id));
+            Assert.assertSame(writeFailure, failure.getCause());
+        } finally {
+            Whitebox.setInternalState(opened, "rocksdb", realDB);
+        }
+        Assert.assertEquals(references, Whitebox.getInternalState(session, "refs"));
+        Assert.assertSame(handle, handles.get(table));
+        Assert.assertFalse(session.hasChanges());
+        session.commit();
+        Assert.assertArrayEquals(getBytes("old"), session.get(table, getBytes("m")));
+        Assert.assertArrayEquals(getBytes("old"), session.get(table, getBytes("n")));
+        Assert.assertNull(session.get(table, getBytes("z")));
+        store.clearOlapTable(id);
+        Assert.assertNull(session.keyRange(table));
+        Assert.assertSame(handle, handles.get(table));
+        Assert.assertEquals(references, Whitebox.getInternalState(session, "refs"));
+    }
+
+    @Test
+    public void testToplingOlapClearPreservesOtherThreadPendingWrites() throws Exception {
+        Id id = IdGenerator.of(123L);
+        String path = DB_PATH + "/olap-clear-concurrent";
+        String table = "store+ap_123";
+        RocksDBSessions olap = new RocksDBStdSessions(FakeObjects.newConfig(), "db", "olap", path, path);
+        CountDownLatch pending = new CountDownLatch(1);
+        CountDownLatch commit = new CountDownLatch(1);
+        AtomicReference<Throwable> failure = new AtomicReference<>();
+        Thread writer = new Thread(() -> {
+            try {
+                RocksDBSessions.Session session = olap.session();
+                session.put(table, getBytes("m"), getBytes("pending replacement"));
+                session.put(table, getBytes("z"), getBytes("pending outside range"));
+                pending.countDown();
+                Assert.assertTrue(commit.await(30, TimeUnit.SECONDS));
+                session.commit();
+            } catch (Throwable e) {
+                failure.set(e);
+            } finally {
+                pending.countDown();
+                try {
+                    olap.close();
+                } catch (Throwable e) {
+                    failure.compareAndSet(null, e);
+                }
+            }
+        }, "olap-pending-writer");
+        try {
+            RocksDBStore.RocksDBGraphStore store = new RocksDBStore.RocksDBGraphStore(
+                    new RocksDBStoreProvider(), "db", "store") {
+                @Override
+                protected RocksDBSessions db(HugeType type) {
+                    return type == HugeType.OLAP ? olap : super.db(type);
+                }
+            };
+            Whitebox.setInternalState(store, "sessions", olap);
+            Whitebox.setInternalState(store, "toplingProvider", true);
+            store.createOlapTable(id);
+            RocksDBSessions.Session session = olap.session();
+            session.put(table, getBytes("m"), getBytes("old"));
+            session.put(table, getBytes("n"), getBytes("old"));
+            session.commit();
+            writer.start();
+            Assert.assertTrue(pending.await(30, TimeUnit.SECONDS));
+            Assert.assertNull(failure.get());
+            Assert.assertArrayEquals(getBytes("old"), session.get(table, getBytes("m")));
+            Assert.assertNull(session.get(table, getBytes("z")));
+
+            store.clearOlapTable(id);
+            // The other worker cannot commit until the clear has returned.
+            Assert.assertNull(session.keyRange(table));
+            commit.countDown();
+            writer.join(TimeUnit.SECONDS.toMillis(30));
+            Assert.assertFalse(writer.isAlive());
+            Assert.assertNull(failure.get());
+            Assert.assertArrayEquals(getBytes("pending replacement"), session.get(table, getBytes("m")));
+            Assert.assertArrayEquals(getBytes("pending outside range"), session.get(table, getBytes("z")));
+            Assert.assertNull(session.get(table, getBytes("n")));
+        } finally {
+            commit.countDown();
+            writer.join(TimeUnit.SECONDS.toMillis(30));
+            Assert.assertFalse("Writer must stop before closing its database", writer.isAlive());
+            olap.close();
+            FileUtils.deleteDirectory(new File(path));
+        }
+    }
+
+    private void assertOlapClear(boolean topling, boolean separate) throws Exception {
+        Id id = IdGenerator.of(123L);
+        String path = DB_PATH + "/olap-clear-" + topling + "-" + separate;
+        RocksDBSessions olap = new RocksDBStdSessions(FakeObjects.newConfig(), "db", "olap", path, path);
+        RocksDBStore.RocksDBGraphStore store = new RocksDBStore.RocksDBGraphStore(
+                new RocksDBStoreProvider(), "db", "store") {
+            @Override
+            protected RocksDBSessions db(HugeType type) {
+                return type == HugeType.OLAP ? olap : super.db(type);
+            }
+        };
+        Whitebox.setInternalState(store, "sessions", separate ? this.rocks : olap);
+        Whitebox.setInternalState(store, "toplingProvider", topling);
+        try {
+            olap.session();
+            store.createOlapTable(id);
+            String table = "store+ap_123";
+            olap.createTable(TABLE, "other-olap");
+            RocksDBSessions.Session session = olap.session();
+            session.put(table, getBytes("m"), getBytes("old"));
+            session.put(table, getBytes("n"), getBytes("old"));
+            session.put("other-olap", getBytes("other"), getBytes("retained"));
+            session.commit();
+            session.put(TABLE, getBytes("main"), getBytes("retained main"));
+            session.commit();
+            AtomicReference<?> owner = Whitebox.getInternalState(olap, "rocksdb");
+            Object opened = owner.get();
+            Map<String, ?> handles = Whitebox.getInternalState(opened, "cfHandles");
+            Object handle = handles.get(table);
+            Assert.assertNotNull(handle);
+            for (int i = 0; i < 2; i++) {
+                // Pending keys lie outside the stored range, or in an empty table.
+                if (topling) {
+                    session.put(table, getBytes("a"), getBytes("pending first"));
+                    session.put(table, getBytes("z"), getBytes("pending last"));
+                }
+                store.clearOlapTable(id);
+                if (topling) {
+                    Assert.assertSame(handle, handles.get(table));
+                } else {
+                    Assert.assertNotSame(handle, handles.get(table));
+                    handle = handles.get(table);
+                }
+                Assert.assertTrue(olap.existsTable(table));
+                Assert.assertNull(session.keyRange(table));
+                session.commit();
+                Assert.assertNull(session.keyRange(table));
+                Assert.assertArrayEquals(getBytes("retained"), session.get("other-olap", getBytes("other")));
+                Assert.assertArrayEquals(getBytes("retained main"), session.get(TABLE, getBytes("main")));
+            }
+            session.put(table, getBytes("new"), getBytes("after clear"));
+            session.commit();
+            olap.close();
+            RocksDBSessions reopened = new RocksDBStdSessions(FakeObjects.newConfig(), "db", "olap", path, path,
+                                                               ImmutableList.of(TABLE, table, "other-olap"));
+            try {
+                RocksDBSessions.Session reader = reopened.session();
+                Assert.assertArrayEquals(getBytes("after clear"), reader.get(table, getBytes("new")));
+                Assert.assertArrayEquals(getBytes("retained"), reader.get("other-olap", getBytes("other")));
+            } finally {
+                reopened.close();
+            }
+        } finally {
+            olap.close();
+            FileUtils.deleteDirectory(new File(path));
+        }
+    }
+
     private void assertAdapterTruncatePreservesBackendVersion(boolean topling) throws Exception {
         RocksDBStoreProvider provider = new RocksDBStoreProvider();
         RocksDBStore store = new RocksDBStore.RocksDBSystemStore(provider, "db", "store") {