Fix/journal event sequence gaps

community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java

@@ -73,17 +73,26 @@ public List<AuditEntry> getAuditHistory() {
     @Override
     public Result writeEntry(AuditEntry auditEntry) {
         RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
+        // Read once rather than per branch: the journal append and the audit write shape are two halves of one
+        // model. With events, the audit record is the change's current state and the journal is the history;
+        // without them, the audit record set is itself the history.
         if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS)) {
-            return txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
+            Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
                 ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class);
-                // Read once rather than per branch: the journal append and the audit write shape are two halves of
-                // one model. With events, the audit record is the change's current state and the journal is the
-                // history; without them, the audit record set is itself the history.
                 JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
                 journalEventStore.write(clientSession, journalEvent);
                 return auditRepository.save(clientSession, auditEntry);
 
             });
+            // Spends the stream position, and only a committed transaction may reach this line. In general a
+            // normal return from wrapInTransaction does NOT mean commit — a FailedStep result is returned
+            // after a rollback, without an exception. It is sound here because this operation returns a
+            // Result, which can never be a FailedStep, so the commit branch is the only graceful path; a
+            // failing commit is caught and rethrown as DatabaseTransactionException. Keep that true: an
+            // operation that could return a failed step would silently burn a position and gap the stream,
+            // and a contiguous sequence is what lets a consumer tell "in flight" from "lost".
+            journalEventSequencer.confirm();
+            return result;
         } else {
             return auditRepository.append(auditEntry);
         }

community/flamingock-mongodb-sync-auditstore/src/test/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistenceJournalTest.java

@@ -199,6 +199,27 @@ void auditFailureRollsBackJournalEvent() {
         assertTrue(storedEvents().isEmpty(), "the journal event must not survive a failed audit write");
     }
 
+    @Test
+    @DisplayName("journal enabled: a failed write leaves no gap — its stream position is handed out again")
+    void failedWriteLeavesNoGapInTheStream() {
+        FeatureFlag.enable(Features.JOURNAL_EVENTS);
+        JournalEventSequencer sequencer = new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID);
+
+        MongoDBSyncAuditRepository failingRepository = mock(MongoDBSyncAuditRepository.class);
+        doThrow(new IllegalStateException("audit write failed"))
+                .when(failingRepository).save(any(ClientSession.class), any(AuditEntry.class));
+        MongoDBSyncAuditPersistence failing = persistenceFor(failingRepository, sequencer);
+        assertThrows(DatabaseTransactionException.class, () -> failing.writeEntry(auditEntry("change-1")));
+
+        // Same sequencer: the aborted attempt must not have spent position 1.
+        persistenceFor(auditRepository, sequencer).writeEntry(auditEntry("change-1"));
+
+        List<JournalEvent<AuditEntry>> events = storedEvents();
+        assertEquals(1, events.size());
+        assertEquals(1L, events.get(0).getStreamSequence(),
+                "the stream must stay contiguous, so consumers can tell in-flight from lost");
+    }
+
     // ----------------------------- helpers -----------------------------
 
     /**
@@ -209,7 +230,11 @@ void auditFailureRollsBackJournalEvent() {
      * gated.
      */
     private MongoDBSyncAuditPersistence persistenceFor(MongoDBSyncAuditRepository repository) {
-        JournalEventSequencer sequencer = new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID);
+        return persistenceFor(repository, new JournalEventSequencerFactory(journalEventStore).forStream(STREAM_ID));
+    }
+
+    private MongoDBSyncAuditPersistence persistenceFor(MongoDBSyncAuditRepository repository,
+                                                       JournalEventSequencer sequencer) {
         MongoDBSyncAuditPersistence persistence = new MongoDBSyncAuditPersistence(
                 new CommunityConfiguration(), repository, journalEventStore, sequencer, txWrapper, true);
         persistence.initialize(RunnerId.generate());

core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/transaction/TransactionWrapper.java

@@ -104,7 +104,11 @@ public interface TransactionWrapper {
      *                       transaction-scoped dependencies
      * @param operation      the work to execute within the transaction
      * @return whatever the operation returned — including a failed step, when the operation reported
-     *         its failure that way
+     *         its failure that way. <strong>A normal return does not imply the transaction committed</strong>:
+     *         a failed step is returned after a rollback, without an exception. Callers that must know the
+     *         work is durable — to advance a counter, publish, or acknowledge — cannot infer it from a
+     *         normal return alone; they either check the result themselves, or rely on an operation whose
+     *         return type cannot be a failed step.
      * @throws DatabaseTransactionException if the operation throws, or if the transaction itself cannot
      *                                      be started, committed or rolled back
      */

core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencer.java

@@ -23,26 +23,55 @@
 import java.time.Instant;
 import java.util.UUID;
 
+/**
+ * Hands out stream positions for a single stream, in memory — safe because the stage lock guarantees one
+ * writer per stream.
+ * <p>
+ * A position is only spent once the caller confirms the event was durably written. Until then the same
+ * position is handed out again, so a write that fails and aborts leaves no hole in the stream. That matters
+ * beyond tidiness: a contiguous {@code streamSequence} is what lets a consumer reconstruct order and tell
+ * "still in flight" from "lost", so gaps must never be produced by ordinary failure handling.
+ * <p>
+ * If a caller forgets to confirm a write that did land, the reused position collides with the unique
+ * {@code (streamId, streamSequence)} index — a loud failure rather than a silent duplicate.
+ */
 public class JournalEventSequencer {
     private final String streamId;
     private long nextSequence;
+    private boolean pendingConfirmation;
 
     JournalEventSequencer(String streamId, long initialSequence) {
         this.streamId = streamId;
         this.nextSequence = initialSequence;   // seeded from outside
     }
 
+    /**
+     * Builds the next event <em>without</em> spending its stream position; call {@link #confirm()} once the
+     * event is durably written.
+     */
     public JournalEvent<AuditEntry> newEvent(AuditEntry payload) {
         return getAuditEntryJournalEvent(payload, JournalEventType.CHANGE_STATE);
     }
 
+    /**
+     * Marks the position handed out by the last {@link #newEvent} as durably written, moving the stream on.
+     * A no-op if nothing is outstanding.
+     */
+    public void confirm() {
+        if (pendingConfirmation) {
+            nextSequence++;
+            pendingConfirmation = false;
+        }
+    }
+
     @NotNull
     private JournalEvent<AuditEntry> getAuditEntryJournalEvent(AuditEntry payload, JournalEventType type) {
+        pendingConfirmation = true;
         return new JournalEvent<>(
                 UUID.randomUUID().toString(),   // eventId
                 type,
                 streamId,
-                nextSequence++,                 // in-memory, safe: distributed lock covers it
+                nextSequence,                   // spent only on confirm(), so a failed write leaves no gap
                 Instant.now(),                  // occurredAt
                 payload);
     }