Do not persist intra-block writes

This stops us writing rows to the database that are overwritten in the same block. Benchmarking says this isn't substantially faster, but it can save space.

This isn't safe to enable if you use Rosetta, which relies on these intermediate rows.

Change-Id: I88660fa4f0ee481e06808f66d98a86a3eb3722db

bench/Chainweb/Pact/Backend/Bench.hs

@@ -53,6 +53,7 @@ import Chainweb.MerkleLogHash
 import Chainweb.Pact.Backend.RelationalCheckpointer
 import Chainweb.Pact.Backend.Types
 import Chainweb.Pact.Backend.Utils
+import Chainweb.Pact.Service.Types
 import Chainweb.Pact.Types
 import Chainweb.Test.TestVersions
 import Chainweb.Utils.Bench
@@ -146,7 +147,7 @@ cpWithBench torun =
         let neverLogger = genericLogger Error (\_ -> return ())
         !sqliteEnv <- openSQLiteConnection dbFile chainwebPragmas
         !cenv <-
-          initRelationalCheckpointer defaultModuleCacheLimit sqliteEnv neverLogger testVer testChainId
+          initRelationalCheckpointer defaultModuleCacheLimit sqliteEnv DoNotPersistIntraBlockWrites neverLogger testVer testChainId
         return $ NoopNFData (sqliteEnv, cenv)
 
     teardown (NoopNFData (sqliteEnv, _cenv)) = closeSQLiteConnection sqliteEnv

bench/Chainweb/Pact/Backend/ForkingBench.hs

@@ -123,13 +123,15 @@ data BenchConfig = BenchConfig
     -- ^ whether or not to validate the blocks as part of the benchmark
   , compact :: Compact
     -- ^ whether or not to compact the pact database prior to benchmarking
+  , persistIntraBlockWrites :: IntraBlockPersistence
   }
 
 defBenchConfig :: BenchConfig
 defBenchConfig = BenchConfig
   { numPriorBlocks = 100
   , validate = DontValidate
   , compact = DontCompact
+  , persistIntraBlockWrites = PersistIntraBlockWrites
   }
 
 data Compact = DoCompact | DontCompact
@@ -144,20 +146,22 @@ bench rdb = C.bgroup "PactService" $
     , doubleForkingBench
     ] ++ map (oneBlock defBenchConfig) [1, 10, 50, 100]
       ++ map (oneBlock validateCfg) [0, 1, 10, 50, 100]
+      ++ map (oneBlock validateCfg { persistIntraBlockWrites = DoNotPersistIntraBlockWrites })
+        [0, 1, 10, 50, 100]
       ++ map (oneBlock compactCfg) [0, 1, 10, 50, 100]
       ++ map (oneBlock compactValidateCfg) [1, 10, 50, 100]
   where
     validateCfg = defBenchConfig { validate = DoValidate }
     compactCfg = defBenchConfig { compact = DoCompact }
     compactValidateCfg = compactCfg { validate = DoValidate }
 
-    forkingBench = withResources rdb 10 Quiet DontCompact
+    forkingBench = withResources rdb 10 Quiet DontCompact PersistIntraBlockWrites
         $ \mainLineBlocks pdb bhdb nonceCounter pactQueue _ ->
             C.bench "forkingBench"  $ C.whnfIO $ do
               let (T3 _ join1 _) = mainLineBlocks !! 5
               void $ playLine pdb bhdb 5 join1 pactQueue nonceCounter
 
-    doubleForkingBench = withResources rdb 10 Quiet DontCompact
+    doubleForkingBench = withResources rdb 10 Quiet DontCompact PersistIntraBlockWrites
         $ \mainLineBlocks pdb bhdb nonceCounter pactQueue _ ->
             C.bench "doubleForkingBench"  $ C.whnfIO $ do
               let (T3 _ join1 _) = mainLineBlocks !! 5
@@ -167,7 +171,7 @@ bench rdb = C.bgroup "PactService" $
               void $ playLine pdb bhdb forkLength2 join1 pactQueue nonceCounter
 
     oneBlock :: BenchConfig -> Int -> C.Benchmark
-    oneBlock cfg txCount = withResources rdb cfg.numPriorBlocks Error cfg.compact go
+    oneBlock cfg txCount = withResources rdb cfg.numPriorBlocks Error cfg.compact cfg.persistIntraBlockWrites go
       where
         go _mainLineBlocks _pdb _bhdb _nonceCounter pactQueue txsPerBlock = do
           C.bench name $ C.whnfIO $ do
@@ -178,6 +182,7 @@ bench rdb = C.bgroup "PactService" $
                [ "txCount=" ++ show txCount
                , "validate=" ++ show (cfg.validate == DoValidate)
                , "compact=" ++ show (cfg.compact == DoCompact)
+               , "persist=" ++ show cfg.persistIntraBlockWrites
                ]
           ++ "]"
 
@@ -276,9 +281,10 @@ withResources :: ()
   -> Word64
   -> LogLevel
   -> Compact
+  -> IntraBlockPersistence
   -> RunPactService
   -> C.Benchmark
-withResources rdb trunkLength logLevel compact f = C.envWithCleanup create destroy unwrap
+withResources rdb trunkLength logLevel compact p f = C.envWithCleanup create destroy unwrap
   where
 
     unwrap ~(NoopNFData (Resources {..})) =
@@ -314,6 +320,7 @@ withResources rdb trunkLength logLevel compact f = C.envWithCleanup create destr
         reqQ <- newPactQueue pactQueueSize
         a <- async $ runPactService version cid l reqQ mempool bhdb pdb sqlEnv testPactServiceConfig
             { _pactBlockGasLimit = 180_000
+            , _pactPersistIntraBlockWrites = p
             }
 
         return (a, reqQ)

src/Chainweb/Chainweb.hs

@@ -159,7 +159,7 @@ import Chainweb.Mempool.P2pConfig
 import Chainweb.Miner.Config
 import qualified Chainweb.OpenAPIValidation as OpenAPIValidation
 import Chainweb.Pact.RestAPI.Server (PactServerData(..))
-import Chainweb.Pact.Service.Types (PactServiceConfig(..))
+import Chainweb.Pact.Service.Types (PactServiceConfig(..), IntraBlockPersistence(..))
 import Chainweb.Pact.Validations
 import Chainweb.Payload.PayloadStore
 import Chainweb.Payload.PayloadStore.RocksDB
@@ -427,8 +427,12 @@ withChainwebInternal conf logger peer serviceSock rocksDb pactDbDir backupDir re
       , _pactBlockGasLimit = maybe id min maxGasLimit (_configBlockGasLimit conf)
       , _pactLogGas = _configLogGas conf
       , _pactModuleCacheLimit = _configModuleCacheLimit conf
-      , _pactFullHistoryRequired = _configRosetta conf -- this could be OR'd with other things that require full history
       , _pactEnableLocalTimeout = _configEnableLocalTimeout conf
+      , _pactFullHistoryRequired = _configFullHistoricPactState conf
+      , _pactPersistIntraBlockWrites =
+          if _configFullHistoricPactState conf
+          then PersistIntraBlockWrites
+          else DoNotPersistIntraBlockWrites
       }
 
     pruningLogger :: T.Text -> logger

src/Chainweb/Chainweb/Configuration.hs

@@ -66,6 +66,7 @@ module Chainweb.Chainweb.Configuration
 , configThrottling
 , configReorgLimit
 , configRosetta
+, configFullHistoricPactState
 , configBackup
 , configServiceApi
 , configOnlySyncPact
@@ -394,6 +395,7 @@ data ChainwebConfiguration = ChainwebConfiguration
     , _configReorgLimit :: !RewindLimit
     , _configPreInsertCheckTimeout :: !Micros
     , _configAllowReadsInLocal :: !Bool
+    , _configFullHistoricPactState :: !Bool
     , _configRosetta :: !Bool
     , _configBackup :: !BackupConfig
     , _configServiceApi :: !ServiceApiConfig
@@ -421,6 +423,11 @@ validateChainwebConfiguration c = do
     validateBackupConfig (_configBackup c)
     unless (c ^. chainwebVersion . versionDefaults . disablePeerValidation) $
         validateP2pConfiguration (_configP2p c)
+    when (_configRosetta c && not (_configFullHistoricPactState c)) $
+        throwError $ T.unwords
+            [ "To enable rosetta, full historic pact state must also be enabled or"
+            , "the Rosetta index will be incomplete."
+            ]
     validateChainwebVersion (_configChainwebVersion c)
 
 validateChainwebVersion :: ConfigValidation ChainwebVersion []
@@ -459,6 +466,7 @@ defaultChainwebConfiguration v = ChainwebConfiguration
     , _configPreInsertCheckTimeout = defaultPreInsertCheckTimeout
     , _configAllowReadsInLocal = False
     , _configRosetta = False
+    , _configFullHistoricPactState = True
     , _configServiceApi = defaultServiceApiConfig
     , _configOnlySyncPact = False
     , _configReadOnlyReplay = False
@@ -486,6 +494,7 @@ instance ToJSON ChainwebConfiguration where
         , "preInsertCheckTimeout" .= _configPreInsertCheckTimeout o
         , "allowReadsInLocal" .= _configAllowReadsInLocal o
         , "rosetta" .= _configRosetta o
+        , "fullHistoricPactState" .= _configFullHistoricPactState o
         , "serviceApi" .= _configServiceApi o
         , "onlySyncPact" .= _configOnlySyncPact o
         , "readOnlyReplay" .= _configReadOnlyReplay o
@@ -517,6 +526,7 @@ instance FromJSON (ChainwebConfiguration -> ChainwebConfiguration) where
         <*< configAllowReadsInLocal ..: "allowReadsInLocal" % o
         <*< configPreInsertCheckTimeout ..: "preInsertCheckTimeout" % o
         <*< configRosetta ..: "rosetta" % o
+        <*< configFullHistoricPactState ..: "fullHistoricPactState" % o
         <*< configServiceApi %.: "serviceApi" % o
         <*< configOnlySyncPact ..: "onlySyncPact" % o
         <*< configReadOnlyReplay ..: "readOnlyReplay" % o
@@ -563,6 +573,9 @@ pChainwebConfiguration = id
     <*< configRosetta .:: boolOption_
         % long "rosetta"
         <> help "Enable the Rosetta endpoints."
+    <*< configFullHistoricPactState .:: boolOption_
+        % long "full-historic-pact-state"
+        <> help "Write full historic Pact state; only enable for custodial or archival nodes."
     <*< configCuts %:: pCutConfig
     <*< configServiceApi %:: pServiceApiConfig
     <*< configMining %:: pMiningConfig