Skip to content

Implement Proper Async Task Management and Supervision#10

Description

@devfire

馃攧 CONCURRENCY NIGHTMARE

Location: main.rs:82-83

Problem

let _result = tokio::try_join!(udp_intake_handle, llm_processing_handle)?;

Issues

  • No task supervision or restart logic
  • Silent failures ignored
  • No graceful shutdown handling

Suggested Fix

// Implement proper task supervision
struct TaskSupervisor {
    tasks: HashMap<String, JoinHandle<()>>,
    shutdown: Arc<AtomicBool>,
}

impl TaskSupervisor {
    pub async fn supervise_task<F, Fut>(
        &mut self,
        name: &str,
        task_fn: F
    )
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<(), String>> + Send + 'static,
    {
        let shutdown = Arc::clone(&self.shutdown);
        let task_name = name.to_string();

        let handle = tokio::spawn(async move {
            loop {
                if shutdown.load(Ordering::Relaxed) {
                    break;
                }

                match task_fn().await {
                    Ok(()) => break,
                    Err(e) => {
                        error!("Task {} failed: {}", task_name, e);
                        tokio::time::sleep(Duration::from_millis(1000)).await;
                    }
                }
            }
        });

        self.tasks.insert(name.to_string(), handle);
    }

    pub async fn shutdown(self) {
        self.shutdown.store(true, Ordering::Relaxed);

        for (name, handle) in self.tasks {
            if let Err(e) = handle.await {
                error!("Task {} failed during shutdown: {}", name, e);
            }
        }
    }
}

Priority: High - Critical for system reliability and proper resource cleanup in production environments.

Metadata

Metadata

Assignees

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions