From 76d084690c20bb363cdff7996681ec06b78f0efd Mon Sep 17 00:00:00 2001 From: bercianor Date: Mon, 12 Oct 2026 15:48:46 +0200 Subject: [PATCH] feat(mongodb-sync): add independent history appenders --- .../internal/MongoDBSyncAuditPersistence.java | 61 +++- ...ongoDBSyncAuditPersistenceJournalTest.java | 115 +++++++- .../MongoDBSyncHistoryAppenderTest.java | 278 ++++++++++++++++++ 3 files changed, 448 insertions(+), 6 deletions(-) create mode 100644 community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncHistoryAppenderTest.java diff --git a/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java b/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java index fc3d976fd..7fd33a67a 100644 --- a/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java +++ b/community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Objects; +import java.util.UUID; public class MongoDBSyncAuditPersistence extends AbstractCommunityAuditPersistence { @@ -115,9 +116,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.write(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; }); // Spends the stream position, and only a committed transaction may reach this line. In general a // normal return from wrapExecution does NOT mean commit — a FailedStep result is returned @@ -138,10 +142,57 @@ private Result writeJournalAndAuditInTransaction(AuditEntry auditEntry) { * accepted best-effort behavior for non-transactional deployments. */ private Result writeJournalAndAuditWithoutTransaction(AuditEntry auditEntry) { - JournalEvent journalEvent = journalEventSequencer.newEvent(auditEntry); - journalEventStore.write(journalEvent); + appendEventFrom(auditEntry, null); journalEventSequencer.confirm(); return auditRepository.save(auditEntry); } + /** + * Appends supplied audit history using its execution/change/state key, independently of the journal flag. + * + * @param entry historical audit entry to preserve + * @return the repository write result + */ + @Override + public Result append(AuditEntry entry) { + return auditRepository.append(entry); + } + + /** + * Durably appends a CHANGE_STATE event to this persistence stream without writing audit records. + * Transaction-capable deployments use an owned session, independent of any ambient business transaction. + * Other deployments use a standalone journal insert. The stream position is confirmed only after success. + * + * @param payload source audit entry preserved in the event + * @return the durable journal write result + * @throws IllegalStateException if journal events are disabled or a contribution reports failure + */ + @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) { + if (!FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) { + throw new IllegalStateException("Journal history requires JOURNAL_EVENTS to be enabled"); + } + JournalEvent event = journalEventSequencer.newEvent(payload); + Result result = session == null ? journalEventStore.write(event) : journalEventStore.write(session, event); + if (result.isError()) { + throw new IllegalStateException("MongoDB history write failed", ((Result.Error) result).getError()); + } + return result; + } + } diff --git a/community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistenceJournalTest.java b/community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistenceJournalTest.java index 4740d39d5..785333cc9 100644 --- a/community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistenceJournalTest.java +++ b/community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistenceJournalTest.java @@ -30,6 +30,11 @@ import io.flamingock.internal.common.core.journal.JournalEvent; import io.flamingock.internal.common.core.journal.JournalEventType; import io.flamingock.internal.common.mongodb.MongoDBJournalEventMapper; +import io.flamingock.internal.common.mongodb.MongoDBAuditMapper; +import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper; +import io.flamingock.internal.core.context.BasicRuntimeContext; +import io.flamingock.internal.util.Result; +import org.bson.Document; import io.flamingock.internal.core.configuration.community.CommunityConfiguration; import io.flamingock.internal.core.journal.JournalEventSequencer; import io.flamingock.internal.core.journal.JournalEventSequencerFactory; @@ -58,6 +63,10 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doCallRealMethod; /** * Covers the {@code Features.JOURNAL_EVENTS} gate in {@link MongoDBSyncAuditPersistence}, and the transaction @@ -306,6 +315,109 @@ void nonTransactionalAuditFailurePreservesAndConfirmsJournalEvent() { assertEquals("change-2", auditRepository.getAuditHistory().get(0).getChangeId()); } + @Test + @DisplayName("historical audit append preserves all states independently of the journal flag") + void historicalAppendPreservesAllStatesWithJournalOnAndOff() { + MongoDBSyncAuditPersistence persistence = freshPersistence(true); + AuditEntry started = auditEntry("history", AuditEntry.Status.STARTED); + AuditEntry applied = auditEntry("history", AuditEntry.Status.APPLIED); + assertFalse(persistence.append(started).isError()); + FeatureFlag.enable(Features.JOURNAL_EVENTS); + assertFalse(persistence.append(applied).isError()); + List history = persistence.getAuditHistory(); + assertEquals(2, history.size()); + assertTrue(history.stream().anyMatch(entry -> entry.getState() == AuditEntry.Status.STARTED)); + assertTrue(history.stream().anyMatch(entry -> entry.getState() == AuditEntry.Status.APPLIED)); + assertTrue(history.stream().anyMatch(entry -> started.getExecutionId().equals(entry.getExecutionId()))); + assertTrue(history.stream().anyMatch(entry -> applied.getExecutionId().equals(entry.getExecutionId()))); + assertFalse(persistence.append(started).isError()); + assertEquals(2, persistence.getAuditHistory().size()); + assertTrue(storedEvents().isEmpty()); + } + + @Test + @DisplayName("source journal append preserves payload and persistence stream in both transaction modes") + void sourceAppendPreservesPayloadAndTailInBothTransactionModes() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + AuditEntry payload = auditEntry("source-history"); + MongoDBAuditMapper payloadMapper = + new MongoDBAuditMapper<>(() -> new MongoDBDocumentHelper(new Document())); + for (boolean transactional : new boolean[]{true, false}) { + MongoDBSyncAuditPersistence persistence = freshPersistence(transactional); + assertFalse(persistence.appendEventFrom(payload).isError()); + assertTrue(persistence.getAuditHistory().isEmpty()); + } + List> events = storedEvents(); + assertEquals(2, events.size()); + events.sort(java.util.Comparator.comparingLong(JournalEvent::getStreamSequence)); + for (int index = 0; index < events.size(); index++) { + JournalEvent event = events.get(index); + assertEquals(STREAM_ID, event.getStreamId()); + assertEquals(index + 1L, 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()); + assertEquals(payloadMapper.toDocument(payload).getDocument(), + payloadMapper.toDocument(event.getData()).getDocument()); + } + } + + @Test + @DisplayName("source journal commit survives rollback of the ambient business transaction") + void sourceAppendIsIndependentOfAmbientBusinessTransaction() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = freshPersistence(true); + AuditEntry payload = auditEntry("independent-history"); + assertThrows(DatabaseTransactionException.class, () -> txWrapper.wrapExecution( + new BasicRuntimeContext("ambient-business"), context -> { + ClientSession ambient = context.getContext().getRequiredDependencyValue(ClientSession.class); + database.getCollection(AUDIT_COLLECTION).insertOne(ambient, new Document("business", true)); + assertFalse(persistence.appendEventFrom(payload).isError()); + throw new IllegalStateException("rollback business"); + })); + assertTrue(persistence.getAuditHistory().isEmpty()); + assertEquals(1, storedEvents().size()); + assertEquals(STREAM_ID, storedEvents().get(0).getStreamId()); + assertEquals(1L, storedEvents().get(0).getStreamSequence()); + } + + @Test + @DisplayName("source contribution error rolls back its transaction and retry reuses the durable position") + void sourceContributionErrorRollsBackAndRetryLeavesNoGap() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + journalEventStore.initialize(true); + MongoDBSyncJournalEventStore failingJournal = spy(journalEventStore); + MongoDBSyncAuditPersistence persistence = new MongoDBSyncAuditPersistence( + new CommunityConfiguration(), auditRepository, failingJournal, + new JournalEventSequencerFactory(failingJournal).forStream(STREAM_ID), true, txWrapper, true); + persistence.initialize(RunnerId.generate()); + doReturn(new Result.Error(new IllegalArgumentException("journal"))) + .when(failingJournal).write(any(ClientSession.class), any()); + assertThrows(DatabaseTransactionException.class, + () -> persistence.appendEventFrom(auditEntry("source-retry"))); + assertTrue(storedEvents().isEmpty()); + assertTrue(persistence.getAuditHistory().isEmpty()); + doCallRealMethod().when(failingJournal).write(any(ClientSession.class), any()); + assertFalse(persistence.appendEventFrom(auditEntry("source-retry")).isError()); + assertEquals(1, storedEvents().size()); + assertEquals(1L, storedEvents().get(0).getStreamSequence()); + assertTrue(persistence.getAuditHistory().isEmpty()); + } + + private MongoDBSyncAuditPersistence freshPersistence(boolean transactional) { + if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) { + journalEventStore.initialize(true); + } + MongoDBSyncAuditPersistence persistence = new MongoDBSyncAuditPersistence( + new CommunityConfiguration(), auditRepository, journalEventStore, + new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID), + transactional, transactional ? txWrapper : null, true); + persistence.initialize(RunnerId.generate()); + return persistence; + } + // ----------------------------- helpers ----------------------------- /** @@ -333,7 +445,8 @@ private MongoDBSyncAuditPersistence persistenceWithoutTransactions( MongoDBSyncJournalEventStore journalStore, JournalEventSequencer sequencer) { MongoDBSyncAuditPersistence persistence = new MongoDBSyncAuditPersistence( - 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-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncHistoryAppenderTest.java b/community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncHistoryAppenderTest.java new file mode 100644 index 000000000..b36ca8141 --- /dev/null +++ b/community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncHistoryAppenderTest.java @@ -0,0 +1,278 @@ +/* + * 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.sync.internal; + +import com.mongodb.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 io.flamingock.internal.util.id.RunnerId; +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.assertNotEquals; +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.ArgumentMatchers.same; +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 MongoDBSyncHistoryAppenderTest { + private static final String STAGE = "normal-stage"; + private final MongoDBSyncAuditRepository audit = mock(MongoDBSyncAuditRepository.class); + private final MongoDBSyncJournalEventStore journal = mock(MongoDBSyncJournalEventStore.class); + private final JournalEventSequencerFactory factory = new JournalEventSequencerFactory(journal); + private final Boundary wrapper = new Boundary(); + 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 Result journalResult = Result.OK(); + private RuntimeException journalFailure; + + @AfterEach + void resetFlag() { + FeatureFlag.remove(Features.JOURNAL_EVENTS); + } + + @Test + void auditAppendPreservesHistoricalEntryRegardlessOfJournalFlag() { + for (boolean enabled : new boolean[]{false, true}) { + if (enabled) { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + } + MongoDBSyncAuditPersistence persistence = persistence(true); + Result expected = Result.OK(); + when(audit.append(entry)).thenReturn(expected); + assertSame(expected, persistence.append(entry)); + verify(audit).append(same(entry)); + verifyNoInteractions(journal); + assertEquals(0, wrapper.commits); + clearInvocations(audit); + } + } + + @Test + void journalOffRejectsTransactionalSourceAppendWithoutSpendingSequence() { + assertJournalOffRejectsSourceAppend(true); + } + + @Test + void journalOffRejectsStandaloneSourceAppendWithoutSpendingSequence() { + assertJournalOffRejectsSourceAppend(false); + } + + private void assertJournalOffRejectsSourceAppend(boolean transactional) { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(transactional); + FeatureFlag.disable(Features.JOURNAL_EVENTS); + assertEquals("Journal history requires JOURNAL_EVENTS to be enabled", + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)).getMessage()); + verifyNoInteractions(journal, audit); + assertTrue(durable.isEmpty()); + assertEquals(0, wrapper.commits); + + FeatureFlag.enable(Features.JOURNAL_EVENTS); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(1, durable.size()); + assertEquals(STAGE, durable.get(0).getStreamId()); + assertEquals(1L, durable.get(0).getStreamSequence()); + assertSame(entry, durable.get(0).getData()); + assertEquals(transactional ? 1 : 0, wrapper.commits); + verifyNoInteractions(audit); + } + + @Test + void sourceAppendUsesPersistenceStreamAndCommitsOnlyJournal() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(true); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(1, wrapper.commits); + assertEquals(1, durable.size()); + JournalEvent event = durable.get(0); + assertSame(entry, event.getData()); + assertEquals(STAGE, event.getStreamId()); + assertEquals(1L, 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()); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(2L, durable.get(1).getStreamSequence()); + assertNotEquals(event.getEventId(), durable.get(1).getEventId()); + assertNotEquals(wrapper.sessionIds.get(0), wrapper.sessionIds.get(1)); + verifyNoInteractions(audit); + } + + @Test + void sourceAppendWithoutTransactionsWritesOnlyJournal() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(false); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(0, wrapper.commits); + assertEquals(1, durable.size()); + assertEquals(STAGE, durable.get(0).getStreamId()); + assertSame(entry, durable.get(0).getData()); + verifyNoInteractions(audit); + } + + @Test + void failedSourceContributionAbortsCommitAndRetryLeavesNoGap() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(true); + journalResult = new Result.Error(new IllegalArgumentException("journal")); + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)); + assertEquals(0, wrapper.commits); + assertTrue(durable.isEmpty()); + journalResult = Result.OK(); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(1L, durable.get(0).getStreamSequence()); + verifyNoInteractions(audit); + } + + @Test + void sourceWriteAndCommitExceptionsLeaveNoDurableEvent() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(true); + journalFailure = new IllegalStateException("journal"); + assertSame(journalFailure, assertThrows(IllegalStateException.class, + () -> persistence.appendEventFrom(entry))); + journalFailure = null; + wrapper.failCommit = true; + assertEquals("commit", assertThrows(IllegalStateException.class, + () -> persistence.appendEventFrom(entry)).getMessage()); + assertTrue(durable.isEmpty()); + assertEquals(0, wrapper.commits); + wrapper.failCommit = false; + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(1L, durable.get(0).getStreamSequence()); + verifyNoInteractions(audit); + } + + @Test + void failedNonTransactionalSourceWriteDoesNotAdvanceTail() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(false); + journalResult = new Result.Error(new IllegalArgumentException("journal")); + assertThrows(IllegalStateException.class, () -> persistence.appendEventFrom(entry)); + assertTrue(durable.isEmpty()); + journalResult = Result.OK(); + assertFalse(persistence.appendEventFrom(entry).isError()); + assertEquals(1L, durable.get(0).getStreamSequence()); + verifyNoInteractions(audit); + } + + @Test + void normalWritesRetainPersistenceStageAndJointTransaction() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(true); + assertFalse(persistence.writeEntry(entry).isError()); + verify(audit).save(same(wrapper.session), same(entry)); + assertEquals(STAGE, durable.get(0).getStreamId()); + assertEquals(1, wrapper.commits); + } + + @Test + void auditContributionErrorAbortsJointWrite() { + FeatureFlag.enable(Features.JOURNAL_EVENTS); + MongoDBSyncAuditPersistence persistence = persistence(true); + when(audit.save(any(ClientSession.class), any())).thenReturn(new Result.Error(new RuntimeException("audit"))); + assertThrows(IllegalStateException.class, () -> persistence.writeEntry(entry)); + assertTrue(durable.isEmpty()); + assertEquals(0, wrapper.commits); + when(audit.save(any(ClientSession.class), any())).thenReturn(Result.OK()); + assertFalse(persistence.writeEntry(entry).isError()); + assertEquals(1L, durable.get(0).getStreamSequence()); + } + + private MongoDBSyncAuditPersistence persistence(boolean transactional) { + when(journal.getLastEventByStream(anyString())).thenAnswer(invocation -> { + String stream = invocation.getArgument(0); + return durable.stream().filter(event -> stream.equals(event.getStreamId())) + .reduce((first, last) -> last); + }); + when(journal.write(any(ClientSession.class), any())).thenAnswer(invocation -> { + assertSame(wrapper.session, invocation.getArgument(0)); + staged = invocation.getArgument(1); + return writeResult(); + }); + when(journal.write(any(JournalEvent.class))).thenAnswer(invocation -> { + staged = invocation.getArgument(0); + Result result = writeResult(); + if (!result.isError()) { + durable.add(staged); + } + return result; + }); + when(audit.save(any(ClientSession.class), any())).thenReturn(Result.OK()); + MongoDBSyncAuditPersistence persistence = new MongoDBSyncAuditPersistence( + new CommunityConfiguration(), audit, journal, factory.forStream(STAGE), transactional, + transactional ? wrapper : null, true); + persistence.initialize(RunnerId.generate()); + clearInvocations(audit, journal); + return persistence; + } + + private Result writeResult() { + if (journalFailure != null) { + throw journalFailure; + } + return journalResult; + } + + private class Boundary implements ExecutionWrapper { + private final ClientSession session = mock(ClientSession.class); + private final List sessionIds = new ArrayList<>(); + private int commits; + private boolean failCommit; + + @Override + public R wrapExecution(C context, Function operation) { + sessionIds.add(context.getSessionId()); + context.addDependency(session); + staged = null; + R result = operation.apply(context); + if (failCommit) { + throw new IllegalStateException("commit"); + } + durable.add(staged); + commits++; + return result; + } + } +}