zyr/FT546MWS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: coordinate stable shadow snapshots

Browse Source
This commit is contained in:
zgw
2026-09-03 11:27:26 +08:00
parent 03ddf3f386
commit c6a8b4f3b2
4 changed files with 394 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/PublishedShadowSnapshot.java Normal file
View File

@@ -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");
}
}

9
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSkipReason.java Normal file
View File

@@ -0,0 +1,9 @@
package com.ruoyi.algorithm.shadow;
public enum ShadowSkipReason {
NOT_READY,
BUSY,
STALE,
WAREHOUSE,
CONCURRENT
}

244
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinator.java Normal file
View File

@@ -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> 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<RefreshTicket> 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) {
}
}

121
ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinatorTest.java Normal file
View File

@@ -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();
}
}