From 9fd4e1ff99c552959978a39d1b853539fd1a4e9b Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 26 Jul 2026 16:44:53 +0300 Subject: [PATCH] Harden 1C saved-state writes and cache consistency --- plugins/1c/connector/adapter_1c_server.py | 351 +++++++++++++++++++++- tests/1c/test_object_property_write.py | 35 ++- tests/1c/test_payload_codec.py | 70 ++++- 3 files changed, 440 insertions(+), 16 deletions(-) diff --git a/plugins/1c/connector/adapter_1c_server.py b/plugins/1c/connector/adapter_1c_server.py index 94eda0d..aa77e77 100644 --- a/plugins/1c/connector/adapter_1c_server.py +++ b/plugins/1c/connector/adapter_1c_server.py @@ -15865,6 +15865,83 @@ def metadata_cache_invalidate(payload: dict[str, Any]) -> dict[str, Any]: return {"schema": "onec_metadata_cache_invalidate.v1", "status": "ok", "dry_run": bool(dry_run), "counts": {"bases": len(base_ids), "deleted": deleted}} +def invalidate_adapter_caches_after_saved_state_change(base_id: str, *, reason: str) -> dict[str, Any]: + """Invalidate persistent and process-local views after a committed saved-state mutation.""" + + normalized_base_id = str(base_id or "").strip() + persistent: dict[str, Any] + try: + persistent = metadata_cache_invalidate({"base_id": normalized_base_id, "dry_run": False}) + except Exception as exc: + persistent = {"status": "error", "diagnostics": {"message": str(exc)}} + + runtime_counts = { + "base_root_metadata": 0, + "data_schema": 0, + "extension_manifests": 0, + } + with BASE_ROOT_METADATA_CACHE_LOCK: + root_keys = [ + key + for key in BASE_ROOT_METADATA_CACHE + if isinstance(key, tuple) and key and str(key[0]).strip() == normalized_base_id + ] + for key in root_keys: + BASE_ROOT_METADATA_CACHE.pop(key, None) + runtime_counts["base_root_metadata"] = len(root_keys) + + normalized_casefold = normalized_base_id.casefold() + with DATA_SCHEMA_CACHE_LOCK: + schema_keys = [] + for key in DATA_SCHEMA_CACHE: + try: + cached_selector = json.loads(key) + except (TypeError, ValueError): + continue + if str(cached_selector.get("base_id") or "").casefold() == normalized_casefold: + schema_keys.append(key) + for key in schema_keys: + DATA_SCHEMA_CACHE.pop(key, None) + runtime_counts["data_schema"] = len(schema_keys) + + manifest_cache = globals().get("EXTENSION_MANIFEST_CACHE") + manifest_lock = globals().get("EXTENSION_MANIFEST_CACHE_LOCK") + if isinstance(manifest_cache, dict): + def clear_manifest_entries() -> int: + manifest_keys = [ + key + for key in manifest_cache + if isinstance(key, tuple) and key and str(key[0]).strip() == normalized_base_id + ] + for key in manifest_keys: + manifest_cache.pop(key, None) + return len(manifest_keys) + + if manifest_lock is not None: + with manifest_lock: + runtime_counts["extension_manifests"] = clear_manifest_entries() + else: + runtime_counts["extension_manifests"] = clear_manifest_entries() + + persistent_status = str(persistent.get("status") or "error") + return { + "schema": "onec_saved_state_cache_invalidation.v1", + "status": "ok" if persistent_status == "ok" else "partial", + "base_id": normalized_base_id, + "reason": reason, + "persistent": { + "status": persistent_status, + "deleted": int(((persistent.get("counts") or {}).get("deleted") or 0)), + }, + "runtime": runtime_counts, + **( + {"diagnostics": persistent.get("diagnostics")} + if persistent_status != "ok" and persistent.get("diagnostics") + else {} + ), + } + + def metadata_module_owner_cache_prune(payload: dict[str, Any]) -> dict[str, Any]: method = "metadata.module_owner_cache.prune" payload = normalize_object_selector_aliases(payload, method) @@ -17455,7 +17532,10 @@ def apply_saved_state_prepare_copy( try: cursor = conn.cursor(as_dict=True) placeholders = ",".join(["%s"] * len(names)) - cursor.execute(f"SELECT FileName, PartNo FROM dbo.[{target_table}] WHERE FileName IN ({placeholders})", tuple(names)) + cursor.execute( + f"SELECT FileName, PartNo FROM dbo.[{target_table}] WITH (UPDLOCK, HOLDLOCK) WHERE FileName IN ({placeholders})", + tuple(names), + ) collisions = [{key: jsonable(value) for key, value in row.items()} for row in cursor.fetchall()] if collisions: conn.rollback() @@ -17521,6 +17601,10 @@ def apply_saved_state_prepare_copy( "source": {"kind": "live_sql", "database": config["database"], "table": source_table}, "target": {"table": target_table}, "counts": {"inserted_rows": inserted, "file_names": len(names)}, + "cache_invalidation": invalidate_adapter_caches_after_saved_state_change( + base_id, + reason="saved_state_prepare", + ), "duration_ms": int((time.time() - started) * 1000), } @@ -19225,7 +19309,7 @@ def apply_storage_file_bytes_single_part( try: cursor = conn.cursor(as_dict=True) cursor.execute( - f"SELECT PartNo, BinaryData FROM dbo.[{table}] WHERE FileName = %s ORDER BY PartNo", + f"SELECT PartNo, BinaryData FROM dbo.[{table}] WITH (UPDLOCK, HOLDLOCK) WHERE FileName = %s ORDER BY PartNo", (file_name,), ) rows = cursor.fetchall() @@ -19337,6 +19421,10 @@ def apply_storage_file_bytes_single_part( if verified and semantic.get("status") not in {"ok", "skipped"}: result["status"] = "semantic_verification_failed" result["applied"] = False + result["cache_invalidation"] = invalidate_adapter_caches_after_saved_state_change( + base_id, + reason="saved_state_payload_apply", + ) return result @@ -28174,6 +28262,7 @@ def extension_source_matches(source: dict[str, Any], extension_guid: str | None) EXTENSION_MANIFEST_CACHE: dict[tuple[str, str], dict[str, Any]] = {} +EXTENSION_MANIFEST_CACHE_LOCK = threading.Lock() def extension_root_key_from_zipped_info(data: bytes) -> str: @@ -28320,7 +28409,8 @@ def live_extension_manifests(base_id: str, *, extension_guid: str | None = None, diagnostics.append({"extension": row.get("name"), "status": "missing_root_cas_key"}) continue cache_key = (base_id, root_key) - cached = EXTENSION_MANIFEST_CACHE.get(cache_key) + with EXTENSION_MANIFEST_CACHE_LOCK: + cached = EXTENSION_MANIFEST_CACHE.get(cache_key) if cached: manifests.append(cached) continue @@ -28333,7 +28423,8 @@ def live_extension_manifests(base_id: str, *, extension_guid: str | None = None, except Exception as exc: diagnostics.append({"extension": row.get("name"), "root_cas_key": root_key, "status": "parse_error", "diagnostics": {"message": str(exc)}}) continue - EXTENSION_MANIFEST_CACHE[cache_key] = manifest + with EXTENSION_MANIFEST_CACHE_LOCK: + EXTENSION_MANIFEST_CACHE[cache_key] = manifest manifests.append(manifest) return manifests, diagnostics @@ -36951,6 +37042,186 @@ def deterministic_member_guid(parent_guid: str, member_kind: str, member_name: s return str(uuid.uuid5(uuid.UUID(str(parent_guid)), seed)).lower() +def config_tree_scalar_occurrences(tree: Any) -> list[dict[str, str]]: + occurrences: list[dict[str, str]] = [] + + def walk(node: Any, path: tuple[int, ...]) -> None: + if isinstance(node, dict) and node.get("type") in {"atom", "string"}: + occurrences.append( + { + "path": ".".join(str(part) for part in path), + "value": str(node.get("value") or ""), + } + ) + return + for index, child in enumerate(config_tree_list_items(node)): + walk(child, (*path, index)) + + walk(tree, ()) + return occurrences + + +def config_tree_set_scalar(tree: Any, path: tuple[int, ...], value: str) -> bool: + target = config_tree_item_at_path(tree, path) + if not isinstance(target, dict) or target.get("type") not in {"atom", "string"}: + return False + target["value"] = str(value) + return True + + +def metadata_member_record_identity_layout(tree: Any, guid: str) -> dict[str, Any]: + identity = config_tree_identity_records(tree).get(str(guid or "").strip().lower()) + if not identity: + return {"status": "not_found", "error": "member_identity_not_found"} + try: + marker_path = tuple( + int(part) + for part in str(identity.get("evidence_path") or "").split(".") + if part != "" + ) + except ValueError: + return {"status": "unsupported", "error": "invalid_identity_evidence_path"} + if not marker_path: + return {"status": "unsupported", "error": "invalid_identity_evidence_path"} + parent_path = marker_path[:-1] + marker_index = marker_path[-1] + marker = config_tree_list_items(config_tree_item_at_path(tree, marker_path)) + siblings = config_tree_list_items(config_tree_item_at_path(tree, parent_path)) + if len(marker) != 3 or marker_index + 3 >= len(siblings): + return {"status": "unsupported", "error": "member_identity_layout_unsupported"} + guid_path = (*marker_path, 2) + name_path = (*parent_path, marker_index + 1) + synonym_container_path = (*parent_path, marker_index + 2) + comment_path = (*parent_path, marker_index + 3) + synonym_items = config_tree_list_items(config_tree_item_at_path(tree, synonym_container_path)) + synonym_paths: dict[str, tuple[int, ...]] = {} + for index in range(1, len(synonym_items) - 1, 2): + language = config_tree_scalar(synonym_items[index]) + value_node = synonym_items[index + 1] + if language and isinstance(value_node, dict) and value_node.get("type") in {"atom", "string"}: + synonym_paths[language] = (*synonym_container_path, index + 1) + return { + "status": "ok", + "identity": identity, + "guid_path": guid_path, + "name_path": name_path, + "synonym_paths": synonym_paths, + "comment_path": comment_path, + "identity_paths": { + ".".join(str(part) for part in guid_path), + ".".join(str(part) for part in name_path), + ".".join(str(part) for part in comment_path), + *( + ".".join(str(part) for part in path) + for path in synonym_paths.values() + ), + }, + } + + +def metadata_member_record_shape_sha1(tree: Any, guid: str) -> str | None: + from parser.payload import serialize_brace_tree + + normalized = clone_form_structural_node(tree, {}) + layout = metadata_member_record_identity_layout(normalized, guid) + if layout.get("status") != "ok": + return None + if not config_tree_set_scalar(normalized, layout["guid_path"], ""): + return None + if not config_tree_set_scalar(normalized, layout["name_path"], ""): + return None + if not config_tree_set_scalar(normalized, layout["comment_path"], ""): + return None + for language, path in (layout.get("synonym_paths") or {}).items(): + if not config_tree_set_scalar(normalized, path, f""): + return None + return hashlib.sha1(serialize_brace_tree(normalized).encode("utf-8")).hexdigest() + + +def clone_metadata_member_record( + tree: Any, + *, + template_guid: str, + new_guid: str, + new_name: str, + new_synonym: str, + new_comment: str, +) -> tuple[Any | None, dict[str, Any]]: + """Clone a declared member while changing only its explicit identity scalars.""" + + cloned = clone_form_structural_node(tree, {}) + layout = metadata_member_record_identity_layout(cloned, template_guid) + if layout.get("status") != "ok": + return None, { + "status": "blocked", + "error": layout.get("error") or "template_identity_layout_unsupported", + } + identity = layout.get("identity") if isinstance(layout.get("identity"), dict) else {} + template_name = str(identity.get("name") or "") + identity_paths = set(layout.get("identity_paths") or set()) + stale_candidates = { + str(template_guid or "").strip().lower(), + template_name, + } + stale_references = [ + occurrence + for occurrence in config_tree_scalar_occurrences(cloned) + if occurrence["path"] not in identity_paths + and occurrence["value"] in stale_candidates + ] + if stale_references: + return None, { + "status": "blocked", + "error": "template_identity_referenced_outside_identity_fields", + "stale_references": stale_references[:20], + } + + changes_ok = [ + config_tree_set_scalar(cloned, layout["guid_path"], new_guid), + config_tree_set_scalar(cloned, layout["name_path"], new_name), + config_tree_set_scalar(cloned, layout["comment_path"], new_comment), + ] + synonym_values: dict[str, str] = {} + for language, path in (layout.get("synonym_paths") or {}).items(): + value = new_synonym if language == "ru" else new_name + synonym_values[language] = value + changes_ok.append(config_tree_set_scalar(cloned, path, value)) + if not all(changes_ok): + return None, { + "status": "blocked", + "error": "template_identity_scalar_update_failed", + } + + cloned_identities = config_tree_identity_records(cloned) + new_identity = cloned_identities.get(new_guid) + if not new_identity or template_guid in cloned_identities: + return None, { + "status": "blocked", + "error": "cloned_identity_verification_failed", + "identities": sorted(cloned_identities), + } + template_shape_sha1 = metadata_member_record_shape_sha1(tree, template_guid) + cloned_shape_sha1 = metadata_member_record_shape_sha1(cloned, new_guid) + shape_preserved = bool( + template_shape_sha1 + and cloned_shape_sha1 + and template_shape_sha1 == cloned_shape_sha1 + ) + return ( + cloned if shape_preserved else None, + { + "status": "ok" if shape_preserved else "blocked", + "error": None if shape_preserved else "member_settings_shape_changed", + "identity_fields_changed": ["guid", "name", "synonyms", "comment"], + "synonyms": synonym_values, + "template_shape_sha1": template_shape_sha1, + "cloned_shape_sha1": cloned_shape_sha1, + "settings_preserved": shape_preserved, + "stale_references": [], + }, + ) + + def nested_metadata_guid_references( base_id: str, guids: Iterable[str], @@ -42714,14 +42985,33 @@ def metadata_object_member_add(payload: dict[str, Any]) -> dict[str, Any]: resolved = candidates[0] template_member = resolved["template"] - replacements = { - str(template_member.get("guid") or ""): new_guid, - str(template_member.get("name") or ""): new_name, - str(resolved["synonym"].get("current") or ""): new_synonym, - } - if "new_member_comment" in payload: - replacements[str(resolved["comment"].get("current") or "")] = str(payload.get("new_member_comment") or "") - cloned_node = clone_form_structural_node(resolved["record"]["node"], replacements) + new_comment = str(payload.get("new_member_comment") or "") + cloned_node, clone_validation = clone_metadata_member_record( + resolved["record"]["node"], + template_guid=str(template_member.get("guid") or "").lower(), + new_guid=new_guid, + new_name=new_name, + new_synonym=new_synonym, + new_comment=new_comment, + ) + if cloned_node is None or clone_validation.get("status") != "ok": + return { + "schema": "onec_metadata_object_member_add.v1", + "method": method, + "status": "blocked", + "error": clone_validation.get("error") or "unsafe_attribute_template", + "base_id": base_id, + "container": {"ref": container_ref, "scope": container_scope}, + "template": { + key: template_member.get(key) + for key in ("kind", "name", "ref") + if template_member.get(key) is not None + }, + "clone_validation": clone_validation, + "diagnostics": { + "message": "The template record must preserve every non-identity setting and must not reference its old identity outside declared identity fields.", + }, + } proposal = changes_propose( { "base_id": base_id, @@ -42751,7 +43041,15 @@ def metadata_object_member_add(payload: dict[str, Any]) -> dict[str, Any]: }, "container": {"ref": container_ref, "scope": container_scope}, "template": {key: template_member.get(key) for key in ("kind", "name", "ref") if template_member.get(key) is not None}, - "requested_member": {"kind": "Attribute", "name": new_name, "synonym": new_synonym, "guid": new_guid, "ref": requested_ref}, + "requested_member": { + "kind": "Attribute", + "name": new_name, + "synonym": new_synonym, + "comment": new_comment, + "guid": new_guid, + "ref": requested_ref, + }, + "clone_validation": clone_validation, "proposal": proposal, "write_mode": {"target": "saved_state", "active_configuration_write": False, "sql_write_performed": False}, } @@ -42790,17 +43088,42 @@ def metadata_object_member_add(payload: dict[str, Any]) -> dict[str, Any]: readback_tree = parse_config_tree_from_bytes(readback_data or b"") if not readback_error else None identity = config_tree_identity_records(readback_tree).get(new_guid) if readback_tree is not None else None readback_record = config_tree_declared_record_for_guid(readback_tree, new_guid) if readback_tree is not None else {"status": "error"} + readback_comment = ( + config_tree_identity_property_target(readback_tree, new_guid, "comment") + if readback_tree is not None + else {"status": "error"} + ) + readback_shape_sha1 = ( + metadata_member_record_shape_sha1(readback_record.get("node"), new_guid) + if readback_record.get("status") == "ok" + else None + ) + settings_preserved = bool( + readback_shape_sha1 + and readback_shape_sha1 == clone_validation.get("cloned_shape_sha1") + ) verified = bool( identity and identity.get("name") == new_name and (identity.get("synonyms") or {}).get("ru") == new_synonym + and readback_comment.get("status") == "ok" + and readback_comment.get("current") == new_comment and readback_record.get("status") == "ok" and readback_record.get("parent_path") == resolved["record"].get("parent_path") + and settings_preserved ) result["semantic_verification"] = { "status": "ok" if verified else "mismatch", - "member": {"kind": "Attribute", "name": (identity or {}).get("name"), "synonym": ((identity or {}).get("synonyms") or {}).get("ru")}, + "member": { + "kind": "Attribute", + "name": (identity or {}).get("name"), + "synonym": ((identity or {}).get("synonyms") or {}).get("ru"), + "comment": readback_comment.get("current"), + }, "container_match": readback_record.get("parent_path") == resolved["record"].get("parent_path"), + "settings_preserved": settings_preserved, + "expected_shape_sha1": clone_validation.get("cloned_shape_sha1"), + "actual_shape_sha1": readback_shape_sha1, } if mode == "apply_and_verify": result["status"] = "verified" if verified else "verification_failed" diff --git a/tests/1c/test_object_property_write.py b/tests/1c/test_object_property_write.py index b7d6822..e5e2508 100644 --- a/tests/1c/test_object_property_write.py +++ b/tests/1c/test_object_property_write.py @@ -75,7 +75,7 @@ def declared_attribute_tree(*, include_new: bool = False) -> dict[str, Any]: seq(atom(1), atom(0), atom(new_guid)), string("КодПоставщика"), seq(atom(1), string("ru"), string("Код поставщика")), - string("Комментарий реквизита"), + string(""), atom("type-settings"), ) ) @@ -437,9 +437,42 @@ def test_object_member_add_plan_clones_attribute_and_generates_guid(monkeypatch) assert result["requested_member"]["guid"] == new_guid assert result["requested_member"]["ref"] == "Catalog.Номенклатура.Attribute.КодПоставщика" assert result["container"] == {"ref": "Catalog.Номенклатура", "scope": "object"} + assert result["requested_member"]["comment"] == "" + assert result["clone_validation"]["settings_preserved"] is True + assert result["clone_validation"]["stale_references"] == [] assert append["parent_path"] == "4" assert cloned_identities[new_guid]["name"] == "КодПоставщика" assert cloned_identities[new_guid]["synonyms"]["ru"] == "Код поставщика" + cloned_comment = adapter.config_tree_identity_property_target( + append["node"], + new_guid, + "comment", + ) + assert cloned_comment["current"] == "" + + +def test_object_member_clone_blocks_template_identity_references_outside_identity_fields() -> None: + template = seq( + seq(atom(1), atom(0), atom("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")), + string("Артикул"), + seq(atom(1), string("ru"), string("Артикул товара")), + string("Комментарий реквизита"), + seq(atom("type-settings"), string("Артикул")), + ) + + cloned, validation = adapter.clone_metadata_member_record( + template, + template_guid="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + new_guid="bbbbbbbb-cccc-dddd-eeee-ffffffffffff", + new_name="КодПоставщика", + new_synonym="Код поставщика", + new_comment="", + ) + + assert cloned is None + assert validation["status"] == "blocked" + assert validation["error"] == "template_identity_referenced_outside_identity_fields" + assert validation["stale_references"] == [{"path": "4.1", "value": "Артикул"}] def test_object_member_add_scopes_tabular_column_guid_and_ref(monkeypatch) -> None: diff --git a/tests/1c/test_payload_codec.py b/tests/1c/test_payload_codec.py index ffa95dd..649b99a 100644 --- a/tests/1c/test_payload_codec.py +++ b/tests/1c/test_payload_codec.py @@ -26043,7 +26043,7 @@ def test_saved_state_apply_blocks_unsafe_form_payload_rewrite() -> None: def test_saved_state_apply_updates_single_part_with_backup(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: original = b"old-payload" replacement = b"new-payload" - state = {"data": original, "committed": False, "rolled_back": False} + state = {"data": original, "committed": False, "rolled_back": False, "sql": []} class FakeCursor: rowcount = 0 @@ -26052,6 +26052,7 @@ def test_saved_state_apply_updates_single_part_with_backup(monkeypatch: pytest.M self.store = store def execute(self, sql: str, params: tuple[Any, ...]) -> None: + self.store["sql"].append(sql) if sql.strip().upper().startswith("SELECT"): self.rowcount = 1 return @@ -26088,6 +26089,11 @@ def test_saved_state_apply_updates_single_part_with_backup(monkeypatch: pytest.M "read_storage_file_bytes", lambda *args, **kwargs: (state["data"], {"database": "upo_test"}, None), ) + monkeypatch.setattr( + adapter_server, + "invalidate_adapter_caches_after_saved_state_change", + lambda base_id, *, reason: {"status": "ok", "base_id": base_id, "reason": reason}, + ) proposal = { "schema": "onec_change_proposal.v1", @@ -26111,11 +26117,73 @@ def test_saved_state_apply_updates_single_part_with_backup(monkeypatch: pytest.M assert state["data"] == replacement assert state["committed"] is True assert state["rolled_back"] is False + assert "WITH (UPDLOCK, HOLDLOCK)" in state["sql"][0] + assert result["cache_invalidation"]["reason"] == "saved_state_payload_apply" assert Path(result["backup"]["path"]).exists() backup = json.loads(Path(result["backup"]["path"]).read_text(encoding="utf-8")) assert backup["original"]["payload_hex"] == original.hex() +def test_saved_state_cache_invalidation_clears_persistent_and_runtime_views(monkeypatch: pytest.MonkeyPatch) -> None: + base_id = "upo_test" + other_base_id = "other" + data_key = adapter_server.data_schema_cache_key( + {"base_id": base_id, "ref": "Catalog.Номенклатура"} + ) + other_data_key = adapter_server.data_schema_cache_key( + {"base_id": other_base_id, "ref": "Catalog.Номенклатура"} + ) + monkeypatch.setattr( + adapter_server, + "metadata_cache_invalidate", + lambda payload: { + "status": "ok", + "counts": {"deleted": 7}, + "base_id": payload["base_id"], + }, + ) + monkeypatch.setattr( + adapter_server, + "BASE_ROOT_METADATA_CACHE", + { + (base_id, "Config"): {"cached_at": 1}, + (other_base_id, "Config"): {"cached_at": 1}, + }, + ) + monkeypatch.setattr( + adapter_server, + "DATA_SCHEMA_CACHE", + { + data_key: {"cached_at": 1}, + other_data_key: {"cached_at": 1}, + }, + ) + monkeypatch.setattr( + adapter_server, + "EXTENSION_MANIFEST_CACHE", + { + (base_id, "root-a"): {"status": "ok"}, + (other_base_id, "root-b"): {"status": "ok"}, + }, + ) + + result = adapter_server.invalidate_adapter_caches_after_saved_state_change( + base_id, + reason="test", + ) + + assert result["status"] == "ok" + assert result["persistent"]["deleted"] == 7 + assert result["runtime"] == { + "base_root_metadata": 1, + "data_schema": 1, + "extension_manifests": 1, + } + assert list(adapter_server.BASE_ROOT_METADATA_CACHE) == [(other_base_id, "Config")] + assert list(adapter_server.DATA_SCHEMA_CACHE) == [other_data_key] + assert list(adapter_server.EXTENSION_MANIFEST_CACHE) == [(other_base_id, "root-b")] + + def test_saved_state_apply_runs_form_element_semantic_verification(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: original = b"old-payload" replacement = b"new-payload"