Add weighted fair scheduling for vqueues via scope rules

cli/src/commands/rules/list.rs

@@ -41,7 +41,7 @@ async fn list(env: &CliEnv, opts: &List) -> Result<()> {
     let client = DataFusionHttpClient::new(env).await?;
     let rows: Vec<RuleRow> = client
         .run_json_query(
-            "SELECT pattern, concurrency, description, disabled, version, last_modified \
+            "SELECT pattern, concurrency, scheduling_weight, description, disabled, version, last_modified \
              FROM sys_rules ORDER BY pattern"
                 .to_string(),
         )
@@ -57,13 +57,14 @@ async fn list(env: &CliEnv, opts: &List) -> Result<()> {
         table.set_styled_header(vec![
             "PATTERN",
             "CONCURRENCY",
+            "WEIGHT",
             "DISABLED",
             "DESCRIPTION",
             "VERSION",
             "LAST MODIFIED",
         ]);
     } else {
-        table.set_styled_header(vec!["PATTERN", "CONCURRENCY", "DISABLED"]);
+        table.set_styled_header(vec!["PATTERN", "CONCURRENCY", "WEIGHT", "DISABLED"]);
     }
 
     for row in rows {
@@ -73,6 +74,7 @@ async fn list(env: &CliEnv, opts: &List) -> Result<()> {
             table.add_row(vec![
                 Cell::new(row.pattern),
                 Cell::new(render_concurrency(row.concurrency)),
+                Cell::new(row.scheduling_weight.unwrap_or(1)),
                 Cell::new(disabled),
                 Cell::new(row.description.unwrap_or_default()),
                 Cell::new(row.version),
@@ -82,6 +84,7 @@ async fn list(env: &CliEnv, opts: &List) -> Result<()> {
             table.add_row(vec![
                 Cell::new(row.pattern),
                 Cell::new(render_concurrency(row.concurrency)),
+                Cell::new(row.scheduling_weight.unwrap_or(1)),
                 Cell::new(disabled),
             ]);
         }

cli/src/commands/rules/mod.rs

@@ -53,6 +53,8 @@ pub(crate) struct RuleRow {
     #[serde(default)]
     pub concurrency: Option<u32>,
     #[serde(default)]
+    pub scheduling_weight: Option<u32>,
+    #[serde(default)]
     pub description: Option<String>,
     pub disabled: bool,
     pub version: u32,
@@ -65,6 +67,11 @@ impl RuleRow {
     fn concurrency(&self) -> Option<NonZeroU32> {
         self.concurrency.and_then(NonZeroU32::new)
     }
+
+    /// The scheduling weight as a `NonZeroU32` (the runtime shape).
+    fn scheduling_weight(&self) -> Option<std::num::NonZeroU32> {
+        self.scheduling_weight.and_then(std::num::NonZeroU32::new)
+    }
 }
 
 /// Renders a concurrency limit for display (`unlimited` when unset).
@@ -88,7 +95,7 @@ pub(crate) async fn fetch_rule(
     canonical_pattern: &str,
 ) -> Result<Option<RuleRow>> {
     let query = format!(
-        "SELECT pattern, concurrency, description, disabled, version, last_modified \
+        "SELECT pattern, concurrency, scheduling_weight, description, disabled, version, last_modified \
          FROM sys_rules WHERE pattern = '{}'",
         escape_sql(canonical_pattern)
     );
@@ -142,7 +149,8 @@ pub(crate) async fn toggle_disabled(env: &CliEnv, pattern: &str, disabled: bool)
     let client = AdminClient::new(env).await?;
     let request = UpsertRuleRequest {
         pattern,
-        limits: UserLimits::new(current.concurrency()),
+        limits: UserLimits::new(current.concurrency())
+            .with_scheduling_weight(current.scheduling_weight()),
         description: current.description.clone(),
         disabled,
         precondition: Precondition::Matches(Version::from(current.version)),

cli/src/commands/rules/set.rs

@@ -37,6 +37,14 @@ pub struct Set {
     #[clap(long)]
     unlimited: bool,
 
+    /// Scheduling weight (>= 1) for this rule's scope in the weighted
+    /// round-robin scheduler: weight N receives N dispatch slots per cycle
+    /// relative to weight-1 groups. On an existing rule, omitting this leaves
+    /// the current weight unchanged. Requires the vqueues scheduler; only
+    /// scope-level exact patterns are consulted.
+    #[clap(long)]
+    weight: Option<NonZeroU32>,
+
     /// Description for the rule
     #[clap(long)]
     description: Option<String>,
@@ -62,7 +70,8 @@ pub async fn run_set(State(env): State<CliEnv>, opts: &Set) -> Result<()> {
                 None
             } else {
                 opts.concurrency
-            }),
+            })
+            .with_scheduling_weight(opts.weight),
             description: opts.description.clone(),
             disabled: opts.disabled,
             precondition: Precondition::DoesNotExist,
@@ -82,9 +91,10 @@ pub async fn run_set(State(env): State<CliEnv>, opts: &Set) -> Result<()> {
                 .description
                 .clone()
                 .or_else(|| rule.description.clone());
+            let scheduling_weight = opts.weight.or_else(|| rule.scheduling_weight());
             UpsertRuleRequest {
                 pattern,
-                limits: UserLimits::new(concurrency),
+                limits: UserLimits::new(concurrency).with_scheduling_weight(scheduling_weight),
                 description,
                 disabled: rule.disabled,
                 precondition: Precondition::Matches(Version::from(rule.version)),

crates/limiter/src/rule_book.rs

@@ -483,6 +483,7 @@ mod tests {
         RuleUpsert {
             limits: UserLimits {
                 concurrency: NonZeroU32::new(concurrency),
+                scheduling_weight: None,
             },
             description: None,
             disabled: false,
@@ -516,6 +517,7 @@ mod tests {
             PersistedRule {
                 limits: UserLimits {
                     concurrency: NonZeroU32::new(1000),
+                    scheduling_weight: None,
                 },
                 description: Some("global default".to_owned()),
                 disabled: false,
@@ -528,6 +530,7 @@ mod tests {
             PersistedRule {
                 limits: UserLimits {
                     concurrency: NonZeroU32::new(10),
+                    scheduling_weight: None,
                 },
                 description: None,
                 disabled: true,

crates/limiter/src/user_limits.rs

@@ -45,11 +45,35 @@ pub struct UserLimits {
     )]
     #[cfg_attr(feature = "schema", schema(value_type = Option<u32>, minimum = 1))]
     pub concurrency: Option<NonZeroU32>,
+
+    /// Scheduling weight of this rule's scope in the scheduler's weighted
+    /// round-robin: a scope with weight N receives N dispatch slots per cycle
+    /// relative to weight-1 groups, regardless of how many queues it has.
+    /// `None` means the default weight of 1. Only scope-level exact patterns
+    /// are consulted. Requires the vqueues scheduler.
+    ///
+    /// Upserts replace the whole limits object: omitting this field resets
+    /// the weight (the CLI preserves it by re-reading the current rule).
+    #[cfg_attr(feature = "bilrost", bilrost(tag(2)))]
+    #[cfg_attr(
+        feature = "serde",
+        serde(default, skip_serializing_if = "Option::is_none")
+    )]
+    #[cfg_attr(feature = "schema", schema(value_type = Option<u32>, minimum = 1))]
+    pub scheduling_weight: Option<NonZeroU32>,
 }
 
 impl UserLimits {
     pub fn new(concurrency: Option<NonZeroU32>) -> Self {
-        Self { concurrency }
+        Self {
+            concurrency,
+            scheduling_weight: None,
+        }
+    }
+
+    pub fn with_scheduling_weight(mut self, scheduling_weight: Option<NonZeroU32>) -> Self {
+        self.scheduling_weight = scheduling_weight;
+        self
     }
 }
 
@@ -69,3 +93,36 @@ pub enum RuleUpdate {
     /// Remove a rule by its pattern.
     Remove { pattern: RulePattern<ReString> },
 }
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZeroU32;
+
+    use bilrost::{Message, OwnedMessage};
+
+    use super::*;
+
+    /// Wire compatibility: pre-weight encodings (concurrency only) decode with
+    /// `scheduling_weight = None`, and a default weight is omitted on the wire
+    /// so old readers see the exact pre-weight byte layout.
+    #[test]
+    fn scheduling_weight_wire_compat() {
+        // "old" shape: only tag(1) present
+        let old = UserLimits::new(NonZeroU32::new(100));
+        let old_bytes = old.encode_to_bytes();
+        let decoded = <UserLimits as OwnedMessage>::decode(old_bytes.clone()).unwrap();
+        assert_eq!(decoded.scheduling_weight, None);
+        assert_eq!(decoded.concurrency, NonZeroU32::new(100));
+
+        // new shape round-trips
+        let new = UserLimits::new(NonZeroU32::new(100)).with_scheduling_weight(NonZeroU32::new(10));
+        let new_bytes = new.encode_to_bytes();
+        let decoded = <UserLimits as OwnedMessage>::decode(new_bytes).unwrap();
+        assert_eq!(decoded.scheduling_weight, NonZeroU32::new(10));
+
+        // None weight encodes identically to the old shape (old readers
+        // decoding new writers see no unknown data)
+        let new_default = UserLimits::new(NonZeroU32::new(100));
+        assert_eq!(new_default.encode_to_bytes(), old_bytes);
+    }
+}

crates/storage-query-datafusion/src/rules/row.rs

@@ -23,6 +23,9 @@ pub(crate) fn append_rule_row(
     if let Some(concurrency) = rule.limits.concurrency {
         row.concurrency(concurrency.get());
     }
+    if let Some(weight) = rule.limits.scheduling_weight {
+        row.scheduling_weight(weight.get());
+    }
     if let Some(description) = rule.description.as_deref() {
         row.description(description);
     }

crates/storage-query-datafusion/src/rules/schema.rs

@@ -20,6 +20,10 @@ define_table!(sys_rules(
     /// rule does not constrain concurrency.
     concurrency: DataType::UInt32,
 
+    /// Scheduling weight of this rule's scope in the weighted round-robin
+    /// scheduler. Null means the default weight of 1.
+    scheduling_weight: DataType::UInt32,
+
     /// Free-form description set by the operator.
     description: DataType::LargeUtf8,
 

crates/types/src/config/common.rs

@@ -662,6 +662,14 @@ experimental! {
     /// Since v1.7.0
     scoped_virtual_objects,
 
+    /// # Enables scope inheritance along the call chain
+    ///
+    /// When enabled, a child invocation created via ctx.call/ctx.send inherits
+    /// its caller's scope when the child target carries no scope of its own,
+    /// so scope-level rules govern the whole call tree rooted at a scoped
+    /// ingress invocation. Requires `vqueues`.
+    scope_inheritance,
+
     /// # Enables Kafka header support for scoped invocations
     ///
     /// When enabled, Kafka subscriptions read `x-restate-scope` and

crates/types/src/partitions/features.rs

@@ -49,6 +49,11 @@ pub enum PartitionFeatureChange {
     ///
     /// *Since v1.7.0*
     EnableUniqueRandomSeeds = 3,
+    /// Child invocations created via ctx.call inherit their caller's scope
+    /// when the child target has none.
+    ///
+    /// *Since v1.7.0*
+    EnableScopeInheritance = 4,
 }
 
 impl PartitionFeatureChange {
@@ -65,6 +70,7 @@ impl PartitionFeatureChange {
             Self::EnableJournalV2 => &RESTATE_VERSION_1_6_0,
             Self::EnableVqueues => &RESTATE_VERSION_1_7_0,
             Self::EnableUniqueRandomSeeds => &RESTATE_VERSION_1_7_0,
+            Self::EnableScopeInheritance => &RESTATE_VERSION_1_7_0,
         }
     }
 
@@ -77,6 +83,9 @@ impl PartitionFeatureChange {
             Self::EnableUniqueRandomSeeds => {
                 !std::mem::replace(&mut features.unique_random_seeds, true)
             }
+            Self::EnableScopeInheritance => {
+                !std::mem::replace(&mut features.scope_inheritance, true)
+            }
         }
     }
 }
@@ -121,6 +130,11 @@ pub struct PersistedFeatures {
     /// *Since v1.7.0*
     #[bilrost(tag(3))]
     pub unique_random_seeds: bool,
+    /// Child invocations inherit their caller's scope when they have none.
+    ///
+    /// *Since v1.7.0*
+    #[bilrost(tag(4))]
+    pub scope_inheritance: bool,
 }
 
 impl PersistedFeatures {
@@ -132,6 +146,7 @@ impl PersistedFeatures {
             self.journal_v2.then_some("journal_v2"),
             self.vqueues.then_some("vqueues"),
             self.unique_random_seeds.then_some("unique_random_seeds"),
+            self.scope_inheritance.then_some("scope_inheritance"),
         ]
         .into_iter()
         .flatten()

crates/vqueues/src/lib.rs

@@ -21,7 +21,7 @@ use std::time::Duration;
 pub use cache::{VQueueHandle, VQueuesMeta, VQueuesMetaCache};
 pub use metric_definitions::describe_metrics;
 pub use restate_worker_api::{ResourceKind, SchedulingStatus, VQueueSchedulerStatus};
-pub use scheduler::{ResourceManager, SchedulerService};
+pub use scheduler::{ResourceManager, SchedulerService, ScopeWeights, scope_weight_resolver};
 pub use util::*;
 
 use smallvec::SmallVec;