Use broadcast TChan for campaign events

Replace the campaign event queue with an STM broadcast TChan so subscribers
keep the existing events-after-subscribe behavior without the Chan root
cursor retention issue. Producers now write through STM, and campaign/UI/SSE
listeners subscribe with dupTChan/readTChan.

Avoid retaining full VM snapshots in queued TestFalsified/TestOptimized
events by publishing VM-free EchidnaTest copies. The canonical testRefs
state still keeps the VM so shrinking, reports, traces, and final JSON
continue to work.

Author: elopez

lib/Echidna.hs

@@ -1,6 +1,5 @@
 module Echidna where
 
-import Control.Concurrent (newChan)
 import Control.Monad.Catch (MonadThrow(..))
 import Control.Monad.IO.Class (liftIO)
 import Data.IORef (newIORef)
@@ -14,6 +13,7 @@ import Data.Text qualified as T
 import System.Console.ANSI (hNowSupportsANSI)
 import System.FilePath ((</>))
 import System.IO (stderr, stdout, hPutStrLn)
+import UnliftIO.STM (newBroadcastTChanIO)
 
 import EVM (cheatCode)
 import EVM.ABI (AbiValue(AbiAddress))
@@ -124,7 +124,7 @@ mkEnv :: EConfig -> BuildOutput -> [EchidnaTest] -> World -> Maybe SlitherInfo -
 mkEnv cfg buildOutput tests world slitherInfo = do
   codehashMap <- newIORef mempty
   chainId <- Onchain.fetchChainIdFrom cfg.rpcUrl
-  eventQueue <- newChan
+  eventQueue <- newBroadcastTChanIO
   coverageRefInit <- newIORef mempty
   coverageRefRuntime <- newIORef mempty
   corpusRef <- newIORef mempty

lib/Echidna/Campaign.hs

@@ -27,6 +27,7 @@ import Data.Text (Text, unpack)
 import Data.Time (LocalTime)
 import Data.Vector qualified as V
 import System.Random (mkStdGen)
+import UnliftIO.STM (TChan, atomically, dupTChan, readTChan)
 
 import EVM (cheatCode)
 import EVM.ABI (getAbi, AbiType(AbiAddressType, AbiTupleType), AbiValue(AbiAddress, AbiTuple), abiValueType)
@@ -120,7 +121,7 @@ runSymWorker callback vm dict workerId _ name = do
   cfg <- asks (.cfg)
   let nworkers = getNFuzzWorkers cfg.campaignConf -- getNFuzzWorkers, NOT getNWorkers
   eventQueue <- asks (.eventQueue)
-  chan <- liftIO $ dupChan eventQueue
+  chan <- liftIO $ atomically $ dupTChan eventQueue
 
   flip runStateT initialState $
     flip evalRandT (mkStdGen effectiveSeed) $ do -- unused but needed for callseq
@@ -644,7 +645,7 @@ updateOpenTest vm reproducer test = do
                            , result
                            , workerId
                            }
-          pushWorkerEvent (TestFalsified test')
+          pushWorkerEvent (TestFalsified (test' { Test.vm = Nothing }))
           pure $ Just test'
 
         IntValue value' | value' > value -> do
@@ -653,7 +654,7 @@ updateOpenTest vm reproducer test = do
                            , vm = Just vm
                            , result
                            }
-          pushWorkerEvent (TestOptimized test')
+          pushWorkerEvent (TestOptimized (test' { Test.vm = Nothing }))
           pure $ Just test'
           where
           value =
@@ -686,7 +687,7 @@ spawnListener handler = do
   cfg <- asks (.cfg)
   let nworkers = getNWorkers cfg.campaignConf
   eventQueue <- asks (.eventQueue)
-  chan <- liftIO $ dupChan eventQueue
+  chan <- liftIO $ atomically $ dupTChan eventQueue
   stopVar <- liftIO newEmptyMVar
   liftIO $ void $ forkFinally (listenerLoop handler chan nworkers) (const $ putMVar stopVar ())
   pure stopVar
@@ -697,14 +698,14 @@ listenerLoop
   :: (MonadIO m)
   => ((LocalTime, CampaignEvent) -> m ())
   -- ^ a function that handles the events
-  -> Chan (LocalTime, CampaignEvent)
+  -> TChan (LocalTime, CampaignEvent)
   -- ^ event channel
   -> Int
   -- ^ number of workers which have to stop before loop exits
   -> m ()
 listenerLoop handler chan !workersAlive =
   when (workersAlive > 0) $ do
-    event <- liftIO $ readChan chan
+    event <- liftIO $ atomically $ readTChan chan
     handler event
     case event of
       (_, WorkerEvent _ _ (WorkerStopped _)) -> listenerLoop handler chan (workersAlive - 1)

lib/Echidna/Server.hs

@@ -1,6 +1,6 @@
 module Echidna.Server where
 
-import Control.Concurrent
+import Control.Concurrent (MVar, forkIO, putMVar)
 import Control.Monad (forever, when, void)
 import Data.Aeson
 import Data.Binary.Builder (fromLazyByteString)
@@ -40,11 +40,11 @@ instance ToJSON SSE where
 runSSEServer :: MVar () -> Env -> Word16 -> Int -> IO ()
 runSSEServer serverStopVar env port nworkers = do
   aliveRef <- newIORef nworkers
-  sseChan <- dupChan env.eventQueue
+  sseChan <- atomically $ dupTChan env.eventQueue
   broadcastChan <- newBroadcastTChanIO
 
   void . forkIO . forever $ do
-    event@(_, campaignEvent) <- readChan sseChan
+    event@(_, campaignEvent) <- atomically $ readTChan sseChan
     let eventName = \case
           WorkerEvent _ _ workerEvent ->
             case workerEvent of

lib/Echidna/Types/Config.hs

@@ -1,12 +1,12 @@
 module Echidna.Types.Config where
 
-import Control.Concurrent (Chan)
 import Data.Aeson.Key (Key)
 import Data.IORef (IORef)
 import Data.Set (Set)
 import Data.Text (Text)
 import Data.Time (LocalTime)
 import Data.Word (Word64)
+import UnliftIO.STM (TChan)
 
 import EVM.Dapp (DappInfo)
 import EVM.Fetch qualified as Fetch
@@ -78,7 +78,7 @@ data Env = Env
 
   -- | Shared between all workers. Events are fairly rare so contention is
   -- minimal.
-  , eventQueue :: Chan (LocalTime, CampaignEvent)
+  , eventQueue :: TChan (LocalTime, CampaignEvent)
 
   , testRefs :: [IORef EchidnaTest]
   , coverageRefInit :: IORef CoverageMap

lib/Echidna/UI.hs

@@ -28,6 +28,7 @@ import System.Signal
 import UnliftIO
   ( MonadUnliftIO, IORef, newIORef, readIORef, hFlush, stdout , writeIORef, timeout)
 import UnliftIO.Concurrent hiding (killThread, threadDelay)
+import UnliftIO.STM (atomically, writeTChan)
 
 import EVM.Fetch qualified
 import EVM.Types (Addr, Contract, VM, VMType(Concrete), W256)
@@ -108,7 +109,7 @@ ui vm dict initialCorpus cliSelectedContract = do
       let forwardEvent = void . writeBChanNonBlocking uiChannel . EventReceived
 
       -- Attach the log/event forwarder before workers start so early worker
-      -- events (like startup logs) are not lost by dupChan.
+      -- events (like startup logs) are not lost by dupTChan.
       uiEventsForwarderStopVar <- spawnListener forwardEvent
       workers <- spawnWorkers
 
@@ -181,7 +182,7 @@ ui vm dict initialCorpus cliSelectedContract = do
 
       let forwardEvent ev = putStrLn =<< runReaderT (ppLogLine vm ev) env
       -- Attach the log/event forwarder before workers start so early worker
-      -- events (like startup logs) are not lost by dupChan.
+      -- events (like startup logs) are not lost by dupTChan.
       uiEventsForwarderStopVar <- spawnListener forwardEvent
       workers <- spawnWorkers
 
@@ -271,7 +272,8 @@ ui vm dict initialCorpus cliSelectedContract = do
         _ -> pure ()
 
       time <- liftIO getTimestamp
-      writeChan env.eventQueue (time, WorkerEvent workerId workerType (WorkerStopped stopReason))
+      liftIO $ atomically $
+        writeTChan env.eventQueue (time, WorkerEvent workerId workerType (WorkerStopped stopReason))
 
     pure (threadId, stateRef)
 

lib/Echidna/Worker.hs

@@ -1,10 +1,10 @@
 module Echidna.Worker where
 
-import Control.Concurrent
 import Control.Monad.Reader (MonadReader, MonadIO, liftIO, ask)
 import Control.Monad.State.Strict(MonadState(..), gets)
 import Data.Aeson
 import Data.Text (unpack)
+import UnliftIO.STM (atomically, writeTChan)
 
 import Echidna.ABI (encodeSig)
 import Echidna.Types.Campaign
@@ -47,7 +47,7 @@ pushWorkerEvent event = do
 pushCampaignEvent :: Env -> CampaignEvent -> IO ()
 pushCampaignEvent env event = do
   time <- liftIO getTimestamp
-  writeChan env.eventQueue (time, event)
+  atomically $ writeTChan env.eventQueue (time, event)
 
 ppCampaignEvent :: CampaignEvent -> String
 ppCampaignEvent = \case