Skip to content

fix: make soroban event processing idempotent #1342

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
d1a-m3n wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: make soroban event processing idempotent #1342

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
97e6abd
fix: make soroban event processing idempotent
d1a-m3n Aug 30, 2026
ae79336
fix: restore unimplemented pause and resume behavior
d1a-m3n Aug 30, 2026
a91f0a3
Merge branch 'main' into fix/fee-collected-transaction
d1a-m3n Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 10 additions & 62 deletions backend/src/controllers/stream.controller.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,6 @@ import {
getClaimableFromChain,
isStale,
topUpStream,
pauseStream as sorobanPauseStream,
resumeStream as sorobanResumeStream,
} from "../services/sorobanService.js";
import type { AuthenticatedRequest } from "../types/auth.types.js";
import { parseStreamId } from "../lib/stream-id.js";
Expand Down Expand Up @@ -831,36 +829,11 @@ export const pauseStream = async (req: Request, res: Response) => {
});
}

try {
// Call Soroban service to verify the pause operation would succeed
const result = await sorobanPauseStream(
authReq.user.publicKey,
parsedStreamId,
);

logger.info(
`Stream ${parsedStreamId} pause simulated by ${authReq.user.publicKey}`,
);

return res.status(200).json({
success: true,
streamId: parsedStreamId,
txHash: result.txHash,
stream,
});
} catch (sorobanError) {
logger.error(
`Soroban pause failed for stream ${parsedStreamId}:`,
sorobanError,
);
return res.status(400).json({
error: "Failed to pause stream on chain",
message:
sorobanError instanceof Error
? sorobanError.message
: "Unknown error",
});
}
return res.status(501).json({
error: "Not Implemented",
message:
"Pausing streams is not currently supported because the on-chain transaction is not yet submitted.",
});
} catch (error) {
logger.error("Error pausing stream:", error);
return res.status(500).json({ error: "Internal server error" });
Expand Down Expand Up @@ -914,36 +887,11 @@ export const resumeStream = async (req: Request, res: Response) => {
});
}

try {
// Call Soroban service to verify the resume operation would succeed
const result = await sorobanResumeStream(
authReq.user.publicKey,
parsedStreamId,
);

logger.info(
`Stream ${parsedStreamId} resume simulated by ${authReq.user.publicKey}`,
);

return res.status(200).json({
success: true,
streamId: parsedStreamId,
txHash: result.txHash,
stream,
});
} catch (sorobanError) {
logger.error(
`Soroban resume failed for stream ${parsedStreamId}:`,
sorobanError,
);
return res.status(400).json({
error: "Failed to resume stream on chain",
message:
sorobanError instanceof Error
? sorobanError.message
: "Unknown error",
});
}
return res.status(501).json({
error: "Not Implemented",
message:
"Resuming streams is not currently supported because the on-chain transaction is not yet submitted.",
});
} catch (error) {
logger.error("Error resuming stream:", error);
return res.status(500).json({ error: "Internal server error" });
Expand Down
54 changes: 28 additions & 26 deletions backend/src/workers/soroban-event-worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1013,39 +1013,41 @@ export class SorobanEventWorker {
const token = decodeAddress(body["token"]);
const timestamp = Math.floor(Date.now() / 1000);

const existingEvent = await prisma.streamEvent.findUnique({
where: {
transactionHash_eventType: {
transactionHash: event.txHash,
eventType: "FEE_COLLECTED",
},
},
select: { id: true },
});
if (existingEvent) {
logger.warn(
`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=FEE_COLLECTED`,
);
} else {
await prisma.streamEvent.upsert({
await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
const existingEvent = await tx.streamEvent.findUnique({
where: {
transactionHash_eventType: {
transactionHash: event.txHash,
eventType: "FEE_COLLECTED",
},
},
create: {
streamId,
eventType: "FEE_COLLECTED",
amount: feeAmount,
transactionHash: event.txHash,
ledgerSequence: event.ledger,
timestamp,
metadata: JSON.stringify({ treasury, token }),
},
update: {},
select: { id: true },
});
}
if (existingEvent) {
logger.warn(
`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=FEE_COLLECTED`,
);
} else {
await tx.streamEvent.upsert({
where: {
transactionHash_eventType: {
transactionHash: event.txHash,
eventType: "FEE_COLLECTED",
},
},
create: {
streamId,
eventType: "FEE_COLLECTED",
amount: feeAmount,
transactionHash: event.txHash,
ledgerSequence: event.ledger,
timestamp,
metadata: JSON.stringify({ treasury, token }),
},
update: {},
});
}
});

// Broadcast to admin channel for treasury reporting
sseService.broadcastToAdmin("stream.fee_collected", {
Expand Down
2 changes: 1 addition & 1 deletion backend/tests/integration/pause-resume.regression.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ describe('Regression #804: Pause/resume controller duplicate StreamEvent', () =>
.post(`/v1/streams/${streamId}/pause`)
.set('Authorization', `Bearer ${token}`);

expect(pauseRes.status).toBe(200);
expect(pauseRes.status).toBe(501);

// Controller should NOT write to DB for PAUSED event
expect(mockPrisma.streamEvent.create).not.toHaveBeenCalled();
Expand Down
112 changes: 50 additions & 62 deletions backend/tests/integration/stream-actions.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,41 +94,35 @@ describe('stream action routes', () => {
mockPrisma.streamEvent.count.mockResolvedValue(0);
});

it('POST /v1/streams/:streamId/pause pauses an active sender-owned stream', async () => {
const sender = makeKeypair();
const token = await getValidJwt(sender);

mockPrisma.stream.findUnique.mockResolvedValue({
streamId: 7,
sender: sender.publicKey(),
recipient: makeKeypair().publicKey(),
isActive: true,
isPaused: false,
pausedAt: null,
totalPausedDuration: 0,
});
mockPauseStream.mockResolvedValue({ txHash: 'pause-tx-hash' });
mockPrisma.stream.update.mockResolvedValue({
streamId: 7,
isActive: true,
isPaused: true,
pausedAt: 1700000000,
totalPausedDuration: 0,
});
it('POST /v1/streams/:streamId/pause returns 501 when pausing is not implemented', async () => {
const sender = makeKeypair();
const token = await getValidJwt(sender);

mockPrisma.stream.findUnique.mockResolvedValue({
streamId: 7,
sender: sender.publicKey(),
recipient: makeKeypair().publicKey(),
isActive: true,
isPaused: false,
pausedAt: null,
totalPausedDuration: 0,
});

const response = await request(app)
.post('/v1/streams/7/pause')
.set('Authorization', `Bearer ${token}`);
const response = await request(app)
.post('/v1/streams/7/pause')
.set('Authorization', `Bearer ${token}`);

expect(response.status).toBe(200);
expect(response.body).toMatchObject({
success: true,
streamId: 7,
txHash: 'pause-tx-hash',
});
expect(mockPauseStream).toHaveBeenCalledWith(sender.publicKey(), 7n);
expect(response.status).toBe(501);
expect(response.body).toMatchObject({
error: 'Not Implemented',
message:
'Pausing streams is not currently supported because the on-chain transaction is not yet submitted.',
});

expect(mockPauseStream).not.toHaveBeenCalled();
expect(mockPrisma.stream.update).not.toHaveBeenCalled();
});

it('rejects a raw signed transaction bearer token without a JWT', async () => {
const sender = makeKeypair();
const rawToken = buildSignedTransaction(sender, '00'.repeat(32));
Expand All @@ -144,41 +138,35 @@ describe('stream action routes', () => {
});
});

it('POST /v1/streams/:streamId/resume resumes a paused sender-owned stream', async () => {
const sender = makeKeypair();
const token = await getValidJwt(sender);

mockPrisma.stream.findUnique.mockResolvedValue({
streamId: 9,
sender: sender.publicKey(),
recipient: makeKeypair().publicKey(),
isActive: true,
isPaused: true,
pausedAt: Math.floor(Date.now() / 1000) - 30,
totalPausedDuration: 10,
});
mockResumeStream.mockResolvedValue({ txHash: 'resume-tx-hash' });
mockPrisma.stream.update.mockResolvedValue({
streamId: 9,
isActive: true,
isPaused: false,
pausedAt: null,
totalPausedDuration: 40,
});
it('POST /v1/streams/:streamId/resume returns 501 when resuming is not implemented', async () => {
const sender = makeKeypair();
const token = await getValidJwt(sender);

mockPrisma.stream.findUnique.mockResolvedValue({
streamId: 9,
sender: sender.publicKey(),
recipient: makeKeypair().publicKey(),
isActive: true,
isPaused: true,
pausedAt: Math.floor(Date.now() / 1000) - 30,
totalPausedDuration: 10,
});

const response = await request(app)
.post('/v1/streams/9/resume')
.set('Authorization', `Bearer ${token}`);
const response = await request(app)
.post('/v1/streams/9/resume')
.set('Authorization', `Bearer ${token}`);

expect(response.status).toBe(200);
expect(response.body).toMatchObject({
success: true,
streamId: 9,
txHash: 'resume-tx-hash',
});
expect(mockResumeStream).toHaveBeenCalledWith(sender.publicKey(), 9n);
expect(response.status).toBe(501);
expect(response.body).toMatchObject({
error: 'Not Implemented',
message:
'Resuming streams is not currently supported because the on-chain transaction is not yet submitted.',
});

expect(mockResumeStream).not.toHaveBeenCalled();
expect(mockPrisma.stream.update).not.toHaveBeenCalled();
});

it('POST /v1/streams/:streamId/withdraw withdraws the claimable amount for the recipient', async () => {
const recipient = makeKeypair();
const token = await getValidJwt(recipient);
Expand Down
Loading
Loading