Re-index the events table with a combined topic/contract and ID (#510)

* Drop P22 integration tests

Author: Shaptic

.github/workflows/stellar-rpc.yml

@@ -90,14 +90,12 @@ jobs:
     strategy:
       matrix:
         os: [ ubuntu-22.04 ]
-        protocol-version: [ 22, 23 ]
+        protocol-version: [ 23 ]
     runs-on: ${{ matrix.os }}
     env:
       STELLAR_RPC_INTEGRATION_TESTS_ENABLED: true
       STELLAR_RPC_INTEGRATION_TESTS_CORE_MAX_SUPPORTED_PROTOCOL: ${{ matrix.protocol-version }}
       STELLAR_RPC_INTEGRATION_TESTS_CAPTIVE_CORE_BIN: /usr/bin/stellar-core
-      PROTOCOL_22_CORE_DEBIAN_PKG_VERSION: 23.0.1-2670.050eacf11.focal
-      PROTOCOL_22_CORE_DOCKER_IMG: stellar/stellar-core:23.0.1-2670.050eacf11.focal
       PROTOCOL_23_CORE_DEBIAN_PKG_VERSION: 23.0.1-2670.050eacf11.focal
       PROTOCOL_23_CORE_DOCKER_IMG: stellar/stellar-core:23.0.1-2670.050eacf11.focal
 

CHANGELOG.md

@@ -2,9 +2,15 @@
 
 ## Unreleased
 
+**This release includes a database migration.** This migration may take up to an hour to build new indices, depending on your hardware; please deploy accordingly.
+
 ### Added
 - The RPC's `Client` now has a new method `LoadAccount` which returns an SDK-compatible `Account` interface for a public key ([#481](https://github.com/stellar/stellar-rpc/pull/481/)).
 
+### Fixed
+- `getLedgers` performance has been improved ([#505](https://github.com/stellar/stellar-rpc/pull/505)).
+- `getEvents` performance has been drastically improved ([#510](https://github.com/stellar/stellar-rpc/pull/510)).
+
 
 ## [v23.0.2](https://github.com/stellar/stellar-rpc/compare/v23.0.1...v23.0.2)
 

Makefile

@@ -76,6 +76,9 @@ go-test: build-libs
 
 test: go-test rust-test
 
+bench:
+	go test -run=None -bench=. ./...
+
 clean:
 	cargo clean
 	go clean ./...

cmd/stellar-rpc/docker/Dockerfile

@@ -13,7 +13,7 @@ ENV RUSTUP_HOME=/rust/.rust
 ENV PATH="/usr/local/go/bin:$CARGO_HOME/bin:${PATH}"
 ENV DEBIAN_FRONTEND=noninteractive
 RUN apt-get update
-RUN apt-get install -y build-essential
+RUN apt-get install -y build-essential jq
 RUN apt-get clean
 
 RUN curl https://sh.rustup.rs -sSf | sh -s -- -y --default-toolchain $RUST_TOOLCHAIN_VERSION

cmd/stellar-rpc/internal/db/event.go

@@ -305,7 +305,7 @@ func (eventHandler *eventHandler) GetEvents(
 	contractIDs [][]byte,
 	topics NestedTopicArray,
 	eventTypes []int,
-	f ScanFunction,
+	scanner ScanFunction,
 ) error {
 	start := time.Now()
 
@@ -348,6 +348,7 @@ func (eventHandler *eventHandler) GetEvents(
 	rows, err := eventHandler.db.Query(ctx, rowQ)
 	if err != nil {
 		eventHandler.log.
+			WithError(err).
 			WithField("duration", time.Since(start)).
 			WithField("start", cursorRange.Start.String()).
 			WithField("end", cursorRange.End.String()).
@@ -363,56 +364,53 @@ func (eventHandler *eventHandler) GetEvents(
 
 	defer rows.Close()
 
-	foundRows := false
-	for rows.Next() {
-		foundRows = true
-		var row struct {
-			eventCursorID   string `db:"id"`
-			eventData       []byte `db:"event_data"`
-			transactionHash []byte `db:"transaction_hash"`
-			ledgerCloseTime int64  `db:"ledger_close_time"`
-		}
+	type rowResult struct {
+		eventCursorID   string
+		eventData       []byte
+		transactionHash []byte
+		ledgerCloseTime int64
+	}
 
-		err = rows.Scan(&row.eventCursorID, &row.eventData, &row.transactionHash, &row.ledgerCloseTime)
-		if err != nil {
-			return fmt.Errorf("failed to scan row: %w", err)
+	foundRows := 0
+	for rows.Next() {
+		foundRows++
+
+		row := rowResult{}
+		if err := rows.Scan(
+			&row.eventCursorID,
+			&row.eventData,
+			&row.transactionHash,
+			&row.ledgerCloseTime,
+		); err != nil {
+			return errors.Join(err, errors.New("failed to scan row"))
 		}
 
-		id, eventData, ledgerCloseTime := row.eventCursorID, row.eventData, row.ledgerCloseTime
+		id, ledgerCloseTime := row.eventCursorID, row.ledgerCloseTime
 		transactionHash := row.transactionHash
 		cur, err := protocol.ParseCursor(id)
 		if err != nil {
 			return errors.Join(err, errors.New("failed to parse cursor"))
 		}
 
-		var eventXDR xdr.DiagnosticEvent
-		err = xdr.SafeUnmarshal(eventData, &eventXDR)
-		if err != nil {
+		var event xdr.DiagnosticEvent
+		if err := event.UnmarshalBinary(row.eventData); err != nil {
 			return errors.Join(err, errors.New("failed to decode event"))
 		}
+
 		txHash := xdr.Hash(transactionHash)
-		if !f(eventXDR, cur, ledgerCloseTime, &txHash) {
+		if !scanner(event, cur, ledgerCloseTime, &txHash) {
 			return nil
 		}
 	}
-	if !foundRows {
-		eventHandler.log.
-			WithField("duration", time.Since(start)).
-			WithField("start", cursorRange.Start.String()).
-			WithField("end", cursorRange.End.String()).
-			WithField("contractIds", encodedContractIDs).
-			WithField("eventTypes", eventTypes).
-			WithField("Topics", topics).
-			Debugf(
-				"No events found for ledger range",
-			)
-	}
 
 	eventHandler.log.
-		WithField("startLedgerSequence", cursorRange.Start.Ledger).
-		WithField("endLedgerSequence", cursorRange.End.Ledger).
 		WithField("duration", time.Since(start)).
-		Debugf("Fetched and decoded all the events with filters - contractIDs: %v ", encodedContractIDs)
+		WithField("start", cursorRange.Start.String()).
+		WithField("end", cursorRange.End.String()).
+		WithField("contractIds", encodedContractIDs).
+		WithField("eventTypes", eventTypes).
+		WithField("Topics", topics).
+		Debugf("Found %d events for ledger range", foundRows)
 
 	return rows.Err()
 }

cmd/stellar-rpc/internal/db/sqlmigrations/06_topic_indices.sql

@@ -0,0 +1,13 @@
+-- +migrate Up
+DROP INDEX idx_contract_id;
+DROP INDEX idx_topic1;
+
+CREATE INDEX idx_id_contract_id ON events (contract_id, id);
+CREATE INDEX idx_id_topic1 ON events (topic1, id);
+
+-- +migrate Down
+DROP INDEX idx_id_contract_id;
+DROP INDEX idx_id_topic1;
+
+CREATE INDEX idx_contract_id ON events (contract_id);
+CREATE INDEX idx_topic1 ON events (topic1);

cmd/stellar-rpc/internal/methods/get_events.go

@@ -170,7 +170,7 @@ func (h eventsRPCHandler) getEvents(ctx context.Context, request protocol.GetEve
 	eventTypes := combineEventTypes(request.Filters)
 
 	// Scan function to apply filters
-	eventScanFunction := func(
+	var eventScanFunction db.ScanFunction = func(
 		event xdr.DiagnosticEvent, cursor protocol.Cursor, ledgerCloseTimestamp int64, txHash *xdr.Hash,
 	) bool {
 		if request.Matches(event) {

cmd/stellar-rpc/internal/methods/get_events_test.go

@@ -1030,15 +1030,17 @@ func BenchmarkGetEvents(b *testing.B) {
 	now := time.Now().UTC()
 
 	writer := db.NewReadWriter(log, dbx, interfaces.MakeNoOpDeamon(), 10, 10, passphrase)
-	write, err := writer.NewTx(ctx)
-	require.NoError(b, err)
-	ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
 
 	for i := range []int{1, 2, 3} {
+		write, err := writer.NewTx(ctx)
+		require.NoError(b, err)
+		ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
+
 		txMeta := getTxMetaWithContractEvents(contractID)
 		ledgerCloseMeta := ledgerCloseMetaWithEvents(uint32(i), now.Unix(), txMeta...)
 		require.NoError(b, ledgerW.InsertLedger(ledgerCloseMeta), "ingestion failed for ledger ")
 		require.NoError(b, eventW.InsertEvents(ledgerCloseMeta), "ingestion failed for events ")
+
 		require.NoError(b, write.Commit(ledgerCloseMeta, nil))
 	}