fix(lingxing): allow bounded configurable business request pacing

2026-10-13 11:46:57 +08:00
parent 6e9466d0b4
commit 39f914b3b5
4 changed files with 109 additions and 5 deletions
@@ -18,9 +18,10 @@ public record LingxingEgressProperties(@DefaultValue("OFF") CollectionMode colle
        environmentLabel = environmentLabel == null ? "test" : environmentLabel;
        minStartInterval = minStartInterval == null ? Duration.ofSeconds(5) : minStartInterval;
        cooldownFallback = cooldownFallback == null ? Duration.ofMinutes(5) : cooldownFallback;
-        if (minStartInterval.compareTo(Duration.ofSeconds(5)) < 0 || minStartInterval.compareTo(Duration.ofMinutes(5)) > 0
+        Duration minimumInterval = collectionMode == CollectionMode.VALIDATION ? Duration.ofSeconds(5) : Duration.ofMillis(1);
+        if (minStartInterval.compareTo(minimumInterval) < 0 || minStartInterval.compareTo(Duration.ofMinutes(5)) > 0
            || cooldownFallback.compareTo(Duration.ofSeconds(5)) < 0 || cooldownFallback.compareTo(Duration.ofDays(1)) > 0) {
-            throw new IllegalArgumentException("领星出口间隔至少5秒，冷却配置须为有界正值");
+            throw new IllegalArgumentException("领星 VALIDATION 出口间隔至少5秒，其他模式至少1毫秒；间隔和冷却配置须为有界正值");
        }
        if (budgetAuthorityBaseUri != null && (budgetAuthorityBaseUri.getHost() == null
            || !("http".equals(budgetAuthorityBaseUri.getScheme()) || "https".equals(budgetAuthorityBaseUri.getScheme()))
@@ -4,6 +4,8 @@ import com.baiyue.integration.config.LingxingEgressProperties;
 import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.*;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import java.time.Clock;
 import java.time.Instant;
 import java.util.concurrent.Semaphore;
@@ -12,6 +14,7 @@ import java.util.concurrent.TimeUnit;
 /** Local serialization supplements, never replaces, the externally attested window and durable budget. */
@Component
 public final class LingxingEgressGate {
+    private static final Logger log = LoggerFactory.getLogger(LingxingEgressGate.class);
    private final LingxingEgressProperties properties;
    private final LingxingBudgetAuthority authority;
    private final Clock clock;
@@ -73,15 +76,20 @@ public final class LingxingEgressGate {
            try { result = call.run(handle); }
            catch (LingxingWmsException error) { failure = error; }
            catch (RuntimeException error) { failure = failure("LOCAL_PROCESSING_FAILED"); }
+            Instant callFinishedAt = clock.instant();
            CompletionRequest completion = new CompletionRequest(context, operation, hash, handle.actualHttpAttempts,
                failure == null ? null : failure.code(), failure == null ? Integer.valueOf(200) : failure.upstreamStatus(),
                failure == null ? Integer.valueOf(200) : failure.sourceCode(), paused, cooldownUntil);
-            try { authority.complete(completion); }
+            boolean completionConfirmed = false;
+            try { authority.complete(completion); completionConfirmed = true; }
            catch (RuntimeException error) {
                recoverableRatePause = false; paused = true;
                // Unconfirmed completion must remain distinguishable even when the provider already failed.
                failure = failure("BUDGET_COMPLETION_UNCONFIRMED");
            }
+            log.debug("lingxingAttempt operation={} start={} end={} upstreamStatus={} sourceCode={} errorCode={} completionConfirmed={}",
+                operation, handle.startedAt, callFinishedAt, completion.upstreamStatus(), completion.sourceCode(),
+                completion.errorCode(), completionConfirmed);
            if (failure != null) throw failure.withEgress(handle.metadata());
            return result;
        } catch (InterruptedException ignored) {
@@ -112,12 +120,13 @@ public final class LingxingEgressGate {
        private final EgressContext context;
        private final Instant expiresAt;
        private int actualHttpAttempts;
+        private Instant startedAt;
        private AttemptHandle(EgressContext context, Instant expiresAt) { this.context = context; this.expiresAt = expiresAt; }
        /** Immediately before send, after signing. One permit can never send twice. */
        public void startHttp() {
            check(context);
            if (actualHttpAttempts != 0 || !clock.instant().isBefore(expiresAt)) throw failure("ATTEMPT_NOT_SENDABLE");
-            actualHttpAttempts = 1; lastStart = clock.instant();
+            actualHttpAttempts = 1; lastStart = clock.instant(); startedAt = lastStart;
        }
        public void rateLimited(Instant retryAfter) {
            Instant fallback = clock.instant().plus(properties.cooldownFallback());
@@ -0,0 +1,60 @@
+package com.baiyue.integration.config;
+
+import com.baiyue.integration.dto.lingxing.LingxingEgressDtos.CollectionMode;
+import org.junit.jupiter.api.Test;
+import java.time.Duration;
+import java.util.List;
+import static org.assertj.core.api.Assertions.*;
+
+class LingxingEgressPropertiesTest {
+    private LingxingEgressProperties properties(CollectionMode mode, Duration interval, Duration cooldown) {
+        return new LingxingEgressProperties(mode, "test", null, null, null, null, interval, cooldown);
+    }
+
+    @Test void businessAcceptsPositiveMillisecondIntervalsIncludingProbeSteps() {
+        for (long millis : new long[]{1, 1500, 2000, 3000, 5000, 300000}) {
+            assertThat(properties(CollectionMode.BUSINESS, Duration.ofMillis(millis), null).minStartInterval())
+                    .isEqualTo(Duration.ofMillis(millis));
+        }
+    }
+
+    @Test void businessRejectsSubMillisecondNonPositiveAndExcessiveIntervals() {
+        for (Duration interval : List.of(Duration.ofMillis(-1), Duration.ZERO,
+                Duration.ofNanos(999999), Duration.ofMinutes(5).plusMillis(1))) {
+            assertThatThrownBy(() -> properties(CollectionMode.BUSINESS, interval, null))
+                    .isInstanceOf(IllegalArgumentException.class);
+        }
+    }
+
+    @Test void validationRetainsFiveSecondMinimumAndAllModesDefaultToFiveSeconds() {
+        for (CollectionMode mode : CollectionMode.values()) {
+            assertThat(properties(mode, null, null).minStartInterval()).isEqualTo(Duration.ofSeconds(5));
+            if (mode == CollectionMode.VALIDATION) {
+                assertThatThrownBy(() -> properties(mode, Duration.ofMillis(4999), null))
+                        .isInstanceOf(IllegalArgumentException.class);
+                assertThat(properties(mode, Duration.ofSeconds(5), null).minStartInterval())
+                        .isEqualTo(Duration.ofSeconds(5));
+            }
+        }
+        assertThat(properties(null, null, null).collectionMode()).isEqualTo(CollectionMode.OFF);
+    }
+
+    @Test void switchingBusinessOffCanRetainItsThreeSecondInterval() {
+        var business = properties(CollectionMode.BUSINESS, Duration.ofSeconds(3), null);
+        assertThat(properties(CollectionMode.OFF, business.minStartInterval(), null).minStartInterval())
+                .isEqualTo(Duration.ofSeconds(3));
+        assertThatThrownBy(() -> properties(CollectionMode.OFF, Duration.ZERO, null))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+
+    @Test void businessIntervalChangeDoesNotRelaxCooldownBounds() {
+        for (Duration cooldown : List.of(Duration.ofMillis(4999), Duration.ofDays(1).plusMillis(1))) {
+            assertThatThrownBy(() -> properties(CollectionMode.BUSINESS, Duration.ofMillis(1), cooldown))
+                    .isInstanceOf(IllegalArgumentException.class);
+        }
+        for (Duration cooldown : List.of(Duration.ofSeconds(5), Duration.ofDays(1))) {
+            assertThat(properties(CollectionMode.BUSINESS, Duration.ofMillis(1), cooldown).cooldownFallback())
+                    .isEqualTo(cooldown);
+        }
+    }
+}
@@ -13,6 +13,40 @@ import static org.assertj.core.api.Assertions.*;
 import static org.junit.jupiter.api.Assertions.assertThrows;

 class LingxingEgressGateTest {
+    @Test void debugTimingRecordsActualCallAndCompletionWithoutContextOrPayload() {
+        var logger = (ch.qos.logback.classic.Logger) org.slf4j.LoggerFactory.getLogger(LingxingEgressGate.class);
+        var previous = logger.getLevel();
+        var appender = new ch.qos.logback.core.read.ListAppender<ch.qos.logback.classic.spi.ILoggingEvent>();
+        appender.start(); logger.addAppender(appender); logger.setLevel(ch.qos.logback.classic.Level.DEBUG);
+        try {
+            var gate = gate();
+            Object body = new Object();
+            assertThat(gate.<Object>execute(context("private-attempt"), Operation.OUTBOUND_DETAIL, HASH, handle -> {
+                handle.startHttp(); clock.advanceMillis(123); return body;
+            })).isSameAs(body);
+            assertThat(appender.list).hasSize(1);
+            assertThat(appender.list.get(0).getFormattedMessage())
+                .contains("operation=OUTBOUND_DETAIL", "start=" + NOW, "end=" + NOW.plusMillis(123),
+                    "upstreamStatus=200", "errorCode=null", "completionConfirmed=true")
+                .doesNotContain("demo-account", "private-attempt", "demo-grant", HASH);
+            ledger.failComplete = true;
+            var failure = assertThrows(LingxingWmsException.class, () -> gate.execute(context("private-failure"),
+                Operation.OUTBOUND_DETAIL, HASH, handle -> {
+                    handle.startHttp(); clock.advanceMillis(50); throw rateLimit();
+                }));
+            assertThat(failure.code()).isEqualTo("LINGXING_WMS_BUDGET_COMPLETION_UNCONFIRMED");
+            assertThat(appender.list).hasSize(2);
+            assertThat(appender.list.get(1).getFormattedMessage())
+                .contains("upstreamStatus=429", "errorCode=RATE_LIMIT", "completionConfirmed=false")
+                .doesNotContain("private-failure", "synthetic rate limit", "demo-account", HASH);
+            logger.setLevel(ch.qos.logback.classic.Level.INFO);
+            ledger.failComplete = false;
+            gate().execute(context("debug-disabled"), Operation.WAREHOUSES, HASH, handle -> body);
+            assertThat(appender.list).hasSize(2);
+        } finally {
+            logger.detachAppender(appender); appender.stop(); logger.setLevel(previous);
+        }
+    }
    static final Instant NOW = Instant.parse("2026-10-12T00:00:00Z");
    static final String HASH = "a".repeat(64);
    private final MutableClock clock = new MutableClock(NOW);
@@ -159,7 +193,7 @@ class LingxingEgressGateTest {
        } finally { release.countDown(); executor.shutdownNow(); }
    }
    @Test void intervalCannotBeConfiguredBelowFiveSecondsAndPropertiesHideGrant() {
-        assertThatThrownBy(() -> new LingxingEgressProperties(null, null, null, null, null, null, Duration.ofMillis(4999), null)).isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(() -> new LingxingEgressProperties(CollectionMode.VALIDATION, null, null, null, null, null, Duration.ofMillis(4999), null)).isInstanceOf(IllegalArgumentException.class);
        assertThat(props(CollectionMode.VALIDATION, "test").toString()).doesNotContain("demo-grant", "demo-account", "127.0.0.1");
    }
    @Test void knownBudgetDenialDoesNotPermanentlyPauseOrConsumeHttp() {