Preserve outbound backlog across outages, fix backoff quiet period
This commit is contained in:
@@ -23,7 +23,7 @@ import HomeAssistant.Controller (HASSEff (..), Service)
|
|||||||
import Network.WebSockets (Connection)
|
import Network.WebSockets (Connection)
|
||||||
|
|
||||||
-- | Shared runtime state: inbound is a broadcast channel (controllers
|
-- | Shared runtime state: inbound is a broadcast channel (controllers
|
||||||
-- read from 'dupTChanIO' copies), outbound queues service calls for the
|
-- read from 'dupTChan' copies), outbound queues service calls for the
|
||||||
-- writer, conn holds the current websocket (Nothing before first connect).
|
-- writer, conn holds the current websocket (Nothing before first connect).
|
||||||
data Bus = Bus
|
data Bus = Bus
|
||||||
{ busInbound :: TChan Value
|
{ busInbound :: TChan Value
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import Control.Concurrent.STM
|
|||||||
, writeTChan
|
, writeTChan
|
||||||
, writeTVar
|
, writeTVar
|
||||||
)
|
)
|
||||||
|
import Control.Exception (onException)
|
||||||
import Control.Exception.Annotated (throw)
|
import Control.Exception.Annotated (throw)
|
||||||
import Control.Lens ((^?))
|
import Control.Lens ((^?))
|
||||||
import Control.Monad (forever)
|
import Control.Monad (forever)
|
||||||
@@ -38,7 +39,8 @@ readerAction host port token bus =
|
|||||||
subscribe bus conn
|
subscribe bus conn
|
||||||
atomically $ writeTVar (busConn bus) (Just conn)
|
atomically $ writeTVar (busConn bus) (Just conn)
|
||||||
putStrLn "[reader] connected"
|
putStrLn "[reader] connected"
|
||||||
receiveLoop bus conn
|
-- Unpublish on exit so the writer blocks and the backlog survives the outage.
|
||||||
|
receiveLoop bus conn `onException` atomically (writeTVar (busConn bus) Nothing)
|
||||||
|
|
||||||
handshake :: WS.Connection -> String -> IO ()
|
handshake :: WS.Connection -> String -> IO ()
|
||||||
handshake conn token = do
|
handshake conn token = do
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ defaultBackoff = Backoff 0.1 30 30
|
|||||||
|
|
||||||
-- | Delay before the @attempt@-th restart: doubles from base, clamped at cap.
|
-- | Delay before the @attempt@-th restart: doubles from base, clamped at cap.
|
||||||
backoffDelay :: Backoff -> Int -> NominalDiffTime
|
backoffDelay :: Backoff -> Int -> NominalDiffTime
|
||||||
backoffDelay (Backoff base cap _) attempt = go (attempt - 1) base
|
backoffDelay (Backoff base cap _) attempt = go (max 0 (attempt - 1)) base
|
||||||
where
|
where
|
||||||
go 0 d = d
|
go 0 d = d
|
||||||
go n d = go (n - 1) (min cap (d * 2))
|
go n d = go (n - 1) (min cap (d * 2))
|
||||||
@@ -68,13 +68,13 @@ supervised name backoff action = go 1
|
|||||||
[ Handler $ \(f :: Fatal) -> pure (Just (Left f))
|
[ Handler $ \(f :: Fatal) -> pure (Just (Left f))
|
||||||
, Handler $ \(e :: SomeException) -> pure (Just (Right e))
|
, Handler $ \(e :: SomeException) -> pure (Just (Right e))
|
||||||
]
|
]
|
||||||
|
end <- getCurrentTime
|
||||||
case outcome of
|
case outcome of
|
||||||
Nothing -> error "unreachable: supervised action returned"
|
Nothing -> error "unreachable: supervised action returned"
|
||||||
Just (Left f) -> throw f
|
Just (Left f) -> throw f
|
||||||
Just (Right e) -> do
|
Just (Right e) -> do
|
||||||
putStrLn $ "[" <> T.unpack name <> "] attempt " <> show attempt <> " crashed: " <> displayException e
|
putStrLn $ "[" <> T.unpack name <> "] attempt " <> show attempt <> " crashed: " <> displayException e
|
||||||
let delay = backoffDelay backoff attempt
|
let delay = backoffDelay backoff attempt
|
||||||
putStrLn $ "[" <> T.unpack name <> "] restarting in " <> show delay <> "s"
|
putStrLn $ "[" <> T.unpack name <> "] restarting in " <> show delay
|
||||||
threadDelay (round (realToFrac delay * 1000000 :: Double))
|
threadDelay (round (realToFrac delay * 1000000 :: Double))
|
||||||
end <- getCurrentTime
|
|
||||||
go (nextAttempt backoff (diffUTCTime end start) attempt)
|
go (nextAttempt backoff (diffUTCTime end start) attempt)
|
||||||
|
|||||||
+2
-1
@@ -3,7 +3,7 @@ module RuntimeSpec (spec) where
|
|||||||
|
|
||||||
import Control.Concurrent (threadDelay)
|
import Control.Concurrent (threadDelay)
|
||||||
import Control.Concurrent.Async (async)
|
import Control.Concurrent.Async (async)
|
||||||
import Control.Concurrent.STM (atomically, readTChan, writeTChan)
|
import Control.Concurrent.STM (atomically, isEmptyTChan, readTChan, writeTChan)
|
||||||
import Data.Aeson (Value, object, (.=))
|
import Data.Aeson (Value, object, (.=))
|
||||||
import Data.Text (Text)
|
import Data.Text (Text)
|
||||||
import HomeAssistant.Controller (light, lightController)
|
import HomeAssistant.Controller (light, lightController)
|
||||||
@@ -24,6 +24,7 @@ spec = describe "runController" $ do
|
|||||||
svc2 <- atomically (readTChan (busOutbound bus))
|
svc2 <- atomically (readTChan (busOutbound bus))
|
||||||
svc1 `shouldBe` light True
|
svc1 `shouldBe` light True
|
||||||
svc2 `shouldBe` light False
|
svc2 `shouldBe` light False
|
||||||
|
atomically (isEmptyTChan (busOutbound bus)) `shouldReturn` True
|
||||||
|
|
||||||
doorEvent :: Text -> Value
|
doorEvent :: Text -> Value
|
||||||
doorEvent state = object
|
doorEvent state = object
|
||||||
|
|||||||
Reference in New Issue
Block a user