rust_store_core/command/
claim.rs

1//! claim 命令规划：并发领取原语（outbox / 任务分派的底层原语）。
2//!
3//! 语义：在 filter 命中的行中按 sort 取前 limit 行，**原子**写入领取标记（update），
4//! 并返回被领取的完整文档。原子性由 SQL 侧行锁（`FOR UPDATE SKIP LOCKED`）或
5//! 单写者（SQLite）在方言翻译层保证（见 `dialect::write::claim`）；core 只产出
6//! 命令序列，不做 IO。
7//!
8//! 计划形状（Host 契约）：
9//!
10//! ```json
11//! { "mode": "atomic_commands", "schema": "Outbox",
12//!   "commands": [
13//!     { "kind": "claim", "source": "...", "database": null, "schema": null,
14//!       "collection": "outbox", "filter": {...}, "sort": [["_id", 1]],
15//!       "limit": 10, "update": {...}, "nowStamp": 0 },
16//!     { "kind": "find", "source": "...", "database": null, "schema": null,
17//!       "collection": "outbox", "filter": { "_id": { "$in": "___claimed_ids___" } },
18//!       "projection": null, "placeholder": "___claimed_ids___" }
19//!   ] }
20//! ```
21//!
22//! `atomic_commands`：全部 commands 必须在**同一连接同一事务**内顺序执行；前一条
23//! `claim` 的结果集 `_id` 数组填充后一条的 [`CLAIMED_IDS`] 占位符（整值替换，Host
24//! 契约同 `{{phase1.ids}}`）。ids 为空（0 行命中）时 Host 跳过后续命令、直接返回
25//! 空集。mode 为新增值，不与既有 `two_phase` 语义重叠。
26//!
27//! MongoDB 无 claim 方言路径：Host 的 Mongo 执行器对 `kind: "claim"` 显式拒绝
28//! （宿主侧抛 PushdownUnsupportedError），core 不产出该路径。
29
30use serde_json::{json, Value};
31
32use crate::command::cmd::{cmd_claim, cmd_find};
33use crate::command::mutate::{build_raw_update, build_set_data, has_raw_operators};
34use crate::command::write_links::attach_write_links;
35use crate::command::{
36    ensure_context, forbid_t2q, CLAIMED_IDS, ERR_NO_BATCH_WRITE, MODE_ATOMIC_COMMANDS,
37};
38use crate::permission::{can_write_schema, Context};
39use crate::rbac::{ensure_write, merge_row_condition, WriteAction};
40use crate::schema::{Registry, Schema};
41use crate::types::{validate_condition, validate_condition_shape};
42
43/// claim 计划：领取 filter 命中的前 limit 行并写入领取标记。
44///
45/// `now` 由 Host 注入（core 无时钟与随机源）；`sort` 缺省 `[["_id", 1]]`（FIFO）；
46/// `limit <= 0` 显式拒绝（禁静默取 1——领取语义下静默改值会掩盖调用方错误）。
47#[allow(clippy::too_many_arguments)]
48pub fn plan_claim(
49    schema_name: &str,
50    registry: &Registry,
51    ctx: Option<&Context>,
52    filter: &Value,
53    update: &Value,
54    sort: Option<&Value>,
55    limit: i64,
56    now: i64,
57) -> Result<Value, String> {
58    ensure_context(registry, ctx)?;
59    // 并发原语属 DB 独有能力：text2query 档收缩（同 triggers / $pipeline 直通先例）
60    forbid_t2q(registry, "claim")?;
61    let schema = registry.get(schema_name)?;
62    validate_condition(filter)?;
63    validate_condition_shape(schema, filter, registry.profile())?;
64    if limit <= 0 {
65        return Err(format!(
66            "claim 的 limit 必须 > 0，实际 {limit}（禁静默取 1）"
67        ));
68    }
69    if update.as_object().map(|o| o.is_empty()).unwrap_or(true) {
70        return Err("claim 的 update 必须为非空对象（领取必须写入标记）".to_string());
71    }
72    // 拒写清单 + 静态写权限（updateMany 同款：不走 creator 探针）
73    if let Some(c) = ctx {
74        let deny_write =
75            crate::permission::has_any_role(c, &registry.role_rules().deny_write_roles);
76        if deny_write || !can_write_schema(registry.role_rules(), schema, ctx) {
77            return Err(ERR_NO_BATCH_WRITE.to_string());
78        }
79    }
80    // RBAC 表级写判定（deny-wins）+ 行级条件合并（与 updateMany 语义一致：
81    // ownerOnly / condition 并入 filter，匹配不到即 0 行）
82    ensure_write(registry, schema, ctx, WriteAction::Update)?;
83    let eff_condition =
84        merge_row_condition(registry, schema, ctx, "update", Some(filter.clone()))
85            .unwrap_or_else(|| json!({}));
86
87    let update_doc = if has_raw_operators(update) {
88        build_raw_update(registry, schema, ctx, update, now)
89    } else {
90        let mut set_data = build_set_data(registry, schema, ctx, update);
91        if schema.timestamps {
92            set_data.insert("updatedAt".to_string(), json!(now));
93        }
94        json!({ "$set": Value::Object(set_data) })
95    };
96    ensure_fields_declared(schema, &update_doc)?;
97
98    let sort_v = sort.cloned().unwrap_or_else(|| json!([["_id", 1]]));
99    let claim_cmd = cmd_claim(schema, &eff_condition, &sort_v, limit, &update_doc, now);
100    let mut find_cmd = cmd_find(
101        schema,
102        &json!({ "_id": { "$in": CLAIMED_IDS } }),
103        None,
104    );
105    if let Some(o) = find_cmd.as_object_mut() {
106        o.insert("placeholder".to_string(), json!(CLAIMED_IDS));
107    }
108    let plan = json!({
109        "mode": MODE_ATOMIC_COMMANDS,
110        "schema": schema_name,
111        "commands": [claim_cmd, find_cmd],
112    });
113    attach_write_links(plan, registry, &[schema_name], registry.write_link_policy())
114}
115
116/// update_doc 中的目标字段必须为 schema 已声明字段（豁免 `_id` 与 timestamps 自动列）。
117///
118/// claim 是宿主编程原语：未知列名属编程错误，规划期显式拒绝（不静默透传、
119/// 不依赖驱动层报错——D2 绝不静默）。
120pub(super) fn ensure_fields_declared(schema: &Schema, update_doc: &Value) -> Result<(), String> {
121    let exempt = |f: &str| {
122        f == "_id" || (schema.timestamps && (f == "createdAt" || f == "updatedAt"))
123    };
124    let Some(ops) = update_doc.as_object() else {
125        return Ok(());
126    };
127    for (op, body) in ops {
128        // 仅校验已知写操作符的字段体；其余（非法操作符 / 混合形态）由方言翻译层显式报错
129        if !matches!(op.as_str(), "$set" | "$inc" | "$unset") {
130            continue;
131        }
132        let Some(fields) = body.as_object() else {
133            continue;
134        };
135        for k in fields.keys() {
136            if !exempt(k) && !schema.fields.contains_key(k) {
137                return Err(format!(
138                    "并发原语（claim/compareAndSet）的 update 字段 \"{k}\" 未在 schema 声明（D2：规划期拒绝，不落库报错）"
139                ));
140            }
141        }
142    }
143    Ok(())
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149    use crate::command::plan_compare_and_set;
150    use crate::schema::Registry;
151    use serde_json::json;
152
153    fn registry() -> Registry {
154        let mut r = Registry::new();
155        r.register(&json!({
156            "name": "Outbox", "collection": "outbox", "timestamps": false,
157            "fields": {
158                "status": { "type": "string" },
159                "lockedBy": { "type": "string" },
160                "lockedAt": { "type": "int" }
161            },
162            "relations": {}
163        }))
164        .expect("schema 注册失败");
165        r
166    }
167
168    // ── 计划形状（atomic_commands 契约） ──
169
170    #[test]
171    fn plan_claim_shape() {
172        let r = registry();
173        let plan = plan_claim(
174            "Outbox",
175            &r,
176            None,
177            &json!({ "status": "pending" }),
178            &json!({ "lockedBy": "w1", "lockedAt": 1 }),
179            None,
180            10,
181            123,
182        )
183        .expect("plan");
184        assert_eq!(plan["mode"], json!(MODE_ATOMIC_COMMANDS));
185        assert_eq!(plan["schema"], json!("Outbox"));
186        let cmds = plan["commands"].as_array().unwrap();
187        assert_eq!(cmds.len(), 2);
188        // 命令一：claim（定位四元组 + filter + 缺省 sort + limit + update + nowStamp）
189        assert_eq!(cmds[0]["kind"], json!("claim"));
190        assert_eq!(cmds[0]["collection"], json!("outbox"));
191        assert_eq!(cmds[0]["sort"], json!([["_id", 1]]), "缺省排序 FIFO");
192        assert_eq!(cmds[0]["limit"], json!(10));
193        assert_eq!(cmds[0]["nowStamp"], json!(123));
194        assert!(cmds[0]["source"].is_string());
195        // 命令二：find 取回被领行（占位符整值替换，契约同 {{phase1.ids}}）
196        assert_eq!(cmds[1]["kind"], json!("find"));
197        assert_eq!(
198            cmds[1]["filter"]["_id"]["$in"],
199            json!(CLAIMED_IDS),
200            "占位符逐字 ___claimed_ids___"
201        );
202        assert_eq!(cmds[1]["placeholder"], json!(CLAIMED_IDS));
203    }
204
205    #[test]
206    fn plan_cas_shape() {
207        let r = registry();
208        let plan = plan_compare_and_set(
209            "Outbox",
210            &r,
211            None,
212            &json!({ "_id": "l1", "lockedBy": "w1" }),
213            &json!({ "lockedAt": 2 }),
214            456,
215        )
216        .expect("plan");
217        assert_eq!(plan["mode"], json!(MODE_ATOMIC_COMMANDS));
218        let cmds = plan["commands"].as_array().unwrap();
219        assert_eq!(cmds.len(), 1, "CAS 单命令也走 atomic_commands");
220        assert_eq!(cmds[0]["kind"], json!("compareAndSet"));
221        assert_eq!(cmds[0]["nowStamp"], json!(456));
222    }
223
224    // ── 错误面（步骤 5：文案前缀断言） ──
225
226    #[test]
227    fn plan_claim_unknown_schema_errs() {
228        let r = registry();
229        let err = plan_claim(
230            "NoSuch",
231            &r,
232            None,
233            &json!({}),
234            &json!({ "lockedBy": "w1" }),
235            None,
236            1,
237            0,
238        )
239        .unwrap_err();
240        assert!(!err.is_empty());
241    }
242
243    #[test]
244    fn plan_claim_nonpositive_limit_errs() {
245        let r = registry();
246        for limit in [0i64, -3] {
247            let err = plan_claim(
248                "Outbox",
249                &r,
250                None,
251                &json!({}),
252                &json!({ "lockedBy": "w1" }),
253                None,
254                limit,
255                0,
256            )
257            .unwrap_err();
258            assert!(err.contains("limit 必须 > 0"), "{err}");
259        }
260    }
261
262    #[test]
263    fn plan_claim_empty_update_errs() {
264        let r = registry();
265        let err = plan_claim(
266            "Outbox",
267            &r,
268            None,
269            &json!({}),
270            &json!({}),
271            None,
272            1,
273            0,
274        )
275        .unwrap_err();
276        assert!(err.contains("非空对象"), "{err}");
277    }
278
279    #[test]
280    fn plan_claim_undeclared_update_field_errs() {
281        let r = registry();
282        let err = plan_claim(
283            "Outbox",
284            &r,
285            None,
286            &json!({}),
287            &json!({ "noSuchField": 1 }),
288            None,
289            1,
290            0,
291        )
292        .unwrap_err();
293        assert!(err.contains("未在 schema 声明"), "{err}");
294    }
295
296    #[test]
297    fn plan_cas_undeclared_update_field_errs() {
298        let r = registry();
299        let err = plan_compare_and_set(
300            "Outbox",
301            &r,
302            None,
303            &json!({}),
304            &json!({ "noSuchField": 1 }),
305            0,
306        )
307        .unwrap_err();
308        assert!(err.contains("未在 schema 声明"), "{err}");
309    }
310}
rust_store_core/command/claim.rs

rust_store_core/command/
claim.rs
