From 5ce7655d613689223050715ff3f8fe5c456ebf11 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Wed, 30 Sep 2026 05:58:38 +0100 Subject: Take a delivery's fields by name, dropping none A message from another node was taken only in its input's exact shape (the link's STREAM shape, fields and order alike; over HTTP, every field). So a board whose nodes are upgraded one after the other stopped both ways until both were, and an output and its input whose fields differ only in order were refused for good (docs/design/014). An input now takes a message when it has every field the sender sends, by name, of the same type or one that holds its values (a u32 in a number, text in bytes); a field only the input has is 0 or empty. One that sends a field the input lacks is refused: taking it would drop the value, and with its number heard the value would never come again (two boards apart for good). So add fields to receivers first and take them from senders first, and nothing stops or is lost. The link carries a sender's fields in a new frame, FIELDS (0x0d: names, shape, u8 count, each {type, name length, name}): after a REFUSED for another shape the sender puts it in its packets (after OFFER, before the records, so a receiver that knows no FIELDS reads what it read before) until the stream is taken, the first time at once. A receiver seeing another shape waits for the packet's FIELDS before refusing, and asks its host with them (PureLinkHost.input takes them); REFUSED says 4 when the input lacks a field. Deliveries carry the fields and the host lays each record out anew (pure_module_takes, pure_module_recast, pure_module_fields, over pure_send_recast's rule). A stream whose sender starts again in another shape lets go what it held; the sender sends it again laid out anew. A refusal already queued for a stream now says the last reason judged, not the first. Over HTTP a delivery (X-Pure-Source) may lack fields; an unknown member is still refused. test-link named: a stream of another shape naming its fields is taken, 200 stored in 47 ms with every delivery naming them; an input that does not take them refuses (4) and the sender backs off. check.py server_fields: five outputs, three linked and two over HTTP, fields in another order taken, an input with a field more taking messages without it, one lacking a field refusing until it starts again with it, then taking all. server_upgrade's new receiver declares its fields in another order. 012 names the frame and the rules. --- core/module.c | 65 +++++++++++++ core/module.h | 18 +++- docs/design/012-link-protocol.md | 20 +++- docs/design/014-fields-by-name.md | 77 +++++++++++++++ link/link-protocol.c | 136 ++++++++++++++++++++++----- link/link-protocol.h | 26 ++++- link/link.c | 10 +- link/link.h | 4 + link/test-link.c | 84 ++++++++++++++++- serve/serve.c | 43 +++++++-- tests/check.py | 193 +++++++++++++++++++++++++++++++++++++- 11 files changed, 629 insertions(+), 47 deletions(-) create mode 100644 docs/design/014-fields-by-name.md diff --git a/core/module.c b/core/module.c index 97270ef..ea6809e 100644 --- a/core/module.c +++ b/core/module.c @@ -1211,6 +1211,71 @@ uint32_t pure_module_shape(const PureModule *m, uint32_t input) { return h; } +size_t pure_module_fields(const PureModule *m, uint32_t input, unsigned char out[PURE_FIELDS_NAMED]) { + const PureInputShape *s = &m->input[input]; + size_t at = 0; + out[at++] = (unsigned char)s->field_count; + for (uint32_t f = 0; f < s->field_count; ++f) { + const PureField *field = &m->field[s->field_start + f]; + size_t length = strlen(field->name); + out[at++] = field->type; + out[at++] = (unsigned char)length; + memcpy(out + at, field->name, length); + at += length; + } + return at; +} + +/* Fields as a sender names them, laid out: their count, or -1 when they are + * not well named (no rows: a sender sends none). */ +static int named_fields(const unsigned char *bytes, size_t size, PureField *fields) { + if (!size || bytes[0] > PURE_MODULE_FIELDS) return -1; + unsigned count = bytes[0]; + size_t at = 1; + for (unsigned f = 0; f < count; ++f) { + size_t length = size - at >= 2 ? bytes[at + 1] : 0; + if (size - at < 2 || !length || length >= PURE_MODULE_NAME || size - at < 2 + length || !bytes[at] || + bytes[at] >= PURE_FIELD_ROWS) + return -1; + fields[f] = (PureField){.type = bytes[at]}; + memcpy(fields[f].name, bytes + at + 2, length); + for (unsigned k = 0; k < f; ++k) + if (!strcmp(fields[k].name, fields[f].name)) return -1; + at += 2 + length; + } + if (at != size) return -1; + pure_fields_lay_out(fields, count); + return (int)count; +} + +bool pure_module_takes(const PureModule *m, uint32_t input, const unsigned char *fields, size_t size) { + PureField sent[PURE_MODULE_FIELDS]; + int count = named_fields(fields, size, sent); + if (count < 0) return false; + const PureInputShape *s = &m->input[input]; + for (int k = 0; k < count; ++k) { + uint32_t f = 0; + while (f < s->field_count && strcmp(m->field[s->field_start + f].name, sent[k].name)) ++f; + if (f == s->field_count) return false; /* its value would be dropped */ + uint8_t type = m->field[s->field_start + f].type, from = sent[k].type; + if (type != from && !(type == PURE_FIELD_NUMBER && from == PURE_FIELD_U32) && + !(type == PURE_FIELD_BYTES && from == PURE_FIELD_TEXT)) + return false; + } + return true; +} + +PureResult pure_module_recast(const PureModule *m, uint32_t input, const unsigned char *fields, size_t size, + PureSpan payload, unsigned char **out, size_t *out_size) { + PureField sent[PURE_MODULE_FIELDS]; + int count = named_fields(fields, size, sent); + *out = NULL; + *out_size = 0; + if (count < 0) return PURE_STORE_ERROR; + const PureInputShape *s = &m->input[input]; + return pure_send_recast(sent, (uint32_t)count, &m->field[s->field_start], s->field_count, payload, out, out_size); +} + int pure_module_view(const PureModule *m, const char *name) { for (uint32_t i = 0; i < m->views; ++i) if (!m->view_table[i] && !strcmp(m->view_names[i], name)) return (int)i; diff --git a/core/module.h b/core/module.h index ad31821..2cac3b3 100644 --- a/core/module.h +++ b/core/module.h @@ -217,8 +217,24 @@ int pure_module_view(const PureModule *, const char *name); int pure_module_column(const PureModule *, uint32_t table, const char *name); /* What a message of an input looks like, as 32 bits: its fields' names, * types and places (FNV-1a). A sender's output and the input it sends to - * across nodes must agree on it (link/link.h). */ + * across nodes agree on it, or the input takes the output's fields by name + * (link/link.h, docs/design/014-fields-by-name.md). */ uint32_t pure_module_shape(const PureModule *, uint32_t input); +/* An input's (or output's) fields as a sender names them to a receiver: u8 + * count, each field's {u8 type, u8 name length, name}. Its size (at most + * PURE_FIELDS_NAMED), written to out. */ +enum { PURE_FIELDS_NAMED = 1 + PURE_MODULE_FIELDS * (2 + PURE_MODULE_NAME) }; +size_t pure_module_fields(const PureModule *, uint32_t input, unsigned char out[PURE_FIELDS_NAMED]); +/* Whether input takes messages laid out with the fields named (as + * pure_module_fields names them): it has every one by name, of the same type + * or one that holds all its values (a u32 in a number, text in bytes). A + * field it has that they lack is 0 or empty; none of theirs is dropped. */ +bool pure_module_takes(const PureModule *, uint32_t input, const unsigned char *fields, size_t size); +/* A payload laid out with the fields named, laid out anew for input by + * name (pure_module_takes says it can be; out is malloc'd). STORE_ERROR + * when payload is not laid out with them. */ +PureResult pure_module_recast(const PureModule *, uint32_t input, const unsigned char *fields, size_t size, + PureSpan payload, unsigned char **out, size_t *out_size); /* The index state a number view shows as its position (a view that is just * `at.position`), or -1: what a host moves with pure_app_point. */ int pure_module_position(const PureModule *, uint32_t view); diff --git a/docs/design/012-link-protocol.md b/docs/design/012-link-protocol.md index 99d7a91..373c618 100644 --- a/docs/design/012-link-protocol.md +++ b/docs/design/012-link-protocol.md @@ -19,7 +19,8 @@ run은 0이 아닌 무작위 u64다. - **받는 쪽**: 받아들일 보내는 쪽을 (source 이름, 공개 키) 목록으로 가진다. - **스트림**: source 이름과 input 이름의 쌍이다. 각 이름은 1–64 B다. 메시지의 shape(u32)가 함께 간다. 받는 쪽 입력의 - shape와 같아야 한다. 링크에게 shape는 비교만 하는 값이고, 뜻은 모듈 형식([008](008-module-format.md))이 정한다. + shape와 같거나, 보내는 쪽이 FIELDS로 이름을 댄 필드를 입력이 모두 가져야 한다([014](014-fields-by-name.md)). 링크에게 + shape와 필드 목록은 비교하고 나르기만 하는 값이고, 뜻은 모듈 형식([008](008-module-format.md))이 정한다. - **기록**: 스트림 안에서 1부터 연속된 번호(u64)를 가진 바이트열이다. 0–16 MiB다. 바이트는 보내는 앱이 저장한 그대로이고, 링크는 읽지 않는다. - **들은 번호**: 받는 쪽 앱이 적용하고 저장한 마지막 기록 번호다. @@ -110,7 +111,8 @@ PADDING도 그렇게 끝을 채운다. 이름은 u8 길이(1–64)와 그 바이 | 0x09 | OFFER | 보냄 | u32 에폭, 공개 키 32 B | | 0x0a | ACCEPT | 답 | u32 에폭, 공개 키 32 B, 답하는 OFFER 공개 키의 앞 4 B | | 0x0b | LOST | 답 | u64 패킷 번호 | -| 0x0c | REFUSED | 답 | names, u8 이유(1: 그 이름의 메시지 입력이 없다, 2: 입력의 shape가 다르다, 3: 이 키는 그 source로 보낼 수 없다) | +| 0x0c | REFUSED | 답 | names, u8 이유(1: 그 이름의 메시지 입력이 없다, 2: 입력의 shape가 다르다, 3: 이 키는 그 source로 보낼 수 없다, 4: 입력이 보낸 필드를 다 갖지 않는다) | +| 0x0d | FIELDS | 보냄 | names, u32 shape, u8 필드 수, 필드마다 {u8 타입, u8 이름 길이, 이름} | - RECORD와 RECORDS는 그 앞의 가장 가까운 STREAM의 스트림에 속한다. 앞에 STREAM이 없으면 버린다. - RECORDS는 번호가 이어진 온전한 기록들이다. LEB128은 7비트씩, 낮은 것부터, 10 B까지다. 기록 하나가 패킷 끝을 넘으면 그 @@ -171,11 +173,16 @@ STORED의 들은 번호 이하인 기록은 덜어 낸다. 보내는 앱은 그 **거절.** 자기 스트림에 대한 REFUSED를 들으면 기록을 보내지 않는다. +- 이유가 2(다른 shape)나 4이고 필드를 댈 수 있으면, 그 뒤 패킷마다 FIELDS를 싣는다. 자리는 OFFER 다음, 기록 앞이다. 처음 + 이유 2를 들었을 때는 기다리지 않고 곧바로 FIELDS를 실은 패킷 하나로 시도한다. - 기다린 뒤 기록을 실은 패킷 하나로 다시 시도한다. - 기다림은 1초에서 시작해 거절될 때마다 두 배로 늘고, 64초에서 멈춘다. - 시도한 패킷을 확인한 답에 REFUSED가 없으면 거절이 끝난다. - STORED를 들어도 끝나며, 이때 기다림도 처음으로 돌아간다. - 시도한 패킷보다 앞선 패킷들에 대한 REFUSED는 무시한다. +- 시도한 패킷을 잃었다고 판단하거나 에폭을 잃어 날던 패킷을 모두 놓으면, 다음 패킷이 다시 시도한다. 받는 쪽이 다시 떠서 시도가 + 아무에게도 닿지 않은 경우다([105](../research/105-growing-a-shared-board.md)). +- 스트림이 받아들여지면 FIELDS를 그만 싣는다. **경로 MTU**(DPLPMTUD, RFC 8899). @@ -229,7 +236,12 @@ STORED의 들은 번호 이하인 기록은 덜어 낸다. 보내는 앱은 그 - 들은 번호 + 65,536보다 큰 기록은 버린다. 그 안의 기록도 메모리(16 MiB)를 넘으면 받지 않는다. 보내는 쪽은 저장 소식이 없으니 다시 보낸다. - 들은 번호 이하의 기록이 다시 오면 STORED를 다시 보낸다. -- 받는 쪽은 들은 번호 다음부터 번호 순서로 온전한 기록을 앱에 넘긴다. 앱이 적용하고 저장한 뒤에만 STORED의 들은 번호를 올린다. +- STREAM의 shape가 입력과 다르면 곧바로 거절하지 않고 그 패킷의 FIELDS를 본다. 같은 스트림의 FIELDS가 있고 입력이 그 필드를 + 모두 같은 이름, 같은 타입이나 넓은 타입(u32를 number로, text를 bytes로)으로 가지면 스트림을 받는다. 입력에만 있는 필드는 0이나 + 빈 값이 된다. 아니면 이유 4다. FIELDS가 없으면 이유 2다. 이미 있는 스트림에 다른 shape가 오면, 앱에 넘긴 전달이 없을 때 쥐던 + 기록을 놓고 새로 판단한다(보내는 쪽이 새 배치로 다시 보낸다). +- 받는 쪽은 들은 번호 다음부터 번호 순서로 온전한 기록을 앱에 넘긴다. FIELDS로 받은 스트림의 기록은 그 필드 목록과 함께 넘기고, + 앱이 입력의 배치로 옮겨 적용한다. 앱이 적용하고 저장한 뒤에만 STORED의 들은 번호를 올린다. 틈에서 멈추고, 틈 뒤의 기록을 버렸으면 깃발 1을 세운다. - 앱이 가득 차 더 받지 못하면(앱의 메모리 예산, [013](013-memory-budget.md)) 받는 쪽은 받지 못한 기록을 쥔 채 잠시 뒤 다시 넘긴다. 그동안 보내는 STORED에 깃발 2를 세운다. 다시 넘기는 간격은 지역 선택이지만, 보내는 쪽이 저장을 기다리는 1초보다 @@ -338,7 +350,7 @@ LOST는 그 패킷의 번호를 담아 에폭 0의 키로 봉인한다. 답 run ## 이 명세와 구현 - 구현은 `link/link-protocol.c`(상태, 소켓 없음)와 `link/link.c`(UDP, GSO/GRO, 스레드)다. -- 전송 코드는 C 3,161줄이다(프로토콜 2,574, UDP 587). 011의 한도는 5,000줄이다. +- 전송 코드는 C 3,272줄이다(프로토콜 2,670, UDP 602). 011의 한도는 5,000줄이다. - 이 명세를 쓰면서 구현의 구멍 셋을 찾아 고쳤다. 셋 다 `link/test-link.c`에 검사가 있다. 메모의 곁 자리와 허가의 크기까지 [102](../research/102-adoption-and-spec.md)에 있다. - 받는 쪽이 스트림을 받지 않을 때 보내는 쪽이 아무 말도 듣지 못해 매초 기록을 다시 보냈다. REFUSED를 더했다. diff --git a/docs/design/014-fields-by-name.md b/docs/design/014-fields-by-name.md new file mode 100644 index 0000000..6db07aa --- /dev/null +++ b/docs/design/014-fields-by-name.md @@ -0,0 +1,77 @@ +# 014 — 필드 이름으로 받는 전달 + +작성일: 2026-09-30. 상태: 구현. +[105](../research/105-growing-a-shared-board.md)(두 노드가 함께 쓰는 보드를 키우며)에서 시작한다. + +## 문제 + +다른 노드에서 온 메시지는 받는 입력과 모양이 똑같아야 받았다. 링크는 STREAM의 shape(필드의 이름과 타입, 그 순서까지)를 +입력의 것과 비교하고, 다르면 REFUSED(다른 모양)로 거절한다. HTTP 전달은 JSON이라 이름으로 읽지만, 빠진 필드가 있으면 400 +`missing_field`로 거절한다. + +- **한 노드씩 올리면 전달이 멈춘다.** 카드에 필드 하나(`who`)를 더한 보드에서, 먼저 올린 노드와 아직 옛 노드가 서로의 + 보내기를 모두 거절한다. 두 노드가 다 올라갈 때까지 아무 변경도 오가지 않는다. +- **순서만 달라도 영영 거절된다.** 출력과 입력을 따로 선언하므로(`output changed`, `input remote`) 필드 순서가 어긋나기 쉽다. + 같은 필드라도 순서가 다르면 거절되고, 이유는 보내는 쪽 stderr에 한 줄 남을 뿐이다. + +## 결정 + +**받는 입력은 보낸 필드를 모두 가지면 받는다.** 이름으로 맞춘다. + +- 보낸 쪽에 없는 입력의 필드는 0이나 빈 값이다. 새 입력이 옛 보내는 쪽의 메시지를 받는다. +- 입력에 없는 필드를 보내면 받지 않는다. 받으면 그 값을 버리게 된다. 옛 받는 쪽이 새 필드를 버린 채 적용하고 번호를 들으면, 나중에 + 올라가도 그 값은 다시 오지 않는다. 보드라면 두 노드가 영영 다른 판을 갖는다. 그래서 그 메시지는 받는 쪽이 올라갈 때까지 기다린다. +- 같은 이름은 같은 종류여야 한다. 넓히는 것은 된다(u32를 number로, text를 bytes로). 좁히는 것(number를 u32로, bytes를 text로)은 + 값을 잃을 수 있어 받지 않는다. +- 순서는 상관없다. + +그래서 필드를 **더할 때는 받는 쪽부터**, **뺄 때는 보내는 쪽부터** 올리면 멈추지도 잃지도 않는다. 한 노드가 보내고 받는 +보드에서는, 먼저 올린 노드가 옛 노드의 변경을 받는 동안 제 새 필드는 옛 노드가 올라갈 때까지 기다린다. + +이것은 대기 중인 보내기를 옮기는 규칙([105](../research/105-growing-a-shared-board.md), `pure_send_recast`)과 같은 규칙에서 "버리지 +않는다"를 더한 것이다. 보내는 앱이 제 출력에서 필드를 뺀 것은 그 앱의 결정이라 기다리는 보내기에서도 뺀다. 받는 쪽이 모르는 +필드를 버리는 것은 누구의 결정도 아니다. + +## 링크 + +프레임을 하나 더한다([012](012-link-protocol.md)). + +| 종류 | 이름 | 방향 | 배치 | +|---:|---|---|---| +| 0x0d | FIELDS | 보냄 | names, u32 shape, u8 필드 수, 필드마다 {u8 타입, u8 이름 길이, 이름} | + +- 보내는 쪽은 REFUSED(2: 다른 모양)를 들으면, 스트림이 받아들여질 때까지 패킷마다 FIELDS를 싣는다. 자리는 OFFER 다음, 기록 + 앞이다. 이 프레임을 모르는 받는 쪽은 거기서 읽기를 멈추므로, 그 앞의 RESPONSE와 OFFER는 읽고 기록은 읽지 않는다(지금처럼 + 거절된 스트림이다). +- 받는 쪽은 STREAM의 shape가 입력과 다를 때 곧바로 거절하지 않고 그 패킷의 FIELDS를 기다린다. 같은 스트림의 FIELDS가 오면 + 호스트가 결정의 규칙대로 판단한다. 받으면 스트림을 열고(뒤의 기록도 받는다), 아니면 REFUSED(4: 입력이 보낸 필드를 다 + 갖지 않는다)다. 패킷이 끝나도록 FIELDS가 없으면 REFUSED(2)다. +- 전달에는 스트림의 shape가 따라간다. 호스트는 입력과 다른 shape의 기록을 입력의 배치로 옮긴 뒤 적용한다. +- 이미 있는 스트림에 다른 shape가 오면(보내는 쪽이 새 모듈로 다시 떴다) 새 스트림처럼 판단한다. +- 옛 보내는 쪽은 FIELDS를 보내지 않으므로 지금처럼 거절된다. +- FIELDS는 한 패킷에 다른 프레임과 함께 들어갈 때만 싣는다. 필드가 아주 많거나 이름이 길어 1,200 B 패킷에 들어가지 않으면 + 전처럼 거절된다(필드 64개의 이름이 평균 15 B를 넘을 때다). +- 받은 스트림의 모양이 바뀌면(보내는 쪽이 새 모듈로 다시 떴다) 받는 쪽은 쥐던 기록을 놓는다. 보내는 쪽은 받았다는 답을 + 들었지만 저장 소식이 없으니 1초 뒤 새 배치로 다시 보낸다. 모양이 바뀔 때 한 번 1초가 걸린다. + +## HTTP + +`X-Pure-Source`가 있는 전달의 JSON은 빠진 필드를 0이나 빈 값으로 둔다. 모르는 멤버는 지금처럼 거절한다(버리지 않는다). +클라이언트의 요청은 지금처럼 모든 필드가 있어야 한다. + +## 검사 + +- test-link: 다른 모양의 스트림이 FIELDS 뒤에 받아들여져 모든 기록이 저장된다. 호스트가 받지 않으면 REFUSED(4)가 남는다. +- check.py: + - 필드 순서만 다른 출력과 입력이 이어진다. + - 새 입력이 옛 출력의 메시지를 받는다(빠진 필드는 빈 값). + - 옛 입력은 새 필드를 보내는 출력을 거절하고, 올라가면 받는다(보낸 값을 잃지 않는다). + - HTTP 전달도 같다. +- 보드: 한 노드씩 올리는 동안 옛 노드의 변경이 새 노드로 가고, 새 노드의 변경은 옛 노드가 올라간 뒤 간다. 끝에 두 판이 같다. + +## 중단 기준 + +| 조건 | 결정 | +|---|---| +| FIELDS가 들어간 패킷 때문에 옛 받는 쪽과의 링크가 전보다 나빠진다(거절 외의 것이 막힌다) | FIELDS를 따로 된 패킷으로 보낸다 | +| 옮기는 비용이 전달의 처리량을 눈에 띄게 줄인다 | 같은 모양이면 옮기지 않는다(이미 그렇다). 다른 모양의 비용을 잰다 | diff --git a/link/link-protocol.c b/link/link-protocol.c index 4fa5aab..169705f 100644 --- a/link/link-protocol.c +++ b/link/link-protocol.c @@ -19,7 +19,7 @@ enum { DATAGRAM_MOST = PURE_LINK_DATAGRAM_MOST, BODY_MOST = DATAGRAM_MOST - HEAD * at least this much of it fits. */ enum { PART_LEAST = 64 }; enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, F_CHALLENGE = 7, - F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12 }; + F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12, F_FIELDS = 13 }; /* STORED's flags: what came after a gap was not kept; the app is full (no * room for more now), and what came is kept for when it has room. */ enum { STORED_GAP = 1, STORED_FULL = 2 }; @@ -307,6 +307,11 @@ struct PureLinkSender { * answer with no REFUSED. */ unsigned refusal, tried; uint64_t refused_until, refused_wait, tried_pn; + /* The fields its records are laid out with, named (FIELDS) in each + * packet from a refusal for another shape until the stream is taken. */ + unsigned char *fields; + size_t fields_size; + bool naming; PureLinkCounts counts; /* Epochs (forward secrecy). Keys come from the secret of the epoch it * sends in (epoch 0's is the two nodes' static secret), each made from @@ -696,6 +701,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui unsigned flags = p[names + 8]; uint32_t credit = get32(p + names + 9); l->refusal = l->tried = 0, l->refused_until = l->refused_wait = 0; /* the stream is taken, and stored */ + l->naming = false; if (heard > l->stored) { while (l->count && record_at(l, 0)->seq <= heard) { Record *r = record_at(l, 0); @@ -725,8 +731,12 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui static void refused(PureLinkSender *l, unsigned why, uint64_t now) { if (l->refusal && l->tried != TRIED) return; l->refusal = why ? why : PURE_LINK_NO_INPUT; - l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST; - l->refused_until = now + l->refused_wait; + /* Another shape, and its fields not named yet: named in a packet at once. */ + bool naming = (why == PURE_LINK_OTHER_SHAPE || why == PURE_LINK_FIELDS_LACKED) && l->fields; + bool at_once = naming && !l->naming; + l->naming = l->naming || naming; + if (!at_once) l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST; + l->refused_until = at_once ? 0 : now + l->refused_wait; l->tried = 0; ++l->counts.refused; for (size_t k = 0; k < l->count; ++k) { @@ -804,7 +814,8 @@ void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_ } else if (type != F_PING) break; /* padding, or what this side does not know: the rest */ } - if (l->refusal && l->tried == TRIED && !refusing) l->refusal = l->tried = 0, l->refused_until = 0; /* taken now */ + if (l->refusal && l->tried == TRIED && !refusing) /* taken now */ + l->refusal = l->tried = 0, l->refused_until = 0, l->naming = false; } /* A packet of what is not sent yet, in order, within the receiver's credit @@ -831,6 +842,17 @@ static size_t fill(PureLinkSender *l, unsigned char *body, Flight *f, bool pad, memcpy(body + at + 5, l->offer_public, PURE_SEAL_KEY); at += EPOCH_FRAME; } + /* The fields, for a receiver whose input takes another shape: after the + * frames one that knows no FIELDS reads, before the records. */ + size_t names = 2 + strlen(l->source) + strlen(l->input); + if (l->naming && at + 1 + names + 4 + l->fields_size <= most) { + body[at++] = F_FIELDS; + at += put_names(body + at, l->source, l->input); + put32(body + at, l->shape); + at += 4; + memcpy(body + at, l->fields, l->fields_size); + at += l->fields_size; + } if (l->renegotiating) { /* no record under the static keys */ l->responding = 0; return at; @@ -1196,6 +1218,15 @@ uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; } size_t pure_link_sender_held(const PureLinkSender *l) { return l->held; } +bool pure_link_sender_fields(PureLinkSender *l, const unsigned char *fields, size_t size) { + unsigned char *kept = size ? malloc(size) : NULL; + if (size && !kept) return false; + if (size) memcpy(kept, fields, size); + free(l->fields); + l->fields = kept, l->fields_size = size; + return true; +} + void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) { *out = l->counts; out->stored = l->stored; @@ -1213,6 +1244,7 @@ void pure_link_sender_free(PureLinkSender *l) { explicit_bzero(l->epoch_secret, sizeof(l->epoch_secret)); explicit_bzero(l->before_secret, sizeof(l->before_secret)); explicit_bzero(l->offer_private, sizeof(l->offer_private)); + free(l->fields); free(l); } @@ -1298,6 +1330,8 @@ typedef struct Stream { size_t held_first, held_count, held_capacity, held_bytes; Run *run; /* the run that sent to it last: where STORED goes */ uint64_t run_id; + unsigned char *fields; /* the fields its records are laid out with, when its input took them by name */ + size_t fields_size; } Stream; struct Active { uint32_t peer; /* its place among the peers */ @@ -1720,6 +1754,7 @@ static void let_go(PureLinkReceiver *s, Active *a) { while (a->streams[k]->held_count) drop_first(a->streams[k]); free(a->streams[k]->held); free(a->streams[k]->line); + free(a->streams[k]->fields); free(a->streams[k]); } free(a->streams); @@ -1845,30 +1880,56 @@ static bool allowed(const PureLinkReceiver *s, const Peer *peer, const char *sou return false; } +/* Whose fields a stream's records are laid out with, kept (NULL, 0: its + * input's). False when memory runs out. */ +static bool keep_fields(Stream *st, const unsigned char *fields, size_t size) { + unsigned char *kept = fields && size ? malloc(size) : NULL; + if (fields && size && !kept) return false; + if (kept) memcpy(kept, fields, size); + free(st->fields); + st->fields = kept, st->fields_size = kept ? size : 0; + return true; +} + /* The stream of a peer's source to input, made the first time: NULL when it - * is refused (why: the source is not allowed from the peer, or the host has - * no input that takes such messages), or memory runs out (why 0). */ + * is refused (why: the source is not allowed from the peer, the host has no + * input that takes such messages, one of another shape unless it takes the + * fields named, which come with FIELDS), or memory runs out (why 0). A + * stream the sender now sends in another shape (it started again with + * another module) starts over with what it holds let go: the sender sends + * those again, laid out anew. */ static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape, - unsigned *why) { + const unsigned char *fields, size_t fields_size, unsigned *why) { *why = 0; - for (unsigned k = 0; k < a->stream_count; ++k) { - Stream *st = a->streams[k]; - if (strcmp(st->input_name, input) || strcmp(st->source, source)) continue; - if (st->shape != shape) *why = PURE_LINK_OTHER_SHAPE; - return *why ? NULL : st; - } - int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) - : -PURE_LINK_NOT_ALLOWED; + Stream *st = NULL; + for (unsigned k = 0; k < a->stream_count && !st; ++k) + if (!strcmp(a->streams[k]->input_name, input) && !strcmp(a->streams[k]->source, source)) st = a->streams[k]; + if (st && st->shape == shape) return st; + int index = !allowed(s, &s->peers[a->peer], source) ? -PURE_LINK_NOT_ALLOWED : + s->host.input(s->host.context, input, shape, fields, fields ? fields_size : 0); if (index < 0) { - *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED ? (unsigned)-index : PURE_LINK_NO_INPUT; + *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED || index == -PURE_LINK_FIELDS_LACKED + ? (unsigned)-index : PURE_LINK_NO_INPUT; return NULL; } + if (st) { /* another shape now */ + if (st->out) { *why = PURE_LINK_OTHER_SHAPE; return NULL; } /* its last delivery first */ + if (!keep_fields(st, fields, fields_size)) return NULL; + while (st->held_count) drop_first(st); + st->line_head = st->line_size = 0; + st->line_count = 0; + st->tail = st->known = false; + st->input = (uint32_t)index; + st->shape = shape; + return st; + } size_t capacity = a->stream_capacity; - Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st; + Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2); if (!streams) return NULL; a->streams = streams; a->stream_capacity = (unsigned)capacity; if (!(st = calloc(1, sizeof(*st)))) return NULL; + if (!keep_fields(st, fields, fields_size)) { free(st); return NULL; } st->active = a; st->input = (uint32_t)index; st->shape = shape; @@ -2023,7 +2084,8 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) { size_t bytes = st->line_size - st->line_head; /* the whole line, when it is a batch or less */ if (n > BATCH) for (n = 0, bytes = 0; n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes); - PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st}; + PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st, + st->fields, st->fields_size}; if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; st->out_count = n, st->out_bytes = bytes; @@ -2052,7 +2114,8 @@ static bool hand_over(PureLinkReceiver *s, Stream *st) { memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size); at += 4 + held_at(st, k)->size; } - PureLinkDelivery d = {st->input, st->source, held_at(st, first)->seq, (unsigned)n, s->batch, bytes, st}; + PureLinkDelivery d = {st->input, st->source, held_at(st, first)->seq, (unsigned)n, s->batch, bytes, st, + st->fields, st->fields_size}; if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; ++s->counts.deliveries; @@ -2215,8 +2278,10 @@ static void lost(PureLinkReceiver *s, const Peer *peer, const Active *a, uint64_ * to its run: once a turn. */ static void refuse(PureLinkReceiver *s, Run *r, const unsigned char *names, size_t size, unsigned why) { for (size_t k = 0; k < s->refusal_count; ++k) - if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size)) + if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size)) { + s->refusals[k].frame[1 + size] = (unsigned char)why; /* said once, as last judged (named fields, say) */ return; + } void *more = room(s->refusals, &s->refusal_capacity, s->refusal_count, sizeof(*s->refusals), 4); if (!more) return; s->refusals = more; @@ -2318,6 +2383,10 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack Stream *st = NULL; bool eliciting = false, records = false; const unsigned char *offer = NULL; /* taken after the frames: whether records came with it */ + /* A stream of another shape: refused at the end unless this packet's + * FIELDS names fields its input takes. */ + const unsigned char *other = NULL; + size_t other_names = 0; for (size_t at = 0; at < size;) { unsigned type = plain[at++]; if (type == F_STREAM) { @@ -2325,14 +2394,36 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack size_t names = get_names(plain + at, size - at, source, input); if (!names || at + names + 4 > size) break; unsigned why; - st = stream_of(s, a, source, input, get32(plain + at + names), &why); + st = stream_of(s, a, source, input, get32(plain + at + names), NULL, 0, &why); if (st) { st->run = r; st->run_id = r->run; - } else if (why) + } else if (why == PURE_LINK_OTHER_SHAPE) + other = plain + at, other_names = names; + else if (why) refuse(s, r, plain + at, names, why); at += names + 4; eliciting = true; + } else if (type == F_FIELDS) { + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; + size_t names = get_names(plain + at, size - at, source, input), length = 1; + if (!names || at + names + 5 > size) break; + const unsigned char *fields = plain + at + names + 4; + for (unsigned f = 0; f < fields[0] && length != SIZE_MAX; ++f) /* each {type, name length, name} */ + length = at + names + 4 + length + 2 <= size ? length + 2 + fields[length + 1] : SIZE_MAX; + if (length == SIZE_MAX || at + names + 4 + length > size) break; + if (other && names == other_names && !memcmp(plain + at, other, names + 4)) { + unsigned why; + st = stream_of(s, a, source, input, get32(plain + at + names), fields, length, &why); + if (st) { + st->run = r; + st->run_id = r->run; + } else if (why) + refuse(s, r, other, other_names, why); + other = NULL; + } + at += names + 4 + length; + eliciting = true; } else if (type == F_RECORD) { if (at + RECORD_HEAD - 1 > size) break; uint32_t length = get16(plain + at + 16); @@ -2384,6 +2475,7 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack } else break; /* padding, or what this side does not know: the rest */ } + if (other) refuse(s, r, other, other_names, PURE_LINK_OTHER_SHAPE); /* its fields not named */ if (offer) offer_taken(s, a, r, get32(offer), offer + 4, e, !records, now); if (eliciting) { r->ack_due = true; diff --git a/link/link-protocol.h b/link/link-protocol.h index e4d3080..feff543 100644 --- a/link/link-protocol.h +++ b/link/link-protocol.h @@ -111,6 +111,12 @@ * sender sends one packet of them a wait from then (a * second, twice as long each time, 64 s at most) until one * is stored + * 0x0d FIELDS source, input, u32 shape (as STREAM), then the fields the + * stream's records are laid out with (core/module.h's + * pure_module_fields): after a REFUSED for another shape, + * in every packet (after OFFER, before records) until the + * stream is taken. A receiver whose input has every field + * takes it by name (docs/design/014-fields-by-name.md) * 0x00 PADDING to the end of the packet: a full packet sent in a burst is * a whole datagram * Losses are detected as RFC 9002 says, with RACK's patience for packets @@ -129,9 +135,10 @@ enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_DATAGRAM_MOST = 1452, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 48, PURE_LINK_RECORD_MOST = 16 << 20 }; /* Why a receiver refuses a stream (REFUSED): no input of that name takes - * messages, the input takes messages of another shape, or the sender's key - * may not send as that source. */ -enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3 }; + * messages, the input takes messages of another shape (send FIELDS), the + * sender's key may not send as that source, or the input lacks fields the + * stream sends (it would drop them). */ +enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3, PURE_LINK_FIELDS_LACKED = 4 }; /* Counts for a host's report: packets and records sent, records sent again, * packets lost and acknowledged, packets received, deliveries handed over, @@ -193,6 +200,9 @@ unsigned pure_link_sender_refusal(const PureLinkSender *); /* How often it moves on to a new epoch (ns), and after how many bytes sent * (an hour and a GiB at first). */ void pure_link_sender_epochs(PureLinkSender *, uint64_t every_ns, uint64_t bytes); +/* The fields its records are laid out with (pure_module_fields), for a + * receiver whose input takes another shape (FIELDS); none at first. */ +bool pure_link_sender_fields(PureLinkSender *, const unsigned char *fields, size_t size); /* The last number queued; the last the receiver stored. */ uint64_t pure_link_sender_queued(const PureLinkSender *); uint64_t pure_link_sender_stored(const PureLinkSender *); @@ -206,7 +216,9 @@ void pure_link_sender_free(PureLinkSender *); * pure_link_idle). A stream (a source to an input) goes to the host's input * its name and shape find: its number, or -PURE_LINK_OTHER_SHAPE when it * takes another shape, or -1 (PURE_LINK_NO_INPUT) when none takes messages - * by that name; a stream refused is said to be (REFUSED). Whole + * by that name; asked again with the fields the sender names (FIELDS), + * its number when it takes them by name, else -PURE_LINK_FIELDS_LACKED. A + * stream refused is said to be (REFUSED). Whole * records of a stream, in order, go to deliver as one delivery; the host * applies them from the one after the source's last heard number, stops at * a gap, saves, and then says so with a done (pure_link_receiver_turn). @@ -220,9 +232,13 @@ typedef struct { const unsigned char *records; /* count of: u32le size, bytes (the receiver's: the host's during the call) */ size_t size; void *token; /* for its done */ + /* The fields the records are laid out with when the input took them by + * name (the host lays each out anew for it), else NULL. */ + const unsigned char *fields; + size_t fields_size; } PureLinkDelivery; typedef struct { - int (*input)(void *context, const char *name, uint32_t shape); + int (*input)(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t fields_size); bool (*deliver)(void *context, const PureLinkDelivery *); void *context; } PureLinkHost; diff --git a/link/link.c b/link/link.c index ad9376c..8f36e21 100644 --- a/link/link.c +++ b/link/link.c @@ -210,6 +210,13 @@ uint64_t pure_link_delivered(PureLink *l) { return stored; } +bool pure_link_fields(PureLink *l, const unsigned char *fields, size_t size) { + pthread_mutex_lock(&l->lock); + bool kept = pure_link_sender_fields(l->sender, fields, size); + pthread_mutex_unlock(&l->lock); + return kept; +} + size_t pure_link_held(PureLink *l) { pthread_mutex_lock(&l->lock); size_t held = pure_link_sender_held(l->sender); @@ -380,7 +387,8 @@ static void *run_link(void *context) { "trying again\"}\n", l->host, l->port, l->input, l->source, refusal == PURE_LINK_NOT_ALLOWED ? "this node's key may not send as that source" : refusal == PURE_LINK_OTHER_SHAPE ? "its input takes messages of another shape" - : "it has no input of that name taking messages"); + : refusal == PURE_LINK_FIELDS_LACKED ? "its input lacks fields this output sends (it would drop them)" + : "it has no input of that name taking messages"); } return NULL; } diff --git a/link/link.h b/link/link.h index 64846ce..c8d8635 100644 --- a/link/link.h +++ b/link/link.h @@ -18,6 +18,10 @@ typedef struct PureLink PureLink; PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size); void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]); +/* The fields its sends are laid out with (core/module.h's + * pure_module_fields): a receiver whose input takes another shape may take + * them by name (docs/design/014-fields-by-name.md). */ +bool pure_link_fields(PureLink *, const unsigned char *fields, size_t size); bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size); uint64_t pure_link_queued(PureLink *); /* The last number the receiver saved; any thread. */ diff --git a/link/test-link.c b/link/test-link.c index 66899ff..2e7509a 100644 --- a/link/test-link.c +++ b/link/test-link.c @@ -102,6 +102,8 @@ typedef struct { * handed records meanwhile, and how long after it had room again. */ uint64_t full_at, full_ms, full_until, full_tries, full_late; bool later_open; /* the app has an input "later" (as "order") */ + bool takes; /* its input takes the fields a sender of another shape names (NAMED) */ + uint64_t named_deliveries; /* deliveries that came with those fields */ } World; static void put_packet(World *w, bool to_receiver, const unsigned char *bytes, size_t size) { @@ -146,10 +148,15 @@ static bool answer(void *context, const unsigned char *packet, size_t size, cons return true; } -static int input_of(void *context, const char *name, uint32_t shape) { +/* The fields a sender of another shape (0x5eee) names; opaque to the link. */ +static const unsigned char NAMED[] = {2, 4, 4, 'i', 't', 'e', 'm', 1, 5, 'c', 'o', 'u', 'n', 't'}; + +static int input_of(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) { const World *w = context; if (strcmp(name, "order") && !(w && w->later_open && !strcmp(name, "later"))) return -PURE_LINK_NO_INPUT; - return shape == 0x5eed ? 3 : -PURE_LINK_OTHER_SHAPE; + if (shape == 0x5eed) return 3; + if (!fields) return -PURE_LINK_OTHER_SHAPE; + return w && w->takes && size == sizeof(NAMED) && !memcmp(fields, NAMED, size) ? 3 : -PURE_LINK_FIELDS_LACKED; } /* Applied from the one after the last heard, in order; a gap stops it. */ @@ -158,6 +165,7 @@ static bool deliver(void *context, const PureLinkDelivery *d) { CHECK(d->input == 3 && !strcmp(d->source, "test.ordered"), "a delivery to input %u from %s", d->input, d->source); if (w->done_count == sizeof(w->done) / sizeof(w->done[0])) return false; /* busy: tried again */ ++w->deliveries; + w->named_deliveries += d->fields && d->fields_size == sizeof(NAMED) && !memcmp(d->fields, NAMED, sizeof(NAMED)); if (w->full_ms && !w->full_until && w->heard >= w->full_at) w->full_until = w->now + w->full_ms * MS; if (w->now < w->full_until) { ++w->full_tries; @@ -682,8 +690,8 @@ typedef struct { PureLinkDone done[64]; size_t done_count; } Outputs; -static int output_input(void *context, const char *name, uint32_t shape) { - (void)context; +static int output_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) { + (void)context, (void)fields, (void)size; return shape != 0x5eed ? -1 : !strcmp(name, "one") ? 0 : !strcmp(name, "two") ? 1 : !strcmp(name, "three") ? 2 : -1; } static bool output_deliver(void *context, const PureLinkDelivery *d) { @@ -958,6 +966,73 @@ static void refused_restarted(const char *dir) { free(w.packets); } +/* A stream of another shape whose sender names its fields (FIELDS): the + * receiver's input takes them by name and every record is stored, each + * delivery naming them for the host to lay the records out anew; an input + * that does not take them refuses the stream (it would drop fields), and the + * sender tries a packet a wait as for any refusal. One that names none stays + * refused for another shape (refusals, above). */ +static void named(const char *dir) { + for (int takes = 1; takes >= 0; --takes) { + unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST]; + PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b); + char memo[512]; + snprintf(memo, sizeof(memo), "%s/b.key.links", dir); + unlink(memo); + World w = {.packets = malloc(PACKETS * sizeof(Packet)), .takes = takes}; + if (!w.packets) exit(1); + PureLinkReceiver *r = receiver_of(&w, b, public_b, public_a, memo, 120000); + PureLinkSender *s = pure_link_sender("test.ordered", "order", 0x5eee, a, public_a); + if (!s || !pure_link_sender_peer(s, public_b) || !pure_link_sender_fields(s, NAMED, sizeof(NAMED))) exit(1); + static unsigned char bytes[20200]; + for (uint64_t seq = 1; seq <= 200; ++seq) { + size_t size = record_size(seq); + for (size_t k = 0; k < size; ++k) bytes[k] = record_byte(seq, k); + if (!pure_link_sender_queue(s, seq, bytes, size)) exit(1); + } + uint64_t number, packets = 0, stored_at = 0; + for (; w.now < 30000 * MS; w.now += MS) { + for (size_t size; (size = pure_link_sender_packet(s, out, false, w.now, &number)); ++packets) + pure_link_receiver_take(r, out, size, "sender0", 7, w.now); + size_t told = w.done_count; + memcpy(w.told, w.done, told * sizeof(*w.done)); + w.done_count = 0; + pure_link_receiver_turn(r, w.told, told, w.now, answer, &w); + for (size_t k = 0; k < w.count;) { + if (w.packets[k].at > w.now) { + ++k; + continue; + } + pure_link_sender_take(s, w.packets[k].bytes, w.packets[k].size, w.now); + w.packets[k] = w.packets[--w.count]; + } + pure_link_sender_tick(s, w.now); + if (!stored_at && pure_link_sender_stored(s) == 200) stored_at = w.now; + } + PureLinkCounts c; + pure_link_sender_counts(s, &c); + unsigned why = pure_link_sender_refusal(s); + printf("{\"case\":\"named\",\"takes\":%d,\"why\":%u,\"refused\":%llu,\"packets\":%llu,\"stored\":%llu," + "\"stored_ms\":%llu,\"named_deliveries\":%llu}\n", takes, why, (unsigned long long)c.refused, + (unsigned long long)packets, (unsigned long long)pure_link_sender_stored(s), + (unsigned long long)(stored_at / MS), (unsigned long long)w.named_deliveries); + if (takes) + CHECK(!why && c.refused == 1 && stored_at && stored_at < 1000 * MS && w.heard == 200 && + w.named_deliveries == w.deliveries, + "named and taken: why %u, refused %llu, stored at %llu ms, %llu of %llu deliveries named", why, + (unsigned long long)c.refused, (unsigned long long)(stored_at / MS), + (unsigned long long)w.named_deliveries, (unsigned long long)w.deliveries); + else /* refused at once, named at once, then at 1, 3, 7 and 15 s */ + CHECK(why == PURE_LINK_FIELDS_LACKED && c.refused == 6 && !pure_link_sender_stored(s) && packets <= 20, + "named and not taken: why %u, refused %llu, %llu packets", why, (unsigned long long)c.refused, + (unsigned long long)packets); + pure_link_sender_free(s); + pure_link_receiver_free(r); + pure_seal_key_free(a), pure_seal_key_free(b); + free(w.packets); + } +} + /* The largest send (PURE_LINK_RECORD_MOST) goes whole, over a receiver that * never heard of the stream before; one byte more is not taken to send. */ typedef struct { World *w; PureLinkSender *s; } Largest; @@ -1089,6 +1164,7 @@ int main(void) { outputs(dir); refusals(dir); refused_restarted(dir); + named(dir); largest(dir); char command[600]; snprintf(command, sizeof(command), "rm -rf %s", dir); diff --git a/serve/serve.c b/serve/serve.c index d06f430..319f165 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -843,8 +843,17 @@ static const char *convert(Connection *c) { memset(b.seen, 0, sizeof(b.seen)); if (!why) why = pure_json_message(&b, json, size, &at); at = pure_json_space(json, at, size); - for (uint32_t f = 0; !why && f < shape->field_count; ++f) - if (!b.seen[f] && !c->given[f]) why = "missing_field"; + /* A delivery from another node may lack fields its sender's module has + * not (docs/design/014-fields-by-name.md): they are 0 or empty. A + * member naming no field is refused all the same: it would be dropped. */ + for (uint32_t f = 0; !why && f < shape->field_count; ++f) { + if (b.seen[f] || c->given[f]) continue; + const PureField *field = &c->host->module->field[shape->field_start + f]; + if (!delivered || field->type == PURE_FIELD_ROWS) why = "missing_field"; + else if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER) + memset(b.payload + field->offset, 0, field->type == PURE_FIELD_NUMBER ? 8 : 4); + else put_u32(b.table + 8 * f, 0), put_u32(b.table + 8 * f + 4, 0); + } if (!why && c->batch > 1 && b.rows_size) why = "delivery"; /* an output sends no rows */ } if (!why && c->batch > 1 && (at == size || json[at++] != ']')) why = "json"; @@ -1374,6 +1383,9 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect uint64_t first = get_u64(delivery), heard = pure_app_heard(app, source); unsigned count = get_u32(delivery + 8); const unsigned char *at = effect->value.data + 4 + DELIVERY; + /* Records laid out with other fields (named after them) are laid out anew for the input. */ + size_t records = header & (STORED_RECORDS - 1), named = effect->value.size - 4 - DELIVERY - records; + const unsigned char *fields = named ? at + records : NULL; out->gap = false; out->fault = PURE_FAULT_NONE; for (unsigned i = 0; i < count; ++i) { @@ -1385,7 +1397,14 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect out->gap = true; break; } - PureResult status = pure_app_deliver(app, route->target, source, first + i, record, NULL); + unsigned char *recast = NULL; + size_t recast_size = 0; + if (fields && pure_module_recast(app->module, route->target, fields, named, record, &recast, &recast_size) != + PURE_OK) + break; /* not laid out as its fields were named: not applied, sent again */ + PureResult status = pure_app_deliver(app, route->target, source, first + i, + recast ? (PureSpan){recast, recast_size} : record, NULL); + free(recast); if (status != PURE_OK && !app_refused(app)) { /* not applied: sent again */ out->fault = app->fault; break; @@ -2375,12 +2394,15 @@ static void stop_shard(Host *host) { } /* The input a link's stream goes to: one taking messages (not keyed) of the - * sender's shape. */ -static int link_input(void *context, const char *name, uint32_t shape) { + * sender's shape, or, named its fields, one that takes them by name + * (pure_module_takes: each record is laid out anew for it). */ +static int link_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) { const PureModule *m = ((Owner *)context)->executor.app.module; int index = pure_module_input(m, name); if (index < 0 || m->input[index].kind != PURE_KIND_MESSAGE || m->input[index].keyed) return -PURE_LINK_NO_INPUT; - return pure_module_shape(m, (uint32_t)index) == shape ? index : -PURE_LINK_OTHER_SHAPE; + if (pure_module_shape(m, (uint32_t)index) == shape) return index; + if (!fields) return -PURE_LINK_OTHER_SHAPE; + return pure_module_takes(m, (uint32_t)index, fields, size) ? index : -PURE_LINK_FIELDS_LACKED; } /* Records a link delivered, queued at the owner as a request is: applied @@ -2388,7 +2410,7 @@ static int link_input(void *context, const char *name, uint32_t shape) { static PureLinkListener *link_listener; static bool link_deliver(void *context, const PureLinkDelivery *d) { Owner *owner = context; - size_t size = 4 + DELIVERY + d->size, length = strlen(d->source); + size_t size = 4 + DELIVERY + d->size + d->fields_size, length = strlen(d->source); unsigned char *ingress = malloc(size); LinkJob *link = malloc(sizeof(*link)); if (!ingress || !link || d->size > (STORED_RECORDS - 1) || length > PURE_SOURCE) { @@ -2403,6 +2425,7 @@ static bool link_deliver(void *context, const PureLinkDelivery *d) { ingress[16] = (unsigned char)length; memcpy(ingress + 17, d->source, length); memcpy(ingress + 4 + DELIVERY, d->records, d->size); + if (d->fields_size) memcpy(ingress + 4 + DELIVERY + d->size, d->fields, d->fields_size); /* laid out with those */ *link = (LinkJob){link_listener, d->token, ingress}; Job job = {.route = {.kind = ROUTE_INPUT, .target = d->input}, .link = link, .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}}; @@ -2451,6 +2474,12 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen ? pure_link_open(equals + 1, source, pure_module_shape(m, (uint32_t)index), seal_key, own_public, error, sizeof(error)) : NULL; + unsigned char named[PURE_FIELDS_NAMED]; + if (link && !pure_link_fields(link, named, pure_module_fields(m, (uint32_t)index, named))) { + pure_link_close(link); + link = NULL; + snprintf(error, sizeof(error), "no memory"); + } if (!link) { fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error); return false; diff --git a/tests/check.py b/tests/check.py index b377e20..7ccafc6 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1339,9 +1339,9 @@ UPGRADED_RECEIVER = """state orders = 0 state noted = 0 state stock = 0 input order = message { - note: text item: text count: u32 + note: text } on order { orders = orders + 1 @@ -1360,8 +1360,9 @@ def server_upgrade(aup, pure, work): to.city}; started again with ordered {note, item, count} it recasts them by name (note empty, to.city gone) where it was refused before (review 004's F3), and a module without the output is refused naming it. The old - receiver refuses the new shape, so the sends wait; the upgraded receiver - takes all eight, once each.""" + receiver lacks note, so it refuses them (it would drop the field) and + they wait; the upgraded receiver, its fields in another order, takes all + eight by name, once each.""" import select import socket import time @@ -1439,6 +1440,191 @@ def server_upgrade(aup, pure, work): assert stats['a']['link_refused'] >= 1, stats['a'] return {'recast_waiting': 5, 'delivered': 8, 'refused_while_old': stats['a']['link_refused']} + +FIELDS_SENDER = """input buy = message { + item: text + count: u32 +} +output swapped = message { + count: u32 + item: text +} +output plain = message { + item: text + count: u32 +} +output noted = message { + item: text + count: u32 + note: text +} +output over_http = message { + item: text + count: u32 +} +output noted_http = message { + item: text + count: u32 + note: text +} +on buy { + send swapped { count: buy.count, item: buy.item } + send plain { item: buy.item, count: buy.count } + send noted { item: buy.item, count: buy.count, note: "kept" } + send over_http { item: buy.item, count: buy.count } + send noted_http { item: buy.item, count: buy.count, note: "kept" } +} +""" + + +def fields_receiver(narrow_notes): + """Inputs for FIELDS_SENDER's outputs; narrow ones lack note unless narrow_notes.""" + note = ' note: text\n' if narrow_notes else '' + kept = lambda name: f' if {name}.note == "kept" {{ {name}_kept = {name}_kept + 1 }}\n' if narrow_notes else '' + return (f"""state swapped_n = 0 +state swapped_total = 0 +state wider_n = 0 +state wider_blank = 0 +state narrow_n = 0 +state narrow_kept = 0 +state wider_http_n = 0 +state wider_http_blank = 0 +state narrow_http_n = 0 +state narrow_http_kept = 0 +input swapped_in = message {{ + item: text + count: u32 +}} +input wider = message {{ + note: text + item: text + count: u32 +}} +input narrow = message {{ + item: text + count: u32 +{note}}} +input wider_http = message {{ + item: text + count: u32 + note: text +}} +input narrow_http = message {{ + item: text + count: u32 +{note}}} +on swapped_in {{ + swapped_n = swapped_n + 1 + swapped_total = swapped_total + swapped_in.count +}} +on wider {{ + wider_n = wider_n + 1 + if wider.note == "" {{ wider_blank = wider_blank + 1 }} +}} +on narrow {{ + narrow_n = narrow_n + 1 +{kept('narrow')}}} +on wider_http {{ + wider_http_n = wider_http_n + 1 + if wider_http.note == "" {{ wider_http_blank = wider_http_blank + 1 }} +}} +on narrow_http {{ + narrow_http_n = narrow_http_n + 1 +{kept('narrow_http')}}} +view counts = "{{swapped_n}} {{swapped_total}} {{wider_n}} {{wider_blank}} {{narrow_n}} {{narrow_kept}} {{wider_http_n}} {{wider_http_blank}} {{narrow_http_n}} {{narrow_http_kept}}" +""") + + +def server_fields(aup, pure, work): + """Deliveries take their fields by name (docs/design/014-fields-by-name.md). + A sender's five outputs, three linked and two over HTTP, to one receiver: + fields in another order are taken; an input with a field more takes + messages without it (empty); an input lacking a field the output sends + refuses it, over the link (REFUSED, fields lacked) and over HTTP + (json_field), so nothing is dropped, and once the receiver starts again + with the field, every waiting message comes, the field kept.""" + import select + import socket + import time + modules = {} + for name, text in (('sender', FIELDS_SENDER), ('receiver-1', fields_receiver(False)), + ('receiver-2', fields_receiver(True))): + source = work / f'fields-{name}.aup' + source.write_text(text) + compile_module(aup, source, source.with_suffix('.pmod')) + modules[name] = source.with_suffix('.pmod') + + def free_port(kind=socket.SOCK_STREAM): + with socket.socket(socket.AF_INET, kind) as probe: + probe.bind(('127.0.0.1', 0)) + return probe.getsockname()[1] + + def start(args, port): + process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + ready, _, _ = select.select([process.stdout], [], [], 10) + line = process.stdout.readline() if ready else b'' + assert line, f'host did not start: {process.stderr.read(2000) if process.poll() is not None else ""}' + return process, json.loads(line) + + def stop(process): + process.terminate() + _, error = process.communicate(timeout=30) + assert process.returncode == 0, error[-1000:] + return error.decode() + + def counts(port): + return [int(n) for n in http(port, 'GET', '/views/counts')[3].split()] + + keys = {} + for name in ('sender', 'receiver'): + process, info = start([pure / 'build/pure-serve', '--module', modules['receiver-1'], '--store', + work / 'fields-key', '--key', work / f'fields-{name}.key'], 0) + keys[name] = info['public_key'] + stop(process) + port_a, port_b, link_port = free_port(), free_port(), free_port(socket.SOCK_DGRAM) + links = {'swapped': 'swapped_in', 'plain': 'wider', 'noted': 'narrow'} + sending = [pure / 'build/pure-serve', '--module', modules['sender'], '--store', work / 'fields-a', '--key', + work / 'fields-sender.key', '--name', 'up', + '--send', f'over_http=http://127.0.0.1:{port_b}/inputs/wider_http', + '--send', f'noted_http=http://127.0.0.1:{port_b}/inputs/narrow_http', + '--seal', f'over_http={keys["receiver"]}', '--seal', f'noted_http={keys["receiver"]}'] + for output, input_name in links.items(): + sending += ['--send', f'{output}=pure://127.0.0.1:{link_port}/{input_name}', '--seal', f'{output}={keys["receiver"]}'] + + def receiving(module): + args = [pure / 'build/pure-serve', '--module', modules[module], '--store', work / 'fields-b', '--key', + work / 'fields-receiver.key', '--link-port', link_port] + for output in (*links, 'over_http', 'noted_http'): + args += ['--sender', f'up.{output}={keys["sender"]}'] + return args + + running, said = {}, '' + try: + running['b'] = start(receiving('receiver-1'), port_b)[0] + running['a'] = start(sending, port_a)[0] + for i in range(3): + assert post(port_a, '/inputs/buy', {'item': f'i{i}', 'count': i + 1})[0] == 200 + deadline = time.time() + 20 + while counts(port_b)[:4] + counts(port_b)[6:8] != [3, 6, 3, 3, 3, 3] and time.time() < deadline: + time.sleep(0.05) + got = counts(port_b) + assert got == [3, 6, 3, 3, 0, 0, 3, 3, 0, 0], got # the narrow inputs took none: nothing is dropped + time.sleep(1.5) # the refused link tried with its fields named: the sender has said why + stop(running.pop('b')) + running['b'] = start(receiving('receiver-2'), port_b)[0] + deadline = time.time() + 30 + while counts(port_b)[4:6] + counts(port_b)[8:] != [3, 3, 3, 3] and time.time() < deadline: + time.sleep(0.1) + got = counts(port_b) + assert got == [3, 6, 3, 3, 3, 3, 3, 3, 3, 3], got + finally: + for name, process in running.items(): + text = stop(process) + said += text if name == 'a' else '' + assert 'its input lacks fields this output sends' in said, said[-2000:] + return {'links': 3, 'wires': 2, 'waited_then_taken': 6} + def server_links_independent(aup, pure, work): """A receiver written from docs/design/012-link-protocol.md alone, in Python with the cryptography package and no code of link/: a Pure node @@ -3146,6 +3332,7 @@ def main(): report['server_memory'] = server_memory(aup, pure, work) report['server_windows'] = server_windows(aup, pure, work) report['server_upgrade'] = server_upgrade(aup, pure, work) + report['server_fields'] = server_fields(aup, pure, work) report['server_links_independent'] = server_links_independent(aup, pure, work) report['server_links_moved'] = server_links_moved(aup, pure, work) report['server_webhook'] = server_webhook(aup, pure, work) -- cgit v1.2.3