@@ -565,22 +565,14 @@ executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = r
565565autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
566566autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
567567 let sessions = ipSessions conf
568− tvar = ipTransactionGraph conf
569568 id1 <- nextRandom
570569 id2 <- nextRandom
571− atomically $ do
570+ commitLock_ sessionId conf $ \graph -> do
572571 eSession <- sessionForSessionId sessionId sessions
573572 case eSession of
574573 Left err -> pure (Left err)
575574 Right session -> do
576− graph <- readTVar tvar
577− case Graph.autoMergeToHead (id1, id2) (Sess.disconnectedTransaction session) headName' strat graph of
578− Left err -> pure (Left err)
579− Right (discon', graph') -> do
580− writeTVar tvar graph'
581− let newSession = Session discon' (Sess.schemaName session)
582− STMMap.insert newSession sessionId sessions
583− execute
575+ pure $ Graph.autoMergeToHead (id1, id2) (Sess.disconnectedTransaction session) headName' strat graph
584576autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
585577
586578-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr` modify the DatabaseContext but cannot be purely implemented.
@@ -677,7 +669,9 @@ executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excMaybe $ do
677669 case eRefreshedGraph of
678670 Left err -> pure (Left (DatabaseLoadError err))
679671 Right refreshedGraph -> do
672+ --snip it
680673 eGraph <- executeGraphExprSTM_ dbWrittenByOtherProcess freshId sessionId session sessions graphExpr refreshedGraph graphTvar
674+ --snip it
681675 case eGraph of
682676 Left err -> pure (Left err)
683677 Right newGraph -> do
@@ -921,3 +915,72 @@ disconnectedTransaction_ sessionId (InProcessConnection conf) = do
921915 Nothing -> error "No such session"
922916 Just (Sess.Session discon _) -> pure discon
923917disconnectedTransaction_ _ _= error "remote connection used"
918+
919+-- wrap an commit-based graph evaluation in file locking
920+commitLock_ :: SessionId ->
921+ InProcessConnectionConf ->
922+ (TransactionGraph ->
923+ STM (Either RelationalError (DisconnectedTransaction, TransactionGraph))) ->
924+ IO (Either RelationalError ())
925+commitLock_ sessionId conf stmBlock = do
926+ let sessions = ipSessions conf
927+ strat = ipPersistenceStrategy conf
928+ mScriptSession = ipScriptSession conf
929+ graphTvar = ipTransactionGraph conf
930+ clientNodes = ipClientNodes conf
931+ mLockFileH = ipLocks conf
932+ lockHandler body = case mLockFileH of
933+ Nothing -> body False
934+ Just (lockFileH, lockMVar) ->
935+ let acquireLocks = do
936+ lastWrittenDigest <- takeMVar lockMVar
937+ lockFile lockFileH WriteLock
938+ latestDigest <- readGraphTransactionIdDigest strat
939+ pure (latestDigest /= lastWrittenDigest)
940+ releaseLocks _ = do
941+ --still holding the lock- get the latest digest
942+ gDigest <- readGraphTransactionIdDigest strat
943+ unlockFile lockFileH
944+ putMVar lockMVar gDigest
945+ in bracket acquireLocks releaseLocks body
946+ manip <- lockHandler $ \dbWrittenByOtherProcess -> atomically $ do
947+ eSession <- sessionForSessionId sessionId sessions
948+ --handle graph update by other process
949+ oldGraph <- readTVar graphTvar
950+ case eSession of
951+ Left err -> pure (Left err)
952+ Right session -> do
953+ let dbdir = case strat of
954+ MinimalPersistence x -> x
955+ CrashSafePersistence x -> x
956+ _ -> error "accessing dbdir on non-persisted connection"
957+ eRefreshedGraph <- if dbWrittenByOtherProcess then
958+ unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
959+ else
960+ pure (Right oldGraph)
961+ case eRefreshedGraph of
962+ Left err -> pure (Left (DatabaseLoadError err))
963+ Right refreshedGraph -> do
964+ eGraph <- stmBlock refreshedGraph
965+ case eGraph of
966+ Left err -> pure (Left err)
967+ Right (discon', graph') -> do
968+ writeTVar graphTvar graph'
969+ let newSession = Session discon' (Sess.schemaName session)
970+ STMMap.insert newSession sessionId sessions
971+ case transactionForId (Sess.parentId session) oldGraph of
972+ Left err -> pure $ Left err
973+ Right previousTrans -> do
974+ (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
975+ nodesToNotify <- toList (STMSet.stream nodes)
976+ pure $ Right (evaldNots, nodesToNotify, graph')
977+ --handle notification firing
978+ case manip of
979+ Left err -> pure (Left err)
980+ Right (notsToFire, nodesToNotify, newGraph) -> do
981+ --update filesystem database, if necessary
982+ processTransactionGraphPersistence strat newGraph
983+ sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
984+ pure (Right ())
985+
986+