feat: run shadow v3 without legacy waiting

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> legacyResult = new CompletableFuture<>();
+    private final Runnable onLegacyExit;
+    private final ShadowRuntimeMetrics metrics;
+    private final CompletableFuture<LegacyResult> 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) {
     }
 }

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<String> legacyResult = new CompletableFuture<>();
+    private final Runnable onLegacyExit;
+    private final ShadowRuntimeMetrics metrics;
+    private final CompletableFuture<LegacyResult> 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) {
+    }
 }

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
+}

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<Instant> 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<ShadowSnapshotCoordinator.RefreshTicket> 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<ShadowSnapshotCoordinator.RefreshTicket> 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);
     }
 }

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<MetricValue> 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<Row> 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<String> 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<Row> rows;
+
+        private Batch(List<Row> rows) {
+            this.rows = List.copyOf(rows);
+        }
+
+        public List<Row> 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();
+        }
+    }
+}

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> 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;

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

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}
     </insert>
+
+    <update id="upsertRuntimeMetric">
+        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 &gt; #{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});
+    </update>
 </mapper>