Repository: hasura/graphql-engine
File: server/src-lib/Hasura/Eventing/EventTrigger.hs (Haskell)

Code before:
    popEventsBatch = do
      let run = runExceptT . Q.runTx pool (Q.RepeatableRead, Just Q.ReadWrite)
      run (fetchEvents fetchBatchSize) >>= \case
          Left err -> do
            L.unLogger logger $ EventInternalErr err
            return []
          Right events -> do
            saveLockedEvents events
            return events

    -- After the events are fetched from the DB, we store the locked events
    -- in a hash set(order doesn't matter and look ups are faster) in the
    -- event engine context
    saveLockedEvents :: [Event] -> IO ()
    saveLockedEvents evts =
      liftIO $ atomically $ do
        lockedEvents <- readTVar _eeCtxLockedEvents
        let evtsIds = map eId evts
        let newLockedEvents = Set.union lockedEvents (Set.fromList evtsIds)
        writeTVar _eeCtxLockedEvents newLockedEvents

    removeEventFromLockedEvents :: EventId -> IO ()
    removeEventFromLockedEvents eventId = do
      liftIO $ atomically $ do
        lockedEvents <- readTVar _eeCtxLockedEvents
        writeTVar _eeCtxLockedEvents $ Set.delete eventId lockedEvents

    -- work on this batch of events while prefetching the next. Recurse after we've forked workers
    -- for each in the batch, minding the requested pool size.
    go :: [Event] -> Int -> Bool -> IO void
    go events !fullFetchCount !alreadyWarned = do
      -- process events ASAP until we've caught up; only then can we sleep
      when (null events) $ sleep _eeCtxFetchInterval

      -- Prefetch next events payload while concurrently working through our current batch.
      -- NOTE: we probably don't need to prefetch so early, but probably not
      -- worth the effort for something more fine-tuned
      eventsNext <- withAsync popEventsBatch $ \eventsNextA -> do
        -- process approximately in order, minding HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE:
        forM_ events $ \event -> do
             runReaderT (withEventEngineCtx eeCtx $ (processEvent event)) (logger, httpMgr)
             -- removing an event from the _eeCtxLockedEvents after the event has
             -- been processed
             removeEventFromLockedEvents (eId event)
        wait eventsNextA

      let lenEvents = length events
      if | lenEvents == fetchBatchSize -> do
             -- If we've seen N fetches in a row from the DB come back full (i.e. only limited
             -- by our LIMIT clause), then we say we're clearly falling behind:
             let clearlyBehind = fullFetchCount >= 3

Review comment (suggestion):
It's probably fine, but can you add a strictness annotation here?:

```
writeTVar _eeCtxLockedEvents $! Set.delete eventId lockedEvents
                              ^
```
I'm always a little worried about writing thunks to mutable variables

Code after:
    popEventsBatch = do
      let run = runExceptT . Q.runTx pool (Q.RepeatableRead, Just Q.ReadWrite)
      run (fetchEvents fetchBatchSize) >>= \case
          Left err -> do
            L.unLogger logger $ EventInternalErr err
            return []
          Right events -> do
            saveLockedEvents events
            return events

    -- After the events are fetched from the DB, we store the locked events
    -- in a hash set(order doesn't matter and look ups are faster) in the
    -- event engine context
    saveLockedEvents :: [Event] -> IO ()
    saveLockedEvents evts =
      liftIO $ atomically $ do
        lockedEvents <- readTVar _eeCtxLockedEvents
        let evtsIds = map eId evts
        let newLockedEvents = Set.union lockedEvents (Set.fromList evtsIds)
        writeTVar _eeCtxLockedEvents $! newLockedEvents

    removeEventFromLockedEvents :: EventId -> IO ()
    removeEventFromLockedEvents eventId = do
      liftIO $ atomically $ do
        lockedEvents <- readTVar _eeCtxLockedEvents
        writeTVar _eeCtxLockedEvents $! Set.delete eventId lockedEvents

    -- work on this batch of events while prefetching the next. Recurse after we've forked workers
    -- for each in the batch, minding the requested pool size.
    go :: [Event] -> Int -> Bool -> IO void
    go events !fullFetchCount !alreadyWarned = do
      -- process events ASAP until we've caught up; only then can we sleep
      when (null events) $ sleep _eeCtxFetchInterval

      -- Prefetch next events payload while concurrently working through our current batch.
      -- NOTE: we probably don't need to prefetch so early, but probably not
      -- worth the effort for something more fine-tuned
      eventsNext <- withAsync popEventsBatch $ \eventsNextA -> do
        -- process approximately in order, minding HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE:
        forM_ events $ \event -> do
             runReaderT (withEventEngineCtx eeCtx $ (processEvent event)) (logger, httpMgr)
             -- removing an event from the _eeCtxLockedEvents after the event has
             -- been processed
             removeEventFromLockedEvents (eId event)
        wait eventsNextA

      let lenEvents = length events
      if | lenEvents == fetchBatchSize -> do
             -- If we've seen N fetches in a row from the DB come back full (i.e. only limited
             -- by our LIMIT clause), then we say we're clearly falling behind:
             let clearlyBehind = fullFetchCount >= 3