diff --git a/backend/app/services/ogm_harvest/harvest.py b/backend/app/services/ogm_harvest/harvest.py index 744f4ae..a61f455 100644 --- a/backend/app/services/ogm_harvest/harvest.py +++ b/backend/app/services/ogm_harvest/harvest.py @@ -10,6 +10,7 @@ from app.services.ogm_harvest.importer import OGMResourceImporter from app.services.ogm_harvest.repo_sync import OGMRepoSync from app.services.ogm_harvest.repository import OGMHarvestRepository +from app.services.ogm_harvest.search_index import sync_repo_search from app.services.thumbnail_refresh_service import refresh_thumbnail_cache_for_changed_resources logger = logging.getLogger(__name__) @@ -78,6 +79,15 @@ async def harvest_repo( ogm_run_id=run_id, progress_meta={"head_sha": head_sha, "repo_action": sync_result.action}, ) + await repo.update_harvest_run( + ogm_id=run_id, + ogm_stats_json={ + **stats, + "stage": "search_index", + "updated_at": datetime.utcnow().isoformat() + "Z", + }, + ) + stats["search_index"] = await sync_repo_search(repo_name) changed_thumbnail_ids = sorted(importer.changed_thumbnail_resource_ids) stats["thumbnail_sources_changed"] = len(changed_thumbnail_ids) if changed_thumbnail_ids: diff --git a/backend/app/services/ogm_harvest/search_index.py b/backend/app/services/ogm_harvest/search_index.py new file mode 100644 index 0000000..60b892a --- /dev/null +++ b/backend/app/services/ogm_harvest/search_index.py @@ -0,0 +1,32 @@ +"""Run bounded bulk indexing in a process isolated from the harvest worker loop.""" + +import asyncio +import json +import sys +from pathlib import Path + + +async def sync_repo_search(repo_name: str) -> dict: + script = Path(__file__).resolve().parents[3] / "scripts" / "sync_ogm_search_index.py" + process = await asyncio.create_subprocess_exec( + sys.executable, + str(script), + "--repo", + repo_name, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + try: + output, _ = await process.communicate() + except BaseException: + if process.returncode is None: + process.kill() + await process.wait() + raise + lines = output.decode("utf-8", errors="replace").splitlines() + if process.returncode: + raise RuntimeError("OGM search synchronization failed: " + "\n".join(lines[-10:])) + for line in reversed(lines): + if line.startswith("SEARCH_SYNC_RESULT "): + return json.loads(line.removeprefix("SEARCH_SYNC_RESULT ")) + raise RuntimeError("OGM search synchronization returned no result") diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 6f5c4b6..8abefeb 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "opengeometadata-api" -version = "0.10.0" +version = "0.10.1" description = "Search, harvest, and delivery API for OpenGeoMetadata Aardvark records" requires-python = ">=3.11" authors = [ diff --git a/backend/scripts/sync_ogm_search_index.py b/backend/scripts/sync_ogm_search_index.py new file mode 100644 index 0000000..11b27ab --- /dev/null +++ b/backend/scripts/sync_ogm_search_index.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Synchronize harvested database records into the live index without replacing it. + +Run in a separate process: the bulk preparation helper temporarily substitutes +prefetched enrichment lookups, so it must not share an event loop with API work. +""" + +import argparse +import asyncio +import json +import logging +import os +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.services.cache_service import CacheService # noqa: E402 +from scripts import reindex_atomic as bulk # noqa: E402 + +logger = logging.getLogger(__name__) + + +async def sync_search(repo_name: str | None = None, batch_size: int = 1000) -> dict: + if batch_size < 1: + raise ValueError("batch_size must be positive") + index_name = os.getenv("ELASTICSEARCH_INDEX", "opengeometadata_api") + stats = {"processed": 0, "indexed": 0, "deleted": 0, "errors": 0} + last_id = None + await bulk.database.connect() + try: + # Do not create an empty index accidentally when a hostname/index is misconfigured. + if not await bulk.es.indices.exists(index=index_name): + raise RuntimeError(f"Search index does not exist: {index_name}") + while True: + conditions = [] + params = {"limit": batch_size} + if repo_name is not None: + conditions.append("""EXISTS ( + SELECT 1 FROM ogm_resource_state s + WHERE s.ogm_resource_id = r.id AND s.ogm_repo_name = :repo_name + )""") + params["repo_name"] = repo_name + if last_id is not None: + conditions.append("r.id > :last_id") + params["last_id"] = last_id + where = " AND ".join(conditions) or "TRUE" + rows = await bulk.database.fetch_all( + f"""SELECT r.id, lower(coalesce(nullif(r.b1g_publication_state_s, ''), + nullif(r.publication_state, ''), 'published')) AS publication_state + FROM resources r WHERE {where} ORDER BY r.id LIMIT :limit""", + params, + ) + if not rows: + break + last_id = str(rows[-1]["id"]) + published = [str(r["id"]) for r in rows if r["publication_state"] == "published"] + unpublished = [str(r["id"]) for r in rows if r["publication_state"] != "published"] + documents = await bulk._prepare_documents_for_chunk(published) + result = await bulk._bulk_index_documents(index_name, documents, batch_size, 2) + stats["processed"] += len(rows) + stats["indexed"] += result["indexed"] + stats["errors"] += len(published) - result["indexed"] + if unpublished: + result = await bulk.es.bulk( + operations=[ + {"delete": {"_index": index_name, "_id": rid}} for rid in unpublished + ], + refresh=False, + ) + stats["errors"] += max(0, len(unpublished) - len(result["items"])) + for item in result["items"]: + status = item["delete"]["status"] + if status in (200, 404): + stats["deleted"] += int(status == 200) + else: + stats["errors"] += 1 + logger.info("Search synchronization repo=%s stats=%s", repo_name, stats) + await bulk.es.indices.refresh(index=index_name) + await CacheService().invalidate_tags({"search", "suggest"}) + if stats["errors"]: + raise RuntimeError(f"Search synchronization incomplete: {stats}") + return stats + finally: + await bulk.database.disconnect() + await bulk.es.close() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", default=None) + parser.add_argument("--batch-size", type=int, default=1000) + args = parser.parse_args() + logging.getLogger().setLevel(logging.WARNING) + logger.setLevel(logging.INFO) + stats = asyncio.run(sync_search(args.repo, args.batch_size)) + print("SEARCH_SYNC_RESULT " + json.dumps(stats), flush=True) + + +if __name__ == "__main__": + main() diff --git a/backend/tests/scripts/test_sync_ogm_search_index.py b/backend/tests/scripts/test_sync_ogm_search_index.py new file mode 100644 index 0000000..151dcfa --- /dev/null +++ b/backend/tests/scripts/test_sync_ogm_search_index.py @@ -0,0 +1,81 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from scripts import sync_ogm_search_index as sync + + +@pytest.mark.asyncio +async def test_sync_indexes_every_batch_deletes_unpublished_and_invalidates_search(monkeypatch): + db = MagicMock() + db.connect = AsyncMock() + db.disconnect = AsyncMock() + db.fetch_all = AsyncMock( + side_effect=[ + [ + {"id": "a", "publication_state": "published"}, + {"id": "b", "publication_state": "draft"}, + ], + [{"id": "c", "publication_state": "published"}], + [], + ] + ) + es = MagicMock() + es.indices.exists = AsyncMock(return_value=True) + es.indices.refresh = AsyncMock() + es.bulk = AsyncMock(return_value={"items": [{"delete": {"status": 200}}]}) + es.close = AsyncMock() + prepare = AsyncMock(side_effect=[[("a", {})], [("c", {})]]) + index = AsyncMock(return_value={"indexed": 1}) + cache = MagicMock() + cache.invalidate_tags = AsyncMock() + monkeypatch.setattr(sync.bulk, "database", db) + monkeypatch.setattr(sync.bulk, "es", es) + monkeypatch.setattr(sync.bulk, "_prepare_documents_for_chunk", prepare) + monkeypatch.setattr(sync.bulk, "_bulk_index_documents", index) + monkeypatch.setattr(sync, "CacheService", lambda: cache) + monkeypatch.setenv("ELASTICSEARCH_INDEX", "test-index") + + result = await sync.sync_search("repo", batch_size=2) + + assert result == {"processed": 3, "indexed": 2, "deleted": 1, "errors": 0} + assert db.fetch_all.await_args_list[1].args[1] == { + "limit": 2, + "repo_name": "repo", + "last_id": "b", + } + assert [call.args[0] for call in prepare.await_args_list] == [["a"], ["c"]] + es.bulk.assert_awaited_once_with( + operations=[{"delete": {"_index": "test-index", "_id": "b"}}], refresh=False + ) + es.indices.refresh.assert_awaited_once_with(index="test-index") + cache.invalidate_tags.assert_awaited_once_with({"search", "suggest"}) + db.disconnect.assert_awaited_once() + es.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_sync_refuses_to_report_success_when_documents_are_missing(monkeypatch): + db = MagicMock( + connect=AsyncMock(), + disconnect=AsyncMock(), + fetch_all=AsyncMock( + side_effect=[ + [{"id": "a", "publication_state": "published"}], + [], + ] + ), + ) + es = MagicMock(close=AsyncMock()) + es.indices.exists = AsyncMock(return_value=True) + es.indices.refresh = AsyncMock() + monkeypatch.setattr(sync.bulk, "database", db) + monkeypatch.setattr(sync.bulk, "es", es) + monkeypatch.setattr(sync.bulk, "_prepare_documents_for_chunk", AsyncMock(return_value=[])) + monkeypatch.setattr(sync.bulk, "_bulk_index_documents", AsyncMock(return_value={"indexed": 0})) + cache = MagicMock(invalidate_tags=AsyncMock()) + monkeypatch.setattr(sync, "CacheService", lambda: cache) + with pytest.raises(RuntimeError, match="incomplete"): + await sync.sync_search("repo") + cache.invalidate_tags.assert_awaited_once() + es.close.assert_awaited_once() diff --git a/backend/tests/services/test_ogm_search_index.py b/backend/tests/services/test_ogm_search_index.py new file mode 100644 index 0000000..10504a5 --- /dev/null +++ b/backend/tests/services/test_ogm_search_index.py @@ -0,0 +1,74 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.services.ogm_harvest import search_index + + +@pytest.mark.asyncio +async def test_sync_repo_search_uses_argument_list_and_reads_result(monkeypatch): + process = MagicMock(returncode=0) + process.communicate = AsyncMock( + return_value=(b'log\nSEARCH_SYNC_RESULT {"indexed": 42}\n', None) + ) + start = AsyncMock(return_value=process) + monkeypatch.setattr(search_index.asyncio, "create_subprocess_exec", start) + assert await search_index.sync_repo_search("repo with spaces") == {"indexed": 42} + assert start.await_args.args[-2:] == ("--repo", "repo with spaces") + + +@pytest.mark.asyncio +async def test_sync_repo_search_propagates_index_failure(monkeypatch): + process = MagicMock(returncode=1) + process.communicate = AsyncMock(return_value=(b"index failed", None)) + monkeypatch.setattr( + search_index.asyncio, "create_subprocess_exec", AsyncMock(return_value=process) + ) + with pytest.raises(RuntimeError, match="index failed"): + await search_index.sync_repo_search("repo") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("index_fails", [False, True]) +async def test_harvest_requires_search_sync_before_reporting_success(monkeypatch, index_fails): + from types import SimpleNamespace + + from app.services.ogm_harvest import harvest + + repo = MagicMock() + for name in ( + "upsert_repo", + "mark_repo_harvest_started", + "cancel_other_running_runs", + "update_harvest_run", + "finalize_harvest_run", + "mark_repo_harvest_completed", + ): + setattr(repo, name, AsyncMock()) + repo.create_harvest_run = AsyncMock(return_value=1) + importer = MagicMock(changed_thumbnail_resource_ids=set()) + importer.upsert_stream = AsyncMock(return_value={"imported": 1, "errors": 0}) + syncer = MagicMock() + syncer.ensure_repo.return_value = SimpleNamespace( + head_sha="abc", action="pull", repo_dir="/tmp" + ) + dump = MagicMock() + dump.finalize.return_value = SimpleNamespace(run_dir="/tmp/run") + search = AsyncMock(return_value={"indexed": 1, "errors": 0}) + if index_fails: + search.side_effect = RuntimeError("index unavailable") + monkeypatch.setattr(harvest, "OGMHarvestRepository", lambda: repo) + monkeypatch.setattr(harvest, "OGMResourceImporter", lambda **kw: importer) + monkeypatch.setattr(harvest, "OGMRepoSync", lambda **kw: syncer) + monkeypatch.setattr(harvest, "OGMHarvestDumpWriter", lambda **kw: dump) + monkeypatch.setattr(harvest, "read_aardvark_records", lambda _: []) + monkeypatch.setattr(harvest, "sync_repo_search", search) + if index_fails: + with pytest.raises(RuntimeError, match="index unavailable"): + await harvest.harvest_repo("repo") + assert repo.finalize_harvest_run.await_args.kwargs["ogm_status"] == "failed" + else: + result = await harvest.harvest_repo("repo") + assert result["stats"]["search_index"] == {"indexed": 1, "errors": 0} + assert repo.finalize_harvest_run.await_args.kwargs["ogm_status"] == "success" + search.assert_awaited_once_with("repo") diff --git a/backend/uv.lock b/backend/uv.lock index d506356..5fb9beb 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -1642,7 +1642,7 @@ wheels = [ [[package]] name = "opengeometadata-api" -version = "0.10.0" +version = "0.10.1" source = { virtual = "." } dependencies = [ { name = "aiohttp" }, diff --git a/docs/deployment.md b/docs/deployment.md index 9cee544..be8e8ac 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -780,3 +780,26 @@ For deployment issues: 3. Verify server connectivity: `kamal details` 4. Review this documentation 5. Check Kamal documentation: https://kamal-deploy.org/ + +### Keep OGM harvests searchable + +Each repository harvest bulk-synchronizes its database records to Elasticsearch +before reporting success. Indexing uses keyset pagination, keeps the live index +available, removes records whose effective publication state is unpublished, and +invalidates search/suggestion response caches after refresh. Index failures make +the harvest fail visibly; rerunning the harvest retries synchronization. + +For a manual reconciliation inside the running backend container: + +```bash +python /app/backend/scripts/sync_ogm_search_index.py --repo gov.usgs +# Omit --repo to synchronize every resource already present in PostgreSQL. +``` + +The helper runs in a separate process because bulk enrichment preparation uses +process-local prefetched lookups. It neither drops the live index nor deletes +PostgreSQL records. Database rows removed entirely are outside this helper's scope. + +Dashboard harvested totals count repository memberships, so shared IDs can appear +more than once. Its published total covers repository-tagged database records; +search also includes published records without a repository tag. diff --git a/docs/upstream_reconciliation.md b/docs/upstream_reconciliation.md index cdb56b2..3ae8ac8 100644 --- a/docs/upstream_reconciliation.md +++ b/docs/upstream_reconciliation.md @@ -13,7 +13,7 @@ upstream `backend/` subtree recorded in `config/upstream-backend-source.env`. | Upstream tag containing that commit | `v0.7.16` | | Recorded backend split | `1cf222b377be4ba1d8468fa864724542e866eb3e` | | Review ceiling | Upstream `0.8.11` (`4254aa3`) | -| OGM product version | Independent; currently `0.10.0` | +| OGM product version | Independent; currently `0.10.1` | The review ceiling is not a promise that this product becomes upstream `0.8.11`. OGM releases and upstream source provenance are separate concepts.