diff --git a/.github/embedded-source.json b/.github/embedded-source.json new file mode 100644 index 00000000..41755189 --- /dev/null +++ b/.github/embedded-source.json @@ -0,0 +1,8 @@ +{ + "seekdb_repository": "https://github.com/cms-cms/seekdb.git", + "seekdb_sha": "b255464c3df78d1b8f769bcb71448e5737940fcc", + "bindings_sha": "5e8eece63205dccfa9500f98893c91d1f50eb4f1", + "fix_commit": "48b80f5193bcac0afafba4d4820aebdf1103bc69", + "platform": "linux-x86_64", + "python": "3.11" +} diff --git a/.github/scripts/build-embedded-wheel.sh b/.github/scripts/build-embedded-wheel.sh new file mode 100755 index 00000000..7f469273 --- /dev/null +++ b/.github/scripts/build-embedded-wheel.sh @@ -0,0 +1,83 @@ +#!/usr/bin/env bash +# CI-local Linux wheel: not a portable manylinux release or a PyPI publication. +set -Eeuo pipefail +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +: "${RUNNER_TEMP:?use an isolated GitHub runner temporary directory}" +: "${EMBEDDED_WHEEL_DIR:?wheel output/cache directory is required}" +[[ "$(uname -s)/$(uname -m)" == Linux/x86_64 ]] +PYTHON="$(command -v python3)" +"$PYTHON" -c 'import sys; assert sys.version_info[:2] == (3, 11)' +read -r SEEKDB_REPOSITORY SEEKDB_SHA BINDINGS_SHA FIX_COMMIT < <( + "$PYTHON" -c 'import json, sys; p=json.load(open(sys.argv[1])); print(p["seekdb_repository"], p["seekdb_sha"], p["bindings_sha"], p["fix_commit"])' \ + "$REPO_ROOT/.github/embedded-source.json" +) +[[ "$SEEKDB_REPOSITORY" == "https://github.com/cms-cms/seekdb.git" ]] +[[ "$SEEKDB_SHA" =~ ^[0-9a-f]{40}$ && "$BINDINGS_SHA" =~ ^[0-9a-f]{40}$ && "$FIX_COMMIT" =~ ^[0-9a-f]{40}$ ]] +BUILD_ROOT="$(mktemp -d "$RUNNER_TEMP/embedded-source.XXXXXX")" +mkdir -p "$EMBEDDED_WHEEL_DIR" +[[ -z "$(ls -A "$EMBEDDED_WHEEL_DIR")" ]] || { echo 'refusing to overwrite wheel evidence' >&2; exit 1; } + +checkout_exact() { + local repository="$1" revision="$2" target="$3" + git init -q "$target" + git -C "$target" remote add origin "$repository" + git -C "$target" config remote.origin.promisor true + git -C "$target" config remote.origin.partialclonefilter blob:none + # Full ancestry is needed to prove the kernel contains the required fix. + # Historical source blobs are not: fetch them lazily for the exact checkout. + git -C "$target" fetch --filter=blob:none --no-tags origin "$revision" + git -C "$target" checkout --detach FETCH_HEAD + [[ "$(git -C "$target" rev-parse HEAD)" == "$revision" ]] +} +checkout_exact "$SEEKDB_REPOSITORY" "$SEEKDB_SHA" "$BUILD_ROOT/seekdb" +git -C "$BUILD_ROOT/seekdb" merge-base --is-ancestor "$FIX_COMMIT" HEAD +checkout_exact https://github.com/oceanbase/seekdb-bindings.git "$BINDINGS_SHA" "$BUILD_ROOT/bindings" +git -C "$BUILD_ROOT/bindings" submodule update --init --depth=1 deps/mariadb-connector-c + +# Match the upstream seekdb-bindings Ubuntu source-build recipe. These are +# packaging/client tools, not dependencies of the seekdb compilation target. +DEP_PROFILE="$BUILD_ROOT/seekdb/deps/init/oceanbase.el9.x86_64.deps" +sed -i -E '/^[^#]*(target=(obshell|obdeploy)|(lib)?obclient-)/d' "$DEP_PROFILE" + +# Install the source's exact Rust toolchain once, before parallel compilation; +# never let two Rust targets race while auto-installing the same components. +RUST_TOOLCHAIN="$("$PYTHON" -c 'import sys,tomllib; print(tomllib.load(open(sys.argv[1],"rb"))["toolchain"]["channel"])' \ + "$BUILD_ROOT/seekdb/rust/rust-toolchain.toml")" +[[ "$RUST_TOOLCHAIN" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]] +export RUSTUP_HOME="$BUILD_ROOT/rustup" CARGO_HOME="$BUILD_ROOT/cargo" +rustup toolchain install "$RUST_TOOLCHAIN" --profile minimal --no-self-update +export RUSTUP_TOOLCHAIN="$RUST_TOOLCHAIN" RUSTUP_AUTO_INSTALL=0 +rustup run "$RUST_TOOLCHAIN" cargo --version +BUILD_JOBS="$(nproc)" +(( BUILD_JOBS <= 3 )) || BUILD_JOBS=3 +MEMORY_JOBS="$(awk '/MemTotal:/ {n=int(($2/1024-3072)/4096); print (n>0?n:1)}' /proc/meminfo)" +(( BUILD_JOBS <= MEMORY_JOBS )) || BUILD_JOBS="$MEMORY_JOBS" +export CMAKE_BUILD_PARALLEL_LEVEL="$BUILD_JOBS" CARGO_BUILD_JOBS="$BUILD_JOBS" +( + cd "$BUILD_ROOT/seekdb" + ./build.sh release --init -DBUILD_EMBED_MODE=ON --make -j"$BUILD_JOBS" +) +mapfile -t binaries < <(find "$BUILD_ROOT/seekdb/build_release" -type f -name seekdb -executable) +[[ "${#binaries[@]}" == 1 ]] +SEEKDB_BIN="${binaries[0]}" +strip --strip-debug --strip-unneeded "$SEEKDB_BIN" +"$SEEKDB_BIN" -V 2>&1 | tee "$EMBEDDED_WHEEL_DIR/seekdb-version.txt" +grep -F "$SEEKDB_SHA" "$EMBEDDED_WHEEL_DIR/seekdb-version.txt" + +# Do not call the upstream wheel target (it selects an unqualified `python`). +# Build only the driver, then build a CPython 3.11 wheel with this exact venv. +"$PYTHON" -m venv "$BUILD_ROOT/venv" +BUILD_PYTHON="$BUILD_ROOT/venv/bin/python" +"$BUILD_PYTHON" -m pip install --disable-pip-version-check \ + 'pip==25.3' 'build==1.3.0' 'cmake==3.31.6' 'nanobind==3.0.0' 'scikit-build-core==0.11.6' 'ninja==1.11.1.4' +export PATH="$BUILD_ROOT/venv/bin:$PATH" +cmake -S "$BUILD_ROOT/bindings" -B "$BUILD_ROOT/bindings/build" \ + -DCMAKE_BUILD_TYPE=Release -DBUILD_TESTING=OFF -DSEEKDB_BUILD_PYTHON=OFF \ + -DSEEKDB_BIN="$SEEKDB_BIN" +cmake --build "$BUILD_ROOT/bindings/build" --target seekdb --parallel "$BUILD_JOBS" +"$BUILD_PYTHON" -m pip wheel --no-deps --no-build-isolation \ + --config-settings="cmake.define.Python_EXECUTABLE=$BUILD_PYTHON" \ + --config-settings="cmake.define.Python_FIND_VIRTUALENV=ONLY" \ + --wheel-dir "$EMBEDDED_WHEEL_DIR" "$BUILD_ROOT/bindings/python" +"$PYTHON" "$REPO_ROOT/.github/scripts/embedded_wheel_manifest.py" create \ + "$REPO_ROOT/.github/embedded-source.json" "$EMBEDDED_WHEEL_DIR" "$SEEKDB_BIN" diff --git a/.github/scripts/embedded_wheel_manifest.py b/.github/scripts/embedded_wheel_manifest.py new file mode 100644 index 00000000..07c1c38b --- /dev/null +++ b/.github/scripts/embedded_wheel_manifest.py @@ -0,0 +1,75 @@ +"""Bind a CI-local wheel to pinned sources and verify the installed payload.""" + +import hashlib +import importlib.metadata +import json +import platform +import subprocess +import sys +import zipfile +from pathlib import Path + + +def digest(path): + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def wheel_payload(wheel): + with zipfile.ZipFile(wheel) as archive, archive.open("pylibseekdb/seekdb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def verify(pin, directory): + manifest = json.loads((directory / "manifest.json").read_text()) + if manifest["source"] != pin: + raise ValueError("cached wheel source identity mismatch") + if platform.system() != "Linux" or platform.machine() != "x86_64" or sys.version_info[:2] != (3, 11): + raise ValueError("wheel requires Linux x86_64 CPython 3.11") + wheels = list(directory.glob("*.whl")) + if len(wheels) != 1 or wheels[0].name != manifest["wheel"]: + raise ValueError("expected exactly the manifest wheel") + if digest(wheels[0]) != manifest["wheel_sha256"]: + raise ValueError("wheel checksum mismatch") + if wheel_payload(wheels[0]) != manifest["binary_sha256"]: + raise ValueError("wheel contains a different SeekDB binary") + version = (directory / "seekdb-version.txt").read_text() + if pin["seekdb_sha"] not in version or version != manifest["revision_output"]: + raise ValueError("SeekDB full REVISION mismatch") + return manifest + + +def main(): + mode, pin_path, directory_name, *rest = sys.argv[1:] + pin = json.loads(Path(pin_path).read_text()) + directory = Path(directory_name) + if mode == "create": + (binary_name,) = rest + (wheel,) = directory.glob("*.whl") + manifest = { + "source": pin, + "wheel": wheel.name, + "wheel_sha256": digest(wheel), + "binary_sha256": digest(Path(binary_name)), + "revision_output": (directory / "seekdb-version.txt").read_text(), + } + (directory / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") + elif mode not in {"verify", "installed"}: + raise ValueError(f"unknown verification mode: {mode}") + manifest = verify(pin, directory) + if mode == "installed": + binary = Path(importlib.metadata.distribution("pylibseekdb").locate_file("pylibseekdb/seekdb")) + if digest(binary) != manifest["binary_sha256"]: + raise ValueError("installed SeekDB differs from source-built wheel (dependency resync?)") + # The executable is the verified local wheel payload, never shell input. + version = subprocess.check_output( # noqa: S603 + [str(binary), "-V"], text=True, stderr=subprocess.STDOUT + ) + if pin["seekdb_sha"] not in version: + raise ValueError("installed SeekDB does not report pinned full REVISION") + print(f"[EMBEDDED_WHEEL] installed={binary}") + print(json.dumps(manifest, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7fc585d3..54799c26 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -100,7 +100,11 @@ jobs: bash .github/scripts/check-pytest-summary.sh pytest.log integration-test: - runs-on: ubuntu-latest + # Build and test the CI-local embedded wheel on the same native image. + runs-on: ${{ matrix.test_mode == 'embedded' && 'ubuntu-22.04' || 'ubuntu-latest' }} + timeout-minutes: 180 + env: + EMBEDDED_WHEEL_DIR: ${{ github.workspace }}/.embedded-wheel strategy: fail-fast: false # Allow other jobs to continue if one fails matrix: @@ -137,6 +141,53 @@ jobs: with: python-version: '3.11' + - name: Restore exact embedded source wheel + if: matrix.test_mode == 'embedded' + id: embedded-wheel + uses: actions/cache/restore@v4 + with: + path: .embedded-wheel + # Never reuse a wheel built from another source or build recipe. + key: embedded-source-ubuntu22-x86_64-cp311-${{ hashFiles('.github/embedded-source.json', '.github/scripts/build-embedded-wheel.sh', '.github/scripts/embedded_wheel_manifest.py') }} + + - name: Install embedded source build dependencies + if: matrix.test_mode == 'embedded' && steps.embedded-wheel.outputs.cache-hit != 'true' + run: | + sudo apt-get update + sudo apt-get install -y git wget rpm rpm2cpio cpio make build-essential binutils m4 libtool-bin libncurses5 libaio1 libssl-dev zlib1g-dev + + - name: Build embedded wheel from pinned fixed sources + if: matrix.test_mode == 'embedded' && steps.embedded-wheel.outputs.cache-hit != 'true' + run: bash .github/scripts/build-embedded-wheel.sh + + - name: Verify and install the exact embedded source wheel + if: matrix.test_mode == 'embedded' + run: | + set -euo pipefail + sudo apt-get update + sudo apt-get install -y libaio1 + python .github/scripts/embedded_wheel_manifest.py verify .github/embedded-source.json "$EMBEDDED_WHEEL_DIR" + uv pip install --python .venv/bin/python --no-deps --reinstall "$EMBEDDED_WHEEL_DIR"/*.whl + .venv/bin/python .github/scripts/embedded_wheel_manifest.py installed .github/embedded-source.json "$EMBEDDED_WHEEL_DIR" + + - name: Save verified embedded source wheel + if: matrix.test_mode == 'embedded' && steps.embedded-wheel.outputs.cache-hit != 'true' + uses: actions/cache/save@v4 + with: + path: .embedded-wheel + key: ${{ steps.embedded-wheel.outputs.cache-primary-key }} + + - name: Record embedded wheel provenance + if: matrix.test_mode == 'embedded' + uses: actions/upload-artifact@v4 + with: + name: embedded-wheel-provenance + include-hidden-files: true + path: | + .embedded-wheel/manifest.json + .embedded-wheel/seekdb-version.txt + if-no-files-found: error + - name: Start OceanBase container if: matrix.test_mode == 'oceanbase' uses: oceanbase/setup-oceanbase-ce@v1 @@ -171,10 +222,23 @@ jobs: log_level=INFO fi set -o pipefail - uv run pytest tests/integration_tests/ -v --log-cli-level=${log_level} \ + # Keep the verified source wheel; a normal uv sync would reinstall + # the released pylibseekdb from uv.lock. + uv run --no-sync pytest tests/integration_tests/ -v --log-cli-level=${log_level} \ -k "${{ matrix.test_mode }}" | tee pytest.log bash .github/scripts/check-pytest-summary.sh pytest.log + - name: Upload embedded database diagnostics + if: always() && matrix.test_mode == 'embedded' + uses: actions/upload-artifact@v4 + with: + name: embedded-database-diagnostics + include-hidden-files: true + path: | + .seekdb-test-data/ + pytest.log + if-no-files-found: warn + - name: Run examples for ${{ matrix.test_mode }} env: MODE: ${{ matrix.test_mode }} diff --git a/src/pyseekdb/client/__init__.py b/src/pyseekdb/client/__init__.py index 03827b28..a242e869 100644 --- a/src/pyseekdb/client/__init__.py +++ b/src/pyseekdb/client/__init__.py @@ -12,7 +12,7 @@ - Returns: _AdminClientProxy (database operations only) All factories use the underlying ServerAPI implementations: -- SeekdbEmbeddedClient - Local seekdb (requires pylibseekdb, Linux only) +- SeekdbEmbeddedClient - Local seekdb (requires seekdb or legacy pylibseekdb) - RemoteServerClient - Remote server via pymysql (supports both seekdb Server and OceanBase Server) """ @@ -136,7 +136,7 @@ def _create_server_client( return SeekdbEmbeddedClient(path=default_path, database=database, **kwargs) raise ValueError( - "Default embedded mode is not available because pylibseekdb could not be imported. " + "Default embedded mode is not available because a SeekDB Python binding could not be imported. " "Please provide host/port parameters to use RemoteServerClient." ) @@ -145,7 +145,7 @@ def __getattr__(name: str) -> Any: """ Lazily expose optional embedded client symbols. - This avoids importing pylibseekdb during `import pyseekdb`, which can crash + This avoids importing a native embedded binding during `import pyseekdb`, which can crash on unsupported interpreter/platform combinations (e.g. some Python 3.14 CI environments). """ if name == "SeekdbEmbeddedClient": diff --git a/src/pyseekdb/client/client_seekdb_embedded.py b/src/pyseekdb/client/client_seekdb_embedded.py index 42efc8ea..31836431 100644 --- a/src/pyseekdb/client/client_seekdb_embedded.py +++ b/src/pyseekdb/client/client_seekdb_embedded.py @@ -1,8 +1,9 @@ """ Embedded mode client - based on seekdb -Note: Only available when pylibseekdb is installed (Linux only) +Available when the current ``seekdb`` binding or legacy ``pylibseekdb`` is installed. """ +import importlib import logging import os import threading @@ -12,25 +13,30 @@ import pymysql from pymysql.cursors import DictCursor -# Try to import pylibseekdb - it may not be available on all platforms -try: - import pylibseekdb as seekdb # type: ignore[import-not-found] - - _PYLIBSEEKDB_AVAILABLE = True -except ImportError: - seekdb = None # type: ignore[assignment] - _PYLIBSEEKDB_AVAILABLE = False - from .admin_client import DEFAULT_TENANT from .client_base import BaseClient from .database import Database from .sql_utils import render_sql_with_params + +def _load_embedded_runtime() -> tuple[Any | None, str | None]: + """Prefer the current ``seekdb`` binding, with legacy compatibility.""" + for distribution in ("seekdb", "pylibseekdb"): + try: + return importlib.import_module(distribution), distribution + except ImportError: + pass + return None, None + + +seekdb, _EMBEDDED_RUNTIME_DISTRIBUTION = _load_embedded_runtime() +_PYLIBSEEKDB_AVAILABLE = seekdb is not None + logger = logging.getLogger(__name__) class _NativeEmbeddedBackend: - """Connect through pylibseekdb's native Python connection wrapper.""" + """Connect through the embedded binding's native connection wrapper.""" supports_dbapi_cursor = False @@ -54,7 +60,7 @@ class _PyMySQLEmbeddedBackend: @staticmethod def connect(instance: Any, database: str, connection_kwargs: dict[str, Any]) -> pymysql.Connection: if instance is None: - raise RuntimeError("pylibseekdb returned no SeekdbInstance for a PyMySQL embedded connection") + raise RuntimeError("the embedded binding returned no SeekdbInstance for a PyMySQL connection") options = dict(instance.connection_options()) kwargs = dict(connection_kwargs) @@ -80,7 +86,7 @@ def is_connection_open(connection: Any) -> bool: def _create_embedded_backend() -> _NativeEmbeddedBackend | _PyMySQLEmbeddedBackend: - """Select the safest connection backend from pylibseekdb's capabilities.""" + """Select the safest connection backend exposed by the installed binding.""" instance_type = getattr(seekdb, "SeekdbInstance", None) if instance_type is not None and callable(getattr(instance_type, "connection_options", None)): return _PyMySQLEmbeddedBackend() @@ -90,7 +96,7 @@ def _create_embedded_backend() -> _NativeEmbeddedBackend | _PyMySQLEmbeddedBacke class SeekdbEmbeddedClient(BaseClient): """Embedded seekdb client (lazy connection) - Note: Only available on Linux platforms. pylibseekdb dependency is Linux-only. + Availability follows the installed embedded binding and its supported platform. """ def __init__(self, path: str = "./seekdb.db", database: str = "test", **kwargs): @@ -102,13 +108,13 @@ def __init__(self, path: str = "./seekdb.db", database: str = "test", **kwargs): database: database name Raises: - RuntimeError: If pylibseekdb is not available + RuntimeError: If no embedded binding is available """ - # Check if pylibseekdb is available + # Check if an embedded binding is available. if not _PYLIBSEEKDB_AVAILABLE or seekdb is None: raise RuntimeError( - "Embedded Client is not available because pylibseekdb is not available. " - "Please install pylibseekdb (Linux only) or use RemoteServerClient (host/port) instead." + "Embedded Client is not available because no SeekDB Python binding was found. " + "Please install seekdb (or legacy pylibseekdb), or use RemoteServerClient (host/port) instead." ) self.path = os.path.abspath(path) @@ -169,7 +175,7 @@ def _ensure_connection(self) -> Any: return self._connection def _cleanup(self) -> None: - """Close the connection and its owning pylibseekdb instance.""" + """Close the connection and its owning embedded runtime instance.""" with self._connection_lock: connection = self._connection instance = self._instance diff --git a/tests/integration_tests/conftest.py b/tests/integration_tests/conftest.py index 5eda1f80..e1d207b1 100644 --- a/tests/integration_tests/conftest.py +++ b/tests/integration_tests/conftest.py @@ -4,6 +4,7 @@ """ import contextlib +import importlib import os import sys from pathlib import Path @@ -45,12 +46,20 @@ # ==================== Client Factory Functions ==================== +def require_embedded_runtime(): + """Require the current seekdb binding or its legacy pylibseekdb name.""" + for module_name in ("seekdb", "pylibseekdb"): + try: + importlib.import_module(module_name) + return + except ImportError: + continue + pytest.skip("seekdb embedded package is not installed") + + def create_embedded_client(): """Create an embedded client instance.""" - try: - import pylibseekdb # noqa: F401 - except ImportError: - pytest.skip("seekdb embedded package is not installed") + require_embedded_runtime() return pyseekdb.Client(path=SEEKDB_PATH, database=SEEKDB_DATABASE) @@ -100,10 +109,7 @@ def create_oceanbase_client(): # ==================== AdminClient Factory Functions ==================== def create_embedded_admin_client(): """Create an embedded admin client instance.""" - try: - import pylibseekdb # noqa: F401 - except ImportError: - pytest.skip("seekdb embedded package is not installed") + require_embedded_runtime() return pyseekdb.AdminClient(path=SEEKDB_PATH) diff --git a/tests/integration_tests/embedded_fulltext_support.py b/tests/integration_tests/embedded_fulltext_support.py new file mode 100644 index 00000000..19514013 --- /dev/null +++ b/tests/integration_tests/embedded_fulltext_support.py @@ -0,0 +1,409 @@ +"""Isolated, dependency-free harness; pyseekdb is imported only in the worker. + +Synthetic deterministic documents (not the original SciFact benchmark). The +scan/freezer uses the same SQL shape as the Linux issue #1382 reproducer, without +a vector index. Public hybrid_search coverage lives in the separate first pass. +""" + +from __future__ import annotations + +import argparse +import collections +import importlib.metadata +import importlib.util +import json +import platform +import re +import threading +import time +import traceback +from pathlib import Path + +ROWS = 24515 +WORKERS = 8 +SECONDS = 75 +INTERVAL = 0.120 +WRITE_BATCH = 32 +QUERIES = 300 +PASSES = 4 +TEXT = "the database retrieval benchmark" +TABLE = "`c$v2$16d82ab3afe311f186fe000000000000`" +SEARCH = ( + f"SELECT `_id`, MATCH(`document`) AGAINST('{TEXT}') AS relevance FROM {TABLE} " + f"WHERE MATCH(`document`) AGAINST('{TEXT}') ORDER BY relevance DESC LIMIT 10" +) +TRACE = re.compile(r"\bY[A-Za-z0-9]+-[A-Za-z0-9]+-\d+-\d+\b") + + +def document(ordinal): + """Every row matches the broad query while cohorts supply varied queries.""" + return ( + f"{TEXT} exercises full text postings during concurrent memtable release " + f"cohort{ordinal % QUERIES:03d} document ordinal{ordinal:08d}" + ) + + +def error_details(error): + """Retain wrapper/cause messages and only report explicit numeric codes.""" + chain, seen, codes = [], set(), [] + while error is not None and id(error) not in seen: + seen.add(id(error)) + chain.append(f"{type(error).__name__}: {error}") + if error.args and isinstance(error.args[0], int): + codes.append(error.args[0]) + codes.extend(int(value) for value in re.findall(r"\bcode[=:]\s*(-?\d+)\b", str(error))) + # Released bindings may expose a RuntimeError rather than numeric args, + # e.g. "execute sql failed OB_ERR_UNEXPECTED(4016): %s". + codes.extend(int(value) for value in re.findall(r"\bOB_[A-Z0-9_]+\((-?\d+)\)", str(error))) + error = error.__cause__ or error.__context__ + text = "\n".join(chain) + return {"exception_chain": chain, "error_codes": sorted(set(codes)), "trace_ids": sorted(set(TRACE.findall(text)))} + + +def capture_logs(root, evidence, label, traces=(), budget=128 * 1024 * 1024, output_limit=2 * 1024 * 1024): + """Scan once under a byte budget; prioritize errors over routine context. + + Each priority buffer is bounded by output_limit (at most 4x that limit in + memory). A full low-priority buffer must not stop the scan: the failing + trace can occur later in a large, noisy rotated log. + """ + if budget < 0 or output_limit <= 0: + raise ValueError("invalid diagnostic byte budget") + trace_patterns = [value.encode() for value in traces if value] + error_patterns = [ + b"-4016", + b"data/schema type does not match", + b"failed to get next row from memtable scanner", + b"ob_memtable_key.h", + ] + files = sorted( + (p for p in root.rglob("seekdb.log*") if p.is_file() and not p.is_symlink()), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + inventory = [] + priorities = ("trace-error", "trace", "error", "memtable-context") + buffers = [bytearray() for _ in priorities] + matched = [0] * len(priorities) + capped = [False] * len(priorities) + + def append(priority, content): + available = output_limit - len(buffers[priority]) + if len(content) > available: + capped[priority] = True + buffers[priority].extend(content[:available]) + + for path in files: + size = path.stat().st_size + count = min(size, budget) + inventory.append({"path": str(path), "size": size, "read_bytes": count}) + if not count: + continue + with path.open("rb") as stream: + stream.seek(size - count) + previous, following, remaining = collections.deque(maxlen=5), 0, count + context_priority = 3 + while remaining > 0: + line = stream.readline(min(65536, remaining)) + if not line: + break + remaining -= len(line) + is_trace = any(pattern in line for pattern in trace_patterns) + is_error = any(pattern in line for pattern in error_patterns) + priority = ( + 0 + if is_trace and is_error + else 1 + if is_trace + else 2 + if is_error + else 3 + if b"read_barrier_:true" in line + else None + ) + if priority is not None: + # Preserve the matching line before optional context so + # a huge preceding SQL line cannot consume the budget. + context = f"\nFILE={path}\n".encode() + line + b"[preceding context]\n" + b"".join(previous) + matched[priority] += 1 + append(priority, context) + following, context_priority = 5, priority + elif following: + append(context_priority, line) + following -= 1 + previous.append(line) + budget -= count + written = 0 + with (evidence / f"{label}-db-context.log").open("wb") as output: + for content in buffers: + part = content[: max(0, output_limit - written)] + output.write(part) + written += len(part) + (evidence / f"{label}-log-inventory.json").write_text( + json.dumps( + { + "files": inventory, + "context_bytes": written, + "context_capped": written >= output_limit, + "priority_matches": dict(zip(priorities, matched, strict=True)), + "priority_capped": dict(zip(priorities, capped, strict=True)), + "scan_capped": any(item["read_bytes"] < item["size"] for item in inventory), + }, + indent=2, + ) + ) + + +class Recorder: + """Flush errors immediately and snapshot the first failure before recycling.""" + + def __init__(self, root, evidence): + self.root, self.evidence = root, evidence + self.lock = threading.Lock() + self.errors = 0 + self.traces = set() + self.first = None + + def error(self, error, phase, client=None, **details): + event = {"time": time.time(), "phase": phase, **details, **error_details(error)} + if not event["trace_ids"] and client is not None: + try: + # Immediately on the failing worker's own connection; mark it + # as best-effort, not proof of a particular internal SDK SQL. + rows = client._server._execute("SELECT last_trace_id() FROM DUAL") + event["trace_ids"] = TRACE.findall(str(rows)) + event["trace_source"] = "same_connection_last_trace_id_best_effort" + except Exception as lookup_error: + event["trace_lookup_error"] = str(lookup_error) + with self.lock: + self.errors += 1 + self.traces.update(event["trace_ids"]) + first = self.first is None + if first: + self.first = event + # Keep a bounded sample but never lose the total failure count. + if self.errors <= 200: + with (self.evidence / "errors.jsonl").open("a") as stream: + stream.write(json.dumps(event) + "\n") + if first: + (self.evidence / "first-error.json").write_text(json.dumps(event, indent=2)) + if first: + print(f"[FTS][FIRST_ERROR] {json.dumps(event)}", flush=True) + capture_logs(self.root, self.evidence, "early", event["trace_ids"]) + + +def validate_report(report, scenario): + """A zero worker exit alone is insufficient with native binding teardown.""" + assert report.get("scenario") == scenario, report + assert report.get("completed") is True and report.get("errors") == 0, report + assert report.get("rows") == ROWS, report + if scenario == "first-pass": + assert report.get("first_pass_successes") == QUERIES, report + assert report.get("fts_successes") == QUERIES * PASSES, report + else: + counts = report.get("worker_successes", []) + assert len(counts) == WORKERS and all(value > 0 for value in counts), report + assert report.get("freeze_successes", 0) > 0, report + assert report.get("elapsed_seconds", 0) >= SECONDS, report + + +def first_pass(client, recorder, report): + import pyseekdb + + collection = client.create_collection( + "fts_first_pass", + configuration=pyseekdb.HNSWConfiguration(dimension=8, distance="cosine"), + embedding_function=None, + ) + vector = [float(i + 1) / 8 for i in range(8)] + for start in range(0, ROWS, 100): + indexes = range(start, min(start + 100, ROWS)) + collection.upsert( + ids=[f"doc-{i}" for i in indexes], + documents=[document(i) for i in indexes], + embeddings=[vector for _ in indexes], + metadatas=[{"ordinal": i} for i in indexes], + ) + if start % 1000 == 0: + print(f"[FTS] ingest={min(start + 100, ROWS)}/{ROWS}", flush=True) + assert collection.count() == ROWS + collection.refresh_index() + report.update(rows=ROWS, first_pass_successes=0, fts_successes=0) + # No FTS warm-up, sleep, retry or reopen before pass zero. + for pass_id in range(PASSES): + for query_id in range(QUERIES): + try: + result = collection.hybrid_search( + query={"where_document": {"$contains": f"{TEXT} cohort{query_id:03d}"}, "n_results": 10}, + knn=None, + n_results=10, + include=["documents", "metadatas"], + ) + assert result.get("ids") and result["ids"][0], result + report["fts_successes"] += 1 + if pass_id == 0: + report["first_pass_successes"] += 1 + except Exception as error: + recorder.error( + error, + "hybrid_search", + client, + pass_id=pass_id, + query_id=query_id, + query_text=f"{TEXT} cohort{query_id:03d}", + ) + print(f"[FTS] pass={pass_id} successes={report['fts_successes']} errors={recorder.errors}", flush=True) + + +def insert_batch(client, start, size): + # IDs/documents are generated here, not user-controlled SQL inputs. + values = ",".join(f"('doc-{i}', '{document(i)}', '{{\"ordinal\":{i}}}')" for i in range(start, start + size)) + client._server._execute(f"INSERT INTO {TABLE} (`_id`, `document`, `metadata`) VALUES {values}") + + +def scan_freeze(client, make_client, recorder, report): + client._server._execute( + f"CREATE TABLE {TABLE} (`__pk_increment` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, " + "`_id` VARCHAR(128) NOT NULL, `document` LONGTEXT NOT NULL, `metadata` JSON NOT NULL, " + "UNIQUE KEY uk_id (`_id`), FULLTEXT KEY idx_document (`document`) WITH PARSER space)" + ) + for start in range(0, ROWS, 100): + insert_batch(client, start, min(100, ROWS - start)) + if start % 1000 == 0: + print(f"[FTS] ingest={min(start + 100, ROWS)}/{ROWS}", flush=True) + counts = client._server._execute(f"SELECT COUNT(*) AS row_count FROM {TABLE}") + count = counts[0]["row_count"] if isinstance(counts[0], dict) else counts[0][0] + assert count == ROWS, counts + schema = client._server._execute(f"SHOW CREATE TABLE {TABLE}") + assert "FULLTEXT" in str(schema).upper() and "VECTOR" not in str(schema).upper(), schema + report.update( + rows=ROWS, + schema=str(schema), + search_sql=SEARCH, + write_batch_size=WRITE_BATCH, + worker_successes=[0] * WORKERS, + freeze_successes=0, + ) + stop = threading.Event() + deadline = [0.0] + started = [0.0] + + def start_window(): + started[0] = time.monotonic() + deadline[0] = started[0] + SECONDS + + barrier = threading.Barrier(WORKERS + 2, action=start_window) + + def worker(number): + connection = None + try: + connection = make_client() + connection._server._execute("SELECT 1") + barrier.wait(timeout=60) + while time.monotonic() < deadline[0] and not stop.is_set(): + started = time.monotonic() + try: + if number == WORKERS: + start = ROWS + report["freeze_successes"] * WRITE_BATCH + insert_batch(connection, start, WRITE_BATCH) + connection._server._execute("ALTER SYSTEM MINOR FREEZE") + report["freeze_successes"] += 1 + else: + rows = connection._server._execute(SEARCH) + assert rows, "full-text scan returned no rows" + report["worker_successes"][number] += 1 + except Exception as error: + recorder.error(error, "minor-freeze" if number == WORKERS else "scan", connection, worker=number) + if number == WORKERS: + stop.set() # No valid pressure if writes/freezes cannot execute. + if number == WORKERS: + stop.wait(max(0, INTERVAL - (time.monotonic() - started))) + except Exception as error: + recorder.error(error, "worker-setup", connection, worker=number) + barrier.abort() + stop.set() + finally: + if connection is not None: + try: + connection.close() + except Exception as error: + recorder.error(error, "worker-close", worker=number) + + threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(WORKERS + 1)] + for thread in threads: + thread.start() + try: + barrier.wait(timeout=60) + while not stop.wait(min(5, max(0, deadline[0] - time.monotonic()))) and time.monotonic() < deadline[0]: + print( + f"[FTS] scans={sum(report['worker_successes'])} freezes={report['freeze_successes']} " + f"errors={recorder.errors}", + flush=True, + ) + finally: + for thread in threads: + thread.join(timeout=65) + assert not any(thread.is_alive() for thread in threads), "FTS worker did not drain" + report["elapsed_seconds"] = time.monotonic() - started[0] + + +def run(scenario, root, evidence): + import pyseekdb + + runtime_distribution = "seekdb" if importlib.util.find_spec("seekdb") else "pylibseekdb" + + report = { + "scenario": scenario, + "completed": False, + "platform": platform.platform(), + "pyseekdb": importlib.metadata.version("pyseekdb"), + "embedded_runtime": runtime_distribution, + "embedded_runtime_version": importlib.metadata.version(runtime_distribution), + "corpus": "synthetic deterministic / 24515 documents", + "workers": WORKERS, + "duration_seconds": SECONDS, + "freeze_interval_seconds": INTERVAL, + } + recorder = Recorder(root, evidence) + client = None + try: + db_dir = root / "database" + assert not db_dir.exists(), "fresh database required" + + def make_client(): + return pyseekdb.Client(path=str(db_dir), database="test") + + client = make_client() + report["database_version"] = str(client._server._execute("SELECT VERSION()")) + client._server._execute("ALTER SYSTEM SET max_syslog_file_count = 50") + print(f"[FTS] start {json.dumps(report)}", flush=True) + if scenario == "first-pass": + first_pass(client, recorder, report) + else: + scan_freeze(client, make_client, recorder, report) + report["completed"] = True + except Exception as error: + recorder.error(error, "scenario", client) + traceback.print_exc() + finally: + if client is not None: + try: + client.close() + except Exception as error: + recorder.error(error, "close") + try: + capture_logs(root, evidence, "final", recorder.traces) + except Exception as error: + recorder.error(error, "diagnostics") + report.update(errors=recorder.errors, first_error=recorder.first, trace_ids=sorted(recorder.traces)) + (evidence / "summary.json").write_text(json.dumps(report, indent=2)) + validate_report(report, scenario) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--scenario", choices=["first-pass", "scan-freeze"], required=True) + parser.add_argument("--root", type=Path, required=True) + parser.add_argument("--evidence", type=Path, required=True) + options = parser.parse_args() + run(options.scenario, options.root, options.evidence) diff --git a/tests/integration_tests/test_embedded_fulltext_stability.py b/tests/integration_tests/test_embedded_fulltext_stability.py new file mode 100644 index 00000000..292412a3 --- /dev/null +++ b/tests/integration_tests/test_embedded_fulltext_stability.py @@ -0,0 +1,69 @@ +"""Fresh-ingestion and memtable-release regressions for seekdb issue #1382. + +Each case runs in a fresh process and database directory, including with legacy +process-global pylibseekdb. Do not rerun failed queries to obtain a passing result. +""" + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +import pytest +from embedded_fulltext_support import validate_report + + +@pytest.mark.parametrize("scenario", ["first-pass", "scan-freeze"]) +def test_embedded_fulltext_stability(scenario): + """Exercise the public SDK first pass and an accelerated kernel stress case.""" + repo = Path(__file__).resolve().parents[2] + data_root = Path(os.environ.get("SEEKDB_TEST_DATA_ROOT", repo / ".seekdb-test-data")) + data_root.mkdir(parents=True, exist_ok=True) + root = Path(tempfile.mkdtemp(prefix=f"fts-{scenario}-", dir=data_root)) + evidence_root = Path(os.environ.get("SEEKDB_TEST_ARTIFACT_ROOT", data_root / "artifacts")) + evidence = evidence_root / root.name + evidence.mkdir(parents=True, exist_ok=False) + command = [ + sys.executable, + str(Path(__file__).with_name("embedded_fulltext_support.py")), + "--scenario", + scenario, + "--root", + str(root), + "--evidence", + str(evidence), + ] + # A whole-case timeout includes ingestion and connection setup; the pressure + # window itself is 75 seconds, starting only after every worker is ready. + with (evidence / "worker.log").open("w") as log: + # Fixed interpreter/helper and generated paths, never shell input. + process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT) # noqa: S603 + started = time.monotonic() + try: + with (evidence / "worker.log").open() as progress: + while process.poll() is None: + if time.monotonic() - started > 900: + raise TimeoutError(f"FTS worker exceeded 900s; evidence={evidence}") + # Stream bounded real output rather than hiding all ingestion + # and pressure activity inside pytest until the case ends. + text = progress.read(16384) + if text: + print(text, end="", flush=True) + time.sleep(1) + print(progress.read(), end="", flush=True) + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=10) + report_path = evidence / "summary.json" + assert report_path.is_file(), f"worker produced no terminal report; evidence={evidence}" + report = json.loads(report_path.read_text()) + validate_report(report, scenario) + assert process.returncode == 0, f"worker rc={process.returncode}; evidence={evidence}" + # Remove only our generated DB after the process has exited and evidence has + # been collected. Failed cases retain their DB for investigation. + shutil.rmtree(root) diff --git a/tests/integration_tests/test_get_or_create_collection_multiprocess.py b/tests/integration_tests/test_get_or_create_collection_multiprocess.py index 248a5028..a2379618 100644 --- a/tests/integration_tests/test_get_or_create_collection_multiprocess.py +++ b/tests/integration_tests/test_get_or_create_collection_multiprocess.py @@ -144,22 +144,25 @@ def _refresh_collection(client_config: dict[str, Any], collection_name: str) -> collection.refresh_index() -def _require_embedded_pylibseekdb() -> None: - """Require embedded pylibseekdb.""" +def _require_embedded_seekdb() -> None: + """Require the current seekdb binding or its legacy pylibseekdb name.""" + distribution = "seekdb" try: - import pylibseekdb # noqa: F401 + importlib.import_module(distribution) except ImportError: - pytest.skip("seekdb embedded package is not installed") + distribution = "pylibseekdb" + try: + importlib.import_module(distribution) + except ImportError: + pytest.skip("seekdb embedded package is not installed") try: - installed_version = Version(importlib.metadata.version("pylibseekdb")) + installed_version = Version(importlib.metadata.version(distribution)) except importlib.metadata.PackageNotFoundError: - pytest.skip("pylibseekdb is not installed") + pytest.skip(f"{distribution} is not installed") if installed_version < MIN_PYLIBSEEKDB_VERSION: - pytest.skip( - f"embedded multiprocess tests require pylibseekdb >= {MIN_PYLIBSEEKDB_VERSION}, got {installed_version}" - ) + pytest.skip(f"embedded multiprocess tests require seekdb >= {MIN_PYLIBSEEKDB_VERSION}, got {installed_version}") def _run_processes( @@ -486,7 +489,7 @@ def _build_client_config(mode: str) -> tuple[dict[str, Any], Path | None, Any]: temp_db_path: Path | None = None if mode == "embedded": - _require_embedded_pylibseekdb() + _require_embedded_seekdb() SEEKDB_TEST_DATA_ROOT.mkdir(parents=True, exist_ok=True) temp_db_path = Path(tempfile.mkdtemp(prefix="seekdb-mp-", dir=SEEKDB_TEST_DATA_ROOT)) client_config = {"mode": "embedded", "path": str(temp_db_path), "database": database}