Fix Multi-Nodes tests + Clean unwinding process

.github/workflows/applications.yml

@@ -477,7 +477,7 @@ jobs:
         ghc: ${{ matrix.ghc }}
         use-freeze-file: ${{ matrix.use-freeze-file }}
     - name: Run compaction-tests
-      timeout-minutes: 7
+      timeout-minutes: 10
       run: |
           ulimit -n 10000
           compaction-tests
@@ -505,7 +505,7 @@ jobs:
         ghc: ${{ matrix.ghc }}
         use-freeze-file: ${{ matrix.use-freeze-file }}
     - name: Run multi-node-network-tests
-      timeout-minutes: 10
+      timeout-minutes: 15
       run: |
           ulimit -n 10000
           multi-node-network-tests

src/Chainweb/BlockWeight.hs

@@ -57,7 +57,7 @@ newtype BlockWeight = BlockWeight HashDifficulty
         ( Hashable
         , ToJSON, FromJSON, ToJSONKey, FromJSONKey
         , AdditiveSemigroup, AdditiveAbelianSemigroup
-        , Num
+        , Num, Bounded
         )
 
 instance MerkleHashAlgorithm a => IsMerkleLogEntry a ChainwebHashTag BlockWeight where

src/Chainweb/Chainweb/MinerResources.hs

@@ -105,24 +105,23 @@ withMiningCoordination logger conf cdb inner
         c503 <- newIORef 0
         c403 <- newIORef 0
         l <- newIORef (_coordinationUpdateStreamLimit coordConf)
-        fmap thd . runConcurrently $ (,,)
-            <$> Concurrently (prune t m c503 c403)
-            <*> Concurrently (mapConcurrently_ (primeWork m) cids)
-            <*> Concurrently (inner . Just $ MiningCoordination
-                { _coordLogger = logger
-                , _coordCutDb = cdb
-                , _coordState = t
-                , _coordLimit = _coordinationReqLimit coordConf
-                , _coord503s = c503
-                , _coord403s = c403
-                , _coordConf = coordConf
-                , _coordUpdateStreamCount = l
-                , _coordPrimedWork = m
-                , _coordTargetFork =
-                    if _coordinationTargetForkOverride coordConf
-                    then pred $ max 1 (_versionForkNumber v)
-                    else _versionForkNumber v
-                })
+        withAsync (prune t m c503 c403) (\_ -> do
+            withAsync (mapConcurrently_ (primeWork m) cids) (\_ -> do
+                inner (Just  MiningCoordination
+                    { _coordLogger = logger
+                    , _coordCutDb = cdb
+                    , _coordState = t
+                    , _coordLimit = _coordinationReqLimit coordConf
+                    , _coord503s = c503
+                    , _coord403s = c403
+                    , _coordConf = coordConf
+                    , _coordUpdateStreamCount = l
+                    , _coordPrimedWork = m
+                    , _coordTargetFork =
+                        if _coordinationTargetForkOverride coordConf
+                        then pred $ max 1 (_versionForkNumber v)
+                        else _versionForkNumber v
+                    })))
   where
     coordConf = _miningCoordination conf
     inNodeConf = _miningInNode conf

src/Chainweb/CutDB.hs

@@ -241,15 +241,6 @@ cutHashesTable rdb = Casify $ newTable rdb valueCodec keyCodec ["CutHashes"]
         (runGetS $ (,,) <$> decodeCutHeightBe <*> decodeBlockWeightBe <*> decodeCutId)
     valueCodec = Codec encodeToByteString decodeStrictOrThrow'
 
--- -------------------------------------------------------------------------- --
--- Exceptions
-
-data CutDbStopped = CutDbStopped
-    deriving (Eq, Show, Generic)
-
-instance Exception CutDbStopped where
-  fromException = asyncExceptionFromException
-  toException = asyncExceptionToException
 
 -- -------------------------------------------------------------------------- --
 -- Cut DB
@@ -518,10 +509,12 @@ fastForwardCutDb cutDb = do
 --
 stopCutDb :: CutDb tbl -> IO ()
 stopCutDb db = do
+    pQueueEnd (_cutDbQueue db)
+    void $ waitCatch (_cutDbAsync db)
+
     currentCut <- readTVarIO (_cutDbCut db)
     unless (_cutDbReadOnly db) $
         casInsert (_cutDbCutStore db) (cutToCutHashes Nothing currentCut)
-    cancelWith (_cutDbAsync db) CutDbStopped
 
 -- | Lookup the BlockHeaders for a CutHashes structure. Throws an exception if
 -- the lookup for some BlockHash in the input CutHashes.

src/Data/PQueue.hs

@@ -1,6 +1,8 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE ExistentialQuantification #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE LambdaCase #-}
+
 
 -- |
 -- Module: Data.PQueue
@@ -18,10 +20,14 @@ module Data.PQueue
 , pQueueRemove
 , pQueueIsEmpty
 , pQueueSize
+, pQueueEnd
+, pQueueEndNice
 ) where
 
 import Control.Concurrent.STM
+import Control.Concurrent.Async (AsyncCancelled(..))
 import Control.Monad
+import Control.Exception
 
 import Data.Ord
 import qualified Data.Map as M
@@ -38,11 +44,14 @@ import Numeric.Natural
 -- items in the queue. An item of low priority my starve in the queue if higher
 -- priority items are added at a rate at least as high as items are removed.
 --
+
+data PQueueElement a = PQueueEOF | PQueueData a
+
 data PQueue a =
-    forall p k. (Ord p, Ord k) =>
-    PQueue (TVar (M.Map (Down p, k) a)) (TVar (S.Set k)) (a -> p) (a -> k) (Maybe Natural)
+    forall p k. (Ord p, Bounded p, Ord k, Bounded k) =>
+    PQueue (TVar (M.Map (Down p, k) (PQueueElement a))) (TVar (S.Set k)) (a -> p) (a -> k) (Maybe Natural)
 
-newEmptyPQueue :: (Ord p, Ord k) => (a -> p) -> (a -> k) -> Maybe Natural -> IO (PQueue a)
+newEmptyPQueue :: (Ord p, Bounded p, Ord k, Bounded k)  => (a -> p) -> (a -> k) -> Maybe Natural -> IO (PQueue a)
 newEmptyPQueue getPrio getKey maybeMaxLen = PQueue
     <$> newTVarIO mempty
     <*> newTVarIO mempty
@@ -60,7 +69,7 @@ pQueueInsert (PQueue mv sv getPrio getKey maybeMaxLen) a =
         then return ()
         else do
             let s' = S.insert k s
-            let m' = M.insert (Down $ getPrio a, k) a m
+            let m' = M.insert (Down $ getPrio a, k) (PQueueData a) m
             let fixup (maxlen :: Natural) = if M.size m' > fromIntegral (2 * maxlen)
                     then let (keep, dontkeep) = M.splitAt (fromIntegral maxlen) m'
                     in (foldl' (flip (S.delete . snd)) s' (M.keys dontkeep), keep)
@@ -75,10 +84,21 @@ pQueueIsEmpty (PQueue mv _ _ _ _) = M.null <$!> readTVarIO mv
 pQueueSize :: PQueue a -> IO Natural
 pQueueSize (PQueue mv _ _ _ _) = fromIntegral . M.size <$!> readTVarIO mv
 
+-- End the queue by pushing a top priority EOF
+pQueueEnd :: PQueue a -> IO ()
+pQueueEnd (PQueue mv _ _ _ _) = atomically $ modifyTVar mv $ M.insert (Down maxBound, minBound) PQueueEOF
+
+-- End the queue by pushing a Lowest priority EOF
+pQueueEndNice :: PQueue a -> IO ()
+pQueueEndNice (PQueue mv _ _ _ _) = atomically $ modifyTVar mv $ M.insert (Down minBound, maxBound) PQueueEOF
+
 -- | If the queue is empty it blocks and races for new items
 --
 pQueueRemove :: PQueue a -> IO a
-pQueueRemove (PQueue mv sv _getPrio _getKey _) = atomically run
+pQueueRemove (PQueue mv sv _getPrio _getKey _) =
+  atomically run >>= \case
+    PQueueEOF -> throwIO AsyncCancelled
+    PQueueData a -> return a
   where
     run = do
         m <- readTVar mv

src/P2P/TaskQueue.hs

@@ -91,9 +91,14 @@ newtype TaskId = TaskId T.Text
     deriving (Show, Eq, Ord, Generic)
     deriving newtype (IsString)
 
+instance Bounded TaskId where
+    minBound = TaskId T.empty
+    maxBound = TaskId $ T.singleton maxBound
+
 newtype Priority = Priority Int
     deriving (Show, Eq, Ord, Generic)
     deriving anyclass (NFData, Hashable)
+    deriving newtype (Bounded)
 
 newtype AttemptsCount = AttemptsCount Natural
     deriving (Show, Eq, Ord, Generic)

test/lib/Chainweb/Test/MultiNode.hs

@@ -63,6 +63,7 @@ import Data.ByteString.Base16 qualified as Base16
 import Chainweb.Pact.Backend.PactState.EmbeddedSnapshot (Snapshot(..))
 import Data.Aeson (ToJSON)
 import Data.Foldable
+import Data.Maybe
 import Data.Hashable
 import qualified Data.HashMap.Strict as HM
 import qualified Data.HashSet as HS
@@ -87,7 +88,6 @@ import System.Directory (createDirectoryIfMissing)
 import System.FilePath
 import System.IO.Temp
 import System.LogLevel
-import System.Timeout
 
 import Test.Tasty.HUnit
 
@@ -245,18 +245,30 @@ harvestConsensusState _ _ _ (Replayed _ _) =
     error "harvestConsensusState: doesn't work when replaying, replays don't do consensus"
 harvestConsensusState logger stateVar nid (StartedChainweb cw) = do
     runChainweb cw (\_ -> return ()) `finally` do
-        logFunctionText logger Info "write sample data"
+        logFunctionText logger' Info "Node main threads ended"
+
+        -- At this point, Warp/Servant server is supposed to be closed.
+        -- But Warp doesn't kill existing connections. As such, other nodes
+        -- can continue to push new Cuts, despite we would like to freeze a final Cut.
+
+        -- A workaround is to early stop the CutDB, and not wait for the node to do it
+        -- naturally during unwinding.
+        stopCutDb (cw ^. chainwebCutResources . cutsCutDb)
+
+        logFunctionText logger' Info "write sample data"
         modifyMVar_ stateVar $
             sampleConsensusState
                 nid
                 (view (chainwebCutResources . cutsCutDb . cutDbWebBlockHeaderDb) cw)
                 (view (chainwebCutResources . cutsCutDb) cw)
-        logFunctionText logger Info "shutdown node"
+        logFunctionText logger' Info "shutdown node"
+    where
+        logger' = addLabel ("node", toText nid) logger
 
 multiNode
     :: LogLevel
     -> (T.Text -> IO ())
-    -> MVar PeerInfo
+    -> MVar (Maybe PeerInfo)
     -> ChainwebConfiguration
     -> RocksDb
     -> FilePath
@@ -269,9 +281,9 @@ multiNode loglevel write bootstrapPeerInfoVar conf rdb pactDbDir nid inner = do
             withChainweb conf logger namespacedNodeRocksDb (pactDbDir </> show nid) backupTmpDir False $ \cw -> do
                 case cw of
                     StartedChainweb cw' ->
-                        when (nid == bootstrapNodeId) $ putMVar bootstrapPeerInfoVar
+                        when (nid == bootstrapNodeId) $ putMVar bootstrapPeerInfoVar $ Just
                             $ view (chainwebPeer . peerResPeer . peerInfo) cw'
-                    Replayed _ _ -> return ()
+                    Replayed _ _ -> when (nid == bootstrapNodeId) $ putMVar bootstrapPeerInfoVar Nothing
                 inner nid cw
   where
     logger :: GenericLogger
@@ -313,7 +325,7 @@ runNodes loglevel write v confBuilders rdb pactDbDir inner = do
             | i == 0 ->
                 return $ multiBootstrapConfig baseConf
             | otherwise ->
-                setBootstrapPeerInfo <$> readMVar bootstrapPortVar <*> pure baseConf
+                maybe baseConf (`setBootstrapPeerInfo` baseConf) <$> readMVar bootstrapPortVar
 
         multiNode loglevel write bootstrapPortVar (confBuilder conf) rdb pactDbDir (NodeId i) inner
 
@@ -330,9 +342,11 @@ runNodesForSeconds
     -> FilePath
     -> (forall logger. NodeId -> StartedChainweb logger -> IO ())
     -> IO ()
-runNodesForSeconds loglevel write v confBuilders (Seconds seconds) rdb pactDbDir inner = do
-    void $ timeout (int seconds * 1_000_000)
-        $ runNodes loglevel write v confBuilders rdb pactDbDir inner
+runNodesForSeconds loglevel write v confBuilders (Seconds seconds) rdb pactDbDir inner =
+    runNodes loglevel write v confBuilders rdb pactDbDir innerWithTimeout
+    where
+        innerWithTimeout:: NodeId -> StartedChainweb a -> IO()
+        innerWithTimeout nid cw = void $ race (inner nid cw) $ threadDelay (int seconds * 1_000_000)
 
 -- | Ensure that we can compact a live node(s).
 --
@@ -617,19 +631,18 @@ replayTest loglevel v n rdb pactDbDir step = do
         tastylog $ "phase 3... replaying"
         let replayInitialHeight = 5
         firstReplayCompleteRef <- newIORef False
-        runNodesForSeconds loglevel logFun v
+        runNodes loglevel logFun v
             (replicate n
                 $ multiConfig n
                     & mapped . configCuts . cutInitialBlockHeightLimit
                         .~ Just replayInitialHeight
                     & mapped . configOnlySyncPact .~ True)
-            (Seconds 20) rdb pactDbDir $ \nid cw -> case cw of
+            rdb pactDbDir $ \nid cw -> case cw of
                 Replayed l (Just u) -> do
                     writeIORef firstReplayCompleteRef True
                     _ <- flip HM.traverseWithKey (_cutMap l) $ \cid bh ->
                         assertEqual ("lower chain " <> sshow cid) replayInitialHeight (view blockHeight bh)
-                    -- TODO: this is flaky, presumably because a node's cutdb
-                    -- is not being cancelled synchronously enough
+
                     assertEqual "upper cut" (_stateCutMap state2 HM.! nid) u
                     _ <- flip HM.traverseWithKey (_cutMap u) $ \cid bh ->
                         assertGe ("upper chain " <> sshow cid) (Actual $ view blockHeight bh) (Expected replayInitialHeight)
@@ -640,7 +653,7 @@ replayTest loglevel v n rdb pactDbDir step = do
         let fastForwardHeight = 10
         tastylog $ "phase 4... replaying with fast-forward limit"
         secondReplayCompleteRef <- newIORef False
-        runNodesForSeconds loglevel logFun v
+        runNodes loglevel logFun v
             (replicate n
                 $ multiConfig n
                     & mapped . configCuts . cutInitialBlockHeightLimit
@@ -649,7 +662,7 @@ replayTest loglevel v n rdb pactDbDir step = do
                         .~ Just fastForwardHeight
                     & mapped . configOnlySyncPact .~ True
                 )
-            (Seconds 20) rdb pactDbDir $ \_ cw -> case cw of
+            rdb pactDbDir $ \_ cw -> case cw of
                 Replayed l (Just u) -> do
                     writeIORef secondReplayCompleteRef True
                     _ <- flip HM.traverseWithKey (_cutMap l) $ \cid bh ->