fix(service): release per-stream subscription connection on unsubscribe

Tie each per-stream subscription's dedicated Postgres connection to its
SubscriptionId via an onRemove cleanup Task, released best-effort on
removeSubscription so unsubscribe stays total. Fixes the connection leak
that exhausted the B1ms connection ceiling under stream-subscription churn.

Implements ADR-0063.

Closes #683

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: NickSeagullBot

core/service/Service/EventStore/Postgres/Internal.hs

@@ -809,9 +809,15 @@ subscribeToStreamEventsImpl ops cfg store entityName streamId callback =
       Sessions.selectMaxGlobalPosition
         |> Sessions.run conn
         |> Task.mapError (toText .> StorageFailure)
+
+    -- Release the dedicated per-stream connection when this subscription is
+    -- removed. Owned by the SubscriptionId, run best-effort on unsubscribe
+    -- (ADR-0063 §3).
+    let releaseConnection = Hasql.release connection |> Task.fromIO
+
     subscriptionId <-
       store
-        |> SubscriptionStore.addStreamSubscriptionFromPosition entityName streamId currentMaxPosition callback
+        |> SubscriptionStore.addStreamSubscriptionWithCleanup entityName streamId currentMaxPosition releaseConnection callback
         |> Task.mapError (\err -> SubscriptionError (SubscriptionId "stream") (err |> toText))
     Log.debug [fmt|Subscription created: #{toText subscriptionId}|] |> Task.ignoreError
     Task.yield subscriptionId

core/service/Service/EventStore/Postgres/SubscriptionStore.hs

@@ -6,6 +6,7 @@ module Service.EventStore.Postgres.SubscriptionStore (
   addGlobalSubscriptionFromPosition,
   addStreamSubscription,
   addStreamSubscriptionFromPosition,
+  addStreamSubscriptionWithCleanup,
   addEntitySubscription,
   addEntitySubscriptionFromPosition,
   getStreamSubscriptions,
@@ -21,6 +22,7 @@ import Core
 import Json qualified
 import Log qualified
 import Map qualified
+import Prelude qualified
 import Service.Event (EntityName, Event (..), StreamPosition)
 import Service.Event.EventMetadata (EventMetadata (..))
 import Service.EventStore.Core (SubscriptionId (..))
@@ -41,9 +43,23 @@ type SubscriptionCallback =
 data SubscriptionInfo = SubscriptionInfo
   { callback :: SubscriptionCallback,
     startingGlobalPosition :: Maybe StreamPosition,
-    entityNameFilter :: Maybe EntityName
+    entityNameFilter :: Maybe EntityName,
+    -- | Run once when this subscription is removed. Releases any dedicated
+    -- resource the subscription owns (the per-stream LISTEN connection). A
+    -- no-op ('Task.yield unit') for subscriptions that own no dedicated resource.
+    onRemove :: Task Text Unit
   }
-  deriving (Show)
+
+
+-- Hand-written Show: the callback and onRemove are Task/closure fields with no
+-- Show instance, so they are elided. (ADR-0063 §1.)
+instance Show SubscriptionInfo where
+  show info =
+    "SubscriptionInfo {startingGlobalPosition = "
+      ++ Prelude.show info.startingGlobalPosition
+      ++ ", entityNameFilter = "
+      ++ Prelude.show info.entityNameFilter
+      ++ ", callback = <function>, onRemove = <task>}"
 
 
 type Subscriptions =
@@ -68,7 +84,7 @@ new = do
 addGlobalSubscription :: SubscriptionCallback -> SubscriptionStore -> Task Error SubscriptionId
 addGlobalSubscription callback store = do
   subId <- Uuid.generate |> Task.map (\result -> result |> toText |> SubscriptionId)
-  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = Nothing, entityNameFilter = Nothing}
+  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = Nothing, entityNameFilter = Nothing, onRemove = Task.yield unit}
   store.globalSubscriptions
     |> ConcurrentVar.modify (Map.set subId subscriptionInfo)
   Task.yield subId
@@ -78,7 +94,7 @@ addGlobalSubscriptionFromPosition ::
   Maybe StreamPosition -> SubscriptionCallback -> SubscriptionStore -> Task Error SubscriptionId
 addGlobalSubscriptionFromPosition startingPosition callback store = do
   subId <- Uuid.generate |> Task.map (\result -> result |> toText |> SubscriptionId)
-  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = startingPosition, entityNameFilter = Nothing}
+  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = startingPosition, entityNameFilter = Nothing, onRemove = Task.yield unit}
   store.globalSubscriptions
     |> ConcurrentVar.modify (Map.set subId subscriptionInfo)
   Task.yield subId
@@ -88,7 +104,7 @@ addStreamSubscription ::
   EntityName -> StreamId -> SubscriptionCallback -> SubscriptionStore -> Task Error SubscriptionId
 addStreamSubscription entityName streamId callback store = do
   subId <- Uuid.generate |> Task.map (toText .> SubscriptionId)
-  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = Nothing, entityNameFilter = Just entityName}
+  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = Nothing, entityNameFilter = Just entityName, onRemove = Task.yield unit}
   store.streamSubscriptions |> ConcurrentVar.modify \subscriptionsMap -> do
     let currentSubscriptions = subscriptionsMap |> Map.getOrElse streamId Map.empty
     subscriptionsMap |> Map.set streamId (currentSubscriptions |> Map.set subId subscriptionInfo)
@@ -104,7 +120,30 @@ addStreamSubscriptionFromPosition ::
   Task Error SubscriptionId
 addStreamSubscriptionFromPosition entityName streamId startingPosition callback store = do
   subId <- Uuid.generate |> Task.map (toText .> SubscriptionId)
-  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = startingPosition, entityNameFilter = Just entityName}
+  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = startingPosition, entityNameFilter = Just entityName, onRemove = Task.yield unit}
+  store.streamSubscriptions |> ConcurrentVar.modify \subscriptionsMap -> do
+    let currentSubscriptions = subscriptionsMap |> Map.getOrElse streamId Map.empty
+    subscriptionsMap |> Map.set streamId (currentSubscriptions |> Map.set subId subscriptionInfo)
+  Task.yield subId
+
+
+addStreamSubscriptionWithCleanup ::
+  EntityName ->
+  StreamId ->
+  Maybe StreamPosition ->
+  Task Text Unit ->
+  SubscriptionCallback ->
+  SubscriptionStore ->
+  Task Error SubscriptionId
+addStreamSubscriptionWithCleanup entityName streamId startingPosition onRemove callback store = do
+  subId <- Uuid.generate |> Task.map (toText .> SubscriptionId)
+  let subscriptionInfo =
+        SubscriptionInfo
+          { callback,
+            startingGlobalPosition = startingPosition,
+            entityNameFilter = Just entityName,
+            onRemove
+          }
   store.streamSubscriptions |> ConcurrentVar.modify \subscriptionsMap -> do
     let currentSubscriptions = subscriptionsMap |> Map.getOrElse streamId Map.empty
     subscriptionsMap |> Map.set streamId (currentSubscriptions |> Map.set subId subscriptionInfo)
@@ -115,7 +154,7 @@ addEntitySubscription ::
   EntityName -> SubscriptionCallback -> SubscriptionStore -> Task Error SubscriptionId
 addEntitySubscription entityName callback store = do
   subId <- Uuid.generate |> Task.map (toText .> SubscriptionId)
-  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = Nothing, entityNameFilter = Just entityName}
+  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = Nothing, entityNameFilter = Just entityName, onRemove = Task.yield unit}
   store.entitySubscriptions |> ConcurrentVar.modify \subscriptionsMap -> do
     let currentSubscriptions = subscriptionsMap |> Map.getOrElse entityName Map.empty
     subscriptionsMap |> Map.set entityName (currentSubscriptions |> Map.set subId subscriptionInfo)
@@ -130,7 +169,7 @@ addEntitySubscriptionFromPosition ::
   Task Error SubscriptionId
 addEntitySubscriptionFromPosition entityName startingPosition callback store = do
   subId <- Uuid.generate |> Task.map (toText .> SubscriptionId)
-  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = startingPosition, entityNameFilter = Just entityName}
+  let subscriptionInfo = SubscriptionInfo {callback, startingGlobalPosition = startingPosition, entityNameFilter = Just entityName, onRemove = Task.yield unit}
   store.entitySubscriptions |> ConcurrentVar.modify \subscriptionsMap -> do
     let currentSubscriptions = subscriptionsMap |> Map.getOrElse entityName Map.empty
     subscriptionsMap |> Map.set entityName (currentSubscriptions |> Map.set subId subscriptionInfo)
@@ -209,8 +248,39 @@ dispatch streamId message store = do
   allCallbacks |> AsyncTask.runAllIgnoringErrors
 
 
+-- | Find a subscription's 'onRemove' cleanup action across the global, stream,
+-- and entity maps. Returns 'Task.yield unit' when the id is unknown, so removing
+-- an already-removed or never-registered id is a no-op rather than an error.
+findOnRemove :: SubscriptionId -> SubscriptionStore -> Task Error (Task Text Unit)
+findOnRemove subId store = do
+  globalSubs <- store.globalSubscriptions |> ConcurrentVar.peek
+  streamSubsMap <- store.streamSubscriptions |> ConcurrentVar.peek
+  entitySubsMap <- store.entitySubscriptions |> ConcurrentVar.peek
+  -- Search the global map, then every stream's map, then every entity's map for
+  -- the id. SubscriptionIds are unique, so at most one map holds it.
+  let lookupIn subs acc =
+        case acc of
+          Just info -> Just info
+          Nothing -> subs |> Map.get subId
+  let fromStream = streamSubsMap |> Map.values |> Array.reduce lookupIn Nothing
+  let fromEntity = entitySubsMap |> Map.values |> Array.reduce lookupIn Nothing
+  let found =
+        case globalSubs |> Map.get subId of
+          Just info -> Just info
+          Nothing -> case fromStream of
+            Just info -> Just info
+            Nothing -> fromEntity
+  case found of
+    Just info -> Task.yield info.onRemove
+    Nothing -> Task.yield (Task.yield unit)
+
+
 removeSubscription :: SubscriptionId -> SubscriptionStore -> Task Error Unit
 removeSubscription subId store = do
+  -- Read the subscription's cleanup BEFORE deleting, so the cleanup is taken
+  -- (not left in place) and a double-remove cannot run it twice.
+  cleanup <- store |> findOnRemove subId
+
   -- Remove from global subscriptions
   store.globalSubscriptions
     |> ConcurrentVar.modify (Map.remove subId)
@@ -225,4 +295,11 @@ removeSubscription subId store = do
     subscriptionsMap
       |> Map.mapValues (\entitySubs -> entitySubs |> Map.remove subId)
 
-  Task.yield ()
+  -- Run the cleanup best-effort: a failed release is logged and swallowed so
+  -- unsubscribe stays total (ADR-0063 §2, design goal 4).
+  cleanup
+    |> Task.recover \err -> do
+      Log.warn [fmt|Subscription #{toText subId} cleanup failed: #{err}|]
+        |> Task.ignoreError
+      Task.yield unit
+    |> Task.mapError (\_ -> OtherError)