Skip to content

feat(mongoreactive): add journal event persistence #955

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bercianor wants to merge 3 commits into
base: master
Choose a base branch
from
Open

feat(mongoreactive): add journal event persistence #955

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c4fe8d1
feat(mongoreactive): add journal event persistence
bercianor Sep 3, 2026
17d7acf
fix(mongoreactive): address journal review feedback
bercianor Sep 9, 2026
47c8187
Merge branch 'master' into feat/mongodb-reactive-journal-events
bercianor Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,3 +103,5 @@ test_*.db-shm
# Generic SQLite artifacts
*.db-wal
*.db-shm
# Local Pi runtime state
.atl/

@davidfrigolet davidfrigolet Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Stray, unrelated to journal events. Not present in #948/#949 or the sync PR. Please drop this line from this PR.

@bercianor bercianor Sep 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This one is intentional. It ignores a directory generated by my local development environment that should never be tracked.

I agree it is unrelated to journal events, but I do not think it warrants a separate PR for a single repository-hygiene entry. It does not affect the implementation or runtime behavior.

97 changes: 72 additions & 25 deletions ...itstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,22 +21,31 @@
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoDatabase;
import io.flamingock.externalsystem.mongodb.reactive.api.MongoDBReactiveExternalSystem;
import io.flamingock.internal.common.core.audit.AuditPersistenceFactory;
import io.flamingock.internal.common.core.audit.AuditReader;
import io.flamingock.internal.common.core.context.ContextResolver;
import io.flamingock.internal.common.core.error.FlamingockException;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.CommunityAuditStore;
import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence;
import io.flamingock.internal.core.external.store.lock.community.CommunityLockService;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.core.journal.JournalEventSequencerFactory;
import io.flamingock.internal.util.Constants;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.TimeService;
import io.flamingock.internal.util.id.RunnerId;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditPersistence;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditRepository;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveJournalEventStore;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveLockService;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

import static io.flamingock.internal.common.mongodb.journal.JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME;
import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME;
import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME;

Expand All @@ -46,15 +55,19 @@ public class MongoDBReactiveAuditStore implements CommunityAuditStore {

protected RunnerId runnerId;
private CommunityConfigurable communityConfiguration;
private MongoDBReactiveAuditPersistence persistence;
private CommunityAuditPersistence persistence;
private MongoDBReactiveLockService lockService;
private MongoDatabase database;
private String auditRepositoryName = DEFAULT_AUDIT_STORE_NAME;
private String lockRepositoryName = DEFAULT_LOCK_STORE_NAME;
private String journalRepositoryName = DEFAULT_JOURNAL_STORE_NAME;
private ReadConcern readConcern = ReadConcern.MAJORITY;
private ReadPreference readPreference = ReadPreference.primary();
private WriteConcern writeConcern = WriteConcern.MAJORITY.withJournal(true);
private boolean autoCreate = true;
private MongoDBReactiveAuditRepository auditRepository;
private MongoDBReactiveJournalEventStore journalEventStore;
private JournalEventSequencerFactory journalEventSequencerFactory;

private MongoDBReactiveAuditStore(MongoDBReactiveExternalSystem mongoDBTargetSystem) {
this.mongoDBTargetSystem = mongoDBTargetSystem;
Expand Down Expand Up @@ -89,6 +102,11 @@ public MongoDBReactiveAuditStore withLockRepositoryName(String lockRepositoryNam
return this;
}

public MongoDBReactiveAuditStore withJournalRepositoryName(String journalRepositoryName) {
this.journalRepositoryName = journalRepositoryName;
return this;
}

public MongoDBReactiveAuditStore withReadConcern(ReadConcern readConcern) {
this.readConcern = readConcern;
return this;
Expand All @@ -114,40 +132,57 @@ public void initialize(ContextResolver baseContext) {
runnerId = baseContext.getRequiredDependencyValue(RunnerId.class);
communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class);
database = mongoDBTargetSystem.getMongoDatabase();
this.validate();
this.validate();

auditRepository = new MongoDBReactiveAuditRepository(
database, auditRepositoryName, readConcern, readPreference, writeConcern);
journalEventStore = new MongoDBReactiveJournalEventStore(
database, journalRepositoryName, readConcern, readPreference, writeConcern);
journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore);

lockService = new MongoDBReactiveLockService(
database,
lockRepositoryName,
readConcern,
readPreference,
writeConcern,
TimeService.getDefault()
);
lockService.initialize(autoCreate);
}

@Override
public synchronized CommunityAuditPersistence getPersistence() {
if (persistence == null) {
persistence = new MongoDBReactiveAuditPersistence(
public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory() {
return stageId -> {
auditRepository.initialize(autoCreate);

@davidfrigolet davidfrigolet Sep 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

auditRepository.initialize(autoCreate) (and journalEventStore.initialize below) run here, then again in stagePersistence.initialize(runnerId) -> doInitialize(). Both are guarded so it's harmless, but the sync sibling only initializes inside persistence.doInitialize(). Consolidate to one site. Keep the journal-store init before forStream(stageId) though — the manual-schema validation test depends on that ordering.

@bercianor bercianor Sep 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I looked into this properly. We cannot just move these calls into doInitialize() with the current flow: forStream(stageId) immediately reads the last persisted journal event, but doInitialize() runs afterwards. That would make the sequencer read the journal before its schema is
created or validated, especially breaking the intended autoCreate=false validation boundary.

The duplicate calls are idempotent and currently preserve the required order:

audit initialization → journal initialization/validation → sequencer creation

I agree the lifecycle is not clean. The right solution is a cross-audit-store refactor: persistence should receive the stage ID and sequencer factory, initialize the repositories, and then create the sequencer itself.

That refactor should be applied consistently across the audit stores, not only MongoDB Reactive. It is therefore better handled separately from this PR, rather than introducing a broader lifecycle-contract change here.

if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) {
journalEventStore.initialize(autoCreate);
}
JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
MongoDBReactiveAuditPersistence stagePersistence = new MongoDBReactiveAuditPersistence(
communityConfiguration,
database,
auditRepositoryName,
readConcern,
readPreference,
writeConcern,
auditRepository,
journalEventStore,
journalEventSequencer,
mongoDBTargetSystem.getTxWrapper(),
autoCreate
);
persistence.initialize(runnerId);
}
return persistence;
stagePersistence.initialize(runnerId);
if (persistence == null) {
persistence = stagePersistence;
}
return stagePersistence;
};
}

@Override
public synchronized CommunityLockService getLockService() {
if (lockService == null) {
lockService = new MongoDBReactiveLockService(
database,
lockRepositoryName,
readConcern,
readPreference,
writeConcern,
TimeService.getDefault()
);
lockService.initialize(autoCreate);
public AuditReader getAuditReader() {
auditRepository.initialize(autoCreate);
return () -> auditRepository.getAuditHistory();
}

}
@Override
public synchronized CommunityLockService getLockService() {
return lockService;
}

Expand All @@ -165,10 +200,22 @@ private void validate() {
throw new FlamingockException("The 'lockRepositoryName' property is required.");
}

if (journalRepositoryName == null || journalRepositoryName.trim().isEmpty()) {
throw new FlamingockException("The 'journalRepositoryName' property is required.");
}

if (auditRepositoryName.trim().equalsIgnoreCase(lockRepositoryName.trim())) {
throw new FlamingockException("The 'auditRepositoryName' and 'lockRepositoryName' properties must not be the same.");
}

if (journalRepositoryName.trim().equalsIgnoreCase(auditRepositoryName.trim())) {
throw new FlamingockException("The 'journalRepositoryName' and 'auditRepositoryName' properties must not be the same.");
}

if (journalRepositoryName.trim().equalsIgnoreCase(lockRepositoryName.trim())) {
throw new FlamingockException("The 'journalRepositoryName' and 'lockRepositoryName' properties must not be the same.");
}

if (readConcern == null) {
throw new FlamingockException("The 'readConcern' property is required.");
}
Expand Down
77 changes: 45 additions & 32 deletions ...n/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,65 +15,78 @@
*/
package io.flamingock.store.mongodb.reactive.internal;

import com.mongodb.ReadConcern;
import com.mongodb.ReadPreference;
import com.mongodb.WriteConcern;
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoDatabase;
import io.flamingock.internal.common.core.audit.AuditEntry;
import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.journal.JournalEvent;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.core.context.BasicRuntimeContext;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.Result;
import io.flamingock.internal.util.id.RunnerId;

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class MongoDBReactiveAuditPersistence extends AbstractCommunityAuditPersistence {

private MongoDBReactiveAuditor auditor;
private final MongoDatabase database;
private final String auditCollectionName;
private final ReadConcern readConcern;
private final ReadPreference readPreference;
private final WriteConcern writeConcern;
private final MongoDBReactiveAuditRepository auditRepository;
private final MongoDBReactiveJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final TransactionWrapper txWrapper;
private final boolean autoCreate;


public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration,
MongoDatabase database,
String auditCollectionName,
ReadConcern readConcern,
ReadPreference readPreference,
WriteConcern writeConcern,
boolean autoCreate) {
MongoDBReactiveAuditRepository auditRepository,
MongoDBReactiveJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
TransactionWrapper txWrapper,
boolean autoCreate) {
super(localConfiguration);
this.database = database;
this.auditCollectionName = auditCollectionName;
this.readConcern = readConcern;
this.readPreference = readPreference;
this.writeConcern = writeConcern;
this.auditRepository = auditRepository;
this.journalEventStore = journalEventStore;
this.journalEventSequencer = journalEventSequencer;
this.txWrapper = txWrapper;
this.autoCreate = autoCreate;
}

@Override
protected void doInitialize(RunnerId runnerId) {
//Auditor
auditor = new MongoDBReactiveAuditor(database, auditCollectionName, readConcern, readPreference, writeConcern);
auditor.initialize(autoCreate);
auditRepository.initialize(autoCreate);
if (FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false)) {
journalEventStore.initialize(autoCreate);
}
}


@Override
public List<AuditEntry> getAuditHistory() {
return auditor.getAuditHistory();
return auditRepository.getAuditHistory();
}

@Override
public Result writeEntry(AuditEntry auditEntry) {
return auditor.writeEntry(auditEntry);
}
if (FeatureFlag.isDisabled(Features.JOURNAL_EVENTS, false)) {
return auditRepository.append(auditEntry);
}

if (journalEventStore == null || journalEventSequencer == null || txWrapper == null) {
throw new IllegalStateException("MongoDB reactive journal writes require a transaction wrapper and sequencer");
}

RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
journalEventStore.append(clientSession, journalEvent);
return auditRepository.save(clientSession, auditEntry);
});

// Result cannot represent FailedStep. The transaction wrapper has therefore committed successfully
// whenever control reaches this line; only then is the in-memory stream position spent.
journalEventSequencer.confirm();
return result;
}
}
Loading
Loading