Skip to content

feat: poll-driven Q subscriptions and serving from an embedded runtime - #20

Merged
protocolstardust merged 12 commits into
RayforceDB:masterfrom
ihrfv:feat/q-subscribe-core-2.11
Oct 12, 2026
Merged

protocolstardust merged 12 commits into
RayforceDB:masterfrom
ihrfv:feat/q-subscribe-core-2.11

Conversation

@ihrfv

@ihrfv ihrfv commented Oct 9, 2026

Copy link
Copy Markdown
Contributor

Runs on the core and rayforce-q already pinned on master (core v2.11.0 dc9b12a, rayforce-q 2.2.0 139e4fd); no submodule moves. The branch is three commits behind master (the table column validation fix) and merges cleanly.

What

Subscriptions and the Q wire

  • QConnection::attach hands a connection to the runtime's event loop and returns a Subscription that receives frames the peer pushes unsolicited
  • Poll is a borrow of the loop the runtime owns; Poll::serve_q(port) registers rayforce-q's listener on it, so a runtime that defines upd is rayforce -q embedded
  • env::bind_vary / env::bind_unary bind a Rust handler type under the name a publisher calls; the generated trampoline borrows the arguments and catches panics
  • q::encode is the mirror of q::decode_response, so a process can publish over a transport it owns
  • rayforce-sys compiles q_server.c beside q.c and binds the poll and env symbols from the core's own headers

Serving the native IPC

  • Poll::serve_ipc(port) returns a ServeIpc builder with budget and clock. Its start listens through .sys.listen, installs a guard over every request, and returns an IpcServer
  • A request is cancelled through the core's interrupt when it runs past the budget, or when its client closes the connection. A client's own cancel, the urgent byte, works as before
  • IpcServer::drain hands back each request's end: the whole text, how long it was evaluated, and Answered, Failed(ErrorCode) or Cancelled(Budget | Hangup | Client)

Build

  • The vendored submodules are addressed over https. With git@github.com: in .gitmodules, every git = "https://..." consumer of this crate fails to fetch: Cargo checks submodules out through libgit2 and offers it no SSH key

Why

q.c's client is blocking request/response: a frame a publisher pushes between two requests is consumed by the next q_send as its own reply, and every response after it is one frame out of step. The only fix is for the poll to own the reads and route frames by message type, which is what rayforce-q 2.1.0's q_server.c does — hence the floor on that version.

The native server evaluates one request at a time, and the core stops a request only when its own client cancels it. In an RDB built on this crate, queries that ran 120 to 214 s held every other client, and the writer on the Q wire, for that long. A client-side timeout does not help: the client drops the connection and the query runs on.

How the IPC guard works

  • When .ipc.on.sync or .ipc.on.async is set, the core passes it each request's payload in place of evaluating it, but only if the hook is a Rayfall lambda (hook_lookup, src/core/ipc.c). So the hook is a one-line lambda over a native function bound with env::bind_unary
  • A watcher thread calls ray_request_interrupt when a request passes the budget, or when a non-blocking peek on the request's socket reads zero bytes. The socket comes from .ipc.handle and ray_poll_get
  • The watcher and the guard share a lock, and the guard clears the core's interrupt flag under it when a request ends, so an interrupt never lands on whatever the loop evaluates next
  • No change to the core

Limits

  • The Q wire never reads the hooks, so a writer there is never cancelled. It can still wait for every request queued ahead of it: when the running request ends, the core takes what is ready on the two wires in an order of its own, and took a queued query before the push in about one run in four. The writer test pins that bound
  • One IPC server per process, because the hooks are globals. A client can redefine them, so a server that has to hold its budget should serve clients in restricted mode
  • Dropping IpcServer or QListener stops the guard, not the serving: the public header has no ray_poll_deregister, so the socket closes with the runtime
  • A cancel that lands while the core interns symbols in bulk can corrupt the shared symbol table (Cancelling a symbol-domain batch intern can corrupt the shared domain rayforce#715). This change makes cancels routine, so that fix matters more
  • Hangup detection has been run on macOS only; CI covers Linux

Verification

At this branch's HEAD, on macOS:

  • cargo fmt --all -- --check and cargo clippy --workspace --all-targets -- -D warnings: clean
  • cargo test --workspace: 168 passed, 0 failed, including 12 in q_subscribe.rs, 3 in q_serve.rs and 8 in ipc_serve.rs
  • cargo package -p rayforce-sys --list carries q_server.c / q_server.h
  • Poll, Subscription and QListener are !Send/!Sync, each pinned by a compile_fail doctest paired with a control. IpcServer carries the same marker but no doctest yet

🤖 Generated with Claude Code

ihrfv added 12 commits October 7, 2026 12:29
- compile rayforce-q's q_server.c beside q.c, and bind ray_poll_get,
  ray_fn_unary, ray_fn_vary, ray_env_bind and ray_env_bind_flat from
  the core's own poll.h and env.h (CORE_PRIVATE_HEADERS / INTERNAL_FNS)
- q.c's client is blocking request/response, so an unsolicited frame is
  read as the next reply; q_server.c routes frames by type off a poll
- headers rather than a hand-copied block: C linkage matches on name
  alone, so a drifted copy is UB with no diagnostic, and ray_selector is
  a struct this crate dereferences
- rayforce-q 2.1.0 is a hard floor (q_conn_* is absent in 2.0.0); the
  vendored checkout is already past it
- q_server.c/h join the crate's include list and CI's packaged-crate
  check, so a published build.rs can find them
- a handler is a type implementing env::VaryFn / env::UnaryFn, bound
  under a name with env::bind_vary / env::bind_unary
- the core takes a bare extern "C" fn with no user-data slot, so a
  closure cannot be bound; a monomorphized trampoline per type makes the
  function address carry the identity
- the trampoline borrows the argument array and catches panics, so the
  FFI contract is not on every implementor
- both binders can refuse (capped table, reserved names), so both return
  codes are checked
- QConnection::attach hands the socket to the runtime's poll and returns
  a Subscription; Poll is a borrow of the loop the runtime owns
- the blocking client reads a pushed frame as the next reply, so a
  subscription is impossible unless the poll owns the reads
- Poll pins no heap handle and checks the runtime is live before every
  dereference; a pinned handle would keep a selector alive that nothing
  is left to close
- attach forgets the QConnection to suppress q_close and releases the
  heap handle by hand; missing it strands the heap until an unrelated
  Runtime::new is refused
- a Subscription is identified by selector id, rx data pointer and a
  claim token: ids are slot indices reused on disconnect, so an id
  compare alone closes the live peer that inherited the slot
- ipc.md gains a subscription section: why the blocking client cannot
  express one, and the Poll / attach / bind shape that can
- changelog entries sit under a fresh Unreleased heading above the
  shipped 1.1.0; Poll and Subscription being !Send is noted there rather
  than by editing the released list
- q::encode turns a Value into a complete wire message, the mirror of
  q::decode_response, so a process can publish over a socket it owns
- the header is always stamped SYNC; a publisher retypes byte 1 in place
- there is no native dict branch, so a true q dict (type 99) means
  splicing the marker between separately encoded keys and values
- Poll and Subscription get the same compile_fail pair plus control as
  Value; both carry a runtime generation checked with the same load
- Subscription<'static> is named on purpose: written bare the snippet
  fails for a missing lifetime, a pass for the wrong reason
- one marker per block: a combined block keeps passing when only one
  regresses (verified with an unsafe impl on each alone)
- the Q wire builders move to tests/support/wire.rs; each tests/ file is
  its own crate, so they lived twice and a third copy was due for the
  listener tests
- the server-side pieces come with them: a char vector, a long atom, and
  a frame reader that returns the message type with the body
- q.rs keeps its own copy until next touched, so this moves code rather
  than rewriting it
- Poll::serve_q(port) registers rayforce-q's listener on the runtime's
  poll and returns a QListener: sync strings are evaluated and answered,
  async (upd; payload) dispatches to whatever upd names
- a runtime defining upd in Rayfall and serving a port is rayforce -q
  embedded, which an RDB written from Rust could not get before
- QListener is a handle, not an owner: the public header has no
  ray_poll_deregister, so the socket closes with the runtime
- !Send/!Sync like every other handle, with the same compile_fail pair
- tests drive the listener from raw TCP peers on helper threads; a
  port-in-use test was dropped because the core binds with address reuse
- reverts 61a9553: with `git@github.com:` in .gitmodules, every
  `git = "https://..."` consumer of this crate fails to fetch, because
  Cargo checks a dependency's submodules out through libgit2 and offers
  it no SSH key
- https reads a public submodule from any clone, SSH origin or not, and
  a container or a CI runner without a key needs no url rewrite and no
  `net.git-fetch-with-cli`
- the README, the installation page, the changelog and the workflow
  comments drop the SSH note they carried
- tests/support/ipc.rs speaks the wire `rayforce -p` serves: the 2-byte
  handshake, a SYNC frame holding one string, and an answer decoded as
  far as a test reads it, a long or an error code
- on raw sockets, because a TcpClient needs a runtime on its thread and
  the one a process may hold is the server's under test
- libc as a dev-dependency, for the one thing std cannot send: the TCP
  urgent byte that is the core's cancel
- Poll::serve_ipc(port) is a builder, with budget and clock, whose start
  listens through `.sys.listen` and installs a guard; IpcServer::drain
  hands back each request's end: the whole text, how long it was
  evaluated, and Answered, Failed(code) or Cancelled(by whom)
- the server evaluates one request at a time and the core stops one only
  when its own client cancels, so a request that ran for minutes held
  every other client for minutes, a writer on the Q wire included
- the guard is the core's .ipc.on.sync/.async hooks, a one-line lambda
  over a native function, since the core calls a hook only when it is a
  lambda; a watcher thread interrupts past the budget or when a peek
  reads the client gone, under a lock that keeps an interrupt from
  landing after the request ends
- the Q wire never reads the hooks, so its writer is never cancelled; it
  can still wait for every request queued ahead of it, because the core
  orders the two wires itself, and the writer test pins that bound
- one per process, the hooks being globals; dropping the handle stops
  the guard, not the serving, as with QListener
feat(serve): serve the native IPC with a budget on every request
@protocolstardust
protocolstardust merged commit cd23e5c into RayforceDB:master Oct 12, 2026
2 checks passed
@ihrfv
ihrfv deleted the feat/q-subscribe-core-2.11 branch October 12, 2026 21:05
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants