[CBR-475] Fix race conditions in SQlite

pkgs/default.nix

@@ -16985,6 +16985,7 @@ x509-store
 testHaskellDepends = [
 acid-state
 aeson
+async
 base
 bytestring
 cardano-crypto

wallet-new/cardano-sl-wallet-new.cabal

@@ -474,6 +474,7 @@ test-suite wallet-unit-tests
   hs-source-dirs:     server test/unit
 
   build-depends:      acid-state
+                    , async
                     , base
                     , bytestring
                     , cardano-crypto

wallet-new/src/Cardano/Wallet/Kernel/DB/Sqlite.hs

@@ -66,7 +66,7 @@ import qualified Data.Foldable as Foldable
 import           Data.List.NonEmpty (NonEmpty (..), nonEmpty)
 import qualified Data.List.NonEmpty as NonEmpty
 import qualified Data.Map as M
-import           Data.Time.Units (Second, fromMicroseconds, toMicroseconds)
+import           Data.Time.Units (fromMicroseconds, toMicroseconds)
 import           Database.Beam.Migrate (CheckedDatabaseSettings, DataType (..),
                      Migration, MigrationSteps, boolean, collectChecks,
                      createTable, evaluateDatabase, executeMigration, field,
@@ -79,7 +79,6 @@ import           Cardano.Wallet.Kernel.DB.TxMeta.Types (AccountFops (..),
                      FilterOperation (..), Limit (..), Offset (..),
                      SortCriteria (..), SortDirection (..), Sorting (..))
 import qualified Cardano.Wallet.Kernel.DB.TxMeta.Types as Kernel
-import           Cardano.Wallet.WalletLayer.ExecutionTimeLimit
 
 import qualified Pos.Chain.Txp as Txp
 import qualified Pos.Core as Core
@@ -600,7 +599,7 @@ getTxMetas conn (Offset offset) (Limit limit) accountFops mbAddress fopTxId fopT
         Left e -> throwIO $ Kernel.StorageFailure (toException e)
         Right Nothing -> return ([], Just 0)
         Right (Just (meta, inputs, outputs)) ->  do
-            eiCount <- limitExecutionTimeTo (25 :: Second) (\ _ -> ()) $ ignoreLeft $ Sqlite.runDBAction $ runBeamSqlite conn $
+            eiCount <- Sqlite.runDBAction $ runBeamSqlite conn $
                 case mbAddress of
                     Nothing   -> SQL.runSelectReturningOne $ SQL.select metaQueryC
                     Just addr -> SQL.runSelectReturningOne $ SQL.select $ metaQueryWithAddrC addr
@@ -613,13 +612,6 @@ getTxMetas conn (Offset offset) (Limit limit) accountFops mbAddress fopTxId fopT
             return (txMeta, count)
 
     where
-        ignoreLeft :: IO (Either a b) -> IO (Either () b)
-        ignoreLeft m  = do
-            x <- m
-            case x of
-                Left _  -> return $ Left ()
-                Right r -> return $ Right r
-
         filters meta = do
             SQL.guard_ $ filterAccs meta accountFops
             SQL.guard_ $ applyFilter (_txMetaTableId meta) fopTxId

wallet-new/src/Cardano/Wallet/Kernel/DB/TxMeta.hs

@@ -7,22 +7,27 @@ module Cardano.Wallet.Kernel.DB.TxMeta (
   , openMetaDB
   ) where
 
+import           Universum
+
+import           Control.Concurrent.MVar (withMVar)
+
 import qualified Cardano.Wallet.Kernel.DB.Sqlite as ConcreteStorage
 import           Cardano.Wallet.Kernel.DB.TxMeta.Types as Types
-import           Universum
 
 -- Concrete instantiation of 'MetaDBHandle'
 
 openMetaDB :: FilePath -> IO MetaDBHandle
 openMetaDB fp = do
     conn <- ConcreteStorage.newConnection fp
+    lock <- newMVar conn
     return MetaDBHandle {
-          closeMetaDB   = ConcreteStorage.closeMetaDB conn
-        , migrateMetaDB = ConcreteStorage.unsafeMigrateMetaDB conn
-        , clearMetaDB   = ConcreteStorage.clearMetaDB conn
-        , getTxMeta     = ConcreteStorage.getTxMeta conn
-        , putTxMeta     = ConcreteStorage.putTxMeta conn
-        , putTxMetaT    = ConcreteStorage.putTxMetaT conn
-        , getAllTxMetas = ConcreteStorage.getAllTxMetas conn
-        , getTxMetas    = ConcreteStorage.getTxMetas conn
+          closeMetaDB   = withMVar lock ConcreteStorage.closeMetaDB
+        , migrateMetaDB = withMVar lock ConcreteStorage.unsafeMigrateMetaDB
+        , clearMetaDB   = withMVar lock ConcreteStorage.clearMetaDB
+        , getTxMeta     = \t w a -> withMVar lock $ \c -> ConcreteStorage.getTxMeta c t w a
+        , putTxMeta     = \ t    -> withMVar lock $ \c -> ConcreteStorage.putTxMeta c t
+        , putTxMetaT    = \ t    -> withMVar lock $ \c -> ConcreteStorage.putTxMetaT c t
+        , getAllTxMetas = withMVar lock ConcreteStorage.getAllTxMetas
+        , getTxMetas    = \o l af w t time s
+                                 -> withMVar lock $ \c -> ConcreteStorage.getTxMetas c o l af w t time s
         }

wallet-new/test/unit/TxMetaStorageSpecs.hs

@@ -11,6 +11,7 @@ import           Universum
 
 import qualified Cardano.Wallet.Kernel.DB.Sqlite as SQlite
 import           Cardano.Wallet.Kernel.DB.TxMeta
+import           Control.Concurrent.Async
 import           Control.Exception.Safe (bracket)
 import qualified Data.List as List
 import qualified Data.List.NonEmpty as NonEmpty
@@ -25,7 +26,7 @@ import           Serokell.Util.Text (listJsonIndent, pairF)
 import           Test.Hspec (expectationFailure, shouldContain, shouldThrow)
 import           Test.Hspec.QuickCheck (prop)
 import           Test.QuickCheck (Arbitrary, Gen, arbitrary, forAll, suchThat,
-                     vectorOf)
+                     vectorOf, withMaxSuccess)
 import           Test.QuickCheck.Monadic (assert, monadicIO, pick, run)
 import           Util.Buildable (ShowThroughBuild (..))
 import           Util.Buildable.Hspec
@@ -167,9 +168,94 @@ sortByCreationAt direction = sortBy sortFn
 hasDupes :: Ord a => [a] -> Bool
 hasDupes xs = length (Set.fromList xs) /= List.length xs
 
+
+threadRead :: Int -> MetaDBHandle -> IO ()
+threadRead times hdl = do
+    let getNoFilters = getTxMetas hdl (Offset 0) (Limit 100) Everything Nothing NoFilterOp NoFilterOp Nothing
+    replicateM_ times getNoFilters
+
+threadWrite :: [TxMeta] -> MetaDBHandle -> IO ()
+threadWrite metas hdl = do
+    let f meta = do
+            putTxMetaT hdl meta `shouldReturn` Tx
+    mapM_ f metas
+
+-- here we try to add the same tx 2 times. The second must fail, but without crashing
+-- anything, as this is a no-op.
+threadWriteWithNoOp :: [TxMeta] -> MetaDBHandle -> IO ()
+threadWriteWithNoOp metas hdl = do
+    let f meta = do
+            putTxMetaT hdl meta `shouldReturn` Tx
+            putTxMetaT hdl meta `shouldReturn` No
+    mapM_ f metas
+
 -- | Specs which tests the persistent storage and API provided by 'TxMeta'.
 txMetaStorageSpecs :: Spec
 txMetaStorageSpecs = do
+    describe "synchronization" $ do
+        it "synchronized with 2 write workers and no-ops" $ withMaxSuccess 5 $  monadicIO $ do
+            -- beware of the big data.
+            testMetas <- pick (genMetas 2000)
+            let metas = unSTB <$> testMetas
+                (meta0, meta1) = splitAt (div 2000 2) metas
+            run $ withTemporaryDb $ \hdl -> do
+                t0 <- async $ threadWriteWithNoOp meta0 hdl
+                t1 <- async $ threadWriteWithNoOp meta1 hdl
+                traverse_ wait [t0, t1]
+
+        it "synchronized with 2 write workers and no-ops: correct count" $ withMaxSuccess 5 $  monadicIO $ do
+            -- beware of the big data.
+            testMetas <- pick (genMetas 200)
+            let metas = unSTB <$> testMetas
+                (meta0, meta1) = splitAt (div 200 2) metas
+            run $ withTemporaryDb $ \hdl -> do
+                t0 <- async $ threadWriteWithNoOp meta0 hdl
+                t1 <- async $ threadWriteWithNoOp meta1 hdl
+                traverse_ wait [t0, t1]
+                (ls, _count) <- getTxMetas hdl (Offset 0) (Limit 300) Everything Nothing NoFilterOp NoFilterOp Nothing
+                length ls `shouldBe` 200
+
+        it "synchronized with 2 write workers" $ withMaxSuccess 5 $  monadicIO $ do
+            -- beware of the big data.
+            testMetas <- pick (genMetas 2000)
+            let metas = unSTB <$> testMetas
+                (meta0, meta1) = splitAt (div 2000 2) metas
+            run $ withTemporaryDb $ \hdl -> do
+                t0 <- async $ threadWrite meta0 hdl
+                t1 <- async $ threadWrite meta1 hdl
+                traverse_ wait [t0, t1]
+
+        it "synchronized with 2 write workers: correct count" $ withMaxSuccess 5 $  monadicIO $ do
+            -- beware of the big data.
+            testMetas <- pick (genMetas 200)
+            let metas = unSTB <$> testMetas
+                (meta0, meta1) = splitAt (div 200 2) metas
+            run $ withTemporaryDb $ \hdl -> do
+                t0 <- async $ threadWrite meta0 hdl
+                t1 <- async $ threadWrite meta1 hdl
+                traverse_ wait [t0, t1]
+                (ls, _count) <- getTxMetas hdl (Offset 0) (Limit 300) Everything Nothing NoFilterOp NoFilterOp Nothing
+                length ls `shouldBe` 200
+
+        it "synchronized 1 write and 1 read workers" $ withMaxSuccess 5 $  monadicIO $ do
+            -- beware of the big data.
+            testMetas <- pick (genMetas 2000)
+            let metas = unSTB <$> testMetas
+            run $ withTemporaryDb $ \hdl -> do
+                t0 <- async $ threadWriteWithNoOp metas hdl
+                t1 <- async $ threadRead 2000 hdl
+                traverse_ wait [t0, t1]
+
+        it "synchronized 1 write and 1 read workers: correct count" $ withMaxSuccess 5 $  monadicIO $ do
+            -- beware of the big data.
+            testMetas <- pick (genMetas 200)
+            let metas = unSTB <$> testMetas
+            run $ withTemporaryDb $ \hdl -> do
+                t0 <- async $ threadWriteWithNoOp metas hdl
+                t1 <- async $ threadRead 200 hdl
+                traverse_ wait [t0, t1]
+                (ls, _count) <- getTxMetas hdl (Offset 0) (Limit 300) Everything Nothing NoFilterOp NoFilterOp Nothing
+                length ls `shouldBe` 200
 
     describe "SQlite transactions" $ do
         it "throws an exception when tx with double spending" $ monadicIO $ do