fix(linux-core): keep a finished call from cancelling its caller

linux/crates/core/src/owned_calls.rs

@@ -112,9 +112,13 @@ impl OwnedCalls {
         });
 
         // Dropping the caller is itself a cancel: the work is already
-        // running on the server and has to be stopped, not forgotten.
-        let _guard = cancel.drop_guard();
-        match receiver.await {
+        // running on the server and has to be stopped, not forgotten. A
+        // call that returns leaves the token alone, since the caller may
+        // hand the same options to its next call.
+        let guard = cancel.drop_guard();
+        let received = receiver.await;
+        drop(guard.disarm());
+        match received {
             Ok(outcome) => outcome,
             Err(_) => {
                 spawned.abort();
@@ -190,8 +194,10 @@ impl OwnedCalls {
             let _ = sender.send(outcome);
         });
 
-        let _guard = cancel.drop_guard();
-        match receiver.await {
+        let guard = cancel.drop_guard();
+        let received = receiver.await;
+        drop(guard.disarm());
+        match received {
             Ok(outcome) => outcome,
             Err(_) => {
                 spawned.abort();
@@ -617,6 +623,44 @@ mod tests {
         );
     }
 
+    /// One operation hands the same options to each of its calls, as a
+    /// page does for its columns, its row identity and its rows.
+    #[tokio::test]
+    async fn a_finished_call_leaves_the_callers_options_usable() {
+        let calls = OwnedCalls::new(Duration::from_secs(1));
+        let (engine, _sent) = Engine::new(StopDelivery::OutOfBand(StopScope::Statement));
+        let cancel = CancellationToken::new();
+        let shared = options(None, &cancel);
+
+        let first = calls
+            .run_read(
+                shared.clone(),
+                Connection::new().0,
+                engine.clone(),
+                |resource, _stop| async move { (resource, Ok(1)) },
+            )
+            .await;
+        let write = calls
+            .run_write(
+                shared.clone(),
+                Connection::new().0,
+                engine.clone(),
+                Arc::new(WriteLedger::new(crate::write_ledger::WriteAtomicity::Transactional)),
+                |resource, _stop| async move { (resource, Ok(WriteReport::default())) },
+            )
+            .await;
+        let second = calls
+            .run_read(shared, Connection::new().0, engine, |resource, _stop| async move {
+                (resource, Ok(2))
+            })
+            .await;
+
+        assert_eq!(first.expect("the first call"), 1);
+        assert!(write.is_ok(), "{write:?}");
+        assert_eq!(second.expect("a later call with the same options"), 2);
+        assert!(!cancel.is_cancelled(), "a finished call cancelled its caller's token");
+    }
+
     #[tokio::test(start_paused = true)]
     async fn a_deadline_stops_the_work_and_reads_as_a_timeout() {
         let calls = OwnedCalls::new(Duration::from_secs(5));

linux/crates/test-support/src/conformance/cases/metadata.rs

@@ -44,3 +44,27 @@ pub async fn lists_created_table_with_primary_key_and_unique_identity<E: Conform
     connection.close().await?;
     Ok(())
 }
+
+const SHARED_CASE: &str = "calls_share_one_call_options";
+
+/// One operation hands the same options to each of its calls, as a page
+/// does for its columns, its row identity and its rows: a call that
+/// finishes leaves them usable for the next.
+pub async fn calls_share_one_call_options<E: ConformanceEngine>() -> Result<(), ConformanceFailure> {
+    let engine = E::start().await?;
+    let connection = connect(SHARED_CASE, &engine, writable()).await?;
+    let table = create_scratch_table(&engine, connection.as_ref(), "shared").await?;
+    let options = within(CALL_DEADLINE);
+
+    let columns = connection.fetch_columns(&table, options.clone()).await;
+    let identity = connection.row_identity(&table, options.clone()).await;
+    let objects = connection.list_objects(&ObjectScope::DefaultSchema, options).await;
+
+    expect(
+        SHARED_CASE,
+        columns.is_ok() && identity.is_ok() && objects.is_ok(),
+        || format!("calls sharing one set of options: columns {columns:?}, identity {identity:?}, objects {objects:?}"),
+    )?;
+    connection.close().await?;
+    Ok(())
+}

linux/crates/test-support/src/conformance/connection_suite_macro.rs

@@ -150,6 +150,11 @@ macro_rules! connection_conformance_suite {
             lists_created_table_with_primary_key_and_unique_identity,
             $crate::conformance::cases::metadata::lists_created_table_with_primary_key_and_unique_identity::<$engine>
         }
+        $crate::__conformance_case! {
+            $docker,
+            calls_share_one_call_options,
+            $crate::conformance::cases::metadata::calls_share_one_call_options::<$engine>
+        }
         $crate::__conformance_case! {
             $docker,
             query_bound_times_out_within_deadline_plus_cancel_grace,