xdr and horizon: Optimize compress-marshaling of ledger keys (#4071)

benchmarks/xdr_test.go

@@ -132,3 +132,57 @@ func BenchmarkXDRUnsafeMarshalBase64WithEncodingBuffer(b *testing.B) {
 		_, _ = e.UnsafeMarshalBase64(xdrInput)
 	}
 }
+
+var ledgerKeys = []xdr.LedgerKey{
+	{
+		Type: xdr.LedgerEntryTypeAccount,
+		Account: &xdr.LedgerKeyAccount{
+			AccountId: xdr.MustAddress("GAOQJGUAB7NI7K7I62ORBXMN3J4SSWQUQ7FOEPSDJ322W2HMCNWPHXFB"),
+		},
+	},
+	{
+		Type: xdr.LedgerEntryTypeTrustline,
+		TrustLine: &xdr.LedgerKeyTrustLine{
+			AccountId: xdr.MustAddress("GAOQJGUAB7NI7K7I62ORBXMN3J4SSWQUQ7FOEPSDJ322W2HMCNWPHXFB"),
+			Asset:     xdr.MustNewCreditAsset("EUR", "GAOQJGUAB7NI7K7I62ORBXMN3J4SSWQUQ7FOEPSDJ322W2HMCNWPHXFB").ToTrustLineAsset(),
+		},
+	},
+	{
+		Type: xdr.LedgerEntryTypeOffer,
+		Offer: &xdr.LedgerKeyOffer{
+			SellerId: xdr.MustAddress("GAOQJGUAB7NI7K7I62ORBXMN3J4SSWQUQ7FOEPSDJ322W2HMCNWPHXFB"),
+			OfferId:  xdr.Int64(3),
+		},
+	},
+	{
+		Type: xdr.LedgerEntryTypeData,
+		Data: &xdr.LedgerKeyData{
+			AccountId: xdr.MustAddress("GAOQJGUAB7NI7K7I62ORBXMN3J4SSWQUQ7FOEPSDJ322W2HMCNWPHXFB"),
+			DataName:  "foobar",
+		},
+	},
+	{
+		Type: xdr.LedgerEntryTypeClaimableBalance,
+		ClaimableBalance: &xdr.LedgerKeyClaimableBalance{
+			BalanceId: xdr.ClaimableBalanceId{
+				Type: 0,
+				V0:   &xdr.Hash{0xca, 0xfe, 0xba, 0xbe},
+			},
+		},
+	},
+	{
+		Type: xdr.LedgerEntryTypeLiquidityPool,
+		LiquidityPool: &xdr.LedgerKeyLiquidityPool{
+			LiquidityPoolId: xdr.PoolId{0xca, 0xfe, 0xba, 0xbe},
+		},
+	},
+}
+
+func BenchmarkXDRMarshalCompress(b *testing.B) {
+	e := xdr.NewEncodingBuffer()
+	for i := 0; i < b.N; i++ {
+		for _, lk := range ledgerKeys {
+			_, _ = e.LedgerKeyUnsafeMarshalBinaryCompress(lk)
+		}
+	}
+}

ingest/checkpoint_change_reader.go

@@ -35,6 +35,8 @@ type CheckpointChangeReader struct {
 	totalRead      int64
 	totalSize      int64
 
+	encodingBuffer *xdr.EncodingBuffer
+
 	// This should be set to true in tests only
 	disableBucketListHashValidation bool
 	sleep                           func(time.Duration)
@@ -109,16 +111,17 @@ func NewCheckpointChangeReader(
 	}
 
 	return &CheckpointChangeReader{
-		ctx:        ctx,
-		has:        &has,
-		archive:    archive,
-		tempStore:  tempStore,
-		sequence:   sequence,
-		readChan:   make(chan readResult, msrBufferSize),
-		streamOnce: sync.Once{},
-		closeOnce:  sync.Once{},
-		done:       make(chan bool),
-		sleep:      time.Sleep,
+		ctx:            ctx,
+		has:            &has,
+		archive:        archive,
+		tempStore:      tempStore,
+		sequence:       sequence,
+		readChan:       make(chan readResult, msrBufferSize),
+		streamOnce:     sync.Once{},
+		closeOnce:      sync.Once{},
+		done:           make(chan bool),
+		encodingBuffer: xdr.NewEncodingBuffer(),
+		sleep:          time.Sleep,
 	}, nil
 }
 
@@ -367,7 +370,8 @@ LoopBucketEntry:
 				}
 
 				// We're using compressed keys here
-				keyBytes, e := key.MarshalBinaryCompress()
+				// safe, since we are converting to string right away
+				keyBytes, e := r.encodingBuffer.LedgerKeyUnsafeMarshalBinaryCompress(key)
 				if e != nil {
 					r.readChan <- r.error(
 						errors.Wrapf(e, "Error marshaling XDR record %d of hash '%s'", n, hash.String()),
@@ -421,7 +425,8 @@ LoopBucketEntry:
 		}
 
 		// We're using compressed keys here
-		keyBytes, e := key.MarshalBinaryCompress()
+		// Safe, since we are converting to string right away
+		keyBytes, e := r.encodingBuffer.LedgerKeyUnsafeMarshalBinaryCompress(key)
 		if e != nil {
 			r.readChan <- r.error(
 				errors.Wrapf(

xdr/account_id.go

@@ -64,28 +64,17 @@ func (aid *AccountId) LedgerKey() (ret LedgerKey) {
 	return
 }
 
-// MarshalBinaryCompress marshals AccountId to []byte but unlike
-// MarshalBinary() it removes all unnecessary bytes, exploting the fact
-// that XDR is padding data to 4 bytes in union discriminants etc.
-// It's primary use is in ingest/io.StateReader that keep LedgerKeys in
-// memory so this function decrease memory requirements.
-//
-// Warning, do not use UnmarshalBinary() on data encoded using this method!
-func (aid AccountId) MarshalBinaryCompress() ([]byte, error) {
-	m := []byte{byte(aid.Type)}
-
+func (e *EncodingBuffer) accountIdCompressEncodeTo(aid AccountId) error {
+	if err := e.xdrEncoderBuf.WriteByte(byte(aid.Type)); err != nil {
+		return err
+	}
 	switch aid.Type {
 	case PublicKeyTypePublicKeyTypeEd25519:
-		pk, err := aid.Ed25519.MarshalBinary()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, pk...)
+		_, err := e.xdrEncoderBuf.Write(aid.Ed25519[:])
+		return err
 	default:
 		panic("Unknown type")
 	}
-
-	return m, nil
 }
 
 func MustAddress(address string) AccountId {

xdr/asset.go

@@ -268,44 +268,6 @@ func (a Asset) StringCanonical() string {
 	return fmt.Sprintf("%s:%s", c, i)
 }
 
-// MarshalBinaryCompress marshals Asset to []byte but unlike
-// MarshalBinary() it removes all unnecessary bytes, exploting the fact
-// that XDR is padding data to 4 bytes in union discriminants etc.
-// It's primary use is in ingest/io.StateReader that keep LedgerKeys in
-// memory so this function decrease memory requirements.
-//
-// Warning, do not use UnmarshalBinary() on data encoded using this method!
-func (a Asset) MarshalBinaryCompress() ([]byte, error) {
-	m := []byte{byte(a.Type)}
-
-	var err error
-	var code []byte
-	var issuer []byte
-
-	switch a.Type {
-	case AssetTypeAssetTypeNative:
-		return m, nil
-	case AssetTypeAssetTypeCreditAlphanum4:
-		code = []byte(strings.TrimRight(string(a.AlphaNum4.AssetCode[:]), "\x00"))
-		issuer, err = a.AlphaNum4.Issuer.MarshalBinary()
-		if err != nil {
-			return nil, err
-		}
-	case AssetTypeAssetTypeCreditAlphanum12:
-		code = []byte(strings.TrimRight(string(a.AlphaNum12.AssetCode[:]), "\x00"))
-		issuer, err = a.AlphaNum12.Issuer.MarshalBinary()
-		if err != nil {
-			panic(err)
-		}
-	default:
-		panic(fmt.Errorf("Unknown asset type: %v", a.Type))
-	}
-
-	m = append(m, code...)
-	m = append(m, issuer...)
-	return m, nil
-}
-
 // Equals returns true if `other` is equivalent to `a`
 func (a Asset) Equals(other Asset) bool {
 	if a.Type != other.Type {

xdr/asset_test.go

@@ -3,12 +3,12 @@ package xdr_test
 import (
 	"testing"
 
+	. "github.com/stellar/go/xdr"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
-	. "github.com/stellar/go/xdr"
 )
 
 var _ = Describe("xdr.Asset#Extract()", func() {

xdr/claimable_balance_id.go

@@ -1,27 +1,14 @@
 package xdr
 
-// MarshalBinaryCompress marshals ClaimableBalanceId to []byte but unlike
-// MarshalBinary() it removes all unnecessary bytes, exploiting the fact
-// that XDR is padding data to 4 bytes in union discriminants etc.
-// It's primary use is in ingest/io.StateReader that keep LedgerKeys in
-// memory so this function decrease memory requirements.
-//
-// Warning, do not use UnmarshalBinary() on data encoded using this method!
-func (cb ClaimableBalanceId) MarshalBinaryCompress() ([]byte, error) {
-	m := []byte{byte(cb.Type)}
-
+func (e *EncodingBuffer) claimableBalanceCompressEncodeTo(cb ClaimableBalanceId) error {
+	if err := e.xdrEncoderBuf.WriteByte(byte(cb.Type)); err != nil {
+		return err
+	}
 	switch cb.Type {
 	case ClaimableBalanceIdTypeClaimableBalanceIdTypeV0:
-		hash, err := cb.V0.MarshalBinary()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, hash...)
-	// TODO fix before Protocol 18
-	// case ClaimableBalanceIdTypeClaimableBalanceIdTypeFromPoolRevoke:
+		_, err := e.xdrEncoderBuf.Write(cb.V0[:])
+		return err
 	default:
 		panic("Unknown type")
 	}
-
-	return m, nil
 }

xdr/ledger_key.go

@@ -3,7 +3,6 @@ package xdr
 import (
 	"encoding/base64"
 	"fmt"
-	"strings"
 )
 
 // LedgerKey implements the `Keyer` interface
@@ -107,7 +106,7 @@ func (key *LedgerKey) SetClaimableBalance(balanceID ClaimableBalanceId) error {
 	return nil
 }
 
-// SetL LquidityPool mutates `key` such that it represents the identity of a
+// SetLiquidityPool mutates `key` such that it represents the identity of a
 // liquidity pool.
 func (key *LedgerKey) SetLiquidityPool(poolID PoolId) error {
 	data := LedgerKeyLiquidityPool{poolID}
@@ -120,74 +119,39 @@ func (key *LedgerKey) SetLiquidityPool(poolID PoolId) error {
 	return nil
 }
 
-// MarshalBinaryCompress marshals LedgerKey to []byte but unlike
-// MarshalBinary() it removes all unnecessary bytes, exploting the fact
-// that XDR is padding data to 4 bytes in union discriminants etc.
-// It's primary use is in ingest/io.StateReader that keep LedgerKeys in
-// memory so this function decrease memory requirements.
-//
-// Warning, do not use UnmarshalBinary() on data encoded using this method!
-//
-// Optimizations:
-// - Writes a single byte for union discriminants vs 4 bytes.
-// - Removes type and code padding for Asset.
-func (key LedgerKey) MarshalBinaryCompress() ([]byte, error) {
-	m := []byte{byte(key.Type)}
+func (e *EncodingBuffer) ledgerKeyCompressEncodeTo(key LedgerKey) error {
+	if err := e.xdrEncoderBuf.WriteByte(byte(key.Type)); err != nil {
+		return err
+	}
 
 	switch key.Type {
 	case LedgerEntryTypeAccount:
-		account, err := key.Account.AccountId.MarshalBinaryCompress()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, account...)
+		return e.accountIdCompressEncodeTo(key.Account.AccountId)
 	case LedgerEntryTypeTrustline:
-		account, err := key.TrustLine.AccountId.MarshalBinaryCompress()
-		if err != nil {
-			return nil, err
+		if err := e.accountIdCompressEncodeTo(key.TrustLine.AccountId); err != nil {
+			return err
 		}
-		m = append(m, account...)
-		asset, err := key.TrustLine.Asset.MarshalBinaryCompress()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, asset...)
+		return e.assetTrustlineCompressEncodeTo(key.TrustLine.Asset)
 	case LedgerEntryTypeOffer:
-		seller, err := key.Offer.SellerId.MarshalBinaryCompress()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, seller...)
-		offer, err := key.Offer.OfferId.MarshalBinary()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, offer...)
+		// We intentionally don't encode the SellerID since the OfferID is enough
+		// (it's unique to the network)
+		return key.Offer.OfferId.EncodeTo(e.encoder)
 	case LedgerEntryTypeData:
-		account, err := key.Data.AccountId.MarshalBinaryCompress()
-		if err != nil {
-			return nil, err
+		if err := e.accountIdCompressEncodeTo(key.Data.AccountId); err != nil {
+			return err
 		}
-		m = append(m, account...)
-		dataName := []byte(strings.TrimRight(string(key.Data.DataName), "\x00"))
-		m = append(m, dataName...)
+		dataName := trimRightZeros([]byte(key.Data.DataName))
+		_, err := e.xdrEncoderBuf.Write(dataName)
+		return err
 	case LedgerEntryTypeClaimableBalance:
-		cBalance, err := key.ClaimableBalance.BalanceId.MarshalBinaryCompress()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, cBalance...)
+		return e.claimableBalanceCompressEncodeTo(key.ClaimableBalance.BalanceId)
 	case LedgerEntryTypeLiquidityPool:
-		cBalance, err := key.LiquidityPool.LiquidityPoolId.MarshalBinary()
-		if err != nil {
-			return nil, err
-		}
-		m = append(m, cBalance...)
+		_, err := e.xdrEncoderBuf.Write(key.LiquidityPool.LiquidityPoolId[:])
+		return err
 	default:
 		panic("Unknown type")
 	}
 
-	return m, nil
 }
 
 // MarshalBinaryBase64 marshals XDR into a binary form and then encodes it

xdr/ledger_key_test.go

@@ -1,18 +1,18 @@
-package xdr_test
+package xdr
 
 import (
 	"encoding/base64"
 	"testing"
 
-	"github.com/stellar/go/xdr"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestLedgerKeyTrustLineBinaryMaxLength(t *testing.T) {
-	key := &xdr.LedgerKey{}
+	key := &LedgerKey{}
 	err := key.SetTrustline(
-		xdr.MustAddress("GBFLTCDLOE6YQ74B66RH3S2UW5I2MKZ5VLTM75F4YMIWUIXRIFVNRNIF"),
-		xdr.MustNewCreditAsset("123456789012", "GBFLTCDLOE6YQ74B66RH3S2UW5I2MKZ5VLTM75F4YMIWUIXRIFVNRNIF").ToTrustLineAsset(),
+		MustAddress("GBFLTCDLOE6YQ74B66RH3S2UW5I2MKZ5VLTM75F4YMIWUIXRIFVNRNIF"),
+		MustNewCreditAsset("123456789012", "GBFLTCDLOE6YQ74B66RH3S2UW5I2MKZ5VLTM75F4YMIWUIXRIFVNRNIF").ToTrustLineAsset(),
 	)
 	assert.NoError(t, err)
 
@@ -22,3 +22,19 @@ func TestLedgerKeyTrustLineBinaryMaxLength(t *testing.T) {
 	bcompressed := base64.StdEncoding.EncodeToString(compressed)
 	assert.Equal(t, len(bcompressed), 124)
 }
+
+func TestTrimRightZeros(t *testing.T) {
+	require.Equal(t, []byte(nil), trimRightZeros(nil))
+	require.Equal(t, []byte{}, trimRightZeros([]byte{}))
+	require.Equal(t, []byte{}, trimRightZeros([]byte{0x0}))
+	require.Equal(t, []byte{}, trimRightZeros([]byte{0x0, 0x0}))
+	require.Equal(t, []byte{0x1}, trimRightZeros([]byte{0x1}))
+	require.Equal(t, []byte{0x1}, trimRightZeros([]byte{0x1, 0x0}))
+	require.Equal(t, []byte{0x1}, trimRightZeros([]byte{0x1, 0x0, 0x0}))
+	require.Equal(t, []byte{0x1}, trimRightZeros([]byte{0x1, 0x0, 0x0, 0x0}))
+	require.Equal(t, []byte{0x1, 0x2}, trimRightZeros([]byte{0x1, 0x2}))
+	require.Equal(t, []byte{0x1, 0x2}, trimRightZeros([]byte{0x1, 0x2, 0x0}))
+	require.Equal(t, []byte{0x1, 0x2}, trimRightZeros([]byte{0x1, 0x2, 0x0, 0x0}))
+	require.Equal(t, []byte{0x0, 0x2}, trimRightZeros([]byte{0x0, 0x2, 0x0, 0x0}))
+	require.Equal(t, []byte{0x0, 0x2, 0x0, 0x1}, trimRightZeros([]byte{0x0, 0x2, 0x0, 0x1, 0x0}))
+}