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

Code before:
--   - don't fetch more than N at a time (since that can mean: space leak, less
--     effective scale out, possible double sends for events we've checked out 
--     on exit (TODO clean shutdown procedure))
--   - try not to cause webhook workers to stall waiting on DB fetch
--   - limit webhook HTTP concurrency per HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE
processEventQueue
  :: (HasVersion) => L.Logger L.Hasura -> LogEnvHeaders -> HTTP.Manager-> Q.PGPool
  -> IO SchemaCache -> EventEngineCtx 
  -> IO void
processEventQueue logger logenv httpMgr pool getSchemaCache EventEngineCtx{..} = do
  events0 <- popEventsBatch
  go events0 0 False
  where
    fetchBatchSize = 100
    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 -> 
            return events

    go :: [Event] -> Int -> Bool -> IO void
    go events !fullFetchCnt !alreadyWarned = do
      -- process events ASAP until we've caught up; then we can 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 -> 
          mask_ $ do
            atomically $ do  -- block until < HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE threads:
              capacity <- readTVar _eeCtxEventThreadsCapacity
              check $ capacity > 0
              writeTVar _eeCtxEventThreadsCapacity (capacity - 1) 
            let restoreCapacity = liftIO $ atomically $ 
                  modifyTVar' _eeCtxEventThreadsCapacity (+ 1)
            t <- async $ flip runReaderT (logger, httpMgr) $ 
                    processEvent event `finally` restoreCapacity
            link t

        -- return when next batch ready; some 'processEvent' threads may be running.
        wait eventsNextA

      let lenEvents = length events
      if | lenEvents == fetchBatchSize -> do
             -- When we choose to warn is fairly arbitrary:

Review comment (suggestion):
@jberryman I am slighly confused how are we capping the memory/threads in flight. IIUC, we keep fetching events and get blocked on processing them because of `HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE`

Code after:
--   - don't fetch more than N at a time (since that can mean: space leak, less
--     effective scale out, possible double sends for events we've checked out 
--     on exit (TODO clean shutdown procedure))
--   - try not to cause webhook workers to stall waiting on DB fetch
--   - limit webhook HTTP concurrency per HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE
processEventQueue
  :: (HasVersion) => L.Logger L.Hasura -> LogEnvHeaders -> HTTP.Manager-> Q.PGPool
  -> IO SchemaCache -> EventEngineCtx 
  -> IO void
processEventQueue logger logenv httpMgr pool getSchemaCache EventEngineCtx{..} = do
  events0 <- popEventsBatch
  go events0 0 False
  where
    fetchBatchSize = 100
    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 -> 
            return events

    -- 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 -> 
          mask_ $ do
            atomically $ do  -- block until < HASURA_GRAPHQL_EVENTS_HTTP_POOL_SIZE threads:
              capacity <- readTVar _eeCtxEventThreadsCapacity
              check $ capacity > 0
              writeTVar _eeCtxEventThreadsCapacity (capacity - 1) 
            -- since there is some capacity in our worker threads, we can launch another:
            let restoreCapacity = liftIO $ atomically $ 
                  modifyTVar' _eeCtxEventThreadsCapacity (+ 1)
            t <- async $ flip runReaderT (logger, httpMgr) $ 
                    processEvent event `finally` restoreCapacity
            link t

        -- return when next batch ready; some 'processEvent' threads may be running.
        wait eventsNextA