Live Market Pipeline And Processor Machinery

This document describes the runtime path for live market data inside backend-app, from subscription resolution through strategy execution, trade execution, persistence, and server publishing.

For the broader module map and runtime boundaries around this path, see Architecture Overview.

Scope

The pipeline described here spans:

• startup task ordering in backend-app
• live feed collection and fan-out
• Processor, StrategyExecutor, and StrategyExecution
• trade execution and trade persistence
• raw market-data persistence through the recorder
• event publishing to backend-server through DataBridgeLauncher

It is the source-of-truth overview for the processor-side machinery that the recent EMA and volume_breakout runtime tests exercise.

Diagrams

The following diagrams illustrate the processor and strategy pipeline:

Diagram	Description
Data Flow	Three-tier architecture: Processor → Executor → Execution
Route State Machine	Route lifecycle: Active, Stopping, Degraded, Gone, Fail-Fast
Command Loop	Single-owner control plane: command sources, channel, snapshot
Mailbox Hierarchy	Two-level mailbox architecture and start barrier
Degradation & Fail-Fast	Overflow detection, deduplication, escalation decision

High-Level Flow

QuantApp startup
  -> AddNewFeedsTask / AddNewStrategiesTask
  -> LoadInstrumentNamesTask / SetSubscribedInstrumentsTask
  -> StartDataFlowTask
  -> StartDataBridgeTask

feedRepository.subscribedInstrumentsFlow()
  -> StartDataFlowTask.startFeed()
  -> liveFeed.feed(dataSourceIdentifier, instrumentIds)
  -> merged Flow<FeedData<MarketData>>
       -> Processor
       -> DataRecorder.feedConsumer

Processor
  -> routeFeedData() (sequential batch processing)
  -> StrategyMailbox (Channel<MarketData>) per strategy
  -> StrategyExecutor per strategy
       -> InstrumentMailbox (ProcessorMailbox) per instrument
       -> StrategyExecution per strategy+instrument
  -> StrategyOutput
  -> StrategySignalEvent
  -> TradeExecutor
  -> tradeEvents

DataRecorder
  -> raw market data to QuestDB
  -> trade events to OrdersRepository

DataBridgeLauncher
  -> strategy outputs
  -> strategy signals
  -> market feed events
  -> trade events
  -> open positions
  -> readiness and errors

AuthStateCoordinator
  -> RecoveryManager.recoveryFlow().recoverable() for auth prompts
  -> clearLogicalRecovery() for obsolete prompts
  -> ResumableDataSource.resumeIfStopped() on OAuth Valid

ProvisionedTokenStateCoordinator
  -> ResumableDataSource.resumeIfStopped() on UPSTOX_ANALYTICS Valid

Startup Sequence

QuantApp groups StartupTasks by phase and runs one phase at a time. Tasks inside the same phase run concurrently.

The current startup order is:

1. AddNewFeedsTask and AddNewStrategiesTask
2. LoadInstrumentNamesTask and SetSubscribedInstrumentsTask
3. StartDataFlowTask
4. StartAuthStateTask
5. StartDataBridgeTask

Why this order matters:

• feeds and strategies must exist before subscription resolution starts
• datasource subscription state must be restored before live streaming begins
• the live feed must exist before the server-forwarding bridge starts collecting from it
• token-state side effects (recovery publication, obsolete-prompt cleanup, datasource resume) publish through RecoveryManager.errorFlow, a replaying StateFlow, so they are retained until the data bridge subscribes — StartAuthStateTask and StartDataBridgeTask have no hard ordering dependency between them

Token Broker Connection

The tokenBroker() Ktor module runs before core() and establishes the kRPC connection to backend-token-broker. Both kRPC clients are wrapped in ResilientTokenBrokerService and ResilientProvisionedTokenBrokerService, which provide automatic reconnect with exponential backoff and shared state flows per broker. If the token broker is unavailable at startup, the resilient decorators retry the connection rather than crashing the pod — once the broker becomes reachable, the token-state stream reconnects and the app recovers without a restart.

UpstoxApiDecorator uses awaitValidToken() to suspend HTTP requests when the token state is AwaitingAuth. AuthStateCoordinator collects recoveryFlow(broker) wrapped in RecoveryManager.recoverable() to surface auth prompts in the UI, watches tokenStateUpdates(broker) to clear obsolete prompts via clearLogicalRecovery(), and resumes datasources via ResumableDataSource.resumeIfStopped() when the OAuth token transitions back to Valid (either through user auth or out-of-band refresh).

The market data feed and historical candle data authenticate with the provisioned UPSTOX_ANALYTICS token instead of OAuth. ProvisionedTokenStateCoordinator (started by the same StartAuthStateTask) resumes the feed datasource via ResumableDataSource.resumeIfStopped() when UPSTOX_ANALYTICS transitions back to Valid after a Secret rotation — the analytics token has no OAuth flow, so while it is Failed the feed reports terminal DEGRADED until the operator rotates the Secret.

Feed Reception And Fan-Out

StartDataFlowTask.startFeed() is the main entry point for live market data inside backend-app.

It does three things:

1. Watches feedRepository.subscribedInstrumentsFlow().
2. Resolves each persisted feed group to a DataSourceIdentifier and calls liveFeed.feed(dataSourceIdentifier, instrumentIds).
3. Merges the resulting per-feed flows into one shared Flow<FeedData<MarketData>>.

Important details:

• persisted subscribed instruments come from KVStore and are resolved through DataStore
• each feed flow is wrapped with RecoveryManager.recoverable()
• the merged feed is shared lazily and fanned out to every FeedConsumer

The two main consumers are:

• Processor, which turns ticks into outputs, signals, and trades
• DataRecorder.feedConsumer, which persists raw market data to QuestDB

LiveFeed.feed(...) resolves the concrete datasource through DataSourceFactory, optionally narrows it with the requested instrument list, and then collects the datasource stream.

The main datasource implementations are currently UpstoxDataSource and BitFlyerDataSource. SetSubscribedInstrumentsTask pushes the restored instrument set into each datasource once at startup via LiveFeed.applySubscribedInstruments().

DataSourceFactory receives BrokerApiResolver rather than token broker services directly — it has no auth knowledge. UpstoxApiDecorator (constructed by the resolver) creates three HttpClient.withConfig instances: one for live trading (positions, holdings, live HFT order placement) that suspends on OAuthTokenBrokerService.awaitValidToken(), one for the market data feed and historical candle data that suspends on ProvisionedTokenBrokerService.awaitValidToken(UPSTOX_ANALYTICS), and one for sandbox HFT order placement that suspends on ProvisionedTokenBrokerService.awaitValidToken(UPSTOX_SANDBOX). All three clients use the shared UpstoxErrorParser for 401 classification — it reads the raw response body from the SavedHttpResponse cache and classifies UDAPI100050 as a token error, triggering invalidation of the appropriate token type.

Datasource Health

DataSource exposes both isReady(): Boolean (binary readiness) and health(): DataSourceHealth (structured health with HealthStatus and optional message). DataSourceHealth carries a HealthStatus enum — HEALTHY, DEGRADED, BLOCKED — where DEGRADED means the datasource is partially functional (allows readiness) and BLOCKED means it is non-functional (fails readiness).

UpstoxDataSource overrides health() to report DEGRADED with a terminalFailureMessage when a permanent token failure occurs (TokenUnavailableException from OAuthTokenState.Failed). This prevents infinite re-auth loops for configuration errors while still allowing the pod to report readiness — the feed is degraded, not dead. The datasource does not observe token state itself; AuthStateCoordinator triggers recovery externally via ResumableDataSource.resumeIfStopped() when the token becomes Valid again. UpstoxDataSource clears the terminal failure as soon as the upstream reconnects (when isReadyState becomes true), not only on tick emission — so even a quiet market after auth recovery reports HEALTHY.

Upstox Tick-to-MarketData Conversion

UpstoxDataSource converts raw WebSocket ticks into MarketData instances. Each MarketData carries a per-tick volume delta — not the cumulative total.

The Upstox feed provides vtt (volume traded today) as a cumulative counter that resets at the start of each trading day. To produce the per-tick delta, the datasource maintains a lastTradeStateMap keyed by InstrumentId and computes volume = current.vtt - previous.vtt.

Day-boundary detection. When a new trading day starts, vtt resets and the delta would produce a negative volume. The datasource detects day boundaries with two conditions (either triggers a reset):

1. vtt-decrease: current.vtt < previous.vtt — a cumulative daily counter can only decrease if the day changed. This is the primary trigger.
2. time-delta: current.ltt - previous.ltt > DAY_BOUNDARY_THRESHOLD_MILLIS (12 hours) — catches the case where opening auction volume already exceeds the previous day's total, so vtt does not decrease even though a new day started. Short-circuit evaluation means this computation only runs when vtt did not already detect the boundary.

When a day boundary is detected, durationSinceLast resets to Duration.ZERO and volume is computed differently depending on the scenario:

Scenario	previous	effectivePrevious	Volume	Rationale
Normal tick	non-null	non-null	current.vtt - previous.vtt	Standard delta
Day boundary	non-null	null (vtt decreased or time delta exceeded)	current.vtt	Opening auction volume — the first tick already carries real volume
Cold start (no prior state)	null	null	0.0	Cannot distinguish pre-trading from mid-day restart

Processor Contract

Processor.consume() is the orchestration boundary between live feed input and strategy execution.

At a high level it:

• runs under a coroutineScope with a CompletableDeferred<Nothing> fatal failure signal
• processes feed batches sequentially through routeFeedData()
• watches strategyRepository.strategyInstrumentIds and enqueues RouteCommand.Reconcile commands
• runs a single-owner RouteCommandLoop that handles all route mutations (reconciliation and degradation)
• launches route-side jobs in a supervised child scope (routeSideScope)
• collects strategy outputs and signals into the processor's shared output streams

flowchart TD
    FD["FeedData<MarketData>"]

    subgraph processor ["Processor (backend-app)"]
        FR["FeedRouter — routeFeedData()\nreads routesByInstrument snapshot\ntrySend to strategy mailboxes\noverflow → requestDegrade()"]
        RCL["RouteCommandLoop\nsole writer of routingSnapshot\nReconcile · Degrade · TeardownStrategy\nInstrumentRouteAdded · Removed"]
        RL["ReconcileLoop\nstrategyInstrumentIds → Reconcile"]
        FW["FatalWatcher\nfatalFailure → coroutineScope fails → pod restart"]
        RS["RoutingSnapshot (@Volatile)\ndesired · routesByInstrument · routesByStrategy\nStrategyRoute: routeInstanceId · strategy · mailbox · executor · routeJobs · state"]
    end

    SM["Strategy Mailbox\nChannel<MarketData> · cap 1024\nProcessor-owned"]

    subgraph executor ["StrategyExecutor (backend-processor) — one per strategy route"]
        ER["Router coroutine\nmailbox.receiveAsFlow() → deliver() trySend"]
        subgraph routes ["Instrument Routes"]
            IR["InstrumentRoute\nProcessorMailbox (Channel, cap 1024)\nStrategyExecution (runningFold)\nBridge jobs: output → _strategyOutput · signals → _signals"]
            IR2["InstrumentRoute … (one per instrument)"]
        end
    end

    subgraph execution ["StrategyExecution (backend-strategy)"]
        MDF["marketDataFlow (ProcessorMailbox.flow)"] --> RF["runningFold (accumulate state)"] --> SO["strategyOutput (MutableSharedFlow)"]
        RF --> SS["strategySignals"]
    end

    FD --> FR
    RL -->|"Reconcile"| RCL
    FR -->|"Degrade"| RCL
    FR -.->|"reads"| RS
    RCL -.->|"writes"| RS
    RS --> SM
    SM -->|"receiveAsFlow"| ER
    ER --> IR
    ER --> IR2
    IR --> MDF
    IR -.->|"onRouteOverflow\nonInstrumentRoute*"| FR

    classDef feed fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    classDef processorNode fill:#e8eaf6,stroke:#283593,color:#1a237e
    classDef fatal fill:#ffebee,stroke:#c62828,color:#b71c1c
    classDef mailbox fill:#fff3e0,stroke:#e65100,color:#bf360c
    classDef executorNode fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c
    classDef executionNode fill:#e8f5e9,stroke:#2e7d32,color:#1b5e20

    class FD feed
    class FR,RCL,RL,RS processorNode
    class FW fatal
    class SM mailbox
    class ER,IR,IR2 executorNode
    class MDF,RF,SO,SS executionNode

    style processor fill:#e8eaf6,stroke:#283593,color:#1a237e
    style executor fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c
    style execution fill:#e8f5e9,stroke:#2e7d32,color:#1b5e20
    style routes fill:#ede7f6,stroke:#7b1fa2,color:#4a148c

Processor-Owned Routing

The processor maintains a single @Volatile routing snapshot that is the source of truth for all routing decisions:

RoutingSnapshot
  ├── desired: List<StrategyInstrumentIds>  (repository state, updated only by Reconcile)
  ├── routesByInstrument: ImmutableMap<InstrumentId, ImmutableSet<StrategyId>>  (Eclipse Collections)
  └── routesByStrategy: ImmutableMap<StrategyId, StrategyRoute>  (Eclipse Collections)

StrategyRoute
  ├── strategy: Strategy
  ├── mailbox: Channel<MarketData>       (strategy-level mailbox)
  ├── executor: StrategyExecutor
  ├── routeJobs: RouteJobs              (signalsJob, outputJob, tradeJob, stoppingWatcherJob?)
  └── state: RouteState (Active / Degraded / Stopping)

RouteState
  ├── Active
  ├── Degraded(cause, degradedAt, hasOpenPositions)
  └── Stopping(retainedInstrumentIds)   (strategy removed but positions still open)

Single-owner mutation: Only the RouteCommandLoop coroutine mutates routingSnapshot. The FeedRouter and ReconcileLoop coroutines only read it or enqueue commands. The @Volatile annotation ensures visibility across the writer/reader boundary. The routesByInstrument and routesByStrategy maps use Eclipse Collections' ImmutableMap, which provides structural immutability — the maps cannot be modified after construction, and every mutation produces a new ImmutableMap instance via newWithKeyValue() / newWithoutKey().

Route commands are sent through an unlimited channel:

private val routeCommands = Channel<RouteCommand>(Channel.UNLIMITED)

This is a control queue, not a back-pressure boundary — it must never overflow.

Feed Routing

The FeedRouter coroutine processes feed data sequentially:

1. For each MarketData in a FeedData batch, look up the instrument in routesByInstrument
2. For each strategy mapped to that instrument, trySend into the strategy's mailbox
3. If trySend fails (mailbox full), call requestDegrade(strategyId, routeInstanceId, cause) — this deduplicates against pendingDegradation by route incarnation before enqueuing a RouteCommand.Degrade command
4. Skip routes that are in Degraded state (both Active and Stopping routes receive ticks)

This replaces the previous flatMapMerge { it.data.asFlow() } approach. The key property is that trySend is non-suspending: a slow consumer never blocks the router — it degrades the affected route instead.

Pending degradation deduplication: Because the route is not marked Degraded until the command loop processes the Degrade command, a naive implementation would enqueue a new Degrade command on every failed trySend, building an unbounded control-plane backlog. This applies to both strategy-mailbox overflow (detected in routeFeedData()) and instrument-mailbox overflow (detected in the executor's router and signaled via onRouteOverflow).

To prevent this, the processor centralizes all degrade requests through a requestDegrade(strategyId, routeInstanceId, cause) helper. This helper uses a ConcurrentHashMap.newKeySet<Long>()-backed pendingDegradation set keyed by route instance ID (not strategy ID) for atomic check-and-add: add() returns false if the route incarnation is already pending, so only the first overflow for a given route incarnation enqueues a Degrade command. Keying by route instance ID prevents a stale degrade from an old route incarnation from degrading a newly recreated route for the same strategy. The command loop removes the route instance from pendingDegradation when processing Degrade, TeardownStrategy, or immediate teardown in Reconcile. If a Degrade command arrives for a route whose routeInstanceId no longer matches (the route was recreated), the command is silently discarded.

flowchart LR
    subgraph sources ["Command Sources"]
        RL["ReconcileLoop\nstrategyInstrumentIds watcher\n→ Reconcile(desired)"]
        FR["FeedRouter\nstrategy mailbox overflow\n→ Degrade(strategyId, routeInstanceId)"]
        SE["StrategyExecutor\nonRouteOverflow → Degrade\nonInstrumentRoute* → InstrumentRoute*"]
        SW["StoppingWatcher\nhasOpenPositions filter\n→ TeardownStrategy(strategyId)"]
        RJ["Route-Side Jobs\nsignals/output/trade failure\n→ Degrade(strategyId, routeInstanceId)"]
    end

    CH["routeCommands\nChannel.UNLIMITED\nReconcile · Degrade · TeardownStrategy\nInstrumentRouteAdded · InstrumentRouteRemoved"]

    subgraph loop ["RouteCommandLoop (sole writer)"]
        RC["Reconcile\nRemove / Reactivate / Add routes\nRebuild instrument index"]
        DG["Degrade\nMark Degraded · Close mailbox\nCancel jobs + executor\nRebuild index · Check open positions"]
        TD["TeardownStrategy\nClose mailbox · Cancel jobs\nCancel executor · Remove route"]
        IR["InstrumentRoute*\nRebuild instrument index\nfrom confirmed executor state"]
    end

    RS["RoutingSnapshot\n(@Volatile)"]

    RL --> CH
    FR --> CH
    SE --> CH
    SW --> CH
    RJ --> CH
    CH --> RC
    CH --> DG
    CH --> TD
    CH --> IR
    loop --> RS

    classDef source fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    classDef channel fill:#fff3e0,stroke:#e65100,color:#bf360c
    classDef handler fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c
    classDef snapshot fill:#e8f5e9,stroke:#2e7d32,color:#1b5e20

    class RL,FR,SE,SW,RJ source
    class CH channel
    class RC,DG,TD,IR handler
    class RS snapshot

    style sources fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    style loop fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c

Route Command Loop

The RouteCommandLoop coroutine is the single owner of route state mutations:

for (command in routeCommands) {
    routingSnapshot = when (command) {
        RouteCommand.Reconcile          -> reconcileRoutes(current, desired, routeSideScope)
        RouteCommand.Degrade            -> degradeRoute(current, strategyId, routeInstanceId, cause, fatalFailure)
        RouteCommand.TeardownStrategy   -> teardownStrategy(current, strategyId)
        RouteCommand.InstrumentRouteAdded,
        RouteCommand.InstrumentRouteRemoved -> rebuild routing index from confirmed executor state
    }
}

On RouteCommand.Reconcile:

1. Removal: strategies in the current snapshot but absent from the desired repository state:
2. If the executor has open positions → transition to Stopping(retainedInstrumentIds), launch a StoppingWatcher that posts TeardownStrategy when positions flatten
3. If no open positions → immediate teardown: close mailbox, cancel jobs, cancel executor, remove from snapshot
4. Reactivation: Stopping routes whose strategies reappear in desired are transitioned back to Active (the stopping watcher is cancelled)
5. Addition: strategies in the desired state but absent from the current snapshot get a new strategy mailbox, executor, and route-side jobs
6. Instrument index rebuild: routesByInstrument is rebuilt solely from confirmed executor state (executor.currentInstrumentIds()). The executor notifies the processor of instrument route changes via InstrumentRouteAdded and InstrumentRouteRemoved commands, so the index is always rebuilt after the executor has confirmed the change. Both Active and Stopping routes use the same source: executor current state. Degraded routes are excluded. desired is stored in the snapshot and carried forward by Degrade, TeardownStrategy, InstrumentRouteAdded, and InstrumentRouteRemoved.

On RouteCommand.Degrade:

1. Mark the route as Degraded(cause, hasOpenPositions) (works for both Active and Stopping routes — a Stopping route that overflows with open positions must still reach the fail-fast path)
2. Close the strategy mailbox (terminates the executor's input)
3. Cancel the route-side jobs (signals, output, trade, stoppingWatcher)
4. Cancel the executor (executor.cancel())
5. Rebuild the instrument index using rebuildRoutesByInstrument(routesByStrategy)
6. If the degraded route has open positions → failProcessor() completes the fatal failure signal

On RouteCommand.TeardownStrategy:

1. Close the strategy mailbox
2. Cancel the route-side jobs (signals, output, trade, stoppingWatcher)
3. Cancel the executor
4. Remove the route from the snapshot
5. Rebuild the instrument index

On RouteCommand.InstrumentRouteAdded / RouteCommand.InstrumentRouteRemoved:

1. Rebuild the instrument routing index from confirmed executor state — the executor has confirmed the instrument route change, so executor.currentInstrumentIds() is now up-to-date
2. No changes to routesByStrategy or desired — this is a pure index refresh

These commands are enqueued by the executor's onInstrumentRouteAdded and onInstrumentRouteRemoved callbacks. InstrumentRouteAdded fires after the executor has created a new per-instrument route, launched per-route bridge jobs, started the execution, and updated strategyExecutions — ensuring the processor only routes ticks to instruments whose output/signal fan-in is fully wired. InstrumentRouteRemoved fires when the executor removes an instrument route — either immediately (no open position) or after a retained instrument's position flattens. In both cases, the executor cancels the per-route bridge jobs and updates strategyExecutions before emitting the callback, so the processor rebuilds the routing index from consistent executor state.ithout these commands, the processor's routing index would be derived from a stale snapshot of executor state, causing ticks to be dropped (for additions) or routed to closed routes (for removals).

Batch Ordering

There are two separate ordering rules:

• Batches are processed sequentially because the FeedRouter coroutine collects the feed flow in order.
• Ordering across instruments inside the same batch is not guaranteed, because routeFeedData() iterates over feedData.data and routes each instrument independently.

That means:

• a later batch cannot overtake an earlier batch
• two instruments inside one batch may be observed in either order

The event-contract tests in backend-app deliberately assert the parts of ordering that are guaranteed and avoid asserting the parts that are not.

StrategyExecutor

Each StrategyExecutor owns one strategy and manages one StrategyExecution<T> per active instrument.

Its responsibilities are:

• receive all market data for its strategy through a single Channel<MarketData> (the strategy mailbox)
• fan out internally to per-instrument ProcessorMailbox instances (bounded Channel, capacity 1024)
• signal overflow via onRouteOverflow — the processor decides what to do (degrade the route)
• expose hasOpenPositions(): Boolean so the processor can decide escalation severity
• expose hasOpenPositionsFlow: Flow<Boolean> for reactive position monitoring (used by StoppingWatcher)
• expose cancel() to stop all executor-owned coroutines when the route is torn down
• expose start() to begin the router and instrument reconciliation loop after the processor has subscribed to outputs/signals

Strategy Mailbox Contract

The executor's public contract is:

StrategyExecutor(
    strategyMailbox: Channel<MarketData>,
    instrumentIdsFlow: Flow<List<Pair<InstrumentId, DataAvailabilityWindow>>>,
    historicalDataProvider: HistoricalDataProvider,
    onRouteOverflow: (InstrumentId, Throwable) -> Unit,
    onInstrumentRouteAdded: (InstrumentId) -> Unit,
    onInstrumentRouteRemoved: (InstrumentId) -> Unit,
)

The processor provides the strategy mailbox, the overflow callback, and the instrument-route-lifecycle callbacks. The executor does not own route state — it only signals overflow conditions and per-instrument route lifecycle events (added/removed). The processor uses these callbacks to maintain a routing index derived from confirmed executor state, ensuring ticks are only routed to instruments the executor is ready to receive.

The executor is constructed in a stopped state — no coroutines are launched until start() is called. The processor subscribes to executor.strategyOutput and executor.signals before calling executor.start(), ensuring no emissions are lost when instrument routes are added and their executions start producing output.

Internal Routing

The executor runs a single router coroutine (launched by start()) that:

1. Collects from strategyMailbox.receiveAsFlow()
2. Looks up the instrument in its instrumentRoutes map
3. Delivers to the per-instrument ProcessorMailbox via deliver() (which uses trySend)
4. If deliver() fails, calls onRouteOverflow(instrumentId, cause)

Per-Route Bridge Jobs

Each instrument route has two bridge jobs that forward the execution's output and signals to the executor's shared flows:

1. outputBridgeJob: collects from execution.strategyOutput and emits to _strategyOutput
2. signalBridgeJob: collects from execution.strategySignals and emits to _signals

These bridge jobs replace the previous executor-wide flatMapLatest { merge() } fan-in. The per-route approach eliminates the re-subscription gap: when a new instrument route is added, the bridge jobs are launched with CoroutineStart.UNDISPATCHED before execution.start() is called. UNDISPATCHED ensures the coroutine executes synchronously up to its first suspension point (the collect(...) call), so the subscription is established before start() returns. When the first tick arrives and the execution produces its first output/signal, the bridge jobs are already subscribed. No emissions are lost.

When an instrument route is removed, the bridge jobs are cancelled before the execution is cancelled, ensuring clean teardown.

Runtime Lifecycle Behavior

The runtime-lifecycle tests currently document an important behavior contract:

• removing an instrument or strategy does not immediately kill an execution if that execution still owns an open position
• the retained execution stays alive long enough to emit the managed exit or reversal
• re-adding before cleanup preserves the existing retained execution
• re-adding after cleanup creates a fresh execution and re-runs initialization/backfill

This behavior is intentional and is one of the main reasons the processor runtime-lifecycle suite exists.

Executor Lifecycle

The executor owns its CoroutineScope (created with SupervisorJob as a child of the calling coroutine's job). All internal coroutines (router, instrument reconciliation, per-route bridge jobs) are launched in this scope.

The executor is constructed in a stopped state. The processor must call start() after subscribing to executor.strategyOutput and executor.signals (via launchRouteJobs). The processor's route-side collector jobs (signals, output, trade) are launched with CoroutineStart.UNDISPATCHED, ensuring their subscriptions are established synchronously before start() is called. start() launches the router coroutine and the instrumentIdsFlow collector, which triggers the first reconcileExecutions and begins routing ticks.

Calling cancel() on the executor cancels its scope, which stops all internal coroutines. The processor calls executor.cancel() in the immediate teardown path, the deferred teardown path (TeardownStrategy), and the degradation path.

StrategyExecution

StrategyExecution<T> is the concrete state machine for one strategy on one instrument.

It:

• pre-fills historical state through HistoricalDataProvider
• runs a runningFold over live MarketData
• emits one StrategyOutput for each accepted live tick
• derives long and short signal streams from state transitions

In production, HistoricalDataProvider is backed by Recorder.historicalDataProvider, so strategy backfill reads from QuestDB.

StrategyExecution is constructed in a stopped state — the runningFold and signal collection coroutines are not launched until start() is called. The per-route bridge jobs in StrategyExecutor are launched with CoroutineStart.UNDISPATCHED before start() is called, ensuring their subscriptions are established synchronously. This makes the start() barrier deterministic: no early emissions can be lost because the subscribers are guaranteed to be attached before any data flows.

flowchart LR
    subgraph overflow ["Overflow Sources"]
        A["Strategy Mailbox Overflow\nrouteFeedData() → trySend fails\nChannel<MarketData> · cap 1024"]
        B["Instrument Mailbox Overflow\nexecutor router → deliver() fails\nProcessorMailbox · cap 1024"]
        C["Route-Side Job Failure\nsignals/output/trade collector\nthrows non-CancellationException"]
    end

    subgraph dedup ["Dedup Gate"]
        RD["requestDegrade()\npendingDegradation: ConcurrentHashMap.newKeySet()\nkeyed by routeInstanceId\nAt most 1 pending Degrade per route incarnation\nStale incarnations automatically rejected"]
    end

    subgraph cmdloop ["Command Loop"]
        DR["degradeRoute()\n① Verify routeInstanceId matches\n② Mark route state = Degraded\n③ Close strategy mailbox\n④ Cancel route-side jobs\n⑤ Cancel executor\n⑥ Rebuild routesByInstrument\n⑦ Remove from pendingDegradation\n⑧ Check hasOpenPositions()"]
    end

    A --> RD
    B --> RD
    C --> RD
    RD -->|"RouteCommand.Degrade"| DR
    DR -->|"!hasOpenPositions"| OK["No Open Positions\nRoute removed from routingSnapshot\nSibling strategies continue\nProcessor remains healthy"]
    DR -->|"hasOpenPositions"| FF["Open Positions Present\nfailProcessor() → fatalFailure\nProcess terminates\nKubernetes restarts pod"]

    classDef overflow fill:#ffebee,stroke:#c62828,color:#b71c1c
    classDef dedup fill:#fff3e0,stroke:#e65100,color:#bf360c
    classDef cmd fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c
    classDef ok fill:#e8f5e9,stroke:#2e7d32,color:#1b5e20
    classDef fail fill:#ffcdd2,stroke:#c62828,color:#b71c1c

    class A,B,C overflow
    class RD dedup
    class DR cmd
    class OK ok
    class FF fail

    style overflow fill:#ffebee,stroke:#c62828,color:#b71c1c
    style dedup fill:#fff3e0,stroke:#e65100,color:#bf360c
    style cmdloop fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c

Degradation And Fail-Fast

The degradation path is:

1. Overflow detected → requestDegrade(strategyId, routeInstanceId, cause) — atomically checks pendingDegradation (keyed by route instance ID) and enqueues RouteCommand.Degrade only if not already pending for that route incarnation. This covers both strategy-mailbox overflow (in routeFeedData()) and instrument-mailbox overflow (via onRouteOverflow callback from the executor).
2. Command processed on the single-owner RouteCommandLoop — pendingDegradation is cleared. If the route's routeInstanceId no longer matches the command's (the route was recreated), the stale command is silently discarded:
3. Route marked Degraded(cause, hasOpenPositions)
4. Strategy mailbox closed (terminates the executor's input)
5. Route-side jobs (signals, output, trade) cancelled
6. Executor cancelled (executor.cancel())
7. Escalation decision:
8. Open positions → failProcessor() completes the fatalFailure signal → ProcessorFatalWatcher propagates the exception → coroutineScope fails → Kubernetes detects failure → pod restart (fail-fast)
9. No open positions → route is removed from active routing, processor stays healthy for sibling strategies

This means one degraded strategy does not stall sibling strategies unless it has open positions that require the processor to fail-fast.

Strategy Removal With Open Positions

When a strategy is removed from the repository while its executor has open positions:

1. The route transitions to Stopping(retainedInstrumentIds) instead of being torn down immediately (routes already in Stopping are skipped to prevent duplicate watchers)
2. The strategy mailbox stays open — the executor continues receiving ticks for retained instruments
3. A StoppingWatcher coroutine monitors executor.hasOpenPositionsFlow and posts RouteCommand.TeardownStrategy when positions flatten. The watcher double-checks executor.hasOpenPositions() before posting to guard against premature false emissions from hasOpenPositionsFlow before executor initialization
4. routeFeedData() continues routing to Stopping routes (only Degraded routes are skipped)
5. The instrument index uses executor.currentInstrumentIds() for both Active and Stopping routes (the index is rebuilt from confirmed executor state via InstrumentRouteAdded/InstrumentRouteRemoved commands)
6. When a retained instrument's position flattens inside the executor, the per-route bridge jobs are cancelled, the execution is cancelled, strategyExecutions is updated synchronously, and onInstrumentRouteRemoved enqueues RouteCommand.InstrumentRouteRemoved, which rebuilds the routing index to remove the flattened instrument
7. When a new instrument route is created inside the executor, per-route bridge jobs are launched before execution.start(), strategyExecutions is updated synchronously, and onInstrumentRouteAdded enqueues RouteCommand.InstrumentRouteAdded, which rebuilds the routing index to include the new instrument
8. When TeardownStrategy is processed: mailbox closed, jobs cancelled, executor cancelled, route removed
9. If a Stopping route's mailbox overflows, degradeRoute() handles it — if it has open positions, fail-fast is triggered

If a Stopping route's strategy reappears in the repository (re-added), the route transitions back to Active and the stopping watcher is cancelled.

Route-side jobs are launched in a supervised child scope (routeSideScope) with inline try/catch: if a signals, output, or trade job fails with a real exception (not CancellationException), it enqueues a RouteCommand.Degrade for that strategy.

stateDiagram-v2
    [*] --> Active

    Active --> Stopping : removed from repo + hasOpenPositions
    Stopping --> Active : strategy reappears in repo
    Active --> Degraded : mailbox overflow / route-side job failure
    Active --> Gone : removed from repo + !hasOpenPositions
    Stopping --> Degraded : overflow while stopping
    Stopping --> Gone : TeardownStrategy (positions flattened)
    Degraded --> Gone : reconciliation removes route
    Degraded --> FailFast : hasOpenPositions

    classDef active fill:#c8e6c9,stroke:#2e7d32,color:#1b5e20
    classDef stopping fill:#ffe0b2,stroke:#e65100,color:#bf360c
    classDef degraded fill:#ffccbc,stroke:#d84315,color:#bf360c
    classDef gone fill:#eceff1,stroke:#546e7a,color:#37474f
    classDef failfast fill:#ffcdd2,stroke:#c62828,color:#b71c1c

    class Active active
    class Stopping stopping
    class Degraded degraded
    class Gone gone
    class FailFast failfast

Readiness And Health

The processor's readiness and health are route-aware:

Condition	isReady()	componentHealth()
No core jobs	false	INACTIVE
Any core job not active (fatalWatcher, routeCommandLoop, feedRouter, reconcileLoop)	false	BLOCKED
Degraded route with open positions	false	BLOCKED
Degraded route without open positions	true	DEGRADED
Stopping route (positions winding down)	true	DEGRADED
All routes active	true	HEALTHY

Readiness reflects processor-core health, not every route-side job. A degraded route without open positions does not make the processor unready. A stopping route (strategy removed but positions still open) does not make the processor unready — it is still functioning correctly, just winding down.

flowchart TD
    subgraph hierarchy ["Mailbox Hierarchy"]
        FD["FeedData<MarketData>"] --> PR["Processor · routeFeedData()"] -->|"trySend"| SM["Strategy Mailbox\nChannel<MarketData> · cap 1024\nowned by Processor (backend-app)"]
        SM -->|"receiveAsFlow"| ER["StrategyExecutor · Router coroutine"] -->|"deliver() trySend"| IM["Instrument Mailbox\nProcessorMailbox · Channel · cap 1024\nowned by StrategyExecutor (backend-processor)"] -->|"mailbox.flow"| SX["StrategyExecution\nrunningFold → strategyOutput · signals"]
    end

    SM -.->|"overflow"| D1["requestDegrade()"]
    IM -.->|"overflow"| D2["onRouteOverflow()"]

    classDef data fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    classDef processor fill:#e8eaf6,stroke:#283593,color:#1a237e
    classDef mailbox fill:#fff3e0,stroke:#e65100,color:#bf360c
    classDef executor fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c
    classDef execution fill:#e8f5e9,stroke:#2e7d32,color:#1b5e20
    classDef overflow fill:#ffebee,stroke:#c62828,color:#b71c1c

    class FD data
    class PR processor
    class SM,IM mailbox
    class ER executor
    class SX execution
    class D1,D2 overflow

    style hierarchy fill:#fafafa,stroke:#9e9e9e,color:#424242

flowchart TD
    S1["① Processor · launchRouteJobs()\nsignalsJob · outputJob · tradeJob\nlaunched with CoroutineStart.UNDISPATCHED\n→ collect() reached synchronously"] --> S2["② strategyExecutor.start()\nLaunches Router + instrumentIdsFlow collector\nProcessor already subscribed to outputs/signals"] --> S3["③ reconcileExecutions() — per instrument\na. Construct StrategyExecution (stopped)\nb. Launch bridge jobs (UNDISPATCHED) → subscribe synchronously\nc. execution.start() → pipelines launch after subscribers attached"] --> S4["④ Executor callbacks → RouteCommand.InstrumentRoute*\nonInstrumentRouteAdded / onInstrumentRouteRemoved\nRouteCommandLoop rebuilds routesByInstrument from executor state"]

    classDef step1 fill:#e3f2fd,stroke:#1565c0,color:#0d47a1
    classDef step2 fill:#e8eaf6,stroke:#283593,color:#1a237e
    classDef step3 fill:#f3e5f5,stroke:#6a1b9a,color:#4a148c
    classDef step4 fill:#e8f5e9,stroke:#2e7d32,color:#1b5e20

    class S1 step1
    class S2 step2
    class S3 step3
    class S4 step4

Mailbox Hierarchy

The routing model uses a two-level mailbox architecture:

Level	Type	Capacity	Owner	Location
Strategy mailbox	Channel<MarketData>	1024	Processor	backend-app (raw Channel)
Instrument mailbox	ProcessorMailbox (wraps Channel<MarketData>)	1024	StrategyExecutor	backend-processor (internal)

The strategy mailbox is a raw Channel in backend-app because ProcessorMailbox is internal to backend-processor. The executor's instrument mailboxes use ProcessorMailbox since they're within the same module.

Both levels use trySend for non-suspending delivery. Overflow is detected immediately via ChannelResult.isFailure, not by suspending the sender. This means a slow consumer never blocks the router — it degrades the affected route instead.

Timing And Event Contracts

There are three time concepts that matter across the pipeline:

1. StrategyOutput.time

This is the logical strategy time:

• for EMA, it is the active strategy bucket time
• for volume_breakout, it is the synthetic volume-bar time

2. StrategyOutput.occurrenceTime

This is the live tick time that caused the output.

This field is the canonical way to recover the real market event time that produced an output.

3. StrategySignalEvent.time

Signals currently use the app's emission time, not the source tick time.

The processor-path tests assert this as an emission window:

• signal time must fall between "emission started" and "emission observed"

Signals link back to outputs through strategyOutputId, so downstream code can recover the source tick context from the linked StrategyOutput.occurrenceTime.

Strategy-Specific Runtime Semantics

Two strategy timing families currently matter at the processor boundary.

EMA

ema is live and intra-bucket:

• the active EMA bucket is keyed from the incoming live tick time
• entries can happen before the next timeframe rollover tick arrives
• reversals are expected to emit exit first and opposite entry second

Volume Breakout

volume_breakout is close-gated:

• partial live volume bars may update the latest StrategyOutput
• entry and exit signals emit only when the current synthetic volume bar closes
• delayed re-entry and managed exits are part of the contract

Trade Execution And Trade Persistence

Trade execution is downstream of signals, not outputs.

The runtime flow is:

1. Processor emits StrategySignalEvents.
2. TradeExecutor consumes those signals using the strategy-specific trade settings selected from ordersRepository.tradeSettingUpdates.
3. TradeExecutor calls the broker and emits tradeEvents.
4. DataRecorder.tradeConsumer persists those trade events through OrdersRepository.saveOrder(...).

Important behavioral contract:

• outputs and signals can still exist even when a trade is not executed
• missing trade settings or broker failures should not suppress outputs or signals
• those situations only affect trade execution and trade-event emission

The resilience tests in backend-app cover those cases explicitly.

Recorder Path

DataRecorder.feedConsumer is the raw market-data persistence path.

It:

• flattens the feed stream with flatMapMerge
• converts it to a ReceiveChannel
• passes the channel to Recorder.recordMarketData()

Ingester then writes QuestDB rows with:

• instrument_id
• duration_since_last
• ltp
• volume
• ts

Trade persistence is separate from QuestDB ingestion. Trade events are stored through OrdersRepository in the main application datastore.

Server Publishing Through DataBridgeLauncher

DataBridgeLauncher.launch() starts one coroutine per event family and forwards app state to backend-server.

The main forwarded streams are:

• strategy signal events
• strategy output events
• market feed events
• trade events
• open positions
• readiness
• error snapshots
• recovery events

Three details matter operationally:

• DataBridgeLauncher does not reuse the shared feed flow from StartDataFlowTask; it opens fresh liveFeed.feed(...) collectors for market-feed forwarding
• readiness published to the server is the diagnostic health set, not the pod-readiness set
• DataBridgeLauncher does not keep a top-level eager kRPC client; launch() starts one lazy background supervisor job that repeatedly creates a fresh session with Arrow retry, marks readiness from the active session, and cancels cleanly on ApplicationStopPreparing. All bridge traffic goes to ApplicationConfig.INTERNAL_EVENTS_HOST/PORT/PATH.
• AuthStateCoordinator runs alongside DataBridgeLauncher at app-lifetime scope, started by StartAuthStateTask. It collects recoveryFlow().recoverable() for auth prompt publication, watches tokenStateUpdates() for obsolete-prompt cleanup, and resumes datasources when the OAuth token recovers.
• ProvisionedTokenStateCoordinator is started by the same StartAuthStateTask. It resumes the market-feed datasource via ResumableDataSource.resumeIfStopped() when UPSTOX_ANALYTICS transitions to Valid after a Secret rotation (the analytics token powers the market feed and historical candle data).

Current Guarantees And Non-Guarantees

Guaranteed:

• feed batches are processed sequentially
• EMA remains live and intra-bucket
• volume_breakout remains close-gated
• reversal-capable strategies emit exit before opposite entry
• open-position removals retain executions until managed cleanup completes
• a slow consumer never blocks the router (non-suspending trySend delivery)
• route degradation with open positions triggers processor fail-fast
• UpstoxDataSource detects day boundaries via vtt-decrease or time-delta heuristic, preventing negative volume deltas at session open and preserving opening auction volume
• DEGRADED datasources do not flip pod readiness red — HealthStatus.DEGRADED.allowsReadiness returns true, so a feed with a terminal token failure reports degraded health without blocking the readiness endpoint

Not guaranteed:

• deterministic ordering across instruments inside the same batch
• reuse of the same live-feed collector between local processing and bridge publishing
• correct day-boundary detection for non-equity Upstox segments (NSE_COM, MCX_FO) whose inter-session gaps may be shorter than the 12-hour threshold (QAPP-113)

Token Auth And Recovery Flow

When the token broker transitions a broker's state to AwaitingAuth, the recovery pipeline activates:

1. OAuthTokenBrokerService.recoveryFlow(AuthBroker.UPSTOX) emits RecoverableErrorException, collected by AuthStateCoordinator via RecoveryManager.recoverable()
2. RecoveryManager publishes the error to errorFlow, surfaced by DataBridge to backend-server
3. The UI displays the auth URL and accepts the auth code
4. The code flows back through RecoveryRouter → DataBridge.recoveryEvents() → RecoveryManager.recover() → TokenRecoveryAction.recover() → OAuthTokenBrokerService.submitAuthCode()
5. The token broker exchanges the code for an AccessToken, transitions state to Valid
6. AuthStateCoordinator detects the Valid transition and calls ResumableDataSource.resumeIfStopped(); awaitValidToken() also resumes in UpstoxApiDecorator, and the market data feed reconnects

RecoveryManager coalesces by recoveryUrl — concurrent AwaitingAuth emissions for the same broker share one recoveryToken UUID and one published error. The recover() method removes only the resolved error and preserves all other active recoverable errors across all recovery keys. Out-of-band transitions (token becomes Valid without explicit recovery) are handled by clearLogicalRecovery(), which removes the stale prompt, resumes suspended .recoverable() collectors, and cleans up the logical recovery entry.

Terminal Token Failure

Permanent OAuth errors (e.g., UDAPI100069 — wrong client ID/secret) throw NonRecoverableTokenException from UpstoxTokenExchangeStrategy.exchange(). The TokenManager surfaces this as OAuthTokenState.Failed, and awaitValidToken() throws TokenUnavailableException. UpstoxDataSource catches this in its outer .catch handler, sets a terminalFailureMessage (reporting DEGRADED health), and does not rethrow — this contains the failure and prevents infinite retry loops. The terminalFailureMessage is cleared only after the datasource successfully receives and maps upstream data following recovery.

Blackout Period

UpstoxTokenExchangeStrategy enforces a blackout period (3:30 AM–8:00 AM IST) via isTokenRequestAllowed(). During this window, submitAuthCode() throws TokenBlackoutException. Tokens expire at 3:30 AM IST daily; the expiry timer in TokenManager proactively transitions Valid → AwaitingAuth without waiting for a 401.

Provisioned Token Flow

Provisioned tokens — UPSTOX_SANDBOX for sandbox HFT order placement and UPSTOX_ANALYTICS for the market data feed and historical candle data — follow a different lifecycle than OAuth tokens:

• Tokens are mounted from a Kubernetes Secret as JSON files at /var/run/secrets/token-broker/provisioned
• ProvisionedTokenManager loads and validates the files at startup; if the file is missing, expired, or unparseable, the manager enters Failed and starts a 30-second poll loop
• On 401, UpstoxApiDecorator's sandbox client calls ProvisionedTokenBrokerService.invalidateToken(UPSTOX_SANDBOX) which re-reads the file; if the fingerprint is unchanged, it transitions to Failed and the poller watches for Secret rotation
• The analytics client follows the same pattern: on 401 it calls invalidateToken(UPSTOX_ANALYTICS). While the handle is Failed, awaitValidToken() throws TokenUnavailableException, surfacing as terminal DEGRADED in UpstoxDataSource; ProvisionedTokenStateCoordinator resumes the feed when the handle returns to Valid after the Secret is rotated
• When the operator rotates the Secret, the kubelet refreshes the mounted files and the poller recovers on its next cycle — no pod restart is required
• ProvisionedTokenManager uses a mutex-owned state machine: becomeValidUnderLock cancels any running poller and old expiry timer, publishes Valid, and starts a new expiry timer; becomeFailedUnderLock cancels the expiry timer, publishes Failed, and ensures exactly one poller is running
• The expiry timer reacquires the lock and checks token identity before transitioning to Failed — stale timers from old tokens are silently ignored
• Transient I/O errors (file temporarily unreadable) clear lastFingerprint to null, so the poller will detect when the same bytes reappear after the I/O issue resolves

Feed And Repository Separation

FeedRepository 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.

Public Instruments

Broker.publicInstruments(action) is a top-level extension in backend-broker 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 OAuthTokenBrokerService connections.

Code Map

If you need to trace the machinery in code, start here:

• backend-app: startup tasks, feed fan-out, processor wiring, data bridge launcher, token broker connection, LiveFeed (live datasource runtime), AuthStateCoordinator (OAuth auth recovery → datasource resume), ProvisionedTokenStateCoordinator (analytics-token recovery → market-feed resume), ResilientTokenBrokerService + ResilientProvisionedTokenBrokerService (kRPC reconnect + shared state flows)
• backend-app (Processor.kt): RoutingSnapshot, StrategyRoute (with routeInstanceId), RouteCommand (Reconcile/Degrade/TeardownStrategy), RouteState (Active/Degraded/Stopping), RouteJobs, CoreJobs (private types), feed routing, route command loop, reconciliation, degradation, stopping lifecycle, and fail-fast logic
• backend-token-broker: TokenManager (token state machine, expiry timer, blackout, write-through persistence), OAuthTokenBrokerServiceImpl (kRPC service), ProvisionedTokenManager (mutex-owned state machine with @Volatile lastFingerprint, file-backed polling recovery, expiry timer with identity check), ProvisionedTokenBrokerServiceImpl (kRPC service), UpstoxTokenExchangeStrategy (OAuth token exchange, error classification), AuthBroker (type-safe key for auth-requiring brokers), ProvisionedTokenHandle (type-safe key for provisioned-token handles)
• backend-broker: BrokerApiResolver (compile-time-safe Broker→BrokerApi mapping, requires both OAuthTokenBrokerService and ProvisionedTokenBrokerService), UpstoxApiDecorator (three HttpClient.withConfig instances: OAuth for live trading, UPSTOX_ANALYTICS for market feed + historical, UPSTOX_SANDBOX for sandbox HFT; UpstoxErrorParser for shared 401 classification), UpstoxPublicInstrumentCatalog (auth-free instrument fetch), Broker.publicInstruments() extension
• backend-processor: StrategyExecutor (with start(), per-route bridge jobs, InstrumentRoute), ProcessorMailbox (internal)
• backend-strategy: StrategyExecution (with start()), EmaStrategyExecution, VolumeBreakoutStrategyExecution
• backend-trade-executor: TradeExecutor (receives BrokerApiResolver)
• backend-datasources: UpstoxDataSource (tick-to-MarketData conversion, vtt delta, day-boundary detection, DEGRADED health on terminal token failure, ResumableDataSource), BitFlyerDataSource, DataSourceFactory (receives BrokerApiResolver)
• backend-repository: FeedRepository (storage/KV-only, no datasource dependency)
• backend-recovery: RecoveryManager (coalescing by recoveryUrl, clearLogicalRecovery() for out-of-band transitions, targeted recover() that preserves unrelated errors)
• backend-recorder: DataRecorder, Recorder, Ingester

Related Documents

• Architecture Overview
• Backend Conventions — BrokerApiResolver wiring, FeedRepository split, RecoveryManager coalescing
• Testing Guide
• GitHub Issue Workflow