fix(webapp): dedupe realtimeStreams array push on stream create (#3653)

## Summary

The PUT handler at `/realtime/v1/streams/:runId/:target/:streamId` ran
`taskRun.update({ realtimeStreams: { push: streamId } })` on every call,
even when the `streamId` was already present. SDK call patterns that
re-initialize the same stream key on every chunk produce a per-write row
UPDATE, duplicate entries pile up in the array, and the row-lock + TOAST
rewrite cost grows unbounded on long-running stream sessions.

## Fix

Mirror the sibling append handler: read the array first and only push
when the `streamId` isn't already present. Identical behavior for
first-time stream creation; repeat creates short-circuit to a single
indexed read. The dashboard's per-run stream listing keeps working
because the first create still records the entry.

## Test plan

- [ ] A fresh PUT for a new `(run, streamId)` adds the entry to the
array
- [ ] A repeat PUT for the same pair leaves the array unchanged
- [ ] 404 is returned when the run doesn't exist; 400 when the run is
completed
This commit is contained in:
Eric Allam
2026-05-18 10:19:20 +01:00
committed by GitHub
parent 9623e88b05
commit f88d4018cc
2 changed files with 25 additions and 9 deletions
@@ -0,0 +1,6 @@
---
area: webapp
type: fix
---
Dedupe the `realtimeStreams` array push on `PUT /realtime/v1/streams/:runId/:target/:streamId` so repeat stream-init calls for the same `(run, streamId)` skip the row UPDATE, mirroring the existing append handler.
@@ -62,31 +62,41 @@ const { action } = createActionApiRoute(
if (request.method === "PUT") {
// This is the "create" endpoint
const updatedRun = await prisma.taskRun.update({
const target = await prisma.taskRun.findFirst({
where: {
friendlyId: targetId,
runtimeEnvironmentId: authentication.environment.id,
},
data: {
realtimeStreams: {
push: params.streamId,
},
},
select: {
id: true,
realtimeStreams: true,
realtimeStreamsVersion: true,
completedAt: true,
},
});
if (updatedRun.completedAt) {
if (!target) {
return new Response("Run not found", { status: 404 });
}
if (target.completedAt) {
return new Response("Cannot initialize a realtime stream on a completed run", {
status: 400,
});
}
if (!target.realtimeStreams.includes(params.streamId)) {
await prisma.taskRun.update({
where: { id: target.id },
data: {
realtimeStreams: { push: params.streamId },
},
});
}
const realtimeStream = getRealtimeStreamInstance(
authentication.environment,
updatedRun.realtimeStreamsVersion,
target.realtimeStreamsVersion,
basinContext
);
@@ -94,7 +104,7 @@ const { action } = createActionApiRoute(
return json(
{
version: updatedRun.realtimeStreamsVersion,
version: target.realtimeStreamsVersion,
},
{ status: 202, headers: responseHeaders }
);