refactor: specialize exception handling across all services

1 files changed, 22 insertions, 4 deletions

diff --git a/services/portfolio-manager/src/portfolio_manager/main.py b/services/portfolio-manager/src/portfolio_manager/main.py
index 6cf248f..6ca7b1b 100644
--- a/services/portfolio-manager/src/portfolio_manager/main.py
+++ b/services/portfolio-manager/src/portfolio_manager/main.py
@@ -2,6 +2,8 @@
 
 import asyncio
 
+import sqlalchemy.exc
+
 from shared.broker import RedisBroker
 from shared.db import Database
 from shared.events import Event, OrderEvent
@@ -52,8 +54,12 @@ async def snapshot_loop(
     while True:
         try:
             await save_snapshot(db, tracker, notifier, log)
+        except (sqlalchemy.exc.OperationalError, ConnectionError, TimeoutError) as exc:
+            log.warning("snapshot_db_error", error=str(exc))
+        except (ValueError, KeyError, TypeError) as exc:
+            log.warning("snapshot_data_error", error=str(exc))
         except Exception as exc:
-            log.error("snapshot_failed", error=str(exc))
+            log.error("snapshot_failed", error=str(exc), exc_info=True)
         await asyncio.sleep(interval_hours * 3600)
 
 
@@ -112,8 +118,12 @@ async def run() -> None:
                     service="portfolio-manager", event_type="order"
                 ).inc()
             await broker.ack(ORDERS_STREAM, GROUP, msg_id)
+        except (ValueError, KeyError, TypeError) as exc:
+            log.warning("pending_parse_error", error=str(exc), msg_id=msg_id)
+            await broker.ack(ORDERS_STREAM, GROUP, msg_id)
+            metrics.errors_total.labels(service="portfolio-manager", error_type="validation").inc()
         except Exception as exc:
-            log.error("pending_process_failed", error=str(exc), msg_id=msg_id)
+            log.error("pending_process_failed", error=str(exc), msg_id=msg_id, exc_info=True)
             metrics.errors_total.labels(service="portfolio-manager", error_type="processing").inc()
 
     try:
@@ -138,13 +148,21 @@ async def run() -> None:
                             service="portfolio-manager", event_type="order"
                         ).inc()
                     await broker.ack(ORDERS_STREAM, GROUP, msg_id)
+                except (ValueError, KeyError, TypeError) as exc:
+                    log.warning("message_parse_error", error=str(exc), msg_id=msg_id)
+                    await broker.ack(ORDERS_STREAM, GROUP, msg_id)
+                    metrics.errors_total.labels(
+                        service="portfolio-manager", error_type="validation"
+                    ).inc()
                 except Exception as exc:
-                    log.exception("message_processing_failed", error=str(exc), msg_id=msg_id)
+                    log.error(
+                        "message_processing_failed", error=str(exc), msg_id=msg_id, exc_info=True
+                    )
                     metrics.errors_total.labels(
                         service="portfolio-manager", error_type="processing"
                     ).inc()
     except Exception as exc:
-        log.error("fatal_error", error=str(exc))
+        log.error("fatal_error", error=str(exc), exc_info=True)
         await notifier.send_error(str(exc), "portfolio-manager")
         raise
     finally: