block-forging: async exception safety

When the block forger thread adds a new block, the adding thread might
be killed by an async exception.  If that happens, the block forger will
get 'Nothing' when `blockProcessed` returns, and it can exit.

ouroboros-consensus/src/Ouroboros/Consensus/Node/Tracers.hs

@@ -324,6 +324,10 @@ data TraceForgeEvent blk
     -- this warrants a warning.
   | TraceDidntAdoptBlock SlotNo blk
 
+    -- | We did not adopt the block we produced, because the adoption thread
+    -- died.  Most likely because of an async exception.
+  | TraceAdoptionThreadDied SlotNo blk
+
     -- | We forged a block that is invalid according to the ledger in the
     -- ChainDB. This means there is an inconsistency between the mempool
     -- validation and the ledger validation. This is a serious error!

ouroboros-consensus/src/Ouroboros/Consensus/NodeKernel.hs

@@ -425,28 +425,31 @@ forkBlockForging IS{..} blockForging =
         uninterruptibleMask_ $ do
           result <- lift $ ChainDB.addBlockAsync chainDB noPunish newBlock
           -- Block until we have processed the block
-          curTip <- lift $ atomically $ ChainDB.blockProcessed result
-
-          -- Check whether we adopted our block
-          when (curTip /= blockPoint newBlock) $ do
-            isInvalid <- lift $ atomically $
-              ($ blockHash newBlock) . forgetFingerprint <$>
-              ChainDB.getIsInvalidBlock chainDB
-            case isInvalid of
-              Nothing ->
-                trace $ TraceDidntAdoptBlock currentSlot newBlock
-              Just reason -> do
-                trace $ TraceForgedInvalidBlock currentSlot newBlock reason
-                -- We just produced a block that is invalid according to the
-                -- ledger in the ChainDB, while the mempool said it is valid.
-                -- There is an inconsistency between the two!
-                --
-                -- Remove all the transactions in that block, otherwise we'll
-                -- run the risk of forging the same invalid block again. This
-                -- means that we'll throw away some good transactions in the
-                -- process.
-                lift $ removeTxs mempool (map (txId . txForgetValidated) txs)
-            exitEarly
+          mbCurTip <- lift $ atomically $ ChainDB.blockProcessed result
+          case mbCurTip of
+            Nothing -> trace (TraceAdoptionThreadDied currentSlot newBlock)
+                    >> exitEarly
+            Just curTip -> do
+              -- Check whether we adopted our block
+              when (curTip /= blockPoint newBlock) $ do
+                isInvalid <- lift $ atomically $
+                  ($ blockHash newBlock) . forgetFingerprint <$>
+                  ChainDB.getIsInvalidBlock chainDB
+                case isInvalid of
+                  Nothing ->
+                    trace $ TraceDidntAdoptBlock currentSlot newBlock
+                  Just reason -> do
+                    trace $ TraceForgedInvalidBlock currentSlot newBlock reason
+                    -- We just produced a block that is invalid according to the
+                    -- ledger in the ChainDB, while the mempool said it is valid.
+                    -- There is an inconsistency between the two!
+                    --
+                    -- Remove all the transactions in that block, otherwise we'll
+                    -- run the risk of forging the same invalid block again. This
+                    -- means that we'll throw away some good transactions in the
+                    -- process.
+                    lift $ removeTxs mempool (map (txId . txForgetValidated) txs)
+                exitEarly
 
           -- We successfully produced /and/ adopted a block
           --

ouroboros-consensus/src/Ouroboros/Consensus/Storage/ChainDB/API.hs

@@ -410,7 +410,7 @@ data AddBlockPromise m blk = AddBlockPromise
       -- NOTE: Even when the result is 'False', 'getIsFetched' might still
       -- return 'True', e.g., the block was older than @k@, but it has been
       -- downloaded and stored on disk before.
-    , blockProcessed     :: STM m (Point blk)
+    , blockProcessed     :: STM m (Maybe (Point blk))
       -- ^ Use this 'STM' transaction to wait until the block has been
       -- processed: the block has been written to disk and chain selection has
       -- been performed for the block, /unless/ the block is from the future.
@@ -420,6 +420,8 @@ data AddBlockPromise m blk = AddBlockPromise
       -- wasn't adopted. We might have adopted a longer chain of which the
       -- added block is a part, but not the tip.
       --
+      -- It returns 'Nothing' if the thread adding the block died.
+      --
       -- NOTE: When the block is from the future, chain selection for the
       -- block won't be performed until the block is no longer in the future,
       -- which might take some time. For that reason, this transaction will
@@ -437,7 +439,7 @@ addBlockWaitWrittenToDisk chainDB punish blk = do
 
 -- | Add a block synchronously: wait until the block has been processed (see
 -- 'blockProcessed'). The new tip of the ChainDB is returned.
-addBlock :: IOLike m => ChainDB m blk -> InvalidBlockPunishment m -> blk -> m (Point blk)
+addBlock :: IOLike m => ChainDB m blk -> InvalidBlockPunishment m -> blk -> m (Maybe (Point blk))
 addBlock chainDB punish blk = do
     promise <- addBlockAsync chainDB punish blk
     atomically $ blockProcessed promise

ouroboros-consensus/src/Ouroboros/Consensus/Storage/ChainDB/Impl/Background.hs

@@ -535,7 +535,13 @@ addBlockRunner
 addBlockRunner cdb@CDB{..} = forever $ do
     let trace = traceWith cdbTracer . TraceAddBlockEvent
     trace $ PoppedBlockFromQueue RisingEdge
-    blkToAdd <- getBlockToAdd cdbBlocksToAdd
-    trace $ PoppedBlockFromQueue $ FallingEdgeWith $
-      blockRealPoint $ blockToAdd blkToAdd
-    addBlockSync cdb blkToAdd
+    -- if the `addBlockSync` does not complete because it was killed by an async
+    -- exception (or it errored), notify the blocked thread
+    bracketOnError (getBlockToAdd cdbBlocksToAdd)
+                   (\blkToAdd -> atomically $ do
+                     _ <- tryPutTMVar (varBlockProcessed blkToAdd) Nothing
+                     closeBlocksToAdd cdbBlocksToAdd)
+                   (\blkToAdd -> do
+                     trace $ PoppedBlockFromQueue $ FallingEdgeWith $
+                             blockRealPoint $ blockToAdd blkToAdd
+                     addBlockSync cdb blkToAdd)

ouroboros-consensus/src/Ouroboros/Consensus/Storage/ChainDB/Impl/ChainSel.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE BangPatterns         #-}
 {-# LANGUAGE FlexibleContexts     #-}
 {-# LANGUAGE GADTs                #-}
 {-# LANGUAGE LambdaCase           #-}
@@ -342,8 +343,8 @@ addBlockSync cdb@CDB {..} BlockToAdd { blockToAdd = b, .. } = do
     -- | Fill in the 'TMVar' for the 'varBlockProcessed' of the block's
     -- 'AddBlockPromise' with the given tip.
     deliverProcessed :: Point blk -> m ()
-    deliverProcessed tip = atomically $
-        putTMVar varBlockProcessed tip
+    deliverProcessed !tip = atomically $
+        putTMVar varBlockProcessed (Just tip)
 
 -- | Return 'True' when the given header should be ignored when adding it
 -- because it is too old, i.e., we wouldn't be able to switch to a chain

ouroboros-consensus/src/Ouroboros/Consensus/Storage/ChainDB/Impl/Types.hs

@@ -46,6 +46,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types (
   , addBlockToAdd
   , getBlockToAdd
   , newBlocksToAdd
+  , closeBlocksToAdd
     -- * Trace types
   , NewTipInfo (..)
   , TraceAddBlockEvent (..)
@@ -61,6 +62,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types (
   ) where
 
 import           Control.Tracer
+import           Data.Foldable (traverse_)
 import           Data.Map.Strict (Map)
 import           Data.Maybe.Strict (StrictMaybe (..))
 import           Data.Typeable
@@ -450,7 +452,7 @@ data BlockToAdd m blk = BlockToAdd
   , blockToAdd            :: !blk
   , varBlockWrittenToDisk :: !(StrictTMVar m Bool)
     -- ^ Used for the 'blockWrittenToDisk' field of 'AddBlockPromise'.
-  , varBlockProcessed     :: !(StrictTMVar m (Point blk))
+  , varBlockProcessed     :: !(StrictTMVar m (Maybe (Point blk)))
     -- ^ Used for the 'blockProcessed' field of 'AddBlockPromise'.
   }
 
@@ -492,6 +494,14 @@ addBlockToAdd tracer (BlocksToAdd queue) punish blk = do
 getBlockToAdd :: IOLike m => BlocksToAdd m blk -> m (BlockToAdd m blk)
 getBlockToAdd (BlocksToAdd queue) = atomically $ readTBQueue queue
 
+-- | Flush the 'BlocksToAdd' queue and notify the waiting threads.
+--
+closeBlocksToAdd :: IOLike m => BlocksToAdd m blk -> STM m ()
+closeBlocksToAdd (BlocksToAdd queue) = do
+  as <- flushTBQueue queue
+  traverse_ (\a -> tryPutTMVar (varBlockProcessed a) Nothing) as
+
+
 {-------------------------------------------------------------------------------
   Trace types
 -------------------------------------------------------------------------------}