services/horizon: Reingest from precomputed TxMeta

ingest/ledgerbackend/buffered_storage_backend.go

@@ -18,12 +18,10 @@ import (
 var _ LedgerBackend = (*BufferedStorageBackend)(nil)
 
 type BufferedStorageBackendConfig struct {
-	LedgerBatchConfig datastore.DataStoreSchema
-	DataStore         datastore.DataStore
-	BufferSize        uint32
-	NumWorkers        uint32
-	RetryLimit        uint32
-	RetryWait         time.Duration
+	BufferSize uint32        `toml:"buffer_size"`
+	NumWorkers uint32        `toml:"num_workers"`
+	RetryLimit uint32        `toml:"retry_limit"`
+	RetryWait  time.Duration `toml:"retry_wait"`
 }
 
 // BufferedStorageBackend is a ledger backend that reads from a storage service.
@@ -45,7 +43,7 @@ type BufferedStorageBackend struct {
 }
 
 // NewBufferedStorageBackend returns a new BufferedStorageBackend instance.
-func NewBufferedStorageBackend(ctx context.Context, config BufferedStorageBackendConfig) (*BufferedStorageBackend, error) {
+func NewBufferedStorageBackend(ctx context.Context, config BufferedStorageBackendConfig, dataStore datastore.DataStore) (*BufferedStorageBackend, error) {
 	if config.BufferSize == 0 {
 		return nil, errors.New("buffer size must be > 0")
 	}
@@ -54,17 +52,17 @@ func NewBufferedStorageBackend(ctx context.Context, config BufferedStorageBacken
 		return nil, errors.New("number of workers must be <= BufferSize")
 	}
 
-	if config.DataStore == nil {
+	if dataStore == nil {
 		return nil, errors.New("no DataStore provided")
 	}
 
-	if config.LedgerBatchConfig.LedgersPerFile <= 0 {
+	if dataStore.GetSchema(ctx).LedgersPerFile <= 0 {
 		return nil, errors.New("ledgersPerFile must be > 0")
 	}
 
 	bsBackend := &BufferedStorageBackend{
 		config:    config,
-		dataStore: config.DataStore,
+		dataStore: dataStore,
 	}
 
 	return bsBackend, nil

ingest/ledgerbackend/buffered_storage_backend_test.go

@@ -44,29 +44,21 @@ func createBufferedStorageBackendConfigForTesting() BufferedStorageBackendConfig
 	param := make(map[string]string)
 	param["destination_bucket_path"] = "testURL"
 
-	ledgerBatchConfig := datastore.DataStoreSchema{
-		LedgersPerFile:    1,
-		FilesPerPartition: 64000,
-	}
-
-	dataStore := new(datastore.MockDataStore)
-
 	return BufferedStorageBackendConfig{
-		LedgerBatchConfig: ledgerBatchConfig,
-		DataStore:         dataStore,
-		BufferSize:        100,
-		NumWorkers:        5,
-		RetryLimit:        3,
-		RetryWait:         time.Microsecond,
+		BufferSize: 100,
+		NumWorkers: 5,
+		RetryLimit: 3,
+		RetryWait:  time.Microsecond,
 	}
 }
 
 func createBufferedStorageBackendForTesting() BufferedStorageBackend {
 	config := createBufferedStorageBackendConfigForTesting()
 
+	dataStore := new(datastore.MockDataStore)
 	return BufferedStorageBackend{
 		config:    config,
-		dataStore: config.DataStore,
+		dataStore: dataStore,
 	}
 }
 
@@ -86,6 +78,10 @@ func createMockdataStore(t *testing.T, start, end, partitionSize, count uint32)
 		}
 		mockDataStore.On("GetFile", mock.Anything, objectName).Return(readCloser, nil)
 	}
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    count,
+		FilesPerPartition: partitionSize,
+	})
 
 	t.Cleanup(func() {
 		mockDataStore.AssertExpectations(t)
@@ -128,13 +124,17 @@ func createLCMBatchReader(start, end, count uint32) io.ReadCloser {
 func TestNewBufferedStorageBackend(t *testing.T) {
 	ctx := context.Background()
 	config := createBufferedStorageBackendConfigForTesting()
-
-	bsb, err := NewBufferedStorageBackend(ctx, config)
+	mockDataStore := new(datastore.MockDataStore)
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    uint32(1),
+		FilesPerPartition: partitionSize,
+	})
+	bsb, err := NewBufferedStorageBackend(ctx, config, mockDataStore)
 	assert.NoError(t, err)
 
-	assert.Equal(t, bsb.dataStore, config.DataStore)
-	assert.Equal(t, uint32(1), bsb.config.LedgerBatchConfig.LedgersPerFile)
-	assert.Equal(t, uint32(64000), bsb.config.LedgerBatchConfig.FilesPerPartition)
+	assert.Equal(t, bsb.dataStore, mockDataStore)
+	assert.Equal(t, uint32(1), bsb.dataStore.GetSchema(ctx).LedgersPerFile)
+	assert.Equal(t, uint32(64000), bsb.dataStore.GetSchema(ctx).FilesPerPartition)
 	assert.Equal(t, uint32(100), bsb.config.BufferSize)
 	assert.Equal(t, uint32(5), bsb.config.NumWorkers)
 	assert.Equal(t, uint32(3), bsb.config.RetryLimit)
@@ -210,12 +210,14 @@ func TestCloudStorageGetLedger_MultipleLedgerPerFile(t *testing.T) {
 	lcmArray := createLCMForTesting(startLedger, endLedger)
 	bsb := createBufferedStorageBackendForTesting()
 	ctx := context.Background()
-	bsb.config.LedgerBatchConfig.LedgersPerFile = uint32(2)
 	ledgerRange := BoundedRange(startLedger, endLedger)
 
 	mockDataStore := createMockdataStore(t, startLedger, endLedger, partitionSize, 2)
 	bsb.dataStore = mockDataStore
-
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    uint32(2),
+		FilesPerPartition: partitionSize,
+	})
 	assert.NoError(t, bsb.PrepareRange(ctx, ledgerRange))
 	assert.Eventually(t, func() bool { return len(bsb.ledgerBuffer.ledgerQueue) == 2 }, time.Second*5, time.Millisecond*50)
 
@@ -451,6 +453,10 @@ func TestLedgerBufferClose(t *testing.T) {
 
 	mockDataStore := new(datastore.MockDataStore)
 	partition := ledgerPerFileCount*partitionSize - 1
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    ledgerPerFileCount,
+		FilesPerPartition: partitionSize,
+	})
 
 	objectName := fmt.Sprintf("FFFFFFFF--0-%d/%08X--%d.xdr.zstd", partition, math.MaxUint32-3, 3)
 	afterPrepareRange := make(chan struct{})
@@ -483,7 +489,10 @@ func TestLedgerBufferBoundedObjectNotFound(t *testing.T) {
 
 	mockDataStore := new(datastore.MockDataStore)
 	partition := ledgerPerFileCount*partitionSize - 1
-
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    ledgerPerFileCount,
+		FilesPerPartition: partitionSize,
+	})
 	objectName := fmt.Sprintf("FFFFFFFF--0-%d/%08X--%d.xdr.zstd", partition, math.MaxUint32-3, 3)
 	mockDataStore.On("GetFile", mock.Anything, objectName).Return(io.NopCloser(&bytes.Buffer{}), os.ErrNotExist).Once()
 	t.Cleanup(func() {
@@ -509,7 +518,10 @@ func TestLedgerBufferUnboundedObjectNotFound(t *testing.T) {
 
 	mockDataStore := new(datastore.MockDataStore)
 	partition := ledgerPerFileCount*partitionSize - 1
-
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    ledgerPerFileCount,
+		FilesPerPartition: partitionSize,
+	})
 	objectName := fmt.Sprintf("FFFFFFFF--0-%d/%08X--%d.xdr.zstd", partition, math.MaxUint32-3, 3)
 	iteration := &atomic.Int32{}
 	cancelAfter := int32(bsb.config.RetryLimit) + 2
@@ -551,7 +563,10 @@ func TestLedgerBufferRetryLimit(t *testing.T) {
 	})
 
 	bsb.dataStore = mockDataStore
-
+	mockDataStore.On("GetSchema", mock.Anything).Return(datastore.DataStoreSchema{
+		LedgersPerFile:    ledgerPerFileCount,
+		FilesPerPartition: partitionSize,
+	})
 	assert.NoError(t, bsb.PrepareRange(context.Background(), ledgerRange))
 
 	bsb.ledgerBuffer.wg.Wait()

ingest/ledgerbackend/ledger_buffer.go

@@ -83,19 +83,19 @@ func (bsb *BufferedStorageBackend) newLedgerBuffer(ledgerRange Range) (*ledgerBu
 	// but for easier conceptualization, len(taskQueue) can be interpreted as both pending and in-flight tasks
 	// where we assume the workers are empty and not processing any tasks.
 	for i := 0; i <= int(bsb.config.BufferSize); i++ {
-		ledgerBuffer.pushTaskQueue()
+		ledgerBuffer.pushTaskQueue(ctx)
 	}
 
 	return ledgerBuffer, nil
 }
 
-func (lb *ledgerBuffer) pushTaskQueue() {
+func (lb *ledgerBuffer) pushTaskQueue(ctx context.Context) {
 	// In bounded mode, don't queue past the end ledger
 	if lb.nextTaskLedger > lb.ledgerRange.to && lb.ledgerRange.bounded {
 		return
 	}
 	lb.taskQueue <- lb.nextTaskLedger
-	lb.nextTaskLedger += lb.config.LedgerBatchConfig.LedgersPerFile
+	lb.nextTaskLedger += lb.dataStore.GetSchema(ctx).LedgersPerFile
 }
 
 // sleepWithContext returns true upon sleeping without interruption from the context
@@ -155,15 +155,15 @@ func (lb *ledgerBuffer) worker(ctx context.Context) {
 				// Thus, the number of tasks decreases by 1 and the priority queue length increases by 1.
 				// This keeps the overall total the same (<= BufferSize). As long as the the ledger buffer invariant
 				// was maintained in the previous state, it is still maintained during this state transition.
-				lb.storeObject(ledgerObject, sequence)
+				lb.storeObject(ctx, ledgerObject, sequence)
 				break
 			}
 		}
 	}
 }
 
 func (lb *ledgerBuffer) downloadLedgerObject(ctx context.Context, sequence uint32) ([]byte, error) {
-	objectKey := lb.config.LedgerBatchConfig.GetObjectKeyFromSequenceNumber(sequence)
+	objectKey := lb.dataStore.GetSchema(ctx).GetObjectKeyFromSequenceNumber(sequence)
 
 	reader, err := lb.dataStore.GetFile(ctx, objectKey)
 	if err != nil {
@@ -180,7 +180,7 @@ func (lb *ledgerBuffer) downloadLedgerObject(ctx context.Context, sequence uint3
 	return objectBytes, nil
 }
 
-func (lb *ledgerBuffer) storeObject(ledgerObject []byte, sequence uint32) {
+func (lb *ledgerBuffer) storeObject(ctx context.Context, ledgerObject []byte, sequence uint32) {
 	lb.priorityQueueLock.Lock()
 	defer lb.priorityQueueLock.Unlock()
 
@@ -198,7 +198,7 @@ func (lb *ledgerBuffer) storeObject(ledgerObject []byte, sequence uint32) {
 	for lb.ledgerPriorityQueue.Len() > 0 && lb.currentLedger == uint32(lb.ledgerPriorityQueue.Peek().startLedger) {
 		item := lb.ledgerPriorityQueue.Pop()
 		lb.ledgerQueue <- item.payload
-		lb.currentLedger += lb.config.LedgerBatchConfig.LedgersPerFile
+		lb.currentLedger += lb.dataStore.GetSchema(ctx).LedgersPerFile
 	}
 }
 
@@ -215,7 +215,7 @@ func (lb *ledgerBuffer) getFromLedgerQueue(ctx context.Context) (xdr.LedgerClose
 			// Thus len(ledgerQueue) decreases by 1 and the number of tasks increases by 1.
 			// The overall sum below remains the same:
 			// len(taskQueue) + len(ledgerQueue) + ledgerPriorityQueue.Len() <= bsb.config.BufferSize
-			lb.pushTaskQueue()
+			lb.pushTaskQueue(ctx)
 
 			lcmBatch := xdr.LedgerCloseMetaBatch{}
 			decoder := compressxdr.NewXDRDecoder(compressxdr.DefaultCompressor, &lcmBatch)