diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/builder/AbstractChangeRunnerBuilder.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/builder/AbstractChangeRunnerBuilder.java index 47d685a01..1b5ffee0b 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/builder/AbstractChangeRunnerBuilder.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/builder/AbstractChangeRunnerBuilder.java @@ -58,6 +58,7 @@ import io.flamingock.internal.core.builder.runner.RunnerBuilder; import io.flamingock.internal.core.builder.runner.RunnerFactory; import io.flamingock.internal.core.change.filter.ChangeFilter; +import io.flamingock.internal.core.system.CoreSystemChangeFilter; import io.flamingock.internal.util.CollectionUtil; import io.flamingock.internal.util.Property; import io.flamingock.internal.util.id.RunnerId; @@ -279,6 +280,11 @@ private LoadedPipeline loadPipeline(FlamingockMetadata flamingockMetadata) { .flatMap(List::stream) .collect(Collectors.toList()); + // Keeps the journal-backfill / audit-cleanup system changes out of the runtime pipeline while + // JOURNAL_EVENTS is disabled - see CoreSystemChangeFilter for why this can't be done where they're + // contributed (compile time, a different process from where the flag actually lives). + changeFiltersFromPlugins.add(CoreSystemChangeFilter.instance()); + return LoadedPipeline.builder() .addFilters(changeFiltersFromPlugins) .addPreviewPipeline(flamingockMetadata.getPipeline()) diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java index 7f47700b1..3c8483f0b 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java @@ -15,18 +15,35 @@ */ package io.flamingock.internal.core.journal; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Owns one sequencer per stream for this factory's lifetime. Persistence instances using the same + * stream share confirmed sequence progress. The stage lock still serializes event writes; + * sharing a sequencer does not make the allocation/write/confirmation cycle concurrent-safe. + */ public class JournalEventSequencerFactory { private final JournalEventReader journalEventReader; + private final Map sequencersByStream = new ConcurrentHashMap<>(); public JournalEventSequencerFactory(JournalEventReader journalEventReader) { this.journalEventReader = journalEventReader; } + /** + * Returns the shared sequencer for a stream, seeded from its durable tail on first access. + * + * @param streamId stream whose sequence positions are allocated + * @return the same sequencer for every request for this stream within this factory + */ public JournalEventSequencer forStream(String streamId) { - long initialSequence = journalEventReader.getLastEventByStream(streamId) - .map(e -> e.getStreamSequence() + 1) - .orElse(1L); - return new JournalEventSequencer(streamId, initialSequence); + return sequencersByStream.computeIfAbsent(streamId, id -> { + long initialSequence = journalEventReader.getLastEventByStream(id) + .map(e -> e.getStreamSequence() + 1) + .orElse(1L); + return new JournalEventSequencer(id, initialSequence); + }); } } diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/AuditCleanupChange.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/AuditCleanupChange.java index 52cf2786b..250b509f8 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/AuditCleanupChange.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/AuditCleanupChange.java @@ -28,32 +28,16 @@ * The system change that collapses the audit store to one record per change, once the journal holds the * history that the audit store used to carry. * - *

NOT YET WIRED — read this before assuming it runs

- * This class is deliberately not contributed to any pipeline. Nothing registers it as a system - * change, so it never appears in a stage and never executes. It exists so that the per-store - * {@code AuditCompactor} implementations have a consumer ready and reviewed, and so the eventual wiring is - * a small, obvious change rather than a large one. - *

- * Two things must happen before it can be wired, in this order: - *

    - *
  1. The journal backfill system change must exist. Compaction deletes the audit ledger; - * the backfill is what first preserves it as {@code JournalEvent}s. If this change ever runs before - * that one has, the change history is gone — not in the audit store, not in the journal, and - * unrecoverable. The ordering guarantee is real once both exist (the system stage sorts by - * {@code order} and runs before every other stage), but it cannot protect against a backfill that - * does not exist.
    2. - *
  2. The target-system problem must be settled. Every change is resolved against a target system - * ({@code ChangeProcessStrategyFactory} → {@code TargetSystemManager#getTargetSystem}, which throws - * on an unknown id), and there is no default or null-object target system. Audit cleanup operates on - * the audit store, not on a target system, so it has no id to supply. The Mongock importer sidesteps - * this by taking one from {@code @MongockSupport(targetSystem = ...)}; there is no equivalent here.
    3. - *
- * When wiring, append a system change to {@code FlamingockAnnotationProcessorPlugin#findAnnotatedChanges()} - * — mirroring {@code MongockAnnotationProcessorPlugin#getImporterChange()} — with - * {@code setSystem(true)} and an {@code order} after the backfill's. Reserved sequence: Mongock import - * {@code 00100}, journal backfill {@code 00200}, audit cleanup {@code 00300}. Those orders are compared as - * plain strings and nothing enforces uniqueness, so the backfill's author has to be told which value is - * taken. + *

Wiring

+ * Contributed unconditionally by {@code CoreSystemChangeProvider}, consumed by + * {@code FlamingockAnnotationProcessorPlugin#findAnnotatedChanges()} — mirroring + * {@code MongockAnnotationProcessorPlugin#getImporterChange()} — with {@code setSystem(true)} and + * {@code order = "00300"}, after the journal backfill's {@code "00200"} (Mongock import is + * {@code "00100"}). The system stage sorts by {@code order} and runs before every other stage, so the + * backfill is guaranteed to run first and this change never races it. Its {@link AuditCompactor} dependency + * is already registered by {@code CommunityChangeRunnerBuilder#contributeEditionDependencies}. The + * target-system problem this javadoc used to describe is resolved: audit-store-only system changes resolve + * against the reserved {@code PipelineHelper#AUDIT_STORE_TARGET_SYSTEM} alias. * *

Why it fails instead of skipping when the journal is off

* With {@code Features.JOURNAL_EVENTS} disabled no journal event has ever been written, so the audit ledger @@ -67,6 +51,8 @@ */ public class AuditCleanupChange { + public static final String CHANGE_ID = "compact-audit-store"; + private static final Logger logger = FlamingockLoggerFactory.getLogger("AuditCleanup"); /** diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/CoreSystemChangeFilter.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/CoreSystemChangeFilter.java new file mode 100644 index 000000000..e66dffdcf --- /dev/null +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/CoreSystemChangeFilter.java @@ -0,0 +1,57 @@ +/* + * 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.internal.core.system; + +import io.flamingock.internal.common.core.feature.Features; +import io.flamingock.internal.core.change.filter.ChangeFilter; +import io.flamingock.internal.core.change.loaded.AbstractLoadedChange; +import io.flamingock.internal.util.FeatureFlag; + +/** + * Excludes the journal-backfill and audit-cleanup system changes from the runtime pipeline while + * {@link Features#JOURNAL_EVENTS} is disabled. + *

+ * They are contributed unconditionally at compile time - {@code CoreSystemChangeProvider} has no way to + * see the runtime flag, since it runs inside javac, a different process. This filter is the runtime-side + * half: it runs in the same process where {@link FeatureFlag} state actually lives, so it can keep them + * out of the executable pipeline while the flag is off. + *

+ * Excluding rather than letting them run and throw (their own fail-fast behavior, kept for when they are + * genuinely in the pipeline but something is otherwise wrong) matters because of what "ran and threw" does + * to the audit store: nothing, since the failure is thrown before any audit write - but letting every + * application startup reach that throw would turn an internal, off-by-default flag into a hard failure for + * every application, which defeats the point of it being off by default. Excluded here, the change simply + * never attempts to run and never gets an audit record, so the first startup after the flag is finally + * enabled sees it for the first time and runs it then - never "skipped but recorded forever". + */ +public final class CoreSystemChangeFilter implements ChangeFilter { + + private static final CoreSystemChangeFilter INSTANCE = new CoreSystemChangeFilter(); + + public static CoreSystemChangeFilter instance() { + return INSTANCE; + } + + private CoreSystemChangeFilter() { + } + + @Override + public boolean filter(AbstractLoadedChange change) { + String id = change.getId(); + boolean gated = JournalBackfillChange.CHANGE_ID.equals(id) || AuditCleanupChange.CHANGE_ID.equals(id); + return !gated || FeatureFlag.isEnabled(Features.JOURNAL_EVENTS); + } +} diff --git a/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/JournalBackfillChange.java b/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/JournalBackfillChange.java index f93592841..5cf3a5659 100644 --- a/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/JournalBackfillChange.java +++ b/core/flamingock-core/src/main/java/io/flamingock/internal/core/system/JournalBackfillChange.java @@ -17,6 +17,7 @@ import io.flamingock.api.annotations.NonLockGuarded; import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditPersistence; import io.flamingock.internal.common.core.audit.AuditPersistenceFactory; import io.flamingock.internal.common.core.error.FlamingockException; import io.flamingock.internal.common.core.feature.Features; @@ -33,15 +34,20 @@ /** * Copies historical audit entries into local journal streams without modifying audit history. - * Registration and execution order are separate concerns; this change only retains its identity to - * exclude its own audit records. + * This change only retains its identity to exclude its own audit records. + *

+ * Contributed unconditionally by {@code CoreSystemChangeProvider}, consumed by + * {@code FlamingockAnnotationProcessorPlugin#findAnnotatedChanges()} with {@code setSystem(true)} and + * {@code order = "00200"} — after the Mongock importer's {@code "00100"} and before audit cleanup's + * {@code "00300"}. Resolved against the reserved {@code PipelineHelper#AUDIT_STORE_TARGET_SYSTEM} target + * system, since this change operates on the audit store itself. */ public class JournalBackfillChange { public static final String CHANGE_ID = "backfill-audit-history-to-journal"; private static final Logger logger = FlamingockLoggerFactory.getLogger("JournalBackfill"); - private final Map journalPersistenceByStage = new HashMap<>(); + private final Map persistenceByStage = new HashMap<>(); private AuditPersistenceFactory persistenceFactory; /** @@ -62,11 +68,17 @@ public void journalBackfill(@NonLockGuarded AuditPersistenceFactory persisten logger.info("Backfilling audit history into journal streams"); this.persistenceFactory = persistenceFactory; - for (AuditEntry auditEntry : persistenceFactory.get(SYSTEM_STAGE_ID).getAuditHistory()) { + // Routed through the same cache as every per-entry lookup below - not just the general + // "one less instantiation" saving the review comment on PR #982 flagged, but required for + // correctness on DynamoDB: a persistence instance owns a journal-sequence baseline, and a second, + // independently-built instance for this same stage starts from a stale baseline, handing out a + // sequence number a later write (e.g. the next system change's own STARTED entry) has already + // consumed - a hard DynamoDB transaction conflict, not a mere inefficiency. + for (AuditEntry auditEntry : getPersistence(SYSTEM_STAGE_ID).getAuditHistory()) { if (CHANGE_ID.equals(auditEntry.getChangeId())) { continue; } - JournalHistoryAppender persistence = getJournalPersistence(auditEntry); + JournalHistoryAppender persistence = (JournalHistoryAppender) getPersistence(auditEntry.getStageId()); Result result = persistence.appendEventFrom(auditEntry); if (!(result instanceof Result.Ok)) { Throwable cause = result instanceof Result.Error ? ((Result.Error) result).getError() : null; @@ -76,8 +88,7 @@ public void journalBackfill(@NonLockGuarded AuditPersistenceFactory persisten logger.info("Audit history backfilled into journal streams"); } - private JournalHistoryAppender getJournalPersistence(AuditEntry auditEntry) { - return journalPersistenceByStage.computeIfAbsent(auditEntry.getStageId(), - stageId -> (JournalHistoryAppender) persistenceFactory.get(stageId)); + private AuditPersistence getPersistence(String stageId) { + return persistenceByStage.computeIfAbsent(stageId, persistenceFactory::get); } } diff --git a/core/flamingock-core/src/test/java/io/flamingock/internal/core/journal/JournalEventSequencerFactoryTest.java b/core/flamingock-core/src/test/java/io/flamingock/internal/core/journal/JournalEventSequencerFactoryTest.java new file mode 100644 index 000000000..fecf56a9f --- /dev/null +++ b/core/flamingock-core/src/test/java/io/flamingock/internal/core/journal/JournalEventSequencerFactoryTest.java @@ -0,0 +1,95 @@ +/* + * 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.internal.core.journal; + +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.journal.JournalEvent; +import io.flamingock.internal.common.core.journal.JournalEventType; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class JournalEventSequencerFactoryTest { + + private final JournalEventReader reader = mock(JournalEventReader.class); + private final AuditEntry payload = mock(AuditEntry.class); + private final JournalEventSequencerFactory factory = new JournalEventSequencerFactory(reader); + + @Test + @DisplayName("shares confirmed sequence progress between consumers of the same stream") + void sharesSequenceProgressBetweenConsumers() { + when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.empty()); + JournalEventSequencer outer = factory.forStream("stage-1"); + assertEquals(1L, outer.newEvent(payload).getStreamSequence()); + outer.confirm(); + + when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.of(event("stage-1", 1L))); + JournalEventSequencer inner = factory.forStream("stage-1"); + assertEquals(2L, inner.newEvent(payload).getStreamSequence()); + inner.confirm(); + + assertEquals(3L, outer.newEvent(payload).getStreamSequence()); + outer.confirm(); + assertSame(outer, inner); + assertSame(outer, factory.forStream("stage-1")); + assertEquals(4L, factory.forStream("stage-1").newEvent(payload).getStreamSequence()); + } + + @Test + @DisplayName("seeds independent streams from their own durable tails") + void seedsIndependentStreamsFromDurableTails() { + when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.of(event("stage-1", 7L))); + when(reader.getLastEventByStream("stage-2")).thenReturn(Optional.empty()); + JournalEventSequencer first = factory.forStream("stage-1"); + JournalEventSequencer second = factory.forStream("stage-2"); + + assertNotSame(first, second); + JournalEvent firstEvent = first.newEvent(payload); + assertEquals("stage-1", firstEvent.getStreamId()); + assertEquals(8L, firstEvent.getStreamSequence()); + first.confirm(); + JournalEvent secondEvent = second.newEvent(payload); + assertEquals("stage-2", secondEvent.getStreamId()); + assertEquals(1L, secondEvent.getStreamSequence()); + assertEquals(9L, first.newEvent(payload).getStreamSequence()); + } + + @Test + @DisplayName("reuses an unconfirmed position through another consumer") + void reusesUnconfirmedPosition() { + when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.empty()); + JournalEventSequencer first = factory.forStream("stage-1"); + assertEquals(1L, first.newEvent(payload).getStreamSequence()); + + JournalEventSequencer second = factory.forStream("stage-1"); + assertEquals(1L, second.newEvent(payload).getStreamSequence()); + second.confirm(); + assertEquals(2L, first.newEvent(payload).getStreamSequence()); + } + + private JournalEvent event(String streamId, long sequence) { + return new JournalEvent<>("event-" + sequence, "key-" + sequence, JournalEventType.CHANGE_STATE, + streamId, sequence, Instant.now(), payload); + } +} diff --git a/core/flamingock-processor/src/main/java/io/flamingock/core/processor/FlamingockAnnotationProcessorPlugin.java b/core/flamingock-processor/src/main/java/io/flamingock/core/processor/FlamingockAnnotationProcessorPlugin.java index 1f696a063..ca8084a07 100644 --- a/core/flamingock-processor/src/main/java/io/flamingock/core/processor/FlamingockAnnotationProcessorPlugin.java +++ b/core/flamingock-processor/src/main/java/io/flamingock/core/processor/FlamingockAnnotationProcessorPlugin.java @@ -16,6 +16,7 @@ package io.flamingock.core.processor; import io.flamingock.api.annotations.Change; +import io.flamingock.core.processor.system.CoreSystemChangeProvider; import io.flamingock.internal.common.core.processor.AnnotationProcessorPlugin; import io.flamingock.internal.common.core.util.LoggerPreProcessor; import io.flamingock.internal.common.core.processor.ChangeDiscoverer; @@ -26,6 +27,7 @@ import javax.lang.model.element.ElementKind; import javax.lang.model.element.TypeElement; import java.util.Collection; +import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.stream.Collectors; @@ -44,13 +46,19 @@ public void initialize(RoundEnvironment roundEnv, LoggerPreProcessor logger) { @Override public Collection findAnnotatedChanges() { logger.info("Searching for code-based changes (Java classes annotated with @Change annotation)"); - return roundEnv.getElementsAnnotatedWith(Change.class) + List changes = roundEnv.getElementsAnnotatedWith(Change.class) .stream() .filter(e -> e.getKind() == ElementKind.CLASS) .map(e -> (TypeElement) e) .map(this::buildCodePreviewChange) .filter(Objects::nonNull) .collect(Collectors.toList()); + + // Unconditional, unlike the Mongock importer: every audit store needs its history preserved to the + // journal and then compacted, regardless of which optional modules are present. + changes.addAll(CoreSystemChangeProvider.getSystemChanges()); + + return changes; } private CodePreviewChange buildCodePreviewChange(TypeElement typeElement) { diff --git a/core/flamingock-processor/src/main/java/io/flamingock/core/processor/system/CoreSystemChangeProvider.java b/core/flamingock-processor/src/main/java/io/flamingock/core/processor/system/CoreSystemChangeProvider.java new file mode 100644 index 000000000..e04472916 --- /dev/null +++ b/core/flamingock-processor/src/main/java/io/flamingock/core/processor/system/CoreSystemChangeProvider.java @@ -0,0 +1,100 @@ +/* + * 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.core.processor.system; + +import io.flamingock.internal.common.core.audit.AuditPersistenceFactory; +import io.flamingock.internal.common.core.change.RecoveryDescriptor; +import io.flamingock.internal.common.core.change.TargetSystemDescriptor; +import io.flamingock.internal.common.core.pipeline.PipelineHelper; +import io.flamingock.internal.common.core.preview.CodePreviewChange; +import io.flamingock.internal.common.core.preview.PreviewConstructor; +import io.flamingock.internal.common.core.preview.PreviewMethod; +import io.flamingock.internal.common.core.preview.builder.CodePreviewChangeBuilder; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * Contributes the core system changes that must run on every application startup, unconditionally - unlike + * the Mongock importer, they are not discovered from a user annotation, because they do not depend on any + * module the user chose to add: every audit store needs its history preserved to the journal, then + * compacted to one record per change. + *

+ * Deliberately references {@code io.flamingock.internal.core.system.JournalBackfillChange}, + * {@code AuditCleanupChange} and {@code AuditCompactor} by fully-qualified string name rather than by + * class literal. {@code flamingock-processor} must stay free of a compile dependency on + * {@code flamingock-core}: this module's own compile classpath doubles as every consuming project's + * annotation-processor classpath, so pulling flamingock-core in here would leak it onto that classpath for + * every project that merely applies the processor - changing what else the processor can see and resolve + * while it runs (see {@code MultiModuleProcessingIT#builderClass} for a validator that depends on + * flamingock-core being absent from exactly this classpath). The identifiers below must be kept in sync by + * hand with their source classes. + *

+ * Reserved order sequence: Mongock import {@code "00100"} (conditional, contributed by + * {@code MongockAnnotationProcessorPlugin} when {@code mongock-support} is present), journal backfill + * {@code "00200"}, audit cleanup {@code "00300"}. Both changes here resolve against the reserved + * {@link PipelineHelper#AUDIT_STORE_TARGET_SYSTEM} alias, since they operate on the audit store itself + * rather than on a target system the user declared. + */ +public final class CoreSystemChangeProvider { + + // Must match io.flamingock.internal.core.system.JournalBackfillChange.CHANGE_ID / .class.getName() + private static final String JOURNAL_BACKFILL_CHANGE_ID = "backfill-audit-history-to-journal"; + private static final String JOURNAL_BACKFILL_CLASS = "io.flamingock.internal.core.system.JournalBackfillChange"; + private static final String JOURNAL_BACKFILL_APPLY_METHOD = "journalBackfill"; + + // Must match io.flamingock.internal.core.system.AuditCleanupChange.CHANGE_ID / .class.getName() + private static final String AUDIT_CLEANUP_CHANGE_ID = "compact-audit-store"; + private static final String AUDIT_CLEANUP_CLASS = "io.flamingock.internal.core.system.AuditCleanupChange"; + private static final String AUDIT_CLEANUP_APPLY_METHOD = "compactAuditStore"; + + // Must match io.flamingock.internal.core.external.store.AuditCompactor.class.getName() + private static final String AUDIT_COMPACTOR_CLASS = "io.flamingock.internal.core.external.store.AuditCompactor"; + + private static final TargetSystemDescriptor AUDIT_STORE_TARGET_SYSTEM = + new TargetSystemDescriptor(PipelineHelper.AUDIT_STORE_TARGET_SYSTEM); + + private CoreSystemChangeProvider() { + } + + public static List getSystemChanges() { + return Arrays.asList( + buildSystemChange(JOURNAL_BACKFILL_CHANGE_ID, "00200", JOURNAL_BACKFILL_CLASS, + JOURNAL_BACKFILL_APPLY_METHOD, AuditPersistenceFactory.class.getName()), + buildSystemChange(AUDIT_CLEANUP_CHANGE_ID, "00300", AUDIT_CLEANUP_CLASS, + AUDIT_CLEANUP_APPLY_METHOD, AUDIT_COMPACTOR_CLASS)); + } + + private static CodePreviewChange buildSystemChange(String id, + String order, + String sourceClass, + String applyMethod, + String applyMethodParam) { + CodePreviewChangeBuilder builder = CodePreviewChangeBuilder.instance(); + builder.setId(id); + builder.setOrder(order); + builder.setAuthor("flamingock-team"); + builder.setTargetSystem(AUDIT_STORE_TARGET_SYSTEM); + builder.setTransactionalFlag(true); + builder.setSystem(true); + builder.setRecovery(RecoveryDescriptor.getDefault()); + builder.setSourceClassPath(sourceClass); + builder.setConstructor(PreviewConstructor.getDefault()); + builder.setApplyMethod(new PreviewMethod(applyMethod, Collections.singletonList(applyMethodParam))); + return builder.build(); + } +} diff --git a/core/flamingock-processor/src/main/java/io/flamingock/core/processor/util/FlamingockMetadataMerger.java b/core/flamingock-processor/src/main/java/io/flamingock/core/processor/util/FlamingockMetadataMerger.java index 91232c00f..1036799e1 100644 --- a/core/flamingock-processor/src/main/java/io/flamingock/core/processor/util/FlamingockMetadataMerger.java +++ b/core/flamingock-processor/src/main/java/io/flamingock/core/processor/util/FlamingockMetadataMerger.java @@ -219,7 +219,7 @@ public static boolean pruneDeletedClasses(FlamingockMetadata target, if (target.getOrphanChanges() != null && !target.getOrphanChanges().isEmpty()) { List kept = new ArrayList<>(); for (CodePreviewChange c : target.getOrphanChanges()) { - if (c.getSource() != null && typeExists.test(c.getSource())) { + if (isResolvable(c.getSource(), typeExists)) { kept.add(c); } else { pruned = true; @@ -383,6 +383,23 @@ private static void appendToStage(PreviewStage stage, CodePreviewChange change) stage.setChanges(next); } + /** + * Package of the core-contributed system changes ({@code CoreSystemChangeProvider}) - journal backfill + * and audit cleanup. Exempt from the resolvability check below: unlike a user's own {@code @Change} + * class, their source lives in {@code flamingock-core}, not in the module being compiled, so a module + * that depends only on {@code flamingock-core-api} (common in multi-module projects where one module + * declares changes and another assembles the runner) would never resolve them and see them wrongly + * pruned as "deleted" on every round. They need no staleness detection anyway: unlike a deleted + * {@code @Change} class, which stops appearing in-round, these are re-contributed unconditionally by + * every round of every module, so they are never actually stale. + */ + private static final String CORE_SYSTEM_CHANGE_PACKAGE = "io.flamingock.internal.core.system."; + + /** Resolvable if its class exists in the compiled module, or it's a core system change - see above. */ + private static boolean isResolvable(String src, java.util.function.Predicate typeExists) { + return src != null && (src.startsWith(CORE_SYSTEM_CHANGE_PACKAGE) || typeExists.test(src)); + } + private static boolean pruneStage(PreviewStage stage, java.util.function.Predicate typeExists) { Collection current = stage.getChanges(); if (current == null || current.isEmpty()) return false; @@ -390,8 +407,7 @@ private static boolean pruneStage(PreviewStage stage, java.util.function.Predica boolean pruned = false; for (AbstractPreviewChange c : current) { if (c instanceof CodePreviewChange) { - String src = c.getSource(); - if (src == null || !typeExists.test(src)) { + if (!isResolvable(c.getSource(), typeExists)) { pruned = true; continue; } diff --git a/core/flamingock-processor/src/test/java/io/flamingock/core/processor/IncrementalProcessingIT.java b/core/flamingock-processor/src/test/java/io/flamingock/core/processor/IncrementalProcessingIT.java index 84ef50a2d..23f0f3515 100644 --- a/core/flamingock-processor/src/test/java/io/flamingock/core/processor/IncrementalProcessingIT.java +++ b/core/flamingock-processor/src/test/java/io/flamingock/core/processor/IncrementalProcessingIT.java @@ -60,6 +60,9 @@ */ class IncrementalProcessingIT { + /** The journal-backfill and audit-cleanup system changes, present unconditionally alongside user changes. */ + private static final int CORE_SYSTEM_CHANGES = 2; + @TempDir Path workDir; @@ -81,7 +84,7 @@ void incrementalChangeOnlyPreservesExistingMetadata() throws Exception { Path metadataR1 = metadataPathIn(round1Out); assertTrue(Files.exists(metadataR1), "round 1 must produce metadata.json"); FlamingockMetadata afterR1 = readMetadata(metadataR1); - assertEquals(2, totalChanges(afterR1.getPipeline()), + assertEquals(2 + CORE_SYSTEM_CHANGES, totalChanges(afterR1.getPipeline()), "round 1 must record both changes"); // ROUND 2: simulate incremental compile of only ChangeOne. Pre-seed CLASS_OUTPUT @@ -96,7 +99,7 @@ void incrementalChangeOnlyPreservesExistingMetadata() throws Exception { assertTrue(ok2, "round 2 must succeed even without @EnableFlamingock in the round"); FlamingockMetadata afterR2 = readMetadata(metadataPathIn(round2Out)); - assertEquals(2, totalChanges(afterR2.getPipeline()), + assertEquals(2 + CORE_SYSTEM_CHANGES, totalChanges(afterR2.getPipeline()), "round 2 must preserve previously-known change id-2"); List ids = changeIds(afterR2.getPipeline()); assertTrue(ids.contains("id-1"), "id-1 (recompiled) present"); @@ -135,8 +138,10 @@ void changeWithoutExistingMetadataIsHeldAsOrphan() throws Exception { FlamingockMetadata metadata = readMetadata(metadataPath); assertNull(metadata.getPipeline(), "no pipeline yet — only orphans"); assertNotNull(metadata.getOrphanChanges()); - assertEquals(1, metadata.getOrphanChanges().size()); - assertEquals("id-x", metadata.getOrphanChanges().get(0).getId()); + // With no @EnableFlamingock yet, the core system changes have nowhere to be routed + // either, so they land as orphans alongside the user's change. + assertEquals(1 + CORE_SYSTEM_CHANGES, metadata.getOrphanChanges().size()); + assertTrue(metadata.getOrphanChanges().stream().anyMatch(c -> "id-x".equals(c.getId()))); assertFalse(metadata.isStrictStageMapping(), "strictStageMapping defaults to false until @EnableFlamingock sets it"); } @@ -154,7 +159,7 @@ void orphanFromRound1IsRehomedByRound2() throws Exception { boolean ok1 = compile(src, round1Out, Collections.singletonList(changeSrc)); assertTrue(ok1); FlamingockMetadata r1 = readMetadata(metadataPathIn(round1Out)); - assertEquals(1, r1.getOrphanChanges().size()); + assertEquals(1 + CORE_SYSTEM_CHANGES, r1.getOrphanChanges().size()); // Round 2: introduce Config.java with covering stage; only Config + previous metadata. writeFile(src, "com/example/Config.java", configClass()); @@ -170,7 +175,7 @@ void orphanFromRound1IsRehomedByRound2() throws Exception { assertNotNull(r2.getPipeline(), "round 2 builds the pipeline"); assertTrue(r2.getOrphanChanges() == null || r2.getOrphanChanges().isEmpty(), "orphan should have been rehomed into the new stage"); - assertEquals(1, totalChanges(r2.getPipeline())); + assertEquals(1 + CORE_SYSTEM_CHANGES, totalChanges(r2.getPipeline())); assertTrue(changeIds(r2.getPipeline()).contains("id-x")); } @@ -190,7 +195,7 @@ void lastRoundPrunesDeletedChange() throws Exception { boolean ok1 = compile(src, round1Out, allJavaFilesIn(src)); assertTrue(ok1); FlamingockMetadata r1 = readMetadata(metadataPathIn(round1Out)); - assertEquals(2, totalChanges(r1.getPipeline())); + assertEquals(2 + CORE_SYSTEM_CHANGES, totalChanges(r1.getPipeline())); // Round 2: simulate deletion — remove the source file. Recompile remaining sources + // seed previous metadata. Last-round pruning should drop id-del. @@ -202,7 +207,7 @@ void lastRoundPrunesDeletedChange() throws Exception { assertTrue(ok2); FlamingockMetadata r2 = readMetadata(metadataPathIn(round2Out)); - assertEquals(1, totalChanges(r2.getPipeline()), + assertEquals(1 + CORE_SYSTEM_CHANGES, totalChanges(r2.getPipeline()), "deleted @Change should be pruned at the last round"); assertTrue(changeIds(r2.getPipeline()).contains("id-keep")); assertFalse(changeIds(r2.getPipeline()).contains("id-del")); @@ -225,7 +230,7 @@ void pruneHandlesNestedClassBinaryNames() throws Exception { FlamingockMetadata md = readMetadata(metadataPathIn(out)); // Without the fix, the prune step would wrongly drop the nested change here. - assertEquals(1, totalChanges(md.getPipeline()), + assertEquals(1 + CORE_SYSTEM_CHANGES, totalChanges(md.getPipeline()), "nested @Change must NOT be pruned (binary name with $ converts to canonical for getTypeElement)"); assertTrue(changeIds(md.getPipeline()).contains("id-nested")); // Confirm the persisted source carries the binary form, so the conversion is exercised. @@ -249,7 +254,7 @@ void cleanBuildPlacesChangeInSubPackageOfStage() throws Exception { assertTrue(ok); FlamingockMetadata md = readMetadata(metadataPathIn(out)); - assertEquals(1, totalChanges(md.getPipeline()), + assertEquals(1 + CORE_SYSTEM_CHANGES, totalChanges(md.getPipeline()), "sub-package change must be placed via covers semantics, not dropped"); assertTrue(changeIds(md.getPipeline()).contains("id-sub")); assertTrue(md.getOrphanChanges() == null || md.getOrphanChanges().isEmpty()); @@ -270,7 +275,7 @@ void droppedStageMovesChangesToOrphans() throws Exception { boolean ok1 = compile(src, round1Out, allJavaFilesIn(src)); assertTrue(ok1); FlamingockMetadata r1 = readMetadata(metadataPathIn(round1Out)); - assertEquals(2, totalChanges(r1.getPipeline())); + assertEquals(2 + CORE_SYSTEM_CHANGES, totalChanges(r1.getPipeline())); // Round 2: rewrite Config.java to drop the 'removed' stage (keeping 'kept'). // Recompile all sources — Gradle does this when a non-isolating processor's input changes. @@ -283,8 +288,9 @@ void droppedStageMovesChangesToOrphans() throws Exception { assertTrue(ok2); FlamingockMetadata r2 = readMetadata(metadataPathIn(round2Out)); - // 'kept' stage covers com.example.kept → id-keep stays placed there. - assertEquals(1, totalChanges(r2.getPipeline())); + // 'kept' stage covers com.example.kept → id-keep stays placed there, alongside the + // core system changes, which are unaffected by the stage drop. + assertEquals(1 + CORE_SYSTEM_CHANGES, totalChanges(r2.getPipeline())); assertTrue(changeIds(r2.getPipeline()).contains("id-keep")); // 'removed' is gone; id-rm has no covering stage → ends up in orphans. assertNotNull(r2.getOrphanChanges()); diff --git a/core/flamingock-processor/src/test/java/io/flamingock/core/processor/MultiModuleProcessingIT.java b/core/flamingock-processor/src/test/java/io/flamingock/core/processor/MultiModuleProcessingIT.java index 907a393f8..c17a77386 100644 --- a/core/flamingock-processor/src/test/java/io/flamingock/core/processor/MultiModuleProcessingIT.java +++ b/core/flamingock-processor/src/test/java/io/flamingock/core/processor/MultiModuleProcessingIT.java @@ -101,7 +101,9 @@ void twoModulesAggregateIntoComposite() throws Exception { List stageNames = stageNames(composite.getPipeline()); assertTrue(stageNames.contains("changes"), "the 'changes' stage from at least one module should appear: " + stageNames); - assertEquals(2, totalChanges(composite.getPipeline()), + // +2: the two core system changes, contributed unconditionally by each module and + // deduplicated to one pair in the composite system stage. + assertEquals(4, totalChanges(composite.getPipeline()), "composite contains both modules' changes"); List ids = changeIds(composite.getPipeline()); assertTrue(ids.contains("id-A1")); diff --git a/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterJournalTest.java b/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterJournalTest.java deleted file mode 100644 index d7549f9c9..000000000 --- a/legacy/mongock-importer-dynamodb/src/test/java/io/flamingock/importer/mongock/dynamodb/DynamoDBImporterJournalTest.java +++ /dev/null @@ -1,216 +0,0 @@ -/* - * 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.importer.mongock.dynamodb; - -import io.flamingock.api.annotations.EnableFlamingock; -import io.flamingock.api.annotations.Stage; -import io.flamingock.common.test.mongock.MongockChangeEntry; -import io.flamingock.common.test.mongock.MongockChangeState; -import io.flamingock.common.test.mongock.MongockChangeType; -import io.flamingock.core.kit.TestKit; -import io.flamingock.dynamodb.kit.DynamoDBTableFactory; -import io.flamingock.dynamodb.kit.DynamoDBTestKit; -import io.flamingock.internal.common.core.audit.AuditEntry; -import io.flamingock.internal.common.core.feature.Features; -import io.flamingock.internal.common.core.journal.JournalEvent; -import io.flamingock.internal.util.FeatureFlag; -import io.flamingock.internal.util.constants.CommunityPersistenceConstants; -import io.flamingock.internal.util.dynamodb.DynamoDBUtil; -import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity; -import io.flamingock.internal.util.dynamodb.entities.journal.DynamoDBJournalEventMapper; -import io.flamingock.internal.util.dynamodb.entities.journal.JournalEventEntity; -import io.flamingock.store.dynamodb.DynamoDBAuditStore; -import io.flamingock.support.mongock.annotations.MongockSupport; -import io.flamingock.targetsystem.dynamodb.DynamoDBTargetSystem; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.DisplayName; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.GenericContainer; -import org.testcontainers.junit.jupiter.Container; -import org.testcontainers.junit.jupiter.Testcontainers; -import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; -import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; -import software.amazon.awssdk.enhanced.dynamodb.TableSchema; -import software.amazon.awssdk.enhanced.dynamodb.model.ScanEnhancedRequest; -import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; -import software.amazon.awssdk.regions.Region; -import software.amazon.awssdk.services.dynamodb.DynamoDbClient; - -import java.net.URI; -import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneId; -import java.util.Arrays; -import java.util.Date; -import java.util.List; -import java.util.stream.Collectors; - -import static io.flamingock.internal.common.core.metadata.Constants.DEFAULT_MONGOCK_ORIGIN; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; - -@Testcontainers -@MongockSupport(targetSystem = "dynamodb-target-system") -@EnableFlamingock(stages = {@Stage(location = "io.flamingock.importer.mongock.dynamodb.changes")}) -class DynamoDBImporterJournalTest { - - private static final String LEGACY_STAGE_ID = "flamingock-legacy-stage"; - private static final String IMPORTED_BEFORE_CHANGE_ID = "create-users-table_before"; - private static final String IMPORTED_CHANGE_ID = "create-users-table"; - private static final Instant HISTORICAL_TIMESTAMP = Instant.parse("2025-06-19T05:43:57.132Z"); - - @Container - static final GenericContainer dynamoDBContainer = new GenericContainer<>("amazon/dynamodb-local:latest") - .withExposedPorts(8000); - - private DynamoDbClient client; - private DynamoDBTargetSystem targetSystem; - private DynamoDBMongockTestHelper mongockTestHelper; - private TestKit testKit; - - @BeforeEach - void setUp() { - String endpoint = String.format("http://%s:%d", - dynamoDBContainer.getHost(), - dynamoDBContainer.getMappedPort(8000)); - client = DynamoDbClient.builder() - .endpointOverride(URI.create(endpoint)) - .region(Region.US_EAST_1) - .httpClient(UrlConnectionHttpClient.builder().build()) - .credentialsProvider(StaticCredentialsProvider.create( - AwsBasicCredentials.create("dummy", "dummy"))) - .build(); - - DynamoDBTableFactory.createMongockTable(client, DEFAULT_MONGOCK_ORIGIN); - mongockTestHelper = new DynamoDBMongockTestHelper(client, DEFAULT_MONGOCK_ORIGIN); - targetSystem = new DynamoDBTargetSystem("dynamodb-target-system", client); - testKit = DynamoDBTestKit.create( - client, - DynamoDBAuditStore.from(targetSystem)); - } - - @AfterEach - void tearDown() { - FeatureFlag.remove(Features.JOURNAL_EVENTS); - mongockTestHelper.reset(); - testKit.cleanUp(); - client.close(); - } - - @Test - @DisplayName("journal-enabled imports retain current audit records and publish full payloads on the legacy stream") - void journalEnabledImportKeepsCurrentAuditRecordsAndUsesLegacyStream() { - FeatureFlag.enable(Features.JOURNAL_EVENTS); - mongockTestHelper.write(new MongockChangeEntry( - "legacy-execution-before", - IMPORTED_BEFORE_CHANGE_ID, - "mongock", - Date.from(Instant.parse("2025-06-19T05:43:57.094Z")), - MongockChangeState.EXECUTED, - MongockChangeType.BEFORE_EXECUTION, - "io.mongock.examples.mongodb.standalone.mondogb.sync.migration.initializer.ClientInitializerChangeUnit", - "beforeExecution", - "legacy-before-metadata", - 25L, - "legacy-host", - null, - false, - null)); - mongockTestHelper.write(new MongockChangeEntry( - "legacy-execution", - IMPORTED_CHANGE_ID, - "mongock", - Date.from(HISTORICAL_TIMESTAMP), - MongockChangeState.EXECUTED, - MongockChangeType.EXECUTION, - "io.mongock.examples.mongodb.standalone.mondogb.sync.migration.initializer.ClientInitializerChangeUnit", - "apply", - "legacy-metadata", - 23L, - "legacy-host", - null, - false, - null)); - - Instant runStartedAt = Instant.now(); - testKit.createBuilder() - .addTargetSystem(targetSystem) - .build() - .run(); - Instant runFinishedAt = Instant.now(); - - List importedAuditRecords = storedAuditRecords().stream() - .filter(entry -> Arrays.asList(IMPORTED_BEFORE_CHANGE_ID, IMPORTED_CHANGE_ID) - .contains(entry.getChangeId())) - .collect(Collectors.toList()); - assertEquals(2, importedAuditRecords.size()); - assertTrue(importedAuditRecords.stream().anyMatch(entry -> - IMPORTED_BEFORE_CHANGE_ID.equals(entry.getPartitionKey()))); - assertTrue(importedAuditRecords.stream().anyMatch(entry -> - IMPORTED_CHANGE_ID.equals(entry.getPartitionKey())), - "flag ON must use the changeId-only current-state key for imported audit entries"); - - List> importedEvents = storedEvents().stream() - .filter(event -> Arrays.asList(IMPORTED_BEFORE_CHANGE_ID, IMPORTED_CHANGE_ID) - .contains(event.getData().getChangeId())) - .collect(Collectors.toList()); - assertEquals(2, importedEvents.size()); - assertTrue(importedEvents.stream().allMatch(event -> LEGACY_STAGE_ID.equals(event.getStreamId()))); - assertTrue(importedEvents.stream().allMatch(event -> - !event.getData().getStageId().equals(event.getStreamId()))); - assertEquals(Arrays.asList(1L, 2L), importedEvents.stream() - .map(JournalEvent::getStreamSequence) - .sorted() - .collect(Collectors.toList())); - - JournalEvent importedEvent = importedEvents.stream() - .filter(event -> IMPORTED_CHANGE_ID.equals(event.getData().getChangeId())) - .findFirst() - .orElseThrow(() -> new AssertionError("Expected the imported execution event")); - AuditEntry importedEntry = importedEvent.getData(); - assertEquals(LocalDateTime.ofInstant(HISTORICAL_TIMESTAMP, ZoneId.systemDefault()), - importedEntry.getCreatedAt()); - assertEquals("legacy-execution", importedEntry.getExecutionId()); - assertEquals("legacy-metadata", importedEntry.getMetadata()); - assertEquals(23L, importedEntry.getExecutionMillis()); - assertEquals("legacy-host", importedEntry.getExecutionHostname()); - assertTrue(!importedEvent.getOccurredAt().isBefore(runStartedAt)); - assertTrue(!importedEvent.getOccurredAt().isAfter(runFinishedAt)); - } - - private List storedAuditRecords() { - return new DynamoDBUtil(client) - .getEnhancedClient() - .table(CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME, - TableSchema.fromBean(AuditEntryEntity.class)) - .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) - .items() - .stream() - .collect(Collectors.toList()); - } - - private List> storedEvents() { - return new DynamoDBUtil(client) - .getEnhancedClient() - .table("flamingockJournalEvents", TableSchema.fromBean(JournalEventEntity.class)) - .scan(ScanEnhancedRequest.builder().consistentRead(true).build()) - .items() - .stream() - .map(DynamoDBJournalEventMapper::fromEntity) - .collect(Collectors.toList()); - } -} diff --git a/legacy/mongock-importer-mongodb/src/test/java/io/flamingock/importer/mongock/mongodb/MongoDBImporterTest.java b/legacy/mongock-importer-mongodb/src/test/java/io/flamingock/importer/mongock/mongodb/MongoDBImporterTest.java index 3d78c038f..305602ff3 100644 --- a/legacy/mongock-importer-mongodb/src/test/java/io/flamingock/importer/mongock/mongodb/MongoDBImporterTest.java +++ b/legacy/mongock-importer-mongodb/src/test/java/io/flamingock/importer/mongock/mongodb/MongoDBImporterTest.java @@ -118,6 +118,23 @@ void GIVEN_allMongockChangeUnitsAlreadyExecuted_WHEN_migratingToFlamingockCommun flamingock.run(); + List persistedSnapshot = auditHelper.getAuditEntriesSorted().stream() + .map(entry -> entry.getChangeId() + "|" + entry.getState() + "|" + + entry.getType() + "|" + entry.getStageId()) + .collect(java.util.stream.Collectors.toList()); + assertEquals(java.util.Arrays.asList( + "system-change-00001_before|APPLIED|MONGOCK_BEFORE|flamingock-legacy-stage", + "system-change-00001|APPLIED|MONGOCK_EXECUTION|flamingock-legacy-stage", + "mongock-change-1_before|APPLIED|MONGOCK_BEFORE|flamingock-legacy-stage", + "mongock-change-1|APPLIED|MONGOCK_EXECUTION|flamingock-legacy-stage", + "mongock-change-2|APPLIED|MONGOCK_EXECUTION|flamingock-legacy-stage", + "migration-mongock-to-flamingock-community|STARTED|STANDARD_CODE|flamingock-system-stage", + "migration-mongock-to-flamingock-community|APPLIED|STANDARD_CODE|flamingock-system-stage", + "create-users-collection-with-index|STARTED|STANDARD_CODE|changes", + "create-users-collection-with-index|APPLIED|STANDARD_CODE|changes", + "seed-users|STARTED|STANDARD_CODE|changes", + "seed-users|APPLIED|STANDARD_CODE|changes"), persistedSnapshot); + // Verify audit sequence: 11 total entries as shown in actual execution // Final state only: imported and freshly-executed changes alike, once STARTED is filtered out auditHelper.verifyAuditFinalStateSequence( diff --git a/legacy/mongock-support/src/main/java/io/flamingock/support/mongock/MongockImportChange.java b/legacy/mongock-support/src/main/java/io/flamingock/support/mongock/MongockImportChange.java index 6ce28c1bb..74005383a 100644 --- a/legacy/mongock-support/src/main/java/io/flamingock/support/mongock/MongockImportChange.java +++ b/legacy/mongock-support/src/main/java/io/flamingock/support/mongock/MongockImportChange.java @@ -21,12 +21,13 @@ import io.flamingock.internal.common.core.pipeline.PipelineHelper; import io.flamingock.internal.common.core.audit.AuditEntry; import io.flamingock.internal.common.core.audit.AuditHistoryReader; -import io.flamingock.internal.common.core.audit.AuditWriter; import io.flamingock.internal.common.core.error.FlamingockException; import io.flamingock.internal.common.core.pipeline.PipelineDescriptor; +import io.flamingock.internal.core.external.store.AuditHistoryAppender; import io.flamingock.internal.core.external.targets.TargetSystemManager; import io.flamingock.internal.core.external.targets.operations.TargetSystemOps; import io.flamingock.internal.core.external.targets.operations.TransactionalTargetSystemOps; +import io.flamingock.internal.util.Result; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -65,7 +66,8 @@ public void importHistory(@Named("change.targetSystem.id") String targetSystemId List legacyHistory = legacyHistoryReader.getAuditHistory(); boolean ignoreUnknownEntries = resolveIgnoreUnknownEntries(ignoreUnknownEntriesPropertyValue); validate(legacyHistory, targetSystemId, emptyOriginAllowedPropertyValue); - AuditWriter auditWriter = auditPersistenceFactory.get(PipelineHelper.LEGACY_STAGE_ID); + AuditHistoryAppender auditHistoryAppender = + (AuditHistoryAppender) auditPersistenceFactory.get(PipelineHelper.LEGACY_STAGE_ID); legacyHistory.forEach(auditEntryFromOrigin -> { Optional stageId = pipelineHelper.findStageId(auditEntryFromOrigin); if (!stageId.isPresent()) { @@ -84,7 +86,13 @@ public void importHistory(@Named("change.targetSystem.id") String targetSystemId AuditEntry auditEntryWithStageId = auditEntryFromOrigin.copyWithNewIdAndStageId( pipelineHelper.getStorableChangeId(auditEntryFromOrigin), stageId.get()); - auditWriter.writeEntry(auditEntryWithStageId); + Result result = auditHistoryAppender.append(auditEntryWithStageId); + if (result instanceof Result.Error) { + throw new FlamingockException(String.format( + "Error importing audit entry with changeId[%s] into stage[%s] for targetSystem[%s].", + auditEntryWithStageId.getChangeId(), stageId.get(), targetSystemId), + ((Result.Error) result).getError()); + } }); } diff --git a/legacy/mongock-support/src/test/java/io/flamingock/support/mongock/MongockImportChangeTest.java b/legacy/mongock-support/src/test/java/io/flamingock/support/mongock/MongockImportChangeTest.java new file mode 100644 index 000000000..422fea274 --- /dev/null +++ b/legacy/mongock-support/src/test/java/io/flamingock/support/mongock/MongockImportChangeTest.java @@ -0,0 +1,68 @@ +/* + * 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.support.mongock; + +import io.flamingock.internal.common.core.audit.AuditEntry; +import io.flamingock.internal.common.core.audit.AuditPersistenceFactory; +import io.flamingock.internal.common.core.audit.AuditHistoryReader; +import io.flamingock.internal.common.core.error.FlamingockException; +import io.flamingock.internal.common.core.pipeline.PipelineDescriptor; +import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence; +import io.flamingock.internal.core.external.targets.TargetSystemManager; +import io.flamingock.internal.core.external.targets.operations.TransactionalTargetSystemOps; +import io.flamingock.internal.util.Result; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.Optional; + +import static io.flamingock.internal.common.core.audit.AuditReaderType.MONGOCK; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class MongockImportChangeTest { + @Test + void failedAppendIncludesEntryIdentityAndOriginalCause() { + TargetSystemManager manager = mock(TargetSystemManager.class); + TransactionalTargetSystemOps ops = mock(TransactionalTargetSystemOps.class); + AuditHistoryReader reader = mock(AuditHistoryReader.class); + PipelineDescriptor pipeline = mock(PipelineDescriptor.class); + CommunityAuditPersistence persistence = mock(CommunityAuditPersistence.class); + AuditPersistenceFactory factory = stageId -> { + assertEquals("flamingock-legacy-stage", stageId); + return persistence; + }; + AuditEntry entry = mock(AuditEntry.class); + RuntimeException cause = new RuntimeException("write failed"); + when(manager.getTargetSystem("mongo")).thenReturn(ops); + when(ops.getAuditAuditReader(MONGOCK)).thenReturn(Optional.of(reader)); + when(reader.getAuditHistory()).thenReturn(Collections.singletonList(entry)); + when(entry.getChangeId()).thenReturn("change-1"); + when(entry.copyWithNewIdAndStageId("change-1", "stage-1")).thenReturn(entry); + when(pipeline.getStageByChange("change-1")).thenReturn(Optional.of("stage-1")); + when(persistence.append(any(AuditEntry.class))).thenReturn(new Result.Error(cause)); + + FlamingockException failure = assertThrows(FlamingockException.class, + () -> new MongockImportChange().importHistory("mongo", manager, factory, pipeline, null, null, null)); + assertSame(cause, failure.getCause()); + assertEquals("Error importing audit entry with changeId[change-1] into stage[stage-1] for targetSystem[mongo].", + failure.getMessage()); + } +}