diff --git a/docs/m7-crypto-l2-capture.md b/docs/m7-crypto-l2-capture.md index f44c7c0..f0ec417 100644 --- a/docs/m7-crypto-l2-capture.md +++ b/docs/m7-crypto-l2-capture.md @@ -91,6 +91,13 @@ operator action. ## Configuration and CLI +On Windows with long-path support disabled, use extended-length local paths in +the JSON configuration, for example `"hot_root": "\\\\?\\D:\\capture-hot"`. +Apply the same form to the archive and restore roots. Journal and temporary +filenames can exceed the legacy path limit even with a short root. The physical +volume probe maps this spelling to the same drive device and still checks the +underlying disks; UNC shares and non-drive device paths remain unsupported. + Example configuration, with operator-selected absolute paths: ```json diff --git a/src/quant_data_kit/capture_v2/storage.py b/src/quant_data_kit/capture_v2/storage.py index 5a367ec..e6c88be 100644 --- a/src/quant_data_kit/capture_v2/storage.py +++ b/src/quant_data_kit/capture_v2/storage.py @@ -93,7 +93,14 @@ class VolumeDiskExtents(ctypes.Structure): _fields_ = [("count", wintypes.DWORD), ("extents", DiskExtent * 1)] anchor = Path(path).resolve(strict=True).anchor.rstrip("\\/") - if len(anchor) != 2 or anchor[1] != ":": + # Extended-length local paths retain this prefix after resolve(). Only the + # volume device handle needs the ordinary drive form; keep file paths long. + anchor = anchor.removeprefix("\\\\?\\") + if ( + len(anchor) != 2 + or anchor[0] not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + or anchor[1] != ":" + ): raise CapturePausedError(f"cannot resolve Windows volume anchor for {path}") volume_path = rf"\\.\{anchor}" create_file = ctypes.windll.kernel32.CreateFileW diff --git a/src/quant_data_kit/capture_v2/transport.py b/src/quant_data_kit/capture_v2/transport.py index e1c4ef5..57254ce 100644 --- a/src/quant_data_kit/capture_v2/transport.py +++ b/src/quant_data_kit/capture_v2/transport.py @@ -70,7 +70,9 @@ def __init__(self, connection: object) -> None: self._connection = connection async def send(self, payload: bytes) -> None: - await self._connection.send(payload) # type: ignore[attr-defined] + # Subscription JSON is a UTF-8 text message. Sending bytes directly uses + # a binary frame, which OKX rejects as an invalid latency probe. + await self._connection.send(payload.decode("utf-8")) # type: ignore[attr-defined] async def receive(self, *, timeout_seconds: float) -> bytes: try: diff --git a/tests/test_capture_v2_collector_cli_transport.py b/tests/test_capture_v2_collector_cli_transport.py index 3060c7c..5149551 100644 --- a/tests/test_capture_v2_collector_cli_transport.py +++ b/tests/test_capture_v2_collector_cli_transport.py @@ -684,10 +684,10 @@ def test_websocket_transport_wrapper_bytes_text_timeout_and_tls( class Underlying: def __init__(self, values: list[Any]) -> None: self.values = values - self.sent: list[bytes] = [] + self.sent: list[str] = [] self.closed = False - async def send(self, payload: bytes) -> None: + async def send(self, payload: str) -> None: self.sent.append(payload) async def recv(self): @@ -712,7 +712,7 @@ async def close(self): asyncio.run(wrapped.receive(timeout_seconds=0.001)) asyncio.run(wrapped.send(b"subscribe")) asyncio.run(wrapped.close()) - assert underlying.sent == [b"subscribe"] and underlying.closed + assert underlying.sent == ["subscribe"] and underlying.closed connector = WebsocketsConnector() with pytest.raises(ValidationError, match="wss"): diff --git a/tests/test_capture_websocket_text.py b/tests/test_capture_websocket_text.py new file mode 100644 index 0000000..1889292 --- /dev/null +++ b/tests/test_capture_websocket_text.py @@ -0,0 +1,26 @@ +import asyncio + +from websockets.asyncio.client import connect +from websockets.asyncio.server import serve + +from quant_data_kit.capture_v2.transport import _WebsocketsConnection + + +def test_subscription_is_a_text_frame_on_the_wire(): + async def exercise(): + received = asyncio.get_running_loop().create_future() + + async def handle(connection): + received.set_result(await connection.recv()) + + async with serve(handle, "127.0.0.1", 0) as server: + port = server.sockets[0].getsockname()[1] + async with connect(f"ws://127.0.0.1:{port}", proxy=None) as connection: + wrapped = _WebsocketsConnection(connection) + payload = b'{"op":"subscribe","args":[{"channel":"books","instId":"BTC-USDT"}]}' + await wrapped.send(payload) + observed = await asyncio.wait_for(received, timeout=3) + assert isinstance(observed, str) + assert observed.encode("utf-8") == payload + + asyncio.run(exercise()) diff --git a/tests/test_capture_windows_extended_paths.py b/tests/test_capture_windows_extended_paths.py new file mode 100644 index 0000000..3ba099d --- /dev/null +++ b/tests/test_capture_windows_extended_paths.py @@ -0,0 +1,51 @@ +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from quant_data_kit.capture_v2 import storage + + +@pytest.mark.parametrize("anchor", ["D:\\", "\\\\?\\D:\\"]) +def test_local_drive_paths_open_the_same_physical_volume(anchor, monkeypatch): + from ctypes import wintypes + + class ResolvedPath: + def __init__(self, value): + self.anchor = anchor + + def resolve(self, *, strict): + assert strict + return self + + class OpenVolume: + def __call__(self, device, *_args): + assert device == "\\\\.\\D:" + # An OS denial still fails closed after resolving either spelling. + return wintypes.HANDLE(-1).value + + monkeypatch.setattr(storage, "Path", ResolvedPath) + monkeypatch.setattr( + storage.ctypes, + "windll", + SimpleNamespace(kernel32=SimpleNamespace(CreateFileW=OpenVolume())), + raising=False, + ) + with pytest.raises(storage.CapturePausedError, match="cannot open volume"): + storage._windows_volume_identity(Path("unused")) + + +@pytest.mark.parametrize( + "anchor", ["\\\\server\\share\\", "\\\\?\\UNC\\server\\share\\", "\\\\?\\Volume{abc}\\", "1:\\"] +) +def test_unsupported_windows_anchors_do_not_reach_device_probe(anchor, monkeypatch): + class ResolvedPath: + def __init__(self, value): + self.anchor = anchor + + def resolve(self, *, strict): + return self + + monkeypatch.setattr(storage, "Path", ResolvedPath) + with pytest.raises(storage.CapturePausedError, match="cannot resolve Windows volume anchor"): + storage._windows_volume_identity(Path("unused"))