-
Notifications
You must be signed in to change notification settings - Fork 156
/
Api.hs
413 lines (377 loc) · 13.2 KB
/
Api.hs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE TypeApplications #-}
{-# LANGUAGE NoImplicitPrelude #-}
module Cardano.DbSync.Api (
SyncEnv (..),
LedgerEnv (..),
SyncOptions (..),
InsertOptions (..),
ConsistentLevel (..),
RunMigration,
defaultInsertOptions,
setConsistentLevel,
getConsistentLevel,
isConsistent,
getIsSyncFixed,
setIsFixedAndMigrate,
getRanIndexes,
runIndexMigrations,
mkSyncEnvFromConfig,
replaceConnection,
verifySnapshotPoint,
getTrace,
getInsertOptions,
getBackend,
hasLedgerState,
getLatestPoints,
getSlotHash,
getDbLatestBlockInfo,
getDbTipBlockNo,
getCurrentTipBlockNo,
generateNewEpochEvents,
logDbState,
convertToPoint,
) where
import Cardano.BM.Trace (Trace, logInfo)
import qualified Cardano.Chain.Genesis as Byron
import Cardano.Crypto.ProtocolMagic (ProtocolMagicId (..))
import qualified Cardano.Db as DB
import Cardano.DbSync.Cache
import Cardano.DbSync.Config.Cardano
import Cardano.DbSync.Config.Shelley
import Cardano.DbSync.Config.Types
import Cardano.DbSync.Error
import Cardano.DbSync.LedgerState
import Cardano.DbSync.LocalStateQuery
import Cardano.DbSync.Types
import Cardano.DbSync.Util
import qualified Cardano.Ledger.BaseTypes as Ledger
import qualified Cardano.Ledger.Shelley.Genesis as Shelley
import Cardano.Prelude
import Cardano.Slotting.Slot (EpochNo (..), SlotNo (..), WithOrigin (..))
import Control.Concurrent.Class.MonadSTM.Strict (
StrictTVar,
newTBQueueIO,
newTVarIO,
readTVar,
readTVarIO,
writeTVar,
)
import Control.Concurrent.Class.MonadSTM.Strict.TBQueue (StrictTBQueue)
import Control.Monad.Trans.Maybe (MaybeT (..))
import qualified Data.Strict.Maybe as Strict
import Data.Time.Clock (UTCTime, getCurrentTime)
import Database.Persist.Postgresql (ConnectionString)
import Database.Persist.Sql (SqlBackend)
import Ouroboros.Consensus.Block.Abstract (HeaderHash, Point (..), fromRawHash)
import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
import Ouroboros.Consensus.Node.ProtocolInfo (ProtocolInfo)
import Ouroboros.Network.Block (BlockNo (..), Point (..))
import Ouroboros.Network.Magic (NetworkMagic (..))
import qualified Ouroboros.Network.Point as Point
data SyncEnv = SyncEnv
{ envProtocol :: !SyncProtocol
, envNetworkMagic :: !NetworkMagic
, envSystemStart :: !SystemStart
, envConnString :: ConnectionString
, envRunDelayedMigration :: RunMigration
, envBackend :: !(StrictTVar IO (Strict.Maybe SqlBackend))
, envConsistentLevel :: !(StrictTVar IO ConsistentLevel)
, envIsFixed :: !(StrictTVar IO Bool)
, envIndexes :: !(StrictTVar IO Bool)
, envOptions :: !SyncOptions
, envCache :: !Cache
, envOfflineWorkQueue :: !(StrictTBQueue IO PoolFetchRetry)
, envOfflineResultQueue :: !(StrictTBQueue IO FetchResult)
, envEpochState :: !(StrictTVar IO EpochState)
, envEpochSyncTime :: !(StrictTVar IO UTCTime)
, envNoLedgerEnv :: !NoLedgerStateEnv -- only used when configured without ledger state.
, envLedger :: !LedgerEnv
}
type RunMigration = DB.MigrationToRun -> IO ()
data ConsistentLevel = Consistent | DBAheadOfLedger | Unchecked
deriving (Show, Eq)
setConsistentLevel :: SyncEnv -> ConsistentLevel -> IO ()
setConsistentLevel env cst = do
logInfo (getTrace env) $ "Setting ConsistencyLevel to " <> textShow cst
atomically $ writeTVar (envConsistentLevel env) cst
getConsistentLevel :: SyncEnv -> IO ConsistentLevel
getConsistentLevel env =
readTVarIO (envConsistentLevel env)
isConsistent :: SyncEnv -> IO Bool
isConsistent env = do
cst <- getConsistentLevel env
case cst of
Consistent -> pure True
_ -> pure False
getIsSyncFixed :: SyncEnv -> IO Bool
getIsSyncFixed = readTVarIO . envIsFixed
setIsFixedAndMigrate :: SyncEnv -> IO ()
setIsFixedAndMigrate env = do
envRunDelayedMigration env DB.Fix
atomically $ writeTVar (envIsFixed env) True
getRanIndexes :: SyncEnv -> IO Bool
getRanIndexes env = do
readTVarIO $ envIndexes env
runIndexMigrations :: SyncEnv -> IO ()
runIndexMigrations env = do
haveRan <- readTVarIO $ envIndexes env
unless haveRan $ do
envRunDelayedMigration env DB.Indexes
logInfo trce "Indexes were created"
atomically $ writeTVar (envIndexes env) True
where
trce = getTrace env
data SyncOptions = SyncOptions
{ soptExtended :: !Bool
, soptAbortOnInvalid :: !Bool
, soptCache :: !Bool
, soptLedger :: !Bool
, soptSkipFix :: !Bool
, soptOnlyFix :: !Bool
, soptInsertOptions :: !InsertOptions
, snapshotEveryFollowing :: !Word64
, snapshotEveryLagging :: !Word64
}
data InsertOptions = InsertOptions
{ ioMultiAssets :: !Bool
, ioMetadata :: !Bool
, ioPlutusExtra :: !Bool
}
defaultInsertOptions :: InsertOptions
defaultInsertOptions = InsertOptions True True True
replaceConnection :: SyncEnv -> SqlBackend -> IO ()
replaceConnection env sqlBackend = do
atomically $ writeTVar (envBackend env) $ Strict.Just sqlBackend
data EpochState = EpochState
{ esInitialized :: !Bool
, esEpochNo :: !(Strict.Maybe EpochNo)
}
initEpochState :: EpochState
initEpochState =
EpochState
{ esInitialized = False
, esEpochNo = Strict.Nothing
}
generateNewEpochEvents :: SyncEnv -> SlotDetails -> STM [LedgerEvent]
generateNewEpochEvents env details = do
!oldEpochState <- readTVar (envEpochState env)
writeTVar (envEpochState env) newEpochState
pure $ maybeToList (newEpochEvent oldEpochState)
where
currentEpochNo :: EpochNo
currentEpochNo = sdEpochNo details
newEpochEvent :: EpochState -> Maybe LedgerEvent
newEpochEvent oldEpochState =
case esEpochNo oldEpochState of
Strict.Nothing -> Just $ LedgerStartAtEpoch currentEpochNo
Strict.Just oldEpoch ->
if currentEpochNo == 1 + oldEpoch
then Just $ LedgerNewEpoch currentEpochNo (getSyncStatus details)
else Nothing
newEpochState :: EpochState
newEpochState =
EpochState
{ esInitialized = True
, esEpochNo = Strict.Just currentEpochNo
}
getTrace :: SyncEnv -> Trace IO Text
getTrace = leTrace . envLedger
getInsertOptions :: SyncEnv -> InsertOptions
getInsertOptions = soptInsertOptions . envOptions
getSlotHash :: SqlBackend -> SlotNo -> IO [(SlotNo, ByteString)]
getSlotHash backend = DB.runDbIohkNoLogging backend . DB.querySlotHash
getBackend :: SyncEnv -> IO SqlBackend
getBackend env = do
mBackend <- readTVarIO $ envBackend env
case mBackend of
Strict.Just conn -> pure conn
Strict.Nothing -> panic "sql connection not initiated"
hasLedgerState :: SyncEnv -> Bool
hasLedgerState = soptLedger . envOptions
getDbLatestBlockInfo :: SqlBackend -> IO (Maybe TipInfo)
getDbLatestBlockInfo backend = do
runMaybeT $ do
block <- MaybeT $ DB.runDbIohkNoLogging backend DB.queryLatestBlock
-- The EpochNo, SlotNo and BlockNo can only be zero for the Byron
-- era, but we need to make the types match, hence `fromMaybe`.
pure $
TipInfo
{ bHash = DB.blockHash block
, bEpochNo = EpochNo . fromMaybe 0 $ DB.blockEpochNo block
, bSlotNo = SlotNo . fromMaybe 0 $ DB.blockSlotNo block
, bBlockNo = BlockNo . fromMaybe 0 $ DB.blockBlockNo block
}
getDbTipBlockNo :: SyncEnv -> IO (Point.WithOrigin BlockNo)
getDbTipBlockNo env =
getBackend env
>>= getDbLatestBlockInfo
<&> maybe Point.Origin (Point.At . bBlockNo)
logDbState :: SyncEnv -> IO ()
logDbState env = do
backend <- getBackend env
mblk <- getDbLatestBlockInfo backend
case mblk of
Nothing -> logInfo (getTrace env) "Cardano.Db is empty"
Just tip -> logInfo (getTrace env) $ mconcat ["Cardano.Db tip is at ", showTip tip]
where
showTip :: TipInfo -> Text
showTip tipInfo =
mconcat
[ "slot "
, DB.textShow (unSlotNo $ bSlotNo tipInfo)
, ", block "
, DB.textShow (unBlockNo $ bBlockNo tipInfo)
]
getCurrentTipBlockNo :: SyncEnv -> IO (WithOrigin BlockNo)
getCurrentTipBlockNo env = do
backend <- getBackend env
maybeTip <- getDbLatestBlockInfo backend
case maybeTip of
Just tip -> pure $ At (bBlockNo tip)
Nothing -> pure Origin
mkSyncEnv ::
Trace IO Text ->
ConnectionString ->
SyncOptions ->
ProtocolInfo IO CardanoBlock ->
Ledger.Network ->
NetworkMagic ->
SystemStart ->
LedgerStateDir ->
Bool ->
Bool ->
RunMigration ->
IO SyncEnv
mkSyncEnv trce connSring syncOptions protoInfo nw nwMagic systemStart dir ranAll forcedIndexes runMigration = do
ledgerEnv <-
mkLedgerEnv
trce
protoInfo
dir
nw
systemStart
(soptAbortOnInvalid syncOptions)
(snapshotEveryFollowing syncOptions)
(snapshotEveryLagging syncOptions)
cache <- if soptCache syncOptions then newEmptyCache 250000 else pure uninitiatedCache
backendVar <- newTVarIO Strict.Nothing
consistentLevelVar <- newTVarIO Unchecked
fixDataVar <- newTVarIO ranAll
indexesVar <- newTVarIO forcedIndexes
owq <- newTBQueueIO 100
orq <- newTBQueueIO 100
epochVar <- newTVarIO initEpochState
epochSyncTime <- newTVarIO =<< getCurrentTime
noLegdState <- mkNoLedgerStateEnv trce systemStart
pure $
SyncEnv
{ envProtocol = SyncProtocolCardano
, envNetworkMagic = nwMagic
, envSystemStart = systemStart
, envConnString = connSring
, envRunDelayedMigration = runMigration
, envBackend = backendVar
, envOptions = syncOptions
, envConsistentLevel = consistentLevelVar
, envIsFixed = fixDataVar
, envIndexes = indexesVar
, envCache = cache
, envOfflineWorkQueue = owq
, envOfflineResultQueue = orq
, envEpochState = epochVar
, envEpochSyncTime = epochSyncTime
, envNoLedgerEnv = noLegdState
, envLedger = ledgerEnv
}
mkSyncEnvFromConfig ::
Trace IO Text ->
ConnectionString ->
SyncOptions ->
LedgerStateDir ->
GenesisConfig ->
Bool ->
Bool ->
RunMigration ->
IO (Either SyncNodeError SyncEnv)
mkSyncEnvFromConfig trce connSring syncOptions dir genCfg ranAll forcedIndexes runMigration =
case genCfg of
GenesisCardano _ bCfg sCfg _
| unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
pure . Left . NECardanoConfig $
mconcat
[ "ProtocolMagicId "
, DB.textShow (unProtocolMagicId $ Byron.configProtocolMagicId bCfg)
, " /= "
, DB.textShow (Shelley.sgNetworkMagic $ scConfig sCfg)
]
| Byron.gdStartTime (Byron.configGenesisData bCfg) /= Shelley.sgSystemStart (scConfig sCfg) ->
pure . Left . NECardanoConfig $
mconcat
[ "SystemStart "
, DB.textShow (Byron.gdStartTime $ Byron.configGenesisData bCfg)
, " /= "
, DB.textShow (Shelley.sgSystemStart $ scConfig sCfg)
]
| otherwise ->
Right
<$> mkSyncEnv
trce
connSring
syncOptions
(mkProtocolInfoCardano genCfg [])
(Shelley.sgNetworkId $ scConfig sCfg)
(NetworkMagic . unProtocolMagicId $ Byron.configProtocolMagicId bCfg)
(SystemStart . Byron.gdStartTime $ Byron.configGenesisData bCfg)
dir
ranAll
forcedIndexes
runMigration
-- | 'True' is for in memory points and 'False' for on disk
getLatestPoints :: SyncEnv -> IO [(CardanoPoint, Bool)]
getLatestPoints env = do
if hasLedgerState env
then do
snapshotPoints <- listKnownSnapshots $ envLedger env
verifySnapshotPoint env snapshotPoints
else do
-- Brings the 5 latest.
dbBackend <- getBackend env
lastPoints <- DB.runDbIohkNoLogging dbBackend DB.queryLatestPoints
pure $ mapMaybe convert lastPoints
where
convert (Nothing, _) = Nothing
convert (Just slot, bs) = convertToDiskPoint (SlotNo slot) bs
verifySnapshotPoint :: SyncEnv -> [SnapshotPoint] -> IO [(CardanoPoint, Bool)]
verifySnapshotPoint env snapPoints =
catMaybes <$> mapM validLedgerFileToPoint snapPoints
where
validLedgerFileToPoint :: SnapshotPoint -> IO (Maybe (CardanoPoint, Bool))
validLedgerFileToPoint (OnDisk lsf) = do
backend <- getBackend env
hashes <- getSlotHash backend (lsfSlotNo lsf)
let valid = find (\(_, h) -> lsfHash lsf == hashToAnnotation h) hashes
case valid of
Just (slot, hash) | slot == lsfSlotNo lsf -> pure $ convertToDiskPoint slot hash
_ -> pure Nothing
validLedgerFileToPoint (InMemory pnt) = do
case pnt of
GenesisPoint -> pure Nothing
BlockPoint slotNo hsh -> do
backend <- getBackend env
hashes <- getSlotHash backend slotNo
let valid = find (\(_, dbHash) -> getHeaderHash hsh == dbHash) hashes
case valid of
Just (dbSlotNo, _) | slotNo == dbSlotNo -> pure $ Just (pnt, True)
_ -> pure Nothing
convertToDiskPoint :: SlotNo -> ByteString -> Maybe (CardanoPoint, Bool)
convertToDiskPoint slot hashBlob = (,False) <$> convertToPoint slot hashBlob
convertToPoint :: SlotNo -> ByteString -> Maybe CardanoPoint
convertToPoint slot hashBlob =
Point . Point.block slot <$> convertHashBlob hashBlob
where
convertHashBlob :: ByteString -> Maybe (HeaderHash CardanoBlock)
convertHashBlob = Just . fromRawHash (Proxy @CardanoBlock)