diff --git a/docs/contracts/run-cancel.md b/docs/contracts/run-cancel.md new file mode 100644 index 0000000..a987b78 --- /dev/null +++ b/docs/contracts/run-cancel.md @@ -0,0 +1,139 @@ +# run 취소 — 명시적 cancel·CANCELED 확정·상태별 사용량 (task 1.84, #224) + +브라우저 `POST /api/ai-runs/{runId}/cancel`(browser-run-v1 `cancel/cases.json`)의 구현이다. 계약 의미는 바꾸지 않았다. +상태 조회 `GET /api/ai-runs/{runId}`는 1.86(`RunReadService.status`) 그대로다(`cancelable` = PENDING|RUNNING). + +코드: `com.proovy.domain.run.cancel` — `RunCancelService`(진입점), `RunCancelStore`(잠금·조건부 전환), `RunCancelFinalizer`(CANCELED 확정), +`RunCancelProperties`(`proovy.run.cancel.*`). 경로는 1.86 `BrowserRunController`에 추가했다. **schema 변경 없음**(V1 cancel 열·trigger 그대로, migration 추가 없음 — V5는 1.85 `recovery_count`). + +## 판정 순서 (`cancel/cases.json` precedence) + +`401(filter) → body 있음 RUN4001(run 조회 전, `{}`·어떤 필드든) → run 접근 RUN4041 → 상태` + +- body는 원문 `String`으로 받는다(Content-Type 무관). 비어 있지 않은 body면 거부한다. +- RUN4041: 없는 run·타인 run·대문자/형식이 다른 runId·Note DELETING을 구분하지 않는 같은 body(1.86 `accessibleRun` 재사용, 잠금 뒤 다시 확인). + +| 잠금 뒤 상태 | 응답 `{status, outcome}` | 같은 transaction의 쓰기 | +|---|---|---| +| PENDING | `CANCELED`, `accepted` | `PENDING∧attempt=0 → CANCELED`, message CANCELED, `run.canceled`, 보상 `PENDING_CANCELED` | +| RUNNING·RETRYING | `CANCELING`, `accepted` | `→ CANCELING`(현재 attempt·fence 조건, `cancel_requested_from` = 이전 상태, grant 폐기), `run.canceling` | +| CANCELING | `CANCELING`, `already_canceling` | 없음 (확정 확인을 한 번 깨운다) | +| COMPLETED·FAILED·CANCELED | 그 상태, `already_terminal` | 없음 | + +잠금: `notes FOR SHARE → ai_runs FOR UPDATE → credit_balance`(1.82와 같은 Note 우선 순서). 직렬화·교착·연결 실패만 최대 3번 다시 연다. +coordinator claim은 `SKIP LOCKED` + `status='PENDING' AND attempt=0`이라 PENDING 취소와 하나만 이긴다(claim이 이기면 RUNNING 취소 경로). + +## CANCELED 확정 (`RunCancelFinalizer`) + +취소 commit 직후 한 번, 그 뒤 `sweep-interval`마다 DB의 CANCELING run(오래된 요청부터 `batch-size`)을 확인한다. 재시작·다른 인스턴스의 취소도 +이어받고, 확정은 `status='CANCELING'` 조건부라 여러 인스턴스가 겹쳐도 하나만 이긴다. run별 동시 확인은 하나다. + +한 번의 확인 = 현재 attempt·fence·cause로 Brain `cancel`(멱등). 처음이면 취소 전달, 그 뒤로는 ack `state`가 정리 상태 조회다. + +| ack state | 판정 | +|---|---| +| `terminal` (run.canceled 저장, 또는 취소 전 끝난 결과) | CANCELED | +| `absent` (record 없음 = 미시작) | run lease가 만료됐고 이 인스턴스에서 실행 중이 아니면 CANCELED. 살아 있는 worker가 아직 stream을 보낼 수 있으면 기다린다. 확정 transaction에서 run을 잠근 뒤 lease 만료를 **다시** 본다(Brain 호출 동안 늦은 heartbeat가 갱신했으면 기다림) | +| `canceling`·`executing`·호출 실패(5xx·timeout·409 등)·Brain client 없음 | 기다린다 | +| (모든 경우) `cancel_requested_at + cleanup-timeout` 경과 | WARN 뒤 CANCELED (안전 timeout) | + +- timeout 시각은 **DB `now()` 기준**이다(`cancel_requested_at = now()`로 쓰고, 확인 SELECT가 남은 시간을 DB에서 계산). 인스턴스 사이 JVM 시계 차이가 확정 시점을 바꾸지 않는다. +- 이미 timeout이 지난 run은 **Brain을 부르지 않고** 확정한다. 남은 시간이 client `request-timeout`보다 짧으면 Brain 응답을 남은 시간까지만 기다린다(hang인 Brain이 확정을 늦추지 않음). + +확정 transaction(하나): `CANCELING → CANCELED`(lease·grant 해제), message CANCELED, `run.canceled{cause, message}`, 1.77 `settle(canceled(from, cause))`, +`RunClaimStore.reserveRuntimeCleanup`(attempt ≥ 1, 지연은 `proovy.run.coordinator.runtime-cleanup-delay`, coordinator가 꺼진 인스턴스는 0s). +cleanup 예약이 같은 transaction이라 commit 뒤 응답 유실·process 사망·예약 SQL 오류로 `runtime_cleanup_at`이 NULL로 남는 CANCELED는 없다(예약이 실패하면 확정 전체가 rollback되고 다시 확인한다). +commit 뒤 `RunCoordinator.release(runId)` — commit 여부를 모르는 재시도가 이미 CANCELED를 보더라도 부른다. coordinator 패키지는 고치지 않았다(공개 hook만). + +- **실패 행**: 확인이 예외로 끝난 run은 `failure-backoff`(기본 30s) 동안 주기 확인 SQL에서 빠진다(`run_id <> ALL(…)`) — 계속 실패하는 오래된 행이 + oldest-first batch를 막지 않는다. 일시 DB 오류는 WARN, 그 밖(불변조건 위반: 열린 assistant placeholder 없음·seq 불일치 등)은 ERROR, counter + `proovy.run.cancel.check_failures{kind=transient|invariant}`. 같은 run의 취소 API 재요청(`wake`)은 backoff와 무관하게 곧바로 확인한다. + 불변조건 위반으로 계속 실패하는 run은 timeout 확정도 같은 transaction을 타므로 CANCELING에 남고 슬롯·Note 자리를 쥔다 — **운영 수동 개입 대상**(ERROR·counter로 감지). +- counter `proovy.run.cancel.confirmed{reason=brain_terminal|brain_absent|cleanup_timeout}`. 주기 SELECT 자체가 실패하면 처음 한 번 WARN, 회복 시 INFO(DB 장애 중 로그 반복 없음). + +- 확정 전까지 run은 비terminal이라 **전역 슬롯(1.83 DB 수)과 Note Active Run 자리(`ux_ai_runs_note_active`)를 계속 차지한다**. 같은 Note 새 요청은 RUN4091. +- CANCELING 동안: Activity는 열린 Activity의 `activity.canceled`만(1.80), terminal·`failByServer`·`expireDeadline`은 NOT_ACCEPTING(1.82), deadline sweep 대상 아님. + worker는 stream을 마저 소비해 wind-down 표식을 저장하고 status `canceling`/`run.canceled`로 OBSOLETE로 끝난다(1.83). +- 스레드: 내부 executor 2개(`runCancel-*`), Spring `Executor`·`TaskScheduler` bean 아님. Brain 호출 중 transaction·row lock 없음. + stop 때 `inFlight`·backoff를 비워 stop → start(context restart) 뒤에도 run이 거부되지 않는다. +- lifecycle phase는 web server start/stop·graceful shutdown보다 앞(`DEFAULT_PHASE − 4096`): 요청을 받기 전에 시작하고 요청이 멈춘 뒤에 멈춘다. + 그 밖 구간(executor 없음)의 `wake`는 버려지고 주기 확인(다른 인스턴스·재시작 포함)이 줍는다. **`enabled=false`만으로 구성한 배포에서는 그런 run이 CANCELING에 남는다** — 점검·테스트 전용. + +## 사용량 (1.77 정책표 그대로) + +| 취소 | 순사용량 | +|---|---| +| PENDING에서 (어떤 cause든) | 0 — `PENDING_CANCELED` 한 번 | +| RUNNING·RETRYING에서 `user_requested` | 1 — 차감 유지(Brain 정리 확인·timeout 무관) | + +ledger PK(request_id)로 PENDING 취소 보상과 deadline·Server 실패 보상이 겹쳐도 한 번이다(경합 테스트). + +## 설정 (`proovy.run.cancel.*`, `@DefaultValue` — application.yaml 변경 없음) + +| key | 기본 | 의미 | +|---|---|---| +| `enabled` | true | 주기 확인. 끄면 취소 직후 한 번만 확인(점검·테스트용) | +| `sweep-interval` | 2s | CANCELING 재확인 주기(run당 Brain cancel 1회) | +| `cleanup-timeout` | 2m | Brain 정리 확인 상한. 근거: Brain wind-down 15s, 끊긴 wind-down sweep은 writer lease 만료 뒤, Brain tail 4 × lease TTL 30s. **S14 확정** | +| `batch-size` | 20 | 주기 한 번 확인 수 | +| `failure-backoff` | 30s | 확인이 실패한 run을 주기 확인에서 빼 두는 시간 | + +**부하(S14)**: 주기 확인은 leader·lease 게이팅 없이 모든 인스턴스가 돈다 — N 인스턴스 × batch × (1/`sweep-interval`)만큼 Brain cancel(멱등)이 나간다 +(기본 20행·2s면 인스턴스당 최대 초당 10회). 정확성 문제는 없고(확정은 조건부), coordinator lease 보유자 전용 sweep 또는 `FOR UPDATE SKIP LOCKED` 분할은 +S14 부하 인수에서 정한다. + +## 반복 요청 제한 (1.79 재사용 — 사용자 결정: 추가 상한은 후속 태스크) + +이 태스크 범위는 1.79 RUN4291(`proovy.run.admission.rate-limit`, 기본 20건/1분, 임시값) 재사용뿐이다. 테스트로 고정한 성질: +새 requestId마다 세고, **취소·실패·clarification으로 순사용량 0이 된 run도 센다**(취소로 우회 불가), 같은 requestId 재조회(replay)는 세지 않고, +다른 사용자에 영향 없음. + +**후속 태스크(S14 수치와 함께)** — task 84 문구 대비 남은 차이와 제안: + +1. 반복 실패/clarification 남용을 "구분"하는 별도 제한이 없다 — 모든 새 요청을 같은 창으로 센다. 무료 결과(보상된 run) 누적 상한(예: 1시간 N건)이 없어 + 20건/분 안에서는 무료 Brain 실행을 계속 만들 수 있다. +2. "입력·준비 상태 보완 요청" 구별이 없다 — 보완 요청도 같은 창으로 센다(더 엄격한 제한이 없으므로 부당하게 막히지도 않는다). +3. 제안: 보상된 run을 browser fingerprint(`run_request_bindings`) 기준으로 세는 추가 상한(같은 fingerprint 반복만 제한 → 보완 요청은 fingerprint가 달라 제외). + browser-run-v1 오류 code는 RUN4291 그대로 쓸 수 있다. 창·상한 수치는 기존 20건/1분과 함께 S14에서 정한다. + +## 1.85 회수 결정표 중 1.84 소유 항목 + +1.85(#225, `docs/contracts/run-coordinator.md` stale 회수 절)가 1.84에 넘긴 두 항목의 판정이다. + +- **`FINAL_CANCEL_PENDING`(Brain canceling인데 Server RUNNING|RETRYING)** — 1.84 흐름에서는 생기지 않는다. Brain `cancel`을 부르는 곳은 + `RunCancelFinalizer` 하나이고, 대상은 `status = 'CANCELING'` 행만 읽어 고른다(취소 API는 CANCELING을 commit한 **뒤** `wake`, sweep도 같은 조건). + CANCELING → RUNNING|RETRYING 전이는 없으므로 Server가 Brain보다 늦게 취소 상태가 되는 순서가 없다. 그래서 1.84는 이 경우를 수습하지 않는다 + (coordinator 패키지 변경 금지이기도 하다). 이 counter·WARN이 보이면 Server 밖에서 Brain cancel이 들어온 것(운영 수동 호출, S13 `asset_deleting` 연결 오류 등)이고, + run은 1.85 규칙대로 deadline sweep이 timeout으로 끝낸다. +- **`FINAL_CHECK`·재전달 소진 실패 직전 Brain cancel** — **이 PR에서 하지 않는다.** 실패시키는 곳이 `RunExecution`(coordinator 패키지, 이 태스크에서 수정 금지)이고, + internal-run-v1 `CancelCause`(`user_requested`·`asset_deleting`·`budget_exhausted`)에 Server 실패를 뜻하는 값이 없어 계약 추가가 먼저다. + 그동안 슬롯 over-commit은 1.85 r185-3대로 Brain 자체 상한에 기댄다. 후속(S13, Brain cleanup·lease 만료 sweep과 함께): coordinator가 실패 직전 + `BrainRunClient.cancel`(1.84가 쓰는 client 그대로, 멱등)을 best-effort로 부르고, 응답과 무관하게 `failByServer`한다 — 확정을 Brain 응답에 묶지 않는다. +- 1.85 merge 뒤 확인: 회수 후보·인수는 RUNNING|RETRYING만이고 heartbeat 조건은 CANCELING을 포함하므로, 인수 worker가 도는 동안 취소된 run은 + lease가 살아 있어 `absent` 확정이 기다리고(위 표), worker는 OBSOLETE로 끝난다. 1.84 코드 변경은 필요 없었다. + +## 검증 + +- `RunCancelPostgresTest`(실제 Postgres + 실제 HTTP + 실제 `BrainRunClient`·loopback 가짜 Brain, coordinator 끔): `cancel/cases.json` 9 case를 production + endpoint로(fixture body 일치, 404 body 동일), 대문자 ID·DELETING Note·401, PENDING 취소(payload = `canceled-pending` fixture, 보상 1회, Brain 미호출, 슬롯 해제), + 동시 cancel 6건, PENDING 취소 ↔ claim 8회, PENDING 취소 ↔ deadline·Server 실패, RUNNING 취소(늦은 terminal·Activity·deadline 차단, RUN4091, 슬롯 계산, + Brain cancel 인자, 확정 뒤 차감 유지·cleanup 예약), cleanup timeout(정리 미확인·Brain 503), absent + lease, 완료 ↔ 취소, deadline ↔ 취소, SSE 단절 ≠ 취소, + 반복 요청 제한, absent 확정의 잠금 뒤 lease 재확인(Brain 호출 중 heartbeat 갱신), timeout이 지난 run의 Brain 미호출·hang Brain의 남은 시간 상한, + 확정 실패 run의 backoff(확정·cleanup 예약 전체 rollback, 재요청 `wake`는 즉시 확인). 두 class는 `@DirtiesContext`로 끝난 context의 sweep을 닫는다. +- `RunCancelCoordinatorPostgresTest`(실제 coordinator worker, 슬롯 1): wind-down `activity.canceled` 저장, worker OBSOLETE 뒤에도 CANCELING이 슬롯을 잡아 + 다음 run이 PENDING, Brain 정리 확인 뒤 CANCELED → 다음 run dispatch, 취소 run 재전달 없음. event 순서는 `canceled-running` fixture와 같다. + +## 범위 밖·인계 + +- **1.87**: 경로 `POST /api/ai-runs/{runId}/cancel`(body 없음, `ApiResponse`). 실제 Brain cancel 왕복·wind-down 시간 실측, timeout 수치 확인. +- **1.85**: CANCELING run(lease 만료 포함)은 이 경로 소유다. 1.85 회수는 RUNNING|RETRYING만 다룬다. +- **S13**: `asset_deleting` cause 취소 연결(이 store의 `cancelPending`·`markCanceling`이 cause를 받는다), Brain runtime `cleanup` 호출, + `RunCancelStore.inTransactions`는 호출자 transaction이 열려 있으면 거기 참여하고 **재시도하지 않는다**(RunTerminalService와 같은 규칙) — 재시도가 필요하면 + S13 쪽 transaction 밖에서 부른다. commit 뒤 `RunCancelFinalizer.wake`도 S13이 부른다. + 1.85 `FINAL_CHECK`·재전달 소진 실패 직전 best-effort Brain cancel(위 1.85 절, `CancelCause` 계약 추가 필요). +- **후속 태스크(S14 수치와 함께)**: 보상된 run의 같은 fingerprint 반복 추가 상한(위 반복 요청 제한 절). 수치 확정 대상: `cleanup-timeout` 2m, + `sweep-interval` 2s, 1.79 rate limit 20건/1분, 추가 상한의 창·건수. +- **S08**: 정지 버튼·draft 유지(서버는 자동 재실행·queue·draft 전송을 만들지 않는다). +- Brain cancel 409(fence 불일치·다른 epoch)는 기다림으로 처리해 timeout 뒤 확정된다(새 attempt 금지 전제상 발생하지 않아야 한다). +- 후속 정리: 재시도 SQLSTATE 판정(`isRetryable`)이 RunTerminalService·RunAdmissionService·RunCancelStore 세 곳에 있다 — 공통 helper(추가형 파일)로 모으는 후속을 둔다. + `cancel_cause` 해석은 `RunCancelCause.Converter`를 재사용한다. diff --git a/src/main/java/com/proovy/domain/run/cancel/RunCancelConfig.java b/src/main/java/com/proovy/domain/run/cancel/RunCancelConfig.java new file mode 100644 index 0000000..3e7f526 --- /dev/null +++ b/src/main/java/com/proovy/domain/run/cancel/RunCancelConfig.java @@ -0,0 +1,9 @@ +package com.proovy.domain.run.cancel; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Configuration; + +@Configuration +@EnableConfigurationProperties(RunCancelProperties.class) +public class RunCancelConfig { +} diff --git a/src/main/java/com/proovy/domain/run/cancel/RunCancelFinalizer.java b/src/main/java/com/proovy/domain/run/cancel/RunCancelFinalizer.java new file mode 100644 index 0000000..94c5492 --- /dev/null +++ b/src/main/java/com/proovy/domain/run/cancel/RunCancelFinalizer.java @@ -0,0 +1,367 @@ +package com.proovy.domain.run.cancel; + +import com.proovy.domain.run.brain.InternalRunContractViolation; +import com.proovy.domain.run.brain.RunCancelAck; +import com.proovy.domain.run.brain.RunCancelRequest; +import com.proovy.domain.run.cancel.RunCancelStore.Canceled; +import com.proovy.domain.run.coordinator.RunClaimStore; +import com.proovy.domain.run.coordinator.RunCoordinator; +import com.proovy.domain.run.coordinator.RunCoordinatorProperties; +import com.proovy.domain.run.entity.AiRunStatus; +import com.proovy.domain.run.entity.RunCancelCause; +import com.proovy.global.infra.brain.BrainCallException; +import com.proovy.global.infra.brain.BrainRunClient; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.context.SmartLifecycle; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * CANCELING run의 CANCELED 확정(task 1.84, docs/plan/05·11). 취소 API commit 직후 한 번, 그 뒤 {@code sweep-interval}마다 + * DB의 CANCELING run을 다시 확인한다 — 재시작·다른 인스턴스가 받은 취소도 이어받는다. 확정은 조건부 전환이라 여러 인스턴스가 + * 겹쳐도 하나만 이긴다. + *

한 번의 확인

+ * 현재 attempt·fence로 Brain {@code cancel}을 보낸다. 멱등이라 처음이면 취소 전달, 그 뒤로는 정리 상태 조회가 된다 + * (ack {@code state}). 판정: + * + * 어느 경우든 {@code cancel_requested_at + cleanup-timeout}이 지나면 확인 없이 CANCELED로 확정한다(안전 timeout). 시각은 DB + * {@code now()} 기준이고, 이미 지났으면 Brain을 부르지 않으며, 남은 시간이 client request-timeout보다 짧으면 호출을 그만큼만 기다린다. + * 확정 transaction 안에서 runtime cleanup을 예약하고({@link RunClaimStore#reserveRuntimeCleanup}) commit 뒤 이 인스턴스의 worker를 + * 멈춘다({@link RunCoordinator#release}). 그 전까지 run은 비terminal이라 전역 슬롯과 Note Active Run 자리를 계속 차지한다. + *

실패 행

+ * 확인이 예외로 끝난 run은 {@code failure-backoff} 동안 주기 확인에서 빼 오래된 실패 행이 batch를 막지 않게 한다(ERROR·metric). + * 확정이 불변조건 위반(열린 assistant placeholder 없음 등)으로 계속 실패하면 운영 수동 개입 대상이다. + *

스레드

+ * 내부 executor 하나({@code runCancel-*})이고 Spring {@code Executor}·{@code TaskScheduler} bean으로 노출하지 않는다(전역 기본 bean을 + * 바꾸지 않기 위해, 1.83·1.86과 같은 관례). run마다 동시에 확인 하나만 돈다. Brain 호출은 client의 request-timeout으로 끝나고 + * 그동안 DB transaction·row lock을 쥐지 않는다. lifecycle phase는 web server보다 앞이라 요청을 받기 전에 시작하고 요청이 멈춘 뒤에 + * 멈춘다. 그 밖의 구간(executor 없음)에 들어온 {@code wake}는 버려지고 주기 확인(다른 인스턴스·재시작 포함)이 줍는다 — + * 그래서 {@code enabled=false}로만 구성한 배포에서는 그 run이 CANCELING에 남는다(점검·테스트 전용 설정). + */ +@Slf4j +@Component +public class RunCancelFinalizer implements SmartLifecycle { + + private static final int THREADS = 2; + /** web server start/stop(Boot {@code DEFAULT_PHASE - 2048})·graceful shutdown(−1024)보다 앞. */ + private static final int PHASE = SmartLifecycle.DEFAULT_PHASE - 4096; + + enum BrainView { + TERMINAL, + ABSENT, + WINDING_DOWN, + UNKNOWN + } + + /** @param untilTimeout DB {@code now()} 기준 cleanup-timeout까지 남은 시간(지났으면 0) */ + private record CancelingRun(UUID runId, int attempt, String fenceToken, RunCancelCause cause, + Duration untilTimeout, boolean leaseExpired) { + } + + private enum Confirmation { + CONFIRMED, + ALREADY_DONE, + LEASE_LIVE + } + + private final RunCancelStore store; + private final RunCancelProperties properties; + private final JdbcTemplate jdbc; + private final ObjectProvider brain; + private final ObjectProvider coordinator; + private final ObjectProvider coordinatorProperties; + private final RunClaimStore claims; + private final MeterRegistry metrics; + private final Set inFlight = ConcurrentHashMap.newKeySet(); + /** 확인이 실패한 run → 주기 확인에 다시 넣을 시각(System.nanoTime). */ + private final Map backoff = new ConcurrentHashMap<>(); + private final AtomicBoolean sweepFailing = new AtomicBoolean(); + + private volatile ScheduledThreadPoolExecutor executor; + + public RunCancelFinalizer(RunCancelStore store, RunCancelProperties properties, JdbcTemplate jdbc, + TransactionTemplate tx, ObjectProvider brain, + ObjectProvider coordinator, + ObjectProvider coordinatorProperties, + ObjectProvider claims, ObjectProvider metrics) { + this.store = store; + this.properties = properties; + this.jdbc = jdbc; + this.brain = brain; + this.coordinator = coordinator; + this.coordinatorProperties = coordinatorProperties; + // coordinator가 꺼진 인스턴스에서도 같은 공개 hook(cleanup 예약 SQL)을 쓴다. + this.claims = claims.getIfAvailable(() -> new RunClaimStore(jdbc, tx)); + this.metrics = metrics.getIfAvailable(SimpleMeterRegistry::new); + } + + // ------------------------------------------------------------------ + // lifecycle + // ------------------------------------------------------------------ + + @Override + public synchronized void start() { + if (executor != null) { + return; + } + AtomicInteger sequence = new AtomicInteger(); + ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(THREADS, runnable -> { + Thread thread = new Thread(runnable, "runCancel-" + sequence.incrementAndGet()); + thread.setDaemon(true); + return thread; + }); + pool.setRemoveOnCancelPolicy(true); + if (properties.enabled()) { + long period = properties.sweepInterval().toMillis(); + pool.scheduleWithFixedDelay(this::sweep, period, period, TimeUnit.MILLISECONDS); + } + executor = pool; + log.info("run cancel finalizer started: sweep={}, sweepInterval={}, cleanupTimeout={}", properties.enabled(), + properties.sweepInterval(), properties.cleanupTimeout()); + } + + /** 확인 중인 Brain 호출은 request-timeout 안에 끝난다. 남은 CANCELING run은 다음 인스턴스의 sweep이 이어받는다. */ + @Override + public synchronized void stop() { + ScheduledThreadPoolExecutor pool = executor; + if (pool == null) { + return; + } + executor = null; + pool.shutdownNow(); + // shutdownNow가 버린 queue 작업은 finally를 돌지 않는다 — 다시 start하면 그 run이 영구히 거부되지 않게 비운다 + inFlight.clear(); + backoff.clear(); + try { + if (!pool.awaitTermination(5, TimeUnit.SECONDS)) { + log.warn("run cancel finalizer did not stop in time: inFlight={}", inFlight.size()); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + @Override + public boolean isRunning() { + return executor != null; + } + + @Override + public int getPhase() { + return PHASE; + } + + // ------------------------------------------------------------------ + // 진입점 + // ------------------------------------------------------------------ + + /** 취소 API가 CANCELING을 commit한 직후(또는 이미 CANCELING인 run에 다시 취소가 왔을 때) 곧바로 한 번 확인한다. */ + public void wake(UUID runId) { + submit(runId); + } + + /** DB의 CANCELING run을 오래된 취소 요청부터 확인한다(run별 동시 확인 하나, backoff 중인 실패 행 제외). */ + void sweep() { + try { + long now = System.nanoTime(); + backoff.values().removeIf(until -> until - now <= 0); + String excluded = backoff.keySet().stream().map(UUID::toString) + .collect(Collectors.joining(",", "{", "}")); + List canceling = jdbc.queryForList(""" + SELECT run_id FROM ai_runs WHERE status = 'CANCELING' AND run_id <> ALL (CAST(? AS uuid[])) + ORDER BY cancel_requested_at, run_id LIMIT ? + """, UUID.class, excluded, properties.batchSize()); + if (sweepFailing.compareAndSet(true, false)) { + log.info("cancel sweep recovered"); + } + canceling.forEach(this::submit); + } catch (RuntimeException e) { + // DB 장애 동안 주기마다 같은 WARN을 남기지 않는다(처음 한 번, 회복 시 INFO) + if (sweepFailing.compareAndSet(false, true)) { + log.warn("cancel sweep failed: cause={}", e.getClass().getSimpleName()); + } else { + log.debug("cancel sweep still failing: cause={}", e.getClass().getSimpleName()); + } + } + } + + private void submit(UUID runId) { + ScheduledThreadPoolExecutor pool = executor; + if (pool == null || !inFlight.add(runId)) { + return; + } + try { + pool.execute(() -> { + try { + check(runId); + } catch (RuntimeException e) { + failed(runId, e); + } finally { + inFlight.remove(runId); + } + }); + } catch (RejectedExecutionException e) { + inFlight.remove(runId); + } + } + + // ------------------------------------------------------------------ + // 확인·확정 + // ------------------------------------------------------------------ + + /** + * 실패한 확인: 예외 message에는 SQL·값이 들어갈 수 있어 class 이름만 남긴다. {@code failure-backoff} 동안 주기 확인에서 빼고 + * (취소 API의 {@code wake}는 그대로 확인한다), 잠금 뒤 불변조건 위반이면 ERROR다. + */ + private void failed(UUID runId, RuntimeException e) { + boolean transientFailure = RunCancelStore.isRetryable(e); + backoff.put(runId, System.nanoTime() + properties.failureBackoff().toNanos()); + metrics.counter("proovy.run.cancel.check_failures", "kind", transientFailure ? "transient" : "invariant") + .increment(); + if (transientFailure) { + log.warn("cancel check failed: runId={}, cause={}", runId, e.getClass().getSimpleName()); + } else { + log.error("cancel check failed, backing off: runId={}, cause={}", runId, e.getClass().getSimpleName()); + } + } + + /** 한 번 확인하고, 확정 조건이면 CANCELED로 바꾼다. 확정했으면 true. */ + boolean check(UUID runId) { + Optional found = jdbc.query(""" + SELECT attempt, fence_token, cancel_cause, + GREATEST(0, CEIL(EXTRACT(EPOCH FROM + (cancel_requested_at + make_interval(secs => ?) - now())) * 1000))::bigint, + lease_expires_at IS NULL OR lease_expires_at <= now() + FROM ai_runs WHERE run_id = ? AND status = 'CANCELING' + """, (rs, row) -> new CancelingRun(runId, rs.getInt(1), rs.getString(2), RunCancelStore.cause(rs.getString(3)), + Duration.ofMillis(rs.getLong(4)), rs.getBoolean(5)), + properties.cleanupTimeout().toMillis() / 1000.0, runId).stream().findFirst(); + if (found.isEmpty()) { + backoff.remove(runId); + return false; + } + CancelingRun run = found.get(); + long timeoutAt = System.nanoTime() + run.untilTimeout().toNanos(); + // timeout이 이미 지났으면 Brain 응답을 기다리지 않는다(hang인 Brain이 확정·다른 확인을 늦추지 않게) + BrainView view = run.untilTimeout().isZero() ? BrainView.UNKNOWN : askBrain(run, run.untilTimeout()); + String reason = switch (view) { + case TERMINAL -> "brain_terminal"; + case ABSENT -> run.leaseExpired() && !executingLocally(runId) ? "brain_absent" : null; + case WINDING_DOWN, UNKNOWN -> null; + }; + if (reason == null && timeoutAt - System.nanoTime() <= 0) { + log.warn("brain cleanup not confirmed before timeout, canceling anyway: runId={}, lastView={}", runId, + view); + reason = "cleanup_timeout"; + } + return reason != null && confirm(runId, reason); + } + + private BrainView askBrain(CancelingRun run, Duration wait) { + BrainRunClient client = brain.getIfAvailable(); + if (client == null) { + return BrainView.UNKNOWN; + } + try { + // client request-timeout 안에서, timeout까지 남은 시간보다 오래 기다리지 않는다 + RunCancelAck ack = client.cancel(RunCancelRequest.of(run.runId().toString(), run.attempt(), + run.fenceToken(), run.cause())).block(wait); + if (ack == null) { + return BrainView.UNKNOWN; + } + log.debug("brain cancel ack: runId={}, outcome={}, state={}", run.runId(), ack.outcome(), ack.state()); + return switch (ack.state()) { + case TERMINAL -> BrainView.TERMINAL; + case ABSENT -> BrainView.ABSENT; + case CANCELING, EXECUTING -> BrainView.WINDING_DOWN; + }; + } catch (BrainCallException e) { + log.info("brain cancel failed: runId={}, kind={}, httpStatus={}", run.runId(), e.kind(), e.httpStatus()); + return BrainView.UNKNOWN; + } catch (InternalRunContractViolation e) { + log.error("cancel request cannot be built from the run row: runId={}", run.runId()); + return BrainView.UNKNOWN; + } catch (IllegalStateException e) { + // block(wait) 초과: cleanup-timeout이 남은 시간 안에 왔다 + log.info("brain cancel did not answer before the cleanup timeout: runId={}", run.runId()); + return BrainView.UNKNOWN; + } + } + + /** + * CANCELING → CANCELED + runtime cleanup 예약을 한 transaction으로. {@code brain_absent}는 잠근 뒤 run lease 만료를 다시 본다 + * (Brain 호출 동안 늦은 heartbeat가 lease를 갱신했으면 그 worker가 아직 START를 보낼 수 있다). commit 여부를 모르는 재시도가 + * 이미 CANCELED를 보더라도 worker 정지는 한다. + */ + private boolean confirm(UUID runId, String reason) { + boolean requireExpiredLease = "brain_absent".equals(reason); + Duration cleanupDelay = cleanupDelay(); + Confirmation result = store.inTransactions(() -> { + Optional locked = store.lock(runId); + if (locked.isEmpty() || locked.get().status() != AiRunStatus.CANCELING) { + return Confirmation.ALREADY_DONE; + } + if (requireExpiredLease && !leaseExpired(runId)) { + return Confirmation.LEASE_LIVE; + } + Canceled canceled = store.confirmCanceled(locked.get()); + claims.reserveRuntimeCleanup(runId, cleanupDelay); + log.info("cancellation confirmed: runId={}, reason={}, lastSeq={}", runId, reason, canceled.lastSeq()); + return Confirmation.CONFIRMED; + }); + if (result == Confirmation.LEASE_LIVE) { + log.info("brain record absent but the run lease was renewed, waiting: runId={}", runId); + return false; + } + backoff.remove(runId); + RunCoordinator local = coordinator.getIfAvailable(); + if (local != null) { + local.release(runId); + } + if (result == Confirmation.CONFIRMED) { + metrics.counter("proovy.run.cancel.confirmed", "reason", reason).increment(); + return true; + } + return false; + } + + /** 잠근 run 행의 lease가 끝났는지(DB 시각). */ + private boolean leaseExpired(UUID runId) { + return Boolean.TRUE.equals(jdbc.queryForObject( + "SELECT lease_expires_at IS NULL OR lease_expires_at <= now() FROM ai_runs WHERE run_id = ?", + Boolean.class, runId)); + } + + private boolean executingLocally(UUID runId) { + RunCoordinator local = coordinator.getIfAvailable(); + return local != null && local.isExecuting(runId); + } + + private Duration cleanupDelay() { + RunCoordinatorProperties coordination = coordinatorProperties.getIfAvailable(); + return coordination == null ? Duration.ZERO : coordination.runtimeCleanupDelay(); + } +} diff --git a/src/main/java/com/proovy/domain/run/cancel/RunCancelProperties.java b/src/main/java/com/proovy/domain/run/cancel/RunCancelProperties.java new file mode 100644 index 0000000..94e0a71 --- /dev/null +++ b/src/main/java/com/proovy/domain/run/cancel/RunCancelProperties.java @@ -0,0 +1,41 @@ +package com.proovy.domain.run.cancel; + +import java.time.Duration; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.DefaultValue; + +/** + * {@code proovy.run.cancel.*} — RUNNING/RETRYING 취소의 CANCELED 확정(task 1.84). 기본값은 {@code @DefaultValue}다. + * + * @param enabled CANCELING run 주기 확정 작업을 돌린다. 끄면 취소 API 직후 한 번만 확인하고, 그 뒤 확정은 다른 인스턴스의 + * 주기 작업에 맡긴다(테스트·점검용. 운영은 켠다) + * @param sweepInterval CANCELING run을 다시 확인하는 주기. 매번 Brain {@code cancel}(멱등)을 다시 보내 그 ack state로 정리를 확인한다 + * @param cleanupTimeout {@code cancel_requested_at}부터 Brain 정리 확인을 기다리는 상한. 넘으면 확인 없이 CANCELED로 확정한다. + * 기본 2분 = Brain의 끊긴 wind-down 정리 상한(writer lease TTL 30s × 4, internal-run-v1 tail)보다 길게 — + * S14 배포 값과 함께 확정 + * @param batchSize 주기 한 번에 확인하는 CANCELING run 수(오래된 취소 요청부터) + * @param failureBackoff 확인이 예외로 끝난 run을 주기 확인에서 빼 두는 시간(실패 행이 batch를 막지 않게) + */ +@ConfigurationProperties(prefix = "proovy.run.cancel") +public record RunCancelProperties( + @DefaultValue("true") boolean enabled, + @DefaultValue("2s") Duration sweepInterval, + @DefaultValue("2m") Duration cleanupTimeout, + @DefaultValue("20") int batchSize, + @DefaultValue("30s") Duration failureBackoff) { + + public RunCancelProperties { + requirePositive(sweepInterval, "sweep-interval"); + requirePositive(cleanupTimeout, "cleanup-timeout"); + requirePositive(failureBackoff, "failure-backoff"); + if (batchSize < 1 || batchSize > 200) { + throw new IllegalArgumentException("proovy.run.cancel.batch-size must be within 1..200"); + } + } + + private static void requirePositive(Duration value, String name) { + if (value == null || value.isNegative() || value.isZero()) { + throw new IllegalArgumentException("proovy.run.cancel." + name + " must be positive"); + } + } +} diff --git a/src/main/java/com/proovy/domain/run/cancel/RunCancelService.java b/src/main/java/com/proovy/domain/run/cancel/RunCancelService.java new file mode 100644 index 0000000..40b45df --- /dev/null +++ b/src/main/java/com/proovy/domain/run/cancel/RunCancelService.java @@ -0,0 +1,77 @@ +package com.proovy.domain.run.cancel; + +import com.proovy.domain.run.cancel.RunCancelStore.LockedRun; +import com.proovy.domain.run.contract.CancelOutcome; +import com.proovy.domain.run.contract.PublicRunStatus; +import com.proovy.domain.run.contract.RunCancelResult; +import com.proovy.domain.run.entity.RunCancelCause; +import com.proovy.domain.run.stream.RunReadService; +import com.proovy.global.exception.BusinessException; +import com.proovy.global.response.ErrorCode; +import java.util.UUID; +import org.springframework.stereotype.Service; + +/** + * 사용자 명시적 취소 {@code POST /api/ai-runs/{runId}/cancel}(task 1.84). 판정 순서는 browser-run-v1 {@code cancel/cases.json} + * precedence다: 인증(filter) → body 있음 RUN4001(run 조회 전) → run 접근 RUN4041 → 상태. + *
    + *
  • PENDING: 즉시 CANCELED + 한 번 보상, 같은 transaction. Brain을 부르지 않는다(claim과는 하나만 이긴다).
    • + *
  • RUNNING·RETRYING: CANCELING + {@code run.canceling}. commit 뒤 {@link RunCancelFinalizer}가 Brain cancel·정리 확인 또는 + * 안전 timeout 뒤 CANCELED로 확정한다. 차감은 유지된다.
    • + *
  • CANCELING: {@code already_canceling}, terminal: {@code already_terminal}(멱등, 아무것도 바꾸지 않는다).
    • + *
+ * 브라우저 연결 단절은 취소가 아니다 — 취소는 이 경로뿐이다. 취소 뒤 자동 재실행·서버 queue·draft 전송은 없다. + */ +@Service +public class RunCancelService { + + private final RunReadService reads; + private final RunCancelStore store; + private final RunCancelFinalizer finalizer; + + public RunCancelService(RunReadService reads, RunCancelStore store, RunCancelFinalizer finalizer) { + this.reads = reads; + this.store = store; + this.finalizer = finalizer; + } + + private record Decision(PublicRunStatus status, CancelOutcome outcome, boolean canceling) { + } + + /** + * @param body 요청 body 원문. 비어 있지 않은 body는 어떤 내용이든 RUN4001이다(cancel은 body가 없다, `{}`도 거부) + */ + public RunCancelResult cancel(long userId, String runId, String body) { + if (body != null) { + throw new BusinessException(ErrorCode.RUN4001); + } + // 공개 경로의 접근 규칙(1.86): 소문자 UUID·소유자·Note ACTIVE. 없음·타인·삭제 중은 같은 RUN4041이다. + if (reads.accessibleRun(userId, runId).isEmpty()) { + throw new BusinessException(ErrorCode.RUN4041); + } + UUID id = UUID.fromString(runId); + Decision decision = store.inTransactions(() -> { + // 사전 조회와 잠금 사이에 Note 삭제가 시작됐으면 없는 run과 같다 + LockedRun run = store.lock(id).filter(locked -> locked.userId() == userId && locked.noteActive()) + .orElseThrow(() -> new BusinessException(ErrorCode.RUN4041)); + return switch (run.status()) { + case PENDING -> { + store.cancelPending(run, RunCancelCause.USER_REQUESTED); + yield new Decision(PublicRunStatus.CANCELED, CancelOutcome.ACCEPTED, false); + } + case RUNNING, RETRYING -> { + store.markCanceling(run, RunCancelCause.USER_REQUESTED); + yield new Decision(PublicRunStatus.CANCELING, CancelOutcome.ACCEPTED, true); + } + case CANCELING -> new Decision(PublicRunStatus.CANCELING, CancelOutcome.ALREADY_CANCELING, true); + case COMPLETED, FAILED, CANCELED -> + new Decision(PublicRunStatus.from(run.status()), CancelOutcome.ALREADY_TERMINAL, false); + }; + }); + if (decision.canceling()) { + // commit 뒤: Brain 호출은 transaction 밖이다. 실패해도 주기 확인이 이어받는다. + finalizer.wake(id); + } + return new RunCancelResult(id.toString(), decision.status(), decision.outcome()); + } +} diff --git a/src/main/java/com/proovy/domain/run/cancel/RunCancelStore.java b/src/main/java/com/proovy/domain/run/cancel/RunCancelStore.java new file mode 100644 index 0000000..56a9c56 --- /dev/null +++ b/src/main/java/com/proovy/domain/run/cancel/RunCancelStore.java @@ -0,0 +1,227 @@ +package com.proovy.domain.run.cancel; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.proovy.domain.run.contract.CancelCause; +import com.proovy.domain.run.contract.PublicRunStatus; +import com.proovy.domain.run.contract.RunCanceledPayload; +import com.proovy.domain.run.contract.RunCancelingPayload; +import com.proovy.domain.run.contract.ServerFrameType; +import com.proovy.domain.run.entity.AiRunStatus; +import com.proovy.domain.run.entity.RunCancelCause; +import com.proovy.domain.run.event.RunEventStore; +import com.proovy.domain.run.event.TerminalRunEvent; +import com.proovy.domain.usage.service.RequestUsageService; +import com.proovy.domain.usage.service.RunTerminalOutcome; +import com.proovy.domain.usage.service.UsageCompensationResult; +import java.sql.SQLException; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; +import java.util.function.Supplier; +import lombok.extern.slf4j.Slf4j; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * 취소 전환의 DB 경계(task 1.84). 모든 쓰기는 호출자가 {@link #inTransactions} 안에서 {@link #lock}으로 잡은 행에 대한 + * 조건부 UPDATE이고, 0 row면 불변조건 위반으로 rollback한다. + *

+ * 잠금 순서는 1.82 terminal 확정과 같다: {@code notes FOR SHARE → ai_runs FOR UPDATE → (usage) credit_balance FOR UPDATE}. + * 접수(1.79)·asset 삭제와 같은 Note 우선 순서라 서로 기다릴 뿐 교착하지 않는다. coordinator claim은 {@code SKIP LOCKED}로 + * 이 transaction이 잡은 PENDING run을 건너뛰고, 조건 {@code status = 'PENDING' AND attempt = 0}이라 취소와 claim은 하나만 이긴다. + * 

+ * PENDING             --cancel-----------------> CANCELED  (+ message CANCELED, run.canceled, 보상 PENDING_CANCELED)
+ * RUNNING|RETRYING    --cancel-----------------> CANCELING (+ run.canceling, grant 폐기. lease·슬롯은 유지)
+ * CANCELING           --Brain 정리 확인·timeout--> CANCELED  (+ message CANCELED, run.canceled, 차감 유지)
+ *
+ */ +@Slf4j +@Component +public class RunCancelStore { + + /** serialization_failure, deadlock_detected, lock_not_available (+ 연결 실패 08xxx) */ + private static final Set RETRYABLE_SQL_STATES = Set.of("40001", "40P01", "55P03"); + private static final int MAX_TRANSACTIONS = 3; + + private static final String LOCK_NOTE = """ + SELECT n.status FROM notes n WHERE n.note_id = (SELECT note_id FROM ai_runs WHERE run_id = ?) FOR SHARE + """; + private static final String LOCK_RUN = """ + SELECT user_id, request_id, note_id, assistant_message_id, status, attempt, fence_token, cancel_cause, + cancel_requested_from, last_event_seq + FROM ai_runs WHERE run_id = ? FOR UPDATE + """; + private static final String CANCELED = """ + status = 'CANCELED', completed_at = ?, lease_owner = NULL, lease_expires_at = NULL, grant_digest = NULL, + grant_expires_at = NULL, last_event_seq = last_event_seq + 1 + """; + + private final JdbcTemplate jdbc; + private final TransactionTemplate tx; + private final ObjectMapper objectMapper; + private final RunEventStore events; + private final RequestUsageService usage; + + public RunCancelStore(JdbcTemplate jdbc, TransactionTemplate tx, ObjectMapper objectMapper, RunEventStore events, + RequestUsageService usage) { + this.jdbc = jdbc; + this.tx = tx; + this.objectMapper = objectMapper; + this.events = events; + this.usage = usage; + } + + /** + * 잠근 run. {@code noteActive}는 Note가 ACTIVE인지(DELETING이면 공개 경로는 없는 run과 같다). + * + * @param cancelRequestedFrom CANCELING·CANCELED일 때만 + */ + record LockedRun(UUID runId, long userId, String requestId, long noteId, boolean noteActive, + UUID assistantMessageId, AiRunStatus status, int attempt, String fenceToken, + RunCancelCause cancelCause, AiRunStatus cancelRequestedFrom, long lastEventSeq) { + } + + /** 결과 CANCELED 전환. */ + record Canceled(long lastSeq, UsageCompensationResult compensation) { + } + + /** Note → run 순서로 잠근다. run이나 Note가 없으면(삭제 cascade) empty. transaction 안에서만 부른다. */ + Optional lock(UUID runId) { + List note = jdbc.queryForList(LOCK_NOTE, String.class, runId); + if (note.isEmpty()) { + return Optional.empty(); + } + boolean noteActive = "ACTIVE".equals(note.get(0)); + return jdbc.query(LOCK_RUN, (rs, row) -> new LockedRun(runId, rs.getLong(1), rs.getString(2), rs.getLong(3), + noteActive, rs.getObject(4, UUID.class), AiRunStatus.valueOf(rs.getString(5)), rs.getInt(6), + rs.getString(7), cause(rs.getString(8)), status(rs.getString(9)), rs.getLong(10)), runId) + .stream().findFirst(); + } + + /** PENDING(attempt 0) → CANCELED + 보상. claim이 먼저 이겼으면 잠금 뒤 RUNNING이라 여기 오지 않는다. */ + Canceled cancelPending(LockedRun run, RunCancelCause cause) { + Instant now = now(); + long lastSeq = transition(run, "cancel_cause = ?, cancel_requested_from = 'PENDING', cancel_requested_at = ?, " + + CANCELED, "status = 'PENDING' AND attempt = 0", + cause.wireValue(), time(now), time(now)); + return writeCanceled(run, cause, AiRunStatus.PENDING, lastSeq, now); + } + + /** + * RUNNING|RETRYING(현재 attempt·fence) → CANCELING + {@code run.canceling}. 이후 이 fence의 Activity는 열린 Activity의 + * {@code activity.canceled}만, terminal은 하나도 받지 않는다(1.80·1.82). lease·heartbeat는 그대로라 worker가 stream을 마저 + * 소비하고, 비terminal이라 전역 슬롯·Note Active Run 자리를 계속 차지한다. 이전 grant는 폐기한다(plan 12). + * {@code cancel_requested_at}은 DB {@code now()}다 — cleanup-timeout을 다른 인스턴스가 DB 시각으로 비교한다. + */ + void markCanceling(LockedRun run, RunCancelCause cause) { + Instant now = now(); + int updated = jdbc.update(""" + UPDATE ai_runs SET status = 'CANCELING', cancel_cause = ?, cancel_requested_from = status, + cancel_requested_at = now(), grant_digest = NULL, grant_expires_at = NULL + WHERE run_id = ? AND status IN ('RUNNING', 'RETRYING') AND attempt = ? AND fence_token = ? + """, cause.wireValue(), run.runId(), run.attempt(), run.fenceToken()); + requireOne(updated, "canceling transition lost its guard", run); + events.appendServerEvent(run.runId(), ServerFrameType.RUN_CANCELING, + new RunCancelingPayload(run.assistantMessageId().toString(), PublicRunStatus.CANCELING), now); + } + + /** CANCELING → CANCELED. 사용량은 1.77 정책표대로(사용자 취소는 차감 유지). */ + Canceled confirmCanceled(LockedRun run) { + Instant now = now(); + long lastSeq = transition(run, CANCELED, "status = 'CANCELING'", time(now)); + return writeCanceled(run, run.cancelCause(), run.cancelRequestedFrom(), lastSeq, now); + } + + private Canceled writeCanceled(LockedRun run, RunCancelCause cause, AiRunStatus from, long lastSeq, + Instant now) { + int message = jdbc.update(""" + UPDATE chat_messages SET status = 'CANCELED', completed_at = ? + WHERE message_id = ? AND note_id = ? AND role = 'ASSISTANT' AND status IN ('PENDING', 'STREAMING') + """, time(now), run.assistantMessageId(), run.noteId()); + requireOne(message, "assistant message is not an open placeholder", run); + + JsonNode payload = objectMapper.valueToTree( + RunCanceledPayload.of(run.assistantMessageId().toString(), CancelCause.from(cause))); + events.appendTerminalEvents(run.runId(), lastSeq, List.of(TerminalRunEvent.runCanceled(payload)), now); + + UsageCompensationResult compensation = usage.settle(run.userId(), run.requestId(), + RunTerminalOutcome.canceled(from, cause)); + log.info("run canceled: runId={}, from={}, cause={}, compensation={}", run.runId(), from, cause.wireValue(), + compensation.outcome()); + return new Canceled(lastSeq, compensation); + } + + /** 조건부 terminal 전환. 잠금 뒤 판정과 같은 조건이라 0 row면 불변조건 위반이다(rollback). */ + private long transition(LockedRun run, String set, String guard, Object... setArgs) { + Object[] args = Arrays.copyOf(setArgs, setArgs.length + 1); + args[setArgs.length] = run.runId(); + List returned = jdbc.queryForList("UPDATE ai_runs SET " + set + " WHERE run_id = ? AND " + guard + + " RETURNING last_event_seq", Long.class, args); + if (returned.size() != 1) { + throw new IllegalStateException("cancel transition lost its guard for run " + run.runId()); + } + return returned.get(0); + } + + /** + * 자기 transaction에서 실행한다. 직렬화·교착·연결 실패만 같은 입력으로 새 transaction을 연다(판정을 잠금 뒤 다시 하므로 + * commit 여부를 모르는 실패 뒤에도 멱등이다 — 이미 바뀐 상태를 보고 ALREADY_*로 답한다). 호출자 transaction이 열려 있으면 + * 거기에 참여하고 재시도하지 않는다 — rollback-only가 된 바깥 transaction 안에서 다시 시도할 수 없다(RunTerminalService와 같은 규칙). + */ + T inTransactions(Supplier work) { + boolean outer = TransactionSynchronizationManager.isActualTransactionActive(); + for (int attempt = 1; ; attempt++) { + try { + return tx.execute(status -> work.get()); + } catch (RuntimeException e) { + if (outer || attempt >= MAX_TRANSACTIONS || !isRetryable(e)) { + throw e; + } + log.info("cancel transaction {} failed, retrying: cause={}", attempt, e.getClass().getSimpleName()); + } + } + } + + static boolean isRetryable(Throwable failure) { + for (Throwable cause = failure; cause != null; cause = cause.getCause()) { + if (cause instanceof SQLException sql && sql.getSQLState() != null) { + String state = sql.getSQLState(); + return RETRYABLE_SQL_STATES.contains(state) || state.startsWith("08"); + } + } + return false; + } + + private static void requireOne(int rows, String problem, LockedRun run) { + if (rows != 1) { + throw new IllegalStateException(problem + " (run " + run.runId() + ")"); + } + } + + private static final RunCancelCause.Converter CAUSES = new RunCancelCause.Converter(); + + static RunCancelCause cause(String wire) { + return CAUSES.convertToEntityAttribute(wire); + } + + private static AiRunStatus status(String value) { + return value == null ? null : AiRunStatus.valueOf(value); + } + + static Instant now() { + return Instant.now().truncatedTo(ChronoUnit.MICROS); + } + + private static OffsetDateTime time(Instant instant) { + return OffsetDateTime.ofInstant(instant, ZoneOffset.UTC); + } +} diff --git a/src/main/java/com/proovy/domain/run/controller/BrowserRunController.java b/src/main/java/com/proovy/domain/run/controller/BrowserRunController.java index 869c9d7..0bdbb0a 100644 --- a/src/main/java/com/proovy/domain/run/controller/BrowserRunController.java +++ b/src/main/java/com/proovy/domain/run/controller/BrowserRunController.java @@ -2,7 +2,9 @@ import com.fasterxml.jackson.databind.JsonNode; import com.proovy.domain.run.admission.RunAdmissionService; +import com.proovy.domain.run.cancel.RunCancelService; import com.proovy.domain.run.contract.ConversationPage; +import com.proovy.domain.run.contract.RunCancelResult; import com.proovy.domain.run.contract.RunStatus; import com.proovy.domain.run.stream.RunReadService; import com.proovy.domain.run.stream.RunStreamService; @@ -26,7 +28,7 @@ *

* 스트림 시작 전 거부는 Accept가 {@code text/event-stream}이어도 JSON {@code ApiResponse}다 * ({@link BrowserRunExceptionHandler}, 접수 거부는 {@code RunAdmissionExceptionHandler}). 시작 뒤에는 terminal event로 끝난다. - * 취소({@code POST /api/ai-runs/{runId}/cancel})는 1.84가 이 controller에 추가한다. + * 취소({@code POST /api/ai-runs/{runId}/cancel})는 동기 JSON이다(1.84). 브라우저 연결 단절은 취소가 아니다. */ @RestController @Tag(name = "Browser Run", description = "대화 요청 접수·SSE replay·run 상태·대화 조회 (browser-run-v1)") @@ -38,11 +40,14 @@ public class BrowserRunController { private final RunAdmissionService admission; private final RunStreamService streams; private final RunReadService reads; + private final RunCancelService cancels; - public BrowserRunController(RunAdmissionService admission, RunStreamService streams, RunReadService reads) { + public BrowserRunController(RunAdmissionService admission, RunStreamService streams, RunReadService reads, + RunCancelService cancels) { this.admission = admission; this.streams = streams; this.reads = reads; + this.cancels = cancels; } @PostMapping("/api/conversations") @@ -73,6 +78,17 @@ public ApiResponse status(@Parameter(hidden = true) @AuthenticationPr return ApiResponse.success(reads.status(principal.getUserId(), runId)); } + @PostMapping("/api/ai-runs/{runId}/cancel") + @Operation(summary = "run 취소", + description = "body 없이 보낸다(`{}`도 RUN4001). PENDING은 즉시 CANCELED(보상), RUNNING은 CANCELING 뒤 Brain 정리 확인 " + + "또는 안전 timeout에 CANCELED(차감 유지). 이미 CANCELING·terminal이면 같은 상태를 멱등으로 돌려준다.") + public ApiResponse cancel(@Parameter(hidden = true) @AuthenticationPrincipal UserPrincipal principal, + @PathVariable String runId, + @RequestBody(required = false) String body) { + // body는 원문으로 받는다: 내용·Content-Type과 무관하게 body가 있으면 run 조회 전에 RUN4001이다. + return ApiResponse.success(cancels.cancel(principal.getUserId(), runId, body)); + } + @GetMapping("/api/conversations/{noteId}/messages") @Operation(summary = "대화 조회", description = "before(messageId)보다 오래된 메시지 중 최신 limit(1..100, 기본 50)개를 createdAt 오름차순으로. " diff --git a/src/test/java/com/proovy/domain/run/cancel/CancelFakeBrain.java b/src/test/java/com/proovy/domain/run/cancel/CancelFakeBrain.java new file mode 100644 index 0000000..95fbe28 --- /dev/null +++ b/src/test/java/com/proovy/domain/run/cancel/CancelFakeBrain.java @@ -0,0 +1,320 @@ +package com.proovy.domain.run.cancel; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.proovy.domain.run.brain.InternalRunRequest; +import com.proovy.domain.run.brain.InternalRunWire; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +/** + * 취소 테스트용 Brain 내부 API(internal-run-v1) HTTP 가짜. 실제 {@code BrainRunClient}가 loopback 임시 port로 부른다. + *

    + *
  • stream: {@code activity.started}(activity-1) 뒤 cancel이 올 때까지 keep-alive로 붙잡는다. cancel을 받으면 열린 Activity의 + * {@code activity.canceled}를 보내고 terminal 없이 끝낸다(Brain wind-down). {@link #complete}면 clarification terminal.
    • + *
  • cancel: run별 {@link CancelMode}. 기본 {@code CONFIRM}은 첫 호출에 accepted/canceling, 그 뒤 already_terminal/terminal.
    • + *
  • status: record 상태대로 absent·executing·canceling·terminal(run.canceled 또는 clarification).
    • + *
+ */ +final class CancelFakeBrain implements AutoCloseable { + + enum CancelMode { + /** 첫 호출 accepted(canceling), 정리가 끝난 뒤 already_terminal(terminal) */ + CONFIRM, + /** 정리를 끝내지 않는다 — 언제나 canceling */ + NEVER_CONFIRM, + /** record 없음 — unknown_run(absent) */ + ABSENT, + /** 503 */ + UNAVAILABLE, + /** 응답하지 않는다(close까지) — hang인 Brain */ + HANG + } + + enum Record { + ABSENT, + EXECUTING, + CANCELING, + CANCELED, + COMPLETED + } + + private static final ObjectMapper JSON = new ObjectMapper(); + private static final Path MIRROR = mirror(); + + private final HttpServer server; + private final ExecutorService executor = Executors.newCachedThreadPool(); + private final Map modes = new ConcurrentHashMap<>(); + private final Map records = new ConcurrentHashMap<>(); + private final Map lastRequest = new ConcurrentHashMap<>(); + private final Map windDown = new ConcurrentHashMap<>(); + private final Map completeOnRelease = new ConcurrentHashMap<>(); + private final Map open = new ConcurrentHashMap<>(); + private final List streamCalls = new CopyOnWriteArrayList<>(); + private final List cancelCalls = new CopyOnWriteArrayList<>(); + private final Map beforeCancelAck = new ConcurrentHashMap<>(); + private final CountDownLatch closed = new CountDownLatch(1); + + CancelFakeBrain() { + try { + server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + server.setExecutor(executor); + server.createContext("/internal/v1/runs/", this::handle); + server.start(); + } + + String baseUrl() { + return "http://127.0.0.1:" + server.getAddress().getPort(); + } + + void mode(String runId, CancelMode mode) { + modes.put(runId, mode); + } + + /** stream 없이 Brain record가 실행 중인 것처럼 둔다(coordinator 없이 RUNNING run을 만든 테스트). */ + void executing(String runId, int attempt, String fenceToken) { + lastRequest.put(runId, JSON.createObjectNode().put("run_id", runId).put("attempt", attempt) + .put("fence_token", fenceToken).put("_fingerprint", "f".repeat(64))); + records.put(runId, Record.EXECUTING); + } + + /** 붙잡은 stream을 clarification terminal로 끝낸다(취소 전 완료 경합용). */ + void complete(String runId) { + completeOnRelease.put(runId, true); + latch(runId).countDown(); + } + + /** 다음 cancel 호출 하나가 응답하기 직전에 한 번 실행한다(Brain 호출 중의 Server 쪽 변화 재현). */ + void beforeNextCancelAck(String runId, Runnable action) { + beforeCancelAck.put(runId, action); + } + + long streamCalls(String runId) { + return streamCalls.stream().filter(runId::equals).count(); + } + + List cancelCalls(String runId) { + return cancelCalls.stream().filter(call -> runId.equals(call.get("run_id").asText())).toList(); + } + + Record record(String runId) { + return records.getOrDefault(runId, Record.ABSENT); + } + + @Override + public void close() { + closed.countDown(); + windDown.values().forEach(CountDownLatch::countDown); + server.stop(0); + executor.shutdownNow(); + } + + // ------------------------------------------------------------------ + + private CountDownLatch latch(String runId) { + return windDown.computeIfAbsent(runId, key -> new CountDownLatch(1)); + } + + private void handle(HttpExchange exchange) throws IOException { + try { + String[] path = exchange.getRequestURI().getPath().split("/"); + String runId = path[4]; + if ("GET".equals(exchange.getRequestMethod())) { + json(exchange, 200, status(runId)); + return; + } + byte[] body = exchange.getRequestBody().readAllBytes(); + if (path.length == 6 && "stream".equals(path[5])) { + stream(exchange, runId, body); + } else if (path.length == 6 && "cancel".equals(path[5])) { + cancel(exchange, runId, JSON.readTree(body)); + } else { + json(exchange, 404, JSON.createObjectNode().put("error", "not_found")); + } + } finally { + exchange.close(); + } + } + + private void stream(HttpExchange exchange, String runId, byte[] body) throws IOException { + streamCalls.add(runId); + JsonNode wire = JSON.readTree(body); + InternalRunRequest request = InternalRunWire.read(body, InternalRunRequest.class); + ((ObjectNode) wire).put("_fingerprint", request.fingerprint().sha256()); + lastRequest.put(runId, wire); + records.putIfAbsent(runId, Record.EXECUTING); + exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, 0); + OutputStream out = exchange.getResponseBody(); + open.put(runId, true); + try { + if (records.get(runId) == Record.EXECUTING) { + frame(out, bind(load("activity-v1/valid/input-analysis-started.json"), wire, "activity-1:started")); + } + CountDownLatch latch = latch(runId); + while (!latch.await(300, TimeUnit.MILLISECONDS)) { + out.write(": keep-alive\n\n".getBytes(StandardCharsets.UTF_8)); + out.flush(); + } + if (Boolean.TRUE.equals(completeOnRelease.get(runId)) && records.get(runId) == Record.EXECUTING) { + records.put(runId, Record.COMPLETED); + frame(out, clarification(wire)); + } else if (records.get(runId) == Record.CANCELING) { + ObjectNode canceled = bind(load("activity-v1/valid/input-analysis-started.json"), wire, + "activity-1:canceled"); + canceled.put("type", "activity.canceled"); + canceled.set("payload", JSON.createObjectNode().put("kind", "input_analysis") + .put("activity_id", "activity-1")); + frame(out, canceled); + // 정리를 끝내고 run.canceled를 저장한 뒤 terminal 없이 stream을 닫는다 + if (modes.getOrDefault(runId, CancelMode.CONFIRM) == CancelMode.CONFIRM) { + records.put(runId, Record.CANCELED); + } + } + out.flush(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (IOException disconnected) { + // client가 구독을 끊었다 + } finally { + open.remove(runId); + } + } + + private void cancel(HttpExchange exchange, String runId, JsonNode request) throws IOException { + cancelCalls.add(request); + CancelMode mode = modes.getOrDefault(runId, CancelMode.CONFIRM); + if (mode == CancelMode.HANG) { + try { + closed.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return; + } + Runnable before = beforeCancelAck.remove(runId); + if (before != null) { + before.run(); + } + ObjectNode ack = JSON.createObjectNode().put("contract_version", "internal-run-v1").put("run_id", runId); + switch (mode) { + case UNAVAILABLE -> { + json(exchange, 503, JSON.createObjectNode().put("error", "unavailable")); + return; + } + case ABSENT -> ack.put("outcome", "unknown_run").put("state", "absent"); + default -> { + Record record = records.getOrDefault(runId, Record.ABSENT); + switch (record) { + case ABSENT -> ack.put("outcome", "unknown_run").put("state", "absent"); + case EXECUTING -> { + records.put(runId, Record.CANCELING); + latch(runId).countDown(); + ack.put("outcome", "accepted").put("state", "canceling"); + } + case CANCELING -> { + if (mode == CancelMode.CONFIRM && !open.containsKey(runId)) { + // stream 없는 실행은 다음 확인 때 정리가 끝나 있다 + records.put(runId, Record.CANCELED); + ack.put("outcome", "already_terminal").put("state", "terminal"); + } else { + ack.put("outcome", "already_canceling").put("state", "canceling"); + } + } + case CANCELED, COMPLETED -> ack.put("outcome", "already_terminal").put("state", "terminal"); + } + } + } + json(exchange, 200, ack); + } + + private ObjectNode status(String runId) { + ObjectNode response = JSON.createObjectNode().put("contract_version", "internal-run-v1").put("run_id", runId); + JsonNode last = lastRequest.get(runId); + Record record = records.getOrDefault(runId, Record.ABSENT); + if (last == null || record == Record.ABSENT) { + return response.put("state", "absent"); + } + response.put("attempt", last.get("attempt").asInt()).put("fence_token", last.get("fence_token").asText()) + .put("request_fingerprint", last.get("_fingerprint").asText()); + switch (record) { + case EXECUTING -> response.put("state", "executing"); + case CANCELING -> response.put("state", "canceling"); + case CANCELED -> { + ObjectNode terminal = bind(JSON.createObjectNode(), last, "terminal"); + terminal.put("contract_version", "brain-source-v1").put("type", "run.canceled"); + terminal.set("payload", JSON.createObjectNode().put("cause", "user_requested")); + response.put("state", "terminal").set("terminal", terminal); + } + case COMPLETED -> response.put("state", "terminal").set("terminal", clarification(last)); + default -> throw new IllegalStateException(); + } + return response; + } + + private static ObjectNode clarification(JsonNode request) { + return bind(load("terminal-v2/pairs/clarification.json").get("source"), request, "terminal"); + } + + private static ObjectNode bind(JsonNode source, JsonNode request, String suffix) { + ObjectNode bound = source.deepCopy(); + String runId = request.get("run_id").asText(); + bound.put("run_id", runId).put("attempt", request.get("attempt").asInt()) + .put("fence_token", request.get("fence_token").asText()) + .put("source_event_id", runId + ":" + suffix) + .put("occurred_at", Instant.now().truncatedTo(ChronoUnit.MILLIS).toString()); + return bound; + } + + private static void frame(OutputStream out, JsonNode event) throws IOException { + String text = "event: " + event.get("type").asText() + "\ndata: " + JSON.writeValueAsString(event) + "\n\n"; + out.write(text.getBytes(StandardCharsets.UTF_8)); + out.flush(); + } + + private static void json(HttpExchange exchange, int status, JsonNode body) throws IOException { + byte[] bytes = JSON.writeValueAsBytes(body); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(status, bytes.length); + exchange.getResponseBody().write(bytes); + } + + private static JsonNode load(String relative) { + try (InputStream in = Files.newInputStream(MIRROR.resolve(relative))) { + return JSON.readTree(in); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static Path mirror() { + try { + return Path.of(CancelFakeBrain.class.getClassLoader().getResource("contracts/brain").toURI()); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } +} diff --git a/src/test/java/com/proovy/domain/run/cancel/RunCancelCoordinatorPostgresTest.java b/src/test/java/com/proovy/domain/run/cancel/RunCancelCoordinatorPostgresTest.java new file mode 100644 index 0000000..7518d56 --- /dev/null +++ b/src/test/java/com/proovy/domain/run/cancel/RunCancelCoordinatorPostgresTest.java @@ -0,0 +1,196 @@ +package com.proovy.domain.run.cancel; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.proovy.domain.run.admission.RunAdmission; +import com.proovy.domain.run.admission.RunAdmissionService; +import com.proovy.domain.run.cancel.CancelFakeBrain.CancelMode; +import com.proovy.domain.run.coordinator.RunClaimStore; +import com.proovy.domain.run.coordinator.RunCoordinator; +import com.proovy.support.db.PostgresTestDatabase; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.UUID; +import java.util.function.BooleanSupplier; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +/** + * 취소(1.84)와 실제 coordinator(1.83) worker의 end-to-end: 실제 {@code BrainRunClient} + loopback 가짜 Brain + 실제 Postgres. + * 전역 슬롯 1이라 CANCELING run이 Brain 정리 확인 전까지 다음 run의 dispatch를 막는지(슬롯 유지) 본다. + */ +@Tag("postgres") +// 끝난 class의 context를 닫아 finalizer sweep이 닫힌 DB를 계속 두드리지 않게 한다 +@DirtiesContext +@ActiveProfiles("test") +@SpringBootTest(properties = { + "gcs.project-id=proovy-test", + "gcs.endpoint=http://127.0.0.1:1", + "jwt.secret=test-jwt-secret-test-jwt-secret-0123456789", + "oauth.kakao.client-id=test", "oauth.kakao.client-secret=test", + "oauth.naver.client-id=test", "oauth.naver.client-secret=test", + "openrouter.api.key=test", + "spring.data.redis.host=127.0.0.1", "spring.data.redis.port=1", + "spring.task.scheduling.pool.size=1", + "proovy.subscription.plan-transition-fixed-delay-ms=3600000", + "proovy.subscription.auto-renew-fixed-delay-ms=3600000", + "proovy.brain.auth.mode=shared-token", + "proovy.brain.auth.shared-token=t184-shared-token", + "proovy.brain.stream-idle-timeout=3s", + "proovy.run.coordinator.enabled=true", + "proovy.run.coordinator.instance-id=t184-coordinator", + "proovy.run.coordinator.max-concurrent-runs=1", + "proovy.run.coordinator.poll-interval=100ms", + "proovy.run.coordinator.heartbeat-interval=300ms", + "proovy.run.coordinator.lease-ttl=2s", + "proovy.run.coordinator.deadline-sweep-interval=200ms", + "proovy.run.coordinator.min-dispatch-remaining=0s", + "proovy.run.coordinator.redelivery.initial-backoff=50ms", + "proovy.run.coordinator.redelivery.max-backoff=200ms", + "proovy.run.coordinator.shutdown-timeout=3s", + "proovy.run.cancel.sweep-interval=200ms", + "proovy.run.cancel.cleanup-timeout=60s" +}) +class RunCancelCoordinatorPostgresTest { + + private static final Duration WAIT = Duration.ofSeconds(20); + + private static PostgresTestDatabase db; + private static CancelFakeBrain brain; + + @BeforeAll + static void setUp() { + db = PostgresTestDatabase.createMigrated(); + db.runScript("db/seed/product-seed.sql"); + brain = new CancelFakeBrain(); + } + + @DynamicPropertySource + static void properties(DynamicPropertyRegistry registry) { + registry.add("spring.datasource.url", () -> db.jdbcUrl()); + registry.add("spring.datasource.username", PostgresTestDatabase::adminUser); + registry.add("spring.datasource.password", PostgresTestDatabase::adminPassword); + registry.add("proovy.brain.base-url", () -> brain.baseUrl()); + } + + @AfterAll + static void tearDown() { + if (brain != null) { + brain.close(); + } + if (db != null) { + db.close(); + } + } + + @Autowired RunAdmissionService admission; + @Autowired RunCancelService cancels; + @Autowired RunCoordinator coordinator; + @Autowired RunClaimStore claims; + @Autowired JdbcTemplate jdbc; + @Autowired ObjectMapper mapper; + + @Test + @DisplayName("실행 중 취소: worker는 wind-down Activity를 저장하고 OBSOLETE로 끝나지만 슬롯은 CANCELED 확정까지 유지, 확정 뒤 다음 run dispatch") + void cancelingRunKeepsTheGlobalSlotUntilBrainCleanupIsConfirmed() { + long first = newUser(); + RunAdmission running = admit(first); + String runId = running.runId().toString(); + brain.mode(runId, CancelMode.NEVER_CONFIRM); + await(() -> eventTypes(running.runId()).contains("activity.started")); + assertThat(coordinator.isExecuting(running.runId())).isTrue(); + + assertThat(cancels.cancel(first, runId, null).outcome().wireValue()).isEqualTo("accepted"); + // Brain cancel → stream이 열린 Activity의 canceled 표식을 보내고 terminal 없이 닫힌다 → worker는 status(canceling)로 OBSOLETE + await(() -> eventTypes(running.runId()).contains("activity.canceled")); + await(() -> !coordinator.isExecuting(running.runId())); + assertThat(status(running.runId())).isEqualTo("CANCELING"); + + // 정리 확인 전: 전역 슬롯(1)을 계속 차지해 다른 사용자의 run이 dispatch되지 않는다 + RunAdmission waiting = admit(newUser()); + sleep(1500); + assertThat(claims.countActiveExecutions()).isEqualTo(1); + assertThat(status(waiting.runId())).isEqualTo("PENDING"); + assertThat(brain.streamCalls(waiting.runId().toString())).isZero(); + + // Brain 정리 완료 → CANCELED 확정 → 슬롯 반환 → 대기 run dispatch + brain.mode(runId, CancelMode.CONFIRM); + await(() -> "CANCELED".equals(status(running.runId()))); + assertThat(eventTypes(running.runId())).containsExactly("run.created", "activity.started", "run.canceling", + "activity.canceled", "run.canceled"); + assertThat(jdbc.queryForObject("SELECT runtime_cleanup_at IS NOT NULL FROM ai_runs WHERE run_id = ?", + Boolean.class, running.runId())).isTrue(); + await(() -> brain.streamCalls(waiting.runId().toString()) == 1); + brain.complete(waiting.runId().toString()); + await(() -> "COMPLETED".equals(status(waiting.runId()))); + + // 취소된 run은 다시 전달되지 않았다(자동 재실행 없음) + assertThat(brain.streamCalls(runId)).isEqualTo(1); + assertThat(jdbc.queryForObject("SELECT attempt FROM ai_runs WHERE run_id = ?", Integer.class, + running.runId())).isEqualTo(1); + } + + // ------------------------------------------------------------------ + + private RunAdmission admit(long user) { + ObjectNode body = mapper.createObjectNode(); + body.put("requestId", UUID.randomUUID().toString()); + body.put("text", "x^2-5x+6=0을 풀어줘 " + UUID.randomUUID()); + return admission.admit(user, body); + } + + private long newUser() { + String key = UUID.randomUUID().toString(); + long user = jdbc.queryForObject(""" + INSERT INTO users (nickname, provider, provider_user_id, email, created_at, updated_at) + VALUES ('t184', 'KAKAO', ?, ?, now(), now()) RETURNING user_id + """, Long.class, key, key + "@example.test"); + jdbc.update(""" + INSERT INTO credit_balance (user_id, daily_free_credit, daily_free_limit, daily_expires_at, free_credit, + paid_credit, created_at, updated_at) + VALUES (?, 100, 100, now() + interval '1 day', 0, 0, now(), now()) + """, user); + return user; + } + + private String status(UUID runId) { + return jdbc.queryForObject("SELECT status FROM ai_runs WHERE run_id = ?", String.class, runId); + } + + private List eventTypes(UUID runId) { + return jdbc.queryForList("SELECT event_type FROM ai_run_events WHERE run_id = ? ORDER BY seq", String.class, + runId); + } + + private static void sleep(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + + private static void await(BooleanSupplier condition) { + Instant until = Instant.now().plus(WAIT); + while (!condition.getAsBoolean()) { + if (Instant.now().isAfter(until)) { + throw new AssertionError("condition not met within " + WAIT); + } + sleep(50); + } + } +} diff --git a/src/test/java/com/proovy/domain/run/cancel/RunCancelPostgresTest.java b/src/test/java/com/proovy/domain/run/cancel/RunCancelPostgresTest.java new file mode 100644 index 0000000..1042543 --- /dev/null +++ b/src/test/java/com/proovy/domain/run/cancel/RunCancelPostgresTest.java @@ -0,0 +1,929 @@ +package com.proovy.domain.run.cancel; + +import static com.proovy.domain.run.contract.BrowserRunContractTestSupport.load; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.proovy.domain.auth.service.AccessTokenBlacklistService; +import com.proovy.domain.auth.service.JwtTokenProvider; +import com.proovy.domain.run.admission.RunAdmission; +import com.proovy.domain.run.admission.RunAdmissionService; +import com.proovy.domain.run.brain.InternalRunWire; +import com.proovy.domain.run.brain.SourceEvent; +import com.proovy.domain.run.cancel.CancelFakeBrain.CancelMode; +import com.proovy.domain.run.contract.RunCancelResult; +import com.proovy.domain.run.coordinator.ClaimedRun; +import com.proovy.domain.run.coordinator.RunClaimStore; +import com.proovy.domain.run.entity.RunFailureCode; +import com.proovy.domain.run.event.ActivityAppendResult; +import com.proovy.domain.run.event.RunEventStore; +import com.proovy.domain.run.terminal.RunTerminalService; +import com.proovy.domain.run.terminal.TerminalSettlement; +import com.proovy.domain.run.terminal.TerminalSettlement.Outcome; +import com.proovy.domain.usage.service.RequestUsageService; +import com.proovy.global.exception.BusinessException; +import com.proovy.global.response.ErrorCode; +import com.proovy.support.db.PostgresTestDatabase; +import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.function.BooleanSupplier; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.server.LocalServerPort; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * 사용자 취소(1.84)를 실제 Postgres + 실제 HTTP(임의 port) + 실제 {@code BrainRunClient}·loopback 가짜 Brain으로 검사한다. + * coordinator는 끈다 — PENDING은 그대로 남고, RUNNING은 claim SQL로 만든다(worker와의 end-to-end는 + * {@link RunCancelCoordinatorPostgresTest}). 판정표(cancel/cases.json)는 production endpoint로 돌린다. + *

+ * cleanup-timeout 5s, 확인 주기 200ms, 사용자별 접수 제한 3건/1분(반복 요청 제한 검사용 — 다른 테스트는 사용자마다 3건 이하). + */ +@Tag("postgres") +// 끝난 class의 context를 닫아 finalizer sweep이 닫힌 DB를 계속 두드리지 않게 한다 +@DirtiesContext +@ActiveProfiles("test") +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = { + "gcs.project-id=proovy-test", + "gcs.endpoint=http://127.0.0.1:1", + "jwt.secret=test-jwt-secret-test-jwt-secret-0123456789", + "oauth.kakao.client-id=test", "oauth.kakao.client-secret=test", + "oauth.naver.client-id=test", "oauth.naver.client-secret=test", + "openrouter.api.key=test", + "spring.data.redis.host=127.0.0.1", "spring.data.redis.port=1", + "spring.task.scheduling.pool.size=1", + "proovy.subscription.plan-transition-fixed-delay-ms=3600000", + "proovy.subscription.auto-renew-fixed-delay-ms=3600000", + "proovy.brain.auth.mode=shared-token", + "proovy.brain.auth.shared-token=t184-shared-token", + "proovy.run.coordinator.enabled=false", + "proovy.run.admission.rate-limit.max-requests=3", + "proovy.run.admission.rate-limit.window=1m", + "proovy.run.cancel.sweep-interval=200ms", + "proovy.run.cancel.cleanup-timeout=5s" +}) +class RunCancelPostgresTest { + + private static final String FENCE = "fence-t184"; + private static final Duration WAIT = Duration.ofSeconds(20); + private static final Duration CLEANUP_TIMEOUT = Duration.ofSeconds(5); + private static final long DAILY_CREDITS = 100; + private static final ObjectMapper RAW = new ObjectMapper(); + + private static PostgresTestDatabase db; + private static CancelFakeBrain brain; + + @BeforeAll + static void setUp() { + db = PostgresTestDatabase.createMigrated(); + db.runScript("db/seed/product-seed.sql"); + brain = new CancelFakeBrain(); + } + + @DynamicPropertySource + static void properties(DynamicPropertyRegistry registry) { + registry.add("spring.datasource.url", () -> db.jdbcUrl()); + registry.add("spring.datasource.username", PostgresTestDatabase::adminUser); + registry.add("spring.datasource.password", PostgresTestDatabase::adminPassword); + registry.add("proovy.brain.base-url", () -> brain.baseUrl()); + } + + @AfterAll + static void tearDown() { + if (brain != null) { + brain.close(); + } + if (db != null) { + db.close(); + } + } + + @LocalServerPort int port; + @MockitoBean AccessTokenBlacklistService blacklist; + @Autowired JwtTokenProvider tokens; + @Autowired RunAdmissionService admission; + @Autowired RunCancelService cancels; + @Autowired RunTerminalService terminals; + @Autowired RunEventStore events; + @Autowired RequestUsageService usage; + @Autowired JdbcTemplate jdbc; + @Autowired TransactionTemplate tx; + @Autowired ObjectMapper mapper; + + private final HttpClient http = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1) + .connectTimeout(Duration.ofSeconds(5)).build(); + + // ------------------------------------------------------------------ + // 판정표 (browser-run-v1 cancel/cases.json) — production endpoint + // ------------------------------------------------------------------ + + @Test + @DisplayName("cancel/cases.json 9 case를 실제 HTTP endpoint로: body 검사가 run 조회보다 먼저, 멱등·terminal·404 동일 body") + void decisionTableRunsThroughProductionEndpoint() throws Exception { + JsonNode table = load("cancel/cases.json"); + List notFoundBodies = new ArrayList<>(); + for (JsonNode testCase : table.get("cases")) { + String name = testCase.get("name").asText(); + long owner = newUser(); + UUID runId = switch (testCase.get("run").asText()) { + case "running" -> runningRun(owner, CancelMode.NEVER_CONFIRM); + case "canceling" -> { + UUID id = runningRun(owner, CancelMode.NEVER_CONFIRM); + cancels.cancel(owner, id.toString(), null); + yield id; + } + case "completed" -> completedRun(owner); + case "pending" -> admit(owner).runId(); + case "missing" -> UUID.randomUUID(); + case "other_user" -> admit(newUser()).runId(); + default -> throw new IllegalArgumentException(name); + }; + JsonNode rawBody = testCase.get("rawBody"); + HttpResponse response = postCancel(token(owner), runId.toString(), + rawBody.isNull() ? null : rawBody.asText()); + JsonNode expected = testCase.get("expected"); + assertThat(response.statusCode()).as(name + " " + response.body()).isEqualTo(expected.get("http").asInt()); + assertThat(response.headers().firstValue("Content-Type").orElse("")).as(name).startsWith("application/json"); + JsonNode body = RAW.readTree(response.body()); + if (expected.has("fixture")) { + ObjectNode fixture = (ObjectNode) load(expected.get("fixture").asText() + ".json").get("body").deepCopy(); + ((ObjectNode) fixture.get("result")).put("runId", runId.toString()); + assertThat(body).as(name).isEqualTo(fixture); + assertThat(body.at("/result/status").asText()).isEqualTo(expected.get("status").asText()); + } else { + assertThat(body.get("isSuccess").asBoolean()).as(name).isFalse(); + assertThat(body.get("code").asText()).as(name).isEqualTo(expected.get("code").asText()); + if (expected.get("http").asInt() == 404) { + notFoundBodies.add(response.body()); + } + } + if (rawBody.isNull() && "running".equals(testCase.get("run").asText())) { + assertThat(status(runId)).isEqualTo("CANCELING"); + } + if (!rawBody.isNull() && !"missing".equals(testCase.get("run").asText())) { + // body 거부는 아무것도 바꾸지 않는다 + assertThat(status(runId)).as(name).isIn("RUNNING", "PENDING"); + } + } + // 없는 run·타인 run은 같은 404 body(존재 비노출) + assertThat(notFoundBodies).hasSize(2); + assertThat(notFoundBodies.get(0)).isEqualTo(notFoundBodies.get(1)); + } + + @Test + @DisplayName("대문자·형식 다른 runId·삭제 중 Note의 run·인증 없음은 상태를 바꾸지 않는다") + void accessEdgesDoNotTouchTheRun() throws Exception { + long user = newUser(); + UUID runId = admit(user).runId(); + assertThat(postCancel(token(user), runId.toString().toUpperCase(), null).statusCode()).isEqualTo(404); + assertThat(postCancel(token(user), "not-a-run", null).statusCode()).isEqualTo(404); + HttpResponse anonymous = http.send(HttpRequest.newBuilder(uri("/api/ai-runs/" + runId + "/cancel")) + .POST(HttpRequest.BodyPublishers.noBody()).timeout(Duration.ofSeconds(10)).build(), + HttpResponse.BodyHandlers.ofString()); + assertThat(anonymous.statusCode()).isEqualTo(401); + assertThat(status(runId)).isEqualTo("PENDING"); + + jdbc.update("UPDATE notes SET status = 'DELETING', deletion_requested_at = now() WHERE note_id = ?", + noteOf(runId)); + HttpResponse deleting = postCancel(token(user), runId.toString(), null); + assertThat(deleting.statusCode()).isEqualTo(404); + assertThat(RAW.readTree(deleting.body()).get("code").asText()).isEqualTo("RUN4041"); + assertThat(status(runId)).isEqualTo("PENDING"); + } + + // ------------------------------------------------------------------ + // PENDING 취소 + // ------------------------------------------------------------------ + + @Test + @DisplayName("PENDING 취소: 한 transaction에 CANCELED·message·run.canceled·보상 1회, Brain 호출 없음, Note 슬롯 즉시 해제") + void pendingCancelIsImmediateCompensatedOnceAndNeverReachesBrain() throws Exception { + long user = newUser(); + RunAdmission run = admit(user); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS - charged(run.requestId())); + + JsonNode result = postCancelJson(token(user), run.runId()); + assertThat(result.get("status").asText()).isEqualTo("CANCELED"); + assertThat(result.get("outcome").asText()).isEqualTo("accepted"); + + Map row = runRow(run.runId()); + assertThat(row.get("status")).isEqualTo("CANCELED"); + assertThat(row.get("attempt")).isEqualTo(0); + assertThat(row.get("cancel_cause")).isEqualTo("user_requested"); + assertThat(row.get("cancel_requested_from")).isEqualTo("PENDING"); + assertThat(row.get("completed_at")).isNotNull(); + assertThat(row.get("runtime_cleanup_at")).isNull(); + assertThat(messageStatus(run.assistantMessageId())).isEqualTo("CANCELED"); + assertThat(eventTypes(run.runId())).containsExactly("run.created", "run.canceled"); + + // 저장 payload는 canceled-pending stream fixture의 run.canceled와 같은 모양 + JsonNode fixture = load("streams/valid/canceled-pending.json").get("frames").get(1).get("payload"); + ObjectNode expected = fixture.deepCopy(); + expected.put("assistantMessageId", run.assistantMessageId().toString()); + assertThat(eventPayload(run.runId(), "run.canceled")).isEqualTo(expected); + + assertThat(compensations(run.requestId())).containsExactly("PENDING_CANCELED"); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + assertThat(brain.cancelCalls(run.runId().toString())).isEmpty(); + assertThat(brain.streamCalls(run.runId().toString())).isZero(); + + JsonNode status = getJson(token(user), "/api/ai-runs/" + run.runId()); + assertThat(status.at("/result/status").asText()).isEqualTo("CANCELED"); + assertThat(status.at("/result/cancelable").asBoolean()).isFalse(); + + // 중복 cancel은 멱등이고 다시 보상하지 않는다 + JsonNode again = postCancelJson(token(user), run.runId()); + assertThat(again.get("outcome").asText()).isEqualTo("already_terminal"); + assertThat(again.get("status").asText()).isEqualTo("CANCELED"); + assertThat(compensations(run.requestId())).hasSize(1); + + // Active Run 자리가 풀려 같은 Note에 새 요청을 받는다. 취소된 run은 자동 재실행되지 않는다. + RunAdmission next = admit(user, String.valueOf(run.noteId())); + assertThat(next.runId()).isNotEqualTo(run.runId()); + assertThat(status(run.runId())).isEqualTo("CANCELED"); + } + + @Test + @DisplayName("같은 PENDING run에 동시 cancel 6건: accepted 1·already_terminal 5, 보상·run.canceled 하나") + void concurrentPendingCancelsSettleOnce() throws Exception { + long user = newUser(); + RunAdmission run = admit(user); + List outcomes = race(6, i -> () -> cancels.cancel(user, run.runId().toString(), null).outcome() + .wireValue()); + assertThat(outcomes).containsOnly("accepted", "already_terminal"); + assertThat(outcomes.stream().filter("accepted"::equals)).hasSize(1); + assertThat(eventTypes(run.runId())).containsExactly("run.created", "run.canceled"); + assertThat(compensations(run.requestId())).containsExactly("PENDING_CANCELED"); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + } + + @Test + @DisplayName("PENDING 취소와 coordinator claim 경합: 하나만 이긴다 — 취소가 이기면 미전달·보상, claim이 이기면 CANCELING·무보상") + void pendingCancelRacesClaimAndExactlyOneWins() throws Exception { + RunClaimStore claims = new RunClaimStore(jdbc, tx); + String owner = "t184-claimer"; + for (int round = 0; round < 8; round++) { + failOtherPendingRuns(); + long user = newUser(); + RunAdmission run = admit(user); + brain.mode(run.runId().toString(), CancelMode.NEVER_CONFIRM); + assertThat(claims.acquireCoordinatorLease(owner, Duration.ofSeconds(30))).isTrue(); + String fence = "fence-race-" + round; + List results = race(2, i -> i == 0 + ? () -> claims.claimNext(owner, fence, Duration.ofSeconds(30), Duration.ZERO) + : () -> cancels.cancel(user, run.runId().toString(), null)); + @SuppressWarnings("unchecked") + Optional claimed = (Optional) results.get(0); + String status = status(run.runId()); + if (claimed.isPresent()) { + assertThat(claimed.get().runId()).isEqualTo(run.runId()); + // claim 뒤 취소: 실행 중 취소 경로, 차감 유지 + assertThat(status).isIn("CANCELING", "CANCELED"); + assertThat(runRow(run.runId()).get("cancel_requested_from")).isEqualTo("RUNNING"); + assertThat(compensations(run.requestId())).isEmpty(); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS - charged(run.requestId())); + } else { + assertThat(status).isEqualTo("CANCELED"); + assertThat(runRow(run.runId()).get("attempt")).isEqualTo(0); + assertThat(compensations(run.requestId())).containsExactly("PENDING_CANCELED"); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + } + claims.releaseCoordinatorLease(owner); + } + failOtherPendingRuns(); + } + + @Test + @DisplayName("PENDING 취소와 deadline 만료·Server 실패 경합: terminal 하나, 보상은 request당 한 번") + void pendingCancelRacesDeadlineAndServerFailureCompensatingOnce() throws Exception { + for (int round = 0; round < 6; round++) { + long user = newUser(); + boolean deadlineRace = round % 2 == 0; + UUID runId = insertPendingRun(user, Instant.now().plusMillis(deadlineRace ? 300 : 600_000)); + if (deadlineRace) { + Thread.sleep(400); + } + List results = race(2, i -> i == 0 + ? () -> deadlineRace ? terminals.expireDeadline(runId) + : terminals.failByServer(runId, 0, null, RunFailureCode.INTERNAL_ERROR, true) + : () -> cancels.cancel(user, runId.toString(), null)); + TerminalSettlement settlement = (TerminalSettlement) results.get(0); + String requestId = (String) runRow(runId).get("request_id"); + String status = status(runId); + if (settlement.outcome() == Outcome.COMMITTED) { + assertThat(status).isEqualTo("FAILED"); + assertThat(compensations(requestId)).containsExactly("INFRASTRUCTURE_FAILURE"); + } else { + assertThat(settlement.outcome()).isEqualTo(Outcome.NOT_ACCEPTING); + assertThat(status).isEqualTo("CANCELED"); + assertThat(compensations(requestId)).containsExactly("PENDING_CANCELED"); + } + assertThat(eventTypes(runId)).hasSize(1); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + } + } + + // ------------------------------------------------------------------ + // RUNNING/RETRYING 취소 + // ------------------------------------------------------------------ + + @Test + @DisplayName("RUNNING 취소: CANCELING 동안 늦은 terminal·Activity·deadline 차단, 새 run·슬롯 유지 → Brain 정리 확인 뒤 CANCELED, 차감 유지") + void runningCancelFencesLateResultsAndHoldsSlotsUntilBrainConfirms() throws Exception { + long user = newUser(); + UUID runId = runningRun(user, CancelMode.NEVER_CONFIRM); + RunAdmission admitted = admissionOf(runId); + jdbc.update("UPDATE ai_runs SET status = 'RETRYING', retry_count = 1 WHERE run_id = ?", runId); + + JsonNode result = postCancelJson(token(user), runId); + assertThat(result.get("status").asText()).isEqualTo("CANCELING"); + assertThat(result.get("outcome").asText()).isEqualTo("accepted"); + Map row = runRow(runId); + assertThat(row.get("cancel_requested_from")).isEqualTo("RETRYING"); + assertThat(row.get("lease_owner")).isEqualTo("t184-worker"); + assertThat(eventTypes(runId)).containsExactly("run.created", "activity.started", "run.canceling"); + ObjectNode canceling = mapper.createObjectNode().put("assistantMessageId", admitted.assistantMessageId().toString()) + .put("status", "CANCELING"); + assertThat(eventPayload(runId, "run.canceling")).isEqualTo(canceling); + assertThat(messageStatus(admitted.assistantMessageId())).isEqualTo("PENDING"); + + // 명시적 Brain cancel: 현재 attempt·fence·cause + await(() -> !brain.cancelCalls(runId.toString()).isEmpty()); + JsonNode call = brain.cancelCalls(runId.toString()).get(0); + assertThat(call.get("attempt").asInt()).isEqualTo(1); + assertThat(call.get("fence_token").asText()).isEqualTo(FENCE); + assertThat(call.get("cause").asText()).isEqualTo("user_requested"); + + // 늦은 결과 차단 (1.80·1.82 규칙, 이 run에서 실제로) + assertThat(terminals.settle(clarificationEvent(runId)).outcome()).isEqualTo(Outcome.NOT_ACCEPTING); + assertThat(terminals.failByServer(runId, 1, FENCE, RunFailureCode.INTERNAL_ERROR, true).outcome()) + .isEqualTo(Outcome.NOT_ACCEPTING); + assertThat(terminals.expireDeadline(runId).outcome()).isEqualTo(Outcome.NOT_ACCEPTING); + assertThat(events.appendActivity(activity(runId, "activity-2", "activity.started")).outcome()) + .isEqualTo(ActivityAppendResult.Outcome.NOT_ACCEPTING); + assertThat(events.appendActivity(activity(runId, "activity-1", "activity.canceled")).outcome()) + .isEqualTo(ActivityAppendResult.Outcome.APPENDED); + + // 정리 확인 전: 상태 CANCELING, 같은 Note 새 요청 거부, 전역 슬롯 계산에 포함 + Thread.sleep(800); + assertThat(status(runId)).isEqualTo("CANCELING"); + assertThatThrownBy(() -> admit(user, String.valueOf(admitted.noteId()))) + .isInstanceOf(BusinessException.class) + .satisfies(e -> assertThat(((BusinessException) e).getErrorCode()).isEqualTo(ErrorCode.RUN4091)); + assertThat(jdbc.queryForObject(""" + SELECT count(*) FROM ai_runs WHERE run_id = ? AND status IN ('RUNNING', 'RETRYING', 'CANCELING') + AND attempt >= 1""", Long.class, runId)).isEqualTo(1); + JsonNode status = getJson(token(user), "/api/ai-runs/" + runId); + assertThat(status.at("/result/status").asText()).isEqualTo("CANCELING"); + assertThat(status.at("/result/cancelable").asBoolean()).isFalse(); + assertThat(postCancelJson(token(user), runId).get("outcome").asText()).isEqualTo("already_canceling"); + + // Brain이 정리를 끝내면 다음 확인에서 CANCELED + brain.mode(runId.toString(), CancelMode.CONFIRM); + awaitStatus(runId, "CANCELED"); + row = runRow(runId); + assertThat(row.get("lease_owner")).isNull(); + assertThat(row.get("runtime_cleanup_at")).isNotNull(); + assertThat(Duration.between(instant(row.get("cancel_requested_at")), instant(row.get("completed_at")))) + .isLessThan(CLEANUP_TIMEOUT); + assertThat(messageStatus(admitted.assistantMessageId())).isEqualTo("CANCELED"); + assertThat(eventTypes(runId)).containsExactly("run.created", "activity.started", "run.canceling", + "activity.canceled", "run.canceled"); + JsonNode canceled = eventPayload(runId, "run.canceled"); + assertThat(canceled.get("cause").asText()).isEqualTo("user_requested"); + assertThat(canceled.get("status").asText()).isEqualTo("CANCELED"); + // RUNNING 이후 사용자 취소는 차감 유지(순사용량 1) + assertThat(compensations(admitted.requestId())).isEmpty(); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS - charged(admitted.requestId())); + + // 정리 뒤 같은 Note 새 요청 가능. 취소된 run은 다시 실행되지 않는다. + admit(user, String.valueOf(admitted.noteId())); + assertThat(brain.streamCalls(runId.toString())).isZero(); + assertThat(postCancelJson(token(user), runId).get("outcome").asText()).isEqualTo("already_terminal"); + } + + @Test + @DisplayName("Brain 정리 미확인·Brain 장애: cleanup-timeout까지 CANCELING을 유지하고 그 뒤 CANCELED로 확정") + void cleanupTimeoutConfirmsWithoutBrainAcknowledgement() throws Exception { + long first = newUser(); + long second = newUser(); + UUID neverConfirms = runningRun(first, CancelMode.NEVER_CONFIRM); + UUID unavailable = runningRun(second, CancelMode.UNAVAILABLE); + cancels.cancel(first, neverConfirms.toString(), null); + cancels.cancel(second, unavailable.toString(), null); + + Thread.sleep(CLEANUP_TIMEOUT.toMillis() - 1500); + assertThat(status(neverConfirms)).isEqualTo("CANCELING"); + assertThat(status(unavailable)).isEqualTo("CANCELING"); + assertThat(brain.cancelCalls(neverConfirms.toString()).size()).isGreaterThan(3); + + awaitStatus(neverConfirms, "CANCELED"); + awaitStatus(unavailable, "CANCELED"); + for (UUID runId : List.of(neverConfirms, unavailable)) { + Map row = runRow(runId); + assertThat(Duration.between(instant(row.get("cancel_requested_at")), instant(row.get("completed_at")))) + .isGreaterThanOrEqualTo(CLEANUP_TIMEOUT); + assertThat(eventTypes(runId)).endsWith("run.canceling", "run.canceled"); + assertThat(compensations((String) row.get("request_id"))).isEmpty(); + } + } + + @Test + @DisplayName("Brain record absent: run lease가 살아 있는 동안(worker가 아직 보낼 수 있음) 기다리고, lease 만료 뒤 확정") + void absentBrainRecordWaitsForTheRunLease() throws Exception { + long user = newUser(); + UUID runId = runningRun(user, CancelMode.ABSENT); + cancels.cancel(user, runId.toString(), null); + Thread.sleep(1200); + assertThat(status(runId)).isEqualTo("CANCELING"); + assertThat(brain.cancelCalls(runId.toString())).isNotEmpty(); + + jdbc.update("UPDATE ai_runs SET lease_expires_at = now() WHERE run_id = ?", runId); + awaitStatus(runId, "CANCELED"); + Map row = runRow(runId); + assertThat(Duration.between(instant(row.get("cancel_requested_at")), instant(row.get("completed_at")))) + .isLessThan(CLEANUP_TIMEOUT); + } + + @Test + @DisplayName("Brain absent 응답 동안 늦은 heartbeat가 lease를 갱신하면 잠금 뒤 다시 보고 확정하지 않는다") + void absentConfirmationRechecksTheLeaseUnderTheRunLock() throws Exception { + long user = newUser(); + UUID runId = runningRun(user, CancelMode.ABSENT); + jdbc.update("UPDATE ai_runs SET lease_expires_at = now() WHERE run_id = ?", runId); + // 확인 전 읽기는 lease 만료를 보고, Brain 응답 직전 worker heartbeat가 lease를 되살린다 + brain.beforeNextCancelAck(runId.toString(), () -> jdbc.update( + "UPDATE ai_runs SET lease_expires_at = now() + interval '1 minute' WHERE run_id = ?", runId)); + cancels.cancel(user, runId.toString(), null); + await(() -> !brain.cancelCalls(runId.toString()).isEmpty()); + Thread.sleep(1000); + assertThat(status(runId)).isEqualTo("CANCELING"); + + jdbc.update("UPDATE ai_runs SET lease_expires_at = now() WHERE run_id = ?", runId); + awaitStatus(runId, "CANCELED"); + assertThat(runRow(runId).get("runtime_cleanup_at")).isNotNull(); + } + + @Test + @DisplayName("cleanup-timeout이 지난 run은 Brain을 부르지 않고 확정, hang인 Brain은 남은 시간까지만 기다린다") + void timedOutRunsDoNotWaitForBrain() throws Exception { + long user = newUser(); + UUID overdue = runningRun(user, CancelMode.HANG); + // 다른 인스턴스가 1분 전에 CANCELING으로 바꾼 run(그 인스턴스는 사라짐) + jdbc.update(""" + UPDATE ai_runs SET status = 'CANCELING', cancel_cause = 'user_requested', cancel_requested_from = status, + cancel_requested_at = now() - interval '1 minute', grant_digest = NULL, grant_expires_at = NULL + WHERE run_id = ?""", overdue); + awaitStatus(overdue, "CANCELED"); + assertThat(brain.cancelCalls(overdue.toString())).isEmpty(); + + // hang: client request-timeout(10s)이 아니라 cleanup-timeout(5s)에 확정된다 + UUID hanging = runningRun(newUser(), CancelMode.HANG); + cancels.cancel(runRowUser(hanging), hanging.toString(), null); + awaitStatus(hanging, "CANCELED"); + Map row = runRow(hanging); + assertThat(Duration.between(instant(row.get("cancel_requested_at")), instant(row.get("completed_at")))) + .isGreaterThanOrEqualTo(CLEANUP_TIMEOUT).isLessThan(Duration.ofSeconds(8)); + } + + @Test + @DisplayName("확정이 계속 실패하는 run은 backoff로 주기 확인에서 빠지고(전부 rollback, cleanup 예약 없음), 취소 API의 확인은 그대로") + void failingConfirmationBacksOffAndRollsBackAtomically() throws Exception { + long user = newUser(); + UUID runId = runningRun(user, CancelMode.CONFIRM); + jdbc.execute(""" + CREATE FUNCTION t184_block_cancel() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.run_id = '%s' AND NEW.status = 'CANCELED' THEN + RAISE EXCEPTION 't184 blocked' USING ERRCODE = 'P0001'; + END IF; + RETURN NEW; + END $$""".formatted(runId)); + jdbc.execute("CREATE TRIGGER t184_block_cancel BEFORE UPDATE ON ai_runs FOR EACH ROW " + + "EXECUTE FUNCTION t184_block_cancel()"); + try { + cancels.cancel(user, runId.toString(), null); + // 첫 확인 canceling → 다음 sweep에서 terminal → 확정 실패 → backoff(30s) + await(() -> brain.cancelCalls(runId.toString()).size() >= 2); + Thread.sleep(400); + int calls = brain.cancelCalls(runId.toString()).size(); + Thread.sleep(1500); + assertThat(brain.cancelCalls(runId.toString())).hasSize(calls); + Map row = runRow(runId); + assertThat(row.get("status")).isEqualTo("CANCELING"); + assertThat(row.get("runtime_cleanup_at")).isNull(); + assertThat(messageStatus(admissionOf(runId).assistantMessageId())).isNotEqualTo("CANCELED"); + } finally { + jdbc.execute("DROP TRIGGER t184_block_cancel ON ai_runs"); + jdbc.execute("DROP FUNCTION t184_block_cancel()"); + } + // 다시 온 취소 요청은 backoff와 무관하게 곧바로 확인한다 + assertThat(postCancelJson(token(user), runId).get("outcome").asText()).isEqualTo("already_canceling"); + awaitStatus(runId, "CANCELED"); + assertThat(runRow(runId).get("runtime_cleanup_at")).isNotNull(); + } + + @Test + @DisplayName("RUNNING 완료와 취소 경합: 결과 확정과 취소 중 하나만, 순사용량은 승자 정책대로") + void runningCompletionRacesCancel() throws Exception { + for (int round = 0; round < 6; round++) { + long user = newUser(); + UUID runId = runningRun(user, CancelMode.CONFIRM); + RunAdmission admitted = admissionOf(runId); + List results = race(2, i -> i == 0 + ? () -> terminals.settle(clarificationEvent(runId)) + : () -> cancels.cancel(user, runId.toString(), null)); + TerminalSettlement settlement = (TerminalSettlement) results.get(0); + if (settlement.outcome() == Outcome.COMMITTED) { + assertThat(status(runId)).isEqualTo("COMPLETED"); + assertThat(((RunCancelResult) results.get(1)).outcome().wireValue()) + .isEqualTo("already_terminal"); + assertThat(eventTypes(runId)).doesNotContain("run.canceling", "run.canceled"); + // clarification은 보상(순사용량 0) + assertThat(compensations(admitted.requestId())).containsExactly("CLARIFICATION_ONLY"); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + } else { + assertThat(settlement.outcome()).isEqualTo(Outcome.NOT_ACCEPTING); + awaitStatus(runId, "CANCELED"); + assertThat(eventTypes(runId)).doesNotContain("run.completed", "response.page.started"); + assertThat(compensations(admitted.requestId())).isEmpty(); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS - charged(admitted.requestId())); + } + } + } + + @Test + @DisplayName("RUNNING deadline 만료와 취소 경합: timeout 실패(보상)와 취소(무보상) 중 하나만") + void runningDeadlineRacesCancel() throws Exception { + for (int round = 0; round < 6; round++) { + long user = newUser(); + UUID runId = insertPendingRun(user, Instant.now().plusMillis(400)); + claim(runId); + brain.executing(runId.toString(), 1, FENCE); + Thread.sleep(500); + List results = race(2, i -> i == 0 + ? () -> terminals.expireDeadline(runId) + : () -> cancels.cancel(user, runId.toString(), null)); + TerminalSettlement settlement = (TerminalSettlement) results.get(0); + String requestId = (String) runRow(runId).get("request_id"); + if (settlement.outcome() == Outcome.COMMITTED) { + assertThat(status(runId)).isEqualTo("FAILED"); + assertThat(runRow(runId).get("failure_code")).isEqualTo("timeout"); + assertThat(compensations(requestId)).containsExactly("INFRASTRUCTURE_FAILURE"); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + } else { + assertThat(settlement.outcome()).isEqualTo(Outcome.NOT_ACCEPTING); + // deadline이 지나도 CANCELING은 취소 경로가 끝낸다 + awaitStatus(runId, "CANCELED"); + assertThat(compensations(requestId)).isEmpty(); + assertThat(credits(user)).isEqualTo(DAILY_CREDITS - charged(requestId)); + } + } + } + + // ------------------------------------------------------------------ + // 단절 ≠ 취소, 반복 요청 제한 + // ------------------------------------------------------------------ + + @Test + @DisplayName("브라우저 SSE 단절은 취소가 아니다: run 상태·cancel 필드·Brain cancel 호출 없음") + void browserDisconnectIsNotCancel() throws Exception { + long user = newUser(); + UUID runId = runningRun(user, CancelMode.CONFIRM); + HttpResponse stream = http.send(HttpRequest.newBuilder(uri("/api/ai-runs/" + runId + "/events")) + .header("Authorization", "Bearer " + token(user)).header("Accept", "text/event-stream").GET().build(), + HttpResponse.BodyHandlers.ofInputStream()); + assertThat(stream.statusCode()).isEqualTo(200); + try (InputStream body = stream.body()) { + assertThat(body.read(new byte[64])).isPositive(); + } + Thread.sleep(1000); + Map row = runRow(runId); + assertThat(row.get("status")).isEqualTo("RUNNING"); + assertThat(row.get("cancel_requested_at")).isNull(); + assertThat(brain.cancelCalls(runId.toString())).isEmpty(); + } + + @Test + @DisplayName("반복 요청 제한(1.79 RUN4291 재사용): 새 requestId·취소 뒤 재요청도 세고, 같은 requestId 재전송과 다른 사용자는 막지 않는다") + void repeatedRequestLimitIsNotBypassedByNewRequestIdsOrCancellation() throws Exception { + long user = newUser(); + long other = newUser(); + List runs = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + RunAdmission run = admit(user); + runs.add(run); + // 실패·취소처럼 순사용량 0으로 끝난 요청도 제한 창에 남는다 + assertThat(cancels.cancel(user, run.runId().toString(), null).outcome().wireValue()).isEqualTo("accepted"); + } + assertThat(credits(user)).isEqualTo(DAILY_CREDITS); + assertThatThrownBy(() -> admit(user)) + .isInstanceOf(BusinessException.class) + .satisfies(e -> assertThat(((BusinessException) e).getErrorCode()).isEqualTo(ErrorCode.RUN4291)); + // 같은 requestId·같은 body 재조회는 새 실행이 아니라 기존 결과(제한 대상 아님) + RunAdmission last = runs.get(2); + RunAdmission replayed = admission.admit(user, requestBody(last.requestId(), null, textOf(last))); + assertThat(replayed.outcome()).isEqualTo(RunAdmission.Outcome.REPLAYED); + assertThat(replayed.runId()).isEqualTo(last.runId()); + // 다른 사용자는 영향 없음 + assertThat(admit(other).outcome()).isEqualTo(RunAdmission.Outcome.CREATED); + assertThat(jdbc.queryForObject("SELECT count(*) FROM ai_runs WHERE user_id = ?", Long.class, user)) + .isEqualTo(3); + } + + // ------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------ + + private final Map admissions = new ConcurrentHashMap<>(); + private final Map texts = new ConcurrentHashMap<>(); + + private RunAdmission admit(long user) { + return admit(user, null); + } + + private RunAdmission admit(long user, String noteId) { + String text = "x^2-5x+6=0을 풀어줘 " + UUID.randomUUID(); + RunAdmission run = admission.admit(user, requestBody(UUID.randomUUID().toString(), noteId, text)); + admissions.put(run.runId(), run); + texts.put(run.runId(), text); + return run; + } + + private RunAdmission admissionOf(UUID runId) { + return admissions.get(runId); + } + + private String textOf(RunAdmission run) { + return texts.get(run.runId()); + } + + private ObjectNode requestBody(String requestId, String noteId, String text) { + ObjectNode body = mapper.createObjectNode(); + body.put("requestId", requestId); + body.put("text", text); + if (noteId != null) { + body.put("noteId", noteId); + } + return body; + } + + /** 접수한 run을 coordinator claim처럼 RUNNING으로 만들고 Activity 하나를 저장한다. Brain record는 실행 중. */ + private UUID runningRun(long user, CancelMode mode) { + UUID runId = admit(user).runId(); + brain.mode(runId.toString(), mode); + claim(runId); + brain.executing(runId.toString(), 1, FENCE); + ActivityAppendResult started = events.appendActivity(activity(runId, "activity-1", "activity.started")); + assertThat(started.outcome()).isEqualTo(ActivityAppendResult.Outcome.APPENDED); + return runId; + } + + private UUID completedRun(long user) { + UUID runId = admit(user).runId(); + claim(runId); + assertThat(terminals.settle(clarificationEvent(runId)).outcome()).isEqualTo(Outcome.COMMITTED); + return runId; + } + + private void claim(UUID runId) { + jdbc.update(""" + UPDATE ai_runs SET status = 'RUNNING', attempt = 1, fence_token = ?, started_at = now(), + lease_owner = 't184-worker', lease_expires_at = now() + interval '1 minute', heartbeat_at = now() + WHERE run_id = ? AND status = 'PENDING' + """, FENCE, runId); + } + + /** claim 경합 테스트가 다른 테스트의 PENDING run을 집지 않게 미리 끝낸다. */ + private void failOtherPendingRuns() { + for (UUID pending : jdbc.queryForList("SELECT run_id FROM ai_runs WHERE status = 'PENDING'", UUID.class)) { + terminals.failByServer(pending, 0, null, RunFailureCode.INTERNAL_ERROR, true); + } + } + + private SourceEvent activity(UUID runId, String activityId, String type) { + ObjectNode envelope = (ObjectNode) load("../activity-v1/valid/input-analysis-started.json").deepCopy(); + envelope.put("run_id", runId.toString()).put("attempt", 1).put("fence_token", FENCE) + .put("source_event_id", runId + ":" + activityId + ":" + type) + .put("occurred_at", Instant.now().truncatedTo(ChronoUnit.MILLIS).toString()) + .put("type", type); + ObjectNode payload = (ObjectNode) envelope.get("payload"); + payload.put("activity_id", activityId); + if (!"activity.started".equals(type)) { + payload.remove("label"); + payload.remove("response_page_id"); + } + return InternalRunWire.read(envelope, SourceEvent.class); + } + + private SourceEvent clarificationEvent(UUID runId) { + ObjectNode source = (ObjectNode) load("../terminal-v2/pairs/clarification.json").get("source").deepCopy(); + source.put("run_id", runId.toString()).put("attempt", 1).put("fence_token", FENCE) + .put("source_event_id", runId + ":terminal"); + return InternalRunWire.read(source, SourceEvent.class); + } + + private long newUser() { + String key = UUID.randomUUID().toString(); + long user = jdbc.queryForObject(""" + INSERT INTO users (nickname, provider, provider_user_id, email, created_at, updated_at) + VALUES ('t184', 'KAKAO', ?, ?, now(), now()) RETURNING user_id + """, Long.class, key, key + "@example.test"); + jdbc.update(""" + INSERT INTO credit_balance (user_id, daily_free_credit, daily_free_limit, daily_expires_at, free_credit, + paid_credit, created_at, updated_at) + VALUES (?, ?, ?, now() + interval '1 day', 0, 0, now(), now()) + """, user, DAILY_CREDITS, DAILY_CREDITS); + return user; + } + + /** 접수가 남기는 모양의 PENDING run·최초 차감(deadline을 짧게 잡는 경합 검사용). */ + private UUID insertPendingRun(long user, Instant deadline) { + return tx.execute(status -> { + long note = jdbc.queryForObject(""" + INSERT INTO notes (user_id, title, created_at, updated_at) VALUES (?, 't184', now(), now()) + RETURNING note_id + """, Long.class, user); + UUID userMessage = UUID.randomUUID(); + UUID assistantMessage = UUID.randomUUID(); + jdbc.update(""" + INSERT INTO chat_messages (message_id, note_id, role, status, user_input, plain_text, created_at) + VALUES (?, ?, 'USER', 'COMPLETED', '{"text":"풀어줘"}', '풀어줘', now()), + (?, ?, 'ASSISTANT', 'PENDING', NULL, NULL, now() + interval '1 microsecond') + """, userMessage, note, assistantMessage, note); + UUID runId = UUID.randomUUID(); + String requestId = UUID.randomUUID().toString(); + jdbc.update(""" + INSERT INTO ai_runs (run_id, request_id, user_id, note_id, user_message_id, assistant_message_id, + request_fingerprint, request_snapshot, policy_snapshot, budget_policy_version, + request_deadline_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, '{"input":{"text":"풀어줘"}}', '{"limits":{}}', 'budget-v1', ?, now()) + """, runId, requestId, user, note, userMessage, assistantMessage, "f".repeat(64), + OffsetDateTime.ofInstant(deadline.truncatedTo(ChronoUnit.MICROS), ZoneOffset.UTC)); + usage.chargeRequest(user, requestId, runId); + return runId; + }); + } + + private Map runRow(UUID runId) { + return jdbc.queryForMap("SELECT * FROM ai_runs WHERE run_id = ?", runId); + } + + private long runRowUser(UUID runId) { + return jdbc.queryForObject("SELECT user_id FROM ai_runs WHERE run_id = ?", Long.class, runId); + } + + private String status(UUID runId) { + return jdbc.queryForObject("SELECT status FROM ai_runs WHERE run_id = ?", String.class, runId); + } + + private long noteOf(UUID runId) { + return jdbc.queryForObject("SELECT note_id FROM ai_runs WHERE run_id = ?", Long.class, runId); + } + + private String messageStatus(UUID messageId) { + return jdbc.queryForObject("SELECT status FROM chat_messages WHERE message_id = ?", String.class, messageId); + } + + private List eventTypes(UUID runId) { + return jdbc.queryForList("SELECT event_type FROM ai_run_events WHERE run_id = ? ORDER BY seq", String.class, + runId); + } + + private JsonNode eventPayload(UUID runId, String type) throws Exception { + return RAW.readTree(jdbc.queryForObject( + "SELECT payload::text FROM ai_run_events WHERE run_id = ? AND event_type = ?", String.class, runId, + type)); + } + + private List compensations(String requestId) { + return jdbc.queryForList("SELECT reason FROM request_usage_compensations WHERE request_id = ?", String.class, + requestId); + } + + /** 접수가 차감한 credit(플랜·요청 단위에 따라 1이 아닐 수 있다). */ + private long charged(String requestId) { + return jdbc.queryForObject("SELECT credit_amount FROM request_usage_charges WHERE request_id = ?", Long.class, + requestId); + } + + private long credits(long user) { + return jdbc.queryForObject("SELECT daily_free_credit + free_credit + paid_credit FROM credit_balance " + + "WHERE user_id = ?", Long.class, user); + } + + private static Instant instant(Object timestamp) { + return timestamp instanceof OffsetDateTime odt ? odt.toInstant() : ((java.sql.Timestamp) timestamp).toInstant(); + } + + private void awaitStatus(UUID runId, String expected) { + await(() -> expected.equals(status(runId))); + } + + private static void await(BooleanSupplier condition) { + Instant until = Instant.now().plus(WAIT); + while (!condition.getAsBoolean()) { + if (Instant.now().isAfter(until)) { + throw new AssertionError("condition not met within " + WAIT); + } + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + } + + private String token(long user) { + return tokens.generateTokens(user).accessToken(); + } + + private URI uri(String path) { + return URI.create("http://127.0.0.1:" + port + path); + } + + private HttpResponse postCancel(String token, String runId, String body) throws Exception { + HttpRequest.Builder request = HttpRequest.newBuilder(uri("/api/ai-runs/" + runId + "/cancel")) + .header("Authorization", "Bearer " + token).timeout(Duration.ofSeconds(10)); + if (body == null) { + request.POST(HttpRequest.BodyPublishers.noBody()); + } else { + request.header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString(body)); + } + return http.send(request.build(), HttpResponse.BodyHandlers.ofString()); + } + + private JsonNode postCancelJson(String token, UUID runId) throws Exception { + HttpResponse response = postCancel(token, runId.toString(), null); + assertThat(response.statusCode()).as(response.body()).isEqualTo(200); + return RAW.readTree(response.body()).get("result"); + } + + private JsonNode getJson(String token, String path) throws Exception { + HttpResponse response = http.send(HttpRequest.newBuilder(uri(path)) + .header("Authorization", "Bearer " + token).timeout(Duration.ofSeconds(10)).GET().build(), + HttpResponse.BodyHandlers.ofString()); + assertThat(response.statusCode()).as(path + " " + response.body()).isEqualTo(200); + return RAW.readTree(response.body()); + } + + private interface Task { + Callable at(int index); + } + + private static List race(int workers, Task task) throws Exception { + CyclicBarrier barrier = new CyclicBarrier(workers); + ExecutorService pool = Executors.newFixedThreadPool(workers); + try { + List> futures = new ArrayList<>(); + for (int i = 0; i < workers; i++) { + Callable work = task.at(i); + futures.add(pool.submit(() -> { + barrier.await(); + return work.call(); + })); + } + List results = new ArrayList<>(); + for (Future future : futures) { + results.add(future.get()); + } + return results; + } finally { + pool.shutdownNow(); + } + } +}