packages feed

cachix 1.9.0 → 1.9.1

raw patch · 12 files changed

+403/−48 lines, 12 files

Files

CHANGELOG.md view
@@ -5,6 +5,37 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [1.9.1] - 2025-10-20++### Changed++- daemon: implement priority queue to prioritize push tasks over queries+- daemon: double the number of workers to improve throughput++### Fixed++- daemon: fix shutdown order of the narinfo batcher++## [1.9.0] - 2025-10-16++### Added++- daemon: `cachix daemon push` displays upload progress+- daemon: `cachix daemon push` can read store paths from standard input @joshuaspence+- daemon: add `--wait` flag to `cachix daemon push` to block until completion @joshuaspence+- daemon: add `--no-remote-stop` option to prevent remote daemon shutdown+- daemon: narinfo batching and caching for significantly improved push performance++### Changed++- daemon: store paths can be separated by any whitespace character @joshuaspence+- deps: support for GHC 9.10 @roberth++### Fixed++- daemon: symlinks are now properly resolved when pushing paths @joshuaspence+- daemon: `cachix daemon push` returns correct exit codes on failure+ ## [1.8.0] - 2025-07-11  ### Added
cachix.cabal view
@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               cachix-version:            1.9.0+version:            1.9.1 synopsis:   Command-line client for Nix binary cache hosting https://cachix.org @@ -48,7 +48,7 @@   --   ghc-options:       -Wpartial-fields    if impl(ghc >=9.4)-    ghc-options: -Wredundant-strictness-flags -Wforall-identifier+    ghc-options: -Wredundant-strictness-flags    default-language:   GHC2021 @@ -98,6 +98,7 @@     Cachix.Daemon.ShutdownLatch     Cachix.Daemon.SocketStore     Cachix.Daemon.Subscription+    Cachix.Daemon.TaskQueue     Cachix.Daemon.TTLCache     Cachix.Daemon.Types     Cachix.Daemon.Types.Daemon@@ -107,6 +108,7 @@     Cachix.Daemon.Types.PushEvent     Cachix.Daemon.Types.PushManager     Cachix.Daemon.Types.SocketStore+    Cachix.Daemon.Types.TaskQueue     Cachix.Daemon.Worker     Cachix.Deploy.Activate     Cachix.Deploy.ActivateCommand@@ -249,6 +251,7 @@     Daemon.PostBuildHookSpec     Daemon.ProtocolSpec     Daemon.PushManagerSpec+    Daemon.TaskQueueSpec     Daemon.TTLCacheSpec     DeploySpec     InstallationModeSpec
src/Cachix/Client/NixConf.hs view
@@ -117,7 +117,7 @@   render :: a -> Text  instance NixConfOps NixConf where-  readLines predicate (NixConf xs) = foldl f [] xs+  readLines predicate (NixConf xs) = foldl' f [] xs     where       f :: [Text] -> NixConfLine -> [Text]       f prev next = prev <> fromMaybe [] (predicate next)
src/Cachix/Daemon.hs view
@@ -122,8 +122,10 @@   -- Start the narinfo batch processor   PushManager.startBatchProcessor daemonPushManager +  -- Double the number of workers to keep queues filled with jobs+  let numWorkers = Options.numJobs daemonPushOptions * 2   Worker.startWorkers-    (Options.numJobs daemonPushOptions)+    numWorkers     (PushManager.pmTaskQueue daemonPushManager)     (liftIO . PushManager.runPushManager daemonPushManager . PushManager.handleTask)     >>= (liftIO . putMVar daemonWorkerThreads)@@ -277,11 +279,11 @@ shutdownGracefully = do   DaemonEnv {..} <- ask +  -- Stop the narinfo batch processor first (before closing the queue)+  liftIO $ PushManager.stopBatchProcessor daemonPushManager+   -- Stop the push manager and wait for any remaining paths to be uploaded   shutdownPushManager daemonPushManager--  -- Stop the narinfo batch processor-  liftIO $ PushManager.stopBatchProcessor daemonPushManager    -- Stop worker threads   withTakeMVar daemonWorkerThreads Worker.stopWorkers
src/Cachix/Daemon/PushManager.hs view
@@ -53,13 +53,13 @@ import Cachix.Daemon.NarinfoQuery qualified as NarinfoQuery import Cachix.Daemon.Protocol qualified as Protocol import Cachix.Daemon.PushManager.PushJob qualified as PushJob+import Cachix.Daemon.TaskQueue import Cachix.Daemon.Types.Log (Logger) import Cachix.Daemon.Types.PushEvent (PushEvent (..), PushEventMessage (..), newPushRetryStatus) import Cachix.Daemon.Types.PushManager import Cachix.Types.BinaryCache qualified as BinaryCache import Conduit qualified as C import Control.Concurrent.Async qualified as Async-import Control.Concurrent.STM.TBMQueue import Control.Concurrent.STM.TVar import Control.Monad.Catch qualified as E import Control.Monad.IO.Unlift (MonadUnliftIO)@@ -68,6 +68,7 @@ import Data.ByteString qualified as BS import Data.HashMap.Strict qualified as HashMap import Data.IORef+import Data.Sequence qualified as Seq import Data.Set qualified as Set import Data.Text qualified as T import Data.Time (UTCTime, diffUTCTime, getCurrentTime, secondsToNominalDiffTime)@@ -86,18 +87,14 @@   pmPushJobs <- newTVarIO mempty   pmPendingJobCount <- newTVarIO 0   pmStorePathIndex <- newTVarIO mempty-  pmTaskQueue <- newTBMQueueIO 100_000+  pmTaskQueue <- atomically newTaskQueue   pmTaskSemaphore <- QSem.newQSem (numJobs pushOptions)   pmLastEventTimestamp <- newTVarIO =<< getCurrentTime   let pmOnPushEvent id pushEvent = updateTimestampTVar pmLastEventTimestamp >> onPushEvent id pushEvent    -- Create query manager with callback that queues ProcessQueryResponse tasks   let batchCallback requestId response = do-        atomically $ do-          result <- tryWriteTBMQueue pmTaskQueue $ HandleMissingPathsResponse requestId response-          case result of-            Just True -> return ()-            _ -> retry -- Queue is full, keep trying+        atomically $ writeTask pmTaskQueue $ HandleMissingPathsResponse requestId response   pmNarinfoQueryManager <- NarinfoQuery.new batchOptions batchCallback    return $ PushManagerEnv {..}@@ -110,7 +107,7 @@   atomicallyWithTimeout timeoutOptions pmLastEventTimestamp $ do     pendingJobs <- readTVar pmPendingJobCount     check (pendingJobs <= 0)-  atomically $ closeTBMQueue pmTaskQueue+  atomically $ closeTaskQueue pmTaskQueue  -- | Start the batch processor for narinfo queries startBatchProcessor :: (MonadUnliftIO m, Katip.KatipContext m) => PushManagerEnv -> m ()@@ -141,12 +138,12 @@   let queueJob = do         modifyTVar' pmPushJobs $ HashMap.insert (pushId pushJob) pushJob         incrementTVar pmPendingJobCount-        res <- tryWriteTBMQueue pmTaskQueue $ QueryMissingPaths (pushId pushJob)+        res <- tryWriteTask pmTaskQueue $ QueryMissingPaths (pushId pushJob)         case res of           Just True -> return True-          _ -> retry+          _ -> return False -  didQueue <- liftIO $ atomically $ queueJob `orElse` return False+  didQueue <- liftIO $ atomically queueJob    if didQueue     then return $ Just $ pushId pushJob@@ -195,7 +192,7 @@     let pj = HashMap.adjust f pushId jobs     (HashMap.lookup pushId pj, pj) -modifyPushJobs :: [Protocol.PushRequestId] -> (PushJob -> PushJob) -> PushManager ()+modifyPushJobs :: (Foldable f) => f Protocol.PushRequestId -> (PushJob -> PushJob) -> PushManager () modifyPushJobs pushIds f = do   pushJobs <- asks pmPushJobs   liftIO $ atomically $ modifyTVar' pushJobs $ \pushJobs' ->@@ -223,9 +220,9 @@   let addToQueue storePath = do         isDuplicate <- HashMap.member storePath <$> readTVar pmStorePathIndex         unless isDuplicate $-          writeTBMQueue pmTaskQueue (PushPath storePath)+          writeTask pmTaskQueue (PushPath storePath) -        modifyTVar' pmStorePathIndex $ HashMap.insertWith (<>) storePath [pushId]+        modifyTVar' pmStorePathIndex $ HashMap.insertWith (<>) storePath (Seq.singleton pushId)    transactionally $ map addToQueue storePaths @@ -235,11 +232,11 @@   liftIO $ atomically $ do     modifyTVar' storePathIndex $ HashMap.delete storePath -lookupStorePathIndex :: FilePath -> PushManager [Protocol.PushRequestId]+lookupStorePathIndex :: FilePath -> PushManager (Seq.Seq Protocol.PushRequestId) lookupStorePathIndex storePath = do   storePathIndex <- asks pmStorePathIndex   references <- liftIO $ readTVarIO storePathIndex-  return $ fromMaybe [] (HashMap.lookup storePath references)+  return $ fromMaybe Seq.empty (HashMap.lookup storePath references)  checkPushJobCompleted :: Protocol.PushRequestId -> PushManager () checkPushJobCompleted pushId = do@@ -483,7 +480,7 @@    lift $ removePushJob pushId -sendStorePathEvent :: [Protocol.PushRequestId] -> PushEventMessage -> PushManager ()+sendStorePathEvent :: (Foldable f) => f Protocol.PushRequestId -> PushEventMessage -> PushManager () sendStorePathEvent pushIds msg = do   timestamp <- liftIO getCurrentTime   sendPushEvent <- asks pmOnPushEvent
src/Cachix/Daemon/Subscription.hs view
@@ -6,12 +6,13 @@ import Data.Aeson as Aeson (ToJSON) import Data.HashMap.Strict (HashMap) import Data.HashMap.Strict qualified as HashMap+import Data.Sequence qualified as Seq import Network.Socket qualified as Socket import Protolude  data SubscriptionManager k v = SubscriptionManager-  { managerSubscriptions :: TVar (HashMap k [Subscription v]),-    managerGlobalSubscriptions :: TVar [Subscription v],+  { managerSubscriptions :: TVar (HashMap k (Seq.Seq (Subscription v))),+    managerGlobalSubscriptions :: TVar (Seq.Seq (Subscription v)),     managerEvents :: TBMQueue (k, v)   } @@ -24,7 +25,7 @@ newSubscriptionManager :: IO (SubscriptionManager k v) newSubscriptionManager = do   subscriptions <- newTVarIO HashMap.empty-  globalSubscriptions <- newTVarIO []+  globalSubscriptions <- newTVarIO Seq.empty   events <- newTBMQueueIO 10000   pure $ SubscriptionManager subscriptions globalSubscriptions events @@ -38,26 +39,26 @@ subscribeToAll manager subscription =   liftIO $ atomically $ subscribeToAllSTM manager subscription -getSubscriptionsFor :: (Hashable k, MonadIO m) => SubscriptionManager k v -> k -> m [Subscription v]+getSubscriptionsFor :: (Hashable k, MonadIO m) => SubscriptionManager k v -> k -> m (Seq.Seq (Subscription v)) getSubscriptionsFor manager key =   liftIO $ atomically $ getSubscriptionsForSTM manager key  subscribeToSTM :: (Hashable k) => SubscriptionManager k v -> k -> Subscription v -> STM () subscribeToSTM manager key subscription = do   subscriptions <- readTVar $ managerSubscriptions manager-  let subscriptions' = HashMap.insertWith (<>) key [subscription] subscriptions+  let subscriptions' = HashMap.insertWith (<>) key (Seq.singleton subscription) subscriptions   writeTVar (managerSubscriptions manager) subscriptions'  subscribeToAllSTM :: SubscriptionManager k v -> Subscription v -> STM () subscribeToAllSTM manager subscription = do   subscriptions <- readTVar $ managerGlobalSubscriptions manager-  let subscriptions' = subscription : subscriptions+  let subscriptions' = subscription Seq.<| subscriptions   writeTVar (managerGlobalSubscriptions manager) subscriptions' -getSubscriptionsForSTM :: (Hashable k) => SubscriptionManager k v -> k -> STM [Subscription v]+getSubscriptionsForSTM :: (Hashable k) => SubscriptionManager k v -> k -> STM (Seq.Seq (Subscription v)) getSubscriptionsForSTM manager key = do   subscriptions <- readTVar $ managerSubscriptions manager-  pure $ HashMap.lookupDefault [] key subscriptions+  pure $ HashMap.lookupDefault Seq.empty key subscriptions  -- Events @@ -95,7 +96,7 @@   globalSubscriptions <- liftIO $ readTVarIO $ managerGlobalSubscriptions manager   subscriptions <- liftIO $ readTVarIO $ managerSubscriptions manager -  forM_ (concat subscriptions <> globalSubscriptions) $ \case+  forM_ (fold subscriptions <> globalSubscriptions) $ \case     SubSocket queue sock -> do       atomically $ closeTBMQueue queue       Socket.close sock
+ src/Cachix/Daemon/TaskQueue.hs view
@@ -0,0 +1,58 @@+module Cachix.Daemon.TaskQueue+  ( TaskQueue,+    newTaskQueue,+    writeTask,+    tryWriteTask,+    readTask,+    closeTaskQueue,+  )+where++import Cachix.Daemon.Types.TaskQueue (Prioritized (..), TaskQueue (..))+import Control.Concurrent.STM.TVar+import Data.PQueue.Max qualified as MaxQueue+import Protolude++-- | Create a new empty task queue+newTaskQueue :: STM (TaskQueue a)+newTaskQueue = TaskQueue <$> newTVar (MaxQueue.empty, 0, False)++-- | Write an item to the queue (always succeeds unless queue is closed)+writeTask :: (Ord a) => TaskQueue a -> a -> STM ()+writeTask (TaskQueue tvar) task = do+  (queue, seqNum, closed) <- readTVar tvar+  unless closed $ do+    let prioritized = Prioritized task seqNum+        nextSeq = seqNum + 1 -- Word32 wraps around naturally on overflow+    writeTVar tvar (MaxQueue.insert prioritized queue, nextSeq, closed)++-- | Try to write an item to the queue+-- Returns Just True if successful, Just False if closed, Nothing if would block (never happens)+tryWriteTask :: (Ord a) => TaskQueue a -> a -> STM (Maybe Bool)+tryWriteTask (TaskQueue tvar) task = do+  (queue, seqNum, closed) <- readTVar tvar+  if closed+    then return (Just False)+    else do+      let prioritized = Prioritized task seqNum+          nextSeq = seqNum + 1+      writeTVar tvar (MaxQueue.insert prioritized queue, nextSeq, closed)+      return (Just True)++-- | Read an item from the queue+-- Returns Nothing if queue is closed, retries if queue is empty but open+readTask :: (Ord a) => TaskQueue a -> STM (Maybe a)+readTask (TaskQueue tvar) = do+  (queue, seqNum, closed) <- readTVar tvar+  case MaxQueue.maxView queue of+    Nothing ->+      if closed+        then return Nothing+        else retry+    Just (prioritized, queue') -> do+      writeTVar tvar (queue', seqNum, closed)+      return (Just (pTask prioritized))++-- | Close the queue+closeTaskQueue :: TaskQueue a -> STM ()+closeTaskQueue (TaskQueue tvar) = modifyTVar' tvar $ \(queue, seqNum, _) -> (queue, seqNum, True)
src/Cachix/Daemon/Types/PushManager.hs view
@@ -23,23 +23,37 @@ import Cachix.Daemon.Protocol qualified as Protocol import Cachix.Daemon.Types.Log (Logger) import Cachix.Daemon.Types.PushEvent (PushEvent (..))-import Control.Concurrent.STM.TBMQueue+import Cachix.Daemon.Types.TaskQueue (TaskQueue) import Control.Concurrent.STM.TVar import Control.Monad.Catch import Control.Monad.IO.Unlift (MonadUnliftIO) import Data.HashMap.Strict (HashMap)+import Data.Sequence qualified as Seq import Data.Time (UTCTime) import Katip qualified import Protolude +-- | A task to be processed by the PushManager+-- Tasks are prioritized by their Ord instance data Task   = QueryMissingPaths Protocol.PushRequestId   | HandleMissingPathsResponse Protocol.PushRequestId NarinfoQuery.NarinfoResponse   | PushPath FilePath+  deriving stock (Eq) +-- | Ord instance that only compares constructors for priority+-- PushPath > HandleMissingPathsResponse > QueryMissingPaths+instance Ord Task where+  compare = comparing taskPriority+    where+      taskPriority :: Task -> Int+      taskPriority (PushPath _) = 3+      taskPriority (HandleMissingPathsResponse _ _) = 2+      taskPriority (QueryMissingPaths _) = 1+ type PushJobStore = TVar (HashMap Protocol.PushRequestId PushJob) -type StorePathIndex = TVar (HashMap FilePath [Protocol.PushRequestId])+type StorePathIndex = TVar (HashMap FilePath (Seq.Seq Protocol.PushRequestId))  -- TODO: a lot of the logic surrounding deduping, search, and job tracking could be replaced by sqlite. -- sqlite can run in-memory if we don't need persistence.@@ -51,8 +65,8 @@     -- | A mapping of store paths to push requests.     -- Used when deduplicating pushes.     pmStorePathIndex :: StorePathIndex,-    -- | FIFO queue of push tasks.-    pmTaskQueue :: TBMQueue Task,+    -- | Priority queue of push tasks.+    pmTaskQueue :: TaskQueue Task,     -- | A semaphore to control task concurrency.     pmTaskSemaphore :: QSem,     -- | A callback for push events.
+ src/Cachix/Daemon/Types/TaskQueue.hs view
@@ -0,0 +1,37 @@+module Cachix.Daemon.Types.TaskQueue+  ( TaskQueue (..),+    Prioritized (..),+  )+where++import Control.Concurrent.STM.TVar+import Data.Int (Int32)+import Data.PQueue.Max (MaxQueue)+import Protolude++-- | Internal wrapper that adds FIFO ordering within same-priority items+-- Items are ordered first by their task priority, then by insertion order (FIFO)+-- Uses circular sequence comparison to handle wraparound correctly+data Prioritized a = Prioritized+  { pTask :: !a,+    pSequence :: !Int32+  }+  deriving stock (Eq)++instance (Ord a) => Ord (Prioritized a) where+  compare p1 p2 =+    case compare (pTask p1) (pTask p2) of+      EQ -> compareSeq (pSequence p1) (pSequence p2)+      other -> other+    where+      -- Circular sequence comparison: older (smaller) sequences come first+      -- Uses signed arithmetic to handle wraparound (works for sequences up to 2^31 apart)+      compareSeq s1 s2 =+        let diff = s2 - s1+         in compare diff 0++-- | A priority queue that maintains TBMQueue-compatible API+-- The queue is unbounded and prioritizes items by their Ord instance+-- Items with the same priority are returned in FIFO order+-- Sequence counter wraps around after 2^32 inserts+newtype TaskQueue a = TaskQueue (TVar (MaxQueue (Prioritized a), Int32, Bool))
src/Cachix/Daemon/Worker.hs view
@@ -7,25 +7,25 @@   ) where +import Cachix.Daemon.TaskQueue import Control.Concurrent.Async qualified as Async-import Control.Concurrent.STM.TBMQueue import Control.Immortal qualified as Immortal import Control.Monad.IO.Unlift (MonadUnliftIO) import Katip qualified import Protolude  startWorkers ::-  (MonadUnliftIO m, Katip.KatipContext m) =>+  (MonadUnliftIO m, Katip.KatipContext m, Ord a) =>   Int ->-  TBMQueue a ->+  TaskQueue a ->   (a -> m ()) ->   m [Immortal.Thread] startWorkers numWorkers queue f = do   replicateM numWorkers (startWorker queue f)  startWorker ::-  (MonadUnliftIO m, Katip.KatipContext m) =>-  TBMQueue a ->+  (MonadUnliftIO m, Katip.KatipContext m, Ord a) =>+  TaskQueue a ->   (a -> m ()) ->   m Immortal.Thread startWorker queue f = do@@ -49,11 +49,11 @@   Katip.logFM Katip.ErrorS $ Katip.ls (toS $ displayException err :: Text) logWorkerException _ = return () -runWorker :: (MonadIO m) => TBMQueue a -> (a -> m ()) -> m ()+runWorker :: (MonadIO m, Ord a) => TaskQueue a -> (a -> m ()) -> m () runWorker queue f = loop   where     loop = do-      mres <- liftIO $ atomically $ readTBMQueue queue+      mres <- liftIO $ atomically $ readTask queue        case mres of         Nothing -> return ()
test/Daemon/PushManagerSpec.hs view
@@ -17,7 +17,7 @@ import Control.Monad (fail) import Control.Retry (defaultRetryStatus) import Data.Set qualified as Set-import Data.Time (getCurrentTime)+import Data.Time (diffUTCTime, getCurrentTime) import Hercules.CNix qualified as CNix import Protolude import Servant.Auth.Client (Token (Token))@@ -154,15 +154,24 @@        it "shuts down after jobs complete" $ withPushManager $ \pm -> do         let paths = ["foo"]-        let longTimeoutOptions = TimeoutOptions {toTimeout = 10.0, toPollingInterval = 0.1}+        let longTimeoutOptions = TimeoutOptions {toTimeout = 1.0, toPollingInterval = 0.1} -        _ <- runPushManager pm $ do+        Just _ <- runPushManager pm $ do           let request = Protocol.PushRequest {Protocol.storePaths = paths, Protocol.subscribeToUpdates = False}-          addPushJob request+          pushId <- addPushJob request+          let pathSet = Set.fromList paths+              closure = PushJob.ResolvedClosure pathSet pathSet+          for_ pushId $ \pid -> resolvePushJob pid closure+          return pushId +        startTime <- getCurrentTime         concurrently_ (stopPushManager longTimeoutOptions pm) $           runPushManager pm $             for_ paths pushStorePathDone+        endTime <- getCurrentTime++        let elapsed = diffUTCTime endTime startTime+        elapsed `shouldSatisfy` (< 0.5)        it "shuts down on job stall" $         withPushManager $ \pm -> do
+ test/Daemon/TaskQueueSpec.hs view
@@ -0,0 +1,203 @@+module Daemon.TaskQueueSpec where++import Cachix.Daemon.TaskQueue+import Cachix.Daemon.Types.TaskQueue (Prioritized (..))+import Protolude+import Test.Hspec++data PriorityTask = PriorityTask+  { priority :: Int,+    taskId :: Int+  }+  deriving stock (Eq, Show)++instance Ord PriorityTask where+  compare t1 t2 = compare (priority t1) (priority t2)++spec :: Spec+spec = do+  describe "TaskQueue" $ do+    describe "basic operations" $ do+      it "creates an empty queue" $ do+        queue <- atomically newTaskQueue+        result <- atomically $ tryWriteTask queue (PriorityTask 1 1)+        result `shouldBe` Just True++      it "writes and reads a single task" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          writeTask q (PriorityTask 1 1)+          return q+        task <- atomically $ readTask queue+        task `shouldBe` Just (PriorityTask 1 1)++      it "tryWriteTask returns Just True on success" $ do+        queue <- atomically newTaskQueue+        result <- atomically $ tryWriteTask queue (PriorityTask 1 1)+        result `shouldBe` Just True++      it "tryWriteTask returns Just False when closed" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          closeTaskQueue q+          return q+        result <- atomically $ tryWriteTask queue (PriorityTask 1 1)+        result `shouldBe` Just False++    describe "priority ordering" $ do+      it "returns higher priority items first" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          writeTask q (PriorityTask 1 1)+          writeTask q (PriorityTask 3 2)+          writeTask q (PriorityTask 2 3)+          return q++        task1 <- atomically $ readTask queue+        task1 `shouldBe` Just (PriorityTask 3 2)++        task2 <- atomically $ readTask queue+        task2 `shouldBe` Just (PriorityTask 2 3)++        task3 <- atomically $ readTask queue+        task3 `shouldBe` Just (PriorityTask 1 1)++      it "maintains FIFO ordering within same priority" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          writeTask q (PriorityTask 1 1)+          writeTask q (PriorityTask 1 2)+          writeTask q (PriorityTask 1 3)+          return q++        task1 <- atomically $ readTask queue+        task1 `shouldBe` Just (PriorityTask 1 1)++        task2 <- atomically $ readTask queue+        task2 `shouldBe` Just (PriorityTask 1 2)++        task3 <- atomically $ readTask queue+        task3 `shouldBe` Just (PriorityTask 1 3)++      it "handles mixed priorities with FIFO within each priority level" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          writeTask q (PriorityTask 2 1)+          writeTask q (PriorityTask 1 2)+          writeTask q (PriorityTask 2 3)+          writeTask q (PriorityTask 3 4)+          writeTask q (PriorityTask 1 5)+          return q++        task1 <- atomically $ readTask queue+        task1 `shouldBe` Just (PriorityTask 3 4)++        task2 <- atomically $ readTask queue+        task2 `shouldBe` Just (PriorityTask 2 1)++        task3 <- atomically $ readTask queue+        task3 `shouldBe` Just (PriorityTask 2 3)++        task4 <- atomically $ readTask queue+        task4 `shouldBe` Just (PriorityTask 1 2)++        task5 <- atomically $ readTask queue+        task5 `shouldBe` Just (PriorityTask 1 5)++    describe "queue lifecycle" $ do+      it "writeTask is a no-op when queue is closed" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          closeTaskQueue q+          writeTask q (PriorityTask 1 1)+          return q+        task <- atomically $ readTask queue+        task `shouldBe` Nothing++      it "readTask returns Nothing when queue is closed and empty" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          closeTaskQueue q+          return q+        task <- atomically $ readTask (queue :: TaskQueue PriorityTask)+        task `shouldBe` Nothing++      it "can read existing items after queue is closed" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          writeTask q (PriorityTask 1 1)+          writeTask q (PriorityTask 2 2)+          closeTaskQueue q+          return q++        task1 <- atomically $ readTask queue+        task1 `shouldBe` Just (PriorityTask 2 2)++        task2 <- atomically $ readTask queue+        task2 `shouldBe` Just (PriorityTask 1 1)++        task3 <- atomically $ readTask queue+        task3 `shouldBe` Nothing++      it "close is idempotent" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          closeTaskQueue q+          closeTaskQueue q+          closeTaskQueue q+          return q+        task <- atomically $ readTask (queue :: TaskQueue PriorityTask)+        task `shouldBe` Nothing++    describe "edge cases" $ do+      it "handles single item queue correctly" $ do+        queue <- atomically $ do+          q <- newTaskQueue+          writeTask q (PriorityTask 5 1)+          return q+        task <- atomically $ readTask queue+        task `shouldBe` Just (PriorityTask 5 1)++      it "maintains ordering with large number of items" $ do+        let items = [1 .. 100] :: [Int]+        queue <- atomically $ do+          q <- newTaskQueue+          forM_ items $ \i -> writeTask q (PriorityTask (i `mod` 5) i)+          return q++        results <- replicateM 100 (atomically $ readTask queue)+        let justResults = catMaybes results+        length justResults `shouldBe` 100++        let grouped = groupBy (\a b -> priority a == priority b) justResults+        forM_ grouped $ \grp -> do+          let priorities = map priority grp+          priorities `shouldSatisfy` \case+            [] -> True+            (p : rest) -> all (== p) rest++    describe "sequence number overflow" $ do+      it "handles wraparound from maxBound to minBound correctly" $ do+        let maxSeq = maxBound :: Int32+            minSeq = minBound :: Int32+            taskOld = Prioritized (PriorityTask 1 1) maxSeq+            taskNew = Prioritized (PriorityTask 1 2) minSeq+        taskOld `compare` taskNew `shouldBe` GT++      it "maintains FIFO across overflow boundary" $ do+        let nearMax = maxBound - 2 :: Int32+            tasks =+              [ Prioritized (PriorityTask 1 4) (nearMax + 3),+                Prioritized (PriorityTask 1 2) (nearMax + 1),+                Prioritized (PriorityTask 1 1) nearMax,+                Prioritized (PriorityTask 1 3) (nearMax + 2)+              ]+        let sorted = sortBy (comparing Down) tasks+        map (taskId . pTask) sorted `shouldBe` [1, 2, 3, 4]++      it "circular comparison works for sequences far apart" $ do+        let s1 = 100 :: Int32+            s2 = s1 + 1000000+            taskOld = Prioritized (PriorityTask 1 1) s1+            taskNew = Prioritized (PriorityTask 1 2) s2+        taskOld `compare` taskNew `shouldBe` GT