shibuya-core 0.9.0.1 → 0.9.0.2

raw patch · 5 files changed

+122/−93 lines, 5 filesdep ~effectfuldep ~streamly-coredep ~unliftioPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: effectful, streamly-core, unliftio

API changes (from Hackage documentation)

Files

• CHANGELOG.md +16/−0
• shibuya-core.cabal +21/−1
• src/Shibuya/Internal/Runner/Master.hs +11/−75
• test-gc/Main.hs +70/−0
• test/Shibuya/RunnerSpec.hs +4/−17

@@ -1,5 +1,21 @@ # Changelog +## 0.9.0.2 — 2026-09-19++### Bug Fixes++- Remove the unused linked master mailbox actor. Its inbox had no senders, so+  a caller that started an idle application and retained only `waitApp` could+  receive `ExceptionInLinkedThread ... thread blocked indefinitely in an STM+  transaction` during major garbage collection. `Master` now owns only the+  NQE supervisor and metrics registry; processor supervision, failure+  propagation, metrics access, and explicit shutdown behavior are unchanged.++### Other Changes++- Add a process-isolated garbage-collection regression that exercises bare+  `runApp`/`waitApp` liveness under the normal optimized test profile.+ ## 0.9.0.1 — 2026-09-15  ### Other Changes

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-core-version: 0.9.0.1+version: 0.9.0.2 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).@@ -150,3 +150,23 @@     time,     unliftio,     unordered-containers,++-- Isolate the bare-waitApp liveness probe so process exit cleans up without+-- retaining the master (which would conceal the regression).+test-suite shibuya-core-gc-test+  import: warnings+  default-language: GHC2024+  type: exitcode-stdio-1.0+  hs-source-dirs: test-gc+  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N2++  build-depends:+    base ^>=4.21.0.0,+    effectful,+    shibuya-core,+    streamly-core,+    unliftio,

@@ -2,11 +2,10 @@ -- No PVP guarantees: anything here may change or disappear in any release. -- Application authors should import "Shibuya" instead. ----- Master process - central coordinator for queue processors.+-- Master handle - owns the shared supervisor and metrics registry for queue processors. -- Provides supervision, metrics collection, and control API. -- -- Architecture:--- - Master is an NQE Process that handles control messages -- - Holds a Supervisor for managing child processors -- - Maintains TVar MetricsMap for O(1) metrics access -- - Processors register their metrics TVars with the Master@@ -15,9 +14,6 @@     Master (..),     MasterState (..), -    -- * Control Messages-    MasterMessage (..),-     -- * Starting the Master     startMaster,     stopMaster,@@ -34,13 +30,7 @@   ) where -import Control.Concurrent.NQE.Process-  ( Inbox,-    Listen,-    Process (..),-    newInbox,-    receive,-  )+import Control.Concurrent.NQE.Process (Process (..)) import Control.Concurrent.NQE.Supervisor (Strategy (..), Supervisor) import Control.Concurrent.NQE.Supervisor qualified as Supervisor import Control.Concurrent.STM@@ -50,7 +40,6 @@     newTVarIO,     readTVar,   )-import Control.Monad (forever) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Effectful (Eff, IOE, liftIO, (:>))@@ -62,16 +51,7 @@     sampleMetrics,   ) import Shibuya.Prelude-import UnliftIO (Async, async, cancel, link)---- | Messages for the master process.-data MasterMessage-  = -- | Register a processor's metrics handle-    RegisterProcessor !ProcessorId !MetricsHandle !(Listen ())-  | -- | Unregister a processor-    UnregisterProcessor !ProcessorId !(Listen ())-  | -- | Shutdown all processors-    Shutdown !(Listen ())+import UnliftIO (cancel)  -- | Master state held in TVars. data MasterState = MasterState@@ -86,19 +66,15 @@   }   deriving (Generic) --- | Master handle - provides access to the master process.-data Master = Master-  { -- | The async handle for the master-    handle :: !(Async ()),-    -- | Direct access to master state-    state :: !MasterState,-    -- | Inbox for sending messages-    inbox :: !(Inbox MasterMessage)+-- | Master handle - owns the shared supervisor and metrics registry.+newtype Master = Master+  { -- | Direct access to master state+    state :: MasterState   }   deriving (Generic)  -- | Start the master process.--- Returns a handle for interacting with the master.+-- Returns a handle for accessing shared application state. -- The caller is responsible for calling stopMaster when done. startMaster :: (IOE :> es) => Strategy -> Eff es Master startMaster strategy = liftIO $ do@@ -111,52 +87,12 @@         IgnoreGraceful -> True         IgnoreAll -> False         Notify _ -> False-      masterState = MasterState metricsMapVar sup propagate--  masterInbox <- newInbox--  -- Start master loop-  masterHandle <- async $ masterLoop masterState masterInbox-  link masterHandle--  pure-    Master-      { handle = masterHandle,-        state = masterState,-        inbox = masterInbox-      }+  pure Master {state = MasterState metricsMapVar sup propagate}  -- | Stop the master and all child processors.--- Cancels the supervisor first (which cancels all children via NQE's stopAll),--- then cancels the master message loop.+-- Cancels the supervisor, which cancels all children via NQE's stopAll. stopMaster :: (IOE :> es) => Master -> Eff es ()-stopMaster master = liftIO $ do-  -- Cancel the supervisor first - this triggers NQE's stopAll which cancels all children-  cancel (getProcessAsync master.state.supervisor)-  -- Then cancel the master message loop-  cancel master.handle---- | The master process main loop.-masterLoop :: MasterState -> Inbox MasterMessage -> IO ()-masterLoop state inbox = forever $ do-  msg <- receive inbox-  handleMessage state msg---- | Handle a single master message.-handleMessage :: MasterState -> MasterMessage -> IO ()-handleMessage state msg = case msg of-  RegisterProcessor pid metricsHandle respond -> do-    atomically $ do-      modifyTVar' state.metrics $ Map.insert pid metricsHandle-    atomically $ respond ()-  UnregisterProcessor pid respond -> do-    atomically $ do-      modifyTVar' state.metrics $ Map.delete pid-    atomically $ respond ()-  Shutdown respond -> do-    -- Clear all processors-    atomically $ modifyTVar' state.metrics (const Map.empty)-    atomically $ respond ()+stopMaster master = liftIO $ cancel (getProcessAsync master.state.supervisor)  -- | Get metrics for all processors. getAllMetrics :: (IOE :> es) => Master -> Eff es MetricsMap

@@ -0,0 +1,70 @@+{-# LANGUAGE OverloadedStrings #-}++-- This probe owns its process: retaining an AppHandle for cleanup would hide+-- the bug, and a weak pointer cannot guarantee cleanup after collection.+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Exception (SomeException, displayException, try)+import Control.Monad (replicateM_)+import Effectful (liftIO, runEff)+import Shibuya.Adapter (Adapter (..))+import Shibuya.App+  ( AppConfig (..),+    ProcessorId (..),+    SupervisionStrategy (..),+    defaultAppConfig,+    mkProcessor,+    runApp,+    waitApp,+  )+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Telemetry.Effect (runTracingNoop)+import Streamly.Data.Stream qualified as Stream+import System.Exit (die)+import System.Mem (performMajorGC)+import UnliftIO qualified as UIO++main :: IO ()+main = do+  started <- newEmptyMVar+  outcome <- try @SomeException $ UIO.timeout 5_000_000 $ UIO.race (gcWindow started) (worker started)+  case outcome of+    Left err -> die ("FAIL: bare waitApp died: " <> displayException err)+    Right Nothing -> die "FAIL: the GC observation window did not finish within five seconds"+    Right (Just (Right ())) -> die "FAIL: waitApp returned while the idle processor should still be running"+    Right (Just (Left ())) -> putStrLn "PASS: bare waitApp survives major collections"++gcWindow :: MVar () -> IO ()+gcWindow started = do+  -- Do not let a slow startup pass without exercising a running application.+  takeMVar started+  replicateM_ 5 $ do+    threadDelay 100_000+    performMajorGC+  -- Leave time for a linked exception from the last collection to arrive.+  threadDelay 200_000++worker :: MVar () -> IO ()+worker started = runEff $ runTracingNoop $ do+  -- A timer keeps the ingester alive just as a real adapter's poll delay does.+  -- An unreachable STM retry here would introduce a separate deadlock.+  let adapter =+        Adapter+          { adapterName = "gc-regression:idle",+            source = Stream.fromEffect $ liftIO $ do+              threadDelay 60_000_000+              fail "the idle adapter unexpectedly woke during the GC probe",+            shutdown = pure ()+          }+  result <-+    runApp+      defaultAppConfig {strategy = IgnoreFailures}+      [(ProcessorId "idle", mkProcessor adapter (\_ -> pure AckOk))]+  case result of+    Left err -> liftIO $ die ("runApp failed: " <> show err)+    Right app -> do+      liftIO $ putMVar started ()+      -- Deliberately no metrics server, retained handle, or subsequent stopApp.+      waitApp app

@@ -70,11 +70,7 @@           Left err -> pure $ Left err           Right appHandle -> do             waitApp appHandle-            -- Stop the app to cancel the (always-linked) master coordinator.-            -- Without this the idle master blocks forever on its mailbox and the-            -- RTS eventually raises BlockedIndefinitelyOnSTM, which propagates-            -- through the link as a flaky ExceptionInLinkedThread landing on-            -- whichever test happens to be running when GC fires.+            -- Stop the supervisor and its children so they do not outlive this test.             stopApp appHandle             pure $ Right () @@ -107,8 +103,7 @@             pure (decs, Left err)           Right appHandle -> do             waitApp appHandle-            -- See note in "processes messages from mock adapter": stop the app so-            -- the linked master does not deadlock and flake a later test.+            -- Stop the supervisor and its children so they do not outlive this test.             stopApp appHandle             decs <- liftIO $ readIORef tracking.trackedDecisions             pure (decs, Right ())@@ -141,11 +136,7 @@           Left err -> pure $ Left err           Right appHandle -> do             waitApp appHandle-            -- Stop the app to cancel the (always-linked) master coordinator.-            -- Without this the idle master blocks forever on its mailbox and the-            -- RTS eventually raises BlockedIndefinitelyOnSTM, which propagates-            -- through the link as a flaky ExceptionInLinkedThread landing on-            -- whichever test happens to be running when GC fires.+            -- Stop the supervisor and its children so they do not outlive this test.             stopApp appHandle             pure $ Right () @@ -200,11 +191,7 @@           Left err -> pure $ Left err           Right appHandle -> do             waitApp appHandle-            -- Stop the app to cancel the (always-linked) master coordinator.-            -- Without this the idle master blocks forever on its mailbox and the-            -- RTS eventually raises BlockedIndefinitelyOnSTM, which propagates-            -- through the link as a flaky ExceptionInLinkedThread landing on-            -- whichever test happens to be running when GC fires.+            -- Stop the supervisor and its children so they do not outlive this test.             stopApp appHandle             pure $ Right ()