fix(store): preflight legacy graph prefixes

docs/store-graphid-upgrade.md

@@ -0,0 +1,34 @@
+# 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, 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 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
+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
+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.

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,8 @@
 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;
 import org.apache.hugegraph.store.cmd.HgCmdClient;
@@ -52,6 +55,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 +179,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 +272,30 @@ public LogStorage createLogStorage(final String uri, final RaftOptions raftOptio
         return this.started;
     }
 
+    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(",");
+        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;
     }

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;
         }
     }
 
@@ -1490,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,

hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java

@@ -21,21 +21,35 @@
 
 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;
 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;
 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 +113,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 +277,90 @@ 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("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);
+            HugeConfig config = RocksDBFactory.getInstance().getHugeConfig();
+            if (config != null) {
+                RocksDBSession.initOptions(config, null, null, cfOptions, cfOptions);
+            }
+            List<byte[]> names = RocksDB.listColumnFamilies(options, existingDbPath);
+            List<ColumnFamilyDescriptor> descriptors = new ArrayList<>();
+            for (byte[] name : names) {
+                descriptors.add(new ColumnFamilyDescriptor(name, cfOptions));
+            }
+            List<ColumnFamilyHandle> handles = new ArrayList<>();
+            RocksDB db = null;
+            try {
+                db = RocksDB.openReadOnly(dbOptions, existingDbPath, descriptors, handles);
+                Map<String, ColumnFamilyHandle> 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("Can't inspect legacy GraphID data in partition " + partitionId +
+                                       " at " + existingDbPath, e);
+        }
+    }
+
+    private static void checkLegacyGraphIdData(RocksDB db,
+                                               Map<String, ColumnFamilyHandle> 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
      *

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);

hugegraph-store/hg-store-rocksdb/README.md

@@ -0,0 +1,46 @@
+<!--
+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.
+-->
+
+# 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.
+
+`RocksDBFactory.createGraphDB(path, name, version, Consumer<String> 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<String> validator)` first copies
+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
+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.

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,13 +186,22 @@ 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<String> existingDatabaseValidator) {
         if (closing.get()) {
             throw new RuntimeException("db closed");
         }
         operateLock.writeLock().lock();
         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);