vaijosh commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r4069890187
########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudStorageEventListener.java: ########## @@ -0,0 +1,3026 @@ +/* + * 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; + + private 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 Roaring bitmap). */ + 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. */ + private 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). + */ + private 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"; + + /** + * Convenience constructor with startup hydration enabled and default read-miss guard window. + * + * @param dataRoots configured store data roots (typically parsed from comma-separated + * {@code app.data-path}) + */ + public CloudStorageEventListener(List<String> dataRoots) { Review Comment: Thanks @bitflicker64. I have addressed this. ########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudSyncTracker.java: ########## @@ -0,0 +1,257 @@ +/* + * 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.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.hugegraph.rocksdb.access.RocksDBFactory.LiveSstFile; +import org.roaringbitmap.longlong.Roaring64NavigableMap; + +/** + * Tracks which RocksDB SST files are confirmed present in cloud storage. + * + * <p>SST file names are monotonic per-DB file numbers ({@code 000123.sst}), so the natural key is + * the integer file number rather than the string path. Sync state is kept as one + * {@link Roaring64NavigableMap} per DB — each set bit means "SST file number N for this DB is + * confirmed present in cloud". Roaring bitmaps stay compact even as low numbers are cleared after + * old files are deleted, and file numbers are unique across column families within a DB, so a + * single bitmap per DB is sufficient (which fits RocksDB delete events not carrying a CF name) + * + * <p>This tracker is the linchpin of the delete-guard invariant: a superseded cloud object is + * deleted only once every live SST file of that DB is confirmed here. + * + * <p><b>Epoch-based stale-callback protection:</b> each DB has a monotone epoch counter that is + * incremented by {@link #clearDb}. Callers that want their confirmation to survive a concurrent + * recreation must first call {@link #currentEpoch} before the upload, then pass the captured + * epoch to {@link #markConfirmedIfEpoch}. A late callback carrying an old epoch is silently + * dropped, preventing file-number reuse after DB recreation from producing stale confirmations. + * + * <p>{@link Roaring64NavigableMap} is not thread-safe, so all access to a per-DB bitmap is + * synchronized on the bitmap instance. + */ +public final class CloudSyncTracker { + + /** Holds the bitmap and the epoch it was created under. */ + private static final class DbState { + final long epoch; + final Roaring64NavigableMap bitmap; Review Comment: Thanks @bitflicker64. I have addressed this. -- 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]
