diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowCompletion.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowCompletion.java index 7e9b805..b2b90b1 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowCompletion.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowCompletion.java @@ -1,29 +1,51 @@ package com.ruoyi.algorithm.shadow; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; public final class InboundShadowCompletion { private static final InboundShadowCompletion REJECTED = - new InboundShadowCompletion(false, 0L); + new InboundShadowCompletion(false, false, 0L, null, null); + private final boolean tracked; private final boolean accepted; + private final long startedNanos; private final long deadlineNanos; - private final CompletableFuture legacyResult = new CompletableFuture<>(); + private final Runnable onLegacyExit; + private final ShadowRuntimeMetrics metrics; + private final CompletableFuture legacyResult; + private final AtomicBoolean finished = new AtomicBoolean(); - static InboundShadowCompletion accepted(int timeoutSeconds) { - return new InboundShadowCompletion( - true, System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds)); + static InboundShadowCompletion tracked(boolean accepted, int timeoutSeconds, + Runnable onLegacyExit, ShadowRuntimeMetrics metrics) { + long startedNanos = System.nanoTime(); + return new InboundShadowCompletion(true, accepted, startedNanos, + startedNanos + TimeUnit.SECONDS.toNanos(timeoutSeconds), + Objects.requireNonNull(onLegacyExit, "onLegacyExit"), + Objects.requireNonNull(metrics, "metrics")); } public static InboundShadowCompletion rejected() { return REJECTED; } - private InboundShadowCompletion(boolean accepted, long deadlineNanos) { + private InboundShadowCompletion(boolean tracked, boolean accepted, long deadlineNanos, + Runnable onLegacyExit, ShadowRuntimeMetrics metrics) { + this(tracked, accepted, 0L, deadlineNanos, onLegacyExit, metrics); + } + + private InboundShadowCompletion(boolean tracked, boolean accepted, long startedNanos, + long deadlineNanos, Runnable onLegacyExit, ShadowRuntimeMetrics metrics) { + this.tracked = tracked; this.accepted = accepted; + this.startedNanos = startedNanos; this.deadlineNanos = deadlineNanos; + this.onLegacyExit = onLegacyExit; + this.metrics = metrics; + this.legacyResult = accepted ? new CompletableFuture<>() : null; } public boolean accepted() { @@ -31,19 +53,44 @@ public final class InboundShadowCompletion { } public void complete(String resultCode, String spaceId) { - if (accepted) { - legacyResult.complete(new LegacyResult(resultCode, spaceId)); - } + finish(new LegacyResult(resultCode, spaceId, null)); + } + + public void fail(Throwable failure) { + finish(new LegacyResult(null, null, failure)); } LegacyResult awaitLegacyResult() throws Exception { + if (!accepted) { + throw new IllegalStateException("Skipped shadow completion has no Legacy result"); + } long remainingNanos = deadlineNanos - System.nanoTime(); - if (remainingNanos <= 0) { + if (remainingNanos <= 0L) { throw new TimeoutException("Shadow inbound observation deadline expired"); } return legacyResult.get(remainingNanos, TimeUnit.NANOSECONDS); } - record LegacyResult(String resultCode, String spaceId) { + private void finish(LegacyResult result) { + if (!tracked || !finished.compareAndSet(false, true)) { + return; + } + try { + if (accepted) { + legacyResult.complete(result); + } + metrics.recordLegacyDuration(System.nanoTime() - startedNanos); + } catch (RuntimeException ignored) { + // Shadow bookkeeping must never escape into a Legacy request. + } finally { + try { + onLegacyExit.run(); + } catch (RuntimeException ignored) { + // Shadow bookkeeping must never escape into a Legacy request. + } + } + } + + record LegacyResult(String resultCode, String spaceId, Throwable failure) { } } diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowCompletion.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowCompletion.java index 0953232..4ca2f81 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowCompletion.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowCompletion.java @@ -1,29 +1,51 @@ package com.ruoyi.algorithm.shadow; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; public final class OutboundShadowCompletion { private static final OutboundShadowCompletion REJECTED = - new OutboundShadowCompletion(false, 0L); + new OutboundShadowCompletion(false, false, 0L, null, null); + private final boolean tracked; private final boolean accepted; + private final long startedNanos; private final long deadlineNanos; - private final CompletableFuture legacyResult = new CompletableFuture<>(); + private final Runnable onLegacyExit; + private final ShadowRuntimeMetrics metrics; + private final CompletableFuture legacyResult; + private final AtomicBoolean finished = new AtomicBoolean(); - static OutboundShadowCompletion accepted(int timeoutSeconds) { - return new OutboundShadowCompletion( - true, System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds)); + static OutboundShadowCompletion tracked(boolean accepted, int timeoutSeconds, + Runnable onLegacyExit, ShadowRuntimeMetrics metrics) { + long startedNanos = System.nanoTime(); + return new OutboundShadowCompletion(true, accepted, startedNanos, + startedNanos + TimeUnit.SECONDS.toNanos(timeoutSeconds), + Objects.requireNonNull(onLegacyExit, "onLegacyExit"), + Objects.requireNonNull(metrics, "metrics")); } public static OutboundShadowCompletion rejected() { return REJECTED; } - private OutboundShadowCompletion(boolean accepted, long deadlineNanos) { + private OutboundShadowCompletion(boolean tracked, boolean accepted, long deadlineNanos, + Runnable onLegacyExit, ShadowRuntimeMetrics metrics) { + this(tracked, accepted, 0L, deadlineNanos, onLegacyExit, metrics); + } + + private OutboundShadowCompletion(boolean tracked, boolean accepted, long startedNanos, + long deadlineNanos, Runnable onLegacyExit, ShadowRuntimeMetrics metrics) { + this.tracked = tracked; this.accepted = accepted; + this.startedNanos = startedNanos; this.deadlineNanos = deadlineNanos; + this.onLegacyExit = onLegacyExit; + this.metrics = metrics; + this.legacyResult = accepted ? new CompletableFuture<>() : null; } public boolean accepted() { @@ -31,16 +53,44 @@ public final class OutboundShadowCompletion { } public void complete(String resultCode) { - if (accepted) { - legacyResult.complete(resultCode); - } + finish(new LegacyResult(resultCode, null)); } - String awaitLegacyResult() throws Exception { + public void fail(Throwable failure) { + finish(new LegacyResult(null, failure)); + } + + LegacyResult awaitLegacyResult() throws Exception { + if (!accepted) { + throw new IllegalStateException("Skipped shadow completion has no Legacy result"); + } long remainingNanos = deadlineNanos - System.nanoTime(); - if (remainingNanos <= 0) { + if (remainingNanos <= 0L) { throw new TimeoutException("Shadow outbound observation deadline expired"); } return legacyResult.get(remainingNanos, TimeUnit.NANOSECONDS); } + + private void finish(LegacyResult result) { + if (!tracked || !finished.compareAndSet(false, true)) { + return; + } + try { + if (accepted) { + legacyResult.complete(result); + } + metrics.recordLegacyDuration(System.nanoTime() - startedNanos); + } catch (RuntimeException ignored) { + // Shadow bookkeeping must never escape into a Legacy request. + } finally { + try { + onLegacyExit.run(); + } catch (RuntimeException ignored) { + // Shadow bookkeeping must never escape into a Legacy request. + } + } + } + + record LegacyResult(String resultCode, Throwable failure) { + } } diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowFailureStage.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowFailureStage.java new file mode 100644 index 0000000..9174c7d --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowFailureStage.java @@ -0,0 +1,10 @@ +package com.ruoyi.algorithm.shadow; + +public enum ShadowFailureStage { + STARTUP, + PLAN, + LEGACY_WAIT, + COMPARE, + REFRESH, + METRIC_FLUSH +} diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java index 70c6449..da337c6 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java @@ -1,102 +1,360 @@ package com.ruoyi.algorithm.shadow; +import com.ruoyi.algorithm.integration.SchedulingFacade; +import com.ruoyi.algorithm.shadow.mapper.ShadowInputMapper; +import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; import java.util.Objects; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; @Component public class ShadowObservationGateway { private static final Logger LOG = LoggerFactory.getLogger(ShadowObservationGateway.class); + private static final Duration RECOVERY_COOLDOWN = Duration.ofMinutes(1); + private static final Duration[] RETRY_DELAYS = { + Duration.ofSeconds(1), Duration.ofSeconds(2), Duration.ofSeconds(4) + }; private final ShadowObservationProperties properties; private final ShadowPlanningEngine engine; + private final ShadowInputMapper inputMapper; + private final ShadowSnapshotCoordinator coordinator; + private final ShadowRuntimeMetrics metrics; + private final Clock clock; + private final Sleeper sleeper; private final ThreadPoolExecutor executor; + private final String instanceId; + private final String warehouseId; + private final AtomicBoolean started = new AtomicBoolean(); + private final AtomicBoolean recoveryQueued = new AtomicBoolean(); + private final AtomicLong lastGeneration = new AtomicLong(); + private final AtomicReference recoveryNotBefore = + new AtomicReference<>(Instant.MIN); - public ShadowObservationGateway(ShadowObservationProperties properties, ShadowPlanningEngine engine) { + @Autowired + public ShadowObservationGateway(ShadowObservationProperties properties, + ShadowPlanningEngine engine, ShadowInputMapper inputMapper, + SchedulingFacade schedulingFacade, ShadowRuntimeMetrics metrics) { + this(properties, engine, inputMapper, + new ShadowSnapshotCoordinator(resolveWarehouseId(properties, schedulingFacade), + Duration.ofSeconds(properties.getSnapshotMaxAgeSeconds()), Clock.systemUTC()), + metrics, Clock.systemUTC(), ShadowObservationGateway::sleep, + newExecutor(), UUID.randomUUID().toString()); + } + + ShadowObservationGateway(ShadowObservationProperties properties, + ShadowPlanningEngine engine, ShadowInputMapper inputMapper, + ShadowSnapshotCoordinator coordinator, ShadowRuntimeMetrics metrics, + Clock clock, Sleeper sleeper, ThreadPoolExecutor executor, String instanceId) { this.properties = Objects.requireNonNull(properties, "properties"); this.engine = engine; - this.executor = new ThreadPoolExecutor( - 1, 1, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<>(), runnable -> { - Thread thread = new Thread(runnable, "ft546-shadow-observer"); - thread.setDaemon(true); - return thread; - }, new ThreadPoolExecutor.AbortPolicy()); + this.inputMapper = inputMapper; + this.coordinator = Objects.requireNonNull(coordinator, "coordinator"); + this.metrics = Objects.requireNonNull(metrics, "metrics"); + this.clock = Objects.requireNonNull(clock, "clock"); + this.sleeper = Objects.requireNonNull(sleeper, "sleeper"); + this.executor = Objects.requireNonNull(executor, "executor"); + this.instanceId = Objects.requireNonNull(instanceId, "instanceId"); + this.warehouseId = coordinator.configuredWarehouseId(); + } + + @PostConstruct + public void start() { + if (available() && started.compareAndSet(false, true)) { + requestRecovery(true, ShadowFailureStage.STARTUP); + } } public InboundShadowCompletion observeInbound(String taskNo, String trayNo) { if (!available()) { return InboundShadowCompletion.rejected(); } - InboundShadowCompletion completion = - InboundShadowCompletion.accepted(properties.getTimeoutSeconds()); + long startedNanos = System.nanoTime(); + ShadowSnapshotCoordinator.Admission admission = coordinator.enter(clock.instant()); + Runnable onLegacyExit = () -> finishLegacy(admission); + if (!admission.accepted()) { + metrics.recordAdmission(admission.skipReason(), System.nanoTime() - startedNanos); + return InboundShadowCompletion.tracked( + false, properties.getTimeoutSeconds(), onLegacyExit, metrics); + } + + InboundShadowCompletion completion = InboundShadowCompletion.tracked( + true, properties.getTimeoutSeconds(), onLegacyExit, metrics); try { - executor.execute(() -> observeInbound(taskNo, trayNo, completion)); + executor.execute(() -> observeInbound(taskNo, trayNo, admission, completion)); + metrics.recordAdmission(null, System.nanoTime() - startedNanos); return completion; - } catch (RejectedExecutionException exception) { - LOG.debug("Shadow inbound observation skipped because worker is busy for task {}", taskNo); - return InboundShadowCompletion.rejected(); + } catch (RejectedExecutionException rejected) { + coordinator.finishObservation(admission); + metrics.recordAdmission(ShadowSkipReason.BUSY, System.nanoTime() - startedNanos); + return InboundShadowCompletion.tracked( + false, properties.getTimeoutSeconds(), onLegacyExit, metrics); } } public OutboundShadowCompletion observeProjectOutbound(String stockoutDetailId) { return observeOutbound(stockoutDetailId, - () -> engine.planProjectOutbound(stockoutDetailId)); + snapshot -> engine.planProjectOutbound(snapshot, stockoutDetailId)); } public OutboundShadowCompletion observeExactStockOutbound(String stockoutDetailId, String stockId) { return observeOutbound(stockoutDetailId, - () -> engine.planExactStockOutbound(stockoutDetailId, stockId)); + snapshot -> engine.planExactStockOutbound(snapshot, stockoutDetailId, stockId)); } - private void observeInbound(String taskNo, String trayNo, InboundShadowCompletion completion) { - try { - InboundShadowObservation observation = engine.planInbound(taskNo, trayNo); - InboundShadowCompletion.LegacyResult legacy = completion.awaitLegacyResult(); - engine.compareInbound(observation, legacy.resultCode(), legacy.spaceId()); - } catch (InterruptedException exception) { - Thread.currentThread().interrupt(); - LOG.warn("Shadow inbound observation interrupted for task {}", taskNo); - } catch (Exception exception) { - LOG.warn("Shadow inbound observation failed open for task {}: {}", taskNo, exception.getMessage()); - } - } - - private OutboundShadowCompletion observeOutbound(String stockoutDetailId, - OutboundPlanner planner) { + private OutboundShadowCompletion observeOutbound(String stockoutDetailId, OutboundPlanner planner) { if (!available()) { return OutboundShadowCompletion.rejected(); } - OutboundShadowCompletion completion = - OutboundShadowCompletion.accepted(properties.getTimeoutSeconds()); + long startedNanos = System.nanoTime(); + ShadowSnapshotCoordinator.Admission admission = coordinator.enter(clock.instant()); + Runnable onLegacyExit = () -> finishLegacy(admission); + if (!admission.accepted()) { + metrics.recordAdmission(admission.skipReason(), System.nanoTime() - startedNanos); + return OutboundShadowCompletion.tracked( + false, properties.getTimeoutSeconds(), onLegacyExit, metrics); + } + + OutboundShadowCompletion completion = OutboundShadowCompletion.tracked( + true, properties.getTimeoutSeconds(), onLegacyExit, metrics); try { - executor.execute(() -> { - try { - OutboundShadowObservation observation = planner.plan(); - engine.compareOutbound(observation, completion.awaitLegacyResult()); - } catch (InterruptedException exception) { - Thread.currentThread().interrupt(); - LOG.warn("Shadow outbound observation interrupted for detail {}", stockoutDetailId); - } catch (Exception exception) { - LOG.warn("Shadow outbound observation failed open for detail {}: {}", - stockoutDetailId, exception.getMessage()); - } - }); + executor.execute(() -> observeOutbound( + stockoutDetailId, admission, completion, planner)); + metrics.recordAdmission(null, System.nanoTime() - startedNanos); return completion; - } catch (RejectedExecutionException exception) { - LOG.debug("Shadow outbound observation skipped because worker is busy for detail {}", stockoutDetailId); - return OutboundShadowCompletion.rejected(); + } catch (RejectedExecutionException rejected) { + coordinator.finishObservation(admission); + metrics.recordAdmission(ShadowSkipReason.BUSY, System.nanoTime() - startedNanos); + return OutboundShadowCompletion.tracked( + false, properties.getTimeoutSeconds(), onLegacyExit, metrics); } } + private void observeInbound(String taskNo, String trayNo, + ShadowSnapshotCoordinator.Admission admission, InboundShadowCompletion completion) { + long startedNanos = System.nanoTime(); + ShadowFailureStage stage = ShadowFailureStage.PLAN; + try { + InboundShadowObservation observation = + engine.planInbound(admission.snapshot(), taskNo, trayNo); + stage = ShadowFailureStage.LEGACY_WAIT; + InboundShadowCompletion.LegacyResult legacy = completion.awaitLegacyResult(); + if (legacy.failure() != null) { + return; + } + if (!coordinator.isComparable(admission)) { + metrics.increment(ShadowSkipReason.CONCURRENT); + return; + } + stage = ShadowFailureStage.COMPARE; + engine.compareInbound(observation, legacy.resultCode(), legacy.spaceId()); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + recordWorkerFailure(taskNo, admission, stage, interrupted, startedNanos); + } catch (Exception shadowFailure) { + recordWorkerFailure(taskNo, admission, stage, shadowFailure, startedNanos); + } finally { + coordinator.finishObservation(admission); + requestRecovery(false, ShadowFailureStage.REFRESH); + } + } + + private void observeOutbound(String stockoutDetailId, + ShadowSnapshotCoordinator.Admission admission, OutboundShadowCompletion completion, + OutboundPlanner planner) { + long startedNanos = System.nanoTime(); + ShadowFailureStage stage = ShadowFailureStage.PLAN; + try { + OutboundShadowObservation observation = planner.plan(admission.snapshot()); + stage = ShadowFailureStage.LEGACY_WAIT; + OutboundShadowCompletion.LegacyResult legacy = completion.awaitLegacyResult(); + if (legacy.failure() != null) { + return; + } + if (!coordinator.isComparable(admission)) { + metrics.increment(ShadowSkipReason.CONCURRENT); + return; + } + stage = ShadowFailureStage.COMPARE; + engine.compareOutbound(observation, legacy.resultCode()); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + recordWorkerFailure(stockoutDetailId, admission, stage, interrupted, startedNanos); + } catch (Exception shadowFailure) { + recordWorkerFailure(stockoutDetailId, admission, stage, shadowFailure, startedNanos); + } finally { + coordinator.finishObservation(admission); + requestRecovery(false, ShadowFailureStage.REFRESH); + } + } + + private void finishLegacy(ShadowSnapshotCoordinator.Admission admission) { + coordinator.exit(admission); + requestRecovery(false, ShadowFailureStage.REFRESH); + } + + private void requestRecovery(boolean ignoreCooldown, ShadowFailureStage failureStage) { + if (!available()) { + return; + } + Instant now = clock.instant(); + if (!ignoreCooldown && now.isBefore(recoveryNotBefore.get())) { + return; + } + if (!recoveryQueued.compareAndSet(false, true)) { + return; + } + try { + executor.execute(() -> runRecoveryTask(failureStage)); + } catch (RejectedExecutionException rejected) { + recoveryQueued.set(false); + } + } + + private void runRecoveryTask(ShadowFailureStage failureStage) { + try { + refreshWithBoundedRecovery(failureStage); + } finally { + flushMetrics(); + recoveryQueued.set(false); + if (available() && coordinator.activeLegacyCount() == 0 + && coordinator.phase() != ShadowSnapshotCoordinator.Phase.READY + && !clock.instant().isBefore(recoveryNotBefore.get())) { + requestRecovery(false, ShadowFailureStage.REFRESH); + } + } + } + + private void refreshWithBoundedRecovery(ShadowFailureStage failureStage) { + for (int attempt = 0; attempt <= RETRY_DELAYS.length; attempt++) { + if (attempt > 0) { + try { + sleeper.sleep(RETRY_DELAYS[attempt - 1]); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + metrics.recordError(failureStage); + warnFailure("CONTROL", 0L, failureStage, interrupted, 0L); + return; + } + } + + Optional optionalTicket; + try { + optionalTicket = tryBeginRefreshAfterQuietWindow(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + metrics.recordError(failureStage); + warnFailure("CONTROL", 0L, failureStage, interrupted, 0L); + return; + } + if (optionalTicket.isEmpty()) { + return; + } + + ShadowSnapshotCoordinator.RefreshTicket ticket = optionalTicket.orElseThrow(); + long generation = 0L; + long startedNanos = System.nanoTime(); + try { + generation = nextGeneration(); + PublishedShadowSnapshot snapshot = engine.refreshSnapshot(generation, instanceId); + if (coordinator.publish(ticket, snapshot, clock.instant())) { + recoveryNotBefore.set(Instant.MIN); + return; + } + metrics.increment(ShadowSkipReason.CONCURRENT); + } catch (Exception shadowFailure) { + coordinator.failRefresh(ticket, failureDescription(failureStage, shadowFailure)); + metrics.recordError(failureStage); + warnFailure("CONTROL", generation, failureStage, shadowFailure, startedNanos); + } + } + recoveryNotBefore.set(clock.instant().plus(RECOVERY_COOLDOWN)); + } + + private Optional tryBeginRefreshAfterQuietWindow() + throws InterruptedException { + if (coordinator.activeLegacyCount() != 0) { + return Optional.empty(); + } + Instant quietSince = coordinator.quietSince(); + if (quietSince != null) { + Instant quietUntil = quietSince.plusMillis(properties.getQuietIntervalMillis()); + Instant now = clock.instant(); + if (now.isBefore(quietUntil)) { + sleeper.sleep(Duration.between(now, quietUntil)); + } + } + return coordinator.tryBeginRefresh(clock.instant()); + } + + private long nextGeneration() { + ShadowSnapshotDescriptor descriptor = inputMapper.selectPublishedSnapshot(warehouseId); + long databaseGeneration = descriptor == null ? 0L : descriptor.getSnapshotGeneration(); + if (databaseGeneration == Long.MAX_VALUE) { + throw new IllegalStateException("shadow snapshot generation is exhausted"); + } + return lastGeneration.updateAndGet(current -> + Math.max(current + 1L, Math.max(1L, databaseGeneration + 1L))); + } + + private void flushMetrics() { + ShadowRuntimeMetrics.Batch batch = metrics.drain(); + if (batch.totalCount() == 0L) { + return; + } + Instant flushedAt = clock.instant(); + try { + for (ShadowRuntimeMetrics.Row row : batch.rows()) { + inputMapper.upsertRuntimeMetric(warehouseId, row.metricName(), + row.bucketUpperMicros(), row.count(), row.totalMicros(), + row.maxMicros(), flushedAt); + } + } catch (Exception shadowFailure) { + metrics.recordError(ShadowFailureStage.METRIC_FLUSH); + warnFailure("CONTROL", 0L, ShadowFailureStage.METRIC_FLUSH, + shadowFailure, 0L); + } + } + + private void recordWorkerFailure(String requestId, + ShadowSnapshotCoordinator.Admission admission, ShadowFailureStage stage, + Exception failure, long startedNanos) { + metrics.recordError(stage); + warnFailure(requestId, admission.snapshot().generation(), stage, failure, startedNanos); + } + + private void warnFailure(String requestId, long generation, + ShadowFailureStage stage, Throwable failure, long startedNanos) { + long elapsedMicros = startedNanos == 0L ? 0L + : TimeUnit.NANOSECONDS.toMicros(Math.max(0L, System.nanoTime() - startedNanos)); + LOG.warn("shadow_failure requestId={} warehouse={} snapshotGeneration={} stage={} " + + "exceptionClass={} elapsedMicros={}", + requestId, warehouseId, generation, stage, + failure.getClass().getName(), elapsedMicros); + } + + private String failureDescription(ShadowFailureStage stage, Throwable failure) { + return stage.name() + ':' + failure.getClass().getName(); + } + private boolean available() { - return properties.isEnabled() && engine != null && !executor.isShutdown(); + return properties.isEnabled() && engine != null && inputMapper != null && !executor.isShutdown(); } @PreDestroy @@ -104,8 +362,38 @@ public class ShadowObservationGateway { executor.shutdownNow(); } + private static String resolveWarehouseId(ShadowObservationProperties properties, + SchedulingFacade schedulingFacade) { + String warehouseId = properties.getWarehouseId(); + if (warehouseId == null || warehouseId.isBlank()) { + warehouseId = Objects.requireNonNull(schedulingFacade, "schedulingFacade").defaultWarehouseId(); + } + if (warehouseId == null || warehouseId.isBlank()) { + throw new IllegalStateException("shadow warehouse is not configured"); + } + return warehouseId.trim(); + } + + private static ThreadPoolExecutor newExecutor() { + return new ThreadPoolExecutor( + 1, 1, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1), runnable -> { + Thread thread = new Thread(runnable, "ft546-shadow-observer"); + thread.setDaemon(true); + return thread; + }, new ThreadPoolExecutor.AbortPolicy()); + } + + private static void sleep(Duration duration) throws InterruptedException { + TimeUnit.NANOSECONDS.sleep(duration.toNanos()); + } + + @FunctionalInterface + interface Sleeper { + void sleep(Duration delay) throws InterruptedException; + } + @FunctionalInterface private interface OutboundPlanner { - OutboundShadowObservation plan(); + OutboundShadowObservation plan(PublishedShadowSnapshot snapshot); } } diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowRuntimeMetrics.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowRuntimeMetrics.java new file mode 100644 index 0000000..46f9ce2 --- /dev/null +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowRuntimeMetrics.java @@ -0,0 +1,157 @@ +package com.ruoyi.algorithm.shadow; + +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReferenceArray; + +@Component +public final class ShadowRuntimeMetrics { + private static final long[] ADMISSION_BUCKETS = { + 100L, 250L, 500L, 1000L, 2500L, 5000L, 10000L, Long.MAX_VALUE + }; + private static final long[] LEGACY_DURATION_BUCKETS = { + 100L, 250L, 500L, 1000L, 2500L, 5000L, 10000L, + 25000L, 50000L, 100000L, 250000L, 500000L, 1000000L, Long.MAX_VALUE + }; + private static final String[] COUNTER_NAMES = counterNames(); + private static final int ADMISSION_OFFSET = COUNTER_NAMES.length; + private static final int LEGACY_OFFSET = ADMISSION_OFFSET + ADMISSION_BUCKETS.length; + + private final AtomicReferenceArray values = + new AtomicReferenceArray<>(LEGACY_OFFSET + LEGACY_DURATION_BUCKETS.length); + + public ShadowRuntimeMetrics() { + for (int index = 0; index < values.length(); index++) { + values.set(index, MetricValue.ZERO); + } + } + + public void recordAdmission(ShadowSkipReason reason, long elapsedNanos) { + record(reason == null ? 0 : 1 + reason.ordinal(), 0L); + record(ADMISSION_OFFSET + bucketIndex(ADMISSION_BUCKETS, toMicros(elapsedNanos)), + toMicros(elapsedNanos)); + } + + public void recordLegacyDuration(long elapsedNanos) { + long elapsedMicros = toMicros(elapsedNanos); + record(LEGACY_OFFSET + bucketIndex(LEGACY_DURATION_BUCKETS, elapsedMicros), elapsedMicros); + } + + public void increment(ShadowSkipReason reason) { + if (reason != null) { + record(1 + reason.ordinal(), 0L); + } + } + + public void recordError(ShadowFailureStage stage) { + Objects.requireNonNull(stage, "stage"); + record(1 + ShadowSkipReason.values().length + stage.ordinal(), 0L); + } + + public Batch drain() { + List rows = new ArrayList<>(values.length()); + for (int index = 0; index < values.length(); index++) { + MetricValue value = values.getAndSet(index, MetricValue.ZERO); + if (value.count() == 0) { + continue; + } + if (index < COUNTER_NAMES.length) { + rows.add(new Row(COUNTER_NAMES[index], 0L, + value.count(), value.totalMicros(), value.maxMicros())); + } else if (index < LEGACY_OFFSET) { + int bucket = index - ADMISSION_OFFSET; + rows.add(new Row("ADMISSION_US", ADMISSION_BUCKETS[bucket], + value.count(), value.totalMicros(), value.maxMicros())); + } else { + int bucket = index - LEGACY_OFFSET; + rows.add(new Row("LEGACY_DURATION_US", LEGACY_DURATION_BUCKETS[bucket], + value.count(), value.totalMicros(), value.maxMicros())); + } + } + return new Batch(rows); + } + + private void record(int index, long elapsedMicros) { + while (true) { + MetricValue current = values.get(index); + MetricValue next = current.add(elapsedMicros); + if (values.compareAndSet(index, current, next)) { + return; + } + } + } + + private static int bucketIndex(long[] buckets, long value) { + for (int index = 0; index < buckets.length; index++) { + if (value <= buckets[index]) { + return index; + } + } + return buckets.length - 1; + } + + private static long toMicros(long elapsedNanos) { + return TimeUnit.NANOSECONDS.toMicros(Math.max(0L, elapsedNanos)); + } + + private static String[] counterNames() { + List names = new ArrayList<>(); + names.add("ADMITTED"); + Arrays.stream(ShadowSkipReason.values()) + .map(reason -> "SKIP_" + reason.name()) + .forEach(names::add); + Arrays.stream(ShadowFailureStage.values()) + .map(stage -> "ERROR_" + stage.name()) + .forEach(names::add); + return names.toArray(String[]::new); + } + + private record MetricValue(long count, long totalMicros, long maxMicros) { + private static final MetricValue ZERO = new MetricValue(0L, 0L, 0L); + + private MetricValue add(long elapsedMicros) { + return new MetricValue(count + 1L, totalMicros + elapsedMicros, + Math.max(maxMicros, elapsedMicros)); + } + } + + public record Row(String metricName, long bucketUpperMicros, long count, + long totalMicros, long maxMicros) { + } + + public static final class Batch { + private final List rows; + + private Batch(List rows) { + this.rows = List.copyOf(rows); + } + + public List rows() { + return rows; + } + + public long count(String metricName) { + return rows.stream() + .filter(row -> row.metricName().equals(metricName) && row.bucketUpperMicros() == 0L) + .mapToLong(Row::count) + .sum(); + } + + public long bucketCount(String metricName, long bucketUpperMicros) { + return rows.stream() + .filter(row -> row.metricName().equals(metricName) + && row.bucketUpperMicros() == bucketUpperMicros) + .mapToLong(Row::count) + .sum(); + } + + public long totalCount() { + return rows.stream().mapToLong(Row::count).sum(); + } + } +} 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 index d462bbb..80d1aec 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinator.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinator.java @@ -6,6 +6,7 @@ import java.time.Clock; import java.util.Objects; import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; /** @@ -14,11 +15,10 @@ import java.util.concurrent.atomic.AtomicReference; 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 Clock clock; + private final AtomicLong refreshLeaseSequence = new AtomicLong(-1L); private final AtomicReference state = new AtomicReference<>( new State(Phase.INITIALIZING, 0L, 0, null, null, null, null)); @@ -126,8 +126,9 @@ public final class ShadowSnapshotCoordinator { if (current.activeLegacyCount() != 0 || current.leasedEpoch() != null) { return Optional.empty(); } + long refreshLease = refreshLeaseSequence.getAndDecrement(); State next = new State(Phase.BUSY, current.epoch(), current.activeLegacyCount(), current.published(), - REFRESH_LEASE, current.quietSince() == null ? now : current.quietSince(), current.lastFailure()); + refreshLease, current.quietSince() == null ? now : current.quietSince(), current.lastFailure()); if (state.compareAndSet(current, next)) { return Optional.of(new RefreshTicket(this, next)); } @@ -141,17 +142,28 @@ public final class ShadowSnapshotCoordinator { 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); + if (state.compareAndSet(expected, next)) { + return true; + } + failRefresh(ticket, "refresh invalidated by concurrent Legacy activity"); + return false; } 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); + while (true) { + State current = state.get(); + if (!Objects.equals(current.leasedEpoch(), ticket.state.leasedEpoch())) { + return; + } + State next = new State(Phase.UNAVAILABLE, current.epoch(), current.activeLegacyCount(), + current.published(), null, current.quietSince(), failure); + if (state.compareAndSet(current, next)) { + return; + } + } } public boolean isComparable(Admission admission) { @@ -186,6 +198,10 @@ public final class ShadowSnapshotCoordinator { return state.get().lastFailure(); } + String configuredWarehouseId() { + return configuredWarehouseId; + } + public static final class Admission { private final ShadowSnapshotCoordinator owner; private final boolean accepted; diff --git a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/mapper/ShadowInputMapper.java b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/mapper/ShadowInputMapper.java index f75fb5c..1b7bb2b 100644 --- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/mapper/ShadowInputMapper.java +++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/mapper/ShadowInputMapper.java @@ -37,4 +37,12 @@ public interface ShadowInputMapper { @Param("sourceSignature") String sourceSignature, @Param("resultCode") String resultCode, @Param("reason") String reason); + + void upsertRuntimeMetric(@Param("warehouseId") String warehouseId, + @Param("metricName") String metricName, + @Param("bucketUpperMicros") long bucketUpperMicros, + @Param("metricCount") long metricCount, + @Param("metricTotalMicros") long metricTotalMicros, + @Param("metricMaxMicros") long metricMaxMicros, + @Param("flushedAt") Instant flushedAt); } diff --git a/ruoyi-admin/src/main/resources/mapper/algorithm/ShadowInputMapper.xml b/ruoyi-admin/src/main/resources/mapper/algorithm/ShadowInputMapper.xml index 03f7504..aab0e9e 100644 --- a/ruoyi-admin/src/main/resources/mapper/algorithm/ShadowInputMapper.xml +++ b/ruoyi-admin/src/main/resources/mapper/algorithm/ShadowInputMapper.xml @@ -112,4 +112,26 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" and state.synchronized_at = #{snapshotSynchronizedAt} and convert(varchar(64), state.source_signature, 2) = #{sourceSignature} + + + merge alg.shadow_runtime_metric with (holdlock) as target + using (select #{warehouseId} as warehouse_id, + #{metricName} as metric_name, + #{bucketUpperMicros} as bucket_upper_micros) as source + on target.warehouse_id = source.warehouse_id + and target.metric_name = source.metric_name + and target.bucket_upper_micros = source.bucket_upper_micros + when matched then update set + metric_count = target.metric_count + #{metricCount}, + metric_total_micros = target.metric_total_micros + #{metricTotalMicros}, + metric_max_micros = case when target.metric_max_micros > #{metricMaxMicros} + then target.metric_max_micros else #{metricMaxMicros} end, + flushed_at = #{flushedAt} + when not matched then insert ( + warehouse_id, metric_name, bucket_upper_micros, + metric_count, metric_total_micros, metric_max_micros, flushed_at) + values ( + source.warehouse_id, source.metric_name, source.bucket_upper_micros, + #{metricCount}, #{metricTotalMicros}, #{metricMaxMicros}, #{flushedAt}); + diff --git a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java index 55b5c66..18428f7 100644 --- a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java +++ b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowObservationGatewayTest.java @@ -1,15 +1,30 @@ package com.ruoyi.algorithm.shadow; +import com.ruoyi.algorithm.shadow.mapper.ShadowInputMapper; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.after; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.timeout; @@ -17,100 +32,433 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; class ShadowObservationGatewayTest { - private ShadowObservationGateway gateway; + private static final Instant NOW = Instant.parse("2026-09-03T00:00:00Z"); + private static final PublishedShadowSnapshot SNAPSHOT = + new PublishedShadowSnapshot(42L, "01", NOW, "ABC123"); + + private final List gateways = new ArrayList<>(); + private final List executors = new ArrayList<>(); @AfterEach - void shutDownGateway() { - if (gateway != null) { - gateway.shutdown(); - } + void shutDownGateways() { + gateways.forEach(ShadowObservationGateway::shutdown); + executors.forEach(ThreadPoolExecutor::shutdownNow); } @Test - void disabledShadowNeverInvokesThePlanningEngine() { - ShadowObservationProperties properties = new ShadowObservationProperties(); + void completionTokensExitLegacyActivityExactlyOnceOnSuccessFailureAndDuplicates() throws Exception { + ShadowRuntimeMetrics metrics = new ShadowRuntimeMetrics(); + AtomicInteger inboundExits = new AtomicInteger(); + InboundShadowCompletion inbound = InboundShadowCompletion.tracked( + true, 3, inboundExits::incrementAndGet, metrics); + + inbound.complete("10", "SPACE-1"); + inbound.complete("10", "SPACE-2"); + inbound.fail(new IllegalStateException("late failure")); + + assertThat(inboundExits).hasValue(1); + assertThat(inbound.awaitLegacyResult().spaceId()).isEqualTo("SPACE-1"); + + AtomicInteger outboundExits = new AtomicInteger(); + OutboundShadowCompletion outbound = OutboundShadowCompletion.tracked( + true, 3, outboundExits::incrementAndGet, metrics); + IllegalStateException failure = new IllegalStateException("legacy failed"); + + outbound.fail(failure); + outbound.fail(new IllegalStateException("again")); + outbound.complete("10"); + + assertThat(outboundExits).hasValue(1); + assertThat(outbound.awaitLegacyResult().failure()).isSameAs(failure); + } + + @Test + void enabledSkippedTokenStillExitsActivityWhileDisabledTokenIsAReusableNoOp() { + ShadowRuntimeMetrics metrics = new ShadowRuntimeMetrics(); + AtomicInteger exits = new AtomicInteger(); + InboundShadowCompletion skipped = InboundShadowCompletion.tracked( + false, 3, exits::incrementAndGet, metrics); + + skipped.complete("10", "SPACE-1"); + skipped.fail(new IllegalStateException("ignored")); + + assertThat(skipped.accepted()).isFalse(); + assertThat(exits).hasValue(1); + assertThat(InboundShadowCompletion.rejected()) + .isSameAs(InboundShadowCompletion.rejected()); + assertThat(OutboundShadowCompletion.rejected()) + .isSameAs(OutboundShadowCompletion.rejected()); + + OutboundShadowCompletion failingCallback = OutboundShadowCompletion.tracked( + false, 3, () -> { throw new IllegalStateException("callback failed"); }, metrics); + assertThatCode(() -> failingCallback.fail(new IllegalStateException("legacy failed"))) + .doesNotThrowAnyException(); + } + + @Test + void startRefreshesGenerationOneAndPublishesReadyWithoutBlockingTheCaller() throws Exception { ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); - gateway = new ShadowObservationGateway(properties, engine); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + CountDownLatch refreshCalled = new CountDownLatch(1); + when(engine.refreshSnapshot(1L, "INSTANCE-1")).thenAnswer(invocation -> { + refreshCalled.countDown(); + return new PublishedShadowSnapshot(1L, "01", NOW, "ABC001"); + }); + ShadowSnapshotCoordinator coordinator = coordinator(); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { }); + + gateway.start(); + + assertThat(refreshCalled.await(1, TimeUnit.SECONDS)).isTrue(); + verify(engine, timeout(1000)).refreshSnapshot(1L, "INSTANCE-1"); + assertThat(waitForPhase(coordinator, ShadowSnapshotCoordinator.Phase.READY)).isTrue(); + } + + @Test + void validInboundPlansFromLeaseWaitsOnlyOnWorkerAndComparesLegacyResult() throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); + InboundShadowObservation observation = inboundObservation(); + when(engine.planInbound(SNAPSHOT, "TASK-1", "TRAY-1")).thenReturn(observation); + CountDownLatch compared = new CountDownLatch(1); + doAnswer(invocation -> { + compared.countDown(); + return null; + }).when(engine).compareInbound(observation, "10", "LEGACY-SPACE"); + stubSuccessfulRecovery(engine, mapper, 43L); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW.plusSeconds(2), ZoneOffset.UTC), delay -> { }); InboundShadowCompletion completion = gateway.observeInbound("TASK-1", "TRAY-1"); - completion.complete("10", "SPACE-1"); - verify(engine, never()).planInbound("TASK-1", "TRAY-1"); + assertThat(completion.accepted()).isTrue(); + verify(mapper, never()).upsertRuntimeMetric(anyString(), anyString(), anyLong(), anyLong(), + anyLong(), anyLong(), any(Instant.class)); + completion.complete("10", "LEGACY-SPACE"); + assertThat(compared.await(1, TimeUnit.SECONDS)).isTrue(); + verify(engine).planInbound(SNAPSHOT, "TASK-1", "TRAY-1"); + verify(engine).compareInbound(observation, "10", "LEGACY-SPACE"); } @Test - void planningFailureIsContainedOnTheBackgroundWorker() { - ShadowObservationProperties properties = enabledProperties(); - ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); - when(engine.planInbound("TASK-2", "TRAY-2")) - .thenThrow(new IllegalStateException("algorithm objects are unavailable")); - gateway = new ShadowObservationGateway(properties, engine); - - InboundShadowCompletion completion = gateway.observeInbound("TASK-2", "TRAY-2"); - completion.complete("10", "SPACE-2"); - - verify(engine, timeout(1000)).planInbound("TASK-2", "TRAY-2"); - verify(engine, never()).compareInbound( - org.mockito.ArgumentMatchers.any(), - org.mockito.ArgumentMatchers.any(), - org.mockito.ArgumentMatchers.any()); - } - - @Test - void missingEngineFailsOpenForOutboundObservation() { - gateway = new ShadowObservationGateway(enabledProperties(), null); - - OutboundShadowCompletion completion = - gateway.observeExactStockOutbound("DETAIL-1", "STOCK-1"); - completion.complete("10"); - - assertThat(completion.accepted()).isFalse(); - } - - @Test - void planningSubmissionNeverWaitsAndConcurrentObservationIsDropped() throws Exception { + void concurrentLegacyCallReturnsImmediatelyAndInvalidatesTheFirstComparison() throws Exception { ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); CountDownLatch planningStarted = new CountDownLatch(1); CountDownLatch releasePlanning = new CountDownLatch(1); - InboundShadowObservation observation = new InboundShadowObservation( - true, "W01", "TASK-3", "READY", "SPACE-SHADOW", "proposal"); - when(engine.planInbound("TASK-3", "TRAY-3")).thenAnswer(invocation -> { + when(engine.planInbound(SNAPSHOT, "TASK-2", "TRAY-2")).thenAnswer(invocation -> { planningStarted.countDown(); - if (!releasePlanning.await(5, TimeUnit.SECONDS)) { - throw new IllegalStateException("test planning was not released"); - } - return observation; + releasePlanning.await(1, TimeUnit.SECONDS); + return inboundObservation(); }); - gateway = new ShadowObservationGateway(enabledProperties(), engine); - ExecutorService legacyCaller = Executors.newSingleThreadExecutor(); + stubSuccessfulRecovery(engine, mapper, 43L); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW.plusSeconds(2), ZoneOffset.UTC), delay -> { }); - try { - Future submitted = - legacyCaller.submit(() -> gateway.observeInbound("TASK-3", "TRAY-3")); - InboundShadowCompletion first = submitted.get(1, TimeUnit.SECONDS); - assertThat(first.accepted()).isTrue(); - assertThat(planningStarted.await(1, TimeUnit.SECONDS)).isTrue(); + InboundShadowCompletion first = gateway.observeInbound("TASK-2", "TRAY-2"); + assertThat(planningStarted.await(1, TimeUnit.SECONDS)).isTrue(); + long started = System.nanoTime(); + OutboundShadowCompletion concurrent = gateway.observeProjectOutbound("DETAIL-2"); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started); - OutboundShadowCompletion concurrent = gateway.observeProjectOutbound("DETAIL-3"); + assertThat(concurrent.accepted()).isFalse(); + assertThat(elapsedMillis).isLessThan(1000L); + concurrent.complete("10"); + first.complete("10", "LEGACY-SPACE"); + releasePlanning.countDown(); - assertThat(concurrent.accepted()).isFalse(); - concurrent.complete("10"); - verify(engine, never()).planProjectOutbound("DETAIL-3"); + verify(engine, after(500).never()).compareInbound(any(), anyString(), anyString()); + verify(engine, never()).planProjectOutbound(any(), anyString()); + assertThat(coordinator.activeLegacyCount()).isZero(); + } - first.complete("10", "SPACE-LEGACY"); - releasePlanning.countDown(); + @Test + void rejectedWorkerSubmissionReturnsTrackedSkipAndReleasesAdmission() throws Exception { + ThreadPoolExecutor fullExecutor = executor(); + CountDownLatch holdWorker = new CountDownLatch(1); + fullExecutor.execute(() -> await(holdWorker)); + fullExecutor.execute(() -> { }); + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); + ShadowObservationGateway gateway = gateway(engine, mock(ShadowInputMapper.class), coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { }, fullExecutor); - verify(engine, timeout(2000)).compareInbound(observation, "10", "SPACE-LEGACY"); - } finally { - releasePlanning.countDown(); - legacyCaller.shutdownNow(); + InboundShadowCompletion completion = gateway.observeInbound("TASK-3", "TRAY-3"); + completion.complete("10", "SPACE-3"); + + assertThat(completion.accepted()).isFalse(); + assertThat(coordinator.activeLegacyCount()).isZero(); + verify(engine, never()).planInbound(any(), anyString(), anyString()); + holdWorker.countDown(); + } + + @Test + void planningFailureIsWorkerOnlyAndRecordedByFixedStage() throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); + when(engine.planInbound(SNAPSHOT, "TASK-4", "TRAY-4")) + .thenThrow(new IllegalStateException("sensitive business payload")); + CountDownLatch planMetric = metricLatch(mapper, "ERROR_PLAN"); + stubSuccessfulRecovery(engine, mapper, 43L); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW.plusSeconds(2), ZoneOffset.UTC), delay -> { }); + + InboundShadowCompletion completion = gateway.observeInbound("TASK-4", "TRAY-4"); + completion.complete("10", "SPACE-4"); + + assertThat(planMetric.await(1, TimeUnit.SECONDS)).isTrue(); + verify(engine, never()).compareInbound(any(), anyString(), anyString()); + } + + @Test + void comparisonFailureIsWorkerOnlyAndRecordedByFixedStage() throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); + InboundShadowObservation observation = inboundObservation(); + when(engine.planInbound(SNAPSHOT, "TASK-8", "TRAY-8")).thenReturn(observation); + org.mockito.Mockito.doThrow(new IllegalStateException("comparison failed")) + .when(engine).compareInbound(observation, "10", "SPACE-8"); + CountDownLatch compareMetric = metricLatch(mapper, "ERROR_COMPARE"); + stubSuccessfulRecovery(engine, mapper, 43L); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW.plusSeconds(2), ZoneOffset.UTC), delay -> { }); + + InboundShadowCompletion completion = gateway.observeInbound("TASK-8", "TRAY-8"); + completion.complete("10", "SPACE-8"); + + assertThat(compareMetric.await(1, TimeUnit.SECONDS)).isTrue(); + verify(engine).compareInbound(observation, "10", "SPACE-8"); + } + + @Test + void legacyWaitTimeoutIsRecordedWithoutClosingCallerOwnedActivity() throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); + when(engine.planInbound(SNAPSHOT, "TASK-5", "TRAY-5")).thenReturn(inboundObservation()); + CountDownLatch waitMetric = metricLatch(mapper, "ERROR_LEGACY_WAIT"); + ShadowObservationProperties properties = enabledProperties(); + properties.setTimeoutSeconds(1); + ShadowObservationGateway gateway = gateway(properties, engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW, ZoneOffset.UTC), delay -> { }, executor()); + + InboundShadowCompletion completion = gateway.observeInbound("TASK-5", "TRAY-5"); + + assertThat(waitMetric.await(2, TimeUnit.SECONDS)).isTrue(); + assertThat(coordinator.activeLegacyCount()).isEqualTo(1); + completion.fail(new IllegalStateException("legacy failed later")); + assertThat(coordinator.activeLegacyCount()).isZero(); + } + + @Test + void projectAndExactStockOutboundUseTheLeasedGeneration() throws Exception { + assertOutboundUsesLease(false); + assertOutboundUsesLease(true); + } + + @Test + void refreshRetriesWithOneTwoFourSecondBackoffAndCoalescesRecoveryForOneMinute() throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch failedCycle = new CountDownLatch(1); + when(engine.refreshSnapshot(anyLong(), eq("INSTANCE-1"))).thenAnswer(invocation -> { + int attempt = attempts.incrementAndGet(); + if (attempt <= 4) { + if (attempt == 4) { + failedCycle.countDown(); + } + throw new IllegalStateException("refresh failed"); + } + return new PublishedShadowSnapshot(invocation.getArgument(0), "01", + NOW.plusSeconds(61), "ABC999"); + }); + MutableClock clock = new MutableClock(NOW); + List delays = new ArrayList<>(); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator(), + new ShadowRuntimeMetrics(), clock, delay -> { + delays.add(delay); + clock.advance(delay); + }); + + gateway.start(); + + assertThat(failedCycle.await(1, TimeUnit.SECONDS)).isTrue(); + verify(engine, timeout(1000).times(4)).refreshSnapshot(anyLong(), eq("INSTANCE-1")); + assertThat(delays).containsExactly(Duration.ofSeconds(1), Duration.ofSeconds(2), Duration.ofSeconds(4)); + + InboundShadowCompletion beforeMinute = gateway.observeInbound("TASK-6", "TRAY-6"); + beforeMinute.complete("10", "SPACE-6"); + verify(engine, after(200).times(4)).refreshSnapshot(anyLong(), eq("INSTANCE-1")); + + clock.advance(Duration.ofSeconds(61)); + CountDownLatch recovered = metricLatch(mapper, "SKIP_NOT_READY"); + InboundShadowCompletion afterMinute = gateway.observeInbound("TASK-7", "TRAY-7"); + afterMinute.complete("10", "SPACE-7"); + + verify(engine, timeout(1000).times(5)).refreshSnapshot(anyLong(), eq("INSTANCE-1")); + assertThat(recovered.await(1, TimeUnit.SECONDS)).isTrue(); + } + + private void assertOutboundUsesLease(boolean exactStock) throws Exception { + ShadowPlanningEngine engine = mock(ShadowPlanningEngine.class); + ShadowInputMapper mapper = mock(ShadowInputMapper.class); + ShadowSnapshotCoordinator coordinator = coordinatorWithSnapshot(); + OutboundShadowObservation observation = new OutboundShadowObservation( + true, 42L, "01", NOW, "ABC123", "DETAIL-7", "READY", + "STOCK-7", "SPACE-7", "diagnostic"); + if (exactStock) { + when(engine.planExactStockOutbound(SNAPSHOT, "DETAIL-7", "STOCK-7")).thenReturn(observation); + } else { + when(engine.planProjectOutbound(SNAPSHOT, "DETAIL-7")).thenReturn(observation); } + CountDownLatch compared = new CountDownLatch(1); + doAnswer(invocation -> { + compared.countDown(); + return null; + }).when(engine).compareOutbound(observation, "10"); + stubSuccessfulRecovery(engine, mapper, 43L); + ShadowObservationGateway gateway = gateway(engine, mapper, coordinator, + new ShadowRuntimeMetrics(), Clock.fixed(NOW.plusSeconds(2), ZoneOffset.UTC), delay -> { }); + + OutboundShadowCompletion completion = exactStock + ? gateway.observeExactStockOutbound("DETAIL-7", "STOCK-7") + : gateway.observeProjectOutbound("DETAIL-7"); + completion.complete("10"); + + assertThat(completion.accepted()).isTrue(); + assertThat(compared.await(1, TimeUnit.SECONDS)).isTrue(); + if (exactStock) { + verify(engine).planExactStockOutbound(SNAPSHOT, "DETAIL-7", "STOCK-7"); + } else { + verify(engine).planProjectOutbound(SNAPSHOT, "DETAIL-7"); + } + } + + private ShadowObservationGateway gateway(ShadowPlanningEngine engine, ShadowInputMapper mapper, + ShadowSnapshotCoordinator coordinator, ShadowRuntimeMetrics metrics, Clock clock, + ShadowObservationGateway.Sleeper sleeper) { + return gateway(enabledProperties(), engine, mapper, coordinator, metrics, clock, sleeper, executor()); + } + + private ShadowObservationGateway gateway(ShadowPlanningEngine engine, ShadowInputMapper mapper, + ShadowSnapshotCoordinator coordinator, ShadowRuntimeMetrics metrics, Clock clock, + ShadowObservationGateway.Sleeper sleeper, ThreadPoolExecutor executor) { + return gateway(enabledProperties(), engine, mapper, coordinator, metrics, clock, sleeper, executor); + } + + private ShadowObservationGateway gateway(ShadowObservationProperties properties, + ShadowPlanningEngine engine, ShadowInputMapper mapper, ShadowSnapshotCoordinator coordinator, + ShadowRuntimeMetrics metrics, Clock clock, ShadowObservationGateway.Sleeper sleeper, + ThreadPoolExecutor executor) { + ShadowObservationGateway gateway = new ShadowObservationGateway(properties, engine, mapper, + coordinator, metrics, clock, sleeper, executor, "INSTANCE-1"); + gateways.add(gateway); + return gateway; + } + + private ThreadPoolExecutor executor() { + ThreadPoolExecutor executor = new ThreadPoolExecutor( + 1, 1, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1), + new ThreadPoolExecutor.AbortPolicy()); + executors.add(executor); + return executor; + } + + private ShadowSnapshotCoordinator coordinator() { + return new ShadowSnapshotCoordinator("01", Duration.ofSeconds(60), Clock.fixed(NOW, ZoneOffset.UTC)); + } + + private ShadowSnapshotCoordinator coordinatorWithSnapshot() { + ShadowSnapshotCoordinator coordinator = coordinator(); + ShadowSnapshotCoordinator.RefreshTicket ticket = coordinator.tryBeginRefresh(NOW).orElseThrow(); + assertThat(coordinator.publish(ticket, SNAPSHOT, NOW)).isTrue(); + return coordinator; + } + + private void stubSuccessfulRecovery(ShadowPlanningEngine engine, ShadowInputMapper mapper, long generation) { + ShadowSnapshotDescriptor descriptor = new ShadowSnapshotDescriptor(); + descriptor.setSnapshotGeneration(generation - 1); + when(mapper.selectPublishedSnapshot("01")).thenReturn(descriptor); + when(engine.refreshSnapshot(generation, "INSTANCE-1")).thenReturn( + new PublishedShadowSnapshot(generation, "01", NOW.plusSeconds(2), "ABC124")); + } + + private CountDownLatch metricLatch(ShadowInputMapper mapper, String metricName) { + CountDownLatch latch = new CountDownLatch(1); + doAnswer(invocation -> { + if (metricName.equals(invocation.getArgument(1))) { + latch.countDown(); + } + return null; + }).when(mapper).upsertRuntimeMetric(anyString(), anyString(), anyLong(), anyLong(), + anyLong(), anyLong(), any(Instant.class)); + return latch; + } + + private InboundShadowObservation inboundObservation() { + return new InboundShadowObservation(true, 42L, "01", NOW, "ABC123", + "TASK-1", "READY", "SHADOW-SPACE", "diagnostic"); } private ShadowObservationProperties enabledProperties() { ShadowObservationProperties properties = new ShadowObservationProperties(); properties.setEnabled(true); + properties.setWarehouseId("01"); properties.setTimeoutSeconds(3); + properties.setQuietIntervalMillis(1000); return properties; } + + private boolean waitForPhase(ShadowSnapshotCoordinator coordinator, + ShadowSnapshotCoordinator.Phase expected) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (System.nanoTime() < deadline) { + if (coordinator.phase() == expected) { + return true; + } + Thread.sleep(5L); + } + return coordinator.phase() == expected; + } + + private static void await(CountDownLatch latch) { + try { + latch.await(5, TimeUnit.SECONDS); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + } + + private static final class MutableClock extends Clock { + private Instant instant; + + private MutableClock(Instant instant) { + this.instant = instant; + } + + private void advance(Duration duration) { + instant = instant.plus(duration); + } + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return instant; + } + } } diff --git a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowRuntimeMetricsTest.java b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowRuntimeMetricsTest.java new file mode 100644 index 0000000..d8579b2 --- /dev/null +++ b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowRuntimeMetricsTest.java @@ -0,0 +1,81 @@ +package com.ruoyi.algorithm.shadow; + +import org.junit.jupiter.api.Test; + +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ShadowRuntimeMetricsTest { + + @Test + void recordsFixedCountersAndHistogramBucketsThenDrainsAtomically() { + ShadowRuntimeMetrics metrics = new ShadowRuntimeMetrics(); + + metrics.recordAdmission(ShadowSkipReason.BUSY, 750_000L); + metrics.recordAdmission(null, 1_500_000L); + metrics.recordLegacyDuration(1_500_000L); + metrics.recordError(ShadowFailureStage.PLAN); + + ShadowRuntimeMetrics.Batch batch = metrics.drain(); + + assertThat(batch.count("SKIP_BUSY")).isEqualTo(1); + assertThat(batch.count("ADMITTED")).isEqualTo(1); + assertThat(batch.count("ERROR_PLAN")).isEqualTo(1); + assertThat(batch.bucketCount("ADMISSION_US", 1000L)).isEqualTo(1); + assertThat(batch.bucketCount("ADMISSION_US", 2500L)).isEqualTo(1); + assertThat(batch.bucketCount("LEGACY_DURATION_US", 2500L)).isEqualTo(1); + assertThat(metrics.drain().totalCount()).isZero(); + assertThatThrownBy(() -> batch.rows().clear()) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void recordsOverflowAndDoesNotLoseConcurrentCounterUpdates() throws Exception { + ShadowRuntimeMetrics metrics = new ShadowRuntimeMetrics(); + int workers = 4; + int perWorker = 1000; + ExecutorService executor = Executors.newFixedThreadPool(workers); + try { + for (int worker = 0; worker < workers; worker++) { + executor.submit(() -> { + for (int index = 0; index < perWorker; index++) { + metrics.increment(ShadowSkipReason.CONCURRENT); + } + }); + } + } finally { + executor.shutdown(); + assertThat(executor.awaitTermination(2, TimeUnit.SECONDS)).isTrue(); + } + + metrics.recordAdmission(null, TimeUnit.SECONDS.toNanos(2)); + metrics.recordLegacyDuration(TimeUnit.SECONDS.toNanos(2)); + ShadowRuntimeMetrics.Batch batch = metrics.drain(); + + assertThat(batch.count("SKIP_CONCURRENT")).isEqualTo((long) workers * perWorker); + assertThat(batch.bucketCount("ADMISSION_US", Long.MAX_VALUE)).isEqualTo(1); + assertThat(batch.bucketCount("LEGACY_DURATION_US", Long.MAX_VALUE)).isEqualTo(1); + } + + @Test + void metricPersistenceUsesOneFixedShadowOwnedSqlStatement() throws Exception { + String xml; + try (InputStream stream = Thread.currentThread().getContextClassLoader() + .getResourceAsStream("mapper/algorithm/ShadowInputMapper.xml")) { + assertThat(stream).isNotNull(); + xml = new String(stream.readAllBytes(), StandardCharsets.UTF_8).toLowerCase(); + } + int start = xml.indexOf(""); + int end = xml.indexOf("", start); + String statement = xml.substring(start, end); + + assertThat(statement).contains("merge alg.shadow_runtime_metric") + .doesNotContain("dbo.", "${"); + } +} 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 index e840a77..c7ea2de 100644 --- a/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinatorTest.java +++ b/ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowSnapshotCoordinatorTest.java @@ -61,6 +61,7 @@ class ShadowSnapshotCoordinatorTest { assertThat(coordinator.publish(refresh, snapshot(2, "01", NOW), NOW)).isFalse(); assertThat(coordinator.phase()).isNotEqualTo(ShadowSnapshotCoordinator.Phase.READY); coordinator.exit(concurrent); + assertThat(coordinator.tryBeginRefresh(NOW)).isPresent(); } @Test @@ -130,6 +131,19 @@ class ShadowSnapshotCoordinatorTest { assertThat(coordinator.phase()).isEqualTo(ShadowSnapshotCoordinator.Phase.READY); } + @Test + void staleRefreshTicketCannotCancelANewerRefreshLease() { + ShadowSnapshotCoordinator coordinator = coordinatorAt("01"); + ShadowSnapshotCoordinator.RefreshTicket first = coordinator.tryBeginRefresh(NOW).orElseThrow(); + coordinator.failRefresh(first, "first failed"); + ShadowSnapshotCoordinator.RefreshTicket second = coordinator.tryBeginRefresh(NOW).orElseThrow(); + + coordinator.failRefresh(first, "late duplicate failure"); + + assertThat(coordinator.phase()).isEqualTo(ShadowSnapshotCoordinator.Phase.BUSY); + assertThat(coordinator.publish(second, snapshot(9, "01", NOW), NOW)).isTrue(); + } + private ShadowSnapshotCoordinator coordinatorAt(String warehouseId) { return new ShadowSnapshotCoordinator(warehouseId, Duration.ofSeconds(60)); }