105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
"""Publication barrier and provenance shared by SQLite and Neo4j.
|
|
|
|
Builders work in a staging directory. Readers hold a shared process lock while
|
|
using a snapshot; publishers hold its exclusive counterpart. A durable pending
|
|
marker makes a process crash fail closed until the next successful sync.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from contextlib import closing, contextmanager
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
from tools.rag.common import DATA_DIR
|
|
|
|
VERSION_FILE = 'index_version.json'
|
|
PENDING_FILE = '.sync-pending.json'
|
|
LOCK_FILE = '.index-publication.lock'
|
|
|
|
|
|
class IndexUnavailableError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def fsync_directory(path: Path) -> None:
|
|
descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def durable_json(path: Path, value: dict) -> None:
|
|
temporary = path.with_suffix(path.suffix + '.tmp')
|
|
with temporary.open('w', encoding='utf-8') as handle:
|
|
json.dump(value, handle, ensure_ascii=False, indent=2)
|
|
handle.write('\n'); handle.flush(); os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
fsync_directory(path.parent)
|
|
|
|
|
|
@contextmanager
|
|
def publication_lock(data_dir: Path, *, exclusive: bool = False):
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
with (data_dir / LOCK_FILE).open('a+') as handle:
|
|
try:
|
|
flags = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH | fcntl.LOCK_NB
|
|
fcntl.flock(handle.fileno(), flags)
|
|
except BlockingIOError as error:
|
|
raise IndexUnavailableError('Prebieha aktivácia novej verzie indexov.') from error
|
|
try:
|
|
yield
|
|
finally:
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def read_version(data_dir: Path = DATA_DIR) -> dict | None:
|
|
path = data_dir / VERSION_FILE
|
|
if not path.exists():
|
|
return None # Pre-migration index; first coordinated sync adds metadata.
|
|
try:
|
|
value = json.loads(path.read_text())
|
|
if not isinstance(value, dict) or not value.get('version'):
|
|
raise ValueError('Missing version')
|
|
return value
|
|
except (ValueError, OSError) as error:
|
|
raise IndexUnavailableError('Neplatná verzia vyhľadávacieho indexu.') from error
|
|
|
|
|
|
def sqlite_version(db_file: Path) -> dict | None:
|
|
with closing(sqlite3.connect(db_file.resolve().as_uri() + '?mode=ro', uri=True)) as conn:
|
|
if not conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='index_state'").fetchone():
|
|
return None
|
|
row = conn.execute('SELECT metadata_json FROM index_state WHERE id=1').fetchone()
|
|
try:
|
|
return json.loads(row[0]) if row else None
|
|
except ValueError as error:
|
|
raise IndexUnavailableError('Neplatná verzia SQLite indexu.') from error
|
|
|
|
|
|
@contextmanager
|
|
def index_snapshot(data_dir: Path = DATA_DIR, *, db_file: Path | None = None):
|
|
with publication_lock(data_dir):
|
|
if (data_dir / PENDING_FILE).exists():
|
|
raise IndexUnavailableError('Synchronizácia indexov nebola dokončená; zopakujte synchronizáciu.')
|
|
version = read_version(data_dir)
|
|
if db_file is not None and version is not None and sqlite_version(db_file) != version:
|
|
raise IndexUnavailableError('SQLite index nezodpovedá aktívnej verzii dát.')
|
|
yield version
|
|
|
|
|
|
def rag_snapshot(function):
|
|
@wraps(function)
|
|
def wrapped(*args, **kwargs):
|
|
db_file = Path(args[0] if args else kwargs.get('db_path', kwargs.get('db_file')))
|
|
with index_snapshot(db_file.parent, db_file=db_file) as version:
|
|
result = function(*args, **kwargs)
|
|
if version is not None:
|
|
result['index_version'] = version
|
|
return result
|
|
return wrapped
|