perf: add events.createBatch() for batch event creation#641
perf: add events.createBatch() for batch event creation#641pranaygp wants to merge 1 commit intographite-base/641from
Conversation
🦋 Changeset detectedLatest commit: 7ad380a The changes in this PR will be included in the next version bump. This PR includes changesets to release 17 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
🧪 E2E Test ResultsNo test result files found. ❌ Some E2E test jobs failed:
Check the workflow run for details. |
|
Warning This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
This stack of pull requests is managed by Graphite. Learn more about stacking. |
pranaygp
changed the base branch from
pranaygp/perf-phase-3b-atomic-events
to
graphite-base/641
pranaygp
force-pushed
the
pranaygp/perf-batch-events
branch
from
853963d to
752bc6b
Compare
pranaygp
changed the base branch from
graphite-base/641
to
pranaygp/perf-phase-3b-atomic-events
pranaygp
changed the base branch from
pranaygp/perf-phase-3b-atomic-events
to
graphite-base/641
pranaygp
force-pushed
the
pranaygp/perf-batch-events
branch
from
f15434b to
32f21aa
Compare
pranaygp
force-pushed
the
graphite-base/641
branch
from
5090eed to
faecbea
Compare
pranaygp
changed the base branch from
graphite-base/641
to
pranaygp/perf-phase-3b-atomic-events
![vercel[bot]](https://avatars.githubusercontent.com/in/8329?s=60&v=4){kind=small}
| // Update run status for run_completed events | ||
| if (runCompletedEvents.length > 0) { | ||
| const completedData = (runCompletedEvents[0] as any).eventData as { | ||
| output?: any; | ||
| }; | ||
| await drizzle | ||
| .update(Schema.runs) | ||
| .set({ | ||
| status: 'completed', | ||
| output: completedData.output as SerializedContent | undefined, | ||
| completedAt: now, | ||
| updatedAt: now, | ||
| }) | ||
| .where(eq(Schema.runs.runId, effectiveRunId)); | ||
| } | ||
|
|
||
| // Update run status for run_failed events | ||
| if (runFailedEvents.length > 0) { | ||
| const failedData = (runFailedEvents[0] as any).eventData as { | ||
| error: any; | ||
| errorCode?: string; | ||
| }; | ||
| const errorMessage = | ||
| typeof failedData.error === 'string' | ||
| ? failedData.error | ||
| : (failedData.error?.message ?? 'Unknown error'); | ||
| // Store structured error as JSON for deserializeRunError to parse | ||
| const errorJson = JSON.stringify({ | ||
| message: errorMessage, | ||
| stack: failedData.error?.stack, | ||
| code: failedData.errorCode, | ||
| }); | ||
| await drizzle | ||
| .update(Schema.runs) | ||
| .set({ | ||
| status: 'failed', | ||
| error: errorJson, | ||
| completedAt: now, | ||
| updatedAt: now, | ||
| }) | ||
| .where(eq(Schema.runs.runId, effectiveRunId)); | ||
| } | ||
|
|
||
| // Update run status for run_cancelled events | ||
| if (runCancelledEvents.length > 0) { | ||
| await drizzle | ||
| .update(Schema.runs) | ||
| .set({ | ||
| status: 'cancelled', | ||
| completedAt: now, | ||
| updatedAt: now, | ||
| }) | ||
| .where(eq(Schema.runs.runId, effectiveRunId)); | ||
| } | ||
|
|
||
| // Update run status for run_paused events | ||
| if (runPausedEvents.length > 0) { | ||
| await drizzle | ||
| .update(Schema.runs) | ||
| .set({ | ||
| status: 'paused', | ||
| updatedAt: now, | ||
| }) | ||
| .where(eq(Schema.runs.runId, effectiveRunId)); | ||
| } | ||
|
|
||
| // Update run status for run_resumed events | ||
| if (runResumedEvents.length > 0) { | ||
| await drizzle | ||
| .update(Schema.runs) | ||
| .set({ | ||
| status: 'running', | ||
| updatedAt: now, | ||
| }) | ||
| .where(eq(Schema.runs.runId, effectiveRunId)); | ||
| } |
There was a problem hiding this comment.
The createBatch method is missing the hook cleanup logic that the create method performs when processing terminal run events (run_completed, run_failed, run_cancelled). This causes inconsistent behavior and prevents hook tokens from being reused.
View Details
📝 Patch Details
diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts
index 4d56f53..2075c10 100644
--- a/packages/world-postgres/src/storage.ts
+++ b/packages/world-postgres/src/storage.ts
@@ -777,6 +777,10 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
updatedAt: now,
})
.where(eq(Schema.runs.runId, effectiveRunId));
+ // Delete all hooks for this run to allow token reuse
+ await drizzle
+ .delete(Schema.hooks)
+ .where(eq(Schema.hooks.runId, effectiveRunId));
}
// Update run status for run_failed events
@@ -804,6 +808,10 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
updatedAt: now,
})
.where(eq(Schema.runs.runId, effectiveRunId));
+ // Delete all hooks for this run to allow token reuse
+ await drizzle
+ .delete(Schema.hooks)
+ .where(eq(Schema.hooks.runId, effectiveRunId));
}
// Update run status for run_cancelled events
@@ -816,6 +824,10 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
updatedAt: now,
})
.where(eq(Schema.runs.runId, effectiveRunId));
+ // Delete all hooks for this run to allow token reuse
+ await drizzle
+ .delete(Schema.hooks)
+ .where(eq(Schema.hooks.runId, effectiveRunId));
}
// Update run status for run_paused events
Analysis
Missing hook cleanup in createBatch() for terminal run events
What fails: The createBatch() method in packages/world-postgres/src/storage.ts fails to delete hooks when processing terminal run events (run_completed, run_failed, run_cancelled), unlike the create() method which properly cleans up hooks. This causes hook tokens to never be released for reuse, leading to potential token exhaustion in systems processing many workflow runs.
How to reproduce:
- Call
createBatch()with events including ahook_createdevent - Process a batch with a terminal event (run_completed, run_failed, or run_cancelled)
- Check the database hooks table - hooks will still exist for that run
- Attempt to create and complete another workflow run with the same hook token
- Observe that the old hooks are not cleaned up, preventing token reuse
Result: The hooks table retains entries for completed runs indefinitely. Comparing to the create() method behavior at lines 416-419, 451-454, and 471-474, these hooks should be deleted after status updates, but are not in createBatch().
Expected: Both create() and createBatch() should consistently delete hooks when terminal run events occur (lines 766-841 in createBatch should include hook deletion after each terminal status update, matching the pattern in create()).
Root cause: Missing drizzle.delete(Schema.hooks).where(eq(Schema.hooks.runId, effectiveRunId)) statements after the runCompletedEvents, runFailedEvents, and runCancelledEvents status update blocks in createBatch().
Implementation note: Added hook deletion logic to match the exact pattern used in the create() method, ensuring consistent behavior across both APIs and allowing hook tokens to be reused for future workflow runs.
| const runData = (eventData as any).eventData as { | ||
| deploymentId: string; | ||
| workflowName: string; | ||
| input: any[]; |
There was a problem hiding this comment.
Missing hook cleanup logic in createBatch() method for terminal run events (run_completed, run_failed, run_cancelled)
View Details
📝 Patch Details
diff --git a/packages/world-postgres/src/storage.ts b/packages/world-postgres/src/storage.ts
index e904510..96a9370 100644
--- a/packages/world-postgres/src/storage.ts
+++ b/packages/world-postgres/src/storage.ts
@@ -843,6 +843,10 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
updatedAt: now,
})
.where(eq(Schema.runs.runId, effectiveRunId));
+ // Delete all hooks for this run to allow token reuse
+ await drizzle
+ .delete(Schema.hooks)
+ .where(eq(Schema.hooks.runId, effectiveRunId));
}
// Update run status for run_failed events
@@ -870,6 +874,10 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
updatedAt: now,
})
.where(eq(Schema.runs.runId, effectiveRunId));
+ // Delete all hooks for this run to allow token reuse
+ await drizzle
+ .delete(Schema.hooks)
+ .where(eq(Schema.hooks.runId, effectiveRunId));
}
// Update run status for run_cancelled events
@@ -882,6 +890,10 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
updatedAt: now,
})
.where(eq(Schema.runs.runId, effectiveRunId));
+ // Delete all hooks for this run to allow token reuse
+ await drizzle
+ .delete(Schema.hooks)
+ .where(eq(Schema.hooks.runId, effectiveRunId));
}
// Update run status for run_paused events
Analysis
Issue Description
The create() method in the createEventsStorage function properly cleans up hooks when handling terminal run events to allow hook token reuse. However, the createBatch() method was missing this cleanup logic for all three terminal event types.
Root Cause
In the create() method (lines ~428-502), each terminal event handler includes hook cleanup:
run_completed(line ~446): Deletes hooks viadrizzle.delete(Schema.hooks).where(...)run_failed(line ~479): Deletes hooks viadrizzle.delete(Schema.hooks).where(...)run_cancelled(line ~502): Deletes hooks viadrizzle.delete(Schema.hooks).where(...)
In the createBatch() method (lines 838-883), the corresponding event handlers update the run status but do not delete hooks:
runCompletedEvents(line 838-846): Missing hook cleanuprunFailedEvents(line 849-872): Missing hook cleanuprunCancelledEvents(line 875-883): Missing hook cleanup
This inconsistency causes hook tokens to never be released for reuse in batch operations, while they are properly released in single event operations.
Fix Applied
Added hook cleanup logic to all three terminal event handlers in createBatch():
-
runCompletedEvents (line ~838): Added:
// Delete all hooks for this run to allow token reuse await drizzle .delete(Schema.hooks) .where(eq(Schema.hooks.runId, effectiveRunId));
-
runFailedEvents (line ~849): Added:
// Delete all hooks for this run to allow token reuse await drizzle .delete(Schema.hooks) .where(eq(Schema.hooks.runId, effectiveRunId));
-
runCancelledEvents (line ~875): Added:
// Delete all hooks for this run to allow token reuse await drizzle .delete(Schema.hooks) .where(eq(Schema.hooks.runId, effectiveRunId));
These additions ensure consistency between the create() and createBatch() methods, allowing hook tokens to be properly reused regardless of whether events are created individually or in batches.
Impact
- Prevents hook token accumulation in the database
- Enables proper token reuse across multiple workflow runs
- Ensures consistent behavior between single and batch event operations
- Maintains data integrity by cleaning up resources when runs reach terminal states
| }; | ||
| } | ||
|
|
||
| export async function createWorkflowRunEventBatch( |
There was a problem hiding this comment.
Missing CreateEventRequest type import in packages/world-vercel/src/events.ts
View Details
📝 Patch Details
diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts
index 137f5af..9ac7cbe 100644
--- a/packages/world-vercel/src/events.ts
+++ b/packages/world-vercel/src/events.ts
@@ -1,6 +1,7 @@
import {
type AnyEventRequest,
type CreateEventParams,
+ type CreateEventRequest,
type Event,
type EventResult,
EventSchema,
Analysis
The function createWorkflowRunEventBatch at line 198 of packages/world-vercel/src/events.ts uses CreateEventRequest[] as a parameter type, but the CreateEventRequest type was not imported from the @workflow/world package. This would cause a TypeScript compilation error: "Cannot find name 'CreateEventRequest'".
The fix adds type CreateEventRequest to the import statement from @workflow/world, which is where the type is exported (packages/world/src/events.ts:270). The CreateEventRequest type is defined as Exclude<AnyEventRequest, RunCreatedEventRequest> and is used to represent event requests that can be created after a workflow run has already been initiated.
This is a straightforward type import fix that resolves the compilation error without changing any logic or functionality.
🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1 participant
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.5 noreply@anthropic.com