From 3fcfb5984dd477212ba758c61bd9730c1baa3ecf Mon Sep 17 00:00:00 2001 From: Peter Bower <37089506+pbower@users.noreply.github.com> Date: Sat, 19 Sep 2026 23:03:37 +0100 Subject: [PATCH] 1. Add Python CI 2. Improve Windows/Mac compatibility --- .github/workflows/ci.yml | 51 ++++++++++++++++++++++ .github/workflows/release.yml | 80 +++++++++++++++++++++++++++++++--- python/Cargo.toml | 6 ++- python/README.md | 3 ++ python/src/input.rs | 6 +++ python/src/listeners.rs | 12 ++++- python/src/output.rs | 5 +++ python/src/uri.rs | 53 +++++++++++++++------- python/tests/test_accept.py | 7 +++ python/tests/test_files.py | 16 +++++++ python/tests/test_network.py | 12 ++++- python/tests/test_tls_wires.py | 2 +- rust/src/constants.rs | 16 ++++--- 13 files changed, 236 insertions(+), 33 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 20a6a9a..b144e21 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -105,6 +105,57 @@ jobs: - name: cargo test --all-targets run: cargo test --all-targets --features "tcp,uds,stdio,websocket,http,quic,webtransport,mmap,csv,json,parquet,zstd,snappy,protocol,msgpack,protobuf,tls" + test-native-platforms: + name: Native Rust and Python tests (${{ matrix.platform }}) + runs-on: ${{ matrix.runner }} + strategy: + fail-fast: false + matrix: + include: + - platform: linux-x86_64 + runner: ubuntu-latest + unix-features: ",uds,mmap" + - platform: macos-x86_64 + runner: macos-15-intel + unix-features: ",uds,mmap" + - platform: macos-aarch64 + runner: macos-14 + unix-features: ",uds,mmap" + - platform: windows-x86_64 + runner: windows-latest + unix-features: "" + defaults: + run: + shell: bash + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@nightly + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - uses: Swatinem/rust-cache@v2 + with: + workspaces: | + rust + python + - name: Test native Rust library + working-directory: rust + env: + FEATURES: tcp,tls,http,websocket,quic,webtransport,stdio,csv,json,parquet,zstd,snappy,protocol,msgpack,protobuf,datetime,decimal,extended_categorical,extended_numeric_types${{ matrix.unix-features }} + run: cargo test --lib --features "$FEATURES" + - name: Install Python build and test dependencies + run: python -m pip install maturin pytest pyarrow polars duckdb + - name: Build Python wheel + run: python -m maturin build -m python/Cargo.toml --out dist --interpreter python + - name: Install Python wheel + run: python -m pip install dist/*.whl + - name: Test Python API (Unix) + if: runner.os != 'Windows' + run: python -m pytest python/tests/ -v --basetemp=/tmp/ls-py-tests + - name: Test Python API (Windows) + if: runner.os == 'Windows' + run: python -m pytest python/tests/ -v + doc: name: cargo doc runs-on: ubuntu-latest diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 175a269..88f8bd2 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -9,9 +9,9 @@ name: Release # workflow_dispatch runs the same build and audit, then offers a crate dry-run # and a TestPyPI upload for exercising the pipeline without a real release. # -# Wheels cover Linux (x86_64, aarch64) and macOS (x86_64, aarch64). Windows is -# not built: the uds and mmap paths are POSIX-only, so a Windows port is tracked -# separately. +# Wheels cover Linux and macOS (x86_64, aarch64), plus Windows x86_64. +# Each wheel is tested natively before publication. Windows excludes mmap +# and Unix-domain sockets; Linux and macOS retain both. on: push: @@ -64,10 +64,14 @@ jobs: wheels-macos: name: Wheels (macos ${{ matrix.target }}) - runs-on: macos-14 + runs-on: ${{ matrix.runner }} strategy: matrix: - target: [x86_64, aarch64] + include: + - target: x86_64 + runner: macos-15-intel + - target: aarch64 + runner: macos-14 steps: - uses: actions/checkout@v4 - uses: PyO3/maturin-action@v1 @@ -80,6 +84,24 @@ jobs: name: wheels-macos-${{ matrix.target }} path: dist + wheels-windows: + name: Wheels (windows x86_64) + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - uses: PyO3/maturin-action@v1 + with: + target: x86_64 + rust-toolchain: nightly + args: --release --out dist ${{ env.MANIFEST }} + - uses: actions/upload-artifact@v4 + with: + name: wheels-windows-x86_64 + path: dist + sdist: name: Source distribution runs-on: ubuntu-latest @@ -117,6 +139,50 @@ jobs: fi echo "binary audit passed" + # Install every release wheel on its native OS and architecture. + test-python: + name: Python tests (${{ matrix.platform }}) + runs-on: ${{ matrix.runner }} + needs: [wheels-linux, wheels-macos, wheels-windows] + strategy: + fail-fast: false + matrix: + include: + - platform: linux-x86_64 + runner: ubuntu-latest + - platform: linux-aarch64 + runner: ubuntu-24.04-arm + - platform: macos-x86_64 + runner: macos-15-intel + - platform: macos-aarch64 + runner: macos-14 + - platform: windows-x86_64 + runner: windows-latest + defaults: + run: + shell: bash + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - uses: actions/download-artifact@v4 + with: + name: wheels-${{ matrix.platform }} + path: dist + - name: Install wheel and test dependencies + run: | + set -euo pipefail + python -m pip install dist/*.whl + python -m pip install pytest pyarrow polars duckdb + - name: Run Python tests (Unix) + if: runner.os != 'Windows' + # Keep socket paths below macOS's Unix socket path-length limit. + run: python -m pytest python/tests/ -v --basetemp=/tmp/ls-py-tests + - name: Run Python tests (Windows) + if: runner.os == 'Windows' + run: python -m pytest python/tests/ -v + # Tag push -> crates.io, gated by the `crates-io` environment approval. publish-crate: name: Publish crate to crates.io @@ -147,7 +213,7 @@ jobs: name: Publish lightstream-io to PyPI if: startsWith(github.ref, 'refs/tags/v') runs-on: ubuntu-latest - needs: [wheels-linux, wheels-macos, sdist, audit] + needs: [wheels-linux, wheels-macos, wheels-windows, sdist, audit, test-python] environment: name: pypi url: https://pypi.org/project/lightstream-io/ @@ -181,7 +247,7 @@ jobs: name: Publish to TestPyPI if: github.event_name == 'workflow_dispatch' runs-on: ubuntu-latest - needs: [wheels-linux, wheels-macos, sdist, audit] + needs: [wheels-linux, wheels-macos, wheels-windows, sdist, audit, test-python] environment: name: testpypi url: https://test.pypi.org/project/lightstream-io/ diff --git a/python/Cargo.toml b/python/Cargo.toml index 3af7349..0f2e433 100644 --- a/python/Cargo.toml +++ b/python/Cargo.toml @@ -19,7 +19,7 @@ name = "lightstream_py" crate-type = ["cdylib", "rlib"] [dependencies] -lightstream = { version = "0.7", path = "../rust", features = ["csv", "datetime", "decimal", "extended_categorical", "extended_numeric_types", "json", "mmap", "http", "parquet", "protocol", "quic", "snappy", "stdio", "tcp", "tls", "uds", "webtransport", "websocket", "zstd"] } +lightstream = { version = "0.7", path = "../rust", features = ["csv", "datetime", "decimal", "extended_categorical", "extended_numeric_types", "json", "http", "parquet", "protocol", "quic", "snappy", "stdio", "tcp", "tls", "webtransport", "websocket", "zstd"] } # The categorical and numeric feature set mirrors the minarrow-py build. # minarrow-pyo3's dictionary-index conversion needs the extended features, # and they flow through lightstream's flags so its match arms gate in step @@ -38,6 +38,10 @@ tokio = { version = "1", features = ["fs", "io-std", "net", "rt-multi-thread", " tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } wtransport = "0.7" +# Unix-only transports stay out of Windows wheels. +[target.'cfg(unix)'.dependencies] +lightstream = { version = "0.7", path = "../rust", features = ["mmap", "uds"] } + [features] default = [] extension-module = ["pyo3/extension-module"] diff --git a/python/README.md b/python/README.md index 6ccd3c4..e5b2f05 100644 --- a/python/README.md +++ b/python/README.md @@ -10,6 +10,9 @@ Move Arrow tables between processes, services and storage from Python without ad pip install lightstream-io ``` +Unix-domain sockets and memory-mapped reads are available on Linux and macOS. +Windows uses buffered file reads; `uds://` and explicit `mmap=True` requests are unsupported. + ## Usage Everything is `read` or `write`. diff --git a/python/src/input.rs b/python/src/input.rs index ecc68e0..781fd61 100644 --- a/python/src/input.rs +++ b/python/src/input.rs @@ -27,6 +27,7 @@ use lightstream::models::readers::chunked::csv::ChunkedCsvReader; use lightstream::models::readers::chunked::parquet::ChunkedParquetReader; use lightstream::models::readers::csv::CsvReader; use lightstream::models::readers::ipc::file_table::FileTableReader; +#[cfg(unix)] use lightstream::models::readers::ipc::mmap_table::MmapTableReader; use lightstream::models::readers::http::HttpTableReader; use lightstream::models::readers::parallel::tcp::TcpParallelTableReader; @@ -35,6 +36,7 @@ use lightstream::models::readers::parquet::load_parquet_table; use lightstream::models::readers::quic::QuicTableReader; use lightstream::models::readers::stdio::StdinTableReader; use lightstream::models::readers::tcp::TcpTableReader; +#[cfg(unix)] use lightstream::models::readers::uds::UdsTableReader; use lightstream::models::readers::websocket::WebSocketTableReader; use lightstream::models::readers::webtransport::WebTransportTableReader; @@ -79,6 +81,7 @@ pub enum FileIO { reader: FileTableReader, cursor: usize, }, + #[cfg(unix)] IpcMmap { reader: MmapTableReader, cursor: usize, @@ -118,6 +121,7 @@ pub enum ArrowIO { TcpParallel(TcpParallelTableReader), Ws(WebSocketTableReader), Http(HttpTableReader), + #[cfg(unix)] Uds(UdsTableReader), Quic(QuicTableReader), Wt(WebTransportTableReader), @@ -139,6 +143,7 @@ impl ArrowIO { } ArrowIO::Ws(reader) => runtime().block_on(reader.read_next()), ArrowIO::Http(reader) => runtime().block_on(reader.read_next()), + #[cfg(unix)] ArrowIO::Uds(reader) => runtime().block_on(reader.read_next()), ArrowIO::Quic(reader) => runtime().block_on(reader.read_next()), ArrowIO::Wt(reader) => runtime().block_on(reader.read_next()), @@ -227,6 +232,7 @@ impl FileIO { *cursor += 1; Ok(Some(table)) } + #[cfg(unix)] FileIO::IpcMmap { reader, cursor } => { if *cursor >= reader.num_batches() { return Ok(None); diff --git a/python/src/listeners.rs b/python/src/listeners.rs index d85a678..6e76a0b 100644 --- a/python/src/listeners.rs +++ b/python/src/listeners.rs @@ -18,15 +18,20 @@ //! and later accepting calls on the endpoint reuse that identity. use std::collections::HashMap; +#[cfg(unix)] use std::fs; use std::io; use std::net; use std::net::ToSocketAddrs; -use std::path::{Path, PathBuf}; +use std::path::Path; +#[cfg(unix)] +use std::path::PathBuf; use std::sync::{Arc, Mutex, OnceLock}; use pyo3::PyResult; -use tokio::net::{TcpListener, UnixListener}; +use tokio::net::TcpListener; +#[cfg(unix)] +use tokio::net::UnixListener; use wtransport::endpoint::endpoint_side::Server; use crate::errors::{TransportError, to_py_err}; @@ -43,6 +48,7 @@ enum ListenerKey { Wss(String), Http(String), Https(String), + #[cfg(unix)] Uds(PathBuf), Quic(String), Wt(String), @@ -50,6 +56,7 @@ enum ListenerKey { enum BoundListener { Tcp(Arc), + #[cfg(unix)] Uds(Arc), Quic(Arc), Wt(Arc>), @@ -100,6 +107,7 @@ pub fn https(url: &str) -> PyResult> { /// it on the first call. A stale socket file left by an earlier /// process is removed before the bind. Call within the runtime /// context. +#[cfg(unix)] pub fn uds(path: &Path) -> PyResult> { let mut map = registry().lock().expect("listener registry lock"); if let Some(BoundListener::Uds(listener)) = map.get(&ListenerKey::Uds(path.to_path_buf())) { diff --git a/python/src/output.rs b/python/src/output.rs index ce9a7b7..a8c0a0b 100644 --- a/python/src/output.rs +++ b/python/src/output.rs @@ -32,6 +32,7 @@ use lightstream::models::writers::parquet::write_parquet_table; use lightstream::models::writers::http::HttpTableWriter; use lightstream::models::writers::stdio::StdoutTableWriter; use lightstream::models::writers::tcp::TcpTableWriter; +#[cfg(unix)] use lightstream::models::writers::uds::UdsTableWriter; use lightstream::models::writers::websocket::WebSocketTableWriter; use lightstream::traits::parallel_transport_writer::ParallelTransportWriter; @@ -48,6 +49,7 @@ use pyo3::exceptions::PyValueError; use tokio::io::{AsyncWrite, ReadHalf, WriteHalf}; use tokio::net::TcpStream; use tokio::net::tcp::OwnedWriteHalf as TcpOwnedWriteHalf; +#[cfg(unix)] use tokio::net::unix::OwnedWriteHalf as UdsOwnedWriteHalf; use tokio_rustls::server::TlsStream as ServerTlsStream; @@ -188,6 +190,7 @@ pub enum ArrowIO { compression: Option, writer: Option, }, + #[cfg(unix)] Uds { link: Link, compression: Option, @@ -402,6 +405,7 @@ impl ArrowIO { .write_table(table.clone()) .await } + #[cfg(unix)] ArrowIO::Uds { link, compression, @@ -533,6 +537,7 @@ impl ArrowIO { Some(writer) => writer.finish().await, None => Ok(()), }, + #[cfg(unix)] ArrowIO::Uds { writer, .. } => match writer.as_mut() { Some(writer) => writer.finish().await, None => Ok(()), diff --git a/python/src/uri.rs b/python/src/uri.rs index 3b88d53..5c61192 100644 --- a/python/src/uri.rs +++ b/python/src/uri.rs @@ -27,6 +27,7 @@ use lightstream::models::readers::chunked::csv::{ChunkedCsvReadOptions, ChunkedC use lightstream::models::readers::chunked::parquet::ChunkedParquetReader; use lightstream::models::readers::csv::CsvReader; use lightstream::models::readers::ipc::file_table::FileTableReader; +#[cfg(unix)] use lightstream::models::readers::ipc::mmap_table::MmapTableReader; use lightstream::models::readers::http::HttpTableReader; use lightstream::models::readers::parallel::tcp::TcpParallelTableReader; @@ -34,6 +35,7 @@ use lightstream::models::readers::json::JsonReader; use lightstream::models::readers::lightstream::LightstreamReader; use lightstream::models::readers::stdio::StdinTableReader; use lightstream::models::readers::tcp::TcpTableReader; +#[cfg(unix)] use lightstream::models::readers::uds::UdsTableReader; use lightstream::models::readers::quic::QuicTableReader; use lightstream::enums::IPCMessageProtocol; @@ -43,6 +45,7 @@ use lightstream::models::streams::websocket::{WsRead, WsWrite}; use lightstream::models::transports::http::HttpTransport; use lightstream::models::transports::quic::QuicTransport; use lightstream::models::transports::tcp::TcpTransport; +#[cfg(unix)] use lightstream::models::transports::uds::UdsTransport; use lightstream::models::transports::webtransport::WebTransport; use lightstream::models::transports::websocket::WebSocketTransport; @@ -63,6 +66,7 @@ use crate::{input, output}; /// Files at or above this size default to the mmap IPC reader. Smaller /// files default to the buffered reader, where the mapping overhead /// outweighs the zero-copy gain. +#[cfg(unix)] const MMAP_DEFAULT_THRESHOLD: u64 = 64 * 1024 * 1024; /// Default row count per decoded batch for the CSV and JSON readers. @@ -134,6 +138,7 @@ enum Endpoint { Wss(String), Http(String), Https(String), + #[cfg(unix)] Uds(PathBuf), Quic(String), Wt(String), @@ -149,7 +154,12 @@ fn resolve_endpoint(uri: &str) -> PyResult { "tcp" => Ok(Endpoint::Tcp(rest.to_string())), "ws" => Ok(Endpoint::Ws(uri.to_string())), "http" => Ok(Endpoint::Http(uri.to_string())), + #[cfg(unix)] "uds" => Ok(Endpoint::Uds(PathBuf::from(rest))), + #[cfg(not(unix))] + "uds" => Err(TransportError::new_err( + "Unix-domain sockets are not supported on this platform", + )), "quic" => Ok(Endpoint::Quic(rest.to_string())), "wt" => Ok(Endpoint::Wt(rest.to_string())), "wss" => Ok(Endpoint::Wss(uri.to_string())), @@ -524,6 +534,7 @@ pub fn resolve_source( .map_err(|e| to_py_err(IoError::Io(e)))? } } + #[cfg(unix)] Endpoint::Uds(path) => { if accept { let listener = { @@ -784,6 +795,7 @@ pub fn resolve_source( .map(input::ArrowIO::Tcp) .map_err(|e| to_py_err(IoError::Io(e)))? } + #[cfg(unix)] Endpoint::Uds(path) => { if accept { let listener = { @@ -968,6 +980,12 @@ pub fn resolve_source( "out_of_core reads through the buffered reader, so it cannot combine with mmap=True", )); } + #[cfg(not(unix))] + if mmap == Some(true) { + return Err(FormatError::new_err( + "mmap is not supported on this platform; use buffered file reads", + )); + } let delimiter_byte = resolve_delimiter(delimiter)?; if path.is_dir() { @@ -1015,22 +1033,25 @@ pub fn resolve_source( reject_inapplicable(resolved_format, delimiter, header, batch_size)?; let file_io = match resolved_format { Format::Ipc => { - let size = fs::metadata(&path) - .map_err(|e| FormatError::new_err(e.to_string()))? - .len(); - // `out_of_core` routes to the buffered reader, which keeps its pages - // reclaimable for datasets larger than RAM. - // TODO: select the mmap reader here once it regains out-of-core streaming. - let use_mmap = mmap.unwrap_or(size >= MMAP_DEFAULT_THRESHOLD) && !out_of_core; - if use_mmap { - let reader = MmapTableReader::open(&path) - .map_err(|e| FormatError::new_err(e.to_string()))?; - input::FileIO::IpcMmap { reader, cursor: 0 } - } else { - let reader = FileTableReader::open(&path) - .map_err(|e| FormatError::new_err(e.to_string()))?; - input::FileIO::Ipc { reader, cursor: 0 } + #[cfg(unix)] + { + let size = fs::metadata(&path) + .map_err(|e| FormatError::new_err(e.to_string()))? + .len(); + // Out-of-core reads use the buffered reader to keep pages reclaimable. + let use_mmap = mmap.unwrap_or(size >= MMAP_DEFAULT_THRESHOLD) && !out_of_core; + if use_mmap { + let reader = MmapTableReader::open(&path) + .map_err(|e| FormatError::new_err(e.to_string()))?; + return Ok(input::Source::File(input::FileIO::IpcMmap { + reader, + cursor: 0, + })); + } } + let reader = FileTableReader::open(&path) + .map_err(|e| FormatError::new_err(e.to_string()))?; + input::FileIO::Ipc { reader, cursor: 0 } } Format::Parquet => input::FileIO::Parquet { path, done: false }, Format::Csv => { @@ -1231,6 +1252,7 @@ pub fn resolve_target( .map_err(|e| to_py_err(IoError::Io(e)))? } } + #[cfg(unix)] Endpoint::Uds(path) => { if accept { let listener = { @@ -1506,6 +1528,7 @@ pub fn resolve_target( writer: None, } } + #[cfg(unix)] Endpoint::Uds(path) => { let link = if accept { let listener = { diff --git a/python/tests/test_accept.py b/python/tests/test_accept.py index 7b5856c..62f505b 100644 --- a/python/tests/test_accept.py +++ b/python/tests/test_accept.py @@ -13,6 +13,8 @@ steals the accept, because a refused connection consumes nothing. """ +import sys + import socket import threading import time @@ -102,10 +104,12 @@ def test_tcp_arrow_accepting_reader(): run_accepting_reader(f"tcp://127.0.0.1:{free_port()}") +@pytest.mark.skipif(sys.platform == "win32", reason="Unix-domain sockets are Unix-only") def test_uds_arrow_accepting_writer(tmp_path): run_accepting_writer(f"uds://{tmp_path / 'writer.sock'}") +@pytest.mark.skipif(sys.platform == "win32", reason="Unix-domain sockets are Unix-only") def test_uds_arrow_accepting_reader(tmp_path): run_accepting_reader(f"uds://{tmp_path / 'reader.sock'}") @@ -153,6 +157,7 @@ def serve(): assert frames[1].payload == b"\x07" +@pytest.mark.skipif(sys.platform == "win32", reason="Unix-domain sockets are Unix-only") def test_uds_lightstream_accepting_writer(tmp_path): uri = f"uds://{tmp_path / 'tlv.sock'}" @@ -226,6 +231,7 @@ def serve(): assert frames[1].payload == b"\x09" +@pytest.mark.skipif(sys.platform == "win32", reason="Unix-domain sockets are Unix-only") def test_accepting_writer_serves_requests_back_to_back(tmp_path): uri = f"uds://{tmp_path / 'serve.sock'}" @@ -249,6 +255,7 @@ def serve(): assert not thread.is_alive() +@pytest.mark.skipif(sys.platform == "win32", reason="Unix-domain sockets are Unix-only") def test_clients_queue_while_the_listener_is_busy(tmp_path): uri = f"uds://{tmp_path / 'queue.sock'}" diff --git a/python/tests/test_files.py b/python/tests/test_files.py index 575bfd0..229977d 100644 --- a/python/tests/test_files.py +++ b/python/tests/test_files.py @@ -10,6 +10,8 @@ Table / ChunkedTable output split, and the error surface. """ +import sys + import gc from decimal import Decimal @@ -77,6 +79,7 @@ def test_read_all_after_drain_returns_empty_table(tmp_path): assert result.n_rows == 0 +@pytest.mark.skipif(sys.platform == "win32", reason="mmap is Unix-only") def test_mmap_route_matches_buffered_route(tmp_path): path = str(tmp_path / "quotes.arrow") original = sample_table() @@ -427,3 +430,16 @@ def test_batch_size_on_ipc_raises(tmp_path): def test_multichar_delimiter_raises(tmp_path): with pytest.raises(ValueError, match="single ASCII character"): ls.read(str(tmp_path / "quotes.csv"), delimiter="::") + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows platform behavior") +def test_windows_mmap_rejected_and_buffered_reads_work(tmp_path): + path = str(tmp_path / "quotes.arrow") + original = sample_table() + with ls.write(path) as writer: + writer.write(original) + for options in ({}, {"mmap": False}, {"out_of_core": True}): + with ls.read(path, **options) as reader: + assert pa.table(reader.read_all()).equals(original) + with pytest.raises(ls.FormatError, match="mmap is not supported on this platform"): + ls.read(path, mmap=True) diff --git a/python/tests/test_network.py b/python/tests/test_network.py index a73f474..35e7acf 100644 --- a/python/tests/test_network.py +++ b/python/tests/test_network.py @@ -64,6 +64,8 @@ def tcp_relay(): @pytest.fixture def uds_relay(tmp_path): + if sys.platform == "win32": + pytest.skip("Unix-domain sockets are Unix-only") sock_path = str(tmp_path / "relay.sock") listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) listener.bind(sock_path) @@ -156,6 +158,7 @@ def test_stdio_pipeline(): assert reader_proc.stdout.strip() == "5" +@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX sed") def test_stdio_csv_pipeline_through_sed(): producer_code = ( "import lightstream as ls, pyarrow as pa\n" @@ -173,7 +176,7 @@ def test_stdio_csv_pipeline_through_sed(): [sys.executable, "-c", producer_code], stdout=subprocess.PIPE ) sed = subprocess.Popen( - ["sed", "-u", "s/a/A/"], stdin=producer.stdout, stdout=subprocess.PIPE + ["sed", "s/a/A/"], stdin=producer.stdout, stdout=subprocess.PIPE ) consumer = subprocess.run( [sys.executable, "-c", consumer_code], @@ -265,3 +268,10 @@ def test_lightstream_over_ws_and_http_reaches_connection(): ls.read(uri, protocol="lightstream") with pytest.raises(ls.LightstreamError): ls.write(uri, protocol="lightstream") + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows platform behavior") +@pytest.mark.parametrize("opener", [ls.read, ls.write]) +def test_windows_uds_rejected(opener): + with pytest.raises(ls.TransportError, match="Unix-domain sockets are not supported"): + opener("uds:///unsupported.sock") diff --git a/python/tests/test_tls_wires.py b/python/tests/test_tls_wires.py index 7874bdd..9c9c311 100644 --- a/python/tests/test_tls_wires.py +++ b/python/tests/test_tls_wires.py @@ -292,4 +292,4 @@ def test_tls_arguments_rejected_on_plain_wires(tls_pair): with pytest.raises(ls.TransportError, match="apply to the quic, wt, wss, and https"): ls.read("tcp://127.0.0.1:9", tls_ca=cert) with pytest.raises(ls.TransportError, match="apply to the quic, wt, wss, and https"): - ls.write("uds:///tmp/none.sock", accept=True, tls_cert=cert, tls_key=key) + ls.write("tcp://127.0.0.1:9", accept=True, tls_cert=cert, tls_key=key) diff --git a/rust/src/constants.rs b/rust/src/constants.rs index ba2dcc2..e61f517 100644 --- a/rust/src/constants.rs +++ b/rust/src/constants.rs @@ -87,12 +87,16 @@ pub fn inmemory_chunk_size() -> usize { /// Default stream arena capacity. /// -/// 2 GiB of virtual address space per arena. -/// With Vec64/MAllocPg64 backing, physical memory is committed -/// only as bytes are written, so the reservation is cheap under normal -/// Linux overcommit. Each stream decoder and, under the `arena` feature, -/// each file reader holds one arena. -pub const DEFAULT_ARENA_CAPACITY: usize = 2 * 1024 * 1024 * 1024; +/// Linux uses 2 GiB per arena to take advantage of normal overcommit. +/// Other platforms use 64 MiB: a large allocation can consume commit +/// budget even before its pages are touched, particularly on Windows. +/// Each stream decoder and, under the `arena` feature, each file reader +/// holds one arena. Larger frames grow a dedicated generation on demand. +pub const DEFAULT_ARENA_CAPACITY: usize = if cfg!(target_os = "linux") { + 2 * 1024 * 1024 * 1024 +} else { + 64 * 1024 * 1024 +}; /// Stream arena capacity in bytes. Override with /// `LIGHTSTREAM_ARENA_CAPACITY` on hosts where per-stream virtual