JAMES-4231 Add S3 and file blob range reads

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() {
 
     }
-}
+}

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.
@@ -115,6 +116,13 @@ public Optional<ContentEncoding> contentEncoding() {
         public BlobMetadata withContentEncoding(ContentEncoding contentEncoding) {
             return withMetadata(ContentEncoding.NAME, contentEncoding.asValue());
         }
+
+        public BlobMetadata withTotalObjectSize(long totalObjectSize) {
+            return new BlobMetadata(ImmutableMap.<BlobMetadataName, BlobMetadataValue>builder()
+                .putAll(underlyingMap)
+                .put(TOTAL_OBJECT_SIZE, new BlobMetadataValue(Long.toString(totalObjectSize)))
+                .buildKeepingLast());
+        }
     }
 
 
@@ -303,4 +311,65 @@ default Publisher<BlobId> 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.
+     *
+     * <p>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.</p>
+     */
+    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.
+     *
+     * <p>{@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.</p>
+     *
+     * @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<Blob> readRange(BucketName bucketName, BlobId blobId, long start, Optional<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(end, totalSize);
+
+                BlobMetadata metadata = bytesBlob.metadata().withTotalObjectSize(totalSize);
+                return BytesBlob.of(Arrays.copyOfRange(bytesBlob.payload(), from, toExclusive), metadata);
+            });
+    }
+
+    /**
+     * Validates range bounds independently of the object size for default and optimized implementations.
+     */
+    static void validateRange(long start, Optional<Long> 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 int rangeStart(long start, long totalSize) {
+        Preconditions.checkArgument(totalSize > 0, "Cannot read a range from an empty object");
+        Preconditions.checkArgument(start < totalSize,
+            "'start' (%s) must be smaller than the object size (%s)", start, totalSize);
+        return (int) start;
+    }
+
+    private static int rangeEndExclusive(Optional<Long> end, long totalSize) {
+        return end.map(value -> value >= totalSize - 1 ? (int) totalSize : (int) (value + 1))
+            .orElse((int) totalSize);
+    }
 }

server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobStoreDAORangeReadContract.java

@@ -0,0 +1,191 @@
+/****************************************************************
+ * 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.Optional;
+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, Optional.of(99L)).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, 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, Optional.empty()).block())
+            .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    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, Optional.of(3000L)).block())
+            .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    default void readRangeShouldRejectInvalidRange() {
+        assertThatThrownBy(() -> testee().readRange(TEST_BUCKET_NAME, createSampleBlob(), 300, Optional.of(299L)).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, Optional.of(9L)).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, 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, 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, Optional.of(0L)).block())
+            .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @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, Optional.of(0L)).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, Optional.of(9L)).block())
+            .isInstanceOf(ObjectNotFoundException.class);
+    }
+
+    @Test
+    default void readRangeShouldReturnWholeObject() throws IOException {
+        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, Optional.of(1000L)).block())
+            .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    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, Optional.empty()).block())
+            .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test
+    default void readRangeShouldFailForMissingBucket() {
+        BlobId blobId = createSampleBlob();
+
+        assertThatThrownBy(() -> testee().readRange(BucketName.of("missing-" + UUID.randomUUID()), blobId, 0, Optional.of(9L)).block())
+            .isInstanceOf(ObjectNotFoundException.class);
+    }
+}

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() {
 
     }
-}
+}

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,37 @@ public Publisher<BytesBlob> readBytes(BucketName bucketName, BlobId blobId) {
             .subscribeOn(Schedulers.boundedElastic());
     }
 
+    @Override
+    public Mono<Blob> readRange(BucketName bucketName, BlobId blobId, long start, Optional<Long> end) {
+        BlobStoreDAO.validateRange(start, end);
+
+        return Mono.<Blob>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<Void> save(BucketName bucketName, BlobId blobId, Blob blob) {
         return switch (blob) {