fix: clarify shadow refresh recovery contract
This commit is contained in:
@@ -2,6 +2,7 @@ package com.ruoyi.algorithm.shadow;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.Clock;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
@@ -17,6 +18,7 @@ public final class ShadowSnapshotCoordinator {
|
||||
|
||||
private final String configuredWarehouseId;
|
||||
private final Duration snapshotMaxAge;
|
||||
private final Clock clock;
|
||||
private final AtomicReference<State> state = new AtomicReference<>(
|
||||
new State(Phase.INITIALIZING, 0L, 0, null, null, null, null));
|
||||
|
||||
@@ -25,8 +27,13 @@ public final class ShadowSnapshotCoordinator {
|
||||
}
|
||||
|
||||
public ShadowSnapshotCoordinator(String configuredWarehouseId, Duration snapshotMaxAge) {
|
||||
this(configuredWarehouseId, snapshotMaxAge, Clock.systemUTC());
|
||||
}
|
||||
|
||||
public ShadowSnapshotCoordinator(String configuredWarehouseId, Duration snapshotMaxAge, Clock clock) {
|
||||
this.configuredWarehouseId = Objects.requireNonNull(configuredWarehouseId, "configuredWarehouseId").trim();
|
||||
this.snapshotMaxAge = Objects.requireNonNull(snapshotMaxAge, "snapshotMaxAge");
|
||||
this.clock = Objects.requireNonNull(clock, "clock");
|
||||
if (snapshotMaxAge.isNegative() || snapshotMaxAge.isZero()) {
|
||||
throw new IllegalArgumentException("snapshotMaxAge must be positive");
|
||||
}
|
||||
@@ -85,7 +92,7 @@ public final class ShadowSnapshotCoordinator {
|
||||
return;
|
||||
}
|
||||
int nextActiveCount = current.activeLegacyCount() - 1;
|
||||
Instant nextQuietSince = nextActiveCount == 0 ? Instant.now() : current.quietSince();
|
||||
Instant nextQuietSince = nextActiveCount == 0 ? clock.instant() : current.quietSince();
|
||||
State next = new State(current.phase(), current.epoch(), nextActiveCount, current.published(),
|
||||
current.leasedEpoch(), nextQuietSince, current.lastFailure());
|
||||
if (state.compareAndSet(current, next)) {
|
||||
|
||||
@@ -4,6 +4,8 @@ import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneOffset;
|
||||
import java.time.Clock;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@@ -75,9 +77,31 @@ class ShadowSnapshotCoordinatorTest {
|
||||
ShadowSnapshotCoordinator.Admission wrongWarehouse = coordinator.enter(NOW);
|
||||
assertThat(wrongWarehouse.accepted()).isFalse();
|
||||
assertThat(wrongWarehouse.skipReason()).isEqualTo(ShadowSkipReason.WAREHOUSE);
|
||||
assertThat(coordinator.phase()).isEqualTo(ShadowSnapshotCoordinator.Phase.BUSY);
|
||||
coordinator.exit(wrongWarehouse);
|
||||
}
|
||||
|
||||
@Test
|
||||
void staleOrMismatchedEntryRecoversThroughRefresh() {
|
||||
ShadowSnapshotCoordinator coordinator = coordinatorAt("01");
|
||||
publish(coordinator, snapshot(7, "02", NOW));
|
||||
|
||||
ShadowSnapshotCoordinator.Admission skipped = coordinator.enter(NOW);
|
||||
coordinator.exit(skipped);
|
||||
assertThat(coordinator.phase()).isEqualTo(ShadowSnapshotCoordinator.Phase.BUSY);
|
||||
assertThat(coordinator.tryBeginRefresh(NOW)).isPresent();
|
||||
}
|
||||
|
||||
@Test
|
||||
void exitUsesInjectedClockForQuietSince() {
|
||||
Instant exitTime = NOW.plusSeconds(17);
|
||||
ShadowSnapshotCoordinator coordinator = new ShadowSnapshotCoordinator(
|
||||
"01", Duration.ofSeconds(60), Clock.fixed(exitTime, ZoneOffset.UTC));
|
||||
ShadowSnapshotCoordinator.Admission admission = coordinator.enter(NOW);
|
||||
coordinator.exit(admission);
|
||||
assertThat(coordinator.quietSince()).isEqualTo(exitTime);
|
||||
}
|
||||
|
||||
@Test
|
||||
void treatsNumericWarehouseIdsAsEquivalent() {
|
||||
ShadowSnapshotCoordinator coordinator = coordinatorAt("01");
|
||||
|
||||
Reference in New Issue
Block a user