Port the Java HA sender to the Rust, Python, and C/C++ clients#6
Draft
javier wants to merge 12 commits into
Draft
Port the Java HA sender to the Rust, Python, and C/C++ clients#6javier wants to merge 12 commits into
javier wants to merge 12 commits into
Conversation
Emit trade_id = <worker>-<1-based sequence> as a VARCHAR on each row, so completeness/gaps can be checked independent of timestamps and, with DEDUP UPSERT KEYS(timestamp, trade_id) on a single-worker client-timestamped run, store-and-forward replay after a failover is idempotent (drops the at-least-once boundary duplicates). High-cardinality, so a string column, not a symbol.
Add a Row id and dedup section covering the trade_id format, the completeness check, idempotent store-and-forward replay via DEDUP UPSERT KEYS(timestamp, trade_id), and the client-side-timestamp requirement.
Sender: - Detect the QWP upgrade "HTTP/1.1 400 Bad request" that a missing/invalid ILP token produces and append an explicit auth hint wherever it surfaces (endpoint-failed, worker error, probe-stopped). - Probe now reports the LIVE serving role via `switch status` each poll instead of the QWP handshake SERVER_INFO, which only refreshes on a reconnect and so goes stale after an in-place primary<->replica switch. Watchdog: - Monitor the primary's currentRole, not just health: react to a silent demotion to REPLICA as well as an unreachable node. - Bootstrap a fresh cluster: if no node is primary, promote the first healthy one in list order. - Fail over to the first HEALTHY node in preference order; adopt a node that is already PRIMARY instead of forcing another switch. - Add a configurable grace period (QDB_WD_GRACE_PERIOD, default 5s) so a manual operator switch can settle before the watchdog acts. - Never exit on its own; keep retrying when nothing can be promoted. - Remove the now-unused get_http_code and STARTUP_RETRIES. Update README and the sample env accordingly.
Build/target: - pom default questdb.client.version -> 1.3.6-SNAPSHOT (required; QWP and the lifecycle switch features are not stable earlier, and it is not on Central). - Drop the RECONNECT_BUDGET_EXHAUSTED switch case in both senders; that kind was removed from SenderConnectionEvent.Kind in 1.3.6, so it no longer compiles. Diagnostics: - Replace the brittle "400 Bad request" string-match with the client's typed signals: QwpAuthFailedException (401/403), WebSocketUpgradeException.isRoleMismatch() (421 = endpoint is a replica, no primary to accept writes), and status-400 (missing/malformed token). The listener passes the real throwable, not its message. Timeouts: - --retry-timeout now also drives QWP (reconnectMaxDurationMillis), not just ILP's retryTimeoutMillis, so one flag means the same "keep retrying" budget on both. - New --connect-timeout-ms (default 3000, 0 = OS default): connectTimeoutMillis on the senders and connect_timeout= on the probe, so a black-holed host fails over in seconds instead of riding the OS connect timeout. ILP left unbounded/unchanged. - Watchdog: make the role/health poll timeouts configurable (QDB_WD_CHECK_CONNECT_TIMEOUT / QDB_WD_CHECK_MAX_TIME; defaults 2/3, no behaviour change). Update README and the watchdog sample env accordingly.
The probe fallback printed only "node=(none)" and dropped the role entirely when `switch status` returned nothing, which was worse than before. Restore the role/node/zone from getServerInfo() on every line (it tracks connect + failover), and append the authoritative live role from `switch status` only when present. When switch status yields no live role, log why at most once per 30s: an error status+message, a returned batch with no *role* column (names listed), or no row batch at all -- so the empty case is diagnosable instead of silent.
…dshake The probe printed getServerInfo()'s role as the headline 'served by role=', but that is the QWP handshake role, refreshed only on connect/failover. After an in-place promotion (the read connection never drops) it stayed stuck at REPLICA even though the node was serving reads as PRIMARY. Make switch status' current_role the authoritative role shown, append '(switching -> ROLE)' while a switch is in flight, and demote getServerInfo() to supplying node/zone only -- its handshake role is now a clearly labelled fallback used solely when the status query is unavailable. target=any is unchanged, so replica-fallback reads still work.
The watchdog Configuration section only named a few QDB_WD_* vars inline and deferred the rest to the env example; enumerate all 14 (required + optional) with defaults and purpose. Also update the probe example to the current 'served by role=<current_role> node=... zone=...' format, covering the '(switching -> ROLE)' in-flight case and the labelled handshake fallback.
Add --protocol qwpudp for fire-and-forget UDP datagram ingest to the UDP port (:9007). It is ingest-only, unauthenticated (auth flags are warned about and ignored), and single-endpoint with no failover, store-and-forward, TLS, or query client (the probe is skipped). The worker flushes datagrams every --batch-size rows and, for a single worker, stamps rows client-side; multiple workers use server-side atNow(). Document the transport, including its best-effort nature and the burst/packet-loss behaviour, with guidance to keep --batch-size small for UDP.
Port the Java CsvParallelSender to Rust, built on the local questdb-rs (c-questdb-client) as a path dependency. Mirrors the Java client: - qwp (WebSocket) with per-worker store-and-forward, transactional commit cadence, and multi-host failover. - qwpudp (UDP) fire-and-forget datagrams: ingest-only, unauthenticated, single-endpoint, with upfront single-address validation. - ilp (HTTP) legacy transport. - Same connection options (address list, token / basic auth, TLS, zone, enterprise durable ack, reconnect budget) and timestamp semantics (single-worker client micros vs multi-worker at_now). - A Reader-based probe (qwp only) polling the latest ingested timestamp and the serving node's live role via 'switch status', with target=any replica-fallback reads and automatic failover. Batching is driven by explicit flush() calls since the Rust client has no auto-flush by design. Validated end-to-end against a local server, including a primary-crash failover with gap-free store-and-forward handoff.
Port the Java/Rust CsvParallelSender to Python on the questdb client's QWP + egress build (the sm_qwp_dataframe_bench branch, built from source; the PyPI release is ILP-only). csv_parallel_sender.py mirrors the other ports: qwp (WebSocket) with per-worker store-and-forward and manual flush cadence, qwpudp, and ilp, plus a probe using the pooled query Client (select ... limit -1 + switch status role reporting, target=any, failover). dataframe_demo.py shows pandas ingestion (Sender.dataframe), polars ingestion (Client.dataframe), and egress to pandas/polars via Client.query(...).to_pandas()/.to_polars(). Validated against a local server: qwp/ilp/udp ingest, a primary-crash failover with gap-free store-and-forward handoff, and row-threshold auto-flush. README documents the build-from-source requirement and the Python-specific differences.
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
1 participant
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.
This branch ports the Java
CsvParallelSender— the QWP high-availability sender (failover, store-and-forward, dual ingest/query, and the shared connection options) — to the three clients built on the C/Rust core (the "CRusty" clients): Rust, Python, and C/C++.Each port keeps the Java design (CSV replay loop, per-worker senders, timestamp / O3 semantics) and mirrors the same
--protocoltransports, connection options, and flags. Per-language differences are documented in each folder'sREADME.md.Status
rust/) — complete. QWP (WebSocket) with per-worker store-and-forward and multi-host failover, QWP/UDP, and ILP/HTTP; aReader-based probe (latest ingested timestamp + live serving role viaswitch status,target=anyreplica-fallback reads, automatic failover). Validated end-to-end, including a primary-crash failover with a gap-free store-and-forward handoff.python/) — complete. The same three transports plus a pooled queryClient, and a pandas/polars ingestion + egress demo (dataframe_demo.py). Built on the client's QWP + egress branch (built from source; the PyPI release is ILP-only). Validated end-to-end, including a primary-crash failover and auto-flush.c/) — pending.Draft while C/C++ is outstanding.