From 80e88ce69d0ec047a5b3b58a2b808e50572eb736 Mon Sep 17 00:00:00 2001 From: belowzeroff Date: Wed, 16 Sep 2026 08:55:32 -0400 Subject: [PATCH 1/3] fix(q): include inet declarations in driver tests --- test/driver.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/test/driver.c b/test/driver.c index 2c2e139..7f6f771 100644 --- a/test/driver.c +++ b/test/driver.c @@ -32,8 +32,9 @@ #include #include #include -#include +#include #include +#include #include #include #include From 9089ccb4046ac24757cddc013bb37379fc724e6d Mon Sep 17 00:00:00 2001 From: belowzeroff Date: Fri, 2 Oct 2026 03:19:55 -0400 Subject: [PATCH 2/3] fix(q): reject unsupported timespan values --- q.c | 2 -- test/driver.c | 26 ++++++++++++++++++++++++++ 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/q.c b/q.c index 596e45f..67554ea 100644 --- a/q.c +++ b/q.c @@ -938,7 +938,6 @@ static ray_t *q_des_obj(uint8_t **buf, int64_t *len) { case -Q_KJ: return q_des_atom_i(buf, len, RAY_I64, 8); case -Q_KP: - case -Q_KN: return q_des_atom_i(buf, len, RAY_TIMESTAMP, 8); case -Q_KD: return q_des_atom_i(buf, len, RAY_DATE, 4); @@ -1019,7 +1018,6 @@ static ray_t *q_des_obj(uint8_t **buf, int64_t *len) { case Q_KJ: return q_des_vec_i(buf, len, RAY_I64, 8); case Q_KP: - case Q_KN: return q_des_vec_i(buf, len, RAY_TIMESTAMP, 8); case Q_KD: return q_des_vec_i(buf, len, RAY_DATE, 4); diff --git a/test/driver.c b/test/driver.c index 7f6f771..626aa0a 100644 --- a/test/driver.c +++ b/test/driver.c @@ -451,6 +451,32 @@ static int run_codec_selftest(void) { release_any(r); } + /* Q timespans (KN) are durations, not timestamps. Rayforce has no + * duration type, so reject them rather than silently changing meaning. */ + const struct { + const char *name; + uint8_t body[6]; + int64_t body_len; + } unsupported_temporals[] = { + {"timespan atom", {0xF0, 0, 0, 0, 0, 0}, 1}, + {"timespan vector", {0x10, 0, 0, 0, 0, 0}, 6}, + }; + for (size_t i = 0; + i < sizeof unsupported_temporals / sizeof unsupported_temporals[0]; + i++) { + err[0] = '\0'; + r = q_decode((uint8_t *)unsupported_temporals[i].body, + unsupported_temporals[i].body_len, 0, err, sizeof err); + if (r == NULL || !RAY_IS_ERR(r) || + strcmp(ray_err_code(r), "q: unsu") != 0) { + fprintf(stderr, "codec selftest: %s was not rejected: %s\n", + unsupported_temporals[i].name, + r && RAY_IS_ERR(r) ? ray_err_code(r) : err); + failures++; + } + release_any(r); + } + int listener = socket(AF_INET, SOCK_STREAM, 0); if (listener < 0) { perror("codec selftest: handshake socket"); From a10458afc2f3567b1df1f3b6157def00350523be Mon Sep 17 00:00:00 2001 From: belowzeroff Date: Fri, 2 Oct 2026 03:19:58 -0400 Subject: [PATCH 3/3] fix(q): receive large frames incrementally --- q_server.c | 64 ++++++++++++++++++++++++++++++++++++++++++---- test/run.sh | 73 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 5 deletions(-) diff --git a/q_server.c b/q_server.c index 349dd4e..6f44f47 100644 --- a/q_server.c +++ b/q_server.c @@ -69,6 +69,7 @@ typedef struct { #define Q_MSG_RESPONSE 2 #define Q_CAP_MAX 3 /* matches q.c client capability */ #define Q_MAX_BODY ((int64_t)256 << 20) /* reject absurd frames (256 MiB) */ +#define Q_RX_CHUNK_SIZE ((int64_t)64 << 10) #define Q_MAX_HANDSHAKE 512 /* credential blob upper bound */ /* Per-connection rx state. `hdr` carries the current frame's header between @@ -85,6 +86,10 @@ typedef struct { uint8_t sync_ready; /* its RESPONSE has been deposited below */ ray_t *sync_resp; int sync_timeout_ms; /* q_conn_send round-trip budget; <= 0 blocks */ + uint8_t *rx_body; + int64_t rx_body_len; + int64_t rx_body_cap; + int64_t rx_body_total; } q_conn_t; static void q_release_any(ray_t *obj) { @@ -246,6 +251,11 @@ static ray_t *q_read_header(ray_poll_t *poll, ray_selector_t *sel) { ray_poll_deregister(poll, sel->id); return NULL; } + free(cd->rx_body); + cd->rx_body = NULL; + cd->rx_body_len = 0; + cd->rx_body_cap = 0; + cd->rx_body_total = body; if (body == 0) { if (cd->hdr.msgtype != 0) /* sync -> identity reply */ q_send_result((ray_sock_t)sel->fd, NULL); @@ -253,23 +263,66 @@ static ray_t *q_read_header(ray_poll_t *poll, ray_selector_t *sel) { return NULL; } sel->rx.read_fn = q_read_body; - ray_poll_rx_request(poll, sel, body); + ray_poll_rx_request(poll, sel, + body < Q_RX_CHUNK_SIZE ? body : Q_RX_CHUNK_SIZE); return NULL; } static ray_t *q_read_body(ray_poll_t *poll, ray_selector_t *sel) { q_conn_t *cd = (q_conn_t *)sel->data; - int64_t body = (int64_t)cd->hdr.size - (int64_t)sizeof(q_header_t); - if (!sel->rx.buf || sel->rx.buf->offset < body) + if (!sel->rx.buf || sel->rx.buf->offset <= 0) + return NULL; + + int64_t received = sel->rx.buf->offset; + int64_t needed = cd->rx_body_len + received; + if (needed < cd->rx_body_len || needed > cd->rx_body_total) { + ray_poll_deregister(poll, sel->id); return NULL; + } + if (needed > cd->rx_body_cap) { + int64_t cap = cd->rx_body_cap + ? cd->rx_body_cap + : (cd->rx_body_total < Q_RX_CHUNK_SIZE + ? cd->rx_body_total + : Q_RX_CHUNK_SIZE); + while (cap < needed) { + if (cap > cd->rx_body_total / 2) { + cap = cd->rx_body_total; + break; + } + cap *= 2; + } + uint8_t *body = (uint8_t *)realloc(cd->rx_body, (size_t)cap); + if (body == NULL) { + ray_poll_deregister(poll, sel->id); + return NULL; + } + cd->rx_body = body; + cd->rx_body_cap = cap; + } + memcpy(cd->rx_body + cd->rx_body_len, sel->rx.buf->data, + (size_t)received); + cd->rx_body_len = needed; + + if (cd->rx_body_len < cd->rx_body_total) { + int64_t left = cd->rx_body_total - cd->rx_body_len; + ray_poll_rx_request(poll, sel, + left < Q_RX_CHUNK_SIZE ? left : Q_RX_CHUNK_SIZE); + return NULL; + } /* q_decode fully materializes the request into ray_t objects, so the rx * buffer is free to reuse the moment it returns. */ q_header_t hdr = cd->hdr; int64_t id = sel->id; char err[128] = {0}; - ray_t *req = - q_decode(sel->rx.buf->data, body, hdr.compressed, err, sizeof err); + ray_t *req = q_decode(cd->rx_body, cd->rx_body_len, hdr.compressed, err, + sizeof err); + free(cd->rx_body); + cd->rx_body = NULL; + cd->rx_body_len = 0; + cd->rx_body_cap = 0; + cd->rx_body_total = 0; sel->rx.read_fn = q_read_header; ray_poll_rx_request(poll, sel, (int64_t)sizeof(q_header_t)); @@ -326,6 +379,7 @@ static void q_on_close(ray_poll_t *poll, ray_selector_t *sel) { * mid-round-trip) would otherwise leak. */ if (cd->sync_resp) q_release_any(cd->sync_resp); + free(cd->rx_body); free(cd); sel->data = NULL; } diff --git a/test/run.sh b/test/run.sh index b99d21b..73c814f 100755 --- a/test/run.sh +++ b/test/run.sh @@ -141,6 +141,79 @@ with socket.create_connection((host, port), 1.0) as s: raise SystemExit("unknown message type was executed or answered") PY +echo "checking multi-chunk Q frame decoding..." +python3 - "$HOST" "$SERVERPORT" <<'PY' +import socket +import struct +import sys + +host, port = sys.argv[1], int(sys.argv[2]) +count = 40_000 +values = list(range(count)) +request = bytes([6, 0]) + struct.pack(" 32 * 1024: + raise SystemExit( + f"partial-frame test: server reserved {delta_kib} KiB before body arrived" + ) + +print(f"partial-frame memory check ok ({delta_kib} KiB growth)") +PY +fi + # ---- Leg 2: real-q interop against Rayforce server find_q if [[ -n "$QBIN" ]]; then