cas-store (empty) → 1.0.0
raw patch · 11 files changed
+2158/−0 lines, 11 filesdep +aesondep +asyncdep +base
Dependencies added: aeson, async, base, bytestring, cas-hashable, cas-store, containers, cryptonite, directory, filepath, hashable, hinotify, hostname, kqueue, lens, monad-control, path, path-io, random, safe-exceptions, sqlite-simple, store, tar, tasty, tasty-hunit, text, unix
Files
- LICENSE +21/−0
- cas-store.cabal +80/−0
- changelog.md +0/−0
- src/Data/CAS/ContentStore.hs +1155/−0
- src/Data/CAS/ContentStore/Notify.hs +28/−0
- src/Data/CAS/ContentStore/Notify/BSD.hs +78/−0
- src/Data/CAS/ContentStore/Notify/Linux.hs +60/−0
- src/Data/CAS/Lock.hs +120/−0
- src/Data/CAS/RemoteCache.hs +134/−0
- test/CAS/ContentStore.hs +472/−0
- test/Test.hs +10/−0
+ LICENSE view
@@ -0,0 +1,21 @@+MIT License++Copyright (c) 2017 Tweag I/O++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ cas-store.cabal view
@@ -0,0 +1,80 @@+Name: cas-store+Version: 1.0.0+Synopsis: A content-addressed storage+Description:+ A content-addressed storage supporting a remote caching. The API mainly consists of the cacheKleisliIO function which takes a (a -> m b) function+ and runs it only if the store doesn't already contain a result for it. Part of the funflow ecosystem.+License: MIT+License-file: LICENSE+Author: Tom Nielsen, Nicholas Clarke, Andreas Herrmann+Maintainer: yves.pares@tweag.io+build-type: Simple+Cabal-Version: >= 1.10+homepage: https://github.com/tweag/funflow+bug-reports: https://github.com/tweag/funflow+category: Control+Tested-With: GHC == 7.8.4, GHC == 7.10.2, GHC == 7.10.3, GHC == 8.0.1++extra-source-files:+ changelog.md++Library+ ghc-options: -Wall -fno-warn-type-defaults+ hs-source-dirs: src+ default-language: Haskell2010++ Exposed-modules: Data.CAS.ContentStore+ , Data.CAS.Lock+ , Data.CAS.RemoteCache+ Other-modules: Data.CAS.ContentStore.Notify+ Build-depends:+ base >= 4.6 && <5+ , aeson >= 1.2.3.0+ , async+ , bytestring+ , cas-hashable >= 1.0.0 && < 2+ , containers+ , cryptonite+ , directory+ , filepath+ , hashable+ , hostname+ , lens+ , monad-control+ , path+ , path-io+ , random+ , safe-exceptions+ , sqlite-simple+ , store+ , tar+ , text+ , unix+ if os(linux)+ CPP-options: -DOS_Linux+ Other-modules: Data.CAS.ContentStore.Notify.Linux+ Build-depends: hinotify >= 0.3.9+ else+ if os(darwin) || os(freebsd)+ CPP-options: -DOS_BSD+ Other-modules: Data.CAS.ContentStore.Notify.BSD+ Build-depends: kqueue++Test-suite unit-tests+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ hs-source-dirs: test+ main-is: Test.hs+ other-modules: CAS.ContentStore+ ghc-options: -Wall -threaded+ build-depends: base+ , async+ , cas-hashable+ , cas-store+ , containers+ , path+ , path-io+ , safe-exceptions+ , tasty+ , tasty-hunit+ , unix
+ changelog.md view
+ src/Data/CAS/ContentStore.hs view
@@ -0,0 +1,1155 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}++-- | Hash addressed store in file system.+--+-- Associates a key ('Data.CAS.ContentHashable.ContentHash')+-- with an item in the store. An item can either be+-- 'Data.CAS.ContentStore.Missing',+-- 'Data.CAS.ContentStore.Pending', or+-- 'Data.CAS.ContentStore.Complete'.+-- The state is persisted in the file system.+--+-- Items are stored under a path derived from their hash. Therefore,+-- there can be no two copies of the same item in the store.+-- If two keys are associated with the same item, then there will be+-- only one copy of that item in the store.+--+-- The store is thread-safe and multi-process safe.+--+-- It is assumed that the user that the process is running under is the owner+-- of the store root, or has permission to create it if missing.+--+-- It is assumed that the store root and its immediate contents are not modified+-- externally. The contents of pending items may be modified externally.+--+-- __Implementation notes:__+--+-- The hash of an item can only be determined once it is completed.+-- If that hash already exists in the store, then the new item is discarded.+--+-- Store state is persisted in the file-system:+--+-- * Pending items are stored writable under the path @pending-\<key>@.+-- * Complete items are stored read-only under the path @item-\<hash>@,+-- with a link under @complete-\<key>@ pointing to that directory.+module Data.CAS.ContentStore+ (+ -- * Open/Close+ withStore+ , open+ , close++ -- * High-level API+ , CacherM (..)+ , Cacher+ , defaultCacherWithIdent+ , defaultIOCacherWithIdent+ , cacheKleisliIO+ , putInStore+ , contentPath+ + -- * List Contents+ , listAll+ , listPending+ , listComplete+ , listItems++ -- * Query/Lookup+ , query+ , isMissing+ , isPending+ , isComplete+ , lookup+ , lookupOrWait+ , waitUntilComplete++ -- * Construct Items+ , constructOrAsync+ , constructOrWait+ , constructIfMissing+ , withConstructIfMissing+ , markPending+ , markComplete++ -- * Remove Contents+ , removeFailed+ , removeForcibly+ , removeItemForcibly++ -- * Aliases+ , assignAlias+ , lookupAlias+ , removeAlias+ , listAliases++ -- * Metadata+ , getBackReferences+ , setInputs+ , getInputs+ , setMetadata+ , getMetadata+ , createMetadataFile+ , getMetadataFile++ -- * Accessors+ , itemHash+ , itemPath+ , itemRelPath+ , contentItem+ , contentFilename+ , root++ -- * Types+ , ContentStore+ , Item+ , Content (..)+ , (^</>)+ , Alias (..)+ , Status (..)+ , Status_+ , Update (..)+ , StoreError (..)+ ) where+++import Prelude hiding (lookup)++import Control.Arrow (second)+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Exception.Safe+import Control.Lens+import Control.Monad (forM_, forever, unless,+ void, when, (<=<), (>=>),+ mzero)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Control (MonadBaseControl)+import Crypto.Hash (hashUpdate)+import Data.Aeson (FromJSON, ToJSON)+import Data.Bits (complement)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as C8+import Data.CAS.ContentStore.Notify+import Data.Foldable (asum)+import qualified Data.Hashable+import Data.List (foldl', stripPrefix)+import Data.Maybe (fromMaybe, listToMaybe)+import Data.Monoid ((<>))+import qualified Data.Store+import Data.String (IsString (..))+import qualified Data.Text as T+import Data.Typeable (Typeable)+import Data.Void+import qualified Database.SQLite.Simple as SQL+import qualified Database.SQLite.Simple.FromField as SQL+import qualified Database.SQLite.Simple.ToField as SQL+import GHC.Generics (Generic)+import Path+import Path.IO+import System.Directory (removePathForcibly)+import System.FilePath (dropTrailingPathSeparator)+import System.IO (Handle, IOMode (..),+ openFile)+import System.Posix.Files+import System.Posix.Types++import Data.CAS.ContentHashable (ContentHash,+ ContentHashable (..),+ DirectoryContent (..),+ contentHashUpdate_fingerprint,+ decodeHash, encodeHash, pathToHash,+ toBytes)+import Data.CAS.Lock+import qualified Data.CAS.RemoteCache as Remote+++-- | Status of an item in the store.+data Status missing pending complete+ = Missing missing+ -- ^ The item does not exist, yet.+ | Pending pending+ -- ^ The item is under construction and not ready for consumption.+ | Complete complete+ -- ^ The item is complete and ready for consumption.+ deriving (Eq, Show)++type Status_ = Status () () ()++-- | Update about the status of a pending item.+data Update+ = Completed Item+ -- ^ The item is now completed and ready for consumption.+ | Failed+ -- ^ Constructing the item failed.+ deriving (Eq, Show)++-- | Errors that can occur when interacting with the store.+data StoreError+ = NotPending ContentHash+ -- ^ An item is not under construction when it should be.+ | AlreadyPending ContentHash+ -- ^ An item is already under construction when it should be missing.+ | AlreadyComplete ContentHash+ -- ^ An item is already complete when it shouldn't be.+ | CorruptedLink ContentHash FilePath+ -- ^ The link under the given hash points to an invalid path.+ | FailedToConstruct ContentHash+ -- ^ A failure occurred while waiting for the item to be constructed.+ | IncompatibleStoreVersion (Path Abs Dir) Int Int+ -- ^ @IncompatibleStoreVersion storeDir actual expected@+ -- The given store has a version number that is incompatible.+ | MalformedMetadataEntry ContentHash SQL.SQLData+ -- ^ @MalformedMetadataEntry hash key@+ -- The metadata entry for the give @hash@, @key@ pair is malformed.+ deriving (Show, Typeable)+instance Exception StoreError where+ displayException = \case+ NotPending hash ->+ "The following input hash is not pending '"+ ++ C8.unpack (encodeHash hash)+ ++ "'."+ AlreadyPending hash ->+ "The following input hash is already pending '"+ ++ C8.unpack (encodeHash hash)+ ++ "'."+ AlreadyComplete hash ->+ "The following input hash is already completed '"+ ++ C8.unpack (encodeHash hash)+ ++ "'."+ CorruptedLink hash fp ->+ "The completed input hash '"+ ++ C8.unpack (encodeHash hash)+ ++ "' points to an invalid store item '"+ ++ fp+ ++ "'."+ FailedToConstruct hash ->+ "Failed to construct the input hash '"+ ++ C8.unpack (encodeHash hash)+ ++ "'."+ IncompatibleStoreVersion storeDir actual expected ->+ "The store in '"+ ++ fromAbsDir storeDir+ ++ "' has version "+ ++ show actual+ ++ ". This software expects version "+ ++ show expected+ ++ ". No automatic migration is available, \+ \please use a fresh store location."+ MalformedMetadataEntry hash key ->+ "The metadtaa entry for hash '"+ ++ C8.unpack (encodeHash hash)+ ++ "' under key '"+ ++ show key+ ++ "' is malformed."++-- | A hash addressed store on the file system.+data ContentStore = ContentStore+ { storeRoot :: Path Abs Dir+ -- ^ Root directory of the content store.+ -- The process must be able to create this directory if missing,+ -- change permissions, and create files and directories within.+ , storeLock :: Lock+ -- ^ Write lock on store metadata to ensure multi thread and process safety.+ -- The lock is taken when item state is changed or queried.+ , storeNotifier :: Notifier+ -- ^ Used to watch for updates on store items.+ , storeDb :: SQL.Connection+ -- ^ Connection to the metadata SQLite database.+ }++-- | A completed item in the 'ContentStore'.+data Item = Item { itemHash :: ContentHash }+ deriving (Eq, Ord, Show, Generic)++instance Monad m => ContentHashable m Item where+ contentHashUpdate ctx item =+ flip contentHashUpdate_fingerprint item+ >=> pure . flip hashUpdate (toBytes $ itemHash item)+ $ ctx++instance FromJSON Item+instance ToJSON Item+instance Data.Hashable.Hashable Item+instance Data.Store.Store Item++-- | File or directory within a content store 'Item'.+data Content t where+ All :: Item -> Content Dir+ (:</>) :: Item -> Path Rel t -> Content t+infixr 5 :</>+deriving instance Eq (Content t)+deriving instance Show (Content t)+instance Monad m => ContentHashable m (Content Dir) where+ contentHashUpdate ctx x = case x of+ All i ->+ flip contentHashUpdate_fingerprint x+ >=> flip contentHashUpdate i+ $ ctx+ i :</> p ->+ flip contentHashUpdate_fingerprint x+ >=> flip contentHashUpdate i+ >=> flip contentHashUpdate p+ $ ctx+instance Monad m => ContentHashable m (Content File) where+ contentHashUpdate ctx x = case x of+ i :</> p ->+ flip contentHashUpdate_fingerprint x+ >=> flip contentHashUpdate i+ >=> flip contentHashUpdate p+ $ ctx++-- | Append to the path within a store item.+(^</>) :: Content Dir -> Path Rel t -> Content t+All item ^</> path = item :</> path+(item :</> dir) ^</> path = item :</> dir </> path+infixl 4 ^</>+ +newtype Alias = Alias { unAlias :: T.Text }+ deriving (ContentHashable IO, Eq, Ord, Show, SQL.FromField, SQL.ToField, Data.Store.Store)++-- | The root directory of the store.+root :: ContentStore -> Path Abs Dir+root = storeRoot++-- | The scoped path to a content item within the store.+itemRelPath :: Item -> Path Rel Dir+itemRelPath (Item x) = prefixHashPath itemPrefix x++-- | The store path of a completed item.+itemPath :: ContentStore -> Item -> Path Abs Dir+itemPath store = mkItemPath store . itemHash++-- | Store item containing the given content.+contentItem :: Content t -> Item+contentItem (All i) = i+contentItem (i :</> _) = i++contentFilename :: Content File -> Path Rel File+contentFilename (_ :</> relPath) = filename relPath++-- | The absolute path to content within the store.+contentPath :: ContentStore -> Content t -> Path Abs t+contentPath store (All item) = itemPath store item+contentPath store (item :</> dir) = itemPath store item </> dir++-- | @open root@ opens a store under the given root directory.+--+-- The root directory is created if necessary.+--+-- It is not safe to have multiple store objects+-- refer to the same root directory.+open :: Path Abs Dir -> IO ContentStore+open storeRoot = do+ createDirIfMissing True storeRoot+ storeLock <- openLock (lockPath storeRoot)+ withLock storeLock $ withWritableStoreRoot storeRoot $ do+ storeDb <- SQL.open (fromAbsFile $ dbPath storeRoot)+ initDb storeRoot storeDb+ createDirIfMissing True (metadataPath storeRoot)+ storeNotifier <- initNotifier+ return ContentStore {..}++-- | Free the resources associated with the given store object.+--+-- The store object may not be used afterwards.+close :: ContentStore -> IO ()+close store = do+ closeLock (storeLock store)+ killNotifier (storeNotifier store)+ SQL.close (storeDb store)++-- | Open the store under the given root and perform the given action.+-- Closes the store once the action is complete+--+-- See also: 'Data.CAS.ContentStore.open'+withStore :: (MonadIO m, MonadMask m)+ => Path Abs Dir -> (ContentStore -> m a) -> m a+withStore root' = bracket (liftIO $ open root') (liftIO . close)++-- | List all elements in the store+-- @(pending keys, completed keys, completed items)@.+listAll :: MonadIO m => ContentStore -> m ([ContentHash], [ContentHash], [Item])+listAll ContentStore {storeRoot} = liftIO $+ foldr go ([], [], []) . fst <$> listDir storeRoot+ where+ go d prev@(builds, outs, items) = fromMaybe prev $ asum+ [ parsePending d >>= \x -> Just (x:builds, outs, items)+ , parseComplete d >>= \x -> Just (builds, x:outs, items)+ , parseItem d >>= \x -> Just (builds, outs, x:items)+ ]+ parsePending :: Path Abs Dir -> Maybe ContentHash+ parsePending = pathToHash <=< stripPrefix pendingPrefix . extractDir+ parseComplete :: Path Abs Dir -> Maybe ContentHash+ parseComplete = pathToHash <=< stripPrefix completePrefix . extractDir+ parseItem :: Path Abs Dir -> Maybe Item+ parseItem = fmap Item . pathToHash <=< stripPrefix itemPrefix . extractDir+ extractDir :: Path Abs Dir -> FilePath+ extractDir = dropTrailingPathSeparator . fromRelDir . dirname++-- | List all pending keys in the store.+listPending :: MonadIO m => ContentStore -> m [ContentHash]+listPending = fmap (^._1) . listAll++-- | List all completed keys in the store.+listComplete :: MonadIO m => ContentStore -> m [ContentHash]+listComplete = fmap (^._2) . listAll++-- | List all completed items in the store.+listItems :: MonadIO m => ContentStore -> m [Item]+listItems = fmap (^._3) . listAll++-- | Query the state of the item under the given key.+query :: MonadIO m => ContentStore -> ContentHash -> m (Status () () ())+query store hash = liftIO . withStoreLock store $+ internalQuery store hash >>= pure . \case+ Missing _ -> Missing ()+ Pending _ -> Pending ()+ Complete _ -> Complete ()++-- | Check if there is no complete or pending item under the given key.+isMissing :: MonadIO m => ContentStore -> ContentHash -> m Bool+isMissing store hash = (== Missing ()) <$> query store hash++-- | Check if there is a pending item under the given key.+isPending :: MonadIO m => ContentStore -> ContentHash -> m Bool+isPending store hash = (== Pending ()) <$> query store hash++-- | Check if there is a completed item under the given key.+isComplete :: MonadIO m => ContentStore -> ContentHash -> m Bool+isComplete store hash = (== Complete ()) <$> query store hash++-- | Query the state under the given key and return the item if completed.+-- Doesn't block if the item is pending.+lookup :: MonadIO m => ContentStore -> ContentHash -> m (Status () () Item)+lookup store hash = liftIO . withStoreLock store $+ internalQuery store hash >>= \case+ Missing () -> return $ Missing ()+ Pending _ -> return $ Pending ()+ Complete item -> return $ Complete item++-- | Query the state under the given key and return the item if completed.+-- Return an 'Control.Concurrent.Async' to await an update, if pending.+lookupOrWait+ :: MonadIO m+ => ContentStore+ -> ContentHash+ -> m (Status () (Async Update) Item)+lookupOrWait store hash = liftIO . withStoreLock store $+ internalQuery store hash >>= \case+ Complete item -> return $ Complete item+ Missing () -> return $ Missing ()+ Pending _ -> Pending <$> internalWatchPending store hash++-- | Query the state under the given key and return the item once completed.+-- Blocks if the item is pending.+-- Returns 'Nothing' if the item is missing, or failed to be completed.+waitUntilComplete :: MonadIO m => ContentStore -> ContentHash -> m (Maybe Item)+waitUntilComplete store hash = lookupOrWait store hash >>= \case+ Complete item -> return $ Just item+ Missing () -> return Nothing+ Pending a -> liftIO (wait a) >>= \case+ Completed item -> return $ Just item+ Failed -> return Nothing++-- | Atomically query the state under the given key and mark pending if missing.+--+-- Returns @'Complete' item@ if the item is complete.+-- Returns @'Pending' async@ if the item is pending, where @async@ is an+-- 'Control.Concurrent.Async' to await updates on.+-- Returns @'Missing' buildDir@ if the item was missing, and is now pending.+-- It should be constructed in the given @buildDir@,+-- and then marked as complete using 'markComplete'.+constructOrAsync+ :: forall m remoteCache.+ (MonadIO m, MonadBaseControl IO m, MonadMask m, Remote.Cacher m remoteCache)+ => ContentStore+ -> remoteCache+ -> ContentHash+ -> m (Status (Path Abs Dir) (Async Update) Item)+constructOrAsync store remoteCacher hash =+ constructIfMissing store remoteCacher hash >>= \case+ Complete item -> return $ Complete item+ Missing path -> return $ Missing path+ Pending _ -> Pending <$> liftIO (internalWatchPending store hash)++-- | Atomically query the state under the given key and mark pending if missing.+-- Wait for the item to be completed, if already pending.+-- Throws a 'FailedToConstruct' error if construction fails.+--+-- Returns @'Complete' item@ if the item is complete.+-- Returns @'Missing' buildDir@ if the item was missing, and is now pending.+-- It should be constructed in the given @buildDir@,+-- and then marked as complete using 'markComplete'.+constructOrWait+ :: (MonadIO m, MonadMask m, MonadBaseControl IO m, Remote.Cacher m remoteCache)+ => ContentStore+ -> remoteCache+ -> ContentHash+ -> m (Status (Path Abs Dir) Void Item)+constructOrWait store remoteCacher hash = constructOrAsync store remoteCacher hash >>= \case+ Pending a -> liftIO (wait a) >>= \case+ Completed item -> return $ Complete item+ -- XXX: Consider extending 'Status' with a 'Failed' constructor.+ -- If the store contains metadata as well, it could keep track of the+ -- number of failed attempts and further details about the failure.+ -- If an external task is responsible for the failure, the client could+ -- choose to resubmit a certain number of times.+ Failed -> liftIO . throwIO $ FailedToConstruct hash+ Complete item -> return $ Complete item+ Missing dir -> return $ Missing dir++-- | Atomically query the state under the given key and mark pending if missing.+constructIfMissing+ :: (MonadIO m, MonadBaseControl IO m, MonadMask m, Remote.Cacher m remoteCache)+ => ContentStore+ -> remoteCache+ -> ContentHash+ -> m (Status (Path Abs Dir) () Item)+constructIfMissing store remoteCacher hash = withStoreLock store $+ internalQuery store hash >>= \case+ Complete item -> return $ Complete item+ Missing () -> withWritableStore store $ do+ let destDir :: Path Abs Dir = mkItemPath store hash+ Remote.pull remoteCacher hash destDir >>= \case+ Remote.PullOK () -> return $ Complete (Item hash)+ Remote.NotInCache ->+ Missing <$> liftIO (internalMarkPending store hash)+ Remote.PullError _ ->+ Missing <$> liftIO (internalMarkPending store hash)+ Pending _ -> return $ Pending ()++-- | Atomically query the state under the given key and mark pending if missing.+-- Execute the given function to construct the item, mark as complete on success+-- and remove on failure. Forcibly removes if an uncaught exception occurs+-- during item construction.+withConstructIfMissing+ :: (MonadIO m, MonadBaseControl IO m, MonadMask m, Remote.Cacher m remoteCache)+ => ContentStore+ -> remoteCache+ -> ContentHash+ -> (Path Abs Dir -> m (Either e a))+ -> m (Status e () (Maybe a, Item))+withConstructIfMissing store remoteCacher hash f =+ bracketOnError+ (constructIfMissing store remoteCacher hash)+ (\case+ Missing _ -> removeForcibly store hash+ _ -> return ())+ (\case+ Pending () -> return (Pending ())+ Complete item -> return (Complete (Nothing, item))+ Missing fp -> f fp >>= \case+ Left e -> do+ removeFailed store hash+ return (Missing e)+ Right x -> do+ item <- markComplete store hash+ _ <- Remote.push remoteCacher (itemHash item) (Just hash) (itemPath store item)+ return (Complete (Just x, item)))++-- | Mark a non-existent item as pending.+--+-- Creates the build directory and returns its path.+--+-- See also: 'Data.CAS.ContentStore.constructIfMissing'.+markPending :: MonadIO m => ContentStore -> ContentHash -> m (Path Abs Dir)+markPending store hash = liftIO . withStoreLock store $+ internalQuery store hash >>= \case+ Complete _ -> throwIO (AlreadyComplete hash)+ Pending _ -> throwIO (AlreadyPending hash)+ Missing () -> withWritableStore store $+ internalMarkPending store hash++-- | Mark a pending item as complete.+markComplete :: MonadIO m => ContentStore -> ContentHash -> m Item+markComplete store inHash = liftIO . withStoreLock store $+ internalQuery store inHash >>= \case+ Missing () -> throwIO (NotPending inHash)+ Complete _ -> throwIO (AlreadyComplete inHash)+ Pending build -> withWritableStore store $ liftIO $ do+ do+ let metadataDir = mkMetadataDirPath store inHash+ exists <- doesDirExist metadataDir+ when exists $+ unsetWritableRecursively metadataDir+ -- XXX: Hashing large data can take some time,+ -- could we avoid locking the store for all that time?+ outHash <- contentHash (DirectoryContent build)+ let out = mkItemPath store outHash+ link' = mkCompletePath store inHash+ doesDirExist out >>= \case+ True -> removePathForcibly (fromAbsDir build)+ False -> do+ renameDir build out+ unsetWritableRecursively out+ rel <- makeRelative (parent link') out+ let from' = dropTrailingPathSeparator $ fromAbsDir link'+ to' = dropTrailingPathSeparator $ fromRelDir rel+ createSymbolicLink to' from'+ addBackReference store inHash (Item outHash)+ pure $! Item outHash++-- | Remove a pending item.+--+-- It is the callers responsibility to ensure that no other threads or processes+-- will attempt to access the item's contents afterwards.+removeFailed :: MonadIO m => ContentStore -> ContentHash -> m ()+removeFailed store hash = liftIO . withStoreLock store $+ internalQuery store hash >>= \case+ Missing () -> throwIO (NotPending hash)+ Complete _ -> throwIO (AlreadyComplete hash)+ Pending build -> withWritableStore store $+ removePathForcibly (fromAbsDir build)++-- | Remove a key association independent of the corresponding item state.+-- Do nothing if no item exists under the given key.+--+-- It is the callers responsibility to ensure that no other threads or processes+-- will attempt to access the contents afterwards.+--+-- Note, this will leave an orphan item behind if no other keys point to it.+-- There is no garbage collection mechanism in place at the moment.+removeForcibly :: MonadIO m => ContentStore -> ContentHash -> m ()+removeForcibly store hash = liftIO . withStoreLock store $ withWritableStore store $+ internalQuery store hash >>= \case+ Missing () -> pure ()+ Pending build -> liftIO $ removePathForcibly (fromAbsDir build)+ Complete _out -> liftIO $+ removePathForcibly $+ dropTrailingPathSeparator $ fromAbsDir $ mkCompletePath store hash+ -- XXX: This will leave orphan store items behind.+ -- Add GC in some form.++-- | Remove a completed item in the store.+-- Do nothing if not completed.+--+-- It is the callers responsibility to ensure that no other threads or processes+-- will attempt to access the contents afterwards.+--+-- Note, this will leave keys pointing to that item dangling.+-- There is no garbage collection mechanism in place at the moment.+removeItemForcibly :: MonadIO m => ContentStore -> Item -> m ()+removeItemForcibly store item = liftIO . withStoreLock store $ withWritableStore store $+ removePathForcibly (fromAbsDir $ itemPath store item)+ -- XXX: Remove dangling links.+ -- Add back-references in some form.++-- We need this orphan instance here so cas-hash doesn't depend on sqlite+instance SQL.FromField ContentHash where+ fromField f = do+ bs <- SQL.fromField f+ case decodeHash bs of+ Just h -> pure h+ Nothing -> mzero++instance SQL.ToField ContentHash where+ toField = SQL.toField . encodeHash++-- | Link the given alias to the given item.+-- If the alias existed before it is overwritten.+assignAlias :: MonadIO m => ContentStore -> Alias -> Item -> m ()+assignAlias store alias item =+ liftIO . withStoreLock store $ withWritableStore store $ do+ hash <- contentHash alias+ SQL.executeNamed (storeDb store)+ "INSERT OR REPLACE INTO\+ \ aliases\+ \ VALUES\+ \ (:hash, :dest, :name)"+ [ ":hash" SQL.:= hash+ , ":dest" SQL.:= itemHash item+ , ":name" SQL.:= alias+ ]++-- | Lookup an item under the given alias.+-- Returns 'Nothing' if the alias does not exist.+lookupAlias :: MonadIO m => ContentStore -> Alias -> m (Maybe Item)+lookupAlias store alias =+ liftIO . withStoreLock store $ do+ hash <- contentHash alias+ r <- SQL.queryNamed (storeDb store)+ "SELECT dest FROM aliases\+ \ WHERE\+ \ hash = :hash"+ [ ":hash" SQL.:= hash ]+ pure $! listToMaybe $ Item . SQL.fromOnly <$> r++-- | Remove the given alias.+removeAlias :: MonadIO m => ContentStore -> Alias -> m ()+removeAlias store alias =+ liftIO . withStoreLock store $ withWritableStore store $ do+ hash <- contentHash alias+ SQL.executeNamed (storeDb store)+ "DELETE FROM aliases\+ \ WHERE\+ \ hash = :hash"+ [ ":hash" SQL.:= hash ]++-- | List all aliases and the respective items.+listAliases :: MonadIO m => ContentStore -> m [(Alias, Item)]+listAliases store = liftIO . withStoreLock store $+ fmap (map (second Item)) $+ SQL.query_ (storeDb store)+ "SELECT name, dest FROM aliases"++-- | Get all hashes that resulted in the given item.+getBackReferences :: MonadIO m => ContentStore -> Item -> m [ContentHash]+getBackReferences store (Item outHash) = liftIO . withStoreLock store $+ map SQL.fromOnly <$> SQL.queryNamed (storeDb store)+ "SELECT hash FROM backrefs\+ \ WHERE\+ \ dest = :out"+ [ ":out" SQL.:= outHash ]++-- | Define the input items to a subtree.+setInputs :: MonadIO m => ContentStore -> ContentHash -> [Item] -> m ()+setInputs store hash items = liftIO $+ withStoreLock store $+ withWritableStore store $+ internalQuery store hash >>= \case+ Pending _ -> forM_ items $ \(Item input) ->+ SQL.executeNamed (storeDb store)+ "INSERT OR REPLACE INTO\+ \ inputs (hash, input)\+ \ VALUES\+ \ (:hash, :input)"+ [ ":hash" SQL.:= hash+ , ":input" SQL.:= input+ ]+ _ -> throwIO $ NotPending hash++-- | Get the input items to a subtree if any were defined.+getInputs :: MonadIO m => ContentStore -> ContentHash -> m [Item]+getInputs store hash = liftIO . withStoreLock store $+ map (Item . SQL.fromOnly) <$> SQL.queryNamed (storeDb store)+ "SELECT input FROM inputs\+ \ WHERE\+ \ hash = :hash"+ [ ":hash" SQL.:= hash ]++-- | Set a metadata entry on an item.+setMetadata :: (SQL.ToField k, SQL.ToField v, MonadIO m )+ => ContentStore -> ContentHash -> k -> v -> m ()+setMetadata store hash k v = liftIO $+ withStoreLock store $+ withWritableStore store $+ SQL.executeNamed (storeDb store)+ "INSERT OR REPLACE INTO\+ \ metadata (hash, key, value)\+ \ VALUES\+ \ (:hash, :key, :value)"+ [ ":hash" SQL.:= hash+ , ":key" SQL.:= k+ , ":value" SQL.:= v+ ]++-- | Retrieve a metadata entry on an item, or 'Nothing' if missing.+getMetadata :: (SQL.ToField k, SQL.FromField v, MonadIO m)+ => ContentStore -> ContentHash -> k -> m (Maybe v)+getMetadata store hash k = liftIO . withStoreLock store $ do+ r <- SQL.queryNamed (storeDb store)+ "SELECT value FROM metadata\+ \ WHERE\+ \ (hash = :hash AND key = :key)"+ [ ":hash" SQL.:= hash+ , ":key" SQL.:= k+ ]+ case r of+ [] -> pure Nothing+ [[v]] -> pure $ Just v+ _ -> throwIO $ MalformedMetadataEntry hash (SQL.toField k)++-- | Create and open a new metadata file on a pending item in write mode.+createMetadataFile+ :: MonadIO m+ => ContentStore -> ContentHash -> Path Rel File -> m (Path Abs File, Handle)+createMetadataFile store hash file = liftIO . withStoreLock store $+ internalQuery store hash >>= \case+ Pending _ -> do+ let path = mkMetadataFilePath store hash file+ createDirIfMissing True (parent path)+ handle <- openFile (fromAbsFile path) WriteMode+ pure (path, handle)+ _ -> throwIO $ NotPending hash++-- | Return the path to a metadata file if it exists.+getMetadataFile+ :: MonadIO m+ => ContentStore -> ContentHash -> Path Rel File -> m (Maybe (Path Abs File))+getMetadataFile store hash file = liftIO . withStoreLock store $ do+ let path = mkMetadataFilePath store hash file+ exists <- doesFileExist path+ if exists then+ pure $ Just path+ else+ pure Nothing++----------------------------------------------------------------------+-- Internals++lockPath :: Path Abs Dir -> Path Abs Dir+lockPath = (</> [reldir|lock|])++dbPath :: Path Abs Dir -> Path Abs File+dbPath = (</> [relfile|metadata.db|])++metadataPath :: Path Abs Dir -> Path Abs Dir+metadataPath = (</> [reldir|metadata|])++-- | Holds a lock on the global 'MVar' and on the global lock file+-- for the duration of the given action.+withStoreLock :: MonadBaseControl IO m => ContentStore -> m a -> m a+withStoreLock store = withLock (storeLock store)++prefixHashPath :: C8.ByteString -> ContentHash -> Path Rel Dir+prefixHashPath pref hash+ | Just dir <- Path.parseRelDir $ C8.unpack $ pref <> encodeHash hash+ = dir+ | otherwise = error+ "[Data.CAS.ContentStore.prefixHashPath] \+ \Failed to construct hash path."++pendingPrefix, completePrefix, hashPrefix, itemPrefix :: IsString s => s+pendingPrefix = "pending-"+completePrefix = "complete-"+hashPrefix = "hash-"+itemPrefix = "item-"++-- | Return the full build path for the given input hash.+mkPendingPath :: ContentStore -> ContentHash -> Path Abs Dir+mkPendingPath ContentStore {storeRoot} hash =+ storeRoot </> prefixHashPath pendingPrefix hash++-- | Return the full link path for the given input hash.+mkCompletePath :: ContentStore -> ContentHash -> Path Abs Dir+mkCompletePath ContentStore {storeRoot} hash =+ storeRoot </> prefixHashPath completePrefix hash++-- | Return the full store path to the given output hash.+mkItemPath :: ContentStore -> ContentHash -> Path Abs Dir+mkItemPath ContentStore {storeRoot} hash =+ storeRoot </> prefixHashPath itemPrefix hash++-- | Return the full store path to the given metadata directory.+mkMetadataDirPath :: ContentStore -> ContentHash -> Path Abs Dir+mkMetadataDirPath ContentStore {storeRoot} hash =+ metadataPath storeRoot </> prefixHashPath hashPrefix hash++-- | Return the full store path to the given metadata file.+mkMetadataFilePath+ :: ContentStore -> ContentHash -> Path Rel File -> Path Abs File+mkMetadataFilePath store hash file =+ mkMetadataDirPath store hash </> file++-- | Query the state under the given key without taking a lock.+internalQuery+ :: MonadIO m+ => ContentStore+ -> ContentHash+ -> m (Status () (Path Abs Dir) Item)+internalQuery store inHash = liftIO $ do+ let build = mkPendingPath store inHash+ link' = mkCompletePath store inHash+ buildExists <- doesDirExist build+ if buildExists then+ pure $! Pending build+ else do+ linkExists <- doesDirExist link'+ if linkExists then do+ out <- readSymbolicLink+ (dropTrailingPathSeparator $ fromAbsDir link')+ case pathToHash =<< stripPrefix itemPrefix out of+ Nothing -> throwIO $ CorruptedLink inHash out+ Just outHash -> return $ Complete (Item outHash)+ else+ pure $! Missing ()++-- | Create the build directory for the given input hash+-- and make the metadata directory writable if it exists.+internalMarkPending :: ContentStore -> ContentHash -> IO (Path Abs Dir)+internalMarkPending store hash = do+ let dir = mkPendingPath store hash+ createDir dir+ setDirWritable dir+ let metadataDir = mkMetadataDirPath store hash+ metadirExists <- doesDirExist metadataDir+ when metadirExists $+ setWritableRecursively metadataDir+ return dir++-- | Watch the build directory of the pending item under the given key.+-- The returned 'Async' completes after the item is completed or failed.+internalWatchPending+ :: ContentStore+ -> ContentHash+ -> IO (Async Update)+internalWatchPending store hash = do+ let build = mkPendingPath store hash+ -- Add an inotify/kqueue watch and give a signal on relevant events.+ let notifier = storeNotifier store+ signal <- newEmptyMVar+ -- Signal the listener. If the 'MVar' is full,+ -- the listener didn't handle earlier signals, yet.+ let giveSignal = void $ tryPutMVar signal ()+ watch <- addDirWatch notifier (fromAbsDir build) giveSignal+ -- Additionally, poll on regular intervals.+ -- Inotify/Kqueue don't cover all cases, e.g. network filesystems.+ ticker <- async $ forever $ threadDelay 3007000 >> giveSignal+ let stopWatching = do+ cancel ticker+ removeDirWatch watch+ -- Listen to the signal asynchronously,+ -- and query the status when it fires.+ -- If the status changed, fill in the update.+ update <- newEmptyMVar+ let query' = liftIO . withStoreLock store $ internalQuery store hash+ loop = takeMVar signal >> query' >>= \case+ Pending _ -> loop+ Complete item -> tryPutMVar update $ Completed item+ Missing () -> tryPutMVar update Failed+ void $ async loop+ -- Wait for the update asynchronously.+ -- Stop watching when it arrives.+ async $ takeMVar update <* stopWatching++setRootDirWritable :: MonadIO m => Path Abs Dir -> m ()+setRootDirWritable storeRoot = liftIO $+ setFileMode (fromAbsDir storeRoot) writableRootDirMode++writableRootDirMode :: FileMode+writableRootDirMode = writableDirMode++setRootDirReadOnly :: MonadIO m => Path Abs Dir -> m ()+setRootDirReadOnly storeRoot = liftIO $+ setFileMode (fromAbsDir storeRoot) readOnlyRootDirMode++readOnlyRootDirMode :: FileMode+readOnlyRootDirMode = writableDirMode `intersectFileModes` allButWritableMode++withWritableStoreRoot :: (MonadMask m, MonadIO m) => Path Abs Dir -> m a -> m a+withWritableStoreRoot storeRoot =+ bracket_ (setRootDirWritable storeRoot) (setRootDirReadOnly storeRoot)++withWritableStore :: (MonadMask m, MonadIO m) => ContentStore -> m a -> m a+withWritableStore ContentStore {storeRoot} =+ withWritableStoreRoot storeRoot++setDirWritable :: Path Abs Dir -> IO ()+setDirWritable fp = setFileMode (fromAbsDir fp) writableDirMode++writableDirMode :: FileMode+writableDirMode = foldl' unionFileModes nullFileMode+ [ directoryMode, ownerModes+ , groupReadMode, groupExecuteMode+ , otherReadMode, otherExecuteMode+ ]++-- | Set write permissions on the given path.+setWritable :: Path Abs t -> IO ()+setWritable fp = do+ mode <- fileMode <$> getFileStatus (toFilePath fp)+ setFileMode (toFilePath fp) $ mode `unionFileModes` ownerWriteMode++-- | Unset write permissions on the given path.+unsetWritable :: Path Abs t -> IO ()+unsetWritable fp = do+ mode <- fileMode <$> getFileStatus (toFilePath fp)+ setFileMode (toFilePath fp) $ mode `intersectFileModes` allButWritableMode++allButWritableMode :: FileMode+allButWritableMode = complement $ foldl' unionFileModes nullFileMode+ [ownerWriteMode, groupWriteMode, otherWriteMode]++-- | Set write permissions on all items in a directory tree recursively.+setWritableRecursively :: Path Abs Dir -> IO ()+setWritableRecursively = walkDir $ \dir _ files -> do+ mapM_ setWritable files+ setWritable dir+ return $ WalkExclude []++-- | Unset write permissions on all items in a directory tree recursively.+unsetWritableRecursively :: Path Abs Dir -> IO ()+unsetWritableRecursively = walkDir $ \dir _ files -> do+ mapM_ unsetWritable files+ unsetWritable dir+ return $ WalkExclude []++storeVersion :: Int+storeVersion = 1++-- | Initialize the database.+initDb :: Path Abs Dir -> SQL.Connection -> IO ()+initDb storeDir db = do+ [[version]] <- SQL.query_ db "PRAGMA user_version"+ if version == 0 then+ SQL.execute_ db $+ "PRAGMA user_version = " <> fromString (show storeVersion)+ else+ unless (version == storeVersion) $+ throwIO $ IncompatibleStoreVersion storeDir version storeVersion+ -- Aliases to items.+ SQL.execute_ db+ "CREATE TABLE IF NOT EXISTS\+ \ aliases\+ \ ( hash TEXT PRIMARY KEY\+ \ , dest TEXT NOT NULL\+ \ , name TEXT NOT NULL\+ \ )"+ -- Back-references from items @dest@ to hashes @hash@.+ SQL.execute_ db+ "CREATE TABLE IF NOT EXISTS\+ \ backrefs\+ \ ( hash TEXT PRIMARY KEY\+ \ , dest TEXT NOT NULL\+ \ )"+ -- Inputs @input@ to hashes @hash@.+ SQL.execute_ db+ "CREATE TABLE IF NOT EXISTS\+ \ inputs\+ \ ( hash TEXT NOT NULL\+ \ , input TEXT NOT NULL\+ \ , UNIQUE (hash, input)\+ \ )"+ -- Arbitrary metadata on hashes.+ SQL.execute_ db+ "CREATE TABLE IF NOT EXISTS\+ \ metadata\+ \ ( hash TEXT NOT NULL\+ \ , key TEXT NOT NULL\+ \ , value TEXT\+ \ , PRIMARY KEY(hash, key)\+ \ )"++-- | Adds a link between input hash and the output hash.+--+-- Assumes that the store is locked and writable.+addBackReference :: ContentStore -> ContentHash -> Item -> IO ()+addBackReference store inHash (Item outHash) =+ SQL.executeNamed (storeDb store)+ "INSERT OR REPLACE INTO\+ \ backrefs (hash, dest)\+ \ VALUES\+ \ (:in, :out)"+ [ ":in" SQL.:= inHash+ , ":out" SQL.:= outHash+ ]++-- | A cacher is responsible for controlling how steps are cached.+data CacherM m i o =+ NoCache -- ^ This step cannot be cached (default).+ | Cache+ { -- | Function to encode the input into a content+ -- hash.+ -- This function additionally takes an+ -- 'identities' which gets incorporated into+ -- the cacher.+ cacherKey :: Int -> i -> m ContentHash+ , cacherStoreValue :: o -> ByteString+ -- | Attempt to read the cache value back. May throw exceptions.+ , cacherReadValue :: ByteString -> o+ }++-- | A pure 'CacherM'+type Cacher = CacherM Identity++-- | Constructs a 'Cacher' that will use hashability of input and+-- serializability of output to make a step cacheable+defaultCacherWithIdent :: forall m i o.+ (ContentHashable m i, Data.Store.Store o)+ => Int -- ^ Seed for the cacher+ -> CacherM m i o+defaultCacherWithIdent ident = Cache+ { cacherKey = \i ident' -> contentHash (ident', ident, i)+ , cacherStoreValue = Data.Store.encode+ , cacherReadValue = Data.Store.decodeEx+ }++-- | Looks for a @CacherM IO@, then lifts it+defaultIOCacherWithIdent :: (MonadIO m, ContentHashable IO i, Data.Store.Store o)+ => Int -- ^ Seed for the cacher+ -> CacherM m i o+defaultIOCacherWithIdent ident = c{cacherKey = \x i -> liftIO $ cacherKey c x i}+ where c = defaultCacherWithIdent ident+++-- | Caches a Kleisli of some MonadIO action in the store given the required+-- properties+cacheKleisliIO+ :: (MonadIO m, MonadBaseControl IO m, MonadMask m, Remote.Cacher m remoteCache)+ => Maybe Int -- ^ This can be used to disambiguate the same program run in+ -- multiple configurations. If Nothing, then it means this+ -- program has no identity, this implies that steps will be+ -- executed without cache, even if 'Cache' has been given.+ -> CacherM m i o+ -> ContentStore+ -> remoteCache+ -> (i -> m o)+ -> i+ -> m o+cacheKleisliIO confIdent c@Cache{} store remoteCacher f i+ | Just confIdent' <- confIdent = do+ chash <- cacherKey c confIdent' i+ let computeAndStore fp = do+ res <- f i -- Do the actual computation+ liftIO $ BS.writeFile (toFilePath $ fp </> [relfile|out|])+ . cacherStoreValue c $ res+ return $ Right res+ readItem item = do+ bs <- liftIO . BS.readFile $ simpleOutPath item+ return . cacherReadValue c $ bs+ withConstructIfMissing store remoteCacher chash computeAndStore >>= \case+ Missing e -> absurd e+ Pending _ ->+ liftIO (waitUntilComplete store chash) >>= \case+ Just item -> readItem item+ Nothing -> throwM $ FailedToConstruct chash+ Complete (Just a, _) -> return a+ Complete (_, item) -> readItem item+ where+ simpleOutPath item =+ toFilePath $ itemPath store item </> [relfile|out|]+cacheKleisliIO _ _ _ _ f i = f i++-- | Caches an action that writes content-addressed data to the store. Returns+-- the Item of the written content.+putInStore+ :: (MonadIO m, MonadMask m, MonadBaseControl IO m+ ,Remote.Cacher m remoteCacher+ ,ContentHashable IO t)+ => ContentStore+ -> remoteCacher+ -> (ContentHash -> m ()) -- ^ In case an exception occurs+ -> (Path Abs Dir -> t -> m ()) -- ^ The action that writes to the new store+ -- directory+ -> t+ -> m Item -- ^ The Item in the store to which @t@ has been written+putInStore store remoteCacher ifExc f x = do+ chash <- liftIO $ contentHash x+ constructOrWait store remoteCacher chash >>= \case+ Pending y -> absurd y+ Complete item -> return item+ Missing fp ->+ do+ f fp x+ finalItem <- markComplete store chash+ _ <- Remote.push remoteCacher (itemHash finalItem) (Just chash) (itemPath store finalItem)+ pure finalItem+ `onException`+ (do ifExc chash+ removeFailed store chash)
+ src/Data/CAS/ContentStore/Notify.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE CPP #-}++-- | Generic file change notifier library for unix-based systems.+--+-- This library abstracts over specific implementations for BSD and linux+-- systems.+--+-- It provides facilities to watch specific directories for the following changes:+-- - File moves+-- - File deletion+-- - Attribute changes.+module Data.CAS.ContentStore.Notify+ ( Notifier+ , initNotifier+ , killNotifier++ , Watch+ , addDirWatch+ , removeDirWatch+ ) where++#ifdef OS_Linux+import Data.CAS.ContentStore.Notify.Linux+#else+# ifdef OS_BSD+import Data.CAS.ContentStore.Notify.BSD+# endif+#endif
+ src/Data/CAS/ContentStore/Notify/BSD.hs view
@@ -0,0 +1,78 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Implementation of filesystem watch for BSD-based systems using KQueue.+module Data.CAS.ContentStore.Notify.BSD+ ( Notifier+ , initNotifier+ , killNotifier++ , Watch+ , addDirWatch+ , removeDirWatch+ ) where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Exception.Safe+import Control.Monad+import Data.Typeable+import Foreign.Ptr+import System.KQueue+import System.Posix.IO+import System.Posix.Types++type Notifier = KQueue++data NotifierException+ = RequiresThreadedRuntime+ deriving (Show, Typeable)+instance Exception NotifierException where+ displayException = \case+ -- XXX: Compile time check?+ RequiresThreadedRuntime ->+ "Threaded runtime required! Please rebuild with -threaded flag."++initNotifier :: IO Notifier+initNotifier = do+ unless rtsSupportsBoundThreads $ throwIO RequiresThreadedRuntime+ kqueue++killNotifier :: Notifier -> IO ()+killNotifier _ = return ()++data Watch = Watch KQueue Fd (Async ())++addDirWatch :: Notifier -> FilePath -> IO () -> IO Watch+addDirWatch kq dir f = do+ fd <- openFd dir ReadOnly Nothing defaultFileFlags+ a <- async $ do+ let event = KEvent+ { ident = fromIntegral fd+ , evfilter = EvfiltVnode+ , flags = [EvAdd]+ , fflags = [NoteDelete, NoteAttrib, NoteRename, NoteRevoke]+ , data_ = 0+ , udata = nullPtr+ }+ loop = do+ chgs <- kevent kq [event] 1 Nothing+ unless (null chgs) f+ loop+ loop `finally` closeFd fd+ return $! Watch kq fd a++removeDirWatch :: Watch -> IO ()+removeDirWatch (Watch kq fd a) = do+ closeFd fd+ let event = KEvent+ { ident = fromIntegral fd+ , evfilter = EvfiltVnode+ , flags = [EvDelete]+ , fflags = []+ , data_ = 0+ , udata = nullPtr+ }+ void (kevent kq [event] 0 Nothing)+ `catch` \(_ :: KQueueException) -> return ()+ cancel a
+ src/Data/CAS/ContentStore/Notify/Linux.hs view
@@ -0,0 +1,60 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Implementation of filesystem watching functionality for linux based on+-- inotify.+module Data.CAS.ContentStore.Notify.Linux+ ( Notifier+ , initNotifier+ , killNotifier++ , Watch+ , addDirWatch+ , removeDirWatch+ ) where++import Control.Exception.Safe (catch)+#if MIN_VERSION_hinotify(0,3,10)+import qualified Data.ByteString.Char8 as BS+#endif+import System.INotify++type Notifier = INotify++initNotifier :: IO Notifier+initNotifier = initINotify++killNotifier :: Notifier -> IO ()+killNotifier = killINotify++type Watch = WatchDescriptor++addDirWatch :: Notifier -> FilePath -> IO () -> IO Watch+addDirWatch inotify dir f = addWatch inotify mask dir' $ \case+ Attributes True Nothing -> f+ MovedSelf True -> f+ DeletedSelf -> f+ _ -> return ()+ where+ mask = [Attrib, MoveSelf, DeleteSelf, OnlyDir]+#if MIN_VERSION_hinotify(0,3,10)+ dir' = BS.pack dir+#else+ dir' = dir+#endif++removeDirWatch :: Watch -> IO ()+removeDirWatch w =+ -- When calling `addWatch` on a path that is already being watched,+ -- inotify will not create a new watch, but amend the existing watch+ -- and return the same watch descriptor.+ -- Therefore, the watch might already have been removed at this point,+ -- which will cause an 'IOError'.+ -- Fortunately, all event handlers to a file are called at once.+ -- So, that removing the watch here will not cause another handler+ -- to miss out on the event.+ -- Note, that this may change when adding different event handlers,+ -- that remove the watch under different conditions.+ removeWatch w+ `catch` \(_::IOError) -> return ()
+ src/Data/CAS/Lock.hs view
@@ -0,0 +1,120 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | Thread and process write lock.+--+-- Allows synchronisation between threads and processes.+-- Uses an 'MVar' for synchronisation between threads+-- and fcntl write locks for synchronisation between processes.+--+-- Only ever have one 'Lock' object per lock file per process!+module Data.CAS.Lock+ ( Lock+ , openLock+ , closeLock+ , withLock+ ) where++import Control.Concurrent+import Control.Exception.Safe+import Control.Monad (unless)+import Control.Monad.Trans.Control (MonadBaseControl, liftBaseOp_)+import Network.HostName (getHostName)+import Path+import Path.IO+import System.Posix.Files+import System.Posix.IO+import System.Posix.Process+import System.Random++-- | Thread and process write lock.+--+-- Only ever have one 'Lock' object per lock file per process!+data Lock = Lock+ { lockMVar :: MVar ()+ , lockDir :: Path Abs Dir+ }++-- | Open the lock file and create a lock object.+--+-- This does not acquire the lock.+--+-- Only ever have one 'Lock' object per lock file per process!+openLock :: Path Abs Dir -> IO Lock+openLock dir = do+ mvar <- newMVar ()+ createDirIfMissing True dir+ return $! Lock+ { lockMVar = mvar+ , lockDir = dir+ }++-- | Close the lock file.+--+-- Does not release the lock.+--+-- Blocks if the lock is taken.+closeLock :: Lock -> IO ()+closeLock lock = do+ takeMVar (lockMVar lock)++-- | Acquire the lock for the duration of the given action and release after.+withLock :: MonadBaseControl IO m => Lock -> m a -> m a+withLock lock = liftBaseOp_ $ \action ->+ withMVar (lockMVar lock) $ \() ->+ bracket_ (acquireDirLock $ lockDir lock) (releaseDirLock $ lockDir lock) $+ action++----------------------------------------------------------------------+-- Internals++-- | Generate unique (per process) filename.+--+-- Combines the host name and process ID.+getUniqueFileName :: IO (Path Rel File)+getUniqueFileName = do+ hostName <- getHostName+ pid <- getProcessID+ parseRelFile $ hostName ++ show pid++lockFileName :: Path Rel File+lockFileName = [relfile|lock|]++-- | Acquire the lock.+--+-- Uses an algorithm that is described in the man-page of open (2) in the+-- last paragraph to @O_EXCL@ in release 4.14 of the Linux man-pages project.+--+-- Creates a file under a unique (per process) filename.+-- Attempts to hard-link that file to a common lock path.+-- If the operation succeeds, then the lock was acquired.+-- If not, but if the link count of the file under the unique filename+-- increased to two, then the lock was acquired.+-- Otherwise, another process holds the lock and this process waits+-- and retries.+acquireDirLock :: Path Abs Dir -> IO ()+acquireDirLock dir = do+ file <- getUniqueFileName+ let path = dir </> file+ fd <- createFile (fromAbsFile path) ownerWriteMode+ closeFd fd+ r <- try $ createLink (fromAbsFile path) (fromAbsFile $ dir </> lockFileName)+ case r of+ Right () -> return ()+ Left (_::IOError) -> do+ count <- linkCount <$> getFileStatus (fromAbsFile path)+ unless (count == 2) $ do+ delay <- randomRIO (50000, 100000)+ threadDelay delay+ acquireDirLock dir++-- | Release the lock.+--+-- Unlinks the file under the unique file name and the common lock file.+releaseDirLock :: Path Abs Dir -> IO ()+releaseDirLock dir = do+ file <- getUniqueFileName+ let path = dir </> file+ removeLink (fromAbsFile $ dir </> lockFileName)+ removeLink (fromAbsFile path)
+ src/Data/CAS/RemoteCache.hs view
@@ -0,0 +1,134 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}++-- |+-- This module defines the remote caching mechanism of funflow which is used to+-- keep several funflow stores (possibly on different machines) in sync.+module Data.CAS.RemoteCache+ ( Cacher(..)+ , PullResult(..), PushResult(..), AliasResult(..)+ , NoCache(..), memoryCache+ , pullAsArchive, pushAsArchive+ ) where++import qualified Codec.Archive.Tar as Tar+import Control.Concurrent.MVar+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.ByteString.Lazy (ByteString)+import Data.CAS.ContentHashable+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Path+++-- |+-- The result of a tentative pull from the remote cache+data PullResult a+ = PullOK a+ | NotInCache+ | PullError String+ deriving (Eq, Ord, Show)++-- |+-- The result of a tentative push to the remote cache+data PushResult+ = PushOK+ | PushError String+ deriving (Eq, Ord, Show)++data AliasResult+ = AliasOK+ | TargetNotInCache+ | AliasError String++-- |+-- A simple mechanism for remote-caching.+--+-- Provides a way to push a path to the cache and pull it back.+--+-- No assumption is made on the availability of a store path. In particular,+-- pushing a path to the cache doesn't mean that we can pull it back.+class Monad m => Cacher m a where+ push ::+ a+ -> ContentHash -- ^ "Primary" key: hash of the content+ -> Maybe ContentHash -- ^ "Secondary" key: hash of the dependencies+ -> Path Abs Dir -- ^ Path to the content+ -> m PushResult+ pull :: a -> ContentHash -> Path Abs Dir -> m (PullResult ())++-- |+-- Push the path as an archive to the remote cache+pushAsArchive ::+ MonadIO m+ => (ContentHash -> ContentHash -> m (Either String ())) -- ^ How to create the aliases+ -> (ContentHash -> ByteString -> m PushResult) -- ^ How to push the content+ -> ContentHash -- ^ Primary key+ -> Maybe ContentHash -- ^ Secondary key+ -> Path Abs Dir+ -> m PushResult+pushAsArchive alias pushArchive primaryKey mSecondaryKey path = do+ archive <- liftIO $ Tar.write <$> Tar.pack (toFilePath path) ["."]+ pushArchive primaryKey archive >>= \case+ PushError e -> pure $ PushError e+ res ->+ case mSecondaryKey of+ Just secondaryKey ->+ alias primaryKey secondaryKey >>= \case+ Left err -> pure $ PushError err+ Right () -> pure res+ Nothing -> pure res++pullAsArchive ::+ MonadIO m+ => (ContentHash -> m (PullResult ByteString))+ -> ContentHash+ -> Path Abs Dir+ -> m (PullResult ())+pullAsArchive pullArchive hash path =+ pullArchive hash >>= \case+ PullOK archive -> do+ liftIO $ Tar.unpack (toFilePath path) $ Tar.read archive+ pure $ PullOK ()+ NotInCache -> pure NotInCache+ PullError e -> pure $ PullError e++-- |+-- A dummy remote cache implementation which does nothing+data NoCache = NoCache++instance Monad m => Cacher m NoCache where+ pull _ _ _ = pure NotInCache+ push _ _ _ _ = pure PushOK++-- |+-- An in-memory cache, for testing purposes+data MemoryCache = MemoryCache (MVar (Map ContentHash ByteString))+instance MonadIO m => Cacher m MemoryCache where+ pull (MemoryCache cacheVar) = pullAsArchive $ \hash -> do+ cacheMap <- liftIO $ readMVar cacheVar+ case Map.lookup hash cacheMap of+ Nothing -> pure NotInCache+ Just x -> pure (PullOK x)+ push (MemoryCache cacheVar) = pushAsArchive alias $ \hash content -> do+ liftIO $ modifyMVar_+ cacheVar+ (\cacheMap -> pure $ Map.insert hash content cacheMap)+ pure PushOK+ where+ alias from to = liftIO $ Right <$> modifyMVar_ cacheVar+ (\cacheMap -> pure $ Map.insert to (cacheMap Map.! from) cacheMap)++memoryCache :: MonadIO m => m MemoryCache+memoryCache = liftIO $ MemoryCache <$> newMVar mempty++-- |+-- If 'a' is a 'Cacher' then 'Maybe a' is a cacher such that 'Just x' behavies+-- like 'x' and 'Nothing' doesn't cache anything+instance Cacher m a => Cacher m (Maybe a) where+ pull (Just x) = pull x+ pull Nothing = pull NoCache++ push (Just x) = push x+ push Nothing = push NoCache
+ test/CAS/ContentStore.hs view
@@ -0,0 +1,472 @@+{-# OPTIONS_GHC -fno-warn-unused-do-bind #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TypeApplications #-}++module CAS.ContentStore+ ( tests+ ) where++import Control.Concurrent.Async+import Control.Exception.Safe (tryAny)+import Data.CAS.ContentHashable (contentHash)+import Data.CAS.ContentStore (ContentStore)+import qualified Data.CAS.ContentStore as ContentStore+import qualified Data.CAS.RemoteCache as Remote+import Control.Monad (void)+import Data.Maybe (catMaybes)+import qualified Data.Set as Set+import Path+import Path.IO+import System.Posix.Files+import Test.Tasty+import Test.Tasty.HUnit++tests :: TestTree+tests = testGroup "Content Store"++ [ testCase "initialise fresh store" $+ withTmpDir $ \dir -> do+ let root = dir </> [reldir|store|]+ ContentStore.withStore root $ \_ ->+ doesDirExist @IO root+ @? "store root exists"++ , testCase "initialise existing store" $+ withTmpDir $ \dir -> do+ let root = dir </> [reldir|store|]+ createDir root+ ContentStore.withStore root $ \_ ->+ doesDirExist @IO root+ @? "store root exists"++ , testCase "store is not writable" $+ withEmptyStore $ \store -> do+ let root = ContentStore.root store+ isNotWritable root+ @? "store not writable"++ , testCase "subtree stages" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)++ missing <- ContentStore.query store hash+ missing @?= ContentStore.Missing ()+ missing' <- ContentStore.lookup store hash+ missing' @?= ContentStore.Missing ()++ subtree <- ContentStore.markPending store hash+ let dir = subtree </> [reldir|dir|]+ file = dir </> [relfile|file|]+ expectedContent = "Hello World"+ pending <- ContentStore.query store hash+ pending @?= ContentStore.Pending ()+ pending' <- ContentStore.lookup store hash+ pending' @?= ContentStore.Pending ()+ doesDirExist @IO subtree+ @? "pending subtree exists"+ isWritable subtree+ @? "pending subtree is writable"+ createDir dir+ writeFile (fromAbsFile file) expectedContent+ do+ content <- readFile (fromAbsFile file)+ content @?= expectedContent++ item <- ContentStore.markComplete store hash+ let itemDir = ContentStore.itemPath store item+ file' = itemDir </> [relfile|dir/file|]+ complete <- ContentStore.query store hash+ complete @?= ContentStore.Complete ()+ complete' <- ContentStore.lookup store hash+ complete' @?= ContentStore.Complete item+ doesDirExist @IO itemDir+ @? "complete subtree exists"+ isNotWritable itemDir+ @? "complete subtree is not writable"+ isNotWritable file'+ @? "complete file is not writable"+ do+ content <- readFile (fromAbsFile file')+ content @?= expectedContent++ , testCase "await construction" $+ Remote.memoryCache >>= \myCache ->+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)++ ContentStore.constructOrAsync store myCache hash >>= \case+ ContentStore.Pending _ ->+ assertFailure "missing already under construction"+ ContentStore.Complete _ ->+ assertFailure "missing already complete"+ ContentStore.Missing _ ->+ return ()++ a <- ContentStore.constructOrAsync store Remote.NoCache hash >>= \case+ ContentStore.Missing _ -> do+ assertFailure "under construction still missing"+ undefined+ ContentStore.Complete _ -> do+ assertFailure "under construction already complete"+ undefined+ ContentStore.Pending a ->+ return a++ b <- ContentStore.lookupOrWait store hash >>= \case+ ContentStore.Missing _ -> do+ assertFailure "under construction still missing"+ undefined+ ContentStore.Complete _ -> do+ assertFailure "under construction already complete"+ undefined+ ContentStore.Pending b ->+ return b++ item <- ContentStore.markComplete store hash++ item' <- wait a+ item' @?= ContentStore.Completed item++ item'' <- wait b+ item'' @?= ContentStore.Completed item++ ContentStore.constructOrAsync store Remote.NoCache hash >>= \case+ ContentStore.Missing _ -> do+ assertFailure "complete still missing"+ ContentStore.Pending _ -> do+ assertFailure "complete still under construction"+ ContentStore.Complete _ -> do+ return ()++ , testCase "await failure" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)++ ContentStore.constructOrAsync store Remote.NoCache hash >>= \case+ ContentStore.Pending _ ->+ assertFailure "missing already under construction"+ ContentStore.Complete _ ->+ assertFailure "missing already complete"+ ContentStore.Missing _ ->+ return ()++ a <- ContentStore.constructOrAsync store Remote.NoCache hash >>= \case+ ContentStore.Missing _ -> do+ assertFailure "under construction still missing"+ undefined+ ContentStore.Complete _ -> do+ assertFailure "under construction already complete"+ undefined+ ContentStore.Pending a ->+ return a++ b <- ContentStore.lookupOrWait store hash >>= \case+ ContentStore.Missing _ -> do+ assertFailure "under construction still missing"+ undefined+ ContentStore.Complete _ -> do+ assertFailure "under construction already complete"+ undefined+ ContentStore.Pending b ->+ return b++ ContentStore.removeFailed store hash++ item' <- wait a+ item' @?= ContentStore.Failed++ item'' <- wait b+ item'' @?= ContentStore.Failed++ ContentStore.constructOrAsync store Remote.NoCache hash >>= \case+ ContentStore.Pending _ -> do+ assertFailure "failed still under construction"+ ContentStore.Complete _ -> do+ assertFailure "failed already complete"+ ContentStore.Missing _ -> do+ return ()++ , testCase "construct if missing" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ let file = [relfile|file|]+ expectedContent = "Hello World"++ ContentStore.constructIfMissing store Remote.NoCache hash >>= \case+ ContentStore.Pending () ->+ assertFailure "missing already under construction"+ ContentStore.Complete _ ->+ assertFailure "missing already complete"+ ContentStore.Missing subtree -> do+ isWritable subtree+ @? "under construction not writable"+ writeFile (fromAbsFile $ subtree </> file) expectedContent++ ContentStore.constructIfMissing store Remote.NoCache hash >>= \case+ ContentStore.Missing _ ->+ assertFailure "under construction still missing"+ ContentStore.Complete _ ->+ assertFailure "under construction already complete"+ ContentStore.Pending () ->+ void $ ContentStore.markComplete store hash++ ContentStore.constructIfMissing store Remote.NoCache hash >>= \case+ ContentStore.Missing _ ->+ assertFailure "complete still missing"+ ContentStore.Pending () ->+ assertFailure "complete still under construction"+ ContentStore.Complete item -> do+ let subtree = ContentStore.itemPath store item+ isNotWritable (subtree </> file)+ @? "complete still writable"+ content <- readFile (fromAbsFile $ subtree </> file)+ content @?= expectedContent++ , testCase "remove failed" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ subtree <- ContentStore.markPending store hash+ ContentStore.removeFailed store hash+ not <$> doesDirExist @IO subtree+ @? "subtree was removed"++ , testCase "forcibly remove" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ subtree <- ContentStore.markPending store hash++ ContentStore.removeForcibly store hash+ not <$> doesDirExist @IO subtree+ @? "remove under construction"++ ContentStore.removeForcibly store hash+ not <$> doesDirExist @IO subtree+ @? "remove missing"++ subtree' <- ContentStore.markPending store hash+ void $ ContentStore.markComplete store hash+ ContentStore.removeForcibly store hash+ not <$> doesDirExist @IO subtree'+ @? "remove complete"++ , testCase "subtree state is persisted" $+ withTmpDir $ \dir -> do+ let root = dir </> [reldir|store|]+ hash <- contentHash ("test" :: String)++ do+ ContentStore.withStore root $ \store ->+ void $ ContentStore.markPending store hash++ -- Imagine the process terminates and the store is closed++ do+ ContentStore.withStore root $ \store -> do+ underConstruction <- ContentStore.query store hash+ underConstruction @?= ContentStore.Pending ()+ void $ ContentStore.markComplete store hash++ -- Imagine the process terminates and the store is closed++ do+ ContentStore.withStore root $ \store -> do+ complete <- ContentStore.query store hash+ complete @?= ContentStore.Complete ()++ , testCase "mark complete before under construction fails" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ ContentStore.markComplete store hash+ `shouldFail` "complete before under construction"++ , testCase "mark complete after complete fails" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ void $ ContentStore.markPending store hash+ void $ ContentStore.markComplete store hash+ ContentStore.markComplete store hash+ `shouldFail` "complete after complete"++ , testCase "mark under construction after under construction fails" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ void $ ContentStore.markPending store hash+ void $ ContentStore.markPending store hash+ `shouldFail` "under construction after under construction"++ , testCase "mark under construction after complete fails" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ void $ ContentStore.markPending store hash+ void $ ContentStore.markComplete store hash+ void $ ContentStore.markPending store hash+ `shouldFail` "under construction after complete"++ , testCase "remove missing fails" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ ContentStore.removeFailed store hash+ `shouldFail` "remove non existent"++ , testCase "remove complete fails" $+ withEmptyStore $ \store -> do+ hash <- contentHash ("test" :: String)+ void $ ContentStore.markPending store hash+ void $ ContentStore.markComplete store hash+ ContentStore.removeFailed store hash+ `shouldFail` "remove complete"++ , testCase "list store contents" $+ withEmptyStore $ \store -> do+ [a, b, c, d] <- mapM contentHash ["a", "b", "c", "d" :: String]+ void $ mapM (ContentStore.markPending store) [a, b, c, d]+ mapM_ (ContentStore.markComplete store) [a, b]++ (pendings, completes, items) <- ContentStore.listAll store+ let all' = pendings ++ completes+ Set.fromList all' @?= Set.fromList [a, b, c, d]+ Set.size (Set.fromList all') @?= 4++ underContsruction <- ContentStore.listPending store+ Set.fromList underContsruction @?= Set.fromList [c, d]++ complete <- ContentStore.listComplete store+ Set.fromList complete @?= Set.fromList [a, b]++ items' <- catMaybes <$> mapM (ContentStore.waitUntilComplete store) [a, b]+ Set.fromList items @?= Set.fromList items'++ , testCase "store aliases" $+ withEmptyStore $ \store -> do+ let fp = [relfile|test|]+ contentA = "itemA"+ contentB = "itemB"+ itemA <- do+ hash <- contentHash ("a" :: String)+ dir <- ContentStore.markPending store hash+ writeFile (fromAbsFile $ dir </> fp) contentA+ ContentStore.markComplete store hash+ itemB <- do+ hash <- contentHash ("b" :: String)+ dir <- ContentStore.markPending store hash+ writeFile (fromAbsFile $ dir </> fp) contentB+ ContentStore.markComplete store hash+ let aliasA = ContentStore.Alias "aliasA"+ aliasB = ContentStore.Alias "aliasB"++ do+ r <- ContentStore.lookupAlias store aliasA+ r @?= Nothing++ ContentStore.assignAlias store aliasA itemA+ ContentStore.assignAlias store aliasB itemB+ do+ r <- ContentStore.lookupAlias store aliasA+ r @?= Just itemA+ do+ r <- ContentStore.lookupAlias store aliasB+ r @?= Just itemB++ ContentStore.assignAlias store aliasB itemA+ do+ r <- ContentStore.lookupAlias store aliasA+ r @?= Just itemA+ do+ r <- ContentStore.lookupAlias store aliasB+ r @?= Just itemA++ ContentStore.removeAlias store aliasA+ ContentStore.removeAlias store aliasB+ do+ r <- ContentStore.lookupAlias store aliasA+ r @?= Nothing+ do+ r <- ContentStore.lookupAlias store aliasB+ r @?= Nothing++ , testCase "Remote caching (constructOrAsync)" $ do+ cacher <- Remote.memoryCache+ hash <- contentHash ("test" :: String)+ let file = [relfile|file|]+ expectedContent = "Hello World"++ -- Populate the remote cache+ withEmptyStore $ \store -> do+ ContentStore.constructOrAsync store cacher hash >>= \case+ ContentStore.Pending _ ->+ assertFailure "missing already under construction"+ ContentStore.Complete _ ->+ assertFailure "missing already complete"+ ContentStore.Missing subtree -> do+ isWritable subtree @? "under construction not writable"+ writeFile (fromAbsFile $ subtree </> file) expectedContent+ Remote.push cacher hash Nothing subtree++ -- Expects having the item in cache+ withEmptyStore $ \store -> do+ ContentStore.constructOrAsync store cacher hash >>= \case+ ContentStore.Pending _ ->+ assertFailure "missing already under construction"+ ContentStore.Complete _ -> pure ()+ ContentStore.Missing _ -> assertFailure "Not found in the cache"++ , testCase "Remote caching (withConstructIfMissing)" $ do+ cacher <- Remote.memoryCache+ hash <- contentHash ("test" :: String)+ let file = [relfile|file|]+ expectedContent = "Hello World"+ doWrite subtree = do+ isWritable subtree @? "under construction not writable"+ writeFile (fromAbsFile $ subtree </> file) expectedContent+ return $ Right ()+ + -- Populates the remote cache+ withEmptyStore $ \store -> do+ ContentStore.withConstructIfMissing store cacher hash doWrite >>= \case+ ContentStore.Missing _ -> assertFailure "not found in the cache"+ ContentStore.Pending _ ->+ assertFailure "missing already under construction"+ -- ContentStore.waitUntilComplete store hash >>= \case+ -- Just item -> return ()+ -- Nothing -> assertFailure "item construction failed"+ ContentStore.Complete _ -> pure ()+ + -- Expects having the item in the remote cache+ withEmptyStore $ \store -> do+ ContentStore.withConstructIfMissing store cacher hash+ (const $ assertFailure "should not try to write the file a second time") >>= \case+ ContentStore.Missing _ -> assertFailure "Not found in the cache"+ ContentStore.Pending _ ->+ assertFailure "missing already under construction"+ ContentStore.Complete _ -> pure ()++ ]++shouldFail :: IO a -> String -> IO ()+shouldFail m msg = tryAny m >>= \case+ Left _ -> return ()+ Right _ -> assertFailure msg++withTmpDir :: (Path Abs Dir -> IO a) -> IO a+withTmpDir = withSystemTempDir "funflow-test"++withEmptyStore :: (ContentStore -> IO a) -> IO a+withEmptyStore k = withTmpDir $ \dir ->+ ContentStore.withStore (dir </> [reldir|store|]) k++isNotWritable :: Path Abs t -> IO Bool+isNotWritable path = do+ mode <- fileMode <$> getFileStatus (toFilePath path)+ return $! nullFileMode == (mode `intersectFileModes` writeModes)+ where+ writeModes =+ ownerWriteMode+ `unionFileModes`+ groupWriteMode+ `unionFileModes`+ otherWriteMode++isWritable :: Path Abs t -> IO Bool+isWritable = fmap not . isNotWritable
+ test/Test.hs view
@@ -0,0 +1,10 @@+import qualified CAS.ContentStore+import Test.Tasty++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "Unit Tests"+ [ CAS.ContentStore.tests+ ]