Log a warning for slow signal updates to subscribers

.github/workflows/kuksa_databroker_build.yml

@@ -24,6 +24,9 @@ on:
       QUAY_IO_USERNAME:
         required: true
   workflow_dispatch:
+  # Run every Sunday night to check regressions, for example from clippy
+  schedule:
+    - cron: "0 4 * * 0"
 
 # suffix to avoid cancellation when running from release workflow
 concurrency:

.gitignore

@@ -5,3 +5,5 @@ __pycache__
 databroker/thirdparty
 databroker-cli/thirdparty
 .venv/
+databroker/databroker_bin.cdx.json
+databroker-cli/databroker-cli_bin.cdx.json

databroker-cli/Cargo.toml

@@ -13,7 +13,7 @@
 
 [package]
 name = "databroker-cli"
-version = "0.5.0"
+version = "0.5.0-dev.1"
 authors = ["Eclipse KUKSA Project"]
 edition = "2021"
 license = "Apache-2.0"

databroker-proto/Cargo.toml

@@ -13,7 +13,7 @@
 
 [package]
 name = "databroker-proto"
-version = "0.5.0"
+version = "0.5.0-dev.1"
 authors = ["Eclipse KUKSA Project"]
 edition = "2021"
 license = "Apache-2.0"

databroker/Cargo.toml

@@ -13,7 +13,7 @@
 
 [package]
 name = "databroker"
-version = "0.5.0"
+version = "0.5.0-dev.1"
 authors = ["Eclipse KUKSA Project"]
 edition = "2021"
 license = "Apache-2.0"

databroker/src/broker.rs

@@ -1111,7 +1111,7 @@ pub enum EntryReadAccess<'a> {
     Err(&'a Metadata, ReadError),
 }
 
-impl<'a> EntryReadAccess<'a> {
+impl EntryReadAccess<'_> {
     pub fn datapoint(&self) -> Result<&Datapoint, ReadError> {
         match self {
             Self::Entry(entry) => Ok(&entry.datapoint),
@@ -1151,7 +1151,7 @@ pub struct EntryReadIterator<'a, 'b> {
     permissions: &'b Permissions,
 }
 
-impl<'a, 'b> Iterator for EntryReadIterator<'a, 'b> {
+impl<'a> Iterator for EntryReadIterator<'a, '_> {
     type Item = EntryReadAccess<'a>;
 
     #[inline]
@@ -1167,7 +1167,7 @@ impl<'a, 'b> Iterator for EntryReadIterator<'a, 'b> {
     }
 }
 
-impl<'a, 'b> DatabaseReadAccess<'a, 'b> {
+impl DatabaseReadAccess<'_, '_> {
     pub fn get_entry_by_id(&self, id: i32) -> Result<&Entry, ReadError> {
         match self.db.entries.get(&id) {
             Some(entry) => match self.permissions.can_read(&entry.metadata.path) {
@@ -1203,7 +1203,7 @@ impl<'a, 'b> DatabaseReadAccess<'a, 'b> {
     }
 }
 
-impl<'a, 'b> DatabaseWriteAccess<'a, 'b> {
+impl DatabaseWriteAccess<'_, '_> {
     pub fn update_by_path(
         &mut self,
         path: &str,
@@ -1397,7 +1397,7 @@ impl Database {
     }
 }
 
-impl<'a, 'b> query::CompilationInput for DatabaseReadAccess<'a, 'b> {
+impl query::CompilationInput for DatabaseReadAccess<'_, '_> {
     fn get_datapoint_type(&self, path: &str) -> Result<DataType, query::CompilationError> {
         match self.get_metadata_by_path(path) {
             Some(metadata) => Ok(metadata.data_type.to_owned()),
@@ -1670,10 +1670,13 @@ impl<'a, 'b> AuthorizedAccess<'a, 'b> {
             .await
             .add_change_subscription(subscription);
 
-        let stream = BroadcastStream::new(receiver).filter_map(|result| match result {
+        let stream = BroadcastStream::new(receiver).filter_map(move |result| match result {
             Ok(message) => Some(message),
             Err(err) => {
-                debug!("Lagged entries: {}", err);
+                warn!(
+                    "Slow subscriber with capacity {} lagged and missed signal updates: {}",
+                    channel_capacity, err
+                );
                 None
             }
         });

lib/Cargo.toml

@@ -4,7 +4,8 @@ resolver = "2"
 members = [
     "common",
     "kuksa",
-    "sdv"
+    "sdv",
+    "databroker-examples"
 ]
 
 [workspace.dependencies]