(self, worker: JoinHandle<()>, forwarder: JoinHandle<()>)
| 246 | } |
| 247 | |
| 248 | pub(super) fn wrap(self, worker: JoinHandle<()>, forwarder: JoinHandle<()>) -> JoinHandle<()> { |
| 249 | tokio::spawn(async move { |
| 250 | let _ = worker.await; |
| 251 | let _ = forwarder.await; |
| 252 | if self |
| 253 | .should_auto_save |
| 254 | .load(std::sync::atomic::Ordering::Acquire) |
| 255 | { |
| 256 | if let Some(persistence) = &self.persistence { |
| 257 | persistence.auto_save_if_enabled().await; |
| 258 | } |
| 259 | } |
| 260 | // Stream run reached a terminal state in-process (worker + |
| 261 | // forwarder both joined) — drop its loop checkpoint. Only a |
| 262 | // crash (this task never completes) leaves one for resume. |
| 263 | if let Some(persistence) = &self.persistence { |
| 264 | persistence |
| 265 | .clear_loop_checkpoint(self.cleanup.run_id()) |
| 266 | .await; |
| 267 | } |
| 268 | self.cleanup.clear_cancel_token().await; |
| 269 | self.cleanup.finish().await; |
| 270 | }) |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | #[cfg(test)] |
no test coverage detected