feat(core): pipelined egress reader

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "6.1.0"
+current_version = "7.0.0"
 commit = false
 tag = false
 

.claude/skills/review-pr/SKILL.md

@@ -130,6 +130,20 @@ Group the callsites from 2.5b by execution context. Typical contexts in this cod
 
 Every entry on this list must be reviewed in Step 3.
 
+### 2.5e Build profile facts
+
+**This sub-step runs at every level, including levels 0 and 1 where the rest of Step 2.5 is skipped.** A single `Cargo.toml` setting can flip the panic-safety story for the entire crate; agents must reason from the actual profile, not from defaults.
+
+Read `questdb-rs/Cargo.toml` and `questdb-rs-ffi/Cargo.toml` and record, with file:line citations:
+
+- **panic strategy** per profile (`[profile.release]`, `[profile.dev]`). If `panic = "abort"` in either, **every `catch_unwind` in that crate is a no-op for that profile** and every reachable panic is a process abort. Agents 2, 3, and 4 (and the level-0 inline review) must not credit `catch_unwind` as a panic guard under `panic = "abort"`. The only acceptable defense under abort-panic is proving no panic path exists.
+- **overflow-checks** per profile. If `overflow-checks = false` in release (the default), integer overflow wraps silently in release builds instead of panicking — bugs that look like panics in test builds disappear into wrong values in production. State which mode applies.
+- **`[profile.*.package.*]` overrides** if present — a per-dependency profile can reintroduce unwinding for one crate even when the workspace defaults to abort.
+- **`#[global_allocator]`** if defined anywhere in the workspace. A custom allocator changes the OOM behavior (some abort, some unwind, some return null).
+- **lto / codegen-units / strip** — informational; flag if they look unusual.
+
+A review without this section is incomplete. State the panic mode in one line at the top of every Step 3 agent prompt so the agent reasons from the right premise.
+
 ## Step 3: Parallel review
 
 Every agent receives:
@@ -141,6 +155,7 @@ Every agent receives:
 - **Bugs at callsites outside the diff outrank bugs inside the diff.** A confirmed bug in a file the PR did not touch but that calls a changed symbol is a P0 finding.
 - **"Looks correct in isolation" is not a valid conclusion.** Before clearing a changed symbol, the agent must walk the callsite inventory from 2.5b and explicitly state, per callsite, whether the new behavior is still correct there.
 - **The diff is the entry point, not the scope.** If the change surface map shows the symbol is reachable from N other files, the review covers N+1 files.
+- **Crate-wide settings affect untouched code.** A change to `Cargo.toml` (panic strategy, allocator, feature defaults, MSRV, profile overrides), a new `#[global_allocator]`, or a new `panic_handler` retroactively changes the safety story for every existing function in the crate — not just the diff. When `Cargo.toml`, build scripts, or workspace-level config files appear in the diff, the review covers the panic/allocation/overflow contract of the **entire affected crate**, not just the touched lines. The same applies when 2.5e records a profile fact (e.g. `panic = "abort"`) that invalidates existing safety patterns in untouched code.
 - A single finding of the form "in `test_line_sender.cpp` the new behavior of `line_sender_buffer_column_f64` causes Y" is worth more than five findings inside the diff.
 
 ### Agents
@@ -160,7 +175,12 @@ Launch the following agents in parallel.
 - **C++ exceptions escaping into C:** the C++ wrapper (`include/questdb/ingress/*.hpp`) is reachable from pure-C callers via inline forwarders. Any path where the wrapper can throw (`std::bad_alloc`, `std::system_error`, user-defined `throw`) and reach a C caller is undefined behavior. Verify wrapper functions called from C are `noexcept` or only invoked from C++ contexts.
 - **SIGPIPE on broken sockets:** writing to a closed peer raises SIGPIPE by default on Linux/macOS, killing the process. Verify TCP/HTTP write paths set `MSG_NOSIGNAL` or mask SIGPIPE.
 
-Every fallible operation must use `Result`/`Option` with proper error propagation. Every `extern "C"` function must wrap its body in `catch_unwind` or prove no panic path exists.
+**Panic strategy is the foundation.** Before reasoning about any panic guard, look up the `panic` setting from Step 2.5e:
+
+- **Under `panic = "abort"`**, `catch_unwind` is a no-op — it cannot catch anything because nothing unwinds. Every reachable panic is a process abort regardless of where the `catch_unwind` is placed. The only acceptable defense is *proving no panic path exists*: front-load every length check, replace `unwrap`/`expect`/indexing on wire-derived or caller-supplied values with `Result`-returning equivalents, validate before allocating, use `checked_*` arithmetic. A `catch_unwind` wrapper in this mode is misleading documentation, not a safety net — flag it if it gives the reader false confidence.
+- **Under `panic = "unwind"`**, every `extern "C"` function must wrap its body in `catch_unwind` AND every `Drop` impl on the unwind path must be panic-free (double-panic aborts the process). Fallible operations must use `Result`/`Option` with proper error propagation.
+
+State which panic mode applies in the agent's first sentence. Every panic-related finding must be evaluated under the actual mode, not the textbook one.
 
 **Agent 3 — FFI boundary safety:** Check every `#[no_mangle]` / `extern "C"` function. Verify: NULL pointer checks on all pointer arguments, proper error propagation across the FFI boundary (no panics escaping into C), correct ownership transfer semantics (who allocates, who frees), buffer length validation, string encoding correctness (UTF-8 ↔ C strings, NUL handling), and that the C header (`include/questdb/ingress/line_sender.h`) and C++ wrapper (`include/questdb/ingress/line_sender.hpp` + the split `line_sender_core.hpp` / `line_sender_array.hpp` / `line_sender_decimal.hpp`) accurately reflect the Rust implementation. If `cbindgen.toml` is involved, verify generated output matches handwritten headers.
 
@@ -250,10 +270,12 @@ Review the diff for:
 - All `unsafe` blocks have documented safety invariants
 - No undefined behavior: dangling pointers, use-after-free, double-free, data races
 - Proper `Send`/`Sync` bounds on public types
-- No panics that can escape FFI boundaries (every `extern "C"` function uses `catch_unwind` or proves panics are impossible)
+- No panics that can escape FFI boundaries — and the meaning of "escape" depends on the panic strategy (see Step 2.5e). Under `panic = "abort"`, `catch_unwind` is a no-op and *every* reachable panic is a fatal escape; the FFI function must prove no panic path exists. Under `panic = "unwind"`, every `extern "C"` function must wrap its body in `catch_unwind`.
 
 ### Crash surface
-Anything that aborts the Rust side aborts the host process. Beyond panics, check for:
+Anything that aborts the Rust side aborts the host process. The first check is the panic strategy itself — everything else is downstream of it.
+
+- **Panic strategy** (from Step 2.5e): under `panic = "abort"`, the entire `catch_unwind` defense collapses — every panic across the entire crate is fatal. Verify the profile before crediting any panic guard. A finding that says "the panic at X is caught by `catch_unwind` at Y" is incorrect under abort-panic.
 - Direct aborts: `std::process::abort()`, `libc::abort()`, `std::intrinsics::abort()`
 - Allocation-failure aborts: any allocation sized by an untrusted length parameter must validate the bound before allocating (Rust's default allocator aborts on OOM)
 - Stack overflow: unbounded recursion, recursive `Drop` impls, deeply nested untrusted input

.gitmodules

@@ -0,0 +1,4 @@
+[submodule "questdb"]
+	path = questdb
+	url = https://github.com/questdb/questdb.git
+	branch = master

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.15.0)
-project(c-questdb-client VERSION 6.1.0)
+project(c-questdb-client VERSION 7.0.0)
 
 set(CPACK_PROJECT_NAME ${PROJECT_NAME})
 set(CPACK_PROJECT_VERSION ${PROJECT_VERSION})
@@ -31,20 +31,74 @@ option(
     "Build shared library dependencies instead of static."
     OFF)
 
+# Opt in to the synchronous WebSocket egress reader (`line_reader_*`
+# C/C++ surface). Pulls in the `tungstenite` + `zstd` Rust crates as
+# transitive dependencies, so downstreams that only need the line
+# sender can flip this OFF to keep the resulting library minimal.
+# Defaults ON for the in-tree build because the `line_reader_*`
+# examples and tests need it; external consumers who add this
+# project via `add_subdirectory` should set
+# `-DQUESTDB_ENABLE_READER=OFF` if they don't want the surface.
+option(
+    QUESTDB_ENABLE_READER
+    "Enable the synchronous WebSocket egress reader (line_reader_* API). Adds tungstenite+zstd."
+    ON)
+
+# When QUESTDB_TESTS_AND_EXAMPLES is enabled the reader-related
+# examples and tests need the reader API present. Refusing to build
+# in that combination is friendlier than producing a confusing link
+# error from missing `line_reader_*` symbols.
+if(QUESTDB_TESTS_AND_EXAMPLES AND NOT QUESTDB_ENABLE_READER)
+    message(FATAL_ERROR
+        "QUESTDB_TESTS_AND_EXAMPLES=ON requires QUESTDB_ENABLE_READER=ON: "
+        "the line_reader_* examples and tests would fail to link without it.")
+endif()
+
+# Compile in the `tls_verify=false` (sender) / `tls_verify=unsafe_off`
+# (egress reader) escape hatch. ON by default to preserve the legacy
+# behaviour of the shipped C ABI; security-conscious distributions can
+# flip it OFF (`-DQUESTDB_ENABLE_INSECURE_SKIP_VERIFY=OFF`) to harden
+# the resulting library — `line_sender_opts_tls_verify` then disappears
+# from the symbol table and `tls_verify=unsafe_off` in a connect string
+# is rejected at parse time.
+option(
+    QUESTDB_ENABLE_INSECURE_SKIP_VERIFY
+    "Compile in support for tls_verify off. Allows downstream code to disable TLS certificate verification at runtime."
+    ON)
+
+option(
+    QUESTDB_SANITIZE
+    "Build the C/C++ tests with -fsanitize=address,undefined."
+    OFF)
+
 # Build static and dynamic lib written in Rust by invoking `cargo`.
 # Imports `questdb_client` target.
 add_subdirectory(corrosion)
-corrosion_import_crate(
-    MANIFEST_PATH questdb-rs-ffi/Cargo.toml
-    LOCKED)   # Use `Cargo.lock`
+set(QUESTDB_CARGO_FEATURES "")
+if(QUESTDB_ENABLE_READER)
+    list(APPEND QUESTDB_CARGO_FEATURES sync-reader-ws)
+endif()
+if(QUESTDB_ENABLE_INSECURE_SKIP_VERIFY)
+    list(APPEND QUESTDB_CARGO_FEATURES insecure-skip-verify)
+endif()
+if(QUESTDB_CARGO_FEATURES)
+    corrosion_import_crate(
+        MANIFEST_PATH questdb-rs-ffi/Cargo.toml
+        LOCKED
+        FEATURES ${QUESTDB_CARGO_FEATURES})
+else()
+    corrosion_import_crate(
+        MANIFEST_PATH questdb-rs-ffi/Cargo.toml
+        LOCKED)
+endif()
 target_include_directories(
     questdb_client INTERFACE
     ${CMAKE_CURRENT_SOURCE_DIR}/include)
 if(WIN32)
     set_target_properties(
         questdb_client-shared
         PROPERTIES
-        DEFINE_SYMBOL "LINESENDER_DYN_LIB")
+        DEFINE_SYMBOL "QUESTDB_CLIENT_DYN_LIB")
     target_link_libraries(
         questdb_client-shared
         INTERFACE wsock32 ws2_32 ntdll crypt32 Secur32 Ncrypt)
@@ -82,6 +136,27 @@ function(set_compile_flags TARGET_NAME)
     endif()
 endfunction()
 
+function(apply_sanitizers TARGET_NAME)
+    if(NOT QUESTDB_SANITIZE)
+        return()
+    endif()
+    if(MSVC)
+        message(WARNING
+            "QUESTDB_SANITIZE is not supported on MSVC (its ASan runtime is "
+            "not validated against the rustc-built library); skipping.")
+    else()
+        target_compile_options(
+            ${TARGET_NAME} PRIVATE
+            -fsanitize=address,undefined
+            -fno-sanitize-recover=all
+            -fno-omit-frame-pointer
+            -g)
+        target_link_options(
+            ${TARGET_NAME} PRIVATE
+            -fsanitize=address,undefined)
+    endif()
+endfunction()
+
 # Examples
 function(compile_example TARGET_NAME)
     list(POP_FRONT ARGV)
@@ -187,6 +262,24 @@ if (QUESTDB_TESTS_AND_EXAMPLES)
     compile_example(
         line_sender_cpp_example_decimal_binary
         examples/line_sender_cpp_example_decimal_binary.cpp)
+    compile_example(
+        line_reader_c_example_from_conf
+        examples/line_reader_c_example_from_conf.c)
+    compile_example(
+        line_reader_cpp_example_from_conf
+        examples/line_reader_cpp_example_from_conf.cpp)
+    compile_example(
+        line_reader_c_example_with_binds
+        examples/line_reader_c_example_with_binds.c)
+    compile_example(
+        line_reader_cpp_example_with_binds
+        examples/line_reader_cpp_example_with_binds.cpp)
+    compile_example(
+        line_reader_cpp_example_columns
+        examples/line_reader_cpp_example_columns.cpp)
+    compile_example(
+        line_reader_c_example_columns
+        examples/line_reader_c_example_columns.c)
 
     # Include Rust tests as part of the tests run
     add_test(
@@ -207,16 +300,64 @@ if (QUESTDB_TESTS_AND_EXAMPLES)
             ${TARGET_NAME}
             PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
         set_compile_flags(${TARGET_NAME})
+        apply_sanitizers(${TARGET_NAME})
         add_test(
             NAME ${TARGET_NAME}
             COMMAND ${TARGET_NAME})
+        if(QUESTDB_SANITIZE AND NOT MSVC)
+            # Leak detection is off: the Rust library is not instrumented and
+            # legitimately retains process-global allocations at exit, which
+            # LSan cannot tell apart from an FFI leak without a suppression
+            # file. ASan's use-after-free / out-of-bounds / double-free checks
+            # stay active.
+            set_tests_properties(
+                ${TARGET_NAME} PROPERTIES
+                ENVIRONMENT
+                "ASAN_OPTIONS=detect_leaks=0;UBSAN_OPTIONS=print_stacktrace=1:halt_on_error=1")
+        endif()
     endfunction()
 
     compile_test(
         test_line_sender
         cpp_test/mock_server.cpp
         cpp_test/test_line_sender.cpp)
 
+    # Live-broker integration tests for the egress reader. Skips per-test
+    # if no broker is reachable on QDB_LIVE_BROKER_HOST:QDB_LIVE_BROKER_HTTP_PORT
+    # (defaults: localhost:9000), so this test is safe to wire into ctest
+    # even on machines without a broker.
+    compile_test(
+        test_line_reader
+        cpp_test/test_line_reader.cpp)
+
+    # Broker-independent smoke test for the line_reader FFI. Targets a
+    # guaranteed-closed port (127.0.0.1:1) and asserts the FFI surfaces
+    # a non-NULL error that can be inspected and freed. Uses standard
+    # exit-code semantics so SIGSEGV / SIGABRT correctly fail the test
+    # (the previous WILL_FAIL-based smoke treated any non-zero exit as a
+    # pass, and inverted to a failure when QuestDB happened to be running
+    # on the developer's machine).
+    compile_test(
+        line_reader_c_smoke
+        cpp_test/smoke_line_reader.c)
+
+    # Broker-independent C++ tests for the line_reader FFI. Covers the
+    # error-handling surface, parser rejection paths, the connect-failure
+    # path against a closed port, NULL-idempotent free / close functions,
+    # `from_env`, and the C++ `line_reader_error` exception wrapper.
+    compile_test(
+        test_line_reader_offline
+        cpp_test/test_line_reader_offline.cpp)
+
+    # Mock-server-driven C++ tests for the line_reader FFI. Drives the
+    # reader against an in-process WebSocket + QWP1 mock so the
+    # column-getter / bind-encoding / server_info / error-code / stats
+    # surface that previously needed a live broker now runs in CI.
+    compile_test(
+        test_line_reader_mock
+        cpp_test/qwp_mock_server.cpp
+        cpp_test/test_line_reader_mock.cpp)
+
     # System testing Python3 script.
     # This will download the latest QuestDB instance from Github,
     # thus will also require a Java 11 installation to run the tests.