JAMES-4209 Improve S3 data recovery handling + JAMES-4224 entropy 256 -> 128 bits

backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java

@@ -40,14 +40,13 @@ public class CassandraConfiguration {
     private static final Logger LOGGER = LoggerFactory.getLogger(CassandraConfiguration.class);
 
     public enum BlobRecoveryMode {
-        NONE, SYNCHRONOUS, ASYNCHRONOUS;
+        NONE, ENABLED;
 
         public static BlobRecoveryMode parse(String value) {
             return switch (value.toLowerCase()) {
                 case "none" -> NONE;
-                case "synchronous" -> SYNCHRONOUS;
-                case "asynchronous" -> ASYNCHRONOUS;
-                default -> throw new IllegalArgumentException("Unknown blob recovery mode: '" + value + "'. Expected none, synchronous or asynchronous");
+                case "enabled" -> ENABLED;
+                default -> throw new IllegalArgumentException("Unknown blob recovery mode: '" + value + "'. Expected none or enabled");
             };
         }
     }

docs/modules/servers/pages/distributed/operate/backup.adoc

@@ -119,28 +119,38 @@ same API. Consult your provider's documentation for the exact tooling.
 == Message content recovery from S3
 
 Even with versioning, you may face the worst case: *the Cassandra metadata is lost but the object store
-survives*. Because blobs alone do not tell which user a message belonged to, James can optionally write,
-next to each stored message, a small `recovery/<headerBlobId>` *sidecar* blob holding the matching
-`bodyBlobId`. The blob garbage collection is aware of these sidecars and never deletes a live one.
+survives*. Because blobs alone do not tell which user a message belonged to, James can optionally record
+the id of a message body as *metadata of its header blob*, and suffix header blob ids with `_hdr` so that
+they can be told apart when walking the store. No extra object is written, and the blob garbage collection
+needs no special handling.
 
-Recording of the sidecars is controlled in `cassandra.properties`:
+Recording of the recovery information is controlled in `cassandra.properties`:
 
 [source,properties]
 ----
-# none (default), synchronous or asynchronous
-mailbox.blob.recovery.mode=synchronous
+# none (default) or enabled
+mailbox.blob.recovery.mode=enabled
 ----
 
-* `synchronous` &mdash; the sidecar is written as part of message storage; a failure fails the delivery.
-* `asynchronous` &mdash; the sidecar is written in the background; failures are only logged.
-* `none` &mdash; no sidecar is written, and content recovery is not possible.
+* `enabled` &mdash; the body blob id is written along with the header blob, as part of message storage;
+a failure fails the delivery.
+* `none` &mdash; nothing is recorded, and content recovery is not possible.
+
+==== What enabling it costs
+
+`enabled` is not free. James can no longer write the two blobs of a message in parallel: the body has to
+be stored first, so that its blob id can be attached to the header blob as metadata. **Saving a mail
+becomes a sequential body-then-header operation**, and write latency grows by one full object round trip
+on every append path &mdash; LMTP delivery, IMAP APPEND, JMAP import. Size that against your delivery
+latency budget before turning it on.
 
 When recovery is needed, run the dedicated `org.apache.james.S3RecoveryMain` entrypoint. It reuses the
 regular mailbox, DAO and blob store modules and the existing `blobstore.properties` (so AES encryption
 and compression are applied transparently), but starts neither the protocol servers nor RabbitMQ. It
-walks the object store, reads each `recovery/` sidecar, rebuilds the message from its header and body
-blobs, reads the `Delivered-To` recipients, and appends the message into a `Restored-messages` mailbox
-of each local recipient.
+walks the header blobs of the object store, reads the body blob id from their metadata, rebuilds each
+message from its header and body blobs, reads the `Delivered-To` recipients, and appends the message into
+a `Restored-messages` mailbox of each local recipient. A blob whose id ends in `_hdr` but that carries no
+recovery metadata is simply skipped and counted apart.
 
 Run it by overriding the container entrypoint main class (Cassandra and S3 must be reachable):
 
@@ -163,19 +173,25 @@ restricts recovery to messages whose `Date` header is strictly after the given i
 ... org.apache.james.S3RecoveryMain --restore-after=2026-01-01T00:00:00Z
 ----
 
-An optional `--header-blob-prefix=<prefix>` argument (also settable via the
-`RECOVERY_HEADER_BLOB_PREFIX` environment variable or the `recovery.header.blob.prefix` system
-property) narrows the walk to the recovery sidecars whose header blob id starts with the given prefix,
-pushing the filter down to S3's `ListObjectsV2`. Header blob ids are generation-aware
-(`family_generation_...`), so a clever admin can pass e.g. `1_42_` to iterate solely the latest
-generation instead of scanning the whole bucket:
+The optional `--family=<int>` and `--generation=<long>` arguments (also settable via the
+`RECOVERY_FAMILY` and `RECOVERY_GENERATION` environment variables, or the `recovery.family` and
+`recovery.generation` system properties) narrow the walk to a single generation of the generation-aware
+blob ids (`family_generation_...`), pushing the filter down to S3's `ListObjectsV2` rather than filtering
+a full listing. Both default to walking the whole bucket.
+
+Because the generation is the *second* component of a blob id, there is no listing prefix for a generation
+on its own: `--generation` requires `--family`.
 
 [source,bash]
 ----
-... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_
+... org.apache.james.S3RecoveryMain --family=1 --generation=42
 ----
 
-The dominant cost of a recovery run is not the listing but the per-message work: three blob reads plus
+Deployments configured with the MinIO blob id strategy separate the family and generation with `/` rather
+than `_`. Declare it with `--minio-separator` (or `RECOVERY_MINIO_SEPARATOR`, or
+`recovery.minio.separator`), which turns the prefix above into `1/42/`.
+
+The dominant cost of a recovery run is not the listing but the per-message work: two blob reads plus
 a full re-store of each message through the mailbox. The `--concurrency=<n>` argument (default `8`, also
 settable via the `RECOVERY_CONCURRENCY` environment variable or the `recovery.concurrency` system
 property) controls how many messages are restored in parallel and is therefore the main lever on
@@ -190,16 +206,16 @@ back off if they become the bottleneck:
 === Scaling recovery: shard by generation
 
 For very large recoveries, a single process is limited by its own concurrency and offers no easy resume
-point. Because `--header-blob-prefix` scopes a run to a slice of the key space, you can *shard* the
-recovery across several independent processes &mdash; typically one per blob generation
-(`family_generation_`) &mdash; and run them in parallel, each with its own concurrency:
+point. Because `--family` and `--generation` scope a run to a slice of the key space, you can *shard* the
+recovery across several independent processes &mdash; typically one per blob generation &mdash; and run
+them in parallel, each with its own concurrency:
 
 [source,bash]
 ----
 # On different hosts / containers, in parallel
-... org.apache.james.S3RecoveryMain --header-blob-prefix=1_40_ --concurrency=16
-... org.apache.james.S3RecoveryMain --header-blob-prefix=1_41_ --concurrency=16
-... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ --concurrency=16
+... org.apache.james.S3RecoveryMain --family=1 --generation=40 --concurrency=16
+... org.apache.james.S3RecoveryMain --family=1 --generation=41 --concurrency=16
+... org.apache.james.S3RecoveryMain --family=1 --generation=42 --concurrency=16
 ----
 
 The shards are disjoint (a given header blob belongs to exactly one generation), so they never restore
@@ -208,7 +224,7 @@ can be re-run on its own without redoing the others.
 
 Notes:
 
-* Restored messages are re-stored (and get a fresh `recovery/` sidecar), so re-running the recovery
+* Restored messages are re-stored, and thus get a fresh header blob of their own, so re-running the recovery
 restores them again. Restore into an empty deployment, or clean up between runs.
 * The search index is not populated during recovery. Run a
 xref:distributed/operate/cli.adoc#_re_indexing[re-indexing] afterwards if search is needed.

docs/modules/servers/partials/configure/jvm.adoc

@@ -99,6 +99,29 @@ lines at most.
 Lowering it only affects previews computed afterwards. Those already stored keep their length until the
 message is reindexed, so the space comes back progressively rather than at once.
 
+== Change the entropy of the blobId
+
+By default a blobId carries 128 bits of entropy: the leading 128 bits of the SHA-256 of the content it
+addresses for deduplicated blobs, and as many random bits for the randomly generated ones. The property
+`james.blobid.entropy` changes that, in bits.
+
+Ex in `jvm.properties`
+----
+james.blobid.entropy=256
+----
+
+Optional. Integer, a multiple of 8 within [96, 256]. Defaults to 128.
+
+The length of an id is paid everywhere it is stored. The object key itself, but above all the Cassandra
+columns referencing it, which hold one id per message rather than one per blob: 256 bits takes a body
+blob id from 28 to 50 characters. Ids below the full 256 bits are also left unpadded, which is worth two
+more characters.
+
+128 bits is ample for a content addressed store. The birthday bound puts a collision at `n^2/2^129`, ie.
+1.5e-19 for ten billion blobs, twenty orders of magnitude below the silent error rate of the storage
+underneath; and truncating a cryptographic hash to its leading bits is standard practice (NIST SP
+800-107, FIPS 180-4). `256` spells ids out the way releases up to 3.9.x did.
+
 == Improve listing support for MinIO
 
 Due to blobs being stored in folder, adding `/` in blobs name emulates folder and avoids blobs to be all stored in a
@@ -304,4 +327,4 @@ Ex in  `jvm.properties`
 james.mailbox.handleRecent=false
 ----
 
-Defaults to true (no breaking changes)
+Defaults to true (no breaking changes)

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

@@ -41,7 +41,6 @@
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
 import java.util.Date;
 import java.util.List;
 import java.util.Optional;
@@ -56,6 +55,7 @@
 import org.apache.james.backends.cassandra.utils.ProfileLocator;
 import org.apache.james.blob.api.BlobId;
 import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreCacheCallback;
 import org.apache.james.blob.api.BlobStoreDAO;
 import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
 import org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.Attachments;
@@ -68,8 +68,6 @@
 import org.apache.james.mailbox.model.StringBackedAttachmentId;
 import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
 import org.apache.james.mailbox.store.mail.model.MailboxMessage;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
@@ -92,13 +90,12 @@
 import reactor.util.function.Tuple2;
 
 public class CassandraMessageDAOV3 {
-    private static final Logger LOGGER = LoggerFactory.getLogger(CassandraMessageDAOV3.class);
     private static final byte[] EMPTY_BYTE_ARRAY = {};
 
     private final CassandraAsyncExecutor cassandraAsyncExecutor;
     private final BlobStore blobStore;
-    private final BlobStoreDAO blobStoreDAO;
     private final BlobId.Factory blobIdFactory;
+    private final MessageContentSaver messageContentSaver;
     private final PreparedStatement insert;
     private final PreparedStatement delete;
     private final PreparedStatement select;
@@ -114,11 +111,11 @@ public class CassandraMessageDAOV3 {
     @Inject
     public CassandraMessageDAOV3(CqlSession session, CassandraTypesProvider typesProvider, BlobStore blobStore,
                                  BlobStoreDAO blobStoreDAO, BlobId.Factory blobIdFactory,
-                                 CassandraConfiguration cassandraConfiguration) {
+                                 CassandraConfiguration cassandraConfiguration, BlobStoreCacheCallback cacheCallback) {
         this.cassandraAsyncExecutor = new CassandraAsyncExecutor(session);
         this.blobStore = blobStore;
-        this.blobStoreDAO = blobStoreDAO;
         this.blobIdFactory = blobIdFactory;
+        this.messageContentSaver = messageContentSaver(blobStore, blobStoreDAO, blobIdFactory, cassandraConfiguration, cacheCallback);
 
         this.insert = prepareInsert(session);
         this.delete = prepareDelete(session);
@@ -134,6 +131,15 @@ public CassandraMessageDAOV3(CqlSession session, CassandraTypesProvider typesPro
         this.optimisticConsistencyLevelProfile = JamesExecutionProfiles.getOptimisticConsistencyLevelProfile(session);
     }
 
+    private static MessageContentSaver messageContentSaver(BlobStore blobStore, BlobStoreDAO blobStoreDAO,
+                                                          BlobId.Factory blobIdFactory, CassandraConfiguration configuration,
+                                                          BlobStoreCacheCallback cacheCallback) {
+        return switch (configuration.getBlobRecoveryMode()) {
+            case NONE -> new DefaultMessageContentSaver(blobStore);
+            case ENABLED -> new ContentRecoveryMessageContentSaver(blobStore, blobStoreDAO, blobIdFactory, cacheCallback);
+        };
+    }
+
     private PreparedStatement prepareSelect(CqlSession session) {
         return session.prepare(selectFrom(TABLE_NAME)
             .all()
@@ -210,31 +216,10 @@ public long size() {
                     }
                 };
 
-                Mono<BlobId> headerFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), headerContent, SIZE_BASED));
-                Mono<BlobId> bodyFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource, LOW_COST));
-
-                return headerFuture.zipWith(bodyFuture)
-                    .flatMap(pair -> saveRecovery(pair.getT1(), pair.getT2()).thenReturn(pair));
+                return messageContentSaver.saveContent(headerContent, bodyByteSource);
             });
     }
 
-    private Mono<Void> saveRecovery(BlobId headerId, BlobId bodyId) {
-        return switch (configuration.getBlobRecoveryMode()) {
-            case NONE -> Mono.empty();
-            case SYNCHRONOUS -> writeRecoveryBlob(headerId, bodyId);
-            case ASYNCHRONOUS -> Mono.fromRunnable(() ->
-                writeRecoveryBlob(headerId, bodyId)
-                    .subscribeOn(Schedulers.parallel())
-                    .subscribe(ignored -> { }, e -> LOGGER.error("Failed to save recovery blob for header={} body={}", headerId.asString(), bodyId.asString(), e)));
-        };
-    }
-
-    private Mono<Void> writeRecoveryBlob(BlobId headerId, BlobId bodyId) {
-        BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + headerId.asString());
-        BlobStoreDAO.BytesBlob content = BlobStoreDAO.BytesBlob.of(bodyId.asString().getBytes(StandardCharsets.UTF_8));
-        return Mono.from(blobStoreDAO.save(blobStore.getDefaultBucketName(), recoveryBlobId, content));
-    }
-
     private BoundStatement boundWriteStatement(MailboxMessage message, Tuple2<BlobId, BlobId> pair) {
         CassandraMessageId messageId = (CassandraMessageId) message.getMessageId();
         BoundStatement boundStatement = insert.bind()