diff --git a/.env.example b/.env.example index 8a30db3..d2c8300 100644 --- a/.env.example +++ b/.env.example @@ -12,6 +12,18 @@ DB_MIGRATION_PASSWORD= # React 개발 서버 또는 배포 Client 주소를 쉼표로 구분합니다. CORS_ALLOWED_ORIGINS=http://localhost:3000,http://localhost:5173 +# AI Runtime은 fowoco/ai의 versioned internal API가 배포된 환경에서만 켭니다. +# Server는 모델 Provider를 직접 호출하지 않으며, 아래 credential은 환경 Secret으로만 주입합니다. +AI_RUNTIME_ENABLED=false +AI_RUNTIME_ENDPOINT=http://127.0.0.1:8000/internal/v1/analyses +# AI_RUNTIME_SERVICE_CREDENTIAL= +AI_RUNTIME_CONNECT_TIMEOUT=2s +AI_RUNTIME_OVERALL_TIMEOUT=15s +AI_RUNTIME_MAX_RESPONSE_BYTES=1048576 +AI_RUNTIME_MAX_CONCURRENT_CALLS=8 +AI_RUNTIME_CIRCUIT_BREAKER_FAILURE_THRESHOLD=5 +AI_RUNTIME_CIRCUIT_BREAKER_OPEN_DURATION=30s + # Transactional Outbox worker 설정입니다. # 일반 실행에서는 켜 두며, 운영 장애 조사 중 자동 처리를 멈춰야 할 때만 false로 둡니다. OUTBOX_ENABLED=true diff --git a/build.gradle b/build.gradle index c0d931b..da5a982 100644 --- a/build.gradle +++ b/build.gradle @@ -37,6 +37,7 @@ dependencies { testImplementation 'org.springframework.boot:spring-boot-starter-security-test' testImplementation 'org.springframework.boot:spring-boot-starter-validation-test' testImplementation 'org.springframework.boot:spring-boot-starter-webmvc-test' + testImplementation 'org.wiremock:wiremock-standalone:3.13.2' testCompileOnly 'org.projectlombok:lombok' testRuntimeOnly 'org.junit.platform:junit-platform-launcher' testAnnotationProcessor 'org.projectlombok:lombok' diff --git a/docs/ai-runtime-contract.md b/docs/ai-runtime-contract.md index 6d89c86..3efef89 100644 --- a/docs/ai-runtime-contract.md +++ b/docs/ai-runtime-contract.md @@ -3,9 +3,9 @@ 이 문서는 `fowoco/server`가 별도 배포되는 `fowoco/ai` Runtime을 호출할 때 지켜야 하는 최소 계약과 방어 규칙을 설명합니다. -현재 단계는 **HTTP 연결 전 계약 기반**입니다. 실제 `/internal/v1/analyses` OpenAPI와 -Structured Output JSON Schema의 원본은 `fowoco/ai`가 소유하며, 원본 계약이 release되면 -Server의 `RemoteAiRuntimeClient`와 fixture를 그 version에 맞춰 연결합니다. +Server에는 `/internal/v1/analyses`를 호출하는 HTTP Adapter까지 구현되어 있습니다. +다만 실제 OpenAPI와 Structured Output JSON Schema의 원본은 `fowoco/ai`가 소유하므로, +AI 저장소에서 같은 `contractVersion`을 release하기 전까지 실제 호출은 기본적으로 꺼 둡니다. ## 초보자용 한 줄 설명 @@ -19,7 +19,10 @@ AiRunWorker (#24, 후속) 2. AiRuntimeClient transport를 정확히 한 번 호출 3. 응답 ID·version·worker·workflow·slot 재검사 → FakeAiRuntimeClient (test) - → RemoteAiRuntimeClient (#8 후속) + → RemoteAiRuntimeClient + 1. deadline·bulkhead·circuit breaker 적용 + 2. Bearer 인증과 추적 header 전달 + 3. 응답 크기 제한과 strict JSON parsing → POST /internal/v1/analyses (fowoco/ai) ``` @@ -128,16 +131,80 @@ Worker Link token을 추가하지 않습니다. - `FakeAiRuntimeClient`: `src/test`에만 있으며 응답이나 예외를 순서대로 예약합니다. - `ValidatingAiRuntimeClient`: transport 앞뒤에서 같은 방어 검증을 수행합니다. -- `RemoteAiRuntimeClient`: 아직 없습니다. AI 원본 계약 release 후 추가합니다. +- `RemoteAiRuntimeClient`: 설정이 켜진 환경에서만 AI Runtime을 HTTP로 한 번 호출합니다. +- `DisabledAiRuntimeClient`: 기본 구현이며, 실수로 호출하면 `RUNTIME_DISABLED`로 즉시 + 실패합니다. LM Studio나 모델 Provider로 우회하지 않습니다. -실제 HTTP 연결 PR에서는 다음을 추가로 검증합니다. +WireMock 계약 테스트는 다음 동작을 검증합니다. -1. `Service-Authorization`, `X-Request-Id`, `traceparent` 전달 -2. 알 수 없는 JSON field와 body size 제한 -3. connect/read/overall deadline -4. circuit breaker와 concurrency bulkhead -5. HTTP·parsing·contract 오류의 안정적인 분류 -6. contract fixture와 WireMock 통합 테스트 +1. `Authorization: Bearer `, `X-Request-Id`, `traceparent` 전달 +2. 문서와 같은 camelCase 요청 JSON 사용 +3. 알 수 없는 JSON field와 제한보다 큰 응답 거부 +4. connect timeout과 요청·응답 전체 deadline +5. circuit breaker와 동시 호출 수 bulkhead +6. HTTP·parsing·contract 오류의 안정적인 `AiRuntimeFailureCode` 분류 +7. 실패 응답에도 HTTP 요청이 한 번만 발생하는지 확인 Remote Client는 자동 HTTP retry를 하지 않습니다. 다시 호출하려면 #24가 먼저 새로운 AiAttempt를 DB에 기록해야 합니다. + +## 실행 설정 + +평소 local 실행과 아직 AI 계약이 배포되지 않은 환경에서는 아래 기본값을 유지합니다. + +```dotenv +AI_RUNTIME_ENABLED=false +``` + +AI Runtime 계약이 배포된 통합 환경에서는 배포 Secret과 함께 설정합니다. + +```dotenv +AI_RUNTIME_ENABLED=true +AI_RUNTIME_ENDPOINT=https://ai.example.com/internal/v1/analyses +AI_RUNTIME_SERVICE_CREDENTIAL=<배포 환경 Secret> +``` + +`AI_RUNTIME_SERVICE_CREDENTIAL`은 Git, 로그, 오류 응답에 남기지 않습니다. Server가 표준 +`Authorization: Bearer ...` 형식으로 조립합니다. `X-Request-Id`는 분석 요청의 +`requestId`와 같고, 상위 요청의 유효한 W3C `traceparent`가 있으면 그대로 전달합니다. + +| 설정 | 기본값 | 의미 | +| --- | --- | --- | +| `AI_RUNTIME_CONNECT_TIMEOUT` | `2s` | AI 서버에 TCP 연결을 맺을 수 있는 최대 시간 | +| `AI_RUNTIME_OVERALL_TIMEOUT` | `15s` | 연결·요청·응답 수신 전체의 Server 상한 | +| `AI_RUNTIME_MAX_RESPONSE_BYTES` | `1048576` | 응답을 메모리에 받기 전 적용하는 최대 크기 | +| `AI_RUNTIME_MAX_CONCURRENT_CALLS` | `8` | Server 한 인스턴스가 동시에 보내는 최대 호출 수 | +| `AI_RUNTIME_CIRCUIT_BREAKER_FAILURE_THRESHOLD` | `5` | 연속 장애 후 호출을 잠시 막는 기준 | +| `AI_RUNTIME_CIRCUIT_BREAKER_OPEN_DURATION` | `30s` | 차단 후 시험 호출까지 기다리는 시간 | + +요청의 `deadlineMs`와 `AI_RUNTIME_OVERALL_TIMEOUT` 중 더 짧은 값을 사용합니다. 따라서 +상위 AiRun이 허용한 시간보다 오래 기다리지 않습니다. + +## 장애가 발생하면 + +| 상황 | 안전한 실패 코드 | 처리 방향 | +| --- | --- | --- | +| 기능 비활성화 | `RUNTIME_DISABLED` | 설정과 AI 계약 release 확인 | +| 동시 호출 한도 초과 | `BULKHEAD_FULL` | #24가 새 AiAttempt로 재시도 여부 결정 | +| 회로 차단 중 | `CIRCUIT_OPEN` | Runtime 복구 대기 | +| 전체 제한시간 초과 | `DEADLINE_EXCEEDED` | 자동 재시도하지 않음 | +| 서비스 인증 실패 | `AUTHENTICATION_FAILED` | 배포 Secret과 audience/scope 확인 | +| `429` | `RATE_LIMITED` | Runtime 정책 확인 후 명시적 재시도 | +| `5xx` | `RUNTIME_UNAVAILABLE` | Runtime 상태 확인 | +| 큰 응답 | `RESPONSE_TOO_LARGE` | 계약과 응답 크기 조사 | +| 잘못된 JSON | `RESPONSE_PARSING_FAILED` | contract version과 schema 조사 | +| 네트워크 오류 | `TRANSPORT_FAILURE` | DNS·TLS·네트워크 상태 확인 | + +오류 메시지에는 credential, endpoint query, 응답 원문을 넣지 않습니다. #24는 안전한 실패 +코드, `requestId`, version, latency만 AiAttempt 진단값으로 저장합니다. + +## AI 저장소와 연결하는 순서 + +1. `fowoco/ai`가 `/internal/v1/analyses` OpenAPI와 JSON Schema를 versioned release로 냅니다. +2. Server의 camelCase fixture와 AI 원본 계약이 같은지 consumer contract test로 확인합니다. +3. staging에 service credential과 endpoint를 Secret으로 주입합니다. +4. 정상·`401`·`429`·`5xx`·timeout smoke test를 통과시킵니다. +5. 그 후에만 `AI_RUNTIME_ENABLED=true`를 적용합니다. + +계약이 다르면 임시 필드나 호환되지 않는 JSON을 Server에 추가하지 않고, 양쪽 저장소에서 +`contractVersion`을 합의한 다음 fixture를 함께 갱신합니다. diff --git a/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeCallException.java b/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeCallException.java new file mode 100644 index 0000000..5f47af5 --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeCallException.java @@ -0,0 +1,29 @@ +package com.fowoco.server.aiintegration.application.error; + +import java.util.Objects; + +/** + * Stable transport failure that does not expose a credential, URI, or Runtime response body. + */ +public final class AiRuntimeCallException extends RuntimeException { + + private final AiRuntimeFailureCode failureCode; + + public AiRuntimeCallException(AiRuntimeFailureCode failureCode, String safeMessage) { + super(safeMessage); + this.failureCode = Objects.requireNonNull(failureCode, "failureCode must not be null"); + } + + public AiRuntimeCallException( + AiRuntimeFailureCode failureCode, + String safeMessage, + Throwable cause + ) { + super(safeMessage, cause); + this.failureCode = Objects.requireNonNull(failureCode, "failureCode must not be null"); + } + + public AiRuntimeFailureCode failureCode() { + return failureCode; + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeFailureCode.java b/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeFailureCode.java index 4387644..d261ed7 100644 --- a/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeFailureCode.java +++ b/src/main/java/com/fowoco/server/aiintegration/application/error/AiRuntimeFailureCode.java @@ -12,5 +12,16 @@ public enum AiRuntimeFailureCode { KNOWLEDGE_VERSION_MISMATCH, UNEXPECTED_WORKER_REFERENCE, UNEXPECTED_WORKFLOW, - UNEXPECTED_SLOT + UNEXPECTED_SLOT, + CORE_VALUE_MISMATCH, + RUNTIME_DISABLED, + BULKHEAD_FULL, + CIRCUIT_OPEN, + DEADLINE_EXCEEDED, + AUTHENTICATION_FAILED, + RATE_LIMITED, + RUNTIME_UNAVAILABLE, + RESPONSE_TOO_LARGE, + RESPONSE_PARSING_FAILED, + TRANSPORT_FAILURE } diff --git a/src/main/java/com/fowoco/server/aiintegration/application/model/AiRuntimeCallContext.java b/src/main/java/com/fowoco/server/aiintegration/application/model/AiRuntimeCallContext.java new file mode 100644 index 0000000..377506d --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/application/model/AiRuntimeCallContext.java @@ -0,0 +1,27 @@ +package com.fowoco.server.aiintegration.application.model; + +import java.util.Locale; +import java.util.regex.Pattern; + +/** + * Transport metadata propagated to the AI Runtime but excluded from the JSON request body. + */ +public record AiRuntimeCallContext(String traceParent) { + + private static final Pattern TRACEPARENT = Pattern.compile( + "^00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$" + ); + + public AiRuntimeCallContext { + if (traceParent != null) { + traceParent = traceParent.trim().toLowerCase(Locale.ROOT); + if (!TRACEPARENT.matcher(traceParent).matches()) { + throw new IllegalArgumentException("traceParent must be a valid W3C traceparent value"); + } + } + } + + public static AiRuntimeCallContext withoutTrace() { + return new AiRuntimeCallContext(null); + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/application/port/AiRuntimeClient.java b/src/main/java/com/fowoco/server/aiintegration/application/port/AiRuntimeClient.java index c55a03f..e40d062 100644 --- a/src/main/java/com/fowoco/server/aiintegration/application/port/AiRuntimeClient.java +++ b/src/main/java/com/fowoco/server/aiintegration/application/port/AiRuntimeClient.java @@ -2,6 +2,7 @@ import com.fowoco.server.aiintegration.application.model.AiAnalysisRequest; import com.fowoco.server.aiintegration.application.model.AiAnalysisResponse; +import com.fowoco.server.aiintegration.application.model.AiRuntimeCallContext; /** * Server-owned port for one attempt against a separately deployed AI Runtime. @@ -11,5 +12,9 @@ @FunctionalInterface public interface AiRuntimeClient { - AiAnalysisResponse analyze(AiAnalysisRequest request); + AiAnalysisResponse analyze(AiAnalysisRequest request, AiRuntimeCallContext context); + + default AiAnalysisResponse analyze(AiAnalysisRequest request) { + return analyze(request, AiRuntimeCallContext.withoutTrace()); + } } diff --git a/src/main/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidator.java b/src/main/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidator.java index d1fcf60..4be3fb1 100644 --- a/src/main/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidator.java +++ b/src/main/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidator.java @@ -72,9 +72,11 @@ public void validateResponse(AiAnalysisRequest request, AiAnalysisResponse respo } Map> allowedSlotsByWorkflow = allowedSlotsByWorkflow(request); - Set allowedWorkers = request.maskedInput().workers().stream() - .map(MaskedWorkerContext::workerRef) - .collect(java.util.stream.Collectors.toUnmodifiableSet()); + Map allowedWorkers = request.maskedInput().workers().stream() + .collect(java.util.stream.Collectors.toUnmodifiableMap( + MaskedWorkerContext::workerRef, + worker -> worker + )); Set candidateRefs = new HashSet<>(); response.candidates().forEach(candidate -> validateCandidate(candidate, allowedWorkers, allowedSlotsByWorkflow, candidateRefs)); @@ -147,14 +149,15 @@ private void validateResponseVersions(AiAnalysisRequest request, AiRuntimeVersio private void validateCandidate( AiCandidate candidate, - Set allowedWorkers, + Map allowedWorkers, Map> allowedSlotsByWorkflow, Set candidateRefs ) { if (!CANDIDATE_REF.matcher(candidate.candidateRef()).matches() || !candidateRefs.add(candidate.candidateRef())) { reject(AiRuntimeFailureCode.INVALID_RESPONSE_CONTRACT, "AI Runtime candidate reference is invalid."); } - if (!allowedWorkers.contains(candidate.workerRef())) { + MaskedWorkerContext worker = allowedWorkers.get(candidate.workerRef()); + if (worker == null) { reject(AiRuntimeFailureCode.UNEXPECTED_WORKER_REFERENCE, "AI Runtime returned an unknown worker reference."); } Set allowedSlots = allowedSlotsByWorkflow.get(candidate.workflowId()); @@ -169,6 +172,7 @@ private void validateCandidate( validateAllowedSlot(key, allowedSlots); privacyPolicy.validateText(value, 4_000, true); }); + validateCoreValues(worker, candidate); Set missingSlots = new HashSet<>(); candidate.missingSlots().forEach(slot -> { validateAllowedSlot(slot, allowedSlots); @@ -178,6 +182,18 @@ private void validateCandidate( }); } + private void validateCoreValues(MaskedWorkerContext worker, AiCandidate candidate) { + String returnedStayExpiryDate = candidate.extractedSlots().get("stay_expiry_date"); + if (returnedStayExpiryDate != null + && worker.stayExpiryDate() != null + && !worker.stayExpiryDate().toString().equals(returnedStayExpiryDate)) { + reject( + AiRuntimeFailureCode.CORE_VALUE_MISMATCH, + "AI Runtime changed a Server-owned core value." + ); + } + } + private void validateAllowedSlot(String slot, Set allowedSlots) { privacyPolicy.validateKey(slot); validateIdentifier(slot, AiRuntimeFailureCode.INVALID_RESPONSE_CONTRACT); diff --git a/src/main/java/com/fowoco/server/aiintegration/application/validation/ValidatingAiRuntimeClient.java b/src/main/java/com/fowoco/server/aiintegration/application/validation/ValidatingAiRuntimeClient.java index dbd8da6..6fb8651 100644 --- a/src/main/java/com/fowoco/server/aiintegration/application/validation/ValidatingAiRuntimeClient.java +++ b/src/main/java/com/fowoco/server/aiintegration/application/validation/ValidatingAiRuntimeClient.java @@ -2,6 +2,7 @@ import com.fowoco.server.aiintegration.application.model.AiAnalysisRequest; import com.fowoco.server.aiintegration.application.model.AiAnalysisResponse; +import com.fowoco.server.aiintegration.application.model.AiRuntimeCallContext; import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; import java.util.Objects; @@ -19,9 +20,9 @@ public ValidatingAiRuntimeClient(AiRuntimeClient delegate, AiRuntimeContractVali } @Override - public AiAnalysisResponse analyze(AiAnalysisRequest request) { + public AiAnalysisResponse analyze(AiAnalysisRequest request, AiRuntimeCallContext context) { validator.validateRequest(request); - AiAnalysisResponse response = delegate.analyze(request); + AiAnalysisResponse response = delegate.analyze(request, context); validator.validateResponse(request, response); return response; } diff --git a/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeCircuitBreaker.java b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeCircuitBreaker.java new file mode 100644 index 0000000..fbd2dcc --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeCircuitBreaker.java @@ -0,0 +1,82 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import com.fowoco.server.aiintegration.application.error.AiRuntimeCallException; +import com.fowoco.server.aiintegration.application.error.AiRuntimeFailureCode; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Objects; + +/** + * Small consecutive-failure circuit breaker for one AI Runtime endpoint. + */ +final class AiRuntimeCircuitBreaker { + + private final int failureThreshold; + private final Duration openDuration; + private final Clock clock; + + private State state = State.CLOSED; + private int consecutiveFailures; + private Instant reopenAt = Instant.EPOCH; + private boolean halfOpenProbeInFlight; + + AiRuntimeCircuitBreaker(int failureThreshold, Duration openDuration, Clock clock) { + if (failureThreshold < 1) { + throw new IllegalArgumentException("failureThreshold must be positive"); + } + if (openDuration == null || openDuration.isZero() || openDuration.isNegative()) { + throw new IllegalArgumentException("openDuration must be positive"); + } + this.failureThreshold = failureThreshold; + this.openDuration = openDuration; + this.clock = Objects.requireNonNull(clock, "clock must not be null"); + } + + synchronized void beforeCall() { + Instant now = clock.instant(); + if (state == State.OPEN && !now.isBefore(reopenAt)) { + state = State.HALF_OPEN; + halfOpenProbeInFlight = false; + } + if (state == State.OPEN || (state == State.HALF_OPEN && halfOpenProbeInFlight)) { + throw new AiRuntimeCallException( + AiRuntimeFailureCode.CIRCUIT_OPEN, + "AI Runtime circuit is open." + ); + } + if (state == State.HALF_OPEN) { + halfOpenProbeInFlight = true; + } + } + + synchronized void recordSuccess() { + state = State.CLOSED; + consecutiveFailures = 0; + halfOpenProbeInFlight = false; + reopenAt = Instant.EPOCH; + } + + synchronized void recordFailure() { + if (state == State.HALF_OPEN) { + open(); + return; + } + consecutiveFailures++; + if (consecutiveFailures >= failureThreshold) { + open(); + } + } + + private void open() { + state = State.OPEN; + halfOpenProbeInFlight = false; + reopenAt = clock.instant().plus(openDuration); + } + + private enum State { + CLOSED, + OPEN, + HALF_OPEN + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeHttpConfiguration.java b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeHttpConfiguration.java new file mode 100644 index 0000000..9aa4f69 --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeHttpConfiguration.java @@ -0,0 +1,57 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; +import com.fowoco.server.aiintegration.application.validation.AiRuntimeContractValidator; +import com.fowoco.server.aiintegration.application.validation.ValidatingAiRuntimeClient; +import java.net.http.HttpClient; +import java.time.Clock; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import tools.jackson.databind.DeserializationFeature; +import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.PropertyNamingStrategies; + +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(AiRuntimeProperties.class) +public class AiRuntimeHttpConfiguration { + + @Bean + public AiRuntimeClient aiRuntimeClient( + AiRuntimeProperties properties, + AiRuntimeContractValidator validator, + ObjectMapper applicationObjectMapper, + Clock clock + ) { + if (!properties.isEnabled()) { + return new DisabledAiRuntimeClient(); + } + properties.validateEnabledConfiguration(); + + ObjectMapper contractObjectMapper = applicationObjectMapper.rebuild() + .propertyNamingStrategy(PropertyNamingStrategies.LOWER_CAMEL_CASE) + .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) + .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS) + .build(); + HttpClient httpClient = HttpClient.newBuilder() + .connectTimeout(properties.getConnectTimeout()) + .followRedirects(HttpClient.Redirect.NEVER) + .build(); + AiRuntimeCircuitBreaker circuitBreaker = new AiRuntimeCircuitBreaker( + properties.getCircuitBreakerFailureThreshold(), + properties.getCircuitBreakerOpenDuration(), + clock + ); + AiRuntimeClient remote = new RemoteAiRuntimeClient( + properties.getEndpoint(), + properties.authorizationHeader(), + properties.getOverallTimeout(), + properties.getMaxResponseBytes(), + properties.getMaxConcurrentCalls(), + httpClient, + contractObjectMapper, + circuitBreaker + ); + return new ValidatingAiRuntimeClient(remote, validator); + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeProperties.java b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeProperties.java new file mode 100644 index 0000000..25a703e --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeProperties.java @@ -0,0 +1,145 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import java.net.URI; +import java.time.Duration; +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "app.ai-runtime") +public final class AiRuntimeProperties { + + private static final int MIN_RESPONSE_BYTES = 1_024; + private static final int MAX_RESPONSE_BYTES = 10 * 1_024 * 1_024; + + private boolean enabled; + private URI endpoint = URI.create("http://127.0.0.1:8000/internal/v1/analyses"); + private String serviceCredential; + private Duration connectTimeout = Duration.ofSeconds(2); + private Duration overallTimeout = Duration.ofSeconds(15); + private int maxResponseBytes = 1_048_576; + private int maxConcurrentCalls = 8; + private int circuitBreakerFailureThreshold = 5; + private Duration circuitBreakerOpenDuration = Duration.ofSeconds(30); + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public URI getEndpoint() { + return endpoint; + } + + public void setEndpoint(URI endpoint) { + this.endpoint = requireHttpEndpoint(endpoint); + } + + public void setServiceCredential(String serviceCredential) { + this.serviceCredential = serviceCredential; + } + + public Duration getConnectTimeout() { + return connectTimeout; + } + + public void setConnectTimeout(Duration connectTimeout) { + this.connectTimeout = requirePositive(connectTimeout, "connectTimeout"); + } + + public Duration getOverallTimeout() { + return overallTimeout; + } + + public void setOverallTimeout(Duration overallTimeout) { + this.overallTimeout = requirePositive(overallTimeout, "overallTimeout"); + } + + public int getMaxResponseBytes() { + return maxResponseBytes; + } + + public void setMaxResponseBytes(int maxResponseBytes) { + if (maxResponseBytes < MIN_RESPONSE_BYTES || maxResponseBytes > MAX_RESPONSE_BYTES) { + throw new IllegalArgumentException("maxResponseBytes must be between 1 KiB and 10 MiB"); + } + this.maxResponseBytes = maxResponseBytes; + } + + public int getMaxConcurrentCalls() { + return maxConcurrentCalls; + } + + public void setMaxConcurrentCalls(int maxConcurrentCalls) { + if (maxConcurrentCalls < 1 || maxConcurrentCalls > 100) { + throw new IllegalArgumentException("maxConcurrentCalls must be between 1 and 100"); + } + this.maxConcurrentCalls = maxConcurrentCalls; + } + + public int getCircuitBreakerFailureThreshold() { + return circuitBreakerFailureThreshold; + } + + public void setCircuitBreakerFailureThreshold(int circuitBreakerFailureThreshold) { + if (circuitBreakerFailureThreshold < 1 || circuitBreakerFailureThreshold > 100) { + throw new IllegalArgumentException("circuitBreakerFailureThreshold must be between 1 and 100"); + } + this.circuitBreakerFailureThreshold = circuitBreakerFailureThreshold; + } + + public Duration getCircuitBreakerOpenDuration() { + return circuitBreakerOpenDuration; + } + + public void setCircuitBreakerOpenDuration(Duration circuitBreakerOpenDuration) { + this.circuitBreakerOpenDuration = requirePositive( + circuitBreakerOpenDuration, + "circuitBreakerOpenDuration" + ); + } + + String authorizationHeader() { + if (serviceCredential == null || serviceCredential.isBlank()) { + throw new IllegalStateException( + "AI_RUNTIME_SERVICE_CREDENTIAL must be configured when AI Runtime is enabled" + ); + } + if (serviceCredential.indexOf('\r') >= 0 || serviceCredential.indexOf('\n') >= 0) { + throw new IllegalStateException("AI Runtime service credential contains an invalid character"); + } + return "Bearer " + serviceCredential.trim(); + } + + void validateEnabledConfiguration() { + requireHttpEndpoint(endpoint); + authorizationHeader(); + requirePositive(connectTimeout, "connectTimeout"); + requirePositive(overallTimeout, "overallTimeout"); + } + + private static URI requireHttpEndpoint(URI value) { + if (value == null + || !value.isAbsolute() + || (!"http".equalsIgnoreCase(value.getScheme()) + && !"https".equalsIgnoreCase(value.getScheme())) + || value.getHost() == null + || value.getUserInfo() != null + || value.getQuery() != null + || value.getFragment() != null) { + throw new IllegalArgumentException("endpoint must be an absolute HTTP(S) URI without credentials or query"); + } + return value; + } + + private static Duration requirePositive(Duration value, String field) { + if (value == null || value.isZero() || value.isNegative()) { + throw new IllegalArgumentException(field + " must be positive"); + } + if (value.compareTo(Duration.ofMillis(100)) < 0) { + throw new IllegalArgumentException(field + " must be at least 100ms"); + } + return value; + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/DisabledAiRuntimeClient.java b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/DisabledAiRuntimeClient.java new file mode 100644 index 0000000..32b89e3 --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/DisabledAiRuntimeClient.java @@ -0,0 +1,19 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import com.fowoco.server.aiintegration.application.error.AiRuntimeCallException; +import com.fowoco.server.aiintegration.application.error.AiRuntimeFailureCode; +import com.fowoco.server.aiintegration.application.model.AiAnalysisRequest; +import com.fowoco.server.aiintegration.application.model.AiAnalysisResponse; +import com.fowoco.server.aiintegration.application.model.AiRuntimeCallContext; +import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; + +final class DisabledAiRuntimeClient implements AiRuntimeClient { + + @Override + public AiAnalysisResponse analyze(AiAnalysisRequest request, AiRuntimeCallContext context) { + throw new AiRuntimeCallException( + AiRuntimeFailureCode.RUNTIME_DISABLED, + "AI Runtime integration is disabled." + ); + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/LimitedByteArrayBodyHandler.java b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/LimitedByteArrayBodyHandler.java new file mode 100644 index 0000000..9d12d2e --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/LimitedByteArrayBodyHandler.java @@ -0,0 +1,97 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import java.io.ByteArrayOutputStream; +import java.net.http.HttpResponse; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Flow; + +/** + * Cancels the HTTP body subscription before an oversized Runtime response is fully buffered. + */ +final class LimitedByteArrayBodyHandler implements HttpResponse.BodyHandler { + + private final int maxBytes; + + LimitedByteArrayBodyHandler(int maxBytes) { + if (maxBytes < 1) { + throw new IllegalArgumentException("maxBytes must be positive"); + } + this.maxBytes = maxBytes; + } + + @Override + public HttpResponse.BodySubscriber apply(HttpResponse.ResponseInfo responseInfo) { + return new LimitedByteArrayBodySubscriber(maxBytes); + } + + static final class ResponseTooLargeException extends RuntimeException { + ResponseTooLargeException() { + super("AI Runtime response exceeded the configured size limit."); + } + } + + private static final class LimitedByteArrayBodySubscriber + implements HttpResponse.BodySubscriber { + + private final int maxBytes; + private final ByteArrayOutputStream output; + private final CompletableFuture body = new CompletableFuture<>(); + private Flow.Subscription subscription; + private int receivedBytes; + + private LimitedByteArrayBodySubscriber(int maxBytes) { + this.maxBytes = maxBytes; + this.output = new ByteArrayOutputStream(Math.min(maxBytes, 8_192)); + } + + @Override + public CompletionStage getBody() { + return body; + } + + @Override + public void onSubscribe(Flow.Subscription subscription) { + if (this.subscription != null) { + subscription.cancel(); + return; + } + this.subscription = subscription; + subscription.request(1); + } + + @Override + public void onNext(List item) { + try { + for (ByteBuffer buffer : item) { + int nextBytes = buffer.remaining(); + if (nextBytes > maxBytes - receivedBytes) { + subscription.cancel(); + body.completeExceptionally(new ResponseTooLargeException()); + return; + } + byte[] chunk = new byte[nextBytes]; + buffer.get(chunk); + output.writeBytes(chunk); + receivedBytes += nextBytes; + } + subscription.request(1); + } catch (RuntimeException exception) { + subscription.cancel(); + body.completeExceptionally(exception); + } + } + + @Override + public void onError(Throwable throwable) { + body.completeExceptionally(throwable); + } + + @Override + public void onComplete() { + body.complete(output.toByteArray()); + } + } +} diff --git a/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/RemoteAiRuntimeClient.java b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/RemoteAiRuntimeClient.java new file mode 100644 index 0000000..3b35105 --- /dev/null +++ b/src/main/java/com/fowoco/server/aiintegration/infrastructure/http/RemoteAiRuntimeClient.java @@ -0,0 +1,290 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import com.fowoco.server.aiintegration.application.error.AiRuntimeCallException; +import com.fowoco.server.aiintegration.application.error.AiRuntimeFailureCode; +import com.fowoco.server.aiintegration.application.model.AiAnalysisRequest; +import com.fowoco.server.aiintegration.application.model.AiAnalysisResponse; +import com.fowoco.server.aiintegration.application.model.AiRuntimeCallContext; +import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.time.Duration; +import java.util.EnumSet; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import tools.jackson.core.JacksonException; +import tools.jackson.databind.ObjectMapper; + +/** + * Exactly-once transport attempt against the separately deployed fowoco/ai Runtime. + */ +public final class RemoteAiRuntimeClient implements AiRuntimeClient { + + private static final String AUTHORIZATION = "Authorization"; + private static final String REQUEST_ID = "X-Request-Id"; + private static final String TRACEPARENT = "traceparent"; + private static final Set CIRCUIT_FAILURES = EnumSet.of( + AiRuntimeFailureCode.DEADLINE_EXCEEDED, + AiRuntimeFailureCode.RATE_LIMITED, + AiRuntimeFailureCode.RUNTIME_UNAVAILABLE, + AiRuntimeFailureCode.RESPONSE_TOO_LARGE, + AiRuntimeFailureCode.RESPONSE_PARSING_FAILED, + AiRuntimeFailureCode.TRANSPORT_FAILURE + ); + + private final URI endpoint; + private final String authorizationHeader; + private final Duration overallTimeout; + private final int maxResponseBytes; + private final HttpClient httpClient; + private final ObjectMapper objectMapper; + private final Semaphore bulkhead; + private final AiRuntimeCircuitBreaker circuitBreaker; + + RemoteAiRuntimeClient( + URI endpoint, + String authorizationHeader, + Duration overallTimeout, + int maxResponseBytes, + int maxConcurrentCalls, + HttpClient httpClient, + ObjectMapper objectMapper, + AiRuntimeCircuitBreaker circuitBreaker + ) { + this.endpoint = Objects.requireNonNull(endpoint, "endpoint must not be null"); + this.authorizationHeader = requireText(authorizationHeader, "authorizationHeader"); + this.overallTimeout = requirePositive(overallTimeout, "overallTimeout"); + if (maxResponseBytes < 1) { + throw new IllegalArgumentException("maxResponseBytes must be positive"); + } + if (maxConcurrentCalls < 1) { + throw new IllegalArgumentException("maxConcurrentCalls must be positive"); + } + this.maxResponseBytes = maxResponseBytes; + this.httpClient = Objects.requireNonNull(httpClient, "httpClient must not be null"); + this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null"); + this.bulkhead = new Semaphore(maxConcurrentCalls); + this.circuitBreaker = Objects.requireNonNull(circuitBreaker, "circuitBreaker must not be null"); + } + + @Override + public AiAnalysisResponse analyze(AiAnalysisRequest request, AiRuntimeCallContext context) { + Objects.requireNonNull(request, "request must not be null"); + Objects.requireNonNull(context, "context must not be null"); + long startedNanos = System.nanoTime(); + if (!bulkhead.tryAcquire()) { + throw new AiRuntimeCallException( + AiRuntimeFailureCode.BULKHEAD_FULL, + "AI Runtime concurrency limit is full." + ); + } + + boolean circuitPermitAcquired = false; + try { + circuitBreaker.beforeCall(); + circuitPermitAcquired = true; + + long remainingMillis = remainingMillis(request, startedNanos); + AiAnalysisRequest outboundRequest = withRemainingDeadline(request, remainingMillis); + byte[] requestBody = serialize(outboundRequest); + remainingMillis = remainingMillis(request, startedNanos); + + HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(endpoint) + .timeout(Duration.ofMillis(remainingMillis)) + .header("Content-Type", "application/json") + .header("Accept", "application/json") + .header(AUTHORIZATION, authorizationHeader) + .header(REQUEST_ID, request.requestId().toString()) + .POST(HttpRequest.BodyPublishers.ofByteArray(requestBody)); + if (context.traceParent() != null) { + requestBuilder.header(TRACEPARENT, context.traceParent()); + } + + HttpResponse response = execute(requestBuilder.build(), remainingMillis); + AiAnalysisResponse result = decodeResponse(response); + remainingMillis(request, startedNanos); + circuitBreaker.recordSuccess(); + return result; + } catch (AiRuntimeCallException exception) { + if (circuitPermitAcquired) { + if (CIRCUIT_FAILURES.contains(exception.failureCode())) { + circuitBreaker.recordFailure(); + } else { + circuitBreaker.recordSuccess(); + } + } + throw exception; + } catch (RuntimeException exception) { + if (circuitPermitAcquired) { + circuitBreaker.recordFailure(); + } + throw new AiRuntimeCallException( + AiRuntimeFailureCode.TRANSPORT_FAILURE, + "AI Runtime transport failed.", + exception + ); + } finally { + bulkhead.release(); + } + } + + private HttpResponse execute(HttpRequest request, long timeoutMillis) { + CompletableFuture> future = httpClient.sendAsync( + request, + new LimitedByteArrayBodyHandler(maxResponseBytes) + ); + try { + return future.get(timeoutMillis, TimeUnit.MILLISECONDS); + } catch (TimeoutException exception) { + future.cancel(true); + throw new AiRuntimeCallException( + AiRuntimeFailureCode.DEADLINE_EXCEEDED, + "AI Runtime deadline was exceeded.", + exception + ); + } catch (InterruptedException exception) { + future.cancel(true); + Thread.currentThread().interrupt(); + throw new AiRuntimeCallException( + AiRuntimeFailureCode.TRANSPORT_FAILURE, + "AI Runtime call was interrupted.", + exception + ); + } catch (ExecutionException exception) { + throw classifyExecutionFailure(exception.getCause()); + } + } + + private AiAnalysisResponse decodeResponse(HttpResponse response) { + int status = response.statusCode(); + if (status < 200 || status >= 300) { + throw classifyStatus(status); + } + try { + return objectMapper.readValue(response.body(), AiAnalysisResponse.class); + } catch (JacksonException exception) { + throw new AiRuntimeCallException( + AiRuntimeFailureCode.RESPONSE_PARSING_FAILED, + "AI Runtime response JSON is invalid.", + exception + ); + } + } + + private byte[] serialize(AiAnalysisRequest request) { + try { + return objectMapper.writeValueAsBytes(request); + } catch (JacksonException exception) { + throw new AiRuntimeCallException( + AiRuntimeFailureCode.INVALID_REQUEST_CONTRACT, + "AI Runtime request JSON could not be created.", + exception + ); + } + } + + private AiRuntimeCallException classifyExecutionFailure(Throwable cause) { + Throwable failure = unwrap(cause); + if (failure instanceof HttpTimeoutException + || failure instanceof java.util.concurrent.TimeoutException) { + return new AiRuntimeCallException( + AiRuntimeFailureCode.DEADLINE_EXCEEDED, + "AI Runtime deadline was exceeded.", + failure + ); + } + if (failure instanceof LimitedByteArrayBodyHandler.ResponseTooLargeException) { + return new AiRuntimeCallException( + AiRuntimeFailureCode.RESPONSE_TOO_LARGE, + "AI Runtime response exceeded the configured size limit.", + failure + ); + } + return new AiRuntimeCallException( + AiRuntimeFailureCode.TRANSPORT_FAILURE, + "AI Runtime transport failed.", + failure + ); + } + + private AiRuntimeCallException classifyStatus(int status) { + if (status == 401 || status == 403) { + return new AiRuntimeCallException( + AiRuntimeFailureCode.AUTHENTICATION_FAILED, + "AI Runtime service authentication failed." + ); + } + if (status == 429) { + return new AiRuntimeCallException( + AiRuntimeFailureCode.RATE_LIMITED, + "AI Runtime rate limit was reached." + ); + } + if (status >= 500) { + return new AiRuntimeCallException( + AiRuntimeFailureCode.RUNTIME_UNAVAILABLE, + "AI Runtime is unavailable." + ); + } + return new AiRuntimeCallException( + AiRuntimeFailureCode.INVALID_REQUEST_CONTRACT, + "AI Runtime rejected the request contract." + ); + } + + private long remainingMillis(AiAnalysisRequest request, long startedNanos) { + long configuredMillis = Math.min(request.deadlineMs(), overallTimeout.toMillis()); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos); + long remainingMillis = configuredMillis - elapsedMillis; + if (remainingMillis < 100) { + throw new AiRuntimeCallException( + AiRuntimeFailureCode.DEADLINE_EXCEEDED, + "AI Runtime deadline was exceeded." + ); + } + return remainingMillis; + } + + private AiAnalysisRequest withRemainingDeadline(AiAnalysisRequest request, long remainingMillis) { + return new AiAnalysisRequest( + request.requestId(), + request.attemptId(), + request.contractVersion(), + request.requiredKnowledgeVersion(), + remainingMillis, + request.maskedInput() + ); + } + + private static Throwable unwrap(Throwable throwable) { + Throwable current = throwable; + while ((current instanceof ExecutionException + || current instanceof java.util.concurrent.CompletionException) + && current.getCause() != null) { + current = current.getCause(); + } + return current; + } + + private static String requireText(String value, String field) { + if (value == null || value.isBlank()) { + throw new IllegalArgumentException(field + " must not be blank"); + } + return value; + } + + private static Duration requirePositive(Duration value, String field) { + if (value == null || value.isZero() || value.isNegative()) { + throw new IllegalArgumentException(field + " must be positive"); + } + return value; + } +} diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index fa70e39..d5db19a 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -37,6 +37,16 @@ springdoc: path: /swagger-ui.html app: + ai-runtime: + enabled: ${AI_RUNTIME_ENABLED:false} + endpoint: ${AI_RUNTIME_ENDPOINT:http://127.0.0.1:8000/internal/v1/analyses} + service-credential: ${AI_RUNTIME_SERVICE_CREDENTIAL:} + connect-timeout: ${AI_RUNTIME_CONNECT_TIMEOUT:2s} + overall-timeout: ${AI_RUNTIME_OVERALL_TIMEOUT:15s} + max-response-bytes: ${AI_RUNTIME_MAX_RESPONSE_BYTES:1048576} + max-concurrent-calls: ${AI_RUNTIME_MAX_CONCURRENT_CALLS:8} + circuit-breaker-failure-threshold: ${AI_RUNTIME_CIRCUIT_BREAKER_FAILURE_THRESHOLD:5} + circuit-breaker-open-duration: ${AI_RUNTIME_CIRCUIT_BREAKER_OPEN_DURATION:30s} reliability: outbox: enabled: ${OUTBOX_ENABLED:true} diff --git a/src/test/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidatorTest.java b/src/test/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidatorTest.java index b51cdde..5449b6a 100644 --- a/src/test/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidatorTest.java +++ b/src/test/java/com/fowoco/server/aiintegration/application/validation/AiRuntimeContractValidatorTest.java @@ -155,6 +155,23 @@ void rejectsSensitiveCandidateValueAndKeepsExceptionMessageSafe() { }); } + @Test + void rejectsCandidateThatChangesServerOwnedStayExpiryDate() { + AiCandidate changedDate = new AiCandidate( + "candidate-changed-date", + WORKER_REF, + WORKFLOW_ID, + Map.of("stay_expiry_date", "2099-01-01"), + List.of("contract_end_date", "monthly_wage"), + BigDecimal.ONE + ); + + assertFailure( + () -> validator.validateResponse(validRequest(), responseWithCandidate(changedDate)), + AiRuntimeFailureCode.CORE_VALUE_MISMATCH + ); + } + private AiAnalysisResponse responseWithVersions(AiRuntimeVersions versions) { return new AiAnalysisResponse( REQUEST_ID, diff --git a/src/test/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeCircuitBreakerTest.java b/src/test/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeCircuitBreakerTest.java new file mode 100644 index 0000000..7c234f5 --- /dev/null +++ b/src/test/java/com/fowoco/server/aiintegration/infrastructure/http/AiRuntimeCircuitBreakerTest.java @@ -0,0 +1,69 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.fowoco.server.aiintegration.application.error.AiRuntimeCallException; +import com.fowoco.server.aiintegration.application.error.AiRuntimeFailureCode; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import org.junit.jupiter.api.Test; + +class AiRuntimeCircuitBreakerTest { + + @Test + void allowsOneHalfOpenProbeAndClosesAfterSuccess() { + MutableClock clock = new MutableClock(); + AiRuntimeCircuitBreaker breaker = new AiRuntimeCircuitBreaker( + 1, + Duration.ofSeconds(30), + clock + ); + breaker.beforeCall(); + breaker.recordFailure(); + + assertCircuitOpen(breaker); + clock.advance(Duration.ofSeconds(30)); + + breaker.beforeCall(); + assertCircuitOpen(breaker); + breaker.recordSuccess(); + breaker.beforeCall(); + breaker.recordSuccess(); + } + + private void assertCircuitOpen(AiRuntimeCircuitBreaker breaker) { + assertThatThrownBy(breaker::beforeCall) + .isInstanceOfSatisfying( + AiRuntimeCallException.class, + exception -> assertThat(exception.failureCode()) + .isEqualTo(AiRuntimeFailureCode.CIRCUIT_OPEN) + ); + } + + private static final class MutableClock extends Clock { + + private Instant instant = Instant.parse("2026-07-26T00:00:00Z"); + + void advance(Duration duration) { + instant = instant.plus(duration); + } + + @Override + public ZoneId getZone() { + return ZoneId.of("UTC"); + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return instant; + } + } +} diff --git a/src/test/java/com/fowoco/server/aiintegration/infrastructure/http/RemoteAiRuntimeClientWireMockTest.java b/src/test/java/com/fowoco/server/aiintegration/infrastructure/http/RemoteAiRuntimeClientWireMockTest.java new file mode 100644 index 0000000..7d30f40 --- /dev/null +++ b/src/test/java/com/fowoco/server/aiintegration/infrastructure/http/RemoteAiRuntimeClientWireMockTest.java @@ -0,0 +1,206 @@ +package com.fowoco.server.aiintegration.infrastructure.http; + +import static com.fowoco.server.aiintegration.support.AiRuntimeContractFixture.REQUEST_ID; +import static com.fowoco.server.aiintegration.support.AiRuntimeContractFixture.validRequest; +import static com.fowoco.server.aiintegration.support.AiRuntimeContractFixture.validResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.absent; +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.equalTo; +import static com.github.tomakehurst.wiremock.client.WireMock.exactly; +import static com.github.tomakehurst.wiremock.client.WireMock.post; +import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; +import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.fowoco.server.aiintegration.application.error.AiRuntimeCallException; +import com.fowoco.server.aiintegration.application.error.AiRuntimeFailureCode; +import com.fowoco.server.aiintegration.application.model.AiAnalysisResponse; +import com.fowoco.server.aiintegration.application.model.AiRuntimeCallContext; +import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; +import com.github.tomakehurst.wiremock.WireMockServer; +import java.net.URI; +import java.net.http.HttpClient; +import java.time.Clock; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import tools.jackson.databind.DeserializationFeature; +import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.PropertyNamingStrategies; + +class RemoteAiRuntimeClientWireMockTest { + + private static final String PATH = "/internal/v1/analyses"; + private static final String TRACEPARENT = + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + + private final ObjectMapper objectMapper = new ObjectMapper().rebuild() + .propertyNamingStrategy(PropertyNamingStrategies.LOWER_CAMEL_CASE) + .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) + .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS) + .build(); + + private WireMockServer wireMock; + + @BeforeEach + void startWireMock() { + wireMock = new WireMockServer(wireMockConfig().dynamicPort()); + wireMock.start(); + } + + @AfterEach + void stopWireMock() { + if (wireMock != null) { + wireMock.stop(); + } + } + + @Test + void sendsCanonicalHeadersAndCamelCaseBodyThenParsesResponse() throws Exception { + wireMock.stubFor(post(urlEqualTo(PATH)) + .withHeader("Authorization", equalTo("Bearer test-service-credential")) + .withHeader("X-Request-Id", equalTo(REQUEST_ID.toString())) + .withHeader("traceparent", equalTo(TRACEPARENT)) + .withRequestBody(com.github.tomakehurst.wiremock.client.WireMock.matchingJsonPath( + "$.requestId", + equalTo(REQUEST_ID.toString()) + )) + .withRequestBody(com.github.tomakehurst.wiremock.client.WireMock.matchingJsonPath( + "$.maskedInput.workers[0].stayExpiryDate" + )) + .withRequestBody(com.github.tomakehurst.wiremock.client.WireMock.matchingJsonPath( + "$.request_id", + absent() + )) + .willReturn(jsonResponse(objectMapper.writeValueAsString(validResponse())))); + + AiAnalysisResponse response = client(1_048_576, 8, 5, Duration.ofSeconds(30)) + .analyze(validRequest(), new AiRuntimeCallContext(TRACEPARENT)); + + assertThat(response).isEqualTo(validResponse()); + wireMock.verify(exactly(1), postRequestedFor(urlEqualTo(PATH))); + } + + @Test + void doesNotRetryAndOpensCircuitAfterConsecutiveRuntimeFailures() { + wireMock.stubFor(post(urlEqualTo(PATH)).willReturn(aResponse().withStatus(503))); + AiRuntimeClient client = client(1_048_576, 8, 2, Duration.ofSeconds(30)); + + assertFailureCode(client, AiRuntimeFailureCode.RUNTIME_UNAVAILABLE); + assertFailureCode(client, AiRuntimeFailureCode.RUNTIME_UNAVAILABLE); + assertFailureCode(client, AiRuntimeFailureCode.CIRCUIT_OPEN); + + wireMock.verify(exactly(2), postRequestedFor(urlEqualTo(PATH))); + } + + @Test + void rejectsUnknownResponseFieldWithStableParsingFailure() throws Exception { + String response = objectMapper.writeValueAsString(validResponse()); + String responseWithUnknownField = response.substring(0, response.length() - 1) + + ",\"unexpected\":\"value\"}"; + wireMock.stubFor(post(urlEqualTo(PATH)).willReturn(jsonResponse(responseWithUnknownField))); + + assertFailureCode( + client(1_048_576, 8, 5, Duration.ofSeconds(30)), + AiRuntimeFailureCode.RESPONSE_PARSING_FAILED + ); + } + + @Test + void enforcesOverallDeadline() { + wireMock.stubFor(post(urlEqualTo(PATH)) + .willReturn(jsonResponse("{}").withFixedDelay(500))); + + AiRuntimeClient client = client( + 1_048_576, + 8, + 5, + Duration.ofMillis(200) + ); + + assertFailureCode(client, AiRuntimeFailureCode.DEADLINE_EXCEEDED); + wireMock.verify(exactly(1), postRequestedFor(urlEqualTo(PATH))); + } + + @Test + void cancelsOversizedResponseBody() { + wireMock.stubFor(post(urlEqualTo(PATH)) + .willReturn(jsonResponse("\"" + "x".repeat(2_000) + "\""))); + + assertFailureCode( + client(1_024, 8, 5, Duration.ofSeconds(30)), + AiRuntimeFailureCode.RESPONSE_TOO_LARGE + ); + } + + @Test + void rejectsConcurrentCallWhenBulkheadIsFull() throws Exception { + wireMock.stubFor(post(urlEqualTo(PATH)) + .willReturn(jsonResponse(objectMapper.writeValueAsString(validResponse())) + .withFixedDelay(500))); + AiRuntimeClient client = client(1_048_576, 1, 5, Duration.ofSeconds(30)); + + CompletableFuture first = CompletableFuture.supplyAsync( + () -> client.analyze(validRequest()) + ); + awaitReceivedRequest(); + + assertFailureCode(client, AiRuntimeFailureCode.BULKHEAD_FULL); + assertThat(first.get(2, TimeUnit.SECONDS)).isEqualTo(validResponse()); + wireMock.verify(exactly(1), postRequestedFor(urlEqualTo(PATH))); + } + + private RemoteAiRuntimeClient client( + int maxResponseBytes, + int maxConcurrentCalls, + int failureThreshold, + Duration overallTimeout + ) { + HttpClient httpClient = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(1)) + .followRedirects(HttpClient.Redirect.NEVER) + .build(); + return new RemoteAiRuntimeClient( + URI.create(wireMock.baseUrl() + PATH), + "Bearer test-service-credential", + overallTimeout, + maxResponseBytes, + maxConcurrentCalls, + httpClient, + objectMapper, + new AiRuntimeCircuitBreaker( + failureThreshold, + Duration.ofSeconds(30), + Clock.systemUTC() + ) + ); + } + + private void assertFailureCode(AiRuntimeClient client, AiRuntimeFailureCode expected) { + assertThatThrownBy(() -> client.analyze(validRequest())) + .isInstanceOfSatisfying( + AiRuntimeCallException.class, + exception -> assertThat(exception.failureCode()).isEqualTo(expected) + ); + } + + private void awaitReceivedRequest() throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (wireMock.getAllServeEvents().isEmpty() && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertThat(wireMock.getAllServeEvents()).hasSize(1); + } + + private com.github.tomakehurst.wiremock.client.ResponseDefinitionBuilder jsonResponse(String body) { + return aResponse() + .withStatus(200) + .withHeader("Content-Type", "application/json") + .withBody(body); + } +} diff --git a/src/test/java/com/fowoco/server/aiintegration/support/FakeAiRuntimeClient.java b/src/test/java/com/fowoco/server/aiintegration/support/FakeAiRuntimeClient.java index 5e85c5d..2ade3dd 100644 --- a/src/test/java/com/fowoco/server/aiintegration/support/FakeAiRuntimeClient.java +++ b/src/test/java/com/fowoco/server/aiintegration/support/FakeAiRuntimeClient.java @@ -2,6 +2,7 @@ import com.fowoco.server.aiintegration.application.model.AiAnalysisRequest; import com.fowoco.server.aiintegration.application.model.AiAnalysisResponse; +import com.fowoco.server.aiintegration.application.model.AiRuntimeCallContext; import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; import java.util.ArrayDeque; import java.util.ArrayList; @@ -31,7 +32,7 @@ public void enqueueFailure(RuntimeException exception) { } @Override - public AiAnalysisResponse analyze(AiAnalysisRequest request) { + public AiAnalysisResponse analyze(AiAnalysisRequest request, AiRuntimeCallContext context) { receivedRequests.add(request); Function script = scripts.pollFirst(); if (script == null) {