# Plan: Implementing `JobJoinHandle::await` (Job Result Awaiting)
## Goal
Allow callers who enqueue a job to await its result. `Queue::add()` and `Queue::add_with()` should return a `JobJoinHandle<D, R>` that can be `.await`ed to get the job's return value (or error).
```rust
let handle = queue.add("my-job", &data).await?;
let result: R = handle.waitUntilFinished().await?; // blocks until worker calls done()/failed()
```
## How BullMQ Does It
BullMQ's `job.waitUntilFinished(queueEvents, ttl?)` works in two parts:
1. **Event listener**: Registers listeners for `completed:{jobId}` and `failed:{jobId}` events on a `QueueEvents` instance. `QueueEvents` runs a loop calling `XREAD BLOCK` on the `bull:{queue}:events` Redis stream, parsing each entry and emitting per-job events.
2. **Poll fallback**: Immediately after registering the listener, it calls `isFinished-3.lua` to check if the job already completed (race condition where job finishes before listener attaches). If finished, resolves immediately with the stored `returnvalue` or `failedReason` from the job hash.
The completion event is already emitted by `moveToFinished-14.lua` (line 195):
```lua
rcall("XADD", eventStreamKey, "*", "event", ARGV[5], "jobId", jobId, ARGV[3], ARGV[4], "prev", "active")
-- ARGV[4] = JSON-serialized value
```
**This XADD already exists in the bullrs Lua scripts**, so no Lua changes are needed.
## Architecture
### Shared `QueueEvents` listener (recommended)
A single background task per queue reads from the events stream with `XREAD BLOCK` and dispatches results to waiting `JobJoinHandle`s via a shared map of oneshot channels.
```
Queue::add()
├─ enqueues job via Lua
├─ inserts tx into shared Arc<Mutex<HashMap<jobId, oneshot::Sender>>>
└─ returns JobJoinHandle { rx, ... }
QueueEvents background task (one per Queue):
loop {
XREAD BLOCK on bull:{queue}:events
for each entry where event == "completed" | "failed":
if waiters.remove(jobId) -> tx:
tx.send(result)
}
JobJoinHandle::await:
rx.await (+ optional timeout)
// fallback: if channel closed without value, poll Redis directly
```
## Implementation Steps
### Step 1: Add `isFinished` Lua script
Create `lua/commands/isFinished-3.lua` (copy from BullMQ). This is used for the race-condition fallback when a job completes before the events listener starts.
- **Keys**: `completed`, `failed`, `jobId` hash
- **Args**: `jobId`, `returnValue?` (1 = include value)
- **Returns**: `{status, value?}` where status: 0=not finished, 1=completed, 2=failed, -1=missing
Create `src/luacommands/is_finished.rs` implementing `InvokeLuaScript`.
### Step 2: Create `QueueEvents` struct
New file: `src/queue/events.rs`
```rust
pub(crate) struct QueueEvents {
/// Map of job IDs to oneshot senders waiting for results
waiters: Arc<Mutex<HashMap<String, oneshot::Sender<JobResult>>>>,
/// Handle to the background XREAD loop
task_handle: JoinHandle<()>,
}
enum JobResult {
Completed(String), // JSON return value
Failed(String), // error reason
}
```
The background task:
- Uses a dedicated Redis connection (not from the pool, or a separate pool connection)
- Calls `XREAD BLOCK 5000 STREAMS bull:{queue}:events {last_id}` in a loop
- Parses stream entries for `completed`/`failed` events
- Looks up `jobId` in `waiters` map, sends result via oneshot if found
- Starts with `last_id = "$"` (only new events) since we register waiters before enqueue returns
### Step 3: Integrate `QueueEvents` into `Queue`
Add a lazily-initialized `QueueEvents` to `Queue`:
```rust
pub struct Queue<D, R> {
name: QueueName,
pool: Pool,
events: OnceCell<QueueEvents>, // initialized on first add() that wants a handle
phantom: PhantomData<(D, R)>,
}
```
Or alternatively, always start it. The `OnceCell` approach avoids the overhead when nobody uses join handles.
### Step 4: Rework `JobJoinHandle`
```rust
pub struct JobJoinHandle<R> {
rx: oneshot::Receiver<JobResult>,
// Fallback fields for race condition check
queue_name: QueueName,
pool: Pool,
id: String,
phantom: PhantomData<R>,
}
impl<R: DeserializeOwned> JobJoinHandle<R> {
/// Wait for the job to complete and return its result.
pub async fn result(self) -> Result<R, JobAwaitError> {
match self.rx.await {
Ok(JobResult::Completed(json)) => Ok(serde_json::from_str(&json)?),
Ok(JobResult::Failed(reason)) => Err(JobAwaitError::JobFailed(reason)),
Err(_) => {
// Channel closed (events task dropped) - fall back to polling
self.poll_result().await
}
}
}
/// Wait with a timeout.
pub async fn result_timeout(self, timeout: Duration) -> Result<R, JobAwaitError> {
tokio::time::timeout(timeout, self.result()).await?
}
/// Fallback: check Redis directly via isFinished script
async fn poll_result(&self) -> Result<R, JobAwaitError> { ... }
}
```
**Note**: Drop the `D` generic from `JobJoinHandle` — it's not needed since we only read the return value, not the input data.
### Step 5: Update `Queue::add` / `Queue::add_with` return types
Change from `Result<String, AddJobErr>` to `Result<JobJoinHandle<R>, AddJobErr>`.
The job ID is still accessible via `handle.id()`. For callers who only need the ID, add `JobJoinHandle::id() -> &str`.
This is a **breaking change** to the public API. The `JobJoinHandle` should expose `.id()` so existing code can adapt easily:
```rust
// Before:
let id = queue.add("job", &data).await?;
// After:
let handle = queue.add("job", &data).await?;
let id = handle.id();
```
### Step 6: Error types
New error enum in `src/error.rs`:
```rust
pub enum JobAwaitError {
/// The job failed with the given reason
JobFailed(String),
/// The job was not found (removed or expired)
JobNotFound,
/// Timed out waiting for result
Timeout,
/// Failed to deserialize the return value
Deserialize(serde_json::Error),
/// Redis error during fallback poll
Redis(redis::RedisError),
/// The events listener was dropped before the job completed
EventsListenerDropped,
}
```
## Race Condition Handling
The critical race: job completes *before* the XREAD listener starts or *before* the waiter is registered.
**Solution**: Register the oneshot sender in the waiters map *before* the Lua enqueue script runs. Sequence:
1. `queue.add()` initializes `QueueEvents` if needed (starts XREAD loop)
2. Register `(jobId, tx)` in waiters map
3. Execute the add job Lua script
4. Return `JobJoinHandle { rx, ... }`
Since `XREAD` with `$` only gets new events, and the events listener is started before the job is enqueued, no events can be missed. If the events listener was started later, we'd need the `isFinished` fallback — keep it anyway for robustness (e.g., reconnection scenarios).
Actually, there's a subtlety: we don't know the `jobId` before enqueue (it's generated by Redis `INCR`). So we can't register the waiter before enqueue. The correct sequence is:
1. Ensure `QueueEvents` XREAD loop is running (reading from `$` = current stream tail)
2. Enqueue job → get `jobId`
3. Register `(jobId, tx)` in waiters map
4. **Immediately** call `isFinished` to check if the job already completed between steps 2 and 3
5. If already finished, send result through `tx` directly
6. Return `JobJoinHandle { rx, ... }`
This matches BullMQ's approach exactly.
## Files to Create/Modify
| `lua/commands/isFinished-3.lua` | Create | Lua script to check job status |
| `src/luacommands/is_finished.rs` | Create | Rust wrapper for isFinished script |
| `src/luacommands/mod.rs` | Modify | Register new script |
| `src/queue/events.rs` | Create | QueueEvents background stream reader |
| `src/queue/mod.rs` | Modify | Add events field to Queue |
| `src/job/join_handle.rs` | Modify | Full implementation |
| `src/job/mod.rs` | Modify | Update exports (drop D generic) |
| `src/queue/basics.rs` | Modify | Return JobJoinHandle from add/add_with |
| `src/error.rs` | Modify | Add JobAwaitError |
| `src/lib.rs` | Modify | Update re-exports |
## Dependencies
May need to add:
- `tokio::sync::oneshot` (already have tokio)
- `tokio::sync::OnceCell` or `std::sync::OnceLock` for lazy QueueEvents init
## Open Questions
1. **Should `Queue::add` always return `JobJoinHandle`, or should there be separate methods?** Returning it always is simpler and the overhead is one hashmap insert + oneshot channel allocation. Callers can just drop the handle if they don't need the result.
2. **Should `JobJoinHandle` implement `Future` directly?** Implementing `Future` would allow `handle.await` syntax. The alternative is `handle.result().await` which is more explicit and allows adding `.result_timeout()`. Recommend explicit `.result()` method — implementing `Future` correctly with the fallback poll is tricky and `.result()` is clearer.
3. **Cleanup of waiters for dropped handles**: If a `JobJoinHandle` is dropped without being awaited, the oneshot `tx` will linger in the map until the event arrives. Add cleanup: when the XREAD loop sends to a closed channel, just discard. Optionally, implement `Drop` on `JobJoinHandle` to remove itself from the map (requires sharing the map reference).
4. **`Queue` is currently `Clone`** — adding `OnceCell<QueueEvents>` with an `Arc` interior should preserve this. The events listener should be shared across clones. Consider wrapping the events state in `Arc` at the Queue level.