agentq 0.1.0

An embeddable job queue with idempotency keys, per-lane backpressure and bounded concurrency, for agent tool calls.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

use std::sync::Arc;

use tokio::sync::Semaphore;
use tokio::sync::mpsc::Receiver;

use crate::handle::Outcome;
use crate::job::Job;
use crate::state::{JobGuard, State, StateMap, WaiterMap};

pub(crate) fn spawn_worker(
    mut receiver: Receiver<Job>,
    semaphore: Arc<Semaphore>,
    statemap: StateMap,
    waiters: WaiterMap,
) {
    tokio::spawn(async move {
        while let Some(job) = receiver.recv().await {
            let Ok(permit) = semaphore.clone().acquire_owned().await else {
                break;
            };
            let statemap = statemap.clone();
            let waiters = waiters.clone();

            tokio::spawn(async move {
                let _permit = permit;
                {
                    let mut statemap = statemap.lock().unwrap_or_else(|e| e.into_inner());
                    statemap.insert(job.key.clone(), State::Processing);
                }

                let mut guard = JobGuard {
                    statemap,
                    waiters,
                    key: job.key.clone(),
                    outcome: None,
                };

                guard.outcome = Some(match (job.func)().await {
                    Ok(output) => Outcome::Completed { output },
                    Err(e) => Outcome::Failed {
                        reason: e.to_string(),
                    },
                });
            });
        }
    });
}