Support maxParallelism of 0 to enable pausing workpools #145
Conversation
📝 WalkthroughWalkthroughAdded a public Workpool.kick(ctx) method and a component-level Changes
sequenceDiagram
rect rgba(200,230,255,0.4)
participant Client as Workpool (Client)
participant Runner as RunMutationCtx
participant Component as Kick Mutation Handler
participant Globals as Workpool Globals / kickMainLoop
end
Client->>Runner: kick(ctx) calls ctx.runMutation("kick", {config})
Runner->>Component: invoke kick mutation handler with config
Component->>Component: validateConfig(config)
alt config valid
Component->>Globals: kickMainLoop(ctx, "kick", config)
Globals-->>Component: completes loop (maybe updates globals)
Component-->>Runner: return null
Runner-->>Client: resolved Promise<void>
else invalid
Component-->>Runner: throw validation error
Runner-->>Client: rejects with error
end
🎯 3 (Moderate) | ⏱️ ~20 minutes
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
📜 Recent review detailsConfiguration used: defaults Review profile: CHILL Plan: Pro ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (2)
🧰 Additional context used🧬 Code graph analysis (2)src/component/lib.ts (3)
src/client/index.ts (3)
🔇 Additional comments (4)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
is this my coderabbit or yours... |
|
Tip For best results, initiate chat on the files or code changes. @danthegoodman1 I can't view images directly, but I'm here to help! Could you describe what you're seeing in the screenshot or let me know what question you have? If you're asking about whether this is your CodeRabbit instance or a different one - I'm CodeRabbit responding on this PR. If you have questions about the review, configuration, or anything else related to this PR, I'm happy to help! |
commit: |
|
One large issue: that value override only lasts for the duration of the isolate 😭 Without changing the behavior of "always use the code's options", the only way to achieve a permanent change in behavior is to look up a current value every time from the DB with a query/helper function. That would definitely have tons of serialization aborts when ever a value is changed too. |
|
Marked draft as there's discussion around making the DB the source of truth rather than the code. |
|
Following up here - when you specify the value via env var with a runtime-initialized workpool, this works, correct? |
|
I hadn’t tested using an env var in the ‘new Workpool’ (sry mobile) since we spoke but I’ll test that shortly |
|
@ianmacartney it does not appear to. I set: Then triggered a self-invoking action: Then |
|
actually I'm stupid and dumb and didn't use my version of it XDDDDDDDDDDDDD one sec |
|
Note: you need to do another enqueue for the value to get passed in. So if there aren't any enqueues you'd need to set the env var and change it in the db |
|
Ideally there'd be a |
|
Yeah sorry that example wasn't 100% accurate, I hav ea helper function like: |
|
Ok so it seems like you can set it to 0 initially, and it will stay paused, but once you turn it up to a positive number, then try to reduce it to 0 again, it will not stop in my observations. I also am seeing a maxParallelism 1 in the DB so something it's setting it. |
I need to go to bed... or maybe opus needs to review me XD |
|
Alright so I added a public |
|
@ianmacartney is it possible to update that above PR npm install to the current commit? |
|
done |
|
That seems to have done the trick, now you can |
|
Testing proceedure: /**
* Name MUST be less than 28 characters (convex only allows 40 char long env vars)
*/
function getMaxParallelism(name: string, defaultValue: number): number {
try {
const envValue = process.env[`WRKPL_${name}_MAX_P`]
if (envValue) {
const parsed = parseInt(envValue, 10)
if (!isNaN(parsed) && parsed >= 0) {
return parsed
}
}
} catch {
// Env vars unavailable during schema evaluation
}
return defaultValue
}
export const testWorkpool = new Workpool(components.testWorkpool, {
retryActionsByDefault: true,
defaultRetryBehavior: {
maxAttempts: 100,
base: 1,
initialBackoffMs: 5000,
},
maxParallelism: getMaxParallelism("TEST", 1),
})export const testAction = internalAction({
args: {
delay: v.number(),
},
handler: async (ctx, args) => {
console.log(`Test action with delay: ${args.delay} ${process.env.WRKPL_TEST_MAX_P}`)
if (args.delay === 0) {
console.log("Not enqueueing with 0 delay, kicking", process.env.WRKPL_TEST_MAX_P)
await testWorkpool.kick(ctx)
return Ok()
}
console.log("Enqueuing action")
await testWorkpool.enqueueAction(
ctx,
internal.test_action.testAction,
{
delay: args.delay,
},
{
runAfter: args.delay,
}
)
return Ok()
},
})
|
|
I do still think something that doesn't require env vars would be better, but this does "work", and it lays the groundwork for supporting 0 parallelism anyway |
|
@ianmacartney this is ready for a final review when you have time! (sorry I don't have permissions to request review) |
2 participants
Support maxParallelism of 0 to enable pausing workpools. It appears like this mostly was supported other than just some validation and falsy checks (otherwise it handles 0 fine).
All tests passing.
Note: I did use Opus 4.5 to investigate what was needed to support maxParallelism of 0, but I investigated it and wrote the code myself.
For devs:
The best way to use this is to just set it:
The next time anything runs, it will ensure it's synced with your code and pull in the latest value (including an enqueue).
Closes #133
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.
🫡
Summary by CodeRabbit
Release Notes
New Features
kickmethod to workpool for triggering execution.Bug Fixes
✏️ Tip: You can customize this high-level summary in your review settings.