Replace mutating status with query

Author: axman6

lib/PSR/Events/Interface.hs

@@ -14,6 +14,7 @@ import Data.Maybe (isNothing)
 import Data.Text (Text)
 import Data.Time.Clock (UTCTime)
 import GHC.Generics (Generic)
+import PSR.Types (BlockStatus)
 import PlutusLedgerApi.Common (Data, MajorProtocolVersion, PlutusLedgerLanguage)
 
 data EventType
@@ -34,6 +35,7 @@ data Event = Event
     , slotNo :: C.SlotNo
     , createdAt :: UTCTime
     , payload :: EventPayload
+    , blockStatus :: BlockStatus
     }
     deriving (Generic)
 

lib/PSR/Storage/SQLite.hs

@@ -19,7 +19,6 @@ import PSR.Storage.SQLite.GetEvents qualified as GetEvents
 import PSR.Storage.SQLite.Instances ()
 import PSR.Storage.SQLite.Metrics (SqliteMetrics (..), initialiseMetrics)
 import PSR.Storage.SQLite.Utils
-import PSR.Types (BlockStatus (..))
 import PlutusLedgerApi.Common (MajorProtocolVersion (..))
 
 withSqliteStorage :: FilePath -> Int -> (Storage -> IO ()) -> IO ()
@@ -34,7 +33,7 @@ mkStorage confirmationDepth metrics pool = do
     withResource pool initSchema
     pure $ Storage{..}
   where
-    getEvents = GetEvents.getEvents metrics.getEvents_select pool
+    getEvents = GetEvents.getEvents metrics.getEvents_select pool confirmationDepth
 
     -- NOTE: The block may not always exist in our database. And it may not be
     -- possible to get the proper BlockHeader on a rollback event.
@@ -44,7 +43,6 @@ mkStorage confirmationDepth metrics pool = do
         let colsKnown =
                 [ col "slot_no" slotNo
                 , col "hash" hash
-                , col "status" BSUnknown
                 ]
             cols =
                 case mBlockNo of
@@ -61,24 +59,6 @@ mkStorage confirmationDepth metrics pool = do
     createBlockIfNotExists conn (BlockHeader slotNo hash blockNo) =
         createBlockIfNotExistsUtil conn slotNo hash (Just blockNo)
 
-    commitBlock :: Connection -> BlockNo -> IO ()
-    commitBlock conn blockNoToCommit = do
-        let q = "UPDATE block SET status = :set_status WHERE block_no = :block_no AND status = :prev_status;"
-        executeNamed metrics.setBlockStatus_update conn q $
-            [ ":set_status" := BSCommitted
-            , ":prev_status" := BSUnknown
-            , ":block_no" := blockNoToCommit
-            ]
-
-    cancelBlocksAfterSlot :: Connection -> SlotNo -> IO [Hash BlockHeader]
-    cancelBlocksAfterSlot conn slotNo = do
-        let q = "UPDATE block SET status = :set_status WHERE slot_no > :slot_no RETURNING hash;"
-        fmap (fmap fromOnly) $
-            queryNamed metrics.setBlockStatus_update conn q $
-                [ ":set_status" := BSCancelled
-                , ":slot_no" := slotNo
-                ]
-
     getOrCreateCostModelParamsId :: Connection -> MajorProtocolVersion -> CostModel -> IO Integer
     getOrCreateCostModelParamsId conn (MajorProtocolVersion v) costModel = do
         version <- mkVersion64 $ fromIntegral v
@@ -157,20 +137,28 @@ mkStorage confirmationDepth metrics pool = do
     addRollbackEvent slotNo hash =
         withResource pool $ \conn -> withTransaction conn $ do
             createPartialBlockIfNotExists conn slotNo hash
-            blocksCancelled <- cancelBlocksAfterSlot conn slotNo
-            let params =
-                    [ col "block_hash" hash
-                    , col "blocks_cancelled" blocksCancelled
+            [Only event_id] <-
+                sqlInsertReturning
+                    metrics.addRollbackEvent_insert
+                    conn
+                    "rollback_event"
+                    [col "block_hash" hash]
+                    ["event_id"]
+            let q' =
+                    "INSERT INTO rollback_block (event_id, block_hash) \
+                    \SELECT :event_id, hash FROM block WHERE slot_no > :slot_no \
+                    \RETURNING block_hash;"
+            fmap fromOnly
+                <$> queryNamed
+                    metrics.addRollbackBlock_insert
+                    conn
+                    q'
+                    [ ":event_id" := (event_id :: Integer)
+                    , ":slot_no" := slotNo
                     ]
-            sqlInsert
-                metrics.addRollbackEvent_insert
-                conn
-                "rollback_event"
-                params
-            pure blocksCancelled
 
     addSelectionEvent :: BlockHeader -> IO ()
-    addSelectionEvent blockHeader@(BlockHeader _ hash blockNo) =
+    addSelectionEvent blockHeader@(BlockHeader _ hash _) =
         withResource pool $ \conn -> withTransaction conn $ do
             void $ createBlockIfNotExists conn blockHeader
             let params = [col "block_hash" hash]
@@ -179,7 +167,6 @@ mkStorage confirmationDepth metrics pool = do
                 conn
                 "selection_event"
                 params
-            commitBlock conn (blockNo - fromIntegral confirmationDepth)
 
     getExecutionContexts :: [FilterBy] -> IO [(BlockHeader, ExecutionContextId, ExecutionContext)]
     getExecutionContexts filters =

lib/PSR/Storage/SQLite/GetEvents.hs

@@ -26,9 +26,10 @@ import PSR.Events.Interface (
 import PSR.Metrics qualified as Metrics
 import PSR.Storage.SQLite.Instances ()
 import PSR.Storage.SQLite.Utils
+import PSR.Types (BlockStatus)
 
-getEvents :: Metrics.Summary -> Pool Connection -> EventFilterParams -> IO [Event]
-getEvents getEvents_select pool EventFilterParams{..} =
+getEvents :: Metrics.Summary -> Pool Connection -> Int -> EventFilterParams -> IO [Event]
+getEvents getEvents_select pool confirmationDepth EventFilterParams{..} =
     withResource pool $ \conn -> withTransaction conn $ do
         let
             -- see `docs/specification.md` for default values
@@ -80,15 +81,20 @@ getEvents getEvents_select pool EventFilterParams{..} =
                 "SELECT b.slot_no, b.hash, b.block_no, \
                 \ CASE \
                 \   WHEN ec.block_hash IS NOT NULL THEN 'execution' \
-                \   WHEN c.block_hash IS NOT NULL THEN 'rollback' \
+                \   WHEN re.block_hash IS NOT NULL THEN 'rollback' \
                 \   WHEN s.block_hash IS NOT NULL THEN 'selection' \
                 \ END, \
-                \ COALESCE(ee.created_at, c.created_at, s.created_at), \
+                \ COALESCE(ee.created_at, re.created_at, s.created_at), \
                 \ json(ee.trace_logs), \
-                \ c.blocks_cancelled, \
+                \ rb.block_hashes, \
                 \ ee.eval_error, \
                 \ ee.exec_budget_cpu, \
                 \ ee.exec_budget_mem, \
+                \ CASE \
+                \   WHEN rb.block_hashes IS NOT NULL THEN 'cancelled' \
+                \   WHEN b_max.max_slot_no  IS NOT NULL AND (b_max.max_slot_no - :confirmation_depth) > b.slot_no THEN 'committed' \
+                \   ELSE 'unknown'\
+                \ END,\
                 \ ec.transaction_hash, \
                 \ ec.target_script_hash, \
                 \ ec.target_script_name, \
@@ -103,20 +109,28 @@ getEvents getEvents_select pool EventFilterParams{..} =
                 \ ec.exec_budget_max_mem, \
                 \ cmp.params \
                 \ FROM block b \
-                \ LEFT JOIN execution_context ec ON ec.block_hash = b.hash \
-                \ LEFT JOIN execution_event ee ON ee.context_id = ec.context_id \
+                \ LEFT JOIN execution_context ec  ON ec.block_hash = b.hash \
+                \ LEFT JOIN execution_event   ee  ON ee.context_id = ec.context_id \
                 \ LEFT JOIN cost_model_params cmp ON cmp.params_id = ec.cost_model_params_id \
-                \ LEFT JOIN rollback_event c ON c.block_hash = b.hash \
-                \ LEFT JOIN selection_event s ON s.block_hash = b.hash "
+                \ LEFT JOIN rollback_event    re  ON re.block_hash = b.hash \
+                \ LEFT JOIN \
+                \(SELECT event_id, string_agg(block_hash, ' ') AS block_hashes FROM rollback_block) \
+                \                             rb  ON re.event_id   = rb.event_id \
+                \ LEFT JOIN selection_event   s   ON s.block_hash  = b.hash  \
+                \ JOIN (SELECT max(slot_no) as max_slot_no from block) b_max"
                     <> whereQuery
-                    <> " ORDER BY COALESCE(ee.created_at, c.created_at, s.created_at) ASC \
+                    <> " ORDER BY COALESCE(ee.created_at, re.created_at, s.created_at) ASC \
                        \ LIMIT :limit \
                        \ OFFSET :offset"
 
-            parameters = whereParams <> [":limit" := limitParameter, ":offset" := offsetParameter]
-
-        rows <- queryNamed getEvents_select conn eventsQuery parameters
-        pure $ rowToEvent <$> rows
+            parameters =
+                whereParams
+                    <> [ ":limit" := limitParameter
+                       , ":offset" := offsetParameter
+                       , ":confirmation_depth" := confirmationDepth
+                       ]
+        fmap rowToEvent
+            <$> queryNamed getEvents_select conn eventsQuery parameters
   where
     rowToEvent ::
         ( ( SlotNo
@@ -129,6 +143,7 @@ getEvents getEvents_select pool EventFilterParams{..} =
           , Maybe EvalError
           , Maybe Integer
           , Maybe Integer
+          , BlockStatus
           )
             :. Maybe ExecutionContext
         ) ->
@@ -144,6 +159,7 @@ getEvents getEvents_select pool EventFilterParams{..} =
                 , evalError
                 , mExBudgetCpu
                 , mExBudgetMem
+                , blockStatus
                 )
                 :. mExecutionContext
             ) =
@@ -161,7 +177,7 @@ getEvents getEvents_select pool EventFilterParams{..} =
                         context <- mExecutionContext
                         blockNo <- mBlockNo
                         pure $ ExecutionPayload blockNo $ ExecutionEventPayload{..}
-                    Rollback -> pure $ RollbackPayload (maybe [] id blocksCancelled)
+                    Rollback -> pure $ RollbackPayload (fromMaybe [] blocksCancelled)
                     Selection -> SelectionPayload <$> mBlockNo
              in
                 Event{..}

lib/PSR/Storage/SQLite/Instances.hs

@@ -224,3 +224,21 @@ maybeField =
         fromField >=> \case
             Just v -> pure v
             _ -> pure Nothing
+
+instance
+    ( FromField a
+    , FromField b
+    , FromField c
+    , FromField d
+    , FromField e
+    , FromField f
+    , FromField g
+    , FromField h
+    , FromField i
+    , FromField j
+    , FromField k
+    ) =>
+    FromRow (a, b, c, d, e, f, g, h, i, j, k)
+    where
+    fromRow =
+        (,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field <*> field <*> field <*> field <*> field <*> field <*> field

lib/PSR/Storage/SQLite/Metrics.hs

@@ -5,13 +5,13 @@ import PSR.Storage.SQLite.Instances ()
 
 data SqliteMetrics = SqliteMetrics
     { createBlockIfNotExists_insert :: Summary
-    , setBlockStatus_update :: Summary
     , getOrCreateCostModelParamsId_insert :: Summary
     , getOrCreateCostModelParamsId_select :: Summary
     , setOrCreateBlockId_insert :: Summary
     , setOrCreateBlockId_select :: Summary
     , addExecutionEvent_insert :: Summary
     , addRollbackEvent_insert :: Summary
+    , addRollbackBlock_insert :: Summary
     , addSelectionEvent_insert :: Summary
     , getEvents_select :: Summary
     , getExecutionContextByNameOrScriptHash_select :: Summary
@@ -23,10 +23,6 @@ initialiseMetrics = do
         regSummary
             "sqlite_createBlockIfNotExists_insert"
             "Execution time of createBlockIfNotExists insert query"
-    setBlockStatus_update <-
-        regSummary
-            "sqlite_setBlockStatus_update"
-            "Execution time of setBlockStatus update query"
     getOrCreateCostModelParamsId_select <-
         regSummary
             "sqlite_getOrCreateCostModelParamsId_select"
@@ -51,6 +47,10 @@ initialiseMetrics = do
         regSummary
             "sqlite_addRollbackEvent_insert"
             "Execution time of addRollbackEvent insert query"
+    addRollbackBlock_insert <-
+        regSummary
+            "sqlite_addRollbackBlock_insert"
+            "Execution time of addRollbackEvent block insert query"
     addSelectionEvent_insert <-
         regSummary
             "sqlite_addSelectionEvent_insert"

lib/PSR/Storage/SQLite/Utils.hs

@@ -64,7 +64,7 @@ initSchema conn = withTransaction conn $ do
 
 -- "col" is a convenience sugar over (:=)
 col :: (ToField v) => Text -> v -> NamedParam
-col k v = (T.cons ':' k) := v
+col k v = T.cons ':' k := v
 
 getParamKeys :: [NamedParam] -> [Text]
 getParamKeys params = [k | (k := _) <- params]