diff --git a/.github/workflows/pd-store-ci.yml b/.github/workflows/pd-store-ci.yml index 1a6825e7e4..1ec0c66ae5 100644 --- a/.github/workflows/pd-store-ci.yml +++ b/.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 diff --git a/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh b/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh index bdb09ba162..133026e594 100755 --- a/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh +++ b/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") diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java index d69b36bd7c..0a45b05f8e 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java @@ -230,6 +230,22 @@ 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. + * Default throws, like {@link #scanOrdered}, so adding this compaction-lock helper does + * not break downstream implementations of this public interface that predate it. + */ + default boolean tryLockCompactionRange(int id) { + throw new UnsupportedOperationException( + "Compaction-range locking is not supported"); + } + + default void unlockCompactionRange(int id) { + throw new UnsupportedOperationException( + "Compaction-range locking is not supported"); + } + void awaitAndSetLock(int id, int expectedValue, int value) throws InterruptedException, TimeoutException; @@ -239,6 +255,18 @@ void awaitAndSetLock(int id, int expectedValue, int value) throws InterruptedExc String getLockPath(int partitionId); + /** + * The path lock state for {@code path} as set by {@link #lock} / {@link #unlock} + * ({@link #compactionCanStart} or {@link #doing}), or {@code null} if {@code path} has + * never been locked. Default throws, like {@link #scanOrdered}, so adding this + * compaction-lock helper does not break downstream implementations of this public + * interface that predate it. + */ + default AtomicInteger getPathLockState(String path) { + throw new UnsupportedOperationException( + "Compaction-range locking is not supported"); + } + List getPartitionIds(String graph); @NotThreadSafe 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 f9ee79252d..38bad3b9e8 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 @@ -34,12 +34,16 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; 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 +54,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; @@ -107,6 +114,7 @@ import org.apache.hugegraph.store.raft.RaftOperation; import org.apache.hugegraph.store.term.Bits; import org.apache.hugegraph.store.term.HgPair; +import org.apache.hugegraph.store.util.DefaultThreadFactory; import org.apache.hugegraph.store.util.ExecutorUtil; import org.apache.hugegraph.store.util.HgStoreException; import org.apache.hugegraph.structure.BaseElement; @@ -138,6 +146,12 @@ public class BusinessHandlerImpl implements BusinessHandler { private static final ConcurrentMap pathLock = new ConcurrentHashMap<>(); private static final ConcurrentMap 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 compactionRangeLock = + new ConcurrentHashMap<>(); // Default core thread count private static final int compactionThreadCount = 64; private static final int compactionMaxThreadCount = 256; @@ -154,6 +168,33 @@ 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; + /* Retry cadence for a cleanPartition() compactRange() pass that had to be skipped because + compactionRangeLock was still held by a snapshot save after compactionRangeLockWaitMillis. + Not final so tests can shorten it via setCleanPartitionCompactRetryDelayMillis() rather + than waiting out the real production value. */ + @Setter @Getter + private static long cleanPartitionCompactRetryDelayMillis = 30_000; + /* Bounds how many times a skipped cleanPartition() compactRange() pass is retried before + it is abandoned and logged, rather than dropped silently forever. Not final so tests can + shrink it via setMaxCleanPartitionCompactRetries() to exercise the exhausted-retries path + quickly. */ + @Setter @Getter + private static int maxCleanPartitionCompactRetries = 5; + private static final ScheduledExecutorService cleanPartitionCompactRetryScheduler = + new ScheduledThreadPoolExecutor(1, new DefaultThreadFactory(PoolNames.COMPACT_RETRY)); + public BusinessHandlerImpl(PartitionManager partitionManager) { this.partitionManager = partitionManager; this.provider = partitionManager.getPdProvider(); @@ -1175,7 +1216,6 @@ public boolean cleanPartition(String graph, int partId, long startKey, long endK if (partition == null) { return true; } - log.info("cleanPartition: graph {}, part id: {}, {} -> {}, cleanType:{}", graph, partId, startKey, endKey, cleanType); @@ -1190,6 +1230,13 @@ public boolean cleanPartition(String graph, int partId, long startKey, long endK taskManager.putAsyncTask(cleanTask); Utils.runInThread(() -> { + // The range lock only guards the actual compactRange() call inside + // cleanPartition(Partition, Function) below - not this whole delete pass, which + // scans every key in the partition and can take minutes. Holding the range lock + // for that long would fail every onSnapshotSave on this partition with EBUSY for + // the duration, and a slow snapshot save could then make this one-shot + // post-split/move cleanup time out and get dropped entirely instead of just its + // compaction step. cleanPartition(partition, code -> { // in range boolean flag = code >= startKey && code < endKey; @@ -1264,11 +1311,110 @@ private boolean cleanPartition(Partition partition, } op.getDBSession().close(); } - op.compactRange(); + int partId = partition.getId(); + ReentrantLock rangeLock = + compactionRangeLock.computeIfAbsent(partId, k -> new ReentrantLock()); + boolean rangeLocked = false; + boolean needsRetry = false; + try { + rangeLocked = rangeLock.tryLock(compactionRangeLockWaitMillis, TimeUnit.MILLISECONDS); + if (rangeLocked) { + op.compactRange(); + } else { + // The delete pass above already committed, so the cleanup itself succeeded - + // only compaction needs to be retried here. Log and move on rather than failing + // the whole cleanup over a busy snapshot save. A bounded retry is scheduled + // below so the compaction still eventually runs. + log.warn("Partition {}-{} skip cleanPartition compactRange, snapshot save " + + "still in progress after {}ms wait", partition.getGraphName(), partId, + compactionRangeLockWaitMillis); + needsRetry = true; + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("Partition {}-{} cleanPartition interrupted while waiting for snapshot " + + "range lock, will retry compactRange", partition.getGraphName(), partId); + needsRetry = true; + } finally { + if (rangeLocked) { + rangeLock.unlock(); + } + } + if (needsRetry) { + scheduleCleanPartitionCompactRetry(partition, 1); + } log.info("Partition {}-{} cleanPartition end", partition.getGraphName(), partition.getId()); return true; } + /** + * Retries a cleanPartition() compactRange() pass that needs to run again because + * compactionRangeLock was still held by a snapshot save. The original op's session was + * already closed by the time the range-lock section ran, so each attempt opens a fresh + * session rather than reusing it. Gives up and logs once maxCleanPartitionCompactRetries + * is exhausted, or if the partition has since been destroyed. + */ + private void scheduleCleanPartitionCompactRetry(Partition partition, int attempt) { + String graph = partition.getGraphName(); + int partId = partition.getId(); + cleanPartitionCompactRetryScheduler.schedule(() -> { + if (HgStoreEngine.getInstance().getPartitionEngine(partId) == null) { + log.warn("Partition {}-{} abandoning cleanPartition compactRange retry {}/{}, " + + "partition no longer exists", graph, partId, attempt, + maxCleanPartitionCompactRetries); + return; + } + ReentrantLock rangeLock = + compactionRangeLock.computeIfAbsent(partId, k -> new ReentrantLock()); + boolean rangeLocked = false; + boolean needsRetry = false; + try { + rangeLocked = rangeLock.tryLock(compactionRangeLockWaitMillis, + TimeUnit.MILLISECONDS); + if (rangeLocked) { + SessionOperator retryOp = getSession(graph, partId).sessionOp(); + try { + retryOp.compactRange(); + log.info("Partition {}-{} cleanPartition compactRange retry {}/{} " + + "succeeded", graph, partId, attempt, + maxCleanPartitionCompactRetries); + } finally { + retryOp.getDBSession().close(); + } + } else { + log.warn("Partition {}-{} cleanPartition compactRange retry {}/{} needs " + + "another retry, snapshot save still in progress after {}ms wait", + graph, partId, attempt, maxCleanPartitionCompactRetries, + compactionRangeLockWaitMillis); + needsRetry = true; + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("Partition {}-{} cleanPartition compactRange retry {}/{} interrupted " + + "while waiting for snapshot range lock", graph, partId, attempt, + maxCleanPartitionCompactRetries); + needsRetry = true; + } catch (Exception e) { + log.error("Partition {}-{} abandoning cleanPartition compactRange retry {}/{} " + + "after failure", graph, partId, attempt, maxCleanPartitionCompactRetries, + e); + } finally { + if (rangeLocked) { + rangeLock.unlock(); + } + } + if (needsRetry) { + if (attempt < maxCleanPartitionCompactRetries) { + scheduleCleanPartitionCompactRetry(partition, attempt + 1); + } else { + log.warn("Partition {}-{} abandoning cleanPartition compactRange after {} " + + "retries, snapshot save still busy", graph, partId, + maxCleanPartitionCompactRetries); + } + } + }, cleanPartitionCompactRetryDelayMillis, TimeUnit.MILLISECONDS); + } + @Override public boolean deletePartition(String graph, int partId) { try { @@ -1413,12 +1559,50 @@ public boolean dbCompaction(String graphName, int id, String tableName) { pathLock.putIfAbsent(path, new AtomicInteger(compactionCanStart)); compactionState.putIfAbsent(id, new AtomicInteger(0)); log.info("Partition {} dbCompaction started", id); + ReentrantLock rangeLock = + compactionRangeLock.computeIfAbsent(id, + k -> new ReentrantLock()); if (tableName.isEmpty()) { + // Take the path lock first: it can block for up to timeoutMillis + // (6h) waiting on an earlier compaction/snapshot cycle for this + // partition, so it must not be held while also reserving the + // range lock below - that would let a blocked dbCompaction pin + // the range lock and fail every onSnapshotSave for the partition + // until the wait ends. lock(path); - setState(id, doing); - log.info("Partition {}-{} got lock, dbCompaction start", id, path); - op.compactRange(); - setState(id, compactionDone); + boolean rangeLocked; + try { + rangeLocked = rangeLock.tryLock(compactionRangeLockWaitMillis, + TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("Partition {} dbCompaction interrupted while waiting " + + "for snapshot range lock", id); + // lock(path) above already succeeded, so the path lock is + // definitely held here and must be released, otherwise later + // compactions for this partition would block until the path + // lock timeout. + unlock(path); + return; + } + if (!rangeLocked) { + // A snapshot save is still reserving this partition's range lock + // after the wait. Skip this compaction pass rather than block - + // callers of dbCompaction(). + 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 @@ -1436,7 +1620,29 @@ public boolean dbCompaction(String graphName, int id, String tableName) { } setAndNotifyState(id, compactionDone); } else { - op.compactRange(tableName); + // No path lock in this branch: only guard the actual + // compactRange(tableName) call against a concurrent snapshot save. + boolean rangeLocked = false; + try { + rangeLocked = rangeLock.tryLock(compactionRangeLockWaitMillis, + TimeUnit.MILLISECONDS); + if (rangeLocked) { + op.compactRange(tableName); + } else { + log.warn("Partition {} skip dbCompaction({}), snapshot " + + "save still in progress after {}ms wait", id, + tableName, compactionRangeLockWaitMillis); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("Partition {} dbCompaction({}) interrupted while " + + "waiting for snapshot range lock", id, tableName); + return; + } finally { + if (rangeLocked) { + rangeLock.unlock(); + } + } } } log.info("Partition {}-{} dbCompaction end", id, path); @@ -1484,6 +1690,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 { @@ -1514,6 +1734,11 @@ public AtomicInteger getState(int id) { return l; } + @Override + public AtomicInteger getPathLockState(String path) { + return pathLock.get(path); + } + private AtomicInteger setState(int id, int state) { AtomicInteger l = compactionState.get(id); l.set(state); diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java index c272701308..33a662813c 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java @@ -29,6 +29,7 @@ public class PoolNames { public static final String I_JOB = "hg-i-job"; public static final String U_JOB = "hg-u-job"; public static final String COMPACT = "hg-compact"; + public static final String COMPACT_RETRY = "hg-compact-retry"; public static final String HEARTBEAT = "hg-heartbeat"; public static final String P_HEARTBEAT = "hg-p-heartbeat"; diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java index cb88814936..2386a1f227 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java +++ b/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 config) { 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 73821e4971..922b23b5a0 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 @@ -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(); } diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java index 3f26b8eedd..495a675393 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java @@ -24,7 +24,6 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.Checksum; import org.apache.commons.io.FileUtils; @@ -94,35 +93,41 @@ public void onSnapshotSave(final SnapshotWriter writer) throws HgStoreException final String snapshotDir = writer.getPath(); if (partitionEngine != null) { Integer groupId = partitionEngine.getGroupId(); - AtomicInteger state = businessHandler.getState(groupId); - if (state != null && state.get() == BusinessHandler.doing) { - return; + if (!businessHandler.tryLockCompactionRange(groupId)) { + throw new HgStoreException(HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL, + String.format( + "Partition %d snapshot save failed: " + + "compaction in progress", groupId)); } - // rocks db snapshot - final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; - businessHandler.saveSnapshot(graphSnapshotDir, "", groupId); - - List files = new ArrayList<>(); - File dir = new File(graphSnapshotDir); - File rootDirFile = new File(writer.getPath()); - // add all files in data dir - findFileList(dir, rootDirFile, files); - - // load snapshot by learner ?? - for (String file : files) { - String checksum = calculateChecksum(writer.getPath() + File.separator + file); - if (checksum.length() != 0) { - LocalFileMetaOutter.LocalFileMeta meta = - LocalFileMetaOutter.LocalFileMeta.newBuilder() - .setChecksum(checksum) - .build(); - writer.addFile(file, meta); - } else { - writer.addFile(file); + try { + // rocks db snapshot + final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; + businessHandler.saveSnapshot(graphSnapshotDir, "", groupId); + + List files = new ArrayList<>(); + File dir = new File(graphSnapshotDir); + File rootDirFile = new File(writer.getPath()); + // add all files in data dir + findFileList(dir, rootDirFile, files); + + // load snapshot by learner ?? + for (String file : files) { + String checksum = calculateChecksum(writer.getPath() + File.separator + file); + if (checksum.length() != 0) { + LocalFileMetaOutter.LocalFileMeta meta = + LocalFileMetaOutter.LocalFileMeta.newBuilder() + .setChecksum(checksum) + .build(); + writer.addFile(file, meta); + } else { + writer.addFile(file); + } } + // should_not_load wound not sync to learner + markShouldNotLoad(writer, true); + } finally { + businessHandler.unlockCompactionRange(groupId); } - // should_not_load wound not sync to learner - markShouldNotLoad(writer, true); } } @@ -169,6 +174,7 @@ private String calculateChecksum(String path) { public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) throws HgStoreException { final String snapshotDir = reader.getPath(); + final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; // No need to load locally saved snapshots if (shouldNotLoad(reader)) { @@ -176,8 +182,15 @@ public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) thr return; } + if (!new File(graphSnapshotDir).isDirectory()) { + throw new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, + String.format( + "Raft %d snapshot is corrupt, data dir %s is " + + "missing", partitionEngine.getGroupId(), + graphSnapshotDir)); + } + // Use snapshot directly - final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; log.info("Raft {} begin loadSnapshot, {}", partitionEngine.getGroupId(), graphSnapshotDir); businessHandler.loadSnapshot(graphSnapshotDir, "", partitionEngine.getGroupId(), committedIndex); diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java index 9284361395..1311fa127e 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java @@ -37,6 +37,7 @@ public class HgStoreException extends RuntimeException { public static final int EC_RKDB_EXPORT_SNAPSHOT_FAIL = 1214; public static final int EC_RKDB_IMPORT_SNAPSHOT_FAIL = 1215; public static final int EC_RKDB_TRANSFER_SNAPSHOT_FAIL = 1216; + public static final int EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL = 1217; public static final int EC_METRIC_FAIL = 1401; private static final long serialVersionUID = 5193624480997934335L; private final int code; diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index ed91e011e3..cb9433e903 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -242,6 +242,12 @@ ${basedir}/target/classes/ + + false **/CoreSuiteTest.java **/BatchGraphIsolationTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java index c222557962..3d78821a63 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java @@ -57,7 +57,6 @@ import org.junit.Test; import org.mockito.Mockito; -import com.alipay.sofa.jraft.util.StorageOptionsFactory; import com.google.protobuf.ByteString; public class BatchGraphIsolationTest { @@ -78,7 +77,6 @@ public static void setup() throws IOException { Map rocksdbConfig = new HashMap<>(); rocksdbConfig.put("rocksdb.write_buffer_size", "1048576"); - StorageOptionsFactory.releaseAllOptions(); RaftRocksdbOptions.initRocksdbGlobalConfig(rocksdbConfig); BusinessHandlerImpl.initRocksdb(rocksdbConfig, null); diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java index 68530367a0..6afd046e18 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java @@ -17,13 +17,14 @@ package org.apache.hugegraph.store.core; +import org.apache.hugegraph.store.core.snapshot.HgSnapshotHandlerTest; +import org.junit.runner.RunWith; +import org.junit.runners.Suite; + import lombok.extern.slf4j.Slf4j; -// TODO: uncomment it until all test can run free. -//@RunWith(Suite.class) -//@Suite.SuiteClasses({ +// TODO: uncomment the rest of these classes once they can run free of each other. // HgCmdClientTest.class, -// HgSnapshotHandlerTest.class, // RaftUtilsTest.class, // RaftOperationTest.class, // UnsafeUtilTest.class, @@ -41,8 +42,10 @@ // PartitionInstructionProcessorTest.class, // // Try to put it last // HgBusinessImplTest.class -//}) - +@RunWith(Suite.class) +@Suite.SuiteClasses({ + HgSnapshotHandlerTest.class +}) @Slf4j public class CoreSuiteTest { diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java index bce07dea5b..ee5cec32a0 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java @@ -28,13 +28,12 @@ import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.meta.ShardGroup; import org.apache.hugegraph.store.options.HgStoreEngineOptions; +import org.apache.hugegraph.store.options.JobOptions; import org.apache.hugegraph.store.options.RaftRocksdbOptions; import org.apache.hugegraph.store.pd.FakePdServiceProvider; import org.junit.AfterClass; import org.junit.BeforeClass; -import com.alipay.sofa.jraft.util.StorageOptionsFactory; - import lombok.extern.slf4j.Slf4j; /** @@ -61,6 +60,11 @@ public static void initEngine() { options.setGrpcAddress("127.0.0.1:6511"); options.setRaftAddress("127.0.0.1:6510"); options.setDataTransfer(new DataManagerImpl()); + JobOptions jobOptions = new JobOptions(); + jobOptions.setUninterruptibleCore(2); + jobOptions.setUninterruptibleMax(8); + jobOptions.setUninterruptibleQueueSize(1024); + options.setJobConfig(jobOptions); options.setFakePdOptions(new HgStoreEngineOptions.FakePdOptions() {{ setStoreList("127.0.0.1"); @@ -70,7 +74,6 @@ public static void initEngine() { }}); if (initCount == 0) { - StorageOptionsFactory.releaseAllOptions(); RaftRocksdbOptions.initRocksdbGlobalConfig(options.getRocksdbConfig()); initCount++; } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/raft/PartitionStateMachineTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/raft/PartitionStateMachineTest.java new file mode 100644 index 0000000000..3f25e63ae0 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/raft/PartitionStateMachineTest.java @@ -0,0 +1,158 @@ +/* + * 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.raft; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +import java.lang.reflect.Field; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; + +import org.apache.hugegraph.store.HgStoreEngine; +import org.apache.hugegraph.store.raft.PartitionStateMachine; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; +import org.apache.hugegraph.store.util.ExecutorUtil; +import org.apache.hugegraph.store.util.HgStoreException; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.alipay.sofa.jraft.Status; +import com.alipay.sofa.jraft.error.RaftError; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; + +/** + * Covers the EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL -> RaftError.EBUSY mapping at the + * PartitionStateMachine.onSnapshotSave(SnapshotWriter, Closure) boundary, per the PR review + * comment on #3164. Existing coverage (SnapshotHandlerTest) only asserts the exception thrown + * by SnapshotHandler directly - nothing invoked PartitionStateMachine.onSnapshotSave() itself + * and captured the resulting Closure status. A regression that mapped busy failures to EIO + * instead of EBUSY would still pass there, while in production it would cause jRaft to + * incorrectly escalate to restartRaftNode() instead of simply retrying. + */ +public class PartitionStateMachineTest { + + private static boolean createdTestExecutor; + + /** + * onSnapshotSave() submits its work to HgStoreEngine's static uninterruptibleJobs executor. + * Install a lightweight one via reflection instead of going through the full + * HgStoreEngine.init() bootstrap (rpc server, raft rpc, PD registration, etc.), which would + * conflict with the singleton lifecycle other suites rely on. Guarded so a real init() that + * runs first (e.g. if this ever shares a fork with a StoreEngineTestBase-derived suite) is + * left untouched. + */ + @BeforeClass + public static void ensureUninterruptibleJobsExecutor() throws Exception { + if (HgStoreEngine.getUninterruptibleJobs() == null) { + Field field = HgStoreEngine.class.getDeclaredField("uninterruptibleJobs"); + field.setAccessible(true); + field.set(null, ExecutorUtil.createExecutor("test-psm-u-job", 2, 4, 16, true)); + createdTestExecutor = true; + } + } + + @AfterClass + public static void shutDownTestExecutor() { + if (createdTestExecutor) { + ((ThreadPoolExecutor) HgStoreEngine.getUninterruptibleJobs()).shutdownNow(); + } + } + + @Test + public void testOnSnapshotSaveMapsBusyFailureToEbusy() throws Exception { + SnapshotHandler mockSnapshotHandler = mock(SnapshotHandler.class); + doThrow(new HgStoreException(HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL, + "Partition 0 snapshot save failed: compaction in progress")) + .when(mockSnapshotHandler).onSnapshotSave(any()); + + PartitionStateMachine stateMachine = new PartitionStateMachine(0, mockSnapshotHandler); + Status status = runOnSnapshotSave(stateMachine); + + assertEquals("a busy compaction-range lock must map to EBUSY, not EIO, so jRaft's " + + "snapshot scheduler retries instead of escalating to restartRaftNode()", + RaftError.EBUSY, status.getRaftError()); + } + + @Test + public void testOnSnapshotSaveMapsOrdinaryFailureToEio() throws Exception { + SnapshotHandler mockSnapshotHandler = mock(SnapshotHandler.class); + doThrow(new HgStoreException(HgStoreException.EC_RKDB_EXPORT_SNAPSHOT_FAIL, "disk full")) + .when(mockSnapshotHandler).onSnapshotSave(any()); + + PartitionStateMachine stateMachine = new PartitionStateMachine(0, mockSnapshotHandler); + Status status = runOnSnapshotSave(stateMachine); + + assertEquals("a non-busy save failure must still escalate as EIO", + RaftError.EIO, status.getRaftError()); + } + + /** + * Confirms onSnapshotSave()'s internal lock is released via its finally block even when + * snapshotHandler.onSnapshotSave() throws - the same unconditional-unlock class of bug + * fixed for dbCompaction()'s rangeLock (see HgSnapshotHandlerTest). If the lock were left + * held, every subsequent onSnapshotSave() call on this state machine would hang forever. + */ + @Test + public void testOnSnapshotSaveReleasesLockOnFailure() throws Exception { + SnapshotHandler mockSnapshotHandler = mock(SnapshotHandler.class); + doThrow(new HgStoreException(HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL, + "Partition 0 snapshot save failed: compaction in progress")) + .when(mockSnapshotHandler).onSnapshotSave(any()); + + PartitionStateMachine stateMachine = new PartitionStateMachine(0, mockSnapshotHandler); + runOnSnapshotSave(stateMachine); + + Lock internalLock = getInternalLock(stateMachine); + // The done callback runs before the worker's finally block releases the lock. + assertTrue("onSnapshotSave's finally block must release its lock even when " + + "snapshotHandler.onSnapshotSave() throws, or every later snapshot save " + + "attempt on this partition would hang forever", + internalLock.tryLock(5, TimeUnit.SECONDS)); + internalLock.unlock(); + } + + private static Status runOnSnapshotSave(PartitionStateMachine stateMachine) + throws InterruptedException { + SnapshotWriter stubWriter = mock(SnapshotWriter.class); + AtomicReference result = new AtomicReference<>(); + CountDownLatch latch = new CountDownLatch(1); + + stateMachine.onSnapshotSave(stubWriter, status -> { + result.set(status); + latch.countDown(); + }); + + assertTrue("onSnapshotSave's done closure must be invoked", + latch.await(5, TimeUnit.SECONDS)); + return result.get(); + } + + private static Lock getInternalLock(PartitionStateMachine stateMachine) throws Exception { + Field field = PartitionStateMachine.class.getDeclaredField("lock"); + field.setAccessible(true); + return (Lock) field.get(stateMachine); + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java index ff5ef24acf..e32f1df498 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java @@ -18,30 +18,56 @@ package org.apache.hugegraph.store.core.snapshot; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import java.io.File; import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.pd.grpc.pulse.CleanType; +import org.apache.hugegraph.rocksdb.access.ScanIterator; +import org.apache.hugegraph.store.PartitionEngine; +import org.apache.hugegraph.store.UnitTestBase; +import org.apache.hugegraph.store.business.BusinessHandler; +import org.apache.hugegraph.store.business.BusinessHandlerImpl; +import org.apache.hugegraph.store.consts.PoolNames; import org.apache.hugegraph.store.core.StoreEngineTestBase; import org.apache.hugegraph.store.meta.Partition; +import org.apache.hugegraph.store.meta.asynctask.AbstractAsyncTask; +import org.apache.hugegraph.store.meta.asynctask.AsyncTask; +import org.apache.hugegraph.store.meta.asynctask.AsyncTaskState; import org.apache.hugegraph.store.snapshot.HgSnapshotHandler; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; +import org.apache.hugegraph.store.util.HgStoreException; import org.junit.Before; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import com.alipay.sofa.jraft.entity.RaftOutter; import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; import com.google.protobuf.Message; - public class HgSnapshotHandlerTest extends StoreEngineTestBase { private static HgSnapshotHandler hgSnapshotHandlerUnderTest; + @Rule + public TemporaryFolder tmpDir = new TemporaryFolder(); + @Before public void setUp() throws IOException { hgSnapshotHandlerUnderTest = new HgSnapshotHandler(createPartitionEngine(0)); @@ -184,4 +210,669 @@ public void testFindFileList() { // Verify the results } + + /** + * Test that onSnapshotLoad skips loading (rather than throwing) when should_not_load is + * present but data/ is missing: a locally-saved snapshot deliberately has no data/ dir + * since nothing was meant to load, and should_not_load is checked before the data/ dir. + */ + @Test + public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentButDataMissing() + throws Exception { + // Arrange: snapshot dir has should_not_load but NO data/ subdirectory. + File snapDir = tmpDir.newFolder("snapshot-corrupt"); + File shouldNotLoad = new File(snapDir, "should_not_load"); + Files.write(shouldNotLoad.toPath(), "saved snapshot".getBytes(StandardCharsets.UTF_8)); + // data/ deliberately not created + + SnapshotHandler handler = new SnapshotHandler(createPartitionEngine(1)); + SnapshotReader stubReader = stubReader(snapDir.getAbsolutePath()); + + // Must not throw; should return early at the should_not_load check. + handler.onSnapshotLoad(stubReader, 0L); + } + + /** + * Test that onSnapshotLoad skips loading when snapshot is locally saved (both flags present). + */ + @Test + public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentAndDataExists() throws Exception { + // Arrange: a healthy local snapshot, both should_not_load and data/ present. + File snapDir = tmpDir.newFolder("snapshot-healthy"); + File shouldNotLoad = new File(snapDir, "should_not_load"); + Files.write(shouldNotLoad.toPath(), "saved snapshot".getBytes(StandardCharsets.UTF_8)); + FileUtils.forceMkdir(new File(snapDir, "data")); + + SnapshotHandler handler = new SnapshotHandler(createPartitionEngine(2)); + SnapshotReader stubReader = stubReader(snapDir.getAbsolutePath()); + + // Must not throw; should return early at the should_not_load + data-exists check. + handler.onSnapshotLoad(stubReader, 0L); + } + + /** + * Test that the compaction-range lock used by onSnapshotSave to guard against a concurrent + * compactRange() call is mutually exclusive and releasable, using the real BusinessHandlerImpl + * rather than a mock, so the actual lock instance backing the check is exercised. The + * concurrent attempt runs on a separate thread because the lock is a ReentrantLock: the + * owning thread can always re-acquire it, so checking from the same thread would not + * exercise exclusion. In production dbCompaction() and onSnapshotSave() run on different + * executor threads, which is what this mirrors. + */ + @Test + public void testCompactionRangeLockIsMutuallyExclusiveAndReleasable() throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 3; + + assertEquals("first reservation must succeed", true, + businessHandler.tryLockCompactionRange(partitionId)); + + AtomicBoolean concurrentResult = new AtomicBoolean(); + Thread other = new Thread( + () -> concurrentResult.set(businessHandler.tryLockCompactionRange(partitionId))); + other.start(); + other.join(); + assertEquals("a concurrent reservation from another thread must fail while the first " + + "is held", false, concurrentResult.get()); + + businessHandler.unlockCompactionRange(partitionId); + + assertEquals("reservation must succeed again once released", true, + businessHandler.tryLockCompactionRange(partitionId)); + businessHandler.unlockCompactionRange(partitionId); + } + + /** + * Test that dbCompaction() gives up and skips its pass, rather than blocking forever, when + * a snapshot save is still holding compactionRangeLock after the configured wait. Shortens + * compactionRangeLockWaitMillis for the duration of the test so it does not have to wait out + * the real production timeout, and restores it afterward so other tests are unaffected. + */ + @Test + public void testDbCompactionSkipsWhenRangeLockStillHeldAfterWait() throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 4; + createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(200); + try { + // Simulate a snapshot save that is still in progress. + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.dbCompaction("graph0", partitionId); + + // dbCompaction() runs on compactionPool asynchronously; give it time to hit the + // shortened wait and skip, then confirm it never reached the compacting state + // (doing = -1, set right after the range lock would have been acquired). + Thread.sleep(1000); + assertEquals("dbCompaction must never reach the compacting state while the range " + + "lock is held by the snapshot save", 0, + businessHandler.getState(partitionId).get()); + + // The range lock must still belong to the snapshot save - dbCompaction skipping + // must not have released a lock it never acquired. Check from another thread since + // the lock is a ReentrantLock and the owning (main) thread could always re-acquire it. + AtomicBoolean concurrentResult = new AtomicBoolean(); + Thread other = new Thread(() -> concurrentResult.set( + businessHandler.tryLockCompactionRange(partitionId))); + other.start(); + other.join(); + assertFalse("a concurrent reservation attempt must still fail", concurrentResult.get()); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + } + } + + /** + * Test that dbCompaction() releases the path lock when it is interrupted while waiting + * on compactionRangeLock, rather than leaking it. Before the fix, an InterruptedException + * thrown out of rangeLock.tryLock() propagated straight to the outer catch (which only + * logs), skipping unlock(path) - so every later compaction for that partition would block + * until the 6-hour path-lock timeout. Interrupts a real compactionPool worker thread while + * it is parked in tryLock() (identified by stack trace, since the pool is shared), then + * confirms a second dbCompaction() call is able to complete instead of hanging behind the + * still-held path lock. + */ + @Test + public void testDbCompactionReleasesPathLockWhenInterruptedWaitingForRangeLock() + throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 5; + createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + // Long enough that the worker thread is still parked in tryLock() when interrupted, + // rather than racing a real timeout. + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(60_000); + try { + // Simulate a snapshot save that is still in progress, forcing dbCompaction() onto + // the tryLock(wait) path rather than acquiring the range lock immediately. + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.dbCompaction("graph0", partitionId); + + // dbCompaction() runs asynchronously on compactionPool; give the submitted task + // time to acquire the path lock and start waiting on the range lock. + Thread other = awaitCompactionPoolWorkerBlockedInRangeLockWait(); + assertNotNull("dbCompaction task must be parked in the range lock wait", other); + + // lock(path) must run before the range lock wait, so the path lock is genuinely + // in the doing state here - otherwise the InterruptedException handler's + // unlock(path) below would not actually be releasing anything it holds. + String pathBeforeInterrupt = businessHandler.getLockPath(partitionId); + AtomicInteger pathLockBeforeInterrupt = + businessHandler.getPathLockState(pathBeforeInterrupt); + assertNotNull("path lock must have been initialized before the range lock wait", + pathLockBeforeInterrupt); + assertEquals("path lock must be in the doing state while parked in the range " + + "lock wait, proving lock(path) runs before it rather than after", + BusinessHandler.doing, pathLockBeforeInterrupt.get()); + + other.interrupt(); + // Give the interrupted task time to run its InterruptedException handling and + // return. + Thread.sleep(500); + + // The path lock must have been released by the interrupted task's + // InterruptedException handler, directly confirming the fix rather than relying + // solely on the second dbCompaction() call below to prove it indirectly. + String path = businessHandler.getLockPath(partitionId); + AtomicInteger pathLockState = businessHandler.getPathLockState(path); + assertNotNull("path lock must have been initialized by the interrupted task", + pathLockState); + assertEquals("path lock must be released, not left in the doing state, after the " + + "interrupted task returns", + BusinessHandler.compactionCanStart, pathLockState.get()); + + // The snapshot save still owns the range lock, unaffected by dbCompaction's + // interrupt. + AtomicBoolean concurrentRangeLockResult = new AtomicBoolean(); + Thread rangeLockCheck = new Thread(() -> concurrentRangeLockResult.set( + businessHandler.tryLockCompactionRange(partitionId))); + rangeLockCheck.start(); + rangeLockCheck.join(); + assertFalse("range lock must still belong to the snapshot save", + concurrentRangeLockResult.get()); + + // The path lock, however, must have been released by the interrupted task - + // otherwise this second dbCompaction() call would block on lock(path) until the + // 6-hour timeout instead of reaching the compacting state below once the range + // lock is released. Ownership of the range lock passes to the second + // dbCompaction() call's own worker thread, which acquires and releases it itself - + // do not touch compactionRangeLock again after this point. + businessHandler.unlockCompactionRange(partitionId); + businessHandler.dbCompaction("graph0", partitionId); + + // Compaction on the test's near-empty RocksDB completes almost immediately, so the + // transient "doing" state cannot be reliably observed here - poll for the + // terminal compactionDone state instead. What this proves is that dbCompaction() + // was able to acquire lock(path) at all: before the fix, the leaked path lock + // would have made this call hang on lock(path) until the 6-hour timeout instead + // of ever reaching compactionDone. + long start = System.currentTimeMillis(); + while (businessHandler.getState(partitionId).get() != BusinessHandler.compactionDone && + System.currentTimeMillis() - start < 5000) { + Thread.sleep(50); + } + assertEquals("second dbCompaction() must complete, proving the path lock was " + + "released rather than leaked by the interrupted task", + BusinessHandler.compactionDone, businessHandler.getState(partitionId).get()); + } finally { + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + } + } + + /** + * Test that dbCompaction() with a specific tableName also respects compactionRangeLock, + * instead of calling op.compactRange(tableName) unconditionally regardless of whether a + * snapshot save currently holds the lock. Before the fix (addressing a PR review comment + * on #3164), the tableName.isEmpty() check gated only the pathLock/state-tracking logic - + * the tableName branch's op.compactRange(tableName) call sat outside the + * if (rangeLocked) block entirely, so it ran immediately even while a snapshot save was + * in progress, never touching rangeLock at all. Confirms the tableName-branch worker + * thread is actually found parked in rangeLock.tryLock(), the same way the empty-tableName + * path is in testDbCompactionReleasesPathLockWhenInterruptedWaitingForRangeLock - proof + * the fix nested the tableName branch inside the range lock's wait/hold rather than + * leaving it unguarded. + */ + @Test + public void testDbCompactionWithTableNameRespectsRangeLock() throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 6; + createPartitionEngine(partitionId); + businessHandler.createTable("graph0", partitionId, UnitTestBase.DEFAULT_TEST_TABLE); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + // Long enough that the worker thread is still parked in tryLock() when observed, + // rather than racing a real timeout. + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(60_000); + try { + // Simulate a snapshot save that is still in progress, forcing dbCompaction() onto + // the tryLock(wait) path rather than compacting the table immediately. + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.dbCompaction("graph0", partitionId, UnitTestBase.DEFAULT_TEST_TABLE); + + Thread other = awaitPoolWorkerBlockedInRangeLockWait(PoolNames.COMPACT); + assertNotNull("dbCompaction(tableName) task must wait on the range lock instead " + + "of calling compactRange(tableName) unconditionally while a " + + "snapshot save holds it", + other); + other.interrupt(); + // Give the interrupted task time to run its InterruptedException handling and + // return, so it does not linger holding the pool thread into later tests. + Thread.sleep(200); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + } + } + + /** + * Test that cleanPartition()'s trailing compactRange() call waits on compactionRangeLock, + * even though the delete/scan pass ahead of it does not. Per the PR review comment on + * #3164, the lock must NOT span the whole async cleanup - that scan can take minutes, and + * holding the range lock for that long would fail every onSnapshotSave on the partition + * with EBUSY for the duration. Only the actual compactRange() call (inside the private + * cleanPartition(Partition, Function) overload) is guarded. Confirms the async worker is + * found parked in rangeLock.tryLock() while a snapshot save holds the lock, proving + * compactRange() is still guarded even though the delete pass ahead of it is not. + */ + @Test + public void testCleanPartitionRespectsRangeLockDuringAsyncWork() throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 7; + createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(60_000); + try { + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.cleanPartition("graph0", partitionId, 0, 10, + CleanType.CLEAN_TYPE_KEEP_RANGE); + + Thread other = awaitPoolWorkerBlockedInRangeLockWait("JRaft-Closure-Executor-"); + assertNotNull("cleanPartition's async task must be parked waiting for the range " + + "lock at its trailing compactRange() call while a snapshot save " + + "holds it, proving that call is still guarded even though the " + + "delete/scan pass ahead of it is not", + other); + other.interrupt(); + Thread.sleep(200); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + } + } + + /** + * Test that cleanPartition() still completes its delete/scan pass and marks its async + * task SUCCESS even when the trailing compactRange() call gives up after the range lock + * is still held past the configured wait - per the PR review comment on #3164, a busy + * compaction step must be logged and skipped rather than dropping the whole one-shot + * cleanup. Also confirms skipping compactRange() does not release a lock it never + * acquired (the same unconditional-unlock class of bug fixed for dbCompaction()'s + * rangeLock, see testDbCompactionReleasesPathLockWhenInterruptedWaitingForRangeLock). + * Shortens compactionRangeLockWaitMillis so the test does not wait out the real + * production timeout. + */ + @Test + public void testCleanPartitionSkipsCompactRangeWhenRangeLockStillHeldAfterWait() + throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 8; + PartitionEngine partitionEngine = createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(200); + try { + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.cleanPartition("graph0", partitionId, 0, 10, + CleanType.CLEAN_TYPE_KEEP_RANGE); + + // cleanPartition() runs asynchronously via Utils.runInThread(); give it time to + // finish its (unlocked) delete pass, hit the shortened wait on compactRange(), + // skip it, and mark the async task SUCCESS. + Thread.sleep(1000); + + List tasks = + partitionEngine.getTaskManager().scanAsyncTasks(partitionId, "graph0"); + assertEquals("exactly one CleanTask must have been recorded", 1, tasks.size()); + assertEquals("cleanPartition's delete pass must succeed and mark its task " + + "SUCCESS even though compactRange() was skipped - only the " + + "compaction step is allowed to be skipped, not the whole cleanup", + AsyncTaskState.SUCCESS, + ((AbstractAsyncTask) tasks.get(0)).getState()); + + // The range lock must still belong to the snapshot save - skipping compactRange() + // must not have released a lock it never acquired. Check from another thread + // since the lock is a ReentrantLock and the owning (main) thread could always + // re-acquire it. + AtomicBoolean concurrentResult = new AtomicBoolean(); + Thread other = new Thread(() -> concurrentResult.set( + businessHandler.tryLockCompactionRange(partitionId))); + other.start(); + other.join(); + assertFalse("a concurrent reservation attempt must still fail", concurrentResult.get()); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + } + } + + /** + * Test that a skipped cleanPartition() compactRange() pass is retried - per the user's + * follow-up request on the PR #3164 review thread, a bounded retry must be scheduled so + * the compaction is not silently dropped forever. Holds the range lock the whole time so + * the scheduled retry attempt also has to wait on it, and confirms a retry worker (running + * on the dedicated PoolNames.COMPACT_RETRY pool) is found parked in the same + * ReentrantLock#tryLock() wait the original attempt used - proving the retry actually ran + * and attempted compactRange() again, rather than the skip being a dead end. + */ + @Test + public void testCleanPartitionCompactRangeRetryAttemptsAgainAfterSkip() + throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 10; + createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + long originalRetryDelayMillis = BusinessHandlerImpl.getCleanPartitionCompactRetryDelayMillis(); + int originalMaxRetries = BusinessHandlerImpl.getMaxCleanPartitionCompactRetries(); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(200); + BusinessHandlerImpl.setCleanPartitionCompactRetryDelayMillis(300); + try { + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.cleanPartition("graph0", partitionId, 0, 10, + CleanType.CLEAN_TYPE_KEEP_RANGE); + + // First, the original async worker must hit the shortened wait and skip. Then, + // after the retry delay, the dedicated retry pool must attempt the lock again - + // give this up to 3s total (200ms wait + 300ms delay + generous margin). + Thread other = awaitPoolWorkerBlockedInRangeLockWait(PoolNames.COMPACT_RETRY, 3000); + assertNotNull("a scheduled retry must attempt compactRange() again while the " + + "range lock is still held, proving the skipped compaction is not " + + "simply dropped", other); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + BusinessHandlerImpl.setCleanPartitionCompactRetryDelayMillis(originalRetryDelayMillis); + BusinessHandlerImpl.setMaxCleanPartitionCompactRetries(originalMaxRetries); + } + } + + /** + * Test that the compactRange() retry is bounded rather than retrying forever - per the + * user's explicit request for a "bounded retry". Shrinks maxCleanPartitionCompactRetries + * to 1 so only a single retry attempt is scheduled after the original skip; holds the + * range lock throughout so every attempt (original + the one retry) is forced to skip. + * Confirms the one retry attempt happens (a worker is found parked on + * PoolNames.COMPACT_RETRY), then confirms no further retry is ever scheduled by checking + * no such worker reappears after that attempt also times out and gives up. + */ + @Test + public void testCleanPartitionCompactRangeRetryStopsAfterMaxAttempts() + throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 11; + createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + long originalRetryDelayMillis = BusinessHandlerImpl.getCleanPartitionCompactRetryDelayMillis(); + int originalMaxRetries = BusinessHandlerImpl.getMaxCleanPartitionCompactRetries(); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(200); + BusinessHandlerImpl.setCleanPartitionCompactRetryDelayMillis(300); + BusinessHandlerImpl.setMaxCleanPartitionCompactRetries(1); + try { + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.cleanPartition("graph0", partitionId, 0, 10, + CleanType.CLEAN_TYPE_KEEP_RANGE); + + Thread firstRetry = + awaitPoolWorkerBlockedInRangeLockWait(PoolNames.COMPACT_RETRY, 3000); + assertNotNull("the single allowed retry attempt must still happen", firstRetry); + + // Wait for that one retry attempt to finish timing out on its own 200ms wait and + // abandon (i.e. leave the TIMED_WAITING/tryLock state), so the next poll below + // cannot mistake this same still-in-flight attempt for a second one. + long deadline = System.currentTimeMillis() + 3000; + while (isBlockedInRangeLockTryLock(firstRetry) && + System.currentTimeMillis() < deadline) { + Thread.sleep(20); + } + + // Then wait well past another retry-delay window - since + // maxCleanPartitionCompactRetries is 1, no second retry attempt must ever be + // scheduled, so no new worker should appear parked on the range lock. + Thread secondRetry = + awaitPoolWorkerBlockedInRangeLockWait(PoolNames.COMPACT_RETRY, 1500); + assertNull("a bounded retry must give up after maxCleanPartitionCompactRetries " + + "attempts instead of retrying forever", secondRetry); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + BusinessHandlerImpl.setCleanPartitionCompactRetryDelayMillis(originalRetryDelayMillis); + BusinessHandlerImpl.setMaxCleanPartitionCompactRetries(originalMaxRetries); + } + } + + /** + * Test that a scheduled compactRange() retry abandons itself if the partition has since + * been destroyed, rather than reaching into a torn-down partition. Holds the range lock so + * the original attempt skips and schedules a retry, then destroys the partition engine + * before the retry delay elapses. Confirms the retry never even attempts the lock (no + * worker parked on PoolNames.COMPACT_RETRY), since the abandon check runs before the + * tryLock() call. + * + *

The retry delay (1500ms) is set well beyond both the initial skip's own wait (200ms) + * and the sleep before destroy (600ms), so the destroy is guaranteed to land in the gap + * between "skip has happened, retry is scheduled" and "retry fires" regardless of scheduler + * jitter under load - avoiding the flaky window where a too-short delay let the retry fire + * (and start its own tryLock wait) before the destroy/abandon-check could beat it there.

+ */ + @Test + public void testCleanPartitionCompactRangeRetryAbandonsWhenPartitionDestroyed() + throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 12; + createPartitionEngine(partitionId); + long originalWaitMillis = BusinessHandlerImpl.getCompactionRangeLockWaitMillis(); + long originalRetryDelayMillis = BusinessHandlerImpl.getCleanPartitionCompactRetryDelayMillis(); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(200); + BusinessHandlerImpl.setCleanPartitionCompactRetryDelayMillis(1500); + try { + assertTrue("snapshot save must reserve the range lock", + businessHandler.tryLockCompactionRange(partitionId)); + + businessHandler.cleanPartition("graph0", partitionId, 0, 10, + CleanType.CLEAN_TYPE_KEEP_RANGE); + + // Give the original attempt time to hit its shortened wait and skip, scheduling a + // retry ~1500ms out, then destroy the partition well before that retry fires. + Thread.sleep(600); + getStoreEngine().destroyPartitionEngine(partitionId, List.of("graph0")); + + Thread retry = awaitPoolWorkerBlockedInRangeLockWait(PoolNames.COMPACT_RETRY, 3000); + assertNull("a retry must abandon itself once the partition no longer exists, " + + "rather than attempting compactRange() on a torn-down partition", + retry); + } finally { + businessHandler.unlockCompactionRange(partitionId); + BusinessHandlerImpl.setCompactionRangeLockWaitMillis(originalWaitMillis); + BusinessHandlerImpl.setCleanPartitionCompactRetryDelayMillis(originalRetryDelayMillis); + } + } + + /** + * Test the race this whole lock was introduced for: while cleanPartition()'s async work + * holds compactionRangeLock (simulating its real compactRange() call in progress), + * onSnapshotSave() must not block or corrupt data - it must fail fast with an + * HgStoreException coded EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL. That exact code is what + * PartitionStateMachine#onSnapshotSave inspects to report RaftError.EBUSY instead of + * RaftError.EIO, so jRaft's snapshot scheduler retries independently rather than the + * failure escalating to reportError()/restartRaftNode(). Reserves the lock on a separate + * thread (rather than driving a real cleanPartition()) to isolate the assertion to + * onSnapshotSave's side of the race - a separate thread is required because + * compactionRangeLock is a ReentrantLock, which would let onSnapshotSave's tryLock + * silently re-enter if called from the same (main) thread that reserved it, defeating the + * point of the test. This mirrors how cleanPartition() and onSnapshotSave() genuinely run + * on different executor threads in production. + */ + @Test + public void testOnSnapshotSaveFailsWithBusyCodeWhileCleanPartitionHoldsRangeLock() + throws Exception { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 9; + PartitionEngine partitionEngine = createPartitionEngine(partitionId); + + // The lock must be reserved and released by the SAME thread, since it is a + // ReentrantLock - so the "cleanPartition worker" thread is kept alive across both + // halves of the test via these latches, rather than the main thread reserving it and + // a throwaway thread (illegally) releasing it. + AtomicBoolean lockReserved = new AtomicBoolean(); + CountDownLatch lockAcquired = new CountDownLatch(1); + CountDownLatch releaseNow = new CountDownLatch(1); + Thread cleanPartitionWorker = new Thread(() -> { + lockReserved.set(businessHandler.tryLockCompactionRange(partitionId)); + lockAcquired.countDown(); + try { + releaseNow.await(); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + } + businessHandler.unlockCompactionRange(partitionId); + }); + cleanPartitionWorker.start(); + lockAcquired.await(); + assertTrue("cleanPartition's compactRange() must reserve the range lock", + lockReserved.get()); + try { + SnapshotHandler snapshotHandler = new SnapshotHandler(partitionEngine); + String snapshotPath = tmpDir.newFolder("snapshot-busy-" + partitionId) + .getAbsolutePath(); + SnapshotWriter stubWriter = stubWriter(snapshotPath); + + HgStoreException ex = assertThrows( + "onSnapshotSave must throw while cleanPartition holds the range lock, " + + "rather than blocking or racing saveSnapshot's checkpoint against " + + "cleanPartition's compactRange()", + HgStoreException.class, + () -> snapshotHandler.onSnapshotSave(stubWriter)); + + assertEquals("the busy code must be EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL - this is the " + + "exact code PartitionStateMachine#onSnapshotSave checks to report " + + "RaftError.EBUSY (transient, jRaft retries) instead of RaftError.EIO " + + "(escalates to restartRaftNode())", + HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL, ex.getCode()); + assertTrue("exception message must mention compaction is in progress", + ex.getMessage().contains("compaction in progress")); + + // The failed onSnapshotSave must not have released a lock it never acquired - + // cleanPartition's compactRange() must still hold it. Check from another thread + // since the lock is a ReentrantLock and the owning (main) thread could always + // re-acquire it. + AtomicBoolean concurrentResult = new AtomicBoolean(); + Thread other = new Thread(() -> concurrentResult.set( + businessHandler.tryLockCompactionRange(partitionId))); + other.start(); + other.join(); + assertFalse("range lock must still belong to cleanPartition's compactRange()", + concurrentResult.get()); + } finally { + releaseNow.countDown(); + cleanPartitionWorker.join(); + } + } + + private static SnapshotWriter stubWriter(String path) { + return new SnapshotWriter() { + @Override public boolean saveMeta(RaftOutter.SnapshotMeta meta) { return false; } + @Override public boolean addFile(String fileName, Message fileMeta) { return false; } + @Override public boolean removeFile(String fileName) { return false; } + @Override public void close(boolean keepDataOnError) {} + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } + + /** + * Polls the compactionPool worker threads for one parked inside ReentrantLock#tryLock + * (the compactionRangeLock wait in dbCompaction()), up to 5s. The pool is shared/static, so + * this cannot target the task directly - it identifies the right worker by stack trace + * instead. tryLock(timeout, unit) parks via LockSupport.parkNanos, which reports as + * TIMED_WAITING rather than WAITING. + */ + private static Thread awaitCompactionPoolWorkerBlockedInRangeLockWait() + throws InterruptedException { + return awaitPoolWorkerBlockedInRangeLockWait(PoolNames.COMPACT); + } + + /** + * Same as {@link #awaitCompactionPoolWorkerBlockedInRangeLockWait()}, but generalized to + * any thread-name prefix - cleanPartition()'s async work runs on jraft's shared + * "JRaft-Closure-Executor-" pool (via Utils.runInThread()) rather than compactionPool. + */ + private static Thread awaitPoolWorkerBlockedInRangeLockWait(String threadNamePrefix) + throws InterruptedException { + return awaitPoolWorkerBlockedInRangeLockWait(threadNamePrefix, 5000); + } + + /** + * Same as {@link #awaitPoolWorkerBlockedInRangeLockWait(String)}, but with a caller-chosen + * timeout - used to assert the *absence* of a blocked worker (e.g. after a bounded retry + * has exhausted its attempts) without waiting out the default 5s. + */ + private static Thread awaitPoolWorkerBlockedInRangeLockWait(String threadNamePrefix, + long maxWaitMillis) + throws InterruptedException { + long start = System.currentTimeMillis(); + while (System.currentTimeMillis() - start < maxWaitMillis) { + for (Thread t : Thread.getAllStackTraces().keySet()) { + if (t.getName().startsWith(threadNamePrefix) && + t.getState() == Thread.State.TIMED_WAITING && + isBlockedInRangeLockTryLock(t)) { + return t; + } + } + Thread.sleep(50); + } + return null; + } + + private static boolean isBlockedInRangeLockTryLock(Thread t) { + boolean inBusinessHandlerImpl = false; + boolean inLockSupportPark = false; + for (StackTraceElement frame : t.getStackTrace()) { + String className = frame.getClassName(); + if (className.equals(BusinessHandlerImpl.class.getName())) { + inBusinessHandlerImpl = true; + } else if (className.equals("java.util.concurrent.locks.LockSupport")) { + inLockSupportPark = true; + } + } + return inBusinessHandlerImpl && inLockSupportPark; + } + + private static SnapshotReader stubReader(String path) { + return new SnapshotReader() { + @Override public RaftOutter.SnapshotMeta load() { return null; } + @Override public String generateURIForCopy() { return null; } + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java new file mode 100644 index 0000000000..ca888f7035 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java @@ -0,0 +1,175 @@ +/* + * 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.snapshot; + +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.contains; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.util.Set; + +import org.apache.hugegraph.store.HgStoreEngine; +import org.apache.hugegraph.store.PartitionEngine; +import org.apache.hugegraph.store.business.BusinessHandler; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; +import org.apache.hugegraph.store.util.HgStoreException; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import com.alipay.sofa.jraft.entity.RaftOutter; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; +import com.google.protobuf.Message; + +public class SnapshotHandlerTest { + + @Rule + public TemporaryFolder tmpDir = new TemporaryFolder(); + + /** + * When the compaction-range lock cannot be reserved (a compaction is actively running), + * onSnapshotSave must throw immediately without touching saveSnapshot. The exception signals + * jRaft, which will retry the snapshot later. jRaft's snapshot scheduler runs independently + * and frequently (default 300s, user config 1800s), so the next attempt will succeed once + * compaction releases the lock. + */ + @Test + public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.tryLockCompactionRange(0)).thenReturn(false); + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotWriter stubWriter = stubWriter("/tmp/snapshot"); + + HgStoreException ex = assertThrows( + "onSnapshotSave must throw when the compaction-range lock is held", + HgStoreException.class, + () -> handler.onSnapshotSave(stubWriter)); + + assertTrue("Exception message must mention the partition", + ex.getMessage().contains("0")); + assertTrue("Exception message must mention compaction is in progress", + ex.getMessage().contains("compaction in progress")); + verify(mockBusinessHandler, never()).saveSnapshot(any(), any(), anyInt()); + } + + /** + * When the compaction-range lock is free, onSnapshotSave must reserve it, call saveSnapshot, + * and release the lock afterwards. + */ + @Test + public void testOnSnapshotSaveCallsSaveSnapshotWhenNotBusy() throws Exception { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + final String snapshotPath = tmpDir.newFolder("snap-not-busy").getAbsolutePath(); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.tryLockCompactionRange(0)).thenReturn(true); + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotWriter stubWriter = stubWriter(snapshotPath); + + handler.onSnapshotSave(stubWriter); + + // Verify: saveSnapshot was called with concrete path containing expected data dir + String expectedDataDir = snapshotPath + File.separator + "data"; + verify(mockBusinessHandler).saveSnapshot( + contains(expectedDataDir), // Must contain the snapshot path + /data + eq(""), // graphName (empty string) + eq(0)); // groupId (partition 0) + verify(mockBusinessHandler).unlockCompactionRange(0); + } + + /** + * When should_not_load is absent (the common corruption variant: leader crashed + * mid-checkpoint with no flag ever written) and data/ is missing, onSnapshotLoad + * must throw a diagnostic naming the corrupt snapshot directory, rather than + * falling through to businessHandler.loadSnapshot. + */ + @Test + public void testOnSnapshotLoadThrowsWhenShouldNotLoadAbsentAndDataMissing() throws Exception { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + when(mockEngine.getGroupId()).thenReturn(3); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + + String snapshotPath = tmpDir.newFolder("snapshot-no-flag-no-data").getAbsolutePath(); + // should_not_load deliberately not created; data/ deliberately not created + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotReader stubReader = stubReader(snapshotPath); + + HgStoreException ex = assertThrows( + "onSnapshotLoad must throw when data/ is missing, flag or no flag", + HgStoreException.class, + () -> handler.onSnapshotLoad(stubReader, 0L)); + + assertTrue("Exception message must name the corrupt snapshot directory", + ex.getMessage().contains(snapshotPath)); + } + + private static SnapshotWriter stubWriter(String path) { + return new SnapshotWriter() { + @Override public boolean saveMeta(RaftOutter.SnapshotMeta meta) { return false; } + @Override public boolean addFile(String fileName, Message fileMeta) { return false; } + @Override public boolean removeFile(String fileName) { return false; } + @Override public void close(boolean keepDataOnError) {} + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } + + private static SnapshotReader stubReader(String path) { + return new SnapshotReader() { + @Override public RaftOutter.SnapshotMeta load() { return null; } + @Override public String generateURIForCopy() { return null; } + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java index f3b1f31d29..be24270ea7 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java @@ -17,13 +17,17 @@ package org.apache.hugegraph.store.raftcore; +import org.apache.hugegraph.store.core.raft.PartitionStateMachineTest; +import org.apache.hugegraph.store.core.snapshot.SnapshotHandlerTest; import org.junit.runner.RunWith; import org.junit.runners.Suite; @RunWith(Suite.class) @Suite.SuiteClasses({ BytesCarrierTest.class, - ZeroByteStringHelperTest.class + ZeroByteStringHelperTest.class, + SnapshotHandlerTest.class, + PartitionStateMachineTest.class }) public class RaftSuiteTest {