DBLayer for pending LocalTxSubmissions

lib/core/src/Cardano/DB/Sqlite.hs

@@ -34,6 +34,7 @@ module Cardano.DB.Sqlite
     -- * Helpers
     , chunkSize
     , dbChunked
+    , dbChunkedFor
     , dbChunked'
     , handleConstraint
     , unsafeRunQuery
@@ -617,7 +618,16 @@ dbChunked
     => ([record] -> SqlPersistT IO b)
     -> [record]
     -> SqlPersistT IO ()
-dbChunked = chunkedM (chunkSizeFor @record)
+dbChunked = dbChunkedFor @record
+
+-- | Like 'dbChunked', but generalized for the case where the input list is not
+-- the same type as the record.
+dbChunkedFor
+    :: forall record a b. PersistEntity record
+    => ([a] -> SqlPersistT IO b)
+    -> [a]
+    -> SqlPersistT IO ()
+dbChunkedFor = chunkedM (chunkSizeFor @record)
 
 -- | Like 'dbChunked', but allows bundling elements with a 'Key'. Useful when
 -- used with 'repsertMany'.

lib/core/src/Cardano/Wallet/DB.hs

@@ -1,4 +1,6 @@
 {-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE ExistentialQuantification #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
@@ -55,7 +57,15 @@ import Cardano.Wallet.Primitive.Types.Coin
 import Cardano.Wallet.Primitive.Types.Hash
     ( Hash )
 import Cardano.Wallet.Primitive.Types.Tx
-    ( TransactionInfo, Tx (..), TxMeta, TxStatus )
+    ( LocalTxSubmissionStatus
+    , SealedTx
+    , TransactionInfo
+    , Tx (..)
+    , TxMeta
+    , TxStatus
+    )
+import Control.DeepSeq
+    ( NFData )
 import Control.Monad.IO.Class
     ( MonadIO )
 import Control.Monad.Trans.Except
@@ -64,6 +74,8 @@ import Data.Quantity
     ( Quantity (..) )
 import Data.Word
     ( Word32, Word8 )
+import GHC.Generics
+    ( Generic )
 
 import qualified Data.List as L
 
@@ -242,6 +254,23 @@ data DBLayer m s k = forall stm. (MonadIO stm, MonadFail stm) => DBLayer
         --
         -- If the wallet doesn't exist, this operation returns an error.
 
+    , putLocalTxSubmission
+        :: PrimaryKey WalletId
+        -> Hash "Tx"
+        -> SealedTx
+        -> SlotNo
+        -> ExceptT ErrNoSuchWallet stm ()
+        -- ^ Add or update a transaction in the local submission pool with the
+        -- most recent submission slot.
+
+    , readLocalTxSubmissionPending
+        :: PrimaryKey WalletId
+        -> stm [LocalTxSubmissionStatus SealedTx]
+        -- ^ List all transactions from the local submission pool which are
+        -- still pending as of the latest checkpoint of the given wallet. The
+        -- slot numbers for first submission and most recent submission are
+        -- included.
+
     , updatePendingTxForExpiry
         :: PrimaryKey WalletId
         -> SlotNo
@@ -330,8 +359,8 @@ newtype ErrWalletAlreadyExists
 -- functions like 'enqueueCheckpoint' needs to be associated to a corresponding
 -- wallet. Some other may not because they are information valid for all wallets
 -- (like for instance, the last known network tip).
-newtype PrimaryKey key = PrimaryKey key
-    deriving (Show, Eq, Ord)
+newtype PrimaryKey key = PrimaryKey { unPrimaryKey :: key }
+    deriving (Show, Eq, Ord, Generic, NFData)
 
 -- | Clean a database by removing all wallets.
 cleanDB :: DBLayer m s k -> m ()

lib/core/src/Cardano/Wallet/DB/MVar.hs

@@ -37,15 +37,18 @@ import Cardano.Wallet.DB.Model
     , mIsStakeKeyRegistered
     , mListCheckpoints
     , mListWallets
+    , mPrune
     , mPutCheckpoint
     , mPutDelegationCertificate
     , mPutDelegationRewardBalance
+    , mPutLocalTxSubmission
     , mPutPrivateKey
     , mPutTxHistory
     , mPutWalletMeta
     , mReadCheckpoint
     , mReadDelegationRewardBalance
     , mReadGenesisParameters
+    , mReadLocalTxSubmissionPending
     , mReadPrivateKey
     , mReadTxHistory
     , mReadWalletMeta
@@ -65,7 +68,7 @@ import Cardano.Wallet.Primitive.Types.Hash
 import Cardano.Wallet.Primitive.Types.Tx
     ( TransactionInfo (..) )
 import Control.DeepSeq
-    ( NFData, deepseq )
+    ( NFData, force )
 import Control.Monad.Trans.Except
     ( ExceptT (..) )
 import Data.Functor.Identity
@@ -90,9 +93,9 @@ newDBLayer timeInterpreter = do
                                       Wallets
         -----------------------------------------------------------------------}
 
-        { initializeWallet = \pk cp meta txs gp -> ExceptT $ do
-            cp `deepseq` meta `deepseq`
-                alterDB errWalletAlreadyExists db (mInitializeWallet pk cp meta txs gp)
+        { initializeWallet = \pk cp meta txs gp -> ExceptT $
+                alterDB errWalletAlreadyExists db $
+                mInitializeWallet pk cp meta txs gp
 
         , removeWallet = ExceptT . alterDB errNoSuchWallet db . mRemoveWallet
 
@@ -102,30 +105,35 @@ newDBLayer timeInterpreter = do
                                     Checkpoints
         -----------------------------------------------------------------------}
 
-        , putCheckpoint = \pk cp -> ExceptT $ do
-            cp `deepseq` alterDB errNoSuchWallet db (mPutCheckpoint pk cp)
+        , putCheckpoint = \pk cp -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mPutCheckpoint pk cp
 
         , readCheckpoint = readDB db . mReadCheckpoint
 
         , listCheckpoints = readDB db . mListCheckpoints
 
         , rollbackTo = \pk pt -> ExceptT $
-            alterDB errNoSuchWallet db (mRollbackTo pk pt)
+            alterDB errNoSuchWallet db $
+            mRollbackTo pk pt
 
-        , prune = \_ _ -> error "MVar.prune: not implemented"
+        , prune = \pk epochStability -> ExceptT $
+                alterDB errNoSuchWallet db $
+                mPrune pk epochStability
 
         {-----------------------------------------------------------------------
                                    Wallet Metadata
         -----------------------------------------------------------------------}
 
-        , putWalletMeta = \pk meta -> ExceptT $ do
-            meta `deepseq` alterDB errNoSuchWallet db (mPutWalletMeta pk meta)
+        , putWalletMeta = \pk meta -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mPutWalletMeta pk meta
 
         , readWalletMeta = readDB db . mReadWalletMeta timeInterpreter
 
-        , putDelegationCertificate = \pk cert sl -> ExceptT $ do
-            cert `deepseq` sl `deepseq`
-                alterDB errNoSuchWallet db (mPutDelegationCertificate pk cert sl)
+        , putDelegationCertificate = \pk cert sl -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mPutDelegationCertificate pk cert sl
 
         , isStakeKeyRegistered =
             ExceptT . alterDB errNoSuchWallet db . mIsStakeKeyRegistered
@@ -134,8 +142,9 @@ newDBLayer timeInterpreter = do
                                      Tx History
         -----------------------------------------------------------------------}
 
-        , putTxHistory = \pk txh -> ExceptT $ do
-            txh `deepseq` alterDB errNoSuchWallet db (mPutTxHistory pk txh)
+        , putTxHistory = \pk txh -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mPutTxHistory pk txh
 
         , readTxHistory = \pk minWithdrawal order range mstatus ->
             readDB db $
@@ -168,20 +177,30 @@ newDBLayer timeInterpreter = do
                                        Keystore
         -----------------------------------------------------------------------}
 
-        , putPrivateKey = \pk prv -> ExceptT $ do
-            prv `deepseq` alterDB errNoSuchWallet db (mPutPrivateKey pk prv)
+        , putPrivateKey = \pk prv -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mPutPrivateKey pk prv
 
         , readPrivateKey = readDB db . mReadPrivateKey
 
         {-----------------------------------------------------------------------
                                        Pending Tx
         -----------------------------------------------------------------------}
 
-        , updatePendingTxForExpiry = \pk tip -> ExceptT $ do
-            alterDB errNoSuchWallet db (mUpdatePendingTxForExpiry pk tip)
+        , putLocalTxSubmission = \pk txid tx sl -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mPutLocalTxSubmission pk txid tx sl
 
-        , removePendingOrExpiredTx = \pk tid -> ExceptT $ do
-            alterDB errCannotRemovePendingTx db (mRemovePendingOrExpiredTx pk tid)
+        , readLocalTxSubmissionPending =
+            readDB db . mReadLocalTxSubmissionPending
+
+        , updatePendingTxForExpiry = \pk tip -> ExceptT $
+            alterDB errNoSuchWallet db $
+            mUpdatePendingTxForExpiry pk tip
+
+        , removePendingOrExpiredTx = \pk tid -> ExceptT $
+            alterDB errCannotRemovePendingTx db $
+            mRemovePendingOrExpiredTx pk tid
 
         {-----------------------------------------------------------------------
                                  Protocol Parameters
@@ -208,7 +227,8 @@ newDBLayer timeInterpreter = do
 
 -- | Apply an operation to the model database, then update the mutable variable.
 alterDB
-    :: (Err (PrimaryKey WalletId) -> Maybe err)
+    :: (NFData s, NFData xprv)
+    => (Err (PrimaryKey WalletId) -> Maybe err)
     -- ^ Error type converter
     -> MVar (Database (PrimaryKey WalletId) s xprv)
     -- ^ The database variable
@@ -218,14 +238,15 @@ alterDB
 alterDB convertErr db op = modifyMVar db (bubble . op)
   where
     bubble (Left e, db') = case convertErr e of
-        Just e' -> pure (db', Left e')
+        Just e' -> pure (force db', Left e')
         Nothing -> throwIO $ MVarDBError e
-    bubble (Right a, db') = pure (db', Right a)
+    bubble (Right a, db') = pure (force db', Right a)
 
 -- | Run a query operation on the model database. Any error results are turned
 -- into a runtime exception.
 readDB
-    :: MVar (Database (PrimaryKey WalletId) s xprv)
+    :: (NFData s, NFData xprv)
+    => MVar (Database (PrimaryKey WalletId) s xprv)
     -- ^ The database variable
     -> ModelOp (PrimaryKey WalletId) s xprv a
     -- ^ Operation to run on the database