datastore

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 5, 2026 License: Apache-2.0, Apache-2.0 Imports: 33 Imported by: 3

Documentation

Index

Constants

View Source 
const (
	Version = "1.0"
)

Variables

View Source 
var ErrNoLedgerFiles = errors.New("no ledger files found")
View Source 
var ErrNoValidLedgerFiles = errors.New("no valid ledger files found on the data store")

Functions

func FindLatestLedgerSequence 

func FindLatestLedgerSequence(ctx context.Context, datastore DataStore) (uint32, error)

FindLatestLedgerSequence returns the absolute latest ledger sequence number stored in the datastore.

func FindLatestLedgerUpToSequence 

func FindLatestLedgerUpToSequence(ctx context.Context, datastore DataStore,
	end uint32, schema DataStoreSchema) (uint32, error)

FindLatestLedgerUpToSequence finds the latest ledger sequence number that is less than or equal to a given 'end' sequence.

func FindOldestLedgerSequence 

func FindOldestLedgerSequence(ctx context.Context, datastore DataStore, schema DataStoreSchema) (uint32, error)

FindOldestLedgerSequence finds the oldest existing ledger in the datastore. It uses a binary search on the range of all known ledgers (from sequence 2 to the latest) to efficiently locate the first existing ledger file.

func GetLedgerFileExtension 

func GetLedgerFileExtension(ctx context.Context, dataStore DataStore) (string, error)

func LedgerFileIter 

func LedgerFileIter(ctx context.Context, ds DataStore, startAfter,
	stopAfter string) iter.Seq2[LedgerFile, error]

LedgerFileIter returns an iterator (iter.Seq2) over ledger files whose file paths fall in the lexicographic range (startAfter, stopAfter]. Paths are listed in ascending lexicographic order.

Both bounds are optional:

  • If startAfter == "", iteration begins at the first file.
  • If stopAfter == "", the upper bound is unbounded.

If both bounds are provided and stopAfter <= startAfter, the range is empty and the iterator yields a single error.

The iterator stops when:

  • all matching file paths have been consumed,
  • the context is canceled, or
  • an error occurs.

func ParseRangeFromObjectKey 

func ParseRangeFromObjectKey(base string) (uint32, uint32, error)

ParseRangeFromObjectKey extracts the [low, high] ledger sequence range from a datastore object key. The expected filename format is defined by SEP-54: https://github.com/stellar/stellar-protocol/blob/master/ecosystem/sep-0054.md#key-format

Types

type ConfigMismatchError 

type ConfigMismatchError struct {
	Diffs []string
}

func (*ConfigMismatchError) Error 

func (e *ConfigMismatchError) Error() string

type DataStore 

type DataStore interface {
	GetFileMetadata(ctx context.Context, path string) (map[string]string, error)
	GetFileLastModified(ctx context.Context, filePath string) (time.Time, error)
	GetFile(ctx context.Context, path string) (io.ReadCloser, error)
	PutFile(ctx context.Context, path string, in io.WriterTo, metaData map[string]string) error
	PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo, metaData map[string]string) (bool, error)
	Exists(ctx context.Context, path string) (bool, error)
	Size(ctx context.Context, path string) (int64, error)
	ListFilePaths(ctx context.Context, options ListFileOptions) ([]string, error)
	Close() error
}

DataStore defines an interface for interacting with data storage

func FromGCSClient 

func FromGCSClient(ctx context.Context, client *storage.Client, bucketPath string) (DataStore, error)

func FromS3Client 

func FromS3Client(ctx context.Context, client *s3.Client, bucketPath string) (DataStore, error)

func NewDataStore 

func NewDataStore(ctx context.Context, datastoreConfig DataStoreConfig) (DataStore, error)

NewDataStore factory, it creates a new DataStore based on the config type

func NewFilesystemDataStore added in v0.2.0 

func NewFilesystemDataStore(ctx context.Context, datastoreConfig DataStoreConfig) (DataStore, error)

NewFilesystemDataStore creates a new FilesystemDataStore from configuration.

func NewFilesystemDataStoreWithPath added in v0.2.0 

func NewFilesystemDataStoreWithPath(basePath string) (DataStore, error)

NewFilesystemDataStoreWithPath creates a FilesystemDataStore with the given base path.

func NewGCSDataStore 

func NewGCSDataStore(ctx context.Context, dataStoreConfig DataStoreConfig) (DataStore, error)

func NewS3DataStore 

func NewS3DataStore(ctx context.Context, datastoreConfig DataStoreConfig) (DataStore, error)

type DataStoreConfig 

type DataStoreConfig struct {
	Type              string            `toml:"type"`
	Params            map[string]string `toml:"params"`
	Schema            DataStoreSchema   `toml:"schema"`
	NetworkPassphrase string
	Compression       string
}

DataStoreConfig defines user-provided configuration used to initialize a DataStore.

type DataStoreSchema 

type DataStoreSchema struct {
	LedgersPerFile    uint32 `toml:"ledgers_per_file"`
	FilesPerPartition uint32 `toml:"files_per_partition"`
	FileExtension     string // Optional – for backward (zstd) compatibility only
}

func LoadSchema 

func LoadSchema(ctx context.Context, dataStore DataStore, cfg DataStoreConfig) (DataStoreSchema, error)

LoadSchema reads the datastore manifest from the given DataStore and returns its schema configuration.

func (DataStoreSchema) GetObjectKeyFromSequenceNumber 

func (ec DataStoreSchema) GetObjectKeyFromSequenceNumber(ledgerSeq uint32) string

GetObjectKeyFromSequenceNumber generates the object key name from the ledger sequence number based on configuration.

func (DataStoreSchema) GetSequenceNumberEndBoundary 

func (ec DataStoreSchema) GetSequenceNumberEndBoundary(ledgerSeq uint32) uint32

Returns the associated end ledger for a given ledgerSeq based on the datastore configuration for LedgersPerFile and a ceiling of MaxUint32.

func (DataStoreSchema) GetSequenceNumberStartBoundary 

func (ec DataStoreSchema) GetSequenceNumberStartBoundary(ledgerSeq uint32) uint32

type DatastoreManifest 

type DatastoreManifest struct {
	NetworkPassphrase string `json:"networkPassphrase"`
	Version           string `json:"version"`
	Compression       string `json:"compression"`
	LedgersPerFile    uint32 `json:"ledgersPerBatch"`
	FilesPerPartition uint32 `json:"batchesPerPartition"`
}

DatastoreManifest represents the persisted configuration stored in the object store.

func PublishConfig 

func PublishConfig(ctx context.Context, dataStore DataStore, cfg DataStoreConfig) (DatastoreManifest, bool, error)

PublishConfig ensures that a datastore manifest exists and matches the provided configuration. If the manifest is missing, it creates one. Returns the manifest, whether it was created, and any error encountered.

type FilesystemDataStore added in v0.2.0 

type FilesystemDataStore struct {
	// contains filtered or unexported fields
}

FilesystemDataStore implements DataStore for local filesystem storage.

Note: This implementation is not recommended for production use. It is intended for development and testing purposes only.

This implementation does not support storing metadata. The metaData parameter in PutFile and PutFileIfNotExists is ignored, and GetFileMetadata always returns an empty map.

Concurrent writes to the same file path are not safe and may result in data corruption. Callers must ensure proper synchronization when writing to the same path from multiple processes.

func (*FilesystemDataStore) Close added in v0.2.0 

func (f *FilesystemDataStore) Close() error

Close is a no-op for FilesystemDataStore as it doesn't maintain persistent connections.

func (*FilesystemDataStore) Exists added in v0.2.0 

func (f *FilesystemDataStore) Exists(ctx context.Context, path string) (bool, error)

Exists checks if a file exists in the filesystem.

func (*FilesystemDataStore) GetFile added in v0.2.0 

func (f *FilesystemDataStore) GetFile(ctx context.Context, path string) (io.ReadCloser, error)

GetFile returns a reader for the file at the given path.

func (*FilesystemDataStore) GetFileLastModified added in v0.2.0 

func (f *FilesystemDataStore) GetFileLastModified(ctx context.Context, path string) (time.Time, error)

GetFileLastModified returns the last modification time of a file.

func (*FilesystemDataStore) GetFileMetadata added in v0.2.0 

func (f *FilesystemDataStore) GetFileMetadata(ctx context.Context, path string) (map[string]string, error)

GetFileMetadata returns an empty map as filesystem storage does not support metadata.

func (*FilesystemDataStore) ListFilePaths added in v0.2.0 

func (f *FilesystemDataStore) ListFilePaths(ctx context.Context, options ListFileOptions) ([]string, error)

ListFilePaths lists file paths matching the given options. Results are returned in lexicographical order (matching GCS/S3 behavior).

func (*FilesystemDataStore) PutFile added in v0.2.0 

func (f *FilesystemDataStore) PutFile(ctx context.Context, path string, in io.WriterTo, metaData map[string]string) error

PutFile writes a file to the filesystem.

func (*FilesystemDataStore) PutFileIfNotExists added in v0.2.0 

func (f *FilesystemDataStore) PutFileIfNotExists(
	ctx context.Context, path string, in io.WriterTo, metaData map[string]string,
) (bool, error)

PutFileIfNotExists writes a file only if it doesn't already exist.

func (*FilesystemDataStore) Size added in v0.2.0 

func (f *FilesystemDataStore) Size(ctx context.Context, path string) (int64, error)

Size returns the size of a file in bytes.

type GCSDataStore 

type GCSDataStore struct {
	// contains filtered or unexported fields
}

GCSDataStore implements DataStore for GCS

func (GCSDataStore) Close 

func (b GCSDataStore) Close() error

Close closes the GCS client connection.

func (GCSDataStore) Exists 

func (b GCSDataStore) Exists(ctx context.Context, pth string) (bool, error)

Exists checks if a file exists in the GCS bucket.

func (GCSDataStore) GetFile 

func (b GCSDataStore) GetFile(ctx context.Context, filePath string) (io.ReadCloser, error)

GetFile retrieves a file from the GCS bucket.

func (GCSDataStore) GetFileAttrs 

func (b GCSDataStore) GetFileAttrs(ctx context.Context, filePath string) (*storage.ObjectAttrs, error)

func (GCSDataStore) GetFileLastModified 

func (b GCSDataStore) GetFileLastModified(ctx context.Context, filePath string) (time.Time, error)

GetFileLastModified retrieves the last modified time of a file in the GCS bucket.

func (GCSDataStore) GetFileMetadata 

func (b GCSDataStore) GetFileMetadata(ctx context.Context, filePath string) (map[string]string, error)

GetFileMetadata retrieves the metadata for the specified file in the GCS bucket.

func (GCSDataStore) ListFilePaths 

func (b GCSDataStore) ListFilePaths(ctx context.Context, options ListFileOptions) ([]string, error)

ListFilePaths lists up to 'limit' file paths under the provided prefix. Returned paths are relative to the bucket prefix. and ordered lexicographically ascending as provided by the backend. If limit <= 0, implementations default to a cap of 1,000; values > 1,000 are capped to 1,000.

func (GCSDataStore) PutFile 

func (b GCSDataStore) PutFile(ctx context.Context, filePath string, in io.WriterTo, metaData map[string]string) error

PutFile uploads a file to GCS

func (GCSDataStore) PutFileIfNotExists 

func (b GCSDataStore) PutFileIfNotExists(ctx context.Context, filePath string, in io.WriterTo, metaData map[string]string) (bool, error)

PutFileIfNotExists uploads a file to GCS only if it doesn't already exist.

func (GCSDataStore) Size 

func (b GCSDataStore) Size(ctx context.Context, pth string) (int64, error)

Size retrieves the size of a file in the GCS bucket.

type LedgerFile 

type LedgerFile struct {
	Key  string
	High uint32
	Low  uint32
}

type ListFileOptions 

type ListFileOptions struct {
	// Prefix filters the results to only include keys that start with this string.
	Prefix string

	// StartAfter specifies the key from which to begin listing. The returned keys will be
	// lexicographically greater than this value.
	StartAfter string

	// Limit restricts the number of keys returned. A value of 0 will use the default limit,
	// and any value above listFilePathsMaxLimit will be automatically capped.
	Limit uint32
}

ListFileOptions controls how ListFilePaths enumerates objects.

type MetaData 

type MetaData struct {
	StartLedger          uint32
	EndLedger            uint32
	StartLedgerCloseTime int64
	EndLedgerCloseTime   int64
	ProtocolVersion      uint32
	CoreVersion          string
	NetworkPassPhrase    string
	CompressionType      string
	Version              string
}

func NewMetaDataFromMap 

func NewMetaDataFromMap(data map[string]string) (MetaData, error)

func (MetaData) ToMap 

func (m MetaData) ToMap() map[string]string

type MockDataStore 

type MockDataStore struct {
	mock.Mock
}

MockDataStore is a mock implementation for the Storage interface.

func (*MockDataStore) Close 

func (m *MockDataStore) Close() error

func (*MockDataStore) Exists 

func (m *MockDataStore) Exists(ctx context.Context, path string) (bool, error)

func (*MockDataStore) GetFile 

func (m *MockDataStore) GetFile(ctx context.Context, path string) (io.ReadCloser, error)

func (*MockDataStore) GetFileLastModified 

func (m *MockDataStore) GetFileLastModified(ctx context.Context, filePath string) (time.Time, error)

func (*MockDataStore) GetFileMetadata 

func (m *MockDataStore) GetFileMetadata(ctx context.Context, path string) (map[string]string, error)

func (*MockDataStore) GetSchema 

func (m *MockDataStore) GetSchema() DataStoreSchema

func (*MockDataStore) ListFilePaths 

func (m *MockDataStore) ListFilePaths(ctx context.Context, options ListFileOptions) ([]string, error)

func (*MockDataStore) PutFile 

func (m *MockDataStore) PutFile(ctx context.Context, path string, in io.WriterTo, metadata map[string]string) error

func (*MockDataStore) PutFileIfNotExists 

func (m *MockDataStore) PutFileIfNotExists(ctx context.Context, path string, in io.WriterTo, metadata map[string]string) (bool, error)

func (*MockDataStore) Size 

func (m *MockDataStore) Size(ctx context.Context, path string) (int64, error)

type S3DataStore 

type S3DataStore struct {
	// contains filtered or unexported fields
}

S3DataStore implements DataStore for AWS S3 and S3-compatible services.

func (S3DataStore) Close 

func (b S3DataStore) Close() error

Close does nothing for S3ObjectStore as it does not maintain a persistent connection.

func (S3DataStore) Exists 

func (b S3DataStore) Exists(ctx context.Context, filePath string) (bool, error)

Exists checks if a file exists in the S3-compatible bucket.

func (S3DataStore) GetFile 

func (b S3DataStore) GetFile(ctx context.Context, filePath string) (io.ReadCloser, error)

GetFile retrieves a file from the S3-compatible bucket.

func (S3DataStore) GetFileLastModified 

func (b S3DataStore) GetFileLastModified(ctx context.Context, filePath string) (time.Time, error)

GetFileLastModified retrieves the last modified time of a file in the S3-compatible bucket.

func (S3DataStore) GetFileMetadata 

func (b S3DataStore) GetFileMetadata(ctx context.Context, filePath string) (map[string]string, error)

GetFileMetadata retrieves the metadata for the specified file in the S3-compatible bucket.

func (S3DataStore) HeadObject 

func (b S3DataStore) HeadObject(ctx context.Context, filePath string) (*s3.HeadObjectOutput, error)

func (S3DataStore) ListFilePaths 

func (b S3DataStore) ListFilePaths(ctx context.Context, options ListFileOptions) ([]string, error)

ListFilePaths lists up to 'limit' file paths under the provided prefix. Returned paths are relative to the bucket prefix. and ordered lexicographically ascending as provided by the backend. If limit <= 0, implementations default to a cap of 1,000; values > 1,000 are capped to 1,000.

func (S3DataStore) PutFile 

func (b S3DataStore) PutFile(ctx context.Context, filePath string, in io.WriterTo, metaData map[string]string) error

PutFile uploads a file to S3-compatible bucket

func (S3DataStore) PutFileIfNotExists 

func (b S3DataStore) PutFileIfNotExists(ctx context.Context, filePath string, in io.WriterTo, metaData map[string]string) (bool, error)

PutFileIfNotExists uploads a file to S3-compatible bucket only if it doesn't already exist.

func (S3DataStore) Size 

func (b S3DataStore) Size(ctx context.Context, filePath string) (int64, error)

Size retrieves the size of a file in the S3-compatible bucket.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.