[testnet] Add missing app ID; submit task outcomes in the correct order. (#5425) (#5427)

Backport of #5425.

## Motivation

Some apps may expect task outcomes to be submitted in the order in which
they were requested.

## Proposal

Order them correctly in the task processor.

Also, add the application ID when rescheduling a query.

## Test Plan

A test was added.

## Release Plan

- Release a new SDK.

## Links

- PR to main: #5425
- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: afck

examples/task-processor/src/service.rs

@@ -70,6 +70,16 @@ impl QueryRoot {
         *self.state.task_count.get()
     }
 
+    /// Returns the stored results in order.
+    async fn results(&self) -> Vec<String> {
+        let count = self.state.results.count();
+        self.state
+            .results
+            .read_front(count)
+            .await
+            .unwrap_or_default()
+    }
+
     /// Returns the pending tasks and callback requests for the task processor.
     async fn next_actions(
         &self,

linera-service/src/cli_wrappers/wallet.rs

@@ -1839,6 +1839,15 @@ impl<A> ApplicationWrapper<A> {
         self.run_graphql_query(&format!("mutation {{ {mutation} }}"))
             .await
     }
+
+    pub async fn multiple_mutate(&self, mutations: &[String]) -> Result<Value> {
+        let mut out = String::from("mutation {\n");
+        for (index, mutation) in mutations.iter().enumerate() {
+            out = format!("{}  u{}: {}\n", out, index, mutation);
+        }
+        out.push_str("}\n");
+        self.run_graphql_query(&out).await
+    }
 }
 
 impl<A> From<String> for ApplicationWrapper<A> {

linera-service/src/task_processor.rs

@@ -14,15 +14,15 @@ use std::{
 };
 
 use async_graphql::InputType as _;
-use futures::{stream::StreamExt, FutureExt};
+use futures::{future, stream::StreamExt, FutureExt};
 use linera_base::{
     data_types::{TimeDelta, Timestamp},
     identifiers::{ApplicationId, ChainId},
     task_processor::{ProcessorActions, TaskOutcome},
 };
 use linera_core::{client::ChainClient, node::NotificationStream, worker::Reason};
 use serde_json::json;
-use tokio::{io::AsyncWriteExt, process::Command, select, sync::mpsc};
+use tokio::{io::AsyncWriteExt, process::Command, select, sync::mpsc, task::JoinHandle};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info};
 
@@ -56,6 +56,7 @@ pub struct TaskProcessor<Env: linera_core::Environment> {
     update_receiver: mpsc::UnboundedReceiver<Update>,
     deadlines: BinaryHeap<Deadline>,
     operators: OperatorMap,
+    last_task_handles: BTreeMap<ApplicationId, JoinHandle<()>>,
 }
 
 impl<Env: linera_core::Environment> TaskProcessor<Env> {
@@ -77,12 +78,13 @@ impl<Env: linera_core::Environment> TaskProcessor<Env> {
             last_requested_callbacks: BTreeMap::new(),
             chain_client,
             cancellation_token,
+            notifications,
             outcome_sender,
             outcome_receiver,
-            notifications,
+            update_receiver,
             deadlines: BinaryHeap::new(),
             operators,
-            update_receiver,
+            last_task_handles: BTreeMap::new(),
         }
     }
 
@@ -144,12 +146,12 @@ impl<Env: linera_core::Environment> TaskProcessor<Env> {
         self.application_ids = update.application_ids;
 
         // Process actions for newly added applications
-        let new_apps: Vec<_> = self
+        let new_apps = self
             .application_ids
             .iter()
             .filter(|app_id| !old_app_set.contains(app_id))
             .cloned()
-            .collect();
+            .collect::<Vec<_>>();
         if !new_apps.is_empty() {
             self.process_actions(new_apps).await;
         }
@@ -188,7 +190,7 @@ impl<Env: linera_core::Environment> TaskProcessor<Env> {
                     // Retry in at most 1 minute.
                     self.deadlines.push(Reverse((
                         now.saturating_add(TimeDelta::from_secs(60)),
-                        None,
+                        Some(application_id),
                     )));
                     continue;
                 }
@@ -198,22 +200,51 @@ impl<Env: linera_core::Environment> TaskProcessor<Env> {
                 self.deadlines
                     .push(Reverse((timestamp, Some(application_id))));
             }
-            for task in actions.execute_tasks {
+            if !actions.execute_tasks.is_empty() {
                 let sender = self.outcome_sender.clone();
                 let operators = self.operators.clone();
-                tokio::spawn(async move {
-                    if let Err(e) = Self::execute_task(
-                        application_id,
-                        task.operator,
-                        task.input,
-                        sender,
-                        operators,
-                    )
-                    .await
-                    {
-                        error!("Error executing task for {application_id}: {e}");
+                let previous = self.last_task_handles.remove(&application_id);
+                let handle = tokio::spawn(async move {
+                    // Spawn all tasks concurrently and join them.
+                    let handles: Vec<_> = actions
+                        .execute_tasks
+                        .into_iter()
+                        .map(|task| {
+                            let operators = operators.clone();
+                            tokio::spawn(Self::execute_task(
+                                application_id,
+                                task.operator,
+                                task.input,
+                                operators,
+                            ))
+                        })
+                        .collect();
+                    let results = future::join_all(handles).await;
+                    // Wait for the previous batch to finish sending outcomes first.
+                    if let Some(previous) = previous {
+                        if let Err(error) = previous.await {
+                            error!(%application_id, %error, "Task panicked");
+                        }
+                    }
+                    // Submit outcomes in the original order.
+                    for result in results {
+                        match result {
+                            Ok(Ok(outcome)) => {
+                                if sender.send((application_id, outcome)).is_err() {
+                                    error!("Outcome receiver dropped for {application_id}");
+                                    break;
+                                }
+                            }
+                            Ok(Err(error)) => {
+                                error!(%application_id, %error, "Error executing task");
+                            }
+                            Err(error) => {
+                                error!(%application_id, %error, "Task panicked");
+                            }
+                        }
                     }
                 });
+                self.last_task_handles.insert(application_id, handle);
             }
         }
     }
@@ -222,9 +253,8 @@ impl<Env: linera_core::Environment> TaskProcessor<Env> {
         application_id: ApplicationId,
         operator: String,
         input: String,
-        sender: mpsc::UnboundedSender<(ApplicationId, TaskOutcome)>,
         operators: OperatorMap,
-    ) -> Result<(), anyhow::Error> {
+    ) -> Result<TaskOutcome, anyhow::Error> {
         let binary_path = operators
             .get(&operator)
             .ok_or_else(|| anyhow::anyhow!("unsupported operator: {}", operator))?;
@@ -250,8 +280,7 @@ impl<Env: linera_core::Environment> TaskProcessor<Env> {
             output: String::from_utf8_lossy(&output.stdout).into(),
         };
         debug!("Done executing task for {application_id}");
-        sender.send((application_id, outcome))?;
-        Ok(())
+        Ok(outcome)
     }
 
     async fn query_actions(

linera-service/tests/local_net_tests.rs

@@ -1389,6 +1389,102 @@ async fn test_node_service_with_task_processor() -> Result<()> {
     Ok(())
 }
 
+/// Test that task processor outcomes are submitted in the order they were requested,
+/// even when a later task finishes before an earlier one.
+#[cfg(feature = "storage-service")]
+#[test_log::test(tokio::test)]
+async fn test_task_processor_outcome_ordering() -> Result<()> {
+    use std::{io::Write, os::unix::fs::PermissionsExt};
+
+    use linera_base::{abi::ContractAbi, identifiers::ApplicationId};
+
+    struct TaskProcessorAbi;
+
+    impl ContractAbi for TaskProcessorAbi {
+        type Operation = ();
+        type Response = ();
+    }
+
+    let _guard = INTEGRATION_TEST_GUARD.lock().await;
+    tracing::info!("Starting test {}", test_name!());
+
+    let config = LocalNetConfig::new_test(Database::Service, Network::Grpc);
+    let (mut net, client) = config.instantiate().await?;
+    let chain = client.load_wallet()?.default_chain().unwrap();
+
+    // Publish and create the task-processor example application.
+    let example_dir = ClientWrapper::example_path("task-processor")?;
+    let app_id_str = client
+        .project_publish(example_dir, vec![], None, &())
+        .await?;
+    let app_id: ApplicationId = app_id_str.trim().parse()?;
+
+    // Create a slow operator that sleeps before echoing its input.
+    let tmp_dir = tempfile::tempdir()?;
+    let slow_path = tmp_dir.path().join("slow-operator");
+    {
+        let mut file = std::fs::File::create(&slow_path)?;
+        writeln!(file, "#!/bin/sh")?;
+        writeln!(file, "sleep 1")?;
+        writeln!(file, "cat")?;
+    }
+    std::fs::set_permissions(&slow_path, std::fs::Permissions::from_mode(0o755))?;
+
+    // Create a fast operator that echoes immediately.
+    let fast_path = tmp_dir.path().join("fast-operator");
+    {
+        let mut file = std::fs::File::create(&fast_path)?;
+        writeln!(file, "#!/bin/sh")?;
+        writeln!(file, "cat")?;
+    }
+    std::fs::set_permissions(&fast_path, std::fs::Permissions::from_mode(0o755))?;
+
+    // Start the node service with both operators.
+    let port = get_node_port().await;
+    let operators = vec![
+        ("slow".to_string(), slow_path),
+        ("fast".to_string(), fast_path),
+    ];
+    let mut node_service = client
+        .run_node_service_with_options(port, ProcessInbox::Skip, &[app_id], &operators, false)
+        .await?;
+
+    node_service.ensure_is_running()?;
+
+    // Subscribe to notifications for the chain.
+    let mut notifications = Box::pin(node_service.notifications(chain).await?);
+
+    let app = node_service.make_application(&chain, &app_id.with_abi::<TaskProcessorAbi>())?;
+
+    // Submit both tasks in a single block: slow first, then fast.
+    // The slow task takes longer but should have its result stored first.
+    app.multiple_mutate(&[
+        r#"requestTask(operator: "slow", input: "slow_result")"#.to_string(),
+        r#"requestTask(operator: "fast", input: "fast_result")"#.to_string(),
+    ])
+    .await?;
+
+    // Wait for the block containing the RequestTask operations.
+    notifications.wait_for_block(None).await?;
+
+    // Wait for the two blocks containing the StoreResult operations.
+    notifications.wait_for_block(None).await?;
+    notifications.wait_for_block(None).await?;
+
+    let task_count: u64 = app.query_json("taskCount").await?;
+    assert_eq!(task_count, 2);
+
+    // Verify the results are in request order (slow first, fast second),
+    // not completion order (which would be fast first).
+    let results: Vec<String> = app.query_json("results").await?;
+    assert_eq!(results, vec!["slow_result", "fast_result"]);
+
+    net.ensure_is_running().await?;
+    net.terminate().await?;
+
+    Ok(())
+}
+
 /// Test that the node service read-only mode disables mutations and prevents query-triggered operations.
 #[cfg(feature = "storage-service")]
 #[test_log::test(tokio::test)]