vaijosh commented on code in PR #3081: URL: https://github.com/apache/hugegraph/pull/3081#discussion_r4044300841
########## 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: Addressed ########## 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: Addressed -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
