diff --git a/lib/pbench/server/api/resources/__init__.py b/lib/pbench/server/api/resources/__init__.py
index 2d37e8296d..694d941265 100644
--- a/lib/pbench/server/api/resources/__init__.py
+++ b/lib/pbench/server/api/resources/__init__.py
@@ -5,6 +5,7 @@
 from json.decoder import JSONDecodeError
 from logging import Logger
 from typing import Any, Callable, Dict, List, NamedTuple, Optional, Union
+import uuid
 
 from dateutil import parser as date_parser
 from flask import request
@@ -32,11 +33,11 @@
 
 
 class APIAbort(Exception):
-    """
-    Used to report an error and abort if there is a failure in processing of API request.
+    """Used to report an error and abort if there is a failure in processing of
+    API request.
     """
 
-    def __init__(self, http_status: int, message: str = None):
+    def __init__(self, http_status: int, message: Optional[str] = None):
         self.http_status = http_status
         self.message = message if message else HTTPStatus(http_status).phrase
 
@@ -47,9 +48,43 @@ def __str__(self) -> str:
         return self.message
 
 
-class UnauthorizedAccess(APIAbort):
+class APIInternalError(APIAbort):
+    """Used to report a server internal error with a UUID value that connects
+    the string reported to the client with a server log entry to aid analysis
+    by an SRE
     """
-    The user is not authorized for the requested operation on the specified
+
+    def __init__(self, details: str):
+        """Construct an "internal server error" exception object.
+
+        This exception is raised and will be used in the API _dispatch method
+        to generate a JSON error message payload to the client. This message
+        contains a UUID string that uniquely identifies this internal server
+        error context. An internal server log message is generated, with
+        traceback, also containing that UUID value so that a server
+        administrator can investigate the cause of the error.
+
+        NOTE: We want to return minimal explanations of internal errors to the
+        client while capturing detailed information for server developers to
+        determine what happened.
+
+        NOTE: We use a fully formatted "details" message here for convenience;
+        we will report this with `logger.exception`, which is never disabled,
+        so deferring the formatting would have no value.
+
+        Args:
+            details: A detailed message to be logged when this exception is caught
+        """
+        u = uuid.uuid4()
+        super().__init__(
+            http_status=HTTPStatus.INTERNAL_SERVER_ERROR,
+            message=f"Internal Pbench Server Error: log reference {u}",
+        )
+        self.details = f"Internal error {u}: {details}"
+
+
+class UnauthorizedAccess(APIAbort):
+    """The user is not authorized for the requested operation on the specified
     resource.
     """
 
@@ -72,8 +107,7 @@ def __str__(self) -> str:
 
 
 class UnauthorizedAdminAccess(UnauthorizedAccess):
-    """
-    A refinement of the UnauthorizedAccess exception where ADMIN access is
+    """A refinement of the UnauthorizedAccess exception where ADMIN access is
     required and we have no associated resource owner and access.
     """
 
@@ -96,9 +130,7 @@ def __str__(self) -> str:
 
 
 class SchemaError(APIAbort):
-    """
-    Generic base class for errors in processing a JSON schema.
-    """
+    """Generic base class for errors in processing a JSON schema."""
 
     def __init__(self, http_status: int = HTTPStatus.BAD_REQUEST):
         super().__init__(http_status=http_status)
@@ -108,9 +140,8 @@ def __str__(self) -> str:
 
 
 class UnverifiedUser(SchemaError):
-    """
-    Attempt by an unauthenticated client to reference a username in a query. An
-    unauthenticated client does not have the right to look up any username.
+    """Attempt by an unauthenticated client to reference a username in a query.
+    An unauthenticated client does not have the right to look up any username.
 
     HTTPStatus.UNAUTHORIZED tells the client that the operation might succeed
     if the request is retried with authentication. (A UI might redirect to a
@@ -126,17 +157,14 @@ def __str__(self):
 
 
 class InvalidRequestPayload(SchemaError):
-    """
-    A required client JSON input document is missing.
-    """
+    """A required client JSON input document is missing."""
 
     def __str__(self) -> str:
         return "Invalid request payload"
 
 
 class UnsupportedAccessMode(SchemaError):
-    """
-    Unsupported values for user or access, or an unsupported combination of
+    """Unsupported values for user or access, or an unsupported combination of
     both.
     """
 
@@ -150,9 +178,8 @@ def __str__(self) -> str:
 
 
 class MissingParameters(SchemaError):
-    """
-    One or more required JSON keys are missing, or the values are unexpectedly
-    empty.
+    """One or more required JSON keys are missing, or the values are
+    unexpectedly empty.
     """
 
     def __init__(self, keys: List[str]):
@@ -164,9 +191,7 @@ def __str__(self):
 
 
 class BadQueryParam(SchemaError):
-    """
-    One or more unrecognized URL query parameters were specified.
-    """
+    """One or more unrecognized URL query parameters were specified."""
 
     def __init__(self, keys: List[str]):
         super().__init__()
@@ -177,8 +202,8 @@ def __str__(self):
 
 
 class RepeatedQueryParam(SchemaError):
-    """
-    A URL query parameter key was repeated, but Pbench supports only one value.
+    """A URL query parameter key was repeated, but Pbench supports only one
+    value.
     """
 
     def __init__(self, key: str):
@@ -190,13 +215,10 @@ def __str__(self):
 
 
 class ConversionError(SchemaError):
-    """
-    Used to report an invalid parameter type
-    """
+    """Used to report an invalid parameter type"""
 
     def __init__(self, value: Any, expected_type: str, **kwargs):
-        """
-        Construct a ConversionError exception
+        """Construct a ConversionError exception
 
         Args:
             value: The value we tried to convert
@@ -212,13 +234,10 @@ def __str__(self):
 
 
 class DatasetConversionError(SchemaError):
-    """
-    Used to report an invalid dataset name.
-    """
+    """Used to report an invalid dataset name."""
 
     def __init__(self, value: str, **kwargs):
-        """
-        Construct a DatasetConversionError exception. This is modeled after
+        """Construct a DatasetConversionError exception. This is modeled after
         DatasetNotFound, but is within the SchemaError exception hierarchy.
 
         Args:
@@ -233,9 +252,7 @@ def __str__(self):
 
 
 class KeywordError(SchemaError):
-    """
-    Used to report an unrecognized keyword value.
-    """
+    """Used to report an unrecognized keyword value."""
 
     def __init__(
         self,
@@ -245,8 +262,7 @@ def __init__(
         *,
         keywords: List[str] = [],
     ):
-        """
-        Construct a KeywordError exception
+        """Construct a KeywordError exception
 
         Args:
             parameter: The Parameter defining the keywords
@@ -266,13 +282,10 @@ def __str__(self):
 
 
 class ListElementError(SchemaError):
-    """
-    Used to report an unrecognized list element value.
-    """
+    """Used to report an unrecognized list element value."""
 
     def __init__(self, parameter: "Parameter", bad: List[str]):
-        """
-        Construct a ListElementError exception
+        """Construct a ListElementError exception
 
         Args:
             parameter: The Parameter defining the list
@@ -292,8 +305,7 @@ def __str__(self):
 
 
 def convert_date(value: str, _) -> datetime:
-    """
-    Convert a date/time string to a datetime.datetime object.
+    """Convert a date/time string to a datetime.datetime object.
 
     Args:
         value: String representation of date/time
@@ -312,9 +324,8 @@ def convert_date(value: str, _) -> datetime:
 
 
 def convert_username(value: Union[str, None], _) -> Union[str, None]:
-    """
-    Validate that the user object referenced by the username string exists, and
-    return the internal representation of that user.
+    """Validate that the user object referenced by the username string exists,
+    and return the internal representation of that user.
 
     We do not want an unauthenticated client to be able to distinguish between
     "invalid user" (ConversionError here) and "valid user I can't access" (some
@@ -1558,15 +1569,18 @@ def _dispatch(
                     method,
                     ApiParams(body=body_params, query=query_params, uri=uri_params),
                 )
+            except APIInternalError as e:
+                self.logger.exception("{} {}", api_name, e.details)
+                abort(e.http_status, message=str(e))
             except APIAbort as e:
                 self.logger.exception("{} {}", api_name, e)
                 abort(e.http_status, message=str(e))
-            except Exception as e:
-                self.logger.exception("{} API error: {}", api_name, e)
-                abort(
-                    HTTPStatus.INTERNAL_SERVER_ERROR,
-                    message=HTTPStatus.INTERNAL_SERVER_ERROR.phrase,
-                )
+            except Exception:
+                # Construct an APIInternalError to get the UUID and standard return
+                # message.
+                x = APIInternalError("Unexpected validation exception")
+                self.logger.exception("{} {}", api_name, x.details)
+                abort(x.http_status, message=str(x))
         else:
             params = ApiParams(None, None, None)
 
@@ -1583,12 +1597,15 @@ def _dispatch(
             except UnauthorizedAccess as e:
                 self.logger.warning("{}: {}", api_name, e)
                 abort(e.http_status, message=str(e))
-            except Exception as e:
-                self.logger.exception("{}: {}", api_name, e)
-                abort(
-                    HTTPStatus.INTERNAL_SERVER_ERROR,
-                    message=HTTPStatus.INTERNAL_SERVER_ERROR.phrase,
-                )
+            except APIInternalError as e:
+                self.logger.exception("{} {}", api_name, e.details)
+                abort(e.http_status, message=str(e))
+            except Exception:
+                # Construct an APIInternalError to get the UUID and standard return
+                # message.
+                x = APIInternalError("Unexpected authorize exception")
+                self.logger.exception("{} {}", api_name, x.details)
+                abort(x.http_status, message=str(x))
 
         audit = None
 
@@ -1629,7 +1646,11 @@ def _dispatch(
                     attributes=auditing["attributes"],
                 )
             return response
+        except APIInternalError as e:
+            self.logger.exception("{} {}", api_name, e.details)
+            abort(e.http_status, message=str(e))
         except APIAbort as e:
+            self.logger.error("{} {}", api_name, e)
             if auditing["finalize"]:
                 attr = auditing.get("attributes", {"message": str(e)})
                 try:
@@ -1640,7 +1661,7 @@ def _dispatch(
                         attributes=attr,
                     )
                 except Exception:
-                    self.logger.exception(
+                    self.logger.error(
                         "Unexpected exception on audit: {}", json.dumps(auditing)
                     )
             abort(e.http_status, message=str(e))
@@ -1657,10 +1678,8 @@ def _dispatch(
                     reason=AuditReason.INTERNAL,
                     attributes=attr,
                 )
-            abort(
-                HTTPStatus.INTERNAL_SERVER_ERROR,
-                message=HTTPStatus.INTERNAL_SERVER_ERROR.phrase,
-            )
+            x = APIInternalError("Unexpected exception")
+            abort(x.http_status, message=x.message)
 
     def _get(self, args: ApiParams, request: Request, context: ApiContext) -> Response:
         """
diff --git a/lib/pbench/server/api/resources/datasets_metadata.py b/lib/pbench/server/api/resources/datasets_metadata.py
index 527ebfde7e..a1e70b87f6 100644
--- a/lib/pbench/server/api/resources/datasets_metadata.py
+++ b/lib/pbench/server/api/resources/datasets_metadata.py
@@ -1,5 +1,6 @@
 from http import HTTPStatus
 from logging import Logger
+from typing import Any
 
 from flask.json import jsonify
 from flask.wrappers import Request, Response
@@ -186,7 +187,7 @@ def _put(
         # Now update the metadata, which may occur in multiple SQL operations
         # across namespaces. Make a best attempt to update all even if we
         # encounter an unexpected error.
-        fail = False
+        fail: dict[str, Any] = {}
         for k, v in metadata.items():
             native_key = Metadata.get_native_key(k)
             user_id = None
@@ -195,10 +196,10 @@ def _put(
             try:
                 Metadata.setvalue(key=k, value=v, dataset=dataset, user_id=user_id)
             except MetadataError as e:
-                self.logger.warning("Unable to update key {} = {!r}: {}", k, v, str(e))
-                fail = True
+                fail[k] = str(e)
 
-        if fail:
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
-        results = self._get_dataset_metadata(dataset, list(metadata.keys()))
+        results = {
+            "metadata": self._get_dataset_metadata(dataset, list(metadata.keys())),
+            "errors": fail,
+        }
         return jsonify(results)
diff --git a/lib/pbench/server/api/resources/query_apis/__init__.py b/lib/pbench/server/api/resources/query_apis/__init__.py
index 80e5ca9597..4e45f9b3b2 100644
--- a/lib/pbench/server/api/resources/query_apis/__init__.py
+++ b/lib/pbench/server/api/resources/query_apis/__init__.py
@@ -2,6 +2,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from http import HTTPStatus
+import json
 from logging import Logger
 import re
 from typing import Any, Callable, Iterator, List, Optional
@@ -21,6 +22,7 @@
     ApiAuthorizationType,
     ApiBase,
     ApiContext,
+    APIInternalError,
     ApiMethod,
     ApiParams,
     ApiSchema,
@@ -30,7 +32,7 @@
 )
 from pbench.server.auth.auth import Auth
 from pbench.server.database.models.audit import AuditReason, AuditStatus
-from pbench.server.database.models.datasets import Dataset, Metadata
+from pbench.server.database.models.datasets import Dataset, Metadata, States
 from pbench.server.database.models.template import Template
 from pbench.server.database.models.users import User
 
@@ -382,11 +384,9 @@ def _call(self, method: Callable, params: ApiParams, context: ApiContext):
             self.preprocess(params, context)
             self.logger.debug("PREPROCESS returns {}", context)
         except UnauthorizedAccess as e:
-            self.logger.warning("{}", e)
             raise APIAbort(e.http_status, str(e))
         except KeyError as e:
-            self.logger.exception("{} problem in preprocess, missing {}", klasname, e)
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError(f"problem in preprocess, missing {e}") from e
         try:
             # prepare payload for Elasticsearch query
             es_request = self.assemble(params, context)
@@ -398,8 +398,7 @@ def _call(self, method: Callable, params: ApiParams, context: ApiContext):
                 es_request.get("kwargs").get("json"),
             )
         except Exception as e:
-            self.logger.exception("{} assembly failed: {}", klasname, e)
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError("Elasticsearch assembly error") from e
 
         try:
             # perform the Elasticsearch query
@@ -412,7 +411,7 @@ def _call(self, method: Callable, params: ApiParams, context: ApiContext):
             es_response.raise_for_status()
             json_response = es_response.json()
         except requests.exceptions.HTTPError as e:
-            self.logger.exception(
+            self.logger.error(
                 "{} HTTP error {} from Elasticsearch request: {}",
                 klasname,
                 e,
@@ -423,53 +422,34 @@ def _call(self, method: Callable, params: ApiParams, context: ApiContext):
                 f"Elasticsearch query failure {e.response.reason} ({e.response.status_code})",
             )
         except requests.exceptions.ConnectionError:
-            self.logger.exception(
+            self.logger.error(
                 "{}: connection refused during the Elasticsearch request", klasname
             )
             raise APIAbort(
                 HTTPStatus.BAD_GATEWAY, "Network problem, could not reach Elasticsearch"
             )
         except requests.exceptions.Timeout:
-            self.logger.exception(
+            self.logger.error(
                 "{}: connection timed out during the Elasticsearch request", klasname
             )
             raise APIAbort(
                 HTTPStatus.GATEWAY_TIMEOUT,
                 "Connection timed out, could reach Elasticsearch",
             )
-        except requests.exceptions.InvalidURL:
-            self.logger.exception(
-                "{}: invalid url {} during the Elasticsearch request", klasname, url
-            )
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+        except requests.exceptions.InvalidURL as e:
+            raise APIInternalError(f"Invalid Elasticsearch URL {url}") from e
         except Exception as e:
-            self.logger.exception(
-                "{}: exception {} occurred during the Elasticsearch request",
-                klasname,
-                type(e).__name__,
-            )
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError("Unexpected backend error") from e
 
         try:
             # postprocess Elasticsearch response
             return self.postprocess(json_response, context)
         except PostprocessError as e:
-            msg = f"{klasname}: the query postprocessor was unable to complete: {e}"
-            self.logger.warning("{}", msg)
-            raise APIAbort(e.status, str(e.message))
-        except KeyError as e:
-            self.logger.error("{}: missing Elasticsearch key {}", klasname, e)
-            raise APIAbort(
-                HTTPStatus.INTERNAL_SERVER_ERROR, f"Missing Elasticsearch key {e}"
-            )
+            msg = f"{klasname}: {str(e)}"
+            self.logger.error("{}", msg)
+            raise APIAbort(e.status, msg)
         except Exception as e:
-            self.logger.exception(
-                "{}: unexpected problem postprocessing Elasticsearch response {}: {}",
-                klasname,
-                json_response,
-                e,
-            )
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError("Unexpected backend exception") from e
 
     def _post(
         self, params: ApiParams, request: Request, context: ApiContext
@@ -742,7 +722,6 @@ def _post(
         Returns:
             Response to return to client
         """
-        klasname = self.__class__.__name__
 
         # Our schema requires a valid dataset and uses it to authorize access;
         # therefore the unconditional dereference is assumed safe.
@@ -752,7 +731,8 @@ def _post(
 
         if self.require_stable and dataset.state.mutating:
             raise APIAbort(
-                HTTPStatus.CONFLICT, f"Dataset state {dataset.state.name} is mutating"
+                HTTPStatus.CONFLICT,
+                f"Dataset state {dataset.state.friendly!r} is mutating",
             )
 
         map = Metadata.getvalue(dataset=dataset, key=Metadata.INDEX_MAP)
@@ -775,16 +755,12 @@ def _post(
                 )
                 report = self._analyze_bulk(results)
             except Exception as e:
-                self.logger.exception(
-                    "{}: exception {} occurred during the Elasticsearch request",
-                    klasname,
-                    type(e).__name__,
-                )
-                raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+                raise APIInternalError("Unexpected backend error") from e
         elif self.require_map:
             raise APIAbort(
                 HTTPStatus.CONFLICT,
-                f"Dataset {self.action} requires Indexed state ({dataset.state.name})",
+                f"Dataset {self.action} requires {States.INDEXED.friendly!r} "
+                f"dataset but state is {dataset.state.friendly!r}",
             )
         else:
             report = BulkResults(errors=0, count=0, report={})
@@ -804,33 +780,21 @@ def _post(
             attributes["message"] = str(e)
             auditing["status"] = AuditStatus.WARNING
             auditing["reason"] = AuditReason.INTERNAL
-            self.logger.exception(
-                "{}: exception {} occurred during bulk operation completion",
-                klasname,
-                type(e).__name__,
-            )
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError("Unexpected completion error") from e
 
         # Return the summary document as the success response, or abort with an
-        # internal error if we weren't 100% successful. Some elasticsearch
-        # documents may have been affected, but the client will be able to try
-        # again.
+        # internal error if we tried to operate on Elasticsearch documents but
+        # experienced total failure. Either way, the operation can be retried
+        # if some documents failed to update.
         #
         # TODO: switching to `pyesbulk` will automatically handle retrying on
         # non-terminal errors, but this requires some cleanup work on the
         # pyesbulk side.
-        if report.errors > 0:
+        if report.count and report.errors == report.count:
             auditing["status"] = AuditStatus.WARNING
             auditing["reason"] = AuditReason.INTERNAL
             attributes["message"] = f"Unable to {self.action} some indexed documents"
-            raise APIAbort(
-                HTTPStatus.INTERNAL_SERVER_ERROR,
-                f"Failed to update {report.errors} out of {report.count} documents",
+            raise APIInternalError(
+                f"Failed to {self.action} any of {report.count} Elasticsearch documents: {json.dumps(report.report)}"
             )
-        self.logger.info(
-            "{}:dataset {}: {} successful document actions",
-            klasname,
-            dataset,
-            report.count,
-        )
         return jsonify(summary)
diff --git a/lib/pbench/server/api/resources/query_apis/datasets/__init__.py b/lib/pbench/server/api/resources/query_apis/datasets/__init__.py
index 714133c9e7..0747e26d98 100644
--- a/lib/pbench/server/api/resources/query_apis/datasets/__init__.py
+++ b/lib/pbench/server/api/resources/query_apis/datasets/__init__.py
@@ -1,12 +1,11 @@
-from http import HTTPStatus
 from logging import Logger
 from typing import AnyStr, List, NoReturn, Union
 
 from pbench.server import JSON, PbenchServerConfig
 from pbench.server.api.resources import (
-    APIAbort,
     ApiAuthorizationType,
     ApiContext,
+    APIInternalError,
     ApiParams,
     ApiSchema,
     ParamType,
@@ -129,12 +128,14 @@ def get_index(self, dataset: Dataset, root_index_name: AnyStr) -> AnyStr:
         try:
             index_map = Metadata.getvalue(dataset=dataset, key=Metadata.INDEX_MAP)
         except MetadataError as exc:
-            self.logger.error("{}", str(exc))
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            self.logger.error("{}", exc)
+            raise APIInternalError(f"Required metadata {Metadata.INDEX_MAP} missing")
 
         if index_map is None:
             self.logger.error("Index map metadata has no value")
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError(
+                f"Required metadata {Metadata.INDEX_MAP} has no value"
+            )
 
         index_keys = [key for key in index_map if root_index_name in key]
         indices = ",".join(index_keys)
diff --git a/lib/pbench/server/api/resources/query_apis/datasets/datasets_mappings.py b/lib/pbench/server/api/resources/query_apis/datasets/datasets_mappings.py
index 83c980d1b3..f4442b2703 100644
--- a/lib/pbench/server/api/resources/query_apis/datasets/datasets_mappings.py
+++ b/lib/pbench/server/api/resources/query_apis/datasets/datasets_mappings.py
@@ -1,4 +1,3 @@
-from http import HTTPStatus
 from logging import Logger
 
 from flask import jsonify
@@ -6,10 +5,10 @@
 
 from pbench.server import OperationCode, PbenchServerConfig
 from pbench.server.api.resources import (
-    APIAbort,
     ApiAuthorizationType,
     ApiBase,
     ApiContext,
+    APIInternalError,
     ApiMethod,
     ApiParams,
     ApiSchema,
@@ -116,5 +115,4 @@ def _get(
             # construct response object
             return jsonify(result)
         except TemplateNotFound as e:
-            self.logger.exception(str(e))
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+            raise APIInternalError("Unexpected template error") from e
diff --git a/lib/pbench/server/api/resources/query_apis/datasets/namespace_and_rows.py b/lib/pbench/server/api/resources/query_apis/datasets/namespace_and_rows.py
index 5cfc9a8365..724f26d9f2 100644
--- a/lib/pbench/server/api/resources/query_apis/datasets/namespace_and_rows.py
+++ b/lib/pbench/server/api/resources/query_apis/datasets/namespace_and_rows.py
@@ -3,8 +3,8 @@
 
 from pbench.server import JSON, OperationCode, PbenchServerConfig
 from pbench.server.api.resources import (
-    APIAbort,
     ApiAuthorizationType,
+    APIInternalError,
     ApiMethod,
     ApiParams,
     ApiSchema,
@@ -80,11 +80,8 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSON:
 
         try:
             mappings = self.get_mappings(document)
-        except TemplateNotFound:
-            self.logger.exception(
-                f"Document template {document_index!r} not found in the database."
-            )
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+        except TemplateNotFound as e:
+            raise APIInternalError("Unexpected template error") from e
 
         result = self.get_aggregatable_fields(mappings)
 
@@ -277,11 +274,8 @@ def assemble(self, params: ApiParams, context: ApiContext) -> JSON:
 
         try:
             mappings = self.get_mappings(document)
-        except TemplateNotFound:
-            self.logger.exception(
-                f"Document template {document_index!r} not found in the database."
-            )
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR)
+        except TemplateNotFound as e:
+            raise APIInternalError("Unexpected template error") from e
 
         # Prepare list of filters to apply for ES query
         es_filter = [{"match": {"run.id": dataset.resource_id}}]
diff --git a/lib/pbench/server/api/resources/server_configuration.py b/lib/pbench/server/api/resources/server_configuration.py
index 07da7c1676..6aaf7513d3 100644
--- a/lib/pbench/server/api/resources/server_configuration.py
+++ b/lib/pbench/server/api/resources/server_configuration.py
@@ -10,6 +10,7 @@
     ApiAuthorizationType,
     ApiBase,
     ApiContext,
+    APIInternalError,
     ApiMethod,
     ApiParams,
     ApiSchema,
@@ -102,7 +103,7 @@ def _get(
             else:
                 return jsonify(ServerConfig.get_all())
         except ServerConfigError as e:
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR, str(e)) from e
+            raise APIInternalError(f"Error reading server configuration {key}") from e
 
     def _put_key(self, params: ApiParams, context: ApiContext) -> Response:
         """
@@ -132,10 +133,7 @@ def _put_key(self, params: ApiParams, context: ApiContext) -> Response:
         except KeyError:
             # This "isn't possible" given the Flask mapping rules, but try
             # to report it gracefully instead of letting the KeyError fly.
-            raise APIAbort(
-                HTTPStatus.INTERNAL_SERVER_ERROR,
-                f"Found URI parameters {sorted(params.uri)} not including 'key'",
-            )
+            raise APIAbort(HTTPStatus.BAD_REQUEST, message="Missing parameter 'key'")
 
         # If we have a key in the URL, then we need a "value" for it, which
         # we can take either from a query parameter or from the JSON
@@ -165,7 +163,7 @@ def _put_key(self, params: ApiParams, context: ApiContext) -> Response:
         except ServerConfigBadValue as e:
             raise APIAbort(HTTPStatus.BAD_REQUEST, str(e)) from e
         except ServerConfigError as e:
-            raise APIAbort(HTTPStatus.INTERNAL_SERVER_ERROR, str(e)) from e
+            raise APIInternalError(f"Error setting server configuration {key}") from e
         return jsonify({key: value})
 
     def _put_body(self, params: ApiParams, context: ApiContext) -> Response:
@@ -195,19 +193,17 @@ def _put_body(self, params: ApiParams, context: ApiContext) -> Response:
 
         failures = []
         response = {}
-        fail_status = HTTPStatus.BAD_REQUEST
         for k, v in params.body.items():
             try:
                 c = ServerConfig.set(key=k, value=v)
                 response[c.key] = c.value
             except ServerConfigBadValue as e:
                 failures.append(str(e))
-            except ServerConfigError as e:
-                fail_status = HTTPStatus.INTERNAL_SERVER_ERROR
-                self.logger.warning(str(e))
-                failures.append(str(e))
+            except Exception as e:
+                self.logger.warning("{}", e)
+                raise APIInternalError(f"Error setting server configuration {k}")
         if failures:
-            raise APIAbort(fail_status, message=", ".join(failures))
+            raise APIAbort(HTTPStatus.BAD_REQUEST, message=", ".join(failures))
         return jsonify(response)
 
     def _put(
diff --git a/lib/pbench/test/unit/server/conftest.py b/lib/pbench/test/unit/server/conftest.py
index 75d3da1956..aeee5759ae 100644
--- a/lib/pbench/test/unit/server/conftest.py
+++ b/lib/pbench/test/unit/server/conftest.py
@@ -1,8 +1,11 @@
 import datetime
 import hashlib
+from http import HTTPStatus
+import logging
 import os
 from pathlib import Path
 from posix import stat_result
+import re
 import shutil
 from stat import ST_MTIME
 import tarfile
@@ -13,6 +16,7 @@
 from freezegun import freeze_time
 import jwt
 import pytest
+from requests import Response
 
 from pbench.common import MetadataLog
 from pbench.common.logger import get_pbench_logger
@@ -158,6 +162,27 @@ def make_logger(server_config):
     return get_pbench_logger("TEST", server_config)
 
 
+@pytest.fixture()
+def capinternal(caplog):
+    def compare(message: str, response: Optional[Response]):
+        uuid = r"[a-zA-Z\d]{8}-([a-zA-Z\d]{4}-){3}[a-zA-Z\d]{12}"
+        name = r"\w+\s+"
+        external = re.compile(f"Internal Pbench Server Error: log reference {uuid}")
+        internal = re.compile(f"{name}Internal error {uuid}: {message}")
+        if response:
+            assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
+            assert external.match(response.json["message"])
+        for r in caplog.get_records("call"):
+            if r.levelno == logging.ERROR:
+                if internal.match(str(r.msg)):
+                    return
+        assert (
+            False
+        ), f"Expected pattern {internal.pattern!r} not logged at level 'ERROR': {[str(r.msg) for r in caplog.get_records('call')]}"
+
+    return compare
+
+
 @pytest.fixture()
 def db_session(request, server_config, make_logger):
     """
@@ -171,10 +196,12 @@ def db_session(request, server_config, make_logger):
     client fixture is also selected, but we'll still remove the DB afterwards.
 
     Args:
+        request: Access to pytest's request context
         server_config: pbench-server.cfg fixture
+        make_logger: produce a Pbench Server logger
     """
     if "client" not in request.fixturenames:
-        Database.init_db(server_config, None)
+        Database.init_db(server_config, make_logger)
     yield
     Database.db_session.remove()
 
diff --git a/lib/pbench/test/unit/server/query_apis/test_datasets_delete.py b/lib/pbench/test/unit/server/query_apis/test_datasets_delete.py
index d5d9d30786..9fc4321f5d 100644
--- a/lib/pbench/test/unit/server/query_apis/test_datasets_delete.py
+++ b/lib/pbench/test/unit/server/query_apis/test_datasets_delete.py
@@ -152,7 +152,13 @@ def test_query(
             Dataset.query(name=owner)
 
     def test_partial(
-        self, client, get_document_map, monkeypatch, server_config, pbench_token
+        self,
+        client,
+        capinternal,
+        get_document_map,
+        monkeypatch,
+        server_config,
+        pbench_token,
     ):
         """
         Check the delete API when some document updates fail. We expect an
@@ -165,10 +171,8 @@ def test_partial(
             f"{server_config.rest_uri}/datasets/delete/{ds.resource_id}",
             headers={"authorization": f"Bearer {pbench_token}"},
         )
-
-        # Verify the report and status
-        assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
-        assert response.json["message"] == "Failed to update 3 out of 31 documents"
+        assert response.status_code == HTTPStatus.OK
+        assert response.json == {"ok": 28, "failure": 3}
 
         # Verify that the Dataset still exists
         Dataset.query(name="drb")
@@ -211,6 +215,7 @@ def test_no_index(
     def test_exception(
         self,
         attach_dataset,
+        capinternal,
         client,
         monkeypatch,
         get_document_map,
@@ -230,7 +235,7 @@ def fake_bulk(
             raise_on_error: bool = True,
             raise_on_exception: bool = True,
         ):
-            raise elasticsearch.helpers.BulkIndexError
+            raise elasticsearch.helpers.BulkIndexError("test")
 
         monkeypatch.setattr("elasticsearch.helpers.streaming_bulk", fake_bulk)
 
@@ -240,5 +245,4 @@ def fake_bulk(
         )
 
         # Verify the failure
-        assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
-        assert response.json["message"] == HTTPStatus.INTERNAL_SERVER_ERROR.phrase
+        capinternal("Unexpected backend error", response)
diff --git a/lib/pbench/test/unit/server/query_apis/test_datasets_mappings.py b/lib/pbench/test/unit/server/query_apis/test_datasets_mappings.py
index 433c47e1d4..ae038c1cc6 100644
--- a/lib/pbench/test/unit/server/query_apis/test_datasets_mappings.py
+++ b/lib/pbench/test/unit/server/query_apis/test_datasets_mappings.py
@@ -104,11 +104,10 @@ def test_with_no_index_document(self, client, server_config):
                 == "Unrecognized keyword ['bad_data_object_view'] given for parameter dataset_view; allowed keywords are ['contents', 'iterations', 'summary', 'timeseries']"
             )
 
-    def test_with_db_error(self, client, server_config):
+    def test_with_db_error(self, capinternal, client, server_config):
         """
         Check the index mappings API if there is an error connecting to sql database.
         """
         with client:
             response = client.get(f"{server_config.rest_uri}/datasets/mappings/summary")
-            assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
-            assert response.json["message"] == HTTPStatus.INTERNAL_SERVER_ERROR.phrase
+            capinternal("Unexpected template error", response)
diff --git a/lib/pbench/test/unit/server/query_apis/test_datasets_publish.py b/lib/pbench/test/unit/server/query_apis/test_datasets_publish.py
index 2438a4e1f2..83430bf6c2 100644
--- a/lib/pbench/test/unit/server/query_apis/test_datasets_publish.py
+++ b/lib/pbench/test/unit/server/query_apis/test_datasets_publish.py
@@ -154,10 +154,8 @@ def test_partial(
             headers={"authorization": f"Bearer {pbench_token}"},
             query_string=self.PAYLOAD,
         )
-
-        # Verify the report and status
-        assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
-        assert response.json["message"] == "Failed to update 3 out of 31 documents"
+        assert response.status_code == HTTPStatus.OK
+        assert response.json == {"ok": 28, "failure": 3}
 
         # Verify that the Dataset access didn't change
         dataset = Dataset.query(name="drb")
@@ -199,7 +197,7 @@ def test_no_index(
         # Verify the report and status
         assert response.status_code == HTTPStatus.CONFLICT
         assert response.json == {
-            "message": "Dataset update requires Indexed state (INDEXED)"
+            "message": "Dataset update requires 'Indexed' dataset but state is 'Indexed'"
         }
 
     def test_exception(
@@ -210,6 +208,7 @@ def test_exception(
         get_document_map,
         pbench_token,
         server_config,
+        capinternal,
     ):
         """
         Check the publish API response if the bulk helper throws an exception.
@@ -224,7 +223,7 @@ def fake_bulk(
             raise_on_error: bool = True,
             raise_on_exception: bool = True,
         ):
-            raise elasticsearch.helpers.BulkIndexError
+            raise elasticsearch.helpers.BulkIndexError("test")
 
         monkeypatch.setattr("elasticsearch.helpers.streaming_bulk", fake_bulk)
 
@@ -235,5 +234,4 @@ def fake_bulk(
         )
 
         # Verify the failure
-        assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
-        assert response.json["message"] == HTTPStatus.INTERNAL_SERVER_ERROR.phrase
+        capinternal("Unexpected backend error", response)
diff --git a/lib/pbench/test/unit/server/test_datasets_metadata.py b/lib/pbench/test/unit/server/test_datasets_metadata.py
index 0ca4463c87..6b96ee24c8 100644
--- a/lib/pbench/test/unit/server/test_datasets_metadata.py
+++ b/lib/pbench/test/unit/server/test_datasets_metadata.py
@@ -8,7 +8,7 @@
 from pbench.server.database.models.datasets import Dataset, DatasetNotFound
 
 
-class TestDatasetsMetadata:
+class TestDatasetsMetadataGet:
     @pytest.fixture()
     def query_get_as(
         self, client, server_config, more_datasets, provide_metadata, get_token
@@ -54,51 +54,6 @@ def query_api(
 
         return query_api
 
-    @pytest.fixture()
-    def query_put_as(
-        self, client, server_config, more_datasets, provide_metadata, get_token
-    ):
-        """
-        Helper fixture to perform the API query and validate an expected
-        return status.
-
-        Args:
-            client: Flask test API client fixture
-            server_config: Pbench config fixture
-            more_datasets: Dataset construction fixture
-            provide_metadata: Dataset metadata fixture
-            get_token: Pbench token fixture
-        """
-
-        def query_api(
-            ds_name: str, payload: JSON, username: str, expected_status: HTTPStatus
-        ) -> requests.Response:
-            headers = None
-            try:
-                dataset = Dataset.query(name=ds_name).resource_id
-            except DatasetNotFound:
-                dataset = ds_name  # Allow passing deliberately bad value
-            if username:
-                token = get_token(username)
-                headers = {"authorization": f"bearer {token}"}
-            response = client.put(
-                f"{server_config.rest_uri}/datasets/metadata/{dataset}",
-                headers=headers,
-                json=payload,
-            )
-            assert response.status_code == expected_status
-
-            # We need to log out to avoid "duplicate auth token" errors on the
-            # test case which does a PUT followed by two GETs.
-            if username:
-                client.post(
-                    f"{server_config.rest_uri}/logout",
-                    headers={"authorization": f"bearer {token}"},
-                )
-            return response
-
-        return query_api
-
     def test_get_no_dataset(self, query_get_as):
         response = query_get_as(
             "foobar",
@@ -254,6 +209,53 @@ def test_get_bad_query_2(self, query_get_as):
         )
         assert response.json == {"message": "Unknown URL query keys: controller,plugh"}
 
+
+class TestDatasetsMetadataPut(TestDatasetsMetadataGet):
+    @pytest.fixture()
+    def query_put_as(
+        self, client, server_config, more_datasets, provide_metadata, get_token
+    ):
+        """
+        Helper fixture to perform the API query and validate an expected
+        return status.
+
+        Args:
+            client: Flask test API client fixture
+            server_config: Pbench config fixture
+            more_datasets: Dataset construction fixture
+            provide_metadata: Dataset metadata fixture
+            get_token: Pbench token fixture
+        """
+
+        def query_api(
+            ds_name: str, payload: JSON, username: str, expected_status: HTTPStatus
+        ) -> requests.Response:
+            headers = None
+            try:
+                dataset = Dataset.query(name=ds_name).resource_id
+            except DatasetNotFound:
+                dataset = ds_name  # Allow passing deliberately bad value
+            if username:
+                token = get_token(username)
+                headers = {"authorization": f"bearer {token}"}
+            response = client.put(
+                f"{server_config.rest_uri}/datasets/metadata/{dataset}",
+                headers=headers,
+                json=payload,
+            )
+            assert response.status_code == expected_status
+
+            # We need to log out to avoid "duplicate auth token" errors on the
+            # test case which does a PUT followed by two GETs.
+            if username:
+                client.post(
+                    f"{server_config.rest_uri}/logout",
+                    headers={"authorization": f"bearer {token}"},
+                )
+            return response
+
+        return query_api
+
     @pytest.mark.parametrize("uri", ("/datasets/metadata/", "/datasets/metadata"))
     def test_put_missing_uri_param(self, client, server_config, pbench_token, uri):
         """
@@ -402,6 +404,43 @@ def test_put_invalid_deletion(self, query_get_as, query_put_as):
             "user.one": None,
         }
 
+    def test_put_set_errors(self, capinternal, monkeypatch, query_get_as, query_put_as):
+        """Test a partial success. We set a scalar value on a key and then try
+        to set a nested value: i.e., with "global.dashboard.nested = False", we
+        attempt to set "global.dashboard.nested.dummy". We expect this to fail,
+        but we expect other values to succeed. This partial success is reported
+        as an "internal error" (for lack of any better representation in HTTP)
+        with a message indicating the specific failures.
+        """
+        query_put_as(
+            "drb",
+            {"metadata": {"global.dashboard.nested": False}},
+            "drb",
+            HTTPStatus.OK,
+        )
+        response = query_put_as(
+            "drb",
+            {
+                "metadata": {
+                    "global.dashboard.seen": False,
+                    "global.dashboard.nested.dummy": True,
+                    "dataset.name": "foo",
+                }
+            },
+            "drb",
+            HTTPStatus.INTERNAL_SERVER_ERROR,
+        )
+        response = query_get_as(
+            "foo", {"metadata": "global,dataset.name"}, "drb", HTTPStatus.OK
+        )
+        assert response.json == {
+            "global": {
+                "contact": "me@example.com",
+                "dashboard": {"seen": False, "nested": False},
+            },
+            "dataset.name": "foo",
+        }
+
     def test_put(self, query_get_as, query_put_as):
         response = query_put_as(
             "drb",
@@ -409,7 +448,10 @@ def test_put(self, query_get_as, query_put_as):
             "drb",
             HTTPStatus.OK,
         )
-        assert response.json == {"global.saved": True, "global.seen": False}
+        assert response.json == {
+            "metadata": {"global.saved": True, "global.seen": False},
+            "errors": {},
+        }
         response = query_get_as(
             "drb", {"metadata": "global,dataset.access"}, "drb", HTTPStatus.OK
         )
@@ -466,7 +508,10 @@ def test_put_user(self, query_get_as, query_put_as):
             "drb",
             HTTPStatus.OK,
         )
-        assert response.json == {"user.favorite": True, "user.tag": "AWS"}
+        assert response.json == {
+            "metadata": {"user.favorite": True, "user.tag": "AWS"},
+            "errors": {},
+        }
         audit = Audit.query()
         assert len(audit) == 2
         assert audit[0].id == 1
@@ -502,7 +547,10 @@ def test_put_user(self, query_get_as, query_put_as):
             "test",
             HTTPStatus.OK,
         )
-        assert response.json == {"user.favorite": False, "user.tag": "RHEL"}
+        assert response.json == {
+            "metadata": {"user.favorite": False, "user.tag": "RHEL"},
+            "errors": {},
+        }
         response = query_put_as(
             "fio_1",
             {"metadata": {"user.favorite": False, "user.tag": "BAD"}},
diff --git a/lib/pbench/test/unit/server/test_schema.py b/lib/pbench/test/unit/server/test_schema.py
index 16e1e27324..e947966094 100644
--- a/lib/pbench/test/unit/server/test_schema.py
+++ b/lib/pbench/test/unit/server/test_schema.py
@@ -1,12 +1,15 @@
 from http import HTTPStatus
+import json
 from typing import Callable, Union
 
 from dateutil import parser as date_parser
+from flask.wrappers import Response
 import pytest
 
 from pbench.server import OperationCode
 from pbench.server.api.resources import (
     APIAbort,
+    APIInternalError,
     ConversionError,
     InvalidRequestPayload,
     KeywordError,
@@ -75,6 +78,16 @@ def test_exceptions(self, create_user):
         assert str(p) == "Postprocessing error returning 400: 'really bad' [None]"
         assert p.status == HTTPStatus.BAD_REQUEST
 
+    def test_internal_error(self, capinternal, make_logger):
+        x = APIInternalError("my error")
+        make_logger.error(f"TEST {x.details}")
+        r = Response(
+            response=json.dumps({"message": x.message}),
+            mimetype="application/json",
+            status=HTTPStatus.INTERNAL_SERVER_ERROR,
+        )
+        capinternal("my error", r)
+
 
 class TestParamType:
     """
diff --git a/lib/pbench/test/unit/server/test_server_configuration.py b/lib/pbench/test/unit/server/test_server_configuration.py
index e05bb8e095..13217d4892 100644
--- a/lib/pbench/test/unit/server/test_server_configuration.py
+++ b/lib/pbench/test/unit/server/test_server_configuration.py
@@ -92,7 +92,7 @@ def test_put_bad_uri_key(self, server_config):
         routing.
         """
         put = ServerConfiguration(server_config, logging.getLogger("test"))
-        with pytest.raises(APIAbort, match=r"Found URI parameters \['foo', 'plugh'\]"):
+        with pytest.raises(APIAbort, match="Missing parameter 'key'"):
             put._put_key(ApiParams(uri={"plugh": "xyzzy", "foo": "bar"}), context=None)
 
     def test_put_missing_value(self, query_put):