Skip to content

Commit ingestion transactions outside the ledger-range cache lock #1036

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cjonas9 merged 11 commits into from
Oct 2, 2026
Merged

Commit ingestion transactions outside the ledger-range cache lock #1036

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
a6efc89
commit ingestion transactions outside the ledger-range cache lock
cjonas9 Sep 23, 2026
fcac5a9
publish both cache bounds after commit; getEvents validates in its ow…
cjonas9 Sep 30, 2026
77af9cc
drop the lazy-fill race guard; every commit publishes the oldest now
cjonas9 Sep 30, 2026
540d3e3
take both snapshot bounds from the snapshot; the cache only supplies …
cjonas9 Oct 2, 2026
e2e7276
prime the range cache at open and drop the lazy fills
cjonas9 Oct 2, 2026
ee6ce71
begin the read snapshot before taking the cache lock
cjonas9 Oct 2, 2026
1abad39
note the single-writer assumption behind publish ordering
cjonas9 Oct 2, 2026
d18ce67
revert the lint config change; suppress contextcheck at the two test …
cjonas9 Oct 2, 2026
224d46b
tighten the range cache: fail on an empty oldest, one cached range pe…
cjonas9 Oct 2, 2026
e101bc0
close the session when priming fails; clamp getTransaction latest to …
cjonas9 Oct 2, 2026
d922a1f
test that an event snapshot survives a trim committed underneath it
cjonas9 Oct 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 14 additions & 5 deletions cmd/stellar-rpc/internal/methods/get_events.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ type eventsRPCHandler struct {
maxLimit uint
defaultLimit uint
logger *log.Entry
ledgerReader store.LedgerReader
}

func combineContractIDs(filters []protocol.EventFilter) ([][]byte, error) {
Expand Down Expand Up @@ -128,7 +127,19 @@ func (h eventsRPCHandler) getEvents(ctx context.Context, request protocol.GetEve
}
}

ledgerRange, err := h.ledgerReader.GetLedgerRange(ctx)
// Range and scan share one snapshot, so a validated startLedger cannot be
// trimmed away between the check and the query.
readTx, err := h.dbReader.NewTx(ctx)
if err != nil {
return protocol.GetEventsResponse{}, &jrpc2.Error{
Code: jrpc2.InternalError, Message: err.Error(),
}
}
defer func() {
_ = readTx.Done()
}()

ledgerRange, err := readTx.GetLedgerRange(ctx)
if err != nil {
return protocol.GetEventsResponse{}, &jrpc2.Error{
Code: jrpc2.InternalError, Message: err.Error(),
Expand Down Expand Up @@ -231,7 +242,7 @@ func (h eventsRPCHandler) getEvents(ctx context.Context, request protocol.GetEve
return uint(len(results)) < limit, nil
}

err = h.dbReader.GetEvents(ctx, cursorRange, contractIDs, topics, eventTypes, eventViewScanFunction)
err = readTx.GetEvents(ctx, cursorRange, contractIDs, topics, eventTypes, eventViewScanFunction)
switch {
case procErr != nil:
return protocol.GetEventsResponse{}, procErr
Expand Down Expand Up @@ -398,14 +409,12 @@ func NewGetEventsHandler(
dbReader store.EventReader,
maxLimit uint,
defaultLimit uint,
ledgerReader store.LedgerReader,
) jrpc2.Handler {
eventsHandler := eventsRPCHandler{
dbReader: dbReader,
maxLimit: maxLimit,
defaultLimit: defaultLimit,
logger: logger,
ledgerReader: ledgerReader,
}
return NewHandler(eventsHandler.getEvents)
}
13 changes: 7 additions & 6 deletions cmd/stellar-rpc/internal/methods/get_events_differential_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,15 +230,16 @@ type legacyEventEntry struct {

// legacyGetEvents is the pre-view handler over ledger-derived events: the
// frozen matcher and renderer on what legacyScanEvents finds.
func legacyGetEvents(ctx context.Context, h eventsRPCHandler, request protocol.GetEventsRequest,
func legacyGetEvents(ctx context.Context, h eventsRPCHandler, ledgerReader store.LedgerReader,
request protocol.GetEventsRequest,
) (protocol.GetEventsResponse, error) {
if err := request.Valid(h.maxLimit); err != nil {
return protocol.GetEventsResponse{}, &jrpc2.Error{
Code: jrpc2.InvalidParams, Message: err.Error(),
}
}

ledgerRange, err := h.ledgerReader.GetLedgerRange(ctx)
ledgerRange, err := ledgerReader.GetLedgerRange(ctx)
if err != nil {
return protocol.GetEventsResponse{}, &jrpc2.Error{
Code: jrpc2.InternalError, Message: err.Error(),
Expand Down Expand Up @@ -289,7 +290,7 @@ func legacyGetEvents(ctx context.Context, h eventsRPCHandler, request protocol.G
}
}

found, err := legacyScanEvents(ctx, h.ledgerReader, cursorRange, &request, limit)
found, err := legacyScanEvents(ctx, ledgerReader, cursorRange, &request, limit)
if err != nil {
return protocol.GetEventsResponse{}, &jrpc2.Error{
Code: jrpc2.InvalidRequest, Message: err.Error(),
Expand Down Expand Up @@ -686,14 +687,14 @@ type eventsDifferential = differential[protocol.GetEventsRequest, protocol.GetEv
// handler. The small default limit lands inside the corpus.
func newEventsDifferential(testDB *sqlitedb.DB) eventsDifferential {
h := eventsRPCHandler{
dbReader: sqlitedb.NewEventReader(log.DefaultLogger, testDB, passphrase),
dbReader: sqlitedb.NewEventReader(log.DefaultLogger, testDB),
maxLimit: 10000,
defaultLimit: 10,
ledgerReader: sqlitedb.NewLedgerReader(testDB),
}
ledgerReader := sqlitedb.NewLedgerReader(testDB)
return eventsDifferential{
want: func(ctx context.Context, req protocol.GetEventsRequest) (protocol.GetEventsResponse, error) {
return legacyGetEvents(ctx, h, req)
return legacyGetEvents(ctx, h, ledgerReader, req)
},
got: h.getEvents,
}
Expand Down
36 changes: 12 additions & 24 deletions cmd/stellar-rpc/internal/methods/get_events_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ func TestGetEvents(t *testing.T) {
write, err := writer.NewTx(ctx)
require.NoError(t, err)
ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

var txMeta []xdr.TransactionMeta
txMeta = append(txMeta, transactionMetaWithEvents(
Expand All @@ -71,7 +71,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
_, err = handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand All @@ -95,7 +94,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

contractID := xdr.ContractId([32]byte{})
var txMeta []xdr.TransactionMeta
Expand Down Expand Up @@ -124,7 +123,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand Down Expand Up @@ -184,7 +182,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

var txMeta []xdr.TransactionMeta
contractIDs := []xdr.ContractId{
Expand Down Expand Up @@ -216,7 +214,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand Down Expand Up @@ -250,7 +247,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

var txMeta []xdr.TransactionMeta
contractID := xdr.ContractId([32]byte{})
Expand Down Expand Up @@ -279,7 +276,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand Down Expand Up @@ -389,7 +385,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

var txMeta []xdr.TransactionMeta
contractID := xdr.ContractId([32]byte{})
Expand Down Expand Up @@ -421,7 +417,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}

id := protocol.Cursor{Ledger: 1, Tx: 5, Op: 0, Event: 0}.String()
Expand Down Expand Up @@ -584,7 +579,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

contractID := xdr.ContractId([32]byte{})
otherContractID := xdr.ContractId([32]byte{1})
Expand Down Expand Up @@ -643,7 +638,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand Down Expand Up @@ -707,7 +701,7 @@ func TestGetEvents(t *testing.T) {
write, err := writer.NewTx(ctx)
require.NoError(t, err)
ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

contractID := xdr.ContractId([32]byte{})
txMeta := []xdr.TransactionMeta{
Expand Down Expand Up @@ -744,7 +738,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand Down Expand Up @@ -796,7 +789,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

contractID := xdr.ContractId([32]byte{})
var txMeta []xdr.TransactionMeta
Expand All @@ -821,7 +814,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
StartLedger: 1,
Expand Down Expand Up @@ -879,7 +871,7 @@ func TestGetEvents(t *testing.T) {
require.NoError(t, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

contractID := xdr.ContractId([32]byte{})
datas := []xdr.ScSymbol{
Expand Down Expand Up @@ -933,7 +925,6 @@ func TestGetEvents(t *testing.T) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}
results, err := handler.getEvents(context.TODO(), protocol.GetEventsRequest{
Pagination: &protocol.PaginationOptions{
Expand Down Expand Up @@ -1019,7 +1010,7 @@ func BenchmarkGetEventsTopicFilters(b *testing.B) {
require.NoError(b, err)

ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)

const (
totalEvents = 5000
Expand Down Expand Up @@ -1081,7 +1072,6 @@ func BenchmarkGetEventsTopicFilters(b *testing.B) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 1000,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}

req := protocol.GetEventsRequest{
Expand All @@ -1108,7 +1098,7 @@ func BenchmarkGetEvents(b *testing.B) {
ctx := context.TODO()
log := log.DefaultLogger
log.SetLevel(logrus.TraceLevel)
store := sqlitedb.NewEventReader(log, dbx, passphrase)
store := sqlitedb.NewEventReader(log, dbx)
contractID := xdr.ContractId([32]byte{})
now := time.Now().UTC()

Expand All @@ -1131,7 +1121,6 @@ func BenchmarkGetEvents(b *testing.B) {
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}

request := protocol.GetEventsRequest{
Expand Down Expand Up @@ -1246,7 +1235,7 @@ func setupTwoContractEventsHandler(t *testing.T) (eventsRPCHandler, xdr.Contract
write, err := writer.NewTx(ctx)
require.NoError(t, err)
ledgerW, eventW := write.LedgerWriter(), write.EventWriter()
store := sqlitedb.NewEventReader(logger, dbx, passphrase)
store := sqlitedb.NewEventReader(logger, dbx)

txMeta := []xdr.TransactionMeta{
transactionMetaWithEvents(
Expand Down Expand Up @@ -1274,7 +1263,6 @@ func setupTwoContractEventsHandler(t *testing.T) (eventsRPCHandler, xdr.Contract
dbReader: store,
maxLimit: 10000,
defaultLimit: 100,
ledgerReader: sqlitedb.NewLedgerReader(dbx),
}, contractA, contractB
}

Expand Down
4 changes: 4 additions & 0 deletions cmd/stellar-rpc/internal/methods/get_transaction.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,10 @@ func GetTransaction(
Message: fmt.Sprintf("unable to get ledger range: %v", err),
}
}
// The cache publishes after the commit, so a just-found tx can be newer than the cached latest.
if getTxErr == nil && tx.Ledger.Sequence > storeRange.LastLedger.Sequence {
storeRange.LastLedger = tx.Ledger
}

response := protocol.GetTransactionResponse{
LatestLedger: storeRange.LastLedger.Sequence,
Expand Down
5 changes: 2 additions & 3 deletions cmd/stellar-rpc/internal/rpcv1/daemon/daemon.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -400,7 +400,7 @@ func createJSONRPCHandler(cfg *config.Config, logger *supportlog.Entry, daemon *
Logger: logger,
LedgerReader: sqlitedb.NewLedgerReader(daemon.db),
TransactionReader: sqlitedb.NewTransactionReader(logger, daemon.db, cfg.NetworkPassphrase),
EventReader: sqlitedb.NewEventReader(logger, daemon.db, cfg.NetworkPassphrase),
EventReader: sqlitedb.NewEventReader(logger, daemon.db),
PreflightGetter: daemon.preflightWorkerPool,
DataStoreLedgerReader: dataStoreLedgerReader,
})
Expand Down Expand Up @@ -540,8 +540,7 @@ func (d *Daemon) mustBackfill(cfg *config.Config, feeWindows *feewindow.FeeWindo
d.logger.WithError(err).Fatal("failed to backfill ledgers")
}

// Clear the DB cache and fee windows so they re-populate from the database
d.db.ResetCache()
// Reset the fee windows so they re-populate from the database
feeWindows.Reset()
}

Expand Down
2 changes: 1 addition & 1 deletion cmd/stellar-rpc/internal/rpcv1/jsonrpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ func NewJSONRPCHandler(cfg *config.Config, params HandlerParams) Handler {
DataStoreLedgerReader: params.DataStoreLedgerReader,

GetEventsHandler: methods.NewGetEventsHandler(params.Logger, params.EventReader,
cfg.MaxEventsLimit, cfg.DefaultEventsLimit, params.LedgerReader),
cfg.MaxEventsLimit, cfg.DefaultEventsLimit),

NetworkPassphrase: cfg.NetworkPassphrase,
FriendbotURL: cfg.FriendbotURL,
Expand Down
Loading
Loading