packages feed

shibuya-core 0.9.0.2 → 0.9.0.3

raw patch · 5 files changed

+271/−7 lines, 5 filesdep ~effectfuldep ~nqedep ~streamly-corePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: effectful, nqe, streamly-core, unliftio

API changes (from Hackage documentation)

Files

CHANGELOG.md view
@@ -1,5 +1,29 @@ # Changelog +## 0.9.0.3 — 2026-09-20++### Bug Fixes++- Start the NQE supervisor without linking it to the thread that called `runApp`.+  NQE links every process it starts, and a supervisor with no children left can+  only be woken through a mailbox reachable solely through the application handle.+  A caller whose processors had all finished, halted or failed, and which dropped+  the handle and kept running, was therefore killed at the next major garbage+  collection by `ExceptionInLinkedThread ... thread blocked indefinitely in an STM+  transaction`. Such a supervisor is now simply collected.+- Deliver a processor failure to the caller exactly once under `StopAllOnFailure`.+  The supervisor's link re-delivered the failure that the processor's own link had+  already delivered, so a caller that handled the first `ExceptionInLinkedThread`+  could be killed by a second one moments later. Sibling shutdown and propagation+  are unchanged; they never depended on the supervisor's link.++### Other Changes++- Add a second process-isolated garbage-collection regression,+  `shibuya-core-gc-finished-test`, covering an application that has finished and+  whose handle has been dropped, and a lifecycle test asserting a single failure+  delivery. `cabal test shibuya-core` now runs three suites.+ ## 0.9.0.2 — 2026-09-19  ### Bug Fixes
shibuya-core.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-core-version: 0.9.0.2+version: 0.9.0.3 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).@@ -167,6 +167,25 @@   build-depends:     base ^>=4.21.0.0,     effectful,+    shibuya-core,+    streamly-core,+    unliftio,++test-suite shibuya-core-gc-finished-test+  import: warnings+  default-language: GHC2024+  type: exitcode-stdio-1.0+  hs-source-dirs: test-gc+  main-is: Finished.hs+  ghc-options:+    -threaded+    -rtsopts+    -with-rtsopts=-N2++  build-depends:+    base ^>=4.21.0.0,+    effectful,+    nqe,     shibuya-core,     streamly-core,     unliftio,
src/Shibuya/Internal/Runner/Master.hs view
@@ -30,7 +30,7 @@   ) where -import Control.Concurrent.NQE.Process (Process (..))+import Control.Concurrent.NQE.Process (Process (..), newMailbox) import Control.Concurrent.NQE.Supervisor (Strategy (..), Supervisor) import Control.Concurrent.NQE.Supervisor qualified as Supervisor import Control.Concurrent.STM@@ -51,7 +51,7 @@     sampleMetrics,   ) import Shibuya.Prelude-import UnliftIO (cancel)+import UnliftIO (async, cancel)  -- | Master state held in TVars. data MasterState = MasterState@@ -76,10 +76,20 @@ -- | Start the master process. -- Returns a handle for accessing shared application state. -- The caller is responsible for calling stopMaster when done.+--+-- The supervisor is deliberately not linked to the calling thread, which is why+-- this assembles the 'Process' itself instead of using 'Supervisor.supervisor':+-- NQE's @process@ always links. With no children left the supervisor can only be+-- woken through its mailbox, and the mailbox is reachable solely through this+-- handle, so a link would turn a dropped handle into an 'ExceptionInLinkedThread'+-- in the caller at the next major garbage collection. Unlinked, such a supervisor+-- is simply collected. Processor failures still reach the caller, exactly once,+-- through the per-processor links installed when 'propagateFailures' is set. startMaster :: (IOE :> es) => Strategy -> Eff es Master startMaster strategy = liftIO $ do-  -- Create supervisor-  sup <- Supervisor.supervisor strategy+  (inbox, mailbox) <- newMailbox+  supAsync <- async (Supervisor.supervisorProcess strategy inbox)+  let sup = Process supAsync mailbox    metricsMapVar <- newTVarIO Map.empty   let propagate = case strategy of
+ test-gc/Finished.hs view
@@ -0,0 +1,137 @@+{-# LANGUAGE OverloadedStrings #-}++-- This probe owns its process for the same reason as Main.hs: retaining an+-- AppHandle for cleanup would hide the bug. It covers the state Main.hs cannot:+-- an application whose processors have all finished, so the supervisor has no+-- children, while the thread that called runApp keeps running.+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.NQE.Supervisor (Strategy (IgnoreAll), supervisor)+import Control.Exception (displayException, throwIO)+import Control.Monad (forM, replicateM_, unless, void)+import Data.List (isInfixOf)+import Effectful (IOE, liftIO, runEff)+import Shibuya.Adapter (Adapter (..))+import Shibuya.App+  ( AppConfig (..),+    ProcessorId (..),+    QueueProcessor (..),+    SupervisionStrategy (..),+    defaultAppConfig,+    mkBatchProcessor,+    mkProcessor,+    runApp,+    waitApp,+  )+import Shibuya.Batch (BatchConfig (..), ackAll, defaultBatchConfig)+import Shibuya.Core.Ack (AckDecision (..), HaltReason (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested, mkIngested)+import Shibuya.Core.Types (MessageId (..), mkEnvelope)+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))+import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)+import Streamly.Data.Stream qualified as Stream+import System.Exit (die, exitFailure)+import System.Mem (performMajorGC)+import UnliftIO qualified as UIO++type Processors = [(ProcessorId, QueueProcessor '[Tracing, IOE])]++main :: IO ()+main = do+  survived <- forM scenarios $ \(name, strat, processors) -> do+    -- Each scenario gets its own thread: a linked supervisor targets whichever+    -- thread calls runApp, and that is the thread a regression kills.+    outcome <- observe (finishThenKeepRunning strat processors)+    case outcome of+      Nothing -> False <$ putStrLn ("FAIL [" <> name <> "]: the observation did not finish within ten seconds")+      Just (Left err) -> False <$ putStrLn ("FAIL [" <> name <> "]: caller died after its application finished: " <> displayException err)+      Just (Right ()) -> True <$ putStrLn ("PASS [" <> name <> "]: caller survives major collections after its application finished")+  detectable <- control+  unless (and survived && detectable) exitFailure+  where+    observe action = UIO.timeout 10_000_000 $ UIO.withAsync action UIO.waitCatch++    -- The defect itself, rebuilt from NQE alone: a linked supervisor with no+    -- children whose handle is dropped. It MUST kill its caller. If it ever+    -- stops doing so, this build cannot detect the failure class at all and the+    -- PASS lines above prove nothing, so the suite fails rather than pass vacuously.+    control = do+      outcome <- observe $ do+        void (supervisor IgnoreAll)+        keepRunningThroughCollections+      case outcome of+        Just (Left err)+          | "blocked indefinitely" `isInfixOf` displayException err ->+              True <$ putStrLn "PASS [control]: a linked childless supervisor still kills its caller, so the scenarios above are meaningful"+        other -> False <$ putStrLn ("FAIL [control]: a linked childless supervisor no longer kills its caller (" <> maybe "timed out" (either displayException (const "survived")) other <> "); re-establish a reproducer before trusting this suite")++scenarios :: [(String, SupervisionStrategy, Processors)]+scenarios =+  [ ("finite source, IgnoreFailures", IgnoreFailures, [(ProcessorId "finished", mkProcessor (finite 0) ok)]),+    ("finite source, StopAllOnFailure", StopAllOnFailure, [(ProcessorId "finished", mkProcessor (finite 0) ok)]),+    ("failed source, IgnoreFailures", IgnoreFailures, [(ProcessorId "failed", mkProcessor failedSource ok)]),+    ("handler halt on a live idle source, IgnoreFailures", IgnoreFailures, [(ProcessorId "halted", mkProcessor oneThenIdle halt)]),+    ("handler halt on a live idle source, StopAllOnFailure", StopAllOnFailure, [(ProcessorId "halted", mkProcessor oneThenIdle halt)]),+    ( "serial, concurrent and batch processors together, StopAllOnFailure",+      StopAllOnFailure,+      [ (ProcessorId "serial", mkProcessor (finite 5) ok),+        (ProcessorId "concurrent", (mkProcessor (finite 50) ok) {ordering = Unordered, concurrency = Async 4}),+        (ProcessorId "batch", mkBatchProcessor (finite 7) (\_ _ -> pure (ackAll AckOk)) defaultBatchConfig {batchSize = 2, batchTimeout = 0.1})+      ]+    )+  ]+  where+    ok _ = pure AckOk+    halt _ = pure (AckHalt (HaltFatal "halt on purpose"))++-- | Run an application to its end on this thread, let the handle go out of+-- scope, and keep running. Deliberately no stopApp and no retained reference.+finishThenKeepRunning :: SupervisionStrategy -> Processors -> IO ()+finishThenKeepRunning strat processors = do+  runEff $ runTracingNoop $ do+    result <- runApp defaultAppConfig {strategy = strat, inboxSize = 10} processors+    case result of+      Left err -> liftIO $ die ("runApp failed: " <> show err)+      Right app -> waitApp app+  keepRunningThroughCollections++keepRunningThroughCollections :: IO ()+keepRunningThroughCollections = do+  replicateM_ 5 $ do+    threadDelay 100_000+    performMajorGC+  -- Leave time for a linked exception from the last collection to arrive.+  threadDelay 200_000++message :: Int -> Ingested '[Tracing, IOE] String+message n = mkIngested (mkEnvelope (MessageId "gc-regression") ("message-" <> show n)) (AckHandle $ \_ -> pure ())++finite :: Int -> Adapter '[Tracing, IOE] String+finite count =+  Adapter+    { adapterName = "gc-regression:finite",+      source = Stream.fromList (map message [1 .. count]),+      shutdown = pure ()+    }++failedSource :: Adapter '[Tracing, IOE] String+failedSource =+  Adapter+    { adapterName = "gc-regression:failed",+      source = Stream.fromEffect (liftIO (throwIO (userError "the source failed on purpose"))),+      shutdown = pure ()+    }++-- | One message, then a quiet queue: the source stays alive but produces nothing.+oneThenIdle :: Adapter '[Tracing, IOE] String+oneThenIdle =+  Adapter+    { adapterName = "gc-regression:one-then-idle",+      source = Stream.unfoldrM step (0 :: Int),+      shutdown = pure ()+    }+  where+    step 0 = pure (Just (message 0, 1))+    step _ = liftIO (threadDelay 60_000_000) >> pure Nothing
test/Shibuya/App/LifecycleSpec.hs view
@@ -5,7 +5,9 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.NQE.Supervisor (Strategy (..)) import Control.Concurrent.STM (readTVarIO)-import Data.IORef (modifyIORef', newIORef, readIORef)+import Control.Exception (SomeException, mask, try)+import Control.Monad (join, void)+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef) import Data.Map.Strict qualified as Map import Data.Text qualified as Text import Data.Time (UTCTime (..), diffUTCTime, fromGregorian, getCurrentTime)@@ -13,7 +15,7 @@ import Shibuya.Adapter (Adapter (..)) import Shibuya.App   ( AppConfig (..),-    QueueProcessor,+    QueueProcessor (..),     ShutdownConfig (..),     SupervisionStrategy (..),     defaultAppConfig,@@ -197,6 +199,39 @@     countB <- readIORef countBRef     countB `shouldSatisfy` (< 50) +  it "StopAllOnFailure delivers one processor failure to the caller exactly once" $ do+    let failing = mkProcessor (failingAfterAdapter 0 "single-delivery failure") alwaysAckOk+    result <-+      UIO.timeout 15_000_000 $+        UIO.withAsync (countLinkedDeliveries [(ProcessorId "single-delivery", failing)]) UIO.wait+    result `shouldBe` Just 1++  it "StopAllOnFailure delivers one failure exactly once while cancelling busy siblings" $ do+    -- Siblings of every runner shape are cancelled by the supervisor; their+    -- cancellation must not surface as further exceptions in the caller.+    let failing = mkProcessor (failingAfterAdapter 3 "sibling-delivery failure") alwaysAckOk+        serialSibling = mkProcessor infiniteAdapter alwaysAckOk+        asyncSibling = (mkProcessor infiniteAdapter alwaysAckOk) {ordering = Unordered, concurrency = Async 4}+        keyedSibling = (mkProcessor infiniteAdapter alwaysAckOk) {ordering = PartitionedInOrder, concurrency = Ahead 4}+        batchSibling =+          mkBatchProcessor+            infiniteAdapter+            (\_info _msgs -> pure (ackAll AckOk))+            defaultBatchConfig {batchSize = 2, batchTimeout = 0.1}+    result <-+      UIO.timeout 15_000_000 $+        UIO.withAsync+          ( countLinkedDeliveries+              [ (ProcessorId "sibling-delivery", failing),+                (ProcessorId "serial-sibling", serialSibling),+                (ProcessorId "async-sibling", asyncSibling),+                (ProcessorId "keyed-sibling", keyedSibling),+                (ProcessorId "batch-sibling", batchSibling)+              ]+          )+          UIO.wait+    result `shouldBe` Just 1+   it "IgnoreFailures isolates a failing processor" $ do     countBRef <- newIORef (0 :: Int) @@ -242,6 +277,45 @@   case result of     Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) >> error "unreachable"     Right app -> pure app++-- | Run the processors under 'StopAllOnFailure' on the calling thread and count+-- the linked-thread exceptions that thread receives.+--+-- The deliveries are asynchronous exceptions, so one that lands between two+-- handlers escapes both. Everything therefore runs under 'mask' and waits only+-- inside 'restore', wrapped in base's 'try' (UnliftIO's deliberately ignores+-- asynchronous exceptions): none can arrive between iterations, and the count is+-- exact. The handle is retained until the end so garbage collection plays no part.+--+-- The first delivery gets a generous deadline, because a loaded machine may be+-- slow to schedule the failing processor; only the search for a /second/+-- delivery uses a short quiet window. A duplicate follows the first within+-- milliseconds, since both come from the same failure.+countLinkedDeliveries :: [(ProcessorId, QueueProcessor '[Tracing, IOE])] -> IO Int+countLinkedDeliveries processors =+  mask $ \restore -> do+    stopRef <- newIORef (pure ())+    started <-+      try @SomeException $+        restore $+          runEff $+            runTracingNoop $ do+              app <- runAppOrFail StopAllOnFailure 10 processors+              liftIO $+                writeIORef stopRef $+                  runEff $+                    runTracingNoop $+                      void (stopAppGracefully (ShutdownConfig {drainTimeout = 1}) app)+    let arrivesWithin :: Int -> IO Bool+        arrivesWithin micros = either (const True) (const False) <$> try @SomeException (restore (threadDelay micros))+        countExtra :: Int -> IO Int+        countExtra n = do+          another <- arrivesWithin 500_000+          if another then countExtra (n + 1) else pure n+    first <- either (const (pure True)) (const (arrivesWithin 10_000_000)) started+    deliveries <- if first then countExtra 1 else pure 0+    join (readIORef stopRef)+    pure deliveries  processorMetrics ::   (IOE :> es) =>