diff --git a/docs/storage-lifecycle.md b/docs/storage-lifecycle.md index ddf7731a0a..cfffc44173 100644 --- a/docs/storage-lifecycle.md +++ b/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. diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/bin/arthas.sh b/hugegraph-server/hugegraph-dist/src/assembly/static/bin/arthas.sh index 3b20dcdc64..1486f06c52 100755 --- a/hugegraph-server/hugegraph-dist/src/assembly/static/bin/arthas.sh +++ b/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 diff --git a/hugegraph-server/hugegraph-dist/src/assembly/travis/test-arthas-cli.sh b/hugegraph-server/hugegraph-dist/src/assembly/travis/test-arthas-cli.sh index 767d58805c..d339cea1f9 100644 --- a/hugegraph-server/hugegraph-dist/src/assembly/travis/test-arthas-cli.sh +++ b/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. diff --git a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java index d28030b36c..d166d11dcd 100644 --- a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java +++ b/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 diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java index 314406838f..9d7ea30962 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java +++ b/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 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 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 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 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") {