diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java index 9c0ffb7e5ce..e4b272b46d5 100644 --- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java +++ b/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"); }; } } diff --git a/docs/modules/servers/pages/distributed/operate/backup.adoc b/docs/modules/servers/pages/distributed/operate/backup.adoc index 92550ab1743..ea508c461af 100644 --- a/docs/modules/servers/pages/distributed/operate/backup.adoc +++ b/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/` *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` — the sidecar is written as part of message storage; a failure fails the delivery. -* `asynchronous` — the sidecar is written in the background; failures are only logged. -* `none` — no sidecar is written, and content recovery is not possible. +* `enabled` — the body blob id is written along with the header blob, as part of message storage; +a failure fails the delivery. +* `none` — 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 — 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=` 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=` and `--generation=` 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=` 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 — typically one per blob generation -(`family_generation_`) — 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 — typically one per blob generation — 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. diff --git a/docs/modules/servers/partials/configure/jvm.adoc b/docs/modules/servers/partials/configure/jvm.adoc index e2418d40af3..59b2c1fe01f 100644 --- a/docs/modules/servers/partials/configure/jvm.adoc +++ b/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) \ No newline at end of file +Defaults to true (no breaking changes) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java index 0b2eea3657a..375d61bc93e 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java +++ b/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 headerFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), headerContent, SIZE_BASED)); - Mono 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 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 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 pair) { CassandraMessageId messageId = (CassandraMessageId) message.getMessageId(); BoundStatement boundStatement = insert.bind() diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaver.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaver.java new file mode 100644 index 00000000000..ccf944732d1 --- /dev/null +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaver.java @@ -0,0 +1,95 @@ +/**************************************************************** + * 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 org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST; + +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.blob.api.BlobStoreDAO.BlobMetadata; +import org.apache.james.blob.api.BlobStoreDAO.BlobMetadataName; +import org.apache.james.blob.api.BlobStoreDAO.BlobMetadataValue; +import org.apache.james.blob.api.BlobStoreDAO.BytesBlob; + +import com.google.common.io.ByteSource; + +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; +import reactor.util.function.Tuples; + +/** + * Carries the recovery information within the header blob itself, rather than in a companion object. + * + *

The body is written first, then the headers are written under a randomly generated blob id carrying + * the body blob id as metadata. Recovering a message therefore only requires walking the header blobs of + * the bucket: the {@value #HEADER_BLOB_ID_SUFFIX} suffix tells them apart, and the + * {@code body-blob-id} metadata points at their body.

+ * + *

The header blob id is random rather than content addressed, so that a header and its recovery + * information stay paired. Headers are consequently not deduplicated, which they hardly ever were.

+ * + *

Headers go through the {@link BlobStoreDAO} rather than the {@link BlobStore} because only the + * former exposes metadata. That DAO is the decorated one, so compression and encryption still apply; the + * caching a {@code SIZE_BASED} save would have performed is restored by {@link BlobStoreCacheCallback}.

+ */ +public class ContentRecoveryMessageContentSaver implements MessageContentSaver { + public static final String HEADER_BLOB_ID_SUFFIX = "_hdr"; + public static final BlobMetadataName BODY_BLOB_ID = new BlobMetadataName("body-blob-id"); + + private final BlobStore blobStore; + private final BlobStoreDAO blobStoreDAO; + private final BlobId.Factory blobIdFactory; + private final BlobStoreCacheCallback cacheCallback; + + public ContentRecoveryMessageContentSaver(BlobStore blobStore, BlobStoreDAO blobStoreDAO, + BlobId.Factory blobIdFactory, BlobStoreCacheCallback cacheCallback) { + this.blobStore = blobStore; + this.blobStoreDAO = blobStoreDAO; + this.blobIdFactory = blobIdFactory; + this.cacheCallback = cacheCallback; + } + + @Override + public Mono> saveContent(byte[] headerBytes, ByteSource bodyByteSource) { + return Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource, LOW_COST)) + .flatMap(bodyId -> saveHeaders(headerBytes, bodyId) + .map(headerId -> Tuples.of(headerId, bodyId))); + } + + private Mono saveHeaders(byte[] headerBytes, BlobId bodyId) { + BlobId headerId = generateHeaderBlobId(); + BlobMetadata metadata = BlobMetadata.empty() + .withMetadata(BODY_BLOB_ID, new BlobMetadataValue(bodyId.asString())); + + return Mono.from(blobStoreDAO.save(blobStore.getDefaultBucketName(), headerId, BytesBlob.of(headerBytes, metadata))) + .then(Mono.from(cacheCallback.cacheIfNeeded(headerId, headerBytes))) + .thenReturn(headerId); + } + + /** + * Leaves the family and generation prefixes to the configured {@link BlobId.Factory}, so that the + * header blob stays generation aware and is garbage collected like any other blob. + */ + private BlobId generateHeaderBlobId() { + return blobIdFactory.random().withSuffix(HEADER_BLOB_ID_SUFFIX); + } +} diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/DefaultMessageContentSaver.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/DefaultMessageContentSaver.java new file mode 100644 index 00000000000..b575547df1e --- /dev/null +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/DefaultMessageContentSaver.java @@ -0,0 +1,53 @@ +/**************************************************************** + * 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 org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST; +import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED; + +import jakarta.inject.Inject; + +import org.apache.james.blob.api.BlobId; +import org.apache.james.blob.api.BlobStore; + +import com.google.common.io.ByteSource; + +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; + +/** + * Writes the headers and the body as two distinct blobs, without any recovery information. + */ +public class DefaultMessageContentSaver implements MessageContentSaver { + private final BlobStore blobStore; + + @Inject + public DefaultMessageContentSaver(BlobStore blobStore) { + this.blobStore = blobStore; + } + + @Override + public Mono> saveContent(byte[] headerBytes, ByteSource bodyByteSource) { + Mono headerFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), headerBytes, SIZE_BASED)); + Mono bodyFuture = Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource, LOW_COST)); + + return headerFuture.zipWith(bodyFuture); + } +} diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageContentSaver.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageContentSaver.java new file mode 100644 index 00000000000..682734a15e7 --- /dev/null +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageContentSaver.java @@ -0,0 +1,39 @@ +/**************************************************************** + * 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 org.apache.james.blob.api.BlobId; + +import com.google.common.io.ByteSource; + +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; + +/** + * Saves the content of a message: its headers and its body. + * + * Implementations decide which recovery policy, if any, is applied alongside the content write. + */ +public interface MessageContentSaver { + /** + * @return the blob id of the headers (T1) and the blob id of the body (T2). + */ + Mono> saveContent(byte[] headerBytes, ByteSource bodyByteSource); +} diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java index a13b5293fba..d460ac5b94b 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java @@ -41,6 +41,7 @@ import org.apache.james.backends.cassandra.StatementRecorder; import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; 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.blob.api.PlainBlobId; import org.apache.james.blob.cassandra.BlobTables; @@ -835,7 +836,8 @@ private CassandraMessageDAOV3 messageDAO(CassandraCluster cassandraCluster) { mock(BlobStore.class), mock(BlobStoreDAO.class), new PlainBlobId.Factory(), - CassandraConfiguration.DEFAULT_CONFIGURATION); + CassandraConfiguration.DEFAULT_CONFIGURATION, + BlobStoreCacheCallback.NOOP); } private CassandraThreadDAO threadDAO(CassandraCluster cassandraCluster) { diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java index 9a3067e1fc0..652a120bcc9 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java @@ -20,7 +20,6 @@ import static org.apache.james.mailbox.store.mail.model.MailboxMessage.EMPTY_SAVE_DATE; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.charset.StandardCharsets; import java.util.Collection; @@ -38,9 +37,9 @@ import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition; 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.blob.api.BucketName; -import org.apache.james.blob.api.ObjectNotFoundException; import org.apache.james.blob.api.PlainBlobId; import org.apache.james.blob.cassandra.CassandraBlobDataDefinition; import org.apache.james.blob.cassandra.CassandraBlobStoreDAO; @@ -136,7 +135,8 @@ private CassandraMessageDAOV3 buildTestee(CassandraCluster cassandra, CassandraC blobStore, blobStoreDAO, blobIdFactory, - configuration); + configuration, + BlobStoreCacheCallback.NOOP); } @Test @@ -172,47 +172,78 @@ void blobReferencesShouldBeEmptyByDefault() { } @Test - void saveShouldNotWriteRecoveryBlobByDefault() throws Exception { + void saveShouldNotCarryRecoveryInformationByDefault() throws Exception { message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); Tuple2 blobIds = testee.save(message).block(); - BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + blobIds.getT1().asString()); - assertThatThrownBy(() -> Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT, recoveryBlobId)).block()) - .isInstanceOf(ObjectNotFoundException.class); + assertThat(Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT, blobIds.getT1())).block() + .metadata().get(ContentRecoveryMessageContentSaver.BODY_BLOB_ID)) + .isEmpty(); } @Test - void saveShouldWriteRecoveryBlobWhenSynchronousMode(CassandraCluster cassandra) throws Exception { - CassandraConfiguration conf = CassandraConfiguration.builder() - .blobRecoveryMode(CassandraConfiguration.BlobRecoveryMode.SYNCHRONOUS) - .build(); - CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra, conf); + void saveShouldNotSuffixHeaderBlobIdByDefault() throws Exception { + message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); + + Tuple2 blobIds = testee.save(message).block(); + + assertThat(blobIds.getT1().asString()) + .doesNotEndWith(ContentRecoveryMessageContentSaver.HEADER_BLOB_ID_SUFFIX); + } + + @Test + void saveShouldCarryBodyBlobIdAsHeaderMetadataWhenRecoveryEnabled(CassandraCluster cassandra) throws Exception { + CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra, recoveryEnabled()); message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); Tuple2 blobIds = testeeWithRecovery.save(message).block(); - BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + blobIds.getT1().asString()); - byte[] recoveryContent = Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT, recoveryBlobId)).block().payload(); - assertThat(new String(recoveryContent, StandardCharsets.UTF_8)).isEqualTo(blobIds.getT2().asString()); + assertThat(Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT, blobIds.getT1())).block() + .metadata().get(ContentRecoveryMessageContentSaver.BODY_BLOB_ID)) + .contains(new BlobStoreDAO.BlobMetadataValue(blobIds.getT2().asString())); } @Test - void saveShouldWriteRecoveryBlobWhenAsynchronousMode(CassandraCluster cassandra) throws Exception { - CassandraConfiguration conf = CassandraConfiguration.builder() - .blobRecoveryMode(CassandraConfiguration.BlobRecoveryMode.ASYNCHRONOUS) - .build(); - CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra, conf); + void saveShouldSuffixHeaderBlobIdWhenRecoveryEnabled(CassandraCluster cassandra) throws Exception { + CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra, recoveryEnabled()); message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); Tuple2 blobIds = testeeWithRecovery.save(message).block(); - BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + blobIds.getT1().asString()); - // Asynchronous: give the parallel scheduler a moment to complete - assertThat(Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT, recoveryBlobId)) - .retryWhen(reactor.util.retry.Retry.fixedDelay(10, java.time.Duration.ofMillis(100))) - .block().payload()) - .isEqualTo(blobIds.getT2().asString().getBytes(StandardCharsets.UTF_8)); + assertThat(blobIds.getT1().asString()) + .endsWith(ContentRecoveryMessageContentSaver.HEADER_BLOB_ID_SUFFIX); + } + + @Test + void headerBlobIdShouldBeRandomWhenRecoveryEnabled(CassandraCluster cassandra) throws Exception { + CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra, recoveryEnabled()); + message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); + Tuple2 blobIds = testeeWithRecovery.save(message).block(); + + message = createMessage(messageId2, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); + Tuple2 otherBlobIds = testeeWithRecovery.save(message).block(); + + assertThat(blobIds.getT1()).isNotEqualTo(otherBlobIds.getT1()); + } + + @Test + void saveShouldStoreRetrievableMessageWhenRecoveryEnabled(CassandraCluster cassandra) throws Exception { + CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra, recoveryEnabled()); + message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); + + testeeWithRecovery.save(message).block(); + + MessageRepresentation representation = + toMessage(testeeWithRecovery.retrieveMessage(messageIdWithMetadata, MessageMapper.FetchType.FULL)); + assertThat(IOUtils.toString(representation.getContent().getInputStream(), StandardCharsets.UTF_8)) + .isEqualTo(CONTENT); + } + + private CassandraConfiguration recoveryEnabled() { + return CassandraConfiguration.builder() + .blobRecoveryMode(CassandraConfiguration.BlobRecoveryMode.ENABLED) + .build(); } @Test diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaverTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaverTest.java new file mode 100644 index 00000000000..463a918c470 --- /dev/null +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaverTest.java @@ -0,0 +1,123 @@ +/**************************************************************** + * 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 org.apache.james.mailbox.cassandra.mail.ContentRecoveryMessageContentSaver.HEADER_BLOB_ID_SUFFIX; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.nio.charset.StandardCharsets; +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.stream.Stream; + +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.blob.api.BucketName; +import org.apache.james.blob.api.PlainBlobId; +import org.apache.james.server.blob.deduplication.GenerationAwareBlobId; +import org.apache.james.server.blob.deduplication.MinIOGenerationAwareBlobId; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import com.google.common.io.ByteSource; + +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; + +/** + * The header blob id is not a free-form string: the garbage collection reads its generation back out of + * it, and the recovery runner filters on its {@value ContentRecoveryMessageContentSaver#HEADER_BLOB_ID_SUFFIX} + * suffix and on the family and generation prefix pushed down to the object store as a listing prefix. + * + *

Those three properties have to hold for every {@link BlobId.Factory} a deployment may be configured + * with, which is what this pins down.

+ */ +class ContentRecoveryMessageContentSaverTest { + private static final byte[] HEADER_BYTES = "Subject: test\r\n\r\n".getBytes(StandardCharsets.UTF_8); + private static final ByteSource BODY = ByteSource.wrap("body".getBytes(StandardCharsets.UTF_8)); + + /** + * 2026-09-04T00:00:00Z is exactly 690 times the default 30 days generation duration, which keeps the + * expected prefixes below readable rather than recomputed from the formula under test. + */ + private static final Clock CLOCK = Clock.fixed(Instant.parse("2026-09-04T00:00:00Z"), ZoneOffset.UTC); + + static Stream blobIdFactories() { + return Stream.of( + Arguments.of("PlainBlobId", new PlainBlobId.Factory(), ""), + Arguments.of("GenerationAwareBlobId", + new GenerationAwareBlobId.Factory(CLOCK, new PlainBlobId.Factory(), GenerationAwareBlobId.Configuration.DEFAULT), + "1_690_"), + Arguments.of("MinIOGenerationAwareBlobId", + new MinIOGenerationAwareBlobId.Factory(CLOCK, GenerationAwareBlobId.Configuration.DEFAULT, new PlainBlobId.Factory()), + "1/690/")); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("blobIdFactories") + void headerBlobIdShouldBeSuffixed(String name, BlobId.Factory blobIdFactory, String expectedPrefix) { + assertThat(saveContent(blobIdFactory).getT1().asString()) + .endsWith(HEADER_BLOB_ID_SUFFIX); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("blobIdFactories") + void headerBlobIdShouldCarryFamilyAndGenerationOfItsFactory(String name, BlobId.Factory blobIdFactory, String expectedPrefix) { + assertThat(saveContent(blobIdFactory).getT1().asString()) + .startsWith(expectedPrefix); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("blobIdFactories") + void headerBlobIdShouldRoundTripThroughItsFactory(String name, BlobId.Factory blobIdFactory, String expectedPrefix) { + String headerBlobId = saveContent(blobIdFactory).getT1().asString(); + + assertThat(blobIdFactory.parse(headerBlobId).asString()) + .isEqualTo(headerBlobId); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("blobIdFactories") + void headerBlobIdShouldNotRepeatItself(String name, BlobId.Factory blobIdFactory, String expectedPrefix) { + assertThat(saveContent(blobIdFactory).getT1()) + .isNotEqualTo(saveContent(blobIdFactory).getT1()); + } + + private Tuple2 saveContent(BlobId.Factory blobIdFactory) { + BlobStoreDAO blobStoreDAO = mock(BlobStoreDAO.class); + when(blobStoreDAO.save(any(BucketName.class), any(BlobId.class), any(BlobStoreDAO.Blob.class))).thenReturn(Mono.empty()); + + BlobStore blobStore = mock(BlobStore.class); + when(blobStore.getDefaultBucketName()).thenReturn(BucketName.DEFAULT); + when(blobStore.save(any(BucketName.class), any(ByteSource.class), any(BlobStore.StoragePolicy.class))) + .thenReturn(Mono.just(blobIdFactory.of("body"))); + + return new ContentRecoveryMessageContentSaver(blobStore, blobStoreDAO, blobIdFactory, BlobStoreCacheCallback.NOOP) + .saveContent(HEADER_BYTES, BODY) + .block(); + } +} diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java index 3688acb50ff..eda113d31d5 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java @@ -27,6 +27,7 @@ import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; 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.blob.api.BucketName; import org.apache.james.blob.api.PlainBlobId; @@ -97,6 +98,7 @@ public static Module commonModules(CqlSession session, CassandraTypesProvider ty binder -> binder.bind(ModSeqProvider.class).to(CassandraModSeqProvider.class), binder -> binder.bind(ACLMapper.class).to(CassandraACLMapper.class), binder -> binder.bind(BlobId.Factory.class).toInstance(new PlainBlobId.Factory()), + binder -> binder.bind(BlobStoreCacheCallback.class).toInstance(BlobStoreCacheCallback.NOOP), binder -> binder.bind(BlobStore.class).toProvider(() -> CassandraBlobStoreFactory.forTesting(session, new RecordingMetricFactory()).passthrough()), binder -> binder.bind(BlobStoreDAO.class).toProvider(() -> { PlainBlobId.Factory blobIdFactory = new PlainBlobId.Factory(); diff --git a/server/apps/distributed-app/README.adoc b/server/apps/distributed-app/README.adoc index 33717877980..aa0aa9a2141 100644 --- a/server/apps/distributed-app/README.adoc +++ b/server/apps/distributed-app/README.adoc @@ -109,17 +109,26 @@ Note that binding ports below 1024 requires administrative rights. == S3 blob store recovery When the S3 (or compatible) object store survives but the Cassandra mailbox structure is lost, messages -can be rebuilt from the blob store alone, provided `mailbox.blob.recovery.mode` was set to `synchronous` -or `asynchronous` in `cassandra.properties` while messages were being stored. In that mode James writes, -next to each stored message, a `recovery/` sidecar blob holding the matching `bodyBlobId`. +can be rebuilt from the blob store alone, provided `mailbox.blob.recovery.mode` was set to `enabled` in +`cassandra.properties` while messages were being stored. In that mode James stores the id of a message +body as metadata of its header blob, and suffixes header blob ids with `_hdr` so that they can be told +apart when walking the store. No extra object is written. -Recovery is packaged as an alternate main class, `org.apache.james.S3RecoveryMain`, that reuses the same -mailbox, DAO and blob store modules and the same `blobstore.properties` (so AES encryption and +=== Cost of enabling it + +`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 -- LMTP delivery, IMAP APPEND, JMAP import. + +Recovery itself is packaged as an alternate main class, `org.apache.james.S3RecoveryMain`, that reuses the +same mailbox, DAO and blob store modules and the same `blobstore.properties` (so AES encryption and compression are applied transparently). It starts neither the protocol servers nor RabbitMQ. -It walks the blob 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. +It walks the header blobs of the blob store, reads the body blob id from their metadata, rebuilds each +message from its header and body, 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 entrypoint main class (Cassandra and S3 must be reachable, RabbitMQ is not needed): @@ -152,17 +161,24 @@ environment variable, or the `restore.messages.after` system property: $ java ... org.apache.james.S3RecoveryMain --restore-after=2026-01-01T00:00:00Z ---- -An optional `--header-blob-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, and lets S3 filter server-side. -Because header blob ids are generation-aware (`family_generation_...`), a clever admin can pass e.g. -`1_42_` to iterate solely the latest generation instead of scanning the whole bucket: +The `--family=` and `--generation=` 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. They are pushed down +to the object store as a listing prefix, so the store filters server-side rather than James 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] ---- -$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ +$ java ... org.apache.james.S3RecoveryMain --family=1 --generation=42 ---- +Deployments configured with the MinIO blob id strategy separate the family and generation with `/` rather +than `_`. Say so with `--minio-separator` (or `RECOVERY_MINIO_SEPARATOR`, or `recovery.minio.separator`), +which turns the prefix above into `1/42/`. + The `--concurrency=` 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. The per-message work (blob reads plus a full re-store through the mailbox) dominates recovery @@ -179,15 +195,15 @@ twice, throughput scales with their number, and a failed shard can be re-run on [source] ---- -$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_40_ --concurrency=16 -$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_41_ --concurrency=16 -$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ --concurrency=16 +$ java ... org.apache.james.S3RecoveryMain --family=1 --generation=40 --concurrency=16 +$ java ... org.apache.james.S3RecoveryMain --family=1 --generation=41 --concurrency=16 +$ java ... org.apache.james.S3RecoveryMain --family=1 --generation=42 --concurrency=16 ---- Notes: -* Restored messages are re-stored (and get a fresh `recovery/` sidecar), so re-running the recovery -restores them again. Restore into an empty deployment, or clean up between runs. +* 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 module is not started during recovery, so restored messages are not indexed on the fly. Run a re-indexing afterwards if search is needed. diff --git a/server/apps/distributed-app/sample-configuration/jvm.properties b/server/apps/distributed-app/sample-configuration/jvm.properties index ce4087873b9..dabfc953ae4 100644 --- a/server/apps/distributed-app/sample-configuration/jvm.properties +++ b/server/apps/distributed-app/sample-configuration/jvm.properties @@ -101,6 +101,13 @@ jmx.remote.x.mlet.allow.getMBeansFromURL=false # messageFastViewProjection table. Only applies to previews computed afterwards. # james.jmap.preview.length=128 +# Bits of entropy carried by a blobId: the SHA-256 is truncated to that many leading bits, and randomly +# generated ids draw that many. A multiple of 8 within [96, 256], defaults to 128. +# 256 spells ids out the way releases up to 3.9.x did, taking a body blobId from 28 to 50 chars in the +# object key and in every Cassandra column referencing it, for no practical collision benefit: 128 bits +# already puts a collision at 1.5e-19 for ten billion blobs. +# james.blobid.entropy=256 + # Count of octet from which hashing shall be done out of the IO threads in deduplicating blob store # james.deduplicating.blobstore.thread.switch.threshold=32768 # Count of octet from which streams are buffered to files and not to memory diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java index 75ca00fb589..35cb07148ed 100644 --- a/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java +++ b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java @@ -34,13 +34,19 @@ * {@code --restore-after=} program argument, the {@code RESTORE_MESSAGES_AFTER} * environment variable, or the {@code restore.messages.after} system property.

* - *

The optional {@code headerBlobPrefix} narrows the walk to the recovery sidecars whose header blob - * id starts with the given prefix. Because header blob ids are generation-aware - * ({@code family_generation_...}), a clever admin can pass e.g. {@code 1_42_} to iterate solely the - * latest generation instead of scanning the whole bucket. It defaults to the empty string (all - * recovery sidecars) and can be provided as a {@code --header-blob-prefix=} program argument, - * the {@code RECOVERY_HEADER_BLOB_PREFIX} environment variable, or the {@code recovery.header.blob.prefix} - * system property.

+ *

The optional {@code family} and {@code generation} narrow the walk to a single generation of the + * generation aware blob ids, and are pushed down to the object store as a listing prefix rather than + * filtered client side. Recovering a large deployment is therefore best sharded by generation, one + * process each. Because the generation is the second component of a blob id, there is no prefix for a + * generation alone: {@code --generation} requires {@code --family}. They come from + * {@code --family=} / {@code --generation=}, the {@code RECOVERY_FAMILY} / + * {@code RECOVERY_GENERATION} environment variables, or the {@code recovery.family} / + * {@code recovery.generation} system properties, and default to walking the whole bucket.

+ * + *

The prefix separator depends on the configured blob id strategy: {@code _} for + * {@code GenerationAwareBlobId}, {@code /} for {@code MinIOGenerationAwareBlobId}. Deployments using the + * latter must say so with {@code --minio-separator}, the {@code RECOVERY_MINIO_SEPARATOR} environment + * variable, or the {@code recovery.minio.separator} system property.

* *

The {@code concurrency} controls how many messages are restored in parallel. Since the dominant * cost is the per-message work (blob reads plus a full re-store through the mailbox), this is the main @@ -48,29 +54,55 @@ * as a {@code --concurrency=} program argument, the {@code RECOVERY_CONCURRENCY} environment * variable, or the {@code recovery.concurrency} system property.

*/ -public record RecoveryConfiguration(Optional restoreAfter, String headerBlobPrefix, int concurrency) { +public record RecoveryConfiguration(Optional restoreAfter, Optional family, + Optional generation, boolean minioSeparator, int concurrency) { public static final int DEFAULT_CONCURRENCY = 8; + private static final String GENERATION_AWARE_SEPARATOR = "_"; + private static final String MINIO_SEPARATOR = "/"; private static final String RESTORE_AFTER_ARG = "--restore-after="; private static final String RESTORE_AFTER_ENV = "RESTORE_MESSAGES_AFTER"; private static final String RESTORE_AFTER_PROPERTY = "restore.messages.after"; - private static final String HEADER_BLOB_PREFIX_ARG = "--header-blob-prefix="; - private static final String HEADER_BLOB_PREFIX_ENV = "RECOVERY_HEADER_BLOB_PREFIX"; - private static final String HEADER_BLOB_PREFIX_PROPERTY = "recovery.header.blob.prefix"; + private static final String FAMILY_ARG = "--family="; + private static final String FAMILY_ENV = "RECOVERY_FAMILY"; + private static final String FAMILY_PROPERTY = "recovery.family"; + private static final String GENERATION_ARG = "--generation="; + private static final String GENERATION_ENV = "RECOVERY_GENERATION"; + private static final String GENERATION_PROPERTY = "recovery.generation"; + private static final String MINIO_SEPARATOR_ARG = "--minio-separator"; + private static final String MINIO_SEPARATOR_ENV = "RECOVERY_MINIO_SEPARATOR"; + private static final String MINIO_SEPARATOR_PROPERTY = "recovery.minio.separator"; private static final String CONCURRENCY_ARG = "--concurrency="; private static final String CONCURRENCY_ENV = "RECOVERY_CONCURRENCY"; private static final String CONCURRENCY_PROPERTY = "recovery.concurrency"; public RecoveryConfiguration { Preconditions.checkArgument(concurrency > 0, "'concurrency' must be strictly positive"); + Preconditions.checkArgument(generation.isEmpty() || family.isPresent(), + "'" + GENERATION_ARG + "' requires '" + FAMILY_ARG + "': the generation is the second component of a blob id, " + + "there is no listing prefix for a generation on its own"); } public static RecoveryConfiguration parse(String[] args) { return new RecoveryConfiguration( option(args, RESTORE_AFTER_ARG, RESTORE_AFTER_ENV, RESTORE_AFTER_PROPERTY).map(RecoveryConfiguration::parseInstant), - option(args, HEADER_BLOB_PREFIX_ARG, HEADER_BLOB_PREFIX_ENV, HEADER_BLOB_PREFIX_PROPERTY).orElse(""), + option(args, FAMILY_ARG, FAMILY_ENV, FAMILY_PROPERTY).map(RecoveryConfiguration::parseFamily), + option(args, GENERATION_ARG, GENERATION_ENV, GENERATION_PROPERTY).map(RecoveryConfiguration::parseGeneration), + flag(args, MINIO_SEPARATOR_ARG, MINIO_SEPARATOR_ENV, MINIO_SEPARATOR_PROPERTY), option(args, CONCURRENCY_ARG, CONCURRENCY_ENV, CONCURRENCY_PROPERTY).map(RecoveryConfiguration::parseConcurrency).orElse(DEFAULT_CONCURRENCY)); } + /** + * The listing prefix restricting the walk to the requested family and generation, empty when the + * whole bucket is to be walked. + */ + public String headerBlobPrefix() { + String separator = minioSeparator ? MINIO_SEPARATOR : GENERATION_AWARE_SEPARATOR; + return family + .map(familyValue -> familyValue + separator + + generation.map(generationValue -> generationValue + separator).orElse("")) + .orElse(""); + } + private static Optional option(String[] args, String argPrefix, String envName, String propertyName) { return Arrays.stream(args) .filter(arg -> arg.startsWith(argPrefix)) @@ -82,6 +114,15 @@ private static Optional option(String[] args, String argPrefix, String e .filter(value -> !value.isEmpty()); } + private static boolean flag(String[] args, String argName, String envName, String propertyName) { + return Arrays.asList(args).contains(argName) + || Optional.ofNullable(System.getenv(envName)) + .or(() -> Optional.ofNullable(System.getProperty(propertyName))) + .map(String::trim) + .map(Boolean::parseBoolean) + .orElse(false); + } + private static Instant parseInstant(String value) { try { return Instant.parse(value); @@ -91,6 +132,28 @@ private static Instant parseInstant(String value) { } } + private static int parseFamily(String value) { + try { + int family = Integer.parseInt(value); + Preconditions.checkArgument(family > 0); + return family; + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Invalid '" + FAMILY_ARG + "' value: '" + value + + "'. Expected a strictly positive integer", e); + } + } + + private static long parseGeneration(String value) { + try { + long generation = Long.parseLong(value); + Preconditions.checkArgument(generation >= 0); + return generation; + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Invalid '" + GENERATION_ARG + "' value: '" + value + + "'. Expected a non negative integer", e); + } + } + private static int parseConcurrency(String value) { try { int concurrency = Integer.parseInt(value); diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java index f693149fcac..96b1c042742 100644 --- a/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java +++ b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java @@ -20,11 +20,10 @@ package org.apache.james; import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST; -import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED; -import static org.apache.james.blob.api.BlobStoreDAO.RECOVERY_BLOB_PREFIX; +import static org.apache.james.mailbox.cassandra.mail.ContentRecoveryMessageContentSaver.BODY_BLOB_ID; +import static org.apache.james.mailbox.cassandra.mail.ContentRecoveryMessageContentSaver.HEADER_BLOB_ID_SUFFIX; import java.io.ByteArrayInputStream; -import java.nio.charset.StandardCharsets; import java.util.Date; import java.util.List; import java.util.Optional; @@ -61,17 +60,22 @@ import reactor.core.scheduler.Schedulers; /** - * Walks the blob store looking for {@code recovery/} sidecars (written by - * {@code CassandraMessageDAOV3} when {@code mailbox.blob.recovery.mode} is enabled) and restores the - * associated messages into a {@code Restored-messages} mailbox of each local {@code Delivered-To} recipient. + * Walks the header blobs of the blob store (written by {@code CassandraMessageDAOV3} when + * {@code mailbox.blob.recovery.mode} is enabled) and restores the associated messages into a + * {@code Restored-messages} mailbox of each local {@code Delivered-To} recipient. * - *

Reads go through the configured {@link BlobStore}, so AES decryption and decompression are applied + *

Header blobs are told apart by their {@code _hdr} suffix and carry the id of their body blob as + * metadata. A blob matching the suffix without that metadata is simply skipped, so an unlucky collision + * costs nothing.

+ * + *

Reads go through the decorated {@link BlobStoreDAO}, so AES decryption and decompression are applied * transparently, exactly as the write path did.

*/ public class S3RecoveryService { - public record Report(long processed, long restored, long skippedByDate, long skippedNoLocalUser, long failed) { + public record Report(long processed, long restored, long skippedByDate, long skippedNoLocalUser, + long skippedNoRecoveryInfo, long failed) { public static Report empty() { - return new Report(0, 0, 0, 0, 0); + return new Report(0, 0, 0, 0, 0, 0); } Report merge(Report other) { @@ -79,6 +83,7 @@ Report merge(Report other) { restored + other.restored, skippedByDate + other.skippedByDate, skippedNoLocalUser + other.skippedNoLocalUser, + skippedNoRecoveryInfo + other.skippedNoRecoveryInfo, failed + other.failed); } } @@ -89,10 +94,11 @@ private record RecoveredMessage(List recipients, Optional dat private static final Logger LOGGER = LoggerFactory.getLogger(S3RecoveryService.class); private static final String RESTORE_MAILBOX = "Restored-messages"; private static final String DELIVERED_TO = "Delivered-To"; - private static final Report RESTORED = new Report(1, 1, 0, 0, 0); - private static final Report SKIPPED_BY_DATE = new Report(1, 0, 1, 0, 0); - private static final Report SKIPPED_NO_LOCAL_USER = new Report(1, 0, 0, 1, 0); - private static final Report FAILED = new Report(1, 0, 0, 0, 1); + private static final Report RESTORED = new Report(1, 1, 0, 0, 0, 0); + private static final Report SKIPPED_BY_DATE = new Report(1, 0, 1, 0, 0, 0); + private static final Report SKIPPED_NO_LOCAL_USER = new Report(1, 0, 0, 1, 0, 0); + private static final Report SKIPPED_NO_RECOVERY_INFO = new Report(1, 0, 0, 0, 1, 0); + private static final Report FAILED = new Report(1, 0, 0, 0, 0, 1); private final BlobStore blobStore; private final BlobStoreDAO blobStoreDAO; @@ -116,39 +122,43 @@ public S3RecoveryService(BlobStore blobStore, BlobStoreDAO blobStoreDAO, BlobId. public Mono run() { BucketName bucket = blobStore.getDefaultBucketName(); - String prefix = RECOVERY_BLOB_PREFIX + configuration.headerBlobPrefix(); + String prefix = configuration.headerBlobPrefix(); LOGGER.info("Starting S3 recovery on bucket {} (prefix: {}, restore after: {}, concurrency: {})", bucket.asString(), prefix, configuration.restoreAfter(), configuration.concurrency()); return Flux.from(blobStoreDAO.listBlobs(bucket, prefix)) - .map(BlobId::asString) - .flatMap(recoveryKey -> restoreOne(bucket, recoveryKey), configuration.concurrency()) + .filter(blobId -> blobId.asString().endsWith(HEADER_BLOB_ID_SUFFIX)) + .flatMap(headerBlobId -> restoreOne(bucket, headerBlobId), configuration.concurrency()) .reduce(Report.empty(), Report::merge) .doOnNext(report -> LOGGER.info("S3 recovery finished: {}", report)); } - private Mono restoreOne(BucketName bucket, String recoveryKey) { - String headerKey = recoveryKey.substring(RECOVERY_BLOB_PREFIX.length()); - BlobId headerBlobId = blobIdFactory.parse(headerKey); - BlobId recoveryBlobId = blobIdFactory.parse(recoveryKey); - - return Mono.from(blobStoreDAO.readBytes(bucket, recoveryBlobId)) - .map(sidecar -> blobIdFactory.parse(new String(sidecar.payload(), StandardCharsets.UTF_8).trim())) - .flatMap(bodyBlobId -> recover(bucket, headerBlobId, bodyBlobId)) - .flatMap(this::restore) + private Mono restoreOne(BucketName bucket, BlobId headerBlobId) { + return Mono.from(blobStoreDAO.readBytes(bucket, headerBlobId)) + .flatMap(headerBlob -> bodyBlobId(headerBlob) + .map(bodyBlobId -> recover(bucket, headerBlob.payload(), bodyBlobId) + .flatMap(this::restore)) + .orElseGet(() -> { + LOGGER.debug("Skipping {}: no recovery information", headerBlobId.asString()); + return Mono.just(SKIPPED_NO_RECOVERY_INFO); + })) .onErrorResume(error -> { - LOGGER.error("Failed to recover message from {}", recoveryKey, error); + LOGGER.error("Failed to recover message from {}", headerBlobId.asString(), error); return Mono.just(FAILED); }); } - private Mono recover(BucketName bucket, BlobId headerBlobId, BlobId bodyBlobId) { - return Mono.from(blobStore.readBytes(bucket, headerBlobId, SIZE_BASED)) - .flatMap(headerBytes -> { - MessageHeaders headers = parseHeaders(headerBytes); - return Mono.from(blobStore.readBytes(bucket, bodyBlobId, LOW_COST)) - .map(bodyBytes -> new RecoveredMessage(headers.recipients(), headers.date(), - new HeaderAndBodyByteContent(headerBytes, bodyBytes))); - }); + private Optional bodyBlobId(BlobStoreDAO.BytesBlob headerBlob) { + return headerBlob.metadata() + .get(BODY_BLOB_ID) + .map(BlobStoreDAO.BlobMetadataValue::value) + .map(blobIdFactory::parse); + } + + private Mono recover(BucketName bucket, byte[] headerBytes, BlobId bodyBlobId) { + MessageHeaders headers = parseHeaders(headerBytes); + return Mono.from(blobStore.readBytes(bucket, bodyBlobId, LOW_COST)) + .map(bodyBytes -> new RecoveredMessage(headers.recipients(), headers.date(), + new HeaderAndBodyByteContent(headerBytes, bodyBytes))); } private Mono restore(RecoveredMessage message) { diff --git a/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java b/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java index e2205b8e3a5..bc587b70b12 100644 --- a/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java +++ b/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java @@ -45,14 +45,50 @@ void parseShouldRejectInvalidInstant() { } @Test - void parseShouldDefaultHeaderBlobPrefixToEmpty() { + void headerBlobPrefixShouldBeEmptyWhenNoFamily() { assertThat(RecoveryConfiguration.parse(new String[] {}).headerBlobPrefix()).isEmpty(); } @Test - void parseShouldReadHeaderBlobPrefixArgument() { - assertThat(RecoveryConfiguration.parse(new String[] {"--header-blob-prefix=1_42_"}).headerBlobPrefix()) - .isEqualTo("1_42_"); + void headerBlobPrefixShouldOnlyCarryFamilyWhenNoGeneration() { + assertThat(RecoveryConfiguration.parse(new String[] {"--family=1"}).headerBlobPrefix()) + .isEqualTo("1_"); + } + + @Test + void headerBlobPrefixShouldCarryFamilyAndGeneration() { + assertThat(RecoveryConfiguration.parse(new String[] {"--family=1", "--generation=690"}).headerBlobPrefix()) + .isEqualTo("1_690_"); + } + + @Test + void headerBlobPrefixShouldUseSlashWhenMinioSeparator() { + assertThat(RecoveryConfiguration.parse(new String[] {"--family=1", "--generation=690", "--minio-separator"}).headerBlobPrefix()) + .isEqualTo("1/690/"); + } + + @Test + void parseShouldRejectGenerationWithoutFamily() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--generation=690"})) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectNonNumericFamily() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--family=one"})) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectNonPositiveFamily() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--family=0"})) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectNonNumericGeneration() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--family=1", "--generation=latest"})) + .isInstanceOf(IllegalArgumentException.class); } @Test diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobId.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobId.java index 815907914f0..93917db2c21 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobId.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobId.java @@ -25,7 +25,19 @@ interface Factory { BlobId of(String id); BlobId parse(String id); + + BlobIdEncoding encoding(); + + default BlobId random() { + return of(encoding().encode(BlobIdEntropy.randomBytes())); + } + + default BlobId ofHash(byte[] hash) { + return of(encoding().encode(BlobIdEntropy.truncate(hash))); + } } String asString(); + + BlobId withSuffix(String suffix); } diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobIdEncoding.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobIdEncoding.java new file mode 100644 index 00000000000..655c2d365a0 --- /dev/null +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobIdEncoding.java @@ -0,0 +1,71 @@ +/**************************************************************** + * 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 java.util.Optional; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.io.BaseEncoding; + +/** + * How the payload of a blob id is spelled out, as set by the {@code james.blob.id.hash.encoding} system + * property. + * + *

Truncated ids are left unpadded: they exist to be short, and padding them back up would give away + * part of what {@link BlobIdEntropy} saved. Ids at full entropy keep the padding of their encoding, so + * that a deployment pinned there keeps spelling ids the way releases up to 3.9.x did.

+ */ +public class BlobIdEncoding { + public static final String ENCODING_PROPERTY = "james.blob.id.hash.encoding"; + private static final BaseEncoding DEFAULT_ENCODING = BaseEncoding.base64Url(); + + public static BlobIdEncoding fromSystemProperties() { + return new BlobIdEncoding(Optional.ofNullable(System.getProperty(ENCODING_PROPERTY)) + .map(BlobIdEncoding::baseEncodingFrom) + .orElse(DEFAULT_ENCODING)); + } + + @VisibleForTesting + static BaseEncoding baseEncodingFrom(String encodingType) { + return switch (encodingType) { + case "base16", "hex" -> BaseEncoding.base16(); + case "base32" -> BaseEncoding.base32(); + case "base32Hex" -> BaseEncoding.base32Hex(); + case "base64" -> BaseEncoding.base64(); + case "base64Url" -> BaseEncoding.base64Url(); + default -> throw new IllegalArgumentException("Unknown encoding type: " + encodingType); + }; + } + + private final BaseEncoding encoding; + + @VisibleForTesting + BlobIdEncoding(BaseEncoding encoding) { + if (BlobIdEntropy.entropyBits() == BlobIdEntropy.MAX_ENTROPY_BITS) { + this.encoding = encoding; + } else { + this.encoding = encoding.omitPadding(); + } + } + + public String encode(byte[] payload) { + return encoding.encode(payload); + } +} diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobIdEntropy.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobIdEntropy.java new file mode 100644 index 00000000000..d88c4b96d3c --- /dev/null +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobIdEntropy.java @@ -0,0 +1,101 @@ +/**************************************************************** + * 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 java.security.SecureRandom; +import java.util.Arrays; +import java.util.Optional; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; + +/** + * How many bits of entropy a blob id carries, as set by the {@code james.blobid.entropy} system property. + * + *

Defaults to {@value #DEFAULT_ENTROPY_BITS} bits: the birthday bound puts a collision at + * {@code n^2/2^129}, ie. 1.5e-19 for ten billion blobs, and truncating a cryptographic hash to its + * leading bits is standard practice (NIST SP 800-107, FIPS 180-4). {@value #MAX_ENTROPY_BITS}, the full + * SHA-256 output, spells ids out the way releases up to 3.9.x did.

+ */ +public class BlobIdEntropy { + public static final String ENTROPY_BITS_PROPERTY = "james.blobid.entropy"; + public static final int DEFAULT_ENTROPY_BITS = 128; + @VisibleForTesting + static final int MIN_ENTROPY_BITS = 96; + /** The full SHA-256 output: the longest an id can usefully get. */ + static final int MAX_ENTROPY_BITS = 256; + private static final int BITS_PER_BYTE = 8; + + private static final SecureRandom SECURE_RANDOM = new SecureRandom(); + private static final int ENTROPY_BITS = parse(System.getProperty(ENTROPY_BITS_PROPERTY)); + + @VisibleForTesting + static int parse(String value) { + return Optional.ofNullable(value) + .map(String::trim) + .filter(trimmed -> !trimmed.isEmpty()) + .map(BlobIdEntropy::parseBits) + .orElse(DEFAULT_ENTROPY_BITS); + } + + private static int parseBits(String value) { + try { + int bits = Integer.parseInt(value); + Preconditions.checkArgument(bits % BITS_PER_BYTE == 0, + "'%s' must be a multiple of %s, got %s", ENTROPY_BITS_PROPERTY, BITS_PER_BYTE, bits); + Preconditions.checkArgument(bits >= MIN_ENTROPY_BITS && bits <= MAX_ENTROPY_BITS, + "'%s' must be within [%s, %s], got %s", ENTROPY_BITS_PROPERTY, MIN_ENTROPY_BITS, MAX_ENTROPY_BITS, bits); + return bits; + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid '" + ENTROPY_BITS_PROPERTY + "' value: '" + value + "'. Expected a bit count, eg. 128 or 256", e); + } + } + + public static int entropyBits() { + return ENTROPY_BITS; + } + + public static int entropyBytes() { + return ENTROPY_BITS / BITS_PER_BYTE; + } + + public static byte[] randomBytes() { + return randomBytes(entropyBytes()); + } + + @VisibleForTesting + static byte[] randomBytes(int entropyBytes) { + byte[] bytes = new byte[entropyBytes]; + SECURE_RANDOM.nextBytes(bytes); + return bytes; + } + + public static byte[] truncate(byte[] hash) { + return truncate(hash, entropyBytes()); + } + + @VisibleForTesting + static byte[] truncate(byte[] hash, int entropyBytes) { + if (hash.length <= entropyBytes) { + return hash; + } + return Arrays.copyOf(hash, entropyBytes); + } +} diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreCacheCallback.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreCacheCallback.java new file mode 100644 index 00000000000..01bd51ea5ac --- /dev/null +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreCacheCallback.java @@ -0,0 +1,43 @@ +/**************************************************************** + * 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 org.reactivestreams.Publisher; + +import reactor.core.publisher.Mono; + +/** + * Populates the blob store cache for a blob that was written through {@link BlobStoreDAO} rather than + * through {@link BlobStore}. + * + *

Callers needing the metadata of a blob have to go through {@link BlobStoreDAO}, which sits below the + * caching decorator and thus knows nothing of {@link BlobStore.StoragePolicy}. This callback gives them + * back the caching that a {@code SIZE_BASED} save would have performed.

+ * + *

The caller vouches for the blob being worth caching: stored in the default bucket, and semantically + * what a non-{@code LOW_COST} storage policy expresses. Implementations remain free to decline, typically + * on payload size.

+ */ +@FunctionalInterface +public interface BlobStoreCacheCallback { + BlobStoreCacheCallback NOOP = (blobId, bytes) -> Mono.empty(); + + Publisher cacheIfNeeded(BlobId blobId, byte[] bytes); +} 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 223dafe3d8a..46af6ebed14 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 @@ -54,8 +54,6 @@ *

See {@code docs/modules/servers/partials/architecture/blobstore.adoc} for more details.

*/ public interface BlobStoreDAO { - String RECOVERY_BLOB_PREFIX = "recovery/"; - record BlobMetadataName(String name) { private static final CharMatcher CHAR_MATCHER = CharMatcher.inRange('a', 'z') .or(CharMatcher.inRange('A', 'Z')) @@ -295,7 +293,8 @@ public ByteSourceBlob asByteSource() { Publisher listBlobs(BucketName bucketName); /** - * Lists the blobs of a bucket whose id starts with the given prefix (eg. {@link #RECOVERY_BLOB_PREFIX}). + * Lists the blobs of a bucket whose id starts with the given prefix (eg. {@code 1_690_} to restrict the + * listing to a single generation of a generation aware blob id). * *

The default implementation filters the full listing. Connectors able to push the prefix down to * their backend (eg. S3 {@code ListObjectsV2}) should override this for efficiency.

diff --git a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/PlainBlobId.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/PlainBlobId.java index 0022827252b..d285d9fb68a 100644 --- a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/PlainBlobId.java +++ b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/PlainBlobId.java @@ -24,6 +24,15 @@ public record PlainBlobId(String id) implements BlobId { public static class Factory implements BlobId.Factory { + private final BlobIdEncoding encoding; + + public Factory() { + this(BlobIdEncoding.fromSystemProperties()); + } + + public Factory(BlobIdEncoding encoding) { + this.encoding = encoding; + } @Override public PlainBlobId of(String id) { @@ -35,10 +44,20 @@ public PlainBlobId of(String id) { public PlainBlobId parse(String id) { return of(id); } + + @Override + public BlobIdEncoding encoding() { + return encoding; + } } @Override public String asString() { return id; } + + @Override + public PlainBlobId withSuffix(String suffix) { + return new PlainBlobId(id + suffix); + } } diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobIdEncodingTest.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobIdEncodingTest.java new file mode 100644 index 00000000000..0c948938e3f --- /dev/null +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobIdEncodingTest.java @@ -0,0 +1,88 @@ +/**************************************************************** + * 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.BlobIdEncoding.ENCODING_PROPERTY; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.nio.charset.StandardCharsets; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import com.google.common.io.BaseEncoding; + +class BlobIdEncodingTest { + private static final byte[] PAYLOAD = "payload".getBytes(StandardCharsets.UTF_8); + + @BeforeEach + @AfterEach + void clearProperty() { + System.clearProperty(ENCODING_PROPERTY); + } + + @Test + void blobIdFactoryCreationShouldFailOnInvalidProperty() { + System.setProperty(ENCODING_PROPERTY, "blobIdFactoryCreationShouldFailOnInvalidProperty"); + + assertThatThrownBy(PlainBlobId.Factory::new) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Unknown encoding type: blobIdFactoryCreationShouldFailOnInvalidProperty"); + } + + @ParameterizedTest + @ValueSource(strings = {"base16", "hex", "base32", "base32Hex", "base64", "base64Url"}) + void blobIdFactoryCreationShouldAcceptSupportedEncodings(String encoding) { + System.setProperty(ENCODING_PROPERTY, encoding); + + assertThatCode(PlainBlobId.Factory::new).doesNotThrowAnyException(); + } + + @Test + void shouldDefaultToBase64Url() { + assertThat(BlobIdEncoding.fromSystemProperties().encode(PAYLOAD)) + .isEqualTo(BaseEncoding.base64Url().omitPadding().encode(PAYLOAD)); + } + + @Test + void shouldOmitPaddingAtTruncatedEntropy() { + assertThat(BlobIdEncoding.fromSystemProperties().encode(PAYLOAD)) + .doesNotEndWith("="); + } + + @Test + void shouldHonourTheConfiguredEncoding() { + System.setProperty(ENCODING_PROPERTY, "base16"); + + assertThat(BlobIdEncoding.fromSystemProperties().encode(PAYLOAD)) + .isEqualTo(BaseEncoding.base16().encode(PAYLOAD)); + } + + @Test + void hexShouldBeAnAliasOfBase16() { + assertThat(BlobIdEncoding.baseEncodingFrom("hex")) + .isEqualTo(BlobIdEncoding.baseEncodingFrom("base16")); + } +} diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobIdEntropyTest.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobIdEntropyTest.java new file mode 100644 index 00000000000..febe3605a26 --- /dev/null +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/BlobIdEntropyTest.java @@ -0,0 +1,147 @@ +/**************************************************************** + * 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.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import com.google.common.io.BaseEncoding; + +class BlobIdEntropyTest { + @Test + void parseShouldReturnDefaultWhenNotSet() { + assertThat(BlobIdEntropy.parse(null)).isEqualTo(BlobIdEntropy.DEFAULT_ENTROPY_BITS); + } + + @Test + void parseShouldReturnDefaultWhenBlank() { + assertThat(BlobIdEntropy.parse(" ")).isEqualTo(BlobIdEntropy.DEFAULT_ENTROPY_BITS); + } + + @Test + void parseShouldAcceptTruncatedValue() { + assertThat(BlobIdEntropy.parse("128")).isEqualTo(128); + } + + @Test + void parseShouldAcceptTheFullHashLength() { + assertThat(BlobIdEntropy.parse("256")).isEqualTo(BlobIdEntropy.MAX_ENTROPY_BITS); + } + + @Test + void defaultShouldBeTruncated() { + assertThat(BlobIdEntropy.DEFAULT_ENTROPY_BITS).isEqualTo(128); + } + + @Test + void parseShouldRejectNonNumericValue() { + assertThatThrownBy(() -> BlobIdEntropy.parse("many")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectValueThatIsNotAByteCount() { + assertThatThrownBy(() -> BlobIdEntropy.parse("130")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectValueBelowTheSafetyFloor() { + assertThatThrownBy(() -> BlobIdEntropy.parse("88")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectValueAboveTheHashLength() { + assertThatThrownBy(() -> BlobIdEntropy.parse("512")) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void randomBytesShouldHonourEntropyLength() { + assertThat(BlobIdEntropy.randomBytes()).hasSize(BlobIdEntropy.entropyBytes()); + } + + @Test + void randomBytesShouldNotRepeatItself() { + assertThat(BlobIdEntropy.randomBytes()).isNotEqualTo(BlobIdEntropy.randomBytes()); + } + + @Test + void truncateShouldKeepLeadingBytes() { + byte[] hash = new byte[BlobIdEntropy.entropyBytes() + 4]; + for (int i = 0; i < hash.length; i++) { + hash[i] = (byte) i; + } + + byte[] truncated = BlobIdEntropy.truncate(hash); + + assertThat(truncated).hasSize(BlobIdEntropy.entropyBytes()); + for (int i = 0; i < truncated.length; i++) { + assertThat(truncated[i]).isEqualTo((byte) i); + } + } + + @Test + void truncateShouldLeaveShorterHashesUntouched() { + byte[] hash = new byte[] {1, 2, 3}; + + assertThat(BlobIdEntropy.truncate(hash)).isEqualTo(hash); + } + + @Nested + class NinetySixBits { + static final int ENTROPY_BYTES = 96 / 8; + // How the factory spells a truncated id: base64url, unpadded. + static final BaseEncoding ENCODING = BaseEncoding.base64Url().omitPadding(); + + @Test + void parseShouldAcceptTheSafetyFloor() { + assertThat(BlobIdEntropy.parse("96")).isEqualTo(BlobIdEntropy.MIN_ENTROPY_BITS); + } + + @Test + void randomBytesShouldDrawTwelveBytes() { + assertThat(BlobIdEntropy.randomBytes(ENTROPY_BYTES)).hasSize(ENTROPY_BYTES); + } + + @Test + void truncateShouldKeepTheTwelveLeadingBytesOfASha256() { + byte[] hash = new byte[32]; + for (int i = 0; i < hash.length; i++) { + hash[i] = (byte) i; + } + + assertThat(BlobIdEntropy.truncate(hash, ENTROPY_BYTES)) + .isEqualTo(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11}); + } + + @Test + void idsShouldSpellOutSixteenCharacters() { + byte[] hash = new byte[32]; + + assertThat(ENCODING.encode(BlobIdEntropy.truncate(hash, ENTROPY_BYTES))) + .hasSize(16); + } + } +} diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeduplicationBlobStoreContract.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeduplicationBlobStoreContract.java index cd4fd2ee588..69fa24586d1 100644 --- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeduplicationBlobStoreContract.java +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/DeduplicationBlobStoreContract.java @@ -24,18 +24,16 @@ import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED; import static org.apache.james.blob.api.BlobStoreContract.SHORT_BYTEARRAY; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.io.ByteArrayInputStream; import java.util.stream.Stream; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import com.google.common.io.BaseEncoding; + import reactor.core.publisher.Mono; public interface DeduplicationBlobStoreContract { @@ -47,6 +45,10 @@ static Stream storagePolicies() { } String SHORT_STRING = "toto"; + /** The SHA-256 of {@link #SHORT_STRING}, which every content addressed store spells its id from. */ + byte[] SHORT_STRING_HASH = BaseEncoding.base64Url().decode("MfemXjFVhqwZi9eYtmKc5JA9CJlHbVdBqfMuLlIbamY="); + /** That hash, truncated and spelled at the entropy the tests run with: "MfemXjFVhqwZi9eYtmKc5A" at the default 128 bits. */ + String SHORT_STRING_BLOB_ID = new BlobIdEncoding(BaseEncoding.base64Url()).encode(BlobIdEntropy.truncate(SHORT_STRING_HASH)); BlobStore testee(); @@ -54,25 +56,6 @@ static Stream storagePolicies() { BlobStore createBlobStore(); - @BeforeEach - default void beforeEach() { - System.clearProperty("james.blob.id.hash.encoding"); - } - - @AfterEach - default void afterEach() { - System.clearProperty("james.blob.id.hash.encoding"); - } - - @Test - default void deduplicationBlobstoreCreationShouldFailOnInvalidProperty() { - System.setProperty("james.blob.id.hash.encoding", "deduplicationBlobstoreCreationShouldFailOnInvalidProperty"); - - assertThatThrownBy(this::createBlobStore) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Unknown encoding type: deduplicationBlobstoreCreationShouldFailOnInvalidProperty"); - } - @ParameterizedTest @MethodSource("storagePolicies") default void saveShouldReturnBlobIdOfString(BlobStore.StoragePolicy storagePolicy) { @@ -81,7 +64,7 @@ default void saveShouldReturnBlobIdOfString(BlobStore.StoragePolicy storagePolic BlobId blobId = Mono.from(store.save(defaultBucketName, SHORT_STRING, storagePolicy)).block(); - assertThat(blobId).isEqualTo(blobIdFactory().parse("MfemXjFVhqwZi9eYtmKc5JA9CJlHbVdBqfMuLlIbamY=")); + assertThat(blobId).isEqualTo(blobIdFactory().parse(SHORT_STRING_BLOB_ID)); } @ParameterizedTest @@ -92,7 +75,7 @@ default void saveShouldReturnBlobId(BlobStore.StoragePolicy storagePolicy) { BlobId blobId = Mono.from(store.save(defaultBucketName, SHORT_BYTEARRAY, storagePolicy)).block(); - assertThat(blobId).isEqualTo(blobIdFactory().parse("MfemXjFVhqwZi9eYtmKc5JA9CJlHbVdBqfMuLlIbamY=")); + assertThat(blobId).isEqualTo(blobIdFactory().parse(SHORT_STRING_BLOB_ID)); } @ParameterizedTest @@ -105,6 +88,6 @@ default void saveShouldReturnBlobIdOfInputStream(BlobStore.StoragePolicy storage // This fix is ok because it will only affect deduplication, after this change the same content might be assigned a different blobid // and thus might be duplicated in the store. No data can be lost since no api allows for externally deterministic blob id construction // before this change. - assertThat(blobId).isEqualTo(blobIdFactory().of("MfemXjFVhqwZi9eYtmKc5JA9CJlHbVdBqfMuLlIbamY=")); + assertThat(blobId).isEqualTo(blobIdFactory().of(SHORT_STRING_BLOB_ID)); } } diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/PlainBlobIdTest.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/PlainBlobIdTest.java index da991b0d287..0a33b09dcee 100644 --- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/PlainBlobIdTest.java +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/PlainBlobIdTest.java @@ -22,11 +22,18 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Arrays; + import org.junit.jupiter.api.Test; +import com.google.common.io.BaseEncoding; + import nl.jqno.equalsverifier.EqualsVerifier; class PlainBlobIdTest { + // Mirrors how the factory spells ids at the default, truncated, entropy: base64url, unpadded. + private static final BaseEncoding ENCODING = BaseEncoding.base64Url().omitPadding(); + private static final PlainBlobId.Factory BLOB_ID_FACTORY = new PlainBlobId.Factory(); @@ -53,4 +60,43 @@ void fromShouldThrowOnEmpty() { assertThatThrownBy(() -> BLOB_ID_FACTORY.parse("")) .isInstanceOf(IllegalArgumentException.class); } + + @Test + void randomShouldDrawTheConfiguredEntropy() { + assertThat(ENCODING.decode(BLOB_ID_FACTORY.random().asString())) + .hasSize(BlobIdEntropy.entropyBytes()); + } + + @Test + void randomShouldNotRepeatItself() { + assertThat(BLOB_ID_FACTORY.random()) + .isNotEqualTo(BLOB_ID_FACTORY.random()); + } + + @Test + void ofHashShouldTruncateToTheConfiguredEntropy() { + byte[] hash = new byte[BlobIdEntropy.entropyBytes() + 8]; + + assertThat(ENCODING.decode(BLOB_ID_FACTORY.ofHash(hash).asString())) + .hasSize(BlobIdEntropy.entropyBytes()); + } + + @Test + void ofHashShouldKeepTheLeadingBytesOfTheHash() { + byte[] hash = new byte[BlobIdEntropy.entropyBytes() + 8]; + for (int i = 0; i < hash.length; i++) { + hash[i] = (byte) i; + } + + assertThat(ENCODING.decode(BLOB_ID_FACTORY.ofHash(hash).asString())) + .isEqualTo(Arrays.copyOf(hash, BlobIdEntropy.entropyBytes())); + } + + @Test + void ofHashShouldBeContentAddressed() { + byte[] hash = new byte[] {1, 2, 3}; + + assertThat(BLOB_ID_FACTORY.ofHash(hash)) + .isEqualTo(BLOB_ID_FACTORY.ofHash(hash)); + } } diff --git a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/TestBlobId.java b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/TestBlobId.java index 26e398eb385..59b7a7d1bec 100644 --- a/server/blob/blob-api/src/test/java/org/apache/james/blob/api/TestBlobId.java +++ b/server/blob/blob-api/src/test/java/org/apache/james/blob/api/TestBlobId.java @@ -22,6 +22,8 @@ import java.util.Objects; public class TestBlobId implements BlobId { + private static final BlobIdEncoding ENCODING = BlobIdEncoding.fromSystemProperties(); + public static class Factory implements BlobId.Factory { @Override @@ -33,6 +35,11 @@ public BlobId of(String id) { public BlobId parse(String id) { return of(id); } + + @Override + public BlobIdEncoding encoding() { + return ENCODING; + } } private final String rawValue; @@ -46,6 +53,11 @@ public String asString() { return rawValue; } + @Override + public TestBlobId withSuffix(String suffix) { + return new TestBlobId(rawValue + suffix); + } + @Override public final boolean equals(Object o) { if (o instanceof TestBlobId) { diff --git a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CachedBlobStore.java b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CachedBlobStore.java index b6c7bdb7246..aaeaae51ee2 100644 --- a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CachedBlobStore.java +++ b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CachedBlobStore.java @@ -33,6 +33,7 @@ import org.apache.commons.io.IOUtils; 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.BucketName; import org.apache.james.blob.api.ObjectNotFoundException; import org.apache.james.blob.api.ObjectStoreIOException; @@ -47,7 +48,7 @@ import reactor.core.publisher.Mono; -public class CachedBlobStore implements BlobStore { +public class CachedBlobStore implements BlobStore, BlobStoreCacheCallback { private static class ReadAheadInputStream { @@ -361,6 +362,14 @@ private Mono saveInCache(BlobId blobId, byte[] bytes) { return Mono.from(cache.cache(blobId, bytes)); } + @Override + public Publisher cacheIfNeeded(BlobId blobId, byte[] bytes) { + if (isAbleToCache(bytes)) { + return saveInCache(blobId, bytes); + } + return Mono.empty(); + } + private boolean isAbleToCache(BucketName bucketName, byte[] bytes, StoragePolicy storagePolicy) { return isAbleToCache(bucketName, storagePolicy) && isAbleToCache(bytes); } diff --git a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3WithMinIOGenerationAwareBlobIdTest.java b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3WithMinIOGenerationAwareBlobIdTest.java index a13a9e772c7..ef1e6b7dbff 100644 --- a/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3WithMinIOGenerationAwareBlobIdTest.java +++ b/server/blob/blob-s3/src/test/java/org/apache/james/blob/objectstorage/aws/S3WithMinIOGenerationAwareBlobIdTest.java @@ -125,8 +125,8 @@ void saveShouldReturnBlobIdOfString(BlobStore.StoragePolicy storagePolicy) { BlobId blobId = Mono.from(store.save(defaultBucketName, "toto", storagePolicy)).block(); String blobIdString = blobId.asString(); - // Then: BlobId string and parsed BlobId should match expectations - assertThat(blobIdString).isEqualTo("1/628/M/f/emXjFVhqwZi9eYtmKc5JA9CJlHbVdBqfMuLlIbamY="); + // Then: BlobId string and parsed BlobId should match expectations, at the default 128 bits of entropy + assertThat(blobIdString).isEqualTo("1/628/M/f/emXjFVhqwZi9eYtmKc5A"); assertThat(blobId).isEqualTo(blobIdFactory().parse(blobIdString)); } diff --git a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java index f9a9ea10faf..b7523c2af87 100644 --- a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java +++ b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java @@ -25,12 +25,10 @@ import java.time.Clock; import java.time.Instant; import java.util.Collection; -import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; -import java.util.stream.Collectors; import org.apache.james.blob.api.BlobId; import org.apache.james.blob.api.BlobReferenceSource; @@ -41,7 +39,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.annotations.VisibleForTesting; import com.google.common.base.MoreObjects; import com.google.common.hash.BloomFilter; import com.google.common.hash.Funnel; @@ -54,8 +51,6 @@ public class BloomFilterGCAlgorithm { private static final Logger LOGGER = LoggerFactory.getLogger(BloomFilterGCAlgorithm.class); private static final Funnel BLOOM_FILTER_FUNNEL = Funnels.stringFunnel(StandardCharsets.US_ASCII); - @VisibleForTesting - static boolean RECOVERY_AWARE = Boolean.parseBoolean(System.getProperty("james.gc.recover.aware", "true")); public static class Context { @@ -278,7 +273,6 @@ public Mono gc(int expectedBlobCount, int deletionWindowSize, double ass private Mono gc(BloomFilter bloomFilter, BucketName bucketName, Context context, int deletionWindowSize) { return Flux.from(blobStoreDAO.listBlobs(bucketName)) - .filter(blobId -> !RECOVERY_AWARE || !blobId.asString().startsWith(BlobStoreDAO.RECOVERY_BLOB_PREFIX)) .doOnNext(blobId -> context.incrementBlobCount()) .flatMap(blobId -> Mono.fromCallable(() -> blobIdFactory.parse(blobId.asString()))) .filter(blobId -> { @@ -296,13 +290,8 @@ private Mono gc(BloomFilter bloomFilter, BucketName bucket private Mono handlePagedDeletion(BucketName bucketName, Context context, Flux blobIdFlux) { return blobIdFlux.collectList() - .flatMap(orphanBlobIds -> { - Mono deleteRecoverySidecars = RECOVERY_AWARE - ? Mono.from(blobStoreDAO.delete(bucketName, toRecoveryBlobIds(orphanBlobIds))) - : Mono.empty(); - - return Mono.from(blobStoreDAO.delete(bucketName, (Collection) orphanBlobIds)) - .then(deleteRecoverySidecars) + .flatMap(orphanBlobIds -> + Mono.from(blobStoreDAO.delete(bucketName, (Collection) orphanBlobIds)) .then(Mono.fromCallable(() -> { context.incrementGCedBlobCount(orphanBlobIds.size()); return Result.COMPLETED; @@ -310,14 +299,7 @@ private Mono handlePagedDeletion(BucketName bucketName, Context context, LOGGER.error("Error when gc orphan blob", error); context.incrementErrorCount(); return Mono.just(Result.PARTIAL); - }); - }); - } - - private List toRecoveryBlobIds(List blobIds) { - return blobIds.stream() - .map(blobId -> blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + blobId.asString())) - .collect(Collectors.toList()); + })); } private Mono> populatedBloomFilter(int expectedBlobCount, double associatedProbability, Context context) { diff --git a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java index ddd01b08f6c..7f370d6b2c5 100644 --- a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java +++ b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java @@ -27,7 +27,7 @@ import java.util.Optional; import org.apache.james.blob.api.BlobId; -import org.apache.james.blob.api.BlobStoreDAO; +import org.apache.james.blob.api.BlobIdEncoding; import org.apache.james.util.DurationParser; import com.google.common.annotations.VisibleForTesting; @@ -129,11 +129,6 @@ public GenerationAwareBlobId of(String id) { @Override public GenerationAwareBlobId parse(String id) { - // Recovery sidecar keys (eg. recovery/1_2_blobId) do not follow the family_generation_blobId - // layout: keep them as a plain, non-generation-aware blob id preserving the original string. - if (id.startsWith(BlobStoreDAO.RECOVERY_BLOB_PREFIX)) { - return decorateWithoutGeneration(id); - } int separatorIndex1 = id.indexOf('_'); if (separatorIndex1 == -1 || separatorIndex1 == id.length() - 1) { return decorateWithoutGeneration(id); @@ -149,6 +144,11 @@ public GenerationAwareBlobId parse(String id) { return new GenerationAwareBlobId(generation, family, wrapped); } + @Override + public BlobIdEncoding encoding() { + return delegate.encoding(); + } + private GenerationAwareBlobId decorateWithoutGeneration(String id) { return new GenerationAwareBlobId(NO_GENERATION, NO_FAMILY, delegate.parse(id)); } @@ -188,6 +188,11 @@ public String asString() { return family + "_" + generation + "_" + delegate.asString(); } + @Override + public GenerationAwareBlobId withSuffix(String suffix) { + return new GenerationAwareBlobId(generation, family, delegate.withSuffix(suffix)); + } + @Override public boolean inActiveGeneration(Configuration configuration, Instant now) { return configuration.getFamily() == this.family && diff --git a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/MinIOGenerationAwareBlobId.java b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/MinIOGenerationAwareBlobId.java index c612e17e532..dc34fedb1e1 100644 --- a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/MinIOGenerationAwareBlobId.java +++ b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/MinIOGenerationAwareBlobId.java @@ -29,6 +29,7 @@ import jakarta.inject.Inject; import org.apache.james.blob.api.BlobId; +import org.apache.james.blob.api.BlobIdEncoding; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.MoreObjects; @@ -73,6 +74,11 @@ public BlobId parse(String id) { } } + @Override + public BlobIdEncoding encoding() { + return delegate.encoding(); + } + private static String injectFoldersInBlobId(String blobIdPart) { int folderDepthToCreate = 2; if (blobIdPart.length() > folderDepthToCreate) { @@ -120,6 +126,11 @@ public String asString() { return family + "/" + generation + "/" + delegate.asString(); } + @Override + public MinIOGenerationAwareBlobId withSuffix(String suffix) { + return new MinIOGenerationAwareBlobId(generation, family, delegate.withSuffix(suffix)); + } + @Override public boolean inActiveGeneration(GenerationAwareBlobId.Configuration configuration, Instant now) { return configuration.getFamily() == this.family && diff --git a/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/DeDuplicationBlobStore.scala b/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/DeDuplicationBlobStore.scala index 25651d63c97..202d4007679 100644 --- a/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/DeDuplicationBlobStore.scala +++ b/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/DeDuplicationBlobStore.scala @@ -20,8 +20,8 @@ package org.apache.james.server.blob.deduplication import com.google.common.base.Preconditions -import com.google.common.hash.{HashCode, Hashing, HashingInputStream} -import com.google.common.io.{BaseEncoding, ByteSource, FileBackedOutputStream} +import com.google.common.hash.{Hashing, HashingInputStream} +import com.google.common.io.{ByteSource, FileBackedOutputStream} import jakarta.inject.{Inject, Named} import org.apache.commons.io.IOUtils import org.apache.james.blob.api.BlobStore.BlobIdProvider @@ -43,31 +43,12 @@ object DeDuplicationBlobStore { val FILE_THRESHOLD = Integer.parseInt(System.getProperty("james.deduplicating.blobstore.file.threshold", "10240")) val THREAD_SWITCH_THRESHOLD = Integer.parseInt(System.getProperty("james.deduplicating.blobstore.thread.switch.threshold", "32768")); - private def baseEncodingFrom(encodingType: String): BaseEncoding = encodingType match { - case "base16" => - BaseEncoding.base16 - case "hex" => - BaseEncoding.base16 - case "base64" => - BaseEncoding.base64 - case "base64Url" => - BaseEncoding.base64Url - case "base32" => - BaseEncoding.base32 - case "base32Hex" => - BaseEncoding.base32Hex - case _ => - throw new IllegalArgumentException("Unknown encoding type: " + encodingType) - } } class DeDuplicationBlobStore @Inject()(blobStoreDAO: BlobStoreDAO, @Named(BlobStore.DEFAULT_BUCKET_NAME_QUALIFIER) defaultBucketName: BucketName, blobIdFactory: BlobId.Factory) extends BlobStore { - private val HASH_BLOB_ID_ENCODING_TYPE_PROPERTY = "james.blob.id.hash.encoding" - private val HASH_BLOB_ID_ENCODING_DEFAULT = BaseEncoding.base64Url - private val baseEncoding = Option(System.getProperty(HASH_BLOB_ID_ENCODING_TYPE_PROPERTY)).map(DeDuplicationBlobStore.baseEncodingFrom).getOrElse(HASH_BLOB_ID_ENCODING_DEFAULT) override def save(bucketName: BucketName, data: Array[Byte], storagePolicy: BlobStore.StoragePolicy): Publisher[BlobId] = { save(bucketName, data, withBlobIdFromArray, storagePolicy) @@ -108,7 +89,7 @@ class DeDuplicationBlobStore @Inject()(blobStoreDAO: BlobStoreDAO, (fileBackedOutputStream: FileBackedOutputStream) => SMono.fromCallable(() => { IOUtils.copy(hashingInputStream, fileBackedOutputStream) - (blobIdFactory.of(base64(hashingInputStream.hash)), fileBackedOutputStream.asByteSource.openStream()) + (blobIdFactory.ofHash(hashingInputStream.hash.asBytes), fileBackedOutputStream.asByteSource.openStream()) }).asJava() Mono.using[(BlobId, InputStream),FileBackedOutputStream]( @@ -123,29 +104,21 @@ class DeDuplicationBlobStore @Inject()(blobStoreDAO: BlobStoreDAO, private def withBlobIdFromByteSource: BlobIdProvider[ByteSource] = data => Mono.fromCallable(() => data.hash(Hashing.sha256())) .subscribeOn(Schedulers.boundedElastic()) - .map(base64) - .map(blobIdFactory.of) + .map(hash => blobIdFactory.ofHash(hash.asBytes)) .map(blobId => Tuples.of(blobId, data)) private def withBlobIdFromArray: BlobIdProvider[Array[Byte]] = data => { if (data.length < THREAD_SWITCH_THRESHOLD) { - val code = Hashing.sha256.hashBytes(data) - val blobId = blobIdFactory.of(base64(code)) + val blobId = blobIdFactory.ofHash(Hashing.sha256.hashBytes(data).asBytes) Mono.just(Tuples.of(blobId, data)) } else { SMono.fromCallable(() => { - val code = Hashing.sha256.hashBytes(data) - val blobId = blobIdFactory.of(base64(code)) + val blobId = blobIdFactory.ofHash(Hashing.sha256.hashBytes(data).asBytes) Tuples.of(blobId, data) }) } } - private def base64(hashCode: HashCode) = { - val bytes = hashCode.asBytes - baseEncoding.encode(bytes) - } - override def save(bucketName: BucketName, data: InputStream, blobIdProvider: BlobIdProvider[InputStream], diff --git a/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/PassThroughBlobStore.scala b/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/PassThroughBlobStore.scala index 9d5c677b4c4..a5493e8bd0b 100644 --- a/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/PassThroughBlobStore.scala +++ b/server/blob/blob-storage-strategy/src/main/scala/org/apache/james/server/blob/deduplication/PassThroughBlobStore.scala @@ -20,7 +20,6 @@ package org.apache.james.server.blob.deduplication import java.io.InputStream -import java.util.UUID import com.google.common.base.Preconditions import com.google.common.io.ByteSource @@ -88,11 +87,11 @@ class PassThroughBlobStore @Inject()(blobStoreDAO: BlobStoreDAO, } private def withBlobId: BlobIdProvider[InputStream] = data => - SMono.just(Tuples.of(blobIdFactory.of(UUID.randomUUID.toString), data)) + SMono.just(Tuples.of(blobIdFactory.random(), data)) private def withBlobIdByteArray: BlobIdProvider[Array[Byte]] = data => - SMono.just(Tuples.of(blobIdFactory.of(UUID.randomUUID.toString), data)) + SMono.just(Tuples.of(blobIdFactory.random(), data)) private def withBlobIdByteSource: BlobIdProvider[ByteSource] = data => - SMono.just(Tuples.of(blobIdFactory.of(UUID.randomUUID.toString), data)) + SMono.just(Tuples.of(blobIdFactory.random(), data)) override def readBytes(bucketName: BucketName, blobId: BlobId): Publisher[Array[Byte]] = { Preconditions.checkNotNull(bucketName) diff --git a/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java b/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java index 46f398611f0..39397f1f145 100644 --- a/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java +++ b/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java @@ -47,7 +47,6 @@ import org.apache.james.utils.UpdatableTickingClock; import org.awaitility.Awaitility; import org.awaitility.core.ConditionFactory; -import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; @@ -84,11 +83,6 @@ default void setUp() { CLOCK.setInstant(NOW.toInstant()); } - @AfterEach - default void tearDown() { - BloomFilterGCAlgorithm.RECOVERY_AWARE = false; - } - default BlobStore blobStore() { return new DeDuplicationBlobStore(blobStoreDAO(), DEFAULT_BUCKET, GENERATION_AWARE_BLOB_ID_FACTORY); } @@ -237,83 +231,6 @@ default void allOrphanBlobIdsShouldRemovedAfterMultipleRunningTimesGC() { }); } - @Test - default void gcShouldPreserveRecoverySidecarOfReferencedBlobWhenRecoveryAware() { - // Without RECOVERY_AWARE the recovery sidecar has NO_FAMILY → inActiveGeneration()=false - // → treated as orphan and deleted even though its parent blob is alive. This tests the fix. - BloomFilterGCAlgorithm.RECOVERY_AWARE = true; - BlobStore blobStore = blobStore(); - BlobId referencedId = Mono.from(blobStore.save(DEFAULT_BUCKET, UUID.randomUUID().toString(), BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block(); - BlobId recoveryBlobId = GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + referencedId.asString()); - Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId, BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block(); - - when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.just(referencedId)); - CLOCK.setInstant(NOW.plusMonths(2).toInstant()); - - Context context = new Context(EXPECTED_BLOB_COUNT, ASSOCIATED_PROBABILITY); - Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT, DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block(); - - assertThat(blobStore.read(DEFAULT_BUCKET, referencedId)).isNotNull(); - assertThat(Mono.from(blobStoreDAO().readBytes(DEFAULT_BUCKET, recoveryBlobId)).block()).isNotNull(); - } - - @Test - default void gcShouldDeleteRecoverySidecarOfReferencedBlobWhenNotRecoveryAware() { - // Documents the unsafe behavior when the flag is off: recovery sidecar of a live blob is GC-ed. - BlobStore blobStore = blobStore(); - BlobId referencedId = Mono.from(blobStore.save(DEFAULT_BUCKET, UUID.randomUUID().toString(), BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block(); - BlobId recoveryBlobId = GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + referencedId.asString()); - Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId, BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block(); - - when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.just(referencedId)); - CLOCK.setInstant(NOW.plusMonths(2).toInstant()); - - Context context = new Context(EXPECTED_BLOB_COUNT, ASSOCIATED_PROBABILITY); - Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT, DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block(); - - assertThat(blobStore.read(DEFAULT_BUCKET, referencedId)).isNotNull(); - assertThatThrownBy(() -> Mono.from(blobStoreDAO().readBytes(DEFAULT_BUCKET, recoveryBlobId)).block()) - .isInstanceOf(ObjectNotFoundException.class); - } - - @Test - default void gcShouldDeleteRecoverySidecarAlongsideOrphanBlobWhenRecoveryAware() { - BloomFilterGCAlgorithm.RECOVERY_AWARE = true; - BlobStore blobStore = blobStore(); - BlobId orphanId = Mono.from(blobStore.save(DEFAULT_BUCKET, UUID.randomUUID().toString(), BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block(); - BlobId recoveryBlobId = GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + orphanId.asString()); - Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId, BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block(); - - when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.empty()); - CLOCK.setInstant(NOW.plusMonths(2).toInstant()); - - Context context = new Context(EXPECTED_BLOB_COUNT, ASSOCIATED_PROBABILITY); - Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT, DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block(); - - assertThatThrownBy(() -> blobStore.read(DEFAULT_BUCKET, orphanId)) - .isInstanceOf(ObjectNotFoundException.class); - assertThatThrownBy(() -> Mono.from(blobStoreDAO().readBytes(DEFAULT_BUCKET, recoveryBlobId)).block()) - .isInstanceOf(ObjectNotFoundException.class); - } - - @Test - default void gcShouldNotCountRecoveryBlobsInStatsWhenRecoveryAware() { - BloomFilterGCAlgorithm.RECOVERY_AWARE = true; - BlobStore blobStore = blobStore(); - BlobId orphanId = Mono.from(blobStore.save(DEFAULT_BUCKET, UUID.randomUUID().toString(), BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block(); - BlobId recoveryBlobId = GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + orphanId.asString()); - Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId, BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block(); - - when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.empty()); - CLOCK.setInstant(NOW.plusMonths(2).toInstant()); - - Context context = new Context(EXPECTED_BLOB_COUNT, ASSOCIATED_PROBABILITY); - Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT, DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block(); - - assertThat(context.snapshot().getBlobCount()).isEqualTo(1); - assertThat(context.snapshot().getGcedBlobCount()).isEqualTo(1); - } - @Test default void gcShouldHandlerErrorWhenException() { when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.empty()); diff --git a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java index d71e7e44ade..5dd427a2a61 100644 --- a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java +++ b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/mailbox/CassandraMailboxModule.java @@ -34,6 +34,7 @@ import org.apache.james.backends.cassandra.components.CassandraDataDefinition; import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; import org.apache.james.blob.api.BlobReferenceSource; +import org.apache.james.blob.api.BlobStoreCacheCallback; import org.apache.james.events.EventListener; import org.apache.james.eventsourcing.Event; import org.apache.james.eventsourcing.eventstore.JsonEventSerializer; @@ -142,6 +143,7 @@ import com.google.inject.Scopes; import com.google.inject.TypeLiteral; import com.google.inject.multibindings.Multibinder; +import com.google.inject.multibindings.OptionalBinder; import com.google.inject.name.Names; public class CassandraMailboxModule extends AbstractModule { @@ -163,6 +165,9 @@ protected void configure() { bind(CassandraMailboxPathV3DAO.class).in(Scopes.SINGLETON); bind(CassandraMailboxRecentsDAO.class).in(Scopes.SINGLETON); bind(CassandraMessageDAOV3.class).in(Scopes.SINGLETON); + // Overridden by CachedBlobStore when the blob store cache is enabled. + OptionalBinder.newOptionalBinder(binder(), BlobStoreCacheCallback.class) + .setDefault().toInstance(BlobStoreCacheCallback.NOOP); bind(CassandraMessageIdDAO.class).in(Scopes.SINGLETON); bind(CassandraMessageIdToImapUidDAO.class).in(Scopes.SINGLETON); bind(CassandraUserMailboxRightsDAO.class).in(Scopes.SINGLETON); diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/blobstore/BlobStoreCacheModulesChooser.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/blobstore/BlobStoreCacheModulesChooser.java index 3b1c2755965..0bc09c2ca56 100644 --- a/server/container/guice/distributed/src/main/java/org/apache/james/modules/blobstore/BlobStoreCacheModulesChooser.java +++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/blobstore/BlobStoreCacheModulesChooser.java @@ -29,6 +29,7 @@ import org.apache.james.backends.cassandra.components.CassandraDataDefinition; import org.apache.james.backends.cassandra.init.configuration.InjectionNames; import org.apache.james.blob.api.BlobStore; +import org.apache.james.blob.api.BlobStoreCacheCallback; import org.apache.james.blob.api.MetricableBlobStore; import org.apache.james.blob.cassandra.cache.BlobStoreCache; import org.apache.james.blob.cassandra.cache.CachedBlobStore; @@ -48,6 +49,7 @@ import com.google.inject.Scopes; import com.google.inject.Singleton; import com.google.inject.multibindings.Multibinder; +import com.google.inject.multibindings.OptionalBinder; import com.google.inject.name.Names; public class BlobStoreCacheModulesChooser { @@ -67,6 +69,12 @@ static class CacheEnabledModule extends AbstractModule { protected void configure() { bind(CassandraBlobStoreCache.class).in(Scopes.SINGLETON); bind(BlobStoreCache.class).to(CassandraBlobStoreCache.class); + bind(CachedBlobStore.class).in(Scopes.SINGLETON); + + // Lets writes performed through the BlobStoreDAO, which knows nothing of StoragePolicy, + // still populate the cache. + OptionalBinder.newOptionalBinder(binder(), BlobStoreCacheCallback.class) + .setBinding().to(CachedBlobStore.class); Multibinder.newSetBinder(binder(), CassandraDataDefinition.class, Names.named(InjectionNames.CACHE)) .addBinding() diff --git a/server/mailrepository/mailrepository-blob/src/main/scala/org/apache/james/mailrepository/blob/BlobMailRepositoryFactory.scala b/server/mailrepository/mailrepository-blob/src/main/scala/org/apache/james/mailrepository/blob/BlobMailRepositoryFactory.scala index a0208ffdff3..976e8691b8c 100644 --- a/server/mailrepository/mailrepository-blob/src/main/scala/org/apache/james/mailrepository/blob/BlobMailRepositoryFactory.scala +++ b/server/mailrepository/mailrepository-blob/src/main/scala/org/apache/james/mailrepository/blob/BlobMailRepositoryFactory.scala @@ -37,6 +37,10 @@ class MailRepositoryBlobIdFactory( override def of(id: String): BlobId = blobIdFactory.of(url.getPath.subPath(id).asString()) + // Random and content addressed ids go through `of`, hence through the url prefix, on their own. + override def encoding(): BlobIdEncoding = + blobIdFactory.encoding() + } class BlobMailRepositoryFactory(blobStoreDao: BlobStoreDAO, diff --git a/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java b/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java index 88edbd757d3..d16a5aa895e 100644 --- a/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java +++ b/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java @@ -31,6 +31,7 @@ import org.apache.james.backends.cassandra.components.CassandraDataDefinition; import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition; +import org.apache.james.blob.api.BlobStoreCacheCallback; import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.PlainBlobId; import org.apache.james.blob.cassandra.CassandraBlobDataDefinition; @@ -170,7 +171,8 @@ void setUp(CassandraCluster cassandra) { .passthrough(), Mockito.mock(BlobStoreDAO.class), new PlainBlobId.Factory(), - CassandraConfiguration.DEFAULT_CONFIGURATION); + CassandraConfiguration.DEFAULT_CONFIGURATION, + BlobStoreCacheCallback.NOOP); testee = new MetaDataFixInconsistenciesService(imapUidDAO, pop3MetadataStore, cassandraMessageDAOV3); } diff --git a/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java b/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java index e0e31212b64..67faea05ff0 100644 --- a/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java +++ b/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java @@ -35,6 +35,7 @@ import org.apache.james.backends.cassandra.components.CassandraDataDefinition; import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration; import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition; +import org.apache.james.blob.api.BlobStoreCacheCallback; import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.blob.api.PlainBlobId; import org.apache.james.blob.cassandra.CassandraBlobDataDefinition; @@ -188,7 +189,8 @@ void setUp(CassandraCluster cassandra) { .passthrough(), Mockito.mock(BlobStoreDAO.class), new PlainBlobId.Factory(), - CassandraConfiguration.DEFAULT_CONFIGURATION); + CassandraConfiguration.DEFAULT_CONFIGURATION, + BlobStoreCacheCallback.NOOP); MetaDataFixInconsistenciesService fixInconsistenciesService = new MetaDataFixInconsistenciesService(imapUidDAO, pop3MetadataStore, cassandraMessageDAOV3); taskManager = new MemoryTaskManager(new Hostname("foo")); diff --git a/upgrade-instructions.md b/upgrade-instructions.md index 8ca73e0c8d5..2dd12243b57 100644 --- a/upgrade-instructions.md +++ b/upgrade-instructions.md @@ -22,6 +22,24 @@ Change list: - [JAMES-4210 SMTP AuthHook deprecation](#james-4210-smtp-authhook-deprecation) - [JAMES-4210 POP3 USER/PASS requires TLS by default](#james-4210-pop3-userpass-requires-tls-by-default) - [JAMES-4210 ManageSieve SASL adoption](#james-4210-managesieve-sasl-adoption) + - [JAMES-4225 Blob ids default to 128 bits of entropy](#james-4225-blob-ids-default-to-128-bits-of-entropy) + +### JAMES-4225 Blob ids default to 128 bits of entropy + +Date: 05/09/2026 + +Concerned products: all products using a blob store + +Blob ids now carry 128 bits of entropy instead of the full 256 bits of a SHA-256, which shortens a body +blob id from 50 to 28 characters in the object key and in every metadata store column referencing it. + +Existing blobs stay readable. However, content stored under the previous spelling will not deduplicate +against its shorter counterpart until it is rewritten. To keep the previous behaviour, add to +`jvm.properties`: + +``` +james.blobid.entropy=256 +``` ### JAMES-4210 POP3 USER/PASS requires TLS by default