fix: add missing workflow fucntionality for queues #4165
Conversation
PR Review: Add missing workflow functionality for queuesI've reviewed this PR which adds support for workflow queue message completion and improves workflow context access patterns. Here's my comprehensive analysis: ✅ Strengths
🔍 Code Quality Issues1. Typo in PR titleThe PR title has "fucntionality" instead of "functionality" 2. Indentation inconsistency (rivetkit-typescript/packages/rivetkit/src/workflow/context.ts:48-68)Lines 48-68 have inconsistent indentation. The 3. Test indentation (rivetkit-typescript/packages/workflow-engine/tests/messages.test.ts:985)Line 985 has an extra tab before 🏗️ Architecture & Design1. Message completion flow is well-designedThe implementation properly handles completion through multiple paths:
2. History replay handling (workflow-engine/src/context.ts:1156-1206)The 3. Multi-name listen supportThe addition of array support for
|
Merge activity
|
# Description Please include a summary of the changes and the related issue. Please also include relevant motivation and context. ## Type of change - [ ] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected) - [ ] This change requires a documentation update ## How Has This Been Tested? Please describe the tests that you ran to verify your changes. ## Checklist: - [ ] My code follows the style guidelines of this project - [ ] I have performed a self-review of my code - [ ] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [ ] My changes generate no new warnings - [ ] I have added tests that prove my fix is effective or that my feature works - [ ] New and existing unit tests pass locally with my changes
NathanFlurry
force-pushed
the
02-09-chore_frontend_add_default_vite_sentry_url
branch
from
f45f5d5 to
eb969c8
Compare
NathanFlurry
force-pushed
the
02-09-chore_add_missing_workflow_functionality
branch
from
33b1455 to
80f43be
Compare
![graphite-app[bot]](https://avatars.githubusercontent.com/in/158384?s=60&v=4){kind=small}
| async step<T>( | ||
| nameOrConfig: string | Parameters<WorkflowContextInterface["step"]>[0], | ||
| run?: () => Promise<T>, | ||
| ): Promise<T> { | ||
| if (typeof nameOrConfig === "string") { | ||
| if (!run) { | ||
| throw new Error("Step run function missing"); | ||
| } | ||
| return await this.#wrapActive(() => | ||
| this.#inner.step(nameOrConfig, () => | ||
| this.#withActorAccess(run), | ||
| ), | ||
| ); | ||
| } | ||
| return await this.#wrapActive(() => | ||
| this.#inner.step(nameOrConfig, () => | ||
| this.#withActorAccess(run), | ||
| ), | ||
| ); | ||
| } |
There was a problem hiding this comment.
The step method is incorrectly indented. Remove the extra indentation at the beginning of the method definition to align it with other class methods.
Spotted by Graphite Agent (based on CI logs)
Is this helpful? React 👍 or 👎 to let us know.
| import { Loop } from "@rivetkit/workflow-engine"; | ||
| import { actor } from "@/actor/mod"; | ||
| import { db } from "@/db/mod"; | ||
| import { WORKFLOW_GUARD_KV_KEY } from "@/workflow/constants"; | ||
| import { workflow, workflowQueueName } from "@/workflow/mod"; | ||
| import type { registry } from "./registry"; |
There was a problem hiding this comment.
The imports are not sorted according to Biome's rules. They should be sorted alphabetically. Run 'biome check --apply' to automatically fix the import sorting.
Spotted by Graphite Agent (based on CI logs)
Is this helpful? React 👍 or 👎 to let us know.
# Description Please include a summary of the changes and the related issue. Please also include relevant motivation and context. ## Type of change - [ ] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected) - [ ] This change requires a documentation update ## How Has This Been Tested? Please describe the tests that you ran to verify your changes. ## Checklist: - [ ] My code follows the style guidelines of this project - [ ] I have performed a self-review of my code - [ ] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [ ] My changes generate no new warnings - [ ] I have added tests that prove my fix is effective or that my feature works - [ ] New and existing unit tests pass locally with my changes
graphite-app
bot
deleted the
02-09-chore_add_missing_workflow_functionality
branch
| import type { Client } from "@/client/client"; | ||
| import type { Registry } from "@/registry"; | ||
| import type { AnyDatabaseProvider, InferDatabaseClient } from "@/actor/database"; |
There was a problem hiding this comment.
Imports are not sorted alphabetically. Biome linter typically requires imports to be sorted. Reorder these imports alphabetically.
Spotted by Graphite Agent (based on CI logs)
Is this helpful? React 👍 or 👎 to let us know.
| async step<T>( | ||
| nameOrConfig: string | Parameters<WorkflowContextInterface["step"]>[0], | ||
| run?: () => Promise<T>, | ||
| ): Promise<T> { | ||
| if (typeof nameOrConfig === "string") { | ||
| if (!run) { | ||
| throw new Error("Step run function missing"); | ||
| } | ||
| return await this.#wrapActive(() => | ||
| this.#inner.step(nameOrConfig, () => | ||
| this.#withActorAccess(run), | ||
| ), | ||
| ); | ||
| } | ||
| return await this.#wrapActive(() => | ||
| this.#inner.step(nameOrConfig, () => | ||
| this.#withActorAccess(run), | ||
| ), | ||
| ); | ||
| } |
There was a problem hiding this comment.
Indentation is inconsistent in the step method implementation. The return statement and its content are indented more than they should be. Fix the indentation to match the surrounding code structure.
Spotted by Graphite Agent (based on CI logs)
Is this helpful? React 👍 or 👎 to let us know.
| import { db } from "@/db/mod"; | ||
| import { WORKFLOW_GUARD_KV_KEY } from "@/workflow/constants"; | ||
| import { workflow, workflowQueueName } from "@/workflow/mod"; | ||
| import type { registry } from "./registry"; |
There was a problem hiding this comment.
Imports are not sorted alphabetically. Biome linter typically requires imports to be sorted. Reorder these imports alphabetically.
Spotted by Graphite Agent (based on CI logs)
Is this helpful? React 👍 or 👎 to let us know.
| @@ -68,7 +112,63 @@ for (const mode of modes) { | |||
| "wait-message", | |||
| "my-message", | |||
| ); | |||
| return message; | |||
| return message.body; | |||
| }; | |||
|
|
|||
| const result = await runWorkflow( | |||
| "wf-1", | |||
| workflow, | |||
| undefined, | |||
| driver, | |||
| { | |||
| mode, | |||
| }, | |||
| ).result; | |||
|
|
|||
| expect(result.state).toBe("completed"); | |||
| expect(result.output).toBe("hello"); | |||
| }); | |||
|
|
|||
| it("listen should return a completable message handle", async () => { | |||
| const completions: Array<{ id: string; response?: unknown }> = []; | |||
| const pending = [ | |||
| buildMessagePayload("my-message", "hello", "msg-1") as { | |||
| id: string; | |||
| name: string; | |||
| data: unknown; | |||
| sentAt: number; | |||
| complete?: (response?: unknown) => Promise<void>; | |||
| }, | |||
| ]; | |||
|
|
|||
| const messageDriver: WorkflowMessageDriver = { | |||
| async loadMessages() { | |||
| return pending.map((message) => ({ | |||
| ...message, | |||
| complete: async (response?: unknown) => { | |||
| completions.push({ id: message.id, response }); | |||
| }, | |||
| })); | |||
| }, | |||
| async addMessage(message) { | |||
| pending.push(message); | |||
| }, | |||
| async deleteMessages(messageIds) { | |||
| const deleted = new Set(messageIds); | |||
| const remaining = pending.filter( | |||
| (message) => !deleted.has(message.id), | |||
| ); | |||
| pending.length = 0; | |||
| pending.push(...remaining); | |||
| return messageIds; | |||
| }, | |||
| }; | |||
| driver.messageDriver = messageDriver; | |||
|
|
|||
| const workflow = async (ctx: WorkflowContextInterface) => { | |||
| const message = await ctx.listen<string>("wait-message", "my-message"); | |||
| await message.complete({ ok: true }); | |||
| return message.body; | |||
There was a problem hiding this comment.
Fix indentation and formatting in the new test cases for message handling. Ensure consistent spacing and indentation throughout.
Spotted by Graphite Agent (based on CI logs)
Is this helpful? React 👍 or 👎 to let us know.
1 participant
Description
Please include a summary of the changes and the related issue. Please also include relevant motivation and context.
Type of change
How Has This Been Tested?
Please describe the tests that you ran to verify your changes.
Checklist: