This is an automated email from the ASF dual-hosted git repository. chibenwa pushed a commit to branch 3.9.x in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 6e842bb605a22481297d25643c2a137abead100e Author: Benoit TELLIER <[email protected]> AuthorDate: Wed Jun 17 00:22:05 2026 +0200 [FIX] Enhance SolveMailboxInconsistencies to fix duplicated mailboxPathV3 registration and recover from failed mailboxPathV3 deletion upon rename --- .../cassandra/mail/CassandraMailboxMapper.java | 8 +- .../cassandra/mail/CassandraMailboxPathV3DAO.java | 21 +++++ .../task/SolveMailboxInconsistenciesService.java | 101 ++++++++++++++++++--- .../cassandra/mail/CassandraMailboxMapperTest.java | 38 +++++++- .../SolveMailboxInconsistenciesServiceTest.java | 39 ++++++-- 5 files changed, 180 insertions(+), 27 deletions(-) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java index eceec29a4b..b5fd64291e 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapper.java @@ -233,8 +233,12 @@ public class CassandraMailboxMapper implements MailboxMapper { return mailboxPathV3DAO.save(cassandraMailbox) .handle(ReactorUtils.raiseErrorIfFalse(() -> new MailboxExistsException(cassandraMailbox.generateAssociatedPath().asString()))) - .flatMap(applied -> deletePreviousMailboxPathReference(previousPath) - .then(persistMailboxEntity(cassandraMailbox))) + // Additive writes first (the new path reference and the projection), subtractive write + // last (drop the old path reference). At any crash point the mailbox thus stays + // reachable under its new path with a consistent projection; only a stale old path + // reference may linger (the delete is retried to absorb transient failures). + .flatMap(applied -> persistMailboxEntity(cassandraMailbox) + .then(deletePreviousMailboxPathReference(previousPath))) .thenReturn(cassandraId); } diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxPathV3DAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxPathV3DAO.java index 22c0f31c59..7663a18ec0 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxPathV3DAO.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxPathV3DAO.java @@ -63,6 +63,7 @@ public class CassandraMailboxPathV3DAO { private final PreparedStatement delete; private final PreparedStatement insert; private final PreparedStatement select; + private final PreparedStatement selectWriteTime; private final PreparedStatement selectUser; private final PreparedStatement selectAll; private final CqlSession session; @@ -75,6 +76,7 @@ public class CassandraMailboxPathV3DAO { this.insert = prepareInsert(); this.delete = prepareDelete(); this.select = prepareSelect(); + this.selectWriteTime = prepareSelectWriteTime(); this.selectUser = prepareSelectUser(); this.selectAll = prepareSelectAll(); this.lwtProfile = JamesExecutionProfiles.getLWTProfile(session); @@ -109,6 +111,15 @@ public class CassandraMailboxPathV3DAO { .build()); } + private PreparedStatement prepareSelectWriteTime() { + return session.prepare(selectFrom(TABLE_NAME) + .writeTime(MAILBOX_ID) + .where(column(NAMESPACE).isEqualTo(bindMarker(NAMESPACE)), + column(USER).isEqualTo(bindMarker(USER)), + column(MAILBOX_NAME).isEqualTo(bindMarker(MAILBOX_NAME))) + .build()); + } + private PreparedStatement prepareSelectUser() { return session.prepare(selectFrom(TABLE_NAME) .columns(MAILBOX_ID, UIDVALIDITY, MAILBOX_NAME) @@ -139,6 +150,16 @@ public class CassandraMailboxPathV3DAO { .switchIfEmpty(ReactorUtils.executeAndEmpty(() -> logGhostMailboxFailure(mailboxPath))); } + public Mono<Long> writeTime(MailboxPath mailboxPath) { + BoundStatement statement = selectWriteTime.bind() + .set(NAMESPACE, mailboxPath.getNamespace(), TypeCodecs.TEXT) + .set(USER, sanitizeUser(mailboxPath.getUser()), TypeCodecs.TEXT) + .set(MAILBOX_NAME, mailboxPath.getName(), TypeCodecs.TEXT); + + return cassandraAsyncExecutor.executeSingleRow(setExecutionProfileIfNeeded(statement, STRONG)) + .map(row -> row.getLong(0)); + } + public Flux<Mailbox> listUserMailboxes(String namespace, Username user, JamesExecutionProfiles.ConsistencyChoice consistencyChoice) { BoundStatementBuilder statementBuilder = selectUser.boundStatementBuilder() .set(NAMESPACE, namespace, TypeCodecs.TEXT) diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesService.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesService.java index d1d5fcf414..0a76fab0d9 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesService.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesService.java @@ -37,6 +37,7 @@ import org.apache.james.mailbox.cassandra.mail.CassandraMailboxDAO; import org.apache.james.mailbox.cassandra.mail.CassandraMailboxPathV3DAO; import org.apache.james.mailbox.model.Mailbox; import org.apache.james.mailbox.model.MailboxId; +import org.apache.james.mailbox.model.MailboxPath; import org.apache.james.task.Task; import org.apache.james.task.Task.Result; import org.slf4j.Logger; @@ -61,9 +62,7 @@ public class SolveMailboxInconsistenciesService { return NO_INCONSISTENCY; } // Path entry references another mailbox. - return new ConflictingEntryInconsistency(ConflictingEntry.builder() - .mailboxDaoEntry(mailboxEntry) - .mailboxPathDaoEntry(mailboxByPath)); + return new ConflictingEntryInconsistency(mailboxEntry, mailboxByPath); }) .defaultIfEmpty(new OrphanMailboxDAOEntry(mailboxEntry)); } @@ -76,9 +75,7 @@ public class SolveMailboxInconsistenciesService { return NO_INCONSISTENCY; } // Mailbox references another path - return new ConflictingEntryInconsistency(ConflictingEntry.builder() - .mailboxDaoEntry(mailboxById) - .mailboxPathDaoEntry(mailboxByPathEntry)); + return new ConflictingEntryInconsistency(mailboxById, mailboxByPathEntry); }) .defaultIfEmpty(new OrphanMailboxPathDAOEntry(mailboxByPathEntry)); } @@ -181,19 +178,87 @@ public class SolveMailboxInconsistenciesService { * See https://github.com/apache/james-project/blob/master/src/site/markdown/server/manage-webadmin.md#correcting-ghost-mailbox */ private static class ConflictingEntryInconsistency implements Inconsistency { - private final ConflictingEntry conflictingEntry; + private final Mailbox mailboxDaoEntry; + private final Mailbox mailboxPathEntry; - private ConflictingEntryInconsistency(ConflictingEntry conflictingEntry) { - this.conflictingEntry = conflictingEntry; + private ConflictingEntryInconsistency(Mailbox mailboxDaoEntry, Mailbox mailboxPathEntry) { + this.mailboxDaoEntry = mailboxDaoEntry; + this.mailboxPathEntry = mailboxPathEntry; } @Override public Mono<Result> fix(Context context, CassandraMailboxDAO mailboxDAO, CassandraMailboxPathV3DAO pathV3DAO) { + if (mailboxDaoEntry.getMailboxId().equals(mailboxPathEntry.getMailboxId())) { + // Same mailbox referenced by two different paths. This happens when a rename (or any + // partial path update) failed to drop the old path reference. As both registrations + // point to the same mailbox id, the messages (keyed by id) are safe regardless of the + // path we keep: we keep the most recently written path reference, realign the + // projection on it and drop the stale one. + return fixSameMailboxConflict(context, mailboxDAO, pathV3DAO); + } + + return reportConflict(context); + } + + // Auto-resolution is restricted to the case where BOTH the conflicting path entry and the + // projection's path are registered to the same mailbox id: only then are they two genuine + // aliases of a single mailbox, hence safe to deduplicate without data loss. If the + // projection's path is unregistered or held by another mailbox (e.g. a reference loop), we + // fall back to the conservative reporting so an admin can merge. + // + // As fixes are applied sequentially, we re-read the current state first: a previous fix may + // already have reconciled this mailbox, in which case there is nothing left to do. + private Mono<Result> fixSameMailboxConflict(Context context, CassandraMailboxDAO mailboxDAO, CassandraMailboxPathV3DAO pathV3DAO) { + CassandraId mailboxId = (CassandraId) mailboxPathEntry.getMailboxId(); + MailboxPath conflictingPath = mailboxPathEntry.generateAssociatedPath(); + + return pathV3DAO.retrieve(conflictingPath, STRONG) + .filter(stillRegistered -> stillRegistered.getMailboxId().equals(mailboxId)) + .flatMap(conflictingEntry -> mailboxDAO.retrieveMailbox(mailboxId) + .flatMap(currentProjection -> resolveSameMailboxConflict(context, mailboxDAO, pathV3DAO, currentProjection, conflictingEntry))) + // The conflicting path entry is gone (already reconciled): nothing left to do. + .switchIfEmpty(Mono.just(Result.COMPLETED)); + } + + private Mono<Result> resolveSameMailboxConflict(Context context, CassandraMailboxDAO mailboxDAO, CassandraMailboxPathV3DAO pathV3DAO, + Mailbox currentProjection, Mailbox conflictingEntry) { + MailboxPath projectionPath = currentProjection.generateAssociatedPath(); + MailboxPath conflictingPath = conflictingEntry.generateAssociatedPath(); + if (projectionPath.equals(conflictingPath)) { + // The projection now matches this path: no longer inconsistent. + return Mono.just(Result.COMPLETED); + } + return pathV3DAO.retrieve(projectionPath, STRONG) + .filter(projectionRegistration -> projectionRegistration.getMailboxId().equals(currentProjection.getMailboxId())) + .flatMap(projectionRegistration -> Mono.zip( + pathV3DAO.writeTime(conflictingPath), + pathV3DAO.writeTime(projectionPath)) + .flatMap(writeTimes -> { + boolean projectionWins = writeTimes.getT2() >= writeTimes.getT1(); + Mailbox winner = projectionWins ? currentProjection : conflictingEntry; + Mailbox loser = projectionWins ? conflictingEntry : currentProjection; + return mailboxDAO.save(winner) + .then(pathV3DAO.delete(loser.generateAssociatedPath())) + .then(Mono.fromRunnable(() -> { + LOGGER.info("Inconsistency fixed for mailbox {}: kept path {}, dropped stale path {}", + winner.getMailboxId().serialize(), + winner.generateAssociatedPath().asString(), + loser.generateAssociatedPath().asString()); + context.addFixedInconsistency(winner.getMailboxId()); + })) + .thenReturn(Result.COMPLETED); + })) + .switchIfEmpty(Mono.defer(() -> reportConflict(context))); + } + + private Mono<Result> reportConflict(Context context) { LOGGER.error("MailboxDAO contains mailbox {} {} which conflict with corresponding registration {} {}. " + "We recommend merging these mailboxes together to prevent mail data loss.", - conflictingEntry.getMailboxDaoEntry().getMailboxId(), conflictingEntry.getMailboxDaoEntry().getMailboxPath(), - conflictingEntry.getMailboxPathDaoEntry().getMailboxId(), conflictingEntry.getMailboxPathDaoEntry().getMailboxPath()); - context.addConflictingEntries(conflictingEntry); + mailboxDaoEntry.getMailboxId(), mailboxDaoEntry.generateAssociatedPath(), + mailboxPathEntry.getMailboxId(), mailboxPathEntry.generateAssociatedPath()); + context.addConflictingEntries(ConflictingEntry.builder() + .mailboxDaoEntry(mailboxDaoEntry) + .mailboxPathDaoEntry(mailboxPathEntry)); return Mono.just(Result.PARTIAL); } } @@ -413,8 +478,16 @@ public class SolveMailboxInconsistenciesService { private Flux<Result> processMailboxPathDaoInconsistencies(Context context) { return mailboxPathV3DAO.listAll() .flatMap(this::detectMailboxPathDaoInconsistency, DEFAULT_CONCURRENCY) - .flatMap(inconsistency -> inconsistency.fix(context, mailboxDAO, mailboxPathV3DAO), DEFAULT_CONCURRENCY) - .doOnNext(any -> context.incrementProcessedMailboxPathEntries()); + .doOnNext(any -> context.incrementProcessedMailboxPathEntries()) + // Detect every inconsistency first, then fix them one at a time. Resolving a same-mailbox + // conflict may realign the projection, so a fully materialized detection set fixed + // sequentially prevents a fix from racing with the detection or resolution of a sibling + // path entry. Consistent entries are filtered out to keep the materialized set small; + // each fix re-confirms the inconsistency against the current state before acting. + .filter(inconsistency -> inconsistency != NO_INCONSISTENCY) + .collectList() + .flatMapMany(inconsistencies -> Flux.fromIterable(inconsistencies) + .concatMap(inconsistency -> inconsistency.fix(context, mailboxDAO, mailboxPathV3DAO))); } private Flux<Result> processMailboxDaoInconsistencies(Context context) { diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapperTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapperTest.java index e13e452c7e..c27ffd68b0 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapperTest.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxMapperTest.java @@ -464,10 +464,14 @@ class CassandraMailboxMapperTest { doQuietly(() -> testee.rename(inboxRenamed).block()); + // rename writes additively first (new path reference + projection) and removes the old + // path reference last. When that removal fails, the projection already reflects the + // rename, so the mailbox is consistent under its new identity; the old INBOX path + // reference merely lingers (the delete is retried to absorb transient failures). SoftAssertions.assertSoftly(Throwing.consumer(softly -> { softly(softly) .assertThat(testee.findMailboxById(inboxId).block()) - .isEqualTo(inbox); + .isEqualTo(inboxRenamed); softly(softly) .assertThat(testee.findMailboxByPath(inboxPath).block()) .isEqualTo(inbox); @@ -479,6 +483,38 @@ class CassandraMailboxMapperTest { })); } + @Test + void renameThenFailToDeleteMailboxPathShouldExposeRenamedMailbox(CassandraCluster cassandra) { + Mailbox inbox = testee.create(inboxPath, UID_VALIDITY).block(); + CassandraId inboxId = (CassandraId) inbox.getMailboxId(); + Mailbox inboxRenamed = createInboxRenamedMailbox(inboxId); + + cassandra.getConf() + .registerScenario(fail() + .times(TRY_COUNT_BEFORE_FAILURE) + .whenQueryStartsWith("DELETE FROM mailboxpathv3 WHERE namespace=:namespace AND user=:user AND mailboxname=:mailboxname IF EXISTS")); + + doQuietly(() -> testee.rename(inboxRenamed).block()); + + // Counterpart of the previous test: even though the old path cleanup failed, the rename + // is fully observable under its new identity (by id and by renamed path). With the legacy + // ordering - projection written last, after the delete - the projection would still read + // INBOX here, contradicting the already committed renamed path reference. + SoftAssertions.assertSoftly(Throwing.consumer(softly -> { + softly(softly) + .assertThat(testee.findMailboxById(inboxId).block()) + .isEqualTo(inboxRenamed); + softly(softly) + .assertThat(testee.findMailboxByPath(inboxPathRenamed).block()) + .isEqualTo(inboxRenamed); + softly.assertThat(testee.findMailboxWithPathLike(inboxRenamedSearchQuery) + .collectList().block()) + .hasOnlyOneElementSatisfying(searchMailbox -> softly(softly) + .assertThat(searchMailbox) + .isEqualTo(inboxRenamed)); + })); + } + @Disabled("JAMES-3056 returning two mailboxes with same name and id") @Test void renameThenFailToDeleteMailboxPathShouldBeConsistentWhenFindAll(CassandraCluster cassandra) { diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java index 2d3437a9a9..8179fc883e 100644 --- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/task/SolveMailboxInconsistenciesServiceTest.java @@ -154,12 +154,13 @@ class SolveMailboxInconsistenciesServiceTest { } @Test - void fixMailboxInconsistenciesShouldReturnPartialWhenDAOMisMatchOnPath() { + void fixMailboxInconsistenciesShouldReturnCompletedWhenDAOMisMatchOnPath() { + // Same mailbox id referenced by two paths: auto-resolvable without data loss. mailboxDAO.save(MAILBOX).block(); mailboxPathV3DAO.save(MAILBOX_NEW_PATH).block(); assertThat(testee.fixMailboxInconsistencies(new Context()).block()) - .isEqualTo(Result.PARTIAL); + .isEqualTo(Result.COMPLETED); } @Test @@ -248,14 +249,14 @@ class SolveMailboxInconsistenciesServiceTest { testee.fixMailboxInconsistencies(context).block(); + // The orphan mailbox pass re-registers the projection path, then the same-id conflict pass + // keeps the most recent path and drops the stale one: two fixes, no conflicting entry. assertThat(context.snapshot()) .isEqualTo(Context.builder() .processedMailboxEntries(1) .processedMailboxPathEntries(2) .addFixedInconsistencies(CASSANDRA_ID_1) - .addConflictingEntry(ConflictingEntry.builder() - .mailboxDaoEntry(MAILBOX) - .mailboxPathDaoEntry(MAILBOX_NEW_PATH)) + .addFixedInconsistencies(CASSANDRA_ID_1) .build() .snapshot()); } @@ -335,8 +336,8 @@ class SolveMailboxInconsistenciesServiceTest { @Test void fixMailboxInconsistenciesShouldAlterStateWhenDaoMisMatchOnPath() { - // Note that CASSANDRA_ID_1 becomes usable - // However in order to avoid data loss, merging CASSANDRA_ID_1 and CASSANDRA_ID_2 is still required + // Same mailbox id referenced by two paths (a partial rename leftover): the solver keeps the + // most recently written path and drops the stale one, leaving a single consistent registration. mailboxDAO.save(MAILBOX).block(); mailboxPathV3DAO.save(MAILBOX_NEW_PATH).block(); @@ -346,9 +347,27 @@ class SolveMailboxInconsistenciesServiceTest { softly.assertThat(mailboxDAO.retrieveAllMailboxes().collectList().block()) .containsExactlyInAnyOrder(MAILBOX); softly.assertThat(mailboxPathV3DAO.listAll().collectList().block()) - .containsExactlyInAnyOrder( - MAILBOX_NEW_PATH, - MAILBOX); + .containsExactlyInAnyOrder(MAILBOX); + }); + } + + @Test + void fixMailboxInconsistenciesShouldKeepMostRecentPathWhenSameMailboxRegisteredUnderTwoPaths() { + // Real partial rename leftover: the projection points to the new path, and both the old and + // the new path are still registered to the same id. The most recently written path (the + // rename target) wins and the stale old path reference is dropped. + mailboxDAO.save(MAILBOX_NEW_PATH).block(); + mailboxPathV3DAO.save(MAILBOX).block(); + mailboxPathV3DAO.save(MAILBOX_NEW_PATH).block(); + + Result result = testee.fixMailboxInconsistencies(new Context()).block(); + + SoftAssertions.assertSoftly(softly -> { + softly.assertThat(result).isEqualTo(Result.COMPLETED); + softly.assertThat(mailboxDAO.retrieveAllMailboxes().collectList().block()) + .containsExactlyInAnyOrder(MAILBOX_NEW_PATH); + softly.assertThat(mailboxPathV3DAO.listAll().collectList().block()) + .containsExactlyInAnyOrder(MAILBOX_NEW_PATH); }); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
