Skip to content

feat(couchbase): add journal event persistence #949

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
davidfrigolet wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat(couchbase): add journal event persistence #949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
101 changes: 79 additions & 22 deletions ...couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/CouchbaseAuditStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,38 +18,58 @@
import com.couchbase.client.core.io.CollectionIdentifier;
import com.couchbase.client.java.Bucket;
import com.couchbase.client.java.Cluster;
import com.couchbase.client.java.transactions.TransactionAttemptContext;
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.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.TimeService;
import io.flamingock.internal.util.constants.CommunityPersistenceConstants;
import io.flamingock.internal.common.couchbase.journal.JournalEventPersistenceConstants;
import io.flamingock.internal.util.id.RunnerId;
import io.flamingock.store.couchbase.internal.CouchbaseAuditPersistence;
import io.flamingock.store.couchbase.internal.CouchbaseAuditor;
import io.flamingock.store.couchbase.internal.CouchbaseJournalEventStore;
import io.flamingock.store.couchbase.internal.CouchbaseLockService;
import io.flamingock.externalsystem.couchbase.api.CouchbaseExternalSystem;

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

public class CouchbaseAuditStore implements CommunityAuditStore {

private final CouchbaseExternalSystem targetSystem;
private final Cluster cluster;
private final String bucketName;
private RunnerId runnerId;
private CommunityConfigurable communityConfiguration;
private CouchbaseAuditPersistence persistence;
private CouchbaseLockService lockService;
private Bucket bucket;
private String scopeName = CollectionIdentifier.DEFAULT_SCOPE;
private String auditRepositoryName = CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME;
private String lockRepositoryName = CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME;
private String journalRepositoryName = JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME;
private boolean autoCreate = true;


private CouchbaseAuditStore(Cluster cluster, String bucketName) {
this.cluster = cluster;
this.bucketName = bucketName;
private CouchbaseAuditor auditor;
private CouchbaseJournalEventStore journalEventStore;
private JournalEventSequencerFactory journalEventSequencerFactory;


private CouchbaseAuditStore(CouchbaseExternalSystem targetSystem) {
// Cannot resolve targetSystem.getTxWrapper() here: the target system's own initialize() — which is
// what sets it — runs later than this constructor (called eagerly when the caller builds this audit
// store), so it would still be null at this point. Kept as a live reference and resolved lazily in
// getPersistenceFactory(), by which point both the target system and this audit store are initialized.
this.targetSystem = targetSystem;
this.cluster = targetSystem.getCluster();
this.bucketName = targetSystem.getBucketName();
}

/**
Expand All @@ -63,7 +83,7 @@ private CouchbaseAuditStore(Cluster cluster, String bucketName) {
* @return a new audit store bound to the same Couchbase instance as the target system
*/
public static CouchbaseAuditStore from(CouchbaseExternalSystem targetSystem) {
return new CouchbaseAuditStore(targetSystem.getCluster(), targetSystem.getBucketName());
return new CouchbaseAuditStore(targetSystem);
}

@Override
Expand All @@ -86,45 +106,70 @@ public CouchbaseAuditStore withLockRepositoryName(String lockRepositoryName) {
return this;
}

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

public CouchbaseAuditStore withAutoCreate(boolean autoCreate) {
this.autoCreate = autoCreate;
return this;
}

@Override
public void initialize(ContextResolver baseContext) {
this.validate();
runnerId = baseContext.getRequiredDependencyValue(RunnerId.class);
communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class);
this.validate();

auditor = new CouchbaseAuditor(cluster, bucket);
journalEventStore = new CouchbaseJournalEventStore(cluster, bucket);
journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore);

lockService = new CouchbaseLockService(cluster, bucket, TimeService.getDefault());
lockService.initialize(autoCreate, scopeName, lockRepositoryName);
}

@Override
public synchronized CommunityAuditPersistence getPersistence() {
if (persistence == null) {
persistence = new CouchbaseAuditPersistence(
public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory() {
return stageId -> {
JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
CouchbaseAuditPersistence persistence = new CouchbaseAuditPersistence(
communityConfiguration,
cluster,
bucket,
auditor,
journalEventStore,
journalEventSequencer,
targetSystem.getTxWrapper(),
scopeName,
auditRepositoryName,
journalRepositoryName,
autoCreate);
persistence.initialize(runnerId);
}
return persistence;
return persistence;
};
}

@Override
public CommunityAuditPersistence getPersistence() {
throw new UnsupportedOperationException("getPersistence shouldn't be called at Couchbase audit store; use getPersistenceFactory(stageId)");
}

@Override
public AuditReader getAuditReader() {
auditor.initialize(autoCreate, scopeName, auditRepositoryName);
return () -> auditor.getAuditHistory();
}

@Override
public synchronized CommunityLockService getLockService() {
if (lockService == null) {
lockService = new CouchbaseLockService(cluster, bucket, TimeService.getDefault());
lockService.initialize(
autoCreate,
scopeName,
lockRepositoryName);
}
return lockService;
}

@Override
public Set<Class<?>> getNonGuardedTypes() {
return Collections.singleton(TransactionAttemptContext.class);
}

private void validate() {

if (cluster == null) {
Expand Down Expand Up @@ -152,8 +197,20 @@ 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.");
}
}
}
67 changes: 55 additions & 12 deletions ...store/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,54 +15,97 @@
*/
package io.flamingock.store.couchbase.internal;

import com.couchbase.client.java.Bucket;
import com.couchbase.client.java.Cluster;
import com.couchbase.client.java.transactions.TransactionAttemptContext;
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.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.context.BasicRuntimeContext;
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.List;

public class CouchbaseAuditPersistence extends AbstractCommunityAuditPersistence {

private final Cluster cluster;
private final Bucket bucket;
private final CouchbaseAuditor auditor;
private final CouchbaseJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final TransactionWrapper txWrapper;
private final String scopeName;
private final String auditRepositoryName;
private final String journalRepositoryName;
private final boolean autoCreate;

private CouchbaseAuditor auditor;


public CouchbaseAuditPersistence(CommunityConfigurable localConfiguration,
Cluster cluster,
Bucket bucket,
CouchbaseAuditor auditor,
CouchbaseJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
TransactionWrapper txWrapper,
String scopeName,
String auditRepositoryName,
String journalRepositoryName,
boolean autoCreate) {
super(localConfiguration);
this.cluster = cluster;
this.bucket = bucket;
this.auditor = auditor;
this.journalEventStore = journalEventStore;
this.journalEventSequencer = journalEventSequencer;
this.txWrapper = txWrapper;
this.scopeName = scopeName;
this.auditRepositoryName = auditRepositoryName;
this.journalRepositoryName = journalRepositoryName;
this.autoCreate = autoCreate;
}

@Override
protected void doInitialize(RunnerId runnerId) {
auditor = new CouchbaseAuditor(cluster, bucket);
auditor.initialize(autoCreate, scopeName, auditRepositoryName);
// Creating the collection/indexes is what brings the journal collection into existence, so skipping
// this keeps it from ever appearing while the flag is off. It must stay in step with the append in
// writeEntry: skipping setup while still appending would let ctx.insert create the collection
// implicitly and without indexes, voiding the stream-position and eventId-lookup guarantees.
FeatureFlag.ifEnabled(Features.JOURNAL_EVENTS, () -> journalEventStore.initialize(autoCreate, scopeName, journalRepositoryName));
}


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

@Override
public Result writeEntry(AuditEntry auditEntry) {
return auditor.writeEntry(auditEntry);
// 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)) {
RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
TransactionAttemptContext ctx = runtimeContext.getContext().getRequiredDependencyValue(TransactionAttemptContext.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
journalEventStore.contributeToTransaction(ctx, journalEvent);
return auditor.contributeToTransaction(ctx, auditEntry);
});
// Spends the stream position, and only a committed transaction attempt may reach this line. A
// normal return from wrapInTransaction does NOT in general mean commit — CouchbaseTxWrapper
// returns normally after a deliberate rollback too, when the operation's result is a FailedStep.
// It is sound here because this operation returns a Result, which can never be a FailedStep, so
// the only way to return normally is a committed attempt; a failing attempt is caught and
// rethrown as TransactionFailedException (see CouchbaseTxWrapper — it doesn't yet wrap that as
// DatabaseTransactionException, a known deviation from the TransactionWrapper contract, tracked
// separately from this ticket). 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 auditor.append(auditEntry);
}
}
}
Loading
Loading