xdr: comprehensive XDR decoding optimization

.gitignore

@@ -19,6 +19,7 @@ debug
 *.index
 *.xdr
 *.db
+xdr/testdata/benchmark-ledgers.xdr.zst
 *.conf
 *.lock
 .proto_checksums

benchmarks/xdr_test.go

@@ -37,14 +37,10 @@ var gxdrInput = func() gxdr.TransactionEnvelope {
 }()
 
 func BenchmarkXDRUnmarshalWithReflection(b *testing.B) {
-	var (
-		r  bytes.Reader
-		te xdr.TransactionEnvelope
-	)
+	var te xdr.TransactionEnvelope
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
-		r.Reset(input)
-		_, _ = xdr3.Unmarshal(&r, &te)
+		_, _ = xdr3.Unmarshal(input, &te)
 	}
 }
 

go.mod

@@ -2,6 +2,8 @@ module github.com/stellar/go-stellar-sdk
 
 go 1.24.0
 
+toolchain go1.24.9
+
 require (
 	cloud.google.com/go/storage v1.42.0
 	github.com/BurntSushi/toml v1.3.2
@@ -35,7 +37,7 @@ require (
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.17.0
-	github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2
+	github.com/stellar/go-xdr v0.0.0-20260106211653-82e681bfd3f3
 	github.com/stretchr/testify v1.10.0
 	github.com/tyler-smith/go-bip39 v0.0.0-20180618194314-52158e4697b8
 	github.com/xdrpp/goxdr v0.1.1

go.sum

@@ -453,8 +453,8 @@ github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.17.0 h1:I5txKw7MJasPL/BrfkbA0Jyo/oELqVmux4pR/UxOMfI=
 github.com/spf13/viper v1.17.0/go.mod h1:BmMMMLQXSbcHK6KAOiFLz0l5JHrU89OdIRHvsk0+yVI=
-github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2 h1:OzCVd0SV5qE3ZcDeSFCmOWLZfEWZ3Oe8KtmSOYKEVWE=
-github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2/go.mod h1:yoxyU/M8nl9LKeWIoBrbDPQ7Cy+4jxRcWcOayZ4BMps=
+github.com/stellar/go-xdr v0.0.0-20260106211653-82e681bfd3f3 h1:uPJuFJUZl57C0I9/3ZPKXl5q6G8SskeVt9H5mJkaXXg=
+github.com/stellar/go-xdr v0.0.0-20260106211653-82e681bfd3f3/go.mod h1:ZSIhPj0Ya41YoY0K4msNqwEQyJW/kCzjNAcZJH+uQO0=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=

historyarchive/archive_test.go

@@ -517,7 +517,7 @@ func TestXdrDecode(t *testing.T) {
 	assert.Equal(t, len(xdrbytes), 152)
 
 	var tmp xdr.BucketEntry
-	n, err := xdr.Unmarshal(bytes.NewReader(xdrbytes[:]), &tmp)
+	n, err := xdr.Unmarshal(xdrbytes[:], &tmp)
 	fmt.Printf("Decoded %d bytes\n", n)
 	if err != nil {
 		panic(err)

ingest/change_compactor_test.go

@@ -428,9 +428,9 @@ func (s *TestChangeCompactorExistingRestoredSuite) SetupTest() {
 					Type:       xdr.ScAddressTypeScAddressTypeContract,
 					ContractId: &xdr.ContractId{0xca, 0xfe},
 				},
-				Key:        xdr.ScVal{Type: xdr.ScValTypeScvBool, B: &val},
+				Key:        xdr.ScVal{Type: xdr.ScValTypeScvBool, B: val},
 				Durability: xdr.ContractDataDurabilityPersistent,
-				Val:        xdr.ScVal{Type: xdr.ScValTypeScvBool, B: &val},
+				Val:        xdr.ScVal{Type: xdr.ScValTypeScvBool, B: val},
 			},
 		},
 	}

ingest/change_test.go

@@ -216,15 +216,14 @@ func TestSortChanges(t *testing.T) {
 }
 
 func createContractDataEntry() *xdr.ContractDataEntry {
-	scVal := true
 	return &xdr.ContractDataEntry{
 		Contract: xdr.ScAddress{
 			Type:       xdr.ScAddressTypeScAddressTypeContract,
 			ContractId: &xdr.ContractId{0xca},
 		},
 		Key: xdr.ScVal{
 			Type: xdr.ScValTypeScvBool,
-			B:    &scVal,
+			B:    true,
 		},
 	}
 }

ingest/checkpoint_change_reader_test.go

@@ -79,10 +79,9 @@ func (s *CheckpointChangeReaderTestSuite) TearDownTest() {
 // TestSimple test reading buckets with a single live entry.
 func (s *CheckpointChangeReaderTestSuite) TestSimple() {
 	meta := metaEntry(23)
-	liveType := xdr.BucketListTypeLive
 	meta.MetaEntry.Ext = xdr.BucketMetadataExt{
 		V:              1,
-		BucketListType: &liveType,
+		BucketListType: xdr.BucketListTypeLive,
 	}
 	curr1 := createXdrStream(
 		meta,
@@ -117,10 +116,9 @@ func (s *CheckpointChangeReaderTestSuite) TestSimple() {
 
 func (s *CheckpointChangeReaderTestSuite) TestReadAfterClose() {
 	meta := metaEntry(23)
-	liveType := xdr.BucketListTypeLive
 	meta.MetaEntry.Ext = xdr.BucketMetadataExt{
 		V:              1,
-		BucketListType: &liveType,
+		BucketListType: xdr.BucketListTypeLive,
 	}
 	curr1 := createXdrStream(
 		meta,
@@ -169,10 +167,9 @@ func (s *CheckpointChangeReaderTestSuite) TestReadAfterClose() {
 
 func (s *CheckpointChangeReaderTestSuite) TestContextCanceled() {
 	meta := metaEntry(23)
-	liveType := xdr.BucketListTypeLive
 	meta.MetaEntry.Ext = xdr.BucketMetadataExt{
 		V:              1,
-		BucketListType: &liveType,
+		BucketListType: xdr.BucketListTypeLive,
 	}
 	curr1 := createXdrStream(
 		meta,
@@ -513,10 +510,9 @@ func (s *CheckpointChangeReaderTestSuite) TestMalformedProtocol11BucketNoMeta()
 // TestMalformedBucketListType ensures the checkpoint change reader asserts its reading from the live bucketlist
 func (s *CheckpointChangeReaderTestSuite) TestMalformedBucketListType() {
 	meta := metaEntry(23)
-	hotArchiveType := xdr.BucketListTypeHotArchive
 	meta.MetaEntry.Ext = xdr.BucketMetadataExt{
 		V:              1,
-		BucketListType: &hotArchiveType,
+		BucketListType: xdr.BucketListTypeHotArchive,
 	}
 	curr1 := createXdrStream(
 		meta,

ingest/hot_archive_iterator_test.go

@@ -109,14 +109,13 @@ var hasWithHotArchiveExample = `{
 }`
 
 func hotArchiveMetaEntry(version uint32) xdr.HotArchiveBucketEntry {
-	listType := xdr.BucketListTypeHotArchive
 	return xdr.HotArchiveBucketEntry{
 		Type: xdr.HotArchiveBucketEntryTypeHotArchiveMetaentry,
 		MetaEntry: &xdr.BucketMetadata{
 			LedgerVersion: xdr.Uint32(version),
 			Ext: xdr.BucketMetadataExt{
 				V:              1,
-				BucketListType: &listType,
+				BucketListType: xdr.BucketListTypeHotArchive,
 			},
 		},
 	}
@@ -383,15 +382,14 @@ func (h *HotArchiveIteratorTestSuite) TestMissingBucketListType() {
 }
 
 func (h *HotArchiveIteratorTestSuite) TestInvalidBucketListType() {
-	listType := xdr.BucketListTypeLive
 	curr1 := createXdrStream(
 		xdr.HotArchiveBucketEntry{
 			Type: xdr.HotArchiveBucketEntryTypeHotArchiveMetaentry,
 			MetaEntry: &xdr.BucketMetadata{
 				LedgerVersion: xdr.Uint32(24),
 				Ext: xdr.BucketMetadataExt{
 					V:              1,
-					BucketListType: &listType,
+					BucketListType: xdr.BucketListTypeLive,
 				},
 			},
 		},

ingest/ledgerbackend/buffered_meta_pipe_reader.go

@@ -1,7 +1,7 @@
 package ledgerbackend
 
 import (
-	"bufio"
+	"bytes"
 	"io"
 	"time"
 
@@ -54,29 +54,28 @@ type metaResult struct {
 //     while previous ledger are being processed.
 //   - Limits memory usage in case of large ledgers are closed by the network.
 //
-// Internally, it keeps two buffers: bufio.Reader with binary ledger data and
-// buffered channel with unmarshaled xdr.LedgerCloseMeta objects ready for
-// processing. The first buffer removes overhead time connected to reading from
-// a file. The second buffer allows unmarshaling binary data into XDR objects
-// (which can be a bottleneck) while clients are processing previous ledgers.
+// Internally, it reads framed XDR data directly into a reusable buffer and uses
+// a reusable Decoder for optimized decoding. The buffered channel stores unmarshaled
+// xdr.LedgerCloseMeta objects ready for processing, allowing unmarshaling to
+// proceed while clients process previous ledgers.
 //
 // Finally, when a large ledger (larger than binary buffer) is closed it waits
 // until xdr.LedgerCloseMeta objects channel is empty. This prevents memory
 // exhaustion when network closes a series a large ledgers.
 type bufferedLedgerMetaReader struct {
-	r       *bufio.Reader
-	c       chan metaResult
-	decoder *xdr3.Decoder
+	r           io.Reader
+	c           chan metaResult
+	decoder     *xdr3.Decoder
+	frameBuffer bytes.Buffer
 }
 
 // newBufferedLedgerMetaReader creates a new meta reader that will shutdown
 // when stellar-core terminates.
 func newBufferedLedgerMetaReader(reader io.Reader) *bufferedLedgerMetaReader {
-	r := bufio.NewReaderSize(reader, metaPipeBufferSize)
 	return &bufferedLedgerMetaReader{
 		c:       make(chan metaResult, ledgerReadAheadBufferSize),
-		r:       r,
-		decoder: xdr3.NewDecoder(r),
+		r:       reader,
+		decoder: xdr3.NewDecoder(nil),
 	}
 }
 
@@ -86,21 +85,41 @@ func newBufferedLedgerMetaReader(reader io.Reader) *bufferedLedgerMetaReader {
 //   - The next ledger available in the buffer exceeds the meta pipe buffer size.
 //     In such case the method will block until LedgerCloseMeta buffer is empty.
 func (b *bufferedLedgerMetaReader) readLedgerMetaFromPipe() (*xdr.LedgerCloseMeta, error) {
-	frameLength, err := xdr.ReadFrameLength(b.decoder)
+	frameLength, err := xdr.ReadFrameLength(b.r)
 	if err != nil {
-		return nil, errors.Wrap(err, "error reading frame length")
+		if err == io.EOF {
+			return nil, err
+		}
+		return nil, errors.Wrap(err, "reading frame length")
 	}
 
 	for frameLength > metaPipeBufferSize && len(b.c) > 0 {
 		// Wait for LedgerCloseMeta buffer to be cleared to minimize memory usage.
 		<-time.After(time.Second)
 	}
 
+	// Read frame data directly into reusable buffer
+	b.frameBuffer.Reset()
+	b.frameBuffer.Grow(int(frameLength))
+	n, err := b.frameBuffer.ReadFrom(io.LimitReader(b.r, int64(frameLength)))
+	if err != nil {
+		return nil, errors.Wrap(err, "reading frame data")
+	}
+	if n != int64(frameLength) {
+		return nil, errors.Errorf("read %d bytes, expected %d", n, frameLength)
+	}
+
+	// Decode using reusable Decoder for optimized performance
 	var xlcm xdr.LedgerCloseMeta
-	_, err = xlcm.DecodeFrom(b.decoder, xdr3.DecodeDefaultMaxDepth)
+	b.decoder.Reset(b.frameBuffer.Bytes())
+	bytesRead, err := b.decoder.Decode(&xlcm)
 	if err != nil {
 		return nil, errors.Wrap(err, "unmarshaling framed LedgerCloseMeta")
 	}
+	if bytesRead != int(frameLength) {
+		return nil, errors.Errorf("unmarshaled %d bytes, expected %d", bytesRead, frameLength)
+	}
+
 	return &xlcm, nil
 }
 

ingest/sac/contract_data.go

@@ -340,7 +340,7 @@ func metadataObjFromAsset(isNative bool, code, issuer string) (*xdr.ScMap, error
 			},
 			Val: xdr.ScVal{
 				Type: xdr.ScValTypeScvU32,
-				U32:  &decimalVal,
+				U32:  decimalVal,
 			},
 		},
 		xdr.ScMapEntry{
@@ -576,7 +576,7 @@ func BalanceInt128ToContractData(assetContractId, holderID [32]byte, amt xdr.Int
 			},
 			Val: xdr.ScVal{
 				Type: xdr.ScValTypeScvBool,
-				B:    &trueIc,
+				B:    trueIc,
 			},
 		},
 		xdr.ScMapEntry{
@@ -586,7 +586,7 @@ func BalanceInt128ToContractData(assetContractId, holderID [32]byte, amt xdr.Int
 			},
 			Val: xdr.ScVal{
 				Type: xdr.ScValTypeScvBool,
-				B:    &trueIc,
+				B:    trueIc,
 			},
 		},
 	}

processors/token_transfer/contract_events_test.go

@@ -4,9 +4,10 @@ import (
 	"fmt"
 	"testing"
 
+	"google.golang.org/protobuf/proto"
+
 	"github.com/stellar/go-stellar-sdk/keypair"
 	"github.com/stellar/go-stellar-sdk/strkey"
-	"google.golang.org/protobuf/proto"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -136,7 +137,7 @@ func createScMap(keyValuePairs ...interface{}) xdr.ScVal {
 			val := xdr.Uint64(v)
 			valueScVal = xdr.ScVal{
 				Type: xdr.ScValTypeScvU64,
-				U64:  &val,
+				U64:  val,
 			}
 		default:
 			panic(fmt.Sprintf("unsupported value type: %T", value))
@@ -1715,7 +1716,7 @@ func TestValidContractEventsV4(t *testing.T) {
 					val := xdr.Uint64(id)
 					muxedIdVal = xdr.ScVal{
 						Type: xdr.ScValTypeScvU64,
-						U64:  &val,
+						U64:  val,
 					}
 				case "text":
 					text := tc.memoValue.(string)
@@ -2009,7 +2010,7 @@ func TestV4InvalidEvents(t *testing.T) {
 				mapEntries := xdr.ScMap{
 					{
 						Key: createSymbol("to_muxed_id"),
-						Val: xdr.ScVal{Type: xdr.ScValTypeScvU64, U64: &[]xdr.Uint64{12345}[0]},
+						Val: xdr.ScVal{Type: xdr.ScValTypeScvU64, U64: xdr.Uint64(12345)},
 					},
 				}
 

processors/token_transfer/muxed_info.go

@@ -13,7 +13,7 @@ func NewMuxedInfoFromMemo(m xdr.Memo) *MuxedInfo {
 	case xdr.MemoTypeMemoNone:
 		return nil
 	case xdr.MemoTypeMemoId:
-		id := uint64(*m.Id)
+		id := uint64(m.Id)
 		return NewMuxedInfoFromId(id)
 	case xdr.MemoTypeMemoText:
 		protoMemo.Content = &MuxedInfo_Text{

processors/token_transfer/token_transfer_processor.go

@@ -535,12 +535,13 @@ func (p *EventsProcessor) accountCreateEvents(tx ingest.LedgerTransaction, opInd
 func (p *EventsProcessor) mergeAccountEvents(tx ingest.LedgerTransaction, opIndex uint32, op xdr.Operation, result xdr.OperationResult) ([]*TokenTransferEvent, error) {
 	res := result.Tr.MustAccountMergeResult()
 	// If there is no transfer of XLM from source account to destination (i.e. src account is empty), then no need to generate a transfer event
-	if res.SourceAccountBalance == nil {
+	sourceBalance, ok := res.GetSourceAccountBalance()
+	if !ok {
 		return nil, nil
 	}
 	opSrcAcc := operationSourceAccount(tx, op)
 	destAcc := op.Body.MustDestination()
-	amt := amount.String64Raw(*res.SourceAccountBalance)
+	amt := amount.String64Raw(sourceBalance)
 	event, err := p.mintOrBurnOrTransferEvent(tx, &opIndex, xlmAsset, opSrcAcc.Address(), destAcc.Address(), amt, true)
 	if err != nil {
 		return nil, err