Batching Monadic Requests

with Free Interpreters

A love story of free monad & free applicative

Cary Robbins

June 6, 2019

Prerequisite Concepts

Applicative
Monad
Free (Monad and Applicative)
GADTs (data Foo a where)
Transformers (ReaderT, ExceptT)
Natural transformations (f a -> g a)
Universal quantification (forall, Data.Some)

Does this only work in Haskell?

Using just Haskell2010!

jk jk

`{-# LANGUAGE AllowAmbiguousTypes, ConstraintKinds, DataKinds, DefaultSignatures, DeriveAnyClass, DeriveGeneric, DerivingStrategies, DerivingVia, DuplicateRecordFields, FlexibleContexts, FlexibleInstances, GADTs, GeneralizedNewtypeDeriving, KindSignatures, LambdaCase, NamedFieldPuns, OverloadedLabels, OverloadedStrings, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UndecidableInstances #-}`

Can you imagine a world without language extensions?

Free Library

Just using plain ol' free

Problem

Business Inc. needs to interact with the Google Calendar API

However...

We're getting rate limited!

Solution

Google provides a batch API, let's just use that!

A quick derail into monads


runThings = do
  x <- thingX
  y <- thingY
  z <- thingZ
  pure (x, y, z)


runThings =
  thingX >>= \x ->
    thingY >>= \y ->
      thingZ >>= \z ->
        pure (x, y, z)

Applicative to the rescue?


runThings = do
  x <- thingX
  y <- thingY
  z <- thingZ
  pure (x, y, z)


runThings =
  (,,)
    <$> thingX
    <*> thingY
    <*> thingZ

Business Inc. Command Requirements

InsertEvent
- Attempt insert; on 409, fall back to update
UpdateEvent
- Attempt update; on 404, fall back to insert
DeleteEvent
- Cancel the event via a delete, permitting 404 and 410
EventExists
- Return true if the event exists and is not cancelled, false otherwise

Business Inc. Scope Creep

Mutable commands (insert, update, delete) should also -

Report what action was taken (e.g. updated via insert fallback)
Return the Google calendar event id for verification

Insert Event Prototype


insertEvent
  :: G.Client         -- ^ Google API client
  -> BEvent           -- ^ Business event
  -> IO BInsertResult -- ^ Business result
insertEvent gClient bEvent = do
  G.insert gClient bEvent >>= \case
    Right res -> pure $ Inserted $ key res
    Left e0 -> do
      when (status e0 /= 409) $ throwIO e0
      G.update gClient bEvent >>= \case
        Right res -> pure $ UpdatedViaInsert $ key res
        Left e1 -> throwIO e1

Can you spot the problem?

Taking a step back

How do we even batch anything in the first place?

Google Calendar Algebra


(insertRes, getRes) <- G.runClient gClient $
  (,)
    <$> G.mkRequest (G.EventsInsert gTok calId event)
    <*> G.mkRequest (G.EventsGet    gTok calId eid)

Google Calendar Algebra


type ActionResult a = Action (Either ClientFailure a)
data Action a where
CalendarsGet
:: AccessToken
-> KeyOf Calendar
-> ActionResult (Maybe (Keyed Calendar))
EventsInsert
:: AccessToken
-> KeyOf Calendar
-> CalendarEvent
-> ActionResult (Keyed CalendarEvent)
-- and so on

Free Applicative Refresher


module Control.Applicative.Free where

data Ap f a where
  Pure :: a -> Ap f a
  Ap   :: f a -> Ap f (a -> b) -> Ap f b

liftAp :: f a -> Ap f a

Free Applicative Request


import qualified Control.Applicative.Free as FreeAp

type Request = FreeAp.Ap Action

mkRequest :: Action a -> Request a
mkRequest = FreeAp.liftAp

Google Calendar Client


newtype Client f = Client
  { runClient :: forall a. Request a -> f a }

Now just implement it!

Making a batched client

Dependency inversion isn't just for OO


mkBatchedClient :: forall m. (MonadIO m) => HttpHandler m -> Client m


_

Making a batched client

Dependency inversion isn't just for OO


mkBatchedClient :: forall m. (MonadIO m) => HttpHandler m -> Client m

newtype HttpHandler f = HttpHandler
  { runHttpHandler :: RequestPayload -> f BatchResponse }

HttpHandler


mkBatchedHttpHandler
  :: (MonadIO m, MonadThrow m) => HTTPClient.Manager -> HttpHandler m


_

HttpHandler


mkBatchedHttpHandler
  :: (MonadIO m, MonadThrow m) => HTTPClient.Manager -> HttpHandler m

httpHandlerWithCounter
  :: (MonadIO m) => IORef Int -> HttpHandler m -> HttpHandler m

The crucial command


data Command a = Command { key :: UUID, action :: Action a }




_

The crucial command


data Command a = Command { key :: UUID, action :: Action a }

data RequestPayload = RequestPayload
  { boundary :: ByteString
  , commands :: [Some Command]
  }

The client for real


mkBatchedClient
  :: forall m. (MonadIO m) => HttpHandler m -> Client m
mkBatchedClient httpHandler = Client {runClient}
  where
  runClient :: forall a. Request a -> m a
  runClient req = do
    cmd <- runApComposed genUUID req

Backtrack to runAp


runAp :: Applicative g => (forall x. f x -> g x) -> Ap f a -> g a


runApComposed
  :: (Applicative g, Applicative h)
  => (forall x. f x -> g (h x)) -> Ap f a -> g (h a)


-- Create a unique UUID for each request
genUUID :: Action a -> m (FreeAp.Ap Command a)
genUUID a = do
  u <- liftIO UUID.nextRandom
  pure $ FreeAp.liftAp $ Command u a

The Client for real


mkBatchedClient
  :: forall m. (MonadIO m) => HttpHandler m -> Client m
mkBatchedClient httpHandler logger = Client {runClient}
  where
  runClient :: forall a. Request a -> m a
  runClient req = do
    cmd <- runApComposed genUUID req
    -- Accumulate all the commands into a list
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd

Backtrack to runAp_


runAp_ :: Monoid m => (forall a. f a -> m) -> Ap f b -> m

-- What were we doing again?
runAp_ (\c -> [This c]) cmd

-- Specialized
runAp_ :: (forall a. Command a -> [Some Command])
       -> Ap Command b
       -> [Some Command]

The client for real


mkBatchedClient
  :: forall m. (MonadIO m) => HttpHandler m -> Client m
mkBatchedClient httpHandler logger = Client {runClient}
  where
  runClient :: forall a. Request a -> m a
  runClient req = do
    cmd <- runApComposed genUUID req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let reader = FreeAp.runAp mkReader cmd

Batched reader

Build reader functions which can decode from our response map


mkReader
  :: Command a
  -> ReaderT (Map UUID EncodedResponse) DecodeResult a
mkReader (Command uuid action) = ReaderT $ \env ->
  decodeResponse action $ env ! uuid

How does decodeResponse work?


class Decoder a where
  decode :: EncodedResponse -> DecodeResult a


decodeResponse :: Action a -> EncodedResponse -> Response.DecodeResult a
decodeResponse action r = case action of
  CalendarsGet        {} -> Decoder.decode r
  EventsInsert        {} -> Decoder.decode r
  EventsInsertWithKey {} -> Decoder.decode r
  EventsUpdate        {} -> Decoder.decode r
  EventsDelete        {} -> Decoder.decode r
  EventsGet           {} -> Decoder.decode r

The client for real


mkBatchedClient
  :: forall m. (MonadIO m) => HttpHandler m -> Client m
mkBatchedClient httpHandler logger = Client {runClient}
  where
  runClient :: forall a. Request a -> m a
  runClient req = do
    cmd <- runApComposed genUUID req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let reader = FreeAp.runAp mkReader cmd
    env <- runCommands cmds

runCommands


runCommands :: [Some Command] -> m (Map UUID EncodedResponse)
runCommands cmds = do
  mconcat <$> traverse runChunk (List.chunksOf batchSize cmds)
  where
  batchSize = 50 -- Google's maximum batch size

  runChunk :: [Some Command] -> m (Map UUID EncodedResponse)

The Client for real


mkBatchedClient
  :: forall m. (MonadIO m) => HttpHandler m -> Client m
mkBatchedClient httpHandler logger = Client {runClient}
  where
  runClient :: forall a. Request a -> m a
  runClient req = do
    cmd <- runApComposed genUUID req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let reader = FreeAp.runAp mkReader cmd
    env <- runCommands cmds
    case Response.unDecodeResult $ runReaderT reader env of
      Right x -> pure x
      Left e -> throwIO e

Yay, we can batch!

...Applicatives?

I thought we were batching monads?

Free monads aren't the solution


module Control.Monad.Free where

data Free f a =
    Pure a
  | Free (f (Free f a))

But they could be part of the solution!

What if...


 BInsertEvent BEventExists BUpdateEvent
      ▼           ▼            ▼     
    insert      exists       update  ▶ [insert, exists, update]
                                                  ▼            
                                               Request         
                                                  ▼             
    Conflict    Pure(true)  Not Found ◀        Response
      ▼                       ▼      
     >>=                     >>=     
    update                  insert   ▶    [update, insert]
                                                  ▼             
                                               Request         
                                                  ▼             
   Pure(ok)                Pure(ok)  ◀        Response

Business Inc Algebra


type BActionResult a = BAction (Either BFailure a)

data BAction a where
  BInsertEvent :: UserId -> BEvent   -> BActionResult BInsertResult
  BDeleteEvent :: UserId -> BEventId -> BActionResult BDeleteResult
  BEventExists :: UserId -> BEventId -> BActionResult Bool
  BUpdateEvent :: UserId -> BEvent   -> BActionResult BUpdateResult

type BRequest = FreeAp.Ap BAction

mkBRequest :: BAction a -> BRequest a
mkBRequest = FreeAp.liftAp

Business Inc Client


newtype BClient f = BClient
  { runBClient :: forall a. BRequest a -> f a }

mkGoogleBusinessClient
  :: forall m. (MonadIO m)
  => TokenProvider m -> G.Client m -> BClient m
mkGoogleBusinessClient tokenProvider gClient = BClient {runBClient}
  where
  runBClient :: BRequest a -> m a
  runBClient req = do
    cmd <- runApComposed genUUID req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    

_

Business Inc Client


newtype BClient f = BClient
  { runBClient :: forall a. BRequest a -> f a }

mkGoogleBusinessClient
  :: forall m. (MonadIO m)
  => TokenProvider m -> G.Client m -> BClient m
mkGoogleBusinessClient tokenProvider gClient = BClient {runBClient}
  where
  runBClient :: BRequest a -> m a
  runBClient req = do
    cmd <- runApComposed genUUID req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let userIds = cmds <&> \(This (BCommand _ a)) -> getUserIdFromBAction a
    tokens <- runTokenProvider tokenProvider userIds
    let inputs = mkInputs tokens cmds

Type aliases for sanity


-- | Alias for a Google Request; represents a single "step" in the interpreter.
type GStep = G.Request

-- | Represents a monadic "program" containing several GSteps.
-- The interpreter will batch intermediate GSteps of multiple GLogic programs.
type GLogic = FreeMonad.Free GStep

-- | GLogic lifted to deal with failures.
type GAttempt = ExceptT BFailure GLogic

-- | Interpreter result of a completed GLogic program.
data IResult a = IResult { action :: BAction a, value :: a }

-- | Inputs into the interpreter
type Inputs = Map UUID (GLogic (Some IResult))

Initializing the inputs


type Inputs = Map UUID (GLogic (Some IResult))

mkInputs :: Map UserId G.AccessToken -> [Some BCommand] -> Inputs
mkInputs tokens cmds =
  Map.fromList $
    cmds <&> \(This (BCommand cmdId a)) ->
      (cmdId, This <$> mkGLogic tokens a)

Initializing the inputs


mkGLogic :: Map UserId G.AccessToken -> BAction a -> GLogic (IResult a)
mkGLogic tokens = \case
  action@(BInsertEvent uid event) -> go action insertEvent uid event
  action@(BDeleteEvent uid eid)   -> go action deleteEvent uid eid
  action@(BEventExists uid eid)   -> go action eventExists uid eid
  action@(BUpdateEvent uid event) -> go action updateEvent uid event
  where
  go :: BActionResult b
     -> (G.AccessToken -> c -> GAttempt b)
     -> UserId
     -> c
     -> GLogic (IResult (Either BFailure b))
  go action f uid c = fmap (IResult action) . runExceptT $ f (tok ! uid) c

Inserting an event with update fallback


insertEvent :: G.AccessToken -> BEvent -> GAttempt BInsertResult
insertEvent tok bEvent = do
  let gEvent = fromBEvent bEvent
  gAttempt (G.EventsInsertWithKey tok calId gEvent) >>= \case
    Right (G.Keyed k _) -> pure $ Inserted k
    Left e0 -> do
      when ((status e0) /= 409) $ throwError $ toBFailure e0
      gAttempt (G.EventsUpdate tok calId gEvent) >>= \case
        Right (G.Keyed k _) -> pure $ UpdatedViaInsert k
        Left e1 -> throwError $ toBFailure e1

Making a Business Client


mkGoogleBusinessClient
  :: forall m. (MonadIO m)
  => TokenProvider m -> G.Client m -> BClient m
mkGoogleBusinessClient tokenProvider gClient = BClient {runBClient}
  where
  runBClient :: BRequest a -> m a
  runBClient req = do
    cmd <- runApComposed mkCmd req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let userIds = cmds <&> \(This (BCommand _ a)) -> getUserIdFromBAction a
    tokens <- runTokenProvider tokenProvider userIds
    let inputs = mkInputs tokens cmds
    results <- runInterpreter inputs

The Interpreter


runInterpreter :: Inputs -> m (Map UUID (Some IResult))
runInterpreter inputs =
  case allComplete inputs of
    Just results -> pure results


-- Returns Nothing if there are any Incomplete values; otherwise returns
-- the final result map.
allComplete :: Inputs -> Maybe (Map UUID (Some IResult))
allComplete = traverse $ \case
  FreeMonad.Pure v -> Just v
  FreeMonad.Free _ -> Nothing


runInterpreter :: Inputs -> m (Map UUID (Some IResult))
runInterpreter inputs =
  case allComplete inputs of
    Just results -> pure results
    Nothing -> do
      -- Collect the next round of steps to be executed.
      let steps = collectSteps inputs

collectSteps :: Inputs -> Map UUID (GStep (GLogic (Some IResult)))
collectSteps = Map.mapMaybe $ \case
  FreeMonad.Free x -> Just x
  FreeMonad.Pure _ -> Nothing

The interpreter loop 🎉


runInterpreter :: Inputs -> m (Map UUID (Some IResult))
runInterpreter inputs =
  case allComplete inputs of
    Just results -> pure results
    Nothing -> do
      let steps :: Map UUID (GStep (GLogic (Some IResult))) = collectSteps inputs
      let step  :: GStep (Map UUID (GLogic (Some IResult))) = sequenceA steps
             -- :: GStep Inputs
      response :: Inputs <- G.runClient gClient step
      runInterpreter $ response <> inputs

Making a Business Client


mkGoogleBusinessClient
  :: forall m. (MonadIO m)
  => TokenProvider m -> G.Client m -> BClient m
mkGoogleBusinessClient tokenProvider gClient = BClient {runBClient}
  where
  runBClient :: BRequest a -> m a
  runBClient req = do
    cmd <- runApComposed mkCmd req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let userIds = cmds <&> \(This (BCommand _ a)) -> getUserIdFromBAction a
    tokens <- runTokenProvider tokenProvider userIds
    let inputs = mkInputs tokens cmds
    results <- runInterpreter inputs
    FreeAp.runAp (extractor results) cmd

The extractor


extractor :: Map UUID (Some IResult) -> BCommand a -> m a
extractor results (BCommand k action) = case action of
  BInsertEvent {} -> case results ! k of
    This (IResult (BInsertEvent {}) x) -> pure x
    This (IResult other _) -> throwTypeMismatch action other

  BDeleteEvent {} -> case res of
    This (IResult (BDeleteEvent {}) x) -> pure x
    This (IResult other _) -> throwTypeMismatch action other

  -- and so on, the compiler will check for exhaustiveness

Making a Business Client


mkGoogleBusinessClient
  :: forall m. (MonadIO m)
  => TokenProvider m -> G.Client m -> BClient m
mkGoogleBusinessClient tokenProvider gClient = BClient {runBClient}
  where
  runBClient :: BRequest a -> m a
  runBClient req = do
    cmd <- runApComposed mkCmd req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let userIds = cmds <&> \(This (BCommand _ a)) -> getUserIdFromBAction a
    tokens <- runTokenProvider tokenProvider userIds
    let inputs = mkInputs tokens cmds
    results <- runInterpreter inputs
    FreeAp.runAp (extractor results) cmd

Making a Business Client


mkGoogleBusinessClient
  :: forall m. (MonadIO m)
  => TokenProvider m -> G.Client m -> BClient m
mkGoogleBusinessClient tokenProvider gClient = BClient {runBClient}
  where
  runBClient :: BRequest a -> m a
  runBClient req = do
    cmd <- runApComposed mkCmd req
    let cmds = FreeAp.runAp_ (\c -> [This c]) cmd
    let userIds = cmds <&> \(This (BCommand _ a)) -> getUserIdFromBAction a
    tokens <- runTokenProvider tokenProvider userIds
    let inputs = mkInputs tokens cmds
    results <- runInterpreter inputs
    FreeAp.runAp (extractor results) cmd

Recap

Represented our request albegras using GADTs

Used Free Applicative to batch groups of requests

Used Free Monad of Free Applicative to build batchable programs

Inspect the next steps of our programs with an interpreter,

batching steps together as much as possible

Continue this process until all programs have completed

Don't forget to write tests!

Testing


it "should work with small inputs" $ do
  (gClient, counter) <- newGClientWithCounter
  let bClient = mkGoogleBusinessClient defaultTokenProvider gClient
  [e1, e2, e3] <- replicateM 3 randomBEvent
  res <- runBClient bClient $
    (,,)
      <$> mkBRequest (BInsertEvent userId e1)
      <*> mkBRequest (BUpdateEvent userId e2)
      <*> mkBRequest (BUpdateEvent userId e3)
  readIORef counter `shouldReturn` 2
  res `shouldBe`
    ( Right $ Inserted          $ getExtId e1
    , Right $ InsertedViaUpdate $ getExtId e2
    , Right $ InsertedViaUpdate $ getExtId e3
    )

Testing


it "should work with large, dynamic inputs" $ do
  (gClient, counter) <- newGClientWithCounter
  let bClient = mkGoogleBusinessClient defaultTokenProvider gClient
  actions :: [Some BAction] <- replicateM 100 randomBAction
  let req :: FreeAp.Ap BAction Dynamic = sequenceA $ dynBRequest <$> actions
  res :: [Dynamic] <- runBClient bClient req
  length res `shouldBe` 100
  readIORef counter >>= `shouldSatisfy` (<= 4)
  traverse_ validateDynResponse $ zip actions res

Testing


validateDynResponse :: (Some BAction, Dynamic) -> IO ()
validateDynResponse (sa, d) = withSome sa $ \case
  x@(BInsertEvent {}) -> expectJust $ proxyDynResponse x d
  x@(BEventExists {}) -> expectJust $ proxyDynResponse x d
  x@(BDeleteEvent {}) -> expectJust $ proxyDynResponse x d
  x@(BUpdateEvent {}) -> expectJust $ proxyDynResponse x d

proxyDynResponse :: (Typeable a) => BAction a -> Dynamic -> Maybe a
proxyDynResponse _ = fromDynamic

proxyDynResponse :: (Typeable a) => proxy a -> Dynamic -> Maybe a
proxyDynResponse _ = fromDynamic

It works!


Batmon.Google.Calendar.BClientIT
  BClient
    should batch efficiently
    should work with large inputs
Batmon.Google.Calendar.GClientIT
  GClient
    should batch multiple requests
    EventsInsertWithKey should work
    EventsGet should return Nothing if not exists

Finished in 8.7644 seconds
5 examples, 0 failures

Thank you!

http://caryrobbins.com/batmon-talk

Files

slides.gen.md

Latest commit

History

slides.gen.md

File metadata and controls

Batching Monadic Requests

with Free Interpreters

Prerequisite Concepts

Does this only work in Haskell?

Using just Haskell2010!

Free Library

Problem

Solution

A quick derail into monads

Applicative to the rescue?

Business Inc. Command Requirements

Business Inc. Scope Creep

Insert Event Prototype

Taking a step back

Google Calendar Algebra

Google Calendar Algebra

Free Applicative Refresher

Free Applicative Request

Google Calendar Client

Now just implement it!

Making a batched client

Making a batched client

HttpHandler

HttpHandler

The crucial command

The crucial command

The client for real

Backtrack to runAp

The Client for real

Backtrack to runAp_

The client for real

Batched reader

How does decodeResponse work?

The client for real

runCommands

The Client for real

Yay, we can batch!

Free monads aren't the solution

What if...

Business Inc Algebra

Business Inc Client

Business Inc Client

Type aliases for sanity

Initializing the inputs

Initializing the inputs

Inserting an event with update fallback

Making a Business Client

The Interpreter

The interpreter loop 🎉

Making a Business Client

The extractor

Making a Business Client

Making a Business Client

Recap

Testing

Testing

Testing

It works!

Thank you!