diff --git a/src/lib.rs b/src/lib.rs index 42af154..158225a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3391,7 +3391,7 @@ impl OrchestrationContext { } /// Clear all keys from the KV store. - /// + /// /// Emits a `KeyValuesCleared` history event. After this call, /// `get_kv_value(key)` returns `None` for all keys. pub fn clear_all_kv_values(&self) { @@ -3408,8 +3408,13 @@ impl OrchestrationContext { /// turn are never prunable (they haven't been acked yet). /// /// Returns the number of keys cleared. + /// + /// The keys are cleared in sorted order so that the emitted `ClearKeyValue` + /// actions are deterministic: `kv_metadata` is a `HashMap`, whose iteration + /// order differs between runs, and replay compares actions against recorded + /// events in order (fixes microsoft/duroxide#57). pub fn prune_kv_values_updated_before(&self, updated_before_ms: u64) -> usize { - let keys_to_clear: Vec = { + let mut keys_to_clear: Vec = { let inner = self.inner.lock().unwrap(); inner .kv_metadata @@ -3419,6 +3424,8 @@ impl OrchestrationContext { .map(|(key, _)| key.clone()) .collect() }; + // Deterministic emission order across replays (see doc comment above). + keys_to_clear.sort(); let count = keys_to_clear.len(); for key in keys_to_clear { self.clear_kv_value(key); diff --git a/tests/kv_store_tests.rs b/tests/kv_store_tests.rs index b57c86a..c76052a 100644 --- a/tests/kv_store_tests.rs +++ b/tests/kv_store_tests.rs @@ -13,6 +13,7 @@ mod common; use duroxide::runtime::{self, OrchestrationStatus, registry::ActivityRegistry}; +use duroxide::providers::Provider; use duroxide::{ActivityContext, OrchestrationContext, OrchestrationRegistry}; use std::sync::Arc; use std::time::Duration; @@ -2573,6 +2574,80 @@ async fn kv_prune_old_values_by_timestamp() { rt.shutdown(None).await; } +/// Regression test for https://github.com/microsoft/duroxide/issues/57: +/// prune_kv_values_updated_before must emit ClearKeyValue actions in a +/// deterministic (sorted) key order. kv_metadata is a HashMap whose iteration +/// order varies between runs, and the replay engine compares emitted actions +/// against recorded events in order — unsorted emission fails replays with +/// `nondeterministic: kv clear mismatch`. +/// +/// Strategy: exec 1 sets four keys in deliberately unsorted insertion order, +/// then ContinueAsNew to exec 2, which prunes them all. The KeyValueCleared +/// events recorded in exec 2's history must appear in sorted key order. +#[tokio::test] +async fn kv_prune_emits_clears_in_sorted_key_order() { + let store = Arc::new( + duroxide::providers::sqlite::SqliteProvider::new_in_memory() + .await + .unwrap(), + ); + let activities = ActivityRegistry::builder().build(); + let orchestrations = OrchestrationRegistry::builder() + .register("PruneOrder", |ctx: OrchestrationContext, input: String| async move { + let n: u32 = input.parse().unwrap_or(0); + if n == 0 { + // Exec 1: set keys in unsorted insertion order, then CAN. + for key in ["zeta", "alpha", "mike", "beta"] { + ctx.set_kv_value(key, "v"); + } + ctx.continue_as_new("1".to_string()).await + } else { + // Exec 2: snapshot has the four keys with their exec-1 timestamps. + // Sleep briefly so that "now" is definitely after those timestamps. + ctx.schedule_timer(Duration::from_millis(50)).await; + let now = ctx + .utc_now() + .await + .map_err(|e| e.to_string())? + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + let removed = ctx.prune_kv_values_updated_before(now); + Ok(format!("removed:{removed}")) + } + }) + .build(); + + let rt = runtime::Runtime::start_with_store(store.clone(), activities, orchestrations).await; + let client = duroxide::Client::new(store.clone()); + client.start_orchestration("kv-prune-order", "PruneOrder", "0").await.unwrap(); + + let status = client + .wait_for_orchestration("kv-prune-order", Duration::from_secs(10)) + .await + .unwrap(); + match status { + OrchestrationStatus::Completed { output, .. } => assert_eq!(output, "removed:4"), + OrchestrationStatus::Failed { details, .. } => panic!("failed: {}", details.display_message()), + _ => panic!("unexpected status"), + } + + // The recorded KeyValueCleared events must be in sorted key order so that a + // later replay of the same turn observes the same action sequence. + let mgmt = store.as_management_capability().expect("ProviderAdmin required"); + let history = mgmt.read_history("kv-prune-order").await.unwrap(); + let cleared: Vec = history + .iter() + .filter_map(|e| match &e.kind { + duroxide::EventKind::KeyValueCleared { key } => Some(key.clone()), + _ => None, + }) + .collect(); + assert_eq!(cleared, vec!["alpha", "beta", "mike", "zeta"]); + + rt.shutdown(None).await; +} + /// prune_kv_values_updated_before with threshold 0 does nothing (all timestamps >= 0). #[tokio::test] async fn kv_prune_with_zero_threshold_does_nothing() {