Handle gateway collect stats and other errors

src/dstack/_internal/core/backends/aws/compute.py

@@ -1291,6 +1291,19 @@ def _get_vpc_id_subnets_ids_by_vpc_name_or_error(
     "L-DB2E81BA": "G/OnDemand",
 }
 
+# `GetServiceQuota` errors that say nothing about dstack: the quota stays unknown and
+# the offer availability stays `UNKNOWN`. Any other error code is reported, since it
+# may mean the request is malformed.
+_EXPECTED_QUOTA_ERROR_CODES = {
+    "408",  # request timed out
+    "TooManyRequestsException",  # rate limits
+    "AccessDeniedException",  # no servicequotas:GetServiceQuota permission
+    "AuthFailure",  # invalid, expired, or deactivated credentials
+    "InvalidClientTokenId",
+    "RequestExpired",
+    "UnrecognizedClientException",
+}
+
 
 def _get_regions_to_quotas(
     session: boto3.Session, regions: List[str]
@@ -1302,14 +1315,16 @@ def get_region_quotas(region_name: str, client: botocore.client.BaseClient) -> D
                 resp = client.get_service_quota(ServiceCode="ec2", QuotaCode=quota_code)
                 region_quotas[quota_class] = resp["Quota"]["Value"]
             except botocore.exceptions.ClientError as e:
-                if "TooManyRequestsException" in str(e):
+                error_code = e.response.get("Error", {}).get("Code", "")
+                if error_code in _EXPECTED_QUOTA_ERROR_CODES:
                     logger.warning(
-                        "Failed to get quota %s in %s due to rate limits",
+                        "Failed to get quota %s in %s: %s",
                         quota_code,
                         region_name,
+                        e,
                     )
                 else:
-                    logger.exception(e)
+                    logger.exception("Failed to get quota %s in %s", quota_code, region_name)
         return region_quotas
 
     regions_to_quotas = {}

src/dstack/_internal/server/background/pipeline_tasks/instances/check.py

@@ -432,7 +432,7 @@ def _check_instance_inner(
     try:
         remove_dangling_tasks_from_instance(shim_client, instance)
     except Exception as exc:
-        logger.exception("%s: error removing dangling tasks: %s", fmt(instance), exc)
+        logger.warning("%s: error removing dangling tasks: %s", fmt(instance), exc)
 
     # There should be no shim API calls after this function call since it can request shim restart.
     _maybe_install_components(instance, shim_client)

src/dstack/_internal/server/background/scheduled_tasks/gateways.py

@@ -1,5 +1,6 @@
 import asyncio
 
+import httpx
 from sqlalchemy import select
 
 from dstack._internal.core.errors import SSHError
@@ -60,4 +61,9 @@ async def _process_connection(conn: GatewayConnection):
         logger.warning("Connection to gateway %s failed: %s", conn.ip_address, e)
         return
 
-    await conn.try_collect_stats()
+    try:
+        await conn.try_collect_stats()
+    except httpx.HTTPError as e:
+        logger.warning("Failed to collect stats from gateway %s: %r", conn.ip_address, e)
+    except Exception:
+        logger.exception("Failed to collect stats from gateway %s", conn.ip_address)

src/dstack/_internal/server/services/locking.py

@@ -9,6 +9,10 @@
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
 
+from dstack._internal.utils.logging import get_logger
+
+logger = get_logger(__name__)
+
 KeyT = TypeVar("KeyT")
 
 
@@ -134,12 +138,12 @@ async def advisory_lock_ctx(
 
     To prevent unreleased locks:
 
-    1. When possible, prefer using `pg_advisory_xact_lock` instead of this context manager.
+    * When possible, prefer using `pg_advisory_xact_lock` instead of this context manager.
        `pg_advisory_xact_lock` is automatically released at the end of transaction.
 
-    1. Prefer using `AsyncConnection` as `bind`.
+    * Prefer using `AsyncConnection` as `bind`.
 
-    1. If using `AsyncSession` as `bind`, **do not** commit before exiting from the context manager.
+    * If using `AsyncSession` as `bind`, **do not** commit before exiting from the context manager.
        Committing will prompt `AsyncSession` to start a new transaction for releasing the lock,
        which may be assigned to a different database connection, which will fail to release.
     """
@@ -150,7 +154,7 @@ async def advisory_lock_ctx(
         yield
     finally:
         if dialect_name == "postgresql":
-            await bind.execute(select(func.pg_advisory_unlock(string_to_lock_id(resource))))
+            await _release_advisory_lock(bind, resource)
 
 
 @asynccontextmanager
@@ -165,7 +169,7 @@ async def try_advisory_lock_ctx(
         yield locked
     finally:
         if dialect_name == "postgresql" and locked:
-            await bind.execute(select(func.pg_advisory_unlock(string_to_lock_id(resource))))
+            await _release_advisory_lock(bind, resource)
 
 
 _in_memory_locker = InMemoryResourceLocker()
@@ -203,3 +207,18 @@ async def _wait_to_lock_many(
         if not left_to_lock:
             return
         await asyncio.sleep(delay)
+
+
+async def _release_advisory_lock(
+    bind: Union[AsyncConnection, AsyncSession], resource: str
+) -> None:
+    """
+    Release an advisory lock, tolerating failures.
+    Releasing typically fails with `PendingRollbackError` because the connection has been
+    invalidated. In this case Postgres has already dropped the lock along with the
+    session and there is nothing left to release.
+    """
+    try:
+        await bind.execute(select(func.pg_advisory_unlock(string_to_lock_id(resource))))
+    except Exception as e:
+        logger.warning("Failed to release advisory lock on %s: %r", resource, e)