You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/21 07:43:06 UTC
(maka) branch rust-preview updated: fix(ui): persist complete drafts with actual read admission
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new 4ccaa3afe6 fix(ui): persist complete drafts with actual read admission
4ccaa3afe6 is described below
commit 4ccaa3afe6f96d848ecd8ecfabae4722519757bd
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Oct 21 15:42:56 2026 +0800
fix(ui): persist complete drafts with actual read admission
Remove duplicate checkpoint content and namespace quotas. Reserve caller-owned content before reading it from the same SQLite snapshot, and migrate editor and preference consumers while preserving original CAS and cleanup semantics.
Generated-by: Codex
---
platform/event-log/README.md | 2 +
platform/event-log/README.zh-CN.md | 2 +
platform/event-log/src/applications/checkpoints.rs | 48 ++--
.../src/applications/checkpoints/tests.rs | 303 +++++++++++++++++----
platform/ui-runtime/README.md | 2 +-
platform/ui-runtime/README.zh-CN.md | 2 +-
platform/ui-runtime/src/checkpoint.rs | 12 +-
platform/ui-runtime/src/checkpoint/tests.rs | 156 ++++++++++-
platform/ui-runtime/src/checkpoint/value.rs | 38 ++-
platform/ui-runtime/src/checkpoint/worker.rs | 19 +-
10 files changed, 477 insertions(+), 107 deletions(-)
diff --git a/platform/event-log/README.md b/platform/event-log/README.md
index 2f99bd1b40..32bb3e5926 100644
--- a/platform/event-log/README.md
+++ b/platform/event-log/README.md
@@ -25,6 +25,8 @@ Owns the SQLite event store and State Root lease. It persists application source
Saved application inspection uses `RootOwner::open_existing` and `EventLog::for_application_inspection`. It requires existing private coordination state and the original exclusive Root/database leases, reads committed application facts including retained WAL content, and closes without changing stored members or bytes. It performs no migration or derived initialization; missing state fails without repair. Only application queries on the owned connection belong to this opening. CLI applic [...]
+Native UI checkpoints retain complete content under the original persisted deployment and instance, without separate content or accumulated-namespace byte quotas. Reads obtain the content byte length and caller-owned admission before materializing the same record within one SQLite snapshot; the reservation travels with the returned record. Writes preserve source/format identity, revision comparison and empty CAS tombstones. Live memory admission belongs to the caller; filesystem and comm [...]
+
Original operation contracts are stored once by digest. Operation requests and receipt references carry that identity, so journal reads, caller history and UI delivery do not repeatedly copy service schemas. Reads verify the exact original request fingerprint; admission still validates the complete contract and resource requirements.
The operation journal retains complete admitted requests and returned values without separate request, result or receipt-reference size quotas. Exact retries and reopened readers recover the original outcome and cleanup status.
diff --git a/platform/event-log/README.zh-CN.md b/platform/event-log/README.zh-CN.md
index 2102e922ce..1dfaa43edc 100644
--- a/platform/event-log/README.zh-CN.md
+++ b/platform/event-log/README.zh-CN.md
@@ -25,6 +25,8 @@
已保存应用的检查使用 `RootOwner::open_existing` 和 `EventLog::for_application_inspection`，要求已有的私有协调状态，并持有原有的独占 Root／数据库租约。检查读取已提交的应用事实，包括保留在 WAL 中的内容，关闭后不改变存储成员或字节；不执行迁移或派生初始化，缺失状态直接报错，不做修复。该打开方式仅供原连接上的应用查询使用。CLI 的 application read 与默认更新预览共用此入口；实际应用更新时重新打开普通写入所有者，并重新计算变更。
+原生 UI 检查点在原始持久化部署和实例下保存完整正文，不设置独立的正文或命名空间累计字节配额。读取在同一 SQLite 快照内先取得正文字节长度和调用方准入，再加载该记录；预留资源随记录一起返回。写入保留来源／格式身份、修订号比较和空内容 CAS 墓碑。实时内存准入由调用方负责，文件系统与提交失败仍明确报告为存储失败。
+
原始操作契约按摘要只存储一次。操作请求和回执引用只携带该身份，日志读取、调用方历史与 UI 传递不再逐次复制服务 schema。读取仍核对原始请求指纹，执行准入仍校验完整契约和资源要求。
操作日志完整保存已受理的请求和返回值，不另设请求、结果或回执引用的大小额度。精确重试和重开后的读取恢复原始结果与清理状态。
diff --git a/platform/event-log/src/applications/checkpoints.rs b/platform/event-log/src/applications/checkpoints.rs
index 6d0fe3ebdb..6a2a3846b6 100644
--- a/platform/event-log/src/applications/checkpoints.rs
+++ b/platform/event-log/src/applications/checkpoints.rs
@@ -28,10 +28,8 @@ use serde_json::Value;
use sqlx::{Connection, Row};
use std::sync::Arc;
-pub const MAX_CONTENT: usize = 1024 * 1024;
pub const MAX_METADATA: usize = 32 * 1024;
pub const MAX_SOURCE: usize = 4096;
-const MAX_NAMESPACE: usize = 40 * 1024 * 1024;
const MAX_REVISION: u64 = (1 << 53) - 1;
#[derive(Clone, Debug, PartialEq)]
@@ -86,29 +84,44 @@ impl EventLog {
}
}
impl Checkpoints {
- pub async fn read(&self, key: &str) -> Result<Option<Checkpoint>, StoreError> {
+ /// Admit the original content allocation before fetching it. The callback
+ /// and payload read share one SQLite snapshot; its token follows the result.
+ pub async fn read<T: Send + 'static>(
+ &self,
+ key: &str,
+ admit: impl FnOnce(usize) -> Result<T, StoreError> + Send + 'static,
+ ) -> Result<Option<(Checkpoint, T)>, StoreError> {
key_valid(key)?;
self.store.validate_root()?;
let deployment = self.deployment.clone();
let instance = self.instance.clone();
let key = key.to_owned();
self.store.connection.run(move |connection| Box::pin(async move {
+ let mut tx = connection.begin().await?;
+ let bytes: Option<i64> = sqlx::query_scalar("SELECT length(CAST(content AS BLOB)) FROM application_ui_checkpoints WHERE deployment=? AND instance=? AND key=?")
+ .bind(&deployment).bind(&instance).bind(&key).fetch_optional(&mut *tx).await?;
+ let Some(bytes) = bytes else {
+ tx.commit().await?;
+ return Ok(None);
+ };
+ let bytes = usize::try_from(bytes).map_err(|_| invalid("invalid UI checkpoint content length"))?;
+ let token = admit(bytes)?;
let row = sqlx::query("SELECT revision,
CASE WHEN length(CAST(format AS BLOB)) BETWEEN 1 AND 128 THEN format END AS format,
CASE WHEN length(CAST(data_source_json AS BLOB)) <= 4096 THEN data_source_json END AS source,
CASE WHEN length(CAST(metadata_json AS BLOB)) <= 32768 THEN metadata_json END AS metadata,
- CASE WHEN length(content) <= 1048576 THEN content END AS content
+ content
FROM application_ui_checkpoints WHERE deployment=? AND instance=? AND key=?")
- .bind(deployment).bind(instance).bind(key).fetch_optional(connection).await?;
- row.map(|row| {
- let revision: i64 = row.try_get("revision")?;
- if !(1..=MAX_REVISION as i64).contains(&revision) { return Err(invalid("invalid UI checkpoint revision")); }
- let format = row.try_get::<Option<String>, _>("format")?.ok_or_else(|| invalid("invalid UI checkpoint format"))?;
- let data_source = serde_json::from_str(&row.try_get::<Option<String>, _>("source")?.ok_or_else(|| invalid("UI checkpoint source exceeds budget"))?)?;
- let metadata = serde_json::from_str(&row.try_get::<Option<String>, _>("metadata")?.ok_or_else(|| invalid("UI checkpoint metadata exceeds budget"))?)?;
- let content = row.try_get::<Option<Vec<u8>>, _>("content")?.ok_or_else(|| invalid("UI checkpoint content exceeds budget"))?;
- Ok(Checkpoint { revision: revision as u64, format, data_source, metadata, content })
- }).transpose()
+ .bind(deployment).bind(instance).bind(key).fetch_one(&mut *tx).await?;
+ let revision: i64 = row.try_get("revision")?;
+ if !(1..=MAX_REVISION as i64).contains(&revision) { return Err(invalid("invalid UI checkpoint revision")); }
+ let format = row.try_get::<Option<String>, _>("format")?.ok_or_else(|| invalid("invalid UI checkpoint format"))?;
+ let data_source = serde_json::from_str(&row.try_get::<Option<String>, _>("source")?.ok_or_else(|| invalid("UI checkpoint source exceeds budget"))?)?;
+ let metadata = serde_json::from_str(&row.try_get::<Option<String>, _>("metadata")?.ok_or_else(|| invalid("UI checkpoint metadata exceeds budget"))?)?;
+ let content: Vec<u8> = row.try_get("content")?;
+ if content.len() != bytes { return Err(invalid("UI checkpoint content length changed")); }
+ tx.commit().await?;
+ Ok(Some((Checkpoint { revision: revision as u64, format, data_source, metadata, content }, token)))
})).await
}
@@ -127,7 +140,6 @@ impl Checkpoints {
self.store.validate_root()?;
if format.is_empty()
|| format.len() > 128
- || content.len() > MAX_CONTENT
|| expected.is_some_and(|value| value == 0 || value >= MAX_REVISION)
{
return Err(invalid("invalid or oversized UI checkpoint"));
@@ -138,7 +150,6 @@ impl Checkpoints {
let instance = self.instance.clone();
let key = key.to_owned();
let format = format.to_owned();
- let size = 256 + key.len() + format.len() + source.len() + metadata.len() + content.len();
self.store.connection.run(move |connection| Box::pin(async move {
let mut tx = connection.begin_with("BEGIN IMMEDIATE").await?;
let current: Option<(i64, Option<String>, Option<String>)> = sqlx::query_as("SELECT revision, CASE WHEN length(CAST(format AS BLOB)) BETWEEN 1 AND 128 THEN format END, CASE WHEN length(CAST(data_source_json AS BLOB)) <= 4096 THEN data_source_json END FROM application_ui_checkpoints WHERE deployment=? AND instance=? AND key=?")
@@ -151,11 +162,6 @@ impl Checkpoints {
if current.is_some_and(|(_, previous_format, previous_source)| previous_format.as_deref() != Some(format.as_str()) || previous_source.as_deref() != Some(source.as_str())) {
return Err(invalid("UI checkpoint format or data source changed"));
}
- let occupied: i64 = sqlx::query_scalar("SELECT COALESCE(SUM(256 + length(CAST(key AS BLOB)) + length(CAST(format AS BLOB)) + length(CAST(data_source_json AS BLOB)) + length(CAST(metadata_json AS BLOB)) + length(content)), 0) FROM application_ui_checkpoints WHERE deployment=? AND instance=? AND key<>?")
- .bind(&deployment).bind(&instance).bind(&key).fetch_one(&mut *tx).await?;
- if occupied < 0 || size > MAX_NAMESPACE.saturating_sub(occupied as usize) {
- return Err(invalid("UI checkpoint namespace exceeds 40 MiB"));
- }
let revision = actual.unwrap_or(0) + 1;
sqlx::query("INSERT INTO application_ui_checkpoints (deployment,instance,key,revision,format,data_source_json,metadata_json,content) VALUES (?,?,?,?,?,?,?,?) ON CONFLICT(deployment,instance,key) DO UPDATE SET revision=excluded.revision,metadata_json=excluded.metadata_json,content=excluded.content")
.bind(deployment).bind(instance).bind(key).bind(revision as i64).bind(format).bind(source).bind(metadata).bind(content).execute(&mut *tx).await?;
diff --git a/platform/event-log/src/applications/checkpoints/tests.rs b/platform/event-log/src/applications/checkpoints/tests.rs
index 7efd8d2907..b2eaa316f5 100644
--- a/platform/event-log/src/applications/checkpoints/tests.rs
+++ b/platform/event-log/src/applications/checkpoints/tests.rs
@@ -92,11 +92,15 @@ async fn original_ui_namespace_survives_source_changes_and_root_reopen_without_c
.await
.unwrap();
assert_eq!(revision, 1);
- let saved = checkpoints.read("conversation").await.unwrap().unwrap();
+ let saved = checkpoints
+ .read("conversation", |_| Ok(()))
+ .await
+ .unwrap()
+ .unwrap();
let updated = source(&store, deployment, 1).await;
let replacement = store.ui_checkpoints(&updated, &instance()).await.unwrap();
assert_eq!(
- replacement.read("conversation").await.unwrap(),
+ replacement.read("conversation", |_| Ok(())).await.unwrap(),
Some(saved.clone())
);
let other = source(&store, Uuid::new_v4(), 0).await;
@@ -105,7 +109,7 @@ async fn original_ui_namespace_survives_source_changes_and_root_reopen_without_c
.ui_checkpoints(&other, &instance())
.await
.unwrap()
- .read("conversation")
+ .read("conversation", |_| Ok(()))
.await
.unwrap()
.is_none()
@@ -130,7 +134,7 @@ async fn original_ui_namespace_survives_source_changes_and_root_reopen_without_c
.is_err()
);
assert_eq!(
- replacement.read("conversation").await.unwrap(),
+ replacement.read("conversation", |_| Ok(())).await.unwrap(),
Some(saved.clone())
);
}
@@ -148,7 +152,10 @@ async fn original_ui_namespace_survives_source_changes_and_root_reopen_without_c
.ui_checkpoints(&updated, &instance())
.await
.unwrap();
- assert_eq!(checkpoints.read("conversation").await.unwrap(), Some(saved));
+ assert_eq!(
+ checkpoints.read("conversation", |_| Ok(())).await.unwrap(),
+ Some(saved)
+ );
drop(checkpoints);
reopened.shutdown().await.unwrap();
}
@@ -183,7 +190,7 @@ async fn concurrent_mounts_cannot_replace_a_saved_or_cleared_original_checkpoint
a.as_ref().err().or(b.as_ref().err()).unwrap(),
StoreError::RevisionConflict { .. }
));
- let current = first.read("draft").await.unwrap().unwrap();
+ let (current, ()) = first.read("draft", |_| Ok(())).await.unwrap().unwrap();
first
.write(
"draft",
@@ -223,25 +230,119 @@ async fn concurrent_mounts_cannot_replace_a_saved_or_cleared_original_checkpoint
);
assert!(
first
- .read("draft")
+ .read("draft", |_| Ok(()))
.await
.unwrap()
.unwrap()
+ .0
.content
.is_empty()
);
}
#[tokio::test]
-async fn raw_editor_limit_metadata_quota_and_corrupt_records_are_bounded_before_reading() {
+async fn whole_checkpoint_above_former_content_and_namespace_limits_survives_reopen() {
+ let (dir, store) = store().await;
+ let store = Arc::new(store);
+ let original = source(&store, Uuid::new_v4(), 0).await;
+ let checkpoints = store.ui_checkpoints(&original, &instance()).await.unwrap();
+ // Eight UTF-8 bytes per unit: this tests whole text beyond both old limits.
+ let body = "甲🙂z".repeat(41 * 1024 * 1024 / 8).into_bytes();
+ let data = json!({"domain":"original-host", "namespace":"original-agent"});
+ let metadata = json!({"owner":"original-editor", "selection":[1,3]});
+ let written = checkpoints
+ .write(
+ "large-draft",
+ None,
+ "editor@1",
+ &data,
+ &metadata,
+ body.clone(),
+ )
+ .await;
+ drop(checkpoints);
+ store.shutdown().await.unwrap();
+ drop(store);
+
+ // On the original implementation, still settle the owner before asserting
+ // admission. Reopen only after a successful write so refusal stays bounded.
+ let saved = if written.is_ok() {
+ let namespaces = RootNamespaces {
+ ownership: dir.path().join("owners"),
+ control: dir.path().join("control"),
+ };
+ let root = Arc::new(RootOwner::open(&dir.path().join("root"), &namespaces).unwrap());
+ let reopened = Arc::new(EventLog::for_root(root).await.unwrap());
+ let checkpoints = reopened
+ .ui_checkpoints(&original, &instance())
+ .await
+ .unwrap();
+ let saved = checkpoints.read("large-draft", |_| Ok(())).await;
+ drop(checkpoints);
+ reopened.shutdown().await.unwrap();
+ saved
+ } else {
+ Ok(None)
+ };
+
+ assert!(
+ written.is_ok(),
+ "whole checkpoint write failed: {written:?}"
+ );
+ assert_eq!(written.unwrap(), 1);
+ let (saved, ()) = saved
+ .expect("whole checkpoint read failed")
+ .expect("saved checkpoint missing");
+ assert_eq!(saved.revision, 1);
+ assert_eq!(saved.format, "editor@1");
+ assert_eq!(saved.data_source, data);
+ assert_eq!(saved.metadata, metadata);
+ assert!(
+ saved.content == body,
+ "whole checkpoint bytes differ: got {}, expected {}",
+ saved.content.len(),
+ body.len()
+ );
+}
+
+struct ReadCharge {
+ bytes: usize,
+ live: Arc<std::sync::atomic::AtomicUsize>,
+}
+impl ReadCharge {
+ fn new(bytes: usize, live: Arc<std::sync::atomic::AtomicUsize>) -> Self {
+ live.fetch_add(bytes, std::sync::atomic::Ordering::SeqCst);
+ Self { bytes, live }
+ }
+}
+impl Drop for ReadCharge {
+ fn drop(&mut self) {
+ self.live
+ .fetch_sub(self.bytes, std::sync::atomic::Ordering::SeqCst);
+ }
+}
+
+#[tokio::test]
+async fn admission_rejection_and_invalid_headers_release_original_read_ownership() {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
let (_dir, store) = store().await;
let store = Arc::new(store);
let source = source(&store, Uuid::new_v4(), 0).await;
let checkpoints = store.ui_checkpoints(&source, &instance()).await.unwrap();
- let body = vec![b'"'; MAX_CONTENT];
+ assert!(
+ checkpoints
+ .read("absent", |_| -> Result<(), StoreError> {
+ panic!("absent checkpoint must not request admission")
+ })
+ .await
+ .unwrap()
+ .is_none()
+ );
+ let body = "甲🙂z".repeat(2048).into_bytes();
checkpoints
.write(
- "limit",
+ "draft",
None,
"v1",
&Value::Null,
@@ -250,23 +351,6 @@ async fn raw_editor_limit_metadata_quota_and_corrupt_records_are_bounded_before_
)
.await
.unwrap();
- assert_eq!(
- checkpoints.read("limit").await.unwrap().unwrap().content,
- body
- );
- assert!(
- checkpoints
- .write(
- "large",
- None,
- "v1",
- &Value::Null,
- &Value::Null,
- vec![0; MAX_CONTENT + 1]
- )
- .await
- .is_err()
- );
assert!(
checkpoints
.write(
@@ -280,43 +364,156 @@ async fn raw_editor_limit_metadata_quota_and_corrupt_records_are_bounded_before_
.await
.is_err()
);
- for index in 1..39 {
- checkpoints
- .write(
- &format!("item-{index}"),
- None,
- "v1",
- &Value::Null,
- &Value::Null,
- body.clone(),
- )
- .await
- .unwrap();
- }
+ let requested = Arc::new(AtomicUsize::new(0));
+ let request = requested.clone();
+ let rejected = checkpoints
+ .read("draft", move |bytes| -> Result<(), StoreError> {
+ request.store(bytes, Ordering::SeqCst);
+ Err(invalid("test allocation admission rejected"))
+ })
+ .await;
+ assert_eq!(requested.load(Ordering::SeqCst), body.len());
assert!(
- checkpoints
- .write("overflow", None, "v1", &Value::Null, &Value::Null, body)
- .await
- .is_err()
+ rejected
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("test allocation admission rejected")
);
- checkpoints
- .write("limit", Some(1), "v1", &Value::Null, &Value::Null, vec![])
+
+ let live = Arc::new(AtomicUsize::new(0));
+ let charge = live.clone();
+ let loaded = checkpoints
+ .read("draft", move |bytes| Ok(ReadCharge::new(bytes, charge)))
.await
+ .unwrap()
.unwrap();
+ assert_eq!(loaded.0.content, body);
+ assert_eq!(live.load(Ordering::SeqCst), body.len());
+ drop(loaded);
+ assert_eq!(live.load(Ordering::SeqCst), 0);
+
+ // Invalid bounded headers are still rejected after successful admission,
+ // and each error returns the original allocation token.
+ for (query, value, diagnostic) in [
+ (
+ "UPDATE application_ui_checkpoints SET format=?, data_source_json='null', metadata_json='null', revision=1 WHERE key='draft'",
+ "x".repeat(129),
+ "invalid UI checkpoint format",
+ ),
+ (
+ "UPDATE application_ui_checkpoints SET format='v1', data_source_json=?, metadata_json='null', revision=1 WHERE key='draft'",
+ json!("x".repeat(MAX_SOURCE)).to_string(),
+ "UI checkpoint source exceeds budget",
+ ),
+ (
+ "UPDATE application_ui_checkpoints SET format='v1', data_source_json='null', metadata_json=?, revision=1 WHERE key='draft'",
+ json!("x".repeat(MAX_METADATA)).to_string(),
+ "UI checkpoint metadata exceeds budget",
+ ),
+ (
+ "UPDATE application_ui_checkpoints SET format='v1', data_source_json='null', metadata_json='null', revision=? WHERE key='draft'",
+ "9007199254740992".into(),
+ "invalid UI checkpoint revision",
+ ),
+ ] {
+ store
+ .connection
+ .run(move |connection| {
+ Box::pin(async move {
+ sqlx::query(query).bind(value).execute(connection).await?;
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+ let charge = live.clone();
+ let result = checkpoints
+ .read("draft", move |bytes| Ok(ReadCharge::new(bytes, charge)))
+ .await;
+ assert!(result.err().unwrap().to_string().contains(diagnostic));
+ assert_eq!(live.load(Ordering::SeqCst), 0);
+ }
+ drop(checkpoints);
+ store.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn admission_and_payload_share_original_snapshot_during_a_real_concurrent_write() {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
+ let (dir, store) = store().await;
+ let store = Arc::new(store);
+ let source = source(&store, Uuid::new_v4(), 0).await;
+ let checkpoints = store.ui_checkpoints(&source, &instance()).await.unwrap();
+ let original = "甲🙂z".as_bytes().to_vec();
+ let replacement = "larger replacement 甲🙂z".repeat(8).into_bytes();
checkpoints
.write(
- "overflow",
+ "draft",
None,
"v1",
&Value::Null,
&Value::Null,
- vec![0; MAX_CONTENT],
+ original.clone(),
)
.await
.unwrap();
- store.connection.run(|connection| Box::pin(async move {
- sqlx::query("UPDATE application_ui_checkpoints SET content=zeroblob(1048577) WHERE key='limit'").execute(&mut *connection).await?;
- Ok(())
- })).await.unwrap();
- assert!(checkpoints.read("limit").await.is_err());
+ // WAL lets a second connection commit while the original read transaction
+ // holds its snapshot. The callback executes exactly between length and fetch.
+ store
+ .connection
+ .run(|connection| {
+ Box::pin(async move {
+ sqlx::query("PRAGMA journal_mode=WAL")
+ .execute(connection)
+ .await?;
+ Ok(())
+ })
+ })
+ .await
+ .unwrap();
+ let database = dir.path().join("root").join(crate::root::ROOT_DATABASE);
+ let changed = replacement.clone();
+ let live = Arc::new(AtomicUsize::new(0));
+ let charge = live.clone();
+ let loaded = checkpoints
+ .read("draft", move |bytes| {
+ let writer = rusqlite::Connection::open(database)?;
+ writer.execute(
+ "UPDATE application_ui_checkpoints SET content=?, revision=2 WHERE key='draft'",
+ [changed],
+ )?;
+ Ok(ReadCharge::new(bytes, charge))
+ })
+ .await
+ .unwrap()
+ .unwrap();
+ let next = checkpoints.read("draft", Ok).await.unwrap().unwrap();
+ let original_revision = loaded.0.revision;
+ let original_bytes_match = loaded.0.content == original;
+ let original_charge = loaded.1.bytes;
+ let original_live = live.load(Ordering::SeqCst);
+ drop(loaded);
+ let released = live.load(Ordering::SeqCst);
+ let revision = checkpoints
+ .write("draft", Some(2), "v1", &Value::Null, &Value::Null, vec![])
+ .await
+ .unwrap();
+ let empty = checkpoints.read("draft", Ok).await.unwrap().unwrap();
+ drop(checkpoints);
+ store.shutdown().await.unwrap();
+
+ assert_eq!(original_revision, 1);
+ assert!(original_bytes_match);
+ assert_eq!(original_charge, original.len());
+ assert_eq!(original_live, original.len());
+ assert_eq!(released, 0);
+ assert_eq!(next.0.revision, 2);
+ assert_eq!(next.0.content, replacement);
+ assert_eq!(next.1, next.0.content.len());
+ assert_eq!(revision, 3);
+ assert_eq!(empty.0.revision, 3);
+ assert!(empty.0.content.is_empty());
+ assert_eq!(empty.1, 0);
}
diff --git a/platform/ui-runtime/README.md b/platform/ui-runtime/README.md
index af554568cd..8e5580deb7 100644
--- a/platform/ui-runtime/README.md
+++ b/platform/ui-runtime/README.md
@@ -29,6 +29,6 @@ Committed frames and accepted calls share immutable JSON state and props snapsho
Native defaults and editing do not await JavaScript. Components declare their supported native behavior in committed props; business input is queued in order. Native observations have bounded publication custody: enqueueing is not publication, and only the original logical commit acknowledges an observation. Rejection retains its exact value for retry after a real business/source change. Only complete replaceable snapshots may coalesce; command receipts and terminal outcomes remain ordered.
-Native editors own their buffers and expose immutable captures at an exact owner and revision. Multiline initialization, edits and local text reads carry complete representable UTF-8 text without a separate per-command or per-buffer size ceiling; transport framing and aggregate resource admission still apply. Single-line fields and opt-in checkpoint storage retain their own policies. A mount keeps one native recipe implementation type; changing the implementation requires a new child key [...]
+Native editors own their buffers and expose immutable captures at an exact owner and revision. Multiline initialization, edits and local text reads carry complete representable UTF-8 text without a separate per-command or per-buffer size ceiling; transport framing and aggregate resource admission still apply. Single-line fields retain their own policies. Opt-in editor checkpoints persist complete text; native capture and restore use the presenter’s shared byte budget, with restore admiss [...]
[Source](src/lib.rs) · [Cargo manifest](Cargo.toml)
diff --git a/platform/ui-runtime/README.zh-CN.md b/platform/ui-runtime/README.zh-CN.md
index d1bb7fdb3e..3af503a94b 100644
--- a/platform/ui-runtime/README.zh-CN.md
+++ b/platform/ui-runtime/README.zh-CN.md
@@ -29,6 +29,6 @@
原生默认动作与编辑不等待 JavaScript。组件在已提交 props 中声明支持的原生行为，业务输入按顺序入队。原生观察具有有界发布托管：入队不等于发布，只有原逻辑提交才能确认观察。拒绝候选保留精确原值，在业务或来源实际变化后重试。只有完整且可替代的快照允许合并；命令回执和终局结果保持顺序。
-原生编辑器拥有缓冲区，并按精确的所有者和修订号提供不可变捕获。多行初始化、编辑和本地文本读取承载完整、可表示的 UTF-8 文本，不设独立的单命令或缓冲区大小上限；传输分帧和聚合资源准入仍然适用。单行字段和可选检查点存储保留各自策略。同一挂载固定一种原生 recipe 实现类型；切换实现需要新子键或重新挂载。类型不匹配会在改变已提交原生所有者之前拒绝候选。异步变更沿用原始回执和调用方身份。操作反馈携带已确认的调用方提交信息（若存在），让组件直接记录精确历史位置；该位置本身不证明业务结果或原生编辑已完成。替换组件必须等待已接受工作和原生资源完成清理，丢弃帧或结果观察者并不足够。相关实现见 `actions`、`operations`、`editor`、`checkpoint` 和 `publication`。
+原生编辑器拥有缓冲区，并按精确的所有者和修订号提供不可变捕获。多行初始化、编辑和本地文本读取承载完整、可表示的 UTF-8 文本，不设独立的单命令或缓冲区大小上限；传输分帧和聚合资源准入仍然适用。单行字段保留各自策略。可选编辑器检查点保存完整文本；原生捕获和恢复使用呈现器的共享字节预算，恢复在同一数据库快照内先准入、再读取正文。小型 JSON 偏好保留独立的值策略。同一挂载固定一种原生 recipe 实现类型；切换实现需要新子键或重新挂载。类型不匹配会在改变已提交原生所有者之前拒绝候选。异步变更沿用原始回执和调用方身份。操作反馈携带已确认的调用方提交信息（若存在），让组件直接记录精确历史位置；该位置本身不证明业务结果或原生编辑已完成。替换组件必须等待已接受工作和原生资源完成清理，丢弃�
��或结果观察者并不足够。相关实现见 `actions`、`operations`、`editor`、`checkpoint` 和 `publication`。
[源码](src/lib.rs) · [Cargo 配置](Cargo.toml)
diff --git a/platform/ui-runtime/src/checkpoint.rs b/platform/ui-runtime/src/checkpoint.rs
index 9de027e310..9844156a7a 100644
--- a/platform/ui-runtime/src/checkpoint.rs
+++ b/platform/ui-runtime/src/checkpoint.rs
@@ -29,7 +29,7 @@ use crate::{
use maka_capabilities::call::Resources;
use maka_event_log::{
EventLog,
- applications::checkpoints::{Checkpoints, MAX_CONTENT, MAX_METADATA, MAX_SOURCE},
+ applications::checkpoints::{Checkpoints, MAX_METADATA, MAX_SOURCE},
};
use maka_plugin_core::{composition::InstanceId, runtime::Mount};
use schemars::JsonSchema;
@@ -351,16 +351,18 @@ impl EditorCheckpoint {
return;
}
let result = (|| {
- if snapshot.text.bytes > MAX_CONTENT {
- return Err("editor checkpoint content exceeds 1 MiB".to_owned());
- }
let bytes = serde_json::to_vec(&saved).map_err(|error| error.to_string())?;
if bytes.len() > MAX_METADATA {
return Err("editor checkpoint metadata exceeds 32 KiB".into());
}
+ let size = snapshot
+ .text
+ .bytes
+ .checked_add(bytes.len())
+ .ok_or("editor checkpoint size overflow")?;
let memory = self
.budget
- .reserve(Kind::Encoded, snapshot.text.bytes + bytes.len())
+ .reserve(Kind::Encoded, size)
.map_err(|error| error.to_string())?;
let text = text();
if text.len() != snapshot.text.bytes {
diff --git a/platform/ui-runtime/src/checkpoint/tests.rs b/platform/ui-runtime/src/checkpoint/tests.rs
index 15acb47e72..38cba6ff96 100644
--- a/platform/ui-runtime/src/checkpoint/tests.rs
+++ b/platform/ui-runtime/src/checkpoint/tests.rs
@@ -160,8 +160,17 @@ async fn final_coalesced_snapshot_flushes_before_cleanup_and_restores_after_root
until(&mut editor, &policy, EditorCheckpoint::accepts_input).await;
let blocker = rusqlite::Connection::open(dir.path().join("root").join(ROOT_DATABASE)).unwrap();
blocker.execute_batch("BEGIN IMMEDIATE").unwrap();
+ let final_body = format!(
+ "{text}100\n{}FINAL COMPLETE DRAFT",
+ "保留中文😀 draft bytes.\n".repeat(50_000)
+ );
+ assert!(final_body.len() > 1024 * 1024);
for revision in 2..=100 {
- let body = format!("{text}{revision}");
+ let body = if revision == 100 {
+ final_body.clone()
+ } else {
+ format!("{text}{revision}")
+ };
editor.save(
&snapshot("old-owner", revision, &body),
&body,
@@ -217,7 +226,12 @@ async fn final_coalesced_snapshot_flushes_before_cleanup_and_restores_after_root
let Phase::Restore(loaded) = &editor.phase else {
unreachable!()
};
- assert_eq!(loaded.text, format!("{text}100"));
+ assert!(
+ loaded.text.as_bytes() == final_body.as_bytes(),
+ "reopened checkpoint differs: got {} bytes, expected {}",
+ loaded.text.len(),
+ final_body.len()
+ );
assert_eq!(loaded.saved.metadata, policy.metadata);
assert_eq!(loaded.saved.snapshot.text.revision, 100);
assert_eq!(
@@ -231,27 +245,129 @@ async fn final_coalesced_snapshot_flushes_before_cleanup_and_restores_after_root
clear_text: false,
});
assert!(editor.restoration("other-owner", &policy).is_none());
- assert_eq!(
- editor.restoration("new-owner", &policy).unwrap().text(),
- format!("{text}100")
+ assert!(
+ editor
+ .restoration("new-owner", &policy)
+ .unwrap()
+ .text()
+ .as_bytes()
+ == final_body.as_bytes(),
+ "restoration must retain every original draft byte"
);
editor.restored();
assert!(!editor.accepts_input());
- let body = format!("{text}100");
- editor.save(&snapshot("new-owner", 1, &body), &body, &policy.metadata);
+ editor.save(
+ &snapshot("new-owner", 1, &final_body),
+ &final_body,
+ &policy.metadata,
+ );
until(&mut editor, &policy, EditorCheckpoint::accepts_input).await;
editor.close();
drop(editor);
resources.finish().await.unwrap();
- let saved = store
+ let (saved, ()) = store
.ui_checkpoints(&source, &instance())
.await
.unwrap()
- .read(&policy.key)
+ .read(&policy.key, |_| Ok(()))
.await
.unwrap()
.unwrap();
assert_eq!(saved.metadata["snapshot"]["text"]["owner"], "new-owner");
+ assert!(
+ saved.content.as_slice() == final_body.as_bytes(),
+ "new owner must persist the complete restored draft"
+ );
+ assert_eq!(budget.usage(), (0, 0, 0));
+ store.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn checkpoint_load_admits_actual_content_and_preserves_records_when_budget_is_exhausted() {
+ let (_dir, store) = store().await;
+ let source = source(&store, Uuid::new_v4(), 0).await;
+ let policy = policy();
+ let checkpoints = store.ui_checkpoints(&source, &instance()).await.unwrap();
+ let large = format!("甲👩💻z{}COMPLETE END", "保留中文😀。\n".repeat(70_000));
+ assert!(large.len() > 1024 * 1024);
+ let mut version = None;
+ for (body, fits) in [("甲👩💻z", true), (large.as_str(), false)] {
+ let saved = Saved {
+ snapshot: snapshot("stored-owner", 1, body),
+ metadata: policy.metadata.clone(),
+ };
+ version = Some(
+ checkpoints
+ .write(
+ &policy.key,
+ version,
+ &policy.stored_format(),
+ &policy.data_source,
+ &serde_json::to_value(&saved).unwrap(),
+ body.as_bytes().to_vec(),
+ )
+ .await
+ .unwrap(),
+ );
+ let resources = Arc::new(Resources::default());
+ let budget = NativeBudget::new(crate::budget::BudgetLimits {
+ encoded_bytes: 128 * 1024,
+ ..Default::default()
+ });
+ let mut editor = EditorCheckpoint::new(
+ Some(scope(&store, &source).await),
+ &policy,
+ &resources,
+ budget.clone(),
+ );
+ until(&mut editor, &policy, |e| !matches!(e.phase, Phase::Loading)).await;
+ let loaded = match &editor.phase {
+ Phase::Restore(value) => {
+ Some((value.text.as_bytes() == body.as_bytes(), value.version))
+ }
+ _ => None,
+ };
+ let failed = match &editor.phase {
+ Phase::Failed {
+ writing,
+ diagnostic,
+ } => Some((*writing, diagnostic.clone())),
+ _ => None,
+ };
+ let retained_bytes = budget.usage().0;
+ editor.close();
+ drop(editor);
+ resources.finish().await.unwrap();
+ assert_eq!(budget.usage(), (0, 0, 0));
+ if fits {
+ assert_eq!(
+ loaded,
+ Some((true, version.unwrap())),
+ "a small checkpoint must not reserve the former whole-content ceiling"
+ );
+ assert_eq!(failed, None);
+ assert!(retained_bytes > 0 && retained_bytes < 128 * 1024);
+ } else {
+ assert_eq!(loaded, None);
+ assert!(
+ matches!(failed, Some((false, ref diagnostic))
+ if diagnostic.contains("native resource budget exhausted")),
+ "large checkpoint load must fail its actual native admission: {failed:?}"
+ );
+ assert_eq!(retained_bytes, 0);
+ }
+ let (record, ()) = checkpoints
+ .read(&policy.key, |_| Ok(()))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(record.revision, version.unwrap());
+ assert!(
+ record.content.as_slice() == body.as_bytes(),
+ "load admission changed stored draft bytes"
+ );
+ assert_eq!(record.metadata, serde_json::to_value(&saved).unwrap());
+ }
store.shutdown().await.unwrap();
}
@@ -272,7 +388,7 @@ async fn an_older_ack_cannot_hide_a_newer_capture_failure_and_retry_accounts_for
let body = "甲👩💻z";
editor.save(&snapshot("owner", 1, body), body, &policy.metadata);
let mut invalid = snapshot("owner", 2, body);
- invalid.text.bytes = MAX_CONTENT + 1;
+ invalid.text.bytes = body.len() + 1;
editor.save(&invalid, body, &policy.metadata);
until(&mut editor, &policy, |e| e.saved_sequence == 1).await;
assert!(!editor.accepts_input());
@@ -344,10 +460,11 @@ async fn source_mismatch_and_conflicting_writers_cannot_overwrite_saved_content(
.ui_checkpoints(&source, &instance())
.await
.unwrap()
- .read(&policy.key)
+ .read(&policy.key, |_| Ok(()))
.await
.unwrap()
.unwrap()
+ .0
.metadata["snapshot"]["text"]["owner"],
"first"
);
@@ -370,7 +487,10 @@ async fn concurrent_preference_reads_are_not_rejected_by_a_worker_count() {
let source = source(&store, Uuid::new_v4(), 0).await;
let recipe_scope = scope(&store, &source).await;
let resources = Arc::new(Resources::default());
- let budget = NativeBudget::default();
+ let budget = NativeBudget::new(crate::budget::BudgetLimits {
+ encoded_bytes: 128 * 1024,
+ ..Default::default()
+ });
let mut preferences: Vec<_> = (0..16)
.map(|index| {
ValueCheckpoint::new(
@@ -490,7 +610,11 @@ async fn preference_busy_keeps_the_original_write_result_and_retirement_drains_i
};
assert_eq!(value, Some(json!({"original": true})));
let checkpoints = scope.open(&store).unwrap();
- let saved = checkpoints.read(&props.key).await.unwrap().unwrap();
+ let (saved, ()) = checkpoints
+ .read(&props.key, |_| Ok(()))
+ .await
+ .unwrap()
+ .unwrap();
assert_eq!(saved.revision, revision);
assert_eq!(
serde_json::from_slice::<Value>(&saved.content).unwrap(),
@@ -510,7 +634,11 @@ async fn preference_busy_keeps_the_original_write_result_and_retirement_drains_i
blocker.execute_batch("COMMIT").unwrap();
resources.finish().await.unwrap();
assert_eq!(budget.usage(), (0, 0, 0));
- let saved = checkpoints.read(&props.key).await.unwrap().unwrap();
+ let (saved, ()) = checkpoints
+ .read(&props.key, |_| Ok(()))
+ .await
+ .unwrap()
+ .unwrap();
assert_eq!(
serde_json::from_slice::<Value>(&saved.content).unwrap(),
json!({"retired": true})
diff --git a/platform/ui-runtime/src/checkpoint/value.rs b/platform/ui-runtime/src/checkpoint/value.rs
index d9f1ff4f04..4c7b788507 100644
--- a/platform/ui-runtime/src/checkpoint/value.rs
+++ b/platform/ui-runtime/src/checkpoint/value.rs
@@ -142,26 +142,46 @@ impl ValueCheckpoint {
loop {
let sequence = command.sequence();
let result: Result<ValueCheckpointState, String> = async {
- let _bytes = budget
- .reserve(Kind::Encoded, super::MAX_CONTENT + 32 * 1024)
- .map_err(|e| e.to_string())?;
if checkpoints.is_none() {
checkpoints = Some(scope.open(&store)?);
}
let checkpoints = checkpoints.as_ref().unwrap();
match command {
ValueCheckpointCommand::Read { .. } => {
- let saved =
- checkpoints.read(&key).await.map_err(|e| e.to_string())?;
+ let budget = budget.clone();
+ let saved = checkpoints
+ .read(&key, move |content_bytes| {
+ if content_bytes > 16384 {
+ return Err(
+ maka_event_log::StoreError::InvalidTransition(
+ "UI preference format is incompatible".into(),
+ ),
+ );
+ }
+ budget
+ .reserve(
+ Kind::Encoded,
+ content_bytes
+ + super::MAX_METADATA
+ + super::MAX_SOURCE
+ + 128,
+ )
+ .map_err(|error| {
+ maka_event_log::StoreError::InvalidTransition(
+ error.to_string(),
+ )
+ })
+ })
+ .await
+ .map_err(|e| e.to_string())?;
match saved {
None => Ok(ValueCheckpointState::Ready {
revision: None,
value: None,
}),
- Some(saved) => {
+ Some((saved, _bytes)) => {
if saved.format != format
|| saved.data_source != Value::Null
- || saved.content.len() > 16384
{
return Err(
"UI preference format is incompatible".into()
@@ -181,11 +201,15 @@ impl ValueCheckpoint {
value,
..
} => {
+ let mut bytes = budget
+ .reserve(Kind::Encoded, 16384)
+ .map_err(|e| e.to_string())?;
let content =
serde_json::to_vec(&value).map_err(|e| e.to_string())?;
if content.len() > 16384 {
return Err("UI preferences exceed 16 KiB".into());
}
+ bytes.shrink_to(content.len());
let revision = checkpoints
.write(
&key,
diff --git a/platform/ui-runtime/src/checkpoint/worker.rs b/platform/ui-runtime/src/checkpoint/worker.rs
index 5bbc2e3030..b84eb10e5a 100644
--- a/platform/ui-runtime/src/checkpoint/worker.rs
+++ b/platform/ui-runtime/src/checkpoint/worker.rs
@@ -129,16 +129,25 @@ async fn run(
if checkpoints.is_none() {
checkpoints = Some(scope.open(&store).map_err(diagnostic)?);
}
- let mut memory = budget
- .reserve(Kind::Encoded, MAX_CONTENT + MAX_METADATA + MAX_SOURCE + 128)
- .map_err(diagnostic)?;
+ let budget = budget.clone();
let record = checkpoints
.as_ref()
.unwrap()
- .read(&policy.key)
+ .read(&policy.key, move |content_bytes| {
+ let size = content_bytes
+ .checked_add(MAX_METADATA + MAX_SOURCE + 128)
+ .ok_or_else(|| {
+ maka_event_log::StoreError::InvalidTransition(
+ "editor checkpoint size overflow".into(),
+ )
+ })?;
+ budget.reserve(Kind::Encoded, size).map_err(|error| {
+ maka_event_log::StoreError::InvalidTransition(error.to_string())
+ })
+ })
.await
.map_err(diagnostic)?;
- let Some(record) = record else {
+ let Some((record, mut memory)) = record else {
return Ok(None);
};
if record.format != policy.stored_format()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org