diff --git a/integration-service/src/main/java/com/baiyue/integration/config/LingxingEgressProperties.java b/integration-service/src/main/java/com/baiyue/integration/config/LingxingEgressProperties.java index b09ec18ac..b59ace7b4 100644 --- a/integration-service/src/main/java/com/baiyue/integration/config/LingxingEgressProperties.java +++ b/integration-service/src/main/java/com/baiyue/integration/config/LingxingEgressProperties.java @@ -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())) diff --git a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingEgressGate.java b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingEgressGate.java index 6058e53f7..e01f5be97 100644 --- a/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingEgressGate.java +++ b/integration-service/src/main/java/com/baiyue/integration/provider/lingxing/LingxingEgressGate.java @@ -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()); diff --git a/integration-service/src/test/java/com/baiyue/integration/config/LingxingEgressPropertiesTest.java b/integration-service/src/test/java/com/baiyue/integration/config/LingxingEgressPropertiesTest.java new file mode 100644 index 000000000..1da7522cc --- /dev/null +++ b/integration-service/src/test/java/com/baiyue/integration/config/LingxingEgressPropertiesTest.java @@ -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); + } + } +} diff --git a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingEgressGateTest.java b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingEgressGateTest.java index 82d7712b0..c7a2da739 100644 --- a/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingEgressGateTest.java +++ b/integration-service/src/test/java/com/baiyue/integration/provider/lingxing/LingxingEgressGateTest.java @@ -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(); + appender.start(); logger.addAppender(appender); logger.setLevel(ch.qos.logback.classic.Level.DEBUG); + try { + var gate = gate(); + Object body = new Object(); + assertThat(gate.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() {