Check for eligibility on every removed item

Before, the DRR scheduler checked for eligibility if the current head item
was removed. This was not enough since a vqueue might have been not eligible
because it was running out of concurrency tokens. A removed item could have
released the concurrency token and thereby it could make the vqueue eligible
again if the vqueue contains more items. Hence, we should check on every
removed item whether this event had an impact on the vqueue's eligibility.

Author: tillrohrmann

crates/storage-api/src/vqueue_table/entry.rs

@@ -16,6 +16,7 @@ use restate_types::state_mut::ExternalStateMutation;
 use restate_types::vqueue::{
     EffectivePriority, NewEntryPriority, VQueueId, VQueueInstance, VQueueParent,
 };
+use std::fmt::{Debug, Formatter};
 
 use crate::StorageError;
 
@@ -36,7 +37,7 @@ pub enum EntryKind {
 
 // Using u128 would have added an extra unnecessary 8 bytes due to alignment
 // requirements (u128 is 0x10 aligned and it forces the struct to be 0x10 aligned)
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct EntryId([u8; 16]);
 
 impl EntryId {
@@ -61,6 +62,15 @@ impl EntryId {
     }
 }
 
+impl Debug for EntryId {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        // display inner field as u128 to make it a bit easier to read
+        f.debug_tuple("EntryId")
+            .field(&u128::from_be_bytes(self.0))
+            .finish()
+    }
+}
+
 impl From<&InvocationId> for EntryId {
     #[inline]
     fn from(id: &InvocationId) -> Self {

crates/types/src/clock/unique_timestamp.rs

@@ -8,6 +8,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::fmt::{Debug, Formatter};
 use std::num::NonZeroU64;
 use std::time::SystemTime;
 
@@ -50,7 +51,7 @@ pub enum Error {
 /// The timestamp is represented as a 64-bit unsigned integer. The upper 42 bits
 /// represent the physical time in milliseconds since [`RESTATE_EPOCH`], and the
 /// lower 22 bits represent the logical clock count.
-#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[derive(Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
 #[repr(transparent)]
 pub struct UniqueTimestamp(NonZeroU64);
 
@@ -132,6 +133,17 @@ impl UniqueTimestamp {
     }
 }
 
+impl Debug for UniqueTimestamp {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        // Output unique timestamp as a pair of physical and logical timestamps. Convert the physical
+        // timestamp from the RESTATE_EPOCH to the UNIX_EPOCH.
+        f.debug_struct("UniqueTimestamp")
+            .field("physical", &self.to_unix_millis().as_u64())
+            .field("logical", &self.logical_raw())
+            .finish()
+    }
+}
+
 impl From<SystemTime> for UniqueTimestamp {
     fn from(value: SystemTime) -> Self {
         // The assumption is that SystemTime will always be > RESTATE_EPOCH

crates/vqueues/src/scheduler/vqueue_state.rs

@@ -274,7 +274,7 @@ where
     }
 
     pub fn check_eligibility(
-        &mut self,
+        &self,
         now: UniqueTimestamp,
         meta: &VQueueMeta,
         config: &VQueueConfig,