Bug: Reception of multiple events doesn't work

[artemsheinacn]

Description

@bharatGoswami8 @darkwisebear
It looks like com is not able to send/receive multiple events and/or events are being lost or overwritten.

[2026/03/26 10:49:04.9920876][266283][adas-secondary][DEBUG] Creating mw com producer for /feo/com/AgentAgt102ToScheduler...
[2026/03/26 10:49:05.0956654][266283][adas-secondary][DEBUG] Sending Hello for 4...
[2026/03/26 10:49:05.0963579][266283][adas-secondary][DEBUG] Sending Hello for 5...

[2026/03/26 10:49:03.6916338][264651][adas-primary][DEBUG] Creating mw com consumer for /feo/com/AgentAgt102ToScheduler...
[2026/03/26 10:49:05.1177299][264651][adas-primary][DEBUG] Got signal ActivityHello(5)

The hello signal from activity 4 is lost despite being sent after consumer has been connected. I'm using:

subscription
            .receive(
                self.sample_container
                    .take()
                    .expect("concurrent receive calls detected on the same subscription"),
                1,
                1,
            )
            .await

to receive events one-by-one and SampleContainer configured to hold 10 samples (SampleContainer::new(10)), also config file specifies 20 sample slots.

"events": [
            {
              "eventName": "signal",
              "numberOfSampleSlots": 20,
              "maxSubscribers": 1
            }
          ]

What could be the issue here?

Error Occurrence Rate

Reproducible

How to reproduce

I can try to reproduce it in com-api-example if that helps.

Supporting Information

No response

Classification

minor

Affected Version

0.5

Expected Closure Version

0.5

Category

• Safety Related
• Security Related

[bharatGoswami8]

Hi @artemsheinacn ,
Could you please share the branch where I can view the code and run the application? This would help me check the issue promptly.

[artemsheinacn]

I made a patch which reproduces the issue which is I think at least similar to what I'm having. Run it with:

bazelisk run //score/mw/com/example/com-api-example:com-api-example

And you should see it failed, because it has received 2 events when the API call sets max_samples for try_receive to just 1.

diff --git a/score/mw/com/example/com-api-example/basic-consumer-producer.rs b/score/mw/com/example/com-api-example/basic-consumer-producer.rs
index 1c2d60b3..2cb05dc9 100644
--- a/score/mw/com/example/com-api-example/basic-consumer-producer.rs
+++ b/score/mw/com/example/com-api-example/basic-consumer-producer.rs
@@ -84,10 +84,8 @@ impl<R: Runtime> VehicleMonitor<R> {
     /// # Panics
     ///
     /// Panics if unwrapping the sample from the buffer fails.
-    pub fn read_tire_data(&self) -> Result<String> {
-        let mut sample_buf = SampleContainer::new(3);
-
-        match self.tire_subscriber.try_receive(&mut sample_buf, 1) {
+    pub fn read_tire_data<'a>(&'a self, sample_buf: &mut SampleContainer<<<<R as Runtime>::Subscriber<Tire> as Subscriber<Tire, R>>::Subscription as Subscription<Tire, R>>::Sample<'a>>) -> Result<String> {
+        match self.tire_subscriber.try_receive(sample_buf, 1) {
             Ok(0) => Err(Error::Fail),
             Ok(x) => {
                 let sample = sample_buf.pop_front().unwrap();
@@ -175,14 +173,24 @@ fn run_with_runtime<R: Runtime>(name: &str, runtime: &R) {
     let monitor = VehicleMonitor::new(consumer, producer).unwrap();
     let tire_pressure = 5.0;
     println!("Setting tire pressure to {tire_pressure}");
-    for i in 0..5 {
-        monitor
-            .write_tire_data(Tire {
-                pressure: tire_pressure + i as f32,
-            })
-            .unwrap();
-        let tire_data = monitor.read_tire_data().unwrap();
-        println!("{tire_data}");
+    {
+        let mut sample_buf = SampleContainer::new(3);
+        for i in 0..5 {
+            monitor
+                .write_tire_data(Tire {
+                    pressure: tire_pressure + i as f32,
+                })
+                .unwrap();
+            monitor
+                .write_tire_data(Tire {
+                    pressure: tire_pressure + i as f32,
+                })
+                .unwrap();
+            let tire_data = monitor.read_tire_data(&mut sample_buf).unwrap();
+            println!("{tire_data}");
+            let tire_data = monitor.read_tire_data(&mut sample_buf).unwrap();
+            println!("{tire_data}");
+        }
     }
     //unoffer returns producer back, so if needed it can be used further
     match monitor.producer.unoffer() {

[bharatGoswami8]

Hi @artemsheinacn,
I tested this using the COM‑API example (demo) application with the async flow, and I believe I understand the scenario you described. Please let me know if my understanding is correct:

The producer offers the service and the consumer subscribes to it. The producer then starts sending data. After ~2 seconds, I started the consumer to receive data. The latest data I received was 3, which corresponds to the third sample sent by the producer.
Your concern is that when the consumer calls the receive method, it should also receive the first two samples that were sent earlier.

Starting async subscription test with Lola runtime
[SENDER] Sent sample with pressure: 1.00 psi
[RECEIVER] Async data processor started
[SENDER] Sent sample with pressure: 2.00 psi
[SENDER] Sent sample with pressure: 3.00 psi
[RECEIVER] Received 1 samples
[RECEIVER]   Sample: 3.00 psi
[SENDER] Sent sample with pressure: 4.00 psi
[RECEIVER] Received 1 samples
[RECEIVER]   Sample: 4.00 psi
[SENDER] Sent sample with pressure: 5.00 psi
[RECEIVER] Received 1 samples
[RECEIVER]   Sample: 5.00 psi
[SENDER] Sent sample with pressure: 6.00 psi
[RECEIVER] Received 1 samples
[RECEIVER]   Sample: 6.00 psi
[SENDER] Sent sample with pressure: 7.00 psi
[RECEIVER] Received 1 samples
[RECEIVER]   Sample: 7.00 psi
[SENDER] Sent sample with pressure: 8.00 psi
[SENDER] Sent sample with pressure: 9.00 psi
[SENDER] Sent sample with pressure: 10.00 psi
Successfully unoffered the service
=== Async subscription test with Lola runtime completed ===

[artemsheinacn]

@bharatGoswami8 yes, exactly

[bharatGoswami8]

Hi @artemsheinacn , Checking issue with mw::com (LoLa)

[bharatGoswami8]

@artemsheinacn
LoLa returns samples ordered by timestamp, always starting from the newest available slot.

The internal algorithm (ReferenceNextEvent) works as follows:
It searches for the slot with the largest timestamp in the range (last_reference_time, upper_limit).
On each successive call within the same receive, upper_limit is narrowed to the previously found timestamp, so the next search returns the next-newest sample.
After the call completes, last_reference_time is advanced to the newest delivered timestamp, so those samples are never returned again.
So if the producer has sent samples with pressures 1.00 → 2.00 → 3.00 (T1 < T2 < T3) before the first receive, and max_samples = 2:

1st iteration → (0, MAX) → 3.00 (T3)
2nd iteration → (0, T3) → 2.00 (T2)
→ Delivered: 2.00 and 3.00 (the 2 newest). Sample 1.00 (T1) is permanently skipped because last_reference_time advances to T3 after this call.

So if you are expecting all the new sample then you need to request for max_sample to 20.
e.g. from shared code by you -

subscription
            .receive(
                self.sample_container
                    .take()
                    .expect("concurrent receive calls detected on the same subscription"),
                20,
                20,
            )
            .await

I will update the documentation in COM-API side for same.

[artemsheinacn]

@bharatGoswami8 So it means if I need to do reading one-by-one I have to make my own Stream on top of receive future which will yield every event returned with the last receive call, right?

[bharatGoswami8]

@artemsheinacn , We are in discussion to add stream API support in COM-API.

You can join coming Monday Rust-COM-API at 10 AM CET.

[bharatGoswami8]

@bharatGoswami8 So it means if I need to do reading one-by-one I have to make my own Stream on top of receive future which will yield every event returned with the last receive call, right?

But currently if you want to read all the latest available sample then you can make new_sample and max_sample parameter equal to numberOfSampleSlots

[artemsheinacn]

@bharatGoswami8 unfortunately I can't join at 10am on Monday.

But currently if you want to read all the latest available sample then you can make new_sample and max_sample parameter equal to numberOfSampleSlots

According to the documentation new_sample is a Minimum number of new events before resolution, so it means it won't get ready until I have 20 samples, but that's not what I need, I need minimum 1, so I will need to use new_sample=1, max_sample=20 and then deal with a collection of samples of variable length if I get it right.

Also, could you also take a look at the issue where max_samples argument is not respected in com-api-example with a patch I have above? Just so I understand why it doesn't work as expected

[bharatGoswami8]

Hi @artemsheinacn

According to the documentation new_sample is a Minimum number of new events before resolution, so it means it won't get ready until I have 20 samples, but that's not what I need, I need minimum 1, so I will need to use new_sample=1, max_sample=20 and then deal with a collection of samples of variable length if I get it right.

Yes Future of async will not resolve until Minimum Sample received and i think for this kind of use-case stream API will be helpful.

Also, could you also take a look at the issue where max_samples argument is not respected in com-api-example with a patch I have above? Just so I understand why it doesn't work as expected

if you mean to try_receive(sample_buf, 1) this API call then this is sync API which will return immediately returning sample up to Max_sample which is here 1, but it can return less also if that many new sample not available or return no new sample if not available.

So this method with polling you can use till stream API available, and the sample_buf can be updated by try_receive with latest sample returned.