Documentation
¶
Overview ¶
Package expingest contains the new ingestion system for horizon. It currently runs completely independent of the old one, that means that the new system can be ledgers behind/ahead the old system.
Index ¶
Constants ¶
const ( // CurrentVersion reflects the latest version of the ingestion // algorithm. This value is stored in KV store and is used to decide // if there's a need to reprocess the ledger state or reingest data. // // Version history: // - 1: Initial version // - 2: Added the orderbook, offers processors and distributed ingestion. // - 3: Fixed a bug that could potentialy result in invalid state // (#1722). Update the version to clear the state. // - 4: Fixed a bug in AccountSignersChanged method. // - 5: Added trust lines. // - 6: Added accounts and accounts data. // - 7: Fixes a bug in AccountSignersChanged method. // - 8: Fixes AccountSigners processor to remove preauth tx signer // when preauth tx is failed. // - 9: Fixes a bug in asset stats processor that counted unauthorized // trustlines. // - 10: Fixes a bug in meta processing (fees are now processed before // everything else). CurrentVersion = 10 )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
CoreSession *db.Session
PaydexCoreURL string
HistorySession *db.Session
HistoryArchiveURL string
TempSet io.TempSet
DisableStateVerification bool
// MaxStreamRetries determines how many times the reader will retry when encountering
// errors while streaming xdr bucket entries from the history archive.
// Set MaxStreamRetries to 0 if there should be no retry attempts
MaxStreamRetries int
OrderBookGraph *orderbook.OrderBookGraph
}
type LoggingLedgerReporter ¶
LoggingLedgerReporter logs the progress of a session running its ledger pipelines
func (*LoggingLedgerReporter) OnEndLedger ¶
func (lr *LoggingLedgerReporter) OnEndLedger(err error, shutdown bool)
OnEndLedger logs that the session has finished processing the ledger
func (*LoggingLedgerReporter) OnLedgerTransaction ¶
func (lr *LoggingLedgerReporter) OnLedgerTransaction()
OnLedgerTransaction records that the session has processed a transaction from the ledger
func (*LoggingLedgerReporter) OnNewLedger ¶
func (lr *LoggingLedgerReporter) OnNewLedger(sequence uint32)
OnNewLedger logs that the session has started reading a new ledger
type LoggingStateReporter ¶
type LoggingStateReporter struct {
Log *logpkg.Entry
Interval int
// contains filtered or unexported fields
}
LoggingStateReporter logs the progress of a session running its state pipelines
func (*LoggingStateReporter) OnEndState ¶
func (lr *LoggingStateReporter) OnEndState(err error, shutdown bool)
OnEndState logs that the session has finished processing the history archive snapshot
func (*LoggingStateReporter) OnStartState ¶
func (lr *LoggingStateReporter) OnStartState(sequence uint32)
OnStartState logs that the session has started reading from the history archive snapshot
func (*LoggingStateReporter) OnStateEntry ¶
func (lr *LoggingStateReporter) OnStateEntry()
OnStateEntry logs that the session has processed an entry from the history archive snapshot
type System ¶
type System struct {
// contains filtered or unexported fields
}
func (*System) Run ¶
func (s *System) Run()
Run starts ingestion system. Ingestion system supports distributed ingestion that means that Horizon ingestion can be running on multiple machines and only one, random node will lead the ingestion.
It needs to support cartesian product of the following run scenarios cases: - Init from empty state (1a) and resuming from existing state (1b). - Ingestion system version has been upgraded (2a) or not (2b). - Current node is leading ingestion (3a) or not (3b).
We always clear state when ingestion system is upgraded so 2a and 2b are included in 1a.
We ensure that only one instance is a leader because in each round instances try to acquire a lock on `LastLedgerExpIngest value in key value store and only one instance will be able to acquire it. This happens in both initial processing and ledger processing. So this solves 3a and 3b in both 1a and 1b.
Finally, 1a and 1b are tricky because we need to keep the latest version of order book graph in memory of each Horizon instance. To solve this: * For state init:
- If instance is a leader, we update the order book graph by running state pipeline normally.
- If instance is NOT a leader, we build a graph from offers present in a database. We completely omit state pipeline in this case.
* For resuming:
- If instances is a leader, it runs full ledger pipeline, including updating a database.
- If instances is a NOT leader, it runs ledger pipeline without updating a a database so order book graph is updated but database is not overwritten.
func (*System) StateReady ¶
StateReady returns true if the ingestion system has finished running it's state pipelines
Source Files
Directories