Repository: stellar/go
Pull request #5311 (2024-05-13 21:02:34)

exp/services/ledgerexporter: refactoring LedgerMetaArchive
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What
Refactoring and enhancements in LedgerMetaArchive:
-  Moved `createLedgerCloseMeta` test helper function out of LedgerCloseMetaArchive.
- Added unit tests for LedgerCloseMetaArchive `GetLedger` method.
- Removed getter methods from LedgerCloseMetaArchive to align with go convention.




Patch:
diff --git a/exp/services/ledgerexporter/internal/exportmanager.go b/exp/services/ledgerexporter/internal/exportmanager.go
index 5ee06b8c..5e6c87bb 100644
--- a/exp/services/ledgerexporter/internal/exportmanager.go
+++ b/exp/services/ledgerexporter/internal/exportmanager.go
@@ -15,7 +15,7 @@ import (
 type ExportManager struct {
 	config             datastore.LedgerBatchConfig
 	ledgerBackend      ledgerbackend.LedgerBackend
-	currentMetaArchive *datastore.LedgerMetaArchive
+	currentMetaArchive *LedgerMetaArchive
 	queue              UploadQueue
 	latestLedgerMetric *prometheus.GaugeVec
 }
@@ -47,7 +47,7 @@ func (e *ExportManager) AddLedgerCloseMeta(ctx context.Context, ledgerCloseMeta
 	// Determine the object key for the given ledger sequence
 	objectKey := e.config.GetObjectKeyFromSequenceNumber(ledgerSeq)
 
-	if e.currentMetaArchive != nil && e.currentMetaArchive.GetObjectKey() != objectKey {
+	if e.currentMetaArchive != nil && e.currentMetaArchive.ObjectKey != objectKey {
 		return errors.New("Current meta archive object key mismatch")
 	}
 	if e.currentMetaArchive == nil {
@@ -61,14 +61,14 @@ func (e *ExportManager) AddLedgerCloseMeta(ctx context.Context, ledgerCloseMeta
 		}
 
 		// Create a new LedgerMetaArchive and add it to the map.
-		e.currentMetaArchive = datastore.NewLedgerMetaArchive(objectKey, ledgerSeq, endSeq)
+		e.currentMetaArchive = NewLedgerMetaArchive(objectKey, ledgerSeq, endSeq)
 	}
 
-	if err := e.currentMetaArchive.AddLedger(ledgerCloseMeta); err != nil {
+	if err := e.currentMetaArchive.Data.AddLedger(ledgerCloseMeta); err != nil {
 		return errors.Wrapf(err, "failed to add ledger %d", ledgerSeq)
 	}
 
-	if ledgerSeq >= e.currentMetaArchive.GetEndLedgerSequence() {
+	if ledgerSeq >= uint32(e.currentMetaArchive.Data.EndSequence) {
 		// Current archive is full, send it for upload
 		if err := e.queue.Enqueue(ctx, e.currentMetaArchive); err != nil {
 			return err
diff --git a/exp/services/ledgerexporter/internal/ledger_meta_archive.go b/exp/services/ledgerexporter/internal/ledger_meta_archive.go
new file mode 100644
index 00000000..5215e235
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/ledger_meta_archive.go
@@ -0,0 +1,24 @@
+package ledgerexporter
+
+import (
+	"github.com/stellar/go/xdr"
+)
+
+// LedgerMetaArchive represents a file with metadata and binary data.
+type LedgerMetaArchive struct {
+	// file name
+	ObjectKey string
+	// Actual binary data
+	Data xdr.LedgerCloseMetaBatch
+}
+
+// NewLedgerMetaArchive creates a new LedgerMetaArchive instance.
+func NewLedgerMetaArchive(key string, startSeq uint32, endSeq uint32) *LedgerMetaArchive {
+	return &LedgerMetaArchive{
+		ObjectKey: key,
+		Data: xdr.LedgerCloseMetaBatch{
+			StartSequence: xdr.Uint32(startSeq),
+			EndSequence:   xdr.Uint32(endSeq),
+		},
+	}
+}
diff --git a/exp/services/ledgerexporter/internal/queue.go b/exp/services/ledgerexporter/internal/queue.go
index 5b46b202..372ccb00 100644
--- a/exp/services/ledgerexporter/internal/queue.go
+++ b/exp/services/ledgerexporter/internal/queue.go
@@ -4,12 +4,11 @@ import (
 	"context"
 
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/stellar/go/support/datastore"
 )
 
 // UploadQueue is a queue of LedgerMetaArchive objects which are scheduled for upload
 type UploadQueue struct {
-	metaArchiveCh     chan *datastore.LedgerMetaArchive
+	metaArchiveCh     chan *LedgerMetaArchive
 	queueLengthMetric prometheus.Gauge
 }
 
@@ -23,13 +22,13 @@ func NewUploadQueue(size int, prometheusRegistry *prometheus.Registry) UploadQue
 	})
 	prometheusRegistry.MustRegister(queueLengthMetric)
 	return UploadQueue{
-		metaArchiveCh:     make(chan *datastore.LedgerMetaArchive, size),
+		metaArchiveCh:     make(chan *LedgerMetaArchive, size),
 		queueLengthMetric: queueLengthMetric,
 	}
 }
 
 // Enqueue will add an upload task to the queue. Enqueue may block if the queue is full.
-func (u UploadQueue) Enqueue(ctx context.Context, archive *datastore.LedgerMetaArchive) error {
+func (u UploadQueue) Enqueue(ctx context.Context, archive *LedgerMetaArchive) error {
 	u.queueLengthMetric.Inc()
 	select {
 	case u.metaArchiveCh <- archive:
@@ -40,7 +39,7 @@ func (u UploadQueue) Enqueue(ctx context.Context, archive *datastore.LedgerMetaA
 }
 
 // Dequeue will pop a task off the queue. Dequeue may block if the queue is empty.
-func (u UploadQueue) Dequeue(ctx context.Context) (*datastore.LedgerMetaArchive, bool, error) {
+func (u UploadQueue) Dequeue(ctx context.Context) (*LedgerMetaArchive, bool, error) {
 	select {
 	case <-ctx.Done():
 		return nil, false, ctx.Err()
diff --git a/exp/services/ledgerexporter/internal/uploader.go b/exp/services/ledgerexporter/internal/uploader.go
index c1f80d64..94344ee1 100644
--- a/exp/services/ledgerexporter/internal/uploader.go
+++ b/exp/services/ledgerexporter/internal/uploader.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/stellar/go/support/compressxdr"
 	"github.com/stellar/go/support/datastore"
 )
@@ -78,19 +79,19 @@ func (r *writerToRecorder) WriteTo(w io.Writer) (int64, error) {
 }
 
 // Upload uploads the serialized binary data of ledger TxMeta to the specified destination.
-func (u Uploader) Upload(ctx context.Context, metaArchive *datastore.LedgerMetaArchive) error {
-	logger.Infof("Uploading: %s", metaArchive.GetObjectKey())
+func (u Uploader) Upload(ctx context.Context, metaArchive *LedgerMetaArchive) error {
+	logger.Infof("Uploading: %s", metaArchive.ObjectKey)
 	startTime := time.Now()
-	numLedgers := strconv.FormatUint(uint64(metaArchive.GetLedgerCount()), 10)
+	numLedgers := strconv.FormatUint(uint64(len(metaArchive.Data.LedgerCloseMetas)), 10)
 
 	xdrEncoder := compressxdr.NewXDREncoder(compressxdr.DefaultCompressor, &metaArchive.Data)
 
 	writerTo := &writerToRecorder{
 		WriterTo: xdrEncoder,
 	}
-	ok, err := u.dataStore.PutFileIfNotExists(ctx, metaArchive.GetObjectKey(), writerTo)
+	ok, err := u.dataStore.PutFileIfNotExists(ctx, metaArchive.ObjectKey, writerTo)
 	if err != nil {
-		return errors.Wrapf(err, "error uploading %s", metaArchive.GetObjectKey())
+		return errors.Wrapf(err, "error uploading %s", metaArchive.ObjectKey)
 	}
 	alreadyExists := strconv.FormatBool(!ok)
 
diff --git a/ingest/ledgerbackend/buffered_storage_backend.go b/ingest/ledgerbackend/buffered_storage_backend.go
index c3c1007e..41ff4a94 100644
--- a/ingest/ledgerbackend/buffered_storage_backend.go
+++ b/ingest/ledgerbackend/buffered_storage_backend.go
@@ -36,12 +36,12 @@ type BufferedStorageBackend struct {
 	// ledgerBuffer is the buffer for LedgerCloseMeta data read in parallel.
 	ledgerBuffer *ledgerBuffer
 
-	dataStore         datastore.DataStore
-	prepared          *Range // Non-nil if any range is prepared
-	closed            bool   // False until the core is closed
-	ledgerMetaArchive *datastore.LedgerMetaArchive
-	nextLedger        uint32
-	lastLedger        uint32
+	dataStore  datastore.DataStore
+	prepared   *Range // Non-nil if any range is prepared
+	closed     bool   // False until the core is closed
+	lcmBatch   xdr.LedgerCloseMetaBatch
+	nextLedger uint32
+	lastLedger uint32
 }
 
 // NewBufferedStorageBackend returns a new BufferedStorageBackend instance.
@@ -62,12 +62,9 @@ func NewBufferedStorageBackend(ctx context.Context, config BufferedStorageBacken
 		return nil, errors.New("ledgersPerFile must be > 0")
 	}
 
-	ledgerMetaArchive := datastore.NewLedgerMetaArchive("", 0, 0)
-
 	bsBackend := &BufferedStorageBackend{
-		config:            config,
-		dataStore:         config.DataStore,
-		ledgerMetaArchive: ledgerMetaArchive,
+		config:    config,
+		dataStore: config.DataStore,
 	}
 
 	return bsBackend, nil
@@ -98,19 +95,19 @@ func (bsb *BufferedStorageBackend) GetLatestLedgerSequence(ctx context.Context)
 // Otherwise will continuously load in the next LedgerCloseMetaBatch until found.
 func (bsb *BufferedStorageBackend) getBatchForSequence(ctx context.Context, sequence uint32) error {
 	// Sequence inside the current cached LedgerCloseMetaBatch
-	if sequence >= bsb.ledgerMetaArchive.GetStartLedgerSequence() && sequence <= bsb.ledgerMetaArchive.GetEndLedgerSequence() {
+	if sequence >= uint32(bsb.lcmBatch.StartSequence) && sequence <= uint32(bsb.lcmBatch.EndSequence) {
 		return nil
 	}
 
 	// Sequence is before the current LedgerCloseMetaBatch
 	// Does not support retrieving LedgerCloseMeta before the current cached batch
-	if sequence < bsb.ledgerMetaArchive.GetStartLedgerSequence() {
-		return errors.New("requested sequence preceeds current LedgerCloseMetaBatch")
+	if sequence < uint32(bsb.lcmBatch.StartSequence) {
+		return errors.New("requested sequence precedes current LedgerCloseMetaBatch")
 	}
 
 	// Sequence is beyond the current LedgerCloseMetaBatch
 	var err error
-	bsb.ledgerMetaArchive.Data, err = bsb.ledgerBuffer.getFromLedgerQueue(ctx)
+	bsb.lcmBatch, err = bsb.ledgerBuffer.getFromLedgerQueue(ctx)
 	if err != nil {
 		return errors.Wrap(err, "failed getting next ledger batch from queue")
 	}
@@ -165,7 +162,7 @@ func (bsb *BufferedStorageBackend) GetLedger(ctx context.Context, sequence uint3
 		return xdr.LedgerCloseMeta{}, err
 	}
 
-	ledgerCloseMeta, err := bsb.ledgerMetaArchive.GetLedger(sequence)
+	ledgerCloseMeta, err := bsb.lcmBatch.GetLedger(sequence)
 	if err != nil {
 		return xdr.LedgerCloseMeta{}, err
 	}
diff --git a/support/datastore/ledger_meta_archive.go b/support/datastore/ledger_meta_archive.go
deleted file mode 100644
index 7942a8bd..00000000
--- a/support/datastore/ledger_meta_archive.go
+++ /dev/null
@@ -1,96 +0,0 @@
-package datastore
-
-import (
-	"fmt"
-
-	"github.com/stellar/go/xdr"
-)
-
-// LedgerMetaArchive represents a file with metadata and binary data.
-type LedgerMetaArchive struct {
-	// file name
-	ObjectKey string
-	// Actual binary data
-	Data xdr.LedgerCloseMetaBatch
-}
-
-// NewLedgerMetaArchive creates a new LedgerMetaArchive instance.
-func NewLedgerMetaArchive(key string, startSeq uint32, endSeq uint32) *LedgerMetaArchive {
-	return &LedgerMetaArchive{
-		ObjectKey: key,
-		Data: xdr.LedgerCloseMetaBatch{
-			StartSequence: xdr.Uint32(startSeq),
-			EndSequence:   xdr.Uint32(endSeq),
-		},
-	}
-}
-
-// AddLedger adds a LedgerCloseMeta to the archive.
-func (f *LedgerMetaArchive) AddLedger(ledgerCloseMeta xdr.LedgerCloseMeta) error {
-	if ledgerCloseMeta.LedgerSequence() < uint32(f.Data.StartSequence) ||
-		ledgerCloseMeta.LedgerSequence() > uint32(f.Data.EndSequence) {
-		return fmt.Errorf("ledger sequence %d is outside valid range [%d, %d]",
-			ledgerCloseMeta.LedgerSequence(), f.Data.StartSequence, f.Data.EndSequence)
-	}
-
-	if len(f.Data.LedgerCloseMetas) > 0 {
-		lastSequence := f.Data.LedgerCloseMetas[len(f.Data.LedgerCloseMetas)-1].LedgerSequence()
-		if ledgerCloseMeta.LedgerSequence() != lastSequence+1 {
-			return fmt.Errorf("ledgers must be added sequentially: expected sequence %d, got %d",
-				lastSequence+1, ledgerCloseMeta.LedgerSequence())
-		}
-	}
-	f.Data.LedgerCloseMetas = append(f.Data.LedgerCloseMetas, ledgerCloseMeta)
-	return nil
-}
-
-// GetLedgerCount returns the number of ledgers currently in the archive.
-func (f *LedgerMetaArchive) GetLedgerCount() uint32 {
-	return uint32(len(f.Data.LedgerCloseMetas))
-}
-
-// GetStartLedgerSequence returns the starting ledger sequence of the archive.
-func (f *LedgerMetaArchive) GetStartLedgerSequence() uint32 {
-	return uint32(f.Data.StartSequence)
-}
-
-// GetEndLedgerSequence returns the ending ledger sequence of the archive.
-func (f *LedgerMetaArchive) GetEndLedgerSequence() uint32 {
-	return uint32(f.Data.EndSequence)
-}
-
-// GetObjectKey returns the object key of the archive.
-func (f *LedgerMetaArchive) GetObjectKey() string {
-	return f.ObjectKey
-}
-
-func (f *LedgerMetaArchive) GetLedger(sequence uint32) (xdr.LedgerCloseMeta, error) {
-	if sequence < uint32(f.Data.StartSequence) || sequence > uint32(f.Data.EndSequence) {
-		return xdr.LedgerCloseMeta{}, fmt.Errorf("ledger sequence %d is outside valid range [%d, %d]",
-			sequence, f.Data.StartSequence, f.Data.EndSequence)
-	}
-
-	ledgerIndex := sequence - f.GetStartLedgerSequence()
-	if ledgerIndex >= uint32(len(f.Data.LedgerCloseMetas)) {
-		return xdr.LedgerCloseMeta{}, fmt.Errorf("LedgerCloseMeta for sequence %d not found", sequence)
-	}
-	return f.Data.LedgerCloseMetas[ledgerIndex], nil
-}
-
-func CreateLedgerCloseMeta(ledgerSeq uint32) xdr.LedgerCloseMeta {
-	return xdr.LedgerCloseMeta{
-		V: int32(0),
-		V0: &xdr.LedgerCloseMetaV0{
-			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
-				Header: xdr.LedgerHeader{
-					LedgerSeq: xdr.Uint32(ledgerSeq),
-				},
-			},
-			TxSet:              xdr.TransactionSet{},
-			TxProcessing:       nil,
-			UpgradesProcessing: nil,
-			ScpInfo:            nil,
-		},
-		V1: nil,
-	}
-}
diff --git a/xdr/ledger_close_meta_batch.go b/xdr/ledger_close_meta_batch.go
new file mode 100644
index 00000000..7fb212d8
--- /dev/null
+++ b/xdr/ledger_close_meta_batch.go
@@ -0,0 +1,41 @@
+package xdr
+
+import (
+	"fmt"
+)
+
+// GetLedger retrieves the LedgerCloseMeta for a given sequence number.
+// It returns an error if LedgerCloseMeta for the sequence number is not found in the batch.
+func (s *LedgerCloseMetaBatch) GetLedger(sequence uint32) (LedgerCloseMeta, error) {
+
+	if sequence < uint32(s.StartSequence) || sequence > uint32(s.EndSequence) {
+		return LedgerCloseMeta{}, fmt.Errorf("ledger sequence %d is outside the "+
+			"valid range of ledger sequences [%d, %d] this batch holds",
+			sequence, s.StartSequence, s.EndSequence)
+	}
+
+	ledgerIndex := sequence - uint32(s.StartSequence)
+	if ledgerIndex >= uint32(len(s.LedgerCloseMetas)) {
+		return LedgerCloseMeta{}, fmt.Errorf("LedgerCloseMeta for sequence %d not found in the batch", sequence)
+	}
+	return s.LedgerCloseMetas[ledgerIndex], nil
+}
+
+// AddLedger adds a LedgerCloseMeta to the batch.
+func (s *LedgerCloseMetaBatch) AddLedger(ledgerCloseMeta LedgerCloseMeta) error {
+	if ledgerCloseMeta.LedgerSequence() < uint32(s.StartSequence) ||
+		ledgerCloseMeta.LedgerSequence() > uint32(s.EndSequence) {
+		return fmt.Errorf("ledger sequence %d is outside valid range [%d, %d]",
+			ledgerCloseMeta.LedgerSequence(), s.StartSequence, s.EndSequence)
+	}
+
+	if len(s.LedgerCloseMetas) > 0 {
+		lastSequence := s.LedgerCloseMetas[len(s.LedgerCloseMetas)-1].LedgerSequence()
+		if ledgerCloseMeta.LedgerSequence() != lastSequence+1 {
+			return fmt.Errorf("ledgers must be added sequentially: expected sequence %d, got %d",
+				lastSequence+1, ledgerCloseMeta.LedgerSequence())
+		}
+	}
+	s.LedgerCloseMetas = append(s.LedgerCloseMetas, ledgerCloseMeta)
+	return nil
+}


Test patch:
diff --git a/exp/services/ledgerexporter/internal/exportmanager_test.go b/exp/services/ledgerexporter/internal/exportmanager_test.go
index f3b302b7..a84e20fb 100644
--- a/exp/services/ledgerexporter/internal/exportmanager_test.go
+++ b/exp/services/ledgerexporter/internal/exportmanager_test.go
@@ -15,8 +15,27 @@ import (
 	"github.com/stellar/go/ingest/ledgerbackend"
 	"github.com/stellar/go/support/collections/set"
 	"github.com/stellar/go/support/datastore"
+	"github.com/stellar/go/xdr"
 )
 
+func createLedgerCloseMeta(ledgerSeq uint32) xdr.LedgerCloseMeta {
+	return xdr.LedgerCloseMeta{
+		V: int32(0),
+		V0: &xdr.LedgerCloseMetaV0{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
+				Header: xdr.LedgerHeader{
+					LedgerSeq: xdr.Uint32(ledgerSeq),
+				},
+			},
+			TxSet:              xdr.TransactionSet{},
+			TxProcessing:       nil,
+			UpgradesProcessing: nil,
+			ScpInfo:            nil,
+		},
+		V1: nil,
+	}
+}
+
 func TestExporterSuite(t *testing.T) {
 	suite.Run(t, new(ExportManagerSuite))
 }
@@ -57,7 +76,7 @@ func (s *ExportManagerSuite) TestRun() {
 	expectedKeys := set.NewSet[string](10)
 	for i := start; i <= end; i++ {
 		s.mockBackend.On("GetLedger", s.ctx, i).
-			Return(datastore.CreateLedgerCloseMeta(i), nil)
+			Return(createLedgerCloseMeta(i), nil)
 		key := config.GetObjectKeyFromSequenceNumber(i)
 		expectedKeys.Add(key)
 	}
@@ -104,7 +123,7 @@ func (s *ExportManagerSuite) TestRunContextCancel() {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	s.mockBackend.On("GetLedger", mock.Anything, mock.Anything).
-		Return(datastore.CreateLedgerCloseMeta(1), nil)
+		Return(createLedgerCloseMeta(1), nil)
 
 	go func() {
 		<-time.After(time.Second * 1)
@@ -164,7 +183,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 	start := uint32(0)
 	end := uint32(255)
 	for i := start; i <= end; i++ {
-		require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), datastore.CreateLedgerCloseMeta(i)))
+		require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(i)))
 
 		key := config.GetObjectKeyFromSequenceNumber(i)
 		expectedKeys.Add(key)
@@ -188,8 +207,8 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
 		cancel()
 	}()
 
-	require.NoError(s.T(), exporter.AddLedgerCloseMeta(ctx, datastore.CreateLedgerCloseMeta(1)))
-	err = exporter.AddLedgerCloseMeta(ctx, datastore.CreateLedgerCloseMeta(2))
+	require.NoError(s.T(), exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(1)))
+	err = exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(2))
 	require.EqualError(s.T(), err, "context canceled")
 }
 
@@ -200,7 +219,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaKeyMismatch() {
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
 	require.NoError(s.T(), err)
 
-	require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), datastore.CreateLedgerCloseMeta(16)))
-	require.EqualError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), datastore.CreateLedgerCloseMeta(21)),
+	require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(16)))
+	require.EqualError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(21)),
 		"Current meta archive object key mismatch")
 }
diff --git a/exp/services/ledgerexporter/internal/queue_test.go b/exp/services/ledgerexporter/internal/queue_test.go
index 4f5fdd26..0676f359 100644
--- a/exp/services/ledgerexporter/internal/queue_test.go
+++ b/exp/services/ledgerexporter/internal/queue_test.go
@@ -6,7 +6,6 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
-	"github.com/stellar/go/support/datastore"
 	"github.com/stretchr/testify/require"
 )
 
@@ -32,9 +31,9 @@ func getMetricValue(metric prometheus.Metric) *dto.Metric {
 func TestQueue(t *testing.T) {
 	queue := NewUploadQueue(3, prometheus.NewRegistry())
 
-	require.NoError(t, queue.Enqueue(context.Background(), datastore.NewLedgerMetaArchive("test", 1, 1)))
-	require.NoError(t, queue.Enqueue(context.Background(), datastore.NewLedgerMetaArchive("test", 2, 2)))
-	require.NoError(t, queue.Enqueue(context.Background(), datastore.NewLedgerMetaArchive("test", 3, 3)))
+	require.NoError(t, queue.Enqueue(context.Background(), NewLedgerMetaArchive("test", 1, 1)))
+	require.NoError(t, queue.Enqueue(context.Background(), NewLedgerMetaArchive("test", 2, 2)))
+	require.NoError(t, queue.Enqueue(context.Background(), NewLedgerMetaArchive("test", 3, 3)))
 
 	require.Equal(t, float64(3), getMetricValue(queue.queueLengthMetric).GetGauge().GetValue())
 	queue.Close()
@@ -43,19 +42,19 @@ func TestQueue(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, ok)
 	require.Equal(t, float64(2), getMetricValue(queue.queueLengthMetric).GetGauge().GetValue())
-	require.Equal(t, uint32(1), l.GetStartLedgerSequence())
+	require.Equal(t, uint32(1), uint32(l.Data.StartSequence))
 
 	l, ok, err = queue.Dequeue(context.Background())
 	require.NoError(t, err)
 	require.True(t, ok)
 	require.Equal(t, float64(1), getMetricValue(queue.queueLengthMetric).GetGauge().GetValue())
-	require.Equal(t, uint32(2), l.GetStartLedgerSequence())
+	require.Equal(t, uint32(2), uint32(l.Data.StartSequence))
 
 	l, ok, err = queue.Dequeue(context.Background())
 	require.NoError(t, err)
 	require.True(t, ok)
 	require.Equal(t, float64(0), getMetricValue(queue.queueLengthMetric).GetGauge().GetValue())
-	require.Equal(t, uint32(3), l.GetStartLedgerSequence())
+	require.Equal(t, uint32(3), uint32(l.Data.StartSequence))
 
 	l, ok, err = queue.Dequeue(context.Background())
 	require.NoError(t, err)
diff --git a/exp/services/ledgerexporter/internal/uploader_test.go b/exp/services/ledgerexporter/internal/uploader_test.go
index da0cc6d2..b31397c1 100644
--- a/exp/services/ledgerexporter/internal/uploader_test.go
+++ b/exp/services/ledgerexporter/internal/uploader_test.go
@@ -41,9 +41,9 @@ func (s *UploaderSuite) TestUpload() {
 
 func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 	key, start, end := "test-1-100", uint32(1), uint32(100)
-	archive := datastore.NewLedgerMetaArchive(key, start, end)
+	archive := NewLedgerMetaArchive(key, start, end)
 	for i := start; i <= end; i++ {
-		_ = archive.AddLedger(datastore.CreateLedgerCloseMeta(i))
+		_ = archive.Data.AddLedger(createLedgerCloseMeta(i))
 	}
 
 	var capturedBuf bytes.Buffer
@@ -61,7 +61,7 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 	require.NoError(s.T(), dataUploader.Upload(context.Background(), archive))
 
 	expectedCompressedLength := capturedBuf.Len()
-	var decodedArchive datastore.LedgerMetaArchive
+	var decodedArchive LedgerMetaArchive
 	xdrDecoder := compressxdr.NewXDRDecoder(compressxdr.DefaultCompressor, &decodedArchive.Data)
 
 	decoder := xdrDecoder
@@ -161,7 +161,7 @@ func (s *UploaderSuite) TestUploadPutError() {
 
 func (s *UploaderSuite) testUploadPutError(putOkReturnVal bool) {
 	key, start, end := "test-1-100", uint32(1), uint32(100)
-	archive := datastore.NewLedgerMetaArchive(key, start, end)
+	archive := NewLedgerMetaArchive(key, start, end)
 
 	s.mockDataStore.On("PutFileIfNotExists", context.Background(), key,
 		mock.Anything).Return(putOkReturnVal, errors.New("error in PutFileIfNotExists"))
@@ -209,7 +209,7 @@ func (s *UploaderSuite) TestRunChannelClose() {
 	go func() {
 		key, start, end := "test", uint32(1), uint32(100)
 		for i := start; i <= end; i++ {
-			s.Assert().NoError(queue.Enqueue(s.ctx, datastore.NewLedgerMetaArchive(key, i, i)))
+			s.Assert().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive(key, i, i)))
 		}
 		<-time.After(time.Second * 2)
 		queue.Close()
@@ -225,7 +225,7 @@ func (s *UploaderSuite) TestRunContextCancel() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 
-	s.Assert().NoError(queue.Enqueue(s.ctx, datastore.NewLedgerMetaArchive("test", 1, 1)))
+	s.Assert().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test", 1, 1)))
 
 	go func() {
 		<-time.After(time.Second * 2)
@@ -240,7 +240,7 @@ func (s *UploaderSuite) TestRunUploadError() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 
-	s.Assert().NoError(queue.Enqueue(s.ctx, datastore.NewLedgerMetaArchive("test", 1, 1)))
+	s.Assert().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test", 1, 1)))
 	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test",
 		mock.Anything).Return(false, errors.New("Put error"))
 
diff --git a/ingest/ledgerbackend/buffered_storage_backend_test.go b/ingest/ledgerbackend/buffered_storage_backend_test.go
index 063e234c..1aca8306 100644
--- a/ingest/ledgerbackend/buffered_storage_backend_test.go
+++ b/ingest/ledgerbackend/buffered_storage_backend_test.go
@@ -22,6 +22,24 @@ import (
 var partitionSize = uint32(64000)
 var ledgerPerFileCount = uint32(1)
 
+func createLedgerCloseMeta(ledgerSeq uint32) xdr.LedgerCloseMeta {
+	return xdr.LedgerCloseMeta{
+		V: int32(0),
+		V0: &xdr.LedgerCloseMetaV0{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
+				Header: xdr.LedgerHeader{
+					LedgerSeq: xdr.Uint32(ledgerSeq),
+				},
+			},
+			TxSet:              xdr.TransactionSet{},
+			TxProcessing:       nil,
+			UpgradesProcessing: nil,
+			ScpInfo:            nil,
+		},
+		V1: nil,
+	}
+}
+
 func createBufferedStorageBackendConfigForTesting() BufferedStorageBackendConfig {
 	param := make(map[string]string)
 	param["destination_bucket_path"] = "testURL"
@@ -45,12 +63,10 @@ func createBufferedStorageBackendConfigForTesting() BufferedStorageBackendConfig
 
 func createBufferedStorageBackendForTesting() BufferedStorageBackend {
 	config := createBufferedStorageBackendConfigForTesting()
-	ledgerMetaArchive := datastore.NewLedgerMetaArchive("", 0, 0)
 
 	return BufferedStorageBackend{
-		config:            config,
-		dataStore:         config.DataStore,
-		ledgerMetaArchive: ledgerMetaArchive,
+		config:    config,
+		dataStore: config.DataStore,
 	}
 }
 
@@ -81,7 +97,7 @@ func createMockdataStore(t *testing.T, start, end, partitionSize, count uint32)
 func createLCMForTesting(start, end uint32) []xdr.LedgerCloseMeta {
 	var lcmArray []xdr.LedgerCloseMeta
 	for i := start; i <= end; i++ {
-		lcmArray = append(lcmArray, datastore.CreateLedgerCloseMeta(i))
+		lcmArray = append(lcmArray, createLedgerCloseMeta(i))
 	}
 
 	return lcmArray
@@ -90,7 +106,7 @@ func createLCMForTesting(start, end uint32) []xdr.LedgerCloseMeta {
 func createTestLedgerCloseMetaBatch(startSeq, endSeq, count uint32) xdr.LedgerCloseMetaBatch {
 	var ledgerCloseMetas []xdr.LedgerCloseMeta
 	for i := uint32(0); i < count; i++ {
-		ledgerCloseMetas = append(ledgerCloseMetas, datastore.CreateLedgerCloseMeta(startSeq+uint32(i)))
+		ledgerCloseMetas = append(ledgerCloseMetas, createLedgerCloseMeta(startSeq+uint32(i)))
 	}
 	return xdr.LedgerCloseMetaBatch{
 		StartSequence:    xdr.Uint32(startSeq),
diff --git a/support/datastore/ledger_meta_archive_test.go b/support/datastore/ledger_meta_archive_test.go
deleted file mode 100644
index 26eeadc3..00000000
--- a/support/datastore/ledger_meta_archive_test.go
+++ /dev/null
@@ -1,84 +0,0 @@
-package datastore
-
-import (
-	"fmt"
-	"testing"
-
-	"github.com/stellar/go/xdr"
-	"github.com/stretchr/testify/require"
-)
-
-func createLedgerCloseMeta(ledgerSeq uint32) xdr.LedgerCloseMeta {
-	return xdr.LedgerCloseMeta{
-		V0: &xdr.LedgerCloseMetaV0{
-			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
-				Header: xdr.LedgerHeader{
-					LedgerSeq: xdr.Uint32(ledgerSeq),
-				},
-			},
-		},
-	}
-}
-
-func TestLedgerMetaArchive_AddLedgerValidRange(t *testing.T) {
-
-	tests := []struct {
-		name     string
-		startSeq uint32
-		endSeq   uint32
-		seqNum   uint32
-		errMsg   string
-	}{
-		{startSeq: 10, endSeq: 100, seqNum: 10, errMsg: ""},
-		{startSeq: 10, endSeq: 100, seqNum: 11, errMsg: ""},
-		{startSeq: 10, endSeq: 100, seqNum: 99, errMsg: ""},
-		{startSeq: 10, endSeq: 100, seqNum: 100, errMsg: ""},
-		{startSeq: 10, endSeq: 100, seqNum: 9, errMsg: "ledger sequence 9 is outside valid range [10, 100]"},
-		{startSeq: 10, endSeq: 100, seqNum: 101, errMsg: "ledger sequence 101 is outside valid range [10, 100]"},
-	}
-	for _, tt := range tests {
-		t.Run(fmt.Sprintf("range [%d, %d]: Add seq %d", tt.startSeq, tt.endSeq, tt.seqNum),
-			func(t *testing.T) {
-				f := NewLedgerMetaArchive("", tt.startSeq, tt.endSeq)
-				err := f.AddLedger(createLedgerCloseMeta(tt.seqNum))
-				if tt.errMsg != "" {
-					require.EqualError(t, err, tt.errMsg)
-				} else {
-					require.NoError(t, err)
-				}
-			})
-	}
-}
-func TestLedgerMetaArchive_AddLedgerSequential(t *testing.T) {
-	var start, end uint32 = 1, 100
-	f := NewLedgerMetaArchive("", start, end+100)
-
-	// Add ledgers sequentially
-	for i := start; i <= end; i++ {
-		require.NoError(t, f.AddLedger(createLedgerCloseMeta(i)))
-	}
-
-	// Test out of sequence
-	testCases := []struct {
-		ledgerSeq      uint32
-		expectedErrMsg string
-	}{
-		{
-			end + 2,
-			fmt.Sprintf("ledgers must be added sequentially: expected sequence %d, got %d", end+1, end+2),
-		},
-		{
-			end,
-			fmt.Sprintf("ledgers must be added sequentially: expected sequence %d, got %d", end+1, end),
-		},
-		{
-			end - 1,
-			fmt.Sprintf("ledgers must be added sequentially: expected sequence %d, got %d", end+1, end-1),
-		},
-	}
-
-	for _, tc := range testCases {
-		err := f.AddLedger(createLedgerCloseMeta(tc.ledgerSeq))
-		require.EqualError(t, err, tc.expectedErrMsg)
-	}
-}
diff --git a/xdr/ledger_close_meta_batch_test.go b/xdr/ledger_close_meta_batch_test.go
new file mode 100644
index 00000000..66197db5
--- /dev/null
+++ b/xdr/ledger_close_meta_batch_test.go
@@ -0,0 +1,183 @@
+package xdr
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLedgerMetaArchive_AddLedgerValidRange(t *testing.T) {
+
+	tests := []struct {
+		name     string
+		startSeq uint32
+		endSeq   uint32
+		seqNum   uint32
+		errMsg   string
+	}{
+		{startSeq: 10, endSeq: 100, seqNum: 10, errMsg: ""},
+		{startSeq: 10, endSeq: 100, seqNum: 11, errMsg: ""},
+		{startSeq: 10, endSeq: 100, seqNum: 99, errMsg: ""},
+		{startSeq: 10, endSeq: 100, seqNum: 100, errMsg: ""},
+		{startSeq: 10, endSeq: 100, seqNum: 9, errMsg: "ledger sequence 9 is outside valid range [10, 100]"},
+		{startSeq: 10, endSeq: 100, seqNum: 101, errMsg: "ledger sequence 101 is outside valid range [10, 100]"},
+	}
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("range [%d, %d]: Add seq %d", tt.startSeq, tt.endSeq, tt.seqNum),
+			func(t *testing.T) {
+				f := LedgerCloseMetaBatch{StartSequence: Uint32(tt.startSeq), EndSequence: Uint32(tt.endSeq)}
+				err := f.AddLedger(LedgerCloseMeta{
+					V: int32(0),
+					V0: &LedgerCloseMetaV0{
+						LedgerHeader: LedgerHeaderHistoryEntry{
+							Header: LedgerHeader{
+								LedgerSeq: Uint32(tt.seqNum),
+							},
+						},
+						TxSet:              TransactionSet{},
+						TxProcessing:       nil,
+						UpgradesProcessing: nil,
+						ScpInfo:            nil,
+					},
+					V1: nil,
+				})
+				if tt.errMsg != "" {
+					require.EqualError(t, err, tt.errMsg)
+				} else {
+					require.NoError(t, err)
+				}
+			})
+	}
+}
+func TestLedgerMetaArchive_AddLedgerSequential(t *testing.T) {
+	var start, end uint32 = 1, 100
+	f := LedgerCloseMetaBatch{StartSequence: Uint32(start), EndSequence: Uint32(end + 100)}
+
+	// Add ledgers sequentially
+	for i := start; i <= end; i++ {
+		require.NoError(t, f.AddLedger(LedgerCloseMeta{
+			V: int32(0),
+			V0: &LedgerCloseMetaV0{
+				LedgerHeader: LedgerHeaderHistoryEntry{
+					Header: LedgerHeader{
+						LedgerSeq: Uint32(i),
+					},
+				},
+				TxSet:              TransactionSet{},
+				TxProcessing:       nil,
+				UpgradesProcessing: nil,
+				ScpInfo:            nil,
+			},
+			V1: nil,
+		}))
+	}
+
+	// Test out of sequence
+	testCases := []struct {
+		ledgerSeq      uint32
+		expectedErrMsg string
+	}{
+		{
+			end + 2,
+			fmt.Sprintf("ledgers must be added sequentially: expected sequence %d, got %d", end+1, end+2),
+		},
+		{
+			end,
+			fmt.Sprintf("ledgers must be added sequentially: expected sequence %d, got %d", end+1, end),
+		},
+		{
+			end - 1,
+			fmt.Sprintf("ledgers must be added sequentially: expected sequence %d, got %d", end+1, end-1),
+		},
+	}
+
+	for _, tc := range testCases {
+		err := f.AddLedger(LedgerCloseMeta{
+			V: int32(0),
+			V0: &LedgerCloseMetaV0{
+				LedgerHeader: LedgerHeaderHistoryEntry{
+					Header: LedgerHeader{
+						LedgerSeq: Uint32(tc.ledgerSeq),
+					},
+				},
+				TxSet:              TransactionSet{},
+				TxProcessing:       nil,
+				UpgradesProcessing: nil,
+				ScpInfo:            nil,
+			},
+			V1: nil,
+		})
+		require.EqualError(t, err, tc.expectedErrMsg)
+	}
+}
+
+func TestGetLedger(t *testing.T) {
+	var start, end uint32 = 121, 1300
+	f := LedgerCloseMetaBatch{StartSequence: Uint32(start), EndSequence: Uint32(end)}
+
+	for i := start; i <= end-10; i++ {
+		f.LedgerCloseMetas = append(f.LedgerCloseMetas, LedgerCloseMeta{
+			V: int32(0),
+			V0: &LedgerCloseMetaV0{
+				LedgerHeader: LedgerHeaderHistoryEntry{
+					Header: LedgerHeader{
+						LedgerSeq: Uint32(i),
+					},
+				},
+				TxSet:              TransactionSet{},
+				TxProcessing:       nil,
+				UpgradesProcessing: nil,
+				ScpInfo:            nil,
+			},
+			V1: nil,
+		})
+	}
+
+	testCases := []struct {
+		name           string
+		ledgerSeq      uint32
+		expectedErrMsg string
+	}{
+		{
+			name:           "LedgerSequenceInRange",
+			ledgerSeq:      start,
+			expectedErrMsg: "",
+		},
+		{
+			name:           "LedgerSequenceInRange",
+			ledgerSeq:      start + 10,
+			expectedErrMsg: "",
+		},
+		{
+			name:      "LedgerSequenceAboveRange",
+			ledgerSeq: end + 1,
+			expectedErrMsg: fmt.Sprintf("ledger sequence %d is outside the valid range "+
+				"of ledger sequences [%d, %d] this batch holds", end+1, start, end),
+		},
+		{
+			name:      "LedgerSequenceBelowRange",
+			ledgerSeq: start - 1,
+			expectedErrMsg: fmt.Sprintf("ledger sequence %d is outside the valid range "+
+				"of ledger sequences [%d, %d] this batch holds", start-1, start, end),
+		},
+		{
+			name:           "LedgerCloseMetaNotFound",
+			ledgerSeq:      end - 5,
+			expectedErrMsg: fmt.Sprintf("LedgerCloseMeta for sequence %d not found in the batch", end-5),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			archive, err := f.GetLedger(tc.ledgerSeq)
+			if tc.expectedErrMsg != "" {
+				require.EqualError(t, err, tc.expectedErrMsg)
+				require.Equal(t, archive, LedgerCloseMeta{})
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, archive.LedgerSequence(), tc.ledgerSeq)
+			}
+		})
+	}
+}