bitflicker64 commented on code in PR #3081:
URL: https://github.com/apache/hugegraph/pull/3081#discussion_r3910386853
##########
.licenserc.yaml:
##########
@@ -66,6 +66,7 @@ header: # `header` section is configurations for source codes
license header.
- '**/*.log'
- '**/*.txt'
- '**/*.json'
+ - '**/*.ipynb'
Review Comment:
⚠️ This repo-wide header exemption exists to land one file that isn't part
of the feature.
The only `.ipynb` in the PR is
`docker/cloud-storage/notebooks/aml_sql_showcase_hugegraph.ipynb` — 1720 lines,
43 cells of anti-money-laundering Gremlin traversals that download
`LI-Large_Trans.csv` / `HI-Large_accounts.csv` from Kaggle. It drives the
Gremlin/REST API; it never touches SST offload, hydration or the cloud
provider. It's also absent from the PR description: "Infrastructure & Testing"
commits to the MinIO compose stack and `test-graph-queries-and-sst.sh`, both of
which are here and both of which do cover the end-to-end durability check.
Drop the notebook and revert this line. As written it permanently exempts
every future notebook in the tree from the ASF header check, which is a real
cost paid for a demo that belongs in hugegraph-doc or a gist.
##########
hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/AppConfig.java:
##########
@@ -312,6 +636,113 @@ public class RocksdbConfig {
private Map<String, String> rocksdb = new HashMap<>();
}
+ /**
+ * Spring {@link ConfigurationProperties} wrapper for the common cloud
storage properties.
+ *
+ * <p>Provider-specific keys (e.g. {@code cloud.storage.s3.bucket}) are
read
+ * directly from the Spring {@link Environment} at conversion time, so
this class
+ * has zero knowledge of any specific cloud provider.
+ *
+ * <pre>
+ * cloud:
+ * storage:
+ * enabled: true
+ * provider: s3 # selects which sub-namespace to forward to
the provider
+ * path-prefix: hugegraph
+ * s3: # all keys here are forwarded as-is to the S3
provider
+ * bucket: my-bucket
+ * region: us-east-1
+ * access-key: ${AWS_ACCESS_KEY_ID}
+ * secret-key: ${AWS_SECRET_ACCESS_KEY}
+ * </pre>
+ */
+ @Data
+ @Configuration
+ @ConfigurationProperties(prefix = "cloud.storage")
+ public class CloudStorageSpringConfig {
+
+ private boolean enabled = false;
+ private String provider = "s3";
+ private String pathPrefix = "hugegraph";
+ private boolean startupHydrationEnabled = true;
+ private long readMissGuardWindowMs = 3000L;
+ // Whole-file upload retries after a first failure. Default 3 under
the primary-durability
+ private int uploadRetryMaxAttempts = 3;
+ private long uploadRetryInitialDelayMs = 1_000L;
+ private long uploadRetryMaxDelayMs = 60_000L;
+ // Backpressure high-watermark on the pending-upload backlog; 0
(default) disables it.
+ // Opt-in: when > 0 the throttle parks RocksDB's flush/compaction
thread (up to 30s/event),
+ // which under a sustained cloud outage can stall memtable flushes /
stop writes.
+ private int uploadBackpressureHighWatermark = 0;
+ // Max DLQ entries before oldest are evicted (bounds memory/disk under
a prolonged outage).
+ private int dlqMaxSize = 100_000;
+ // Debounce window (ms) for the per-SST metadata sync; <= 0 disables
debouncing.
+ private long metadataSyncDebounceMs = 1_000L;
+ // Force a metadata publish once this many SST uploads accumulate
unmirrored; <= 0 disables.
+ private int metadataSyncMaxUnpublished = 32;
+ // Stable per-node identity for the cloud key scope. Blank =>
persisted-in-data-dir scope
+ // (seeded from network address). Set for guaranteed recovery after IP
drift / disk loss.
+ private String nodeId = "";
+
+ /**
+ * Injected by Spring; used to read {@code cloud.storage.<provider>.*}
properties
+ * without coupling this class to any specific provider.
+ */
+ @Autowired
+ @EqualsAndHashCode.Exclude
+ @ToString.Exclude
+ private Environment environment;
+
+ /** Converts this Spring-bound config into a plain {@link
CloudStorageConfig} POJO. */
+ public CloudStorageConfig toCloudStorageConfig() {
+ CloudStorageConfig cfg = new CloudStorageConfig();
+ cfg.setEnabled(enabled);
+ cfg.setProvider(provider);
+ cfg.setPathPrefix(pathPrefix);
+ cfg.setStartupHydrationEnabled(startupHydrationEnabled);
+ cfg.setReadMissGuardWindowMs(readMissGuardWindowMs);
+ cfg.setUploadRetryMaxAttempts(uploadRetryMaxAttempts);
+ cfg.setUploadRetryInitialDelayMs(uploadRetryInitialDelayMs);
+ cfg.setUploadRetryMaxDelayMs(uploadRetryMaxDelayMs);
+
cfg.setUploadBackpressureHighWatermark(uploadBackpressureHighWatermark);
+ cfg.setDlqMaxSize(dlqMaxSize);
+ cfg.setMetadataSyncDebounceMs(metadataSyncDebounceMs);
+ cfg.setMetadataSyncMaxUnpublished(metadataSyncMaxUnpublished);
+ cfg.setNodeId(nodeId);
+ cfg.setProviderProperties(readProviderProperties());
+ return cfg;
+ }
+
+ /**
+ * Reads all {@code cloud.storage.<provider>.*} keys from the Spring
Environment
+ * and returns them as a flat map with the provider sub-prefix
stripped.
+ *
+ * <p>For example, with {@code provider=s3}, the YAML key
+ * {@code cloud.storage.s3.bucket} becomes {@code bucket} in the
returned map.
+ */
+ private Map<String, String> readProviderProperties() {
Review Comment:
⚠️ Spring Boot already does this in one call.
This method casts to `AbstractEnvironment`, streams `getPropertySources()`,
filters to `EnumerablePropertySource`, flat-maps `getPropertyNames()`,
string-prefix-filters, de-dupes and re-reads each key. `hg-store-node` is on
spring-boot 2.5.14, so:
```java
Map<String, String> props = Binder.get(environment)
.bind("cloud.storage." + provider, Bindable.mapOf(String.class,
String.class))
.orElseGet(Collections::emptyMap);
```
Same kebab-case keys the `S3CloudStorageConfig.KEY_*` constants expect
(`bucket`, `access-key`, ...), about 25 lines down to 3, and the
`AbstractEnvironment` / `EnumerablePropertySource` imports go with it.
Worth noting it isn't only shorter: `Binder` applies relaxed binding, so
`CLOUD_STORAGE_S3_SECRET_KEY` and friends resolve. A raw
`key.startsWith("cloud.storage.s3.")` scan over property *names* won't match
environment-variable-style names, which is exactly how credentials tend to
arrive in the container deployment this PR ships.
##########
hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/cloud/CloudStorageEventListener.java:
##########
@@ -0,0 +1,3150 @@
+/*
+ * 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.concurrent.locks.LockSupport;
+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;
+
+ /**
+ * When {@code > 0}, {@link #onTableFileCreated} slows the RocksDB
flush/compaction thread while
+ * the pending-upload backlog exceeds this watermark, so ingestion cannot
outrun the cloud
+ * mirror. The backlog is the executor's queued/active uploads + the retry
queue's in-flight
+ * retries + a bounded DLQ enqueue rate (see {@link
#dlqEnqueueRateBacklog()}). {@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;
+
+ /**
+ * Window over which the DLQ enqueue rate (uploads that exhausted their
retries and became
+ * local-only) is measured for backpressure. The count of DLQ enqueues
observed in the trailing
+ * window is added — capped at {@link #backpressureHighWatermark} — to the
backpressure backlog,
+ * so a sustained cloud outage that keeps pushing uploads to the DLQ
throttles ingestion, while a
+ * static post-recovery DLQ (rate 0) does not.
+ */
+ private static final long DLQ_ENQUEUE_RATE_WINDOW_MS = 1_000L;
+
+ /** Guards the DLQ enqueue-rate sample below (touched from every
backpressure poll). */
+ private final Object dlqRateLock = new Object();
+ /** Wall-clock time of the last DLQ enqueue-rate sample; {@code 0} until
first primed. */
+ private long lastDlqRateSampleMs = 0L;
+ /** {@link CloudUploadRetryQueue#getDlqEnqueuedTotal()} captured at the
last sample. */
+ private long lastDlqEnqueuedTotalAtSample = 0L;
+ /** Cached bounded DLQ enqueue-rate contribution to the backpressure
backlog. */
+ private int dlqRateBacklogContribution = 0;
+
+ /**
+ * 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) {
+ this(dataRoots, true, DEFAULT_READ_MISS_GUARD_WINDOW_MS, null);
+ }
+
+ /**
+ * @param dataRoots configured store data roots
+ */
+ public CloudStorageEventListener(List<String> dataRoots,
+ boolean startupHydrationEnabled) {
+ this(dataRoots, startupHydrationEnabled,
DEFAULT_READ_MISS_GUARD_WINDOW_MS, null);
+ }
+
+ /**
+ * @param dataRoots configured store data roots
+ * @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(List<String> dataRoots,
+ boolean startupHydrationEnabled,
+ long readMissGuardWindowMs) {
+ this(dataRoots, startupHydrationEnabled, readMissGuardWindowMs, null);
+ }
+
+ /**
+ * @param dataRoots configured store data roots
+ * @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(List<String> dataRoots,
+ boolean startupHydrationEnabled,
+ long readMissGuardWindowMs,
+ CloudUploadRetryQueue retryQueue) {
+ this(dataRoots, startupHydrationEnabled, readMissGuardWindowMs,
retryQueue,
+ new CloudSyncTracker(), 0);
+ }
+
+ /**
+ * @param dataRoots configured store data roots
+ * @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.
+ * @param backpressureHighWatermark {@code > 0} to slow ingestion while
pending-upload backlog
+ * exceeds this value; {@code 0} disables
backpressure.
+ */
+ public CloudStorageEventListener(List<String> dataRoots,
+ boolean startupHydrationEnabled,
+ long readMissGuardWindowMs,
+ CloudUploadRetryQueue retryQueue,
+ CloudSyncTracker syncTracker,
+ int backpressureHighWatermark) {
+ this(dataRoots, startupHydrationEnabled, readMissGuardWindowMs,
retryQueue, syncTracker,
+ backpressureHighWatermark, null);
+ }
+
+ /**
+ * Multi-root constructor for comma-separated app.data-path configuration.
+ *
+ * @param dataRoots configured store data roots (absolute, normalised)
+ * @param storeScopePrefix optional per-store key prefix to isolate cloud
objects
+ */
+ public CloudStorageEventListener(List<String> dataRoots,
+ boolean startupHydrationEnabled,
+ long readMissGuardWindowMs,
+ CloudUploadRetryQueue retryQueue,
+ CloudSyncTracker syncTracker,
+ int backpressureHighWatermark,
+ String storeScopePrefix) {
+ this(dataRoots, startupHydrationEnabled, readMissGuardWindowMs,
retryQueue, syncTracker,
+ backpressureHighWatermark, storeScopePrefix, Tuning.defaults());
+ }
+
+ /**
+ * Fully-parameterised constructor. Prefer this in production wiring:
passing {@link Tuning}
+ * makes the listener 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 storeScopePrefix optional per-store key prefix to isolate cloud
objects
+ * @param tuning debounce / backlog-bound tuning (never {@code null}; use
+ * {@link Tuning#defaults()} for defaults)
+ */
+ public CloudStorageEventListener(List<String> dataRoots,
+ boolean startupHydrationEnabled,
+ long readMissGuardWindowMs,
+ CloudUploadRetryQueue retryQueue,
+ CloudSyncTracker syncTracker,
+ int backpressureHighWatermark,
+ String storeScopePrefix,
+ Tuning tuning) {
+ // 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.backpressureHighWatermark = Math.max(0,
backpressureHighWatermark);
+ this.storeScopePrefix = normaliseKeyPrefix(storeScopePrefix);
+ this.uploadExecutor = sharedUploadExecutor();
+
+ Tuning t = tuning != null ? tuning : Tuning.defaults();
+ this.metadataSyncDebounceMs = t.metadataSyncDebounceMs;
+ this.metadataSyncMaxUnpublished = t.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();
+ }
+
+ /**
+ * Immutable tuning bundle for the listener's post-upload metadata sync
behaviour. Grouping these
+ * into one params object keeps the constructor readable and lets the
listener be fully configured
+ * at construction time. Build with {@link #builder()}; unset knobs fall
back to the documented
+ * defaults.
+ */
+ public static final class Tuning {
+
+ private final long metadataSyncDebounceMs;
+ private final int metadataSyncMaxUnpublished;
+
+ private Tuning(Builder b) {
+ this.metadataSyncDebounceMs = b.metadataSyncDebounceMs;
+ this.metadataSyncMaxUnpublished = b.metadataSyncMaxUnpublished;
+ }
+
+ /** Tuning with all defaults (equivalent to constructing the listener
with no {@code set*}). */
+ public static Tuning defaults() {
+ return builder().build();
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static final class Builder {
+
+ private long metadataSyncDebounceMs =
DEFAULT_METADATA_SYNC_DEBOUNCE_MS;
+ private int metadataSyncMaxUnpublished =
DEFAULT_METADATA_SYNC_MAX_UNPUBLISHED;
+
+ /** @see CloudStorageEventListener#setMetadataSyncDebounceMs(long)
*/
+ public Builder metadataSyncDebounceMs(long ms) {
+ this.metadataSyncDebounceMs = ms;
+ return this;
+ }
+
+ /** @see
CloudStorageEventListener#setMetadataSyncMaxUnpublished(int) */
+ public Builder metadataSyncMaxUnpublished(int maxUnpublished) {
+ this.metadataSyncMaxUnpublished = maxUnpublished;
+ return this;
+ }
+
+ public Tuning build() {
+ return new Tuning(this);
+ }
+ }
+ }
+
+ 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 {
+ Path dir = pendingTruncateDir();
+ Files.createDirectories(dir);
+ Path marker = pendingTruncateMarkerPath(prefix);
+ 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);
+ }
+ try (FileChannel dirCh = FileChannel.open(dir,
StandardOpenOption.READ)) {
+ dirCh.force(true);
+ } catch (IOException e) {
+ if (WINDOWS_OS) {
+ log.debug("Directory fsync of {} not supported on this
platform: {}",
+ dir, e.getMessage());
+ } else {
+ throw e;
+ }
+ }
+ log.debug("Cloud pending-truncate marker persisted (fsync'd): db={},
prefix={}",
+ dbName, prefix);
+ }
+
+ private void removePendingDeleteMarker(String prefix) {
+ try {
+ Files.deleteIfExists(pendingDeleteMarkerPath(prefix));
+ } catch (IOException e) {
+ log.debug("Failed to remove pending-delete marker for prefix={}:
{}",
+ prefix, e.getMessage());
+ }
+ }
+
+ /** Whether a local pending-remote-delete marker exists for {@code
prefix}. */
+ boolean hasPendingDeleteMarker(String prefix) {
+ return Files.exists(pendingDeleteMarkerPath(prefix));
+ }
+
+ private void removePendingTruncateMarker(String prefix) {
+ try {
+ Files.deleteIfExists(pendingTruncateMarkerPath(prefix));
+ } catch (IOException e) {
+ log.debug("Failed to remove pending-truncate marker for prefix={}:
{}",
+ prefix, e.getMessage());
+ }
+ }
+
+ /**
+ * Completes a pending remote delete under a per-prefix lock: if the
marker is still present,
+ * ensure the tombstone exists, purge the prefix, and on success drop the
tombstone + marker.
+ * Returns {@code true} when cleanup is confirmed complete (marker absent
after this call).
+ *
+ * <p>The lock + marker re-check make the purge run at most once and ONLY
while the marker is
+ * present. A re-created DB's {@link #onDBOpening} calls this before
returning (so the open
+ * blocks until cleanup finishes), and its uploads happen only afterwards
— so the async retry,
+ * which also takes the lock and skips when the marker is gone, can never
purge freshly-written
+ * data for the re-created generation.
+ */
+ private boolean tryCompletePendingDelete(CloudStorageProvider provider,
String dbName,
+ String prefix) {
+ Object lock = pendingDeleteLocks.computeIfAbsent(prefix, k -> new
Object());
+ synchronized (lock) {
+ if (!hasPendingDeleteMarker(prefix)) {
+ return true; // already cleaned up by another caller
+ }
+ ensureTombstonePresent(provider, dbName, prefix);
+ if (purgeRemotePrefix(provider, dbName, prefix)) {
+ deleteTombstoneBestEffort(provider, dbName, prefix);
+ removePendingDeleteMarker(prefix);
+ pendingDeleteLocks.remove(prefix);
+ log.info("Cloud pending-delete cleanup completed for db={}
(prefix={})",
+ dbName, prefix);
+ return true;
+ }
+ return false;
+ }
+ }
+
+ /**
+ * Retries the remote purge for a deleted DB with capped exponential
backoff — the same safety
+ * level as the truncate purge. A {@code null} provider is treated as a
failed attempt and
+ * rescheduled (transient reconfiguration window) rather than abandoned.
While retries are
+ * pending, the local pending-delete marker keeps {@link #onDBOpening}
from re-hydrating stale
+ * data. After {@link #MAX_DELETE_PURGE_RETRIES} failures it logs loudly
for operator action; the
+ * marker and tombstone remain, so re-hydration stays blocked until manual
cleanup.
+ */
+ private void scheduleDeletePurgeRetry(String dbName, String prefix, int
attempt) {
+ if (attempt > MAX_DELETE_PURGE_RETRIES) {
+ log.error("Cloud DB delete purge for db={} still failing after {}
attempt(s) — remote "
+ + "prefix '{}' may retain stale objects. The local
pending-delete marker and "
+ + "tombstone remain (re-hydration stays blocked); manual
cleanup is required.",
+ dbName, MAX_DELETE_PURGE_RETRIES, prefix);
+ return;
+ }
+ long delay = Math.min(DELETE_PURGE_RETRY_BASE_MS * (1L << (attempt -
1)), 8_000L);
+ ScheduledExecutorService scheduler = metadataSyncScheduler();
+ if (scheduler == null) {
+ log.error("Cloud DB delete purge retry for db={} cannot be
scheduled (subsystem shutting "
+ + "down) — local marker preserved; remote prefix '{}'
may retain stale "
+ + "objects until next startup.", dbName, prefix);
+ return;
+ }
+ try {
+ scheduler.schedule(() -> {
+ CloudStorageProvider provider =
CloudStorageProviderFactory.getActiveProvider();
+ if (provider == null) {
+ log.warn("Cloud DB delete purge retry for db={} found no
active provider "
+ + "(attempt {}); rescheduling.", dbName, attempt);
+ scheduleDeletePurgeRetry(dbName, prefix, attempt + 1);
+ return;
+ }
+ // tryCompletePendingDelete is a no-op (returns true) if the
marker was already
+ // cleared (e.g. by an inline open-time cleanup), so the retry
never purges a prefix
+ // a re-created DB may have taken over.
+ if (!tryCompletePendingDelete(provider, dbName, prefix)) {
+ scheduleDeletePurgeRetry(dbName, prefix, attempt + 1);
+ }
+ }, delay, TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ log.error("Cloud DB delete purge retry could not be scheduled for
db={} (scheduler "
+ + "shutting down) — local marker preserved; remote
prefix '{}' may retain "
+ + "stale objects.", dbName, prefix);
+ }
+ }
+
+ /** Writes the delete-tombstone if it is not already present
(best-effort). */
+ private void ensureTombstonePresent(CloudStorageProvider provider, String
dbName, String prefix) {
+ String tombstoneKey = prefix + DB_TOMBSTONE_SUFFIX;
+ Path tmp = null;
+ try {
+ if (provider.fileExists(tombstoneKey)) {
+ return;
+ }
+ 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 (deferred): db={}, key={}",
dbName, tombstoneKey);
+ } catch (Exception e) {
+ log.warn("Deferred tombstone write failed for db={}, key={}: {}",
+ dbName, tombstoneKey, e.getMessage());
+ } finally {
+ if (tmp != null) {
+ try {
+ Files.deleteIfExists(tmp);
+ } catch (IOException ignore) {
+ // best-effort temp cleanup
+ }
+ }
+ }
+ }
+
+ /** On startup, schedule a bounded purge retry for each leftover
pending-delete marker. */
+ private void processPendingDeleteMarkersOnStartup() {
+ Path dir = pendingDeleteDir();
+ if (!Files.isDirectory(dir)) {
+ return;
+ }
+ try (Stream<Path> markers = Files.list(dir)) {
+ markers.forEach(marker -> {
+ String encoded = marker.getFileName() == null ? "" :
marker.getFileName().toString();
+ String prefix;
+ try {
+ prefix = Files.readString(marker).trim();
+ } catch (IOException e) {
+ log.warn("Failed to read pending-delete marker {}: {}",
marker, e.getMessage());
+ return;
+ }
+ if (prefix.isEmpty()) {
+ return;
+ }
+ String decoded = decodePendingDeleteMarkerName(encoded);
+ if (decoded == null || !decoded.equals(prefix)) {
+ log.error("Cloud startup: ignoring invalid pending-delete
marker {} (payload does "
+ + "not match encoded prefix)", marker);
+ return;
+ }
+ if (isPendingDeletePrefixInScope(prefix)) {
+ log.error("Cloud startup: ignoring out-of-scope
pending-delete marker {} for "
+ + "prefix='{}'", marker, prefix);
+ return;
+ }
+ String inferredDbName = inferDbNameFromPrefix(prefix);
+ log.warn("Cloud startup: found pending remote-delete marker
for prefix={} — "
+ + "scheduling deferred purge to prevent stale-data
re-hydration.", prefix);
+ scheduleDeletePurgeRetry(inferredDbName, prefix, 1);
+ });
+ } catch (IOException e) {
+ log.warn("Failed to scan pending-delete markers in {}: {}", dir,
e.getMessage());
+ }
+ }
+
+ /** On startup, schedule retries for leftover pending-truncate markers. */
+ private void processPendingTruncateMarkersOnStartup() {
+ Path dir = pendingTruncateDir();
+ if (!Files.isDirectory(dir)) {
+ return;
+ }
+ try (Stream<Path> markers = Files.list(dir)) {
+ markers.forEach(marker -> {
+ String encoded = marker.getFileName() == null ? "" :
marker.getFileName().toString();
+ String prefix;
+ try {
+ prefix = Files.readString(marker).trim();
+ } catch (IOException e) {
+ log.warn("Failed to read pending-truncate marker {}: {}",
+ marker, e.getMessage());
+ return;
+ }
+ if (prefix.isEmpty()) {
+ return;
+ }
+ String decoded = decodePendingDeleteMarkerName(encoded);
+ if (decoded == null || !decoded.equals(prefix)) {
+ log.error("Cloud startup: ignoring invalid
pending-truncate marker {} "
+ + "(payload does not match encoded prefix)",
marker);
+ return;
+ }
+ if (isPendingDeletePrefixInScope(prefix)) {
+ log.error("Cloud startup: ignoring out-of-scope
pending-truncate marker {} "
+ + "for prefix='{}'", marker, prefix);
+ return;
+ }
+ String inferredDbName = inferDbNameFromPrefix(prefix);
+ log.warn("Cloud startup: found pending truncate marker for
prefix={} — scheduling "
+ + "deferred purge retry.", prefix);
+ scheduleTruncatePurgeRetry(inferredDbName, prefix);
+ });
+ } catch (IOException e) {
+ log.warn("Failed to scan pending-truncate markers in {}: {}", dir,
e.getMessage());
+ }
+ }
+
+ /** Decodes the marker filename (Base64 URL, no padding) back to the
original remote prefix. */
+ private static String decodePendingDeleteMarkerName(String encoded) {
+ if (encoded == null || encoded.isEmpty()) {
+ return null;
+ }
+ try {
+ byte[] bytes = java.util.Base64.getUrlDecoder().decode(encoded);
+ return new String(bytes, java.nio.charset.StandardCharsets.UTF_8);
+ } catch (IllegalArgumentException e) {
+ return null;
+ }
+ }
+
+ /** Validates that a pending-delete prefix is relative, normalized, and in
this listener's scope. */
+ private boolean isPendingDeletePrefixInScope(String prefix) {
+ if (prefix == null || prefix.isEmpty()) {
+ return true;
+ }
+ String normalized = prefix.replace('\\', '/');
+ if (normalized.startsWith("/") || normalized.contains("//")) {
+ return true;
+ }
+ String[] parts = normalized.split("/");
+ for (String p : parts) {
+ if (p.isEmpty() || ".".equals(p) || "..".equals(p)) {
+ return true;
+ }
+ }
+ if (storeScopePrefix.isEmpty()) {
+ return false;
+ }
+ String scopedPrefix = storeScopePrefix + "/";
+ return !normalized.equals(storeScopePrefix) &&
!normalized.startsWith(scopedPrefix);
+ }
+
+ /** Best-effort DB-name inference for startup marker logs/scheduling. */
+ private String inferDbNameFromPrefix(String prefix) {
+ if (prefix == null || prefix.isEmpty()) {
+ return prefix;
+ }
+ try {
+ String stripped = stripStoreScope(prefix);
+ return stripped.isEmpty() ? prefix : stripped;
+ } catch (IllegalArgumentException e) {
+ return prefix;
+ }
+ }
+
+ /**
+ * A truncate failed partway (drop/create threw after {@link
#onDBTruncateBegin}). Clear the
+ * "truncating" suppression so cloud metadata mirroring, object deletion,
and the delete guard
+ * resume immediately — but do NOT purge remote state: the data intended
for clearing may still
+ * be present locally and must remain recoverable from cloud.
+ */
+ @Override
+ public void onDBTruncateAbort(String dbName, String dbPath) {
+ truncatingDbs.remove(dbName);
+ // Drop the begin-timestamp so the (successful-truncate) grace period
does not suppress
+ // mirroring for a truncate that never completed.
+ truncationTimes.remove(dbName);
+ log.warn("Cloud truncate aborted for db={} — cleared 'truncating'
suppression without "
+ + "purging remote state (data may still be present locally)",
dbName);
+ }
+
+ /**
+ * Upper bound on how long a truncate may legitimately stay "in progress".
A truncate is a
+ * drop+recreate of column families and completes in well under this
window; anything longer
+ * means the completion callback ({@link #onDBTruncated}) never fired.
+ */
+ private static final long MAX_TRUNCATION_DURATION_MS = 60_000L;
+
+ /** Effective max-truncation window; overridable in tests to avoid a 60 s
wait. */
+ private volatile long maxTruncationDurationMs = MAX_TRUNCATION_DURATION_MS;
+
+ /** Test seam: shrink the stale-latch window so self-healing can be
exercised quickly. */
+ void setMaxTruncationDurationMsForTest() {
+ this.maxTruncationDurationMs = 50L;
+ }
+
+ /**
+ * Returns {@code true} while a truncate is actively in progress for
{@code dbName}.
+ *
+ * <p><b>Self-healing latch.</b> {@code truncatingDbs} is set by {@link
#onDBTruncateBegin} and
+ * normally cleared by {@link #onDBTruncated}. However {@code
RocksDBSession.truncate()} invokes
+ * {@code dropTables}/{@code createTables} (which throw the unchecked
{@code DBStoreException})
+ * between the begin and completion notifications without a {@code
finally}, so a failure there
+ * leaves the completion callback un-invoked and the latch stuck. A stuck
latch would silently
+ * disable metadata mirroring, cloud object deletion, and the delete guard
for that DB for the
+ * lifetime of the process. To bound that blast radius, treat the latch as
stale (and clear it)
+ * once the recorded truncation start is older than {@link
#MAX_TRUNCATION_DURATION_MS}.
+ */
+ boolean isActivelyTruncating(String dbName) {
+ if (!truncatingDbs.contains(dbName)) {
+ return false;
+ }
+ Long startedAt = truncationTimes.get(dbName);
+ if (startedAt != null
+ && System.currentTimeMillis() - startedAt >
maxTruncationDurationMs) {
+ truncatingDbs.remove(dbName);
+ log.warn("Cleared stale 'truncating' flag for db={} ({} ms elapsed
with no completion "
+ + "callback) — resuming cloud metadata mirroring and
cleanup", dbName,
+ System.currentTimeMillis() - startedAt);
+ return false;
+ }
+ return true;
+ }
+
+ /** Test seam exposing {@link #isInTruncationGracePeriod} for assertions.
*/
+ @SuppressWarnings("SameParameterValue")
+ boolean isInTruncationGracePeriodForTest(String dbName) {
+ return isInTruncationGracePeriod(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.
+ *
+ * @return {@code true} if the purge completed without error; {@code
false} if it failed.
+ * The caller uses the return value to decide whether it is safe
to remove the
+ * tombstone — the tombstone must survive any partial purge.
+ */
+ private boolean 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);
+ }
+ return true;
+ } catch (IOException e) {
+ log.warn("Cloud DB purge failed for db={}, prefix={}: {}",
+ dbName, prefix, e.getMessage());
+ return false;
+ }
+ }
+
+ /**
+ * Pins and asynchronously uploads the newly created SST file to the
active cloud
+ * storage provider.
+ *
+ * @param dbNameOrPath RocksDB instance name (partition id) or DB
directory path
+ * @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 dbNameOrPath, String cfName,
+ String filePath, long fileSize) {
+ String dbName = resolveDbName(dbNameOrPath, filePath);
+ recordDb(dbName, parentDir(filePath));
+ CloudStorageMetrics.registerDatabaseMetrics(dbName);
+ String remoteKey = toRelativeKey(filePath);
+
+ // Capture the epoch before we hand off to a background thread. The
async callback
+ // uses markConfirmedIfEpoch so a late confirmation after clearDb() +
DB recreation
+ // with reused file numbers is silently dropped instead of producing
stale durability state.
+ long uploadEpoch = syncTracker.currentEpoch(dbName);
+
+ CloudStorageProvider provider =
CloudStorageProviderFactory.getActiveProvider();
+ if (provider == null) {
+ CloudStorageMetrics.recordUploadFailure(dbName, cfName,
"NoActiveProvider");
+ if (retryQueue != null) {
+ try {
+ Path pinned = pinForAsyncUpload(filePath);
+ retryQueue.submitPinned(dbName, cfName, pinned.toString(),
filePath, remoteKey,
+ uploadEpoch,
+ new IOException("no active cloud
provider"));
+ log.warn("Cloud upload deferred due to missing provider:
db={}, cf={}, path={}"
+ + " (staged pin routed to retry queue)", dbName,
cfName, filePath);
+ } catch (Exception pinError) {
+ log.warn("Cloud upload deferred due to missing provider
and staging failed: "
+ + "db={}, cf={}, path={} — routing original SST
to retry queue: {}",
+ dbName, cfName, filePath, pinError.getMessage());
+ retryQueue.submit(dbName, cfName, filePath, remoteKey,
uploadEpoch,
+ new IOException("no active cloud
provider", pinError));
+ }
+ } else {
+ log.warn("Cloud upload skipped: no active provider and no
retry queue: db={}, cf={}, "
+ + "path={}", dbName, cfName, filePath);
+ }
+ applyBackpressure(dbName);
+ return;
+ }
+
+ 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) {
+ // Pass the captured epoch so a successful retry is confirmed
via the epoch-guarded
+ // callback; the plain submit() would use epoch 0 and be
silently dropped.
+ retryQueue.submit(dbName, cfName, filePath, remoteKey,
uploadEpoch, e);
+ }
+ applyBackpressure(dbName);
+ return;
+ }
+
+ try {
+ uploadExecutor.execute(new SstUploadTask(provider, dbName, cfName,
pinned, filePath,
+ remoteKey, uploadEpoch,
fileSize));
+ } catch (RejectedExecutionException e) {
+ // Queue is full. Keep the hard-link pin alive and submit IT to
the retry queue so
+ // the file survives even if RocksDB compacts the original SST
before the retry fires.
+ // Pass both the pinned path (for the actual upload) and the
original SST path (for
+ // confirmation and cleanup) so CloudSyncTracker can parse the
file number correctly.
+ CloudStorageMetrics.recordUploadFailure(dbName, cfName,
"UploadQueueFull");
+ log.error("Cloud upload dispatch rejected (queue full): db={},
cf={}, path={}",
+ dbName, cfName, filePath);
+ if (retryQueue != null) {
+ retryQueue.submitPinned(dbName, cfName, pinned.toString(),
filePath, remoteKey,
+ uploadEpoch,
+ new IOException("cloud upload dispatch
queue full", e));
+ } else {
+ // No retry queue — nothing more we can do; clean up the pin.
+ try {
+ Files.deleteIfExists(pinned);
+ } catch (IOException ioe) {
+ log.debug("Failed to cleanup staged upload file {}: {}",
pinned, ioe.getMessage());
+ }
+ }
+ }
+
+ // 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);
+ }
+
+ /**
+ * A single asynchronous SST upload, as a typed {@link Runnable} rather
than a lambda so that a
+ * forced {@code shutdownNow()} of the shared upload executor returns the
actual task objects
+ * (a {@link ThreadPoolExecutor} returns the exact runnables it had
queued). {@link
+ * #handoffOnShutdown()} then routes any never-run task into the retry
queue / DLQ so an SST that
+ * was accepted locally but not yet mirrored is not silently dropped from
the durability
+ * pipeline. {@code run()} preserves the original inline behaviour exactly.
+ */
+ private final class SstUploadTask implements Runnable {
+
+ private final CloudStorageProvider provider;
+ private final String dbName;
+ private final String cfName;
+ private final Path pinned;
+ private final String filePath;
+ private final String remoteKey;
+ private final long uploadEpoch;
+ private final long fileSize;
+
+ SstUploadTask(CloudStorageProvider provider, String dbName, String
cfName, Path pinned,
+ String filePath, String remoteKey, long uploadEpoch,
long fileSize) {
+ this.provider = provider;
+ this.dbName = dbName;
+ this.cfName = cfName;
+ this.pinned = pinned;
+ this.filePath = filePath;
+ this.remoteKey = remoteKey;
+ this.uploadEpoch = uploadEpoch;
+ this.fileSize = fileSize;
+ }
+
+ @Override
+ public void run() {
+ long startTimeMs = System.currentTimeMillis();
+ // When the upload fails and is handed to the retry queue,
OWNERSHIP of the pinned
+ // hard-link transfers to the retry/DLQ lifecycle — it must NOT be
deleted here, or a
+ // retry firing after RocksDB has compacted away the original SST
would find no source
+ // and be silently dropped (retry intent lost). We only delete the
pin when we still
+ // own it (successful upload, or no retry queue to hand off to).
+ boolean pinHandedOff = false;
+ try {
+ // ---- Upload proper: ONLY a failure here is an upload
failure that should retry. ----
+ try {
+ provider.uploadFile(pinned.toString(), remoteKey);
+ } 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) {
+ // Hand the SURVIVING pinned hard-link to the retry
queue (submitPinned), not
+ // the original SST path (plain submit): compaction
may delete the original
+ // before the retry fires, but the pin keeps the
byteset alive.
+ // sourceSstPath=filePath is still used for
epoch-guarded confirmation and
+ // staging cleanup on success.
+ retryQueue.submitPinned(dbName, cfName,
pinned.toString(), filePath,
+ remoteKey, uploadEpoch, e);
+ pinHandedOff = true;
+ }
+ return;
+ }
+ // ---- Post-upload: the upload SUCCEEDED. Errors below (e.g.
a null metadata-sync
+ // scheduler during a shutdown race) must NOT be reclassified
as an upload failure
+ // and re-uploaded/DLQ'd; the SST is already durable in cloud.
----
+ syncTracker.markConfirmedIfEpoch(dbName, filePath,
uploadEpoch);
+ long syncLatencyMs = System.currentTimeMillis() - startTimeMs;
+ CloudStorageMetrics.recordSyncLatency(dbName, syncLatencyMs);
+ try {
+ // Coalesce the per-SST metadata sync: at most one publish
per debounce window
+ // per DB, with a trailing sync. (Truncation/grace
suppression is handled inside
+ // requestDebouncedMetadataSync.)
+ requestDebouncedMetadataSync(provider, dbName);
+ } catch (Exception e) {
+ log.warn("Cloud upload succeeded but post-upload
metadata-sync scheduling "
+ + "failed (SST is durable; CURRENT/MANIFEST will
catch up on the next "
+ + "sync): db={}, path={}: {}", dbName, filePath,
e.getMessage());
+ }
+ log.debug("Cloud upload success: db={}, cf={}, path={},
size={}, latencyMs={}",
+ dbName, cfName, filePath, fileSize, syncLatencyMs);
+ } finally {
+ if (!pinHandedOff) {
+ try {
+ Files.deleteIfExists(pinned);
+ } catch (IOException e) {
+ log.debug("Failed to cleanup staged upload file {}:
{}",
+ pinned, e.getMessage());
+ }
+ }
+ }
+ }
+
+ /**
+ * Called for a task that was queued but never ran because the
executor was force-stopped at
+ * shutdown. Hands the still-pinned file to the retry queue so its
upload intent survives:
+ * the pin is kept alive (not deleted) so the retry/DLQ can upload it
even if RocksDB has
+ * since compacted the original SST away.
+ */
+ void handoffOnShutdown() {
+ if (retryQueue == null) {
+ // No retry queue: best-effort cleanup of the pin; nothing
else we can do.
+ try {
+ Files.deleteIfExists(pinned);
+ } catch (IOException ignore) {
+ // best-effort
+ }
+ return;
+ }
+ log.warn("Cloud upload task not executed before shutdown; handing
off to retry/DLQ: "
+ + "db={}, cf={}, path={}", dbName, cfName, filePath);
+ retryQueue.submitPinned(dbName, cfName, pinned.toString(),
filePath, remoteKey,
+ uploadEpoch,
+ new IOException("upload executor stopped
before task ran"));
+ }
+ }
+
+ /**
+ * 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);
+ // Use the staging dir that lives on the same filesystem as the source
SST to avoid
+ // cross-device hard-link failures in multi-disk deployments.
+ String matchingRoot =
findMatchingDataRoot(source.toAbsolutePath().normalize().toString());
+ Path stagingDir = Paths.get(matchingRoot, ".cloud-upload-staging");
+ Files.createDirectories(stagingDir);
+ String fileName = source.getFileName().toString();
+ Path staged = stagingDir.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) {
Review Comment:
⚠️ Backpressure is inert as shipped, and nobody asked for it.
`upload-backpressure-high-watermark` is `0` in both `application.yml` files,
and this method and `dlqEnqueueRateBacklog` (L1910) both return immediately
when it's `<= 0`. Nothing in #3079 or the PR description asks for ingestion
throttling, and the yaml comment this PR adds argues against switching it on:
"the throttle parks RocksDB's flush/compaction thread (up to 30s per SST
event), which under a sustained cloud outage can stall memtable flushes / stop
writes for the partition."
Deleting the path removes `applyBackpressure`, `pendingUploadBacklog`,
`dlqEnqueueRateBacklog`, the `dlqRateLock` monitor plus the four
`lastDlqRateSample*` / `dlqRateBacklogContribution` fields,
`BACKPRESSURE_MAX_WAIT_MS`, `BACKPRESSURE_POLL_MS`,
`DLQ_ENQUEUE_RATE_WINDOW_MS`, the `backpressureHighWatermark` constructor
parameter threaded through two constructors,
`CloudUploadRetryQueue.getDlqEnqueuedTotal()`, and a config key in two files.
The `cloud_storage_retry_queue_size` and
`cloud_storage_dlq_persistence_healthy` gauges you already register surface the
same "durability is degrading" signal without parking a RocksDB flush thread.
Add the throttle back when an operator asks for one and can say what it should
do.
##########
hugegraph-store/hg-store-cloud-s3/src/main/java/org/apache/hugegraph/store/cloud/s3/S3CloudStorageProvider.java:
##########
@@ -0,0 +1,1100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.store.cloud.s3;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.store.cloud.CloudStorageConfig;
+import org.apache.hugegraph.store.cloud.CloudStorageNonRetryableException;
+import org.apache.hugegraph.store.cloud.CloudStorageProvider;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.jetbrains.annotations.NotNull;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.http.ContentStreamProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3ClientBuilder;
+import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
+import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.ListMultipartUploadsRequest;
+import software.amazon.awssdk.services.s3.model.ListMultipartUploadsResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.MultipartUpload;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Object;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+/**
+ * Amazon S3 (and S3-compatible) implementation of {@link
CloudStorageProvider}.
+ *
+ * <h3>Activation</h3>
+ * Place {@code hg-store-cloud-s3-*.jar} on the classpath and configure:
+ * <pre>
+ * cloud:
+ * storage:
+ * enabled: true
+ * provider: s3
+ * s3:
+ * bucket: my-bucket
+ * region: us-east-1
+ * </pre>
+ *
+ * <h3>Credentials</h3>
+ * <ul>
+ * <li>If {@code cloud.storage.s3.access-key} / {@code secret-key} are set,
+ * they are used directly.</li>
+ * <li>Otherwise the standard AWS Default Credentials chain is followed
+ * (env vars, instance profile, ~/.aws/credentials, etc.).</li>
+ * </ul>
+ *
+ * <h3>S3-compatible endpoints (MinIO, Ceph, etc.)</h3>
+ * Set {@code cloud.storage.s3.endpoint} to the custom HTTP/HTTPS endpoint URL.
+ *
+ * <h3>Large-file (multipart) uploads</h3>
+ * S3 limits a single PUT to 5 GB. Files larger than
+ * {@link #MULTIPART_THRESHOLD_BYTES} ({@value #MULTIPART_THRESHOLD_BYTES} MB)
+ * are automatically split into {@link #PART_SIZE_BYTES} ({@value
#PART_SIZE_MB} MB)
+ * chunks and uploaded using the S3 Multipart Upload API.
+ * Each chunk is logged individually so progress is visible for very large
files.
+ *
+ * <h3>Multipart part retry tuning</h3>
+ * Tune part-level retry behavior via typed S3 keys:
+ * <pre>
+ * cloud:
+ * storage:
+ * s3:
+ * multipart-part-retry-max-attempts: 5
+ * multipart-part-retry-base-backoff-ms: 1500
+ * multipart-exhausted-direct-dlq: false
+ * </pre>
+ * These options apply only to multipart chunks, not to whole-file retry/DLQ
policy
+ * in {@code CloudUploadRetryQueue}.
+ *
+ * <h3>Timing metrics</h3>
+ * Every upload logs the file size, elapsed time, and throughput at INFO level:
+ * <pre>
+ * S3 upload complete: db/000042.sst | size=64.0 MB | elapsed=830 ms |
throughput=77.11 MB/s
+ * </pre>
+ */
+@Slf4j
+public class S3CloudStorageProvider implements CloudStorageProvider {
+
+ /** Provider name as referenced in {@link
CloudStorageConfig#getProvider()}. */
+ public static final String PROVIDER_NAME = "s3";
+
+ /**
+ * Files larger than this are uploaded via multipart.
+ * S3's hard per-PUT limit is 5 GB; we start multipart well below that.
+ */
+ static final long MULTIPART_THRESHOLD_BYTES = 512L * 1024 * 1024; // 512
MB
+
+ /**
+ * Size of each multipart chunk.
+ * S3 minimum part size is 5 MB (except for the last part).
+ */
+ static final long PART_SIZE_BYTES = 512L * 1024 * 1024; // 512
MB
+ static final int PART_SIZE_MB = 512;
+
+ /**
+ * Upper bound on a single retry backoff sleep. Keeps a large configured
{@code max-attempts}
+ * (or base backoff) from parking an upload thread for hours/days and from
overflowing the
+ * {@code 1L << n} shift used to compute the exponential delay.
+ */
+ private static final long MAX_RETRY_BACKOFF_MS = 60_000L;
+
+ /** Upper bound on configured part-upload retry attempts, to keep the
backoff bounded. */
+ private static final int MAX_PART_UPLOAD_RETRIES = 20;
+
+ private S3Client s3Client;
+ private String bucket;
+ private String pathPrefix;
+ private int partUploadMaxRetries =
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_MAX_ATTEMPTS;
+ private long partUploadRetryBaseBackoffMs =
+ S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_BASE_BACKOFF_MS;
+ private boolean multipartExhaustedDirectDlq = false;
+ private boolean multipartStaleAbortOnInit =
+ S3CloudStorageConfig.DEFAULT_MULTIPART_STALE_ABORT_ON_INIT;
+
+ // -----------------------------------------------------------------------
+ // CloudStorageProvider
+ // -----------------------------------------------------------------------
+
+ @Override
+ public String providerName() {
+ return PROVIDER_NAME;
+ }
+
+ @Override
+ public void init(CloudStorageConfig config) {
+ Map<String, String> props = config.getProviderProperties();
+ if (props == null || props.isEmpty()) {
+ throw new IllegalArgumentException("S3 provider selected but
providerProperties are empty");
+ }
+
+ this.bucket = props.get(S3CloudStorageConfig.KEY_BUCKET);
+ if (this.bucket == null || this.bucket.isBlank()) {
+ throw new IllegalArgumentException("S3 bucket is required:
cloud.storage.s3.bucket");
+ }
+ this.pathPrefix = config.getPathPrefix();
+ this.initRetryConfig(props);
+
+ S3ClientBuilder builder = S3Client.builder();
+
+ // Credentials
+ String ak = props.get(S3CloudStorageConfig.KEY_ACCESS_KEY);
+ String sk = props.get(S3CloudStorageConfig.KEY_SECRET_KEY);
+ if (ak != null && !ak.isEmpty() && sk != null && !sk.isEmpty()) {
+ builder.credentialsProvider(
+
StaticCredentialsProvider.create(AwsBasicCredentials.create(ak, sk)));
+ } else {
+
builder.credentialsProvider(DefaultCredentialsProvider.builder().build());
+ }
+
+ // Region
+ String region = props.get(S3CloudStorageConfig.KEY_REGION);
+ if (region != null && !region.isEmpty()) {
+ builder.region(Region.of(region));
+ }
+
+ // Custom endpoint (MinIO, Ceph, LocalStack …)
+ String endpoint = props.get(S3CloudStorageConfig.KEY_ENDPOINT);
+ if (endpoint != null && !endpoint.isEmpty()) {
+ builder.endpointOverride(URI.create(endpoint));
+ // Path-style required for most non-AWS S3 services
+ builder.serviceConfiguration(
+
software.amazon.awssdk.services.s3.S3Configuration.builder()
+
.pathStyleAccessEnabled(true)
+
.build());
+ }
+
+ // Close any client from a previous init() so a re-initialization
(e.g. Spring context
+ // restart, which re-runs the same singleton provider instance) does
not leak the old
+ // client's connection pool and SDK threads.
+ if (this.s3Client != null) {
+ try {
+ this.s3Client.close();
+ } catch (Exception e) {
+ log.warn("Failed to close previous S3 client on re-init: {}",
e.getMessage());
+ }
+ }
+ this.s3Client = builder.build();
+ log.info("S3CloudStorageProvider initialized: bucket='{}',
region='{}', endpoint='{}', "
+ + "partRetryMaxAttempts={}, partRetryBaseBackoffMs={}, "
+ + "multipartExhaustedDirectDlq={}",
+ bucket, region, endpoint,
+ this.partUploadMaxRetries,
+ this.partUploadRetryBaseBackoffMs,
+ this.multipartExhaustedDirectDlq);
+ // Blast-radius guard: only sweep when explicitly enabled AND a
non-empty pathPrefix scopes
+ // the listing. With an empty prefix the sweep would span the entire
bucket and could abort
+ // in-flight multipart uploads owned by other writers/applications
sharing it.
+ if (this.multipartStaleAbortOnInit
+ && this.pathPrefix != null && !this.pathPrefix.isBlank()) {
+ abortStaleMultipartUploads();
+ } else {
+ log.info("Skipping init-time stale-multipart sweep (enabled={},
pathPrefix='{}'): "
+ + "rely on an S3 AbortIncompleteMultipartUpload lifecycle
rule for cleanup",
+ this.multipartStaleAbortOnInit, this.pathPrefix);
+ }
+ }
+
+ /**
+ * Sweeps for incomplete multipart uploads older than 24 h and aborts them.
+ * A JVM crash (SIGKILL, OOM) after {@code createMultipartUpload} but
before the guarding
+ * {@code abortMultipartUpload} in {@link #uploadMultipart} leaves
orphaned parts in S3
+ * indefinitely. This best-effort sweep runs once at provider
initialisation so that stale
+ * uploads from a previous crashed instance are cleaned up before any new
uploads begin.
+ * Operators should also configure an {@code
AbortIncompleteMultipartUpload} S3 lifecycle
+ * rule (e.g. 1 day) as a second line of defence for crashes that occur
before the next
+ * provider initialisation.
+ */
+ private void abortStaleMultipartUploads() {
+ try {
+ String prefix = pathPrefix != null ? pathPrefix : "";
+ // Defense in depth: never run an unscoped (whole-bucket) sweep
even if reached directly.
+ if (prefix.isBlank()) {
+ log.warn("Refusing stale-multipart sweep with an empty prefix
(would span the "
+ + "entire bucket and could abort unrelated uploads)");
+ return;
+ }
+ long cutoff = System.currentTimeMillis() -
TimeUnit.DAYS.toMillis(1);
+ String keyMarker = null;
+ String uploadIdMarker = null;
+ ListMultipartUploadsResponse resp;
+ do {
+ ListMultipartUploadsRequest.Builder req =
+
ListMultipartUploadsRequest.builder().bucket(bucket).prefix(prefix);
+ if (keyMarker != null && !keyMarker.isEmpty()) {
+ req.keyMarker(keyMarker).uploadIdMarker(uploadIdMarker);
+ }
+ resp = s3Client.listMultipartUploads(req.build());
+ for (MultipartUpload u : resp.uploads()) {
+ if (u.initiated() != null && u.initiated().toEpochMilli()
< cutoff) {
+ try {
+ s3Client.abortMultipartUpload(
+ AbortMultipartUploadRequest.builder()
+ .bucket(bucket)
+ .key(u.key())
+
.uploadId(u.uploadId())
+ .build());
+ log.info("Aborted stale multipart upload: key={}
uploadId={}",
+ u.key(), u.uploadId());
+ } catch (Exception abortEx) {
+ log.warn("Failed to abort stale multipart upload:
key={} uploadId={}: {}",
+ u.key(), u.uploadId(),
abortEx.getMessage());
+ }
+ }
+ }
+ keyMarker = resp.nextKeyMarker();
+ uploadIdMarker = resp.nextUploadIdMarker();
+ if (Boolean.TRUE.equals(resp.isTruncated())
+ && (keyMarker == null || keyMarker.isEmpty())) {
+ // Some S3-compatible gateways return isTruncated=true
without a usable
+ // nextKeyMarker. Re-issuing the request without a marker
would refetch the
+ // first page forever and hang init(); stop the
best-effort sweep early instead.
+ log.warn("Stale-multipart sweep: response truncated but no
continuation marker "
+ + "returned; stopping early to avoid an infinite
list loop");
+ break;
+ }
+ } while (Boolean.TRUE.equals(resp.isTruncated()));
+ } catch (Exception e) {
+ log.warn("Stale multipart upload sweep failed (non-critical): {}",
e.getMessage());
+ }
+ }
+
+ private void initRetryConfig(Map<String, String> props) {
+ int retryMaxAttempts = parseIntOrDefault(
+
props.get(S3CloudStorageConfig.KEY_MULTIPART_RETRY_MAX_ATTEMPTS)
+ );
+ if (retryMaxAttempts <= 0) {
+ log.warn("Invalid
cloud.storage.s3.multipart-part-retry-max-attempts={} "
+ + "(must be > 0), using default {}",
+ retryMaxAttempts,
+
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_MAX_ATTEMPTS);
+ this.partUploadMaxRetries =
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_MAX_ATTEMPTS;
+ } else if (retryMaxAttempts > MAX_PART_UPLOAD_RETRIES) {
+ log.warn("cloud.storage.s3.multipart-part-retry-max-attempts={}
exceeds the supported "
+ + "maximum {}; clamping (per-attempt backoff is capped at
{} ms)",
+ retryMaxAttempts, MAX_PART_UPLOAD_RETRIES,
MAX_RETRY_BACKOFF_MS);
+ this.partUploadMaxRetries = MAX_PART_UPLOAD_RETRIES;
+ } else {
+ this.partUploadMaxRetries = retryMaxAttempts;
+ }
+
+ long retryBaseBackoffMs = parseLongOrDefault(
+
props.get(S3CloudStorageConfig.KEY_MULTIPART_RETRY_BASE_BACKOFF_MS)
+ );
+ if (retryBaseBackoffMs <= 0L) {
+ log.warn("Invalid
cloud.storage.s3.multipart-part-retry-base-backoff-ms={} "
+ + "(must be > 0), using default {}",
+ retryBaseBackoffMs,
+
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_BASE_BACKOFF_MS);
+ this.partUploadRetryBaseBackoffMs =
+
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_BASE_BACKOFF_MS;
+ } else {
+ this.partUploadRetryBaseBackoffMs = retryBaseBackoffMs;
+ }
+
+ this.multipartExhaustedDirectDlq = parseBooleanOrDefault(
+
props.get(S3CloudStorageConfig.KEY_MULTIPART_EXHAUSTED_DIRECT_DLQ));
+
+ String staleAbort =
props.get(S3CloudStorageConfig.KEY_MULTIPART_STALE_ABORT_ON_INIT);
+ this.multipartStaleAbortOnInit = (staleAbort == null ||
staleAbort.isBlank())
+ ? S3CloudStorageConfig.DEFAULT_MULTIPART_STALE_ABORT_ON_INIT
+ : Boolean.parseBoolean(staleAbort.trim());
+ }
+
+ private static int parseIntOrDefault(String value) {
+ if (value == null || value.isBlank()) {
+ return
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_MAX_ATTEMPTS;
+ }
+ try {
+ return Integer.parseInt(value.trim());
+ } catch (NumberFormatException e) {
+ log.warn("Invalid {}={}, using default {}",
+ "cloud.storage.s3.multipart-part-retry-max-attempts",
value,
+
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_MAX_ATTEMPTS);
+ return
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_MAX_ATTEMPTS;
+ }
+ }
+
+ private static long parseLongOrDefault(String value) {
+ if (value == null || value.isBlank()) {
+ return
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_BASE_BACKOFF_MS;
+ }
+ try {
+ return Long.parseLong(value.trim());
+ } catch (NumberFormatException e) {
+ log.warn("Invalid {}={}, using default {}",
+ "cloud.storage.s3.multipart-part-retry-base-backoff-ms",
value,
+
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_BASE_BACKOFF_MS);
+ return
S3CloudStorageConfig.DEFAULT_MULTIPART_PART_RETRY_BASE_BACKOFF_MS;
+ }
+ }
+
+ private static boolean parseBooleanOrDefault(String value) {
+ if (value == null || value.isBlank()) {
+ return false;
+ }
+ return Boolean.parseBoolean(value.trim());
+ }
+
+ /**
+ * Uploads a local file to S3.
+ *
+ * <p>Files ≤ {@link #MULTIPART_THRESHOLD_BYTES} use a single PUT
request.
+ * Larger files are split into {@link #PART_SIZE_BYTES} chunks and
uploaded via
+ * the S3 Multipart Upload API, which is required for files larger than 5
GB.
+ *
+ * <p>Timing and throughput are always logged at INFO level after the
upload
+ * completes (or each part for multipart uploads).
+ */
+ @Override
+ public void uploadFile(String localPath, String remoteKey) throws
IOException {
+ java.nio.file.Path path = Paths.get(localPath);
+ long fileSize;
+ try {
+ fileSize = Files.size(path);
+ } catch (IOException e) {
+ throw new IOException("Cannot stat local file: " + localPath, e);
+ }
+
+ String fullKey = buildKey(remoteKey);
+ long startNs = System.nanoTime();
+
+ if (fileSize > MULTIPART_THRESHOLD_BYTES) {
+ uploadMultipart(path, fileSize, fullKey);
+ } else {
+ uploadSinglePart(path, fullKey, localPath);
+ }
+
+ long elapsedMs = (System.nanoTime() - startNs) / 1_000_000;
+ double throughputMBps = elapsedMs > 0
+ ? (fileSize / 1_048_576.0) / (elapsedMs /
1000.0)
+ : 0.0;
+ log.info("S3 upload complete: {} | size={} | elapsed={} ms |
throughput={} MB/s",
+ remoteKey,
+ humanSize(fileSize),
+ elapsedMs,
+ String.format(Locale.US, "%.2f", throughputMBps));
+ }
+
+ @Override
+ public void deleteFile(String remoteKey) throws IOException {
+ String fullKey = buildKey(remoteKey);
+ try {
+ s3Client.deleteObject(
+
DeleteObjectRequest.builder().bucket(bucket).key(fullKey).build());
+ log.debug("S3 delete: s3://{}/{}", bucket, fullKey);
+ } catch (SdkException e) {
+ throw classifySdkException("deleteObject", fullKey, e);
+ }
+ }
+
+ /**
+ * Deletes all objects under a prefix using S3's DeleteObjects (batch
delete) API.
+ *
+ * <p>Much more efficient than individual deletes, especially for prefixes
with many objects.
+ * Handles pagination internally if the prefix contains more than 1000
objects.
+ *
+ * @param remoteDirPrefix directory/prefix inside bucket (without provider
pathPrefix)
+ * @return number of objects deleted
+ * @throws IOException on I/O or network failure
+ */
+ @Override
+ public int deletePrefix(String remoteDirPrefix) throws IOException {
+ String fullPrefix = buildKey(remoteDirPrefix == null ? "" :
remoteDirPrefix);
+ int totalDeleted = 0;
+
+ try {
+ String token = null;
+ do {
+ ListObjectsV2Request.Builder listReq =
+
ListObjectsV2Request.builder().bucket(bucket).prefix(fullPrefix);
+ if (token != null) {
+ listReq.continuationToken(token);
+ }
+ ListObjectsV2Response listResp =
s3Client.listObjectsV2(listReq.build());
+
+ List<ObjectIdentifier> toDelete = new ArrayList<>();
+ for (S3Object obj : listResp.contents()) {
+ String key = obj.key();
+ if (key != null && !key.endsWith("/")) {
+
toDelete.add(ObjectIdentifier.builder().key(key).build());
+ }
+ }
+
+ if (!toDelete.isEmpty()) {
+ try {
+ DeleteObjectsResponse deleteResp =
s3Client.deleteObjects(
+ DeleteObjectsRequest.builder()
+ .bucket(bucket)
+
.delete(software.amazon.awssdk.services.s3.model.Delete.builder()
+
.objects(toDelete)
+
.build())
+ .build());
+ totalDeleted += deleteResp.deleted().size();
+ log.debug("S3 batch delete: deleted {} objects from
prefix {}",
+ deleteResp.deleted().size(),
remoteDirPrefix);
+
+ // S3 returns HTTP 200 even when individual keys fail;
always inspect.
+ if (!deleteResp.errors().isEmpty()) {
+ log.warn("S3 batch delete partial failure: {}/{}
key(s) failed in "
+ + "prefix '{}' — retrying individually",
+ deleteResp.errors().size(),
toDelete.size(), remoteDirPrefix);
+ List<String> stillFailed = new ArrayList<>();
+ for
(software.amazon.awssdk.services.s3.model.S3Error err
+ : deleteResp.errors()) {
+ log.warn(" S3 DeleteObjects error: key={}
code={} message={}",
+ err.key(), err.code(), err.message());
+ try {
+
s3Client.deleteObject(DeleteObjectRequest.builder()
+
.bucket(bucket)
+
.key(err.key())
+
.build());
+ totalDeleted++;
+ log.debug("S3 individual retry delete
succeeded: key={}",
+ err.key());
+ } catch (SdkException ex) {
+ stillFailed.add(err.key());
+ log.warn("S3 individual retry delete
failed: key={}: {}",
+ err.key(), ex.getMessage());
+ }
+ }
+ if (!stillFailed.isEmpty()) {
+ throw new IOException(
+ "S3 DeleteObjects: " +
stillFailed.size()
+ + " key(s) could not be deleted from
prefix '"
+ + remoteDirPrefix + "': " +
stillFailed);
+ }
+ }
+ } catch (SdkException e) {
+ log.warn("S3 batch delete failed for prefix='{}': {}",
+ fullPrefix, e.getMessage());
+ // Fall back to individual deletes for any remaining
objects.
+ // Failures are collected and re-thrown so callers
(e.g. purgeRemotePrefix)
+ // can correctly preserve the tombstone guard when the
purge is incomplete.
+ List<String> stillFailed = new ArrayList<>();
+ for (ObjectIdentifier obj : toDelete) {
+ try {
+
s3Client.deleteObject(DeleteObjectRequest.builder()
+
.bucket(bucket)
+
.key(obj.key())
+
.build());
+ totalDeleted++;
+ } catch (SdkException ex) {
+ stillFailed.add(obj.key());
+ log.debug("S3 fallback delete failed for
key='{}': {}",
+ obj.key(), ex.getMessage());
+ }
+ }
+ if (!stillFailed.isEmpty()) {
+ throw new IOException(
+ "S3 fallback delete: " + stillFailed.size()
+ + " key(s) could not be deleted from
prefix '"
+ + remoteDirPrefix + "': " + stillFailed);
+ }
+ }
+ }
+
+ token = listResp.nextContinuationToken();
+ // Some S3-compatible gateways return isTruncated=true without
a usable
+ // continuation token. Unlike the best-effort multipart sweep
(which can stop
+ // early), a prefix purge that silently stops here would
report success while
+ // objects remain — and the caller (onDBDeleted / truncate
purge) would then remove
+ // its tombstone guard, leaving stale objects that can be
re-hydrated as live data.
+ // Fail loudly so the purge is marked incomplete and the guard
is preserved.
+ if (Boolean.TRUE.equals(listResp.isTruncated())
+ && (token == null || token.isEmpty())) {
+ throw new IOException(
+ "S3 deletePrefix: listing for prefix '" +
remoteDirPrefix
+ + "' is truncated but returned no continuation
token; the purge is "
+ + "incomplete and cannot be confirmed. Deleted " +
totalDeleted
+ + " object(s) so far.");
+ }
+ } while (token != null && !token.isEmpty());
+
+ if (totalDeleted > 0) {
+ log.info("S3 prefix delete completed: prefix={}, deleted={}",
remoteDirPrefix, totalDeleted);
+ }
+ return totalDeleted;
+
+ } catch (SdkException e) {
+ throw classifySdkException("deletePrefix", fullPrefix, e);
+ }
+ }
+
+ @Override
+ public boolean fileExists(String remoteKey) throws IOException {
+ String fullKey = buildKey(remoteKey);
+ try {
+
s3Client.headObject(HeadObjectRequest.builder().bucket(bucket).key(fullKey).build());
+ return true;
+ } catch (NoSuchKeyException e) {
+ return false;
+ } catch (AwsServiceException e) {
+ // A missing bucket is a misconfiguration, not a missing key.
Treat it as a hard error
+ // so callers (e.g. tombstone check in preHydrateDbFiles) surface
the problem rather
+ // than silently skipping hydration and starting with an empty
database.
+ String errorCode = e.awsErrorDetails() != null ?
e.awsErrorDetails().errorCode() : "";
+ if ("NoSuchBucket".equals(errorCode) ||
"InvalidBucketName".equals(errorCode)) {
+ throw new IOException(
+ "S3 bucket not found or misconfigured (key=" + fullKey
+ "): " + errorCode,
+ e);
+ }
+ // Some S3-compatible providers return generic service exceptions
for 404.
+ if (e.statusCode() == 404) {
+ return false;
+ }
+ throw classifySdkException("headObject", fullKey, e);
+ } catch (SdkException e) {
+ throw classifySdkException("headObject", fullKey, e);
+ }
+ }
+
+ @Override
+ public List<String> listFiles(String remoteDirPrefix) throws IOException {
+ String fullPrefix = buildKey(remoteDirPrefix == null ? "" :
remoteDirPrefix);
+ List<String> keys = new ArrayList<>();
+ try {
+ String token = null;
+ do {
+ ListObjectsV2Request.Builder req =
+
ListObjectsV2Request.builder().bucket(bucket).prefix(fullPrefix);
+ if (token != null) {
+ req.continuationToken(token);
+ }
+ ListObjectsV2Response resp =
s3Client.listObjectsV2(req.build());
+ for (S3Object obj : resp.contents()) {
+ String key = obj.key();
+ if (key == null || key.endsWith("/")) {
+ continue;
+ }
+ keys.add(stripPathPrefix(key));
+ }
+ token = resp.nextContinuationToken();
+ // A truncated listing with no continuation token would
silently return a PARTIAL
+ // key set. Startup hydration relies on a complete listing
(e.g. to find CURRENT);
+ // a partial set could let a DB open on incomplete local
state. Fail loudly so the
+ // caller blocks rather than proceeding on a partial listing.
+ if (Boolean.TRUE.equals(resp.isTruncated())
+ && (token == null || token.isEmpty())) {
+ throw new IOException(
+ "S3 listFiles: listing for prefix '" + fullPrefix
+ "' is truncated "
+ + "but returned no continuation token; refusing to
return a partial "
+ + "listing (" + keys.size() + " key(s) seen so
far).");
+ }
+ } while (token != null && !token.isEmpty());
+ return keys;
+ } catch (SdkException e) {
+ throw classifySdkException("listObjectsV2", fullPrefix, e);
+ }
+ }
+
+ @Override
+ public void downloadFile(String remoteKey, String localPath) throws
IOException {
+ String fullKey = buildKey(remoteKey);
+ long startNs = System.nanoTime();
+ Path destinationPath = Paths.get(localPath);
+ try {
+
+ s3Client.getObject(
+
GetObjectRequest.builder().bucket(bucket).key(fullKey).build(),
+ destinationPath);
+ } catch (SdkException e) {
+ throw classifySdkException("getObject", fullKey, e);
+ }
+ long elapsedMs = (System.nanoTime() - startNs) / 1_000_000;
+ long fileSize = 0;
+ try {
+ fileSize = Files.size(destinationPath);
+ } catch (IOException ignored) {
+ // best-effort; don't fail download reporting
+ }
+ double throughputMBps = elapsedMs > 0
+ ? (fileSize / 1_048_576.0) / (elapsedMs /
1000.0)
+ : 0.0;
+ log.info("S3 download complete: {} | size={} | elapsed={} ms |
throughput={} MB/s",
+ remoteKey,
+ humanSize(fileSize),
+ elapsedMs,
+ String.format(Locale.US, "%.2f", throughputMBps));
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (s3Client != null) {
+ s3Client.close();
+ s3Client = null;
+ log.info("S3CloudStorageProvider closed");
+ }
+ }
+
+ // -----------------------------------------------------------------------
+ // Internal – upload strategies
+ // -----------------------------------------------------------------------
+
+ /**
+ * Single-PUT upload for files ≤ {@link #MULTIPART_THRESHOLD_BYTES}, with
bounded
+ * exponential-backoff retry on transient failures (using the same tuning
as multipart parts).
+ * Without this, a single transient network blip on the common small-SST
path would surface
+ * immediately and, with whole-file retries disabled, go straight to the
DLQ.
+ */
+ private void uploadSinglePart(java.nio.file.Path path, String fullKey,
+ String localPath) throws IOException {
+ IOException last = null;
+ for (int attempt = 1; attempt <= this.partUploadMaxRetries; attempt++)
{
+ try {
+ s3Client.putObject(
+
PutObjectRequest.builder().bucket(bucket).key(fullKey).build(),
+ path);
+ return;
+ } catch (SdkException e) {
+ IOException classified = classifySdkException("putObject",
fullKey, e);
+ if (classified instanceof CloudStorageNonRetryableException) {
+ throw classified;
+ }
+ last = classified;
+ if (attempt >= this.partUploadMaxRetries) {
+ break;
+ }
+ long backoffMs = retryBackoffMs(attempt);
+ log.warn("S3 single-PUT retry: attempt={}/{} key={} reason={}
nextBackoffMs={}",
+ attempt, this.partUploadMaxRetries, fullKey,
+ classified.getMessage(), backoffMs);
+ sleepQuietly(backoffMs);
+ }
+ }
+ throw new IOException(
+ "S3 upload failed for local='" + localPath + "' key='" +
fullKey + "' after "
+ + this.partUploadMaxRetries + " attempt(s)", last);
+ }
+
+ /**
+ * Multipart upload for files > {@link #MULTIPART_THRESHOLD_BYTES}.
+ *
+ * <p>Each part is logged individually so that progress of multi-hour
uploads
+ * is visible in the server log:
+ * <pre>
+ * S3 multipart part 1/41 uploaded: size=512.0 MB | elapsed=6 230 ms |
throughput=82.18 MB/s
+ * S3 multipart part 2/41 uploaded: size=512.0 MB | elapsed=6 050 ms |
throughput=84.63 MB/s
+ * ...
+ * S3 multipart upload completed: key=hugegraph/hgstore-data/000099.sst
| parts=41
+ * </pre>
+ *
+ * <p>If any part fails the multipart upload is aborted (to avoid
incomplete-upload storage
+ * charges) and an {@link IOException} is thrown.
+ */
+ private void uploadMultipart(java.nio.file.Path path, long fileSize,
Review Comment:
⚠️ This multipart driver and its retry layer are re-implementing the SDK you
already depend on.
`init()` (L176-248) builds the `S3Client` with no
`overrideConfiguration(...)`, so AWS SDK v2 defaults are live: every
`uploadPart` is already retried 3× with full-jitter exponential backoff on 5xx,
throttling and IO errors. `uploadOnePartWithRetry` (L880) then wraps that in a
second loop, so a part can be attempted up to 12 times before anyone hears
about it.
Smallest fix, no new dependency: delete `uploadOnePartWithRetry`,
`retryBackoffMs`, `sleepQuietly`, `MAX_PART_UPLOAD_RETRIES` and
`MAX_RETRY_BACKOFF_MS`, and set the count you want on the client instead —
`ClientOverrideConfiguration.builder().retryPolicy(RetryPolicy.builder().numRetries(n).build())`.
Lazier still: `software.amazon.awssdk:s3-transfer-manager` from the same
2.33.8 BOM, over an `S3AsyncClient` built with `.multipartEnabled(true)`,
collapses `uploadFile` to
```java
tm.uploadFile(UploadFileRequest.builder()
.source(path)
.putObjectRequest(b -> b.bucket(bucket).key(fullKey))
.build())
.completionFuture().join();
```
That deletes `uploadMultipart`, `uploadOnePart`, `openBoundedPartStream`,
`LimitedInputStream` and the three `cloud.storage.s3.multipart-part-retry-*`
knobs in both `application.yml` files — roughly 400 of this file's 1100 lines.
Separately, `abortStaleMultipartUploads` (L259) can go too: its own javadoc
already tells operators to configure the S3 `AbortIncompleteMultipartUpload`
lifecycle rule, which is the platform doing this for free and doesn't need an
init-time `ListMultipartUploads` sweep.
--
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]
