New VQueueId

Author: AhmedSoliman

Cargo.lock

@@ -221,6 +221,9 @@ impl InvocationBuilder {
         };
 
         let invocation_id = InvocationId::generate_or_else(&invocation_target, None, || {
+            // todo: reconcile this with the world of scopes+limit keys.
+            // In particular, the scatter-width of partition keys and how scopes can be assigned to
+            // ingested items from kafka.
             partitioner::HashPartitioner::compute_partition_key(seed)
         });
 

crates/ingress-kafka/src/builder.rs

@@ -1823,7 +1823,7 @@ mod tests {
     };
     use restate_types::schema::service::ServiceMetadata;
     use restate_types::service_protocol::ServiceProtocolVersion;
-    use restate_types::vqueue::{VQueueId, VQueueInstance, VQueueParent};
+    use restate_types::vqueue::VQueueId;
     use tempfile::tempdir;
     use test_log::test;
     use tokio::sync::mpsc;
@@ -2385,10 +2385,9 @@ mod tests {
 
         // Create an invocation state machine and register it with an in-flight notification proposal
         let mut ism = InvocationStateMachine::create(
-            Some(VQueueId::new(
-                VQueueParent::default_unlimited(),
+            Some(VQueueId::custom(
                 invocation_id.partition_key(),
-                VQueueInstance::Default,
+                invocation_id.to_string(),
             )),
             Permit::new_empty(),
             invocation_target.clone(),

crates/invoker-impl/src/lib.rs

@@ -22,16 +22,12 @@ use restate_storage_api::vqueue_table::{
 };
 use restate_types::clock::UniqueTimestamp;
 use restate_types::identifiers::PartitionKey;
-use restate_types::vqueue::{EffectivePriority, VQueueId, VQueueInstance, VQueueParent};
+use restate_types::vqueue::{EffectivePriority, VQueueId};
 
 use crate::PartitionStore;
 
 fn test_qid() -> VQueueId {
-    VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(1000u64),
-        VQueueInstance::from_raw(1),
-    )
+    VQueueId::custom(PartitionKey::from(1000u64), "1")
 }
 
 fn entry_card(id: u8) -> EntryCard {
@@ -120,11 +116,7 @@ fn verify_priority_ordering(db: &crate::PartitionDb) {
 
 /// Test: Within the same priority, items are ordered by visible_at then created_at.
 async fn ordering_within_same_priority<W: WriteVQueueTable>(txn: &mut W) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(2000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(2000, "1");
 
     // All entries have UserDefault priority, but different visible_at and created_at
     // visible_at ordering: Now (0) < At(ts)
@@ -155,11 +147,7 @@ async fn ordering_within_same_priority<W: WriteVQueueTable>(txn: &mut W) {
 }
 
 fn verify_ordering_within_same_priority(db: &crate::PartitionDb) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(2000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(2000, "1");
     let mut reader = db.new_inbox_reader(&qid);
     let items = collect_cursor(&mut reader);
 
@@ -173,11 +161,7 @@ fn verify_ordering_within_same_priority(db: &crate::PartitionDb) {
 
 /// Test: Running and inbox stages are separate namespaces.
 async fn running_and_inbox_are_separate<W: WriteVQueueTable>(txn: &mut W) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(3000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(3000, "1");
 
     // Put one entry in Run stage
     let run_entry = entry_card(10);
@@ -191,11 +175,7 @@ async fn running_and_inbox_are_separate<W: WriteVQueueTable>(txn: &mut W) {
 }
 
 fn verify_running_and_inbox_are_separate(db: &crate::PartitionDb) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(3000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(3000, "1");
 
     // Running reader should only see the Run stage entry
     let mut run_reader = db.new_run_reader(&qid);
@@ -217,11 +197,7 @@ fn verify_running_and_inbox_are_separate(db: &crate::PartitionDb) {
 
 /// Test: seek_after positions cursor strictly after the given item.
 async fn seek_after_works<W: WriteVQueueTable>(txn: &mut W) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(4000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(4000, "1");
 
     // Insert entries in order
     let entries: Vec<_> = (1..=5).map(entry_card).collect();
@@ -231,11 +207,7 @@ async fn seek_after_works<W: WriteVQueueTable>(txn: &mut W) {
 }
 
 fn verify_seek_after_works(db: &crate::PartitionDb) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(4000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(4000, "1");
 
     let entries: Vec<_> = (1..=5).map(entry_card).collect();
 
@@ -255,11 +227,7 @@ fn verify_seek_after_works(db: &crate::PartitionDb) {
 
 /// Test: Empty queue returns None from peek.
 fn verify_empty_queue_returns_none(db: &crate::PartitionDb) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(99),
-        PartitionKey::from(9999u64), // unused partition key
-        VQueueInstance::from_raw(99),
-    );
+    let qid = VQueueId::custom(9999, "99");
 
     let mut run_reader = db.new_run_reader(&qid);
     run_reader.seek_to_first();
@@ -279,18 +247,9 @@ fn verify_empty_queue_returns_none(db: &crate::PartitionDb) {
 /// Test: Different vqueues (different parent/instance) are isolated.
 async fn vqueue_isolation<W: WriteVQueueTable>(txn: &mut W) {
     let pkey = PartitionKey::from(5000u64);
-
-    let qid1 = VQueueId::new(VQueueParent::from_raw(1), pkey, VQueueInstance::from_raw(1));
-    let qid2 = VQueueId::new(
-        VQueueParent::from_raw(1),
-        pkey,
-        VQueueInstance::from_raw(2), // different instance
-    );
-    let qid3 = VQueueId::new(
-        VQueueParent::from_raw(2), // different parent
-        pkey,
-        VQueueInstance::from_raw(1),
-    );
+    let qid1 = VQueueId::custom(pkey, "1");
+    let qid2 = VQueueId::custom(pkey, "2");
+    let qid3 = VQueueId::custom(pkey, "3");
 
     txn.put_inbox_entry(&qid1, Stage::Inbox, &entry_card(1));
     txn.put_inbox_entry(&qid2, Stage::Inbox, &entry_card(2));
@@ -300,9 +259,9 @@ async fn vqueue_isolation<W: WriteVQueueTable>(txn: &mut W) {
 fn verify_vqueue_isolation(db: &crate::PartitionDb) {
     let pkey = PartitionKey::from(5000u64);
 
-    let qid1 = VQueueId::new(VQueueParent::from_raw(1), pkey, VQueueInstance::from_raw(1));
-    let qid2 = VQueueId::new(VQueueParent::from_raw(1), pkey, VQueueInstance::from_raw(2));
-    let qid3 = VQueueId::new(VQueueParent::from_raw(2), pkey, VQueueInstance::from_raw(1));
+    let qid1 = VQueueId::custom(pkey, "1");
+    let qid2 = VQueueId::custom(pkey, "2");
+    let qid3 = VQueueId::custom(pkey, "3");
 
     // Each queue should only see its own entry
     let mut reader1 = db.new_inbox_reader(&qid1);
@@ -326,11 +285,7 @@ fn verify_vqueue_isolation(db: &crate::PartitionDb) {
 /// This verifies that the inbox reader (which uses a tailing iterator) can see
 /// items that were added after the reader was created, when re-seeking.
 async fn tailing_iterator_sees_new_items_on_reseek(rocksdb: &mut PartitionStore) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(6000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(6000, "1");
 
     // Insert initial entries
     let entry1 = entry_card(1);
@@ -388,11 +343,7 @@ async fn tailing_iterator_sees_new_items_on_reseek(rocksdb: &mut PartitionStore)
 /// This verifies that when using seek_after to resume iteration, newly added
 /// items that sort after the seek position are visible.
 async fn tailing_iterator_sees_new_items_on_seek_after(rocksdb: &mut PartitionStore) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(7000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(7000, "1");
 
     // Insert initial entries with different priorities
     let entry_high = entry_card_with_priority(1, EffectivePriority::System);
@@ -440,11 +391,7 @@ async fn tailing_iterator_sees_new_items_on_seek_after(rocksdb: &mut PartitionSt
 /// This verifies that when an item is deleted while the reader is open,
 /// re-seeking will not return the deleted item.
 async fn deleted_items_not_visible_after_reseek(rocksdb: &mut PartitionStore) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(8000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(8000, "1");
 
     // Insert initial entries
     let entry1 = entry_card(1);
@@ -492,11 +439,7 @@ async fn deleted_items_not_visible_after_reseek(rocksdb: &mut PartitionStore) {
 /// This verifies that deleted items are not returned when using seek_after
 /// to resume iteration past a certain point.
 async fn deleted_items_not_visible_after_seek_after(rocksdb: &mut PartitionStore) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(8500u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(8500, "1");
 
     // Insert entries
     let entry1 = entry_card(1);
@@ -540,11 +483,7 @@ async fn deleted_items_not_visible_after_seek_after(rocksdb: &mut PartitionStore
 /// This tests a more complex scenario where items are both added and removed
 /// while the reader is open.
 async fn concurrent_enqueue_and_delete(rocksdb: &mut PartitionStore) {
-    let qid = VQueueId::new(
-        VQueueParent::from_raw(1),
-        PartitionKey::from(9000u64),
-        VQueueInstance::from_raw(1),
-    );
+    let qid = VQueueId::custom(9000, "1");
 
     // Insert initial entries with different priorities
     let entry_high = entry_card_with_priority(10, EffectivePriority::System);

crates/partition-store/src/tests/vqueue_table_test/mod.rs

@@ -13,7 +13,7 @@ use restate_storage_api::vqueue_table::{
 };
 use restate_types::clock::UniqueTimestamp;
 use restate_types::identifiers::{InvocationId, PartitionKey, WithPartitionKey as _};
-use restate_types::vqueue::{EffectivePriority, VQueueId, VQueueInstance, VQueueParent};
+use restate_types::vqueue::{EffectivePriority, VQueueId};
 
 use crate::TableKind;
 use crate::keys::{EncodeTableKey, KeyKind, define_table_key};
@@ -61,25 +61,22 @@ impl From<&InvocationId> for EntryStateKey {
 
 #[derive(Debug, Clone, PartialEq, bilrost::Message)]
 pub struct EntryStateHeader {
+    #[bilrost(1)]
+    pub qid: VQueueId,
     /// Unknown is an invalid state, this will be set to None when the invocation
     /// leaves the queue.
-    #[bilrost(1)]
-    pub stage: Stage,
     #[bilrost(2)]
-    pub queue_parent: u32,
-    #[bilrost(3)]
-    pub queue_instance: u32,
+    pub stage: Stage,
     // current entry card details
-    #[bilrost(4)]
+    #[bilrost(3)]
     pub effective_priority: EffectivePriority,
-    #[bilrost(5)]
+    #[bilrost(4)]
     pub visible_at: VisibleAt,
-    #[bilrost(6)]
+    #[bilrost(5)]
     pub created_at: UniqueTimestamp,
 }
 
 pub struct OwnedHeader {
-    pub(crate) partition_key: PartitionKey,
     pub(crate) kind: EntryKind,
     pub(crate) id: EntryId,
 
@@ -91,24 +88,12 @@ impl AsEntryStateHeader for OwnedHeader {
         self.kind
     }
 
-    fn stage(&self) -> Stage {
-        self.inner.stage
-    }
-
-    fn queue_parent(&self) -> VQueueParent {
-        VQueueParent::from_raw(self.inner.queue_parent)
-    }
-
-    fn queue_instance(&self) -> VQueueInstance {
-        VQueueInstance::from_raw(self.inner.queue_instance)
+    fn vqueue_id(&self) -> &VQueueId {
+        &self.inner.qid
     }
 
-    fn vqueue_id(&self) -> VQueueId {
-        VQueueId::new(
-            self.queue_parent(),
-            self.partition_key,
-            self.queue_instance(),
-        )
+    fn stage(&self) -> Stage {
+        self.inner.stage
     }
 
     fn current_entry_card(&self) -> EntryCard {
@@ -136,15 +121,7 @@ impl<E> AsEntryStateHeader for OwnedEntryState<E> {
         self.header.stage()
     }
 
-    fn queue_parent(&self) -> VQueueParent {
-        self.header.queue_parent()
-    }
-
-    fn queue_instance(&self) -> VQueueInstance {
-        self.header.queue_instance()
-    }
-
-    fn vqueue_id(&self) -> VQueueId {
+    fn vqueue_id(&self) -> &VQueueId {
         self.header.vqueue_id()
     }
 
@@ -180,20 +157,8 @@ impl<E> AsEntryState for OwnedEntryState<E> {
 //         self.inner.stage
 //     }
 //
-//     fn queue_parent(&self) -> VQueueParent {
-//         VQueueParent::from_raw(self.inner.queue_parent)
-//     }
-//
-//     fn queue_instance(&self) -> VQueueInstance {
-//         VQueueInstance::from_raw(self.inner.queue_instance)
-//     }
-//
-//     fn vqueue_id(&self) -> VQueueId {
-//         VQueueId::new(
-//             self.queue_parent(),
-//             self.partition_key,
-//             self.queue_instance(),
-//         )
+//     fn vqueue_id(&self) -> &VQueueId {
+//         self.vqueue_id
 //     }
 //
 //     fn current_entry_card(&self) -> EntryCard {