chore(gas): add support for find_workflows

[NathanFlurry]

No description provided.

[vercel]

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

Project	Deployment	Review	Updated (UTC)
rivetkit-serverless	Error		Jan 6, 2026 2:31am

3 Skipped Deployments

Project	Deployment	Review	Updated (UTC)
rivet-cloud	Ignored		Jan 6, 2026 2:31am
rivet-inspector	Ignored	Preview	Jan 6, 2026 2:31am
rivet-site	Ignored	Preview	Jan 6, 2026 2:31am

[NathanFlurry]

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

• feat(engine): return error for pending actors #3730 : 2 dependent PRs (#3675 , #3740 )
• chore(gas): add support for find_workflows #3739 👈 (View in Graphite)
• chore(docs): document abort signal #3673
• chore(rivetkit): clean up logging configuration #3672
• chore(rivetkit): RIVET_EXPOSE_ERRORS #3671
• chore(rivetkit): specialize context types for better compiler errors #3669
• fix(rivetkit): fix c.client not respecting correct origin #3668
• fix(rivetkit): allow providing no args to createRivetKit #3667
• 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: Add support for find_workflows

Summary

This PR adds a batched version of find_workflow called find_workflows that allows finding multiple workflows by (name, tags) pairs in a single database transaction. The implementation follows the existing pattern and adds the method to all context types.

---

Issues Found

🔴 Critical: Metrics API Inconsistency (line 1022)

The new find_workflows method uses a different metrics API than the existing find_workflow method:

• Old method (line 992-994): Uses .with_label_values(&[workflow_name]).observe(dt)
• New method (line 1022): Uses .record(dt, &[KeyValue::new("workflow_name", "batch")])

Impact: This inconsistency suggests either:

1. A migration is in progress between two metrics libraries
2. An unintentional mixing of APIs that will cause runtime issues

Recommendation: Use the same metrics API as the existing find_workflow method for consistency:

metrics::FIND_WORKFLOWS_DURATION
    .with_label_values(&["batch"])  // or perhaps "multiple"
    .observe(dt);

---

🟡 Performance: Concurrent vs Sequential Execution

The implementation runs all queries concurrently within a single transaction:

let futures = queries.iter().map(|(workflow_name, tags)| {
    self.find_workflow_inner(workflow_name, tags, &tx)
});
try_join_all(futures).await

Concern: Each find_workflow_inner call performs a streaming query that iterates through database entries (lines 378-410). Running these concurrently within the same transaction may:

• Not provide significant performance benefits if the database serializes access
• Potentially increase transaction contention
• Make it harder to reason about transaction lifecycle

Questions:

• Have you benchmarked this approach vs sequential execution?
• Does UniversalDB support concurrent operations within a single transaction effectively?

Alternative consideration: If performance is critical, consider documenting the expected use case and query count limits.

---

🟡 Code Quality: Missing Documentation

The doc comments are identical across all context types but lack important details:

/// Finds the first incomplete workflow for each (name, tags) pair in a single batch transaction.

Recommended additions:

/// Finds the first incomplete workflow for each (name, tags) pair in a single batch transaction.
///
/// Returns a vector of optional workflow IDs in the same order as the input queries.
/// `None` is returned for queries where no matching workflow was found.
///
/// # Arguments
/// * `queries` - Slice of (workflow_name, tags) pairs to search for
///
/// # Example
/// ```ignore
/// let queries = [
///     ("my-workflow", json\!({"env": "prod"})),
///     ("other-workflow", json\!({"region": "us-east"})),
/// ];
/// let results = ctx.find_workflows(&queries).await?;
/// ```

---

🟢 Minor: Return Value Documentation

While the return type Vec<Option<Id>> is clear from the signature, it's not immediately obvious that:

• The order matches the input order
• None indicates no workflow found (vs an error)

Consider adding a comment or example showing this behavior.

---

Positive Observations

✅ Consistent API design - The method is added to all context types (Activity, Operation, Standalone, Test)

✅ Proper instrumentation - Uses #[tracing::instrument(skip_all)] for observability

✅ Transaction safety - Correctly uses a single transaction for all queries, ensuring consistency

✅ Error handling - Properly propagates errors with context

✅ Code reuse - Leverages existing find_workflow_inner logic

---

Test Coverage

⚠️ No tests found for the new find_workflows method.

Recommendation: Add integration tests covering:

1. Multiple queries returning all Some values
2. Mixed results with some None values
3. Empty query array
4. Order preservation of results
5. Transaction rollback behavior on error

Example test location: engine/packages/gasoline/tests/workflows/find_workflows_test.rs

---

Security & Performance

✅ No obvious security concerns
✅ No SQL injection risks (using parameterized queries)
✅ Proper use of database transaction for ACID guarantees

---

Recommendations Summary

1. Fix metrics API inconsistency (critical)
2. Add comprehensive documentation with examples
3. Add integration tests for the new functionality
4. Consider documenting performance characteristics and intended use cases
5. Verify concurrent query execution is optimal for UniversalDB

---

Conclusion

The implementation is solid and follows existing patterns well. The main concern is the metrics API inconsistency which should be resolved before merging. Adding tests and better documentation would significantly improve the PR quality.

[MasterPtato]

Nit: should be its own metric

[graphite-app]

Merge activity

• Jan 6, 2:19 AM UTC: NathanFlurry added this pull request to the Graphite merge queue.
• Jan 6, 2:20 AM UTC: The Graphite merge queue removed this pull request due to downstack failures on PR #3667.
• Jan 6, 2:20 AM UTC: The Graphite merge queue removed this pull request due to downstack failures on PR #3667.
• Jan 6, 2:21 AM UTC: NathanFlurry added this pull request to the Graphite merge queue.
• Jan 6, 2:22 AM UTC: The Graphite merge queue removed this pull request due to downstack failures on PR #3667.
• Jan 6, 2:22 AM UTC: The Graphite merge queue removed this pull request due to downstack failures on PR #3667.
• Jan 6, 2:31 AM UTC: NathanFlurry added this pull request to the Graphite merge queue.
• Jan 6, 2:31 AM UTC: CI is running for this pull request on a draft pull request (#3754) due to your merge queue CI optimization settings.
• Jan 6, 2:32 AM UTC: Merged by the Graphite merge queue via draft PR: #3754.