diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/PublishedShadowSnapshot.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/PublishedShadowSnapshot.java new file mode 100644 index 0000000..fda09e0 --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/PublishedShadowSnapshot.java @@ -0,0 +1,20 @@ +package com.ruoyi.algorithm.shadow; + +import java.time.Instant; +import java.util.Objects; + +/** + * Immutable metadata identifying one completed shadow read-model generation. + */ +public record PublishedShadowSnapshot( + long generation, + String warehouseId, + Instant synchronizedAt, + String sourceSignature) { + + public PublishedShadowSnapshot { + Objects.requireNonNull(warehouseId, "warehouseId"); + Objects.requireNonNull(synchronizedAt, "synchronizedAt"); + Objects.requireNonNull(sourceSignature, "sourceSignature"); + } +} diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSkipReason.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSkipReason.java new file mode 100644 index 0000000..62cee42 --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSkipReason.java @@ -0,0 +1,9 @@ +package com.ruoyi.algorithm.shadow; + +public enum ShadowSkipReason { + NOT_READY, + BUSY, + STALE, + WAREHOUSE, + CONCURRENT +} diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinator.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinator.java new file mode 100644 index 0000000..400a41d --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinator.java @@ -0,0 +1,244 @@ +package com.ruoyi.algorithm.shadow; + +import java.time.Duration; +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Coordinates one JVM-local published snapshot without making the Legacy path wait. + */ +public final class ShadowSnapshotCoordinator { + + private static final Duration DEFAULT_SNAPSHOT_MAX_AGE = Duration.ofSeconds(60); + private static final long REFRESH_LEASE = Long.MIN_VALUE; + + private final String configuredWarehouseId; + private final Duration snapshotMaxAge; + private final AtomicReference state = new AtomicReference<>( + new State(Phase.INITIALIZING, 0L, 0, null, null, null, null)); + + public ShadowSnapshotCoordinator(String configuredWarehouseId) { + this(configuredWarehouseId, DEFAULT_SNAPSHOT_MAX_AGE); + } + + public ShadowSnapshotCoordinator(String configuredWarehouseId, Duration snapshotMaxAge) { + this.configuredWarehouseId = Objects.requireNonNull(configuredWarehouseId, "configuredWarehouseId").trim(); + this.snapshotMaxAge = Objects.requireNonNull(snapshotMaxAge, "snapshotMaxAge"); + if (snapshotMaxAge.isNegative() || snapshotMaxAge.isZero()) { + throw new IllegalArgumentException("snapshotMaxAge must be positive"); + } + } + + public enum Phase { + INITIALIZING, + READY, + BUSY, + UNAVAILABLE + } + + public Admission enter(Instant now) { + Objects.requireNonNull(now, "now"); + while (true) { + State current = state.get(); + long nextEpoch = current.epoch() + 1; + int nextActiveCount = current.activeLegacyCount() + 1; + boolean canLease = current.phase() == Phase.READY + && current.activeLegacyCount() == 0 + && current.leasedEpoch() == null; + PublishedShadowSnapshot snapshot = canLease ? current.published() : null; + ShadowSkipReason skipReason = null; + if (canLease && !isFresh(snapshot, now)) { + skipReason = ShadowSkipReason.STALE; + } else if (canLease && !WarehouseIdNormalizer.equivalent( + configuredWarehouseId, snapshot.warehouseId())) { + skipReason = ShadowSkipReason.WAREHOUSE; + } else if (!canLease) { + skipReason = skipReason(current); + } + + boolean accepted = skipReason == null; + Phase nextPhase = current.phase() == Phase.READY ? Phase.BUSY : current.phase(); + Long nextLeasedEpoch = current.leasedEpoch(); + if (accepted) { + nextLeasedEpoch = nextEpoch; + } + Instant nextQuietSince = current.activeLegacyCount() == 0 ? null : current.quietSince(); + State next = new State(nextPhase, nextEpoch, nextActiveCount, current.published(), + nextLeasedEpoch, nextQuietSince, current.lastFailure()); + if (state.compareAndSet(current, next)) { + return new Admission(this, accepted, skipReason, snapshot, nextEpoch); + } + } + } + + public void exit(Admission admission) { + if (admission == null || admission.owner != this + || !admission.exited.compareAndSet(false, true)) { + return; + } + while (true) { + State current = state.get(); + if (current.activeLegacyCount() == 0) { + return; + } + int nextActiveCount = current.activeLegacyCount() - 1; + Instant nextQuietSince = nextActiveCount == 0 ? Instant.now() : current.quietSince(); + State next = new State(current.phase(), current.epoch(), nextActiveCount, current.published(), + current.leasedEpoch(), nextQuietSince, current.lastFailure()); + if (state.compareAndSet(current, next)) { + return; + } + } + } + + public void finishObservation(Admission admission) { + if (admission == null || admission.owner != this || !admission.accepted + || !admission.observationFinished.compareAndSet(false, true)) { + return; + } + while (true) { + State current = state.get(); + if (!Objects.equals(current.leasedEpoch(), admission.epoch)) { + return; + } + State next = new State(current.phase(), current.epoch(), current.activeLegacyCount(), + current.published(), null, current.quietSince(), current.lastFailure()); + if (state.compareAndSet(current, next)) { + return; + } + } + } + + public Optional tryBeginRefresh(Instant now) { + Objects.requireNonNull(now, "now"); + while (true) { + State current = state.get(); + if (current.activeLegacyCount() != 0 || current.leasedEpoch() != null) { + return Optional.empty(); + } + State next = new State(Phase.BUSY, current.epoch(), current.activeLegacyCount(), current.published(), + REFRESH_LEASE, current.quietSince() == null ? now : current.quietSince(), current.lastFailure()); + if (state.compareAndSet(current, next)) { + return Optional.of(new RefreshTicket(this, next)); + } + } + } + + public boolean publish(RefreshTicket ticket, PublishedShadowSnapshot snapshot, Instant now) { + if (ticket == null || ticket.owner != this || snapshot == null || now == null) { + return false; + } + State expected = ticket.state; + State next = new State(Phase.READY, expected.epoch(), expected.activeLegacyCount(), snapshot, + null, expected.quietSince() == null ? now : expected.quietSince(), null); + return state.compareAndSet(expected, next); + } + + public void failRefresh(RefreshTicket ticket, String failure) { + if (ticket == null || ticket.owner != this) { + return; + } + State expected = ticket.state; + State next = new State(Phase.UNAVAILABLE, expected.epoch(), expected.activeLegacyCount(), + expected.published(), null, expected.quietSince(), failure); + state.compareAndSet(expected, next); + } + + public boolean isComparable(Admission admission) { + if (admission == null || admission.owner != this || !admission.accepted + || admission.observationFinished.get()) { + return false; + } + State current = state.get(); + return current.epoch() == admission.epoch + && current.leasedEpoch() != null + && current.leasedEpoch() == admission.epoch + && current.published() == admission.snapshot; + } + + public Phase phase() { + return state.get().phase(); + } + + public int activeLegacyCount() { + return state.get().activeLegacyCount(); + } + + public long epoch() { + return state.get().epoch(); + } + + public Instant quietSince() { + return state.get().quietSince(); + } + + public String lastFailure() { + return state.get().lastFailure(); + } + + public static final class Admission { + private final ShadowSnapshotCoordinator owner; + private final boolean accepted; + private final ShadowSkipReason skipReason; + private final PublishedShadowSnapshot snapshot; + private final long epoch; + private final AtomicBoolean exited = new AtomicBoolean(); + private final AtomicBoolean observationFinished = new AtomicBoolean(); + + private Admission(ShadowSnapshotCoordinator owner, boolean accepted, ShadowSkipReason skipReason, + PublishedShadowSnapshot snapshot, long epoch) { + this.owner = owner; + this.accepted = accepted; + this.skipReason = skipReason; + this.snapshot = snapshot; + this.epoch = epoch; + } + + public boolean accepted() { + return accepted; + } + + public ShadowSkipReason skipReason() { + return skipReason; + } + + public PublishedShadowSnapshot snapshot() { + return snapshot; + } + } + + public static final class RefreshTicket { + private final ShadowSnapshotCoordinator owner; + private final State state; + + private RefreshTicket(ShadowSnapshotCoordinator owner, State state) { + this.owner = owner; + this.state = state; + } + } + + private ShadowSkipReason skipReason(State current) { + if (current.phase() != Phase.READY || current.leasedEpoch() != null || current.activeLegacyCount() != 0) { + return current.phase() == Phase.INITIALIZING || current.phase() == Phase.UNAVAILABLE + ? ShadowSkipReason.NOT_READY : ShadowSkipReason.BUSY; + } + return ShadowSkipReason.NOT_READY; + } + + private boolean isFresh(PublishedShadowSnapshot snapshot, Instant now) { + return !snapshot.synchronizedAt().plus(snapshotMaxAge).isBefore(now); + } + + private record State( + Phase phase, + long epoch, + int activeLegacyCount, + PublishedShadowSnapshot published, + Long leasedEpoch, + Instant quietSince, + String lastFailure) { + } +} diff --git a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinatorTest.java b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinatorTest.java new file mode 100644 index 0000000..61801a3 --- /dev/null +++ b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinatorTest.java @@ -0,0 +1,121 @@ +package com.ruoyi.algorithm.shadow; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; + +class ShadowSnapshotCoordinatorTest { + + private static final Instant NOW = Instant.parse("2026-09-03T00:00:00Z"); + + @Test + void startsNotReadyAndAllowsOnlyOneObservationLease() { + ShadowSnapshotCoordinator coordinator = coordinatorAt("01"); + + ShadowSnapshotCoordinator.Admission beforePublish = coordinator.enter(NOW); + assertThat(beforePublish.accepted()).isFalse(); + assertThat(beforePublish.skipReason()).isEqualTo(ShadowSkipReason.NOT_READY); + coordinator.exit(beforePublish); + + publish(coordinator, snapshot(1, "01", NOW)); + ShadowSnapshotCoordinator.Admission first = coordinator.enter(NOW); + ShadowSnapshotCoordinator.Admission second = coordinator.enter(NOW); + + assertThat(first.accepted()).isTrue(); + assertThat(first.snapshot()).isEqualTo(snapshot(1, "01", NOW)); + assertThat(second.accepted()).isFalse(); + assertThat(second.skipReason()).isEqualTo(ShadowSkipReason.BUSY); + + coordinator.finishObservation(first); + coordinator.finishObservation(first); + assertThat(coordinator.activeLegacyCount()).isEqualTo(2); + + coordinator.exit(second); + coordinator.exit(second); + coordinator.exit(first); + coordinator.exit(first); + assertThat(coordinator.activeLegacyCount()).isZero(); + } + + @Test + void concurrentLegacyEntryInvalidatesRefreshPublicationAndComparison() { + ShadowSnapshotCoordinator coordinator = coordinatorAt("01"); + publish(coordinator, snapshot(1, "01", NOW)); + + ShadowSnapshotCoordinator.Admission first = coordinator.enter(NOW); + ShadowSnapshotCoordinator.Admission later = coordinator.enter(NOW); + assertThat(first.accepted()).isTrue(); + assertThat(coordinator.isComparable(first)).isFalse(); + coordinator.exit(later); + coordinator.finishObservation(first); + coordinator.exit(first); + + ShadowSnapshotCoordinator.RefreshTicket refresh = coordinator.tryBeginRefresh(NOW).orElseThrow(); + ShadowSnapshotCoordinator.Admission concurrent = coordinator.enter(NOW); + + assertThat(coordinator.publish(refresh, snapshot(2, "01", NOW), NOW)).isFalse(); + assertThat(coordinator.phase()).isNotEqualTo(ShadowSnapshotCoordinator.Phase.READY); + coordinator.exit(concurrent); + } + + @Test + void skipsStaleSnapshotsAndMismatchedWarehouses() { + ShadowSnapshotCoordinator coordinator = coordinatorAt("01"); + + publish(coordinator, snapshot(3, "01", NOW.minusSeconds(61))); + ShadowSnapshotCoordinator.Admission stale = coordinator.enter(NOW); + assertThat(stale.accepted()).isFalse(); + assertThat(stale.skipReason()).isEqualTo(ShadowSkipReason.STALE); + coordinator.exit(stale); + + publish(coordinator, snapshot(4, "02", NOW)); + ShadowSnapshotCoordinator.Admission wrongWarehouse = coordinator.enter(NOW); + assertThat(wrongWarehouse.accepted()).isFalse(); + assertThat(wrongWarehouse.skipReason()).isEqualTo(ShadowSkipReason.WAREHOUSE); + coordinator.exit(wrongWarehouse); + } + + @Test + void treatsNumericWarehouseIdsAsEquivalent() { + ShadowSnapshotCoordinator coordinator = coordinatorAt("01"); + publish(coordinator, snapshot(5, "1", NOW)); + + ShadowSnapshotCoordinator.Admission admission = coordinator.enter(NOW); + + assertThat(admission.accepted()).isTrue(); + coordinator.exit(admission); + } + + @Test + void failedRefreshBecomesUnavailableUntilAValidPublication() { + ShadowSnapshotCoordinator coordinator = coordinatorAt("01"); + ShadowSnapshotCoordinator.RefreshTicket refresh = coordinator.tryBeginRefresh(NOW).orElseThrow(); + + coordinator.failRefresh(refresh, "signature changed"); + + assertThat(coordinator.phase()).isEqualTo(ShadowSnapshotCoordinator.Phase.UNAVAILABLE); + assertThat(coordinator.lastFailure()).isEqualTo("signature changed"); + ShadowSnapshotCoordinator.Admission unavailable = coordinator.enter(NOW); + assertThat(unavailable.skipReason()).isEqualTo(ShadowSkipReason.NOT_READY); + coordinator.exit(unavailable); + + publish(coordinator, snapshot(6, "01", NOW)); + assertThat(coordinator.phase()).isEqualTo(ShadowSnapshotCoordinator.Phase.READY); + } + + private ShadowSnapshotCoordinator coordinatorAt(String warehouseId) { + return new ShadowSnapshotCoordinator(warehouseId, Duration.ofSeconds(60)); + } + + private static PublishedShadowSnapshot snapshot(long generation, String warehouseId, Instant synchronizedAt) { + return new PublishedShadowSnapshot(generation, warehouseId, synchronizedAt, "signature-" + generation); + } + + private static void publish(ShadowSnapshotCoordinator coordinator, PublishedShadowSnapshot snapshot) { + ShadowSnapshotCoordinator.RefreshTicket refresh = coordinator.tryBeginRefresh(NOW).orElseThrow(); + assertThat(coordinator.publish(refresh, snapshot, NOW)).isTrue(); + } +}