vaijosh commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r3631833129
########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudStorageEventListener.java: ########## @@ -0,0 +1,1318 @@ +/* + * 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.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +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.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import java.util.stream.Stream; + +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 path of the store's data root directory. */ + private final String dataRoot; + + /** 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; + + /** + * 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; + + /** + * When {@code > 0}, {@link #onTableFileCreated} slows the RocksDB flush/compaction thread while + * the number of not-yet-durable uploads (retry-queue in-flight + DLQ) exceeds this watermark, + * so ingestion cannot outrun the cloud mirror. {@code 0} disables backpressure. + */ + private final int backpressureHighWatermark; + + + /** Upper bound on how long a single {@link #onTableFileCreated} call will block for backpressure. */ + private static final long BACKPRESSURE_MAX_WAIT_MS = 30_000L; + private static final long BACKPRESSURE_POLL_MS = 50L; + + /** 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; + private static final ThreadPoolExecutor SHARED_UPLOAD_EXECUTOR = + new ThreadPoolExecutor( + ASYNC_UPLOAD_THREADS, + ASYNC_UPLOAD_THREADS, + 60L, + TimeUnit.SECONDS, + new ArrayBlockingQueue<>(ASYNC_UPLOAD_QUEUE_CAPACITY), + newUploadThreadFactory(), + new ThreadPoolExecutor.AbortPolicy()); + private final ThreadPoolExecutor uploadExecutor; + private final Path uploadStagingDir; + + // ----------------------------------------------------------------------- + // Metadata (CURRENT/MANIFEST/OPTIONS[/WAL]) mirroring & consistent restore + // ----------------------------------------------------------------------- + + + /** + * When {@code true} ({@code wal-mode: wal}), the active WAL {@code *.log} segments are mirrored + * alongside the metadata and replayed on restore. When {@code false} ({@code wal-mode: flush}), + * no WAL is mirrored. + */ + private final boolean walModeEnabled; + + /** 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<>(); + + /** + * 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; + + /** + * Sentinel object key written to the DB prefix during database deletion. + * {@link #preHydrateDbFiles} checks for this key and skips hydration when present, preventing + * a newly-recreated DB from ingesting data that belonged to a previous deleted generation. + */ + static final String DB_TOMBSTONE_FILE = "_DELETED"; + + /** + * @param dataRoot absolute path of the store's data directory + * (value of {@code app.data-path}, resolved to an absolute path). + */ + public CloudStorageEventListener(String dataRoot) { + this(dataRoot, true, DEFAULT_READ_MISS_GUARD_WINDOW_MS, null); + } + + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled) { + this(dataRoot, startupHydrationEnabled, DEFAULT_READ_MISS_GUARD_WINDOW_MS, null); + } + + /** + * @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) + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs) { + this(dataRoot, startupHydrationEnabled, readMissGuardWindowMs, null); + } + + /** + * @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). + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs, + CloudUploadRetryQueue retryQueue) { + this(dataRoot, startupHydrationEnabled, readMissGuardWindowMs, retryQueue, + new CloudSyncTracker(), 0); + } + + /** + * Full constructor. + * + * @param syncTracker tracks SST files confirmed present in cloud; the delete guard + * uses it to avoid deleting a superseded object before its + * replacements are durable. Must be shared with the retry queue. + * @param backpressureHighWatermark {@code > 0} to slow ingestion while the pending-upload backlog + * exceeds this value; {@code 0} disables backpressure. + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs, + CloudUploadRetryQueue retryQueue, + CloudSyncTracker syncTracker, + int backpressureHighWatermark) { + this(dataRoot, startupHydrationEnabled, readMissGuardWindowMs, retryQueue, syncTracker, + backpressureHighWatermark, false); + } + + /** + * Full constructor including metadata-mirroring parameters. + * + * @param walModeEnabled {@code true} for {@code wal-mode: wal} (mirror + replay WAL); + * {@code false} for {@code wal-mode: flush} (force flush, no WAL) + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs, + CloudUploadRetryQueue retryQueue, + CloudSyncTracker syncTracker, + int backpressureHighWatermark, + boolean walModeEnabled) { + this(dataRoot, startupHydrationEnabled, readMissGuardWindowMs, retryQueue, syncTracker, + backpressureHighWatermark, walModeEnabled, null); + } + + /** + * Full constructor including metadata-mirroring and per-store key namespace parameters. + * + * @param storeScopePrefix optional per-store key prefix to isolate cloud objects across + * distributed store nodes sharing the same bucket/path-prefix. + */ + public CloudStorageEventListener(String dataRoot, + boolean startupHydrationEnabled, + long readMissGuardWindowMs, + CloudUploadRetryQueue retryQueue, + CloudSyncTracker syncTracker, + int backpressureHighWatermark, + boolean walModeEnabled, + String storeScopePrefix) { + String normalised = Paths.get(dataRoot).toAbsolutePath().normalize().toString(); + // Strip trailing separator so substring arithmetic is consistent. + this.dataRoot = normalised.endsWith(File.separator) + ? normalised.substring(0, normalised.length() - 1) + : normalised; + this.startupHydrationEnabled = startupHydrationEnabled; + this.readMissGuardWindowMs = Math.max(0L, readMissGuardWindowMs); + this.readMissAttemptTs = new ConcurrentHashMap<>(); + this.retryQueue = retryQueue; + this.syncTracker = syncTracker != null ? syncTracker : new CloudSyncTracker(); + this.backpressureHighWatermark = Math.max(0, backpressureHighWatermark); + this.walModeEnabled = walModeEnabled; + this.storeScopePrefix = normaliseKeyPrefix(storeScopePrefix); + this.uploadStagingDir = Paths.get(this.dataRoot, ".cloud-upload-staging"); + this.uploadExecutor = SHARED_UPLOAD_EXECUTOR; + } + + private static ThreadFactory newUploadThreadFactory() { + return r -> { + Thread t = new Thread(r, "cloud-upload-dispatch"); + t.setDaemon(true); + return t; + }; + } + + // ----------------------------------------------------------------------- + // RocksdbChangedListener + // ----------------------------------------------------------------------- + + @Override + public void onDBOpening(String dbName, String dbPath) { + if (!startupHydrationEnabled) { + return; + } + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + 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()); + try { + 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 temp file then atomically move into place so a crash mid-download + // never leaves RocksDB reading a truncated SST at the expected path. + Path tmp = localPath.resolveSibling(localPath.getFileName() + ".hydrate"); + provider.downloadFile(remoteKey, tmp.toString()); + Files.move(tmp, localPath, java.nio.file.StandardCopyOption.ATOMIC_MOVE); + syncTracker.markConfirmed(dbName, live.getAbsolutePath()); + restored++; + } catch (IOException e) { + log.warn("Cloud read-miss restore failed: db={}, key={}, reason={}", + dbName, remoteKey, e.getMessage()); + } + } + 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; + } + uploadExistingSstFiles(provider, dbName, dbPath); + 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 + * ({@value #DB_TOMBSTONE_FILE}) to the DB's remote 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) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + String tombstoneKey = dbPrefix(dbPath) + "/" + DB_TOMBSTONE_FILE; + 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); + + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + purgeRemotePrefix(provider, dbName, dbPrefix(dbPath)); + } + + /** + * 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 + "::")); + } + + @Override + public void onDBTruncated(String dbName, String dbPath) { + truncatingDbs.add(dbName); + try { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + purgeRemotePrefix(provider, dbName, dbPrefix(dbPath)); + } finally { + truncationTimes.put(dbName, System.currentTimeMillis()); + truncatingDbs.remove(dbName); + } + } + + /** + * Checks if a DB is within the grace period after truncation, during which + * metadata syncs should be suppressed to prevent re-uploading purged data. + */ + private boolean isInTruncationGracePeriod(String dbName) { + Long truncationTime = truncationTimes.get(dbName); + if (truncationTime == null) { + return false; + } + long elapsed = System.currentTimeMillis() - truncationTime; + if (elapsed > TRUNCATION_GRACE_PERIOD_MS) { + // Grace period expired, remove the record + truncationTimes.remove(dbName); + return false; + } + return true; + } + + /** + * Deletes every remote object under {@code prefix} using an optimized prefix-level delete + * if available, falling back to individual file deletion if necessary. + * This is called during DB destruction to prevent a recreated DB from hydrating stale data. + */ + private void purgeRemotePrefix(CloudStorageProvider provider, String dbName, String prefix) { + String normalizedPrefix = prefix.endsWith("/") ? prefix : prefix + "/"; + try { + int deleted = provider.deletePrefix(normalizedPrefix); + if (deleted > 0) { + log.info("Cloud DB purge completed: db={}, prefix={}, deleted={}", + dbName, prefix, deleted); + } + } catch (IOException e) { + log.warn("Cloud DB purge failed for db={}, prefix={}: {}", + dbName, prefix, e.getMessage()); + } + } + + /** + * Pins and asynchronously uploads the newly created SST file to the active cloud + * storage provider. + * + * @param dbName RocksDB instance name (partition id) + * @param cfName column-family name + * @param filePath absolute local path of the new SST file + * @param fileSize file size in bytes (informational) + */ + @Override + public void onTableFileCreated(String dbName, String cfName, + String filePath, long fileSize) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + recordDb(dbName, parentDir(filePath)); + CloudStorageMetrics.registerDatabaseMetrics(dbName); + String remoteKey = toRelativeKey(filePath); + + Path pinned; + try { + pinned = pinForAsyncUpload(filePath); + } catch (Exception e) { + String errorType = e.getClass().getSimpleName(); + CloudStorageMetrics.recordUploadFailure(dbName, cfName, errorType); + // Hard link failed — no copy fallback, no extra disk use. Route original SST path + // to retry queue; retry will upload directly from the original file if still present. + log.warn("Cloud upload staging (hard link failed): db={}, cf={}, path={} " + + "— routing original SST to retry queue: {}", + dbName, cfName, filePath, e.getMessage()); + if (retryQueue != null) { + retryQueue.submit(dbName, cfName, filePath, remoteKey, e); + } + applyBackpressure(dbName); + return; + } + + try { + uploadExecutor.execute(() -> { + long startTimeMs = System.currentTimeMillis(); + try { + provider.uploadFile(pinned.toString(), remoteKey); + syncTracker.markConfirmed(dbName, filePath); + long syncLatencyMs = System.currentTimeMillis() - startTimeMs; + CloudStorageMetrics.recordSyncLatency(dbName, syncLatencyMs); + // Skip metadata sync if DB is being truncated or in grace period after truncation + // to allow purge to complete cleanly without metadata files being re-uploaded. + if (!truncatingDbs.contains(dbName) && !isInTruncationGracePeriod(dbName)) { + syncMetadataSnapshotInline(provider, dbName); + } + log.debug("Cloud upload success: db={}, cf={}, path={}, size={}, latencyMs={}", + dbName, cfName, filePath, fileSize, syncLatencyMs); + } catch (Exception e) { + String errorType = e.getClass().getSimpleName(); + CloudStorageMetrics.recordUploadFailure(dbName, cfName, errorType); + log.error("Cloud upload failed (will retry on next compaction): " + + "db={}, cf={}, path={}, error={}", dbName, cfName, + filePath, e.getMessage()); + if (retryQueue != null) { + // Keep retry semantics unchanged: retries target the original SST path. + retryQueue.submit(dbName, cfName, filePath, remoteKey, e); + } + } finally { + try { + Files.deleteIfExists(pinned); + } catch (IOException e) { + log.debug("Failed to cleanup staged upload file {}: {}", + pinned, e.getMessage()); + } + } + }); + } catch (RejectedExecutionException e) { + try { + Files.deleteIfExists(pinned); + } catch (IOException ioe) { + log.debug("Failed to cleanup staged upload file {}: {}", pinned, ioe.getMessage()); + } + CloudStorageMetrics.recordUploadFailure(dbName, cfName, "UploadQueueFull"); + log.error("Cloud upload dispatch rejected (queue full): db={}, cf={}, path={}", + dbName, cfName, filePath); + if (retryQueue != null) { + retryQueue.submit(dbName, cfName, filePath, remoteKey, + new IOException("cloud upload dispatch queue full", e)); + } + } + + // Apply backpressure AFTER handling this file so the flush/compaction thread slows down + // while the cloud mirror is behind, preventing ingestion from outrunning durability. + applyBackpressure(dbName); + } + + /** + * Creates a stable hard-link snapshot of the SST file for async upload, so the upload worker + * can read a consistent source even after RocksDB deletes the original during compaction. + * + * <p>Hard links share the same inode — no extra data blocks are consumed. If the original is + * deleted by compaction before the worker runs, the hard-link still holds the inode alive so + * the upload can proceed normally. + * + * <p>If the hard link fails (e.g. cross-device mount, filesystem hard-link limits), an + * {@link IOException} is thrown. The caller ({@link #onTableFileCreated}) catches this and + * routes the upload to the retry queue using the original SST path — no copy is made and no + * extra disk space is consumed. The retry will succeed as long as the original file still + * exists when it fires; if it has been compacted away the retry queue silently drops it. + */ + private Path pinForAsyncUpload(String filePath) throws IOException { + Path source = Paths.get(filePath); + Files.createDirectories(uploadStagingDir); + String fileName = source.getFileName().toString(); + Path staged = uploadStagingDir.resolve(fileName + ".upload-" + System.nanoTime()); + try { + Files.createLink(staged, source); + return staged; + } catch (Exception linkEx) { + throw new IOException( + "Hard link failed; upload will be retried from original SST path: " + + linkEx.getMessage(), linkEx); + } + } + + /** + * Blocks the calling (RocksDB flush/compaction) thread while the pending-upload backlog exceeds + * {@link #backpressureHighWatermark}, up to {@link #BACKPRESSURE_MAX_WAIT_MS}. This is the + * durability-tier backpressure: it keeps at-risk local-only data bounded. + */ + private void applyBackpressure(String dbName) { + if (backpressureHighWatermark <= 0 || retryQueue == null) { + return; + } + long waited = 0L; + boolean logged = false; + while (pendingUploadBacklog() > backpressureHighWatermark + && waited < BACKPRESSURE_MAX_WAIT_MS) { + if (!logged) { + log.warn("Cloud upload backpressure: db={}, backlog={} > watermark={}, " + + "slowing ingestion", dbName, pendingUploadBacklog(), + backpressureHighWatermark); + logged = true; + } + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(BACKPRESSURE_POLL_MS)); + if (Thread.currentThread().isInterrupted()) { + return; + } + waited += BACKPRESSURE_POLL_MS; + } + if (logged) { + log.info("Cloud upload backpressure released: db={}, backlog={}, waitedMs={}", + dbName, pendingUploadBacklog(), waited); + } + } + + private int pendingUploadBacklog() { + int backlog = uploadExecutor.getQueue().size() + uploadExecutor.getActiveCount(); + if (retryQueue != null) { + backlog += retryQueue.getInFlightCount() + retryQueue.getDlqSize(); + } + return backlog; + } + + /** + * Removes the deleted SST file from the active cloud storage provider. + * + * @param dbName RocksDB instance name (partition id) + * @param cfName column-family name + * @param filePath absolute local path of the deleted SST file + */ + @Override + public void onTableFileDeleted(String dbName, String cfName, String filePath) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + // Skip file deletion during truncation or grace period; the entire DB prefix will be purged anyway + if (truncatingDbs.contains(dbName) || isInTruncationGracePeriod(dbName)) { + log.debug("Skipping delete during truncation: db={}, path={}", dbName, filePath); + return; + } + // DATA-LOSS GUARD: never delete a superseded cloud object until every SST file currently + // live in this DB is confirmed present in cloud. + if (!ensureLiveSetUploaded(provider, dbName)) { + log.warn("Delete skipped (live set not fully durable in cloud): db={}, filePath={}", + dbName, filePath); + return; + } + + // MANIFEST-BEFORE-DELETE INVARIANT: publish an updated MANIFEST+CURRENT that reflects + // the post-compaction live set before removing the old SST from cloud. Without this, a + // crash between the SST delete and the next metadata-sync attempt leaves a cloud + // MANIFEST that references an object that no longer exists, making recovery impossible. + // + // We use syncMetadataSnapshotInline (no MemTable flush) because: + // (a) this callback fires on a RocksDB compaction thread — calling flushSession(wait=true) + // here would deadlock against RocksDB background execution; + // (b) a flush is not needed: by the time onTableFileDeleted fires the compaction is + // complete, new SSTs are already on disk and uploaded, and the MANIFEST already + // reflects the post-compaction state. + if (!syncMetadataSnapshotInline(provider, dbName)) { + log.warn("Delete skipped (metadata sync failed, MANIFEST not yet updated): " + + "db={}, filePath={}", dbName, filePath); + return; + } + + String remoteKey = toRelativeKey(filePath); + try { + provider.deleteFile(remoteKey); + syncTracker.clearConfirmed(dbName, filePath); + log.debug("Cloud delete success: db={}, cf={}, path={}", dbName, cfName, filePath); + } catch (Exception e) { + // Non-fatal: log and continue. + log.error("Cloud delete failed: db={}, cf={}, path={}", dbName, cfName, filePath, e); + } + } + + /** + * Ensures every live SST file of {@code dbName} is confirmed present in cloud, uploading any + * that are not yet confirmed. Returns {@code true} only when the entire live set is durable. + * + * <h4>Fast path (common case)</h4> + * {@link #preHydrateDbFiles} seeds {@link #syncTracker} from the cloud listing on startup, + * and {@link #onTableFileCreated} marks every successful upload confirmed. By the time + * {@link #onTableFileDeleted} fires the entire live set is normally already confirmed, so + * {@link CloudSyncTracker#allConfirmed} returns {@code true} after one lock acquisition with + * zero cloud I/O — regardless of live-set size. + * + * <h4>Why we must check the whole live set, not just the deleted {@code filePath}</h4> + * {@code filePath} is the <em>old</em> compaction input being removed from cloud. Its bit + * being set in the bitmap only confirms it was uploaded previously — it says nothing about + * whether the <em>compaction outputs</em> (the replacement files, which form the new live + * set) are also in cloud. Deleting the input before the outputs are durable would leave + * data irretrievably lost on a node crash. + * + * <h4>Slow path (rare)</h4> + * Entered only when {@code allConfirmed} returns {@code false} (e.g. a previous upload + * failed). Files present locally are uploaded directly without a {@code fileExists} probe + * (idempotent PUT). Files absent locally and not confirmed are treated as not-yet-durable + * and the delete is held. + */ + private boolean ensureLiveSetUploaded(CloudStorageProvider provider, String dbName) { + return ensureLiveSetUploaded(provider, dbName, + RocksDBFactory.getInstance().getLiveSstFiles(dbName)); + } + + /** Testable seam: caller supplies the live-file set instead of reading the RocksDB singleton. */ + boolean ensureLiveSetUploaded(CloudStorageProvider provider, String dbName, + List<LiveSstFile> liveFiles) { + // Fast path: single lock acquisition, zero cloud I/O — the common case. + if (syncTracker.allConfirmed(dbName, liveFiles)) { + return true; + } + + // Slow path: one or more live files not yet confirmed — find and upload them. + log.info("Checking live set durability: db={}, liveFileCount={}", dbName, liveFiles.size()); + + java.util.List<String> unconfirmedFiles = new java.util.ArrayList<>(); + boolean allDurable = true; + + for (LiveSstFile live : liveFiles) { + String localPath = live.getAbsolutePath(); + if (syncTracker.isConfirmed(dbName, localPath)) { + continue; + } + unconfirmedFiles.add(localPath); + Path localFile = Paths.get(localPath); + String remoteKey = toRelativeKey(localPath); + if (!Files.exists(localFile)) { + // File absent locally and absent from bitmap. Startup hydration (preHydrateDbFiles) + // seeds the bitmap from the cloud listing, so reaching here means this file is + // genuinely not durable in cloud — hold the delete. + allDurable = false; + log.warn("Delete guard: live file absent locally and not confirmed in cloud: " + + "db={}, filePath={}", dbName, localPath); + continue; + } + try { + // Upload directly — no fileExists probe (idempotent PUT is cheaper than a probe). + provider.uploadFile(localPath, remoteKey); + syncTracker.markConfirmed(dbName, localPath); + } catch (Exception e) { + allDurable = false; + log.warn("Delete guard: failed to upload live file: db={}, filePath={}, error={}", + dbName, localPath, e.getMessage()); + if (retryQueue != null) { + retryQueue.submit(dbName, live.getCfName(), localPath, remoteKey, e); + } + } + } + + if (!unconfirmedFiles.isEmpty()) { + CloudStorageMetrics.incrementDeleteGuardReupload(dbName); + if (allDurable) { + log.info("Re-uploaded {} previously unconfirmed live files (delete proceeding): " + + "db={}, files={}", unconfirmedFiles.size(), dbName, unconfirmedFiles); + } else { + log.warn("Re-upload attempted for {} unconfirmed live files (some still not durable, " + + "delete held): db={}, files={}", unconfirmedFiles.size(), dbName, + unconfirmedFiles); + } + } + + return allDurable; + } + + /** + * A compaction changed this DB's live SST set; mirror metadata immediately (event-driven). + * + * <p>Note: RocksDB delivers the DB <em>directory path</em> here (from {@code db.getName()}), + * not the logical name, so we resolve it against {@link #dbNameByDir}. + */ + @Override + public void onCompacted(String dbNameOrPath) { + CloudStorageProvider provider = CloudStorageProviderFactory.getActiveProvider(); + if (provider == null) { + return; + } + String normalised = Paths.get(dbNameOrPath).toAbsolutePath().normalize().toString(); + String dbName = dbNameByDir.getOrDefault(normalised, dbNameOrPath); + // Skip metadata sync during truncation or grace period to allow purge to complete cleanly + if (!truncatingDbs.contains(dbName) && !isInTruncationGracePeriod(dbName)) { + syncMetadataSnapshotInline(provider, dbName); + } + } + + // ----------------------------------------------------------------------- + // Metadata mirroring + // ----------------------------------------------------------------------- + + /** Records the resolved DB directory for a logical DB name (idempotent). */ + private void recordDb(String dbName, String dbDir) { + if (dbName == null || dbDir == null) { + return; + } + String normalised = Paths.get(dbDir).toAbsolutePath().normalize().toString(); + dbNameByDir.put(normalised, dbName); + } + + /** Absolute, normalised parent directory of a file path (the DB dir of an SST). */ + private String parentDir(String filePath) { + Path parent = Paths.get(filePath).toAbsolutePath().normalize().getParent(); + return parent == null ? null : parent.toString(); + } + + + /** + * Captures a consistent metadata snapshot and mirrors it to cloud without first flushing the + * MemTable. Safe to call from any thread, including a RocksDB event/compaction callback. + * + * <p>By the time a compaction event fires, the MANIFEST already reflects the post-compaction + * live SST set — a checkpoint captures that consistent state without needing a flush. In + * {@code wal} mode the WAL tail is included in the checkpoint, so un-flushed data is still + * recoverable. In {@code flush} mode un-flushed MemTable data written since the last sync is + * not captured, but that is acceptable here because the goal is to publish a MANIFEST that + * does not reference any cloud object that is about to be deleted. + * + * <p>Package-private for testability. + */ + boolean syncMetadataSnapshotInline(CloudStorageProvider provider, String dbName) { Review Comment: Addressed ########## hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/AppConfig.java: ########## @@ -116,6 +145,119 @@ public void init() { "0".equals(rocksdb.get("write_buffer_size"))) { rocksdb.put("write_buffer_size", Long.toString(totalMemory / 1000)); } + + // ---- Cloud storage initialization ---- + initCloudStorage(); + } + + /** + * Initialises the cloud storage provider (if enabled) and registers + * {@link CloudStorageEventListener} with {@link RocksDBFactory} so that + * SST file creation/deletion events are forwarded to cloud storage. + * + * <p>The resolved absolute data-path is passed to the listener so that + * S3 object keys are relative to the store's storage root rather than + * being full container-specific absolute paths. + */ + private void initCloudStorage() { + CloudStorageConfig cfg = cloudStorageSpringConfig.toCloudStorageConfig(); + if (!cfg.isEnabled()) { + log.info("Cloud storage disabled (cloud.storage.enabled=false)"); + return; + } + try { + CloudStorageProviderFactory.initialize(cfg); + String resolvedDataRoot = + Paths.get(dataPath).toAbsolutePath().normalize().toString(); + + // Shared sync tracker: the listener's delete guard and the retry queue's success + // callback both update it, so a superseded cloud object is deleted only once every + // live SST file of that DB is confirmed present in cloud. + CloudSyncTracker syncTracker = new CloudSyncTracker(); + + CloudUploadRetryQueue retryQueue = new CloudUploadRetryQueue( + cfg.getUploadRetryMaxAttempts(), + cfg.getUploadRetryInitialDelayMs(), + cfg.getUploadRetryMaxDelayMs(), + resolvedDataRoot, + syncTracker::markConfirmed); + this.cloudUploadRetryQueue = retryQueue; + + String storeScopePrefix = buildCloudStoreScopePrefix(); + + CloudStorageEventListener listener = new CloudStorageEventListener( + resolvedDataRoot, + cfg.isStartupHydrationEnabled(), + cfg.getReadMissGuardWindowMs(), + retryQueue, + syncTracker, + cfg.getUploadBackpressureHighWatermark(), + "wal".equalsIgnoreCase(cfg.getWalMode()), + storeScopePrefix); + + // Initialize metrics if MeterRegistry is available + if (meterRegistry != null) { + CloudStorageMetrics.init(meterRegistry, syncTracker); + } + + RocksDBFactory.getInstance().addRocksdbChangedListener(listener); + log.info("Cloud storage provider '{}' registered with RocksDBFactory " + + "(dataRoot='{}', storeScopePrefix='{}', startupHydration={}, " + + "readMissHydration=true, " + + "readMissGuardWindowMs={}, uploadRetryMaxAttempts={}, " + + "uploadRetryInitialDelayMs={}, uploadRetryMaxDelayMs={})", + cfg.getProvider(), resolvedDataRoot, storeScopePrefix, + cfg.isStartupHydrationEnabled(), + cfg.getReadMissGuardWindowMs(), + cfg.getUploadRetryMaxAttempts(), + cfg.getUploadRetryInitialDelayMs(), + cfg.getUploadRetryMaxDelayMs()); + } catch (Exception e) { 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]
