00c9dc7 · project-m36 · git.iko.soy

git.iko.soy/project-m36

on master help sign in

3noch found that automerge was not writing all new transactions to disk

00c9dc7AgentM9 years ago

project-m36.cabal+1−1

compare with master

unifiedsplit

@@ -482,7 +482,7 @@ Test-Suite test-multiprocess-access

482 Hs-Source-Dirs: ./src/bin, ., test/

483 CPP-Options: -DPROJECTM36_VERSION="v0.1"

484

485−Test-Suite test-automerge

485+Test-Suite test-transactiongraph-automerge

486 Default-Language: Haskell2010

487 type: exitcode-stdio-1.0

488 main-is: test/TransactionGraph/Automerge.hs

src/lib/ProjectM36/Client.hs+21−15

@@ -128,6 +128,7 @@ import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)

128import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)

129import Control.Distributed.Process (NodeId(..), reconnect)

130

131+import Data.Either (isRight)

131import Data.UUID.V4 (nextRandom)

132import Data.UUID.V4 (nextRandom)

132import Data.Word

133import Data.Word

133import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)

134import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)

@@ -580,7 +581,8 @@ autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do

580 Right session -> do

581 Right session -> do

581 case Graph.autoMergeToHead (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph of

582 case Graph.autoMergeToHead (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph of

582 Left err -> pure (Left err)

583 Left err -> pure (Left err)

584+ Right (discon', graph') -> do

583− Right (discon', graph') -> pure (Right (discon', graph', True))

585+ pure (Right (discon', graph', [id1, id2, id3]))

584autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')

586autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')

585

587

586-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.

588-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.

@@ -645,7 +647,13 @@ executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do

645 let discon = Sess.disconnectedTransaction session

647 let discon = Sess.disconnectedTransaction session

646 case evalGraphOp freshId discon updatedGraph graphExpr of

648 case evalGraphOp freshId discon updatedGraph graphExpr of

647 Left err -> pure (Left err)

649 Left err -> pure (Left err)

650+ Right (discon', graph') -> do

651+ --if freshId appears in the graph, then we need to pass it on

652+ let transIds = if isRight (transactionForId freshId graph') then

653+ [freshId]

654+ else

655+ []

648− Right (discon', graph') -> pure (Right (discon', graph', isCommit graphExpr))

656+ pure (Right (discon', graph', transIds))

649{-

657{-

650executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do

658executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do

651 let strat = ipPersistenceStrategy conf

659 let strat = ipPersistenceStrategy conf

@@ -768,10 +776,10 @@ rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId con

768rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)

776rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)

769

777

770-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.

778-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.

771−processTransactionGraphPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

779+processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()

772−processTransactionGraphPersistence NoPersistence _ = pure ()

780+processTransactionGraphPersistence NoPersistence _ _ = pure ()

773−processTransactionGraphPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph >> pure ()

781+processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()

774−processTransactionGraphPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph >> pure ()

782+processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()

775

783

776readGraphTransactionIdDigest :: PersistenceStrategy -> IO (LockFileHash)

784readGraphTransactionIdDigest :: PersistenceStrategy -> IO (LockFileHash)

777readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"

785readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"

@@ -947,13 +955,11 @@ disconnectedTransaction_ sessionId (InProcessConnection conf) = do

947 Just (Sess.Session discon _) -> pure discon

955 Just (Sess.Session discon _) -> pure discon

948disconnectedTransaction_ _ _= error "remote connection used"

956disconnectedTransaction_ _ _= error "remote connection used"

949

957

950−type ProcessCommit = Bool

951−

952-- wrap a graph evaluation in file locking

958-- wrap a graph evaluation in file locking

953commitLock_ :: SessionId ->

959commitLock_ :: SessionId ->

954 InProcessConnectionConf ->

960 InProcessConnectionConf ->

955 (TransactionGraph ->

961 (TransactionGraph ->

956− STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, ProcessCommit))) ->

962+ STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->

957 IO (Either RelationalError ())

963 IO (Either RelationalError ())

958commitLock_ sessionId conf stmBlock = do

964commitLock_ sessionId conf stmBlock = do

959 let sessions = ipSessions conf

965 let sessions = ipSessions conf

@@ -998,25 +1004,25 @@ commitLock_ sessionId conf stmBlock = do

998 eGraph <- stmBlock refreshedGraph

1004 eGraph <- stmBlock refreshedGraph

999 case eGraph of

1005 case eGraph of

1000 Left err -> pure (Left err)

1006 Left err -> pure (Left err)

1001− Right (discon', graph', processCommit) -> do

1007+ Right (discon', graph', transactionIdsToPersist) -> do

1002 writeTVar graphTvar graph'

1008 writeTVar graphTvar graph'

1003 let newSession = Session discon' (Sess.schemaName session)

1009 let newSession = Session discon' (Sess.schemaName session)

1004 STMMap.insert newSession sessionId sessions

1010 STMMap.insert newSession sessionId sessions

1005 case transactionForId (Sess.parentId session) oldGraph of

1011 case transactionForId (Sess.parentId session) oldGraph of

1006 Left err -> pure $ Left err

1012 Left err -> pure $ Left err

1007 Right previousTrans -> do

1013 Right previousTrans -> do

1008− if processCommit then do

1014+ if not (Prelude.null transactionIdsToPersist) then do

1009 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes

1015 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes

1010 nodesToNotify <- toList (STMSet.stream nodes)

1016 nodesToNotify <- toList (STMSet.stream nodes)

1011− pure $ Right (evaldNots, nodesToNotify, graph')

1017+ pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)

1012− else pure (Right (M.empty, [], graph'))

1018+ else pure (Right (M.empty, [], graph', []))

1013

1019

1014 --handle notification firing

1020 --handle notification firing

1015 case manip of

1021 case manip of

1016 Left err -> pure (Left err)

1022 Left err -> pure (Left err)

1017− Right (notsToFire, nodesToNotify, newGraph) -> do

1023+ Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do

1018 --update filesystem database, if necessary

1024 --update filesystem database, if necessary

1019− processTransactionGraphPersistence strat newGraph

1025+ processTransactionGraphPersistence strat transactionIdsToPersist newGraph

1020 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire

1026 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire

1021 pure (Right ())

1027 pure (Right ())

1022{-

1028{-

src/lib/ProjectM36/TransactionGraph.hs+3−1

@@ -440,12 +440,13 @@ createUnionMergeTransaction newId strategy graph (t1,t2) = do

440 atomFuncs <- unionMergeAtomFunctions preference (atomFunctions contextA) (atomFunctions contextB)

441 notifs <- unionMergeMaps preference (notifications contextA) (notifications contextB)

442 types <- unionMergeTypeConstructorMapping preference (typeConstructorMapping contextA) (typeConstructorMapping contextB)

443+ dbcFuncs <- unionMergeDatabaseContextFunctions preference (dbcFunctions contextA) (dbcFunctions contextB)

443 -- TODO: add merge of subschemas

444 -- TODO: add merge of subschemas

444 let newContext = DatabaseContext {

445 let newContext = DatabaseContext {

445 inclusionDependencies = incDeps,

446 inclusionDependencies = incDeps,

446 relationVariables = relVars,

447 relationVariables = relVars,

447 atomFunctions = atomFuncs,

448 atomFunctions = atomFuncs,

448− dbcFunctions = undefined,

449+ dbcFunctions = dbcFuncs,

449 notifications = notifs,

450 notifications = notifs,

450 typeConstructorMapping = types

451 typeConstructorMapping = types

451 }

452 }

@@ -507,6 +508,7 @@ autoMergeToHead (tempBranchTransId, tempCommitTransId, mergeTransId) discon merg

507 let rel = runReader (evalRelationalExpr (RelationVariable "s" ())) (mkRelationalExprState $ D.concreteDatabaseContext discon'''')

508 let rel = runReader (evalRelationalExpr (RelationVariable "s" ())) (mkRelationalExprState $ D.concreteDatabaseContext discon'''')

508 traceShowM rel

509 traceShowM rel

509-}

510-}

511+

510 pure (discon'''', graph'''')

512 pure (discon'''', graph'''')

511

513

512

514

src/lib/ProjectM36/TransactionGraph/Merge.hs+5

@@ -83,3 +83,8 @@ unionMergeTypeConstructorMapping prefer typesA typesB = do

83 PreferNeither -> Left StrategyViolatesTypeConstructorMergeError

84 ) [] (S.toList allFuncNames)

86+unionMergeDatabaseContextFunctions :: MergePreference -> DatabaseContextFunctions -> DatabaseContextFunctions -> Either MergeError DatabaseContextFunctions

87+unionMergeDatabaseContextFunctions prefer funcsA funcsB = case prefer of

88+ PreferFirst -> pure $ HS.union funcsA funcsB

89+ PreferSecond -> pure $ HS.union funcsB funcsA

90+ PreferNeither -> pure $ HS.union funcsA funcsB

src/lib/ProjectM36/TransactionGraph/Persist.hs+15−8

@@ -70,7 +70,8 @@ bootstrapDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (Handle,

70bootstrapDatabaseDir sync dbdir bootstrapGraph = do

71 createDirectory dbdir

72 lockFileH <- openLockFile dbdir

73+ let allTransIds = map transactionId (S.toList (transactionsForGraph bootstrapGraph))

73− digest <- bracket_ (lockFile lockFileH WriteLock) (unlockFile lockFileH) (transactionGraphPersist sync dbdir bootstrapGraph)

74+ digest <- bracket_ (lockFile lockFileH WriteLock) (unlockFile lockFileH) (transactionGraphPersist sync dbdir allTransIds bootstrapGraph)

74 pure (lockFileH, digest)

75 pure (lockFileH, digest)

76openLockFile :: FilePath -> IO (Handle)

77openLockFile :: FilePath -> IO (Handle)

@@ -85,19 +86,25 @@ incrementally updates an existing database directory

85-assume that all non-head transactions have already been written because this is an incremental (and concurrent!) write method

86-assume that all non-head transactions have already been written because this is an incremental (and concurrent!) write method

86--store the head names with a symlink to the transaction under "heads"

87--store the head names with a symlink to the transaction under "heads"

87-}

88-}

88−transactionGraphPersist :: DiskSync -> FilePath -> TransactionGraph -> IO LockFileHash

89+transactionGraphPersist :: DiskSync -> FilePath -> [TransactionId] -> TransactionGraph -> IO LockFileHash

89−transactionGraphPersist sync destDirectory graph = do

90+transactionGraphPersist sync destDirectory transIds graph = do

90− transactionHeadTransactionsPersist sync destDirectory graph

91+ transactionsPersist sync transIds destDirectory graph

91 --write graph file

92 --write graph file

92 newDigest <- writeGraphTransactionIdFile sync destDirectory graph

93 newDigest <- writeGraphTransactionIdFile sync destDirectory graph

93 --write heads file

94 --write heads file

94 transactionGraphHeadsPersist sync destDirectory graph

95 transactionGraphHeadsPersist sync destDirectory graph

95 pure newDigest

96 pure newDigest

97−-- | The incremental writer which only writes from the set of heads. New heads must be written on every commit. Most heads will already be written on every commit.

98+-- | The incremental writer writes the transactions ids specified by the second argument.

99+

100+-- There was a bug here via #128 because automerge added multiple transactions to the graph but this function used to only write the head transactions from the graph. Automerge creates multiple transactions, so these are now passed in as the second argument.

98−transactionHeadTransactionsPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()

101+transactionsPersist :: DiskSync -> [TransactionId] -> FilePath -> TransactionGraph -> IO ()

99−transactionHeadTransactionsPersist sync destDirectory graphIn = mapM_ (writeTransaction sync destDirectory) $ M.elems (transactionHeadsForGraph graphIn)

102+transactionsPersist sync transIds destDirectory graphIn = mapM_ writeTrans transIds

100−

103+ where writeTrans tid = do

104+ case transactionForId tid graphIn of

105+ Left err -> error ("writeTransaction: " ++ show err)

106+ Right trans -> writeTransaction sync destDirectory trans

107+

101{-

108{-

102write graph heads to a file which can be atomically swapped

109write graph heads to a file which can be atomically swapped

103-}

110-}

test/TransactionGraph/Automerge.hs+21−1

@@ -6,6 +6,9 @@ import qualified Data.Set as S

6import TutorialD.Interpreter.TestBase

8import System.Exit

9+import System.IO.Temp

10+import System.FilePath

11+import Control.Exception.Base

10main :: IO ()

13main :: IO ()

11main = do

14main = do

@@ -14,7 +17,8 @@ main = do

15testList :: Test

18testList :: Test

16testList = TestList [testAutomergeSuccess,

19testList = TestList [testAutomergeSuccess,

17− testAutomergeFailure]

20+ testAutomergeFailure,

21+ testAutomergeReconnect]

19checkEither :: IO (Either RelationalError a) -> IO a

23checkEither :: IO (Either RelationalError a) -> IO a

20checkEither io = do

24checkEither io = do

@@ -67,5 +71,21 @@ testAutomergeFailure = TestCase $ do

68 assertEqual "merge failure" (Left (InclusionDependencyCheckError "s_pkey")) mergeRes

72 assertEqual "merge failure" (Left (InclusionDependencyCheckError "s_pkey")) mergeRes

74+--reported as #128

75+testAutomergeReconnect :: Test

76+testAutomergeReconnect = TestCase $ withSystemTempDirectory "m36testdb" $ \tempdir -> do

77+ let repro = do

78+ conn <- unsafeLeftCrash =<< connectProjectM36 (InProcessConnectionInfo (CrashSafePersistence (tempdir </> "test.db")) emptyNotificationCallback [])

79+ sess <- unsafeLeftCrash =<< createSessionAtHead conn "master"

80+ autoMergeToHead sess conn UnionMergeStrategy "master"

81+ -- commit sess conn

82+

83+ unsafeLeftCrash :: Show e => Either e a -> IO a

84+ unsafeLeftCrash = either (throwIO . userError . show) pure

85+ _ <- repro

86+ _ <- repro

87+ pure ()

88+

89+

test/TransactionGraph/Persist.hs+1−1

@@ -57,7 +57,7 @@ testDBSimplePersistence = TestCase $ withSystemTempDirectory "m36testdb" $ \temp

57 Left err -> assertFailure (show err)

58 Right (_, graph') -> do

59 --persist the new graph

60− _ <- transactionGraphPersist NoDiskSync dbdir graph'

60+ _ <- transactionGraphPersist NoDiskSync dbdir [freshId'] graph'

61 --reload the graph from the filesystem and confirm that the transaction is present

62 graphErr <- transactionGraphLoad dbdir emptyTransactionGraph Nothing

63 let mapEq graphArg = S.map transactionId (transactionsForGraph graphArg)