diff --git a/crates/executor-core/src/sdk/kv.rs b/crates/executor-core/src/sdk/kv.rs index f5024e6..45e2002 100644 --- a/crates/executor-core/src/sdk/kv.rs +++ b/crates/executor-core/src/sdk/kv.rs @@ -5,6 +5,8 @@ //! widgets.set("k", #{ n: 1 }); //! let v = widgets.get("k"); // value or () if absent //! if widgets.has("k") { ... } +//! widgets.set_if("k", (), #{ n: 0 }); // insert only if ABSENT -> bool +//! widgets.set_if("k", old, new); // swap only if current == old -> bool //! widgets.delete("k"); // bool (was-present) //! let page = widgets.list(); // returns #{ keys: [...], next_cursor: () } //! ``` @@ -105,6 +107,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Option { + if expected.is_unit() { + None + } else { + Some(dynamic_to_json(expected)) + } +} + fn register_get(engine: &mut RhaiEngine) { engine.register_fn( "get", @@ -145,6 +160,27 @@ fn register_set(engine: &mut RhaiEngine) { ); } +fn register_set_if(engine: &mut RhaiEngine) { + // `handle.set_if(key, expected, new)` — CAS. `expected = ()` means "only if + // absent". Returns `true` if the swap happened, `false` if the precondition + // failed. + engine.register_fn( + "set_if", + |handle: &mut KvHandle, + key: &str, + expected: Dynamic, + new: Dynamic| + -> Result> { + let h = handle.clone(); + let exp = expected_from_dynamic(&expected); + let new = dynamic_to_json(&new); + block_on("kv", async move { + h.service.set_if(&h.cx, &h.collection, key, exp, new).await + }) + }, + ); +} + fn register_has(engine: &mut RhaiEngine) { engine.register_fn( "has", @@ -243,6 +279,24 @@ fn register_group_set(engine: &mut RhaiEngine) { ); } +fn register_group_set_if(engine: &mut RhaiEngine) { + engine.register_fn( + "set_if", + |handle: &mut GroupKvHandle, + key: &str, + expected: Dynamic, + new: Dynamic| + -> Result> { + let h = handle.clone(); + let exp = expected_from_dynamic(&expected); + let new = dynamic_to_json(&new); + block_on("kv", async move { + h.service.set_if(&h.cx, &h.collection, key, exp, new).await + }) + }, + ); +} + fn register_group_has(engine: &mut RhaiEngine) { engine.register_fn( "has", diff --git a/crates/executor-core/tests/sdk_kv.rs b/crates/executor-core/tests/sdk_kv.rs index c68ea37..775ac1d 100644 --- a/crates/executor-core/tests/sdk_kv.rs +++ b/crates/executor-core/tests/sdk_kv.rs @@ -52,6 +52,27 @@ impl KvService for InMemoryKv { Ok(()) } + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: Value, + ) -> Result { + let mut data = self.data.lock().await; + let k = (cx.app_id, collection.to_string(), key.to_string()); + let matches = match (&expected, data.get(&k)) { + (None, None) => true, + (Some(exp), Some(cur)) => exp == cur, + _ => false, + }; + if matches { + data.insert(k, new); + } + Ok(matches) + } + async fn delete(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result { Ok(self .data @@ -167,6 +188,27 @@ async fn kv_set_then_get_round_trip() { assert_eq!(body, json!({ "n": 1 })); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn kv_set_if_compare_and_swap() { + let engine = make_engine(); + let app = AppId::new(); + // `set_if(key, (), new)` inserts only when absent; `set_if(key, expected, + // new)` swaps only on match. Returns a bool each time. + let src = r#" + let c = kv::collection("counters"); + let first = c.set_if("n", (), 1); // absent -> inserts, true + let dup = c.set_if("n", (), 2); // present -> false + let bad = c.set_if("n", 99, 3); // mismatch -> false + let good = c.set_if("n", 1, 3); // match -> swaps, true + #{ first: first, dup: dup, bad: bad, good: good, final: c.get("n") } + "#; + let body = run_script(engine, src, baseline_request(app)).await; + assert_eq!( + body, + json!({ "first": true, "dup": false, "bad": false, "good": true, "final": 3 }) + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn kv_get_missing_returns_unit() { let engine = make_engine(); diff --git a/crates/manager-core/src/group_kv_repo.rs b/crates/manager-core/src/group_kv_repo.rs index e53ecdc..139225c 100644 --- a/crates/manager-core/src/group_kv_repo.rs +++ b/crates/manager-core/src/group_kv_repo.rs @@ -39,6 +39,18 @@ pub trait GroupKvRepo: Send + Sync { value: serde_json::Value, ) -> Result, GroupKvRepoError>; + /// Compare-and-swap. Atomically writes `new` iff the current value equals + /// `expected` (`expected = None` → iff the key is ABSENT). Returns `true` + /// when the write happened. One SQL statement — atomic without a tx. + async fn set_if( + &self, + group_id: GroupId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result; + /// Returns the deleted value if present, `None` if the row didn't exist. async fn delete( &self, @@ -143,6 +155,43 @@ impl GroupKvRepo for PostgresGroupKvRepo { Ok(row.and_then(|(v,)| v)) } + async fn set_if( + &self, + group_id: GroupId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + let affected = match expected { + Some(exp) => sqlx::query( + "UPDATE group_kv_entries SET value = $4, updated_at = NOW() \ + WHERE group_id = $1 AND collection = $2 AND key = $3 AND value = $5", + ) + .bind(group_id.into_inner()) + .bind(collection) + .bind(key) + .bind(new) + .bind(exp) + .execute(&self.pool) + .await? + .rows_affected(), + None => sqlx::query( + "INSERT INTO group_kv_entries (group_id, collection, key, value) \ + VALUES ($1, $2, $3, $4) \ + ON CONFLICT (group_id, collection, key) DO NOTHING", + ) + .bind(group_id.into_inner()) + .bind(collection) + .bind(key) + .bind(new) + .execute(&self.pool) + .await? + .rows_affected(), + }; + Ok(affected == 1) + } + async fn delete( &self, group_id: GroupId, diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index f42a461..6347fee 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/crates/manager-core/src/group_kv_service.rs @@ -263,6 +263,66 @@ impl GroupKvService for GroupKvServiceImpl { Ok(()) } + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + let group_id = self.owning_group(cx, collection).await?; + let encoded_len = serde_json::to_vec(&new) + .map(|v| v.len()) + .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?; + if encoded_len > self.max_value_bytes { + return Err(GroupKvError::ValueTooLarge { + limit: self.max_value_bytes, + actual: encoded_len, + }); + } + self.check_write(cx, group_id).await?; + // Same quota rules as `set` (best-effort pre-check; the swap itself is + // atomic in the repo). A swap that inserts a NEW key (expected absent) + // must fit the row ceiling; the byte ceiling uses the projected total. + let existing = self.repo.get(group_id, collection, key).await?; + if expected.is_none() && existing.is_none() { + let count = self.repo.count_rows(group_id).await?; + if count >= self.max_rows { + return Err(GroupKvError::QuotaExceeded { + limit: usize::try_from(self.max_rows).unwrap_or(usize::MAX), + actual: usize::try_from(count).unwrap_or(usize::MAX), + }); + } + } + let old_len = existing + .as_ref() + .and_then(|v| serde_json::to_vec(v).ok()) + .map_or(0u64, |b| b.len() as u64); + let used = self.repo.total_bytes(group_id).await?; + let projected = used.saturating_sub(old_len) + encoded_len as u64; + if projected > self.max_total_bytes { + return Err(GroupKvError::TotalBytesQuotaExceeded { + limit: self.max_total_bytes, + actual: projected, + }); + } + let op = if expected.is_some() { + "update" + } else { + "insert" + }; + let swapped = self + .repo + .set_if(group_id, collection, key, expected, new.clone()) + .await?; + if swapped { + self.emit_shared(cx, group_id, op, collection, key, Some(new)) + .await; + } + Ok(swapped) + } + async fn delete( &self, cx: &SdkCallCx, @@ -348,6 +408,27 @@ mod tests { .insert((group_id, collection.to_string(), key.to_string()), value)) } + async fn set_if( + &self, + group_id: GroupId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + let mut data = self.data.lock().await; + let k = (group_id, collection.to_string(), key.to_string()); + let matches = match (&expected, data.get(&k)) { + (None, None) => true, + (Some(exp), Some(cur)) => exp == cur, + _ => false, + }; + if matches { + data.insert(k, new); + } + Ok(matches) + } + async fn delete( &self, group_id: GroupId, @@ -643,4 +724,56 @@ mod tests { let err = kv.get(&cx, "", "k").await.unwrap_err(); assert!(matches!(err, GroupKvError::InvalidCollection)); } + + #[tokio::test] + async fn set_if_is_compare_and_swap_and_writes_fail_closed() { + let app = AppId::new(); + let group = GroupId::new(); + let mut resolver = FakeResolver::default(); + resolver.map.insert((app, "catalog".into()), group); + let kv = svc(resolver); + let cx = cx_with(app, Some(owner())); + + // Insert-if-absent, then swap-if-equal (as the per-app path). + assert!(kv + .set_if(&cx, "catalog", "k", None, serde_json::json!(1)) + .await + .unwrap()); + assert!(!kv + .set_if(&cx, "catalog", "k", None, serde_json::json!(2)) + .await + .unwrap()); + assert!(!kv + .set_if( + &cx, + "catalog", + "k", + Some(serde_json::json!(9)), + serde_json::json!(3) + ) + .await + .unwrap()); + assert!(kv + .set_if( + &cx, + "catalog", + "k", + Some(serde_json::json!(1)), + serde_json::json!(3) + ) + .await + .unwrap()); + assert_eq!( + kv.get(&cx, "catalog", "k").await.unwrap(), + Some(serde_json::json!(3)) + ); + + // A shared-collection CAS still fails closed for an anonymous writer. + let anon = cx_with(app, None); + let err = kv + .set_if(&anon, "catalog", "k", None, serde_json::json!(4)) + .await + .unwrap_err(); + assert!(matches!(err, GroupKvError::Forbidden)); + } } diff --git a/crates/manager-core/src/kv_repo.rs b/crates/manager-core/src/kv_repo.rs index 750d1bb..c7a1a20 100644 --- a/crates/manager-core/src/kv_repo.rs +++ b/crates/manager-core/src/kv_repo.rs @@ -39,6 +39,19 @@ pub trait KvRepo: Send + Sync { value: serde_json::Value, ) -> Result, KvRepoError>; + /// Compare-and-swap. Atomically writes `new` iff the current value equals + /// `expected` (`expected = None` → iff the key is ABSENT). Returns `true` + /// when the write happened. A single SQL statement, so it is atomic without a + /// transaction. + async fn set_if( + &self, + app_id: AppId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result; + /// Returns the deleted value if present, `None` if the row didn't /// exist. The caller turns the `bool was-present` part into the /// SDK's return value; the `Option` part feeds the @@ -131,6 +144,45 @@ impl KvRepo for PostgresKvRepo { Ok(row.and_then(|(v,)| v)) } + async fn set_if( + &self, + app_id: AppId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + let affected = match expected { + // Swap iff the current value matches (JSONB `=` is semantic equality). + Some(exp) => sqlx::query( + "UPDATE kv_entries SET value = $4, updated_at = NOW() \ + WHERE app_id = $1 AND collection = $2 AND key = $3 AND value = $5", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .bind(new) + .bind(exp) + .execute(&self.pool) + .await? + .rows_affected(), + // Insert iff absent. + None => sqlx::query( + "INSERT INTO kv_entries (app_id, collection, key, value) \ + VALUES ($1, $2, $3, $4) \ + ON CONFLICT (app_id, collection, key) DO NOTHING", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .bind(new) + .execute(&self.pool) + .await? + .rows_affected(), + }; + Ok(affected == 1) + } + async fn delete( &self, app_id: AppId, diff --git a/crates/manager-core/src/kv_service.rs b/crates/manager-core/src/kv_service.rs index 71b07e8..00a2f4f 100644 --- a/crates/manager-core/src/kv_service.rs +++ b/crates/manager-core/src/kv_service.rs @@ -186,6 +186,59 @@ impl KvService for KvServiceImpl { Ok(()) } + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + validate_collection(collection)?; + let encoded_len = serde_json::to_vec(&new) + .map(|v| v.len()) + .map_err(|e| KvError::Backend(format!("encode value: {e}")))?; + if encoded_len > self.max_value_bytes { + return Err(KvError::ValueTooLarge { + limit: self.max_value_bytes, + actual: encoded_len, + }); + } + self.check_write(cx).await?; + // `insert` when the precondition was "absent", else `update`. The op is + // only emitted on a successful swap. + let op = if expected.is_some() { + "update" + } else { + "insert" + }; + let swapped = self + .repo + .set_if(cx.app_id, collection, key, expected, new.clone()) + .await?; + if swapped { + // Non-transactional with the write (same caveat as `set`). + if let Err(e) = self + .events + .emit( + cx, + ServiceEvent { + source: "kv", + op, + collection: Some(collection.to_string()), + key: Some(key.to_string()), + payload: Some(new), + old_payload: None, + }, + ) + .await + { + tracing::error!(error = %e, source = "kv", op, event_emit_failure = true, "event emit failed"); + } + } + Ok(swapped) + } + async fn delete(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result { validate_collection(collection)?; self.check_write(cx).await?; @@ -283,6 +336,28 @@ mod tests { .insert((app_id, collection.to_string(), key.to_string()), value)) } + async fn set_if( + &self, + app_id: AppId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + // Holding the map lock makes the compare-and-set atomic in-memory. + let mut data = self.data.lock().await; + let k = (app_id, collection.to_string(), key.to_string()); + let matches = match (&expected, data.get(&k)) { + (None, None) => true, // insert-if-absent + (Some(exp), Some(cur)) => exp == cur, // swap-if-equal + _ => false, + }; + if matches { + data.insert(k, new); + } + Ok(matches) + } + async fn delete( &self, app_id: AppId, @@ -464,6 +539,63 @@ mod tests { assert!(matches!(err, KvError::InvalidCollection)); } + #[tokio::test] + async fn set_if_is_compare_and_swap() { + let kv = svc(); + let cx = anon_cx(AppId::new()); + // Insert-if-absent: expected () swaps only when the key is missing. + assert!( + kv.set_if(&cx, "cas", "k", None, serde_json::json!(1)) + .await + .unwrap(), + "insert-if-absent must succeed on a missing key" + ); + assert!( + !kv.set_if(&cx, "cas", "k", None, serde_json::json!(2)) + .await + .unwrap(), + "insert-if-absent must fail once the key exists" + ); + assert_eq!( + kv.get(&cx, "cas", "k").await.unwrap(), + Some(serde_json::json!(1)) + ); + + // Swap-if-equal: wrong expected → no-op; correct expected → swaps. + assert!( + !kv.set_if( + &cx, + "cas", + "k", + Some(serde_json::json!(99)), + serde_json::json!(3) + ) + .await + .unwrap(), + "a mismatched expected must not swap" + ); + assert_eq!( + kv.get(&cx, "cas", "k").await.unwrap(), + Some(serde_json::json!(1)) + ); + assert!( + kv.set_if( + &cx, + "cas", + "k", + Some(serde_json::json!(1)), + serde_json::json!(3) + ) + .await + .unwrap(), + "a matching expected must swap" + ); + assert_eq!( + kv.get(&cx, "cas", "k").await.unwrap(), + Some(serde_json::json!(3)) + ); + } + /// Load-bearing: a script with `cx.app_id = A` must NOT see /// entries inserted under `cx.app_id = B`. This is the cross-app /// isolation boundary; getting this wrong is a security diff --git a/crates/shared/src/group_kv.rs b/crates/shared/src/group_kv.rs index 99b550a..216d118 100644 --- a/crates/shared/src/group_kv.rs +++ b/crates/shared/src/group_kv.rs @@ -37,6 +37,25 @@ pub trait GroupKvService: Send + Sync { value: serde_json::Value, ) -> Result<(), GroupKvError>; + /// Compare-and-swap on a shared-collection key. Atomically writes `new` only + /// if the current value equals `expected` — or, when `expected` is `None`, + /// only if the key is ABSENT. Returns `true` on swap, `false` if the + /// precondition failed. Same reads-open / writes-authed + quota rules as + /// [`Self::set`]. Default: unsupported. + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + let _ = (cx, collection, key, expected, new); + Err(GroupKvError::Backend( + "set_if is not supported by this group-kv implementation".into(), + )) + } + async fn delete( &self, cx: &SdkCallCx, diff --git a/crates/shared/src/kv.rs b/crates/shared/src/kv.rs index a0cb4d9..456ec98 100644 --- a/crates/shared/src/kv.rs +++ b/crates/shared/src/kv.rs @@ -36,6 +36,25 @@ pub trait KvService: Send + Sync { value: serde_json::Value, ) -> Result<(), KvError>; + /// Compare-and-swap. Atomically writes `new` only if the current value + /// equals `expected` — or, when `expected` is `None`, only if the key is + /// ABSENT (insert-if-absent). Returns `true` if the swap happened, `false` + /// if the precondition failed (value differed / key already present). + /// Default: unsupported, so only storage-backed impls opt in. + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, + ) -> Result { + let _ = (cx, collection, key, expected, new); + Err(KvError::Backend( + "set_if is not supported by this kv implementation".into(), + )) + } + async fn delete(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result; async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result; diff --git a/docs/sdk-shape.md b/docs/sdk-shape.md index 225afa3..dd55370 100644 --- a/docs/sdk-shape.md +++ b/docs/sdk-shape.md @@ -57,8 +57,18 @@ returned by an `::collection(name)` constructor: let widgets = kv::collection("widgets"); widgets.set("k", "v"); let v = widgets.get("k"); +// Compare-and-swap (v1.2, Track A M5): atomic, returns a bool. +widgets.set_if("k", (), "v0"); // insert only if ABSENT +widgets.set_if("k", "v0", "v1"); // swap only if current == "v0" ``` +`set_if(key, expected, new)` is an atomic compare-and-swap: an `expected` of +`()` means "only if the key is absent" (insert-if-absent); any other `expected` +swaps only when the stored value equals it. Returns `true` if the write +happened, `false` if the precondition failed — the primitive for lock-free +counters / optimistic concurrency. Available on both `kv::collection` and +`kv::shared_collection`. + Not `kv::set("widgets", "k", "v")`. The handle is a Rhai custom type the service registers; method calls bind to that type. This: