fix(gas): fix batch listen, fix history for graceful signal send in workflows

[MasterPtato]

No description provided.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Preview	Comments	Updated (UTC)
rivet-site	Ready	Preview	Comment	Nov 18, 2025 2:33am
rivetkit-serverless	Ready	Preview	Comment	Nov 18, 2025 2:33am

2 Skipped Deployments

Project	Deployment	Preview	Comments	Updated (UTC)
rivet-cloud	Ignored			Nov 18, 2025 2:33am
rivet-inspector	Ignored	Preview		Nov 18, 2025 2:33am

[MasterPtato]

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.
Learn more

• fix(pb): rewrite runner wf to handle batch signals #3480
• fix(gas): fix batch listen, fix history for graceful signal send in workflows #3479 👈 (View in Graphite)
• feat(gas): add batch listen #3478
• chore(gas): add test for history with holes #3477
• chore: change streamingmode to exact for ranges with limits #3476
• fix(gas): optimize loops #3475
• chore: rename retry -> hibernate in runner protocol #3473
• fix(udb): update backoff algo #3469
• chore: write/update docs #3468
• chore: add debug scripts #3467
• main

---

How to use the Graphite Merge Queue

Add the label merge-queue to this PR to add it to the merge queue.

You must have a Graphite account in order to use the merge queue. Sign up using this link.

An organization admin has enabled the Graphite Merge Queue in this repository.

Please do not merge from GitHub as this will restart CI on PRs being processed by the merge queue.

This stack of pull requests is managed by Graphite. Learn more about stacking.

[claude]

PR Review: Fix batch listen, fix history for graceful signal send in workflows

Summary

This PR introduces two main improvements to the gasoline workflow engine:

1. Graceful signal sending - adds graceful_not_found() method to signal builders that returns Ok(None) instead of erroring when the target workflow is not found
2. Batch listen fix - corrects signal batching to properly sort by timestamp and handle indexed input chunks

Code Quality & Best Practices

Strengths:

• Consistent API design across both common/signal.rs and workflow/signal.rs builders
• Good documentation with doc comments explaining behavior (lines 107-108, 118-119 in both files)
• Follows existing error handling patterns using Result<Option<Id>>
• Proper tracing/logging throughout with structured logging as per CLAUDE.md guidelines
• Loop optimization improvements maintain good performance characteristics

Suggestions:

1. Type annotation consistency (gasoline/src/db/kv/mod.rs:1945):

   let mut signals = futures_util::stream::iter(owned_filter.clone())

   The mut is added here, but signals were already being collected into a Vec. This is correct now for the sort operation on line 1980, but consider adding a comment explaining why this needs to be mutable.

2. Error message clarity (api-peer/src/actors/delete.rs:74-77):
   The warning message could be more specific about which operation failed:

   tracing::warn!(
       actor_id=?path.actor_id,
       "actor workflow not found when sending destroy signal, likely already stopped"
   );

Potential Bugs & Issues

Critical:

1. Database index insertion logic (gasoline/src/db/kv/debug.rs:931-939 and gasoline/src/db/kv/mod.rs:1505-1512):

   if let Some(input_chunks) = current_event.indexed_input_chunks.get_mut(key.index) {
       input_chunks.push(entry);
   } else {
       current_event.indexed_input_chunks.insert(key.index, vec![entry]);
   }

   This is a good fix! The previous code had a logic error where it would only push to existing indices but never create new ones. This could have caused data loss in signal batching. The fix properly handles both cases.

2. Timestamp consistency (gasoline/src/db/kv/mod.rs:1980-2012):
   Good catch fixing the timestamp! The code now:

   • Sorts signals by key.ts (line 1980)
   • Uses a single now timestamp for all acks (line 1982, 2011)
   • Applies limit AFTER sorting but BEFORE processing (line 1995)

   This ensures consistent ordering and prevents race conditions.

Minor:

3. Event type mismatch fix (gasoline/src/db/kv/keys/history.rs:1645):

   EventType::Signal -> EventType::Signals

   Good catch! This was incorrectly using the singular form for batch operations.

Performance Considerations

Improvements:

1. Loop event upsertion parallelization (gasoline/src/ctx/workflow.rs:872-978):
   The new approach defers loop event upserts to run in parallel with the next iteration. This is a clever optimization that should reduce latency in tight loops. The 4-way join on lines 910-940 is well-documented with comments explaining the parallel execution strategy.

2. Signal batching optimization (gasoline/src/db/kv/mod.rs:1995):
   Applying .take(limit) after sorting but before processing is more efficient than the previous approach of sorting after collection.

Concerns:

1. The buffer_unordered(1024) on line 2063 could cause memory pressure if processing 1024 signals simultaneously. Consider if this limit should be configurable or tied to the input limit parameter.

Security Concerns

No security issues identified. The changes:

• Properly handle missing workflows without exposing sensitive information
• Maintain transactional integrity with conflict ranges
• Don't introduce new attack surfaces

Test Coverage

Missing test scenarios:

1. Test that graceful_not_found() properly creates and replays removed events in workflow history
2. Test signal batching with timestamps in non-chronological order to verify sorting
3. Test the indexed input chunks fix with multiple signals in a batch
4. Test loop event upsertion parallelization with rapid iterations

Recommended test additions:

// Test graceful signal send with replay
#[test]
async fn test_graceful_signal_not_found_replay() {
    // First run: target doesn't exist, should create removed event
    // Second run: should replay removed event correctly
}

// Test batch signal ordering
#[test]
async fn test_batch_signal_timestamp_ordering() {
    // Create signals with out-of-order timestamps
    // Verify they're received in timestamp order
}

Additional Notes

1. Display text changes (engine/src/util/wf/mod.rs): The changes from "signal receive" to "signal recv" for consistency are good, but ensure this doesn't break any external tooling that parses these strings.

2. Logging level change (universaldb/src/driver/rocksdb/transaction_conflict_tracker.rs:67): Changing from info! to debug! for conflict logging is appropriate as this could be noisy in production.

3. TypeScript comment (runner-protocol/src/index.ts:56-58): The TODO comment about renaming createTs to update_ts should be addressed in a future PR to maintain naming consistency per CLAUDE.md conventions.

Verdict

LGTM with minor suggestions

This is a solid improvement to the workflow engine. The graceful signal sending feature is well-implemented with proper history tracking for replay consistency. The batch listen fixes address real bugs in signal ordering and indexed chunk handling. The loop optimization should provide measurable performance improvements.

The code follows project conventions, has good documentation, and maintains backward compatibility. Recommended to merge after addressing the test coverage gaps.

---

Recommendation: Approve with suggestion to add tests before merge.

[pkg-pr-new]

More templates

• example-ai-agent
• example-background-jobs
• example-better-auth-external-db
• example-bots
• chat-room
• example-cloudflare-workers
• example-cloudflare-workers-hono
• example-counter
• example-counter-serverless
• example-crdt
• example-cursors
• example-cursors-raw-websocket
• example-database
• example-deno
• example-sqlite
• example-elysia
• example-express
• example-freestyle
• example-game
• example-hono
• example-hono-bun
• example-hono-react
• example-kitchen-sink
• example-next-js
• example-rate
• example-raw-fetch-handler
• example-raw-websocket-handler
• example-raw-websocket-handler-proxy
• example-react
• starter
• example-stream
• example-sync
• example-tenant
• example-trpc
• example-workflows

@rivetkit/cloudflare-workers

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/cloudflare-workers@3479

@rivetkit/db

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/db@3479

@rivetkit/framework-base

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/framework-base@3479

@rivetkit/next-js

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/next-js@3479

@rivetkit/react

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/react@3479

rivetkit

pnpm add https://pkg.pr.new/rivet-dev/rivet/rivetkit@3479

@rivetkit/sql-loader

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/sql-loader@3479

@rivetkit/engine-runner

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/engine-runner@3479

@rivetkit/engine-runner-protocol

pnpm add https://pkg.pr.new/rivet-dev/rivet/@rivetkit/engine-runner-protocol@3479

commit: 0ca2fa2