From bfe8cbba1bac53071d47c489878986982d0fc805 Mon Sep 17 00:00:00 2001 From: jp170na Date: Mon, 26 Oct 2026 13:08:45 +0100 Subject: [PATCH] =?UTF-8?q?Automatick=C3=A1=20synchroniz=C3=A1cia=20wiki?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 79 ++++--- app/main.py | 15 +- app/routes.py | 209 +++++------------- app/security.py | 2 + docker-compose.yml | 17 ++ test/conftest.py | 18 ++ test/test_api.py | 295 ++++++-------------------- test/test_index_sync.py | 13 +- test/test_sync_jobs.py | 147 +++++++++++++ test/test_wiki_git.py | 81 +++++++ tools/rag/rebuild_index.py | 62 ++---- tools/rag/tests/conftest.py | 6 + tools/rag/tests/test_rebuild_index.py | 39 ++-- tools/sync_indexes.py | 4 +- tools/sync_jobs.py | 173 +++++++++++++++ tools/wiki_git.py | 67 ++++++ 16 files changed, 740 insertions(+), 487 deletions(-) create mode 100644 test/test_sync_jobs.py create mode 100644 test/test_wiki_git.py create mode 100644 tools/sync_jobs.py create mode 100644 tools/wiki_git.py diff --git a/README.md b/README.md index beb1149..bff97f4 100644 --- a/README.md +++ b/README.md @@ -583,8 +583,11 @@ V koreňovom priečinku vytvor `.env`: WEBHOOK_SECRET= SYNC_API_KEY= SEARCH_API_KEY=<ďalšia náhodná hodnota s minimálne 32 znakmi> -EXPECTED_GITEA_REPOSITORY=KEMT/zpwiki -WEBHOOK_PULL_GIT=false +EXPECTED_GITEA_REPOSITORY=jp170na/zpwiki +EXPECTED_GITEA_BRANCH=master +WEBHOOK_PULL_GIT=true +ZPWIKI_SSH_KEY=/absolutna/cesta/k/ssh_klucu +ZPWIKI_KNOWN_HOSTS=/absolutna/cesta/k/known_hosts # Voliteľné EMBEDDING_MODEL=intfloat/multilingual-e5-small @@ -980,51 +983,77 @@ tools/rag/evaluation/json_files/rag_answer_overrides.json Pôvodný frozen `questions.json` tým ostáva nezmenený. -## Manuálna synchronizácia +## Manuálna synchronizácia a stav úloh -Manuálne reindexovanie cez zabezpečený endpoint: +`POST /sync` uloží požiadavku do persistentnej fronty a ihneď vráti HTTP 202. +Indexovanie prebieha na pozadí; prijatie požiadavky ešte neznamená dokončenie. ```bash curl -X POST http://127.0.0.1:8000/sync \ -H "Content-Type: application/json" \ -H "X-API-Key: $SYNC_API_KEY" \ - -d '{"pull_git": false}' + -d '{"pull_git": true}' ``` -## Gitea webhook +Odpoveď obsahuje `job_id` a `status_url`, napríklad `/sync/jobs/`. +Stav načítaj rovnakým `SYNC_API_KEY`: -Endpoint: - -```text -POST /webhook/gitea +```bash +curl http://127.0.0.1:8000/sync/jobs/ -H "X-API-Key: $SYNC_API_KEY" +curl http://127.0.0.1:8000/sync/status -H "X-API-Key: $SYNC_API_KEY" ``` -Webhook overuje: +Úloha má stav `queued`, `running`, `succeeded` alebo `failed`. Pole `phase` +rozlišuje čakanie na zámok, Git, budovanie a publikovanie indexov. Úspešná +úloha obsahuje v `result` počty, trvanie a spoločnú verziu SQLite a Neo4j. +Prehľad obsahuje aktívne úlohy, posledný úspech a posledné zlyhanie. -- HMAC-SHA256 podpis, -- typ Gitea udalosti, -- očakávaný repozitár, -- maximálnu veľkosť payloadu, -- stav reindexovacieho zámku. +Fronta `data/sync_jobs.sqlite` prežije reštart. Procesový zámok povoľuje +iba jedného pracovníka; rozpracovaná úloha sa po páde zopakuje. Push prijatý +po začiatku obnovy vytvorí následnú úlohu. Ďalšie čakajúce požiadavky sa +zlúčia, pričom požiadavka na stiahnutie z Gitu sa zachová. Konflikt s CLI +obnovou spôsobí čakanie, nie stratu požiadavky. Bežná chyba sa skúsi najviac +trikrát; následne zostane viditeľný stav `failed` a možno poslať nový `/sync`. +CLI `python -m tools.rag.rebuild_index --pull` zostáva synchronný. -Očakávaný repozitár: - -```dotenv -EXPECTED_GITEA_REPOSITORY=KEMT/zpwiki -``` - -Ak je povolené: +## Gitea webhook a prístup ku Gitu + +Endpoint `POST /webhook/gitea` overuje HMAC-SHA256 nad pôvodným telom, +veľkosť payloadu, udalosť `push`, repozitár a konkrétnu vetvu z `ref`: ```dotenv +EXPECTED_GITEA_REPOSITORY=jp170na/zpwiki +EXPECTED_GITEA_BRANCH=master WEBHOOK_PULL_GIT=true ``` -pred reindexovaním sa vykoná: +Push inej vetvy a zmazanie vetvy sa ignorujú bez obnovy. `X-Gitea-Delivery` +zabraňuje opakovanému zaradeniu rovnakého doručenia. Platný push vráti HTTP +202 s odkazom na stav úlohy; nečaká na dokončenie indexovania. + +Git aktualizácia overí zhodu `origin` s očakávaným repozitárom, aktívnu +vetvu a neprítomnosť lokálnych zmien vrátane nesledovaných súborov. Potom vykoná: ```bash -git pull --ff-only +git fetch --no-tags --no-recurse-submodules origin refs/heads/master:refs/remotes/origin/master +git merge --ff-only refs/remotes/origin/master ``` +Cieľom je aktuálna verzia nakonfigurovanej vetvy; oneskorený webhook +neprepína wiki naspäť na starší commit. Odlišná história sa automaticky +neprepisuje. Pri chybe Gitu zostávajú predchádzajúce indexy zachované. + +Pre tento wiki server treba SSH. V `.env` nastav absolútne hostiteľské cesty +`ZPWIKI_SSH_KEY` a `ZPWIKI_KNOWN_HOSTS`. Compose pripája iba tieto dva súbory +na čítanie do `/run/secrets/`; kľúč sa nekopíruje do image ani repozitára. +Použi kľúč s prístupom k wiki, ideálne deploy kľúč iba na čítanie, a vopred +over záznam servera v `known_hosts`. SSH používa `BatchMode` a povinné +striktné overenie servera; pri chybe si interaktívne nepýta heslo. + +Na Gitea nastav push webhook na verejne dosiahnuteľnú adresu +`https:///webhook/gitea` s rovnakým `WEBHOOK_SECRET`. Lokálna +adresa `127.0.0.1` nie je zo vzdialeného Gitea servera dosiahnuteľná. + ## Bezpečnosť Vyhľadávanie podporuje API key: diff --git a/app/main.py b/app/main.py index 45389f0..73a7fea 100644 --- a/app/main.py +++ b/app/main.py @@ -35,6 +35,9 @@ from app.security import ( ) from tools.rag.embedding_utils import embed_query from tools.retrieval_contract import error_result +from tools.rag.common import DATA_DIR +from tools.sync_jobs import SyncJobs +from tools.rag.rebuild_index import rebuild_index logger = logging.getLogger(__name__) @@ -66,7 +69,17 @@ async def lifespan( "úspešného warm-upu." ) - yield + def run_sync(**kwargs): + result = rebuild_index(**kwargs) + application.state.embedding_ready = True + return result + + application.state.sync_jobs = SyncJobs(DATA_DIR, runner=run_sync) + application.state.sync_jobs.start() + try: + yield + finally: + await asyncio.to_thread(application.state.sync_jobs.stop) app = FastAPI( diff --git a/app/routes.py b/app/routes.py index a0bbccf..5bb5ea1 100644 --- a/app/routes.py +++ b/app/routes.py @@ -1,8 +1,8 @@ from __future__ import annotations -import asyncio import json import logging +import re import sqlite3 from typing import ( Any, @@ -41,11 +41,8 @@ from tools.rag.common import ( ZPWIKI_ROOT, ) from tools.rag.rag_utils import build_rag_context -from tools.rag.rebuild_index import ( - ReindexInProgressError, - rebuild_index, -) from tools.rag.search_utils import search_database +from tools.wiki_git import expected_branch from tools.publication import DEFAULT_PUBLISHED_ONLY from tools.tool_health import tools_readiness from tools.retrieval_contract import RetrievalResponse, RetrievalSource, RetrievalErrorResponse @@ -75,10 +72,6 @@ INTERNAL_ERROR_DETAIL = ( "Nastala interná chyba servera." ) -REINDEX_FAILED_DETAIL = ( - "Reindexovanie zlyhalo." -) - WEBHOOK_CONFIGURATION_DETAIL = ( "Webhook nie je správne nakonfigurovaný." ) @@ -164,8 +157,8 @@ class SyncRequest( pull_git: bool = Field( default=False, description=( - "Pred reindexovaním vykoná " - "git pull --ff-only." + "Pred reindexovaním stiahne očakávanú vetvu " + "a aktualizuje ju iba fast-forward." ), ) @@ -255,59 +248,34 @@ class SearchResponse(RetrievalResponse): index_version: dict[str, Any] | None = None -class SyncResponse(BaseModel): - status: Literal[ - "ok" - ] - +class SyncAcceptedResponse(BaseModel): + status: Literal['accepted'] + job_id: str + status_url: str pull_git: bool - - duration_seconds: float - - counts: dict[ - str, - int, - ] - index_version: dict[str, Any] | None = None - graph_counts: dict[str, int] | None = None + coalesced: bool + duplicate: bool + event: str | None = None + repository: str | None = None + verified_by: Literal['hmac_sha256'] | None = None -class WebhookSuccessResponse( - BaseModel -): - status: Literal[ - "ok" - ] - +class WebhookIgnoredResponse(BaseModel): + status: Literal['ignored'] + reason: Literal['unsupported_event', 'unexpected_branch', 'deleted_branch'] event: str - repository: str - - verified_by: Literal[ - "hmac_sha256" - ] - - duration_seconds: float - - counts: dict[ - str, - int, - ] - index_version: dict[str, Any] | None = None - graph_counts: dict[str, int] | None = None -class WebhookIgnoredResponse( - BaseModel -): - status: Literal[ - "ignored" - ] - - reason: Literal[ - "unsupported_event" - ] - - event: str +def enqueue_sync(request: Request, **kwargs) -> dict: + try: + accepted = request.app.state.sync_jobs.enqueue(**kwargs) + except (sqlite3.Error, OSError): + logger.exception('Prijatie synchronizácie do fronty zlyhalo.') + raise HTTPException(status_code=503, detail='Synchronizačná fronta nie je dostupná') + job = accepted['job'] + return {'status': 'accepted', 'job_id': job['id'], + 'status_url': f"/sync/jobs/{job['id']}", 'pull_git': job['pull_git'], + 'coalesced': accepted['coalesced'], 'duplicate': accepted['duplicate']} def raise_retrieval_error( @@ -366,36 +334,6 @@ def raise_retrieval_error( ) from error -def raise_reindex_error( - error: Exception, -) -> NoReturn: - if isinstance( - error, - ReindexInProgressError, - ): - raise HTTPException( - status_code=( - status.HTTP_409_CONFLICT - ), - detail=str( - error - ), - ) from error - - logger.exception( - "Reindexovanie zlyhalo." - ) - - raise HTTPException( - status_code=( - status.HTTP_500_INTERNAL_SERVER_ERROR - ), - detail=( - REINDEX_FAILED_DETAIL - ), - ) from error - - def webhook_secret() -> str: try: return validate_secret( @@ -709,47 +647,34 @@ def search( ) ], include_in_schema=False, - response_model=SyncResponse, + response_model=SyncAcceptedResponse, + status_code=202, ) def sync( request: SyncRequest, http_request: Request, ) -> dict[str, Any]: - try: - result = rebuild_index( - pull_git=( - request.pull_git - ) - ) - http_request.app.state.embedding_ready = True + return enqueue_sync(http_request, pull_git=request.pull_git) - except Exception as error: - raise_reindex_error( - error - ) - return { - "status": "ok", - "pull_git": ( - request.pull_git - ), - "index_version": result.get('index_version'), - "graph_counts": result.get('graph_counts'), - "duration_seconds": ( - result[ - "duration_seconds" - ] - ), - "counts": result[ - "counts" - ], - } +@router.get('/sync/status', dependencies=[Depends(require_sync_api_key)], include_in_schema=False) +def sync_status(request: Request) -> dict: + return request.app.state.sync_jobs.overview() + + +@router.get('/sync/jobs/{job_id}', dependencies=[Depends(require_sync_api_key)], include_in_schema=False) +def sync_job(job_id: str, request: Request) -> dict: + job = request.app.state.sync_jobs.get(job_id) + if job is None: + raise HTTPException(status_code=404, detail='Synchronizačná úloha neexistuje') + return job @router.post( "/webhook/gitea", + status_code=202, response_model=( - WebhookSuccessResponse + SyncAcceptedResponse | WebhookIgnoredResponse ), response_model_exclude_unset=True, @@ -893,39 +818,21 @@ async def gitea_webhook( ), ) - try: - result = await asyncio.to_thread( - rebuild_index, - pull_git=( - webhook_should_pull_git() - ), - ) - request.app.state.embedding_ready = True - - except Exception as error: - raise_reindex_error( - error - ) - + ref = payload.get('ref') + if not isinstance(ref, str) or not ref.startswith('refs/heads/'): + raise HTTPException(status_code=400, detail='Push payload musí obsahovať ref vetvy') + if ref != f'refs/heads/{expected_branch()}': + return {'status': 'ignored', 'reason': 'unexpected_branch', 'event': x_gitea_event} + revision = payload.get('after') + if payload.get('deleted') is True or (isinstance(revision, str) and revision and set(revision) == {'0'}): + return {'status': 'ignored', 'reason': 'deleted_branch', 'event': x_gitea_event} + if revision is not None and (not isinstance(revision, str) or not re.fullmatch(r'[0-9a-fA-F]{40}|[0-9a-fA-F]{64}', revision)): + raise HTTPException(status_code=400, detail='Neplatná Git revízia v push payload') + delivery = request.headers.get('X-Gitea-Delivery') + if delivery is not None and (not delivery.strip() or len(delivery) > 200 or any(ord(c) < 32 for c in delivery)): + raise HTTPException(status_code=400, detail='Neplatné ID webhook doručenia') return { - "status": "ok", - "event": ( - x_gitea_event - ), - "repository": ( - repository_name - ), - "verified_by": ( - "hmac_sha256" - ), - "index_version": result.get('index_version'), - "graph_counts": result.get('graph_counts'), - "duration_seconds": ( - result[ - "duration_seconds" - ] - ), - "counts": result[ - "counts" - ], + **enqueue_sync(request, pull_git=webhook_should_pull_git(), requested_revision=revision, + delivery_key=f'{repository_name.casefold()}:{delivery}' if delivery else None), + 'event': x_gitea_event, 'repository': repository_name, 'verified_by': 'hmac_sha256', } diff --git a/app/security.py b/app/security.py index 12ac494..1dbc95e 100644 --- a/app/security.py +++ b/app/security.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import hmac import os +from tools.wiki_git import expected_branch from typing import Any from fastapi import ( @@ -182,6 +183,7 @@ def validate_security_configuration() -> None: ) expected_gitea_repository() + expected_branch() # Validujeme aj voliteľnú boolean # konfiguráciu už pri štarte aplikácie. diff --git a/docker-compose.yml b/docker-compose.yml index fd7e4fe..b489da2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,6 +15,10 @@ services: CHUNK_MIN_TOKENS: "80" CHUNK_TOKEN_ENCODING: cl100k_base HF_HOME: /cache/huggingface + GIT_SSH_COMMAND: >- + ssh -F /dev/null -i /run/secrets/zpwiki_ssh_key + -o IdentitiesOnly=yes -o BatchMode=yes -o StrictHostKeyChecking=yes + -o UserKnownHostsFile=/run/secrets/zpwiki_known_hosts -o ConnectTimeout=15 NEO4J_URI: bolt://neo4j:7687 NEO4J_USER: ${NEO4J_USER:-neo4j} NEO4J_PASSWORD: ${NEO4J_PASSWORD} @@ -24,8 +28,21 @@ services: - ./data:/app/data - ../zpwiki:/zpwiki - hf-cache:/cache/huggingface + - type: bind + source: ${ZPWIKI_SSH_KEY:-${HOME}/.ssh/id_ed25519} + target: /run/secrets/zpwiki_ssh_key + read_only: true + bind: + create_host_path: false + - type: bind + source: ${ZPWIKI_KNOWN_HOSTS:-${HOME}/.ssh/known_hosts} + target: /run/secrets/zpwiki_known_hosts + read_only: true + bind: + create_host_path: false restart: unless-stopped + stop_grace_period: 5m neo4j: image: neo4j:2026.09.0 diff --git a/test/conftest.py b/test/conftest.py index bf67c00..665bac1 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -41,3 +41,21 @@ def security_environment( "WEBHOOK_PULL_GIT", "false", ) + monkeypatch.setenv('EXPECTED_GITEA_BRANCH', 'master') + + +@pytest.fixture(autouse=True) +def isolated_sync_queue(monkeypatch, tmp_path): + """API tests never start a real index rebuild or share production job state.""" + import app.main as main + from tools.sync_jobs import SyncJobs + + class PausedJobs(SyncJobs): + def __init__(self, data_dir, **kwargs): + super().__init__(tmp_path / 'jobs', runner=lambda **kw: + main.rebuild_index(pull_git=kw['pull_git'])) + + def start(self): + pass + + monkeypatch.setattr(main, 'SyncJobs', PausedJobs) diff --git a/test/test_api.py b/test/test_api.py index 1f80c68..1eaece7 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -482,126 +482,40 @@ def test_sync_rejects_wrong_api_key( ) -def test_sync_accepts_valid_api_key( - client: TestClient, - monkeypatch: pytest.MonkeyPatch, -) -> None: - monkeypatch.setattr( - routes, - "rebuild_index", - lambda pull_git=False: ( - fake_rebuild_result() - ), - ) - - response = client.post( - "/sync", - headers={ - "X-API-Key": ( - SYNC_API_KEY - ), - }, - json={ - "pull_git": False, - }, - ) - - assert ( - response.status_code - == 200 - ) - - assert ( - response.json()[ - "status" - ] - == "ok" - ) +def test_sync_accepts_valid_api_key(client, monkeypatch): + monkeypatch.setattr(main, 'rebuild_index', lambda **kw: fake_rebuild_result()) + response = client.post('/sync', headers={'X-API-Key': SYNC_API_KEY}, json={'pull_git': False}) + assert response.status_code == 202 + assert response.json()['status'] == 'accepted' + assert main.app.state.sync_jobs.run_once() + result = client.get(response.json()['status_url'], headers={'X-API-Key': SYNC_API_KEY}) + assert result.json()['status'] == 'succeeded' + assert result.json()['result'] == fake_rebuild_result() -def test_sync_returns_conflict_when_reindex_is_running( - client: TestClient, - monkeypatch: pytest.MonkeyPatch, -) -> None: - def busy( - *args, - **kwargs, - ): - raise ( - ReindexInProgressError( - "Reindexovanie už prebieha" - ) - ) - - monkeypatch.setattr( - routes, - "rebuild_index", - busy, - ) - - response = client.post( - "/sync", - headers={ - "X-API-Key": ( - SYNC_API_KEY - ), - }, - json={ - "pull_git": False, - }, - ) - - assert ( - response.status_code - == 409 - ) +def test_sync_waits_when_reindex_is_running(client, monkeypatch): + def busy(**kw): raise ReindexInProgressError('Reindexovanie už prebieha') + monkeypatch.setattr(main, 'rebuild_index', busy) + response = client.post('/sync', headers={'X-API-Key': SYNC_API_KEY}, json={}) + assert response.status_code == 202 + main.app.state.sync_jobs.run_once() + job = client.get(response.json()['status_url'], headers={'X-API-Key': SYNC_API_KEY}).json() + assert job['status'] == 'queued' and job['phase'] == 'waiting_for_lock' -def test_sync_returns_generic_500_when_rebuild_fails( - client: TestClient, - monkeypatch: pytest.MonkeyPatch, -) -> None: - def failed_rebuild( - *args, - **kwargs, - ): - raise RuntimeError( - "private rebuild error" - ) - - monkeypatch.setattr( - routes, - "rebuild_index", - failed_rebuild, - ) - - response = client.post( - "/sync", - headers={ - "X-API-Key": ( - SYNC_API_KEY - ), - }, - json={ - "pull_git": False, - }, - ) - - assert ( - response.status_code - == 500 - ) - - assert response.json()[ - "detail" - ] == ( - routes.REINDEX_FAILED_DETAIL - ) - - assert ( - "private rebuild error" - not in response.text - ) +def test_sync_records_failure_without_exposing_private_error(client, monkeypatch): + def fail(**kw): raise RuntimeError('private rebuild error') + monkeypatch.setattr(main, 'rebuild_index', fail) + response = client.post('/sync', headers={'X-API-Key': SYNC_API_KEY}, json={}) + assert response.status_code == 202 + queue = main.app.state.sync_jobs + for _ in range(3): + with queue.connect() as conn: conn.execute('UPDATE jobs SET available_at=0') + queue.run_once() + status = client.get(response.json()['status_url'], headers={'X-API-Key': SYNC_API_KEY}) + assert status.json()['status'] == 'failed' + assert status.json()['reason'] == 'sync_failed' + assert 'private rebuild error' not in status.text def test_webhook_rejects_invalid_signature( @@ -609,6 +523,7 @@ def test_webhook_rejects_invalid_signature( ) -> None: body = json.dumps( { + "ref": "refs/heads/master", "repository": { "full_name": ( "KEMT/zpwiki" @@ -675,6 +590,7 @@ def test_webhook_requires_event_header( ) -> None: body = json.dumps( { + "ref": "refs/heads/master", "repository": { "full_name": ( "KEMT/zpwiki" @@ -725,13 +641,14 @@ def test_webhook_ignores_non_push_event( ) monkeypatch.setattr( - routes, + main, "rebuild_index", fake_rebuild, ) body = json.dumps( { + "ref": "refs/heads/master", "repository": { "full_name": ( "KEMT/zpwiki" @@ -780,6 +697,7 @@ def test_webhook_rejects_unexpected_repository( ) -> None: body = json.dumps( { + "ref": "refs/heads/master", "repository": { "full_name": ( "OTHER/repository" @@ -819,7 +737,7 @@ def test_webhook_accepts_signed_push( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( - routes, + main, "rebuild_index", lambda pull_git=False: ( fake_rebuild_result() @@ -828,6 +746,7 @@ def test_webhook_accepts_signed_push( body = json.dumps( { + "ref": "refs/heads/master", "repository": { "full_name": ( "KEMT/zpwiki" @@ -858,7 +777,7 @@ def test_webhook_accepts_signed_push( assert ( response.status_code - == 200 + == 202 ) payload = response.json() @@ -878,122 +797,26 @@ def test_webhook_accepts_signed_push( ) -def test_webhook_returns_conflict_when_reindex_is_running( - client: TestClient, - monkeypatch: pytest.MonkeyPatch, -) -> None: - def busy( - *args, - **kwargs, - ): - raise ( - ReindexInProgressError( - "Reindexovanie už prebieha" - ) - ) - - monkeypatch.setattr( - routes, - "rebuild_index", - busy, - ) - - body = json.dumps( - { - "repository": { - "full_name": ( - "KEMT/zpwiki" - ), - } - } - ).encode( - "utf-8" - ) - - response = client.post( - "/webhook/gitea", - content=body, - headers={ - "Content-Type": ( - "application/json" - ), - "X-Gitea-Event": ( - "push" - ), - "X-Gitea-Signature": ( - sign( - body - ) - ), - }, - ) - - assert ( - response.status_code - == 409 - ) +def test_webhook_queues_while_reindex_is_running(client, monkeypatch): + def busy(**kw): raise ReindexInProgressError('Reindexovanie už prebieha') + monkeypatch.setattr(main, 'rebuild_index', busy) + body = json.dumps({'repository': {'full_name': 'KEMT/zpwiki'}, 'ref': 'refs/heads/master'}).encode() + response = client.post('/webhook/gitea', content=body, + headers={'X-Gitea-Event': 'push', 'X-Gitea-Signature': sign(body)}) + assert response.status_code == 202 + main.app.state.sync_jobs.run_once() + job = main.app.state.sync_jobs.get(response.json()['job_id']) + assert job['status'] == 'queued' and job['phase'] == 'waiting_for_lock' -def test_webhook_returns_generic_500_when_rebuild_fails( - client: TestClient, - monkeypatch: pytest.MonkeyPatch, -) -> None: - def failed_rebuild( - *args, - **kwargs, - ): - raise RuntimeError( - "private webhook error" - ) - - monkeypatch.setattr( - routes, - "rebuild_index", - failed_rebuild, - ) - - body = json.dumps( - { - "repository": { - "full_name": ( - "KEMT/zpwiki" - ), - } - } - ).encode( - "utf-8" - ) - - response = client.post( - "/webhook/gitea", - content=body, - headers={ - "Content-Type": ( - "application/json" - ), - "X-Gitea-Event": ( - "push" - ), - "X-Gitea-Signature": ( - sign( - body - ) - ), - }, - ) - - assert ( - response.status_code - == 500 - ) - - assert response.json()[ - "detail" - ] == ( - routes.REINDEX_FAILED_DETAIL - ) - - assert ( - "private webhook error" - not in response.text - ) +def test_webhook_rebuild_failure_is_visible_in_job(client, monkeypatch): + def fail(**kw): raise RuntimeError('private webhook error') + monkeypatch.setattr(main, 'rebuild_index', fail) + body = json.dumps({'repository': {'full_name': 'KEMT/zpwiki'}, 'ref': 'refs/heads/master'}).encode() + response = client.post('/webhook/gitea', content=body, + headers={'X-Gitea-Event': 'push', 'X-Gitea-Signature': sign(body)}) + assert response.status_code == 202 + main.app.state.sync_jobs.run_once() + status = client.get(response.json()['status_url'], headers={'X-API-Key': SYNC_API_KEY}) + assert status.json()['phase'] == 'retry_wait' + assert 'private webhook error' not in status.text diff --git a/test/test_index_sync.py b/test/test_index_sync.py index 8b05ff7..d70e9c0 100644 --- a/test/test_index_sync.py +++ b/test/test_index_sync.py @@ -222,17 +222,20 @@ def test_sync_api_exposes_common_version_and_graph_counts(monkeypatch, endpoint) 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: { + monkeypatch.setattr(main, '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() + body = json.dumps({'repository': {'full_name': 'KEMT/zpwiki'}, 'ref': 'refs/heads/master'}).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} + assert response.status_code == 202 + main.app.state.sync_jobs.run_once() + job = client.get(response.json()['status_url'], headers={'X-API-Key': 's' * 64}).json() + assert job['status'] == 'succeeded' + assert job['result']['index_version'] == version + assert job['result']['graph_counts'] == {'nodes': 2, 'relationships': 1} diff --git a/test/test_sync_jobs.py b/test/test_sync_jobs.py new file mode 100644 index 0000000..347066e --- /dev/null +++ b/test/test_sync_jobs.py @@ -0,0 +1,147 @@ +import hashlib +import hmac +import json +import threading +import time +from concurrent.futures import ThreadPoolExecutor + +import pytest +from fastapi.testclient import TestClient + +import app.main as main +from tools.sync_jobs import SyncJobs +from tools.rag.rebuild_index import acquire_reindex_lock + + +def test_concurrent_enqueues_are_durable_and_coalesced(tmp_path): + jobs = SyncJobs(tmp_path) + with ThreadPoolExecutor(max_workers=8) as pool: + accepted = list(pool.map(lambda n: jobs.enqueue(pull_git=n % 2 == 0, + delivery_key=f'delivery-{n}', requested_revision=str(n)), range(24))) + assert len({row['job']['id'] for row in accepted}) == 1 + job = SyncJobs(tmp_path).get(accepted[0]['job']['id']) + assert job['request_count'] == 24 and job['pull_git'] + duplicate = SyncJobs(tmp_path).enqueue(pull_git=True, delivery_key='delivery-1') + assert duplicate['duplicate'] and duplicate['job']['request_count'] == 24 + + +def test_push_during_running_job_schedules_followup_and_workers_do_not_overlap(tmp_path): + entered, release = threading.Event(), threading.Event() + phases = [] + def runner(**kw): + kw['progress']('building_indexes') + phases.append('run') + entered.set() + assert release.wait(5) + return {'index_version': {'version': str(len(phases))}} + jobs = SyncJobs(tmp_path, runner=runner) + first = jobs.enqueue(pull_git=True)['job']['id'] + worker = threading.Thread(target=jobs.run_once) + worker.start() + try: + assert entered.wait(5) + assert jobs.get(first)['phase'] == 'building_indexes' + second = jobs.enqueue(pull_git=True)['job']['id'] + third = jobs.enqueue(pull_git=True)['job']['id'] + assert second != first and third == second + assert not SyncJobs(tmp_path, runner=runner).run_once() + finally: + release.set() + worker.join(5) + assert jobs.get(first)['status'] == 'succeeded' + assert jobs.run_once() and jobs.get(second)['status'] == 'succeeded' + assert len(phases) == 2 + + +def test_interrupted_job_recovers_and_preserves_delivery_dedup(tmp_path): + jobs = SyncJobs(tmp_path) + job_id = jobs.enqueue(pull_git=True, delivery_key='one')['job']['id'] + with jobs.connect() as conn: + conn.execute("UPDATE jobs SET status='running',phase='publishing_indexes'") + recovered = SyncJobs(tmp_path, runner=lambda **kw: {'index_version': {'version': 'recovered'}}) + assert recovered.run_once() + assert recovered.get(job_id)['status'] == 'succeeded' + assert recovered.enqueue(pull_git=True, delivery_key='one')['duplicate'] + + +def test_cli_lock_defers_instead_of_losing_job(tmp_path): + def runner(**kw): + with acquire_reindex_lock(tmp_path / '.reindex.lock'): + return {'counts': {'documents': 1}} + jobs = SyncJobs(tmp_path, runner=runner) + job_id = jobs.enqueue(pull_git=False)['job']['id'] + with acquire_reindex_lock(tmp_path / '.reindex.lock'): + assert jobs.run_once() + job = jobs.get(job_id) + assert job['status'] == 'queued' and job['attempts'] == 0 + with jobs.connect() as conn: + conn.execute('UPDATE jobs SET available_at=0') + jobs.run_once() + assert jobs.get(job_id)['status'] == 'succeeded' + + +def test_background_worker_processes_and_stops(tmp_path): + completed = threading.Event() + def runner(**kw): + completed.set() + return {'counts': {'documents': 1}} + jobs = SyncJobs(tmp_path, runner=runner) + job_id = jobs.enqueue(pull_git=False)['job']['id'] + jobs.start() + try: + assert completed.wait(5) + finally: + jobs.stop() + assert not jobs.thread.is_alive() and jobs.get(job_id)['status'] == 'succeeded' + + +@pytest.fixture +def api(monkeypatch): + monkeypatch.setattr(main, 'embed_query', lambda _: None) + with TestClient(main.app) as client: + yield client + + +def post_push(api, *, ref='refs/heads/master', delivery='one', **extra): + body = json.dumps({'repository': {'full_name': 'KEMT/zpwiki'}, 'ref': ref, + 'after': 'a' * 40, **extra}).encode() + signature = hmac.new(('w' * 64).encode(), body, hashlib.sha256).hexdigest() + return api.post('/webhook/gitea', content=body, headers={ + 'X-Gitea-Event': 'push', 'X-Gitea-Signature': signature, 'X-Gitea-Delivery': delivery}) + + +def test_api_returns_before_slow_rebuild_and_rejects_unauthed_status(api, monkeypatch): + import app.routes as routes + def forbidden(**kw): + pytest.fail('HTTP request must not execute the rebuild') + monkeypatch.setattr(main, 'rebuild_index', forbidden) + response = post_push(api) + assert response.status_code == 202 + assert api.get(response.json()['status_url']).status_code == 401 + assert api.get('/sync/status').status_code == 401 + assert api.get(response.json()['status_url'], headers={'X-API-Key': 's' * 64}).json()['status'] == 'queued' + assert api.get('/sync/jobs/missing', headers={'X-API-Key': 's' * 64}).status_code == 404 + + +def test_api_deduplicates_delivery_and_enables_git_pull(api, monkeypatch): + monkeypatch.setenv('WEBHOOK_PULL_GIT', 'true') + first = post_push(api) + repeated = post_push(api) + assert first.json()['job_id'] == repeated.json()['job_id'] + assert repeated.json()['duplicate'] and first.json()['pull_git'] + assert main.app.state.sync_jobs.get(first.json()['job_id'])['requested_revision'] == 'a' * 40 + + +@pytest.mark.parametrize('extra,reason', [ + ({'ref': 'refs/heads/other'}, 'unexpected_branch'), + ({'deleted': True}, 'deleted_branch'), ({'after': '0' * 40}, 'deleted_branch')]) +def test_other_branches_and_deletions_do_not_queue(api, extra, reason): + response = post_push(api, **extra) + assert response.status_code == 202 and response.json()['reason'] == reason + assert main.app.state.sync_jobs.overview()['active'] == [] + + +@pytest.mark.parametrize('extra', [{'ref': None}, {'ref': 'refs/tags/v1'}, {'after': 'not-a-commit'}]) +def test_malformed_push_is_rejected(api, extra): + assert post_push(api, **extra).status_code == 400 + assert main.app.state.sync_jobs.overview()['active'] == [] diff --git a/test/test_wiki_git.py b/test/test_wiki_git.py new file mode 100644 index 0000000..3111a41 --- /dev/null +++ b/test/test_wiki_git.py @@ -0,0 +1,81 @@ +import subprocess +from pathlib import Path + +import pytest + +import tools.wiki_git as wiki_git + + +@pytest.mark.parametrize('value', ['', '../master', '-main', 'a..b', 'a b', 'a.lock', 'a/@{x', 'a//b']) +def test_invalid_branch_configuration_is_rejected(monkeypatch, value): + monkeypatch.setenv('EXPECTED_GITEA_BRANCH', value) + with pytest.raises(RuntimeError, match='EXPECTED_GITEA_BRANCH'): + wiki_git.expected_branch() + + +@pytest.mark.parametrize('url', ['git@git.kemt.fei.tuke.sk:jp170na/zpwiki.git', + 'ssh://git@git.kemt.fei.tuke.sk/jp170na/zpwiki.git', 'https://git.kemt.fei.tuke.sk/jp170na/zpwiki']) +def test_repository_identity_from_transport_url(url): + assert wiki_git.repository_from_url(url) == 'jp170na/zpwiki' + + +def test_real_git_fetch_fast_forward_and_local_change_protection(tmp_path, monkeypatch): + run = subprocess.run + def git(root, *args): + return run(['git', '-C', str(root), *args], capture_output=True, text=True, check=True).stdout.strip() + bare, author, checkout = (tmp_path / name for name in ('remote.git', 'author', 'checkout')) + run(['git', 'init', '--bare', str(bare)], check=True, capture_output=True) + run(['git', 'clone', str(bare), str(author)], check=True, capture_output=True) + git(author, 'checkout', '-b', 'master') + git(author, 'config', 'user.name', 'Sync test') + git(author, 'config', 'user.email', 'sync@example.test') + (author / 'README.md').write_text('old') + git(author, 'add', '.') + git(author, 'commit', '-m', 'initial') + git(author, 'push', 'origin', 'master') + run(['git', 'clone', str(bare), str(checkout)], check=True, capture_output=True) + git(checkout, 'remote', 'set-url', 'origin', 'git@example.test:KEMT/zpwiki.git') + (author / 'README.md').write_text('new') + git(author, 'commit', '-am', 'update') + git(author, 'push', 'origin', 'master') + wanted = git(author, 'rev-parse', 'HEAD') + # Exercise real Git fetch/merge against a local bare remote. Only transport + # is redirected, so the production origin/branch validation still runs. + def local_transport(args, **kw): + if args[:2] == ['git', 'fetch']: + args = [str(bare) if arg == 'origin' else arg for arg in args] + return run(args, **kw) + monkeypatch.setattr(wiki_git.subprocess, 'run', local_transport) + wiki_git.update_wiki(checkout) + assert git(checkout, 'rev-parse', 'HEAD') == wanted + assert (checkout / 'README.md').read_text() == 'new' + (checkout / 'untracked.md').write_text('local notes') + with pytest.raises(RuntimeError, match='lokálne zmeny'): + wiki_git.update_wiki(checkout) + (checkout / 'untracked.md').unlink() + (checkout / 'README.md').write_text('local modification') + with pytest.raises(RuntimeError, match='lokálne zmeny'): + wiki_git.update_wiki(checkout) + assert (checkout / 'README.md').read_text() == 'local modification' + git(checkout, 'restore', 'README.md') + monkeypatch.setenv('EXPECTED_GITEA_BRANCH', 'other') + with pytest.raises(RuntimeError, match='Aktívna Git vetva'): + wiki_git.update_wiki(checkout) + monkeypatch.setenv('EXPECTED_GITEA_BRANCH', 'master') + monkeypatch.setenv('EXPECTED_GITEA_REPOSITORY', 'OTHER/zpwiki') + with pytest.raises(RuntimeError, match='origin nezodpovedá'): + wiki_git.update_wiki(checkout) + monkeypatch.setenv('EXPECTED_GITEA_REPOSITORY', 'KEMT/zpwiki') + git(checkout, 'config', 'user.name', 'Local author') + git(checkout, 'config', 'user.email', 'local@example.test') + (checkout / 'local.md').write_text('local commit') + git(checkout, 'add', 'local.md') + git(checkout, 'commit', '-m', 'local') + local_head = git(checkout, 'rev-parse', 'HEAD') + (author / 'README.md').write_text('remote update') + git(author, 'commit', '-am', 'remote') + git(author, 'push', 'origin', 'master') + with pytest.raises(RuntimeError, match='Git operácia merge zlyhala'): + wiki_git.update_wiki(checkout) + assert git(checkout, 'rev-parse', 'HEAD') == local_head + assert (checkout / 'local.md').read_text() == 'local commit' diff --git a/tools/rag/rebuild_index.py b/tools/rag/rebuild_index.py index 4fede02..21a5fde 100644 --- a/tools/rag/rebuild_index.py +++ b/tools/rag/rebuild_index.py @@ -5,12 +5,11 @@ import errno import fcntl import json import os -import subprocess import sys import time from contextlib import contextmanager from pathlib import Path -from typing import Iterator, TextIO +from typing import Callable, Iterator, TextIO from rich import print @@ -22,10 +21,10 @@ if str(PROJECT_ROOT) not in sys.path: from tools.rag.common import DATA_DIR, ZPWIKI_ROOT +from tools.wiki_git import update_wiki REINDEX_LOCK_FILE = DATA_DIR / ".reindex.lock" -GIT_PULL_TIMEOUT_SECONDS = 120 class ReindexInProgressError(RuntimeError): @@ -95,47 +94,7 @@ def acquire_reindex_lock( def git_pull( repo_path: Path = ZPWIKI_ROOT, ) -> None: - if not repo_path.exists(): - raise RuntimeError( - f"ZPWIKI_ROOT neexistuje: {repo_path}" - ) - - if not (repo_path / ".git").exists(): - raise RuntimeError( - f"Nie je to git repozitár: {repo_path}" - ) - - try: - result = subprocess.run( - [ - "git", - "pull", - "--ff-only", - ], - cwd=repo_path, - text=True, - capture_output=True, - timeout=GIT_PULL_TIMEOUT_SECONDS, - check=False, - ) - - except subprocess.TimeoutExpired as error: - raise RuntimeError( - "Git pull prekročil limit " - f"{GIT_PULL_TIMEOUT_SECONDS} sekúnd" - ) from error - - if result.stdout: - print(result.stdout.strip()) - - if result.stderr: - print(result.stderr.strip()) - - if result.returncode != 0: - raise RuntimeError( - "Git pull zlyhal s návratovým kódom " - f"{result.returncode}" - ) + update_wiki(repo_path) def rebuild_index( @@ -144,9 +103,12 @@ def rebuild_index( lock_file: Path | None = None, wiki_root: Path | None = None, data_dir: Path | None = None, + progress: Callable[[str], None] | None = None, ) -> dict: root = wiki_root or ZPWIKI_ROOT target = data_dir or DATA_DIR + if progress: + progress('waiting_for_lock') with acquire_reindex_lock(lock_file or target / '.reindex.lock'): start = time.monotonic() @@ -156,10 +118,16 @@ def rebuild_index( ) if pull_git: + if progress: + progress('git_pull') git_pull(root) from tools.sync_indexes import synchronize_indexes - result = synchronize_indexes(root, target) + if progress: + progress('building_indexes') + result = synchronize_indexes(root, target, progress=progress) + else: + result = synchronize_indexes(root, target) duration = round( time.monotonic() - start, @@ -184,8 +152,8 @@ def main() -> None: "--pull", action="store_true", help=( - "Pred reindexovaním spustí " - "git pull --ff-only." + "Pred reindexovaním stiahne očakávanú vetvu " + "a aktualizuje ju iba fast-forward." ), ) diff --git a/tools/rag/tests/conftest.py b/tools/rag/tests/conftest.py index 6b583b1..5202712 100644 --- a/tools/rag/tests/conftest.py +++ b/tools/rag/tests/conftest.py @@ -11,6 +11,12 @@ if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) +@pytest.fixture(autouse=True) +def isolated_job_volume(monkeypatch, tmp_path): + import app.main as main + monkeypatch.setattr(main, 'DATA_DIR', tmp_path / 'sync-jobs') + + @pytest.fixture(autouse=True) def security_environment( monkeypatch: pytest.MonkeyPatch, diff --git a/tools/rag/tests/test_rebuild_index.py b/tools/rag/tests/test_rebuild_index.py index 5729ff6..722b84b 100644 --- a/tools/rag/tests/test_rebuild_index.py +++ b/tools/rag/tests/test_rebuild_index.py @@ -7,6 +7,7 @@ from pathlib import Path import pytest import tools.rag.rebuild_index as rebuild +import tools.wiki_git as wiki_git def hold_lock(lock_path: str, ready, release) -> None: @@ -89,26 +90,22 @@ def test_rebuild_pulls_and_locks_the_requested_snapshot_paths(tmp_path, monkeypa assert result['index_version']['version'] == 'alternate' -def test_git_pull_uses_fast_forward_only( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - repo = tmp_path / "repo" - (repo / ".git").mkdir(parents=True) - captured = {} - +def test_git_pull_uses_explicit_branch_and_fast_forward_only(tmp_path, monkeypatch): + repo = tmp_path / 'repo' + (repo / '.git').mkdir(parents=True) + captured = [] def fake_run(args, **kwargs): - captured["args"] = args - captured["kwargs"] = kwargs - return subprocess.CompletedProcess(args, 0, stdout="OK", stderr="") - - monkeypatch.setattr(rebuild.subprocess, "run", fake_run) - + captured.append((args, kwargs)) + output = {'remote': 'git@git.example:KEMT/zpwiki.git', 'branch': 'master', 'status': ''}.get(args[1], '') + return subprocess.CompletedProcess(args, 0, stdout=output, stderr='') + monkeypatch.setattr(wiki_git.subprocess, 'run', fake_run) + monkeypatch.setenv('EXPECTED_GITEA_BRANCH', 'master') rebuild.git_pull(repo) - - assert captured["args"] == ["git", "pull", "--ff-only"] - assert captured["kwargs"]["cwd"] == repo - assert captured["kwargs"]["timeout"] == rebuild.GIT_PULL_TIMEOUT_SECONDS + assert captured[-2][0] == ['git', 'fetch', '--no-tags', '--no-recurse-submodules', + 'origin', 'refs/heads/master:refs/remotes/origin/master'] + assert captured[-1][0] == ['git', 'merge', '--ff-only', 'refs/remotes/origin/master'] + assert all(kw['cwd'] == repo and kw['timeout'] == 120 and kw['env']['GIT_TERMINAL_PROMPT'] == '0' + for _, kw in captured) def test_git_pull_rejects_non_repository(tmp_path: Path) -> None: @@ -129,9 +126,9 @@ def test_git_pull_reports_timeout( def timeout(*args, **kwargs): raise subprocess.TimeoutExpired(cmd="git pull", timeout=120) - monkeypatch.setattr(rebuild.subprocess, "run", timeout) + monkeypatch.setattr(wiki_git.subprocess, "run", timeout) - with pytest.raises(RuntimeError, match="prekročil limit"): + with pytest.raises(RuntimeError, match="prekročila limit"): rebuild.git_pull(repo) @@ -143,7 +140,7 @@ def test_git_pull_reports_nonzero_exit( (repo / ".git").mkdir(parents=True) monkeypatch.setattr( - rebuild.subprocess, + wiki_git.subprocess, "run", lambda *args, **kwargs: subprocess.CompletedProcess( args[0], diff --git a/tools/sync_indexes.py b/tools/sync_indexes.py index e8d4615..1abe68a 100644 --- a/tools/sync_indexes.py +++ b/tools/sync_indexes.py @@ -142,11 +142,13 @@ def publish_snapshot(data_dir: Path, stage: Path, version: dict, payload: dict, return graph_summary -def synchronize_indexes(wiki_root: Path, data_dir: Path) -> dict: +def synchronize_indexes(wiki_root: Path, data_dir: Path, *, progress=None) -> 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) + if progress: + progress('publishing_indexes') settings = get_neo4j_settings() with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: driver.verify_connectivity() diff --git a/tools/sync_jobs.py b/tools/sync_jobs.py new file mode 100644 index 0000000..aaa3987 --- /dev/null +++ b/tools/sync_jobs.py @@ -0,0 +1,173 @@ +"""Durable synchronization queue, shared by API processes on the data volume.""" +from __future__ import annotations + +import json +import logging +import sqlite3 +import threading +import time +import uuid +from contextlib import closing +from pathlib import Path +from typing import Callable + +from tools.rag.rebuild_index import ReindexInProgressError, acquire_reindex_lock, rebuild_index + +logger = logging.getLogger(__name__) + + +class SyncJobs: + def __init__(self, data_dir: Path, *, runner: Callable | None = None): + self.data_dir = Path(data_dir) + self.data_dir.mkdir(parents=True, exist_ok=True) + self.path = self.data_dir / 'sync_jobs.sqlite' + self.runner = runner or rebuild_index + self.stopping = threading.Event() + self.thread: threading.Thread | None = None + with closing(self.connect()) as conn, conn: + conn.executescript(''' + CREATE TABLE IF NOT EXISTS jobs ( + id TEXT PRIMARY KEY, status TEXT NOT NULL, phase TEXT NOT NULL, + pull_git INTEGER NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL, + started_at REAL, finished_at REAL, requested_revision TEXT, + request_count INTEGER NOT NULL DEFAULT 1, attempts INTEGER NOT NULL DEFAULT 0, + available_at REAL NOT NULL DEFAULT 0, reason TEXT, result_json TEXT); + CREATE INDEX IF NOT EXISTS jobs_status ON jobs(status, created_at); + CREATE TABLE IF NOT EXISTS deliveries ( + delivery_key TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES jobs(id)); + ''') + + def connect(self): + conn = sqlite3.connect(self.path, timeout=15) + conn.row_factory = sqlite3.Row + conn.execute('PRAGMA foreign_keys=ON') + return conn + + @staticmethod + def public(row): + if row is None: + return None + value = dict(row) + value['pull_git'] = bool(value['pull_git']) + result = value.pop('result_json') + value['result'] = json.loads(result) if result else None + return value + + def enqueue(self, *, pull_git: bool, delivery_key: str | None = None, + requested_revision: str | None = None) -> dict: + now = time.time() + with closing(self.connect()) as conn, conn: + conn.execute('BEGIN IMMEDIATE') + if delivery_key: + duplicate = conn.execute('SELECT job_id FROM deliveries WHERE delivery_key=?', + (delivery_key,)).fetchone() + if duplicate: + return {'job': self.public(conn.execute('SELECT * FROM jobs WHERE id=?', + (duplicate['job_id'],)).fetchone()), 'coalesced': False, 'duplicate': True} + # Never merge a new request into a running job: its source snapshot + # may already exist. One queued follow-up absorbs further pushes. + queued = conn.execute("SELECT * FROM jobs WHERE status='queued' ORDER BY created_at LIMIT 1").fetchone() + if queued: + job_id = queued['id'] + conn.execute('''UPDATE jobs SET pull_git=MAX(pull_git,?), updated_at=?, + requested_revision=COALESCE(?,requested_revision), request_count=request_count+1 + WHERE id=?''', (int(pull_git), now, requested_revision, job_id)) + else: + job_id = uuid.uuid4().hex + conn.execute('''INSERT INTO jobs(id,status,phase,pull_git,created_at,updated_at,requested_revision) + VALUES(?,'queued','queued',?,?,?,?)''', + (job_id, int(pull_git), now, now, requested_revision)) + if delivery_key: + conn.execute('INSERT INTO deliveries VALUES(?,?)', (delivery_key, job_id)) + job = self.public(conn.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()) + return {'job': job, 'coalesced': queued is not None, 'duplicate': False} + + def get(self, job_id: str): + with closing(self.connect()) as conn: + return self.public(conn.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()) + + def overview(self): + with closing(self.connect()) as conn: + active = [self.public(row) for row in conn.execute( + "SELECT * FROM jobs WHERE status IN ('queued','running') ORDER BY created_at")] + success = self.public(conn.execute( + "SELECT * FROM jobs WHERE status='succeeded' ORDER BY finished_at DESC LIMIT 1").fetchone()) + failure = self.public(conn.execute( + "SELECT * FROM jobs WHERE status='failed' ORDER BY finished_at DESC LIMIT 1").fetchone()) + return {'active': active, 'last_success': success, 'last_failure': failure} + + def phase(self, job_id: str, phase: str): + with closing(self.connect()) as conn, conn: + conn.execute('UPDATE jobs SET phase=?,updated_at=? WHERE id=?', (phase, time.time(), job_id)) + + def run_once(self) -> bool: + """The process lock covers recovery, claim and execution, including retries.""" + try: + with acquire_reindex_lock(self.data_dir / '.sync-worker.lock'): + return self._run_locked() + except ReindexInProgressError: + return False + + def _run_locked(self) -> bool: + now = time.time() + with closing(self.connect()) as conn, conn: + conn.execute('BEGIN IMMEDIATE') + # No other worker can be running while this process holds the lock. + # A persisted running row therefore belongs to an interrupted process. + conn.execute("""UPDATE jobs SET status='queued',phase='recovered',reason='interrupted', + updated_at=? WHERE status='running'""", (now,)) + row = conn.execute("""SELECT * FROM jobs WHERE status='queued' AND available_at<=? + ORDER BY created_at LIMIT 1""", (now,)).fetchone() + if row is None: + return False + job_id = row['id'] + conn.execute("""UPDATE jobs SET status='running',phase='starting',started_at=?, + updated_at=?,attempts=attempts+1,reason=NULL WHERE id=?""", (now, now, job_id)) + try: + result = self.runner(pull_git=bool(row['pull_git']), data_dir=self.data_dir, + progress=lambda phase: self.phase(job_id, phase)) + except ReindexInProgressError: + # A CLI rebuild may own the pipeline lock. Keep the HTTP request + # durable and try again; this is not a failed synchronization attempt. + with closing(self.connect()) as conn, conn: + conn.execute("""UPDATE jobs SET status='queued',phase='waiting_for_lock', + attempts=attempts-1,available_at=?,updated_at=? WHERE id=?""", + (time.time() + 1, time.time(), job_id)) + except Exception: + logger.exception('Synchronizácia %s zlyhala.', job_id) + retry = row['attempts'] + 1 < 3 + with closing(self.connect()) as conn, conn: + conn.execute('''UPDATE jobs SET status=?,phase=?,reason='sync_failed', + updated_at=?,finished_at=?,available_at=? WHERE id=?''', + ('queued' if retry else 'failed', 'retry_wait' if retry else 'failed', + time.time(), None if retry else time.time(), time.time() + 5 * (row['attempts'] + 1), job_id)) + else: + with closing(self.connect()) as conn, conn: + conn.execute("""UPDATE jobs SET status='succeeded',phase='complete',result_json=?, + updated_at=?,finished_at=?,reason=NULL WHERE id=?""", + (json.dumps(result, ensure_ascii=False), time.time(), time.time(), job_id)) + return True + + def _loop(self): + while not self.stopping.is_set(): + try: + worked = self.run_once() + except Exception: + logger.exception('Chyba pracovníka synchronizačnej fronty.') + worked = False + if not worked: + self.stopping.wait(0.5) + + def start(self): + if self.thread and self.thread.is_alive(): + return + self.stopping.clear() + self.thread = threading.Thread(target=self._loop, name='index-sync', daemon=True) + self.thread.start() + + def stop(self): + self.stopping.set() + if self.thread: + # Compose grants five minutes to finish publication before shutdown. + # An interrupted job remains durable and is recovered on startup. + self.thread.join(timeout=285) diff --git a/tools/wiki_git.py b/tools/wiki_git.py new file mode 100644 index 0000000..dfe06db --- /dev/null +++ b/tools/wiki_git.py @@ -0,0 +1,67 @@ +"""Non-interactive, branch-specific Git updates of the configured wiki.""" +from __future__ import annotations + +import os +import re +import subprocess +from pathlib import Path +from urllib.parse import urlsplit + +GIT_TIMEOUT_SECONDS = 120 + + +def expected_branch() -> str: + branch = os.getenv('EXPECTED_GITEA_BRANCH', 'master').strip() + if (not branch or branch.startswith(('-', '/')) or branch.endswith(('/', '.')) + or '..' in branch or '@{' in branch or branch == '@' + or re.search(r'[\s\x00-\x1f\x7f~^:?*\[\\]', branch) + or any(not part or part.startswith('.') or part.endswith('.lock') for part in branch.split('/'))): + raise RuntimeError('EXPECTED_GITEA_BRANCH nie je platný názov Git vetvy') + return branch + + +def repository_from_url(url: str) -> str | None: + if '://' in url: + parsed = urlsplit(url) + if parsed.scheme not in {'ssh', 'https', 'http', 'git'}: + return None + path = parsed.path + elif re.match(r'[^/\s]+:[^\s]+', url): + path = url.split(':', 1)[1] + else: + return None + path = path.strip('/').removesuffix('.git') + return path if len(path.split('/')) == 2 else None + + +def update_wiki(repo_path: Path) -> None: + if not repo_path.exists(): + raise RuntimeError(f'ZPWIKI_ROOT neexistuje: {repo_path}') + if not (repo_path / '.git').exists(): + raise RuntimeError(f'Nie je to git repozitár: {repo_path}') + environment = {**os.environ, 'GIT_TERMINAL_PROMPT': '0'} + + def git(*args): + try: + result = subprocess.run(['git', *args], cwd=repo_path, env=environment, + text=True, capture_output=True, timeout=GIT_TIMEOUT_SECONDS, check=False) + except subprocess.TimeoutExpired as error: + raise RuntimeError(f'Git aktualizácia prekročila limit {GIT_TIMEOUT_SECONDS} sekúnd') from error + if result.returncode: + # Remote URLs and credentials can occur in Git stderr; do not place + # them in API responses or ordinary application logs. + raise RuntimeError(f'Git operácia {args[0]} zlyhala s návratovým kódom {result.returncode}') + return result.stdout.strip() + + expected = os.getenv('EXPECTED_GITEA_REPOSITORY', '').strip() + origin = repository_from_url(git('remote', 'get-url', 'origin')) + if not expected or origin is None or origin.casefold() != expected.casefold(): + raise RuntimeError('Git origin nezodpovedá EXPECTED_GITEA_REPOSITORY') + branch = expected_branch() + if git('branch', '--show-current') != branch: + raise RuntimeError('Aktívna Git vetva nezodpovedá EXPECTED_GITEA_BRANCH') + if git('status', '--porcelain', '--untracked-files=normal'): + raise RuntimeError('Wiki obsahuje lokálne zmeny; automatická aktualizácia ich neprepíše') + git('fetch', '--no-tags', '--no-recurse-submodules', 'origin', + f'refs/heads/{branch}:refs/remotes/origin/{branch}') + git('merge', '--ff-only', f'refs/remotes/origin/{branch}')