summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-06 01:33:16 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-06 01:33:37 +0200
commitf2da33a648f86e34ddbbb5f6bec124828ef2a847 (patch)
tree670cbf62ff92598847e067b5835713dfd854dfb5 /packages/meshbay-node/src/meshbay_node/indexer/indexer.py
parentfff1974edf19cf1186e0f49da5f8a4d237bcb13e (diff)
downloadmeshbay-f2da33a648f86e34ddbbb5f6bec124828ef2a847.tar.gz
feat(node): indexing v2 — partial-read hashing for files above 40 MB
Files above 40 MB are no longer read in full. Instead, blake3 hashes 45 MB of samples (first 20 MB + last 20 MB + 5 MB at 50% offset). Files at or below 40 MB are unchanged (full read, hash_version 1). A new `hash_version` field on IndexEntry (default 1) travels on the wire and through the cache so both versions coexist without breaking existing nodes or clients. The IndexCache auto-migrates its schema on open (ALTER TABLE), so no manual step is required on upgrade. A standalone migration script is available in QE/migration/ for operators who want to preview or force a full re-hash. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer/indexer.py')
-rw-r--r--packages/meshbay-node/src/meshbay_node/indexer/indexer.py67
1 files changed, 50 insertions, 17 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index f78465b..8376c23 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -94,6 +94,32 @@ def _is_indexable_size(path: Path, size: int) -> bool:
_HASH_CHUNK = 8 * 1024 * 1024 # 8 MB streaming hash chunks
+_PARTIAL_THRESHOLD = 40 * 1024 * 1024 # files above this use partial-read hashing
+_PARTIAL_HEAD = 20 * 1024 * 1024
+_PARTIAL_TAIL = 20 * 1024 * 1024
+_PARTIAL_MID = 5 * 1024 * 1024
+
+
+def _feed(hasher, f, nbytes: int) -> None:
+ remaining = nbytes
+ while remaining > 0:
+ chunk = f.read(min(_HASH_CHUNK, remaining))
+ if not chunk:
+ break
+ hasher.update(chunk)
+ remaining -= len(chunk)
+
+
+def _partial_hash(file_path: Path, size: int) -> str:
+ hasher = blake3.blake3()
+ with open(long_path(file_path), "rb") as f:
+ _feed(hasher, f, _PARTIAL_HEAD)
+ f.seek(size - _PARTIAL_TAIL)
+ _feed(hasher, f, _PARTIAL_TAIL)
+ f.seek(size // 2)
+ _feed(hasher, f, _PARTIAL_MID)
+ return hasher.hexdigest()
+
@dataclass
class IndexProgress:
@@ -162,29 +188,33 @@ def _size_files(files: list[Path]) -> list[tuple[Path, int]]:
def _scan_file(root: Root, file_path: Path) -> IndexEntry | None:
"""Compute IndexEntry for a file. Blocking — run in executor.
- Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.)
- don't require loading the whole file into memory."""
+ Files <= 40 MB are hashed in full (hash_version 1). Files > 40 MB use a
+ 45 MB partial read — first 20 MB, last 20 MB, 5 MB at 50% — for
+ hash_version 2."""
if not _is_indexable(file_path):
return None
try:
stat = file_path.stat()
if not _is_indexable_size(file_path, stat.st_size):
return None
- hasher = blake3.blake3()
- # long_path is a no-op off Windows; there it is what lets a deep media
- # library past MAX_PATH.
- with open(long_path(file_path), "rb") as f:
- while chunk := f.read(_HASH_CHUNK):
- hasher.update(chunk)
+ if stat.st_size > _PARTIAL_THRESHOLD:
+ hex_hash = _partial_hash(file_path, stat.st_size)
+ hv = 2
+ else:
+ hasher = blake3.blake3()
+ with open(long_path(file_path), "rb") as f:
+ while chunk := f.read(_HASH_CHUNK):
+ hasher.update(chunk)
+ hex_hash = hasher.hexdigest()
+ hv = 1
return IndexEntry(
- id=hasher.hexdigest(),
- # Stored exactly as the filesystem gave it: this is the string that
- # opens the file. Normalization is for comparison only.
+ id=hex_hash,
name=file_path.name,
path=_virtual_dir(root, file_path),
size=stat.st_size,
type=_detect_type(file_path),
added_at=int(stat.st_mtime),
+ hash_version=hv,
)
except (OSError, PermissionError, ValueError) as e:
log.warning("Cannot index %s: %s", file_path, e)
@@ -375,10 +405,8 @@ class DirectoryIndexer:
async def _hash_or_cached(self, root: Root, file_path: Path) -> IndexEntry | None:
"""
Cache-aware replacement for a bare _scan_file() call: skips the
- content read entirely when this path's (size, mtime) still match
- what was hashed last time — the difference between a redundant full
- rehash of a 100+ GB library on every restart and a stat()-only pass.
- The only place that decides to actually read a file's bytes.
+ content read entirely when this path's (size, mtime, hash_version)
+ still match what was hashed last time.
"""
if not _is_indexable(file_path):
return None
@@ -389,8 +417,11 @@ class DirectoryIndexer:
if not _is_indexable_size(file_path, st.st_size):
return None
+ expected_hv = 2 if st.st_size > _PARTIAL_THRESHOLD else 1
+
if self._cache is not None:
- cached = await self._cache.lookup(str(file_path), st.st_size, st.st_mtime)
+ cached = await self._cache.lookup(
+ str(file_path), st.st_size, st.st_mtime, expected_hv)
if cached is not None:
return IndexEntry(
id=cached.hash,
@@ -399,13 +430,15 @@ class DirectoryIndexer:
size=st.st_size,
type=cached.type,
added_at=cached.added_at,
+ hash_version=cached.hash_version,
)
loop = asyncio.get_event_loop()
entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path)
if entry and self._cache is not None:
await self._cache.put(str(file_path), st.st_size, st.st_mtime,
- entry.id, entry.type, entry.added_at)
+ entry.id, entry.type, entry.added_at,
+ entry.hash_version)
return entry
def _report_collisions(self) -> None: