soroban-rpc: Ingest temp ledger entry evictions

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

@@ -46,6 +46,34 @@ func (s *Service) ingestLedgerEntryChanges(ctx context.Context, reader ingest.Ch
 	return ctx.Err()
 }
 
+func (s *Service) ingestTempLedgerEntryEvictions(
+	ctx context.Context,
+	evictedTempLedgerKeys []xdr.LedgerKey,
+	tx db.WriteTx,
+) error {
+	startTime := time.Now()
+	writer := tx.LedgerEntryWriter()
+	counts := map[string]int{}
+
+	for _, key := range evictedTempLedgerKeys {
+		if err := writer.DeleteLedgerEntry(key); err != nil {
+			return err
+		}
+		counts["evicted_"+key.Type.String()]++
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+	}
+
+	for evictionType, count := range counts {
+		s.ledgerStatsMetric.
+			With(prometheus.Labels{"type": evictionType}).Add(float64(count))
+	}
+	s.ingestionDurationMetric.
+		With(prometheus.Labels{"type": "evicted_temp_ledger_entries"}).Observe(time.Since(startTime).Seconds())
+	return ctx.Err()
+}
+
 func ingestLedgerEntryChange(writer db.LedgerEntryWriter, change ingest.Change) error {
 	if change.Post == nil {
 		ledgerKey, err := xdr.GetLedgerKeyFromData(change.Pre.Data)

cmd/soroban-rpc/internal/ingest/mock_db_test.go

@@ -0,0 +1,78 @@
+package ingest
+
+import (
+	"context"
+
+	"github.com/stellar/go/xdr"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/stellar/soroban-tools/cmd/soroban-rpc/internal/db"
+)
+
+var (
+	_ db.ReadWriter        = (*MockDB)(nil)
+	_ db.WriteTx           = (*MockTx)(nil)
+	_ db.LedgerEntryWriter = (*MockLedgerEntryWriter)(nil)
+	_ db.LedgerWriter      = (*MockLedgerWriter)(nil)
+)
+
+type MockDB struct {
+	mock.Mock
+}
+
+func (m MockDB) NewTx(ctx context.Context) (db.WriteTx, error) {
+	args := m.Called(ctx)
+	return args.Get(0).(db.WriteTx), args.Error(1)
+}
+
+func (m MockDB) GetLatestLedgerSequence(ctx context.Context) (uint32, error) {
+	args := m.Called(ctx)
+	return args.Get(0).(uint32), args.Error(1)
+}
+
+type MockTx struct {
+	mock.Mock
+}
+
+func (m MockTx) LedgerEntryWriter() db.LedgerEntryWriter {
+	args := m.Called()
+	return args.Get(0).(db.LedgerEntryWriter)
+}
+
+func (m MockTx) LedgerWriter() db.LedgerWriter {
+	args := m.Called()
+	return args.Get(0).(db.LedgerWriter)
+}
+
+func (m MockTx) Commit(ledgerSeq uint32) error {
+	args := m.Called(ledgerSeq)
+	return args.Error(0)
+}
+
+func (m MockTx) Rollback() error {
+	args := m.Called()
+	return args.Error(0)
+}
+
+type MockLedgerEntryWriter struct {
+	mock.Mock
+}
+
+func (m MockLedgerEntryWriter) UpsertLedgerEntry(entry xdr.LedgerEntry) error {
+	args := m.Called(entry)
+	return args.Error(0)
+}
+
+func (m MockLedgerEntryWriter) DeleteLedgerEntry(key xdr.LedgerKey) error {
+	args := m.Called(key)
+	return args.Error(0)
+}
+
+type MockLedgerWriter struct {
+	mock.Mock
+}
+
+func (m MockLedgerWriter) InsertLedger(ledger xdr.LedgerCloseMeta) error {
+	args := m.Called(ledger)
+	return args.Error(0)
+}

cmd/soroban-rpc/internal/ingest/service.go

@@ -43,6 +43,12 @@ type Config struct {
 }
 
 func NewService(cfg Config) *Service {
+	service := newService(cfg)
+	startService(service, cfg)
+	return service
+}
+
+func newService(cfg Config) *Service {
 	// ingestionDurationMetric is a metric for measuring the latency of ingestion
 	ingestionDurationMetric := prometheus.NewSummaryVec(prometheus.SummaryOpts{
 		Namespace: cfg.Daemon.MetricsNamespace(), Subsystem: "ingest", Name: "ledger_ingestion_duration_seconds",
@@ -67,7 +73,6 @@ func NewService(cfg Config) *Service {
 
 	cfg.Daemon.MetricsRegistry().MustRegister(ingestionDurationMetric, latestLedgerMetric, ledgerStatsMetric)
 
-	ctx, done := context.WithCancel(context.Background())
 	service := &Service{
 		logger:                  cfg.Logger,
 		db:                      cfg.DB,
@@ -76,11 +81,17 @@ func NewService(cfg Config) *Service {
 		ledgerBackend:           cfg.LedgerBackend,
 		networkPassPhrase:       cfg.NetworkPassPhrase,
 		timeout:                 cfg.Timeout,
-		done:                    done,
 		ingestionDurationMetric: ingestionDurationMetric,
 		latestLedgerMetric:      latestLedgerMetric,
 		ledgerStatsMetric:       ledgerStatsMetric,
 	}
+
+	return service
+}
+
+func startService(service *Service, cfg Config) {
+	ctx, done := context.WithCancel(context.Background())
+	service.done = done
 	service.wg.Add(1)
 	panicGroup := util.UnrecoverablePanicGroup.Log(cfg.Logger)
 	panicGroup.Go(func() {
@@ -104,7 +115,6 @@ func NewService(cfg Config) *Service {
 			service.logger.WithError(err).Fatal("could not run ingestion")
 		}
 	})
-	return service
 }
 
 type Service struct {
@@ -247,6 +257,16 @@ func (s *Service) ingest(ctx context.Context, sequence uint32) error {
 		return err
 	}
 
+	// EvictedTemporaryLedgerKeys will include both temporary ledger keys which
+	// have been evicted and their corresponding expiration ledger entries
+	evictedTempLedgerKeys, err := ledgerCloseMeta.EvictedTemporaryLedgerKeys()
+	if err != nil {
+		return err
+	}
+	if err := s.ingestTempLedgerEntryEvictions(ctx, evictedTempLedgerKeys, tx); err != nil {
+		return err
+	}
+
 	if err := s.ingestLedgerCloseMeta(tx, ledgerCloseMeta); err != nil {
 		return err
 	}

cmd/soroban-rpc/internal/ingest/service_test.go

@@ -2,16 +2,22 @@ package ingest
 
 import (
 	"context"
+	"encoding/hex"
 	"errors"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/stellar/go/ingest/ledgerbackend"
+	"github.com/stellar/go/network"
 	supportlog "github.com/stellar/go/support/log"
+	"github.com/stellar/go/xdr"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/stellar/soroban-tools/cmd/soroban-rpc/internal/daemon/interfaces"
 	"github.com/stellar/soroban-tools/cmd/soroban-rpc/internal/db"
+	"github.com/stellar/soroban-tools/cmd/soroban-rpc/internal/events"
+	"github.com/stellar/soroban-tools/cmd/soroban-rpc/internal/transactions"
 )
 
 type ErrorReadWriter struct {
@@ -55,3 +61,183 @@ func TestRetryRunningIngestion(t *testing.T) {
 	assert.Error(t, lastErr)
 	assert.ErrorContains(t, lastErr, "could not get latest ledger sequence")
 }
+
+func TestIngestion(t *testing.T) {
+	mockDB := &MockDB{}
+	mockLedgerBackend := &ledgerbackend.MockDatabaseBackend{}
+	daemon := interfaces.MakeNoOpDeamon()
+	config := Config{
+		Logger:            supportlog.New(),
+		DB:                mockDB,
+		EventStore:        events.NewMemoryStore(daemon, network.TestNetworkPassphrase, 1),
+		TransactionStore:  transactions.NewMemoryStore(daemon, network.TestNetworkPassphrase, 1),
+		LedgerBackend:     mockLedgerBackend,
+		Daemon:            daemon,
+		NetworkPassPhrase: network.TestNetworkPassphrase,
+	}
+	sequence := uint32(3)
+	service := newService(config)
+	mockTx := &MockTx{}
+	mockLedgerEntryWriter := &MockLedgerEntryWriter{}
+	mockLedgerWriter := &MockLedgerWriter{}
+	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()
+
+	src := xdr.MustAddress("GBXGQJWVLWOYHFLVTKWV5FGHA3LNYY2JQKM7OAJAUEQFU6LPCSEFVXON")
+	firstTx := xdr.TransactionEnvelope{
+		Type: xdr.EnvelopeTypeEnvelopeTypeTx,
+		V1: &xdr.TransactionV1Envelope{
+			Tx: xdr.Transaction{
+				Fee:           1,
+				SourceAccount: src.ToMuxedAccount(),
+			},
+		},
+	}
+	firstTxHash, err := network.HashTransactionInEnvelope(firstTx, network.TestNetworkPassphrase)
+	assert.NoError(t, err)
+
+	baseFee := xdr.Int64(100)
+	tempKey := xdr.ScSymbol("TEMPKEY")
+	persistentKey := xdr.ScSymbol("TEMPVAL")
+	contractIDBytes, err := hex.DecodeString("df06d62447fd25da07c0135eed7557e5a5497ee7d15b7fe345bd47e191d8f577")
+	assert.NoError(t, err)
+	var contractID xdr.Hash
+	copy(contractID[:], contractIDBytes)
+	contractAddress := xdr.ScAddress{
+		Type:       xdr.ScAddressTypeScAddressTypeContract,
+		ContractId: &contractID,
+	}
+	operationChanges := xdr.LedgerEntryChanges{
+		{
+			Type: xdr.LedgerEntryChangeTypeLedgerEntryState,
+			State: &xdr.LedgerEntry{
+				LastModifiedLedgerSeq: 1,
+				Data: xdr.LedgerEntryData{
+					Type: xdr.LedgerEntryTypeContractData,
+					ContractData: &xdr.ContractDataEntry{
+						Contract: contractAddress,
+						Key: xdr.ScVal{
+							Type: xdr.ScValTypeScvSymbol,
+							Sym:  &persistentKey,
+						},
+						Durability: xdr.ContractDataDurabilityPersistent,
+					},
+				},
+			},
+		},
+		{
+			Type: xdr.LedgerEntryChangeTypeLedgerEntryUpdated,
+			Updated: &xdr.LedgerEntry{
+				LastModifiedLedgerSeq: 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,
+					},
+				},
+			},
+		},
+	}
+	evictedPersistentLedgerEntry := xdr.LedgerEntry{
+		LastModifiedLedgerSeq: 123,
+		Data: xdr.LedgerEntryData{
+			Type: xdr.LedgerEntryTypeContractData,
+			ContractData: &xdr.ContractDataEntry{
+				Contract: contractAddress,
+				Key: xdr.ScVal{
+					Type: xdr.ScValTypeScvSymbol,
+					Sym:  &persistentKey,
+				},
+				Durability: xdr.ContractDataDurabilityTemporary,
+			},
+		},
+	}
+	evictedTempLedgerKey := xdr.LedgerKey{
+		Type: xdr.LedgerEntryTypeContractData,
+		ContractData: &xdr.LedgerKeyContractData{
+			Contract: contractAddress,
+			Key: xdr.ScVal{
+				Type: xdr.ScValTypeScvSymbol,
+				Sym:  &tempKey,
+			},
+			Durability: xdr.ContractDataDurabilityTemporary,
+		},
+	}
+	ledger := xdr.LedgerCloseMeta{
+		V: 2,
+		V2: &xdr.LedgerCloseMetaV2{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{Header: xdr.LedgerHeader{LedgerVersion: 10}},
+			TxSet: xdr.GeneralizedTransactionSet{
+				V: 1,
+				V1TxSet: &xdr.TransactionSetV1{
+					PreviousLedgerHash: xdr.Hash{1, 2, 3},
+					Phases: []xdr.TransactionPhase{
+						{
+							V0Components: &[]xdr.TxSetComponent{
+								{
+									Type: xdr.TxSetComponentTypeTxsetCompTxsMaybeDiscountedFee,
+									TxsMaybeDiscountedFee: &xdr.TxSetComponentTxsMaybeDiscountedFee{
+										BaseFee: &baseFee,
+										Txs: []xdr.TransactionEnvelope{
+											firstTx,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			TxProcessing: []xdr.TransactionResultMeta{
+				{
+					Result:        xdr.TransactionResultPair{TransactionHash: firstTxHash},
+					FeeProcessing: xdr.LedgerEntryChanges{},
+					TxApplyProcessing: xdr.TransactionMeta{
+						V: 3,
+						V3: &xdr.TransactionMetaV3{
+							Operations: []xdr.OperationMeta{
+								{
+									Changes: operationChanges,
+								},
+							},
+						},
+					},
+				},
+			},
+			UpgradesProcessing:             []xdr.UpgradeEntryMeta{},
+			EvictedTemporaryLedgerKeys:     []xdr.LedgerKey{evictedTempLedgerKey},
+			EvictedPersistentLedgerEntries: []xdr.LedgerEntry{evictedPersistentLedgerEntry},
+		},
+	}
+	mockLedgerBackend.On("GetLedger", ctx, sequence).
+		Return(ledger, nil).Once()
+	mockLedgerEntryWriter.On("UpsertLedgerEntry", operationChanges[1].MustUpdated()).
+		Return(nil).Once()
+	evictedPresistentLedgerKey, err := evictedPersistentLedgerEntry.LedgerKey()
+	assert.NoError(t, err)
+	mockLedgerEntryWriter.On("DeleteLedgerEntry", evictedPresistentLedgerKey).
+		Return(nil).Once()
+	mockLedgerEntryWriter.On("DeleteLedgerEntry", evictedTempLedgerKey).
+		Return(nil).Once()
+	mockLedgerWriter.On("InsertLedger", ledger).
+		Return(nil).Once()
+	assert.NoError(t, service.ingest(ctx, sequence))
+
+	mockDB.AssertExpectations(t)
+	mockTx.AssertExpectations(t)
+	mockLedgerEntryWriter.AssertExpectations(t)
+	mockLedgerWriter.AssertExpectations(t)
+	mockLedgerBackend.AssertExpectations(t)
+}

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
-	github.com/stellar/go v0.0.0-20230901002747-400590c5cce4
+	github.com/stellar/go v0.0.0-20230905170723-6e18a2f2f583
 	github.com/stretchr/testify v1.8.4
 	golang.org/x/mod v0.12.0
 )