Migrate data to new transactions table

.github/workflows/golangci-lint.yml

@@ -29,7 +29,7 @@ jobs:
           make build-libpreflight
 
       - name: Run golangci-lint
-        uses: golangci/golangci-lint-action@537aa1903e5d359d0b27dbc19ddd22c5087f3fbc # version v3.2.0
+        uses: golangci/golangci-lint-action@a4f60bb28d35aeee14e6880718e0c85ff1882e64 # version v6.0.1
         with:
           version: v1.52.2 # this is the golangci-lint version
           args: --issues-exit-code=0 # exit without errors for now - won't fail the build

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

@@ -40,6 +40,7 @@ const (
 	defaultReadTimeout                    = 5 * time.Second
 	defaultShutdownGracePeriod            = 10 * time.Second
 	inMemoryInitializationLedgerLogPeriod = 1_000_000
+	transactionsTableMigrationDoneMetaKey = "TransactionsTableMigrationDone"
 )
 
 type Daemon struct {
@@ -187,49 +188,7 @@ func MustNew(cfg *config.Config) *Daemon {
 		}, metricsRegistry),
 	}
 
-	eventStore := events.NewMemoryStore(
-		daemon,
-		cfg.NetworkPassphrase,
-		cfg.EventLedgerRetentionWindow,
-	)
-	feewindows := feewindow.NewFeeWindows(cfg.ClassicFeeStatsLedgerRetentionWindow, cfg.SorobanFeeStatsLedgerRetentionWindow, cfg.NetworkPassphrase)
-
-	// initialize the stores using what was on the DB
-	readTxMetaCtx, cancelReadTxMeta := context.WithTimeout(context.Background(), cfg.IngestionTimeout)
-	defer cancelReadTxMeta()
-	// NOTE: We could optimize this to avoid unnecessary ingestion calls
-	//       (the range of txmetas can be larger than the individual store retention windows)
-	//       but it's probably not worth the pain.
-	var initialSeq uint32
-	var currentSeq uint32
-	err = db.NewLedgerReader(dbConn).StreamAllLedgers(readTxMetaCtx, func(txmeta xdr.LedgerCloseMeta) error {
-		currentSeq = txmeta.LedgerSequence()
-		if initialSeq == 0 {
-			initialSeq = currentSeq
-			logger.WithFields(supportlog.F{
-				"seq": currentSeq,
-			}).Info("initializing in-memory store")
-		} else if (currentSeq-initialSeq)%inMemoryInitializationLedgerLogPeriod == 0 {
-			logger.WithFields(supportlog.F{
-				"seq": currentSeq,
-			}).Debug("still initializing in-memory store")
-		}
-		if err := eventStore.IngestEvents(txmeta); err != nil {
-			logger.WithError(err).Fatal("could not initialize event memory store")
-		}
-		if err := feewindows.IngestFees(txmeta); err != nil {
-			logger.WithError(err).Fatal("could not initialize fee stats")
-		}
-		return nil
-	})
-	if err != nil {
-		logger.WithError(err).Fatal("could not obtain txmeta cache from the database")
-	}
-	if currentSeq != 0 {
-		logger.WithFields(supportlog.F{
-			"seq": currentSeq,
-		}).Info("finished initializing in-memory store")
-	}
+	feewindows, eventStore := daemon.mustInitializeStorage(cfg)
 
 	onIngestionRetry := func(err error, dur time.Duration) {
 		logger.WithError(err).Error("could not run ingestion. Retrying")
@@ -282,7 +241,7 @@ func MustNew(cfg *config.Config) *Daemon {
 		Logger:            logger,
 		LedgerReader:      db.NewLedgerReader(dbConn),
 		LedgerEntryReader: db.NewLedgerEntryReader(dbConn),
-		TransactionReader: db.NewTransactionReader(logger, dbConn, cfg.NetworkPassphrase),
+		TransactionReader: db.NewTransactionReader(logger, dbConn.SessionInterface, cfg.NetworkPassphrase),
 		PreflightGetter:   preflightWorkerPool,
 	})
 
@@ -317,6 +276,69 @@ func MustNew(cfg *config.Config) *Daemon {
 	return daemon
 }
 
+// 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,
+	)
+	feewindows := feewindow.NewFeeWindows(cfg.ClassicFeeStatsLedgerRetentionWindow, cfg.SorobanFeeStatsLedgerRetentionWindow, cfg.NetworkPassphrase)
+
+	readTxMetaCtx, cancelReadTxMeta := context.WithTimeout(context.Background(), cfg.IngestionTimeout)
+	defer cancelReadTxMeta()
+	var initialSeq uint32
+	var currentSeq uint32
+	dataMigrations, err := db.BuildMigrations(readTxMetaCtx, d.logger, d.db, cfg)
+	if err != nil {
+		d.logger.WithError(err).Fatal("could not build migrations")
+	}
+	// NOTE: We could optimize this to avoid unnecessary ingestion calls
+	//       (the range of txmetas can be larger than the individual store retention windows)
+	//       but it's probably not worth the pain.
+	err = db.NewLedgerReader(d.db).StreamAllLedgers(readTxMetaCtx, func(txmeta xdr.LedgerCloseMeta) error {
+		currentSeq = txmeta.LedgerSequence()
+		if initialSeq == 0 {
+			initialSeq = currentSeq
+			d.logger.WithFields(supportlog.F{
+				"seq": currentSeq,
+			}).Info("initializing in-memory store")
+		} else if (currentSeq-initialSeq)%inMemoryInitializationLedgerLogPeriod == 0 {
+			d.logger.WithFields(supportlog.F{
+				"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")
+		}
+		// TODO: clean up once we remove the in-memory storage.
+		//       (we should only stream over the required range)
+		if r := dataMigrations.ApplicableRange(); r.IsLedgerIncluded(currentSeq) {
+			if err := dataMigrations.Apply(readTxMetaCtx, txmeta); err != nil {
+				d.logger.WithError(err).Fatal("could not run migrations")
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		d.logger.WithError(err).Fatal("could not obtain txmeta cache from the database")
+	}
+	if err := dataMigrations.Commit(readTxMetaCtx); err != nil {
+		d.logger.WithError(err).Fatal("could not commit data migrations")
+	}
+
+	if currentSeq != 0 {
+		d.logger.WithFields(supportlog.F{
+			"seq": currentSeq,
+		}).Info("finished initializing in-memory store")
+	}
+
+	return feewindows, eventStore
+}
+
 func (d *Daemon) Run() {
 	d.logger.WithFields(supportlog.F{
 		"addr": d.server.Addr,

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

@@ -17,11 +17,12 @@ import (
 	"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
-var migrations embed.FS
+//go:embed sqlmigrations/*.sql
+var sqlMigrations embed.FS
 
 var ErrEmptyDB = errors.New("DB is empty")
 
@@ -52,7 +53,14 @@ type dbCache struct {
 
 type DB struct {
 	db.SessionInterface
-	cache dbCache
+	cache *dbCache
+}
+
+func (db *DB) Clone() *DB {
+	return &DB{
+		SessionInterface: db.SessionInterface.Clone(),
+		cache:            db.cache,
+	}
 }
 
 func openSQLiteDB(dbFilePath string) (*db.Session, error) {
@@ -65,9 +73,9 @@ func openSQLiteDB(dbFilePath string) (*db.Session, error) {
 		return nil, errors.Wrap(err, "open failed")
 	}
 
-	if err = runMigrations(session.DB.DB, "sqlite3"); err != nil {
+	if err = runSQLMigrations(session.DB.DB, "sqlite3"); err != nil {
 		_ = session.Close()
-		return nil, errors.Wrap(err, "could not run migrations")
+		return nil, errors.Wrap(err, "could not run SQL migrations")
 	}
 	return session, nil
 }
@@ -79,7 +87,7 @@ func OpenSQLiteDBWithPrometheusMetrics(dbFilePath string, namespace string, sub
 	}
 	result := DB{
 		SessionInterface: db.RegisterMetrics(session, namespace, sub, registry),
-		cache: dbCache{
+		cache: &dbCache{
 			ledgerEntries: newTransactionalCache(),
 		},
 	}
@@ -93,28 +101,50 @@ func OpenSQLiteDB(dbFilePath string) (*DB, error) {
 	}
 	result := DB{
 		SessionInterface: session,
-		cache: dbCache{
+		cache: &dbCache{
 			ledgerEntries: newTransactionalCache(),
 		},
 	}
 	return &result, nil
 }
 
-func getLatestLedgerSequence(ctx context.Context, q db.SessionInterface, cache *dbCache) (uint32, error) {
-	sql := sq.Select("value").From(metaTableName).Where(sq.Eq{"key": latestLedgerSequenceMetaKey})
+func getMetaBool(ctx context.Context, q db.SessionInterface, key string) (bool, error) {
+	valueStr, err := getMetaValue(ctx, q, key)
+	if err != nil {
+		return false, err
+	}
+	return strconv.ParseBool(valueStr)
+}
+
+func setMetaBool(ctx context.Context, q db.SessionInterface, key string) error {
+	query := sq.Replace(metaTableName).
+		Values(key, "true")
+	_, err := q.Exec(ctx, query)
+	return err
+}
+
+func getMetaValue(ctx context.Context, q db.SessionInterface, key string) (string, error) {
+	sql := sq.Select("value").From(metaTableName).Where(sq.Eq{"key": key})
 	var results []string
 	if err := q.Select(ctx, &results, sql); err != nil {
-		return 0, err
+		return "", err
 	}
 	switch len(results) {
 	case 0:
-		return 0, ErrEmptyDB
+		return "", ErrEmptyDB
 	case 1:
 		// expected length on an initialized DB
 	default:
-		return 0, fmt.Errorf("multiple entries (%d) for key %q in table %q", len(results), latestLedgerSequenceMetaKey, metaTableName)
+		return "", fmt.Errorf("multiple entries (%d) for key %q in table %q", len(results), latestLedgerSequenceMetaKey, metaTableName)
+	}
+	return results[0], nil
+}
+
+func getLatestLedgerSequence(ctx context.Context, q db.SessionInterface, cache *dbCache) (uint32, error) {
+	latestLedgerStr, err := getMetaValue(ctx, q, latestLedgerSequenceMetaKey)
+	if err != nil {
+		return 0, err
 	}
-	latestLedgerStr := results[0]
 	latestLedger, err := strconv.ParseUint(latestLedgerStr, 10, 32)
 	if err != nil {
 		return 0, err
@@ -125,7 +155,7 @@ func getLatestLedgerSequence(ctx context.Context, q db.SessionInterface, cache *
 	// Otherwise, the write-through cache won't get updated until the first ingestion commit
 	cache.Lock()
 	if cache.latestLedgerSeq == 0 {
-		// Only update the cache if value is missing (0), otherwise
+		// Only update the cache if the value is missing (0), otherwise
 		// we may end up overwriting the entry with an older version
 		cache.latestLedgerSeq = result
 	}
@@ -192,20 +222,21 @@ func NewReadWriter(
 }
 
 func (rw *readWriter) GetLatestLedgerSequence(ctx context.Context) (uint32, error) {
-	return getLatestLedgerSequence(ctx, rw.db, &rw.db.cache)
+	return getLatestLedgerSequence(ctx, rw.db.SessionInterface, rw.db.cache)
 }
 
 func (rw *readWriter) NewTx(ctx context.Context) (WriteTx, error) {
-	txSession := rw.db.Clone()
+	txSession := rw.db.SessionInterface.Clone()
 	if err := txSession.Begin(ctx); err != nil {
 		return nil, err
 	}
 	stmtCache := sq.NewStmtCache(txSession.GetTx())
 
 	db := rw.db
 	writer := writeTx{
-		globalCache: &db.cache,
+		globalCache: db.cache,
 		postCommit: func() error {
+			// TODO: this is sqlite-only, it shouldn't be here
 			_, err := db.ExecRaw(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
 			return err
 		},
@@ -308,12 +339,12 @@ func (w writeTx) Rollback() error {
 	}
 }
 
-func runMigrations(db *sql.DB, dialect string) error {
+func runSQLMigrations(db *sql.DB, dialect string) error {
 	m := &migrate.AssetMigrationSource{
-		Asset: migrations.ReadFile,
+		Asset: sqlMigrations.ReadFile,
 		AssetDir: func() func(string) ([]string, error) {
 			return func(path string) ([]string, error) {
-				dirEntry, err := migrations.ReadDir(path)
+				dirEntry, err := sqlMigrations.ReadDir(path)
 				if err != nil {
 					return nil, err
 				}
@@ -325,7 +356,7 @@ func runMigrations(db *sql.DB, dialect string) error {
 				return entries, nil
 			}
 		}(),
-		Dir: "migrations",
+		Dir: "sqlmigrations",
 	}
 	_, err := migrate.ExecMax(db, dialect, m, migrate.Up, 0)
 	return err

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

@@ -10,6 +10,7 @@ import (
 	"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"
 )
 
@@ -113,8 +114,8 @@ func NewTestDB(tb testing.TB) *DB {
 		assert.NoError(tb, db.Close())
 	})
 	return &DB{
-		SessionInterface: db,
-		cache: dbCache{
+		SessionInterface: db.SessionInterface,
+		cache: &dbCache{
 			ledgerEntries: newTransactionalCache(),
 		},
 	}

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

@@ -341,12 +341,12 @@ func NewLedgerEntryReader(db *DB) LedgerEntryReader {
 }
 
 func (r ledgerEntryReader) GetLatestLedgerSequence(ctx context.Context) (uint32, error) {
-	return getLatestLedgerSequence(ctx, r.db, &r.db.cache)
+	return getLatestLedgerSequence(ctx, r.db.SessionInterface, r.db.cache)
 }
 
 // NewCachedTx() caches all accessed ledger entries and select statements. If many ledger entries are accessed, it will grow without bounds.
 func (r ledgerEntryReader) NewCachedTx(ctx context.Context) (LedgerEntryReadTx, error) {
-	txSession := r.db.Clone()
+	txSession := r.db.SessionInterface.Clone()
 	// We need to copy the cached ledger entries locally when we start the transaction
 	// since otherwise we would break the consistency between the transaction and the cache.
 
@@ -360,7 +360,7 @@ func (r ledgerEntryReader) NewCachedTx(ctx context.Context) (LedgerEntryReadTx,
 	}
 	cacheReadTx := r.db.cache.ledgerEntries.newReadTx()
 	return &ledgerEntryReadTx{
-		globalCache:            &r.db.cache,
+		globalCache:            r.db.cache,
 		stmtCache:              sq.NewStmtCache(txSession.GetTx()),
 		latestLedgerSeqCache:   r.db.cache.latestLedgerSeq,
 		ledgerEntryCacheReadTx: &cacheReadTx,
@@ -370,14 +370,14 @@ func (r ledgerEntryReader) NewCachedTx(ctx context.Context) (LedgerEntryReadTx,
 }
 
 func (r ledgerEntryReader) NewTx(ctx context.Context) (LedgerEntryReadTx, error) {
-	txSession := r.db.Clone()
+	txSession := r.db.SessionInterface.Clone()
 	if err := txSession.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}); err != nil {
 		return nil, err
 	}
 	r.db.cache.RLock()
 	defer r.db.cache.RUnlock()
 	return &ledgerEntryReadTx{
-		globalCache:          &r.db.cache,
+		globalCache:          r.db.cache,
 		latestLedgerSeqCache: r.db.cache.latestLedgerSeq,
 		tx:                   txSession,
 		buffer:               xdr.NewEncodingBuffer(),