This is an automated email from the ASF dual-hosted git repository. quantranhong1999 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/james-project.git
commit f9b02d9039f8eefd5d385c174e04d9fcf5e7a321 Author: Benoit TELLIER <[email protected]> AuthorDate: Fri Sep 4 18:07:09 2026 +0200 JAMES-4209 Inline body blob id in header blob metadata Key design decision: - DUPLICATE headers: needed for recovery info unicity - Use of BlobStoreDAO: apply the above AND allow passing metadata Impact: - full scan of the generation needed - Need to explicitly ask to cache headers - Save-in-sequence body-then-header is needed... - 2 object instead of 3 thus limiting dramatically pressure on S3 store metadata - Simplify GC: no side car handling --- .../init/configuration/CassandraConfiguration.java | 7 +- .../servers/pages/distributed/operate/backup.adoc | 70 +++++++---- .../cassandra/mail/CassandraMessageDAOV3.java | 11 +- .../mail/ContentRecoveryMessageContentSaver.java | 77 +++++++----- .../cassandra/CassandraMailboxManagerTest.java | 4 +- .../cassandra/mail/CassandraMessageDAOV3Test.java | 83 +++++++++---- .../ContentRecoveryMessageContentSaverTest.java | 136 +++++++++++++++++++++ .../mailbox/cassandra/mail/utils/GuiceUtils.java | 2 + server/apps/distributed-app/README.adoc | 54 +++++--- .../org/apache/james/RecoveryConfiguration.java | 87 +++++++++++-- .../java/org/apache/james/S3RecoveryService.java | 78 ++++++------ .../apache/james/RecoveryConfigurationTest.java | 44 ++++++- .../org/apache/james/blob/api/BlobStoreDAO.java | 5 +- .../blob/deduplication/BloomFilterGCAlgorithm.java | 24 +--- .../blob/deduplication/GenerationAwareBlobId.java | 6 - .../BloomFilterGCAlgorithmContract.java | 83 ------------- 16 files changed, 493 insertions(+), 278 deletions(-) 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 9c0ffb7e5c..e4b272b46d 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 92550ab174..ea508c461a 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/<headerBlobId>` *sidecar* blob holding the matching -`bodyBlobId`. The blob garbage collection is aware of these sidecars and never deletes a live one. +survives*. Because blobs alone do not tell which user a message belonged to, James can optionally record +the id of a message body as *metadata of its header blob*, and suffix header blob ids with `_hdr` so that +they can be told apart when walking the store. No extra object is written, and the blob garbage collection +needs no special handling. -Recording of the sidecars is controlled in `cassandra.properties`: +Recording of the recovery information is controlled in `cassandra.properties`: [source,properties] ---- -# none (default), synchronous or asynchronous -mailbox.blob.recovery.mode=synchronous +# none (default) or enabled +mailbox.blob.recovery.mode=enabled ---- -* `synchronous` — 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=<prefix>` argument (also settable via the -`RECOVERY_HEADER_BLOB_PREFIX` environment variable or the `recovery.header.blob.prefix` system -property) narrows the walk to the recovery sidecars whose header blob id starts with the given prefix, -pushing the filter down to S3's `ListObjectsV2`. Header blob ids are generation-aware -(`family_generation_...`), so a clever admin can pass e.g. `1_42_` to iterate solely the latest -generation instead of scanning the whole bucket: +The optional `--family=<int>` and `--generation=<long>` arguments (also settable via the +`RECOVERY_FAMILY` and `RECOVERY_GENERATION` environment variables, or the `recovery.family` and +`recovery.generation` system properties) narrow the walk to a single generation of the generation-aware +blob ids (`family_generation_...`), pushing the filter down to S3's `ListObjectsV2` rather than filtering +a full listing. Both default to walking the whole bucket. + +Because the generation is the *second* component of a blob id, there is no listing prefix for a generation +on its own: `--generation` requires `--family`. [source,bash] ---- -... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ +... org.apache.james.S3RecoveryMain --family=1 --generation=42 ---- -The dominant cost of a recovery run is not the listing but the per-message work: three blob reads plus +Deployments configured with the MinIO blob id strategy separate the family and generation with `/` rather +than `_`. Declare it with `--minio-separator` (or `RECOVERY_MINIO_SEPARATOR`, or +`recovery.minio.separator`), which turns the prefix above into `1/42/`. + +The dominant cost of a recovery run is not the listing but the per-message work: two blob reads plus a full re-store of each message through the mailbox. The `--concurrency=<n>` argument (default `8`, also settable via the `RECOVERY_CONCURRENCY` environment variable or the `recovery.concurrency` system property) controls how many messages are restored in parallel and is therefore the main lever on @@ -190,16 +206,16 @@ back off if they become the bottleneck: === Scaling recovery: shard by generation For very large recoveries, a single process is limited by its own concurrency and offers no easy resume -point. Because `--header-blob-prefix` scopes a run to a slice of the key space, you can *shard* the -recovery across several independent processes — 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/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 e5032a139f..375d61bc93 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 @@ -55,6 +55,7 @@ import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor; 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; @@ -110,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.blobIdFactory = blobIdFactory; - this.messageContentSaver = messageContentSaver(blobStore, blobStoreDAO, blobIdFactory, cassandraConfiguration); + this.messageContentSaver = messageContentSaver(blobStore, blobStoreDAO, blobIdFactory, cassandraConfiguration, cacheCallback); this.insert = prepareInsert(session); this.delete = prepareDelete(session); @@ -131,11 +132,11 @@ public class CassandraMessageDAOV3 { } private static MessageContentSaver messageContentSaver(BlobStore blobStore, BlobStoreDAO blobStoreDAO, - BlobId.Factory blobIdFactory, CassandraConfiguration configuration) { + BlobId.Factory blobIdFactory, CassandraConfiguration configuration, + BlobStoreCacheCallback cacheCallback) { return switch (configuration.getBlobRecoveryMode()) { case NONE -> new DefaultMessageContentSaver(blobStore); - case SYNCHRONOUS, ASYNCHRONOUS -> new ContentRecoveryMessageContentSaver(blobStore, blobStoreDAO, - blobIdFactory, configuration.getBlobRecoveryMode()); + case ENABLED -> new ContentRecoveryMessageContentSaver(blobStore, blobStoreDAO, blobIdFactory, cacheCallback); }; } 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 index 74c9175790..c6a7c05443 100644 --- 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 @@ -19,69 +19,80 @@ package org.apache.james.mailbox.cassandra.mail; -import java.nio.charset.StandardCharsets; +import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST; -import org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration.BlobRecoveryMode; import org.apache.james.blob.api.BlobId; +import org.apache.james.blob.api.BlobIdEntropy; import org.apache.james.blob.api.BlobStore; +import org.apache.james.blob.api.BlobStoreCacheCallback; import org.apache.james.blob.api.BlobStoreDAO; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +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.base.Preconditions; +import com.google.common.io.BaseEncoding; import com.google.common.io.ByteSource; import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; import reactor.util.function.Tuple2; +import reactor.util.function.Tuples; /** - * Delegates the content write, then materializes the recovery information as a sidecar blob: - * the body blob id, stored under the header blob id prefixed by {@link BlobStoreDAO#RECOVERY_BLOB_PREFIX}. + * Carries the recovery information within the header blob itself, rather than in a companion object. * - * The sidecar write is either awaited ({@link BlobRecoveryMode#SYNCHRONOUS}) or performed on the side - * ({@link BlobRecoveryMode#ASYNCHRONOUS}). + * <p>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.</p> + * + * <p>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.</p> + * + * <p>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}.</p> */ public class ContentRecoveryMessageContentSaver implements MessageContentSaver { - private static final Logger LOGGER = LoggerFactory.getLogger(ContentRecoveryMessageContentSaver.class); + public static final String HEADER_BLOB_ID_SUFFIX = "_hdr"; + public static final BlobMetadataName BODY_BLOB_ID = new BlobMetadataName("body-blob-id"); + private static final BaseEncoding BLOB_ID_ENCODING = BaseEncoding.base64Url().omitPadding(); - private final MessageContentSaver delegate; private final BlobStore blobStore; private final BlobStoreDAO blobStoreDAO; private final BlobId.Factory blobIdFactory; - private final BlobRecoveryMode recoveryMode; + private final BlobStoreCacheCallback cacheCallback; public ContentRecoveryMessageContentSaver(BlobStore blobStore, BlobStoreDAO blobStoreDAO, - BlobId.Factory blobIdFactory, BlobRecoveryMode recoveryMode) { - Preconditions.checkArgument(recoveryMode != BlobRecoveryMode.NONE, - "%s does not handle %s: rely on the delegate alone instead", ContentRecoveryMessageContentSaver.class.getSimpleName(), BlobRecoveryMode.NONE); - this.delegate = new DefaultMessageContentSaver(blobStore); + BlobId.Factory blobIdFactory, BlobStoreCacheCallback cacheCallback) { this.blobStore = blobStore; this.blobStoreDAO = blobStoreDAO; this.blobIdFactory = blobIdFactory; - this.recoveryMode = recoveryMode; + this.cacheCallback = cacheCallback; } @Override public Mono<Tuple2<BlobId, BlobId>> saveContent(byte[] headerBytes, ByteSource bodyByteSource) { - return delegate.saveContent(headerBytes, bodyByteSource) - .flatMap(pair -> saveRecovery(pair.getT1(), pair.getT2()).thenReturn(pair)); + return Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource, LOW_COST)) + .flatMap(bodyId -> saveHeaders(headerBytes, bodyId) + .map(headerId -> Tuples.of(headerId, bodyId))); } - private Mono<Void> saveRecovery(BlobId headerId, BlobId bodyId) { - return switch (recoveryMode) { - 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<BlobId> 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); } - private Mono<Void> writeRecoveryBlob(BlobId headerId, BlobId bodyId) { - BlobId recoveryBlobId = blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + headerId.asString()); - BlobStoreDAO.BytesBlob content = BlobStoreDAO.BytesBlob.of(bodyId.asString().getBytes(StandardCharsets.UTF_8)); - return Mono.from(blobStoreDAO.save(blobStore.getDefaultBucketName(), recoveryBlobId, content)); + /** + * 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.of(BLOB_ID_ENCODING.encode(BlobIdEntropy.randomBytes()) + HEADER_BLOB_ID_SUFFIX); } } 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 a13b5293fb..d460ac5b94 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.CassandraClusterExtension; 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 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai 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 9a3067e1fc..652a120bcc 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 @@ package org.apache.james.mailbox.cassandra.mail; 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.init.configuration.CassandraConfigura 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 @@ class CassandraMessageDAOV3Test { blobStore, blobStoreDAO, blobIdFactory, - configuration); + configuration, + BlobStoreCacheCallback.NOOP); } @Test @@ -172,47 +172,78 @@ class CassandraMessageDAOV3Test { } @Test - void saveShouldNotWriteRecoveryBlobByDefault() throws Exception { + void saveShouldNotCarryRecoveryInformationByDefault() throws Exception { message = createMessage(messageId, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); Tuple2<BlobId, BlobId> 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<BlobId, BlobId> 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<BlobId, BlobId> 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<BlobId, BlobId> 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<BlobId, BlobId> blobIds = testeeWithRecovery.save(message).block(); + + message = createMessage(messageId2, threadId, CONTENT, BODY_START, NO_ATTACHMENT, EMPTY_SAVE_DATE); + Tuple2<BlobId, BlobId> 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 0000000000..ca1e449f64 --- /dev/null +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/ContentRecoveryMessageContentSaverTest.java @@ -0,0 +1,136 @@ +/**************************************************************** + * 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.BlobIdEntropy; +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.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 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. + * + * <p>Those three properties have to hold for every {@link BlobId.Factory} a deployment may be configured + * with, which is what this pins down.</p> + */ +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)); + private static final BaseEncoding BLOB_ID_ENCODING = BaseEncoding.base64Url().omitPadding(); + + /** + * 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<Arguments> 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()); + } + + @Test + void headerBlobIdShouldDrawTheConfiguredEntropy() { + String headerBlobId = saveContent(new PlainBlobId.Factory()).getT1().asString(); + String randomPart = headerBlobId.substring(0, headerBlobId.length() - HEADER_BLOB_ID_SUFFIX.length()); + + assertThat(BLOB_ID_ENCODING.decode(randomPart)) + .hasSize(BlobIdEntropy.entropyBytes()); + } + + private Tuple2<BlobId, BlobId> 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 3688acb50f..eda113d31d 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.CassandraTypesProvider; 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 class GuiceUtils { 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 3371787798..aa0aa9a214 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/<headerBlobId>` 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=<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=<int>` and `--generation=<long>` arguments (also settable via the `RECOVERY_FAMILY` and +`RECOVERY_GENERATION` environment variables, or the `recovery.family` and `recovery.generation` system +properties) narrow the walk to a single generation of the generation-aware blob ids. 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=<n>` argument (default `8`, also settable via the `RECOVERY_CONCURRENCY` environment variable or the `recovery.concurrency` system property) controls how many messages are restored in parallel. 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/src/main/java/org/apache/james/RecoveryConfiguration.java b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java index 75ca00fb58..35cb07148e 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 @@ import com.google.common.base.Preconditions; * {@code --restore-after=<ISO-8601 instant>} program argument, the {@code RESTORE_MESSAGES_AFTER} * environment variable, or the {@code restore.messages.after} system property.</p> * - * <p>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=<prefix>} program argument, - * the {@code RECOVERY_HEADER_BLOB_PREFIX} environment variable, or the {@code recovery.header.blob.prefix} - * system property.</p> + * <p>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=<int>} / {@code --generation=<long>}, 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.</p> + * + * <p>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.</p> * * <p>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 @@ import com.google.common.base.Preconditions; * as a {@code --concurrency=<n>} program argument, the {@code RECOVERY_CONCURRENCY} environment * variable, or the {@code recovery.concurrency} system property.</p> */ -public record RecoveryConfiguration(Optional<Instant> restoreAfter, String headerBlobPrefix, int concurrency) { +public record RecoveryConfiguration(Optional<Instant> restoreAfter, Optional<Integer> family, + Optional<Long> 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<String> option(String[] args, String argPrefix, String envName, String propertyName) { return Arrays.stream(args) .filter(arg -> arg.startsWith(argPrefix)) @@ -82,6 +114,15 @@ public record RecoveryConfiguration(Optional<Instant> restoreAfter, String heade .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 @@ public record RecoveryConfiguration(Optional<Instant> restoreAfter, String heade } } + 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 f693149fca..96b1c04274 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.publisher.Mono; 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. * - * <p>Reads go through the configured {@link BlobStore}, so AES decryption and decompression are applied + * <p>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.</p> + * + * <p>Reads go through the decorated {@link BlobStoreDAO}, so AES decryption and decompression are applied * transparently, exactly as the write path did.</p> */ 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 @@ public class S3RecoveryService { restored + other.restored, skippedByDate + other.skippedByDate, skippedNoLocalUser + other.skippedNoLocalUser, + skippedNoRecoveryInfo + other.skippedNoRecoveryInfo, failed + other.failed); } } @@ -89,10 +94,11 @@ public class S3RecoveryService { 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 class S3RecoveryService { public Mono<Report> 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<Report> 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<Report> 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<RecoveredMessage> 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<BlobId> bodyBlobId(BlobStoreDAO.BytesBlob headerBlob) { + return headerBlob.metadata() + .get(BODY_BLOB_ID) + .map(BlobStoreDAO.BlobMetadataValue::value) + .map(blobIdFactory::parse); + } + + private Mono<RecoveredMessage> 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<Report> 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 e2205b8e3a..bc587b70b1 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 @@ class RecoveryConfigurationTest { } @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/BlobStoreDAO.java b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java index 223dafe3d8..46af6ebed1 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 @@ import reactor.core.publisher.Flux; * <p>See {@code docs/modules/servers/partials/architecture/blobstore.adoc} for more details.</p> */ 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 interface BlobStoreDAO { Publisher<BlobId> 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). * * <p>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.</p> 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 f9a9ea10fa..b7523c2af8 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.nio.charset.StandardCharsets; 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.apache.james.task.Task.Result; 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<CharSequence> 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 class BloomFilterGCAlgorithm { private Mono<Result> gc(BloomFilter<CharSequence> 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 @@ public class BloomFilterGCAlgorithm { private Mono<Result> handlePagedDeletion(BucketName bucketName, Context context, Flux<BlobId> blobIdFlux) { return blobIdFlux.collectList() - .flatMap(orphanBlobIds -> { - Mono<Void> 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 @@ public class BloomFilterGCAlgorithm { LOGGER.error("Error when gc orphan blob", error); context.incrementErrorCount(); return Mono.just(Result.PARTIAL); - }); - }); - } - - private List<BlobId> toRecoveryBlobIds(List<BlobId> blobIds) { - return blobIds.stream() - .map(blobId -> blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + blobId.asString())) - .collect(Collectors.toList()); + })); } private Mono<BloomFilter<CharSequence>> 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 ddd01b08f6..67fd492b16 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,6 @@ import java.util.Objects; import java.util.Optional; import org.apache.james.blob.api.BlobId; -import org.apache.james.blob.api.BlobStoreDAO; import org.apache.james.util.DurationParser; import com.google.common.annotations.VisibleForTesting; @@ -129,11 +128,6 @@ public class GenerationAwareBlobId implements BlobId, GenerationAware { @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); 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 46f398611f..39397f1f14 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.task.Task; 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 @@ public interface BloomFilterGCAlgorithmContract { 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 @@ public interface BloomFilterGCAlgorithmContract { }); } - @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()); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
