bitflicker64 commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r4125584338
########## hugegraph-store/hg-store-common/src/main/java/org/apache/hugegraph/store/cloud/CloudStorageProviderFactory.java: ########## @@ -0,0 +1,220 @@ +/* + * 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.cloud; + +import java.io.IOException; +import java.util.Map; +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; + +import lombok.Getter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Factory for {@link CloudStorageProvider} instances. + * + * <p>Providers are discovered at class-loading time via {@link ServiceLoader}. + * Any JAR that includes + * {@code META-INF/services/org.apache.hugegraph.store.cloud.CloudStorageProvider} + * is automatically picked up when it is present on the classpath. + * + * <p>Usage: + * <pre> + * CloudStorageConfig cfg = ...; // populated from application.yml + * CloudStorageProvider provider = CloudStorageProviderFactory.initialize(cfg); + * // later: + * CloudStorageProvider active = CloudStorageProviderFactory.getActiveProvider(); + * </pre> + */ +public final class CloudStorageProviderFactory { + + private static final Logger log = LoggerFactory.getLogger(CloudStorageProviderFactory.class); + + /** All discovered providers keyed by {@link CloudStorageProvider#providerName()}. */ + private static final Map<String, CloudStorageProvider> REGISTRY = new ConcurrentHashMap<>(); + + /** The currently active (initialized) provider; null when disabled or not yet initialized. + * -- GETTER -- + * Returns the currently active provider, or + * if cloud storage is + * disabled or + * has not yet been called. + */ + @Getter + private static volatile CloudStorageProvider activeProvider; Review Comment: Important: should not exist. The provider is set once in `@PostConstruct` and never reconfigured, yet it lives in a mutable global. The listener and retry queue call `getActiveProvider()` at 18 sites with 16 `provider == null` branches. Those branches feed a second "provider unavailable" retry budget: `DEFAULT_MAX_PROVIDER_UNAVAILABLE_RETRIES`, `computeProviderUnavailableDelay` and two `*ForTest` setters in `CloudUploadRetryQueue`, plus `TRUNCATE_PROVIDER_UNAVAILABLE_MAX_DELAY_MS` in the listener. `AppConfig` returns before `initialize` when cloud storage is disabled, so the disabled branch here never runs. `reset()` and `setActiveProviderForTest()` are test hooks in main code. Keep the SPI (#3079 asks for it). Resolve the provider once and pass it to both constructors as a `final` field: ```java CloudStorageProvider provider = ServiceLoader.load(CloudStorageProvider.class).stream() .map(ServiceLoader.Provider::get) .filter(p -> p.providerName().equals(cfg.getProvider())) .findFirst() .orElseThrow(() -> new IllegalArgumentException( "No cloud storage provider '" + cfg.getProvider() + "' on the classpath")); provider.init(cfg); ``` That deletes the global, the null branches, the second retry budget and the test hooks. ########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudUploadRetryQueue.java: ########## @@ -0,0 +1,1159 @@ +/* + * 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.node.cloud; + +import java.io.BufferedReader; +import java.io.BufferedWriter; +import java.io.Closeable; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.hugegraph.store.cloud.CloudStorageProvider; +import org.apache.hugegraph.store.cloud.CloudStorageNonRetryableException; +import org.apache.hugegraph.store.cloud.CloudStorageProviderFactory; + +import lombok.Getter; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; + +/** + * Asynchronous retry queue with a file-backed dead-letter queue (DLQ) for failed + * cloud SST uploads. + * + * <h3>Retry strategy</h3> + * <p>The number of whole-file retries is configured by the caller via + * {@code cloud.storage.upload-retry-max-attempts}, which defaults to {@code 3} (whole-file retries + * enabled). {@code maxAttempts == 0} is an opt-in mode — not the shipped default — for deployments + * whose provider already performs sufficient internal retries (e.g. the S3 provider's underlying + * AWS SDK client, which retries transient per-request/per-part failures on its own). + * <ul> + * <li>When {@code maxAttempts == 0}, failures go directly to the DLQ with no whole-file retry.</li> + * <li>When {@code maxAttempts > 0} (the default), the first retry is scheduled after + * {@code initialDelayMs} milliseconds.</li> + * <li>Subsequent retries use exponential backoff: {@code delay = initialDelayMs * 2^(attempt-1)}, + * capped at {@code maxDelayMs}.</li> + * <li>After {@code maxAttempts} consecutive failures the task is moved to the DLQ.</li> + * <li>If the local SST file no longer exists at retry time (compacted away by RocksDB) + * the retry is silently dropped — there is nothing to upload.</li> + * </ul> + * + * <h3>Dead-letter queue (DLQ)</h3> + * <ul> + * <li>In-memory: {@link #getDlqEntries()} returns a snapshot.</li> + * <li>On-disk: entries are appended to {@code <dataRoot>/.cloud-upload-dlq.tsv} in a + * tab-separated format so they survive JVM restarts. Existing entries are loaded + * back into memory at construction time.</li> + * <li>{@link #replayDlq()} retries all DLQ entries immediately; successful uploads are + * removed and the file is rewritten. Failed entries are re-submitted to the retry + * queue with a fresh attempt cycle.</li> + * </ul> + * + * <h3>Thread safety</h3> + * All public methods are thread-safe. The in-memory DLQ is a {@link ConcurrentLinkedDeque}; + * disk I/O is serialised onto the internal {@link ScheduledExecutorService} (for appends) + * or synchronised explicitly (for rewrites). + * + * <h3>Lifecycle</h3> + * Call {@link #close()} (e.g. from a Spring {@code @PreDestroy} method) to shut down the + * background executor gracefully. + */ +@Slf4j +public class CloudUploadRetryQueue implements Closeable { + + // ----------------------------------------------------------------------- + // Constants + // ----------------------------------------------------------------------- + + /** Name of the on-disk DLQ file, relative to {@code dataRoot}. */ + static final String DLQ_FILE_NAME = ".cloud-upload-dlq.tsv"; + + private static final String DLQ_COMMENT = + "# Cloud SST upload dead-letter queue – tab-separated: " + + "failedAt\\tattemptCount\\tdbName\\tcfName\\tfilePath\\tremoteKey\\tlastError" + + "\\tsourceSstPath\\tuploadEpoch"; + + /** Max provider-unavailable reschedules before surfacing a terminal local-only DLQ signal. */ + private static final int DEFAULT_MAX_PROVIDER_UNAVAILABLE_RETRIES = 256; + /** Backoff cap for provider-unavailable reschedules (ms), independent of upload-attempt cap. */ + private static final long DEFAULT_PROVIDER_UNAVAILABLE_MAX_DELAY_MS = 30_000L; + + // ----------------------------------------------------------------------- + // Configuration + // ----------------------------------------------------------------------- + + private final int maxAttempts; + private final long initialDelayMs; + private final long maxDelayMs; + private volatile int maxProviderUnavailableRetries = + DEFAULT_MAX_PROVIDER_UNAVAILABLE_RETRIES; + private volatile long providerUnavailableMaxDelayMs = + DEFAULT_PROVIDER_UNAVAILABLE_MAX_DELAY_MS; + + /** + * Invoked with {@code (dbName, sourceSstPath, uploadEpoch)} whenever an upload succeeds via + * retry or DLQ replay, so the caller can mark the file confirmed-present in cloud. + * The {@code uploadEpoch} is the {@link CloudSyncTracker} epoch captured at submission time; + * the callback should use {@link CloudSyncTracker#markConfirmedIfEpoch} to drop stale callbacks. + */ + @FunctionalInterface + public interface UploadConfirmedCallback { + void onConfirmed(String dbName, String sourceSstPath, long uploadEpoch); + } + + private final UploadConfirmedCallback onUploadConfirmed; + + /** + * Invoked with {@code dbName} after an upload becomes durable via retry or DLQ replay, so the + * caller can (debounced) publish CURRENT/MANIFEST for that DB. Without this, a retry that + * succeeds during a quiet period would leave the SST in cloud but the mirrored CURRENT stale + * until some unrelated sync fires (or never, on an idle DB) — widening the recovery point. + */ + @FunctionalInterface + public interface MetadataSyncTrigger { + void onUploadDurable(String dbName); + } + + /** Optional; set via {@link #setMetadataSyncTrigger} after the listener is constructed. */ + @Setter + private volatile MetadataSyncTrigger metadataSyncTrigger; + + // ----------------------------------------------------------------------- + // State + // ----------------------------------------------------------------------- + + /** In-memory dead-letter queue – holds tasks that exhausted all retries. */ + private final Deque<FailedUploadTask> dlq = new ConcurrentLinkedDeque<>(); + + /** + * Retries that are scheduled but have not started running yet. A scheduled task is added here + * before {@code scheduler.schedule(...)} and removed the moment it starts executing. On + * {@link #close()}, a forced {@code shutdownNow()} silently discards the executor's queued + * (delayed) tasks — but those wrappers are opaque, so we cannot recover their payload from the + * returned runnables. Tracking the payload here lets us move any retry that never ran into the + * DLQ, so retry intent (the only remaining upload path for a compaction-raced file) is not lost. + */ + private final Set<RetryContext> pendingRetries = ConcurrentHashMap.newKeySet(); + + /** + * Retries that have STARTED executing but not yet finished. A task moves from + * {@link #pendingRetries} to this set when it begins running, and is removed when it completes + * (success, DLQ, or reschedule). On {@link #close()}, a retry still here after the shutdown + * timeout is one hung in {@code provider.uploadFile(...)}; it is persisted to the DLQ so its + * upload intent survives even a forced shutdown. DLQ replay is idempotent (idempotent PUT + + * epoch-guarded confirmation), so a duplicate entry for a retry that actually completed is safe. + */ + private final Set<RetryContext> inFlightRetries = ConcurrentHashMap.newKeySet(); + + /** + * Upper bound on DLQ entries (in memory and, amortized, on disk). Without a cap, a prolonged + * provider outage with {@code maxAttempts == 0} routes every failed SST upload straight to the + * DLQ, growing process memory and the on-disk file without bound (OOM / disk exhaustion). + * When the cap is exceeded the oldest entries are evicted; those files remain recoverable via + * the delete guard and startup SST backfill, which re-upload any live SST missing from cloud. + */ + static final int DEFAULT_MAX_DLQ_SIZE = 100_000; + private volatile int maxDlqSize = DEFAULT_MAX_DLQ_SIZE; + + /** Count of DLQ entries evicted due to the size cap (monitoring / tests). */ + private final AtomicInteger droppedDlqEntries = new AtomicInteger(0); + + /** Appends since the last on-disk compaction, used to amortize {@link #rewriteDlqFile}. */ + private final AtomicInteger appendsSinceRewrite = new AtomicInteger(0); + + /** + * DLQ on-disk persistence health. The in-memory DLQ is authoritative at runtime, but the + * on-disk file is what survives a process restart. If a persist (append/rewrite) fails, retry + * intent for the failed entry would be lost on a crash before the next successful persist. + * We flip this to {@code false} and increment {@link #dlqPersistenceFailures} so operators can + * alert on a "degraded durability" state via {@link #isDlqPersistenceHealthy()} rather than + * treating a silently-swallowed IO error as healthy. It recovers to {@code true} on the next + * successful persist. Live SSTs remain recoverable via the delete guard and startup backfill, + * so this is a degraded signal, not hard data loss. + */ + @Getter + private volatile boolean dlqPersistenceHealthy = true; + + /** Cumulative count of DLQ persistence failures (monitoring / tests). */ + private final AtomicInteger dlqPersistenceFailures = new AtomicInteger(0); + + /** Absolute path of the on-disk DLQ file. */ + private final Path dlqFile; + + /** Background thread pool used to schedule retries. */ + private final ScheduledExecutorService scheduler; + + /** Counter of in-flight retry tasks (for testing / monitoring). */ + private final AtomicInteger inFlightCount = new AtomicInteger(0); + + /** Serialises all DLQ file mutations (append and rewrite) across scheduler threads. */ + private final ReentrantLock dlqFileLock = new ReentrantLock(); + + // ----------------------------------------------------------------------- + // Constructor + // ----------------------------------------------------------------------- + + /** + * Fully-configured constructor. Prefer this over the {@link #setMaxDlqSize(int)} setter when the + * cap is known at construction time: setting it here applies the cap <em>before</em> the on-disk + * DLQ is loaded, so a large persisted file is bounded to the configured cap during the load + * itself — avoiding the post-load trim-and-rewrite the setter must perform. + * + * @param maxAttempts maximum whole-file upload retries after a first failure. + * {@code 0} means no retries – failures go directly to DLQ + * (provider is expected to handle its own retries internally). + * Positive values enable exponential-backoff whole-file retries. + * @param initialDelayMs delay before the first retry, in milliseconds; clamped to ≥ 100. + * Ignored when {@code maxAttempts == 0}. + * @param maxDelayMs upper bound for exponential backoff delay; clamped to + * ≥ {@code initialDelayMs}. Ignored when {@code maxAttempts == 0}. + * @param dataRoot absolute path of the store's data directory – the DLQ file is + * written here + * @param onUploadConfirmed epoch-aware callback invoked on every successful retry/DLQ-replay + * upload; pass {@code null} for none. + * @param maxDlqSize maximum number of DLQ entries retained before the oldest are evicted; + * a non-positive value falls back to {@link #DEFAULT_MAX_DLQ_SIZE} so a + * misconfiguration cannot turn the DLQ into a zero-capacity queue. + */ + public CloudUploadRetryQueue(int maxAttempts, long initialDelayMs, + long maxDelayMs, String dataRoot, + UploadConfirmedCallback onUploadConfirmed, int maxDlqSize) { + this.maxAttempts = Math.max(0, maxAttempts); + this.initialDelayMs = Math.max(100L, initialDelayMs); + this.maxDelayMs = Math.max(this.initialDelayMs, maxDelayMs); + this.onUploadConfirmed = onUploadConfirmed != null + ? onUploadConfirmed + : (db, path, epoch) -> { }; + this.dlqFile = Paths.get(dataRoot, DLQ_FILE_NAME); + // Apply the cap before loadDlqFromDisk() so the load bounds the in-memory DLQ (and the + // rewritten on-disk file) to the configured value directly. A non-positive value keeps the + // default cap set at field initialisation, matching setMaxDlqSize()'s guard. + if (maxDlqSize <= 0) { + log.warn("Ignoring non-positive DLQ max size {}; keeping default {}", + maxDlqSize, this.maxDlqSize); + } else { + this.maxDlqSize = maxDlqSize; + } + + java.util.concurrent.ScheduledThreadPoolExecutor sched = + new java.util.concurrent.ScheduledThreadPoolExecutor(2, r -> { + Thread t = new Thread(r, "cloud-upload-retry"); + t.setDaemon(true); + return t; + }); + // On shutdown(), cancel not-yet-due delayed retries instead of holding the executor open + // until their (possibly minutes-long) delay elapses. close() then rescues these pending + // retries into the DLQ via drainPendingRetriesToDlq(), so their upload intent is preserved + // without a slow, pointless wait for retries that will never get to run. + sched.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + this.scheduler = sched; + + loadDlqFromDisk(); + } + + // ----------------------------------------------------------------------- + // Public API + // ----------------------------------------------------------------------- + + /** + * Submits a failed upload to the retry queue or directly to the DLQ. + * + * <p>If {@code maxAttempts == 0} (opt-in mode, not the shipped default), the task goes directly + * to the DLQ — the provider is expected to handle its own retries internally. + * + * <p>If {@code maxAttempts > 0} (the default is {@code 3}), the first retry is scheduled after + * {@code initialDelayMs} ms with exponential backoff. After all attempts are exhausted the task + * is moved to the DLQ. + * + * <p>If the file no longer exists at retry time (compacted away by RocksDB) the task is + * silently dropped. + * + * @param dbName RocksDB instance name (partition id) + * @param cfName column-family name + * @param filePath absolute local path of the SST file (also used for confirmation) + * @param remoteKey remote object key (relative to bucket root) + * @param cause the exception from the first upload attempt + */ + public void submit(String dbName, String cfName, String filePath, + String remoteKey, Exception cause) { + submitPinned(dbName, cfName, filePath, filePath, remoteKey, 0L, cause); + } + + /** + * Epoch-aware variant of {@link #submit}. Callers that use an epoch-guarded confirmation + * callback ({@link CloudSyncTracker#markConfirmedIfEpoch}) MUST use this overload and pass the + * epoch captured before the upload was initiated. The plain {@link #submit} hardcodes epoch + * {@code 0}, which never matches a live {@link CloudSyncTracker} epoch (they start at 1), so a + * retried upload confirmed through it would be silently dropped. + * + * @param uploadEpoch the {@link CloudSyncTracker} epoch captured at upload-initiation time + */ + public void submit(String dbName, String cfName, String filePath, + String remoteKey, long uploadEpoch, Exception cause) { + submitPinned(dbName, cfName, filePath, filePath, remoteKey, uploadEpoch, cause); + } + + /** + * Submits a failed upload where the upload file is a hard-linked staging copy that + * differs from the original source SST. + * + * <p>The retry will upload from {@code pinnedPath} (the staging hard-link). On success, + * {@code onUploadConfirmed} is called with {@code sourceSstPath} (the original {@code *.sst} + * path), and the staging file at {@code pinnedPath} is deleted. + * + * @param dbName RocksDB instance name (partition id) + * @param cfName column-family name + * @param pinnedPath path of the staging hard-link to upload from + * @param sourceSstPath original {@code *.sst} path used for confirmation and cleanup + * @param remoteKey remote object key (relative to bucket root) + * @param cause the exception from the first upload attempt + */ + public void submitPinned(String dbName, String cfName, String pinnedPath, + String sourceSstPath, String remoteKey, Exception cause) { + submitPinned(dbName, cfName, pinnedPath, sourceSstPath, remoteKey, 0L, cause); + } + + /** + * Epoch-aware variant. The {@code uploadEpoch} is passed through to the + * {@link UploadConfirmedCallback} so that stale callbacks after DB recreation are discarded. + */ + public void submitPinned(String dbName, String cfName, String pinnedPath, + String sourceSstPath, String remoteKey, + long uploadEpoch, Exception cause) { + if (cause instanceof CloudStorageNonRetryableException || maxAttempts == 0) { + moveToDlq(dbName, cfName, pinnedPath, sourceSstPath, remoteKey, uploadEpoch, 0, + cause.getMessage() != null ? cause.getMessage() : "non-retryable"); + return; + } + // This method is invoked from RocksDB flush/compaction callbacks (via + // CloudStorageEventListener). It must never throw back across that JNI/event boundary. + // When the scheduler is shutting down, scheduler.schedule(...) throws + // RejectedExecutionException — in that race we persist the task directly to the DLQ + // (best-effort disk append) so the retry intent is not lost, and return normally. + try { + scheduleRetry(dbName, cfName, pinnedPath, sourceSstPath, remoteKey, uploadEpoch, + 1, 0); + } catch (RejectedExecutionException e) { + log.warn("Cloud upload retry scheduling rejected (scheduler shutting down); " + + "persisting directly to DLQ: db={}, cf={}, path={}", + dbName, cfName, pinnedPath); + moveToDlq(dbName, cfName, pinnedPath, sourceSstPath, remoteKey, uploadEpoch, 1, + "retry scheduling rejected during shutdown: " + + (cause.getMessage() != null ? cause.getMessage() : cause.toString())); + } + } + + /** + * Returns an unmodifiable snapshot of all current DLQ entries. + */ + public List<FailedUploadTask> getDlqEntries() { + return List.copyOf(dlq); + } + + /** + * Returns the number of entries currently in the DLQ. + */ + public int getDlqSize() { + return dlq.size(); + } + + /** + * Returns the number of DLQ entries evicted because the DLQ size cap was exceeded (e.g. during + * a prolonged provider outage). A non-zero value indicates local-only files whose DLQ record + * was dropped; they remain recoverable via the delete guard and startup SST backfill. + */ + public int getDroppedDlqCount() { + return droppedDlqEntries.get(); + } + + /** Cumulative number of DLQ on-disk persistence failures since startup (monitoring / tests). */ + public int getDlqPersistenceFailureCount() { + return dlqPersistenceFailures.get(); + } + + /** Fires the metadata-sync trigger (if wired), swallowing any error — the upload is durable. */ + private void triggerMetadataSync(String dbName) { + MetadataSyncTrigger trigger = this.metadataSyncTrigger; + if (trigger == null) { + return; + } + try { + trigger.onUploadDurable(dbName); + } catch (Exception e) { + log.warn("Post-upload metadata-sync trigger failed for db={} (upload is durable, " + + "CURRENT/MANIFEST will catch up on the next sync): {}", dbName, + e.getMessage()); + } + } + + /** + * Sets the maximum number of DLQ entries retained before the oldest are evicted. A + * non-positive value is ignored (keeps the current cap) so a misconfiguration cannot turn the + * DLQ into a zero-capacity queue that drops every entry. + */ + public void setMaxDlqSize(int maxSize) { + if (maxSize <= 0) { + log.warn("Ignoring non-positive DLQ max size {}; keeping {}", maxSize, this.maxDlqSize); + return; + } + this.maxDlqSize = maxSize; + // The startup load bounds the in-memory DLQ to the DEFAULT cap (the configured value is not + // known during construction). If the operator configured a SMALLER cap, apply it now so a + // large persisted file loaded at startup is trimmed to the configured bound and the on-disk + // file is rewritten to match — otherwise up to DEFAULT_MAX_DLQ_SIZE stale entries could + // linger in memory and on disk despite a tighter configuration. + int evicted = 0; + while (dlq.size() > maxSize) { + if (dlq.pollFirst() == null) { + break; + } + evicted++; + } + if (evicted > 0) { + int total = droppedDlqEntries.addAndGet(evicted); + rewriteDlqFile(); + log.warn("DLQ: max size lowered to {} — evicted {} oldest entry(ies) to honor the " + + "configured cap (total dropped={}).", maxSize, evicted, total); + } + } + + /** Test seam: cap provider-unavailable reschedules so terminal behavior is deterministic. */ + @SuppressWarnings("SameParameterValue") + void setMaxProviderUnavailableRetriesForTest(int retries) { + this.maxProviderUnavailableRetries = Math.max(1, retries); + } + + /** Test seam: speed up provider-unavailable retry cadence for deterministic timing tests. */ + @SuppressWarnings("SameParameterValue") + void setProviderUnavailableMaxDelayMsForTest(long maxDelayMs) { + this.providerUnavailableMaxDelayMs = Math.max(10L, maxDelayMs); + } + + /** + * Returns the number of retry tasks currently scheduled or executing. + */ + public int getInFlightCount() { + return inFlightCount.get(); + } + + /** + * Replays all DLQ entries immediately by attempting to re-upload each file. + * + * <ul> + * <li>Entries that succeed are removed from the in-memory DLQ and the DLQ file is + * rewritten without them.</li> + * <li>Entries whose local file no longer exists are silently dropped.</li> + * <li>Entries that fail are re-submitted to the retry queue with a fresh attempt cycle + * (they are removed from the DLQ for now and will re-enter it only if all retries + * fail again).</li> + * </ul> + */ + public void replayDlq() { Review Comment: Important: should not exist. `replayDlq()` has no caller outside `CloudUploadRetryQueueTest`: no endpoint, no scheduled call, no startup call. `loadDlqFromDisk()` reads `.cloud-upload-dlq.tsv` at startup and logs "call replayDlq() to retry them", but no operator path can do that. The loaded entries sit in memory unused. Recovery already happens without the file: `createGraphDB` fires `onDBCreated` on every open, and `uploadExistingSstFiles` uploads every live `*.sst` missing from the bucket. The PR's docs say evicted entries "stay recoverable via the delete guard and startup SST backfill". That leaves about 380 lines writing a file nobody reads: - `loadDlqFromDisk`, `appendDlqEntryToDisk`, `rewriteDlqFile`, `fsyncDir` - `serialize`/`deserialize`/`escape`/`unescape` - `dlqFileLock` - the `dlqPersistenceHealthy` flag, gauge and `AppConfig` binding - this method Requested change: keep the in-memory DLQ count for visibility and delete the persistence and replay. If replay is wanted, call `replayDlq()` once after `loadDlqFromDisk()` in the constructor so the file has a reader. ########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudStorageEventListener.java: ########## @@ -0,0 +1,2913 @@ +/* + * 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.node.cloud; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Stream; + +import lombok.Getter; + +import lombok.Setter; + +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.LiveSstFile; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.MetadataSnapshot; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.RocksdbChangedListener; +import org.apache.hugegraph.rocksdb.access.RocksDBSession; +import org.apache.hugegraph.store.cloud.CloudStorageProvider; +import org.apache.hugegraph.store.cloud.CloudStorageProviderFactory; + +import lombok.extern.slf4j.Slf4j; + +/** + * {@link RocksdbChangedListener} that bridges RocksDB table-file lifecycle events + * to the active {@link CloudStorageProvider}. + * + * <p>When cloud storage is enabled: + * <ul> + * <li>{@link #onDBCreated} uploads any SST files that already exist in the DB directory + * (e.g. surviving from a previous run), triggers a non-blocking MemTable flush so + * WAL-recovered or recently-written data is written to SST files (completion is signalled + * event-driven via {@link #onTableFileCreated}), then mirrors metadata inline.</li> + * <li>{@link #onTableFileCreated} pins newly created SST files via a hard link, dispatches + * upload work to a bounded background executor, and mirrors metadata after upload completes. + * If the hard link fails (e.g. cross-device mount, filesystem limits), the upload is routed + * directly to the retry queue using the original SST path — no copy is made, so no extra + * disk space is consumed.</li> + * <li>{@link #onTableFileDeleted} mirrors metadata first, then removes the superseded SST object.</li> + * </ul> + * + * <h3>Remote key construction</h3> + * The remote key is derived by stripping the {@code dataRoot} prefix from the absolute + * local file path. This keeps the object layout clean and independent of the container + * filesystem layout: + * <pre> + * dataRoot = /hugegraph-store/storage + * filePath = /hugegraph-store/storage/hgstore-metadata/000008.sst + * remoteKey = store-127.0.0.1_8501/hgstore-metadata/000008.sst + * (with path-prefix "hugegraph") → hugegraph/store-127.0.0.1_8501/hgstore-metadata/000008.sst + * </pre> + * + * This listener is registered with {@link RocksDBFactory} during application startup + * (see {@link org.apache.hugegraph.store.node.AppConfig}). + */ +@Slf4j +public class CloudStorageEventListener implements RocksdbChangedListener { + + /** Absolute, normalised paths of configured store data roots (one per configured partition root). */ + private final String primaryDataRoot; + private final List<String> allDataRoots; + + /** Optional per-store namespace prefix prepended to every remote cloud key. */ + private final String storeScopePrefix; + + static final long DEFAULT_READ_MISS_GUARD_WINDOW_MS = 3000L; + + private final boolean startupHydrationEnabled; + private final long readMissGuardWindowMs; + private final Map<String, Long> readMissAttemptTs; + + /** + * Tracks which {@code (dbName, remoteKey)} pairs are currently being restored by + * {@link #restoreMissingLiveFiles} so a thundering herd of concurrent read misses on the + * same cold SST does not download the same object many times over. This is a best-effort + * de-duplication; correctness against concurrent restores of the same file is guaranteed by + * the per-attempt unique temp name plus an atomic replace-move, not by this set. + */ + private final Set<String> inFlightRestores = ConcurrentHashMap.newKeySet(); + + /** + * Optional retry queue; when non-null, upload failures are submitted here instead + * of just being logged. When null, failures are only logged (no retry). + */ + private final CloudUploadRetryQueue retryQueue; + + /** Tracks which SST files are confirmed present in cloud (per-DB confirmed-file-number set). */ + private final CloudSyncTracker syncTracker; + + /** + * Hard health flag for pending-delete marker durability. Flips to {@code false} when a marker + * cannot be durably persisted (write + fsync) — a state in which a DB delete during a + * provider-unavailable window may be unguarded against re-hydration after a crash — and back to + * {@code true} once a marker is durably persisted again. Surfaced via + * {@link CloudStorageMetricsConst#DELETE_MARKER_HEALTHY}. {@code volatile}: written from delete + * callbacks, read from metric-scrape threads. + * -- GETTER -- + * Whether pending-delete marker persistence is currently healthy. + * indicates a + * marker could not be durably written, so a delete during a provider-unavailable window may not + * survive a crash as a hydration guard. Bound to + * <p> + * . + + */ + @Getter + private volatile boolean deleteMarkerHealthy = true; + + /** Directory fsync is not supported on Windows directory handles. */ + private static final boolean WINDOWS_OS = + System.getProperty("os.name", "").toLowerCase().contains("win"); + + /** Bounded async upload dispatcher so RocksDB callbacks return quickly. */ + private static final int ASYNC_UPLOAD_THREADS = 2; + private static final int ASYNC_UPLOAD_QUEUE_CAPACITY = 256; + /** + * Backing field for {@link #sharedUploadExecutor()}. Not {@code final}: it is recreated on + * demand after {@link #shutdownSharedUploadExecutor} clears it, so a Spring context restart in + * the same JVM binds new listeners to a live executor instead of a TERMINATED one (which would + * reject every upload via AbortPolicy and silently divert all SSTs to the DLQ). + */ + private static ThreadPoolExecutor sharedUploadExecutor; Review Comment: Important: should not exist. Production builds one listener, but the upload pool and metadata-sync scheduler are `static`, with lazy re-creation in `sharedUploadExecutor()` / `metadataSyncScheduler()`. A static `uploadSubsystemShuttingDown` gate stops the scheduler from coming back during drain. Because the gate lets `metadataSyncScheduler()` return `null`, four call sites carry a `scheduler == null` branch (L899, L1165, L1796, L2114), and `requestDebouncedMetadataSync` checks the gate again. The stated reason is "a Spring context restart in the same JVM", which hg-store-node does not do. Requested change: create both executors as `private final` instance fields in the constructor. Add a `close()` that calls `uploadExecutor.shutdown()`, `awaitTermination(...)` and then `scheduler.shutdownNow()`, and call it from `AppConfig.onDestroy()`. A schedule after shutdown throws `RejectedExecutionException`, which every call site already catches. The two static fields, both lazy getters, the gate and the four null branches go. ########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/AppConfig.java: ########## @@ -312,6 +631,100 @@ public class RocksdbConfig { private Map<String, String> rocksdb = new HashMap<>(); } + /** + * Spring {@link ConfigurationProperties} wrapper for the common cloud storage properties. + * + * <p>Provider-specific keys (e.g. {@code cloud.storage.s3.bucket}) are read + * directly from the Spring {@link Environment} at conversion time, so this class + * has zero knowledge of any specific cloud provider. + * + * <pre> + * cloud: + * storage: + * enabled: true + * provider: s3 # selects which sub-namespace to forward to the provider + * path-prefix: hugegraph + * s3: # all keys here are forwarded as-is to the S3 provider + * bucket: my-bucket + * region: us-east-1 + * access-key: ${AWS_ACCESS_KEY_ID} + * secret-key: ${AWS_SECRET_ACCESS_KEY} + * </pre> + */ + @Data + @Configuration + @ConfigurationProperties(prefix = "cloud.storage") + public class CloudStorageSpringConfig { Review Comment: Important: replace or collapse. This class re-declares all 12 fields of `CloudStorageConfig` with the same defaults, and `toCloudStorageConfig()` copies them one setter at a time. A new knob means three edits, and the two sets of defaults can drift. `CloudStorageConfig` is already a Lombok `@Data` bean, so bind it directly: ```java CloudStorageConfig cfg = Binder.get(environment) .bind("cloud.storage", CloudStorageConfig.class) .orElseGet(CloudStorageConfig::new); cfg.setProviderProperties(readProviderProperties(cfg.getProvider())); ``` That deletes this inner `@Configuration` class and the copy method. Relaxed binding (`CLOUD_STORAGE_ENABLED` and so on) still works. ########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudStorageEventListener.java: ########## @@ -0,0 +1,2913 @@ +/* + * 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.node.cloud; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Stream; + +import lombok.Getter; + +import lombok.Setter; + +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.LiveSstFile; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.MetadataSnapshot; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.RocksdbChangedListener; +import org.apache.hugegraph.rocksdb.access.RocksDBSession; +import org.apache.hugegraph.store.cloud.CloudStorageProvider; +import org.apache.hugegraph.store.cloud.CloudStorageProviderFactory; + +import lombok.extern.slf4j.Slf4j; + +/** + * {@link RocksdbChangedListener} that bridges RocksDB table-file lifecycle events + * to the active {@link CloudStorageProvider}. + * + * <p>When cloud storage is enabled: + * <ul> + * <li>{@link #onDBCreated} uploads any SST files that already exist in the DB directory + * (e.g. surviving from a previous run), triggers a non-blocking MemTable flush so + * WAL-recovered or recently-written data is written to SST files (completion is signalled + * event-driven via {@link #onTableFileCreated}), then mirrors metadata inline.</li> + * <li>{@link #onTableFileCreated} pins newly created SST files via a hard link, dispatches + * upload work to a bounded background executor, and mirrors metadata after upload completes. + * If the hard link fails (e.g. cross-device mount, filesystem limits), the upload is routed + * directly to the retry queue using the original SST path — no copy is made, so no extra + * disk space is consumed.</li> + * <li>{@link #onTableFileDeleted} mirrors metadata first, then removes the superseded SST object.</li> + * </ul> + * + * <h3>Remote key construction</h3> + * The remote key is derived by stripping the {@code dataRoot} prefix from the absolute + * local file path. This keeps the object layout clean and independent of the container + * filesystem layout: + * <pre> + * dataRoot = /hugegraph-store/storage + * filePath = /hugegraph-store/storage/hgstore-metadata/000008.sst + * remoteKey = store-127.0.0.1_8501/hgstore-metadata/000008.sst + * (with path-prefix "hugegraph") → hugegraph/store-127.0.0.1_8501/hgstore-metadata/000008.sst + * </pre> + * + * This listener is registered with {@link RocksDBFactory} during application startup + * (see {@link org.apache.hugegraph.store.node.AppConfig}). + */ +@Slf4j +public class CloudStorageEventListener implements RocksdbChangedListener { + + /** Absolute, normalised paths of configured store data roots (one per configured partition root). */ + private final String primaryDataRoot; + private final List<String> allDataRoots; + + /** Optional per-store namespace prefix prepended to every remote cloud key. */ + private final String storeScopePrefix; + + static final long DEFAULT_READ_MISS_GUARD_WINDOW_MS = 3000L; + + private final boolean startupHydrationEnabled; + private final long readMissGuardWindowMs; + private final Map<String, Long> readMissAttemptTs; + + /** + * Tracks which {@code (dbName, remoteKey)} pairs are currently being restored by + * {@link #restoreMissingLiveFiles} so a thundering herd of concurrent read misses on the + * same cold SST does not download the same object many times over. This is a best-effort + * de-duplication; correctness against concurrent restores of the same file is guaranteed by + * the per-attempt unique temp name plus an atomic replace-move, not by this set. + */ + private final Set<String> inFlightRestores = ConcurrentHashMap.newKeySet(); + + /** + * Optional retry queue; when non-null, upload failures are submitted here instead + * of just being logged. When null, failures are only logged (no retry). + */ + private final CloudUploadRetryQueue retryQueue; + + /** Tracks which SST files are confirmed present in cloud (per-DB confirmed-file-number set). */ + private final CloudSyncTracker syncTracker; + + /** + * Hard health flag for pending-delete marker durability. Flips to {@code false} when a marker + * cannot be durably persisted (write + fsync) — a state in which a DB delete during a + * provider-unavailable window may be unguarded against re-hydration after a crash — and back to + * {@code true} once a marker is durably persisted again. Surfaced via + * {@link CloudStorageMetricsConst#DELETE_MARKER_HEALTHY}. {@code volatile}: written from delete + * callbacks, read from metric-scrape threads. + * -- GETTER -- + * Whether pending-delete marker persistence is currently healthy. + * indicates a + * marker could not be durably written, so a delete during a provider-unavailable window may not + * survive a crash as a hydration guard. Bound to + * <p> + * . + + */ + @Getter + private volatile boolean deleteMarkerHealthy = true; + + /** Directory fsync is not supported on Windows directory handles. */ + private static final boolean WINDOWS_OS = + System.getProperty("os.name", "").toLowerCase().contains("win"); + + /** Bounded async upload dispatcher so RocksDB callbacks return quickly. */ + private static final int ASYNC_UPLOAD_THREADS = 2; + private static final int ASYNC_UPLOAD_QUEUE_CAPACITY = 256; + /** + * Backing field for {@link #sharedUploadExecutor()}. Not {@code final}: it is recreated on + * demand after {@link #shutdownSharedUploadExecutor} clears it, so a Spring context restart in + * the same JVM binds new listeners to a live executor instead of a TERMINATED one (which would + * reject every upload via AbortPolicy and silently divert all SSTs to the DLQ). + */ + private static ThreadPoolExecutor sharedUploadExecutor; + private final ThreadPoolExecutor uploadExecutor; + + /** + * Shutdown gate for the upload subsystem. Set true at the start of + * {@link #shutdownSharedUploadExecutor} so in-flight upload tasks draining during shutdown do + * NOT schedule new trailing metadata syncs or resurrect the (about-to-be-torn-down) + * {@link #metadataSyncScheduler}. Reset to false when a fresh executor is (re)created for a + * Spring context restart in the same JVM. + */ + private static volatile boolean uploadSubsystemShuttingDown = false; + + /** Returns the shared upload executor, (re)creating it if absent or already shut down. */ + private static synchronized ThreadPoolExecutor sharedUploadExecutor() { + if (sharedUploadExecutor == null || sharedUploadExecutor.isShutdown()) { + // A fresh executor means a new lifecycle (e.g. context restart) — reopen the gate. + uploadSubsystemShuttingDown = false; + sharedUploadExecutor = new ThreadPoolExecutor( + ASYNC_UPLOAD_THREADS, + ASYNC_UPLOAD_THREADS, + 60L, + TimeUnit.SECONDS, + new ArrayBlockingQueue<>(ASYNC_UPLOAD_QUEUE_CAPACITY), + newUploadThreadFactory(), + new ThreadPoolExecutor.AbortPolicy()); + } + return sharedUploadExecutor; + } + + // ----------------------------------------------------------------------- + // Metadata-sync debounce + // ----------------------------------------------------------------------- + // onTableFileCreated fires once per flushed/compacted SST; syncing metadata inline on every + // one triggers a full RocksDB checkpoint + S3 list/prune per SST. These fields coalesce those + // high-frequency syncs into at most one per debounce window per DB, with a trailing sync so + // the final state is always eventually published even if writes stop mid-window. Event-driven + // callers that need an immediate publish (delete guard, onCompacted, onDBCreated) still call + // syncMetadataSnapshotInline directly and are NOT debounced. + + /** Default debounce window for the post-upload metadata sync. */ + static final long DEFAULT_METADATA_SYNC_DEBOUNCE_MS = 1_000L; + + /** + * Single shared scheduler for trailing (deferred) metadata syncs. Not {@code final}: like the + * upload executor it is recreated on demand after {@link #shutdownSharedUploadExecutor} clears + * it, so a Spring context restart in the same JVM gets a live scheduler. Otherwise every + * post-restart trailing sync would be rejected and fall back to an inline publish (a checkpoint + * per SST), defeating debouncing exactly when a burst is most likely. + */ + private static ScheduledExecutorService metadataSyncScheduler; + + /** + * Returns the metadata-sync scheduler, (re)creating it if absent or already shut down. + * + * <p>During upload-subsystem shutdown it must NOT resurrect the scheduler — otherwise an + * in-flight upload task draining after {@link #shutdownSharedUploadExecutor} tore the scheduler + * down would create a brand-new one and schedule a trailing sync that fires after provider + * teardown, leaking the scheduler across a context stop/start. Returns the current field + * (possibly {@code null}) in that case; callers must tolerate a {@code null}. + */ + private static synchronized ScheduledExecutorService metadataSyncScheduler() { + if (uploadSubsystemShuttingDown) { + return metadataSyncScheduler; + } + if (metadataSyncScheduler == null || metadataSyncScheduler.isShutdown()) { + metadataSyncScheduler = Executors.newScheduledThreadPool(1, r -> { + Thread t = new Thread(r, "cloud-metadata-sync"); + t.setDaemon(true); + return t; + }); + } + return metadataSyncScheduler; + } + + /** Effective debounce window; overridable in tests. + * -- SETTER -- + * Sets the debounce window (ms) for the per-SST metadata sync. Values + * disable + * debouncing (publish metadata on every SST upload, the pre-debounce behavior). + */ + @Setter + private volatile long metadataSyncDebounceMs; + + /** + * Backlog bound on the debounce: the time window alone lets an unbounded number of SSTs be + * uploaded-but-not-yet-mirrored during a heavy-ingestion burst, widening the cloud recovery + * point (a crash + local-disk-loss in that window loses the flushed SSTs whose manifest was + * not yet republished). Once this many uploads accumulate without a metadata publish for a DB, + * a publish is forced immediately regardless of the time window, bounding RPO by count as well + * as by time. {@code <= 0} disables the count bound (time-only debounce). + */ + static final int DEFAULT_METADATA_SYNC_MAX_UNPUBLISHED = 32; + /** + * -- SETTER -- + * Sets the maximum number of uploaded-but-unmirrored SSTs tolerated before a metadata publish + * is forced regardless of the debounce window (bounds the cloud recovery point by count during + * heavy-ingestion bursts). Values + * disable the count bound (time-only debounce). + */ + @Setter + private volatile int metadataSyncMaxUnpublished; + + /** Last time a post-upload metadata sync completed, per DB (epoch millis). */ + private final Map<String, Long> lastMetadataSyncMs = new ConcurrentHashMap<>(); + + /** Count of SST uploads confirmed but not yet reflected in a published manifest, per DB. */ + private final Map<String, java.util.concurrent.atomic.AtomicInteger> unpublishedUploads = + new ConcurrentHashMap<>(); + + /** DBs with a trailing metadata sync already scheduled (coalescing guard). */ + private final Set<String> pendingMetadataSync = ConcurrentHashMap.newKeySet(); + + // ----------------------------------------------------------------------- + // Metadata (CURRENT/MANIFEST/OPTIONS) mirroring & consistent restore + // ----------------------------------------------------------------------- + + /** Resolved DB directory -> logical DB name, so {@link #onCompacted} resolves path events. */ + private final Map<String, String> dbNameByDir = new ConcurrentHashMap<>(); + + /** + * Tracks which DBs are currently being truncated. While a DB is in this set, + * metadata sync operations are skipped to allow the purge to complete cleanly + * without new metadata files being re-uploaded. + */ + private final Set<String> truncatingDbs = ConcurrentHashMap.newKeySet(); + + /** + * Tracks the timestamp of recent truncations (DB name -> truncation time in ms). + * Used to suppress metadata syncs for a grace period after truncation, allowing + * pending RocksDB background operations and callbacks to complete without + * re-uploading metadata that was just purged. + */ + private final Map<String, Long> truncationTimes = new ConcurrentHashMap<>(); + + /** Per-DB mutexes to serialize metadata capture/publication/pruning. */ + private final Map<String, Object> metadataSyncLocks = new ConcurrentHashMap<>(); + + /** + * Per-remote-prefix mutexes serializing pending-delete cleanup (inline at open vs. the async + * retry). Ensures the purge runs at most once and only while the marker is present, so it can + * never delete data a re-created DB uploads to the same prefix after cleanup completes. + */ + private final Map<String, Object> pendingDeleteLocks = new ConcurrentHashMap<>(); + + /** Last successfully published RocksDB generation per DB. */ + private final Map<String, Long> lastPublishedMetadataGeneration = new ConcurrentHashMap<>(); + + /** + * Grace period (ms) after truncation during which metadata syncs are suppressed. + * This allows pending RocksDB background callbacks to complete without re-uploading + * metadata that was purged during truncation. + */ + private static final long TRUNCATION_GRACE_PERIOD_MS = 5_000L; + + /** + * Suffix appended to the DB prefix (not inside it) when a database is deleted. + * Placing the tombstone as a sibling of the data prefix means the + * {@link #purgeRemotePrefix} call in {@link #onDBDeleted} cannot accidentally remove it + * while stale SST or metadata objects remain, so {@link #preHydrateDbFiles} can still + * detect the deleted generation and skip hydration. + * + * <p>Example: data prefix = {@code store-host_8500/hugegraph/db}, + * tombstone key = {@code store-host_8500/hugegraph/db_DELETED}. + */ + static final String DB_TOMBSTONE_SUFFIX = "_DELETED"; + + /** + * Fully-parameterised constructor: the listener is completely configured the moment it is + * constructed, so there is no window in which a registered listener can be observed with the + * tuning setters not yet applied. The equivalent {@code set*} methods remain for tests and + * runtime overrides. + * + * @param dataRoots configured store data roots (typically parsed from comma-separated + * {@code app.data-path}) + * @param readMissGuardWindowMs guard window in ms for repeated read-miss hydration attempts + * for the same db/table pair (cloud.storage.read-miss-guard-window-ms) + * @param retryQueue optional {@link CloudUploadRetryQueue}; when non-null, upload failures + * are retried asynchronously and eventually moved to the dead-letter queue. + * Pass {@code null} to disable retries (failures are only logged). + * @param syncTracker tracks SST files confirmed present in cloud; the delete guard uses it + * to avoid deleting a superseded object before replacements are durable. + * Must be shared with the retry queue. Pass {@code null} for a fresh + * {@link CloudSyncTracker}. + * @param storeScopePrefix optional per-store key prefix to isolate cloud objects + * @param metadataSyncDebounceMs debounce window (ms) for the per-SST metadata sync; see + * {@link #setMetadataSyncDebounceMs(long)} + * @param metadataSyncMaxUnpublished backlog bound on the debounce; see + * {@link #setMetadataSyncMaxUnpublished(int)} + */ + public CloudStorageEventListener(List<String> dataRoots, + boolean startupHydrationEnabled, + long readMissGuardWindowMs, + CloudUploadRetryQueue retryQueue, + CloudSyncTracker syncTracker, + String storeScopePrefix, + long metadataSyncDebounceMs, + int metadataSyncMaxUnpublished) { + // Fail fast on a missing/empty data-root list: primaryDataRoot is derived from index 0 + // below, and every key<->path conversion depends on at least one root. Without this guard + // the constructor would throw an opaque IndexOutOfBoundsException at + // allDataRoots.get(0), which is far harder to diagnose than a config error. + if (dataRoots == null || dataRoots.isEmpty()) { + throw new IllegalArgumentException( + "CloudStorageEventListener requires at least one data root; none configured. " + + "Check the store data-path configuration (e.g. rocksdb.data_path / " + + "raft.path) used to derive the cloud storage data roots."); + } + // Normalize each root + this.allDataRoots = new ArrayList<>(); + for (String root : dataRoots) { + String normalised = Paths.get(root).toAbsolutePath().normalize().toString(); + // Strip trailing separator so substring arithmetic is consistent. + normalised = normalised.endsWith(File.separator) + ? normalised.substring(0, normalised.length() - 1) + : normalised; + this.allDataRoots.add(normalised); + } + this.primaryDataRoot = this.allDataRoots.get(0); + this.startupHydrationEnabled = startupHydrationEnabled; + this.readMissGuardWindowMs = Math.max(0L, readMissGuardWindowMs); + this.readMissAttemptTs = new ConcurrentHashMap<>(); + this.retryQueue = retryQueue; + this.syncTracker = syncTracker != null ? syncTracker : new CloudSyncTracker(); + this.storeScopePrefix = normaliseKeyPrefix(storeScopePrefix); + this.uploadExecutor = sharedUploadExecutor(); + this.metadataSyncDebounceMs = metadataSyncDebounceMs; + this.metadataSyncMaxUnpublished = metadataSyncMaxUnpublished; + + // A crash may leave local pending-delete markers whose remote cleanup never completed. + // Kick off bounded async retries for each so stale remote data is eventually purged even if + // the affected DB is never re-opened (an open would otherwise be the only trigger). + processPendingDeleteMarkersOnStartup(); + // Truncate purge intent is also crash-durable via local markers. + processPendingTruncateMarkersOnStartup(); + } + + private static ThreadFactory newUploadThreadFactory() { + return r -> { + Thread t = new Thread(r, "cloud-upload-dispatch"); + t.setDaemon(true); + return t; + }; + } + + /** + * Shuts down the shared SST upload executor, waiting up to {@code timeout} for in-flight + * uploads to complete. Called from {@link org.apache.hugegraph.store.node.AppConfig#onDestroy()} + * so uploads started before shutdown have a chance to finish before the JVM exits. + */ + public static void shutdownSharedUploadExecutor(long timeout, + java.util.concurrent.TimeUnit unit) { + ThreadPoolExecutor executor; + synchronized (CloudStorageEventListener.class) { + // Raise the gate BEFORE draining: in-flight upload tasks that finish during the drain + // call requestDebouncedMetadataSync(), which now short-circuits so it cannot resurrect + // the metadata-sync scheduler we are about to tear down. + uploadSubsystemShuttingDown = true; + executor = sharedUploadExecutor; + // Clear the executor field so a later listener construction (Spring context restart in + // the same JVM) lazily recreates a live instance rather than reusing a terminated one. + sharedUploadExecutor = null; + } + // Drain the upload executor FIRST, while the metadata-sync scheduler is still alive (so any + // inline metadata publish an in-flight task performs can still run). The gate prevents new + // trailing syncs from being scheduled. + if (executor != null) { + executor.shutdown(); + try { + if (!executor.awaitTermination(timeout, unit)) { + log.warn("CloudStorageEventListener: shared upload executor did not terminate " + + "within {}{}; forcing shutdown", timeout, unit); + handoffDroppedUploadTasks(executor.shutdownNow()); + } + } catch (InterruptedException e) { + handoffDroppedUploadTasks(executor.shutdownNow()); + Thread.currentThread().interrupt(); + } + } + // Only AFTER the upload executor has fully drained do we tear down the metadata-sync + // scheduler — no in-flight upload task can recreate it now (executor is terminated and the + // gate is up). + ScheduledExecutorService scheduler; + synchronized (CloudStorageEventListener.class) { + scheduler = metadataSyncScheduler; + metadataSyncScheduler = null; + } + if (scheduler != null) { + scheduler.shutdownNow(); + } + } + + /** + * Routes upload tasks discarded by a forced {@code shutdownNow()} into their retry queue / DLQ + * so an SST accepted locally but never mirrored is not silently lost from the durability + * pipeline. Only {@link SstUploadTask}s carry the metadata needed for handoff; any other + * runnable (there should be none) is ignored. + */ + private static void handoffDroppedUploadTasks(java.util.List<Runnable> dropped) { + if (dropped == null || dropped.isEmpty()) { + return; + } + int handed = 0; + for (Runnable r : dropped) { + if (r instanceof SstUploadTask) { + try { + ((SstUploadTask) r).handoffOnShutdown(); + handed++; + } catch (Exception e) { + log.warn("Failed to hand off a dropped upload task to retry/DLQ: {}", + e.getMessage()); + } + } + } + if (handed > 0) { + log.warn("CloudStorageEventListener: handed off {} queued upload task(s) to retry/DLQ " + + "during shutdown so their upload intent survives.", handed); + } + } + + /** + * Finds which configured data roots contain the given file path. + * Returns the matching root, or the primary root if no exact match is found. + * + * <p>Uses {@link Path#startsWith} so that {@code /data/store1} never incorrectly + * matches {@code /data/store10/...} (a raw string prefix match would). + * + * @param filePath absolute file path + * @return the matched configured data root for this file + */ + private String findMatchingDataRoot(String filePath) { + Path fileNormalized = Paths.get(filePath).toAbsolutePath().normalize(); + for (String root : allDataRoots) { + if (fileNormalized.startsWith(Paths.get(root))) { + return root; + } + } + // Fallback to primary root (should not happen in normal operation) + return primaryDataRoot; + } + + // ----------------------------------------------------------------------- + // RocksdbChangedListener + // ----------------------------------------------------------------------- + + @Override + public void onDBOpening(String dbName, String dbPath) { + if (!startupHydrationEnabled) { + return; + } + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + // Pending-delete guard: a prior delete's remote cleanup may not be confirmed (provider was + // down, or the purge failed and is being retried). Checked BEFORE the provider-null return + // so a re-create during a provider outage still cannot re-hydrate the deleted generation. + String prefix = dbPrefix(dbPath); + if (hasPendingDeleteMarker(prefix)) { + if (provider != null) { + // Finish cleanup inline (under the per-prefix lock) BEFORE the open proceeds, so the + // re-created DB's later uploads cannot be purged by the async retry. If it succeeds + // the marker is removed and it is safe to hydrate (the prefix is now empty). + tryCompletePendingDelete(provider, dbName, prefix); + } + if (hasPendingDeleteMarker(prefix)) { + log.warn("Cloud pre-hydration skipped for db={}: remote-delete cleanup not yet " + + "confirmed — opening fresh and blocking re-hydration of deleted data.", + dbName); + return; + } + } + if (provider == null) { + return; + } + preHydrateDbFiles(provider, dbName, dbPath); + } + + /** + * Called when a read returns null in RocksDB. + * + * <p>We restore only the SST files that RocksDB references as <em>live</em> in its manifest but + * that are physically missing on local disk, downloading each back to its <em>exact original + * path</em> so RocksDB finds it on the next access. This deliberately avoids + * {@code ingestExternalFile}, which would (a) risk placing a file into the wrong column family + * and (b) assign a fresh sequence number that can resurrect deleted keys. Restricting to the + * live set also guarantees superseded / compacted-away objects are never resurrected. + * + * <p>Note: a genuine key-not-found also arrives here as {@code value == null}; in that case no + * live file is missing and we return {@code false} without any cloud I/O. + */ + @Override + public boolean onReadMiss(RocksDBSession session, String table, byte[] key) { + String dbName = session.getGraphName(); + if (!shouldAttemptReadMissHydration(dbName, table)) { + return false; + } + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return false; + } + int restored = restoreMissingLiveFiles(provider, dbName, + RocksDBFactory.getInstance().getLiveSstFiles(dbName)); + if (restored > 0) { + log.info("Cloud read-miss hydration: restored {} missing live SST file(s) for db={}", + restored, dbName); + return true; + } + return false; + } + + /** + * Downloads back any SST files that are live in RocksDB's manifest but missing on local disk, + * writing each to its original path. Returns the number of files restored. + * + * <p>Package-private testable seam: caller supplies the live-file set. + */ + int restoreMissingLiveFiles(CloudStorageProvider provider, String dbName, + List<LiveSstFile> liveFiles) { + int restored = 0; + for (LiveSstFile live : liveFiles) { + Path localPath = Paths.get(live.getAbsolutePath()); + if (Files.exists(localPath)) { + continue; + } + String remoteKey = toRelativeKey(live.getAbsolutePath()); + // De-dup concurrent restores of the same object (thundering herd on a cold read + // miss). If another thread already holds the slot, skip: it will produce the file. + String restoreGuardKey = dbName + "::" + remoteKey; + if (!inFlightRestores.add(restoreGuardKey)) { + continue; + } + try { + // Re-check after acquiring the slot: a concurrent restore may have just finished. + if (Files.exists(localPath)) { + continue; + } + if (!provider.fileExists(remoteKey)) { + log.warn("Cloud read-miss: live file missing locally AND absent in cloud: " + + "db={}, key={}", dbName, remoteKey); + continue; + } + Files.createDirectories(localPath.getParent()); + // Download to a UNIQUE sibling temp file, then atomically move into place. The + // per-thread/per-attempt suffix prevents two concurrent restorers from writing the + // same temp file (which would interleave into a corrupt SST), and REPLACE_EXISTING + // makes a late second mover a harmless idempotent overwrite. A crash mid-download + // never leaves RocksDB reading a truncated SST at the expected path. + Path tmp = localPath.resolveSibling( + localPath.getFileName() + ".hydrate-" + Thread.currentThread().getId() + + "-" + System.nanoTime()); + try { + provider.downloadFile(remoteKey, tmp.toString()); + try { + Files.move(tmp, localPath, + java.nio.file.StandardCopyOption.ATOMIC_MOVE, + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + } catch (java.nio.file.AtomicMoveNotSupportedException ex) { + // Cross-filesystem / FS without atomic-move support: fall back to a + // non-atomic replace so restore still succeeds. + Files.move(tmp, localPath, + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + } + } catch (IOException e) { + deleteIfExistsQuietly(tmp, "read-miss restore temp cleanup"); + throw e; + } + syncTracker.markConfirmed(dbName, live.getAbsolutePath()); + restored++; + } catch (IOException e) { + log.warn("Cloud read-miss restore failed: db={}, key={}, reason={}", + dbName, remoteKey, e.getMessage()); + } finally { + inFlightRestores.remove(restoreGuardKey); + } + } + return restored; + } + + /** + * Called when a new RocksDB instance is opened for the first time. + * + * <p>Uploads any SST files that already exist in {@code dbPath} (e.g. from a previous run) + * and then triggers a MemTable flush so that WAL-recovered data is also written to + * SST files and eventually forwarded here via {@link #onTableFileCreated}. + * + * @param dbName logical name of the graph / partition + * @param dbPath absolute path of the RocksDB directory + */ + @Override + public void onDBCreated(String dbName, String dbPath) { + recordDb(dbName, dbPath); + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + // Do not propagate upload failures: the session is already live in dbSessionMap at + // this point, so throwing here would cause the createGraphDB caller to receive an + // error while the DB is actually open and usable by other threads (split-brain). + try { + uploadExistingSstFiles(provider, dbName, dbPath); + } catch (Exception e) { + log.warn("Cloud initial-upload failed for db={}: {} — DB is open, " + + "existing SSTs may not be in cloud yet", dbName, e.getMessage()); + } + flushDb(dbName); + // Mirror metadata immediately after initial upload/flush to keep cloud state recoverable. + syncMetadataSnapshotInline(provider, dbName); + } + + /** + * Called just before the local RocksDB directory is removed. Writes a small tombstone object + * (key = {@code dbPrefix + }{@value #DB_TOMBSTONE_SUFFIX}) outside the data prefix so that + * any subsequent {@link #preHydrateDbFiles} call for the same path will detect the deleted + * generation and skip hydration rather than re-ingesting stale objects. + * + * <p>This callback fires while the session is still in a pending-destroy list (refcount may + * be non-zero). The tombstone write is best-effort: a failure is logged but does not block + * the deletion. The cloud purge in {@link #onDBDeleted} provides a second line of defence. + * + * @param dbName logical graph/partition name + * @param dbPath absolute path of the RocksDB directory being destroyed + */ + @Override + public void onDBDeleteBegin(String dbName, String dbPath) { + String prefix = dbPrefix(dbPath); + // ALWAYS persist a LOCAL pending-delete marker first. This is the durable anti-resurrection + // guard that survives a provider-unavailable window: even if we cannot write the remote + // tombstone or purge below, the marker blocks hydration of this DB (see onDBOpening) and + // drives async cleanup until confirmed. It is removed once the remote purge succeeds. + // + // Marker durability is a hard precondition for a safe delete: if we cannot fsync it, a crash + // could lose the guard and let stale remote SST/metadata be re-hydrated as live data. So on a + // persistence failure we flip the health signal to degraded and HOLD delete progression + // (throw) rather than proceeding unguarded — the caller can retry once local storage recovers. + try { + writePendingDeleteMarker(dbName, prefix); + } catch (IOException e) { + deleteMarkerHealthy = false; + log.error("Cloud pending-delete marker could not be durably persisted for db={} " + + "prefix={}: {} — HOLDING delete to avoid an unguarded delete that a crash " + + "could let re-hydrate. Delete-marker health is DEGRADED.", + dbName, prefix, e.getMessage()); + throw new IllegalStateException( + "Cannot durably persist pending-delete marker for db=" + dbName + + "; holding delete to preserve the anti-resurrection guard", e); + } + + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + // No provider now — the local marker keeps hydration blocked and onDBDeleted will + // schedule the retry that writes the tombstone + purges once a provider returns. + log.warn("Cloud DB delete begin with no active provider: db={} — persisted local " + + "pending-delete marker to block re-hydration until cleanup completes.", dbName); + return; + } + // Tombstone lives OUTSIDE the data prefix (sibling, not child) so it is not + // accidentally deleted by purgeRemotePrefix when SST objects are still present. + String tombstoneKey = prefix + DB_TOMBSTONE_SUFFIX; + Path tmp = null; + try { + tmp = Files.createTempFile("hgstore-tombstone-", ".tmp"); + Files.write(tmp, "deleted".getBytes(java.nio.charset.StandardCharsets.UTF_8)); + provider.uploadFile(tmp.toString(), tombstoneKey); + log.info("Cloud DB tombstone written: db={}, key={}", dbName, tombstoneKey); + } catch (Exception e) { + log.warn("Cloud DB tombstone write failed (onDBDeleted will still purge): " + + "db={}, key={}, reason={}", dbName, tombstoneKey, e.getMessage()); + } finally { + if (tmp != null) { + try { + Files.deleteIfExists(tmp); + } catch (IOException ignore) { + // best-effort temp-file cleanup + } + } + } + } + + /** + * Called after the local RocksDB directory has been physically removed. Purges all cloud + * objects under the DB prefix (SSTs, metadata, tombstone) so a future creation at the same + * path starts with a clean remote state. Also clears all in-memory state for this DB. + * + * <p>The purge is best-effort: individual delete failures are logged at DEBUG level and do + * not throw. Any objects that survive the purge are neutralised by the tombstone check in + * {@link #preHydrateDbFiles}: the next open will find the tombstone (or an empty prefix if + * the purge was complete), skip hydration, and clean up any leftovers. + * + * @param dbName logical graph/partition name + * @param dbPath absolute path of the now-deleted RocksDB directory + */ + @Override + public void onDBDeleted(String dbName, String dbPath) { + // Clear in-memory tracking so no stale state bleeds into a recreated DB. + syncTracker.clearDb(dbName); + readMissAttemptTs.entrySet().removeIf(e -> e.getKey().startsWith(dbName + "::")); + dbNameByDir.values().removeIf(dbName::equals); + metadataSyncLocks.remove(dbName); + lastPublishedMetadataGeneration.remove(dbName); + truncationTimes.remove(dbName); + truncatingDbs.remove(dbName); + deferredSstDeletes.removeIf(t -> dbName.equals(t.dbName)); + lastMetadataSyncMs.remove(dbName); + pendingMetadataSync.remove(dbName); + unpublishedUploads.remove(dbName); + // Remove per-DB meters so meter cardinality does not grow without bound as DBs churn. + CloudStorageMetrics.unregisterDatabaseMetrics(dbName); + + String prefix = dbPrefix(dbPath); + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + // Provider unavailable: cannot purge now. The local pending-delete marker (written in + // onDBDeleteBegin) keeps hydration blocked; schedule a bounded retry that purges once a + // provider returns, so stale remote data cannot be resurrected. + log.warn("Cloud DB delete with no active provider: db={} — scheduling deferred remote " + + "purge; local pending-delete marker guards re-hydration meanwhile.", dbName); + scheduleDeletePurgeRetry(dbName, prefix, 1); + return; + } + boolean purgeSucceeded = purgeRemotePrefix(provider, dbName, prefix); + if (purgeSucceeded) { + // Remote data is gone: drop the sibling tombstone and the local marker (cleanup done). + deleteTombstoneBestEffort(provider, dbName, prefix); + removePendingDeleteMarker(prefix); + } else { + // Purge failed: stale SST objects may remain. Keep the tombstone AND the local marker so + // a future onDBOpening still blocks re-hydration, and retry the purge (bounded) — same + // safety level as the truncate purge, instead of a single best-effort attempt. + log.warn("Cloud DB purge failed for db={}: tombstone + local marker preserved to guard " + + "re-hydration; scheduling bounded retries.", dbName); + scheduleDeletePurgeRetry(dbName, prefix, 1); + } + } + + /** Best-effort deletion of the sibling delete-tombstone once the prefix purge has succeeded. */ + private void deleteTombstoneBestEffort(CloudStorageProvider provider, String dbName, + String prefix) { + String tombstoneKey = prefix + DB_TOMBSTONE_SUFFIX; + try { + provider.deleteFile(tombstoneKey); + log.debug("Cloud DB tombstone cleaned up: db={}, key={}", dbName, tombstoneKey); + } catch (IOException e) { + log.debug("Cloud DB tombstone cleanup failed (non-critical): db={}, key={}: {}", + dbName, tombstoneKey, e.getMessage()); + } + } + + /** + * Called after a RocksDB has been truncated (all data cleared but directory preserved). + * Purges all cloud objects under the DB prefix (SSTs, metadata) so the remote state matches + * the now-empty local state. Also clears all in-memory sync tracking for this DB. + * + * <p>This is triggered by graph.clear() operations to ensure cloud storage is cleaned up + * when the graph data is cleared. + * + * @param dbName logical graph/partition name + * @param dbPath absolute path of the RocksDB directory + */ + @Override + public void onDBTruncateBegin(String dbName, String dbPath) { + truncatingDbs.add(dbName); + truncationTimes.put(dbName, System.currentTimeMillis()); + syncTracker.clearDb(dbName); + readMissAttemptTs.entrySet().removeIf(e -> e.getKey().startsWith(dbName + "::")); + deferredSstDeletes.removeIf(t -> dbName.equals(t.dbName)); + } + + @Override + public void onDBTruncated(String dbName, String dbPath) { + truncatingDbs.add(dbName); + try { + String prefix = dbPrefix(dbPath); + try { + writePendingTruncateMarker(dbName, prefix); + } catch (IOException e) { + log.error("Cloud truncate marker persist failed for db={}, prefix={}: {} — " + + "remote purge intent is not crash-durable until the next successful " + + "persist.", + dbName, prefix, e.getMessage()); + } + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + // Provider can be transiently unavailable during reconfiguration; keep the bounded + // retry chain alive so stale pre-truncate objects are still purged once it returns. + log.warn("Cloud truncate purge for db={} found no active provider; scheduling " + + "bounded retries for prefix={}", dbName, prefix); + scheduleTruncatePurgeRetry(dbName, prefix); + return; + } + boolean purged = purgeRemotePrefix(provider, dbName, prefix); + if (purged) { + removePendingTruncateMarker(prefix); + } + if (!purged) { + // Do NOT ignore a failed purge: stale remote SST/metadata still describe the + // pre-clear generation and could be re-hydrated after a restart + disk loss, + // silently resurrecting data the operator explicitly cleared. A transient cloud + // error is the common cause, so retry the purge (bounded, with backoff) on the + // scheduler. If every attempt fails we log loudly for operator intervention rather + // than reporting a clean truncate. + log.error("Cloud truncate purge failed for db={} (prefix={}): stale remote objects " + + "remain; scheduling bounded retries to prevent resurrection of cleared " + + "data on a future restore.", dbName, prefix); + scheduleTruncatePurgeRetry(dbName, prefix); + } + } finally { + truncationTimes.put(dbName, System.currentTimeMillis()); + truncatingDbs.remove(dbName); + } + } + + /** Max asynchronous retries for a failed truncate purge before giving up (logging loudly). */ + private static final int MAX_TRUNCATE_PURGE_RETRIES = 5; + /** Base backoff (ms) for truncate-purge retries; doubles each attempt up to a small cap. */ + private static final long TRUNCATE_PURGE_RETRY_BASE_MS = 500L; + /** Cap for provider-unavailable truncate-retry backoff (ms), separate from purge failures. */ + private static final long TRUNCATE_PROVIDER_UNAVAILABLE_MAX_DELAY_MS = 8_000L; + + /** + * Retries a failed truncate purge asynchronously with capped exponential backoff. Each attempt + * refreshes the truncation grace window so metadata mirroring stays suppressed until the remote + * prefix is confirmed clean, avoiding a re-mirror of not-yet-purged stale objects. After + * {@link #MAX_TRUNCATE_PURGE_RETRIES} failures it logs an error and stops — the stale objects + * then require manual cleanup, but the failure is at least visible rather than silent. + */ + private void scheduleTruncatePurgeRetry(String dbName, String prefix) { + scheduleTruncatePurgeRetry(dbName, prefix, 1, 0); + } + + private void scheduleTruncatePurgeRetry(String dbName, String prefix, int attempt, + int providerUnavailableRetries) { + if (attempt > MAX_TRUNCATE_PURGE_RETRIES) { + log.error("Cloud truncate purge for db={} still failing after {} attempt(s) — remote " + + "prefix '{}' may retain stale objects that could be re-hydrated on restore. " + + "Manual cleanup of that prefix is required.", + dbName, MAX_TRUNCATE_PURGE_RETRIES, prefix); + return; + } + int exp = providerUnavailableRetries > 0 ? providerUnavailableRetries - 1 : attempt - 1; + exp = Math.min(Math.max(exp, 0), 30); + long cap = providerUnavailableRetries > 0 + ? TRUNCATE_PROVIDER_UNAVAILABLE_MAX_DELAY_MS + : 8_000L; + long delay = Math.min(TRUNCATE_PURGE_RETRY_BASE_MS * (1L << exp), cap); + // Keep the grace window fresh so we do not mirror new metadata over the still-stale prefix. + truncationTimes.put(dbName, System.currentTimeMillis()); + ScheduledExecutorService scheduler = metadataSyncScheduler(); + if (scheduler == null) { + // Upload subsystem is shutting down; the scheduler will not be resurrected. The remote + // prefix may retain stale objects — surface it rather than NPE on a null scheduler. + log.error("Cloud truncate purge retry for db={} cannot be scheduled (subsystem " + + "shutting down) — remote prefix '{}' may retain stale objects.", + dbName, prefix); + return; + } + try { + scheduler.schedule(() -> { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + // Provider temporarily unavailable (e.g. reconfiguration window). This is NOT a + // remote purge failure, so do not consume the bounded purge-attempt budget here; + // otherwise a long outage can exhaust retries before the provider returns and + // leave stale pre-truncate objects behind indefinitely. + log.warn("Cloud truncate purge retry for db={} found no active provider " + + "(attempt {}, providerUnavailableRetries={}); rescheduling " + + "without consuming retry budget.", + dbName, attempt, providerUnavailableRetries); + scheduleTruncatePurgeRetry(dbName, prefix, attempt, + providerUnavailableRetries + 1); + return; + } + if (purgeRemotePrefix(provider, dbName, prefix)) { + removePendingTruncateMarker(prefix); + log.info("Cloud truncate purge succeeded on retry for db={} (attempt {})", + dbName, attempt); + } else { + scheduleTruncatePurgeRetry(dbName, prefix, attempt + 1, 0); + } + }, delay, TimeUnit.MILLISECONDS); + } catch (RejectedExecutionException e) { + log.error("Cloud truncate purge retry could not be scheduled for db={} (scheduler " + + "shutting down) — remote prefix '{}' may retain stale objects.", + dbName, prefix); + } + } + + // ----------------------------------------------------------------------- + // DB-deletion remote cleanup: local pending-delete marker + bounded purge retry + // ----------------------------------------------------------------------- + + /** Local subdirectory of the primary data root holding pending-remote-delete markers. */ + static final String PENDING_DELETE_DIR = ".cloud-pending-delete"; + /** Local subdirectory holding pending-remote-truncate purge markers. */ + static final String PENDING_TRUNCATE_DIR = ".cloud-pending-truncate"; + /** Max async retries for a failed delete purge before giving up (logging loudly). */ + private static final int MAX_DELETE_PURGE_RETRIES = 8; + /** Base backoff (ms) for delete-purge retries; doubles each attempt up to a small cap. */ + private static final long DELETE_PURGE_RETRY_BASE_MS = 500L; + /** Retry delay for superseded-SST deletes deferred by transient metadata/cloud failures. */ + private static final long DEFERRED_SST_DELETE_RETRY_MS = 500L; + /** Emit one WARN when deferred delete backlog crosses this size; otherwise keep DEBUG-only. */ + private static final int DEFERRED_SST_DELETE_WARN_THRESHOLD = 64; + + /** Pending superseded-SST deletes held until metadata/cloud preconditions are satisfied. */ + private final Set<DeferredSstDelete> deferredSstDeletes = ConcurrentHashMap.newKeySet(); + /** Coalescing guard so only one deferred-delete retry task is scheduled at a time. */ + private final AtomicBoolean deferredSstDeleteRetryScheduled = new AtomicBoolean(false); + /** Ensures backlog WARN is emitted only once per threshold crossing. */ + private final AtomicBoolean deferredSstDeleteBacklogWarned = new AtomicBoolean(false); + + /** Immutable payload for a deferred superseded-SST delete. */ + private static final class DeferredSstDelete { + + final String dbName; + final String filePath; + + DeferredSstDelete(String dbName, String filePath) { + this.dbName = dbName; + this.filePath = filePath; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof DeferredSstDelete)) { + return false; + } + DeferredSstDelete that = (DeferredSstDelete) o; + return java.util.Objects.equals(this.dbName, that.dbName) + && java.util.Objects.equals(this.filePath, that.filePath); + } + + @Override + public int hashCode() { + return java.util.Objects.hash(this.dbName, this.filePath); + } + } + + private Path pendingDeleteDir() { + return Paths.get(primaryDataRoot, PENDING_DELETE_DIR); + } + + private Path pendingTruncateDir() { + return Paths.get(primaryDataRoot, PENDING_TRUNCATE_DIR); + } + + private static void deleteIfExistsQuietly(Path path, String context) { + try { + Files.deleteIfExists(path); + } catch (IOException cleanupError) { + log.debug("Failed to cleanup temp file during {}: path={}, reason={}", + context, path, cleanupError.getMessage()); + } + } + + /** + * Marker file path for a remote prefix. The remote prefix contains '/', so the filename is a + * URL-safe Base64 of the prefix — reversible (for the startup scan) and collision-free. + */ + private Path pendingDeleteMarkerPath(String prefix) { + return pendingDeleteDir().resolve(encodeMarkerName(prefix)); + } + + private Path pendingTruncateMarkerPath(String prefix) { + return pendingTruncateDir().resolve(encodeMarkerName(prefix)); + } + + private static String encodeMarkerName(String prefix) { + return java.util.Base64.getUrlEncoder().withoutPadding() + .encodeToString(prefix.getBytes(java.nio.charset.StandardCharsets.UTF_8)); + } + + /** + * Crash-safely persists a local pending-remote-delete marker for {@code prefix}. This is the + * durable anti-resurrection guard, so it is NOT best-effort: the marker file is written and + * fsynced, and the containing directory is fsynced so the new directory entry itself survives a + * crash. Throws {@link IOException} if durability cannot be guaranteed — the caller must then + * hold delete progression rather than proceed with an unguarded delete. + */ + @SuppressWarnings("ResultOfMethodCallIgnored") + private void writePendingDeleteMarker(String dbName, String prefix) throws IOException { + Path dir = pendingDeleteDir(); + Files.createDirectories(dir); + Path marker = pendingDeleteMarkerPath(prefix); + // Write + fsync the marker file so its bytes are on stable storage. + try (FileChannel ch = FileChannel.open(marker, StandardOpenOption.CREATE, + StandardOpenOption.WRITE, + StandardOpenOption.TRUNCATE_EXISTING)) { + ch.write(java.nio.ByteBuffer.wrap( + prefix.getBytes(java.nio.charset.StandardCharsets.UTF_8))); + ch.force(true); + } + // fsync the directory so the new/renamed marker entry itself is durable (a file fsync does + // not guarantee the directory entry pointing at it survives a crash on many filesystems). + try (FileChannel dirCh = FileChannel.open(dir, StandardOpenOption.READ)) { + dirCh.force(true); + } catch (IOException e) { + // Windows rejects directory-handle fsync; elsewhere treat it as a real durability error. + if (WINDOWS_OS) { + log.debug("Directory fsync of {} not supported on this platform: {}", + dir, e.getMessage()); + } else { + throw e; + } + } + deleteMarkerHealthy = true; + log.debug("Cloud pending-delete marker persisted (fsync'd): db={}, prefix={}", dbName, prefix); + } + + /** + * Crash-safely persists a local pending-truncate marker for {@code prefix}. This keeps remote + * truncate-purge intent durable across process restarts while the provider is unavailable. + */ + @SuppressWarnings("ResultOfMethodCallIgnored") + private void writePendingTruncateMarker(String dbName, String prefix) throws IOException { Review Comment: Important: replace or collapse. The delete and truncate marker code exists twice, and the copies differ only in the directory and the log text: - This method matches `writePendingDeleteMarker` line for line (file fsync, directory fsync, Windows skip). The only extra there is `deleteMarkerHealthy = true`. - `processPendingTruncateMarkersOnStartup` matches `processPendingDeleteMarkersOnStartup` except for the log text and the final schedule call. - `pending*Dir`, `pending*MarkerPath` and `removePending*Marker` come in pairs. Requested change: one `writeMarker(Path dir, String prefix)`, one `removeMarker(Path dir, String prefix)` and one `scanMarkers(Path dir, BiConsumer<String, String> onMarker)` taking `(dbName, prefix)`. Each caller passes its directory and callback, and the delete caller sets `deleteMarkerHealthy` itself. That removes about 90 lines. While there: the file name is Base64(prefix), the body repeats the prefix, and the startup scan drops the marker unless the two match. Decode the name and write an empty file. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
