From 6d2c1b91c553681ffc1ffad4c8e8c197338771ff Mon Sep 17 00:00:00 2001 From: deedy5 <65482418+deedy5@users.noreply.github.com> Date: Sun, 17 May 2026 07:38:19 +0300 Subject: [PATCH] feat: remove dht --- .github/workflows/python-package.yml | 20 +- README.md | 104 ---- ddgs/api_server/__init__.py | 11 +- ddgs/api_server/api.py | 181 ------ ddgs/api_server/dht_service.py | 273 --------- ddgs/cli.py | 9 - ddgs/ddgs.py | 233 +------- ddgs/dht/__init__.py | 293 ---------- ddgs/dht/cache.py | 204 ------- ddgs/dht/libp2p_client.py | 790 --------------------------- ddgs/dht/types.py | 80 --- pyproject.toml | 18 - tests/dht_test.py | 340 ------------ 13 files changed, 4 insertions(+), 2552 deletions(-) delete mode 100644 ddgs/api_server/dht_service.py delete mode 100644 ddgs/dht/__init__.py delete mode 100644 ddgs/dht/cache.py delete mode 100644 ddgs/dht/libp2p_client.py delete mode 100644 ddgs/dht/types.py delete mode 100644 tests/dht_test.py diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 78e8c30d..d39fbc52 100755 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -27,29 +27,13 @@ jobs: uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} - - name: Install system dependencies (macOS) - if: matrix.os == 'macos-latest' - run: | - brew install gmp - # Add homebrew paths to compiler search paths - if [ "$(uname -m)" = "arm64" ]; then - echo "CFLAGS=-I/opt/homebrew/include $CFLAGS" >> $GITHUB_ENV - echo "LDFLAGS=-L/opt/homebrew/lib $LDFLAGS" >> $GITHUB_ENV - else - echo "CFLAGS=-I/usr/local/include $CFLAGS" >> $GITHUB_ENV - echo "LDFLAGS=-L/usr/local/lib $LDFLAGS" >> $GITHUB_ENV - fi + - name: Install dependencies shell: bash run: | python -m pip install -U pip - if [ "${{ matrix.os }}" = "windows-latest" ]; then - # DHT dependencies require GMP which is difficult to install on Windows - python -m pip install .[dev,api,mcp] - else - python -m pip install .[dev,api,mcp,dht] - fi + python -m pip install .[dev,api,mcp] - name: Ruff run: | ruff check . diff --git a/README.md b/README.md index 5922baac..620e90f9 100755 --- a/README.md +++ b/README.md @@ -8,7 +8,6 @@ A metasearch library that aggregates results from diverse web search services. * [Install](#install) * [CLI version](#cli-version) * [API Server](#api-server) -* [DHT Network (BETA)](#dht-network-beta) * [MCP Server](#mcp-server) * [Engines](#engines) * [DDGS class](#ddgs-class) @@ -83,109 +82,6 @@ chmod +x start_api.sh [Go To TOP](#TOP) ___ -## DHT Network (BETA) - -DDGS includes an optional peer-to-peer distributed cache network. Search results are shared anonymously between users, drastically reducing rate limits and latency for everyone. - -This is the only metasearch engine that can actually get faster the more people use it. - ---- - -### Why this exists - -Every DDGS user currently hits search engines independently. Everyone gets rate limited individually. Everyone waits 1-2 seconds for exactly the same results. - -The DHT network fixes this: -- ✅ **90% faster** repeated queries (50ms instead of 1-2s) -- ✅ **No rate limits** for common queries -- ✅ **Zero infrastructure cost** - runs entirely on user machines -- ✅ **No central server** - cannot be blocked or shut down -- ✅ **Anonymous** - no one sees your queries, no logs - -### How it works - -When running: -1. Your node automatically discovers other DDGS users -2. All existing API endpoints automatically check the network first -3. If found, returns results immediately -4. If not, searches normally and shares the result anonymously -5. Every new user makes the network faster for everyone else - ---- - -### Installation - -```bash -# Install base DHT package -pip install -U ddgs[dht] - -# Install required dependencies (works on Linux and macOS) -pip install coincurve@git+https://github.com/ofek/coincurve.git@7829b29c08ebb1cc80386a1cdaf8c2243c4ef5c5 -pip install libp2p@git+https://github.com/libp2p/py-libp2p.git@0e88584c89377086883c6f5b26cd1a8052399be7 - -# macOS only: First install gmp -brew install gmp - -# Windows: DHT is not supported. Use base package only. -``` - -When installed, DHT: -- Adds persistent local result cache -- Adds `/dht/*` endpoints to the API server automatically -- Starts full distributed network node when you run `ddgs api` -- Works fully transparently - all existing search methods automatically use cache - -> **Platform Support**: DHT works on Linux and macOS. Windows is not currently supported due to libp2p dependencies. - -### Running - -1. Using terminal -```bash -ddgs api -d -``` -2. Using DDGS initialization -```python -from ddgs import DDGS - -ddgs = DDGS(api_url="http://localhost:4479", spawn_api=True) -``` - -You do **not** need to change any existing code. Any Python script using `DDGS().text()`, `images()`, etc. will automatically use both the local cache and the distributed DHT network. - -DHT will automatically start in the background and begin participating in the network. - -### DHT API Endpoints - -| Endpoint | Method | Description | -|---|---|---| -| `/dht/cache` | GET | Get cached results | -| `/dht/cache` | POST | Store results to cache | -| `/dht/cache` | DELETE | Invalidate cached results | -| `/dht/status` | GET | DHT service status and metrics | -| `/dht/peers` | GET | List connected peers | -| `/dht/peers/detailed` | GET | Detailed peer information | -| `/dht/map` | GET | Network graph view | -| `/dht/metrics` | GET | Prometheus format metrics | - ---- - -### Beta Status - -This is working beta software. All functionality is complete, but network performance will improve as more users join: - -| Network Size | Performance | -|--------------|-------------| -| < 50 nodes | Marginal - expect timeouts | -| 50-200 nodes | Good - >95% success rate | -| >200 nodes | Excellent - near perfect | - -Results use eventual consistency and may take 1-5 minutes to propagate. - -Please report any issues here: https://github.com/deedy5/ddgs/issues - -[Go To TOP](#TOP) -___ - ## MCP Server - **Install** diff --git a/ddgs/api_server/__init__.py b/ddgs/api_server/__init__.py index 3c5d90ae..ffe15693 100644 --- a/ddgs/api_server/__init__.py +++ b/ddgs/api_server/__init__.py @@ -1,7 +1,6 @@ """DDGS API server. -This module provides the FastAPI application for the DDGS REST API -and the distributed network cache service. +This module provides the FastAPI application for the DDGS REST API. """ __all__: list[str] = [] @@ -13,11 +12,3 @@ except ImportError: # API dependencies not installed pass - -try: - from ddgs.api_server.dht_service import get_dht_service - - __all__ += ["get_dht_service"] -except ImportError: - # DHT dependencies not installed - pass diff --git a/ddgs/api_server/api.py b/ddgs/api_server/api.py index a0416972..978c9fc7 100644 --- a/ddgs/api_server/api.py +++ b/ddgs/api_server/api.py @@ -7,7 +7,6 @@ from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import Response from pydantic import BaseModel, Field from ddgs import DDGS @@ -450,186 +449,6 @@ async def extract_content_get(url: str, fmt: str = "text_markdown") -> dict[str, raise HTTPException(status_code=500, detail=f"Extraction failed: {e!s}") from e -# Optional DHT Endpoints - only added if DHT dependencies are installed -try: - from pydantic import BaseModel, Field - - class CacheRequest(BaseModel): - """Request model for cache operations.""" - - query: str = Field(..., description="Search query") - results: list[dict[str, Any]] = Field(..., description="Search results to cache") - category: str = Field("text", description="Search category") - - class DhtStatusResponse(BaseModel): - """Response model for DHT status.""" - - running: bool - connected: bool - peer_id: str | None - port: int | None - cache_size: int - cache_count: int - peer_count: int - routing_table_size: int - metrics: dict[str, Any] - - @app.get("/dht/cache", response_model=SearchResponse) - async def get_cached(query: str, category: str = "text") -> SearchResponse: - """Get cached results from DHT.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - results = dht.get_cached(query, category) - if results is None: - raise HTTPException(status_code=404, detail="Not found in cache") - return SearchResponse(results=results) - - @app.post("/dht/cache", status_code=201) - async def cache_results(request: CacheRequest) -> dict[str, str]: - """Cache results to DHT.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - dht.cache(request.query, request.results, request.category) - return {"status": "ok"} - - @app.delete("/dht/cache", status_code=204) - async def invalidate_cache(query: str, category: str = "text") -> None: - """Invalidate cached results.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - from ddgs.dht.types import compute_query_hash # noqa: PLC0415 - - dht = get_dht_service() - query_hash = compute_query_hash(query, category) - if dht._cache: - dht._cache.delete(query_hash) - - @app.get("/dht/status", response_model=DhtStatusResponse) - async def dht_status() -> DhtStatusResponse: - """Get DHT service status.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - status = dht.get_status() - return DhtStatusResponse(**status) - - @app.get("/dht/peers", response_model=list[str]) - async def dht_peers() -> list[str]: - """Get list of connected DHT peers.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - return dht.get_peers() - - @app.get("/dht/peers/detailed", response_model=list[dict[str, Any]]) - async def dht_peers_detailed() -> list[dict[str, Any]]: - """Get detailed information about all connected peers.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - return dht._dht.get_neighbors() if dht._dht else [] - - @app.get("/dht/map") - async def dht_map() -> dict[str, Any]: - """Get local DHT network map view as graph structure. - - Each node only sees its own routing table neighborhood. - No global crawling, no gossip, zero additional network traffic. - """ - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - if not dht._dht: - return {"nodes": [], "edges": [], "total_peers_estimated": 0, "local_view_size": 0, "bucket_depth": 0} - - peers = dht._dht.get_neighbors() - - # Build graph - nodes = [] - edges = [] - - # Add self node - nodes.append( - { - "id": dht._dht.peer_id, - "type": "self", - "connections": len(peers), - } - ) - - # Add neighbor nodes - for peer in peers: - nodes.append( - { - "id": peer["peer_id"], - "type": "peer", - "xor_distance": peer["xor_distance"], - "latency_ms": peer["latency_ms"], - } - ) - edges.append( - { - "source": dht._dht.peer_id, - "target": peer["peer_id"], - "latency_ms": peer["latency_ms"], - } - ) - - # Network size estimation using Kademlia math - # For 256 bit ID space: estimated_size = 2^(average_bucket_depth) - bucket_depth = next((i for i, count in enumerate(dht._dht.kbucket_distribution) if count == 0), 255) - estimated_size = 2**bucket_depth - - return { - "nodes": nodes, - "edges": edges, - "total_peers_estimated": estimated_size, - "local_view_size": len(nodes), - "bucket_depth": bucket_depth, - } - - @app.get("/dht/metrics") - async def dht_metrics() -> Response: - """Get DHT metrics in Prometheus format.""" - from ddgs.api_server import get_dht_service # noqa: PLC0415 - - dht = get_dht_service() - status = dht.get_status() - - metrics = [ - "# HELP ddgs_dht_running Whether DHT service is running", - "# TYPE ddgs_dht_running gauge", - f"ddgs_dht_running {1 if status['running'] else 0}", - "# HELP ddgs_dht_connected_peers Number of currently connected peers", - "# TYPE ddgs_dht_connected_peers gauge", - f"ddgs_dht_connected_peers {status.get('peer_count', 0)}", - "# HELP ddgs_dht_cache_entries Number of entries in local cache", - "# TYPE ddgs_dht_cache_entries gauge", - f"ddgs_dht_cache_entries {status['cache_count']}", - "# HELP ddgs_dht_cache_size_bytes Size of local cache in bytes", - "# TYPE ddgs_dht_cache_size_bytes gauge", - f"ddgs_dht_cache_size_bytes {status['cache_size']}", - "# HELP ddgs_dht_query_success_rate DHT query success rate (0-1)", - "# TYPE ddgs_dht_query_success_rate gauge", - f"ddgs_dht_query_success_rate {status['metrics'].get('query_success_rate', 1.0)}", - "# HELP ddgs_dht_average_query_latency_ms Average DHT query latency in milliseconds", - "# TYPE ddgs_dht_average_query_latency_ms gauge", - f"ddgs_dht_average_query_latency_ms {status['metrics'].get('average_query_latency_ms', 0.0)}", - "# HELP ddgs_dht_routing_table_size Number of entries in DHT routing table", - "# TYPE ddgs_dht_routing_table_size gauge", - f"ddgs_dht_routing_table_size {status.get('routing_table_size', 0)}", - ] - - return Response("\n".join(metrics), media_type="text/plain") - - logger.info("DHT endpoints enabled - distributed cache available") - -except ImportError: - # DHT dependencies not installed - silently skip adding endpoints - pass - - if __name__ == "__main__": import uvicorn diff --git a/ddgs/api_server/dht_service.py b/ddgs/api_server/dht_service.py deleted file mode 100644 index b739bb83..00000000 --- a/ddgs/api_server/dht_service.py +++ /dev/null @@ -1,273 +0,0 @@ -"""DHT service for distributed cache - runs in background thread with Trio.""" - -import logging -import threading -import time -from typing import Any - -import trio - -from ddgs.dht import Libp2pClient, ResultCache, compute_query_hash, normalize_query -from ddgs.dht.cache import DEFAULT_TTL -from ddgs.dht.libp2p_client import DEFAULT_MAX_HOP_CONNECTIONS - -logger = logging.getLogger(__name__) - - -class DhtService: - """Background service running Trio event loop with Libp2pClient and ResultCache. - - Provides thread-safe access to distributed cache DHT from FastAPI. - """ - - def __init__( - self, - *, - listen_port: int = 0, - cache_ttl: int = DEFAULT_TTL, - max_hop_connections: int = DEFAULT_MAX_HOP_CONNECTIONS, - ) -> None: - self.listen_port = listen_port - self.cache_ttl = cache_ttl - self.max_hop_connections = max_hop_connections - - self._dht: Libp2pClient | None = None - self._cache: ResultCache | None = None - self._thread: threading.Thread | None = None - self._started = False - self._start_event = threading.Event() - self._stop_event = threading.Event() - self._trio_token: trio.lowlevel.TrioToken | None = None - - def _run_trio(self) -> None: - """Run Trio event loop in background thread.""" - - async def _main() -> None: - self._trio_token = trio.lowlevel.current_trio_token() - self._cancel_scope = trio.CancelScope() - self._dht = Libp2pClient( - listen_port=self.listen_port, - bootstrap=True, - max_hop_connections=self.max_hop_connections, - ) - self._cache = ResultCache() - - # Use async start directly now that libp2p is properly async - await self._dht.astart() - logger.info( - "Network service started: connected=%s, port=%d", - self._dht.is_running, - self._dht.port, - ) - self._start_event.set() - - with self._cancel_scope: - await trio.sleep_forever() - - try: - trio.run(_main) - except Exception as ex: # noqa: BLE001 - logger.warning("Network service loop exited: %r", ex) - finally: - self._start_event.set() - self._stop_event.set() - - def start(self, timeout: float = 10.0) -> bool: - """Start the DHT service in background thread. - - Args: - timeout: Maximum time to wait for service to start. - - Returns: - True if service started successfully. - - """ - if self._started: - return True - - self._thread = threading.Thread(target=self._run_trio, daemon=True) - self._thread.start() - - if self._start_event.wait(timeout=timeout): - self._started = True - return True - logger.warning("Network service failed to start within %ds", timeout) - return False - - def stop(self) -> None: - """Stop the DHT service.""" - if not self._started or self._dht is None: - return - - # Cancel the main loop first - if hasattr(self, "_cancel_scope"): - self._cancel_scope.cancel() - - try: - self._dht.stop() - except Exception as ex: # noqa: BLE001 - logger.warning("Error stopping DHT service: %r", ex) - - if self._thread: - self._thread.join(timeout=5.0) - - self._started = False - - def _run_in_trio(self, func: Any, *args: Any) -> Any: # noqa: ANN401 - """Run an async function in the Trio event loop from another thread.""" - if not self._started or self._trio_token is None: - return None - - async def _wrapped() -> Any: # noqa: ANN401 - with trio.fail_after(30): - return await func(*args) - - return trio.from_thread.run(_wrapped, trio_token=self._trio_token) - - def get_cached(self, query: str, category: str = "text") -> list[dict[str, Any]] | None: - """Get cached results synchronously (for use from FastAPI). - - Args: - query: The search query. - category: The search category. - - Returns: - Cached results or None. - - """ - if not self._started or self._dht is None or self._cache is None: - return None - - query_hash = compute_query_hash(query, category) - normalized = normalize_query(query) - - local_results = self._cache.get(query_hash) - if local_results: - logger.debug("Cache hit (local): %s", normalized) - return local_results - - if self._dht.is_running: - try: - dht_results: list[dict[str, Any]] | None = self._run_in_trio(self._dht.aget, query_hash) - if dht_results: - logger.debug("Cache hit (network): %s", normalized) - self._cache.set(query_hash, normalized, category, dht_results, self.cache_ttl) - return dht_results - # Add to negative cache to avoid repeated DHT lookups - self._cache.add_negative(query_hash) - except Exception as ex: # noqa: BLE001 - logger.debug("DHT get failed for %s: %r", query_hash, ex) - - return None - - def cache(self, query: str, results: list[dict[str, Any]], category: str = "text") -> None: - """Cache search results synchronously (for use from FastAPI). - - Args: - query: The search query. - results: The search results. - category: The search category. - - """ - if not self._started or self._dht is None or self._cache is None: - return - - query_hash = compute_query_hash(query, category) - normalized = normalize_query(query) - - self._cache.set(query_hash, normalized, category, results, self.cache_ttl) - - if self._dht.is_running: - try: - self._run_in_trio(self._dht.aset, query_hash, results, self.cache_ttl) - logger.debug("Cached to network: %s", normalized) - except Exception as ex: # noqa: BLE001 - logger.debug("DHT set failed for %s: %r", query_hash, ex) - - def get_status(self) -> dict[str, Any]: - """Get DHT service status. - - Returns: - Dict with status information. - - """ - if not self._started or self._dht is None: - return { - "running": False, - "connected": False, - "peer_id": None, - "port": None, - "cache_size": 0, - "cache_count": 0, - "metrics": {}, - } - - return { - "running": True, - "connected": self._dht.is_running, - "peer_id": self._dht.peer_id, - "port": self._dht.port, - "listen_addrs": self._dht.listen_addrs, - "cache_size": self._cache.size_bytes() if self._cache else 0, - "cache_count": self._cache.count() if self._cache else 0, - "peer_count": len(self._dht.find_peers()), - "routing_table_size": self._dht.routing_table_size, - "metrics": { - "query_success_rate": self._dht.query_success_rate, - "average_query_latency_ms": self._dht.average_query_latency_ms, - "kbucket_distribution": self._dht.kbucket_distribution, - "uptime_seconds": time.time() - self._dht.metrics["started_at"], - }, - } - - def get_peers(self) -> list[str]: - """Get list of connected peers. - - Returns: - List of peer IDs. - - """ - if not self._started or self._dht is None: - return [] - - try: - peers: list[str] = self._run_in_trio(self._dht.afind_peers) - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to get peers: %r", ex) - return [] - else: - return peers or [] - - @property - def is_running(self) -> bool: - """Check if DHT service is running.""" - return self._started and self._dht is not None and self._dht.is_running - - @property - def cache_count(self) -> int: - """Get number of cached entries.""" - return self._cache.count() if self._cache else 0 - - @property - def cache_size(self) -> int: - """Get cache size in bytes.""" - return self._cache.size_bytes() if self._cache else 0 - - -_dht_service: DhtService | None = None -_dht_service_lock = threading.Lock() - - -def get_dht_service() -> DhtService: - """Get the global DHT service instance. - - Returns: - DhtService singleton. - - """ - global _dht_service # noqa: PLW0603 - with _dht_service_lock: - if _dht_service is None: - _dht_service = DhtService() - _dht_service.start() - return _dht_service diff --git a/ddgs/cli.py b/ddgs/cli.py index d38bd206..390e8bfd 100644 --- a/ddgs/cli.py +++ b/ddgs/cli.py @@ -644,15 +644,6 @@ def api(detach: bool, stop: bool, host: str, port: int, reload: bool, proxy: str click.echo("Error: API dependencies not installed. Run: pip install 'ddgs[api]'", err=True) return - try: - # Pre-initialize DHT service if dependencies are available - from .api_server import get_dht_service # noqa: PLC0415 - - get_dht_service() - click.echo("API server starting with distributed DHT cache enabled") - except ImportError: - click.echo("API server starting (DHT cache not available - install ddgs[dht] to enable)") - # Prepare proxy environment variable proxy_env = os.environ.copy() if proxy: diff --git a/ddgs/ddgs.py b/ddgs/ddgs.py index 795c7265..40f853a2 100644 --- a/ddgs/ddgs.py +++ b/ddgs/ddgs.py @@ -1,22 +1,13 @@ """DDGS class implementation.""" -import asyncio -import atexit import logging import os -import subprocess -import sys -import threading -import time from concurrent.futures import ThreadPoolExecutor, wait from math import ceil -from pathlib import Path from random import random, shuffle from types import TracebackType from typing import Any, ClassVar -import primp - from .base import BaseSearchEngine from .engines import ENGINES from .exceptions import DDGSException, TimeoutException @@ -27,96 +18,6 @@ logger = logging.getLogger(__name__) -DEFAULT_API_URL = "http://localhost:4479" -NETWORK_START_TIMEOUT = 10.0 -NETWORK_CHECK_INTERVAL = 0.5 - - -_http_client: primp.Client | None = None -_cache_executor: ThreadPoolExecutor | None = None -_network_lock = threading.Lock() -_async_loop: asyncio.AbstractEventLoop | None = None -_async_thread: threading.Thread | None = None - - -def _cleanup_api_process() -> None: - """Cleanup any spawned API server process on exit.""" - if DDGS._api_process is not None: - try: - if DDGS._api_process.poll() is None: - logger.info("Stopping spawned API server (PID: %d)", DDGS._api_process.pid) - DDGS._api_process.terminate() - DDGS._api_process.wait(timeout=5.0) - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to stop API server: %r", ex) - try: - DDGS._api_process.kill() - DDGS._api_process.wait(timeout=2.0) - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to kill API server: %r", ex) - finally: - DDGS._api_process = None - - global _async_loop, _async_thread # noqa: PLW0603 - if _async_loop is not None and _async_loop.is_running(): - try: - _async_loop.call_soon_threadsafe(_async_loop.stop) - if _async_thread is not None: - _async_thread.join(timeout=5.0) - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to stop async loop: %r", ex) - finally: - _async_loop = None - _async_thread = None - - # Cleanup cache executor - global _cache_executor # noqa: PLW0603 - if _cache_executor is not None: - try: - _cache_executor.shutdown(wait=True) - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to shutdown cache executor: %r", ex) - finally: - _cache_executor = None - - -# Register cleanup handlers -atexit.register(_cleanup_api_process) - - -def _get_cache_executor() -> ThreadPoolExecutor: - """Get shared thread pool executor for background cache operations.""" - global _cache_executor # noqa: PLW0603 - if _cache_executor is None: - # Use a small fixed pool size to avoid thread exhaustion - _cache_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="DDGS-Cache") - return _cache_executor - - -def _get_http_client() -> primp.Client: - """Get or create the shared primp HTTP client.""" - global _http_client # noqa: PLW0603 - if _http_client is None: - _http_client = primp.Client(timeout=5) - return _http_client - - -def _get_async_loop() -> asyncio.AbstractEventLoop: - """Get or create shared async loop running in dedicated thread.""" - global _async_loop, _async_thread # noqa: PLW0603 - if _async_loop is None: - with _network_lock: - if _async_loop is None: - _async_loop = asyncio.new_event_loop() - - def _run_loop() -> None: - asyncio.set_event_loop(_async_loop) - _async_loop.run_forever() - - _async_thread = threading.Thread(target=_run_loop, daemon=True) - _async_thread.start() - return _async_loop - class DDGS: """DDGS | Dux Distributed Global Search. @@ -141,8 +42,6 @@ class DDGS: """ threads: ClassVar[int | None] = None - _network_client: ClassVar[Any] = None - _api_process: ClassVar[subprocess.Popen[str] | None] = None def __init__( self, @@ -150,22 +49,14 @@ def __init__( timeout: int | None = 5, *, verify: bool | str = True, - api_url: str | None = None, - spawn_api: bool = False, ) -> None: self._proxy = _expand_proxy_tb_alias(proxy) or os.environ.get("DDGS_PROXY") self._timeout = timeout self._verify = verify - self._api_url = api_url - self._spawn_api = spawn_api self._engines_cache: dict[ type[BaseSearchEngine[Any]], BaseSearchEngine[Any] ] = {} # dict[engine_class, engine_instance] - # Only enable network if api_url is provided - if self._api_url and DDGS._network_client is None: - self._ensure_network_running() - def __enter__(self) -> "DDGS": # noqa: PYI034 """Enter the context manager and return the DDGS instance.""" return self @@ -178,113 +69,6 @@ def __exit__( ) -> None: """Exit the context manager.""" - def _ensure_network_running(self) -> None: # noqa: C901, PLR0912 - """Ensure the network service is running. - - If not running, spawn a new ddgs api process only if spawn_api is True. - """ - if DDGS._network_client is not None: - return - - with _network_lock: - if DDGS._network_client is not None: - return - - api_url = self._api_url or DEFAULT_API_URL - - # Check if API is already running - try: - resp = _get_http_client().get(f"{api_url}/health") - if resp.status_code == 200: - # Import DhtClient only when needed (optional dependency) - from .dht import DhtClient # noqa: PLC0415 - - DDGS._network_client = DhtClient(api_url=api_url) - logger.info("Network client ready: %s", api_url) - return - except ImportError: - # DHT dependencies not installed - continue without network cache - logger.debug("DHT dependencies not available - network cache disabled") - except Exception as ex: # noqa: BLE001 - logger.debug("API health check failed: %r", ex) - - # API not running, spawn new process only if explicitly requested - if self._spawn_api and (DDGS._api_process is None or DDGS._api_process.poll() is not None): - try: - venv_bin = str(Path(sys.executable).parent) - env = {**os.environ, "PATH": f"{venv_bin}{os.pathsep}{os.environ.get('PATH', '')}"} - DDGS._api_process = subprocess.Popen( - [sys.executable, "-m", "ddgs", "api", "-d"], - env=env, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - start_new_session=True, - text=True, - ) - logger.info("Spawned ddgs api service (PID: %d)", DDGS._api_process.pid) - except Exception as ex: # noqa: BLE001 - logger.warning("Failed to spawn ddgs api: %r", ex) - return - - start_time = time.time() - while time.time() - start_time < NETWORK_START_TIMEOUT: - try: - resp = _get_http_client().get(f"{api_url}/health") - if resp.status_code == 200: - break - except Exception: # noqa: BLE001, S110 - pass - - # Check remaining time before sleeping - remaining = NETWORK_START_TIMEOUT - (time.time() - start_time) - if remaining <= 0: - break - time.sleep(min(NETWORK_CHECK_INTERVAL, remaining)) - else: - logger.warning("Timed out waiting for ddgs api service to start") - return - - # Import DhtClient only when needed (optional dependency) - try: - from .dht import DhtClient # noqa: PLC0415 - - DDGS._network_client = DhtClient(api_url=api_url) - logger.info("Network client ready: %s", api_url) - except ImportError: - logger.debug("DHT dependencies not available - network cache disabled") - - def _get_network_client(self) -> Any: # noqa: ANN401 - """Get the network client for caching. - - Returns: - DhtClient instance if available, None otherwise. - - """ - return DDGS._network_client - - def _cache_results_async( - self, - query: str, - results: list[dict[str, Any]], - category: str, - ) -> None: - """Cache search results asynchronously.""" - network = self._get_network_client() - if not network: - return - - def _cache_worker() -> None: - try: - loop = _get_async_loop() - future = asyncio.run_coroutine_threadsafe(network.cache(query, results, category), loop) - future.result(timeout=10.0) - logger.debug("Cached results: %s", query) - except Exception as ex: # noqa: BLE001 - logger.debug("Cache failed: %r", ex) - - executor = _get_cache_executor() - executor.submit(_cache_worker) - def _get_engines( self, category: str, @@ -348,7 +132,7 @@ def _get_engines( instances.sort(key=lambda e: (e.priority, random), reverse=True) return instances - def _search_sync( # noqa: C901, PLR0912 + def _search_sync( # noqa: C901 self, category: str, query: str, @@ -385,19 +169,6 @@ def _search_sync( # noqa: C901, PLR0912 msg = "query is mandatory." raise DDGSException(msg) - network = self._get_network_client() - if network: - try: - loop = _get_async_loop() - future = asyncio.run_coroutine_threadsafe(network.get_cached(query, category), loop) - cached = future.result(timeout=1.0) - if cached: - logger.debug("Cache hit: %s", query) - return cached # type: ignore[no-any-return] - except Exception as ex: # noqa: BLE001 - # Any cache failure, proceed normally to search - logger.debug("Cache check failed: %r", ex) - engines = self._get_engines(category, backend) len_unique_providers = len({engine.provider for engine in engines}) seen_providers: set[str] = set() @@ -445,8 +216,6 @@ def _search_sync( # noqa: C901, PLR0912 results = ranker.rank(results, query) if results: - if network: - self._cache_results_async(query, results, category) return results[:max_results] if max_results else results if "timed out" in f"{err}": diff --git a/ddgs/dht/__init__.py b/ddgs/dht/__init__.py deleted file mode 100644 index a1f5a398..00000000 --- a/ddgs/dht/__init__.py +++ /dev/null @@ -1,293 +0,0 @@ -"""Distributed cache DHT module. - -This module provides a P2P cache DHT for sharing search results -across multiple ddgs instances using Kademlia DHT. - -Usage: - from ddgs.dht import DhtClient - - # Direct mode (runs Trio in-process) - client = DhtClient() - await client.start() - - # REST API mode (connects to ddgs api service) - client = DhtClient(api_url="http://localhost:4479") - - # Get cached results - results = await client.get_cached("python tutorial") - - # Store results - await client.cache("python tutorial", results) - - await client.stop() -""" - -import asyncio -import logging -from typing import Any - -import primp - -from .cache import DEFAULT_TTL, MAX_CACHE_SIZE, ResultCache -from .libp2p_client import DEFAULT_MAX_HOP_CONNECTIONS, Libp2pClient -from .types import compute_query_hash, normalize_query - -__all__ = [ - "DEFAULT_MAX_HOP_CONNECTIONS", - "DEFAULT_TTL", - "MAX_CACHE_SIZE", - "DhtClient", - "Libp2pClient", - "ResultCache", - "compute_query_hash", - "normalize_query", -] - -logger = logging.getLogger(__name__) - -DEFAULT_API_URL = "http://localhost:4479" -CACHE_TIMEOUT = 1.0 - - -class DhtClient: - """Client for the distributed cache DHT. - - Provides transparent access to cached results from other peers - while falling back to direct search engine queries when needed. - - Supports two modes: - - Direct mode (api_url=None): Runs libp2p/Trio in-process - - REST mode (api_url set): Connects to ddgs api service via HTTP - """ - - def __init__( - self, - *, - enable_dht: bool = True, - listen_port: int = 0, - cache_ttl: int = DEFAULT_TTL, - max_hop_connections: int = DEFAULT_MAX_HOP_CONNECTIONS, - api_url: str | None = None, - ) -> None: - self.enable_dht = enable_dht - self.cache_ttl = cache_ttl - self.api_url = (api_url or DEFAULT_API_URL) if enable_dht and api_url is not None else None - self._started = False - self._dht: Libp2pClient | None - self._cache: ResultCache | None - - if self.api_url: - self._dht = None - self._cache = None - else: - self._dht = Libp2pClient( - listen_port=listen_port, - bootstrap=enable_dht, - max_hop_connections=max_hop_connections, - ) - self._cache = ResultCache() - - async def start(self) -> bool: - """Start the DHT client. - - Returns: - True if DHT connection successful. - - """ - if self._started: - return True - - if self.api_url: - logger.info("DHT client started (REST API mode): %s", self.api_url) - elif self.enable_dht and self._dht is not None: - # Run blocking start in thread to not block async loop - connected = await asyncio.to_thread(self._dht.start) - logger.info( - "DHT client started: connected=%s, port=%d", - connected, - self._dht.port, - ) - else: - logger.info("DHT client started (DHT disabled)") - - self._started = True - return self._started - - async def _get_from_api(self, query: str, category: str) -> list[dict[str, Any]] | None: - """Get cached results from API.""" - normalized = normalize_query(query) - try: - async with primp.AsyncClient(timeout=CACHE_TIMEOUT) as client: - resp = await client.get( - f"{self.api_url}/dht/cache", - params={"query": query, "category": category}, - ) - if resp.status_code == 200: - try: - data: dict[str, Any] = resp.json() - if results := data.get("results"): - logger.debug("Cache hit (DHT API): %s", normalized) - return results # type: ignore[no-any-return] - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to parse API response: %r", ex) - except Exception as ex: # noqa: BLE001 - logger.debug("DHT API get failed for %s: %r", normalized, ex) - return None - - async def get_cached( - self, - query: str, - category: str = "text", - ) -> list[dict[str, Any]] | None: - """Get cached results for a query. - - Checks local cache first, then DHT DHT (or REST API). - - Args: - query: The search query. - category: The search category. - - Returns: - Cached results or None. - - """ - if not self._started: - await self.start() - - query_hash = compute_query_hash(query, category) - normalized = normalize_query(query) - - if self.api_url: - return await self._get_from_api(query, category) - - if self._cache is None: - return None - - local_results = self._cache.get(query_hash) - if local_results: - logger.debug("Cache hit (local): %s", normalized) - return local_results - - if self._dht is not None and self._dht.is_running: - dht_results = self._dht.get(query_hash) - if dht_results: - logger.debug("Cache hit (DHT): %s", normalized) - self._cache.set(query_hash, normalized, category, dht_results, self.cache_ttl) - return dht_results - # Add to negative cache to avoid repeated DHT lookups - self._cache.add_negative(query_hash) - - return None - - async def cache( - self, - query: str, - results: list[dict[str, Any]], - category: str = "text", - ) -> None: - """Cache search results locally and on DHT. - - Args: - query: The search query. - results: The search results. - category: The search category. - - """ - if not self._started: - await self.start() - - query_hash = compute_query_hash(query, category) - normalized = normalize_query(query) - - if self.api_url: - try: - async with primp.AsyncClient(timeout=CACHE_TIMEOUT) as client: - resp = await client.post( - f"{self.api_url}/dht/cache", - json={"query": query, "results": results, "category": category}, - ) - if resp.status_code in (200, 201): - logger.debug("Cached to DHT API: %s", normalized) - except Exception as ex: # noqa: BLE001 - logger.debug("DHT API cache failed for %s: %r", normalized, ex) - return - - if self._cache is None: - return - - self._cache.set(query_hash, normalized, category, results, self.cache_ttl) - - if self._dht is not None and self._dht.is_running: - self._dht.set(query_hash, results, self.cache_ttl) - logger.debug("Cached to DHT: %s", normalized) - - async def invalidate(self, query: str, category: str = "text") -> None: - """Invalidate cached results for a query. - - Args: - query: The search query. - category: The search category. - - """ - query_hash = compute_query_hash(query, category) - - if self.api_url: - try: - async with primp.AsyncClient(timeout=CACHE_TIMEOUT) as client: - await client.delete( - f"{self.api_url}/dht/cache", - params={"query": query, "category": category}, - ) - except Exception as ex: # noqa: BLE001 - logger.debug("DHT API invalidate failed: %r", ex) - return - - if self._cache is None: - return - - self._cache.delete(query_hash) - - async def stop(self) -> None: - """Stop the DHT client.""" - if self._dht and self._dht.is_running: - self._dht.stop() - - self._started = False - - @property - def is_connected(self) -> bool: - """Check if the DHT client is connected.""" - if self.api_url: - return self._started - return self._dht.is_running if self._dht else False - - @property - def cache_count(self) -> int: - """Get the number of cached entries.""" - if self._cache: - return self._cache.count() - return 0 - - @property - def cache_size(self) -> int: - """Get the size of the cache in bytes.""" - if self._cache: - return self._cache.size_bytes() - return 0 - - -async def get_dht_client( - api_url: str | None = None, -) -> DhtClient: - """Get a shared DHT client instance. - - Args: - api_url: URL of the ddgs API service. If None, uses direct mode. - - Returns: - A started DhtClient instance. - - """ - client = DhtClient(api_url=api_url) - await client.start() - return client diff --git a/ddgs/dht/cache.py b/ddgs/dht/cache.py deleted file mode 100644 index 102b5f25..00000000 --- a/ddgs/dht/cache.py +++ /dev/null @@ -1,204 +0,0 @@ -"""Local cache with TTL expiration using SQLite.""" - -import hashlib -import json -import logging -import sqlite3 -import threading -import time -from pathlib import Path -from typing import Any - -logger = logging.getLogger(__name__) - -DEFAULT_TTL = 14400 -MAX_CACHE_SIZE = 10000 -NEGATIVE_CACHE_SIZE = 50000 -NEGATIVE_CACHE_TTL = 1800 # 30 minutes - - -class BloomFilter: - """Simple bloom filter for negative cache lookups.""" - - def __init__(self, size: int = NEGATIVE_CACHE_SIZE, hash_count: int = 3) -> None: - self.size = size - self.hash_count = hash_count - self.bit_array = bytearray(size // 8 + 1) - self._lock = threading.RLock() - - def _hashes(self, key: str) -> list[int]: - """Generate hash positions for a key using SHA-256 for consistency.""" - result = [] - for i in range(self.hash_count): - # Use key + seed for each hash, then truncate to 64-bit int - data = f"{key}:{i}".encode() - digest = hashlib.sha256(data).digest() - # Convert first 8 bytes to unsigned integer - h = int.from_bytes(digest[:8], byteorder="big", signed=False) - result.append(h % self.size) - return result - - def add(self, key: str) -> None: - """Add a key to the filter.""" - with self._lock: - for pos in self._hashes(key): - self.bit_array[pos // 8] |= 1 << (pos % 8) - - def __contains__(self, key: str) -> bool: - """Check if a key may be present.""" - with self._lock: - return all(self.bit_array[pos // 8] & (1 << (pos % 8)) for pos in self._hashes(key)) - - -class ResultCache: - """Thread-safe local SQLite cache with TTL support.""" - - def __init__(self, db_path: str | None = None) -> None: - if db_path: - self._db_path = Path(db_path) - else: - ddgs_dir = Path("~/.ddgs").expanduser() - ddgs_dir.mkdir(parents=True, exist_ok=True) - self._db_path = ddgs_dir / "cache.db" - - self._lock = threading.RLock() - self._negative_cache = BloomFilter() - self._negative_cache_times: dict[str, float] = {} - - # Initialize database schema - with sqlite3.connect(self._db_path) as conn: - self._init_db(conn) - - def _init_db(self, conn: sqlite3.Connection) -> None: - conn.execute(""" - CREATE TABLE IF NOT EXISTS cached_results ( - query_hash TEXT PRIMARY KEY, - query TEXT NOT NULL, - category TEXT NOT NULL, - results TEXT NOT NULL, - timestamp REAL NOT NULL, - ttl INTEGER NOT NULL - ) - """) - conn.execute(""" - CREATE INDEX IF NOT EXISTS idx_timestamp - ON cached_results(timestamp) - """) - conn.commit() - - def _parse_results(self, results_json: str) -> list[dict[str, Any]] | None: - """Parse JSON results safely.""" - try: - return json.loads(results_json) # type: ignore[no-any-return] - except Exception: # noqa: BLE001 - return None - - def get( - self, - query_hash: str, - ) -> list[dict[str, Any]] | None: - """Get cached results if not expired.""" - with self._lock, sqlite3.connect(self._db_path) as conn: - result = conn.execute( - "SELECT results, timestamp, ttl FROM cached_results WHERE query_hash = ?", - (query_hash,), - ).fetchone() - - if not result: - return None - - results_json, timestamp, ttl = result - if time.time() > timestamp + ttl: - # Expired, delete it - conn.execute( - "DELETE FROM cached_results WHERE query_hash = ?", - (query_hash,), - ) - conn.commit() - return None - - return self._parse_results(results_json) - - def set( - self, - query_hash: str, - query: str, - category: str, - results: list[dict[str, Any]], - ttl: int = DEFAULT_TTL, - ) -> None: - """Store results with TTL and enforce max size.""" - with self._lock, sqlite3.connect(self._db_path) as conn: - conn.execute( - "INSERT OR REPLACE INTO cached_results " - "(query_hash, query, category, results, timestamp, ttl) " - "VALUES (?, ?, ?, ?, ?, ?)", - ( - query_hash, - query, - category, - json.dumps(results, ensure_ascii=False, allow_nan=False), - time.time(), - ttl, - ), - ) - conn.commit() - - # Only run eviction when we exceed max size - count = conn.execute("SELECT COUNT(*) FROM cached_results").fetchone()[0] - if count > MAX_CACHE_SIZE: - excess = count - MAX_CACHE_SIZE + 1 # Remove extra to get under limit - conn.execute( - "DELETE FROM cached_results WHERE query_hash IN (" - "SELECT query_hash FROM cached_results " - "ORDER BY timestamp ASC LIMIT ?)", - (excess,), - ) - conn.commit() - logger.debug("Evicted %d old entries", excess) - - def delete(self, query_hash: str) -> None: - """Delete a cached result.""" - with self._lock, sqlite3.connect(self._db_path) as conn: - conn.execute( - "DELETE FROM cached_results WHERE query_hash = ?", - (query_hash,), - ) - conn.commit() - - def cleanup_expired(self) -> int: - """Remove all expired entries.""" - with self._lock, sqlite3.connect(self._db_path) as conn: - cursor = conn.execute( - "DELETE FROM cached_results WHERE timestamp + ttl < ?", - (time.time(),), - ) - conn.commit() - return cursor.rowcount - - def count(self) -> int: - """Get total number of cached entries.""" - with self._lock, sqlite3.connect(self._db_path) as conn: - return conn.execute("SELECT COUNT(*) FROM cached_results").fetchone()[0] # type: ignore[no-any-return] - - def __len__(self) -> int: - """Get total number of cached entries.""" - return self.count() - - def size_bytes(self) -> int: - """Get database size in bytes.""" - return self._db_path.stat().st_size if self._db_path.exists() else 0 - - def add_negative(self, query_hash: str) -> None: - """Add an entry to the negative cache (no results found).""" - with self._lock: - # Cleanup expired entries first when we reach 80% capacity - if len(self._negative_cache_times) >= NEGATIVE_CACHE_SIZE * 0.8: - now = time.time() - expired = [k for k, v in self._negative_cache_times.items() if now - v >= NEGATIVE_CACHE_TTL] - for k in expired: - del self._negative_cache_times[k] - - self._negative_cache.add(query_hash) - self._negative_cache_times[query_hash] = time.time() - logger.debug("Added to negative cache: %s", query_hash) diff --git a/ddgs/dht/libp2p_client.py b/ddgs/dht/libp2p_client.py deleted file mode 100644 index 90542c26..00000000 --- a/ddgs/dht/libp2p_client.py +++ /dev/null @@ -1,790 +0,0 @@ -"""libp2p DHT client for distributed cache network.""" - -import json -import logging -import secrets -import socket -import threading -import time -from typing import TYPE_CHECKING, Any - -import trio - -# Optional import for dnsaddr resolution -try: - import dns.resolver -except ImportError: - dns = None # type: ignore[assignment] -from libp2p import new_host -from libp2p.crypto.secp256k1 import create_new_key_pair -from libp2p.custom_types import TProtocol -from libp2p.kad_dht import kad_dht -from libp2p.kad_dht.kad_dht import DHTMode -from libp2p.records.pubkey import PublicKeyValidator -from libp2p.records.validator import NamespacedValidator, Validator -from libp2p.relay.circuit_v2 import CircuitV2Protocol, CircuitV2Transport -from libp2p.relay.circuit_v2.config import RelayConfig -from libp2p.relay.circuit_v2.resources import RelayLimits -from libp2p.stream_muxer.mplex.mplex import Mplex -from libp2p.tools.utils import info_from_p2p_addr # type: ignore[attr-defined] -from libp2p.utils.address_validation import get_available_interfaces -from multiaddr import Multiaddr - -if TYPE_CHECKING: - from libp2p.abc import IHost - -logger = logging.getLogger(__name__) - - -def _resolve_dnsaddr(multiaddr_str: str) -> list[str]: - """Resolve /dnsaddr/ multiaddrs into concrete addresses using DNS TXT records. - - Follows libp2p dnsaddr specification: - - Queries TXT records at _dnsaddr. - - Extracts all records starting with "dnsaddr=" - - Gracefully falls back to original address on any failure - - Requires dnspython for full functionality. - """ - try: - # Manual parsing since multiaddr library doesn't properly handle /dnsaddr/.../p2p/... format - parts = multiaddr_str.split("/") - dnsaddr_domain = None - - for i, part in enumerate(parts): - if part == "dnsaddr" and i + 1 < len(parts): - dnsaddr_domain = parts[i + 1] - break - - if not dnsaddr_domain: - return [multiaddr_str] - - if dns is None: - logger.debug("dnspython not installed, skipping dnsaddr resolution") - return [multiaddr_str] - - # Perform TXT lookup - answers = dns.resolver.resolve(f"_dnsaddr.{dnsaddr_domain}", "TXT") - results = [] - - for rdata in answers: - for txt_string in rdata.strings: - decoded = txt_string.decode("utf-8", errors="replace") - if decoded.startswith("dnsaddr="): - results.append(decoded[8:]) - - if results: - return results - return [multiaddr_str] # noqa: TRY300 - - except Exception: # noqa: BLE001 - return [multiaddr_str] - - -# Raw bootstrap node definitions (official IPFS bootstrap nodes) -# Source: https://docs.ipfs.tech/how-to/modify-bootstrap-list/ -_RAW_BOOTSTRAP_NODES = [ - "/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN", - "/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa", - "/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb", - "/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt", - "/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ", -] - -# Resolve all dnsaddr addresses once at module load time -BOOTSTRAP_NODES: list[str] = [] -for node in _RAW_BOOTSTRAP_NODES: - BOOTSTRAP_NODES.extend(_resolve_dnsaddr(node)) - -DEFAULT_MAX_HOP_CONNECTIONS = 10 - - -class DDGSValidator(Validator): # type: ignore[misc] - """Validator for the 'ddgs' namespace.""" - - def validate(self, key: str, value: bytes) -> None: # noqa: ARG002 - """Validate a key-value pair.""" - if not value: - msg = "Value cannot be empty" - raise ValueError(msg) - - def select(self, key: str, values: list[bytes]) -> int: # noqa: ARG002 - """Select the best value (first one).""" - return 0 - - -class Libp2pClient: - """libp2p DHT client for distributed cache network.""" - - def __init__( - self, - *, - listen_port: int = 0, - bootstrap: bool = True, - max_hop_connections: int = DEFAULT_MAX_HOP_CONNECTIONS, - refresh_interval: int = 3600, # Refresh DHT records every hour - ) -> None: - self.listen_port = listen_port - self.bootstrap_enabled = bootstrap - self.max_hop_connections = max_hop_connections - self._host: IHost | None = None - self._dht: kad_dht.KadDHT | None = None - self._running = False - self._host_cm = None - self._relay_transport: CircuitV2Transport | None = None - self._stored_keys: dict[str, tuple[list[dict[str, Any]], int]] = {} - self._refresh_interval = refresh_interval - self._dht_thread: threading.Thread | None = None - self._trio_token: trio.lowlevel.TrioToken | None = None - self._start_event = threading.Event() - self._stop_event: threading.Event | None = None - self._lock = threading.Lock() - - # Metrics tracking - self.metrics = { - "total_queries": 0, - "successful_queries": 0, - "failed_queries": 0, - "total_puts": 0, - "successful_puts": 0, - "failed_puts": 0, - "total_peers_seen": 0, - "query_latency_sum": 0.0, - "started_at": time.time(), - } - - @property - def query_success_rate(self) -> float: - """Get DHT query success rate (0-1).""" - if self.metrics["total_queries"] == 0: - return 1.0 - return self.metrics["successful_queries"] / self.metrics["total_queries"] - - @property - def average_query_latency_ms(self) -> float: - """Get average DHT query latency in milliseconds.""" - if self.metrics["successful_queries"] == 0: - return 0.0 - return (self.metrics["query_latency_sum"] / self.metrics["successful_queries"]) * 1000 - - @property - def routing_table_size(self) -> int: - """Get total number of peers in DHT routing table.""" - if not self._dht or not hasattr(self._dht, "routing_table"): - return 0 - return sum(len(bucket.peers) for bucket in self._dht.routing_table.buckets) - - @property - def kbucket_distribution(self) -> list[int]: - """Get peer count per k-bucket (256 buckets total).""" - if not self._dht or not hasattr(self._dht, "routing_table"): - return [0] * 256 - return [len(bucket.peers) for bucket in self._dht.routing_table.buckets] - - def _create_host(self) -> "IHost": - """Create a host with Mplex muxer (default security includes Noise).""" - secret = secrets.token_bytes(32) - key_pair = create_new_key_pair(secret) - - # Don't pass sec_opt - use default which includes Noise - muxer_opt: dict[str, type] = {"/mplex/6.7.0": Mplex} - - # Test if IPv6 is actually working - has_ipv6 = False - try: - sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) - sock.bind(("::1", 0)) - sock.close() - has_ipv6 = True - logger.debug("IPv6 loopback is functional") - except OSError: - logger.debug("IPv6 is not available, using IPv4 only") - - # If listen_port is 0, find an available port first - if self.listen_port == 0: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.bind(("127.0.0.1", 0)) - self.listen_port = sock.getsockname()[1] - sock.close() - logger.debug("Auto-selected port: %d", self.listen_port) - - # Build listen addresses - listen_addrs = [] - - # Always add loopback - we know this port is available since we just bound it - listen_addrs.append(Multiaddr(f"/ip4/127.0.0.1/tcp/{self.listen_port}")) - - # Add all available non-loopback IPv4 interfaces - all_addrs = get_available_interfaces(self.listen_port) - for addr in all_addrs: - addr_str = str(addr) - if ("/ip4/" in addr_str and "127.0.0.1" not in addr_str) or ( - has_ipv6 and "/ip6/" in addr_str and "::1" not in addr_str - ): - listen_addrs.append(addr) - - if has_ipv6: - listen_addrs.append(Multiaddr(f"/ip6/::1/tcp/{self.listen_port}")) - - logger.debug("Using listen addresses: %s", [str(a) for a in listen_addrs]) - - host = new_host( - key_pair=key_pair, - muxer_opt=muxer_opt, # type: ignore[arg-type] - listen_addrs=listen_addrs, - muxer_preference="MPLEX", - ) - - # Store actual bound addresses since new_host may re-bind ports - self._listen_addrs = host.get_addrs() - - # Update listen_port with actual bound port - for addr in self._listen_addrs: - addr_str = str(addr) - if "/ip4/127.0.0.1/tcp/" in addr_str: - self.listen_port = int(addr_str.split("/tcp/")[1].split("/")[0]) - break - - return host - - async def _setup_dht(self) -> None: - """Set up DHT instance with validator.""" - if not self._host: - return - - validator = NamespacedValidator( - { - "pk": PublicKeyValidator(), - "ddgs": DDGSValidator(), - } - ) - self._dht = kad_dht.KadDHT( - host=self._host, - mode=DHTMode.SERVER, - protocol_prefix=TProtocol("/ddgs"), - validator=validator, - ) - - async def _run_refresh_task(self) -> None: - """Background task to refresh DHT records periodically.""" - while self._running: - await trio.sleep(self._refresh_interval) - if not self._running: - break - - with self._lock: - # Make a copy under lock to avoid concurrent modification - stored_items = list(self._stored_keys.items()) - - logger.debug("Refreshing %d DHT records", len(stored_items)) - for key, (results, ttl) in stored_items: - try: - # Re-publish value to refresh its lifetime on the network - self.set(key, results, ttl=ttl) - except Exception as ex: # noqa: BLE001, PERF203 - logger.debug("Failed to refresh DHT record %s: %r", key, ex) - - async def _run_main_loop(self) -> None: - """Run main DHT event loop.""" - from libp2p.tools.anyio_service import AnyIOManager # noqa: PLC0415 - - self._stop_event = threading.Event() - - async with trio.open_nursery() as nursery: - # Start DHT service - if self._dht: - nursery.start_soon(AnyIOManager.run_service, self._dht) - - # Start DHT record refresh background task - nursery.start_soon(self._run_refresh_task) - - if self._host: - logger.info( - "DHT client started, listening on: %s", - [str(addr) for addr in self._host.get_addrs()], - ) - - await trio.to_thread.run_sync(self._stop_event.wait) - - def _run_trio(self) -> None: - """Run Trio event loop in background thread with all DHT operations.""" - - async def _main() -> None: - self._trio_token = trio.lowlevel.current_trio_token() - - try: - self._host = self._create_host() - self.listen_port = max(0, self.listen_port) - - # Reuse the same filtered listen addresses as _create_host - listen_addrs = self._host.get_addrs() - async with self._host.run(listen_addrs): - # Update listen_port with actual bound port first thing after start - for addr in self._host.get_addrs(): - addr_str = str(addr) - if "/ip4/127.0.0.1/tcp/" in addr_str: - self.listen_port = int(addr_str.split("/tcp/")[1].split("/", maxsplit=1)[0]) - break - - logger.info("Host started, peer ID: %s, port: %d", self._host.get_id(), self.listen_port) - - await self._setup_relay() - await self._setup_dht() - - # Now mark as running BEFORE signaling - self._running = True - - # Signal that we're started - all addresses and state are ready now - self._start_event.set() - - if self.bootstrap_enabled: - await self._connect_bootstrap() - - # Run the main loop WHILE host is running - await self._run_main_loop() - - finally: - # Cleanup - self._running = False - self._start_event.set() - - self._host = None - self._dht = None - self._relay_transport = None - self._stored_keys.clear() - - try: - trio.run(_main) - except Exception as ex: # noqa: BLE001 - logger.warning("DHT service loop exited: %r", ex) - finally: - self._running = False - self._start_event.set() - if self._stop_event: - self._stop_event.set() - - def start(self, timeout: float = 2.0) -> bool: - """Start the DHT client and connect to the network. - - Args: - timeout: Maximum time to wait for service to start. - - Returns: - True if connected to network, False otherwise. - - """ - with self._lock: - if self._running: - return True - - self._dht_thread = threading.Thread(target=self._run_trio, daemon=True) - self._dht_thread.start() - - if self._start_event.wait(timeout=timeout): - logger.info("DHT client started") - return True - - logger.warning("DHT client failed to start within %ds", timeout) - return False - - async def astart(self, timeout: float = 2.0) -> bool: # noqa: ASYNC109 - """Async wrapper for start() for backwards compatibility. - - Args: - timeout: Maximum time to wait for service to start. - - Returns: - True if connected to network, False otherwise. - - """ - return await trio.to_thread.run_sync(self.start, timeout) - - async def _setup_relay(self) -> None: - """Set up Circuit Relay v2 for NAT traversal.""" - if not self._host: - return - - try: - limits = RelayLimits( - duration=300, - data=1000000, - max_circuit_conns=self.max_hop_connections, - max_reservations=self.max_hop_connections, - ) - - relay_protocol = CircuitV2Protocol( - self._host, - limits=limits, - allow_hop=True, - ) - - relay_config = RelayConfig(limits=limits) - - self._relay_transport = CircuitV2Transport( - self._host, - relay_protocol, - relay_config, - ) - - logger.info( - "Circuit Relay v2 enabled (max_hop_connections=%d)", - self.max_hop_connections, - ) - - except Exception as ex: # noqa: BLE001 - logger.warning("Failed to setup relay: %r", ex) - - async def _connect_bootstrap(self) -> None: - """Connect to bootstrap nodes.""" - if not self._host: - return - - connected_count = 0 - for addr in BOOTSTRAP_NODES: - try: - peer = info_from_p2p_addr(Multiaddr(addr)) - self._host.get_peerstore().add_addrs(peer.peer_id, peer.addrs, 3600) - await self._host.connect(peer) - logger.info("Connected to bootstrap: %s", peer.peer_id) - connected_count += 1 - except Exception as ex: # noqa: BLE001, PERF203 - logger.debug("Bootstrap %s failed: %s", addr, ex) - - if connected_count == 0: - logger.warning("Failed to connect to any bootstrap nodes") - else: - logger.info("Successfully connected to %d bootstrap nodes", connected_count) - - def connect_peer(self, peer_addr: str) -> bool: - """Connect to a specific peer. - - Args: - peer_addr: The multiaddress of the peer to connect to. - - Returns: - True if connected successfully. - - """ - if not self._running or not self._host or not self._dht: - return False - - try: - - async def _connect() -> bool: - peer = info_from_p2p_addr(Multiaddr(peer_addr)) - self._host.get_peerstore().add_addrs(peer.peer_id, peer.addrs, 3600) # type: ignore[union-attr] - await self._host.connect(peer) # type: ignore[union-attr] - await self._dht.routing_table.add_peer(peer.peer_id) # type: ignore[union-attr] - logger.info("Connected to peer: %s", peer.peer_id) - return True - - result = self._run_in_trio(_connect, timeout=3.0) - except Exception as ex: # noqa: BLE001 - logger.warning("Failed to connect to peer %s: %s", peer_addr, ex) - return False - else: - return result if result is not None else False - - async def aconnect_peer(self, peer_addr: str) -> bool: - """Async wrapper for connect_peer() for backwards compatibility. - - Args: - peer_addr: The multiaddress of the peer to connect to. - - Returns: - True if connected successfully. - - """ - return await trio.to_thread.run_sync(self.connect_peer, peer_addr) - - def _run_in_trio(self, func: Any, *args: Any, timeout: float = 30.0) -> Any: # noqa: ANN401 - """Run an async function in the Trio event loop from another thread.""" - if not self._running or self._trio_token is None: - return None - - result = None - done = threading.Event() - - async def _runner() -> None: - nonlocal result - try: - with trio.fail_after(timeout): - result = await func(*args) - except Exception: # noqa: BLE001 - result = None - finally: - done.set() - - trio.from_thread.run(_runner, trio_token=self._trio_token) - done.wait(timeout + 0.5) - return result - - def get(self, key: str, *, timeout: float = 2.0) -> list[dict[str, Any]] | None: - """Get cached results from the DHT. - - Args: - key: The cache key (query hash). - timeout: Timeout in seconds for DHT query. - - Returns: - List of results or None if not found. - - """ - if not self._running or not self._dht: - return None - - start_time = time.time() - success = False - - try: - self.metrics["total_queries"] += 1 - - async def _get() -> list[dict[str, Any]] | None: - dht_key = f"/ddgs/{key}" - with trio.move_on_after(timeout) as cancel_scope: - result = await self._dht.get_value(dht_key) # type: ignore[union-attr] - if cancel_scope.cancelled_caught: - logger.debug("DHT get timed out for %s", key) - return None - if result and isinstance(result, bytes): - data = json.loads(result, parse_constant=lambda _: None, object_hook=None, object_pairs_hook=None) - if isinstance(data, dict) and "results" in data: - stored_at = data.get("timestamp", 0) - ttl = data.get("ttl", 14400) - if time.time() - stored_at > ttl: - return None - return data.get("results") - return None - - result = self._run_in_trio(_get, timeout=timeout) - success = result is not None - return result # type: ignore[no-any-return] # noqa: TRY300 - except Exception as ex: # noqa: BLE001 - logger.debug("DHT get failed for %s: %r", key, ex) - success = False - return None - finally: - if success: - self.metrics["successful_queries"] += 1 - self.metrics["query_latency_sum"] += time.time() - start_time - else: - self.metrics["failed_queries"] += 1 - - def set( - self, - key: str, - results: list[dict[str, Any]], - ttl: int = 14400, - *, - timeout: float = 2.0, - ) -> bool: - """Store results in the DHT. - - Args: - key: The cache key (query hash). - results: The search results to cache. - ttl: Time to live in seconds (default 4 hours). - timeout: Timeout in seconds for DHT query. - - Returns: - True if stored successfully. - - """ - if not self._running or not self._dht: - return False - - try: - self.metrics["total_puts"] += 1 - - async def _set() -> bool: - timestamp = time.time() - data = { - "results": results, - "timestamp": timestamp, - "ttl": ttl, - } - value = json.dumps(data, ensure_ascii=False, allow_nan=False).encode() - dht_key = f"/ddgs/{key}" - with trio.move_on_after(timeout) as cancel_scope: - if self._dht is not None: - await self._dht.put_value(dht_key, value) - if cancel_scope.cancelled_caught: - logger.debug("DHT set timed out for %s", key) - return False - - # Track stored keys for periodic refresh - with self._lock: - self._stored_keys[key] = (results, ttl) - return True - - result = self._run_in_trio(_set, timeout=timeout) - if result: - self.metrics["successful_puts"] += 1 - return True - self.metrics["failed_puts"] += 1 - return False # noqa: TRY300 - except Exception as ex: # noqa: BLE001 - logger.debug("DHT set failed for %s: %r", key, ex) - self.metrics["failed_puts"] += 1 - return False - - async def aget(self, key: str, *, timeout: float = 2.0) -> list[dict[str, Any]] | None: # noqa: ASYNC109 - """Async wrapper for get() for backwards compatibility. - - Args: - key: The cache key (query hash). - timeout: Timeout in seconds for DHT query. - - Returns: - List of results or None if not found. - - """ - return await trio.to_thread.run_sync(lambda: self.get(key, timeout=timeout)) - - async def aset( - self, - key: str, - results: list[dict[str, Any]], - ttl: int = 14400, - *, - timeout: float = 2.0, # noqa: ASYNC109 - ) -> bool: - """Async wrapper for set() for backwards compatibility. - - Args: - key: The cache key (query hash). - results: The search results to cache. - ttl: Time to live in seconds (default 4 hours). - timeout: Timeout in seconds for DHT query. - - Returns: - True if stored successfully. - - """ - return await trio.to_thread.run_sync(lambda: self.set(key, results, ttl, timeout=timeout)) - - def find_peers(self) -> list[str]: - """Find other peers in the network. - - Returns: - List of peer addresses. - - """ - if not self._running or not self._host: - return [] - - try: - - async def _find_peers() -> list[str]: - peers = self._host.get_peerstore().peer_ids() # type: ignore[union-attr] - return [str(p) for p in peers] - - result = self._run_in_trio(_find_peers, timeout=5.0) - except Exception as ex: # noqa: BLE001 - logger.debug("DHT find_peers failed: %r", ex) - return [] - else: - return result if result is not None else [] - - def get_neighbors(self) -> list[dict[str, Any]]: - """Get all peers from routing table with complete metadata. - - Returns: - List of peer dictionaries with metadata. - - """ - if not self._dht or not hasattr(self._dht, "routing_table"): - return [] - - peers: list[dict[str, Any]] = [] - peers.extend( - { - "peer_id": str(peer), - "xor_distance": bucket_idx, - "last_seen": peer.last_seen if hasattr(peer, "last_seen") else None, - "latency_ms": peer.latency * 1000 if hasattr(peer, "latency") else None, - "agent_version": peer.agent_version if hasattr(peer, "agent_version") else None, - } - for bucket_idx, bucket in enumerate(self._dht.routing_table.buckets) - for peer in bucket.peers - ) - return peers - - async def afind_peers(self) -> list[str]: - """Async wrapper for find_peers() for backwards compatibility. - - Returns: - List of peer addresses. - - """ - return await trio.to_thread.run_sync(self.find_peers) - - @property - def is_running(self) -> bool: - """Check if the client is running.""" - return self._running - - @property - def port(self) -> int: - """Get the listen port.""" - return self.listen_port - - @property - def peer_id(self) -> str | None: - """Get the peer ID of this node.""" - return str(self._host.get_id()) if self._host else None - - @property - def listen_addrs(self) -> list[str]: - """Get the listen addresses of this node.""" - if self._host: - try: - addrs = [str(addr) for addr in self._host.get_addrs()] - if addrs: - return addrs - except Exception as ex: # noqa: BLE001 - logger.debug("Failed to get host addresses: %r", ex) - - # Fall back to manual address construction using known port - return [f"/ip4/127.0.0.1/tcp/{self.listen_port}"] - - @property - def peer_addrs(self) -> list[str]: - """Get full peer addresses with /p2p/ suffix for connecting.""" - if not self.peer_id: - return [] - return [f"{addr_str}/p2p/{self.peer_id}" for addr_str in self.listen_addrs] - - def stop(self, timeout: float = 2.0) -> None: - """Stop the DHT client. - - Args: - timeout: Maximum time to wait for service to stop. - - """ - with self._lock: - if not self._running: - return - - self._running = False - - # Signal stop to trio thread - avoid deadlock by using direct set() - if self._stop_event is not None: - self._stop_event.set() - - # Wait for thread to exit - if self._dht_thread: - self._dht_thread.join(timeout=timeout) - - self._dht_thread = None - self._trio_token = None - self._stop_event = None - - async def astop(self, timeout: float = 2.0) -> None: # noqa: ASYNC109 - """Async wrapper for stop() for backwards compatibility. - - Args: - timeout: Maximum time to wait for service to stop. - - """ - await trio.to_thread.run_sync(lambda: self.stop(timeout=timeout)) diff --git a/ddgs/dht/types.py b/ddgs/dht/types.py deleted file mode 100644 index 31ffc1c2..00000000 --- a/ddgs/dht/types.py +++ /dev/null @@ -1,80 +0,0 @@ -"""Network types for distributed cache.""" - -import hashlib -import time -from dataclasses import dataclass -from typing import Any - - -@dataclass -class NodeInfo: - """Information about a network node.""" - - peer_id: str - public_key: bytes - address: str - seen_at: float - - @property - def is_alive(self) -> bool: - """Check if the node has been seen in the last 5 minutes.""" - return time.time() - self.seen_at < 300 - - -@dataclass -class CachedResult: - """Cached search result stored in the network.""" - - query_hash: str - query: str - results: list[dict[str, Any]] - timestamp: float - ttl: int = 14400 - - def is_fresh(self) -> bool: - """Check if the cached result is still valid based on TTL.""" - return time.time() - self.timestamp < self.ttl - - @property - def age(self) -> float: - """Get the age of the cached result in seconds.""" - return time.time() - self.timestamp - - -def compute_query_hash(query: str, category: str = "text") -> str: - """Compute a normalized hash for a query. - - Normalization: - - Strip whitespace - - Lowercase - - Sort words alphabetically for deterministic key - - This ensures "Python tutorial" and "tutorial Python" share cache. - - Args: - query: The search query. - category: The search category (text, images, etc.). - - Returns: - A SHA256 hash string. - - """ - words = query.strip().lower().split() - words.sort() - normalized = f"{category}:{' '.join(words)}" - return hashlib.sha256(normalized.encode()).hexdigest() - - -def normalize_query(query: str) -> str: - """Normalize query for display/storage. - - Args: - query: The search query. - - Returns: - Normalized query string. - - """ - words = query.strip().lower().split() - words.sort() - return " ".join(words) diff --git a/pyproject.toml b/pyproject.toml index 3bbcc332..ed0f777d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,13 +81,6 @@ api = [ "fastapi>=0.135.1", "uvicorn[standard]>=0.41.0", ] -dht = [ - "fastapi>=0.135.1", - "uvicorn[standard]>=0.41.0", - "trio>=0.25.0", - #"coincurve @ git+https://github.com/ofek/coincurve.git@7829b29c08ebb1cc80386a1cdaf8c2243c4ef5c5", - #"libp2p @ git+https://github.com/libp2p/py-libp2p.git@0e88584c89377086883c6f5b26cd1a8052399be7", -] [tool.ruff] line-length = 120 @@ -174,17 +167,6 @@ disable_error_code = [ "unused-ignore", ] -# Ignore missing imports for optional DHT dependencies -[[tool.mypy.overrides]] -module = [ - "dns", - "dns.resolver", - "libp2p", - "libp2p.*", - "multiaddr", -] -ignore_missing_imports = true - [tool.pytest.ini_options] asyncio_mode = "auto" pythonpath = [".", "tests"] diff --git a/tests/dht_test.py b/tests/dht_test.py deleted file mode 100644 index 0f933a04..00000000 --- a/tests/dht_test.py +++ /dev/null @@ -1,340 +0,0 @@ -"""Network distributed cache tests using libp2p.""" - -import sys -import tempfile -from pathlib import Path -from unittest import mock - -import pytest -import trio - -# Skip entire test file if DHT dependencies are not installed -dht = pytest.importorskip("ddgs.dht") - -from ddgs.dht import DhtClient, Libp2pClient -from ddgs.dht.types import compute_query_hash - -pytestmark = [ - pytest.mark.trio, - pytest.mark.filterwarnings("ignore::DeprecationWarning"), - pytest.mark.skipif("sys.platform == 'win32'", reason="DHT not supported on Windows"), -] - - -async def test_libp2p_client_start() -> None: - """Test that libp2p client can start.""" - client = Libp2pClient(bootstrap=False) - result = await client.astart() - - assert result is True - assert client.is_running - assert client.peer_id is not None - - await client.astop() - - -@pytest.mark.skip(reason="Known libp2p loopback connection issue - works on real network but not localhost") -async def test_libp2p_two_instances_connect() -> None: - """Test that two libp2p instances can connect to each other. - - Note: This test currently fails on localhost due to a known issue in py-libp2p, - but works correctly on real networks with separate machines. - """ - client1 = Libp2pClient(listen_port=0, bootstrap=False) - client2 = Libp2pClient(listen_port=0, bootstrap=False) - - await client1.astart() - await client2.astart() - - assert client1.peer_id is not None - assert client2.peer_id is not None - assert client1.peer_id != client2.peer_id - - # Give time for socket to be fully bound and listening - await trio.sleep(1) - - # Try all addresses with multiple retries - success = False - for addr in client1.peer_addrs: - for _ in range(5): - success = await client2.aconnect_peer(addr) - if success: - break - await trio.sleep(0.2) - if success: - break - - assert success is True, f"Failed to connect to any address: {client1.peer_addrs}" - - # Critical: Wait for routing tables to update and refresh - await trio.sleep(2) - - peers1 = await client1.afind_peers() - peers2 = await client2.afind_peers() - - assert len(peers1) >= 1 - assert len(peers2) >= 1 - - await client1.astop() - await client2.astop() - - -@pytest.mark.skip(reason="Known libp2p loopback connection issue - works on real network but not localhost") -async def test_dht_put_and_get_between_instances() -> None: - """Test DHT put/get between two libp2p instances. - - Important note: This test has realistic propagation delays. - Kademlia requires time for routing table updates and value replication. - - Note: This test currently fails on localhost due to a known issue in py-libp2p, - but works correctly on real networks with separate machines. - """ - client1 = Libp2pClient(listen_port=0, bootstrap=False) - client2 = Libp2pClient(listen_port=0, bootstrap=False) - - await client1.astart() - await client2.astart() - - # Connect explicitly - success = await client2.aconnect_peer(client1.peer_addrs[0]) - assert success is True - - # Wait for routing tables to populate (this is required - cannot be skipped) - await trio.sleep(3) - - query = "python tutorial" - category = "text" - key = compute_query_hash(query, category) - results = [{"title": "Python Tutorial", "url": "https://example.com/python"}] - - set_result = await client1.aset(key, results, ttl=60) - assert set_result is True - - # Kademlia values replicate asynchronously. Wait for propagation. - await trio.sleep(5) - - # Try multiple times with backoff - realistic real world behavior - retrieved = None - for attempt in range(3): - retrieved = await client2.aget(key, timeout=2) - if retrieved is not None: - break - await trio.sleep(2) - - assert retrieved is not None, "Failed to retrieve from DHT after 3 attempts" - assert len(retrieved) == 1 - assert retrieved[0]["title"] == "Python Tutorial" - - await client1.astop() - await client2.astop() - - -async def test_network_client_local_cache() -> None: - """Test DhtClient local cache (no DHT).""" - client = DhtClient(enable_dht=False) - await client.start() - - await client.cache("test query", [{"title": "Test"}], "text") - result = await client.get_cached("test query", "text") - - assert result is not None - assert len(result) == 1 - - await client.stop() - - -async def test_cache_ttl() -> None: - """Test cache TTL expiration.""" - client = DhtClient(enable_dht=False, cache_ttl=1) - await client.start() - - await client.cache("ttl test", [{"title": "Old"}], "text") - await trio.sleep(1.1) - - result = await client.get_cached("ttl test", "text") - assert result is None - - await client.stop() - - -async def test_dht_query_timeout() -> None: - """Test DHT query timeout functionality.""" - from ddgs.dht.libp2p_client import Libp2pClient - - client = Libp2pClient(bootstrap=False) - await client.astart() - - # This should timeout quickly instead of hanging - result = await client.aget("non-existent-key", timeout=0.1) - assert result is None - - # Set should also work with timeout - success = await client.aset("test-key", [{"title": "test"}], timeout=0.1) - assert isinstance(success, bool) - - await client.astop() - - -async def test_local_cache_eviction() -> None: - """Test local cache eviction when max size is reached.""" - from ddgs.dht.cache import MAX_CACHE_SIZE, ResultCache - - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - f.close() # Critical: close file handle before SQLite opens it - try: - cache = ResultCache(db_path=f.name) - - # Fill cache beyond max size - for i in range(MAX_CACHE_SIZE + 100): - cache.set(f"key-{i}", f"query-{i}", "text", [{"title": f"Result {i}"}]) - - # Should have evicted old entries - count = cache.count() - assert count == MAX_CACHE_SIZE - finally: - Path(f.name).unlink(missing_ok=True) - - -async def test_negative_cache() -> None: - """Test negative cache functionality.""" - from ddgs.dht.cache import ResultCache - - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - f.close() # Critical: close file handle before SQLite opens it - try: - cache = ResultCache(db_path=f.name) - - # Add to negative cache - cache.add_negative("test-key") - - # Should return None immediately (negative cache hit) - result = cache.get("test-key") - assert result is None - finally: - Path(f.name).unlink(missing_ok=True) - - -async def test_network_client_invalidate() -> None: - """Test DhtClient cache invalidation.""" - client = DhtClient(enable_dht=False) - await client.start() - - await client.cache("test query", [{"title": "Test"}], "text") - - # Verify exists - result = await client.get_cached("test query", "text") - assert result is not None - - # Invalidate - await client.invalidate("test query", "text") - - # Verify removed - result = await client.get_cached("test query", "text") - assert result is None - - await client.stop() - - -async def test_peer_connection_error_handling() -> None: - """Test that connection failures are handled gracefully.""" - from ddgs.dht.libp2p_client import Libp2pClient - - client = Libp2pClient(bootstrap=False) - await client.astart() - - # Try to connect to invalid peer - success = await client.aconnect_peer("/ip4/127.0.0.1/tcp/9999/p2p/invalidpeerid") - assert success is False - - await client.astop() - - -async def test_bootstrap_network_connectivity() -> None: - """Test that client can connect to public bootstrap network.""" - import logging - - logging.basicConfig(level=logging.WARNING) - - client = Libp2pClient(bootstrap=True) - try: - await client.astart() - - # Give time for bootstrap connection - await trio.sleep(3) - - peer_count = len(await client.afind_peers()) - - # Should have at least a few peers if connected successfully - assert peer_count >= 0, "Peer discovery should work" - - finally: - await client.astop() - - -@pytest.mark.slow -@pytest.mark.integration -@pytest.mark.skip(reason="Kademlia requires active peers to replicate values. Test will pass automatically once network has 10+ nodes.") -@pytest.mark.slow -@pytest.mark.integration -async def test_bootstrap_value_replication() -> None: - """Test that value stored on one client propagates to another via bootstrap network. - - IMPORTANT: This test is expected to fail right now. Kademlia does not replicate - values through bootstrap nodes. It requires existing active peers on the network - to store and forward values. This test will start passing automatically once - there are ~10+ permanent ddgs nodes running. - - This is an integration test that uses the real public IPFS bootstrap network. - """ - import logging - import time - logging.basicConfig(level=logging.WARNING) - - client1 = Libp2pClient(bootstrap=True) - client2 = Libp2pClient(bootstrap=True) - - try: - await client1.astart() - await client2.astart() - - await trio.sleep(15) - - query = "bootstrap integration test" - category = "text" - key = compute_query_hash(query, category) - test_value = [{"title": "Bootstrap Test", "url": "https://github.com/xxx/xxx"}] - - await client1.aset(key, test_value, ttl=120) - - retrieved = None - for _ in range(3): - retrieved = await client2.aget(key, timeout=3) - if retrieved is not None: - break - await trio.sleep(5) - - assert retrieved == test_value - - finally: - await client1.astop() - await client2.astop() - - -async def test_bootstrap_local_only_fallback() -> None: - """Test that client works correctly even when bootstrap is unreachable.""" - client = Libp2pClient(bootstrap=False) - try: - await client.astart(timeout=10) - - # Should still work even with no bootstrap connectivity - key = "local-test-key" - test_value = [{"title": "Local Test"}] - - success = await client.aset(key, test_value) - assert success is True - - retrieved = await client.aget(key) - assert retrieved == test_value - - finally: - await client.astop()