From 2c45f501da418814d3027ae0ff77a90f68b81603 Mon Sep 17 00:00:00 2001 From: davidfrigolet Date: Mon, 12 Oct 2026 16:00:54 +0100 Subject: [PATCH] feat(mongodb-reactive): add audit compaction capability - Implements AuditCompactor for the MongoDB reactive audit store, mirroring the MongoDB sync implementation (#967): no rekey needed since the current and ledger forms share the same auto _id, filtered by changeId, so compaction only deletes superseded documents. - Survivor is addressed by its stored _id (never by attribute matching), avoiding the bug class that bit the original sync attempt - a document's physical identity is tracked alongside its mapped AuditEntry via an IdentityHashMap, since AuditEntry declares no equals(). - Bridges the reactive driver's Publisher-based find()/deleteMany() to the synchronous AuditCompactor#compact() contract via the existing PublisherSync idiom already used throughout this module. - Adds unit tests against a mocked reactive MongoCollection (Publisher driven via a small hand-rolled Subscriber) and the shared AuditCompactionConformance suite wired against a real MongoDB container, plus store-specific assertions that the surviving document keeps its original _id. --- .../reactive/MongoDBReactiveAuditStore.java | 18 +- .../MongoDBReactiveAuditCompactor.java | 155 +++++++++ .../MongoDBReactiveAuditRepository.java | 15 + ...eactiveAuditCompactionConformanceTest.java | 311 ++++++++++++++++++ .../MongoDBReactiveAuditCompactorTest.java | 307 +++++++++++++++++ 5 files changed, 794 insertions(+), 12 deletions(-) create mode 100644 community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactor.java create mode 100644 community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditCompactionConformanceTest.java create mode 100644 community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactorTest.java diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java index 33f840ba4..22f986fc6 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java @@ -240,19 +240,13 @@ public Set> getNonGuardedTypes() { return new HashSet<>(Collections.singletonList(ClientSession.class)); } - /** - * Not implemented yet — audit compaction lands for this store in its own change. - *

- * A temporary stub so that adding the mandatory {@code getAuditCompactor()} to - * {@link CommunityAuditStore} does not break the build before each store has been done. It throws - * rather than silently doing nothing: a no-op would leave the audit store in ledger shape while the - * current-state write path is already active, which surfaces much later and much more confusingly. - */ @Override public AuditCompactor getAuditCompactor() { - return () -> { - throw new UnsupportedOperationException( - "Audit compaction is not yet implemented for " + getId()); - }; + if (auditRepository == null) { + throw new IllegalStateException("AuditStore not initialized - call initialize first"); + } + // No lazy initialize call, unlike the DynamoDB store: the repository builds its collection handle + // in its constructor, and compaction only deletes, so it needs no index to exist. + return auditRepository.getAuditCompactor(); } } diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactor.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactor.java new file mode 100644 index 000000000..423aa6621 --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactor.java @@ -0,0 +1,155 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed 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 io.flamingock.store.mongodb.reactive.internal; + +import com.mongodb.client.model.Filters; +import com.mongodb.client.result.DeleteResult; +import com.mongodb.reactivestreams.client.MongoCollection; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditSnapshotBuilder; +import io.flamingock.internal.common.mongodb.MongoDBAuditMapper; +import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper; +import io.flamingock.internal.core.external.store.AuditCompactor; +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.log.FlamingockLoggerFactory; +import io.flamingock.reactive.util.PublisherSync; +import org.bson.Document; +import org.bson.conversions.Bson; +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_CHANGE_ID; + +/** + * Collapses the MongoDB Reactive Streams audit collection to one document per change. + *

+ * Identical in shape and guarantees to {@code MongoDBSyncAuditCompactor} — same driver family, same + * document model, same auto {@code _id} that compaction never needs to change. The only difference is + * plumbing: the reactive driver returns {@code Publisher}s, and {@code compact()} is synchronous by + * contract, so every call is bridged with {@link PublisherSync}, the same idiom + * {@code MongoDBReactiveAuditRepository} already uses for its own writes and reads. + *

+ * Nothing is ever written. See {@code MongoDBSyncAuditCompactor}'s javadoc for the full reasoning on why + * the survivor is identified by {@code _id} rather than by attributes. + */ +class MongoDBReactiveAuditCompactor implements AuditCompactor { + + private static final Logger logger = FlamingockLoggerFactory.getLogger("MongoDBReactiveAuditCompactor"); + + /** MongoDB's document primary key. Not in {@code AuditEntryFieldConstants} — it is the driver's, not ours. */ + static final String KEY_ID = "_id"; + + private final MongoCollection collection; + private final MongoDBAuditMapper mapper; + + MongoDBReactiveAuditCompactor(MongoCollection collection, + MongoDBAuditMapper mapper) { + this.collection = Objects.requireNonNull(collection, "collection must not be null"); + this.mapper = Objects.requireNonNull(mapper, "mapper must not be null"); + } + + @Override + public Result compact() { + try { + // One pass, reading the whole collection before anything is deleted. Keeping each document's + // _id alongside the entry it maps to is the entire reason this reads documents rather than + // reusing AuditReader.getAuditSnapshotByChangeId() - see MongoDBSyncAuditCompactor's javadoc. + Map idOf = new IdentityHashMap<>(); + Map> byChange = new LinkedHashMap<>(); + for (Document document : PublisherSync.collect(collection.find())) { + AuditEntry entry = mapper.fromDocument(new MongoDBDocumentHelper(document)); + idOf.put(entry, document.get(KEY_ID)); + List entries = byChange.get(entry.getChangeId()); + if (entries == null) { + entries = new ArrayList<>(); + byChange.put(entry.getChangeId(), entries); + } + entries.add(entry); + } + + long removed = 0; + for (Map.Entry> change : byChange.entrySet()) { + // Fail-fast: a failure on one change aborts the rest. Changes are independent and nothing + // is written, so whatever was already compacted stays compacted and a re-run finishes. + removed += compactChange(change.getKey(), change.getValue(), idOf); + } + + logger.debug("Audit compaction finished [changes={} documentsRemoved={}]", byChange.size(), removed); + return Result.OK(); + } catch (RuntimeException exception) { + // Reported rather than thrown, so the calling system change decides. Nothing can have been + // half-done damagingly: every change's survivor is excluded from every delete. + logger.warn("Audit compaction failed", exception); + return new Result.Error(exception); + } + } + + /** + * Removes every document of {@code changeId} except the one holding its current state. + * + * @return how many documents were removed + */ + private long compactChange(String changeId, List entries, Map idOf) { + if (entries.size() == 1) { + return 0; + } + + Object survivorId = idOf.get(selectSurvivor(changeId, entries)); + if (survivorId == null) { + throw new IllegalStateException("Could not resolve the stored document of the current state of" + + " changeId '" + changeId + "'"); + } + + // Both clauses matter. Without the changeId scope this would delete the rest of the collection. + Bson supersededForThisChange = Filters.and( + Filters.eq(KEY_CHANGE_ID, changeId), + Filters.ne(KEY_ID, survivorId)); + + DeleteResult result = PublisherSync.first(collection.deleteMany(supersededForThisChange)); + long removed = result == null ? 0 : result.getDeletedCount(); + logger.debug("Compacted change [changeId={} documentsRemoved={}]", changeId, removed); + return removed; + } + + /** + * Picks the entry holding the change's current effective state, via {@link AuditSnapshotBuilder} rather + * than a reimplementation of "latest wins" - which would resolve the equal-{@code createdAt} case + * arbitrarily instead of by status priority, and so could silently change which changes are considered + * applied. + *

+ * The builder deals in {@link AuditEntry}, but what is needed is the stored document's {@code _id}, so + * entries are tracked back to their documents by identity. Sound because the builder returns the very + * instances it was given and {@code AuditEntry} declares no {@code equals}. + */ + private AuditEntry selectSurvivor(String changeId, List entries) { + AuditSnapshotBuilder builder = new AuditSnapshotBuilder(); + for (AuditEntry entry : entries) { + builder.addEntry(entry); + } + List survivors = builder.buildList(); + if (survivors.size() != 1) { + throw new IllegalStateException("Expected a single effective state for changeId '" + changeId + + "' but the snapshot produced " + survivors.size()); + } + return survivors.get(0); + } +} diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java index 1c6579fca..ae1e503fe 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditRepository.java @@ -25,6 +25,7 @@ import com.mongodb.reactivestreams.client.MongoCollection; import com.mongodb.reactivestreams.client.MongoDatabase; import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.core.external.store.AuditCompactor; import io.flamingock.internal.common.mongodb.CollectionInitializator; import io.flamingock.internal.common.mongodb.MongoDBAuditMapper; import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper; @@ -154,6 +155,20 @@ Result append(AuditEntry auditEntry) { return Result.OK(); } + /** + * Returns the compactor bound to this repository's collection. + *

+ * Public because {@code MongoDBReactiveAuditStore} lives in the parent package, same as + * {@link #getAuditHistory()}. The returned implementation stays package-private; callers only ever see + * the {@link AuditCompactor} contract. + *

+ * Needs no prior {@link #initialize(boolean)}: the collection handle is built in the constructor, and + * compaction only deletes and addresses documents by {@code _id}, so it depends on no index existing. + */ + public AuditCompactor getAuditCompactor() { + return new MongoDBReactiveAuditCompactor(collection, mapper); + } + public List getAuditHistory() { return PublisherSync.collect(collection.find()) .stream() diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditCompactionConformanceTest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditCompactionConformanceTest.java new file mode 100644 index 000000000..c40d7145e --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditCompactionConformanceTest.java @@ -0,0 +1,311 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed 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 io.flamingock.store.mongodb.reactive; + +import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; +import com.mongodb.reactivestreams.client.MongoDatabase; +import io.flamingock.core.kit.audit.AuditEntryTestFactory; +import io.flamingock.core.kit.audit.compaction.AuditCompactionConformance; +import io.flamingock.core.kit.audit.compaction.AuditStorageCompactionFixture; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.context.Dependency; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.core.configuration.community.CommunityConfiguration; +import io.flamingock.internal.core.context.SimpleContext; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.Result; +import io.flamingock.internal.util.id.RunnerId; +import io.flamingock.mongodb.reactive.kit.MongoDBReactiveAuditStorage; +import io.flamingock.reactive.util.PublisherSync; +import io.flamingock.targetsystem.mongodb.reactive.MongoDBReactiveTargetSystem; +import org.bson.Document; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.MongoDBContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Audit compaction against a real MongoDB: the shared contract, plus the guarantees that are specific to + * this store. + *

+ * Mirrors {@code MongoDBSyncAuditCompactionConformanceTest} - same driver family, same document model, + * same reasoning. Reads here go through {@link PublisherSync} because the reactive driver returns + * {@code Publisher}s, but {@code compact()} itself is synchronous by contract either way. + *

+ * The call-sequence guarantees - that only deletes ever reach the collection, that the filter excludes the + * survivor, that a failure is reported rather than thrown - are covered Docker-free by + * {@code io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditCompactorTest}. + */ +@Testcontainers +class MongoDBReactiveAuditCompactionConformanceTest { + + private static final String DATABASE_NAME = "test"; + private static final String AUDIT_COLLECTION = "compactionAuditLog"; + private static final String STAGE_ID = "compaction-stage"; + private static final LocalDateTime T0 = LocalDateTime.of(2026, 1, 1, 12, 0, 0); + private static final LocalDateTime T1 = LocalDateTime.of(2026, 1, 1, 12, 5, 0); + + @Container + static final MongoDBContainer mongoDBContainer = + new MongoDBContainer(DockerImageName.parse("mongo:6")).withReuse(true); + + private MongoClient mongoClient; + private MongoDatabase database; + private MongoDBReactiveAuditStore auditStore; + private MongoDBReactiveAuditStorage auditStorage; + private AuditCompactionConformance conformance; + + @BeforeEach + void setUp() { + mongoClient = MongoClients.create(mongoDBContainer.getConnectionString()); + database = mongoClient.getDatabase(DATABASE_NAME); + + auditStore = storeFor(new MongoDBReactiveTargetSystem("mongodb", mongoClient, DATABASE_NAME)); + + // Forces the audit collection and its unique (executionId, changeId, state) index into existence, + // which happens lazily in MongoDBReactiveAuditPersistence.doInitialize rather than in the store's + // own initialize. Without this the collection would be created indexless by the test kit's raw + // inserts, the store would not behave as it does in production, and indexesAreUntouched below + // would compare _id_ with _id_ and prove nothing. + auditStore.getPersistenceFactory().get(STAGE_ID); + + auditStorage = new MongoDBReactiveAuditStorage(database, AUDIT_COLLECTION); + conformance = new AuditCompactionConformance( + new AuditStorageCompactionFixture(auditStore, auditStorage, STAGE_ID)); + } + + @AfterEach + void tearDown() { + // FeatureFlag is process-global and every test in this module shares the JVM. The current-state + // write property restores it itself, but leaving it enabled on an unexpected failure path would + // silently change how the next test's store behaves. + FeatureFlag.remove(Features.JOURNAL_EVENTS); + if (mongoClient != null) { + PublisherSync.complete(database.drop()); + mongoClient.close(); + } + } + + // --------------------------------------------------------------------------------------------- + // The shared contract + // --------------------------------------------------------------------------------------------- + + @Test + @DisplayName("seeding round-trips faithfully") + void seedingRoundTripsFaithfully() { + conformance.verifySeedingRoundTripsFaithfully(); + } + + @Test + @DisplayName("the audit snapshot is unchanged by compaction") + void snapshotPreserved() { + conformance.verifySnapshotPreserved(); + } + + @Test + @DisplayName("exactly one record survives per change, and it is the right one") + void oneRecordPerChange() { + conformance.verifyOneRecordPerChange(); + } + + @Test + @DisplayName("compaction is a fixpoint") + void isFixpoint() { + conformance.verifyIsFixpoint(); + } + + @Test + @DisplayName("the ordinary current-state write still works after compaction") + void subsequentCurrentStateWriteSucceeds() { + conformance.verifySubsequentCurrentStateWriteSucceeds(); + } + + @Test + @DisplayName("changes absent from the pipeline are preserved") + void unknownChangesPreserved() { + conformance.verifyUnknownChangesPreserved(); + } + + @Test + @DisplayName("a change that already has one record is left untouched") + void alreadySingleRecordUntouched() { + conformance.verifyAlreadySingleRecordUntouched(); + } + + @Test + @DisplayName("compacting an empty store is a no-op") + void emptyStoreIsNoOp() { + conformance.verifyEmptyStoreIsNoOp(); + } + + @Test + @DisplayName("an exact createdAt tie is broken by status priority") + void timestampTieBreaksOnStatusPriority() { + conformance.verifyTimestampTieBreaksOnStatusPriority(); + } + + @Test + @DisplayName("system-change entries keep their flag") + void systemChangeEntriesPreserved() { + conformance.verifySystemChangeEntriesPreserved(); + } + + @Test + @DisplayName("a ledger larger than one batch is fully compacted") + void largeLedgerIsCompacted() { + conformance.verifyLargeLedgerIsCompacted(300, 3); + } + + // --------------------------------------------------------------------------------------------- + // MongoDB-specific + // --------------------------------------------------------------------------------------------- + + @Test + @DisplayName("the surviving document is never rewritten — it keeps its original _id") + void survivorKeepsItsOriginalId() { + seedLedger("change-a"); + Object survivorIdBefore = idOf("change-a", AuditEntry.Status.APPLIED); + assertNotNull(survivorIdBefore, "the document that should survive was not seeded"); + + requireCompacted(); + + List remaining = rawDocumentsFor("change-a"); + assertEquals(1, remaining.size(), "expected one surviving document, found " + remaining.size()); + assertEquals(survivorIdBefore, remaining.get(0).get("_id"), + "the survivor must be the original document, untouched — a changed _id means compaction" + + " rewrote it instead of merely deleting its siblings"); + } + + @Test + @DisplayName("compaction leaves the collection's indexes untouched") + void indexesAreUntouched() { + seedLedger("change-a"); + List before = indexNames(); + assertTrue(before.size() > 1, + "setup must have created the unique index, otherwise this test proves nothing: " + before); + + requireCompacted(); + + assertEquals(before, indexNames(), "compaction must not create, drop or alter any index"); + } + + @Test + @DisplayName("the current-state write works after compaction with transactions disabled") + void currentStateWriteWorksAfterCompactionWithTransactionsDisabled() { + // A dimension DynamoDB does not have. With transactions off the persistence takes the + // non-transactional save(AuditEntry) overload, which carries the same one-document-per-change + // expectation as the transactional one. The shared property only exercises the default, enabled path. + MongoDBReactiveAuditStore nonTransactionalStore = storeFor( + new MongoDBReactiveTargetSystem("mongodb", mongoClient, DATABASE_NAME) + .withTransactionsSupported(false)); + seedLedger("change-a"); + requireCompacted(nonTransactionalStore); + + FeatureFlag.enable(Features.JOURNAL_EVENTS); + try { + Result written = nonTransactionalStore.getPersistenceFactory().get(STAGE_ID) + .writeEntry(AuditEntryTestFactory.createDeterministicAuditEntry( + "exec-after", "change-a", AuditEntry.Status.APPLIED, T1.plusMinutes(5), false)); + assertTrue(!written.isError(), "the non-transactional current-state write must still work"); + assertEquals(1, rawDocumentsFor("change-a").size(), + "the write must have replaced the single document, not added a second"); + } finally { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + } + } + + @Test + @DisplayName("asking an uninitialized store for a compactor fails loudly") + void getAuditCompactorBeforeInitializeThrows() { + MongoDBReactiveAuditStore uninitialized = MongoDBReactiveAuditStore.from( + new MongoDBReactiveTargetSystem("mongodb", mongoClient, DATABASE_NAME)); + + assertThrows(IllegalStateException.class, uninitialized::getAuditCompactor, + "contract clause 13: the compactor is only available after initialize"); + } + + // --------------------------------------------------------------------------------------------- + // Helpers + // --------------------------------------------------------------------------------------------- + + private MongoDBReactiveAuditStore storeFor(MongoDBReactiveTargetSystem targetSystem) { + SimpleContext context = new SimpleContext(); + context.addDependency(new Dependency(RunnerId.class, RunnerId.generate("compaction-conformance"))); + context.addDependency(new Dependency(CommunityConfiguration.class, new CommunityConfiguration())); + targetSystem.initialize(context); + + MongoDBReactiveAuditStore store = MongoDBReactiveAuditStore.from(targetSystem) + .withAuditRepositoryName(AUDIT_COLLECTION) + .withLockRepositoryName("compactionLock") + .withJournalRepositoryName("compactionJournal"); + store.initialize(context); + return store; + } + + /** Seeds a ledger-shaped history for one change: STARTED then APPLIED, same execution. */ + private void seedLedger(String changeId) { + auditStorage.addAuditEntry(AuditEntryTestFactory.createDeterministicAuditEntry( + "exec-1", changeId, AuditEntry.Status.STARTED, T0, false)); + auditStorage.addAuditEntry(AuditEntryTestFactory.createDeterministicAuditEntry( + "exec-1", changeId, AuditEntry.Status.APPLIED, T1, false)); + } + + private void requireCompacted() { + requireCompacted(auditStore); + } + + private void requireCompacted(MongoDBReactiveAuditStore store) { + Result result = store.getAuditCompactor().compact(); + if (result.isError()) { + throw new AssertionError("compact() failed: " + ((Result.Error) result).getError()); + } + } + + /** Raw documents, so {@code _id} is visible — the shared fixture maps it away. */ + private List rawDocumentsFor(String changeId) { + return PublisherSync.collect( + database.getCollection(AUDIT_COLLECTION).find(new Document("changeId", changeId))); + } + + private Object idOf(String changeId, AuditEntry.Status state) { + Document found = PublisherSync.first( + database.getCollection(AUDIT_COLLECTION) + .find(new Document("changeId", changeId).append("state", state.name()))); + return found == null ? null : found.get("_id"); + } + + private List indexNames() { + List names = new java.util.ArrayList<>(); + for (Document index : PublisherSync.collect(database.getCollection(AUDIT_COLLECTION).listIndexes())) { + names.add(index.getString("name")); + } + java.util.Collections.sort(names); + return names; + } +} diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactorTest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactorTest.java new file mode 100644 index 000000000..d6723e11f --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditCompactorTest.java @@ -0,0 +1,307 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed 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 io.flamingock.store.mongodb.reactive.internal; + +import com.mongodb.MongoClientSettings; +import com.mongodb.client.result.DeleteResult; +import com.mongodb.reactivestreams.client.FindPublisher; +import com.mongodb.reactivestreams.client.MongoCollection; +import io.flamingock.api.RecoveryStrategy; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditTxType; +import io.flamingock.internal.common.mongodb.MongoDBAuditMapper; +import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper; +import io.flamingock.internal.util.Result; +import org.bson.BsonDocument; +import org.bson.Document; +import org.bson.conversions.Bson; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +import java.time.LocalDateTime; +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Unit coverage for {@link MongoDBReactiveAuditCompactor} against a mocked reactive {@code MongoCollection}. + *

+ * Mirrors {@code MongoDBSyncAuditCompactorTest} test-for-test — same driver family, same document model, + * same reasoning for why the survivor is addressed by {@code _id} rather than by attributes. The only + * difference is that {@code find()} and {@code deleteMany()} return {@code Publisher}s here, driven + * synchronously in these tests via a hand-rolled {@link Publisher} rather than a reactive test library, to + * keep this module's test dependencies unchanged. + *

+ * The end state these tests imply is covered against a real MongoDB by + * {@code io.flamingock.store.mongodb.reactive.MongoDBReactiveAuditCompactionConformanceTest}. + */ +class MongoDBReactiveAuditCompactorTest { + + private static final String CHANGE_A = "change-a"; + private static final String CHANGE_B = "change-b"; + private static final LocalDateTime T0 = LocalDateTime.of(2026, 1, 1, 12, 0, 0); + private static final LocalDateTime T1 = LocalDateTime.of(2026, 1, 1, 12, 5, 0); + + private final MongoDBAuditMapper mapper = + new MongoDBAuditMapper<>(() -> new MongoDBDocumentHelper(new Document())); + + @Test + @DisplayName("one delete per change that has superseded documents") + void deletesOneBatchPerChangeThatNeedsIt() { + MongoCollection collection = collectionHolding( + stored("id-a1", CHANGE_A, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-a2", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T1), + stored("id-b1", CHANGE_B, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-b2", CHANGE_B, "exec-1", AuditEntry.Status.APPLIED, T1)); + + assertOk(new MongoDBReactiveAuditCompactor(collection, mapper).compact()); + + Mockito.verify(collection, Mockito.times(2)).deleteMany(Mockito.any(Bson.class)); + } + + @Test + @DisplayName("nothing is ever written — only the read and the deletes reach the collection") + void neverWritesAnything() { + MongoCollection collection = collectionHolding( + stored("id-a1", CHANGE_A, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-a2", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T1)); + + assertOk(new MongoDBReactiveAuditCompactor(collection, mapper).compact()); + + // Proves three contract clauses by construction rather than by inspecting state afterwards: the + // survivor is preserved verbatim because it is never rewritten, no index or schema is altered, and + // no journal event is emitted. There is no other call the compactor can make. + Mockito.verify(collection).find(); + Mockito.verify(collection).deleteMany(Mockito.any(Bson.class)); + Mockito.verifyNoMoreInteractions(collection); + } + + @Test + @DisplayName("the delete filter is scoped to the change and excludes the survivor by _id") + void theDeleteFilterIsScopedAndExcludesTheSurvivor() { + MongoCollection collection = collectionHolding( + stored("id-superseded", CHANGE_A, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-survivor", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T1)); + + assertOk(new MongoDBReactiveAuditCompactor(collection, mapper).compact()); + + String rendered = capturedFilter(collection).toJson(); + // Scope: without it, `_id != survivor` would match almost the whole collection. + assertTrue(rendered.contains("\"changeId\"") && rendered.contains(CHANGE_A), + "the filter must be scoped to the change: " + rendered); + // Exclusion: by identity, so there is exactly one predicate and no attribute combination to get + // wrong. `id-survivor` must be the one spared, not the superseded document. + assertTrue(rendered.contains("$ne") && rendered.contains("id-survivor"), + "the survivor must be excluded by _id: " + rendered); + assertTrue(!rendered.contains("id-superseded"), + "the superseded document must not appear in the exclusion: " + rendered); + } + + @Test + @DisplayName("on an exact createdAt tie the higher status priority survives") + void survivorIsChosenByStatusPriorityOnTie() { + MongoCollection collection = collectionHolding( + stored("id-applied", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T0), + stored("id-rolled-back", CHANGE_A, "exec-1", AuditEntry.Status.ROLLED_BACK, T0)); + + assertOk(new MongoDBReactiveAuditCompactor(collection, mapper).compact()); + + String rendered = capturedFilter(collection).toJson(); + assertTrue(rendered.contains("id-rolled-back"), + "ROLLED_BACK outranks APPLIED on an equal createdAt; selecting on createdAt alone would" + + " resolve this arbitrarily: " + rendered); + } + + @Test + @DisplayName("a change that already has one document is left alone") + void singleDocumentChangeIsLeftAlone() { + MongoCollection collection = collectionHolding( + stored("id-a1", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T0)); + + assertOk(new MongoDBReactiveAuditCompactor(collection, mapper).compact()); + + Mockito.verify(collection).find(); + Mockito.verify(collection, Mockito.never()).deleteMany(Mockito.any(Bson.class)); + } + + @Test + @DisplayName("an empty collection is left completely alone") + void emptyCollectionDoesNothing() { + MongoCollection collection = collectionHolding(); + + assertOk(new MongoDBReactiveAuditCompactor(collection, mapper).compact()); + + Mockito.verify(collection).find(); + Mockito.verifyNoMoreInteractions(collection); + } + + @Test + @DisplayName("a failed delete is reported, not thrown") + void failureIsReportedNotThrown() { + MongoCollection collection = collectionHolding( + stored("id-a1", CHANGE_A, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-a2", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T1)); + Publisher failingDelete = errorPublisher(new IllegalStateException("delete failed")); + Mockito.when(collection.deleteMany(Mockito.any(Bson.class))).thenReturn(failingDelete); + + assertTrue(new MongoDBReactiveAuditCompactor(collection, mapper).compact().isError(), + "the caller decides what a failed compaction means"); + } + + @Test + @DisplayName("the first failing change stops the rest") + void failFastStopsAtTheFirstFailingChange() { + MongoCollection collection = collectionHolding( + stored("id-a1", CHANGE_A, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-a2", CHANGE_A, "exec-1", AuditEntry.Status.APPLIED, T1), + stored("id-b1", CHANGE_B, "exec-1", AuditEntry.Status.STARTED, T0), + stored("id-b2", CHANGE_B, "exec-1", AuditEntry.Status.APPLIED, T1)); + Publisher failingDelete = errorPublisher(new IllegalStateException("delete failed")); + Mockito.when(collection.deleteMany(Mockito.any(Bson.class))).thenReturn(failingDelete); + + assertTrue(new MongoDBReactiveAuditCompactor(collection, mapper).compact().isError()); + + // One attempt only. Nothing was written, so anything already compacted stays compacted and a + // re-run finishes the job. + Mockito.verify(collection, Mockito.times(1)).deleteMany(Mockito.any(Bson.class)); + } + + @Test + @DisplayName("a failed read is reported without deleting anything") + void readFailureIsReportedWithoutDeleting() { + @SuppressWarnings("unchecked") + MongoCollection collection = Mockito.mock(MongoCollection.class); + Mockito.when(collection.find()).thenThrow(new IllegalStateException("read failed")); + + assertTrue(new MongoDBReactiveAuditCompactor(collection, mapper).compact().isError()); + + // Relevant because a document missing its state makes the mapping or aggregation throw. Failing + // before any delete is what makes that safe rather than destructive. + Mockito.verify(collection, Mockito.never()).deleteMany(Mockito.any(Bson.class)); + } + + // --------------------------------------------------------------------------------------------- + // Helpers + // --------------------------------------------------------------------------------------------- + + @SuppressWarnings("unchecked") + private static MongoCollection collectionHolding(Document... documents) { + MongoCollection collection = Mockito.mock(MongoCollection.class); + FindPublisher findPublisher = Mockito.mock(FindPublisher.class); + driveOnSubscribe(findPublisher, Arrays.asList(documents), null); + Mockito.when(collection.find()).thenReturn(findPublisher); + Mockito.when(collection.deleteMany(Mockito.any(Bson.class))) + .thenAnswer(invocation -> resultPublisher(DeleteResult.acknowledged(1))); + return collection; + } + + /** A {@code Publisher} mock that replays {@code items} (or fails with {@code error}) on subscribe. */ + @SuppressWarnings("unchecked") + private static void driveOnSubscribe(Publisher publisherMock, List items, Throwable error) { + Mockito.doAnswer(invocation -> { + Subscriber subscriber = invocation.getArgument(0); + subscriber.onSubscribe(new Subscription() { + @Override + public void request(long n) { + } + + @Override + public void cancel() { + } + }); + if (error != null) { + subscriber.onError(error); + } else { + for (T item : items) { + subscriber.onNext(item); + } + subscriber.onComplete(); + } + return null; + }).when(publisherMock).subscribe(Mockito.any(Subscriber.class)); + } + + private static Publisher resultPublisher(T result) { + @SuppressWarnings("unchecked") + Publisher publisher = Mockito.mock(Publisher.class); + driveOnSubscribe(publisher, java.util.Collections.singletonList(result), null); + return publisher; + } + + private static Publisher errorPublisher(Throwable error) { + @SuppressWarnings("unchecked") + Publisher publisher = Mockito.mock(Publisher.class); + driveOnSubscribe(publisher, java.util.Collections.emptyList(), error); + return publisher; + } + + /** + * Renders the captured delete filter to BSON. + *

+ * The driver here is 4.0.0, where {@code Bson} has only the two-argument + * {@code toBsonDocument(Class, CodecRegistry)} — the no-argument overload arrived in 4.2. + */ + private static BsonDocument capturedFilter(MongoCollection collection) { + ArgumentCaptor captor = ArgumentCaptor.forClass(Bson.class); + Mockito.verify(collection).deleteMany(captor.capture()); + return captor.getValue() + .toBsonDocument(Document.class, MongoClientSettings.getDefaultCodecRegistry()); + } + + /** A stored audit document, with the {@code _id} the real collection would have assigned. */ + private Document stored(String id, + String changeId, + String executionId, + AuditEntry.Status status, + LocalDateTime createdAt) { + Document document = mapper.toDocument(new AuditEntry( + executionId, + "test-stage", + changeId, + "test-author", + createdAt, + status, + AuditEntry.ChangeType.STANDARD_CODE, + "TestChangeClass", + "testMethod", + "TestSourceFile", + 0L, + "localhost", + null, + false, + null, + AuditTxType.NON_TX, + "test-target-system", + "001", + RecoveryStrategy.MANUAL_INTERVENTION, + null)).getDocument(); + document.put(MongoDBReactiveAuditCompactor.KEY_ID, id); + return document; + } + + private static void assertOk(Result result) { + if (result.isError()) { + throw new AssertionError("compact() failed: " + ((Result.Error) result).getError()); + } + } +}