diff --git a/cmd/stellar-rpc/internal/methods/get_events_differential_test.go b/cmd/stellar-rpc/internal/methods/get_events_differential_test.go index 60137a79e..4726f13b6 100644 --- a/cmd/stellar-rpc/internal/methods/get_events_differential_test.go +++ b/cmd/stellar-rpc/internal/methods/get_events_differential_test.go @@ -4,9 +4,11 @@ import ( "context" "encoding/json" "fmt" + "io" "math" "slices" "strconv" + "strings" "testing" "time" @@ -14,6 +16,7 @@ import ( "github.com/pkg/errors" "github.com/stretchr/testify/require" + "github.com/stellar/go-stellar-sdk/ingest" protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" "github.com/stellar/go-stellar-sdk/strkey" "github.com/stellar/go-stellar-sdk/support/log" @@ -29,11 +32,14 @@ import ( // matching it with the SDK's GetEventsRequest.Matches to matching and // rendering off the zero-copy DiagnosticEventView. // -// The reference below is the pre-migration path, with the SDK matcher frozen -// here so it cannot drift under a dependency bump. Its one adaptation is -// decoding the view the store now hands out. Pagination and the DB-level -// prefilters are unchanged by the migration and shared with production. -// Shared machinery lives in differential_test.go. +// The reference below is the pre-migration matcher and renderer, frozen here +// so they cannot drift under a dependency bump, fed from the ledgers rather +// than the events table: it re-derives each window's events from the raw +// LedgerCloseMeta that store.LedgerReader.ScanLedgers yields, indexed by the +// writer's own cursor rules. The two sides share only pagination. The events +// table, its prefilters and the view path stand on one side, the ledger scan +// and the decode-based extraction on the other, and each is the other's +// oracle. Shared machinery lives in differential_test.go. // // ---- the frozen matcher (protocol.GetEventsRequest.Matches and below) ---- @@ -222,9 +228,8 @@ type legacyEventEntry struct { txHash *xdr.Hash } -// legacyGetEvents is the pre-view handler: each event decoded, then the frozen matcher and renderer. -// -//nolint:cyclop,funlen // frozen reference; it mirrors the production handler's shape by design +// 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, ) (protocol.GetEventsResponse, error) { if err := request.Valid(h.maxLimit); err != nil { @@ -271,39 +276,20 @@ func legacyGetEvents(ctx context.Context, h eventsRPCHandler, request protocol.G } } - found := make([]legacyEventEntry, 0, limit) - - contractIDs, err := combineContractIDs(request.Filters) - if err != nil { + // The filter validation stays for error parity; the prefilters it builds are + // production's alone, since the reference reads ledgers, not the events table. + if _, err := combineContractIDs(request.Filters); err != nil { return protocol.GetEventsResponse{}, &jrpc2.Error{ Code: jrpc2.InvalidParams, Message: err.Error(), } } - - topics, err := combineTopics(request.Filters) - if err != nil { + if _, err := combineTopics(request.Filters); err != nil { return protocol.GetEventsResponse{}, &jrpc2.Error{ Code: jrpc2.InvalidParams, Message: err.Error(), } } - eventTypes := combineEventTypes(request.Filters) - - var scan store.ViewScanFunction = func( - eventView xdr.DiagnosticEventView, cursor protocol.Cursor, ledgerCloseTimestamp int64, txHash *xdr.Hash, - ) (bool, error) { - // The store hands out views now; decode into the struct the old path received. - var event xdr.DiagnosticEvent - if err := event.UnmarshalBinary([]byte(eventView)); err != nil { - return false, err - } - if legacyRequestMatches(&request, event) { - found = append(found, legacyEventEntry{cursor, ledgerCloseTimestamp, event, txHash}) - } - return uint(len(found)) < limit, nil - } - - err = h.dbReader.GetEvents(ctx, cursorRange, contractIDs, topics, eventTypes, scan) + found, err := legacyScanEvents(ctx, h.ledgerReader, cursorRange, &request, limit) if err != nil { return protocol.GetEventsResponse{}, &jrpc2.Error{ Code: jrpc2.InvalidRequest, Message: err.Error(), @@ -346,6 +332,110 @@ func legacyGetEvents(ctx context.Context, h eventsRPCHandler, request protocol.G }, nil } +// legacyScanEvents re-derives the window's events from the ledgers: one +// ScanLedgers pass, each ledger decoded inside the loop (its bytes are on +// loan) and indexed by legacyLedgerEvents, then the frozen matcher, up to limit. +func legacyScanEvents( + ctx context.Context, reader store.LedgerReader, window protocol.CursorRange, + request *protocol.GetEventsRequest, limit uint, +) ([]legacyEventEntry, error) { + found := make([]legacyEventEntry, 0, limit) + if window.End.Ledger <= window.Start.Ledger { + return found, nil + } + // The events table is keyed by cursor strings, so the window is the same compare. + first, end := window.Start.String(), window.End.String() + for ledger, err := range reader.ScanLedgers(ctx, window.Start.Ledger, window.End.Ledger-1) { + if err != nil { + return nil, err + } + var lcm xdr.LedgerCloseMeta + if err := lcm.UnmarshalBinary(ledger.Raw); err != nil { + return nil, err + } + entries, err := legacyLedgerEvents(lcm) + if err != nil { + return nil, err + } + for _, entry := range entries { + id := entry.cursor.String() + if id < first || id >= end || !legacyRequestMatches(request, entry.event) { + continue + } + found = append(found, entry) + if uint(len(found)) >= limit { + return found, nil + } + } + } + return found, nil +} + +// legacyLedgerEvents is the writer's indexing (sqlitedb InsertEvents) over a +// decoded ledger: every transaction's events, failed ones included, wrapped as +// the DiagnosticEvent the table stores, in id order. +func legacyLedgerEvents(lcm xdr.LedgerCloseMeta) ([]legacyEventEntry, error) { + if lcm.CountTransactions() == 0 { + return nil, nil + } + reader, err := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta(passphrase, lcm) + if err != nil { + return nil, err + } + defer reader.Close() + + seq, closeTime := lcm.LedgerSequence(), lcm.LedgerCloseTime() + var entries []legacyEventEntry + var beforeAll, afterAll uint32 // per-ledger stage counters; afterTx restarts per transaction + for { + tx, err := reader.Read() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return nil, err + } + all, err := tx.GetTransactionEvents() + if err != nil { + return nil, err + } + add := func(cursor protocol.Cursor, event xdr.ContractEvent) { + entries = append(entries, legacyEventEntry{ + cursor: cursor, + ledgerCloseTimestamp: closeTime, + event: xdr.DiagnosticEvent{InSuccessfulContractCall: tx.Successful(), Event: event}, + txHash: &tx.Hash, + }) + } + var afterTx uint32 + for _, event := range all.TransactionEvents { + txIdx, opIdx, err := store.StageSentinels(event.Stage, tx.Index) + if err != nil { + return nil, err + } + counter := &afterTx + switch event.Stage { + case xdr.TransactionEventStageTransactionEventStageBeforeAllTxs: + counter = &beforeAll + case xdr.TransactionEventStageTransactionEventStageAfterAllTxs: + counter = &afterAll + case xdr.TransactionEventStageTransactionEventStageAfterTx: // numbered within its transaction + } + add(protocol.Cursor{Ledger: seq, Tx: txIdx, Op: opIdx, Event: *counter}, event.Event) + *counter++ + } + for op, events := range all.OperationEvents { + for i, event := range events { + add(protocol.Cursor{Ledger: seq, Tx: tx.Index, Op: uint32(op), Event: uint32(i)}, event) + } + } + } + slices.SortFunc(entries, func(a, b legacyEventEntry) int { + return strings.Compare(a.cursor.String(), b.cursor.String()) + }) + return entries, nil +} + // // ---- the corpus ---- // diff --git a/cmd/stellar-rpc/internal/methods/get_latest_ledger.go b/cmd/stellar-rpc/internal/methods/get_latest_ledger.go index a646ce39d..f54f47126 100644 --- a/cmd/stellar-rpc/internal/methods/get_latest_ledger.go +++ b/cmd/stellar-rpc/internal/methods/get_latest_ledger.go @@ -27,7 +27,7 @@ func NewGetLatestLedgerHandler(ledgerReader store.LedgerReader) jrpc2.Handler { } var response protocol.GetLatestLedgerResponse var parseErr error - found, err := ledgerReader.WithLedgerRaw(ctx, latestSequence, func(raw []byte) error { + found, err := store.WithLedgerRaw(ctx, ledgerReader, latestSequence, func(raw []byte) error { response, parseErr = latestLedgerResponse(xdr.LedgerCloseMetaView(raw), latestSequence) return parseErr }) diff --git a/cmd/stellar-rpc/internal/methods/get_latest_ledger_test.go b/cmd/stellar-rpc/internal/methods/get_latest_ledger_test.go index cefecd0ff..caaabfc3f 100644 --- a/cmd/stellar-rpc/internal/methods/get_latest_ledger_test.go +++ b/cmd/stellar-rpc/internal/methods/get_latest_ledger_test.go @@ -3,6 +3,7 @@ package methods import ( "context" "errors" + "iter" "testing" "github.com/creachadair/jrpc2" @@ -36,35 +37,12 @@ func (ledgerReader *ConstantLedgerReader) NewTx(_ context.Context) (store.Ledger return nil, errors.New("mock NewTx error") } -func (ledgerReader *ConstantLedgerReader) GetLedger(_ context.Context, - sequence uint32, -) (xdr.LedgerCloseMeta, bool, error) { - return createLedger(expectedLatestLedgerHashBytes, - sequence, - expectedLatestLedgerCloseTime), - true, nil -} - -func (ledgerReader *ConstantLedgerReader) WithLedgerRaw( - _ context.Context, sequence uint32, fn store.WithLedgerRawFn, -) (bool, error) { - lcm := createLedger(expectedLatestLedgerHashBytes, - sequence, - expectedLatestLedgerCloseTime) - raw, err := lcm.MarshalBinary() - if err != nil { - return false, err - } - return true, fn(raw) -} - -func (ledgerReader *ConstantLedgerReader) StreamLedgerRange( - _ context.Context, - _ uint32, - _ uint32, - _ store.StreamLedgerFn, -) error { - return nil +func (ledgerReader *ConstantLedgerReader) ScanLedgers( + _ context.Context, start, end uint32, +) iter.Seq2[store.RawLedger, error] { + return store.ScanLedgersFrom(start, end, func(seq uint32) (xdr.LedgerCloseMeta, bool, error) { + return createLedger(expectedLatestLedgerHashBytes, seq, expectedLatestLedgerCloseTime), true, nil + }) } func MakeTxSet() xdr.GeneralizedTransactionSet { diff --git a/cmd/stellar-rpc/internal/methods/get_ledgers.go b/cmd/stellar-rpc/internal/methods/get_ledgers.go index 0bc12932e..961134a09 100644 --- a/cmd/stellar-rpc/internal/methods/get_ledgers.go +++ b/cmd/stellar-rpc/internal/methods/get_ledgers.go @@ -199,36 +199,35 @@ func (h ledgersHandler) fetchLedgers( limit := end - start + 1 result := make([]protocol.LedgerInfo, 0, limit) - addToResult := func(ledgers []store.LedgerMetadataChunk) error { - // Transform them all into JSON responses. - for _, chunk := range ledgers { - if len(result) >= int(limit) { - break + // appendLedger renders one ledger's raw LedgerCloseMeta into the page + appendLedger := func(seq uint32, raw []byte) error { + info, err := parseLedgerInfo(raw, format) + if err != nil { + return &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: fmt.Sprintf("error processing ledger %d: %v", seq, err), } + } + result = append(result, info) + return nil + } - info, err := parseLedgerInfo(chunk, format) + fetchFromLocalDB := func(start, end uint32) error { + for ledger, err := range readTx.ScanLedgers(ctx, start, end) { if err != nil { - seq, _ := xdr.LedgerCloseMetaView(chunk.Lcm).LedgerSequence() return &jrpc2.Error{ Code: jrpc2.InternalError, - Message: fmt.Sprintf("error processing ledger %d: %v", seq, err), + Message: fmt.Sprintf("error fetching ledgers from db: %v", err), } } - result = append(result, info) - } - return nil - } - - fetchFromLocalDB := func(start, end uint32) error { - ledgers, err := readTx.BatchGetLedgers(ctx, start, end) - if err != nil { - return &jrpc2.Error{ - Code: jrpc2.InternalError, - Message: fmt.Sprintf("error fetching ledgers from db: %v", err), + if len(result) >= int(limit) { + break + } + if aerr := appendLedger(ledger.Sequence, ledger.Raw); aerr != nil { + return aerr } } - - return addToResult(ledgers) + return nil } fetchFromDatastore := func(start, end uint32) error { @@ -245,16 +244,23 @@ func (h ledgersHandler) fetchLedgers( Message: fmt.Sprintf("error fetching ledgers from datastore: %v", err), } } - - // Convert deserialized structures into serialized ones. - chunks, err := metaToChunk(ledgers) - if err != nil { - return &jrpc2.Error{ - Code: jrpc2.InternalError, - Message: fmt.Sprintf("error serializing ledgers: %v", err), + // Serialize lazily so a short page never marshals the ledgers past it. + for i := range ledgers { + if len(result) >= int(limit) { + break + } + raw, merr := ledgers[i].MarshalBinary() + if merr != nil { + return &jrpc2.Error{ + Code: jrpc2.InternalError, + Message: fmt.Sprintf("error serializing ledgers: %v", merr), + } + } + if aerr := appendLedger(ledgers[i].LedgerSequence(), raw); aerr != nil { + return aerr } } - return addToResult(chunks) + return nil } var err error @@ -280,8 +286,8 @@ func (h ledgersHandler) fetchLedgers( // parseLedgerInfo extracts and formats the ledger metadata and header // information. In the error case, it returns a jrcp2.Error. -func parseLedgerInfo(ledger store.LedgerMetadataChunk, format string) (protocol.LedgerInfo, error) { - view := xdr.LedgerCloseMetaView(ledger.Lcm) +func parseLedgerInfo(raw []byte, format string) (protocol.LedgerInfo, error) { + view := xdr.LedgerCloseMetaView(raw) sequence, err := view.LedgerSequence() if err != nil { return protocol.LedgerInfo{}, err @@ -294,6 +300,15 @@ func parseLedgerInfo(ledger store.LedgerMetadataChunk, format string) (protocol. if err != nil { return protocol.LedgerInfo{}, err } + // The header is a slice of raw, not a decode: the wire format base64s it as-is. + headerView, err := view.LedgerHeader() + if err != nil { + return protocol.LedgerInfo{}, err + } + headerRaw, err := headerView.Raw() + if err != nil { + return protocol.LedgerInfo{}, err + } var hashXdr xdr.Hash copy(hashXdr[:], hash) ledgerInfo := protocol.LedgerInfo{ @@ -306,36 +321,14 @@ func parseLedgerInfo(ledger store.LedgerMetadataChunk, format string) (protocol. switch format { case protocol.FormatJSON: var convErr error - ledgerInfo.LedgerMetadataJSON, ledgerInfo.LedgerHeaderJSON, convErr = ledgerToJSON(&ledger) + ledgerInfo.LedgerMetadataJSON, ledgerInfo.LedgerHeaderJSON, convErr = ledgerToJSON(raw, headerRaw) if convErr != nil { return ledgerInfo, convErr } default: - ledgerInfo.LedgerMetadata = base64.StdEncoding.EncodeToString(ledger.Lcm) - ledgerInfo.LedgerHeader = base64.StdEncoding.EncodeToString(ledger.HeaderRaw) + ledgerInfo.LedgerMetadata = base64.StdEncoding.EncodeToString(raw) + ledgerInfo.LedgerHeader = base64.StdEncoding.EncodeToString(headerRaw) } return ledgerInfo, nil } - -func metaToChunk(meta []xdr.LedgerCloseMeta) ([]store.LedgerMetadataChunk, error) { - result := make([]store.LedgerMetadataChunk, 0, len(meta)) - for _, lcm := range meta { - raw, err := lcm.MarshalBinary() - if err != nil { - return nil, err - } - - headerRaw, err := lcm.LedgerHeaderHistoryEntry().MarshalBinary() - if err != nil { - return nil, err - } - - result = append(result, store.LedgerMetadataChunk{ - Lcm: raw, - HeaderRaw: headerRaw, - }) - } - - return result, nil -} diff --git a/cmd/stellar-rpc/internal/methods/get_ledgers_test.go b/cmd/stellar-rpc/internal/methods/get_ledgers_test.go index f13a7fb81..404488d96 100644 --- a/cmd/stellar-rpc/internal/methods/get_ledgers_test.go +++ b/cmd/stellar-rpc/internal/methods/get_ledgers_test.go @@ -2,6 +2,7 @@ package methods import ( "context" + "encoding/base64" "encoding/json" "errors" "testing" @@ -331,6 +332,18 @@ func createLedgerCloseMeta(ledgerSeq uint32) xdr.LedgerCloseMeta { } } +// rawLedgers builds what ScanLedgers yields for the given sequences. +func rawLedgers(t *testing.T, sequences []uint32) []store.RawLedger { + t.Helper() + out := make([]store.RawLedger, 0, len(sequences)) + for _, seq := range sequences { + raw, err := createLedgerCloseMeta(seq).MarshalBinary() + require.NoError(t, err) + out = append(out, store.RawLedger{Sequence: seq, Raw: raw}) + } + return out +} + func getLedgerRange(sequences []uint32) []xdr.LedgerCloseMeta { ledgers := make([]xdr.LedgerCloseMeta, 0, len(sequences)) for _, seq := range sequences { @@ -400,10 +413,8 @@ func TestGetLedgers(t *testing.T) { FirstLedger: 2, }, nil) if len(tc.expectLocal) > 0 { - ledgerChunks, err := metaToChunk(getLedgerRange(tc.expectLocal)) - require.NoError(t, err) - mockReaderTx.On("BatchGetLedgers", ctx, tc.expectLocal[0], tc.expectLocal[len(tc.expectLocal)-1]). - Return(ledgerChunks, nil) + mockReaderTx.On("ScanLedgers", ctx, tc.expectLocal[0], tc.expectLocal[len(tc.expectLocal)-1]). + Return(rawLedgers(t, tc.expectLocal), nil) } if len(tc.expectDatastore) > 0 { @@ -437,8 +448,20 @@ func TestFetchLedgersErrors(t *testing.T) { t.Run("DB error", func(t *testing.T) { mockTx := new(MockLedgerReaderTx) - mockTx.On("BatchGetLedgers", ctx, uint32(150), uint32(151)). - Return([]store.LedgerMetadataChunk(nil), errors.New("db error")) + mockTx.On("ScanLedgers", ctx, uint32(150), uint32(151)). + Return([]store.RawLedger(nil), errors.New("db error")) + + handler := ledgersHandler{} + _, err := handler.fetchLedgers(ctx, 150, 151, "default", mockTx, localRange) + require.Error(t, err) + require.Contains(t, err.Error(), "db error") + mockTx.AssertExpectations(t) + }) + + t.Run("DB error after a full page", func(t *testing.T) { + mockTx := new(MockLedgerReaderTx) + mockTx.On("ScanLedgers", ctx, uint32(150), uint32(151)). + Return(rawLedgers(t, []uint32{150, 151}), errors.New("db error")) handler := ledgersHandler{} _, err := handler.fetchLedgers(ctx, 150, 151, "default", mockTx, localRange) @@ -475,10 +498,10 @@ func TestFetchLedgersErrors(t *testing.T) { }) } -// TestGetLedgers_EmptyBatchGetLedgersResult is a regression test that ensures -// when GetLedgerRange reports data but BatchGetLedgers returns an empty slice, +// TestGetLedgers_EmptyScanResult is a regression test that ensures +// when GetLedgerRange reports data but ScanLedgers yields nothing, // getLedgers returns an empty page with a stable cursor and does not panic. -func TestGetLedgers_EmptyBatchGetLedgersResult(t *testing.T) { +func TestGetLedgers_EmptyScanResult(t *testing.T) { ctx := t.Context() t.Run("empty result with cursor", func(t *testing.T) { @@ -499,9 +522,9 @@ func TestGetLedgers_EmptyBatchGetLedgersResult(t *testing.T) { mockReader.On("NewTx", ctx).Return(mockReaderTx, nil) mockReaderTx.On("Done").Return(nil) mockReaderTx.On("GetLedgerRange", ctx).Return(localRange, nil) - // BatchGetLedgers returns empty slice even though GetLedgerRange indicates data exists - mockReaderTx.On("BatchGetLedgers", ctx, uint32(151), uint32(155)). - Return([]store.LedgerMetadataChunk{}, nil) + // ScanLedgers yields nothing even though GetLedgerRange indicates data exists + mockReaderTx.On("ScanLedgers", ctx, uint32(151), uint32(155)). + Return([]store.RawLedger{}, nil) request := protocol.GetLedgersRequest{ Pagination: &protocol.LedgerPaginationOptions{ @@ -539,9 +562,9 @@ func TestGetLedgers_EmptyBatchGetLedgersResult(t *testing.T) { mockReader.On("NewTx", ctx).Return(mockReaderTx, nil) mockReaderTx.On("Done").Return(nil) mockReaderTx.On("GetLedgerRange", ctx).Return(localRange, nil) - // BatchGetLedgers returns empty slice even though GetLedgerRange indicates data exists - mockReaderTx.On("BatchGetLedgers", ctx, uint32(100), uint32(104)). - Return([]store.LedgerMetadataChunk{}, nil) + // ScanLedgers yields nothing even though GetLedgerRange indicates data exists + mockReaderTx.On("ScanLedgers", ctx, uint32(100), uint32(104)). + Return([]store.RawLedger{}, nil) request := protocol.GetLedgersRequest{ StartLedger: 100, @@ -558,3 +581,21 @@ func TestGetLedgers_EmptyBatchGetLedgersResult(t *testing.T) { mockReaderTx.AssertExpectations(t) }) } + +// TestParseLedgerInfo_HeaderMatchesFullDecode pins that the header parseLedgerInfo slices +// off the raw bytes equals what a full decode re-marshals, on every LCM wire version. +func TestParseLedgerInfo_HeaderMatchesFullDecode(t *testing.T) { + for _, version := range []int32{0, 1, 2} { + lcm := diffLCM(t, version, 101) + raw, err := lcm.MarshalBinary() + require.NoError(t, err) + wantHeader, err := lcm.LedgerHeaderHistoryEntry().MarshalBinary() + require.NoError(t, err) + + info, err := parseLedgerInfo(raw, protocol.FormatBase64) + require.NoError(t, err, "version %d", version) + assert.Equal(t, uint32(101), info.Sequence) + assert.Equal(t, base64.StdEncoding.EncodeToString(wantHeader), info.LedgerHeader, "version %d", version) + assert.Equal(t, base64.StdEncoding.EncodeToString(raw), info.LedgerMetadata, "version %d", version) + } +} diff --git a/cmd/stellar-rpc/internal/methods/get_transactions.go b/cmd/stellar-rpc/internal/methods/get_transactions.go index 719667b16..a81704f05 100644 --- a/cmd/stellar-rpc/internal/methods/get_transactions.go +++ b/cmd/stellar-rpc/internal/methods/get_transactions.go @@ -72,39 +72,16 @@ func (h transactionsRPCHandler) initializePagination( return *start, requestCursor, limit, nil } -// readLedgerPage borrows ledgerSeq's raw LedgerCloseMeta and runs the page -// extraction inside the loan. Extraction errors are already *jrpc2.Error and -// pass through unchanged; only the read's own outcomes are classified here. -func (h transactionsRPCHandler) readLedgerPage( - ctx context.Context, ledgerSeq uint32, readTx store.LedgerReaderTx, - start toid.ID, txns *[]protocol.TransactionInfo, limit uint, format string, -) (*toid.ID, bool, error) { - var cursor *toid.ID - var done bool - var procErr error - found, err := readTx.WithLedgerRaw(ctx, ledgerSeq, func(raw []byte) error { - cursor, done, procErr = h.processTransactionsInLedger(raw, start, txns, limit, format) - return procErr - }) - switch { - case procErr != nil: - return nil, false, procErr - case err != nil: - return nil, false, &jrpc2.Error{ - Code: jrpc2.InternalError, - Message: err.Error(), - } - case !found: - return nil, false, &jrpc2.Error{ - Code: jrpc2.InvalidParams, - Message: fmt.Sprintf("database does not contain metadata for ledger: %d", ledgerSeq), - } +// missingLedger is the wire error for a ledger the scan did not yield. +func missingLedger(seq uint32) *jrpc2.Error { + return &jrpc2.Error{ + Code: jrpc2.InvalidParams, + Message: fmt.Sprintf("database does not contain metadata for ledger: %d", seq), } - return cursor, done, nil } // processTransactionsInLedger extracts the page's worth of transactions from -// raw — a marshaled LedgerCloseMeta on loan from WithLedgerRaw — through the +// raw — a marshaled LedgerCloseMeta on loan from ScanLedgers — through the // SDK's zero-copy views. Every byte field below aliases raw until it is // base64- or JSON-encoded into txns; nothing that aliases raw outlives this call. func (h transactionsRPCHandler) processTransactionsInLedger( @@ -264,25 +241,37 @@ func (h transactionsRPCHandler) getTransactionsByLedgerSequence(ctx context.Cont txns := make([]protocol.TransactionInfo, 0, limit) var done bool cursor := toid.New(0, 0, 0) - // Bound the walk the way getEvents bounds its scan (LedgerScanLimit): over a - // sparse range the response is a short page and the client pages on from the - // returned cursor, instead of the handler walking unboundedly toward the tip. - endLedger := min(int64(ledgerRange.LastLedger.Sequence), int64(start.LedgerSequence)+LedgerScanLimit-1) - for ledgerSeq := start.LedgerSequence; int64(ledgerSeq) <= endLedger; ledgerSeq++ { - if ledgerSeq < 0 { + if start.LedgerSequence < 0 { + return protocol.GetTransactionsResponse{}, &jrpc2.Error{ + Code: jrpc2.InvalidParams, + Message: "cursor ledger sequence cannot be negative", + } + } + // Cap the scan at LedgerScanLimit so a sparse range returns a short page instead of walking to the tip. + first := uint32(start.LedgerSequence) // a non-negative int32, so first+LedgerScanLimit cannot overflow + last := min(ledgerRange.LastLedger.Sequence, first+LedgerScanLimit-1) + next := first + for entry, serr := range readTx.ScanLedgers(ctx, first, last) { + if serr != nil { return protocol.GetTransactionsResponse{}, &jrpc2.Error{ - Code: jrpc2.InvalidParams, - Message: "cursor ledger sequence cannot be negative", + Code: jrpc2.InternalError, + Message: serr.Error(), } } - cursor, done, err = h.readLedgerPage( - ctx, uint32(ledgerSeq), readTx, start, &txns, limit, request.Format) + if entry.Sequence != next { + return protocol.GetTransactionsResponse{}, missingLedger(next) + } + cursor, done, err = h.processTransactionsInLedger(entry.Raw, start, &txns, limit, request.Format) if err != nil { return protocol.GetTransactionsResponse{}, err } if done { break } + next++ + } + if !done && next <= last { + return protocol.GetTransactionsResponse{}, missingLedger(next) } // A caught-up poller's cursor points at or past the tip. The walk then diff --git a/cmd/stellar-rpc/internal/methods/get_transactions_differential_test.go b/cmd/stellar-rpc/internal/methods/get_transactions_differential_test.go index b0ec747b7..26ed2983c 100644 --- a/cmd/stellar-rpc/internal/methods/get_transactions_differential_test.go +++ b/cmd/stellar-rpc/internal/methods/get_transactions_differential_test.go @@ -79,10 +79,11 @@ func legacyGetTransactionsByLedgerSequence( } } var ledger xdr.LedgerCloseMeta - found, gerr := readTx.WithLedgerRaw(ctx, uint32(ledgerSeq), ledger.UnmarshalBinary) + found, gerr := store.WithLedgerRaw(ctx, readTx, uint32(ledgerSeq), ledger.UnmarshalBinary) if gerr != nil { return protocol.GetTransactionsResponse{}, &jrpc2.Error{Code: jrpc2.InternalError, Message: gerr.Error()} - } else if !found { + } + if !found { return protocol.GetTransactionsResponse{}, &jrpc2.Error{ Code: jrpc2.InvalidParams, Message: fmt.Sprintf("database does not contain metadata for ledger: %d", ledgerSeq), diff --git a/cmd/stellar-rpc/internal/methods/get_transactions_test.go b/cmd/stellar-rpc/internal/methods/get_transactions_test.go index dd9ba5a66..b8b1114f6 100644 --- a/cmd/stellar-rpc/internal/methods/get_transactions_test.go +++ b/cmd/stellar-rpc/internal/methods/get_transactions_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "iter" "testing" "github.com/creachadair/jrpc2" @@ -377,26 +378,10 @@ func setupDBNoTxs(t *testing.T, numLedgers int) *sqlitedb.DB { } // sparseLedgerReader serves an arbitrarily wide range of empty ledgers, -// counting point reads, so a test can observe how far the handler walks. +// counting ledgers served, so a test can observe how far the handler walks. type sparseLedgerReader struct { latest uint32 - gets int -} - -func (r *sparseLedgerReader) GetLedger(_ context.Context, seq uint32) (xdr.LedgerCloseMeta, bool, error) { - r.gets++ - return createEmptyTestLedger(seq), true, nil -} - -func (r *sparseLedgerReader) WithLedgerRaw( - _ context.Context, seq uint32, fn store.WithLedgerRawFn, -) (bool, error) { - r.gets++ - raw, err := createEmptyTestLedger(seq).MarshalBinary() - if err != nil { - return false, err - } - return true, fn(raw) + served int } func (r *sparseLedgerReader) GetLedgerRange(context.Context) (store.LedgerRange, error) { @@ -406,12 +391,14 @@ func (r *sparseLedgerReader) GetLedgerRange(context.Context) (store.LedgerRange, }, nil } -func (r *sparseLedgerReader) BatchGetLedgers(context.Context, uint32, uint32) ([]store.LedgerMetadataChunk, error) { - return nil, nil -} - -func (r *sparseLedgerReader) StreamLedgerRange(context.Context, uint32, uint32, store.StreamLedgerFn) error { - return nil +// ScanLedgers yields every sequence in range. +func (r *sparseLedgerReader) ScanLedgers( + _ context.Context, start, end uint32, +) iter.Seq2[store.RawLedger, error] { + return store.ScanLedgersFrom(start, end, func(seq uint32) (xdr.LedgerCloseMeta, bool, error) { + r.served++ + return createEmptyTestLedger(seq), true, nil + }) } func (r *sparseLedgerReader) GetLatestLedgerSequence(context.Context) (uint32, error) { @@ -422,6 +409,46 @@ func (r *sparseLedgerReader) NewTx(context.Context) (store.LedgerReaderTx, error func (r *sparseLedgerReader) Done() error { return nil } +// TestGetTransactions_ScanGaps covers the two branches that replaced the not-found +// point read: a gap the scan jumps over, and a range that runs dry before its end. +func TestGetTransactions_ScanGaps(t *testing.T) { + for name, tc := range map[string]struct { + yield []uint32 // sequences the scan hands back for a requested [1, 3] + wantSeq uint32 // the ledger the error must name + }{ + "gap in the middle": {[]uint32{1, 3}, 2}, + "range runs dry": {[]uint32{1, 2}, 3}, + "nothing at all": {nil, 1}, + } { + t.Run(name, func(t *testing.T) { + ctx := t.Context() + mockReader := new(MockLedgerReader) + mockTx := new(MockLedgerReaderTx) + mockReader.On("NewTx", ctx).Return(mockTx, nil) + mockTx.On("Done").Return(nil) + mockTx.On("GetLedgerRange", ctx).Return(store.LedgerRange{ + FirstLedger: store.LedgerInfo{Sequence: 1, CloseTime: 100}, + LastLedger: store.LedgerInfo{Sequence: 3, CloseTime: 300}, + }, nil) + mockTx.On("ScanLedgers", ctx, uint32(1), uint32(3)).Return(rawLedgers(t, tc.yield), nil) + + handler := transactionsRPCHandler{ + ledgerReader: mockReader, + maxLimit: 100, + defaultLimit: 10, + networkPassphrase: NetworkPassphrase, + } + _, err := handler.getTransactionsByLedgerSequence( + ctx, protocol.GetTransactionsRequest{StartLedger: 1}) + require.Error(t, err) + assert.Equal(t, + fmt.Sprintf("[%d] database does not contain metadata for ledger: %d", + jrpc2.InvalidParams, tc.wantSeq), + err.Error()) + }) + } +} + func TestGetTransactions_SparseRangeCapsAtLedgerScanLimit(t *testing.T) { reader := &sparseLedgerReader{latest: 50_000} handler := transactionsRPCHandler{ @@ -436,7 +463,7 @@ func TestGetTransactions_SparseRangeCapsAtLedgerScanLimit(t *testing.T) { require.NoError(t, err) assert.Empty(t, response.Transactions) - assert.Equal(t, LedgerScanLimit, reader.gets, "the walk stops at the scan limit, not the latest ledger") + assert.Equal(t, LedgerScanLimit, reader.served, "the walk stops at the scan limit, not the latest ledger") assert.Equal(t, toid.New(LedgerScanLimit, 0, 1).String(), response.Cursor, "the cursor points at the last scanned ledger so the client can page on") assert.Equal(t, uint32(50_000), response.LatestLedger) diff --git a/cmd/stellar-rpc/internal/methods/json.go b/cmd/stellar-rpc/internal/methods/json.go index a173076d4..f3d667363 100644 --- a/cmd/stellar-rpc/internal/methods/json.go +++ b/cmd/stellar-rpc/internal/methods/json.go @@ -36,16 +36,16 @@ func transactionToJSON(tx store.Transaction) ( return result, envelope, resultMeta, nil } -func ledgerToJSON(chunk *store.LedgerMetadataChunk) ([]byte, []byte, error) { +func ledgerToJSON(lcm, headerRaw []byte) ([]byte, []byte, error) { var err error var closeMetaJSON, headerJSON []byte - closeMetaJSON, err = xdr2json.ConvertBytes(xdr.LedgerCloseMeta{}, chunk.Lcm) + closeMetaJSON, err = xdr2json.ConvertBytes(xdr.LedgerCloseMeta{}, lcm) if err != nil { return nil, nil, err } - headerJSON, err = xdr2json.ConvertBytes(xdr.LedgerHeaderHistoryEntry{}, chunk.HeaderRaw) + headerJSON, err = xdr2json.ConvertBytes(xdr.LedgerHeaderHistoryEntry{}, headerRaw) if err != nil { return nil, nil, err } diff --git a/cmd/stellar-rpc/internal/methods/mocks.go b/cmd/stellar-rpc/internal/methods/mocks.go index 76291e021..41f99410c 100644 --- a/cmd/stellar-rpc/internal/methods/mocks.go +++ b/cmd/stellar-rpc/internal/methods/mocks.go @@ -2,6 +2,7 @@ package methods import ( "context" + "iter" "github.com/stretchr/testify/mock" @@ -22,16 +23,17 @@ type MockLedgerReader struct { mock.Mock } +// GetLedger is the mock's stubbing surface; it is not on store.LedgerReader. func (m *MockLedgerReader) GetLedger(ctx context.Context, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { args := m.Called(ctx, sequence) return args.Get(0).(xdr.LedgerCloseMeta), args.Bool(1), args.Error(2) //nolint:forcetypeassert } -func (m *MockLedgerReader) WithLedgerRaw( - ctx context.Context, sequence uint32, fn store.WithLedgerRawFn, -) (bool, error) { - args := m.Called(ctx, sequence, fn) - return args.Bool(0), args.Error(1) +// ScanLedgers routes each sequence through GetLedger, so tests keep stubbing by sequence. +func (m *MockLedgerReader) ScanLedgers(ctx context.Context, start, end uint32) iter.Seq2[store.RawLedger, error] { + return store.ScanLedgersFrom(start, end, func(seq uint32) (xdr.LedgerCloseMeta, bool, error) { + return m.GetLedger(ctx, seq) + }) } func (m *MockLedgerReader) GetLedgerRange(ctx context.Context) (store.LedgerRange, error) { @@ -39,13 +41,6 @@ func (m *MockLedgerReader) GetLedgerRange(ctx context.Context) (store.LedgerRang return args.Get(0).(store.LedgerRange), args.Error(1) //nolint:forcetypeassert } -func (m *MockLedgerReader) StreamLedgerRange(ctx context.Context, startLedger, endLedger uint32, - f store.StreamLedgerFn, -) error { - args := m.Called(ctx, startLedger, endLedger, f) - return args.Error(0) -} - func (m *MockLedgerReader) NewTx(ctx context.Context) (store.LedgerReaderTx, error) { args := m.Called(ctx) return args.Get(0).(store.LedgerReaderTx), args.Error(1) //nolint:forcetypeassert @@ -65,17 +60,23 @@ func (m *MockLedgerReaderTx) GetLedgerRange(ctx context.Context) (store.LedgerRa return args.Get(0).(store.LedgerRange), args.Error(1) //nolint:forcetypeassert } -func (m *MockLedgerReaderTx) BatchGetLedgers(ctx context.Context, start, end uint32, -) ([]store.LedgerMetadataChunk, error) { +// ScanLedgers returns the stubbed ledgers as a stream; a non-nil error is yielded alone. +func (m *MockLedgerReaderTx) ScanLedgers( + ctx context.Context, start, end uint32, +) iter.Seq2[store.RawLedger, error] { args := m.Called(ctx, start, end) - return args.Get(0).([]store.LedgerMetadataChunk), args.Error(1) //nolint:forcetypeassert -} - -func (m *MockLedgerReaderTx) WithLedgerRaw( - ctx context.Context, sequence uint32, fn store.WithLedgerRawFn, -) (bool, error) { - args := m.Called(ctx, sequence, fn) - return args.Bool(0), args.Error(1) + ledgers := args.Get(0).([]store.RawLedger) //nolint:forcetypeassert + err := args.Error(1) + return func(yield func(store.RawLedger, error) bool) { + for _, l := range ledgers { + if !yield(l, nil) { + return + } + } + if err != nil { + yield(store.RawLedger{}, err) + } + } } func (m *MockLedgerReaderTx) Done() error { diff --git a/cmd/stellar-rpc/internal/methods/simulate_transaction.go b/cmd/stellar-rpc/internal/methods/simulate_transaction.go index f9d6a3b0c..0130ddf71 100644 --- a/cmd/stellar-rpc/internal/methods/simulate_transaction.go +++ b/cmd/stellar-rpc/internal/methods/simulate_transaction.go @@ -488,7 +488,7 @@ func getLatestLedgerPreflightInfo( ) (latestLedgerPreflightInfo, error) { // Obtain every preflight ledger field from the same meta so sequence and // close time cannot come from different ingestion snapshots. - closeMeta, ok, err := ledgerReader.GetLedger(ctx, latestLedger) + closeMeta, ok, err := store.GetLedger(ctx, ledgerReader, latestLedger) if err != nil { return latestLedgerPreflightInfo{}, err } diff --git a/cmd/stellar-rpc/internal/methods/simulate_transaction_test.go b/cmd/stellar-rpc/internal/methods/simulate_transaction_test.go index 04a4e5375..72c466b3e 100644 --- a/cmd/stellar-rpc/internal/methods/simulate_transaction_test.go +++ b/cmd/stellar-rpc/internal/methods/simulate_transaction_test.go @@ -422,8 +422,8 @@ func TestSimulateTransactionReturnsErrorForUnknownLedgerMetaVersion(t *testing.T ledgerReader := &MockLedgerReader{} handler := NewSimulateTransactionHandler(logger, ledgerReader, nil, panicPreflightGetter{}, xdr.DecodeOptions{}) - // V=99 is deliberately out of range to trigger the default error branch. - unknownVersionMeta := xdr.LedgerCloseMeta{V: 99} + // V0 is encodable but predates the versions the handler reads, so it reaches the default branch. + unknownVersionMeta := xdr.LedgerCloseMeta{V: 0, V0: &xdr.LedgerCloseMetaV0{}} ledgerReader. On("GetLatestLedgerSequence", mock.Anything). @@ -453,7 +453,7 @@ func TestSimulateTransactionReturnsErrorForUnknownLedgerMetaVersion(t *testing.T simResp, ok := resp.(protocol.SimulateTransactionResponse) require.True(t, ok) - require.Equal(t, "latest ledger (77) meta has unexpected version (99)", simResp.Error) + require.Equal(t, "latest ledger (77) meta has unexpected version (0)", simResp.Error) require.Equal(t, uint32(77), simResp.LatestLedger) ledgerReader.AssertExpectations(t) @@ -636,6 +636,8 @@ func TestSimulateTransactionThreadsUseUpgradedAuth(t *testing.T) { LedgerHeader: xdr.LedgerHeaderHistoryEntry{ Header: xdr.LedgerHeader{LedgerVersion: 23}, }, + // GeneralizedTransactionSet has no zero arm; the ledger must round-trip through bytes. + TxSet: xdr.GeneralizedTransactionSet{V: 1, V1TxSet: &xdr.TransactionSetV1{}}, TotalByteSizeOfLiveSorobanState: 100, }, } diff --git a/cmd/stellar-rpc/internal/methods/util.go b/cmd/stellar-rpc/internal/methods/util.go index 44b161ec5..347109fd6 100644 --- a/cmd/stellar-rpc/internal/methods/util.go +++ b/cmd/stellar-rpc/internal/methods/util.go @@ -17,7 +17,7 @@ func getProtocolVersion( } // obtain bucket size - closeMeta, ok, err := ledgerReader.GetLedger(ctx, latestLedger) + closeMeta, ok, err := store.GetLedger(ctx, ledgerReader, latestLedger) if err != nil { return 0, err } diff --git a/cmd/stellar-rpc/internal/methods/util_test.go b/cmd/stellar-rpc/internal/methods/util_test.go index f98e60a8d..13467b086 100644 --- a/cmd/stellar-rpc/internal/methods/util_test.go +++ b/cmd/stellar-rpc/internal/methods/util_test.go @@ -12,6 +12,7 @@ import ( "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/host" "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv1/sqlitedb" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/store" ) func BenchmarkGetProtocolVersion(b *testing.B) { @@ -19,7 +20,7 @@ func BenchmarkGetProtocolVersion(b *testing.B) { daemon := host.MakeNoOpDaemon() ledgerReader := sqlitedb.NewLedgerReader(dbx) - _, exists, err := ledgerReader.GetLedger(b.Context(), 1) + _, exists, err := store.GetLedger(b.Context(), ledgerReader, 1) require.NoError(b, err) assert.False(b, exists) @@ -43,7 +44,7 @@ func TestGetProtocolVersion(t *testing.T) { daemon := host.MakeNoOpDaemon() ledgerReader := sqlitedb.NewLedgerReader(dbx) - _, exists, err := ledgerReader.GetLedger(t.Context(), 1) + _, exists, err := store.GetLedger(t.Context(), ledgerReader, 1) require.NoError(t, err) assert.False(t, exists) diff --git a/cmd/stellar-rpc/internal/rpcv1/daemon/daemon.go b/cmd/stellar-rpc/internal/rpcv1/daemon/daemon.go index 53f9d2710..3516975f0 100644 --- a/cmd/stellar-rpc/internal/rpcv1/daemon/daemon.go +++ b/cmd/stellar-rpc/internal/rpcv1/daemon/daemon.go @@ -483,33 +483,32 @@ func (d *Daemon) mustInitializeStorage(cfg *config.Config) *feewindow.FeeWindows // 3. Apply all migrations, including fee stat analysis. // var initialSeq, currentSeq uint32 - err = sqlitedb.NewLedgerReader(d.db).StreamLedgerRange( - readTxMetaCtx, - ledgerSeqRange.First, - ledgerSeqRange.Last, - func(txMeta xdr.LedgerCloseMeta) error { - currentSeq = txMeta.LedgerSequence() - if initialSeq == 0 { - initialSeq = currentSeq - d.logger. - WithField("first", initialSeq). - WithField("last", ledgerSeqRange.Last). - Info("Initializing in-memory store") - } else if (currentSeq-initialSeq)%inMemoryInitializationLedgerLogPeriod == 0 { - d.logger. - WithField("seq", currentSeq). - WithField("last", ledgerSeqRange.Last). - Debug("Still initializing in-memory store") - } - - if err := dataMigrations.Apply(readTxMetaCtx, txMeta); err != nil { - d.logger.WithError(err).Fatal("could not apply migration for ledger ", currentSeq) - } + reader := sqlitedb.NewLedgerReader(d.db) + for l, err := range reader.ScanLedgers(readTxMetaCtx, ledgerSeqRange.First, ledgerSeqRange.Last) { + if err != nil { + d.logger.WithError(err).Fatal("Could not obtain txmeta cache from the database") + } + var txMeta xdr.LedgerCloseMeta + if err := txMeta.UnmarshalBinary(l.Raw); err != nil { + d.logger.WithError(err).Fatal("could not decode ledger ", l.Sequence) + } + currentSeq = txMeta.LedgerSequence() + if initialSeq == 0 { + initialSeq = currentSeq + d.logger. + WithField("first", initialSeq). + WithField("last", ledgerSeqRange.Last). + Info("Initializing in-memory store") + } else if (currentSeq-initialSeq)%inMemoryInitializationLedgerLogPeriod == 0 { + d.logger. + WithField("seq", currentSeq). + WithField("last", ledgerSeqRange.Last). + Debug("Still initializing in-memory store") + } - return nil - }) - if err != nil { - d.logger.WithError(err).Fatal("Could not obtain txmeta cache from the database") + if err := dataMigrations.Apply(readTxMetaCtx, txMeta); err != nil { + d.logger.WithError(err).Fatal("could not apply migration for ledger ", currentSeq) + } } if err := dataMigrations.Commit(readTxMetaCtx); err != nil { diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger.go index b04fa2a0e..59342ba22 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger.go @@ -4,8 +4,8 @@ package sqlitedb import ( "context" "database/sql" - "errors" "fmt" + "iter" sq "github.com/Masterminds/squirrel" @@ -32,6 +32,7 @@ type LedgerWriter interface { type readDB interface { Select(ctx context.Context, dest any, query sq.Sqlizer) error + Query(ctx context.Context, query sq.Sqlizer) (*db.Rows, error) } type ledgerReader struct { @@ -51,52 +52,9 @@ func (l ledgerReaderTx) GetLedgerRange(ctx context.Context) (store.LedgerRange, return getLedgerRangeWithoutCache(ctx, l.tx) } -// BatchGetLedgers fetches ledgers in batches from the db. -func (l ledgerReaderTx) BatchGetLedgers( - ctx context.Context, - start, end uint32, -) ([]store.LedgerMetadataChunk, error) { - if start > end { - return nil, errors.New("batch size must be greater than zero") - } - sql := sq.Select("meta"). - From(ledgerCloseMetaTableName). - Where(sq.And{ - sq.GtOrEq{"sequence": start}, - sq.LtOrEq{"sequence": end}, - }) - - results := make([][]byte, 0, end-start+1) - if err := l.tx.Select(ctx, &results, sql); err != nil { - return nil, err - } - - batch := make([]store.LedgerMetadataChunk, len(results)) - for i, meta := range results { - headerView, err := xdr.LedgerCloseMetaView(meta).LedgerHeader() - if err != nil { - return nil, err - } - headerRaw, err := headerView.Raw() - if err != nil { - return nil, err - } - batch[i] = store.LedgerMetadataChunk{HeaderRaw: headerRaw, Lcm: meta} - } - - return batch, nil -} - -// WithLedgerRaw lends the ledger's stored meta blob without decoding it. The -// blob is ours to lend: database/sql clones each BLOB scanned into a *[]byte. -func (l ledgerReaderTx) WithLedgerRaw( - ctx context.Context, sequence uint32, fn store.WithLedgerRawFn, -) (bool, error) { - meta, found, err := getLedgerRawFromDB(ctx, l.tx, sequence) - if err != nil || !found { - return found, err - } - return true, fn(meta) +// ScanLedgers reads inside the reader's transaction. +func (l ledgerReaderTx) ScanLedgers(ctx context.Context, start, end uint32) iter.Seq2[store.RawLedger, error] { + return scanLedgers(ctx, l.tx, start, end) } func (l ledgerReaderTx) Done() error { @@ -122,47 +80,49 @@ func (r ledgerReader) NewTx(ctx context.Context) (store.LedgerReaderTx, error) { return tx, nil } -// StreamLedgerRange runs f over inclusive (startLedger, endLedger) (until f errors or signals it's done). -func (r ledgerReader) StreamLedgerRange( - ctx context.Context, - startLedger uint32, - endLedger uint32, - f store.StreamLedgerFn, -) error { - sql := sq.Select("meta").From(ledgerCloseMetaTableName). - Where(sq.GtOrEq{"sequence": startLedger}). - Where(sq.LtOrEq{"sequence": endLedger}). - OrderBy("sequence asc") - - q, err := r.db.Query(ctx, sql) - if err != nil { - return err - } - defer q.Close() - for q.Next() { - var closeMeta xdr.LedgerCloseMeta - if err = q.Scan(&closeMeta); err != nil { - return err - } - if err = f(closeMeta); err != nil { - return err - } - } - return q.Err() +// ScanLedgers reads the pooled connection: no snapshot, the store as it stands. +func (r ledgerReader) ScanLedgers(ctx context.Context, start, end uint32) iter.Seq2[store.RawLedger, error] { + return scanLedgers(ctx, r.db, start, end) } -// GetLedger fetches a single ledger from the db. -func (r ledgerReader) GetLedger(ctx context.Context, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { - return getLedgerFromDB(ctx, r.db, sequence) -} +// scanLedgers yields the stored ledgers in [start, end] ascending, one row at a +// time. Absent sequences are simply not yielded. +func scanLedgers(ctx context.Context, q readDB, start, end uint32) iter.Seq2[store.RawLedger, error] { + return func(yield func(store.RawLedger, error) bool) { + if start > end { + return + } + // The primary-key range plan is one B-tree seek, for a scan of one too. + stmt := sq.Select("sequence", "meta").From(ledgerCloseMetaTableName). + Where(sq.GtOrEq{"sequence": start}).Where(sq.LtOrEq{"sequence": end}).OrderBy("sequence asc") -// WithLedgerRaw lends the ledger's stored meta blob without decoding it. -func (r ledgerReader) WithLedgerRaw(ctx context.Context, sequence uint32, fn store.WithLedgerRawFn) (bool, error) { - meta, found, err := getLedgerRawFromDB(ctx, r.db, sequence) - if err != nil || !found { - return found, err + rows, err := q.Query(ctx, stmt) + if err != nil { + yield(store.RawLedger{}, err) + return + } + // Runs on an early break too, which is how the consumer ends the scan. + defer rows.Close() + + for rows.Next() { + if err := ctx.Err(); err != nil { + yield(store.RawLedger{}, err) + return + } + var seq uint32 + var meta sql.RawBytes // the driver's buffer, valid until the next Next: RawLedger's loan + if err := rows.Scan(&seq, &meta); err != nil { + yield(store.RawLedger{}, err) + return + } + if !yield(store.RawLedger{Sequence: seq, Raw: []byte(meta)}, nil) { + return + } + } + if err := rows.Err(); err != nil { + yield(store.RawLedger{}, err) + } } - return true, fn(meta) } // GetLedgerRange pulls the min/max ledger sequence numbers from the meta table. @@ -361,19 +321,6 @@ func (l ledgerWriter) trimLedgers(latestLedgerSeq uint32, retentionWindow uint32 return err } -// getLedgerFromDB fetches a single ledger from the database. -func getLedgerFromDB(ctx context.Context, db readDB, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { - meta, found, err := getLedgerRawFromDB(ctx, db, sequence) - if err != nil || !found { - return xdr.LedgerCloseMeta{}, false, err - } - var lcm xdr.LedgerCloseMeta - if err := lcm.UnmarshalBinary(meta); err != nil { - return xdr.LedgerCloseMeta{}, false, err - } - return lcm, true, nil -} - // getLedgerRawFromDB fetches a single ledger's meta blob. The bytes are owned: // the driver and database/sql each copy the BLOB out of SQLite's memory. func getLedgerRawFromDB(ctx context.Context, db readDB, sequence uint32) ([]byte, bool, error) { diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger_test.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger_test.go index eb81dd7dc..31b412320 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger_test.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/ledger_test.go @@ -43,14 +43,13 @@ func createLedger(ledgerSequence uint32) xdr.LedgerCloseMeta { func assertLedgerRange(t *testing.T, reader LedgerReader, start, end uint32) { ctx := t.Context() - var allLedgers []xdr.LedgerCloseMeta - err := reader.StreamLedgerRange(ctx, start-1, end+1, func(txmeta xdr.LedgerCloseMeta) error { - allLedgers = append(allLedgers, txmeta) - return nil - }) - require.NoError(t, err) + var allLedgers [][]byte + for l, err := range reader.ScanLedgers(ctx, start-1, end+1) { + require.NoError(t, err) + allLedgers = append(allLedgers, bytes.Clone(l.Raw)) // the loan forbids retaining Raw + } for i := start - 1; i <= end+1; i++ { - ledger, exists, err := reader.GetLedger(ctx, i) + ledger, exists, err := store.GetLedger(ctx, reader, i) require.NoError(t, err) if i < start || i > end { assert.False(t, exists) @@ -64,9 +63,7 @@ func assertLedgerRange(t *testing.T, reader LedgerReader, start, end uint32) { require.NoError(t, err) assert.Equal(t, expectedBinary, ledgerBinary) - ledgerBinary, err = allLedgers[0].MarshalBinary() - require.NoError(t, err) - assert.Equal(t, expectedBinary, ledgerBinary) + assert.Equal(t, expectedBinary, allLedgers[0]) allLedgers = allLedgers[1:] } assert.Empty(t, allLedgers) @@ -77,7 +74,7 @@ func TestLedgers(t *testing.T) { daemon := host.MakeNoOpDaemon() reader := NewLedgerReader(db) - _, exists, err := reader.GetLedger(t.Context(), 1) + _, exists, err := store.GetLedger(t.Context(), reader, 1) require.NoError(t, err) assert.False(t, exists) @@ -267,7 +264,7 @@ func TestWithLedgerRaw(t *testing.T) { reader := NewLedgerReader(db) var got []byte - found, err := reader.WithLedgerRaw(t.Context(), 42, func(raw []byte) error { + found, err := store.WithLedgerRaw(t.Context(), reader, 42, func(raw []byte) error { got = bytes.Clone(raw) return nil }) @@ -276,7 +273,7 @@ func TestWithLedgerRaw(t *testing.T) { assert.Equal(t, want, got) ran := false - found, err = reader.WithLedgerRaw(t.Context(), 43, func([]byte) error { + found, err = store.WithLedgerRaw(t.Context(), reader, 43, func([]byte) error { ran = true return nil }) @@ -285,6 +282,40 @@ func TestWithLedgerRaw(t *testing.T) { assert.False(t, ran) } +// TestScanLedgers pins what the handlers' gap checks rest on: ascending, duplicate-free, +// within [start, end], absent sequences skipped, and a start below the oldest row served from it. +func TestScanLedgers(t *testing.T) { + db := NewTestDB(t) + for _, seq := range []uint32{10, 11, 13, 14} { // 12 is missing + tx, err := NewReadWriter(logger, db, host.MakeNoOpDaemon(), 15, passphrase).NewTx(t.Context()) + require.NoError(t, err) + lcm := createLedger(seq) + require.NoError(t, tx.LedgerWriter().InsertLedger(lcm)) + require.NoError(t, tx.Commit(lcm, nil)) + } + readTx, err := NewLedgerReader(db).NewTx(t.Context()) + require.NoError(t, err) + defer func() { _ = readTx.Done() }() + + scan := func(start, end uint32) []uint32 { + var got []uint32 + for l, err := range readTx.ScanLedgers(t.Context(), start, end) { + require.NoError(t, err) + var lcm xdr.LedgerCloseMeta + require.NoError(t, lcm.UnmarshalBinary(l.Raw)) + require.Equal(t, l.Sequence, lcm.LedgerSequence(), "Sequence must match the bytes") + got = append(got, l.Sequence) + } + return got + } + assert.Equal(t, []uint32{10, 11, 13, 14}, scan(10, 14), "ascending, and the gap at 12 is silent") + assert.Equal(t, []uint32{11, 13}, scan(11, 13), "bounded on both ends") + assert.Equal(t, []uint32{10, 11}, scan(1, 11), "a start below the oldest row is served from it") + assert.Empty(t, scan(12, 12), "an absent ledger yields nothing") + assert.Empty(t, scan(14, 10), "start above end yields nothing") + assert.Empty(t, scan(20, 30), "beyond the latest row yields nothing") +} + func BenchmarkGetLedgerRange(b *testing.B) { testDB, lcms := setupBenchmarkingDB(b) reader := NewLedgerReader(testDB) @@ -297,7 +328,7 @@ func BenchmarkGetLedgerRange(b *testing.B) { } } -func BenchmarkBatchGetLedgers(b *testing.B) { +func BenchmarkScanLedgers(b *testing.B) { testDB, lcms := setupBenchmarkingDB(b) reader := NewLedgerReader(testDB) readTx, err := reader.NewTx(b.Context()) @@ -308,14 +339,26 @@ func BenchmarkBatchGetLedgers(b *testing.B) { end := start + uint32(batchSize) - 1 for b.Loop() { - ledgers, err := readTx.BatchGetLedgers(b.Context(), start, end) - require.NoError(b, err) - - var hdrFirst, hdrLast xdr.LedgerHeaderHistoryEntry - require.NoError(b, hdrFirst.UnmarshalBinary(ledgers[0].HeaderRaw)) - require.NoError(b, hdrLast.UnmarshalBinary(ledgers[batchSize-1].HeaderRaw)) - assert.EqualValues(b, lcms[0].LedgerSequence(), hdrFirst.Header.LedgerSeq) - assert.EqualValues(b, lcms[batchSize-1].LedgerSequence(), hdrLast.Header.LedgerSeq) + // The header slice is what getLedgers pulls off each scanned ledger. + var first, last xdr.LedgerHeaderHistoryEntry + count := 0 + for entry, err := range readTx.ScanLedgers(b.Context(), start, end) { + require.NoError(b, err) + headerView, herr := xdr.LedgerCloseMetaView(entry.Raw).LedgerHeader() + require.NoError(b, herr) + raw, rerr := headerView.Raw() + require.NoError(b, rerr) + switch count { + case 0: + require.NoError(b, first.UnmarshalBinary(raw)) + case int(batchSize) - 1: + require.NoError(b, last.UnmarshalBinary(raw)) + } + count++ + } + require.Equal(b, int(batchSize), count) + assert.EqualValues(b, lcms[0].LedgerSequence(), first.Header.LedgerSeq) + assert.EqualValues(b, lcms[batchSize-1].LedgerSequence(), last.Header.LedgerSeq) } } diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/mocks.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/mocks.go index 33600cb79..93516b486 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/mocks.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/mocks.go @@ -4,6 +4,7 @@ import ( "context" "errors" "io" + "iter" "github.com/prometheus/client_golang/prometheus" @@ -95,28 +96,14 @@ func NewMockLedgerReader(txn *MockTransactionHandler) *MockLedgerReader { } } -func (m *MockLedgerReader) GetLedger(_ context.Context, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { - lcm, ok := m.txn.ledgerSeqToMeta[sequence] - if !ok { - return xdr.LedgerCloseMeta{}, false, nil - } - return *lcm, true, nil -} - -func (m *MockLedgerReader) WithLedgerRaw(_ context.Context, sequence uint32, fn store.WithLedgerRawFn) (bool, error) { - lcm, ok := m.txn.ledgerSeqToMeta[sequence] - if !ok { - return false, nil - } - rawMeta, err := lcm.MarshalBinary() - if err != nil { - return false, err - } - return true, fn(rawMeta) -} - -func (m *MockLedgerReader) StreamLedgerRange(_ context.Context, _ uint32, _ uint32, _ store.StreamLedgerFn) error { - return nil +func (m *MockLedgerReader) ScanLedgers(_ context.Context, start, end uint32) iter.Seq2[store.RawLedger, error] { + return store.ScanLedgersFrom(start, end, func(seq uint32) (xdr.LedgerCloseMeta, bool, error) { + lcm, ok := m.txn.ledgerSeqToMeta[seq] + if !ok { + return xdr.LedgerCloseMeta{}, false, nil + } + return *lcm, true, nil + }) } func (m *MockLedgerReader) GetLedgerRange(_ context.Context) (store.LedgerRange, error) { diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/transaction_test.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/transaction_test.go index bda0e5558..2dc94f43c 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/transaction_test.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/transaction_test.go @@ -284,7 +284,7 @@ func TestInsertTransactionsBatchingExceedsLimit(t *testing.T) { // Verify ledger was ingested with the correct number of transactions. ledgerReader := NewLedgerReader(testDB) - lcmReadBack, exists, err := ledgerReader.GetLedger(ctx, lcm.LedgerSequence()) + lcmReadBack, exists, err := store.GetLedger(ctx, ledgerReader, lcm.LedgerSequence()) require.NoError(t, err) require.True(t, exists) envelopes := lcmReadBack.TransactionEnvelopes() diff --git a/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader.go b/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader.go index a239d7adb..a59fa6e63 100644 --- a/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader.go +++ b/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader.go @@ -4,7 +4,6 @@ package adapters import ( - "bytes" "context" "errors" "fmt" @@ -12,23 +11,14 @@ import ( "github.com/stellar/go-stellar-sdk/xdr" - "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/chunk" "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/query" "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/stores" - "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/stores/ledger" "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/store" ) -// walkSpanCap bounds the ledger span ledgerReaderTx.WithLedgerRaw primes its -// walk iterator with: one chunk's worth of ledgers, touching at most two -// chunks when the span straddles a boundary, so view.ScanLedgers can resolve -// every reader up front. Handler scan limits (methods.LedgerScanLimit) must -// stay ≤ this cap; the pairing test enforces it. -const walkSpanCap = chunk.LedgersPerChunk - // LedgerReader satisfies store.LedgerReader over the query router. Every // method reads through the request's read view (see WithView); NewTx returns -// a handle that runs its walk against that same view until Done. +// a handle whose scans read that same view until Done. type LedgerReader struct{} func NewLedgerReader() *LedgerReader { @@ -46,35 +36,13 @@ func (r *LedgerReader) GetLatestLedgerSequence(ctx context.Context) (uint32, err return view.LatestLedger(), nil } -func (r *LedgerReader) GetLedger(ctx context.Context, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { - view, err := query.ViewFrom(ctx) - if err != nil { - return xdr.LedgerCloseMeta{}, false, err - } - lcm, found, err := getLedger(view, sequence) - return lcm, found, err -} - -// WithLedgerRaw lends the ledger's raw bytes with no copy: the routed point -// read lends the tier's buffer, whose validity ends with fn — exactly the -// loan's terms. -func (r *LedgerReader) WithLedgerRaw(ctx context.Context, sequence uint32, fn store.WithLedgerRawFn) (bool, error) { +// ScanLedgers reads the request's view; unlike a Tx it takes no snapshot of its own. +func (r *LedgerReader) ScanLedgers(ctx context.Context, start, end uint32) iter.Seq2[store.RawLedger, error] { view, err := query.ViewFrom(ctx) if err != nil { - return false, err - } - if !inWindow(view, sequence) { - return false, nil - } - found := false - err = view.WithLedger(sequence, func(raw []byte) error { - found = true - return fn(raw) - }) - if !found && errors.Is(err, stores.ErrNotFound) { - return false, nil + return func(yield func(store.RawLedger, error) bool) { yield(store.RawLedger{}, err) } } - return found, err + return scanView(ctx, view, start, end) } func (r *LedgerReader) GetLedgerRange(ctx context.Context) (store.LedgerRange, error) { @@ -86,36 +54,6 @@ func (r *LedgerReader) GetLedgerRange(ctx context.Context) (store.LedgerRange, e return lr, err } -func (r *LedgerReader) StreamLedgerRange( - ctx context.Context, startLedger, endLedger uint32, f store.StreamLedgerFn, -) error { - view, err := query.ViewFrom(ctx) - if err != nil { - return err - } - - scan, err := view.ScanLedgers(startLedger, endLedger) - if err != nil { - return err - } - for entry, err := range scan { - if err != nil { - return err - } - if ctx.Err() != nil { - return ctx.Err() - } - var lcm xdr.LedgerCloseMeta - if err := lcm.UnmarshalBinary(entry.Bytes); err != nil { - return fmt.Errorf("adapters: unmarshal ledger %d: %w", entry.Seq, err) - } - if err := f(lcm); err != nil { - return err - } - } - return nil -} - func (r *LedgerReader) NewTx(ctx context.Context) (store.LedgerReaderTx, error) { view, err := query.ViewFrom(ctx) if err != nil { @@ -124,168 +62,74 @@ func (r *LedgerReader) NewTx(ctx context.Context) (store.LedgerReaderTx, error) return &ledgerReaderTx{view: view}, nil } -// ledgerReaderTx satisfies store.LedgerReaderTx over the request's read view -// (the serving wrapper owns and releases it — Done does not). WithLedgerRaw -// serves getTransactions' ascending, contiguous per-ledger walk by pulling -// from a ScanLedgers iterator primed on the first call, see walk; -// GetLedgerRange and BatchGetLedgers read through the same view but never -// touch that iterator. +// ledgerReaderTx satisfies store.LedgerReaderTx over the request's read view. +// The serving wrapper owns and releases the view, so Done has nothing to do. type ledgerReaderTx struct { view *query.ReadView - - // next/stop are the pull ends of the walk iterator; nil until the first - // walk step primes them. - next func() (ledger.Entry, error, bool) - stop func() } -// WithLedgerRaw lends the step's bytes straight from the chunk reader's -// scratch buffer, which the next step overwrites — fn must not retain them. -func (tx *ledgerReaderTx) WithLedgerRaw( - ctx context.Context, sequence uint32, fn store.WithLedgerRawFn, -) (bool, error) { - entry, found, err := tx.walk(ctx, sequence) - if err != nil || !found { - return found, err - } - return true, fn(entry.Bytes) +func (tx *ledgerReaderTx) ScanLedgers(ctx context.Context, start, end uint32) iter.Seq2[store.RawLedger, error] { + return scanView(ctx, tx.view, start, end) } func (tx *ledgerReaderTx) GetLedgerRange(_ context.Context) (store.LedgerRange, error) { return getLedgerRange(tx.view) } -func (tx *ledgerReaderTx) BatchGetLedgers( - ctx context.Context, start, end uint32, -) ([]store.LedgerMetadataChunk, error) { - scan, err := tx.view.ScanLedgers(start, end) - if err != nil { - return nil, err - } - out := make([]store.LedgerMetadataChunk, 0, end-start+1) - for entry, err := range scan { - if err != nil { - return nil, err - } +func (tx *ledgerReaderTx) Done() error { return nil } + +// scanView yields [start, end] off the view with no copy: RawLedger.Raw +// aliases the tier's buffer until the next step. A scan of one, after clamping, +// is the routed point read, so getLatestLedger's I/O stays a pinned lookup, not a chunk walk; +// a hot-store miss inside the window yields nothing, matching v1's absent shape. +func scanView(ctx context.Context, view *query.ReadView, start, end uint32) iter.Seq2[store.RawLedger, error] { + return func(yield func(store.RawLedger, error) bool) { + // The request duration limiter answers the client at the deadline but + // only abandons the handler goroutine; without these checks an + // abandoned scan would keep decoding while holding its read view. if err := ctx.Err(); err != nil { - return nil, err + yield(store.RawLedger{}, err) + return } - headerRaw, err := rawHeaderFromLCMBytes(entry.Bytes) - if err != nil { - return nil, fmt.Errorf("adapters: slice ledger %d header: %w", entry.Seq, err) + // Clamp to the window here rather than let ClampRange answer a start below + // the floor with a *RangeError: not yielding is the shape the handler + // turns into v1's InvalidParams naming the caller's ledger. + start = max(start, view.OldestLedger()) + end = min(end, view.LatestLedger()) + if start > end { + return } - // Entry.Bytes aliases the reader's scratch buffer and is overwritten - // on the next iteration step; clone what we keep. - out = append(out, store.LedgerMetadataChunk{ - HeaderRaw: bytes.Clone(headerRaw), - Lcm: bytes.Clone(entry.Bytes), - }) - } - return out, nil -} - -// rawHeaderFromLCMBytes slices the marshaled LedgerHeaderHistoryEntry out of -// raw LCM bytes. No decode: the consumer (getLedgers) sends the header and -// the metadata as raw bytes. -func rawHeaderFromLCMBytes(raw []byte) ([]byte, error) { - view, err := xdr.LedgerCloseMetaView(raw).LedgerHeader() - if err != nil { - return nil, err - } - return view.Raw() -} - -func (tx *ledgerReaderTx) Done() error { - if tx.stop != nil { - tx.stop() - } - return nil -} - -// walk is the store.LedgerReaderTx walk: one ScanLedgers iterator primed on -// the first call. The returned Entry.Bytes are valid only until the next call. -func (tx *ledgerReaderTx) walk(ctx context.Context, sequence uint32) (ledger.Entry, bool, error) { - // The request duration limiter answers the client at the deadline but only - // abandons the handler goroutine — it cannot stop it. Without this check - // the abandoned getTransactions walk would keep decoding the rest of its - // primed span (up to a whole chunk) while holding its read view, and the - // deletion grace margin is sized assuming walks stop within one iteration - // of their deadline. - if err := ctx.Err(); err != nil { - return ledger.Entry{}, false, err - } - // ClampRange is the only place the servable window is enforced and no - // point-read path calls it, so gate here: without this a view acquired - // between ingestion's commit and its SetLatestLedger could return a ledger - // above the view's frozen latest. - if !inWindow(tx.view, sequence) { - return ledger.Entry{}, false, nil - } - - if tx.next == nil { - // ScanLedgers' end is inclusive; the -1 keeps the span at exactly - // walkSpanCap ledgers, so a chunk-aligned start opens one chunk - // reader, not two. - scan, err := tx.view.ScanLedgers(sequence, min(tx.view.LatestLedger(), sequence+walkSpanCap-1)) + if start == end { + yielded := false + err := view.WithLedger(start, func(raw []byte) error { + yielded = true + yield(store.RawLedger{Sequence: start, Raw: raw}, nil) + return nil + }) + if err != nil && !yielded && !errors.Is(err, stores.ErrNotFound) { + yield(store.RawLedger{}, err) + } + return + } + scan, err := view.ScanLedgers(start, end) if err != nil { - return ledger.Entry{}, false, err + yield(store.RawLedger{}, err) + return } - tx.next, tx.stop = iter.Pull2(scan) - } - - entry, err, ok := tx.next() - if err != nil { - return ledger.Entry{}, false, err - } - if !ok { - // The iterator ran dry: the caller walked sequentially but past the span - // primed above. getTransactions' span cap (methods.LedgerScanLimit) keeps - // its walks inside the span, so reaching this means a new caller without - // that cap. Fail loudly rather than serve a wrong-position read. - return ledger.Entry{}, false, fmt.Errorf( - "adapters: ledger walk exhausted its primed %d-ledger span at ledger %d"+ - " — the calling handler must cap the request's ledger range", - walkSpanCap, sequence) - } - if entry.Seq != sequence { - // The walk contract (ascending, contiguous from the priming sequence) - // was broken. Fail loudly rather than serve the wrong ledger's data. - return ledger.Entry{}, false, fmt.Errorf( - "adapters: non-sequential ledger walk: asked for ledger %d, the walk is at ledger %d", - sequence, entry.Seq) - } - return entry, true, nil -} - -// inWindow reports whether seq falls inside the view's servable window -// [OldestLedger, LatestLedger] — the one gate every point read must apply. -// OldestLedger is always ≥ 2 (the floor sits on a chunk, and chunk 0 starts at -// ledger 2), so this also rejects the sequences chunk.IDFromLedger panics on. -func inWindow(view *query.ReadView, seq uint32) bool { - return seq >= view.OldestLedger() && seq <= view.LatestLedger() -} - -// getLedger is the one-shot point read: window-gated, then one ledger read. A -// hot-store miss inside the window maps to (false, nil), matching v1's -// absent-ledger shape. -func getLedger(view *query.ReadView, sequence uint32) (xdr.LedgerCloseMeta, bool, error) { - if !inWindow(view, sequence) { - return xdr.LedgerCloseMeta{}, false, nil - } - var lcm xdr.LedgerCloseMeta - err := view.WithLedger(sequence, func(raw []byte) error { - if uerr := lcm.UnmarshalBinary(raw); uerr != nil { - return fmt.Errorf("adapters: unmarshal ledger %d: %w", sequence, uerr) + for entry, err := range scan { + if err != nil { + yield(store.RawLedger{}, err) + return + } + if err := ctx.Err(); err != nil { + yield(store.RawLedger{}, err) + return + } + if !yield(store.RawLedger{Sequence: entry.Seq, Raw: entry.Bytes}, nil) { + return + } } - return nil - }) - if errors.Is(err, stores.ErrNotFound) { - return xdr.LedgerCloseMeta{}, false, nil - } - if err != nil { - return xdr.LedgerCloseMeta{}, false, err } - return lcm, true, nil } // getLedgerRange reads the window's edge sequences from the view. Close times diff --git a/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader_test.go b/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader_test.go index c41f305a6..b4db72fc6 100644 --- a/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/adapters/ledger_reader_test.go @@ -102,7 +102,7 @@ func TestGetLedgerRange_BootStampFallsBackThenCaches(t *testing.T) { func TestGetLedger_PointRead(t *testing.T) { ctx, reader, c0, _ := sparseFixture(t) - lcm, ok, err := reader.GetLedger(ctx, c0.FirstLedger()+2) + lcm, ok, err := store.GetLedger(ctx, reader, c0.FirstLedger()+2) require.NoError(t, err) require.True(t, ok) assert.Equal(t, c0.FirstLedger()+2, lcm.LedgerSequence()) @@ -112,7 +112,7 @@ func TestGetLedger_PointRead(t *testing.T) { func TestGetLedger_SubGenesisDoesNotPanic(t *testing.T) { ctx, reader, _, _ := sparseFixture(t) for _, seq := range []uint32{0, 1} { - _, ok, err := reader.GetLedger(ctx, seq) + _, ok, err := store.GetLedger(ctx, reader, seq) assert.NoError(t, err) assert.False(t, ok) } @@ -123,7 +123,7 @@ func TestGetLedger_OutsideWindow(t *testing.T) { // c1.FirstLedger()+1 is committed but above the view's latest; the gate, // not the store, must produce the miss. for _, seq := range []uint32{c0.FirstLedger() - 1, c1.FirstLedger() + 1} { - _, ok, err := reader.GetLedger(ctx, seq) + _, ok, err := store.GetLedger(ctx, reader, seq) assert.NoError(t, err) assert.False(t, ok) } @@ -137,218 +137,157 @@ func TestGetLedger_V1LedgerCloseMeta(t *testing.T) { r.SetLatestLedger(testChunk.FirstLedger(), query.CloseTimeAt(closeTimeFor(testChunk.FirstLedger()))) reader := NewLedgerReader() - lcm, ok, err := reader.GetLedger(viewCtx(t, r), testChunk.FirstLedger()) + lcm, ok, err := store.GetLedger(viewCtx(t, r), reader, testChunk.FirstLedger()) require.NoError(t, err) require.True(t, ok) assert.Equal(t, testChunk.FirstLedger(), lcm.LedgerSequence()) assert.Equal(t, closeTimeFor(testChunk.FirstLedger()), lcm.LedgerCloseTime()) } -func TestStreamLedgerRange(t *testing.T) { - ctx, reader, c0, c1 := sparseFixture(t) +// scanAll drains a Tx scan into sequences and clones of the borrowed bytes. +func scanAll(t *testing.T, tx store.LedgerReaderTx, start, end uint32) ([]uint32, [][]byte) { + t.Helper() var seqs []uint32 - err := reader.StreamLedgerRange(ctx, c0.FirstLedger(), c1.FirstLedger()+500, - func(lcm xdr.LedgerCloseMeta) error { - assert.Equal(t, closeTimeFor(lcm.LedgerSequence()), lcm.LedgerCloseTime()) - seqs = append(seqs, lcm.LedgerSequence()) - return nil - }) - require.NoError(t, err) - assert.Equal(t, append(seqRange(c0.FirstLedger(), c0.FirstLedger()+3), c1.FirstLedger()), seqs, - "streams what is committed, flat across the chunk border, clamped at latest") -} - -func TestStreamLedgerRange_CallbackErrorStopsStream(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - boom := errors.New("boom") - calls := 0 - err := reader.StreamLedgerRange(ctx, c0.FirstLedger(), c0.FirstLedger()+3, - func(xdr.LedgerCloseMeta) error { - calls++ - return boom - }) - assert.ErrorIs(t, err, boom) - assert.Equal(t, 1, calls) -} - -func TestStreamLedgerRange_BelowFloorIsRangeError(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - var rangeErr *query.RangeError - err := reader.StreamLedgerRange(ctx, 2, c0.FirstLedger(), - func(xdr.LedgerCloseMeta) error { return nil }) - require.ErrorAs(t, err, &rangeErr) - assert.Equal(t, uint32(2), rangeErr.Requested) - assert.Equal(t, c0.FirstLedger(), rangeErr.Oldest) -} - -// txGetLedger is the Tx's retired GetLedger: one WithLedgerRaw step, decoded inside the loan. -func txGetLedger(ctx context.Context, tx store.LedgerReaderTx, seq uint32) (xdr.LedgerCloseMeta, bool, error) { - var lcm xdr.LedgerCloseMeta - found, err := tx.WithLedgerRaw(ctx, seq, lcm.UnmarshalBinary) - return lcm, found, err -} - -func TestTxWalk_Contiguous(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - tx, err := reader.NewTx(ctx) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - - for seq := c0.FirstLedger(); seq <= c0.FirstLedger()+3; seq++ { - lcm, ok, err := txGetLedger(context.Background(), tx, seq) + var raws [][]byte + for l, err := range tx.ScanLedgers(context.Background(), start, end) { require.NoError(t, err) - require.True(t, ok, "ledger %d", seq) - assert.Equal(t, seq, lcm.LedgerSequence()) - assert.Equal(t, closeTimeFor(seq), lcm.LedgerCloseTime()) + seqs = append(seqs, l.Sequence) + raws = append(raws, bytes.Clone(l.Raw)) // the loan forbids retaining Raw } + return seqs, raws } -func TestTxWalk_NonSequentialFailsLoudly(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) +func TestTxScanLedgers_YieldsAscendingAndClampsAtLatest(t *testing.T) { + ctx, reader, c0, c1 := sparseFixture(t) tx, err := reader.NewTx(ctx) require.NoError(t, err) defer func() { _ = tx.Done() }() - _, ok, err := txGetLedger(context.Background(), tx, c0.FirstLedger()) - require.NoError(t, err) - require.True(t, ok) - - // Skipping ahead breaks the walk contract; serving the wrong ledger's - // data would be worse than an error. - _, _, err = txGetLedger(context.Background(), tx, c0.FirstLedger()+2) - assert.ErrorContains(t, err, "non-sequential") + seqs, raws := scanAll(t, tx, c0.FirstLedger(), c1.FirstLedger()+500) + want := append(seqRange(c0.FirstLedger(), c0.FirstLedger()+3), c1.FirstLedger()) + assert.Equal(t, want, seqs, "four from chunk 5, one from chunk 6, clamped at latest; the gap between is silent") + for i, raw := range raws { + var lcm xdr.LedgerCloseMeta + require.NoError(t, lcm.UnmarshalBinary(raw)) + assert.Equal(t, want[i], lcm.LedgerSequence()) + assert.Equal(t, closeTimeFor(want[i]), lcm.LedgerCloseTime()) + } } -func TestTxWalk_GuardsBeforePriming(t *testing.T) { - ctx, reader, c0, c1 := sparseFixture(t) +func TestTxScanLedgers_LendsTheSameBytesGetLedgerDecodes(t *testing.T) { + ctx, reader, c0, _ := sparseFixture(t) tx, err := reader.NewTx(ctx) require.NoError(t, err) defer func() { _ = tx.Done() }() - // Sub-genesis and out-of-window sequences return clean misses without - // consuming the walk iterator... - for _, seq := range []uint32{0, 1, c0.FirstLedger() - 1, c1.FirstLedger() + 1} { - _, ok, err := txGetLedger(context.Background(), tx, seq) - assert.NoError(t, err) - assert.False(t, ok) + seqs, raws := scanAll(t, tx, c0.FirstLedger(), c0.FirstLedger()+3) + require.Len(t, seqs, 4) + for i, seq := range seqs { + lcm, ok, err := store.GetLedger(ctx, reader, seq) + require.NoError(t, err) + require.True(t, ok) + want, err := lcm.MarshalBinary() + require.NoError(t, err) + assert.Equal(t, want, raws[i], "ledger %d", seq) } - // ...so a walk primed afterwards still starts at its own sequence. - lcm, ok, err := txGetLedger(context.Background(), tx, c0.FirstLedger()) - require.NoError(t, err) - require.True(t, ok) - assert.Equal(t, c0.FirstLedger(), lcm.LedgerSequence()) } -func TestTxGetLedgerRange_DoesNotDisturbTheWalk(t *testing.T) { - ctx, reader, c0, c1 := sparseFixture(t) +// A start below the retention floor is raised to it rather than surfacing +// ClampRange's *RangeError; the handler's gap check then names the caller's ledger. +func TestTxScanLedgers_BelowFloorStartsAtOldest(t *testing.T) { + ctx, reader, c0, _ := sparseFixture(t) tx, err := reader.NewTx(ctx) require.NoError(t, err) defer func() { _ = tx.Done() }() - // getTransactions reads the range first, then walks; the range's point - // reads must not consume the walk iterator. - lr, err := tx.GetLedgerRange(context.Background()) - require.NoError(t, err) - assert.Equal(t, c0.FirstLedger(), lr.FirstLedger.Sequence) - assert.Equal(t, c1.FirstLedger(), lr.LastLedger.Sequence) - - lcm, ok, err := txGetLedger(context.Background(), tx, c0.FirstLedger()+1) - require.NoError(t, err) - require.True(t, ok) - assert.Equal(t, c0.FirstLedger()+1, lcm.LedgerSequence()) + seqs, _ := scanAll(t, tx, 2, c0.FirstLedger()+1) + assert.Equal(t, seqRange(c0.FirstLedger(), c0.FirstLedger()+1), seqs) } -func TestTxBatchGetLedgers(t *testing.T) { +func TestTxScanLedgers_EmptyRanges(t *testing.T) { ctx, reader, c0, c1 := sparseFixture(t) tx, err := reader.NewTx(ctx) require.NoError(t, err) defer func() { _ = tx.Done() }() - got, err := tx.BatchGetLedgers(context.Background(), c0.FirstLedger(), c1.FirstLedger()+500) - require.NoError(t, err) - require.Len(t, got, 5, "four from chunk 5, one from chunk 6, clamped at latest") - - want := append(seqRange(c0.FirstLedger(), c0.FirstLedger()+3), c1.FirstLedger()) - for i, mc := range got { - var hdr xdr.LedgerHeaderHistoryEntry - require.NoError(t, hdr.UnmarshalBinary(mc.HeaderRaw)) - assert.Equal(t, want[i], uint32(hdr.Header.LedgerSeq)) - // Entry.Bytes is borrowed from the reader's scratch buffer; if the - // adapter forgot to clone, earlier entries would decode to a later - // ledger's bytes by the time the loop finishes. - var lcm xdr.LedgerCloseMeta - require.NoError(t, lcm.UnmarshalBinary(mc.Lcm)) - assert.Equal(t, want[i], lcm.LedgerSequence()) - assert.Equal(t, closeTimeFor(want[i]), lcm.LedgerCloseTime()) + for name, r := range map[string][2]uint32{ + "beyond latest": {c1.FirstLedger() + 100, c1.FirstLedger() + 200}, + "start above end": {c0.FirstLedger() + 2, c0.FirstLedger()}, + "in-window gap": {c0.FirstLedger() + 4, c1.FirstLedger() - 1}, + } { + seqs, _ := scanAll(t, tx, r[0], r[1]) + assert.Empty(t, seqs, name) } } -func TestTxBatchGetLedgers_HeaderMatchesFullDecode(t *testing.T) { - cat := openTestCatalog(t) - r := query.NewRegistry(cat, geometry.NewRetention(0, testChunk)) - rawV1, _ := lcmV1WithClassicTx(t, testChunk.FirstLedger()) - rawV2, _ := lcmWithTxs(t, testChunk.FirstLedger()+1, txSpec{}) - seedHotChunkLCMs(t, cat, r, testChunk, rawV1, rawV2) - r.SetLatestLedger(testChunk.FirstLedger()+1, query.CloseTimeAt(closeTimeFor(testChunk.FirstLedger()+1))) - reader := NewLedgerReader() - - tx, err := reader.NewTx(viewCtx(t, r)) +func TestTxScanLedgers_EmptyStoreYieldsNothing(t *testing.T) { + ctx, reader := emptyFixture(t) + tx, err := reader.NewTx(ctx) require.NoError(t, err) defer func() { _ = tx.Done() }() - got, err := tx.BatchGetLedgers(context.Background(), testChunk.FirstLedger(), testChunk.FirstLedger()+1) - require.NoError(t, err) - require.Len(t, got, 2) - for i, raw := range [][]byte{rawV1, rawV2} { - var full xdr.LedgerCloseMeta - require.NoError(t, full.UnmarshalBinary(raw)) - wantHeader, err := full.LedgerHeaderHistoryEntry().MarshalBinary() - require.NoError(t, err) - assert.Equal(t, wantHeader, got[i].HeaderRaw, - "the header slice must equal what the full decode re-marshals") - assert.Equal(t, raw, got[i].Lcm) - } + seqs, _ := scanAll(t, tx, 2, 1_000_000) + assert.Empty(t, seqs) } -func TestTxBatchGetLedgers_BelowFloorIsRangeError(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - tx, err := reader.NewTx(ctx) +func TestTxScanLedgers_StopsOnCanceledContext(t *testing.T) { + r, first := sharedViewFixture(t) + tx, err := NewLedgerReader().NewTx(viewCtx(t, r)) require.NoError(t, err) defer func() { _ = tx.Done() }() - var rangeErr *query.RangeError - _, err = tx.BatchGetLedgers(context.Background(), 2, c0.FirstLedger()) - require.ErrorAs(t, err, &rangeErr) + t.Run("mid-scan", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var got []uint32 + var scanErr error + for l, err := range tx.ScanLedgers(ctx, first, first+2) { + if err != nil { + scanErr = err + break + } + got = append(got, l.Sequence) + cancel() // the next step must observe it + } + assert.Equal(t, []uint32{first}, got, "the step that ran before cancel is delivered") + assert.ErrorIs(t, scanErr, context.Canceled) + }) + t.Run("before the first step", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + for l, err := range tx.ScanLedgers(ctx, first, first+2) { + assert.ErrorIs(t, err, context.Canceled) + assert.Zero(t, l, "the RawLedger beside an error is zero") + } + }) } -func TestTxBatchGetLedgers_BeyondLatestIsEmpty(t *testing.T) { - ctx, reader, _, c1 := sparseFixture(t) +// The Tx is a snapshot: a scan cut short by a break leaves a later scan on the +// same Tx answering from the same state. +func TestTxScanLedgers_EarlyBreakThenRescan(t *testing.T) { + ctx, reader, c0, _ := sparseFixture(t) tx, err := reader.NewTx(ctx) require.NoError(t, err) defer func() { _ = tx.Done() }() - got, err := tx.BatchGetLedgers(context.Background(), c1.FirstLedger()+100, c1.FirstLedger()+200) - require.NoError(t, err) - assert.Empty(t, got) + for l, err := range tx.ScanLedgers(context.Background(), c0.FirstLedger(), c0.FirstLedger()+3) { + require.NoError(t, err) + assert.Equal(t, c0.FirstLedger(), l.Sequence) + break + } + seqs, _ := scanAll(t, tx, c0.FirstLedger()+1, c0.FirstLedger()+3) + assert.Equal(t, seqRange(c0.FirstLedger()+1, c0.FirstLedger()+3), seqs) } -func TestTxDone_WithAndWithoutPriming(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - +func TestTxDone_IsIdempotent(t *testing.T) { + ctx, reader, _, _ := sparseFixture(t) tx, err := reader.NewTx(ctx) require.NoError(t, err) assert.NoError(t, tx.Done()) assert.NoError(t, tx.Done(), "a second Done must be a no-op, not a double release") - - tx, err = reader.NewTx(ctx) - require.NoError(t, err) - _, _, err = txGetLedger(context.Background(), tx, c0.FirstLedger()) - require.NoError(t, err) - assert.NoError(t, tx.Done()) - assert.NoError(t, tx.Done(), "a second Done must be a no-op, not a double release") } -func TestTxWalk_CrossesChunkBorder(t *testing.T) { +func TestTxScanLedgers_CrossesChunkBorder(t *testing.T) { if testing.Short() { t.Skip("seeds a full 10k-ledger chunk") } @@ -358,81 +297,20 @@ func TestTxWalk_CrossesChunkBorder(t *testing.T) { seedHotLedgers(t, cat, r, c0, seqRange(c0.FirstLedger(), c0.LastLedger())...) seedHotLedgers(t, cat, r, c1, c1.FirstLedger(), c1.FirstLedger()+1) r.SetLatestLedger(c1.FirstLedger()+1, query.CloseTimeAt(closeTimeFor(c1.FirstLedger()+1))) - reader := NewLedgerReader() - - tx, err := reader.NewTx(viewCtx(t, r)) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - - for seq := c0.LastLedger() - 1; seq <= c1.FirstLedger()+1; seq++ { - lcm, ok, err := txGetLedger(context.Background(), tx, seq) - require.NoError(t, err, "ledger %d", seq) - require.True(t, ok, "ledger %d", seq) - assert.Equal(t, seq, lcm.LedgerSequence()) - } -} - -func TestTxWalk_PastSpanCapIsExhaustedNotNonSequential(t *testing.T) { - if testing.Short() { - t.Skip("seeds a full 10k-ledger chunk") - } - cat := openTestCatalog(t) - r := query.NewRegistry(cat, geometry.NewRetention(0, testChunk)) - c0, c1 := testChunk, testChunk+1 - seedHotLedgers(t, cat, r, c0, seqRange(c0.FirstLedger(), c0.LastLedger())...) - seedHotLedgers(t, cat, r, c1, c1.FirstLedger(), c1.FirstLedger()+1, c1.FirstLedger()+2) - r.SetLatestLedger(c1.FirstLedger()+2, query.CloseTimeAt(closeTimeFor(c1.FirstLedger()+2))) - reader := NewLedgerReader() - - tx, err := reader.NewTx(viewCtx(t, r)) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - - start := c0.FirstLedger() - for seq := start; seq < start+walkSpanCap; seq++ { - _, ok, err := txGetLedger(context.Background(), tx, seq) - require.NoError(t, err, "ledger %d", seq) - require.True(t, ok, "ledger %d", seq) - } - - _, _, err = txGetLedger(context.Background(), tx, start+walkSpanCap) - require.ErrorContains(t, err, "exhausted its primed") - assert.NotContains(t, err.Error(), "non-sequential") -} - -func TestBatchGetLedgers_ClonesBorrowedBytes(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - tx, err := reader.NewTx(ctx) + tx, err := NewLedgerReader().NewTx(viewCtx(t, r)) require.NoError(t, err) defer func() { _ = tx.Done() }() - got, err := tx.BatchGetLedgers(context.Background(), c0.FirstLedger(), c0.FirstLedger()+1) - require.NoError(t, err) - require.Len(t, got, 2) - assert.False(t, bytes.Equal(got[0].Lcm, got[1].Lcm), - "two entries decoding identically would mean both alias one scratch buffer") + seqs, _ := scanAll(t, tx, c0.LastLedger()-1, c1.FirstLedger()+1) + assert.Equal(t, seqRange(c0.LastLedger()-1, c1.FirstLedger()+1), seqs) } -func TestWalkSpanCapCoversTheHandlerScanLimit(t *testing.T) { - assert.Equal(t, methods.LedgerScanLimit, int(walkSpanCap), +func TestLedgerScanLimitMatchesChunkSpan(t *testing.T) { + assert.Equal(t, methods.LedgerScanLimit, int(chunk.LedgersPerChunk), "the per-request scan bound has one value (chunk.LedgersPerChunk); "+ "methods.LedgerScanLimit cannot derive from it (shared v1 code), so it is pinned here") } -func TestLedgerReaderTx_BatchGetLedgersStopsOnCanceledContext(t *testing.T) { - r, first := sharedViewFixture(t) - reader := NewLedgerReader() - - ctx, cancel := context.WithCancel(viewCtx(t, r)) - tx, err := reader.NewTx(ctx) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - cancel() - - _, err = tx.BatchGetLedgers(ctx, first, first+2) - assert.ErrorIs(t, err, context.Canceled) -} - // TestGetLedgerRange_SeededWindowReadsNoLedgers is the standing guard on the // close-time path: once both edges are stamped, a served daemon that is not // ingesting must answer getLedgerRange without touching a ledger at all. Each @@ -479,14 +357,14 @@ func TestGetLedgerRange_SeededWindowReadsNoLedgers(t *testing.T) { func TestWithLedgerRaw_LendsTheSameBytesGetLedgerDecodes(t *testing.T) { ctx, reader, c0, _ := sparseFixture(t) - lcm, ok, err := reader.GetLedger(ctx, c0.FirstLedger()) + lcm, ok, err := store.GetLedger(ctx, reader, c0.FirstLedger()) require.NoError(t, err) require.True(t, ok) want, err := lcm.MarshalBinary() require.NoError(t, err) var got []byte - found, err := reader.WithLedgerRaw(ctx, c0.FirstLedger(), func(raw []byte) error { + found, err := store.WithLedgerRaw(ctx, reader, c0.FirstLedger(), func(raw []byte) error { got = bytes.Clone(raw) // the loan forbids retaining raw return nil }) @@ -500,7 +378,7 @@ func TestWithLedgerRaw_MissDoesNotRunFn(t *testing.T) { // below the floor, in an in-window gap, and above latest for _, seq := range []uint32{c0.FirstLedger() - 1, c0.FirstLedger() + 10, c1.FirstLedger() + 1} { ran := false - found, err := reader.WithLedgerRaw(ctx, seq, func([]byte) error { + found, err := store.WithLedgerRaw(ctx, reader, seq, func([]byte) error { ran = true return nil }) @@ -510,88 +388,11 @@ func TestWithLedgerRaw_MissDoesNotRunFn(t *testing.T) { } } -func TestTxWithLedgerRaw_LendsTheSameBytesGetLedgerDecodes(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - rawTx, err := reader.NewTx(ctx) - require.NoError(t, err) - defer func() { _ = rawTx.Done() }() - - for seq := c0.FirstLedger(); seq <= c0.FirstLedger()+3; seq++ { - lcm, ok, err := reader.GetLedger(ctx, seq) - require.NoError(t, err) - require.True(t, ok, "ledger %d", seq) - want, err := lcm.MarshalBinary() - require.NoError(t, err) - - var got []byte - found, err := rawTx.WithLedgerRaw(context.Background(), seq, func(raw []byte) error { - // The loan forbids retaining raw, so clone inside fn. - got = bytes.Clone(raw) - return nil - }) - require.NoError(t, err) - require.True(t, found, "ledger %d", seq) - assert.Equal(t, want, got, "ledger %d", seq) - } -} - -func TestTxWithLedgerRaw_MissDoesNotRunFn(t *testing.T) { - ctx, reader, c0, c1 := sparseFixture(t) - tx, err := reader.NewTx(ctx) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - - for _, seq := range []uint32{0, 1, c0.FirstLedger() - 1, c1.FirstLedger() + 1} { - ran := false - found, err := tx.WithLedgerRaw(context.Background(), seq, func([]byte) error { - ran = true - return nil - }) - assert.NoError(t, err) - assert.False(t, found, "ledger %d", seq) - assert.False(t, ran, "fn must not run for an absent ledger %d", seq) - } -} - func TestWithLedgerRaw_CallbackErrorSurfacesAsFound(t *testing.T) { ctx, reader, c0, _ := sparseFixture(t) boom := errors.New("boom") // found stays true: the ledger WAS there, the caller's own callback failed. - found, err := reader.WithLedgerRaw(ctx, c0.FirstLedger(), func([]byte) error { return boom }) + found, err := store.WithLedgerRaw(ctx, reader, c0.FirstLedger(), func([]byte) error { return boom }) assert.ErrorIs(t, err, boom) assert.True(t, found) } - -func TestTxWithLedgerRaw_CallbackErrorSurfacesAsFound(t *testing.T) { - ctx, reader, c0, _ := sparseFixture(t) - tx, err := reader.NewTx(ctx) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - - boom := errors.New("boom") - // found stays true: the ledger WAS there, the caller's own callback failed. - found, err := tx.WithLedgerRaw(context.Background(), c0.FirstLedger(), - func([]byte) error { return boom }) - assert.ErrorIs(t, err, boom) - assert.True(t, found) -} - -func TestTxWithLedgerRaw_StopsOnCanceledContext(t *testing.T) { - baseCtx, reader, c0, _ := sparseFixture(t) - tx, err := reader.NewTx(baseCtx) - require.NoError(t, err) - defer func() { _ = tx.Done() }() - - ctx, cancel := context.WithCancel(baseCtx) - _, err = tx.WithLedgerRaw(ctx, c0.FirstLedger(), func([]byte) error { return nil }) - require.NoError(t, err) - - cancel() - ran := false - _, err = tx.WithLedgerRaw(ctx, c0.FirstLedger()+1, func([]byte) error { - ran = true - return nil - }) - require.ErrorIs(t, err, context.Canceled) - assert.False(t, ran) -} diff --git a/cmd/stellar-rpc/internal/rpcv2/adapters/view_context_test.go b/cmd/stellar-rpc/internal/rpcv2/adapters/view_context_test.go index ab45e2ced..d6d4cb33d 100644 --- a/cmd/stellar-rpc/internal/rpcv2/adapters/view_context_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/adapters/view_context_test.go @@ -43,7 +43,7 @@ func TestWithView_OneSnapshotPerRequest(t *testing.T) { assert.Equal(t, first+2, got, "the request's snapshot froze at acquisition") - _, found, err := reader.GetLedger(ctx, first+3) + _, found, err := store.GetLedger(ctx, reader, first+3) require.NoError(t, err) assert.False(t, found, "a ledger committed after the snapshot is invisible to this request") @@ -89,7 +89,7 @@ func TestWithView_TxDoneLeavesTheRequestViewAlive(t *testing.T) { require.NoError(t, err) require.NoError(t, tx.Done()) - _, found, err := reader.GetLedger(ctx, first) + _, found, err := store.GetLedger(ctx, reader, first) require.NoError(t, err) assert.True(t, found, "Done releases nothing it does not own; the request's view still serves reads") diff --git a/cmd/stellar-rpc/internal/rpcv2/backfill/process_test.go b/cmd/stellar-rpc/internal/rpcv2/backfill/process_test.go index c0f340d13..364642c65 100644 --- a/cmd/stellar-rpc/internal/rpcv2/backfill/process_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/backfill/process_test.go @@ -32,7 +32,7 @@ import ( // backend) that has the whole requested range. type fullChunkStream struct { t *testing.T - gen func(*testing.T, uint32) []byte + gen func(testing.TB, uint32) []byte } var _ ledgerbackend.LedgerStream = (*fullChunkStream)(nil) @@ -54,7 +54,7 @@ func (s *fullChunkStream) RawLedgers( // assert whether backfillSource actually read from the bulk backend (vs the pack). type fakeBackend struct { t *testing.T - gen func(*testing.T, uint32) []byte + gen func(testing.TB, uint32) []byte tip uint32 tipErr error tipFn func(context.Context) (uint32, error) // overrides tip/tipErr when set diff --git a/cmd/stellar-rpc/internal/rpcv2/chunk/chunk.go b/cmd/stellar-rpc/internal/rpcv2/chunk/chunk.go index 57bb78751..e22e9151a 100644 --- a/cmd/stellar-rpc/internal/rpcv2/chunk/chunk.go +++ b/cmd/stellar-rpc/internal/rpcv2/chunk/chunk.go @@ -23,7 +23,7 @@ const ( // It is also the per-request scan bound: one request never scans more // than one chunk's span of ledgers, touching at most two stores when the // span straddles a boundary. Every scan cap derives from this constant - // (adapters' walkSpanCap, the events pager's window); the one spelling + // (the events pager's window); the one spelling // that cannot — methods.LedgerScanLimit, in shared v1 code that must not // import this package — is pinned equal by a pairing test. LedgersPerChunk uint32 = 10_000 diff --git a/cmd/stellar-rpc/internal/rpcv2/corestate/corestate_test.go b/cmd/stellar-rpc/internal/rpcv2/corestate/corestate_test.go index 9596fb217..4b6c7293d 100644 --- a/cmd/stellar-rpc/internal/rpcv2/corestate/corestate_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/corestate/corestate_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "iter" "net/http" "net/http/httptest" "net/url" @@ -210,24 +211,14 @@ func (s stubLedgerReader) GetLatestLedgerSequence(context.Context) (uint32, erro return s.latest, nil } -func (s stubLedgerReader) GetLedger(context.Context, uint32) (xdr.LedgerCloseMeta, bool, error) { - return xdr.LedgerCloseMeta{}, false, errors.New("unused") -} - -func (s stubLedgerReader) WithLedgerRaw(context.Context, uint32, store.WithLedgerRawFn) (bool, error) { - return false, errors.New("unused") +func (s stubLedgerReader) ScanLedgers(context.Context, uint32, uint32) iter.Seq2[store.RawLedger, error] { + return func(yield func(store.RawLedger, error) bool) { yield(store.RawLedger{}, errors.New("unused")) } } func (s stubLedgerReader) GetLedgerRange(context.Context) (store.LedgerRange, error) { return store.LedgerRange{}, errors.New("unused") } -func (s stubLedgerReader) StreamLedgerRange( - context.Context, uint32, uint32, store.StreamLedgerFn, -) error { - return errors.New("unused") -} - func (s stubLedgerReader) NewTx(context.Context) (store.LedgerReaderTx, error) { return nil, errors.New("unused") } diff --git a/cmd/stellar-rpc/internal/rpcv2/eventsapi/v1_parity_test.go b/cmd/stellar-rpc/internal/rpcv2/eventsapi/v1_parity_test.go index 2b876eb86..e98a11ce2 100644 --- a/cmd/stellar-rpc/internal/rpcv2/eventsapi/v1_parity_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/eventsapi/v1_parity_test.go @@ -18,7 +18,6 @@ import ( "github.com/creachadair/jrpc2" "github.com/creachadair/jrpc2/handler" - "github.com/creachadair/jrpc2/server" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -157,10 +156,7 @@ func multiTxLCMBytes(t *testing.T, seq uint32, closeTime int64, txs []parityTx) } func newLocalClient(t *testing.T, h jrpc2.Handler) *jrpc2.Client { - t.Helper() - local := server.NewLocal(handler.Map{protocol.GetEventsMethodName: h}, nil) - t.Cleanup(func() { _ = local.Client.Close() }) - return local.Client + return rpcv2test.NewLocalClient(t, handler.Map{protocol.GetEventsMethodName: h}) } // newV1Client seeds an rpcv1 sqlite store from lcms and serves the shared v1 diff --git a/cmd/stellar-rpc/internal/rpcv2/feereplay.go b/cmd/stellar-rpc/internal/rpcv2/feereplay.go index 89032bed9..0497d0d9c 100644 --- a/cmd/stellar-rpc/internal/rpcv2/feereplay.go +++ b/cmd/stellar-rpc/internal/rpcv2/feereplay.go @@ -19,12 +19,8 @@ import ( // // The replay is the fees-only consumer of the shared walk: raw bytes from the // query router's read view → ExtractLedgerTxParts → FeesFromTxParts, events -// never computed. It deliberately does NOT go through store.LedgerReader / -// StreamLedgerFn — that seam exists for the JSON-RPC handlers and hands out -// parsed xdr.LedgerCloseMeta, which would force a full LCM decode per -// replayed ledger (the only full decode anywhere on the v2 path, paid up to -// 1,000 times per restart) just to walk data the view extractor reads for -// free off raw bytes. +// never computed. It holds the view itself, so it scans it directly rather than +// through the store.LedgerReader adapter, which takes the view from a request ctx. func replayFeeWindows(registry *query.Registry, windows *feewindow.FeeWindows, lastCommitted uint32) error { if windows == nil { return nil // no fee consumer (tests) → nothing to refill diff --git a/cmd/stellar-rpc/internal/rpcv2/ledger_reads_bench_test.go b/cmd/stellar-rpc/internal/rpcv2/ledger_reads_bench_test.go new file mode 100644 index 000000000..8fd4242f4 --- /dev/null +++ b/cmd/stellar-rpc/internal/rpcv2/ledger_reads_bench_test.go @@ -0,0 +1,226 @@ +package rpcv2 + +// BenchmarkLedgerReads times the shared ledger read methods, getTransactions, +// getLedgers and getLatestLedger, over real ledgers on three backends: a v2 +// hot chunk, a v2 frozen pack and v1 sqlite. Requests cross an in-memory +// JSON-RPC pipe, so a row covers the handler, the response encode and, on v2, +// the per-request read view, but not HTTP. Opt-in: benchLedgersEnv names a v1 +// sqlite DB holding the ledgers (the v1 daemon's DB is one); unset skips. +// +// STELLAR_RPC_BENCH_SQLITE=$PWD/soroban_rpc_full.sqlite go test ./cmd/stellar-rpc/internal/rpcv2/ \ +// -run '^$' -bench BenchmarkLedgerReads -benchmem -count 5 | tee before.txt +// +// The v2 stores are seeded from the DB once per backend and -count, before the +// timer. Compare two builds with benchstat before.txt after.txt; the resp-bytes +// metric must not move between them. + +import ( + "bytes" + "context" + "encoding/json" + "iter" + "os" + "testing" + "time" + + "github.com/creachadair/jrpc2" + "github.com/creachadair/jrpc2/handler" + "github.com/stretchr/testify/require" + + "github.com/stellar/go-stellar-sdk/network" + protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" + + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/methods" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv1/sqlitedb" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/adapters" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/catalog" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/chunk" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/geometry" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/query" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/rpcv2test" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/rpcv2/stores/hotchunk" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/store" +) + +const benchLedgersEnv = "STELLAR_RPC_BENCH_SQLITE" + +func BenchmarkLedgerReads(b *testing.B) { + f := openBenchLedgers(b) + backends := []struct { + name string + open func(*testing.B, benchLedgers) *jrpc2.Client + }{ + {"v2-hot", benchV2Hot}, + {"v2-cold", benchV2Cold}, + {"v1-sqlite", benchV1}, + } + for _, be := range backends { + b.Run(be.name, func(b *testing.B) { + start := time.Now() + client := be.open(b, f) + b.Logf("%s ready in %s", be.name, time.Since(start).Round(time.Second)) + for _, row := range benchRows(f.first.Sequence) { + b.Run(row.name, func(b *testing.B) { benchCall(b, client, row) }) + } + }) + } +} + +// benchRow is one request shape; its name is what benchstat pairs across builds. +type benchRow struct { + name string + method string + params any +} + +// benchRows: getTransactions at its default page, getLedgers at v2's cap and at +// v1 pubnet's, each in both encodings, and the getLatestLedger point read. +func benchRows(first uint32) []benchRow { + page := func(limit uint) *protocol.LedgerPaginationOptions { + return &protocol.LedgerPaginationOptions{Limit: limit} + } + var rows []benchRow + for _, enc := range []struct{ name, format string }{{"xdr", ""}, {"json", protocol.FormatJSON}} { + rows = append(rows, + benchRow{ + "getTransactions-50/" + enc.name, protocol.GetTransactionsMethodName, + protocol.GetTransactionsRequest{StartLedger: first, Pagination: page(50), Format: enc.format}, + }, + benchRow{ + "getLedgers-20/" + enc.name, protocol.GetLedgersMethodName, + protocol.GetLedgersRequest{StartLedger: first, Pagination: page(20), Format: enc.format}, + }, + benchRow{ + "getLedgers-200/" + enc.name, protocol.GetLedgersMethodName, + protocol.GetLedgersRequest{StartLedger: first, Pagination: page(200), Format: enc.format}, + }, + ) + } + return append(rows, + benchRow{"getLatestLedger", protocol.GetLatestLedgerMethodName, protocol.GetLatestLedgerRequest{}}) +} + +// benchCall times one request shape and reports the response size; RawMessage +// reuses its buffer, so the client-side copy is not in allocs/op. +func benchCall(b *testing.B, c *jrpc2.Client, row benchRow) { + b.ReportAllocs() + var raw json.RawMessage + for b.Loop() { + if err := c.CallResult(context.Background(), row.method, row.params, &raw); err != nil { + b.Fatal(err) + } + } + b.ReportMetric(float64(len(raw)), "resp-bytes") +} + +// benchLedgers is the fixture: a v1 sqlite DB and the ledger run it holds, in one chunk. +type benchLedgers struct { + db *sqlitedb.DB + first, last store.LedgerInfo +} + +func openBenchLedgers(b *testing.B) benchLedgers { + path := os.Getenv(benchLedgersEnv) + if path == "" { + b.Skipf("set %s to a v1 sqlite DB holding real ledgers", benchLedgersEnv) + } + db, err := sqlitedb.OpenSQLiteDB(path) + require.NoError(b, err) + b.Cleanup(func() { _ = db.Close() }) + lr, err := sqlitedb.NewLedgerReader(db).GetLedgerRange(context.Background()) + require.NoError(b, err) + require.Equal(b, chunk.IDFromLedger(lr.FirstLedger.Sequence), chunk.IDFromLedger(lr.LastLedger.Sequence), + "the fixture's ledgers must lie in one chunk") + return benchLedgers{db: db, first: lr.FirstLedger, last: lr.LastLedger} +} + +func (f benchLedgers) chunkID() chunk.ID { return chunk.IDFromLedger(f.first.Sequence) } + +// chunkRun pads the fixture with zero-tx ledgers from the chunk's first ledger, +// where hot ingestion and cold packs must start. Padding close times follow +// pubnet's cadence. +func (f benchLedgers) chunkRun(b *testing.B) iter.Seq[[]byte] { + const cadence = 5 // seconds between ledgers + return func(yield func([]byte) bool) { + for seq := f.chunkID().FirstLedger(); seq < f.first.Sequence; seq++ { + closeTime := f.first.CloseTime - cadence*int64(f.first.Sequence-seq) + if !yield(rpcv2test.ZeroTxLCMBytesAt(b, seq, closeTime)) { + return + } + } + for raw := range f.raw(b) { + if !yield(raw) { + return + } + } + } +} + +// raw streams the fixture's ledgers as marshaled bytes; the v2 seeders assign +// sequences by position, so the run must be contiguous. +func (f benchLedgers) raw(b *testing.B) iter.Seq[[]byte] { + return func(yield func([]byte) bool) { + next := f.first.Sequence + reader := sqlitedb.NewLedgerReader(f.db) + for l, err := range reader.ScanLedgers(context.Background(), f.first.Sequence, f.last.Sequence) { + require.NoError(b, err) + require.Equal(b, next, l.Sequence, "the fixture must be one contiguous run") + next++ + if !yield(bytes.Clone(l.Raw)) { // Raw is on loan until the next step + return + } + } + } +} + +// benchV2Hot serves over a hot chunk holding the fixture. +func benchV2Hot(b *testing.B, f benchLedgers) *jrpc2.Client { + cat, _ := rpcv2test.OpenTestCatalog(b, geometry.ChunksPerTxhashIndex) + var live *hotchunk.DB + rpcv2test.SeedHotChunkSeq(b, cat, f.chunkID(), func(db *hotchunk.DB) { live = db }, f.chunkRun(b)) + return benchV2Client(b, cat, f.chunkID(), live, f.last.Sequence) +} + +// benchV2Cold serves over a frozen pack holding the fixture, with the tip on +// its last real ledger like the other backends; the registry's live chunk is +// the empty one after it, created the way ingestion does. +func benchV2Cold(b *testing.B, f benchLedgers) *jrpc2.Client { + cat, _ := rpcv2test.OpenTestCatalog(b, geometry.ChunksPerTxhashIndex) + c := f.chunkID() + rpcv2test.WriteFrozenLedgerPackSeq(b, cat, c, f.chunkRun(b)) + live, err := openHotDBForChunk(cat, c+1, silentLogger()) + require.NoError(b, err) + return benchV2Client(b, cat, c, live, f.last.Sequence) +} + +// benchV2Client opens the serving registry as startup does and serves the +// handlers behind the per-request read view. +func benchV2Client( + b *testing.B, cat *catalog.Catalog, floor chunk.ID, live *hotchunk.DB, lastCommitted uint32, +) *jrpc2.Client { + registry, err := query.OpenRegistry(cat, geometry.NewRetention(0, floor), live, lastCommitted) + require.NoError(b, err) + b.Cleanup(registry.Close) + require.NoError(b, adapters.SeedCloseTimes(registry)) + handlers := benchHandlers(adapters.NewLedgerReader()) + for name, h := range handlers { + handlers[name] = wrapAdapterRequest(h, registry) + } + return rpcv2test.NewLocalClient(b, handlers) +} + +// benchV1 serves over the fixture DB itself. +func benchV1(b *testing.B, f benchLedgers) *jrpc2.Client { + return rpcv2test.NewLocalClient(b, benchHandlers(sqlitedb.NewLedgerReader(f.db))) +} + +// benchHandlers is the three read methods over reader, with v1 pubnet's page caps. +func benchHandlers(reader store.LedgerReader) handler.Map { + logger := silentLogger() + return handler.Map{ + protocol.GetTransactionsMethodName: methods.NewGetTransactionsHandler( + logger, reader, 200, 50, network.PublicNetworkPassphrase), + protocol.GetLedgersMethodName: methods.NewGetLedgersHandler(reader, 200, 50, nil, logger), + protocol.GetLatestLedgerMethodName: methods.NewGetLatestLedgerHandler(reader), + } +} diff --git a/cmd/stellar-rpc/internal/rpcv2/query/registry.go b/cmd/stellar-rpc/internal/rpcv2/query/registry.go index 8a951f7cf..07744aca0 100644 --- a/cmd/stellar-rpc/internal/rpcv2/query/registry.go +++ b/cmd/stellar-rpc/internal/rpcv2/query/registry.go @@ -303,7 +303,7 @@ type ReadView struct { // chunk boundary the hot key can flip ready before the handle publishes, and a // prune can retire a chunk whose handle this view already loaded. What keeps // skewed reads correct are the gates downstream: every lookup is checked -// against the view's window (the adapters' inWindow / windowGatedIndex), and a +// against the view's window (the adapters' scanView clamp / windowGatedIndex), and a // chunk without a serving store resolves to ErrUnavailable. A read path that // skips those gates cannot lean on this ordering. (See the design's Read views // section.) diff --git a/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go b/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go index 0c6ff3992..d07b9de27 100644 --- a/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go +++ b/cmd/stellar-rpc/internal/rpcv2/rpcv2test/rpcv2test.go @@ -9,11 +9,14 @@ import ( "context" "encoding/binary" "io" + "iter" "os" "path/filepath" "slices" "testing" + "github.com/creachadair/jrpc2" + "github.com/creachadair/jrpc2/server" "github.com/sirupsen/logrus" "github.com/stretchr/testify/require" @@ -44,14 +47,14 @@ func SilentLogger() *supportlog.Entry { // artifacts under another, cpi-wide tx-hash indexes, silent logger, closed on // test cleanup. Returns the catalog and the artifact root. Most tests pass // geometry.ChunksPerTxhashIndex for cpi. -func OpenTestCatalog(t *testing.T, cpi uint32) (*catalog.Catalog, string) { +func OpenTestCatalog(t testing.TB, cpi uint32) (*catalog.Catalog, string) { t.Helper() return OpenTestCatalogWith(t, cpi, SilentLogger()) } // OpenTestCatalogWith is OpenTestCatalog with the caller's logger, for tests // that assert on what the catalog logs. -func OpenTestCatalogWith(t *testing.T, cpi uint32, logger *supportlog.Entry) (*catalog.Catalog, string) { +func OpenTestCatalogWith(t testing.TB, cpi uint32, logger *supportlog.Entry) (*catalog.Catalog, string) { t.Helper() artifactRoot := t.TempDir() idxLayout, err := geometry.NewTxHashIndexLayout(cpi) @@ -73,14 +76,25 @@ func OpenTestCatalogWith(t *testing.T, cpi uint32, logger *supportlog.Entry) (*c // func(db *hotchunk.DB) { registry.PublishHandle(c, db) }, // lcms...) func SeedHotChunkLCMs( - t *testing.T, cat *catalog.Catalog, c chunk.ID, publish func(*hotchunk.DB), lcms ...[]byte, + t testing.TB, cat *catalog.Catalog, c chunk.ID, publish func(*hotchunk.DB), lcms ...[]byte, +) { + t.Helper() + SeedHotChunkSeq(t, cat, c, publish, slices.Values(lcms)) +} + +// SeedHotChunkSeq is SeedHotChunkLCMs over a stream of raw LCMs, for fixtures +// too large to hold at once. +func SeedHotChunkSeq( + t testing.TB, cat *catalog.Catalog, c chunk.ID, publish func(*hotchunk.DB), lcms iter.Seq[[]byte], ) { t.Helper() db, err := hotchunk.Open(cat.Layout().HotChunkPath(c), c, SilentLogger()) require.NoError(t, err) t.Cleanup(func() { _ = db.Close() }) - for i, raw := range lcms { - IngestLedger(t, db, c.FirstLedger()+uint32(i), raw) + seq := c.FirstLedger() + for raw := range lcms { + IngestLedger(t, db, seq, raw) + seq++ } require.NoError(t, cat.FlipHotReady(c)) publish(db) @@ -90,20 +104,37 @@ func SeedHotChunkLCMs( // lcms from the chunk's first ledger, and flips the chunk's ledgers artifact // frozen. AppendLedger stores raw bytes, so marker payloads work as well as // real LCMs. -func WriteFrozenLedgerPack(t *testing.T, cat *catalog.Catalog, c chunk.ID, lcms ...[]byte) { +func WriteFrozenLedgerPack(t testing.TB, cat *catalog.Catalog, c chunk.ID, lcms ...[]byte) { + t.Helper() + WriteFrozenLedgerPackSeq(t, cat, c, slices.Values(lcms)) +} + +// WriteFrozenLedgerPackSeq is WriteFrozenLedgerPack over a stream of raw LCMs. +func WriteFrozenLedgerPackSeq(t testing.TB, cat *catalog.Catalog, c chunk.ID, lcms iter.Seq[[]byte]) { t.Helper() path := cat.Layout().LedgerPackPath(c) require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o755)) w, err := ledger.NewColdWriter(path, c.FirstLedger(), ledger.ColdWriterOptions{}) require.NoError(t, err) defer func() { _ = w.Close() }() - for i, raw := range lcms { - require.NoError(t, w.AppendLedger(c.FirstLedger()+uint32(i), raw)) + seq := c.FirstLedger() + for raw := range lcms { + require.NoError(t, w.AppendLedger(seq, raw)) + seq++ } require.NoError(t, w.Commit()) require.NoError(t, cat.FlipChunkFrozen(c, geometry.KindLedgers)) } +// NewLocalClient serves handlers over an in-memory JSON-RPC pipe: a request +// crosses the handler and the response encode, but not HTTP. +func NewLocalClient(t testing.TB, handlers jrpc2.Assigner) *jrpc2.Client { + t.Helper() + local := server.NewLocal(handlers, nil) + t.Cleanup(func() { _ = local.Close() }) + return local.Client +} + // SymbolContractEvent returns a contract event whose topics and data // are ScSymbols: the shape event-query tests build fixtures from and // later match against by the data label. @@ -131,7 +162,7 @@ func SymbolContractEvent(contractID xdr.ContractId, data string, topics ...strin // shared ExtractLedgerTxParts walk, then hotchunk's atomic write over its // output. Tests that just need ledgers in a hot DB use this instead of // hand-running the walk. -func IngestLedger(t *testing.T, db *hotchunk.DB, seq uint32, raw []byte) { +func IngestLedger(t testing.TB, db *hotchunk.DB, seq uint32, raw []byte) { t.Helper() txParts, err := sdkingest.ExtractLedgerTxParts(xdr.LedgerCloseMetaView(raw)) require.NoError(t, err) @@ -155,14 +186,14 @@ func RetentionFor(t *testing.T, cat *catalog.Catalog, size uint32) geometry.Rete // ZeroTxLCMBytes returns the marshaled bytes of a minimal, zero-transaction // LedgerCloseMeta (V2) for ledger seq — the fixture ingestion/backfill/lifecycle // tests feed in when they need a valid but empty ledger. -func ZeroTxLCMBytes(t *testing.T, seq uint32) []byte { +func ZeroTxLCMBytes(t testing.TB, seq uint32) []byte { t.Helper() return ZeroTxLCMBytesAt(t, seq, 0) } // ZeroTxLCMBytesAt is ZeroTxLCMBytes with an explicit close time, for tests // that assert against a known close-time value. -func ZeroTxLCMBytesAt(t *testing.T, seq uint32, closeTimeUnix int64) []byte { +func ZeroTxLCMBytesAt(t testing.TB, seq uint32, closeTimeUnix int64) []byte { t.Helper() return V2LCMBytes(t, seq, closeTimeUnix, nil, nil) } @@ -172,7 +203,7 @@ func ZeroTxLCMBytesAt(t *testing.T, seq uint32, closeTimeUnix int64) []byte { // at all when empty), and processing as the apply results. Every V2 fixture // builder delegates here so the skeleton exists once. func V2LCMBytes( - t *testing.T, seq uint32, closeTimeUnix int64, + t testing.TB, seq uint32, closeTimeUnix int64, envelopes []xdr.TransactionEnvelope, processing []xdr.TransactionResultMetaV1, ) []byte { t.Helper() diff --git a/cmd/stellar-rpc/internal/store/ledger.go b/cmd/stellar-rpc/internal/store/ledger.go index c5ad5a4b8..7fda2b7a4 100644 --- a/cmd/stellar-rpc/internal/store/ledger.go +++ b/cmd/stellar-rpc/internal/store/ledger.go @@ -3,6 +3,7 @@ package store import ( "context" "errors" + "iter" protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" "github.com/stellar/go-stellar-sdk/xdr" @@ -11,8 +12,6 @@ import ( // ErrEmptyDB is returned when the storage backend holds no ledgers yet. var ErrEmptyDB = errors.New("DB is empty") -type StreamLedgerFn func(xdr.LedgerCloseMeta) error - // LedgerInfo identifies one ledger: its sequence number and close time. type LedgerInfo struct { Sequence uint32 @@ -35,10 +34,10 @@ func (lr LedgerRange) ToLedgerSeqRange() protocol.LedgerSeqRange { // LedgerReader is the serving-side read contract every storage backend // implements. Handlers depend on this interface, never on a concrete backend. type LedgerReader interface { - GetLedger(ctx context.Context, sequence uint32) (xdr.LedgerCloseMeta, bool, error) - WithLedgerRaw(ctx context.Context, sequence uint32, fn WithLedgerRawFn) (found bool, err error) + // ScanLedgers reads whatever the request already sees (the live store on + // v1 and the request's read view on v2), without its own snapshot. + LedgerScanner GetLedgerRange(ctx context.Context) (LedgerRange, error) - StreamLedgerRange(ctx context.Context, startLedger uint32, endLedger uint32, f StreamLedgerFn) error NewTx(ctx context.Context) (LedgerReaderTx, error) GetLatestLedgerSequence(ctx context.Context) (uint32, error) } @@ -47,27 +46,84 @@ type LedgerReader interface { // are valid only inside the call, read-only. Copy whatever outlives fn. type WithLedgerRawFn func(raw []byte) error -// LedgerReaderTx is a read-only snapshot of the ledger store. Call Done to -// release it. -// -// WithLedgerRaw is a walk, not a free-form point read: calls must be -// ascending and contiguous from the first call's sequence, and read at most -// methods.LedgerScanLimit ledgers per Tx. The v1 (SQL) backend accepts any -// pattern, while the v2 backend only walks from a forward iterator primed on -// the first call. fn borrows the marshaled LCM under WithLedgerRawFn's loan -// terms. found=false means fn never ran; fn's own error comes back verbatim -// with found=true. +// LedgerScanner is the contract's one read idiom, embedded by LedgerReader and LedgerReaderTx. +type LedgerScanner interface { + // ScanLedgers yields the stored ledgers in [start, end], ascending, as one + // forward read that ends when the loop body breaks. Ledgers the source + // does not hold, below its oldest, above its latest, or missing inside the + // range, are not yielded, so callers should check the sequences they receive. + // A non-nil error ends the stream and the RawLedger beside it is zero. + // The body runs inside the source's read (a v2 hot range scan holds the + // store's read lock across every step): decode and copy, never call back into the store. + ScanLedgers(ctx context.Context, start, end uint32) iter.Seq2[RawLedger, error] +} + +// WithLedgerRaw lends one ledger's bytes under RawLedger's loan: a scan of one. +// found is false when the ledger is absent and fn never ran; fn's own error +// comes back verbatim with found true. +func WithLedgerRaw(ctx context.Context, s LedgerScanner, seq uint32, fn WithLedgerRawFn) (bool, error) { + for l, err := range s.ScanLedgers(ctx, seq, seq) { + if err != nil { + return false, err + } + return true, fn(l.Raw) + } + return false, nil +} + +// GetLedger decodes one ledger, or reports it absent. +func GetLedger(ctx context.Context, s LedgerScanner, seq uint32) (xdr.LedgerCloseMeta, bool, error) { + var lcm xdr.LedgerCloseMeta + found, err := WithLedgerRaw(ctx, s, seq, lcm.UnmarshalBinary) + if err != nil { + return xdr.LedgerCloseMeta{}, false, err + } + return lcm, found, nil +} + +// ScanLedgersFrom is the inverse adapter: a scan over a per-sequence lookup, for +// a source that only has point reads. get reports an absent ledger as false. +func ScanLedgersFrom( + start, end uint32, get func(seq uint32) (xdr.LedgerCloseMeta, bool, error), +) iter.Seq2[RawLedger, error] { + return func(yield func(RawLedger, error) bool) { + for seq := start; seq <= end; seq++ { + lcm, found, err := get(seq) + if err != nil { + yield(RawLedger{}, err) + return + } + if found { + raw, err := lcm.MarshalBinary() + if err != nil { + yield(RawLedger{}, err) + return + } + if !yield(RawLedger{Sequence: seq, Raw: raw}, nil) { + return + } + } + if seq == end { // seq++ would wrap at MaxUint32 + return + } + } + } +} + +// RawLedger is one ledger as ScanLedgers yields it. Raw is the read-only LCM +// bytes on loan, valid only inside the loop body that received it, overwritten +// by the next step. Copy whatever outlives the body. +type RawLedger struct { + Sequence uint32 + Raw []byte +} + +// LedgerReaderTx is a read-only snapshot of the ledger store: GetLedgerRange +// and every ScanLedgers on it answer from the same committed state. Call Done +// to release it, however the loops ended. type LedgerReaderTx interface { - WithLedgerRaw(ctx context.Context, sequence uint32, fn WithLedgerRawFn) (bool, error) + LedgerScanner + // GetLedgerRange is the snapshot's oldest and latest ledger. GetLedgerRange(ctx context.Context) (LedgerRange, error) - BatchGetLedgers(ctx context.Context, start uint32, end uint32) ([]LedgerMetadataChunk, error) Done() error } - -// LedgerMetadataChunk is one ledger as getLedgers serves it: the marshaled -// LedgerCloseMeta plus the marshaled LedgerHeaderHistoryEntry sliced out of -// it. Both stay raw bytes because the XDR wire format base64s them as-is. -type LedgerMetadataChunk struct { - HeaderRaw []byte - Lcm []byte -} diff --git a/cmd/stellar-rpc/internal/store/ledger_test.go b/cmd/stellar-rpc/internal/store/ledger_test.go new file mode 100644 index 000000000..dc9103386 --- /dev/null +++ b/cmd/stellar-rpc/internal/store/ledger_test.go @@ -0,0 +1,36 @@ +package store + +import ( + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/stellar/go-stellar-sdk/xdr" +) + +// TestScanLedgersFrom_EndsAtMaxUint32 pins that an inclusive scan ending at +// MaxUint32 stops after the endpoint instead of wrapping to 0, whether or not +// the endpoint is present. +func TestScanLedgersFrom_EndsAtMaxUint32(t *testing.T) { + const end = uint32(math.MaxUint32) + var asked []uint32 + get := func(seq uint32) (xdr.LedgerCloseMeta, bool, error) { + asked = append(asked, seq) + if seq == end { + return xdr.LedgerCloseMeta{}, false, nil + } + lcm := xdr.LedgerCloseMeta{V: 0, V0: &xdr.LedgerCloseMetaV0{}} + lcm.V0.LedgerHeader.Header.LedgerSeq = xdr.Uint32(seq) + return lcm, true, nil + } + + var got []uint32 + for l, err := range ScanLedgersFrom(end-1, end, get) { + require.NoError(t, err) + got = append(got, l.Sequence) + } + assert.Equal(t, []uint32{end - 1, end}, asked) + assert.Equal(t, []uint32{end - 1}, got) +} diff --git a/cmd/stellar-rpc/internal/xdr2json/conversion.go b/cmd/stellar-rpc/internal/xdr2json/conversion.go index 321476c6b..e7cd1c981 100644 --- a/cmd/stellar-rpc/internal/xdr2json/conversion.go +++ b/cmd/stellar-rpc/internal/xdr2json/conversion.go @@ -5,6 +5,7 @@ package xdr2json // See preflight.go for add'l explanations: // Note: no blank lines allowed. #include +#include #include "../../lib/xdr2json.h" #cgo windows,amd64 LDFLAGS: -L${SRCDIR}/../../../../target/x86_64-pc-windows-gnu/release-with-panic-unwind/ -lxdr2json -lntdll -static -lws2_32 -lbcrypt -luserenv #cgo darwin,amd64 LDFLAGS: -L${SRCDIR}/../../../../target/x86_64-apple-darwin/release-with-panic-unwind/ -lxdr2json -ldl -lm @@ -103,7 +104,6 @@ func ConvertJSON(xdr any, js json.RawMessage) ([]byte, error) { } func convertAnyBytes(xdrTypeName string, field []byte) (json.RawMessage, error) { - var jsonStr, errStr string goRawXdr := CXDR(field) defer FreeGoXDR(goRawXdr) @@ -113,14 +113,14 @@ func convertAnyBytes(xdrTypeName string, field []byte) (json.RawMessage, error) result := C.xdr_to_json(b, goRawXdr) defer C.free_conversion_result(result) - jsonStr = C.GoString(result.json) - errStr = C.GoString(result.error) - - if errStr != "" { - return json.RawMessage(jsonStr), errors.New(errStr) + if errStr := C.GoString(result.error); errStr != "" { + return json.RawMessage(C.GoString(result.json)), errors.New(errStr) } - - return json.RawMessage(jsonStr), nil + if result.json == nil { + return json.RawMessage{}, nil + } + // One copy out of C; GoString followed by []byte(string) would copy the JSON twice. + return C.GoBytes(unsafe.Pointer(result.json), C.int(C.strlen(result.json))), nil } // CXDR is ripped directly from preflight.go to avoid a dependency.