Action times out if client is listening

[chrisstaite-menlo]

The addition of client_action_timeout_s means that an action times out if a client isn't actively polling for it. However, the RBE WaitExecution allows passive polling of an action. This is not considered and therefore long actions time out an signal down this channel that they have timed out. If there is an active WaitExecution for an action then it should not be subject to the client action timeout.

[MarcusSorealheis]

Wow! You found the smoking gun (I think). Thank you so much Chris!

We will look into this right away.

[chrisstaite-menlo]

I think this is handled by the code in OperationSubscriber::changed for store_awaited_action_db, however this same logic doesn't apply in memory_awaited_action_db.

[chrisstaite-menlo]

This might fix it:

diff --git a/nativelink-scheduler/src/memory_awaited_action_db.rs b/nativelink-scheduler/src/memory_awaited_action_db.rs
index ef703329..55b797a0 100644
--- a/nativelink-scheduler/src/memory_awaited_action_db.rs
+++ b/nativelink-scheduler/src/memory_awaited_action_db.rs
@@ -452,11 +452,21 @@ impl<I: InstantWrapper, NowFn: Fn() -> I + Clone + Send + Sync> AwaitedActionDbI
                     }
                 }
                 ActionEvent::ClientKeepAlive(client_id) => {
-                    let maybe_size = self
+                    if let Some(client_awaited_action) = self
                         .client_operation_to_awaited_action
-                        .size_for_key(&client_id)
-                        .await;
-                    if maybe_size.is_none() {
+                        .get(&client_id)
+                        .await
+                    {
+                        if let Some(awaited_action_sender) = self
+                            .operation_id_to_awaited_action
+                            .get(&client_awaited_action.operation_id)
+                        {
+                            awaited_action_sender.send_if_modified(|awaited_action| {
+                                awaited_action.update_client_keep_alive((self.now_fn)().now());
+                                false
+                            });
+                        }
+                    } else {
                         event!(
                             Level::ERROR,
                             ?client_id,