From c818bb87eaaf705ca4b4c1f4b5ded8cd0cf859ff Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Thu, 23 May 2024 15:40:56 -0700 Subject: [PATCH] Add new `getTransactions` endpoint and database backend for transactions (#174) * Increase `getTransaction` window via ingestion into a local database (#141) * Add common interface between in-memory and on-disk transaction storage * Revert "Add common interface between in-memory and on-disk transaction storage" This reverts commit 3196ade050bfa3ac300dab37fadaa89b2bc08cc4. * Add initial implementation of DB-backed tx store * Drop all references to the in-memory transaction store * Actually filter for the tx hash you want :facepalm: * Drop remaining reference to transaction store * Update stellar/go to latest master * Add a bunch of logging/verbose errors, another column to DB * Propogate contexts, fix all bugs; e2e flow works! * Add metric for tx ingestion * Split TransactionHandler into separate readers and writers * Move tx reads into their own isolated, read-only db tx * Add in-memory cache for ledger ranges * Remove cache: read/write are separate so it doesn't work * Add trimming of transactions table based on retention window * Tests work (but dont pass)! Plus some fixes based on test cases :+1: * Tidy up the go.mod file * Fixup the way ledger ranges are handled * Add prometheus metrics back * Drop columns and logging that aren't useful * Incorporate errors into ledger range retrieval * Return instead of log errors in getTransaction * Fixup tests: an empty db isn't a fetch error * Make 24hr the new default transaction retention window, fixup errors * Expect the correct retention window in tests * Use raw byte hash over hex for 1/2 the column space * Rename interfaces to avoid overloading 'Transaction' term * PR feedback: variable rename * Separate tx parsing into its own function * Modify queries to not use transactions * Pass around the logger rather than using the global * Fixup all of the tests to pass a logger * Use subqueries to get around () sqlite limitations * Use .Select() to simplify querying * Add db unit test but it's failing :( * Prefer custom error over EOF or DB nonsense * Prepare randoms in advance, writer for the ledger as well * Pass daemon to db to hook in metrics * Check another error in e2e test * Feedback: use no-op interface, drop metric * Refactor method tests to use 'fake' tx/ledger backend * Simplify code a bit * Fixup mocking interface * Move test function back, fixup metrics and tests * Add `getTransactions` endpoint (#136) * Add cursor for getTransactions * Add validation of pagination and range * implement getTransactions handler - 1 * Add getLedgers * Read a single ledger instead of getting all ledgers * Add ParseCursor method for tx cursor * Cursor changes - 1 * Cursor changes - 2 * Cursor changes - 3 * revert go-mod changes * Use reader.Seek() * Use reader.Seek() - 2 * Add config-options for tx pagination limits * Fix failing cursor test * Go mod changes * Go mod changes - 2 * Add unittests - 1 * Update go.mod and go.sum * Add integration tests * Update go.mod and go.sum * Add ledgerSeq to error string * Add docstrings * Change transactions limits * add defensive check for error other than EOF * add defensive check for error other than EOF - 2 * Change ledger sequence to uint32 * Add comments/docstrings * Include only cursor in response * Use toid instead of new cursor * Revert cursor changes * Return cursor as string in result * Refactor reader.Seek error handling * Small refactoring * Remove startLedger check * Remove endLedger * Import fix * Fix failing tests * Refactor to use new transaction db * Refactor mocks * Refactor unittests for using the new db changes * Refactor integration test * Add config vars for max requests and request duration * Fix failing test * Use transactionInfo struct instead of db.Transactions * Start indexing from 1 instead of 0 for toid * Operation index from 1 * Add lines to make sure structs implement interfaces * Remove omitempty * rename test func * make txInfo struct public and convert string cursor to int * Use sendSuccessfulTransaction helper func * Convert cursor to string in request * Change jrpc response codes * Change ledger close meta code to invalid-params * Revert back to InvalidParams error for reader.Read() * Refactor if-else statement * Refactor if-else statement - 2 * Add documentation on opting into the new db backend (#182) * Refactor getFeeStats changes to add db changes * Remove code making `GetLedgerRange` conform to interface (#186) It's unnecessary because we aren't using that interface anymore. * Add changelog entry for `getTransactions` (#183) * Add more changelog details on new endpoint * Fixup spacing --------- Co-authored-by: George Co-authored-by: George Kudrayvtsev --- CHANGELOG.md | 39 +- cmd/soroban-rpc/internal/config/config.go | 4 + cmd/soroban-rpc/internal/config/options.go | 49 +- cmd/soroban-rpc/internal/daemon/daemon.go | 39 +- .../internal/daemon/interfaces/noOpDaemon.go | 2 +- cmd/soroban-rpc/internal/db/db.go | 91 +++- cmd/soroban-rpc/internal/db/ledger.go | 7 +- cmd/soroban-rpc/internal/db/ledger_test.go | 32 +- .../internal/db/ledgerentry_test.go | 44 +- .../db/migrations/02_transactions.sql | 13 + cmd/soroban-rpc/internal/db/mocks.go | 110 +++++ cmd/soroban-rpc/internal/db/transaction.go | 304 ++++++++++++ .../internal/db/transaction_test.go | 205 ++++++++ cmd/soroban-rpc/internal/events/events.go | 6 +- .../internal/ingest/mock_db_test.go | 22 +- cmd/soroban-rpc/internal/ingest/service.go | 30 +- .../internal/ingest/service_test.go | 12 +- cmd/soroban-rpc/internal/jsonrpc.go | 46 +- .../ledgerbucketwindow/ledgerbucketwindow.go | 9 +- .../internal/methods/get_fee_stats.go | 12 +- .../internal/methods/get_transaction.go | 33 +- .../internal/methods/get_transaction_test.go | 317 ++++++------- .../internal/methods/get_transactions.go | 241 ++++++++++ .../internal/methods/get_transactions_test.go | 302 ++++++++++++ cmd/soroban-rpc/internal/methods/health.go | 26 +- .../internal/methods/send_transaction.go | 20 +- .../internal/preflight/preflight_test.go | 11 +- .../internal/test/get_transactions_test.go | 101 ++++ cmd/soroban-rpc/internal/test/health_test.go | 2 +- cmd/soroban-rpc/internal/test/integration.go | 5 +- .../internal/test/transaction_test.go | 23 +- .../internal/transactions/transactions.go | 209 --------- .../transactions/transactions_test.go | 436 ------------------ 33 files changed, 1837 insertions(+), 965 deletions(-) create mode 100644 cmd/soroban-rpc/internal/db/migrations/02_transactions.sql create mode 100644 cmd/soroban-rpc/internal/db/mocks.go create mode 100644 cmd/soroban-rpc/internal/db/transaction.go create mode 100644 cmd/soroban-rpc/internal/db/transaction_test.go create mode 100644 cmd/soroban-rpc/internal/methods/get_transactions.go create mode 100644 cmd/soroban-rpc/internal/methods/get_transactions_test.go create mode 100644 cmd/soroban-rpc/internal/test/get_transactions_test.go delete mode 100644 cmd/soroban-rpc/internal/transactions/transactions.go delete mode 100644 cmd/soroban-rpc/internal/transactions/transactions_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 76829873..f6902465 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,44 @@ # Changelog ## Unreleased -n/a + +### Added +* Transactions will now be stored in a database rather than in memory ([#174](https://github.com/stellar/soroban-rpc/pull/174)). + +You can opt-in to longer transaction retention by setting `--transaction-retention-window` / `TRANSACTION_RETENTION_WINDOW` to a higher number of ledgers. This will also retain corresponding number of ledgers in the database. Keep in mind, of course, that this will cause an increase in disk usage for the growing database. + +* There is a new `getTransactions` endpoint with the following API ([#136](https://github.com/stellar/soroban-rpc/pull/136)): + +```typescript +interface Request { + startLedger: number; // uint32 + pagination?: { + cursor?: string; + limit?: number; // uint + } +} + +interface Response { + transactions: Transaction[]; // see below + latestLedger: number; // uint32 + latestLedgerCloseTimestamp: number; // int64 + oldestLedger: number; // uint32 + oldestLedgerCloseTimestamp: number; // int64 + cursor: string; +} + +interface Transaction { + status: boolean; // whether or not the transaction succeeded + applicationOrder: number; // int32, index of the transaction in the ledger + feeBump: boolean; // if it's a fee-bump transaction + envelopeXdr: string; // TransactionEnvelope XDR + resultXdr: string; // TransactionResult XDR + resultMetaXdr: string; // TransactionMeta XDR + ledger: number; // uint32, ledger sequence with this transaction + createdAt: int64; // int64, UNIX timestamp the transaction's inclusion + diagnosticEventsXdr?: string[]; // if failed, DiagnosticEvent XDRs +} +``` ## [v21.2.0](https://github.com/stellar/soroban-rpc/compare/v21.1.0...v21.2.0) diff --git a/cmd/soroban-rpc/internal/config/config.go b/cmd/soroban-rpc/internal/config/config.go index 139144b0..2ce48459 100644 --- a/cmd/soroban-rpc/internal/config/config.go +++ b/cmd/soroban-rpc/internal/config/config.go @@ -25,6 +25,7 @@ type Config struct { CheckpointFrequency uint32 CoreRequestTimeout time.Duration DefaultEventsLimit uint + DefaultTransactionsLimit uint EventLedgerRetentionWindow uint32 FriendbotURL string HistoryArchiveURLs []string @@ -33,6 +34,7 @@ type Config struct { LogFormat LogFormat LogLevel logrus.Level MaxEventsLimit uint + MaxTransactionsLimit uint MaxHealthyLedgerLatency time.Duration NetworkPassphrase string PreflightWorkerCount uint @@ -50,6 +52,7 @@ type Config struct { RequestBacklogGetLatestLedgerQueueLimit uint RequestBacklogGetLedgerEntriesQueueLimit uint RequestBacklogGetTransactionQueueLimit uint + RequestBacklogGetTransactionsQueueLimit uint RequestBacklogSendTransactionQueueLimit uint RequestBacklogSimulateTransactionQueueLimit uint RequestBacklogGetFeeStatsTransactionQueueLimit uint @@ -62,6 +65,7 @@ type Config struct { MaxGetLatestLedgerExecutionDuration time.Duration MaxGetLedgerEntriesExecutionDuration time.Duration MaxGetTransactionExecutionDuration time.Duration + MaxGetTransactionsExecutionDuration time.Duration MaxSendTransactionExecutionDuration time.Duration MaxSimulateTransactionExecutionDuration time.Duration MaxGetFeeStatsExecutionDuration time.Duration diff --git a/cmd/soroban-rpc/internal/config/options.go b/cmd/soroban-rpc/internal/config/options.go index 80d5068e..50f0321b 100644 --- a/cmd/soroban-rpc/internal/config/options.go +++ b/cmd/soroban-rpc/internal/config/options.go @@ -211,18 +211,22 @@ func (cfg *Config) options() ConfigOptions { }, { Name: "event-retention-window", - Usage: fmt.Sprintf("configures the event retention window expressed in number of ledgers,"+ - " the default value is %d which corresponds to about 24 hours of history", ledgerbucketwindow.DefaultEventLedgerRetentionWindow), + Usage: fmt.Sprintf( + "configures the event retention window expressed in number of ledgers,"+ + " the default value is %d which corresponds to about 24 hours of history", + ledgerbucketwindow.DefaultEventLedgerRetentionWindow), ConfigKey: &cfg.EventLedgerRetentionWindow, DefaultValue: uint32(ledgerbucketwindow.DefaultEventLedgerRetentionWindow), Validate: positive, }, { Name: "transaction-retention-window", - Usage: "configures the transaction retention window expressed in number of ledgers," + - " the default value is 1440 which corresponds to about 2 hours of history", + Usage: fmt.Sprintf( + "configures the transaction retention window expressed in number of ledgers,"+ + " the default value is %d which corresponds to about 24 hours of history", + ledgerbucketwindow.OneDayOfLedgers), ConfigKey: &cfg.TransactionLedgerRetentionWindow, - DefaultValue: uint32(1440), + DefaultValue: uint32(ledgerbucketwindow.OneDayOfLedgers), Validate: positive, }, { @@ -261,6 +265,28 @@ func (cfg *Config) options() ConfigOptions { return nil }, }, + { + Name: "max-transactions-limit", + Usage: "Maximum amount of transactions allowed in a single getTransactions response", + ConfigKey: &cfg.MaxTransactionsLimit, + DefaultValue: uint(200), + }, + { + Name: "default-transactions-limit", + Usage: "Default cap on the amount of transactions included in a single getTransactions response", + ConfigKey: &cfg.DefaultTransactionsLimit, + DefaultValue: uint(50), + Validate: func(co *ConfigOption) error { + if cfg.DefaultTransactionsLimit > cfg.MaxTransactionsLimit { + return fmt.Errorf( + "default-transactions-limit (%v) cannot exceed max-transactions-limit (%v)", + cfg.DefaultTransactionsLimit, + cfg.MaxTransactionsLimit, + ) + } + return nil + }, + }, { Name: "max-healthy-ledger-latency", Usage: "maximum ledger latency (i.e. time elapsed since the last known ledger closing time) considered to be healthy" + @@ -344,6 +370,13 @@ func (cfg *Config) options() ConfigOptions { DefaultValue: uint(1000), Validate: positive, }, + { + TomlKey: strutils.KebabToConstantCase("request-backlog-get-transactions-queue-limit"), + Usage: "Maximum number of outstanding GetTransactions requests", + ConfigKey: &cfg.RequestBacklogGetTransactionsQueueLimit, + DefaultValue: uint(1000), + Validate: positive, + }, { TomlKey: strutils.KebabToConstantCase("request-backlog-send-transaction-queue-limit"), Usage: "Maximum number of outstanding SendTransaction requests", @@ -419,6 +452,12 @@ func (cfg *Config) options() ConfigOptions { ConfigKey: &cfg.MaxGetTransactionExecutionDuration, DefaultValue: 5 * time.Second, }, + { + TomlKey: strutils.KebabToConstantCase("max-get-transactions-execution-duration"), + Usage: "The maximum duration of time allowed for processing a getTransactions request. When that time elapses, the rpc server would return -32001 and abort the request's execution", + ConfigKey: &cfg.MaxGetTransactionsExecutionDuration, + DefaultValue: 5 * time.Second, + }, { TomlKey: strutils.KebabToConstantCase("max-send-transaction-execution-duration"), Usage: "The maximum duration of time allowed for processing a sendTransaction request. When that time elapses, the rpc server would return -32001 and abort the request's execution", diff --git a/cmd/soroban-rpc/internal/daemon/daemon.go b/cmd/soroban-rpc/internal/daemon/daemon.go index d3432391..c38ccde9 100644 --- a/cmd/soroban-rpc/internal/daemon/daemon.go +++ b/cmd/soroban-rpc/internal/daemon/daemon.go @@ -19,6 +19,7 @@ import ( "github.com/stellar/go/ingest/ledgerbackend" supporthttp "github.com/stellar/go/support/http" supportlog "github.com/stellar/go/support/log" + "github.com/stellar/go/support/ordered" "github.com/stellar/go/support/storage" "github.com/stellar/go/xdr" @@ -28,8 +29,8 @@ import ( "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/events" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/feewindow" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ingest" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/preflight" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/transactions" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/util" ) @@ -135,7 +136,6 @@ func newCaptiveCore(cfg *config.Config, logger *supportlog.Entry) (*ledgerbacken func MustNew(cfg *config.Config) *Daemon { logger := supportlog.New() logger.SetLevel(cfg.LogLevel) - if cfg.LogFormat == config.LogFormatJSON { logger.UseJSONFormatter() } @@ -151,7 +151,7 @@ func MustNew(cfg *config.Config) *Daemon { } if len(cfg.HistoryArchiveURLs) == 0 { - logger.Fatal("no history archives url were provided") + logger.Fatal("no history archives URLs were provided") } historyArchive, err := historyarchive.NewArchivePool( @@ -192,11 +192,6 @@ func MustNew(cfg *config.Config) *Daemon { cfg.NetworkPassphrase, cfg.EventLedgerRetentionWindow, ) - transactionStore := transactions.NewMemoryStore( - daemon, - cfg.NetworkPassphrase, - cfg.TransactionLedgerRetentionWindow, - ) feewindows := feewindow.NewFeeWindows(cfg.ClassicFeeStatsLedgerRetentionWindow, cfg.SorobanFeeStatsLedgerRetentionWindow, cfg.NetworkPassphrase) // initialize the stores using what was on the DB @@ -222,9 +217,6 @@ func MustNew(cfg *config.Config) *Daemon { if err := eventStore.IngestEvents(txmeta); err != nil { logger.WithError(err).Fatal("could not initialize event memory store") } - if err := transactionStore.IngestTransactions(txmeta); err != nil { - logger.WithError(err).Fatal("could not initialize transaction memory store") - } if err := feewindows.IngestFees(txmeta); err != nil { logger.WithError(err).Fatal("could not initialize fee stats") } @@ -242,12 +234,27 @@ func MustNew(cfg *config.Config) *Daemon { onIngestionRetry := func(err error, dur time.Duration) { logger.WithError(err).Error("could not run ingestion. Retrying") } - maxRetentionWindow := max(cfg.EventLedgerRetentionWindow, cfg.TransactionLedgerRetentionWindow, cfg.ClassicFeeStatsLedgerRetentionWindow, cfg.SorobanFeeStatsLedgerRetentionWindow) + + // Take the larger of (event retention, tx retention) and then the smaller + // of (tx retention, default event retention) if event retention wasn't + // specified, for some reason...? + maxRetentionWindow := ordered.Max(cfg.EventLedgerRetentionWindow, cfg.TransactionLedgerRetentionWindow) + if cfg.EventLedgerRetentionWindow <= 0 { + maxRetentionWindow = ordered.Min( + maxRetentionWindow, + ledgerbucketwindow.DefaultEventLedgerRetentionWindow) + } ingestService := ingest.NewService(ingest.Config{ - Logger: logger, - DB: db.NewReadWriter(dbConn, maxLedgerEntryWriteBatchSize, maxRetentionWindow), + Logger: logger, + DB: db.NewReadWriter( + logger, + dbConn, + daemon, + maxLedgerEntryWriteBatchSize, + maxRetentionWindow, + cfg.NetworkPassphrase, + ), EventStore: eventStore, - TransactionStore: transactionStore, NetworkPassPhrase: cfg.NetworkPassphrase, Archive: historyArchive, LedgerBackend: core, @@ -271,11 +278,11 @@ func MustNew(cfg *config.Config) *Daemon { jsonRPCHandler := internal.NewJSONRPCHandler(cfg, internal.HandlerParams{ Daemon: daemon, EventStore: eventStore, - TransactionStore: transactionStore, FeeStatWindows: feewindows, Logger: logger, LedgerReader: db.NewLedgerReader(dbConn), LedgerEntryReader: db.NewLedgerEntryReader(dbConn), + TransactionReader: db.NewTransactionReader(logger, dbConn, cfg.NetworkPassphrase), PreflightGetter: preflightWorkerPool, }) diff --git a/cmd/soroban-rpc/internal/daemon/interfaces/noOpDaemon.go b/cmd/soroban-rpc/internal/daemon/interfaces/noOpDaemon.go index e73689a5..78e6cdbf 100644 --- a/cmd/soroban-rpc/internal/daemon/interfaces/noOpDaemon.go +++ b/cmd/soroban-rpc/internal/daemon/interfaces/noOpDaemon.go @@ -24,7 +24,7 @@ func MakeNoOpDeamon() *noOpDaemon { } func (d *noOpDaemon) MetricsRegistry() *prometheus.Registry { - return d.metricsRegistry + return prometheus.NewRegistry() // so that you can register metrics many times } func (d *noOpDaemon) MetricsNamespace() string { diff --git a/cmd/soroban-rpc/internal/db/db.go b/cmd/soroban-rpc/internal/db/db.go index 428f29fe..227e2115 100644 --- a/cmd/soroban-rpc/internal/db/db.go +++ b/cmd/soroban-rpc/internal/db/db.go @@ -15,7 +15,9 @@ import ( "github.com/stellar/go/support/db" "github.com/stellar/go/support/errors" + "github.com/stellar/go/support/log" "github.com/stellar/go/xdr" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" ) //go:embed migrations/*.sql @@ -34,8 +36,10 @@ type ReadWriter interface { } type WriteTx interface { + TransactionWriter() TransactionWriter LedgerEntryWriter() LedgerEntryWriter LedgerWriter() LedgerWriter + Commit(ledgerSeq uint32) error Rollback() error } @@ -130,21 +134,60 @@ func getLatestLedgerSequence(ctx context.Context, q db.SessionInterface, cache * return result, nil } +type ReadWriterMetrics struct { + TxIngestDuration, TxCount prometheus.Observer +} + type readWriter struct { + log *log.Entry db *DB maxBatchSize int ledgerRetentionWindow uint32 + passphrase string + + metrics ReadWriterMetrics } -// NewReadWriter constructs a new ReadWriter instance and configures -// the size of ledger entry batches when writing ledger entries -// and the retention window for how many historical ledgers are -// recorded in the database. -func NewReadWriter(db *DB, maxBatchSize int, ledgerRetentionWindow uint32) ReadWriter { +// NewReadWriter constructs a new readWriter instance and configures the size of +// ledger entry batches when writing ledger entries and the retention window for +// how many historical ledgers are recorded in the database, hooking up metrics +// for various DB ops. +func NewReadWriter( + log *log.Entry, + db *DB, + daemon interfaces.Daemon, + maxBatchSize int, + ledgerRetentionWindow uint32, + networkPassphrase string, +) ReadWriter { + // a metric for measuring latency of transaction store operations + txDurationMetric := prometheus.NewSummaryVec(prometheus.SummaryOpts{ + Namespace: daemon.MetricsNamespace(), Subsystem: "transactions", + Name: "operation_duration_seconds", + Help: "transaction store operation durations, sliding window = 10m", + Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, + }, + []string{"operation"}, + ) + txCountMetric := prometheus.NewSummary(prometheus.SummaryOpts{ + Namespace: daemon.MetricsNamespace(), Subsystem: "transactions", + Name: "count", + Help: "count of transactions ingested, sliding window = 10m", + Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, + }) + + daemon.MetricsRegistry().MustRegister(txDurationMetric, txCountMetric) + return &readWriter{ + log: log, db: db, maxBatchSize: maxBatchSize, ledgerRetentionWindow: ledgerRetentionWindow, + passphrase: networkPassphrase, + metrics: ReadWriterMetrics{ + TxIngestDuration: txDurationMetric.With(prometheus.Labels{"operation": "ingest"}), + TxCount: txCountMetric, + }, } } @@ -158,16 +201,18 @@ func (rw *readWriter) NewTx(ctx context.Context) (WriteTx, error) { return nil, err } stmtCache := sq.NewStmtCache(txSession.GetTx()) + db := rw.db - return writeTx{ + writer := writeTx{ globalCache: &db.cache, postCommit: func() error { _, err := db.ExecRaw(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") return err }, - tx: txSession, - stmtCache: stmtCache, - ledgerWriter: ledgerWriter{stmtCache: stmtCache}, + tx: txSession, + stmtCache: stmtCache, + ledgerRetentionWindow: rw.ledgerRetentionWindow, + ledgerWriter: ledgerWriter{stmtCache: stmtCache}, ledgerEntryWriter: ledgerEntryWriter{ stmtCache: stmtCache, buffer: xdr.NewEncodingBuffer(), @@ -175,8 +220,18 @@ func (rw *readWriter) NewTx(ctx context.Context) (WriteTx, error) { ledgerEntryCacheWriteTx: db.cache.ledgerEntries.newWriteTx(rw.maxBatchSize), maxBatchSize: rw.maxBatchSize, }, - ledgerRetentionWindow: rw.ledgerRetentionWindow, - }, nil + txWriter: transactionHandler{ + log: rw.log, + db: txSession, + stmtCache: stmtCache, + passphrase: rw.passphrase, + }, + } + writer.txWriter.RegisterMetrics( + rw.metrics.TxIngestDuration, + rw.metrics.TxCount) + + return writer, nil } type writeTx struct { @@ -186,6 +241,7 @@ type writeTx struct { stmtCache *sq.StmtCache ledgerEntryWriter ledgerEntryWriter ledgerWriter ledgerWriter + txWriter transactionHandler ledgerRetentionWindow uint32 } @@ -197,6 +253,10 @@ func (w writeTx) LedgerWriter() LedgerWriter { return w.ledgerWriter } +func (w writeTx) TransactionWriter() TransactionWriter { + return &w.txWriter +} + func (w writeTx) Commit(ledgerSeq uint32) error { if err := w.ledgerEntryWriter.flush(); err != nil { return err @@ -205,9 +265,14 @@ func (w writeTx) Commit(ledgerSeq uint32) error { if err := w.ledgerWriter.trimLedgers(ledgerSeq, w.ledgerRetentionWindow); err != nil { return err } + if err := w.txWriter.trimTransactions(ledgerSeq, w.ledgerRetentionWindow); err != nil { + return err + } - _, err := sq.Replace(metaTableName).RunWith(w.stmtCache). - Values(latestLedgerSequenceMetaKey, fmt.Sprintf("%d", ledgerSeq)).Exec() + _, err := sq.Replace(metaTableName). + Values(latestLedgerSequenceMetaKey, fmt.Sprintf("%d", ledgerSeq)). + RunWith(w.stmtCache). + Exec() if err != nil { return err } diff --git a/cmd/soroban-rpc/internal/db/ledger.go b/cmd/soroban-rpc/internal/db/ledger.go index e0bf7fd1..97887281 100644 --- a/cmd/soroban-rpc/internal/db/ledger.go +++ b/cmd/soroban-rpc/internal/db/ledger.go @@ -79,8 +79,11 @@ func (l ledgerWriter) trimLedgers(latestLedgerSeq uint32, retentionWindow uint32 return nil } cutoff := latestLedgerSeq + 1 - retentionWindow - deleteSQL := sq.StatementBuilder.RunWith(l.stmtCache).Delete(ledgerCloseMetaTableName).Where(sq.Lt{"sequence": cutoff}) - _, err := deleteSQL.Exec() + _, err := sq.StatementBuilder. + RunWith(l.stmtCache). + Delete(ledgerCloseMetaTableName). + Where(sq.Lt{"sequence": cutoff}). + Exec() return err } diff --git a/cmd/soroban-rpc/internal/db/ledger_test.go b/cmd/soroban-rpc/internal/db/ledger_test.go index bbbfdbee..f6ebd70b 100644 --- a/cmd/soroban-rpc/internal/db/ledger_test.go +++ b/cmd/soroban-rpc/internal/db/ledger_test.go @@ -2,13 +2,20 @@ package db import ( "context" + "path" "testing" "github.com/stretchr/testify/assert" + "github.com/stellar/go/network" + "github.com/stellar/go/support/log" "github.com/stellar/go/xdr" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" ) +var passphrase = network.FutureNetworkPassphrase +var logger = log.DefaultLogger + func createLedger(ledgerSequence uint32) xdr.LedgerCloseMeta { return xdr.LedgerCloseMeta{ V: 1, @@ -59,6 +66,7 @@ func assertLedgerRange(t *testing.T, reader LedgerReader, start, end uint32) { func TestLedgers(t *testing.T) { db := NewTestDB(t) + daemon := interfaces.MakeNoOpDeamon() reader := NewLedgerReader(db) _, exists, err := reader.GetLedger(context.Background(), 1) @@ -67,7 +75,7 @@ func TestLedgers(t *testing.T) { for i := 1; i <= 10; i++ { ledgerSequence := uint32(i) - tx, err := NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err := NewReadWriter(logger, db, daemon, 150, 15, passphrase).NewTx(context.Background()) assert.NoError(t, err) assert.NoError(t, tx.LedgerWriter().InsertLedger(createLedger(ledgerSequence))) assert.NoError(t, tx.Commit(ledgerSequence)) @@ -78,7 +86,7 @@ func TestLedgers(t *testing.T) { assertLedgerRange(t, reader, 1, 10) ledgerSequence := uint32(11) - tx, err := NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err := NewReadWriter(logger, db, daemon, 150, 15, passphrase).NewTx(context.Background()) assert.NoError(t, err) assert.NoError(t, tx.LedgerWriter().InsertLedger(createLedger(ledgerSequence))) assert.NoError(t, tx.Commit(ledgerSequence)) @@ -86,10 +94,28 @@ func TestLedgers(t *testing.T) { assertLedgerRange(t, reader, 1, 11) ledgerSequence = uint32(12) - tx, err = NewReadWriter(db, 150, 5).NewTx(context.Background()) + tx, err = NewReadWriter(logger, db, daemon, 150, 5, passphrase).NewTx(context.Background()) assert.NoError(t, err) assert.NoError(t, tx.LedgerWriter().InsertLedger(createLedger(ledgerSequence))) assert.NoError(t, tx.Commit(ledgerSequence)) assertLedgerRange(t, reader, 8, 12) } + +func NewTestDB(tb testing.TB) *DB { + tmp := tb.TempDir() + dbPath := path.Join(tmp, "db.sqlite") + db, err := OpenSQLiteDB(dbPath) + if err != nil { + assert.NoError(tb, db.Close()) + } + tb.Cleanup(func() { + assert.NoError(tb, db.Close()) + }) + return &DB{ + SessionInterface: db, + cache: dbCache{ + ledgerEntries: newTransactionalCache(), + }, + } +} diff --git a/cmd/soroban-rpc/internal/db/ledgerentry_test.go b/cmd/soroban-rpc/internal/db/ledgerentry_test.go index 1d5d6646..34b1b41e 100644 --- a/cmd/soroban-rpc/internal/db/ledgerentry_test.go +++ b/cmd/soroban-rpc/internal/db/ledgerentry_test.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "math/rand" - "path" "sync" "testing" "time" @@ -12,7 +11,9 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/stellar/go/support/log" "github.com/stellar/go/xdr" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" ) func getLedgerEntryAndLatestLedgerSequenceWithErr(db *DB, key xdr.LedgerKey) (bool, xdr.LedgerEntry, uint32, *uint32, error) { @@ -44,13 +45,18 @@ func getLedgerEntryAndLatestLedgerSequence(t require.TestingT, db *DB, key xdr.L return present, entry, latestSeq, expSeq } +func makeReadWriter(db *DB, batchSize, retentionWindow int) ReadWriter { + return NewReadWriter(log.DefaultLogger, db, interfaces.MakeNoOpDeamon(), + batchSize, uint32(retentionWindow), passphrase) +} + func TestGoldenPath(t *testing.T) { db := NewTestDB(t) // Check that we get an empty DB error _, err := NewLedgerEntryReader(db).GetLatestLedgerSequence(context.Background()) assert.Equal(t, ErrEmptyDB, err) - tx, err := NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err := makeReadWriter(db, 150, 15).NewTx(context.Background()) assert.NoError(t, err) writer := tx.LedgerEntryWriter() @@ -97,7 +103,7 @@ func TestGoldenPath(t *testing.T) { assert.Equal(t, ledgerSequence, obtainedLedgerSequence) // Do another round, overwriting the ledger entry - tx, err = NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err = makeReadWriter(db, 150, 15).NewTx(context.Background()) assert.NoError(t, err) writer = tx.LedgerEntryWriter() eight := xdr.Uint32(8) @@ -115,7 +121,7 @@ func TestGoldenPath(t *testing.T) { assert.Equal(t, eight, *obtainedEntry.Data.ContractData.Val.U32) // Do another round, deleting the ledger entry - tx, err = NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err = makeReadWriter(db, 150, 15).NewTx(context.Background()) assert.NoError(t, err) writer = tx.LedgerEntryWriter() assert.NoError(t, err) @@ -139,7 +145,7 @@ func TestDeleteNonExistentLedgerEmpty(t *testing.T) { // Simulate a ledger which creates and deletes a ledger entry // which would result in trying to delete a ledger entry which isn't there - tx, err := NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err := makeReadWriter(db, 150, 15).NewTx(context.Background()) assert.NoError(t, err) writer := tx.LedgerEntryWriter() @@ -212,7 +218,7 @@ func TestReadTxsDuringWriteTx(t *testing.T) { assert.Equal(t, ErrEmptyDB, err) // Start filling the DB with a single entry (enforce flushing right away) - tx, err := NewReadWriter(db, 0, 15).NewTx(context.Background()) + tx, err := makeReadWriter(db, 0, 15).NewTx(context.Background()) assert.NoError(t, err) writer := tx.LedgerEntryWriter() @@ -293,7 +299,7 @@ func TestWriteTxsDuringReadTxs(t *testing.T) { assert.NoError(t, err) // Start filling the DB with a single entry (enforce flushing right away) - tx, err := NewReadWriter(db, 0, 15).NewTx(context.Background()) + tx, err := makeReadWriter(db, 0, 15).NewTx(context.Background()) assert.NoError(t, err) writer := tx.LedgerEntryWriter() @@ -396,7 +402,7 @@ func TestConcurrentReadersAndWriter(t *testing.T) { }, } } - rw := NewReadWriter(db, 10, 15) + rw := makeReadWriter(db, 10, 15) for ledgerSequence := uint32(0); ledgerSequence < 1000; ledgerSequence++ { tx, err := rw.NewTx(context.Background()) assert.NoError(t, err) @@ -499,7 +505,7 @@ func benchmarkLedgerEntry(b *testing.B, cached bool) { }, } key, entry := getContractDataLedgerEntry(b, data) - tx, err := NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err := makeReadWriter(db, 150, 15).NewTx(context.Background()) assert.NoError(b, err) assert.NoError(b, tx.LedgerEntryWriter().UpsertLedgerEntry(entry)) expLedgerKey, err := entryKeyToTTLEntryKey(key) @@ -557,7 +563,7 @@ func BenchmarkLedgerUpdate(b *testing.B) { const numEntriesPerOp = 3500 b.ResetTimer() for i := 0; i < b.N; i++ { - tx, err := NewReadWriter(db, 150, 15).NewTx(context.Background()) + tx, err := makeReadWriter(db, 150, 15).NewTx(context.Background()) assert.NoError(b, err) writer := tx.LedgerEntryWriter() for j := 0; j < numEntriesPerOp; j++ { @@ -567,21 +573,3 @@ func BenchmarkLedgerUpdate(b *testing.B) { assert.NoError(b, tx.Commit(uint32(i+1))) } } - -func NewTestDB(tb testing.TB) *DB { - tmp := tb.TempDir() - dbPath := path.Join(tmp, "db.sqlite") - db, err := OpenSQLiteDB(dbPath) - if err != nil { - assert.NoError(tb, db.Close()) - } - tb.Cleanup(func() { - assert.NoError(tb, db.Close()) - }) - return &DB{ - SessionInterface: db, - cache: dbCache{ - ledgerEntries: newTransactionalCache(), - }, - } -} diff --git a/cmd/soroban-rpc/internal/db/migrations/02_transactions.sql b/cmd/soroban-rpc/internal/db/migrations/02_transactions.sql new file mode 100644 index 00000000..64720164 --- /dev/null +++ b/cmd/soroban-rpc/internal/db/migrations/02_transactions.sql @@ -0,0 +1,13 @@ +-- +migrate Up + +-- indexing table to find transactions in ledgers by hash +CREATE TABLE transactions ( + hash BLOB PRIMARY KEY, -- 32-byte binary + ledger_sequence INTEGER NOT NULL, + application_order INTEGER NOT NULL +); + +CREATE INDEX index_ledger_sequence ON transactions(ledger_sequence); + +-- +migrate Down +drop table transactions cascade; diff --git a/cmd/soroban-rpc/internal/db/mocks.go b/cmd/soroban-rpc/internal/db/mocks.go new file mode 100644 index 00000000..d2dfba4f --- /dev/null +++ b/cmd/soroban-rpc/internal/db/mocks.go @@ -0,0 +1,110 @@ +package db + +import ( + "context" + "io" + + "github.com/prometheus/client_golang/prometheus" + + "github.com/stellar/go/ingest" + "github.com/stellar/go/xdr" + + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" +) + +type mockTransactionHandler struct { + passphrase string + + ledgerRange ledgerbucketwindow.LedgerRange + txs map[string]ingest.LedgerTransaction + txHashToMeta map[string]*xdr.LedgerCloseMeta + ledgerSeqToMeta map[uint32]*xdr.LedgerCloseMeta +} + +func NewMockTransactionStore(passphrase string) *mockTransactionHandler { + return &mockTransactionHandler{ + passphrase: passphrase, + txs: make(map[string]ingest.LedgerTransaction), + txHashToMeta: make(map[string]*xdr.LedgerCloseMeta), + ledgerSeqToMeta: make(map[uint32]*xdr.LedgerCloseMeta), + } +} + +func (txn *mockTransactionHandler) InsertTransactions(lcm xdr.LedgerCloseMeta) error { + txn.ledgerSeqToMeta[lcm.LedgerSequence()] = &lcm + + reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(txn.passphrase, lcm) + if err != nil { + return err + } + + for { + tx, err := reader.Read() + if err == io.EOF { + break + } else if err != nil { + return err + } + + h := tx.Result.TransactionHash.HexString() + txn.txs[h] = tx + txn.txHashToMeta[h] = &lcm + } + + if lcmSeq := lcm.LedgerSequence(); lcmSeq < txn.ledgerRange.FirstLedger.Sequence || + txn.ledgerRange.FirstLedger.Sequence == 0 { + txn.ledgerRange.FirstLedger.Sequence = lcmSeq + txn.ledgerRange.FirstLedger.CloseTime = lcm.LedgerCloseTime() + } + + if lcmSeq := lcm.LedgerSequence(); lcmSeq > txn.ledgerRange.LastLedger.Sequence { + txn.ledgerRange.LastLedger.Sequence = lcmSeq + txn.ledgerRange.LastLedger.CloseTime = lcm.LedgerCloseTime() + } + + return nil +} + +// GetLedgerRange pulls the min/max ledger sequence numbers from the database. +func (txn *mockTransactionHandler) GetLedgerRange(ctx context.Context) (ledgerbucketwindow.LedgerRange, error) { + return txn.ledgerRange, nil +} + +func (txn *mockTransactionHandler) GetTransaction(ctx context.Context, hash xdr.Hash) ( + Transaction, ledgerbucketwindow.LedgerRange, error, +) { + if tx, ok := txn.txs[hash.HexString()]; !ok { + return Transaction{}, txn.ledgerRange, ErrNoTransaction + } else { + itx, err := ParseTransaction(*txn.txHashToMeta[hash.HexString()], tx) + return itx, txn.ledgerRange, err + } +} + +func (txn *mockTransactionHandler) RegisterMetrics(_, _ prometheus.Observer) {} + +type mockLedgerReader struct { + txn mockTransactionHandler +} + +func NewMockLedgerReader(txn *mockTransactionHandler) *mockLedgerReader { + return &mockLedgerReader{ + txn: *txn, + } +} + +func (m *mockLedgerReader) GetLedger(ctx context.Context, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { + lcm, ok := m.txn.ledgerSeqToMeta[sequence] + if !ok { + return xdr.LedgerCloseMeta{}, false, nil + } + return *lcm, true, nil +} + +func (m *mockLedgerReader) StreamAllLedgers(ctx context.Context, f StreamLedgerFn) error { + return nil +} + +var _ TransactionReader = &mockTransactionHandler{} +var _ TransactionWriter = &mockTransactionHandler{} +var _ LedgerReader = &mockLedgerReader{} diff --git a/cmd/soroban-rpc/internal/db/transaction.go b/cmd/soroban-rpc/internal/db/transaction.go new file mode 100644 index 00000000..c703c39e --- /dev/null +++ b/cmd/soroban-rpc/internal/db/transaction.go @@ -0,0 +1,304 @@ +package db + +import ( + "context" + "encoding/hex" + "fmt" + "time" + + sq "github.com/Masterminds/squirrel" + "github.com/prometheus/client_golang/prometheus" + + "github.com/stellar/go/ingest" + "github.com/stellar/go/support/db" + "github.com/stellar/go/support/errors" + "github.com/stellar/go/support/log" + "github.com/stellar/go/xdr" + + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" +) + +const transactionTableName = "transactions" + +var ErrNoTransaction = errors.New("no transaction with this hash exists") + +type Transaction struct { + Result []byte // XDR encoded xdr.TransactionResult + Meta []byte // XDR encoded xdr.TransactionMeta + Envelope []byte // XDR encoded xdr.TransactionEnvelope + Events [][]byte // XDR encoded xdr.DiagnosticEvent + FeeBump bool + ApplicationOrder int32 + Successful bool + Ledger ledgerbucketwindow.LedgerInfo +} + +// TransactionWriter is used during ingestion to write LCM. +type TransactionWriter interface { + InsertTransactions(lcm xdr.LedgerCloseMeta) error + RegisterMetrics(ingest, count prometheus.Observer) +} + +// TransactionReader provides all of the public ways to read from the DB. +type TransactionReader interface { + GetTransaction(ctx context.Context, hash xdr.Hash) (Transaction, ledgerbucketwindow.LedgerRange, error) + GetLedgerRange(ctx context.Context) (ledgerbucketwindow.LedgerRange, error) +} + +type transactionHandler struct { + log *log.Entry + db db.SessionInterface + stmtCache *sq.StmtCache + passphrase string + + ingestMetric, countMetric prometheus.Observer +} + +func NewTransactionReader(log *log.Entry, db db.SessionInterface, passphrase string) TransactionReader { + return &transactionHandler{log: log, db: db, passphrase: passphrase} +} + +func (txn *transactionHandler) InsertTransactions(lcm xdr.LedgerCloseMeta) error { + start := time.Now() + txCount := lcm.CountTransactions() + L := txn.log. + WithField("ledger_seq", lcm.LedgerSequence()). + WithField("tx_count", txCount) + + defer func() { + if txn.ingestMetric != nil { + txn.ingestMetric.Observe(time.Since(start).Seconds()) + txn.countMetric.Observe(float64(txCount)) + } + }() + + if txn.stmtCache == nil { + return errors.New("TransactionWriter incorrectly initialized without stmtCache") + } else if txCount == 0 { + return nil + } + + reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(txn.passphrase, lcm) + if err != nil { + return errors.Wrapf(err, + "failed to open transaction reader for ledger %d", + lcm.LedgerSequence()) + } + + transactions := make(map[xdr.Hash]ingest.LedgerTransaction, txCount) + for i := 0; i < txCount; i++ { + tx, err := reader.Read() + if err != nil { + return errors.Wrapf(err, "failed reading tx %d", i) + } + + // For fee-bump transactions, we store lookup entries for both the outer + // and inner hashes. + if tx.Envelope.IsFeeBump() { + transactions[tx.Result.InnerHash()] = tx + } + transactions[tx.Result.TransactionHash] = tx + } + + query := sq.Insert(transactionTableName). + Columns("hash", "ledger_sequence", "application_order") + for hash, tx := range transactions { + query = query.Values(hash[:], lcm.LedgerSequence(), tx.Index) + } + _, err = query.RunWith(txn.stmtCache).Exec() + + L.WithError(err). + WithField("duration", time.Since(start)). + Infof("Ingested %d transaction lookups", len(transactions)) + + return err +} + +func (txn *transactionHandler) RegisterMetrics(ingest, count prometheus.Observer) { + txn.ingestMetric = ingest + txn.countMetric = count +} + +// trimTransactions removes all transactions which fall outside the ledger retention window. +func (txn *transactionHandler) trimTransactions(latestLedgerSeq uint32, retentionWindow uint32) error { + if latestLedgerSeq+1 <= retentionWindow { + return nil + } + + cutoff := latestLedgerSeq + 1 - retentionWindow + _, err := sq.StatementBuilder. + RunWith(txn.stmtCache). + Delete(transactionTableName). + Where(sq.Lt{"ledger_sequence": cutoff}). + Exec() + return err +} + +// GetLedgerRange pulls the min/max ledger sequence numbers from the database. +func (txn *transactionHandler) GetLedgerRange(ctx context.Context) (ledgerbucketwindow.LedgerRange, error) { + var ledgerRange ledgerbucketwindow.LedgerRange + + // + // We use subqueries alongside a UNION ALL stitch in order to select the min + // and max from the ledger table in a single query and get around sqlite's + // limitations with parentheses (see https://stackoverflow.com/a/22609948). + // + newestQ := sq. + Select("m1.meta"). + FromSelect( + sq. + Select("meta"). + From(ledgerCloseMetaTableName). + OrderBy("sequence ASC"). + Limit(1), + "m1", + ) + sql, args, err := sq. + Select("m2.meta"). + FromSelect( + sq. + Select("meta"). + From(ledgerCloseMetaTableName). + OrderBy("sequence DESC"). + Limit(1), + "m2", + ).ToSql() + if err != nil { + return ledgerRange, errors.Wrap(err, "couldn't build ledger range query") + } + + var lcms []xdr.LedgerCloseMeta + if err = txn.db.Select(ctx, &lcms, newestQ.Suffix("UNION ALL "+sql, args...)); err != nil { + return ledgerRange, errors.Wrap(err, "couldn't query ledger range") + } else if len(lcms) < 2 { + // There is almost certainly a row, but we want to avoid a race condition + // with ingestion as well as support test cases from an empty DB, so we need + // to sanity check that there is in fact a result. Note that no ledgers in + // the database isn't an error, it's just an empty range. + return ledgerRange, nil + } + + lcm1, lcm2 := lcms[0], lcms[1] + ledgerRange.FirstLedger.Sequence = lcm1.LedgerSequence() + ledgerRange.FirstLedger.CloseTime = lcm1.LedgerCloseTime() + ledgerRange.LastLedger.Sequence = lcm2.LedgerSequence() + ledgerRange.LastLedger.CloseTime = lcm2.LedgerCloseTime() + + txn.log.Debugf("Database ledger range: [%d, %d]", + ledgerRange.FirstLedger.Sequence, ledgerRange.LastLedger.Sequence) + return ledgerRange, nil +} + +// GetTransaction conforms to the interface in +// methods/get_transaction.go#NewGetTransactionHandler so that it can be used +// directly against the RPC handler. +// +// Errors occur if there are issues with the DB connection or the XDR is +// corrupted somehow. If the transaction is not found, io.EOF is returned. +func (txn *transactionHandler) GetTransaction(ctx context.Context, hash xdr.Hash) ( + Transaction, ledgerbucketwindow.LedgerRange, error, +) { + start := time.Now() + tx := Transaction{} + + ledgerRange, err := txn.GetLedgerRange(ctx) + if err != nil && err != ErrEmptyDB { + return tx, ledgerRange, err + } + + lcm, ingestTx, err := txn.getTransactionByHash(ctx, hash) + if err != nil { + return tx, ledgerRange, err + } + tx, err = ParseTransaction(lcm, ingestTx) + if err != nil { + return tx, ledgerRange, err + } + + txn.log. + WithField("txhash", hex.EncodeToString(hash[:])). + WithField("duration", time.Since(start)). + Debugf("Fetched and encoded transaction from ledger %d", lcm.LedgerSequence()) + + return tx, ledgerRange, nil +} + +// getTransactionByHash actually performs the DB ops to cross-reference a +// transaction hash with a particular set of ledger close meta and parses out +// the relevant transaction efficiently by leveraging the `application_order` db +// field. +// +// Note: Caller must do input sanitization on the hash. +func (txn *transactionHandler) getTransactionByHash(ctx context.Context, hash xdr.Hash) ( + xdr.LedgerCloseMeta, ingest.LedgerTransaction, error, +) { + var rows []struct { + TxIndex int `db:"application_order"` + Lcm xdr.LedgerCloseMeta `db:"meta"` + } + rowQ := sq. + Select("t.application_order", "lcm.meta"). + From(fmt.Sprintf("%s t", transactionTableName)). + Join(fmt.Sprintf("%s lcm ON (t.ledger_sequence = lcm.sequence)", ledgerCloseMetaTableName)). + Where(sq.Eq{"t.hash": []byte(hash[:])}). + Limit(1) + + if err := txn.db.Select(ctx, &rows, rowQ); err != nil { + return xdr.LedgerCloseMeta{}, ingest.LedgerTransaction{}, + errors.Wrapf(err, "db read failed for txhash %s", hex.EncodeToString(hash[:])) + } else if len(rows) < 1 { + return xdr.LedgerCloseMeta{}, ingest.LedgerTransaction{}, ErrNoTransaction + } + + txIndex, lcm := rows[0].TxIndex, rows[0].Lcm + reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(txn.passphrase, lcm) + reader.Seek(txIndex - 1) + if err != nil { + return lcm, ingest.LedgerTransaction{}, + errors.Wrapf(err, "failed to index to tx %d in ledger %d (txhash=%s)", + txIndex, lcm.LedgerSequence(), hash) + } + + ledgerTx, err := reader.Read() + return lcm, ledgerTx, err +} + +func ParseTransaction(lcm xdr.LedgerCloseMeta, ingestTx ingest.LedgerTransaction) (Transaction, error) { + var tx Transaction + var err error + + // + // On-the-fly ingestion: extract all of the fields, return best effort. + // + tx.FeeBump = ingestTx.Envelope.IsFeeBump() + tx.ApplicationOrder = int32(ingestTx.Index) + tx.Successful = ingestTx.Result.Successful() + tx.Ledger = ledgerbucketwindow.LedgerInfo{ + Sequence: lcm.LedgerSequence(), + CloseTime: lcm.LedgerCloseTime(), + } + + if tx.Result, err = ingestTx.Result.Result.MarshalBinary(); err != nil { + return tx, errors.Wrap(err, "couldn't encode transaction Result") + } + if tx.Meta, err = ingestTx.UnsafeMeta.MarshalBinary(); err != nil { + return tx, errors.Wrap(err, "couldn't encode transaction UnsafeMeta") + } + if tx.Envelope, err = ingestTx.Envelope.MarshalBinary(); err != nil { + return tx, errors.Wrap(err, "couldn't encode transaction Envelope") + } + if events, diagErr := ingestTx.GetDiagnosticEvents(); diagErr == nil { + tx.Events = make([][]byte, 0, len(events)) + for i, event := range events { + bytes, ierr := event.MarshalBinary() + if ierr != nil { + return tx, errors.Wrapf(ierr, "couldn't encode transaction DiagnosticEvent %d", i) + } + tx.Events = append(tx.Events, bytes) + } + } else { + return tx, errors.Wrap(diagErr, "couldn't encode transaction DiagnosticEvents") + } + + return tx, nil +} diff --git a/cmd/soroban-rpc/internal/db/transaction_test.go b/cmd/soroban-rpc/internal/db/transaction_test.go new file mode 100644 index 00000000..068f793d --- /dev/null +++ b/cmd/soroban-rpc/internal/db/transaction_test.go @@ -0,0 +1,205 @@ +package db + +import ( + "context" + "encoding/hex" + "math/rand" + "testing" + + "github.com/sirupsen/logrus" + "github.com/stellar/go/network" + "github.com/stellar/go/support/log" + "github.com/stellar/go/xdr" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestTransactionNotFound(t *testing.T) { + db := NewTestDB(t) + log := log.DefaultLogger + log.SetLevel(logrus.TraceLevel) + + reader := NewTransactionReader(log, db, passphrase) + _, _, err := reader.GetTransaction(context.TODO(), xdr.Hash{}) + require.Error(t, err, ErrNoTransaction) +} + +func TestTransactionFound(t *testing.T) { + db := NewTestDB(t) + ctx := context.TODO() + log := log.DefaultLogger + log.SetLevel(logrus.TraceLevel) + + writer := NewReadWriter(log, db, interfaces.MakeNoOpDeamon(), 10, 10, passphrase) + write, err := writer.NewTx(ctx) + require.NoError(t, err) + + lcms := []xdr.LedgerCloseMeta{ + txMeta(1234, true), + txMeta(1235, true), + txMeta(1236, true), + txMeta(1237, true), + } + + ledgerW, txW := write.LedgerWriter(), write.TransactionWriter() + for _, lcm := range lcms { + require.NoError(t, ledgerW.InsertLedger(lcm), "ingestion failed for ledger %+v", lcm.V1) + require.NoError(t, txW.InsertTransactions(lcm), "ingestion failed for ledger %+v", lcm.V1) + } + require.NoError(t, write.Commit(lcms[len(lcms)-1].LedgerSequence())) + + // check 404 case + reader := NewTransactionReader(log, db, passphrase) + _, _, err = reader.GetTransaction(ctx, xdr.Hash{}) + require.Error(t, err, ErrNoTransaction) + + // check all 200 cases + for _, lcm := range lcms { + h := lcm.TransactionHash(0) + tx, lRange, err := reader.GetTransaction(ctx, h) + require.NoError(t, err, "failed to find txhash %s in db", hex.EncodeToString(h[:])) + assert.EqualValues(t, 1234+100, lRange.FirstLedger.Sequence) + assert.EqualValues(t, 1237+100, lRange.LastLedger.Sequence) + assert.EqualValues(t, 1, tx.ApplicationOrder) + + expectedEnvelope, err := lcm.TransactionEnvelopes()[0].MarshalBinary() + require.NoError(t, err) + assert.Equal(t, expectedEnvelope, tx.Envelope) + } +} + +func BenchmarkTransactionFetch(b *testing.B) { + db := NewTestDB(b) + ctx := context.TODO() + log := log.DefaultLogger + + writer := NewReadWriter(log, db, interfaces.MakeNoOpDeamon(), 100, 1_000_000, passphrase) + write, err := writer.NewTx(ctx) + require.NoError(b, err) + + // ingest 100k tx rows + lcms := make([]xdr.LedgerCloseMeta, 0, 100_000) + for i := uint32(0); i < uint32(cap(lcms)); i++ { + lcms = append(lcms, txMeta(1234+i, i%2 == 0)) + } + + ledgerW, txW := write.LedgerWriter(), write.TransactionWriter() + for _, lcm := range lcms { + require.NoError(b, ledgerW.InsertLedger(lcm)) + require.NoError(b, txW.InsertTransactions(lcm)) + } + require.NoError(b, write.Commit(lcms[len(lcms)-1].LedgerSequence())) + reader := NewTransactionReader(log, db, passphrase) + + randoms := make([]int, b.N) + for i := 0; i < b.N; i++ { + randoms[i] = rand.Intn(len(lcms)) + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + r := randoms[i] + tx, _, err := reader.GetTransaction(ctx, lcms[r].TransactionHash(0)) + require.NoError(b, err) + assert.Equal(b, r%2 == 0, tx.Successful) + } +} + +// +// Structure creation methods below. +// + +func txHash(acctSeq uint32) xdr.Hash { + envelope := txEnvelope(acctSeq) + hash, err := network.HashTransactionInEnvelope(envelope, passphrase) + if err != nil { + panic(err) + } + return hash +} + +func txEnvelope(acctSeq uint32) xdr.TransactionEnvelope { + envelope, err := xdr.NewTransactionEnvelope(xdr.EnvelopeTypeEnvelopeTypeTx, xdr.TransactionV1Envelope{ + Tx: xdr.Transaction{ + Fee: 1, + SeqNum: xdr.SequenceNumber(acctSeq), + SourceAccount: xdr.MustMuxedAddress("MA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVAAAAAAAAAAAAAJLK"), + }, + }) + if err != nil { + panic(err) + } + return envelope +} + +func transactionResult(successful bool) xdr.TransactionResult { + code := xdr.TransactionResultCodeTxBadSeq + if successful { + code = xdr.TransactionResultCodeTxSuccess + } + opResults := []xdr.OperationResult{} + return xdr.TransactionResult{ + FeeCharged: 100, + Result: xdr.TransactionResultResult{ + Code: code, + Results: &opResults, + }, + } +} + +func txMeta(acctSeq uint32, successful bool) xdr.LedgerCloseMeta { + envelope := txEnvelope(acctSeq) + txProcessing := []xdr.TransactionResultMeta{ + { + TxApplyProcessing: xdr.TransactionMeta{ + V: 3, + Operations: &[]xdr.OperationMeta{}, + V3: &xdr.TransactionMetaV3{}, + }, + Result: xdr.TransactionResultPair{ + TransactionHash: txHash(acctSeq), + Result: transactionResult(successful), + }, + }, + } + components := []xdr.TxSetComponent{ + { + Type: xdr.TxSetComponentTypeTxsetCompTxsMaybeDiscountedFee, + TxsMaybeDiscountedFee: &xdr.TxSetComponentTxsMaybeDiscountedFee{ + BaseFee: nil, + Txs: []xdr.TransactionEnvelope{envelope}, + }, + }, + } + + return xdr.LedgerCloseMeta{ + V: 1, + V1: &xdr.LedgerCloseMetaV1{ + LedgerHeader: xdr.LedgerHeaderHistoryEntry{ + Header: xdr.LedgerHeader{ + ScpValue: xdr.StellarValue{ + CloseTime: xdr.TimePoint(ledgerCloseTime(acctSeq + 100)), + }, + LedgerSeq: xdr.Uint32(acctSeq + 100), + }, + }, + TxProcessing: txProcessing, + TxSet: xdr.GeneralizedTransactionSet{ + V: 1, + V1TxSet: &xdr.TransactionSetV1{ + PreviousLedgerHash: xdr.Hash{1}, + Phases: []xdr.TransactionPhase{{ + V: 0, + V0Components: &components, + }}, + }, + }, + }, + } +} + +func ledgerCloseTime(ledgerSequence uint32) int64 { + return int64(ledgerSequence)*25 + 100 +} diff --git a/cmd/soroban-rpc/internal/events/events.go b/cmd/soroban-rpc/internal/events/events.go index fcea2a2b..b724fbf9 100644 --- a/cmd/soroban-rpc/internal/events/events.go +++ b/cmd/soroban-rpc/internal/events/events.go @@ -202,7 +202,7 @@ func (m *MemoryStore) IngestEvents(ledgerCloseMeta xdr.LedgerCloseMeta) error { } bucket := ledgerbucketwindow.LedgerBucket[[]event]{ LedgerSeq: ledgerCloseMeta.LedgerSequence(), - LedgerCloseTimestamp: int64(ledgerCloseMeta.LedgerHeaderHistoryEntry().Header.ScpValue.CloseTime), + LedgerCloseTimestamp: ledgerCloseMeta.LedgerCloseTime(), BucketContent: events, } m.lock.Lock() @@ -267,8 +267,8 @@ func readEvents(networkPassphrase string, ledgerCloseMeta xdr.LedgerCloseMeta) ( } // GetLedgerRange returns the first and latest ledger available in the store. -func (m *MemoryStore) GetLedgerRange() ledgerbucketwindow.LedgerRange { +func (m *MemoryStore) GetLedgerRange() (ledgerbucketwindow.LedgerRange, error) { m.lock.RLock() defer m.lock.RUnlock() - return m.eventsByLedger.GetLedgerRange() + return m.eventsByLedger.GetLedgerRange(), nil } diff --git a/cmd/soroban-rpc/internal/ingest/mock_db_test.go b/cmd/soroban-rpc/internal/ingest/mock_db_test.go index cb883eac..6e57658d 100644 --- a/cmd/soroban-rpc/internal/ingest/mock_db_test.go +++ b/cmd/soroban-rpc/internal/ingest/mock_db_test.go @@ -3,9 +3,11 @@ package ingest import ( "context" - "github.com/stellar/go/xdr" + "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/mock" + "github.com/stellar/go/xdr" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" ) @@ -44,6 +46,11 @@ func (m MockTx) LedgerWriter() db.LedgerWriter { return args.Get(0).(db.LedgerWriter) } +func (m MockTx) TransactionWriter() db.TransactionWriter { + args := m.Called() + return args.Get(0).(db.TransactionWriter) +} + func (m MockTx) Commit(ledgerSeq uint32) error { args := m.Called(ledgerSeq) return args.Error(0) @@ -76,3 +83,16 @@ func (m MockLedgerWriter) InsertLedger(ledger xdr.LedgerCloseMeta) error { args := m.Called(ledger) return args.Error(0) } + +type MockTransactionWriter struct { + mock.Mock +} + +func (m MockTransactionWriter) InsertTransactions(ledger xdr.LedgerCloseMeta) error { + args := m.Called(ledger) + return args.Error(0) +} + +func (m MockTransactionWriter) RegisterMetrics(ingest, count prometheus.Observer) { + m.Called(ingest, count) +} diff --git a/cmd/soroban-rpc/internal/ingest/service.go b/cmd/soroban-rpc/internal/ingest/service.go index 41116a59..54ccbef4 100644 --- a/cmd/soroban-rpc/internal/ingest/service.go +++ b/cmd/soroban-rpc/internal/ingest/service.go @@ -22,7 +22,6 @@ import ( "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/util" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/events" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/transactions" ) const ( @@ -35,7 +34,6 @@ type Config struct { Logger *log.Entry DB db.ReadWriter EventStore *events.MemoryStore - TransactionStore *transactions.MemoryStore FeeWindows *feewindow.FeeWindows NetworkPassPhrase string Archive historyarchive.ArchiveInterface @@ -84,7 +82,6 @@ func newService(cfg Config) *Service { logger: cfg.Logger, db: cfg.DB, eventStore: cfg.EventStore, - transactionStore: cfg.TransactionStore, feeWindows: cfg.FeeWindows, ledgerBackend: cfg.LedgerBackend, networkPassPhrase: cfg.NetworkPassPhrase, @@ -137,7 +134,6 @@ type Service struct { logger *log.Entry db db.ReadWriter eventStore *events.MemoryStore - transactionStore *transactions.MemoryStore feeWindows *feewindow.FeeWindows ledgerBackend backends.LedgerBackend timeout time.Duration @@ -202,10 +198,9 @@ func (s *Service) maybeFillEntriesFromCheckpoint(ctx context.Context, archive hi nextLedgerSeq := curLedgerSeq + 1 prepareRangeCtx, cancelPrepareRange := context.WithTimeout(ctx, s.timeout) defer cancelPrepareRange() - if err = s.ledgerBackend.PrepareRange(prepareRangeCtx, backends.UnboundedRange(nextLedgerSeq)); err != nil { - return nextLedgerSeq, checkPointFillErr, err - } - return nextLedgerSeq, checkPointFillErr, nil + return nextLedgerSeq, + checkPointFillErr, + s.ledgerBackend.PrepareRange(prepareRangeCtx, backends.UnboundedRange(nextLedgerSeq)) } } @@ -292,6 +287,7 @@ func (s *Service) ingest(ctx context.Context, sequence uint32) error { if err := s.ingestLedgerEntryChanges(ctx, reader, tx, 0); err != nil { return err } + if err := reader.Close(); err != nil { return err } @@ -313,10 +309,13 @@ func (s *Service) ingest(ctx context.Context, sequence uint32) error { if err := tx.Commit(sequence); err != nil { return err } - s.logger.Debugf("Ingested ledger %d", sequence) + s.logger. + WithField("duration", time.Since(startTime).Seconds()). + Debugf("Ingested ledger %d", sequence) s.metrics.ingestionDurationMetric. - With(prometheus.Labels{"type": "total"}).Observe(time.Since(startTime).Seconds()) + With(prometheus.Labels{"type": "total"}). + Observe(time.Since(startTime).Seconds()) s.metrics.latestLedgerMetric.Set(float64(sequence)) return nil } @@ -327,13 +326,18 @@ func (s *Service) ingestLedgerCloseMeta(tx db.WriteTx, ledgerCloseMeta xdr.Ledge return err } s.metrics.ingestionDurationMetric. - With(prometheus.Labels{"type": "ledger_close_meta"}).Observe(time.Since(startTime).Seconds()) + With(prometheus.Labels{"type": "ledger_close_meta"}). + Observe(time.Since(startTime).Seconds()) - if err := s.eventStore.IngestEvents(ledgerCloseMeta); err != nil { + startTime = time.Now() + if err := tx.TransactionWriter().InsertTransactions(ledgerCloseMeta); err != nil { return err } + s.metrics.ingestionDurationMetric. + With(prometheus.Labels{"type": "transactions"}). + Observe(time.Since(startTime).Seconds()) - if err := s.transactionStore.IngestTransactions(ledgerCloseMeta); err != nil { + if err := s.eventStore.IngestEvents(ledgerCloseMeta); err != nil { return err } diff --git a/cmd/soroban-rpc/internal/ingest/service_test.go b/cmd/soroban-rpc/internal/ingest/service_test.go index d5b86879..78f42494 100644 --- a/cmd/soroban-rpc/internal/ingest/service_test.go +++ b/cmd/soroban-rpc/internal/ingest/service_test.go @@ -18,7 +18,6 @@ import ( "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/events" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/feewindow" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/transactions" ) type ErrorReadWriter struct { @@ -47,7 +46,6 @@ func TestRetryRunningIngestion(t *testing.T) { Logger: supportlog.New(), DB: &ErrorReadWriter{}, EventStore: nil, - TransactionStore: nil, NetworkPassPhrase: "", Archive: nil, LedgerBackend: nil, @@ -72,7 +70,6 @@ func TestIngestion(t *testing.T) { Logger: supportlog.New(), DB: mockDB, EventStore: events.NewMemoryStore(daemon, network.TestNetworkPassphrase, 1), - TransactionStore: transactions.NewMemoryStore(daemon, network.TestNetworkPassphrase, 1), FeeWindows: feewindow.NewFeeWindows(1, 1, network.TestNetworkPassphrase), LedgerBackend: mockLedgerBackend, Daemon: daemon, @@ -83,12 +80,14 @@ func TestIngestion(t *testing.T) { mockTx := &MockTx{} mockLedgerEntryWriter := &MockLedgerEntryWriter{} mockLedgerWriter := &MockLedgerWriter{} + mockTxWriter := &MockTransactionWriter{} ctx := context.Background() mockDB.On("NewTx", ctx).Return(mockTx, nil).Once() mockTx.On("Commit", sequence).Return(nil).Once() mockTx.On("Rollback").Return(nil).Once() mockTx.On("LedgerEntryWriter").Return(mockLedgerEntryWriter).Twice() mockTx.On("LedgerWriter").Return(mockLedgerWriter).Once() + mockTx.On("TransactionWriter").Return(mockTxWriter).Once() src := xdr.MustAddress("GBXGQJWVLWOYHFLVTKWV5FGHA3LNYY2JQKM7OAJAUEQFU6LPCSEFVXON") firstTx := xdr.TransactionEnvelope{ @@ -244,8 +243,7 @@ func TestIngestion(t *testing.T) { EvictedPersistentLedgerEntries: []xdr.LedgerEntry{evictedPersistentLedgerEntry}, }, } - mockLedgerBackend.On("GetLedger", ctx, sequence). - Return(ledger, nil).Once() + mockLedgerBackend.On("GetLedger", ctx, sequence).Return(ledger, nil).Once() mockLedgerEntryWriter.On("UpsertLedgerEntry", operationChanges[1].MustUpdated()). Return(nil).Once() evictedPresistentLedgerKey, err := evictedPersistentLedgerEntry.LedgerKey() @@ -254,8 +252,8 @@ func TestIngestion(t *testing.T) { Return(nil).Once() mockLedgerEntryWriter.On("DeleteLedgerEntry", evictedTempLedgerKey). Return(nil).Once() - mockLedgerWriter.On("InsertLedger", ledger). - Return(nil).Once() + mockLedgerWriter.On("InsertLedger", ledger).Return(nil).Once() + mockTxWriter.On("InsertTransactions", ledger).Return(nil).Once() assert.NoError(t, service.ingest(ctx, sequence)) mockDB.AssertExpectations(t) diff --git a/cmd/soroban-rpc/internal/jsonrpc.go b/cmd/soroban-rpc/internal/jsonrpc.go index 9151d629..014a75a1 100644 --- a/cmd/soroban-rpc/internal/jsonrpc.go +++ b/cmd/soroban-rpc/internal/jsonrpc.go @@ -23,7 +23,6 @@ import ( "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/feewindow" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/methods" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/network" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/transactions" ) // maxHTTPRequestSize defines the largest request size that the http handler @@ -48,8 +47,8 @@ func (h Handler) Close() { type HandlerParams struct { EventStore *events.MemoryStore - TransactionStore *transactions.MemoryStore FeeStatWindows *feewindow.FeeWindows + TransactionReader db.TransactionReader LedgerEntryReader db.LedgerEntryReader LedgerReader db.LedgerReader Logger *log.Entry @@ -137,13 +136,10 @@ func NewJSONRPCHandler(cfg *config.Config, params HandlerParams) Handler { }, } - // Get the largest history window - var ledgerRangeGetter methods.LedgerRangeGetter = params.EventStore - var retentionWindow = cfg.EventLedgerRetentionWindow - if cfg.TransactionLedgerRetentionWindow > cfg.EventLedgerRetentionWindow { - retentionWindow = cfg.TransactionLedgerRetentionWindow - ledgerRangeGetter = params.TransactionStore - } + // While we transition from in-memory to database-oriented history storage, + // the on-disk (transaction) retention window will always be larger than the + // in-memory (events) one. + var retentionWindow = cfg.TransactionLedgerRetentionWindow handlers := []struct { methodName string @@ -153,15 +149,17 @@ func NewJSONRPCHandler(cfg *config.Config, params HandlerParams) Handler { requestDurationLimit time.Duration }{ { - methodName: "getHealth", - underlyingHandler: methods.NewHealthCheck(retentionWindow, ledgerRangeGetter, cfg.MaxHealthyLedgerLatency), + methodName: "getHealth", + underlyingHandler: methods.NewHealthCheck( + retentionWindow, params.TransactionReader, cfg.MaxHealthyLedgerLatency), longName: "get_health", queueLimit: cfg.RequestBacklogGetHealthQueueLimit, requestDurationLimit: cfg.MaxGetHealthExecutionDuration, }, { - methodName: "getEvents", - underlyingHandler: methods.NewGetEventsHandler(params.EventStore, cfg.MaxEventsLimit, cfg.DefaultEventsLimit), + methodName: "getEvents", + underlyingHandler: methods.NewGetEventsHandler( + params.EventStore, cfg.MaxEventsLimit, cfg.DefaultEventsLimit), longName: "get_events", queueLimit: cfg.RequestBacklogGetEventsQueueLimit, requestDurationLimit: cfg.MaxGetEventsExecutionDuration, @@ -203,28 +201,38 @@ func NewJSONRPCHandler(cfg *config.Config, params HandlerParams) Handler { }, { methodName: "getTransaction", - underlyingHandler: methods.NewGetTransactionHandler(params.TransactionStore), + underlyingHandler: methods.NewGetTransactionHandler(params.Logger, params.TransactionReader), longName: "get_transaction", queueLimit: cfg.RequestBacklogGetTransactionQueueLimit, requestDurationLimit: cfg.MaxGetTransactionExecutionDuration, }, { - methodName: "sendTransaction", - underlyingHandler: methods.NewSendTransactionHandler(params.Daemon, params.Logger, params.TransactionStore, cfg.NetworkPassphrase), + methodName: "getTransactions", + underlyingHandler: methods.NewGetTransactionsHandler(params.Logger, params.LedgerReader, params.TransactionReader, cfg.MaxTransactionsLimit, cfg.DefaultTransactionsLimit, cfg.NetworkPassphrase), + longName: "get_transactions", + queueLimit: cfg.RequestBacklogGetTransactionsQueueLimit, + requestDurationLimit: cfg.MaxGetTransactionsExecutionDuration, + }, + { + methodName: "sendTransaction", + underlyingHandler: methods.NewSendTransactionHandler( + params.Daemon, params.Logger, params.TransactionReader, cfg.NetworkPassphrase), longName: "send_transaction", queueLimit: cfg.RequestBacklogSendTransactionQueueLimit, requestDurationLimit: cfg.MaxSendTransactionExecutionDuration, }, { - methodName: "simulateTransaction", - underlyingHandler: methods.NewSimulateTransactionHandler(params.Logger, params.LedgerEntryReader, params.LedgerReader, params.Daemon, params.PreflightGetter), + methodName: "simulateTransaction", + underlyingHandler: methods.NewSimulateTransactionHandler( + params.Logger, params.LedgerEntryReader, params.LedgerReader, + params.Daemon, params.PreflightGetter), longName: "simulate_transaction", queueLimit: cfg.RequestBacklogSimulateTransactionQueueLimit, requestDurationLimit: cfg.MaxSimulateTransactionExecutionDuration, }, { methodName: "getFeeStats", - underlyingHandler: methods.NewGetFeeStatsHandler(params.FeeStatWindows, ledgerRangeGetter), + underlyingHandler: methods.NewGetFeeStatsHandler(params.FeeStatWindows, params.TransactionReader, params.Logger), longName: "get_fee_stats", queueLimit: cfg.RequestBacklogGetFeeStatsTransactionQueueLimit, requestDurationLimit: cfg.MaxGetFeeStatsExecutionDuration, diff --git a/cmd/soroban-rpc/internal/ledgerbucketwindow/ledgerbucketwindow.go b/cmd/soroban-rpc/internal/ledgerbucketwindow/ledgerbucketwindow.go index 7225a6b3..ee91427c 100644 --- a/cmd/soroban-rpc/internal/ledgerbucketwindow/ledgerbucketwindow.go +++ b/cmd/soroban-rpc/internal/ledgerbucketwindow/ledgerbucketwindow.go @@ -20,9 +20,12 @@ type LedgerBucket[T any] struct { BucketContent T } -// DefaultEventLedgerRetentionWindow represents the max number of ledgers we would like to keep -// an incoming event in memory. The value was calculated to align with (roughly) 24 hours window. -const DefaultEventLedgerRetentionWindow = 17280 +// OneDayOfLedgers is (roughly) a 24 hour window of ledgers. +const OneDayOfLedgers = 17280 + +// DefaultEventLedgerRetentionWindow represents the max number of ledgers we +// would like to keep an incoming event in memory. +const DefaultEventLedgerRetentionWindow = OneDayOfLedgers // NewLedgerBucketWindow creates a new LedgerBucketWindow func NewLedgerBucketWindow[T any](retentionWindow uint32) *LedgerBucketWindow[T] { diff --git a/cmd/soroban-rpc/internal/methods/get_fee_stats.go b/cmd/soroban-rpc/internal/methods/get_fee_stats.go index e1f1182b..0e02896a 100644 --- a/cmd/soroban-rpc/internal/methods/get_fee_stats.go +++ b/cmd/soroban-rpc/internal/methods/get_fee_stats.go @@ -4,7 +4,9 @@ import ( "context" "github.com/creachadair/jrpc2" + "github.com/stellar/go/support/log" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/feewindow" ) @@ -56,12 +58,18 @@ type GetFeeStatsResult struct { } // NewGetFeeStatsHandler returns a handler obtaining fee statistics -func NewGetFeeStatsHandler(windows *feewindow.FeeWindows, ledgerRangeGetter LedgerRangeGetter) jrpc2.Handler { +func NewGetFeeStatsHandler(windows *feewindow.FeeWindows, reader db.TransactionReader, logger *log.Entry) jrpc2.Handler { return NewHandler(func(ctx context.Context) (GetFeeStatsResult, error) { + ledgerInfo, err := reader.GetLedgerRange(ctx) + if err != nil { // still not fatal + logger.WithError(err). + Error("could not fetch ledger range") + } + result := GetFeeStatsResult{ SorobanInclusionFee: convertFeeDistribution(windows.SorobanInclusionFeeWindow.GetFeeDistribution()), InclusionFee: convertFeeDistribution(windows.ClassicFeeWindow.GetFeeDistribution()), - LatestLedger: ledgerRangeGetter.GetLedgerRange().LastLedger.Sequence, + LatestLedger: ledgerInfo.LastLedger.Sequence, } return result, nil }) diff --git a/cmd/soroban-rpc/internal/methods/get_transaction.go b/cmd/soroban-rpc/internal/methods/get_transaction.go index 166cb152..399243e3 100644 --- a/cmd/soroban-rpc/internal/methods/get_transaction.go +++ b/cmd/soroban-rpc/internal/methods/get_transaction.go @@ -4,13 +4,14 @@ import ( "context" "encoding/base64" "encoding/hex" + "errors" "fmt" "github.com/creachadair/jrpc2" + "github.com/stellar/go/support/log" "github.com/stellar/go/xdr" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/transactions" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" ) const ( @@ -66,11 +67,12 @@ type GetTransactionRequest struct { Hash string `json:"hash"` } -type transactionGetter interface { - GetTransaction(hash xdr.Hash) (transactions.Transaction, bool, ledgerbucketwindow.LedgerRange) -} - -func GetTransaction(getter transactionGetter, request GetTransactionRequest) (GetTransactionResponse, error) { +func GetTransaction( + ctx context.Context, + log *log.Entry, + reader db.TransactionReader, + request GetTransactionRequest, +) (GetTransactionResponse, error) { // parse hash if hex.DecodedLen(len(request.Hash)) != len(xdr.Hash{}) { return GetTransactionResponse{}, &jrpc2.Error{ @@ -88,16 +90,22 @@ func GetTransaction(getter transactionGetter, request GetTransactionRequest) (Ge } } - tx, found, storeRange := getter.GetTransaction(txHash) + tx, storeRange, err := reader.GetTransaction(ctx, txHash) + response := GetTransactionResponse{ LatestLedger: storeRange.LastLedger.Sequence, LatestLedgerCloseTime: storeRange.LastLedger.CloseTime, OldestLedger: storeRange.FirstLedger.Sequence, OldestLedgerCloseTime: storeRange.FirstLedger.CloseTime, } - if !found { + if errors.Is(err, db.ErrNoTransaction) { response.Status = TransactionStatusNotFound return response, nil + } else if err != nil { + log.WithError(err). + WithField("hash", txHash). + Errorf("failed to fetch transaction") + return response, err } response.ApplicationOrder = tx.ApplicationOrder @@ -110,17 +118,16 @@ func GetTransaction(getter transactionGetter, request GetTransactionRequest) (Ge response.ResultMetaXdr = base64.StdEncoding.EncodeToString(tx.Meta) response.DiagnosticEventsXDR = base64EncodeSlice(tx.Events) + response.Status = TransactionStatusFailed if tx.Successful { response.Status = TransactionStatusSuccess - } else { - response.Status = TransactionStatusFailed } return response, nil } // NewGetTransactionHandler returns a get transaction json rpc handler -func NewGetTransactionHandler(getter transactionGetter) jrpc2.Handler { +func NewGetTransactionHandler(logger *log.Entry, getter db.TransactionReader) jrpc2.Handler { return NewHandler(func(ctx context.Context, request GetTransactionRequest) (GetTransactionResponse, error) { - return GetTransaction(getter, request) + return GetTransaction(ctx, logger, getter, request) }) } diff --git a/cmd/soroban-rpc/internal/methods/get_transaction_test.go b/cmd/soroban-rpc/internal/methods/get_transaction_test.go index dadf145f..960adf82 100644 --- a/cmd/soroban-rpc/internal/methods/get_transaction_test.go +++ b/cmd/soroban-rpc/internal/methods/get_transaction_test.go @@ -1,18 +1,162 @@ package methods import ( + "context" "encoding/hex" "testing" - "github.com/stellar/go/xdr" + "github.com/sirupsen/logrus" "github.com/stretchr/testify/require" "github.com/stellar/go/network" + "github.com/stellar/go/support/log" + "github.com/stellar/go/xdr" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/transactions" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" ) +func TestGetTransaction(t *testing.T) { + var ( + ctx = context.TODO() + log = log.DefaultLogger + store = db.NewMockTransactionStore("passphrase") + ) + log.SetLevel(logrus.DebugLevel) + + _, err := GetTransaction(ctx, log, store, GetTransactionRequest{"ab"}) + require.EqualError(t, err, "[-32602] unexpected hash length (2)") + _, err = GetTransaction(ctx, log, store, GetTransactionRequest{"foo "}) + require.EqualError(t, err, "[-32602] incorrect hash: encoding/hex: invalid byte: U+006F 'o'") + + hash := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + tx, err := GetTransaction(ctx, log, store, GetTransactionRequest{hash}) + require.NoError(t, err) + require.Equal(t, GetTransactionResponse{Status: TransactionStatusNotFound}, tx) + + meta := txMeta(1, true) + require.NoError(t, store.InsertTransactions(meta)) + + xdrHash := txHash(1) + hash = hex.EncodeToString(xdrHash[:]) + tx, err = GetTransaction(ctx, log, store, GetTransactionRequest{hash}) + require.NoError(t, err) + + expectedTxResult, err := xdr.MarshalBase64(meta.V1.TxProcessing[0].Result.Result) + require.NoError(t, err) + expectedEnvelope, err := xdr.MarshalBase64(txEnvelope(1)) + require.NoError(t, err) + expectedTxMeta, err := xdr.MarshalBase64(meta.V1.TxProcessing[0].TxApplyProcessing) + require.NoError(t, err) + require.Equal(t, GetTransactionResponse{ + Status: TransactionStatusSuccess, + LatestLedger: 101, + LatestLedgerCloseTime: 2625, + OldestLedger: 101, + OldestLedgerCloseTime: 2625, + ApplicationOrder: 1, + FeeBump: false, + EnvelopeXdr: expectedEnvelope, + ResultXdr: expectedTxResult, + ResultMetaXdr: expectedTxMeta, + Ledger: 101, + LedgerCloseTime: 2625, + DiagnosticEventsXDR: []string{}, + }, tx) + + // ingest another (failed) transaction + meta = txMeta(2, false) + require.NoError(t, store.InsertTransactions(meta)) + + // the first transaction should still be there + tx, err = GetTransaction(ctx, log, store, GetTransactionRequest{hash}) + require.NoError(t, err) + require.Equal(t, GetTransactionResponse{ + Status: TransactionStatusSuccess, + LatestLedger: 102, + LatestLedgerCloseTime: 2650, + OldestLedger: 101, + OldestLedgerCloseTime: 2625, + ApplicationOrder: 1, + FeeBump: false, + EnvelopeXdr: expectedEnvelope, + ResultXdr: expectedTxResult, + ResultMetaXdr: expectedTxMeta, + Ledger: 101, + LedgerCloseTime: 2625, + DiagnosticEventsXDR: []string{}, + }, tx) + + // the new transaction should also be there + xdrHash = txHash(2) + hash = hex.EncodeToString(xdrHash[:]) + + expectedTxResult, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].Result.Result) + require.NoError(t, err) + expectedEnvelope, err = xdr.MarshalBase64(txEnvelope(2)) + require.NoError(t, err) + expectedTxMeta, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].TxApplyProcessing) + require.NoError(t, err) + + tx, err = GetTransaction(ctx, log, store, GetTransactionRequest{hash}) + require.NoError(t, err) + require.Equal(t, GetTransactionResponse{ + Status: TransactionStatusFailed, + LatestLedger: 102, + LatestLedgerCloseTime: 2650, + OldestLedger: 101, + OldestLedgerCloseTime: 2625, + ApplicationOrder: 1, + FeeBump: false, + EnvelopeXdr: expectedEnvelope, + ResultXdr: expectedTxResult, + ResultMetaXdr: expectedTxMeta, + Ledger: 102, + LedgerCloseTime: 2650, + DiagnosticEventsXDR: []string{}, + }, tx) + + // Test Txn with events + meta = txMetaWithEvents(3, true) + require.NoError(t, store.InsertTransactions(meta)) + + xdrHash = txHash(3) + hash = hex.EncodeToString(xdrHash[:]) + + expectedTxResult, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].Result.Result) + require.NoError(t, err) + expectedEnvelope, err = xdr.MarshalBase64(txEnvelope(3)) + require.NoError(t, err) + expectedTxMeta, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].TxApplyProcessing) + require.NoError(t, err) + + diagnosticEvents, err := meta.V1.TxProcessing[0].TxApplyProcessing.GetDiagnosticEvents() + require.NoError(t, err) + expectedEventsMeta, err := xdr.MarshalBase64(diagnosticEvents[0]) + require.NoError(t, err) + + tx, err = GetTransaction(ctx, log, store, GetTransactionRequest{hash}) + require.NoError(t, err) + require.Equal(t, GetTransactionResponse{ + Status: TransactionStatusSuccess, + LatestLedger: 103, + LatestLedgerCloseTime: 2675, + OldestLedger: 101, + OldestLedgerCloseTime: 2625, + ApplicationOrder: 1, + FeeBump: false, + EnvelopeXdr: expectedEnvelope, + ResultXdr: expectedTxResult, + ResultMetaXdr: expectedTxMeta, + Ledger: 103, + LedgerCloseTime: 2675, + DiagnosticEventsXDR: []string{expectedEventsMeta}, + }, tx) +} + +func ledgerCloseTime(ledgerSequence uint32) int64 { + return int64(ledgerSequence)*25 + 100 +} + func txHash(acctSeq uint32) xdr.Hash { envelope := txEnvelope(acctSeq) hash, err := network.HashTransactionInEnvelope(envelope, "passphrase") @@ -23,8 +167,18 @@ func txHash(acctSeq uint32) xdr.Hash { return hash } -func ledgerCloseTime(ledgerSequence uint32) int64 { - return int64(ledgerSequence)*25 + 100 +func txEnvelope(acctSeq uint32) xdr.TransactionEnvelope { + envelope, err := xdr.NewTransactionEnvelope(xdr.EnvelopeTypeEnvelopeTypeTx, xdr.TransactionV1Envelope{ + Tx: xdr.Transaction{ + Fee: 1, + SeqNum: xdr.SequenceNumber(acctSeq), + SourceAccount: xdr.MustMuxedAddress("MA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVAAAAAAAAAAAAAJLK"), + }, + }) + if err != nil { + panic(err) + } + return envelope } func transactionResult(successful bool) xdr.TransactionResult { @@ -135,156 +289,3 @@ func txMetaWithEvents(acctSeq uint32, successful bool) xdr.LedgerCloseMeta { return meta } - -func txEnvelope(acctSeq uint32) xdr.TransactionEnvelope { - envelope, err := xdr.NewTransactionEnvelope(xdr.EnvelopeTypeEnvelopeTypeTx, xdr.TransactionV1Envelope{ - Tx: xdr.Transaction{ - Fee: 1, - SeqNum: xdr.SequenceNumber(acctSeq), - SourceAccount: xdr.MustMuxedAddress("MA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVAAAAAAAAAAAAAJLK"), - }, - }) - if err != nil { - panic(err) - } - return envelope -} - -func TestGetTransaction(t *testing.T) { - store := transactions.NewMemoryStore(interfaces.MakeNoOpDeamon(), "passphrase", 100) - _, err := GetTransaction(store, GetTransactionRequest{"ab"}) - require.EqualError(t, err, "[-32602] unexpected hash length (2)") - _, err = GetTransaction(store, GetTransactionRequest{"foo "}) - require.EqualError(t, err, "[-32602] incorrect hash: encoding/hex: invalid byte: U+006F 'o'") - - hash := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" - tx, err := GetTransaction(store, GetTransactionRequest{hash}) - require.NoError(t, err) - require.Equal(t, GetTransactionResponse{ - Status: TransactionStatusNotFound, - }, tx) - - meta := txMeta(1, true) - err = store.IngestTransactions(meta) - require.NoError(t, err) - - xdrHash := txHash(1) - hash = hex.EncodeToString(xdrHash[:]) - tx, err = GetTransaction(store, GetTransactionRequest{hash}) - require.NoError(t, err) - - expectedTxResult, err := xdr.MarshalBase64(meta.V1.TxProcessing[0].Result.Result) - require.NoError(t, err) - expectedEnvelope, err := xdr.MarshalBase64(txEnvelope(1)) - require.NoError(t, err) - expectedTxMeta, err := xdr.MarshalBase64(meta.V1.TxProcessing[0].TxApplyProcessing) - require.NoError(t, err) - require.Equal(t, GetTransactionResponse{ - Status: TransactionStatusSuccess, - LatestLedger: 101, - LatestLedgerCloseTime: 2625, - OldestLedger: 101, - OldestLedgerCloseTime: 2625, - ApplicationOrder: 1, - FeeBump: false, - EnvelopeXdr: expectedEnvelope, - ResultXdr: expectedTxResult, - ResultMetaXdr: expectedTxMeta, - Ledger: 101, - LedgerCloseTime: 2625, - DiagnosticEventsXDR: []string{}, - }, tx) - - // ingest another (failed) transaction - meta = txMeta(2, false) - err = store.IngestTransactions(meta) - require.NoError(t, err) - - // the first transaction should still be there - tx, err = GetTransaction(store, GetTransactionRequest{hash}) - require.NoError(t, err) - require.Equal(t, GetTransactionResponse{ - Status: TransactionStatusSuccess, - LatestLedger: 102, - LatestLedgerCloseTime: 2650, - OldestLedger: 101, - OldestLedgerCloseTime: 2625, - ApplicationOrder: 1, - FeeBump: false, - EnvelopeXdr: expectedEnvelope, - ResultXdr: expectedTxResult, - ResultMetaXdr: expectedTxMeta, - Ledger: 101, - LedgerCloseTime: 2625, - DiagnosticEventsXDR: []string{}, - }, tx) - - // the new transaction should also be there - xdrHash = txHash(2) - hash = hex.EncodeToString(xdrHash[:]) - - expectedTxResult, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].Result.Result) - require.NoError(t, err) - expectedEnvelope, err = xdr.MarshalBase64(txEnvelope(2)) - require.NoError(t, err) - expectedTxMeta, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].TxApplyProcessing) - require.NoError(t, err) - - tx, err = GetTransaction(store, GetTransactionRequest{hash}) - require.NoError(t, err) - require.NoError(t, err) - require.Equal(t, GetTransactionResponse{ - Status: TransactionStatusFailed, - LatestLedger: 102, - LatestLedgerCloseTime: 2650, - OldestLedger: 101, - OldestLedgerCloseTime: 2625, - ApplicationOrder: 1, - FeeBump: false, - EnvelopeXdr: expectedEnvelope, - ResultXdr: expectedTxResult, - ResultMetaXdr: expectedTxMeta, - Ledger: 102, - LedgerCloseTime: 2650, - DiagnosticEventsXDR: []string{}, - }, tx) - - // Test Txn with events - meta = txMetaWithEvents(3, true) - err = store.IngestTransactions(meta) - require.NoError(t, err) - - xdrHash = txHash(3) - hash = hex.EncodeToString(xdrHash[:]) - - expectedTxResult, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].Result.Result) - require.NoError(t, err) - expectedEnvelope, err = xdr.MarshalBase64(txEnvelope(3)) - require.NoError(t, err) - expectedTxMeta, err = xdr.MarshalBase64(meta.V1.TxProcessing[0].TxApplyProcessing) - require.NoError(t, err) - - diagnosticEvents, err := meta.V1.TxProcessing[0].TxApplyProcessing.GetDiagnosticEvents() - require.NoError(t, err) - expectedEventsMeta, err := xdr.MarshalBase64(diagnosticEvents[0]) - - tx, err = GetTransaction(store, GetTransactionRequest{hash}) - require.NoError(t, err) - require.NoError(t, err) - require.Equal(t, GetTransactionResponse{ - Status: TransactionStatusSuccess, - LatestLedger: 103, - LatestLedgerCloseTime: 2675, - OldestLedger: 101, - OldestLedgerCloseTime: 2625, - ApplicationOrder: 1, - FeeBump: false, - EnvelopeXdr: expectedEnvelope, - ResultXdr: expectedTxResult, - ResultMetaXdr: expectedTxMeta, - Ledger: 103, - LedgerCloseTime: 2675, - DiagnosticEventsXDR: []string{expectedEventsMeta}, - }, tx) - -} diff --git a/cmd/soroban-rpc/internal/methods/get_transactions.go b/cmd/soroban-rpc/internal/methods/get_transactions.go new file mode 100644 index 00000000..f8c38dd8 --- /dev/null +++ b/cmd/soroban-rpc/internal/methods/get_transactions.go @@ -0,0 +1,241 @@ +package methods + +import ( + "context" + "encoding/base64" + "fmt" + "io" + "strconv" + + "github.com/creachadair/jrpc2" + "github.com/creachadair/jrpc2/handler" + "github.com/stellar/go/ingest" + "github.com/stellar/go/support/errors" + "github.com/stellar/go/support/log" + "github.com/stellar/go/toid" + + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" +) + +// TransactionsPaginationOptions defines the available options for paginating through transactions. +type TransactionsPaginationOptions struct { + Cursor string `json:"cursor,omitempty"` + Limit uint `json:"limit,omitempty"` +} + +// GetTransactionsRequest represents the request parameters for fetching transactions within a range of ledgers. +type GetTransactionsRequest struct { + StartLedger uint32 `json:"startLedger"` + Pagination *TransactionsPaginationOptions `json:"pagination,omitempty"` +} + +// isValid checks the validity of the request parameters. +func (req GetTransactionsRequest) isValid(maxLimit uint, ledgerRange ledgerbucketwindow.LedgerRange) error { + if req.Pagination != nil && req.Pagination.Cursor != "" { + if req.StartLedger != 0 { + return errors.New("startLedger and cursor cannot both be set") + } + } else if req.StartLedger < ledgerRange.FirstLedger.Sequence || req.StartLedger > ledgerRange.LastLedger.Sequence { + return errors.Errorf("start ledger must be between the oldest ledger: %d and the latest ledger: %d for this rpc instance.", ledgerRange.FirstLedger.Sequence, ledgerRange.LastLedger.Sequence) + } + + if req.Pagination != nil && req.Pagination.Limit > maxLimit { + return fmt.Errorf("limit must not exceed %d", maxLimit) + } + + return nil +} + +type TransactionInfo struct { + // Successful indicates whether the transaction was successful or not + Successful bool `json:"status"` + // ApplicationOrder is the index of the transaction among all the transactions + // for that ledger. + ApplicationOrder int32 `json:"applicationOrder"` + // FeeBump indicates whether the transaction is a feebump transaction + FeeBump bool `json:"feeBump"` + // EnvelopeXdr is the TransactionEnvelope XDR value. + EnvelopeXdr string `json:"envelopeXdr"` + // ResultXdr is the TransactionResult XDR value. + ResultXdr string `json:"resultXdr"` + // ResultMetaXdr is the TransactionMeta XDR value. + ResultMetaXdr string `json:"resultMetaXdr"` + // DiagnosticEventsXDR is present only if transaction was not successful. + // DiagnosticEventsXDR is a base64-encoded slice of xdr.DiagnosticEvent + DiagnosticEventsXDR []string `json:"diagnosticEventsXdr,omitempty"` + // Ledger is the sequence of the ledger which included the transaction. + Ledger uint32 `json:"ledger"` + // LedgerCloseTime is the unix timestamp of when the transaction was included in the ledger. + LedgerCloseTime int64 `json:"createdAt"` +} + +// GetTransactionsResponse encapsulates the response structure for getTransactions queries. +type GetTransactionsResponse struct { + Transactions []TransactionInfo `json:"transactions"` + LatestLedger uint32 `json:"latestLedger"` + LatestLedgerCloseTime int64 `json:"latestLedgerCloseTimestamp"` + OldestLedger uint32 `json:"oldestLedger"` + OldestLedgerCloseTime int64 `json:"oldestLedgerCloseTimestamp"` + Cursor string `json:"cursor"` +} + +type transactionsRPCHandler struct { + ledgerReader db.LedgerReader + dbReader db.TransactionReader + maxLimit uint + defaultLimit uint + logger *log.Entry + networkPassphrase string +} + +// getTransactionsByLedgerSequence fetches transactions between the start and end ledgers, inclusive of both. +// The number of ledgers returned can be tuned using the pagination options - cursor and limit. +func (h transactionsRPCHandler) getTransactionsByLedgerSequence(ctx context.Context, request GetTransactionsRequest) (GetTransactionsResponse, error) { + ledgerRange, err := h.dbReader.GetLedgerRange(ctx) + if err != nil { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: err.Error(), + } + } + + err = request.isValid(h.maxLimit, ledgerRange) + if err != nil { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InvalidRequest, + Message: err.Error(), + } + } + + // Move start to pagination cursor + start := toid.New(int32(request.StartLedger), 1, 1) + limit := h.defaultLimit + if request.Pagination != nil { + if request.Pagination.Cursor != "" { + cursorInt, err := strconv.ParseInt(request.Pagination.Cursor, 10, 64) + if err != nil { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InvalidParams, + Message: err.Error(), + } + } + + *start = toid.Parse(cursorInt) + // increment tx index because, when paginating, + // we start with the item right after the cursor + start.TransactionOrder++ + } + if request.Pagination.Limit > 0 { + limit = request.Pagination.Limit + } + } + + // Iterate through each ledger and its transactions until limit or end range is reached. + // The latest ledger acts as the end ledger range for the request. + var txns []TransactionInfo + var cursor *toid.ID +LedgerLoop: + for ledgerSeq := start.LedgerSequence; ledgerSeq <= int32(ledgerRange.LastLedger.Sequence); ledgerSeq++ { + // Get ledger close meta from db + ledger, found, err := h.ledgerReader.GetLedger(ctx, uint32(ledgerSeq)) + if err != nil { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: err.Error(), + } + } else if !found { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InvalidParams, + Message: errors.Errorf("ledger close meta not found: %d", ledgerSeq).Error(), + } + } + + // Initialise tx reader. + reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(h.networkPassphrase, ledger) + if err != nil { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: err.Error(), + } + } + + // Move the reader to specific tx idx + startTxIdx := 1 + if ledgerSeq == start.LedgerSequence { + startTxIdx = int(start.TransactionOrder) + if ierr := reader.Seek(startTxIdx - 1); ierr != nil && ierr != io.EOF { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: err.Error(), + } + } + } + + // Decode transaction info from ledger meta + txCount := ledger.CountTransactions() + for i := startTxIdx; i <= txCount; i++ { + cursor = toid.New(int32(ledger.LedgerSequence()), int32(i), 1) + + ingestTx, err := reader.Read() + if err != nil { + if err == io.EOF { + // No more transactions to read. Start from next ledger + break + } + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InvalidParams, + Message: err.Error(), + } + } + + tx, err := db.ParseTransaction(ledger, ingestTx) + if err != nil { + return GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: err.Error(), + } + } + + txInfo := TransactionInfo{ + Successful: tx.Successful, + ApplicationOrder: tx.ApplicationOrder, + FeeBump: tx.FeeBump, + ResultXdr: base64.StdEncoding.EncodeToString(tx.Result), + ResultMetaXdr: base64.StdEncoding.EncodeToString(tx.Meta), + EnvelopeXdr: base64.StdEncoding.EncodeToString(tx.Envelope), + DiagnosticEventsXDR: base64EncodeSlice(tx.Events), + Ledger: tx.Ledger.Sequence, + LedgerCloseTime: tx.Ledger.CloseTime, + } + txns = append(txns, txInfo) + if len(txns) >= int(limit) { + break LedgerLoop + } + } + } + + return GetTransactionsResponse{ + Transactions: txns, + LatestLedger: ledgerRange.LastLedger.Sequence, + LatestLedgerCloseTime: ledgerRange.LastLedger.CloseTime, + OldestLedger: ledgerRange.FirstLedger.Sequence, + OldestLedgerCloseTime: ledgerRange.FirstLedger.CloseTime, + Cursor: cursor.String(), + }, nil +} + +func NewGetTransactionsHandler(logger *log.Entry, ledgerReader db.LedgerReader, dbReader db.TransactionReader, maxLimit, defaultLimit uint, networkPassphrase string) jrpc2.Handler { + transactionsHandler := transactionsRPCHandler{ + ledgerReader: ledgerReader, + dbReader: dbReader, + maxLimit: maxLimit, + defaultLimit: defaultLimit, + logger: logger, + networkPassphrase: networkPassphrase, + } + + return handler.New(func(context context.Context, request GetTransactionsRequest) (GetTransactionsResponse, error) { + return transactionsHandler.getTransactionsByLedgerSequence(context, request) + }) +} diff --git a/cmd/soroban-rpc/internal/methods/get_transactions_test.go b/cmd/soroban-rpc/internal/methods/get_transactions_test.go new file mode 100644 index 00000000..f6f7fea2 --- /dev/null +++ b/cmd/soroban-rpc/internal/methods/get_transactions_test.go @@ -0,0 +1,302 @@ +package methods + +import ( + "context" + "testing" + + "github.com/creachadair/jrpc2" + "github.com/stellar/go/support/errors" + "github.com/stellar/go/toid" + "github.com/stellar/go/xdr" + "github.com/stretchr/testify/assert" + + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" +) + +const ( + NetworkPassphrase string = "passphrase" +) + +// createTestLedger Creates a test ledger with 2 transactions +func createTestLedger(sequence uint32) xdr.LedgerCloseMeta { + sequence = sequence - 100 + meta := txMeta(sequence, true) + meta.V1.TxProcessing = append(meta.V1.TxProcessing, xdr.TransactionResultMeta{ + TxApplyProcessing: xdr.TransactionMeta{ + V: 3, + Operations: &[]xdr.OperationMeta{}, + V3: &xdr.TransactionMetaV3{}, + }, + Result: xdr.TransactionResultPair{ + TransactionHash: txHash(sequence), + Result: transactionResult(false), + }, + }) + return meta +} + +func TestGetTransactions_DefaultLimit(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 10; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + StartLedger: 1, + } + + response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + assert.NoError(t, err) + + // assert latest ledger details + assert.Equal(t, uint32(10), response.LatestLedger) + assert.Equal(t, int64(350), response.LatestLedgerCloseTime) + + // assert pagination + assert.Equal(t, toid.New(5, 2, 1).String(), response.Cursor) + + // assert transactions result + assert.Equal(t, 10, len(response.Transactions)) +} + +func TestGetTransactions_DefaultLimitExceedsLatestLedger(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 3; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + StartLedger: 1, + } + + response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + assert.NoError(t, err) + + // assert latest ledger details + assert.Equal(t, uint32(3), response.LatestLedger) + assert.Equal(t, int64(175), response.LatestLedgerCloseTime) + + // assert pagination + assert.Equal(t, toid.New(3, 2, 1).String(), response.Cursor) + + // assert transactions result + assert.Equal(t, 6, len(response.Transactions)) +} + +func TestGetTransactions_CustomLimit(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 10; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + StartLedger: 1, + Pagination: &TransactionsPaginationOptions{ + Limit: 2, + }, + } + + response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + assert.NoError(t, err) + + // assert latest ledger details + assert.Equal(t, uint32(10), response.LatestLedger) + assert.Equal(t, int64(350), response.LatestLedgerCloseTime) + + // assert pagination + assert.Equal(t, toid.New(1, 2, 1).String(), response.Cursor) + + // assert transactions result + assert.Equal(t, 2, len(response.Transactions)) + assert.Equal(t, uint32(1), response.Transactions[0].Ledger) + assert.Equal(t, uint32(1), response.Transactions[1].Ledger) +} + +func TestGetTransactions_CustomLimitAndCursor(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 10; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + Pagination: &TransactionsPaginationOptions{ + Cursor: toid.New(1, 2, 1).String(), + Limit: 3, + }, + } + + response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + assert.NoError(t, err) + + // assert latest ledger details + assert.Equal(t, uint32(10), response.LatestLedger) + assert.Equal(t, int64(350), response.LatestLedgerCloseTime) + + // assert pagination + assert.Equal(t, toid.New(3, 1, 1).String(), response.Cursor) + + // assert transactions result + assert.Equal(t, 3, len(response.Transactions)) + assert.Equal(t, uint32(2), response.Transactions[0].Ledger) + assert.Equal(t, uint32(2), response.Transactions[1].Ledger) + assert.Equal(t, uint32(3), response.Transactions[2].Ledger) +} + +func TestGetTransactions_InvalidStartLedger(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 3; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + StartLedger: 4, + } + + response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + expectedErr := errors.Errorf("[%d] start ledger must be between the oldest ledger: 1 and the latest ledger: 3 for this rpc instance.", jrpc2.InvalidRequest) + assert.Equal(t, expectedErr.Error(), err.Error()) + assert.Nil(t, response.Transactions) +} + +func TestGetTransactions_LedgerNotFound(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 3; i++ { + // Skip creation of ledger 2 + if i == 2 { + continue + } + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + StartLedger: 1, + } + + response, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + expectedErr := errors.Errorf("[%d] ledger close meta not found: 2", jrpc2.InvalidParams) + assert.Equal(t, expectedErr.Error(), err.Error()) + assert.Nil(t, response.Transactions) +} + +func TestGetTransactions_LimitGreaterThanMaxLimit(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 3; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + StartLedger: 1, + Pagination: &TransactionsPaginationOptions{ + Limit: 200, + }, + } + + _, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + expectedErr := errors.Errorf("[%d] limit must not exceed 100", jrpc2.InvalidRequest) + assert.Equal(t, expectedErr.Error(), err.Error()) +} + +func TestGetTransactions_InvalidCursorString(t *testing.T) { + mockDbReader := db.NewMockTransactionStore(NetworkPassphrase) + mockLedgerReader := db.NewMockLedgerReader(mockDbReader) + for i := 1; i <= 3; i++ { + meta := createTestLedger(uint32(i)) + err := mockDbReader.InsertTransactions(meta) + assert.NoError(t, err) + } + + handler := transactionsRPCHandler{ + ledgerReader: mockLedgerReader, + dbReader: mockDbReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + + request := GetTransactionsRequest{ + Pagination: &TransactionsPaginationOptions{ + Cursor: "abc", + }, + } + + _, err := handler.getTransactionsByLedgerSequence(context.TODO(), request) + expectedErr := errors.Errorf("[%d] strconv.ParseInt: parsing \"abc\": invalid syntax", jrpc2.InvalidParams) + assert.Equal(t, expectedErr.Error(), err.Error()) +} diff --git a/cmd/soroban-rpc/internal/methods/health.go b/cmd/soroban-rpc/internal/methods/health.go index f9b1c50e..267b7206 100644 --- a/cmd/soroban-rpc/internal/methods/health.go +++ b/cmd/soroban-rpc/internal/methods/health.go @@ -6,8 +6,7 @@ import ( "time" "github.com/creachadair/jrpc2" - - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" ) type HealthCheckResult struct { @@ -17,18 +16,22 @@ type HealthCheckResult struct { LedgerRetentionWindow uint32 `json:"ledgerRetentionWindow"` } -type LedgerRangeGetter interface { - GetLedgerRange() ledgerbucketwindow.LedgerRange -} - // NewHealthCheck returns a health check json rpc handler -func NewHealthCheck(retentionWindow uint32, ledgerRangeGetter LedgerRangeGetter, maxHealthyLedgerLatency time.Duration) jrpc2.Handler { +func NewHealthCheck( + retentionWindow uint32, + reader db.TransactionReader, + maxHealthyLedgerLatency time.Duration, +) jrpc2.Handler { return NewHandler(func(ctx context.Context) (HealthCheckResult, error) { - ledgerRange := ledgerRangeGetter.GetLedgerRange() - if ledgerRange.LastLedger.Sequence < 1 { + ledgerRange, err := reader.GetLedgerRange(ctx) + if err != nil || ledgerRange.LastLedger.Sequence < 1 { + extra := "" + if err != nil { + extra = fmt.Sprintf(": %s", err.Error()) + } return HealthCheckResult{}, jrpc2.Error{ Code: jrpc2.InternalError, - Message: "data stores are not initialized", + Message: "data stores are not initialized" + extra, } } @@ -36,7 +39,8 @@ func NewHealthCheck(retentionWindow uint32, ledgerRangeGetter LedgerRangeGetter, lastKnownLedgerLatency := time.Since(lastKnownLedgerCloseTime) if lastKnownLedgerLatency > maxHealthyLedgerLatency { roundedLatency := lastKnownLedgerLatency.Round(time.Second) - msg := fmt.Sprintf("latency (%s) since last known ledger closed is too high (>%s)", roundedLatency, maxHealthyLedgerLatency) + msg := fmt.Sprintf("latency (%s) since last known ledger closed is too high (>%s)", + roundedLatency, maxHealthyLedgerLatency) return HealthCheckResult{}, jrpc2.Error{ Code: jrpc2.InternalError, Message: msg, diff --git a/cmd/soroban-rpc/internal/methods/send_transaction.go b/cmd/soroban-rpc/internal/methods/send_transaction.go index 82d014d1..c8e476be 100644 --- a/cmd/soroban-rpc/internal/methods/send_transaction.go +++ b/cmd/soroban-rpc/internal/methods/send_transaction.go @@ -11,6 +11,7 @@ import ( "github.com/stellar/go/xdr" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" ) // SendTransactionResponse represents the transaction submission response returned Soroban-RPC @@ -44,7 +45,12 @@ type SendTransactionRequest struct { } // NewSendTransactionHandler returns a submit transaction json rpc handler -func NewSendTransactionHandler(daemon interfaces.Daemon, logger *log.Entry, ledgerRangeGetter LedgerRangeGetter, passphrase string) jrpc2.Handler { +func NewSendTransactionHandler( + daemon interfaces.Daemon, + logger *log.Entry, + reader db.TransactionReader, + passphrase string, +) jrpc2.Handler { submitter := daemon.CoreClient() return NewHandler(func(ctx context.Context, request SendTransactionRequest) (SendTransactionResponse, error) { var envelope xdr.TransactionEnvelope @@ -66,11 +72,19 @@ func NewSendTransactionHandler(daemon interfaces.Daemon, logger *log.Entry, ledg } txHash := hex.EncodeToString(hash[:]) - latestLedgerInfo := ledgerRangeGetter.GetLedgerRange().LastLedger + ledgerInfo, err := reader.GetLedgerRange(ctx) + if err != nil { // still not fatal + logger.WithError(err). + WithField("tx", request.Transaction). + Error("could not fetch ledger range") + } + latestLedgerInfo := ledgerInfo.LastLedger + resp, err := submitter.SubmitTransaction(ctx, request.Transaction) if err != nil { logger.WithError(err). - WithField("tx", request.Transaction).Error("could not submit transaction") + WithField("tx", request.Transaction). + Error("could not submit transaction") return SendTransactionResponse{}, &jrpc2.Error{ Code: jrpc2.InternalError, Message: "could not submit transaction to stellar-core", diff --git a/cmd/soroban-rpc/internal/preflight/preflight_test.go b/cmd/soroban-rpc/internal/preflight/preflight_test.go index 2a4a90b9..4e361605 100644 --- a/cmd/soroban-rpc/internal/preflight/preflight_test.go +++ b/cmd/soroban-rpc/internal/preflight/preflight_test.go @@ -8,10 +8,12 @@ import ( "runtime" "testing" + "github.com/stellar/go/network" "github.com/stellar/go/support/log" "github.com/stellar/go/xdr" "github.com/stretchr/testify/require" + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/db" ) @@ -293,15 +295,18 @@ func getDB(t testing.TB, restartDB bool) *db.DB { dbPath := path.Join(t.TempDir(), "soroban_rpc.sqlite") dbInstance, err := db.OpenSQLiteDB(dbPath) require.NoError(t, err) - readWriter := db.NewReadWriter(dbInstance, 100, 10000) + + readWriter := db.NewReadWriter(log.DefaultLogger, dbInstance, interfaces.MakeNoOpDeamon(), + 100, 10000, network.FutureNetworkPassphrase) tx, err := readWriter.NewTx(context.Background()) require.NoError(t, err) + for _, e := range mockLedgerEntries { err := tx.LedgerEntryWriter().UpsertLedgerEntry(e) require.NoError(t, err) } - err = tx.Commit(2) - require.NoError(t, err) + require.NoError(t, tx.Commit(2)) + if restartDB { // Restarting the DB resets the ledger entries write-through cache require.NoError(t, dbInstance.Close()) diff --git a/cmd/soroban-rpc/internal/test/get_transactions_test.go b/cmd/soroban-rpc/internal/test/get_transactions_test.go new file mode 100644 index 00000000..f3da92dd --- /dev/null +++ b/cmd/soroban-rpc/internal/test/get_transactions_test.go @@ -0,0 +1,101 @@ +package test + +import ( + "context" + "testing" + + "github.com/creachadair/jrpc2" + "github.com/creachadair/jrpc2/jhttp" + "github.com/stellar/go/keypair" + "github.com/stellar/go/txnbuild" + "github.com/stretchr/testify/assert" + + "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/methods" +) + +// buildSetOptionsTxParams constructs the parameters necessary for creating a transaction from the given account. +// +// account - the source account from which the transaction will originate. This account provides the starting sequence number. +// +// Returns a fully populated TransactionParams structure. +func buildSetOptionsTxParams(account txnbuild.SimpleAccount) txnbuild.TransactionParams { + params := txnbuild.TransactionParams{ + SourceAccount: &account, + IncrementSequenceNum: true, + Operations: []txnbuild.Operation{ + &txnbuild.SetOptions{HomeDomain: txnbuild.NewHomeDomain("soroban.com")}, + }, + BaseFee: txnbuild.MinBaseFee, + Preconditions: txnbuild.Preconditions{TimeBounds: txnbuild.NewInfiniteTimeout()}, + } + return params +} + +// sendTransactions sends multiple transactions for testing purposes. +// It sends a total of three transactions, each from a new account sequence, and gathers the ledger +// numbers where these transactions were recorded. +// +// t - the testing framework handle for assertions. +// client - the JSON-RPC client used to send the transactions. +// +// Returns a slice of ledger numbers corresponding to where each transaction was recorded. +func sendTransactions(t *testing.T, client *jrpc2.Client) []uint32 { + kp := keypair.Root(StandaloneNetworkPassphrase) + address := kp.Address() + + var ledgers []uint32 + for i := 0; i <= 2; i++ { + account := txnbuild.NewSimpleAccount(address, int64(i)) + tx, err := txnbuild.NewTransaction(buildSetOptionsTxParams(account)) + assert.NoError(t, err) + + txResponse := sendSuccessfulTransaction(t, client, kp, tx) + ledgers = append(ledgers, txResponse.Ledger) + } + return ledgers +} + +func TestGetTransactions(t *testing.T) { + test := NewTest(t, nil) + ch := jhttp.NewChannel(test.sorobanRPCURL(), nil) + client := jrpc2.NewClient(ch, nil) + + ledgers := sendTransactions(t, client) + + // Get transactions across multiple ledgers + var result methods.GetTransactionsResponse + request := methods.GetTransactionsRequest{ + StartLedger: ledgers[0], + } + err := client.CallResult(context.Background(), "getTransactions", request, &result) + assert.NoError(t, err) + assert.Equal(t, len(result.Transactions), 3) + assert.Equal(t, result.Transactions[0].Ledger, ledgers[0]) + assert.Equal(t, result.Transactions[1].Ledger, ledgers[1]) + assert.Equal(t, result.Transactions[2].Ledger, ledgers[2]) + + // Get transactions with limit + request = methods.GetTransactionsRequest{ + StartLedger: ledgers[0], + Pagination: &methods.TransactionsPaginationOptions{ + Limit: 1, + }, + } + err = client.CallResult(context.Background(), "getTransactions", request, &result) + assert.NoError(t, err) + assert.Equal(t, len(result.Transactions), 1) + assert.Equal(t, result.Transactions[0].Ledger, ledgers[0]) + + // Get transactions using previous result's cursor + request = methods.GetTransactionsRequest{ + Pagination: &methods.TransactionsPaginationOptions{ + Cursor: result.Cursor, + Limit: 5, + }, + } + err = client.CallResult(context.Background(), "getTransactions", request, &result) + assert.NoError(t, err) + assert.Equal(t, len(result.Transactions), 2) + assert.Equal(t, result.Transactions[0].Ledger, ledgers[1]) + assert.Equal(t, result.Transactions[1].Ledger, ledgers[2]) +} diff --git a/cmd/soroban-rpc/internal/test/health_test.go b/cmd/soroban-rpc/internal/test/health_test.go index 75e097d4..0840959c 100644 --- a/cmd/soroban-rpc/internal/test/health_test.go +++ b/cmd/soroban-rpc/internal/test/health_test.go @@ -23,7 +23,7 @@ func TestHealth(t *testing.T) { t.Fatalf("rpc call failed: %v", err) } assert.Equal(t, "healthy", result.Status) - assert.Equal(t, uint32(ledgerbucketwindow.DefaultEventLedgerRetentionWindow), result.LedgerRetentionWindow) + assert.Equal(t, uint32(ledgerbucketwindow.OneDayOfLedgers), result.LedgerRetentionWindow) assert.Greater(t, result.OldestLedger, uint32(0)) assert.Greater(t, result.LatestLedger, uint32(0)) assert.GreaterOrEqual(t, result.LatestLedger, result.OldestLedger) diff --git a/cmd/soroban-rpc/internal/test/integration.go b/cmd/soroban-rpc/internal/test/integration.go index fc22c986..03e86237 100644 --- a/cmd/soroban-rpc/internal/test/integration.go +++ b/cmd/soroban-rpc/internal/test/integration.go @@ -177,7 +177,8 @@ func (i *Test) launchDaemon(coreBinaryPath string) { config.LogLevel = logrus.DebugLevel config.SQLiteDBPath = path.Join(i.t.TempDir(), "soroban_rpc.sqlite") config.IngestionTimeout = 10 * time.Minute - config.EventLedgerRetentionWindow = ledgerbucketwindow.DefaultEventLedgerRetentionWindow + config.EventLedgerRetentionWindow = ledgerbucketwindow.OneDayOfLedgers + config.TransactionLedgerRetentionWindow = ledgerbucketwindow.OneDayOfLedgers config.CheckpointFrequency = checkpointFrequency config.MaxHealthyLedgerLatency = time.Second * 10 config.PreflightEnableDebug = true @@ -210,7 +211,7 @@ func (i *Test) launchDaemon(coreBinaryPath string) { time.Sleep(time.Second) } if !success { - i.t.Fatalf("LedgerEntryStorage failed to sync in 1 minute") + i.t.Fatal("LedgerEntryStorage failed to sync in 1 minute") } } diff --git a/cmd/soroban-rpc/internal/test/transaction_test.go b/cmd/soroban-rpc/internal/test/transaction_test.go index 372e4ee2..f838ad6b 100644 --- a/cmd/soroban-rpc/internal/test/transaction_test.go +++ b/cmd/soroban-rpc/internal/test/transaction_test.go @@ -286,8 +286,7 @@ func sendSuccessfulTransaction(t *testing.T, client *jrpc2.Client, kp *keypair.F request := methods.SendTransactionRequest{Transaction: b64} var result methods.SendTransactionResponse - err = client.CallResult(context.Background(), "sendTransaction", request, &result) - assert.NoError(t, err) + assert.NoError(t, client.CallResult(context.Background(), "sendTransaction", request, &result)) expectedHashHex, err := tx.HashHex(StandaloneNetworkPassphrase) assert.NoError(t, err) @@ -297,7 +296,7 @@ func sendSuccessfulTransaction(t *testing.T, client *jrpc2.Client, kp *keypair.F var txResult xdr.TransactionResult err := xdr.SafeUnmarshalBase64(result.ErrorResultXDR, &txResult) assert.NoError(t, err) - fmt.Printf("error: %#v\n", txResult) + t.Logf("error: %#v\n", txResult) } assert.NotZero(t, result.LatestLedger) assert.NotZero(t, result.LatestLedgerCloseTime) @@ -305,24 +304,24 @@ func sendSuccessfulTransaction(t *testing.T, client *jrpc2.Client, kp *keypair.F response := getTransaction(t, client, expectedHashHex) if !assert.Equal(t, methods.TransactionStatusSuccess, response.Status) { var txResult xdr.TransactionResult - err := xdr.SafeUnmarshalBase64(response.ResultXdr, &txResult) - assert.NoError(t, err) - fmt.Printf("error: %#v\n", txResult) + assert.NoError(t, xdr.SafeUnmarshalBase64(response.ResultXdr, &txResult)) + t.Logf("error: %#v\n", txResult) + var txMeta xdr.TransactionMeta - err = xdr.SafeUnmarshalBase64(response.ResultMetaXdr, &txMeta) - assert.NoError(t, err) + assert.NoError(t, xdr.SafeUnmarshalBase64(response.ResultMetaXdr, &txMeta)) + if txMeta.V == 3 && txMeta.V3.SorobanMeta != nil { if len(txMeta.V3.SorobanMeta.Events) > 0 { - fmt.Println("Contract events:") + t.Log("Contract events:") for i, e := range txMeta.V3.SorobanMeta.Events { - fmt.Printf(" %d: %s\n", i, e) + t.Logf(" %d: %s\n", i, e) } } if len(txMeta.V3.SorobanMeta.DiagnosticEvents) > 0 { - fmt.Println("Diagnostic events:") + t.Log("Diagnostic events:") for i, d := range txMeta.V3.SorobanMeta.DiagnosticEvents { - fmt.Printf(" %d: %s\n", i, d) + t.Logf(" %d: %s\n", i, d) } } } diff --git a/cmd/soroban-rpc/internal/transactions/transactions.go b/cmd/soroban-rpc/internal/transactions/transactions.go deleted file mode 100644 index e5abe55c..00000000 --- a/cmd/soroban-rpc/internal/transactions/transactions.go +++ /dev/null @@ -1,209 +0,0 @@ -package transactions - -import ( - "sync" - "time" - - "github.com/prometheus/client_golang/prometheus" - "github.com/stellar/go/ingest" - "github.com/stellar/go/xdr" - - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" -) - -type transaction struct { - bucket *ledgerbucketwindow.LedgerBucket[[]xdr.Hash] - result []byte // encoded XDR of xdr.TransactionResult - meta []byte // encoded XDR of xdr.TransactionMeta - envelope []byte // encoded XDR of xdr.TransactionEnvelope - feeBump bool - successful bool - applicationOrder int32 -} - -// MemoryStore is an in-memory store of Stellar transactions. -type MemoryStore struct { - // networkPassphrase is an immutable string containing the - // Stellar network passphrase. - // Accessing networkPassphrase does not need to be protected - // by the lock - networkPassphrase string - lock sync.RWMutex - transactions map[xdr.Hash]transaction - transactionsByLedger *ledgerbucketwindow.LedgerBucketWindow[[]xdr.Hash] - transactionDurationMetric *prometheus.SummaryVec - transactionCountMetric prometheus.Summary -} - -// NewMemoryStore creates a new MemoryStore. -// The retention window is in units of ledgers. -// All events occurring in the following ledger range -// [ latestLedger - retentionWindow, latestLedger ] -// will be included in the MemoryStore. If the MemoryStore -// is full, any transactions from new ledgers will evict -// older entries outside the retention window. -func NewMemoryStore(daemon interfaces.Daemon, networkPassphrase string, retentionWindow uint32) *MemoryStore { - window := ledgerbucketwindow.NewLedgerBucketWindow[[]xdr.Hash](retentionWindow) - - // transactionDurationMetric is a metric for measuring latency of transaction store operations - transactionDurationMetric := prometheus.NewSummaryVec(prometheus.SummaryOpts{ - Namespace: daemon.MetricsNamespace(), Subsystem: "transactions", Name: "operation_duration_seconds", - Help: "transaction store operation durations, sliding window = 10m", - Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, - }, - []string{"operation"}, - ) - transactionCountMetric := prometheus.NewSummary(prometheus.SummaryOpts{ - Namespace: daemon.MetricsNamespace(), Subsystem: "transactions", Name: "count", - Help: "count of transactions ingested, sliding window = 10m", - Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, - }) - daemon.MetricsRegistry().MustRegister(transactionDurationMetric, transactionCountMetric) - - return &MemoryStore{ - networkPassphrase: networkPassphrase, - transactions: make(map[xdr.Hash]transaction), - transactionsByLedger: window, - transactionDurationMetric: transactionDurationMetric, - transactionCountMetric: transactionCountMetric, - } -} - -// IngestTransactions adds new transactions from the given ledger into the store. -// As a side effect, transactions which fall outside the retention window are -// removed from the store. -func (m *MemoryStore) IngestTransactions(ledgerCloseMeta xdr.LedgerCloseMeta) error { - startTime := time.Now() - reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(m.networkPassphrase, ledgerCloseMeta) - if err != nil { - return err - } - - txCount := ledgerCloseMeta.CountTransactions() - transactions := make([]transaction, txCount) - hashes := make([]xdr.Hash, 0, txCount) - hashMap := map[xdr.Hash]transaction{} - var bucket ledgerbucketwindow.LedgerBucket[[]xdr.Hash] - - for i := 0; i < txCount; i++ { - tx, err := reader.Read() - if err != nil { - return err - } - transactions[i] = transaction{ - bucket: &bucket, - feeBump: tx.Envelope.IsFeeBump(), - applicationOrder: int32(tx.Index), - successful: tx.Result.Result.Successful(), - } - if transactions[i].result, err = tx.Result.Result.MarshalBinary(); err != nil { - return err - } - if transactions[i].meta, err = tx.UnsafeMeta.MarshalBinary(); err != nil { - return err - } - if transactions[i].envelope, err = tx.Envelope.MarshalBinary(); err != nil { - return err - } - if transactions[i].feeBump { - innerHash := tx.Result.InnerHash() - hashMap[innerHash] = transactions[i] - hashes = append(hashes, innerHash) - } - hashMap[tx.Result.TransactionHash] = transactions[i] - hashes = append(hashes, tx.Result.TransactionHash) - } - bucket = ledgerbucketwindow.LedgerBucket[[]xdr.Hash]{ - LedgerSeq: ledgerCloseMeta.LedgerSequence(), - LedgerCloseTimestamp: int64(ledgerCloseMeta.LedgerHeaderHistoryEntry().Header.ScpValue.CloseTime), - BucketContent: hashes, - } - - m.lock.Lock() - defer m.lock.Unlock() - evicted, err := m.transactionsByLedger.Append(bucket) - if err != nil { - return err - } - if evicted != nil { - // garbage-collect evicted entries - for _, evictedTxHash := range evicted.BucketContent { - delete(m.transactions, evictedTxHash) - } - } - for hash, tx := range hashMap { - m.transactions[hash] = tx - } - m.transactionDurationMetric.With(prometheus.Labels{"operation": "ingest"}).Observe(time.Since(startTime).Seconds()) - m.transactionCountMetric.Observe(float64(txCount)) - return nil -} - -type Transaction struct { - Result []byte // XDR encoded xdr.TransactionResult - Meta []byte // XDR encoded xdr.TransactionMeta - Envelope []byte // XDR encoded xdr.TransactionEnvelope - Events [][]byte // XDR encoded xdr.DiagnosticEvent - FeeBump bool - ApplicationOrder int32 - Successful bool - Ledger ledgerbucketwindow.LedgerInfo -} - -// GetLedgerRange returns the first and latest ledger available in the store. -func (m *MemoryStore) GetLedgerRange() ledgerbucketwindow.LedgerRange { - m.lock.RLock() - defer m.lock.RUnlock() - return m.transactionsByLedger.GetLedgerRange() -} - -// GetTransaction obtains a transaction from the store and whether it's present and the current store range -func (m *MemoryStore) GetTransaction(hash xdr.Hash) (Transaction, bool, ledgerbucketwindow.LedgerRange) { - startTime := time.Now() - m.lock.RLock() - defer m.lock.RUnlock() - storeRange := m.transactionsByLedger.GetLedgerRange() - internalTx, ok := m.transactions[hash] - if !ok { - return Transaction{}, false, storeRange - } - - var txMeta xdr.TransactionMeta - err := txMeta.UnmarshalBinary(internalTx.meta) - if err != nil { - return Transaction{}, false, storeRange - } - - txEvents, err := txMeta.GetDiagnosticEvents() - if err != nil { - return Transaction{}, false, storeRange - } - - events := make([][]byte, 0, len(txEvents)) - - for _, e := range txEvents { - diagnosticEventXDR, err := e.MarshalBinary() - if err != nil { - return Transaction{}, false, storeRange - } - events = append(events, diagnosticEventXDR) - } - - tx := Transaction{ - Result: internalTx.result, - Meta: internalTx.meta, - Envelope: internalTx.envelope, - Events: events, - FeeBump: internalTx.feeBump, - Successful: internalTx.successful, - ApplicationOrder: internalTx.applicationOrder, - Ledger: ledgerbucketwindow.LedgerInfo{ - Sequence: internalTx.bucket.LedgerSeq, - CloseTime: internalTx.bucket.LedgerCloseTimestamp, - }, - } - - m.transactionDurationMetric.With(prometheus.Labels{"operation": "get"}).Observe(time.Since(startTime).Seconds()) - return tx, true, storeRange -} diff --git a/cmd/soroban-rpc/internal/transactions/transactions_test.go b/cmd/soroban-rpc/internal/transactions/transactions_test.go deleted file mode 100644 index 4b801835..00000000 --- a/cmd/soroban-rpc/internal/transactions/transactions_test.go +++ /dev/null @@ -1,436 +0,0 @@ -package transactions - -import ( - "encoding/hex" - "fmt" - "math" - "runtime" - "testing" - "time" - - "github.com/stellar/go/network" - "github.com/stellar/go/xdr" - "github.com/stretchr/testify/require" - - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/daemon/interfaces" - "github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow" -) - -func expectedTransaction(t *testing.T, ledger uint32, feeBump bool) Transaction { - tx := Transaction{ - FeeBump: feeBump, - ApplicationOrder: 1, - Ledger: expectedLedgerInfo(ledger), - Events: [][]byte{}, - } - var err error - tx.Result, err = transactionResult(ledger, feeBump).MarshalBinary() - require.NoError(t, err) - tx.Meta, err = xdr.TransactionMeta{ - V: 3, - Operations: &[]xdr.OperationMeta{}, - V3: &xdr.TransactionMetaV3{}, - }.MarshalBinary() - require.NoError(t, err) - tx.Envelope, err = txEnvelope(ledger, feeBump).MarshalBinary() - require.NoError(t, err) - return tx -} - -func expectedLedgerInfo(ledgerSequence uint32) ledgerbucketwindow.LedgerInfo { - return ledgerbucketwindow.LedgerInfo{ - Sequence: ledgerSequence, - CloseTime: ledgerCloseTime(ledgerSequence), - } - -} - -func expectedStoreRange(startLedger uint32, endLedger uint32) ledgerbucketwindow.LedgerRange { - return ledgerbucketwindow.LedgerRange{ - FirstLedger: expectedLedgerInfo(startLedger), - LastLedger: expectedLedgerInfo(endLedger), - } -} - -func txHash(ledgerSequence uint32, feebump bool) xdr.Hash { - envelope := txEnvelope(ledgerSequence, feebump) - hash, err := network.HashTransactionInEnvelope(envelope, "passphrase") - if err != nil { - panic(err) - } - - return hash -} - -func ledgerCloseTime(ledgerSequence uint32) int64 { - return int64(ledgerSequence)*25 + 100 -} - -func transactionResult(ledgerSequence uint32, feeBump bool) xdr.TransactionResult { - if feeBump { - return xdr.TransactionResult{ - FeeCharged: 100, - Result: xdr.TransactionResultResult{ - Code: xdr.TransactionResultCodeTxFeeBumpInnerFailed, - InnerResultPair: &xdr.InnerTransactionResultPair{ - TransactionHash: txHash(ledgerSequence, false), - Result: xdr.InnerTransactionResult{ - Result: xdr.InnerTransactionResultResult{ - Code: xdr.TransactionResultCodeTxBadSeq, - }, - }, - }, - }, - } - } - return xdr.TransactionResult{ - FeeCharged: 100, - Result: xdr.TransactionResultResult{ - Code: xdr.TransactionResultCodeTxBadSeq, - }, - } -} - -func txMeta(ledgerSequence uint32, feeBump bool) xdr.LedgerCloseMeta { - envelope := txEnvelope(ledgerSequence, feeBump) - persistentKey := xdr.ScSymbol("TEMPVAL") - contractIDBytes, _ := hex.DecodeString("df06d62447fd25da07c0135eed7557e5a5497ee7d15b7fe345bd47e191d8f577") - var contractID xdr.Hash - copy(contractID[:], contractIDBytes) - contractAddress := xdr.ScAddress{ - Type: xdr.ScAddressTypeScAddressTypeContract, - ContractId: &contractID, - } - xdrTrue := true - operationChanges := xdr.LedgerEntryChanges{ - { - Type: xdr.LedgerEntryChangeTypeLedgerEntryState, - State: &xdr.LedgerEntry{ - LastModifiedLedgerSeq: xdr.Uint32(ledgerSequence - 1), - Data: xdr.LedgerEntryData{ - Type: xdr.LedgerEntryTypeContractData, - ContractData: &xdr.ContractDataEntry{ - Contract: contractAddress, - Key: xdr.ScVal{ - Type: xdr.ScValTypeScvSymbol, - Sym: &persistentKey, - }, - Durability: xdr.ContractDataDurabilityPersistent, - Val: xdr.ScVal{ - Type: xdr.ScValTypeScvBool, - B: &xdrTrue, - }, - }, - }, - }, - }, - { - Type: xdr.LedgerEntryChangeTypeLedgerEntryUpdated, - Updated: &xdr.LedgerEntry{ - LastModifiedLedgerSeq: xdr.Uint32(ledgerSequence - 1), - Data: xdr.LedgerEntryData{ - Type: xdr.LedgerEntryTypeContractData, - ContractData: &xdr.ContractDataEntry{ - Contract: xdr.ScAddress{ - Type: xdr.ScAddressTypeScAddressTypeContract, - ContractId: &contractID, - }, - Key: xdr.ScVal{ - Type: xdr.ScValTypeScvSymbol, - Sym: &persistentKey, - }, - Durability: xdr.ContractDataDurabilityPersistent, - Val: xdr.ScVal{ - Type: xdr.ScValTypeScvBool, - B: &xdrTrue, - }, - }, - }, - }, - }, - } - txProcessing := []xdr.TransactionResultMeta{ - { - TxApplyProcessing: xdr.TransactionMeta{ - V: 3, - Operations: &[]xdr.OperationMeta{ - { - Changes: operationChanges, - }, - }, - V3: &xdr.TransactionMetaV3{}, - }, - Result: xdr.TransactionResultPair{ - TransactionHash: txHash(ledgerSequence, feeBump), - Result: transactionResult(ledgerSequence, feeBump), - }, - }, - } - - components := []xdr.TxSetComponent{ - { - Type: xdr.TxSetComponentTypeTxsetCompTxsMaybeDiscountedFee, - TxsMaybeDiscountedFee: &xdr.TxSetComponentTxsMaybeDiscountedFee{ - BaseFee: nil, - Txs: []xdr.TransactionEnvelope{ - envelope, - }, - }, - }, - } - return xdr.LedgerCloseMeta{ - V: 1, - V1: &xdr.LedgerCloseMetaV1{ - LedgerHeader: xdr.LedgerHeaderHistoryEntry{ - Header: xdr.LedgerHeader{ - ScpValue: xdr.StellarValue{ - CloseTime: xdr.TimePoint(ledgerCloseTime(ledgerSequence)), - }, - LedgerSeq: xdr.Uint32(ledgerSequence), - }, - }, - TxProcessing: txProcessing, - TxSet: xdr.GeneralizedTransactionSet{ - V: 1, - V1TxSet: &xdr.TransactionSetV1{ - PreviousLedgerHash: xdr.Hash{1}, - Phases: []xdr.TransactionPhase{ - { - V: 0, - V0Components: &components, - }, - }, - }, - }, - }, - } -} - -func txMetaWithEvents(ledgerSequence uint32, feeBump bool) xdr.LedgerCloseMeta { - tx := txMeta(ledgerSequence, feeBump) - contractIDBytes, _ := hex.DecodeString("df06d62447fd25da07c0135eed7557e5a5497ee7d15b7fe345bd47e191d8f577") - var contractID xdr.Hash - copy(contractID[:], contractIDBytes) - counter := xdr.ScSymbol("COUNTER") - - tx.V1.TxProcessing[0].TxApplyProcessing.V3 = &xdr.TransactionMetaV3{ - SorobanMeta: &xdr.SorobanTransactionMeta{ - Events: []xdr.ContractEvent{{ - ContractId: &contractID, - Type: xdr.ContractEventTypeContract, - Body: xdr.ContractEventBody{ - V: 0, - V0: &xdr.ContractEventV0{ - Topics: []xdr.ScVal{{ - Type: xdr.ScValTypeScvSymbol, - Sym: &counter, - }}, - Data: xdr.ScVal{ - Type: xdr.ScValTypeScvSymbol, - Sym: &counter, - }, - }, - }, - }}, - ReturnValue: xdr.ScVal{ - Type: xdr.ScValTypeScvSymbol, - Sym: &counter, - }, - }, - } - - return tx -} - -func txEnvelope(ledgerSequence uint32, feeBump bool) xdr.TransactionEnvelope { - var envelope xdr.TransactionEnvelope - var err error - if feeBump { - envelope, err = xdr.NewTransactionEnvelope(xdr.EnvelopeTypeEnvelopeTypeTxFeeBump, xdr.FeeBumpTransactionEnvelope{ - Tx: xdr.FeeBumpTransaction{ - Fee: 10, - FeeSource: xdr.MustMuxedAddress("MA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVAAAAAAAAAAAAAJLK"), - InnerTx: xdr.FeeBumpTransactionInnerTx{ - Type: xdr.EnvelopeTypeEnvelopeTypeTx, - V1: &xdr.TransactionV1Envelope{ - Tx: xdr.Transaction{ - Fee: 1, - SeqNum: xdr.SequenceNumber(ledgerSequence + 90), - SourceAccount: xdr.MustMuxedAddress("MA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVAAAAAAAAAAAAAJLK"), - }, - }, - }, - }, - }) - } else { - envelope, err = xdr.NewTransactionEnvelope(xdr.EnvelopeTypeEnvelopeTypeTx, xdr.TransactionV1Envelope{ - Tx: xdr.Transaction{ - Fee: 1, - SeqNum: xdr.SequenceNumber(ledgerSequence + 90), - SourceAccount: xdr.MustMuxedAddress("MA7QYNF7SOWQ3GLR2BGMZEHXAVIRZA4KVWLTJJFC7MGXUA74P7UJVAAAAAAAAAAAAAJLK"), - }, - }) - } - if err != nil { - panic(err) - } - return envelope -} - -func requirePresent(t *testing.T, store *MemoryStore, feeBump bool, ledgerSequence, firstSequence, lastSequence uint32) { - tx, ok, storeRange := store.GetTransaction(txHash(ledgerSequence, false)) - require.True(t, ok) - require.Equal(t, expectedTransaction(t, ledgerSequence, feeBump), tx) - require.Equal(t, expectedStoreRange(firstSequence, lastSequence), storeRange) - if feeBump { - tx, ok, storeRange = store.GetTransaction(txHash(ledgerSequence, true)) - require.True(t, ok) - require.Equal(t, expectedTransaction(t, ledgerSequence, feeBump), tx) - require.Equal(t, expectedStoreRange(firstSequence, lastSequence), storeRange) - } -} - -func TestIngestTransactions(t *testing.T) { - // Use a small retention window to test eviction - store := NewMemoryStore(interfaces.MakeNoOpDeamon(), "passphrase", 3) - - _, ok, storeRange := store.GetTransaction(txHash(1, false)) - require.False(t, ok) - require.Equal(t, ledgerbucketwindow.LedgerRange{}, storeRange) - - // Insert ledger 1 - require.NoError(t, store.IngestTransactions(txMeta(1, false))) - requirePresent(t, store, false, 1, 1, 1) - require.Len(t, store.transactions, 1) - - // Insert ledger 2 - require.NoError(t, store.IngestTransactions(txMeta(2, true))) - requirePresent(t, store, false, 1, 1, 2) - requirePresent(t, store, true, 2, 1, 2) - require.Len(t, store.transactions, 3) - - // Insert ledger 3 - require.NoError(t, store.IngestTransactions(txMeta(3, false))) - requirePresent(t, store, false, 1, 1, 3) - requirePresent(t, store, true, 2, 1, 3) - requirePresent(t, store, false, 3, 1, 3) - require.Len(t, store.transactions, 4) - - // Now we have filled the memory store - - // Insert ledger 4, which will cause the window to move and evict ledger 1 - require.NoError(t, store.IngestTransactions(txMeta(4, false))) - requirePresent(t, store, true, 2, 2, 4) - requirePresent(t, store, false, 3, 2, 4) - requirePresent(t, store, false, 4, 2, 4) - - _, ok, storeRange = store.GetTransaction(txHash(1, false)) - require.False(t, ok) - require.Equal(t, expectedStoreRange(2, 4), storeRange) - require.Equal(t, uint32(3), store.transactionsByLedger.Len()) - require.Len(t, store.transactions, 4) - - // Insert ledger 5, which will cause the window to move and evict ledger 2 - require.NoError(t, store.IngestTransactions(txMeta(5, false))) - requirePresent(t, store, false, 3, 3, 5) - requirePresent(t, store, false, 4, 3, 5) - requirePresent(t, store, false, 5, 3, 5) - - _, ok, storeRange = store.GetTransaction(txHash(2, false)) - require.False(t, ok) - require.Equal(t, expectedStoreRange(3, 5), storeRange) - require.Equal(t, uint32(3), store.transactionsByLedger.Len()) - require.Len(t, store.transactions, 3) - - _, ok, storeRange = store.GetTransaction(txHash(2, true)) - require.False(t, ok) - require.Equal(t, expectedStoreRange(3, 5), storeRange) - require.Equal(t, uint32(3), store.transactionsByLedger.Len()) - require.Len(t, store.transactions, 3) -} - -func TestGetTransactionsWithEventData(t *testing.T) { - store := NewMemoryStore(interfaces.MakeNoOpDeamon(), "passphrase", 100) - - // Insert ledger 1 - metaWithEvents := txMetaWithEvents(1, false) - require.NoError(t, store.IngestTransactions(metaWithEvents)) - require.Len(t, store.transactions, 1) - - // check events data - tx, ok, _ := store.GetTransaction(txHash(1, false)) - require.True(t, ok) - require.NotNil(t, tx.Events) - require.Len(t, tx.Events, 1) - - events, err := metaWithEvents.V1.TxProcessing[0].TxApplyProcessing.GetDiagnosticEvents() - require.NoError(t, err) - eventBytes, err := events[0].MarshalBinary() - require.NoError(t, err) - require.Equal(t, eventBytes, tx.Events[0]) -} - -func stableHeapInUse() int64 { - var ( - m = runtime.MemStats{} - prevInUse uint64 - prevNumGC uint32 - ) - - for { - runtime.GC() - - // Sleeping to allow GC to run a few times and collect all temporary data. - time.Sleep(100 * time.Millisecond) - - runtime.ReadMemStats(&m) - - // Considering heap stable if recent cycle collected less than 10KB. - if prevNumGC != 0 && m.NumGC > prevNumGC && math.Abs(float64(m.HeapInuse-prevInUse)) < 10*1024 { - break - } - - prevInUse = m.HeapInuse - prevNumGC = m.NumGC - } - - return int64(m.HeapInuse) -} - -func byteCountBinary(b int64) string { - const unit = 1024 - if b < unit { - return fmt.Sprintf("%d B", b) - } - div, exp := int64(unit), 0 - for n := b / unit; n >= unit; n /= unit { - div *= unit - exp++ - } - return fmt.Sprintf("%.1f %ciB", float64(b)/float64(div), "KMGTPE"[exp]) -} - -func BenchmarkIngestTransactionsMemory(b *testing.B) { - roundsNumber := uint32(b.N * 100000) - // Use a small retention window to test eviction - store := NewMemoryStore(interfaces.MakeNoOpDeamon(), "passphrase", roundsNumber) - - heapSizeBefore := stableHeapInUse() - - for i := uint32(0); i < roundsNumber; i++ { - // Insert ledger i - require.NoError(b, store.IngestTransactions(txMeta(i, false))) - } - heapSizeAfter := stableHeapInUse() - b.ReportMetric(float64(heapSizeAfter), "bytes/100k_transactions") - b.Logf("Memory consumption for %d transactions %v", roundsNumber, byteCountBinary(heapSizeAfter-heapSizeBefore)) - - // we want to generate 500*20000 transactions total, to cover the expected daily amount of transactions. - projectedTransactionCount := int64(500 * 20000) - projectedMemoryUtiliztion := (heapSizeAfter - heapSizeBefore) * projectedTransactionCount / int64(roundsNumber) - b.Logf("Projected memory consumption for %d transactions %v", projectedTransactionCount, byteCountBinary(projectedMemoryUtiliztion)) - b.ReportMetric(float64(projectedMemoryUtiliztion), "bytes/10M_transactions") - - // add another call to store to prevent the GC from collecting. - store.GetTransaction(xdr.Hash{}) -}