@@ -8,6 +8,7 @@ module ProjectM36.Client
88 ConnectionError,
99 connectProjectM36,
1010 close,
11+ closeRemote_,
1112 executeRelationalExpr,
1213 executeDatabaseContextExpr,
1314 executeDatabaseContextIOExpr,
@@ -47,6 +48,8 @@ module ProjectM36.Client
4748 callTestTimeout_,
4849 RelationCardinality(..),
4950 TransactionGraphOperator(..),
51+ transactionGraph_,
52+ disconnectedTransaction_,
5053 TransGraphRelationalExpr,
5154 TransactionIdLookup(..),
5255 TransactionIdHeadBacktrack(..),
@@ -83,11 +86,12 @@ import qualified ProjectM36.DisconnectedTransaction as Discon
8386import ProjectM36.Relation (typesAsRelation)
8487import ProjectM36.AtomFunctionBody (initScriptSession, ScriptSession)
8588import qualified ProjectM36.Relation as R
89+import Network.Transport (Transport(closeTransport))
8690import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
87−import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess)
91+import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess, closeLocalNode)
8892import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
8993import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
90−import Control.Distributed.Process (NodeId(..))
94+import Control.Distributed.Process (NodeId(..), reconnect)
9195
9296import Data.UUID.V4 (nextRandom)
9397import Control.Concurrent.STM
@@ -108,8 +112,6 @@ import GHC.Generics (Generic)
108112import Control.DeepSeq (force)
109113import System.IO
110114
111−--import Debug.Trace
112−
113115type Hostname = String
114116
115117type Port = Word16
@@ -165,8 +167,25 @@ defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (c
165167-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
166168type ClientNodes = STMSet.Set ProcessId
167169
168−data Connection = InProcessConnection PersistenceStrategy ClientNodes Sessions (TVar TransactionGraph) (Maybe ScriptSession) |
169− RemoteProcessConnection LocalNode ProcessId
170+-- internal structure specific to in-process connections
171+data InProcessConnectionConf = InProcessConnectionConf {
172+ ipPersistenceStrategy :: PersistenceStrategy,
173+ ipClientNodes :: ClientNodes,
174+ ipSessions :: Sessions,
175+ ipTransactionGraph :: TVar TransactionGraph,
176+ ipScriptSession :: Maybe ScriptSession,
177+ ipLocalNode :: LocalNode,
178+ ipTransport :: Transport -- we hold onto this so that we can close it gracefully
179+ }
180+
181+data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
182+ rLocalNode :: LocalNode,
183+ rProcessId :: ProcessId, --remote processId
184+ rTransport :: Transport-- the TCP socket transport
185+ }
186+
187+data Connection = InProcessConnection InProcessConnectionConf |
188+ RemoteProcessConnection RemoteProcessConnectionConf
170189
171190-- | There are several reasons why a connection can fail.
172191data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
@@ -178,12 +197,14 @@ data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
178197remoteDBLookupName :: DatabaseName -> String
179198remoteDBLookupName = (++) "db-"
180199
181−commonLocalNode :: IO (Either ConnectionError LocalNode)
182−commonLocalNode = do
200+createLocalNode :: IO (LocalNode, Transport)
201+createLocalNode = do
183202 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
184203 case eLocalTransport of
185− Left err -> pure (Left $ IOExceptionError err)
186− Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right
204+ Left err -> error ("failed to create transport: " ++ show err)
205+ Right localTransport -> do
206+ localNode <- newLocalNode localTransport initRemoteTable
207+ pure (localNode, localTransport)
187208
188209notificationListener :: NotificationCallback -> Process ()
189210notificationListener callback = do
@@ -200,12 +221,8 @@ notificationListener callback = do
200221 --say "LISTENER THREAD EXIT"
201222 pure ()
202223
203−startNotificationListener :: NotificationCallback -> IO (ProcessId)
204−startNotificationListener callback = do
205− eLocalNode <- commonLocalNode
206− case eLocalNode of
207− Left err -> error ("Failed to start local notification listener: " ++ show err)
208− Right localNode -> forkProcess localNode (notificationListener callback)
224+startNotificationListener :: LocalNode -> NotificationCallback -> IO (ProcessId)
225+startNotificationListener localNode callback = forkProcess localNode (notificationListener callback)
209226
210227createScriptSession :: [String] -> IO (Maybe ScriptSession)
211228createScriptSession ghcPkgPaths = do
@@ -227,9 +244,17 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
227244 graphTvar <- newTVarIO freshGraph
228245 clientNodes <- STMSet.newIO
229246 sessions <- STMMap.newIO
230− notificationPid <- startNotificationListener notificationCallback
247+ (localNode, transport) <- createLocalNode
248+ notificationPid <- startNotificationListener localNode notificationCallback
231249 mScriptSession <- createScriptSession ghcPkgPaths
232− let conn = InProcessConnection strat clientNodes sessions graphTvar mScriptSession
250+ let conn = InProcessConnection (InProcessConnectionConf {
251+ ipPersistenceStrategy =strat,
252+ ipClientNodes = clientNodes,
253+ ipSessions = sessions,
254+ ipTransactionGraph = graphTvar,
255+ ipScriptSession = mScriptSession,
256+ ipLocalNode = localNode,
257+ ipTransport = transport })
233258 addClientNode conn notificationPid
234259 pure (Right conn)
235260 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
@@ -237,25 +262,23 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
237262
238263connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
239264 connStatus <- newEmptyMVar
240− eLocalNode <- commonLocalNode
241− notificationListenerPid <- startNotificationListener notificationCallback
265+ (localNode, transport) <- createLocalNode
242266 let dbName = remoteDBLookupName databaseName
243− putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
244− case eLocalNode of
245− Left err -> pure (Left err)
246− Right localNode -> do
247− runProcess localNode $ do
248− mServerProcessId <- whereisRemote serverNodeId dbName
249− case mServerProcessId of
250− Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
251− Just serverProcessId -> do
252− loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
253− if not loginConfirmation then
254− liftIO $ putMVar connStatus (Left LoginError)
255− else do
256− liftIO $ putMVar connStatus (Right $ RemoteProcessConnection localNode serverProcessId)
257− status <- takeMVar connStatus
258− pure status
267+ --putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
268+ notificationListenerPid <- startNotificationListener localNode notificationCallback
269+ runProcess localNode $ do
270+ --even a failed connection retains an open socket!
271+ mServerProcessId <- whereisRemote serverNodeId dbName
272+ case mServerProcessId of
273+ Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
274+ Just serverProcessId -> do
275+ loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
276+ if not loginConfirmation then
277+ liftIO $ putMVar connStatus (Left LoginError)
278+ else do
279+ liftIO $ putMVar connStatus (Right $ RemoteProcessConnection (RemoteProcessConnectionConf {rLocalNode = localNode, rProcessId = serverProcessId, rTransport = transport}))
280+ status <- takeMVar connStatus
281+ pure status
259282
260283connectPersistentProjectM36 :: PersistenceStrategy ->
261284 DiskSync ->
@@ -277,21 +300,33 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc
277300 sessions <- STMMap.newIO
278301 clientNodes <- STMSet.newIO
279302 mScriptSession <- createScriptSession ghcPkgPaths
280− let conn = InProcessConnection strat clientNodes sessions tvarGraph mScriptSession
281− notificationPid <- startNotificationListener notificationCallback
303+ (localNode, transport) <- createLocalNode
304+ let conn = InProcessConnection (InProcessConnectionConf {
305+ ipPersistenceStrategy = strat,
306+ ipClientNodes = clientNodes,
307+ ipSessions = sessions,
308+ ipTransactionGraph = tvarGraph,
309+ ipScriptSession = mScriptSession,
310+ ipLocalNode = localNode,
311+ ipTransport = transport
312+ })
313+
314+ notificationPid <- startNotificationListener localNode notificationCallback
282315 addClientNode conn notificationPid
283316 pure (Right conn)
284317
285318-- | Create a new session at the transaction id and return the session's Id.
286319createSessionAtCommit :: TransactionId -> Connection -> IO (Either RelationalError SessionId)
287−createSessionAtCommit commitId conn@(InProcessConnection _ _ _ _ _) = do
320+createSessionAtCommit commitId conn@(InProcessConnection _) = do
288321 newSessionId <- nextRandom
289322 atomically $ do
290323 createSessionAtCommit_ commitId newSessionId conn
291−createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)
324+createSessionAtCommit uuid conn@(RemoteProcessConnection _) = remoteCall conn (CreateSessionAtCommit uuid)
292325
293326createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
294−createSessionAtCommit_ commitId newSessionId (InProcessConnection _ _ sessions graphTvar _) = do
327+createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
328+ let sessions = ipSessions conf
329+ graphTvar = ipTransactionGraph conf
295330 graph <- readTVar graphTvar
296331 case transactionForId commitId graph of
297332 Left err -> pure (Left err)
@@ -303,39 +338,53 @@ createSessionAtCommit_ commitId newSessionId (InProcessConnection _ _ sessions g
303338 Nothing -> do
304339 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
305340 pure $ Right newSessionId
306−createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionAtCommit_ called on remote connection"
341+createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
307342
308343-- | Call 'createSessionAtHead' with a transaction graph's head's name to create a new session pinned to that head. This function returns a 'SessionId' which can be used in other function calls to reference the point in the transaction graph.
309344createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)
310−createSessionAtHead headn conn@(InProcessConnection _ _ _ graphTvar _) = do
345+createSessionAtHead headn conn@(InProcessConnection conf) = do
346+ let graphTvar = ipTransactionGraph conf
311347 newSessionId <- nextRandom
312348 atomically $ do
313349 graph <- readTVar graphTvar
314350 case transactionForHead headn graph of
315351 Nothing -> pure $ Left (NoSuchHeadNameError headn)
316352 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
317−createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)
353+createSessionAtHead headn conn@(RemoteProcessConnection _) = remoteCall conn (CreateSessionAtHead headn)
318354
319355-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
320356addClientNode :: Connection -> ProcessId -> IO ()
321−addClientNode (RemoteProcessConnection _ _) _ = error "addClientNode called on remote connection"
322−addClientNode (InProcessConnection _ clientNodes _ _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)
357+addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
358+addClientNode (InProcessConnection conf) newProcessId = atomically (STMSet.insert newProcessId (ipClientNodes conf))
323359
324360-- | Discards a session, eliminating any uncommitted changes present in the session.
325361closeSession :: SessionId -> Connection -> IO ()
326−closeSession sessionId (InProcessConnection _ _ sessions _ _) = do
327− atomically $ STMMap.delete sessionId sessions
328−closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)
362+closeSession sessionId (InProcessConnection conf) = do
363+ atomically $ STMMap.delete sessionId (ipSessions conf)
364+closeSession sessionId conn@(RemoteProcessConnection _) = remoteCall conn (CloseSession sessionId)
329365-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.
330366close :: Connection -> IO ()
331−close (InProcessConnection _ _ sessions _ _) = atomically $ do
367+close (InProcessConnection conf) = do
368+ atomically $ do
369+ let sessions = ipSessions conf
332370 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)
333371 pure ()
372+ closeLocalNode (ipLocalNode conf)
373+ closeTransport (ipTransport conf)
334374
335−close (RemoteProcessConnection localNode serverProcessId) = do
336− runProcessResult localNode $ do
337− call serverProcessId Logout
338−
375+close conn@(RemoteProcessConnection conf) = do
376+ _ <- (remoteCall conn Logout) :: IO Bool
377+ --putStrLn ("closing " ++ show (localNodeId (rLocalNode conf)))
378+
379+ closeLocalNode (rLocalNode conf)
380+ closeTransport (rTransport conf)
381+
382+--used only by the server EntryPoints
383+closeRemote_ :: Connection -> IO ()
384+closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
385+closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
386+
387+ --we need to actually close the localNode's connection to the remote
339388--within the database server, we must catch and handle all exception lest they take down the database process- this handling might be different for other use-cases
340389--exceptions should generally *NOT* be thrown from any Project:M36 code paths, but third-party code such as AtomFunction scripts could conceivably throw undefined, etc.
341390
@@ -367,14 +416,17 @@ safeLogin login procId = do
367416 Right val -> pure val
368417
369418remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
370−remoteCall (InProcessConnection _ _ _ _ _) _ = error "remoteCall called on local connection"
371−remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do
419+remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
420+remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
372421 ret <- safeCall serverProcessId arg
373422 case ret of
374423 Left err -> error ("server died: " ++ show err)
375424 Right ret' -> case ret' of
376425 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
377426 Right val -> pure val
427+ where
428+ localNode = rLocalNode conf
429+ serverProcessId = rProcessId conf
378430
379431sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
380432sessionForSessionId sessionId sessions = do
@@ -403,26 +455,29 @@ sessionAndSchema sessionId sessions = do
403455 Right schema -> pure (Right (session, schema))
404456
405457currentSchemaName :: SessionId -> Connection -> IO (Maybe SchemaName)
406−currentSchemaName sessionId (InProcessConnection _ _ sessions _ _) = atomically $ do
458+currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
459+ let sessions = ipSessions conf
407460 eSession <- sessionForSessionId sessionId sessions
408461 case eSession of
409462 Left _ -> pure Nothing
410463 Right session -> pure (Just (Sess.schemaName session))
411−currentSchemaName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
464+currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
412465
413466setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Maybe RelationalError)
414−setCurrentSchemaName sessionId (InProcessConnection _ _ sessions _ _) sname = atomically $ do
467+setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
468+ let sessions = ipSessions conf
415469 eSession <- sessionForSessionId sessionId sessions
416470 case eSession of
417471 Left _ -> pure Nothing
418472 Right session -> case Sess.setSchemaName sname session of
419473 Left err -> pure (Just err)
420474 Right newSession -> STMMap.insert newSession sessionId sessions >> pure Nothing
421−setCurrentSchemaName sessionId conn@(RemoteProcessConnection _ _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
475+setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
422476
423477-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
424478executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
425−executeRelationalExpr sessionId (InProcessConnection _ _ sessions _ _) expr = excEither $ atomically $ do
479+executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
480+ let sessions = ipSessions conf
426481 eSession <- sessionAndSchema sessionId sessions
427482 case eSession of
428483 Left err -> pure $ Left err
@@ -436,11 +491,12 @@ executeRelationalExpr sessionId (InProcessConnection _ _ sessions _ _) expr = ex
436491 Right expr'' -> case evalState (RE.evalRelationalExpr expr'') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session)) of
437492 Left err -> pure (Left err)
438493 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
439−executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
494+executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
440495
441496-- | Execute a database context expression in the context of the session and connection. Database expressions modify the current session's disconnected transaction but cannot modify the transaction graph.
442497executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Maybe RelationalError)
443−executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) expr = excMaybe $ atomically $ do
498+executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excMaybe $ atomically $ do
499+ let sessions = ipSessions conf
444500 eSession <- sessionAndSchema sessionId sessions
445501 case eSession of
446502 Left err -> pure $ Just err
@@ -461,12 +517,14 @@ executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) expr
461517 STMMap.insert newSession sessionId sessions
462518 pure Nothing
463519
464−executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
520+executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
465521
466522-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr` modify the DatabaseContext but cannot be purely implemented.
467523--this is almost completely identical to executeDatabaseContextExpr above
468524executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Maybe RelationalError)
469−executeDatabaseContextIOExpr sessionId (InProcessConnection _ _ sessions _ scriptSession) expr = excMaybe $ do
525+executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excMaybe $ do
526+ let sessions = ipSessions conf
527+ scriptSession = ipScriptSession conf
470528 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
471529 case eSession of
472530 Left err -> pure $ Just err
@@ -480,7 +538,7 @@ executeDatabaseContextIOExpr sessionId (InProcessConnection _ _ sessions _ scrip
480538 newSession = Session newDiscon (Sess.schemaName session)
481539 atomically $ STMMap.insert newSession sessionId sessions
482540 pure Nothing
483−executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
541+executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
484542
485543executeGraphExprSTM_ :: TransactionId -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
486544executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar = do
@@ -508,7 +566,11 @@ executeCommitExprSTM_ oldContext newContext nodes = do
508566
509567-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
510568executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
511−executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graphTvar _) graphExpr = excMaybe $ do
569+executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excMaybe $ do
570+ let strat = ipPersistenceStrategy conf
571+ clientNodes = ipClientNodes conf
572+ sessions = ipSessions conf
573+ graphTvar = ipTransactionGraph conf
512574 freshId <- nextRandom
513575 manip <- atomically $ do
514576 eSession <- sessionForSessionId sessionId sessions
@@ -535,9 +597,9 @@ executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graph
535597 --update filesystem database, if necessary
536598 --this should really grab a lock at the beginning of the method to be threadsafe
537599 processPersistence strat newGraph
538− sendNotifications nodesToNotify notsToFire
600+ sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
539601 return Nothing
540−executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
602+executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
541603
542604{-
543605commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
@@ -550,17 +612,19 @@ commitSTM_ freshUUID sessionId sessions graph = do
550612
551613-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
552614executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
553−executeTransGraphRelationalExpr _ (InProcessConnection _ _ _ graphTvar _) tgraphExpr = excEither . atomically $ do
615+executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
616+ let graphTvar = ipTransactionGraph conf
554617 graph <- readTVar graphTvar
555618 case evalTransGraphRelationalExpr tgraphExpr graph of
556619 Left err -> pure (Left err)
557620 Right relExpr -> case evalState (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState RE.emptyDatabaseContext) of
558621 Left err -> pure (Left err)
559622 Right rel -> pure (force (Right rel))
560−executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _ _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
623+executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
561624
562625executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Maybe RelationalError)
563−executeSchemaExpr sessionId (InProcessConnection _ _ sessions _ _) schemaExpr = atomically $ do
626+executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
627+ let sessions = ipSessions conf
564628 eSession <- sessionAndSchema sessionId sessions
565629 case eSession of
566630 Left err -> pure (Just err)
@@ -575,26 +639,23 @@ executeSchemaExpr sessionId (InProcessConnection _ _ sessions _ _) schemaExpr =
575639 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas) (Sess.schemaName session)
576640 STMMap.insert newSession sessionId sessions
577641 pure Nothing
578−executeSchemaExpr sessionId conn@(RemoteProcessConnection _ _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
642+executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
579643
580644-- | After modifying a session, 'commit' the transaction to the transaction graph at the head which the session is referencing. This will also trigger checks for any notifications which need to be propagated.
581645commit :: SessionId -> Connection -> IO (Maybe RelationalError)
582−commit sessionId conn@(InProcessConnection _ _ _ _ _) = executeGraphExpr sessionId conn Commit
583−commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
646+commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
647+commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
584648
585−sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()
586−sendNotifications pids nots = mapM_ sendNots pids
649+sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
650+sendNotifications pids localNode nots = mapM_ sendNots pids
587651 where
588652 sendNots remoteClientPid = do
589− eLocalNode <- commonLocalNode
590− case eLocalNode of
591− Left err -> error ("Failed to get local node: " ++ show err)
592− Right localNode -> when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
653+ when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
593654
594655-- | Discard any changes made in the current session. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.
595656rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
596−rollback sessionId conn@(InProcessConnection _ _ _ _ _) = executeGraphExpr sessionId conn Rollback
597−rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
657+rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
658+rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
598659
599660processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
600661processPersistence NoPersistence _ = return ()
@@ -603,11 +664,12 @@ processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist
603664
604665-- | Return a relation whose type would match that of the relational expression if it were executed. This is useful for checking types and validating a relational expression's types.
605666typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
606−typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
607−typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
667+typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
668+typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
608669
609670typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
610−typeForRelationalExprSTM sessionId (InProcessConnection _ _ sessions _ _) relExpr = do
671+typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
672+ let sessions = ipSessions conf
611673 eSession <- sessionAndSchema sessionId sessions
612674 case eSession of
613675 Left err -> pure $ Left err
@@ -624,7 +686,8 @@ typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-l
624686
625687-- | Return a 'Map' of the database's constraints at the context of the session and connection.
626688inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
627−inclusionDependencies sessionId (InProcessConnection _ _ sessions _ _) = do
689+inclusionDependencies sessionId (InProcessConnection conf) = do
690+ let sessions = ipSessions conf
628691 atomically $ do
629692 eSession <- sessionAndSchema sessionId sessions
630693 case eSession of
@@ -636,11 +699,12 @@ inclusionDependencies sessionId (InProcessConnection _ _ sessions _ _) = do
636699 else
637700 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
638701
639−inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
702+inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
640703
641704-- | Return an optimized database expression which is logically equivalent to the input database expression. This function can be used to determine which expression will actually be evaluated.
642705planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
643−planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) dbExpr = do
706+planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
707+ let sessions = ipSessions conf
644708 atomically $ do
645709 eSession <- sessionAndSchema sessionId sessions
646710 case eSession of
@@ -650,7 +714,7 @@ planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) dbEx
650714 else -- don't show any optimization because the current optimization infrastructure relies on access to the base context- this probably underscores the need for each schema to have its own DatabaseContext, even if it is generated on-the-fly
651715 pure (Right dbExpr)
652716
653−planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
717+planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
654718
655719-- | Return a relation which represents the current state of the global transaction graph. The attributes are
656720-- * current- boolean attribute representing whether or not the current session references this transaction
@@ -658,7 +722,9 @@ planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =
658722-- * id- id attribute of the transaction
659723-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
660724transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
661−transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar _) = do
725+transactionGraphAsRelation sessionId (InProcessConnection conf) = do
726+ let sessions = ipSessions conf
727+ tvar = ipTransactionGraph conf
662728 atomically $ do
663729 eSession <- sessionForSessionId sessionId sessions
664730 case eSession of
@@ -667,10 +733,11 @@ transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar _) =
667733 graph <- readTVar tvar
668734 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
669735
670−transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveTransactionGraph sessionId)
736+transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
671737
672738relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
673−relationVariablesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
739+relationVariablesAsRelation sessionId (InProcessConnection conf) = do
740+ let sessions = ipSessions conf
674741 atomically $ do
675742 eSession <- sessionAndSchema sessionId sessions
676743 case eSession of
@@ -684,17 +751,18 @@ relationVariablesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = d
684751 Left err -> pure (Left err)
685752 Right relvars -> pure $ R.relationVariablesAsRelation relvars
686753
687−relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
754+relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
688755
689756-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
690757headTransactionId :: SessionId -> Connection -> IO (Maybe TransactionId)
691−headTransactionId sessionId (InProcessConnection _ _ sessions _ _) = do
758+headTransactionId sessionId (InProcessConnection conf) = do
759+ let sessions = ipSessions conf
692760 atomically $ do
693761 eSession <- sessionForSessionId sessionId sessions
694762 case eSession of
695763 Left _ -> pure Nothing
696764 Right session -> pure $ Just (Sess.parentId session)
697−headTransactionId sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
765+headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
698766
699767headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)
700768headNameSTM_ sessionId sessions graphTvar = do
@@ -708,12 +776,15 @@ headNameSTM_ sessionId sessions graphTvar = do
708776
709777-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
710778headName :: SessionId -> Connection -> IO (Maybe HeadName)
711−headName sessionId (InProcessConnection _ _ sessions graphTvar _) = do
779+headName sessionId (InProcessConnection conf) = do
780+ let sessions = ipSessions conf
781+ graphTvar = ipTransactionGraph conf
712782 atomically (headNameSTM_ sessionId sessions graphTvar)
713−headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)
783+headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
714784
715785atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
716−atomTypesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
786+atomTypesAsRelation sessionId (InProcessConnection conf) = do
787+ let sessions = ipSessions conf
717788 atomically $ do
718789 eSession <- sessionForSessionId sessionId sessions
719790 case eSession of
@@ -722,9 +793,25 @@ atomTypesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
722793 case typesAsRelation (typeConstructorMapping (Sess.concreteDatabaseContext session)) of
723794 Left err -> pure (Left err)
724795 Right rel -> pure (Right rel)
725−atomTypesAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
796+atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
726797
727798--used only for testing- we expect this to throw an exception
728799callTestTimeout_ :: SessionId -> Connection -> IO Bool
729−callTestTimeout_ _ (InProcessConnection _ _ _ _ _) = error "bad testing call"
730−callTestTimeout_ sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (TestTimeout sessionId)
800+callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
801+callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
802+
803+--used in tests only
804+transactionGraph_ :: Connection -> IO TransactionGraph
805+transactionGraph_ (InProcessConnection conf) = atomically $ readTVar (ipTransactionGraph conf)
806+transactionGraph_ _ = error "remote connection used"
807+
808+--used in tests only
809+disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
810+disconnectedTransaction_ sessionId (InProcessConnection conf) = do
811+ let sessions = ipSessions conf
812+ mSession <- atomically $ do
813+ STMMap.lookup sessionId sessions
814+ case mSession of
815+ Nothing -> error "No such session"
816+ Just (Sess.Session discon _) -> pure discon
817+disconnectedTransaction_ _ _= error "remote connection used"