Shared Startup Module (`galaxy_startup`)

Issue: #709 Status: Accepted

Problem

All 5 Python services duplicate identical startup patterns:

25-line structlog configuration block (identical across all 5)
8-line signal handler + shutdown_event pattern (identical across 4 gRPC services)
11-line run_health_server() function (identical across 4 gRPC services)
Task cancellation boilerplate (cancel() + try/await/except CancelledError)

Any startup change requires updating 5 files identically.

Solution

Create services/shared/galaxy_startup/__init__.py with 4 shared functions (not a class — functions are simpler and sufficient).

API

`configure_logging(log_level: str) -> None`

Configures logging.basicConfig and structlog.configure with the standard Galaxy processor chain:

filter_by_level, add_logger_name, add_log_level
PositionalArgumentsFormatter, TimeStamper(fmt="iso")
StackInfoRenderer, format_exc_info, UnicodeDecoder
JSONRenderer

Uses stdlib.BoundLogger, dict context class, stdlib.LoggerFactory, cache_logger_on_first_use=True.

`setup_shutdown_handler() -> tuple[asyncio.Event, Callable[[], None]]`

Registers SIGTERM and SIGINT signal handlers that set an asyncio.Event. Returns (shutdown_event, set_shutting_down) where set_shutting_down is a callable that sets the event.

The caller provides their own set_shutting_down from health.py — this function only provides the signal wiring and shutdown event. The returned set_shutting_down simply sets the event (useful for signal-only shutdown coordination).

Note: api-gateway does NOT use this — it has custom shutdown logic that also sets app.state.shutting_down.

`run_health_server(app, port: int) -> None`

Async function that wraps uvicorn Config + Server.serve():

host="0.0.0.0", log_level="warning"
The app argument is whatever Starlette/ASGI app the service creates via galaxy_health.create_health_app()

Meant to be used as asyncio.create_task(run_health_server(app, port)).

Note: api-gateway does NOT use this — it runs its own uvicorn with SSL and different log level.

`cancel_tasks(*tasks: asyncio.Task) -> None`

Async function that cancels each task and awaits it, catching CancelledError:

for task in tasks:
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

Usage

4 gRPC services (galaxy, physics, players, tick-engine)

from galaxy_startup import configure_logging, setup_shutdown_handler, run_health_server, cancel_tasks

configure_logging(settings.log_level)
logger = structlog.get_logger()

async def main():
    # ... service-specific init ...
    shutdown_event, _ = setup_shutdown_handler()
    health_app = create_health_app(...)  # service-specific
    health_task = asyncio.create_task(run_health_server(health_app, settings.http_port))
    await shutdown_event.wait()
    set_shutting_down()  # from .health import set_shutting_down
    await grpc_server.stop(grace=5)
    await cancel_tasks(health_task)

api-gateway

Only uses configure_logging() and cancel_tasks(). Keeps its own signal handler (sets app.state.shutting_down).

from galaxy_startup import configure_logging, cancel_tasks

configure_logging(settings.log_level)

async def main():
    # ... keeps existing signal handler ...
    await cancel_tasks(ws_task, version_task, automation_task)

File Location

services/shared/galaxy_startup/__init__.py — follows existing pattern of galaxy_health, galaxy_auth, galaxy_config.

Dependencies

structlog (already in all service requirements)
uvicorn (already in all gRPC service requirements)
Standard library: logging, sys, signal, asyncio

Shared Startup Module (galaxy_startup)