Repository navigation
feat(orchestrator): keep per-state request membership records for listing a queue's requests by state - #891
Conversation
auditbenchai
left a comment
There was a problem hiding this comment.
No actionable comments were generated in this review. 🎉
Merge readiness: ⛔ Not ready to merge
Pre-merge checks: ✅ 5 · ❌ 1
❌ Failed checks: 1 error
| Check name | Status | Explanation | Resolution |
|---|---|---|---|
| CI checks | ❌ | 3 of 22 failing: Required Checks, Validator Extension Test, E2E Integration Test. | Open the failing runs on GitHub and fix them before merging. |
✅ Passed checks: 5 passed
| Check name | Status | Explanation |
|---|---|---|
| Merge conflicts | ✅ | Merges cleanly into chenghan.ying/sqv2-dashboard-metrics. |
| Secrets | ✅ | No keys, passwords or tokens in the changed code. |
| Blocking issues | ✅ | No critical or high-severity issues in the changed lines. |
| Tests | ✅ | Tests were changed alongside the code. |
| PR size | ✅ | 802 lines changed. |
ℹ️ Review info
- Commit:
296d400 - Verdict: ✅ Pass
- Findings: none
- Review: local checks plus AI review of the changed lines
- Scope: Reviewed 0/17 files (of 17 total) — 0 from cache, 0 needed no AI review. Found 0 finding(s), 0 potential secret(s).
💬 Reply to any inline comment and Audit Bench Ai will recheck that code and answer in the thread. Mention @auditbench in the conversation to ask about the whole PR.
🛠️ Open the full report and fix with AI
Thanks for using Audit Bench Ai! Free pull request reviews for open source: paste any public PR at auditbenchai.com/scan.
Reviewed by Audit Bench Ai · AI code review for security, logic and performance · Review any public PR free
296d400 to
a1159dc
Compare
| -- is a PK-prefix scan, so no secondary index is needed. Rows are written in the same | ||
| -- transaction as the request row they mirror and removed when the request reaches a | ||
| -- terminal state, so the table only ever holds in-queue requests. | ||
| CREATE TABLE IF NOT EXISTS queue_request_state ( |
There was a problem hiding this comment.
why is this in orchestrator? why do we need it? shouldn't this be in gateway later?
There was a problem hiding this comment.
we need it for the queue-depth panels on dashboard, like how many requests are in each state per queue right now, and how long the oldest has been waiting. Adding a new index will break the key-value storage contract so I prefer to add a new table.
gw is also fine, we can leverage the RequestLog to maintain the table, will do it in another PR
a1159dc to
26b97c0
Compare
…osen operation results
…ad of result=error
…l outcome, and end-to-end land time
…ed paths per batch as histograms
…issing row still counts as an error
… so process.* counter wildcards cannot match their buckets
…o a queue's requests can be listed by state
… with a no-op upsert so concurrent writes in one queue cannot deadlock
…lation level for rolling out membership records
26b97c0 to
b17440e
Compare
Why?
Operators need queue depth: how many requests are in each state, and how long the oldest one has been there. Event metrics cannot provide this. It is a property of stored state, and it matters most when a queue is stuck and no events are flowing.
The
requesttable is keyed by(queue, id), so "requests of queue Q in state S" is not a key-based read. Adding a(queue, state)secondary index would makeRequestStoredepend on a query-by-attribute, which the storage contract rules out. The sanctioned shape is a first-class mapping store keyed by the lookup attribute, the same pattern asQueueBatchStateandChangeRecord.This PR adds that mapping for requests. A periodic job can then report per-state request counts and the oldest time-in-state with key-prefix scans only.
What?
entity.QueueRequestState{Queue, State, RequestID, EnteredAtMs}: one record per in-queue request, under its current state.entity.InQueueRequestStates: every non-terminal request state. A test enforces that each valid state is either in this list or terminal.storage.QueueRequestStateStore.List(ctx, state), with its mock.GetQueueRequestStateStore()on the orchestrator storage aggregate.queue_request_state, PK(queue, state, request_id). It leads withqueue, as the queue-shard linter requires, and has no secondary index.Listis a primary-key-prefix scan.CreateandUpdateit deletes the request's record from every other in-queue state andINSERT IGNOREs it under the new one, so a version-only update keepsentered_at_ms. A terminal state leaves no record, so the table only ever holds in-queue requests. A failed CAS or duplicate create rolls back with no record change.mysql.NewStorage(db, scope, WithQueueRequestStateRecords())enables the records. It is off by default, so a deployment that upgrades before creating the table keeps working.Listreturns an error instead of reporting every queue as idle.AGENTS.md, and the store/table map indoc/rfc/service-scoped-extensions.md.Exact records maintained by the store, rather than advisory records maintained in core.**
QueueBatchStaterecords are moved bysubmitqueue/orchestrator/core/batcharound the batch CAS, are advisory, and need repair on redelivery. Here the backend keeps the record in the same transaction as the request write. That way:The cost is that a backend must be able to update the request and its record atomically. A key-value backend without local transactions would need its own write ordering.
Test Plan
TestRequestStore_WritesQueueRequestStateRecordscovers:startedin the same transactionTestQueueRequestStateStore_List: rows, empty, query error, row error.TestDisabledQueueRequestStateStore_ListFails.TestNewStorage_QueueRequestStateRecords: off by default, on with the option.TestInQueueRequestStates: every valid state is either in-queue or terminal.TestStorage_QueueRequestStateRecordsFollowRequestWritescovers:EnteredAtMsRun locally:
Issue