Fix memory memory leak

- Tuple to strict pair
- Self-recursive loops with more knot tying <- this was the thing

src/AFRP.hs

@@ -20,6 +20,7 @@ module AFRP
   , (>>|)
   , toEvent
   , lMerge
+  , Pair(..)
   , Request(..)
   , edge
   , dropFirst
@@ -48,34 +49,40 @@ import Data.UUID (UUID)
 import qualified Data.Set as S
 import qualified Data.Text as T
 
+data Pair a b = Pair !a !b
+
 data Request = Request
   { requestTime :: !UTCTime
   , requestTimeZone :: !TimeZone
   , requestTraceId :: !UUID
-  } deriving Show
+  } deriving (Show, Eq)
 
 -- | The set of entity ids an arrow subscribes to. Static: it does not
 -- change as the machine steps, so the runtime can read it once to build
 -- trigger subscriptions.
 data Mealy eff a b = Mealy
   { entities :: S.Set T.Text
-  , runMealy :: forall m. MonadFix m => (forall x. eff x -> m x) -> Request -> a -> m (b, Mealy eff a b)
+  , runMealy :: forall m. MonadFix m => (forall x. eff x -> m x) -> Request -> a -> m (Pair b (Mealy eff a b))
   }
 
 
 instance Semigroup b => Semigroup (Mealy eff a b) where
   Mealy ast af <> Mealy bst bf = Mealy (ast <> bst) $ \nt r a -> do
-    (x, af') <- af nt r a
-    (x', bf') <- bf nt r a
-    pure (x <> x', af' <> bf')
+    Pair x af' <- af nt r a
+    Pair x' bf' <- bf nt r a
+    pure (Pair (x <> x') (af' <> bf'))
 
 
 instance Monoid b => Monoid (Mealy eff a b) where
-  mempty = Mealy mempty $ \_ _ _ -> pure (mempty, mempty)
+  mempty = m
+    where
+      m = Mealy mempty $ \_ _ _ -> pure (Pair mempty m)
 
 eff :: (Request -> a -> eff b) -> Mealy eff a b
-eff f = Mealy mempty $ \nt req x ->
-  nt (f req x) >>= \b -> pure (b, eff f)
+eff f = m
+  where
+    m = Mealy mempty $ \nt req x ->
+      nt (f req x) >>= \b -> pure (Pair b m)
 
 -- | Override the static entity set of an arrow. Use when a combinator
 -- (e.g. 'switch') hides continuation entities from the runtime's
@@ -84,41 +91,48 @@ withEntities :: S.Set T.Text -> Mealy eff a b -> Mealy eff a b
 withEntities es (Mealy _ f) = Mealy es f
 
 instance Category (Mealy eff) where
-  id = Mealy mempty (\_ _ x -> pure (x, id))
+  id = Mealy mempty (\_ _ x -> pure (Pair x id))
   (Mealy ast f) . (Mealy bst g) = Mealy (ast <> bst) $ \nt t a -> do
-    (b, g') <- g nt t a
-    (c, f') <- f nt t b
-    pure (c, f' . g')
+    Pair b g' <- g nt t a
+    Pair c f' <- f nt t b
+    pure (Pair c (f' . g'))
 
 instance Arrow (Mealy eff) where
-  arr f = Mealy mempty $ \_ _ b -> pure (f b, arr f)
+  arr f = mealy
+    where
+      mealy = Mealy mempty $ \_ _ b -> pure (Pair (f b) mealy)
   first (Mealy st f) = Mealy st $ \nt t (b,d) -> do
-    (c, f') <- f nt t b
-    pure ((c, d), first f')
+    Pair c f' <- f nt t b
+    pure (Pair (c, d) (first f'))
 
 instance ArrowChoice (Mealy eff) where
-  left m@(Mealy st f) = Mealy st $ \nt t -> \case
-    Left b -> do
-      (c, f') <- f nt t b
-      pure (Left c, left f')
-    Right d -> pure (Right d, left m)
+  left (Mealy st f) = lm
+    where
+      lm = Mealy st $ \nt t -> \case
+        Left b -> do
+          Pair c f' <- f nt t b
+          pure (Pair (Left c) (left f'))
+        Right d -> pure (Pair (Right d) lm)
 
-instance ArrowLoop (Mealy eff) where
-  loop (Mealy st f) = Mealy st $ \nt t b -> do
-    ((c,_), f') <- mfix $ \((_,d), _) -> f nt t (b,d)
-    pure (c, loop f')
+-- ArrowLoop is incompatible with strict Pair (strict fields prevent
+-- the lazy knot-tying that mfix requires with loop).
+-- instance ArrowLoop (Mealy eff) where
+--   loop (Mealy st f) = Mealy st $ \nt t b -> do
+--     Pair (c,_) f' <- mfix $ \(Pair (_,d) _) -> f nt t (b,d)
+--     pure (Pair c (loop f'))
 
 instance Functor (Mealy eff a) where
   fmap f (Mealy st g) = Mealy st $ \nt t a -> do
-    (b, g') <- g nt t a
-    pure (f b, fmap f g')
+    Pair b g' <- g nt t a
+    pure (Pair (f b) (fmap f g'))
 
 instance Applicative (Mealy eff a) where
-  pure b = Mealy mempty $ \_ _ _ -> pure (b, pure b)
+  pure b = Mealy mempty $ \_ _ _ -> pure (Pair b (pure b))
   Mealy ast f <*> Mealy bst x = Mealy (ast <> bst) $ \nt t a -> do
-    (f', fNext) <- f nt t a
-    (x', xNext) <- x nt t a
-    pure (f' x', fNext <*> xNext)
+    b' <- x nt t a
+    let Pair x' xNext = b'
+    Pair f' fNext <- f nt t a
+    pure (Pair (f' x') (fNext <*> xNext))
 
 data Event a
   = Tick
@@ -133,8 +147,8 @@ instance Monoid (Event a) where
 
 hold :: a -> Mealy eff (Event a) a
 hold a = Mealy mempty $ \_ _ -> \case
-  Tick -> pure (a, hold a)
-  Event a' -> pure (a', hold a')
+  Tick -> pure (Pair a (hold a))
+  Event a' -> pure (Pair a' (hold a'))
 
 events :: Mealy eff (Event a) (Either () a)
 events = arr $ \case
@@ -150,9 +164,9 @@ tag b ev = b <$ ev
 
 switch :: Mealy eff a (b, Event c) -> (c -> Mealy eff a b) -> Mealy eff a b
 switch (Mealy st f) s = Mealy st $ \nt t a -> do
-  ((b, ev), f') <- f nt t a
+  Pair (b, ev) f' <- f nt t a
   case ev of
-          Tick -> pure (b, switch f' s)
+          Tick -> pure (Pair b (switch f' s))
           Event x -> runMealy (s x) nt t a
 
 sample :: Mealy eff (a, Event b) (Event a)
@@ -163,28 +177,28 @@ preMapAccum f x extract = go x
   where
     go b = Mealy mempty $ \_ _ a ->
       let next = f b a
-      in pure (extract b, go next)
+      in pure (Pair (extract b) (go next))
 
 preMapAccumRequest :: (Request -> x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 preMapAccumRequest f x extract = go x
   where
     go b = Mealy mempty $ \_ t a ->
       let next = f t b a
-      in pure (extract b, go next)
+      in pure (Pair (extract b) (go next))
 
 mapAccum :: (x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 mapAccum f x extract = go x
   where
     go b = Mealy mempty $ \_ _ a ->
       let next = f b a
-      in pure (extract next, go next)
+      in pure (Pair (extract next) (go next))
 
 mapAccumRequest :: (Request -> x -> a -> x) -> x -> (x -> b) -> Mealy eff a b
 mapAccumRequest f x extract = go x
   where
     go b = Mealy mempty $ \_ t a ->
       let next = f t b a
-      in pure (extract next, go next)
+      in pure (Pair (extract next) (go next))
 
 data DelayState x a = DelayState
   { pending :: x
@@ -265,11 +279,11 @@ edge :: Mealy eff Bool (Event ())
 edge = go False
   where
     go True = Mealy mempty $ \_ _ -> \case
-      True -> pure (Tick, go True)
-      False -> pure (Tick, go False)
+      True -> pure (Pair Tick (go True))
+      False -> pure (Pair Tick (go False))
     go False = Mealy mempty $ \_ _ -> \case
-      True -> pure (Event (), go True)
-      False -> pure (Tick, go False)
+      True -> pure (Pair (Event ()) (go True))
+      False -> pure (Pair Tick (go False))
 
 
 -- | Drop the first 'Event' and pass through everything after. Useful for
@@ -280,8 +294,8 @@ dropFirst = go False
   where
     go seen = Mealy mempty $ \_ _ input ->
       case input of
-        Event _ | not seen -> pure (Tick, go True)
-        _ -> pure (input, go seen)
+        Event _ | not seen -> pure (Pair Tick (go True))
+        _ -> pure (Pair input (go seen))
 
 
 duration :: forall eff a. Mealy eff a NominalDiffTime
@@ -340,7 +354,7 @@ fixed seconds = mapAccumRequest go Nothing (maybe [] ((`appEndo` []) . snd))
 
 currentTime :: Mealy eff a LocalTime
 currentTime = Mealy mempty $ \_ Request{requestTime, requestTimeZone} _ ->
-  pure (utcToLocalTime requestTimeZone requestTime, currentTime)
+  pure (Pair (utcToLocalTime requestTimeZone requestTime) currentTime)
 
 
 onEvent :: Mealy eff a () -> Mealy eff (Event a) ()

src/HomeAssistant/Controller.hs

@@ -27,7 +27,7 @@ module HomeAssistant.Controller
   , Light(..)
   ) where
 
-import AFRP (Mealy (..), eff, Event(..), events, filterA, (>>|), toEvent, Request)
+import AFRP (Mealy (..), Pair (..), eff, Event(..), events, filterA, (>>|), toEvent, Request)
 import Control.Arrow (Arrow(..), returnA)
 import Control.Category ((>>>))
 import Data.Aeson (Value, object, (.=))
@@ -39,7 +39,7 @@ import qualified Data.Text.Lens as TL
 import Data.Bool (bool)
 
 data Target = EntityId !T.Text | AreaId !T.Text
-  deriving (Show,Eq)
+  deriving (Show,Eq,Ord)
 
 data Service = Service
   { serviceDomain :: T.Text
@@ -65,9 +65,11 @@ debug = proc x -> do
   returnA -< x
 
 traceEvent :: Show a => HASS (Event a) (Event a)
-traceEvent = Mealy mempty $ \nt req -> \case
-  Event a -> nt (Trace req a) >>= \() -> pure (Event a, traceEvent)
-  Tick -> pure (Tick, traceEvent)
+traceEvent = m
+  where
+    m = Mealy mempty $ \nt req -> \case
+      Event a -> nt (Trace req a) >>= \() -> pure (Pair (Event a) m)
+      Tick -> pure (Pair Tick m)
 
 traceValue :: Show a => HASS a a
 traceValue = proc x -> do

src/HomeAssistant/Runtime.hs

@@ -13,7 +13,7 @@ module HomeAssistant.Runtime
   , runController
   ) where
 
-import AFRP (Event (..), Mealy (..), Request (..))
+import AFRP (Event (..), Mealy (..), Pair (..), Request (..))
 import Control.Concurrent.Async (async, waitAny)
 import Control.Concurrent.STM (atomically, dupTChan, readTChan)
 import Data.Aeson (Value)
@@ -35,7 +35,7 @@ import Control.Monad.Fix (MonadFix)
 import HomeAssistant.Controller.Ruuvi (ruuviController)
 import HomeAssistant.Controller.Children (schoolLightController)
 
-step :: (MonadFix m, MonadIO m) => (forall x. eff x -> m x) -> UUID -> Mealy eff a b -> a -> m (b, Mealy eff a b)
+step :: (MonadFix m, MonadIO m) => (forall x. eff x -> m x) -> UUID -> Mealy eff a b -> a -> m (Pair b (Mealy eff a b))
 step nt trace (Mealy _ f) a = do
   now <- liftIO getCurrentTime
   tz <- liftIO getCurrentTimeZone
@@ -65,7 +65,7 @@ runController bus (Controller name machine _enabled) = do
       msg <- atomically (readTChan inbound)
       uuid <- UUID.V4.nextRandom
       let ns = Namespace [name]
-      (_, f') <- step (runKatipContextT (busLogEnv bus) () ns . channelHassEval bus) uuid f msg
+      Pair _ f' <- step (runKatipContextT (busLogEnv bus) () ns . channelHassEval bus) uuid f msg
       go inbound f'
 
 defaultMain :: IO ()

src/HomeAssistant/Runtime/Connection.hs

@@ -5,13 +5,16 @@ module HomeAssistant.Runtime.Connection
   ( readerAction
   , writerAction
   , encodeService
+  , dedupeBatch
   ) where
 
 import Control.Concurrent.STM
-  ( atomically
+  ( TChan
+  , atomically
   , readTChan
   , readTVar
   , retry
+  , tryReadTChan
   , writeTChan
   , writeTVar
   )
@@ -23,6 +26,8 @@ import Control.Lens ((^?))
 import Control.Monad (forever, forM_)
 import Data.Aeson (Value, eitherDecode, encode, object, (.=))
 import Data.Aeson.Lens (key, _String)
+import Data.List (sort)
+import qualified Data.Map.Strict as M
 import qualified Data.Set as S
 import qualified Data.Text as T
 import Data.Void (Void)
@@ -86,7 +91,7 @@ subscribe bus conn ents =
 -- even when no state changes arrive.
 receiveLoop :: Bus -> WS.Connection -> IO Void
 receiveLoop bus conn = forever $ do
-  winner <- race (threadDelay 1000000) (WS.receiveData conn)
+  winner <- race (threadDelay 1_000_000) (WS.receiveData conn)
   case winner of
     Left () -> atomically $ writeTChan (busInbound bus) Tick
     Right msg -> case eitherDecode msg of
@@ -100,16 +105,41 @@ receiveJSON conn = do
     Left err -> throw (Fatal $ "Invalid JSON from Home Assistant: " <> T.pack err)
     Right x  -> pure x
 
-writerAction :: Bus -> IO Void
-writerAction bus = forever $ do
-  (request, svc) <- atomically $ readTChan (busOutbound bus)
-  conn <- atomically $ readTVar (busConn bus) >>= maybe retry pure
+-- | Floor between sends within a batch: 100ms, so a many-distinct-target
+-- flood still caps at ~10 sends/sec even after dedupe.
+minInterval :: Int
+minInterval = 100000
+
+-- | Non-blocking drain of everything queued on a channel. Returns items
+-- oldest-first (FIFO from the channel), so prepending the blocking
+-- `readTChan` item keeps the whole batch oldest-first for `dedupeBatch`.
+drainTry :: TChan a -> IO [a]
+drainTry chan = go []
+  where
+    go acc = do
+      m <- atomically $ tryReadTChan chan
+      case m of
+        Nothing -> pure (reverse acc)
+        Just x  -> go (x : acc)
+
+sendWithId :: Bus -> WS.Connection -> Request -> Service -> IO ()
+sendWithId bus conn request svc = do
   callId <- generateCallId (busGen bus)
   let textData = encode $ encodeService callId svc
   runKatipContextT (busLogEnv bus) (sl "traceId" (toText (requestTraceId request))) "connection" $
     logFM DebugS (ls textData)
   WS.sendTextData conn textData
 
+writerAction :: Bus -> IO Void
+writerAction bus = forever $ do
+  first <- atomically $ readTChan (busOutbound bus)
+  rest  <- drainTry (busOutbound bus)
+  let deduped = dedupeBatch (first : rest)
+  conn <- atomically $ readTVar (busConn bus) >>= maybe retry pure
+  forM_ deduped $ \(request, svc) -> do
+    sendWithId bus conn request svc
+    threadDelay minInterval
+
 encodeService :: Int -> Service -> Value
 encodeService callId Service{..} = object $
   [ "id"      .= callId
@@ -119,6 +149,18 @@ encodeService callId Service{..} = object $
   , "target"  .= targetObject serviceTarget
   ] <> maybe [] (\d -> ["service_data" .= d]) serviceData
 
+-- | Collapse a drained batch of outbound calls: the newest call per
+-- `(domain, service, sorted-targets)` survives; older duplicates are
+-- dropped. `serviceData` is not part of the key, so a newer `turn_on`
+-- with different brightness supersedes an older one to the same target.
+dedupeBatch :: [(Request, Service)] -> [(Request, Service)]
+dedupeBatch = M.elems . foldl' ins M.empty
+  where
+    ins m (req, svc) = M.insert (dedupeKey svc) (req, svc) m
+
+dedupeKey :: Service -> (T.Text, T.Text, [Target])
+dedupeKey Service{..} = (serviceDomain, serviceName, sort serviceTarget)
+
 -- | A single target encodes as a scalar; multiple encode as a list. Empty
 -- lists are omitted so Home Assistant receives only populated keys.
 targetObject :: [Target] -> Value

test/AFRPSpec.hs

@@ -28,14 +28,14 @@ sec n = UTCTime (toEnum 0) (fromIntegral n)
 runPure :: Mealy Identity a b -> [a] -> [b]
 runPure _ [] = []
 runPure m (a : as) = case runIdentity (AFRP.runMealy m id fakeRequest a) of
-  (b, m') -> b : runPure m' as
+  Pair b m' -> b : runPure m' as
 
 -- | Run a Mealy with a per-step wall clock (seconds since the day-0 epoch).
 runTimed :: Mealy Identity a b -> [(Integer, a)] -> [b]
 runTimed _ [] = []
 runTimed m ((s, a) : as) =
   case runIdentity (AFRP.runMealy m id (Request (sec s) utc nil) a) of
-    (b, m') -> b : runTimed m' as
+    Pair b m' -> b : runTimed m' as
 
 -- | A minimal State monad for observing effectful arrows (e.g. whenA gating).
 newtype St a = St { unSt :: Int -> (a, Int) }
@@ -59,7 +59,7 @@ runStEff m s0 as = go m s0 as
     go _ s [] = ([], s)
     go m' s (a : rest) =
       case unSt (AFRP.runMealy m' id fakeRequest a) s of
-        ((b, m''), s') -> let (bs, s'') = go m'' s' rest in (b : bs, s'')
+        (Pair b m'', s') -> let (bs, s'') = go m'' s' rest in (b : bs, s'')
 
 spec :: Spec
 spec = describe "AFRP" $ do
@@ -569,11 +569,11 @@ sampleSpec = describe "sample" $ do
 
 -- | A stateless arrow carrying a fixed entity set, for testing propagation.
 subscribed :: S.Set T.Text -> Mealy Identity Int Int
-subscribed ents = Mealy ents $ \_ _ a -> pure (a, subscribed ents)
+subscribed ents = Mealy ents $ \_ _ a -> pure (Pair a (subscribed ents))
 
 -- | Same as 'subscribed' but yields a function, for testing '<*>'.
 subscribedF :: S.Set T.Text -> Mealy Identity Int (Int -> Int)
-subscribedF ents = Mealy ents $ \_ _ a -> pure ((a +), subscribedF ents)
+subscribedF ents = Mealy ents $ \_ _ a -> pure (Pair (a +) (subscribedF ents))
 
 entitiesSpec :: Spec
 entitiesSpec = describe "entities" $ do

test/ConnectionSpec.hs

@@ -2,31 +2,78 @@
 
 module ConnectionSpec (spec) where
 
+import AFRP (Request(..))
 import Data.Aeson (object, (.=))
+import Data.Maybe (fromJust)
 import Data.Text (Text)
+import Data.Time (UTCTime (..), utc)
+import Data.UUID (fromString)
 import HomeAssistant.Controller (Service (..), Target(..))
-import HomeAssistant.Runtime.Connection (encodeService)
+import HomeAssistant.Runtime.Connection (encodeService, dedupeBatch)
 import Test.Hspec
 
 spec :: Spec
-spec = describe "encodeService" $ do
-  it "encodes a call_service message" $
-    encodeService 7 (Service "light" "turn_on" Nothing [EntityId "light.bedroom_masse"])
-      `shouldBe` object
-        [ "id" .= (7 :: Int)
-        , "type" .= ("call_service" :: Text)
-        , "domain" .= ("light" :: Text)
-        , "service" .= ("turn_on" :: Text)
-        , "target" .= object ["entity_id" .= ("light.bedroom_masse" :: Text)]
-        ]
+spec = do
+  describe "encodeService" $ do
+    it "encodes a call_service message" $
+      encodeService 7 (Service "light" "turn_on" Nothing [EntityId "light.bedroom_masse"])
+        `shouldBe` object
+          [ "id" .= (7 :: Int)
+          , "type" .= ("call_service" :: Text)
+          , "domain" .= ("light" :: Text)
+          , "service" .= ("turn_on" :: Text)
+          , "target" .= object ["entity_id" .= ("light.bedroom_masse" :: Text)]
+          ]
 
-  it "includes service_data when present" $
-    encodeService 8 (Service "light" "turn_on" (Just (object ["brightness" .= (200 :: Int)])) [EntityId "light.bedroom_masse"])
-      `shouldBe` object
-        [ "id" .= (8 :: Int)
-        , "type" .= ("call_service" :: Text)
-        , "domain" .= ("light" :: Text)
-        , "service" .= ("turn_on" :: Text)
-        , "target" .= object ["entity_id" .= ("light.bedroom_masse" :: Text)]
-        , "service_data" .= object ["brightness" .= (200 :: Int)]
-        ]
+    it "includes service_data when present" $
+      encodeService 8 (Service "light" "turn_on" (Just (object ["brightness" .= (200 :: Int)])) [EntityId "light.bedroom_masse"])
+        `shouldBe` object
+          [ "id" .= (8 :: Int)
+          , "type" .= ("call_service" :: Text)
+          , "domain" .= ("light" :: Text)
+          , "service" .= ("turn_on" :: Text)
+          , "target" .= object ["entity_id" .= ("light.bedroom_masse" :: Text)]
+          , "service_data" .= object ["brightness" .= (200 :: Int)]
+          ]
+
+  describe "dedupeBatch" $ do
+    it "collapses identical calls to one" $
+      let batch = [ (req 1, lightOn [AreaId "x"])
+                  , (req 2, lightOn [AreaId "x"])
+                  , (req 3, lightOn [AreaId "x"])
+                  ]
+      in dedupeBatch batch `shouldBe` [(req 3, lightOn [AreaId "x"])]
+
+    it "keeps same-target different-service calls separate" $
+      let batch = [ (req 1, lightOn  [AreaId "x"])
+                  , (req 2, lightOff [AreaId "x"])
+                  ]
+          result = dedupeBatch batch
+      in length result `shouldBe` 2
+
+    it "newest call wins for the same key" $
+      let batch = [ (req 1, lightOn [AreaId "x"])
+                  , (req 2, lightOn [AreaId "x"])
+                  , (req 3, lightOn [AreaId "x"])
+                  ]
+      in map requestTraceId (map fst (dedupeBatch batch)) `shouldBe`
+           [fromJust (fromString "00000000-0000-0000-0000-000000000003")]
+
+    it "treats target lists in different order as the same key" $
+      let batch = [ (req 1, lightOn [EntityId "a", EntityId "b"])
+                  , (req 2, lightOn [EntityId "b", EntityId "a"])
+                  ]
+      in length (dedupeBatch batch) `shouldBe` 1
+
+req :: Int -> Request
+req n = Request (UTCTime (toEnum 0) (fromIntegral (0 :: Int))) utc
+          (fromJust (fromString uuid))
+  where
+    pad i = replicate (12 - length (show i)) '0' <> show i
+    uuid = "00000000-0000-0000-0000-" <> pad n
+
+lightOn :: [Target] -> Service
+lightOn targets = Service "light" "turn_on" Nothing targets
+
+lightOff :: [Target] -> Service
+lightOff targets = Service "light" "turn_off" Nothing targets

test/Support.hs

@@ -16,7 +16,7 @@ import Data.Aeson (Value, object, (.=))
 import qualified Data.Text as T
 import Data.Time (UTCTime (..), utc)
 import Data.UUID (nil)
-import AFRP (Mealy (..), Request (..))
+import AFRP (Mealy (..), Pair (..), Request (..))
 import HomeAssistant.Controller (HASSEff (..), Service)
 
 fakeRequest :: Request
@@ -52,7 +52,7 @@ runHASS :: Mealy HASSEff a b -> [a] -> [(b, [Service])]
 runHASS _ [] = []
 runHASS m (a : as) =
   case runAcc (runMealy m interp fakeRequest a) [] of
-    ((b, m'), svcs) -> (b, svcs) : runHASS m' as
+    (Pair b m', svcs) -> (b, svcs) : runHASS m' as
 
 services :: [(b, [Service])] -> [[Service]]
 services = map snd