Repository navigation
feat(q): a connection rayforce-q opened takes frames up to 2 GiB - #25
Conversation
Every frame was capped at 256 MiB: the plain client's response, an attached connection's header and decompression. A q publisher's catch-up passed it. Replaying one day, 9 of 695 frames were 273 to 661 MiB uncompressed. The largest came in 116 MiB on the wire, so the header checks let it through and decompression refused it. The limit now follows who opened the connection. One rayforce-q opened trusts the peer it chose and takes Q_MAX_CLIENT_BODY, INT32_MAX less the header, the most a classic header can declare. Capability 3 lets a peer send no more. That is q_exchange, q_decode, q_decode_cols and a connection under q_conn_attach. One a q_serve listener accepted keeps Q_MAX_BODY, 256 MiB, so a stranger still cannot make the process allocate gigabytes. The reply queue in q_send_result stays at 256 MiB. q_decompress takes the limit from its caller. The public q_decode and q_decode_cols keep their signatures and use the client's limit. The rx machine calls q_decode_bounded with the connection's own, declared in q_server.h so q.h gains nothing. driver --frame-limit-selftest builds a 300 MiB long vector, sent once plain and once q-compressed, with a small encoder for q's compression. On a socketpair with a forked peer, the plain client decodes it whole, and an attached connection hands it whole to the bound upd. A header declaring 4 GiB still closes an attached connection. Against driver --serve, the listener still refuses a 300 MiB request. run.sh runs it in the server leg, which needs no q. The codec selftest's oversized compressed body now declares 4 GiB, past both limits.
A pushed frame that failed to decode reached the handler path as an error, which logged "q: async message raised an error" and dropped the frame. The connection stayed up. A subscriber then held a day missing that batch, and nothing made it reconnect and replay it. That is how the frames over the old 256 MiB limit were lost. On a connection this process opened, a frame that decodes to NULL or to an error now closes the connection, with a log line giving the frame's declared size and the decode's reason. A q peer cannot push an error value, so an error there is the decode's own. A subscriber sees the handle gone and reconnects. A push that decodes and whose handler raises is still logged and kept: that is the handler's failure, not the wire's. A response that fails to decode still reaches the waiting q_conn_send as an error. An inbound connection is unchanged. The error object carries only the core's short code, so the reason reads "q: buff" for a buffer underflow, as the codec selftest already matches "q: malf". driver --frame-limit-selftest pushes (`upd;v) cut to half on an attached connection, once plain, where the decode is an error object, and once compressed, where it is NULL: each closes it. The compressed one sent to driver --serve leaves that connection answering.
q_decode_bounded is declared in q_server.h and defined in q.c, which included only q.h, so nothing checked that the two signatures agree. If one changed and the other did not, both files would still compile and link, and q_server.c's call would pass its arguments wrong. q.c now includes q_server.h when __has_include finds it beside q.c. A binding that copies q.c and q.h alone, as INTEGRATING.md tells it to, still builds. With -Wmissing-prototypes, q.c no longer warns about q_decode_bounded.
The core's ray_error keeps only the first 7 bytes of its code, and q.c's decoder passed no message, so its errors read "q: buff", "q: unsu" or "q: symb". The log line that closes an attached connection showed only that, and it gave the size on the wire, which for a refused catch-up frame was 116 MiB against 661 MiB uncompressed. The decoder's errors now go through q_des_err, which puts the text in the message too, where ray_error_msg() reads it whole. The 7-byte codes are unchanged, so a caller matching on them still matches. q_read_body copies the message right after the decode, before another error can replace it, and reads a compressed frame's declared uncompressed size before freeing the body. The close log now reads, for example: "a pushed frame of 103 bytes, 8025 uncompressed, did not decode, closing the connection: q: decompression failed".
…ry element Two regressions passed the frame-limit selftest, each found by changing the code on purpose. A listener handed the client's 2 GiB at decompression passed, because the only listener check sent an uncompressed header, which the header limit refuses first. And a decompression that zeroed every back-reference passed, because the big vector was 7s with zeros between and only its first and last elements were checked. The listener now gets a compressed sync request, (`count;v), a few MiB on the wire and 300 MiB uncompressed, and must answer an error, not the count. The same request with 1000 elements is counted first, so the refusal is the limit's and not a malformed frame's. The big vector now runs a thousand equal elements at a time through 250 values, which still compresses to about 7.3 MiB, and every element is compared. Each check fails against its regression.
The docs said a frame costs about twice its uncompressed size at its peak. Measured on the replayed catch-up's largest frames, decoding added about 4.4 times: a 661 MiB frame peaked at about 3 GB. The body as received, the decompressed copy and the decoded objects are alive together, and the objects alone can be several times the input. They said a frame over its limit closes the connection. Only a header over the limit does. A compressed frame declaring too much is refused at decompression and follows its connection's rule for a frame that does not decode, so a listener's connection stays open. They described the close on an attached connection as a push cut short. It is any frame that does not decode: a sync request from the peer, which gets no reply, a q type rayforce-q does not decode, such as a timespan or a lambda, or memory running out. The failure is permanent for that frame, so a subscriber that replays from the start meets it on every reconnect and should back off. The comment in q_read_body said a q peer cannot push an error value. A kdb+ peer cannot; a rayforce-q peer can, and its error is treated the same. The changelog also records that decode errors now carry their text.
|
@morsel-ray could you please review the changes? |
morsel-ray
left a comment
morsel-ray
left a comment
There was a problem hiding this comment.
Review on exact676d1dd1ce0af2f5fa6aa58c1e361594a39379e0. Read the production delta, frame-limit test paths and relevant documentation, alongside the independent source report.
The role-specific limits and RESPONSE-before-pushed-request-close ordering look coherent in that scope, but the unchanged stable-core pairing currently does not compile, and the allocation-bound documentation overstates the contract. Details inline.
My attempted q-free adapter/server build against pinned core5d5f47c2 stopped at the undeclared call; no large-frame suite passed or ran from that attempt. Separately compiled the codec-only ASan/UBSan probe: complete small compressed body succeeds at its exact cap, fails below it with zero malloc calls; a cap-removal counterfactual fails this control. Replayed the four-byte fixture with a protective external allocation-refusal wrapper to demonstrate the coverage gap without allocating gigabytes. Source/core archive hashes remained unchanged.
No real-q, replay workload, idle-poll push delivery, packet-loss, peak-memory benchmark or combined#24+#25 clearance. Please fix the explicit header dependency, accurate body-cap/error wording and limit-sensitive small fixture before final approval.
| * which says more about a refused frame than its size on the wire. */ | ||
| char reason[256] = {0}; | ||
| if (req != NULL && RAY_IS_ERR(req)) { | ||
| const char *msg = ray_error_msg(); |
There was a problem hiding this comment.
P2 — This new call fails to compile against the existing core v2.11.5 pairing (5d5f47c2): Clang C17 reports undeclared ray_error_msg, then incompatible int-to-pointer initialization. I reproduced it in the adapter/server build before any runtime test. The symbol exists in that archive, but its owning declaration is in core/runtime.h, not the stable public header or this TU's present includes. Please include the owning header explicitly (or otherwise preserve the supported-core contract), rather than depending on newer transitive includes or suppressing the diagnostic. The same pinned archive/headers built and ran #24 successfully.
| header, the most the classic q header can declare. That covers `q_connect`'s | ||
| responses, read through `q_send` or `q_exchange` with `q_decode` or | ||
| `q_decode_cols`, and a connection `q_conn_attach` put under a poll. A | ||
| connection a `q_serve` listener accepted is held to 256 MiB, so a stranger |
There was a problem hiding this comment.
P2 — 256 MiB bounds a frame body, not what an inbound peer can make the process allocate. Even a valid plain numeric vector keeps the receive buffer and newly allocated decoded vector live together; compressed input adds another buffer, and multiple connections/replies add more. The memory-amplification paragraph below correctly acknowledges this. Please remove the allocation-ceiling claim here, in the new q_server.c comment, and in CHANGELOG; describe the wire/decompressed-body cap without promising an RSS/total-allocation bound. No new runtime scan or guard is requested. Also qualify the handler-error wording below/CHANGELOG as async: the sync branch still encodes an error response.
| uint32_t oversized_size = UINT32_MAX; | ||
| memcpy(oversized_compressed, &oversized_size, sizeof oversized_size); | ||
| err[0] = '\0'; | ||
| r = q_decode(oversized_compressed, (int64_t)sizeof oversized_compressed, 1, |
There was a problem hiding this comment.
P3 — This four-byte truncated stream does not distinguish size-limit enforcement from malformed-stream/allocation failure. In a safe external probe, omitting only the decompression cap still satisfies this fixture's assertion when a test allocator refuses the resulting huge allocation. No huge allocation was actually made. A complete valid compressed I64 atom (9-byte body) decodes at cap9 and is rejected before malloc at cap8; that compact case fails specifically when the guard is removed. Please add a valid-small-body/two-cap control through q_decode_bounded, and avoid claiming this truncated fixture establishes pre-allocation refusal. The large valid client/listener tests are separate useful evidence.
ray_error_msg is declared in the core's core/runtime.h, not in the public rayforce.h, so q_server.c did not compile against core v2.11.0, the oldest core rayforce-q supports. Include the owning header instead of relying on a newer core's transitive includes.
…llocates The receive buffer, a decompressed copy and the decoded objects are alive together, over every connection, so the cap bounds one frame's body on the wire and decompressed, nothing more. Say so in INTEGRATING.md, the CHANGELOG and the q_server.c comment, and name the handler-error rule as the async one: a sync request still gets the error as its response. The frame-limit selftest also checks the decompression cap on its own: a valid 9-byte compressed long atom decodes at a cap of 9 and is refused at 8. The truncated fixtures show a cut stream fails, not that the cap holds.
How it looks from the user's side
A process subscribed to a q publisher through
q_connectandq_conn_attachreceives the publisher's catch-up as large frames. Replaying one day from a tick publisher, 9 of 695 frames were 273 to 661 MiB uncompressed. The largest came in 116 MiB on the wire.Before: every frame stopped at 256 MiB. The header checks passed those 9, because they were small on the wire, and decompression refused them. The handler path then logged
q: async message raised an error, dropped the frame and kept the connection open. The subscriber carried on with about 7.5 M rows missing and nothing to make it reconnect.After: a connection rayforce-q opened takes frames up to 2 GiB, and all 695 frames reach
upd. A frame on such a connection that still does not decode closes it with a log line, so the subscriber sees the connection gone and can reconnect instead of keeping a hole.Independent of #24: this branch starts from master and does not touch
q_connect.What changed
The limit follows who opened the connection (first commit).
Q_MAX_CLIENT_BODY,INT32_MAXless the header, the most a classic header can declare and what capability 3 lets a peer send. It applies toq_exchange,q_decode,q_decode_colsand a connection underq_conn_attach. The peer is one the caller chose.Q_MAX_BODYstays 256 MiB for a connection aq_servelistener accepted, so a stranger still cannot make the process allocate gigabytes. The reply queue inq_send_resultstays at 256 MiB too.q_decompresstakes the limit from its caller. The publicq_decodeandq_decode_colskeep their signatures. The rx machine callsq_decode_boundedwith the connection's own limit, declared inq_server.hsoq.hgains nothing.q.cincludesq_server.hwhen__has_includefinds it, so the definition is checked against the declaration, and a binding that copies onlyq.candq.hstill builds.An attached connection closes on a frame it cannot decode (second commit).
q.cdoes not decode (a timespan, a lambda), memory running out, and a sync request from the peer, which gets no reply. The failure is permanent for that frame, so a subscriber replaying from the start meets it on every reconnect and should back off.docs/INTEGRATING.mdsays so.q_conn_sendas an error.Decode errors keep their text. The core's
ray_errorkeeps 7 bytes of its code, and the decoder passed no message, so its errors readq: buff. They now carry the text in the message too, whereray_error_msg()reads it. The codes are unchanged. The close log reads, for example:Memory is the caller's to budget, and
docs/INTEGRATING.mdsays so. While a frame decodes, the body as received, the decompressed copy and the decoded objects are all alive. Measured on the replayed frames, decoding added about 4.4 times the uncompressed size: the 661 MiB frame peaked at about 3 GB.Tests
driver --frame-limit-selftest HOST PORT, run inrun.sh's server leg, so CI covers it without a q binary. A 300 MiB long vector is built in the test, sent once plain and once q-compressed. The driver gains a small encoder for q's compression, which matchesq_decompress; the compressed frame is about 7.3 MiB on the wire. The client side runs on a socketpair with a forked peer.q_exchangeandq_decode, plain and compressed, and every element is right.upd, plain and compressed, and stays open.driver --serve:(`count;v)that decompresses to 300 MiB with an error, after counting the same request with 1000 elements;Each check fails against the regression it guards. Against master's
q.candq_server.c, six fail. With the listener given the client's limit at decompression, the listener check fails. With every back-reference copy zeroed inq_decompress, both compressed 300 MiB checks fail. The selftest peaks at about 1.5 GiB resident and takes 1.5 s.Both pass. Legs that need a q binary are skipped, as in CI.
The codec selftest's oversized compressed body now declares 4 GiB, past both limits, where it declared 256 MiB + 1.
End to end, through a binding: rayforce-rs with this branch merged onto #24, subscribed to a replay of the recorded day. All 695 frames reached
upd(10.49 M rows, where 2.99 M did before), with no error on stderr. The process peaked at 4.8 GB resident, where it peaked at 3.4 GB with the 9 frames dropped.Out of scope