(maka) branch rust-preview updated: fix(host): preserve cancellation across subscription observation

[wy...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git


The following commit(s) were added to refs/heads/rust-preview by this push:
     new ed5f000bc2 fix(host): preserve cancellation across subscription observation
ed5f000bc2 is described below

commit ed5f000bc29bddb4be82db71a4595a0e2ffb9d08
Author: Wang <m4...@gmail.com>
AuthorDate: Tue Oct 20 09:38:21 2026 +0800

    fix(host): preserve cancellation across subscription observation
    
    Keep a consumed Cancel for the next transport phase when the original observer still returns queued output. Verify that the peer sends only one Cancel and that final acknowledgement waits for original resource cleanup.
    
    Generated-by: Codex
---
 .../src/application/transport/server.rs            | 66 +++++++++++++++++++++-
 1 file changed, 65 insertions(+), 1 deletion(-)

diff --git a/platform/host-runtime/src/application/transport/server.rs b/platform/host-runtime/src/application/transport/server.rs
index a3d7b6d39c..2ee0b352c9 100644
--- a/platform/host-runtime/src/application/transport/server.rs
+++ b/platform/host-runtime/src/application/transport/server.rs
@@ -290,7 +290,11 @@ impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> Session<R, W> {
         tokio::select! {biased;
             ()=self.stop.cancelled()=>{self.lost=true;self.scope.as_ref().unwrap().cancellation.cancel();future.await},
             control=self.reader.read::<Control>()=>{
-                if !matches!(control,Ok(Control::Cancel)) {self.lost=true;}
+                if matches!(control,Ok(Control::Cancel)) {
+                    // The original observer may still return an accepted item.
+                    // Preserve the peer's stop for the next protocol phase.
+                    self.pending=Some(Control::Cancel);
+                } else {self.lost=true;}
                 self.scope.as_ref().unwrap().cancellation.cancel();future.await
             },
             result=&mut future=>result,
@@ -655,6 +659,66 @@ mod tests {
         observed.finish().await.unwrap();
     }
     #[tokio::test]
+    async fn cancellation_while_receiving_a_queued_item_survives_until_cleanup() {
+        let (input, mut command) = tokio::io::duplex(128);
+        let (output, reply) = tokio::io::duplex(1024);
+        let mut reply = Reader::new(reply);
+        let scope = scope();
+        let mut resource = scope.resources.reserve().unwrap();
+        resource.start();
+        let (mut items, emitter) = crate::application::subscriptions::items(
+            Arc::new(jsonschema::validator_for(&serde_json::json!({})).unwrap()),
+            scope.cancellation.clone(),
+        );
+        let value = serde_json::json!("already accepted");
+        emitter.emit(value.clone()).await.unwrap();
+        let mut session = Session {
+            reader: Reader::new(input),
+            writer: output,
+            scope: Some(scope.clone()),
+            stop: CancellationToken::new(),
+            lost: false,
+            pending: None,
+        };
+        // Both are ready on the same poll: cancellation wins observation,
+        // but the real subscription queue still returns its accepted item.
+        wire::write(&mut command, &Control::Cancel).await.unwrap();
+        assert_eq!(session.observe(items.recv()).await, Some(value.clone()));
+        assert!(scope.cancellation.is_cancelled());
+        assert!(!session.lost);
+        session
+            .send(&Reply::Item {
+                value: value.clone(),
+            })
+            .await
+            .unwrap();
+        assert!(matches!(
+            reply.read::<Reply>().await.unwrap(),
+            Reply::Item { value: received } if received == value
+        ));
+        // The peer has already sent its one Cancel and now only reads.
+        assert!(matches!(
+            session.control().now_or_never(),
+            Some(Ok(Control::Cancel))
+        ));
+        let finish = session.finish(Reply::Finished {
+            result: Ok(serde_json::Value::Null),
+            cleanup: Ok(()),
+        });
+        tokio::pin!(finish);
+        assert!(finish.as_mut().now_or_never().is_none());
+        resource.complete(Ok(()));
+        finish.await.unwrap();
+        assert!(matches!(
+            reply.read::<Reply>().await.unwrap(),
+            Reply::Finished {
+                cleanup: Ok(()),
+                ..
+            }
+        ));
+        scope.finish().await.unwrap();
+    }
+    #[tokio::test]
     async fn eof_revokes_but_does_not_drop_the_same_admission_observer() {
         let (input, peer) = tokio::io::duplex(64);
         let scope = scope();


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org