feat(indexer): replace polling with Stellar RPC event fetching - #666
Conversation
|
@MJ-RWA is attempting to deploy a commit to the ritik4ever's projects Team on Vercel. A member of the Team first needs to authorize it. |
|
@MJ-RWA Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits. You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀 |
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughThe indexer now supports persisted ledger checkpoints, cursor-based event pagination, configurable fallback polling, startup checkpoint recovery, concurrency and circuit-breaker guards, and expanded tests for pagination, failures, duplicates, and fallback behavior. ChangesIndexer reliability
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant App
participant startIndexer
participant indexEvents
participant StellarRPC
participant indexer_cursor
participant recordEventWithDb
App->>startIndexer: initialize indexer
startIndexer->>indexer_cursor: load last processed ledger
startIndexer->>indexEvents: schedule polling with interval
startIndexer->>indexEvents: run initial indexing attempt
loop polling cycle
indexEvents->>StellarRPC: fetch events from ledger or cursor
StellarRPC-->>indexEvents: return events and pagination cursor
indexEvents->>recordEventWithDb: process events transactionally
recordEventWithDb-->>indexEvents: confirmed recorded
indexEvents->>indexer_cursor: persist checkpoint
end
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (3)
backend/src/services/indexer.test.ts (2)
111-117: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winWall-clock driven polling makes every test timing-dependent.
runOnePollsleeps a fixed 150 ms with a 50 ms interval, so 3–4 poll cycles execute per "one poll". Exact-count assertions (Lines 265, 278) only pass becauselastProcessedLedgerhappens to short-circuit later cycles. Consider driving the indexer withvi.useFakeTimers()+await vi.advanceTimersByTimeAsync(...), or exporting a singleindexEvents()invocation for tests.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/src/services/indexer.test.ts` around lines 111 - 117, Make runOnePoll deterministic by replacing its wall-clock setTimeout wait with Vitest fake timers and an explicit advanceTimersByTimeAsync call, or invoke an exported single-cycle indexEvents operation. Ensure each test run performs exactly one intended polling cycle and keeps the indexer cleanup via stopIndexer.
230-240: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThis test does not exercise a restart.
The body is identical to "saves checkpoint to database after processing events" (Lines 207-217) — one poll, one assertion. To cover restart recovery, call
stopIndexer(), re-runinitIndexeragainst the samedb, and assert the checkpoint is reloaded rather than reset to 0.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/src/services/indexer.test.ts` around lines 230 - 240, Update the “does not reset checkpoint on restart when events are already processed” test to exercise an actual restart: after the first run and checkpoint assertion, call stopIndexer(), reinitialize the indexer with the same db, run polling again, and assert the persisted checkpoint is reloaded rather than reset to zero. Keep the existing event setup and checkpoint expectations intact.backend/src/services/indexer.ts (1)
184-191: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
indexer_cursortable DDL is duplicated across three files with no single source of truth. The sameCREATE TABLE IF NOT EXISTS indexer_cursor (...)statement is independently maintained in the migration SQL, the migration TypeScript wrapper, and the indexer's own startup helper; any future schema change must be applied in all three or they silently drift.
backend/src/services/indexer.ts#L184-L191: replace the inlinedb.exec(...)DDL inensureIndexerCursorTablewith a call to the migration's exportedup(db)function.backend/src/migrations/0003_add_indexer_cursor.ts#L3-L10: keep as the single source of truth for theup()DDL that both the migration runner andindexer.tsinvoke.backend/src/migrations/0003_add_indexer_cursor.sql#L9-L12: keep as the canonical raw-SQL reference (or generate it from the.tsmigration) so all three don't need independent edits.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/src/services/indexer.ts` around lines 184 - 191, The indexer cursor table DDL is duplicated across the startup helper and migration files. In backend/src/services/indexer.ts lines 184-191, replace the inline DDL in ensureIndexerCursorTable with a call to the exported up(db) from backend/src/migrations/0003_add_indexer_cursor.ts; keep that up() implementation as the migration source of truth, and retain backend/src/migrations/0003_add_indexer_cursor.sql lines 9-12 as the canonical raw-SQL reference or generate it from the TypeScript migration without requiring separate schema edits.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/src/config/validateEnv.ts`:
- Around line 58-65: Align the fallback polling interval floor across all three
sites, using one shared constant where possible: update
backend/src/config/validateEnv.ts lines 58-65 and its validation message,
backend/.env.example lines 39-41, and backend/src/services/indexer.ts lines
105-107 so they consistently enforce and document the intended minimum, with the
runtime Math.max using the same floor as the schema.
In `@backend/src/services/indexer.test.ts`:
- Around line 219-228: The test at backend/src/services/indexer.test.ts lines
219-228 must assert the checkpoint behavior exercised by initIndexer, such as
polling with getLatestLedger at 501 and verifying getEvents receives startLedger
501 (or that the cursor remains unchanged when sequence <= 500), instead of
asserting the local event fixture. The test at
backend/src/services/indexer.test.ts lines 369-388 must verify actual
deduplication by querying stream_events for stream_id '1' and event_type
'claimed' and expecting a count of 1, replacing the unconditional
mock-call-length comparison.
- Around line 328-357: Isolate the RPC error-handling tests from later suites by
cleaning up fallback-mode configuration and module-level indexer state. Add an
afterEach that removes INDEXER_FALLBACK_POLLING_ENABLED, calls
vi.resetModules(), and re-imports the indexer entry points used by runOnePoll so
lastProcessedLedger, isIndexing, and the circuit breaker reset between tests.
Preserve the existing test assertions and setup behavior.
In `@backend/src/services/indexer.ts`:
- Around line 286-292: Update indexEventsWithFallback and
indexEventsWithCursorPagination so a completed scan with zero matching events
still advances lastProcessedLedger to currentLedger and persists the checkpoint.
Preserve existing event-processing behavior, but ensure the zero-event path
saves the confirmed currentLedger instead of returning or skipping the
checkpoint update.
- Around line 19-20: Replace the raw environment parsing in indexer.ts with the
validated indexerFallbackPollingEnabled and indexerFallbackPollIntervalMs values
from validateEnv(). Thread these settings through startIndexer() and
initIndexer(), and remove the local FALLBACK_POLLING_ENABLED and
FALLBACK_POLL_INTERVAL_MS constants while preserving the existing fallback
polling behavior.
- Around line 109-125: Update loadCheckpoint’s catch block to log the database
error and fail the current load attempt without assigning lastProcessedLedger =
0, preserving the existing checkpoint so the caller’s retry or circuit-breaker
flow can handle the failure. Keep the ledger-0 fallback only for a successful
query that finds no valid checkpoint.
- Around line 268-305: The single-call fallback path in indexEventsWithFallback
can skip events beyond the first getEvents page. Add cursor-based pagination
using the response cursor, process every returned page before checkpointing, and
set lastProcessedLedger to the highest ledger actually processed rather than
currentLedger; preserve the existing transaction, metrics, and checkpoint
behavior after all pages complete.
---
Nitpick comments:
In `@backend/src/services/indexer.test.ts`:
- Around line 111-117: Make runOnePoll deterministic by replacing its wall-clock
setTimeout wait with Vitest fake timers and an explicit advanceTimersByTimeAsync
call, or invoke an exported single-cycle indexEvents operation. Ensure each test
run performs exactly one intended polling cycle and keeps the indexer cleanup
via stopIndexer.
- Around line 230-240: Update the “does not reset checkpoint on restart when
events are already processed” test to exercise an actual restart: after the
first run and checkpoint assertion, call stopIndexer(), reinitialize the indexer
with the same db, run polling again, and assert the persisted checkpoint is
reloaded rather than reset to zero. Keep the existing event setup and checkpoint
expectations intact.
In `@backend/src/services/indexer.ts`:
- Around line 184-191: The indexer cursor table DDL is duplicated across the
startup helper and migration files. In backend/src/services/indexer.ts lines
184-191, replace the inline DDL in ensureIndexerCursorTable with a call to the
exported up(db) from backend/src/migrations/0003_add_indexer_cursor.ts; keep
that up() implementation as the migration source of truth, and retain
backend/src/migrations/0003_add_indexer_cursor.sql lines 9-12 as the canonical
raw-SQL reference or generate it from the TypeScript migration without requiring
separate schema edits.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 0094d24a-d4e4-46df-8ed1-4282cedcb3ca
⛔ Files ignored due to path filters (2)
backend/package-lock.jsonis excluded by!**/package-lock.jsonpackage-lock.jsonis excluded by!**/package-lock.json
📒 Files selected for processing (7)
backend/.env.examplebackend/src/config/validateEnv.tsbackend/src/migrations/0003_add_indexer_cursor.down.sqlbackend/src/migrations/0003_add_indexer_cursor.sqlbackend/src/migrations/0003_add_indexer_cursor.tsbackend/src/services/indexer.test.tsbackend/src/services/indexer.ts
| // Indexer fallback polling interval validation | ||
| const fallbackPollIntervalSchema = z | ||
| .string() | ||
| .transform((val: string) => parseInt(val, 10)) | ||
| .refine((val: number) => !isNaN(val) && val >= 1000, { | ||
| message: "must be a valid number >= 1000 (minimum 1 second)", | ||
| }); | ||
|
|
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fallback poll interval minimum disagrees across docs, schema, and runtime floor (5000ms vs 1000ms). backend/.env.example documents a 5000ms minimum, but backend/src/config/validateEnv.ts's Zod schema enforces >= 1000 and backend/src/services/indexer.ts's runtime floor is Math.max(1000, ...) — both consistent with each other at 1000ms, but not with the documented 5000ms.
backend/src/config/validateEnv.ts#L58-L65: either update the.refine(val >= 1000, ...)check (and message) to>= 5000if 5000ms is the intended floor, or update the.env.examplecomment to say 1000ms if 1000ms is intended.backend/.env.example#L39-L41: update the "minimum 5000" comment to match whichever floor is chosen invalidateEnv.ts.backend/src/services/indexer.ts#L105-L107: updateMath.max(1000, FALLBACK_POLL_INTERVAL_MS)to use the same floor constant asvalidateEnv.ts(ideally sourced from one shared constant) so all three stay in sync.
📍 Affects 3 files
backend/src/config/validateEnv.ts#L58-L65(this comment)backend/.env.example#L39-L41backend/src/services/indexer.ts#L105-L107
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/config/validateEnv.ts` around lines 58 - 65, Align the fallback
polling interval floor across all three sites, using one shared constant where
possible: update backend/src/config/validateEnv.ts lines 58-65 and its
validation message, backend/.env.example lines 39-41, and
backend/src/services/indexer.ts lines 105-107 so they consistently enforce and
document the intended minimum, with the runtime Math.max using the same floor as
the schema.
| it("loads checkpoint from database on initIndexer", async () => { | ||
| const cid = nextContractId(); | ||
| setupDb(cid, 500); | ||
| const event = makeClaimedEvent({ streamId: "1", ledger: 501 }); | ||
| mockGetEvents.mockResolvedValue({ events: [event] }); | ||
|
|
||
| initIndexer("https://rpc.example.com", cid, "Test SDF Network ; September 2015"); | ||
|
|
||
| expect(event).toBeDefined(); | ||
| }); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Two tests assert conditions that can never fail. Both cases exercise setup but then assert something tautological, so the named behavior is not actually covered.
backend/src/services/indexer.test.ts#L219-L228:expect(event).toBeDefined()only checks a local fixture. Assert the loaded checkpoint instead — e.g. afterinitIndexer, run a poll withgetLatestLedgerat 501 and verifygetEventswas called withstartLedger501 (not 0), or that the cursor row is unchanged whensequence <= 500.backend/src/services/indexer.test.ts#L369-L388:callsAfterSecond >= callsAfterFirstholds unconditionally because the mock call array only grows. Assert real deduplication instead — querySELECT COUNT(*) FROM stream_events WHERE stream_id = '1' AND event_type = 'claimed'and expect1.
📍 Affects 1 file
backend/src/services/indexer.test.ts#L219-L228(this comment)backend/src/services/indexer.test.ts#L369-L388
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/services/indexer.test.ts` around lines 219 - 228, The test at
backend/src/services/indexer.test.ts lines 219-228 must assert the checkpoint
behavior exercised by initIndexer, such as polling with getLatestLedger at 501
and verifying getEvents receives startLedger 501 (or that the cursor remains
unchanged when sequence <= 500), instead of asserting the local event fixture.
The test at backend/src/services/indexer.test.ts lines 369-388 must verify
actual deduplication by querying stream_events for stream_id '1' and event_type
'claimed' and expecting a count of 1, replacing the unconditional
mock-call-length comparison.
| describe("indexer RPC error handling", () => { | ||
| let ledgerSeq = 600; | ||
|
|
||
| beforeEach(() => { | ||
| vi.clearAllMocks(); | ||
| ledgerSeq += 200; | ||
| mockGetLatestLedger.mockResolvedValue({ sequence: ledgerSeq }); | ||
| }); | ||
|
|
||
| it("does not crash when getEvents throws an error", async () => { | ||
| const cid = nextContractId(); | ||
| setupDb(cid, ledgerSeq - 100); | ||
| mockGetEvents.mockRejectedValue(new Error("RPC timeout")); | ||
|
|
||
| await runOnePoll(cid); | ||
|
|
||
| expect(mockRecordEventWithDb).not.toHaveBeenCalled(); | ||
| }); | ||
|
|
||
| it("does not advance checkpoint when getEvents fails", async () => { | ||
| process.env.INDEXER_FALLBACK_POLLING_ENABLED = "true"; | ||
| const cid = nextContractId(); | ||
| setupDb(cid, ledgerSeq - 100); | ||
| mockGetEvents.mockRejectedValue(new Error("RPC error")); | ||
|
|
||
| await runOnePoll(cid); | ||
|
|
||
| const row = db.prepare("SELECT last_ledger_sequence FROM indexer_cursor WHERE id = 1").get() as { last_ledger_sequence: number }; | ||
| expect(row.last_ledger_sequence).toBe(ledgerSeq - 100); | ||
| }); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
INDEXER_FALLBACK_POLLING_ENABLED leaks into subsequent suites, and indexer module state is never reset.
Line 348 sets the env var with no cleanup — the "indexer duplicate prevention" describe (Lines 360+) then runs in fallback mode depending on file order. Separately, indexer.ts keeps lastProcessedLedger, isIndexing, and the circuit breaker in module scope; the two forced RPC failures here push the breaker toward OPEN, after which indexEvents returns early and later tests can pass vacuously. Add an afterEach (already imported at Line 5) to clear the env var, and vi.resetModules() + re-import to isolate indexer state.
🔧 Suggested cleanup
beforeEach(() => {
vi.clearAllMocks();
ledgerSeq += 200;
mockGetLatestLedger.mockResolvedValue({ sequence: ledgerSeq });
});
+
+ afterEach(() => {
+ delete process.env.INDEXER_FALLBACK_POLLING_ENABLED;
+ });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/services/indexer.test.ts` around lines 328 - 357, Isolate the RPC
error-handling tests from later suites by cleaning up fallback-mode
configuration and module-level indexer state. Add an afterEach that removes
INDEXER_FALLBACK_POLLING_ENABLED, calls vi.resetModules(), and re-imports the
indexer entry points used by runOnePoll so lastProcessedLedger, isIndexing, and
the circuit breaker reset between tests. Preserve the existing test assertions
and setup behavior.
| const FALLBACK_POLLING_ENABLED = process.env.INDEXER_FALLBACK_POLLING_ENABLED === "true"; | ||
| const FALLBACK_POLL_INTERVAL_MS = Number(process.env.INDEXER_FALLBACK_POLL_INTERVAL_MS ?? 10000); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP -A5 -B5 'startIndexer\s*\(' --type=ts | grep -v indexer.test.ts
rg -nP 'validateEnv\s*\(' --type=ts -g '!*.test.ts'Repository: ritik4ever/stellar-stream
Length of output: 163
🏁 Script executed:
#!/bin/bash
set -u
printf '--- repo files matching indexer / validateEnv ---\n'
git ls-files | rg '(^|/)indexer\.ts$|(^|/)validateEnv\.ts$|backend/src/config/.*\.ts|backend/src/index\.ts' || true
printf '\n--- backend/src/services/indexer.ts outline ---\n'
ast-grep outline backend/src/services/indexer.ts 2>/dev/null || true
printf '\n--- search validateEnv usages/config structure ---\n'
rg -n "validateEnv|fallbackPollIntervalSchema|INDEXER_FALLBACK|fallbackPollingEnabled|fallbackPollIntervalMs" backend/src || true
printf '\n--- relevant validateEnv.ts content ---\n'
fd -a 'validateEnv\.ts$' backend/src | while read -r f; do
echo "### $f"
wc -l "$f"
sed -n '1,220p' "$f"
doneRepository: ritik4ever/stellar-stream
Length of output: 16364
🏁 Script executed:
#!/bin/bash
set -u
printf '%s\n' '--- backend/src/index.ts tail around validateEnv usage ---'
wc -l backend/src/index.ts
sed -n '1800,1900p' backend/src/index.ts
printf '\n--- backend/src/services/indexer.ts relevant lines ---\n'
sed -n '1,60p' backend/src/services/indexer.ts
sed -n '90,126p' backend/src/services/indexer.ts
sed -n '140,220p' backend/src/services/indexer.ts
printf '\n--- behavioral probes for fallback interval edge cases ---\n'
node - <<'JS'
const inputs = ["", "abc", "-50", "0", "500", "1000", "5000", "NaN"];
for (const raw of inputs) {
const v = Number(raw ?? "10000");
const fallbackEnabled = raw === "true";
const fallbackInterval = Number.isNaN(v) ? "NaN" : Math.max(1000, v);
console.log(JSON.stringify({raw, fallbackEnabled, fallbackInterval, isFinite: Number.isFinite(Number(raw ?? "10000"))}));
}
JS
printf '%s\n' 'setInterval(NaN) does not throw synchronously; it coerces delay to 1 in Node.'Repository: ritik4ever/stellar-stream
Length of output: 8667
Use validated fallback-indexer config instead of raw process.env.
indexer.ts re-parses INDEXER_FALLBACK_POLLING_ENABLED and INDEXER_FALLBACK_POLL_INTERVAL_MS with Number(...), bypassing validateEnv(). A malformed fallback interval produces NaN, and Math.max(1000, NaN) stays NaN; setInterval(..., NaN) is coerced by Node to 1ms, causing rapid polling. Since validateEnv() already exposes indexerFallbackPollingEnabled and indexerFallbackPollIntervalMs, thread that validated config through startIndexer()/initIndexer() instead of reading raw env in backend/src/services/indexer.ts.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/services/indexer.ts` around lines 19 - 20, Replace the raw
environment parsing in indexer.ts with the validated
indexerFallbackPollingEnabled and indexerFallbackPollIntervalMs values from
validateEnv(). Thread these settings through startIndexer() and initIndexer(),
and remove the local FALLBACK_POLLING_ENABLED and FALLBACK_POLL_INTERVAL_MS
constants while preserving the existing fallback polling behavior.
| function loadCheckpoint(db: any): void { | ||
| try { | ||
| const row = db | ||
| .prepare(`SELECT last_ledger_sequence FROM ${INDEXER_CURSOR_TABLE} WHERE id = @id`) | ||
| .get({ id: CHECKPOINT_ROW_ID }) as { last_ledger_sequence: number } | undefined; | ||
|
|
||
| if (row && row.last_ledger_sequence > 0) { | ||
| lastProcessedLedger = row.last_ledger_sequence; | ||
| logger.info({ lastProcessedLedger }, "loaded indexer checkpoint from database"); | ||
| } else { | ||
| logger.info("no checkpoint found, starting from ledger 0"); | ||
| } | ||
| } catch (err) { | ||
| logger.error({ err }, "failed to load indexer checkpoint, starting from ledger 0"); | ||
| lastProcessedLedger = 0; | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Checkpoint load failure silently resets progress to ledger 0.
On any DB error (lock contention, corruption, transient I/O), loadCheckpoint resets lastProcessedLedger = 0 and the indexer resumes "from ledger 0". Stellar RPC only retains a short window of event history (commonly ~7 days), so a checkpoint request starting near ledger 0/1 will either error out against the RPC (startLedger below the oldest retained ledger) or, if the range happens to still be retrievable, trigger reprocessing of a huge backlog of already-handled events — directly undermining the "preventing... duplicate processing" goal stated in the migration's own header comment.
Consider failing the poll cycle (leave lastProcessedLedger untouched, let the circuit breaker/retry handle it) rather than defaulting to 0 on transient read errors.
🛡️ Avoid resetting to 0 on transient errors
} catch (err) {
- logger.error({ err }, "failed to load indexer checkpoint, starting from ledger 0");
- lastProcessedLedger = 0;
+ logger.error({ err }, "failed to load indexer checkpoint; leaving lastProcessedLedger unset and deferring to retry");
+ throw err;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| function loadCheckpoint(db: any): void { | |
| try { | |
| const row = db | |
| .prepare(`SELECT last_ledger_sequence FROM ${INDEXER_CURSOR_TABLE} WHERE id = @id`) | |
| .get({ id: CHECKPOINT_ROW_ID }) as { last_ledger_sequence: number } | undefined; | |
| if (row && row.last_ledger_sequence > 0) { | |
| lastProcessedLedger = row.last_ledger_sequence; | |
| logger.info({ lastProcessedLedger }, "loaded indexer checkpoint from database"); | |
| } else { | |
| logger.info("no checkpoint found, starting from ledger 0"); | |
| } | |
| } catch (err) { | |
| logger.error({ err }, "failed to load indexer checkpoint, starting from ledger 0"); | |
| lastProcessedLedger = 0; | |
| } | |
| } | |
| function loadCheckpoint(db: any): void { | |
| try { | |
| const row = db | |
| .prepare(`SELECT last_ledger_sequence FROM ${INDEXER_CURSOR_TABLE} WHERE id = `@id``) | |
| .get({ id: CHECKPOINT_ROW_ID }) as { last_ledger_sequence: number } | undefined; | |
| if (row && row.last_ledger_sequence > 0) { | |
| lastProcessedLedger = row.last_ledger_sequence; | |
| logger.info({ lastProcessedLedger }, "loaded indexer checkpoint from database"); | |
| } else { | |
| logger.info("no checkpoint found, starting from ledger 0"); | |
| } | |
| } catch (err) { | |
| logger.error({ err }, "failed to load indexer checkpoint; leaving lastProcessedLedger unset and deferring to retry"); | |
| throw err; | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/services/indexer.ts` around lines 109 - 125, Update
loadCheckpoint’s catch block to log the database error and fail the current load
attempt without assigning lastProcessedLedger = 0, preserving the existing
checkpoint so the caller’s retry or circuit-breaker flow can handle the failure.
Keep the ledger-0 fallback only for a successful query that finds no valid
checkpoint.
| async function indexEventsWithFallback(db: any, currentLedger: number): Promise<void> { | ||
| const startLedger = lastProcessedLedger + 1; | ||
| let events; | ||
|
|
||
| try { | ||
| events = await rpcServer.getEvents({ | ||
| startLedger, | ||
| filters: [ | ||
| { | ||
| type: "contract", | ||
| contractIds: [contractId], | ||
| contractIds: [contractId!], | ||
| }, | ||
| ], | ||
| }); | ||
| } catch (err) { | ||
| logger.error({ err }, "RPC getEvents failed in fallback mode"); | ||
| throw err; | ||
| } | ||
|
|
||
| const startLedgerForMetrics = lastProcessedLedger; | ||
| const eventCount = events.events?.length ?? 0; | ||
|
|
||
| if (eventCount === 0) { | ||
| return; | ||
| } | ||
|
|
||
| db.transaction(() => { | ||
| for (const event of events.events || []) { | ||
| processEvent(db, event); | ||
| eventsIndexedTotal.inc(); | ||
| } | ||
|
|
||
| lastProcessedLedger = currentLedger; | ||
| })(); | ||
|
|
||
| saveCheckpoint(db, lastProcessedLedger); | ||
| ledgersScannedTotal.inc(lastProcessedLedger - startLedgerForMetrics); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg 'backend/src/services/indexer.ts|backend/src/validation/schemas.ts|package.json' || true
echo "== indexer outline =="
ast-grep outline backend/src/services/indexer.ts --view expanded || true
echo "== relevant indexer lines =="
nl -ba backend/src/services/indexer.ts | sed -n '230,330p'
echo "== usages of indexEventsWithFallback/indexEventsWithCursorPagination/getEvents =="
rg -n "indexEventsWithFallback|indexEventsWithCursorPagination|getEvents|lastProcessedLedger|currentLedger|cursor" backend/src/services/indexer.ts backend/src -S || trueRepository: ritik4ever/stellar-stream
Length of output: 2468
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== file size =="
wc -l backend/src/services/indexer.ts
echo "== fallback and cursor implementations =="
sed -n '222,380p' backend/src/services/indexer.ts
echo "== fetch/currency/client config for rpc =="
sed -n '1,110p' backend/src/services/indexer.ts
echo "== search for pagination and cursor =="
rg -n "cursor|getEvents|indexEventsWithFallback|indexEventsWithCursorPagination|lastProcessedLedger|currentLedger|checkpoint" backend/src/services/indexer.tsRepository: ritik4ever/stellar-stream
Length of output: 9372
🌐 Web query:
Stellar RPC getEvents pagination default limit max events cursor documentation
💡 Result:
In the Stellar RPC getEvents method, pagination is handled through an optional pagination object passed within the request parameters [1][2]. The pagination object supports the following fields: - limit: An optional integer specifying the maximum number of records to return [1][2]. The allowed range is 1 to 10,000, and if no limit is provided, it defaults to 100 [1][2]. - cursor: An optional string acting as an opaque paging token [1][2]. It is used to fetch the next page of results; each response includes a cursor field that can be provided in a subsequent request to continue pagination [1][2]. When using a cursor, you must omit the startLedger and endLedger parameters from your request [3][4]. If a cursor is not used, you can specify a ledger range using startLedger and endLedger [3]. Note that the 10,000 record limit is hardcoded in Stellar-RPC for performance reasons [1][2].
Citations:
- 1: https://developers.stellar.org/docs/data/apis/rpc/api-reference/structure/pagination
- 2: https://www.alchemy.com/docs/chains/stellar/stellar-api-endpoints/get-events.md
- 3: https://www.alchemy.com/docs/chains/stellar/stellar-api-endpoints/get-events
- 4: https://docs.validationcloud.io/v1/stellar/stellar-rpc-formerly-soroban-api/getevents.md
🌐 Web query:
Stellar RPC Soroban getEvents API object cursor startLedger cursor in response
💡 Result:
The Stellar RPC getEvents method uses a pagination system based on a cursor to navigate through event results [1][2][3]. Key details regarding the relationship between startLedger and the cursor in the request and response are as follows: Request Parameters When making a getEvents request, you can provide either a startLedger or a pagination.cursor, but they are mutually exclusive [1][2][4]: - startLedger: Defines the inclusive ledger sequence number to begin searching from [1][2][5]. - pagination.cursor: An opaque string acting as a paging token used to fetch subsequent pages of results [3]. If a cursor is included in the request, startLedger (and endLedger) must be omitted [1][2][4]. Response Object The getEvents response contains a cursor field [6][7][8]. This cursor is an opaque string that represents the current position in the event stream [3]. To retrieve the next page of events, the client should pass this cursor value back into the pagination.cursor field of a subsequent getEvents request [7][9][3]. Summary of Behavior - Initial Request: Set startLedger to the desired starting point and optionally include a pagination.limit [2][10]. - Subsequent Requests: Extract the cursor from the previous response and include it in the pagination object for the next request [7][3]. Do not include startLedger in these follow-up requests [1][2][4]. - Termination: Pagination is typically complete when no further cursor is returned or the stream reaches the end of the available data [7][10].
Citations:
- 1: https://www.alchemy.com/docs/chains/stellar/stellar-api-endpoints/get-events.md
- 2: https://docs.validationcloud.io/v1/stellar/stellar-rpc-formerly-soroban-api/getevents.md
- 3: https://developers.stellar.org/docs/data/apis/rpc/api-reference/structure/pagination
- 4: https://pub.dev/documentation/stellar_dart/latest/stellar_dart/SorobanRequestGetEvents-class.html
- 5: https://docs.validationcloud.io/v1/stellar/stellar-rpc-formerly-soroban-api/getevents
- 6: https://cdn.jsdelivr.net/npm/stellar-sdk@13.1.0/lib/no-eventsource/rpc/api.d.ts
- 7: https://pub.dev/documentation/stellar_flutter_sdk/latest/stellar_flutter_sdk/GetEventsResponse-class.html
- 8: https://stellar-sdk.readthedocs.io/en/stable/_modules/stellar_sdk/soroban_rpc.html
- 9: https://stellar.github.io/js-soroban-client/Server.html
- 10: https://docs.uniblock.dev/reference/jsonrpc-docs/stellar-soroban/stellar/get-events
🌐 Web query:
site:developers.stellar.org/get-events Soroban RPC GetEventsRequest cursor pagination.cursor
💡 Result:
For Soroban RPC getEvents, pagination.cursor is the opaque paging token used to fetch the next page of results. Set it to the cursor returned in the previous response. If you include cursor, you must omit startLedger and endLedger. The limit for getEvents is 1 to 10000 (default 100). (developers.stellar.org)
Citations:
- 1: https://developers.stellar.org/docs/data/apis/rpc/api-reference/methods/getEvents?utm_source=openai
Loop pages in fallback mode before advancing the checkpoint.
indexEventsWithFallback calls rpcServer.getEvents({ startLedger, filters }) once and then sets lastProcessedLedger = currentLedger, but getEvents defaults to 100 events and only returns a cursor for the next page. If more events exist in the catch-up range, those beyond page one are skipped and can never be recovered because the next poll starts after currentLedger; reuse or copy the cursor-pagination loop and advance the checkpoint to the highest ledger actually processed instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/services/indexer.ts` around lines 268 - 305, The single-call
fallback path in indexEventsWithFallback can skip events beyond the first
getEvents page. Add cursor-based pagination using the response cursor, process
every returned page before checkpointing, and set lastProcessedLedger to the
highest ledger actually processed rather than currentLedger; preserve the
existing transaction, metrics, and checkpoint behavior after all pages complete.
|
|
||
| const startLedgerForMetrics = lastProcessedLedger; | ||
| const eventCount = events.events?.length ?? 0; | ||
|
|
||
| if (eventCount === 0) { | ||
| return; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Checkpoint never advances past currentLedger when no events are found.
Both indexEventsWithFallback (returns early on eventCount === 0 without touching lastProcessedLedger) and indexEventsWithCursorPagination (only updates/saves when totalProcessed > 0) leave the checkpoint stuck at the old value even though currentLedger (fetched via getLatestLedger()) has already moved forward. During a long quiet period with no matching contract events, this means the indexer keeps re-requesting the same growing startLedger…currentLedger range on every poll, and if that idle period exceeds the RPC's retention window, the eventual poll will fail with startLedger below the oldest retained ledger — the indexer becomes permanently stuck and can never recover automatically.
Consider persisting lastProcessedLedger = currentLedger (with a checkpoint save) even when zero events are found, since currentLedger was already confirmed scanned.
Also applies to: 373-377
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/src/services/indexer.ts` around lines 286 - 292, Update
indexEventsWithFallback and indexEventsWithCursorPagination so a completed scan
with zero matching events still advances lastProcessedLedger to currentLedger
and persists the checkpoint. Preserve existing event-processing behavior, but
ensure the zero-event path saves the confirmed currentLedger instead of
returning or skipping the checkpoint update.
|
Hi @MJ-RWA, This PR could not be merged because it has merge conflicts with the target branch. Please resolve the merge conflicts, push the updated changes, and the PR can be reviewed and merged. Thank you! |
2 participants
Summary
Replaces the indexer's fixed-interval event polling with Stellar RPC
getEventsand cursor-based pagination to reduce event processing latency.Changes
Acceptance Criteria
Closes: #599
Summary by CodeRabbit
New Features
Bug Fixes