Repository: stellar/go
Pull request #5264 (2024-03-28 00:53:48)

/exp/services/ledgerexporter: resumable export, check data storage for optimal starting point
<!-- 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

* [x] 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).
* [x] 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

* [x] 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

* Added `ResumableManager`, app will call this during init to apply a best effort attempt to check data storage layer and increase the effective `start` value for captive core ingestion to the boundary aligned starting ledger of the nearest absent object key of data store relative to requested `start`. 
* Enhanced the DataStoreConfig toml model to be more generic key/value params to allow for holding different data store configuration values going forward, the app checks `datastore_config.type` to know which datastore instance to create, passing the `datastore_config.params` to it:
  ```
  [datastore_config]
  type = "GCS"
  
  [datastore_config.params]
  destination_url = "gcs://your-bucket-name"
  ```

### Why

Efficiency of exporter, avoiding exporting files to data store that may already be present.

Closes: https://stellarorg.atlassian.net/browse/HUBBLE-272

### Known limitations




sreuland: I need to run some live integration tests against GCS.
sreuland: > I need to run some live integration tests, building ledgerexporter and running it against GCS bucket.

I ran ledgerexporter on testnet against a real gcs bucket doing a sequence of commands that requested ranges overlapping, to confirm the expected resumable starts were calculated, using `ledgers_per_file=1`:

```
ledgerexporter --config-file config.toml --start 842267 --end 844267
... started core at 842267, uploaded all 2000

ledgerexporter --config-file config.toml --start 842267 --end 844267
... log output stating full range existed on data store already, no core or upload attempted

ledgerexporter --config-file config.toml --start 842267 --end 844277
... log output stating start was adjusted to 844268, uploaded next 10

 ledgerexporter --config-file config.toml --start 842267 --end 844287
... log output stating start was adjusted to 844278, uploaded next 10

 ledgerexporter --config-file config.toml --start 842267 --end 0
... log output stating start was adjusted to 844288, and continued on unbounded upload from there
```

Patch:
diff --git a/exp/services/ledgerexporter/Makefile b/exp/services/ledgerexporter/Makefile
index e4d6f647..5fee16f9 100644
--- a/exp/services/ledgerexporter/Makefile
+++ b/exp/services/ledgerexporter/Makefile
@@ -14,9 +14,15 @@ $(if $(STELLAR_CORE_VERSION), --build-arg STELLAR_CORE_VERSION=$(STELLAR_CORE_VE
 	-t $(DOCKER_IMAGE):$(VERSION) \
 	-t $(DOCKER_IMAGE):latest .
 
-docker-test:
+docker-clean:
+	$(SUDO) docker stop fake-gcs-server || true
+	$(SUDO) docker rm fake-gcs-server || true
+	$(SUDO) rm -rf ${PWD}/storage || true
+	$(SUDO) docker network rm test-network || true
+    
+docker-test: docker-clean
 	# Create temp storage dir
-	$(SUDO) mkdir -p ${PWD}/storage/exporter-test
+	$(SUDO) mkdir -p ${PWD}/storage/exporter-test 
 
 	# Create test network for docker
 	$(SUDO) docker network create test-network
@@ -28,16 +34,13 @@ docker-test:
 	# Run the ledger-exporter
 	$(SUDO) docker run --platform linux/amd64 -t --network test-network\
 		 -e NETWORK=pubnet \
-		 -e ARCHIVE_TARGET=gcs://exporter-test \
+		 -e ARCHIVE_TARGET=exporter-test/test-subpath \
 		 -e START=1000 \
 		 -e END=2000 \
 		 -e STORAGE_EMULATOR_HOST=http://fake-gcs-server:4443 \
 		 $(DOCKER_IMAGE):$(VERSION)
-
-	$(SUDO) docker stop fake-gcs-server
-	$(SUDO) docker rm fake-gcs-server
-	$(SUDO) rm -rf ${PWD}/storage
-	$(SUDO) docker network rm test-network
+		 
+	$(MAKE) docker-clean
 
 docker-push:
 	$(SUDO) docker push $(DOCKER_IMAGE):$(VERSION)
diff --git a/exp/services/ledgerexporter/README.md b/exp/services/ledgerexporter/README.md
index ad8a3dde..5d483f08 100644
--- a/exp/services/ledgerexporter/README.md
+++ b/exp/services/ledgerexporter/README.md
@@ -33,17 +33,24 @@ Exports ledgers continuously starting from --start. In this mode, the end ledger
 ledgerexporter --start <start_ledger> --config-file <config_file_path>
 ```
 
+#### Resumability:
+Exporting a ledger range can be optimized further by enabling resumability if the remote data store supports it.
 
-Starts exporting from a specified number of ledgers before the latest ledger sequence number on the network.
-```bash
-ledgerexporter --from-last <number_of_ledgers> --config-file <config_file_path>
-```
+By default, resumability is disabled, `--resume false`
+
+When enabled, `--resume true`, ledgerexporter will search the remote data store within the requested range, looking for the oldest absent ledger sequence number within range. If abscence is detected, the export range is narrowed to `--start <absent_ledger_sequence>`. 
+This feature requires all ledgers to be present on the remote data store for some (possibly empty) prefix of the requested range and then absent for the (possibly empty) remainder.
 
 ### Configuration (toml):
 
 ```toml
 network = "testnet"  # Options: `testnet` or `pubnet`
-destination_url = "gcs://your-bucket-name"
+
+[datastore_config]
+type = "GCS"
+
+[datastore_config.params]
+destination_bucket_path = "your-bucket-name/<optional_subpaths>"
 
 [exporter_config]
 ledgers_per_file = 64
diff --git a/exp/services/ledgerexporter/config.toml b/exp/services/ledgerexporter/config.toml
index 30c4a4fa..a5608742 100644
--- a/exp/services/ledgerexporter/config.toml
+++ b/exp/services/ledgerexporter/config.toml
@@ -1,7 +1,15 @@
 network = "testnet"
-destination_url = "gcs://exporter-test/ledgers"
+
+[datastore_config]
+type = "GCS"
+
+[datastore_config.params]
+destination_bucket_path = "exporter-test/ledgers"
 
 [exporter_config]
   ledgers_per_file = 1
   files_per_partition = 64000
 
+[stellar_core_config]
+  stellar_core_binary_path = "/usr/local/bin/stellar-core"
+
diff --git a/exp/services/ledgerexporter/docker/start b/exp/services/ledgerexporter/docker/start
index 8cca164d..eca213de 100644
--- a/exp/services/ledgerexporter/docker/start
+++ b/exp/services/ledgerexporter/docker/start
@@ -18,7 +18,12 @@ files_per_partition="${FILES_PER_PARTITION:-64000}"
 # Generate TOML configuration
 cat <<EOF > config.toml
 network = "${NETWORK}"
-destination_url = "${ARCHIVE_TARGET}"
+
+[datastore_config]
+type = "GCS"
+
+[datastore_config.params]
+destination_bucket_path = "${ARCHIVE_TARGET}"
 
 [exporter_config]
   ledgers_per_file = $ledgers_per_file
@@ -29,10 +34,6 @@ EOF
 if [[ -n "$START" || -n "$END" ]]; then
     echo "START: $START END: $END"
     /usr/bin/ledgerexporter --config-file config.toml --start $START --end $END
-# Check if FROM_LAST variable is set
-elif [[ -n "$FROM_LAST" ]]; then
-    echo "FROM_LAST: $FROM_LAST"
-    /usr/bin/ledgerexporter --config-file config.toml --from-last  $FROM_LAST
 else
     echo "Error: No ledger range provided."
     exit 1
diff --git a/exp/services/ledgerexporter/internal/app.go b/exp/services/ledgerexporter/internal/app.go
index 8239b9c7..58816458 100644
--- a/exp/services/ledgerexporter/internal/app.go
+++ b/exp/services/ledgerexporter/internal/app.go
@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/stellar/go/historyarchive"
 
 	"github.com/stellar/go/ingest/ledgerbackend"
 	_ "github.com/stellar/go/network"
@@ -24,41 +25,118 @@ var (
 	logger = log.New().WithField("service", "ledger-exporter")
 )
 
+func NewDataAlreadyExportedError(Start uint32, End uint32) *DataAlreadyExportedError {
+	return &DataAlreadyExportedError{
+		Start: Start,
+		End:   End,
+	}
+}
+
+type DataAlreadyExportedError struct {
+	Start uint32
+	End   uint32
+}
+
+func (m DataAlreadyExportedError) Error() string {
+	return fmt.Sprintf("For export ledger range start=%d, end=%d, the remote storage has all the data, there is no need to continue export", m.Start, m.End)
+}
+
+func NewInvalidDataStoreError(LedgerSequence uint32, LedgersPerFile uint32) *InvalidDataStoreError {
+	return &InvalidDataStoreError{
+		LedgerSequence: LedgerSequence,
+		LedgersPerFile: LedgersPerFile,
+	}
+}
+
+type InvalidDataStoreError struct {
+	LedgerSequence uint32
+	LedgersPerFile uint32
+}
+
+func (m InvalidDataStoreError) Error() string {
+	return fmt.Sprintf("The remote data store has inconsistent data, "+
+		"a resumable starting ledger of %v was identified, "+
+		"but that is not aligned to expected ledgers-per-file of %v. use '--resume false' to bypass",
+		m.LedgerSequence, m.LedgersPerFile)
+}
+
 type App struct {
-	config             Config
+	config             *Config
 	ledgerBackend      ledgerbackend.LedgerBackend
 	dataStore          DataStore
 	exportManager      *ExportManager
 	uploader           Uploader
+	flags              Flags
 	prometheusRegistry *prometheus.Registry
 }
 
-func NewApp() *App {
+func NewApp(flags Flags) *App {
 	logger.SetLevel(log.DebugLevel)
-
-	config := Config{}
-	err := config.LoadConfig()
-	logFatalIf(err, "Could not load configuration")
-
-	app := &App{config: config, prometheusRegistry: prometheus.NewRegistry()}
-	app.prometheusRegistry.MustRegister(
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(
 		collectors.NewProcessCollector(collectors.ProcessCollectorOpts{Namespace: "ledger_exporter"}),
 		collectors.NewGoCollector(),
 	)
+	app := &App{flags: flags, prometheusRegistry: registry}
 	return app
 }
 
-func (a *App) init(ctx context.Context) {
-	a.dataStore = mustNewDataStore(ctx, a.config)
-	a.ledgerBackend = mustNewLedgerBackend(ctx, a.config, a.prometheusRegistry)
+func (a *App) init(ctx context.Context) error {
+	var err error
+	var archive historyarchive.ArchiveInterface
+
+	if a.config, err = NewConfig(ctx, a.flags); err != nil {
+		return errors.Wrap(err, "Could not load configuration")
+	}
+	if archive, err = createHistoryArchiveFromNetworkName(ctx, a.config.Network); err != nil {
+		return err
+	}
+	a.config.ValidateAndSetLedgerRange(ctx, archive)
+
+	if a.dataStore, err = NewDataStore(ctx, a.config.DataStoreConfig, a.config.Network); err != nil {
+		return errors.Wrap(err, "Could not connect to destination data store")
+	}
+	if a.config.Resume {
+		if err = a.applyResumability(ctx,
+			NewResumableManager(a.dataStore, a.config.Network, a.config.LedgerBatchConfig, archive)); err != nil {
+			return err
+		}
+	}
+
+	logger.Infof("Final computed ledger range for backend retrieval and export, start=%d, end=%d", a.config.StartLedger, a.config.EndLedger)
+
+	if a.ledgerBackend, err = newLedgerBackend(ctx, a.config, a.prometheusRegistry); err != nil {
+		return err
+	}
+
 	// TODO: make number of upload workers configurable instead of hard coding it to 1
 	queue := NewUploadQueue(1, a.prometheusRegistry)
-	a.exportManager = NewExportManager(a.config.ExporterConfig, a.ledgerBackend, queue, a.prometheusRegistry)
-	a.uploader = NewUploader(
-		a.dataStore,
-		queue,
-		a.prometheusRegistry,
-	)
+	if a.exportManager, err = NewExportManager(a.config.LedgerBatchConfig, a.ledgerBackend, queue, a.prometheusRegistry); err != nil {
+		return err
+	}
+	a.uploader = NewUploader(a.dataStore, queue, a.prometheusRegistry)
+
+	return nil
+}
+
+func (a *App) applyResumability(ctx context.Context, resumableManager ResumableManager) error {
+	absentLedger, ok, err := resumableManager.FindStart(ctx, a.config.StartLedger, a.config.EndLedger)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return NewDataAlreadyExportedError(a.config.StartLedger, a.config.EndLedger)
+	}
+
+	// TODO - evaluate a more robust validation of remote data for ledgers-per-file consistency
+	// this assumes ValidateAndSetLedgerRange() has conditioned the a.config.StartLedger to be at least > 1
+	if absentLedger > 2 && absentLedger != a.config.LedgerBatchConfig.GetSequenceNumberStartBoundary(absentLedger) {
+		return NewInvalidDataStoreError(absentLedger, a.config.LedgerBatchConfig.LedgersPerFile)
+	}
+	logger.Infof("For export ledger range start=%d, end=%d, the remote storage has some of this data already, will resume at later start ledger of %d", a.config.StartLedger, a.config.EndLedger, absentLedger)
+	a.config.StartLedger = absentLedger
+
+	return nil
 }
 
 func (a *App) close() {
@@ -92,7 +170,15 @@ func (a *App) Run() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	a.init(ctx)
+	if err := a.init(ctx); err != nil {
+		var dataAlreadyExported DataAlreadyExportedError
+		if errors.As(err, &dataAlreadyExported) {
+			logger.Info(err.Error())
+			logger.Info("Shutting down ledger-exporter")
+			return
+		}
+		logger.WithError(err).Fatal("Stopping ledger-exporter")
+	}
 	defer a.close()
 
 	var wg sync.WaitGroup
@@ -133,22 +219,20 @@ func (a *App) Run() {
 	logger.Info("Shutting down ledger-exporter")
 }
 
-func mustNewDataStore(ctx context.Context, config Config) DataStore {
-	dataStore, err := NewDataStore(ctx, fmt.Sprintf("%s/%s", config.DestinationURL, config.Network))
-	logFatalIf(err, "Could not connect to destination data store")
-	return dataStore
-}
-
-// mustNewLedgerBackend Creates and initializes captive core ledger backend
+// newLedgerBackend Creates and initializes captive core ledger backend
 // Currently, only supports captive-core as ledger backend
-func mustNewLedgerBackend(ctx context.Context, config Config, prometheusRegistry *prometheus.Registry) ledgerbackend.LedgerBackend {
-	captiveConfig := config.GenerateCaptiveCoreConfig()
+func newLedgerBackend(ctx context.Context, config *Config, prometheusRegistry *prometheus.Registry) (ledgerbackend.LedgerBackend, error) {
+	captiveConfig, err := config.GenerateCaptiveCoreConfig()
+	if err != nil {
+		return nil, err
+	}
 
 	var backend ledgerbackend.LedgerBackend
-	var err error
 	// Create a new captive core backend
 	backend, err = ledgerbackend.NewCaptive(captiveConfig)
-	logFatalIf(err, "Failed to create captive-core instance")
+	if err != nil {
+		return nil, errors.Wrap(err, "Failed to create captive-core instance")
+	}
 	backend = ledgerbackend.WithMetrics(backend, prometheusRegistry, "ledger_exporter")
 
 	var ledgerRange ledgerbackend.Range
@@ -158,13 +242,8 @@ func mustNewLedgerBackend(ctx context.Context, config Config, prometheusRegistry
 		ledgerRange = ledgerbackend.BoundedRange(config.StartLedger, config.EndLedger)
 	}
 
-	err = backend.PrepareRange(ctx, ledgerRange)
-	logFatalIf(err, "Could not prepare captive core ledger backend")
-	return backend
-}
-
-func logFatalIf(err error, message string, args ...interface{}) {
-	if err != nil {
-		logger.WithError(err).Fatalf(message, args...)
+	if err = backend.PrepareRange(ctx, ledgerRange); err != nil {
+		return nil, errors.Wrap(err, "Could not prepare captive core ledger backend")
 	}
+	return backend, nil
 }
diff --git a/exp/services/ledgerexporter/internal/config.go b/exp/services/ledgerexporter/internal/config.go
index 17e76a5d..0d3cb87c 100644
--- a/exp/services/ledgerexporter/internal/config.go
+++ b/exp/services/ledgerexporter/internal/config.go
@@ -1,8 +1,8 @@
 package ledgerexporter
 
 import (
+	"context"
 	_ "embed"
-	"flag"
 	"os/exec"
 
 	"github.com/stellar/go/historyarchive"
@@ -13,11 +13,20 @@ import (
 
 	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/support/ordered"
+	"github.com/stellar/go/support/storage"
 )
 
 const Pubnet = "pubnet"
 const Testnet = "testnet"
 
+type Flags struct {
+	StartLedger    uint32
+	EndLedger      uint32
+	ConfigFilePath string
+	Resume         bool
+	AdminPort      uint
+}
+
 type StellarCoreConfig struct {
 	NetworkPassphrase     string   `toml:"network_passphrase"`
 	HistoryArchiveUrls    []string `toml:"history_archive_urls"`
@@ -25,94 +34,97 @@ type StellarCoreConfig struct {
 	CaptiveCoreTomlPath   string   `toml:"captive_core_toml_path"`
 }
 
+type DataStoreConfig struct {
+	Type   string            `toml:"type"`
+	Params map[string]string `toml:"params"`
+}
+
 type Config struct {
 	AdminPort int `toml:"admin_port"`
 
 	Network           string            `toml:"network"`
-	DestinationURL    string            `toml:"destination_url"`
-	ExporterConfig    ExporterConfig    `toml:"exporter_config"`
+	DataStoreConfig   DataStoreConfig   `toml:"datastore_config"`
+	LedgerBatchConfig LedgerBatchConfig `toml:"exporter_config"`
 	StellarCoreConfig StellarCoreConfig `toml:"stellar_core_config"`
 
-	//From command-line
-	StartLedger          uint32 `toml:"start"`
-	EndLedger            uint32 `toml:"end"`
-	StartFromLastLedgers uint32 `toml:"from-last"`
+	StartLedger uint32
+	EndLedger   uint32
+	Resume      bool
 }
 
-func (config *Config) LoadConfig() error {
-	// Parse command-line options
-	startLedger := flag.Uint("start", 0, "Starting ledger")
-	endLedger := flag.Uint("end", 0, "Ending ledger (inclusive)")
-	startFromLastNLedger := flag.Uint("from-last", 0, "Start streaming from last N ledgers")
-	adminPort := flag.Int("admin-port", 0, "Admin HTTP port for prometheus metrics")
-
-	configFilePath := flag.String("config-file", "config.toml", "Path to the TOML config file")
-	flag.Parse()
-
-	config.StartLedger = uint32(*startLedger)
-	config.EndLedger = uint32(*endLedger)
-	config.StartFromLastLedgers = uint32(*startFromLastNLedger)
-	config.AdminPort = *adminPort
-
-	// Load config TOML file
-	cfg, err := toml.LoadFile(*configFilePath)
-	if err != nil {
-		return err
-	}
-
-	// Unmarshal TOML data into the Config struct
-	err = cfg.Unmarshal(config)
-	logFatalIf(err, "Error unmarshalling TOML config.")
-	logger.Infof("Config: %v", *config)
-
+func createHistoryArchiveFromNetworkName(ctx context.Context, networkName string) (historyarchive.ArchiveInterface, error) {
 	var historyArchiveUrls []string
-	switch config.Network {
+	switch networkName {
 	case Pubnet:
 		historyArchiveUrls = network.PublicNetworkhistoryArchiveURLs
 	case Testnet:
 		historyArchiveUrls = network.TestNetworkhistoryArchiveURLs
 	default:
-		logger.Fatalf("Invalid network %s", config.Network)
+		return nil, errors.Errorf("Invalid network name %s", networkName)
 	}
 
-	// Retrieve the latest ledger sequence from history archives
-	latestNetworkLedger, err := getLatestLedgerSequenceFromHistoryArchives(historyArchiveUrls)
-	logFatalIf(err, "Failed to retrieve the latest ledger sequence from history archives.")
+	return historyarchive.NewArchivePool(historyArchiveUrls, historyarchive.ArchiveOptions{
+		ConnectOptions: storage.ConnectOptions{
+			UserAgent: "ledger-exporter",
+			Context:   ctx,
+		},
+	})
+}
 
-	// Validate config params
-	err = config.validateAndSetLedgerRange(latestNetworkLedger)
-	logFatalIf(err, "Error validating config params.")
+func getLatestLedgerSequenceFromHistoryArchives(archive historyarchive.ArchiveInterface) (uint32, error) {
+	has, err := archive.GetRootHAS()
+	if err != nil {
+		logger.WithError(err).Warnf("Error getting root HAS from archives")
+		return 0, errors.Wrap(err, "failed to retrieve the latest ledger sequence from any history archive")
+	}
 
-	// Validate and build the appropriate range
-	// TODO: Make it configurable
-	config.adjustLedgerRange()
+	return has.CurrentLedger, nil
+}
 
-	return nil
+func getHistoryArchivesCheckPointFrequency() uint32 {
+	// this could evolve to use other sources for checkpoint freq
+	return historyarchive.DefaultCheckpointFrequency
 }
 
-func (config *Config) validateAndSetLedgerRange(latestNetworkLedger uint32) error {
-	if config.StartFromLastLedgers > 0 && (config.StartLedger > 0 || config.EndLedger > 0) {
-		return errors.New("--from-last cannot be used with --start or --end")
+// 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) {
+	config := &Config{}
+
+	config.StartLedger = uint32(flags.StartLedger)
+	config.EndLedger = uint32(flags.EndLedger)
+	config.Resume = flags.Resume
+
+	logger.Infof("Requested ledger range start=%d, end=%d, resume=%v", config.StartLedger, config.EndLedger, config.Resume)
+
+	var err error
+	if err = config.processToml(flags.ConfigFilePath); err != nil {
+		return nil, err
 	}
+	logger.Infof("Config: %v", *config)
 
-	if config.StartFromLastLedgers > 0 {
-		if config.StartFromLastLedgers > latestNetworkLedger {
-			return errors.Errorf("--from-last %d exceeds latest network ledger %d",
-				config.StartLedger, latestNetworkLedger)
-		}
-		config.StartLedger = latestNetworkLedger - config.StartFromLastLedgers
-		logger.Infof("Setting start ledger to %d, calculated as latest ledger (%d) minus --from-last value (%d)",
-			config.StartLedger, latestNetworkLedger, config.StartFromLastLedgers)
+	return config, nil
+}
+
+// Validates requested ledger range, and will automatically adjust it
+// to be ledgers-per-file boundary aligned
+func (config *Config) ValidateAndSetLedgerRange(ctx context.Context, archive historyarchive.ArchiveInterface) error {
+	latestNetworkLedger, err := getLatestLedgerSequenceFromHistoryArchives(archive)
+
+	if err != nil {
+		return errors.Wrap(err, "Failed to retrieve the latest ledger sequence from history archives.")
 	}
+	logger.Infof("Latest %v ledger sequence was detected as %d", config.Network, latestNetworkLedger)
 
 	if config.StartLedger > latestNetworkLedger {
 		return errors.Errorf("--start %d exceeds latest network ledger %d",
 			config.StartLedger, latestNetworkLedger)
 	}
 
-	// Ensure that the start ledger is at least 2.
-	config.StartLedger = ordered.Max(2, config.StartLedger)
-
 	if config.EndLedger != 0 { // Bounded mode
 		if config.EndLedger < config.StartLedger {
 			return errors.New("invalid --end value, must be >= --start")
@@ -123,39 +135,19 @@ func (config *Config) validateAndSetLedgerRange(latestNetworkLedger uint32) erro
 		}
 	}
 
+	config.adjustLedgerRange()
 	return nil
 }
 
-func (config *Config) adjustLedgerRange() {
-	logger.Infof("Requested ledger range start=%d, end=%d", config.StartLedger, config.EndLedger)
-
-	// Check if either the start or end ledger does not fall on the "LedgersPerFile" boundary
-	// and adjust the start and end ledger accordingly.
-	// Align the start ledger to the nearest "LedgersPerFile" boundary.
-	config.StartLedger = config.StartLedger / config.ExporterConfig.LedgersPerFile * config.ExporterConfig.LedgersPerFile
-
-	// Ensure that the adjusted start ledger is at least 2.
-	config.StartLedger = ordered.Max(2, config.StartLedger)
-
-	// Align the end ledger (for bounded cases) to the nearest "LedgersPerFile" boundary.
-	if config.EndLedger != 0 {
-		// Add an extra batch only if "LedgersPerFile" is greater than 1 and the end ledger doesn't fall on the boundary.
-		if config.ExporterConfig.LedgersPerFile > 1 && config.EndLedger%config.ExporterConfig.LedgersPerFile != 0 {
-			config.EndLedger = (config.EndLedger/config.ExporterConfig.LedgersPerFile + 1) * config.ExporterConfig.LedgersPerFile
-		}
-	}
-
-	logger.Infof("Adjusted ledger range: start=%d, end=%d", config.StartLedger, config.EndLedger)
-}
-
-func (config *Config) GenerateCaptiveCoreConfig() ledgerbackend.CaptiveCoreConfig {
+func (config *Config) GenerateCaptiveCoreConfig() (ledgerbackend.CaptiveCoreConfig, error) {
 	coreConfig := &config.StellarCoreConfig
 
 	// Look for stellar-core binary in $PATH, if not supplied
 	if coreConfig.StellarCoreBinaryPath == "" {
 		var err error
-		coreConfig.StellarCoreBinaryPath, err = exec.LookPath("stellar-core")
-		logFatalIf(err, "Failed to find stellar-core binary")
+		if coreConfig.StellarCoreBinaryPath, err = exec.LookPath("stellar-core"); err != nil {
+			return ledgerbackend.CaptiveCoreConfig{}, errors.Wrap(err, "Failed to find stellar-core binary")
+		}
 	}
 
 	var captiveCoreConfig []byte
@@ -182,16 +174,58 @@ func (config *Config) GenerateCaptiveCoreConfig() ledgerbackend.CaptiveCoreConfi
 	}
 
 	captiveCoreToml, err := ledgerbackend.NewCaptiveCoreTomlFromData(captiveCoreConfig, params)
-	logFatalIf(err, "Failed to create captive-core toml")
+	if err != nil {
+		return ledgerbackend.CaptiveCoreConfig{}, errors.Wrap(err, "Failed to create captive-core toml")
+	}
 
 	return ledgerbackend.CaptiveCoreConfig{
 		BinaryPath:          coreConfig.StellarCoreBinaryPath,
 		NetworkPassphrase:   params.NetworkPassphrase,
 		HistoryArchiveURLs:  params.HistoryArchiveURLs,
-		CheckpointFrequency: historyarchive.DefaultCheckpointFrequency,
+		CheckpointFrequency: getHistoryArchivesCheckPointFrequency(),
 		Log:                 logger.WithField("subservice", "stellar-core"),
 		Toml:                captiveCoreToml,
 		UserAgent:           "ledger-exporter",
 		UseDB:               true,
+	}, nil
+}
+
+func (config *Config) processToml(tomlPath string) error {
+	// Load config TOML file
+	cfg, err := toml.LoadFile(tomlPath)
+	if err != nil {
+		return err
+	}
+
+	// Unmarshal TOML data into the Config struct
+	if err := cfg.Unmarshal(config); err != nil {
+		return errors.Wrap(err, "Error unmarshalling TOML config.")
 	}
+
+	// validate TOML data
+	if config.Network == "" {
+		return errors.New("Invalid TOML config, 'network' must be set, supported values are 'testnet' or 'pubnet'")
+	}
+	return nil
+}
+
+func (config *Config) adjustLedgerRange() {
+
+	// Check if either the start or end ledger does not fall on the "LedgersPerFile" boundary
+	// and adjust the start and end ledger accordingly.
+	// Align the start ledger to the nearest "LedgersPerFile" boundary.
+	config.StartLedger = config.LedgerBatchConfig.GetSequenceNumberStartBoundary(config.StartLedger)
+
+	// Ensure that the adjusted start ledger is at least 2.
+	config.StartLedger = ordered.Max(2, config.StartLedger)
+
+	// Align the end ledger (for bounded cases) to the nearest "LedgersPerFile" boundary.
+	if config.EndLedger != 0 {
+		// Add an extra batch only if "LedgersPerFile" is greater than 1 and the end ledger doesn't fall on the boundary.
+		if config.LedgerBatchConfig.LedgersPerFile > 1 && config.EndLedger%config.LedgerBatchConfig.LedgersPerFile != 0 {
+			config.EndLedger = (config.EndLedger/config.LedgerBatchConfig.LedgersPerFile + 1) * config.LedgerBatchConfig.LedgersPerFile
+		}
+	}
+
+	logger.Infof("Computed effective export boundary ledger range: start=%d, end=%d", config.StartLedger, config.EndLedger)
 }
diff --git a/exp/services/ledgerexporter/internal/datastore.go b/exp/services/ledgerexporter/internal/datastore.go
index a53d9746..0529ba22 100644
--- a/exp/services/ledgerexporter/internal/datastore.go
+++ b/exp/services/ledgerexporter/internal/datastore.go
@@ -3,13 +3,8 @@ package ledgerexporter
 import (
 	"context"
 	"io"
-	"strings"
-
-	"cloud.google.com/go/storage"
-	"google.golang.org/api/option"
 
 	"github.com/stellar/go/support/errors"
-	"github.com/stellar/go/support/url"
 )
 
 // DataStore defines an interface for interacting with data storage
@@ -22,37 +17,12 @@ type DataStore interface {
 	Close() error
 }
 
-// NewDataStore creates a new DataStore based on the destination URL.
-// Currently, only accepts GCS URLs.
-func NewDataStore(ctx context.Context, destinationURL string) (DataStore, error) {
-	parsed, err := url.Parse(destinationURL)
-	if err != nil {
-		return nil, err
-	}
-
-	pth := parsed.Path
-	if parsed.Scheme != "gcs" {
-		return nil, errors.Errorf("Invalid destination URL %s. Expected GCS URL ", destinationURL)
-	}
-
-	// Inside gcs, all paths start _without_ the leading /
-	pth = strings.TrimPrefix(pth, "/")
-	bucketName := parsed.Host
-	prefix := pth
-
-	logger.Infof("creating GCS client for bucket: %s, prefix: %s", bucketName, prefix)
-
-	var options []option.ClientOption
-	client, err := storage.NewClient(ctx, options...)
-	if err != nil {
-		return nil, err
+// NewDataStore factory, it creates a new DataStore based on the config type
+func NewDataStore(ctx context.Context, datastoreConfig DataStoreConfig, network string) (DataStore, error) {
+	switch datastoreConfig.Type {
+	case "GCS":
+		return NewGCSDataStore(ctx, datastoreConfig.Params, network)
+	default:
+		return nil, errors.Errorf("Invalid datastore type %v, not supported", datastoreConfig.Type)
 	}
-
-	// Check the bucket exists
-	bucket := client.Bucket(bucketName)
-	if _, err := bucket.Attrs(ctx); err != nil {
-		return nil, errors.Wrap(err, "failed to retrieve bucket attributes")
-	}
-
-	return &GCSDataStore{client: client, bucket: bucket, prefix: prefix}, nil
 }
diff --git a/exp/services/ledgerexporter/internal/encoder.go b/exp/services/ledgerexporter/internal/encoder.go
new file mode 100644
index 00000000..33909ace
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/encoder.go
@@ -0,0 +1,39 @@
+package ledgerexporter
+
+import (
+	"compress/gzip"
+	"io"
+
+	xdr3 "github.com/stellar/go-xdr/xdr3"
+)
+
+type XDRGzipEncoder struct {
+	XdrPayload interface{}
+}
+
+func (g *XDRGzipEncoder) WriteTo(w io.Writer) (int64, error) {
+	gw := gzip.NewWriter(w)
+	n, err := xdr3.Marshal(gw, g.XdrPayload)
+	if err != nil {
+		return int64(n), err
+	}
+	return int64(n), gw.Close()
+}
+
+type XDRGzipDecoder struct {
+	XdrPayload interface{}
+}
+
+func (d *XDRGzipDecoder) ReadFrom(r io.Reader) (int64, error) {
+	gr, err := gzip.NewReader(r)
+	if err != nil {
+		return 0, err
+	}
+	defer gr.Close()
+
+	n, err := xdr3.Unmarshal(gr, d.XdrPayload)
+	if err != nil {
+		return int64(n), err
+	}
+	return int64(n), nil
+}
diff --git a/exp/services/ledgerexporter/internal/exportmanager.go b/exp/services/ledgerexporter/internal/exportmanager.go
index a918a16e..51c39475 100644
--- a/exp/services/ledgerexporter/internal/exportmanager.go
+++ b/exp/services/ledgerexporter/internal/exportmanager.go
@@ -11,14 +11,8 @@ import (
 	"github.com/stellar/go/xdr"
 )
 
-type ExporterConfig struct {
-	LedgersPerFile    uint32 `toml:"ledgers_per_file"`
-	FilesPerPartition uint32 `toml:"files_per_partition"`
-}
-
-// ExportManager manages the creation and handling of export objects.
 type ExportManager struct {
-	config             ExporterConfig
+	config             LedgerBatchConfig
 	ledgerBackend      ledgerbackend.LedgerBackend
 	currentMetaArchive *LedgerMetaArchive
 	queue              UploadQueue
@@ -26,7 +20,11 @@ type ExportManager struct {
 }
 
 // NewExportManager creates a new ExportManager with the provided configuration.
-func NewExportManager(config ExporterConfig, backend ledgerbackend.LedgerBackend, queue UploadQueue, prometheusRegistry *prometheus.Registry) *ExportManager {
+func NewExportManager(config 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)
+	}
+
 	latestLedgerMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: "ledger_exporter", Subsystem: "export_manager", Name: "latest_ledger",
 		Help: "sequence number of the latest ledger consumed by the export manager",
@@ -38,7 +36,7 @@ func NewExportManager(config ExporterConfig, backend ledgerbackend.LedgerBackend
 		ledgerBackend:      backend,
 		queue:              queue,
 		latestLedgerMetric: latestLedgerMetric,
-	}
+	}, nil
 }
 
 // AddLedgerCloseMeta adds ledger metadata to the current export object
@@ -46,10 +44,8 @@ func (e *ExportManager) AddLedgerCloseMeta(ctx context.Context, ledgerCloseMeta
 	ledgerSeq := ledgerCloseMeta.LedgerSequence()
 
 	// Determine the object key for the given ledger sequence
-	objectKey, err := GetObjectKeyFromSequenceNumber(e.config, ledgerSeq)
-	if err != nil {
-		return errors.Wrapf(err, "failed to get object key for ledger %d", ledgerSeq)
-	}
+	objectKey := e.config.GetObjectKeyFromSequenceNumber(ledgerSeq)
+
 	if e.currentMetaArchive != nil && e.currentMetaArchive.GetObjectKey() != objectKey {
 		return errors.New("Current meta archive object key mismatch")
 	}
@@ -67,13 +63,13 @@ func (e *ExportManager) AddLedgerCloseMeta(ctx context.Context, ledgerCloseMeta
 		e.currentMetaArchive = NewLedgerMetaArchive(objectKey, ledgerSeq, endSeq)
 	}
 
-	if err = e.currentMetaArchive.AddLedger(ledgerCloseMeta); err != nil {
+	if err := e.currentMetaArchive.AddLedger(ledgerCloseMeta); err != nil {
 		return errors.Wrapf(err, "failed to add ledger %d", ledgerSeq)
 	}
 
 	if ledgerSeq >= e.currentMetaArchive.GetEndLedgerSequence() {
 		// Current archive is full, send it for upload
-		if err = e.queue.Enqueue(ctx, e.currentMetaArchive); err != nil {
+		if err := e.queue.Enqueue(ctx, e.currentMetaArchive); err != nil {
 			return err
 		}
 		e.currentMetaArchive = nil
diff --git a/exp/services/ledgerexporter/internal/gcs_datastore.go b/exp/services/ledgerexporter/internal/gcs_datastore.go
index 927130fe..c68b7e7a 100644
--- a/exp/services/ledgerexporter/internal/gcs_datastore.go
+++ b/exp/services/ledgerexporter/internal/gcs_datastore.go
@@ -2,16 +2,20 @@ package ledgerexporter
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"path"
+	"strings"
 
 	"google.golang.org/api/googleapi"
+	"google.golang.org/api/option"
 
 	"cloud.google.com/go/storage"
 
 	"github.com/stellar/go/support/errors"
+	"github.com/stellar/go/support/url"
 )
 
 // GCSDataStore implements DataStore for GCS
@@ -21,6 +25,41 @@ type GCSDataStore struct {
 	prefix string
 }
 
+func NewGCSDataStore(ctx context.Context, params map[string]string, network string) (DataStore, error) {
+	destinationBucketPath, ok := params["destination_bucket_path"]
+	if !ok {
+		return nil, errors.Errorf("Invalid GCS config, no destination_bucket_path")
+	}
+
+	// append the gcs:// scheme to enable usage of the url package reliably to
+	// get parse bucket name which is first path segment as URL.Host
+	gcsBucketURL := fmt.Sprintf("gcs://%s/%s", destinationBucketPath, network)
+	parsed, err := url.Parse(gcsBucketURL)
+	if err != nil {
+		return nil, err
+	}
+
+	// Inside gcs, all paths start _without_ the leading /
+	prefix := strings.TrimPrefix(parsed.Path, "/")
+	bucketName := parsed.Host
+
+	logger.Infof("creating GCS client for bucket: %s, prefix: %s", bucketName, prefix)
+
+	var options []option.ClientOption
+	client, err := storage.NewClient(ctx, options...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check the bucket exists
+	bucket := client.Bucket(bucketName)
+	if _, err := bucket.Attrs(ctx); err != nil {
+		return nil, errors.Wrap(err, "failed to retrieve bucket attributes")
+	}
+
+	return &GCSDataStore{client: client, bucket: bucket, prefix: prefix}, 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)
@@ -84,6 +123,11 @@ func (b GCSDataStore) Size(ctx context.Context, pth string) (int64, error) {
 // Exists checks if a file exists in the GCS bucket.
 func (b GCSDataStore) Exists(ctx context.Context, pth string) (bool, error) {
 	_, err := b.Size(ctx, pth)
+
+	if err == os.ErrNotExist {
+		return false, nil
+	}
+
 	return err == nil, err
 }
 
diff --git a/exp/services/ledgerexporter/internal/ledgerbatch_config.go b/exp/services/ledgerexporter/internal/ledgerbatch_config.go
new file mode 100644
index 00000000..6ac3a3f3
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/ledgerbatch_config.go
@@ -0,0 +1,49 @@
+package ledgerexporter
+
+import (
+	"fmt"
+)
+
+const (
+	fileSuffix = ".xdr.gz"
+)
+
+type LedgerBatchConfig struct {
+	LedgersPerFile    uint32 `toml:"ledgers_per_file"`
+	FilesPerPartition uint32 `toml:"files_per_partition"`
+}
+
+func (ec LedgerBatchConfig) GetSequenceNumberStartBoundary(ledgerSeq uint32) uint32 {
+	if ec.LedgersPerFile == 0 {
+		return 0
+	}
+	return (ledgerSeq / ec.LedgersPerFile) * ec.LedgersPerFile
+}
+
+func (ec LedgerBatchConfig) GetSequenceNumberEndBoundary(ledgerSeq uint32) uint32 {
+	return ec.GetSequenceNumberStartBoundary(ledgerSeq) + ec.LedgersPerFile - 1
+}
+
+// GetObjectKeyFromSequenceNumber generates the object key name from the ledger sequence number based on configuration.
+func (ec LedgerBatchConfig) GetObjectKeyFromSequenceNumber(ledgerSeq uint32) string {
+	var objectKey string
+
+	if ec.FilesPerPartition > 1 {
+		partitionSize := ec.LedgersPerFile * ec.FilesPerPartition
+		partitionStart := (ledgerSeq / partitionSize) * partitionSize
+		partitionEnd := partitionStart + partitionSize - 1
+		objectKey = fmt.Sprintf("%d-%d/", partitionStart, partitionEnd)
+	}
+
+	fileStart := ec.GetSequenceNumberStartBoundary(ledgerSeq)
+	fileEnd := ec.GetSequenceNumberEndBoundary(ledgerSeq)
+	objectKey += fmt.Sprintf("%d", fileStart)
+
+	// Multiple ledgers per file
+	if fileStart != fileEnd {
+		objectKey += fmt.Sprintf("-%d", fileEnd)
+	}
+	objectKey += fileSuffix
+
+	return objectKey
+}
diff --git a/exp/services/ledgerexporter/internal/mock_datastore.go b/exp/services/ledgerexporter/internal/mocks.go
similarity index 70%
rename from exp/services/ledgerexporter/internal/mock_datastore.go
rename to exp/services/ledgerexporter/internal/mocks.go
index 705df45a..3f514a8f 100644
--- a/exp/services/ledgerexporter/internal/mock_datastore.go
+++ b/exp/services/ledgerexporter/internal/mocks.go
@@ -34,10 +34,23 @@ func (m *MockDataStore) PutFile(ctx context.Context, path string, in io.WriterTo
 
 func (m *MockDataStore) PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo) (bool, error) {
 	args := m.Called(ctx, path, in)
-	return args.Bool(0), args.Error(1)
+	return args.Get(0).(bool), args.Error(1)
 }
 
 func (m *MockDataStore) Close() error {
 	args := m.Called()
 	return args.Error(0)
 }
+
+type MockResumableManager struct {
+	mock.Mock
+}
+
+func (m *MockResumableManager) FindStart(ctx context.Context, start, end uint32) (absentLedger uint32, ok bool, err error) {
+	a := m.Called(ctx, start, end)
+	return a.Get(0).(uint32), a.Get(1).(bool), a.Error(2)
+}
+
+// ensure that the MockClient implements ClientInterface
+var _ DataStore = &MockDataStore{}
+var _ ResumableManager = &MockResumableManager{}
diff --git a/exp/services/ledgerexporter/internal/resumablemanager.go b/exp/services/ledgerexporter/internal/resumablemanager.go
new file mode 100644
index 00000000..1d832f2e
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/resumablemanager.go
@@ -0,0 +1,129 @@
+package ledgerexporter
+
+import (
+	"context"
+	"sort"
+
+	"github.com/pkg/errors"
+	"github.com/stellar/go/historyarchive"
+)
+
+type ResumableManager interface {
+	// Given a requested ledger range, return the first absent ledger within the
+	// requested range of [start, end].
+	//
+	// start - begin search inclusive from this ledger, must be greater than 0.
+	// end   - stop search inclusive to this ledger.
+	//
+	// If start=0, invalid, error will be returned.
+	//
+	// If end=0, is provided as a convenience, to allow requesting an effectively
+	// dynamic end value for the range, which will be an approximation of the network's
+	// most recent checkpointed ledger + (2 * checkpoint_frequency).
+	//
+	// return:
+	// absentLedger      - will be non-zero, the oldest ledger sequence between range of [start, end]
+	//                     which is not populated on data store.
+	// ok                - if true, 'absentLedger' has a usable non-zero value, if false, there is no absent ledger in the requested range and 'absentLedger' is set to zero.
+	// err               - the search was cancelled due to this unexpected error, 'absentLedger' and 'ok' return values should be ignored.
+	//
+	// When no error, the two return values will compose the following truth table:
+	//    1. datastore had no data in the requested range: absentLedger={start}, ok=true
+	//    2. datastore had partial data in the requested range: absentLedger={a value > start and <= end}, ok=true
+	//    3. datastore had all data in the requested range: absentLedger=0, ok=false
+	FindStart(ctx context.Context, start, end uint32) (absentLedger uint32, ok bool, err error)
+}
+
+type resumableManagerService struct {
+	network           string
+	ledgerBatchConfig LedgerBatchConfig
+	dataStore         DataStore
+	archive           historyarchive.ArchiveInterface
+}
+
+func NewResumableManager(dataStore DataStore,
+	network string,
+	ledgerBatchConfig LedgerBatchConfig,
+	archive historyarchive.ArchiveInterface) ResumableManager {
+	return &resumableManagerService{
+		ledgerBatchConfig: ledgerBatchConfig,
+		network:           network,
+		dataStore:         dataStore,
+		archive:           archive,
+	}
+}
+
+func (rm resumableManagerService) FindStart(ctx context.Context, start, end uint32) (absentLedger uint32, ok bool, err error) {
+	if start < 1 {
+		return 0, false, errors.New("Invalid start value, must be greater than zero")
+	}
+
+	log := logger.WithField("start", start).WithField("end", end).WithField("network", rm.network)
+
+	networkLatest := uint32(0)
+	if end < 1 {
+		var latestErr error
+		networkLatest, latestErr = getLatestLedgerSequenceFromHistoryArchives(rm.archive)
+		if latestErr != nil {
+			err := errors.Wrap(latestErr, "Resumability of requested export ledger range, was not able to get latest ledger from network")
+			return 0, false, err
+		}
+		networkLatest = networkLatest + (getHistoryArchivesCheckPointFrequency() * 2)
+		logger.Infof("Resumability computed effective latest network ledger including padding of checkpoint frequency to be %d + for network=%v", networkLatest, rm.network)
+
+		if start > networkLatest {
+			// requested to start at a point beyond the latest network, resume not applicable.
+			return 0, false, errors.Errorf("Invalid start value of %v, it is greater than network's latest ledger of %v", start, networkLatest)
+		}
+		end = networkLatest
+	}
+
+	rangeSize := max(int(end-start), 1)
+	var binarySearchError error
+	lowestAbsentIndex := sort.Search(rangeSize, binarySearchCallbackFn(&rm, ctx, start, end, &binarySearchError))
+	if binarySearchError != nil {
+		return 0, false, binarySearchError
+	}
+
+	if lowestAbsentIndex < int(rangeSize) {
+		nearestAbsentLedgerSequence := start + uint32(lowestAbsentIndex)
+		log.Infof("Resumability determined next absent object start key of %d for requested export ledger range", nearestAbsentLedgerSequence)
+		return nearestAbsentLedgerSequence, true, nil
+	}
+
+	// unbounded, and datastore had up to latest network, return that as staring point.
+	if networkLatest > 0 {
+		return networkLatest, true, nil
+	}
+
+	// data store had all ledgers for requested range, no resumability needed.
+	log.Infof("Resumability found no absent object keys in requested ledger range")
+	return 0, false, nil
+}
+
+func binarySearchCallbackFn(rm *resumableManagerService, ctx context.Context, start, end uint32, binarySearchError *error) func(ledgerSequence int) bool {
+	lookupCache := map[string]bool{}
+
+	return func(binarySearchIndex int) bool {
+		if *binarySearchError != nil {
+			// an error has already occured in a callback for the same binary search, exiting
+			return true
+		}
+		objectKeyMiddle := rm.ledgerBatchConfig.GetObjectKeyFromSequenceNumber(start + uint32(binarySearchIndex))
+
+		// there may be small occurrence of repeated queries on same object key once
+		// search narrows down to a range that fits within the ledgers per file
+		// worst case being 'log of ledgers_per_file' queries.
+		middleFoundOnStore, foundInCache := lookupCache[objectKeyMiddle]
+		if !foundInCache {
+			var datastoreErr error
+			middleFoundOnStore, datastoreErr = rm.dataStore.Exists(ctx, objectKeyMiddle)
+			if datastoreErr != nil {
+				*binarySearchError = errors.Wrapf(datastoreErr, "While searching datastore for resumability within export ledger range start=%d, end=%d, was not able to check if object key %v exists on data store", start, end, objectKeyMiddle)
+				return true
+			}
+			lookupCache[objectKeyMiddle] = middleFoundOnStore
+		}
+		return !middleFoundOnStore
+	}
+}
diff --git a/exp/services/ledgerexporter/internal/utils.go b/exp/services/ledgerexporter/internal/utils.go
deleted file mode 100644
index 8b556c58..00000000
--- a/exp/services/ledgerexporter/internal/utils.go
+++ /dev/null
@@ -1,105 +0,0 @@
-package ledgerexporter
-
-import (
-	"compress/gzip"
-	"fmt"
-	"io"
-
-	xdr3 "github.com/stellar/go-xdr/xdr3"
-
-	"github.com/stellar/go/historyarchive"
-	"github.com/stellar/go/support/errors"
-	"github.com/stellar/go/support/storage"
-)
-
-const (
-	fileSuffix = ".xdr.gz"
-)
-
-// GetObjectKeyFromSequenceNumber generates the file name from the ledger sequence number based on configuration.
-func GetObjectKeyFromSequenceNumber(config ExporterConfig, ledgerSeq uint32) (string, error) {
-	var objectKey string
-
-	if config.LedgersPerFile < 1 {
-		return "", errors.Errorf("Invalid ledgers per file (%d): must be at least 1", config.LedgersPerFile)
-	}
-
-	if config.FilesPerPartition > 1 {
-		partitionSize := config.LedgersPerFile * config.FilesPerPartition
-		partitionStart := (ledgerSeq / partitionSize) * partitionSize
-		partitionEnd := partitionStart + partitionSize - 1
-		objectKey = fmt.Sprintf("%d-%d/", partitionStart, partitionEnd)
-	}
-
-	fileStart := (ledgerSeq / config.LedgersPerFile) * config.LedgersPerFile
-	fileEnd := fileStart + config.LedgersPerFile - 1
-	objectKey += fmt.Sprintf("%d", fileStart)
-
-	// Multiple ledgers per file
-	if fileStart != fileEnd {
-		objectKey += fmt.Sprintf("-%d", fileEnd)
-	}
-	objectKey += fileSuffix
-
-	return objectKey, nil
-}
-
-// getLatestLedgerSequenceFromHistoryArchives returns the most recent ledger sequence (checkpoint ledger)
-// number present in the history archives.
-func getLatestLedgerSequenceFromHistoryArchives(historyArchivesURLs []string) (uint32, error) {
-	for _, historyArchiveURL := range historyArchivesURLs {
-		ha, err := historyarchive.Connect(
-			historyArchiveURL,
-			historyarchive.ArchiveOptions{
-				ConnectOptions: storage.ConnectOptions{
-					UserAgent: "ledger-exporter",
-				},
-			},
-		)
-		if err != nil {
-			logger.WithError(err).Warnf("Error connecting to history archive %s", historyArchiveURL)
-			continue // Skip to next archive
-		}
-
-		has, err := ha.GetRootHAS()
-		if err != nil {
-			logger.WithError(err).Warnf("Error getting RootHAS for %s", historyArchiveURL)
-			continue // Skip to next archive
-		}
-
-		return has.CurrentLedger, nil
-	}
-
-	return 0, errors.New("failed to retrieve the latest ledger sequence from any history archive")
-}
-
-type XDRGzipEncoder struct {
-	XdrPayload interface{}
-}
-
-func (g *XDRGzipEncoder) WriteTo(w io.Writer) (int64, error) {
-	gw := gzip.NewWriter(w)
-	n, err := xdr3.Marshal(gw, g.XdrPayload)
-	if err != nil {
-		return int64(n), err
-	}
-	return int64(n), gw.Close()
-}
-
-type XDRGzipDecoder struct {
-	XdrPayload interface{}
-}
-
-func (d *XDRGzipDecoder) ReadFrom(r io.Reader) (int64, error) {
-	gr, err := gzip.NewReader(r)
-	if err != nil {
-		return 0, err
-	}
-	defer gr.Close()
-
-	n, err := xdr3.Unmarshal(gr, d.XdrPayload)
-	if err != nil {
-		return int64(n), err
-	}
-	return int64(n), nil
-}
diff --git a/exp/services/ledgerexporter/main.go b/exp/services/ledgerexporter/main.go
index f1a81e95..63e09498 100644
--- a/exp/services/ledgerexporter/main.go
+++ b/exp/services/ledgerexporter/main.go
@@ -1,8 +1,25 @@
 package main
 
-import exporter "github.com/stellar/go/exp/services/ledgerexporter/internal"
+import (
+	"flag"
+
+	exporter "github.com/stellar/go/exp/services/ledgerexporter/internal"
+)
 
 func main() {
-	app := exporter.NewApp()
+	flags := exporter.Flags{}
+	startLedger := uint(0)
+	endLedger := uint(0)
+	flag.UintVar(&startLedger, "start", 0, "Starting ledger")
+	flag.UintVar(&endLedger, "end", 0, "Ending ledger (inclusive)")
+	flag.StringVar(&flags.ConfigFilePath, "config-file", "config.toml", "Path to the TOML config file")
+	flag.BoolVar(&flags.Resume, "resume", false, "Attempt to find a resumable starting point on remote data store")
+	flag.UintVar(&flags.AdminPort, "admin-port", 0, "Admin HTTP port for prometheus metrics")
+
+	flag.Parse()
+	flags.StartLedger = uint32(startLedger)
+	flags.EndLedger = uint32(endLedger)
+
+	app := exporter.NewApp(flags)
 	app.Run()
 }
diff --git a/go.mod b/go.mod
index f45b68de..fb02c945 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ toolchain go1.22.1
 
 require (
 	cloud.google.com/go/firestore v1.14.0 // indirect
-	cloud.google.com/go/storage v1.37.0
+	cloud.google.com/go/storage v1.40.0
 	firebase.google.com/go v3.12.0+incompatible
 	github.com/2opremio/pretty v0.2.2-0.20230601220618-e1d5758b2a95
 	github.com/BurntSushi/toml v1.3.2
@@ -22,7 +22,7 @@ require (
 	github.com/go-chi/chi v4.1.2+incompatible
 	github.com/go-errors/errors v1.5.1
 	github.com/golang-jwt/jwt v3.2.2+incompatible
-	github.com/google/uuid v1.5.0
+	github.com/google/uuid v1.6.0
 	github.com/gorilla/schema v1.2.0
 	github.com/graph-gophers/graphql-go v1.3.0
 	github.com/guregu/null v4.0.0+incompatible
@@ -52,7 +52,7 @@ require (
 	github.com/stretchr/testify v1.8.4
 	github.com/tyler-smith/go-bip39 v0.0.0-20180618194314-52158e4697b8
 	github.com/xdrpp/goxdr v0.1.1
-	google.golang.org/api v0.157.0
+	google.golang.org/api v0.170.0
 	gopkg.in/gavv/httpexpect.v1 v1.0.0-20170111145843-40724cf1e4a0
 	gopkg.in/square/go-jose.v2 v2.4.1
 	gopkg.in/tylerb/graceful.v1 v1.2.15
@@ -64,10 +64,10 @@ require (
 )
 
 require (
-	cloud.google.com/go/compute v1.23.3 // indirect
+	cloud.google.com/go/compute v1.24.0 // indirect
 	cloud.google.com/go/compute/metadata v0.2.3 // indirect
-	cloud.google.com/go/iam v1.1.5 // indirect
-	cloud.google.com/go/longrunning v0.5.4 // indirect
+	cloud.google.com/go/iam v1.1.7 // indirect
+	cloud.google.com/go/longrunning v0.5.5 // indirect
 	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/certifi/gocertifi v0.0.0-20210507211836-431795d63e8d // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
@@ -75,7 +75,7 @@ require (
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/go-gorp/gorp/v3 v3.1.0 // indirect
-	github.com/go-logr/logr v1.3.0 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gobuffalo/packd v1.0.2 // indirect
 	github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 // indirect
@@ -90,23 +90,23 @@ require (
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.10.0 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect
-	go.opentelemetry.io/otel v1.21.0 // indirect
-	go.opentelemetry.io/otel/metric v1.21.0 // indirect
-	go.opentelemetry.io/otel/trace v1.21.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
+	go.opentelemetry.io/otel v1.24.0 // indirect
+	go.opentelemetry.io/otel/metric v1.24.0 // indirect
+	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/mod v0.13.0 // indirect
 	golang.org/x/tools v0.14.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240122161410-6c6643bf1457 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240314234333-6e1732d8331c // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240311132316-a219d84964c2 // indirect
 	gopkg.in/djherbis/atime.v1 v1.0.0 // indirect
 	gopkg.in/djherbis/stream.v1 v1.3.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
 
 require (
-	cloud.google.com/go v0.112.0 // indirect
+	cloud.google.com/go v0.112.1 // indirect
 	github.com/ajg/form v0.0.0-20160822230020-523a5da1a92f // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/buger/goreplay v1.3.2
@@ -114,10 +114,10 @@ require (
 	github.com/fatih/structs v1.0.0 // indirect
 	github.com/gavv/monotime v0.0.0-20161010190848-47d58efa6955 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/go-querystring v0.0.0-20160401233042-9235644dd9e5 // indirect
-	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
+	github.com/googleapis/gax-go/v2 v2.12.3 // indirect
 	github.com/hashicorp/golang-lru v1.0.2
 	github.com/imkira/go-interpol v1.1.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
@@ -150,14 +150,14 @@ require (
 	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 	golang.org/x/net v0.23.0 // indirect
-	golang.org/x/oauth2 v0.16.0 // indirect
+	golang.org/x/oauth2 v0.18.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/term v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.5.0
 	google.golang.org/appengine v1.6.8 // indirect
-	google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac // indirect
-	google.golang.org/grpc v1.60.1 // indirect
+	google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect
+	google.golang.org/grpc v1.62.1 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
diff --git a/go.sum b/go.sum
index 2ea7e9a0..6d8713de 100644
--- a/go.sum
+++ b/go.sum
@@ -17,26 +17,26 @@ cloud.google.com/go v0.65.0/go.mod h1:O5N8zS7uWy9vkA9vayVHs65eM1ubvY4h553ofrNHOb
 cloud.google.com/go v0.72.0/go.mod h1:M+5Vjvlc2wnp6tjzE102Dw08nGShTscUx2nZMufOKPI=
 cloud.google.com/go v0.74.0/go.mod h1:VV1xSbzvo+9QJOxLDaJfTjx5e+MePCpCWwvftOeQmWk=
 cloud.google.com/go v0.75.0/go.mod h1:VGuuCn7PG0dwsd5XPVm2Mm3wlh3EL55/79EKB6hlPTY=
-cloud.google.com/go v0.112.0 h1:tpFCD7hpHFlQ8yPwT3x+QeXqc2T6+n6T+hmABHfDUSM=
-cloud.google.com/go v0.112.0/go.mod h1:3jEEVwZ/MHU4djK5t5RHuKOA/GbLddgTdVubX1qnPD4=
+cloud.google.com/go v0.112.1 h1:uJSeirPke5UNZHIb4SxfZklVSiWWVqW4oXlETwZziwM=
+cloud.google.com/go v0.112.1/go.mod h1:+Vbu+Y1UU+I1rjmzeMOb/8RfkKJK2Gyxi1X6jJCZLo4=
 cloud.google.com/go/bigquery v1.0.1/go.mod h1:i/xbL2UlR5RvWAURpBYZTtm/cXjCha9lbfbpx4poX+o=
 cloud.google.com/go/bigquery v1.3.0/go.mod h1:PjpwJnslEMmckchkHFfq+HTD2DmtT67aNFKH1/VBDHE=
 cloud.google.com/go/bigquery v1.4.0/go.mod h1:S8dzgnTigyfTmLBfrtrhyYhwRxG72rYxvftPBK2Dvzc=
 cloud.google.com/go/bigquery v1.5.0/go.mod h1:snEHRnqQbz117VIFhE8bmtwIDY80NLUZUMb4Nv6dBIg=
 cloud.google.com/go/bigquery v1.7.0/go.mod h1://okPTzCYNXSlb24MZs83e2Do+h+VXtc4gLoIoXIAPc=
 cloud.google.com/go/bigquery v1.8.0/go.mod h1:J5hqkt3O0uAFnINi6JXValWIb1v0goeZM77hZzJN/fQ=
-cloud.google.com/go/compute v1.23.3 h1:6sVlXXBmbd7jNX0Ipq0trII3e4n1/MsADLK6a+aiVlk=
-cloud.google.com/go/compute v1.23.3/go.mod h1:VCgBUoMnIVIR0CscqQiPJLAG25E3ZRZMzcFZeQ+h8CI=
+cloud.google.com/go/compute v1.24.0 h1:phWcR2eWzRJaL/kOiJwfFsPs4BaKq1j6vnpZrc1YlVg=
+cloud.google.com/go/compute v1.24.0/go.mod h1:kw1/T+h/+tK2LJK0wiPPx1intgdAM3j/g3hFDlscY40=
 cloud.google.com/go/compute/metadata v0.2.3 h1:mg4jlk7mCAj6xXp9UJ4fjI9VUI5rubuGBW5aJ7UnBMY=
 cloud.google.com/go/compute/metadata v0.2.3/go.mod h1:VAV5nSsACxMJvgaAuX6Pk2AawlZn8kiOGuCv6gTkwuA=
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
 cloud.google.com/go/firestore v1.14.0 h1:8aLcKnMPoldYU3YHgu4t2exrKhLQkqaXAGqT0ljrFVw=
 cloud.google.com/go/firestore v1.14.0/go.mod h1:96MVaHLsEhbvkBEdZgfN+AS/GIkco1LRpH9Xp9YZfzQ=
-cloud.google.com/go/iam v1.1.5 h1:1jTsCu4bcsNsE4iiqNT5SHwrDRCfRmIaaaVFhRveTJI=
-cloud.google.com/go/iam v1.1.5/go.mod h1:rB6P/Ic3mykPbFio+vo7403drjlgvoWfYpJhMXEbzv8=
-cloud.google.com/go/longrunning v0.5.4 h1:w8xEcbZodnA2BbW6sVirkkoC+1gP8wS57EUUgGS0GVg=
-cloud.google.com/go/longrunning v0.5.4/go.mod h1:zqNVncI0BOP8ST6XQD1+VcvuShMmq7+xFSzOL++V0dI=
+cloud.google.com/go/iam v1.1.7 h1:z4VHOhwKLF/+UYXAJDFwGtNF0b6gjsW1Pk9Ml0U/IoM=
+cloud.google.com/go/iam v1.1.7/go.mod h1:J4PMPg8TtyurAUvSmPj8FF3EDgY1SPRZxcUGrn7WXGA=
+cloud.google.com/go/longrunning v0.5.5 h1:GOE6pZFdSrTb4KAiKnXsJBtlE6mEyaW44oKyMILWnOg=
+cloud.google.com/go/longrunning v0.5.5/go.mod h1:WV2LAxD8/rg5Z1cNW6FJ/ZpX4E4VnDnoTk0yawPBB7s=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
 cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
@@ -47,8 +47,8 @@ cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohl
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
 cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3fOKtUw0Xmo=
-cloud.google.com/go/storage v1.37.0 h1:WI8CsaFO8Q9KjPVtsZ5Cmi0dXV25zMoX0FklT7c3Jm4=
-cloud.google.com/go/storage v1.37.0/go.mod h1:i34TiT2IhiNDmcj65PqwCjcoUX7Z5pLzS8DEmoiFq1k=
+cloud.google.com/go/storage v1.40.0 h1:VEpDQV5CJxFmJ6ueWNsKxcr1QAYOXEgxDa+sBbJahPw=
+cloud.google.com/go/storage v1.40.0/go.mod h1:Rrj7/hKlG87BLqDJYtwR0fbPld8uJPbQ2ucUMY7Ir0g=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 firebase.google.com/go v3.12.0+incompatible h1:q70KCp/J0oOL8kJ8oV2j3646kV4TB8Y5IvxXC0WT1bo=
 firebase.google.com/go v3.12.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIwjt8toICdV5Wh9ptHs=
@@ -96,8 +96,6 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
-github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4 h1:/inchEIKaYC1Akx+H+gqO04wryn5h75LSazbRlnya1k=
-github.com/cncf/xds/go v0.0.0-20230607035331-e9ce68804cb4/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creachadair/jrpc2 v1.1.0 h1:SgpJf0v1rVCZx68+4APv6dgsTFsIHlpgFD1NlQAWA0A=
 github.com/creachadair/jrpc2 v1.1.0/go.mod h1:5jN7MKwsm8qvgfTsTzLX3JIfidsAkZ1c8DZSQmp+g38=
@@ -121,8 +119,6 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m
 github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-github.com/envoyproxy/protoc-gen-validate v1.0.2 h1:QkIBuU5k+x7/QXPvPPnWXWlCdaBFApVqftFV6k087DA=
-github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE=
 github.com/fatih/structs v1.0.0 h1:BrX964Rv5uQ3wwS+KRUAJCBBw5PQmgJfJ6v4yly5QwU=
 github.com/fatih/structs v1.0.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
@@ -150,8 +146,8 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2
 github.com/go-gorp/gorp/v3 v3.1.0 h1:ItKF/Vbuj31dmV4jxA1qblpSwkl9g1typ24xoe70IGs=
 github.com/go-gorp/gorp/v3 v3.1.0/go.mod h1:dLEjIyyRNiXvNZ8PSmzpt1GsWAUK8kjVhEpjH8TixEw=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY=
-github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
+github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
@@ -197,8 +193,8 @@ github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
-github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
@@ -240,14 +236,14 @@ github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm4
 github.com/google/s2a-go v0.1.7 h1:60BLSyTrOV4/haCDW4zb1guZItoSq8foHCXrAnjBo/o=
 github.com/google/s2a-go v0.1.7/go.mod h1:50CgR4k1jNlWBu4UfS4AcfhVe1r6pdZPygJ3R8F0Qdw=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
-github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/enterprise-certificate-proxy v0.3.2 h1:Vie5ybvEvT75RniqhfFxPRy3Bf7vr3h0cechB90XaQs=
 github.com/googleapis/enterprise-certificate-proxy v0.3.2/go.mod h1:VLSiSSBs/ksPL8kq3OBOQ6WRI2QnaFynd1DCjZ62+V0=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
-github.com/googleapis/gax-go/v2 v2.12.0 h1:A+gCJKdRfqXkr+BIRGtZLibNXf0m1f9E4HG56etFpas=
-github.com/googleapis/gax-go/v2 v2.12.0/go.mod h1:y+aIqrI5eb1YGMVJfuV3185Ts/D7qKpsEkdD5+I6QGU=
+github.com/googleapis/gax-go/v2 v2.12.3 h1:5/zPPDvw8Q1SuXjrqrZslrqT7dL/uJT2CQii/cLCKqA=
+github.com/googleapis/gax-go/v2 v2.12.3/go.mod h1:AKloxT6GtNbaLm8QTNSidHUVsHYcBHwWRvkNFJUQcS4=
 github.com/googleapis/google-cloud-go-testing v0.0.0-20200911160855-bcd43fbb19e8/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
@@ -478,18 +474,18 @@ go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 h1:SpGay3w+nEwMpfVnbqOLH5gY52/foP8RE8UzTZ1pdSE=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1/go.mod h1:4UoMYEZOC0yN/sPGH76KPkkU7zgiEWYWL9vwmbnTJPE=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 h1:aFJWCqJMNjENlcleuuOkGAPH82y0yULBScfXcIEdS24=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1/go.mod h1:sEGXWArGqc3tVa+ekntsN65DmVbVeW+7lTKTjZF3/Fo=
-go.opentelemetry.io/otel v1.21.0 h1:hzLeKBZEL7Okw2mGzZ0cc4k/A7Fta0uoPgaJCr8fsFc=
-go.opentelemetry.io/otel v1.21.0/go.mod h1:QZzNPQPm1zLX4gZK4cMi+71eaorMSGT3A4znnUvNNEo=
-go.opentelemetry.io/otel/metric v1.21.0 h1:tlYWfeo+Bocx5kLEloTjbcDwBuELRrIFxwdQ36PlJu4=
-go.opentelemetry.io/otel/metric v1.21.0/go.mod h1:o1p3CA8nNHW8j5yuQLdc1eeqEaPfzug24uvsyIEJRWM=
-go.opentelemetry.io/otel/sdk v1.21.0 h1:FTt8qirL1EysG6sTQRZ5TokkU8d0ugCj8htOgThZXQ8=
-go.opentelemetry.io/otel/sdk v1.21.0/go.mod h1:Nna6Yv7PWTdgJHVRD9hIYywQBRx7pbox6nwBnZIxl/E=
-go.opentelemetry.io/otel/trace v1.21.0 h1:WD9i5gzvoUPuXIXH24ZNBudiarZDKuekPqi/E8fpfLc=
-go.opentelemetry.io/otel/trace v1.21.0/go.mod h1:LGbsEB0f9LGjN+OZaQQ26sohbOmiMR+BaslueVtS/qQ=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 h1:4Pp6oUg3+e/6M4C0A/3kJ2VYa++dsWVTtGgLVj5xtHg=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0/go.mod h1:Mjt1i1INqiaoZOMGR1RIUJN+i3ChKoFRqzrRQhlkbs0=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 h1:jq9TW8u3so/bN+JPT166wjOI6/vQPF6Xe7nMNIltagk=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0/go.mod h1:p8pYQP+m5XfbZm9fxtSKAbM6oIllS7s2AfxrChvc7iw=
+go.opentelemetry.io/otel v1.24.0 h1:0LAOdjNmQeSTzGBzduGe/rU4tZhMwL5rWgtp9Ku5Jfo=
+go.opentelemetry.io/otel v1.24.0/go.mod h1:W7b9Ozg4nkF5tWI5zsXkaKKDjdVjpD4oAt9Qi/MArHo=
+go.opentelemetry.io/otel/metric v1.24.0 h1:6EhoGWWK28x1fbpA4tYTOWBkPefTDQnb8WSGXlc88kI=
+go.opentelemetry.io/otel/metric v1.24.0/go.mod h1:VYhLe1rFfxuTXLgj4CBiyz+9WYBA8pNGJgDcSFRKBco=
+go.opentelemetry.io/otel/sdk v1.22.0 h1:6coWHw9xw7EfClIC/+O31R8IY3/+EiRFHevmHafB2Gw=
+go.opentelemetry.io/otel/sdk v1.22.0/go.mod h1:iu7luyVGYovrRpe2fmj3CVKouQNdTOkxtLzPvPz1DOc=
+go.opentelemetry.io/otel/trace v1.24.0 h1:CsKnnL4dUAr/0llH9FKuc698G04IrpWV0MQA/Y1YELI=
+go.opentelemetry.io/otel/trace v1.24.0/go.mod h1:HPc3Xr/cOApsBI154IU0OI0HJexz+aw5uPdbs3UCjNU=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -591,8 +587,8 @@ golang.org/x/oauth2 v0.0.0-20200902213428-5d25da1a8d43/go.mod h1:KelEdhl1UZF7XfJ
 golang.org/x/oauth2 v0.0.0-20201109201403-9fd604954f58/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
 golang.org/x/oauth2 v0.0.0-20201208152858-08078c50e5b5/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
 golang.org/x/oauth2 v0.0.0-20210218202405-ba52d332ba99/go.mod h1:KelEdhl1UZF7XfJ4dDtk6s++YSgaE7mD/BuKKDLBl4A=
-golang.org/x/oauth2 v0.16.0 h1:aDkGMBSYxElaoP81NpoUoz2oo2R2wHdZpGToUxfyQrQ=
-golang.org/x/oauth2 v0.16.0/go.mod h1:hqZ+0LWXsiVoZpeld6jVt06P3adbS2Uu911W1SsJv2o=
+golang.org/x/oauth2 v0.18.0 h1:09qnuIAgzdx1XplqJvW6CQqMCtGZykZWcXzPMPUusvI=
+golang.org/x/oauth2 v0.18.0/go.mod h1:Wf7knwG0MPoWIMMBgFlEaSUDaKskp0dCfrlJRJXbBi8=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -756,8 +752,8 @@ google.golang.org/api v0.30.0/go.mod h1:QGmEvQ87FHZNiUVJkT14jQNYJ4ZJjdRF23ZXz513
 google.golang.org/api v0.35.0/go.mod h1:/XrVsuzM0rZmrsbjJutiuftIzeuTQcEeaYcSk/mQ1dg=
 google.golang.org/api v0.36.0/go.mod h1:+z5ficQTmoYpPn8LCUNVpK5I7hwkpjbcgqA7I34qYtE=
 google.golang.org/api v0.40.0/go.mod h1:fYKFpnQN0DsDSKRVRcQSDQNtqWPfM9i+zNPxepjRCQ8=
-google.golang.org/api v0.157.0 h1:ORAeqmbrrozeyw5NjnMxh7peHO0UzV4wWYSwZeCUb20=
-google.golang.org/api v0.157.0/go.mod h1:+z4v4ufbZ1WEpld6yMGHyggs+PmAHiaLNj5ytP3N01g=
+google.golang.org/api v0.170.0 h1:zMaruDePM88zxZBG+NG8+reALO2rfLhe/JShitLyT48=
+google.golang.org/api v0.170.0/go.mod h1:/xql9M2btF85xac/VAm4PsLMTLVGUOpq4BE9R8jyNy8=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -803,12 +799,12 @@ google.golang.org/genproto v0.0.0-20201210142538-e3217bee35cc/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20201214200347-8c77b98c765d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210108203827-ffc7fda8c3d7/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210226172003-ab064af71705/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
-google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac h1:ZL/Teoy/ZGnzyrqK/Optxxp2pmVh+fmJ97slxSRyzUg=
-google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac/go.mod h1:+Rvu7ElI+aLzyDQhpHMFMMltsD6m7nqpuWDd2CwJw3k=
-google.golang.org/genproto/googleapis/api v0.0.0-20240122161410-6c6643bf1457 h1:KHBtwE+eQc3+NxpjmRFlQ3pJQ2FNnhhgB9xOV8kyBuU=
-google.golang.org/genproto/googleapis/api v0.0.0-20240122161410-6c6643bf1457/go.mod h1:4jWUdICTdgc3Ibxmr8nAJiiLHwQBY0UI0XZcEMaFKaA=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac h1:nUQEQmH/csSvFECKYRv6HWEyypysidKl2I6Qpsglq/0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac/go.mod h1:daQN87bsDqDoe316QbbvX60nMoJQa4r6Ds0ZuoAe5yA=
+google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 h1:9+tzLLstTlPTRyJTh+ah5wIMsBW5c4tQwGTN3thOW9Y=
+google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9/go.mod h1:mqHbVIp48Muh7Ywss/AD6I5kNVKZMmAa/QEW58Gxp2s=
+google.golang.org/genproto/googleapis/api v0.0.0-20240314234333-6e1732d8331c h1:kaI7oewGK5YnVwj+Y+EJBO/YN1ht8iTL9XkFHtVZLsc=
+google.golang.org/genproto/googleapis/api v0.0.0-20240314234333-6e1732d8331c/go.mod h1:VQW3tUculP/D4B+xVCo+VgSq8As6wA9ZjHl//pmk+6s=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240311132316-a219d84964c2 h1:9IZDv+/GcI6u+a4jRFRLxQs0RUCfavGfoOgEW6jpkI0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240311132316-a219d84964c2/go.mod h1:UCOku4NytXMJuLQE5VuqA5lX3PcHCBo8pxNyvkf4xBs=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
@@ -825,8 +821,8 @@ google.golang.org/grpc v1.31.1/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
 google.golang.org/grpc v1.34.0/go.mod h1:WotjhfgOW/POjDeRt8vscBtXq+2VjORFy659qA51WJ8=
 google.golang.org/grpc v1.35.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.60.1 h1:26+wFr+cNqSGFcOXcabYC0lUVJVRa2Sb2ortSK7VrEU=
-google.golang.org/grpc v1.60.1/go.mod h1:OlCHIeLYqSSsLi6i49B5QGdzaMZK9+M7LXN2FKz4eGM=
+google.golang.org/grpc v1.62.1 h1:B4n+nfKzOICUXMgyrNd19h/I9oH0L1pizfk1d4zSgTk=
+google.golang.org/grpc v1.62.1/go.mod h1:IWTG0VlJLCh1SkC58F7np9ka9mx/WNkjl4PGJaiq+QE=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=


Test patch:
diff --git a/exp/services/ledgerexporter/internal/app_test.go b/exp/services/ledgerexporter/internal/app_test.go
new file mode 100644
index 00000000..fc63c3a3
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/app_test.go
@@ -0,0 +1,114 @@
+package ledgerexporter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestApplyResumeHasStartError(t *testing.T) {
+	ctx := context.Background()
+	app := &App{}
+	app.config = &Config{StartLedger: 10, EndLedger: 19, Resume: true}
+	mockResumableManager := &MockResumableManager{}
+	mockResumableManager.On("FindStart", ctx, uint32(10), uint32(19)).Return(uint32(0), false, errors.New("start error")).Once()
+
+	err := app.applyResumability(ctx, mockResumableManager)
+	require.ErrorContains(t, err, "start error")
+	mockResumableManager.AssertExpectations(t)
+}
+
+func TestApplyResumeDatastoreComplete(t *testing.T) {
+	ctx := context.Background()
+	app := &App{}
+	app.config = &Config{StartLedger: 10, EndLedger: 19, Resume: true}
+	mockResumableManager := &MockResumableManager{}
+	mockResumableManager.On("FindStart", ctx, uint32(10), uint32(19)).Return(uint32(0), false, nil).Once()
+
+	var alreadyExported *DataAlreadyExportedError
+	err := app.applyResumability(ctx, mockResumableManager)
+	require.ErrorAs(t, err, &alreadyExported)
+	mockResumableManager.AssertExpectations(t)
+}
+
+func TestApplyResumeInvalidDataStoreLedgersPerFileBoundary(t *testing.T) {
+	ctx := context.Background()
+	app := &App{}
+	app.config = &Config{
+		StartLedger:       3,
+		EndLedger:         9,
+		Resume:            true,
+		LedgerBatchConfig: LedgerBatchConfig{LedgersPerFile: 10, FilesPerPartition: 50},
+	}
+	mockResumableManager := &MockResumableManager{}
+	// simulate the datastore has inconsistent data,
+	// with last ledger not aligned to starting boundary
+	mockResumableManager.On("FindStart", ctx, uint32(3), uint32(9)).Return(uint32(6), true, nil).Once()
+
+	var invalidStore *InvalidDataStoreError
+	err := app.applyResumability(ctx, mockResumableManager)
+	require.ErrorAs(t, err, &invalidStore)
+	mockResumableManager.AssertExpectations(t)
+}
+
+func TestApplyResumeWithPartialRemoteDataPresent(t *testing.T) {
+	ctx := context.Background()
+	app := &App{}
+	app.config = &Config{
+		StartLedger:       10,
+		EndLedger:         99,
+		Resume:            true,
+		LedgerBatchConfig: LedgerBatchConfig{LedgersPerFile: 10, FilesPerPartition: 50},
+	}
+	mockResumableManager := &MockResumableManager{}
+	// simulates a data store that had ledger files populated up to seq=49, so the first absent ledger would be 50
+	mockResumableManager.On("FindStart", ctx, uint32(10), uint32(99)).Return(uint32(50), true, nil).Once()
+
+	err := app.applyResumability(ctx, mockResumableManager)
+	require.NoError(t, err)
+	require.Equal(t, app.config.StartLedger, uint32(50))
+	mockResumableManager.AssertExpectations(t)
+}
+
+func TestApplyResumeWithNoRemoteDataPresent(t *testing.T) {
+	ctx := context.Background()
+	app := &App{}
+	app.config = &Config{
+		StartLedger:       10,
+		EndLedger:         99,
+		Resume:            true,
+		LedgerBatchConfig: LedgerBatchConfig{LedgersPerFile: 10, FilesPerPartition: 50},
+	}
+	mockResumableManager := &MockResumableManager{}
+	// simulates a data store that had no data in the requested range
+	mockResumableManager.On("FindStart", ctx, uint32(10), uint32(99)).Return(uint32(2), true, nil).Once()
+
+	err := app.applyResumability(ctx, mockResumableManager)
+	require.NoError(t, err)
+	require.Equal(t, app.config.StartLedger, uint32(2))
+	mockResumableManager.AssertExpectations(t)
+}
+
+func TestApplyResumeWithNoRemoteDataAndRequestFromGenesis(t *testing.T) {
+	// app will coerce config.StartLedger values less than 2 to a min of 2 before applying resumability FindStart
+	// app will validate the response from FindStart to ensure datastore is ledgers-per-file aligned
+	// config.StartLedger=2 is a special genesis case that shouldn't trigger ledgers-per-file validation error
+	ctx := context.Background()
+	app := &App{}
+	app.config = &Config{
+		StartLedger:       2,
+		EndLedger:         99,
+		Resume:            true,
+		LedgerBatchConfig: LedgerBatchConfig{LedgersPerFile: 10, FilesPerPartition: 50},
+	}
+	mockResumableManager := &MockResumableManager{}
+	// simulates a data store that had no data in the requested range
+	mockResumableManager.On("FindStart", ctx, uint32(2), uint32(99)).Return(uint32(2), true, nil).Once()
+
+	err := app.applyResumability(ctx, mockResumableManager)
+	require.NoError(t, err)
+	require.Equal(t, app.config.StartLedger, uint32(2))
+	mockResumableManager.AssertExpectations(t)
+}
diff --git a/exp/services/ledgerexporter/internal/config_test.go b/exp/services/ledgerexporter/internal/config_test.go
index 6a320c74..86f6cfb5 100644
--- a/exp/services/ledgerexporter/internal/config_test.go
+++ b/exp/services/ledgerexporter/internal/config_test.go
@@ -1,20 +1,57 @@
 package ledgerexporter
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
+	"github.com/stellar/go/historyarchive"
 	"github.com/stretchr/testify/require"
 )
 
+func TestNewConfigResumeEnabled(t *testing.T) {
+	ctx := context.Background()
+
+	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.ValidateAndSetLedgerRange(ctx, mockArchive)
+	require.NoError(t, err)
+	require.Equal(t, config.DataStoreConfig.Type, "ABC")
+	require.Equal(t, config.LedgerBatchConfig.FilesPerPartition, uint32(1))
+	require.Equal(t, config.LedgerBatchConfig.LedgersPerFile, uint32(3))
+	require.True(t, config.Resume)
+	url, ok := config.DataStoreConfig.Params["destination_bucket_path"]
+	require.True(t, ok)
+	require.Equal(t, url, "your-bucket-name/subpath")
+}
+
+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"})
+	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})
+	require.ErrorContains(t, err, "Invalid TOML config")
+}
+
 func TestValidateStartAndEndLedger(t *testing.T) {
-	const latestNetworkLedger = 20000
+	const latestNetworkLedger = uint32(20000)
 
-	config := &Config{
-		ExporterConfig: ExporterConfig{
-			LedgersPerFile: 1,
-		},
-	}
 	tests := []struct {
 		name        string
 		startLedger uint32
@@ -67,104 +104,166 @@ func TestValidateStartAndEndLedger(t *testing.T) {
 		},
 	}
 
+	ctx := context.Background()
+	mockArchive := &historyarchive.MockArchive{}
+	mockArchive.On("GetRootHAS").Return(historyarchive.HistoryArchiveState{CurrentLedger: latestNetworkLedger}, nil).Times(len(tests))
+
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			config.StartLedger = tt.startLedger
-			config.EndLedger = tt.endLedger
+			config, err := NewConfig(ctx,
+				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 != "" {
-				require.Equal(t, tt.errMsg, config.validateAndSetLedgerRange(latestNetworkLedger).Error())
+				require.Equal(t, tt.errMsg, err.Error())
 			} else {
-				require.NoError(t, config.validateAndSetLedgerRange(latestNetworkLedger))
+				require.NoError(t, err)
 			}
 		})
 	}
 }
 
-func TestAdjustLedgerRangeBoundedMode(t *testing.T) {
+func TestAdjustedLedgerRangeBoundedMode(t *testing.T) {
 	tests := []struct {
-		name     string
-		config   *Config
-		expected *Config
+		name          string
+		configFile    string
+		start         uint32
+		end           uint32
+		expectedStart uint32
+		expectedEnd   uint32
 	}{
 		{
-			name:     "Min start ledger 2",
-			config:   &Config{StartLedger: 0, EndLedger: 10, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
-			expected: &Config{StartLedger: 2, EndLedger: 10, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
+			name:          "Min start ledger 2",
+			configFile:    "test/1perfile.toml",
+			start:         0,
+			end:           10,
+			expectedStart: 2,
+			expectedEnd:   10,
 		},
 		{
-			name:     "No change, 1 ledger per file",
-			config:   &Config{StartLedger: 2, EndLedger: 2, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
-			expected: &Config{StartLedger: 2, EndLedger: 2, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
+			name:          "No change, 1 ledger per file",
+			configFile:    "test/1perfile.toml",
+			start:         2,
+			end:           2,
+			expectedStart: 2,
+			expectedEnd:   2,
 		},
 		{
-			name:     "Min start ledger2, round up end ledger, 10 ledgers per file",
-			config:   &Config{StartLedger: 0, EndLedger: 1, ExporterConfig: ExporterConfig{LedgersPerFile: 10}},
-			expected: &Config{StartLedger: 2, EndLedger: 10, ExporterConfig: ExporterConfig{LedgersPerFile: 10}},
+			name:          "Min start ledger2, round up end ledger, 10 ledgers per file",
+			configFile:    "test/10perfile.toml",
+			start:         0,
+			end:           1,
+			expectedStart: 2,
+			expectedEnd:   10,
 		},
 		{
-			name:     "Round down start ledger and round up end ledger, 15 ledgers per file ",
-			config:   &Config{StartLedger: 4, EndLedger: 10, ExporterConfig: ExporterConfig{LedgersPerFile: 15}},
-			expected: &Config{StartLedger: 2, EndLedger: 15, ExporterConfig: ExporterConfig{LedgersPerFile: 15}},
+			name:          "Round down start ledger and round up end ledger, 15 ledgers per file ",
+			configFile:    "test/15perfile.toml",
+			start:         4,
+			end:           10,
+			expectedStart: 2,
+			expectedEnd:   15,
 		},
 		{
-			name:     "Round down start ledger and round up end ledger, 64 ledgers per file ",
-			config:   &Config{StartLedger: 400, EndLedger: 500, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
-			expected: &Config{StartLedger: 384, EndLedger: 512, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
+			name:          "Round down start ledger and round up end ledger, 64 ledgers per file ",
+			configFile:    "test/64perfile.toml",
+			start:         400,
+			end:           500,
+			expectedStart: 384,
+			expectedEnd:   512,
 		},
 		{
-			name:     "No change, 64 ledger per file",
-			config:   &Config{StartLedger: 64, EndLedger: 128, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
-			expected: &Config{StartLedger: 64, EndLedger: 128, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
+			name:          "No change, 64 ledger per file",
+			configFile:    "test/64perfile.toml",
+			start:         64,
+			end:           128,
+			expectedStart: 64,
+			expectedEnd:   128,
 		},
 	}
 
+	ctx := context.Background()
+	mockArchive := &historyarchive.MockArchive{}
+	mockArchive.On("GetRootHAS").Return(historyarchive.HistoryArchiveState{CurrentLedger: 500}, nil).Times(len(tests))
+
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			tt.config.adjustLedgerRange()
-			require.EqualValues(t, tt.expected.StartLedger, tt.config.StartLedger)
-			require.EqualValues(t, tt.expected.EndLedger, tt.config.EndLedger)
+			config, err := NewConfig(ctx,
+				Flags{StartLedger: tt.start, EndLedger: tt.end, ConfigFilePath: tt.configFile})
+			require.NoError(t, err)
+			err = config.ValidateAndSetLedgerRange(ctx, mockArchive)
+			require.NoError(t, err)
+			require.EqualValues(t, tt.expectedStart, config.StartLedger)
+			require.EqualValues(t, tt.expectedEnd, config.EndLedger)
 		})
 	}
 }
 
-func TestAdjustLedgerRangeUnBoundedMode(t *testing.T) {
+func TestAdjustedLedgerRangeUnBoundedMode(t *testing.T) {
 	tests := []struct {
-		name     string
-		config   *Config
-		expected *Config
+		name          string
+		configFile    string
+		start         uint32
+		end           uint32
+		expectedStart uint32
+		expectedEnd   uint32
 	}{
 		{
-			name:     "Min start ledger 2",
-			config:   &Config{StartLedger: 0, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
-			expected: &Config{StartLedger: 2, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
+			name:          "Min start ledger 2",
+			configFile:    "test/1perfile.toml",
+			start:         0,
+			end:           0,
+			expectedStart: 2,
+			expectedEnd:   0,
 		},
 		{
-			name:     "No change, 1 ledger per file",
-			config:   &Config{StartLedger: 2, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
-			expected: &Config{StartLedger: 2, ExporterConfig: ExporterConfig{LedgersPerFile: 1}},
+			name:          "No change, 1 ledger per file",
+			configFile:    "test/1perfile.toml",
+			start:         2,
+			end:           0,
+			expectedStart: 2,
+			expectedEnd:   0,
 		},
 		{
-			name:     "Round down start ledger, 15 ledgers per file ",
-			config:   &Config{StartLedger: 4, ExporterConfig: ExporterConfig{LedgersPerFile: 15}},
-			expected: &Config{StartLedger: 2, ExporterConfig: ExporterConfig{LedgersPerFile: 15}},
+			name:          "Round down start ledger, 15 ledgers per file ",
+			configFile:    "test/15perfile.toml",
+			start:         4,
+			end:           0,
+			expectedStart: 2,
+			expectedEnd:   0,
 		},
 		{
-			name:     "Round down start ledger, 64 ledgers per file ",
-			config:   &Config{StartLedger: 400, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
-			expected: &Config{StartLedger: 384, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
+			name:          "Round down start ledger, 64 ledgers per file ",
+			configFile:    "test/64perfile.toml",
+			start:         400,
+			end:           0,
+			expectedStart: 384,
+			expectedEnd:   0,
 		},
 		{
-			name:     "No change, 64 ledger per file",
-			config:   &Config{StartLedger: 64, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
-			expected: &Config{StartLedger: 64, ExporterConfig: ExporterConfig{LedgersPerFile: 64}},
+			name:          "No change, 64 ledger per file",
+			configFile:    "test/64perfile.toml",
+			start:         64,
+			end:           0,
+			expectedStart: 64,
+			expectedEnd:   0,
 		},
 	}
 
+	ctx := context.Background()
+
+	mockArchive := &historyarchive.MockArchive{}
+	mockArchive.On("GetRootHAS").Return(historyarchive.HistoryArchiveState{CurrentLedger: 500}, nil).Times(len(tests))
+
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			tt.config.adjustLedgerRange()
-			require.EqualValues(t, int(tt.expected.StartLedger), int(tt.config.StartLedger))
-			require.EqualValues(t, int(tt.expected.EndLedger), int(tt.config.EndLedger))
+			config, err := NewConfig(ctx,
+				Flags{StartLedger: tt.start, EndLedger: tt.end, ConfigFilePath: tt.configFile})
+			require.NoError(t, err)
+			err = config.ValidateAndSetLedgerRange(ctx, mockArchive)
+			require.NoError(t, err)
+			require.EqualValues(t, tt.expectedStart, config.StartLedger)
+			require.EqualValues(t, tt.expectedEnd, config.EndLedger)
 		})
 	}
 }
diff --git a/exp/services/ledgerexporter/internal/datastore_test.go b/exp/services/ledgerexporter/internal/datastore_test.go
new file mode 100644
index 00000000..d3128a4c
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/datastore_test.go
@@ -0,0 +1,13 @@
+package ledgerexporter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestInvalidStore(t *testing.T) {
+	_, err := NewDataStore(context.Background(), DataStoreConfig{Type: "unknown"}, "test")
+	require.Error(t, err)
+}
diff --git a/exp/services/ledgerexporter/internal/utils_test.go b/exp/services/ledgerexporter/internal/encoder_test.go
similarity index 56%
rename from exp/services/ledgerexporter/internal/utils_test.go
rename to exp/services/ledgerexporter/internal/encoder_test.go
index c11b500c..2bf61bce 100644
--- a/exp/services/ledgerexporter/internal/utils_test.go
+++ b/exp/services/ledgerexporter/internal/encoder_test.go
@@ -2,50 +2,12 @@ package ledgerexporter
 
 import (
 	"bytes"
-	"fmt"
 	"testing"
 
 	"github.com/stellar/go/xdr"
 	"github.com/stretchr/testify/require"
 )
 
-func TestGetObjectKeyFromSequenceNumber(t *testing.T) {
-	testCases := []struct {
-		filesPerPartition uint32
-		ledgerSeq         uint32
-		ledgersPerFile    uint32
-		expectedKey       string
-		expectedError     bool
-	}{
-		{0, 5, 1, "5.xdr.gz", false},
-		{0, 5, 10, "0-9.xdr.gz", false},
-		{2, 5, 0, "", true},
-		{2, 10, 100, "0-199/0-99.xdr.gz", false},
-		{2, 150, 50, "100-199/150-199.xdr.gz", false},
-		{2, 300, 200, "0-399/200-399.xdr.gz", false},
-		{2, 1, 1, "0-1/1.xdr.gz", false},
-		{4, 10, 100, "0-399/0-99.xdr.gz", false},
-		{4, 250, 50, "200-399/250-299.xdr.gz", false},
-		{1, 300, 200, "200-399.xdr.gz", false},
-		{1, 1, 1, "1.xdr.gz", false},
-	}
-
-	for _, tc := range testCases {
-		t.Run(fmt.Sprintf("LedgerSeq-%d-LedgersPerFile-%d", tc.ledgerSeq, tc.ledgersPerFile), func(t *testing.T) {
-			config := ExporterConfig{FilesPerPartition: tc.filesPerPartition, LedgersPerFile: tc.ledgersPerFile}
-			key, err := GetObjectKeyFromSequenceNumber(config, tc.ledgerSeq)
-
-			if tc.expectedError {
-				require.Error(t, err)
-				require.Empty(t, key)
-			} else {
-				require.NoError(t, err)
-				require.Equal(t, tc.expectedKey, key)
-			}
-		})
-	}
-}
-
 func createTestLedgerCloseMetaBatch(startSeq, endSeq uint32, count int) xdr.LedgerCloseMetaBatch {
 	var ledgerCloseMetas []xdr.LedgerCloseMeta
 	for i := 0; i < count; i++ {
diff --git a/exp/services/ledgerexporter/internal/exportmanager_test.go b/exp/services/ledgerexporter/internal/exportmanager_test.go
index b082a851..5eaa38bd 100644
--- a/exp/services/ledgerexporter/internal/exportmanager_test.go
+++ b/exp/services/ledgerexporter/internal/exportmanager_test.go
@@ -2,6 +2,7 @@ package ledgerexporter
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -36,11 +37,20 @@ func (s *ExportManagerSuite) TearDownTest() {
 	s.mockBackend.AssertExpectations(s.T())
 }
 
+func (s *ExportManagerSuite) TestInvalidExportConfig() {
+	config := LedgerBatchConfig{LedgersPerFile: 0, FilesPerPartition: 10}
+	registry := prometheus.NewRegistry()
+	queue := NewUploadQueue(1, registry)
+	_, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.Error(s.T(), err)
+}
+
 func (s *ExportManagerSuite) TestRun() {
-	config := ExporterConfig{LedgersPerFile: 64, FilesPerPartition: 10}
+	config := LedgerBatchConfig{LedgersPerFile: 64, FilesPerPartition: 10}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
-	exporter := NewExportManager(config, &s.mockBackend, queue, registry)
+	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.NoError(s.T(), err)
 
 	start := uint32(0)
 	end := uint32(255)
@@ -48,7 +58,7 @@ func (s *ExportManagerSuite) TestRun() {
 	for i := start; i <= end; i++ {
 		s.mockBackend.On("GetLedger", s.ctx, i).
 			Return(createLedgerCloseMeta(i), nil)
-		key, _ := GetObjectKeyFromSequenceNumber(config, i)
+		key := config.GetObjectKeyFromSequenceNumber(i)
 		expectedKeys.Add(key)
 	}
 
@@ -58,8 +68,8 @@ func (s *ExportManagerSuite) TestRun() {
 	go func() {
 		defer wg.Done()
 		for {
-			v, ok, err := queue.Dequeue(s.ctx)
-			s.Assert().NoError(err)
+			v, ok, dqErr := queue.Dequeue(s.ctx)
+			s.Assert().NoError(dqErr)
 			if !ok {
 				break
 			}
@@ -67,7 +77,7 @@ func (s *ExportManagerSuite) TestRun() {
 		}
 	}()
 
-	err := exporter.Run(s.ctx, start, end)
+	err = exporter.Run(s.ctx, start, end)
 	require.NoError(s.T(), err)
 
 	wg.Wait()
@@ -86,10 +96,11 @@ func (s *ExportManagerSuite) TestRun() {
 }
 
 func (s *ExportManagerSuite) TestRunContextCancel() {
-	config := ExporterConfig{LedgersPerFile: 1, FilesPerPartition: 1}
+	config := LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 1}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
-	exporter := NewExportManager(config, &s.mockBackend, queue, registry)
+	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.NoError(s.T(), err)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	s.mockBackend.On("GetLedger", mock.Anything, mock.Anything).
@@ -102,33 +113,65 @@ func (s *ExportManagerSuite) TestRunContextCancel() {
 
 	go func() {
 		for i := 0; i < 127; i++ {
-			_, ok, err := queue.Dequeue(s.ctx)
-			s.Assert().NoError(err)
+			_, ok, dqErr := queue.Dequeue(s.ctx)
+			s.Assert().NoError(dqErr)
 			s.Assert().True(ok)
 		}
 	}()
 
-	err := exporter.Run(ctx, 0, 255)
+	err = exporter.Run(ctx, 0, 255)
 	require.EqualError(s.T(), err, "failed to add ledgerCloseMeta for ledger 128: context canceled")
 
 }
 
 func (s *ExportManagerSuite) TestRunWithCanceledContext() {
-	config := ExporterConfig{LedgersPerFile: 1, FilesPerPartition: 10}
+	config := LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
-	exporter := NewExportManager(config, &s.mockBackend, queue, registry)
+	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.NoError(s.T(), err)
+
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	err := exporter.Run(ctx, 1, 10)
+	err = exporter.Run(ctx, 1, 10)
 	require.EqualError(s.T(), err, "context canceled")
 }
 
+func (s *ExportManagerSuite) TestGetObjectKeyFromSequenceNumber() {
+	testCases := []struct {
+		filesPerPartition uint32
+		ledgerSeq         uint32
+		ledgersPerFile    uint32
+		expectedKey       string
+	}{
+		{0, 5, 1, "5.xdr.gz"},
+		{0, 5, 10, "0-9.xdr.gz"},
+		{2, 10, 100, "0-199/0-99.xdr.gz"},
+		{2, 150, 50, "100-199/150-199.xdr.gz"},
+		{2, 300, 200, "0-399/200-399.xdr.gz"},
+		{2, 1, 1, "0-1/1.xdr.gz"},
+		{4, 10, 100, "0-399/0-99.xdr.gz"},
+		{4, 250, 50, "200-399/250-299.xdr.gz"},
+		{1, 300, 200, "200-399.xdr.gz"},
+		{1, 1, 1, "1.xdr.gz"},
+	}
+
+	for _, tc := range testCases {
+		s.T().Run(fmt.Sprintf("LedgerSeq-%d-LedgersPerFile-%d", tc.ledgerSeq, tc.ledgersPerFile), func(t *testing.T) {
+			config := LedgerBatchConfig{FilesPerPartition: tc.filesPerPartition, LedgersPerFile: tc.ledgersPerFile}
+			key := config.GetObjectKeyFromSequenceNumber(tc.ledgerSeq)
+			require.Equal(t, tc.expectedKey, key)
+		})
+	}
+}
+
 func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
-	config := ExporterConfig{LedgersPerFile: 1, FilesPerPartition: 10}
+	config := LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
-	exporter := NewExportManager(config, &s.mockBackend, queue, registry)
+	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.NoError(s.T(), err)
+
 	expectedkeys := set.NewSet[string](10)
 	actualKeys := set.NewSet[string](10)
 
@@ -150,9 +193,7 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 	end := uint32(255)
 	for i := start; i <= end; i++ {
 		require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(i)))
-
-		key, err := GetObjectKeyFromSequenceNumber(config, i)
-		require.NoError(s.T(), err)
+		key := config.GetObjectKeyFromSequenceNumber(i)
 		expectedkeys.Add(key)
 	}
 
@@ -162,10 +203,11 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMeta() {
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
-	config := ExporterConfig{LedgersPerFile: 1, FilesPerPartition: 10}
+	config := LedgerBatchConfig{LedgersPerFile: 1, FilesPerPartition: 10}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
-	exporter := NewExportManager(config, &s.mockBackend, queue, registry)
+	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.NoError(s.T(), err)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
@@ -174,15 +216,16 @@ func (s *ExportManagerSuite) TestAddLedgerCloseMetaContextCancel() {
 	}()
 
 	require.NoError(s.T(), exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(1)))
-	err := exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(2))
+	err = exporter.AddLedgerCloseMeta(ctx, createLedgerCloseMeta(2))
 	require.EqualError(s.T(), err, "context canceled")
 }
 
 func (s *ExportManagerSuite) TestAddLedgerCloseMetaKeyMismatch() {
-	config := ExporterConfig{LedgersPerFile: 10, FilesPerPartition: 1}
+	config := LedgerBatchConfig{LedgersPerFile: 10, FilesPerPartition: 1}
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
-	exporter := NewExportManager(config, &s.mockBackend, queue, registry)
+	exporter, err := NewExportManager(config, &s.mockBackend, queue, registry)
+	require.NoError(s.T(), err)
 
 	require.NoError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(16)))
 	require.EqualError(s.T(), exporter.AddLedgerCloseMeta(context.Background(), createLedgerCloseMeta(21)),
diff --git a/exp/services/ledgerexporter/internal/ledgerbatch_config_test.go b/exp/services/ledgerexporter/internal/ledgerbatch_config_test.go
new file mode 100644
index 00000000..cad9249d
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/ledgerbatch_config_test.go
@@ -0,0 +1,36 @@
+package ledgerexporter
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetObjectKeyFromSequenceNumber(t *testing.T) {
+	testCases := []struct {
+		filesPerPartition uint32
+		ledgerSeq         uint32
+		ledgersPerFile    uint32
+		expectedKey       string
+	}{
+		{0, 5, 1, "5.xdr.gz"},
+		{0, 5, 10, "0-9.xdr.gz"},
+		{2, 10, 100, "0-199/0-99.xdr.gz"},
+		{2, 150, 50, "100-199/150-199.xdr.gz"},
+		{2, 300, 200, "0-399/200-399.xdr.gz"},
+		{2, 1, 1, "0-1/1.xdr.gz"},
+		{4, 10, 100, "0-399/0-99.xdr.gz"},
+		{4, 250, 50, "200-399/250-299.xdr.gz"},
+		{1, 300, 200, "200-399.xdr.gz"},
+		{1, 1, 1, "1.xdr.gz"},
+	}
+
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("LedgerSeq-%d-LedgersPerFile-%d", tc.ledgerSeq, tc.ledgersPerFile), func(t *testing.T) {
+			config := LedgerBatchConfig{FilesPerPartition: tc.filesPerPartition, LedgersPerFile: tc.ledgersPerFile}
+			key := config.GetObjectKeyFromSequenceNumber(tc.ledgerSeq)
+			require.Equal(t, tc.expectedKey, key)
+		})
+	}
+}
diff --git a/exp/services/ledgerexporter/internal/queue_test.go b/exp/services/ledgerexporter/internal/queue_test.go
index 1d001765..a791d29e 100644
--- a/exp/services/ledgerexporter/internal/queue_test.go
+++ b/exp/services/ledgerexporter/internal/queue_test.go
@@ -58,6 +58,6 @@ func TestQueue(t *testing.T) {
 
 	l, ok, err = queue.Dequeue(context.Background())
 	require.NoError(t, err)
-	require.False(t, false)
+	require.False(t, ok)
 	require.Nil(t, l)
 }
diff --git a/exp/services/ledgerexporter/internal/resumablemanager_test.go b/exp/services/ledgerexporter/internal/resumablemanager_test.go
new file mode 100644
index 00000000..18913693
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/resumablemanager_test.go
@@ -0,0 +1,273 @@
+package ledgerexporter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stellar/go/historyarchive"
+	"github.com/stretchr/testify/require"
+)
+
+func TestResumability(t *testing.T) {
+
+	tests := []struct {
+		name              string
+		startLedger       uint32
+		endLedger         uint32
+		ledgerBatchConfig LedgerBatchConfig
+		absentLedger      uint32
+		findStartOk       bool
+		networkName       string
+		latestLedger      uint32
+		errorSnippet      string
+		archiveError      error
+	}{
+		{
+			name:         "archive error when resolving network latest",
+			startLedger:  4,
+			endLedger:    0,
+			absentLedger: 0,
+			findStartOk:  false,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test",
+			errorSnippet: "archive error",
+			archiveError: errors.New("archive error"),
+		},
+		{
+			name:         "End ledger same as start, data store has it",
+			startLedger:  4,
+			endLedger:    4,
+			absentLedger: 0,
+			findStartOk:  false,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName: "test",
+		},
+		{
+			name:         "End ledger same as start, data store does not have it",
+			startLedger:  14,
+			endLedger:    14,
+			absentLedger: 14,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName: "test",
+		},
+		{
+			name:         "binary search encounters an error during datastore retrieval",
+			startLedger:  24,
+			endLedger:    24,
+			absentLedger: 0,
+			findStartOk:  false,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test",
+			errorSnippet: "datastore error happened",
+		},
+		{
+			name:         "Data store is beyond boundary aligned start ledger",
+			startLedger:  20,
+			endLedger:    50,
+			absentLedger: 40,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName: "test",
+		},
+		{
+			name:         "Data store is beyond non boundary aligned start ledger",
+			startLedger:  55,
+			endLedger:    85,
+			absentLedger: 80,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName: "test",
+		},
+		{
+			name:         "Data store is beyond start and end ledger",
+			startLedger:  255,
+			endLedger:    275,
+			absentLedger: 0,
+			findStartOk:  false,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName: "test",
+		},
+		{
+			name:         "Data store is not beyond start ledger",
+			startLedger:  95,
+			endLedger:    125,
+			absentLedger: 95,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName: "test",
+		},
+		{
+			name:         "No start ledger provided",
+			startLedger:  0,
+			endLedger:    10,
+			absentLedger: 0,
+			findStartOk:  false,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test",
+			errorSnippet: "Invalid start value",
+		},
+		{
+			name:         "No end ledger provided, data store not beyond start",
+			startLedger:  1145,
+			endLedger:    0,
+			absentLedger: 1145,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test2",
+			latestLedger: uint32(2000),
+		},
+		{
+			name:         "No end ledger provided, data store is beyond start",
+			startLedger:  2145,
+			endLedger:    0,
+			absentLedger: 2250,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test3",
+			latestLedger: uint32(3000),
+		},
+		{
+			name:         "No end ledger provided, data store is beyond start and archive network latest, and partially into checkpoint frequency padding",
+			startLedger:  3145,
+			endLedger:    0,
+			absentLedger: 4070,
+			findStartOk:  true,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test4",
+			latestLedger: uint32(4000),
+		},
+		{
+			name:         "No end ledger provided, start is beyond archive network latest and checkpoint frequency padding",
+			startLedger:  5129,
+			endLedger:    0,
+			absentLedger: 0,
+			findStartOk:  false,
+			ledgerBatchConfig: LedgerBatchConfig{
+				FilesPerPartition: uint32(1),
+				LedgersPerFile:    uint32(10),
+			},
+			networkName:  "test5",
+			latestLedger: uint32(5000),
+			errorSnippet: "Invalid start value of 5129, it is greater than network's latest ledger of 5128",
+		},
+	}
+
+	ctx := context.Background()
+
+	mockDataStore := &MockDataStore{}
+
+	//"End ledger same as start, data store has it"
+	mockDataStore.On("Exists", ctx, "0-9.xdr.gz").Return(true, nil).Once()
+
+	//"End ledger same as start, data store does not have it"
+	mockDataStore.On("Exists", ctx, "10-19.xdr.gz").Return(false, nil).Once()
+
+	//"binary search encounters an error during datastore retrieval",
+	mockDataStore.On("Exists", ctx, "20-29.xdr.gz").Return(false, errors.New("datastore error happened")).Once()
+
+	//"Data store is beyond boundary aligned start ledger"
+	mockDataStore.On("Exists", ctx, "30-39.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "40-49.xdr.gz").Return(false, nil).Once()
+
+	//"Data store is beyond non boundary aligned start ledger"
+	mockDataStore.On("Exists", ctx, "70-79.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "80-89.xdr.gz").Return(false, nil).Once()
+
+	//"Data store is beyond start and end ledger"
+	mockDataStore.On("Exists", ctx, "260-269.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "270-279.xdr.gz").Return(true, nil).Once()
+
+	//"Data store is not beyond start ledger"
+	mockDataStore.On("Exists", ctx, "110-119.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "100-109.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "90-99.xdr.gz").Return(false, nil).Once()
+
+	//"No end ledger provided, data store not beyond start" uses latest from network="test2"
+	mockDataStore.On("Exists", ctx, "1630-1639.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1390-1399.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1260-1269.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1200-1209.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1160-1169.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1170-1179.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1150-1159.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "1140-1149.xdr.gz").Return(false, nil).Once()
+
+	//"No end ledger provided, data store is beyond start" uses latest from network="test3"
+	mockDataStore.On("Exists", ctx, "2630-2639.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "2390-2399.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "2260-2269.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "2250-2259.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "2240-2249.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "2230-2239.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "2200-2209.xdr.gz").Return(true, nil).Once()
+
+	//"No end ledger provided, data store is beyond start and archive network latest, and partially into checkpoint frequency padding" uses latest from network="test4"
+	mockDataStore.On("Exists", ctx, "3630-3639.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "3880-3889.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "4000-4009.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "4060-4069.xdr.gz").Return(true, nil).Once()
+	mockDataStore.On("Exists", ctx, "4090-4099.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "4080-4089.xdr.gz").Return(false, nil).Once()
+	mockDataStore.On("Exists", ctx, "4070-4079.xdr.gz").Return(false, nil).Once()
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockArchive := &historyarchive.MockArchive{}
+			mockArchive.On("GetRootHAS").Return(historyarchive.HistoryArchiveState{CurrentLedger: tt.latestLedger}, tt.archiveError).Once()
+
+			resumableManager := NewResumableManager(mockDataStore, tt.networkName, tt.ledgerBatchConfig, mockArchive)
+			absentLedger, ok, err := resumableManager.FindStart(ctx, tt.startLedger, tt.endLedger)
+			if tt.errorSnippet != "" {
+				require.ErrorContains(t, err, tt.errorSnippet)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tt.absentLedger, absentLedger)
+			require.Equal(t, tt.findStartOk, ok)
+			if tt.endLedger == 0 {
+				// archives are only expected to be called when end = 0
+				mockArchive.AssertExpectations(t)
+			}
+		})
+	}
+
+	mockDataStore.AssertExpectations(t)
+}
diff --git a/exp/services/ledgerexporter/internal/test/10perfile.toml b/exp/services/ledgerexporter/internal/test/10perfile.toml
new file mode 100644
index 00000000..8629bf4b
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/10perfile.toml
@@ -0,0 +1,4 @@
+network = "test"  
+
+[exporter_config]
+ledgers_per_file = 10
\ No newline at end of file
diff --git a/exp/services/ledgerexporter/internal/test/15perfile.toml b/exp/services/ledgerexporter/internal/test/15perfile.toml
new file mode 100644
index 00000000..eb76bac4
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/15perfile.toml
@@ -0,0 +1,4 @@
+network = "test"  
+
+[exporter_config]
+ledgers_per_file = 15
\ No newline at end of file
diff --git a/exp/services/ledgerexporter/internal/test/1perfile.toml b/exp/services/ledgerexporter/internal/test/1perfile.toml
new file mode 100644
index 00000000..cadef50d
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/1perfile.toml
@@ -0,0 +1,4 @@
+network = "test"  
+
+[exporter_config]
+ledgers_per_file = 1
\ No newline at end of file
diff --git a/exp/services/ledgerexporter/internal/test/64perfile.toml b/exp/services/ledgerexporter/internal/test/64perfile.toml
new file mode 100644
index 00000000..f4e30a71
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/64perfile.toml
@@ -0,0 +1,4 @@
+network = "test"  
+
+[exporter_config]
+ledgers_per_file = 64
\ No newline at end of file
diff --git a/exp/services/ledgerexporter/internal/test/no_network.toml b/exp/services/ledgerexporter/internal/test/no_network.toml
new file mode 100644
index 00000000..f5815b9b
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/no_network.toml
@@ -0,0 +1,10 @@
+
+[datastore_config]
+type = "ABC"
+
+[datastore_config.params]
+destination_bucket_path = "your-bucket-name/subpath"
+
+[exporter_config]
+ledgers_per_file = 3
+files_per_partition = 1
\ No newline at end of file
diff --git a/exp/services/ledgerexporter/internal/test/test.toml b/exp/services/ledgerexporter/internal/test/test.toml
new file mode 100644
index 00000000..49d62384
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/test.toml
@@ -0,0 +1,11 @@
+network = "test"  
+
+[datastore_config]
+type = "ABC"
+
+[datastore_config.params]
+destination_bucket_path = "your-bucket-name/subpath"
+
+[exporter_config]
+ledgers_per_file = 3
+files_per_partition = 1
\ No newline at end of file
diff --git a/exp/services/ledgerexporter/internal/test/validate_start_end.toml b/exp/services/ledgerexporter/internal/test/validate_start_end.toml
new file mode 100644
index 00000000..cadef50d
--- /dev/null
+++ b/exp/services/ledgerexporter/internal/test/validate_start_end.toml
@@ -0,0 +1,4 @@
+network = "test"  
+
+[exporter_config]
+ledgers_per_file = 1
\ No newline at end of file