[jopemachine]

📚 Stacked PRs

Part of the AppProxy worker Ingress stack (epic BA-8433). Merge in order (bottom-up):

1. 👉 feat(BA-8443): push ingresses from the coordinator to api-delivery workers #15833 — feat(BA-8443): push ingresses from the coordinator to api-delivery workers ← you are here
2. feat(BA-8444): add the coordinator client that writes ingresses to Traefik etcd #15893 — feat(BA-8444): export ingresses from the coordinator to Traefik etcd

Resolves #15539 (BA-8443)

Summary

• Coordinator API for the manager. Like the worker handler, it is an IngressHandler class with api_handler methods. Each route is wrapped with auth_required("manager"), and the handler takes IngressServiceCtx, a MiddlewareParam carrying only the IngressService:
  • POST /v2/ingress/upsert takes IngressUpsertRequest and answers 204. A success means every node holds every sent ingress, so there is nothing to return.
  • POST /v2/ingress/remove takes IngressRemoveRequest and answers 204.
• IngressService groups ingresses by authority and reads the appproxy-worker group from service discovery. It pushes concurrently, with an AppProxyWorkerClient (feat(BA-8642): add the coordinator client that pushes ingresses to api-delivery workers #15915) on a session from the common ClientPool, to the endpoint of every node labelled ingress-delivery-mode=api for that authority (labels from feat(BA-8635): choose how the AppProxy worker receives ingresses #15890). An authority with no such node is skipped, since its workers read ingresses from Redis. It keeps no ingress state and needs no schema change.
• Service discovery is read on redis, where the workers register; core_redis only moves the coordinator's own registration.
• Failures:
  • The coordinator answers only after every push has settled, so a late push cannot land after the manager's next request.
  • If any push failed, it answers ApiDeliveryWorkerPushFailedError (502). data.failures holds the error body of each failed node:
    • 4xx from the worker → ApiDeliveryWorkerRejectedError; 5xx → ApiDeliveryWorkerServerError; unreachable → ApiDeliveryWorkerUnreachableError. All from feat(BA-8642): add the coordinator client that pushes ingresses to api-delivery workers #15915.
    • Each body carries data.worker_addr; rejected and server errors also carry the worker's status and body.
  • Every push failure is a 502 because one request fans out to several nodes that can fail differently, e.g. one rejects a stale revision while another is down. A single 4xx would misreport the nodes that failed for another reason. The manager tells a rejection apart by failures[].type.
  • A malformed request body is answered 400 and a wrong token 401 by the coordinator itself, before any push.
  • Upsert with the same revision and remove are both safe to retry.

Not in this PR

• The manager-side client that calls these endpoints.

Test plan

• pants check / pants test on tests/unit/appproxy:: and src/ai/backend/appproxy::
• tests/unit/appproxy/coordinator/services/test_ingress.py — pushes go to each authority's api-delivery nodes only; an authority with no node is skipped; a failed node fails the push after every node finishes; every failed node is reported in failures

🤖 Generated with Claude Code

[Copilot]

🟡 Changes recommended

Startup initialization and delivery failure handling contain unresolved defects that can prevent service startup or leave worker state incorrect.

3 open issues

• Assign worker discovery instance to root_ctx before yielding · New
• Wait for all worker requests before propagating failures · New
• Distinguish empty groups from failed discovery lookups · New

What changed in this PR

Adds the AppProxy coordinator’s manager-facing ingress API and stateless delivery to API-mode workers, providing the push path for the ingress stack.

Changes:

• Adds authenticated upsert and remove endpoints with typed request context and responses.
• Groups pushes by authority, discovers API-delivery workers, and uses the shared HTTP client pool.
• Adds service tests for worker selection, skipped authorities, and delivery failures.

File	Description
tests/​unit/​appproxy/​coordinator/​services/​test_ingress.py	Tests worker selection and push failures.
tests/​unit/​appproxy/​coordinator/​services/​BUILD	Registers the new service tests.
tests/​unit/​appproxy/​coordinator/​clients/​worker/​test_client.py	Updates address fixtures and removes obsolete pool tests.
src/​ai/​backend/​common/​dto/​appproxy/​v2/​ingress/​response.py	Adds the coordinator upsert response.
src/​ai/​backend/​appproxy/​coordinator/​types.py	Declares ingress service and delivery dependencies.
src/​ai/​backend/​appproxy/​coordinator/​services/​ingress.py	Discovers workers and dispatches ingress operations.
src/​ai/​backend/​appproxy/​coordinator/​server.py	Wires routes, discovery, client pooling, and service lifecycle.
src/​ai/​backend/​appproxy/​coordinator/​dto/​context.py	Provides request-scoped ingress service access.
src/​ai/​backend/​appproxy/​coordinator/​dto/​__init__.py	Initializes the DTO package.
src/​ai/​backend/​appproxy/​coordinator/​clients/​worker/​client.py	Uses typed addresses and externally managed sessions.
src/​ai/​backend/​appproxy/​coordinator/​api/​ingress.py	Defines manager-authenticated ingress endpoints.
changes/​15833.feature.md	Records the new coordinator ingress API.

_{Review effort: Balanced}