Deliver status information using server-sent events (#1131)

* POC of delivering status information using server-sent events

* deliver only events instead of status lines

* Stream JSON events

* Hide event server behind config

---------

Co-authored-by: Artur Cygan <[email protected]>

lib/Echidna/Campaign.hs

@@ -73,7 +73,7 @@ replayCorpus vm txSeqs =
 -- optional dictionary to generate calls with. Return the 'Campaign' state once
 -- we can't solve or shrink anything.
 runWorker
-  :: (MonadIO m, MonadThrow m, MonadRandom m, MonadReader Env m)
+  :: (MonadIO m, MonadThrow m, MonadReader Env m)
   => StateT WorkerState m ()
   -- ^ Callback to run after each state update (for instrumentation)
   -> VM RealWorld -- ^ Initial VM state

lib/Echidna/Config.hs

@@ -96,6 +96,7 @@ instance FromJSON EConfigWithUsage where
         <*> v ..:? "mutConsts" ..!= defaultMutationConsts
         <*> v ..:? "coverageFormats" ..!= [Txt,Html,Lcov]
         <*> v ..:? "workers"
+        <*> v ..:? "server"
 
       solConfParser = SolConf
         <$> v ..:? "contractAddr"    ..!= defaultContractAddr

lib/Echidna/Server.hs

@@ -0,0 +1,54 @@
+module Echidna.Server where
+
+import Control.Concurrent
+import Control.Monad (when, void)
+import Data.Aeson
+import Data.Binary.Builder (fromLazyByteString)
+import Data.IORef
+import Data.Time (LocalTime)
+import Data.Word (Word16)
+import Network.Wai.EventSource (ServerEvent(..), eventSourceAppIO)
+import Network.Wai.Handler.Warp (run)
+
+import Echidna.Types.Campaign (CampaignEvent (..))
+import Echidna.Types.Config (Env(..))
+
+newtype SSE = SSE (Int, LocalTime, CampaignEvent)
+
+instance ToJSON SSE where
+  toJSON (SSE (workerId, time, event)) =
+    object [ "worker" .= workerId
+           , "timestamp" .= time
+           , "data" .= event
+           ]
+
+runSSEServer :: MVar () -> Env -> Word16 -> Int -> IO ()
+runSSEServer serverStopVar env port nworkers = do
+  aliveRef <- newIORef nworkers
+  sseChan <- dupChan env.eventQueue
+
+  let sseListener = do
+        aliveNow <- readIORef aliveRef
+        if aliveNow == 0 then
+          pure CloseEvent
+        else do
+          event@(_, _, campaignEvent) <- readChan sseChan
+          let eventName = \case
+                TestFalsified _ -> "test_falsified"
+                TestOptimized _ -> "test_optimized"
+                NewCoverage {} -> "new_coverage"
+                TxSequenceReplayed _ _ -> "tx_sequence_replayed"
+                WorkerStopped _ -> "worker_stopped"
+          case campaignEvent of
+            WorkerStopped _ -> do
+              aliveAfter <- atomicModifyIORef' aliveRef (\n -> (n-1, n-1))
+              when (aliveAfter == 0) $ putMVar serverStopVar ()
+            _ -> pure ()
+          pure $ ServerEvent
+            { eventName = Just (eventName campaignEvent)
+            , eventId = Nothing
+            , eventData = [ fromLazyByteString $ encode (SSE event) ]
+            }
+
+  void . forkIO $ do
+    run (fromIntegral port) $ eventSourceAppIO sseListener

lib/Echidna/Types/Campaign.hs

@@ -1,9 +1,10 @@
 module Echidna.Types.Campaign where
 
+import Data.Aeson
 import Data.Map (Map)
 import Data.Text (Text)
 import Data.Text qualified as T
-import Data.Word (Word8)
+import Data.Word (Word8, Word16)
 
 import Echidna.ABI (GenDict, emptyDict, encodeSig)
 import Echidna.Output.Source (CoverageFileType)
@@ -39,6 +40,9 @@ data CampaignConf = CampaignConf
   , coverageFormats :: [CoverageFileType]
     -- ^ List of file formats to save coverage reports
   , workers         :: Maybe Word8
+    -- ^ Number of fuzzing workers
+  , serverPort      :: Maybe Word16
+    -- ^ Server-Sent Events HTTP port number, if missing server is not ran
   }
 
 data CampaignEvent
@@ -51,6 +55,15 @@ data CampaignEvent
   -- this one
   deriving Show
 
+instance ToJSON CampaignEvent where
+  toJSON = \case
+    TestFalsified test -> toJSON test
+    TestOptimized test -> toJSON test
+    NewCoverage coverage numContracts corpusSize ->
+      object [ "coverage" .= coverage, "contracts" .= numContracts, "corpus_size" .= corpusSize]
+    TxSequenceReplayed current total -> object [ "current" .= current, "total" .= total ]
+    WorkerStopped reason -> object [ "reason" .= show reason ]
+
 data WorkerStopReason
   = TestLimitReached
   | TimeLimitReached

lib/Echidna/Types/Test.hs

@@ -1,7 +1,10 @@
+{-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE RecordWildCards #-}
+
 module Echidna.Types.Test where
 
 import Control.Monad.ST (RealWorld)
-import Data.Aeson (ToJSON(..), object)
+import Data.Aeson
 import Data.DoubleWord (Int256)
 import Data.Maybe (maybeToList)
 import Data.Text (Text)
@@ -12,6 +15,7 @@ import EVM.Types (Addr, VM)
 import Echidna.Types (ExecException)
 import Echidna.Types.Signature (SolSignature)
 import Echidna.Types.Tx (Tx, TxResult)
+import GHC.Generics (Generic)
 
 -- | Test mode is parsed from a string
 type TestMode = String
@@ -40,7 +44,7 @@ data TestValue
   = BoolValue Bool
   | IntValue Int256
   | NoValue
-  deriving (Eq, Ord)
+  deriving (Eq, Ord, Generic, ToJSON)
 
 instance Show TestValue where
   show (BoolValue x) = show x
@@ -70,6 +74,19 @@ instance Show TestType where
     CallTest t _         -> show t
     Exploration          -> "Exploration"
 
+instance ToJSON TestType where
+  toJSON = \case
+    PropertyTest name addr ->
+      object [ "type" .= ("property_test" :: String), "name" .= name, "addr" .= addr ]
+    OptimizationTest name addr ->
+      object [ "type" .= ("optimization_test" :: String), "name" .= name, "addr" .= addr ]
+    AssertionTest _ sig addr ->
+      object [ "type" .= ("assertion_test" :: String), "signature" .= sig, "addr" .= addr ]
+    CallTest name _ ->
+      object [ "type" .= ("call_test" :: String), "name" .= name ]
+    Exploration ->
+      object [ "type" .= ("exploration_test" :: String) ]
+
 instance Eq TestState where
   Open    == Open    = True
   Large i == Large j = i == j
@@ -87,6 +104,15 @@ data EchidnaTest = EchidnaTest
   , vm         :: Maybe (VM RealWorld)
   } deriving (Show)
 
+instance ToJSON EchidnaTest where
+  toJSON EchidnaTest{..} = object
+    [ "state" .= state
+    , "type" .= testType
+    , "value" .= value
+    , "reproducer" .= reproducer
+    , "result" .= result
+    ]
+
 isOptimizationTest :: EchidnaTest -> Bool
 isOptimizationTest EchidnaTest{testType = OptimizationTest _ _} = True
 isOptimizationTest _ = False

lib/Echidna/UI.hs

@@ -18,31 +18,30 @@ import Control.Concurrent (killThread, threadDelay)
 import Control.Exception (AsyncException)
 import Control.Monad
 import Control.Monad.Catch
-import Control.Monad.Random.Strict (MonadRandom)
 import Control.Monad.Reader
 import Control.Monad.State.Strict hiding (state)
 import Control.Monad.ST (RealWorld)
+import Data.Binary.Builder
 import Data.ByteString.Lazy qualified as BS
 import Data.List.Split (chunksOf)
 import Data.Map (Map)
 import Data.Maybe (fromMaybe, isJust)
 import Data.Time
 import UnliftIO
-  ( MonadUnliftIO, newIORef, readIORef, atomicWriteIORef, hFlush, stdout
-  , writeIORef, atomicModifyIORef', timeout
-  )
+  ( MonadUnliftIO, newIORef, readIORef, hFlush, stdout , writeIORef, timeout)
 import UnliftIO.Concurrent hiding (killThread, threadDelay)
 
 import EVM.Types (Addr, Contract, VM, W256)
 
 import Echidna.ABI
 import Echidna.Campaign (runWorker)
 import Echidna.Output.JSON qualified
+import Echidna.Server (runSSEServer)
 import Echidna.Types.Campaign
 import Echidna.Types.Config
 import Echidna.Types.Corpus (corpusSize)
 import Echidna.Types.Coverage (scoveragePoints)
-import Echidna.Types.Test (EchidnaTest(..), didFail, isOptimizationTest, TestType, TestState(..))
+import Echidna.Types.Test (EchidnaTest(..), didFail, isOptimizationTest)
 import Echidna.Types.Tx (Tx)
 import Echidna.Types.World (World)
 import Echidna.UI.Report
@@ -57,7 +56,7 @@ data UIEvent =
 -- | Set up and run an Echidna 'Campaign' and display interactive UI or
 -- print non-interactive output in desired format at the end
 ui
-  :: (MonadCatch m, MonadRandom m, MonadReader Env m, MonadUnliftIO m)
+  :: (MonadCatch m, MonadReader Env m, MonadUnliftIO m)
   => VM RealWorld -- ^ Initial VM state
   -> World   -- ^ Initial world state
   -> GenDict
@@ -159,10 +158,11 @@ ui vm world dict initialCorpus = do
 #endif
 
     NonInteractive outputFormat -> do
+      serverStopVar <- newEmptyMVar
 #ifdef INTERACTIVE_UI
       -- Handles ctrl-c, TODO: this doesn't work on Windows
       liftIO $ forM_ [sigINT, sigTERM] $ \sig ->
-        installHandler sig (Catch $ stopWorkers workers) Nothing
+        installHandler sig (Catch $ stopWorkers workers >> putMVar serverStopVar ()) Nothing
 #endif
       let forwardEvent = putStrLn . ppLogLine
       liftIO $ spawnListener env forwardEvent nworkers listenerStopVar
@@ -174,6 +174,10 @@ ui vm world dict initialCorpus = do
             putStrLn $ time <> "[status] " <> line
             hFlush stdout
 
+      case conf.campaignConf.serverPort of
+        Just port -> liftIO $ runSSEServer serverStopVar env port nworkers
+        Nothing -> pure ()
+
       ticker <- liftIO . forkIO . forever $ do
         threadDelay 3_000_000 -- 3 seconds
         printStatus
@@ -186,6 +190,11 @@ ui vm world dict initialCorpus = do
       -- print final status regardless the last scheduled update
       liftIO printStatus
 
+      when (isJust conf.campaignConf.serverPort) $ do
+        -- wait until we send all SSE events
+        liftIO $ putStrLn "Waiting until all SSE are received..."
+        readMVar serverStopVar
+
       states <- liftIO $ workerStates workers
 
       case outputFormat of

package.yaml

@@ -48,6 +48,8 @@ dependencies:
   - yaml
   - http-conduit
   - html-conduit
+  - warp
+  - wai-extra
   - xml-conduit
   - strip-ansi-escape
 

src/Main.hs

@@ -24,7 +24,7 @@ import Data.Text (Text)
 import Data.Time.Clock.System (getSystemTime, systemSeconds)
 import Data.Vector qualified as Vector
 import Data.Version (showVersion)
-import Data.Word (Word8)
+import Data.Word (Word8, Word16)
 import Main.Utf8 (withUtf8)
 import Options.Applicative
 import Paths_echidna (version)
@@ -225,6 +225,7 @@ readFileIfExists path = do
 data Options = Options
   { cliFilePath         :: NE.NonEmpty FilePath
   , cliWorkers          :: Maybe Word8
+  , cliServerPort       :: Maybe Word16
   , cliSelectedContract :: Maybe Text
   , cliConfigFilepath   :: Maybe FilePath
   , cliOutputFormat     :: Maybe OutputFormat
@@ -255,6 +256,9 @@ options = Options
   <*> optional (option auto $ long "workers"
     <> metavar "N"
     <> help "Number of workers to run")
+  <*> optional (option auto $ long "server"
+    <> metavar "PORT"
+    <> help "Run events server on the given port")
   <*> optional (option str $ long "contract"
     <> metavar "CONTRACT"
     <> help "Contract to analyze")
@@ -339,6 +343,7 @@ overrideConfig config Options{..} = do
       , seqLen = fromMaybe campaignConf.seqLen cliSeqLen
       , seed = cliSeed <|> campaignConf.seed
       , workers = cliWorkers <|> campaignConf.workers
+      , serverPort = cliServerPort <|> campaignConf.serverPort
       }
 
     overrideSolConf solConf = solConf

src/test/Tests/Seed.hs

@@ -34,6 +34,7 @@ seedTests =
         , mutConsts = defaultMutationConsts
         , coverageFormats = [Txt,Html,Lcov]
         , workers = Nothing
+        , serverPort = Nothing
         }
       }
       & overrideQuiet

tests/solidity/basic/default.yaml

@@ -89,3 +89,5 @@ rpcUrl: null
 rpcBlock: null
 # number of workers
 workers: 1
+# events server port
+server: null