Keep OGM harvests synchronized with search

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:

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")

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 = [

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()

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()

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")

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.

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.