From a754cd8d07c5fea874c0d927933a5aba33147757 Mon Sep 17 00:00:00 2001 From: bercianor Date: Mon, 12 Oct 2026 15:48:46 +0200 Subject: [PATCH] feat(mongodb-reactive): add independent history appenders --- .../MongoDBReactiveAuditPersistence.java | 69 +++++- ...DBReactiveAuditPersistenceJournalTest.java | 6 +- .../MongoDBReactiveHistoryAppenderTest.java | 206 ++++++++++++++++++ 3 files changed, 270 insertions(+), 11 deletions(-) create mode 100644 community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveHistoryAppenderTest.java diff --git a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java index eed5d1072..4cd190d9a 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Objects; +import java.util.UUID; public class MongoDBReactiveAuditPersistence extends AbstractCommunityAuditPersistence { @@ -99,10 +100,6 @@ public Result writeEntry(AuditEntry auditEntry) { return auditRepository.append(auditEntry); } - if (journalEventStore == null || journalEventSequencer == null) { - throw new IllegalStateException("MongoDB reactive journal writes require a sequencer"); - } - if (!supportsTransactions) { return writeJournalAndAuditWithoutTransaction(auditEntry); } @@ -114,9 +111,12 @@ private Result writeJournalAndAuditInTransaction(AuditEntry auditEntry) { RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId()); Result result = txWrapper.wrapExecution(baseContext, runtimeContext -> { ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class); - JournalEvent journalEvent = journalEventSequencer.newEvent(auditEntry); - journalEventStore.append(clientSession, journalEvent); - return auditRepository.save(clientSession, auditEntry); + appendEventFrom(auditEntry, clientSession); + Result auditResult = auditRepository.save(clientSession, auditEntry); + if (auditResult.isError()) { + throw new IllegalStateException("MongoDB history write failed", ((Result.Error) auditResult).getError()); + } + return auditResult; }); // Result cannot represent FailedStep. The transaction wrapper has therefore committed successfully @@ -131,9 +131,60 @@ private Result writeJournalAndAuditInTransaction(AuditEntry auditEntry) { * audit state without its corresponding historical event. */ private Result writeJournalAndAuditWithoutTransaction(AuditEntry auditEntry) { - JournalEvent journalEvent = journalEventSequencer.newEvent(auditEntry); - journalEventStore.append(journalEvent); + appendEventFrom(auditEntry, null); journalEventSequencer.confirm(); return auditRepository.save(auditEntry); } + + /** + * Preserves supplied historical audit entries independently of the journal feature flag. + * + * @param entry original historical entry + * @return repository append result + */ + @Override + public Result append(AuditEntry entry) { + return auditRepository.append(entry); + } + + /** + * Durably appends only a CHANGE_STATE event to this persistence stream. Transactional deployments use + * an owned boundary independent of the ambient business transaction; standalone deployments use + * a completed journal insert. No audit projection is written. + * + * @param payload original historical entry + * @return durable journal write result + */ + @Override + public Result appendEventFrom(AuditEntry payload) { + + Result result; + if (supportsTransactions) { + RuntimeContext context = new BasicRuntimeContext("append-journal-" + UUID.randomUUID()); + result = txWrapper.wrapExecution(context, runtimeContext -> { + ClientSession session = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class); + return appendEventFrom(payload, session); + }); + } else { + result = appendEventFrom(payload, null); + } + journalEventSequencer.confirm(); + return result; + } + + private Result appendEventFrom(AuditEntry payload, ClientSession session) { + requireJournalEnabled(); + JournalEvent event = journalEventSequencer.newEvent(payload); + Result result = session == null ? journalEventStore.append(event) : journalEventStore.append(session, event); + if (result.isError()) { + throw new IllegalStateException("MongoDB history write failed", ((Result.Error) result).getError()); + } + return result; + } + + private static void requireJournalEnabled() { + if (!FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) { + throw new IllegalStateException("Journal history requires JOURNAL_EVENTS to be enabled"); + } + } } diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java index 94aa63ccf..59a19a2ae 100644 --- a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceJournalTest.java @@ -299,7 +299,8 @@ private MongoDBReactiveAuditPersistence persistenceFor(MongoDBReactiveAuditRepos private MongoDBReactiveAuditPersistence persistenceFor(MongoDBReactiveAuditRepository repository, JournalEventSequencer sequencer) { MongoDBReactiveAuditPersistence persistence = new MongoDBReactiveAuditPersistence( - new CommunityConfiguration(), repository, journalEventStore, sequencer, true, txWrapper, true); + new CommunityConfiguration(), repository, journalEventStore, sequencer, + true, txWrapper, true); persistence.initialize(RunnerId.generate()); return persistence; } @@ -309,7 +310,8 @@ private MongoDBReactiveAuditPersistence persistenceWithoutTransactions( MongoDBReactiveJournalEventStore journalStore, JournalEventSequencer sequencer) { MongoDBReactiveAuditPersistence persistence = new MongoDBReactiveAuditPersistence( - new CommunityConfiguration(), repository, journalStore, sequencer, false, null, true); + new CommunityConfiguration(), repository, journalStore, sequencer, + false, null, true); persistence.initialize(RunnerId.generate()); return persistence; } diff --git a/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveHistoryAppenderTest.java b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveHistoryAppenderTest.java new file mode 100644 index 000000000..e3b6ea2b8 --- /dev/null +++ b/community/flamingock-mongodb-reactive-auditstore/src/test/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveHistoryAppenderTest.java @@ -0,0 +1,206 @@ +/* + * 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.reactivestreams.client.ClientSession; +import io.flamingock.core.kit.audit.AuditEntryTestFactory; +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditTxType; +import io.flamingock.internal.common.core.context.RuntimeContext; +import io.flamingock.internal.common.core.external.ExecutionWrapper; +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; +import io.flamingock.internal.core.configuration.community.CommunityConfiguration; +import io.flamingock.internal.core.journal.JournalEventSequencerFactory; +import io.flamingock.internal.util.FeatureFlag; +import io.flamingock.internal.util.Result; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class MongoDBReactiveHistoryAppenderTest { + private final MongoDBReactiveAuditRepository audit = mock(MongoDBReactiveAuditRepository.class); + private final MongoDBReactiveJournalEventStore journal = mock(MongoDBReactiveJournalEventStore.class); + private final JournalEventSequencerFactory factory = new JournalEventSequencerFactory(journal); + private final AuditEntry entry = AuditEntryTestFactory.createTestAuditEntry( + "change", AuditEntry.Status.APPLIED, AuditTxType.NON_TX, (Class) null); + private final List> durable = new ArrayList<>(); + private JournalEvent staged; + private boolean failCommit; + private int commits; + private Result contribution = Result.OK(); + private final ClientSession session = mock(ClientSession.class); + private final ExecutionWrapper wrapper = new ExecutionWrapper() { + @Override + public R wrapExecution(C context, Function operation) { + context.addDependency(session); + staged = null; + R result = operation.apply(context); + if (failCommit) { + throw new IllegalStateException("commit"); + } + durable.add(staged); + commits++; + return result; + } + }; + + @AfterEach + void resetFlag() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + } + + @Test + void historicalAppendPreservesEntryWithFlagOffAndOn() { + MongoDBReactiveAuditPersistence persistence = persistence(false); + when(audit.append(entry)).thenReturn(Result.OK()); + assertFalse(persistence.append(entry).isError()); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + assertFalse(persistence.append(entry).isError()); + verify(audit, org.mockito.Mockito.times(2)).append(entry); + verifyNoInteractions(journal); + assertEquals(0, commits); + } + + @Test + void disabledJournalRejectsBeforeStorageAccess() { + MongoDBReactiveAuditPersistence persistence = persistence(true); + assertEquals("Journal history requires JOURNAL_EVENTS to be enabled", + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)).getMessage()); + verifyNoInteractions(journal, audit); + assertEquals(0, commits); + assertTrue(durable.isEmpty()); + } + + @Test + void sourceEventUsesPersistenceStreamAndOnlyCommitsJournal() { + MongoDBReactiveAuditPersistence persistence = persistence(true); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(1, commits); + assertEvent(durable.get(0), "normal-stage", 1L); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEvent(durable.get(1), "normal-stage", 2L); + verifyNoInteractions(audit); + } + + @Test + void sourceNonTransactionalWriteIsDurableWithoutWrapper() { + MongoDBReactiveAuditPersistence persistence = persistence(false); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEvent(durable.get(0), "normal-stage", 1L); + assertEquals(0, commits); + verifyNoInteractions(audit); + } + + @Test + void failedContributionAndCommitDoNotSpendSequence() { + MongoDBReactiveAuditPersistence persistence = persistence(true); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + contribution = new Result.Error(new IllegalStateException("insert")); + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)); + assertEquals(0, commits); + assertTrue(durable.isEmpty()); + contribution = Result.OK(); + failCommit = true; + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)); + assertEquals(0, commits); + assertTrue(durable.isEmpty()); + failCommit = false; + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEvent(durable.get(0), "normal-stage", 1L); + verifyNoInteractions(audit); + } + + @Test + void nonTransactionalFailureDoesNotSpendSequence() { + MongoDBReactiveAuditPersistence persistence = persistence(false); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + contribution = new Result.Error(new IllegalStateException("insert")); + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)); + assertTrue(durable.isEmpty()); + contribution = Result.OK(); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEvent(durable.get(0), "normal-stage", 1L); + verifyNoInteractions(audit); + } + + @Test + void normalJointWriteRetainsStageAndAbortsOnAuditError() { + MongoDBReactiveAuditPersistence persistence = persistence(true); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + when(audit.save(session, entry)).thenReturn(new Result.Error(new IllegalStateException("audit"))); + assertThrows(IllegalStateException.class, () -> persistence.writeEntry(entry)); + assertEquals(0, commits); + assertTrue(durable.isEmpty()); + when(audit.save(session, entry)).thenReturn(Result.OK()); + assertFalse(persistence.writeEntry(entry).isError()); + assertEvent(durable.get(0), "normal-stage", 1L); + assertEquals(1, commits); + } + + private void assertEvent(JournalEvent event, String stream, long sequence) { + assertSame(entry, event.getData()); + assertEquals(stream, event.getStreamId()); + assertEquals(sequence, event.getStreamSequence()); + assertEquals(JournalEventType.CHANGE_STATE, event.getEventType()); + assertEquals(JournalEvent.DEFAULT_VERSION, event.getEventVersion()); + assertFalse(event.isAcknowledged()); + assertFalse(event.getEventId().isEmpty()); + assertFalse(event.getIdempotencyKey().isEmpty()); + } + + private MongoDBReactiveAuditPersistence persistence(boolean transactional) { + when(journal.getLastEventByStream(anyString())).thenAnswer(invocation -> { + String stream = invocation.getArgument(0); + return durable.stream().filter(event -> stream.equals(event.getStreamId())).reduce((a, b) -> b); + }); + when(journal.append(any(ClientSession.class), any())).thenAnswer(invocation -> { + assertSame(session, invocation.getArgument(0)); + staged = invocation.getArgument(1); + return contribution; + }); + when(journal.append(any(JournalEvent.class))).thenAnswer(invocation -> { + if (!contribution.isError()) { + durable.add(invocation.getArgument(0)); + } + return contribution; + }); + MongoDBReactiveAuditPersistence persistence = new MongoDBReactiveAuditPersistence( + new CommunityConfiguration(), audit, journal, factory.forStream("normal-stage"), + transactional, transactional ? wrapper : null, true); + clearInvocations(journal); + return persistence; + } +}