Repository: stellar/go
Pull request #5324 (2024-05-28 23:27:41)

exp/services/ledgerexporter: Add metadata to exported files
<!-- 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

The following metadata added to the exported files in GCS:

```
	StartLedger          uint32
	EndLedger            uint32
	StartLedgerCloseTime int64
	EndLedgerCloseTime   int64
	ProtocolVersion      uint32
	CoreVersion          string
	Network              string
	CompressionType      string
	Version              string
```

- `CoreVersion` and `ProtocolVersion` are parsed from the output of the `stellar-core version` command.
- Build process does not include versioning yet, so lexie `Version` will default to "develop". Versioning of Lexie is out of scope for this PR.
- `Network` will be either "pubnet" or "testnet".
- `CompressionType` is the name of the compression algorithm used which also used as the file extension.

Datatstore interface changes:
- Put interface modified to accept a map of strings as input which holds the metadata.
- Get interface returns metadata as a map of strings. The usage of this metadata is not yet defined so the ledgerbackend currently ignores it.

Field names:
```
start-ledger
end-ledger
start-ledger-close-time
end-ledger-close-time
protocol-version
core-version
network
compression-type
version
```

### Why

Addresses [HUBBLE-279](https://stellarorg.atlassian.net/browse/HUBBLE-279)

### Known limitations


[HUBBLE-279]: https://stellarorg.atlassian.net/browse/HUBBLE-279?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ


Patch:
diff --git a/exp/services/ledgerexporter/Makefile b/exp/services/ledgerexporter/Makefile
index 5fee16f9..10bf16e9 100644
--- a/exp/services/ledgerexporter/Makefile
+++ b/exp/services/ledgerexporter/Makefile
@@ -2,13 +2,13 @@ SUDO := $(shell docker version >/dev/null 2>&1 || echo "sudo")
 
 # https://github.com/opencontainers/image-spec/blob/master/annotations.md
 BUILD_DATE := $(shell date -u +%FT%TZ)
-VERSION ?= $(shell git rev-parse --short HEAD)
+VERSION ?= 1.0.0-$(shell git rev-parse --short HEAD)
 DOCKER_IMAGE := stellar/ledger-exporter
 
 docker-build:
 	cd ../../../ && \
 	$(SUDO) docker build --platform linux/amd64 --pull --label org.opencontainers.image.created="$(BUILD_DATE)" \
-	--build-arg VERSION=$(VERSION) \
+    --build-arg GOFLAGS="-ldflags=-X=github.com/stellar/go/exp/services/ledgerexporter/internal.version=$(VERSION)" \
 $(if $(STELLAR_CORE_VERSION), --build-arg STELLAR_CORE_VERSION=$(STELLAR_CORE_VERSION)) \
 	-f exp/services/ledgerexporter/docker/Dockerfile \
 	-t $(DOCKER_IMAGE):$(VERSION) \
diff --git a/exp/services/ledgerexporter/docker/Dockerfile b/exp/services/ledgerexporter/docker/Dockerfile
index a8623067..59e57030 100644
--- a/exp/services/ledgerexporter/docker/Dockerfile
+++ b/exp/services/ledgerexporter/docker/Dockerfile
@@ -9,6 +9,7 @@ RUN go mod download
 
 COPY . ./
 
+ARG GOFLAGS
 RUN go install github.com/stellar/go/exp/services/ledgerexporter
 
 FROM ubuntu:22.04
diff --git a/exp/services/ledgerexporter/internal/app.go b/exp/services/ledgerexporter/internal/app.go
index e6dde872..c6adcd41 100644
--- a/exp/services/ledgerexporter/internal/app.go
+++ b/exp/services/ledgerexporter/internal/app.go
@@ -37,7 +37,8 @@ const (
 )
 
 var (
-	logger = log.New().WithField("service", "ledger-exporter")
+	logger  = log.New().WithField("service", "ledger-exporter")
+	version = "develop"
 )
 
 func NewDataAlreadyExportedError(Start uint32, End uint32) *DataAlreadyExportedError {
@@ -95,13 +96,15 @@ func (a *App) init(ctx context.Context) error {
 	var err error
 	var archive historyarchive.ArchiveInterface
 
+	logger.Infof("Starting Ledger Exporter with version %s", version)
+
 	registry := prometheus.NewRegistry()
 	registry.MustRegister(
 		collectors.NewProcessCollector(collectors.ProcessCollectorOpts{Namespace: "ledger_exporter"}),
 		collectors.NewGoCollector(),
 	)
 
-	if a.config, err = NewConfig(ctx, a.flags); err != nil {
+	if a.config, err = NewConfig(a.flags); err != nil {
 		return errors.Wrap(err, "Could not load configuration")
 	}
 	if archive, err = datastore.CreateHistoryArchiveFromNetworkName(ctx, a.config.Network); err != nil {
@@ -126,7 +129,7 @@ func (a *App) init(ctx context.Context) error {
 	}
 
 	queue := NewUploadQueue(uploadQueueCapacity, registry)
-	if a.exportManager, err = NewExportManager(a.config.LedgerBatchConfig, a.ledgerBackend, queue, registry); err != nil {
+	if a.exportManager, err = NewExportManager(a.config, a.ledgerBackend, queue, registry); err != nil {
 		return err
 	}
 	a.uploader = NewUploader(a.dataStore, queue, registry)
@@ -256,7 +259,7 @@ func (a *App) Run() {
 // newLedgerBackend Creates and initializes captive core ledger backend
 // Currently, only supports captive-core as ledger backend
 func newLedgerBackend(config *Config, prometheusRegistry *prometheus.Registry) (ledgerbackend.LedgerBackend, error) {
-	captiveConfig, err := config.GenerateCaptiveCoreConfig()
+	captiveConfig, err := config.generateCaptiveCoreConfig()
 	if err != nil {
 		return nil, err
 	}
diff --git a/exp/services/ledgerexporter/internal/config.go b/exp/services/ledgerexporter/internal/config.go
index bab184bd..ef062867 100644
--- a/exp/services/ledgerexporter/internal/config.go
+++ b/exp/services/ledgerexporter/internal/config.go
@@ -3,7 +3,9 @@ package ledgerexporter
 import (
 	"context"
 	_ "embed"
+	"fmt"
 	"os/exec"
+	"strings"
 
 	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/go/ingest/ledgerbackend"
@@ -45,15 +47,16 @@ type Config struct {
 	StartLedger uint32
 	EndLedger   uint32
 	Resume      bool
+
+	CoreVersion string
 }
 
 // This will generate the config based on commandline flags and toml
 //
-// ctx                   - the caller context
 // flags                 - command line flags
 //
 // return                - *Config or an error if any range validation failed.
-func NewConfig(ctx context.Context, flags Flags) (*Config, error) {
+func NewConfig(flags Flags) (*Config, error) {
 	config := &Config{}
 
 	config.StartLedger = uint32(flags.StartLedger)
@@ -100,17 +103,21 @@ func (config *Config) ValidateAndSetLedgerRange(ctx context.Context, archive his
 	return nil
 }
 
-func (config *Config) GenerateCaptiveCoreConfig() (ledgerbackend.CaptiveCoreConfig, error) {
+func (config *Config) generateCaptiveCoreConfig() (ledgerbackend.CaptiveCoreConfig, error) {
 	coreConfig := &config.StellarCoreConfig
 
 	// Look for stellar-core binary in $PATH, if not supplied
-	if coreConfig.StellarCoreBinaryPath == "" {
+	if config.StellarCoreConfig.StellarCoreBinaryPath == "" {
 		var err error
-		if coreConfig.StellarCoreBinaryPath, err = exec.LookPath("stellar-core"); err != nil {
+		if config.StellarCoreConfig.StellarCoreBinaryPath, err = exec.LookPath("stellar-core"); err != nil {
 			return ledgerbackend.CaptiveCoreConfig{}, errors.Wrap(err, "Failed to find stellar-core binary")
 		}
 	}
 
+	if err := config.setCoreVersionInfo(); err != nil {
+		return ledgerbackend.CaptiveCoreConfig{}, fmt.Errorf("failed to set stellar-core version info: %w", err)
+	}
+
 	var captiveCoreConfig []byte
 	// Default network config
 	switch config.Network {
@@ -151,6 +158,30 @@ func (config *Config) GenerateCaptiveCoreConfig() (ledgerbackend.CaptiveCoreConf
 	}, nil
 }
 
+// By default, it points to exec.Command, overridden for testing purpose
+var execCommand = exec.Command
+
+// Executes the "stellar-core version" command and parses its output to extract
+// the core version
+// The output of the "version" command is expected to be a multi-line string where the
+// first line is the core version in format "vX.Y.Z-*".
+func (c *Config) setCoreVersionInfo() (err error) {
+	versionCmd := execCommand(c.StellarCoreConfig.StellarCoreBinaryPath, "version")
+	versionOutput, err := versionCmd.Output()
+	if err != nil {
+		return fmt.Errorf("failed to execute stellar-core version command: %w", err)
+	}
+
+	// Split the output into lines
+	rows := strings.Split(string(versionOutput), "\n")
+	if len(rows) == 0 || len(rows[0]) == 0 {
+		return fmt.Errorf("stellar-core version not found")
+	}
+	c.CoreVersion = rows[0]
+	logger.Infof("stellar-core version: %s", c.CoreVersion)
+	return nil
+}
+
 func (config *Config) processToml(tomlPath string) error {
 	// Load config TOML file
 	cfg, err := toml.LoadFile(tomlPath)
diff --git a/exp/services/ledgerexporter/internal/exportmanager.go b/exp/services/ledgerexporter/internal/exportmanager.go
index 55f85b9c..ae85c072 100644
--- a/exp/services/ledgerexporter/internal/exportmanager.go
+++ b/exp/services/ledgerexporter/internal/exportmanager.go
@@ -8,22 +8,21 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/stellar/go/ingest/ledgerbackend"
-	"github.com/stellar/go/support/datastore"
 	"github.com/stellar/go/xdr"
 )
 
 type ExportManager struct {
-	config             datastore.LedgerBatchConfig
+	config             *Config
 	ledgerBackend      ledgerbackend.LedgerBackend
-	currentMetaArchive *LedgerMetaArchive
+	currentMetaArchive *xdr.LedgerCloseMetaBatch
 	queue              UploadQueue
 	latestLedgerMetric *prometheus.GaugeVec
 }
 
 // NewExportManager creates a new ExportManager with the provided configuration.
-func NewExportManager(config datastore.LedgerBatchConfig, backend ledgerbackend.LedgerBackend, queue UploadQueue, prometheusRegistry *prometheus.Registry) (*ExportManager, error) {
-	if config.LedgersPerFile < 1 {
-		return nil, errors.Errorf("Invalid ledgers per file (%d): must be at least 1", config.LedgersPerFile)
+func NewExportManager(config *Config, backend ledgerbackend.LedgerBackend, queue UploadQueue, prometheusRegistry *prometheus.Registry) (*ExportManager, error) {
+	if config.LedgerBatchConfig.LedgersPerFile < 1 {
+		return nil, errors.Errorf("Invalid ledgers per file (%d): must be at least 1", config.LedgerBatchConfig.LedgersPerFile)
 	}
 
 	latestLedgerMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -45,32 +44,32 @@ func (e *ExportManager) AddLedgerCloseMeta(ctx context.Context, ledgerCloseMeta
 	ledgerSeq := ledgerCloseMeta.LedgerSequence()
 
 	// Determine the object key for the given ledger sequence
-	objectKey := e.config.GetObjectKeyFromSequenceNumber(ledgerSeq)
+	objectKey := e.config.LedgerBatchConfig.GetObjectKeyFromSequenceNumber(ledgerSeq)
 
-	if e.currentMetaArchive != nil && e.currentMetaArchive.ObjectKey != objectKey {
-		return errors.New("Current meta archive object key mismatch")
-	}
 	if e.currentMetaArchive == nil {
-		endSeq := ledgerSeq + e.config.LedgersPerFile - 1
-		if ledgerSeq < e.config.LedgersPerFile {
+		endSeq := ledgerSeq + e.config.LedgerBatchConfig.LedgersPerFile - 1
+		if ledgerSeq < e.config.LedgerBatchConfig.LedgersPerFile {
 			// Special case: Adjust the end ledger sequence for the first batch.
 			// Since the start ledger is 2 instead of 0, we want to ensure that the end ledger sequence
 			// does not exceed LedgersPerFile.
 			// For example, if LedgersPerFile is 64, the file name for the first batch should be 0-63, not 2-66.
-			endSeq = e.config.LedgersPerFile - 1
+			endSeq = e.config.LedgerBatchConfig.LedgersPerFile - 1
 		}
 
-		// Create a new LedgerMetaArchive and add it to the map.
-		e.currentMetaArchive = NewLedgerMetaArchive(objectKey, ledgerSeq, endSeq)
+		// Create a new LedgerCloseMetaBatch
+		e.currentMetaArchive = &xdr.LedgerCloseMetaBatch{StartSequence: xdr.Uint32(ledgerSeq), EndSequence: xdr.Uint32(endSeq)}
 	}
 
-	if err := e.currentMetaArchive.Data.AddLedger(ledgerCloseMeta); err != nil {
+	if err := e.currentMetaArchive.AddLedger(ledgerCloseMeta); err != nil {
 		return errors.Wrapf(err, "failed to add ledger %d", ledgerSeq)
 	}
 
-	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 {
+	if ledgerSeq >= uint32(e.currentMetaArchive.EndSequence) {
+		ledgerMetaArchive, err := NewLedgerMetaArchiveFromXDR(e.config, objectKey, *e.currentMetaArchive)
+		if err != nil {
+			return err
+		}
+		if err := e.queue.Enqueue(ctx, ledgerMetaArchive); err != nil {
 			return err
 		}
 		e.currentMetaArchive = nil
diff --git a/exp/services/ledgerexporter/internal/ledger_meta_archive.go b/exp/services/ledgerexporter/internal/ledger_meta_archive.go
index 5215e235..583f6a3e 100644
--- a/exp/services/ledgerexporter/internal/ledger_meta_archive.go
+++ b/exp/services/ledgerexporter/internal/ledger_meta_archive.go
@@ -1,24 +1,43 @@
 package ledgerexporter
 
 import (
+	"github.com/stellar/go/support/compressxdr"
+	"github.com/stellar/go/support/datastore"
 	"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
+	Data      xdr.LedgerCloseMetaBatch
+	metaData  datastore.MetaData
 }
 
-// NewLedgerMetaArchive creates a new LedgerMetaArchive instance.
-func NewLedgerMetaArchive(key string, startSeq uint32, endSeq uint32) *LedgerMetaArchive {
+// NewLedgerMetaArchiveFromXDR creates a new LedgerMetaArchive instance.
+func NewLedgerMetaArchiveFromXDR(config *Config, key string, data xdr.LedgerCloseMetaBatch) (*LedgerMetaArchive, error) {
+	startLedger, err := data.GetLedger(uint32(data.StartSequence))
+	if err != nil {
+		return &LedgerMetaArchive{}, err
+
+	}
+	endLedger, err := data.GetLedger(uint32(data.EndSequence))
+	if err != nil {
+		return &LedgerMetaArchive{}, err
+	}
+
 	return &LedgerMetaArchive{
 		ObjectKey: key,
-		Data: xdr.LedgerCloseMetaBatch{
-			StartSequence: xdr.Uint32(startSeq),
-			EndSequence:   xdr.Uint32(endSeq),
+		Data:      data,
+		metaData: datastore.MetaData{
+			StartLedger:          startLedger.LedgerSequence(),
+			EndLedger:            endLedger.LedgerSequence(),
+			StartLedgerCloseTime: startLedger.LedgerCloseTime(),
+			EndLedgerCloseTime:   endLedger.LedgerCloseTime(),
+			Network:              config.Network,
+			CompressionType:      compressxdr.DefaultCompressor.Name(),
+			ProtocolVersion:      endLedger.ProtocolVersion(),
+			CoreVersion:          config.CoreVersion,
+			Version:              version,
 		},
-	}
+	}, nil
 }
diff --git a/exp/services/ledgerexporter/internal/uploader.go b/exp/services/ledgerexporter/internal/uploader.go
index 78296cde..6d35d292 100644
--- a/exp/services/ledgerexporter/internal/uploader.go
+++ b/exp/services/ledgerexporter/internal/uploader.go
@@ -95,7 +95,7 @@ func (u Uploader) Upload(ctx context.Context, metaArchive *LedgerMetaArchive) er
 	writerTo := &writerToRecorder{
 		WriterTo: xdrEncoder,
 	}
-	ok, err := u.dataStore.PutFileIfNotExists(ctx, metaArchive.ObjectKey, writerTo)
+	ok, err := u.dataStore.PutFileIfNotExists(ctx, metaArchive.ObjectKey, writerTo, metaArchive.metaData.ToMap())
 	if err != nil {
 		return errors.Wrapf(err, "error uploading %s", metaArchive.ObjectKey)
 	}
diff --git a/support/datastore/datastore.go b/support/datastore/datastore.go
index 5592edf6..0a92857b 100644
--- a/support/datastore/datastore.go
+++ b/support/datastore/datastore.go
@@ -14,9 +14,10 @@ type DataStoreConfig struct {
 
 // DataStore defines an interface for interacting with data storage
 type DataStore interface {
+	GetFileMetadata(ctx context.Context, path string) (map[string]string, error)
 	GetFile(ctx context.Context, path string) (io.ReadCloser, error)
-	PutFile(ctx context.Context, path string, in io.WriterTo) error
-	PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo) (bool, error)
+	PutFile(ctx context.Context, path string, in io.WriterTo, metaData map[string]string) error
+	PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo, metaData map[string]string) (bool, error)
 	Exists(ctx context.Context, path string) (bool, error)
 	Size(ctx context.Context, path string) (int64, error)
 	Close() error
diff --git a/support/datastore/gcs_datastore.go b/support/datastore/gcs_datastore.go
index aaf7a55a..c6b22b0f 100644
--- a/support/datastore/gcs_datastore.go
+++ b/support/datastore/gcs_datastore.go
@@ -58,6 +58,18 @@ func FromGCSClient(ctx context.Context, client *storage.Client, bucketPath strin
 	return &GCSDataStore{client: client, bucket: bucket, prefix: prefix}, nil
 }
 
+// GetFileMetadata retrieves the metadata for the specified file in the GCS bucket.
+func (b GCSDataStore) GetFileMetadata(ctx context.Context, filePath string) (map[string]string, error) {
+	filePath = path.Join(b.prefix, filePath)
+	attrs, err := b.bucket.Object(filePath).Attrs(ctx)
+	if err != nil {
+		if errors.Is(err, storage.ErrObjectNotExist) {
+			return nil, os.ErrNotExist
+		}
+	}
+	return attrs.Metadata, nil
+}
+
 // GetFile retrieves a file from the GCS bucket.
 func (b GCSDataStore) GetFile(ctx context.Context, filePath string) (io.ReadCloser, error) {
 	filePath = path.Join(b.prefix, filePath)
@@ -82,8 +94,8 @@ func (b GCSDataStore) GetFile(ctx context.Context, filePath string) (io.ReadClos
 }
 
 // PutFileIfNotExists uploads a file to GCS only if it doesn't already exist.
-func (b GCSDataStore) PutFileIfNotExists(ctx context.Context, filePath string, in io.WriterTo) (bool, error) {
-	err := b.putFile(ctx, filePath, in, &storage.Conditions{DoesNotExist: true})
+func (b GCSDataStore) PutFileIfNotExists(ctx context.Context, filePath string, in io.WriterTo, metaData map[string]string) (bool, error) {
+	err := b.putFile(ctx, filePath, in, &storage.Conditions{DoesNotExist: true}, metaData)
 	if err != nil {
 		if gcsError, ok := err.(*googleapi.Error); ok {
 			switch gcsError.Code {
@@ -101,8 +113,8 @@ func (b GCSDataStore) PutFileIfNotExists(ctx context.Context, filePath string, i
 }
 
 // PutFile uploads a file to GCS
-func (b GCSDataStore) PutFile(ctx context.Context, filePath string, in io.WriterTo) error {
-	err := b.putFile(ctx, filePath, in, nil) // No conditions for regular PutFile
+func (b GCSDataStore) PutFile(ctx context.Context, filePath string, in io.WriterTo, metaData map[string]string) error {
+	err := b.putFile(ctx, filePath, in, nil, metaData) // No conditions for regular PutFile
 
 	if err != nil {
 		if gcsError, ok := err.(*googleapi.Error); ok {
@@ -143,7 +155,7 @@ func (b GCSDataStore) Close() error {
 	return b.client.Close()
 }
 
-func (b GCSDataStore) putFile(ctx context.Context, filePath string, in io.WriterTo, conditions *storage.Conditions) error {
+func (b GCSDataStore) putFile(ctx context.Context, filePath string, in io.WriterTo, conditions *storage.Conditions, metaData map[string]string) error {
 	filePath = path.Join(b.prefix, filePath)
 	o := b.bucket.Object(filePath)
 	if conditions != nil {
@@ -155,6 +167,7 @@ func (b GCSDataStore) putFile(ctx context.Context, filePath string, in io.Writer
 	}
 
 	w := o.NewWriter(ctx)
+	w.Metadata = metaData
 	w.SendCRC32C = true
 	// we must set CRC32C before invoking w.Write() for the first time
 	w.CRC32C = crc32.Checksum(buf.Bytes(), crc32.MakeTable(crc32.Castagnoli))
diff --git a/support/datastore/metadata.go b/support/datastore/metadata.go
new file mode 100644
index 00000000..db734bbd
--- /dev/null
+++ b/support/datastore/metadata.go
@@ -0,0 +1,79 @@
+package datastore
+
+import "strconv"
+
+type MetaData struct {
+	StartLedger          uint32
+	EndLedger            uint32
+	StartLedgerCloseTime int64
+	EndLedgerCloseTime   int64
+	ProtocolVersion      uint32
+	CoreVersion          string
+	Network              string
+	CompressionType      string
+	Version              string
+}
+
+func (m MetaData) ToMap() map[string]string {
+	return map[string]string{
+		"start-ledger":            strconv.FormatUint(uint64(m.StartLedger), 10),
+		"end-ledger":              strconv.FormatUint(uint64(m.EndLedger), 10),
+		"start-ledger-close-time": strconv.FormatInt(m.StartLedgerCloseTime, 10),
+		"end-ledger-close-time":   strconv.FormatInt(m.EndLedgerCloseTime, 10),
+		"protocol-version":        strconv.FormatInt(int64(m.ProtocolVersion), 10),
+		"core-version":            m.CoreVersion,
+		"network":                 m.Network,
+		"compression-type":        m.CompressionType,
+		"version":                 m.Version,
+	}
+}
+func NewMetaDataFromMap(data map[string]string) (MetaData, error) {
+	var metaData MetaData
+
+	if val, ok := data["start-ledger"]; ok {
+		startLedger, err := strconv.ParseUint(val, 10, 32)
+		if err != nil {
+			return metaData, err
+		}
+		metaData.StartLedger = uint32(startLedger)
+	}
+
+	if val, ok := data["end-ledger"]; ok {
+		endLedger, err := strconv.ParseUint(val, 10, 32)
+		if err != nil {
+			return metaData, err
+		}
+		metaData.EndLedger = uint32(endLedger)
+	}
+
+	if val, ok := data["start-ledger-close-time"]; ok {
+		startLedgerCloseTime, err := strconv.ParseInt(val, 10, 64)
+		if err != nil {
+			return metaData, err
+		}
+		metaData.StartLedgerCloseTime = startLedgerCloseTime
+	}
+
+	if val, ok := data["end-ledger-close-time"]; ok {
+		endLedgerCloseTime, err := strconv.ParseInt(val, 10, 64)
+		if err != nil {
+			return metaData, err
+		}
+		metaData.EndLedgerCloseTime = endLedgerCloseTime
+	}
+
+	if val, ok := data["protocol-version"]; ok {
+		protocolVersion, err := strconv.ParseInt(val, 10, 32)
+		if err != nil {
+			return metaData, err
+		}
+		metaData.ProtocolVersion = uint32(protocolVersion)
+	}
+
+	metaData.CoreVersion = data["core-version"]
+	metaData.Network = data["network"]
+	metaData.CompressionType = data["compression-type"]
+	metaData.Version = data["version"]
+
+	return metaData, nil
+}
diff --git a/support/datastore/mocks.go b/support/datastore/mocks.go
index b77040d2..96c15c13 100644
--- a/support/datastore/mocks.go
+++ b/support/datastore/mocks.go
@@ -22,18 +22,23 @@ func (m *MockDataStore) Size(ctx context.Context, path string) (int64, error) {
 	return args.Get(0).(int64), args.Error(1)
 }
 
+func (m *MockDataStore) GetFileMetadata(ctx context.Context, path string) (map[string]string, error) {
+	args := m.Called(ctx, path)
+	return args.Get(0).(map[string]string), args.Error(1)
+}
+
 func (m *MockDataStore) GetFile(ctx context.Context, path string) (io.ReadCloser, error) {
 	args := m.Called(ctx, path)
 	return args.Get(0).(io.ReadCloser), args.Error(1)
 }
 
-func (m *MockDataStore) PutFile(ctx context.Context, path string, in io.WriterTo) error {
-	args := m.Called(ctx, path, in)
+func (m *MockDataStore) PutFile(ctx context.Context, path string, in io.WriterTo, metadata map[string]string) error {
+	args := m.Called(ctx, path, in, metadata)
 	return args.Error(0)
 }
 
-func (m *MockDataStore) PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo) (bool, error) {
-	args := m.Called(ctx, path, in)
+func (m *MockDataStore) PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo, metadata map[string]string) (bool, error) {
+	args := m.Called(ctx, path, in, metadata)
 	return args.Get(0).(bool), args.Error(1)
 }
 


Test patch:
diff --git a/exp/services/ledgerexporter/internal/config_test.go b/exp/services/ledgerexporter/internal/config_test.go
index 13316507..a4ce76cf 100644
--- a/exp/services/ledgerexporter/internal/config_test.go
+++ b/exp/services/ledgerexporter/internal/config_test.go
@@ -3,11 +3,14 @@ package ledgerexporter
 import (
 	"context"
 	"fmt"
+	"os"
+	"os/exec"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
 	"github.com/stellar/go/historyarchive"
+	"github.com/stellar/go/support/errors"
 )
 
 func TestNewConfigResumeEnabled(t *testing.T) {
@@ -16,8 +19,7 @@ func TestNewConfigResumeEnabled(t *testing.T) {
 	mockArchive := &historyarchive.MockArchive{}
 	mockArchive.On("GetRootHAS").Return(historyarchive.HistoryArchiveState{CurrentLedger: 5}, nil).Once()
 
-	config, err := NewConfig(ctx,
-		Flags{StartLedger: 1, EndLedger: 2, ConfigFilePath: "test/test.toml", Resume: true})
+	config, err := NewConfig(Flags{StartLedger: 1, EndLedger: 2, ConfigFilePath: "test/test.toml", Resume: true})
 	config.ValidateAndSetLedgerRange(ctx, mockArchive)
 	require.NoError(t, err)
 	require.Equal(t, config.DataStoreConfig.Type, "ABC")
@@ -30,23 +32,19 @@ func TestNewConfigResumeEnabled(t *testing.T) {
 }
 
 func TestNewConfigResumeDisabled(t *testing.T) {
-	ctx := context.Background()
 
 	mockArchive := &historyarchive.MockArchive{}
 	mockArchive.On("GetRootHAS").Return(historyarchive.HistoryArchiveState{CurrentLedger: 5}, nil).Once()
 
 	// resume disabled by default
-	config, err := NewConfig(ctx,
-		Flags{StartLedger: 1, EndLedger: 2, ConfigFilePath: "test/test.toml"})
+	config, err := NewConfig(Flags{StartLedger: 1, EndLedger: 2, ConfigFilePath: "test/test.toml"})
 	require.NoError(t, err)
 	require.False(t, config.Resume)
 }
 
 func TestInvalidTomlConfig(t *testing.T) {
-	ctx := context.Background()
 
-	_, err := NewConfig(ctx,
-		Flags{StartLedger: 1, EndLedger: 2, ConfigFilePath: "test/no_network.toml", Resume: true})
+	_, err := NewConfig(Flags{StartLedger: 1, EndLedger: 2, ConfigFilePath: "test/no_network.toml", Resume: true})
 	require.ErrorContains(t, err, "Invalid TOML config")
 }
 
@@ -111,8 +109,7 @@ func TestValidateStartAndEndLedger(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			config, err := NewConfig(ctx,
-				Flags{StartLedger: tt.startLedger, EndLedger: tt.endLedger, ConfigFilePath: "test/validate_start_end.toml"})
+			config, err := NewConfig(Flags{StartLedger: tt.startLedger, EndLedger: tt.endLedger, ConfigFilePath: "test/validate_start_end.toml"})
 			require.NoError(t, err)
 			err = config.ValidateAndSetLedgerRange(ctx, mockArchive)
 			if tt.errMsg != "" {
@@ -189,8 +186,7 @@ func TestAdjustedLedgerRangeBoundedMode(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			config, err := NewConfig(ctx,
-				Flags{StartLedger: tt.start, EndLedger: tt.end, ConfigFilePath: tt.configFile})
+			config, err := NewConfig(Flags{StartLedger: tt.start, EndLedger: tt.end, ConfigFilePath: tt.configFile})
 			require.NoError(t, err)
 			err = config.ValidateAndSetLedgerRange(ctx, mockArchive)
 			require.NoError(t, err)
@@ -258,8 +254,7 @@ func TestAdjustedLedgerRangeUnBoundedMode(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			config, err := NewConfig(ctx,
-				Flags{StartLedger: tt.start, EndLedger: tt.end, ConfigFilePath: tt.configFile})
+			config, err := NewConfig(Flags{StartLedger: tt.start, EndLedger: tt.end, ConfigFilePath: tt.configFile})
 			require.NoError(t, err)
 			err = config.ValidateAndSetLedgerRange(ctx, mockArchive)
 			require.NoError(t, err)
@@ -268,3 +263,69 @@ func TestAdjustedLedgerRangeUnBoundedMode(t *testing.T) {
 		})
 	}
 }
+
+var cmdOut = ""
+
+func fakeExecCommand(command string, args ...string) *exec.Cmd {
+	cs := append([]string{"-test.run=TestExecCmdHelperProcess", "--", command}, args...)
+	cmd := exec.Command(os.Args[0], cs...)
+	cmd.Env = append(os.Environ(), "GO_EXEC_CMD_HELPER_PROCESS=1", "CMD_OUT="+cmdOut)
+	return cmd
+}
+
+func TestExecCmdHelperProcess(t *testing.T) {
+	if os.Getenv("GO_EXEC_CMD_HELPER_PROCESS") != "1" {
+		return
+	}
+	fmt.Fprintf(os.Stdout, os.Getenv("CMD_OUT"))
+	os.Exit(0)
+}
+
+func TestSetCoreVersionInfo(t *testing.T) {
+	tests := []struct {
+		name            string
+		commandOutput   string
+		expectedError   error
+		expectedCoreVer string
+	}{
+		{
+			name: "version found",
+			commandOutput: "v20.2.0-2-g6e73c0a88\n" +
+				"rust version: rustc 1.74.1 (a28077b28 2023-12-04)\n" +
+				"soroban-env-host: \n" +
+				"    curr:\n" +
+				"       package version: 20.2.0\n" +
+				"       git version: 1bfc0f2a2ee134efc1e1b0d5270281d0cba61c2e\n" +
+				"       ledger protocol version: 20\n" +
+				"       pre-release version: 0\n" +
+				"       rs-stellar-xdr:\n" +
+				"           package version: 20.1.0\n" +
+				"           git version: 8b9d623ef40423a8462442b86997155f2c04d3a1\n" +
+				"           base XDR git version: b96148cd4acc372cc9af17b909ffe4b12c43ecb6\n",
+			expectedError:   nil,
+			expectedCoreVer: "v20.2.0-2-g6e73c0a88",
+		},
+		{
+			name:            "core version not found",
+			commandOutput:   "",
+			expectedError:   errors.New("stellar-core version not found"),
+			expectedCoreVer: "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			config := Config{}
+
+			cmdOut = tt.commandOutput
+			execCommand = fakeExecCommand
+			err := config.setCoreVersionInfo()
+
+			if tt.expectedError != nil {
+				require.EqualError(t, err, tt.expectedError.Error())
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tt.expectedCoreVer, config.CoreVersion)
+			}
+		})
+	}
+}
diff --git a/exp/services/ledgerexporter/internal/exportmanager_test.go b/exp/services/ledgerexporter/internal/exportmanager_test.go
index d99f88cd..b74af2e1 100644
--- a/exp/services/ledgerexporter/internal/exportmanager_test.go
+++ b/exp/services/ledgerexporter/internal/exportmanager_test.go
@@ -23,7 +23,9 @@ func createLedgerCloseMeta(ledgerSeq uint32) xdr.LedgerCloseMeta {
 		V0: &xdr.LedgerCloseMetaV0{
 			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
 				Header: xdr.LedgerHeader{
-					LedgerSeq: xdr.Uint32(ledgerSeq),
+					LedgerVersion: 21,
+					LedgerSeq:     xdr.Uint32(ledgerSeq),
+					ScpValue:      xdr.StellarValue{CloseTime: xdr.TimePoint(ledgerSeq * 100)},
 				},
 			},
 			TxSet:              xdr.TransactionSet{},
@@ -56,7 +58,7 @@ func (s *ExportManagerSuite) TearDownTest() {
 }
 
 func (s *ExportManagerSuite) TestInvalidExportConfig() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 0, FilesPerPartition: 10}
+	config := &Config{LedgerBatchConfig: datastore.LedgerBatchConfig{LedgersPerFile: 0, FilesPerPartition: 10}}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	_, err := NewExportManager(config, &s.mockBackend, queue, registry)
@@ -64,7 +66,7 @@ func (s *ExportManagerSuite) TestInvalidExportConfig() {
 }
 
 func (s *ExportManagerSuite) TestRun() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 64, FilesPerPartition: 10}
+	config := &Config{LedgerBatchConfig: datastore.LedgerBatchConfig{LedgersPerFile: 64, FilesPerPartition: 10}}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
@@ -77,7 +79,7 @@ func (s *ExportManagerSuite) TestRun() {
 	for i := start; i <= end; i++ {
 		s.mockBackend.On("GetLedger", s.ctx, i).
 			Return(createLedgerCloseMeta(i), nil)
-		key := config.GetObjectKeyFromSequenceNumber(i)
+		key := config.LedgerBatchConfig.GetObjectKeyFromSequenceNumber(i)
 		expectedKeys.Add(key)
 	}
 
@@ -114,7 +116,7 @@ func (s *ExportManagerSuite) TestRun() {
 }
 
 func (s *ExportManagerSuite) TestRunContextCancel() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 1}
+	config := &Config{LedgerBatchConfig: datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 1}}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
@@ -144,7 +146,7 @@ func (s *ExportManagerSuite) TestRunContextCancel() {
 }
 
 func (s *ExportManagerSuite) TestRunWithCanceledContext() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}
+	config := &Config{LedgerBatchConfig: datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
@@ -163,7 +165,7 @@ func (s *ExportManagerSuite) TestRunWithCanceledContext() {
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}
+	config := &Config{LedgerBatchConfig: datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
@@ -191,7 +193,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 	for i := start; i <= end; i++ {
 		s.Require().NoError(exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(i)))
 
-		key := config.GetObjectKeyFromSequenceNumber(i)
+		key := config.LedgerBatchConfig.GetObjectKeyFromSequenceNumber(i)
 		expectedKeys.Add(key)
 	}
 
@@ -201,7 +203,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}
+	config := &Config{LedgerBatchConfig: datastore.LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
@@ -217,15 +219,3 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
 	err = exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(2))
 	s.Require().EqualError(err, "context canceled")
 }
-
-func (s *ExportManagerSuite) TestAddLedgerCloseMetaKeyMismatch() {
-	config := datastore.LedgerBatchConfig{LedgersPerFile: 10, FilesPerPartition: 1}
-	registry := prometheus.NewRegistry()
-	queue := NewUploadQueue(1, registry)
-	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
-	s.Require().NoError(err)
-
-	s.Require().NoError(exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(16)))
-	s.Require().EqualError(exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(21)),
-		"Current meta archive object key mismatch")
-}
diff --git a/exp/services/ledgerexporter/internal/ledger_meta_archive_test.go b/exp/services/ledgerexporter/internal/ledger_meta_archive_test.go
new file mode 100644
index 00000000..3cf263d7
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/ledger_meta_archive_test.go
@@ -0,0 +1,75 @@
+package ledgerexporter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/stellar/go/support/datastore"
+	"github.com/stellar/go/xdr"
+)
+
+func TestNewLedgerMetaArchiveFromXDR(t *testing.T) {
+	config := &Config{
+		Network:     "testnet",
+		CoreVersion: "v1.2.3",
+	}
+	data := xdr.LedgerCloseMetaBatch{
+		StartSequence: 1234,
+		EndSequence:   1234,
+		LedgerCloseMetas: []xdr.LedgerCloseMeta{
+			createLedgerCloseMeta(1234),
+		},
+	}
+
+	archive, err := NewLedgerMetaArchiveFromXDR(config, "key", data)
+
+	require.NoError(t, err)
+	require.NotNil(t, archive)
+
+	// Check if the metadata fields are correctly populated
+	expectedMetaData := datastore.MetaData{
+		StartLedger:          1234,
+		EndLedger:            1234,
+		StartLedgerCloseTime: 1234 * 100,
+		EndLedgerCloseTime:   1234 * 100,
+		Network:              "testnet",
+		CompressionType:      "zstd",
+		ProtocolVersion:      21,
+		CoreVersion:          "v1.2.3",
+		Version:              "develop",
+	}
+
+	require.Equal(t, expectedMetaData, archive.metaData)
+
+	data = xdr.LedgerCloseMetaBatch{
+		StartSequence: 1234,
+		EndSequence:   1237,
+		LedgerCloseMetas: []xdr.LedgerCloseMeta{
+			createLedgerCloseMeta(1234),
+			createLedgerCloseMeta(1235),
+			createLedgerCloseMeta(1236),
+			createLedgerCloseMeta(1237),
+		},
+	}
+
+	archive, err = NewLedgerMetaArchiveFromXDR(config, "key", data)
+
+	require.NoError(t, err)
+	require.NotNil(t, archive)
+
+	// Check if the metadata fields are correctly populated
+	expectedMetaData = datastore.MetaData{
+		StartLedger:          1234,
+		EndLedger:            1237,
+		StartLedgerCloseTime: 1234 * 100,
+		EndLedgerCloseTime:   1237 * 100,
+		Network:              "testnet",
+		CompressionType:      "zstd",
+		ProtocolVersion:      21,
+		CoreVersion:          "v1.2.3",
+		Version:              "develop",
+	}
+
+	require.Equal(t, expectedMetaData, archive.metaData)
+}
diff --git a/exp/services/ledgerexporter/internal/uploader_test.go b/exp/services/ledgerexporter/internal/uploader_test.go
index 5fbf5e22..bb3ca1e2 100644
--- a/exp/services/ledgerexporter/internal/uploader_test.go
+++ b/exp/services/ledgerexporter/internal/uploader_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/stellar/go/support/compressxdr"
 	"github.com/stellar/go/support/datastore"
 	"github.com/stellar/go/support/errors"
+	"github.com/stellar/go/xdr"
 )
 
 var testShutdownDelayTime = 300 * time.Millisecond
@@ -45,6 +46,39 @@ func (s *UploaderSuite) TestUpload() {
 	s.testUpload(true)
 }
 
+func (s *UploaderSuite) TestUploadWithMetadata() {
+	key, start, end := "test-1-100", uint32(1), uint32(100)
+	archive := NewLedgerMetaArchive(key, start, end)
+	for i := start; i <= end; i++ {
+		_ = archive.Data.AddLedger(createLedgerCloseMeta(i))
+	}
+	metadata := datastore.MetaData{
+		StartLedger:          start,
+		EndLedger:            end,
+		StartLedgerCloseTime: 123456789,
+		EndLedgerCloseTime:   987654321,
+		ProtocolVersion:      3,
+		CoreVersion:          "v1.2.3",
+		Network:              "testnet",
+		CompressionType:      "gzip",
+		Version:              "1.0.0",
+	}
+	archive.metaData = metadata
+	var capturedBuf bytes.Buffer
+	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, key, mock.Anything, metadata.ToMap()).
+		Run(func(args mock.Arguments) {
+			_ = args.Get(1).(string)
+			_, err := args.Get(2).(io.WriterTo).WriteTo(&capturedBuf)
+			s.Require().NoError(err)
+		}).Return(true, nil).Once()
+
+	registry := prometheus.NewRegistry()
+	queue := NewUploadQueue(1, registry)
+	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
+	s.Require().NoError(dataUploader.Upload(context.Background(), archive))
+
+}
+
 func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 	key, start, end := "test-1-100", uint32(1), uint32(100)
 	archive := NewLedgerMetaArchive(key, start, end)
@@ -54,7 +88,7 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 
 	var capturedBuf bytes.Buffer
 	var capturedKey string
-	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, key, mock.Anything).
+	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, key, mock.Anything, datastore.MetaData{}.ToMap()).
 		Run(func(args mock.Arguments) {
 			capturedKey = args.Get(1).(string)
 			_, err := args.Get(2).(io.WriterTo).WriteTo(&capturedBuf)
@@ -167,7 +201,7 @@ func (s *UploaderSuite) testUploadPutError(putOkReturnVal bool) {
 	archive := NewLedgerMetaArchive(key, start, end)
 
 	s.mockDataStore.On("PutFileIfNotExists", context.Background(), key,
-		mock.Anything).Return(putOkReturnVal, errors.New("error in PutFileIfNotExists")).Once()
+		mock.Anything, datastore.MetaData{}.ToMap()).Return(putOkReturnVal, errors.New("error in PutFileIfNotExists")).Once()
 
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
@@ -211,7 +245,7 @@ func (s *UploaderSuite) TestRunUntilQueueClose() {
 	for i := 1; i <= 100; i++ {
 		key := fmt.Sprintf("test-%d", i)
 		cur := s.mockDataStore.On("PutFileIfNotExists", mock.Anything,
-			key, mock.Anything).Return(true, nil).Once()
+			key, mock.Anything, mock.Anything).Return(true, nil).Once()
 		if prev != nil {
 			cur.NotBefore(prev)
 		}
@@ -242,11 +276,11 @@ func (s *UploaderSuite) TestRunContextCancel() {
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 
-	first := s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test", mock.Anything).
+	first := s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test", mock.Anything, datastore.MetaData{}.ToMap()).
 		Return(true, nil).Once().Run(func(args mock.Arguments) {
 		cancel()
 	})
-	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test1", mock.Anything).
+	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test1", mock.Anything, datastore.MetaData{}.ToMap()).
 		Return(true, nil).Once().NotBefore(first).Run(func(args mock.Arguments) {
 		ctxArg := args.Get(0).(context.Context)
 		s.Require().NoError(ctxArg.Err())
@@ -273,9 +307,20 @@ func (s *UploaderSuite) TestRunUploadError() {
 	s.Require().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test1", 2, 2)))
 
 	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test",
-		mock.Anything).Return(false, errors.New("Put error")).Once()
+		mock.Anything, mock.Anything).Return(false, errors.New("Put error")).Once()
 
 	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
 	err := dataUploader.Run(context.Background(), testShutdownDelayTime)
 	s.Require().Equal("error uploading test: Put error", err.Error())
 }
+
+func NewLedgerMetaArchive(key string, startSeq uint32, endSeq uint32) *LedgerMetaArchive {
+	return &LedgerMetaArchive{
+		ObjectKey: key,
+		Data: xdr.LedgerCloseMetaBatch{
+			StartSequence: xdr.Uint32(startSeq),
+			EndSequence:   xdr.Uint32(endSeq),
+		},
+		metaData: datastore.MetaData{},
+	}
+}
diff --git a/support/datastore/gcs_test.go b/support/datastore/gcs_test.go
index 9fe23bf2..0ddf60ac 100644
--- a/support/datastore/gcs_test.go
+++ b/support/datastore/gcs_test.go
@@ -96,7 +96,7 @@ func TestGCSPutFile(t *testing.T) {
 	writerTo := &writerToRecorder{
 		WriterTo: bytes.NewReader(content),
 	}
-	err = store.PutFile(context.Background(), "file.txt", writerTo)
+	err = store.PutFile(context.Background(), "file.txt", writerTo, nil)
 	require.NoError(t, err)
 	require.Equal(t, int64(len(content)), writerTo.total)
 
@@ -104,17 +104,25 @@ func TestGCSPutFile(t *testing.T) {
 	require.NoError(t, err)
 	requireReaderContentEquals(t, reader, content)
 
+	metadata, err := store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, map[string]string(nil), metadata)
+
 	otherContent := []byte("other text")
 	writerTo = &writerToRecorder{
 		WriterTo: bytes.NewReader(otherContent),
 	}
-	err = store.PutFile(context.Background(), "file.txt", writerTo)
+	err = store.PutFile(context.Background(), "file.txt", writerTo, nil)
 	require.NoError(t, err)
 	require.Equal(t, int64(len(otherContent)), writerTo.total)
 
 	reader, err = store.GetFile(context.Background(), "file.txt")
 	require.NoError(t, err)
 	requireReaderContentEquals(t, reader, otherContent)
+
+	metadata, err = store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, map[string]string(nil), metadata)
 }
 
 func TestGCSPutFileIfNotExists(t *testing.T) {
@@ -140,7 +148,7 @@ func TestGCSPutFileIfNotExists(t *testing.T) {
 	writerTo := &writerToRecorder{
 		WriterTo: bytes.NewReader(newContent),
 	}
-	ok, err := store.PutFileIfNotExists(context.Background(), "file.txt", writerTo)
+	ok, err := store.PutFileIfNotExists(context.Background(), "file.txt", writerTo, nil)
 	require.NoError(t, err)
 	require.False(t, ok)
 	require.Equal(t, int64(len(newContent)), writerTo.total)
@@ -149,10 +157,14 @@ func TestGCSPutFileIfNotExists(t *testing.T) {
 	require.NoError(t, err)
 	requireReaderContentEquals(t, reader, existingContent)
 
+	metadata, err := store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, map[string]string(nil), metadata)
+
 	writerTo = &writerToRecorder{
 		WriterTo: bytes.NewReader(newContent),
 	}
-	ok, err = store.PutFileIfNotExists(context.Background(), "other-file.txt", writerTo)
+	ok, err = store.PutFileIfNotExists(context.Background(), "other-file.txt", writerTo, nil)
 	require.NoError(t, err)
 	require.True(t, ok)
 	require.Equal(t, int64(len(newContent)), writerTo.total)
@@ -160,6 +172,142 @@ func TestGCSPutFileIfNotExists(t *testing.T) {
 	reader, err = store.GetFile(context.Background(), "other-file.txt")
 	require.NoError(t, err)
 	requireReaderContentEquals(t, reader, newContent)
+
+	metadata, err = store.GetFileMetadata(context.Background(), "other-file.txt")
+	require.NoError(t, err)
+	require.Equal(t, map[string]string(nil), metadata)
+}
+
+func TestGCSPutFileWithMetadata(t *testing.T) {
+	server := fakestorage.NewServer([]fakestorage.Object{})
+	defer server.Stop()
+	server.CreateBucketWithOpts(fakestorage.CreateBucketOpts{
+		Name:                  "test-bucket",
+		VersioningEnabled:     false,
+		DefaultEventBasedHold: false,
+	})
+
+	store, err := FromGCSClient(context.Background(), server.Client(), "test-bucket/objects", "testnet")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, store.Close())
+	})
+
+	// Initial metadata
+	metadataObj := MetaData{StartLedger: 1234,
+		EndLedger:            1234,
+		StartLedgerCloseTime: 1234,
+		EndLedgerCloseTime:   1234,
+		Network:              "testnet",
+		CompressionType:      "zstd",
+		ProtocolVersion:      21,
+		CoreVersion:          "v1.2.3",
+		Version:              "1.0.0",
+	}
+
+	content := []byte("inside the file")
+	writerTo := &writerToRecorder{
+		WriterTo: bytes.NewReader(content),
+	}
+	err = store.PutFile(context.Background(), "file.txt", writerTo, metadataObj.ToMap())
+	require.NoError(t, err)
+	require.Equal(t, int64(len(content)), writerTo.total)
+
+	metadata, err := store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, metadataObj.ToMap(), metadata)
+
+	// Update metadata
+	modifiedMetadataObj := MetaData{
+		StartLedger:          5678,
+		EndLedger:            6789,
+		StartLedgerCloseTime: 1622547800,
+		EndLedgerCloseTime:   1622548900,
+		Network:              "mainnet",
+		CompressionType:      "gzip",
+		ProtocolVersion:      23,
+		CoreVersion:          "v1.4.0",
+		Version:              "2.0.0",
+	}
+
+	otherContent := []byte("other text")
+	writerTo = &writerToRecorder{
+		WriterTo: bytes.NewReader(otherContent),
+	}
+	err = store.PutFile(context.Background(), "file.txt", writerTo, modifiedMetadataObj.ToMap())
+	require.NoError(t, err)
+	require.Equal(t, int64(len(otherContent)), writerTo.total)
+
+	metadata, err = store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, modifiedMetadataObj.ToMap(), metadata)
+}
+
+func TestGCSPutFileIfNotExistsWithMetadata(t *testing.T) {
+	existingContent := []byte("inside the file")
+	server := fakestorage.NewServer([]fakestorage.Object{
+		{
+			ObjectAttrs: fakestorage.ObjectAttrs{
+				BucketName: "test-bucket",
+			},
+			Content: existingContent,
+		},
+	})
+	defer server.Stop()
+
+	store, err := FromGCSClient(context.Background(), server.Client(), "test-bucket/objects", "testnet")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, store.Close())
+	})
+
+	metadataObj := MetaData{StartLedger: 1234,
+		EndLedger:            1234,
+		StartLedgerCloseTime: 1234,
+		EndLedgerCloseTime:   1234,
+		Network:              "testnet",
+		CompressionType:      "zstd",
+		ProtocolVersion:      21,
+		CoreVersion:          "v1.2.3",
+		Version:              "1.0.0",
+	}
+
+	newContent := []byte("overwrite the file")
+	writerTo := &writerToRecorder{
+		WriterTo: bytes.NewReader(newContent),
+	}
+	ok, err := store.PutFileIfNotExists(context.Background(), "file.txt", writerTo, metadataObj.ToMap())
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.Equal(t, int64(len(newContent)), writerTo.total)
+
+	metadata, err := store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, metadataObj.ToMap(), metadata)
+
+	modifiedMetadataObj := MetaData{
+		StartLedger:          5678,
+		EndLedger:            6789,
+		StartLedgerCloseTime: 1622547800,
+		EndLedgerCloseTime:   1622548900,
+		Network:              "mainnet",
+		CompressionType:      "gzip",
+		ProtocolVersion:      23,
+		CoreVersion:          "v1.4.0",
+		Version:              "2.0.0",
+	}
+
+	writerTo = &writerToRecorder{
+		WriterTo: bytes.NewReader(newContent),
+	}
+	ok, err = store.PutFileIfNotExists(context.Background(), "file.txt", writerTo, modifiedMetadataObj.ToMap())
+	require.NoError(t, err)
+	require.False(t, ok)
+	require.Equal(t, int64(len(newContent)), writerTo.total)
+
+	metadata, err = store.GetFileMetadata(context.Background(), "file.txt")
+	require.NoError(t, err)
+	require.Equal(t, metadataObj.ToMap(), metadata)
 }
 
 func TestGCSGetNonExistentFile(t *testing.T) {
@@ -183,6 +331,10 @@ func TestGCSGetNonExistentFile(t *testing.T) {
 
 	_, err = store.GetFile(context.Background(), "other-file.txt")
 	require.ErrorIs(t, err, os.ErrNotExist)
+
+	metadata, err := store.GetFileMetadata(context.Background(), "other-file.txt")
+	require.ErrorIs(t, err, os.ErrNotExist)
+	require.Equal(t, map[string]string(nil), metadata)
 }
 
 func TestGCSGetFileValidatesCRC32C(t *testing.T) {
diff --git a/support/datastore/metadata_test.go b/support/datastore/metadata_test.go
new file mode 100644
index 00000000..6ec6a176
--- /dev/null
+++ b/support/datastore/metadata_test.go
@@ -0,0 +1,89 @@
+package datastore
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetaDataToMap(t *testing.T) {
+
+	var tests = []struct {
+		name     string
+		metaData MetaData
+		expected map[string]string
+	}{
+		{
+			name: "testToMap",
+			metaData: MetaData{
+				StartLedger:          100,
+				EndLedger:            200,
+				StartLedgerCloseTime: 123456789,
+				EndLedgerCloseTime:   987654321,
+				ProtocolVersion:      3,
+				CoreVersion:          "v1.2.3",
+				Network:              "testnet",
+				CompressionType:      "gzip",
+				Version:              "1.0.0",
+			},
+			expected: map[string]string{
+				"start-ledger":            "100",
+				"end-ledger":              "200",
+				"start-ledger-close-time": "123456789",
+				"end-ledger-close-time":   "987654321",
+				"protocol-version":        "3",
+				"core-version":            "v1.2.3",
+				"network":                 "testnet",
+				"compression-type":        "gzip",
+				"version":                 "1.0.0",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := MetaData{
+				StartLedger:          tt.metaData.StartLedger,
+				EndLedger:            tt.metaData.EndLedger,
+				StartLedgerCloseTime: tt.metaData.StartLedgerCloseTime,
+				EndLedgerCloseTime:   tt.metaData.EndLedgerCloseTime,
+				ProtocolVersion:      tt.metaData.ProtocolVersion,
+				CoreVersion:          tt.metaData.CoreVersion,
+				Network:              tt.metaData.Network,
+				CompressionType:      tt.metaData.CompressionType,
+				Version:              tt.metaData.Version,
+			}
+			got := m.ToMap()
+			require.Equal(t, got, tt.expected)
+		})
+	}
+}
+
+func TestNewMetaDataFromMap(t *testing.T) {
+	data := map[string]string{
+		"start-ledger":            "100",
+		"end-ledger":              "200",
+		"start-ledger-close-time": "123456789",
+		"end-ledger-close-time":   "987654321",
+		"protocol-version":        "3",
+		"core-version":            "v1.2.3",
+		"network":                 "testnet",
+		"compression-type":        "gzip",
+		"version":                 "1.0.0",
+	}
+
+	expected := MetaData{
+		StartLedger:          100,
+		EndLedger:            200,
+		StartLedgerCloseTime: 123456789,
+		EndLedgerCloseTime:   987654321,
+		ProtocolVersion:      3,
+		CoreVersion:          "v1.2.3",
+		Network:              "testnet",
+		CompressionType:      "gzip",
+		Version:              "1.0.0",
+	}
+
+	got, err := NewMetaDataFromMap(data)
+	require.NoError(t, err)
+	require.Equal(t, got, expected)
+}