/
Node.hs
331 lines (309 loc) · 12.4 KB
/
Node.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
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE NumericUnderscores #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeApplications #-}
-- | Run the whole Node
--
-- Intended for qualified import.
--
module Ouroboros.Consensus.Node
( DiffusionTracers (..)
, DiffusionArguments (..)
, run
, IsProducer (..)
-- * Exposed by 'run'
, RunNode (..)
, Tracers
, Tracers' (..)
, ChainDB.TraceEvent (..)
, ProtocolInfo (..)
, ChainDbArgs (..)
, NodeArgs (..)
, NodeKernel (..)
, MaxBlockSizeOverride (..)
, MempoolCapacityBytesOverride (..)
, IPSubscriptionTarget (..)
, DnsSubscriptionTarget (..)
, ConnectionId (..)
-- * Internal helpers
, openChainDB
, mkChainDbArgs
, mkNodeArgs
) where
import Codec.Serialise (DeserialiseFailure)
import Control.Monad (when)
import Control.Tracer (Tracer)
import Crypto.Random
import Data.ByteString.Lazy (ByteString)
import Data.Proxy (Proxy (..))
import Data.Time.Clock (secondsToDiffTime)
import Ouroboros.Network.Diffusion
import Ouroboros.Network.Magic
import Ouroboros.Network.NodeToClient (DictVersion (..),
NodeToClientVersion (..), NodeToClientVersionData (..),
nodeToClientCodecCBORTerm)
import Ouroboros.Network.NodeToNode (NodeToNodeVersion (..),
NodeToNodeVersionData (..), nodeToNodeCodecCBORTerm)
import Ouroboros.Network.Protocol.ChainSync.PipelineDecision
(pipelineDecisionLowHighMark)
import Ouroboros.Network.Socket (ConnectionId)
import Ouroboros.Consensus.Block (BlockProtocol)
import Ouroboros.Consensus.BlockchainTime
import Ouroboros.Consensus.ChainSyncClient (ClockSkew (..))
import Ouroboros.Consensus.Ledger.Extended (ExtLedgerState (..))
import Ouroboros.Consensus.Node.DbMarker
import Ouroboros.Consensus.Node.ErrorPolicy
import Ouroboros.Consensus.Node.ProtocolInfo
import Ouroboros.Consensus.Node.Recovery
import Ouroboros.Consensus.Node.Run
import Ouroboros.Consensus.Node.Tracers
import Ouroboros.Consensus.NodeKernel
import Ouroboros.Consensus.NodeNetwork
import Ouroboros.Consensus.Protocol hiding (Protocol)
import Ouroboros.Consensus.Util.IOLike
import Ouroboros.Consensus.Util.Orphans ()
import Ouroboros.Consensus.Util.ResourceRegistry
import Ouroboros.Storage.ChainDB (ChainDB, ChainDbArgs)
import qualified Ouroboros.Storage.ChainDB as ChainDB
import Ouroboros.Storage.EpochInfo (EpochInfo, newEpochInfo)
import Ouroboros.Storage.FS.API.Types
import Ouroboros.Storage.FS.IO (ioHasFS)
import Ouroboros.Storage.ImmutableDB (ValidationPolicy (..))
import Ouroboros.Storage.LedgerDB.DiskPolicy (defaultDiskPolicy)
import Ouroboros.Storage.LedgerDB.InMemory (ledgerDbDefaultParams)
import Ouroboros.Storage.VolatileDB (BlockValidationPolicy (..),
mkBlocksPerFile)
-- | Whether the node produces blocks or not.
data IsProducer
= IsProducer
| IsNotProducer
deriving (Eq, Show)
-- | Start a node.
--
-- This opens the 'ChainDB', sets up the 'NodeKernel' and initialises the
-- network layer.
--
-- This function runs forever unless an exception is thrown.
run
:: forall blk.
RunNode blk
=> Tracers IO ConnectionId blk -- ^ Consensus tracers
-> ProtocolTracers IO ConnectionId blk DeserialiseFailure
-- ^ Protocol tracers
-> Tracer IO (ChainDB.TraceEvent blk) -- ^ ChainDB tracer
-> DiffusionTracers -- ^ Diffusion tracers
-> DiffusionArguments -- ^ Diffusion arguments
-> NetworkMagic
-> FilePath -- ^ Database path
-> ProtocolInfo blk
-> IsProducer
-> (ChainDbArgs IO blk -> ChainDbArgs IO blk)
-- ^ Customise the 'ChainDbArgs'
-> (NodeArgs IO ConnectionId blk -> NodeArgs IO ConnectionId blk)
-- ^ Customise the 'NodeArgs'
-> (ResourceRegistry IO -> NodeKernel IO ConnectionId blk -> IO ())
-- ^ Called on the 'NodeKernel' after creating it, but before the network
-- layer is initialised.
-> IO ()
run tracers protocolTracers chainDbTracer diffusionTracers diffusionArguments
networkMagic dbPath pInfo isProducer customiseChainDbArgs
customiseNodeArgs onNodeKernel = do
either throwM return =<< checkDbMarker
hasFS
mountPoint
(nodeProtocolMagicId (Proxy @blk) cfg)
withRegistry $ \registry -> do
lockDbMarkerFile registry dbPath
btime <- realBlockchainTime
registry
(blockchainTimeTracer tracers)
(nodeStartTime (Proxy @blk) cfg)
(focusSlotLengths slotLengths)
-- When we shut down cleanly, we create a marker file so that the next
-- time we start, we know we don't have to validate the contents of the
-- whole ChainDB. When we shut down with an exception indicating
-- corruption or something going wrong with the file system, we don't
-- create this marker file so that the next time we start, we do a full
-- validation.
lastShutDownWasClean <- hasCleanShutdownMarker hasFS
when lastShutDownWasClean $ removeCleanShutdownMarker hasFS
let customiseChainDbArgs' args
| lastShutDownWasClean
= customiseChainDbArgs args
| otherwise
-- When the last shutdown was not clean, validate the complete
-- ChainDB to detect and recover from any corruptions. This will
-- override the default value /and/ the user-customised value of
-- the 'ChainDB.cdbImmValidation' and the
-- 'ChainDB.cdbVolValidation' fields.
= (customiseChainDbArgs args)
{ ChainDB.cdbImmValidation = ValidateAllEpochs
, ChainDB.cdbVolValidation = ValidateAll
}
-- On a clean shutdown, create a marker in the database folder so that
-- next time we start up, we know we don't have to validate the whole
-- database.
createMarkerOnCleanShutdown hasFS $ do
(_, chainDB) <- allocate registry
(\_ -> openChainDB
chainDbTracer registry btime dbPath cfg initLedger
customiseChainDbArgs')
ChainDB.closeDB
let nodeArgs = customiseNodeArgs $ mkNodeArgs
registry
cfg
initState
tracers
btime
chainDB
isProducer
nodeKernel <- initNodeKernel nodeArgs
onNodeKernel registry nodeKernel
let networkApps = mkNetworkApps nodeArgs nodeKernel
diffusionApplications = mkDiffusionApplications networkApps
runDataDiffusion diffusionTracers
diffusionArguments
diffusionApplications
where
mountPoint = MountPoint dbPath
hasFS = ioHasFS mountPoint
ProtocolInfo
{ pInfoConfig = cfg
, pInfoInitLedger = initLedger
, pInfoInitState = initState
} = pInfo
slotLengths = protocolSlotLengths cfg
nodeToNodeVersionData = NodeToNodeVersionData { networkMagic = networkMagic }
nodeToClientVersionData = NodeToClientVersionData { networkMagic = networkMagic }
mkNetworkApps
:: NodeArgs IO ConnectionId blk
-> NodeKernel IO ConnectionId blk
-> NetworkApplication
IO ConnectionId
ByteString ByteString ByteString ByteString ByteString ByteString
()
mkNetworkApps nodeArgs nodeKernel = consensusNetworkApps
nodeKernel
protocolTracers
(protocolCodecs (getNodeConfig nodeKernel))
(protocolHandlers nodeArgs nodeKernel)
mkDiffusionApplications networkApps = DiffusionApplications
{ daResponderApplication =
simpleSingletonVersions
NodeToNodeV_1
nodeToNodeVersionData
(DictVersion nodeToNodeCodecCBORTerm)
(responderNetworkApplication networkApps)
, daInitiatorApplication =
simpleSingletonVersions
NodeToNodeV_1
nodeToNodeVersionData
(DictVersion nodeToNodeCodecCBORTerm)
(initiatorNetworkApplication networkApps)
, daLocalResponderApplication =
simpleSingletonVersions
NodeToClientV_1
nodeToClientVersionData
(DictVersion nodeToClientCodecCBORTerm)
(localResponderNetworkApplication networkApps)
, daErrorPolicies = consensusErrorPolicy
}
openChainDB
:: forall blk. RunNode blk
=> Tracer IO (ChainDB.TraceEvent blk)
-> ResourceRegistry IO
-> BlockchainTime IO
-> FilePath
-- ^ Database path
-> NodeConfig (BlockProtocol blk)
-> ExtLedgerState blk
-- ^ Initial ledger
-> (ChainDbArgs IO blk -> ChainDbArgs IO blk)
-- ^ Customise the 'ChainDbArgs'
-> IO (ChainDB IO blk)
openChainDB tracer registry btime dbPath cfg initLedger customiseArgs = do
epochInfo <- newEpochInfo $ nodeEpochSize (Proxy @blk) cfg
let args = customiseArgs $
mkChainDbArgs tracer registry btime dbPath cfg initLedger
epochInfo
ChainDB.openDB args
mkChainDbArgs
:: forall blk. RunNode blk
=> Tracer IO (ChainDB.TraceEvent blk)
-> ResourceRegistry IO
-> BlockchainTime IO
-> FilePath
-- ^ Database path
-> NodeConfig (BlockProtocol blk)
-> ExtLedgerState blk
-- ^ Initial ledger
-> EpochInfo IO
-> ChainDbArgs IO blk
mkChainDbArgs tracer registry btime dbPath cfg initLedger
epochInfo = (ChainDB.defaultArgs dbPath)
{ ChainDB.cdbBlocksPerFile = mkBlocksPerFile 1000
, ChainDB.cdbDecodeBlock = nodeDecodeBlock cfg
, ChainDB.cdbDecodeHeader = nodeDecodeHeader cfg
, ChainDB.cdbDecodeChainState = nodeDecodeChainState (Proxy @blk) cfg
, ChainDB.cdbDecodeHash = nodeDecodeHeaderHash (Proxy @blk)
, ChainDB.cdbDecodeLedger = nodeDecodeLedgerState cfg
, ChainDB.cdbDecodeTipInfo = nodeDecodeTipInfo (Proxy @blk)
, ChainDB.cdbEncodeBlock = nodeEncodeBlockWithInfo cfg
, ChainDB.cdbEncodeHeader = nodeEncodeHeader cfg
, ChainDB.cdbEncodeChainState = nodeEncodeChainState (Proxy @blk) cfg
, ChainDB.cdbEncodeHash = nodeEncodeHeaderHash (Proxy @blk)
, ChainDB.cdbEncodeLedger = nodeEncodeLedgerState cfg
, ChainDB.cdbEncodeTipInfo = nodeEncodeTipInfo (Proxy @blk)
, ChainDB.cdbEpochInfo = epochInfo
, ChainDB.cdbHashInfo = nodeHashInfo (Proxy @blk)
, ChainDB.cdbGenesis = return initLedger
, ChainDB.cdbAddHdrEnv = nodeAddHeaderEnvelope (Proxy @blk)
, ChainDB.cdbDiskPolicy = defaultDiskPolicy secParam
, ChainDB.cdbIsEBB = nodeIsEBB
, ChainDB.cdbCheckIntegrity = nodeCheckIntegrity cfg
, ChainDB.cdbParamsLgrDB = ledgerDbDefaultParams secParam
, ChainDB.cdbNodeConfig = cfg
, ChainDB.cdbRegistry = registry
, ChainDB.cdbTracer = tracer
, ChainDB.cdbImmValidation = ValidateMostRecentEpoch
, ChainDB.cdbVolValidation = NoValidation
, ChainDB.cdbGcDelay = secondsToDiffTime 10
, ChainDB.cdbBlockchainTime = btime
}
where
secParam = protocolSecurityParam cfg
mkNodeArgs
:: forall blk. RunNode blk
=> ResourceRegistry IO
-> NodeConfig (BlockProtocol blk)
-> NodeState (BlockProtocol blk)
-> Tracers IO ConnectionId blk
-> BlockchainTime IO
-> ChainDB IO blk
-> IsProducer
-> NodeArgs IO ConnectionId blk
mkNodeArgs registry cfg initState tracers btime chainDB isProducer = NodeArgs
{ tracers
, registry
, maxClockSkew = ClockSkew 1
, cfg
, initState
, btime
, chainDB
, initChainDB = nodeInitChainDB
, blockProduction
, blockFetchSize = nodeBlockFetchSize
, blockMatchesHeader = nodeBlockMatchesHeader
, maxUnackTxs = 100 -- TODO
, maxBlockSize = NoOverride
, mempoolCap = NoMempoolCapacityBytesOverride
, chainSyncPipelining = pipelineDecisionLowHighMark 200 300 -- TODO
}
where
blockProduction = case isProducer of
IsNotProducer -> Nothing
IsProducer -> Just BlockProduction
{ produceDRG = drgNew
, produceBlock = nodeForgeBlock cfg
}