diff options
| -rw-r--r-- | core/module.c | 65 | ||||
| -rw-r--r-- | core/module.h | 18 | ||||
| -rw-r--r-- | docs/design/012-link-protocol.md | 20 | ||||
| -rw-r--r-- | docs/design/014-fields-by-name.md | 77 | ||||
| -rw-r--r-- | link/link-protocol.c | 136 | ||||
| -rw-r--r-- | link/link-protocol.h | 26 | ||||
| -rw-r--r-- | link/link.c | 10 | ||||
| -rw-r--r-- | link/link.h | 4 | ||||
| -rw-r--r-- | link/test-link.c | 84 | ||||
| -rw-r--r-- | serve/serve.c | 43 | ||||
| -rw-r--r-- | tests/check.py | 193 |
11 files changed, 629 insertions, 47 deletions
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) |