Optimise db performance of GetLedgerRange

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

@@ -27,8 +27,7 @@ var sqlMigrations embed.FS
 var ErrEmptyDB = errors.New("DB is empty")
 
 const (
-	metaTableName               = "metadata"
-	latestLedgerSequenceMetaKey = "LatestLedgerSequence"
+	metaTableName = "metadata"
 )
 
 type ReadWriter interface {
@@ -42,13 +41,14 @@ type WriteTx interface {
 	LedgerEntryWriter() LedgerEntryWriter
 	LedgerWriter() LedgerWriter
 
-	Commit(ledgerSeq uint32) error
+	Commit(ledgerCloseMeta xdr.LedgerCloseMeta) error
 	Rollback() error
 }
 
 type dbCache struct {
-	latestLedgerSeq uint32
-	ledgerEntries   transactionalCache // Just like the DB: compress-encoded ledger key -> ledger entry XDR
+	latestLedgerSeq       uint32
+	latestLedgerCloseTime int64
+	ledgerEntries         transactionalCache // Just like the DB: compress-encoded ledger key -> ledger entry XDR
 	sync.RWMutex
 }
 
@@ -129,33 +129,37 @@ func getMetaValue(ctx context.Context, q db.SessionInterface, key string) (strin
 	case 1:
 		// expected length on an initialized DB
 	default:
-		return "", 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), key, 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
+func getLatestLedgerSequence(ctx context.Context, ledgerReader LedgerReader, cache *dbCache) (uint32, error) {
+	cache.RLock()
+	latestLedgerSeqCache := cache.latestLedgerSeq
+	cache.RUnlock()
+
+	if latestLedgerSeqCache != 0 {
+		return latestLedgerSeqCache, nil
 	}
-	latestLedger, err := strconv.ParseUint(latestLedgerStr, 10, 32)
+
+	ledgerRange, err := ledgerReader.GetLedgerRange(ctx)
 	if err != nil {
 		return 0, err
 	}
-	result := uint32(latestLedger)
 
-	// Add missing ledger sequence to the top cache.
+	// Add missing ledger sequence and close time to the top 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 the value is missing (0), otherwise
 		// we may end up overwriting the entry with an older version
-		cache.latestLedgerSeq = result
+		cache.latestLedgerSeq = ledgerRange.LastLedger.Sequence
+		cache.latestLedgerCloseTime = ledgerRange.LastLedger.CloseTime
 	}
 	cache.Unlock()
 
-	return result, nil
+	return ledgerRange.LastLedger.Sequence, nil
 }
 
 type ReadWriterMetrics struct {
@@ -216,7 +220,7 @@ func NewReadWriter(
 }
 
 func (rw *readWriter) GetLatestLedgerSequence(ctx context.Context) (uint32, error) {
-	return getLatestLedgerSequence(ctx, rw.db, rw.db.cache)
+	return getLatestLedgerSequence(ctx, NewLedgerReader(rw.db), rw.db.cache)
 }
 
 func (rw *readWriter) NewTx(ctx context.Context) (WriteTx, error) {
@@ -293,7 +297,10 @@ func (w writeTx) EventWriter() EventWriter {
 	return &w.eventWriter
 }
 
-func (w writeTx) Commit(ledgerSeq uint32) error {
+func (w writeTx) Commit(ledgerCloseMeta xdr.LedgerCloseMeta) error {
+	ledgerSeq := ledgerCloseMeta.LedgerSequence()
+	ledgerCloseTime := ledgerCloseMeta.LedgerCloseTime()
+
 	if err := w.ledgerEntryWriter.flush(); err != nil {
 		return err
 	}
@@ -309,24 +316,17 @@ func (w writeTx) Commit(ledgerSeq uint32) error {
 		return err
 	}
 
-	_, err := sq.Replace(metaTableName).
-		Values(latestLedgerSequenceMetaKey, strconv.FormatUint(uint64(ledgerSeq), 10)).
-		RunWith(w.stmtCache).
-		Exec()
-	if err != nil {
-		return err
-	}
-
 	// We need to make the cache update atomic with the transaction commit.
 	// Otherwise, the cache can be made inconsistent if a write transaction finishes
 	// in between, updating the cache in the wrong order.
 	commitAndUpdateCache := func() error {
 		w.globalCache.Lock()
 		defer w.globalCache.Unlock()
-		if err = w.tx.Commit(); err != nil {
+		if err := w.tx.Commit(); err != nil {
 			return err
 		}
 		w.globalCache.latestLedgerSeq = ledgerSeq
+		w.globalCache.latestLedgerCloseTime = ledgerCloseTime
 		w.ledgerEntryWriter.ledgerEntryCacheWriteTx.commit()
 		return nil
 	}

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

@@ -105,6 +105,39 @@ func (r ledgerReader) GetLedger(ctx context.Context, sequence uint32) (xdr.Ledge
 
 // GetLedgerRange pulls the min/max ledger sequence numbers from the meta table.
 func (r ledgerReader) GetLedgerRange(ctx context.Context) (ledgerbucketwindow.LedgerRange, error) {
+	r.db.cache.RLock()
+	latestLedgerSeqCache := r.db.cache.latestLedgerSeq
+	latestLedgerCloseTimeCache := r.db.cache.latestLedgerCloseTime
+	r.db.cache.RUnlock()
+
+	// Make use of the cached latest ledger seq and close time to query only the oldest ledger details.
+	if latestLedgerSeqCache != 0 {
+		query := sq.Select("meta").
+			From(ledgerCloseMetaTableName).
+			Where(
+				fmt.Sprintf("sequence = (SELECT MIN(sequence) FROM %s)", ledgerCloseMetaTableName),
+			)
+		var lcm []xdr.LedgerCloseMeta
+		if err := r.db.Select(ctx, &lcm, query); err != nil {
+			return ledgerbucketwindow.LedgerRange{}, fmt.Errorf("couldn't query ledger range: %w", err)
+		}
+
+		if len(lcm) == 0 {
+			return ledgerbucketwindow.LedgerRange{}, ErrEmptyDB
+		}
+
+		return ledgerbucketwindow.LedgerRange{
+			FirstLedger: ledgerbucketwindow.LedgerInfo{
+				Sequence:  lcm[0].LedgerSequence(),
+				CloseTime: lcm[0].LedgerCloseTime(),
+			},
+			LastLedger: ledgerbucketwindow.LedgerInfo{
+				Sequence:  latestLedgerSeqCache,
+				CloseTime: latestLedgerCloseTimeCache,
+			},
+		}, nil
+	}
+
 	query := sq.Select("lcm.meta").
 		From(ledgerCloseMetaTableName + " as lcm").
 		Where(sq.Or{
@@ -117,9 +150,8 @@ func (r ledgerReader) GetLedgerRange(ctx context.Context) (ledgerbucketwindow.Le
 		return ledgerbucketwindow.LedgerRange{}, fmt.Errorf("couldn't query ledger range: %w", err)
 	}
 
-	// Empty DB
 	if len(lcms) == 0 {
-		return ledgerbucketwindow.LedgerRange{}, nil
+		return ledgerbucketwindow.LedgerRange{}, ErrEmptyDB
 	}
 
 	return ledgerbucketwindow.LedgerRange{

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

@@ -81,8 +81,10 @@ func TestLedgers(t *testing.T) {
 		ledgerSequence := uint32(i)
 		tx, err := NewReadWriter(logger, db, daemon, 150, 15, passphrase).NewTx(context.Background())
 		require.NoError(t, err)
-		require.NoError(t, tx.LedgerWriter().InsertLedger(createLedger(ledgerSequence)))
-		require.NoError(t, tx.Commit(ledgerSequence))
+
+		ledgerCloseMeta := createLedger(ledgerSequence)
+		require.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+		require.NoError(t, tx.Commit(ledgerCloseMeta))
 		// rolling back after a commit is a no-op
 		require.NoError(t, tx.Rollback())
 	}
@@ -92,16 +94,18 @@ func TestLedgers(t *testing.T) {
 	ledgerSequence := uint32(11)
 	tx, err := NewReadWriter(logger, db, daemon, 150, 15, passphrase).NewTx(context.Background())
 	require.NoError(t, err)
-	require.NoError(t, tx.LedgerWriter().InsertLedger(createLedger(ledgerSequence)))
-	require.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta := createLedger(ledgerSequence)
+	require.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	require.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	assertLedgerRange(t, reader, 1, 11)
 
 	ledgerSequence = uint32(12)
 	tx, err = NewReadWriter(logger, db, daemon, 150, 5, passphrase).NewTx(context.Background())
 	require.NoError(t, err)
-	require.NoError(t, tx.LedgerWriter().InsertLedger(createLedger(ledgerSequence)))
-	require.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta = createLedger(ledgerSequence)
+	require.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	require.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	assertLedgerRange(t, reader, 8, 12)
 }
@@ -126,7 +130,7 @@ func TestGetLedgerRange_NonEmptyDB(t *testing.T) {
 		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()))
+	require.NoError(t, write.Commit(lcms[len(lcms)-1]))
 
 	reader := NewLedgerReader(db)
 	ledgerRange, err := reader.GetLedgerRange(ctx)
@@ -154,7 +158,7 @@ func TestGetLedgerRange_SingleDBRow(t *testing.T) {
 		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()))
+	require.NoError(t, write.Commit(lcms[len(lcms)-1]))
 
 	reader := NewLedgerReader(db)
 	ledgerRange, err := reader.GetLedgerRange(ctx)
@@ -171,7 +175,7 @@ func TestGetLedgerRange_EmptyDB(t *testing.T) {
 
 	reader := NewLedgerReader(db)
 	ledgerRange, err := reader.GetLedgerRange(ctx)
-	require.NoError(t, err)
+	assert.Equal(t, ErrEmptyDB, err)
 	assert.Equal(t, uint32(0), ledgerRange.FirstLedger.Sequence)
 	assert.Equal(t, int64(0), ledgerRange.FirstLedger.CloseTime)
 	assert.Equal(t, uint32(0), ledgerRange.LastLedger.Sequence)
@@ -196,7 +200,7 @@ func BenchmarkGetLedgerRange(b *testing.B) {
 		require.NoError(b, ledgerW.InsertLedger(lcm))
 		require.NoError(b, txW.InsertTransactions(lcm))
 	}
-	require.NoError(b, write.Commit(lcms[len(lcms)-1].LedgerSequence()))
+	require.NoError(b, write.Commit(lcms[len(lcms)-1]))
 	reader := NewLedgerReader(db)
 
 	b.ResetTimer()

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

@@ -140,6 +140,7 @@ type ledgerEntryReadTx struct {
 	stmtCache              *sq.StmtCache
 	latestLedgerSeqCache   uint32
 	ledgerEntryCacheReadTx *transactionalCacheReadTx
+	ledgerReader           LedgerReader
 	tx                     db.SessionInterface
 	buffer                 *xdr.EncodingBuffer
 }
@@ -148,7 +149,7 @@ func (l *ledgerEntryReadTx) GetLatestLedgerSequence() (uint32, error) {
 	if l.latestLedgerSeqCache != 0 {
 		return l.latestLedgerSeqCache, nil
 	}
-	latestLedgerSeq, err := getLatestLedgerSequence(context.Background(), l.tx, l.globalCache)
+	latestLedgerSeq, err := getLatestLedgerSequence(context.Background(), l.ledgerReader, l.globalCache)
 	if err == nil {
 		l.latestLedgerSeqCache = latestLedgerSeq
 	}
@@ -341,7 +342,7 @@ 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, NewLedgerReader(r.db), r.db.cache)
 }
 
 // NewCachedTx() caches all accessed ledger entries and select statements. If many ledger entries are accessed, it will grow without bounds.
@@ -364,22 +365,24 @@ func (r ledgerEntryReader) NewCachedTx(ctx context.Context) (LedgerEntryReadTx,
 		stmtCache:              sq.NewStmtCache(txSession.GetTx()),
 		latestLedgerSeqCache:   r.db.cache.latestLedgerSeq,
 		ledgerEntryCacheReadTx: &cacheReadTx,
+		ledgerReader:           NewLedgerReader(r.db),
 		tx:                     txSession,
 		buffer:                 xdr.NewEncodingBuffer(),
 	}, nil
 }
 
 func (r ledgerEntryReader) NewTx(ctx context.Context) (LedgerEntryReadTx, error) {
 	txSession := r.db.Clone()
+	r.db.cache.RLock()
 	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,
 		latestLedgerSeqCache: r.db.cache.latestLedgerSeq,
 		tx:                   txSession,
+		ledgerReader:         NewLedgerReader(r.db),
 		buffer:               xdr.NewEncodingBuffer(),
 	}, nil
 }

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

@@ -88,7 +88,9 @@ func TestGoldenPath(t *testing.T) {
 	assert.NoError(t, writer.UpsertLedgerEntry(expLegerEntry))
 
 	ledgerSequence := uint32(23)
-	assert.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta := createLedger(ledgerSequence)
+	assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	assert.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	present, obtainedEntry, obtainedLedgerSequence, liveUntilSeq := getLedgerEntryAndLatestLedgerSequence(t, db, key)
 	assert.True(t, present)
@@ -113,7 +115,9 @@ func TestGoldenPath(t *testing.T) {
 	assert.NoError(t, writer.UpsertLedgerEntry(entry))
 
 	ledgerSequence = uint32(24)
-	assert.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta = createLedger(ledgerSequence)
+	assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	assert.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	present, obtainedEntry, obtainedLedgerSequence, liveUntilSeq = getLedgerEntryAndLatestLedgerSequence(t, db, key)
 	assert.True(t, present)
@@ -129,7 +133,9 @@ func TestGoldenPath(t *testing.T) {
 
 	assert.NoError(t, writer.DeleteLedgerEntry(key))
 	ledgerSequence = uint32(25)
-	assert.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta = createLedger(ledgerSequence)
+	assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	assert.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	present, _, obtainedLedgerSequence, liveUntilSeq = getLedgerEntryAndLatestLedgerSequence(t, db, key)
 	assert.False(t, present)
@@ -170,7 +176,9 @@ func TestDeleteNonExistentLedgerEmpty(t *testing.T) {
 	key, _ := getContractDataLedgerEntry(t, data)
 	assert.NoError(t, writer.DeleteLedgerEntry(key))
 	ledgerSequence := uint32(23)
-	assert.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta := createLedger(ledgerSequence)
+	assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	assert.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	// Make sure that the ledger number was submitted
 	obtainedLedgerSequence, err := NewLedgerEntryReader(db).GetLatestLedgerSequence(context.Background())
@@ -271,7 +279,9 @@ func TestReadTxsDuringWriteTx(t *testing.T) {
 
 	// Finish the write transaction and check that the results are present
 	ledgerSequence := uint32(23)
-	assert.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta := createLedger(ledgerSequence)
+	assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	assert.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	obtainedLedgerSequence, err := NewLedgerEntryReader(db).GetLatestLedgerSequence(context.Background())
 	assert.NoError(t, err)
@@ -348,11 +358,11 @@ func TestWriteTxsDuringReadTxs(t *testing.T) {
 
 	// commit the write transaction
 	ledgerSequence := uint32(23)
-	assert.NoError(t, tx.Commit(ledgerSequence))
+	ledgerCloseMeta := createLedger(ledgerSequence)
+	assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+	assert.NoError(t, tx.Commit(ledgerCloseMeta))
 
 	for _, readTx := range []LedgerEntryReadTx{readTx1, readTx2, readTx3} {
-		_, err = readTx.GetLatestLedgerSequence()
-		assert.Equal(t, ErrEmptyDB, err)
 		present, _, _, err := GetLedgerEntry(readTx, key)
 		assert.NoError(t, err)
 		assert.False(t, present)
@@ -416,7 +426,9 @@ func TestConcurrentReadersAndWriter(t *testing.T) {
 				expLegerEntry := getTTLLedgerEntry(expLedgerKey)
 				assert.NoError(t, writer.UpsertLedgerEntry(expLegerEntry))
 			}
-			assert.NoError(t, tx.Commit(ledgerSequence))
+			ledgerCloseMeta := createLedger(ledgerSequence)
+			assert.NoError(t, tx.LedgerWriter().InsertLedger(ledgerCloseMeta))
+			assert.NoError(t, tx.Commit(ledgerCloseMeta))
 			logMessageCh <- fmt.Sprintf("Wrote ledger %d", ledgerSequence)
 			time.Sleep(time.Duration(rand.Int31n(30)) * time.Millisecond)
 		}
@@ -511,7 +523,7 @@ func benchmarkLedgerEntry(b *testing.B, cached bool) {
 	expLedgerKey, err := entryKeyToTTLEntryKey(key)
 	assert.NoError(b, err)
 	assert.NoError(b, tx.LedgerEntryWriter().UpsertLedgerEntry(getTTLLedgerEntry(expLedgerKey)))
-	assert.NoError(b, tx.Commit(2))
+	assert.NoError(b, tx.Commit(createLedger(2)))
 	reader := NewLedgerEntryReader(db)
 	const numQueriesPerOp = 15
 	b.ResetTimer()
@@ -570,6 +582,6 @@ func BenchmarkLedgerUpdate(b *testing.B) {
 			keyUint32 = xdr.Uint32(j)
 			assert.NoError(b, writer.UpsertLedgerEntry(entry))
 		}
-		assert.NoError(b, tx.Commit(uint32(i+1)))
+		assert.NoError(b, tx.Commit(createLedger(uint32(i+1))))
 	}
 }