refactor: specialize exception handling across all services

1 files changed, 30 insertions, 2 deletions

diff --git a/services/order-executor/src/order_executor/main.py b/services/order-executor/src/order_executor/main.py
index 63d93bc..d9e2373 100644
--- a/services/order-executor/src/order_executor/main.py
+++ b/services/order-executor/src/order_executor/main.py
@@ -3,6 +3,8 @@
 import asyncio
 from decimal import Decimal
 
+import aiohttp
+
 from shared.alpaca import AlpacaClient
 from shared.broker import RedisBroker
 from shared.db import Database
@@ -98,8 +100,18 @@ async def run() -> None:
                 if event.type == EventType.SIGNAL:
                     await executor.execute(event.data)
                 await broker.ack(stream, GROUP, msg_id)
+            except (
+                aiohttp.ClientError,
+                ConnectionError,
+                TimeoutError,
+                asyncio.TimeoutError,
+            ) as exc:
+                log.warning("pending_network_error", error=str(exc), msg_id=msg_id)
+            except (ValueError, KeyError, TypeError) as exc:
+                log.warning("pending_parse_error", error=str(exc), msg_id=msg_id)
+                await broker.ack(stream, GROUP, msg_id)
             except Exception as exc:
-                log.error("pending_failed", error=str(exc), msg_id=msg_id)
+                log.error("pending_failed", error=str(exc), msg_id=msg_id, exc_info=True)
 
         while not shutdown.is_shutting_down:
             messages = await broker.read_group(stream, GROUP, CONSUMER, count=10, block=5000)
@@ -114,8 +126,24 @@ async def run() -> None:
                             service="order-executor", event_type="signal"
                         ).inc()
                     await broker.ack(stream, GROUP, msg_id)
+                except (
+                    aiohttp.ClientError,
+                    ConnectionError,
+                    TimeoutError,
+                    asyncio.TimeoutError,
+                ) as exc:
+                    log.warning("process_network_error", error=str(exc))
+                    metrics.errors_total.labels(
+                        service="order-executor", error_type="network"
+                    ).inc()
+                except (ValueError, KeyError, TypeError) as exc:
+                    log.warning("process_parse_error", error=str(exc))
+                    await broker.ack(stream, GROUP, msg_id)
+                    metrics.errors_total.labels(
+                        service="order-executor", error_type="validation"
+                    ).inc()
                 except Exception as exc:
-                    log.error("process_failed", error=str(exc))
+                    log.error("process_failed", error=str(exc), exc_info=True)
                     metrics.errors_total.labels(
                         service="order-executor", error_type="processing"
                     ).inc()