diff options
| -rw-r--r-- | CHECKPOINT.md | 4 | ||||
| -rw-r--r-- | README.md | 10 | ||||
| -rw-r--r-- | docs/design/008-module-format.md | 6 | ||||
| -rw-r--r-- | docs/research/078-app-connections.md | 4 | ||||
| -rw-r--r-- | docs/research/083-delivery-over-http.md | 73 | ||||
| -rw-r--r-- | prototypes/connections/Makefile | 4 | ||||
| -rw-r--r-- | prototypes/connections/deliver.c | 500 | ||||
| -rw-r--r-- | prototypes/connections/deliver.h | 36 | ||||
| -rw-r--r-- | prototypes/connections/module.c | 5 | ||||
| -rw-r--r-- | prototypes/connections/module.h | 5 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 245 | ||||
| -rw-r--r-- | prototypes/connections/test-module.c | 8 |
12 files changed, 852 insertions, 48 deletions
diff --git a/CHECKPOINT.md b/CHECKPOINT.md index 0ab4bfe..b50858d 100644 --- a/CHECKPOINT.md +++ b/CHECKPOINT.md @@ -15,10 +15,10 @@ | 앱 표현 | 상태(수, text와 커서·마크, 레코드의 리스트와 키 있는 리스트, 인덱스), 입력(키, text, 필드 있는 메시지, 타이머), 원자적 이벤트 핸들러, 순수 보기(조건식, 표와 걸러진 표, count·sum·max·min·first), `fn`·`use` 라이브러리, text 비교, 시간(`now`, `every`, `floor`, `%`). 로드할 때 타입·스택·합류·커서와 인덱스의 소속을 모두 검증하고 이벤트의 실행량에 상한이 있다 | 008, aup `docs/design/002-dsl-direction.md`, aup `make check`(파이썬 참조 모델 대조, 진단 132건) | | 영속 | 선언한 상태가 곧 영속 상태다. 트리 저장소에 커밋당 동기화 한 번(헤드를 바이트 바로 앞에 두어 한 연속 구간), 이벤트 묶음마다 저장 한 번. `kill -9` 뒤 확인 응답한 이벤트가 모두 남는다. 레코드·키가 바뀐 모듈로 바꿔도 데이터가 이름으로 옮겨진다 | [076](docs/research/076-one-sync-commit.md), [081](docs/research/081-tail-records.md), 008 상태 영속·교체 | | 인터페이스 | 한 명세가 TUI(셀 단위 갱신), CLI, 문장(NUI), 웹 GUI가 된다. 저장하면 실행 중인 앱이 상태를 이어 새 모듈로 바뀐다(`aup watch`, `vega … --watch`) | Vega `docs/design/001-first-component.md`, Vega `make check`(가상 터미널, 헤드리스 Chromium 포함) | -| HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md) | +| HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) | | 성능(Pi 5) | 문서 서버: 작은 GET 151.6k/s, 1 MiB GET 2,794/s(axum 1,372, drogon 708), 충돌 안전 PUT은 axum-atomic의 5.1–5.8배. 앱 서비스(aup 22줄 대 손으로 쓴 서비스): 내구성 있는 쓰기 약 76배(그 뒤 저장소 커밋을 한 구간으로 바꿔 64연결 21–26k/s, axum·drogon 내구 약 180/s), 20행 JSON GET 145k/s(axum 95k, drogon 144k), 쓰기와 섞인 읽기 149k/s·p99 0.6 ms(axum·drogon 내구 1.1–1.3k/s·p99 111–156 ms), 최대 RSS 3.0–4.1 MiB(axum 6.4, drogon 11.6) | [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) | -남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 앱 라우트의 인증·TLS(문서 +남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 전달의 TLS와 응답을 이벤트로 돌려받기(083 다음 결정), 앱 라우트의 인증·TLS(문서 피어의 TLS는 있다), aup·Vega의 배포 묶음. ## 공개할 작업물과 완료 증거 @@ -6,10 +6,12 @@ in Pure's tree store, every event is a transaction, views are pure functions of terminal UI, a CLI, spoken sentences and a web page through [Vega](/srv/git/vega.git), and an HTTP service through `pure-serve`: views are GETs and inputs are POSTs, with JSON bodies (groups included) read in place; answers are durable, a commit being one contiguous write and one sync; views are read from a snapshot of the last save, so -reads never wait for a sync. Measurements against axum and drogon on a Raspberry Pi 5 are in -[075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), -[080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) and -[082](docs/research/082-json-in-place.md); the status table is [CHECKPOINT.md](CHECKPOINT.md). The text below +reads never wait for a sync. What an app sends is saved with its event and delivered over HTTP in order and in +batches, and a receiving app applies each message once. Measurements against axum and drogon on a Raspberry Pi 5 +are in [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), +[080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md), +[082](docs/research/082-json-in-place.md) and [083](docs/research/083-delivery-over-http.md); the status table is +[CHECKPOINT.md](CHECKPOINT.md). The text below records the earlier document runtime this grew from. A standalone native micro runtime being designed as a small operating diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index e67afa8..07cff23 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -289,6 +289,7 @@ text·리스트가 남을 때만 위치를 유지한다. | `GET /views/이름` | 보기의 값. 수는 가장 짧은 왕복 10진수, text는 UTF-8 그대로(`text/plain`). text 상태를 그대로 보이는 보기(`view page = note`)는 저장된 익스텐트를 빌려준다([079](../research/079-lent-views.md)) | | `GET /views/표?from=N&count=M` | 표의 행: `{"rows":전체,"from":N,"items":[{열: 값}…]}`(`application/json`). 요청한 행의 칸만 계산한다. 값이 없는 칸은 `null` | | `POST /inputs/이름` | 이벤트. text 입력은 본문이 text, 키 입력은 누름(down 뒤 up). 메시지는 고정 필드(u32, 수)를 쿼리(`?points=3`, 그룹 안이면 `?to.zip=7`)로, 가변 필드 하나를 본문으로 받거나, `Content-Type: application/json` 본문의 객체로 필드를 받는다(문자열은 바이트·text, 수는 u32·수, 그룹은 안의 객체). 응답 본문은 핸들러의 `reply`(메시지 필드) | +| `POST /inputs/이름` + `X-Pure-Source`·`X-Pure-Sequence` | 다른 앱의 출력에서 전해진 메시지(`pure_app_deliver`): 번호마다 한 번 적용하고, 이미 들은 번호는 200. 답에 `X-Pure-Batch: 256`을 싣고, `X-Pure-Count: N`이면 본문은 N개 객체의 JSON 배열(연속 번호, 한 묶음에서 차례로 적용) | | `GET`·`PUT /objects/키` | 모듈에 키별 메시지 `show`·`edit`가 있으면 기존 문서 서버(`server.c`)와 같은 계약. `GET`은 `show`, `PUT`(`If-Match`/`If-None-Match`)은 `edit` | - **실패는 실패 종류로 답한다.** 이벤트가 실패하면 `X-Pure-Error`에 종류를 싣는다. `not_found` 404, `exists` 409, @@ -337,6 +338,11 @@ text·리스트가 남을 때만 위치를 유지한다. 이벤트 30 → 50, p50은 오히려 5.4 → 3.5 ms). 끊김은 답을 돌려준 뒤에 온 요청이 하나라도 있어야 세기 시작한다. 저장을 기다리는 동안 줄에 선 요청만으로 끊김을 세면 막 답을 받은 클라이언트가 다음 묶음을 놓치고, 클라이언트가 두 무리로 갈려 번갈아 저장한다(연결 스레드 둘, 64연결에서 묶음당 34개, 15.8–16.0k/s였고, 고친 뒤 61.6개, 22.9–23.9k/s). +- **보냄의 전달.** `--send 출력=http://호스트:포트/경로`로 출력을 받는 곳에 잇는다. 저장이 끝날 때마다 소유자가 새 보냄을 + JSON(필드와 그룹)으로 그 출력의 줄에 넣고, 출력마다 전달 스레드가 순서대로 보낸다(`X-Pure-Source: 이름.출력`, + `X-Pure-Sequence`, `Idempotency-Key`). 2xx 또는 Pure 받는 쪽의 거절이면 전해진 것이고, 소유자가 덜어 낸다. 그 밖에는 쉬었다가 + 다시 보낸다(버리지 않는다). 받는 쪽이 `X-Pure-Batch`를 알리면 밀린 것을 묶어 보낸다([083](../research/083-delivery-over-http.md)). + 출력이 있는 앱은 상태가 없어도 소유자를 둔다. - **보기는 스냅숏에서.** 소유자는 저장이 성공할 때마다, 답을 돌려주기 전에 저장된 상태의 스냅숏을 게시한다. 연결 스레드는 라운드마다(돌려받은 답을 처리한 뒤) 최신 스냅숏을 잡아 자기 읽기용 앱으로 그 라운드의 보기를 모두 평가하고 답을 모아 보낸다. 입력은 그보다 먼저 소유자에게 넘겨, 소유자가 그동안 적용하고 저장한다. 그래서 쓰기의 diff --git a/docs/research/078-app-connections.md b/docs/research/078-app-connections.md index 5e89853..a75cdfd 100644 --- a/docs/research/078-app-connections.md +++ b/docs/research/078-app-connections.md @@ -2,8 +2,8 @@ 2026-09-26. 설계 노트이며 구현 순서를 포함한다. 상태: 1단계(Pure 런타임: 출력, `SEND`, outbox·inbox와 그 저장)와 2단계(aup `output`·`send`)를 2026-09-27에 구현했다([008 앱 사이의 연결](../design/008-module-format.md)). 호스트의 배선과 -전달(3·4단계)은 [로드맵](../roadmap.md) 중기 1(외부로 나가는 호출)과 함께 한다. 외부 API 호출도 같은 outbox 위에 선다(출력을 -HTTP 끝점에 잇는 배선). +전달은 한 프로세스 안이 아니라 곧바로 HTTP로 했다(`pure-serve --send`, [083](083-delivery-over-http.md)). 다른 앱에도 외부 API에도 +같은 outbox와 같은 전달이다. 착수 기준: diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md new file mode 100644 index 0000000..48da056 --- /dev/null +++ b/docs/research/083-delivery-over-http.md @@ -0,0 +1,73 @@ +# 083 — 앱의 보냄을 HTTP로: 저장된 것만, 순서대로, 받는 쪽은 한 번 + +2026-09-27. [로드맵](../roadmap.md) 중기 1 "비동기 아웃바운드 호출"의 첫 단계이자 [078](078-app-connections.md)의 3·4단계. + +착수 기준: + +- 바꿀 결정: 앱이 다른 서비스를 부를 때 핸들러 안에서 HTTP 클라이언트를 쓰고 재시도·중복 제거·outbox를 앱이 짤지, + 보냄을 이벤트의 일부로 저장하고 호스트가 저장 뒤에 전하며 받는 쪽 런타임이 번호로 한 번만 적용할지. +- 남은 질문: 보내는 서버와 받는 서버가 아무 순간에 죽어도, 보내는 앱이 저장한 메시지가 받는 앱에 정확히 한 번씩 + 적용되는가? 전달이 받는 쪽의 저장(동기화)에 묶이면 얼마나 느린가? +- 최소 확인: 두 `pure-serve`를 잇고, 클라이언트 넷이 주문하는 동안 받는 서버와 보내는 서버를 `kill -9`로 죽였다 살린 + 뒤, 받는 앱의 합계와 건수가 보내는 앱이 저장한 것과 같은지 본다. 전달 속도를 잰다. +- 중단 기준: 한 번이라도 합계가 다르면(잃거나 겹치면) 되돌린다. + +## 의심한 관례 + +| 관례 | 이 설계 | +|---|---| +| 외부 호출은 핸들러 안에서 한다. 이벤트가 실패하면 보상 호출을 짠다 | `send`는 이벤트의 일부로 저장되고, 호스트는 저장된 보냄만 전한다. 실패한 이벤트는 보낸 적이 없다 | +| 재시도, 중복 제거, outbox 테이블은 앱 코드다 | 번호, 순서, 재시도, 덜어 내기는 호스트가, 들은 번호는 받는 쪽 런타임이 맡는다. 앱에는 `send` 한 줄이다 | +| 서비스 사이에는 메시지 브로커가 있다 | 보내는 앱의 저장소가 곧 줄이다. 받는 쪽이 저장한 뒤에야 덜어 낸다 | +| 묶음 크기는 설정이다 | 받는 쪽이 한 요청에 받을 수를 알리고(`X-Pure-Batch`), 보내는 쪽은 밀린 만큼 묶는다 | + +## 한 것 + +- **배선.** `pure-serve --send 출력=http://호스트:포트/경로`. 보내는 쪽 이름은 `--name`(없으면 모듈 파일 이름)과 출력 + 이름을 이은 것(`orders.ordered`)이다. 출력이 있는 앱은 상태가 없어도 소유자를 둔다(보냄이 상태이므로). +- **전달.** 저장이 끝날 때마다(시작할 때 저장소에 남은 것도) 소유자가 배선된 출력의 새 보냄을 JSON으로 만들어 그 + 출력의 줄에 넣는다. 출력마다 전달 스레드 하나가 순서대로 `POST`한다(`X-Pure-Source`, `X-Pure-Sequence`, + `Idempotency-Key`). 2xx면 전해졌고, Pure 받는 쪽이 앱의 거절(실패 종류가 있는 4xx)로 답해도 전해졌다(들었고 거절했다). + 408·429·5xx·응답 없음은 0.1초에서 두 배씩 10초까지 쉬었다가, 다른 4xx는 기록하고 10초마다 다시 보낸다. 버리지 않는다. + 전해진 번호까지 소유자가 덜어 내고(`pure_app_trim`), 다음 저장이 저장소에서 지운다. 연결은 이어 쓰고, 끊긴 연결에 보낸 + 것은 새 연결로 한 번 더 보낸다(받는 쪽이 번호로 거른다). +- **받기.** `X-Pure-Source`와 `X-Pure-Sequence`가 붙은 `POST /inputs/이름`을 `pure_app_deliver`로 적용한다. 이미 들은 + 번호는 적용하지 않고 200이다. 앱이 거절하면 들은 것으로 치고 그 실패 종류로 답한다. 답에는 `X-Pure-Batch: 256`을 싣는다. +- **묶음.** 그 헤더를 본 보내는 쪽은 줄에 쌓인 연속 번호를 최대 256개까지 JSON 배열 하나로 보낸다(`X-Pure-Count`). + 받는 쪽은 한 요청의 메시지를 차례로 한 묶음에서 적용하고 한 번 저장한다. 거절한 것은 들은 것으로 치고 넘어가고, 다른 + 실패에서 멈추면 묶음 전체를 다시 받는다(앞부분은 번호로 걸러진다). 제3자 끝점은 이 헤더를 모르므로 하나씩 받는다. +- JSON 본문은 출력의 필드와 그룹 그대로다([082](082-json-in-place.md)의 입력과 같은 모양). 한 그룹의 필드가 떨어져 있는 + 출력, `https`는 시작할 때 거절한다. UTF-8이 아닌 bytes 필드는 JSON으로 만들 수 없어 그 출력이 거기서 멈춘다(보냄은 남는다). + +## 장치가 말한 것 + +Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버는 CPU 2, 클라이언트(파이썬 16스레드)는 나머지. + +| | 하나씩 | 묶음 | +|---|---:|---:| +| 주문 2,992건을 넣는 동안 | 2.49 s(1,200/s) | 1.04 s(2,864/s) | +| 받는 쪽이 다 받은 때 | 7.64 s(**391건/s**) | 1.10 s(주문을 따라감) | +| 받는 쪽이 꺼진 동안 쌓인 20,000건 | — | 첫 묶음 뒤 0.32 s(**60,215건/s**) | + +하나씩 보내면 전달 하나가 받는 쪽의 저장 한 번(SD의 동기화 1–2 ms)과 왕복을 기다린다. 묶으면 저장 한 번이 256건까지 +나눈다. 보내는 앱의 저장이 이미 주문을 묶어 저장하므로, 받는 앱도 같은 비율로 묶어 받는 것이 맞다. + +## 검증 + +- aup `make check`의 `server_links`: 클라이언트 넷이 보내는 앱에 주문하는 동안 받는 서버를, 그다음 보내는 서버를 + `kill -9`로 죽이고 같은 저장소로 다시 띄운다. 그 뒤 수량 150(받는 앱이 거절한다)과 7의 주문을 더 넣는다. 받는 앱의 건수와 + 합계가 보내는 앱이 저장한 것(거절한 한 건을 빼고)과 같아지고, 확인 응답한 주문은 모두 그 안에 있다. 다섯 번(각 약 + 7,300건)과 ASan·UBSan 빌드로 한 번 통과했다. 거절한 주문 뒤의 주문도 전해졌다(막히지 않는다). +- `test-module`의 링크 검사([078](078-app-connections.md), 008)가 런타임 쪽(이벤트의 보냄, 저장 실패, 재시작, 중복·거절)을 + 맡는다. + +## 다음 결정 + +- **TLS.** 제3자 API는 거의 `https`다. 문서 피어의 TLS([055](055-authenticated-peer-transport-cost.md))를 전달 클라이언트에 + 쓰는 것이 다음이다. +- **응답을 이벤트로.** 지금은 보내고 끝이다(알림, 웹훅). 결제처럼 답이 필요한 호출은 응답을 앱의 입력으로 돌려주는 + 배선(`--send 출력=URL --reply 입력`)이 필요하다. +- 저장이 끝날 때마다 outbox 전체를 훑어 새 보냄을 찾는다. 받는 쪽이 오래 꺼져 수만 건이 밀리면 저장마다 그만큼 훑는다. + 마지막으로 넣은 자리에서 이어 훑게 해야 한다. +- 덜어 낸 것은 다음 저장에 지워진다. 요청이 없는 앱은 저장하지 않으므로, 다시 시작하면 전한 것을 한 번 더 보낸다(받는 + 쪽이 걸러 낸다). diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile index 6775e5a..60d83d7 100644 --- a/prototypes/connections/Makefile +++ b/prototypes/connections/Makefile @@ -44,8 +44,8 @@ $(BUILD)/pure-connections-async-gated: server.c $(HOST) flow.c program.c $(CORE) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_ASYNC_EFFECTS -DPURE_HTTP_ASYNC -DPURE_STORE_TEST server.c $(HOST) flow.c program.c $(CORE) test-async-server-gate.c $(LDFLAGS) -pthread $(LDLIBS) -o $@ # The HTTP host for aup modules: persistent connections, epoll, group commits. -$(BUILD)/pure-serve: serve.c flow.c module-store.c $(MODULE) $(CORE) module-store.h $(MODULE_HEADERS) | $(BUILD) - $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@ +$(BUILD)/pure-serve: serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) deliver.h module-store.h $(MODULE_HEADERS) | $(BUILD) + $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@ $(BUILD)/pure-connections-fixed: server.c $(HOST) flow.c $(CORE) $(HEADERS) $(BUILD)/embedded-plan.h | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_EMBEDDED_PROGRAM -I$(BUILD) server.c $(HOST) flow.c $(CORE) $(LDFLAGS) $(LDLIBS) -o $@ diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c new file mode 100644 index 0000000..5f32978 --- /dev/null +++ b/prototypes/connections/deliver.c @@ -0,0 +1,500 @@ +#define _POSIX_C_SOURCE 200809L +#include "deliver.h" +#include "module-text.h" +#include <errno.h> +#include <netdb.h> +#include <netinet/in.h> +#include <netinet/tcp.h> +#include <pthread.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <strings.h> +#include <sys/socket.h> +#include <sys/time.h> +#include <time.h> +#include <unistd.h> + +enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 }; +enum { DELIVERED, AGAIN, REFUSED_BY_PEER }; + +typedef struct { char *body; size_t size; uint64_t seq; } Pending; + +struct PureWire { + const PureModule *module; + uint32_t output; + char host[256], port[8], path[1024], source[PURE_SOURCE + 1]; + pthread_mutex_t lock; + pthread_cond_t changed; + Pending *queue; /* the sends to deliver, in order, from head */ + size_t head, count, capacity; + uint64_t queued, delivered, attempts, deliveries; + unsigned batch; /* how many a request may carry, as the receiver said (X-Pure-Batch); 0: one */ + bool stopping, started, stuck; + pthread_t thread; + void (*notify)(void *); + void *context; + int fd; /* the connection kept for the next send, or -1; changed under the lock */ +}; + +/* ---- A send as JSON: its fields in order, a group's fields as an object. */ + +typedef struct { char *bytes; size_t size, capacity; bool failed; } Text; + +static void put(Text *t, const void *bytes, size_t size) { + if (t->failed) return; + if (size > t->capacity - t->size) { + size_t capacity = t->capacity ? t->capacity : 256; + while (size > capacity - t->size) capacity *= 2; + char *grown = realloc(t->bytes, capacity); + if (!grown) { t->failed = true; return; } + t->bytes = grown; + t->capacity = capacity; + } + memcpy(t->bytes + t->size, bytes, size); + t->size += size; +} + +static void put_string(Text *t, const unsigned char *bytes, size_t size) { + put(t, "\"", 1); + size_t run = 0; + for (size_t i = 0; i < size; ++i) { + unsigned char b = bytes[i]; + if (b != '"' && b != '\\' && b >= 0x20) continue; + put(t, bytes + run, i - run); + char escaped[8]; + if (b == '"' || b == '\\') put(t, (char[]){'\\', (char)b}, 2); + else put(t, escaped, (size_t)snprintf(escaped, sizeof(escaped), "\\u%04x", b)); + run = i + 1; + } + put(t, bytes + run, size - run); + put(t, "\"", 1); +} + +static uint32_t read_u32(const unsigned char *p) { return p[0] | p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; } + +/* The group a field is in: the length of its name before the last dot (0: none). */ +static size_t group_of(const char *name) { + const char *dot = strrchr(name, '.'); + return dot ? (size_t)(dot - name) : 0; +} + +/* Whether the group path open (length open_size) holds the group of name (length size). */ +static bool holds(const char *open, size_t open_size, const char *name, size_t size) { + return open_size <= size && !memcmp(open, name, open_size) && (open_size == size || !open_size || name[open_size] == '.'); +} + +static bool encode(const PureWire *w, PureSpan payload, Text *t) { + const PureModule *m = w->module; + const PureInputShape *shape = &m->input[w->output]; + if (payload.size < shape->prefix) return false; + char open[PURE_MODULE_NAME] = ""; + size_t open_size = 0, at = shape->prefix; + unsigned depth = 0; + bool first[GROUPS + 1] = {true}; + put(t, "{", 1); + for (uint32_t f = 0; f < shape->field_count; ++f) { + const PureField *field = &m->field[shape->field_start + f]; + size_t group = group_of(field->name); + while (depth && !holds(open, open_size, field->name, group)) { + put(t, "}", 1); + --depth; + open_size = group_of(open) < open_size ? group_of(open) : 0; + open[open_size] = 0; + } + while (open_size < group) { + size_t start = open_size ? open_size + 1 : 0, end = start; + while (end < group && field->name[end] != '.') ++end; + if (!first[depth]) put(t, ",", 1); + first[depth] = false; + put_string(t, (const unsigned char *)field->name + start, end - start); + put(t, ":{", 2); + first[++depth] = true; + memcpy(open, field->name, end); + open[open_size = end] = 0; + } + if (!first[depth]) put(t, ",", 1); + first[depth] = false; + const char *key = field->name + (group ? group + 1 : 0); + put_string(t, (const unsigned char *)key, strlen(key)); + put(t, ":", 1); + char digits[32]; + if (field->type == PURE_FIELD_U32) { + put(t, digits, (size_t)snprintf(digits, sizeof(digits), "%u", (unsigned)read_u32(payload.data + field->offset))); + } else if (field->type == PURE_FIELD_NUMBER) { + uint64_t bits = read_u32(payload.data + field->offset) | (uint64_t)read_u32(payload.data + field->offset + 4) << 32; + double value; + memcpy(&value, &bits, sizeof(value)); + put(t, digits, pure_format_number(value, digits)); + } else { + size_t size = field->offset == PURE_FIELD_REST ? payload.size - at : read_u32(payload.data + field->offset); + if (size > payload.size - at) return false; + PureSpan value = {payload.data + at, size}; + if (!pure_utf8_valid(value)) return false; /* bytes that are not text have no JSON form */ + put_string(t, value.data, size); + at += size; + } + } + while (depth--) put(t, "}", 1); + put(t, "}", 1); + return !t->failed; +} + +/* ---- The wire. */ + +PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source, + char *error, size_t size) { + int index = pure_module_input(m, output); + if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) { + snprintf(error, size, "no output '%s'", output); + return NULL; + } + /* A group's fields are together, so each group is one object. */ + const PureInputShape *shape = &m->input[index]; + for (uint32_t f = 1; f < shape->field_count; ++f) { + const char *name = m->field[shape->field_start + f].name; + for (uint32_t e = 0; e + 1 < f; ++e) { + const char *earlier = m->field[shape->field_start + e].name, *between = m->field[shape->field_start + f - 1].name; + size_t shared = 0; + while (earlier[shared] && earlier[shared] == name[shared]) ++shared; + while (shared && earlier[shared - 1] != '.') --shared; /* the groups both are in */ + if (shared && strncmp(between, name, shared)) { + snprintf(error, size, "the fields of output '%s' in one group are not together", output); + return NULL; + } + } + } + size_t length = strlen(source); + if (!length || length > PURE_SOURCE) { + snprintf(error, size, "a source name has 1 to %d bytes", PURE_SOURCE); + return NULL; + } + if (strncmp(url, "http://", 7)) { + snprintf(error, size, strncmp(url, "https://", 8) ? "a URL is http://host[:port]/path" : + "https is not delivered to yet: use http"); + return NULL; + } + const char *host = url + 7, *slash = strchr(host, '/'), *colon = memchr(host, ':', slash ? (size_t)(slash - host) : strlen(host)); + const char *host_end = colon ? colon : slash ? slash : host + strlen(host); + const char *port = colon ? colon + 1 : "80"; + size_t port_size = colon ? (size_t)((slash ? slash : host + strlen(host)) - port) : 2; + if (host_end == host || (size_t)(host_end - host) >= 256 || !port_size || port_size >= 8 || + (slash && strlen(slash) >= 1024)) { + snprintf(error, size, "a URL is http://host[:port]/path"); + return NULL; + } + PureWire *w = calloc(1, sizeof(*w)); + if (!w) { snprintf(error, size, "out of memory"); return NULL; } + w->module = m; + w->output = (uint32_t)index; + memcpy(w->host, host, (size_t)(host_end - host)); + memcpy(w->port, port, port_size); + snprintf(w->path, sizeof(w->path), "%s", slash ? slash : "/"); + memcpy(w->source, source, length + 1); + w->fd = -1; + pthread_mutex_init(&w->lock, NULL); + pthread_cond_init(&w->changed, NULL); + return w; +} + +uint32_t pure_wire_output(const PureWire *w) { return w->output; } +uint64_t pure_wire_queued(const PureWire *w) { return w->queued; } + +uint64_t pure_wire_delivered(PureWire *w) { + pthread_mutex_lock(&w->lock); + uint64_t delivered = w->delivered; + pthread_mutex_unlock(&w->lock); + return delivered; +} + +void pure_wire_counts(PureWire *w, uint64_t *delivered, uint64_t *attempts) { + pthread_mutex_lock(&w->lock); + *delivered = w->deliveries; + *attempts = w->attempts; + pthread_mutex_unlock(&w->lock); +} + +bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) { + if (w->stuck) return false; + Text t = {0}; + if (!encode(w, payload, &t)) { + free(t.bytes); + w->stuck = true; + fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"no JSON form: its bytes are not text\"}\n", w->source, + (unsigned long long)seq); + return false; + } + pthread_mutex_lock(&w->lock); + if (w->head + w->count == w->capacity) { + if (w->head) { + memmove(w->queue, w->queue + w->head, w->count * sizeof(*w->queue)); + w->head = 0; + } else { + size_t capacity = w->capacity ? 2 * w->capacity : 16; + Pending *grown = realloc(w->queue, capacity * sizeof(*grown)); + if (!grown) { + pthread_mutex_unlock(&w->lock); + free(t.bytes); + return false; /* it stays in the outbox: queued after the next save */ + } + w->queue = grown; + w->capacity = capacity; + } + } + w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq}; + w->queued = seq; + pthread_cond_signal(&w->changed); + pthread_mutex_unlock(&w->lock); + return true; +} + +/* ---- HTTP/1.1, one request at a time on a kept connection. */ + +typedef struct { int fd; char bytes[4096]; size_t at, end; } Reader; + +static bool fill(Reader *r) { + if (r->at == r->end) r->at = r->end = 0; + if (r->end == sizeof(r->bytes)) return false; + ssize_t n; + do n = recv(r->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR); + if (n <= 0) return false; + r->end += (size_t)n; + return true; +} + +/* A line without its CRLF, in line (at most size - 1 bytes). */ +static bool read_line(Reader *r, char *line, size_t size) { + for (;;) { + char *end = memchr(r->bytes + r->at, '\n', r->end - r->at); + if (end) { + size_t length = (size_t)(end - (r->bytes + r->at)); + if (length && end[-1] == '\r') --length; + if (length >= size) return false; + memcpy(line, r->bytes + r->at, length); + line[length] = 0; + r->at = (size_t)(end - r->bytes) + 1; + return true; + } + if (r->at) { + memmove(r->bytes, r->bytes + r->at, r->end - r->at); + r->end -= r->at; + r->at = 0; + } + if (!fill(r)) return false; + } +} + +static bool skip(Reader *r, uint64_t size) { + while (size) { + if (r->at == r->end && !fill(r)) return false; + size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size; + r->at += n; + size -= n; + } + return true; +} + +static bool send_all(int fd, const char *bytes, size_t size) { + while (size) { + ssize_t n = send(fd, bytes, size, MSG_NOSIGNAL); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) return false; + bytes += n; + size -= (size_t)n; + } + return true; +} + +static int connect_to(const PureWire *w) { + struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM}, *list; + if (getaddrinfo(w->host, w->port, &hints, &list)) return -1; + int fd = -1; + for (struct addrinfo *a = list; a && fd < 0; a = a->ai_next) { + fd = socket(a->ai_family, a->ai_socktype | SOCK_CLOEXEC, a->ai_protocol); + if (fd < 0) continue; + struct timeval limit = {TIMEOUT_S, 0}; + int one = 1; + setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &limit, sizeof(limit)); + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &limit, sizeof(limit)); + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)); + if (connect(fd, a->ai_addr, a->ai_addrlen)) { + close(fd); + fd = -1; + } + } + freeaddrinfo(list); + return fd; +} + +static void drop_connection(PureWire *w) { + pthread_mutex_lock(&w->lock); + if (w->fd >= 0) close(w->fd); + w->fd = -1; + pthread_mutex_unlock(&w->lock); +} + +/* A fault a Pure receiver answers when its app refused the message. */ +static bool refusal(const char *error) { + static const char *const faults[] = {"divide", "non_finite", "clamp", "limit", "not_found", "encoding", "exists"}; + for (size_t i = 0; i < sizeof(faults) / sizeof(*faults); ++i) if (!strcmp(error, faults[i])) return true; + return false; +} + +/* The answer to one request: its status, whether a Pure receiver refused it, + * and whether the connection may carry the next. */ +static bool read_answer(int fd, int *status, bool *refused, bool *keep, unsigned *batch) { + Reader r = {.fd = fd}; + char line[1024]; + if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false; + *status = atoi(line + 9); + *keep = line[7] == '1'; + *refused = false; + uint64_t length = 0; + bool sized = false, chunked = false; + while (read_line(&r, line, sizeof(line))) { + if (!*line) { + if (chunked) { + for (;;) { + uint64_t chunk; + if (!read_line(&r, line, sizeof(line))) return false; + chunk = strtoull(line, NULL, 16); + if (!chunk) break; + if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false; + } + while (read_line(&r, line, sizeof(line)) && *line) {} + return true; + } + if (*status == 204 || *status == 304 || *status < 200) return true; + if (sized) return skip(&r, length); + *keep = false; /* the body runs to the close */ + while (fill(&r)) r.at = r.end; + return true; + } + char *value = strchr(line, ':'); + if (!value) return false; + *value++ = 0; + while (*value == ' ' || *value == '\t') ++value; + if (!strcasecmp(line, "Content-Length")) { length = strtoull(value, NULL, 10); sized = true; } + else if (!strcasecmp(line, "Transfer-Encoding")) chunked = !strcasecmp(value, "chunked"); + else if (!strcasecmp(line, "Connection")) *keep = strcasecmp(value, "close") != 0 && *keep; + else if (!strcasecmp(line, "X-Pure-Error")) *refused = refusal(value); + else if (!strcasecmp(line, "X-Pure-Batch")) *batch = (unsigned)strtoul(value, NULL, 10); + } + return false; +} + +/* One request for count sends from p (numbered in a row): one object, or + * an array of them for a receiver that takes batches. */ +static int post(PureWire *w, const Pending *p, unsigned count) { + size_t body = count > 1 ? count + 1 : 0; + for (unsigned i = 0; i < count; ++i) body += p[i].size; + char head[2048], batch[32] = ""; + if (count > 1) snprintf(batch, sizeof(batch), "X-Pure-Count: %u\r\n", count); + int size = snprintf(head, sizeof(head), + "POST %s HTTP/1.1\r\nHost: %s:%s\r\nContent-Type: application/json\r\nContent-Length: %zu\r\n" + "X-Pure-Source: %s\r\nX-Pure-Sequence: %llu\r\n%sIdempotency-Key: %s-%llu\r\n\r\n", + w->path, w->host, w->port, body, w->source, (unsigned long long)p->seq, batch, w->source, + (unsigned long long)p->seq); + /* A kept connection the peer closed is tried once more on a new one. */ + for (int attempt = 0; attempt < 2; ++attempt) { + bool fresh = w->fd < 0; + if (fresh) { + int fd = connect_to(w); + if (fd < 0) return AGAIN; + pthread_mutex_lock(&w->lock); + w->fd = fd; + bool stopping = w->stopping; + pthread_mutex_unlock(&w->lock); + if (stopping) return AGAIN; + } + int status; + bool refused, keep, sent = send_all(w->fd, head, (size_t)size); + unsigned batch_size = 0; + for (unsigned i = 0; sent && i < count; ++i) + sent = (!i || count == 1 || send_all(w->fd, ",", 1)) && (i || count == 1 || send_all(w->fd, "[", 1)) && + send_all(w->fd, p[i].body, p[i].size); + if (sent && count > 1) sent = send_all(w->fd, "]", 1); + if (!sent || !read_answer(w->fd, &status, &refused, &keep, &batch_size)) { + drop_connection(w); + if (fresh) return AGAIN; + continue; + } + if (!keep) drop_connection(w); + if (batch_size) w->batch = batch_size; + if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED; + if (status == 408 || status == 429 || status >= 500) return AGAIN; + return REFUSED_BY_PEER; + } + return AGAIN; +} + +static void *deliver(void *context) { + PureWire *w = context; + long pause_ms = 0; + uint64_t reported = 0; + pthread_mutex_lock(&w->lock); + for (;;) { + while (!w->stopping && !w->count) pthread_cond_wait(&w->changed, &w->lock); + if (w->stopping) break; + /* As many as the receiver takes at once; their bodies stay put while + * the queue only grows behind them. */ + unsigned count = w->batch > 1 ? (unsigned)(w->count < w->batch ? w->count : w->batch) : 1; + Pending next[BATCH_MOST]; + if (count > BATCH_MOST) count = BATCH_MOST; + memcpy(next, w->queue + w->head, count * sizeof(*next)); + ++w->attempts; + pthread_mutex_unlock(&w->lock); + int outcome = post(w, next, count); + pthread_mutex_lock(&w->lock); + if (outcome == DELIVERED) { + for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body); + w->head += count; + w->count -= count; + w->delivered = next[count - 1].seq; + w->deliveries += count; + pause_ms = 0; + pthread_mutex_unlock(&w->lock); + if (w->notify) w->notify(w->context); + pthread_mutex_lock(&w->lock); + continue; + } + if (outcome == REFUSED_BY_PEER && reported != next[0].seq) { + fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"the receiver does not take it; trying again\"}\n", + w->source, (unsigned long long)next[0].seq); + reported = next[0].seq; + } + pause_ms = outcome == REFUSED_BY_PEER ? LONGEST_PAUSE_MS : pause_ms ? 2 * pause_ms : FIRST_PAUSE_MS; + if (pause_ms > LONGEST_PAUSE_MS) pause_ms = LONGEST_PAUSE_MS; + struct timespec until; + clock_gettime(CLOCK_REALTIME, &until); + until.tv_sec += pause_ms / 1000; + until.tv_nsec += (pause_ms % 1000) * 1000000; + if (until.tv_nsec >= 1000000000) { ++until.tv_sec; until.tv_nsec -= 1000000000; } + while (!w->stopping && pthread_cond_timedwait(&w->changed, &w->lock, &until) == 0) {} + } + pthread_mutex_unlock(&w->lock); + return NULL; +} + +bool pure_wire_start(PureWire *w, void (*notify)(void *), void *context) { + w->notify = notify; + w->context = context; + w->started = pthread_create(&w->thread, NULL, deliver, w) == 0; + return w->started; +} + +void pure_wire_close(PureWire *w) { + if (!w) return; + pthread_mutex_lock(&w->lock); + w->stopping = true; + if (w->fd >= 0) shutdown(w->fd, SHUT_RDWR); /* a send or an answer waited for ends now */ + pthread_cond_broadcast(&w->changed); + pthread_mutex_unlock(&w->lock); + if (w->started) pthread_join(w->thread, NULL); + if (w->fd >= 0) close(w->fd); + for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body); + free(w->queue); + pthread_cond_destroy(&w->changed); + pthread_mutex_destroy(&w->lock); + free(w); +} diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h new file mode 100644 index 0000000..f720296 --- /dev/null +++ b/prototypes/connections/deliver.h @@ -0,0 +1,36 @@ +#ifndef PURE_DELIVER_H +#define PURE_DELIVER_H +#include "module.h" + +/* Delivering an app's saved sends over HTTP (docs/research/083-delivery-over-http.md). + * A wire takes one output's sends in order: each goes as a JSON object of + * its fields (groups as objects inside) by POST to a URL, with the headers + * X-Pure-Source (the sender's name for the output), X-Pure-Sequence and + * Idempotency-Key, one at a time, again and again until it is delivered: + * a 2xx answer, or a Pure receiver's refusal (a 4xx with X-Pure-Error + * naming a fault: it heard the message and refused it). 408, 429, 5xx and + * no answer are tried again after a pause (0.1 s doubling to 10 s); another + * 4xx is logged and tried again every 10 s. Nothing is ever dropped: what is + * not delivered stays in the app's outbox, and a receiver applies a number + * once, so a send delivered twice (its answer lost) is harmless. */ +typedef struct PureWire PureWire; + +/* A wire for the output named output of m to url (http://host[:port]/path), + * as source; NULL with why in error when it cannot be. */ +PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source, + char *error, size_t size); +uint32_t pure_wire_output(const PureWire *); +/* The owner's side: queue a saved send numbered after the last queued one + * (it is encoded now); false when it cannot be (a bytes field that is not + * UTF-8 has no JSON form), and the wire then stops at it. */ +bool pure_wire_queue(PureWire *, uint64_t seq, PureSpan payload); +uint64_t pure_wire_queued(const PureWire *); +/* The last number delivered; any thread. */ +uint64_t pure_wire_delivered(PureWire *); +/* Deliver on a thread of the wire's own, calling notify after each delivery. */ +bool pure_wire_start(PureWire *, void (*notify)(void *), void *context); +/* Stop (a delivery under way is finished or abandoned) and free. */ +void pure_wire_close(PureWire *); +/* Counts for a host's report. */ +void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts); +#endif diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c index ee72494..5cb7f11 100644 --- a/prototypes/connections/module.c +++ b/prototypes/connections/module.c @@ -1676,7 +1676,8 @@ void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) { } } -PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload) { +PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload, + const uint32_t *placed) { const PureModule *m = app->module; size_t length = source ? strlen(source) : 0; app->duplicate = false; @@ -1700,7 +1701,7 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui app->delivering.source = source; app->delivering.seq = seq; app->delivering.pending = true; - PureResult result = message(app, input, NULL, -1, payload, NULL); + PureResult result = message(app, input, NULL, -1, payload, placed); if (!app->delivering.pending) return result; /* Not applied. Refused by the app (a fault), the message was delivered * all the same: its number is heard, saved like a key event's changes. */ diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h index ed40c15..3441c03 100644 --- a/prototypes/connections/module.h +++ b/prototypes/connections/module.h @@ -316,8 +316,9 @@ void pure_app_trim(PureApp *, uint32_t output, uint64_t through); * before is not applied again: it succeeds with app->duplicate set. The * number counts as heard with the event's changes, and also when the app * refuses the message (a fault: it was delivered, and refused); not when a - * save fails. */ -PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload); + * save fails. placed is as for pure_app_message_fields, or NULL. */ +PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload, + const uint32_t *placed); /* Readers on other threads. Right after a save an app can make a snapshot of * its saved states, for evaluating views anywhere while it keeps applying diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 0029164..2ecb780 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -29,6 +29,7 @@ * for a module with them, one more thread owns the app, and the connection * threads hand it their ready requests and answer what it hands back. Its * bursts take the requests of every thread, one save each. */ +#include "deliver.h" #include "module-store.h" #include "module-text.h" #include "runtime.h" @@ -57,7 +58,7 @@ #include <time.h> #include <unistd.h> -enum { HEADER_LIMIT = 4096, EVENTS = 256 }; +enum { HEADER_LIMIT = 4096, EVENTS = 256, BATCH = 256 }; typedef enum { CLOSED, READING_HEAD, READING_BODY, WAITING, WRITING } Phase; typedef struct Host Host; @@ -89,6 +90,9 @@ typedef struct Connection { bool given[PURE_MODULE_FIELDS], json; uint32_t input; unsigned char *ingress; + char source[PURE_SOURCE + 1]; /* a delivered message's sender (X-Pure-Source), or empty */ + uint64_t sequence; /* and its number (X-Pure-Sequence) */ + unsigned batch; /* how many numbered messages the body holds (X-Pure-Count) */ int64_t expected; size_t body_size, received, in_used; unsigned char *body; @@ -142,6 +146,10 @@ typedef struct { pthread_mutex_t published_lock; PureSnapshot *published; uint64_t published_changes, publications; + /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */ + PureWire *wires[PURE_MODULE_INPUTS]; + unsigned wire_count; + uint64_t trimmed[PURE_MODULE_INPUTS]; } Executor; /* A request made ready: its effect, what it names, and the thread that answers it. */ @@ -159,6 +167,7 @@ typedef struct { size_t count, capacity; uint64_t arrivals; /* requests handed to the owner, ever */ bool stopping; + bool delivered; /* a wire delivered a send: trim it */ Host *hosts; unsigned host_count; } Owner; @@ -269,10 +278,13 @@ static void respond_typed(Connection *c, int status, const char *reason, const c const char *type, const struct iovec *extents, size_t count, size_t size) { char tag[48] = ""; if (revision > 0) snprintf(tag, sizeof(tag), "ETag: \"%" PRId64 "\"\r\n", revision); + /* A delivery's answer tells how many numbered messages one request may carry. */ + char batch[32] = ""; + if (*c->source) snprintf(batch, sizeof(batch), "X-Pure-Batch: %d\r\n", BATCH); int length = snprintf(c->head, sizeof(c->head), "HTTP/1.1 %d %s\r\nContent-Length: %zu\r\n%sContent-Type: %s\r\n" - "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s\r\n", - status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", type, error, tag); + "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s%s\r\n", + status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", type, error, tag, batch); c->out = count < sizeof(c->small) / sizeof(*c->small) ? c->small : malloc((count + 1) * sizeof(*c->out)); c->phase = WRITING; if (!c->out) { c->keep_alive = false; c->out = c->small; count = 0; } /* cannot send it: the close tells */ @@ -396,6 +408,10 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) { if ((c->route == ROUTE_OBJECT && (c->post || *query)) || (c->route == ROUTE_VIEW && (c->put || c->post)) || (c->route == ROUTE_INPUT && !c->post)) return false; bool length_seen = false, condition_seen = false, host_seen = false; + c->source[0] = 0; + c->sequence = 0; + c->batch = 1; + bool counted = false; c->body_size = 0; c->json = false; c->expected = -1; @@ -437,16 +453,34 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) { } else if (!strcasecmp(line, "Connection")) { if (!strcasecmp(value, "close")) c->keep_alive = false; else if (strcasecmp(value, "keep-alive")) return false; + } else if (!strcasecmp(line, "X-Pure-Source")) { + /* A message delivered from another app's output: applied once per number. */ + size_t length = strlen(value); + if (*c->source || !length || length > PURE_SOURCE) return false; + for (size_t i = 0; i < length; ++i) if (value[i] <= ' ' || value[i] > '~') return false; + memcpy(c->source, value, length + 1); + } else if (!strcasecmp(line, "X-Pure-Sequence")) { + if (c->sequence || !number(value, UINT64_MAX, &n) || !n) return false; + c->sequence = n; + } else if (!strcasecmp(line, "X-Pure-Count")) { + /* Numbered messages from X-Pure-Sequence on, as a JSON array. */ + if (counted || !number(value, BATCH, &n) || !n) return false; + c->batch = (unsigned)n; + counted = true; } else if (!strcasecmp(line, "Transfer-Encoding") || !strcasecmp(line, "Expect")) return false; line = next + 2; } + if (!*c->source != !c->sequence || (*c->source && c->route != ROUTE_INPUT) || (counted && !*c->source) || + c->sequence > UINT64_MAX - c->batch) return false; if (c->put) return host_seen && length_seen && condition_seen; if (c->post) return host_seen && !condition_seen; /* no Content-Length: no body */ return host_seen && !condition_seen && c->body_size == 0; } static void put_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); } +static void put_u64(unsigned char *p, uint64_t v) { for (unsigned i = 0; i < 8; ++i) p[i] = (unsigned char)(v >> (8 * i)); } static uint32_t get_u32(const unsigned char *p) { return p[0] | p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; } +static uint64_t get_u64(const unsigned char *p) { return get_u32(p) | (uint64_t)get_u32(p + 4) << 32; } /* A JSON number's text: -?(0|[1-9][0-9]*)(.[0-9]+)?([eE][+-]?[0-9]+)?, its * length (0 if there is none). An integer has neither part after it. */ @@ -512,6 +546,8 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route * const PureInputShape *shape = route->kind == ROUTE_INPUT ? &m->input[found] : NULL; bool message = shape && shape->kind == PURE_KIND_MESSAGE; if (message && shape->keyed) return "keyed"; + if (*c->source && !message) return "delivery"; + if (!host->owner) c->source[0] = 0; /* an app without states or outputs applies it as a message */ if (shape) { /* A body is a text input's text, a message's one variable field, or a * JSON object of a message's fields (two variable fields need one). */ @@ -552,6 +588,10 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route * uint8_t type = m->field[shape->field_start + f].type; if ((type == PURE_FIELD_U32 || type == PURE_FIELD_NUMBER) && !c->given[f]) return "missing_field"; } + for (uint32_t f = 0; c->batch > 1 && f < shape->field_count; ++f) + if (c->given[f]) return "delivery"; /* a batch's fields are all in its objects */ + if (c->batch > 1 && !c->json) return "delivery"; + if (!*c->source) c->batch = 1; return NULL; } @@ -636,6 +676,12 @@ static size_t json_space(const unsigned char *p, size_t at, size_t size) { /* Parsing a JSON body in place: the payload it is in, the placed fields' * table, and which fields it gave. */ #define PLACED (UINT32_C(1) << 31) +/* After the size: a delivered message's number (u64le), how many follow it + * (u32le: a batch) and its source (u8 length and the bytes), in DELIVERY + * bytes before the tables; then a table for each message, a prefix for each + * and the JSON. */ +#define DELIVERED (UINT32_C(1) << 30) +enum { DELIVERY = 80 }; typedef struct { Connection *c; const PureInputShape *shape; @@ -735,16 +781,31 @@ static const char *members(Body *b, unsigned char *json, size_t size, size_t *at * NULL, or the error to answer. */ static const char *convert(Connection *c) { const PureInputShape *shape = &c->host->module->input[c->input]; - Body b = {.c = c, .shape = shape, .table = c->ingress + 4}; - b.payload = b.table + 8 * shape->field_count; + uint32_t delivered = get_u32(c->ingress) & DELIVERED; + size_t table = 8 * (size_t)shape->field_count; + unsigned char *tables = c->ingress + 4 + (delivered ? DELIVERY : 0), *prefixes = tables + c->batch * table; unsigned char *json = c->body; size_t size = c->body_size, at = json_space(json, 0, size); - const char *why = members(&b, json, size, &at, "", 0); - if (why) return why; + /* A batch is an array of its messages' objects. */ + if (c->batch > 1) { + if (at == size || json[at++] != '[') return "json"; + at = json_space(json, at, size); + } + for (unsigned i = 0; i < c->batch; ++i) { + if (i) { + if (at == size || json[at++] != ',') return "json"; + at = json_space(json, at, size); + } + Body b = {.c = c, .shape = shape, .table = tables + i * table, .payload = prefixes + i * c->field_bytes}; + const char *why = members(&b, json, size, &at, "", 0); + if (why) return why; + at = json_space(json, at, size); + for (uint32_t f = 0; f < shape->field_count; ++f) + if (!b.seen[f] && !c->given[f]) return "missing_field"; + } + if (c->batch > 1 && (at == size || json[at++] != ']')) return "json"; if (json_space(json, at, size) != size) return "json"; - for (uint32_t f = 0; f < shape->field_count; ++f) - if (!b.seen[f] && !c->given[f]) return "missing_field"; - put_u32(c->ingress, (uint32_t)(c->field_bytes + c->body_size) | PLACED); + put_u32(c->ingress, (uint32_t)(c->batch * c->field_bytes + c->body_size) | PLACED | delivered); return NULL; } @@ -781,7 +842,10 @@ static void start_request(Connection *c, size_t head_size) { * the payload (the fixed fields, then the body), and room for its JSON. */ bool write = route.kind == ROUTE_INPUT || c->put; size_t table = route.kind == ROUTE_INPUT && c->json ? 8 * (size_t)host->module->input[c->input].field_count : 0; - size_t input_size = route.kind == ROUTE_INPUT ? 4 + table + c->field_bytes + c->body_size : write ? c->body_size : 0; + size_t lead = 4 + (route.kind == ROUTE_INPUT && *c->source ? DELIVERY : 0); + /* A batch has a table and a prefix for each of its messages. */ + size_t input_size = route.kind == ROUTE_INPUT ? lead + c->batch * (table + c->field_bytes) + c->body_size : + write ? c->body_size : 0; PureResult admitted = pure_runtime_accept(&host->runtime, write ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key, route.kind == ROUTE_INPUT ? 0 : c->expected, input_size, &c->ticket); if (admitted != PURE_OK) { @@ -797,9 +861,16 @@ static void start_request(Connection *c, size_t head_size) { unsigned char *ingress = c->ingress = pure_runtime_ingress(c->ticket, NULL); /* NULL for no bytes */ c->body = ingress; if (route.kind == ROUTE_INPUT) { - put_u32(ingress, (uint32_t)(c->field_bytes + c->body_size)); - if (c->field_bytes) memcpy(ingress + 4 + table, c->fields, c->field_bytes); - c->body = ingress + 4 + table + c->field_bytes; + put_u32(ingress, (uint32_t)(c->batch * c->field_bytes + c->body_size) | (lead > 4 ? DELIVERED : 0)); + if (lead > 4) { + size_t length = strlen(c->source); + put_u64(ingress + 4, c->sequence); + put_u32(ingress + 12, c->batch); + ingress[16] = (unsigned char)length; + memcpy(ingress + 17, c->source, length); + } + if (c->field_bytes) memcpy(ingress + lead + c->batch * table, c->fields, c->field_bytes); + c->body = ingress + lead + c->batch * (table + c->field_bytes); } size_t take = c->in_used < c->body_size ? c->in_used : c->body_size; if (take) memcpy(c->body, c->in, take); @@ -1072,8 +1143,10 @@ static PureResult show_view(PureApp *app, const Route *route, size_t available, static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) { const PureInputShape *shape = &app->module->input[route->target]; uint32_t header = get_u32(effect->value.data); - size_t table = header & PLACED ? 8 * (size_t)shape->field_count : 0; - PureSpan payload = {effect->value.data + 4 + table, header & ~PLACED}; + const unsigned char *delivery = effect->value.data + 4; + unsigned batch = header & DELIVERED ? get_u32(delivery + 8) : 1; + size_t lead = 4 + (header & DELIVERED ? DELIVERY : 0), table = header & PLACED ? 8 * (size_t)shape->field_count : 0; + PureSpan payload = {effect->value.data + lead + batch * table, header & ~(PLACED | DELIVERED)}; PureResult status; if (shape->kind == PURE_KIND_TEXT) { status = pure_app_text(app, route->target, payload); @@ -1083,9 +1156,24 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect } else { uint32_t placed[2 * PURE_MODULE_FIELDS]; for (size_t i = 0; table && i < 2 * (size_t)shape->field_count; ++i) - placed[i] = get_u32(effect->value.data + 4 + 4 * i); - status = table ? pure_app_message_fields(app, route->target, NULL, -1, payload, placed) : - pure_app_message(app, route->target, NULL, -1, payload); + placed[i] = get_u32(effect->value.data + lead + 4 * i); + if (header & DELIVERED) { + char source[PURE_SOURCE + 1]; + memcpy(source, delivery + 13, delivery[12]); + source[delivery[12]] = 0; + status = pure_app_deliver(app, route->target, source, get_u64(delivery), payload, table ? placed : NULL); + /* The rest of a batch, in order: an app's refusal was heard and the + * batch goes on; another failure stops it, to be sent again. */ + for (unsigned i = 1; i < batch && (status == PURE_OK || app->fault != PURE_FAULT_NONE); ++i) { + const unsigned char *at = effect->value.data + lead + i * table; + for (size_t k = 0; k < 2 * (size_t)shape->field_count; ++k) placed[k] = get_u32(at + 4 * k); + PureSpan next = {payload.data + i * shape->prefix, payload.size - i * shape->prefix}; + status = pure_app_deliver(app, route->target, source, get_u64(delivery) + i, next, placed); + } + if (batch > 1 && app->fault != PURE_FAULT_NONE) status = PURE_OK; + } else + status = table ? pure_app_message_fields(app, route->target, NULL, -1, payload, placed) : + pure_app_message(app, route->target, NULL, -1, payload); Reply reply = {0}; if (status == PURE_OK) pure_app_reply(app, measure_reply, &reply); out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size}; @@ -1154,6 +1242,35 @@ static void publish(Executor *ex) { ++ex->publications; } +/* The saved sends of wired outputs not queued yet go to their wires, in order; + * a wire that cannot take one takes the rest at a later save. */ +typedef struct { Executor *executor; bool blocked[PURE_MODULE_INPUTS]; } Queueing; + +static bool queue_send(void *context, uint32_t output, uint64_t seq, PureSpan payload) { + Queueing *q = context; + Executor *ex = q->executor; + for (unsigned i = 0; i < ex->wire_count; ++i) + if (pure_wire_output(ex->wires[i]) == output && !q->blocked[i] && seq > pure_wire_queued(ex->wires[i])) + q->blocked[i] = !pure_wire_queue(ex->wires[i], seq, payload); + return true; +} + +static void after_save(Executor *ex) { + publish(ex); + Queueing q = {.executor = ex}; + if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q); +} + +/* What the wires delivered leaves the outbox (with the next save). */ +static void trim_delivered(Executor *ex) { + for (unsigned i = 0; i < ex->wire_count; ++i) { + uint64_t delivered = pure_wire_delivered(ex->wires[i]); + if (delivered <= ex->trimmed[i]) continue; + pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered); + ex->trimmed[i] = delivered; + } +} + /* One save for a burst: its requests are durable with it or fail with it. */ static void save_burst(Executor *ex, Applied *applied, size_t count) { if (!count) return; @@ -1161,7 +1278,7 @@ static void save_burst(Executor *ex, Applied *applied, size_t count) { ex->wrote = ex->wrote || (applied[i].status == PURE_OK && applied[i].input == PURE_INPUT_WRITE); PureResult saved = pure_app_save(&ex->app); ++ex->bursts; - if (saved == PURE_OK) publish(ex); + if (saved == PURE_OK) after_save(ex); for (size_t i = 0; i < count; ++i) if (applied[i].status == PURE_OK) applied[i].status = saved; } @@ -1230,7 +1347,7 @@ static long run_timers(Executor *ex) { } if (soonest < 0 || ex->next[i] - now < soonest) soonest = ex->next[i] - now; } - if (ticked && pure_app_save(&ex->app) == PURE_OK) publish(ex); /* a failed save undoes the ticks */ + if (ticked && pure_app_save(&ex->app) == PURE_OK) after_save(ex); /* a failed save undoes the ticks */ return soonest < 0 ? -1 : (long)ceil(soonest); } @@ -1376,12 +1493,17 @@ static void *run_owner(void *context) { uint64_t answered_at = 0; /* arrivals when the last round's answers were handed back */ for (;;) { pthread_mutex_lock(&owner->lock); - /* Wait for requests, or until the next timer is due. */ + /* Wait for requests, or until the next timer is due; trim what the + * wires delivered meanwhile. */ for (;;) { + bool delivered = owner->delivered; + owner->delivered = false; pthread_mutex_unlock(&owner->lock); long due = run_timers(&owner->executor); + if (delivered) trim_delivered(&owner->executor); pthread_mutex_lock(&owner->lock); if (owner->count || owner->stopping) break; + if (owner->delivered) continue; if (due < 0) { pthread_cond_wait(&owner->waiting, &owner->lock); continue; @@ -1654,6 +1776,44 @@ static void stop_shard(Host *host) { free(host->views); } +/* A wire delivered a send: the owner trims it. */ +static void wire_delivered(void *context) { + Owner *owner = context; + pthread_mutex_lock(&owner->lock); + owner->delivered = true; + pthread_cond_signal(&owner->waiting); + pthread_mutex_unlock(&owner->lock); +} + +/* --send OUTPUT=URL: that output's saved sends go to url, from NAME.OUTPUT. */ +static bool open_wires(Executor *ex, const PureModule *m, const char *const *sends, unsigned count, const char *name) { + for (unsigned i = 0; i < count; ++i) { + const char *equals = strchr(sends[i], '='); + char output[PURE_MODULE_NAME], source[PURE_SOURCE + 2], error[160]; + size_t length = equals ? (size_t)(equals - sends[i]) : 0; + if (!length || length >= sizeof(output)) { + fprintf(stderr, "--send takes OUTPUT=URL, not %s\n", sends[i]); + return false; + } + memcpy(output, sends[i], length); + output[length] = 0; + int written = snprintf(source, sizeof(source), "%s.%s", name, output); + PureWire *wire = written > 0 && written <= PURE_SOURCE ? pure_wire_open(m, output, equals + 1, source, error, sizeof(error)) : NULL; + if (!wire) { + fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error); + return false; + } + for (unsigned k = 0; k < ex->wire_count; ++k) + if (pure_wire_output(ex->wires[k]) == pure_wire_output(wire)) { + fprintf(stderr, "--send %s: the output is wired already\n", sends[i]); + pure_wire_close(wire); + return false; + } + ex->wires[ex->wire_count++] = wire; + } + return true; +} + /* --threads auto: a connection thread for each CPU this process may run on, * less the one the owner of an app with states takes. */ static unsigned auto_threads(bool owner) { @@ -1667,7 +1827,8 @@ static unsigned auto_threads(bool owner) { int main(int argc, char **argv) { static PureModule module; static Owner owner; - const char *module_path = NULL, *store = NULL; + const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS]; + unsigned send_count = 0; unsigned port = 8767, threads = 1; size_t budget = 64u << 20, capacity = 1024; uint64_t timeout_ms = 10000; @@ -1676,6 +1837,8 @@ int main(int argc, char **argv) { if (i + 1 == argc) goto usage; if (!strcmp(argv[i], "--module")) module_path = argv[i + 1]; else if (!strcmp(argv[i], "--store")) store = argv[i + 1]; + else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1]; + else if (!strcmp(argv[i], "--name")) name = argv[i + 1]; else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value; else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value; else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value; @@ -1697,16 +1860,30 @@ int main(int argc, char **argv) { Host *hosts = calloc(threads, sizeof(*hosts)); pthread_t *thread = calloc(threads, sizeof(*thread)), owner_thread; if (!hosts || !thread) return 1; - /* States have one owner on a thread of its own, whose snapshots the - * connection threads read. */ - bool owned = module.states; + /* States (and outputs, whose sends are the app's state) have one owner on + * a thread of its own, whose snapshots the connection threads read. */ + bool outputs = false; + for (uint32_t i = 0; i < module.inputs; ++i) outputs = outputs || module.input[i].kind == PURE_KIND_OUTPUT; + bool owned = module.states || outputs; + /* The sender's name: --name, or the module file's name without .pmod. */ + char stem[PURE_SOURCE + 1]; + if (!name) { + const char *base = strrchr(module_path, '/') ? strrchr(module_path, '/') + 1 : module_path; + size_t length = strlen(base); + if (length > 5 && !strcmp(base + length - 5, ".pmod")) length -= 5; + snprintf(stem, sizeof(stem), "%.*s", (int)length, base); + name = stem; + } if (owned) { if (!start_executor(&owner.executor, &module, show, edit, store) || pthread_mutex_init(&owner.lock, NULL) || pthread_cond_init(&owner.waiting, NULL) || pthread_mutex_init(&owner.executor.published_lock, NULL)) return 1; owner.executor.reads_first = true; owner.executor.publishes = true; - publish(&owner.executor); + if (!open_wires(&owner.executor, &module, sends, send_count, name)) return 2; + after_save(&owner.executor); /* the snapshot, and the sends stored before */ + for (unsigned i = 0; i < owner.executor.wire_count; ++i) + if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; } owner.hosts = hosts; owner.host_count = threads; } @@ -1746,6 +1923,14 @@ int main(int argc, char **argv) { expired += hosts[i].expired; stop_shard(&hosts[i]); } + uint64_t delivered = 0, attempts = 0; + for (unsigned i = 0; i < owner.executor.wire_count; ++i) { + uint64_t d, a; + pure_wire_counts(owner.executor.wires[i], &d, &a); + delivered += d; + attempts += a; + pure_wire_close(owner.executor.wires[i]); /* what was not delivered stays in the outbox */ + } if (owned) { stop_executor(&owner.executor); free(owner.jobs); @@ -1755,14 +1940,14 @@ int main(int argc, char **argv) { fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64 ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64 ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64 - ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n", + ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n", active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false", - owner.executor.publications, shown, groups, grouped); + owner.executor.publications, shown, delivered, attempts, groups, grouped); free(hosts); free(thread); return 0; usage: fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] " - "[--connections 1024] [--timeout-ms 10000]\n", stderr); + "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http://host:port/path ...] [--name NAME]\n", stderr); return 2; } diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c index bb59174..ccb3371 100644 --- a/prototypes/connections/test-module.c +++ b/prototypes/connections/test-module.c @@ -2581,7 +2581,7 @@ static PureResult deliver(PureApp *app, const char *source, uint64_t seq, const memcpy(payload, &count, 4); memcpy(payload + 4, &total, 8); memcpy(payload + 12, item, length); - PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length}); + PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length}, NULL); pure_app_finish(app); return result; } @@ -2618,10 +2618,10 @@ static void links(void) { Outbox o = outbox_of(&a); CHECK(o.count == 1 && o.items[0].output == 2 && o.items[0].seq == 1 && o.items[0].size == 15 && a.values[0] == 1); PureSpan sent = {o.items[0].payload, o.items[0].size}; - CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && !b.duplicate && b.values[0] == 2 && + CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent, NULL) == PURE_OK && !b.duplicate && b.values[0] == 2 && !strcmp(state_text(&b, 1), "pen")); pure_app_finish(&b); - CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && b.duplicate && b.values[0] == 2); + CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent, NULL) == PURE_OK && b.duplicate && b.values[0] == 2); CHECK(deliver(&b, "other", 1, "cup", 5) == PURE_OK && !b.duplicate && b.values[0] == 7); CHECK(deliver(&b, "orders.ordered", 3, "hat", 1) == PURE_OK && b.values[0] == 8); /* numbers may skip */ CHECK(deliver(&b, "orders.ordered", 2, "old", 1) == PURE_OK && b.duplicate && b.values[0] == 8); @@ -2630,7 +2630,7 @@ static void links(void) { CHECK(deliver(&b, "other", 2, "big", 1) == PURE_OK && b.duplicate && b.values[0] == 8); /* Bad deliveries change nothing. */ CHECK(deliver(&b, "", 4, "x", 1) == PURE_INVALID && deliver(&b, "a b", 4, "x", 1) == PURE_INVALID); - CHECK(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent) == PURE_INVALID); + CHECK(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent, NULL) == PURE_INVALID); CHECK(pure_app_message(&a, 2, NULL, -1, sent) == PURE_INVALID); /* Trimmed, a send is gone at once for the host and from the store at the next save. */ pure_app_trim(&a, 2, 1); |