feat: state worker (builtin migration) - #436
Merged
Merged
Conversation
…nd/remove parity) Bumps the crate to edition 2024: the engine source uses let-chains; keeping the port byte-verbatim outweighs matching http/cron's edition 2021.
…p cadence, hot retune
…lace, SDK stub unregister)
…une, restart-tier adapter
…d comments - README parity table: note that trigger metadata is not forwarded to handlers (iii-sdk 0.20 TriggerRequest has no metadata field; same limitation as the http worker). - condition.rs: reword module doc from the http-worker's per-route middleware framing to state's actual context (condition gates change-event fan-out per trigger binding), and point the test comment at the real e2e coverage (condition_false_blocks_null_passes in tests/e2e_state.rs) instead of a nonexistent e2e_condition suite. - functions.rs / configuration.rs: backtick doc references to `#[function]` and to private items (StateCtx::snapshot, on_config_change) instead of broken intra-doc links.
…ltin Ports test_reconfigure_reverts_to_boot_interval_when_cleared, test_boot_interval_is_floored, test_reconfigure_in_memory_is_noop, and test_kv_store_invalid_store_method (renamed from test_builtin_kv_store_invalid_store_method) from the engine's built-in kv.rs test suite; KvStore's save-loop/lock surface is unchanged from the builtin so all four ported as-is.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughThis PR adds a new standalone Rust "state" worker ( ChangesState Worker
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Caller
participant StateFunctions as "state::* functions"
participant StateAdapter
participant TriggerTable
participant Invoker as "SdkInvoker"
participant ConditionFn as "condition function"
Caller->>StateFunctions: state::set(scope, key, value)
StateFunctions->>StateAdapter: set(scope, key, value)
StateAdapter-->>StateFunctions: old_value, new_value
StateFunctions->>TriggerTable: fan_out(event: Created/Updated)
TriggerTable->>TriggerTable: match bindings by scope/key
TriggerTable->>ConditionFn: check_condition(payload)
ConditionFn-->>TriggerTable: true/false/null
TriggerTable->>Invoker: call(function_id, event payload)
Invoker-->>TriggerTable: handler result/error
StateFunctions-->>Caller: StreamSetResult
sequenceDiagram
participant Main as "main.rs"
participant IIIClient
participant Boot as "boot::start"
participant Configuration as "configuration.rs"
participant Adapter as "StateAdapter"
Main->>IIIClient: register_worker
Main->>Configuration: register_config(seed)
Main->>Configuration: fetch_config
Main->>Boot: start(iii, config)
Boot->>Boot: guard_against_builtin_state
Boot->>Adapter: build_adapter(config)
Boot->>Boot: register state trigger + functions
Boot-->>Main: BootHandle
Main->>Configuration: register_config_trigger(ctx, apply_lock)
Configuration-->>Boot: on configuration:updated, reconfigure adapter
Main->>Boot: shutdown()
Boot->>Adapter: destroy()
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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 |
Contributor
skill-check — worker0 verified, 40 skipped (no docs/).
Four for four. Nicely done. |
guibeira
marked this pull request as ready for review
July 8, 2026 11:32
The registry publish gate (collect_worker_interface.py --assert-typed-schemas) rejects state::get, state::delete, and state::list: their handlers return Option<serde_json::Value> (the raw stored value, or null), which schemars auto-extracts to the permissive Nullable_AnyValue schema — no type/properties keyword, flagged as "unknown" by the gate. Override response_format with an explicit typed "JSON value or null" schema (type: [string, number, boolean, object, array, null]) for the three handlers. This only annotates the registered schema; the handlers still return the exact same Option<Value> on the wire. Verified by reproducing the CI collection + assert locally against a live engine + worker: worker-interface.json now passes --assert-non-empty --assert-typed-schemas for all 7 state::* functions. 96 unit tests + 8 e2e tests still green.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
state/src/configuration.rs (1)
89-102: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid retrying a definitive
NOT_FOUNDthroughtrigger_with_retry.
try_get_config_valueroutesconfiguration::getthroughtrigger_with_retry, which retries every error uniformly. ANOT_FOUND(the expected first-boot state) is therefore retriedCONFIG_RETRIEStimes with cumulative backoff before being recognized here. Since bothshould_seed_initial_valueandfetch_configcall this on boot, first-run startup pays this wait twice for a result that is already definitive. Consider short-circuitingNOT_FOUNDbefore the retry loop.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@state/src/configuration.rs` around lines 89 - 102, `try_get_config_value` is retrying a definitive `NOT_FOUND` from `trigger_with_retry`, which causes unnecessary boot delays in both `should_seed_initial_value` and `fetch_config`. Update the `configuration::get` path so `NOT_FOUND` is detected before entering the retry loop, and return `Ok(None)` immediately instead of letting `trigger_with_retry` apply `CONFIG_RETRIES` and backoff. Keep the existing `trigger_with_retry` behavior for other errors, and preserve the `try_get_config_value` return shape.state/src/store.rs (1)
178-209: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider cloning
configonce instead of repeatedly.
config.clone()is called three times (lines 181, 195, 204) to extractstore_method,file_path, andsave_interval_ms. Whileserde_json::Valueclones are Arc-cheap, a single clone at the top would be cleaner.♻️ Optional refactor
pub fn new(config: Option<Value>) -> Self { tracing::debug!("Initializing KvStore with config: {:?}", config); - let store_method = config - .clone() - .and_then(|cfg| { - cfg.get("store_method") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()) - }) - .unwrap_or_else(|| "in_memory".to_string()); + let cfg = config.as_ref(); + let store_method = cfg + .and_then(|c| c.get("store_method").and_then(|v| v.as_str())) + .unwrap_or("in_memory") + .to_string(); - if store_method == "in_memory" { + if store_method == "in_memory" { tracing::warn!( "DO NOT USE IN_MEMORY STORE_METHOD IN PRODUCTION - DATA WILL BE LOST ON SHUTDOWN" ); } - let file_path = config - .clone() - .and_then(|cfg| { - cfg.get("file_path") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()) - }) - .unwrap_or_else(|| "kv_store_data.db".to_string()); + let file_path = cfg + .and_then(|c| c.get("file_path").and_then(|v| v.as_str())) + .unwrap_or("kv_store_data.db") + .to_string(); - let interval = config - .clone() - .and_then(|cfg| cfg.get("save_interval_ms").and_then(|v| v.as_u64())) + let interval = cfg + .and_then(|c| c.get("save_interval_ms").and_then(|v| v.as_u64())) .filter(|&n| n > 0) .unwrap_or(DEFAULT_SAVE_INTERVAL_MS) .max(MIN_SAVE_INTERVAL_MS);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@state/src/store.rs` around lines 178 - 209, Clone config once at the start of KvStore::new and reuse that cloned value to derive store_method, file_path, and save_interval_ms instead of calling config.clone() multiple times. Keep the existing extraction logic inside new, but switch each lookup to read from the single cloned config so the constructor is cleaner and easier to maintain.state/src/events.rs (1)
25-38: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAdd timeout to
SdkInvoker::call
timeout_ms: Nonemeans a hanging trigger handler blocks the spawnedfan_outtask indefinitely, leaking resources.check_conditionincondition.rsalready accepts atimeout_msparameter — pass a configurable timeout here too for consistency.♻️ Proposed fix
pub struct SdkInvoker { pub iii: Arc<IIIClient>, + pub timeout_ms: u64, } #[async_trait::async_trait] impl Invoker for SdkInvoker { async fn call(&self, function_id: &str, payload: Value) -> Result<Value, String> { self.iii .trigger(TriggerRequest { function_id: function_id.to_string(), payload, action: None, - timeout_ms: None, + timeout_ms: Some(self.timeout_ms), }) .await .map_err(|e| e.to_string()) } }The
timeout_msvalue can be sourced fromStateConfigat construction time, similar to howcheck_conditionreceives it.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@state/src/events.rs` around lines 25 - 38, Add a configurable timeout to SdkInvoker::call so TriggerRequest is not created with timeout_ms: None. Thread the timeout in from StateConfig at construction time for SdkInvoker, and pass it through in the trigger call the same way check_condition in condition.rs already uses a timeout_ms value. Update the SdkInvoker constructor/fields and the call method to use that stored timeout consistently.state/src/adapters.rs (1)
587-607: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRemove the
MutexaroundConnectionManagerConnectionManageralready supports shared concurrent use, so the lock just serializes Redis operations and adds avoidable contention. Store it directly and clone it where each method needs a handle.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@state/src/adapters.rs` around lines 587 - 607, The RedisAdapter::new constructor currently wraps ConnectionManager in Arc<Mutex<_>>, but ConnectionManager is already safe for shared concurrent use and the mutex adds unnecessary contention. Update RedisAdapter to store the ConnectionManager directly (or inside an Arc without Mutex if sharing is needed) and adjust the RedisAdapter field and any methods that currently lock publisher to clone and use the connection manager handle instead of serializing access.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@state/src/main.rs`:
- Around line 100-108: The post-boot path in start() can exit early on
register_config_trigger() failure without cleaning up the started boot state.
Wrap the booted section so boot.shutdown() is always called on any error path
after iii_state::boot::start(), including the register_config_trigger() call,
and ensure iii.shutdown_async() still runs during overall teardown; use the
existing boot and register_config_trigger symbols to keep the cleanup localized.
---
Nitpick comments:
In `@state/src/adapters.rs`:
- Around line 587-607: The RedisAdapter::new constructor currently wraps
ConnectionManager in Arc<Mutex<_>>, but ConnectionManager is already safe for
shared concurrent use and the mutex adds unnecessary contention. Update
RedisAdapter to store the ConnectionManager directly (or inside an Arc without
Mutex if sharing is needed) and adjust the RedisAdapter field and any methods
that currently lock publisher to clone and use the connection manager handle
instead of serializing access.
In `@state/src/configuration.rs`:
- Around line 89-102: `try_get_config_value` is retrying a definitive
`NOT_FOUND` from `trigger_with_retry`, which causes unnecessary boot delays in
both `should_seed_initial_value` and `fetch_config`. Update the
`configuration::get` path so `NOT_FOUND` is detected before entering the retry
loop, and return `Ok(None)` immediately instead of letting `trigger_with_retry`
apply `CONFIG_RETRIES` and backoff. Keep the existing `trigger_with_retry`
behavior for other errors, and preserve the `try_get_config_value` return shape.
In `@state/src/events.rs`:
- Around line 25-38: Add a configurable timeout to SdkInvoker::call so
TriggerRequest is not created with timeout_ms: None. Thread the timeout in from
StateConfig at construction time for SdkInvoker, and pass it through in the
trigger call the same way check_condition in condition.rs already uses a
timeout_ms value. Update the SdkInvoker constructor/fields and the call method
to use that stored timeout consistently.
In `@state/src/store.rs`:
- Around line 178-209: Clone config once at the start of KvStore::new and reuse
that cloned value to derive store_method, file_path, and save_interval_ms
instead of calling config.clone() multiple times. Keep the existing extraction
logic inside new, but switch each lookup to read from the single cloned config
so the constructor is cleaner and easier to maintain.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: fa472aef-0ecb-4681-b484-ca0e01049313
⛔ Files ignored due to path filters (1)
state/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (25)
.github/workflows/create-tag.yml.github/workflows/release.ymldocs/adr/0001-state-worker-store-location.mdstate/Cargo.tomlstate/README.mdstate/build.rsstate/iii.worker.yamlstate/skills/SKILL.mdstate/src/adapters.rsstate/src/boot.rsstate/src/condition.rsstate/src/config.rsstate/src/configuration.rsstate/src/events.rsstate/src/functions.rsstate/src/lib.rsstate/src/main.rsstate/src/manifest.rsstate/src/store.rsstate/src/structs.rsstate/src/trigger.rsstate/src/update_ops.rsstate/tests/common/engine.rsstate/tests/common/mod.rsstate/tests/e2e_state.rs
Comment on lines
+100
to
+108
| let boot = iii_state::boot::start(iii.clone(), config.clone()).await?; | ||
| tracing::info!(adapter = %config.effective_adapter_name(), "iii-state ready"); | ||
|
|
||
| // Subscribe to configuration:updated so triggers_enabled/max_value_bytes | ||
| // reload live and a save_interval_ms change respawns the adapter's save | ||
| // loop (see configuration). | ||
| configuration::register_config_trigger(&iii, boot.ctx.clone(), boot.apply_lock.clone()) | ||
| .map_err(anyhow::Error::msg) | ||
| .context("binding configuration trigger")?; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Call boot.shutdown() on the register_config_trigger error path.
If register_config_trigger fails, the ? operator returns early without calling boot.shutdown() or iii.shutdown_async(). The kv adapter's save loop may have pending dirty entries that won't flush, and the engine won't receive a clean disconnect. Wrap the post-boot logic so shutdown always runs.
🛡️ Proposed fix: guarantee cleanup on all post-boot paths
let boot = iii_state::boot::start(iii.clone(), config.clone()).await?;
tracing::info!(adapter = %config.effective_adapter_name(), "iii-state ready");
-// Subscribe to configuration:updated so triggers_enabled/max_value_bytes
-// reload live and a save_interval_ms change respawns the adapter's save
-// loop (see configuration).
-configuration::register_config_trigger(&iii, boot.ctx.clone(), boot.apply_lock.clone())
- .map_err(anyhow::Error::msg)
- .context("binding configuration trigger")?;
-
-tokio::signal::ctrl_c().await?;
-tracing::info!("iii-state shutting down");
-boot.shutdown().await;
-iii.shutdown_async().await;
-Ok(())
+let result = async {
+ // Subscribe to configuration:updated so triggers_enabled/max_value_bytes
+ // reload live and a save_interval_ms change respawns the adapter's save
+ // loop (see configuration).
+ configuration::register_config_trigger(&iii, boot.ctx.clone(), boot.apply_lock.clone())
+ .map_err(anyhow::Error::msg)
+ .context("binding configuration trigger")?;
+
+ tokio::signal::ctrl_c().await?;
+ tracing::info!("iii-state shutting down");
+ Ok::<(), anyhow::Error>(())
+}
+.await;
+
+boot.shutdown().await;
+iii.shutdown_async().await;
+result📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let boot = iii_state::boot::start(iii.clone(), config.clone()).await?; | |
| tracing::info!(adapter = %config.effective_adapter_name(), "iii-state ready"); | |
| // Subscribe to configuration:updated so triggers_enabled/max_value_bytes | |
| // reload live and a save_interval_ms change respawns the adapter's save | |
| // loop (see configuration). | |
| configuration::register_config_trigger(&iii, boot.ctx.clone(), boot.apply_lock.clone()) | |
| .map_err(anyhow::Error::msg) | |
| .context("binding configuration trigger")?; | |
| let boot = iii_state::boot::start(iii.clone(), config.clone()).await?; | |
| tracing::info!(adapter = %config.effective_adapter_name(), "iii-state ready"); | |
| let result = async { | |
| // Subscribe to configuration:updated so triggers_enabled/max_value_bytes | |
| // reload live and a save_interval_ms change respawns the adapter's save | |
| // loop (see configuration). | |
| configuration::register_config_trigger(&iii, boot.ctx.clone(), boot.apply_lock.clone()) | |
| .map_err(anyhow::Error::msg) | |
| .context("binding configuration trigger")?; | |
| tokio::signal::ctrl_c().await?; | |
| tracing::info!("iii-state shutting down"); | |
| Ok::<(), anyhow::Error>(()) | |
| } | |
| .await; | |
| boot.shutdown().await; | |
| iii.shutdown_async().await; | |
| result |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@state/src/main.rs` around lines 100 - 108, The post-boot path in start() can
exit early on register_config_trigger() failure without cleaning up the started
boot state. Wrap the booted section so boot.shutdown() is always called on any
error path after iii_state::boot::start(), including the
register_config_trigger() call, and ensure iii.shutdown_async() still runs
during overall teardown; use the existing boot and register_config_trigger
symbols to keep the cleanup localized.
This was referenced Jul 24, 2026
Merged
This was referenced Jul 31, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
stateregistry worker, replacing the built-iniii-state.statetrigger type and the sixstate::*functions (set,get,delete,update,list,list_groups) with exact id/input/output parity.BuiltinKvStore→KvStore) with a byte-identical on-disk format (rkyv.binper scope), plus theredisadapter with the samestate:<scope>hash layout and atomic Lua update script.state:created/state:updated/state:deleted, scope/key pre-filter, condition functions (only explicitfalseblocks),triggers_enabledlive gate.configurationworker: live (triggers_enabled,max_value_bytes), task-rebuild (save_interval_mssave-loop respawn), restart-tier (adapter).iii-stateis active, with a store-migration hint.Behavior
statetrigger type and all sixstate::*function ids with identical input/output shapes.state_store.dbloads unchanged.metadatanot forwarded (no field on iii-sdk 0.20TriggerRequest, same limitation as the http worker);bridgeadapter not ported (ADR 0001); telemetry counters out of scope.iii-stateworker is still connected.update_ops.rs) compile unchanged.Validation
cd state && cargo test— 96 unit tests green (includes the verbatim-portedupdate_opssuite and the builtin-format kv-store compat test)III_E2E_REQUIRE=1 cargo test --test e2e_stateagainst a live engine (workers: []) — 8/8 e2e cases pass (ops/trigger/config parity, hot-reload)cargo test --test e2e_statewithout an engine — connect-or-skip, greencd state && cargo fmt --checkcd state && cargo clippy --all-targets -- -D warningscargo doc --no-deps— zero warningscd state && cargo build && ./target/debug/state --manifest | head -5python3 .github/scripts/discover_changed_workers.py --base main— state lands in the rust bucketpython3 .github/scripts/validate_worker.py --worker state --base-ref main --source-changed '["state"]'— exit 0, no hard failurespython3 .github/scripts/build_skills_payload.py --worker state --version 0.1.0— collects the skillNotes
state/v0.1.0is post-merge only, via the Create Tag workflow (this PR wiresstateinto create-tag.yml and'state/v*'into release.yml).docs/adr/0001-state-worker-store-location.md) records the store-location decision (store lives in the worker process) and the benchmark waiver.test_python_state/) not yet run — pending before flipping the PR to ready.Summary by CodeRabbit
New Features
Documentation
Tests