Remove persistence from withReliability

.github/workflows/docker.yaml

@@ -19,6 +19,7 @@ jobs:
     strategy:
       matrix:
         target: [ hydra-node, hydra-tui, hydraw, hydra-explorer ]
+        system: [ x86_64-linux, aarch64-linux ]
 
     runs-on: ubuntu-latest
     steps:
@@ -49,7 +50,7 @@ jobs:
       run: |
         IMAGE_NAME=ghcr.io/${{github.repository_owner}}/${{matrix.target}}
         echo "IMAGE_NAME=${IMAGE_NAME}" >> $GITHUB_ENV
-        nix build .#docker-${{ matrix.target }} && docker load < ./result
+        nix build .#packages.${{matrix.system}}.docker-${{ matrix.target }} && docker load < ./result
         # Determine whether we are building a tag and if yes, set a VERSION_NAME
         BUILDING_TAG=${{github.ref_type == 'tag'}}
         [[ ${BUILDING_TAG} = true ]] && \

hydra-node/src/Hydra/Network/Reliability.hs

@@ -80,7 +80,7 @@
 -- become a problem, this can be mitigated by closing and reopening a head.
 module Hydra.Network.Reliability where
 
-import Hydra.Prelude hiding (empty, fromList, length, replicate, zipWith)
+import Hydra.Prelude hiding (empty, fromList, replicate, zipWith)
 
 import Cardano.Binary (serialize')
 import Cardano.Crypto.Util (SignableRepresentation (getSignableRepresentation))
@@ -95,13 +95,12 @@ import Control.Concurrent.Class.MonadSTM (
 import Control.Tracer (Tracer)
 import Data.IntMap qualified as IMap
 import Data.Sequence.Strict ((|>))
-import Data.Sequence.Strict qualified as Seq
+import Data.Vector qualified as Vector
 import Data.Vector (
   Vector,
   elemIndex,
   fromList,
   generate,
-  length,
   replicate,
   zipWith,
   (!?),
@@ -216,18 +215,16 @@ withReliability ::
   (MonadThrow (STM m), MonadThrow m, MonadAsync m) =>
   -- | Tracer for logging messages.
   Tracer m ReliabilityLog ->
-  -- | Our persistence handle
-  MessagePersistence m outbound ->
   -- | Our own party identifier.
   Party ->
   -- | Other parties' identifiers.
   [Party] ->
   -- | Underlying network component providing consuming and sending channels.
   NetworkComponent m (Authenticated (ReliableMsg (Heartbeat inbound))) (ReliableMsg (Heartbeat outbound)) a ->
   NetworkComponent m (Authenticated (Heartbeat inbound)) (Heartbeat outbound) a
-withReliability tracer MessagePersistence{saveAcks, loadAcks, appendMessage, loadMessages} me otherParties withRawNetwork callback action = do
-  acksCache <- loadAcks >>= newTVarIO
-  sentMessages <- loadMessages >>= newTVarIO . Seq.fromList
+withReliability tracer me otherParties withRawNetwork callback action = do
+  acksCache <- newTVarIO (Vector.replicate (length (me : otherParties)) 0)
+  sentMessages <- newTVarIO mempty
   resendQ <- newTQueueIO
   let ourIndex = fromMaybe (error "This cannot happen because we constructed the list with our party inside.") (findPartyIndex me)
   let resend = writeTQueue resendQ
@@ -243,13 +240,10 @@ withReliability tracer MessagePersistence{saveAcks, loadAcks, appendMessage, loa
             case msg of
               Data{} -> do
                 localCounter <- atomically $ cacheMessage msg >> incrementAckCounter
-                saveAcks localCounter
-                appendMessage msg
                 traceWith tracer BroadcastCounter{ourIndex, localCounter}
                 broadcast $ ReliableMsg localCounter msg
               Ping{} -> do
                 localCounter <- readTVarIO acksCache
-                saveAcks localCounter
                 traceWith tracer BroadcastPing{ourIndex, localCounter}
                 broadcast $ ReliableMsg localCounter msg
         }

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

@@ -123,19 +123,17 @@ withNetwork ::
 withNetwork tracer configuration callback action = do
   let localhost = Host{hostname = show host, port}
       me = deriveParty signingKey
-      numberOfParties = length $ me : otherParties
-  messagePersistence <- configureMessagePersistence (contramap Node tracer) persistenceDir numberOfParties
 
   let reliability =
         withFlipHeartbeats $
-          withReliability (contramap Reliability tracer) messagePersistence me otherParties $
+          withReliability (contramap Reliability tracer) me otherParties $
             withAuthentication (contramap Authentication tracer) signingKey otherParties $
               withOuroborosNetwork (contramap Network tracer) localhost peers
 
   withHeartbeat nodeId reliability (callback . mapHeartbeat) $ \network ->
     action network
  where
-  NetworkConfiguration{persistenceDir, signingKey, otherParties, host, port, peers, nodeId} = configuration
+  NetworkConfiguration{signingKey, otherParties, host, port, peers, nodeId} = configuration
 
   mapHeartbeat :: Either Connectivity (Authenticated (Message tx)) -> NetworkEvent (Message tx)
   mapHeartbeat = \case

hydra-node/test/Hydra/Network/ReliabilitySpec.hs

@@ -23,9 +23,7 @@ import Hydra.Network.Message (Connectivity)
 import Hydra.Network.Reliability (MessagePersistence (..), ReliabilityLog (..), ReliableMsg (..), withReliability)
 import Hydra.Node.Network (withFlipHeartbeats)
 import Hydra.Persistence (
-  Persistence (..),
   PersistenceIncremental (..),
-  createPersistence,
   createPersistenceIncremental,
  )
 import System.Directory (doesFileExist)
@@ -99,9 +97,8 @@ spec = parallel $ do
     prop "broadcast messages to the network assigning a sequential id" $ \(messages :: [String]) ->
       let sentMsgs = runSimOrThrow $ do
             sentMessages <- newTVarIO empty
-            persistence <- mockMessagePersistence 1
 
-            withReliability nullTracer persistence alice [] (captureOutgoing sentMessages) noop $ \Network{broadcast} -> do
+            withReliability nullTracer alice [] (captureOutgoing sentMessages) noop $ \Network{broadcast} -> do
               mapM_ (broadcast . Data "node-1") messages
 
             fromList . Vector.toList <$> readTVarIO sentMessages
@@ -119,16 +116,14 @@ spec = parallel $ do
             randomSeed <- newTVarIO $ mkStdGen seed
             aliceToBob <- newTQueueIO
             bobToAlice <- newTQueueIO
-            alicePersistence <- mockMessagePersistence 2
-            bobPersistence <- mockMessagePersistence 2
             let
               -- this is a NetworkComponent that broadcasts authenticated messages
               -- mediated through a read and a write TQueue but drops 0.2 % of them
               aliceFailingNetwork = failingNetwork randomSeed alice (bobToAlice, aliceToBob)
               bobFailingNetwork = failingNetwork randomSeed bob (aliceToBob, bobToAlice)
 
-              bobReliabilityStack = reliabilityStack bobPersistence bobFailingNetwork (captureTraces emittedTraces) "bob" bob [alice]
-              aliceReliabilityStack = reliabilityStack alicePersistence aliceFailingNetwork (captureTraces emittedTraces) "alice" alice [bob]
+              bobReliabilityStack = reliabilityStack bobFailingNetwork (captureTraces emittedTraces) "bob" bob [alice]
+              aliceReliabilityStack = reliabilityStack aliceFailingNetwork (captureTraces emittedTraces) "alice" alice [bob]
 
               runAlice = runPeer aliceReliabilityStack "alice" messagesReceivedByAlice messagesReceivedByBob aliceToBobMessages bobToAliceMessages
               runBob = runPeer bobReliabilityStack "bob" messagesReceivedByBob messagesReceivedByAlice bobToAliceMessages aliceToBobMessages
@@ -150,10 +145,8 @@ spec = parallel $ do
     it "broadcast updates counter from peers" $ do
       let receivedMsgs = runSimOrThrow $ do
             sentMessages <- newTVarIO empty
-            alicePersistence <- mockMessagePersistence 2
             withReliability
               nullTracer
-              alicePersistence
               alice
               [bob]
               ( \incoming action -> do
@@ -173,26 +166,10 @@ spec = parallel $ do
       withTempDir "network-messages-persistence" $ \tmpDir -> do
         let networkMessagesFile = tmpDir <> "/network-messages"
 
-        Persistence{load, save} <- createPersistence $ tmpDir <> "/acks"
-        PersistenceIncremental{loadAll, append} <- createPersistenceIncremental networkMessagesFile
-
-        let messagePersistence =
-              MessagePersistence
-                { loadAcks = do
-                    mloaded <- load
-                    case mloaded of
-                      Nothing -> pure $ replicate (length [alice, bob]) 0
-                      Just acks -> pure acks
-                , saveAcks = save
-                , loadMessages = loadAll
-                , appendMessage = append
-                }
-
         receivedMsgs <- do
           sentMessages <- newTVarIO empty
           withReliability
             nullTracer
-            messagePersistence
             alice
             [bob]
             ( \incoming action -> do
@@ -212,7 +189,6 @@ spec = parallel $ do
         reloadAll networkMessagesFile `shouldReturn` [Data "node-1" msg]
 
         doesFileExist (tmpDir </> "acks") `shouldReturn` True
-        load `shouldReturn` Just (fromList [1, 1])
  where
   runPeer reliability partyName receivedMessageContainer sentMessageContainer messagesToSend expectedMessages =
     reliability (capturePayload receivedMessageContainer) $ \Network{broadcast} -> do
@@ -224,10 +200,10 @@ spec = parallel $ do
         (waitForAllMessages expectedMessages receivedMessageContainer)
         (waitForAllMessages messagesToSend sentMessageContainer)
 
-  reliabilityStack persistence underlyingNetwork tracer nodeId party peers =
+  reliabilityStack underlyingNetwork tracer nodeId party peers =
     withHeartbeat nodeId $
       withFlipHeartbeats $
-        withReliability tracer persistence party peers underlyingNetwork
+        withReliability tracer party peers underlyingNetwork
 
   failingNetwork seed peer (readQueue, writeQueue) callback action =
     withAsync
@@ -260,14 +236,12 @@ noop = const $ pure ()
 aliceReceivesMessages :: [Authenticated (ReliableMsg (Heartbeat msg))] -> [Authenticated (Heartbeat msg)]
 aliceReceivesMessages messages = runSimOrThrow $ do
   receivedMessages <- newTVarIO empty
-  alicePersistence <- mockMessagePersistence 3
 
   let baseNetwork incoming _ = mapM incoming messages
 
       aliceReliabilityStack =
         withReliability
           nullTracer
-          alicePersistence
           alice
           [bob, carol]
           baseNetwork