jp170na/dp-zp-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Zjednotenie synchronizácie indexov

Browse Source
This commit is contained in:
Ján Pták 2026-10-25 15:26:37 +01:00
parent 90abe1558b
commit 86a155691d
18 changed files with 882 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
README.md
View File

@ -669,8 +669,9 @@ docker compose exec zp-agent-api \
## Reindexovanie
Celý proces načíta dokumenty, vytvorí chunky, obnoví FTS5 index a
vytvorí embedding pre každý chunk:
Spoločná synchronizácia pripraví dokumenty, chunky, SQLite/FTS5, embeddingy
a Neo4j graf z jedného snapshotu wiki. Nová generácia sa aktivuje až po
úspešnej príprave oboch indexov:
```bash
docker compose run --rm zp-agent-api python -m tools.rag.rebuild_index
@ -682,12 +683,17 @@ Vzniknú súbory:
data/documents.json
data/chunks.json
data/zp_index.sqlite
data/index_version.json
```
Databáza obsahuje dokumenty, chunky, FTS5 index, metadata a embeddingy.
Pri automatickom reindexovaní sa používa zámok proti súbežnému spusteniu
a atomická výmena databázy po úspešnom vytvorení nového indexu.
Manuálny `/sync`, push webhook a uvedený CLI príkaz používajú rovnakú pipeline.
SQLite, Neo4j a odpovede nástrojov evidujú spoločnú verziu dát a Git revíziu.
Pri krátkom prepnutí sa nové vyhľadávanie dočasne odmietne; chyba grafu
vyvolá rollback. Pri nedokončenom prepnutí po páde zostane vyhľadávanie
nedostupné do úspešného opakovania synchronizácie. Podrobnosti a zotavenie:
[Spoločná synchronizácia indexov](docs/synchronizacia_indexov.md).
## Hybridné vyhľadávanie
@ -837,8 +843,9 @@ docker compose exec zp-agent-api \
```
Builder vytvára constraints, zapisuje uzly a vzťahy a pri štandardnom
spustení obnoví iba časť grafu spravovanú ZP Agentom. Opakované
spustenie preto nevytvára duplicitné uzly.
spustení obnoví iba časť grafu spravovanú ZP Agentom. Zmazanie aj zápis
prebiehajú v jednej transakcii s kontrolou počtov, takže chyba zachová
pôvodný graf. Opakované spustenie nevytvára duplicitné uzly.
Graf je možné vizuálne kontrolovať v Neo4j Browseri. Samostatné
grafové vyhľadávanie je implementované v `tools/graphrag/graph_search.py`.

19
app/routes.py
View File

@ -46,6 +46,7 @@ from tools.rag.rebuild_index import (
rebuild_index,
)
from tools.rag.search_utils import search_database
from tools.index_state import index_snapshot, IndexUnavailableError
logger = logging.getLogger(__name__)
@ -182,6 +183,7 @@ class HealthResponse(BaseModel):
zpwiki_exists: bool
security_configured: bool
index_version: dict[str, Any] | None = None
class RagSource(BaseModel):
@ -231,6 +233,7 @@ class RagResponse(BaseModel):
sources: list[
RagSource
]
index_version: dict[str, Any] | None = None
class SearchResponse(BaseModel):
@ -246,6 +249,7 @@ class SearchResponse(BaseModel):
results: list[
dict[str, Any]
]
index_version: dict[str, Any] | None = None
class SyncResponse(BaseModel):
@ -261,6 +265,8 @@ class SyncResponse(BaseModel):
str,
int,
]
index_version: dict[str, Any] | None = None
graph_counts: dict[str, int] | None = None
class WebhookSuccessResponse(
@ -283,6 +289,8 @@ class WebhookSuccessResponse(
str,
int,
]
index_version: dict[str, Any] | None = None
graph_counts: dict[str, int] | None = None
class WebhookIgnoredResponse(
@ -529,10 +537,17 @@ def health() -> dict[str, Any]:
and zpwiki_exists
and security_configured
)
version = None
try:
with index_snapshot(DB_FILE.parent, db_file=DB_FILE) as version:
pass
except (IndexUnavailableError, sqlite3.Error):
ready = False
return {
"status": "ok",
"ready": ready,
"index_version": version,
"database_exists": (
database_exists
),
@ -708,6 +723,8 @@ def sync(
"pull_git": (
request.pull_git
),
"index_version": result.get('index_version'),
"graph_counts": result.get('graph_counts'),
"duration_seconds": (
result[
"duration_seconds"
@ -890,6 +907,8 @@ async def gitea_webhook(
"verified_by": (
"hmac_sha256"
),
"index_version": result.get('index_version'),
"graph_counts": result.get('graph_counts'),
"duration_seconds": (
result[
"duration_seconds"

93
docs/synchronizacia_indexov.md Normal file
View File

@ -0,0 +1,93 @@
# Spoločná synchronizácia RAG a GraphRAG
`POST /sync`, overený push webhook aj `python -m tools.rag.rebuild_index`
používajú spoločnú pipeline v `tools/sync_indexes.py`. Obnova vyžaduje dostupné
Neo4j a jeho konfiguráciu v prostredí. Pôvodné evaluačné výsledky sa nemenia.
## Priebeh
1. Procesový `.reindex.lock` pripustí iba jednu synchronizáciu nad data volume.
2. Voliteľný `git pull --ff-only` zachováva pôvodné nastavenie synchronizácie.
3. V `data/.sync-stage-*` vznikne kópia pôvodných README súborov. Dokumenty,
chunky, SQLite/FTS5 a embeddingy sa pripravujú z tejto jednej kópie. Aktívne
súbory sa počas prípravy nemenia a existujúce vyhľadávanie môže pokračovať.
4. Overí sa, že sa zdrojové súbory ani Git revízia počas prípravy nezmenili.
5. Pred aktiváciou vzniknú zálohy pôvodných súborov a trvalý marker
`.sync-pending.json`. Exkluzívny publikačný zámok počká na rozbehnutých
čitateľov. Nové požiadavky počas aktivácie dostanú dočasnú nedostupnosť.
6. Jedna explicitná Neo4j transakcia odstráni starý spravovaný graf, zapíše
nový graf a skontroluje počty uzlov/vzťahov oproti pripravenému payloadu.
Verzia grafu sa zapisuje v tej istej transakcii.
7. Pod publikačným zámkom sa aktivujú pripravené lokálne súbory a potvrdí sa
grafová transakcia. Po úspechu sa odstráni marker a sprístupní nová verzia.
SQLite a Neo4j netvoria distribuovanú ACID transakciu. Konzistenciu pri tejto
implementácii zaisťujú spoločný zámok pre podporované vyhľadávacie vstupy,
verzie, zálohy a marker nedokončeného prepnutia. Používanie samotného nízkoúrovňového
`search_graph_with_session` mimo publikačnej ochrany je určené pre kontrolované
interné evaluácie/testy; produkčný volajúci používa chránené `search_graph`.
## Verzia a dohľadateľnosť
Každá generácia má identifikátor `version`, Git revíziu zdrojovej wiki,
UTC čas prípravy, hash pôvodných README súborov, hashe dokumentov/chunkov
a názov embeddingového modelu. Hash zdrojov zachytáva aj necommitnuté zmeny;
samotný Git HEAD preto nie je jediným identifikátorom obsahu.
Rovnaký objekt `index_version` je uložený v:
- `data/index_version.json`,
- SQLite tabuľke `index_state`,
- operatívnom uzle Neo4j `IndexState` s ID `zpwiki`.
Operatívny uzol nie je súčasťou znalostného grafu označeného `graph_scope=zpwiki`.
Verziu vracajú `/sync`, webhook, `/health`, RAG vyhľadávanie a verejná Python
funkcia grafového vyhľadávania. Synchronizácia vracia aj `graph_counts`.
## Chyby a zotavenie
- Chyba prípravy, embeddingov alebo nedostupné Neo4j pred aktiváciou:
pôvodné indexy ostávajú aktívne.
- Chyba pri zápise grafu: celá grafová transakcia sa zruší.
- Chyba pri aktivácii lokálneho súboru: obnovia sa pôvodné súbory a grafová
transakcia sa zruší. Pri neúspešnej obnove marker zostáva.
- Stratené potvrdenie COMMIT: najprv sa kontroluje skutočná verzia v Neo4j.
Ak bola nová verzia potvrdená, staré lokálne dáta sa naslepo neobnovujú.
- Pád procesu alebo neoveriteľný výsledok transakcie: marker a obnovovacie
súbory ostávajú na disku. Vyhľadávanie odmietne nejasný snapshot; RAG API
vracia 503 a `/health` má `ready=false`.
Po odstránení príčiny spusti znova autorizovaný `/sync` alebo:
```bash
docker compose exec zp-agent-api python -m tools.rag.rebuild_index
```
Úspešná synchronizácia vytvorí kompletnú novú generáciu oboch indexov a marker
odstráni. Marker sa nemá ručne mazať ako náhrada opravy. Staršie adresáre
prerušených behov môžu zostať ako podklady na obnovu; bežný úspešný staging
sa automaticky odstraňuje.
## Samostatná obnova Neo4j
`python -m tools.graphrag.build_graphrag` používa rovnaký transakčný zápis.
Zmena grafu sa potvrdí celá alebo sa zruší celá. Ak JSON vstupy zodpovedajú
aktívnej generácii, jej proveniencia sa zachová. Iné vstupy alebo režim
`--no-replace` nemôžu predstierať synchronizovanú verziu; verejný grafový
vstup potom pri aktívnom manifeste vyžaduje spoločnú synchronizáciu.
## Overenie
Regresie pokrývajú chybu uprostred grafového zápisu, chybu po čiastočnej
aktivácii súborov, stratenú odpoveď COMMIT, neznámy výsledok COMMIT, následné
zotavenie, zámok pre čitateľov, nesúlad verzií, zmenu zdroja počas prípravy,
odpovede `/sync` aj podpísaného webhooku a API 503 pri nejasnej generácii.
Skutočné Neo4j testy overujú rollback po odstránení pôvodných uzlov a zápise
časti náhrady v izolovanom scope. Pozitívny prípad overuje aj opakovateľnosť.
Záverečná sada po dokončení opravy: **464 passed, 4 skipped**. Reálny `/sync`
v Docker API obnovil 114 dokumentov, 887 chunkov a 447 grafových uzlov /
879 vzťahov za 87,72 s. Overené sú rovnaké verzie v manifeste, SQLite, Neo4j,
health, skutočnej RAG odpovedi aj grafovom vyhľadávaní. Obsah znalostného
grafu zodpovedá pôvodnej v4; nové meranie presnosti LLM sa nevykonávalo.
Podrobné overenie: [záznam](synchronizacia_indexov_overenie.json).

52
docs/synchronizacia_indexov_overenie.json Normal file
View File

@ -0,0 +1,52 @@
{
"date": "2026-10-25",
"changes": [
"Coordinated RAG/Neo4j snapshot publication for /sync, push webhook and CLI.",
"Atomic Neo4j replacement with rollback, count validation and data-version metadata."
],
"backup_directory": "/tmp/zp-agent-sync-backup-20261025T023515Z",
"automated_tests": {
"passed": 464,
"skipped": 4
},
"live_sync": {
"status": "ok",
"pull_git": false,
"duration_seconds": 87.72,
"counts": {
"documents": 114,
"chunks": 887,
"fts_chunks": 887,
"embedding_chunks": 887,
"tags": 2189,
"categories": 1193
},
"index_version": {
"version": "d4a6fbaa03b44a0a8aabb3cba2ddff36",
"git_revision": "3e892a1d49e9bce118cd03478ce15e01c2bebb4e",
"created_at_utc": "2026-10-25T02:50:27.976522+00:00",
"source_sha256": "007792c12f3467dd3203bb4951ccc197fdc0eef669580f1b1f893baf609ca363",
"documents_sha256": "0542af2a373771cf7c4a95f93d14f4f687f6a0187846892c959aa6efc4bade52",
"chunks_sha256": "437dde589be46e0c30842dd330de22d96a0173d2dbaff9c5f129568cb2bca4c9",
"embedding_model": "intfloat/multilingual-e5-small"
},
"graph_counts": {
"nodes": 447,
"relationships": 879
}
},
"live_verification": {
"ready": true,
"rag_sources": 3,
"graph_status": "ok",
"versions_equal_in_sqlite_neo4j_manifest_and_responses": true,
"knowledge_graph_unchanged_against_frozen_v4": true
},
"verification_scope": [
"Full existing and new regression suite, including real isolated Neo4j rollback.",
"Signed webhook and manual sync response tests.",
"Actual /sync on deployed Docker API without Git pull.",
"Actual authenticated /rag and public graph lookup on the same data generation."
],
"new_llm_accuracy_measurement": false
}

238
test/test_index_sync.py Normal file
View File

@ -0,0 +1,238 @@
"""Failure-injection tests for coordinated publication, not only happy paths."""
from __future__ import annotations
import hashlib
import hmac
import json
import sqlite3
from pathlib import Path
import pytest
from tools import index_state as state
from tools import sync_indexes as sync
class Result:
def __init__(self, value):
self.value = value
def single(self):
return self.value
class Session:
def __init__(self, version):
self.version = version; self.fail_commit = None; self.unavailable = False
def run(self, query, **params):
if self.unavailable:
raise ConnectionError('lost database connection')
return Result({'metadata': json.dumps(self.version)} if self.version else None)
def begin_transaction(self):
return Transaction(self)
class Transaction:
def __init__(self, session):
self.session = session; self.version = session.version
def __enter__(self):
return self
def __exit__(self, *args):
pass # Uncommitted local state is discarded, as in the Neo4j driver.
def commit(self):
if self.session.fail_commit == 'unknown':
self.session.unavailable = True
raise ConnectionError('commit outcome unknown')
self.session.version = self.version
if self.session.fail_commit == 'response_lost':
raise ConnectionError('committed but response lost')
def snapshot(directory: Path, version: dict):
directory.mkdir(parents=True, exist_ok=True)
(directory / 'documents.json').write_text(json.dumps([version]))
(directory / 'chunks.json').write_text(json.dumps([version]))
with sqlite3.connect(directory / 'zp_index.sqlite') as conn:
conn.execute('CREATE TABLE index_state (id INTEGER PRIMARY KEY, metadata_json TEXT)')
conn.execute('INSERT INTO index_state VALUES (1,?)', (json.dumps(version),))
state.durable_json(directory / state.VERSION_FILE, version)
@pytest.fixture
def publication(tmp_path, monkeypatch):
data = tmp_path / 'data'; stage = data / '.sync-stage-test'
old = {'version': 'old', 'git_revision': 'commit-old'}
new = {'version': 'new', 'git_revision': 'commit-new'}
snapshot(data, old); snapshot(stage, new)
monkeypatch.setattr(sync, 'write_graph_transaction', lambda tx, payload, version: setattr(tx, 'version', version))
monkeypatch.setattr(sync, 'graph_counts', lambda tx: ([('Person', 1)], [('HAS_DOCUMENT', 1)]))
return data, stage, old, new, Session(old)
def test_publish_makes_all_versions_agree(publication):
data, stage, old, new, session = publication
result = sync.publish_snapshot(data, stage, new, {}, session)
assert result == {'nodes': 1, 'relationships': 1}
assert state.read_version(data) == state.sqlite_version(data / 'zp_index.sqlite') == session.version == new
with state.index_snapshot(data, db_file=data / 'zp_index.sqlite') as version:
assert version == new
assert not (data / state.PENDING_FILE).exists()
def test_graph_write_failure_keeps_original_files_and_graph(publication, monkeypatch):
data, stage, old, new, session = publication
before = {name: (data / name).read_bytes() for name in sync.FILES}
def fail(tx, payload, version):
tx.version = new
raise RuntimeError('failure halfway through graph write')
monkeypatch.setattr(sync, 'write_graph_transaction', fail)
with pytest.raises(RuntimeError, match='halfway'):
sync.publish_snapshot(data, stage, new, {}, session)
assert session.version == old
assert before == {name: (data / name).read_bytes() for name in sync.FILES}
assert not (data / state.PENDING_FILE).exists()
def test_partial_file_publication_rolls_back_files_and_graph(publication, monkeypatch):
data, stage, old, new, session = publication
before = {name: (data / name).read_bytes() for name in sync.FILES}
replace = sync.replace_file
def fail(source, target):
if target.name == 'zp_index.sqlite':
raise OSError('disk write failed')
replace(source, target)
monkeypatch.setattr(sync, 'replace_file', fail)
with pytest.raises(OSError, match='disk write'):
sync.publish_snapshot(data, stage, new, {}, session)
assert session.version == old
assert before == {name: (data / name).read_bytes() for name in sync.FILES}
assert not (data / state.PENDING_FILE).exists()
def test_lost_commit_response_is_verified_before_restoring_any_files(publication):
data, stage, old, new, session = publication
session.fail_commit = 'response_lost'
sync.publish_snapshot(data, stage, new, {}, session)
assert session.version == state.sqlite_version(data / 'zp_index.sqlite') == new
assert not (data / state.PENDING_FILE).exists()
def test_unknown_commit_blocks_reads_until_next_successful_sync(publication):
data, stage, old, new, session = publication
session.fail_commit = 'unknown'
with pytest.raises(ConnectionError):
sync.publish_snapshot(data, stage, new, {}, session)
assert (stage / 'previous/zp_index.sqlite').exists()
with pytest.raises(state.IndexUnavailableError, match='nebola dokončená'):
with state.index_snapshot(data):
pytest.fail('Mixed generation must never be readable')
session.unavailable = False; session.fail_commit = None
repaired = {'version': 'repaired'}; next_stage = data / '.sync-stage-repaired'
snapshot(next_stage, repaired)
sync.publish_snapshot(data, next_stage, repaired, {}, session)
assert session.version == state.sqlite_version(data / 'zp_index.sqlite') == state.read_version(data) == repaired
assert not (data / state.PENDING_FILE).exists()
def test_read_during_publication_is_unavailable(publication):
data, *_ = publication
with state.publication_lock(data, exclusive=True):
with pytest.raises(state.IndexUnavailableError, match='aktivácia'):
with state.index_snapshot(data):
pytest.fail('Publication barrier bypassed')
def test_reader_rejects_a_replaced_sqlite_database(publication):
data, stage, old, new, session = publication
state.durable_json(data / state.VERSION_FILE, new)
with pytest.raises(state.IndexUnavailableError, match='SQLite'):
with state.index_snapshot(data, db_file=data / 'zp_index.sqlite'):
pass
def test_staging_failure_never_touches_live_indices(tmp_path, monkeypatch):
data = tmp_path / 'data'; snapshot(data, {'version': 'old'})
before = {name: (data / name).read_bytes() for name in sync.FILES}
def fail(wiki, stage):
(stage / 'documents.json').write_text('new incomplete data')
raise RuntimeError('embedding failed')
monkeypatch.setattr(sync, 'prepare_snapshot', fail)
with pytest.raises(RuntimeError, match='embedding failed'):
sync.synchronize_indexes(tmp_path / 'wiki', data)
assert before == {name: (data / name).read_bytes() for name in sync.FILES}
assert not list(data.glob('.sync-stage-*'))
def test_prepare_uses_one_frozen_source_and_records_versions(tmp_path, monkeypatch):
import numpy as np
import tools.rag.build_sqlite_index as indexer
wiki = tmp_path / 'wiki'; page = wiki / 'pages/students/2020/eva/README.md'
page.parent.mkdir(parents=True)
page.write_text('---\ntitle: Eva Nová\npublished: true\ntaxonomy:\n category: [bp2024]\n---\n# Bakalárska práca 2024\n\nNázov: Overovanie synchronizácie indexov\n')
monkeypatch.setattr(indexer, 'embed_passages', lambda texts: np.ones((len(texts), 4), dtype=np.float32))
stage = tmp_path / 'stage'; stage.mkdir()
version, counts, payload = sync.prepare_snapshot(wiki, stage)
assert counts['documents'] == 1 and counts['chunks'] > 0
assert version == state.sqlite_version(stage / 'zp_index.sqlite') == state.read_version(stage)
assert version['documents_sha256'] == sync.digest(stage / 'documents.json')
assert version['chunks_sha256'] == sync.digest(stage / 'chunks.json')
assert payload['people'][0]['name'] == 'Eva Nová'
assert payload['theses'][0]['year'] == 2024
assert payload['documents'][0]['path'] == 'pages/students/2020/eva/README.md'
def test_changed_source_is_not_activated(tmp_path, monkeypatch):
import numpy as np
import tools.rag.build_sqlite_index as indexer
wiki = tmp_path / 'wiki'; page = wiki / 'pages/example/README.md'
page.parent.mkdir(parents=True); page.write_text('---\ntitle: Source\n---\nInitial source text.')
def embed(texts):
page.write_text('Source changed during indexing.')
return np.ones((len(texts), 4), dtype=np.float32)
monkeypatch.setattr(indexer, 'embed_passages', embed)
stage = tmp_path / 'stage'; stage.mkdir()
with pytest.raises(RuntimeError, match='počas indexovania'):
sync.prepare_snapshot(wiki, stage)
def test_api_returns_503_while_publication_is_unresolved(tmp_path, monkeypatch):
from fastapi.testclient import TestClient
import app.main as main
import app.routes as routes
monkeypatch.setattr(main, 'embed_query', lambda q: None)
for name, value in [('WEBHOOK_SECRET', 'w'), ('SEARCH_API_KEY', 'a'), ('SYNC_API_KEY', 's')]:
monkeypatch.setenv(name, value * 64)
data = tmp_path / 'data'; snapshot(data, {'version': 'old'})
state.durable_json(data / state.PENDING_FILE, {'target_version': 'new'})
monkeypatch.setattr(routes, 'DB_FILE', data / 'zp_index.sqlite')
with TestClient(main.app) as client:
response = client.post('/rag', json={'query': 'Aká práca?'},
headers={'Authorization': 'Bearer ' + 'a' * 64})
assert response.status_code == 503
assert client.get('/health').json()['ready'] is False
@pytest.mark.parametrize('endpoint', ['sync', 'webhook/gitea'])
def test_sync_api_exposes_common_version_and_graph_counts(monkeypatch, endpoint):
from fastapi.testclient import TestClient
import app.main as main
import app.routes as routes
monkeypatch.setattr(main, 'embed_query', lambda q: None)
for name, value in [('WEBHOOK_SECRET', 'w'), ('SEARCH_API_KEY', 'a'), ('SYNC_API_KEY', 's')]:
monkeypatch.setenv(name, value * 64)
monkeypatch.setenv('EXPECTED_GITEA_REPOSITORY', 'KEMT/zpwiki')
monkeypatch.setenv('WEBHOOK_PULL_GIT', 'false')
version = {'version': 'new', 'git_revision': 'commit'}
monkeypatch.setattr(routes, 'rebuild_index', lambda **kwargs: {
'duration_seconds': 0.1, 'counts': {'documents': 1}, 'index_version': version,
'graph_counts': {'nodes': 2, 'relationships': 1}})
with TestClient(main.app) as client:
if endpoint == 'sync':
response = client.post('/sync', json={}, headers={'X-API-Key': 's' * 64})
else:
body = json.dumps({'repository': {'full_name': 'KEMT/zpwiki'}}).encode()
signature = hmac.new(('w' * 64).encode(), body, hashlib.sha256).hexdigest()
response = client.post('/webhook/gitea', content=body,
headers={'X-Gitea-Event': 'push', 'X-Gitea-Signature': signature, 'Content-Type': 'application/json'})
assert response.status_code == 200
assert response.json()['index_version'] == version
assert response.json()['graph_counts'] == {'nodes': 2, 'relationships': 1}

95
tools/graphrag/build_graphrag.py
View File

@ -13,6 +13,7 @@ from .graphrag import (
GRAPH_SCOPE,
build_graph_payload,
)
from tools.index_state import DATA_DIR, PENDING_FILE, publication_lock, read_version
PROJECT_ROOT = Path(
@ -33,6 +34,7 @@ CHUNKS_FILE = (
CONSTRAINTS = (
('zpwiki_index_state_unique', 'IndexState', 'id'),
(
"zpwiki_document_path_unique",
"Document",
@ -81,6 +83,45 @@ LEGACY_CONSTRAINTS = (
)
def graph_version(session) -> dict | None:
record = session.run('MATCH (s:IndexState {id:$scope}) RETURN s.metadata_json AS metadata',
scope=GRAPH_SCOPE).single()
return json.loads(record['metadata']) if record and record['metadata'] else None
def write_graph_transaction(tx, payload: dict, *, replace: bool = True,
version: dict | None = None) -> None:
"""All destructive data changes belong to the caller's single transaction."""
if replace:
clear_managed_graph(tx)
for key, query in NODE_QUERIES.items():
write_rows(tx, query, payload[key])
for key, query in RELATIONSHIP_QUERIES.items():
write_rows(tx, query, payload[key])
if replace:
nodes, relationships = graph_counts(tx)
labels = {'people': 'Person', 'document_authors': 'DocumentAuthor', 'documents': 'Document',
'categories': 'Category', 'topics': 'Topic', 'keywords': 'Keyword',
'theses': 'Thesis', 'works': 'Work'}
types = {'person_documents': 'HAS_DOCUMENT', 'document_author_documents': 'AUTHORED_DOCUMENT',
'document_categories': 'IN_CATEGORY', 'document_keywords': 'HAS_KEYWORD',
'document_describes_topics': 'DESCRIBES', 'person_theses': 'AUTHORED',
'person_works': 'HAS_WORK'}
expected_nodes = {labels[key]: len(payload[key]) for key in labels if payload[key]}
expected_relations = {types[key]: len(payload[key]) for key in types if payload[key]}
evidenced = len(payload['thesis_documents']) + len(payload['work_documents'])
if evidenced:
expected_relations['EVIDENCED_BY'] = evidenced
if dict(nodes) != expected_nodes or dict(relationships) != expected_relations:
raise RuntimeError('Neo4j počty nezodpovedajú pripravenému grafu; transakcia sa zruší.')
if version is None:
tx.run('MATCH (s:IndexState {id:$scope}) DELETE s', scope=GRAPH_SCOPE).consume()
else:
# Operational state is deliberately outside the knowledge graph scope.
tx.run('MERGE (s:IndexState {id:$scope}) SET s.metadata_json=$metadata',
scope=GRAPH_SCOPE, metadata=json.dumps(version, sort_keys=True)).consume()
NODE_QUERIES = {
"people": """
UNWIND $rows AS row
@ -508,23 +549,6 @@ def main() -> None:
args = parser.parse_args()
documents = load_json(
DOCUMENTS_FILE
)
chunks = load_json(
CHUNKS_FILE
)
payload = build_graph_payload(
documents,
chunks,
)
print_payload_summary(
payload
)
settings = get_neo4j_settings()
with create_neo4j_driver(
@ -543,28 +567,21 @@ def main() -> None:
session
)
if args.replace:
clear_managed_graph(
session
)
for key, query in (
NODE_QUERIES.items()
):
write_rows(
session,
query,
payload[key],
)
for key, query in (
RELATIONSHIP_QUERIES.items()
):
write_rows(
session,
query,
payload[key],
)
import hashlib
with publication_lock(DATA_DIR, exclusive=True):
if (DATA_DIR / PENDING_FILE).exists():
raise RuntimeError('Nedokončená spoločná synchronizácia; použite tools.rag.rebuild_index.')
payload = build_graph_payload(load_json(DOCUMENTS_FILE), load_json(CHUNKS_FILE))
print_payload_summary(payload)
version = read_version(DATA_DIR)
if not args.replace:
version = None # An incremental merge cannot certify removed source records.
if version and any(hashlib.sha256(path.read_bytes()).hexdigest() != version[key]
for path, key in [(DOCUMENTS_FILE, 'documents_sha256'), (CHUNKS_FILE, 'chunks_sha256')]):
version = None # A standalone rebuild must not claim stale provenance.
with session.begin_transaction() as tx:
write_graph_transaction(tx, payload, replace=args.replace, version=version)
tx.commit()
print_database_summary(
session

30
tools/graphrag/graph_search.py
View File

@ -2600,30 +2600,22 @@ def search_graph(
get_neo4j_settings,
)
from tools.index_state import DATA_DIR, IndexUnavailableError, index_snapshot
from .build_graphrag import graph_version
settings = (
get_neo4j_settings()
)
with create_neo4j_driver(
settings
) as driver:
with index_snapshot(DATA_DIR) as version:
with create_neo4j_driver(settings) as driver:
driver.verify_connectivity()
with driver.session(
database=(
settings.database
)
) as session:
return (
search_graph_with_session(
session,
query,
limit=limit,
max_entities=(
max_entities
),
)
)
with driver.session(database=settings.database) as session:
if version is not None and graph_version(session) != version:
raise IndexUnavailableError('Neo4j graf nezodpovedá aktívnej verzii dát.')
result = search_graph_with_session(session, query, limit=limit, max_entities=max_entities)
if version is not None:
result['index_version'] = version
return result
def main() -> None:

61
tools/graphrag/tests/test_graph_rebuild_atomic.py Normal file
View File

@ -0,0 +1,61 @@
"""Actual Neo4j rollback after DELETE and partial replacement; isolated scopes."""
import os
import uuid
import pytest
from tools.graphrag import build_graphrag as builder
from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings
pytestmark = pytest.mark.skipif(os.getenv('RUN_NEO4J_TESTS') != '1', reason='Requires local Neo4j.')
@pytest.fixture
def managed_graph(monkeypatch):
scope = 'atomic-rebuild-' + uuid.uuid4().hex
monkeypatch.setattr(builder, 'GRAPH_SCOPE', scope)
settings = get_neo4j_settings()
with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session:
session.run('CREATE (:Person {id:$id, name:"Original", graph_scope:$scope})',
id=scope+'old', scope=scope).consume()
session.run('CREATE (:IndexState {id:$scope, metadata_json:$metadata})',
scope=scope, metadata='{"version":"old"}').consume()
try:
yield session, scope
finally:
session.run('MATCH (n) WHERE n.graph_scope=$scope DETACH DELETE n', scope=scope).consume()
session.run('MATCH (s:IndexState {id:$scope}) DELETE s', scope=scope).consume()
def replacement(scope):
payload = {key: [] for key in [*builder.NODE_QUERIES, *builder.RELATIONSHIP_QUERIES]}
payload['people'] = [{'id': scope+'new', 'name': 'Replacement', 'graph_scope': scope}]
payload['documents'] = [{'path': scope+'/p.md', 'graph_scope': scope}]
payload['person_documents'] = [{'person_id': scope+'new', 'document_path': scope+'/p.md'}]
return payload
def test_failed_replacement_restores_deleted_graph_and_version(managed_graph):
session, scope = managed_graph
payload = replacement(scope)
payload['documents'].append({'path': None, 'graph_scope': scope})
with pytest.raises(Exception):
with session.begin_transaction() as tx:
builder.write_graph_transaction(tx, payload, version={'version': 'new'})
tx.commit()
rows = session.run('MATCH (n) WHERE n.graph_scope=$scope RETURN n.name AS name', scope=scope).data()
assert rows == [{'name': 'Original'}]
assert builder.graph_version(session) == {'version': 'old'}
def test_successful_replacement_commits_graph_and_version_together(managed_graph):
session, scope = managed_graph
payload = replacement(scope); version = {'version': 'new', 'git_revision': 'new-commit'}
for _ in range(2):
with session.begin_transaction() as tx:
builder.write_graph_transaction(tx, payload, version=version)
tx.commit()
assert builder.graph_version(session) == version
nodes, relationships = builder.graph_counts(session)
assert dict(nodes) == {'Person': 1, 'Document': 1}
assert dict(relationships) == {'HAS_DOCUMENT': 1}

104
tools/index_state.py Normal file
View File

@ -0,0 +1,104 @@
"""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

15
tools/rag/build_chunks.py
View File

@ -920,19 +920,20 @@ def content_hash(
return hashlib.sha256(value).hexdigest()[:16]
def build_chunks() -> list[dict]:
if not PAGES_ROOT.exists():
def build_chunks(*, wiki_root: Path | None = None, output: Path | None = None) -> list[dict]:
pages = wiki_root / 'pages' if wiki_root is not None else PAGES_ROOT
if not pages.exists():
raise SystemExit(
f"Neexistuje priečinok: {PAGES_ROOT}"
f"Neexistuje priečinok: {pages}"
)
all_chunks: list[dict] = []
document_count = 0
for file_path in sorted(
PAGES_ROOT.glob("**/README.md")
pages.glob("**/README.md")
):
document = load_zpwiki_page(file_path)
document = load_zpwiki_page(file_path, wiki_root=wiki_root)
document_count += 1
drafts = chunk_markdown(
@ -968,7 +969,7 @@ def build_chunks() -> list[dict]:
)
write_json(
CHUNKS_FILE,
output or CHUNKS_FILE,
all_chunks,
)
@ -1008,7 +1009,7 @@ def build_chunks() -> list[dict]:
)
print(
f"[green]Výstup uložený do:[/green] "
f"{CHUNKS_FILE}"
f"{output or CHUNKS_FILE}"
)
if all_chunks:

16
tools/rag/build_sqlite_index.py
View File

@ -626,23 +626,25 @@ def temporary_database_path(
)
def build_database() -> dict[str, int]:
def build_database(*, documents_file: Path | None = None, chunks_file: Path | None = None,
db_file: Path | None = None) -> dict[str, int]:
target = db_file or DB_FILE
documents = read_json(
DOCUMENTS_FILE
documents_file or DOCUMENTS_FILE
)
chunks = read_json(
CHUNKS_FILE
chunks_file or CHUNKS_FILE
)
DB_FILE.parent.mkdir(
target.parent.mkdir(
parents=True,
exist_ok=True,
)
temporary_file = (
temporary_database_path(
DB_FILE
target
)
)
@ -698,7 +700,7 @@ def build_database() -> dict[str, int]:
os.replace(
temporary_file,
DB_FILE,
target,
)
except Exception:
@ -710,7 +712,7 @@ def build_database() -> dict[str, int]:
print(
"[green]SQLite index "
"vytvorený:[/green] "
f"{DB_FILE}"
f"{target}"
)
print(

4
tools/rag/common.py
View File

@ -328,6 +328,8 @@ def resolve_page_title(
def load_zpwiki_page(
file_path: Path,
*,
wiki_root: Path | None = None,
) -> dict[str, Any]:
post = frontmatter.load(file_path)
@ -366,7 +368,7 @@ def load_zpwiki_page(
return {
"path": str(
file_path.relative_to(ZPWIKI_ROOT)
file_path.relative_to(wiki_root or ZPWIKI_ROOT)
),
"title": title,
"categories": categories,

2
tools/rag/rag_utils.py
View File

@ -5,6 +5,7 @@ import json
import sqlite3
from pathlib import Path
from typing import Any
from tools.index_state import rag_snapshot
from tools.rag.rag_document_expansion import (
expand_results_with_exact_document_section,
@ -691,6 +692,7 @@ def _expand_exact_document(
return expand_results_with_exact_document_section(**kwargs)
@rag_snapshot
def build_rag_context(
db_path: Path,
query: str,

29
tools/rag/rebuild_index.py
View File

@ -21,10 +21,7 @@ if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from tools.rag.build_chunks import build_chunks
from tools.rag.build_sqlite_index import build_database
from tools.rag.common import DATA_DIR, DB_FILE, ZPWIKI_ROOT
from tools.rag.scan_zpwiki import scan_pages
from tools.rag.common import DATA_DIR, ZPWIKI_ROOT
REINDEX_LOCK_FILE = DATA_DIR / ".reindex.lock"
@ -144,22 +141,25 @@ def git_pull(
def rebuild_index(
pull_git: bool = False,
*,
lock_file: Path = REINDEX_LOCK_FILE,
lock_file: Path | None = None,
wiki_root: Path | None = None,
data_dir: Path | None = None,
) -> dict:
with acquire_reindex_lock(lock_file):
root = wiki_root or ZPWIKI_ROOT
target = data_dir or DATA_DIR
with acquire_reindex_lock(lock_file or target / '.reindex.lock'):
start = time.monotonic()
print(
f"[green]ZPWIKI_ROOT:[/green] "
f"{ZPWIKI_ROOT}"
f"{root}"
)
if pull_git:
git_pull()
git_pull(root)
documents = scan_pages()
chunks = build_chunks()
counts = build_database()
from tools.sync_indexes import synchronize_indexes
result = synchronize_indexes(root, target)
duration = round(
time.monotonic() - start,
@ -168,10 +168,7 @@ def rebuild_index(
return {
"duration_seconds": duration,
"documents_scanned": len(documents),
"chunks_created": len(chunks),
"counts": counts,
"database_path": str(DB_FILE),
**result,
}
@ -179,7 +176,7 @@ def main() -> None:
parser = argparse.ArgumentParser(
description=(
"Obnoví JSON súbory "
"a SQLite FTS5 index."
"a koordinovane SQLite aj Neo4j index."
)
)

13
tools/rag/scan_zpwiki.py
View File

@ -16,9 +16,10 @@ if str(PROJECT_ROOT) not in sys.path:
from tools.rag.common import DOCUMENTS_FILE, PAGES_ROOT, ZPWIKI_ROOT, load_zpwiki_page, write_json
def scan_pages() -> list[dict]:
if not PAGES_ROOT.exists():
raise SystemExit(f"Neexistuje priečinok: {PAGES_ROOT}")
def scan_pages(*, wiki_root: Path | None = None, output: Path | None = None) -> list[dict]:
pages = wiki_root / 'pages' if wiki_root is not None else PAGES_ROOT
if not pages.exists():
raise SystemExit(f"Neexistuje priečinok: {pages}")
documents = []
metadata_keys = Counter()
@ -26,8 +27,8 @@ def scan_pages() -> list[dict]:
tags = Counter()
authors = Counter()
for file_path in sorted(PAGES_ROOT.glob("**/README.md")):
page = load_zpwiki_page(file_path)
for file_path in sorted(pages.glob("**/README.md")):
page = load_zpwiki_page(file_path, wiki_root=wiki_root)
content = page.pop("content", "")
for key in page["metadata"]:
@ -50,7 +51,7 @@ def scan_pages() -> list[dict]:
}
)
write_json(DOCUMENTS_FILE, documents)
write_json(output or DOCUMENTS_FILE, documents)
print_summary(documents, metadata_keys, categories, tags, authors)
return documents

2
tools/rag/search_utils.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import sqlite3
from pathlib import Path
from typing import Any
from tools.index_state import rag_snapshot
# Tieto importy sú zámerne verejné.
# Zachovávajú spätnú kompatibilitu pre testy
@ -43,6 +44,7 @@ from tools.rag.search_core import (
)
@rag_snapshot
def search_database(
db_file: Path,
query: str,

47
tools/rag/tests/test_rebuild_index.py
View File

@ -54,40 +54,39 @@ def test_rebuild_runs_pipeline_in_correct_order(
monkeypatch.setattr(
rebuild,
"git_pull",
lambda: calls.append("git_pull"),
)
monkeypatch.setattr(
rebuild,
"scan_pages",
lambda: calls.append("scan") or [{"path": "a"}],
)
monkeypatch.setattr(
rebuild,
"build_chunks",
lambda: calls.append("chunks") or [{"chunk_id": "a::0"}],
)
monkeypatch.setattr(
rebuild,
"build_database",
lambda: calls.append("database")
or {
"documents": 1,
"chunks": 1,
"fts_chunks": 1,
"tags": 0,
"categories": 0,
},
lambda root: calls.append("git_pull"),
)
import tools.sync_indexes as sync
monkeypatch.setattr(sync, 'synchronize_indexes', lambda wiki, data:
calls.append('sqlite_and_neo4j') or {'documents_scanned': 1, 'chunks_created': 1,
'counts': {'fts_chunks': 1}, 'index_version': {'version': 'both'}})
result = rebuild.rebuild_index(
pull_git=True,
lock_file=tmp_path / "lock",
)
assert calls == ["git_pull", "scan", "chunks", "database"]
assert calls == ["git_pull", "sqlite_and_neo4j"]
assert result["documents_scanned"] == 1
assert result["chunks_created"] == 1
assert result["counts"]["fts_chunks"] == 1
assert result['index_version'] == {'version': 'both'}
def test_rebuild_pulls_and_locks_the_requested_snapshot_paths(tmp_path, monkeypatch):
import tools.sync_indexes as sync
wiki = tmp_path / 'alternate-wiki'
data = tmp_path / 'alternate-data'
seen = []
monkeypatch.setattr(rebuild, 'git_pull', lambda root: seen.append(('pull', root)))
def synchronized(root, target):
assert (data / '.reindex.lock').exists()
seen.append(('sync', root, target))
return {'index_version': {'version': 'alternate'}}
monkeypatch.setattr(sync, 'synchronize_indexes', synchronized)
result = rebuild.rebuild_index(pull_git=True, wiki_root=wiki, data_dir=data)
assert seen == [('pull', wiki), ('sync', wiki, data)]
assert result['index_version']['version'] == 'alternate'
def test_git_pull_uses_fast_forward_only(

167
tools/sync_indexes.py Normal file
View File

@ -0,0 +1,167 @@
"""Stage SQLite/JSON and publish them with a single Neo4j data transaction.
There is no distributed transaction between a filesystem and Neo4j. Cooperating
readers therefore use the publication barrier. Ordinary failures restore local
files; ambiguous commit/crash leaves a durable marker and readers fail closed.
Re-running sync creates a complete fresh generation and clears that marker.
"""
from __future__ import annotations
import hashlib
import json
import os
import shutil
import sqlite3
import subprocess
import tempfile
import uuid
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from tools.index_state import (PENDING_FILE, VERSION_FILE, durable_json, fsync_directory,
publication_lock, sqlite_version)
from tools.rag.scan_zpwiki import scan_pages
from tools.rag.build_chunks import build_chunks
from tools.rag.build_sqlite_index import build_database
from tools.rag.embedding_utils import embedding_model_name
from tools.graphrag.graphrag import build_graph_payload
from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings
from tools.graphrag.build_graphrag import (create_constraints, graph_version,
write_graph_transaction, graph_counts)
FILES = ('documents.json', 'chunks.json', 'zp_index.sqlite', VERSION_FILE)
def digest(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def source_files(wiki_root: Path) -> dict[str, str]:
if not (wiki_root / 'pages').is_dir():
raise RuntimeError(f'Chýba wiki priečinok pages: {wiki_root}')
return {str(p.relative_to(wiki_root)): digest(p)
for p in sorted((wiki_root / 'pages').rglob('README.md'))}
def git_revision(wiki_root: Path) -> str | None:
if not (wiki_root / '.git').exists():
return None
result = subprocess.run(['git', '-C', str(wiki_root), 'rev-parse', 'HEAD'],
capture_output=True, text=True, timeout=15, check=True)
return result.stdout.strip()
def prepare_snapshot(wiki_root: Path, stage: Path) -> tuple[dict, dict, dict]:
revision = git_revision(wiki_root); sources = source_files(wiki_root)
frozen_wiki = stage / 'wiki'
for relative in sources:
target = frozen_wiki / relative; target.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(wiki_root / relative, target)
if source_files(frozen_wiki) != sources:
raise RuntimeError('Wiki sa počas vytvárania snapshotu zmenila.')
documents = scan_pages(wiki_root=frozen_wiki, output=stage / 'documents.json')
chunks = build_chunks(wiki_root=frozen_wiki, output=stage / 'chunks.json')
counts = build_database(documents_file=stage / 'documents.json', chunks_file=stage / 'chunks.json',
db_file=stage / 'zp_index.sqlite')
version = {'version': uuid.uuid4().hex, 'git_revision': revision,
'created_at_utc': datetime.now(timezone.utc).isoformat(),
'source_sha256': hashlib.sha256(json.dumps(sources, sort_keys=True).encode()).hexdigest(),
'documents_sha256': digest(stage / 'documents.json'), 'chunks_sha256': digest(stage / 'chunks.json'),
'embedding_model': embedding_model_name()}
with closing(sqlite3.connect(stage / 'zp_index.sqlite')) as conn, conn:
conn.execute('CREATE TABLE index_state (id INTEGER PRIMARY KEY CHECK(id=1), metadata_json TEXT NOT NULL)')
conn.execute('INSERT INTO index_state VALUES (1, ?)', (json.dumps(version, sort_keys=True),))
durable_json(stage / VERSION_FILE, version)
if source_files(wiki_root) != sources or git_revision(wiki_root) != revision:
raise RuntimeError('Wiki sa počas indexovania zmenila; pripravené indexy sa neaktivujú.')
return version, counts, build_graph_payload(documents, chunks)
def replace_file(source: Path, target: Path) -> None:
"""Separate operation so tests can inject a publication I/O failure."""
os.replace(source, target)
def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, session) -> dict:
with publication_lock(data_dir, exclusive=True):
had_pending = (data_dir / PENDING_FILE).exists()
old_graph_version = graph_version(session)
originals = []
backup = stage / 'previous'; backup.mkdir()
for name in FILES:
current = data_dir / name
if current.exists():
shutil.copy2(current, backup / name)
with (backup / name).open('rb') as handle:
os.fsync(handle.fileno())
originals.append(name)
fsync_directory(backup)
durable_json(data_dir / PENDING_FILE, {'target_version': version['version'],
'staging_directory': stage.name, 'original_files': originals})
commit_attempted = False
try:
with session.begin_transaction() as tx:
write_graph_transaction(tx, payload, version=version)
node_counts, relation_counts = graph_counts(tx)
graph_summary = {'nodes': sum(n for _, n in node_counts),
'relationships': sum(n for _, n in relation_counts)}
for name in FILES:
# Flush staged contents before the durable directory switch.
with (stage / name).open('rb') as handle:
os.fsync(handle.fileno())
replace_file(stage / name, data_dir / name)
fsync_directory(data_dir)
commit_attempted = True
tx.commit()
except BaseException:
if commit_attempted:
# A lost COMMIT response is not proof of rollback. Never restore
# old SQLite against a possibly committed new graph.
try:
current_graph_version = graph_version(session)
except Exception:
raise # Keep marker + backups; reads stay unavailable.
if current_graph_version == version and sqlite_version(data_dir / 'zp_index.sqlite') == version:
(data_dir / PENDING_FILE).unlink(); fsync_directory(data_dir)
return graph_summary
if current_graph_version != old_graph_version:
raise # Unknown external writer; fail closed.
for name in FILES:
if name in originals:
os.replace(backup / name, data_dir / name)
else:
(data_dir / name).unlink(missing_ok=True)
fsync_directory(data_dir)
if not had_pending:
(data_dir / PENDING_FILE).unlink(); fsync_directory(data_dir)
raise
(data_dir / PENDING_FILE).unlink()
fsync_directory(data_dir)
return graph_summary
def synchronize_indexes(wiki_root: Path, data_dir: Path) -> dict:
data_dir.mkdir(parents=True, exist_ok=True)
stage = Path(tempfile.mkdtemp(prefix='.sync-stage-', dir=data_dir))
try:
version, counts, payload = prepare_snapshot(wiki_root, stage)
settings = get_neo4j_settings()
with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session:
driver.verify_connectivity()
create_constraints(session)
graph_summary = publish_snapshot(data_dir, stage, version, payload, session)
return {'counts': counts, 'documents_scanned': counts['documents'], 'chunks_created': counts['chunks'],
'database_path': str(data_dir / 'zp_index.sqlite'), 'index_version': version,
'graph_counts': graph_summary}
finally:
# Keep recovery material when a crash/ambiguous commit is unresolved.
pending = data_dir / PENDING_FILE
keep_stage = False
if pending.exists():
try:
keep_stage = json.loads(pending.read_text()).get('staging_directory') == stage.name
except (ValueError, OSError):
keep_stage = True
if not keep_stage:
shutil.rmtree(stage)