ingest: size each element once on the extract path, and return element views (+ stress-density benchmark)

CHANGELOG.md

@@ -15,6 +15,7 @@ Official project releases may be found here: https://github.com/stellar/go-stell
 ### New Features
 * xdr: Added `LedgerCloseMetaView.LedgerHeader()`, exposing the version-resolving header accessor that already backs `LedgerSequence`, `LedgerCloseTime`, `LedgerHash`, and `PreviousLedgerHash` ([#5982](https://github.com/stellar/go-stellar-sdk/pull/5982))
 * rpcclient: Add `Client.URL()` to expose the configured RPC server URL ([#5885](https://github.com/stellar/go-stellar-sdk/issues/5885))
+* protocols/rpc: Add getEvents v2 wire types and request validation (`GetEventsV2Request`/`GetEventsV2Response`, `EventFilterV2`, scan statuses, typed `error.data` payloads), transcribed from the [accepted proposal](https://github.com/orgs/stellar/discussions/1872) ([#5971](https://github.com/stellar/go-stellar-sdk/pull/5971))
 
 ### Bug Fixes
 * processors/token_transfer: Accept a `to_muxed_id` bound to `Void` in V4 event data. CAP-0067 specifies that the key is simply absent when there is no muxed destination, and that form already parsed. `Void` is what a contract emits instead if it publishes its event data as a `#[contracttype]` struct with an `Option` field — the natural way to write it before CAP-0086's sparse maps, which omit the key. Such an event previously failed to parse and was dropped from the event stream entirely, silently ([#5983](https://github.com/stellar/go-stellar-sdk/pull/5983))

ingest/extract.go

@@ -58,6 +58,11 @@ func ExtractLedgerTxParts(lcmView xdr.LedgerCloseMetaView) ([]LedgerTxParts, err
 		return nil, err
 	}
 	var out []LedgerTxParts
+	if d.txCount > 0 {
+		// Presize from the O(1) TxProcessing count; an empty ledger keeps
+		// returning the nil slice it always has.
+		out = make([]LedgerTxParts, 0, d.txCount)
+	}
 	for parts, iterErr := range d.TxProcessing() {
 		if iterErr != nil {
 			return nil, fmt.Errorf("ingest: TxProcessing iter: %w", iterErr)
@@ -77,23 +82,31 @@ func ExtractLedgerTxParts(lcmView xdr.LedgerCloseMetaView) ([]LedgerTxParts, err
 	return out, nil
 }
 
-// TxEvents is one transaction's contract events in the flat raw-bytes shape
-// an events indexer consumes, index-aligned with the LedgerTxParts slice it
-// was derived from (the transaction hash lives on the parts element, not
-// here). Every byte slice ALIASES the source LedgerCloseMetaView buffer
-// (zero-copy); callers copy what they retain.
+// TxEvents is one transaction's contract events as trimmed element VIEWS,
+// index-aligned with the LedgerTxParts slice it was derived from (the
+// transaction hash lives on the parts element, not here). Every view ALIASES
+// the source LedgerCloseMetaView buffer (zero-copy); callers copy what they
+// retain.
 //
-//   - TransactionEvents holds the V4 top-level transaction events, each a raw
-//     xdr.TransactionEvent. Read Stage / the inner event zero-copy by wrapping
-//     an element: xdr.TransactionEventView(raw).Stage() / .Event().
-//   - OperationEvents holds, per operation, the raw xdr.ContractEvent bytes.
+//   - TransactionEvents holds the V4 top-level transaction events. Read Stage
+//     or the inner event directly — elem.Stage() / elem.Event() — with no
+//     wrapping step.
+//   - OperationEvents holds, per operation, that operation's contract events.
 //     For V3 SorobanMeta there is a single operation group (the soroban tx has
 //     one op); for V4 there is one group per operation.
 //
+// The element types are views rather than []byte because the views ARE their
+// wire bytes (each is ~[]byte, trimmed to the element's extent), so a consumer
+// that wants bytes writes []byte(elem) for free, while a consumer that wants
+// to read a field is spared re-wrapping. Returning [][]byte instead would cost
+// an extra n-element slice per array — the generated MustAll() hands back
+// []E, and Go cannot retype the outer slice — for no gain at either kind of
+// call site.
+//
 // V0/V1/V2 meta carry no contract events, so both fields are empty.
 type TxEvents struct {
-	TransactionEvents [][]byte   // raw xdr.TransactionEvent (V4 top-level)
-	OperationEvents   [][][]byte // raw xdr.ContractEvent, per operation
+	TransactionEvents []xdr.TransactionEventView // V4 top-level
+	OperationEvents   [][]xdr.ContractEventView  // per operation
 }
 
 // EventsFromTxParts returns the contract events of every transaction, one

ingest/extract_real_ledger_test.go

@@ -102,15 +102,15 @@ func TestEventsFromTxParts_RealLedgerEquivalence(t *testing.T) {
 		for j, refEv := range refEvents.TransactionEvents {
 			refRaw, err := refEv.MarshalBinary()
 			require.NoError(t, err)
-			assert.Equal(t, refRaw, got[i].TransactionEvents[j], "tx %d tx-event %d", i, j)
+			assert.Equal(t, refRaw, []byte(got[i].TransactionEvents[j]), "tx %d tx-event %d", i, j)
 		}
 		require.Len(t, got[i].OperationEvents, len(refEvents.OperationEvents), "tx %d op groups", i)
 		for op, refOpEvents := range refEvents.OperationEvents {
 			require.Len(t, got[i].OperationEvents[op], len(refOpEvents), "tx %d op %d events", i, op)
 			for j, refEv := range refOpEvents {
 				refRaw, err := refEv.MarshalBinary()
 				require.NoError(t, err)
-				assert.Equal(t, refRaw, got[i].OperationEvents[op][j], "tx %d op %d event %d", i, op, j)
+				assert.Equal(t, refRaw, []byte(got[i].OperationEvents[op][j]), "tx %d op %d event %d", i, op, j)
 			}
 		}
 	}
@@ -148,7 +148,7 @@ func TestLedgerTransactionViewRange_RealLedgerEquivalence(t *testing.T) {
 		for j, refEv := range refDiag {
 			refRaw, err := refEv.MarshalBinary()
 			require.NoError(t, err)
-			assert.Equal(t, refRaw, v.DiagnosticEvents[j], "tx %d diagnostic %d", i, j)
+			assert.Equal(t, refRaw, []byte(v.DiagnosticEvents[j]), "tx %d diagnostic %d", i, j)
 		}
 	}
 }

ingest/extract_stress_bench_test.go

@@ -0,0 +1,205 @@
+package ingest_test
+
+import (
+	"testing"
+
+	"github.com/stellar/go-stellar-sdk/ingest"
+	"github.com/stellar/go-stellar-sdk/xdr"
+)
+
+// BenchmarkLedgerTxPartsStressDensity measures the parts/products composition
+// on a crafted ledger far denser than the pubnet fixture: 6,000 transactions
+// (stellar-rpc's hot-ingest stress shape), ~10 contract events each, cycling
+// V3-soroban / V4-soroban / V4-classic metas so every extractor arm runs, in
+// an LCM V2 (TransactionResultMetaV1 TxProcessing — the modern element shape).
+// Each operation also carries a couple of ledger-entry changes and each tx a
+// couple of diagnostic events, so the walk pays a realistic spine, not just
+// event bytes. This is the density where per-element size passes dominate; a
+// TRUE production-shape measurement (real stress ledgers) belongs to the RPC
+// repo and its fixtures.
+func BenchmarkLedgerTxPartsStressDensity(b *testing.B) {
+	const txCount = 6000
+	raw := stressDensityLCM(b, txCount)
+	view := xdr.LedgerCloseMetaView(raw)
+
+	b.Run("parts", func(b *testing.B) {
+		b.ReportAllocs()
+		for i := 0; i < b.N; i++ {
+			txParts, err := ingest.ExtractLedgerTxParts(view)
+			if err != nil {
+				b.Fatal(err)
+			}
+			if len(txParts) != txCount {
+				b.Fatal("bad fixture")
+			}
+		}
+	})
+	b.Run("parts_events", func(b *testing.B) {
+		b.ReportAllocs()
+		for i := 0; i < b.N; i++ {
+			txParts, err := ingest.ExtractLedgerTxParts(view)
+			if err != nil {
+				b.Fatal(err)
+			}
+			txEvents, err := ingest.EventsFromTxParts(txParts)
+			if err != nil {
+				b.Fatal(err)
+			}
+			if len(txEvents) != txCount {
+				b.Fatal("bad fixture")
+			}
+		}
+	})
+	b.Run("all_products", func(b *testing.B) {
+		b.ReportAllocs()
+		for i := 0; i < b.N; i++ {
+			txParts, err := ingest.ExtractLedgerTxParts(view)
+			if err != nil {
+				b.Fatal(err)
+			}
+			txEvents, err := ingest.EventsFromTxParts(txParts)
+			if err != nil {
+				b.Fatal(err)
+			}
+			fees, err := ingest.FeesFromTxParts(txParts)
+			if err != nil {
+				b.Fatal(err)
+			}
+			if len(txEvents) != txCount || len(fees.ClassicFeesPerOp) == 0 || len(fees.SorobanInclusionFees) == 0 {
+				b.Fatal("bad fixture")
+			}
+		}
+	})
+}
+
+// stressDensityLCM builds the stress ledger: txCount transactions cycling
+// three flavors —
+//
+//	i%3 == 0: V3 soroban  (SorobanMeta: fee ext, 10 events, 2 diagnostics)
+//	i%3 == 1: V4 soroban  (1 op: 2 changes + 10 events; 1 tx-level event,
+//	          SorobanMeta with fee ext, 2 diagnostics)
+//	i%3 == 2: V4 classic  (2 ops: 2 changes + 5 unified events each)
+//
+// with per-tx results carrying per-operation results (so the classic fee arm
+// classifies) and distinct hashes.
+func stressDensityLCM(b *testing.B, txCount int) []byte {
+	b.Helper()
+
+	ev := func(s string) xdr.ContractEvent {
+		sym := xdr.ScSymbol(s)
+		amount := xdr.Int64(1_000_000)
+		return xdr.ContractEvent{
+			Type: xdr.ContractEventTypeContract,
+			Body: xdr.ContractEventBody{V: 0, V0: &xdr.ContractEventV0{
+				Topics: []xdr.ScVal{{Type: xdr.ScValTypeScvSymbol, Sym: &sym}},
+				Data:   xdr.ScVal{Type: xdr.ScValTypeScvI64, I64: &amount},
+			}},
+		}
+	}
+	events := func(n int) []xdr.ContractEvent {
+		out := make([]xdr.ContractEvent, n)
+		for i := range out {
+			out[i] = ev("stress-topic-payload")
+		}
+		return out
+	}
+	diags := []xdr.DiagnosticEvent{
+		{InSuccessfulContractCall: true, Event: ev("diag-a")},
+		{InSuccessfulContractCall: true, Event: ev("diag-b")},
+	}
+	changes := func(seed byte) xdr.LedgerEntryChanges {
+		key := func(k byte) xdr.LedgerKey {
+			var ed xdr.Uint256
+			ed[0], ed[31] = seed, k
+			return xdr.LedgerKey{
+				Type: xdr.LedgerEntryTypeAccount,
+				Account: &xdr.LedgerKeyAccount{AccountId: xdr.AccountId{
+					Type: xdr.PublicKeyTypePublicKeyTypeEd25519, Ed25519: &ed,
+				}},
+			}
+		}
+		removed0, removed1 := key(0), key(1)
+		return xdr.LedgerEntryChanges{
+			{Type: xdr.LedgerEntryChangeTypeLedgerEntryRemoved, Removed: &removed0},
+			{Type: xdr.LedgerEntryChangeTypeLedgerEntryRemoved, Removed: &removed1},
+		}
+	}
+	feeExt := xdr.SorobanTransactionMetaExt{V: 1, V1: &xdr.SorobanTransactionMetaExtV1{
+		TotalNonRefundableResourceFeeCharged: 40_000,
+		TotalRefundableResourceFeeCharged:    10_000,
+	}}
+	rv := xdr.ScVal{Type: xdr.ScValTypeScvVoid}
+
+	metaV3 := xdr.TransactionMeta{V: 3, V3: &xdr.TransactionMetaV3{
+		TxChangesBefore: changes(3),
+		SorobanMeta: &xdr.SorobanTransactionMeta{
+			Ext: feeExt, Events: events(10), ReturnValue: rv, DiagnosticEvents: diags,
+		},
+	}}
+	metaV4Soroban := xdr.TransactionMeta{V: 4, V4: &xdr.TransactionMetaV4{
+		Operations:       []xdr.OperationMetaV2{{Changes: changes(4), Events: events(10)}},
+		SorobanMeta:      &xdr.SorobanTransactionMetaV2{Ext: feeExt, ReturnValue: &rv},
+		Events:           []xdr.TransactionEvent{{Stage: xdr.TransactionEventStageTransactionEventStageAfterTx, Event: ev("fee")}},
+		DiagnosticEvents: diags,
+	}}
+	metaV4Classic := xdr.TransactionMeta{V: 4, V4: &xdr.TransactionMetaV4{
+		Operations: []xdr.OperationMetaV2{
+			{Changes: changes(5), Events: events(5)},
+			{Changes: changes(6), Events: events(5)},
+		},
+	}}
+
+	opResults := func(n int) *[]xdr.OperationResult {
+		out := make([]xdr.OperationResult, n)
+		for i := range out {
+			out[i] = xdr.OperationResult{
+				Code: xdr.OperationResultCodeOpInner,
+				Tr: &xdr.OperationResultTr{
+					Type:          xdr.OperationTypePayment,
+					PaymentResult: &xdr.PaymentResult{Code: xdr.PaymentResultCodePaymentSuccess},
+				},
+			}
+		}
+		return &out
+	}
+
+	proc := make([]xdr.TransactionResultMetaV1, txCount)
+	for i := range proc {
+		var meta xdr.TransactionMeta
+		nOps := 1
+		switch i % 3 {
+		case 0:
+			meta = metaV3
+		case 1:
+			meta = metaV4Soroban
+		case 2:
+			meta = metaV4Classic
+			nOps = 2
+		}
+		var hash xdr.Hash
+		hash[0], hash[1], hash[2] = byte(i), byte(i>>8), byte(i>>16)
+		proc[i] = xdr.TransactionResultMetaV1{
+			Result: xdr.TransactionResultPair{
+				TransactionHash: hash,
+				Result: xdr.TransactionResult{
+					FeeCharged: 150_000,
+					Result: xdr.TransactionResultResult{
+						Code:    xdr.TransactionResultCodeTxSuccess,
+						Results: opResults(nOps),
+					},
+				},
+			},
+			TxApplyProcessing: meta,
+		}
+	}
+
+	lcm := xdr.LedgerCloseMeta{V: 2, V2: &xdr.LedgerCloseMetaV2{
+		TxSet:        xdr.GeneralizedTransactionSet{V: 1, V1TxSet: &xdr.TransactionSetV1{}},
+		TxProcessing: proc,
+	}}
+	raw, err := lcm.MarshalBinary()
+	if err != nil {
+		b.Fatal(err)
+	}
+	return raw
+}

ingest/ledger_close_meta_view_nav.go

@@ -130,6 +130,11 @@ type lcmViewDispatch struct {
 	lcm  xdr.LedgerCloseMetaView
 	tp   iter.Seq2[txResultParts, error]
 	envs iter.Seq2[xdr.TransactionEnvelopeView, error]
+	// txCount is the TxProcessing element count, read from the array's length
+	// prefix at dispatch time (an O(1) read) so consumers can presize their
+	// per-transaction output. Best-effort: on a malformed prefix it stays 0 and
+	// the walk itself reports the error — dispatch does not fail for it.
+	txCount int
 }
 
 // dispatchLCMView opens lcm, reads its discriminator, and returns the
@@ -159,6 +164,7 @@ func dispatchLCMView(lcm xdr.LedgerCloseMetaView) (lcmViewDispatch, error) {
 		if err != nil {
 			return lcmViewDispatch{}, fmt.Errorf("ingest: V0 TxProcessing: %w", err)
 		}
+		d.txCount, _ = raw.Count() //nolint:errcheck // best-effort presize; the walk reports the error
 		d.tp = txProcessingPartsMeta(raw)
 		d.envs = v0TxSetEnvelopes(v0.TxSet)
 	case 1:
@@ -170,6 +176,7 @@ func dispatchLCMView(lcm xdr.LedgerCloseMetaView) (lcmViewDispatch, error) {
 		if err != nil {
 			return lcmViewDispatch{}, fmt.Errorf("ingest: V1 TxProcessing: %w", err)
 		}
+		d.txCount, _ = raw.Count() //nolint:errcheck // best-effort presize; the walk reports the error
 		d.tp = txProcessingPartsMeta(raw)
 		d.envs = generalizedEnvelopes("V1", v1.TxSet)
 	case 2:
@@ -181,6 +188,7 @@ func dispatchLCMView(lcm xdr.LedgerCloseMetaView) (lcmViewDispatch, error) {
 		if err != nil {
 			return lcmViewDispatch{}, fmt.Errorf("ingest: V2 TxProcessing: %w", err)
 		}
+		d.txCount, _ = raw.Count() //nolint:errcheck // best-effort presize; the walk reports the error
 		d.tp = txProcessingPartsMetaV1(raw)
 		d.envs = generalizedEnvelopes("V2", v2.TxSet)
 	default: