JAMES-4231 S3 Object Compaction

docs/modules/servers/partials/architecture/blobstore.adoc

@@ -59,6 +59,13 @@ connectors:
   `content-encoding` and the original size. On reads, it uses this metadata to
   transparently decompress the payload. This reduces storage usage and network
   transfer for compressible blob content.
+* `ChunkedBlobStoreDAO` wraps the storage DAO to provide transparent, low-latency
+  access to blobs compacted into multi-slot chunk objects. When a blob identifier
+  represents a virtual slot within a compacted chunk
+  (`<family>_<generation>_chunk<hash>~<offset>~<limit>`), `ChunkedBlobStoreDAO`
+  translates read requests into precise HTTP byte-range queries
+  (`Range: bytes=offset-limit`) against the underlying object store. Reads of
+  uncompacted standalone blobs pass through directly without overhead.
 
 AES and Zstd can be enabled together. In the Guice binding chain, compression
 wraps encryption: `ZstdBlobStoreDAO` delegates to `AESBlobStoreDAO`, which then
@@ -67,6 +74,30 @@ encrypt afterwards; reads decrypt first and decompress afterwards. This ordering
 preserves the benefit of compression, as encrypted payloads are generally not
 compressible.
 
+== S3 Object Compaction
+
+Storing millions of small objects (such as email headers and small bodies) in S3
+leads to increased storage costs, high request fees (PUT/GET), and slow metadata
+listing operations.
+
+James provides an S3 Object Compaction mechanism to mitigate this:
+
+* *Generation-aware packing*: Historical generation blobs that have become immutable
+  are gathered by `BlobCompactionAlgorithm` and packed into large (~100MB) chunk
+  objects in S3.
+* *Binary Chunk Format*: Chunks store slots containing individual blob payloads
+  compressed with Zstandard (Zstd), slot CRC32 checksums, and a trailing footer
+  with slot offset tables.
+* *Atomic reference updates*: Cassandra message metadata tables (`messageV3`,
+  `imapUidTable`, `messageIdTable`) are updated to point to the new virtual chunk
+  slot IDs before the original standalone blobs are deleted from S3.
+* *Two-stage lifecycle*:
+  ** *Initial Compaction*: Packs standalone candidate blobs of a target generation
+     into chunks and updates references.
+  ** *GC Recompaction*: Rewrites sparse chunks (where slots were marked dead due to
+     message expunges) into new dense chunks to reclaim storage space, and merges
+     undersized chunks.
+
 == Logical buckets
 
 `BucketName` is a James logical namespace in the `BlobStoreDAO` contract. It is

docs/modules/servers/partials/operate/webadmin.adoc

@@ -3176,6 +3176,72 @@ Where:
 filter in later runs.
 - *gcedBlobCount* is the count of blobs that were garbage collected.
 
+== Running blob object compaction
+
+In large installations backed by S3 object storage, storing large numbers of small
+blobs can lead to high S3 request billing and slow bucket listings. Object compaction
+allows administrators to pack standalone blobs from historical generations into large
+chunk objects, serving individual blobs via HTTP ranged reads.
+
+=== Initial Compaction
+
+To compact standalone blobs of a completed generation into chunk objects:
+
+....
+curl -XDELETE "http://ip:port/blobs?scope=initial-compaction&generation=1"
+....
+
+link:#_endpoints_returning_a_task[More details about endpoints returning a task].
+
+Query parameters:
+
+- *generation*: (Compulsory) The integer epoch generation to compact.
+- *family*: (Optional) Storage policy family identifier.
+
+The created task has the following additional information:
+
+....
+{
+  "type": "InitialBlobCompactionTask",
+  "generation": 1,
+  "family": 1,
+  "packedBlobs": 12500,
+  "createdChunks": 5,
+  "freedBytes": 524288000
+}
+....
+
+=== GC Recompaction (Purge & Merge)
+
+When messages are expunged over time, slots within compacted chunks become dead.
+GC recompaction scans existing chunks, purges dead slots by rewriting live slots
+into new compact chunks, and merges undersized chunks:
+
+....
+curl -XDELETE "http://ip:port/blobs?scope=gc-compaction&generation=1"
+....
+
+The created task has the following additional information:
+
+....
+{
+  "type": "GCBlobCompactionTask",
+  "generation": 1,
+  "family": 1,
+  "deadPurged": 450,
+  "chunksMerged": 2,
+  "freedBytes": 20971520
+}
+....
+
+=== Full Compaction
+
+To execute both initial compaction and GC recompaction in a single operation:
+
+....
+curl -XDELETE "http://ip:port/blobs?scope=compaction&generation=1"
+....
+
 endif::[]
 == Administrating Recipient rewriting
 

mailbox/cassandra/pom.xml

@@ -86,6 +86,21 @@
             <artifactId>blob-cassandra</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>blob-compaction</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>blob-compaction</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>blob-memory</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>${james.groupId}</groupId>
             <artifactId>event-bus-api</artifactId>

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraBlobIdRepairer.java

@@ -0,0 +1,215 @@
+/****************************************************************
+ * 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.mailbox.cassandra.mail;
+
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.bindMarker;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.selectFrom;
+import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.update;
+import static com.datastax.oss.driver.api.querybuilder.relation.Relation.column;
+import static com.datastax.oss.driver.api.querybuilder.update.Assignment.setColumn;
+import static org.apache.james.mailbox.cassandra.table.CassandraMessageIds.IMAP_UID;
+import static org.apache.james.mailbox.cassandra.table.CassandraMessageIds.MAILBOX_ID;
+import static org.apache.james.mailbox.cassandra.table.CassandraMessageIds.MESSAGE_ID;
+import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.BODY_CONTENT;
+import static org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.HEADER_CONTENT;
+
+import java.util.UUID;
+
+import jakarta.inject.Inject;
+
+import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
+import org.apache.james.blob.api.BlobId;
+import org.apache.james.blob.api.BucketName;
+import org.apache.james.blob.compaction.BlobIdRepairer;
+import org.apache.james.mailbox.cassandra.table.CassandraMessageIdTable;
+import org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table;
+import org.apache.james.mailbox.cassandra.table.MessageIdToImapUid;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.PreparedStatement;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+
+import reactor.core.publisher.Mono;
+
+public class CassandraBlobIdRepairer implements BlobIdRepairer {
+    private static final Logger LOGGER = LoggerFactory.getLogger(CassandraBlobIdRepairer.class);
+
+    private final CassandraAsyncExecutor cassandraAsyncExecutor;
+    private final BlobId.Factory blobIdFactory;
+
+    private final PreparedStatement scanMessageIdTable;
+    private final PreparedStatement selectImapUidSingle;
+    private final PreparedStatement updateMessageIdTableHeader;
+
+    private final PreparedStatement scanImapUidTable;
+    private final PreparedStatement selectMessageIdTableSingle;
+    private final PreparedStatement updateImapUidTableHeader;
+
+    private final PreparedStatement scanMessageV3;
+    private final PreparedStatement selectImapUidByMessageId;
+    private final PreparedStatement updateMessageV3Header;
+
+    @Inject
+    public CassandraBlobIdRepairer(CqlSession session, BlobId.Factory blobIdFactory) {
+        this.cassandraAsyncExecutor = new CassandraAsyncExecutor(session);
+        this.blobIdFactory = blobIdFactory;
+
+        this.scanMessageIdTable = session.prepare(selectFrom(CassandraMessageIdTable.TABLE_NAME)
+            .columns(MAILBOX_ID, IMAP_UID, MESSAGE_ID, HEADER_CONTENT)
+            .limit(1000)
+            .build());
+
+        this.selectImapUidSingle = session.prepare(selectFrom(MessageIdToImapUid.TABLE_NAME)
+            .column(HEADER_CONTENT)
+            .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)),
+                column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)),
+                column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID)))
+            .build());
+
+        this.updateMessageIdTableHeader = session.prepare(update(CassandraMessageIdTable.TABLE_NAME)
+            .set(setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT)))
+            .where(column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)),
+                column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID)))
+            .build());
+
+        this.scanImapUidTable = session.prepare(selectFrom(MessageIdToImapUid.TABLE_NAME)
+            .columns(MESSAGE_ID, MAILBOX_ID, IMAP_UID, HEADER_CONTENT)
+            .limit(1000)
+            .build());
+
+        this.selectMessageIdTableSingle = session.prepare(selectFrom(CassandraMessageIdTable.TABLE_NAME)
+            .column(HEADER_CONTENT)
+            .where(column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)),
+                column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID)))
+            .build());
+
+        this.updateImapUidTableHeader = session.prepare(update(MessageIdToImapUid.TABLE_NAME)
+            .set(setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT)))
+            .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)),
+                column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)),
+                column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID)))
+            .build());
+
+        this.scanMessageV3 = session.prepare(selectFrom(CassandraMessageV3Table.TABLE_NAME)
+            .columns(MESSAGE_ID, HEADER_CONTENT, BODY_CONTENT)
+            .limit(1000)
+            .build());
+
+        this.selectImapUidByMessageId = session.prepare(selectFrom(MessageIdToImapUid.TABLE_NAME)
+            .column(HEADER_CONTENT)
+            .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)))
+            .build());
+
+        this.updateMessageV3Header = session.prepare(update(CassandraMessageV3Table.TABLE_NAME)
+            .set(setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT)))
+            .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)))
+            .build());
+    }
+
+    @Override
+    public Mono<BlobId> repair(BucketName bucketName, BlobId staleBlobId) {
+        String staleStr = staleBlobId.asString();
+
+        return repairFromMessageIdTable(staleStr)
+            .switchIfEmpty(repairFromImapUidTable(staleStr))
+            .switchIfEmpty(repairFromMessageV3(staleStr))
+            .doOnNext(repairedId -> LOGGER.info("Successfully repaired stale blob ID {} with canonical ID {}", staleStr, repairedId.asString()));
+    }
+
+    private Mono<BlobId> repairFromMessageIdTable(String staleStr) {
+        return cassandraAsyncExecutor.executeRows(scanMessageIdTable.bind())
+            .filter(row -> staleStr.equals(row.get(HEADER_CONTENT, TypeCodecs.TEXT)))
+            .concatMap(row -> {
+                UUID mailboxId = row.get(MAILBOX_ID, TypeCodecs.UUID);
+                Long imapUid = row.get(IMAP_UID, TypeCodecs.BIGINT);
+                UUID messageId = row.get(MESSAGE_ID, TypeCodecs.TIMEUUID);
+
+                return cassandraAsyncExecutor.executeSingleRow(selectImapUidSingle.bind()
+                        .set(MESSAGE_ID, messageId, TypeCodecs.TIMEUUID)
+                        .set(MAILBOX_ID, mailboxId, TypeCodecs.UUID)
+                        .set(IMAP_UID, imapUid, TypeCodecs.BIGINT))
+                    .flatMap(imapRow -> {
+                        String canonical = imapRow.get(HEADER_CONTENT, TypeCodecs.TEXT);
+                        if (canonical != null && !canonical.equals(staleStr)) {
+                            return cassandraAsyncExecutor.executeVoid(updateMessageIdTableHeader.bind()
+                                    .set(HEADER_CONTENT, canonical, TypeCodecs.TEXT)
+                                    .set(MAILBOX_ID, mailboxId, TypeCodecs.UUID)
+                                    .set(IMAP_UID, imapUid, TypeCodecs.BIGINT))
+                                .thenReturn(blobIdFactory.parse(canonical));
+                        }
+                        return Mono.empty();
+                    });
+            })
+            .next();
+    }
+
+    private Mono<BlobId> repairFromImapUidTable(String staleStr) {
+        return cassandraAsyncExecutor.executeRows(scanImapUidTable.bind())
+            .filter(row -> staleStr.equals(row.get(HEADER_CONTENT, TypeCodecs.TEXT)))
+            .concatMap(row -> {
+                UUID messageId = row.get(MESSAGE_ID, TypeCodecs.TIMEUUID);
+                UUID mailboxId = row.get(MAILBOX_ID, TypeCodecs.UUID);
+                Long imapUid = row.get(IMAP_UID, TypeCodecs.BIGINT);
+
+                return cassandraAsyncExecutor.executeSingleRow(selectMessageIdTableSingle.bind()
+                        .set(MAILBOX_ID, mailboxId, TypeCodecs.UUID)
+                        .set(IMAP_UID, imapUid, TypeCodecs.BIGINT))
+                    .flatMap(msgRow -> {
+                        String canonical = msgRow.get(HEADER_CONTENT, TypeCodecs.TEXT);
+                        if (canonical != null && !canonical.equals(staleStr)) {
+                            return cassandraAsyncExecutor.executeVoid(updateImapUidTableHeader.bind()
+                                    .set(HEADER_CONTENT, canonical, TypeCodecs.TEXT)
+                                    .set(MESSAGE_ID, messageId, TypeCodecs.TIMEUUID)
+                                    .set(MAILBOX_ID, mailboxId, TypeCodecs.UUID)
+                                    .set(IMAP_UID, imapUid, TypeCodecs.BIGINT))
+                                .thenReturn(blobIdFactory.parse(canonical));
+                        }
+                        return Mono.empty();
+                    });
+            })
+            .next();
+    }
+
+    private Mono<BlobId> repairFromMessageV3(String staleStr) {
+        return cassandraAsyncExecutor.executeRows(scanMessageV3.bind())
+            .filter(row -> staleStr.equals(row.get(HEADER_CONTENT, TypeCodecs.TEXT)))
+            .concatMap(row -> {
+                UUID messageId = row.get(MESSAGE_ID, TypeCodecs.TIMEUUID);
+
+                return cassandraAsyncExecutor.executeRows(selectImapUidByMessageId.bind()
+                        .set(MESSAGE_ID, messageId, TypeCodecs.TIMEUUID))
+                    .filter(imapRow -> {
+                        String canonical = imapRow.get(HEADER_CONTENT, TypeCodecs.TEXT);
+                        return canonical != null && !canonical.equals(staleStr);
+                    })
+                    .next()
+                    .flatMap(imapRow -> {
+                        String canonical = imapRow.get(HEADER_CONTENT, TypeCodecs.TEXT);
+                        return cassandraAsyncExecutor.executeVoid(updateMessageV3Header.bind()
+                                .set(HEADER_CONTENT, canonical, TypeCodecs.TEXT)
+                                .set(MESSAGE_ID, messageId, TypeCodecs.TIMEUUID))
+                            .thenReturn(blobIdFactory.parse(canonical));
+                    });
+            })
+            .next();
+    }
+}