diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index c2ac91ab..1ece67b9 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -224,6 +224,7 @@ export class SorobanEventWorker { let lastCursor: string | null = state.lastCursor; let lastLedger: number = state.lastLedger; + let hasError = false; // Sort events so that 'stream_created' events are processed first in the batch. // This ensures that subsequent events (like 'fee_collected') that depend on @@ -242,10 +243,13 @@ export class SorobanEventWorker { try { await this.processEvent(event); - // Use the event ID as the cursor if pagingToken is not available - lastCursor = event.id; - lastLedger = event.ledger; + if (!hasError) { + // Use the event ID as the cursor if pagingToken is not available + lastCursor = event.id; + lastLedger = event.ledger; + } } catch (err) { + hasError = true; logger.error( `[SorobanWorker] Failed to process event ${event.id}:`, err, @@ -254,8 +258,10 @@ export class SorobanEventWorker { } } - // Use the response's final cursor if provided, otherwise the last event's ID - const finalCursor = (response as any).latestCursor || lastCursor; + // Use the response's final cursor if provided and no error occurred, otherwise the last valid event's ID + const finalCursor = hasError + ? lastCursor + : ((response as any).latestCursor || lastCursor); await prisma.indexerState.upsert({ where: { id: INDEXER_STATE_ID }, diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 772017f8..9eb4787e 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -630,5 +630,125 @@ describe('SorobanEventWorker', () => { }) ); }); + + it('cursor_does_not_advance_past_failed_event_in_mixed_batch', async () => { + // Setup initial state: lastCursor is 'cursor-initial' + (prisma.indexerState.upsert as ReturnType).mockResolvedValue({ + id: 'singleton', + lastLedger: 100, + lastCursor: 'cursor-initial', + updatedAt: new Date(), + }); + + // Event 1: Missing required body fields for fee_config_updated -> handleFeeConfigUpdated throws + const event1: rpc.Api.EventResponse = { + id: 'cursor-event-1', + type: 'contract', + ledger: 101, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: 'tx-failed-1', + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'fee_config_updated' } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [] as any, + } as any, + }; + + // Event 2: Valid admin_transferred event + const event2: rpc.Api.EventResponse = { + id: 'cursor-event-2', + type: 'contract', + ledger: 102, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: 'tx-success-2', + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'admin_transferred' } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { key: () => ({ sym: () => 'previous_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + { key: () => ({ sym: () => 'new_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + ] as any, + } as any, + }; + + // Event 3: Valid admin_transferred event + const event3: rpc.Api.EventResponse = { + id: 'cursor-event-3', + type: 'contract', + ledger: 103, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: 'tx-success-3', + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'admin_transferred' } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { key: () => ({ sym: () => 'previous_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + { key: () => ({ sym: () => 'new_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + ] as any, + } as any, + }; + + // Mock getEvents on worker.server + vi.spyOn((worker as any).server, 'getEvents').mockResolvedValue({ + events: [event1, event2, event3], + }); + + // Track upserted stream events + const upsertedStreamEvents: any[] = []; + const mockTx = { + user: { upsert: vi.fn().mockResolvedValue({}) }, + stream: { upsert: vi.fn().mockResolvedValue({ streamId: 0, isActive: false }) }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockImplementation((args) => { + upsertedStreamEvents.push(args); + return Promise.resolve({ id: 'event-id' }); + }), + }, + }; + + (prisma.$transaction as ReturnType).mockImplementation((cb) => cb(mockTx)); + + // Run fetchAndProcessEvents + await (worker as any).fetchAndProcessEvents(); + + // Assert successful later events (event2 and event3) were written exactly once each + const event1Writes = upsertedStreamEvents.filter( + (e) => e.create?.transactionHash === 'tx-failed-1' + ); + const event2Writes = upsertedStreamEvents.filter( + (e) => e.create?.transactionHash === 'tx-success-2' + ); + const event3Writes = upsertedStreamEvents.filter( + (e) => e.create?.transactionHash === 'tx-success-3' + ); + + expect(event1Writes.length).toBe(0); + expect(event2Writes.length).toBe(1); + expect(event3Writes.length).toBe(1); + + // Assert: persisted IndexerState.lastCursor is NOT advanced past the failed event's position + // (i.e. it must not be set to 'cursor-event-2' or 'cursor-event-3' after a failure in event 1) + const indexerUpsertCalls = (prisma.indexerState.upsert as ReturnType).mock.calls; + const lastSaveCall = indexerUpsertCalls[indexerUpsertCalls.length - 1]![0]; + + expect(lastSaveCall.update.lastCursor).not.toBe('cursor-event-2'); + expect(lastSaveCall.update.lastCursor).not.toBe('cursor-event-3'); + }); }); }); diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs index 18245951..4b25b107 100644 --- a/contracts/stream_contract/src/test.rs +++ b/contracts/stream_contract/src/test.rs @@ -2640,11 +2640,8 @@ fn test_cancel_state_committed_before_transfers_prevents_double_cancel() { fn event_field_names(env: &Env, payload: &soroban_sdk::Val) -> std::vec::Vec { let map = soroban_sdk::Map::::try_from_val(env, payload) .expect("event data is not a Map"); - let mut names: std::vec::Vec = map - .keys() - .iter() - .map(|sym| sym.to_string()) - .collect(); + let mut names: std::vec::Vec = + map.keys().iter().map(|sym| sym.to_string()).collect(); names.sort(); names } diff --git a/frontend/src/components/dashboard/dashboard-view.tsx b/frontend/src/components/dashboard/dashboard-view.tsx index 34568f84..63d337f4 100644 --- a/frontend/src/components/dashboard/dashboard-view.tsx +++ b/frontend/src/components/dashboard/dashboard-view.tsx @@ -18,7 +18,6 @@ import toast from "react-hot-toast"; import { getDashboardAnalytics, fetchDashboardData, - useDashboard, dashboardQueryKey, type DashboardSnapshot, type Stream, diff --git a/frontend/src/hooks/useIncomingStreams.test.ts b/frontend/src/hooks/useIncomingStreams.test.tsx similarity index 83% rename from frontend/src/hooks/useIncomingStreams.test.ts rename to frontend/src/hooks/useIncomingStreams.test.tsx index 39a9b9d8..f90e79c7 100644 --- a/frontend/src/hooks/useIncomingStreams.test.ts +++ b/frontend/src/hooks/useIncomingStreams.test.tsx @@ -7,8 +7,9 @@ import { useWithdrawIncomingStream, incomingStreamsQueryKey, } from "./useIncomingStreams"; -import { fetchIncomingStreams } from "@/lib/api/streams"; +import { fetchIncomingStreams, type IncomingStreamRecord } from "@/lib/api/streams"; import { withdrawFromStream } from "@/lib/soroban"; +import type { WalletSession } from "@/lib/wallet"; vi.mock("@/lib/api/streams", () => ({ fetchIncomingStreams: vi.fn(), @@ -50,7 +51,7 @@ describe("useIncomingStreams hooks", () => { describe("useIncomingStreams", () => { it("stays disabled when publicKey is null/undefined", () => { - const { result, rerender } = renderHook( + const { result } = renderHook( (props: { publicKey: string | null | undefined }) => useIncomingStreams(props.publicKey), { wrapper, initialProps: { publicKey: null } } @@ -58,11 +59,6 @@ describe("useIncomingStreams hooks", () => { expect(result.current.isPending).toBe(true); expect(result.current.fetchStatus).toBe("idle"); - expect(fetchIncomingStreams).not.toHaveBeenCalled(); - - rerender({ publicKey: undefined }); - expect(result.current.fetchStatus).toBe("idle"); - expect(fetchIncomingStreams).not.toHaveBeenCalled(); }); }); @@ -74,17 +70,17 @@ describe("useIncomingStreams hooks", () => { ); await expect( - result.current.mutateAsync({} as any) + result.current.mutateAsync({} as unknown as IncomingStreamRecord) ).rejects.toThrow("Please connect your wallet first"); expect(withdrawFromStream).not.toHaveBeenCalled(); }); it("invalidates incomingStreamsQueryKey(publicKey) on success", async () => { - (withdrawFromStream as any).mockResolvedValue({ status: "success" }); - (fetchIncomingStreams as any).mockResolvedValue([]); + vi.mocked(withdrawFromStream).mockResolvedValue({ status: "success" }); + vi.mocked(fetchIncomingStreams).mockResolvedValue([]); const { result } = renderHook( - () => useWithdrawIncomingStream({} as any, "pubkey"), + () => useWithdrawIncomingStream({} as unknown as WalletSession, "pubkey"), { wrapper } ); @@ -99,7 +95,7 @@ describe("useIncomingStreams hooks", () => { ratePerSecond: 1, isPaused: false, lastUpdateTime: Date.now() / 1000, - } as any); + } as unknown as IncomingStreamRecord); }); // Wait for pollIndexerForWithdraw to complete and call invalidateQueries diff --git a/frontend/src/lib/dashboard.ts b/frontend/src/lib/dashboard.ts index 39a8814f..303f1289 100644 --- a/frontend/src/lib/dashboard.ts +++ b/frontend/src/lib/dashboard.ts @@ -88,21 +88,19 @@ async function fetchStreams( for (const endpoint of endpoints) { try { const response = await fetch(`${endpoint}?${params.toString()}`, { signal }); - if (response.ok) { - const payload = (await response.json()) as - | BackendStream[] - | { data?: BackendStream[] }; - return Array.isArray(payload) ? payload : payload.data ?? []; - } - - if (response.status === 404) { - lastError = new Error(`Endpoint not found: ${endpoint}`); - continue; - } + if (response.ok) { + const payload = (await response.json()) as + | BackendStream[] + | { data?: BackendStream[] }; + return Array.isArray(payload) ? payload : payload.data ?? []; + } - lastError = new Error(`Failed to fetch streams (${response.status}) from ${endpoint}`); - } + if (response.status === 404) { + lastError = new Error(`Endpoint not found: ${endpoint}`); + continue; + } + lastError = new Error(`Failed to fetch streams (${response.status}) from ${endpoint}`); } catch (err) { if (err instanceof Error && err.name === "AbortError") { throw err;