fix(pb): implement events and commands for actor wfs

[MasterPtato]

No description provided.

[vercel]

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

Project	Deployment	Preview	Comments	Updated (UTC)
rivetkit-serverless	Ready	Preview	Comment	Dec 2, 2025 7:12pm

3 Skipped Deployments

Project	Deployment	Preview	Comments	Updated (UTC)
rivet-cloud	Ignored			Dec 2, 2025 7:12pm
rivet-inspector	Ignored	Preview		Dec 2, 2025 7:12pm
rivet-site	Ignored	Preview		Dec 2, 2025 7:12pm

[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

• feat(pegboard): move pegboard serverless to wf #3416 : 2 dependent PRs (#3450 , #3571 )
• fix(runtime): evict runners on SIGTERM, add periodic shutdown progress messages #3550
• fix(pb): implement events and commands for actor wfs #3543 👈 (View in Graphite)
• fix: make all uses of protocol handle v4 <-> v3 #3528
• fix(pb): simplify runner wf #3483
• fix(gas): fix loop forgotten bug due to concurrency #3482
• fix(pb): rewrite runner wf to handle batch signals #3480
• fix(gas): fix batch listen, fix history for graceful signal send in workflows #3479
• 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
• fix: parallelize publish signal awaits #3523
• chore: fix diff between ee, fix plural api query params #3524
• 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: Implement events and commands for actor workflows

I've reviewed this PR which implements MK2 protocol support for actor workflows. Here are my findings:

Strengths

1. Well-structured protocol versioning with clear upgrade/downgrade paths
2. Robust event demuxing with batching and proper cleanup
3. Careful initialization sequence that handles missed commands
4. Good per-runner state management with separate event/command indices

Issues & Concerns

1. Channel abort TODO (actor_event_demuxer.rs:81) - HIGH PRIORITY
The TODO about aborting tasks needs addressing. Aborting may lose in-flight events. Consider graceful shutdown or document why abort is safe.

2. Metadata parsing (conn.rs:285-288, 299-303) - MEDIUM PRIORITY
Silent failures via unwrap_or_default() may hide malformed metadata. Add logging for parse failures.

3. Event deduplication (workflows/actor/mod.rs:466-477)
Consider edge cases: What if events arrive out of order? Risk of loss if runner crashes mid-send?

4. Runner state cleanup (workflows/actor/mod.rs:623-625)
Ensure no race condition where events from removed runner recreate state entry.

5. Missing command ack tracking
Event acks are batched but no similar tracking for command acks. Intentional?

Security

• Actor ownership validation: Good ✓
• Generation checking: Proper filtering ✓

Performance

• Good event batching with 1024 buffer
• Proper transaction isolation levels
• Consider: Loop processes up to 1024 signals - may need periodic yielding

Test Coverage

Consider adding:

• Event deduplication tests
• Missed command recovery tests
• Multi-runner generation change tests

Recommendations

1. HIGH: Address abort TODO or document safety
2. MEDIUM: Add metadata parse failure logging
3. MEDIUM: Add reconnection integration tests

Overall

Solid implementation with good architecture. Main concern is the task abort TODO.

Recommendation: Approve with minor changes

[pkg-pr-new]

More templates

• example-ai-agent
• chat-room
• example-cloudflare-workers
• example-cloudflare-workers-hono
• example-cloudflare-workers-inline-client
• example-counter
• example-counter-next-js
• example-counter-serverless
• example-cursors
• example-cursors-raw-websocket
• example-deno
• example-sqlite
• example-elysia
• example-express
• example-freestyle
• example-hono
• example-hono-bun
• example-hono-react
• example-kitchen-sink
• example-next-js
• quickstart-actions
• quickstart-cross-actor-actions
• quickstart-multi-region
• quickstart-raw-websockets
• quickstart-realtime
• quickstart-scheduling
• quickstart-state
• example-raw-fetch-handler
• example-raw-websocket-handler
• example-raw-websocket-handler-proxy
• example-react
• starter
• example-stream
• example-trpc

@rivetkit/cloudflare-workers

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

@rivetkit/db

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

@rivetkit/framework-base

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

@rivetkit/next-js

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

@rivetkit/react

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

rivetkit

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

@rivetkit/sql-loader

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

@rivetkit/engine-runner

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

@rivetkit/engine-runner-protocol

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

commit: 0f54d2f

[claude]

Pull Request Review: fix(pb): implement events and commands for actor wfs

Overview

This PR implements events and commands handling for actor workflows in the Pegboard system, specifically adding support for the mk2 protocol version. The changes span across multiple packages including pegboard-runner, pegboard, pegboard-gateway, and related infrastructure.

Code Quality & Architecture

Strengths:

1. Good abstraction for protocol versions via the Init enum in conn.rs (lines 203-255)
2. Event batching properly reduces overhead using EVENT_ACK_BATCH_SIZE = 250
3. Proper state management with RunnerState tracking for multiple runners
4. Database transaction isolation with custom instrumentation

Potential Bugs & Issues

1. Missing error handling in metadata parsing (conn.rs:285-288, 299-303): Using unwrap_or_default() silently swallows parsing errors. Consider logging when metadata parsing fails.

2. Unchecked index arithmetic (actor/mod.rs:468-469): The filtering assumes monotonically increasing indices. If the runner sends duplicate or out-of-order events, this could cause issues.

3. Potential race condition (conn.rs:88-117): Between checking if a runner exists/is expired and dispatching the workflow, another connection could modify the runner state. Consider documenting this.

4. Missing protocol version validation (pegboard-gateway/lib.rs:164-174): The code defaults to PROTOCOL_MK1_VERSION when no version is stored, but this could cause issues if a runner reconnects with a different version.

[claude]

Performance Considerations

1. Database read amplification (conn.rs:324-356): Reading all missed commands during init could be expensive for runners offline for long periods. Consider pagination, command expiration, and metrics.

2. Iterator chaining efficiency (keys/runner.rs:804-806): Nested iterators could be optimized using flat_map.

3. Event filtering on every signal (actor/mod.rs:464-473): For high-frequency events, consider maintaining a cursor or checkpointing more aggressively.

Security Concerns

1. No metadata size validation (conn.rs:297-316): A malicious runner could send excessive metadata. Add size validation before chunking.

2. Command replay vulnerability (conn.rs:323-356): Missed commands are replayed without verifying relevance. Consider adding command expiration.

Test Coverage

Missing test scenarios:

• No tests for handle_init with prepopulated actor names
• Missing event deduplication logic tests
• No metadata chunking/splitting tests
• Missing protocol version negotiation edge case tests

Database Schema Concerns

Command persistence: Commands are persisted but there is no cleanup logic. This could lead to unbounded growth over time. Add a cleanup mechanism for acknowledged/expired commands.

[claude]

Style & Convention Issues

1. Import ordering (conn.rs:7-8): Should combine duplicate futures_util imports per CLAUDE.md conventions
2. Formatting inconsistencies: Auto-formatting changes should be in a separate commit

Recommendations

High Priority:

1. Add size validation for metadata to prevent abuse
2. Implement command cleanup/expiration logic
3. Add logging for metadata parsing errors
4. Add tests for the new event handling logic

Medium Priority:

1. Document the protocol version fallback behavior
2. Add metrics for missed command counts and event batch sizes
3. Optimize iterator chaining in metadata key combining
4. Consider pagination for missed commands

Low Priority:

1. Separate formatting changes into their own commit
2. Combine duplicate imports
3. Document the EVENT_ACK_BATCH_SIZE constant

Overall Assessment

This is a solid implementation that handles complex protocol versioning and event management. The code is generally well-structured with proper error handling. However, there are areas that need attention around data validation, performance at scale, and test coverage. The main concerns are unbounded growth (metadata size, command storage) and potential edge cases in event deduplication.

Recommendation: Request changes to address high-priority issues, particularly around data validation and resource management.

[claude]

PR Review: Implement Events and Commands for Actor Workflows

This PR implements a significant refactoring of the actor lifecycle management system to use persistent events and commands for mk2 protocol runners. Here's my comprehensive review:

✅ Strengths

1. Protocol Versioning: Good backward compatibility with mk1 protocol while introducing mk2 improvements

   • Clean separation between mk1 and mk2 handling in conn.rs:181-198
   • Proper fallback to PROTOCOL_MK1_VERSION when version is missing

2. Event Batching: Efficient event acknowledgment with batching (lines actor/mod.rs:586-606)

   • Uses EVENT_ACK_BATCH_SIZE = 250 to reduce message overhead
   • Only sends acks when threshold is exceeded

3. State Management: Well-structured state tracking per runner

   • RunnerState tracks both last_event_idx and last_event_ack_idx
   • LifecycleRunnerState maintains proper event sequencing

4. Actor Event Demuxer: Clean separation of concerns

   • Batches events efficiently (actor_event_demuxer.rs:52)
   • Includes proper garbage collection (actor_event_demuxer.rs:76-89)

⚠️ Issues & Concerns

Critical

1. Missing Error Handling in Connection Init (conn.rs:70)
   If init parsing fails, the connection is dropped without sending a proper error response to the runner. Consider sending an error message before closing.

2. Silent Event Send Failures (actor_event_demuxer.rs:41, 63)
   Events are silently dropped if the channel is full or closed with underscore pattern. This could lead to event loss without any warning. Consider:

   • Logging when sends fail
   • Implementing backpressure or flow control
   • Alerting when events are dropped

3. Unsafe Abort in GC (actor_event_demuxer.rs:84)
   There's a TODO comment indicating uncertainty about task abortion safety. Aborting tasks can leave state inconsistent. Consider:

   • Sending a shutdown signal to the task instead
   • Waiting for graceful completion with a timeout
   • Documenting why abort is safe if it is

4. Race Condition in Protocol Version (pegboard-gateway/lib.rs:165-176)
   The protocol version is read from UDB but uses read_opt with a fallback. If a runner reconnects between mk1 and mk2, there's a small window where the version might be stale or missing.

Medium Priority

5. Transaction Isolation Inconsistency (runtime.rs:244)
   Mixing Snapshot and Serializable reads in the same transaction could lead to subtle consistency issues. The comment explains the intent but the safety should be verified.

6. Unbounded Command Index (runtime.rs:1105)
   The command index grows unboundedly (last_command_idx += input.commands.len()). Consider:

   • What happens when it overflows i64::MAX?
   • Implementing command cleanup/archival

7. Metadata Deserialization (conn.rs:285-288, 300-303)
   Invalid JSON in metadata silently becomes empty with unwrap_or_default(). Consider logging when deserialization fails for debugging.

8. Missing Context in Trace (actor_event_demuxer.rs:38)
   Good logging added, but the ingest method isn't instrumented. Consider adding tracing::instrument for better traceability.

Low Priority

9. Code Formatting: The PR includes unrelated formatting changes in api-builder/middleware.rs:189. This is minor but makes the diff harder to review.

10. Commented-Out Code: Comment formatting change in metrics/buckets.rs:5 seems unnecessary.

📊 Performance Considerations

1. Event Batching: Good use of batching in actor_event_demuxer.rs:52 with buffer size 1024
2. Database Queries: The StreamingMode::WantAll in conn.rs:326 could be expensive if there are many missed commands. Consider pagination.
3. Command Storage: Commands are stored indefinitely in UDB. Consider implementing cleanup based on acknowledgment.

🔒 Security Considerations

1. Metadata Injection: The prepopulate_actor_names and metadata fields allow runners to write arbitrary data to UDB. Ensure proper validation exists upstream.
2. Command Replay: Missed commands are replayed on reconnect (conn.rs:377-385). Verify that commands are idempotent to prevent double-execution issues.

🧪 Testing Recommendations

1. Add tests for event deduplication logic (actor/mod.rs:466-470)
2. Test command index overflow scenarios
3. Test event loss scenarios (what happens if demuxer drops events?)
4. Test mk1 ↔ mk2 protocol migration
5. Test runner reconnection with missed commands

📝 Documentation Gaps

1. The new ActorCommandKey structure needs documentation explaining the indexing scheme
2. The relationship between last_event_idx, last_event_ack_idx, and last_command_idx should be documented
3. Add comments explaining why certain transactions use Snapshot vs Serializable

✨ Suggestions

1. Extract Constants: Magic numbers like 1024 for batch sizes should be constants
2. Metrics: Add metrics for:
   • Events dropped by demuxer
   • Command replay counts
   • Event processing latency
3. Graceful Degradation: Consider what happens when UDB is slow/unavailable during init

Overall Assessment

This is a substantial and well-architected change that modernizes the actor lifecycle system. The core logic is sound, but there are several areas where error handling and edge cases need attention. The critical issues around silent event dropping and task abortion should be addressed before merging.

Recommendation: Request changes to address critical issues, then approve after fixes.

Files needing attention:

• engine/packages/pegboard-runner/src/actor_event_demuxer.rs
• engine/packages/pegboard-runner/src/conn.rs
• engine/packages/pegboard/src/workflows/actor/runtime.rs

[claude]

PR Review: fix(pb): implement events and commands for actor wfs

I've completed a comprehensive review of this PR. This is a significant refactor implementing mk2 protocol support for actor workflows. Overall, the implementation is solid, but I've identified several areas that need attention.

Critical Issues

1. Missing Error Propagation (pegboard-gateway/src/lib.rs:210-213)
The result from send().await? is now being checked, but there's no error handling if the workflow isn't found. Add proper error handling or document why graceful_not_found() is acceptable.

2. Unsafe unwrap_or_default() in Protocol Version (pegboard-gateway/src/lib.rs:172-173)
Silently defaults to MK1 protocol if version isn't found. Log when defaulting to MK1 or consider failing for critical operations.

3. Race Condition in Init Handler (pegboard-runner/src/conn.rs:257-272)
Signal is sent before database writes complete. Workflow may not see prepopulated actor names or metadata. Complete DB writes before sending signal or add synchronization.

4. Silent JSON Parse Failures (pegboard-runner/src/conn.rs:285-288, 299-303)
JSON parsing errors are silently ignored with empty defaults. Add logging for parse failures.

High Priority Issues

5. TODO Comments Need Resolution
Multiple TODOs in pegboard-runner/src/ws_to_tunnel_task.rs - KV operations may be blocking the WebSocket receive loop. Create follow-up issues or resolve in this PR.

6. Missing Tracing (pegboard/src/workflows/actor/runtime.rs:1096-1163)
insert_and_send_commands activity needs structured logging for debugging.

7. HACK Comment (pegboard/src/workflows/actor/runtime.rs:573-574)
Using util::timestamp::now() for create_ts breaks workflow determinism. Use ctx.create_ts() instead.

Medium Priority Issues

8. Unbounded Channels (pegboard-runner/src/actor_event_demuxer.rs:29)
Could cause memory issues. Consider bounded channels with backpressure.

9. Hardcoded EVENT_ACK_BATCH_SIZE
Should this be configurable per deployment?

10. Incomplete Runner State Cleanup
State could accumulate if check_runners fails. Add periodic cleanup or document retention policy.

Positive Observations

• Excellent protocol abstraction with Init enum
• Robust error context throughout
• Proper database subspace management
• Good structured logging (with room for improvement)
• Careful transaction design with isolation levels

Test Coverage

Missing tests for:

• Event demuxer
• Command insertion and acknowledgment
• Protocol version fallback

Summary

Overall: Approve with Changes

Priority actions before merge:

1. Fix init handler race condition
2. Add error logging for JSON parse failures
3. Resolve TODO comments or create follow-up issues
4. Fix HACK with dynamic timestamps

Follow-up issues to create:

• KV operation performance optimization
• Test coverage for event/command system
• Error message sanitization

Great work on this complex refactor!

[claude]

Code Review: Events and Commands Implementation for Actor Workflows

I've reviewed this PR which implements a significant architectural improvement to how actor workflows handle events and commands. Here's my detailed feedback:

---

🎯 Overall Assessment

This is a well-structured PR that introduces important improvements to the actor workflow system, particularly around:

• Persistent command storage in the database
• Event acknowledgment batching for better performance
• Protocol version handling and backward compatibility
• Better separation between mk1 and mk2 protocol handling

Overall Quality: Good ✅

---

💡 Strengths

1. Robust Database Design (engine/packages/pegboard/src/keys/runner.rs)

   • Well-designed key structures for actor commands with proper indexing
   • Good use of chunking for metadata storage (prevents hitting size limits)
   • Proper use of formal keys with serialization/deserialization

2. Event Batching (engine/packages/pegboard/src/workflows/actor/mod.rs:587-607)

   • Efficient event acknowledgment batching (EVENT_ACK_BATCH_SIZE = 250)
   • Only sends acks when threshold is exceeded, reducing network overhead

3. Backward Compatibility

   • Proper fallback to PROTOCOL_MK1_VERSION when protocol version is not found
   • Both mk1 and mk2 paths maintained throughout the codebase
   • Good separation of concerns between protocol versions

4. Error Handling

   • Comprehensive use of context() for error propagation
   • Graceful degradation (e.g., .graceful_not_found() calls)
   • Minimal use of .unwrap() - only .unwrap_or_default() which is safe

5. Concurrency Safety (engine/packages/pegboard-runner/src/actor_event_demuxer.rs)

   • Good use of per-actor channels to avoid contention
   • Proper cleanup with GC of inactive channels
   • Graceful shutdown handling

---

⚠️ Issues & Recommendations

Critical Issues

1. JSON Deserialization with Silent Failures (conn.rs:285-288, 299-303)

let metadata = serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(&data.metadata)
    .unwrap_or_default();

Issue: If JSON parsing fails, the error is silently ignored and defaults to an empty map. This could mask data corruption or protocol issues.

Recommendation:

let metadata = match serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(&data.metadata) {
    Ok(m) => m,
    Err(err) => {
        tracing::warn!(?err, metadata=?data.metadata, "failed to parse actor metadata, using empty map");
        Default::default()
    }
};

2. Race Condition Risk in Event Processing (actor/mod.rs:459-471)

let last_event_idx = state.runner_states.entry(runner_id).or_default().last_event_idx;

let new_events = sig.events
    .iter()
    .filter(|event| {
        event.checkpoint.index > last_event_idx &&
        crate::utils::event_generation(&event.inner) == generation
    });

Issue: Events are filtered by last_event_idx, but if the same runner sends multiple batches in quick succession and the workflow processes them out of order, events could be dropped or duplicated.

Recommendation: Consider adding sequence validation or logging when events arrive out of order to detect this scenario early.

---

Medium Priority Issues

3. Missing Transaction Conflict Handling

In runtime.rs:243-248, the code uses Snapshot isolation to avoid conflicts when checking the queue:

// NOTE: This is not Serializable because we don't want to conflict with other
// inserts/clears to this range
Snapshot,

Concern: While the comment explains the intent, using Snapshot here means the transaction could miss concurrent inserts. This appears intentional, but there's a window where an actor could be added to the queue between the check and the runner allocation scan.

Recommendation: Add a test case to verify this behavior under concurrent load, or add additional validation after the transaction commits.

4. Potential Memory Leak in Event Demuxer (actor_event_demuxer.rs:76-89)

if self.last_gc.elapsed() > GC_INTERVAL {
    self.last_gc = Instant::now();
    
    self.channels.retain(|_, channel| {
        let keep = channel.last_seen.elapsed() < MAX_LAST_SEEN;
        
        if !keep {
            // TODO: Verify aborting is safe here
            channel.handle.abort();
        }
        
        keep
    });
}

Issues:

• Channel last_seen is never updated in the current implementation, so channels are only removed after 30s of no activity
• The TODO comment indicates uncertainty about aborting being safe
• If events continue flowing to an actor, its channel will never be cleaned up

Recommendation:

1. Update last_seen when ingesting events: channel.last_seen = Instant::now(); after sending
2. Consider a more graceful shutdown than .abort(), or document why abort is safe
3. Add metrics to monitor channel count growth

5. Missing Error Context in Handle Init (conn.rs:262-272)

The signal is sent before database operations, which is good for fail-fast, but errors from the database transaction don't include context about which part failed.

Recommendation: Add .context() or wrap sections of the transaction with span instrumentation to help debugging.

---

Minor Issues & Code Quality

6. Inconsistent Logging Levels (conn.rs:288, 303)

Metadata parsing failures should probably be logged as warnings since they represent a protocol/data issue.

7. Code Duplication in spawn_actor (runtime.rs:556-820)

The mk1 vs mk2 command sending logic is duplicated in three places. Consider extracting to a helper function.

8. Magic Numbers

• EVENT_ACK_BATCH_SIZE = 250 - Consider making this configurable or document the rationale
• GC_INTERVAL = 30s, MAX_LAST_SEEN = 30s - Consider making these configurable

9. HACK Comments (runtime.rs:573-575, 597-599)

// HACK: We should not use dynamic timestamp here, but we don't validate if signal data
// changes (like activity inputs) so this is fine for now.
create_ts: util::timestamp::now(),

Recommendation: File a follow-up issue to address this properly (workflow determinism).

---

🔒 Security Considerations

1. Actor Ownership Validation (ws_to_tunnel_task.rs:183-213)

✅ Good validation that actors belong to the correct runner before processing KV requests. This prevents cross-runner data access.

2. TODO: Error Message Exposure (ws_to_tunnel_task.rs:237, 284, etc.)

Multiple TODOs note: // TODO: Don't return actual error?

Recommendation: Sanitize error messages before sending to runners to avoid leaking internal details. This should be addressed before production use.

---

🧪 Testing

Missing Test Coverage:

1. Event ordering edge cases (out-of-order delivery, duplicates)
2. Concurrent actor allocation with queue insertion
3. Event demuxer GC behavior
4. Command acknowledgment with missed commands replay
5. Protocol version migration scenarios (mk1→mk2 upgrades)

Recommendation: Add integration tests for the event/command flow, especially around the persistence and replay logic.

---

📊 Performance Considerations

Positive:

• Event batching reduces signal overhead significantly
• Actor event demuxer prevents head-of-line blocking
• Proper use of database transactions with appropriate isolation levels

Concerns:

• missed_commands query in handle_init scans entire command subspace - could be slow for runners with many actors. Consider pagination or a limit.
• KV operations are synchronous in the WS handler - the TODO mentions adding a queue, which would be beneficial under load

---

📝 Documentation

Good:

• Comments explaining transaction isolation choices
• Workflow step documentation in actor creation sequence
• Clear naming of keys and structures

Could Improve:

• Document the event/command flow lifecycle
• Add module-level documentation for the event demuxer
• Document retry/replay semantics for commands

---

✅ Adherence to Repository Guidelines

Per CLAUDE.md:

• ✅ Uses custom error types with RivetError derive
• ✅ Proper structured logging (tracing with ?field syntax)
• ✅ Lowercase log messages
• ✅ UUID for IDs, i64 timestamps in milliseconds
• ✅ Timestamp naming follows *_at convention (though some use *_ts)
• ✅ Workspace dependencies used correctly
• ⚠️ A few instances don't follow the "imports at top" guideline (inline imports acceptable but check consistency)

---

🎬 Conclusion

This is a solid PR that implements a critical piece of infrastructure. The code quality is generally high with good error handling and thoughtful design.

Blocking Issues: None
High Priority: Address JSON parsing silent failures, verify event ordering guarantees
Medium Priority: Event demuxer improvements, test coverage
Low Priority: Code deduplication, configuration for magic numbers

Recommendation: Approve with minor changes requested ✅

The critical path code is sound, but I'd like to see the JSON parsing improved and some consideration of the event ordering edge cases before merge. The other issues can be addressed in follow-up PRs.

---

Great work on this complex refactoring! The architecture is well-thought-out and the implementation is thorough. 🚀

[claude]

Code Review - PR #3543: fix(pb): implement events and commands for actor wfs

Overview

This PR implements a significant architectural improvement to the Pegboard actor system by introducing proper event and command handling for actor workflows, particularly for the mk2 runner protocol. The changes refactor how actors communicate with runners, moving from signal-based communication to a more robust event-driven model with persistent command storage.

Summary of Changes

Core Changes:

• Actor Workflow (pegboard/src/workflows/actor/mod.rs): Added support for handling batched events from mk2 runners, including event deduplication and acknowledgment batching
• Actor Runtime (pegboard/src/workflows/actor/runtime.rs): New activity InsertAndSendCommands for persisting commands to UniversalDB before sending them to runners
• Pegboard Runner Connection (pegboard-runner/src/conn.rs): Refactored initialization to handle mk2 protocol with prepopulated actor names, metadata storage, and missed command recovery
• Actor Event Demuxer (pegboard-runner/src/actor_event_demuxer.rs): Enhanced to dispatch events with runner_id context for proper routing
• Protocol Changes: Added new protocol structures for command/event wrappers with checkpoints for reliable delivery

---

Positive Aspects ✅

1. Robust Event Tracking: The implementation of per-runner event indices with batch acknowledgment (lines 580-606 in actor/mod.rs) is well-designed and prevents duplicate processing
2. Backward Compatibility: Excellent handling of both mk1 and mk2 protocols throughout the codebase
3. Persistent Commands: Storing commands in UniversalDB before sending ensures they can be recovered after runner reconnections
4. Clean Abstraction: The Init enum pattern (lines 203-255 in conn.rs) provides good protocol version abstraction

---

Issues & Concerns ⚠️

1. Critical: Missing Error Handling in Guard 🔴

Location: engine/packages/guard/src/routing/pegboard_gateway.rs:213

.await?;

if res.is_none() {

The ? operator was added but the error is not propagated properly. If the signal send fails, this will cause a panic rather than graceful error handling.

Recommendation: Review error handling strategy here - should this fail silently, log and continue, or propagate the error?

---

2. Data Integrity: Event Generation Filtering Logic 🟡

Location: engine/packages/pegboard/src/workflows/actor/mod.rs:462-469

let new_events = sig.events
    .iter()
    .filter(|event| {
        event.checkpoint.index > last_event_idx &&
        crate::utils::event_generation(&event.inner) == generation
    });

Issue: Events are filtered by generation but there's no logging when events are dropped due to generation mismatch. This could make debugging difficult.

Recommendation: Add tracing when events are filtered out:

let new_events = sig.events
    .iter()
    .filter(|event| {
        let matches = event.checkpoint.index > last_event_idx &&
            crate::utils::event_generation(&event.inner) == generation;
        if !matches {
            tracing::debug!(
                event_idx=?event.checkpoint.index,
                event_gen=?crate::utils::event_generation(&event.inner),
                current_gen=?generation,
                "filtering out event"
            );
        }
        matches
    });

---

3. Performance: Unbounded Command Storage 🟡

Location: engine/packages/pegboard/src/workflows/actor/runtime.rs:1109-1136

Issue: Commands are written to UniversalDB indefinitely without cleanup. Over time, this could lead to significant storage bloat, especially for long-running actors with many state transitions.

Recommendation: Implement a cleanup mechanism:

• Add a background task to delete acknowledged commands
• Or store only the last N commands per actor
• Document the expected lifecycle of these command records

---

4. Concurrency: Race Condition in Demuxer 🟡

Location: engine/packages/pegboard-runner/src/actor_event_demuxer.rs:48-65

let _ = tx.send(event);

self.channels.insert(
    actor_id,
    Channel {
        tx,
        handle,
        last_seen: Instant::now(),
    },
);

Issue: The first event is sent before the channel is inserted into the map. If another event arrives for the same actor immediately after, it could create a duplicate channel since the first one isn't in the map yet.

Recommendation: Insert the channel first, then send the event:

self.channels.insert(
    actor_id,
    Channel {
        tx: tx.clone(),
        handle,
        last_seen: Instant::now(),
    },
);

let _ = tx.send(event);

---

5. Error Handling: Silent Failures in Metadata Parsing 🟡

Location: engine/packages/pegboard-runner/src/conn.rs:285-288 and 299-303

let metadata = serde_json::from_str::<
    serde_json::Map<String, serde_json::Value>,
>(&data.metadata)
.unwrap_or_default();

Issue: Invalid JSON in metadata fields silently falls back to empty objects. This could hide configuration errors.

Recommendation: Log when parsing fails:

let metadata = serde_json::from_str::<
    serde_json::Map<String, serde_json::Value>,
>(&data.metadata)
.map_err(|e| {
    tracing::warn!(?e, name=?name, "failed to parse actor metadata, using empty object");
    e
})
.unwrap_or_default();

---

6. Logging: Inconsistent Log Levels 🟢

Location: Various files

Issue: Some important state transitions use debug! when they should probably use info!:

• Line 298 in actor/mod.rs: tracing::debug!(actor_id=?input.actor_id, "actor wake");
• Line 282 in actor/mod.rs: tracing::warn!(actor_id=?input.actor_id, "actor lost");

Recommendation: Use info! for significant lifecycle events (wake, allocate, deallocate) and reserve debug! for verbose internal state.

---

7. Code Quality: Formatting Inconsistencies 🟢

Location: Multiple files (e.g., api-builder/src/middleware.rs:189)

Some lines exceed reasonable length limits. While this PR seems to fix some formatting issues, there are still instances of inconsistent formatting.

Note: Per CLAUDE.md, do not run cargo fmt automatically, but be mindful of readability.

---

Security Considerations 🔒

1. ✅ Good: Commands are validated through the protocol's deserialization
2. ✅ Good: Event indices prevent replay attacks
3. ⚠️ Consider: The prepopulated actor names feature could potentially be abused if not properly validated upstream

---

Testing Recommendations 🧪

1. Event Ordering: Add tests to verify events are processed in the correct order even when delivered out of order
2. Reconnection Scenarios: Test that missed commands are properly recovered after runner reconnection
3. Concurrent Actor Events: Test the demuxer with rapid concurrent events for the same actor
4. Command Persistence: Verify commands survive workflow restarts
5. Generation Transitions: Test edge cases around generation increments

---

Performance Considerations ⚡

1. Event Batching: The batch size of 250 (EVENT_ACK_BATCH_SIZE) seems reasonable, but should be tuned based on production metrics
2. Database Writes: Each InsertAndSendCommands call writes multiple UDB entries. Monitor transaction latency
3. Memory: The runner_states HashMap could grow unbounded - the cleanup logic in CheckRunners is good but ensure it's called frequently enough

---

Documentation Needs 📝

1. Add comments explaining the lifecycle of command records in UDB
2. Document the event acknowledgment protocol and batch size rationale
3. Explain the generation matching logic and why events from old generations are dropped
4. Add architecture diagram showing the event/command flow between actors, workflows, and runners

---

Summary

This is a solid architectural improvement that adds much-needed reliability to the actor system. The core design is sound, with good separation of concerns and backward compatibility.

Before Merging:

• ✅ Fix the error handling in guard (Issue [SVC-2555] Set up issue templates #1)
• 🔍 Consider adding cleanup for command storage (Issue [SVC-2504] Fix 5 GB upload limit for local development from Cloudflare #3)
• 🔍 Review the demuxer race condition (Issue [SVC-2483] Remove hardcoded uses of rivet.gg #4)

Post-Merge:

• Add the recommended logging for event filtering
• Improve metadata parsing error visibility
• Add comprehensive integration tests for the new event/command system

Overall Assessment: Approve with minor requested changes ✅

---

Review generated by Claude Code

[claude]

PR Review: fix(pb): implement events and commands for actor wfs

Summary

This PR implements a comprehensive overhaul of the Pegboard actor workflow system, migrating from mk1 to mk2 protocol with improved event/command handling, state management, and protocol version negotiation. The changes are substantial (+1934/-909 lines across 42 files) and introduce important architectural improvements.

🟢 Strengths

Architecture & Design

• Clean protocol migration: The mk1/mk2 protocol split is well-handled with proper version detection and fallback to mk1 for legacy runners
• Event demuxing: The ActorEventDemuxer (pegboard-runner/src/actor_event_demuxer.rs:19-111) provides efficient batching and per-actor task isolation
• Missed commands recovery: Excellent handling of commands that occurred while runner was disconnected (pegboard-runner/src/conn.rs:323-357)
• Runner state tracking: Per-runner state management with proper cleanup (pegboard/src/workflows/actor/mod.rs:66, 107-118)

Code Quality

• Proper error handling: Uses custom error types consistently with the RivetError derive pattern
• Structured logging: Good use of tracing with structured fields (e.g., ?runner_id, ?actor_id)
• Transaction safety: UDB transactions are well-structured with proper conflict range management

🟡 Issues Requiring Attention

Critical Issues

1. Missing event send in actor_event_demuxer (pegboard-runner/src/actor_event_demuxer.rs:37-73)

   • Line 63 sends the event: let _ = tx.send(event);
   • However, line 40-41 only checks existing channels but doesn't update last_seen
   • Impact: Channels will be GC'd even when actively receiving events
   • Fix: Update last_seen when sending to existing channels:

   if let Some(channel) = self.channels.get_mut(&actor_id) {
       channel.last_seen = Instant::now();  // Add this
       let _ = channel.tx.send(event);
   }

2. Protocol version default handling (pegboard-gateway/src/lib.rs:161-174, 305-318)

   • Falls back to PROTOCOL_MK1_VERSION when protocol version key doesn't exist
   • This is correct for backward compatibility, but consider adding a metric or warning log when falling back
   • Recommendation: Add observability:

   let protocol_version = protocol_version_entry.unwrap_or_else(|| {
       tracing::debug\!(runner_id=?runner_id, "protocol version not found, defaulting to mk1");
       PROTOCOL_MK1_VERSION
   });

3. Potential race condition in init_conn (pegboard-runner/src/conn.rs:88-121)

   • Runner expiry check happens before workflow dispatch
   • If runner expires between the check and dispatch, there could be duplicate runner IDs
   • Current mitigation: The unique workflow dispatch should prevent this, but worth verifying
   • Recommendation: Add a comment explaining the race safety

Medium Priority Issues

4. Unused import (pegboard-runner/src/conn.rs:8)

   • TryStreamExt is imported but appears unused based on the diff
   • Fix: Remove if not needed or verify usage

5. Error handling in event dispatch (pegboard-runner/src/actor_event_demuxer.rs:56-59)

   • Errors cause the entire demuxer task to abort for that actor
   • Consider retry logic or more graceful degradation
   • Recommendation: Log error and continue processing other events

6. Memory concerns with command storage (pegboard/src/workflows/actor/runtime.rs:1096-1163)

   • Commands are stored in UDB indefinitely (until explicitly cleared)
   • No TTL or cleanup mechanism visible
   • Question: When/how are old commands cleaned up? Consider adding a comment about lifecycle

7. Hardcoded batch size (pegboard/src/workflows/actor/mod.rs:18)

   • EVENT_ACK_BATCH_SIZE = 250 is hardcoded
   • Consider making this configurable via config for tuning

8. Missing null safety (pegboard/src/workflows/actor/mod.rs:449-462)

   • Line 461: state.runner_states.entry(runner_id).or_default().last_event_idx
   • This is fine, but the filtering logic at 465-470 could skip all events if generations don't match
   • Recommendation: Add a debug log when all events are filtered out

Code Style & Best Practices

9. Formatting inconsistencies (multiple files)

   • Some files have formatting changes (e.g., api-builder/src/middleware.rs:189-196)
   • Per CLAUDE.md, avoid running cargo fmt manually
   • Action: These will be handled by pre-commit hooks

10. Missing documentation (pegboard/src/keys/runner.rs:779-862)

    • New MetadataKey implementation lacks documentation
    • Recommendation: Add doc comments explaining chunking strategy and usage

🔒 Security Considerations

• No security issues identified: Proper use of UDB transactions with conflict ranges
• Good isolation: Actor events are properly isolated per workflow
• Proper validation: Input validation in handle_init with unwrap_or_default for JSON parsing

⚡ Performance Considerations

Positive

• Event batching: The demuxer batches up to 1024 events (actor_event_demuxer.rs:52)
• Efficient lookups: Proper use of HashMap for O(1) channel lookups
• GC mechanism: Periodic cleanup prevents unbounded memory growth

Concerns

• Unbounded command history: Commands stored in UDB without apparent cleanup
• No backpressure: Unbounded channel in demuxer could cause memory issues under heavy load
• Consider: Adding channel capacity limits or flow control

🧪 Test Coverage

• Limited visibility: No new test files added in this PR
• Existing tests modified: guard-core/tests/simple_websocket.rs (formatting only)
• Recommendation: Add integration tests for:
  • mk1 → mk2 protocol negotiation
  • Missed command recovery on reconnect
  • Event ordering and deduplication
  • Runner state cleanup after expiry

📋 Additional Recommendations

1. Add migration guide: Document the mk1→mk2 protocol migration for operators
2. Monitoring: Add metrics for:
   • Event processing latency per actor
   • Command queue depth per runner
   • Protocol version distribution
3. Consider feature flag: Gradual rollout of mk2 protocol with ability to force mk1 for debugging
4. Runner metadata validation: Add size limits to prevent abuse (pegboard-runner/src/conn.rs:297-316)

✅ Approval Recommendation

Approve with minor changes required - The PR is architecturally sound and implements a critical improvement to the actor workflow system. The critical issue (#1) should be fixed before merge, and the recommendations should be considered for follow-up work.

Must Fix Before Merge

• Fix last_seen update in ActorEventDemuxer (Issue [SVC-2555] Set up issue templates #1)
• Remove unused import TryStreamExt (Issue [SVC-2483] Remove hardcoded uses of rivet.gg #4)

Should Consider for Follow-up

• Add observability for protocol version fallback (Issue [SVC-2479] Send cluster events to PostHog #2)
• Document command cleanup lifecycle (Issue Remove servers not in salt #6)
• Add integration tests for protocol negotiation
• Add metrics for event processing

---

Great work on this complex migration! The event/command architecture is much cleaner with mk2. 🚀

[claude]

PR Review: fix(pb): implement events and commands for actor wfs

Overview

This PR implements a significant refactoring of the Pegboard actor workflow system, moving from a simple event forwarding model to a more sophisticated event and command management system for mk2 protocol runners. The changes span 43 files with ~1,900 additions and ~900 deletions.

Code Quality

Strengths:

• Well-structured refactoring with clear separation of concerns
• Proper error handling using the custom error system
• Good use of structured logging with tracing
• Follows repository conventions (hard tabs, lowercase log messages, structured field logging)

Areas for Improvement:

1. Import Organization (engine/packages/pegboard-runner/src/conn.rs:7-8)

   • Duplicate futures_util imports should be consolidated into a single line

2. Error Handling (engine/packages/pegboard-runner/src/conn.rs:285-288)

   • Using unwrap_or_default() when parsing JSON metadata silently swallows errors. Consider logging warnings when deserialization fails.

3. Missing Cleanup (engine/packages/engine/src/util/db.rs)

   • The wf_sqlite_shell function was removed but left an unused import for Path

Potential Bugs

1. Race Condition Risk (engine/packages/pegboard-runner/src/actor_event_demuxer.rs:52-66)

   • The event is sent after spawning the handler task, but before the channel is stored in self.channels. If another event for the same actor arrives before insertion completes, a duplicate task could be spawned. Consider sending the event after insertion.

2. Index Tracking (engine/packages/pegboard/src/workflows/actor/mod.rs:461-472, 580-607)

   • The last_event_idx filtering logic is complex and critical. The code filters events where event.checkpoint.index > last_event_idx, but what happens if events arrive out of order? Consider adding validation or documenting assumptions about event ordering.

3. Missing Runner State Fields

   • In the actor workflow, RunnerState is defined with last_command_idx but the new code references last_event_idx and last_event_ack_idx. Need to verify RunnerState has these fields added.

Performance Considerations

1. Batch Event Processing (engine/packages/pegboard-runner/src/actor_event_demuxer.rs:47-54)

   • Good use of batching with recv_many. Consider making the buffer size (16) configurable.

2. Event Ack Batching (engine/packages/pegboard/src/workflows/actor/mod.rs:587-606)

   • Excellent optimization batching event acknowledgments by 250 events. This will significantly reduce signal overhead.

3. Database Transaction Scope (engine/packages/pegboard-runner/src/conn.rs:274-360)

   • The handle_init function performs a large database transaction. Consider breaking this into smaller transactions if timeouts become an issue, especially for runners with many prepopulated actors.

4. HashMap Usage (engine/packages/pegboard/src/workflows/actor/mod.rs:66)

   • runner_states grows unbounded over actor lifetime. The cleanup at lines 999-1001 helps, but consider periodic garbage collection if actors are long-lived.

Security Concerns

1. Metadata Injection (engine/packages/pegboard-runner/src/conn.rs:285-288, 300-302)

   • Metadata from init packets is parsed from client-provided JSON without validation
   • No size limits on metadata or prepopulated actor names
   • Could potentially store malicious/large payloads in database
   • Recommendation: Add size limits and validation before storing

2. Input Validation (engine/packages/pegboard-runner/src/conn.rs:282-294)

   • The prepopulated actor names come from untrusted client input. Ensure actor name validation (valid DNS subdomain per CLAUDE.md conventions) is applied.

Test Coverage

Missing Test Scenarios:

1. Event deduplication when same event arrives multiple times
2. Out-of-order event handling
3. Command/event synchronization during runner reconnection
4. Metadata size limits
5. Race conditions in ActorEventDemuxer
6. Protocol version migration (mk1 to mk2)

Architecture & Design

Positive:

• Clean separation between mk1 and mk2 protocol handling via Init enum
• Event demultiplexing architecture allows parallel processing per actor
• Command persistence enables reliability across runner reconnections

Concerns:

1. Complexity Growth - The actor workflow (mod.rs) is now over 1000 lines. Consider extracting event handling logic into separate modules.
2. State Management - Multiple sources of truth for actor state (workflow state, UDB keys, runner state). Document the consistency model.
3. Protocol Version Fallback (engine/packages/pegboard-gateway/src/lib.rs:166-174) - The code assumes PROTOCOL_MK1_VERSION when protocol version is missing. This could mask issues.

Specific Recommendations

1. Add tracing instrumentation for the database transaction in handle_init
2. Document the event ordering guarantees - critical for understanding filtering logic
3. Add metrics for event batch sizes, command queue depths per runner, and event deduplication rates
4. Consider using a BTreeMap instead of HashMap for runner_states if you need stable iteration order
5. Add integration tests for the full mk2 protocol flow

Conclusion

This is a substantial and well-architected change that adds important reliability features. The code quality is generally good and follows repository conventions. However, there are concerns around race conditions in event demuxing, security validation for user-provided metadata, and complexity of the actor workflow.

Recommendation: Address the race condition and security concerns before merging. The other issues can be tracked as follow-up improvements.

Priority Issues to Address:

1. High: Fix race condition in ActorEventDemuxer (line 52-66)
2. High: Add validation/size limits for metadata and prepopulated actor names
3. Medium: Document event ordering guarantees
4. Medium: Add missing test coverage
5. Low: Consolidate duplicate imports and clean up unused imports

---

Review generated by Claude Code

[graphite-app]

Merge activity

• Dec 3, 12:33 AM UTC: NathanFlurry added this pull request to the Graphite merge queue.
• Dec 3, 12:34 AM UTC: CI is running for this pull request on a draft pull request (#3577) due to your merge queue CI optimization settings.
• Dec 3, 12:34 AM UTC: Merged by the Graphite merge queue via draft PR: #3577.