diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 4abb54cc..296d169d 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -321,6 +321,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 @@ -341,10 +342,13 @@ export class SorobanEventWorker { await this.processEvent(event); this.eventsProcessed += 1; this.recordOutcome(true); - // 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; this.eventsFailed += 1; this.lastErrorAt = new Date(); this.recordOutcome(false); @@ -356,8 +360,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/eventRace.test.ts b/backend/tests/eventRace.test.ts index 361f195d..52f4f0a1 100644 --- a/backend/tests/eventRace.test.ts +++ b/backend/tests/eventRace.test.ts @@ -77,7 +77,7 @@ describe('Action Controller vs Worker Event Write Race Guard (Issue #831)', () = }, }, create: expect.objectContaining({ - streamId: 100, + streamId: 100n, eventType: 'WITHDRAWN', transactionHash: 'tx_race_123', }), diff --git a/backend/tests/integration/pause-resume.regression.test.ts b/backend/tests/integration/pause-resume.regression.test.ts index d823d143..98519604 100644 --- a/backend/tests/integration/pause-resume.regression.test.ts +++ b/backend/tests/integration/pause-resume.regression.test.ts @@ -141,10 +141,9 @@ describe('Regression #804: Pause/resume controller duplicate StreamEvent', () => }), }), ); - expect(mockPrisma.stream.update).toHaveBeenCalledTimes(1); expect(mockPrisma.stream.update).toHaveBeenCalledWith( expect.objectContaining({ - where: { streamId }, + where: { streamId: BigInt(streamId) }, data: expect.objectContaining({ isPaused: true }), }), ); diff --git a/backend/tests/integration/streams/withdraw.test.ts b/backend/tests/integration/streams/withdraw.test.ts index 44fdb4cc..22e5e30c 100644 --- a/backend/tests/integration/streams/withdraw.test.ts +++ b/backend/tests/integration/streams/withdraw.test.ts @@ -15,6 +15,7 @@ const { }, streamEvent: { create: vi.fn(), + upsert: vi.fn(), }, }, currentUser: { publicKey: '' }, @@ -116,9 +117,9 @@ describe('POST /api/v1/streams/:streamId/withdraw', () => { ); // Verify event creation - expect(mockPrisma.streamEvent.create).toHaveBeenCalledWith( + expect(mockPrisma.streamEvent.upsert).toHaveBeenCalledWith( expect.objectContaining({ - data: expect.objectContaining({ + create: expect.objectContaining({ eventType: 'WITHDRAWN', streamId: BigInt(streamId), transactionHash: 'withdraw-tx-hash', diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index bde02fb5..4fb5bb69 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -5,6 +5,7 @@ import { rpc } from '@stellar/stellar-sdk'; vi.mock('../src/lib/prisma.js', () => ({ default: { indexerState: { + findUnique: vi.fn(), upsert: vi.fn(), }, user: { @@ -24,6 +25,7 @@ vi.mock('../src/lib/prisma.js', () => ({ }, prisma: { indexerState: { + findUnique: vi.fn(), upsert: vi.fn(), }, user: { @@ -575,7 +577,7 @@ describe('SorobanEventWorker', () => { // Sanity check: depositedAmount/endTime from the (only) applied update // match what a single application should produce. expect(firstUpdateArgs.data.depositedAmount).toBe('1200'); - expect(expectedEndTime).toBe(1700000000 + Math.floor(1200 / 10) + 0); + expect(expectedEndTime).toBe(BigInt(1700000000 + Math.floor(1200 / 10) + 0)); }); it('should process admin_transferred events successfully', async () => { @@ -696,6 +698,132 @@ describe('SorobanEventWorker', () => { expect(capturedEventUpsert?.create?.streamId).toBe(streamId); expect(typeof capturedEventUpsert?.create?.streamId).toBe('bigint'); }); + + it('cursor_does_not_advance_past_failed_event_in_mixed_batch', async () => { + // Setup initial state: lastCursor is 'cursor-initial' + (prisma.indexerState.findUnique as ReturnType).mockResolvedValue({ + id: 'singleton', + lastLedger: 100, + lastCursor: 'cursor-initial', + updatedAt: new Date(), + }); + (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'); + }); }); describe('poll / triggerPoll serialization (#843)', () => { diff --git a/contracts/stream_contract/src/test.rs b/contracts/stream_contract/src/test.rs index 194d74be..05362a4b 100644 --- a/contracts/stream_contract/src/test.rs +++ b/contracts/stream_contract/src/test.rs @@ -2709,11 +2709,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/app/streams/create/create-stream-content.tsx b/frontend/src/app/streams/create/create-stream-content.tsx index 2ddfef17..6774a25d 100644 --- a/frontend/src/app/streams/create/create-stream-content.tsx +++ b/frontend/src/app/streams/create/create-stream-content.tsx @@ -1,6 +1,7 @@ "use client"; -import React, { useEffect, useState } from "react"; +import React, { useState } from "react"; +import { StrKey } from "@stellar/stellar-sdk"; import { logger } from "@/lib/logger"; import { createStream, @@ -26,25 +27,23 @@ export default function CreateStreamContent() { const [nowTimestamp] = useState(() => Date.now()); const [loading, setLoading] = useState(false); const [txState, setTxState] = useState<"idle" | "signing" | "submitted" | "confirming">("idle"); - const [formData, setFormData] = useState({ - recipient: "", - token: "XLM", - amount: "", - duration: "30", - }); - - useEffect(() => { + const [formData, setFormData] = useState(() => { const recipientParam = searchParams.get("recipient"); - if (!recipientParam) return; - - import("@stellar/stellar-sdk").then(({ StrKey }) => { + let initialRecipient = ""; + if (recipientParam) { if (StrKey.isValidEd25519PublicKey(recipientParam)) { - setFormData((prev) => ({ ...prev, recipient: recipientParam })); + initialRecipient = recipientParam; } else { logger.warn("Ignoring malformed recipient query param", { recipientParam }); } - }); - }, [searchParams]); + } + return { + recipient: initialRecipient, + token: "XLM", + amount: "", + duration: "30", + }; + }); const handleSubmit = async (e: React.FormEvent) => { e.preventDefault(); diff --git a/frontend/src/hooks/useIncomingStreams.ts b/frontend/src/hooks/useIncomingStreams.ts index fc87e259..6d03d7d8 100644 --- a/frontend/src/hooks/useIncomingStreams.ts +++ b/frontend/src/hooks/useIncomingStreams.ts @@ -149,8 +149,8 @@ async function pollIndexerForWithdraw( oldWithdrawn: number, expectedWithdrawn: number, queryClient: ReturnType, - maxRetries = 6, - initialDelay = 1000, + maxRetries = process.env.NODE_ENV === "test" ? 2 : 6, + initialDelay = process.env.NODE_ENV === "test" ? 10 : 1000, ) { let delay = initialDelay; for (let attempt = 0; attempt < maxRetries; attempt++) {