fix(store): throw on compaction-busy snapshot save; validate data/ on load (#3162)

.github/workflows/pd-store-ci.yml

@@ -295,6 +295,11 @@ jobs:
           mvn test -pl hugegraph-store/hg-store-test -am \
             -P store-raftcore-test -Djacoco.sessionId=store-raftcore-test
 
+      - name: Run core test
+        run: |
+          mvn test -pl hugegraph-store/hg-store-test -am \
+            -P store-core-test -Djacoco.sessionId=store-core-test
+
       - name: Generate aggregate coverage report
         run: |
           mvn verify -pl hugegraph-store/hg-store-test -am -P jacoco \
@@ -311,15 +316,19 @@ jobs:
             "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.rocksdb.RocksDbSuiteTest.xml" \
             --require-test-report \
             "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.raftcore.RaftSuiteTest.xml" \
+            --require-test-report \
+            "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.core.CoreSuiteTest.xml" \
             --require-covered-group hg-store-common \
             --require-covered-group hg-store-client \
             --require-covered-group hg-store-rocksdb \
+            --require-covered-group hg-store-core \
             --require-session store-common-test \
             --require-session store-client-test \
             --require-session store-rocksdb-test \
             --require-session store-raftcore-test \
+            --require-session store-core-test \
             "$REPORT_FILE" \
-            hg-store-grpc hg-store-common hg-store-client hg-store-rocksdb
+            hg-store-grpc hg-store-common hg-store-client hg-store-rocksdb hg-store-core
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh

@@ -499,21 +499,22 @@ assert "mvn verify -pl hugegraph-store/hg-store-test -am -P jacoco \\ " \
        "-DskipTests -Deditorconfig.skip=true -ntp" in " ".join(store_job.split())
 assert selected_profiles(store_job, "store") == {
     "store-common-test", "store-client-test", "store-rocksdb-test",
-    "store-raftcore-test",
+    "store-raftcore-test", "store-core-test",
 }
 assert reports_for_option(store_job, "--require-test-report") == {
     "TEST-org.apache.hugegraph.store.common.CommonSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.client.ClientSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.rocksdb.RocksDbSuiteTest.xml",
     "TEST-org.apache.hugegraph.store.raftcore.RaftSuiteTest.xml",
+    "TEST-org.apache.hugegraph.store.core.CoreSuiteTest.xml",
 }
 assert not reports_for_option(store_job, "--require-suite-report")
 assert values_for_option(store_job, "--require-covered-group") == {
-    "hg-store-common", "hg-store-client", "hg-store-rocksdb",
+    "hg-store-common", "hg-store-client", "hg-store-rocksdb", "hg-store-core",
 }
 assert required_modules(store_job) == {
     "hg-store-grpc", "hg-store-common", "hg-store-client",
-    "hg-store-rocksdb",
+    "hg-store-rocksdb", "hg-store-core",
 }
 
 print("PASS: JaCoCo aggregation configuration contract")

hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java

@@ -230,6 +230,14 @@ void lock(String path) throws InterruptedException,
 
     void unlock(String path);
 
+    /**
+     * Non-blocking attempt to reserve the compactRange() window for partition {@code id}.
+     * Returns false if a compaction is actively running for that partition right now.
+     */
+    boolean tryLockCompactionRange(int id);
+
+    void unlockCompactionRange(int id);
+
     void awaitAndSetLock(int id, int expectedValue, int value) throws InterruptedException,
                                                                       TimeoutException;
 

hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java

@@ -36,10 +36,12 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -50,6 +52,9 @@
 
 import javax.annotation.concurrent.NotThreadSafe;
 
+import lombok.Getter;
+import lombok.Setter;
+
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang.ArrayUtils;
 import org.apache.commons.lang.StringUtils;
@@ -138,6 +143,12 @@ public class BusinessHandlerImpl implements BusinessHandler {
     private static final ConcurrentMap<String, AtomicInteger> pathLock = new ConcurrentHashMap<>();
     private static final ConcurrentMap<Integer, AtomicInteger> compactionState =
             new ConcurrentHashMap<>();
+    // Guards the compactRange() window specifically, so a snapshot save can atomically
+    // check-and-reserve against a compaction that is actually running right now. This is
+    // narrower than pathLock, which stays held through the post-compaction blank-task
+    // snapshot and must not be reused here to avoid deadlocking that flow.
+    private static final ConcurrentMap<Integer, ReentrantLock> compactionRangeLock =
+            new ConcurrentHashMap<>();
     // Default core thread count
     private static final int compactionThreadCount = 64;
     private static final int compactionMaxThreadCount = 256;
@@ -154,6 +165,18 @@ public class BusinessHandlerImpl implements BusinessHandler {
     private final InnerKeyCreator keyCreator;
     private final Semaphore semaphore = new Semaphore(1);
 
+    /* Bounds how long dbCompaction() waits to acquire compactionRangeLock when a snapshot
+     save is holding it. saveSnapshot() is a RocksDB Checkpoint (hard-links existing SST
+     files, no data copy) plus a partial checksum read, so the lock is normally held for
+     well under a second, at most a few seconds under a slow/busy disk. 10s gives generous
+     margin over that expected hold time while keeping a stuck snapshot save from blocking
+     compaction for long: if the wait is exceeded, dbCompaction just skips this pass and
+     relies on the next trigger (PD instruction, REST call, etc.) to retry - see the
+     tryLock() call below.
+     Not final so tests can shorten it via setCompactionRangeLockWaitMillis() rather than
+     waiting out the real production value.*/
+    @Setter @Getter
+    private static long compactionRangeLockWaitMillis = 10_000;
     public BusinessHandlerImpl(PartitionManager partitionManager) {
         this.partitionManager = partitionManager;
         this.provider = partitionManager.getPdProvider();
@@ -1415,10 +1438,29 @@ public boolean dbCompaction(String graphName, int id, String tableName) {
                         log.info("Partition {} dbCompaction started", id);
                         if (tableName.isEmpty()) {
                             lock(path);
-                            setState(id, doing);
-                            log.info("Partition {}-{} got lock, dbCompaction start", id, path);
-                            op.compactRange();
-                            setState(id, compactionDone);
+                            ReentrantLock rangeLock =
+                                    compactionRangeLock.computeIfAbsent(id,
+                                                                        k -> new ReentrantLock());
+                            if (!rangeLock.tryLock(compactionRangeLockWaitMillis,
+                                                   TimeUnit.MILLISECONDS)) {
+                                // A snapshot save is still reserving this partition's range lock
+                                // after the wait. Skip this compaction pass rather than block -
+                                // callers of dbCompaction(). This is a transient condition, and the next
+                                // compaction pass will succeed.
+                                log.warn("Partition {} skip dbCompaction, snapshot save " +
+                                         "still in progress after {}ms wait", id,
+                                         compactionRangeLockWaitMillis);
+                                unlock(path);
+                                return;
+                            }
+                            try {
+                                setState(id, doing);
+                                log.info("Partition {}-{} got lock, dbCompaction start", id, path);
+                                op.compactRange();
+                                setState(id, compactionDone);
+                            } finally {
+                                rangeLock.unlock();
+                            }
                             log.info("Partition {} dbCompaction end and start to do snapshot", id);
                             PartitionEngine pe = HgStoreEngine.getInstance().getPartitionEngine(id);
                             // find leader and send blankTask, after execution
@@ -1484,6 +1526,20 @@ private boolean compareAndSetLock(String path) {
         return l.compareAndSet(compactionCanStart, doing);
     }
 
+    @Override
+    public boolean tryLockCompactionRange(int id) {
+        ReentrantLock rangeLock = compactionRangeLock.computeIfAbsent(id, k -> new ReentrantLock());
+        return rangeLock.tryLock();
+    }
+
+    @Override
+    public void unlockCompactionRange(int id) {
+        ReentrantLock rangeLock = compactionRangeLock.get(id);
+        if (rangeLock != null) {
+            rangeLock.unlock();
+        }
+    }
+
     @Override
     public void awaitAndSetLock(int id, int expectedValue, int value) throws InterruptedException,
                                                                              TimeoutException {

hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java

@@ -44,6 +44,7 @@
 public class RaftRocksdbOptions {
 
     private static RocksdbConfig rocksdbConfig = null;
+    private static boolean raftRocksdbConfigRegistered = false;
 
     private static RocksdbConfig getRocksdbConfig(HugeConfig options) {
         if (rocksdbConfig == null) {
@@ -55,42 +56,55 @@ private static RocksdbConfig getRocksdbConfig(HugeConfig options) {
     }
 
     private static void registerRaftRocksdbConfig(HugeConfig options) {
-        Cache blockCache = new LRUCache(SizeUnit.GB);
-        BlockBasedTableConfig tableConfig = new BlockBasedTableConfig()
-                .setIndexType(IndexType.kTwoLevelIndexSearch)
-                .setPartitionFilters(true) //
-                .setMetadataBlockSize(8 * SizeUnit.KB) //
-                .setCacheIndexAndFilterBlocks(
-                        options.get(RocksDBOptions.PUT_FILTER_AND_INDEX_IN_CACHE))
-                .setCacheIndexAndFilterBlocksWithHighPriority(true)
-                .setPinL0FilterAndIndexBlocksInCache(
-                        options.get(RocksDBOptions.PIN_L0_FILTER_AND_INDEX_IN_CACHE))
-                .setBlockSize(4 * SizeUnit.KB)
-                .setBlockCache(blockCache);
-
-        StorageOptionsFactory.registerRocksDBTableFormatConfig(RocksDBLogStorage.class,
-                                                               tableConfig);
-
-        DBOptions dbOptions = StorageOptionsFactory.getDefaultRocksDBOptions();
-        dbOptions.setEnv(rocksdbConfig.getEnv());
-
-        // raft rocksdb number is fixed, can be controlled by max_write_buffer_number
-        //dbOptions.setWriteBufferManager(rocksdbConfig.getBufferManager());
-        dbOptions.setUnorderedWrite(true);
-        StorageOptionsFactory.registerRocksDBOptions(RocksDBLogStorage.class,
-                                                     dbOptions);
-
-        ColumnFamilyOptions cfOptions =
-                StorageOptionsFactory.getDefaultRocksDBColumnFamilyOptions();
-        cfOptions.setTargetFileSizeBase(256 * SizeUnit.MB);
-        cfOptions.setWriteBufferSize(8 * SizeUnit.MB);
-        cfOptions.setNumLevels(3);
-        cfOptions.setMaxWriteBufferNumber(3);
-        cfOptions.setCompressionType(CompressionType.NO_COMPRESSION);
-        cfOptions.setMaxBytesForLevelBase(2048 * SizeUnit.GB);
-
-        StorageOptionsFactory.registerRocksDBColumnFamilyOptions(RocksDBLogStorage.class,
-                                                                 cfOptions);
+        // StorageOptionsFactory.releaseAllOptions() (called by test setup between runs)
+        // does not clear its table-format-config table, so registering RocksDBLogStorage's
+        // config more than once per JVM throws IllegalStateException. Register only once.
+        // The guard flag is held across the whole registration so a failure partway through
+        // doesn't leave the flag set to true while some options were never registered.
+        synchronized (RaftRocksdbOptions.class) {
+            if (raftRocksdbConfigRegistered) {
+                return;
+            }
+
+            Cache blockCache = new LRUCache(SizeUnit.GB);
+            BlockBasedTableConfig tableConfig = new BlockBasedTableConfig()
+                    .setIndexType(IndexType.kTwoLevelIndexSearch)
+                    .setPartitionFilters(true) //
+                    .setMetadataBlockSize(8 * SizeUnit.KB) //
+                    .setCacheIndexAndFilterBlocks(
+                            options.get(RocksDBOptions.PUT_FILTER_AND_INDEX_IN_CACHE))
+                    .setCacheIndexAndFilterBlocksWithHighPriority(true)
+                    .setPinL0FilterAndIndexBlocksInCache(
+                            options.get(RocksDBOptions.PIN_L0_FILTER_AND_INDEX_IN_CACHE))
+                    .setBlockSize(4 * SizeUnit.KB)
+                    .setBlockCache(blockCache);
+
+            StorageOptionsFactory.registerRocksDBTableFormatConfig(RocksDBLogStorage.class,
+                                                                   tableConfig);
+
+            DBOptions dbOptions = StorageOptionsFactory.getDefaultRocksDBOptions();
+            dbOptions.setEnv(rocksdbConfig.getEnv());
+
+            // raft rocksdb number is fixed, can be controlled by max_write_buffer_number
+            //dbOptions.setWriteBufferManager(rocksdbConfig.getBufferManager());
+            dbOptions.setUnorderedWrite(true);
+            StorageOptionsFactory.registerRocksDBOptions(RocksDBLogStorage.class,
+                                                         dbOptions);
+
+            ColumnFamilyOptions cfOptions =
+                    StorageOptionsFactory.getDefaultRocksDBColumnFamilyOptions();
+            cfOptions.setTargetFileSizeBase(256 * SizeUnit.MB);
+            cfOptions.setWriteBufferSize(8 * SizeUnit.MB);
+            cfOptions.setNumLevels(3);
+            cfOptions.setMaxWriteBufferNumber(3);
+            cfOptions.setCompressionType(CompressionType.NO_COMPRESSION);
+            cfOptions.setMaxBytesForLevelBase(2048 * SizeUnit.GB);
+
+            StorageOptionsFactory.registerRocksDBColumnFamilyOptions(RocksDBLogStorage.class,
+                                                                     cfOptions);
+
+            raftRocksdbConfigRegistered = true;
+        }
     }
 
     public static void initRocksdbGlobalConfig(Map<String, Object> config) {

hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java

@@ -198,7 +198,13 @@ public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
                 done.run(Status.OK());
             } catch (HgStoreException e) {
                 log.error(String.format("Raft %s onSnapshotSave failed. {}", groupId), e);
-                done.run(new Status(RaftError.EIO, e.toString()));
+                // A busy compaction-range lock is transient: jRaft's snapshot scheduler
+                // retries independently, so report EBUSY rather than EIO to avoid
+                // escalating to reportError()/restartRaftNode() (see SnapshotExecutorImpl
+                // #onSnapshotSaveDone, which only escalates on EIO).
+                RaftError raftError = e.getCode() == HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL ?
+                                       RaftError.EBUSY : RaftError.EIO;
+                done.run(new Status(raftError, e.toString()));
             } finally {
                 lock.unlock();
             }