packages feed

funflow 1.5.0 → 1.6.0

raw patch · 33 files changed

+187/−2723 lines, 33 filesdep +cas-hashabledep +cas-storedep +profunctorsdep −hinotifydep −kqueue

Dependencies added: cas-hashable, cas-store, profunctors

Dependencies removed: hinotify, kqueue

Files

TestFunflow.hs view
@@ -7,10 +7,9 @@ import           Control.Arrow.Free import           Control.Exception.Safe import           Control.Funflow-import qualified Control.Funflow.ContentStore                as CS import           Control.Funflow.External.Coordinator.Memory import           Control.Funflow.Pretty-import           Data.Default+import qualified Data.CAS.ContentStore                       as CS import           Data.Monoid                                 ((<>)) import           Path import           Path.IO
app/FFExecutorD.hs view
@@ -11,12 +11,12 @@ import           Control.Concurrent.MVar import           Control.Exception.Base                      (AsyncException (UserInterrupt)) import           Control.Exception.Safe-import qualified Control.Funflow.ContentStore                as CS import           Control.Funflow.External.Coordinator import           Control.Funflow.External.Coordinator.Redis import           Control.Funflow.External.Coordinator.SQLite import           Control.Funflow.External.Executor import           Control.Monad                               (void)+import qualified Data.CAS.ContentStore                       as CS import           Data.Monoid                                 ((<>)) import qualified Database.Redis                              as R import           Network.Socket                              (PortNumber)
funflow.cabal view
@@ -1,5 +1,5 @@ Name:                funflow-Version:             1.5.0+Version:             1.6.0 Synopsis:            Workflows with arrows Description:         An arrow with resumable computations and logging@@ -24,13 +24,10 @@    default-language:  Haskell2010     Exposed-modules:-                     Control.Arrow.AppArrow-                   , Control.Arrow.Async+                     Control.Arrow.Async                    , Control.Arrow.Free                    , Control.Funflow                    , Control.Funflow.Cache.TH-                   , Control.Funflow.ContentHashable-                   , Control.Funflow.ContentStore                    , Control.Funflow.Diagram                    , Control.Funflow.External                    , Control.Funflow.External.Docker@@ -39,22 +36,20 @@                    , Control.Funflow.External.Coordinator.Memory                    , Control.Funflow.External.Coordinator.Redis                    , Control.Funflow.External.Coordinator.SQLite-                   , Control.Funflow.Lock-                   , Control.Funflow.Orphans-                   , Control.Funflow.RemoteCache                    , Control.Funflow.Steps                    , Control.Funflow.Pretty                    , Control.Funflow.Exec.Simple    Other-modules:                      Control.Funflow.Base                    , Control.Funflow.Class-                   , Control.Funflow.ContentStore.Notify    Build-depends:                  base                    >= 4.6 && <5                , Glob                , aeson                   >= 1.2.3.0                , async                , bytestring+               , cas-hashable            >= 1.0.1 && <2+               , cas-store               >= 1.0.1 && <2                , clock                , constraints                , containers@@ -79,6 +74,7 @@                , path-io                , pretty                , process+               , profunctors                , random                , safe-exceptions                , scientific@@ -94,21 +90,13 @@                , unordered-containers                , vector                , yaml-   if os(linux)-     CPP-options: -DOS_Linux-     Other-modules: Control.Funflow.ContentStore.Notify.Linux-     Build-depends: hinotify             >= 0.3.9-   else-     if os(darwin) || os(freebsd)-       CPP-options: -DOS_BSD-       Other-modules: Control.Funflow.ContentStore.Notify.BSD-       Build-depends: kqueue  Executable ffexecutord   default-language:  Haskell2010   main-is: app/FFExecutorD.hs   build-depends:       base >=4.6 && <5                      , bytestring+                     , cas-store                      , clock                      , funflow                      , hedis >= 0.12.5@@ -126,6 +114,7 @@   main-is: TestFunflow.hs   ghc-options:        -Wall -threaded   build-depends:       base >=4.6 && <5+                     , cas-store                      , data-default                      , funflow                      , filepath@@ -141,14 +130,13 @@   default-language:   Haskell2010   hs-source-dirs:     test   main-is:            Test.hs-  other-modules:      Funflow.ContentStore-                      Funflow.SQLiteCoordinator+  other-modules:      Funflow.SQLiteCoordinator                       Funflow.TestFlows                       Control.Arrow.Async.Tests   ghc-options:        -Wall -threaded   build-depends:      base                     , async-                    , containers+                    , cas-store                     , data-default >= 0.7                     , directory                     , filepath
− src/Control/Arrow/AppArrow.hs 
@@ -1,39 +0,0 @@--- | This modules defines the composition of an applicative functor and an--- arrow, which is always an arrow.--module Control.Arrow.AppArrow-  ( AppArrow(..)-  , appArrow-  ) where--import Control.Category-import Control.Arrow-import Prelude          hiding (id, (.))--newtype AppArrow app arr a b = AppArrow { unAppArrow :: app (arr a b) }--instance (Applicative app, Category cat) => Category (AppArrow app cat) where-  id = appArrow id-  AppArrow a1 . AppArrow a2 = AppArrow $ (.) <$> a1 <*> a2--instance (Applicative app, Arrow arr) => Arrow (AppArrow app arr) where-  arr = appArrow . arr-  first (AppArrow a) = AppArrow $ first <$> a-  second (AppArrow a) = AppArrow $ second <$> a-  AppArrow a1 *** AppArrow a2 = AppArrow $ (***) <$> a1 <*> a2--instance (Applicative app, ArrowChoice arr) => ArrowChoice (AppArrow app arr) where-  left (AppArrow a) = AppArrow $ left <$> a-  right (AppArrow a) = AppArrow $ right <$> a-  AppArrow a1 +++ AppArrow a2 = AppArrow $ (+++) <$> a1 <*> a2-  AppArrow a1 ||| AppArrow a2 = AppArrow $ (|||) <$> a1 <*> a2--instance (Applicative f, Arrow arr) => Functor (AppArrow f arr t) where-  fmap f = (>>> arr f)--instance (Applicative app, Arrow arr) => Applicative (AppArrow app arr t) where-  pure = arr . const-  a <*> b = a &&& b >>> arr (uncurry ($))--appArrow :: (Applicative app) => arr a b -> AppArrow app arr a b-appArrow = AppArrow . pure
src/Control/Arrow/Async.hs view
@@ -2,6 +2,9 @@ {-# LANGUAGE FlexibleInstances     #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE UndecidableInstances  #-}+{-# LANGUAGE DerivingVia           #-}+{-# LANGUAGE StandaloneDeriving    #-}+ -- | Asynchronous arrows over monads with MonadBaseControl IO, using --   lifted-async. module Control.Arrow.Async where@@ -14,9 +17,25 @@ import qualified Control.Exception.Safe import           Control.Monad.Trans.Class       (MonadTrans, lift) import           Control.Monad.Trans.Control     (MonadBaseControl)+import qualified Data.Profunctor                 as P+import qualified Data.Profunctor.Mapping         as P+import qualified Data.Profunctor.Traversing      as P import           Prelude                         hiding (id, (.))  newtype AsyncA m a b = AsyncA { runAsyncA :: a -> m b }++deriving via P.WrappedArrow (AsyncA m)+  instance (MonadBaseControl IO m) => P.Profunctor (AsyncA m)+deriving via P.WrappedArrow (AsyncA m)+  instance (MonadBaseControl IO m) => P.Strong (AsyncA m)+deriving via P.WrappedArrow (AsyncA m)+  instance (MonadBaseControl IO m) => P.Choice (AsyncA m)+instance (MonadBaseControl IO m) => P.Traversing (AsyncA m) where+  traverse' (AsyncA f) = AsyncA $ mapConcurrently f+instance (MonadBaseControl IO m) => P.Closed (AsyncA m) where+  closed = P.closedMapping+instance (MonadBaseControl IO m) => P.Mapping (AsyncA m) where+  map' = P.traverseMapping  instance Monad m => Category (AsyncA m) where   id = AsyncA return
src/Control/Arrow/Free.hs view
@@ -12,6 +12,7 @@ {-# LANGUAGE ScopedTypeVariables   #-} {-# LANGUAGE TypeFamilies          #-} {-# LANGUAGE TypeOperators         #-}+{-# LANGUAGE DerivingVia           #-}  -- | Various varieties of free arrow constructions. --@@ -37,26 +38,27 @@   , mapA   , mapSeqA   , filterA+  , hoistErrorChoiceEff+  , expandErrorChoiceEff   , type (~>)   ) where +import           Prelude                (Functor (..))+ import           Control.Arrow-import           Control.Arrow.AppArrow import           Control.Category import           Control.Exception.Safe (Exception, MonadCatch) import qualified Control.Exception.Safe-import           Control.Monad.Trans.Reader-import           Control.Monad.Trans.Writer-import qualified Control.Monad.Trans.Writer.Strict as SW-import           Data.Bool           (Bool)-import           Data.Constraint     (Constraint, Dict (..), mapDict, weaken1,-                                      weaken2)-import           Data.Either         (Either (..))-import           Data.Function       (const, flip, ($))-import           Data.List           (uncons)-import           Data.Maybe          (maybe)-import           Data.Monoid         (Monoid)-import           Data.Tuple          (uncurry)+import           Data.Bool              (Bool)+import           Data.Constraint        (Constraint, Dict (..))+import           Data.Either            (Either (..))+import           Data.Function          (const, flip, ($))+import           Data.List              (uncons)+import           Data.Maybe             (maybe)+import qualified Data.Profunctor        as P+import qualified Data.Profunctor.Cayley as P+import qualified Data.Profunctor.Traversing as P+import           Data.Tuple             (uncurry)  -- | A natural transformation on type constructors of two arguments. type x ~> y = forall a b. x a b -> y a b@@ -77,10 +79,10 @@  -- | Annoying hackery to let us tuple constraints and still use 'effect' --   and 'eval'-class Join ( a :: k -> Constraint) (b :: k -> Constraint) (x :: k) where-  ctx :: Dict (a x, b x)-instance (a x, b x) => Join a b x where-  ctx = Dict+class Join (a :: k -> Constraint) (b :: k -> Constraint) (c :: k -> Constraint) (x :: k) where+  ctx :: (Dict (a x), Dict (b x), Dict (c x))+instance (a x, b x, c x) => Join a b c x where+  ctx = (Dict, Dict, Dict)  -------------------------------------------------------------------------------- -- Arrow@@ -167,17 +169,8 @@ class ArrowError ex a where   try :: a e c -> a e (Either ex c) -instance (ArrowError ex arr) => ArrowError ex (AppArrow (Reader r) arr) where-  try (AppArrow act) = AppArrow $ reader $ \r ->-    try $ runReader act r--instance (ArrowError ex arr, Monoid w) => ArrowError ex (AppArrow (Writer w) arr) where-  try (AppArrow act) = AppArrow $ writer (try a, w)-    where (a, w) = runWriter act--instance (ArrowError ex arr, Monoid w) => ArrowError ex (AppArrow (SW.Writer w) arr) where-  try (AppArrow act) = AppArrow $ SW.writer (try a, w)-    where (a, w) = SW.runWriter act+instance (ArrowError ex arr, Functor f) => ArrowError ex (P.Cayley f arr) where+  try (P.Cayley f) = P.Cayley $ fmap try f  catch :: (ArrowError ex a, ArrowChoice a) => a e c -> a (e, ex) c -> a e c catch a onExc = proc e -> do@@ -192,12 +185,28 @@   => ArrowError ex (Kleisli m) where     try (Kleisli a) = Kleisli $ Control.Exception.Safe.try . a --- | Freely generated arrows with both choice and error handling.+-- | Freely generated arrows with both choice, error handling and ability to+-- traverse structures with Traversals. newtype ErrorChoice ex eff a b = ErrorChoice {-  runErrorChoice :: forall ac. (ArrowChoice ac, ArrowError ex ac)+  runErrorChoice :: forall ac. (ArrowChoice ac, ArrowError ex ac, P.Traversing ac)                  => (eff ~> ac) -> ac a b }+  deriving (P.Profunctor, P.Strong, P.Choice) via P.WrappedArrow (ErrorChoice ex eff) +hoistErrorChoiceEff+  :: (eff ~> eff')+  -> ErrorChoice ex eff a b+  -> ErrorChoice ex eff' a b+hoistErrorChoiceEff f (ErrorChoice ec) = ErrorChoice $ \interp ->+  ec (interp . f)++expandErrorChoiceEff+  :: (forall x y. eff x y -> ErrorChoice ex eff' x y)+  -> ErrorChoice ex eff a b+  -> ErrorChoice ex eff' a b+expandErrorChoiceEff f (ErrorChoice ec) = ErrorChoice $ \interp ->+  ec (\eff -> runErrorChoice (f eff) interp)+ instance Category (ErrorChoice ex eff) where   id = ErrorChoice $ const id   ErrorChoice f . ErrorChoice g = ErrorChoice $ \x -> f x . g x@@ -216,19 +225,22 @@ instance ArrowError ex (ErrorChoice ex eff) where   try (ErrorChoice a) = ErrorChoice $ \f -> try $ a f +instance P.Traversing (ErrorChoice ex eff) where+  traverse' (ErrorChoice a) = ErrorChoice $ \f -> P.traverse' (a f)+  wander trav (ErrorChoice a) = ErrorChoice $ \f -> P.wander trav (a f)+ instance FreeArrowLike (ErrorChoice ex) where-  type Ctx (ErrorChoice ex) = Join ArrowChoice (ArrowError ex)+  type Ctx (ErrorChoice ex) = Join ArrowChoice (ArrowError ex) P.Traversing   effect :: eff a b -> ErrorChoice ex eff a b   effect a = ErrorChoice $ \f -> f a -  eval :: forall eff arr a0 b0. (Join ArrowChoice (ArrowError ex) arr)+  eval :: forall eff arr a0 b0. (Join ArrowChoice (ArrowError ex) P.Traversing arr)        => (eff ~> arr)        -> ErrorChoice ex eff a0 b0        -> arr a0 b0-  eval f a = case (\x -> (mapDict weaken1 x, mapDict weaken2 x)) ctx of-    ( Dict :: Dict (ArrowChoice arr)-      , Dict :: Dict (ArrowError ex arr)-      ) -> runErrorChoice a f+  eval f a = case ctx of+    ( Dict :: Dict (ArrowChoice arr), Dict :: Dict (ArrowError ex arr), Dict :: Dict (P.Traversing arr) )+      -> runErrorChoice a f   --------------------------------------------------------------------------------
src/Control/Funflow.hs view
@@ -1,4 +1,6 @@ {-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes    #-}+{-# LANGUAGE GADTs         #-}  -- | Central Funflow module. --@@ -9,13 +11,14 @@   , type (Base.==>)   , Base.NoEffect   , Base.Flow'(..)-  , Base.Cacher(..)   , Base.ExternalProperties(..)   , Base.MDWriter   , Base.Properties(..)   , Base.EpPurity(..)   , Base.alwaysRecompile-  , Base.defaultCacherWithIdent+  , CS.CacherM(..)+  , CS.Cacher+  , CS.defaultCacherWithIdent   , Cache.defaultCacher   , Cache.defaultCacherLoc     -- * Defines our primitive flow functions@@ -24,16 +27,45 @@   , Class.stepIO   , Class.wrap   , CS.withStore+  , hoistFlowEff+  , expandFlowEff   , module Control.Funflow.Steps   , module Control.Funflow.Exec.Simple   , module Control.Funflow.External   )   where -import qualified Control.Funflow.Base         as Base+import           Control.Arrow.Free+import           Control.Funflow.Base         as Base import qualified Control.Funflow.Cache.TH     as Cache import qualified Control.Funflow.Class        as Class-import qualified Control.Funflow.ContentStore as CS import           Control.Funflow.Exec.Simple import           Control.Funflow.External import           Control.Funflow.Steps+import qualified Data.CAS.ContentStore        as CS++-- | Change the effect type present inside the 'Flow'+hoistFlowEff :: (forall x y. eff x y -> eff' x y) -> Flow eff ex a b -> Flow eff' ex a b+hoistFlowEff f = hoistErrorChoiceEff $ \e -> case e of+  Step p s -> Step p s+  StepIO p s -> StepIO p s+  External p s -> External p s+  PutInStore s -> PutInStore s+  GetFromStore s -> GetFromStore s+  InternalManipulateStore s -> InternalManipulateStore s+  Wrapped p s -> Wrapped p (f s)++-- | Turns each effect in the 'Flow' as a 'Flow' of some other effect+expandFlowEff :: (forall x y. Properties x y -> eff x y -> Flow eff' ex x y)+              -> Flow eff ex a b+              -> Flow eff' ex a b+expandFlowEff f = expandErrorChoiceEff $ \e -> case e of+  Step p s -> Class.step' p s+  StepIO p s -> Class.stepIO' p s+  External p s -> Class.external' p s+  PutInStore s -> Class.putInStore s+  GetFromStore s -> Class.getFromStore s+  InternalManipulateStore s -> Class.internalManipulateStore s+  Wrapped p s -> f p s+-- hoistFlowEff and expandFlowEff are here and not in Control.Funflow.Base to+-- avoid a dependency cycle
src/Control/Funflow/Base.hs view
@@ -16,16 +16,14 @@ import           Control.Arrow                   (Kleisli (..)) import           Control.Arrow.Free import           Control.Exception.Safe          (SomeException)-import           Control.Funflow.ContentHashable-import qualified Control.Funflow.ContentStore    as CS import           Control.Funflow.Diagram import           Control.Funflow.External import           Data.ByteString                 (ByteString)+import           Data.CAS.ContentHashable+import           Data.CAS.ContentStore           as CS import           Data.Default-import           Data.Functor.Identity import           Data.Int                        (Int64) import           Data.Proxy                      (Proxy (..))-import qualified Data.Store                      as Store import qualified Data.Text                       as T import           Path import           Prelude                         hiding (id, (.))@@ -33,30 +31,6 @@  -- | Metadata writer type MDWriter i o = Maybe (i -> o -> [(T.Text, ByteString)])---- | A cacher is responsible for controlling how steps are cached.-data Cacher 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 -> ContentHash-    , cacherStoreValue :: o -> ByteString-      -- | Attempt to read the cache value back. May throw exceptions.-    , cacherReadValue  :: ByteString -> o-    }--defaultCacherWithIdent :: (Store.Store o, ContentHashable Identity i)-                       => Int -- ^ Seed for the cacher-                       -> Cacher i o-defaultCacherWithIdent ident = Cache-  { cacherKey = \i ident' -> runIdentity $ contentHash (ident', ident, i)-  , cacherStoreValue = Store.encode-  , cacherReadValue = Store.decodeEx-  }  data Properties i o = Properties   { -- | Name of this step. Used when describing the step in diagrams
src/Control/Funflow/Cache/TH.hs view
@@ -8,7 +8,7 @@ -- | Template Haskell splices for the funflow cache. module Control.Funflow.Cache.TH where -import           Control.Funflow.Base+import           Data.CAS.ContentStore import           Data.Hashable import           Language.Haskell.TH.Syntax import           System.Random
src/Control/Funflow/Class.hs view
@@ -16,12 +16,12 @@ module Control.Funflow.Class where  import           Control.Arrow-import           Control.Arrow.AppArrow import           Control.Arrow.Free import qualified Control.Funflow.Base            as Base-import           Control.Funflow.ContentHashable-import qualified Control.Funflow.ContentStore    as CS import           Control.Funflow.External+import           Data.CAS.ContentHashable+import qualified Data.CAS.ContentStore           as CS+import qualified Data.Profunctor.Cayley          as P import           Data.Default                    (def) import           Path @@ -67,14 +67,15 @@ wrap = wrap' def  instance ( Applicative app-         , ArrowError ex (AppArrow app (arr eff ex))+         , ArrowError ex (arr eff ex)          , ArrowFlow eff ex (arr eff ex) )-      => ArrowFlow eff ex (AppArrow app (arr eff ex)) where-  step' props f = appArrow $ step' props f-  stepIO' props f = appArrow $ stepIO' props f-  external f = appArrow $ external f-  external' props f = appArrow $ external' props f-  wrap' props eff = appArrow $ wrap' props eff-  putInStore f = appArrow $ putInStore f-  getFromStore f = appArrow $ getFromStore f-  internalManipulateStore f = appArrow $ internalManipulateStore f+      => ArrowFlow eff ex (P.Cayley app (arr eff ex)) where+  step' props f = P.Cayley . pure $ step' props f+  stepIO' props f = P.Cayley . pure $ stepIO' props f+  external f = P.Cayley . pure $ external f+  external' props f = P.Cayley . pure $ external' props f+  wrap' props eff = P.Cayley . pure $ wrap' props eff+  putInStore f = P.Cayley . pure $ putInStore f+  getFromStore f = P.Cayley . pure $ getFromStore f+  internalManipulateStore f = P.Cayley . pure $ internalManipulateStore f+
− src/Control/Funflow/ContentHashable.hs 
@@ -1,549 +0,0 @@-{-# LANGUAGE DefaultSignatures     #-}-{-# LANGUAGE DeriveGeneric         #-}-{-# LANGUAGE FlexibleContexts      #-}-{-# LANGUAGE FlexibleInstances     #-}-{-# LANGUAGE LambdaCase            #-}-{-# LANGUAGE MagicHash             #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings     #-}-{-# LANGUAGE TypeApplications      #-}-{-# LANGUAGE TypeOperators         #-}-{-# LANGUAGE TypeSynonymInstances  #-}-{-# LANGUAGE UnboxedTuples         #-}---- | 'ContentHashable' provides a hashing function suitable for use in the---   Funflow content store.------   This behaves as does a normal hashing function on Haskell types. However,---   on path types, this instead calculates a hash based on the contents of the---   file or directory referenced.------   We also export the 'ExternallyAssuredFile' and 'ExternallyAssuredDirectory'---   types. These instead use the path, file size and modification time to control---   the hash.-module Control.Funflow.ContentHashable-  ( ContentHash-  , toBytes-  , fromBytes-  , ContentHashable (..)-  , contentHashUpdate_binaryFile-  , contentHashUpdate_byteArray#-  , contentHashUpdate_fingerprint-  , contentHashUpdate_primitive-  , contentHashUpdate_storable--  , FileContent (..)-  , DirectoryContent (..)--  , ExternallyAssuredFile(..)-  , ExternallyAssuredDirectory(..)--  , encodeHash-  , decodeHash-  , hashToPath-  , pathToHash--  , SHA256-  , Context-  , Digest-  ) where---import           Control.Exception.Safe           (catchJust)-import           Control.Funflow.Orphans          ()-import           Control.Monad                    (foldM, mzero, (>=>))-import           Control.Monad.IO.Class           (MonadIO, liftIO)-import           Crypto.Hash                      (Context, Digest, SHA256,-                                                   digestFromByteString,-                                                   hashFinalize, hashInit,-                                                   hashUpdate)-import qualified Data.Aeson                       as Aeson-import qualified Data.Aeson.Types                 as Aeson-import           Data.Bits                        (shiftL)-import           Data.ByteArray                   (Bytes, MemView (MemView),-                                                   allocAndFreeze, convert)-import           Data.ByteArray.Encoding          (Base (Base16),-                                                   convertFromBase,-                                                   convertToBase)-import qualified Data.ByteString                  as BS-import           Data.ByteString.Builder.Extra    (defaultChunkSize)-import qualified Data.ByteString.Char8            as C8-import qualified Data.ByteString.Lazy             as BSL-import           Data.Foldable                    (foldlM)-import           Data.Functor.Contravariant-import qualified Data.Hashable-import qualified Data.HashMap.Lazy                as HashMap-import qualified Data.HashSet                     as HashSet-import           Data.Int-import           Data.List                        (sort)-import           Data.List.NonEmpty               (NonEmpty)-import           Data.Map                         (Map)-import qualified Data.Map                         as Map-import           Data.Ratio-import           Data.Scientific-import           Data.Store                       (Store (..), peekException)-import qualified Data.Text                        as T-import qualified Data.Text.Array                  as TA-import qualified Data.Text.Encoding               as TE-import qualified Data.Text.Internal               as T-import qualified Data.Text.Lazy                   as TL-import           Data.Time.Clock                  (UTCTime)-import           Data.Time.Clock.POSIX            (utcTimeToPOSIXSeconds)-import           Data.Typeable-import qualified Data.Vector                      as V-import           Data.Word-import qualified Database.SQLite.Simple.FromField as SQL-import qualified Database.SQLite.Simple.ToField   as SQL-import           Foreign.Marshal.Utils            (with)-import           Foreign.Ptr                      (castPtr)-import           Foreign.Storable                 (Storable, sizeOf)-import           GHC.Fingerprint-import           GHC.Generics-import           GHC.Integer.GMP.Internals        (BigNat (..), Integer (..))-import           GHC.Natural                      (Natural (..))-import           GHC.Prim                         (ByteArray#,-                                                   copyByteArrayToAddr#,-                                                   sizeofByteArray#)-import           GHC.Ptr                          (Ptr (Ptr))-import           GHC.Types                        (IO (IO), Int (I#), Word (W#))-import qualified Path-import qualified Path.Internal-import qualified Path.IO-import           System.IO                        (IOMode (ReadMode),-                                                   withBinaryFile)-import           System.IO.Error                  (isPermissionError)-import           System.IO.Unsafe                 (unsafePerformIO)-import           System.Posix.Files               (fileSize, getFileStatus)---newtype ContentHash = ContentHash { unContentHash :: Digest SHA256 }-  deriving (Eq, Ord, Generic)--instance Aeson.FromJSON ContentHash where-  parseJSON (Aeson.String s)-    | Just h <- decodeHash (TE.encodeUtf8 s) = pure h-    | otherwise = fail "Invalid hash encoding"-  parseJSON invalid-    = Aeson.typeMismatch "ContentHash" invalid-instance Aeson.ToJSON ContentHash where-  toJSON = Aeson.String . TE.decodeUtf8 . encodeHash--instance Data.Hashable.Hashable ContentHash where-  hashWithSalt s = Data.Hashable.hashWithSalt s . encodeHash--instance Show ContentHash where-  showsPrec d h = showParen (d > app_prec)-    $ showString "ContentHash \""-    . (showString $ C8.unpack $ encodeHash h)-    . showString "\""-    where app_prec = 10--instance Store ContentHash where-  size = contramap toBytes size-  peek = fromBytes <$> peek >>= \case-    Nothing -> peekException "Store ContentHash: Illegal digest"-    Just x -> return x-  poke = poke . toBytes--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--toBytes :: ContentHash -> BS.ByteString-toBytes = convert . unContentHash--fromBytes :: BS.ByteString -> Maybe ContentHash-fromBytes bs = ContentHash <$> digestFromByteString bs--hashEncoding :: Base-hashEncoding = Base16---- | File path appropriate encoding of a hash-encodeHash :: ContentHash -> BS.ByteString-encodeHash = convertToBase hashEncoding . toBytes---- | Inverse of 'encodeHash' if given a valid input.------ prop> decodeHash (encodeHash x) = Just x-decodeHash :: BS.ByteString -> Maybe ContentHash-decodeHash bs = case convertFromBase hashEncoding bs of-  Left _  -> Nothing-  Right x -> fromBytes x---- | File path appropriate encoding of a hash-hashToPath :: ContentHash -> Path.Path Path.Rel Path.Dir-hashToPath h =-  case Path.parseRelDir $ C8.unpack $ encodeHash h of-    Nothing -> error-      "[ContentHashable.hashToPath] \-      \Failed to convert hash to directory name"-    Just dir -> dir----- | Inverse of 'hashToPath' if given a valid input.------ prop> pathToHash (hashToPath x) = Just x-pathToHash :: FilePath -> Maybe ContentHash-pathToHash = decodeHash . C8.pack---class Monad m => ContentHashable m a where--  -- | Update a hash context based on the given value.-  ---  -- See 'Crypto.Hash.hashUpdate'.-  ---  -- XXX: Consider swapping the arguments.-  contentHashUpdate :: Context SHA256 -> a -> m (Context SHA256)--  default contentHashUpdate :: (Generic a, GContentHashable m (Rep a))-    => Context SHA256 -> a -> m (Context SHA256)-  contentHashUpdate ctx a = gContentHashUpdate ctx (from a)--  -- | Generate hash of the given value.-  ---  -- See 'Crypto.Hash.hash'.-  contentHash :: a -> m ContentHash-  contentHash x = ContentHash . hashFinalize <$> contentHashUpdate hashInit x----- | Update hash context based on binary in memory representation due to 'Foreign.Storable.Storable'.------ XXX: Do we need to worry about endianness?-contentHashUpdate_storable :: (Monad m, Storable a) => Context SHA256 -> a -> m (Context SHA256)-contentHashUpdate_storable ctx a =-  return . unsafePerformIO $ with a (\p -> pure $! hashUpdate ctx (MemView (castPtr p) (sizeOf a)))---- | Update hash context based on a type's 'GHC.Fingerprint.Type.Fingerprint'.------ The fingerprint is constructed from the library-name, module-name, and name of the type itself.-contentHashUpdate_fingerprint :: (Monad m, Typeable a) => Context SHA256 -> a -> m (Context SHA256)-contentHashUpdate_fingerprint ctx = contentHashUpdate ctx . typeRepFingerprint . typeOf---- | Update hash context by combining 'contentHashUpdate_fingerprint' and 'contentHashUpdate_storable'.--- Intended for primitive types like 'Int'.-contentHashUpdate_primitive :: (Monad m, Typeable a, Storable a) => Context SHA256 -> a -> m (Context SHA256)-contentHashUpdate_primitive ctx a =-  flip contentHashUpdate_fingerprint a >=> flip contentHashUpdate_storable a $ ctx---- | Update hash context based on binary contents of the given file.-contentHashUpdate_binaryFile :: Context SHA256 -> FilePath -> IO (Context SHA256)-contentHashUpdate_binaryFile ctx0 fp = withBinaryFile fp ReadMode $ \h ->-  let go ctx = do-        chunk <- BS.hGetSome h defaultChunkSize-        if BS.null chunk then-          pure ctx-        else-          go $! hashUpdate ctx chunk-  in go ctx0---- | Update hash context based on 'GHC.Prim.ByteArray#'--- by copying into a newly allocated 'Data.ByteArray.Bytes'--- and updating the hash context from there.------ XXX: @'GHC.Prim.byteArrayContents#' :: 'GHC.Prim.ByteArray#' -> 'GHC.Prim.Addr#'@--- could be used together with 'Data.ByteArray.MemView' instead.--- However, 'GHC.Prim.byteArrayContents#' explicitly says, that it is only safe to use--- on a pinned 'GHC.Prim.ByteArray#'.-contentHashUpdate_byteArray# :: ByteArray# -> Int -> Int -> Context SHA256 -> Context SHA256-contentHashUpdate_byteArray# ba (I# off) (I# len) ctx = hashUpdate ctx $-  allocAndFreeze @Bytes (I# len) $ \(Ptr addr) -> IO $ \s ->-    (# copyByteArrayToAddr# ba off addr len s, () #)---- | Update hash context based on the contents of a strict 'Data.Text.Text'.-contentHashUpdate_text :: Context SHA256 -> T.Text -> Context SHA256-contentHashUpdate_text ctx (T.Text arr off_ len_) =-    contentHashUpdate_byteArray# (TA.aBA arr) off len ctx-    where-      off = off_ `shiftL` 1 -- convert from 'Word16' to 'Word8'-      len = len_ `shiftL` 1 -- convert from 'Word16' to 'Word8'--instance Monad m => ContentHashable m Fingerprint where-  contentHashUpdate ctx (Fingerprint a b) = flip contentHashUpdate_storable a >=> flip contentHashUpdate_storable b $ ctx--instance Monad m => ContentHashable m Bool where contentHashUpdate = contentHashUpdate_primitive--instance Monad m => ContentHashable m Char where contentHashUpdate = contentHashUpdate_primitive--instance Monad m => ContentHashable m Int where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Int8 where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Int16 where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Int32 where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Int64 where contentHashUpdate = contentHashUpdate_primitive--instance Monad m => ContentHashable m Word where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Word8 where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Word16 where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Word32 where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Word64 where contentHashUpdate = contentHashUpdate_primitive--instance Monad m => ContentHashable m Float where contentHashUpdate = contentHashUpdate_primitive-instance Monad m => ContentHashable m Double where contentHashUpdate = contentHashUpdate_primitive--instance (ContentHashable m n, Typeable n) => ContentHashable m (Ratio n) where-  contentHashUpdate ctx x =-    flip contentHashUpdate_fingerprint x-    >=> flip contentHashUpdate (numerator x)-    >=> flip contentHashUpdate (denominator x)-    $ ctx--instance Monad m => ContentHashable m Scientific where-  contentHashUpdate ctx x =-    flip contentHashUpdate_fingerprint x-    >=> flip contentHashUpdate (toRational x)-    $ ctx--instance Monad m => ContentHashable m Integer where-  contentHashUpdate ctx n = ($ ctx) $-    flip contentHashUpdate_fingerprint n >=> case n of-      S# i ->-        pure . flip hashUpdate (C8.pack "S") -- tag constructur-        >=> flip contentHashUpdate_storable (I# i) -- hash field-      Jp# (BN# ba) ->-        pure . flip hashUpdate (C8.pack "L") -- tag constructur-        >=> pure . contentHashUpdate_byteArray# ba 0 (I# (sizeofByteArray# ba)) -- hash field-      Jn# (BN# ba) ->-        pure . flip hashUpdate (C8.pack "N") -- tag constructur-        >=> pure . contentHashUpdate_byteArray# ba 0 (I# (sizeofByteArray# ba)) -- hash field--instance Monad m => ContentHashable m Natural where-  contentHashUpdate ctx n = ($ ctx) $-    flip contentHashUpdate_fingerprint n >=> case n of-      NatS# w ->-        pure . flip hashUpdate (C8.pack "S") -- tag constructur-        >=> flip contentHashUpdate_storable (W# w) -- hash field-      NatJ# (BN# ba) ->-        pure . flip hashUpdate (C8.pack "L") -- tag constructur-        >=> pure . contentHashUpdate_byteArray# ba 0 (I# (sizeofByteArray# ba)) -- hash field--instance Monad m => ContentHashable m BS.ByteString where-  contentHashUpdate ctx s =-    flip contentHashUpdate_fingerprint s-    >=> pure . flip hashUpdate s $ ctx--instance Monad m => ContentHashable m BSL.ByteString where-  contentHashUpdate ctx s =-    flip contentHashUpdate_fingerprint s-    >=> pure . flip (BSL.foldlChunks hashUpdate) s $ ctx--instance Monad m => ContentHashable m T.Text where-  contentHashUpdate ctx s =-    flip contentHashUpdate_fingerprint s-    >=> pure . flip contentHashUpdate_text s $ ctx--instance Monad m => ContentHashable m TL.Text where-  contentHashUpdate ctx s =-    flip contentHashUpdate_fingerprint s-    >=> pure . flip (TL.foldlChunks contentHashUpdate_text) s $ ctx--instance (Typeable k, Typeable v, ContentHashable m k, ContentHashable m v)-  => ContentHashable m (Map k v) where-  contentHashUpdate ctx m =-    flip contentHashUpdate_fingerprint m-    >=> flip contentHashUpdate (Map.toList m) $ ctx--instance (Typeable k, Typeable v, ContentHashable m k, ContentHashable m v)-  => ContentHashable m (HashMap.HashMap k v) where-  contentHashUpdate ctx m =-    flip contentHashUpdate_fingerprint m-    -- XXX: The order of the list is unspecified.-    >=> flip contentHashUpdate (HashMap.toList m) $ ctx--instance (Typeable v, ContentHashable m v)-  => ContentHashable m (HashSet.HashSet v) where-  contentHashUpdate ctx s =-    flip contentHashUpdate_fingerprint s-    -- XXX: The order of the list is unspecified.-    >=> flip contentHashUpdate (HashSet.toList s) $ ctx--instance (Typeable a, ContentHashable m a)-  => ContentHashable m [a] where-  contentHashUpdate ctx l =-    flip contentHashUpdate_fingerprint l-    >=> flip (foldM contentHashUpdate) l $ ctx--instance (Typeable a, ContentHashable m a)-  => ContentHashable m (NonEmpty a) where-  contentHashUpdate ctx l =-    flip contentHashUpdate_fingerprint l-    >=> flip (foldlM contentHashUpdate) l $ ctx--instance (Typeable a, ContentHashable m a)-  => ContentHashable m (V.Vector a) where-  contentHashUpdate ctx v =-    flip contentHashUpdate_fingerprint v-    >=> flip (V.foldM' contentHashUpdate) v $ ctx--instance Monad m => ContentHashable m ()-instance (ContentHashable m a, ContentHashable m b) => ContentHashable m (a, b)-instance (ContentHashable m a, ContentHashable m b, ContentHashable m c) => ContentHashable m (a, b, c)-instance (ContentHashable m a, ContentHashable m b, ContentHashable m c, ContentHashable m d) => ContentHashable m (a, b, c, d)-instance (ContentHashable m a, ContentHashable m b, ContentHashable m c, ContentHashable m d, ContentHashable m e) => ContentHashable m (a, b, c, d, e)-instance (Monad m, ContentHashable m a, ContentHashable m b, ContentHashable m c, ContentHashable m d, ContentHashable m e, ContentHashable m f) => ContentHashable m (a, b, c, d, e, f)-instance (Monad m, ContentHashable m a, ContentHashable m b, ContentHashable m c, ContentHashable m d, ContentHashable m e, ContentHashable m f, ContentHashable m g) => ContentHashable m (a, b, c, d, e, f, g)--instance ContentHashable m a => ContentHashable m (Maybe a)--instance (ContentHashable m a, ContentHashable m b) => ContentHashable m (Either a b)--instance Monad m => ContentHashable m Aeson.Value---class Monad m => GContentHashable m f where-  gContentHashUpdate :: Context SHA256 -> f a -> m (Context SHA256)--instance Monad m => GContentHashable m V1 where-  gContentHashUpdate ctx _ = pure ctx--instance Monad m => GContentHashable m U1 where-  gContentHashUpdate ctx U1 = pure ctx--instance ContentHashable m c => GContentHashable m (K1 i c) where-  gContentHashUpdate ctx x = contentHashUpdate ctx (unK1 x)--instance (Constructor c, GContentHashable m f) => GContentHashable m (C1 c f) where-  gContentHashUpdate ctx0 x = gContentHashUpdate nameCtx (unM1 x)-    where nameCtx = hashUpdate ctx0 $ C8.pack (conName x)--instance (Datatype d, GContentHashable m f) => GContentHashable m (D1 d f) where-  gContentHashUpdate ctx0 x = gContentHashUpdate packageCtx (unM1 x)-    where-      datatypeCtx = hashUpdate ctx0 $ C8.pack (datatypeName x)-      moduleCtx = hashUpdate datatypeCtx $ C8.pack (datatypeName x)-      packageCtx = hashUpdate moduleCtx $ C8.pack (datatypeName x)--instance GContentHashable m f => GContentHashable m (S1 s f) where-  gContentHashUpdate ctx x = gContentHashUpdate ctx (unM1 x)--instance (GContentHashable m a, GContentHashable m b) => GContentHashable m (a :*: b) where-  gContentHashUpdate ctx (x :*: y) = gContentHashUpdate ctx x >>= flip gContentHashUpdate y--instance (GContentHashable m a, GContentHashable m b) => GContentHashable m (a :+: b) where-  gContentHashUpdate ctx (L1 x) = gContentHashUpdate ctx x-  gContentHashUpdate ctx (R1 x) = gContentHashUpdate ctx x---- XXX: Do we need this?--- instance GContentHashable (a :.: b) where---   gContentHashUpdate ctx x = _ (unComp1 x)---instance (Monad m, Typeable b, Typeable t) => ContentHashable m (Path.Path b t) where-  contentHashUpdate ctx p@(Path.Internal.Path fp) =-    flip contentHashUpdate_fingerprint p-    >=> flip contentHashUpdate fp-    $ ctx----- | Path to a regular file------ Only the file's content and its executable permission is taken into account--- when generating the content hash. The path itself is ignored.-newtype FileContent = FileContent (Path.Path Path.Abs Path.File)--instance ContentHashable IO FileContent where--  contentHashUpdate ctx (FileContent fp) = do-    exec <- Path.IO.executable <$> Path.IO.getPermissions fp-    ctx' <- if exec then contentHashUpdate ctx () else pure ctx-    contentHashUpdate_binaryFile ctx' (Path.fromAbsFile fp)---- | Path to a directory------ Only the contents of the directory and their path relative to the directory--- are taken into account when generating the content hash.--- The path to the directory is ignored.-newtype DirectoryContent = DirectoryContent (Path.Path Path.Abs Path.Dir)--instance MonadIO m => ContentHashable m DirectoryContent where--  contentHashUpdate ctx0 (DirectoryContent dir0) = liftIO $ do-    (dirs, files) <- Path.IO.listDir dir0-    ctx' <- foldM hashFile ctx0 (sort files)-    foldM hashDir ctx' (sort dirs)-    where-      hashFile ctx fp =-        -- XXX: Do we need to treat symbolic links specially?-        flip contentHashUpdate (Path.filename fp)-        >=> flip contentHashUpdate (FileContent fp)-        $ ctx-      hashDir ctx dir =-        flip contentHashUpdate (Path.dirname dir)-        >=> flip contentHashUpdate (DirectoryContent dir)-        $ ctx--instance Monad m => ContentHashable m UTCTime where-  contentHashUpdate ctx utcTime = let-      secondsSinceEpoch = fromEnum . utcTimeToPOSIXSeconds $ utcTime-    in flip contentHashUpdate_fingerprint utcTime-       >=> flip contentHashUpdate secondsSinceEpoch-         $ ctx---- | Path to a file to be treated as _externally assured_.------   An externally assured file is handled in a somewhat 'cheating' way by---   funflow. The 'ContentHashable' instance for such assumes that some external---   agent guarantees the integrity of the file being referenced. Thus, rather---   than hashing the file contents, we only consider its (absolute) path, size and---   modification time, which can be rapidly looked up from filesystem metadata.------   For a similar approach, see the instance for 'ObjectInBucket' in---   Control.Funflow.AWS.S3, where we exploit the fact that S3 is already---   content hashed to avoid performing any hashing.-newtype ExternallyAssuredFile = ExternallyAssuredFile (Path.Path Path.Abs Path.File)-  deriving (Generic, Show)--instance Aeson.FromJSON ExternallyAssuredFile-instance Aeson.ToJSON ExternallyAssuredFile-instance Store ExternallyAssuredFile--instance ContentHashable IO ExternallyAssuredFile where-  contentHashUpdate ctx (ExternallyAssuredFile fp) = do-    modTime <- Path.IO.getModificationTime fp-    fSize <- fileSize <$> getFileStatus (Path.toFilePath fp)-    flip contentHashUpdate fp-      >=> flip contentHashUpdate modTime-      >=> flip contentHashUpdate_storable fSize-        $ ctx----- | Path to a directory to be treated as _externally assured_.------   For an externally assured directory, we _do_ traverse its contents and verify---   those as we would externally assured files, rather than just relying on the---   directory path. Doing this traversal is pretty cheap, and it's quite likely---   for directory contents to be modified without modifying the contents.------   If an item in the directory cannot be read due to lacking permissions,---   then it will be ignored and not included in the hash. If the flow does not---   have permissions to access the contents of a subdirectory, then these---   contents cannot influence the outcome of a task and it is okay to exclude---   them from the hash. In that case we only hash the name, as that could---   influence the outcome of a task.-newtype ExternallyAssuredDirectory = ExternallyAssuredDirectory (Path.Path Path.Abs Path.Dir)-  deriving (Generic, Show)--instance Aeson.FromJSON ExternallyAssuredDirectory-instance Aeson.ToJSON ExternallyAssuredDirectory-instance Store ExternallyAssuredDirectory--instance ContentHashable IO ExternallyAssuredDirectory where-  contentHashUpdate ctx0 (ExternallyAssuredDirectory dir0) = do-    -- Note that we don't bother looking at the relative directory paths and-    -- including these in the hash. This is because the absolute hash gets-    -- included every time we hash a file.-    (dirs, files) <- Path.IO.listDir dir0-    ctx' <- foldM hashFile ctx0 (sort files)-    foldM hashDir ctx' (sort dirs)-    where-      hashFile ctx fp = contentHashUpdate ctx (ExternallyAssuredFile fp)-        `catchPermissionError` \_ -> contentHashUpdate ctx fp-      hashDir ctx dir = contentHashUpdate ctx (ExternallyAssuredDirectory dir)-        `catchPermissionError` \_ -> contentHashUpdate ctx dir-      catchPermissionError = catchJust $ \e ->-        if isPermissionError e then Just e else Nothing
− src/Control/Funflow/ContentStore.hs 
@@ -1,1032 +0,0 @@-{-# 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 ('Control.Funflow.ContentHashable.ContentHash')--- with an item in the store. An item can either be--- 'Control.Funflow.ContentStore.Missing',--- 'Control.Funflow.ContentStore.Pending', or--- 'Control.Funflow.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 Control.Funflow.ContentStore-  (-  -- * Open/Close-    withStore-  , open-  , close--  -- * 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-  , contentPath-  , 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              (Exception, MonadMask,-                                                      bracket, bracketOnError,-                                                      bracket_,-                                                      displayException, throwIO)-import           Control.Funflow.ContentStore.Notify-import           Control.Funflow.Orphans             ()-import           Control.Lens-import           Control.Monad                       (forM_, forever, unless,-                                                      void, when, (<=<), (>=>))-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 qualified Data.ByteString.Char8               as C8-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           Control.Funflow.ContentHashable     (ContentHash,-                                                      ContentHashable (..),-                                                      DirectoryContent (..),-                                                      contentHashUpdate_fingerprint,-                                                      encodeHash, pathToHash,-                                                      toBytes)-import           Control.Funflow.Lock-import qualified Control.Funflow.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: 'Control.Funflow.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 cacher hash =-  constructIfMissing store cacher 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 cacher hash = constructOrAsync store cacher 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 cacher 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 cacher 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 cacher hash f =-  bracketOnError-    (constructIfMissing store cacher 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 cacher (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: 'Control.Funflow.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.---- | 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-      "[Control.Funflow.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-    ]
− src/Control/Funflow/ContentStore/Notify.hs 
@@ -1,28 +0,0 @@-{-# 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 Control.Funflow.ContentStore.Notify-  ( Notifier-  , initNotifier-  , killNotifier--  , Watch-  , addDirWatch-  , removeDirWatch-  ) where--#ifdef OS_Linux-import           Control.Funflow.ContentStore.Notify.Linux-#else-#  ifdef OS_BSD-import           Control.Funflow.ContentStore.Notify.BSD-#  endif-#endif
− src/Control/Funflow/ContentStore/Notify/BSD.hs 
@@ -1,78 +0,0 @@-{-# LANGUAGE LambdaCase          #-}-{-# LANGUAGE ScopedTypeVariables #-}---- | Implementation of filesystem watch for BSD-based systems using KQueue.-module Control.Funflow.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/Control/Funflow/ContentStore/Notify/Linux.hs 
@@ -1,60 +0,0 @@-{-# LANGUAGE CPP                 #-}-{-# LANGUAGE LambdaCase          #-}-{-# LANGUAGE ScopedTypeVariables #-}---- | Implementation of filesystem watching functionality for linux based on---   inotify.-module Control.Funflow.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/Control/Funflow/Diagram.hs view
@@ -3,6 +3,9 @@ {-# LANGUAGE InstanceSigs          #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE DerivingVia           #-}+{-# LANGUAGE RankNTypes            #-}+ -- | A Diagram is a representation of an arrow with labels. --   No computation is actually performed by a diagram, it just carries --   around a description.@@ -13,6 +16,8 @@ import           Control.Arrow import           Control.Arrow.Free (ArrowError (..)) import           Control.Category+import qualified Data.Profunctor    as P+import qualified Data.Profunctor.Traversing as P import           Data.Proxy         (Proxy (..)) import qualified Data.Text          as T import           Prelude            hiding (id, (.))@@ -34,6 +39,9 @@   Par :: Diagram ex a b -> Diagram ex c d -> Diagram ex (a,c) (b,d)   Fanin :: Diagram ex a c -> Diagram ex b c -> Diagram ex (Either a b) c   Try :: Diagram ex a b -> Diagram ex a (Either ex b)+  Traverse :: Diagram ex a b -> Diagram ex (f a) (f b)+  Wander :: (forall f. (Applicative f) => (a -> f b) -> s -> f t) -> Diagram ex a b -> Diagram ex s t+  deriving (P.Profunctor, P.Strong, P.Choice) via P.WrappedArrow (Diagram ex)  instance Category (Diagram ex) where   id = Node emptyNodeProperties Proxy Proxy@@ -52,6 +60,10 @@  instance ArrowError ex (Diagram ex) where   try = Try++instance P.Traversing (Diagram ex) where+  traverse' = Traverse+  wander = Wander  -- | Construct a labelled node node :: forall arr a b ex. Arrow arr => arr a b -> [T.Text] -> (Diagram ex) a b
src/Control/Funflow/Exec/Simple.hs view
@@ -27,25 +27,22 @@ import           Control.Exception.Safe                      (Exception,                                                               SomeException,                                                               bracket,-                                                              onException,                                                               throwM, try) import           Control.Funflow.Base-import           Control.Funflow.ContentHashable-import qualified Control.Funflow.ContentStore                as CS import           Control.Funflow.External import           Control.Funflow.External.Coordinator import           Control.Funflow.External.Coordinator.Memory import           Control.Funflow.External.Executor           (executeLoop)-import qualified Control.Funflow.RemoteCache                 as Remote+import           Control.Lens                                (Identity (..)) import           Control.Monad.Catch                         (MonadCatch,                                                               MonadMask) import           Control.Monad.IO.Class import           Control.Monad.Trans.Control                 (MonadBaseControl)-import qualified Data.ByteString                             as BS+import           Data.CAS.ContentHashable+import qualified Data.CAS.ContentStore                       as CS+import qualified Data.CAS.RemoteCache                        as Remote import           Data.Foldable                               (traverse_)-import           Data.Maybe import           Data.Monoid                                 ((<>))-import           Data.Void import           Katip import           Path import           System.IO                                   (stderr)@@ -68,35 +65,17 @@           -> Flow eff ex a b           -> a           -> m b-runFlowEx _ cfg store cacher runWrapped confIdent flow input = do+runFlowEx _ cfg store remoteCacher runWrapped confIdent flow input = do     hook <- initialise cfg     runAsyncA (eval (runFlow' hook) flow) input   where-    simpleOutPath item = toFilePath-      $ CS.itemPath store item </> [relfile|out|]-    withStoreCache :: forall i o. Cacher i o+    withStoreCache :: forall i o. CS.Cacher i o                    -> AsyncA m i o -> AsyncA m i o-    withStoreCache c@Cache{} (AsyncA f)-      | Just confIdent' <- confIdent = AsyncA $ \i -> do-      let chash = cacherKey c confIdent' i-          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-      CS.withConstructIfMissing store cacher chash computeAndStore >>= \case-        CS.Missing e -> absurd e-        CS.Pending _ ->-          liftIO (CS.waitUntilComplete store chash) >>= \case-            Just item -> readItem item-            Nothing -> throwM $ CS.FailedToConstruct chash-        CS.Complete (Just a, _) -> return a-        CS.Complete (_, item) -> readItem item-    withStoreCache _ f = f-        +    withStoreCache c (AsyncA f) = AsyncA $+      let c' = c{CS.cacherKey = \ident i -> return $ runIdentity $+                  CS.cacherKey c ident i}+      in CS.cacheKleisliIO confIdent c' store remoteCacher f+           writeMd :: forall i o. ContentHash             -> i             -> o@@ -107,14 +86,13 @@       let kvs = writer i o       in traverse_ (uncurry $ CS.setMetadata store chash) kvs -     runFlow' :: Hook c -> Flow' eff a1 b1 -> AsyncA m a1 b1     runFlow' _ (Step props f) = withStoreCache (cache props)       . AsyncA $ \x -> do           let out = f x           case cache props of-            Cache key _ _ | Just confIdent' <- confIdent ->-              writeMd (key confIdent' x) x out $ mdpolicy props+            CS.Cache key _ _ | Just confIdent' <- confIdent ->+              writeMd (runIdentity $ key confIdent' x) x out $ mdpolicy props             _ -> return ()           return out     runFlow' _ (StepIO props f) = withStoreCache (cache props)@@ -162,24 +140,13 @@               mbStdout <- CS.getMetadataFile store chash [relfile|stdout|]               mbStderr <- CS.getMetadataFile store chash [relfile|stderr|]               throwM $ ExternalTaskFailed td ti mbStdout mbStderr-    runFlow' _ (PutInStore f) = AsyncA $ \x -> katipAddNamespace "putInStore" $ do-      chash <- liftIO $ contentHash x-      CS.constructOrWait store cacher chash >>= \case-        CS.Pending void -> absurd void-        CS.Complete item -> return item-        CS.Missing fp ->-          do-            liftIO $ f fp x-            finalItem <- CS.markComplete store chash-            _ <- Remote.push cacher (CS.itemHash finalItem) (Just chash) (CS.itemPath store finalItem)-            pure finalItem-          `onException`-            (do $(logTM) WarningS . ls $ "Exception in construction: removing " <> show chash-                CS.removeFailed store chash-            )-    runFlow' _ (GetFromStore f) = AsyncA $ \case-      CS.All item -> liftIO . f $ CS.itemPath store item-      item CS.:</> path -> liftIO . f $ CS.itemPath store item </> path+    runFlow' _ (PutInStore f) = AsyncA $+      katipAddNamespace "putInStore"+      . CS.putInStore store remoteCacher+          (\chash -> $(logTM) WarningS . ls $ "Exception in construction: removing " <> show chash)+          (\p -> liftIO . f p)+    runFlow' _ (GetFromStore f) = AsyncA $+      liftIO . f . CS.contentPath store     runFlow' _ (InternalManipulateStore f) = AsyncA $ \i -> liftIO $ f store i     runFlow' _ (Wrapped props w) = withStoreCache (cache props)       $ runWrapped w
src/Control/Funflow/External.hs view
@@ -9,14 +9,15 @@ -- | Definition of external tasks module Control.Funflow.External where -import           Control.Funflow.ContentHashable (ContentHash, ContentHashable, ExternallyAssuredDirectory (..),-                                                  ExternallyAssuredFile (..))-import qualified Control.Funflow.ContentStore    as CS import           Control.Lens.TH import           Data.Aeson                      (FromJSON, ToJSON) #if __GLASGOW_HASKELL__ < 804 import           Data.Semigroup #endif+import           Data.CAS.ContentHashable        (ContentHash, ContentHashable, ExternallyAssuredDirectory (..),+                                                  ExternallyAssuredFile (..))+import qualified Data.CAS.ContentStore           as CS+ import           Data.Store                      (Store) import           Data.String                     (IsString (..)) import qualified Data.Text                       as T
src/Control/Funflow/External/Coordinator.hs view
@@ -16,11 +16,10 @@ module Control.Funflow.External.Coordinator where  import           Control.Exception.Safe-import           Control.Funflow.ContentHashable (ContentHash) import           Control.Funflow.External import           Control.Lens import           Control.Monad.IO.Class          (MonadIO, liftIO)-+import           Data.CAS.ContentHashable        (ContentHash) import           Data.Monoid                     ((<>)) import           Data.Store                      (Store) import           Data.Store.TH                   (makeStore)
src/Control/Funflow/External/Coordinator/Memory.hs view
@@ -7,12 +7,12 @@  import           Control.Concurrent                   (threadDelay) import           Control.Concurrent.STM.TVar-import           Control.Funflow.ContentHashable      (ContentHash) import           Control.Funflow.External import           Control.Funflow.External.Coordinator import           Control.Lens import           Control.Monad.IO.Class               (liftIO) import           Control.Monad.STM+import           Data.CAS.ContentHashable             (ContentHash) import           Data.List                            (find) import qualified Data.Map.Strict                      as M import           System.Clock                         (fromNanoSecs)
src/Control/Funflow/External/Coordinator/Redis.hs view
@@ -14,12 +14,12 @@   , RedisPreconnected (..)   ) where -import qualified Control.Funflow.ContentHashable      as CHash import           Control.Funflow.External import           Control.Funflow.External.Coordinator import           Control.Lens import           Control.Monad.Except import           Control.Monad.Fix                    (fix)+import qualified Data.CAS.ContentHashable             as CHash import           Data.Store import qualified Database.Redis                       as R import           GHC.Conc
src/Control/Funflow/External/Coordinator/SQLite.hs view
@@ -15,14 +15,14 @@  import           Control.Concurrent                   (threadDelay) import           Control.Exception.Safe-import           Control.Funflow.ContentHashable import           Control.Funflow.External import           Control.Funflow.External.Coordinator-import           Control.Funflow.Lock import           Control.Lens import           Control.Monad.IO.Class import qualified Data.Aeson                           as Json import qualified Data.ByteString.Char8                as C8+import           Data.CAS.ContentHashable+import           Data.CAS.Lock import qualified Data.Text                            as T import           Data.Typeable                        (Typeable) import qualified Database.SQLite.Simple               as SQL
src/Control/Funflow/External/Docker.hs view
@@ -16,9 +16,9 @@   ) where  import           Control.Arrow                    (Kleisli (..), second)-import           Control.Funflow.ContentHashable import           Control.Funflow.External import           Control.Monad.Trans.State.Strict+import           Data.CAS.ContentHashable import           Data.Semigroup                   (Semigroup, (<>)) import qualified Data.Text                        as T import           GHC.Generics                     (Generic)
src/Control/Funflow/External/Executor.hs view
@@ -16,16 +16,16 @@ import           Control.Concurrent.Async import           Control.Concurrent.MVar import           Control.Exception.Safe-import qualified Control.Funflow.ContentStore         as CS import           Control.Funflow.External import           Control.Funflow.External.Coordinator-import qualified Control.Funflow.RemoteCache          as Remote import           Control.Lens import           Control.Monad                        (forever, mzero, unless) import           Control.Monad.Trans                  (lift) import           Control.Monad.Trans.Maybe import qualified Data.Aeson                           as Json import qualified Data.ByteString                      as BS+import qualified Data.CAS.ContentStore                as CS+import qualified Data.CAS.RemoteCache                 as Remote import           Data.Foldable                        (for_) import           Data.Maybe                           (isJust, isNothing) import           Data.Monoid                          ((<>))
− src/Control/Funflow/Lock.hs 
@@ -1,120 +0,0 @@-{-# 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 Control.Funflow.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/Control/Funflow/Orphans.hs 
@@ -1,28 +0,0 @@-{-# OPTIONS_GHC -fno-warn-orphans #-}-{-# LANGUAGE FlexibleInstances #-}---- | Dedicated module for orphan instances.-module Control.Funflow.Orphans where--import           Data.Functor.Contravariant-import           Data.Store                 (Store)-import qualified Data.Store                 as Store-import qualified Path                       as Path-import qualified Path.Internal--instance Store (Path.Path Path.Abs Path.File) where-  size = contramap (\(Path.Internal.Path fp) -> fp) Store.size-  peek = Path.Internal.Path <$> Store.peek-  poke = Store.poke . (\(Path.Internal.Path fp) -> fp)-instance Store (Path.Path Path.Abs Path.Dir) where-  size = contramap (\(Path.Internal.Path fp) -> fp) Store.size-  peek = Path.Internal.Path <$> Store.peek-  poke = Store.poke . (\(Path.Internal.Path fp) -> fp)-instance Store (Path.Path Path.Rel Path.File) where-  size = contramap (\(Path.Internal.Path fp) -> fp) Store.size-  peek = Path.Internal.Path <$> Store.peek-  poke = Store.poke . (\(Path.Internal.Path fp) -> fp)-instance Store (Path.Path Path.Rel Path.Dir) where-  size = contramap (\(Path.Internal.Path fp) -> fp) Store.size-  peek = Path.Internal.Path <$> Store.peek-  poke = Store.poke . (\(Path.Internal.Path fp) -> fp)
src/Control/Funflow/Pretty.hs view
@@ -19,6 +19,8 @@   ppDiagram (Par f g) = parens $ ppDiagram f <+>  text "***" <+> ppDiagram g   ppDiagram (Fanin f g) = parens $ ppDiagram f <+>  text "|||" <+> ppDiagram g   ppDiagram (Try f) = parens $ text "try" <+> ppDiagram f+  ppDiagram (Traverse f) = parens $ text "traverse" <+> ppDiagram f+  ppDiagram (Wander _trav f) = parens $ text "wander" <+> ppDiagram f  showFlow :: Flow eff ex a b -> String showFlow = render . ppFlow
− src/Control/Funflow/RemoteCache.hs 
@@ -1,134 +0,0 @@-{-# 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 Control.Funflow.RemoteCache-  ( Cacher(..)-  , PullResult(..), PushResult(..), AliasResult(..)-  , NoCache(..), memoryCache-  , pullAsArchive, pushAsArchive-  ) where--import qualified Codec.Archive.Tar               as Tar-import           Control.Concurrent.MVar-import           Control.Funflow.ContentHashable-import           Control.Monad.IO.Class          (MonadIO, liftIO)-import           Data.ByteString.Lazy            (ByteString)-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
src/Control/Funflow/Steps.hs view
@@ -46,15 +46,15 @@ import           Control.Arrow import           Control.Arrow.Free              (catch) import           Control.Exception.Safe          (Exception, throwM)-import           Control.Funflow.Base            (SimpleFlow, cache,-                                                  defaultCacherWithIdent)+import           Control.Funflow.Base            (SimpleFlow, cache) import           Control.Funflow.Class-import           Control.Funflow.ContentHashable (ContentHashable,+import qualified Control.Funflow.External.Docker as Docker+import           Data.CAS.ContentHashable        (ContentHashable,                                                   DirectoryContent (..),                                                   FileContent (..))-import           Control.Funflow.ContentStore    (Content ((:</>)))-import qualified Control.Funflow.ContentStore    as CS-import qualified Control.Funflow.External.Docker as Docker+import           Data.CAS.ContentStore           (Content ((:</>)),+                                                  defaultCacherWithIdent)+import qualified Data.CAS.ContentStore           as CS import           Data.Default                    (def) import           Data.Foldable                   (for_) import           Data.Store
− test/Funflow/ContentStore.hs 
@@ -1,472 +0,0 @@-{-# OPTIONS_GHC -fno-warn-unused-do-bind #-}-{-# LANGUAGE LambdaCase        #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE QuasiQuotes       #-}-{-# LANGUAGE TypeApplications  #-}--module Funflow.ContentStore-  ( tests-  ) where--import           Control.Concurrent.Async-import           Control.Exception.Safe          (tryAny)-import           Control.Funflow.ContentHashable (contentHash)-import           Control.Funflow.ContentStore    (ContentStore)-import qualified Control.Funflow.ContentStore    as ContentStore-import qualified Control.Funflow.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/Funflow/SQLiteCoordinator.hs view
@@ -12,9 +12,9 @@ import           Control.Concurrent.MVar import           Control.Exception.Safe import           Control.Funflow-import qualified Control.Funflow.ContentStore                as CS import           Control.Funflow.External.Coordinator.SQLite import           Control.Monad+import qualified Data.CAS.ContentStore                as CS import           Data.Semigroup                              ((<>)) import           Data.String                                 (fromString) import           Path
test/Funflow/TestFlows.hs view
@@ -12,11 +12,11 @@ import           Control.Concurrent.Async                    (withAsync) import           Control.Exception.Safe                      hiding (catch) import           Control.Funflow-import           Control.Funflow.ContentStore                (Content ((:</>)))-import qualified Control.Funflow.ContentStore                as CS import           Control.Funflow.External.Coordinator.Memory import           Control.Funflow.External.Executor           (executeLoop) import           Control.Monad                               (when)+import           Data.CAS.ContentStore                       (Content ((:</>)))+import qualified Data.CAS.ContentStore                       as CS import           Data.Default                                (def) import           Data.List                                   (sort) import           Path
test/Test.hs view
@@ -1,5 +1,4 @@ import qualified Control.Arrow.Async.Tests-import qualified Funflow.ContentStore import qualified Funflow.SQLiteCoordinator import qualified Funflow.TestFlows import           Test.Tasty@@ -9,8 +8,7 @@  tests :: TestTree tests = testGroup "Unit Tests"-  [ Funflow.ContentStore.tests-  , Control.Arrow.Async.Tests.tests+  [ Control.Arrow.Async.Tests.tests   , Funflow.TestFlows.tests   , Funflow.SQLiteCoordinator.tests   ]