vaijosh commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r4270502439
########## 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: Addressed ########## 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: Addressed -- 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]
