fix(server): query & backfill perf (#15144)

#### PR Dependency Tree


* **PR #15144** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Document history retention is now explicitly controlled via
caller-provided max-age parameters during pending doc compaction.

* **Improvements**
* Quota state backfilling/reconciliation was improved to reduce
unnecessary work and ensure missing quota states are created in batches.
* Permission context loading now more strictly respects “known” vs
“stale” quota runtime state.

* **Bug Fixes**
* Workspace member responses now populate invite IDs correctly from the
nested user information.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/backend/native/src/backend_runtime/constants.rs

@@ -1,4 +1,3 @@
-pub(super) const DEFAULT_HISTORY_PERIOD_SECONDS: i32 = 7 * 24 * 60 * 60;
 pub(super) const BYOK_LOCAL_LEASE_ACTIVE_PURPOSE: &str = "copilot_byok_local_lease:active";
 pub(super) const BYOK_LOCAL_LEASE_PURPOSE: &str = "copilot_byok_local_lease";
 pub(super) const MAGIC_LINK_OTP_PURPOSE: &str = "magic_link_otp";

packages/backend/native/src/backend_runtime/doc_compactor.rs

@@ -3,9 +3,7 @@ use napi::Result;
 use sqlx::{FromRow, PgPool, Postgres, Row, Transaction};
 use y_octo::Doc;
 
-use super::{
-  BackendRuntime, constants::DEFAULT_HISTORY_PERIOD_SECONDS, error::napi_error, types::RuntimeDocCompactionResult,
-};
+use super::{BackendRuntime, error::napi_error, types::RuntimeDocCompactionResult};
 
 #[derive(FromRow)]
 struct SnapshotRow {
@@ -36,6 +34,7 @@ impl DocCompactorStore {
     doc_id: &str,
     batch_limit: i64,
     history_min_interval_ms: i64,
+    history_max_age_seconds: i64,
   ) -> Result<(i64, bool)> {
     compact_doc(
       self.pool.clone(),
@@ -43,6 +42,7 @@ impl DocCompactorStore {
       doc_id,
       batch_limit,
       history_min_interval_ms,
+      history_max_age_seconds,
     )
     .await
   }
@@ -64,6 +64,14 @@ fn apply_updates(updates: impl IntoIterator<Item = Vec<u8>>) -> Result<Vec<u8>>
     .map_err(|err| napi_error(format!("DocCompactor encode failed: {err}")))
 }
 
+fn checked_milliseconds(value: i64, field: &str) -> Result<Duration> {
+  Duration::try_milliseconds(value).ok_or_else(|| napi_error(format!("DocCompactor {field} is too large")))
+}
+
+fn checked_seconds(value: i64, field: &str) -> Result<Duration> {
+  Duration::try_seconds(value).ok_or_else(|| napi_error(format!("DocCompactor {field} is too large")))
+}
+
 async fn load_snapshot(
   tx: &mut Transaction<'_, Postgres>,
   workspace_id: &str,
@@ -186,27 +194,13 @@ async fn should_create_history(
     return Ok(false);
   }
 
-  Ok(last_timestamp < snapshot.updated_at - Duration::milliseconds(history_min_interval_ms))
-}
+  let min_interval = checked_milliseconds(history_min_interval_ms, "history interval")?;
+  let threshold = snapshot
+    .updated_at
+    .checked_sub_signed(min_interval)
+    .ok_or_else(|| napi_error("DocCompactor history interval is out of range"))?;
 
-async fn history_max_age_seconds(tx: &mut Transaction<'_, Postgres>, workspace_id: &str) -> Result<i32> {
-  let row = sqlx::query(
-    r#"
-    SELECT history_period_seconds
-    FROM effective_workspace_quota_states
-    WHERE workspace_id = $1
-    "#,
-  )
-  .bind(workspace_id)
-  .fetch_optional(&mut **tx)
-  .await
-  .map_err(|err| napi_error(format!("DocCompactor load history quota failed: {err}")))?;
-
-  Ok(
-    row
-      .map(|row| row.get("history_period_seconds"))
-      .unwrap_or(DEFAULT_HISTORY_PERIOD_SECONDS),
-  )
+  Ok(last_timestamp < threshold)
 }
 
 async fn create_history(
@@ -214,13 +208,16 @@ async fn create_history(
   workspace_id: &str,
   doc_id: &str,
   snapshot: &SnapshotRow,
+  max_age_seconds: i64,
 ) -> Result<bool> {
-  let max_age_seconds = history_max_age_seconds(tx, workspace_id).await?;
   if max_age_seconds <= 0 {
     return Ok(false);
   }
 
-  let expired_at = Utc::now() + Duration::seconds(max_age_seconds as i64);
+  let max_age = checked_seconds(max_age_seconds, "history max age")?;
+  let expired_at = Utc::now()
+    .checked_add_signed(max_age)
+    .ok_or_else(|| napi_error("DocCompactor history max age is out of range"))?;
   sqlx::query(
     r#"
     INSERT INTO snapshot_histories
@@ -274,6 +271,7 @@ async fn compact_doc(
   doc_id: &str,
   batch_limit: i64,
   history_min_interval_ms: i64,
+  history_max_age_seconds: i64,
 ) -> Result<(i64, bool)> {
   let mut tx = pool
     .begin()
@@ -317,7 +315,7 @@ async fn compact_doc(
     && let Some(snapshot) = &snapshot
     && should_create_history(&mut tx, snapshot, workspace_id, doc_id, history_min_interval_ms).await?
   {
-    history_created = create_history(&mut tx, workspace_id, doc_id, snapshot).await?;
+    history_created = create_history(&mut tx, workspace_id, doc_id, snapshot, history_max_age_seconds).await?;
   }
 
   let timestamps = updates.iter().map(|update| update.created_at).collect::<Vec<_>>();
@@ -336,13 +334,20 @@ impl BackendRuntime {
   ///
   /// Do not use this for snapshots that will be sent back to yjs clients until
   /// the y-octo/yjs round-trip compatibility issue is resolved.
+  ///
+  /// The caller owns quota reconciliation and must pass a fresh
+  /// history_max_age_seconds value. The compactor intentionally does not read
+  /// effective_workspace_quota_states; if a future caller cannot provide a
+  /// fresh quota state, fail and retry after Node reconciles it.
   #[napi]
+  #[allow(clippy::too_many_arguments)]
   pub async fn compact_pending_doc_updates(
     &self,
     workspace_id: String,
     doc_id: String,
     batch_limit: i64,
     history_min_interval_ms: i64,
+    history_max_age_seconds: i64,
     owner: String,
     lease_ttl_ms: i64,
   ) -> Result<RuntimeDocCompactionResult> {
@@ -352,6 +357,16 @@ impl BackendRuntime {
     if history_min_interval_ms < 0 {
       return Err(napi_error("doc compactor history interval must be non-negative"));
     }
+    if history_max_age_seconds < 0 {
+      return Err(napi_error("doc compactor history max age must be non-negative"));
+    }
+    checked_milliseconds(history_min_interval_ms, "history interval")?;
+    if history_max_age_seconds > 0 {
+      let max_age = checked_seconds(history_max_age_seconds, "history max age")?;
+      Utc::now()
+        .checked_add_signed(max_age)
+        .ok_or_else(|| napi_error("DocCompactor history max age is out of range"))?;
+    }
 
     let lease_key = format!("doc:update:{workspace_id}:{doc_id}");
     let Some(lease) = self.acquire_coordination_lease(lease_key, owner, lease_ttl_ms).await? else {
@@ -366,7 +381,13 @@ impl BackendRuntime {
     };
 
     let result = DocCompactorStore::new(self.pool().await?)
-      .compact_doc(&workspace_id, &doc_id, batch_limit, history_min_interval_ms)
+      .compact_doc(
+        &workspace_id,
+        &doc_id,
+        batch_limit,
+        history_min_interval_ms,
+        history_max_age_seconds,
+      )
       .await;
 
     let released = self