From 7ee5d420ca7301d8038f1fe981dd82623c8961c9 Mon Sep 17 00:00:00 2001 From: Vaibhav Joshi Date: Tue, 18 Aug 2026 15:58:04 +0530 Subject: [PATCH 1/7] fix(store): throw on compaction-busy snapshot save; validate data/ on load (#3162) - Throw HgStoreException in onSnapshotSave when RocksDB compaction is in progress so JRaft retries rather than committing an empty snapshot dir. - In onSnapshotLoad, fall through to the real load path when should_not_load is present but data/ is missing (JVM-killed mid-checkpoint), so JRaft can signal the error and request a fresh snapshot from the leader. - Add unit tests covering both fix paths in HgSnapshotHandlerTest. - Add docker/test/test-snapshot-corruption.sh, a deterministic Docker reproducer that confirms the bug and validates the fix (--fixed mode). Fixes #3162 Co-Authored-By: Claude --- docker/test/test-snapshot-corruption.sh | 314 ++++++++++++++++++ .../store/snapshot/SnapshotHandler.java | 14 +- .../store/core/StoreEngineTestBase.java | 6 + .../core/snapshot/HgSnapshotHandlerTest.java | 164 +++++++++ 4 files changed, 495 insertions(+), 3 deletions(-) create mode 100755 docker/test/test-snapshot-corruption.sh diff --git a/docker/test/test-snapshot-corruption.sh b/docker/test/test-snapshot-corruption.sh new file mode 100755 index 0000000000..6133536106 --- /dev/null +++ b/docker/test/test-snapshot-corruption.sh @@ -0,0 +1,314 @@ + +#!/usr/bin/env bash +# +# 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. + + +# test-snapshot-corruption.sh — deterministic reproducer for the HStore snapshot corruption bug +# +# Requires: Docker Desktop >= 20.10, >= 12 GB allocated to Docker, Docker Compose v2 +# Run from the repo root: +# bash docker/hbase/test/test-snapshot-corruption.sh # confirm bug is present (buggy image) +# bash docker/hbase/test/test-snapshot-corruption.sh --fixed # confirm bug is absent (fixed image) + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../../../.." && pwd)" +COMPOSE_FILE="$SCRIPT_DIR/../../docker-compose-3pd-3store-3server.yml" +HUGEGRAPH_VERSION="${HUGEGRAPH_VERSION:-1.7.0}" +VOLUME_PREFIX="hugegraph-3x3" +STORE_LOG="hugegraph-store.log" +FIXED_MODE=false +[[ "${1:-}" == "--fixed" ]] && FIXED_MODE=true + +RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; NC='\033[0m' +log() { echo -e "${GREEN}[repro]${NC} $*"; } +warn() { echo -e "${YELLOW}[repro]${NC} $*"; } +fail() { echo -e "${RED}[repro] FAIL${NC} $*" >&2; exit 1; } + +# In --fixed mode use the locally-built patched image. +# Build it from source if it doesn't exist yet so the caller only needs --fixed. +PATCHED_IMAGE="hugegraph/store:patched" +DOCKERFILE="$SCRIPT_DIR/../../Dockerfile.store-patched" +PATCHED_JAR="$SCRIPT_DIR/../../hg-store-node-${HUGEGRAPH_VERSION}.jar" +JAR_SOURCE="$REPO_ROOT/hugegraph-store/hg-store-node/target/hg-store-node-${HUGEGRAPH_VERSION}.jar" + +if $FIXED_MODE; then + STORE_IMAGE="${STORE_IMAGE:-$PATCHED_IMAGE}" + if [[ "$STORE_IMAGE" == "$PATCHED_IMAGE" ]] && \ + ! docker image inspect "$PATCHED_IMAGE" >/dev/null 2>&1; then + log "Patched image not found — building from source..." + if [[ ! -f "$JAR_SOURCE" ]]; then + log " Compiling hugegraph-store (this takes a minute)..." + mvn package -pl hugegraph-store/hg-store-node -am -DskipTests -q \ + -f "$REPO_ROOT/pom.xml" + fi + cp "$JAR_SOURCE" "$PATCHED_JAR" + docker build -f "$DOCKERFILE" -t "$PATCHED_IMAGE" \ + "$(dirname "$DOCKERFILE")" >/dev/null + log " Built $PATCHED_IMAGE." + fi +else + STORE_IMAGE="${STORE_IMAGE:-hugegraph/store:${HUGEGRAPH_VERSION}}" +fi + +log "Store image: $STORE_IMAGE (fixed-mode: $FIXED_MODE)" +log "Compose File: $COMPOSE_FILE" + +# If a non-default store image is requested, write a temporary compose override that +# replaces the store image — without modifying the committed compose file. +OVERRIDE_FILE="" +DEFAULT_IMAGE="hugegraph/store:${HUGEGRAPH_VERSION}" +if [ "$STORE_IMAGE" != "$DEFAULT_IMAGE" ]; then + OVERRIDE_FILE="$(mktemp /tmp/snapshot-test-override-XXXXXX.yml)" + cat > "$OVERRIDE_FILE" </dev/null 2>&1; then log "$label up."; return 0; fi + sleep 3 + done + fail "$label not healthy after $((tries * 3))s" +} + +# ── Step 1: Start cluster ───────────────────────────────────────────────────── +log "Step 1: Tearing down any previous run and starting a clean cluster..." +HUGEGRAPH_VERSION="$HUGEGRAPH_VERSION" \ + docker compose $COMPOSE_ARGS down -v --remove-orphans 2>&1 | tail -3 || true +HUGEGRAPH_VERSION="$HUGEGRAPH_VERSION" \ + docker compose $COMPOSE_ARGS up -d \ + --scale server0=0 --scale server1=0 --scale server2=0 \ + 2>&1 | grep -E "Started|Healthy|healthy" | tail -5 || true + +wait_http "http://localhost:8620/v1/health" "pd0" 60 +wait_http "http://localhost:8520/v1/health" "store0" 60 +wait_http "http://localhost:8521/v1/health" "store1" 60 +wait_http "http://localhost:8522/v1/health" "store2" 60 + +# Raft partition dirs are created lazily when the server first registers a graph. +# Start server0 just long enough for init-store to run, then stop it. +# We only need the init_complete flag to be written — we do NOT wait for /versions +# because start-hugegraph.sh has a 120s JVM-ready timeout that can expire on a cold +# distributed cluster, causing the entrypoint to exit and Docker to restart the +# container, resetting the timer indefinitely. +if ! docker exec hg-store0 sh -c 'ls /hugegraph-store/storage/raft/ 2>/dev/null | grep -qE "^[0-9]{5}$"' 2>/dev/null; then + log " Fresh cluster: starting server0 briefly to initialise partitions..." + HUGEGRAPH_VERSION="$HUGEGRAPH_VERSION" \ + HUGEGRAPH_STORE_IMAGE="$STORE_IMAGE" \ + docker compose $COMPOSE_ARGS up -d server0 2>&1 | tail -2 || true + + log " Waiting for init-store to complete (up to 120s)..." + for i in $(seq 1 40); do + if docker exec hg-server0 test -f /hugegraph-server/docker/init_complete 2>/dev/null; then + log " init_complete flag found after ~$((i * 3))s." + break + fi + sleep 3 + done + docker exec hg-server0 test -f /hugegraph-server/docker/init_complete 2>/dev/null \ + || fail "init-store did not complete within 120s" + + # Give Raft groups a moment to create their partition dirs + sleep 5 + docker compose $COMPOSE_ARGS stop server0 2>/dev/null || true + log " server0 stopped — partitions initialised." +fi + +# ── Step 2: Ensure committed snapshots exist on store0 ─────────────────────── +log "Step 2: Flushing + snapshotting all store nodes..." +for port in 8520 8521 8522; do + curl -fsS "http://localhost:${port}/test/flush" >/dev/null && log " :${port} flush OK" + curl -fsS "http://localhost:${port}/test/snapshot" >/dev/null && log " :${port} snapshot triggered" +done +log "Waiting 20s for Raft snapshot commits..." +sleep 20 + +SNAP_COUNT=$(docker run --rm \ + -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ + busybox sh -c 'find /hugegraph-store/storage/raft -name "data" -type d | wc -l') +log "store0 has $SNAP_COUNT committed snapshot data/ directories." +[ "$SNAP_COUNT" -ge 1 ] || fail "No committed snapshots on store0. Retry." + +# ── Step 3: Stop all stores ─────────────────────────────────────────────────── +log "Step 3: Stopping all store nodes..." +docker stop hg-store0 hg-store1 hg-store2 >/dev/null +log "All stores stopped." + +# ── Step 4: Corrupt one partition snapshot on store0 ───────────────────────── +# Two sub-cases of the bug: +# +# Sub-case A (race: state==doing at snapshot-save time) — tested in default mode: +# onSnapshotSave returns early → neither data/ nor should_not_load written. +# Snapshot dir has only __raft_snapshot_meta. +# On load: goes straight to loadSnapshot(missingDir) → "not exists" → stuck. +# Fix 1 (throw instead of return) prevents this snapshot from ever being committed. +# +# Sub-case B (JVM killed after should_not_load but before data/ completes) — tested in --fixed mode: +# Snapshot dir has __raft_snapshot_meta + should_not_load, but no data/. +# On load (buggy): shouldNotLoad() == true → silent return → partition silently has no data. +# On load (fixed): Fix 2 detects data/ is missing → logs warning → falls through to +# loadSnapshot → throws "not exists" → JRaft signals error → leader rescues. +# +log "Step 4: Corrupting one snapshot on store0 (sub-case $( $FIXED_MODE && echo B || echo A ))..." +TARGET=$(docker run --rm \ + -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ + busybox sh -c ' + for meta in $(find /hugegraph-store/storage/raft -name "__raft_snapshot_meta" | sort); do + snap=$(dirname "$meta") + if [ -d "$snap/data" ] && [ -f "$snap/should_not_load" ]; then + echo "$snap"; break + fi + done + ') + +[ -n "$TARGET" ] || fail "No suitable snapshot found (need data/ + should_not_load + __raft_snapshot_meta)" + +PARTITION_ID=$(echo "$TARGET" | grep -oE '/[0-9]{5}/' | head -1 | tr -d '/') +SNAP_NAME=$(basename "$TARGET") + +if $FIXED_MODE; then + # Sub-case B: remove only data/, keep should_not_load. + # Buggy image: shouldNotLoad() fires, silently returns — no error logged. + # Fixed image (Fix 2): detects data/ missing, logs warning, falls through. + log " Target: partition $PARTITION_ID / $SNAP_NAME" + log " Removing data/ only — keeping should_not_load (sub-case B)" + docker run --rm \ + -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ + busybox sh -c "rm -rf '${TARGET}/data' + echo 'Contents after corruption:'; ls '${TARGET}'" +else + # Sub-case A: remove both data/ and should_not_load — exactly what the race produces. + log " Target: partition $PARTITION_ID / $SNAP_NAME" + log " Removing data/ and should_not_load — leaving only __raft_snapshot_meta (sub-case A)" + docker run --rm \ + -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ + busybox sh -c "rm -rf '${TARGET}/data' '${TARGET}/should_not_load' + echo 'Contents after corruption:'; ls '${TARGET}'" +fi + +# ── Step 5: Start store0 alone ──────────────────────────────────────────────── +log "Step 5: Starting store0 alone (no peers — prevents leader snapshot rescue)..." +docker start hg-store0 >/dev/null +log "Polling store0 logs for snapshot load result (up to 90s)..." +for i in $(seq 1 45); do + if docker exec hg-store0 grep -qE "not exists|Fail to init|onSnapshotLoad success|warn.*corrupt" \ + /hugegraph-store/logs/$STORE_LOG 2>/dev/null; then + log " Snapshot load result detected after ~$((i * 2))s." + break + fi + sleep 2 +done +sleep 3 + +APP_LOGS=$(docker exec hg-store0 cat /hugegraph-store/logs/$STORE_LOG 2>/dev/null || true) + +# ── Step 6: Evaluate result ─────────────────────────────────────────────────── +if $FIXED_MODE; then + # Sub-case B: we corrupted should_not_load+data/ (kept should_not_load, removed data/). + # Buggy behaviour: shouldNotLoad() silently returns — "skip to load snapshot" logged, no error. + # Fixed behaviour (Fix 2): detects data/ is missing → logs the warn line → falls through + # → loadSnapshot throws "not exists" → JRaft signals error (visible in logs). + # The key assertion: the warn line IS present, proving Fix 2 caught the corrupt snapshot + # instead of silently accepting it. + log "Step 6: Verifying Fix 2 — should_not_load + missing data/ must be caught, not silently skipped..." + WARN_LINE="should_not_load flag present but data dir" + if grep -q "$WARN_LINE" <<< "$APP_LOGS"; then + echo "" + echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + echo -e "${GREEN} FIX 2 VERIFIED — corrupt snapshot detected, not silently accepted${NC}" + echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + echo "" + echo "Key log lines:" + grep -E "$WARN_LINE|not exists|Fail to init" <<< "$APP_LOGS" | head -6 | sed 's/^/ /' || true + else + fail "Fix 2 did not fire — warn line not found. The corrupt snapshot was silently accepted." + fi +else + log "Step 6: Checking logs for the bug..." + if grep -q "not exists" <<< "$APP_LOGS"; then + echo "" + echo -e "${RED}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + echo -e "${RED} BUG REPRODUCED — partition ${PARTITION_ID} is stuck${NC}" + echo -e "${RED}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + echo "" + echo "Key log lines:" + grep -E "not exists|Fail to init|onSnapshotLoad failed|StateMachine on error" \ + <<< "$APP_LOGS" | head -6 | sed 's/^/ /' || true + else + fail "Expected error lines not found. Check: docker exec hg-store0 cat /hugegraph-store/logs/$STORE_LOG" + fi + + # ── Step 7: Health endpoint still 200 ──────────────────────────────────── + log "Step 7: Health endpoint check..." + HTTP_CODE="000" + for i in $(seq 1 10); do + HTTP_CODE=$(curl -sw "%{http_code}" http://localhost:8520/v1/health -o /dev/null 2>/dev/null || echo "000") + [ "$HTTP_CODE" != "000" ] && break + sleep 3 + done + warn " /v1/health → HTTP $HTTP_CODE (200 = misleading — broken partition is invisible)" + + # ── Step 8: Restart does not recover ───────────────────────────────────── + log "Step 8: Confirming plain restart does not recover partition $PARTITION_ID..." + # Record log line count before restart so we only examine lines written after it. + LOG_LINES_BEFORE=$(docker exec hg-store0 wc -l /hugegraph-store/logs/$STORE_LOG 2>/dev/null | awk '{print $1}' || echo 0) + docker stop hg-store0 >/dev/null + docker start hg-store0 >/dev/null + for i in $(seq 1 45); do + NEW_LINES=$(docker exec hg-store0 \ + awk "NR > $LOG_LINES_BEFORE" /hugegraph-store/logs/$STORE_LOG 2>/dev/null || true) + if grep -qE "not exists|Fail to init" <<< "$NEW_LINES"; then + log " Error lines found after ~$((i * 2))s." + break + fi + sleep 2 + done + POST_RESTART_LOGS=$(docker exec hg-store0 \ + awk "NR > $LOG_LINES_BEFORE" /hugegraph-store/logs/$STORE_LOG 2>/dev/null || true) + if grep -q "not exists" <<< "$POST_RESTART_LOGS"; then + warn " Confirmed: restart loops again. The node is permanently stuck." + else + warn " Error lines not found in post-restart output — JVM may need more time:" + warn " docker exec hg-store0 tail -20 /hugegraph-store/logs/$STORE_LOG" + fi +fi + +# ── Step 9: Restore full store cluster ─────────────────────────────────────── +log "Step 9: Restoring store cluster (store1 + store2)..." +docker start hg-store1 hg-store2 >/dev/null +log " Leader will install a fresh snapshot on store0 for partition $PARTITION_ID." + +echo "" +echo -e "${GREEN} Run complete. Clean up with:${NC}" +echo " docker compose -f docker/docker-compose-3pd-3store-3server.yml down -v" 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..d98dbaa12d 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 @@ -96,7 +96,9 @@ public void onSnapshotSave(final SnapshotWriter writer) throws HgStoreException Integer groupId = partitionEngine.getGroupId(); AtomicInteger state = businessHandler.getState(groupId); if (state != null && state.get() == BusinessHandler.doing) { - return; + throw new HgStoreException( + String.format("Partition %d is busy (compaction in progress), " + + "snapshot save skipped", groupId)); } // rocks db snapshot final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; @@ -172,8 +174,14 @@ public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) thr // No need to load locally saved snapshots if (shouldNotLoad(reader)) { - log.info("skip to load snapshot because of should_not_load flag"); - return; + final String dataDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; + if (new File(dataDir).exists()) { + log.info("skip to load snapshot because of should_not_load flag"); + return; + } + log.warn("Raft {} should_not_load flag present but data dir {} is missing — " + + "snapshot is corrupt, proceeding to load path", + partitionEngine.getGroupId(), dataDir); } // Use snapshot directly 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..a740130da0 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,6 +28,7 @@ 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; @@ -61,6 +62,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"); 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..ebaf28bbf8 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,19 +18,34 @@ package org.apache.hugegraph.store.core.snapshot; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; 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.atomic.AtomicInteger; import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.store.HgStoreEngine; +import org.apache.hugegraph.store.PartitionEngine; +import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.core.StoreEngineTestBase; import org.apache.hugegraph.store.meta.Partition; 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; @@ -42,6 +57,9 @@ 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)); @@ -49,6 +67,152 @@ public void setUp() throws IOException { FileUtils.forceMkdir(new File("/tmp/snapshot/data")); } + // ── Fix 1: onSnapshotSave must throw when compaction is in progress ──────── + + /** + * Before the fix, onSnapshotSave silently returned when state == doing, + * causing JRaft to commit an empty snapshot dir with no data/. + * After the fix it must throw HgStoreException so JRaft retries instead. + */ + @Test + public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { + // Build a SnapshotHandler wired to a mock PartitionEngine whose BusinessHandler + // reports state == doing (compaction active) for partition 0. + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + AtomicInteger doingState = new AtomicInteger(BusinessHandler.doing); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.getState(0)).thenReturn(doingState); + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + + SnapshotWriter stubWriter = stubWriter("/tmp/snapshot"); + + HgStoreException ex = assertThrows( + "onSnapshotSave must throw when state == doing", + HgStoreException.class, + () -> handler.onSnapshotSave(stubWriter)); + + assertTrue("Exception message must mention the partition", + ex.getMessage().contains("0")); + assertTrue("Exception message must describe the cause", + ex.getMessage().contains("compaction in progress")); + } + + /** + * When state is NOT doing (e.g. compactionDone), onSnapshotSave must not throw. + */ + @Test + public void testOnSnapshotSaveDoesNotThrowWhenNotBusy() throws Exception { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + // state == compactionDone (not doing) — save should proceed normally + AtomicInteger doneState = new AtomicInteger(BusinessHandler.compactionDone); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.getState(0)).thenReturn(doneState); + + // saveSnapshot is a no-op via the mock, so we just need it not to throw at the guard + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotWriter stubWriter = stubWriter(tmpDir.newFolder("snap-not-busy").getAbsolutePath()); + + // No exception should propagate from the state guard. + // (saveSnapshot will throw because the mock returns null for it — that's fine, + // we only care the doing-check is not hit.) + try { + handler.onSnapshotSave(stubWriter); + } catch (HgStoreException e) { + assertFalse("Must not be the compaction-busy exception", + e.getMessage().contains("compaction in progress")); + } + } + + // ── Fix 2: onSnapshotLoad must not silently skip a corrupt snapshot ──────── + + /** + * Before the fix, onSnapshotLoad returned silently when should_not_load was present, + * even if data/ was missing — leaving the partition in an undefined state. + * After the fix it must fall through to the real load path and throw, + * so JRaft can signal the error and the leader can install a fresh snapshot. + */ + @Test + public void testOnSnapshotLoadFallsThroughWhenShouldNotLoadPresentButDataMissing() + 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()); + + // The fix causes execution to fall through shouldNotLoad() and call + // businessHandler.loadSnapshot(missingDataDir) which throws HgStoreException. + assertThrows( + "onSnapshotLoad must throw when should_not_load present but data/ missing", + HgStoreException.class, + () -> handler.onSnapshotLoad(stubReader, 0L)); + } + + /** + * When should_not_load is present AND data/ also exists, onSnapshotLoad must + * return early (normal locally-saved snapshot — no load needed). + */ + @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); + } + + // ── Stub helpers ────────────────────────────────────────────────────────── + + 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() {} + }; + } + @Test public void testGetPartitions() { // Run the test From f2b7219fb33c256726963b12a7eec3a9f5b48c0e Mon Sep 17 00:00:00 2001 From: Vaibhav Joshi Date: Wed, 19 Aug 2026 18:55:07 +0530 Subject: [PATCH 2/7] fix(store): throw on compaction-busy snapshot save; validate data/ on load (#3162) - Addressed Review comments --- docker/test/test-snapshot-corruption.sh | 35 +++++++++---------- .../store/snapshot/SnapshotHandler.java | 4 +-- 2 files changed, 18 insertions(+), 21 deletions(-) diff --git a/docker/test/test-snapshot-corruption.sh b/docker/test/test-snapshot-corruption.sh index 6133536106..43f071fa3b 100755 --- a/docker/test/test-snapshot-corruption.sh +++ b/docker/test/test-snapshot-corruption.sh @@ -1,4 +1,3 @@ - #!/usr/bin/env bash # # Licensed to the Apache Software Foundation (ASF) under one or more @@ -27,8 +26,8 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" -REPO_ROOT="$(cd "$SCRIPT_DIR/../../../.." && pwd)" -COMPOSE_FILE="$SCRIPT_DIR/../../docker-compose-3pd-3store-3server.yml" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +COMPOSE_FILE="$SCRIPT_DIR/../docker-compose-3pd-3store-3server.yml" HUGEGRAPH_VERSION="${HUGEGRAPH_VERSION:-1.7.0}" VOLUME_PREFIX="hugegraph-3x3" STORE_LOG="hugegraph-store.log" @@ -40,27 +39,26 @@ log() { echo -e "${GREEN}[repro]${NC} $*"; } warn() { echo -e "${YELLOW}[repro]${NC} $*"; } fail() { echo -e "${RED}[repro] FAIL${NC} $*" >&2; exit 1; } -# In --fixed mode use the locally-built patched image. -# Build it from source if it doesn't exist yet so the caller only needs --fixed. -PATCHED_IMAGE="hugegraph/store:patched" -DOCKERFILE="$SCRIPT_DIR/../../Dockerfile.store-patched" -PATCHED_JAR="$SCRIPT_DIR/../../hg-store-node-${HUGEGRAPH_VERSION}.jar" JAR_SOURCE="$REPO_ROOT/hugegraph-store/hg-store-node/target/hg-store-node-${HUGEGRAPH_VERSION}.jar" if $FIXED_MODE; then - STORE_IMAGE="${STORE_IMAGE:-$PATCHED_IMAGE}" - if [[ "$STORE_IMAGE" == "$PATCHED_IMAGE" ]] && \ - ! docker image inspect "$PATCHED_IMAGE" >/dev/null 2>&1; then + STORE_IMAGE="${STORE_IMAGE:-hugegraph/store:patched}" + if [[ "$STORE_IMAGE" == "hugegraph/store:patched" ]] && \ + ! docker image inspect "hugegraph/store:patched" >/dev/null 2>&1; then log "Patched image not found — building from source..." if [[ ! -f "$JAR_SOURCE" ]]; then log " Compiling hugegraph-store (this takes a minute)..." mvn package -pl hugegraph-store/hg-store-node -am -DskipTests -q \ -f "$REPO_ROOT/pom.xml" fi - cp "$JAR_SOURCE" "$PATCHED_JAR" - docker build -f "$DOCKERFILE" -t "$PATCHED_IMAGE" \ - "$(dirname "$DOCKERFILE")" >/dev/null - log " Built $PATCHED_IMAGE." + BUILD_CTX="$(mktemp -d)" + cp "$JAR_SOURCE" "$BUILD_CTX/hg-store-node-${HUGEGRAPH_VERSION}.jar" + cat > "$BUILD_CTX/Dockerfile" </dev/null + log " Built hugegraph/store:patched." fi else STORE_IMAGE="${STORE_IMAGE:-hugegraph/store:${HUGEGRAPH_VERSION}}" @@ -72,8 +70,7 @@ log "Compose File: $COMPOSE_FILE" # If a non-default store image is requested, write a temporary compose override that # replaces the store image — without modifying the committed compose file. OVERRIDE_FILE="" -DEFAULT_IMAGE="hugegraph/store:${HUGEGRAPH_VERSION}" -if [ "$STORE_IMAGE" != "$DEFAULT_IMAGE" ]; then +if [ "$STORE_IMAGE" != "hugegraph/store:${HUGEGRAPH_VERSION}" ]; then OVERRIDE_FILE="$(mktemp /tmp/snapshot-test-override-XXXXXX.yml)" cat > "$OVERRIDE_FILE" < Date: Thu, 20 Aug 2026 10:04:09 +0530 Subject: [PATCH 3/7] fix(store): throw on compaction-busy snapshot save; validate data/ on load (#3162) - Added comment in test-snapshot-corruption.sh to make clear that its just load-path reproducer for the HStore snapshot corruption bug --- docker/test/test-snapshot-corruption.sh | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/docker/test/test-snapshot-corruption.sh b/docker/test/test-snapshot-corruption.sh index 43f071fa3b..ffba4179e0 100755 --- a/docker/test/test-snapshot-corruption.sh +++ b/docker/test/test-snapshot-corruption.sh @@ -16,12 +16,21 @@ # limitations under the License. -# test-snapshot-corruption.sh — deterministic reproducer for the HStore snapshot corruption bug +# test-snapshot-corruption.sh — load-path reproducer for the HStore snapshot corruption bug +# +# This script covers the LOAD side of the fix (SnapshotHandler.onSnapshotLoad): +# default mode — simulates Sub-case A outcome (missing data/ dir) to verify the load error +# --fixed mode — simulates Sub-case B (should_not_load present, data/ missing) to verify Fix 2 +# +# The SAVE side of the fix (SnapshotHandler.onSnapshotSave throwing when compaction state==doing) +# cannot be reproduced deterministically here: /test/compact submits a background job and returns +# immediately, so the race window is too narrow to hit reliably from a shell script. +# Save-side coverage lives in the unit test: HgSnapshotHandlerTest. # # Requires: Docker Desktop >= 20.10, >= 12 GB allocated to Docker, Docker Compose v2 # Run from the repo root: -# bash docker/hbase/test/test-snapshot-corruption.sh # confirm bug is present (buggy image) -# bash docker/hbase/test/test-snapshot-corruption.sh --fixed # confirm bug is absent (fixed image) +# bash docker/hbase/test/test-snapshot-corruption.sh # confirm load-path bug is present (buggy image) +# bash docker/hbase/test/test-snapshot-corruption.sh --fixed # confirm load-path fix is active (fixed image) set -euo pipefail @@ -165,11 +174,13 @@ log "All stores stopped." # ── Step 4: Corrupt one partition snapshot on store0 ───────────────────────── # Two sub-cases of the bug: # -# Sub-case A (race: state==doing at snapshot-save time) — tested in default mode: -# onSnapshotSave returns early → neither data/ nor should_not_load written. -# Snapshot dir has only __raft_snapshot_meta. +# Sub-case A (race: state==doing at snapshot-save time) — load-path simulated in default mode: +# The actual race cannot be triggered deterministically from a shell script (see header). +# We instead simulate the outcome: manually remove data/ and should_not_load, leaving only +# __raft_snapshot_meta, which is what the race would produce. # On load: goes straight to loadSnapshot(missingDir) → "not exists" → stuck. -# Fix 1 (throw instead of return) prevents this snapshot from ever being committed. +# Fix 1 (throw instead of return in onSnapshotSave) prevents this snapshot from ever being +# committed; this script validates only the resulting load-path error, not the throw itself. # # Sub-case B (JVM killed after should_not_load but before data/ completes) — tested in --fixed mode: # Snapshot dir has __raft_snapshot_meta + should_not_load, but no data/. From 46ce0c44d6d09021462796a02951f910fadbd540 Mon Sep 17 00:00:00 2001 From: Vaibhav Joshi Date: Mon, 31 Aug 2026 21:31:30 +0530 Subject: [PATCH 4/7] fix(store): throw on compaction-busy snapshot save; validate data/ on load (#3162) - Addressed review comments. Added few Unit test cases, -Removed test-snapshot-corruption.sh because scenario is already covered by UTs. --- docker/test/test-snapshot-corruption.sh | 322 ------------------ .../core/snapshot/HgSnapshotHandlerTest.java | 213 ++++-------- .../core/snapshot/SnapshotHandlerTest.java | 130 +++++++ 3 files changed, 189 insertions(+), 476 deletions(-) delete mode 100755 docker/test/test-snapshot-corruption.sh create mode 100644 hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java diff --git a/docker/test/test-snapshot-corruption.sh b/docker/test/test-snapshot-corruption.sh deleted file mode 100755 index ffba4179e0..0000000000 --- a/docker/test/test-snapshot-corruption.sh +++ /dev/null @@ -1,322 +0,0 @@ -#!/usr/bin/env bash -# -# 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. - - -# test-snapshot-corruption.sh — load-path reproducer for the HStore snapshot corruption bug -# -# This script covers the LOAD side of the fix (SnapshotHandler.onSnapshotLoad): -# default mode — simulates Sub-case A outcome (missing data/ dir) to verify the load error -# --fixed mode — simulates Sub-case B (should_not_load present, data/ missing) to verify Fix 2 -# -# The SAVE side of the fix (SnapshotHandler.onSnapshotSave throwing when compaction state==doing) -# cannot be reproduced deterministically here: /test/compact submits a background job and returns -# immediately, so the race window is too narrow to hit reliably from a shell script. -# Save-side coverage lives in the unit test: HgSnapshotHandlerTest. -# -# Requires: Docker Desktop >= 20.10, >= 12 GB allocated to Docker, Docker Compose v2 -# Run from the repo root: -# bash docker/hbase/test/test-snapshot-corruption.sh # confirm load-path bug is present (buggy image) -# bash docker/hbase/test/test-snapshot-corruption.sh --fixed # confirm load-path fix is active (fixed image) - -set -euo pipefail - -SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" -REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" -COMPOSE_FILE="$SCRIPT_DIR/../docker-compose-3pd-3store-3server.yml" -HUGEGRAPH_VERSION="${HUGEGRAPH_VERSION:-1.7.0}" -VOLUME_PREFIX="hugegraph-3x3" -STORE_LOG="hugegraph-store.log" -FIXED_MODE=false -[[ "${1:-}" == "--fixed" ]] && FIXED_MODE=true - -RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; NC='\033[0m' -log() { echo -e "${GREEN}[repro]${NC} $*"; } -warn() { echo -e "${YELLOW}[repro]${NC} $*"; } -fail() { echo -e "${RED}[repro] FAIL${NC} $*" >&2; exit 1; } - -JAR_SOURCE="$REPO_ROOT/hugegraph-store/hg-store-node/target/hg-store-node-${HUGEGRAPH_VERSION}.jar" - -if $FIXED_MODE; then - STORE_IMAGE="${STORE_IMAGE:-hugegraph/store:patched}" - if [[ "$STORE_IMAGE" == "hugegraph/store:patched" ]] && \ - ! docker image inspect "hugegraph/store:patched" >/dev/null 2>&1; then - log "Patched image not found — building from source..." - if [[ ! -f "$JAR_SOURCE" ]]; then - log " Compiling hugegraph-store (this takes a minute)..." - mvn package -pl hugegraph-store/hg-store-node -am -DskipTests -q \ - -f "$REPO_ROOT/pom.xml" - fi - BUILD_CTX="$(mktemp -d)" - cp "$JAR_SOURCE" "$BUILD_CTX/hg-store-node-${HUGEGRAPH_VERSION}.jar" - cat > "$BUILD_CTX/Dockerfile" </dev/null - log " Built hugegraph/store:patched." - fi -else - STORE_IMAGE="${STORE_IMAGE:-hugegraph/store:${HUGEGRAPH_VERSION}}" -fi - -log "Store image: $STORE_IMAGE (fixed-mode: $FIXED_MODE)" -log "Compose File: $COMPOSE_FILE" - -# If a non-default store image is requested, write a temporary compose override that -# replaces the store image — without modifying the committed compose file. -OVERRIDE_FILE="" -if [ "$STORE_IMAGE" != "hugegraph/store:${HUGEGRAPH_VERSION}" ]; then - OVERRIDE_FILE="$(mktemp /tmp/snapshot-test-override-XXXXXX.yml)" - cat > "$OVERRIDE_FILE" </dev/null 2>&1; then log "$label up."; return 0; fi - sleep 3 - done - fail "$label not healthy after $((tries * 3))s" -} - -# ── Step 1: Start cluster ───────────────────────────────────────────────────── -log "Step 1: Tearing down any previous run and starting a clean cluster..." -HUGEGRAPH_VERSION="$HUGEGRAPH_VERSION" \ - docker compose $COMPOSE_ARGS down -v --remove-orphans 2>&1 | tail -3 || true -HUGEGRAPH_VERSION="$HUGEGRAPH_VERSION" \ - docker compose $COMPOSE_ARGS up -d \ - --scale server0=0 --scale server1=0 --scale server2=0 \ - 2>&1 | grep -E "Started|Healthy|healthy" | tail -5 || true - -wait_http "http://localhost:8620/v1/health" "pd0" 60 -wait_http "http://localhost:8520/v1/health" "store0" 60 -wait_http "http://localhost:8521/v1/health" "store1" 60 -wait_http "http://localhost:8522/v1/health" "store2" 60 - -# Raft partition dirs are created lazily when the server first registers a graph. -# Start server0 just long enough for init-store to run, then stop it. -# We only need the init_complete flag to be written — we do NOT wait for /versions -# because start-hugegraph.sh has a 120s JVM-ready timeout that can expire on a cold -# distributed cluster, causing the entrypoint to exit and Docker to restart the -# container, resetting the timer indefinitely. -if ! docker exec hg-store0 sh -c 'ls /hugegraph-store/storage/raft/ 2>/dev/null | grep -qE "^[0-9]{5}$"' 2>/dev/null; then - log " Fresh cluster: starting server0 briefly to initialise partitions..." - HUGEGRAPH_VERSION="$HUGEGRAPH_VERSION" \ - HUGEGRAPH_STORE_IMAGE="$STORE_IMAGE" \ - docker compose $COMPOSE_ARGS up -d server0 2>&1 | tail -2 || true - - log " Waiting for init-store to complete (up to 120s)..." - for i in $(seq 1 40); do - if docker exec hg-server0 test -f /hugegraph-server/docker/init_complete 2>/dev/null; then - log " init_complete flag found after ~$((i * 3))s." - break - fi - sleep 3 - done - docker exec hg-server0 test -f /hugegraph-server/docker/init_complete 2>/dev/null \ - || fail "init-store did not complete within 120s" - - # Give Raft groups a moment to create their partition dirs - sleep 5 - docker compose $COMPOSE_ARGS stop server0 2>/dev/null || true - log " server0 stopped — partitions initialised." -fi - -# ── Step 2: Ensure committed snapshots exist on store0 ─────────────────────── -log "Step 2: Flushing + snapshotting all store nodes..." -for port in 8520 8521 8522; do - curl -fsS "http://localhost:${port}/test/flush" >/dev/null && log " :${port} flush OK" - curl -fsS "http://localhost:${port}/test/snapshot" >/dev/null && log " :${port} snapshot triggered" -done -log "Waiting 20s for Raft snapshot commits..." -sleep 20 - -SNAP_COUNT=$(docker run --rm \ - -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ - busybox sh -c 'find /hugegraph-store/storage/raft -name "data" -type d | wc -l') -log "store0 has $SNAP_COUNT committed snapshot data/ directories." -[ "$SNAP_COUNT" -ge 1 ] || fail "No committed snapshots on store0. Retry." - -# ── Step 3: Stop all stores ─────────────────────────────────────────────────── -log "Step 3: Stopping all store nodes..." -docker stop hg-store0 hg-store1 hg-store2 >/dev/null -log "All stores stopped." - -# ── Step 4: Corrupt one partition snapshot on store0 ───────────────────────── -# Two sub-cases of the bug: -# -# Sub-case A (race: state==doing at snapshot-save time) — load-path simulated in default mode: -# The actual race cannot be triggered deterministically from a shell script (see header). -# We instead simulate the outcome: manually remove data/ and should_not_load, leaving only -# __raft_snapshot_meta, which is what the race would produce. -# On load: goes straight to loadSnapshot(missingDir) → "not exists" → stuck. -# Fix 1 (throw instead of return in onSnapshotSave) prevents this snapshot from ever being -# committed; this script validates only the resulting load-path error, not the throw itself. -# -# Sub-case B (JVM killed after should_not_load but before data/ completes) — tested in --fixed mode: -# Snapshot dir has __raft_snapshot_meta + should_not_load, but no data/. -# On load (buggy): shouldNotLoad() == true → silent return → partition silently has no data. -# On load (fixed): Fix 2 detects data/ is missing → logs warning → falls through to -# loadSnapshot → throws "not exists" → JRaft signals error → leader rescues. -# -log "Step 4: Corrupting one snapshot on store0 (sub-case $( $FIXED_MODE && echo B || echo A ))..." -TARGET=$(docker run --rm \ - -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ - busybox sh -c ' - for meta in $(find /hugegraph-store/storage/raft -name "__raft_snapshot_meta" | sort); do - snap=$(dirname "$meta") - if [ -d "$snap/data" ] && [ -f "$snap/should_not_load" ]; then - echo "$snap"; break - fi - done - ') - -[ -n "$TARGET" ] || fail "No suitable snapshot found (need data/ + should_not_load + __raft_snapshot_meta)" - -PARTITION_ID=$(echo "$TARGET" | grep -oE '/[0-9]{5}/' | head -1 | tr -d '/') -SNAP_NAME=$(basename "$TARGET") - -if $FIXED_MODE; then - # Sub-case B: remove only data/, keep should_not_load. - # Buggy image: shouldNotLoad() fires, silently returns — no error logged. - # Fixed image (Fix 2): detects data/ missing, logs warning, falls through. - log " Target: partition $PARTITION_ID / $SNAP_NAME" - log " Removing data/ only — keeping should_not_load (sub-case B)" - docker run --rm \ - -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ - busybox sh -c "rm -rf '${TARGET}/data' - echo 'Contents after corruption:'; ls '${TARGET}'" -else - # Sub-case A: remove both data/ and should_not_load — exactly what the race produces. - log " Target: partition $PARTITION_ID / $SNAP_NAME" - log " Removing data/ and should_not_load — leaving only __raft_snapshot_meta (sub-case A)" - docker run --rm \ - -v "${VOLUME_PREFIX}_hg-store0-data:/hugegraph-store/storage" \ - busybox sh -c "rm -rf '${TARGET}/data' '${TARGET}/should_not_load' - echo 'Contents after corruption:'; ls '${TARGET}'" -fi - -# ── Step 5: Start store0 alone ──────────────────────────────────────────────── -log "Step 5: Starting store0 alone (no peers — prevents leader snapshot rescue)..." -docker start hg-store0 >/dev/null -log "Polling store0 logs for snapshot load result (up to 90s)..." -for i in $(seq 1 45); do - if docker exec hg-store0 grep -qE "not exists|Fail to init|onSnapshotLoad success|warn.*corrupt" \ - /hugegraph-store/logs/$STORE_LOG 2>/dev/null; then - log " Snapshot load result detected after ~$((i * 2))s." - break - fi - sleep 2 -done -sleep 3 - -APP_LOGS=$(docker exec hg-store0 cat /hugegraph-store/logs/$STORE_LOG 2>/dev/null || true) - -# ── Step 6: Evaluate result ─────────────────────────────────────────────────── -if $FIXED_MODE; then - # Sub-case B: we corrupted should_not_load+data/ (kept should_not_load, removed data/). - # Buggy behaviour: shouldNotLoad() silently returns — "skip to load snapshot" logged, no error. - # Fixed behaviour (Fix 2): detects data/ is missing → logs the warn line → falls through - # → loadSnapshot throws "not exists" → JRaft signals error (visible in logs). - # The key assertion: the warn line IS present, proving Fix 2 caught the corrupt snapshot - # instead of silently accepting it. - log "Step 6: Verifying Fix 2 — should_not_load + missing data/ must be caught, not silently skipped..." - WARN_LINE="should_not_load flag present but data dir" - if grep -q "$WARN_LINE" <<< "$APP_LOGS"; then - echo "" - echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" - echo -e "${GREEN} FIX 2 VERIFIED — corrupt snapshot detected, not silently accepted${NC}" - echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" - echo "" - echo "Key log lines:" - grep -E "$WARN_LINE|not exists|Fail to init" <<< "$APP_LOGS" | head -6 | sed 's/^/ /' || true - else - fail "Fix 2 did not fire — warn line not found. The corrupt snapshot was silently accepted." - fi -else - log "Step 6: Checking logs for the bug..." - if grep -q "not exists" <<< "$APP_LOGS"; then - echo "" - echo -e "${RED}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" - echo -e "${RED} BUG REPRODUCED — partition ${PARTITION_ID} is stuck${NC}" - echo -e "${RED}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" - echo "" - echo "Key log lines:" - grep -E "not exists|Fail to init|onSnapshotLoad failed|StateMachine on error" \ - <<< "$APP_LOGS" | head -6 | sed 's/^/ /' || true - else - fail "Expected error lines not found. Check: docker exec hg-store0 cat /hugegraph-store/logs/$STORE_LOG" - fi - - # ── Step 7: Health endpoint still 200 ──────────────────────────────────── - log "Step 7: Health endpoint check..." - HTTP_CODE="000" - for i in $(seq 1 10); do - HTTP_CODE=$(curl -sw "%{http_code}" http://localhost:8520/v1/health -o /dev/null 2>/dev/null || echo "000") - [ "$HTTP_CODE" != "000" ] && break - sleep 3 - done - warn " /v1/health → HTTP $HTTP_CODE (200 = misleading — broken partition is invisible)" - - # ── Step 8: Restart does not recover ───────────────────────────────────── - log "Step 8: Confirming plain restart does not recover partition $PARTITION_ID..." - # Record log line count before restart so we only examine lines written after it. - LOG_LINES_BEFORE=$(docker exec hg-store0 wc -l /hugegraph-store/logs/$STORE_LOG 2>/dev/null | awk '{print $1}' || echo 0) - docker stop hg-store0 >/dev/null - docker start hg-store0 >/dev/null - for i in $(seq 1 45); do - NEW_LINES=$(docker exec hg-store0 \ - awk "NR > $LOG_LINES_BEFORE" /hugegraph-store/logs/$STORE_LOG 2>/dev/null || true) - if grep -qE "not exists|Fail to init" <<< "$NEW_LINES"; then - log " Error lines found after ~$((i * 2))s." - break - fi - sleep 2 - done - POST_RESTART_LOGS=$(docker exec hg-store0 \ - awk "NR > $LOG_LINES_BEFORE" /hugegraph-store/logs/$STORE_LOG 2>/dev/null || true) - if grep -q "not exists" <<< "$POST_RESTART_LOGS"; then - warn " Confirmed: restart loops again. The node is permanently stuck." - else - warn " Error lines not found in post-restart output — JVM may need more time:" - warn " docker exec hg-store0 tail -20 /hugegraph-store/logs/$STORE_LOG" - fi -fi - -# ── Step 9: Restore full store cluster ─────────────────────────────────────── -log "Step 9: Restoring store cluster (store1 + store2)..." -docker start hg-store1 hg-store2 >/dev/null -log " Leader will install a fresh snapshot on store0 for partition $PARTITION_ID." - -echo "" -echo -e "${GREEN} Run complete. Clean up with:${NC}" -echo " docker compose -f docker/docker-compose-3pd-3store-3server.yml down -v" 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 ebaf28bbf8..72acb2012c 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,11 +18,7 @@ package org.apache.hugegraph.store.core.snapshot; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; -import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; import java.io.File; import java.io.IOException; @@ -31,17 +27,18 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.io.FileUtils; -import org.apache.hugegraph.store.HgStoreEngine; -import org.apache.hugegraph.store.PartitionEngine; -import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.core.StoreEngineTestBase; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.snapshot.HgSnapshotHandler; import org.apache.hugegraph.store.snapshot.SnapshotHandler; import org.apache.hugegraph.store.util.HgStoreException; + +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; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -67,152 +64,6 @@ public void setUp() throws IOException { FileUtils.forceMkdir(new File("/tmp/snapshot/data")); } - // ── Fix 1: onSnapshotSave must throw when compaction is in progress ──────── - - /** - * Before the fix, onSnapshotSave silently returned when state == doing, - * causing JRaft to commit an empty snapshot dir with no data/. - * After the fix it must throw HgStoreException so JRaft retries instead. - */ - @Test - public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { - // Build a SnapshotHandler wired to a mock PartitionEngine whose BusinessHandler - // reports state == doing (compaction active) for partition 0. - PartitionEngine mockEngine = mock(PartitionEngine.class); - HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); - BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); - - AtomicInteger doingState = new AtomicInteger(BusinessHandler.doing); - - when(mockEngine.getGroupId()).thenReturn(0); - when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); - when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); - when(mockBusinessHandler.getState(0)).thenReturn(doingState); - - SnapshotHandler handler = new SnapshotHandler(mockEngine); - - SnapshotWriter stubWriter = stubWriter("/tmp/snapshot"); - - HgStoreException ex = assertThrows( - "onSnapshotSave must throw when state == doing", - HgStoreException.class, - () -> handler.onSnapshotSave(stubWriter)); - - assertTrue("Exception message must mention the partition", - ex.getMessage().contains("0")); - assertTrue("Exception message must describe the cause", - ex.getMessage().contains("compaction in progress")); - } - - /** - * When state is NOT doing (e.g. compactionDone), onSnapshotSave must not throw. - */ - @Test - public void testOnSnapshotSaveDoesNotThrowWhenNotBusy() throws Exception { - PartitionEngine mockEngine = mock(PartitionEngine.class); - HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); - BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); - - // state == compactionDone (not doing) — save should proceed normally - AtomicInteger doneState = new AtomicInteger(BusinessHandler.compactionDone); - - when(mockEngine.getGroupId()).thenReturn(0); - when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); - when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); - when(mockBusinessHandler.getState(0)).thenReturn(doneState); - - // saveSnapshot is a no-op via the mock, so we just need it not to throw at the guard - SnapshotHandler handler = new SnapshotHandler(mockEngine); - SnapshotWriter stubWriter = stubWriter(tmpDir.newFolder("snap-not-busy").getAbsolutePath()); - - // No exception should propagate from the state guard. - // (saveSnapshot will throw because the mock returns null for it — that's fine, - // we only care the doing-check is not hit.) - try { - handler.onSnapshotSave(stubWriter); - } catch (HgStoreException e) { - assertFalse("Must not be the compaction-busy exception", - e.getMessage().contains("compaction in progress")); - } - } - - // ── Fix 2: onSnapshotLoad must not silently skip a corrupt snapshot ──────── - - /** - * Before the fix, onSnapshotLoad returned silently when should_not_load was present, - * even if data/ was missing — leaving the partition in an undefined state. - * After the fix it must fall through to the real load path and throw, - * so JRaft can signal the error and the leader can install a fresh snapshot. - */ - @Test - public void testOnSnapshotLoadFallsThroughWhenShouldNotLoadPresentButDataMissing() - 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()); - - // The fix causes execution to fall through shouldNotLoad() and call - // businessHandler.loadSnapshot(missingDataDir) which throws HgStoreException. - assertThrows( - "onSnapshotLoad must throw when should_not_load present but data/ missing", - HgStoreException.class, - () -> handler.onSnapshotLoad(stubReader, 0L)); - } - - /** - * When should_not_load is present AND data/ also exists, onSnapshotLoad must - * return early (normal locally-saved snapshot — no load needed). - */ - @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); - } - - // ── Stub helpers ────────────────────────────────────────────────────────── - - 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() {} - }; - } - @Test public void testGetPartitions() { // Run the test @@ -348,4 +199,58 @@ public void testFindFileList() { // Verify the results } + + /** + * Test that onSnapshotLoad validates corruption when should_not_load is present but data/ missing. + */ + @Test + public void testOnSnapshotLoadFallsThroughWhenShouldNotLoadPresentButDataMissing() + 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()); + + // The fix causes execution to fall through shouldNotLoad() and call + // businessHandler.loadSnapshot(missingDataDir) which throws HgStoreException. + assertThrows( + "onSnapshotLoad must throw when should_not_load present but data/ missing", + HgStoreException.class, + () -> 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); + } + + 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..a860b82447 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java @@ -0,0 +1,130 @@ +/* + * 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.contains; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +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.SnapshotWriter; +import com.google.protobuf.Message; + +public class SnapshotHandlerTest { + + @Rule + public TemporaryFolder tmpDir = new TemporaryFolder(); + + /** + * When state is doing (compaction in progress), onSnapshotSave must throw + * immediately. 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 snapshot attempt will succeed after compaction completes. + */ + @Test + public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + AtomicInteger doingState = new AtomicInteger(BusinessHandler.doing); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.getState(0)).thenReturn(doingState); + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotWriter stubWriter = stubWriter("/tmp/snapshot"); + + HgStoreException ex = assertThrows( + "onSnapshotSave must throw when state == doing", + 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")); + } + + /** + * When state is NOT doing (e.g. compactionDone or null), onSnapshotSave must not throw. + * It should proceed and call saveSnapshot with concrete path verification. + */ + @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(); + + // state == compactionDone (not doing) — save should proceed immediately + AtomicInteger doneState = new AtomicInteger(BusinessHandler.compactionDone); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.getState(0)).thenReturn(doneState); + + 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) + } + + 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() {} + }; + } +} From 302844cea3eaddc59e33852c2af1321d8aea899b Mon Sep 17 00:00:00 2001 From: Vaibhav Joshi Date: Sat, 5 Sep 2026 13:55:16 +0530 Subject: [PATCH 5/7] fix(store): guard snapshot save against corruption and compaction races (#3162) Addresses review comments: - onSnapshotSave/dbCompaction shared a non-atomic state check, letting saves race with compaction; coordinate both through a dedicated per-partition lock, checked non-blockingly on both sides - add EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL so the busy-save case has its own grep-able error code, and fix the exception text (compaction, not "skipped") and a stray non-ASCII em dash - onSnapshotLoad checks data/ before should_not_load, so a snapshot missing its flag is reported as corrupt instead of failing later with an unrelated RocksDB path error - drop the duplicate jraft/protobuf imports in HgSnapshotHandlerTest - register SnapshotHandlerTest in RaftSuiteTest and HgSnapshotHandlerTest in CoreSuiteTest, and run store-core-test in CI, so both actually execute instead of being skipped by every bound surefire profile --- .github/workflows/pd-store-ci.yml | 11 ++- .../store/business/BusinessHandler.java | 8 ++ .../store/business/BusinessHandlerImpl.java | 46 ++++++++++- .../store/options/RaftRocksdbOptions.java | 10 +++ .../store/snapshot/SnapshotHandler.java | 81 ++++++++++--------- .../store/util/HgStoreException.java | 4 +- .../store/core/BatchGraphIsolationTest.java | 2 - .../hugegraph/store/core/CoreSuiteTest.java | 15 ++-- .../store/core/StoreEngineTestBase.java | 3 - .../core/snapshot/HgSnapshotHandlerTest.java | 50 +++++++++--- .../core/snapshot/SnapshotHandlerTest.java | 75 +++++++++++++---- .../store/raftcore/RaftSuiteTest.java | 4 +- 12 files changed, 225 insertions(+), 84 deletions(-) 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-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..e227808079 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,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; 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..a5aa8388cf 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 @@ -40,6 +40,7 @@ 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; @@ -138,6 +139,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; @@ -1415,10 +1422,27 @@ 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()) { + // A snapshot save is currently reserving this partition's + // range lock. Skip this compaction pass rather than block + // the compactionPool thread on it - the next scheduled/ + // triggered compaction will retry. + log.info("Partition {} skip dbCompaction, snapshot save in " + + "progress", id); + 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 +1508,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 { 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..7287d44fbd 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,6 +56,15 @@ private static RocksdbConfig getRocksdbConfig(HugeConfig options) { } private static void registerRaftRocksdbConfig(HugeConfig options) { + // 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. + synchronized (RaftRocksdbOptions.class) { + if (raftRocksdbConfigRegistered) { + return; + } + raftRocksdbConfigRegistered = true; + } Cache blockCache = new LRUCache(SizeUnit.GB); BlockBasedTableConfig tableConfig = new BlockBasedTableConfig() .setIndexType(IndexType.kTwoLevelIndexSearch) 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 0ca4f4c3e5..86c236fb97 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,37 +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) { - throw new HgStoreException( - String.format("Partition %d is busy (compaction in progress), " + - "snapshot save skipped", groupId)); + 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); } } @@ -171,21 +174,23 @@ 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; + + 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)); + } // No need to load locally saved snapshots if (shouldNotLoad(reader)) { - final File dataDir = new File(snapshotDir + File.separator + SNAPSHOT_DATA_PATH); - if (dataDir.isDirectory()) { - log.info("skip to load snapshot because of should_not_load flag"); - return; - } - log.warn("Raft {} should_not_load flag present but data dir {} is missing — " + - "snapshot is corrupt, proceeding to load path", - partitionEngine.getGroupId(), dataDir); + log.info("skip to load snapshot because of should_not_load flag"); + return; } // 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..38c6c49904 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 @@ -33,11 +33,9 @@ public class HgStoreException extends RuntimeException { public static final int EC_RKDB_DOMERGE_FAIL = 1207; public static final int EC_RKDB_DOGET_FAIL = 1208; public static final int EC_RKDB_PD_FAIL = 1209; - public static final int EC_RKDB_TRUNCATE_FAIL = 1212; 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_METRIC_FAIL = 1401; + public static final int EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL = 1217; private static final long serialVersionUID = 5193624480997934335L; private final int code; 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 a740130da0..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 @@ -34,8 +34,6 @@ import org.junit.AfterClass; import org.junit.BeforeClass; -import com.alipay.sofa.jraft.util.StorageOptionsFactory; - import lombok.extern.slf4j.Slf4j; /** @@ -76,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/snapshot/HgSnapshotHandlerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java index 72acb2012c..54f890bd5c 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 @@ -27,18 +27,15 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.core.StoreEngineTestBase; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.snapshot.HgSnapshotHandler; import org.apache.hugegraph.store.snapshot.SnapshotHandler; import org.apache.hugegraph.store.util.HgStoreException; - -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; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -49,7 +46,6 @@ import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; import com.google.protobuf.Message; - public class HgSnapshotHandlerTest extends StoreEngineTestBase { private static HgSnapshotHandler hgSnapshotHandlerUnderTest; @@ -204,7 +200,7 @@ public void testFindFileList() { * Test that onSnapshotLoad validates corruption when should_not_load is present but data/ missing. */ @Test - public void testOnSnapshotLoadFallsThroughWhenShouldNotLoadPresentButDataMissing() + public void testOnSnapshotLoadThrowsWhenShouldNotLoadPresentButDataMissing() throws Exception { // Arrange: snapshot dir has should_not_load but NO data/ subdirectory. File snapDir = tmpDir.newFolder("snapshot-corrupt"); @@ -215,8 +211,8 @@ public void testOnSnapshotLoadFallsThroughWhenShouldNotLoadPresentButDataMissing SnapshotHandler handler = new SnapshotHandler(createPartitionEngine(1)); SnapshotReader stubReader = stubReader(snapDir.getAbsolutePath()); - // The fix causes execution to fall through shouldNotLoad() and call - // businessHandler.loadSnapshot(missingDataDir) which throws HgStoreException. + // The missing data/ dir is checked before should_not_load, so this throws + // immediately rather than falling through to businessHandler.loadSnapshot. assertThrows( "onSnapshotLoad must throw when should_not_load present but data/ missing", HgStoreException.class, @@ -228,7 +224,7 @@ public void testOnSnapshotLoadFallsThroughWhenShouldNotLoadPresentButDataMissing */ @Test public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentAndDataExists() throws Exception { - // Arrange: a healthy local snapshot — both should_not_load and data/ present. + // 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)); @@ -237,10 +233,42 @@ public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentAndDataExists() throw 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. + // 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); + } + private static SnapshotReader stubReader(String path) { return new SnapshotReader() { @Override public RaftOutter.SnapshotMeta load() { return null; } 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 index a860b82447..ca888f7035 100644 --- 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 @@ -19,15 +19,17 @@ 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 java.util.concurrent.atomic.AtomicInteger; import org.apache.hugegraph.store.HgStoreEngine; import org.apache.hugegraph.store.PartitionEngine; @@ -39,6 +41,7 @@ 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; @@ -48,10 +51,11 @@ public class SnapshotHandlerTest { public TemporaryFolder tmpDir = new TemporaryFolder(); /** - * When state is doing (compaction in progress), onSnapshotSave must throw - * immediately. 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 snapshot attempt will succeed after compaction completes. + * 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() { @@ -59,18 +63,16 @@ public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); - AtomicInteger doingState = new AtomicInteger(BusinessHandler.doing); - when(mockEngine.getGroupId()).thenReturn(0); when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); - when(mockBusinessHandler.getState(0)).thenReturn(doingState); + when(mockBusinessHandler.tryLockCompactionRange(0)).thenReturn(false); SnapshotHandler handler = new SnapshotHandler(mockEngine); SnapshotWriter stubWriter = stubWriter("/tmp/snapshot"); HgStoreException ex = assertThrows( - "onSnapshotSave must throw when state == doing", + "onSnapshotSave must throw when the compaction-range lock is held", HgStoreException.class, () -> handler.onSnapshotSave(stubWriter)); @@ -78,11 +80,12 @@ public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { 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 state is NOT doing (e.g. compactionDone or null), onSnapshotSave must not throw. - * It should proceed and call saveSnapshot with concrete path verification. + * When the compaction-range lock is free, onSnapshotSave must reserve it, call saveSnapshot, + * and release the lock afterwards. */ @Test public void testOnSnapshotSaveCallsSaveSnapshotWhenNotBusy() throws Exception { @@ -92,13 +95,10 @@ public void testOnSnapshotSaveCallsSaveSnapshotWhenNotBusy() throws Exception { final String snapshotPath = tmpDir.newFolder("snap-not-busy").getAbsolutePath(); - // state == compactionDone (not doing) — save should proceed immediately - AtomicInteger doneState = new AtomicInteger(BusinessHandler.compactionDone); - when(mockEngine.getGroupId()).thenReturn(0); when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); - when(mockBusinessHandler.getState(0)).thenReturn(doneState); + when(mockBusinessHandler.tryLockCompactionRange(0)).thenReturn(true); SnapshotHandler handler = new SnapshotHandler(mockEngine); SnapshotWriter stubWriter = stubWriter(snapshotPath); @@ -111,6 +111,38 @@ public void testOnSnapshotSaveCallsSaveSnapshotWhenNotBusy() throws Exception { 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) { @@ -127,4 +159,17 @@ private static SnapshotWriter stubWriter(String path) { @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..3721000ec5 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,15 @@ package org.apache.hugegraph.store.raftcore; +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 }) public class RaftSuiteTest { From 98fdaac6bc30baf87b1e06209eb1fe167171a0b7 Mon Sep 17 00:00:00 2001 From: Vaibhav Joshi Date: Sat, 5 Sep 2026 14:48:30 +0530 Subject: [PATCH 6/7] fix(dist): update JaCoCo self-test contract for store-core-test profile The pd-store-ci.yml store job gained a store-core-test profile and hg-store-core module in a prior commit, but test-check-jacoco-report.sh's hardcoded aggregation contract still asserted the old 4-profile set, breaking CI with an AssertionError on the set-equality checks. --- .../src/assembly/travis/test-check-jacoco-report.sh | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) 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") From 468e234396c9ea61298478209721d6f4f920de42 Mon Sep 17 00:00:00 2001 From: Vaibhav Joshi Date: Fri, 11 Sep 2026 15:45:08 +0530 Subject: [PATCH 7/7] fix(store): address review feedback on snapshot save/load races (#3164) - report EBUSY instead of EIO when a snapshot save is skipped due to an in-progress compaction, so jRaft retries later instead of escalating to a full raft node restart (only EIO triggers that in SnapshotExecutorImpl#onSnapshotSaveDone) - check should_not_load before validating the data/ directory in onSnapshotLoad, so a locally-saved snapshot (which has no data/ by design) is skipped instead of reported as corrupt - keep the raftRocksdbConfigRegistered guard flag unset until registration actually completes, so a failure partway through can be retried instead of being silently swallowed forever - restore EC_RKDB_TRUNCATE_FAIL, EC_RKDB_TRANSFER_SNAPSHOT_FAIL, and EC_METRIC_FAIL, which were unintentionally dropped and would have broken binary compatibility for downstream consumers - stop CoreSuiteTest and BatchGraphIsolationTest from sharing a surefire fork: HgStoreEngine's `closing` flag is set by the former's teardown and never reset, so the latter failed with "store is closing" whenever both ran in the same JVM Updates HgSnapshotHandlerTest's should_not_load/data-missing case to expect a skip rather than a throw, matching the corrected check order. --- .../store/options/RaftRocksdbOptions.java | 76 ++++++++++--------- .../store/raft/PartitionStateMachine.java | 8 +- .../store/snapshot/SnapshotHandler.java | 12 +-- .../store/util/HgStoreException.java | 3 + hugegraph-store/hg-store-test/pom.xml | 6 ++ .../core/snapshot/HgSnapshotHandlerTest.java | 16 ++-- 6 files changed, 68 insertions(+), 53 deletions(-) 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 7287d44fbd..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 @@ -59,48 +59,52 @@ private static void registerRaftRocksdbConfig(HugeConfig options) { // 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; } - 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); } 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 86c236fb97..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 @@ -176,6 +176,12 @@ public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) thr final String snapshotDir = reader.getPath(); final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; + // No need to load locally saved snapshots + if (shouldNotLoad(reader)) { + log.info("skip to load snapshot because of should_not_load flag"); + return; + } + if (!new File(graphSnapshotDir).isDirectory()) { throw new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, String.format( @@ -184,12 +190,6 @@ public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) thr graphSnapshotDir)); } - // No need to load locally saved snapshots - if (shouldNotLoad(reader)) { - log.info("skip to load snapshot because of should_not_load flag"); - return; - } - // Use snapshot directly log.info("Raft {} begin loadSnapshot, {}", partitionEngine.getGroupId(), graphSnapshotDir); businessHandler.loadSnapshot(graphSnapshotDir, "", partitionEngine.getGroupId(), 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 38c6c49904..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 @@ -33,9 +33,12 @@ public class HgStoreException extends RuntimeException { public static final int EC_RKDB_DOMERGE_FAIL = 1207; public static final int EC_RKDB_DOGET_FAIL = 1208; public static final int EC_RKDB_PD_FAIL = 1209; + public static final int EC_RKDB_TRUNCATE_FAIL = 1212; 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/snapshot/HgSnapshotHandlerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java index 54f890bd5c..b60cdfd303 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,7 +18,6 @@ package org.apache.hugegraph.store.core.snapshot; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertThrows; import java.io.File; import java.io.IOException; @@ -35,7 +34,6 @@ import org.apache.hugegraph.store.meta.Partition; 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; @@ -197,10 +195,12 @@ public void testFindFileList() { } /** - * Test that onSnapshotLoad validates corruption when should_not_load is present but data/ missing. + * 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 testOnSnapshotLoadThrowsWhenShouldNotLoadPresentButDataMissing() + public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentButDataMissing() throws Exception { // Arrange: snapshot dir has should_not_load but NO data/ subdirectory. File snapDir = tmpDir.newFolder("snapshot-corrupt"); @@ -211,12 +211,8 @@ public void testOnSnapshotLoadThrowsWhenShouldNotLoadPresentButDataMissing() SnapshotHandler handler = new SnapshotHandler(createPartitionEngine(1)); SnapshotReader stubReader = stubReader(snapDir.getAbsolutePath()); - // The missing data/ dir is checked before should_not_load, so this throws - // immediately rather than falling through to businessHandler.loadSnapshot. - assertThrows( - "onSnapshotLoad must throw when should_not_load present but data/ missing", - HgStoreException.class, - () -> handler.onSnapshotLoad(stubReader, 0L)); + // Must not throw; should return early at the should_not_load check. + handler.onSnapshotLoad(stubReader, 0L); } /**