Merge pull request #725 from input-output-hk/plt-5853-collect-garbage

PLT-5853 Contract store garbage collector

.github/PULL_REQUEST_TEMPLATE.md

@@ -12,6 +12,8 @@ Pre-submit checklist:
     - [ ] Key commits have useful messages
     - [ ] Relevant tickets are mentioned in commit messages
     - [ ] Formatting, PNG optimization, etc. are updated
+    - [ ] Operables are updated with changes to executable command line options.
+    - [ ] Deploy charts updated with changes to operables.
 - PR
     - [ ] Self-reviewed the diff
     - [ ] Useful pull request description

.github/workflows/test.yml

@@ -55,10 +55,13 @@ jobs:
           path: |
             ~/integration-tests.log
             ~/integration-tests.status
+      - if: ${{ steps.cache-results.outputs.cache-hit != 'true' }}
+        name: Build tests
+        run: nix build .#integration-tests --show-trace
       - if: ${{ steps.cache-results.outputs.cache-hit != 'true' }}
         name: Run tests
         run: |
-          nix run .#integration-tests --show-trace -- --strict > ~/integration-tests.log
+          result/bin/marlowe-integration-tests --strict > ~/integration-tests.log
           echo $? > ~/integration-tests.status
           exit $?
         env:
@@ -78,4 +81,4 @@ jobs:
         run: |
           echo "Using cached test results. Test log:"
           cat ~/integration-tests.log
-          exit $(cat ~/integration-tests.status)
+          exit $(cat ~/integration-tests.status)

deploy/marlowe-runtime/templates/marlowe-contract.yaml

@@ -23,8 +23,20 @@ spec:
           value: "3728"
         - name: TRANSFER_PORT
           value: "3729"
+        - name: MARLOWE_CHAIN_SYNC_HOST
+          value: chain-sync-{{ $network }}.{{ $.Values.namespace }}
+        - name: MARLOWE_CHAIN_SYNC_QUERY_PORT
+          value: "3716"
+        - name: SYNC_HOST
+          value: marlowe-sync-{{ . }}.{{ $.Values.namespace }}
+        - name: MARLOWE_BULK_SYNC_PORT
+          value: "3730"
         - name: STORE_DIR
           value: /store
+        - name: MAX_STORE_SIZE
+          value: 214748364800 # 200 GB
+        - name: MIN_CONTRACT_AGE
+          value: 1800 # 30 minutes
         - name: HTTP_PORT
           value: "3787"
         - name: OTEL_EXPORTER_OTLP_ENDPOINT

deploy/marlowe-runtime/templates/marlowe-proxy.yaml

@@ -39,6 +39,8 @@ spec:
           value: "3724"
         - name: MARLOWE_HEADER_SYNC_PORT
           value: "3725"
+        - name: MARLOWE_BULK_SYNC_PORT
+          value: "3730"
         - name: MARLOWE_QUERY_PORT
           value: "3726"
         - name: HTTP_PORT

deploy/marlowe-runtime/templates/marlowe-sync.yaml

@@ -21,6 +21,8 @@ spec:
         value: "3724"
       - name: MARLOWE_HEADER_SYNC_PORT
         value: "3725"
+      - name: MARLOWE_BULK_SYNC_PORT
+        value: "3730"
       - name: MARLOWE_QUERY_PORT
         value: "3726"
       - name: MARLOWE_CHAIN_SYNC_HOST

marlowe-integration-tests/test/Language/Marlowe/Runtime/Integration/Contract.hs

@@ -8,6 +8,7 @@ module Language.Marlowe.Runtime.Integration.Contract where
 
 import Cardano.Api (ScriptData (ScriptDataBytes), hashScriptDataBytes, unsafeHashableScriptData)
 import Colog (HasLog (..), LogAction, Message)
+import Control.Applicative (Applicative (..))
 import Control.Arrow (Arrow (..), returnA)
 import Control.Concurrent.Component
 import Control.Monad (foldM, unless)
@@ -34,14 +35,21 @@ import Language.Marlowe.Object.Types (
   ObjectType (..),
   fromCoreContract,
  )
+import Language.Marlowe.Protocol.BulkSync.Client (serveMarloweBulkSyncClient)
+import Language.Marlowe.Protocol.BulkSync.Server
 import Language.Marlowe.Protocol.Load.Client (MarloweLoadClient, pushContract, serveMarloweLoadClient)
 import Language.Marlowe.Protocol.Transfer.Client (
   MarloweTransferClient (..),
   hoistMarloweTransferClient,
   serveMarloweTransferClient,
  )
 import Language.Marlowe.Runtime.Cardano.Api (fromCardanoDatumHash, toCardanoScriptData)
-import Language.Marlowe.Runtime.ChainSync.Api (DatumHash (..), toDatum)
+import Language.Marlowe.Runtime.ChainSync.Api (
+  ChainSyncQuery (..),
+  DatumHash (..),
+  WithGenesis (..),
+  toDatum,
+ )
 import Language.Marlowe.Runtime.Client (
   exportContract,
   exportIncremental,
@@ -58,6 +66,7 @@ import Network.Protocol.Connection
 import Network.Protocol.Driver.Trace (HasSpanContext (..))
 import Network.Protocol.Peer.Trace (defaultSpanContext)
 import Network.Protocol.Query.Client (QueryClient, serveQueryClient)
+import Network.Protocol.Query.Server (respond)
 import Network.TypedProtocol (unsafeIntToNat)
 import Pipes (each, yield, (>->))
 import qualified Pipes.Internal as PI
@@ -72,8 +81,8 @@ import Test.QuickCheck (Gen, chooseInt, counterexample, forAll)
 import UnliftIO (
   Concurrently (..),
   atomically,
-  concurrently,
   liftIO,
+  race,
   throwIO,
  )
 
@@ -357,6 +366,8 @@ runContractTest test = runResourceT do
         { contractStoreDirectory = resolveWorkspacePath workspace "contract-store"
         , contractStoreStagingDirectory = resolveWorkspacePath workspace "staging-areas"
         , lockingMicrosecondsBetweenRetries = 100_000
+        , minContractAge = 60 -- In seconds
+        , maxStoreSize = 4 * 1024 * 1024 * 1024 -- 4 GB
         }
   let testComponent = proc contractDeps -> do
         Contract.MarloweContract{..} <- Contract.contract -< contractDeps
@@ -375,9 +386,27 @@ runContractTest test = runResourceT do
         Contract.ContractDependencies
           { batchSize = unsafeIntToNat 10
           , contractStore
+          , marloweBulkSyncConnector = directConnector serveMarloweBulkSyncClient $ ServerSource do
+              let idle =
+                    ServerStIdle
+                      { recvMsgRequestNext = \_ -> pure $ SendMsgWait poll
+                      , recvMsgIntersect = \_ -> pure $ SendMsgIntersectNotFound Genesis idle
+                      , recvMsgDone = pure ()
+                      }
+                  poll =
+                    ServerStPoll
+                      { recvMsgPoll = pure $ SendMsgWait poll
+                      , recvMsgCancel = pure idle
+                      }
+              pure $ MarloweBulkSyncServer $ pure idle
+          , chainSyncQueryConnector = directConnector serveQueryClient $ ServerSource $ pure $ respond (liftA2 (,)) \case
+              GetSecurityParameter -> pure 1
+              req -> fail $ "Request not mocked: " <> show req
           }
-  (a, _) <- runNoopEventT $ flip runReaderT testHandle $ concurrently test runTestComponent
-  pure a
+  result <- runNoopEventT $ flip runReaderT testHandle $ race test runTestComponent
+  case result of
+    Left a -> pure a
+    Right _ -> fail "contract component finished before test could finish"
 
 type TestM = ReaderT TestHandle (NoopEventT TestRef AnySelector (ResourceT IO))
 

marlowe-integration/src/Test/Integration/Marlowe/Local.hs

@@ -264,6 +264,8 @@ withLocalMarloweRuntime' MarloweRuntimeOptions{..} test = withRunInIO \runInIO -
           { contractStoreDirectory = resolveWorkspacePath workspace "contract-store"
           , contractStoreStagingDirectory = resolveWorkspacePath workspace "contract-staging-area"
           , lockingMicrosecondsBetweenRetries = 100_000
+          , minContractAge = 60 -- In seconds
+          , maxStoreSize = 4 * 1024 * 1024 * 1024 -- 4 GB
           }
 
     let baseUrl = BaseUrl Http "localhost" webPort ""

marlowe-runtime/changelog.d/20231010_152643_jhbertra_plt_5853_collect_garbage.md

@@ -0,0 +1,3 @@
+### Added
+
+- Garbage collection added to `marlowe-contract`