diff --git a/cmd/stellar-rpc/internal/methods/event_filter.go b/cmd/stellar-rpc/internal/methods/event_filter.go new file mode 100644 index 000000000..e159d3891 --- /dev/null +++ b/cmd/stellar-rpc/internal/methods/event_filter.go @@ -0,0 +1,111 @@ +package methods + +import ( + "bytes" + "fmt" + "slices" + + protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" + "github.com/stellar/go-stellar-sdk/strkey" + "github.com/stellar/go-stellar-sdk/xdr" +) + +// eventFilters is a validated GetEventsRequest.Filters with comparison values in wire form, +// matched against a view's raw bytes the way GetEventsRequest.Matches matches a decoded event. +// Empty matches every event. +type eventFilters []eventFilter + +// eventFilter is one protocol.EventFilter; an empty field matches everything. +type eventFilter struct { + types []xdr.ContractEventType + contractIDs [][]byte + topics []topicFilter +} + +// topicFilter is one protocol.TopicFilter: a nil value is "*", atLeast is a trailing "**". +type topicFilter struct { + values [][]byte + atLeast bool +} + +func compileFilters(in []protocol.EventFilter) (eventFilters, error) { + out := make(eventFilters, 0, len(in)) + for _, f := range in { + c := eventFilter{} + for _, name := range f.EventType.Keys() { + typ, ok := protocol.GetEventTypeXDRFromEventType()[name] + if !ok { // Valid admits only contract and system + return nil, fmt.Errorf("unsupported event type %q", name) + } + c.types = append(c.types, typ) + } + for _, id := range f.ContractIDs { + raw, err := strkey.Decode(strkey.VersionByteContract, id) + if err != nil { // unreachable: Valid decoded it already + return nil, fmt.Errorf("invalid contract ID: %v", id) + } + c.contractIDs = append(c.contractIDs, raw) + } + for _, tf := range f.Topics { + t, err := compileTopicFilter(tf) + if err != nil { + return nil, err + } + c.topics = append(c.topics, t) + } + out = append(out, c) + } + return out, nil +} + +func compileTopicFilter(tf protocol.TopicFilter) (topicFilter, error) { + var t topicFilter + if n := len(tf); n > 0 && tf[n-1].Wildcard != nil && *tf[n-1].Wildcard == protocol.WildCardZeroOrMore { + tf, t.atLeast = tf[:n-1], true + } + t.values = make([][]byte, len(tf)) + for i, s := range tf { + if s.ScVal == nil { // "*", or a segment Valid would have rejected + continue + } + raw, err := s.ScVal.MarshalBinary() + if err != nil { + return topicFilter{}, fmt.Errorf("failed to marshal segment: %w", err) + } + t.values[i] = raw + } + return t, nil +} + +// matchHeader reports whether some filter accepts the type and contract id. It never rejects +// an event match would accept, so a false answer skips reading the topics. +func (fs eventFilters) matchHeader(h eventHead) bool { + return len(fs) == 0 || slices.ContainsFunc(fs, func(f eventFilter) bool { return f.matchHeader(h) }) +} + +// match reports whether some filter accepts the whole event. +func (fs eventFilters) match(h eventHead, topics [][]byte) bool { + return len(fs) == 0 || slices.ContainsFunc(fs, func(f eventFilter) bool { + return f.matchHeader(h) && (len(f.topics) == 0 || + slices.ContainsFunc(f.topics, func(t topicFilter) bool { return t.matches(topics) })) + }) +} + +func (f eventFilter) matchHeader(h eventHead) bool { + return (len(f.types) == 0 || slices.Contains(f.types, h.typ)) && + (len(f.contractIDs) == 0 || slices.ContainsFunc(f.contractIDs, func(id []byte) bool { + return bytes.Equal(id, h.cid) + })) +} + +func (t topicFilter) matches(topics [][]byte) bool { + if len(topics) < len(t.values) || (!t.atLeast && len(topics) != len(t.values)) { + return false + } + for i, want := range t.values { + if want != nil && !bytes.Equal(topics[i], want) { + return false + } + } + return true +} diff --git a/cmd/stellar-rpc/internal/methods/event_filter_test.go b/cmd/stellar-rpc/internal/methods/event_filter_test.go new file mode 100644 index 000000000..c3c4b31a2 --- /dev/null +++ b/cmd/stellar-rpc/internal/methods/event_filter_test.go @@ -0,0 +1,78 @@ +package methods + +import ( + "bytes" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + protocol "github.com/stellar/go-stellar-sdk/protocols/rpc" + "github.com/stellar/go-stellar-sdk/strkey" + "github.com/stellar/go-stellar-sdk/xdr" +) + +// TestEventFilters pins the port of GetEventsRequest.Matches: filters OR, fields AND, +// "*" is any one topic, a trailing "**" relaxes the arity, and matchHeader never rejects +// an event match accepts. +func TestEventFilters(t *testing.T) { + sym := func(s string) xdr.ScVal { return xdr.ScVal{Type: xdr.ScValTypeScvSymbol, Sym: (*xdr.ScSymbol)(&s)} } + raw := func(v xdr.ScVal) []byte { b, err := v.MarshalBinary(); require.NoError(t, err); return b } + seg := func(s string) protocol.SegmentFilter { v := sym(s); return protocol.SegmentFilter{ScVal: &v} } + wild := func(w string) protocol.SegmentFilter { return protocol.SegmentFilter{Wildcard: &w} } + cidA, cidB := bytes.Repeat([]byte{0xAA}, 32), bytes.Repeat([]byte{0xBB}, 32) + strkeyA := strkey.MustEncode(strkey.VersionByteContract, cidA) + strkeyB := strkey.MustEncode(strkey.VersionByteContract, cidB) + topics := func(tf ...protocol.SegmentFilter) []protocol.TopicFilter { return []protocol.TopicFilter{tf} } + contract := protocol.EventTypeSet{protocol.EventTypeContract: nil} + + xferAlice := [][]byte{raw(sym("xfer")), raw(sym("alice"))} + headA := eventHead{typ: xdr.ContractEventTypeContract, cid: cidA} + headB := eventHead{typ: xdr.ContractEventTypeSystem, cid: cidB} + + for name, tc := range map[string]struct { + in []protocol.EventFilter + head eventHead + topics [][]byte + want bool + }{ + "no filters match everything": {nil, headB, nil, true}, + "contract id": {[]protocol.EventFilter{{ContractIDs: []string{strkeyA}}}, headA, nil, true}, + "contract id mismatch": {[]protocol.EventFilter{{ContractIDs: []string{strkeyA}}}, headB, nil, false}, + "type": {[]protocol.EventFilter{{EventType: contract}}, headB, nil, false}, + "fields AND within a filter": { + []protocol.EventFilter{{ + EventType: contract, ContractIDs: []string{strkeyA}, + Topics: topics(seg("xfer"), wild("*")), + }}, headA, xferAlice, true, + }, + "filters OR": { + []protocol.EventFilter{ + {ContractIDs: []string{strkeyA}, Topics: topics(seg("mint"))}, + {Topics: topics(seg("xfer"), seg("alice"))}, + }, headA, xferAlice, true, + }, + "header and topics must come from one filter": { + []protocol.EventFilter{ + {ContractIDs: []string{strkeyA}, Topics: topics(seg("mint"))}, + {ContractIDs: []string{strkeyB}, Topics: topics(seg("xfer"), wild("**"))}, + }, + headA, xferAlice, false, + }, + "exact arity": {[]protocol.EventFilter{{Topics: topics(seg("xfer"))}}, headA, xferAlice, false}, + "trailing ** relaxes": {[]protocol.EventFilter{{Topics: topics(seg("xfer"), wild("**"))}}, headA, xferAlice, true}, + "** needs the prefix": { + []protocol.EventFilter{{Topics: topics(seg("a"), seg("b"), seg("c"), wild("**"))}}, headA, xferAlice, false, + }, + } { + t.Run(name, func(t *testing.T) { + fs, err := compileFilters(tc.in) + require.NoError(t, err) + got := fs.match(tc.head, tc.topics) + assert.Equal(t, tc.want, got) + if got { + assert.True(t, fs.matchHeader(tc.head), "matchHeader must not reject a matching event") + } + }) + } +} diff --git a/cmd/stellar-rpc/internal/methods/get_events.go b/cmd/stellar-rpc/internal/methods/get_events.go index 6ff25135e..13604da29 100644 --- a/cmd/stellar-rpc/internal/methods/get_events.go +++ b/cmd/stellar-rpc/internal/methods/get_events.go @@ -2,9 +2,9 @@ package methods import ( "context" - "encoding/json" + "encoding/base64" "fmt" - "strconv" + "math" "time" "github.com/creachadair/jrpc2" @@ -117,13 +117,6 @@ func combineTopics(filters []protocol.EventFilter) (store.TopicFilters, error) { return topicFilters, nil } -type entry struct { - cursor protocol.Cursor - ledgerCloseTimestamp int64 - event xdr.DiagnosticEvent - txHash *xdr.Hash -} - // TODO: remove this linter exclusions // //nolint:cyclop,funlen @@ -176,8 +169,6 @@ func (h eventsRPCHandler) getEvents(ctx context.Context, request protocol.GetEve } } - found := make([]entry, 0, limit) - contractIDs, err := combineContractIDs(request.Filters) if err != nil { return protocol.GetEventsResponse{}, &jrpc2.Error{ @@ -194,36 +185,60 @@ func (h eventsRPCHandler) getEvents(ctx context.Context, request protocol.GetEve eventTypes := combineEventTypes(request.Filters) - // Scan function to apply filters - var eventScanFunction store.ScanFunction = func( - event xdr.DiagnosticEvent, cursor protocol.Cursor, ledgerCloseTimestamp int64, txHash *xdr.Hash, - ) bool { - if request.Matches(event) { - found = append(found, entry{cursor, ledgerCloseTimestamp, event, txHash}) - } - return uint(len(found)) < limit - } - - err = h.dbReader.GetEvents(ctx, cursorRange, contractIDs, topics, eventTypes, eventScanFunction) + filters, err := compileFilters(request.Filters) if err != nil { return protocol.GetEventsResponse{}, &jrpc2.Error{ - Code: jrpc2.InvalidRequest, Message: err.Error(), + Code: jrpc2.InvalidParams, Message: err.Error(), } } - results := make([]protocol.EventInfo, 0, len(found)) - for _, entry := range found { - info, err := eventInfoForEvent( - entry.event, - entry.cursor, - time.Unix(entry.ledgerCloseTimestamp, 0).UTC().Format(time.RFC3339), - entry.txHash.HexString(), - request.Format, - ) + results := []protocol.EventInfo{} + // procErr keeps the callback's own error out of the reader's wrap, as getTransactions does, + // so a render failure stays a plain (system) error rather than an InvalidRequest. + var procErr error + var eventViewScanFunction store.ViewScanFunction = func( + eventView xdr.DiagnosticEventView, cursor protocol.Cursor, ledgerCloseTimestamp int64, txHash *xdr.Hash, + ) (bool, error) { + var event xdr.ContractEventView + if event, procErr = eventView.Event(); procErr != nil { + return false, procErr + } + // Fields are pulled off the view once and feed both the match and the render. + // Topics wait until type and contract id pass, so a rejected event never sizes them. + head, err := eventHeader(event) + if err != nil { + procErr = errors.Wrap(err, "could not parse event") + return false, procErr + } + if !filters.matchHeader(head) { + return true, nil + } + body, err := eventBody(head.v0) if err != nil { - return protocol.GetEventsResponse{}, errors.Wrap(err, "could not parse event") + procErr = errors.Wrap(err, "could not parse event") + return false, procErr + } + if !filters.match(head, body.topics) { + return true, nil + } + info, err := eventInfo(head, body, cursor, + time.Unix(ledgerCloseTimestamp, 0).UTC().Format(time.RFC3339), txHash.HexString(), request.Format) + if err != nil { + procErr = errors.Wrap(err, "could not parse event") + return false, procErr } results = append(results, info) + return uint(len(results)) < limit, nil + } + + err = h.dbReader.GetEvents(ctx, cursorRange, contractIDs, topics, eventTypes, eventViewScanFunction) + switch { + case procErr != nil: + return protocol.GetEventsResponse{}, procErr + case err != nil: + return protocol.GetEventsResponse{}, &jrpc2.Error{ + Code: jrpc2.InvalidRequest, Message: err.Error(), + } } var cursor string @@ -250,29 +265,88 @@ func (h eventsRPCHandler) getEvents(ctx context.Context, request protocol.GetEve }, nil } -func eventInfoForEvent( - event xdr.DiagnosticEvent, - cursor protocol.Cursor, - ledgerClosedAt, txHash, format string, +// EventInfoFromView renders one stored ContractEvent into the v1 wire type; rpcv2's eventsapi wraps it. +func EventInfoFromView( + ev xdr.ContractEventView, cursor protocol.Cursor, ledgerClosedAt, txHash, format string, ) (protocol.EventInfo, error) { - v0, ok := event.Event.Body.GetV0() - if !ok { - return protocol.EventInfo{}, errors.New("unknown event version") + head, err := eventHeader(ev) + if err != nil { + return protocol.EventInfo{}, errors.Wrap(err, "malformed event") + } + body, err := eventBody(head.v0) + if err != nil { + return protocol.EventInfo{}, errors.Wrap(err, "malformed event") } + return eventInfo(head, body, cursor, ledgerClosedAt, txHash, format) +} - eventType, ok := protocol.GetEventTypeFromEventTypeXDR()[event.Event.Type] - if !ok { - return protocol.EventInfo{}, fmt.Errorf("unknown XDR ContractEventType type: %d", event.Event.Type) +// eventHead is the event's fields above the body, plus the V0 body view to read on demand. +type eventHead struct { + typ xdr.ContractEventType + cid []byte // nil when absent + v0 xdr.ContractEventV0View +} + +// eventV0 is the V0 body's topics and data as raw XDR. +type eventV0 struct { + topics [][]byte + data []byte +} + +// eventHeader locates type, contract id and body in one pass. Fields returns views trimmed +// to their extent, so their bytes are the raw XDR and Raw() would only re-walk them. +func eventHeader(ev xdr.ContractEventView) (eventHead, error) { + f, err := ev.Fields() + if err != nil { + return eventHead{}, err + } + var h eventHead + if h.typ, err = f.Type.Value(); err != nil { + return eventHead{}, err } + if cid, ok, err := f.ContractId.Unwrap(); err != nil { + return eventHead{}, err + } else if ok { + h.cid = []byte(cid) + } + if h.v0, err = f.Body.V0(); err != nil { // fails on a non-V0 body, replacing "unknown event version" + return eventHead{}, err + } + return h, nil +} - ledger, err := strconv.ParseInt(strconv.FormatUint(uint64(cursor.Ledger), 10), 10, 32) +// eventBody locates topics and data in one pass; All returns each topic trimmed. +func eventBody(v0 xdr.ContractEventV0View) (eventV0, error) { + f, err := v0.Fields() + if err != nil { + return eventV0{}, err + } + views, err := f.Topics.All() if err != nil { + return eventV0{}, err + } + b := eventV0{topics: make([][]byte, len(views)), data: []byte(f.Data)} + for i, t := range views { + b.topics[i] = []byte(t) + } + return b, nil +} + +func eventInfo( + head eventHead, body eventV0, cursor protocol.Cursor, ledgerClosedAt, txHash, format string, +) (protocol.EventInfo, error) { + if cursor.Ledger > math.MaxInt32 { return protocol.EventInfo{}, fmt.Errorf("ledger sequence %d exceeds supported range", cursor.Ledger) } + typeName, ok := eventTypeName(head.typ) + if !ok { + return protocol.EventInfo{}, fmt.Errorf("unknown XDR ContractEventType type: %d", head.typ) + } + info := protocol.EventInfo{ - EventType: eventType, - Ledger: int32(ledger), + EventType: typeName, + Ledger: int32(cursor.Ledger), LedgerClosedAt: ledgerClosedAt, ID: cursor.String(), TransactionHash: txHash, @@ -280,53 +354,44 @@ func eventInfoForEvent( TxIndex: cursor.Tx, } + if head.cid != nil { + info.ContractID = strkey.MustEncode(strkey.VersionByteContract, head.cid) + } + + var err error switch format { case protocol.FormatJSON: - // json encode the topic - info.TopicJSON = make([]json.RawMessage, 0, protocol.MaxTopicCount) - for _, topic := range v0.Topics { - topic, err := xdr2json.ConvertInterface(topic) - if err != nil { - return protocol.EventInfo{}, err - } - info.TopicJSON = append(info.TopicJSON, topic) + if info.TopicJSON, err = jsonifySlice(xdr.ScVal{}, body.topics); err != nil { + return protocol.EventInfo{}, err } - - var convErr error - info.ValueJSON, convErr = xdr2json.ConvertInterface(v0.Data) - if convErr != nil { - return protocol.EventInfo{}, convErr + if info.ValueJSON, err = xdr2json.ConvertBytes(xdr.ScVal{}, body.data); err != nil { + return protocol.EventInfo{}, err } - default: - // base64-xdr encode the topic - topic := make([]string, 0, protocol.MaxTopicCount) - for _, segment := range v0.Topics { - seg, err := xdr.MarshalBase64(segment) - if err != nil { - return protocol.EventInfo{}, err - } - topic = append(topic, seg) + info.TopicXDR = make([]string, len(body.topics)) + for i, segment := range body.topics { + info.TopicXDR[i] = base64.StdEncoding.EncodeToString(segment) } - - // base64-xdr encode the data - data, err := xdr.MarshalBase64(v0.Data) - if err != nil { - return protocol.EventInfo{}, err - } - - info.TopicXDR = topic - info.ValueXDR = data + info.ValueXDR = base64.StdEncoding.EncodeToString(body.data) } - if event.Event.ContractId != nil { - info.ContractID = strkey.MustEncode( - strkey.VersionByteContract, - (*event.Event.ContractId)[:]) - } return info, nil } +// eventTypeName is protocol.GetEventTypeFromEventTypeXDR without the per-call map; ok is false outside the enum. +func eventTypeName(t xdr.ContractEventType) (string, bool) { + switch t { + case xdr.ContractEventTypeSystem: + return protocol.EventTypeSystem, true + case xdr.ContractEventTypeContract: + return protocol.EventTypeContract, true + case xdr.ContractEventTypeDiagnostic: + return protocol.EventTypeDiagnostic, true + default: + return "", false + } +} + // NewGetEventsHandler returns a json rpc handler to fetch and filter events func NewGetEventsHandler( logger *log.Entry, diff --git a/cmd/stellar-rpc/internal/methods/get_events_test.go b/cmd/stellar-rpc/internal/methods/get_events_test.go index 2286905fe..8b2f6196f 100644 --- a/cmd/stellar-rpc/internal/methods/get_events_test.go +++ b/cmd/stellar-rpc/internal/methods/get_events_test.go @@ -1279,6 +1279,24 @@ func setupTwoContractEventsHandler(t *testing.T) (eventsRPCHandler, xdr.Contract }, contractA, contractB } +// TestEventTypeName pins the three wire names and that a type outside the enum is refused, not rendered empty. +func TestEventTypeName(t *testing.T) { + for want, typ := range map[string]xdr.ContractEventType{ + protocol.EventTypeSystem: xdr.ContractEventTypeSystem, + protocol.EventTypeContract: xdr.ContractEventTypeContract, + protocol.EventTypeDiagnostic: xdr.ContractEventTypeDiagnostic, + } { + got, ok := eventTypeName(typ) + require.True(t, ok, want) + assert.Equal(t, want, got) + } + + // The views reject this today; the guard is for an enum that gains a member. + got, ok := eventTypeName(xdr.ContractEventType(99)) + assert.False(t, ok, "a type outside the enum must not render") + assert.Empty(t, got) +} + // Filters are OR-ed together: a filter with no contract IDs matches events // from any contract, so it must not be starved by another filter's contract // ID restriction. diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event.go index 06c8d8267..1e319c12b 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event.go @@ -3,6 +3,7 @@ package sqlitedb import ( "context" + "database/sql" "errors" "fmt" "io" @@ -291,7 +292,7 @@ func (eventHandler *eventHandler) trimEvents(latestLedgerSeq uint32, retentionWi // ascending Cursor order. // // If f returns false, the scan terminates early (f will not be applied on -// remaining events in the range). +// remaining events in the range). If f returns an error, the scan aborts with it. // //nolint:funlen,cyclop func (eventHandler *eventHandler) GetEvents( @@ -300,7 +301,7 @@ func (eventHandler *eventHandler) GetEvents( contractIDs [][]byte, topics store.TopicFilters, eventTypes []int, - scanner store.ScanFunction, + scanner store.ViewScanFunction, ) error { start := time.Now() @@ -370,18 +371,15 @@ func (eventHandler *eventHandler) GetEvents( defer rows.Close() - type rowResult struct { - eventCursorID string - eventData []byte - transactionHash []byte + var row struct { + eventCursorID string + // Use sql.RawBytes over []byte to avoid unnecessary copying + eventData sql.RawBytes + transactionHash sql.RawBytes ledgerCloseTime int64 } - foundRows := 0 for rows.Next() { - foundRows++ - - row := rowResult{} if err := rows.Scan( &row.eventCursorID, &row.eventData, @@ -398,15 +396,17 @@ func (eventHandler *eventHandler) GetEvents( return errors.Join(err, errors.New("failed to parse cursor")) } - var event xdr.DiagnosticEvent - if err := event.UnmarshalBinary(row.eventData); err != nil { - return errors.Join(err, errors.New("failed to decode event")) - } + eventView := xdr.DiagnosticEventView(row.eventData) txHash := xdr.Hash(transactionHash) - if !scanner(event, cur, ledgerCloseTime, &txHash) { + keepGoing, err := scanner(eventView, cur, ledgerCloseTime, &txHash) + if err != nil { + return errors.Join(err, errors.New("failed to scan event")) + } + if !keepGoing { return nil } + foundRows++ } eventHandler.log. diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event_test.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event_test.go index 20ced00ae..26ccaa3bc 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event_test.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/event_test.go @@ -276,9 +276,9 @@ func TestInsertEventsBatchingExceedsLimit(t *testing.T) { var count int err = eventReader.GetEvents(ctx, cursorRange, nil, nil, nil, - func(_ xdr.DiagnosticEvent, _ protocol.Cursor, _ int64, _ *xdr.Hash) bool { + func(_ xdr.DiagnosticEventView, _ protocol.Cursor, _ int64, _ *xdr.Hash) (bool, error) { count++ - return true + return true, nil }) require.NoError(t, err) diff --git a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/migration_test.go b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/migration_test.go index 27bce5678..b990ce7c9 100644 --- a/cmd/stellar-rpc/internal/rpcv1/sqlitedb/migration_test.go +++ b/cmd/stellar-rpc/internal/rpcv1/sqlitedb/migration_test.go @@ -82,9 +82,9 @@ func requireLedgerData(t *testing.T, testDB *DB, lcms []xdr.LedgerCloseMeta) { End: protocol.Cursor{Ledger: lcms[len(lcms)-1].LedgerSequence() + 1}, } require.NoError(t, eventReader.GetEvents(t.Context(), cursorRange, nil, nil, nil, - func(xdr.DiagnosticEvent, protocol.Cursor, int64, *xdr.Hash) bool { + func(xdr.DiagnosticEventView, protocol.Cursor, int64, *xdr.Hash) (bool, error) { eventCount++ - return true + return true, nil })) require.Equal(t, 4*len(lcms), eventCount, "expected 4 events per ledger") } diff --git a/cmd/stellar-rpc/internal/rpcv2/eventsapi/conversions_test.go b/cmd/stellar-rpc/internal/rpcv2/eventsapi/conversions_test.go index 240744738..317899c04 100644 --- a/cmd/stellar-rpc/internal/rpcv2/eventsapi/conversions_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/eventsapi/conversions_test.go @@ -466,15 +466,8 @@ func TestResponseScanStatus(t *testing.T) { // Ingest stores contract and system events only, so a diagnostic event // reaching the response means the store is corrupt. -func TestResponseEventType(t *testing.T) { - contract, err := responseEventType(xdr.ContractEventTypeContract) - require.NoError(t, err) - assert.Equal(t, protocol.EventTypeContract, contract) - - system, err := responseEventType(xdr.ContractEventTypeSystem) - require.NoError(t, err) - assert.Equal(t, protocol.EventTypeSystem, system) - - _, err = responseEventType(xdr.ContractEventTypeDiagnostic) - require.ErrorContains(t, err, "stored event has type") +func TestCheckResponseEventType(t *testing.T) { + require.NoError(t, checkResponseEventType(protocol.EventTypeContract)) + require.NoError(t, checkResponseEventType(protocol.EventTypeSystem)) + require.ErrorContains(t, checkResponseEventType(protocol.EventTypeDiagnostic), "stored event has type") } diff --git a/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v1.go b/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v1.go index a9e3f7fbb..5eb7c401e 100644 --- a/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v1.go +++ b/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v1.go @@ -307,25 +307,11 @@ func isMatchAll(f *event.Filter) bool { return true } -// eventInfoV1 mints the v1 response event: the same stored-payload decode as -// eventInfoV2, reshaped into the v1 wire type. +// eventInfoV1 mints the v1 response event: eventInfoV2's output in the v1 wire type. func eventInfoV1(p *event.Payload, format string) (protocol.EventInfo, error) { v2, err := eventInfoV2(p, format) if err != nil { return protocol.EventInfo{}, fmt.Errorf("could not parse event: %w", err) } - return protocol.EventInfo{ - EventType: v2.EventType, - Ledger: v2.Ledger, - LedgerClosedAt: v2.LedgerClosedAt, - ContractID: v2.ContractID, - ID: v2.ID, - OpIndex: v2.OpIndex, - TxIndex: v2.TxIndex, - TransactionHash: v2.TransactionHash, - TopicXDR: v2.TopicXDR, - TopicJSON: v2.TopicJSON, - ValueXDR: v2.ValueXDR, - ValueJSON: v2.ValueJSON, - }, nil + return protocol.EventInfo(v2), nil } diff --git a/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v2.go b/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v2.go index ac6c7b188..b22a33391 100644 --- a/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v2.go +++ b/cmd/stellar-rpc/internal/rpcv2/eventsapi/get_events_v2.go @@ -22,10 +22,10 @@ import ( supportlog "github.com/stellar/go-stellar-sdk/support/log" "github.com/stellar/go-stellar-sdk/xdr" + "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/methods" "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/event" - "github.com/stellar/stellar-rpc/cmd/stellar-rpc/internal/xdr2json" ) // Limits are the page caps an operator configures, which both events @@ -453,88 +453,26 @@ func responseScanStatus(s query.ScanStatus) string { } } -// eventInfoV2 builds one response event from a stored payload. The ID is -// the same TOID-and-index form v1 mints, and topics and value follow the -// request's xdrFormat. +// eventInfoV2 builds one response event from a stored event payload. func eventInfoV2(p *event.Payload, format string) (protocol.EventInfoV2, error) { - var ev xdr.ContractEvent - if err := ev.UnmarshalBinary(p.ContractEventBytes); err != nil { - return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: stored event bytes: %w", err) - } - if ev.Body.V != 0 || ev.Body.V0 == nil { - // Unreachable: UnmarshalBinary above rejects any other discriminant. - // Kept as a guard on the V0 dereferences that follow. - return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: stored event has body version %d", ev.Body.V) - } - - eventType, err := responseEventType(ev.Type) + cursor := protocol.Cursor{Ledger: p.LedgerSequence, Tx: p.TxIdx, Op: p.OpIdx, Event: p.EventIdx} + info, err := methods.EventInfoFromView(xdr.ContractEventView(p.ContractEventBytes), cursor, + time.Unix(p.LedgerClosedAt, 0).UTC().Format(time.RFC3339), p.TxHash.HexString(), format) if err != nil { - return protocol.EventInfoV2{}, err - } - // The response field is int32, so a sequence past its range would go - // out negative. v1 rejects the same shape. - if p.LedgerSequence > math.MaxInt32 { - return protocol.EventInfoV2{}, fmt.Errorf( - "rpcv2: ledger sequence %d exceeds supported range", p.LedgerSequence) - } - - info := protocol.EventInfoV2{ - EventType: eventType, - Ledger: int32(p.LedgerSequence), - LedgerClosedAt: time.Unix(p.LedgerClosedAt, 0).UTC().Format(time.RFC3339), - ID: protocol.Cursor{ - Ledger: p.LedgerSequence, Tx: p.TxIdx, Op: p.OpIdx, Event: p.EventIdx, - }.String(), - OpIndex: p.OpIdx, - TxIndex: p.TxIdx, - TransactionHash: p.TxHash.HexString(), - } - if ev.ContractId != nil { - info.ContractID = strkey.MustEncode(strkey.VersionByteContract, ev.ContractId[:]) - } - - if format == protocol.FormatJSON { - info.TopicJSON = make([]json.RawMessage, 0, len(ev.Body.V0.Topics)) - for i := range ev.Body.V0.Topics { - converted, err := xdr2json.ConvertInterface(ev.Body.V0.Topics[i]) - if err != nil { - return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: topic %d to json: %w", i, err) - } - info.TopicJSON = append(info.TopicJSON, converted) - } - valueJSON, err := xdr2json.ConvertInterface(ev.Body.V0.Data) - if err != nil { - return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: value to json: %w", err) - } - info.ValueJSON = valueJSON - return info, nil + return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: %w", err) } - - info.TopicXDR = make([]string, 0, len(ev.Body.V0.Topics)) - for i := range ev.Body.V0.Topics { - encoded, err := xdr.MarshalBase64(ev.Body.V0.Topics[i]) - if err != nil { - return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: topic %d to xdr: %w", i, err) - } - info.TopicXDR = append(info.TopicXDR, encoded) - } - valueXDR, err := xdr.MarshalBase64(ev.Body.V0.Data) - if err != nil { - return protocol.EventInfoV2{}, fmt.Errorf("rpcv2: value to xdr: %w", err) + // Stricter than v1, which also renders diagnostic events. + if err := checkResponseEventType(info.EventType); err != nil { + return protocol.EventInfoV2{}, err } - info.ValueXDR = valueXDR - return info, nil + return protocol.EventInfoV2(info), nil } -// responseEventType: ingest stores contract and system events only. -func responseEventType(t xdr.ContractEventType) (string, error) { - switch t { - case xdr.ContractEventTypeSystem: - return protocol.EventTypeSystem, nil - case xdr.ContractEventTypeContract: - return protocol.EventTypeContract, nil - default: - return "", fmt.Errorf("rpcv2: stored event has type %d;"+ - " this endpoint serves contract and system events only", t) +// checkResponseEventType: ingest stores contract and system events only, so only diagnostic is refused. +func checkResponseEventType(name string) error { + if name != protocol.EventTypeDiagnostic { + return nil } + return fmt.Errorf("rpcv2: stored event has type %q;"+ + " this endpoint serves contract and system events only", name) } diff --git a/cmd/stellar-rpc/internal/rpcv2/stores/event/extract_test.go b/cmd/stellar-rpc/internal/rpcv2/stores/event/extract_test.go index 4ad5ed8dd..b398c8d05 100644 --- a/cmd/stellar-rpc/internal/rpcv2/stores/event/extract_test.go +++ b/cmd/stellar-rpc/internal/rpcv2/stores/event/extract_test.go @@ -1,6 +1,7 @@ package event_test import ( + "bytes" "fmt" "path" "testing" @@ -598,16 +599,18 @@ func sqliteEventRows(t *testing.T, lcm xdr.LedgerCloseMeta) []sqliteEventRow { var rows []sqliteEventRow reader := sqlitedb.NewEventReader(logger, testDB, testPassphrase) err = reader.GetEvents(ctx, cursorRange, nil, nil, nil, - func(event xdr.DiagnosticEvent, cur protocol.Cursor, closeTime int64, txHash *xdr.Hash) bool { - evBytes, merr := event.Event.MarshalBinary() - require.NoError(t, merr) + func(eventView xdr.DiagnosticEventView, cur protocol.Cursor, closeTime int64, txHash *xdr.Hash) (bool, error) { + ev, verr := eventView.Event() + require.NoError(t, verr) + evBytes, rerr := ev.Raw() + require.NoError(t, rerr) rows = append(rows, sqliteEventRow{ cursor: cur, txHash: *txHash, ledgerCloseTime: closeTime, - contractEventBytes: evBytes, + contractEventBytes: bytes.Clone(evBytes), }) - return true + return true, nil }) require.NoError(t, err, "SQLite GetEvents") return rows diff --git a/cmd/stellar-rpc/internal/store/event.go b/cmd/stellar-rpc/internal/store/event.go index e43c80705..9feaa008a 100644 --- a/cmd/stellar-rpc/internal/store/event.go +++ b/cmd/stellar-rpc/internal/store/event.go @@ -24,12 +24,15 @@ type TopicFilter []TopicCondition // matches a row, the DB event is considered a candidate for further filtering. type TopicFilters []TopicFilter -type ScanFunction func( - event xdr.DiagnosticEvent, +// ViewScanFunction visits one event. eventView aliases the backend's row buffer and is valid +// only for the call, so copy anything that must persist. txHash points at a per-row copy and +// is free to keep. Returning false stops the scan and an error aborts it. +type ViewScanFunction func( + eventView xdr.DiagnosticEventView, cursor protocol.Cursor, ledgerCloseTimestamp int64, txHash *xdr.Hash, -) bool +) (bool, error) // EventReader has all the public methods to fetch events from the backend. type EventReader interface { @@ -39,7 +42,7 @@ type EventReader interface { contractIDs [][]byte, topics TopicFilters, eventTypes []int, - f ScanFunction, + f ViewScanFunction, ) error }