From e04b3e1f6c72c54ce4dcfb53a5c096042f4dc11b Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Sat, 10 Oct 2026 16:49:45 +0530 Subject: [PATCH 01/10] JAMES-4231 Add S3 blob range reads --- .../apache/james/blob/api/BlobStoreDAO.java | 70 ++++++++++++++ .../api/BlobStoreDAORangeReadContract.java | 96 +++++++++++++++++++ .../objectstorage/aws/S3BlobStoreDAO.java | 77 +++++++++++++++ .../aws/S3BlobStoreDAORangeReadTest.java | 81 ++++++++++++++++ 4 files changed, 324 insertions(+) create mode 100644 server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java create mode 100644 server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java index 46af6ebed14..ddbbc01b32e 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java @@ -39,6 +39,7 @@ import com.google.common.io.FileBackedOutputStream; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; /** * James virtual blob store abstraction. @@ -303,4 +304,73 @@ default Publisher listBlobs(BucketName bucketName, String prefix) { return Flux.from(listBlobs(bucketName)) .filter(blobId -> blobId.asString().startsWith(prefix)); } + + BlobMetadataName TOTAL_OBJECT_SIZE = new BlobMetadataName("total-object-size"); + + /** + * Returns the size of the source object from a ranged-read result. + * + *

The value is the complete object size, rather than the range payload size. Implementations that support + * efficient ranged reads must add it to their result metadata. The default ranged-read implementation does so + * after reading the complete object.

+ */ + static long totalObjectSize(Blob blob) { + return blob.metadata().get(TOTAL_OBJECT_SIZE) + .map(BlobMetadataValue::value) + .map(Long::parseLong) + .orElseThrow(() -> new IllegalArgumentException("Ranged-read result is missing total object size metadata")); + } + + /** + * Reads an inclusive byte range while preserving the blob metadata. + * + *

When {@code start} is non-negative, {@code end} is either an inclusive byte offset or {@code -1} to read + * through the end of the object. When {@code start} is negative, it requests that many bytes from the end of the + * object and {@code end} must be {@code -1}. For example, {@code readRange(bucket, id, -65536, -1)} reads up to + * the final 64 KiB. Implementations capable of backend range requests should override this method.

+ * + * @return the requested bytes, its original metadata, and {@link #TOTAL_OBJECT_SIZE} metadata + * @throws IllegalArgumentException when the range is malformed or starts beyond the end of the object + */ + default Mono readRange(BucketName bucketName, BlobId blobId, long start, long end) { + validateRange(start, end); + + return Mono.from(readBytes(bucketName, blobId)) + .map(bytesBlob -> { + long totalSize = bytesBlob.payload().length; + int from = rangeStart(start, totalSize); + int toExclusive = rangeEndExclusive(start, end, totalSize); + + BlobMetadata metadata = bytesBlob.metadata() + .withMetadata(TOTAL_OBJECT_SIZE, new BlobMetadataValue(Long.toString(totalSize))); + return BytesBlob.of(Arrays.copyOfRange(bytesBlob.payload(), from, toExclusive), metadata); + }); + } + + private static void validateRange(long start, long end) { + if (start < 0) { + Preconditions.checkArgument(start != Long.MIN_VALUE, "'start' must be greater than %s", Long.MIN_VALUE); + Preconditions.checkArgument(end == -1, "'end' must be -1 for a suffix range"); + return; + } + Preconditions.checkArgument(end == -1 || end >= start, + "'end' (%s) must be -1 or greater than or equal to 'start' (%s)", end, start); + } + + private static int rangeStart(long start, long totalSize) { + Preconditions.checkArgument(totalSize > 0, "Cannot read a range from an empty object"); + if (start < 0) { + return (int) Math.max(0, totalSize + start); + } + Preconditions.checkArgument(start < totalSize, + "'start' (%s) must be smaller than the object size (%s)", start, totalSize); + return (int) start; + } + + private static int rangeEndExclusive(long start, long end, long totalSize) { + if (start < 0 || end == -1) { + return (int) totalSize; + } + return end >= totalSize - 1 ? (int) totalSize : (int) (end + 1); + } } diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java new file mode 100644 index 00000000000..e8baf86d40f --- /dev/null +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java @@ -0,0 +1,96 @@ +/**************************************************************** + * 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.james.blob.api; + +import static org.apache.james.blob.api.BlobStoreDAOFixture.TEST_BUCKET_NAME; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.UUID; + +import org.junit.jupiter.api.Test; + +import reactor.core.publisher.Mono; + +public interface BlobStoreDAORangeReadContract { + BlobStoreDAO testee(); + + default byte[] sampleContent() { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < 100; i++) { + builder.append(String.format("%010d", i)); + } + return builder.toString().getBytes(StandardCharsets.UTF_8); + } + + default BlobId createSampleBlob() { + BlobId blobId = new TestBlobId.Factory().of(UUID.randomUUID().toString()); + BlobStoreDAO.BlobMetadata metadata = BlobStoreDAO.BlobMetadata.empty() + .withMetadata(new BlobStoreDAO.BlobMetadataName("test-metadata"), new BlobStoreDAO.BlobMetadataValue("value")); + Mono.from(testee().save(TEST_BUCKET_NAME, blobId, BlobStoreDAO.BytesBlob.of(sampleContent(), metadata))).block(); + return blobId; + } + + @Test + default void readRangeShouldReturnAnInclusivePrefixAndPreserveMetadata() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, 99).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 0, 100)); + assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(1000); + assertThat(slice.metadata().get(new BlobStoreDAO.BlobMetadataName("test-metadata"))) + .contains(new BlobStoreDAO.BlobMetadataValue("value")); + } + + @Test + default void readRangeShouldReturnAnInclusiveMiddleRange() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 200, 399).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 200, 400)); + } + + @Test + default void readRangeShouldReturnSuffix() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -100, -1).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 900, 1000)); + } + + @Test + default void readRangeShouldReadThroughTheEndWhenEndIsMinusOne() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 900, -1).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 900, 1000)); + } + + @Test + default void readRangeShouldFailWhenStartIsBeyondObjectSize() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 2000, 3000).block()) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + default void readRangeShouldRejectInvalidRange() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 300, 299).block()) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java index 94392fe6d3f..b8f5c3c0c6d 100644 --- a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java +++ b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java @@ -245,6 +245,66 @@ public Publisher readBytes(BucketName bucketName, BlobId blobId) { .onErrorMap(e -> e.getCause() instanceof OutOfMemoryError, Throwable::getCause); } + @Override + public Mono readRange(BucketName bucketName, BlobId blobId, long start, long end) { + validateRange(start, end); + BucketName resolvedBucketName = bucketNameResolver.resolve(bucketName); + String rangeHeader = formatRangeHeader(start, end); + + return getObjectRangeBytes(resolvedBucketName, blobId, rangeHeader) + .onErrorMap(NoSuchBucketException.class, e -> new ObjectNotFoundException("Bucket not found " + resolvedBucketName.asString(), e)) + .onErrorMap(NoSuchKeyException.class, e -> new ObjectNotFoundException("Blob not found " + blobId.asString() + " in bucket " + resolvedBucketName.asString(), e)) + .onErrorMap(this::isRangeNotSatisfiable, e -> new IllegalArgumentException("Range not satisfiable: " + rangeHeader, e)) + .publishOn(Schedulers.parallel()) + .map(responseBytes -> BytesBlob.of(responseBytes.asByteArrayUnsafe(), metadataForRange(responseBytes.response()))) + .onErrorMap(e -> e.getCause() instanceof OutOfMemoryError, Throwable::getCause); + } + + private static void validateRange(long start, long end) { + if (start < 0) { + Preconditions.checkArgument(start != Long.MIN_VALUE, "'start' must be greater than %s", Long.MIN_VALUE); + Preconditions.checkArgument(end == -1, "'end' must be -1 for a suffix range"); + return; + } + Preconditions.checkArgument(end == -1 || end >= start, + "'end' (%s) must be -1 or greater than or equal to 'start' (%s)", end, start); + } + + private static String formatRangeHeader(long start, long end) { + if (start < 0) { + return "bytes=" + start; + } + if (end == -1) { + return "bytes=" + start + "-"; + } + return "bytes=" + start + "-" + end; + } + + private static boolean isRangeNotSatisfiable(Throwable throwable) { + return Throwables.getCausalChain(throwable) + .stream() + .filter(S3Exception.class::isInstance) + .map(S3Exception.class::cast) + .anyMatch(s3Exception -> s3Exception.statusCode() == 416); + } + + private BlobMetadata metadataForRange(GetObjectResponse response) { + return asBlobMetadata(response.metadata()) + .withMetadata(TOTAL_OBJECT_SIZE, new BlobMetadataValue(Long.toString(extractTotalObjectSize(response)))); + } + + private static long extractTotalObjectSize(GetObjectResponse response) { + String contentRange = response.contentRange(); + if (contentRange != null && contentRange.contains("/")) { + try { + return Long.parseLong(contentRange.substring(contentRange.lastIndexOf('/') + 1)); + } catch (NumberFormatException e) { + throw new ObjectStoreIOException("Invalid content range returned by S3: " + contentRange, e); + } + } + throw new ObjectStoreIOException("S3 did not return a content range for a ranged read"); + } + private Mono> getObjectBytes(BucketName bucketName, BlobId blobId) { return getObjectBytesFromStore(bucketName, blobId) .onErrorResume(e -> e instanceof NoSuchKeyException || e instanceof NoSuchBucketException, e -> { @@ -255,12 +315,29 @@ private Mono> getObjectBytes(BucketName bucketN }); } + private Mono> getObjectRangeBytes(BucketName bucketName, BlobId blobId, String rangeHeader) { + return getObjectRangeBytesFromStore(bucketName, blobId, rangeHeader) + .onErrorResume(e -> e instanceof NoSuchKeyException || e instanceof NoSuchBucketException, e -> { + if (fallbackNamespace.isPresent() && bucketNameResolver.isNameSpace(bucketName)) { + return getObjectRangeBytesFromStore(fallbackNamespace.get(), blobId, rangeHeader); + } + return Mono.error(e); + }); + } + private Mono> getObjectBytesFromStore(BucketName bucketName, BlobId blobId) { return buildGetObjectRequestBuilder(bucketName, blobId) .flatMap(putObjectRequest -> Mono.fromFuture(() -> client.getObject(putObjectRequest.build(), new MinimalCopyBytesResponseTransformer(configuration, blobId)))); } + private Mono> getObjectRangeBytesFromStore(BucketName bucketName, BlobId blobId, String rangeHeader) { + return buildGetObjectRequestBuilder(bucketName, blobId) + .map(builder -> builder.range(rangeHeader)) + .flatMap(getObjectRequestBuilder -> Mono.fromFuture(() -> + client.getObject(getObjectRequestBuilder.build(), new MinimalCopyBytesResponseTransformer(configuration, blobId)))); + } + private Mono buildGetObjectRequestBuilder(BucketName bucketName, BlobId blobId) { GetObjectRequest.Builder baseBuilder = GetObjectRequest.builder() .bucket(bucketName.asString()) diff --git a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java new file mode 100644 index 00000000000..c1645056d95 --- /dev/null +++ b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java @@ -0,0 +1,81 @@ +/**************************************************************** + * 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.james.blob.objectstorage.aws; + +import static org.apache.james.blob.objectstorage.aws.JamesS3MetricPublisher.DEFAULT_S3_METRICS_PREFIX; +import static org.apache.james.blob.objectstorage.aws.S3BlobStoreConfiguration.UPLOAD_RETRY_EXCEPTION_PREDICATE; + +import java.time.Duration; +import java.util.Optional; + +import org.apache.james.blob.api.BlobStoreDAO; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; +import org.apache.james.blob.api.BucketName; +import org.apache.james.blob.api.TestBlobId; +import org.apache.james.metrics.api.NoopGaugeRegistry; +import org.apache.james.metrics.tests.RecordingMetricFactory; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.extension.ExtendWith; + +import reactor.util.retry.Retry; + +@ExtendWith(DockerAwsS3Extension.class) +class S3BlobStoreDAORangeReadTest implements BlobStoreDAORangeReadContract { + private static S3BlobStoreDAO testee; + private static S3ClientFactory clientFactory; + + @BeforeAll + static void setUp(DockerAwsS3Container dockerAwsS3) { + AwsS3AuthConfiguration authConfiguration = AwsS3AuthConfiguration.builder() + .endpoint(dockerAwsS3.getEndpoint()) + .accessKeyId(DockerAwsS3Container.ACCESS_KEY_ID) + .secretKey(DockerAwsS3Container.SECRET_ACCESS_KEY) + .build(); + + S3BlobStoreConfiguration s3Configuration = S3BlobStoreConfiguration.builder() + .authConfiguration(authConfiguration) + .region(dockerAwsS3.dockerAwsS3().region()) + .uploadRetrySpec(Optional.of(Retry.backoff(3, Duration.ofSeconds(1)) + .filter(UPLOAD_RETRY_EXCEPTION_PREDICATE))) + .defaultBucketName(BucketName.DEFAULT) + .build(); + + clientFactory = new S3ClientFactory(s3Configuration, + () -> new JamesS3MetricPublisher(new RecordingMetricFactory(), new NoopGaugeRegistry(), DEFAULT_S3_METRICS_PREFIX)); + testee = new S3BlobStoreDAO(clientFactory, s3Configuration, new TestBlobId.Factory(), S3RequestOption.DEFAULT); + } + + @AfterEach + void tearDown() { + testee.deleteAllBuckets().block(); + } + + @AfterAll + static void tearDownClass() { + clientFactory.close(); + } + + @Override + public BlobStoreDAO testee() { + return testee; + } +} From 045ea5c06b6c465fc3e90c48453e15a2de52c924 Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Sat, 10 Oct 2026 17:05:53 +0530 Subject: [PATCH 02/10] JAMES-4231 Fix S3 range error mapping --- .../apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java index b8f5c3c0c6d..0f7a21ab8d9 100644 --- a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java +++ b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java @@ -254,9 +254,9 @@ public Mono readRange(BucketName bucketName, BlobId blobId, long start, lo return getObjectRangeBytes(resolvedBucketName, blobId, rangeHeader) .onErrorMap(NoSuchBucketException.class, e -> new ObjectNotFoundException("Bucket not found " + resolvedBucketName.asString(), e)) .onErrorMap(NoSuchKeyException.class, e -> new ObjectNotFoundException("Blob not found " + blobId.asString() + " in bucket " + resolvedBucketName.asString(), e)) - .onErrorMap(this::isRangeNotSatisfiable, e -> new IllegalArgumentException("Range not satisfiable: " + rangeHeader, e)) + .onErrorMap(S3BlobStoreDAO::isRangeNotSatisfiable, e -> new IllegalArgumentException("Range not satisfiable: " + rangeHeader, e)) .publishOn(Schedulers.parallel()) - .map(responseBytes -> BytesBlob.of(responseBytes.asByteArrayUnsafe(), metadataForRange(responseBytes.response()))) + .map(responseBytes -> BytesBlob.of(responseBytes.asByteArrayUnsafe(), metadataForRange(responseBytes.response()))) .onErrorMap(e -> e.getCause() instanceof OutOfMemoryError, Throwable::getCause); } From cdb749ed0cefe36962b5bf15dc668c882b6754c4 Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Mon, 12 Oct 2026 05:08:00 +0530 Subject: [PATCH 03/10] JAMES-4231 Empty commit to retrigger CI Retriggers the PR build after an OpenSearch testcontainer infrastructure failure unrelated to the change set. Co-authored-by: CommandCodeBot From 1449386258f8ef3546f8bd2672bfd655d6c7cf2e Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Wed, 14 Oct 2026 14:57:02 +0530 Subject: [PATCH 04/10] JAMES-4231 Replace range size metadata and cover range boundaries --- .../apache/james/blob/api/BlobStoreDAO.java | 10 ++- .../api/BlobStoreDAORangeReadContract.java | 89 +++++++++++++++++++ .../blob/memory/MemoryBlobStoreDAOTest.java | 3 +- .../objectstorage/aws/S3BlobStoreDAO.java | 2 +- 4 files changed, 100 insertions(+), 4 deletions(-) diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java index ddbbc01b32e..5f9c53b4ed4 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java @@ -116,6 +116,13 @@ public Optional contentEncoding() { public BlobMetadata withContentEncoding(ContentEncoding contentEncoding) { return withMetadata(ContentEncoding.NAME, contentEncoding.asValue()); } + + public BlobMetadata withTotalObjectSize(long totalObjectSize) { + return new BlobMetadata(ImmutableMap.builder() + .putAll(underlyingMap) + .put(TOTAL_OBJECT_SIZE, new BlobMetadataValue(Long.toString(totalObjectSize))) + .buildKeepingLast()); + } } @@ -341,8 +348,7 @@ default Mono readRange(BucketName bucketName, BlobId blobId, long start, l int from = rangeStart(start, totalSize); int toExclusive = rangeEndExclusive(start, end, totalSize); - BlobMetadata metadata = bytesBlob.metadata() - .withMetadata(TOTAL_OBJECT_SIZE, new BlobMetadataValue(Long.toString(totalSize))); + BlobMetadata metadata = bytesBlob.metadata().withTotalObjectSize(totalSize); return BytesBlob.of(Arrays.copyOfRange(bytesBlob.payload(), from, toExclusive), metadata); }); } diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java index e8baf86d40f..6ee69c889c7 100644 --- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java @@ -93,4 +93,93 @@ default void readRangeShouldRejectInvalidRange() { assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 300, 299).block()) .isInstanceOf(IllegalArgumentException.class); } + + @Test + default void readRangeShouldReplaceStoredTotalObjectSize() throws IOException { + BlobId blobId = new TestBlobId.Factory().of(UUID.randomUUID().toString()); + BlobStoreDAO.BlobMetadata metadata = BlobStoreDAO.BlobMetadata.empty() + .withMetadata(BlobStoreDAO.TOTAL_OBJECT_SIZE, new BlobStoreDAO.BlobMetadataValue("9999")) + .withMetadata(new BlobStoreDAO.BlobMetadataName("custom"), new BlobStoreDAO.BlobMetadataValue("retained")); + Mono.from(testee().save(TEST_BUCKET_NAME, blobId, BlobStoreDAO.BytesBlob.of(sampleContent(), metadata))).block(); + + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, blobId, 0, 9).block(); + + assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(1000); + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 0, 10)); + assertThat(slice.metadata().get(new BlobStoreDAO.BlobMetadataName("custom"))) + .contains(new BlobStoreDAO.BlobMetadataValue("retained")); + } + + @Test + default void readRangeShouldReturnSingleByte() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 25, 25).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 25, 26)); + } + + @Test + default void readRangeShouldCapEndAtObjectSize() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 900, Long.MAX_VALUE).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 900, 1000)); + } + + @Test + default void readRangeShouldCapSuffixAtObjectSize() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -2000, -1).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(sampleContent()); + assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(1000); + } + + @Test + default void readRangeShouldFailForEmptyObject() { + BlobId blobId = new TestBlobId.Factory().of(UUID.randomUUID().toString()); + Mono.from(testee().save(TEST_BUCKET_NAME, blobId, BlobStoreDAO.BytesBlob.of(new byte[0]))).block(); + + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, blobId, 0, 0).block()) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + default void readRangeShouldFailForMissingObject() { + createSampleBlob(); + BlobId blobId = new TestBlobId.Factory().of(UUID.randomUUID().toString()); + + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, blobId, 0, 9).block()) + .isInstanceOf(ObjectNotFoundException.class); + } + + @Test + default void readRangeShouldReturnWholeObject() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, 999).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(sampleContent()); + } + + @Test + default void readRangeShouldFailWhenStartEqualsObjectSize() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 1000, 1000).block()) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + default void readRangeShouldRejectSuffixWithExplicitEnd() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -100, 99).block()) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + default void readRangeShouldRejectUnrepresentableSuffixLength() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), Long.MIN_VALUE, -1).block()) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + default void readRangeShouldFailForMissingBucket() { + BlobId blobId = createSampleBlob(); + + assertThatThrownBy(() -> testee().readRange(BucketName.of("missing-" + UUID.randomUUID()), blobId, 0, 9).block()) + .isInstanceOf(ObjectNotFoundException.class); + } } diff --git a/server/blob/blob-memory/src/test/java/org/apache/james/blob/memory/MemoryBlobStoreDAOTest.java b/server/blob/blob-memory/src/test/java/org/apache/james/blob/memory/MemoryBlobStoreDAOTest.java index e55fad369d9..96cf88063c0 100644 --- a/server/blob/blob-memory/src/test/java/org/apache/james/blob/memory/MemoryBlobStoreDAOTest.java +++ b/server/blob/blob-memory/src/test/java/org/apache/james/blob/memory/MemoryBlobStoreDAOTest.java @@ -21,11 +21,12 @@ import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; -class MemoryBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +class MemoryBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { private MemoryBlobStoreDAO blobStore; diff --git a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java index 0f7a21ab8d9..45dddc21a84 100644 --- a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java +++ b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java @@ -290,7 +290,7 @@ private static boolean isRangeNotSatisfiable(Throwable throwable) { private BlobMetadata metadataForRange(GetObjectResponse response) { return asBlobMetadata(response.metadata()) - .withMetadata(TOTAL_OBJECT_SIZE, new BlobMetadataValue(Long.toString(extractTotalObjectSize(response)))); + .withTotalObjectSize(extractTotalObjectSize(response)); } private static long extractTotalObjectSize(GetObjectResponse response) { From 8ea019ba036a17d203f096eae730ce332fcc5cbf Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Fri, 16 Oct 2026 09:41:47 +0530 Subject: [PATCH 05/10] JAMES-4231 Reuse S3 DAO test for range read contract --- .../aws/S3BlobStoreDAORangeReadTest.java | 81 ------------------- .../objectstorage/aws/S3BlobStoreDAOTest.java | 3 +- 2 files changed, 2 insertions(+), 82 deletions(-) delete mode 100644 server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java diff --git a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java deleted file mode 100644 index c1645056d95..00000000000 --- a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAORangeReadTest.java +++ /dev/null @@ -1,81 +0,0 @@ -/**************************************************************** - * 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.james.blob.objectstorage.aws; - -import static org.apache.james.blob.objectstorage.aws.JamesS3MetricPublisher.DEFAULT_S3_METRICS_PREFIX; -import static org.apache.james.blob.objectstorage.aws.S3BlobStoreConfiguration.UPLOAD_RETRY_EXCEPTION_PREDICATE; - -import java.time.Duration; -import java.util.Optional; - -import org.apache.james.blob.api.BlobStoreDAO; -import org.apache.james.blob.api.BlobStoreDAORangeReadContract; -import org.apache.james.blob.api.BucketName; -import org.apache.james.blob.api.TestBlobId; -import org.apache.james.metrics.api.NoopGaugeRegistry; -import org.apache.james.metrics.tests.RecordingMetricFactory; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.extension.ExtendWith; - -import reactor.util.retry.Retry; - -@ExtendWith(DockerAwsS3Extension.class) -class S3BlobStoreDAORangeReadTest implements BlobStoreDAORangeReadContract { - private static S3BlobStoreDAO testee; - private static S3ClientFactory clientFactory; - - @BeforeAll - static void setUp(DockerAwsS3Container dockerAwsS3) { - AwsS3AuthConfiguration authConfiguration = AwsS3AuthConfiguration.builder() - .endpoint(dockerAwsS3.getEndpoint()) - .accessKeyId(DockerAwsS3Container.ACCESS_KEY_ID) - .secretKey(DockerAwsS3Container.SECRET_ACCESS_KEY) - .build(); - - S3BlobStoreConfiguration s3Configuration = S3BlobStoreConfiguration.builder() - .authConfiguration(authConfiguration) - .region(dockerAwsS3.dockerAwsS3().region()) - .uploadRetrySpec(Optional.of(Retry.backoff(3, Duration.ofSeconds(1)) - .filter(UPLOAD_RETRY_EXCEPTION_PREDICATE))) - .defaultBucketName(BucketName.DEFAULT) - .build(); - - clientFactory = new S3ClientFactory(s3Configuration, - () -> new JamesS3MetricPublisher(new RecordingMetricFactory(), new NoopGaugeRegistry(), DEFAULT_S3_METRICS_PREFIX)); - testee = new S3BlobStoreDAO(clientFactory, s3Configuration, new TestBlobId.Factory(), S3RequestOption.DEFAULT); - } - - @AfterEach - void tearDown() { - testee.deleteAllBuckets().block(); - } - - @AfterAll - static void tearDownClass() { - clientFactory.close(); - } - - @Override - public BlobStoreDAO testee() { - return testee; - } -} diff --git a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAOTest.java b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAOTest.java index 53c22c08a79..1017d757824 100644 --- a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAOTest.java +++ b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAOTest.java @@ -33,6 +33,7 @@ import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.BucketName; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.apache.james.blob.api.ObjectNotFoundException; @@ -50,7 +51,7 @@ import reactor.util.retry.Retry; @ExtendWith(DockerAwsS3Extension.class) -public class S3BlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +public class S3BlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { private static final BucketName fallbackBucket = BucketName.of("fallback"); private static S3BlobStoreDAO testee; From cefb9a73ea5cfa3cacbfef8e9dc42188b7a9f5b2 Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:52:13 +0530 Subject: [PATCH 06/10] JAMES-4231 Cover range reads across blob store implementations --- .../james/blob/aes/AESBlobStoreDAOTest.java | 5 ++-- .../cassandra/CassandraBlobStoreDAOTest.java | 5 ++-- .../james/blob/file/FileBlobStoreDAOTest.java | 3 ++- .../postgres/PostgresBlobStoreDAOTest.java | 5 ++-- .../blob/zstd/ZstdBlobStoreDAOContract.java | 23 ++++++++++++++++++- 5 files changed, 33 insertions(+), 8 deletions(-) diff --git a/server/blob/blob-aes/src/test/java/org/apache/james/blob/aes/AESBlobStoreDAOTest.java b/server/blob/blob-aes/src/test/java/org/apache/james/blob/aes/AESBlobStoreDAOTest.java index 2cf74dee4e7..d25199420ca 100644 --- a/server/blob/blob-aes/src/test/java/org/apache/james/blob/aes/AESBlobStoreDAOTest.java +++ b/server/blob/blob-aes/src/test/java/org/apache/james/blob/aes/AESBlobStoreDAOTest.java @@ -26,6 +26,7 @@ import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.apache.james.blob.memory.MemoryBlobStoreDAO; import org.junit.jupiter.api.BeforeEach; @@ -34,7 +35,7 @@ import reactor.core.publisher.Mono; -class AESBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +class AESBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { private static final String SAMPLE_SALT = "c603a7327ee3dcbc031d8d34b1096c605feca5e1"; private static final CryptoConfig CRYPTO_CONFIG = CryptoConfig.builder() .salt(SAMPLE_SALT) @@ -87,4 +88,4 @@ void underlyingDataShouldBeEncryptedWhenUsingByteSource() { public void listBucketsShouldReturnBucketsWithNoBlob() { } -} \ No newline at end of file +} diff --git a/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreDAOTest.java b/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreDAOTest.java index 9e829c456be..788b3d5d81d 100644 --- a/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreDAOTest.java +++ b/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/CassandraBlobStoreDAOTest.java @@ -24,6 +24,7 @@ import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.BucketName; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.apache.james.blob.api.TestBlobId; @@ -32,7 +33,7 @@ import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.extension.RegisterExtension; -public class CassandraBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +public class CassandraBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { private static final int CHUNK_SIZE = 10240; @RegisterExtension @@ -66,4 +67,4 @@ public BlobStoreDAO testee() { public void listBucketsShouldReturnBucketsWithNoBlob() { } -} \ No newline at end of file +} diff --git a/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java b/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java index 8de2169cd30..028295db3ab 100644 --- a/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java +++ b/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java @@ -21,6 +21,7 @@ import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.apache.james.blob.api.PlainBlobId; import org.apache.james.server.core.filesystem.FileSystemImpl; @@ -32,7 +33,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; -class FileBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +class FileBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { private FileBlobStoreDAO blobStore; diff --git a/server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreDAOTest.java b/server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreDAOTest.java index 95d17b840cd..c250740fb50 100644 --- a/server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreDAOTest.java +++ b/server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreDAOTest.java @@ -28,6 +28,7 @@ import org.apache.james.backends.postgres.PostgresExtension; import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.apache.james.blob.api.PlainBlobId; import org.apache.james.util.concurrency.ConcurrentTestRunner; @@ -39,7 +40,7 @@ import reactor.core.publisher.Mono; -class PostgresBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +class PostgresBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { static Duration CONCURRENT_TEST_DURATION = Duration.ofMinutes(5); @RegisterExtension @@ -106,4 +107,4 @@ public void concurrentSaveBytesShouldReturnConsistentValues(String description, .operationCount(10) .runSuccessfullyWithin(Duration.ofMinutes(10)); } -} \ No newline at end of file +} diff --git a/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java b/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java index 245b00ad80e..7388d702d29 100644 --- a/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java +++ b/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java @@ -30,11 +30,13 @@ import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.Arrays; import java.util.Random; import java.util.stream.Stream; import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; +import org.apache.james.blob.api.BlobStoreDAORangeReadContract; import org.apache.james.blob.api.MetadataAwareBlobStoreDAOContract; import org.apache.james.metrics.tests.RecordingMetricFactory; import org.junit.jupiter.api.Test; @@ -46,7 +48,7 @@ import reactor.core.publisher.Mono; -public interface ZstdBlobStoreDAOContract extends BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract { +public interface ZstdBlobStoreDAOContract extends BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { BlobStoreDAO underlying(); @@ -61,6 +63,25 @@ default CompressionConfiguration defaultCompressionConfiguration() { .build(); } + @Test + default void readRangeShouldSliceDecompressedContent() throws IOException { + ZstdBlobStoreDAO compressedStore = new ZstdBlobStoreDAO(underlying(), + CompressionConfiguration.builder() + .enabled(true) + .threshold(1) + .build(), + metricFactory()); + Mono.from(compressedStore.save(TEST_BUCKET_NAME, TEST_BLOB_ID, BlobStoreDAO.BytesBlob.of(sampleContent()))).block(); + + assertThat(Mono.from(underlying().readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block().metadata().contentEncoding()) + .contains(BlobStoreDAO.ContentEncoding.ZSTD); + + BlobStoreDAO.Blob slice = compressedStore.readRange(TEST_BUCKET_NAME, TEST_BLOB_ID, 200, 399).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 200, 400)); + assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(sampleContent().length); + } + @Override @Test default void retrieveContentTransferEncodingShouldSucceed() { From 906eeff8d1965d44d2abfd49077ebd2daaeca01a Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Sun, 18 Oct 2026 09:46:34 +0530 Subject: [PATCH 07/10] JAMES-4231 Require non-negative range start offsets --- .../apache/james/blob/api/BlobStoreDAO.java | 22 ++++++------------- .../api/BlobStoreDAORangeReadContract.java | 21 ++++++++---------- .../objectstorage/aws/S3BlobStoreDAO.java | 9 +------- 3 files changed, 17 insertions(+), 35 deletions(-) diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java index 5f9c53b4ed4..bbb2f9ec98d 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java @@ -331,10 +331,9 @@ static long totalObjectSize(Blob blob) { /** * Reads an inclusive byte range while preserving the blob metadata. * - *

When {@code start} is non-negative, {@code end} is either an inclusive byte offset or {@code -1} to read - * through the end of the object. When {@code start} is negative, it requests that many bytes from the end of the - * object and {@code end} must be {@code -1}. For example, {@code readRange(bucket, id, -65536, -1)} reads up to - * the final 64 KiB. Implementations capable of backend range requests should override this method.

+ *

{@code start} must be a non-negative byte offset. {@code end} is either an inclusive byte offset or + * {@code -1} to read through the end of the object. Implementations capable of backend range requests should + * override this method.

* * @return the requested bytes, its original metadata, and {@link #TOTAL_OBJECT_SIZE} metadata * @throws IllegalArgumentException when the range is malformed or starts beyond the end of the object @@ -346,7 +345,7 @@ default Mono readRange(BucketName bucketName, BlobId blobId, long start, l .map(bytesBlob -> { long totalSize = bytesBlob.payload().length; int from = rangeStart(start, totalSize); - int toExclusive = rangeEndExclusive(start, end, totalSize); + int toExclusive = rangeEndExclusive(end, totalSize); BlobMetadata metadata = bytesBlob.metadata().withTotalObjectSize(totalSize); return BytesBlob.of(Arrays.copyOfRange(bytesBlob.payload(), from, toExclusive), metadata); @@ -354,27 +353,20 @@ default Mono readRange(BucketName bucketName, BlobId blobId, long start, l } private static void validateRange(long start, long end) { - if (start < 0) { - Preconditions.checkArgument(start != Long.MIN_VALUE, "'start' must be greater than %s", Long.MIN_VALUE); - Preconditions.checkArgument(end == -1, "'end' must be -1 for a suffix range"); - return; - } + Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); Preconditions.checkArgument(end == -1 || end >= start, "'end' (%s) must be -1 or greater than or equal to 'start' (%s)", end, start); } private static int rangeStart(long start, long totalSize) { Preconditions.checkArgument(totalSize > 0, "Cannot read a range from an empty object"); - if (start < 0) { - return (int) Math.max(0, totalSize + start); - } Preconditions.checkArgument(start < totalSize, "'start' (%s) must be smaller than the object size (%s)", start, totalSize); return (int) start; } - private static int rangeEndExclusive(long start, long end, long totalSize) { - if (start < 0 || end == -1) { + private static int rangeEndExclusive(long end, long totalSize) { + if (end == -1) { return (int) totalSize; } return end >= totalSize - 1 ? (int) totalSize : (int) (end + 1); diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java index 6ee69c889c7..74f76483559 100644 --- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java @@ -69,10 +69,9 @@ default void readRangeShouldReturnAnInclusiveMiddleRange() throws IOException { } @Test - default void readRangeShouldReturnSuffix() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -100, -1).block(); - - assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 900, 1000)); + default void readRangeShouldRejectNegativeStart() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -1, -1).block()) + .isInstanceOf(IllegalArgumentException.class); } @Test @@ -125,11 +124,9 @@ default void readRangeShouldCapEndAtObjectSize() throws IOException { } @Test - default void readRangeShouldCapSuffixAtObjectSize() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -2000, -1).block(); - - assertThat(slice.asBytes().payload()).isEqualTo(sampleContent()); - assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(1000); + default void readRangeShouldRejectNegativeStartWithBoundedEnd() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -1, 0).block()) + .isInstanceOf(IllegalArgumentException.class); } @Test @@ -164,13 +161,13 @@ default void readRangeShouldFailWhenStartEqualsObjectSize() { } @Test - default void readRangeShouldRejectSuffixWithExplicitEnd() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -100, 99).block()) + default void readRangeShouldRejectNegativeEndOtherThanMinusOne() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, -2).block()) .isInstanceOf(IllegalArgumentException.class); } @Test - default void readRangeShouldRejectUnrepresentableSuffixLength() { + default void readRangeShouldRejectMinimumStart() { assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), Long.MIN_VALUE, -1).block()) .isInstanceOf(IllegalArgumentException.class); } diff --git a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java index 45dddc21a84..a99267c7a8d 100644 --- a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java +++ b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java @@ -261,19 +261,12 @@ public Mono readRange(BucketName bucketName, BlobId blobId, long start, lo } private static void validateRange(long start, long end) { - if (start < 0) { - Preconditions.checkArgument(start != Long.MIN_VALUE, "'start' must be greater than %s", Long.MIN_VALUE); - Preconditions.checkArgument(end == -1, "'end' must be -1 for a suffix range"); - return; - } + Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); Preconditions.checkArgument(end == -1 || end >= start, "'end' (%s) must be -1 or greater than or equal to 'start' (%s)", end, start); } private static String formatRangeHeader(long start, long end) { - if (start < 0) { - return "bytes=" + start; - } if (end == -1) { return "bytes=" + start + "-"; } From ba0e480f18aec8efa27677661fb4d5b641f9639f Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Sun, 18 Oct 2026 09:49:47 +0530 Subject: [PATCH 08/10] JAMES-4231 Represent optional range end explicitly --- .../apache/james/blob/api/BlobStoreDAO.java | 22 ++++----- .../api/BlobStoreDAORangeReadContract.java | 47 +++++++++++-------- .../objectstorage/aws/S3BlobStoreDAO.java | 15 +++--- .../blob/zstd/ZstdBlobStoreDAOContract.java | 3 +- 4 files changed, 46 insertions(+), 41 deletions(-) diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java index bbb2f9ec98d..80d67a05a1f 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java @@ -331,14 +331,14 @@ static long totalObjectSize(Blob blob) { /** * Reads an inclusive byte range while preserving the blob metadata. * - *

{@code start} must be a non-negative byte offset. {@code end} is either an inclusive byte offset or - * {@code -1} to read through the end of the object. Implementations capable of backend range requests should - * override this method.

+ *

{@code start} must be a non-negative byte offset. {@code end}, when present, is an inclusive byte offset + * greater than or equal to {@code start}. An empty {@code end} reads through the end of the object. + * Implementations capable of backend range requests should override this method.

* * @return the requested bytes, its original metadata, and {@link #TOTAL_OBJECT_SIZE} metadata * @throws IllegalArgumentException when the range is malformed or starts beyond the end of the object */ - default Mono readRange(BucketName bucketName, BlobId blobId, long start, long end) { + default Mono readRange(BucketName bucketName, BlobId blobId, long start, Optional end) { validateRange(start, end); return Mono.from(readBytes(bucketName, blobId)) @@ -352,10 +352,10 @@ default Mono readRange(BucketName bucketName, BlobId blobId, long start, l }); } - private static void validateRange(long start, long end) { + private static void validateRange(long start, Optional end) { Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); - Preconditions.checkArgument(end == -1 || end >= start, - "'end' (%s) must be -1 or greater than or equal to 'start' (%s)", end, start); + end.ifPresent(value -> Preconditions.checkArgument(value >= start, + "'end' (%s) must be greater than or equal to 'start' (%s)", value, start)); } private static int rangeStart(long start, long totalSize) { @@ -365,10 +365,8 @@ private static int rangeStart(long start, long totalSize) { return (int) start; } - private static int rangeEndExclusive(long end, long totalSize) { - if (end == -1) { - return (int) totalSize; - } - return end >= totalSize - 1 ? (int) totalSize : (int) (end + 1); + private static int rangeEndExclusive(Optional end, long totalSize) { + return end.map(value -> value >= totalSize - 1 ? (int) totalSize : (int) (value + 1)) + .orElse((int) totalSize); } } diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java index 74f76483559..d553e4fd28a 100644 --- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java @@ -26,6 +26,7 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Optional; import java.util.UUID; import org.junit.jupiter.api.Test; @@ -53,7 +54,7 @@ default BlobId createSampleBlob() { @Test default void readRangeShouldReturnAnInclusivePrefixAndPreserveMetadata() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, 99).block(); + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, Optional.of(99L)).block(); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 0, 100)); assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(1000); @@ -63,33 +64,33 @@ default void readRangeShouldReturnAnInclusivePrefixAndPreserveMetadata() throws @Test default void readRangeShouldReturnAnInclusiveMiddleRange() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 200, 399).block(); + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 200, Optional.of(399L)).block(); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 200, 400)); } @Test default void readRangeShouldRejectNegativeStart() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -1, -1).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -1, Optional.empty()).block()) .isInstanceOf(IllegalArgumentException.class); } @Test - default void readRangeShouldReadThroughTheEndWhenEndIsMinusOne() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 900, -1).block(); + default void readRangeShouldReadThroughTheEndWhenEndIsAbsent() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 900, Optional.empty()).block(); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 900, 1000)); } @Test default void readRangeShouldFailWhenStartIsBeyondObjectSize() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 2000, 3000).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 2000, Optional.of(3000L)).block()) .isInstanceOf(IllegalArgumentException.class); } @Test default void readRangeShouldRejectInvalidRange() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 300, 299).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 300, Optional.of(299L)).block()) .isInstanceOf(IllegalArgumentException.class); } @@ -101,7 +102,7 @@ default void readRangeShouldReplaceStoredTotalObjectSize() throws IOException { .withMetadata(new BlobStoreDAO.BlobMetadataName("custom"), new BlobStoreDAO.BlobMetadataValue("retained")); Mono.from(testee().save(TEST_BUCKET_NAME, blobId, BlobStoreDAO.BytesBlob.of(sampleContent(), metadata))).block(); - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, blobId, 0, 9).block(); + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, blobId, 0, Optional.of(9L)).block(); assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(1000); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 0, 10)); @@ -111,21 +112,21 @@ default void readRangeShouldReplaceStoredTotalObjectSize() throws IOException { @Test default void readRangeShouldReturnSingleByte() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 25, 25).block(); + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 25, Optional.of(25L)).block(); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 25, 26)); } @Test default void readRangeShouldCapEndAtObjectSize() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 900, Long.MAX_VALUE).block(); + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 900, Optional.of(Long.MAX_VALUE)).block(); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 900, 1000)); } @Test default void readRangeShouldRejectNegativeStartWithBoundedEnd() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -1, 0).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), -1, Optional.of(0L)).block()) .isInstanceOf(IllegalArgumentException.class); } @@ -134,7 +135,7 @@ default void readRangeShouldFailForEmptyObject() { BlobId blobId = new TestBlobId.Factory().of(UUID.randomUUID().toString()); Mono.from(testee().save(TEST_BUCKET_NAME, blobId, BlobStoreDAO.BytesBlob.of(new byte[0]))).block(); - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, blobId, 0, 0).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, blobId, 0, Optional.of(0L)).block()) .isInstanceOf(IllegalArgumentException.class); } @@ -143,32 +144,40 @@ default void readRangeShouldFailForMissingObject() { createSampleBlob(); BlobId blobId = new TestBlobId.Factory().of(UUID.randomUUID().toString()); - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, blobId, 0, 9).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, blobId, 0, Optional.of(9L)).block()) .isInstanceOf(ObjectNotFoundException.class); } @Test default void readRangeShouldReturnWholeObject() throws IOException { - BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, 999).block(); + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, Optional.of(999L)).block(); assertThat(slice.asBytes().payload()).isEqualTo(sampleContent()); } @Test default void readRangeShouldFailWhenStartEqualsObjectSize() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 1000, 1000).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 1000, Optional.of(1000L)).block()) .isInstanceOf(IllegalArgumentException.class); } @Test - default void readRangeShouldRejectNegativeEndOtherThanMinusOne() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, -2).block()) + default void readRangeShouldReturnWholeObjectWhenEndIsAbsent() throws IOException { + BlobStoreDAO.Blob slice = testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, Optional.empty()).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(sampleContent()); + assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(sampleContent().length); + } + + @Test + default void readRangeShouldRejectNegativeEnd() { + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 0, Optional.of(-1L)).block()) .isInstanceOf(IllegalArgumentException.class); } @Test default void readRangeShouldRejectMinimumStart() { - assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), Long.MIN_VALUE, -1).block()) + assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), Long.MIN_VALUE, Optional.empty()).block()) .isInstanceOf(IllegalArgumentException.class); } @@ -176,7 +185,7 @@ default void readRangeShouldRejectMinimumStart() { default void readRangeShouldFailForMissingBucket() { BlobId blobId = createSampleBlob(); - assertThatThrownBy(() -> testee().readRange(BucketName.of("missing-" + UUID.randomUUID()), blobId, 0, 9).block()) + assertThatThrownBy(() -> testee().readRange(BucketName.of("missing-" + UUID.randomUUID()), blobId, 0, Optional.of(9L)).block()) .isInstanceOf(ObjectNotFoundException.class); } } diff --git a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java index a99267c7a8d..e3947db6733 100644 --- a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java +++ b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java @@ -246,7 +246,7 @@ public Publisher readBytes(BucketName bucketName, BlobId blobId) { } @Override - public Mono readRange(BucketName bucketName, BlobId blobId, long start, long end) { + public Mono readRange(BucketName bucketName, BlobId blobId, long start, java.util.Optional end) { validateRange(start, end); BucketName resolvedBucketName = bucketNameResolver.resolve(bucketName); String rangeHeader = formatRangeHeader(start, end); @@ -260,17 +260,14 @@ public Mono readRange(BucketName bucketName, BlobId blobId, long start, lo .onErrorMap(e -> e.getCause() instanceof OutOfMemoryError, Throwable::getCause); } - private static void validateRange(long start, long end) { + private static void validateRange(long start, java.util.Optional end) { Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); - Preconditions.checkArgument(end == -1 || end >= start, - "'end' (%s) must be -1 or greater than or equal to 'start' (%s)", end, start); + end.ifPresent(value -> Preconditions.checkArgument(value >= start, + "'end' (%s) must be greater than or equal to 'start' (%s)", value, start)); } - private static String formatRangeHeader(long start, long end) { - if (end == -1) { - return "bytes=" + start + "-"; - } - return "bytes=" + start + "-" + end; + private static String formatRangeHeader(long start, java.util.Optional end) { + return "bytes=" + start + "-" + end.map(String::valueOf).orElse(""); } private static boolean isRangeNotSatisfiable(Throwable throwable) { diff --git a/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java b/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java index 7388d702d29..918e6c598f0 100644 --- a/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java +++ b/server/blob/blob-zstd/src/test/java/org/apache/james/blob/zstd/ZstdBlobStoreDAOContract.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.io.InputStream; import java.util.Arrays; +import java.util.Optional; import java.util.Random; import java.util.stream.Stream; @@ -76,7 +77,7 @@ default void readRangeShouldSliceDecompressedContent() throws IOException { assertThat(Mono.from(underlying().readBytes(TEST_BUCKET_NAME, TEST_BLOB_ID)).block().metadata().contentEncoding()) .contains(BlobStoreDAO.ContentEncoding.ZSTD); - BlobStoreDAO.Blob slice = compressedStore.readRange(TEST_BUCKET_NAME, TEST_BLOB_ID, 200, 399).block(); + BlobStoreDAO.Blob slice = compressedStore.readRange(TEST_BUCKET_NAME, TEST_BLOB_ID, 200, Optional.of(399L)).block(); assertThat(slice.asBytes().payload()).isEqualTo(Arrays.copyOfRange(sampleContent(), 200, 400)); assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(sampleContent().length); From 1f1f60afec5f188f657f80506c9195398439cd8a Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Sun, 18 Oct 2026 09:55:14 +0530 Subject: [PATCH 09/10] JAMES-4231 Optimize file blob range reads with positioned IO --- .../james/blob/file/FileBlobStoreDAO.java | 35 ++++++++++++++++ .../james/blob/file/FileBlobStoreDAOTest.java | 42 ++++++++++++++++++- 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java b/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java index b80c1f8d3bb..6272577245c 100644 --- a/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java +++ b/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java @@ -35,10 +35,12 @@ import java.nio.file.NoSuchFileException; import java.nio.file.Path; import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; import java.nio.file.attribute.UserDefinedFileAttributeView; import java.time.Duration; import java.util.Collection; import java.util.Map; +import java.util.Optional; import java.util.stream.Collectors; import java.util.stream.StreamSupport; @@ -124,6 +126,39 @@ public Publisher readBytes(BucketName bucketName, BlobId blobId) { .subscribeOn(Schedulers.boundedElastic()); } + @Override + public Mono readRange(BucketName bucketName, BlobId blobId, long start, Optional end) { + Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); + end.ifPresent(value -> Preconditions.checkArgument(value >= start, + "'end' (%s) must be greater than or equal to 'start' (%s)", value, start)); + + return Mono.fromCallable(() -> { + Path path = getBlobFile(bucketName, blobId).toPath(); + try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) { + long totalSize = channel.size(); + Preconditions.checkArgument(start < totalSize, + "'start' (%s) must be smaller than the object size (%s)", start, totalSize); + long lastByte = end.map(value -> Math.min(value, totalSize - 1)).orElse(totalSize - 1); + long length = lastByte - start + 1; + Preconditions.checkArgument(length <= Integer.MAX_VALUE, "Requested range is too large for an in-memory read: %s", length); + + BlobMetadata metadata = readMetadata(path).withTotalObjectSize(totalSize); + ByteBuffer buffer = ByteBuffer.allocate((int) length); + channel.position(start); + while (buffer.hasRemaining()) { + if (channel.read(buffer) == -1) { + throw new ObjectStoreIOException("Unexpected end of file while reading range from " + path); + } + } + return BytesBlob.of(buffer.array(), metadata); + } + }) + .onErrorMap(NoSuchFileException.class, e -> new ObjectNotFoundException( + String.format("Cannot locate %s within %s", blobId.asString(), bucketName.asString()), e)) + .onErrorMap(IOException.class, e -> new ObjectStoreIOException("IOException occurred while reading blob range", e)) + .subscribeOn(Schedulers.boundedElastic()); + } + @Override public Publisher save(BucketName bucketName, BlobId blobId, Blob blob) { return switch (blob) { diff --git a/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java b/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java index 028295db3ab..39b2968fd0d 100644 --- a/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java +++ b/server/blob/blob-file/src/test/java/org/apache/james/blob/file/FileBlobStoreDAOTest.java @@ -19,6 +19,17 @@ package org.apache.james.blob.file; +import static org.apache.james.blob.api.BlobStoreDAOFixture.TEST_BUCKET_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.Optional; + +import org.apache.james.blob.api.BlobId; import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.BlobStoreDAOContract; import org.apache.james.blob.api.BlobStoreDAORangeReadContract; @@ -36,10 +47,12 @@ class FileBlobStoreDAOTest implements BlobStoreDAOContract, MetadataAwareBlobStoreDAOContract, BlobStoreDAORangeReadContract { private FileBlobStoreDAO blobStore; + private FileSystemImpl fileSystem; @BeforeEach void setUp() throws Exception { - blobStore = new FileBlobStoreDAO(FileSystemImpl.forTesting(), new PlainBlobId.Factory()); + fileSystem = FileSystemImpl.forTesting(); + blobStore = new FileBlobStoreDAO(fileSystem, new PlainBlobId.Factory()); } @Override @@ -47,6 +60,33 @@ public BlobStoreDAO testee() { return blobStore; } + @Test + void readRangeShouldSeekBeyondTwoGigabytesWithoutReadingWholeFile() throws Exception { + BlobId blobId = new PlainBlobId("large-sparse-blob"); + Path bucket = fileSystem.getFile("file://var/blob").toPath().resolve(TEST_BUCKET_NAME.asString()); + Files.createDirectories(bucket); + Path file = bucket.resolve(blobId.asString()); + long offset = (long) Integer.MAX_VALUE + 128; + byte[] content = {1, 2, 3, 4}; + + try { + try (FileChannel channel = FileChannel.open(file, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE, StandardOpenOption.SPARSE)) { + channel.position(offset); + ByteBuffer buffer = ByteBuffer.wrap(content); + while (buffer.hasRemaining()) { + channel.write(buffer); + } + } + + BlobStoreDAO.Blob slice = blobStore.readRange(TEST_BUCKET_NAME, blobId, offset, Optional.of(offset + 3)).block(); + + assertThat(slice.asBytes().payload()).isEqualTo(content); + assertThat(BlobStoreDAO.totalObjectSize(slice)).isEqualTo(offset + content.length); + } finally { + Files.deleteIfExists(file); + } + } + @Override @Test @Disabled("Not supported") From 4855e34c1fd03fa93a285d0d48adad22a487901c Mon Sep 17 00:00:00 2001 From: Hesanda Liyanage <130324291+HesandaLiyanage@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:54:49 +0530 Subject: [PATCH 10/10] JAMES-4231 Share blob range validation across implementations --- .../main/java/org/apache/james/blob/api/BlobStoreDAO.java | 5 ++++- .../java/org/apache/james/blob/file/FileBlobStoreDAO.java | 4 +--- .../james/blob/objectstorage/aws/S3BlobStoreDAO.java | 8 +------- 3 files changed, 6 insertions(+), 11 deletions(-) diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java index 80d67a05a1f..c90b3876f23 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java @@ -352,7 +352,10 @@ default Mono readRange(BucketName bucketName, BlobId blobId, long start, O }); } - private static void validateRange(long start, Optional end) { + /** + * Validates range bounds independently of the object size for default and optimized implementations. + */ + static void validateRange(long start, Optional end) { Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); end.ifPresent(value -> Preconditions.checkArgument(value >= start, "'end' (%s) must be greater than or equal to 'start' (%s)", value, start)); diff --git a/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java b/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java index 6272577245c..a6ba2382c2e 100644 --- a/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java +++ b/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java @@ -128,9 +128,7 @@ public Publisher readBytes(BucketName bucketName, BlobId blobId) { @Override public Mono readRange(BucketName bucketName, BlobId blobId, long start, Optional end) { - Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); - end.ifPresent(value -> Preconditions.checkArgument(value >= start, - "'end' (%s) must be greater than or equal to 'start' (%s)", value, start)); + BlobStoreDAO.validateRange(start, end); return Mono.fromCallable(() -> { Path path = getBlobFile(bucketName, blobId).toPath(); diff --git a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java index e3947db6733..8c5ca298230 100644 --- a/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java +++ b/server/blob/blob-s3/src/main/java/org/apache/james/blob/objectstorage/aws/S3BlobStoreDAO.java @@ -247,7 +247,7 @@ public Publisher readBytes(BucketName bucketName, BlobId blobId) { @Override public Mono readRange(BucketName bucketName, BlobId blobId, long start, java.util.Optional end) { - validateRange(start, end); + BlobStoreDAO.validateRange(start, end); BucketName resolvedBucketName = bucketNameResolver.resolve(bucketName); String rangeHeader = formatRangeHeader(start, end); @@ -260,12 +260,6 @@ public Mono readRange(BucketName bucketName, BlobId blobId, long start, ja .onErrorMap(e -> e.getCause() instanceof OutOfMemoryError, Throwable::getCause); } - private static void validateRange(long start, java.util.Optional end) { - Preconditions.checkArgument(start >= 0, "'start' must be non-negative"); - end.ifPresent(value -> Preconditions.checkArgument(value >= start, - "'end' (%s) must be greater than or equal to 'start' (%s)", value, start)); - } - private static String formatRangeHeader(long start, java.util.Optional end) { return "bytes=" + start + "-" + end.map(String::valueOf).orElse(""); }