PhoenixGenApi — Architecture Deep Dive

Table of Contents

1. High-Level Overview
2. System Topology
3. Supervision Tree
4. Core Data Structures
5. Request Lifecycle
6. Config Management
7. Execution Engine
8. Rate Limiter
9. Permission System
10. Worker Pools & Circuit Breakers
11. Relay Messages
12. Hooks
13. Security
14. Telemetry
15. Multi-Version Support
16. Retry & Fault Tolerance
17. Node Selection Strategies

1. High-Level Overview

PhoenixGenApi is a framework for building API gateways on top of Phoenix Channels (WebSocket). Instead of defining HTTP endpoints, you define function configurations (FunConfig) that map WebSocket events to function calls — locally or remotely across an Elixir cluster.

┌──────────┐    WebSocket     ┌───────────────┐    RPC / Local     ┌──────────────┐
│  Browser  │ ◄─────────────► │  Gateway Node  │ ◄────────────────► │ Service Node │
│  Client   │  Phoenix Ch.    │  (Phoenix app) │    Erlang dist.    │  (your app)  │
└──────────┘                  └───────────────┘                    └──────────────┘

Key design principles:

• Transport-agnostic business logic — Your service functions don't know or care about WebSocket, JSON, or HTTP.
• Cluster-native — Services register their APIs on remote nodes; the gateway discovers them automatically via pull or push.
• Zero HTTP boilerplate — No controllers, no routers, no serializers. A FunConfig struct is the only thing between a WebSocket message and your function.
• Defense in depth — Rate limiting, permission checking, argument validation, MFA allowlists, and push tokens all layer together.

2. System Topology

There are two node roles:

Gateway Node (Phoenix app)

• Runs the Phoenix web server and Channels.
• Holds the ConfigDb (ETS) cache of all registered FunConfig entries.
• Runs the ConfigPuller (periodic pull from services) and ConfigReceiver (accepts pushes).
• Runs the RateLimiter, WorkerPoolSupervisor, and RelayServer.
• Executes requests by looking up the FunConfig, checking rate limits/permissions, then calling the MFA (locally or via RPC).

Service Node (any Elixir app)

• Runs your business logic (MyApp.Api).
• Optionally runs a Supporter module that returns FunConfig structs.
• Can push configs to the gateway on startup (ConfigPusher) or let the gateway pull them periodically.
• Sets client_mode: true to avoid starting gateway-only processes.

Communication Flow

Service Node                          Gateway Node
    │                                      │
    │  1. Push configs (on startup)        │
    │─────────────────────────────────────►│  ConfigReceiver
    │                                      │
    │  2. Periodic pull (every 30s)        │
    │◄─────────────────────────────────────│  ConfigPuller
    │  3. Return FunConfig list             │
    │─────────────────────────────────────►│
    │                                      │
    │  4. Client sends WebSocket request   │
    │         ──────────────►               │  Channel
    │                                      │  Executor
    │  5. RPC to service node              │
    │◄─────────────────────────────────────│
    │  6. Return result                     │
    │─────────────────────────────────────►│
    │                                      │  7. Push result to client
    │         ◄──────────────               │

3. Supervision Tree

The application supervisor (PhoenixGenApi.Supervisor) uses a :rest_for_one strategy. If a child crashes, it and all children started after it are restarted.

PhoenixGenApi.Supervisor (:rest_for_one)
├── PhoenixGenApi.RateLimiter              (GenServer — ETS-backed sliding window)
├── PhoenixGenApi.WorkerPoolSupervisor     (Supervisor — :one_for_one)
│   ├── PhoenixGenApi.WorkerPool           (GenServer — async_pool)
│   │   └── [N x Worker]                  (GenServer — task execution)
│   └── PhoenixGenApi.WorkerPool           (GenServer — stream_pool)
│       └── [N x Worker]                  (GenServer — task execution)
├── PhoenixGenApi.ConfigDb                 (GenServer — ETS-backed FunConfig store)
├── PhoenixGenApi.ConfigPuller             (GenServer — periodic pull scheduler)
├── PhoenixGenApi.ConfigReceiver           (GenServer — push handler + version tracker)
├── PhoenixGenApi.RelayRegistry            (Registry — :duplicate keys, pid dispatch)
└── PhoenixGenApi.RelayServer              (GenServer — serializes ETS relay ops)

When client_mode: true (on service nodes), the supervisor starts zero children — the node only needs the library's structs and the ConfigPusher module (which is stateless).

4. Core Data Structures

FunConfig

The central configuration unit. Each FunConfig maps one request_type to one function call.

%FunConfig{
  request_type: "get_user",        # Unique identifier for this API endpoint
  service: "user_service",         # Service grouping key
  nodes: [:"node1@host"],          # Where to run (:local for gateway-local)
  choose_node_mode: :random,       # :random | :hash | :round_robin | {:hash, key} | {:sticky, key}
  timeout: 5_000,                  # RPC timeout in ms
  mfa: {MyApp.Api, :get_user, []}, # {Module, Function, Args} to call
  arg_types: %{"user_id" => :string},  # Argument validation schema
  arg_orders: ["user_id"],         # Ordered arg list (or :map for map-style)
  response_type: :sync,            # :sync | :async | :stream
  version: "1.0.0",                # Semantic version for multi-version support
  check_permission: false,         # false | :any_authenticated | {:arg, key} | {:role, roles}
  permission_callback: nil,        # Custom {mod, fun, args} for permission checks
  retry: nil,                      # nil | integer() | {:same_node, n} | {:all_nodes, n}
  before_execute: nil,             # {mod, fun} | {mod, fun, extra_args}
  after_execute: nil,              # {mod, fun} | {mod, fun, extra_args}
  disabled: false                  # Soft-delete flag
}

Request

Created from the WebSocket payload by Request.decode!/1. Includes payload size validation (default 1MB), required field checks, and role sanitization.

%Request{
  request_id: "req_123",    # Client-generated unique ID
  request_type: "get_user", # Matches FunConfig.request_type
  service: "user_service",  # Matches FunConfig.service
  user_id: "user_42",       # From socket.assigns or payload
  device_id: "device_1",    # From payload
  args: %{"user_id" => "42"}, # Function arguments
  user_roles: ["admin"],    # For role-based permissions
  version: "1.0.0"          # Optional version override
}

Response

Returned to the client. Has constructors for each response type:

• Response.sync_response(request_id, result) — Standard success
• Response.async_response(request_id) — Acknowledged, processing async
• Response.stream_response(request_id, result, has_more) — Streaming chunk
• Response.stream_end_response(request_id) — Stream complete
• Response.error_response(request_id, error, can_retry) — Error

ServiceConfig

Tells the ConfigPuller how to reach a service node and pull its configs.

%ServiceConfig{
  service: "user_service",
  nodes: [:"node1@host"],
  module: MyApp.GenApi.Supporter,
  function: :get_config,
  args: [],
  version_module: MyApp.GenApi.Supporter,  # Optional: lightweight version check
  version_function: :get_config_version,
  version_args: []
}

PushConfig

Used when a service node pushes its config to the gateway (instead of being pulled).

%PushConfig{
  service: "user_service",
  nodes: [:"node1@host"],
  config_version: "1.2.3",
  fun_configs: [%FunConfig{...}],
  module: MyApp.GenApi.Supporter,  # Optional: for auto-pull after push
  function: :get_config,
  push_token: "secret"             # Authenticates push to gateway
}

5. Request Lifecycle

The complete path from WebSocket message to response:

Client                    Gateway Channel              Executor
  │                            │                          │
  │  push("api", payload)      │                          │
  │───────────────────────────►│                          │
  │                            │                          │
  │                     ┌──────┴──────┐                   │
  │                     │ handle_in/3 │                   │
  │                     │             │                   │
  │                     │ 1. Override │                   │
  │                     │    user_id  │                   │
  │                     │    from     │                   │
  │                     │    socket   │                   │
  │                     │    assigns  │                   │
  │                     │             │                   │
  │                     │ 2. Request  │                   │
  │                     │    .decode! │                   │
  │                     │    (valid.) │                   │
  │                     │             │                   │
  │                     │ 3. Execute  │                   │
  │                     └──────┬──────┘                   │
  │                            │                          │
  │                            │  Executor.execute!(req)  │
  │                            │─────────────────────────►│
  │                            │                          │
  │                            │                    ┌─────┴─────┐
  │                            │                    │ Phase 1:  │
  │                            │                    │ ConfigDb  │
  │                            │                    │ .get/3    │
  │                            │                    │ (ETS)     │
  │                            │                    └─────┬─────┘
  │                            │                          │
  │                            │                    ┌─────┴─────┐
  │                            │                    │ Phase 2:  │
  │                            │                    │ Hooks     │
  │                            │                    │ .run_     │
  │                            │                    │ before/3  │
  │                            │                    └─────┬─────┘
  │                            │                          │
  │                            │                    ┌─────┴─────┐
  │                            │                    │ Phase 3:  │
  │                            │                    │ RateLimit │
  │                            │                    │ + Perms   │
  │                            │                    └─────┬─────┘
  │                            │                          │
  │                            │                    ┌─────┴─────┐
  │                            │                    │ Phase 4-6:│
  │                            │                    │ sync /    │
  │                            │                    │ async /   │
  │                            │                    │ stream    │
  │                            │                    └─────┬─────┘
  │                            │                          │
  │                            │                    ┌─────┴─────┐
  │                            │                    │ Hooks     │
  │                            │                    │ .run_     │
  │                            │                    │ after/4   │
  │                            │                    └─────┬─────┘
  │                            │                          │
  │                            │  {:push, response}       │
  │                            │◄─────────────────────────│
  │                            │                          │
  │  push(socket, "api", resp) │                          │
  │◄───────────────────────────│                          │

Phase Details

Line-by-Line Execution

The complete path from WebSocket message to response, traced through the source. Use this as a debugging map: find the phase, open the file, follow the branches.

Phase 1: Executor.execute!(request)

File: lib/phoenix_gen_api/executor/executor.ex

execute!(request)
  │
  ├─ 1. Record start_time (monotonic microsecond)
  │
  ├─ 2. Emit telemetry [:executor, :request, :start]
  │
  ├─ 3. try do
  │     │
  │     ├─ 4. Resolve version: request.version || "0.0.0"
  │     │
  │     ├─ 5. ConfigDb.get(service, request_type, version)
  │     │     │
  │     │     ├─ {:ok, fun_config}
  │     │     │     └─ execute_with_config!(request, fun_config)  ──→ Phase 2
  │     │     │
  │     │     ├─ {:error, :not_found}
  │     │     │     └─ Log warning
  │     │     │     └─ Return error_response("unsupported function: ... version ...")
  │     │     │
  │     │     └─ {:error, :disabled}
  │     │           └─ Log warning
  │     │           └─ Return error_response("disabled function: ... version ...")
  │     │
  │     ├─ 6. Calculate duration
  │     │
  │     ├─ 7. Extract {success, async} from result
  │     │     └─ %Response{success: s, async: a} → {s, a}
  │     │     └─ _ → {true, false}  (fallback for non-Response tuples)
  │     │
  │     ├─ 8. Emit telemetry [:executor, :request, :stop]
  │     │
  │     └─ 9. Return result
  │
  └─ rescue e
        ├─ Emit telemetry [:executor, :request, :exception]
        └─ reraise e

ConfigDb.get/3 — Direct ETS read (no GenServer call)

File: lib/phoenix_gen_api/config_cache/config_cache.ex

get(service, request_type, version)
  │
  ├─ :ets.lookup_element(__MODULE__, {service, request_type, version}, 2)
  │     │
  │     ├─ config when is_map(config)
  │     │     ├─ config.disabled == true  → {:error, :disabled}
  │     │     └─ otherwise                → {:ok, config}
  │     │
  │     └─ _  → {:error, :not_found}
  │
  └─ rescue ArgumentError  → {:error, :not_found}

Phase 2: Executor.execute_with_config!(request, fun_config)

File: lib/phoenix_gen_api/executor/executor.ex

execute_with_config!(request, fun_config)
  │
  ├─ 1. Log debug: request_id, response_type
  │
  └─ 2. Hooks.run_before(before_execute, request, fun_config)
        │
        ├─ {:ok, new_request, new_fun_config}
        │     └─ do_execute_with_config!(new_request, new_fun_config)  ──→ Phase 3
        │
        └─ {:error, reason}
              ├─ Log warning: request_id, reason
              ├─ Build error_response("hook rejected: ...")
              ├─ Hooks.run_after(after_execute, request, fun_config, response)
              └─ Return response   ←── (Fix #8: was previously discarded)

Hooks.run_before/3

File: lib/phoenix_gen_api/hooks/hook.ex

run_before(nil, request, fun_config)
  └─ {:ok, request, fun_config}   (no-op)

run_before({mod, fun}, request, fun_config)
  └─ execute_hook(:before, mod, fun, [request, fun_config])
       │
       ├─ Telemetry :start
       ├─ Task.async + Task.yield(hook_timeout) with try/rescue inside task
       ├─ Timeout: Task.shutdown(:brutal_kill) → {:error, reason}
       ├─ Telemetry :stop or :exception (kind: :error or :timeout)
       │
       ├─ Returns {:ok, {:ok, new_request, new_fun_config}} → proceed
       ├─ Returns {:ok, {:error, reason}} → abort
       ├─ Returns {:ok, _} → proceed with original request/fun_config
       └─ Returns {:error, reason} → abort

run_before({mod, fun, extra_args}, request, fun_config)
  └─ Same but args = [request, fun_config | extra_args]

Phase 3: Executor.do_execute_with_config!(request, fun_config)

File: lib/phoenix_gen_api/executor/executor.ex

do_execute_with_config!(request, fun_config)
  │
  └─ RateLimiter.check_rate_limit(request)  ──→ Phase 3a
        │
        ├─ :ok
        │     └─ try do
        │           │
        │           ├─ Permission.check_permission!(request, fun_config)  ──→ Phase 3b
        │           │     │
        │           │     ├─ true (permission granted) → continues below
        │           │     │
        │           │     └─ raises PermissionDenied
        │           │           ├─ Build error_response("Permission denied")
        │           │           ├─ Hooks.run_after(after_execute, ...)
        │           │           └─ Return error_response   ←── (Fix #7: no longer reraises)
        │           │
        │           ├─ Dispatch by response_type:
        │           │     │
        │           │     ├─ :sync   → do_call(request, fun_config)         ──→ Phase 4
        │           │     ├─ :async  → async_call(request, fun_config)      ──→ Phase 5
        │           │     ├─ :none   → async_call(request, fun_config)      ──→ Phase 5
        │           │     ├─ :stream → stream_call(request, fun_config)     ──→ Phase 6
        │           │     │
        │           │     └─ other
        │           │           └─ Log error
        │           │           └─ error_response("unsupported response type: ...")
        │           │
        │           ├─ Hooks.run_after(after_execute, request, fun_config, result)
        │           └─ Return result
        │
        └─ error (rate limiter returned error)
              ├─ handle_rate_limit_error(error, request, fun_config)  ──→ Phase 3c
              ├─ Hooks.run_after(after_execute, request, fun_config, result)
              └─ Return result

Phase 3a: RateLimiter.check_rate_limit(request)

File: lib/phoenix_gen_api/rate_limiter/rate_limiter.ex

check_rate_limit(request)
  │
  ├─ enabled?() == false → :ok (rate limiting disabled)
  │
  ├─ Select instance (consistent hashing by request_id)
  │
  ├─ GenServer.call(instance, {:check_rate_limit, request})
  │     │
  │     └─ check_request_limits(request, state)
  │           │
  │           ├─ check_global_limits(request, global_limits)
  │           │     └─ For each global limit:
  │           │           ├─ Extract key_value from request (user_id, device_id, etc.)
  │           │           ├─ check_and_record(:rate_limiter_global, key, limit)
  │           │           │     ├─ Within limit → :ok
  │           │           │     └─ Exceeded → {:error, :rate_limited, details}
  │           │           └─ Emit exceeded telemetry
  │           │
  │           └─ check_api_limits(request, api_limits)
  │                 └─ For each matching API limit:
  │                       ├─ build_api_key(key_value, {service, request_type})
  │                       ├─ check_and_record(:rate_limiter_api, key, limit)
  │                       └─ Same :ok / {:error, :rate_limited, details}
  │
  ├─ Emit check telemetry
  │
  └─ rescue e (rate limiter itself crashed)
        ├─ fail_open?() → Log error, return :ok (allow request)
        └─ otherwise    → {:error, :rate_limiter_error, %{message: ...}}

Phase 3b: Permission.check_permission!(request, fun_config)

File: lib/phoenix_gen_api/permission.ex

check_permission!(request, fun_config)
  │
  └─ if not check_permission(request, fun_config) do
        ├─ Log warning: user_id, request_id, request_type, mode
        └─ raise PermissionDenied
     end
     └─ nil (permission granted, returns nil implicitly)

check_permission/2 dispatch order (most specific first):

1. permission_callback: {mod, fun, args}  → execute_permission_callback(mod, fun, [request | args])
2. check_permission: false                → true (public endpoint)
3. check_permission: :any_authenticated   → true if user_id is non-empty binary
4. check_permission: {:arg, arg_name}     → compare request.args[arg_name] with user_id
5. check_permission: {:role, roles}       → check user_roles ∩ allowed_roles ≠ ∅
6. Invalid check_permission mode          → log error, false
7. Invalid permission_callback format     → log error, fallback to check_permission mode

Phase 3c: handle_rate_limit_error/3

handle_rate_limit_error({:error, :rate_limited, details}, request, _)
  └─ error_response("Rate limit exceeded. Retry after N seconds.") + can_retry: true

handle_rate_limit_error({:error, :rate_limiter_error, details}, request, _)
  └─ Log error (fail-closed)
  └─ error_response("Rate limit service unavailable", can_retry: true)

handle_rate_limit_error({:error, :permission_denied}, request, _)
  └─ Log warning
  └─ error_response("Permission denied")

handle_rate_limit_error(error, request, _)  (catch-all)
  └─ Log error (fail-closed)
  └─ error_response("Rate limit service unavailable", can_retry: true)

Phase 4: do_call(request, fun_config) — Sync Execution

File: lib/phoenix_gen_api/executor/executor.ex

do_call(request, fun_config)
  │
  ├─ 1. ArgumentHandler.convert_args!(fun_config, request)
  │     │
  │     ├─ validate_args!(config, request)
  │     │     ├─ check_extra_args!(args, arg_types, ...)  → raises ArgumentError if extra
  │     │     └─ validate_all_args!(arg_types, args, ...)
  │     │           └─ For each arg: validate_arg!(name, type, params, value, allow_nil, request)
  │     │                 ├─ nil + not allow_nil → raises ArgumentError
  │     │                 └─ arg_validation!(type, value, name, request)
  │     │                       ├─ validate_simple_type!  (boolean, string, num, uuid, etc.)
  │     │                       └─ validate_complex_type! (string with max_bytes, list with max_items, etc.)
  │     │
  │     └─ Convert args to final format
  │           ├─ No arg_types → []
  │           ├─ arg_orders: :map → [converted_map]
  │           ├─ Single arg → Map.values(converted_args)
  │           └─ Multiple args → ordered list by arg_orders
  │
  ├─ 2. Build final_args = predefined_args ++ converted_args ++ info_args
  │     │
  │     └─ info_args(request, fun_config)
  │           ├─ request_info: false → []
  │           └─ request_info: true  → [%{request_id, user_id, device_id, stream_pid?}]
  │
  ├─ 3. Normalize retry config
  │     └─ FunConfig.normalize_retry(retry)
  │           ├─ nil → nil
  │           ├─ n (integer) → {:all_nodes, n}
  │           ├─ {:same_node, n} → {:same_node, n}
  │           └─ {:all_nodes, n} → {:all_nodes, n}
  │
  ├─ 4. Execute
  │     │
  │     ├─ local_service?(fun_config)  → execute_local_with_retry(mod, fun, args, timeout, retry)
  │     │     │
  │     │     ├─ execute_local(mod, fun, args, timeout)
  │     │     │     ├─ function_exported?(mod, fun, arity) check
  │     │     │     ├─ Task.async → apply(mod, fun, args)
  │     │     │     ├─ Task.yield(task, timeout) || Task.shutdown(task)
  │     │     │     │     ├─ {:ok, result} → result
  │     │     │     │     ├─ nil → {:error, "local execution timed out"}
  │     │     │     │     └─ {:exit, reason} → {:error, "local execution failed"}
  │     │     │     └─ not exported → {:error, :function_not_found}
  │     │     │
  │     │     └─ apply_local_retry(result, mod, fun, args, timeout, retry_config, request_id)
  │     │           ├─ retryable_error? && has_retry_remaining?
  │     │           │     ├─ Calculate backoff
  │     │           │     ├─ Process.sleep(backoff_ms)
  │     │           │     ├─ Emit retry telemetry
  │     │           │     └─ Recurse with {mode, n-1}
  │     │           ├─ retryable_error? && exhausted → emit :retry, :exhausted telemetry
  │     │           └─ otherwise → return result
  │     │
  │     └─ remote service  → execute_remote_with_retry(mod, fun, args, fun_config, request, retry)
  │           │
  │           ├─ NodeSelector.get_nodes(fun_config, request)
  │           │     └─ resolve_nodes(config) → select_node(nodes, mode, request)
  │           │           ├─ :random → Enum.random
  │           │           ├─ :hash → hash-based selection
  │           │           ├─ {:hash, key} → hash on specific arg
  │           │           ├─ :round_robin → atomic counter
  │           │           └─ {:sticky, key} → ETS sticky table lookup
  │           │
  │           ├─ execute_remote_with_fallback([node | rest], mod, fun, args, timeout, request_id, _)
  │           │     │
  │           │     ├─ :rpc.call(node, mod, fun, args, timeout)
  │           │     │     ├─ {:badrpc, :timeout} → log warning, try next node
  │           │     │     ├─ {:badrpc, {:EXIT, reason}} → log warning, try next node
  │           │     │     ├─ {:badrpc, reason} → log warning, try next node
  │           │     │     └─ result → return result (success, stops fallback)
  │           │     │
  │           │     └─ execute_remote_with_fallback([], ...) → return last_error || {:error, "no target nodes"}
  │           │
  │           └─ apply_remote_retry(state)
  │                 ├─ {:same_node, n} → retry on same nodes
  │                 ├─ {:all_nodes, n} → retry on ALL nodes (re-resolved)
  │                 └─ _ → return result (no retry or exhausted)
  │
  └─ 5. handle_call_result(result, request_id)
        │
        ├─ {:error, reason} → error_response(request_id, get_error_message(result))
        ├─ {:ok, result}   → sync_response(request_id, result)
        ├─ non-tuple result → sync_response(request_id, result) + warning log
        └─ other tuple      → error_response(request_id, "Unexpected execution result") + error log

sync_call/2 — Wrapper around do_call

sync_call(request, fun_config)
  │
  └─ try do
        do_call(request, fun_config)
      rescue e  → error_response(request_id, get_error_message(e))
      catch :exit, reason  → error_response(request_id, get_error_message(reason))
      catch :throw, reason → error_response(request_id, get_error_message(reason))
      catch kind, reason   → error_response(request_id, get_error_message(reason))
     end

Phase 5: async_call(request, fun_config)

File: lib/phoenix_gen_api/executor/executor.ex

async_call(request, fun_config)
  │
  ├─ 1. Capture receiver = self()
  │
  ├─ 2. Build task fn:
  │     │
  │     └─ try do
  │           result = sync_call(request, fun_config)   ──→ Phase 4 (sync_call)
  │           if response_type != :none:
  │             send(receiver, {:async_call, result})
  │         catch kind, reason:
  │           Log error
  │           if Process.alive?(receiver):   ←── (Fix #9: guard added)
  │             send(receiver, {:async_call, error_response})
  │        end
  │
  ├─ 3. WorkerPool.execute_async(:async_pool, task)
  │     │
  │     ├─ :ok
  │     │     ├─ response_type != :none → async_response(request_id)
  │     │     └─ response_type == :none → {:ok, :no_response}
  │     │
  │     └─ {:error, :queue_full}
  │           └─ error_response("Service temporarily unavailable", can_retry: true)
  │
  └─ 4. Caller receives {:async_call, result} via handle_info later

Phase 6: stream_call(request, fun_config)

File: lib/phoenix_gen_api/executor/executor.ex

stream_call(request, fun_config)
  │
  ├─ 1. Capture receiver = self(), request_id
  │
  ├─ 2. Build task fn:
  │     │
  │     └─ try do
  │           StreamCall.start_link(%{request, fun_config, receiver})
  │           │
  │           ├─ {:ok, pid}
  │           │     ├─ if Process.alive?(receiver):   ←── (Fix #9: guard added)
  │           │     │     send(receiver, {:stream_started, request_id, pid})
  │           │     ├─ Process.monitor(pid)
  │           │     └─ receive {:DOWN, ^ref, :process, ^pid, _} -> :ok
  │           │         after timeout → GenServer.stop(pid, :timeout)
  │           │
  │           └─ {:error, reason}
  │                 └─ if Process.alive?(receiver):   ←── (Fix #9: guard added)
  │                       send(receiver, {:stream_response, error_response(...)})
  │         catch kind, reason:
  │           Log error
  │           if Process.alive?(receiver):   ←── (Fix #9: guard added)
  │             send(receiver, {:stream_response, error_response(...)})
  │        end
  │
  ├─ 3. WorkerPool.execute_async(:async_pool, task)
  │     │
  │     ├─ :ok → stream_response(request_id, :init)   ←── (Fix #1: no blocking receive)
  │     │
  │     └─ {:error, :queue_full}
  │           └─ error_response("Service temporarily unavailable", can_retry: true)
  │
  └─ 4. Caller receives messages later via handle_info:
        ├─ {:stream_started, request_id, pid} → Process.put(stream_call_pid, pid)
        └─ {:stream_response, response} → push to client

StreamCall GenServer lifecycle:

StreamCall.start_link(args)
  │
  ├─ init → {:ok, args, {:continue, :start_stream}}
  │
  ├─ handle_continue(:start_stream, state)
  │     ├─ Executor.sync_call(request, fun_config)   ──→ Phase 4
  │     ├─ error? → send(receiver, {:stream_response, error}) → stop
  │     └─ success → send(receiver, {:stream_response, stream_response(result)}) → continue
  │
  ├─ handle_info({:result, data}, state)
  │     └─ send(receiver, {:stream_response, stream_response(result, true)})
  │
  ├─ handle_info({:last_result, data}, state)
  │     └─ send(receiver, {:stream_response, stream_response(result, false)}) → stop
  │
  ├─ handle_info({:error, error}, state)
  │     └─ send(receiver, {:stream_response, error_response(...)}) → stop
  │
  ├─ handle_info(:complete, state)
  │     └─ send(receiver, {:stream_response, stream_end_response(request_id)}) → stop
  │
  └─ handle_cast(:stop, state)  → send completion → stop

Return to Channel: handle_info clauses

File: lib/phoenix_gen_api.ex

handle_info({:push, result}, socket)
  └─ push(socket, event, result)

handle_info({:stream_started, request_id, pid}, socket)   ←── (Fix #1: new clause)
  └─ Process.put({:phoenix_gen_api, :stream_call_pid, request_id}, pid)

handle_info({:stream_response, result}, socket)
  └─ push(socket, event, result)

handle_info({:async_call, result}, socket)
  └─ push(socket, event, result)

handle_info({:relay_message, result}, socket)
  └─ push(socket, event, result)

Error Handling Summary

6. Config Management

Config management follows a pull or push model (or both).

Pull Model (Gateway-initiated)

The ConfigPuller GenServer runs on the gateway and periodically pulls configs from service nodes.

ConfigPuller (timer-based)
    │
    ├─ For each ServiceConfig:
    │   ├─ Optional: Call version_module.version_function on remote node
    │   │   └─ If version matches stored version → skip (saves bandwidth)
    │   │
    │   ├─ Call module.function(args) on remote node (RPC)
    │   │   └─ Returns list of FunConfig structs
    │   │
    │   ├─ Validate each FunConfig
    │   ├─ Enforce service name
    │   ├─ Ensure version field
    │   └─ ConfigDb.batch_add(configs) → ETS write
    │
    └─ Schedule next pull (with exponential backoff on failures)

Version-based skip: If ServiceConfig.version_module and version_function are set, the puller first makes a lightweight RPC to check if the version changed. If it matches the stored version, the full config pull is skipped entirely.

Exponential backoff: On pull failures, the interval increases (up to a cap) to avoid hammering unreachable nodes.

Push Model (Service-initiated)

On service node startup, ConfigPusher.push_on_startup/3 sends a PushConfig to the gateway's ConfigReceiver.

Service Node                          Gateway Node
    │                                      │
    │  ConfigPusher.push_on_startup        │
    │  (RPC to ConfigReceiver)             │
    │─────────────────────────────────────►│
    │                                      │
    │                               ┌──────┴──────┐
    │                               │ 1. Decode   │
    │                               │ 2. Validate │
    │                               │ 3. Check    │
    │                               │    version  │
    │                               │ 4. Check    │
    │                               │    push_tok │
    │                               │ 5. batch_add│
    │                               │ 6. Register │
    │                               │    puller   │
    │                               └──────┬──────┘
    │                                      │
    │  {:ok, :accepted}                    │
    │◄─────────────────────────────────────│

Auto-pull registration: A PushConfig can include module/function fields. After accepting the push, the ConfigReceiver registers the service with the ConfigPuller for periodic refresh — giving you push-on-startup + pull-for-refresh.

ConfigDb (ETS Store)

ConfigDb is a GenServer wrapping an ETS table. The key is {service, request_type, version} and the value is the FunConfig struct.

Key operations: