feat: consumer lease streaming & message delivery (story 1.7)

[vieiralucas]

Summary

• Add consumer registry to scheduler with per-consumer tokio::sync::mpsc channels
• Implement try_deliver_pending: scans for unleased messages, creates lease + lease_expiry entries atomically via write_batch, delivers to consumers in round-robin
• Implement Lease gRPC RPC with two-channel bridge pattern (scheduler sends ReadyMessage, converter task maps to LeaseResponse)
• Automatic consumer unregistration on stream close via tokio::select! detecting both message arrival and client disconnection

Test plan

• Consumer receives enqueued messages immediately
• Consumer receives pending messages on registration (backlog drain)
• Lease creates entries in leases CF after delivery
• Multiple consumers receive different messages (no duplicates)
• Unregister consumer stops message delivery
• Enqueue 10 messages, lease receives all 10 in FIFO order
• 41 tests pass, clippy clean, cargo fmt clean

---

Summary by cubic

Implements a server‑streaming Lease RPC and a consumer registry to push messages instantly without polling. Adds per‑queue round‑robin delivery with lease‑first, atomic lease+expiry writes, per‑queue visibility timeouts (30s default), and auto cleanup on disconnect (Story 1.7).

• New Features

  • Lease RPC opens a server stream; maps ReadyMessage → LeaseResponse with metadata (fairness_key, weight, throttle_keys, attempt_count, queue_id); unregisters on disconnect via tokio::select.
  • Scheduler registers consumers via tokio mpsc; delivers on enqueue and on registration; scans for unleased messages; writes lease and expiry in one batch using the queue’s visibility_timeout_ms; persists a per‑queue round‑robin index across calls for fairness.

• Bug Fixes

  • Lease‑before‑send to prevent delivery without a lease; on full/closed channels roll back the lease and try next; if none available, keep the message; log list/write/rollback failures with consumer/message IDs; tests cover closed/full channel cases and lease rollback.

^{Written for commit 6e63f64. Summary will update on new commits.}

[cubic-dev-ai]

2 issues found across 6 files

Prompt for AI agents (all issues)

Check if these issues are valid — if so, understand the root cause of each and fix them.


<file name="crates/fila-server/src/service.rs">

<violation number="1" location="crates/fila-server/src/service.rs:43">
P2: `weight` is hardcoded to `0` and `throttle_keys` to `vec![]` instead of carrying over the original message values. The `ReadyMessage` struct is missing these fields, so the data is silently lost during delivery. Consumers will see incorrect metadata. Consider adding `weight` and `throttle_keys` to `ReadyMessage` and populating them from the source `Message`.</violation>
</file>

<file name="crates/fila-core/src/broker/scheduler.rs">

<violation number="1" location="crates/fila-core/src/broker/scheduler.rs:227">
P1: Round-robin `consumer_idx` is a local variable reset to 0 on every `try_deliver_pending` call. Since each enqueue triggers a separate call, messages arriving one-at-a-time will always be offered to the same consumer first, completely defeating the intended round-robin distribution. Promote `consumer_idx` (or a per-queue equivalent) to a `Scheduler` field so the position persists across invocations.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}

[vieiralucas]

@cubic-dev-ai full review

[cubic-dev-ai]

@cubic-dev-ai full review

@vieiralucas I have started the AI code review. It will take a few minutes to complete.

[cubic-dev-ai]

2 issues found across 6 files

Prompt for AI agents (all issues)

Check if these issues are valid — if so, understand the root cause of each and fix them.


<file name="crates/fila-core/src/broker/scheduler.rs">

<violation number="1" location="crates/fila-core/src/broker/scheduler.rs:28">
P2: Global `consumer_rr_idx` is shared across all queues, breaking per-queue round-robin fairness. Deliveries to queue A advance the index that queue B uses, causing uneven distribution. Additionally, `queue_consumers` is collected from a `HashMap` whose iteration order is non-deterministic — any consumer add/remove reshuffles the order, making the persisted index meaningless. Consider using a per-queue round-robin index (e.g., `HashMap<String, usize>`).</violation>

<violation number="2" location="crates/fila-core/src/broker/scheduler.rs:300">
P2: Silently ignoring lease rollback failures can orphan leases and block messages. If `write_batch` (delete) fails here, the lease persists and the message is stuck until natural lease expiry. At minimum, log an error so operators can detect this condition.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.}