questdb/ingress/sender/
qwp_ws_ownership.rs

1/*******************************************************************************
2 *     ___                  _   ____  ____
3 *    / _ \ _   _  ___  ___| |_|  _ \| __ )
4 *   | | | | | | |/ _ \/ __| __| | | |  _ \
5 *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
6 *    \__\_\\__,_|\___||___/\__|____/|____/
7 *
8 *  Copyright (c) 2014-2019 Appsicle
9 *  Copyright (c) 2019-2025 QuestDB
10 *
11 *  Licensed under the Apache License, Version 2.0 (the "License");
12 *  you may not use this file except in compliance with the License.
13 *  You may obtain a copy of the License at
14 *
15 *  http://www.apache.org/licenses/LICENSE-2.0
16 *
17 *  Unless required by applicable law or agreed to in writing, software
18 *  distributed under the License is distributed on an "AS IS" BASIS,
19 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 *  See the License for the specific language governing permissions and
21 *  limitations under the License.
22 *
23 ******************************************************************************/
24
25//! QWP/WebSocket progress ownership.
26
27use std::fmt;
28use std::sync::Arc;
29
30/// Controls whether a QWP/WebSocket [`crate::ingress::Sender`] starts its
31/// background progress runner or requires the caller to drive progress.
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum QwpWsProgress {
34    /// Start a background runner that sends frames, receives ACKs, reconnects,
35    /// and replays as needed. This is the default and matches Java's sender.
36    Background,
37    /// Do not start a background runner. The caller must call
38    /// [`crate::ingress::Sender::drive_once`] or
39    /// [`crate::ingress::Sender::wait`] to advance WebSocket progress.
40    Manual,
41}
42
43/// Structured server-side error observed by a QWP/WebSocket sender.
44///
45/// This mirrors Java's `SenderError` shape, but Rust exposes it through
46/// polling on [`crate::ingress::Sender`] instead of a callback dispatcher.
47#[non_exhaustive]
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct QwpWsSenderError {
50    /// Server-distinguishable error category.
51    pub category: QwpWsErrorCategory,
52    /// Policy the client applied after observing the error.
53    pub applied_policy: QwpWsErrorPolicy,
54    /// Raw QWP status byte. `None` for WebSocket protocol violations, which do
55    /// not carry a QWP status byte.
56    pub status: Option<u8>,
57    /// Human-readable message provided by the server or derived from the
58    /// WebSocket close reason.
59    pub message: Option<String>,
60    /// Server's per-frame QWP message sequence. `None` for WebSocket protocol
61    /// violations, which do not carry a QWP message sequence.
62    pub message_sequence: Option<u64>,
63    /// Inclusive lower bound of the affected frame sequence number span.
64    pub from_fsn: u64,
65    /// Inclusive upper bound of the affected frame sequence number span.
66    pub to_fsn: u64,
67}
68
69/// Producer-thread callback invoked for structured QWP/WebSocket server
70/// diagnostics.
71///
72/// The callback runs synchronously from sender API calls such as
73/// [`crate::ingress::Sender::flush`]. It must not call back into the same
74/// sender. For a terminal diagnostic, the sender's terminal state and
75/// pollable diagnostic are committed before the callback is invoked.
76#[derive(Clone)]
77pub struct QwpWsErrorHandler {
78    handler: Arc<dyn Fn(&QwpWsSenderError) + Send + Sync>,
79}
80
81impl QwpWsErrorHandler {
82    /// Create a handler from a closure.
83    pub fn new<F>(handler: F) -> Self
84    where
85        F: Fn(&QwpWsSenderError) + Send + Sync + 'static,
86    {
87        Self {
88            handler: Arc::new(handler),
89        }
90    }
91
92    pub(crate) fn log_default() -> Self {
93        Self::new(default_qwp_ws_error_handler)
94    }
95
96    pub(crate) fn handle(&self, error: &QwpWsSenderError) {
97        (self.handler)(error);
98    }
99}
100
101impl fmt::Debug for QwpWsErrorHandler {
102    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
103        f.write_str("QwpWsErrorHandler { .. }")
104    }
105}
106
107fn default_qwp_ws_error_handler(error: &QwpWsSenderError) {
108    let status = error
109        .status
110        .map(|status| format!("0x{status:02x}"))
111        .unwrap_or_else(|| "none".to_string());
112    let sequence = error
113        .message_sequence
114        .map(|sequence| sequence.to_string())
115        .unwrap_or_else(|| "none".to_string());
116    let message = error.message.as_deref().unwrap_or("");
117    if error.applied_policy == QwpWsErrorPolicy::Terminal {
118        log::error!(
119            target: "questdb::ingress",
120            "QWP/WebSocket server rejected batch [category={:?}, policy={:?}, status={}, fsn=[{},{}], seq={}, msg={}]",
121            error.category,
122            error.applied_policy,
123            status,
124            error.from_fsn,
125            error.to_fsn,
126            sequence,
127            message
128        );
129    } else {
130        log::warn!(
131            target: "questdb::ingress",
132            "QWP/WebSocket server rejected batch [category={:?}, policy={:?}, status={}, fsn=[{},{}], seq={}, msg={}]",
133            error.category,
134            error.applied_policy,
135            status,
136            error.from_fsn,
137            error.to_fsn,
138            sequence,
139            message
140        );
141    }
142}
143
144/// Lifetime totals reported by a QWP/WebSocket [`crate::ingress::Sender`].
145///
146/// Mirrors the `getTotal*` counters on Java's `QwpWebSocketSender` so the
147/// QuestDB Enterprise e2e harness (questdb-ent/e2e) can compare the same
148/// signal across language bindings. All counts are cumulative from the
149/// moment the sender was constructed: they never reset, and they survive
150/// reconnects.
151#[non_exhaustive]
152#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
153pub struct QwpWsTotals {
154    /// Frames handed to the transport for writing, regardless of whether the
155    /// server has acknowledged them.
156    pub frames_sent: u64,
157    /// Frames handed to the transport while catching up through the
158    /// publication boundary captured at a successful reconnect. Initial
159    /// publication, including recovered store-and-forward frames sent on the
160    /// first connection, is not counted as replay.
161    pub frames_replayed: u64,
162    /// Server responses interpreted as ACKs: ordinary OK, DurableOk, and
163    /// stand-alone DurableAck position notifications.
164    pub acks: u64,
165    /// Reconnect attempts initiated, including ones that returned
166    /// immediately because the retry budget was exhausted.
167    pub reconnect_attempts: u64,
168    /// Reconnect cycles that completed successfully and resumed publication.
169    pub reconnects_succeeded: u64,
170    /// Server-sent Reject responses (any policy: retriable, terminal, durable,
171    /// presend).
172    pub server_errors: u64,
173}
174
175/// Server-distinguishable QWP/WebSocket error category.
176#[non_exhaustive]
177#[derive(Debug, Clone, Copy, PartialEq, Eq)]
178pub enum QwpWsErrorCategory {
179    /// Server-side schema mismatch, such as a missing column, type clash, or
180    /// NOT NULL violation.
181    SchemaMismatch,
182    /// Malformed QWP payload.
183    ParseError,
184    /// Server-side internal failure.
185    InternalError,
186    /// Authentication or authorization failure.
187    SecurityError,
188    /// Non-critical server write failure.
189    WriteError,
190    /// Server reports that the endpoint is temporarily not writable.
191    NotWritable,
192    /// WebSocket protocol violation, including poison-frame escalation.
193    ProtocolViolation,
194    /// Unknown QWP status byte.
195    Unknown,
196}
197
198/// Policy applied by the sender after observing a QWP/WebSocket server error.
199#[non_exhaustive]
200#[derive(Debug, Clone, Copy, PartialEq, Eq)]
201pub enum QwpWsErrorPolicy {
202    /// Tear down the connection, reconnect, and replay from the unresolved
203    /// store-and-forward watermark.
204    Retriable,
205    /// Tear down and replay, but classify separately for diagnostics because
206    /// another endpoint/role may be required.
207    RetriableOther,
208    /// Latch the error as terminal. The bytes remain in store-and-forward
209    /// storage for inspection or a later compatible client/server.
210    Terminal,
211}
212
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub(crate) struct QwpWsRoleReject {
215    pub(crate) role: String,
216    pub(crate) zone: Option<String>,
217}
218
219impl QwpWsRoleReject {
220    pub(crate) fn new(role: &str, zone: Option<&str>) -> Self {
221        Self {
222            role: role.to_string(),
223            zone: zone.map(str::to_string),
224        }
225    }
226
227    pub(crate) fn is_transient(&self) -> bool {
228        self.role.eq_ignore_ascii_case("PRIMARY_CATCHUP")
229    }
230}
questdb/ingress/sender/qwp_ws_ownership.rs

questdb/ingress/sender/
qwp_ws_ownership.rs
