feat(core): add rfq_swap_replacer built-in post-processor

.claude/CODEBASE.md

@@ -64,7 +64,7 @@ See `docs/ARCHITECTURE.md` for the full architecture diagram and detailed compon
    `MostLiquidAlgorithm`, `BellmanFordAlgorithm`, `PathFrankWolfeAlgorithm`, and
    `WaterFillAlgorithm`
    **Post-processors** (`fynd-core/src/post_process/`) — `RoutePostProcessor`s a pool runs on the
-   algorithm's route before validation
+   algorithm's route after the worker validates it; a processed route that fails the checks falls back to the algorithm's route. `rfq_swap_replacer` is built in
 5. **MarketState** (`fynd-core/src/feed/market_data.rs`) — `Arc<RwLock<>>` of all component/token/gas state; accessed via `MarketData` handle
 6. **TychoFeed** (`fynd-core/src/feed/tycho_feed.rs`) — Background task: Tycho WebSocket → MarketState → broadcast events
 7. **Derived Data** (`fynd-core/src/derived/`) — Pre-computed spot prices, component (pool) depths, token gas prices

docs/ARCHITECTURE.md

@@ -187,7 +187,7 @@ Each worker:
 2. Runs a prioritized `select!` loop: shutdown > market events > derived events > solve tasks
 3. Maintains a `ReadinessTracker` for derived data requirements
 4. Calls the algorithm's `find_best_route` with a `SolveRequest`: the local graph, the shared market data, the order, and the pools and tokens the request excluded
-5. Runs the pool's post-processors, in config order, on the route `find_best_route` returned, before it validates the route. Market events also go to each post-processor
+5. Validates the route `find_best_route` returned, then runs the pool's post-processors on it, in config order. If the processed route fails the checks, the worker logs a warning and keeps the algorithm's route. Market events also go to each post-processor
 
 ***
 

docs/guides/server-configuration.md

@@ -196,7 +196,27 @@ All pools solve every incoming order in parallel. Fynd picks the best result acr
 | `timeout_ms`          | `100`           | Maximum time in milliseconds allowed per order processing in this pool |
 | `max_routes`          | _(no limit)_    | Maximum number of candidate routes to evaluate per order               |
 | `connector_tokens`    | _(no restriction)_ | Allowlist of `"0x..."`-prefixed token addresses permitted as intermediate hops. Source and destination are always allowed regardless. Absent = all tokens reachable. |
-| `post_processors`     | _(none)_        | Names of post-processors, in order, that every worker in the pool runs on the route its algorithm returned, before the worker validates the route. The `fynd` binary registers no post-processors; this field is for binaries that embed Fynd. They register the names through `PostProcessorRegistry`. A name that is not registered stops the server from starting |
+| `algorithm_exclude_protocols` | _(no restriction)_ | Protocol systems the pool's algorithm never routes through. Its post-processors still see them. An entry names a protocol system exactly (`"rfq:bebop"`), or a whole family when it ends with `:` (`"rfq:"`). `exclude_protocols`, with the same entries, hides protocols from the whole pool. Without `post_processors` this key does the same as `exclude_protocols` |
+| `post_processors`     | _(none)_        | Names of post-processors, in order, that every worker in the pool runs on the route its algorithm returned, after the worker validates that route. `rfq_swap_replacer` is built into fynd-core; a binary that embeds Fynd registers more names through `PostProcessorRegistry`, and a custom name may not equal a built-in name. If a processed route fails the checks, the worker logs a warning and quotes the algorithm's route. A name that is not registered stops the server from starting |
+
+### Post-processors
+
+`rfq_swap_replacer` replaces one swap, or a run of swaps between two tokens that every path passes through, with one swap through an RFQ component when that raises the net amount out. The server must index `rfq:` or `fallback:rfq:` protocols (`--protocols`, see Including RFQ Protocols), and the pool must not drop them with `exclude_protocols`. Only `rfq:bebop` and `rfq:hashflow` support an input amount smaller than the quoted amount, so they can replace any swap. Every other RFQ protocol can replace only swaps starting at the sell token, where the input amount matches the quoted amount exactly. The default `worker_pools.toml` does not enable it.
+
+```toml
+[pools.most_liquid_rfq]
+algorithm       = "most_liquid"
+post_processors = ["rfq_swap_replacer"]
+```
+
+`algorithm_exclude_protocols` hides protocols from the algorithm only, so `rfq_swap_replacer` still sees them. With it, a pool can reach a protocol only through the replacer. In the example below, Bebop appears in a route only when the replacer adds it, so the pool's quotes show the gain of the replacer alone.
+
+```toml
+[pools.replacer_only]
+algorithm                   = "most_liquid"
+algorithm_exclude_protocols = ["rfq:bebop"]
+post_processors             = ["rfq_swap_replacer"]
+```
 
 ### Connector tokens
 

fynd-core/CLAUDE.md

@@ -9,7 +9,7 @@ applications.
 |-----------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | `algorithm/`          | `Algorithm` trait + built-in `MostLiquidAlgorithm`, `BellmanFordAlgorithm`, `PathFrankWolfeAlgorithm`, `WaterFillAlgorithm`. Pluggable via associated graph types. `AlgorithmConfig` shared by built-ins |
 | `solver.rs`           | `FyndBuilder` assembles the full pipeline (feed + gas + computations + pools + encoder + router). `Solver` runs it                                                                 |
-| `worker_pool/`        | `WorkerPool` manages dedicated OS threads. `SolverWorker` runs a prioritized select loop (shutdown > market events > derived events > tasks). `TaskQueue` is `async_channel`-based. Two rules keep components out of one worker's graph: `should_drop_component` (a free function here) for exclusive components in a `PublicOnly` pool and for the pool's `exclude_protocols`, and `fallback::manager::PammManager` for pAMMs the market does not back. The first holds for the worker's life; the pAMM rule expires, so `PammManager::apply_pamm_admission` re-decides it per market event, in both directions, on `propamm_admissions_total{outcome=admitted|withheld|evicted,pool}`. `price_impact` runs after route finding: it computes `price_impact_bps` from the route token map and each swap's reported spot price, supports linear and split routes, and reads no shared `MarketState`. On failure the worker omits the field, logs at debug, and increments `worker_pool_price_impact_calculations_total{pool,outcome}` |
+| `worker_pool/`        | `WorkerPool` manages dedicated OS threads. `SolverWorker` runs a prioritized select loop (shutdown > market events > derived events > tasks). `TaskQueue` is `async_channel`-based. Two rules govern component admission to a worker's graph: the free function `should_drop_component` excludes exclusive components in a `PublicOnly` pool and the pool's `exclude_protocols`; `fallback::manager::PammManager` excludes pAMMs the market does not back. The algorithm's graph and fallback pricing also exclude `algorithm_exclude_protocols`. Post-processors still see those components and receive market events filtered only by the pool rule. The first holds for the worker's life; the pAMM rule expires, so `PammManager::apply_pamm_admission` re-decides it per market event, in both directions, on `propamm_admissions_total{outcome=admitted|withheld|evicted,pool}`. `price_impact` runs after route finding: it computes `price_impact_bps` from the route token map and each swap's reported spot price, supports linear and split routes, and reads no shared `MarketState`. On failure the worker omits the field, logs at debug, and increments `worker_pool_price_impact_calculations_total{pool,outcome}` |
 | `worker_pool_router/` | `WorkerPoolRouter` allocates the worker pools that serve each order (`allocation`: `OrderClass` matched by `SolverPoolHandle::serves`), fans out to those, drops candidates whose pAMM fallback misses `min_amount_out`, ranks the rest by `amount_out_net_gas` descending; price guard (if enabled) validates in rank order; optionally encodes |
 | `feed/`               | `TychoFeed` (WebSocket → MarketState), `GasPriceFetcher`, `MarketEvent` broadcasting, `ProtocolRegistry`. `component_filter` drops components from one worker's graph topology and incoming events; `exclusivity` classifies exclusive components |
 | `derived/`            | `ComputationManager` runs `SpotPriceComputation` and `TokenGasPriceComputation` in one stage, then `ComponentDepthComputation`, every block. `ReadinessTracker` gates workers until data is fresh |
@@ -18,9 +18,9 @@ applications.
 | `price_guard/`        | Price guard: external price validation for quotes. Sub-modules: `guard` (validation logic), `binance_ws` (Binance WebSocket price provider), `hyperliquid` (Hyperliquid oracle provider), `provider_registry`, `config`, `utils` |
 | `simulation/`         | Encoded-quote simulation, switched on per deployment with `--enable-simulation`. `token_layout` — the storage slots a token keys its balances and allowances on, found by tracing the reads its own view functions make, so a non-standard layout (USDT, stETH shares, a proxy holding state elsewhere, a namespaced mapping such as Base's B20, Solady's seeded keys) can still be funded; the tool crates use it too. `simulator` — runs the quote under `eth_simulateV1` with those overrides, caching one layout per token. A revert logs the full `eth_simulateV1` payload, pinned to the block it ran in, under the opt-in `fynd::simulation_replay` debug target, so the call can be rerun off-line. `revert` — names the error a reverting call carries, replaying it under the call tracer when `eth_simulateV1` drops the payload. `deviation` — the quote-versus-simulation bps figure, shared by the log and the metric |
 | `rpc.rs`              | private — `eth_call` and Tycho-address-to-alloy-address helpers used by `encoding::fee_fetcher` |
-| `replay.rs`           | `replay_route(&Route, &MarketState)` — re-execute an already-built route against a (possibly newer) market state, honoring split fractions and shared-pool depletion. Used by `hindsight` to measure quote-to-execution slippage |
+| `replay.rs`           | `replay_route(&Route, &MarketState)` — re-execute an already-built route against a (possibly newer) market state, honoring split fractions and shared-pool depletion. Used by `hindsight` to measure quote-to-execution slippage. `simulate_swaps` (crate-private) runs a swap list on caller-chosen token and state lookups and returns `SimulatedRoute`, with the amount in, amount out and gas of each swap; `RfqSwapReplacer` uses it |
 | `encoding/`           | `Encoder` wraps `tycho-execution` to produce ABI-encoded calldata (singleSwap, sequentialSwap, Permit2 variants). Each order encodes on its own blocking task, `MAX_CONCURRENT_ENCODES` (32, the same cap `tycho-execution` applies to a batch) at a time, so one order's failure (an RFQ leg whose signed quote does not arrive, say) marks that order `QuoteStatus::EncodingFailed` and leaves every other order's calldata in place. A failure that belongs to one order never fails the call, as with `NoRouteFound` and `PriceCheckFailed`. Optional calldata watermark (`with_calldata_watermark`) appends attribution bytes the EVM ignores; a per-request `EncodingOptions::with_calldata_watermark` (serde-skipped, set by the serving layer) replaces it for that request. Computes `FeeBreakdown` mirroring on-chain `FeeCalculator` logic. `RouterFees`/`SharedRouterFees` hold default + per-client fee rates; `RouterFeeFetcher` refreshes them from the FeeCalculator contract every 5 min. `DisableSlippageTakingSigner` (`disable_slippage_taking.rs`, crate-internal, key from `DISABLE_SLIPPAGE_TAKING_SIGNER_KEY`) signs zero-fee `ClientFee` params when `EncodingOptions::disable_slippage_taking` is set, so the FeeCalculator applies its address's positive-slippage exemption; that also makes the signer the request's fee client (displacing the `tx.origin`/sender fallback in `Encoder::fee_client`) and stamps a `DEFAULT_DEADLINE_WINDOW_SECS` deadline the unsigned path did not have. `DEFAULT_DEADLINE_WINDOW_SECS` (120s, `mod.rs`) is shared with `exclusive_swap.rs`, so a quote carrying both signatures has one expiry |
-| `post_process/`       | `RoutePostProcessor` trait and `PostProcessorRegistry`. A pool config lists post-processor names in `post_processors`; `FyndBuilder::with_post_processors` supplies the registry, and `FyndBuilder` resolves the names, so an unknown name fails the build with `SolverBuildError::UnknownPostProcessor`. Each worker builds its own instances once, on its thread, and holds them in `PostProcessors` (`pipeline.rs`), which runs them. The worker calls `initialize` in `initialize_graph`, passes every filtered market event to `handle_event` after the graph manager, and in `quote` runs them in config order on the algorithm's `RouteResult` before route validation and the route filter check. A post-processor still running 10 ms (`POST_PROCESS_DEADLINE_MARGIN`) before the task deadline is cut off, and the worker keeps its input route, so that route still reaches the router in time. The worker also rejects a processed route that uses a component the pool drops, or (sell order) has `net_amount_out` above its amount out. Each run records `worker_pool_post_processor_duration_seconds{pool,post_processor}` |
+| `post_process/`       | `RoutePostProcessor` trait and `PostProcessorRegistry`. A pool config lists post-processor names in `post_processors`; `FyndBuilder::with_post_processors` supplies the registry, and `FyndBuilder` resolves the names, so an unknown name fails the build with `SolverBuildError::UnknownPostProcessor`. Each worker builds its own instances once, on its thread, and holds them in `PostProcessors` (`pipeline.rs`), which runs them. The worker calls `initialize` in `initialize_graph`, passes every filtered market event to `handle_event` after the graph manager, and in `quote` checks the algorithm's route first (route validation and the route filter, as without post-processors), then runs them in config order on it. `PostProcessRequest::drops_component` gives them the pool's drop rule. A post-processor still running at the post-process deadline, 10 ms (`POST_PROCESS_DEADLINE_MARGIN`) before the task deadline, is cut off, and the worker keeps its input route, so that route still reaches the router in time. A post-processor reads the route by reference and returns `None` to keep it. A processed route that fails route validation, the request route filter, or the worker checks (it uses a component the pool drops, for a sell order has `net_amount_out` above its amount out, or has a `fallback:` leg that fallback pricing rejects) is logged at warn, counted in `worker_pool_post_processed_route_rejected_total{pool}`, and the quote uses the algorithm's route. `rfq_swap_replacer` (`rfq.rs`, crate-private `RfqSwapReplacer`) is built in: `PostProcessorRegistry::resolve` finds it by name, and `with_post_processor` refuses that name. Each run records `worker_pool_post_processor_duration_seconds{pool,post_processor}` and `worker_pool_post_processor_outcomes_total{pool,post_processor,outcome}`, where `outcome` is `replaced`, `kept`, `timed_out` or `skipped` (the post-process deadline passed before it started). A replacement also records `worker_pool_post_processor_gain_bps{pool,post_processor}`, the gain in `net_amount_out` over the route the post-processor received, in bps (not recorded when that net is not positive; it counts a route that the worker checks later reject). `PostProcessors::run` returns a `ProcessedRoute` with the names of the post-processors that replaced the route; the worker puts them on the quote as `OrderQuote::post_processed_by` (empty when the checks reject the route; internal, not on the wire), and the router adds them as the `post_processed_by` label (names joined by `+`, or `none`) to `worker_router_best_quote_pool`, `worker_router_won_volume_gas_token` and `worker_router_unpriced_won_quotes_total` |
 | `types/`              | Core types: `Order`, `Route`, `Swap`, `Quote`, `QuoteRequest`, `BlockInfo`, `EncodingOptions`, `FeeBreakdown`, error types. `RouteExclusionFilter` is the liquidity a request excludes from a route (pools, protocol systems, tokens), empty by default. `MarketState::resolve_route_filter` turns it into the `RouteExclusions` a graph query and an algorithm read, with each protocol system replaced by the pools `MarketState::components_by_protocol` holds for it                                                         |
 
 ## Key Traits

fynd-core/src/encoding/encoder.rs

@@ -30,7 +30,7 @@ use crate::{
         router_fees::{FeeRates, RouterFees, SharedRouterFees},
     },
     fallback::user_data::fallback_user_data,
-    EncodingOptions, FeeBreakdown, OrderQuote, QuoteStatus, SolveError, Transaction,
+    EncodingOptions, FeeBreakdown, OrderQuote, QuoteStatus, Route, SolveError, Transaction,
 };
 
 /// Canonical Permit2 contract address — identical on all EVM chains.
@@ -454,7 +454,10 @@ impl Encoder {
             None => {
                 // Fail fast rather than emit on-chain-invalid unsigned calldata for an
                 // exclusive leg: an exclusive route requires a signature.
-                if has_exclusive_leg(quote) {
+                if quote
+                    .route()
+                    .is_some_and(Route::has_exclusive_leg)
+                {
                     return Err(SolveError::FailedEncoding(
                         "quote routes through an exclusive pool but no signing key is \
                          configured (set EXCLUSIVE_SWAP_CONTROLLER_KEY)"
@@ -1039,17 +1042,6 @@ impl From<EncodingError> for SolveError {
     }
 }
 
-/// Returns whether the quote routes through an exclusive component, i.e. any swap in its route
-/// carries a committed amount.
-fn has_exclusive_leg(quote: &OrderQuote) -> bool {
-    quote.route().is_some_and(|route| {
-        route
-            .swaps()
-            .iter()
-            .any(|s| s.committed_amount_out().is_some())
-    })
-}
-
 #[cfg(test)]
 mod tests {
     use alloy::primitives::{Address as EvmAddress, Bytes as EvmBytes, Signature};

fynd-core/src/feed/protocol_registry.rs

@@ -73,6 +73,19 @@ const PRICE_LEVEL_STREAM_PREFIX: &str = "pricelevelstream:";
 /// `--protocols` entry and the protocol system of the components the client streams.
 pub(crate) const RFQ_PREFIX: &str = "rfq:";
 
+/// Removes the `fallback:` prefix from `protocol_system`, if it has one.
+pub(crate) fn strip_fallback_prefix(protocol_system: &str) -> &str {
+    protocol_system
+        .strip_prefix(FALLBACK_PREFIX)
+        .unwrap_or(protocol_system)
+}
+
+/// Whether `protocol_system` is an RFQ protocol, through its own executor (`rfq:`) or its
+/// fallback router (`fallback:rfq:`).
+pub(crate) fn is_rfq_protocol(protocol_system: &str) -> bool {
+    strip_fallback_prefix(protocol_system).starts_with(RFQ_PREFIX)
+}
+
 /// Marks a `--protocols` entry that drops a protocol system from the list rather than adding one,
 /// e.g. `exclude:vm:fermiswap`.
 pub const EXCLUDE_PREFIX: &str = "exclude:";
@@ -847,10 +860,20 @@ fn get_env(var: &str) -> Result<String, DataFeedError> {
 mod tests {
     use std::sync::{Arc, Mutex};
 
+    use rstest::rstest;
     use tycho_simulation::price_level_stream::config::PRICE_LEVEL_STREAM_FAMILY;
 
     use super::*;
 
+    #[rstest]
+    #[case::rfq("rfq:bebop", true)]
+    #[case::rfq_via_fallback_router("fallback:rfq:bebop", true)]
+    #[case::pamm("fallback:tycho_pamm", false)]
+    #[case::onchain("uniswap_v2", false)]
+    fn test_is_rfq_protocol(#[case] protocol_system: &str, #[case] expected: bool) {
+        assert_eq!(is_rfq_protocol(protocol_system), expected);
+    }
+
     /// A writer that keeps every byte a subscriber formats, so a test can assert on the
     /// rendered log lines.
     #[derive(Clone, Default)]