git.iko.soy/project-m36
on 52182bc help sign in
project-m36/src/lib/ProjectM36/TransactionGraph/Persist.hs
225 lines 11 kB history raw
1module ProjectM36.TransactionGraph.Persist where
2import ProjectM36.Error
3import ProjectM36.TransactionGraph
4import ProjectM36.Transaction
5import ProjectM36.Transaction.Persist
6import ProjectM36.Base
7import ProjectM36.ScriptSession
8import ProjectM36.Persist (writeFileSync, renameSync, DiskSync)
9import ProjectM36.FileLock
10import System.Directory
11import System.FilePath
12import System.IO.Temp
13import Data.Time.Clock.POSIX
14import qualified Data.UUID as U
15import qualified Data.Set as S
16import qualified Data.Map as M
17import qualified Data.Text as T
18import Data.Text.Encoding
19import Control.Monad (foldM)
20import Data.Either (isRight)
21import Data.Maybe (fromMaybe)
22import qualified Data.List as L
23import Control.Exception.Base
24import qualified Data.ByteString as BS
25import qualified Data.Text.Encoding as TE
26import Data.ByteString (ByteString)
27import Data.Monoid
28import qualified Crypto.Hash.SHA256 as SHA256
29import Control.Arrow
30import Data.Time.Clock
31import Data.Text.Read
32import System.FilePath.Glob
34type LockFileHash = ByteString
36{-
37The "m36vX" file at the top-level of the destination directory contains the the transaction graph as a set of transaction ids referencing their parents (1 or more)
38Each Transaction is written to it own directory named by its transaction id. Partially written transactions ids are prefixed with a "." to indicate incompleteness in the graph.
40Persistence requires a POSIX-compliant, journaled-metadata filesystem.
41-}
43expectedVersion :: Int
44expectedVersion = 5
46transactionLogFileName :: FilePath
47transactionLogFileName = "m36v" ++ show expectedVersion
49transactionLogPath :: FilePath -> FilePath
50transactionLogPath dbdir = dbdir </> transactionLogFileName
52headsPath :: FilePath -> FilePath
53headsPath dbdir = dbdir </> "heads"
55lockFilePath :: FilePath -> FilePath
56lockFilePath dbdir = dbdir </> "lockFile"
58{-
59verify that the database directory is valid or bootstrap it
60-note: checking for the existence of every transaction may be prohibitively expensive
61- return error or lock file handle which is already locked with a read lock
62-}
64checkForOtherVersions :: FilePath -> IO (Either PersistenceError ())
65checkForOtherVersions dbdir = do
66 versionMatches <- globDir1 (compile "m36v*") dbdir
67 let otherVersions = L.delete transactionLogFileName (map takeFileName versionMatches)
68 if not (null otherVersions) then
69 pure (Left (WrongDatabaseFormatVersionError transactionLogFileName (head otherVersions)))
70 else
71 pure (Right ())
74setupDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (Either PersistenceError (LockFile, LockFileHash))
75setupDatabaseDir sync dbdir bootstrapGraph = do
76 dbdirExists <- doesDirectoryExist dbdir
77 eWrongVersion <- checkForOtherVersions dbdir
78 case eWrongVersion of
79 Left err -> pure (Left err)
80 Right () -> do
81 m36exists <- doesFileExist (transactionLogPath dbdir)
82 if dbdirExists && m36exists then do
83 --no directories to write, just
84 locker <- openLockFile (lockFilePath dbdir)
85 gDigest <- bracket_ (lockFile locker WriteLock) (unlockFile locker) (readGraphTransactionIdFileDigest dbdir)
86 pure (Right (locker, gDigest))
87 else if not m36exists then
88 Right <$> bootstrapDatabaseDir sync dbdir bootstrapGraph
89 else
90 pure (Left (InvalidDirectoryError dbdir))
91{-
92initialize a database directory with the graph from which to bootstrap- return lock file handle which must be closed by the caller
93-}
94bootstrapDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (LockFile, LockFileHash)
95bootstrapDatabaseDir sync dbdir bootstrapGraph = do
96 createDirectory dbdir
97 locker <- openLockFile (lockFilePath dbdir)
98 let allTransIds = map transactionId (S.toList (transactionsForGraph bootstrapGraph))
99 digest <- bracket_ (lockFile locker WriteLock) (unlockFile locker) (transactionGraphPersist sync dbdir allTransIds bootstrapGraph)
100 pure (locker, digest)
102{-
103incrementally updates an existing database directory
104--algorithm:
105--assume that all transaction data has already been written
106-assume that all non-head transactions have already been written because this is an incremental (and concurrent!) write method
107--store the head names with a symlink to the transaction under "heads"
108-}
109transactionGraphPersist :: DiskSync -> FilePath -> [TransactionId] -> TransactionGraph -> IO LockFileHash
110transactionGraphPersist sync destDirectory transIds graph = do
111 transactionsPersist sync transIds destDirectory graph
112 --write graph file
113 newDigest <- writeGraphTransactionIdFile sync destDirectory graph
114 --write heads file
115 transactionGraphHeadsPersist sync destDirectory graph
116 pure newDigest
118-- | The incremental writer writes the transactions ids specified by the second argument.
120-- 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.
121transactionsPersist :: DiskSync -> [TransactionId] -> FilePath -> TransactionGraph -> IO ()
122transactionsPersist sync transIds destDirectory graphIn = mapM_ writeTrans transIds
123 where writeTrans tid =
124 case transactionForId tid graphIn of
125 Left err -> error ("writeTransaction: " ++ show err)
126 Right trans -> writeTransaction sync destDirectory trans
128{-
129write graph heads to a file which can be atomically swapped
130-}
131--writing the heads in a directory is a synchronization nightmare, so just write the binary to a file and swap atomically
132transactionGraphHeadsPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()
133transactionGraphHeadsPersist sync dbdir graph = do
134 let headFileStr :: (HeadName, Transaction) -> T.Text
135 headFileStr (headName, trans) = headName <> " " <> U.toText (transactionId trans)
136 withTempDirectory dbdir ".heads.tmp" $ \tempHeadsDir -> do
137 let tempHeadsPath = tempHeadsDir </> "heads"
138 headsStrLines = map headFileStr $ M.toList (transactionHeadsForGraph graph)
139 writeFileSync sync tempHeadsPath $ T.intercalate "\n" headsStrLines
140 renameSync sync tempHeadsPath (headsPath dbdir)
142transactionGraphHeadsLoad :: FilePath -> IO [(HeadName,TransactionId)]
143transactionGraphHeadsLoad dbdir = do
144 headsData <- readFile (headsPath dbdir)
145 let headsAssocs = map (\l -> let [headName, uuidStr] = words l in
146 (headName,uuidStr)
147 ) (lines headsData)
148 return [(T.pack headName, uuid) | (headName, Just uuid) <- map (second U.fromString) headsAssocs]
150{-
151load any transactions which are not already part of the incoming transaction graph
152-}
154transactionGraphLoad :: FilePath -> TransactionGraph -> Maybe ScriptSession -> IO (Either PersistenceError TransactionGraph)
155transactionGraphLoad dbdir graphIn mScriptSession = do
156 --optimization: perform tail-bisection search to find last-recorded transaction in the existing stream- replay the rest
157 --read in all missing transactions from transaction directories and add to graph
158 uuidInfo <- readGraphTransactionIdFile dbdir
159 freshHeadsAssoc <- transactionGraphHeadsLoad dbdir
160 case uuidInfo of
161 Left err -> return $ Left err
162 Right info -> do
163 let folder eitherGraph transId = case eitherGraph of
164 Left err -> return $ Left err
165 Right graph -> readTransactionIfNecessary dbdir transId mScriptSession graph
166 loadedGraph <- foldM folder (Right graphIn) (map (\(tid,_,_) -> tid) info)
167 case loadedGraph of
168 Left err -> return $ Left err
169 Right freshGraph -> do
170 let maybeTransHeads = [(headName, transactionForId uuid freshGraph) | (headName, uuid) <- freshHeadsAssoc]
171 freshHeads = M.fromList [(headName,trans) | (headName, Right trans) <- maybeTransHeads]
172 return $ Right $ TransactionGraph freshHeads (transactionsForGraph freshGraph)
174{-
175if the transaction with the TransactionId argument is not yet part of the graph, then read the transaction and add it - this does not update the heads
176-}
177readTransactionIfNecessary :: FilePath -> TransactionId -> Maybe ScriptSession -> TransactionGraph -> IO (Either PersistenceError TransactionGraph)
178readTransactionIfNecessary dbdir transId mScriptSession graphIn =
179 if isRight $ transactionForId transId graphIn then
180 --the transaction is already known and loaded- done
181 return $ Right graphIn
182 else do
183 trans <- readTransaction dbdir transId mScriptSession
184 case trans of
185 Left err -> return $ Left err
186 Right trans' -> return $ Right $ TransactionGraph (transactionHeadsForGraph graphIn) (S.insert trans' (transactionsForGraph graphIn))
188writeGraphTransactionIdFile :: DiskSync -> FilePath -> TransactionGraph -> IO LockFileHash
189writeGraphTransactionIdFile sync destDirectory (TransactionGraph _ transSet) = writeFileSync sync graphFile uuidInfo >> pure digest
190 where
191 graphFile = transactionLogPath destDirectory
192 uuidInfo = T.intercalate "\n" graphLines
193 digest = SHA256.hash (encodeUtf8 uuidInfo)
194 graphLines = S.toList $ S.map graphLine transSet
195 epochTime = realToFrac . utcTimeToPOSIXSeconds . transactionTimestamp :: Transaction -> Double
196 graphLine trans = U.toText (transactionId trans)
197 <> " "
198 <> T.pack (show (epochTime trans))
199 <> " "
200 <> T.intercalate " " (S.toList (S.map U.toText $ transactionParentIds trans))
202readGraphTransactionIdFileDigest :: FilePath -> IO LockFileHash
203readGraphTransactionIdFileDigest dbdir = do
204 let graphTransactionIdData = readUTF8FileOrError (transactionLogPath dbdir)
205 SHA256.hash . encodeUtf8 <$> graphTransactionIdData
207readGraphTransactionIdFile :: FilePath -> IO (Either PersistenceError [(TransactionId, UTCTime, [TransactionId])])
208readGraphTransactionIdFile dbdir = do
209 --read in all transactions' uuids
210 let grapher line = let tid:epochText:parentIds = T.words line in
211 (readUUID tid, readEpoch epochText, map readUUID parentIds)
212 readUUID uuidText = fromMaybe (error "failed to read uuid") (U.fromText uuidText)
213 readEpoch t = posixSecondsToUTCTime (realToFrac (either (error "failed to read epoch") fst (double t)))
214 Right . map grapher . T.lines <$> readUTF8FileOrError (transactionLogPath dbdir)
216--rationale- reading essential database files must fail hard
217readUTF8FileOrError :: FilePath -> IO T.Text
218readUTF8FileOrError pathIn = do
219 eFileBytes <- try (BS.readFile pathIn) :: IO (Either IOError BS.ByteString)
220 case eFileBytes of
221 Left err -> error (show err)
222 Right fileBytes ->
223 case TE.decodeUtf8' fileBytes of
224 Left err -> error (show err)
225 Right utf8Bytes -> pure utf8Bytes