Repository navigation
feat(qwp): send batches of any size, with strict transactions and rollback - #127
Open
bluestreak01 wants to merge 7 commits into
Open
bluestreak01 wants to merge 7 commits into
bluestreak01 wants to merge 7 commits into
Conversation
The WebSocket sender's default byte trigger now sizes every batch to 90% of the frame the connection can carry: the smaller of the server's advertised X-QWP-Max-Batch-Size and the store-and-forward segment payload, or the server's default cap when nothing is advertised. Rows wider than about 2 KiB no longer fail with BatchTooLargeForCapException, in streaming and in transactional mode. auto_flush_bytes=off and a sender with every auto-flush trigger off keep the old behaviour. New symbols count toward the trigger, since their strings ride in the frame's dictionary section. Add strict transactional mode: transaction=strict, or auto_flush=off on WebSocket. Rows go out as deferred frames sized for the connection and only flush() commits, so a transaction can be any size, but it never commits in parts: a row that would take a table to the server's qwp.max.uncommitted.rows (advertised in X-QWP-Max-Uncommitted-Rows, or the 1M default) and a transaction that outgrows sf_max_total_bytes are refused with TransactionTooLargeException before anything reaches the server, leaving the transaction open. close() without flush() discards a strict transaction, and a pooled strict lease discards it on return. Add Sender.rollback() and the QWP rollback control frame. The sender asks for it with X-QWP-Request-Rollback and sends the frame only to a server that confirmed X-QWP-Rollback: enabled; rollback() waits for the ack, so a reconnect can never replay the rolled-back frames. reset() in transactional mode now rolls back frames already sent when the server supports it. Without server support rollback() aborts the transaction, and a strict reset() does too; a non-strict reset() keeps committing those frames on the next flush, as before.
Cover a strict transaction whose batch splits per table within the cap, and a strict pooled lease returned with only unsent rows, which keeps its connection and sends no rollback frame.
reset(), rollback() and returning a strict pooled lease discard the open transaction by publishing the rollback frame. During a server outage they wait up to 30 s for its ack, then throw and leave the sender aborted: every later row, flush, drain and rollback throws, even after the server comes back and acks the replayed rollback frame. Store-and-forward promises that an outage never fails a producer call. The new tests take the server down after a transaction's first frames went out, discard the transaction, commit the next one and bring the server back on the same endpoint. They fail until discarding stops waiting for the server. TxnServer now models the server per connection: wire sequences restart on every connection, and it records which frames a commit frame commits, so a test can assert that discarded rows never commit. A connection that did not confirm rollback reads the rollback frame as an empty deferred frame, as a server without rollback support does. TestWebSocketServer records which connections confirmed rollback.
reset(), rollback() and returning a strict pooled lease waited up to 30 s for the rollback frame's ack, then latched the transaction as aborted: every later row, flush, drain and rollback threw, even after the server came back and acked the replayed rollback frame, and the error blamed missing server rollback support. A transient outage broke the producer for good, which store-and-forward must never do. The wait stood in for a guarantee the producer cannot see: a reconnect that reaches a server without rollback support, such as an older node after a failover, must not replay the rolled-back frames there, as that server reads the rollback frame as an empty deferred frame and commits the frames with the next commit. The I/O loop knows which server each connection reached, so the guarantee moves there. The store-and-forward log already orders the rollback frame ahead of every later frame on every connection. On a connection whose server did not confirm X-QWP-Rollback the loop never sends a rollback frame: once the server acked every commit-bearing frame sent before it, the loop acknowledges the frame itself and recycles the connection. Closing the connection discards the rolled-back frames server-side, as every server does with what a closed connection leaves uncommitted, and the next connection starts past the frame, the way the orphaned-tail retirement already works. Replays after a restart and background drains run through the same loop, so they get the same guard. rollbackOpenTransaction() now only publishes the frame; the timeout, its test setter and the abort on timeout are gone. Sender.rollback() no longer promises to block, and Sender.reset() documents the transactional discard. On a connection that confirmed rollback, trySendBatch adds one cached boolean check per frame and no extra writes. CursorWebSocketSendLoopRollbackTest drives the loop against a model of the server's commits that drops what a closed connection had in flight: the rollback frame never reaches a server without rollback support, the loop waits for commit acks (durable ones in durable-ack mode) before recycling, only the exact rollback frame is held back, consecutive rollbacks, a recovered log, the background drainer, a per-connection capability, and a fuzz test over random transactions and mixed servers (questdb.fuzz.rollback.iterations). A sender-level test fails over to a server without rollback support. testRollbackAckTimeoutAbortsTransaction asserted the abort this commit removes; testUnacknowledgedRollbackDoesNotStallTheSender replaces it. Tests that read the server's frames right after reset(), rollback() or a lease close now wait with drain() or for the rollback frame, since those calls no longer block, and assert what the server committed.
auto_flush=off made a WebSocket sender a strict transaction of any size, while auto_flush_rows=off with auto_flush_interval=off, the same trigger state, built a non-transactional sender with no byte trigger: each flush went out as one batch, which failed past about 2 MiB per table, and close() committed. The builder keyed the mode on the auto_flush=off spelling, and recomputeFrameBudget() turned the byte trigger off for exactly that non-transactional sender. HTTP already treats both spellings alike. The builder now reads the trigger state. A sender with no auto-flush trigger (no row or interval trigger, no positive auto_flush_bytes) is in strict transactional mode unless transaction= picks the mode. transaction=on keeps it non-strict, so close() commits; auto_flush=off used to override it to strict. transaction=off is rejected for every spelling, as nothing but flush() could commit and a non-transactional flush must fit one frame. auto_flush=off with a positive auto_flush_bytes is rejected, as auto_flush=off with auto_flush_rows=N already was. auto_flush_bytes=off keeps its contract: no frame sizing, so each batch must fit one frame per table. recomputeFrameBudget() no longer special-cases the manual-flush sender, which the builder can no longer build. Behaviour change: a sender, or a pool, configured with auto_flush_rows=off and auto_flush_interval=off now discards the rows it did not flush when it closes or returns a lease, like HTTP and auto_flush=off, where it used to commit them. close() and lease return now log a WARN with the row count when they discard an uncommitted strict transaction, so a caller that relied on close() to commit sees it. A lease returned without server rollback support now aborts its transaction before the pool retires the connection, so the retiring close() does not report the same rows again. QwpLargeTransactionTest runs every spelling of "no trigger" (both config strings and both builder forms) through the same strict transaction of any size and the close() discard, and covers transaction=on committing on close, the rejections, auto_flush_bytes as the only trigger, auto_flush_bytes=off, one trigger left, the WARN on close and on lease return with and without server rollback, and an aborted lease retiring quietly. It replaces testManualFlushKeepsOneCommitPerFlush, which pinned the old split. LineSenderBuilderWebSocketTest and QuestDBBuilderTest cover the builder and the pool config paths.
A strict transaction refused a row inconsistently. At the server's qwp.max.uncommitted.rows the row that hit the limit stayed out of the transaction, but at the client's sf_max_total_bytes the row whose auto-flush found no room stayed in it, although at() threw all the same. A caller could not tell whether flush() would commit that row. The auto-flush that finds no room now keeps its rows pending and returns, since the room kept for the commit holds them, and the next row throws. A row whose at() throws TransactionTooLargeException is never part of the transaction, at either limit. flush() still throws when the client buffer cannot hold it, and sends nothing. The errors and the exception's javadoc name the setting to raise instead of suggesting code that commits part of the transaction. A row refused at the client buffer reports the size the transaction would have reached, as a refused flush does, rather than the bytes already sent, which read 0 when nothing had been sent yet.
A transaction=on transaction holds its frames in the store-and-forward buffer until the server commits it, and the server acks none of them before that, so the buffer cannot trim them. The row whose frame no longer fit blocked for sf_append_deadline_millis (30 s by default) and then threw, and close() then committed the frames already sent and dropped the rest, under an error that did not say so. Before a flush that does not fit beside the open transaction's frames, the sender now commits those frames with a bare commit frame and goes on, as the server already does at qwp.max.uncommitted.rows. Every deferred flush leaves room for that commit frame, so it never waits for the buffer to drain. The sender logs a warning on the first part of each transaction it commits this way, and rollback() and reset() warn that the parts already committed stay. Strict mode still refuses such transactions. The client-buffer simulation strict mode uses now tracks transaction=on transactions too. A single flush larger than the whole buffer still blocks, as before; only auto_flush_bytes=off builds one.
Contributor
[PR Coverage check]😍 pass : 357 / 378 (94.44%) file detail
|
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
BatchTooLargeForCapException.transaction=strictcommits only onflush(), at any size.auto_flush=offandauto_flush_rows=off;auto_flush_interval=offboth select strict mode.TransactionTooLargeException.transaction=ontransactions larger thansf_max_total_bytescommit in parts instead of hanging 30 s.Sender.rollback()discards the open transaction, including rows already sent.reset()in transactional mode now discards rows already sent, on servers with rollback support.auto_flush_bytesnow flushes at 90% of the frame size.close()discards unflushed rows instead of committing.qwp.max.uncommitted.rowsper table.transaction=offwith row and interval triggers both off fails at build.The byte trigger now sizes each batch from the frame the connection can carry: the server's advertised
X-QWP-Max-Batch-Size, capped by the store-and-forward segment, or the server's 2 MiB default when nothing is advertised. An explicitauto_flush_bytes=offkeeps the old behaviour: no byte trigger.The builder picks the transaction mode from the auto-flush triggers, not from how they were turned off, so both spellings build the same sender, as they already do on HTTP. A sender with no row or interval trigger and no positive
auto_flush_bytescommits only onflush(): it is strict unlesstransaction=onkeeps it non-strict. Closing it, or returning its pooled lease, withoutflush()discards the open transaction and logs a warning with the row count.Strict mode ships rows as deferred frames like
transaction=on, but refuses a row that would take a table past the server's advertisedX-QWP-Max-Uncommitted-Rows(1M when not advertised), or the transaction past the client buffer, which must hold it until the server commits it; the refused row is never part of the transaction, and the error names the setting to raise. An auto-flush that finds no room keeps its rows pending in the room reserved for the commit, so the next row is the one that fails.close()withoutflush()discards the transaction, and a pooled lease discards it on return.A
transaction=onsender now tracks how its open transaction fills the store-and-forward buffer, as strict mode does. Before a flush that would not fit beside the frames already sent, it commits them with a bare commit frame and logs a warning, as the server commits atqwp.max.uncommitted.rows, androllback()warns that those parts stay; every deferred flush keeps room for that frame, so it never waits for the buffer to drain.rollback()hands the QWP rollback control frame to store-and-forward without waiting for the server, asflush()does, so an outage never stalls or fails it. The frame goes out behind the transaction's frames and ahead of anything sent later, and only to a server that confirmedX-QWP-Rollback: enabled: when a reconnect reaches a server without rollback support, the I/O loop waits for the acks of earlier commits, then closes that connection, which discards the rolled-back frames. Without server support at call time,rollback()aborts the transaction, so the next flush cannot commit the discarded rows.Tandem: questdb/questdb#8343