diff --git a/CHANGELOG.md b/CHANGELOG.md index d0cc0d5..ea77ffc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,26 @@ see [docs/compatibility.md](docs/compatibility.md) for what 0.x guarantees). ## [Unreleased] ### Fixed +- `lnpl serve` / `build_app()` on a repository driver without the + idempotency claim methods (today `lnpl-postgres`) ignored an + `Idempotency-Key` header without warning, so a retried request ran its + workflow again. On such a driver a workflow request carrying the header + (M22) and every `POST /-/events/` delivery, whose CloudEvents `id` + is its key (E8), are now refused with 501 `idempotency-unsupported` + (problem+json) and nothing runs; workflow requests without the header + run as before. `lnpl serve` prints one boot line to stderr, + `lnpl serve: repository driver -- idempotency: + unsupported ...`, and the `GET /-/readyz` 200 body now carries + `"capabilities": {"idempotency": }` (information only, never a + 503). Requests on `--backend fake` are not refused (issue #228). +- `lnpl outbox drain`, `lnpl outbox ack` and `lnpl relay` crashed with an + `AttributeError` traceback on a persistent backend whose driver lacks + `drain_outbox`/`ack_outbox` (today `lnpl-postgres`). They now print one + stderr line, `error: backend does not implement the + outbox drain contract (drain_outbox/ack_outbox) — see docs/backends.md + §3`, close the driver and exit with rc 2 before any drain, ack or + publish call; `--backend fake` keeps its existing rc 2 refusal + (issue #229). - `capability http`'s `retry`/`breaker`/`path` clauses were parsed and compiled but silently dropped before reaching `HttpNetworkDriver` — both `lnpl run`/`lnpl serve` (`cli.py:_open_endpoints`) and @@ -143,6 +163,19 @@ see [docs/compatibility.md](docs/compatibility.md) for what 0.x guarantees). (issue #226). ### Added +- `RepositoryDriver.supports_idempotency()` and + `RepositoryDriver.supports_outbox_drain()` capability predicates + (`lnpl.drivers`). The first is True only when `idempotency_begin`, + `idempotency_finish` and `idempotency_release` are all callable on the + driver, the second only when `drain_outbox` and `ack_outbox` both are, + so an external driver that implements a set is reported capable without + overriding anything and half a set counts as none. `lnpl serve` and the + outbox CLI commands now decide by these predicates (see Fixed). + `RepositoryDriverTCK` gains six cases for the two optional sets: a + driver reporting True must pass the claim/replay/release and drain/ack + behaviour checks, a driver reporting False must define none of the + set's methods, and neither branch skips. `docs/backends.md` gains the + optional-method capability table (issues #228, #229). - `scripts/load_probe.py` (stdlib open-loop load generator) and `docs/postgres-load-ceiling.md` (measured sustained-load ceiling and root cause for the postgres backend) — issue #180. diff --git a/README.ko.md b/README.ko.md index 2bb8ff2..871346f 100644 --- a/README.ko.md +++ b/README.ko.md @@ -198,7 +198,7 @@ workflow Login -> completed (33ms, correlation_id=cid-0001) - OpenAPI가 IR에서 생성되고, 골든 시나리오도 마찬가지다 — 손으로 유지하는 파일이 아니라 컴파일된다. 에이전트 9역할도 전부 구현됐다. -**테스트 5,330여 개 전부 통과**, 그리고 그 스위트가 실제로 실패할 수 있음을 증명하는 +**테스트 5,390여 개 전부 통과**, 그리고 그 스위트가 실제로 실패할 수 있음을 증명하는 77종 뮤테이션 하네스. 둘 다 [검증](#검증)의 명령으로 재현한다. **RFC 65편 — 46편 `Accepted`, RFC-0000은 RFC-0007로 `Superseded`, RFC-0034·RFC-0048·RFC-0049·RFC-0050·RFC-0051·RFC-0052·RFC-0053·RFC-0054·RFC-0055·RFC-0056·RFC-0057·RFC-0058·RFC-0059·RFC-0060·RFC-0061·RFC-0062·RFC-0063·RFC-0064는 `Draft`.** RFC-0007은 @@ -339,7 +339,7 @@ PYTHONPATH=impl .venv/bin/python -m unittest discover -s impl/tests -t impl ``` ``` -Ran 5330 tests in 163.572s +Ran 5394 tests in 154.320s OK ``` diff --git a/README.md b/README.md index 39e4cd6..617a2d1 100644 --- a/README.md +++ b/README.md @@ -244,7 +244,7 @@ All three roadmap phases are complete. - OpenAPI is generated from the IR, and so is the golden scenario — it is compiled, not hand-maintained. All nine agent roles are implemented. -**~5,330 tests, all passing**, plus a 77-mutation harness that proves the suite can +**~5,390 tests, all passing**, plus a 77-mutation harness that proves the suite can actually fail. Both are reproduced by the commands under [Verification](#verification). @@ -409,7 +409,7 @@ PYTHONPATH=impl .venv/bin/python -m unittest discover -s impl/tests -t impl ``` ``` -Ran 5330 tests in 163.572s +Ran 5394 tests in 154.320s OK ``` diff --git a/docs/backends.md b/docs/backends.md index a7eabcd..e55d2e4 100644 --- a/docs/backends.md +++ b/docs/backends.md @@ -261,6 +261,10 @@ CLI 계약(`plugins/lnpl/skills/lnpl-authoring/cli-surface.md`에 상세): 거부한다 — 나머지가 조용히 acked되어 "일부만 성공했다"를 호출자가 메시지 만으로 알 수 없게 되는 일은 없다. +`drain_outbox`/`ack_outbox`는 `RepositoryDriver`의 **선택** 메서드다. 외부 드라이버가 +이 둘을 갖췄는지는 `supports_outbox_drain()`이 알리며, 기능별 대응은 §8 +「선택 메서드와 능력 술어」 표에 있다. + **외부 릴레이가 소유하는 것.** 이 구현은 drain으로 읽고 ack로 지우는 것까지만 한다 — 실제로 카프카·SQS 등에 퍼블리시하는 폴링 퍼블리셔는 cron·systemd 타이머·k8s `CronJob` 같은 프로세스 밖 스케줄러가 소유하며, 그 루프는 @@ -459,6 +463,32 @@ if문이 하나도 없다. 보존). 이 모듈의 "ONE ERROR TYPE OUT" 규칙이 entry-points 경로에도 그대로 적용된다는 뜻이다. +### 선택 메서드와 능력 술어 (이슈 #228/#229) + +`RepositoryDriver`의 필수 메서드만 구현해도 `lnpl run`은 돈다. 그러나 serve와 CLI의 +일부 기능은 **선택 메서드** 묶음을 더 요구하고, 드라이버가 그 묶음을 갖췄는지는 기본 +클래스의 두 **능력 술어**가 알린다. 기본 클래스는 다섯 선택 메서드를 하나도 정의하지 +않는다. 술어의 기본 구현은 묶음의 메서드가 이 드라이버 인스턴스에서 전부 호출 가능한지를 +직접 확인하므로, 메서드를 구현한 외부 드라이버는 술어를 오버라이드하지 않아도 True로 +보고된다. 묶음 중 하나라도 빠지면 False다 — 반쪽 구현은 지원이 아니다. + +| 기능 | 필요한 선택 메서드 | 알리는 술어 | 술어가 False일 때 | +|---|---|---|---| +| `Idempotency-Key` 헤더 — 워크플로 요청 경로와 `POST /-/events/` 소비 경로 | `idempotency_begin` / `idempotency_finish` / `idempotency_release` | `supports_idempotency()` | serve가 요청을 실행하지 않고 501 `idempotency-unsupported`(problem+json)로 거부한다 (이슈 #228) | +| `lnpl outbox drain` / `lnpl outbox ack` | `drain_outbox` / `ack_outbox` | `supports_outbox_drain()` | CLI가 한 줄 오류와 rc 2로 끝낸다 (이슈 #229) | +| `lnpl relay` | `drain_outbox` / `ack_outbox` | `supports_outbox_drain()` | CLI가 한 줄 오류와 rc 2로 끝낸다 (이슈 #229) | +| SSE 구독 `GET //events/` | `read_outbox` | 없음 — `read_outbox`는 필수 기본 메서드다 | 해당 없음 | + +술어는 오버라이드하지 않는다. TCK가 술어의 답과 실제로 있는 메서드를 양쪽으로 맞춰 +보므로, 메서드 없이 True를 내거나 메서드를 둔 채 False를 내는 드라이버는 TCK에서 +실패한다(아래 「TCK로 검증하기」). + +현재 드라이버(2026-10-11 기준): 내장 `sqlite`는 두 술어 모두 True다. +[`lnpl-postgres`](https://github.com/choiyounggi/lnpl-postgres)는 다섯 선택 메서드가 +없어 둘 다 False이며, 구현은 lnpl-postgres#8의 몫이다. `fake`는 저장소 드라이버를 두지 +않으므로 술어를 묻지 않고 기존 동작을 그대로 유지한다 — 멱등성 클레임은 하지 않고, +`lnpl outbox`/`lnpl relay`는 기존대로 rc 2로 거부한다. + ### TCK로 검증하기 외부 드라이버는 `lnpl.testing.RepositoryDriverTCK`를 상속해 자기 CI에서 돌린다: @@ -496,6 +526,16 @@ class MyPostgresDriverTCKTest(RepositoryDriverTCK, unittest.TestCase): 드라이버는 이 TCK를 더 이상 통과하지 못한다.** `SqliteRepositoryDriver`가 이 TCK로 검증되는 예는 `impl/tests/test_driver_contract.py::SqliteDriverTCKTest`다. +**선택 메서드(이슈 #228/#229) — skip이 아니라 declared-unsupported.** TCK는 멱등성 +3케이스(열린 클레임에 다시 클레임하면 `in-progress`, `idempotency_finish` 뒤 클레임은 +저장된 결과를 재생, `idempotency_release` 뒤 클레임은 새로 `started`)와 drain/ack +3케이스(미전달 행을 `seq` 오름차순으로 나열, ack한 행은 다시 drain되지 않음, 같은 `seq` +재-ack는 멱등 성공)를 담는다. 각 케이스는 먼저 해당 술어를 묻는다. True면 동작을 +단언하고, False면 술어가 정말 `bool` False이고 그 묶음의 메서드가 하나도 없음을 단언한 +뒤 통과한다. 어느 쪽도 `skipTest`가 아니므로, 메서드 없이 True를 내거나 묶음의 일부만 +구현한 드라이버는 스킵 뒤에 숨지 못하고 실패한다. 다섯 메서드가 모두 없는 +드라이버(2026-10-11의 `lnpl-postgres`)는 여섯 케이스를 declared-unsupported로 통과한다. + ## 9. SPI: 외부 토큰 프로바이더 등록 (이슈 #119b) 이슈 #119가 지적한 대목: `security role `을 집행하는 역할 클레임이 diff --git a/docs/serving.md b/docs/serving.md index 0fffc64..e4d3bac 100644 --- a/docs/serving.md +++ b/docs/serving.md @@ -174,6 +174,7 @@ linkly가 따르는 규범 문서는 없다. | # | 관측 조건 | HTTP | error `code` | |---|-----------|------|--------------| | M17 | 같은 키로 이미 실행 중(다른 요청이 아직 안 끝남) | 409 | `idempotency-in-progress` | +| M22 | `Idempotency-Key` 헤더가 있고, 설정된 저장소 드라이버(`RepositoryDriver`)의 `supports_idempotency()`가 False — 클레임 메서드 3종(`idempotency_begin`/`idempotency_finish`/`idempotency_release`)을 구현하지 않음(현재 `lnpl-postgres`). 클레임도 워크플로 실행도 하지 않는다 | 501 | `idempotency-unsupported` | 재생(replay)은 새 판정 행이 아니다 — 저장된 첫 실행의 상태코드·바디를 그대로 돌려줄 뿐이다(200이었으면 200, 409 `conflict`였으면 409 @@ -199,10 +200,26 @@ linkly가 따르는 규범 문서는 없다. 복구 수단은 `--idempotency-ttl`(기본 24h)뿐이다 — TTL이 지나면 그 키는 새 미스로 취급된다. 이 창을 없는 척하지 않는다: 짧지만 실재한다. -`--backend fake`에서는 요청마다 빈 저장소가 새로 시딩되므로 클레임을 남길 -곳이 없다 — 이 기능은 **비활성**이고, 서버 기동 시 stderr에 경고를 한 번 -낸다. `Idempotency-Key` 헤더를 보내도 조용히 무시되고 매 요청이 그대로 -실행된다(이 백엔드에서 결과가 요청 간에 남지 않는 것과 같은 이유). +**백엔드별 규칙 (이슈 #228).** `Idempotency-Key`를 지킬 수 있는지는 저장소 +드라이버가 `RepositoryDriver.supports_idempotency()`로 선언한다 +(`docs/backends.md` §8 — 클레임 메서드 3종 `idempotency_begin`/ +`idempotency_finish`/`idempotency_release`를 모두 구현했을 때만 True). +지킬 수 없는 키를 조용히 무시해 같은 요청을 두 번 실행하는 대신, 명시적으로 +실패한다: + +| 백엔드 | `Idempotency-Key`가 오면 | 기동 시 stderr | +|--------|--------------------------|----------------| +| `fake` (`--backend fake`) | 비활성 — 요청마다 빈 저장소가 새로 시딩되어 클레임을 남길 곳이 없으므로 헤더를 무시하고 매 요청을 그대로 실행한다 | `lnpl serve: Idempotency-Key support is disabled ...` 한 줄(D11) | +| `sqlite:` (`supports_idempotency()` True) | 위 계약 그대로 — 클레임·재생·M17 | 없음 | +| 클레임 메서드 3종을 구현하지 않은 드라이버(`supports_idempotency()` False — 현재 `lnpl-postgres`, lnpl-postgres#8 전까지) | 거부 — 501 `idempotency-unsupported`(M22). 클레임도 워크플로 실행도 하지 않는다. 헤더 없는 요청은 그대로 실행된다 | `lnpl serve: repository driver -- idempotency: unsupported ...` 한 줄(아래 운영 표면) | + +501은 RFC 9110 §15.6.2의 뜻 그대로다 — 서버가 이 요청을 이행하는 데 필요한 +기능을 지원하지 않는다. Idempotency-Key draft는 키를 저장할 곳이 없는 서버의 +응답을 규정하지 않는다. 501 본문의 `detail`은 드라이버를 +`.`으로 이름 짓고 운영자가 할 일(구현한 드라이버 설치· +업그레이드, 또는 헤더 없이 재전송)을 적는다. 이벤트 소비 라우트 +(`POST /-/events/`)도 같은 규칙을 따른다 — 그 라우트는 CloudEvents +`id`가 곧 키이므로, 이런 드라이버에서는 모든 요청이 E8(501)이다. ## `ETag` / `If-Match` (이슈 #113) @@ -232,7 +249,8 @@ REST의 PUT/PATCH가 갖는 단일 대상 리소스가 없어서, 이전 GET의 | M20 | `If-Match`가 있고 첫 읽기의 조회 키를 이 요청만으로 미리 평가할 수 없음(참조가 바인딩·네트워크 결과를 가리키거나, 그 값이 이 요청에 없음) | 400 | `precondition-unsupported` | | M21 | `If-Match`가 있고 조회 키는 평가됐는데 그 키의 행이 없음 | 412 | `precondition-failed` | -평가 순서: 인증(M3/M3a/M3b, `_respond` 전에 이미 끝남) → M18(형식) → +평가 순서: 인증(M3/M3a/M3b, `_respond` 전에 이미 끝남) → M22(멱등성 미지원, +`Idempotency-Key`가 있을 때만) → 멱등성 클레임(M17/재생) → M18(형식) → M20(조회 키 미평가) → M21(행 없음) → M19(버전 불일치) → 워크플로 실행. M21은 `If-Match` 없는 같은 요청이 받는 404(M8b)보다 앞선다. @@ -307,6 +325,7 @@ dead-letter할지 기계로 판정해야 하는 대상이 "이 호출자가 뭘 | E5 | 실행 완료 | 200 | — | | E6 | 실행 실패, 데드라인 초과 또는 실패 스텝의 효과가 `RepositoryCall`/`NetworkCall`(`DriverError` 계열), 또는 낙관적 버전 쓰기 충돌(`failure_kind == "write-conflict"`, 이슈 #201 — 실패 스텝이 `set`이라 효과가 `Assignment`여도 여기다) — 일시적, 릴레이는 재시도해야 한다 | 503 + `Retry-After: 1` | `event-retry-later` | | E7 | 실행 실패, 그 외 전부(`Validation` 거부, 명시적 비즈니스/가드 `RunError`(저자가 선언한 `fail ` — `failure_kind == "rejected"`, RFC-0056), create 충돌, `not-found` — 영속 백엔드 읽기 미스(이슈 #197)나 0행 `update`/`delete`(이슈 #215), `id` 없는 create `id-required`(RFC-0055)) — 영구적, 같은 페이로드를 다시 돌려도 같은 결과다 | 422 | `event-rejected` | +| E8 | 저장소 드라이버(`RepositoryDriver`)의 `supports_idempotency()`가 False — CloudEvents `id`가 곧 멱등성 키이므로 이 라우트의 모든 요청이 해당한다. 봉투 검사(E1-E3) 다음, E4보다 먼저 판정하며 클레임도 워크플로 실행도 하지 않는다 | 501 | `idempotency-unsupported` | **멱등성 (D6)** — CloudEvents `id`가 멱등성 키다. `lnpl_idempotency`(이슈 #113)를 **그대로** 재사용한다 — 두 번째 저장소를 만들지 않는다. 200과 @@ -317,7 +336,11 @@ dead-letter할지 기계로 판정해야 하는 대상이 "이 호출자가 뭘 취지를 정면으로 깬다). 그래서 503 다음 같은 `id`가 즉시 재전달되면 "실행 중" 상태가 아직 안 지워졌으므로 E4(409)를 본다 — TTL이 지나야 완전히 새로 시작한다. `fake` 백엔드(멱등성 저장소 없음)는 매번 독립적으로 -실행한다(D9/D11 기존 규약과 동일). +실행한다(D9/D11 기존 규약과 동일). 클레임 메서드 3종을 구현하지 않은 +드라이버(`supports_idempotency()` False)에서는 봉투 검사 다음 E8(501 +`idempotency-unsupported`)로 거부하고 실행하지 않는다(이슈 #228). 레퍼런스 +릴레이 `lnpl relay`는 501을 예상 밖 상태로 보고 ack하지 않은 채 다음 drain에서 +다시 보낸다. **레퍼런스 릴레이 — `lnpl relay`.** `lnpl outbox drain`(발행)과 이 라우트 (소비)를 잇는 최소 구현. `--target http(s)://...`는 브로커 의존 없이 두 @@ -424,7 +447,9 @@ liveness가 재시작시키면 롤링 업데이트/드레이닝이 깨진다. (실패했다면 `ServeError`로 애초에 뜨지 못했으므로, 이 앱이 존재한다는 사실 자체가 통과의 증거다). 2. 영속 백엔드(`--backend sqlite:...`)가 설정돼 있으면 커넥션을 1회 - 획득·해제한다. + 획득·해제한다. 같은 커넥션에서 드라이버의 `supports_idempotency()`도 읽어 + 200 본문의 `capabilities`에 싣는다(이슈 #228) — 검사 항목이 아니라 + 데이터다. False여도 200이다. 3. `--jwt-secret-env`가 지정돼 있으면 그 환경변수가 **지금도** 설정돼 있는지 — 기동 시 검증(`cli.cmd_serve`)과 별개로, 매 프로브마다 다시 읽는다(프로세스가 떠 있는 동안 그 변수가 사라지는 드문 드리프트도 @@ -445,7 +470,7 @@ liveness가 재시작시키면 롤링 업데이트/드레이닝이 깨진다. 200으로 돌아온다. 본문에는 검사 이름만 실린다. SIGTERM은 이 다섯보다 **먼저** 본다 — 받는 즉시 나머지 검사 없이 503이다. -전부 통과하면 200 `{"status": "ok"}`; 하나라도 깨졌으면 503 + +전부 통과하면 200 `{"status": "ok", "capabilities": {"idempotency": }}`; 하나라도 깨졌으면 503 + `application/problem+json`(`code: "not-ready"`)에 **깨진 검사 이름**을 `checks`로 싣는다. 401/403(위 M3/M3a/M3b)과 반대 판단이다 — readyz는 운영자용이지 공격면이 아니므로, 어느 검사가 깨졌는지 감추지 않는다. @@ -458,6 +483,22 @@ SIGTERM은 이 다섯보다 **먼저** 본다 — 받는 즉시 나머지 검사 "checks": ["repository", "jwt-secret-env"]} ``` +**능력 노출과 기동 줄 (이슈 #228).** `capabilities.idempotency`는 저장소 +드라이버의 `supports_idempotency()`가 True일 때만 `true`다 — 영속 백엔드가 +없으면(`fake`) `false`. 능력이 없는 것은 기다리거나 재시작해도 생기지 않는 +정적 사실이라 readiness 실패가 아니다: 200은 그대로이고, 닫힌 목록 다섯은 +늘지 않으며(`capabilities`는 `checks` 이름이 아니다), 503 본문도 바뀌지 +않는다. 같은 사실을 기동 시에도 한 번 알린다 — `make_wsgi_app`(`lnpl serve`와 +`build_app()` 둘 다 거친다)이 저장소를 하나 열어 능력을 읽고 닫으며, 능력이 +없으면 stderr에 한 줄을 낸다: + +```text +lnpl serve: repository driver -- idempotency: unsupported (it does not implement idempotency_begin, idempotency_finish and idempotency_release); requests carrying Idempotency-Key are refused with 501 idempotency-unsupported +``` + +`fake`에는 이 줄이 없다(위 D11 경고가 이미 있다). gunicorn이 `--preload` +없이 워커마다 `build_app()`을 부르면 워커마다 한 줄씩 찍힌다. + **liveness와 readiness를 절대 섞지 않는다.** 재시작이 답인 실패(프로세스가 망가짐)는 healthz로, 트래픽만 끊으면 되는 실패(백엔드/설정 드리프트)는 readyz로 간다 — 둘을 하나로 합치면 트래픽 차단이면 충분한 상황에서 diff --git a/impl/lnpl/cli.py b/impl/lnpl/cli.py index 9887428..5433c45 100644 --- a/impl/lnpl/cli.py +++ b/impl/lnpl/cli.py @@ -901,6 +901,12 @@ def cmd_outbox_drain(args): "(e.g. sqlite:./store.db) — `fake` has no outbox to drain", file=sys.stderr) return 2 + if not repository.supports_outbox_drain(): + print("error: backend %s does not implement the outbox drain " + "contract (drain_outbox/ack_outbox) — see docs/backends.md §3" + % args.backend.split(":", 1)[0], file=sys.stderr) + repository.close() + return 2 try: for emission in repository.drain_outbox(limit=args.limit): sys.stdout.write(json.dumps(emission, ensure_ascii=False) + "\n") @@ -930,6 +936,12 @@ def cmd_outbox_ack(args): "(e.g. sqlite:./store.db) — `fake` has no outbox to ack", file=sys.stderr) return 2 + if not repository.supports_outbox_drain(): + print("error: backend %s does not implement the outbox drain " + "contract (drain_outbox/ack_outbox) — see docs/backends.md §3" + % args.backend.split(":", 1)[0], file=sys.stderr) + repository.close() + return 2 try: repository.ack_outbox(args.seq) return 0 @@ -1091,6 +1103,12 @@ def cmd_relay(args): "(e.g. sqlite:./store.db) — `fake` has no outbox to drain", file=sys.stderr) return 2 + if not repository.supports_outbox_drain(): + print("error: backend %s does not implement the outbox drain " + "contract (drain_outbox/ack_outbox) — see docs/backends.md §3" + % args.backend.split(":", 1)[0], file=sys.stderr) + repository.close() + return 2 try: publisher = open_publisher(args.target) except (ValueError, DriverError) as exc: diff --git a/impl/lnpl/drivers.py b/impl/lnpl/drivers.py index f98c180..4b94336 100644 --- a/impl/lnpl/drivers.py +++ b/impl/lnpl/drivers.py @@ -176,6 +176,14 @@ class PublishRejected(DriverError): # The contracts # -------------------------------------------------------------------------- +# issue #228/#229: the optional method sets `RepositoryDriver`'s two +# capability predicates look for. Neither set has a base-class definition +# (blackboard ruling) — a driver offers a set by defining every method in it. +IDEMPOTENCY_METHODS = ("idempotency_begin", "idempotency_finish", + "idempotency_release") +OUTBOX_DRAIN_METHODS = ("drain_outbox", "ack_outbox") + + class RepositoryDriver: """The `postgres` capability's adapter contract. @@ -195,6 +203,13 @@ class RepositoryDriver: repository driver, `delivery`) from the closed axis table in `capabilities.py`. Absent by default: no report, no diagnostic (`docs/backends.md` §8). + + `supports_idempotency()` / `supports_outbox_drain()` (issue #228/#229) + — methods, not attributes: each reports whether this driver implements + a whole optional method set (`IDEMPOTENCY_METHODS` / + `OUTBOX_DRAIN_METHODS`). The base class defines none of those five + methods; the predicates' default reads the instance, so an + implementation needs no override (`docs/backends.md` §8). """ def seed(self, rows): @@ -316,6 +331,44 @@ def read_outbox(self, event, after_seq=0): """ raise NotImplementedError + def supports_idempotency(self): + """True when this driver implements every idempotency-claim method (issue #228). + + The set is `IDEMPOTENCY_METHODS`: `idempotency_begin`, + `idempotency_finish`, `idempotency_release` (signatures and return + shapes: `SqliteRepositoryDriver`, the reference implementation). + `lnpl serve` needs all three to honour an `Idempotency-Key` header. + + Default: introspection — True only when every method of the set is + callable on this instance, so an external driver that implements + the three is reported capable without overriding this predicate, + and a driver missing any one of them is not (half a set is no + support). Do not override it: `testing.RepositoryDriverTCK` checks + the answer against the methods actually present, in both + directions. `RepositoryDriver` itself defines none of the three — a + base-class stub would make this default report True for every + driver. + """ + return all(callable(getattr(self, name, None)) + for name in IDEMPOTENCY_METHODS) + + def supports_outbox_drain(self): + """True when this driver implements both outbox drain/ack methods (issue #229). + + The set is `OUTBOX_DRAIN_METHODS`: `drain_outbox` and `ack_outbox` + (issue #102 contract: `SqliteRepositoryDriver`'s docstrings). + `lnpl outbox drain` / `lnpl outbox ack` and `lnpl relay` need both. + `read_outbox` (the SSE tail read) is NOT in the set — it is a + required method of this base class. + + Default: introspection, the same rule as `supports_idempotency` — + True only when both are callable on this instance. Do not override + it (`testing.RepositoryDriverTCK` cross-checks it); `RepositoryDriver` + itself defines neither method. + """ + return all(callable(getattr(self, name, None)) + for name in OUTBOX_DRAIN_METHODS) + def close(self): """Release resources. Safe to call more than once.""" raise NotImplementedError diff --git a/impl/lnpl/lexer.py b/impl/lnpl/lexer.py index d110e0d..4223deb 100644 --- a/impl/lnpl/lexer.py +++ b/impl/lnpl/lexer.py @@ -124,7 +124,8 @@ "auth-invalid", "auth-missing", "body-too-large", "body-unreadable", "cloudevents-invalid", "conflict", "cursor-invalid", "deadline-exceeded", "event-rejected", "event-retry-later", - "forbidden", "id-required", "idempotency-in-progress", "limit-invalid", + "forbidden", "id-required", "idempotency-in-progress", + "idempotency-unsupported", "limit-invalid", "method-not-allowed", "not-found", "not-ready", "precondition-failed", "precondition-invalid", "precondition-unsupported", "rate-limited", "read-failed", "shutting-down", "validation-failed", diff --git a/impl/lnpl/testing.py b/impl/lnpl/testing.py index f8c19bd..32e0161 100644 --- a/impl/lnpl/testing.py +++ b/impl/lnpl/testing.py @@ -57,6 +57,16 @@ def make_driver(self): skips the conflict test for a driver that does not offer it, the same way a JDBC TCK's optional-feature tests skip against a driver that reports the feature unsupported. + +The optional method sets (issue #228/#229) — `idempotency_begin`/ +`idempotency_finish`/`idempotency_release` and `drain_outbox`/`ack_outbox` +— are deliberately NOT handled that way. Each of their cases first asks the +driver's capability predicate (`supports_idempotency()` / +`supports_outbox_drain()`): True runs the behaviour assertions, False +asserts the predicate is a real bool and that no method of the set exists +on the driver, then passes. Neither branch skips, so a driver that claims a +set it lacks, or implements half a set, fails instead of hiding behind a +skip. """ import base64 @@ -64,7 +74,8 @@ def make_driver(self): import os from lnpl.cli import _relay_drain_once_via_publisher -from lnpl.drivers import (ConflictError, DriverError, TokenError, +from lnpl.drivers import (IDEMPOTENCY_METHODS, OUTBOX_DRAIN_METHODS, + ConflictError, DriverError, TokenError, WriteConflictError) from lnpl.generators import GeneratorError, run_generator @@ -496,6 +507,126 @@ def test_a_write_from_another_run_before_this_runs_update_still_conflicts(self): self.driver.persist("widget", "w-x2", row) self.assertIn("conflict", str(caught.exception)) + # -- optional method sets (issue #228/#229) ----------------------------- + # + # Declared-unsupported, never skipped (contrast + # `_skip_unless_predicate_supported` above). Each case asks the + # capability predicate first: True -> the behaviour must hold, so a + # driver claiming a set it lacks errors here; False -> the predicate + # must be a real bool and NO method of that set may be callable on the + # driver, and the case passes having asserted exactly that. + + _TCK_NOW_MS = 1000 + _TCK_TTL_MS = 86400000 + + def _assert_declares_none_of(self, predicate_name, method_names): + for name in method_names: + self.assertFalse( + callable(getattr(self.driver, name, None)), + "%s() reports False but the driver implements %s — " + "implement all of %s, or none" + % (predicate_name, name, ", ".join(method_names))) + + def _declares_idempotency(self): + supported = self.driver.supports_idempotency() + self.assertIsInstance(supported, bool) + if not supported: + self._assert_declares_none_of("supports_idempotency", + IDEMPOTENCY_METHODS) + return supported + + def _declares_outbox_drain(self): + supported = self.driver.supports_outbox_drain() + self.assertIsInstance(supported, bool) + if not supported: + self._assert_declares_none_of("supports_outbox_drain", + OUTBOX_DRAIN_METHODS) + return supported + + def _claim(self, key): + return self.driver.idempotency_begin( + "wf.tck", key, self._TCK_NOW_MS, self._TCK_TTL_MS) + + def _record_tck_emission(self, n): + self.driver.record_emission({ + "emission_id": "tck.emit#%d" % n, + "event": "entity.event.tck", + "payload": {"x": n}, + }) + + def test_an_open_idempotency_claim_answers_in_progress_to_a_second_claim(self): + if not self._declares_idempotency(): + return + + self.assertEqual(self._claim("k1"), ("started", None)) + self.assertEqual(self._claim("k1"), ("in-progress", None)) + + def test_a_finished_idempotency_claim_replays_its_stored_result(self): + if not self._declares_idempotency(): + return + self.assertEqual(self._claim("k2"), ("started", None)) + + self.driver.idempotency_finish("wf.tck", "k2", 201, {"id": "x"}) + + status, stored = self._claim("k2") + self.assertEqual(status, "done") + self.assertEqual(tuple(stored), (201, {"id": "x"})) + + def test_a_released_idempotency_claim_can_be_claimed_afresh(self): + if not self._declares_idempotency(): + return + self.assertEqual(self._claim("k3"), ("started", None)) + + self.driver.idempotency_release("wf.tck", "k3") + + self.assertEqual(self._claim("k3"), ("started", None)) + + def test_drain_outbox_lists_every_undelivered_emission_in_seq_order(self): + if not self._declares_outbox_drain(): + return + self._record_tck_emission(1) + self._record_tck_emission(2) + + rows = self.driver.drain_outbox() + + self.assertEqual(len(rows), 2) + for row in rows: + self.assertEqual(set(row), {"seq", "emission_id", "event", + "payload", "created_at"}) + self.assertEqual([row["emission_id"] for row in rows], + ["tck.emit#1", "tck.emit#2"]) + self.assertLess(rows[0]["seq"], rows[1]["seq"]) + self.assertEqual([row["payload"] for row in rows], + [{"x": 1}, {"x": 2}]) + self.assertEqual([row["event"] for row in rows], + ["entity.event.tck", "entity.event.tck"]) + + def test_an_acked_emission_is_never_drained_again(self): + if not self._declares_outbox_drain(): + return + self._record_tck_emission(1) + self._record_tck_emission(2) + first, second = self.driver.drain_outbox() + + self.driver.ack_outbox([first["seq"]]) + + self.assertEqual([row["emission_id"] + for row in self.driver.drain_outbox()], + ["tck.emit#2"]) + self.driver.ack_outbox([second["seq"]]) + self.assertEqual(self.driver.drain_outbox(), []) + + def test_re_acking_a_delivered_seq_is_an_idempotent_success(self): + if not self._declares_outbox_drain(): + return + self._record_tck_emission(1) + (row,) = self.driver.drain_outbox() + self.driver.ack_outbox([row["seq"]]) + + self.driver.ack_outbox([row["seq"]]) + + self.assertEqual(self.driver.drain_outbox(), []) + class CacheDriverTCK: """Mix into a `unittest.TestCase` subclass, override `make_cache()` and diff --git a/impl/lnpl/wsgi.py b/impl/lnpl/wsgi.py index 7ff4b8b..f8556ec 100644 --- a/impl/lnpl/wsgi.py +++ b/impl/lnpl/wsgi.py @@ -37,7 +37,7 @@ from importlib import metadata as importlib_metadata from .drivers import (DriverError, HmacTokenProvider, HttpNetworkDriver, - RotatingHmacTokenProvider, TokenError, + RepositoryDriver, RotatingHmacTokenProvider, TokenError, _http_capabilities, _is_url_literal, audience_for_path, open_cache, open_network, open_repository, open_secret_provider, open_token_provider) @@ -866,6 +866,7 @@ def map_consume_result(result): "conflict": "the request conflicts with the current state of the target resource", "write-conflict": "a concurrent write changed the row since it was read -- retry is safe", "idempotency-in-progress": "a request with this Idempotency-Key is already running", + "idempotency-unsupported": "this server's repository driver cannot honour an idempotency key", "precondition-failed": "the If-Match version no longer matches the stored row", "precondition-invalid": "the If-Match header value is not a recognized ETag", "precondition-unsupported": "this workflow's first read cannot be evaluated against If-Match before it runs", @@ -992,6 +993,37 @@ def _read_json_body(environ, start_response): return payload, None +def _idempotency_mode(repo): + """issue #228: what an idempotency key may do against `repo`. + + A `RepositoryDriver` instance -- the only thing `open_repository` returns + (pinned by test_serve_backend.OpenRepositoryReturnsDriversTest): + "claim" -- `supports_idempotency()` is True: claim/replay (#113). + "refuse" -- it is False: 501 `idempotency-unsupported`, boot line, + readyz `capabilities.idempotency` False. + Anything else -- LEGACY branch, coordinator ruling t2 D1 r2 + (.orchestration/notes/decisions.md:15): + `None` (the `fake` backend) and test doubles such as + `interp.FakeRepository` keep the exact pre-#228 opt-in -- claim iff the + object has `idempotency_begin`; never 501, no boot line, readyz omits + `capabilities`. A plain class that defines its own + `supports_idempotency` is still legacy: the type decides, not the + method (no getattr fallback, ruling S2). + "legacy-claim" -- has `idempotency_begin`. + "legacy-skip" -- does not (includes `None`). + """ + if isinstance(repo, RepositoryDriver): + return "claim" if repo.supports_idempotency() else "refuse" + return "legacy-claim" if hasattr(repo, "idempotency_begin") else "legacy-skip" + + +def _driver_name(repository): + """issue #228: the driver as the operator configured it -- no backend + scheme reaches this module on the `lnpl serve` path (cli.py passes a bare + factory), so the class path is the identifier both entry points share.""" + return type(repository).__module__ + "." + type(repository).__qualname__ + + class LnplWsgiApp: """A PEP-3333 callable bound to one compiled document + routing table. @@ -1101,6 +1133,30 @@ def __init__(self, document, routes, repository_factory=None, "so there is nowhere to durably record a claim (issue " "#113, D11). Use --backend sqlite: to enable it.", file=sys.stderr) + else: + # issue #228: one repository, opened once here, says whether + # this driver can honour Idempotency-Key. Both entry points + # (`serve()`, `build_app()`) construct through here. An + # unreachable store prints nothing: production probes the + # backend before this point, and readyz reports it. + try: + probe = repository_factory() + except DriverError: + probe = None + if probe is not None: + try: + mode = _idempotency_mode(probe) + finally: + probe.close() + if mode == "refuse": + print( + "lnpl serve: repository driver %s -- idempotency: " + "unsupported (it does not implement " + "idempotency_begin, idempotency_finish and " + "idempotency_release); requests carrying " + "Idempotency-Key are refused with 501 " + "idempotency-unsupported" % _driver_name(probe), + file=sys.stderr) def _resolve_trace_context(self, environ): """issue #107: decide this request's `(trace_id, span_id, link, @@ -1711,17 +1767,23 @@ def _readyz(self, start_response): "the server received SIGTERM and is " "shutting down", checks=["shutting-down"])) - broken = self._readyz_broken() + capabilities = {} + broken = self._readyz_broken(capabilities) if broken: return _json_response(start_response, 503, problem(503, "not-ready", "readiness check(s) failed: %s" % ", ".join(broken), checks=broken)) - return _json_response(start_response, 200, {"status": "ok"}, + body = {"status": "ok"} + if capabilities: + # issue #228: data, not a check -- the closed list of five + # (issue #110 D4) is unchanged and never names it. + body["capabilities"] = capabilities + return _json_response(start_response, 200, body, content_type="application/json") - def _readyz_broken(self): + def _readyz_broken(self, capabilities): """issue #110, D4's closed list of five, by name — extending this list is a decision (D4 says so explicitly), not a one-line patch: @@ -1743,6 +1805,12 @@ def _readyz_broken(self): (review C1: readyz is unauthenticated); failure keeps the last good keys and reports `secret-provider`. + Check 2 also fills `capabilities` (issue #228) from the repository it + already opened -- `{"idempotency": bool}` for a RepositoryDriver, + nothing for a legacy test double -- and `{"idempotency": False}` when + no backend is configured. Data for the 200 body, never a broken-check + name. + Returns the broken ones' names, in the order above; empty means ready. """ @@ -1753,7 +1821,12 @@ def _readyz_broken(self): except DriverError: broken.append("repository") else: + mode = _idempotency_mode(repository) + if mode in ("claim", "refuse"): + capabilities["idempotency"] = mode == "claim" repository.close() + else: + capabilities["idempotency"] = False if self.jwt_secret_env and not os.environ.get(self.jwt_secret_env): broken.append("jwt-secret-env") if (isinstance(self.token_provider, RotatingHmacTokenProvider) @@ -1953,12 +2026,22 @@ def _check_if_match(self, doc, workflow_id, payload, repository, if_match, def _respond(self, environ, start_response, doc, workflow_id, payload, correlation_id, repository, claims=None, log_sink=None, trace_ctx=None, idempotency_key=None, if_match=None): - # issue #113, D8/D9/D11: opt-in on the repository object, the same - # `getattr` idiom D12's ETag opt-in uses -- covers "no backend at - # all" (fake -> `repository is None`) and "a backend that never - # implemented it" in one check, no special-casing either. - claim = (idempotency_key is not None and repository is not None - and hasattr(repository, "idempotency_begin")) + # issue #228: the driver declares the capability; `_idempotency_mode` + # (ruling t2 D1 r2) keeps the pre-#228 opt-in for non-driver objects. + mode = (_idempotency_mode(repository) + if idempotency_key is not None else None) + if mode == "refuse": + return _json_response( + start_response, 501, + problem(501, "idempotency-unsupported", + "repository driver %s does not support Idempotency-Key " + "(it does not implement idempotency_begin, " + "idempotency_finish and idempotency_release); install " + "or upgrade to a driver that implements them, or " + "resend the request without the Idempotency-Key header" + % _driver_name(repository), + correlation_id=correlation_id)) + claim = mode in ("claim", "legacy-claim") if claim: now_ms = int(time.time() * 1000) claim_status, stored = repository.idempotency_begin( @@ -2114,11 +2197,21 @@ def _respond_event_consume(self, start_response, route, payload, claims=None): factory = self.repository_factory repository = factory() if factory is not None else None try: - # issue #113, D8/D9's same opt-in `getattr` idiom `_respond` - # uses: covers "no backend" (fake) and "a backend that never - # implemented it" in one check. - claim = (repository is not None - and hasattr(repository, "idempotency_begin")) + # issue #228: same rule as `_respond`; the CloudEvents id is the + # key, so a driver without the claim set refuses every event. + mode = _idempotency_mode(repository) + if mode == "refuse": + return _json_response( + start_response, 501, + problem(501, "idempotency-unsupported", + "repository driver %s does not support idempotency " + "(it does not implement idempotency_begin, " + "idempotency_finish and idempotency_release), and " + "this route uses the CloudEvents id as its " + "idempotency key; install or upgrade to a driver " + "that implements them" % _driver_name(repository), + correlation_id=correlation_id)) + claim = mode in ("claim", "legacy-claim") if claim: now_ms = int(time.time() * 1000) claim_status, stored = repository.idempotency_begin( diff --git a/impl/tests/test_cli_outbox.py b/impl/tests/test_cli_outbox.py index edde44f..e4a9194 100644 --- a/impl/tests/test_cli_outbox.py +++ b/impl/tests/test_cli_outbox.py @@ -22,9 +22,14 @@ import json import os import tempfile +import types import unittest +from importlib import metadata as importlib_metadata +from unittest import mock from lnpl import cli +from lnpl import drivers as drivers_module +from tests.driver_spi_fixture import DemoRepositoryDriver EMIT_SRC = """entity Order field @@ -38,6 +43,51 @@ emit orderPlaced """ +# issue #229: the one line every outbox command prints when its --backend +# driver lacks drain_outbox/ack_outbox (supports_outbox_drain() is False). +NO_DRAIN_CONTRACT = ("error: backend %s does not implement the outbox drain " + "contract (drain_outbox/ack_outbox) — see " + "docs/backends.md §3\n") + +DEMO_DRIVER = importlib_metadata.EntryPoint( + name="demo", value="tests.driver_spi_fixture:make_demo_driver", + group=drivers_module.DRIVERS_ENTRY_POINT_GROUP) + + +def _drivers_registered(*entry_points): + """Patch `lnpl.drivers` discovery to exactly `entry_points` (the + `test_driver_spi.py` idiom).""" + return mock.patch.object(drivers_module.importlib_metadata, + "entry_points", + lambda **_kw: list(entry_points)) + + +def _inline_entry_point(name, driver): + """An `lnpl.drivers` entry point whose factory returns `driver`, so the + test can inspect the very instance the CLI opened.""" + return types.SimpleNamespace( + name=name, value="inline", + group=drivers_module.DRIVERS_ENTRY_POINT_GROUP, + load=lambda: (lambda _arg: driver)) + + +class _DrainOnlyDriver(DemoRepositoryDriver): + """Half of the pair: `drain_outbox` without `ack_outbox`, so + `supports_outbox_drain()` is False (issue #229, boundary).""" + + def __init__(self, arg=None): + super().__init__(arg) + self.calls = [] + self.closed = False + + def drain_outbox(self, limit=None): + self.calls.append(limit) + return [{"seq": 1, "emission_id": "e1", "event": "x", + "payload": {}, "created_at": 0}] + + def close(self): + self.closed = True + class OutboxCliTestCase(unittest.TestCase): @@ -218,5 +268,71 @@ def test_ack_against_fake_is_rejected(self): self.assertIn("fake", err) +class UnsupportedDriverGuardTest(OutboxCliTestCase): + """Issue #229: a real driver without the drain/ack pair is refused + with one stderr line and rc 2 before anything is drained or acked.""" + + def test_drain_against_a_driver_without_the_contract_exits_2_with_one_line(self): + with _drivers_registered(DEMO_DRIVER): + rc, out, err = self.run_cli( + ["outbox", "drain", "--backend", "demo:x"]) + + self.assertEqual(2, rc) + self.assertEqual(NO_DRAIN_CONTRACT % "demo", err) + self.assertEqual("", out) + self.assertNotIn("Traceback", err) + + def test_ack_against_a_driver_without_the_contract_exits_2_with_one_line(self): + with _drivers_registered(DEMO_DRIVER): + rc, out, err = self.run_cli( + ["outbox", "ack", "--backend", "demo:x", "1"]) + + self.assertEqual(2, rc) + self.assertEqual(NO_DRAIN_CONTRACT % "demo", err) + self.assertEqual("", out) + self.assertNotIn("Traceback", err) + + def test_a_driver_with_drain_but_no_ack_is_refused_before_draining(self): + driver = _DrainOnlyDriver("x") + with _drivers_registered(_inline_entry_point("half", driver)): + rc, out, err = self.run_cli( + ["outbox", "drain", "--backend", "half:x"]) + + self.assertEqual(2, rc) + self.assertEqual(NO_DRAIN_CONTRACT % "half", err) + self.assertEqual("", out) + self.assertEqual([], driver.calls) + self.assertTrue(driver.closed) + + def test_fake_backend_keeps_its_own_refusal(self): + for argv in (["outbox", "drain", "--backend", "fake"], + ["outbox", "ack", "--backend", "fake", "1"]): + with self.subTest(argv=argv): + rc, _out, err = self.run_cli(argv) + + self.assertEqual(2, rc) + self.assertIn("persistent --backend", err) + self.assertNotIn("does not implement", err) + + def test_a_supporting_driver_is_not_refused(self): + self.emit_one() + + rc, out, err = self.run_cli( + ["outbox", "drain", "--backend", "sqlite:" + self.db]) + + self.assertEqual(0, rc, err) + self.assertEqual("", err) + self.assertEqual(1, len(out.splitlines())) + + def test_an_unknown_seq_on_a_supporting_driver_maps_to_rc_1(self): + self.emit_one() + + rc, _out, err = self.run_cli( + ["outbox", "ack", "--backend", "sqlite:" + self.db, "999999"]) + + self.assertEqual(1, rc) + self.assertIn("999999", err) + + if __name__ == "__main__": unittest.main() diff --git a/impl/tests/test_cli_relay.py b/impl/tests/test_cli_relay.py index 783218c..bd82878 100644 --- a/impl/tests/test_cli_relay.py +++ b/impl/tests/test_cli_relay.py @@ -19,6 +19,7 @@ import os import tempfile import threading +import types import unittest from importlib import metadata as importlib_metadata from unittest import mock @@ -29,6 +30,7 @@ from lnpl.lower import lower from lnpl.parser import parse from lnpl.wsgi import make_wsgi_app +from tests.driver_spi_fixture import DemoRepositoryDriver PUBLISH_SRC = """entity Order field @@ -257,6 +259,51 @@ def _publishers_registered(*entry_points): name="demo", value="tests.publisher_spi_fixture:make_demo_publisher", group=drivers_module.PUBLISHERS_ENTRY_POINT_GROUP) +# issue #229: the one line every outbox command prints when its --backend +# driver lacks drain_outbox/ack_outbox (supports_outbox_drain() is False). +NO_DRAIN_CONTRACT = ("error: backend %s does not implement the outbox drain " + "contract (drain_outbox/ack_outbox) — see " + "docs/backends.md §3\n") + +DEMO_DRIVER = importlib_metadata.EntryPoint( + name="demo", value="tests.driver_spi_fixture:make_demo_driver", + group=drivers_module.DRIVERS_ENTRY_POINT_GROUP) + + +def _drivers_registered(*entry_points): + """Patch `lnpl.drivers` discovery to exactly `entry_points` (the + `test_driver_spi.py` idiom).""" + return mock.patch.object(drivers_module.importlib_metadata, + "entry_points", + lambda **_kw: list(entry_points)) + + +def _inline_entry_point(name, driver): + """An `lnpl.drivers` entry point whose factory returns `driver`, so the + test can inspect the very instance the CLI opened.""" + return types.SimpleNamespace( + name=name, value="inline", + group=drivers_module.DRIVERS_ENTRY_POINT_GROUP, + load=lambda: (lambda _arg: driver)) + + +class _DrainOnlyDriver(DemoRepositoryDriver): + """Half of the pair: `drain_outbox` without `ack_outbox`, so + `supports_outbox_drain()` is False (issue #229, boundary).""" + + def __init__(self, arg=None): + super().__init__(arg) + self.calls = [] + self.closed = False + + def drain_outbox(self, limit=None): + self.calls.append(limit) + return [{"seq": 1, "emission_id": "e1", "event": "x", + "payload": {}, "created_at": 0}] + + def close(self): + self.closed = True + class RegisteredPublisherSchemeTest(RelayCliTestCase): """R7/R9: a registered scheme publishes through the matched @@ -355,5 +402,34 @@ def test_an_unloadable_registered_scheme_is_rejected_cleanly(self): self.assertEqual(1, len(self.drain())) +class UnsupportedDriverRelayTest(RelayCliTestCase): + """Issue #229: relay refuses a driver without the drain/ack pair + before opening the publisher and before the first drain.""" + + def test_relay_against_a_driver_without_the_contract_exits_2_with_one_line(self): + with _drivers_registered(DEMO_DRIVER): + rc, out, err = self.run_cli( + ["relay", self.publish_source, "--backend", "demo:x", + "--target", "http://127.0.0.1:1", "--once"]) + + self.assertEqual(2, rc) + self.assertNotIn("Traceback", err) + self.assertEqual(NO_DRAIN_CONTRACT % "demo", err) + self.assertEqual("", out) + + def test_relay_refuses_a_drain_only_driver_before_any_drain(self): + driver = _DrainOnlyDriver("x") + with _drivers_registered(_inline_entry_point("half", driver)): + rc, out, err = self.run_cli( + ["relay", self.publish_source, "--backend", "half:x", + "--target", "http://127.0.0.1:1", "--once"]) + + self.assertEqual(2, rc) + self.assertEqual(NO_DRAIN_CONTRACT % "half", err) + self.assertEqual("", out) + self.assertEqual([], driver.calls) + self.assertTrue(driver.closed) + + if __name__ == "__main__": unittest.main() diff --git a/impl/tests/test_driver_contract.py b/impl/tests/test_driver_contract.py index a32b056..27b742e 100644 --- a/impl/tests/test_driver_contract.py +++ b/impl/tests/test_driver_contract.py @@ -17,7 +17,9 @@ import tempfile import unittest -from lnpl.drivers import DriverError, SqliteRepositoryDriver +from lnpl.drivers import (IDEMPOTENCY_METHODS, OUTBOX_DRAIN_METHODS, + DriverError, RepositoryDriver, + SqliteRepositoryDriver) from lnpl.interp import (MASK, Clock, FakeCache, FakeRepository, Interpreter, RunError) from lnpl.lower import lower @@ -25,6 +27,7 @@ from lnpl.repo_policy import default_rows from lnpl.testing import RepositoryDriverTCK +from tests.driver_spi_fixture import DemoRepositoryDriver from tests.fixtures import GUARDED, SECRET_ACCOUNT, VALUE_INVENTORY READ_ONLY = """capability postgres @@ -904,5 +907,254 @@ def test_the_positional_prefix_is_unchanged(self): "correlation_id"]) +class _ExternalLikeDriver(RepositoryDriver): + """R3: an external driver written against the five methods alone — + it overrides neither capability predicate.""" + + def idempotency_begin(self, workflow_id, key, now_ms, ttl_ms): + return ("started", None) + + def idempotency_finish(self, workflow_id, key, http_status, body): + return None + + def idempotency_release(self, workflow_id, key): + return None + + def drain_outbox(self, limit=None): + return [] + + def ack_outbox(self, seqs): + return None + + +class _DrainAckOnlyDriver(RepositoryDriver): + def drain_outbox(self, limit=None): + return [] + + def ack_outbox(self, seqs): + return None + + +class _TwoOfThreeIdempotencyDriver(RepositoryDriver): + def idempotency_begin(self, workflow_id, key, now_ms, ttl_ms): + return ("started", None) + + def idempotency_finish(self, workflow_id, key, http_status, body): + return None + + +class _DrainWithoutAckDriver(RepositoryDriver): + def drain_outbox(self, limit=None): + return [] + + +class _SqliteWithoutRelease(SqliteRepositoryDriver): + """Boundary: a test double withdrawing one inherited method by setting + it to None — the predicate counts callables, not names.""" + + idempotency_release = None + + +class DriverCapabilityPredicateTest(unittest.TestCase): + """Issue #228/#229, design D1/D8: the two predicates read the method + set the instance really has — no override needed, a partial set is + not support.""" + + def _sqlite(self, driver_class): + box = tempfile.TemporaryDirectory() + self.addCleanup(box.cleanup) + driver = driver_class(os.path.join(box.name, "capability.db")) + self.addCleanup(driver.close) + return driver + + def test_the_sqlite_driver_reports_both_capabilities(self): + driver = self._sqlite(SqliteRepositoryDriver) + + self.assertIs(driver.supports_idempotency(), True) + self.assertIs(driver.supports_outbox_drain(), True) + + def test_an_external_driver_with_all_five_methods_is_capable_without_overriding(self): + driver = _ExternalLikeDriver() + + self.assertNotIn("supports_idempotency", vars(_ExternalLikeDriver)) + self.assertNotIn("supports_outbox_drain", vars(_ExternalLikeDriver)) + self.assertIs(driver.supports_idempotency(), True) + self.assertIs(driver.supports_outbox_drain(), True) + + def test_the_drain_ack_pair_alone_reports_outbox_drain_only(self): + driver = _DrainAckOnlyDriver() + + self.assertIs(driver.supports_outbox_drain(), True) + self.assertIs(driver.supports_idempotency(), False) + + def test_the_base_class_reports_neither_and_defines_none_of_the_five(self): + driver = RepositoryDriver() + + self.assertEqual(IDEMPOTENCY_METHODS, ("idempotency_begin", + "idempotency_finish", + "idempotency_release")) + self.assertEqual(OUTBOX_DRAIN_METHODS, ("drain_outbox", "ack_outbox")) + self.assertIs(driver.supports_idempotency(), False) + self.assertIs(driver.supports_outbox_drain(), False) + for name in IDEMPOTENCY_METHODS + OUTBOX_DRAIN_METHODS: + self.assertFalse(hasattr(driver, name), name) + + def test_two_of_the_three_idempotency_methods_is_not_support(self): + self.assertIs(_TwoOfThreeIdempotencyDriver().supports_idempotency(), + False) + + def test_drain_without_ack_is_not_outbox_drain_support(self): + self.assertIs(_DrainWithoutAckDriver().supports_outbox_drain(), False) + + def test_an_inherited_method_set_to_none_withdraws_only_its_own_set(self): + driver = self._sqlite(_SqliteWithoutRelease) + + self.assertIs(driver.supports_idempotency(), False) + self.assertIs(driver.supports_outbox_drain(), True) + + +# issue #228/#229: the six optional-method TCK cases, in the order they +# are defined in `lnpl.testing.RepositoryDriverTCK`. test_driver_spi.py +# imports this tuple for the demo driver's run. +CAPABILITY_TCK_CASES = ( + "test_an_open_idempotency_claim_answers_in_progress_to_a_second_claim", + "test_a_finished_idempotency_claim_replays_its_stored_result", + "test_a_released_idempotency_claim_can_be_claimed_afresh", + "test_drain_outbox_lists_every_undelivered_emission_in_seq_order", + "test_an_acked_emission_is_never_drained_again", + "test_re_acking_a_delivered_seq_is_an_idempotent_success", +) + + +class _ClaimAlwaysStartsDriver(SqliteRepositoryDriver): + """Negative control: a claim that never sees an open claim.""" + + def idempotency_begin(self, workflow_id, key, now_ms, ttl_ms): + return ("started", None) + + +class _FinishForgetsDriver(SqliteRepositoryDriver): + """Negative control: finish records nothing, so nothing replays.""" + + def idempotency_finish(self, workflow_id, key, http_status, body): + return None + + +class _ReleaseKeepsClaimDriver(SqliteRepositoryDriver): + """Negative control: release leaves the claim in-progress.""" + + def idempotency_release(self, workflow_id, key): + return None + + +class _DrainReturnsNothingDriver(SqliteRepositoryDriver): + """Negative control: drain hides undelivered rows.""" + + def drain_outbox(self, limit=None): + return [] + + +class _AckIgnoredDriver(SqliteRepositoryDriver): + """Negative control: ack marks nothing delivered.""" + + def ack_outbox(self, seqs): + return None + + +class _ReAckRaisesDriver(SqliteRepositoryDriver): + """Negative control: re-acking a delivered seq raises instead of + succeeding silently.""" + + def ack_outbox(self, seqs): + undelivered = {row["seq"] for row in self.drain_outbox()} + if any(seq not in undelivered for seq in seqs): + raise DriverError("outbox ack: seq already delivered") + super().ack_outbox(seqs) + + +class _ClaimsDrainWithoutAckDriver(SqliteRepositoryDriver): + """Negative control for the True branch: claims the drain/ack set + while `ack_outbox` is gone.""" + + ack_outbox = None + + def supports_outbox_drain(self): + return True + + +class _PartialIdempotencyDriver(DemoRepositoryDriver): + """Negative control for the False branch: two of the three + idempotency methods, so the default predicate reports False while a + method of the set is present.""" + + def idempotency_begin(self, workflow_id, key, now_ms, ttl_ms): + return ("started", None) + + def idempotency_finish(self, workflow_id, key, http_status, body): + return None + + +class _PartialOutboxDriver(DemoRepositoryDriver): + """Negative control for the False branch: drain without ack.""" + + def drain_outbox(self, limit=None): + return [] + + +class CapabilityTCKDiscriminatesTest(unittest.TestCase): + """`harness-reverse-controls`: each optional-method TCK case passes the + real sqlite driver without skipping and fails a driver built to break + exactly that case; every control asserts `testsRun == 1` and + `skipped == 0`, so neither a vacuous pass nor a skip can satisfy it.""" + + SQLITE_CONTROLS = ( + (_ClaimAlwaysStartsDriver, CAPABILITY_TCK_CASES[0]), + (_FinishForgetsDriver, CAPABILITY_TCK_CASES[1]), + (_ReleaseKeepsClaimDriver, CAPABILITY_TCK_CASES[2]), + (_DrainReturnsNothingDriver, CAPABILITY_TCK_CASES[3]), + (_AckIgnoredDriver, CAPABILITY_TCK_CASES[4]), + (_ReAckRaisesDriver, CAPABILITY_TCK_CASES[5]), + (_ClaimsDrainWithoutAckDriver, CAPABILITY_TCK_CASES[4]), + ) + + DEMO_CONTROLS = ( + (_PartialIdempotencyDriver, CAPABILITY_TCK_CASES[0]), + (_PartialOutboxDriver, CAPABILITY_TCK_CASES[3]), + ) + + def _store_path(self): + box = tempfile.TemporaryDirectory() + self.addCleanup(box.cleanup) + return os.path.join(box.name, "capability-tck.db") + + def test_every_capability_case_passes_the_sqlite_driver_without_skipping(self): + for case in CAPABILITY_TCK_CASES: + with self.subTest(case=case): + path = self._store_path() + result = _run_one_tck_case( + lambda: SqliteRepositoryDriver(path), case) + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.skipped), 0) + self.assertEqual(len(result.failures) + len(result.errors), 0, + (result.failures, result.errors)) + + def test_each_sqlite_negative_control_fails_its_case(self): + for driver_class, case in self.SQLITE_CONTROLS: + with self.subTest(control=driver_class.__name__, case=case): + path = self._store_path() + result = _run_one_tck_case(lambda: driver_class(path), case) + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.skipped), 0) + self.assertEqual(len(result.failures) + len(result.errors), 1) + + def test_a_partial_method_set_fails_the_declared_unsupported_branch(self): + for driver_class, case in self.DEMO_CONTROLS: + with self.subTest(control=driver_class.__name__, case=case): + result = _run_one_tck_case(lambda: driver_class(), case) + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.skipped), 0) + self.assertEqual(len(result.failures), 1) + + if __name__ == "__main__": unittest.main() diff --git a/impl/tests/test_driver_spi.py b/impl/tests/test_driver_spi.py index bef62d4..f75c4f9 100644 --- a/impl/tests/test_driver_spi.py +++ b/impl/tests/test_driver_spi.py @@ -21,6 +21,7 @@ from lnpl.drivers import BACKENDS, DriverError, SqliteRepositoryDriver, open_repository from tests.driver_spi_fixture import DemoRepositoryDriver +from tests.test_driver_contract import CAPABILITY_TCK_CASES, _run_one_tck_case GROUP = drivers_module.DRIVERS_ENTRY_POINT_GROUP @@ -137,3 +138,55 @@ def test_a_same_named_entry_point_never_shadows_builtin_fake(self): driver = open_repository("fake") self.assertIsNone(driver) + + +class _DrainOnlyDemoDriver(DemoRepositoryDriver): + def drain_outbox(self, limit=None): + return [] + + +class _ClaimingDemoDriver(DemoRepositoryDriver): + """Claims idempotency support without implementing any of the three + methods — the TCK must catch it, not skip it.""" + + def supports_idempotency(self): + return True + + +class DemoDriverCapabilityTest(unittest.TestCase): + """Issue #228/#229: the SPI demo driver implements none of the five + optional methods, so it declares neither capability and passes every + optional-method TCK case on the declared-unsupported path. It runs + one case at a time because the demo fails five unrelated TCK cases + (no predicate pushdown, no real transactions — analysis Spike S3).""" + + def test_the_demo_driver_declares_neither_capability(self): + driver = DemoRepositoryDriver() + + self.assertIs(driver.supports_idempotency(), False) + self.assertIs(driver.supports_outbox_drain(), False) + + def test_the_demo_driver_passes_every_case_as_declared_unsupported(self): + for case in CAPABILITY_TCK_CASES: + with self.subTest(case=case): + result = _run_one_tck_case(lambda: DemoRepositoryDriver(), + case) + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.skipped), 0) + self.assertEqual(len(result.failures) + len(result.errors), 0, + (result.failures, result.errors)) + + def test_half_an_outbox_pair_reads_as_unsupported(self): + driver = _DrainOnlyDemoDriver() + + self.assertIs(driver.supports_outbox_drain(), False) + self.assertIs(driver.supports_idempotency(), False) + + def test_claiming_idempotency_without_the_methods_fails_the_tck(self): + result = _run_one_tck_case(lambda: _ClaimingDemoDriver(), + CAPABILITY_TCK_CASES[0]) + + self.assertEqual(result.testsRun, 1) + self.assertEqual(len(result.skipped), 0) + self.assertEqual(len(result.errors), 1) + self.assertIn("AttributeError", result.errors[0][1]) diff --git a/impl/tests/test_ops_surface.py b/impl/tests/test_ops_surface.py index b8f750b..9b61e5c 100644 --- a/impl/tests/test_ops_surface.py +++ b/impl/tests/test_ops_surface.py @@ -127,6 +127,9 @@ def test_normal_healthz_is_200_and_touches_no_repository(self): calls = [] app = make_wsgi_app(_doc(OPEN_SRC), repository_factory=_recording_factory(calls)) + # issue #228: construction opens and closes one repository (the + # boot capability probe); count only what the request does. + calls.clear() status, _headers, body = call_wsgi(app, "GET", "/-/healthz") @@ -140,12 +143,15 @@ def test_normal_readyz_is_200_with_no_backend_configured(self): status, _headers, body = call_wsgi(app, "GET", "/-/readyz") self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) def test_normal_readyz_is_200_with_a_healthy_backend(self): calls = [] app = make_wsgi_app(_doc(OPEN_SRC), repository_factory=_recording_factory(calls)) + # issue #228: construction opens and closes one repository (the + # boot capability probe); count only what the request does. + calls.clear() status, _headers, body = call_wsgi(app, "GET", "/-/readyz") @@ -168,7 +174,7 @@ def test_normal_readyz_needs_no_token_on_a_role_gated_service(self): status, _headers, body = call_wsgi(app, "GET", "/-/readyz") self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) def test_normal_readyz_is_200_when_jwt_secret_env_is_actually_set(self): # The positive twin of the error-case unset test below — proves the @@ -183,7 +189,7 @@ def test_normal_readyz_is_200_when_jwt_secret_env_is_actually_set(self): status, _headers, body = call_wsgi(app, "GET", "/-/readyz") self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) # issue #187: the build_app-path twins of the readyz check 3 cases. # `build_app` used to drop `jwt_secret_env` before `make_wsgi_app`, so @@ -197,7 +203,7 @@ def test_normal_readyz_stays_200_on_a_source_and_backend_only_build_app_deployme self.assertEqual("LNPL_TEST_R9_SECRET", app.jwt_secret_env) self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) def test_normal_readyz_stays_200_when_no_jwt_secret_env_is_configured_at_all(self): with mock.patch.dict(os.environ): @@ -208,7 +214,7 @@ def test_normal_readyz_stays_200_when_no_jwt_secret_env_is_configured_at_all(sel self.assertIsNone(app.jwt_secret_env) self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) class ErrorTest(unittest.TestCase): @@ -307,7 +313,7 @@ def test_normal_readyz_is_200_before_any_signal(self): status, _headers, body = call_wsgi(server.get_app(), "GET", "/-/readyz") self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) def test_error_sigterm_flips_readyz_to_503_with_the_check_named(self): server = self._serve(OPEN_SRC) @@ -403,7 +409,7 @@ def test_normal_recovery_is_200(self): status, body, _err = self._probe(2 * self.FLOOR) self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) def test_normal_rotation_visible_after_one_probe(self): self.source.rotate("jwt", DEMO_SECRET_K1) @@ -439,7 +445,7 @@ def test_boundary_probe_right_after_startup_reports_the_startup_read(self): status, body, _err = self._probe(self.FLOOR - 0.1) self.assertEqual(200, status) - self.assertEqual({"status": "ok"}, body) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, body) self.assertEqual(1, self.source.get_calls) def test_error_failure_is_reported_until_the_floor_passes(self): diff --git a/impl/tests/test_serve_backend.py b/impl/tests/test_serve_backend.py index 3686731..c79b59a 100644 --- a/impl/tests/test_serve_backend.py +++ b/impl/tests/test_serve_backend.py @@ -27,14 +27,17 @@ import unittest from contextlib import redirect_stderr, redirect_stdout +from lnpl import drivers as drivers_module from lnpl.cli import main -from lnpl.drivers import HmacTokenProvider, SqliteRepositoryDriver, audience_for_path +from lnpl.drivers import (HmacTokenProvider, RepositoryDriver, SqliteRepositoryDriver, + audience_for_path, open_repository) from lnpl.lower import lower from lnpl.parser import parse from lnpl.repo_policy import default_rows from lnpl.serve import serve from tests.fixtures import SECRET_ACCOUNT +from tests.test_driver_spi import DEMO_ENTRY_POINT, entry_point, registered REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) SHORTEN = os.path.join(REPO, "examples", "shorten.lnpl") @@ -64,6 +67,22 @@ def compile_source(source, module="m"): return lower(parse(source), module).to_document() +def _non_driver_schemes(): + """issue #228 (ruling t2 D1 r2): the registered `lnpl.drivers` schemes + whose `open_repository` result is NOT a RepositoryDriver -- the only + way a production object could reach wsgi.py's legacy branch.""" + bad = [] + for ep in drivers_module._driver_entry_points(): + repo = open_repository(ep.name + ":unused") + try: + if not isinstance(repo, RepositoryDriver): + bad.append(ep.name) + finally: + if hasattr(repo, "close"): + repo.close() + return sorted(bad) + + class BackendServerTestCase(unittest.TestCase): """One server per test on an ephemeral port; teardown always runs.""" @@ -185,15 +204,18 @@ def test_every_request_gets_and_releases_its_own_connection(self): and released whichever way the request ends.""" port = self.start(compile_file(SHORTEN), repository_factory=self.sqlite_factory(self.store())) + # issue #228: `serve()` -> `make_wsgi_app` opens and closes one + # store for the boot capability probe before any request. + boot = len(self.opened) self.post(port, SHORTEN_PATH, LINK_PAYLOAD, {"Authorization": "Bearer anything"}) self.post(port, SHORTEN_PATH, LINK_PAYLOAD, {"Authorization": "Bearer anything"}) # this one fails (409) - self.wait_until(lambda: len(self.closed) == 2, + self.wait_until(lambda: len(self.closed) == boot + 2, "both requests to release their store") - self.assertEqual(2, len(self.opened)) + self.assertEqual(boot + 2, len(self.opened)) self.assertEqual(self.opened, self.closed, "a request did not release the store it opened") @@ -452,5 +474,31 @@ def test_the_backends_document_exists_and_is_pointed_at(self): self.assertIn("docs/backends.md", self.serving) +class OpenRepositoryReturnsDriversTest(unittest.TestCase): + """issue #228: wsgi.py's legacy branch serves non-RepositoryDriver test + doubles only; this pins that `open_repository` never hands one out.""" + + def test_normal_a_sqlite_spec_opens_a_repository_driver(self): + base = os.path.join(REPO, ".claude", "tmp") + os.makedirs(base, exist_ok=True) + box = tempfile.TemporaryDirectory(prefix="lnpl-t228-", dir=base) + self.addCleanup(box.cleanup) + repo = open_repository("sqlite:" + os.path.join(box.name, "s.db")) + self.addCleanup(repo.close) + self.assertIsInstance(repo, RepositoryDriver) + + def test_normal_every_registered_entry_point_opens_a_repository_driver(self): + with registered(DEMO_ENTRY_POINT): + self.assertEqual(["demo"], [ep.name for ep in drivers_module._driver_entry_points()]) + self.assertEqual([], _non_driver_schemes()) + + def test_error_a_registration_that_returns_a_plain_object_is_caught(self): + with registered(DEMO_ENTRY_POINT, entry_point("plain", "builtins:str")): + self.assertEqual(["plain"], _non_driver_schemes()) + + def test_boundary_the_fake_backend_opens_nothing(self): + self.assertIsNone(open_repository("fake")) + + if __name__ == "__main__": unittest.main() diff --git a/impl/tests/test_wsgi_idempotency_unsupported.py b/impl/tests/test_wsgi_idempotency_unsupported.py new file mode 100644 index 0000000..2961ebf --- /dev/null +++ b/impl/tests/test_wsgi_idempotency_unsupported.py @@ -0,0 +1,343 @@ +"""Issue #228: an `Idempotency-Key` the configured repository driver cannot +honour is refused (501 `idempotency-unsupported`), never silently ignored. + +`_idempotency_mode` (wsgi.py) decides: a `RepositoryDriver` claims when +`supports_idempotency()` is True and refuses when it is False; anything +else (None for the fake backend, `interp.FakeRepository`-style doubles) +takes the LEGACY pre-#228 path (coordinator ruling t2 D1 r2). +""" + +import contextlib +import io +import json +import os +import tempfile +import unittest + +from lnpl import wsgi +from lnpl.drivers import DriverError, SqliteRepositoryDriver +from lnpl.interp import FakeRepository +from lnpl.wsgi import make_wsgi_app + +from tests.driver_spi_fixture import DemoRepositoryDriver +from tests.fixtures import VALUE_PAYMENT +from tests.test_idempotency import PAYMENT_PATH, compile_source +from tests.test_wsgi_contract import call_wsgi +from tests.test_wsgi_event_consume import (EVENTS_PATH_CREATE, SRC, _doc, + _envelope, _post) + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) +DRIVER_NAME = "tests.test_wsgi_idempotency_unsupported._CountingDemoDriver" +PAYLOAD = {"id": "3f2504e0-4f89-41d3-9a0c-0305e82c330b", "amount": 5} + +WRITES = [] # every write-method call, as (method, detail) +PREDICATE_CALLS = [] # one entry per supports_idempotency() call +OPENED = [] # every _CountingDemoDriver instance, in order +CLOSED = [] # every close(), in order +LEGACY_CLAIMS = [] # _LegacyClaimingDouble's claim calls + + +class _CountingDemoDriver(DemoRepositoryDriver): + """A real `RepositoryDriver` whose predicate is False (none of the + three claim methods), recording every write and every predicate call.""" + + def __init__(self, arg=None): + super().__init__(arg) + OPENED.append(self) + + def seed(self, rows): + WRITES.append(("seed", None)) + return super().seed(rows) + + def execute(self, entity_id, operation, key): + WRITES.append(("execute", operation)) + return super().execute(entity_id, operation, key) + + def persist(self, entity_id, key, row): + WRITES.append(("persist", entity_id)) + return super().persist(entity_id, key, row) + + def record_emission(self, emission): + WRITES.append(("record_emission", None)) + return super().record_emission(emission) + + def supports_idempotency(self): + PREDICATE_CALLS.append(1) + return super().supports_idempotency() + + def close(self): + CLOSED.append(self) + return super().close() + + +class _PredicateOnlyDouble(FakeRepository): + """Not a RepositoryDriver, but defines the predicate (answering False): + still LEGACY -- the type decides, not the method.""" + + def supports_idempotency(self): + return False + + +class _LegacyClaimingDouble(FakeRepository): + """Not a RepositoryDriver, but has the claim methods: LEGACY claims.""" + + def idempotency_begin(self, workflow_id, key, now_ms, ttl_ms): + LEGACY_CLAIMS.append(("begin", key)) + return "started", None + + def idempotency_finish(self, workflow_id, key, http_status, body): + LEGACY_CLAIMS.append(("finish", http_status)) + + def idempotency_release(self, workflow_id, key): + LEGACY_CLAIMS.append(("release", key)) + + +def _build(doc, factory=None): + """`make_wsgi_app` with stderr captured -> (app, stderr text).""" + err = io.StringIO() + with contextlib.redirect_stderr(err): + app = make_wsgi_app(doc, repository_factory=factory) + return app, err.getvalue() + + +def _post_payment(app, key=None): + headers = {} if key is None else {"Idempotency-Key": key} + return call_wsgi(app, "POST", PAYMENT_PATH, + body=json.dumps(PAYLOAD).encode("utf-8"), + headers=headers) + + +class _Base(unittest.TestCase): + + def setUp(self): + for log in (WRITES, PREDICATE_CALLS, OPENED, CLOSED, LEGACY_CLAIMS): + del log[:] + + def store_path(self): + base = os.path.join(REPO_ROOT, ".claude", "tmp") + os.makedirs(base, exist_ok=True) + box = tempfile.TemporaryDirectory(prefix="lnpl-t228-", dir=base) + self.addCleanup(box.cleanup) + return os.path.join(box.name, "store.db") + + def assert_nothing_ran(self): + self.assertEqual([], WRITES) + self.assertTrue(all(driver.rows == {} for driver in OPENED)) + + +class RequestPathRefusalTest(_Base): + + def setUp(self): + super().setUp() + self.app, _ = _build(compile_source(VALUE_PAYMENT), + lambda: _CountingDemoDriver()) + + def test_normal_keyed_post_on_a_driver_without_idempotency_is_501_and_runs_nothing(self): + status, headers, body = _post_payment(self.app, key="k1") + self.assertEqual(501, status) + self.assertEqual("application/problem+json", headers["Content-Type"]) + self.assertEqual("idempotency-unsupported", body["code"]) + self.assertEqual(501, body["status"]) + self.assertEqual(wsgi._TITLES["idempotency-unsupported"], body["title"]) + self.assertTrue(body["correlation_id"].startswith("req-")) + self.assertIn(DRIVER_NAME, body["detail"]) + self.assertIn("Idempotency-Key", body["detail"]) + self.assert_nothing_ran() + + def test_boundary_an_empty_key_value_still_counts_as_present(self): + status, _headers, body = _post_payment(self.app, key="") + self.assertEqual(501, status) + self.assertEqual("idempotency-unsupported", body["code"]) + self.assert_nothing_ran() + + def test_boundary_unkeyed_post_on_that_driver_runs_as_before(self): + before = len(PREDICATE_CALLS) + status, _headers, body = _post_payment(self.app) + self.assertEqual(200, status) + self.assertEqual("completed", body["status"]) + self.assertIn(("execute", "create"), WRITES) + # the helper is not consulted without the header + self.assertEqual(before, len(PREDICATE_CALLS)) + + +class ConsumePathRefusalTest(_Base): + + def setUp(self): + super().setUp() + self.app, _ = _build(_doc(SRC), lambda: _CountingDemoDriver()) + + def test_error_consume_on_a_driver_without_idempotency_is_501_and_runs_nothing(self): + status, headers, body = _post( + self.app, EVENTS_PATH_CREATE, + _envelope(event_type="OrderCreated", + data={"id": "ord-228", "amount": 1})) + self.assertEqual(501, status) + self.assertEqual("application/problem+json", headers["Content-Type"]) + self.assertEqual("idempotency-unsupported", body["code"]) + self.assertTrue(body["correlation_id"].startswith("req-")) + self.assertIn(DRIVER_NAME, body["detail"]) + self.assertIn("CloudEvents id", body["detail"]) + self.assert_nothing_ran() + # the route's `finally` still closed the request's driver + self.assertEqual(len(OPENED), len(CLOSED)) + + def test_boundary_an_envelope_without_id_is_still_400(self): + envelope = _envelope(event_type="OrderCreated", + data={"id": "ord-228", "amount": 1}) + del envelope["id"] + status, _headers, body = _post(self.app, EVENTS_PATH_CREATE, envelope) + self.assertEqual(400, status) + self.assertEqual("cloudevents-invalid", body["code"]) + self.assert_nothing_ran() + + +class LegacyAndFakeBoundaryTest(_Base): + + def test_boundary_fake_backend_with_a_key_still_runs_and_never_501s(self): + app, err = _build(compile_source(VALUE_PAYMENT)) + status1, _h1, body1 = _post_payment(app, key="k1") + status2, _h2, body2 = _post_payment(app, key="k1") + self.assertEqual(200, status1) + self.assertEqual(200, status2) + self.assertNotEqual(body1["correlation_id"], body2["correlation_id"]) + self.assertEqual(1, err.count("Idempotency-Key support is disabled")) + self.assertNotIn("idempotency: unsupported", err) + + def test_boundary_a_fake_repository_double_keeps_the_pre_228_no_claim_path(self): + app, _ = _build(compile_source(VALUE_PAYMENT), lambda: FakeRepository()) + status1, _h1, body1 = _post_payment(app, key="k1") + status2, _h2, body2 = _post_payment(app, key="k1") + self.assertEqual(200, status1) + self.assertEqual(200, status2) + self.assertNotEqual(body1["correlation_id"], body2["correlation_id"]) + + def test_boundary_a_double_defining_the_predicate_is_still_legacy(self): + app, _ = _build(compile_source(VALUE_PAYMENT), + lambda: _PredicateOnlyDouble()) + status, _headers, body = _post_payment(app, key="k1") + self.assertEqual(200, status) + self.assertEqual("completed", body["status"]) + + def test_normal_a_legacy_double_with_the_claim_methods_still_claims(self): + app, _ = _build(compile_source(VALUE_PAYMENT), + lambda: _LegacyClaimingDouble()) + status, _headers, _body = _post_payment(app, key="k1") + self.assertEqual(200, status) + self.assertEqual([("begin", "k1"), ("finish", 200)], LEGACY_CLAIMS) + + +class SqlitePathTest(_Base): + + def test_normal_sqlite_keyed_repeat_still_replays(self): + path = self.store_path() + app, _ = _build(compile_source(VALUE_PAYMENT), + lambda: SqliteRepositoryDriver(path)) + status1, _h1, body1 = _post_payment(app, key="k-sqlite") + status2, _h2, body2 = _post_payment(app, key="k-sqlite") + self.assertEqual(200, status1) + self.assertEqual(200, status2) + # replayed: the identical body, same correlation_id included + self.assertEqual(body1, body2) + + + +def _unreachable_store(): + raise DriverError("store unreachable") + + +class ReadyzCapabilityTest(_Base): + + def test_normal_readyz_reports_true_on_sqlite(self): + path = self.store_path() + app, _ = _build(compile_source(VALUE_PAYMENT), + lambda: SqliteRepositoryDriver(path)) + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(200, status) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": True}}, + body) + + def test_normal_readyz_reports_false_on_a_driver_without_it(self): + app, _ = _build(compile_source(VALUE_PAYMENT), + lambda: _CountingDemoDriver()) + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(200, status) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, + body) + + def test_boundary_readyz_reports_false_with_no_backend(self): + app, _ = _build(compile_source(VALUE_PAYMENT)) + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(200, status) + self.assertEqual({"status": "ok", "capabilities": {"idempotency": False}}, + body) + + def test_boundary_readyz_omits_capabilities_for_a_legacy_double(self): + app, _ = _build(compile_source(VALUE_PAYMENT), lambda: FakeRepository()) + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(200, status) + self.assertEqual({"status": "ok"}, body) + + def test_error_readyz_503_is_unchanged_when_the_store_is_unreachable(self): + app, _ = _build(compile_source(VALUE_PAYMENT), _unreachable_store) + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(503, status) + self.assertEqual("not-ready", body["code"]) + self.assertEqual(["repository"], body["checks"]) + self.assertNotIn("capabilities", body) + + def test_boundary_a_shutting_down_server_carries_no_capabilities(self): + app, _ = _build(compile_source(VALUE_PAYMENT)) + app.shutting_down = True + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(503, status) + self.assertEqual(["shutting-down"], body["checks"]) + self.assertNotIn("capabilities", body) + + + +class BootLineTest(_Base): + + def boot_lines(self, err): + return [line for line in err.splitlines() + if "idempotency: unsupported" in line] + + def test_normal_a_driver_without_idempotency_prints_one_line_naming_it(self): + _, err = _build(compile_source(VALUE_PAYMENT), + lambda: _CountingDemoDriver()) + lines = self.boot_lines(err) + self.assertEqual(1, len(lines)) + self.assertTrue(lines[0].startswith( + "lnpl serve: repository driver " + DRIVER_NAME + + " -- idempotency: unsupported"), lines[0]) + + def test_normal_the_boot_probe_closes_what_it_opens(self): + _build(compile_source(VALUE_PAYMENT), lambda: _CountingDemoDriver()) + self.assertEqual(1, len(OPENED)) + self.assertEqual(OPENED, CLOSED) + + def test_boundary_sqlite_prints_no_line(self): + path = self.store_path() + _, err = _build(compile_source(VALUE_PAYMENT), + lambda: SqliteRepositoryDriver(path)) + self.assertEqual([], self.boot_lines(err)) + + def test_boundary_fake_prints_only_the_d11_notice(self): + _, err = _build(compile_source(VALUE_PAYMENT)) + self.assertEqual([], self.boot_lines(err)) + self.assertEqual(1, err.count("Idempotency-Key support is disabled")) + + def test_boundary_a_legacy_double_prints_no_line(self): + _, err = _build(compile_source(VALUE_PAYMENT), lambda: FakeRepository()) + self.assertEqual([], self.boot_lines(err)) + + def test_error_an_unreachable_store_at_boot_prints_nothing_and_construction_succeeds(self): + app, err = _build(compile_source(VALUE_PAYMENT), _unreachable_store) + self.assertEqual([], self.boot_lines(err)) + status, _h, body = call_wsgi(app, "GET", "/-/readyz") + self.assertEqual(503, status) + self.assertEqual(["repository"], body["checks"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/plugins/lnpl/skills/lnpl-authoring/cli-surface.md b/plugins/lnpl/skills/lnpl-authoring/cli-surface.md index 53ddd28..0f90ce5 100644 --- a/plugins/lnpl/skills/lnpl-authoring/cli-surface.md +++ b/plugins/lnpl/skills/lnpl-authoring/cli-surface.md @@ -216,11 +216,13 @@ lnpl outbox ack --backend sqlite: [...] `--backend` 규칙). `fake`는 프로세스 밖 저장소가 없으므로 두 서브커맨드 모두 이 값을 거부한다(rc 2). +`lnpl.drivers`로 등록한 외부 드라이버는 `drain_outbox`/`ack_outbox` 둘 다를 구현해야 한다(`supports_outbox_drain()`이 True). 하나라도 없으면 두 서브커맨드 모두 drain/ack를 부르기 전에 `error: backend does not implement the outbox drain contract (drain_outbox/ack_outbox) — see docs/backends.md §3` 한 줄을 stderr에 찍고 rc 2로 끝난다 — stdout에는 아무것도 나가지 않는다(이슈 #229). 기능별 대응은 `docs/backends.md` §8 「선택 메서드와 능력 술어」. + | 서브커맨드 | 플래그 | 뜻 | |-----------|--------|-----| -| `drain` | `--backend` | 필수. `sqlite:`만 유효 | +| `drain` | `--backend` | 필수. `sqlite:`, 또는 `drain_outbox`/`ack_outbox`를 둘 다 구현한 등록 드라이버(`:`, 아래 참고) | | `drain` | `--limit` | 출력할 최대 개수 (기본 무제한) | -| `ack` | `--backend` | 필수. `sqlite:`만 유효 | +| `ack` | `--backend` | 필수. `sqlite:`, 또는 `drain_outbox`/`ack_outbox`를 둘 다 구현한 등록 드라이버(`:`, 아래 참고) | | `ack` | (위치 인자) `` | delivered로 마킹할 `seq` 값(들). `drain` 출력의 첫 필드 | `drain`은 아직 delivered로 마킹되지 않은 모든 행을 `seq` 오름차순(삽입 순서) @@ -254,11 +256,13 @@ lnpl relay --backend sqlite: --target [--once] | 플래그 | 뜻 | |--------|-----| -| `--backend` | 필수. 드레인할 영속 백엔드(`sqlite:`). `fake`는 outbox가 없어 거부(rc 2) | +| `--backend` | 필수. 드레인할 영속 백엔드 — `sqlite:`, 또는 `drain_outbox`/`ack_outbox`를 둘 다 구현한 등록 드라이버(`:`, 아래 참고). 그 밖의 드라이버와 `fake`(outbox가 없음)는 거부(rc 2) | | `--target` | 필수. `http(s)://`이면 소비 인스턴스의 base URL — 봉투는 `<--target>/-/events/`로 POST된다(바이트 동일). 그 외 스킴(예: `kafka://…`)이면 등록된 `lnpl.publishers` 드라이버가 발행한다 — 미등록 스킴은 rc!=0 + 스킴/등록 목록을 담은 에러(크리덴셜은 절대 로그에 남지 않는다) | | `--once` | 한 번 드레인·POST하고 종료(rc 0) — 테스트나 cron이 미는 모양. 기본은 무한 반복(폴링 간격 1초) | | (위치 인자) `source` | 컴파일만 한다(재실행 아님) — emission의 이벤트 id를 이벤트의 선언된 이름(슬러그/`type`)으로 되돌리는 데만 쓴다 | +`--backend` 드라이버가 `drain_outbox`/`ack_outbox` 둘 다를 구현하지 않으면(`supports_outbox_drain()`이 False) relay는 `--target` 발행자를 열기 전, 첫 드레인 전에 `outbox drain`과 같은 한 줄 오류와 rc 2로 끝난다(이슈 #229). + 봉투: `specversion="1.0"`, `id="outbox-"`(안정값), `source`=모듈명, `type`=이벤트 이름, `data`=emission의 payload. ack 규율(성공 후에만 커밋 — offset-commit discipline): 200 → ack. 422 → ack(재시도해도 같은 결과이므로