Skip to content

Backend Conventions

This document covers backend dependency injection patterns, the backend-data-bridge boundary contracts, broker API wiring, token broker kRPC resilience, recovery manager coalescing, and per-module Kubernetes manifest details for backend-database, backend-recorder, backend-sync, and backend-token-broker.

For the high-level system map, see architecture-overview.md. For the live feed and processor machinery, see live-market-pipeline.md. For Gradle DSL and JKube plugin details, see build-system.md.

Backend Bootstrap & Dependency Injection

Both backend-app and backend-server use an asynchronous, fail-fast bootstrap sequence powered by Ktor 3's Concurrent Modules and Dagger 2.

Fail-Fast Repositories: Key repositories (OrdersRepository, StrategyRepository, PositionsRepository, TokenManager) use private constructors and suspend operator invoke factories. Database pre-population and DataStore reads happen inside these suspendable factories, ensuring instances are never available to the application in a partially initialized state. TokenManager's factory loads the persisted token from DataStore before returning, eliminating the race where an async init block could overwrite a newer in-memory token.

Concurrent Modules: Ktor configuration in application.yaml sets startup: concurrent. The monolithic module setup is split into fine-grained, internal concurrent modules (e.g., core, orders, strategy, positions, application, health).

Provider-First Concurrent DI: For suspendable startup dependencies, the concurrent modules should register providers from no-arg module functions via dependencies { provide { ... } }. In this repo, orders(), strategy(), positions(), and application() follow that pattern so Ktor DI can await the provider itself during concurrent startup.

Application-Scoped DataStore: DataStore is created from the application CoroutineScope. Its background Gel pool refresh flow runs in that scope and closes the active pool on scope completion so a failed startup does not leave background work keeping the process alive.

Plugin-Local Route Registration: In backend-server, route registration stays in the module that installs the owning plugin (http, sse, graphql). Do not reintroduce a separate concurrent routes module; graphQLRoutes() depends on the GraphQL plugin already being installed.

Compile-Time Safety Boundary: Dagger 2 remains the source of truth for the final operational graph. The application module acts as the compile-time safety boundary, assembling the final ApplicationComponent from the instances initialized by earlier concurrent modules.

Terminal Error Propagation: Any failure during repository initialization or StartupTask execution bubbles up to the Ktor engine, causing an immediate process exit. This allows Kubernetes to detect initialization failures as container crashes and trigger automatic pod restarts, preventing "zombie pods".

backend-data-bridge Boundary

backend-data-bridge is the shared boundary module between backend-app and backend-server. It contains transport-agnostic service contracts and DTOs; the actual kRPC server/client wiring lives in the application modules.

API Directions

AppDataBridge is the narrow request/response API exposed by backend-app and consumed by backend-server. Its current responsibility is historical market data lookup via HistoricalMarketDataRequest.

  • backend-app registers it from Application.setAppDataBridge() and implements it in AppDataBridgeImpl, which reads raw historical data from HistoricalDataProvider and rolls it up to the requested interval before returning it.
  • backend-server consumes it through AppDataBridgeManager, which owns a private ConnectionManager that keeps a background connection to backend-app alive.

DataBridge is the reverse-direction long-lived streaming API hosted by backend-server and used by backend-app to push live application state upstream. It carries strategy signals, strategy output, market-feed events, trades, open positions, component-health snapshots, error snapshots, and server-to-app recovery events.

  • backend-app owns the outbound client in DataBridgeLauncher, which starts a background connection-maintenance job that reconnects with retry.
  • backend-server hosts it from Application.setDataBridge(). DataBridge(connection) creates a per-connection DataBridgeImpl, tracks connection lifecycle, and designates the first active connection as the leader for downlink-gated event streams.

DataStream

DataStream is the server-side aggregation surface over all active DataBridge connections. GraphQL/SSE/router code should depend on DataStream rather than individual bridge connections. It merges event flows, combines per-connection component-health lists, and exposes sendRecoveryEvent(...) for server-driven recovery messages back to connected app instances.

Readiness

backend-app keeps two health sets:

  • The plain ReadinessReporter set gates Cohort /readiness.
  • The @DiagnosticHealth set is published over DataBridge and adds per-datasource DataSourceReadinessReporters for observability without making them pod-readiness blockers.

DataSourceReadinessReporter overrides componentHealth() to map dataSource.health() directly to ComponentHealth (supporting DEGRADED status with a diagnostic message). Its isReady() delegates to health().status.allowsReadiness, so DEGRADED datasources still count as ready.

RequiredFeedReadinessReporter aggregates per-feed health using HealthStatus precedence: unknown feed identifiers → BLOCKED, any BLOCKED feed → BLOCKED, any DEGRADED feed → DEGRADED, all HEALTHY → HEALTHY. The key invariant is that DEGRADED feeds do not flip pod readiness red because HealthStatus.DEGRADED.allowsReadiness is true.

Readiness is represented on both sides: DataBridgeLauncher reports whether its forwarding jobs are active, while DataBridgeReadinessReporter on the server reports whether at least one app connection is present.

backend-database Kubernetes Manifest

backend-database/build.gradle.kts reads its db.* and k8s-namespace values via findProperty, so shared values can come from gradle.properties and environment-specific values from gradle.<buildType>.properties.

  • The module deploys the external image from db.image; JKube image building is marked skip(true) for this module.
  • Schema files are rendered into a single dbschema ConfigMap generated from the full src/main/resources/dbschema tree.
  • JKube configMap.items are generated from the same source tree so nested paths such as migrations/*.edgeql are recreated under /dbschema without hardcoding file names in YAML.
  • GEL credentials are injected through the custom envValuesFrom(...) DSL and the gel-env-vars secret.
  • Readiness and liveness probes are declared through httpGetProbe(...), which emits JKube getUrl values like http://:$port/server/status/ready.

backend-database Test Suites

JkubeManifestTest is parameterized across all build types and verifies the rendered JKube output against source files and Gradle properties:

  • dbschema-configmap.yml must contain one ConfigMap whose keys match the full src/main/resources/dbschema tree.
  • dbschema-migrations-configmap.yml must not exist anymore.
  • The geldata-dbschema volume's configMap.items must reconstruct the schema tree's relative paths exactly.
  • Image, namespace, ingress host/path, metrics port/path, env vars, service port, probes, volume mounts, and volumes are checked against gradle.properties, gradle.<buildType>.properties, and the rendered YAML.

DbBootstrapTest is a real-cluster integration suite:

  • Starts K3sContainer, creates the namespace and Gel credentials secret, applies the rendered JKube manifests, and waits for backend-database-0 to become ready.
  • Verifies runtime image, env vars, readiness/liveness probes, metrics annotations, volumes, and mounted schema files inside the real container.
  • Verifies bootstrap by running gel describe schema inside the database pod and comparing its canonical output exactly against backend-database/src/k8sIntegrationTest/resources/expected-schema.gel.txt.

backend-recorder Kubernetes Manifest

backend-recorder/build.gradle.kts reads all recorder.* and k8s-namespace values via extra[...]. Shared values (QuestDB image, credentials secret name, probe paths/delays, direct env var values) come from gradle.properties; environment-specific values (host, ports, metrics port/path) come from gradle.<buildType>.properties.

  • The module deploys the external image from recorder.image; JKube image building is marked skip(true).
  • QuestDB credentials are injected through envValuesFrom(...) using six envValueFromSecretKey entries bound to the qdb-env-vars secret.
  • Readiness and liveness probes are declared through httpGetProbe(...) on the health-check port (recorder.healthCheckPort).
  • Pod annotations, probes, env vars, the questdb-storage PVC volume, and the QuestDB postStart bootstrap hook are expressed in the Gradle DSL.
  • The recorder container's postStart exec hook shell body lives in backend-recorder/src/main/jkube/backend-recorder-post-start.sh; backend-recorder/build.gradle.kts loads it unchanged, injects RECORDER_HTTP_PORT into the pod env, and applies it through postStartExec(...).
  • JKube YAML fragments (backend-recorder-pv.yml, backend-recorder-pvc.yml, backend-recorder-ingress.yml) handle the resources that are cleaner to express as overlay YAML than through the Gradle DSL.

backend-recorder QuestDB Access

  • QuestDbConnectionConfig centralizes the recorder's host, port, credential, and database-name settings.
  • Only the outermost Recorder.invoke(coroutineScope) call resolves QuestDbConnectionConfig.fromEnvironment().
  • Inner layers (RecorderImpl, MarketDataRepository, Ingester, and Retriever) require an explicit QuestDbConnectionConfig, which keeps tests and other call sites deterministic.
  • Future improvement: replace the current explicit config threading with dependency injection so production and tests can bind different QuestDB connection settings without widening recorder internals purely for testability.

backend-recorder Test Suites

JkubeManifestTest is parameterized across all build types and verifies the rendered JKube output against Gradle properties:

  • Image, namespace, ingress host/path, metrics annotations, all six secret env var refs, six direct env vars, service ports (HTTP + PgWire), probes, volume mount, and volume are checked against gradle.properties, gradle.<buildType>.properties, and the rendered YAML.
  • The recorder container's postStart hook command against http://127.0.0.1:$RECORDER_HTTP_PORT/exec is also checked from the rendered YAML.

RecorderDbIntegrationTest is a QuestDB-backed integration suite:

  • Starts the official QuestDBContainer, initializes it from backend-recorder/src/dbIntegrationTest/resources/questdb/market-data-init.sql.
  • Constructs Recorder(coroutineScope, connectionConfig) explicitly from the container endpoints and credentials.
  • Verifies connectivity, time-range retrieval, target-volume retrieval, average-volume aggregation, and deduplication for (instrument_id, ts) rows.

RecorderBootstrapTest is a real-cluster integration suite:

  • Starts K3sContainer, creates the namespace and qdb-env-vars credentials secret, applies the rendered JKube manifests, and waits for backend-recorder-0 to become ready.
  • Verifies runtime image, env vars, readiness/liveness probes, metrics annotations, volume mount, and the postStart hook on the QuestDB pod.
  • There is no exec-level schema assertion yet; bootstrap success is inferred from the pod becoming ready with the expected postStart hook in place.

k8sIntegrationTest runs with maxParallelForks = 1 and depends on k8sResource before execution because it reads manifests from the standard JKube output under build/classes/java/main/META-INF/jkube/kubernetes, not from prepareManifestFixtures.

backend-sync Kubernetes Manifest

backend-sync/build.gradle.kts reads its sync.* and k8s-namespace values via extra[...], reads db.credentials.secret.name via findProperty, and renders src/main/db-init into a db-init ConfigMap.

  • The module builds and deploys its own application image via JKube image config; manifest tests render fixtures with CONTAINER_REGISTRY=test.invalid for deterministic image assertions.
  • Pod annotations plus the main application container's direct env vars are expressed in the Gradle DSL. Secret/field-based env refs (DB_USER, DB_PASSWORD, KUBERNETES_NAMESPACE, HOSTNAME) are injected through controllerResourceConfig { envValuesFrom(...) }.
  • Static backend-sync-deployment.yml still carries the main container probes and the three init containers (backend-sync-db, wait-for-sync-db, wait-for-database).
  • The local Postgres container uses the PostgreSQL 18+ volume layout: the PVC is mounted at /var/lib/postgresql, and the bootstrap SQL ConfigMap is mounted at /docker-entrypoint-initdb.d.
  • Existing volumes initialized with the older /var/lib/postgresql/data layout need recreation or migration before the postgres:18.3 sidecar will start cleanly.

backend-sync Sync Recorder

backend-sync uses Exposed 1.2.0 with the R2DBC PostgreSQL driver to persist sync events. The recorder lives in backend-sync/src/main/kotlin/com/timemanx/quant/server/sync/recorder/SyncEventRecorder.kt and stores rows in the local Postgres instance bootstrapped from backend-sync/src/main/db-init/init.sql.

  • Runtime database access is configured from SYNC_DB_HOST, SYNC_DB_PORT, SYNC_DB_USER, SYNC_DB_PASSWORD, and SYNC_DB_NAME.
  • Application.module() owns the recorder singleton and constructs SyncEventRecorder(host, port, user, password, dbName) explicitly from those runtime values.
  • Future improvement: replace the current explicit production/test database wiring with dependency injection so backend-sync can bind different database connection settings in tests without threading test-specific config through recorder internals.
  • SyncRecords.type is a PostgreSQL enum (SyncType), while SyncRecords.status remains JSONB and stores the serialized SyncStatus payload.
  • The schema-level sync_records_status_valid check requires scheduled_time, validates that scheduled_time, start_time, and completion_time are parseable TIMESTAMPTZ strings when present, and enforces scheduled_time <= start_time <= completion_time plus the completion_time -> start_time dependency.
  • The sync_record_enforce_immutability trigger prevents updates to id, name, and type, and also prevents already-set scheduled_time, start_time, or completion_time values from being changed or removed.
  • SyncEventRecorder.latestEvent(name, type) returns the most recently scheduled sync row for a data source and is used during startup recovery.
  • SyncEventRecorder.record(record) behaves like a compare-and-set update: it narrows the WHERE clause to valid state transitions and throws SyncRecordTransitionRejected when no row matches that guarded update.
  • InstrumentSyncTask resumes from the last persisted state per data source: scheduled rows are reused, started rows are resumed to completion/failure on the same record, and completed/no-history cases schedule the next daily run from refreshTime.
  • InstrumentSyncTask wraps SyncRecordTransitionRejected with task-specific context before rethrowing, so transition failures surface with the data source name and record ID.
  • LiveInstrumentFeedRefresher resolves a sync target by matching DataSourceIdentifier.name against an existing Gel Feed row from dataStore.feeds().
  • addInstruments writes directly to Instrument without a staging table: a pre-flight query fetches existing instrument names, the input sequence is streamed in batches of 2,000 with flatMapMerge(concurrency=6) for parallel inserts, removals are deleted in batched chunks, and change records are written separately (auto-commit per operation). The InstrumentSnapshot type has been removed from the schema.
  • feedInstruments(feedId) uses a concurrent read pattern: a count query determines total pages, then all pages are fetched in parallel with flatMapMerge(concurrency=6) and reassembled in offset order. The syncTime query runs concurrently with the page fetches.
  • gelClient is the GelClientPool instance. All operations call the pool directly without .use {} — the pools are closed once in DataStore.close().
  • The Gel schema has index on (.feed) on Instrument so that all feed-filtered queries use index seeks instead of sequential scans.
  • backend-sync does not create those Feed rows itself; feed bootstrap is currently owned by backend-app startup via AddNewFeedsTask / FeedRepository.addNewFeeds().
  • If a Feed row is missing, backend-sync records a sync Error with Feed not found for data source: ... and retries on the scheduler path.
  • Upstox instrument decode is currently tolerant at the row level: malformed rows are logged and skipped inside UpstoxApiDecorator, and the outer sync can still complete with SyncStatus.Success.
  • Those skipped-row decode warnings are not yet retained in SyncRecords.status; operators must consult application logs to see which instruments were dropped.
  • The bootstrap SQL is mounted into /docker-entrypoint-initdb.d of the local Postgres sidecar, so these constraints only apply on first Postgres initialization; existing backend-sync volumes need an explicit migration if the schema already exists.

backend-sync Test Suites

test combines application logic coverage with JKube fixture assertions:

  • InstrumentSyncTaskTest verifies startup recovery and transition error handling with fake scheduler/recorder seams and a fixed Clock.
  • InstrumentSyncTaskTest also verifies that refresh/storage failures go through onTaskFailed and do not record false success.
  • SyncRowsTest verifies the SyncRow mapping used by the sync UI/API layer.
  • JkubeManifestTest is parameterized across all build types and reads build/test-manifests/<buildType> fixtures rendered with CONTAINER_REGISTRY=test.invalid.

SyncEventRecorderDbIntegrationTest is a Postgres-backed integration suite:

  • Starts PostgreSQLContainer, mounts backend-sync/src/main/db-init/init.sql into /docker-entrypoint-initdb.d.
  • Verifies latest-event lookup, guarded state transitions, time-window event queries, and schema-level transition protections against a real database.

SyncBootstrapTest is a real-cluster integration suite:

  • Starts K3sContainer, creates the Gel credentials secret plus a stub backend-database Service and Deployment, applies the rendered manifests, and waits for the pod to become ready.
  • Verifies runtime image, env vars, probes, init containers, mounted init.sql, and the bootstrapped Postgres schema inside the running pod.

backend-sync:k8sIntegrationTest runs with maxParallelForks = 1 and depends on buildFatJar, k8sBuild, and k8sResource; both k8sBuild and k8sResource are ordered after buildFatJar because JKube needs backend-sync-all.jar, and buildFatJar would otherwise wipe the rendered manifests under build/classes/java/main/META-INF/jkube/kubernetes.

BrokerApiResolver And Broker API Wiring

BrokerApiResolver is the compile-time-safe mapping from the general Broker enum to concrete BrokerApi instances. It lives in backend-broker and requires both OAuthTokenBrokerService and ProvisionedTokenBrokerService in its constructor.

Why Both Token Broker Services Are Required

DefaultBrokerApiResolver constructs all BrokerApi instances eagerly at construction time. UpstoxApiDecorator needs OAuthTokenBrokerService for awaitValidToken() and onUnauthorized on live/feed/history clients, and ProvisionedTokenBrokerService for sandbox HFT order placement. By requiring both services in the resolver's constructor, any module that needs broker API access (e.g., TradeExecutor, DataSourceFactory) transitively guarantees that both token broker connections are available — you cannot construct a BrokerApi without them.

Exhaustive Broker Mapping

DefaultBrokerApiResolver.get(broker) uses an exhaustive when (broker) so that adding a new Broker enum entry is a compile error until a corresponding BrokerApi mapping is provided. This prevents silent no-op fallbacks at runtime.

Auth-Free Instrument Access

Broker.publicInstruments(action) is a top-level extension that fetches instrument catalogs without authentication. UpstoxPublicInstrumentCatalog uses a plain HttpClient (no auth plugins) to fetch from the assets.upstox.com CDN. This is used by backend-sync's LiveInstrumentFeedRefresher instead of DataSourceFactory, since backend-sync does not have token broker service connections.

FeedRepository And LiveFeed Separation

FeedRepository (in backend-repository) is storage/KV-only — it persists FeedIdentifier rows and subscribed-instrument KV state, with no dependency on backend-datasources. The live datasource runtime lives in LiveFeed (in backend-app), which owns DataSourceFactory and the applySubscribedInstruments() method.

AddNewFeedsTask passes the dataSources list from backend-datasources to FeedRepository.addMissingFeeds() for diff-based feed creation. backend-server and backend-repository do not depend on backend-datasources — the live feed runtime is scoped to backend-app only, consistent with the app/server separation invariant.

Token Broker kRPC Pattern

backend-token-broker runs as a standalone Ktor/Netty pod and exposes its services via kRPC at /token-broker. Two service interfaces are exposed: OAuthTokenBrokerService (annotated with @Rpc, protobuf serialization) and ProvisionedTokenBrokerService.

Client-Side Resilience

backend-app connects to the token broker via kRPC in the tokenBroker() Ktor module, which runs before core(). The raw kRPC clients are wrapped in ResilientTokenBrokerService (OAuth) and ResilientProvisionedTokenBrokerService (provisioned), which decorate tokenStateUpdates(broker/handle) with:

  • automatic reconnect with exponential backoff (1 s → 30 s cap, ±20% jitter) using Arrow's Schedule, matching the DataBridgeLauncher reconnect policy
  • shared StateFlow per broker (via shareIn(scope, Eagerly, replay=1)) so that multiple collectors receive the same state — only one kRPC logical stream per broker
  • immediate CancellationException propagation

ResilientTokenStateFlow provides the resilientTokenStateUpdates() extension that applies the Arrow Schedule-based retry with backoff reset on successful emission. If the kRPC WebSocket transport drops, the shared stream re-subscribes automatically rather than going permanently dead.

Resumable datasources are wired into the AuthStateCoordinator via Dagger's @IntoMap with @AuthBrokerKey. Each ResumableDataSource binding is keyed by its AuthBroker, and the coordinator iterates over the injected Map<AuthBroker, ResumableDataSource<*>> to start per-broker token-state collectors. Adding a new broker's resumable datasource requires only a single @Provides @IntoMap @AuthBrokerKey(...) method in DataSourceModule — no changes to the coordinator or DataSourceFactory.

Auth State Coordination

AuthStateCoordinator owns all token-state side effects for the application layer. For each AuthBroker, it manages three behaviors:

  1. Recovery publication — collects OAuthTokenBrokerService.recoveryFlow(broker) wrapped in RecoveryManager.recoverable(), surfacing "please re-authenticate" prompts in the UI.
  2. Obsolete-prompt cleanup — watches OAuthTokenBrokerService.tokenStateUpdates(broker) and calls RecoveryManager.clearLogicalRecovery() when the token transitions away from AwaitingAuth, preventing stale auth prompts from lingering after the token recovers by other means (e.g., auto-refresh).
  3. Datasource resume — watches token-state transitions to Valid and calls ResumableDataSource.resumeIfStopped() on affected datasources so that feeds stuck in a terminal degraded state after auth failure can reconnect.

The coordinator is started via StartAuthStateTask in the startup sequence, before StartDataBridgeTask. ResumableDataSource is an opt-in interface — only datasources that can enter a terminal degraded state (currently UpstoxDataSource) implement it.

Token Health Reporters

Two health reporters surface token state through the @DiagnosticHealth set (published over DataBridge for observability, not pod readiness):

  • OAuthTokenHealthReporter partitions AuthBroker entries into active (present in the ResumableDataSource Dagger map) and inactive at construction time. Active brokers with Valid tokens → HEALTHY; AwaitingAuth/Exchanging → DEGRADED; Failed (non-recoverable) → FAILED (does not allow readiness). Inactive brokers appear in the health message but do not affect aggregate status. If no brokers are active, the reporter returns INACTIVE.
  • ProvisionedTokenHealthReporter iterates all ProvisionedTokenHandle entries. Valid → HEALTHY; Failed (recoverable via Secret rotation) → DEGRADED.

RecoveryManager Coalescing And Lifecycle

RecoveryManager (in backend-recovery) manages recoverable errors that require user action. It has been reworked to fix several lifecycle bugs:

Logical Recovery Coalescing

logicalRecoveries: ConcurrentHashMap<String, Uuid> maps recoveryUrl to recoveryToken UUID. When recoverable() is called, it derives a logicalKey from the error's recoveryUrl and uses computeIfAbsent to reuse the same UUID for concurrent errors targeting the same URL. This ensures that concurrent AwaitingAuth emissions for the same broker produce one published error, not duplicate prompts.

Targeted Recovery

recover() removes only the resolved error from _errorPublisher. It does not rebuild the error set from the recovered UUID's remaining entries — it uses a targeted filter on recoveryToken == uuid. This prevents completing one recovery from wiping unrelated recoverable errors across the system. Empty pending-entry lists are cleaned up by returning null from computeIfPresent, which removes the key entirely.

Out-Of-Band Self-Clear

clearLogicalRecovery(logicalKey) handles the case where a token becomes Valid without explicit recover() being called (e.g., an out-of-band re-auth). It:

  1. removes the logical entry from logicalRecoveries
  2. completes all waiting CompletableDeferreds with true (triggering retry in .recoverable())
  3. removes the pending entries and the bucket
  4. removes the error from errorFlow

This prevents stale auth prompts from persisting indefinitely when the auth condition self-resolves.

Recovery Token Allocation

RecoveryManager.recoverable() allocates the recoveryToken UUID. Callers (including OAuthTokenBrokerService.recoveryFlow()) should pass recoveryToken = null in their RecoverableByUserActionError — RecoveryManager will assign the token based on the logical key. This centralizes token identity in one place and avoids the previous pattern where each recoveryFlow() emission generated a new random UUID, preventing coalescing.