stop blocking offchain queues & update query

Author: Cmdv

cardano-db-sync/src/Cardano/DbSync/OffChain.hs

@@ -29,7 +29,6 @@ import Control.Concurrent.Class.MonadSTM.Strict (
   StrictTBQueue (..),
   flushTBQueue,
   isEmptyTBQueue,
-  readTBQueue,
   writeTBQueue,
  )
 import Control.Monad.Trans.Control (MonadBaseControl)
@@ -84,8 +83,8 @@ loadOffChainWorkQueue ::
 loadOffChainWorkQueue _trce offChainWorkQueue = do
   whenM (liftIO $ atomically (isEmptyTBQueue (lQueue offChainWorkQueue))) $ do
     now <- liftIO Time.getPOSIXTime
-    runnableOffChainPools <- filter (isRunnable now offChainWorkQueue) <$> lGetData offChainWorkQueue now 100
-    liftIO $ mapM_ (queueInsert offChainWorkQueue) runnableOffChainPools
+    runnableOffChainData <- filter (isRunnable now offChainWorkQueue) <$> lGetData offChainWorkQueue now 100
+    liftIO $ mapM_ (queueInsert offChainWorkQueue) runnableOffChainData
   where
     isRunnable :: POSIXTime -> LoadOffChainWorkQueue a m -> a -> Bool
     isRunnable now oCWorkQueue locWq = retryRetryTime (lRetryTime oCWorkQueue locWq) <= now
@@ -172,7 +171,7 @@ runFetchOffChainPoolThread syncEnv = do
       tDelay
       -- load the offChain vote work queue using the db
       _ <- runReaderT (loadOffChainPoolWorkQueue trce (envOffChainPoolWorkQueue syncEnv)) (envBackend syncEnv)
-      poolq <- blockingFlushTBQueue (envOffChainPoolWorkQueue syncEnv)
+      poolq <- atomically $ flushTBQueue (envOffChainPoolWorkQueue syncEnv)
       manager <- Http.newManager tlsManagerSettings
       now <- liftIO Time.getPOSIXTime
       mapM_ (queuePoolInsert <=< fetchOffChainPoolData trce manager now) poolq
@@ -192,7 +191,7 @@ runFetchOffChainVoteThread syncEnv = do
       tDelay
       -- load the offChain pool work queue using the db
       _ <- runReaderT (loadOffChainVoteWorkQueue trce (envOffChainVoteWorkQueue syncEnv)) (envBackend syncEnv)
-      voteq <- blockingFlushTBQueue (envOffChainVoteWorkQueue syncEnv)
+      voteq <- atomically $ flushTBQueue (envOffChainVoteWorkQueue syncEnv)
       manager <- Http.newManager tlsManagerSettings
       now <- liftIO Time.getPOSIXTime
       mapM_ (queueVoteInsert <=< fetchOffChainVoteData trce manager now) voteq
@@ -207,16 +206,6 @@ runFetchOffChainVoteThread syncEnv = do
 tDelay :: IO ()
 tDelay = threadDelay 60_000_000
 
--- -------------------------------------------------------------------------------------------------
-
--- Blocks on an empty queue, but gets all elements in the queue if there is more than one.
-blockingFlushTBQueue :: StrictTBQueue IO a -> IO [a]
-blockingFlushTBQueue queue = do
-  atomically $ do
-    x <- readTBQueue queue
-    xs <- flushTBQueue queue
-    pure $ x : xs
-
 ---------------------------------------------------------------------------------------------------------------------------------
 -- Fetch OffChain data
 ---------------------------------------------------------------------------------------------------------------------------------

cardano-db-sync/src/Cardano/DbSync/OffChain/Http.hs

@@ -172,7 +172,7 @@ httpGetOffChain manager request mHash url =
                 { sovaHash = metadataHash
                 , sovaBytes = respBS
                 , -- TODO: no json format decided for vote metadata yet so we are leaving it blank for now
-                  sovaJson = ""
+                  sovaJson = "{}"
                 , sovaContentType = mContentType
                 }
 

cardano-db-sync/src/Cardano/DbSync/OffChain/Query.hs

@@ -75,8 +75,16 @@ queryNewVoteWorkQueue :: MonadIO m => POSIXTime -> ReaderT SqlBackend m [OffChai
 queryNewVoteWorkQueue now = do
   res <- select $ do
     va <- from $ table @VotingAnchor
-    where_ (notExists $ from (table @OffChainVoteData) >>= \ocvd -> where_ (ocvd ^. OffChainVoteDataVotingAnchorId ==. va ^. VotingAnchorId))
-    where_ (notExists $ from (table @OffChainVoteFetchError) >>= \ocvfe -> where_ (ocvfe ^. OffChainVoteFetchErrorVotingAnchorId ==. va ^. VotingAnchorId))
+    where_
+      ( notExists $
+          from (table @OffChainVoteData) >>= \ocvd ->
+            where_ (ocvd ^. OffChainVoteDataVotingAnchorId ==. va ^. VotingAnchorId)
+      )
+    where_
+      ( notExists $
+          from (table @OffChainVoteFetchError) >>= \ocvfe ->
+            where_ (ocvfe ^. OffChainVoteFetchErrorVotingAnchorId ==. va ^. VotingAnchorId)
+      )
     pure
       ( va ^. VotingAnchorId
       , va ^. VotingAnchorDataHash
@@ -101,6 +109,11 @@ queryOffChainVoteWorkQueue _now = do
         $ table @VotingAnchor
           `innerJoin` table @OffChainVoteFetchError
         `on` (\(va :& ocpfe) -> ocpfe ^. OffChainVoteFetchErrorVotingAnchorId ==. va ^. VotingAnchorId)
+    where_
+      ( notExists $
+          from (table @OffChainVoteData) >>= \ocvd ->
+            where_ (ocvd ^. OffChainVoteDataVotingAnchorId ==. ocpfe ^. OffChainVoteFetchErrorVotingAnchorId)
+      )
     orderBy [desc (ocpfe ^. OffChainVoteFetchErrorFetchTime)]
     pure
       ( ocpfe ^. OffChainVoteFetchErrorFetchTime
@@ -144,8 +157,16 @@ queryNewPoolWorkQueue now = do
           `innerJoin` table @PoolMetadataRef
         `on` (\(ph :& pmr) -> ph ^. PoolHashId ==. pmr ^. PoolMetadataRefPoolId)
     where_ (just (pmr ^. PoolMetadataRefId) `in_` latestRefs)
-    where_ (notExists $ from (table @OffChainPoolData) >>= \pod -> where_ (pod ^. OffChainPoolDataPmrId ==. pmr ^. PoolMetadataRefId))
-    where_ (notExists $ from (table @OffChainPoolFetchError) >>= \pofe -> where_ (pofe ^. OffChainPoolFetchErrorPmrId ==. pmr ^. PoolMetadataRefId))
+    where_
+      ( notExists $
+          from (table @OffChainPoolData) >>= \pod ->
+            where_ (pod ^. OffChainPoolDataPmrId ==. pmr ^. PoolMetadataRefId)
+      )
+    where_
+      ( notExists $
+          from (table @OffChainPoolFetchError) >>= \pofe ->
+            where_ (pofe ^. OffChainPoolFetchErrorPmrId ==. pmr ^. PoolMetadataRefId)
+      )
     pure
       ( ph ^. PoolHashId
       , pmr ^. PoolMetadataRefId