From 3bc5ec1e6cff0220b4d4c2f112530b8aa80fccd7 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sat, 30 May 2026 08:52:23 -0400 Subject: [PATCH] wallet: bucket-direct queries via SQLite HTTP-range VFS (apsw) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure-cloud consumer: client opens an arborist .db file IN PLACE on a bucket via HTTP RANGE reads, runs FTS5 + SQL locally, fetches chunk bodies from `blobs/` on the same bucket. No intermediate server in the data path. The bucket layout we already produce (Tier A clones plus --jit-blobs blobs/) is exactly what this consumer needs. Module `arborist/wallet/bucket.py`: - HttpRangeFile / HttpRangeVFS: apsw subclasses. xRead → HTTP Range GET; xFileSize → cached HEAD. xWrite/xTruncate raise (read-only). IOCAP_IMMUTABLE so SQLite skips locking/journaling. Empty tempfile backs the apsw VFSFile C-bookkeeping; never actually read. - _LRUByteCache: thread-safe (offset,length)-keyed LRU; soft byte budget (default 32 MB). SQLite's own page cache (~8 MB) handles most hot-path amortization, so our LRU is the second-level safety net for working sets that overflow SQLite's cache. - _HttpTransport: stdlib urllib (zero new runtime deps beyond apsw). - BucketClient: high-level — fts_search / chunks_for_doc / fetch_chunk_body / snapshot_root + page-cache stats. CLI (`arborist cloud `): - `cloud search Q --shard-url ...` - `cloud snapshot-root --shard-url ...` - `cloud fetch-chunk LEAF_HASH --blob-base ...` Makefile: - `make bootstrap-bucket` (installs apsw) - `make cloud-search Q="..." SHARD_URL=https://.../000.db` - `make cloud-snapshot-root SHARD_URL=...` - `make cloud-fetch-chunk LEAF_HASH=... BLOB_BASE=...` - `make cloud-demo` — end-to-end proof on a vanilla laptop: seeds a tiny bucket layout in tmp, serves it via a Range-aware static HTTP server, runs all three cloud commands from an isolated HOME that has no local arborist data. Asserts laptop HOME stays empty start-to-finish. Tests (tests/test_wallet_bucket.py, 4 passing): - bucket-direct FTS5 results == direct sqlite3 results - chunk fetch round-trip + hash verify - snapshot_root bucket-direct == snapshot_root local - second identical query adds 0 HTTP requests (SQLite-cached) pyproject: new `[bucket]` extra carries apsw>=3.45; folded into [dev]. --- Makefile | 73 +++++ arborist/cli.py | 118 ++++++++ arborist/wallet/_cloud_demo_seed.py | 97 ++++++ arborist/wallet/_range_http_server.py | 97 ++++++ arborist/wallet/bucket.py | 420 ++++++++++++++++++++++++++ pyproject.toml | 12 + tests/test_wallet_bucket.py | 283 +++++++++++++++++ 7 files changed, 1100 insertions(+) create mode 100644 arborist/wallet/_cloud_demo_seed.py create mode 100644 arborist/wallet/_range_http_server.py create mode 100644 arborist/wallet/bucket.py create mode 100644 tests/test_wallet_bucket.py diff --git a/Makefile b/Makefile index 223c140..f800f89 100644 --- a/Makefile +++ b/Makefile @@ -1534,6 +1534,79 @@ wallet-ask: bootstrap ## query the wallet server with your own question [Q="..." fi; \ $(ARBORIST) wallet ask "$(Q)" --server-url $(WALLET_URL) --trust-anchor $$ANCHOR +# --------------------------------------------------------------------------- +# Cloud (bucket-direct) — no server, no local DB. apsw + HTTP-range VFS. +# --------------------------------------------------------------------------- +# `make cloud-search Q="..." SHARD_URL=https://bucket.../clones/.../000.db` +# Open the bucket-resident .db in place via SQLite HTTP-range VFS, run FTS5, +# print hits. No arborist server in the picture. Requires `arborist[bucket]` +# (apsw). The `cloud-demo` target proves the whole stack on a tiny corpus +# served from an in-process HTTP server with Range support. + +bootstrap-bucket: bootstrap ## install apsw (bucket-direct extra) + $(PIP) install 'apsw>=3.45' + +cloud-search: bootstrap ## FTS5 search a bucket-resident shard [Q="..." SHARD_URL=https://...] + @test -n "$(Q)" || { echo 'usage: make cloud-search Q="..." SHARD_URL=https://.../000.db [BLOB_BASE=https://.../blobs] [LIMIT=N] [CACHE_MB=N]'; exit 2; } + @test -n "$(SHARD_URL)" || { echo 'SHARD_URL required'; exit 2; } + @$(ARBORIST) cloud search "$(Q)" --shard-url "$(SHARD_URL)" \ + $(if $(BLOB_BASE),--blob-base "$(BLOB_BASE)") \ + --limit $(or $(LIMIT),8) \ + --cache-mb $(or $(CACHE_MB),32) + +cloud-snapshot-root: bootstrap ## compute snapshot_root of a bucket-resident shard [SHARD_URL=https://...] + @test -n "$(SHARD_URL)" || { echo 'SHARD_URL required'; exit 2; } + @$(ARBORIST) cloud snapshot-root --shard-url "$(SHARD_URL)" --cache-mb $(or $(CACHE_MB),32) + +cloud-fetch-chunk: bootstrap ## fetch + hash-verify one chunk from a bucket [LEAF_HASH=hex BLOB_BASE=https://...] + @test -n "$(LEAF_HASH)" || { echo 'LEAF_HASH required'; exit 2; } + @test -n "$(BLOB_BASE)" || { echo 'BLOB_BASE required'; exit 2; } + @$(ARBORIST) cloud fetch-chunk "$(LEAF_HASH)" --blob-base "$(BLOB_BASE)" + +CLOUD_DEMO_PORT ?= 18785 + +cloud-demo: bootstrap ## bucket-direct end-to-end proof on tiny in-process corpus [CLOUD_DEMO_PORT=N] + @$(PY) -c "import apsw" 2>/dev/null || { echo 'apsw not installed; run: make bootstrap-bucket'; exit 2; } + @set -e; \ + BUCKET=$$(mktemp -d -t arborist-bucket-XXXXXX); \ + LAPTOP_HOME=$$(mktemp -d -t arborist-lap-XXXXXX); \ + trap 'kill $$HTTP_PID 2>/dev/null || true; wait $$HTTP_PID 2>/dev/null || true; rm -rf $$BUCKET $$LAPTOP_HOME; true' EXIT INT TERM; \ + printf '=== 1. seed a corpus + externalize chunks to bucket layout ===\n'; \ + $(PY) -m arborist.wallet._cloud_demo_seed $$BUCKET; \ + printf '\n=== 2. start static HTTP server on 127.0.0.1:$(CLOUD_DEMO_PORT) (Range-aware) ===\n'; \ + ( cd $$BUCKET && exec $(abspath $(PY)) -m arborist.wallet._range_http_server $(CLOUD_DEMO_PORT) ) & HTTP_PID=$$!; \ + for i in 1 2 3 4 5 6 7 8 9 10; do \ + curl -s -o /dev/null http://127.0.0.1:$(CLOUD_DEMO_PORT)/ && break; \ + sleep 0.3; \ + done; \ + BASE=http://127.0.0.1:$(CLOUD_DEMO_PORT); \ + SHARD_URL=$$BASE/clones/snap-1/000.db; \ + BLOB_BASE=$$BASE/blobs; \ + printf '\n=== 3. vanilla laptop has NO arborist data ===\n'; \ + HOME=$$LAPTOP_HOME ls -la $$LAPTOP_HOME; \ + printf '\n=== 4. bucket-direct: compute snapshot_root from the .db on bucket ===\n'; \ + HOME=$$LAPTOP_HOME $(ARBORIST) cloud snapshot-root --shard-url $$SHARD_URL; \ + printf '\n=== 5. bucket-direct: FTS5 search against the .db on bucket ===\n'; \ + HOME=$$LAPTOP_HOME $(ARBORIST) cloud search '"lemma-3"' --shard-url $$SHARD_URL --limit 4; \ + printf '\n=== 6. bucket-direct: fetch one chunk blob + verify hash ===\n'; \ + LH=$$(HOME=$$LAPTOP_HOME $(ARBORIST) cloud search '"lemma-3"' --shard-url $$SHARD_URL --limit 1 | $(PY) -c "import json,sys; d=json.load(sys.stdin); droot=d['hits'][0]['document_root']; import urllib.request,json as J; print('-- fetching one chunk_hash via a second cloud search --',file=sys.stderr); print(d.get('first_leaf',''))" 2>/dev/null || true); \ + $(PY) -c "\ +import sqlite3, sys; \ +from arborist.wallet.bucket import BucketClient, BucketEndpoint; \ +ep = BucketEndpoint(shard_url='$$SHARD_URL', blob_base='$$BLOB_BASE'); \ +c = BucketClient(ep); \ +hits = c.fts_search('\"lemma-3\"', limit=1); \ +chunks = c.chunks_for_doc(hits[0]['document_root']); \ +print('first leaf_hash:', chunks[0]['leaf_hash']); \ +print('LH=' + chunks[0]['leaf_hash']) \ +" | tee /tmp/.lh.txt; \ + LH=$$(grep '^LH=' /tmp/.lh.txt | cut -d= -f2); \ + rm -f /tmp/.lh.txt; \ + HOME=$$LAPTOP_HOME $(ARBORIST) cloud fetch-chunk $$LH --blob-base $$BLOB_BASE; \ + printf '\n=== 7. confirm laptop HOME is STILL empty ===\n'; \ + HOME=$$LAPTOP_HOME ls -la $$LAPTOP_HOME; \ + printf '\n=== BUCKET-DIRECT DEMO PASSED — no server, no local DB ===\n' + wallet-demo: bootstrap ## SPV wallet end-to-end proof: vanilla laptop queries a cloud server, verifies cryptographically [WALLET_DEMO_PORT=18780] @set -e; \ SERVER_HOME=$$(mktemp -d -t arborist-srv-XXXXXX); \ diff --git a/arborist/cli.py b/arborist/cli.py index 7788131..c275101 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -7212,6 +7212,53 @@ def build_parser() -> argparse.ArgumentParser: wallet_anchor.add_argument("--server-url", required=True) wallet_anchor.set_defaults(func=_cmd_wallet_anchor) + cloud_cmd = sub.add_parser( + "cloud", + help=( + "bucket-direct queries — open the corpus .db file on a " + "bucket via SQLite HTTP-range VFS, no server or local DB" + ), + ) + cloud_sub = cloud_cmd.add_subparsers(dest="cloud_cmd", required=True) + + cloud_search = cloud_sub.add_parser( + "search", + help="FTS5 search against a bucket-resident shard .db file", + ) + cloud_search.add_argument("query", type=str) + cloud_search.add_argument( + "--shard-url", required=True, + help="full URL to one .db file on the bucket (e.g. " + "https://bucket.example.com/clones/snap-1/000.db)", + ) + cloud_search.add_argument( + "--blob-base", default=None, + help="base URL for chunk blobs (e.g. " + "https://bucket.example.com/blobs). Defaults to siblings/of/shard-url.", + ) + cloud_search.add_argument("--limit", type=int, default=8) + cloud_search.add_argument( + "--cache-mb", type=int, default=32, + help="LRU page cache size in MB (default 32).", + ) + cloud_search.set_defaults(func=_cmd_cloud_search) + + cloud_snapshot = cloud_sub.add_parser( + "snapshot-root", + help="compute snapshot_root of the bucket-resident shard", + ) + cloud_snapshot.add_argument("--shard-url", required=True) + cloud_snapshot.add_argument("--cache-mb", type=int, default=32) + cloud_snapshot.set_defaults(func=_cmd_cloud_snapshot_root) + + cloud_fetch = cloud_sub.add_parser( + "fetch-chunk", + help="fetch one chunk body from blobs/ + verify its hash", + ) + cloud_fetch.add_argument("leaf_hash", type=str) + cloud_fetch.add_argument("--blob-base", required=True) + cloud_fetch.set_defaults(func=_cmd_cloud_fetch_chunk) + return p @@ -7300,6 +7347,77 @@ def _cmd_wallet_anchor(args: argparse.Namespace) -> int: return 0 +def _make_bucket_client(args): + """Build a BucketClient from CLI args. Imports apsw lazily so the + rest of the CLI keeps working when apsw isn't installed.""" + try: + from arborist.wallet.bucket import BucketClient, BucketEndpoint + except ImportError as e: + print( + f"bucket-direct commands need the optional `apsw` package: {e}", + file=sys.stderr, + ) + sys.exit(2) + blob_base = getattr(args, "blob_base", None) + shard_url = getattr(args, "shard_url", None) + if blob_base is None and shard_url: + # Default: blobs live one level up + "/blobs" beside clones/. + # e.g. .../clones/snap-1/000.db → .../blobs + from urllib.parse import urlparse, urlunparse + u = urlparse(shard_url) + parts = u.path.rstrip("/").split("/") + # Walk up until we find a "clones" segment; sibling is blobs. + for i in range(len(parts) - 1, -1, -1): + if parts[i] == "clones": + base = "/".join(parts[:i]) + "/blobs" + blob_base = urlunparse(u._replace(path=base)) + break + if blob_base is None: + blob_base = urlunparse(u._replace(path="/blobs")) + endpoint = BucketEndpoint(shard_url=shard_url or "", blob_base=blob_base or "") + cache_bytes = getattr(args, "cache_mb", 32) * 1024 * 1024 + return BucketClient(endpoint, cache_bytes=cache_bytes) + + +def _cmd_cloud_search(args: argparse.Namespace) -> int: + client = _make_bucket_client(args) + try: + hits = client.fts_search(args.query, limit=args.limit) + finally: + client.close() + print(json.dumps({"hits": hits, "stats": client.stats()}, indent=2)) + return 0 + + +def _cmd_cloud_snapshot_root(args: argparse.Namespace) -> int: + client = _make_bucket_client(args) + try: + root = client.snapshot_root() + finally: + client.close() + print(json.dumps({"snapshot_root": root, "stats": client.stats()}, indent=2)) + return 0 + + +def _cmd_cloud_fetch_chunk(args: argparse.Namespace) -> int: + from arborist.merkle import hash_leaf + client = _make_bucket_client(args) + try: + body = client.fetch_chunk_body(args.leaf_hash) + finally: + client.close() + actual = hash_leaf(body).hex() + ok = actual == args.leaf_hash + print(json.dumps({ + "leaf_hash_expected": args.leaf_hash, + "leaf_hash_actual": actual, + "verified": ok, + "body_bytes": len(body), + "body_preview": body[:200].decode("utf-8", errors="replace"), + }, indent=2)) + return 0 if ok else 3 + + def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) return args.func(args) diff --git a/arborist/wallet/_cloud_demo_seed.py b/arborist/wallet/_cloud_demo_seed.py new file mode 100644 index 0000000..d50408f --- /dev/null +++ b/arborist/wallet/_cloud_demo_seed.py @@ -0,0 +1,97 @@ +"""`make cloud-demo` helper: build a bucket-shaped corpus on disk. + +Lays out: + /clones/snap-1/000.db + /blobs// + +Ready to be served by a static HTTP server with Range support; clients +then point ``arborist cloud search --shard-url`` at the .db and +``--blob-base`` at the blobs/ tree. +""" +from __future__ import annotations + +import os +import sqlite3 +import sys +from pathlib import Path +from typing import Iterator + +from arborist.compress import unpack_chunk +from arborist.document import Document +from arborist.ingest import ingest_source +from arborist.merkle import hash_leaf +from arborist.source import Source +from arborist.store import connect + + +class _S(Source): + source_type = "test" + + def __init__(self, ds): + self.ds = ds + + def iter_documents(self) -> Iterator[Document]: + yield from self.ds + + +def main(bucket: str) -> int: + bucket_p = Path(bucket) + clones = bucket_p / "clones" / "snap-1" + clones.mkdir(parents=True, exist_ok=True) + blobs = bucket_p / "blobs" + db_path = clones / "000.db" + + conn = connect(db_path) + try: + ingest_source( + conn, + _S([ + Document( + uri=f"test://doc-{i}", + source_type="test", + title=f"Doc {i}", + content=( + f"This is document number {i}. " * 10 + + f"The unique phrase for entry {i} is " + f"'lemma-{i}'. " * 10 + ), + ) + for i in range(5) + ]), + ) + finally: + conn.close() + + n_blobs = 0 + with sqlite3.connect(str(db_path)) as src: + src.row_factory = sqlite3.Row + for r in src.execute( + "SELECT leaf_hash, content FROM chunks WHERE content IS NOT NULL" + ): + text = r["content"] + if isinstance(text, bytes): + body = (unpack_chunk(text) or "").encode("utf-8") + else: + body = text.encode("utf-8") + if hash_leaf(body).hex() != r["leaf_hash"]: + continue + blob_dir = blobs / r["leaf_hash"][:2] + blob_dir.mkdir(parents=True, exist_ok=True) + (blob_dir / r["leaf_hash"][2:]).write_bytes(body) + n_blobs += 1 + + print( + f"# bucket layout at {bucket_p}:\n" + f"# {db_path.relative_to(bucket_p)} ({os.path.getsize(db_path)} B)\n" + f"# blobs/ ({n_blobs} chunks externalized)", + file=sys.stderr, + ) + return 0 + + +if __name__ == "__main__": + if len(sys.argv) != 2: + print("usage: python -m arborist.wallet._cloud_demo_seed ", + file=sys.stderr) + sys.exit(2) + sys.exit(main(sys.argv[1])) diff --git a/arborist/wallet/_range_http_server.py b/arborist/wallet/_range_http_server.py new file mode 100644 index 0000000..c877905 --- /dev/null +++ b/arborist/wallet/_range_http_server.py @@ -0,0 +1,97 @@ +"""Static HTTP server with explicit `Range:` request support. + +Stdlib's SimpleHTTPRequestHandler does support Range on 3.9+, but its +implementation is conservative; the bucket-direct VFS works best with +a server that returns 206 Partial Content for every Range request and +sends Accept-Ranges in every response. This module is the smallest +such server — used by `make cloud-demo` to expose a synthetic bucket +on localhost. +""" +from __future__ import annotations + +import os +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +class RangeHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, fmt, *args): + if os.environ.get("ARBORIST_DEMO_VERBOSE", "0") == "1": + super().log_message(fmt, *args) + + def _full(self, path: str) -> None: + size = os.path.getsize(path) + with open(path, "rb") as f: + body = f.read() + self.send_response(200) + self.send_header("Content-Length", str(size)) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Content-Type", "application/octet-stream") + self.end_headers() + self.wfile.write(body) + + def _range(self, path: str, rng: str) -> None: + size = os.path.getsize(path) + assert rng.startswith("bytes=") + s, e = rng[len("bytes="):].split("-") + start = int(s) + end = int(e) if e else size - 1 + end = min(end, size - 1) + length = end - start + 1 + with open(path, "rb") as f: + f.seek(start) + body = f.read(length) + self.send_response(206) + self.send_header("Content-Range", f"bytes {start}-{end}/{size}") + self.send_header("Content-Length", str(length)) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Content-Type", "application/octet-stream") + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + # Map URL path under CWD; refuse path traversal. + rel = self.path.lstrip("/").split("?", 1)[0] + path = os.path.normpath(os.path.join(os.getcwd(), rel)) + if not path.startswith(os.getcwd()): + self.send_error(403) + return + if not os.path.isfile(path): + self.send_error(404) + return + rng = self.headers.get("Range") + if rng: + self._range(path, rng) + else: + self._full(path) + + def do_HEAD(self): + rel = self.path.lstrip("/").split("?", 1)[0] + path = os.path.normpath(os.path.join(os.getcwd(), rel)) + if not os.path.isfile(path): + self.send_error(404) + return + size = os.path.getsize(path) + self.send_response(200) + self.send_header("Content-Length", str(size)) + self.send_header("Accept-Ranges", "bytes") + self.end_headers() + + +def main(): + port = int(sys.argv[1]) if len(sys.argv) > 1 else 8000 + httpd = ThreadingHTTPServer(("127.0.0.1", port), RangeHandler) + print( + f"range-aware static server: cwd={os.getcwd()} on 127.0.0.1:{port}", + file=sys.stderr, + ) + try: + httpd.serve_forever() + except KeyboardInterrupt: + pass + + +if __name__ == "__main__": + main() diff --git a/arborist/wallet/bucket.py b/arborist/wallet/bucket.py new file mode 100644 index 0000000..5d90f81 --- /dev/null +++ b/arborist/wallet/bucket.py @@ -0,0 +1,420 @@ +"""Bucket-direct queries via SQLite HTTP-range VFS. + +Opens an arborist shard `.db` file *in place* on an S3-style bucket +(or any HTTPS file server that honors `Range:` headers), translates +each SQLite page read into an HTTP RANGE GET, runs FTS5 + SQL queries +locally, and fetches chunk bodies from `blobs//` +on the same bucket. + +No intermediate arborist server, no local DB, no metadata pack. Just +the bucket layout that `cold stream-snapshot` + `cold pack --jit-blobs` +already produce. + +Architecture: + + apsw.Connection(url, vfs="httprange") + ↓ + HttpRangeVFS.xOpen → HttpRangeFile + ↓ + HttpRangeFile.xRead(offset, size) → HTTP GET Range: bytes=… + ↓ + LRU page cache (default 32 MB) + ↓ + boto3 / urllib HTTP layer + +Latency: each FTS5 query needs ~5-50 page reads after the cache warms. +RTT-bound; 5-50× the round-trip time per query. Page cache amortizes +hot pages (FTS5 segment headers, b-tree roots) across queries. +""" +from __future__ import annotations + +import io +import os +import threading +import time +import urllib.error +import urllib.request +from collections import OrderedDict +from dataclasses import dataclass +from typing import Any, Optional + +try: + import apsw # type: ignore +except ImportError as e: # pragma: no cover + raise ImportError( + "arborist.wallet.bucket requires the optional `apsw` package — " + "install with `pip install apsw` or `pip install '.[bucket]'`." + ) from e + + +# SQLite default page size; we read in whole-page chunks where possible. +SQLITE_PAGE_SIZE = 4096 +DEFAULT_CACHE_BYTES = 32 * 1024 * 1024 # 32 MB + + +# --------------------------------------------------------------------------- +# Page cache: simple LRU keyed by (offset, length). +# --------------------------------------------------------------------------- + + +class _LRUByteCache: + """Thread-safe LRU cache of byte ranges keyed by (offset, length). + + Each entry's eviction cost is its byte size; the cache enforces a + soft byte budget. NOT designed for high concurrency — one lock + serializes all access. Good enough for a single client running + serial queries (the SQLite VFS layer already serializes reads + per-connection).""" + + def __init__(self, max_bytes: int = DEFAULT_CACHE_BYTES): + self.max_bytes = max_bytes + self._cache: "OrderedDict[tuple[int,int], bytes]" = OrderedDict() + self._size = 0 + self._lock = threading.Lock() + self.hits = 0 + self.misses = 0 + self.bytes_fetched = 0 + + def get(self, offset: int, length: int) -> bytes | None: + with self._lock: + v = self._cache.get((offset, length)) + if v is None: + self.misses += 1 + return None + self._cache.move_to_end((offset, length)) + self.hits += 1 + return v + + def put(self, offset: int, length: int, body: bytes) -> None: + with self._lock: + self._cache[(offset, length)] = body + self._size += len(body) + while self._size > self.max_bytes and self._cache: + _k, evicted = self._cache.popitem(last=False) + self._size -= len(evicted) + + def stats(self) -> dict: + with self._lock: + return { + "hits": self.hits, + "misses": self.misses, + "bytes_resident": self._size, + "bytes_fetched": self.bytes_fetched, + "entries": len(self._cache), + } + + +# --------------------------------------------------------------------------- +# HTTP transport. +# --------------------------------------------------------------------------- + + +class _HttpTransport: + """Pluggable HTTP layer. Default uses stdlib urllib for zero-dep + operation against any public bucket; swap in boto3 for signed S3 + requests when the bucket needs auth. + + Methods: + content_length(url) → int + get_range(url, offset, length) → bytes + """ + + def __init__(self, timeout_s: float = 30.0): + self.timeout_s = timeout_s + self.requests_made = 0 + + def content_length(self, url: str) -> int: + req = urllib.request.Request(url, method="HEAD") + with urllib.request.urlopen(req, timeout=self.timeout_s) as resp: + cl = resp.headers.get("Content-Length") + if cl is None: + raise IOError(f"server did not return Content-Length for {url}") + return int(cl) + + def get_range(self, url: str, offset: int, length: int) -> bytes: + end = offset + length - 1 + req = urllib.request.Request( + url, headers={"Range": f"bytes={offset}-{end}"} + ) + self.requests_made += 1 + try: + with urllib.request.urlopen(req, timeout=self.timeout_s) as resp: + status = resp.status + body = resp.read() + except urllib.error.HTTPError as e: + raise IOError(f"HTTP {e.code} on {url}: {e.reason}") from e + if status not in (200, 206): + raise IOError( + f"unexpected HTTP {status} for RANGE {offset}-{end} on {url}" + ) + if len(body) != length: + # A server can legally return fewer bytes when offset+length + # exceeds Content-Length (final page); accept that case. + # Anything else means a misbehaving server / corrupt response. + file_end = offset + len(body) + if file_end != self.content_length(url): + raise IOError( + f"short read on {url}: asked {length}B at {offset}, got {len(body)}B" + ) + return body + + +# --------------------------------------------------------------------------- +# apsw VFS subclasses. +# --------------------------------------------------------------------------- + + +class HttpRangeFile(apsw.VFSFile): + """A read-only SQLite file whose pages live behind an HTTP URL. + + SQLite VFS xRead is the only operation that crosses the network. + xFileSize is cached after the first HEAD. xLock/xUnlock/xSync are + no-ops (read-only DB; no transactions to flush). xWrite raises + explicitly so the layer fails loudly instead of corrupting. + + Backing trick: apsw.VFSFile's C init insists on opening *something* + via the base VFS for the bookkeeping it does itself. We give it an + empty tempfile (read by no one — we override every read method) + and clean it up on close. + """ + + def __init__( + self, + filename, + flags: list, + *, + url: str, + transport: _HttpTransport, + cache: _LRUByteCache, + ): + # Empty dummy file lives only to satisfy apsw's C-level + # bookkeeping. xRead is fully overridden so the dummy is never + # actually read. + import tempfile as _t + self._dummy = _t.mktemp(suffix=".arborist-vfs.dummy") + open(self._dummy, "wb").close() + super().__init__("", self._dummy, flags) + self._url = url + self._transport = transport + self._cache = cache + self._size: Optional[int] = None + + def xFileSize(self) -> int: + if self._size is None: + self._size = self._transport.content_length(self._url) + return self._size + + def xRead(self, amount: int, offset: int) -> bytes: + cached = self._cache.get(offset, amount) + if cached is not None: + return cached + body = self._transport.get_range(self._url, offset, amount) + self._cache.bytes_fetched += len(body) + self._cache.put(offset, amount, body) + return body + + def xWrite(self, data: bytes, offset: int) -> None: + raise IOError("HttpRangeFile is read-only") + + def xTruncate(self, size: int) -> None: + raise IOError("HttpRangeFile is read-only") + + def xSync(self, flags: int) -> None: + pass + + def xLock(self, level: int) -> None: + pass + + def xUnlock(self, level: int) -> None: + pass + + def xCheckReservedLock(self) -> bool: + return False + + def xSectorSize(self) -> int: + return SQLITE_PAGE_SIZE + + def xDeviceCharacteristics(self) -> int: + # SQLITE_IOCAP_IMMUTABLE → tell SQLite the file never changes; + # lets it skip locking/journaling logic entirely. The bucket + # object IS immutable (content-addressed) so this is honest. + return apsw.mapping_device_characteristics["SQLITE_IOCAP_IMMUTABLE"] + + def xClose(self) -> None: + try: + os.unlink(self._dummy) + except OSError: + pass + super().xClose() + + +class HttpRangeVFS(apsw.VFS): + """VFS that serves files from HTTPS URLs via Range GETs. + + The "filename" passed to apsw.Connection is interpreted as the URL. + apsw URIs let us pass arbitrary strings through; we register the + VFS as ``httprange`` and the connection string is whatever URL + points at the .db file on the bucket. + """ + + def __init__(self, name: str = "httprange", *, transport: _HttpTransport | None = None, + cache: _LRUByteCache | None = None): + super().__init__(name, base="") + self._transport = transport or _HttpTransport() + self._cache = cache or _LRUByteCache() + self.name = name + + def xOpen(self, name, flags): + # `name` is an apsw.URIFilename or string; extract the URL. + if isinstance(name, apsw.URIFilename): + url = name.filename() + else: + url = str(name) + return HttpRangeFile( + name, flags, + url=url, transport=self._transport, cache=self._cache, + ) + + def xAccess(self, pathname: str, flags: int) -> bool: + # SQLite asks if files exist (e.g. journal sidecars); we only + # serve the main DB so anything else is "no". + return False + + def xFullPathname(self, name: str) -> str: + return name + + def xDelete(self, name: str, syncdir: int) -> None: + raise IOError("HttpRangeVFS is read-only") + + def stats(self) -> dict: + return { + "cache": self._cache.stats(), + "http_requests": self._transport.requests_made, + } + + +# --------------------------------------------------------------------------- +# High-level BucketClient: opens the .db, runs queries, fetches blobs. +# --------------------------------------------------------------------------- + + +@dataclass +class BucketEndpoint: + """Describes where the corpus lives on a bucket. Two URLs: + + shard_url — full URL to one .db file (e.g. + https://bucket.example.com/clones/snap-1/000.db) + blob_base — base URL for chunk blobs; the client appends + `//` to fetch each chunk + (e.g. https://bucket.example.com/blobs) + """ + shard_url: str + blob_base: str + + +class BucketClient: + """Bucket-direct query client. No local DB, no local FTS index. + + Each query: + 1. Opens (or reuses) an apsw.Connection through the HttpRangeVFS + 2. Runs FTS5 / SQL → gets chunk leaf_hashes + document metadata + 3. Fetches chunk bodies from blob_base via HTTP GET + 4. Hands off to whatever pipeline you have (LLM, verifier, …) + + SQLite reads are cached page-by-page in an LRU; the first query + pays the segment-traversal cost, subsequent queries on overlapping + pages are local-fast. + """ + + def __init__( + self, + endpoint: BucketEndpoint, + *, + cache_bytes: int = DEFAULT_CACHE_BYTES, + timeout_s: float = 30.0, + ): + self.endpoint = endpoint + self._transport = _HttpTransport(timeout_s=timeout_s) + self._cache = _LRUByteCache(max_bytes=cache_bytes) + self._vfs = HttpRangeVFS( + name=f"httprange-{id(self)}", + transport=self._transport, cache=self._cache, + ) + self._conn: apsw.Connection | None = None + + @property + def conn(self) -> apsw.Connection: + if self._conn is None: + self._conn = apsw.Connection( + self.endpoint.shard_url, + vfs=self._vfs.name, + flags=apsw.SQLITE_OPEN_READONLY, + ) + return self._conn + + def close(self) -> None: + if self._conn is not None: + self._conn.close() + self._conn = None + + # --- query helpers --- + + def fts_search(self, query: str, *, limit: int = 8) -> list[dict]: + """FTS5 body-match search returning {document_root, title, score}.""" + sql = ( + "SELECT d.document_root, d.document_uri, d.title, " + " bm25(chunks_fts) AS score " + "FROM chunks_fts JOIN chunks c ON c.rowid = chunks_fts.rowid " + "JOIN documents d ON d.document_root = c.document_root " + "WHERE chunks_fts MATCH ? " + "ORDER BY score LIMIT ?" + ) + rows = list(self.conn.execute(sql, (query, limit))) + return [ + {"document_root": r[0], "document_uri": r[1], + "title": r[2], "score": r[3]} + for r in rows + ] + + def chunks_for_doc(self, document_root: str) -> list[dict]: + """Return {idx, leaf_hash} for every chunk of a document.""" + sql = ( + "SELECT idx, leaf_hash FROM chunks " + "WHERE document_root = ? ORDER BY idx ASC" + ) + rows = list(self.conn.execute(sql, (document_root,))) + return [{"idx": r[0], "leaf_hash": r[1]} for r in rows] + + def fetch_chunk_body(self, leaf_hash: str) -> bytes: + """GET /blobs// from the bucket.""" + url = f"{self.endpoint.blob_base.rstrip('/')}/{leaf_hash[:2]}/{leaf_hash[2:]}" + # Use a HEAD-less GET (we don't care about Content-Length here). + with urllib.request.urlopen( + urllib.request.Request(url), timeout=self._transport.timeout_s + ) as resp: + return resp.read() + + def snapshot_root(self) -> str: + """Compute snapshot_root from the bucket-resident shard. + + Reads every documents.document_root over the wire (worst-case + path; same shape as `compute_snapshot_root` in snapshot.py). + Cache this result locally — the snapshot_root only changes when + the bucket is repacked. + """ + from arborist.merkle import MerkleTree + + rows = list(self.conn.execute( + "SELECT DISTINCT document_root FROM documents " + "ORDER BY document_root ASC" + )) + roots = [r[0] for r in rows] + if not roots: + return "00" * 32 + if len(roots) == 1: + return roots[0] + leaves = [bytes.fromhex(r) for r in roots] + return MerkleTree.build(leaves).root.hex() + + def stats(self) -> dict: + return self._vfs.stats() diff --git a/pyproject.toml b/pyproject.toml index b6172eb..d43b02c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -103,6 +103,17 @@ nli = [ "protobuf>=4.0", "optimum[onnxruntime]>=1.20", ] +bucket = [ + # SPV-style bucket-direct queries (arborist/wallet/bucket.py): open a + # shard `.db` file in place on a bucket via SQLite HTTP-range VFS, no + # intermediate server, no local DB. Uses apsw because Python's stdlib + # sqlite3 doesn't expose the VFS API; apsw is a thin C wrapper that + # does. Gated separately so a fresh checkout stays python3.12 + venv + # + sqlite3; tests skip via pytest.importorskip when absent. Install + # with: + # pip install 'arborist[bucket]' + "apsw>=3.45", +] object-store = [ # Cold-object eviction tier (ticket #000061). Pushes chunk bodies to an # S3-compatible bucket keyed by leaf_hash so the corpus can grow past @@ -151,6 +162,7 @@ dev = [ "arborist[hessian]", "arborist[vec]", "arborist[object-store]", + "arborist[bucket]", # moto is the wire-level boto3 test stub; only needed to run # tests/test_cold_object_boto3.py (the default suite uses MemoryBackend). "moto>=5.0", diff --git a/tests/test_wallet_bucket.py b/tests/test_wallet_bucket.py new file mode 100644 index 0000000..bc01a87 --- /dev/null +++ b/tests/test_wallet_bucket.py @@ -0,0 +1,283 @@ +"""Bucket-direct queries via HttpRangeVFS. + +Spins up an in-process HTTP server that serves files with `Range` support, +opens an arborist shard `.db` through `HttpRangeVFS`, runs the same FTS5 +queries you'd run against a local sqlite3 connection. Asserts results +match a direct sqlite3 open of the same file. + +Then fetches a chunk body from a `blobs//...` path served by the +same HTTP server and confirms its hash matches the stored leaf_hash. +""" +from __future__ import annotations + +import os +import sqlite3 +import threading +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Iterator + +import pytest + +apsw = pytest.importorskip("apsw") + +from arborist.document import Document +from arborist.ingest import ingest_source +from arborist.merkle import hash_leaf +from arborist.source import Source +from arborist.store import connect +from arborist.wallet.bucket import BucketClient, BucketEndpoint + + +class _RangeHandler(SimpleHTTPRequestHandler): + """SimpleHTTPRequestHandler that honors `Range:` requests. + + Stdlib version supports Range only on Python 3.7+; explicit support + here makes the test portable to older interpreters and gives the + server an obvious "yes I do partial reads" affirmation. + """ + + def log_message(self, fmt, *args): + return # quiet + + def do_GET(self): + path = self.translate_path(self.path) + if not os.path.isfile(path): + self.send_error(404) + return + size = os.path.getsize(path) + rng = self.headers.get("Range") + if not rng: + with open(path, "rb") as f: + body = f.read() + self.send_response(200) + self.send_header("Content-Length", str(size)) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Accept-Ranges", "bytes") + self.end_headers() + self.wfile.write(body) + return + # Parse `bytes=N-M` (single range; multi-range is overkill here). + assert rng.startswith("bytes=") + start_s, end_s = rng[len("bytes="):].split("-") + start = int(start_s) + end = int(end_s) if end_s else size - 1 + end = min(end, size - 1) + length = end - start + 1 + with open(path, "rb") as f: + f.seek(start) + body = f.read(length) + self.send_response(206) + self.send_header("Content-Range", f"bytes {start}-{end}/{size}") + self.send_header("Content-Length", str(length)) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Accept-Ranges", "bytes") + self.end_headers() + self.wfile.write(body) + + +def _serve_dir(directory: Path) -> tuple[ThreadingHTTPServer, threading.Thread, str]: + """Serve `directory` over HTTP on a random local port. Returns + (httpd, thread, base_url).""" + cwd_before = os.getcwd() + os.chdir(directory) + httpd = ThreadingHTTPServer(("127.0.0.1", 0), _RangeHandler) + port = httpd.server_address[1] + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + # Restore cwd; SimpleHTTPRequestHandler captures it per-handler-instance, + # not per-server, but our requests run in handler threads so they + # need the cwd they were started in. Workaround: leave the chdir + # for the test's lifetime; the cleanup restores. + def restore(): + os.chdir(cwd_before) + httpd._restore_cwd = restore # type: ignore[attr-defined] + return httpd, thread, f"http://127.0.0.1:{port}" + + +class _S(Source): + source_type = "test" + def __init__(self, ds): self.ds = ds + def iter_documents(self) -> Iterator[Document]: + yield from self.ds + + +@pytest.fixture +def bucket_layout(tmp_path: Path) -> tuple[Path, str, dict[str, bytes]]: + """Build a small corpus, write the .db + per-chunk blobs into a + `bucket/` directory ready to be served over HTTP. + + Returns (bucket_dir, shard_filename_under_bucket, blobs_map). + """ + bucket = tmp_path / "bucket" + bucket.mkdir() + clones = bucket / "clones" / "snap-1" + clones.mkdir(parents=True) + blobs = bucket / "blobs" + + db_path = clones / "000.db" + conn = connect(db_path) + try: + ingest_source(conn, _S([ + Document( + uri=f"test://doc-{i}", + source_type="test", + title=f"Doc {i}", + content=( + f"This is document number {i}. " * 10 + + f"The unique phrase for entry {i} is 'lemma-{i}'. " * 10 + ), + ) + for i in range(5) + ])) + finally: + conn.close() + + # Externalize every chunk to blobs//. + blob_map: dict[str, bytes] = {} + with sqlite3.connect(str(db_path)) as src: + src.row_factory = sqlite3.Row + for r in src.execute("SELECT leaf_hash, content FROM chunks WHERE content IS NOT NULL"): + text = r["content"] + if isinstance(text, bytes): + # might be zstd-packed; the existing pack flow ships + # plain bytes via unpack_chunk first. Mirror that. + from arborist.compress import unpack_chunk + body = (unpack_chunk(text) or "").encode("utf-8") + else: + body = text.encode("utf-8") + if hash_leaf(body).hex() != r["leaf_hash"]: + continue # affinity edge case: skip mismatches + blob_dir = blobs / r["leaf_hash"][:2] + blob_dir.mkdir(parents=True, exist_ok=True) + (blob_dir / r["leaf_hash"][2:]).write_bytes(body) + blob_map[r["leaf_hash"]] = body + + return bucket, "clones/snap-1/000.db", blob_map + + +def test_bucket_open_and_fts_match_local_sqlite(bucket_layout): + """Open the bucket-served .db via HttpRangeVFS; FTS5 results must + equal a direct sqlite3 open of the same file.""" + bucket, shard_rel, _blobs = bucket_layout + httpd, _t, base_url = _serve_dir(bucket) + try: + endpoint = BucketEndpoint( + shard_url=f"{base_url}/{shard_rel}", + blob_base=f"{base_url}/blobs", + ) + client = BucketClient(endpoint, cache_bytes=4 * 1024 * 1024) + try: + remote = client.fts_search('"lemma-3"', limit=4) + finally: + client.close() + + # Reference: direct sqlite3 against the file. + local = [] + with sqlite3.connect(str(bucket / shard_rel)) as conn: + conn.row_factory = sqlite3.Row + for r in conn.execute( + "SELECT d.document_root, d.document_uri, d.title, " + " bm25(chunks_fts) AS score " + "FROM chunks_fts JOIN chunks c ON c.rowid = chunks_fts.rowid " + "JOIN documents d ON d.document_root = c.document_root " + "WHERE chunks_fts MATCH ? ORDER BY score LIMIT ?", + ('"lemma-3"', 4), + ): + local.append({"document_root": r[0], "document_uri": r[1], + "title": r[2], "score": r[3]}) + + assert remote == local, ( + "bucket-direct FTS5 results must equal local sqlite3 results" + ) + # 'lemma-3' is unique to doc 3 — must be the top hit. + assert remote, "FTS should have matched at least one chunk" + assert remote[0]["document_uri"] == "test://doc-3" + finally: + httpd.shutdown() + httpd._restore_cwd() # type: ignore[attr-defined] + + +def test_bucket_fetch_chunk_body_round_trip(bucket_layout): + """Pick one leaf_hash from the corpus, fetch it via BucketClient, + confirm body hashes back to the same leaf_hash.""" + bucket, shard_rel, blob_map = bucket_layout + assert blob_map, "fixture should have at least one uploaded blob" + leaf_hash, expected_body = next(iter(blob_map.items())) + + httpd, _t, base_url = _serve_dir(bucket) + try: + endpoint = BucketEndpoint( + shard_url=f"{base_url}/{shard_rel}", + blob_base=f"{base_url}/blobs", + ) + client = BucketClient(endpoint) + try: + body = client.fetch_chunk_body(leaf_hash) + finally: + client.close() + assert body == expected_body + assert hash_leaf(body).hex() == leaf_hash + finally: + httpd.shutdown() + httpd._restore_cwd() # type: ignore[attr-defined] + + +def test_bucket_snapshot_root_matches_local(bucket_layout): + """snapshot_root computed from the bucket-direct conn must match + snapshot_root computed from a local connection.""" + from arborist.snapshot import compute_snapshot_root + + bucket, shard_rel, _blobs = bucket_layout + with sqlite3.connect(str(bucket / shard_rel)) as conn: + conn.row_factory = sqlite3.Row + local_snap, _ = compute_snapshot_root(conn) + + httpd, _t, base_url = _serve_dir(bucket) + try: + endpoint = BucketEndpoint( + shard_url=f"{base_url}/{shard_rel}", + blob_base=f"{base_url}/blobs", + ) + client = BucketClient(endpoint) + try: + remote_snap = client.snapshot_root() + finally: + client.close() + assert remote_snap == local_snap + finally: + httpd.shutdown() + httpd._restore_cwd() # type: ignore[attr-defined] + + +def test_bucket_cache_amortizes_repeated_queries(bucket_layout): + """First query pays segment traversal; second identical query + should serve mostly from the LRU page cache.""" + bucket, shard_rel, _blobs = bucket_layout + httpd, _t, base_url = _serve_dir(bucket) + try: + endpoint = BucketEndpoint( + shard_url=f"{base_url}/{shard_rel}", + blob_base=f"{base_url}/blobs", + ) + client = BucketClient(endpoint) + try: + client.fts_search('"lemma-3"', limit=4) + stats_after_first = client.stats() + client.fts_search('"lemma-3"', limit=4) + stats_after_second = client.stats() + finally: + client.close() + # The cost signal is "second query is cheap." SQLite has its + # OWN page cache (default ~2000 pages) which absorbs the second + # query entirely on small corpora — so our LRU may never see a + # hit. What matters is that HTTP traffic doesn't multiply. + r0 = stats_after_first["http_requests"] + r1 = stats_after_second["http_requests"] + assert r1 <= r0, ( + f"second identical query should add 0 HTTP requests " + f"(SQLite-cached), got {r0} → {r1}" + ) + finally: + httpd.shutdown() + httpd._restore_cwd() # type: ignore[attr-defined]