Permissive-reading of the state file

This change allows us to read the state file even in the presence of
malformed JSON. This is a kind of "best-effort" hack; we don't ever
_expect_ malformed JSON; but if we find it, the best thing we can try to
do is ignore it and see if things still work correctly. That's what we
do here; when we find an invalid parse we emit a warning.

Author: noonio

hydra-node/hydra-node.cabal

@@ -102,6 +102,7 @@ library
     Hydra.Node.Util
     Hydra.Options
     Hydra.Persistence
+    Hydra.PersistenceLog
     Hydra.Utils
 
   other-modules:   Paths_hydra_node

hydra-node/src/Hydra/Logging/Messages.hs

@@ -15,13 +15,15 @@ import Hydra.Chain.Direct.Handlers (CardanoChainLog)
 import Hydra.Node (HydraNodeLog)
 import Hydra.Node.Network (NetworkLog)
 import Hydra.Options (RunOptions)
+import Hydra.PersistenceLog (PersistenceLog)
 
 data HydraLog tx
   = DirectChain {directChain :: CardanoChainLog}
   | APIServer {api :: APIServerLog}
   | Network {network :: NetworkLog}
   | Node {node :: HydraNodeLog tx}
   | NodeOptions {runOptions :: RunOptions}
+  | Persistence {persistenceLog :: PersistenceLog}
   | EnteringMainloop
   deriving stock (Generic)
 

hydra-node/src/Hydra/Node/Run.hs

@@ -95,7 +95,7 @@ run opts = do
         eventStore@EventStore{eventSource} <-
           prepareEventStore
             =<< mkFileBasedEventStore stateFile
-            =<< createPersistenceIncremental stateFile
+            =<< createPersistenceIncremental (contramap Persistence tracer) stateFile
         -- NOTE: Add any custom sinks here
         let eventSinks :: [EventSink (StateEvent Tx) IO] = []
         wetHydraNode <- hydrate (contramap Node tracer) env ledger initialChainState eventStore eventSinks

hydra-node/src/Hydra/Persistence.hs

@@ -8,6 +8,7 @@ import Conduit (
   ConduitT,
   MonadUnliftIO,
   ResourceT,
+  concatC,
   linesUnboundedAsciiC,
   mapMC,
   runResourceT,
@@ -18,6 +19,8 @@ import Conduit (
 import Control.Monad.Trans.Resource (allocate)
 import Data.Aeson qualified as Aeson
 import Data.ByteString qualified as BS
+import Hydra.Logging (Tracer, traceWith)
+import Hydra.PersistenceLog (PersistenceLog (..))
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.FilePath (takeDirectory)
 import UnliftIO (MVar, newMVar, putMVar, takeMVar, withMVar)
@@ -82,9 +85,10 @@ createPersistenceIncremental ::
   , MonadThrow m
   , FromJSON a
   ) =>
+  Tracer IO PersistenceLog ->
   FilePath ->
   m (PersistenceIncremental a m)
-createPersistenceIncremental fp = do
+createPersistenceIncremental tracer fp = do
   liftIO . createDirectoryIfMissing True $ takeDirectory fp
   mutex <- newMVar ()
   pure $
@@ -97,6 +101,14 @@ createPersistenceIncremental fp = do
       , source = source mutex
       }
  where
+  -- Maybe read the next item from persistence; or, if we failed to
+  -- decode it, we will emit a warning.
+  maybeDecode :: ByteString -> ResourceT m (Maybe a)
+  maybeDecode bs = case Aeson.eitherDecodeStrict' bs of
+    Left e -> do
+      liftIO $ traceWith tracer $ FailedToDecodeJson{reason = show e, filepath = fp, contents = show bs}
+      pure Nothing
+    Right decoded -> pure (Just decoded)
   source :: forall i. MVar () -> ConduitT i a (ResourceT m) ()
   source mutex = do
     liftIO (doesFileExist fp) >>= \case
@@ -108,12 +120,5 @@ createPersistenceIncremental fp = do
         -- NOTE: Read, decode and yield values line by line.
         sourceFileBS fp
           .| linesUnboundedAsciiC
-          .| mapMC
-            ( \bs ->
-                case Aeson.eitherDecodeStrict' bs of
-                  Left e ->
-                    lift . throwIO $
-                      PersistenceException $
-                        "Error when decoding from file " <> fp <> ": " <> show e <> "\n" <> show bs
-                  Right decoded -> pure decoded
-            )
+          .| mapMC maybeDecode
+          .| concatC

hydra-node/src/Hydra/PersistenceLog.hs

@@ -0,0 +1,24 @@
+module Hydra.PersistenceLog where
+
+import Data.Aeson (defaultOptions, genericParseJSON, genericToJSON, tagSingleConstructors)
+import Hydra.Prelude
+
+data PersistenceLog
+  = FailedToDecodeJson {reason :: String, filepath :: FilePath, contents :: String}
+  deriving stock (Eq, Show, Generic)
+
+-- Note: Specific Aeson instances, so that we don't hide the tags when emitting
+-- this log.
+instance ToJSON PersistenceLog where
+  toJSON =
+    genericToJSON
+      defaultOptions
+        { tagSingleConstructors = True
+        }
+
+instance FromJSON PersistenceLog where
+  parseJSON =
+    genericParseJSON
+      defaultOptions
+        { tagSingleConstructors = True
+        }

hydra-node/test/Hydra/Events/FileBasedSpec.hs

@@ -16,6 +16,7 @@ import Hydra.HeadLogic (StateChanged)
 import Hydra.HeadLogic.StateEvent (StateEvent (..))
 import Hydra.Ledger.Cardano (Tx)
 import Hydra.Ledger.Simple (SimpleTx)
+import Hydra.Logging (Verbosity (Verbose), withTracer)
 import Hydra.Persistence (PersistenceIncremental (..), createPersistenceIncremental)
 import Test.Aeson.GenericSpecs (
   defaultSettings,
@@ -80,17 +81,18 @@ spec = do
       forAllShrink genContinuousEvents shrink $ \events -> do
         ioProperty $ do
           withTempDir "hydra-persistence" $ \tmpDir -> do
-            let stateDir = tmpDir <> "/data"
-            PersistenceIncremental{append} <- createPersistenceIncremental stateDir
-            forM_ events append
-            -- Load and store events through the event source interface
-            EventStore{eventSource = src, eventSink = EventSink{putEvent}} <-
-              mkFileBasedEventStore stateDir =<< createPersistenceIncremental stateDir
-            loadedEvents <- getEvents src
-            -- Store all loaded events like the node would do
-            forM_ loadedEvents putEvent
-            pure $
-              loadedEvents === events
+            withTracer (Verbose "hydra-persistence") $ \tracer -> do
+              let stateDir = tmpDir <> "/data"
+              PersistenceIncremental{append} <- createPersistenceIncremental tracer stateDir
+              forM_ events append
+              -- Load and store events through the event source interface
+              EventStore{eventSource = src, eventSink = EventSink{putEvent}} <-
+                mkFileBasedEventStore stateDir =<< createPersistenceIncremental tracer stateDir
+              loadedEvents <- getEvents src
+              -- Store all loaded events like the node would do
+              forM_ loadedEvents putEvent
+              pure $
+                loadedEvents === events
 
 genContinuousEvents :: Gen [StateEvent SimpleTx]
 genContinuousEvents =
@@ -99,6 +101,7 @@ genContinuousEvents =
 withEventSourceAndSink :: (EventSource (StateEvent SimpleTx) IO -> EventSink (StateEvent SimpleTx) IO -> IO b) -> IO b
 withEventSourceAndSink action =
   withTempDir "hydra-persistence" $ \tmpDir -> do
-    let stateDir = tmpDir <> "/data"
-    EventStore{eventSource, eventSink} <- mkFileBasedEventStore stateDir =<< createPersistenceIncremental stateDir
-    action eventSource eventSink
+    withTracer (Verbose "hydra-persistence") $ \tracer -> do
+      let stateDir = tmpDir <> "/data"
+      EventStore{eventSource, eventSink} <- mkFileBasedEventStore stateDir =<< createPersistenceIncremental tracer stateDir
+      action eventSource eventSink

hydra-node/test/Hydra/PersistenceSpec.hs

@@ -7,8 +7,11 @@ import Test.Hydra.Prelude
 
 import Data.Aeson (Value (..))
 import Data.Aeson qualified as Aeson
+import Data.ByteString qualified as BS
 import Data.Text qualified as Text
+import Hydra.Logging (Verbosity (Verbose), traceWith, withTracer, withTracerOutputTo)
 import Hydra.Persistence (Persistence (..), PersistenceIncremental (..), createPersistence, createPersistenceIncremental, loadAll)
+import Hydra.PersistenceLog
 import Test.QuickCheck (checkCoverage, cover, elements, oneof, suchThat, (===))
 import Test.QuickCheck.Gen (listOf)
 import Test.QuickCheck.Monadic (monadicIO, monitor, pick, run)
@@ -35,12 +38,26 @@ spec = do
           pure $ actualResult === Just item
 
   describe "PersistenceIncremental" $ do
-    it "can handle empty files" $ do
+    it "can ignore invalid lines and emits warning" $ do
       withTempDir "hydra-persistence" $ \tmpDir -> do
-        let fp = tmpDir <> "/data"
-        writeFileBS fp ""
-        p <- createPersistenceIncremental fp
-        loadAll p `shouldReturn` ([] :: [Aeson.Value])
+        let logFile = tmpDir <> "/tracer.log"
+        withFile logFile WriteMode $ \hdl -> do
+          withTracerOutputTo hdl "persistence-incremental" $ \tracer -> do
+            let fp = tmpDir <> "/data"
+            writeFileBS fp "\"abc\"\n{\"xyz\": "
+            -- traceWith tracer $ FailedToDecodeJson{reason = "show e", filepath = "fp", contents = "show bs"}
+            p <- createPersistenceIncremental tracer fp
+            loadAll p `shouldReturn` ([Aeson.String "abc"] :: [Aeson.Value])
+        logs <- readFileBS logFile
+        logs `shouldSatisfy` BS.isInfixOf "FailedToDecodeJson"
+
+    it "can handle empty files" $ do
+      withTracer (Verbose "persistence-incremental") $ \tracer -> do
+        withTempDir "hydra-persistence" $ \tmpDir -> do
+          let fp = tmpDir <> "/data"
+          writeFileBS fp ""
+          p <- createPersistenceIncremental tracer fp
+          loadAll p `shouldReturn` ([] :: [Aeson.Value])
 
     it "is consistent after multiple append calls in presence of new-lines" $
       checkCoverage $
@@ -50,24 +67,26 @@ spec = do
           monitor (cover 10 (containsNewLine items) "some item contains a new line")
 
           actualResult <- run $
-            withTempDir "hydra-persistence" $ \tmpDir -> do
-              p <- createPersistenceIncremental $ tmpDir <> "/data"
-              forM_ items $ append p
-              loadAll p
+            withTracer (Verbose "persistence-incremental") $ \tracer -> do
+              withTempDir "hydra-persistence" $ \tmpDir -> do
+                p <- createPersistenceIncremental tracer $ tmpDir <> "/data"
+                forM_ items $ append p
+                loadAll p
           pure $ actualResult === items
 
     it "it cannot load from a different thread once having started appending" $
       monadicIO $ do
         items <- pick $ listOf genPersistenceItem
         moreItems <- pick $ listOf genPersistenceItem `suchThat` ((> 2) . length)
         pure $
-          withTempDir "hydra-persistence" $ \tmpDir -> do
-            p <- createPersistenceIncremental $ tmpDir <> "/data"
-            forM_ items $ append p
-            loadAll p `shouldReturn` items
-            raceLabelled_
-              ("forever-load-all", forever $ threadDelay 0.01 >> loadAll p)
-              ("append-more-items", forM_ moreItems $ \item -> append p item >> threadDelay 0.01)
+          withTracer (Verbose "persistence-incremental") $ \tracer -> do
+            withTempDir "hydra-persistence" $ \tmpDir -> do
+              p <- createPersistenceIncremental tracer $ tmpDir <> "/data"
+              forM_ items $ append p
+              loadAll p `shouldReturn` items
+              raceLabelled_
+                ("forever-load-all", forever $ threadDelay 0.01 >> loadAll p)
+                ("append-more-items", forM_ moreItems $ \item -> append p item >> threadDelay 0.01)
 
 genPersistenceItem :: Gen Aeson.Value
 genPersistenceItem =