feat(worker): add ordering lock, metrics, and health endpoints - #1196
Open
realcodesiman wants to merge 5 commits into
Open
realcodesiman wants to merge 5 commits into
realcodesiman wants to merge 5 commits into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements P1/P2 items B3, B6, B7, B8, B9 from the system-design-review audit.
B3 — conversation ordering lock
The
integrationqueue (concurrency 10-200) could process two webhook deliveries for the same conversation concurrently, risking out-of-order message persistence (a documented race already referenced by anisEchoOfOwnSendcomment). Wraps the conversation-mutating section ofapps/worker/src/integration/handlers/received-message.ts's message-ingest entry points indistributedLock.runExclusive({ key: "conversation-ingest:<id>" }), keyed by the resolved conversationId — a concurrency safety net only, no behavior change for the uncontended case.B6 — recommend isolated production deployments
docker-entrypoint.shalready supportsworker <name>, but the default isworker all(one container, one crash kills every queue). Documents the recommended production split (at leastchat,integration,heavyas separate deployments) — docs only, no runtime change.B7 — prom-client metrics
Added
apps/worker/src/lib/metrics.ts(sharedRegistry,failed_jobs_totalCounter,job_duration_secondsHistogram) and wired it into every one of the 9apps/worker/src/*/worker.tsentrypoints'on("failed")/on("completed")handlers.B8 — jobId in failed/completed logs
Standardized every worker's failed/completed log to carry
jobId: job.idas a structured field (matching the naming already used inai-agent/heavy), instead of only interpolating it into the message text — lets log aggregators correlate a job's failure and completion by exact field match.B9 — health + metrics HTTP endpoints
Added
apps/worker/src/lib/health-server.ts:GET /health(200 while the process's BullMQWorkeris running, 503 once closing) andGET /metrics(prom-client registry).worker allruns each queue as its own OS process, so each worker gets its own default port (CHAT_WORKER_HEALTH_PORT=3301…DEFAULT_WORKER_HEALTH_PORT=3309) to avoid anEADDRINUSEcollision — aworker <name>deployment only ever binds its own port. Also added an unauthenticatedGET /healthtoapps/realtime/src/server.tsvia PartyKit'sonFetch(catches requests that don't match any/parties/:party/:idroute), so it never touches the existingonBeforeRequest403-by-default guard on real party traffic.Verification
pnpm --filter worker check-types && test— 2340/2340 passpnpm --filter realtime check-types— cleanapps/worker/src/chat/worker.tsagainst local Postgres/Redis,curl localhost:3301/health→200 ok,curl localhost:3301/metrics→ real prom-client output,curl localhost:3301/nope→404