services: Implement call to `getLedgers` and start parsing the `xdr.LedgerCloseMeta` #197

marcelosalloum · 2025-06-18T23:40:07Z

What

Update the indexer code to stream from {RPC}.getLedgers and break down the results from ledgerRange -> ledgers -> transactions.

Also, (pseudo) process the ledgers in parallel.

Why

As part of the new indexer implementation.

Future Work

evaluate transactions, filter by participant account & store in the DB
evaluate operations, filter by participant account & store in the DB
generate state changes
unlock channel accounts

I haven't added many tests to the ingest service yet, since this code will still evolve quite a bit.

Issue that this PR addresses

Partially address #128

Checklist

PR Structure

It is not possible to break this PR down into smaller PRs.
This PR does not mix refactoring changes with feature changes.
This PR's title starts with name of package that is most changed in the PR, or all if the changes are broad or impact many packages.

Thoroughness

This PR adds tests for the new functionality or fixes.
All updated queries have been tested (refer to this check if the data set returned by the updated query is expected to be same as the original one).

Release

This is not a breaking change.
This is ready to be tested in development.
The new functionality is gated with a feature flag if this is not ready for production.

marcelosalloum · 2025-06-19T00:17:48Z

internal/services/ingest.go

+	mu := sync.Mutex{}
+	var allTxHashes []string
+	var errs []error
+
+	// Submit tasks to the pool
+	for _, ledger := range getLedgersResponse.Ledgers {
+		ledger := ledger // Create a new variable to avoid closure issues
+		pool.Submit(func() {
+			txHashes, err := m.processLedger(ctx, ledger)
+			mu.Lock()
+			if err != nil {


I plan to use toid to order the transactions in the database. It's calculated like this.

internal/ingest/ingest.go

internal/services/ingest.go

aditya1702 · 2025-06-24T18:59:25Z

internal/services/ingest.go

+func (m *ingestService) fetchNextLedgersBatch(ctx context.Context, startLedger uint32) (GetLedgersResponse, error) {
+	rpcHealth, err := m.rpcService.GetHealth()
+	if err != nil {
+		return GetLedgersResponse{}, fmt.Errorf("getting rpc health: %w", err)


In cases where RPC becomes unhealthy for a few mins and then healthy again, this would stop the ingestion and terminate right? How about we terminate ingestion process only if RPC stays unhealthy for a certain threshold. Otherwise we keep letting it try and fetch the ledgers batch until then.

Hm, good point! I'll work on that 👍

Done in dc0f8a5

aditya1702 · 2025-06-24T19:09:42Z

internal/services/ingest.go

+	if latestLedgerSynced >= rpcNewestLedger {
+		return LedgerSeqRange{}, true
+	}
+	ledgerRange.StartLedger = max(latestLedgerSynced+1, rpcOldestLedger)


If we do a max() here would that mean if wallet-backend is at 5 and rpc's oldest ledger is at 8, it would skip ingesting 6, 7? If we are ingesting using GetLedgers by getting ledgers in bulk, do we want to allow WB catch up to RPC instead of skipping ledgers?

If we do a max() here would that mean if wallet-backend is at 5 and rpc's oldest ledger is at 8, it would skip ingesting 6, 7?

I think we don't have another option here, right? If we have [oldestLedger: 8, newestLedger: 20], we can only ingest within that window, as the RPC doesn't have anything older than ledger 8 available for ingestion.

internal/services/ingest.go

aditya1702 · 2025-06-24T19:18:03Z

internal/services/ingest.go

+	log.Ctx(ctx).Debugf("🚧 Got %d transactions for ledger %d", len(transactions), xdrLedgerCloseMeta.LedgerSequence())
+	txHashes := make([]string, 0, len(transactions))
+	for _, tx := range transactions {
+		txHashes = append(txHashes, tx.Hash.HexString())


Since this is still in TODO, I am assuming this will return a list of types.Transaction structs instead of tx hash strings? Because what would be the use of the tx hashes?

Yes, you're right! I'm returning txHashes because I'd like to log something short. But the output will definitely be actual objects!

… a new IngestStoreModel, since it relates to a different database table.

… used by the stellar-rpc package dependency

…e fetched from the RPC, based on the RPC available ledgers and the local "curstor"

…d limit, since this is how the API is meant to be used.

… transactions.

Copilot

Pull Request Overview

This PR implements a new getLedgers RPC endpoint in the indexer’s RPC service, adds parallel ledger batch ingestion logic to the ingest service (with advisory locking), and introduces a dedicated IngestStoreModel to track the ingestion cursor.

Added GetLedgers to RPCService, updated error handling, and tests in rpc_service_test.go.
Extended ingest service with a new Run loop to fetch and process ledger batches in parallel.
Introduced IngestStoreModel and updated Models to support ingest cursor storage.

Reviewed Changes

Copilot reviewed 15 out of 16 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
internal/services/rpc_service.go	Added `GetLedgers` method and updated `sendRPCRequest` error message
internal/services/rpc_service_test.go	New tests for `GetLedgers` and error path adjustments
internal/services/ingest.go	New `Run` logic with batch fetching, parallel processing, and locking
internal/services/mocks.go	Added `GetLedgers` mock method
internal/services/ingest_test.go	Tests for `getLedgerSeqRange` and `getLedgerTransactions`
internal/indexer/types/types.go	Changed `Transaction.LedgerNumber` from `int64` to `uint32`
internal/data/ingest_store.go	New `IngestStoreModel` for cursor persistence
internal/data/models.go	Registered `IngestStoreModel` in `Models`

Comments suppressed due to low confidence (5)

internal/indexer/types/types.go:21

Changing LedgerNumber from int64 to uint32 is a breaking change in the API contract; downstream consumers expecting signed integers may be impacted.

	LedgerNumber    uint32    `json:"ledgerNumber,omitempty" db:"ledger_number"`

internal/services/rpc_service.go:147

[nitpick] Consider adding a comment for the GetLedgersResponse type alias to document its purpose and linkage to protocol.GetLedgersResponse.

type GetLedgersResponse protocol.GetLedgersResponse

internal/services/ingest_test.go:36

Consider adding tests for fetchNextLedgersBatch to cover cases where RPC indicates ingestion is in sync (ErrAlreadyInSync) and RPC call failures for more complete coverage.

func Test_getLedgerSeqRange(t *testing.T) {

internal/services/rpc_service.go:161

Missing import for "encoding/json"; the call to json.Unmarshal will not compile without importing the package.

	err = json.Unmarshal(resultBytes, &result)

internal/services/ingest.go

marcelosalloum self-assigned this Jun 18, 2025

marcelosalloum requested a review from Copilot June 18, 2025 23:46

This comment was marked as outdated.

Sign in to view

marcelosalloum commented Jun 19, 2025

View reviewed changes

marcelosalloum marked this pull request as ready for review June 19, 2025 00:18

marcelosalloum changed the title ~~[WIP] services: Implement call to getLedgers and start parsing the xdr.LedgerCloseMeta~~ services: Implement call to getLedgers and start parsing the xdr.LedgerCloseMeta Jun 19, 2025

marcelosalloum requested a review from aditya1702 June 19, 2025 00:18

marcelosalloum changed the title ~~services: Implement call to getLedgers and start parsing the xdr.LedgerCloseMeta~~ services: Implement call to getLedgers and start parsing the xdr.LedgerCloseMeta Jun 23, 2025

marcelosalloum mentioned this pull request Jun 23, 2025

State Change Indexer: Create Ingestion Pipeline #128

Open

marcelosalloum added this to the Wallet Backend Indexer milestone Jun 23, 2025

aditya1702 removed this from the Wallet Backend Indexer milestone Jun 23, 2025

aditya1702 reviewed Jun 24, 2025

View reviewed changes

marcelosalloum added 5 commits June 24, 2025 15:47

Install stellar-rpc.

b835043

Migrate the latest ledger synched tracker from the PaymentsModel into…

0d1b9cc

… a new IngestStoreModel, since it relates to a different database table.

Change the LedgerNumber from int64 to uint32, which is the same being…

c618e44

… used by the stellar-rpc package dependency

Implement the rpcService.getLedgers.

afe47c1

Implement the AcquireAdvisoryLock and ReleaseAdvisoryLock functions.

18fe45e

marcelosalloum force-pushed the marcelo/start-indexer branch from 866e082 to 169087f Compare June 24, 2025 22:55

marcelosalloum added 8 commits June 24, 2025 15:55

Implement the getLedgerSeqRange to calculate the Ledgers range to b…

0a04bc0

…e fetched from the RPC, based on the RPC available ledgers and the local "curstor"

Instead of calculating startLedger and lastLedger, use startLedger an…

43967c6

…d limit, since this is how the API is meant to be used.

Temporarily rename Run to DeprecatedRun

acd5df5

Start streaming the Ledgers and logging the start/end ledgers range

1d502af

storing network passphrase in the ingest service.

b29a91e

Fix ticker logic

a0151ac

Start breaking down ledger ranges into individual ledgers and then to…

64434d5

… transactions.

Address Copilot code review suggestions.

2bc8bfe

marcelosalloum force-pushed the marcelo/start-indexer branch from 169087f to 5369197 Compare June 24, 2025 22:58

marcelosalloum added 3 commits June 24, 2025 16:11

Add unit tests for getLedgerTransactions.

b3cd960

Address code review suggestions.

21b7168

Polish TrackRPCServiceHealth.

13aa671

Make the ingestion service resilient against an unhealthy RPC server

dc0f8a5

marcelosalloum force-pushed the marcelo/start-indexer branch from 5369197 to dc0f8a5 Compare June 24, 2025 23:58

marcelosalloum requested review from aditya1702 and Copilot June 25, 2025 00:00

Copilot AI reviewed Jun 25, 2025

View reviewed changes

internal/services/ingest.go Outdated Show resolved Hide resolved

polish

df23028

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

services: Implement call to `getLedgers` and start parsing the `xdr.LedgerCloseMeta` #197

services: Implement call to `getLedgers` and start parsing the `xdr.LedgerCloseMeta` #197

Uh oh!

marcelosalloum commented Jun 18, 2025 •

edited

Loading

Uh oh!

This comment was marked as outdated.

Uh oh!

marcelosalloum Jun 19, 2025

Uh oh!

Uh oh!

Uh oh!

aditya1702 Jun 24, 2025

Uh oh!

marcelosalloum Jun 24, 2025

Uh oh!

marcelosalloum Jun 25, 2025

Uh oh!

aditya1702 Jun 24, 2025

Uh oh!

marcelosalloum Jun 24, 2025

Uh oh!

Uh oh!

aditya1702 Jun 24, 2025

Uh oh!

marcelosalloum Jun 24, 2025

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

services: Implement call to getLedgers and start parsing the xdr.LedgerCloseMeta #197

Are you sure you want to change the base?

services: Implement call to getLedgers and start parsing the xdr.LedgerCloseMeta #197

Uh oh!

Conversation

marcelosalloum commented Jun 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Why

Future Work

Issue that this PR addresses

Checklist

PR Structure

Thoroughness

Release

Uh oh!

This comment was marked as outdated.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

services: Implement call to `getLedgers` and start parsing the `xdr.LedgerCloseMeta` #197

services: Implement call to `getLedgers` and start parsing the `xdr.LedgerCloseMeta` #197

marcelosalloum commented Jun 18, 2025 •

edited

Loading