Standalone QWP/WebSocket error-handler reentrancy can mutate a live flush buffer and silently lose rows

[mtopolnik]

Summary

A standalone QWP/WebSocket sender invokes its Python error_handler synchronously while servicing a sender call. When a pending error notification is drained by flush(), the native call is already holding a Rust reference to the buffer being flushed. The handler can nevertheless mutate that same buffer through Python APIs.

This can silently change which rows the outer flush() publishes. In particular, clearing and repopulating the buffer from the handler can make flush() return success after publishing unrelated rows instead of the rows supplied by its caller.

This predates PR #140; it was identified during that review: #140 (comment)

Call path

1. Python Sender.flush() calls line_sender_flush(sender, c_buf, ...) after releasing the GIL.
2. The C FFI unwraps the arguments into &mut Sender and &mut Buffer, then calls sender.flush(buffer).
3. The QWP/WebSocket flush path calls drain_qwp_ws_error_notifications() before it reads and encodes the buffer.
4. Draining a notification invokes the Python error_handler synchronously on the same thread.
5. The callback reacquires the GIL and can call Buffer.clear(), Buffer.row(), or Sender.row() on the buffer participating in the outer flush.
6. The outer flush resumes, encodes whatever the buffer contains at that point, clears it, and may return success.

Relevant code:

• src/questdb/_client.pyx: Sender.flush() and _check_not_in_own_callback()
• questdb-rs-ffi/src/lib.rs: line_sender_flush()
• questdb-rs/src/ingress/sender.rs: flush_qwp_ws_buffer(), drain_qwp_ws_error_notifications(), and Sender::flush()

Observed behavior

Two forms were reproduced:

• Appending rows from the handler grew the buffer from 434 bytes to 780,459 bytes during the live native call, including a reallocation-capable mutation while Rust retained its buffer borrow.
• Clearing the buffer and adding two rows from the handler caused the outer flush() to return success after publishing those two rows instead of the fifty rows originally passed to it.

The second case is silent row loss: the caller receives a successful return even though its original batch was discarded.

Existing guard does not cover buffer mutation

_check_not_in_own_callback() currently protects sender operations that re-enter or free the live native sender, including flush(), close(), and the QWP/WebSocket progress/watermark methods. The existing reentrancy test covers sender methods such as published_fsn, acked_fsn, drive_once, flush, and close.

It does not prevent mutation through:

• Sender.row() on the sender's internal buffer
• Buffer.clear()
• Buffer.row()
• Buffer.dataframe()
• Buffer.reserve() or any other operation that can mutate/reallocate the same buffer

Both the internal sender buffer and a buffer explicitly supplied to Sender.flush(buffer) need consideration. Buffer._check_not_in_row() is a different guard: it detects mutation while a row is being constructed, not mutation while the buffer is participating in a flush.

Sender.dataframe() over WebSocket uses a separate direct columnar connection and does not mutate the sender's row buffer, so it is not part of this specific buffer-loss mechanism. It may still merit a separate callback-reentrancy policy audit.

Expected behavior

A callback must not be able to mutate a buffer that is participating in an active native flush. An attempted mutation should either:

• fail with QuestDBError(InvalidApiCall) before changing the buffer, or
• be made safe by restructuring callback delivery so no Rust buffer reference is live when Python is invoked.

Operations on an unrelated sender/buffer should remain allowed where safe; callback identity alone should not globally prohibit independent objects.

Test coverage requested

• Handler calls Buffer.clear() during flush(); original rows are not silently discarded.
• Handler calls Buffer.row() / Sender.row() during flush(), including enough data to force growth/reallocation.
• Explicitly supplied buffers are covered as well as the sender's internal buffer.
• Other buffer-mutating entry points, including Buffer.dataframe() and Buffer.reserve(), are audited and tested.
• The existing behavior allowing a shared handler to operate on a different sender remains covered.