Rename WorkerShutdown to BumpStream, also handle shutdown() case (#1056)

* Rename WorkerShutdown to BumpStream, also handle shutdown() case

* Update comment

Author: yuandrew

crates/sdk-core/src/worker/mod.rs

@@ -266,9 +266,9 @@ impl WorkerTrait for Worker {
                 .unregister_worker(self.worker_instance_key);
         }
 
-        // Push a Worker Shutdown message to the workflow activation queue. This ensures that
+        // Push a BumpStream message to the workflow activation queue. This ensures that
         // any pending workflow activation polls will resolve, even if there are no other inputs.
-        self.workflows.send_worker_shutdown();
+        self.workflows.bump_stream();
 
         // Second, we want to stop polling of both activity and workflow tasks
         if let Some(atm) = self.at_task_mgr.as_ref() {

crates/sdk-core/src/worker/workflow/mod.rs

@@ -626,10 +626,10 @@ impl Workflows {
         rx
     }
 
-    /// Send a `WorkerShutdown` message to the workflow stream. This ensures that any pending
-    /// workflow activation polls will resolve during shutdown, even if there are no other inputs.
-    pub(super) fn send_worker_shutdown(&self) {
-        self.send_local(LocalInputs::WorkerShutdown);
+    /// Send a `BumpStream` message to the workflow stream. This ensures that any pending
+    /// workflow activation polls will resolve, like during shutdown, even if there are no other inputs.
+    pub(super) fn bump_stream(&self) {
+        self.send_local(LocalInputs::BumpStream);
     }
 
     /// Query the state of workflow management. Can return `None` if workflow state is shut down.
@@ -654,7 +654,7 @@ impl Workflows {
                 let mut interval = tokio::time::interval(Duration::from_millis(10));
                 loop {
                     interval.tick().await;
-                    let _ = self.get_state_info().await;
+                    self.bump_stream();
                 }
             });
             let (_, jh_res) = tokio::join!(

crates/sdk-core/src/worker/workflow/workflow_stream.rs

@@ -147,9 +147,9 @@ impl WFStream {
                                 });
                                 None
                             }
-                            LocalInputs::WorkerShutdown => {
-                                // Worker shutdown message is used to bump the stream to ensure
-                                // pending workflow activation polls resolve during shutdown
+                            LocalInputs::BumpStream => {
+                                // Ensures pending workflow activation polls resolve for
+                                // scenarios like during shutdown
                                 None
                             }
                         }
@@ -638,7 +638,7 @@ pub(super) enum LocalInputs {
     RequestEviction(RequestEvictMsg),
     HeartbeatTimeout(String),
     GetStateInfo(GetStateInfoMsg),
-    WorkerShutdown,
+    BumpStream,
 }
 impl LocalInputs {
     fn run_id(&self) -> Option<&str> {
@@ -649,7 +649,7 @@ impl LocalInputs {
             LocalInputs::PostActivation(pa) => &pa.run_id,
             LocalInputs::RequestEviction(re) => &re.run_id,
             LocalInputs::HeartbeatTimeout(hb) => hb,
-            LocalInputs::GetStateInfo(_) | LocalInputs::WorkerShutdown => return None,
+            LocalInputs::GetStateInfo(_) | LocalInputs::BumpStream => return None,
         })
     }
 }