Add getEvents backed by DB

cmd/soroban-rpc/internal/daemon/daemon.go

@@ -28,7 +28,6 @@
 	"github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal"
 	"github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/config"
 	"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/ingest"
 	"github.com/stellar/soroban-rpc/cmd/soroban-rpc/internal/ledgerbucketwindow"
@@ -201,13 +200,13 @@
 		}, metricsRegistry),
 	}
 
-	feewindows, eventStore := daemon.mustInitializeStorage(cfg)
+	feewindows := daemon.mustInitializeStorage(cfg)
 
 	onIngestionRetry := func(err error, dur time.Duration) {
 		logger.WithError(err).Error("could not run ingestion. Retrying")
 	}
 
-	// Take the larger of (event retention, tx retention) and then the smaller
+	// Take the largest of (event retention, tx retention) and then the smallest
 	// of (tx retention, default event retention) if event retention wasn't
 	// specified, for some reason...?
 	maxRetentionWindow := ordered.Max(cfg.EventLedgerRetentionWindow, cfg.TransactionLedgerRetentionWindow)
@@ -226,7 +225,6 @@
 			maxRetentionWindow,
 			cfg.NetworkPassphrase,
 		),
-		EventStore:        eventStore,
 		NetworkPassPhrase: cfg.NetworkPassphrase,
 		Archive:           historyArchive,
 		LedgerBackend:     core,
@@ -251,12 +249,12 @@
 
 	jsonRPCHandler := internal.NewJSONRPCHandler(cfg, internal.HandlerParams{
 		Daemon:            daemon,
-		EventStore:        eventStore,
 		FeeStatWindows:    feewindows,
 		Logger:            logger,
 		LedgerReader:      db.NewLedgerReader(dbConn),
 		LedgerEntryReader: db.NewLedgerEntryReader(dbConn),
 		TransactionReader: db.NewTransactionReader(logger, dbConn, cfg.NetworkPassphrase),
+		EventReader:       db.NewEventReader(logger, dbConn, cfg.NetworkPassphrase),
 		PreflightGetter:   preflightWorkerPool,
 	})
 
@@ -301,12 +299,8 @@
 }
 
 // mustInitializeStorage initializes the storage using what was on the DB
-func (d *Daemon) mustInitializeStorage(cfg *config.Config) (*feewindow.FeeWindows, *events.MemoryStore) {
-	eventStore := events.NewMemoryStore(
-		d,
-		cfg.NetworkPassphrase,
-		cfg.EventLedgerRetentionWindow,
-	)
+func (d *Daemon) mustInitializeStorage(cfg *config.Config) *feewindow.FeeWindows {
+
 	feewindows := feewindow.NewFeeWindows(cfg.ClassicFeeStatsLedgerRetentionWindow, cfg.SorobanFeeStatsLedgerRetentionWindow, cfg.NetworkPassphrase)
 
 	readTxMetaCtx, cancelReadTxMeta := context.WithTimeout(context.Background(), cfg.IngestionTimeout)
@@ -332,9 +326,7 @@
 				"seq": currentSeq,
 			}).Debug("still initializing in-memory store")
 		}
-		if err := eventStore.IngestEvents(txmeta); err != nil {
-			d.logger.WithError(err).Fatal("could not initialize event memory store")
-		}
+
 		if err := feewindows.IngestFees(txmeta); err != nil {
 			d.logger.WithError(err).Fatal("could not initialize fee stats")
 		}
@@ -360,7 +352,7 @@
 		}).Info("finished initializing in-memory store")
 	}
 
-	return feewindows, eventStore
+	return feewindows
 }
 
 func (d *Daemon) Run() {

cmd/soroban-rpc/internal/db/db.go

@@ -6,13 +6,16 @@ import (
 	"embed"
 	"errors"
 	"fmt"
-	"strconv"
-	"sync"
-
 	sq "github.com/Masterminds/squirrel"
 	_ "github.com/mattn/go-sqlite3"
 	"github.com/prometheus/client_golang/prometheus"
 	migrate "github.com/rubenv/sql-migrate"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"path"
+	"strconv"
+	"sync"
+	"testing"
 
 	"github.com/stellar/go/support/db"
 	"github.com/stellar/go/support/log"
@@ -305,6 +308,10 @@ func (w writeTx) Commit(ledgerSeq uint32) error {
 		return err
 	}
 
+	if err := w.eventWriter.trimEvents(ledgerSeq, w.ledgerRetentionWindow); err != nil {
+		return err
+	}
+
 	_, err := sq.Replace(metaTableName).
 		Values(latestLedgerSequenceMetaKey, strconv.FormatUint(uint64(ledgerSeq), 10)).
 		RunWith(w.stmtCache).
@@ -366,3 +373,14 @@ func runSQLMigrations(db *sql.DB, dialect string) error {
 	_, err := migrate.ExecMax(db, dialect, m, migrate.Up, 0)
 	return err
 }
+
+func NewTestDB(tb testing.TB) *DB {
+	tmp := tb.TempDir()
+	dbPath := path.Join(tmp, "db.sqlite")
+	db, err := OpenSQLiteDB(dbPath)
+	require.NoError(tb, err)
+	tb.Cleanup(func() {
+		assert.NoError(tb, db.Close())
+	})
+	return db
+}

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

@@ -1,25 +1,31 @@
 package db
 
 import (
+	"context"
+	"fmt"
+	"io"
+	"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/events"
-	"io"
 )
 
 const eventTableName = "events"
 
+// EventWriter is used during ingestion of events from LCM to DB
 type EventWriter interface {
 	InsertEvents(lcm xdr.LedgerCloseMeta) error
 }
 
+// EventReader has all the public methods to fetch events from DB
 type EventReader interface {
-	GetEvents(lcm xdr.LedgerCloseMeta) error
+	GetEvents(ctx context.Context, cursorRange events.CursorRange, contractIDs []string, f ScanFunction) error
+	// GetLedgerRange(ctx context.Context) error
 }
 
 type eventHandler struct {
@@ -30,21 +36,25 @@ type eventHandler struct {
 	ingestMetric, countMetric prometheus.Observer
 }
 
+func NewEventReader(log *log.Entry, db db.SessionInterface, passphrase string) EventReader {
+	return &eventHandler{log: log, db: db, passphrase: passphrase}
+}
+
 func (eventHandler *eventHandler) InsertEvents(lcm xdr.LedgerCloseMeta) error {
 	txCount := lcm.CountTransactions()
 
 	if eventHandler.stmtCache == nil {
-		return errors.New("EventWriter incorrectly initialized without stmtCache")
+		return fmt.Errorf("EventWriter incorrectly initialized without stmtCache")
 	} else if txCount == 0 {
 		return nil
 	}
 
 	var txReader *ingest.LedgerTransactionReader
 	txReader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(eventHandler.passphrase, lcm)
 	if err != nil {
-		return errors.Wrapf(err,
-			"failed to open transaction reader for ledger %d",
-			lcm.LedgerSequence())
+		return fmt.Errorf(
+			"failed to open transaction reader for ledger %d: %w ",
+			lcm.LedgerSequence(), err)
 	}
 	defer func() {
 		closeErr := txReader.Close()
@@ -97,3 +107,99 @@ func (eventHandler *eventHandler) InsertEvents(lcm xdr.LedgerCloseMeta) error {
 
 	return nil
 }
+
+type ScanFunction func(xdr.DiagnosticEvent, events.Cursor, int64, *xdr.Hash) bool
+
+// trimEvents removes all Events which fall outside the ledger retention window.
+func (eventHandler *eventHandler) trimEvents(latestLedgerSeq uint32, retentionWindow uint32) error {
+	if latestLedgerSeq+1 <= retentionWindow {
+		return nil
+	}
+
+	cutoff := latestLedgerSeq + 1 - retentionWindow
+	_, err := sq.StatementBuilder.
+		RunWith(eventHandler.stmtCache).
+		Delete(eventTableName).
+		Where(sq.Lt{"ledger_sequence": cutoff}).
+		Exec()
+	return err
+}
+
+// GetEvents applies f on all the events occurring in the given range with specified contract IDs if provided.
+// The events are returned in sorted ascending Cursor order.
+// If f returns false, the scan terminates early (f will not be applied on
+// remaining events in the range).
+func (eventHandler *eventHandler) GetEvents(
+	ctx context.Context,
+	cursorRange events.CursorRange,
+	contractIDs []string,
+	f ScanFunction,
+) error {
+	start := time.Now()
+
+	var rows []struct {
+		EventCursorID string              `db:"id"`
+		TxIndex       int                 `db:"application_order"`
+		Lcm           xdr.LedgerCloseMeta `db:"meta"`
+	}
+
+	rowQ := sq.
+		Select("e.id", "e.application_order", "lcm.meta").
+		From(eventTableName + " e").
+		Join(ledgerCloseMetaTableName + "lcm ON (e.ledger_sequence = lcm.sequence)").
+		Where(sq.GtOrEq{"e.id": cursorRange.Start.String()}).
+		Where(sq.Lt{"e.id": cursorRange.End.String()}).
+		OrderBy("e.id ASC")
+
+	if len(contractIDs) > 0 {
+		rowQ = rowQ.Where(sq.Eq{"e.contract_id": contractIDs})
+	}
+
+	if err := eventHandler.db.Select(ctx, &rows, rowQ); err != nil {
+		return fmt.Errorf(
+			"db read failed for start ledger cursor= %v contractIDs= %v: %w",
+			cursorRange.Start.String(),
+			contractIDs,
+			err)
+	} else if len(rows) < 1 {
+		return fmt.Errorf("no LCM found with requested event filters")
+	}
+
+	for _, row := range rows {
+		eventCursorID, txIndex, lcm := row.EventCursorID, row.TxIndex, row.Lcm
+		reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(eventHandler.passphrase, lcm)
+		if err != nil {
+			return fmt.Errorf("failed to index to tx %d in ledger %d: %w", txIndex, lcm.LedgerSequence(), err)
+		}
+
+		err = reader.Seek(txIndex - 1)
+		if err != nil {
+			return fmt.Errorf("failed to index to tx %d in ledger %d: %w", txIndex, lcm.LedgerSequence(), err)
+		}
+
+		ledgerCloseTime := lcm.LedgerCloseTime()
+		ledgerTx, err := reader.Read()
+		transactionHash := ledgerTx.Result.TransactionHash
+		diagEvents, diagErr := ledgerTx.GetDiagnosticEvents()
+
+		if diagErr != nil {
+			return fmt.Errorf("db read failed for Event Id %s: %w", eventCursorID, err)
+		}
+
+		// Find events based on filter passed in function f
+		for eventIndex, event := range diagEvents {
+			cur := events.Cursor{Ledger: lcm.LedgerSequence(), Tx: uint32(txIndex), Event: uint32(eventIndex)}
+			if f != nil && !f(event, cur, ledgerCloseTime, &transactionHash) {
+				return nil
+			}
+		}
+	}
+
+	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 ", contractIDs)
+
+	return nil
+}

cmd/soroban-rpc/internal/db/ledger_test.go

@@ -2,11 +2,9 @@ package db
 
 import (
 	"context"
-	"path"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 
 	"github.com/stellar/go/network"
 	"github.com/stellar/go/support/log"
@@ -105,14 +103,3 @@ func TestLedgers(t *testing.T) {
 
 	assertLedgerRange(t, reader, 8, 12)
 }
-
-func NewTestDB(tb testing.TB) *DB {
-	tmp := tb.TempDir()
-	dbPath := path.Join(tmp, "db.sqlite")
-	db, err := OpenSQLiteDB(dbPath)
-	require.NoError(tb, err)
-	tb.Cleanup(func() {
-		assert.NoError(tb, db.Close())
-	})
-	return db
-}