From 3db1aa98af2a1b922961ee1e3853b0337ad5323d Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 02:35:45 +0800 Subject: [PATCH 1/5] fix(store): reject unsafe legacy graph prefixes - inspect reserved-prefix rows before allocating graph IDs - preflight existing databases before partition Raft replay - validate staged snapshots before activation and index publication - reject unknown path access and cover native recovery boundaries --- docs/store-graphid-upgrade.md | 30 ++ .../hugegraph/store/PartitionEngine.java | 21 + .../store/business/BusinessHandlerImpl.java | 14 +- .../hugegraph/store/meta/GraphIdManager.java | 97 +++++ .../store/raft/PartitionStateMachine.java | 9 +- hugegraph-store/hg-store-rocksdb/README.md | 37 ++ .../rocksdb/access/RocksDBSession.java | 79 +++- .../rocksdb/access/RocksDBSessionTest.java | 267 ++++++++++++ hugegraph-store/hg-store-test/pom.xml | 1 + .../store/GraphIdUpgradePreflightTest.java | 390 ++++++++++++++++++ .../store/meta/GraphIDManagerTest.java | 154 +++++++ 11 files changed, 1085 insertions(+), 14 deletions(-) create mode 100644 docs/store-graphid-upgrade.md create mode 100644 hugegraph-store/hg-store-rocksdb/README.md create mode 100644 hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java diff --git a/docs/store-graphid-upgrade.md b/docs/store-graphid-upgrade.md new file mode 100644 index 0000000000..2225eebfbd --- /dev/null +++ b/docs/store-graphid-upgrade.md @@ -0,0 +1,30 @@ +# Store legacy GraphID preflight + +Some older Store writes can have the reserved `FFFE` graph prefix without a +persisted graph-to-ID mapping. Allocating a new graph ID does not rewrite those +rows, so their physical presence does not imply that normal graph reads can see +them. This condition is specific to affected old write paths; it does not mean +that every 1.7 database needs migration. + +Store checks existing graph-data column families before allocating a missing +graph ID, starting a partition's Raft state machine, or activating a snapshot. +If it finds the reserved prefix, it rejects that operation and reports the +partition and column family. Metadata-only column families are excluded. The +checks do not assign an owner to legacy rows or change the persisted ID format. + +The partition preflight resolves existing DB versions without creating a DB, +allocating a store location or pruning older versions. It runs before partition +task/FSM construction and Raft replay; other existing Store bootstrap steps can +already have opened databases. It is not a promise that the entire startup +process leaves every file untouched. + +A rejected snapshot retains the active DB and its index, the incoming snapshot, +and the staged copy for diagnosis. Successful snapshots still use the existing +activation path. An incomplete selected DB version with an older valid DB is +rejected rather than silently skipping its preflight. + +After rejection, preserve the original database, metadata and snapshot evidence +and investigate the affected write/upgrade history before restarting writes or +replay. Do not delete mappings or assign all reserved-prefix rows to one graph: +different graphs may share that prefix. This change provides detection and +refusal, not automatic migration, repair or proof of a complete 1.7 upgrade. diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java index 85e43fea02..c8d6e58c80 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java @@ -22,6 +22,7 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -44,6 +45,7 @@ import org.apache.hugegraph.pd.grpc.MetaTask; import org.apache.hugegraph.pd.grpc.Metapb; import org.apache.hugegraph.pd.raft.RaftReflectionUtil; +import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.business.BusinessHandlerImpl; import org.apache.hugegraph.store.cmd.HgCmdClient; @@ -52,6 +54,7 @@ import org.apache.hugegraph.store.cmd.request.DbCompactionRequest; import org.apache.hugegraph.store.cmd.request.UpdatePartitionRequest; import org.apache.hugegraph.store.listener.PartitionStateListener; +import org.apache.hugegraph.store.meta.GraphIdManager; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.meta.PartitionManager; import org.apache.hugegraph.store.meta.Shard; @@ -175,6 +178,7 @@ public synchronized boolean init(PartitionEngineOptions opts) { return true; } + this.checkLegacyGraphIdData(opts.getGroupId()); log.info("PartitionEngine starting: {}", this); this.taskManager = new TaskManager(storeEngine.getBusinessHandler(), opts.getGroupId()); this.snapshotHandler = new SnapshotHandler(this); @@ -267,6 +271,23 @@ public LogStorage createLogStorage(final String uri, final RaftOptions raftOptio return this.started; } + private void checkLegacyGraphIdData(int partitionId) { + Metapb.PartitionStore location = + this.partitionManager.getStoreMetadata().getPartitionStore(partitionId); + // Looking up or assigning a new store location would write metadata before inspection. + String[] roots = location != null ? new String[]{location.getStoreLocation()} : + this.storeEngine.getOption().getDataPath().split(","); + String dbName = BusinessHandlerImpl.getDbName(partitionId); + for (String root : roots) { + String dataPath = Paths.get(root, HgStoreEngineOptions.DB_Path_Prefix).toString(); + String existing = RocksDBSession.resolveExistingDBPath( + dataPath, dbName, this.storeEngine.getCommittedIndex(partitionId)); + if (existing != null) { + GraphIdManager.checkLegacyGraphIdData(existing, partitionId); + } + } + } + public HgStoreEngine getStoreEngine() { return this.storeEngine; } diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java index bc446afb31..b88447369e 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java @@ -101,6 +101,7 @@ import org.apache.hugegraph.store.grpc.Graphpb.ScanPartitionRequest.Request; import org.apache.hugegraph.store.grpc.Graphpb.ScanPartitionRequest.ScanType; import org.apache.hugegraph.store.grpc.query.DeDupOption; +import org.apache.hugegraph.store.meta.GraphIdManager; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.meta.PartitionManager; import org.apache.hugegraph.store.meta.asynctask.AsyncTaskState; @@ -1195,11 +1196,20 @@ public void saveSnapshot(String snapshotPath, String graph, int partId) throws public void loadSnapshot(String snapshotPath, String graph, int partId, long v1) throws HgStoreException { try (RocksDBSession dbSession = getSession(graph, partId)) { - dbSession.loadSnapshot(snapshotPath, v1); + dbSession.loadSnapshot(snapshotPath, v1, stagedPath -> { + try { + GraphIdManager.checkLegacyGraphIdData(stagedPath, partId); + } catch (HgStoreException e) { + throw new DBStoreException(e.getMessage(), e); + } + }); keyCreator.clearCache(partId); factory.destroyGraphDB(dbSession.getGraphName()); } catch (DBStoreException e) { - throw new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, e.toString()); + HgStoreException failure = + new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, e.toString()); + failure.initCause(e); + throw failure; } } diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java index 76eada307b..ea463aefc9 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java @@ -21,9 +21,11 @@ import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -31,11 +33,21 @@ import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.rocksdb.access.SessionOperator; +import org.apache.hugegraph.store.constant.HugeServerTables; import org.apache.hugegraph.store.meta.base.DBSessionBuilder; import org.apache.hugegraph.store.meta.base.PartitionMetaStore; import org.apache.hugegraph.store.term.Bits; import org.apache.hugegraph.store.util.Asserts; import org.apache.hugegraph.store.util.HgStoreException; +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.InfoLogLevel; +import org.rocksdb.Logger; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksDBException; import com.google.protobuf.Int64Value; import com.google.protobuf.InvalidProtocolBufferException; @@ -99,6 +111,9 @@ public long getGraphIdOrCreate(String graphName) { byte[] key = MetadataKeyHelper.getGraphIDKey(graphName); Int64Value id = get(Int64Value.parser(), key); if (id == null) { + try (RocksDBSession session = this.sessionBuilder.getSession(this.partitionId)) { + checkLegacyGraphIdData(session, this.partitionId); + } id = Int64Value.of(getCId(GRAPH_ID_PREFIX, maxGraphID)); if (id.getValue() == -1) { throw new HgStoreException(HgStoreException.EC_FAIL, @@ -260,6 +275,88 @@ public long releaseGraphId(String graphName) { return gid; } + /** + * Reject unresolved rows written with the 1.7 missing-graph sentinel. + * Their graph ownership cannot be inferred from the shared physical prefix. + */ + public static void checkLegacyGraphIdData(RocksDBSession session, int partitionId) { + var lock = session.getCfHandleReadLock(); + lock.lock(); + try { + checkLegacyGraphIdData(session.getDB(), session.getTables(), partitionId); + } catch (RocksDBException e) { + throw new HgStoreException(HgStoreException.EC_FAIL, + "Can't inspect legacy GraphID data in partition " + partitionId, e); + } finally { + lock.unlock(); + } + } + + public static void checkLegacyGraphIdData(String existingDbPath, int partitionId) { + try (Options options = new Options(); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions(); + DBOptions dbOptions = new DBOptions(); + Logger logger = new Logger(dbOptions) { + @Override + protected void log(InfoLogLevel level, String message) { + // Keep inspection from creating or replacing the DB's LOG file. + } + }) { + dbOptions.setLogger(logger); + List names = RocksDB.listColumnFamilies(options, existingDbPath); + List descriptors = new ArrayList<>(); + for (byte[] name : names) { + descriptors.add(new ColumnFamilyDescriptor(name, cfOptions)); + } + List handles = new ArrayList<>(); + RocksDB db = null; + try { + db = RocksDB.openReadOnly(dbOptions, existingDbPath, descriptors, handles); + Map tables = new HashMap<>(); + for (int i = 0; i < names.size(); i++) { + tables.put(new String(names.get(i), StandardCharsets.UTF_8), handles.get(i)); + } + checkLegacyGraphIdData(db, tables, partitionId); + } finally { + for (ColumnFamilyHandle handle : handles) { + handle.close(); + } + if (db != null) { + db.close(); + } + } + } catch (RocksDBException e) { + throw new HgStoreException(HgStoreException.EC_FAIL, + "Can't inspect legacy GraphID data in partition " + partitionId + + " at " + existingDbPath, e); + } + } + + private static void checkLegacyGraphIdData(RocksDB db, + Map tables, + int partitionId) throws RocksDBException { + byte[] prefix = {(byte) 0xff, (byte) 0xfe}; + for (String table : HugeServerTables.TABLES) { + ColumnFamilyHandle handle = tables.get(table); + if (HugeServerTables.UNKNOWN_TABLE.equals(table) || handle == null) { + continue; + } + try (var iterator = db.newIterator(handle)) { + iterator.seek(prefix); + iterator.status(); + if (iterator.isValid()) { + byte[] key = iterator.key(); + if (key.length >= prefix.length && key[0] == prefix[0] && key[1] == prefix[1]) { + throw new HgStoreException(HgStoreException.EC_FAIL, + "Legacy GraphID data in partition " + partitionId + + ", table " + table + ", prefix 0xFFFE; " + + "stop upgrade and recover the affected data"); + } + } + } + } + } + /** * To maintain compatibility with affected graphs, ensure the g+v table contains no data * diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java index cf74810aec..1ba2e7399c 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java @@ -222,11 +222,11 @@ public void onSnapshotSave(final SnapshotWriter writer, final Closure done) { @Override public boolean onSnapshotLoad(final SnapshotReader reader) { + final long snapshotIndex; try { RaftOutter.SnapshotMeta meta = reader.load(); if (meta != null) { - this.committedIndex = meta.getLastIncludedIndex(); - log.info("onSnapshotLoad committedIndex = {}", this.committedIndex); + snapshotIndex = meta.getLastIncludedIndex(); } else { log.error("onSnapshotLoad failed to get SnapshotMeta"); return false; @@ -241,8 +241,9 @@ public boolean onSnapshotLoad(final SnapshotReader reader) { return false; } try { - snapshotHandler.onSnapshotLoad(reader, this.committedIndex); - log.info("Raft {} onSnapshotLoad success", groupId); + snapshotHandler.onSnapshotLoad(reader, snapshotIndex); + this.committedIndex = snapshotIndex; + log.info("Raft {} onSnapshotLoad success, committedIndex = {}", groupId, snapshotIndex); return true; } catch (HgStoreException e) { log.error(String.format("Raft %s onSnapshotLoad failed. ", groupId), e); diff --git a/hugegraph-store/hg-store-rocksdb/README.md b/hugegraph-store/hg-store-rocksdb/README.md new file mode 100644 index 0000000000..1e4c261d6e --- /dev/null +++ b/hugegraph-store/hg-store-rocksdb/README.md @@ -0,0 +1,37 @@ + + +# RocksDB path and snapshot validation + +`RocksDBSession.resolveExistingDBPath(dbDataPath, graphName, version)` uses the +same version-directory selection as the constructor, without opening, creating, +pruning or renaming databases. It returns `null` when the selected path has no +`CURRENT` and no existing database version is present. If an incomplete selected +version would hide another existing database, it fails without changing either +path. Directory listing or `CURRENT` inspection failures are rejected; unknown +access or I/O state is not treated as an absent database. Paths queued for removal retain the constructor's existing treatment. +The normal constructor continues to open read/write and prune old versions. + +`loadSnapshot(snapshotPath, version, Consumer validator)` first copies +and verifies the staging database, then passes its path to the validator before +replacing the target version. The validator must inspect it read-only. A rejected +validation preserves active databases, target versions and staging evidence; +the exception cause is retained. The original two-argument overload remains +available and uses a no-op validator. + +These APIs provide generic inspection hooks. They do not validate graph ownership, +migrate historical data, or replace deployment-specific recovery and upgrade checks. diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java index f4e7605a7f..319e878896 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java @@ -21,9 +21,13 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; +import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -33,7 +37,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Consumer; import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.commons.io.FileUtils; import org.apache.hugegraph.config.HugeConfig; @@ -73,11 +79,11 @@ public class RocksDBSession implements AutoCloseable, Cloneable { private static final int CPUS = Runtime.getRuntime().availableProcessors(); + private static final String TEMP_SUFFIX = "_temp_"; final Statistics rocksDbStats; final WriteOptions writeOptions; final AtomicInteger refCount; final AtomicBoolean shutdown; - final String tempSuffix = "_temp_"; private final transient String graphName; private final HugeConfig hugeConfig; private final ReentrantReadWriteLock cfHandleLock; @@ -300,7 +306,34 @@ public ReentrantReadWriteLock.ReadLock getCfHandleReadLock() { return this.cfHandleLock.readLock(); } + /** + * Resolve the path the constructor would select without opening or pruning databases. + * Returns null for a new database without CURRENT. An incomplete selected version with + * another existing database is rejected rather than treated as a new database. + */ + public static String resolveExistingDBPath(String dbDataPath, String graphName, long version) { + String path = Paths.get(dbDataPath, graphName).toString(); + String selected = findLatestDBPath(path, version, false); + return hasCurrentFile(selected) ? selected : null; + } + + private static boolean hasCurrentFile(String dbPath) { + try { + return Files.readAttributes(Paths.get(dbPath, "CURRENT"), BasicFileAttributes.class).isRegularFile(); + } catch (NoSuchFileException e) { + return false; + } catch (IOException e) { + throw new DBStoreException(String.format("Unable to inspect database path %s", dbPath), e); + } + } + private String findLatestDBPath(String path, long version) { + String selected = findLatestDBPath(path, version, true); + log.info("{} latest db path {}", this.graphName, selected); + return selected; + } + + private static String findLatestDBPath(String path, long version, boolean prune) { File file = new File(path); int strIndex = file.getName().indexOf("_"); @@ -313,7 +346,18 @@ private String findLatestDBPath(String path, long version) { String prefix = defaultName + "_"; File parentFile = new File(file.getParent()); ArrayList> dbs = new ArrayList<>(); - File[] files = parentFile.listFiles(); + File[] files; + if (prune) { + files = parentFile.listFiles(); + } else { + try (Stream paths = Files.list(parentFile.toPath())) { + files = paths.map(Path::toFile).toArray(File[]::new); + } catch (NoSuchFileException e) { + files = null; + } catch (IOException | UncheckedIOException e) { + throw new DBStoreException(String.format("Unable to list database directory %s", parentFile), e); + } + } if (files != null) { dbs.ensureCapacity(files.length); // search all db path @@ -322,7 +366,7 @@ private String findLatestDBPath(String path, long version) { if (!name.startsWith(prefix) && !name.equals(defaultName)) { continue; } - if (name.endsWith(tempSuffix)) { + if (name.endsWith(TEMP_SUFFIX)) { continue; } long v1 = -1L; @@ -345,6 +389,7 @@ private String findLatestDBPath(String path, long version) { RocksDBFactory factory = RocksDBFactory.getInstance(); // get last index db path String latestDBPath = ""; + boolean existingDatabase = false; if (!dbs.isEmpty()) { dbs.sort((o1, o2) -> o1.getKey().equals(o2.getKey()) ? @@ -363,11 +408,15 @@ private String findLatestDBPath(String path, long version) { String.format("%s_%d_%d", defaultName, pair.getKey(), pair.getValue()); } String curDBPath = Paths.get(parentFile.getPath(), curDBName).toString(); + if (!prune && !factory.findPathInRemovedList(curDBPath) && + hasCurrentFile(curDBPath)) { + existingDatabase = true; + } if (i == dbCount - 1) { latestDBPath = curDBPath; } else { // delete old db, do not delete files in the deletion queue - if (!factory.findPathInRemovedList(curDBPath)) { + if (prune && !factory.findPathInRemovedList(curDBPath)) { try { FileUtils.deleteDirectory(new File(curDBPath)); log.info("delete old dbpath {}", curDBPath); @@ -386,8 +435,10 @@ private String findLatestDBPath(String path, long version) { Paths.get(parentFile.getPath(), String.format("%s_%d", defaultName, version)) .toString(); } - // - log.info("{} latest db path {}", this.graphName, latestDBPath); + if (!prune && existingDatabase && !hasCurrentFile(latestDBPath)) { + throw new DBStoreException("Selected database path %s has no CURRENT while another version exists", + latestDBPath); + } return latestDBPath; } @@ -733,6 +784,16 @@ private boolean verifySnapshot(String snapshotPath) { } public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { + this.loadSnapshot(snapshotPath, v1, path -> { }); + } + + /** + * Validate the verified staging directory before replacing a database version. The validator + * must inspect it read-only. Rejection preserves existing databases and the staging directory. + */ + public void loadSnapshot(String snapshotPath, long v1, + Consumer validator) throws DBStoreException { + E.checkArgumentNotNull(validator, "Snapshot validator can't be null"); long startTime = System.currentTimeMillis(); log.info("begin load snapshot from {}", snapshotPath); cfHandleLock.writeLock().lock(); @@ -750,7 +811,7 @@ public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { String defaultName = dbFile.getName().split("_")[0]; String defaultPath = Paths.get(parent, defaultName).toString(); String newDBPath = String.format("%s_%d", defaultPath, v1); - String tempDBPath = newDBPath + tempSuffix; + String tempDBPath = newDBPath + TEMP_SUFFIX; // first link snapshot to temp dir try { @@ -790,6 +851,8 @@ public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { String.format("failed to verify snapshot %s", tempDBPath)); } + validator.accept(tempDBPath); + // move temp to newDBPath try { FileUtils.deleteDirectory(new File(newDBPath)); @@ -804,7 +867,7 @@ public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { } catch (final Exception e) { log.error("failed to load snapshot from {}", snapshotPath); throw new DBStoreException( - String.format("Fail to write snapshot at path %s", snapshotPath)); + String.format("Fail to write snapshot at path %s", snapshotPath), e); } finally { cfHandleLock.writeLock().unlock(); } diff --git a/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java b/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java index 3f80191e4c..2245452490 100644 --- a/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java +++ b/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java @@ -17,26 +17,59 @@ package org.apache.hugegraph.rocksdb.access; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import java.io.IOException; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.PosixFilePermission; +import java.nio.file.attribute.PosixFilePermissions; +import java.security.MessageDigest; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; import org.apache.commons.configuration2.MapConfiguration; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.OptionSpace; +import org.junit.Assume; import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.rocksdb.Checkpoint; +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.InfoLogLevel; +import org.rocksdb.Logger; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; public class RocksDBSessionTest { private final String graphName = "testDummy"; + @Rule + public TemporaryFolder temporary = new TemporaryFolder(); + @BeforeClass public static void init() { OptionSpace.register("rocksdb", @@ -51,6 +84,240 @@ public static void init() { rFactory.setHugeConfig(hConfig); } + @Test + public void resolveExistingPathPreservesVersionsAndConstructorStillPrunes() throws Exception { + Path root = this.temporary.newFolder("versions").toPath(); + for (String name : new String[]{"db", "db_2", "db_2_3", "db_4_1", "db_99_temp_"}) { + writeNative(root.resolve(name), bytes(name)); + } + Map before = fileHashes(root); + assertEquals(root.resolve("db_4_1").toString(), + RocksDBSession.resolveExistingDBPath(root.toString(), "db", 5L)); + assertEquals(before, fileHashes(root)); + assertArrayEquals(bytes("db_4_1"), readNative(root.resolve("db_4_1"))); + try (RocksDBSession session = new RocksDBSession(smallConfig(), root.toString(), "db", 5L)) { + assertEquals(root.resolve("db_4_1").toString(), session.getDbPath()); + assertArrayEquals(bytes("db_4_1"), session.getDB().get(bytes("key"))); + assertFalse(Files.exists(root.resolve("db"))); + assertFalse(Files.exists(root.resolve("db_2"))); + assertFalse(Files.exists(root.resolve("db_2_3"))); + assertTrue(Files.exists(root.resolve("db_99_temp_"))); + } + } + + @Test + public void resolveNewPathDoesNotCreateDatabase() throws Exception { + Path root = this.temporary.newFolder("missing").toPath(); + Path missing = root.resolve("not-created"); + assertNull(RocksDBSession.resolveExistingDBPath(missing.toString(), "db", 0L)); + assertFalse(Files.exists(missing)); + Files.createDirectory(root.resolve("db")); + assertNull(RocksDBSession.resolveExistingDBPath(root.toString(), "db", 0L)); + assertTrue(Files.isDirectory(root.resolve("db"))); + assertFalse(Files.exists(root.resolve("db/CURRENT"))); + } + + @Test + public void resolveIncompleteNewerVersionPreservesExistingDatabase() throws Exception { + Path root = this.temporary.newFolder("incomplete").toPath(); + writeNative(root.resolve("db_1"), bytes("existing")); + Files.createDirectory(root.resolve("db_2")); + Map before = fileHashes(root); + assertThrows(DBStoreException.class, + () -> RocksDBSession.resolveExistingDBPath(root.toString(), "db", 2L)); + assertTrue(Files.isDirectory(root.resolve("db_2"))); + assertEquals(before, fileHashes(root)); + assertArrayEquals(bytes("existing"), readNative(root.resolve("db_1"))); + } + + @Test + public void resolveRejectsUnreadableParent() throws Exception { + Path root = this.temporary.newFolder("unreadable-parent").toPath(); + assertRestrictedPathRejected(root, root, "-wx------", root); + } + + @Test + public void resolveRejectsUnsearchableDatabase() throws Exception { + Path root = this.temporary.newFolder("unsearchable-db").toPath(); + assertRestrictedPathRejected(root, root.resolve("db_2"), "r--------", root); + } + + @Test + public void resolveRejectsUnknownMissingPathUnderDeniedAncestor() throws Exception { + Path root = this.temporary.newFolder("denied-ancestor").toPath(); + assertRestrictedPathRejected(root, root, "---------", root.resolve("missing")); + } + + private static void assertRestrictedPathRejected(Path root, Path restricted, + String mode, Path requested) throws Exception { + Assume.assumeTrue("POSIX permissions are required for the access regression", + Files.getFileStore(root).supportsFileAttributeView("posix")); + Path database = root.resolve("db_2"); + writeNative(database, bytes("existing")); + Map before = fileHashes(root); + Set original = Files.getPosixFilePermissions(restricted); + try { + Files.setPosixFilePermissions(restricted, PosixFilePermissions.fromString(mode)); + boolean denied = restricted.equals(database) ? !Files.isExecutable(restricted) : + !Files.isReadable(restricted); + Assume.assumeTrue("Permission denial is required; root DAC override may bypass it", denied); + DBStoreException failure = assertThrows(DBStoreException.class, + () -> RocksDBSession.resolveExistingDBPath(requested.toString(), "db", 2L)); + assertTrue(failure.getCause() instanceof IOException); + } finally { + Files.setPosixFilePermissions(restricted, original); + } + assertEquals(before, fileHashes(root)); + assertArrayEquals(bytes("existing"), readNative(database)); + assertFalse(Files.exists(root.resolve("db"))); + assertFalse(Files.exists(root.resolve("missing"))); + } + + @Test + public void rejectedSnapshotPreservesActiveVersionAndStaging() throws Exception { + Path root = this.temporary.newFolder("rejected").toPath(); + Path snapshot = root.resolve("snapshot"); + writeNative(root.resolve("producer"), bytes("candidate")); + createNativeSnapshot(root.resolve("producer"), snapshot); + Path target = root.resolve("db_20"); + Path staging = root.resolve("db_20_temp_"); + IllegalStateException rejection = new IllegalStateException("candidate content rejected"); + AtomicInteger validated = new AtomicInteger(); + AtomicReference> stagedBefore = new AtomicReference<>(); + try (RocksDBSession active = new RocksDBSession(smallConfig(), root.toString(), "db", 0L)) { + active.getDB().put(bytes("key"), bytes("active")); + writeNative(target, bytes("previous-version")); + String activePath = active.getDbPath(); + long sequence = active.getLatestSequenceNumber(); + Map activeBefore = fileHashes(Path.of(activePath)); + Map targetBefore = fileHashes(target); + Map snapshotBefore = fileHashes(snapshot); + DBStoreException failed = assertThrows(DBStoreException.class, () -> + active.loadSnapshot(snapshot.toString(), 20L, path -> { + assertEquals(staging.toString(), path); + try { + assertArrayEquals(bytes("candidate"), readNative(Path.of(path))); + stagedBefore.set(fileHashes(Path.of(path))); + } catch (Exception e) { + throw new AssertionError(e); + } + validated.incrementAndGet(); + throw rejection; + })); + assertSame(rejection, failed.getCause()); + assertEquals(1, validated.get()); + assertEquals(activePath, active.getDbPath()); + assertEquals(sequence, active.getLatestSequenceNumber()); + assertArrayEquals(bytes("active"), active.getDB().get(bytes("key"))); + assertEquals(activeBefore, fileHashes(Path.of(activePath))); + assertEquals(targetBefore, fileHashes(target)); + assertEquals(snapshotBefore, fileHashes(snapshot)); + assertArrayEquals(bytes("previous-version"), readNative(target)); + assertArrayEquals(bytes("candidate"), readNative(staging)); + assertEquals(stagedBefore.get(), fileHashes(staging)); + } + } + + @Test + public void acceptedSnapshotAndOriginalOverloadActivateNativeData() throws Exception { + Path root = this.temporary.newFolder("accepted").toPath(); + Path snapshot = root.resolve("snapshot"); + writeNative(root.resolve("producer"), bytes("candidate")); + createNativeSnapshot(root.resolve("producer"), snapshot); + AtomicInteger validated = new AtomicInteger(); + try (RocksDBSession active = new RocksDBSession(smallConfig(), root.toString(), "db", 0L)) { + active.loadSnapshot(snapshot.toString(), 20L, path -> { + try { + assertArrayEquals(bytes("candidate"), readNative(Path.of(path))); + } catch (Exception e) { + throw new AssertionError(e); + } + validated.incrementAndGet(); + }); + assertEquals(1, validated.get()); + assertFalse(Files.exists(root.resolve("db_20_temp_"))); + assertArrayEquals(bytes("candidate"), readNative(root.resolve("db_20"))); + active.loadSnapshot(snapshot.toString(), 21L); + assertArrayEquals(bytes("candidate"), readNative(root.resolve("db_21"))); + } + try (RocksDBSession reopened = new RocksDBSession(smallConfig(), root.toString(), "db", 21L)) { + assertEquals(root.resolve("db_21").toString(), reopened.getDbPath()); + assertArrayEquals(bytes("candidate"), reopened.getDB().get(bytes("key"))); + } + } + + private static HugeConfig smallConfig() { + Map values = new HashMap<>(); + values.put("rocksdb.optimize_mode", false); + values.put("rocksdb.write_buffer_size", 65536L); + values.put("rocksdb.max_background_jobs", 1); + values.put("rocksdb.max_subcompactions", 1); + values.put("rocksdb.log_level", "ERROR"); + return new HugeConfig(new MapConfiguration(values)); + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } + + private static void writeNative(Path path, byte[] value) throws Exception { + try (Options options = new Options().setCreateIfMissing(true).setWriteBufferSize(65536L); + RocksDB db = RocksDB.open(options, path.toString())) { + db.put(bytes("key"), value); + } + } + + private static void createNativeSnapshot(Path source, Path snapshot) throws Exception { + try (Options options = new Options(); + RocksDB db = RocksDB.open(options, source.toString()); + Checkpoint checkpoint = Checkpoint.create(db)) { + checkpoint.createCheckpoint(snapshot.toString()); + } + } + + private static byte[] readNative(Path path) throws Exception { + try (Options options = new Options(); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions(); + DBOptions dbOptions = new DBOptions(); + Logger logger = new Logger(dbOptions) { + @Override + protected void log(InfoLogLevel level, String message) { + // Native readback must not create or replace the DB's LOG file. + } + }) { + dbOptions.setLogger(logger); + List descriptors = new ArrayList<>(); + for (byte[] name : RocksDB.listColumnFamilies(options, path.toString())) { + descriptors.add(new ColumnFamilyDescriptor(name, cfOptions)); + } + List handles = new ArrayList<>(); + try (RocksDB db = RocksDB.openReadOnly(dbOptions, path.toString(), descriptors, handles)) { + try { + return db.get(handles.get(0), bytes("key")); + } finally { + for (ColumnFamilyHandle handle : handles) { + handle.close(); + } + } + } + } + } + + private static Map fileHashes(Path root) throws Exception { + Map hashes = new TreeMap<>(); + try (Stream paths = Files.walk(root)) { + for (Path path : paths.filter(Files::isRegularFile).collect(java.util.stream.Collectors.toList())) { + byte[] digest = MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(path)); + StringBuilder encoded = new StringBuilder(); + for (byte value : digest) { + encoded.append(String.format("%02x", value)); + } + hashes.put(root.relativize(path).toString(), encoded.toString()); + } + } + return hashes; + } + private static byte[] intToBytesForPartId(int v) { short s = (short) v; ByteBuffer buffer = ByteBuffer.allocate(Short.BYTES).order(ByteOrder.BIG_ENDIAN); diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index 17dbdb32ae..7feeef0eda 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -242,6 +242,7 @@ **/PartitionEngineErrorTest.java **/BatchGraphIsolationTest.java **/GraphIDManagerTest.java + **/GraphIdUpgradePreflightTest.java **/SchemaDriverTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java new file mode 100644 index 0000000000..43b8e24d21 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java @@ -0,0 +1,390 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph.store.core.store; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; + +import org.apache.hugegraph.rocksdb.access.RocksDBSession; +import org.apache.hugegraph.rocksdb.access.SessionOperator; +import org.apache.hugegraph.store.HgStoreEngine; +import org.apache.hugegraph.store.PartitionEngine; +import org.apache.hugegraph.store.business.BusinessHandlerImpl; +import org.apache.hugegraph.store.meta.PartitionManager; +import org.apache.hugegraph.store.meta.ShardGroup; +import org.apache.hugegraph.store.meta.StoreMetadata; +import org.apache.hugegraph.store.options.HgStoreEngineOptions; +import org.apache.hugegraph.store.options.PartitionEngineOptions; +import org.apache.hugegraph.store.raft.PartitionStateMachine; +import org.apache.hugegraph.store.pd.PdProvider; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; +import org.apache.hugegraph.store.term.Bits; +import org.apache.hugegraph.store.util.HgStoreException; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.rocksdb.Checkpoint; +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.RocksDB; + +import com.google.protobuf.Int64Value; + +import com.alipay.sofa.jraft.Status; +import com.alipay.sofa.jraft.closure.ClosureQueueImpl; +import com.alipay.sofa.jraft.closure.LoadSnapshotClosure; +import com.alipay.sofa.jraft.core.FSMCallerImpl; +import com.alipay.sofa.jraft.core.NodeImpl; +import com.alipay.sofa.jraft.core.NodeMetrics; +import com.alipay.sofa.jraft.entity.LogId; +import com.alipay.sofa.jraft.entity.RaftOutter; +import com.alipay.sofa.jraft.option.FSMCallerOptions; +import com.alipay.sofa.jraft.storage.LogManager; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; + +public class GraphIdUpgradePreflightTest { + + private static final int PARTITION_ID = 0; + private static final String DB_NAME = "00000"; + private static final String[] TABLES = {"g+v", "g+oe", "g+ie", "g+index"}; + private static final byte[] VALUE = {42}; + + @Rule + public TemporaryFolder temporary = new TemporaryFolder(); + + @BeforeClass + public static void initRocksdb() { + Map config = new HashMap<>(); + config.put("rocksdb.write_buffer_size", "1048576"); + config.put("rocksdb.bloom_filter_bits_per_key", "10"); + BusinessHandlerImpl.initRocksdb(config, null); + } + + @Test + public void testLegacyDataRejectsBeforeRaftAndPreservesVersions() throws Exception { + Path location = this.temporary.newFolder("existing-location").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME, 1); + writeDatabase(data, DB_NAME + "_4", 65534); + Map before = fileHashes(location); + PartitionEngine engine = partitionEngine(location, true); + PartitionEngineOptions options = new PartitionEngineOptions(); + options.setGroupId(PARTITION_ID); + + HgStoreException failure = assertThrows(HgStoreException.class, () -> engine.init(options)); + + assertTrue(failure.getMessage().contains("GraphID")); + assertEquals(false, field(engine, "started")); + assertNull(engine.getRaftNode()); + assertNull(field(engine, "stateMachine")); + assertNull(engine.getTaskManager()); + assertEquals(before, fileHashes(location)); + } + + @Test + public void testMissingLocationChecksAllConfiguredRootsWithoutWriting() throws Exception { + Path first = this.temporary.newFolder("first-root").toPath(); + Path second = this.temporary.newFolder("second-root").toPath(); + writeDatabase(Files.createDirectory(first.resolve("db")), DB_NAME, 1); + writeDatabase(Files.createDirectory(second.resolve("db")), DB_NAME, 65534); + Map firstBefore = fileHashes(first); + Map secondBefore = fileHashes(second); + PartitionEngine engine = partitionEngine(first, false); + engine.getStoreEngine().getOption().setDataPath(first + "," + second); + PartitionEngineOptions options = new PartitionEngineOptions(); + options.setGroupId(PARTITION_ID); + + assertThrows(HgStoreException.class, () -> engine.init(options)); + + assertNull(engine.getRaftNode()); + assertEquals(firstBefore, fileHashes(first)); + assertEquals(secondBefore, fileHashes(second)); + } + + @Test + public void testLegacySnapshotKeepsActiveAndTargetData() throws Exception { + Path data = this.temporary.newFolder("active-parent").toPath(); + try (RocksDBSession active = session(data, "active")) { + putRows(active, 1); + writeDatabase(data, "active_5", 2); + Path snapshot = snapshot(65534); + Map before = fileHashes(data); + Map incomingBefore = fileHashes(snapshot); + BusinessHandlerImpl handler = handler(active); + assertEquals(1, handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + Map cached = (Map) field(handler.getKeyCreator(), "graphIdCache"); + Object owner = cached.get(PARTITION_ID); + + HgStoreException failure = assertThrows(HgStoreException.class, + () -> handler.loadSnapshot(snapshot.toString(), "", PARTITION_ID, 5L)); + + assertTrue(failure.getMessage().contains("GraphID")); + assertNotNull(failure.getCause()); + assertSame(owner, cached.get(PARTITION_ID)); + assertEquals(before, fileHashes(data)); + assertEquals(incomingBefore, fileHashes(snapshot)); + assertTrue(Files.isDirectory(data.resolve("active_5_temp_"))); + assertRows(active, 1); + assertArrayEquals(Int64Value.of(1L).toByteArray(), active.sessionOp().get("default", + "HUGEGRAPH/GRAPH_ID/test".getBytes(java.nio.charset.StandardCharsets.UTF_8))); + } + } + + @Test + public void testRejectedNativeSnapshotDoesNotPublishEitherIndex() throws Exception { + assertSnapshotIndices(true); + } + + @Test + public void testAcceptedNativeSnapshotPublishesBothIndices() throws Exception { + assertSnapshotIndices(false); + } + + private void assertSnapshotIndices(boolean legacy) throws Exception { + Path data = this.temporary.newFolder("fsm-active").toPath(); + try (RocksDBSession active = session(data, "fsm")) { + putRows(active, 1); + Path snapshot = snapshot(legacy ? 65534 : 2); + Path snapshotRoot = snapshot.getParent(); + BusinessHandlerImpl handler = handler(active); + HgStoreEngine store = mock(HgStoreEngine.class); + PartitionEngine engine = mock(PartitionEngine.class); + when(engine.getStoreEngine()).thenReturn(store); + when(engine.getGroupId()).thenReturn(PARTITION_ID); + when(engine.loadPartitionsFromLocalDb()).thenReturn(Collections.emptyList()); + when(store.getBusinessHandler()).thenReturn(handler); + PartitionStateMachine machine = new PartitionStateMachine(PARTITION_ID, + new SnapshotHandler(engine)); + Field committed = PartitionStateMachine.class.getDeclaredField("committedIndex"); + committed.setAccessible(true); + committed.setLong(machine, 3L); + SnapshotReader reader = mock(SnapshotReader.class); + when(reader.getPath()).thenReturn(snapshotRoot.toString()); + when(reader.load()).thenReturn(RaftOutter.SnapshotMeta.newBuilder() + .setLastIncludedIndex(5L) + .setLastIncludedTerm(1L).build()); + NodeImpl node = mock(NodeImpl.class); + when(node.getNodeMetrics()).thenReturn(new NodeMetrics(false)); + when(node.getGroupId()).thenReturn("graph-id-snapshot-test"); + FSMCallerOptions options = new FSMCallerOptions(); + options.setNode(node); + options.setFsm(machine); + options.setLogManager(mock(LogManager.class)); + options.setClosureQueue(new ClosureQueueImpl("graph-id-snapshot-test")); + options.setBootstrapId(new LogId(3L, 1L)); + options.setDisruptorBufferSize(16); + FSMCallerImpl caller = new FSMCallerImpl(); + assertTrue(caller.init(options)); + AtomicReference result = new AtomicReference<>(); + CountDownLatch done = new CountDownLatch(1); + try { + assertTrue(caller.onSnapshotLoad(new LoadSnapshotClosure() { + @Override + public SnapshotReader start() { + return reader; + } + + @Override + public void run(Status status) { + result.set(status); + done.countDown(); + } + })); + assertTrue(done.await(5, TimeUnit.SECONDS)); + assertEquals(!legacy, result.get().isOk()); + assertEquals(legacy ? 3L : 5L, machine.getCommittedIndex()); + assertEquals(legacy ? 3L : 5L, caller.getLastAppliedIndex()); + assertEquals(legacy ? 3L : 5L, caller.getLastCommittedIndex()); + assertEquals(!legacy, Files.exists(snapshotRoot.resolve("should_not_load"))); + if (legacy) { + assertRows(active, 1); + assertFalse(Files.exists(data.resolve("fsm_5"))); + } else { + assertTrue(Files.isDirectory(data.resolve("fsm_5"))); + try (RocksDBSession restored = session(data, "fsm")) { + assertRows(restored, 2); + BusinessHandlerImpl restoredHandler = handler(restored); + assertEquals(2, restoredHandler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + for (String table : TABLES) { + assertArrayEquals(VALUE, restoredHandler.doGet("test", 0, table, new byte[]{1})); + } + } + } + } finally { + caller.shutdown(); + caller.join(); + } + } + } + + private PartitionEngine partitionEngine(Path location, boolean persistedLocation) { + HgStoreEngine store = mock(HgStoreEngine.class); + PartitionManager manager = mock(PartitionManager.class); + StoreMetadata metadata = mock(StoreMetadata.class); + HgStoreEngineOptions options = new HgStoreEngineOptions(); + options.setDataPath(location.toString()); + when(store.getOption()).thenReturn(options); + when(store.getPartitionManager()).thenReturn(manager); + when(manager.getStoreMetadata()).thenReturn(metadata); + if (persistedLocation) { + when(metadata.getPartitionStore(PARTITION_ID)).thenReturn( + Metapb.PartitionStore.newBuilder().setPartitionId(PARTITION_ID) + .setStoreLocation(location.toString()).build()); + } + return new PartitionEngine(store, mock(ShardGroup.class)); + } + + private static BusinessHandlerImpl handler(RocksDBSession active) { + PartitionManager manager = mock(PartitionManager.class); + PdProvider provider = mock(PdProvider.class); + when(manager.getPdProvider()).thenReturn(provider); + when(manager.hasPartition(anyString(), anyInt())).thenReturn(true); + when(provider.getPartitionByCode(anyString(), anyInt())).thenReturn( + Metapb.Partition.newBuilder().setId(PARTITION_ID).build()); + return new BusinessHandlerImpl(manager) { + @Override + public RocksDBSession getSession(int partitionId) { + return active.clone(); + } + }; + } + + private Path snapshot(int graphId) throws Exception { + Path seed = this.temporary.newFolder("snapshot-seed").toPath(); + Path root = this.temporary.newFolder("snapshot-root").toPath(); + Path data = root.resolve("data"); + try (RocksDBSession session = session(seed, "seed")) { + putRows(session, graphId); + try (Checkpoint checkpoint = Checkpoint.create(session.getDB())) { + checkpoint.createCheckpoint(data.toString()); + } + } + return data; + } + + private static void writeDatabase(Path root, String name, int graphId) { + List handles = new ArrayList<>(); + try (DBOptions options = new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions().setWriteBufferSize(1048576)) { + List descriptors = new ArrayList<>(); + descriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOptions)); + for (String table : TABLES) { + descriptors.add(new ColumnFamilyDescriptor(table.getBytes(java.nio.charset.StandardCharsets.UTF_8), + cfOptions)); + } + try (RocksDB database = RocksDB.open(options, root.resolve(name).toString(), descriptors, handles)) { + for (int i = 1; i < handles.size(); i++) { + database.put(handles.get(i), key(graphId), VALUE); + } + } finally { + handles.forEach(ColumnFamilyHandle::close); + } + } catch (Exception failure) { + throw new IllegalStateException(failure); + } + } + + private static Object field(Object object, String name) throws Exception { + Field field = object.getClass().getDeclaredField(name); + field.setAccessible(true); + return field.get(object); + } + + private static RocksDBSession session(Path root, String name) { + return new RocksDBSession(org.apache.hugegraph.rocksdb.access.RocksDBFactory + .getInstance().getHugeConfig(), root.toString(), name, 0L); + } + + private static void putRows(RocksDBSession session, int graphId) { + session.createTables(TABLES); + SessionOperator operator = session.sessionOp(); + operator.prepare(); + try { + for (String table : TABLES) { + operator.put(table, key(graphId), VALUE); + } + if (graphId != 65534) { + operator.put("default", "HUGEGRAPH/GRAPH_ID/test".getBytes(java.nio.charset.StandardCharsets.UTF_8), + Int64Value.of(graphId).toByteArray()); + } + operator.commit(); + } catch (RuntimeException failure) { + operator.rollback(); + throw failure; + } + } + + private static void assertRows(RocksDBSession session, int graphId) { + for (String table : TABLES) { + assertArrayEquals(VALUE, session.sessionOp().get(table, key(graphId))); + } + } + + private static byte[] key(int graphId) { + byte[] key = {0, 0, 1, 0, 0}; + Bits.putShort(key, 0, graphId); + return key; + } + + private static Map fileHashes(Path root) throws Exception { + Map hashes = new LinkedHashMap<>(); + try (Stream paths = Files.walk(root)) { + for (Path path : (Iterable) paths.filter(Files::isRegularFile) + .sorted()::iterator) { + // Rejected loads deliberately retain a new staging directory. + if (path.toString().contains("_temp_")) { + continue; + } + hashes.put(root.relativize(path).toString(), + Arrays.toString(MessageDigest.getInstance("SHA-256") + .digest(Files.readAllBytes(path)))); + } + } + return hashes; + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java index ac0d5d0d1e..9b3314a722 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java @@ -91,6 +91,160 @@ public void clear() { } } + @Test + public void testLegacyVertexRowsBlockAllocation() { + this.assertLegacyRowsBlockAllocation(VERTEX_TABLE, false); + } + + @Test + public void testLegacyOutEdgeRowsBlockAllocationWithoutVertexTable() { + this.assertLegacyRowsBlockAllocation(OUT_EDGE_TABLE, false); + } + + @Test + public void testLegacyInEdgeRowsBlockAllocationWithoutVertexTable() { + this.assertLegacyRowsBlockAllocation(IN_EDGE_TABLE, false); + } + + @Test + public void testLegacyIndexRowsBlockAllocationWithoutVertexTable() { + this.assertLegacyRowsBlockAllocation(INDEX_TABLE, false); + } + + @Test + public void testLegacyRowsBlockAllocationDespiteExistingMapping() { + this.assertLegacyRowsBlockAllocation(INDEX_TABLE, true); + } + + @Test + public void testLegacyCheckPreservesCleanAndLegalBoundaryData() { + GraphIdManager.maxGraphID = this.previousMaxGraphId; + Set cleanTables = new HashSet<>(this.session.getTables().keySet()); + GraphIdManager.checkLegacyGraphIdData(this.session, PARTITION_ID); + Assert.assertEquals(cleanTables, this.session.getTables().keySet()); + Assert.assertFalse(this.session.tableIsExist(VERTEX_TABLE)); + + GraphIdManager manager = this.newManager(); + manager.updateGraphIds(Collections.singletonMap("boundary", 0xFFFDL)); + this.session.createTables(OUT_EDGE_TABLE); + byte[] key = {(byte) 0xff, (byte) 0xfd, 1, 0, 0}; + byte[] value = {43}; + SessionOperator operator = this.session.sessionOp(); + operator.prepare(); + try { + operator.put(OUT_EDGE_TABLE, key, value); + operator.commit(); + } catch (RuntimeException e) { + operator.rollback(); + throw e; + } + Set tablesBefore = new HashSet<>(this.session.getTables().keySet()); + this.session.flush(true); + Map filesBefore = this.databaseFiles(); + GraphIdManager.checkLegacyGraphIdData(this.session, PARTITION_ID); + GraphIdManager.checkLegacyGraphIdData(this.session.getDbPath(), PARTITION_ID); + Assert.assertEquals(filesBefore, this.databaseFiles()); + Assert.assertEquals(0L, manager.getGraphIdOrCreate("clean-control")); + Assert.assertEquals(0xFFFDL, this.newManager().getGraphId("boundary")); + Assert.assertArrayEquals(value, operator.get(OUT_EDGE_TABLE, key)); + Assert.assertEquals(tablesBefore, this.session.getTables().keySet()); + Assert.assertTrue(this.session.getIteratorMap().isEmpty()); + } + + private void assertLegacyRowsBlockAllocation(String table, boolean existingMapping) { + GraphIdManager.maxGraphID = this.previousMaxGraphId; + GraphIdManager manager = this.newManager(); + if (existingMapping) { + manager.updateGraphIds(Collections.singletonMap("existing", 1L)); + } + this.session.createTables(table); + byte[] key = {(byte) 0xff, (byte) 0xfe, 1, 0, 0}; + byte[] value = {42}; + SessionOperator operator = this.session.sessionOp(); + operator.prepare(); + try { + operator.put(table, key, value); + operator.commit(); + } catch (RuntimeException e) { + operator.rollback(); + throw e; + } + Map metadataBefore = this.metadataRows(); + Set tablesBefore = new HashSet<>(this.session.getTables().keySet()); + this.session.flush(true); + Map filesBefore = this.databaseFiles(); + Assert.assertThrows(HgStoreException.class, + () -> GraphIdManager.checkLegacyGraphIdData(this.session.getDbPath(), + PARTITION_ID)); + Assert.assertEquals(filesBefore, this.databaseFiles()); + HgStoreException failure = Assert.assertThrows(HgStoreException.class, + () -> manager.getGraphIdOrCreate("legacy-first-batch")); + Assert.assertTrue(failure.getMessage().contains("partition " + PARTITION_ID)); + Assert.assertTrue(failure.getMessage().contains("table " + table)); + Assert.assertTrue(failure.getMessage().contains("prefix 0xFFFE")); + Assert.assertEquals(filesBefore, this.databaseFiles()); + Assert.assertEquals(metadataBefore, this.metadataRows()); + Assert.assertEquals(tablesBefore, this.session.getTables().keySet()); + Assert.assertArrayEquals(value, operator.get(table, key)); + Assert.assertEquals(0xFFFE, manager.getGraphId("legacy-first-batch")); + Assert.assertNull(manager.get(Int64Value.parser(), + MetadataKeyHelper.getGraphIDKey("legacy-first-batch"))); + Assert.assertNull(manager.get(Int64Value.parser(), + manager.genCIDSlotKey(GraphIdManager.GRAPH_ID_PREFIX, 0L))); + Assert.assertNull(manager.get(Int64Value.parser(), + MetadataKeyHelper.getCidKey(GraphIdManager.GRAPH_ID_PREFIX))); + if (existingMapping) { + Assert.assertEquals(1L, this.newManager().getGraphId("existing")); + } + this.reopenDatabase(); + Assert.assertArrayEquals(value, this.session.sessionOp().get(table, key)); + Assert.assertEquals(metadataBefore, this.metadataRows()); + Assert.assertThrows(HgStoreException.class, + () -> GraphIdManager.checkLegacyGraphIdData(this.session, PARTITION_ID)); + Assert.assertTrue(this.session.getIteratorMap().isEmpty()); + } + + @Test + public void testLegacyPathCheckDoesNotCreateMissingDatabase() { + File missing = new File(this.dbPath, "not-created"); + Assert.assertFalse(missing.exists()); + Assert.assertThrows(HgStoreException.class, + () -> GraphIdManager.checkLegacyGraphIdData(missing.getAbsolutePath(), + PARTITION_ID)); + Assert.assertFalse(missing.exists()); + } + + private Map databaseFiles() { + Map files = new LinkedHashMap<>(); + java.nio.file.Path db = new File(this.session.getDbPath()).toPath(); + try (var paths = java.nio.file.Files.walk(db)) { + paths.filter(java.nio.file.Files::isRegularFile).sorted().forEach(path -> { + try { + byte[] digest = java.security.MessageDigest.getInstance("SHA-256") + .digest(java.nio.file.Files.readAllBytes(path)); + files.put(db.relativize(path).toString(), java.util.HexFormat.of().formatHex(digest)); + } catch (java.io.IOException | java.security.NoSuchAlgorithmException e) { + throw new AssertionError(e); + } + }); + } catch (java.io.IOException e) { + throw new AssertionError(e); + } + return files; + } + + private Map metadataRows() { + Map rows = new LinkedHashMap<>(); + try (var iterator = this.session.sessionOp().scan("default")) { + while (iterator.hasNext()) { + RocksDBSession.BackendColumn column = iterator.next(); + rows.put(java.util.HexFormat.of().formatHex(column.name), + java.util.HexFormat.of().formatHex(column.value)); + } + } + return rows; + } + @Test public void testAllocateBoundaryIdBeforeWrap() { GraphIdManager manager = this.newManager(); From ac1dba2645fc21694024a3d5139a509034329fb0 Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 04:25:12 +0800 Subject: [PATCH 2/5] fix(store): preflight graph data before cold open - validate retained DB versions before native pruning - reuse existing factory locks and cached sessions - cover the real metadata bootstrap entry point - preserve version data and document metadata limits --- docs/store-graphid-upgrade.md | 11 +- .../store/business/BusinessHandlerImpl.java | 2 +- hugegraph-store/hg-store-rocksdb/README.md | 8 + .../rocksdb/access/RocksDBFactory.java | 10 + .../rocksdb/access/RocksDBSession.java | 15 +- .../store/GraphIdUpgradePreflightTest.java | 219 ++++++++++++++++++ 6 files changed, 259 insertions(+), 6 deletions(-) diff --git a/docs/store-graphid-upgrade.md b/docs/store-graphid-upgrade.md index 2225eebfbd..b9cc849402 100644 --- a/docs/store-graphid-upgrade.md +++ b/docs/store-graphid-upgrade.md @@ -7,16 +7,19 @@ them. This condition is specific to affected old write paths; it does not mean that every 1.7 database needs migration. Store checks existing graph-data column families before allocating a missing -graph ID, starting a partition's Raft state machine, or activating a snapshot. +graph ID, first opening a partition DB during metadata bootstrap, starting a +partition's Raft state machine, or activating a snapshot. If it finds the reserved prefix, it rejects that operation and reports the partition and column family. Metadata-only column families are excluded. The checks do not assign an owner to legacy rows or change the persisted ID format. The partition preflight resolves existing DB versions without creating a DB, allocating a store location or pruning older versions. It runs before partition -task/FSM construction and Raft replay; other existing Store bootstrap steps can -already have opened databases. It is not a promise that the entire startup -process leaves every file untouched. +task/FSM construction and Raft replay. The first native DB open also validates +every existing version not already queued for removal before constructor pruning +or cache publication. Cached sessions are reused without repeating these scans. +Bootstrap can update Store/Raft location metadata before this check; the refusal +preserves data DB versions, not every file touched by the startup process. A rejected snapshot retains the active DB and its index, the incoming snapshot, and the staged copy for diagnosis. Successful snapshots still use the existing diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java index b88447369e..b3dd9f8c40 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java @@ -1500,7 +1500,7 @@ public RocksDBSession getSession(int partId) throws HgStoreException { long version = HgStoreEngine.getInstance().getCommittedIndex(partId); dbSession = factory.createGraphDB(partitionManager.getDbDataPath(partId, dbName), dbName, - version); + version, path -> GraphIdManager.checkLegacyGraphIdData(path, partId)); if (dbSession == null) { log.info("failed to create a new graph db: {}", dbName); throw new HgStoreException(HgStoreException.EC_RKDB_CREATE_FAIL, diff --git a/hugegraph-store/hg-store-rocksdb/README.md b/hugegraph-store/hg-store-rocksdb/README.md index 1e4c261d6e..3946f6bd2d 100644 --- a/hugegraph-store/hg-store-rocksdb/README.md +++ b/hugegraph-store/hg-store-rocksdb/README.md @@ -26,6 +26,14 @@ path. Directory listing or `CURRENT` inspection failures are rejected; unknown access or I/O state is not treated as an absent database. Paths queued for removal retain the constructor's existing treatment. The normal constructor continues to open read/write and prune old versions. +`RocksDBFactory.createGraphDB(path, name, version, Consumer validator)` +validates existing DB versions before creating an uncached native session. It +uses the existing factory creation lock and cache: cached sessions skip validation. +Each existing candidate not queued for removal is passed to the validator through +the read-only resolver before constructor pruning or cache publication. Validation +failure leaves these DB paths intact. The original overloads keep their behavior; +this generic hook does not depend on Store graph-ID rules. + `loadSnapshot(snapshotPath, version, Consumer validator)` first copies and verifies the staging database, then passes its path to the validator before replacing the target version. The validator must inspect it read-only. A rejected diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java index 2e8e0bae68..52b0b66cd9 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java @@ -31,6 +31,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Consumer; import java.util.stream.Collectors; import org.apache.commons.io.FileUtils; @@ -185,6 +186,11 @@ public RocksDBSession createGraphDB(String dbPath, String dbName) { } public RocksDBSession createGraphDB(String dbPath, String dbName, long version) { + return createGraphDB(dbPath, dbName, version, null); + } + + public RocksDBSession createGraphDB(String dbPath, String dbName, long version, + Consumer existingDatabaseValidator) { if (closing.get()) { throw new RuntimeException("db closed"); } @@ -192,6 +198,10 @@ public RocksDBSession createGraphDB(String dbPath, String dbName, long version) try { RocksDBSession dbSession = dbSessionMap.get(dbName); if (dbSession == null) { + if (existingDatabaseValidator != null) { + RocksDBSession.resolveExistingDBPath(dbPath, dbName, version, + existingDatabaseValidator); + } log.info("create rocksdb for {}", dbName); dbSession = new RocksDBSession(this.hugeConfig, dbPath, dbName, version); dbSessionMap.put(dbName, dbSession); diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java index 319e878896..18c6a853b4 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java @@ -312,8 +312,13 @@ public ReentrantReadWriteLock.ReadLock getCfHandleReadLock() { * another existing database is rejected rather than treated as a new database. */ public static String resolveExistingDBPath(String dbDataPath, String graphName, long version) { + return resolveExistingDBPath(dbDataPath, graphName, version, null); + } + + public static String resolveExistingDBPath(String dbDataPath, String graphName, long version, + Consumer existingDatabaseValidator) { String path = Paths.get(dbDataPath, graphName).toString(); - String selected = findLatestDBPath(path, version, false); + String selected = findLatestDBPath(path, version, false, existingDatabaseValidator); return hasCurrentFile(selected) ? selected : null; } @@ -334,6 +339,11 @@ private String findLatestDBPath(String path, long version) { } private static String findLatestDBPath(String path, long version, boolean prune) { + return findLatestDBPath(path, version, prune, null); + } + + private static String findLatestDBPath(String path, long version, boolean prune, + Consumer existingDatabaseValidator) { File file = new File(path); int strIndex = file.getName().indexOf("_"); @@ -411,6 +421,9 @@ private static String findLatestDBPath(String path, long version, boolean prune) if (!prune && !factory.findPathInRemovedList(curDBPath) && hasCurrentFile(curDBPath)) { existingDatabase = true; + if (existingDatabaseValidator != null) { + existingDatabaseValidator.accept(curDBPath); + } } if (i == dbCount - 1) { latestDBPath = curDBPath; diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java index 43b8e24d21..13c3880e6b 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java @@ -47,6 +47,7 @@ import java.util.stream.Stream; import org.apache.hugegraph.rocksdb.access.RocksDBSession; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; import org.apache.hugegraph.rocksdb.access.SessionOperator; import org.apache.hugegraph.store.HgStoreEngine; import org.apache.hugegraph.store.PartitionEngine; @@ -54,6 +55,9 @@ import org.apache.hugegraph.store.meta.PartitionManager; import org.apache.hugegraph.store.meta.ShardGroup; import org.apache.hugegraph.store.meta.StoreMetadata; +import org.apache.hugegraph.store.meta.Store; +import org.apache.hugegraph.store.meta.GraphIdManager; +import org.apache.hugegraph.store.meta.MetadataKeyHelper; import org.apache.hugegraph.store.options.HgStoreEngineOptions; import org.apache.hugegraph.store.options.PartitionEngineOptions; import org.apache.hugegraph.store.raft.PartitionStateMachine; @@ -105,6 +109,221 @@ public static void initRocksdb() { BusinessHandlerImpl.initRocksdb(config, null); } + @Test + public void testBootstrapPreservesOlderDatabaseWhenLatestHasNoCurrent() throws Exception { + Path location = this.temporary.newFolder("bootstrap").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + Files.createDirectory(data.resolve(DB_NAME + "_2")); + Files.write(data.resolve(DB_NAME + "_2/incomplete-copy"), new byte[]{7}); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location)) { + Throwable failure = null; + try { + bootstrap.manager.loadPartition(); + } catch (Throwable caught) { + failure = caught; + } + System.out.println("BOOTSTRAP older-present=" + Files.exists(data.resolve(DB_NAME + "_1")) + + " latest-CURRENT=" + Files.exists(data.resolve(DB_NAME + "_2/CURRENT")) + + " unchanged=" + before.equals(fileHashes(data)) + " failure=" + failure); + assertNotNull("Bootstrap must reject before creating/pruning native databases", failure); + assertTrue(failure.getMessage().contains("no CURRENT")); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + @Test + public void testBootstrapPreservesLegacyOlderVersionBeforePruning() throws Exception { + Path location = this.temporary.newFolder("bootstrap-older-legacy").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 65534); + writeDatabase(data, DB_NAME + "_2", 1); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location)) { + Throwable failure = null; + try { + bootstrap.manager.loadPartition(); + } catch (Throwable caught) { + failure = caught; + } + System.out.println("BOOTSTRAP older-legacy-present=" + + Files.exists(data.resolve(DB_NAME + "_1")) + + " unchanged=" + before.equals(fileHashes(data)) + " failure=" + failure); + assertNotNull("Pruning must not erase an older legacy database", failure); + assertTrue(failure.getMessage().contains("GraphID")); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + @Test + public void testBootstrapPreservesLegacyLatestVersion() throws Exception { + Path location = this.temporary.newFolder("bootstrap-latest-legacy").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + writeDatabase(data, DB_NAME + "_2", 65534); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location)) { + assertThrows(HgStoreException.class, bootstrap.manager::loadPartition); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + @Test + public void testBootstrapLoadsCurrentValidMappingAndData() throws Exception { + Path location = this.temporary.newFolder("bootstrap-valid").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + writeBootstrapMetadata(data.resolve(DB_NAME + "_1"), 1); + try (Bootstrap bootstrap = new Bootstrap(location)) { + bootstrap.manager.loadPartition(); + assertTrue(bootstrap.manager.hasPartition("test", PARTITION_ID)); + assertEquals(1, bootstrap.handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + for (String table : TABLES) { + assertArrayEquals(VALUE, bootstrap.handler.doGet("test", 0, table, new byte[]{1})); + } + } + } + + @Test + public void testFactoryValidatesColdOpenAndReusesCachedSession() throws Exception { + Path data = this.temporary.newFolder("factory-valid").toPath(); + writeDatabase(data, DB_NAME, 1); + java.util.concurrent.atomic.AtomicInteger checks = new java.util.concurrent.atomic.AtomicInteger(); + RocksDBFactory factory = RocksDBFactory.getInstance(); + try { + try (RocksDBSession first = factory.createGraphDB(data.toString(), DB_NAME, 0L, path -> { + checks.incrementAndGet(); + GraphIdManager.checkLegacyGraphIdData(path, PARTITION_ID); + }); + RocksDBSession cached = factory.createGraphDB(data.toString(), DB_NAME, 0L, + path -> org.junit.Assert.fail("Cached session must not be revalidated"))) { + assertEquals(1, checks.get()); + assertSame(first.getDB(), cached.getDB()); + assertRows(cached, 1); + } + } finally { + factory.releaseGraphDB(DB_NAME); + } + } + + @Test + public void testNewPartitionDatabaseStillCreatesNormally() throws Exception { + Path location = this.temporary.newFolder("bootstrap-new").toPath(); + try (Bootstrap bootstrap = new Bootstrap(location); + RocksDBSession session = bootstrap.handler.getSession(PARTITION_ID)) { + assertTrue(Files.isRegularFile(Path.of(session.getDbPath(), "CURRENT"))); + assertTrue(session.getDB().isOwningHandle()); + } + } + + private static void writeBootstrapMetadata(Path databasePath, long graphId) throws Exception { + List handles = new ArrayList<>(); + try (DBOptions options = new DBOptions(); ColumnFamilyOptions cfOptions = new ColumnFamilyOptions()) { + List descriptors = new ArrayList<>(); + descriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOptions)); + for (String table : TABLES) { + descriptors.add(new ColumnFamilyDescriptor(table.getBytes(java.nio.charset.StandardCharsets.UTF_8), + cfOptions)); + } + try (RocksDB database = RocksDB.open(options, databasePath.toString(), descriptors, handles)) { + database.put(handles.get(0), MetadataKeyHelper.getGraphIDKey("test"), + Int64Value.of(graphId).toByteArray()); + database.put(handles.get(0), MetadataKeyHelper.getPartitionKey("test", PARTITION_ID), + Metapb.Partition.newBuilder().setId(PARTITION_ID).setGraphName("test") + .build().toByteArray()); + } finally { + handles.forEach(ColumnFamilyHandle::close); + } + } + } + + @Test + public void testBootstrapChecksActualRootAfterLocationUpdate() throws Exception { + Path location = this.temporary.newFolder("bootstrap-relocated").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + Files.createDirectory(data.resolve(DB_NAME + "_2")); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location, true)) { + assertThrows(org.apache.hugegraph.rocksdb.access.DBStoreException.class, + bootstrap.manager::loadPartition); + // Location discovery precedes native validation; it is not a startup transaction. + assertEquals(location.toString(), bootstrap.manager.getStoreMetadata() + .getPartitionStore(PARTITION_ID) + .getStoreLocation()); + org.mockito.Mockito.verify(bootstrap.manager.getStoreMetadata()) + .savePartitionStore(org.mockito.ArgumentMatchers.any()); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + private static class Bootstrap implements AutoCloseable { + + private final PartitionManager manager; + private final BusinessHandlerImpl handler; + private final Object previousHandler; + private final Field handlerField; + + private Bootstrap(Path location) throws Exception { + this(location, false); + } + + private Bootstrap(Path location, boolean relocate) throws Exception { + Files.createDirectories(location.resolve("raft/" + DB_NAME + "/log")); + HgStoreEngineOptions options = new HgStoreEngineOptions(); + options.setDataPath(location.toString()); + options.setRaftPath(location.toString()); + PdProvider provider = mock(PdProvider.class); + when(provider.getShardGroup(PARTITION_ID)).thenReturn( + Metapb.ShardGroup.newBuilder().setId(PARTITION_ID) + .addShards(Metapb.Shard.newBuilder().setStoreId(1L).build()).build()); + Metapb.Partition partition = Metapb.Partition.newBuilder().setId(PARTITION_ID) + .setGraphName("test").build(); + when(provider.getPartitionByID(anyString(), anyInt())).thenReturn( + new org.apache.hugegraph.store.meta.Partition(partition)); + when(provider.getPartitionByCode(anyString(), anyInt())).thenReturn(partition); + this.manager = new PartitionManager(provider, options); + StoreMetadata metadata = mock(StoreMetadata.class); + when(metadata.getStore()).thenReturn(new Store(Metapb.Store.newBuilder().setId(1L).build())); + AtomicReference saved = new AtomicReference<>( + Metapb.PartitionStore.newBuilder().setPartitionId(PARTITION_ID) + .setStoreLocation(relocate ? location.resolve("previous").toString() : + location.toString()).build()); + when(metadata.getPartitionStore(PARTITION_ID)).thenAnswer(call -> saved.get()); + org.mockito.Mockito.doAnswer(call -> { + saved.set(call.getArgument(0)); + return null; + }).when(metadata).savePartitionStore(org.mockito.ArgumentMatchers.any()); + when(metadata.getPartitionRaft(PARTITION_ID)).thenReturn( + Metapb.PartitionRaft.newBuilder().setPartitionId(PARTITION_ID) + .setRaftLocation(location.toString()).build()); + when(metadata.getPartitionStoreLocation(PARTITION_ID, DB_NAME)) + .thenAnswer(call -> saved.get().getStoreLocation()); + Field metadataField = PartitionManager.class.getDeclaredField("storeMetadata"); + metadataField.setAccessible(true); + metadataField.set(this.manager, metadata); + this.handler = new BusinessHandlerImpl(this.manager); + this.handlerField = HgStoreEngine.class.getDeclaredField("businessHandler"); + this.handlerField.setAccessible(true); + this.previousHandler = this.handlerField.get(HgStoreEngine.getInstance()); + this.handlerField.set(HgStoreEngine.getInstance(), this.handler); + } + + @Override + public void close() throws Exception { + try { + RocksDBFactory.getInstance().releaseGraphDB(DB_NAME); + } finally { + this.handlerField.set(HgStoreEngine.getInstance(), this.previousHandler); + } + } + } + @Test public void testLegacyDataRejectsBeforeRaftAndPreservesVersions() throws Exception { Path location = this.temporary.newFolder("existing-location").toPath(); From 70d03ebd15705996a7cebf0c0b54b958b1af4ba9 Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 05:26:40 +0800 Subject: [PATCH 3/5] fix(store): honor configured cf options - reuse effective options for read-only checks - keep native causes and close failed-open handles - cover level-seven snapshots and corruption errors --- docs/store-graphid-upgrade.md | 3 +- .../hugegraph/store/meta/GraphIdManager.java | 12 +- hugegraph-store/hg-store-rocksdb/README.md | 3 +- .../rocksdb/access/RocksDBSession.java | 31 +++- .../store/GraphIdUpgradePreflightTest.java | 168 ++++++++++++++++++ 5 files changed, 202 insertions(+), 15 deletions(-) diff --git a/docs/store-graphid-upgrade.md b/docs/store-graphid-upgrade.md index b9cc849402..a77798deca 100644 --- a/docs/store-graphid-upgrade.md +++ b/docs/store-graphid-upgrade.md @@ -11,7 +11,8 @@ graph ID, first opening a partition DB during metadata bootstrap, starting a partition's Raft state machine, or activating a snapshot. If it finds the reserved prefix, it rejects that operation and reports the partition and column family. Metadata-only column families are excluded. The -checks do not assign an owner to legacy rows or change the persisted ID format. +checks use the configured column-family options from normal Store opens. They do +not assign an owner to legacy rows or change the persisted ID format. The partition preflight resolves existing DB versions without creating a DB, allocating a store location or pruning older versions. It runs before partition diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java index ea463aefc9..95fd438b0e 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java @@ -31,6 +31,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.rocksdb.access.SessionOperator; import org.apache.hugegraph.store.constant.HugeServerTables; @@ -285,8 +287,7 @@ public static void checkLegacyGraphIdData(RocksDBSession session, int partitionI try { checkLegacyGraphIdData(session.getDB(), session.getTables(), partitionId); } catch (RocksDBException e) { - throw new HgStoreException(HgStoreException.EC_FAIL, - "Can't inspect legacy GraphID data in partition " + partitionId, e); + throw new HgStoreException("Can't inspect legacy GraphID data in partition " + partitionId, e); } finally { lock.unlock(); } @@ -303,6 +304,10 @@ protected void log(InfoLogLevel level, String message) { } }) { dbOptions.setLogger(logger); + HugeConfig config = RocksDBFactory.getInstance().getHugeConfig(); + if (config != null) { + RocksDBSession.initOptions(config, null, null, cfOptions, cfOptions); + } List names = RocksDB.listColumnFamilies(options, existingDbPath); List descriptors = new ArrayList<>(); for (byte[] name : names) { @@ -326,8 +331,7 @@ protected void log(InfoLogLevel level, String message) { } } } catch (RocksDBException e) { - throw new HgStoreException(HgStoreException.EC_FAIL, - "Can't inspect legacy GraphID data in partition " + partitionId + + throw new HgStoreException("Can't inspect legacy GraphID data in partition " + partitionId + " at " + existingDbPath, e); } } diff --git a/hugegraph-store/hg-store-rocksdb/README.md b/hugegraph-store/hg-store-rocksdb/README.md index 3946f6bd2d..86c034daae 100644 --- a/hugegraph-store/hg-store-rocksdb/README.md +++ b/hugegraph-store/hg-store-rocksdb/README.md @@ -35,7 +35,8 @@ failure leaves these DB paths intact. The original overloads keep their behavior this generic hook does not depend on Store graph-ID rules. `loadSnapshot(snapshotPath, version, Consumer validator)` first copies -and verifies the staging database, then passes its path to the validator before +and verifies the staging database with the session's configured column-family +options, then passes its path to the validator before replacing the target version. The validator must inspect it read-only. A rejected validation preserves active databases, target versions and staging evidence; the exception cause is retained. The original two-argument overload remains diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java index 18c6a853b4..867b4ef17c 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java @@ -761,6 +761,7 @@ private boolean verifySnapshot(String snapshotPath) { try { try (final Options options = new Options(); final ColumnFamilyOptions cfOptions = new ColumnFamilyOptions()) { + RocksDBSession.initOptions(this.hugeConfig, null, null, cfOptions, cfOptions); List columnFamilyDescriptorList = new ArrayList<>(); List columnFamilyBytes = RocksDB.listColumnFamilies(options, snapshotPath); @@ -775,15 +776,27 @@ private boolean verifySnapshot(String snapshotPath) { } List columnFamilyHandleList = new ArrayList<>(); - try (final DBOptions dbOptions = new DBOptions(); - final RocksDB db = RocksDB.openReadOnly(dbOptions, snapshotPath, - columnFamilyDescriptorList, - columnFamilyHandleList)) { - for (ColumnFamilyHandle handle : columnFamilyHandleList) { - if (handle == null) { - log.error("verifySnapshot some ColumnFamilyHandle is null in {}", - snapshotPath); - return false; + try (final DBOptions dbOptions = new DBOptions()) { + RocksDB db = null; + try { + db = RocksDB.openReadOnly(dbOptions, snapshotPath, + columnFamilyDescriptorList, + columnFamilyHandleList); + for (ColumnFamilyHandle handle : columnFamilyHandleList) { + if (handle == null) { + log.error("verifySnapshot some ColumnFamilyHandle is null in {}", + snapshotPath); + return false; + } + } + } finally { + for (ColumnFamilyHandle handle : columnFamilyHandleList) { + if (handle != null) { + handle.close(); + } + } + if (db != null) { + db.close(); } } } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java index 13c3880e6b..2c3430b11c 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java @@ -46,6 +46,7 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Stream; +import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.rocksdb.access.RocksDBFactory; import org.apache.hugegraph.rocksdb.access.SessionOperator; @@ -71,11 +72,16 @@ import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.rocksdb.Checkpoint; +import org.rocksdb.CompactRangeOptions; +import org.rocksdb.CompressionType; +import org.rocksdb.FlushOptions; import org.rocksdb.ColumnFamilyDescriptor; import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.ColumnFamilyOptions; import org.rocksdb.DBOptions; +import org.rocksdb.Options; import org.rocksdb.RocksDB; +import org.rocksdb.RocksDBException; import com.google.protobuf.Int64Value; @@ -109,6 +115,168 @@ public static void initRocksdb() { BusinessHandlerImpl.initRocksdb(config, null); } + @Test + public void testNativePreflightFailurePreservesCauseAndFiles() throws Exception { + Path data = this.temporary.newFolder("native-preflight-failure").toPath(); + writeDatabase(data, DB_NAME, 1); + Path database = data.resolve(DB_NAME); + Files.writeString(database.resolve("CURRENT"), "invalid-current"); + Map before = fileHashes(data); + RocksDBException nativeFailure; + try (Options options = new Options()) { + nativeFailure = assertThrows(RocksDBException.class, () -> { + try (RocksDB ignored = RocksDB.openReadOnly(options, database.toString())) { + // Close any unexpectedly successful open before reporting failure. + } + }); + } + HgStoreException rejected = assertThrows(HgStoreException.class, () -> + GraphIdManager.checkLegacyGraphIdData(database.toString(), PARTITION_ID)); + assertEquals(HgStoreException.EC_FAIL, rejected.getCode()); + assertEquals(before, fileHashes(data)); + assertTrue("Expected native cause " + nativeFailure.getStatus().getCode() + + ", got " + rejected.getCause(), + rejected.getCause() instanceof RocksDBException); + assertEquals(nativeFailure.getStatus().getCode(), + ((RocksDBException) rejected.getCause()).getStatus().getCode()); + } + + @Test + public void testEffectiveOptionsPreserveLevelSevenBootstrap() throws Exception { + RocksDBFactory factory = RocksDBFactory.getInstance(); + HugeConfig previous = factory.getHugeConfig(); + try { + useEightLevelConfiguration(); + Path location = this.temporary.newFolder("eight-level-bootstrap").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeEightLevelDatabase(data.resolve(DB_NAME + "_1"), 1); + // The normal session can read this legal configuration before the preflight. + try (RocksDBSession normal = session(data, DB_NAME)) { + assertRows(normal, 1); + } + Map before = fileHashes(data); + GraphIdManager.checkLegacyGraphIdData(data.resolve(DB_NAME + "_1").toString(), PARTITION_ID); + assertEquals(before, fileHashes(data)); + try (Bootstrap bootstrap = new Bootstrap(location)) { + bootstrap.manager.loadPartition(); + assertTrue(bootstrap.manager.hasPartition("test", PARTITION_ID)); + assertEquals(1, bootstrap.handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + for (String table : TABLES) { + assertArrayEquals(VALUE, bootstrap.handler.doGet("test", 0, table, new byte[]{1})); + } + GraphIdManager ids = new GraphIdManager(bootstrap.handler::getSession, PARTITION_ID); + assertTrue(ids.getGraphIdOrCreate("additional") < 65534L); + } + } finally { + factory.setHugeConfig(previous); + } + } + + @Test + public void testEffectiveOptionsRejectLegacyLevelSevenData() throws Exception { + RocksDBFactory factory = RocksDBFactory.getInstance(); + HugeConfig previous = factory.getHugeConfig(); + try { + useEightLevelConfiguration(); + Path location = this.temporary.newFolder("eight-level-legacy").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeEightLevelDatabase(data.resolve(DB_NAME + "_1"), 65534); + Map before = fileHashes(data); + HgStoreException rejected = assertThrows(HgStoreException.class, () -> + GraphIdManager.checkLegacyGraphIdData(data.resolve(DB_NAME + "_1").toString(), PARTITION_ID)); + assertTrue(rejected.getMessage().contains("prefix 0xFFFE")); + assertEquals(before, fileHashes(data)); + try (Bootstrap bootstrap = new Bootstrap(location)) { + assertThrows(HgStoreException.class, bootstrap.manager::loadPartition); + assertEquals(before, fileHashes(data)); + assertFalse(factory.getGraphNames().contains(DB_NAME)); + } + } finally { + factory.setHugeConfig(previous); + } + } + + @Test + public void testEffectiveOptionsAcceptLevelSevenSnapshot() throws Exception { + RocksDBFactory factory = RocksDBFactory.getInstance(); + HugeConfig previous = factory.getHugeConfig(); + try { + useEightLevelConfiguration(); + Path seed = this.temporary.newFolder("eight-level-seed").toPath(); + writeEightLevelDatabase(seed.resolve("seed"), 2); + Path snapshot = seed.resolve("snapshot"); + try (RocksDBSession source = session(seed, "seed"); + Checkpoint checkpoint = Checkpoint.create(source.getDB())) { + checkpoint.createCheckpoint(snapshot.toString()); + } + Map incoming = fileHashes(snapshot); + Path location = this.temporary.newFolder("eight-level-restore").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeEightLevelDatabase(data.resolve(DB_NAME + "_1"), 1); + try (Bootstrap bootstrap = new Bootstrap(location)) { + bootstrap.handler.loadSnapshot(snapshot.toString(), "", PARTITION_ID, 5L); + try (RocksDBSession restored = bootstrap.handler.getSession(PARTITION_ID)) { + assertRows(restored, 2); + assertEquals(2, bootstrap.handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + } + } + assertEquals(incoming, fileHashes(snapshot)); + } finally { + factory.setHugeConfig(previous); + } + } + + private static void useEightLevelConfiguration() { + Map config = new HashMap<>(); + config.put("rocksdb.write_buffer_size", "1048576"); + config.put("rocksdb.bloom_filter_bits_per_key", "10"); + config.put("rocksdb.num_levels", "8"); + config.put("rocksdb.compression_per_level", "[none,none,none,none,none,none,none,none]"); + BusinessHandlerImpl.initRocksdb(config, null); + } + + private static void writeEightLevelDatabase(Path path, int graphId) throws Exception { + List handles = new ArrayList<>(); + try (DBOptions options = new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions().setNumLevels(8) + .setCompressionPerLevel(Collections.nCopies(8, CompressionType.NO_COMPRESSION)) + .setDisableAutoCompactions(true).setWriteBufferSize(1048576); + FlushOptions flush = new FlushOptions().setWaitForFlush(true); + CompactRangeOptions compact = new CompactRangeOptions().setChangeLevel(true).setTargetLevel(7)) { + List descriptors = new ArrayList<>(); + descriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOptions)); + for (String table : TABLES) { + descriptors.add(new ColumnFamilyDescriptor(table.getBytes(java.nio.charset.StandardCharsets.UTF_8), + cfOptions)); + } + RocksDB database = null; + try { + database = RocksDB.open(options, path.toString(), descriptors, handles); + for (int i = 1; i < handles.size(); i++) { + database.put(handles.get(i), key(graphId), VALUE); + } + if (graphId != 65534) { + database.put(handles.get(0), MetadataKeyHelper.getGraphIDKey("test"), + Int64Value.of(graphId).toByteArray()); + database.put(handles.get(0), MetadataKeyHelper.getPartitionKey("test", PARTITION_ID), + Metapb.Partition.newBuilder().setId(PARTITION_ID).setGraphName("test") + .build().toByteArray()); + } + database.flush(flush, handles); + for (int i = 1; i < handles.size(); i++) { + database.compactRange(handles.get(i), null, null, compact); + assertFalse("A real L7 SST is required for this native compatibility regression", + database.getColumnFamilyMetaData(handles.get(i)).levels().get(7).files().isEmpty()); + } + } finally { + handles.forEach(ColumnFamilyHandle::close); + if (database != null) { + database.close(); + } + } + } + } + @Test public void testBootstrapPreservesOlderDatabaseWhenLatestHasNoCurrent() throws Exception { Path location = this.temporary.newFolder("bootstrap").toPath(); From a3cf14679c3294ebc6a8a9ac275217b81643921e Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 14 Oct 2026 05:53:03 +0800 Subject: [PATCH 4/5] fix(store): reuse cached preflight session Inspect the cached primary view before partition startup. Detect legacy rows written with WAL disabled before flush. Close the borrowed session without releasing its native owner. Preserve read-only path checks when the cache is empty. --- .../hugegraph/store/PartitionEngine.java | 10 +++- .../store/GraphIdUpgradePreflightTest.java | 55 +++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java index c8d6e58c80..588f0aebb1 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java @@ -45,6 +45,7 @@ import org.apache.hugegraph.pd.grpc.MetaTask; import org.apache.hugegraph.pd.grpc.Metapb; import org.apache.hugegraph.pd.raft.RaftReflectionUtil; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.business.BusinessHandlerImpl; @@ -272,12 +273,19 @@ public LogStorage createLogStorage(final String uri, final RaftOptions raftOptio } private void checkLegacyGraphIdData(int partitionId) { + String dbName = BusinessHandlerImpl.getDbName(partitionId); + try (RocksDBSession cached = RocksDBFactory.getInstance().queryGraphDB(dbName)) { + if (cached != null) { + // Use the primary's view, including unflushed writes made with WAL disabled. + GraphIdManager.checkLegacyGraphIdData(cached, partitionId); + return; + } + } Metapb.PartitionStore location = this.partitionManager.getStoreMetadata().getPartitionStore(partitionId); // Looking up or assigning a new store location would write metadata before inspection. String[] roots = location != null ? new String[]{location.getStoreLocation()} : this.storeEngine.getOption().getDataPath().split(","); - String dbName = BusinessHandlerImpl.getDbName(partitionId); for (String root : roots) { String dataPath = Paths.get(root, HgStoreEngineOptions.DB_Path_Prefix).toString(); String existing = RocksDBSession.resolveExistingDBPath( diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java index 2c3430b11c..2cbc2b286b 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java @@ -378,6 +378,61 @@ public void testFactoryValidatesColdOpenAndReusesCachedSession() throws Exceptio } } + @Test + public void testCachedMemtableLegacyDataRejectsBeforeRaft() throws Exception { + Path location = this.temporary.newFolder("cached-legacy-location").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + RocksDBFactory factory = RocksDBFactory.getInstance(); + try (RocksDBSession cached = factory.createGraphDB(data.toString(), DB_NAME, 0L)) { + cached.setDisableWAL(true); + putRows(cached, 65534); + // A separate RO instance cannot see WAL-disabled writes still in this primary's memtable. + GraphIdManager.checkLegacyGraphIdData(cached.getDbPath(), PARTITION_ID); + assertThrows(HgStoreException.class, () -> GraphIdManager.checkLegacyGraphIdData(cached, PARTITION_ID)); + int references = cached.getRefCount(); + PartitionEngine engine = partitionEngine(location, true); + PartitionEngineOptions options = new PartitionEngineOptions(); + options.setGroupId(PARTITION_ID); + + HgStoreException failure = assertThrows(HgStoreException.class, () -> engine.init(options)); + + assertTrue(failure.getMessage().contains("GraphID")); + assertNull(engine.getRaftNode()); + assertNull(field(engine, "stateMachine")); + assertNull(engine.getTaskManager()); + assertEquals(references, cached.getRefCount()); + assertTrue(cached.getDB().isOwningHandle()); + assertRows(cached, 65534); + } finally { + factory.releaseGraphDB(DB_NAME); + } + } + + @Test + public void testCachedValidSessionPreflightAvoidsPathResolution() throws Exception { + Path location = this.temporary.newFolder("cached-valid-location").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + RocksDBFactory factory = RocksDBFactory.getInstance(); + try (RocksDBSession cached = factory.createGraphDB(data.toString(), DB_NAME, 0L)) { + cached.setDisableWAL(true); + putRows(cached, 1); + int references = cached.getRefCount(); + PartitionEngine engine = partitionEngine(location, true); + // Cached native handles already identify the active DB; do not resolve another disk view. + when(engine.getStoreEngine().getPartitionManager().getStoreMetadata().getPartitionStore(PARTITION_ID)) + .thenThrow(new AssertionError("Cached preflight must not resolve a DB path")); + var check = PartitionEngine.class.getDeclaredMethod("checkLegacyGraphIdData", int.class); + check.setAccessible(true); + check.invoke(engine, PARTITION_ID); + + assertEquals(references, cached.getRefCount()); + assertTrue(cached.getDB().isOwningHandle()); + assertRows(cached, 1); + } finally { + factory.releaseGraphDB(DB_NAME); + } + } + @Test public void testNewPartitionDatabaseStillCreatesNormally() throws Exception { Path location = this.temporary.newFolder("bootstrap-new").toPath(); From d9ff79fded22b14229a57b3146f9acb02bbfe1a3 Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 14 Oct 2026 06:33:12 +0800 Subject: [PATCH 5/5] chore(store): remove preflight debug prints Remove stdout diagnostics from bootstrap regressions. Retain existing failure messages and data-preservation assertions. Keep production preflight behavior unchanged. --- .../store/core/store/GraphIdUpgradePreflightTest.java | 6 ------ 1 file changed, 6 deletions(-) diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java index 2cbc2b286b..91bd158b28 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java @@ -292,9 +292,6 @@ public void testBootstrapPreservesOlderDatabaseWhenLatestHasNoCurrent() throws E } catch (Throwable caught) { failure = caught; } - System.out.println("BOOTSTRAP older-present=" + Files.exists(data.resolve(DB_NAME + "_1")) + - " latest-CURRENT=" + Files.exists(data.resolve(DB_NAME + "_2/CURRENT")) + - " unchanged=" + before.equals(fileHashes(data)) + " failure=" + failure); assertNotNull("Bootstrap must reject before creating/pruning native databases", failure); assertTrue(failure.getMessage().contains("no CURRENT")); assertEquals(before, fileHashes(data)); @@ -316,9 +313,6 @@ public void testBootstrapPreservesLegacyOlderVersionBeforePruning() throws Excep } catch (Throwable caught) { failure = caught; } - System.out.println("BOOTSTRAP older-legacy-present=" + - Files.exists(data.resolve(DB_NAME + "_1")) + - " unchanged=" + before.equals(fileHashes(data)) + " failure=" + failure); assertNotNull("Pruning must not erase an older legacy database", failure); assertTrue(failure.getMessage().contains("GraphID")); assertEquals(before, fileHashes(data));