diff --git a/config.py b/config.py
index b65559f..a2acdf4 100644
--- a/config.py
+++ b/config.py
@@ -20,6 +20,7 @@
 OPEN_SEARCH_URL = 'http://opensearch-node1:9200'
 OPEN_SEARCH_INDEX = 'scan-explorer'
 OPEN_SEARCH_AGG_BUCKET_LIMIT = 10000
+OPEN_SEARCH_MAX_RESULT_WINDOW = 10000
 
 REDIS_URL = 'redis://redis-backend:6379/4'
 
diff --git a/scan_explorer_service/manifest_factory.py b/scan_explorer_service/manifest_factory.py
index b9d7dec..913a5d0 100644
--- a/scan_explorer_service/manifest_factory.py
+++ b/scan_explorer_service/manifest_factory.py
@@ -1,4 +1,5 @@
 from typing import Dict
+from html import escape
 from iiif_prezi.factory import ManifestFactory, Sequence, Canvas, Image, Annotation, Manifest, Range
 from scan_explorer_service.models import Article, Page, Collection
 from typing import Union
@@ -68,7 +69,9 @@ def get_or_create_canvas(self, page: Page):
 
         if len(page.articles) > 0:
             metadata = {
-                'Abstract': ''.join(f'<a href="https://ui.adsabs.harvard.edu/abs/{str(x.bibcode)}/abstract">{str(x.bibcode)}</a><br/>' for x in page.articles)
+                'Abstract': ''.join(
+                    f'<a href="/abs/{escape(str(article.bibcode))}/abstract">{escape(str(article.bibcode))}</a><br/>'
+                    for article in page.articles)
             }
             canvas.set_metadata(metadata)
 
diff --git a/scan_explorer_service/open_search.py b/scan_explorer_service/open_search.py
index 50c59a7..f2acbec 100644
--- a/scan_explorer_service/open_search.py
+++ b/scan_explorer_service/open_search.py
@@ -138,6 +138,11 @@ def page_os_search(qs: str, page, limit, sort):
     query = create_query_string_query(qs)
     query = set_page_search_fields(query)
     from_number = (page - 1) * limit
+    window = current_app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+    if from_number + limit > window:
+        raise ValueError(
+            f'page {page} at limit {limit} reaches result {from_number + limit}, '
+            f'beyond the searchable window of {window}')
     query['size'] = limit
     query['from'] = from_number
     query['track_total_hits'] = True
diff --git a/scan_explorer_service/tests/test_cache.py b/scan_explorer_service/tests/test_cache.py
index fa76d95..12d8512 100644
--- a/scan_explorer_service/tests/test_cache.py
+++ b/scan_explorer_service/tests/test_cache.py
@@ -414,5 +414,188 @@ def test_ocr_served_from_cache(self, mock_cache_get):
         self.assertIn('text/plain', r.content_type)
 
 
+
+class TestVariantIsolation(TestCaseDatabase):
+    """Two deployments serving different hostnames must not share cached documents."""
+
+    def create_app(self):
+        from scan_explorer_service.app import create_app
+        return create_app(**{
+            'SQLALCHEMY_DATABASE_URI': self.postgresql_url,
+            'TESTING': True,
+            'PROXY_SERVER': 'https://ui.adsabs.harvard.edu:443',
+            'PROXY_PREFIX': '/v1/scan',
+        })
+
+    def setUp(self):
+        super().setUp()
+        cache_mod._redis_client = None
+
+    def tearDown(self):
+        cache_mod._redis_client = None
+        super().tearDown()
+
+    @patch('scan_explorer_service.utils.cache.redis.from_url')
+    def test_variants_do_not_share_cached_manifests(self, mock_from_url):
+        store = {}
+        client = MagicMock()
+        client.ping.return_value = True
+        client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
+        client.get.side_effect = store.get
+        mock_from_url.return_value = client
+
+        bibcode = '1993ASPC...52..132K'
+        ads_manifest = '{"@id":"https://ui.adsabs.harvard.edu:443/v1/scan/..."}'
+
+        cache_mod.cache_set_manifest(bibcode, ads_manifest)
+        self.assertEqual(cache_mod.cache_get_manifest(bibcode), ads_manifest)
+
+        self.app.config['PROXY_SERVER'] = 'https://scixplorer.org:443'
+        self.app.config['PROXY_PREFIX'] = '/v1/scix-scan'
+
+        self.assertIsNone(
+            cache_mod.cache_get_manifest(bibcode),
+            'the SciX deployment must not read the manifest cached by the ADS deployment')
+
+        scix_manifest = '{"@id":"https://scixplorer.org:443/v1/scix-scan/..."}'
+        cache_mod.cache_set_manifest(bibcode, scix_manifest)
+        self.assertEqual(cache_mod.cache_get_manifest(bibcode), scix_manifest)
+
+        self.app.config['PROXY_SERVER'] = 'https://ui.adsabs.harvard.edu:443'
+        self.app.config['PROXY_PREFIX'] = '/v1/scan'
+        self.assertEqual(cache_mod.cache_get_manifest(bibcode), ads_manifest)
+
+    @patch('scan_explorer_service.utils.cache.redis.from_url')
+    def test_variants_do_not_share_cached_searches(self, mock_from_url):
+        store = {}
+        client = MagicMock()
+        client.ping.return_value = True
+        client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
+        client.get.side_effect = store.get
+        mock_from_url.return_value = client
+
+        cache_key = 'ocr:1993ASPC...52..132K:gas'
+        ads_annotations = '{"resources":[{"on":"https://ui.adsabs.harvard.edu:443/v1/scan/canvas/x"}]}'
+
+        cache_mod.cache_set_search(cache_key, ads_annotations)
+        self.assertEqual(cache_mod.cache_get_search(cache_key), ads_annotations)
+
+        self.app.config['PROXY_SERVER'] = 'https://scixplorer.org:443'
+        self.app.config['PROXY_PREFIX'] = '/v1/scix-scan'
+
+        self.assertIsNone(
+            cache_mod.cache_get_search(cache_key),
+            'the SciX deployment must not read content-search annotations cached by ADS')
+
+    @patch('scan_explorer_service.utils.cache.redis.from_url')
+    def test_delete_removes_the_id_from_every_recorded_scope(self, mock_from_url):
+        store = {}
+        scopes = set()
+        client = MagicMock()
+        client.ping.return_value = True
+        client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
+        client.get.side_effect = store.get
+        client.sadd.side_effect = lambda k, v: scopes.add(v)
+        client.smembers.side_effect = lambda k: set(scopes)
+        client.delete.side_effect = lambda k: store.pop(k, None)
+        mock_from_url.return_value = client
+
+        collection_id = 'ApJ0099'
+        cache_mod.cache_set_manifest(collection_id, '{"ads":1}')
+
+        self.app.config['PROXY_SERVER'] = 'https://scixplorer.org:443'
+        self.app.config['PROXY_PREFIX'] = '/v1/scix-scan'
+        cache_mod.cache_set_manifest(collection_id, '{"scix":1}')
+
+        self.assertEqual(len(store), 2)
+        cache_mod.cache_delete_manifest(collection_id)
+        self.assertEqual(store, {}, 'the collection PUT must clear both deployment scopes')
+
+    @patch('scan_explorer_service.utils.cache.redis.from_url')
+    def test_delete_leaves_other_ids_alone(self, mock_from_url):
+        store = {}
+        scopes = set()
+        client = MagicMock()
+        client.ping.return_value = True
+        client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
+        client.sadd.side_effect = lambda k, v: scopes.add(v)
+        client.smembers.side_effect = lambda k: set(scopes)
+        client.delete.side_effect = lambda k: store.pop(k, None)
+        mock_from_url.return_value = client
+
+        cache_mod.cache_set_manifest('ApJ0099', '{"a":1}')
+        cache_mod.cache_set_manifest('ApJ00990', '{"b":1}')
+        cache_mod.cache_set_manifest('*', '{"c":1}')
+
+        cache_mod.cache_delete_manifest('ApJ0099')
+
+        remaining = sorted(k.rsplit(':', 1)[-1] for k in store)
+        self.assertEqual(remaining, ['*', 'ApJ00990'])
+    def _fake_redis(self, mock_from_url):
+        store, scopes = {}, set()
+        client = MagicMock()
+        client.ping.return_value = True
+        client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
+        client.get.side_effect = store.get
+        client.sadd.side_effect = lambda k, v: scopes.add(v)
+        client.smembers.side_effect = lambda k: set(scopes)
+        client.delete.side_effect = lambda *names: [store.pop(n, None) for n in names]
+
+        def _set(name, value, nx=False, ex=None):
+            if ex is None:
+                raise AssertionError('a claim without an expiry would outlive a crashed holder')
+            if nx and name in store:
+                return None
+            store[name] = value
+            return True
+        client.set.side_effect = _set
+        self.redis_client = client
+        mock_from_url.return_value = client
+        return store
+
+    @patch('scan_explorer_service.utils.cache.redis.from_url')
+    def test_bulk_delete_clears_every_id_in_every_scope(self, mock_from_url):
+        store = {}
+        scopes = set()
+        client = MagicMock()
+        client.ping.return_value = True
+        client.setex.side_effect = lambda k, ttl, v: store.__setitem__(k, v)
+        client.sadd.side_effect = lambda k, v: scopes.add(v)
+        client.smembers.side_effect = lambda k: set(scopes)
+        client.delete.side_effect = lambda *names: [store.pop(n, None) for n in names]
+        mock_from_url.return_value = client
+
+        ids = ['ApJ0099', '1988ApJ...333..341R', '1988ApJ...333..352S']
+        for i in ids:
+            cache_mod.cache_set_manifest(i, '{"ads":1}')
+        self.app.config['PROXY_SERVER'] = 'https://scixplorer.org:443'
+        self.app.config['PROXY_PREFIX'] = '/v1/scix-scan'
+        for i in ids:
+            cache_mod.cache_set_manifest(i, '{"scix":1}')
+        cache_mod.cache_set_manifest('untouched', '{"scix":1}')
+
+        self.assertEqual(len(store), 7)
+        cache_mod.cache_delete_manifests(ids)
+
+        remaining = sorted(k.rsplit(':', 1)[-1] for k in store)
+        self.assertEqual(remaining, ['untouched'])
+
+    @patch('scan_explorer_service.utils.cache.redis.from_url')
+    def test_bulk_delete_spans_more_than_one_batch(self, mock_from_url):
+        store = self._fake_redis(mock_from_url)
+        ids = ['id%04d' % i for i in range(cache_mod.DELETE_BATCH_SIZE + 25)]
+        for i in ids:
+            cache_mod.cache_set_manifest(i, '{}')
+        self.assertEqual(len(store), len(ids))
+
+        cache_mod.cache_delete_manifests(ids)
+        leftovers = [k for k in store if k.startswith(cache_mod.MANIFEST_CACHE_PREFIX)]
+        self.assertEqual(leftovers, [], 'every batch must be deleted, not just the first')
+        self.assertGreater(self.redis_client.delete.call_count, 1, 'expected more than one batch')
+
+if __name__ == '__main__':
+    unittest.main()
+
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/scan_explorer_service/tests/test_manifest.py b/scan_explorer_service/tests/test_manifest.py
index 1d43783..5a911ce 100644
--- a/scan_explorer_service/tests/test_manifest.py
+++ b/scan_explorer_service/tests/test_manifest.py
@@ -5,6 +5,7 @@
 from scan_explorer_service.tests.base import TestCaseDatabase
 from scan_explorer_service.models import Base
 import json
+import opensearchpy
 
 class TestManifest(TestCaseDatabase):
 
@@ -62,6 +63,117 @@ def test_get_canvas(self):
         self.assertStatus(r, 200)
         self.assertEqual(data['@type'], 'sc:Canvas')
 
+    def test_canvas_abstract_link_stays_on_the_readers_host(self):
+        """A hardcoded host would send SciX readers to ADS from the 'About this item' panel."""
+        bibcode = self.article.bibcode
+        url = url_for("manifest.get_manifest", id=self.article.id)
+        r = self.client.get(url)
+        self.assertStatus(r, 200)
+        canvases = json.loads(r.data)['sequences'][0]['canvases']
+        values = [m['value'] for c in canvases for m in c.get('metadata', [])]
+        self.assertTrue(values, 'expected canvas metadata to be present')
+        for value in values:
+            self.assertNotIn('http://', value)
+            self.assertNotIn('https://', value)
+            self.assertIn(f'href="/abs/{bibcode}/abstract"', value)
+
+    def test_a_bibcode_cannot_inject_markup_into_canvas_metadata(self):
+        """The manifest is a public document; a malicious bibcode must not become live markup."""
+        hostile = '"><img src=x onerror=alert(1)>'
+        article = Article(bibcode=hostile, collection_id=self.collection.id)
+        self.app.db.session.add(article)
+        self.app.db.session.commit()
+        self.page.articles.append(article)
+        self.app.db.session.commit()
+
+        r = self.client.get(url_for("manifest.get_manifest", id=self.collection.id))
+        self.assertStatus(r, 200)
+        values = [m['value'] for c in json.loads(r.data)['sequences'][0]['canvases']
+                  for m in c.get('metadata', [])]
+        self.assertTrue(values)
+        for value in values:
+            self.assertNotIn(hostile, value, 'the bibcode must not appear unescaped')
+            self.assertNotIn('<img', value)
+            self.assertNotIn('"><', value)
+        self.assertTrue(any('&lt;img' in v for v in values), 'the markup must survive as escaped text')
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_skips_hits_that_have_no_highlight(self, OpenSearch):
+        """A stop word is analyzed away, so every page matches with no highlight to show."""
+        hits = [{'_source': {'page_id': self.page.id, 'volume_id': self.page.collection_id,
+                             'page_label': self.page.label,
+                             'page_number': self.page.volume_running_page_num}}]
+        OpenSearch.return_value.search.return_value = {
+            "hits": {"total": {"value": 1, "relation": "eq"}, "max_score": None, "hits": hits}}
+
+        url = url_for("manifest.search", id=self.article.id, q='the')
+        r = self.client.get(url)
+        data = json.loads(r.data)
+        self.assertStatus(r, 200)
+        self.assertEqual(data['@type'], 'sc:AnnotationList')
+        self.assertEqual(data.get('resources', []), [])
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_reports_an_internal_failure_as_json_500(self, OpenSearch):
+        """Our own failure is a 500, in JSON, and must not leak the exception text."""
+        OpenSearch.return_value.search.side_effect = RuntimeError('could not connect to db.internal')
+
+        url = url_for("manifest.search", id=self.article.id, q='gas')
+        r = self.client.get(url)
+        self.assertStatus(r, 500)
+        self.assertIn('application/json', r.content_type)
+        self.assertNotIn('db.internal', r.data.decode())
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_reports_a_search_outage_as_503(self, OpenSearch):
+        """A backend outage must be a 5xx so it is visible to alerting."""
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.ConnectionError(
+            'N/A', 'connection refused', Exception('refused'))
+
+        url = url_for("manifest.search", id=self.article.id, q='gas')
+        r = self.client.get(url)
+        self.assertStatus(r, 503)
+        self.assertIn('unavailable', json.loads(r.data)['message'].lower())
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_reports_a_rejected_query_as_400(self, OpenSearch):
+        """OpenSearch rejecting the query is the caller's problem, not an outage."""
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.RequestError(
+            400, 'search_phase_execution_exception', {'error': 'bad query'})
+
+        url = url_for("manifest.search", id=self.article.id, q='gas')
+        r = self.client.get(url)
+        self.assertStatus(r, 400)
+
+
+class TestCollectionManifest(TestCaseDatabase):
+
+    @patch('opensearchpy.OpenSearch')
+    def test_a_missing_index_is_our_fault_not_an_outage(self, OpenSearch):
+        """A renamed index must not read as 'OpenSearch is down' forever."""
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.NotFoundError(
+            404, 'index_not_found_exception', {'error': 'no such index'})
+
+        r = self.client.get(url_for("manifest.search", id=self.article.id, q='gas'))
+        self.assertStatus(r, 500)
+        self.assertNotIn('no such index', r.data.decode())
+
+    @patch('opensearchpy.OpenSearch')
+    def test_a_bad_credential_is_our_fault_not_an_outage(self, OpenSearch):
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.AuthenticationException(
+            401, 'security_exception', {'error': 'bad credentials'})
+
+        r = self.client.get(url_for("manifest.search", id=self.article.id, q='gas'))
+        self.assertStatus(r, 500)
+
+    @patch('opensearchpy.OpenSearch')
+    def test_an_opensearch_server_error_is_an_outage(self, OpenSearch):
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.TransportError(
+            503, 'search_phase_execution_exception', {'error': 'overloaded'})
+
+        r = self.client.get(url_for("manifest.search", id=self.article.id, q='gas'))
+        self.assertStatus(r, 503)
+
     @patch('opensearchpy.OpenSearch')
     def test_search_article_with_highlight(self, OpenSearch):
         open_search_highlight_response = {"hits":{"total":{"value":1,"relation":"eq"},"max_score":None,"hits":[{'_source':{'page_id':self.page.id, 'volume_id':self.page.collection_id, 'page_label':self.page.label, 'page_number': self.page.volume_running_page_num}, "highlight":{'text':'some <b>highlighted</b> text'}}]}}
diff --git a/scan_explorer_service/tests/test_metadata.py b/scan_explorer_service/tests/test_metadata.py
index 26e7a9e..6a3e625 100644
--- a/scan_explorer_service/tests/test_metadata.py
+++ b/scan_explorer_service/tests/test_metadata.py
@@ -5,6 +5,7 @@
 from scan_explorer_service.tests.base import TestCaseDatabase
 from scan_explorer_service.models import Base
 import json
+import opensearchpy
 
 class TestMetadata(TestCaseDatabase):
 
@@ -143,6 +144,124 @@ def test_query_parsing_failures(self):
         r = self.client.get(url)
         self.assertStatus(r, 400)
 
+    def test_bad_pagination_is_a_client_error(self):
+        """Invalid page or limit is the caller's mistake, so it must not read as an outage."""
+        for endpoint in ("metadata.article_search", "metadata.collection_search", "metadata.page_search"):
+            for params in ({"limit": -1}, {"limit": 0}, {"page": 0}, {"page": -3}):
+                url = url_for(endpoint, q='volume:1', **params)
+                r = self.client.get(url)
+                self.assertStatus(r, 400, f"{endpoint} with {params}")
+                self.assertIn('application/json', r.content_type)
+
+    @patch('opensearchpy.OpenSearch')
+    def test_deep_pagination_is_rejected_before_opensearch_is_called(self, OpenSearch):
+        """Paging past the result window is the caller's mistake, not a backend outage."""
+        window = self.app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+        limit = 5
+        url = url_for("metadata.page_search", q='volume:1', page=window // limit + 1, limit=limit)
+        r = self.client.get(url)
+        self.assertStatus(r, 400)
+        self.assertIn('searchable window', json.loads(r.data)['message'])
+        OpenSearch.return_value.search.assert_not_called()
+
+    @patch('opensearchpy.OpenSearch')
+    def test_the_last_advertised_page_is_reachable(self, OpenSearch):
+        """pageCount must never name a page the result-window guard would reject."""
+        window = self.app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+        limit = 3
+        OpenSearch.return_value.search.return_value = {
+            "hits": {"total": {"value": 50000, "relation": "eq"}, "max_score": None, "hits": []},
+            "aggregations": {"total_count": {"value": 50000}, "ids": {"buckets": []}}}
+
+        r = self.client.get(url_for("metadata.page_search", q='volume:1', page=1, limit=limit))
+        self.assertStatus(r, 200)
+        advertised = json.loads(r.data)['pageCount']
+        self.assertLessEqual(advertised * limit, window,
+                             'the advertised final page must sit inside the window')
+
+    @patch('opensearchpy.OpenSearch')
+    def test_the_page_exactly_on_the_window_boundary_is_allowed(self, OpenSearch):
+        window = self.app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+        limit = 5
+        OpenSearch.return_value.search.return_value = {
+            "hits": {"total": {"value": 50000, "relation": "eq"}, "max_score": None, "hits": []}}
+
+        url = url_for("metadata.page_search", q='volume:1', page=window // limit, limit=limit)
+        self.assertStatus(self.client.get(url), 200)
+
+    @patch('opensearchpy.OpenSearch')
+    def test_non_numeric_pagination_falls_back_to_the_default(self, OpenSearch):
+        """Deliberate: a page or limit that is not a number is served as the default, not rejected."""
+        OpenSearch.return_value.search.return_value = {
+            "hits": {"total": {"value": 0, "relation": "eq"}, "max_score": None, "hits": []},
+            "aggregations": {"total_count": {"value": 0}, "ids": {"buckets": []}}}
+
+        for params in ({"page": "abc"}, {"limit": "abc"}, {"page": "1.5"}):
+            r = self.client.get(url_for("metadata.article_search", q='volume:1', **params))
+            self.assertStatus(r, 200, str(params))
+
+    def test_a_limit_beyond_the_result_window_is_a_client_error(self):
+        """Without this bound, article_search returns 400 or 200 for one request depending on data."""
+        window = self.app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+        r = self.client.get(url_for("metadata.article_search", q='volume:1', limit=window + 1))
+        self.assertStatus(r, 400)
+
+    @patch('opensearchpy.OpenSearch')
+    def test_deep_article_pages_are_not_capped_when_nothing_matches(self, OpenSearch):
+        """article_search probes the page index for a count; that probe must not cap the caller."""
+        OpenSearch.return_value.search.return_value = {
+            "hits": {"total": {"value": 0, "relation": "eq"}, "max_score": None, "hits": []},
+            "aggregations": {"total_count": {"value": 0}, "ids": {"buckets": []}}}
+
+        url = url_for("metadata.article_search", q='volume:1', page=2001, limit=5)
+        self.assertStatus(self.client.get(url), 200)
+
+    SEARCH_ENDPOINTS = ("metadata.article_search", "metadata.collection_search", "metadata.page_search")
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_outage_is_reported_as_503(self, OpenSearch):
+        """Every search endpoint must report an unreachable backend as an outage."""
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.ConnectionError(
+            'N/A', 'connection refused', Exception('refused'))
+        for endpoint in self.SEARCH_ENDPOINTS:
+            r = self.client.get(url_for(endpoint, q='volume:1'))
+            self.assertStatus(r, 503, endpoint)
+            self.assertIn('unavailable', json.loads(r.data)['message'].lower())
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_misconfiguration_is_ours_not_an_outage(self, OpenSearch):
+        """A missing index is a 500, so alerting sees it and nobody reads it as 'OpenSearch is down'."""
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.NotFoundError(
+            404, 'index_not_found_exception', {'error': 'no such index'})
+        for endpoint in self.SEARCH_ENDPOINTS:
+            r = self.client.get(url_for(endpoint, q='volume:1'))
+            self.assertStatus(r, 500, endpoint)
+            self.assertNotIn('no such index', r.data.decode())
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_internal_failure_does_not_leak_its_text(self, OpenSearch):
+        OpenSearch.return_value.search.side_effect = RuntimeError('could not connect to db.internal')
+        for endpoint in self.SEARCH_ENDPOINTS:
+            r = self.client.get(url_for(endpoint, q='volume:1'))
+            self.assertStatus(r, 500, endpoint)
+            self.assertIn('application/json', r.content_type)
+            self.assertNotIn('db.internal', r.data.decode())
+
+    @patch('opensearchpy.OpenSearch')
+    def test_search_rejected_query_is_a_client_error(self, OpenSearch):
+        OpenSearch.return_value.search.side_effect = opensearchpy.exceptions.RequestError(
+            400, 'search_phase_execution_exception', {'error': 'bad query'})
+        for endpoint in self.SEARCH_ENDPOINTS:
+            r = self.client.get(url_for(endpoint, q='volume:1'))
+            self.assertStatus(r, 400, endpoint)
+
+    @patch('opensearchpy.OpenSearch')
+    def test_ocr_failure_follows_the_same_contract(self, OpenSearch):
+        OpenSearch.return_value.search.side_effect = RuntimeError('could not connect to db.internal')
+        r = self.client.get(url_for("metadata.get_page_ocr", id=self.article.id, page_number=1))
+        self.assertStatus(r, 500)
+        self.assertNotIn('db.internal', r.data.decode())
+
     @patch('opensearchpy.OpenSearch')
     def test_query_parsing_sucess(self, OpenSearch):
         es = OpenSearch.return_value
@@ -234,6 +353,57 @@ def test_put_collection_with_articles(self):
         pages = self.app.db.session.query(Page).filter(Page.collection_id == collection_id).all()
         self.assertEqual(len(pages), 1)
 
+    @patch('scan_explorer_service.views.metadata.cache_delete_manifests')
+    def test_put_collection_invalidates_its_articles(self, mock_delete):
+        """The collection's pages changed, so every article manifest in it is now stale."""
+        collection_json = {
+            'type': 'type',
+            'journal': self.collection.journal,
+            'volume': self.collection.volume,
+            'pages': [{
+                'name': 'pageA',
+                'color_type': 'BW',
+                'page_type': 'Normal',
+                'label': '1',
+                'width': 100,
+                'height': 100,
+                'volume_running_page_num': 1,
+                'articles': [{'bibcode': '2000ApJ...001..001A'}],
+            }]
+        }
+        url = url_for("metadata.put_collection")
+        self.assertStatus(self.client.put(url, json=collection_json), 200)
+
+        mock_delete.assert_called_once()
+        invalidated = set(mock_delete.call_args[0][0])
+        self.assertIn(self.collection.id, invalidated)
+        self.assertIn('2000ApJ...001..001A', invalidated)
+        self.assertIn(self.article.id, invalidated)
+        self.assertIn(self.article2.id, invalidated)
+
+    @patch('scan_explorer_service.views.metadata.cache_delete_manifests')
+    def test_put_page_invalidates_its_articles_and_collection(self, mock_delete):
+        page_json = dict(self.page_json)
+        page_json['articles'] = [{'bibcode': self.article.bibcode}]
+        collection_id = self.collection.id
+        bibcode = self.article.bibcode
+
+        r = self.client.put(url_for("metadata.put_page"), json=page_json)
+        self.assertStatus(r, 200)
+        invalidated = set(mock_delete.call_args[0][0])
+        self.assertIn(collection_id, invalidated)
+        self.assertIn(bibcode, invalidated)
+
+    @patch('scan_explorer_service.views.metadata.cache_delete_manifests')
+    def test_put_article_invalidates_itself_and_its_collection(self, mock_delete):
+        collection_id = self.collection.id
+        r = self.client.put(url_for("metadata.put_article"),
+                            json={'bibcode': '2001ApJ...555..555Z', 'collection_id': collection_id})
+        self.assertStatus(r, 200)
+        invalidated = set(mock_delete.call_args[0][0])
+        self.assertIn('2001ApJ...555..555Z', invalidated)
+        self.assertIn(collection_id, invalidated)
+
     def test_put_collection_deduplicates_articles(self):
         """An article appearing in multiple pages is inserted only once."""
         collection_json = {
diff --git a/scan_explorer_service/tests/test_perf.py b/scan_explorer_service/tests/test_perf.py
index 1bd2762..5af44b6 100644
--- a/scan_explorer_service/tests/test_perf.py
+++ b/scan_explorer_service/tests/test_perf.py
@@ -5,7 +5,7 @@
 from unittest.mock import patch, MagicMock
 from scan_explorer_service.tests.base import TestCaseDatabase
 from scan_explorer_service.models import Article, Base, Collection, Page
-from scan_explorer_service.utils.cache import cache_set_manifest, MANIFEST_CACHE_PREFIX
+from scan_explorer_service.utils.cache import cache_set_manifest, MANIFEST_CACHE_PREFIX, _variant_scope
 from scan_explorer_service.views.image_proxy import fetch_images
 import scan_explorer_service.utils.cache as cache_mod
 
@@ -95,7 +95,7 @@ def mock_delete(key):
     def test_cache_hit_returns_cached_json(self):
         """Verifies that a cached manifest is returned directly without regeneration."""
         mock_r, store = self._mock_redis()
-        store[MANIFEST_CACHE_PREFIX + self.article.id] = ('{"@type":"sc:Manifest","cached":true}', time.monotonic() + 3600)
+        store[MANIFEST_CACHE_PREFIX + _variant_scope() + self.article.id] = ('{"@type":"sc:Manifest","cached":true}', time.monotonic() + 3600)
 
         url = url_for("manifest.get_manifest", id=self.article.id)
         r = self.client.get(url)
@@ -106,7 +106,7 @@ def test_cache_hit_returns_cached_json(self):
     def test_cache_hit_returns_correct_content_type(self):
         """Verifies that cached manifest responses have application/json content type."""
         mock_r, store = self._mock_redis()
-        store[MANIFEST_CACHE_PREFIX + self.collection.id] = ('{"@type":"sc:Manifest"}', time.monotonic() + 3600)
+        store[MANIFEST_CACHE_PREFIX + _variant_scope() + self.collection.id] = ('{"@type":"sc:Manifest"}', time.monotonic() + 3600)
 
         url = url_for("manifest.get_manifest", id=self.collection.id)
         r = self.client.get(url)
@@ -128,12 +128,14 @@ def tracking_setex(key, ttl, val):
         cache_set_manifest(self.article.id, '{"@type":"sc:Manifest"}')
 
         self.assertEqual(len(setex_calls), 1)
-        self.assertEqual(setex_calls[0], MANIFEST_CACHE_PREFIX + self.article.id)
+        self.assertEqual(
+            setex_calls[0],
+            'scan:manifest:http://localhost:8184/v1/scan:' + self.article.id)
 
     def test_cached_manifest_skips_manifest_factory(self):
         """Verifies that manifest_factory is not called when the manifest is cached."""
         mock_r, store = self._mock_redis()
-        store[MANIFEST_CACHE_PREFIX + self.article.id] = ('{"@type":"sc:Manifest"}', time.monotonic() + 3600)
+        store[MANIFEST_CACHE_PREFIX + _variant_scope() + self.article.id] = ('{"@type":"sc:Manifest"}', time.monotonic() + 3600)
 
         with patch('scan_explorer_service.views.manifest.manifest_factory') as mock_factory:
             url = url_for("manifest.get_manifest", id=self.article.id)
@@ -148,7 +150,7 @@ def test_404_not_cached(self):
         url = url_for("manifest.get_manifest", id='nonexistent')
         r = self.client.get(url)
         self.assertStatus(r, 404)
-        self.assertNotIn(MANIFEST_CACHE_PREFIX + 'nonexistent', store)
+        self.assertNotIn(MANIFEST_CACHE_PREFIX + _variant_scope() + 'nonexistent', store)
 
     def test_redis_unavailable_falls_through(self):
         """Verifies that the endpoint still works when Redis is unavailable."""
diff --git a/scan_explorer_service/tests/test_proxy.py b/scan_explorer_service/tests/test_proxy.py
index 51bcf02..80f1109 100644
--- a/scan_explorer_service/tests/test_proxy.py
+++ b/scan_explorer_service/tests/test_proxy.py
@@ -146,6 +146,25 @@ def test_get_thumbnail(self, mock_request):
         assert(response.is_streamed)
         assert(response.status_code == 200)
 
+    @patch('scan_explorer_service.views.image_proxy.item_thumbnail')
+    def test_thumbnail_internal_failure_does_not_leak_its_text(self, mock_thumbnail):
+        mock_thumbnail.side_effect = RuntimeError('could not connect to db.internal')
+        r = self.client.get(url_for('proxy.image_proxy_thumbnail', id=self.article.id, type='article'))
+        self.assertEqual(r.status_code, 500)
+        self.assertNotIn('db.internal', r.data.decode())
+
+    def test_thumbnail_for_an_item_with_no_pages_is_a_client_error(self):
+        r = self.client.get(url_for('proxy.image_proxy_thumbnail', id='NOT.A.REAL.ID', type='article'))
+        self.assertEqual(r.status_code, 400)
+        self.assertIn('No pages found', json.loads(r.data)['Message'])
+
+    @patch('scan_explorer_service.views.image_proxy.get_item')
+    def test_pdf_internal_failure_does_not_leak_its_text(self, mock_get_item):
+        mock_get_item.side_effect = RuntimeError('could not connect to db.internal')
+        r = self.client.get(url_for('proxy.pdf_save', id=self.article.id))
+        self.assertEqual(r.status_code, 500)
+        self.assertNotIn('db.internal', r.data.decode())
+
     def test_get_item(self):
         """Test retrieving an item by its ID"""
         with self.app.app_context():
diff --git a/scan_explorer_service/utils/cache.py b/scan_explorer_service/utils/cache.py
index 3f7c034..3a60fa9 100644
--- a/scan_explorer_service/utils/cache.py
+++ b/scan_explorer_service/utils/cache.py
@@ -3,6 +3,7 @@
 import threading
 import json as json_lib
 from flask import current_app
+from scan_explorer_service.utils.utils import proxy_url
 
 logger = logging.getLogger(__name__)
 
@@ -10,6 +11,8 @@
 MANIFEST_CACHE_PREFIX = 'scan:manifest:'
 SEARCH_CACHE_TTL = 60
 SEARCH_CACHE_PREFIX = 'scan:search:'
+MANIFEST_SCOPE_SET = 'scan:manifest:scopes'
+DELETE_BATCH_SIZE = 500
 
 _redis_client = None
 _redis_lock = threading.Lock()
@@ -40,13 +43,23 @@ def _reset_redis():
     _redis_client = None
 
 
+def _variant_scope():
+    """Return the deployment scope for cache keys.
+
+    Cached documents embed absolute URLs built from PROXY_SERVER and PROXY_PREFIX,
+    so two deployments serving different hostnames must not share cache entries.
+    """
+    server, prefix = proxy_url()
+    return f'{server}/{prefix}:'
+
+
 def _redis_get(prefix, key):
-    """Fetch a cached value by prefix + key, returning None on miss or failure."""
+    """Fetch a cached value by prefix + scope + key, returning None on miss or failure."""
     r = _get_redis()
     if r is None:
         return None
     try:
-        return r.get(prefix + key)
+        return r.get(prefix + _variant_scope() + key)
     except redis.ConnectionError:
         _reset_redis()
         return None
@@ -54,45 +67,60 @@ def _redis_get(prefix, key):
         return None
 
 
-def _redis_set(prefix, key, value, ttl):
-    """Store a value in Redis with the given prefix, key, and TTL."""
+def _redis_set(prefix, key, value, ttl, scope_set=None):
+    """Store a value under prefix + scope + key, recording the scope when scope_set is given."""
     r = _get_redis()
     if r is None:
         return
     try:
-        r.setex(prefix + key, ttl, value)
+        scope = _variant_scope()
+        r.setex(prefix + scope + key, ttl, value)
+        if scope_set:
+            r.sadd(scope_set, scope)
     except redis.ConnectionError:
         _reset_redis()
     except Exception:
         logger.debug("Failed to write cache for key %s%s", prefix, key, exc_info=True)
 
 
-def _redis_delete(prefix, key):
-    """Delete a cached entry by prefix + key."""
-    r = _get_redis()
-    if r is None:
-        return
-    try:
-        r.delete(prefix + key)
-    except redis.ConnectionError:
-        _reset_redis()
-    except Exception:
-        logger.debug("Failed to delete cache for key %s%s", prefix, key, exc_info=True)
-
-
 def cache_get_manifest(key):
     """Fetch a cached manifest JSON string."""
     return _redis_get(MANIFEST_CACHE_PREFIX, key)
 
 
 def cache_set_manifest(key, json_str):
-    """Cache a manifest JSON string with 1-hour TTL."""
-    _redis_set(MANIFEST_CACHE_PREFIX, key, json_str, MANIFEST_CACHE_TTL)
+    """Cache a manifest JSON string with 24-hour TTL."""
+    _redis_set(MANIFEST_CACHE_PREFIX, key, json_str, MANIFEST_CACHE_TTL, MANIFEST_SCOPE_SET)
 
 
 def cache_delete_manifest(key):
-    """Invalidate a cached manifest. Called when a collection is updated via PUT."""
-    _redis_delete(MANIFEST_CACHE_PREFIX, key)
+    """Invalidate one manifest id in every deployment scope that has cached it."""
+    cache_delete_manifests([key])
+
+
+def cache_delete_manifests(keys):
+    """Invalidate several manifest ids in every deployment scope that has cached them.
+
+    A collection PUT reaches only one deployment, but the update applies to all of
+    them, so each id is removed under every scope recorded in MANIFEST_SCOPE_SET.
+    """
+    keys = list(dict.fromkeys(keys))
+    if not keys:
+        return
+    r = _get_redis()
+    if r is None:
+        return
+    try:
+        scopes = set(r.smembers(MANIFEST_SCOPE_SET))
+        scopes.add(_variant_scope())
+        for scope in scopes:
+            names = [MANIFEST_CACHE_PREFIX + scope + key for key in keys]
+            for i in range(0, len(names), DELETE_BATCH_SIZE):
+                r.delete(*names[i:i + DELETE_BATCH_SIZE])
+    except (redis.ConnectionError, redis.TimeoutError):
+        _reset_redis()
+    except Exception:
+        logger.warning("Failed to delete cached manifests for %s", keys, exc_info=True)
 
 
 def cache_get_search(key):
diff --git a/scan_explorer_service/utils/db_utils.py b/scan_explorer_service/utils/db_utils.py
index ccc74bd..d8e418a 100644
--- a/scan_explorer_service/utils/db_utils.py
+++ b/scan_explorer_service/utils/db_utils.py
@@ -80,14 +80,14 @@ def article_thumbnail(session, id):
     page = session.query(Page).join(Article, Page.articles).filter(
                 Article.id == id).order_by(Page.volume_running_page_num.asc()).first()
     if page is None:
-        raise Exception(f"No pages found for article {id}")
+        raise ValueError(f"No pages found for article {id}")
     return page.thumbnail_url
 
 def collection_thumbnail(session, id):
     page = session.query(Page).filter(Page.collection_id == id).order_by(
         Page.volume_running_page_num.asc()).first()
     if page is None:
-        raise Exception(f"No pages found for collection {id}")
+        raise ValueError(f"No pages found for collection {id}")
     return page.thumbnail_url
 
 def page_thumbnail(session, id):
@@ -102,4 +102,4 @@ def item_thumbnail(session, id, type):
     elif type == 'collection':
         return collection_thumbnail(session, id)
     else:
-        raise Exception("Invalid type")
+        raise ValueError("Invalid type")
diff --git a/scan_explorer_service/utils/search_utils.py b/scan_explorer_service/utils/search_utils.py
index a0c0442..fc11c24 100644
--- a/scan_explorer_service/utils/search_utils.py
+++ b/scan_explorer_service/utils/search_utils.py
@@ -1,4 +1,5 @@
 import math
+from flask import current_app
 from scan_explorer_service.models import PageType, PageColor
 import shlex
 import enum
@@ -65,6 +66,13 @@ def parse_query_args(args):
 
     page = args.get('page', 1, int)
     limit = args.get('limit', 10, int)
+    window = current_app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+    if page < 1:
+        raise ValueError('page must be 1 or greater')
+    if limit < 1:
+        raise ValueError('limit must be 1 or greater')
+    if limit > window:
+        raise ValueError(f'limit must be {window} or less')
     sort_raw = args.get('sort')
     sort = parse_sorting_option(sort_raw)
     return qs, qs_dict, page, limit, sort
@@ -134,7 +142,7 @@ def check_page_type(qs_dict: dict):
             if page_type.replace('"','').lower() == p.name.lower():
                 qs_dict[SearchOptions.PageType.value] = p.name
                 return
-        raise Exception("%s is not a valid page type, %s is possible choices"% (page_type, str(valid_types)))
+        raise ValueError("%s is not a valid page type, %s is possible choices"% (page_type, str(valid_types)))
 
 def check_page_color(qs_dict: dict):
     """Validate and normalize the pagecolor filter value to match the PageColor enum."""
@@ -149,7 +157,7 @@ def check_page_color(qs_dict: dict):
             if page_color.replace('"','').lower() == p.name.lower():
                 qs_dict[SearchOptions.PageColor.value] = p.name
                 return
-        raise Exception("%s is not a valid page color, %s is possible choices"% (page_color, str(valid_types)))
+        raise ValueError("%s is not a valid page color, %s is possible choices"% (page_color, str(valid_types)))
 
 def check_project(qs_dict: dict):
     """Validate and normalize the project filter value against known project names."""
@@ -165,7 +173,7 @@ def check_project(qs_dict: dict):
             if project.lower() == p.lower():
                 qs_dict[SearchOptions.Project.value] = p.replace('Microfilm Scanning', 'Historical Literature')
                 return
-        raise Exception("%s is not a valid project, %s is possible choices"% (project, str(valid_types)))
+        raise ValueError("%s is not a valid project, %s is possible choices"% (project, str(valid_types)))
 
 def serialize_os_agg_page_bucket(bucket: dict):
     """Convert an OpenSearch page hit into a page result dict with collection and label info."""
@@ -180,7 +188,8 @@ def serialize_os_agg_page_bucket(bucket: dict):
 def serialize_os_page_result(result: dict, page: int, limit: int, contentQuery):
     """Serialize an OpenSearch page search response into a paginated result dict."""
     total_count = result['hits']['total']['value']
-    page_count = int(math.ceil(min(total_count,10000) / limit))    
+    window = current_app.config.get('OPEN_SEARCH_MAX_RESULT_WINDOW', 10000)
+    page_count = min(int(math.ceil(total_count / limit)), window // limit)
     es_buckets = result['hits']['hits']
 
     return {'page': page, 'pageCount': page_count, 'limit': limit, 'total': total_count, 'query': contentQuery,
@@ -190,7 +199,7 @@ def serialize_os_page_ocr_result(result: dict):
     """Extract the OCR text from an OpenSearch page result, raising if no page is found."""
     es_buckets = result['hits']['hits']
     if len(es_buckets) < 1:
-        raise Exception("No page with those parameters found")
+        raise ValueError("No page with those parameters found")
     return es_buckets[0]['_source']['text']
 
 def serialize_os_agg_collection_bucket(bucket: dict):
diff --git a/scan_explorer_service/views/image_proxy.py b/scan_explorer_service/views/image_proxy.py
index e00f326..8fa3f9f 100644
--- a/scan_explorer_service/views/image_proxy.py
+++ b/scan_explorer_service/views/image_proxy.py
@@ -81,9 +81,11 @@ def image_proxy_thumbnail():
             path = path.replace(remove, '')
 
             return image_proxy(path)
+    except ValueError as e:
+        return jsonify(Message=str(e)), 400
     except Exception as e:
         current_app.logger.exception(f'{e}')
-        return jsonify(Message=str(e)), 400
+        return jsonify(Message='Internal error'), 500
 
 def get_item(session, id):
     """Look up an Article or Collection by ID, raising if neither exists."""
@@ -91,7 +93,7 @@ def get_item(session, id):
                 session.query(Article).filter(Article.id == id).one_or_none()
                 or session.query(Collection).filter(Collection.id == id).one_or_none())
     if not item:
-        raise Exception("ID: " + str(id) + " not found")
+        raise ValueError("ID: " + str(id) + " not found")
 
     return item
 
@@ -102,7 +104,7 @@ def get_pages(item, session, page_start, page_end, page_limit):
     if isinstance(item, Article):
         first_page = item.pages.first()
         if first_page is None:
-            raise Exception(f"No pages found for article {item.id}")
+            raise ValueError(f"No pages found for article {item.id}")
         start_page = first_page.volume_running_page_num
         query = session.query(Page).filter(Page.articles.any(Article.id == item.id),
             Page.volume_running_page_num  >= page_start + start_page - 1,
@@ -226,5 +228,8 @@ def pdf_save():
 
             response = generate_pdf(item, session, page_start, page_end, page_limit, memory_limit)
             return response
-    except Exception as e:
+    except ValueError as e:
         return jsonify(Message=str(e)), 400
+    except Exception as e:
+        current_app.logger.exception(f'{e}')
+        return jsonify(Message='Internal error'), 500
diff --git a/scan_explorer_service/views/manifest.py b/scan_explorer_service/views/manifest.py
index d0923d0..d22f564 100644
--- a/scan_explorer_service/views/manifest.py
+++ b/scan_explorer_service/views/manifest.py
@@ -4,6 +4,7 @@
 from scan_explorer_service.extensions import manifest_factory
 from scan_explorer_service.models import Article, Page, Collection
 from flask_discoverer import advertise
+from scan_explorer_service.views.view_utils import ApiErrors, search_error_response
 from scan_explorer_service.open_search import EsFields, text_search_highlight
 from scan_explorer_service.utils.utils import proxy_url, url_for_proxy
 from scan_explorer_service.utils.cache import (
@@ -104,25 +105,30 @@ def search(id: str):
     if cached is not None:
         return Response(cached, content_type='application/json')
 
-    with current_app.session_scope() as session:
-        item: Union[Article, Collection] = (
-                    session.query(Article).filter(Article.id == id).one_or_none()
-                    or session.query(Collection).filter(Collection.id == id).one_or_none())
-        if item:
-            annotation_list = manifest_factory.annotationList(request.url)
-            annotation_list.resources = []
-
-            es_field = EsFields.article_id if isinstance(item, Article) else EsFields.volume_id
-            results = text_search_highlight(query, es_field, item.id)
-
-            for res in results:
-                annotation = annotation_list.annotation(res['page_id'])
-                canvas_slice_url = url_for_proxy('manifest.get_canvas', page_id=res['page_id'])
-                annotation.on = canvas_slice_url
-                highlight_text = "<br><br>".join(res['highlight']).replace("em>", "b>")
-                annotation.text(highlight_text, format="text/html")
-
-            return to_json_and_cache(annotation_list, cache_set_search, cache_key)
-
-        else:
-            return jsonify(exception='Article or volume not found'), 404
+    try:
+        with current_app.session_scope() as session:
+            item: Union[Article, Collection] = (
+                        session.query(Article).filter(Article.id == id).one_or_none()
+                        or session.query(Collection).filter(Collection.id == id).one_or_none())
+            if item:
+                annotation_list = manifest_factory.annotationList(request.url)
+                annotation_list.resources = []
+
+                es_field = EsFields.article_id if isinstance(item, Article) else EsFields.volume_id
+                results = text_search_highlight(query, es_field, item.id)
+
+                for res in results:
+                    highlight_text = "<br><br>".join(res['highlight']).replace("em>", "b>")
+                    if not highlight_text:
+                        continue
+                    annotation = annotation_list.annotation(res['page_id'])
+                    canvas_slice_url = url_for_proxy('manifest.get_canvas', page_id=res['page_id'])
+                    annotation.on = canvas_slice_url
+                    annotation.text(highlight_text, format="text/html")
+
+                return to_json_and_cache(annotation_list, cache_set_search, cache_key)
+
+            else:
+                return jsonify(exception='Article or volume not found'), 404
+    except Exception as e:
+        return search_error_response(e)
diff --git a/scan_explorer_service/views/metadata.py b/scan_explorer_service/views/metadata.py
index c8f2b49..8d04438 100644
--- a/scan_explorer_service/views/metadata.py
+++ b/scan_explorer_service/views/metadata.py
@@ -4,10 +4,11 @@
 from scan_explorer_service.utils.db_utils import article_get_or_create, article_overwrite, collection_overwrite, page_get_or_create, page_overwrite
 from scan_explorer_service.models import Article, Collection, Page, page_article_association_table
 from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy import or_
 from flask_discoverer import advertise
 from scan_explorer_service.utils.search_utils import *
-from scan_explorer_service.views.view_utils import ApiErrors
-from scan_explorer_service.utils.cache import cache_delete_manifest, cache_get_search, cache_set_search
+from scan_explorer_service.views.view_utils import ApiErrors, search_error_response
+from scan_explorer_service.utils.cache import cache_delete_manifests, cache_get_search, cache_set_search
 from scan_explorer_service.open_search import EsFields, page_os_search, aggregate_search, page_ocr_os_search
 import opensearchpy
 import requests
@@ -64,6 +65,9 @@ def put_article():
             try:
                 article = Article(**json)
                 article_overwrite(session, article)
+                session.commit()
+                stale = [i for i in (article.id, article.collection_id) if i]
+                cache_delete_manifests(stale)
                 return jsonify({'id': article.bibcode}), 200
             except Exception:
                 session.rollback()
@@ -81,6 +85,14 @@ def put_collection():
         with current_app.session_scope() as session:
             try:
                 collection = Collection(**json)
+                stale_article_ids = {
+                    row[0] for row in session.query(Article.id).filter(
+                        or_(
+                            Article.collection_id == collection.id,
+                            Article.pages.any(Page.collection_id == collection.id),
+                        )
+                    ).all()
+                }
                 collection_overwrite(session, collection)
 
                 now = datetime.now(timezone.utc).replace(tzinfo=None)
@@ -132,7 +144,8 @@ def put_collection():
                         pg_insert(page_article_association_table).values(page_article_data).on_conflict_do_nothing()
                     )
                 session.commit()
-                cache_delete_manifest(collection.id)
+                stale_article_ids.update(articles_data.keys())
+                cache_delete_manifests([collection.id, *stale_article_ids])
 
                 return jsonify({'id': collection.id}), 200
             except Exception:
@@ -160,6 +173,10 @@ def put_page():
                 session.add(page)
                 session.commit()
                 session.refresh(page)
+                stale = [a.id for a in page.articles]
+                if page.collection_id:
+                    stale.append(page.collection_id)
+                cache_delete_manifests(stale)
                 return jsonify({'id': page.id}), 200
             except Exception:
                 session.rollback()
@@ -194,17 +211,13 @@ def article_search():
         collection_count = page_count = 0
         if article_count == 0:
             collection_count = aggregate_search(qs, EsFields.volume_id, page, limit, sort)['aggregations']['total_count']['value']
-            page_count = page_os_search(qs, page, limit, sort)['hits']['total']['value']
+            page_count = page_os_search(qs, 1, limit, sort)['hits']['total']['value']
         agg_limit = current_app.config.get("OPEN_SEARCH_AGG_BUCKET_LIMIT", 10000)
         response_data = serialize_os_article_result(result, page, limit, text_query, collection_count, page_count, agg_limit)
         cache_set_search(cache_key, json_lib.dumps(response_data))
         return jsonify(response_data)
-    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
-        current_app.logger.exception(f"OpenSearch error: {e}")
-        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
-        current_app.logger.exception(f"An exception has occurred: {e}")
-        return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
+        return search_error_response(e)
 
 
 @advertise(scopes=['api'], rate_limit=[5000, 3600*24])
@@ -226,11 +239,8 @@ def collection_search():
         response_data = serialize_os_collection_result(result, page, limit, text_query, agg_limit)
         cache_set_search(cache_key, json_lib.dumps(response_data))
         return jsonify(response_data)
-    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
-        current_app.logger.exception(f"OpenSearch error: {e}")
-        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
-        return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
+        return search_error_response(e)
 
 @advertise(scopes=['api'], rate_limit=[5000, 3600*24])
 @bp_metadata.route('/page/search', methods=['GET'])
@@ -250,11 +260,8 @@ def page_search():
         response_data = serialize_os_page_result(result, page, limit, text_query)
         cache_set_search(cache_key, json_lib.dumps(response_data))
         return jsonify(response_data)
-    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
-        current_app.logger.exception(f"OpenSearch error: {e}")
-        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
-        return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
+        return search_error_response(e)
 
 @advertise(scopes=['api'], rate_limit=[5000, 3600*24])
 @bp_metadata.route('/page/ocr', methods=['GET'])
@@ -290,8 +297,5 @@ def get_page_ocr():
             cache_set_search(cache_key, ocr_text)
             return current_app.response_class(ocr_text, content_type='text/plain')
 
-    except (opensearchpy.exceptions.ConnectionError, opensearchpy.exceptions.ConnectionTimeout, opensearchpy.exceptions.TransportError) as e:
-        current_app.logger.exception(f"OpenSearch error: {e}")
-        return jsonify(message='Search service temporarily unavailable', type=ApiErrors.SearchError.value), 503
     except Exception as e:
-        return jsonify(message=str(e), type=ApiErrors.SearchError.value), 400
+        return search_error_response(e)
diff --git a/scan_explorer_service/views/view_utils.py b/scan_explorer_service/views/view_utils.py
index 8bf006c..fe4516d 100644
--- a/scan_explorer_service/views/view_utils.py
+++ b/scan_explorer_service/views/view_utils.py
@@ -1,4 +1,39 @@
 import enum
+import opensearchpy
+from flask import current_app, jsonify
+
 
 class ApiErrors(enum.Enum):
-    SearchError = 1
\ No newline at end of file
+    SearchError = 1
+
+
+def search_error_response(exc):
+    """Map a search failure onto the right status: the caller's fault, ours, or a transient outage.
+
+    A 4xx says the caller must change the request, so it is used only for failures the caller
+    can act on. Our own misconfiguration and unexpected failures are 5xx, which keeps them
+    visible to alerting, and their text stays in the log rather than the response.
+    """
+    if isinstance(exc, opensearchpy.exceptions.RequestError):
+        current_app.logger.warning("OpenSearch rejected the query: %s", exc)
+        return jsonify(message=str(exc), type=ApiErrors.SearchError.value), 400
+
+    if isinstance(exc, (opensearchpy.exceptions.ConnectionError,
+                        opensearchpy.exceptions.ConnectionTimeout)):
+        current_app.logger.exception("OpenSearch unreachable: %s", exc)
+        return jsonify(message='Search service temporarily unavailable',
+                       type=ApiErrors.SearchError.value), 503
+
+    if isinstance(exc, opensearchpy.exceptions.TransportError):
+        status = getattr(exc, 'status_code', None)
+        current_app.logger.exception("OpenSearch transport error %s: %s", status, exc)
+        if isinstance(status, int) and status >= 500:
+            return jsonify(message='Search service temporarily unavailable',
+                           type=ApiErrors.SearchError.value), 503
+        return jsonify(message='Internal error', type=ApiErrors.SearchError.value), 500
+
+    if isinstance(exc, ValueError):
+        return jsonify(message=str(exc), type=ApiErrors.SearchError.value), 400
+
+    current_app.logger.exception("An exception has occurred: %s", exc)
+    return jsonify(message='Internal error', type=ApiErrors.SearchError.value), 500