packages feed

mealstrom (empty) → 0.0.0.1

raw patch · 13 files changed

+1113/−0 lines, 13 filesdep +aesondep +asyncdep +basesetup-changed

Dependencies added: aeson, async, base, bytestring, containers, hashable, list-t, postgresql-simple, resource-pool, stm, stm-containers, tasty, tasty-hunit, text, time, uuid

Files

+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2016 Max Amanshauser++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,154 @@+Mealstrom+=========++Mealstrom is a way of modeling, storing and running (business) processes using PostgreSQL. It is based on an idea that [Jakob Sievers](http://canned.primat.es/) had when we both worked at a payment service provider.++It is a remedy for a number of drawbacks of using relational database systems directly for the same task, while still building on some of their strengths.++You often want to store not just the current state of a process instantiation, but keep a log of all steps taken so far. Obviously you cannot simply update the previous state in a relational database.++* Therefore in a RDBMS you must store *events* in their own right, and have a way to compute the object's current state. You need to implement checks for what constitutes *valid state transitions* manually, again and again for each entity.++* While RDBMS are very powerful, it often feels like you are doing all the work twice, e.g. you write constraints, foreign key checks, triggers etc. and then do all the input validity checking in the client as well, because you do not want to incur the overhead of constantly sending all input to the DB and because relying on parsing the thrown exceptions is often not even possible.++* If you want to make sure that updates are actually applied, keep in mind that database transactions guarantee all-or-nothing handling of your updates, but you do not necessarily know which one of the two happened! Your database connection can drop between when a transaction completes and when control returns to your session. Hence, you need to make all your updates idempotent, and where they are not naturally, you need to add client-generated IDs to your queries (and perhaps use some of the RDBMS' power like triggers). That assumes you actually read the part on transaction isolation in your database manual, because the details are surprisingly tricky and the tiniest mistake can lead to data loss.++In short: If you are not very careful, modeling state transitions in your processes becomes a tangled mess of SQL queries and code, with duplicated functionality and the potential of race conditions and low assurances of correctness.++##Enter Mealstrom++With Mealstrom you model your process as a finite-state automaton, a Mealy machine to be precise. A Mealy machine, in contrast to a Moore machine, is an FSA that attaches effects to transitions instead of states.++Modeling a process as an FSA is the natural way to do it. FSAs have defined states, defined transitions and rules which transitions are permissable in a given state.++You can then create instances of the machine definition and manipulate them using API functions.++A Mealy machine in Mealstrom has the types **State**, **Event** and **Action**, an instance furthermore has a type **Key**. Mealstrom comes with support for `Text` and `UUID` as the Key type. You can have your own Key type, if you make it an instance of `(FSMKey k)` and implement `toText :: k -> Text` and `fromText :: Text -> k`. If you have no preference, it is recommended to use `UUID`.++To persist the machines to PostgreSQL, you need to have Aeson `ToJSON`, `FromJSON` and `Typeable` instances for your four types. Typically, they can be derived generically.++Once you have your four types, you make an instance of `MealyInstance`.++Let's go through an example - A simple system a surgery ward might use to track patients.++```+-- First the language extension and import dance:++{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DeriveAnyClass, DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE MultiWayIf #-}++import Data.Aeson+import Data.Text (Text)+import Data.Typeable+import GHC.Generics++import Mealstrom+import Mealstrom.PostgresJSONStore as PGJSON++type SSN              = Text+data Limb             = Arm | Hand | Leg deriving (Show,Eq,Generic,ToJSON,FromJSON,Typeable)+data PatientStatus    = PatientAdmitted Integer [LimbSurgery]+                      | PatientReleased+                      | PatientDeceased+                  deriving (Show,Eq,Generic,ToJSON,FromJSON,Typeable)+data LimbSurgery      = Removed Limb | Attached Limb deriving (Show,Eq,Generic,ToJSON,FromJSON,Typeable)+data Event            = Operation LimbSurgery | Release | Deceased deriving (Show,Eq,Generic,ToJSON,FromJSON,Typeable)+data Action           = SendBill Integer | SendCondolences deriving (Show,Eq,Generic,ToJSON,FromJSON,Typeable)++instance MealyInstance SSN PatientStatus Event Action++```++There is also a transition function `transition :: (State,Event) -> (State,[Action])`,++as well as an effects function `effects :: Msg Action -> IO Bool`.++You implement `transition` to indicate which transitions are valid and which effects you want to run when a transition occurs.++An `Action` (wrapped in a Msg) is then used to pattern match in `effects` and execute the appropriate code.++NB *Action* is the type you use to represent the *effects* you want to run.++Because the states, events, actions as well as the transition/effects functions are just Haskell data types and code, you can go crazy, but for now let's expand on the simple example above:++```+-- |Calculates current number of specified limb on patient+-- Boldly assumes every patient comes in with full set of limbs+limbsOnPatient :: [LimbSurgery] -> Limb -> Int+limbsOnPatient ops limb =+    foldr (\op acc -> if+            | op == Removed limb  -> acc-1+            | op == Attached limb -> acc+1+            | otherwise           -> acc) 2 ops++cost :: LimbSurgery -> Integer+cost (Removed Arm)   =  5000+cost (Attached Arm)  = 15000+cost (Removed Hand)  =  2000+cost (Attached Hand) =  8000+cost (Removed Leg)   = 12000+cost (Attached Leg)  = 20000++tr (PatientAdmitted bill ls, Operation (Removed l))+    | limbsOnPatient ls l < 1 = error "Cannot remove limb that's not there anymore!"+    | otherwise               = let newbill = bill + cost (Removed l) in+        (PatientAdmitted newbill $ Removed l : ls, [SendCondolences])++tr (PatientAdmitted bill ls, Operation (Attached l))+    | limbsOnPatient ls l > 1 = error "Cannot attach limb, there is no space!"+    | otherwise               = let newbill = bill + cost (Attached l) in+        (PatientAdmitted newbill $ Attached l : ls, [])++tr (PatientAdmitted bill _ls, Release)  = (PatientReleased, [SendBill bill])+tr (PatientAdmitted bill _ls, Deceased) = (PatientDeceased, [SendCondolences, SendBill bill])++tr (PatientReleased, _) = error "Patient escaped, operation invalid."+tr (PatientDeceased, _) = error "Operations on dead patients are not billable"++eff :: Msg Action -> IO Bool+eff (Msg msgId SendCondolences) = putStrLn "not implemented" >> return True+eff (Msg msgId (SendBill bill)) = charge bill :: IO Bool+```+++From wherever you wish to manipulate a Patient instance, you can then use a simple REST-like interface:++```+main = do+    st <- PGJSON.mkStore "host='localhost' port=5432 dbname='butchershop'" "Patient"++    -- You specify transition and effects when creating the Handle for a machine+    -- This is so that you can pass variables to the functions, if you want to.+    let t          = FSMTable tr eff+    let patientFSM = FSMHandle st st t 90 3 :: FSMHandle PostgresJSONStore PostgresJSONStore SSN PatientStatus Event Action++    -- `post` gives you the flexibility of having different start states.+    post patientFSM "123-12-1235" (PatientAdmitted 0 [])+    res <- mkMsgs [Operation (Removed Arm)] >>= patch patientFSM "123-12-1235"++    get patientFSM "123-12-1235"  -- Just (PatientAdmitted 5000 [Removed Arm])+```+++### Reliability+You may have noticed up there, that "patches" are wrapped in Msgs. They are used to give certain reliability guarantees in Mealstrom.++The `FSMAPI` through which you should interact with instances guarantees idempotance. `get` is trivially idempotent, `post` will let you know if the instance already exists and it is safe to retry. Finally, for `patch` you generate a `Msg` using `mkMsg` or `mkMsgs` that wraps an `Event` you want to send to an instance.++Once `patch` returns `True`, you can be assured that the state transition has occurred and the associated Actions are now running asynchronously. You can safely retry `patch`, because when a `msgId` is already known, the message is discarded.++You can run arbitrary effects, they will be retried until a retry limit you set is hit or until they succeed. This means they may happen [more than once] (https://en.wikipedia.org/wiki/Two_Generals'_Problem) or not at all. Failed effects can be retried at any time by calling `recoverAll`.++If, however, you choose to send a Msg to another _MealyInstance_ as an effect, i.e. call `patch` on it in the `effects` function, you can reuse the `msgId` from the first `Msg`. The receiving FSM instance can then even do the same thing, and so on. This way you can form a chain of idempotent updates that will, assuming failures are intermittent, eventually succeed.++### Log+The `FSMAPI` attempt to provide an exception-safe way to work with FSM instances in production. If you want to examine an instances log or alter the past, you can use the functions from the respective stores directly, but have to take care of exceptions yourself.++Lastly, Mealstrom is not a good fit if:++* You require every last bit of performance.+* You do not care particularly whether updates are occasionally lost.+* You require complex, cross-entity queries and/or already have a large amount of query language code, so that the drawbacks cited above do not seem too bad in comparison.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ mealstrom.cabal view
@@ -0,0 +1,79 @@+name:                mealstrom+version:             0.0.0.1+synopsis:            Manipulate FSMs and store them in PostgreSQL.+homepage:            https://github.com/linearray/mealstrom+bug-reports:         https://github.com/linearray/mealstrom/issues+description:++    Mealstrom is a library that allows you to work with Mealy machines,+    a kind of finite-state machine, in Haskell using PostgreSQL for+    persistence.++license:             MIT+license-file:        LICENSE+author:              Max Amanshauser+maintainer:          max@lambdalifting.org+copyright:           Copyright (c) Max Amanshauser 2016+category:            Database, Control+build-type:          Simple+extra-source-files:  README.md+cabal-version:       >=1.10++library+  -- other-modules:       +  -- other-extensions:    +  build-depends:       base              >=4.8       && <4.10+                     , aeson             >= 0.11.2.0 && < 1.1+                     , async             >= 2.1.0    && < 2.2+                     , bytestring        >= 0.10.8.1 && < 0.11+                     , containers        >= 0.5.8.1  && < 0.6+                     , hashable          >= 1.2.4    && < 1.3+                     , list-t            >= 1        && < 2+                     , postgresql-simple >= 0.5.1.2  && < 0.6+                     , resource-pool     >= 0.2.3.2  && < 0.3+                     , stm               >= 2.4.4.1  && < 2.5+                     , stm-containers    >= 0.2.15   && < 0.3+                     , text              >= 1.2.2.1  && < 1.3+                     , time              >= 1.6      && < 1.7+                     , uuid              >= 1.3.12   && < 1.4++  hs-source-dirs:      src+  default-language:    Haskell2010+  exposed-modules:     Mealstrom.FSM,+                       Mealstrom.FSMApi,+                       Mealstrom.FSMEngine,+                       Mealstrom.FSMStore,+                       Mealstrom.FSMTable,+                       Mealstrom.MemoryStore,+                       Mealstrom.PostgresJSONStore,+                       Mealstrom.WALStore+  ghc-options:++test-suite test+  default-language:    Haskell2010+  type:                exitcode-stdio-1.0+  hs-source-dirs:      src+                       test+  main-is:             Main.hs+  ghc-options:         -threaded -with-rtsopts=-N+  build-depends:+                       base              >= 4.8      && < 4.10+                     , aeson             >= 0.11.2.0 && < 1.1+                     , async             >= 2.1.0    && < 2.2+                     , bytestring        >= 0.10.8.1 && < 0.11+                     , hashable          >= 1.2.4    && < 1.3+                     , list-t            >= 1        && < 2+                     , postgresql-simple >= 0.5.1.2  && < 0.6+                     , resource-pool     >= 0.2.3.2  && < 0.3+                     , stm               >= 2.4.4.1  && < 2.5+                     , stm-containers    >= 0.2.15   && < 0.3+                     , text              >= 1.2.2.1  && < 1.3+                     , time              >= 1.6      && < 1.7+                     , tasty             >= 0.11.0.2 && < 0.12+                     , tasty-hunit       >= 0.9.2    && < 1.0+                     , uuid              >= 1.3.12   && < 1.4++source-repository head+  type:     git+  location: git://github.com/linearray/mealstrom.git+
+ src/Mealstrom/FSM.hs view
@@ -0,0 +1,148 @@+{-# LANGUAGE DeriveGeneric, DeriveAnyClass #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}++{-|+Module      : Mealstrom.FSM+Description : Finite State Machine Definitions+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org++These defintions are concerned with the basic functions of+finite state machines, keeping a memory and state transitions.+-}++module Mealstrom.FSM where++import           Data.Aeson+import           Data.Foldable     (asum)+import           Data.Hashable     (Hashable)+import           Data.Maybe        (fromJust, fromMaybe)+import           Data.Text         (Text)+import           Data.Time.Clock+import           Data.Typeable     (Typeable)+import qualified Data.UUID as       UUID+import           Data.UUID         (UUID)+import           Data.UUID.V4+import           GHC.Generics++type MachineTransformer s e a = Machine s e a -> IO (Machine s e a)++-- |A data type that often comes in handy when describing whether+-- updates have succeeded in the backend.+data MealyStatus              = MealyError | Pending | Done deriving (Eq, Show)+++-- |FSMs are uniquely identified by a type k, which must be convertible from/to Text.+class (Hashable k, Eq k) => FSMKey k where+    toText   :: k -> Text+    fromText :: Text -> k++-- |This typeclass is needed to provide a constraint for the FSMStore abstraction.+class (FSMKey k) => MealyInstance k s e a++-- |A change in a FSM is either a (Step Timestamp oldState event newState Actions)+-- or an increase in a counter.+data Change s e a = Step UTCTime s e s [a] | Count Int deriving (Show)++-- |Steps are equal to each other when they originated in the same state+-- received the same event and ended up in the same state+instance (Eq s, Eq e) => Eq (Change s e a) where+    (==) (Count a)             (Count b)             = a == b+    (==) (Step _ os1 e1 ns1 _) (Step _ os2 e2 ns2 _) = (os1 == os2) && (e1 == e2) && (ns1 == ns2)+    (==) (Count _)              Step{}               = False+    (==)  Step{}               (Count _)             = False++data Instance k s e a = Instance {+    key     :: k,+    machine :: Machine s e a+} deriving (Eq,Show,Generic,Typeable)++data Machine s e a = Machine {+    inbox     :: [Msg e],+    outbox    :: [Msg a],+    committed :: [UUID],+    initState :: s,+    currState :: s,+    hist      :: [Change s e a]+} deriving (Eq,Show,Generic,Typeable)++mkEmptyMachine :: s -> Machine s e a+mkEmptyMachine s = Machine [] [] [] s s []++mkEmptyInstance :: k -> s -> Instance k s e a+mkEmptyInstance k s = Instance k (mkEmptyMachine s)++mkInstance :: k -> s -> [Msg e] -> Instance k s e a+mkInstance k s es = Instance k ((mkEmptyMachine s) {inbox = es})+++-- |Type of messages that are sent between FSMs+-- Messages are always identified by UUID.+-- The purpose of Msg is to attach a unique ID to an event, so that+-- certain guarantees can be provided.+data Msg e = Msg {+    msgID       :: Maybe UUID,+    msgContents :: e+} deriving (Show,Eq,Generic)++mkMsg :: t -> IO (Msg t)+mkMsg t = nextRandom >>= \i -> return $ Msg (Just i) t++mkMsgs :: [t] -> IO [Msg t]+mkMsgs = mapM mkMsg++mkBogusMsg :: (Eq t) => t -> Msg t+mkBogusMsg = Msg Nothing++-- |Append a Change to a history.+-- Identical steps are just counted, otherwise they are consed to the history.+histAppend :: (Eq s, Eq e) => Change s e a -> [Change s e a] -> [Change s e a]+histAppend s1 all@(Count i:s2:rest)+    | s1 == s2 = Count (i+1):s2:rest+    | otherwise = s1 : all+histAppend s1 all@(s2:_rest)+    | s1 == s2 = Count 1 : all+    | otherwise = s1 : all+histAppend s ss = s:ss+++-- ##############+-- # JSON Codecs+-- ##############+instance ToJSON UUID where+    toJSON u = toJSON (UUID.toText u)++instance FromJSON UUID where+    parseJSON = withText "UUID" $ \x -> return . fromJust $ UUID.fromText x++instance (ToJSON s, ToJSON e, ToJSON a) => ToJSON (Change s e a) where+    toJSON (Count i) = object [ "count" .= toJSON i]+    toJSON (Step ts os ev ns as) =+        object [+            "timestamp" .= toJSON ts,+            "old_state" .= toJSON os,+            "event"     .= toJSON ev,+            "new_state" .= toJSON ns,+            "actions"   .= toJSON as+        ]++instance (FromJSON s, FromJSON e, FromJSON a) => FromJSON (Change s e a) where+    parseJSON =+        withObject "Change" $ \o ->+            asum [+                Count <$> o .: "count",+                Step  <$> o .: "timestamp" <*> o .: "old_state" <*> o .: "event" <*> o .: "new_state" <*> o .: "actions"+            ]+++-- Other Instances+instance FSMKey Text where+    toText   = id+    fromText = id++instance FSMKey UUID where+    toText     = UUID.toText+    fromText a = fromMaybe (error "Conversion from UUID failed") (UUID.fromText a)
+ src/Mealstrom/FSMApi.hs view
@@ -0,0 +1,115 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE OverloadedStrings #-}++{-|+Module      : Mealstrom.FSMApi+Description : API for FSMs+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org++This is the interface through which you primarily interact with a FSM+from the rest of your program.+-}++module Mealstrom.FSMApi where++import           Control.Concurrent+import           Control.Exception+import           Control.Monad          (void)+import qualified Data.Text           as  Text+import           System.IO+import           System.Timeout++import           Mealstrom.FSM+import           Mealstrom.FSMEngine+import           Mealstrom.FSMStore+import           Mealstrom.FSMTable+import           Mealstrom.WALStore+++data FSMHandle st wal k s e a where+    FSMHandle :: (Eq s, Eq e, Eq a, FSMStore st k s e a, WALStore wal k, FSMKey k) => {+        fsmStore   :: st,                -- ^ Which backend to use for storing FSMs.+        walStore   :: wal,               -- ^ Which backend to use for the WAL.+        fsmTable   :: FSMTable s e a,    -- ^ A table of transitions and effects.+                                         --   This is not in a typeclass, because you may want to use MVars or similar in effects.+                                         --   See the tests for examples.+        effTimeout :: Int,               -- ^ How much time to allow for Actions until they are considered failed.+        retryCount :: Int                -- ^ How often to automatically retry actions.+    } -> FSMHandle st wal k s e a+++get :: forall st wal k s e a . FSMStore st k s e a => FSMHandle st wal k s e a -> k -> IO(Maybe s)+get FSMHandle{..} k = fsmRead fsmStore k (Proxy :: Proxy k s e a)+++-- |Idempotent because of usage of caller-generated keys.+post :: forall st wal k s e a . FSMStore st k s e a =>+        FSMHandle st wal k s e a                                 ->+        k                                                        ->+        s                                                        -> IO Bool+post FSMHandle{..} k s0 =+    fsmCreate fsmStore (mkInstance k s0 [] :: Instance k s e a) >>= \case+        Nothing -> return True+        Just s  -> hPutStrLn stderr s >> return False+++-- |Concurrent updates will be serialised by Postgres.+-- Returns True when the state transition has been successfully computed+-- and actions have been scheduled.+-- Returns False on failure.+patch :: forall st wal k s e a . (FSMStore st k s e a, MealyInstance k s e a, FSMKey k) => FSMHandle st wal k s e a -> k -> [Msg e] -> IO Bool+patch h@FSMHandle{..} k es = do+    openTxn walStore k++    status <- handle (\(e::SomeException) -> hPutStrLn stderr (show e) >> return MealyError)+                     (fsmUpdate fsmStore k ((patchPhase1 fsmTable es) :: MachineTransformer s e a))++    if status /= MealyError+    then recover h k >> return True+    else return False+++-- |Recovering is the process of asynchronously applying Actions. It is performed+-- immediately after the synchronous part of an update and, on failure, retried until it+-- succeeds or the retry limit is hit.+recover :: forall st wal k s e a . (FSMStore st k s e a, MealyInstance k s e a, FSMKey k) => FSMHandle st wal k s e a -> k -> IO ()+recover h@FSMHandle{..} k+    | retryCount == 0 = hPutStrLn stderr $ "Alarma! Recovery retries for " ++ Text.unpack (toText k) ++ " exhausted. Giving up!"+    | otherwise =+        void $ forkFinally (timeout (effTimeout*10^6) (fsmUpdate fsmStore k (patchPhase2 fsmTable :: MachineTransformer s e a))) -- (patchPhase2 fsmTable))+                           (\case Left exn      -> do       -- the damn thing crashed, print log and try again+                                      hPutStrLn stderr $ "Exception occurred while trying to recover " ++ Text.unpack (toText k)+                                      hPrint stderr exn+                                      recover h{retryCount = retryCount - 1} k+                                  Right Nothing -> do       -- We hit the timeout. Try again until we hit the retry limit.+                                      hPutStrLn stderr $ "Timeout while trying to recover " ++ Text.unpack (toText k)+                                      recover h{retryCount = retryCount - 1} k+                                  Right (Just Done)    -> closeTxn walStore k    -- All good.+                                  Right (Just Pending) ->                        -- Some actions did not complete successfully.+                                      recover h{retryCount = retryCount - 1} k)+++-- |During certain long-lasting failures, like network outage, the retry limit of Actions will be exhausted.+-- You should regularly, e.g. ever 10 minutes, call this function to clean up those hard cases.+recoverAll :: forall st wal k s e a . (MealyInstance k s e a) => FSMHandle st wal k s e a -> IO ()+recoverAll h@FSMHandle{..} = do+    wals <- walScan walStore effTimeout+    mapM_ (recover h . walId) wals+++-- |A helper that is sometimes useful+upsert :: forall st wal k s e a . MealyInstance k s e a => FSMStore st k s e a =>+          FSMHandle st wal k s e a -> k -> s -> [Msg e] -> IO ()+upsert h k s es = do+    ms <- get h k+    maybe (post h k s >> void (patch h k es))+          (\_s -> void $ patch h k es)+          ms
+ src/Mealstrom/FSMEngine.hs view
@@ -0,0 +1,82 @@+{-# LANGUAGE RecordWildCards #-}++{-|+Module      : Mealstrom.FSMEngine+Description : Apply changes to the machine and run effects+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org+-}++module Mealstrom.FSMEngine(patchPhase1,patchPhase2) where++import Mealstrom.FSM+import Mealstrom.FSMTable++import Control.Monad (filterM, liftM)+import Data.List+import Data.Time.Clock++-- |patchPhase1 is the part of a "change" to an FSM that happens synchronously.+patchPhase1 :: (Eq s, Eq e) => FSMTable s e a -> [Msg e] -> Machine s e a -> IO (Machine s e a)+patchPhase1 tab es m = getCurrentTime >>= \ts -> eval tab ts (sendMultiple m es)++-- |patchPhase2 is the part of a "change" to an FSM that happens *asynchronously*.+patchPhase2 :: (Eq a) => FSMTable s e a -> Machine s e a -> IO (Machine s e a)+patchPhase2 = apply++-- |Wrapper to send multiple messages at once.+sendMultiple :: Machine s e a -> [Msg e] -> Machine s e a+sendMultiple = foldr (flip send)++-- |See if the message has already been recorded once+-- If not, add it to the inbox.+-- This is where duplicates, resulting from e.g. a crashed client, are filtered out.+send :: Machine s e a -> Msg e -> Machine s e a+send m e =+    let+        msgId (Msg (Just i) _) = i+        ibox = inbox m+    in+        if elem (msgId e) $ map msgId ibox ++ committed m+        then m+        else m {inbox = ibox ++ [e]}++-- |Calculate the state changes in response to a message+eval :: (Eq s, Eq e) => FSMTable s e a -> UTCTime -> Machine s e a -> IO (Machine s e a)+eval FSMTable{..} ts m =+    let+        ibox         = inbox m+        obox         = outbox m+        comm         = committed m+        (ids,events) = foldr (\(Msg (Just i) e) (is,es) -> (i:is,e:es)) ([],[]) ibox+        (newm,as)    = closure transitions ts m events+        asmsgs       = map mkMsg as+    in do+        s <- sequence asmsgs+        return $ newm {inbox = [], outbox = obox ++ s, committed = comm ++ ids}++-- |Take messages from outbox and apply the effects.+-- Failed applications of effects shall remain in the outbox.+apply :: (Eq a) => FSMTable s e a -> Machine s e a -> IO (Machine s e a)+apply FSMTable{..} m = do+    newas <- filterM (liftM not . effects) (outbox m)++    return $ m {outbox = newas}++-- |Apply a list of events to a Memory according to a transition function+closure :: (Eq s, Eq e) => Transitions s e a -> UTCTime -> Machine s e a -> [e] -> (Machine s e a, [a])+closure trans ts m@Machine{..} =+    foldl' (\(mm,oldas) e ->+        let (newm, newas) = step trans ts mm e in+            (newm, oldas ++ newas)+    ) (m,[])++-- |Calculates a new Memory, according to the transition function, for one event.+step :: (Eq s, Eq e) => Transitions s e a -> UTCTime -> Machine s e a -> e -> (Machine s e a, [a])+step trans ts Machine{..} e =+    let+        (newState,as) = trans (currState,e)+        newHist       = histAppend (Step ts currState e newState as) hist+    in+      (Machine inbox outbox committed initState newState newHist, as)
+ src/Mealstrom/FSMStore.hs view
@@ -0,0 +1,23 @@+{-# LANGUAGE MultiParamTypeClasses #-}++{-|+Module      : Mealstrom.FSMStore+Description : Typeclass for FSMStores+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org+-}++module Mealstrom.FSMStore where++import Mealstrom.FSM++data Proxy k s e a = Proxy++-- |Even the Read method needs type parameters 'e' and 'a' because it needs to deserialise the entry from storage.+-- Implementations are expected to not throw exceptions in fsmRead/fsmCreate.+-- Throwing in fsmUpdate is OK.+class FSMStore st k s e a where+    fsmRead   :: st -> k -> Proxy k s e a -> IO (Maybe s)+    fsmCreate :: st -> Instance k s e a -> IO (Maybe String)+    fsmUpdate :: st -> k -> MachineTransformer s e a -> IO MealyStatus
+ src/Mealstrom/FSMTable.hs view
@@ -0,0 +1,25 @@+{-|+Module      : FSMTable+Description : Types for Transitions and Effects+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org+-}++module Mealstrom.FSMTable where++import Mealstrom.FSM++type Transitions s e a = (s,e) -> (s,[a])++-- |Effects are wrapped in Msgs so that the effects function+-- can access the msgId. This is useful when the effects function+-- sends messages of its own, because it can reuse the msgId, thereby+-- creating a message chain with the same Id. Doing so extends guarantees+-- to the receiving FSM.+type Effects a         = Msg a -> IO Bool++data FSMTable s e a    = FSMTable {+    transitions :: Transitions s e a,+    effects     :: Effects a+}
+ src/Mealstrom/MemoryStore.hs view
@@ -0,0 +1,137 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RecordWildCards #-}++{-|+Module      : Mealstrom.MemoryStore+Description : A memory-only storage backend, using STM.+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org+-}+module Mealstrom.MemoryStore (+    MemoryStore,+    mkStore,+    _fsmRead,+    _fsmCreate,+    _fsmUpdate,+    printWal+) where++import           Control.Concurrent.STM+import           Control.Exception+import           Data.Text+import           Data.Time+import qualified ListT+import           STMContainers.Map as Map++import           Mealstrom.FSM+import           Mealstrom.FSMStore+import           Mealstrom.WALStore++instance (MealyInstance k s e a) => FSMStore (MemoryStore k s e a) k s e a where+    fsmRead st k _p = do+        may <- atomically (_fsmRead st k)+        return $ fmap (currState . machine) may+    fsmCreate st a  = atomically $ _fsmCreate st a+    fsmUpdate st k t = _fsmUpdate st k t++instance WALStore (MemoryStore k s e a) k where+    walUpsertIncrement      =              Mealstrom.MemoryStore.walUpsertIncrement+    walDecrement       st k = atomically $ Mealstrom.MemoryStore.walDecrement st k+    walScan                 =              Mealstrom.MemoryStore.walScan++data MemoryStore k s e a where+    MemoryStore :: (MealyInstance k s e a) => {+        memstoreName    :: Text,+        memstoreBacking :: Map k (Instance k s e a),+        memstoreLocks   :: Map k (TMVar ()),+        memstoreWals    :: Map k (UTCTime,Int)+    } -> MemoryStore k s e a++_fsmRead :: MemoryStore k s e a -> k -> STM (Maybe (Instance k s e a))+_fsmRead MemoryStore{..} k = Map.lookup k memstoreBacking >>= \case+    Just a -> return $ Just a+    _      -> return Nothing++-- |For compatibility with the other stores we check existence here+_fsmCreate :: MemoryStore k s e a -> Instance k s e a -> STM (Maybe String)+_fsmCreate MemoryStore{..} ins = do+    exists <- Map.lookup (key ins) memstoreBacking+    maybe (do+            t <- newTMVar ()+            Map.insert t   (key ins) memstoreLocks+            Map.insert ins (key ins) memstoreBacking+            return Nothing+          )+          (\_ -> return $ Just "MemoryStore: Duplicate key")+          exists++-- |We need to use a lock here, because we are in the unfortunate position of+-- having to use IO while performing STM operations, which is not possible.+-- Using the lock we can rest assured no concurrent update operation can progress.+_fsmUpdate :: MemoryStore k s e a -> k -> MachineTransformer s e a -> IO MealyStatus+_fsmUpdate MemoryStore{..} k t =+    let+        m  = memstoreBacking+        ls = memstoreLocks+    in+        atomically (Map.lookup k ls) >>= \lock ->+            maybe (return MealyError)+                  (\l ->+                      bracket_ (atomically $ takeTMVar l)+                               (atomically $ putTMVar l ())+                               (atomically (Map.lookup k m) >>= \res ->+                                   maybe (return MealyError)+                                         (\inst ->+                                             (do+                                                 newMach <- t (machine inst)+                                                 let r = if Prelude.null (outbox newMach) then Done else Pending+                                                 atomically $ Map.insert inst{machine=newMach} k m+                                                 return r+                                             )+                                         ) res)+                  )+                  lock++walUpsertIncrement :: MemoryStore k s e a -> k -> IO ()+walUpsertIncrement MemoryStore{..} k =+    getCurrentTime >>= \t -> atomically $+        Map.lookup k memstoreWals >>= \res ->+            maybe (Map.insert (t,1) k memstoreWals)+                  (\(_oldt,w) -> Map.insert (t,w+1) k memstoreWals)+                  res++walDecrement :: MemoryStore k s e a -> k -> STM ()+walDecrement MemoryStore{..} k =+    Map.lookup k memstoreWals >>= \res ->+        maybe (error "trying to recover non-existing entry")+              (\(t,w) -> Map.insert (t,w-1) k memstoreWals)+              res++walScan :: MemoryStore k s e a -> Int -> IO [WALEntry k]+walScan MemoryStore{..} cutoff =+    getCurrentTime >>= \t -> atomically $+        let xx = addUTCTime (negate (fromInteger (toInteger cutoff) :: NominalDiffTime)) t in++        ListT.fold (\acc (k,(t,w)) -> if t < xx+                                    then return (WALEntry k t w : acc)+                                    else return acc) [] (stream memstoreWals)+++printWal :: MemoryStore k s e a -> k -> IO ()+printWal MemoryStore{..} k =+    atomically (Map.lookup k memstoreWals) >>= \res ->+        maybe (putStrLn "NO WAL")+              print+              res+++mkStore :: (MealyInstance k s e a) => Text -> IO (MemoryStore k s e a)+mkStore name = atomically $ do+    back  <- new :: STM (Map k (Instance k s e a))+    locks <- new :: STM (Map k (TMVar ()))+    wals  <- new :: STM (Map k (UTCTime,Int))+    return $ MemoryStore name back locks wals
+ src/Mealstrom/PostgresJSONStore.hs view
@@ -0,0 +1,267 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE UndecidableInstances #-}++{-|+Module      : Mealstrom.PostgresJSONStore+Description : Main backend for FSMs and WALs.+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org++This module is the main backend for FSMs. Instances are stored in a table+with the name passed as storeName when creating the PostgresJSONStore. WALs use+the same name with "Wal" appended.+-}++module Mealstrom.PostgresJSONStore(+    PostgresJSONStore,+    mkStore,+    _fsmRead,+    _fsmCreate,+    _fsmUpdate+) where+++import           Control.Exception                           (handle,SomeException)+import           Control.Monad                               (void)+import           Database.PostgreSQL.Simple                as PGS+import           Database.PostgreSQL.Simple.FromRow+import           Database.PostgreSQL.Simple.ToField+import           Database.PostgreSQL.Simple.Transaction+import           Database.PostgreSQL.Simple.Types+import           Data.Aeson+import qualified Data.ByteString.Char8                     as DBSC8+import           Data.Int                                    (Int64)+import           Data.Maybe                                  (listToMaybe)+import           Data.Pool+import           Data.Text+import           Data.Time+import           Data.Typeable                        hiding (Proxy)+import           GHC.Generics+import           Database.PostgreSQL.Simple.FromField        (FromField (fromField),+                                                              fromJSONField,+                                                              Conversion)++import           Mealstrom.FSM+import           Mealstrom.FSMStore+import           Mealstrom.WALStore++data PostgresJSONStore = PostgresJSONStore {+    storeConnPool :: Pool Connection,+    storeName     :: Text+}++instance (FromJSON s, FromJSON e, FromJSON a,+          ToJSON   s, ToJSON   e, ToJSON   a,+          Typeable s, Typeable e, Typeable a,+          MealyInstance k s e a)              => FSMStore PostgresJSONStore k s e a where+    fsmRead st k p = Mealstrom.PostgresJSONStore._fsmRead st k p >>= \mi -> return $ fmap (currState . machine) mi+    fsmCreate      = Mealstrom.PostgresJSONStore._fsmCreate+    fsmUpdate      = Mealstrom.PostgresJSONStore._fsmUpdate++instance (FSMKey k) => WALStore PostgresJSONStore k where+    walUpsertIncrement = Mealstrom.PostgresJSONStore.walUpsertIncrement+    walDecrement       = Mealstrom.PostgresJSONStore.walDecrement+    walScan            = Mealstrom.PostgresJSONStore.walScan++-- |We create a database pool (no subpools) of 20 connections that will be closed+-- after 10 seconds of inactivity.+givePool :: IO Connection -> IO (Pool Connection)+givePool creator = createPool creator close 1 10 20+++-- #########+-- # FSM API+-- #########+_fsmRead :: (FromJSON s, FromJSON e, FromJSON a,+             Typeable s, Typeable e, Typeable a,+             MealyInstance k s e a)              =>+             PostgresJSONStore                   ->+             k                                   ->+             Proxy k s e a                       -> IO (Maybe (Instance k s e a))+_fsmRead st k _p =+    withResource (storeConnPool st) (\conn ->+        withTransactionSerializable conn $ do+            el <- _getValue conn (storeName st) (toText k)+            return $ listToMaybe el)+++_fsmCreate :: forall k s e a .+              (ToJSON   s, ToJSON   e, ToJSON   a,+               Typeable s, Typeable e, Typeable a,+               MealyInstance k s e a)              =>+               PostgresJSONStore                   ->+               Instance k s e a                    -> IO (Maybe String)+_fsmCreate st i =+    handle (\(e::SomeException) -> return $ Just (show e))+           (withResource (storeConnPool st) (\conn ->+               withTransactionSerializable conn $ do+                   void $ _postValue conn (storeName st) (toText $ key i) (machine i)+                   return Nothing))+++-- |Postgresql-simple exceptions will be caught by `patch` in FSMApi.hs+-- In principle all transaction isolation levels offered by Postgres are safe+-- here, because we do explicit locking in _getValueForUpdate.+-- However things become more interesting when considering that you can do+-- arbitrary queries in effects, either using the functions in this+-- module or otherwise.++-- We use Serializable here, because it involves no extra cost in our case, and+-- it provides safety when used in arbitrary ways in effects.+-- Hence,+-- * Serializable is recommended and safe.+-- * Repeatable Read, or in PostgreSQL's case Snapshot Isolation, does *not* protect+--   against write skew, which means that if two Actions perform reads and based+--   on the result update data, one of the two updates may be lost.+-- * Read Committed means the usual caveats apply (Nonrepeatable reads, Phantom reads, Write skew…).+--+--   If you are not careful you may end up with wrong data or attempts to insert data+--   with a duplicate ID…+--   Hence, when in doubt, do not lower the isolation level.+_fsmUpdate :: forall k s e a .+              (FromJSON s, FromJSON e, FromJSON a,+               ToJSON   s, ToJSON   e, ToJSON   a,+               Typeable s, Typeable e, Typeable a,+               MealyInstance k s e a)              =>+               PostgresJSONStore                   ->+               k                                   ->+               MachineTransformer s e a            -> IO MealyStatus+_fsmUpdate st k t =+    withResource (storeConnPool st) (\conn ->+        withTransactionSerializable conn $ do+            el <- _getValueForUpdate conn (storeName st) (toText k) :: IO [Instance k s e a]+            let entry = listToMaybe el++            maybe+                (return MealyError)+                (\e -> do+                    newMachine <- t (machine e)+                    void (_postOrUpdateValue conn (storeName st) (toText k) newMachine)+                    return $ if Prelude.null (outbox newMachine) then Done else Pending)+                entry)+++-- #####+-- # WAL+-- #####+_createWalTable :: Connection -> Text -> IO Int64+_createWalTable conn name =+    PGS.execute conn "CREATE TABLE IF NOT EXISTS ? ( id TEXT PRIMARY KEY, date timestamptz NOT NULL, count int NOT NULL )" (Only (Identifier name))++-- |Updates a WALEntry if it exists, inserts a new WALEntry if is is missing.+walUpsertIncrement :: (FSMKey k) => PostgresJSONStore -> k -> IO ()+walUpsertIncrement st i =+    _walExecute st i _walIncrement++walDecrement :: (FSMKey k) => PostgresJSONStore -> k -> IO ()+walDecrement st i =+    _walExecute st i _walDecrement++_walExecute :: (FSMKey k) => PostgresJSONStore -> k -> Query -> IO ()+_walExecute st k q = let tbl = append (storeName st) "Wal" in+    withResource (storeConnPool st) (\conn ->+        withTransactionSerializable conn $ do+            now   <- getCurrentTime+            void $ PGS.execute conn q (Identifier tbl, toText k, now, Identifier tbl))++_walIncrement :: Query+_walIncrement = "INSERT INTO ? VALUES (?,?,1) ON CONFLICT (id) DO UPDATE SET count = ?.count + 1, date = EXCLUDED.date"++_walDecrement :: Query+_walDecrement = "INSERT INTO ? VALUES (?,?,0) ON CONFLICT (id) DO UPDATE SET count = ?.count - 1"+++-- |Returns a list of all transactions that were not successfully terminated+-- and are older than `cutoff`.+walScan :: (FSMKey k) => PostgresJSONStore -> Int -> IO [WALEntry k]+walScan st cutoff = do+    t <- getCurrentTime+    let xx = addUTCTime (negate (fromInteger (toInteger cutoff) :: NominalDiffTime)) t++    withResource (storeConnPool st) (\c ->+        withTransactionSerializable c $+            PGS.query c "SELECT * FROM ? WHERE date < ? AND count > 0" (Identifier $ append (storeName st) "Wal", xx))++-- |Creates a postgresql store+mkStore :: String -> Text -> IO PostgresJSONStore+mkStore connStr name =+    let+        connBS = DBSC8.pack connStr+    in do+        pool <- givePool (PGS.connectPostgreSQL connBS)+        _    <- withResource pool $ flip _createFsmTable name+        _    <- withResource pool $ flip _createWalTable (append name "Wal")+        return $ PostgresJSONStore pool name++_createFsmTable :: Connection -> Text -> IO Int64+_createFsmTable conn name =+    PGS.execute conn "CREATE TABLE IF NOT EXISTS ? ( id text PRIMARY KEY, data jsonb NOT NULL)" (Only (Identifier name))++-- SELECT .. FOR UPDATE locks the rows matching the query. Concurrent+-- (repeatable read and serializable) transactions will block and+-- abort once the new value has been inserted. Since we run effects+-- between SELECT and INSERT, this is what we want.+-- Concurrent SELECTS (without FOR UPDATE) will be unaffected.+_getValue :: (FromRow v) => Connection -> Text -> Text -> IO [v]+_getValue c tbl k =+    PGS.query c "SELECT * FROM ? WHERE id = ?" (Identifier tbl, k)++_getValueForUpdate :: (FromRow v) => Connection -> Text -> Text -> IO [v]+_getValueForUpdate c tbl k =+    PGS.query c "SELECT * FROM ? WHERE id = ? FOR UPDATE" (Identifier tbl, k)++_postOrUpdateValue :: (ToField v) => Connection -> Text -> Text -> v -> IO Int64+_postOrUpdateValue c tbl k v =+    PGS.execute c "INSERT INTO ? VALUES (?,?) ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data" (Identifier tbl, k, v)++_postValue :: (ToField v) => Connection -> Text -> Text -> v -> IO Int64+_postValue c tbl k v =+    PGS.execute c "INSERT INTO ? VALUES (?,?)" (Identifier tbl, k, v)++_deleteValue :: (ToField k) => Connection -> Text -> k -> IO Int64+_deleteValue c tbl k =+    PGS.execute c "DELETE FROM ? WHERE id = ?" (Identifier tbl, k)++_queryValue :: (FromRow v) => Connection -> Text -> Text -> IO [v]+_queryValue c tbl q =+    PGS.query c "SELECT * FROM ? WHERE data @> ?" (Identifier tbl, q)+++-- |Instance to convert one DB row to an instance of Instance ;)+-- users of this module must provide instances for ToJSON, FromJSON for `s`, `e` and `a`.+instance (ToJSON s,   ToJSON e,   ToJSON a)   => ToJSON (Machine s e a)+instance (FromJSON s, FromJSON e, FromJSON a) => FromJSON (Machine s e a)++instance (ToJSON e)   => ToJSON (Msg e)+instance (FromJSON e) => FromJSON (Msg e)++instance (Typeable s, Typeable e, Typeable a,+          FromJSON s, FromJSON e, FromJSON a, FSMKey k) => FromRow (Instance k s e a) where+    fromRow = Instance <$> field <*> field++instance (Typeable s, Typeable e, Typeable a,+          FromJSON s, FromJSON e, FromJSON a) => FromField (Machine s e a) where+    fromField = fromJSONField++instance (Typeable s, Typeable e, Typeable a,+          ToJSON s, ToJSON e, ToJSON a)       => ToField (Machine s e a) where+    toField = toJSONField++instance {-# OVERLAPS #-} (FSMKey k) => ToField k where+    toField k = toField (toText k)++instance {-# OVERLAPS #-} (FSMKey k) => FromField k where+    fromField f mdata = fmap fromText (fromField f mdata :: Conversion Text)++instance (FSMKey k) => FromRow (WALEntry k) where+    fromRow = WALEntry <$> field <*> field <*> field++deriving instance (FSMKey k) => Generic  (WALEntry k)+deriving instance (FSMKey k) => Typeable (WALEntry k)
+ src/Mealstrom/WALStore.hs view
@@ -0,0 +1,33 @@+{-# LANGUAGE MultiParamTypeClasses #-}++{-|+Module      : Mealstrom.WALStore+Description : Store WALEntries+Copyright   : (c) Max Amanshauser, 2016+License     : MIT+Maintainer  : max@lambdalifting.org++A WALStore is anything being able to store WALEntries.+WALEntries indicate how often a recovery process has been started for+an instance.+-}+module Mealstrom.WALStore where++import Data.Time.Clock++class WALStore st k where+    walUpsertIncrement :: st -> k -> IO ()+    walDecrement       :: st -> k -> IO ()+    walScan            :: st -> Int  -> IO [WALEntry k]++data WALEntry k = WALEntry {+    walId    :: k,+    walTime  :: UTCTime,+    walCount :: Int+} deriving (Show,Eq)++openTxn :: WALStore st k => st -> k -> IO ()+openTxn = walUpsertIncrement++closeTxn :: WALStore st k => st -> k -> IO ()+closeTxn = walDecrement
+ test/Main.hs view
@@ -0,0 +1,28 @@+import BasicFSM   (runBasicTests)+import FSM2FSM    (runFSM2FSMTests)+import CounterFSM (runCounterTests)+import Recovery   (runRecoveryTests)+import Timeout    (runTimeoutTests)+import Exception  (runExceptionTests)++import Database.PostgreSQL.Simple+import Database.PostgreSQL.Simple.Types++import Data.ByteString.Char8             as DBSC8++import Test.Tasty++main :: IO ()+main =+    let c = "host='localhost' port=5432 dbname='fsmtest'" in do+        conn <- connectPostgreSQL (DBSC8.pack c)+        _    <- execute_ conn $ Query (DBSC8.pack "DROP SCHEMA public CASCADE; CREATE SCHEMA public;")++        defaultMain $ testGroup "All tests" [+            runBasicTests c,+            runFSM2FSMTests c,+            runCounterTests c,+            runRecoveryTests c,+            runTimeoutTests c,+            runExceptionTests c+            ]