mirror of
https://github.com/luxfi/dex.git
synced 2026-09-24 00:26:31 +00:00
The surface answers for the pools it reads, and stops answering for a header
Three things this gateway said were not true, all of them found by putting it
behind api.lux.cloud and asking it the questions a screen asks.
/v1/admin/* paused the DEX, paused a pool and froze one — "IRREVERSIBLE", its
own word — on the strength of `X-User-Role: admin`, a header the caller writes.
Nothing read the state it wrote: PauseState had no caller outside its own file,
so the freeze froze nothing and the pause paused nothing, per replica, until the
process restarted. A switch wired to nothing is worse than no switch, because in
an incident someone pulls it, reads {"status":"paused"}, and stops looking. The
real gate is checkPauseState in the pool manager, on chain, where a market is.
Deleted rather than authenticated: there is nothing here to protect.
/healthz reported the hosted provider registry alone. This deployment registers
no hosted provider on purpose — the trade API answers ACCESS_DENIED without a
contract — so the platform's own health path answered 503 forever about a
gateway quoting seven chains, and a readiness probe on it holds every replica
out of its own Service while every quote works. It reads the venues now, and
names them.
/v1/quotes is /v1/quote's question without the last step, and it was reading a
different source: the registry, so a chain we settle ourselves answered "no
providers available" while /v1/quote priced the same pair from the same pools a
line away. One source now — venueQuotes returns what each venue holds, best
first, and venueQuote is its head.
Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
@@ -1,397 +0,0 @@
|
||||
// Copyright (C) 2025, Lux Industries Inc. All rights reserved.
|
||||
// See the file LICENSE for licensing terms.
|
||||
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// AdminRole is the required role in X-User-Role header for admin operations.
|
||||
const AdminRole = "admin"
|
||||
|
||||
// pauseStateGlobal is the package-level pause state singleton.
|
||||
// Thread-safe: protected by internal RWMutex.
|
||||
var pauseStateGlobal = &PauseState{
|
||||
poolPaused: make(map[string]*PoolPauseInfo),
|
||||
poolFrozen: make(map[string]*PoolFreezeInfo),
|
||||
}
|
||||
|
||||
// GetPauseState returns the global pause state for external access.
|
||||
func GetPauseState() *PauseState { return pauseStateGlobal }
|
||||
|
||||
// PauseState tracks the current pause/freeze state for the gateway layer.
|
||||
type PauseState struct {
|
||||
mu sync.RWMutex
|
||||
dexPaused bool
|
||||
dexPausedAt time.Time
|
||||
dexPausedBy string
|
||||
poolPaused map[string]*PoolPauseInfo
|
||||
poolFrozen map[string]*PoolFreezeInfo
|
||||
}
|
||||
|
||||
// PoolPauseInfo tracks when and by whom a pool was paused.
|
||||
type PoolPauseInfo struct {
|
||||
PoolID string `json:"poolId"`
|
||||
PausedAt time.Time `json:"pausedAt"`
|
||||
PausedBy string `json:"pausedBy"`
|
||||
}
|
||||
|
||||
// PoolFreezeInfo tracks when and by whom a pool was permanently frozen.
|
||||
type PoolFreezeInfo struct {
|
||||
PoolID string `json:"poolId"`
|
||||
FrozenAt time.Time `json:"frozenAt"`
|
||||
FrozenBy string `json:"frozenBy"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
|
||||
// PoolStatusResponse is the response for GET /v1/admin/pool/status.
|
||||
type PoolStatusResponse struct {
|
||||
PoolID string `json:"poolId"`
|
||||
DEXPaused bool `json:"dexPaused"`
|
||||
Paused bool `json:"paused"`
|
||||
Frozen bool `json:"frozen"`
|
||||
PauseInfo *PoolPauseInfo `json:"pauseInfo,omitempty"`
|
||||
FreezeInfo *PoolFreezeInfo `json:"freezeInfo,omitempty"`
|
||||
}
|
||||
|
||||
// DEXStatusResponse is the response for GET /v1/admin/status.
|
||||
type DEXStatusResponse struct {
|
||||
DEXPaused bool `json:"dexPaused"`
|
||||
DEXPausedAt *time.Time `json:"dexPausedAt,omitempty"`
|
||||
DEXPausedBy string `json:"dexPausedBy,omitempty"`
|
||||
PausedPoolCount int `json:"pausedPoolCount"`
|
||||
FrozenPoolCount int `json:"frozenPoolCount"`
|
||||
PausedPools []*PoolPauseInfo `json:"pausedPools,omitempty"`
|
||||
FrozenPools []*PoolFreezeInfo `json:"frozenPools,omitempty"`
|
||||
}
|
||||
|
||||
// IsDEXPaused returns whether the entire DEX is paused.
|
||||
func (ps *PauseState) IsDEXPaused() bool {
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
return ps.dexPaused
|
||||
}
|
||||
|
||||
// IsPoolPaused returns whether a specific pool is paused.
|
||||
func (ps *PauseState) IsPoolPaused(poolId string) bool {
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
_, ok := ps.poolPaused[poolId]
|
||||
return ok
|
||||
}
|
||||
|
||||
// IsPoolFrozen returns whether a specific pool is permanently frozen.
|
||||
func (ps *PauseState) IsPoolFrozen(poolId string) bool {
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
_, ok := ps.poolFrozen[poolId]
|
||||
return ok
|
||||
}
|
||||
|
||||
// IsOperationBlocked returns an error if operations on the given pool are blocked.
|
||||
func (ps *PauseState) IsOperationBlocked(poolId string) error {
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
if ps.dexPaused {
|
||||
return fmt.Errorf("DEX: paused")
|
||||
}
|
||||
if _, ok := ps.poolFrozen[poolId]; ok {
|
||||
return fmt.Errorf("pool: frozen")
|
||||
}
|
||||
if _, ok := ps.poolPaused[poolId]; ok {
|
||||
return fmt.Errorf("pool: paused")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PauseDEX pauses the entire DEX.
|
||||
func (ps *PauseState) PauseDEX(admin string) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if ps.dexPaused {
|
||||
return fmt.Errorf("DEX already paused")
|
||||
}
|
||||
ps.dexPaused = true
|
||||
ps.dexPausedAt = time.Now()
|
||||
ps.dexPausedBy = admin
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResumeDEX resumes the entire DEX.
|
||||
func (ps *PauseState) ResumeDEX(admin string) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if !ps.dexPaused {
|
||||
return fmt.Errorf("DEX not paused")
|
||||
}
|
||||
ps.dexPaused = false
|
||||
ps.dexPausedAt = time.Time{}
|
||||
ps.dexPausedBy = ""
|
||||
return nil
|
||||
}
|
||||
|
||||
// PausePool pauses a specific pool.
|
||||
func (ps *PauseState) PausePool(poolId, admin string) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if _, ok := ps.poolFrozen[poolId]; ok {
|
||||
return fmt.Errorf("pool is frozen, cannot pause")
|
||||
}
|
||||
if _, ok := ps.poolPaused[poolId]; ok {
|
||||
return fmt.Errorf("pool already paused")
|
||||
}
|
||||
ps.poolPaused[poolId] = &PoolPauseInfo{
|
||||
PoolID: poolId, PausedAt: time.Now(), PausedBy: admin,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResumePool resumes a specific pool.
|
||||
func (ps *PauseState) ResumePool(poolId, admin string) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if _, ok := ps.poolFrozen[poolId]; ok {
|
||||
return fmt.Errorf("pool is frozen, cannot resume")
|
||||
}
|
||||
if _, ok := ps.poolPaused[poolId]; !ok {
|
||||
return fmt.Errorf("pool not paused")
|
||||
}
|
||||
delete(ps.poolPaused, poolId)
|
||||
return nil
|
||||
}
|
||||
|
||||
// FreezePool permanently freezes a pool. IRREVERSIBLE without governance.
|
||||
func (ps *PauseState) FreezePool(poolId, admin, reason string) error {
|
||||
ps.mu.Lock()
|
||||
defer ps.mu.Unlock()
|
||||
if _, ok := ps.poolFrozen[poolId]; ok {
|
||||
return fmt.Errorf("pool already frozen")
|
||||
}
|
||||
delete(ps.poolPaused, poolId)
|
||||
ps.poolFrozen[poolId] = &PoolFreezeInfo{
|
||||
PoolID: poolId, FrozenAt: time.Now(), FrozenBy: admin, Reason: reason,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Status returns the current DEX status.
|
||||
func (ps *PauseState) Status() DEXStatusResponse {
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
resp := DEXStatusResponse{
|
||||
DEXPaused: ps.dexPaused, PausedPoolCount: len(ps.poolPaused), FrozenPoolCount: len(ps.poolFrozen),
|
||||
}
|
||||
if ps.dexPaused {
|
||||
t := ps.dexPausedAt
|
||||
resp.DEXPausedAt = &t
|
||||
resp.DEXPausedBy = ps.dexPausedBy
|
||||
}
|
||||
for _, info := range ps.poolPaused {
|
||||
resp.PausedPools = append(resp.PausedPools, info)
|
||||
}
|
||||
for _, info := range ps.poolFrozen {
|
||||
resp.FrozenPools = append(resp.FrozenPools, info)
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
// PoolStatus returns the status for a specific pool.
|
||||
func (ps *PauseState) PoolStatus(poolId string) PoolStatusResponse {
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
resp := PoolStatusResponse{PoolID: poolId, DEXPaused: ps.dexPaused}
|
||||
if info, ok := ps.poolPaused[poolId]; ok {
|
||||
resp.Paused = true
|
||||
resp.PauseInfo = info
|
||||
}
|
||||
if info, ok := ps.poolFrozen[poolId]; ok {
|
||||
resp.Frozen = true
|
||||
resp.FreezeInfo = info
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
// RegisterAdminRoutes registers all admin routes on the given mux.
|
||||
func RegisterAdminRoutes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/v1/admin/pause", adminAuthWrap(handleAdminPause))
|
||||
mux.HandleFunc("/v1/admin/resume", adminAuthWrap(handleAdminResume))
|
||||
mux.HandleFunc("/v1/admin/pool/pause", adminAuthWrap(handleAdminPoolPause))
|
||||
mux.HandleFunc("/v1/admin/pool/resume", adminAuthWrap(handleAdminPoolResume))
|
||||
mux.HandleFunc("/v1/admin/pool/freeze", adminAuthWrap(handleAdminPoolFreeze))
|
||||
mux.HandleFunc("/v1/admin/pool/status", handleAdminPoolStatus)
|
||||
mux.HandleFunc("/v1/admin/status", handleAdminStatus)
|
||||
}
|
||||
|
||||
func adminAuthWrap(next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
role := r.Header.Get("X-User-Role")
|
||||
if role != AdminRole {
|
||||
writeAdminError(w, http.StatusForbidden, fmt.Errorf("admin role required"))
|
||||
return
|
||||
}
|
||||
next(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
func adminCallerID(r *http.Request) string {
|
||||
if id := r.Header.Get("X-User-ID"); id != "" {
|
||||
return id
|
||||
}
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
func writeAdminJSON(w http.ResponseWriter, status int, data interface{}) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{"success": status < 400, "data": data})
|
||||
}
|
||||
|
||||
func writeAdminError(w http.ResponseWriter, status int, err error) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{"success": false, "error": err.Error()})
|
||||
}
|
||||
|
||||
type poolActionRequest struct {
|
||||
PoolID string `json:"poolId"`
|
||||
Reason string `json:"reason,omitempty"`
|
||||
}
|
||||
|
||||
func handleAdminPause(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
admin := adminCallerID(r)
|
||||
if err := pauseStateGlobal.PauseDEX(admin); err != nil {
|
||||
writeAdminError(w, http.StatusConflict, err)
|
||||
return
|
||||
}
|
||||
log.Printf("ADMIN: DEX PAUSED by %s", admin)
|
||||
writeAdminJSON(w, http.StatusOK, map[string]string{"status": "paused", "action": "dex_pause", "admin": admin})
|
||||
}
|
||||
|
||||
func handleAdminResume(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
admin := adminCallerID(r)
|
||||
if err := pauseStateGlobal.ResumeDEX(admin); err != nil {
|
||||
writeAdminError(w, http.StatusConflict, err)
|
||||
return
|
||||
}
|
||||
log.Printf("ADMIN: DEX RESUMED by %s", admin)
|
||||
writeAdminJSON(w, http.StatusOK, map[string]string{"status": "resumed", "action": "dex_resume", "admin": admin})
|
||||
}
|
||||
|
||||
func handleAdminPoolPause(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
var req poolActionRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("invalid request body: %w", err))
|
||||
return
|
||||
}
|
||||
if req.PoolID == "" {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("poolId is required"))
|
||||
return
|
||||
}
|
||||
admin := adminCallerID(r)
|
||||
if err := pauseStateGlobal.PausePool(req.PoolID, admin); err != nil {
|
||||
writeAdminError(w, http.StatusConflict, err)
|
||||
return
|
||||
}
|
||||
log.Printf("ADMIN: POOL PAUSED pool=%s by=%s", req.PoolID, admin)
|
||||
writeAdminJSON(w, http.StatusOK, map[string]string{"status": "paused", "action": "pool_pause", "poolId": req.PoolID, "admin": admin})
|
||||
}
|
||||
|
||||
func handleAdminPoolResume(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
var req poolActionRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("invalid request body: %w", err))
|
||||
return
|
||||
}
|
||||
if req.PoolID == "" {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("poolId is required"))
|
||||
return
|
||||
}
|
||||
admin := adminCallerID(r)
|
||||
if err := pauseStateGlobal.ResumePool(req.PoolID, admin); err != nil {
|
||||
writeAdminError(w, http.StatusConflict, err)
|
||||
return
|
||||
}
|
||||
log.Printf("ADMIN: POOL RESUMED pool=%s by=%s", req.PoolID, admin)
|
||||
writeAdminJSON(w, http.StatusOK, map[string]string{"status": "resumed", "action": "pool_resume", "poolId": req.PoolID, "admin": admin})
|
||||
}
|
||||
|
||||
func handleAdminPoolFreeze(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
var req poolActionRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("invalid request body: %w", err))
|
||||
return
|
||||
}
|
||||
if req.PoolID == "" {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("poolId is required"))
|
||||
return
|
||||
}
|
||||
if req.Reason == "" {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("reason is required for freeze (regulatory audit trail)"))
|
||||
return
|
||||
}
|
||||
admin := adminCallerID(r)
|
||||
if err := pauseStateGlobal.FreezePool(req.PoolID, admin, req.Reason); err != nil {
|
||||
writeAdminError(w, http.StatusConflict, err)
|
||||
return
|
||||
}
|
||||
log.Printf("ADMIN: POOL FROZEN (IRREVERSIBLE) pool=%s by=%s reason=%q", req.PoolID, admin, req.Reason)
|
||||
writeAdminJSON(w, http.StatusOK, map[string]string{
|
||||
"status": "frozen", "action": "pool_freeze", "poolId": req.PoolID,
|
||||
"admin": admin, "reason": req.Reason, "warning": "This action is IRREVERSIBLE without governance",
|
||||
})
|
||||
}
|
||||
|
||||
func handleAdminPoolStatus(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
poolId := r.URL.Query().Get("poolId")
|
||||
if poolId == "" {
|
||||
path := strings.TrimPrefix(r.URL.Path, "/v1/admin/pool/status")
|
||||
path = strings.TrimPrefix(path, "/")
|
||||
if path != "" {
|
||||
poolId = path
|
||||
}
|
||||
}
|
||||
if poolId == "" {
|
||||
writeAdminError(w, http.StatusBadRequest, fmt.Errorf("poolId is required"))
|
||||
return
|
||||
}
|
||||
writeAdminJSON(w, http.StatusOK, pauseStateGlobal.PoolStatus(poolId))
|
||||
}
|
||||
|
||||
func handleAdminStatus(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
writeAdminError(w, http.StatusMethodNotAllowed, fmt.Errorf("method not allowed"))
|
||||
return
|
||||
}
|
||||
writeAdminJSON(w, http.StatusOK, pauseStateGlobal.Status())
|
||||
}
|
||||
+68
-41
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"math/big"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -172,9 +173,6 @@ func (s *Server) registerRoutes() {
|
||||
|
||||
// Multihop routing routes
|
||||
s.registerMultihopRoutes()
|
||||
|
||||
// Admin pause/freeze routes
|
||||
RegisterAdminRoutes(s.mux)
|
||||
}
|
||||
|
||||
// Response helpers
|
||||
@@ -223,7 +221,15 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := s.requestContext(r)
|
||||
checks := s.router.HealthCheck(ctx)
|
||||
|
||||
healthy := false
|
||||
// The venues count, and they count first. A deployment with no hosted
|
||||
// provider is the ordinary one — the hosted trade API answers ACCESS_DENIED
|
||||
// without a contract, so the pools are read directly — and asking only the
|
||||
// provider registry answers 503 about a gateway that is quoting seven
|
||||
// chains. Anything probing this to decide whether the process can serve
|
||||
// then holds every replica out of its own Service, permanently, while every
|
||||
// quote it would have served works.
|
||||
chains := s.chains.Chains()
|
||||
healthy := len(chains) > 0
|
||||
for _, check := range checks {
|
||||
if check.Healthy {
|
||||
healthy = true
|
||||
@@ -238,6 +244,7 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
s.writeJSON(w, status, map[string]interface{}{
|
||||
"status": map[bool]string{true: "healthy", false: "unhealthy"}[healthy],
|
||||
"chains": chains,
|
||||
"providers": checks,
|
||||
})
|
||||
}
|
||||
@@ -299,7 +306,7 @@ func (s *Server) handleQuote(w http.ResponseWriter, r *http.Request) {
|
||||
// deployment missing a provider — it is a pair no venue here holds.
|
||||
// Saying "no providers available" of our own chain sends a reader
|
||||
// looking for a misconfiguration that is not there.
|
||||
if errors.Is(err, ErrNoProvidersAvailable) && (s.chains.For(asked.ChainID) != nil || (s.venues != nil && len(s.venues.Venues()) > 0)) {
|
||||
if errors.Is(err, ErrNoProvidersAvailable) && s.settles(asked.ChainID) {
|
||||
s.writeError(w, http.StatusNotFound, fmt.Errorf("no venue here holds this pair"))
|
||||
return
|
||||
}
|
||||
@@ -310,14 +317,17 @@ func (s *Server) handleQuote(w http.ResponseWriter, r *http.Request) {
|
||||
s.writeJSON(w, http.StatusOK, quote)
|
||||
}
|
||||
|
||||
// venueQuote asks the venues this deployment settles on, and returns nil when
|
||||
// none of them holds the pair — which is a question for the providers and not
|
||||
// an answer of its own.
|
||||
func (s *Server) venueQuote(ctx context.Context, req QuoteRequest) *SwapQuote {
|
||||
if kept := s.quotes.get(req); kept != nil {
|
||||
return kept
|
||||
}
|
||||
// settles reports whether this deployment reads a chain from its own pools.
|
||||
// Where it does, "no providers available" is never the true answer — it names
|
||||
// an upstream registry the caller did not ask about.
|
||||
func (s *Server) settles(chain ChainID) bool {
|
||||
return s.chains.For(chain) != nil || (s.venues != nil && len(s.venues.Venues()) > 0)
|
||||
}
|
||||
|
||||
// venueQuotes asks the venues this deployment settles on and returns what each
|
||||
// one holds, best first. Empty when none of them holds the pair — which is a
|
||||
// question for the providers and not an answer of its own.
|
||||
func (s *Server) venueQuotes(ctx context.Context, req QuoteRequest) []SwapQuote {
|
||||
venues := s.chains.For(req.ChainID)
|
||||
if venues == nil {
|
||||
venues = s.venues
|
||||
@@ -330,45 +340,47 @@ func (s *Server) venueQuote(ctx context.Context, req QuoteRequest) *SwapQuote {
|
||||
// A V3 pool on Ethereum came back marked V4_NATIVE, which names our own
|
||||
// precompile on somebody else's chain — the one thing a route must never
|
||||
// misreport, since it is what a reader checks to see where their order
|
||||
// goes. The winning venue names itself.
|
||||
quotes, _ := venues.QueryAllVenues(ctx, VenueQuoteRequest{
|
||||
// goes. Each venue names itself.
|
||||
answers, _ := venues.QueryAllVenues(ctx, VenueQuoteRequest{
|
||||
TokenIn: req.TokenIn.Address,
|
||||
TokenOut: req.TokenOut.Address,
|
||||
Amount: req.Amount.String(),
|
||||
Type: map[bool]string{true: "EXACT_INPUT", false: "EXACT_OUTPUT"}[req.IsExactIn],
|
||||
})
|
||||
|
||||
best := -1
|
||||
for i, q := range quotes {
|
||||
if !q.Executable || q.AmountOut == "" || q.AmountOut == "0" {
|
||||
held := make([]SwapQuote, 0, len(answers))
|
||||
for _, a := range answers {
|
||||
out, ok := new(big.Int).SetString(a.AmountOut, 10)
|
||||
if !a.Executable || !ok || out.Sign() <= 0 {
|
||||
continue
|
||||
}
|
||||
if best < 0 {
|
||||
best = i
|
||||
continue
|
||||
}
|
||||
a, aok := new(big.Int).SetString(q.AmountOut, 10)
|
||||
b, bok := new(big.Int).SetString(quotes[best].AmountOut, 10)
|
||||
if aok && bok && a.Cmp(b) > 0 {
|
||||
best = i
|
||||
}
|
||||
gas, _ := new(big.Int).SetString(a.GasEstimate, 10)
|
||||
held = append(held, SwapQuote{
|
||||
TokenIn: TokenAmount{Token: req.TokenIn, Amount: req.Amount},
|
||||
TokenOut: TokenAmount{Token: req.TokenOut, Amount: out},
|
||||
Route: []PoolHop{{PoolType: a.Venue, TokenIn: req.TokenIn, TokenOut: req.TokenOut}},
|
||||
GasEstimate: gas,
|
||||
ProviderName: a.Venue,
|
||||
})
|
||||
}
|
||||
if best < 0 {
|
||||
sort.SliceStable(held, func(i, j int) bool {
|
||||
return held[i].TokenOut.Amount.Cmp(held[j].TokenOut.Amount) > 0
|
||||
})
|
||||
return held
|
||||
}
|
||||
|
||||
// venueQuote is the best of those, kept for the moment a pool holds still.
|
||||
func (s *Server) venueQuote(ctx context.Context, req QuoteRequest) *SwapQuote {
|
||||
if kept := s.quotes.get(req); kept != nil {
|
||||
return kept
|
||||
}
|
||||
held := s.venueQuotes(ctx, req)
|
||||
if len(held) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
won := quotes[best]
|
||||
out, _ := new(big.Int).SetString(won.AmountOut, 10)
|
||||
gas, _ := new(big.Int).SetString(won.GasEstimate, 10)
|
||||
quote := &SwapQuote{
|
||||
TokenIn: TokenAmount{Token: req.TokenIn, Amount: req.Amount},
|
||||
TokenOut: TokenAmount{Token: req.TokenOut, Amount: out},
|
||||
Route: []PoolHop{{PoolType: won.Venue, TokenIn: req.TokenIn, TokenOut: req.TokenOut}},
|
||||
GasEstimate: gas,
|
||||
ProviderName: won.Venue,
|
||||
}
|
||||
s.quotes.put(req, quote)
|
||||
return quote
|
||||
best := held[0]
|
||||
s.quotes.put(req, &best)
|
||||
return &best
|
||||
}
|
||||
|
||||
func (s *Server) handleQuotes(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -384,8 +396,23 @@ func (s *Server) handleQuotes(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
ctx := s.requestContext(r)
|
||||
quotes, err := s.router.GetAllQuotes(ctx, s.convertQuoteRequest(req))
|
||||
asked := s.convertQuoteRequest(req)
|
||||
|
||||
// Every answer rather than the best one, and from the same source: this is
|
||||
// /v1/quote's question, asked without the last step. Reading it from the
|
||||
// provider registry alone made a chain we settle ourselves report "no
|
||||
// providers available" while /v1/quote priced the same pair from its pools.
|
||||
if held := s.venueQuotes(ctx, asked); len(held) > 0 {
|
||||
s.writeJSON(w, http.StatusOK, held)
|
||||
return
|
||||
}
|
||||
|
||||
quotes, err := s.router.GetAllQuotes(ctx, asked)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrNoProvidersAvailable) && s.settles(asked.ChainID) {
|
||||
s.writeError(w, http.StatusNotFound, fmt.Errorf("no venue here holds this pair"))
|
||||
return
|
||||
}
|
||||
s.writeError(w, http.StatusInternalServerError, err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,167 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"math/big"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The deployment this exercises is the ordinary one: venues over pools, and no
|
||||
// hosted provider at all. The hosted trade API answers ACCESS_DENIED without a
|
||||
// contract, so a provider that always refuses is worse than none — which means
|
||||
// every path a client uses has to be correct with an EMPTY registry, and three
|
||||
// of them were not.
|
||||
func newPoolOnlyServer() *Server {
|
||||
unit := new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil)
|
||||
|
||||
pools := NewV4Venue("")
|
||||
pools.AddPool(testLUSD, testWETH,
|
||||
"0x0000000000000000000000000000000000000000000000000000000000000002",
|
||||
30,
|
||||
new(big.Int).Mul(big.NewInt(5_000_000), unit),
|
||||
new(big.Int).Mul(big.NewInt(1666), unit))
|
||||
|
||||
return NewServer(
|
||||
NewRouter(NewRegistry(), true),
|
||||
DefaultServerConfig(),
|
||||
WithChainVenues(&ChainRouters{byChain: map[ChainID]*VenueRouter{
|
||||
ChainIDLux: NewVenueRouter(pools),
|
||||
}}),
|
||||
)
|
||||
}
|
||||
|
||||
func ask(s *Server, method, path string, body interface{}) *httptest.ResponseRecorder {
|
||||
var r *http.Request
|
||||
if body == nil {
|
||||
r = httptest.NewRequest(method, path, nil)
|
||||
} else {
|
||||
b, _ := json.Marshal(body)
|
||||
r = httptest.NewRequest(method, path, bytes.NewReader(b))
|
||||
r.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
w := httptest.NewRecorder()
|
||||
s.mux.ServeHTTP(w, r)
|
||||
return w
|
||||
}
|
||||
|
||||
// /healthz reported the hosted provider registry and nothing else, so the
|
||||
// ordinary deployment answered 503 forever. Anything probing it to decide
|
||||
// whether a replica can serve then holds every replica out of its own Service
|
||||
// while every quote it would have served works.
|
||||
func TestHealthIsTheVenues(t *testing.T) {
|
||||
w := ask(newPoolOnlyServer(), http.MethodGet, "/healthz", nil)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("healthz with venues and no provider: got %d, want 200: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
var got struct {
|
||||
Data struct {
|
||||
Status string `json:"status"`
|
||||
Chains []ChainID `json:"chains"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
|
||||
t.Fatalf("body: %v", err)
|
||||
}
|
||||
if got.Data.Status != "healthy" {
|
||||
t.Errorf("status = %q, want healthy", got.Data.Status)
|
||||
}
|
||||
if len(got.Data.Chains) != 1 || got.Data.Chains[0] != ChainIDLux {
|
||||
t.Errorf("chains = %v, want [%d]", got.Data.Chains, ChainIDLux)
|
||||
}
|
||||
}
|
||||
|
||||
// /v1/quotes read the provider registry alone, so a chain we settle ourselves
|
||||
// answered "no providers available" — an error about an upstream account —
|
||||
// while /v1/quote priced the very same pair from the very same pools.
|
||||
func TestQuotesReadThePools(t *testing.T) {
|
||||
s := newPoolOnlyServer()
|
||||
asked := quoteRequest{
|
||||
ChainID: uint64(ChainIDLux),
|
||||
TokenIn: testWETH,
|
||||
TokenOut: testLUSD,
|
||||
Amount: "1000000000000000000",
|
||||
IsExactIn: true,
|
||||
}
|
||||
|
||||
one := ask(s, http.MethodPost, "/v1/quote", asked)
|
||||
if one.Code != http.StatusOK {
|
||||
t.Fatalf("/v1/quote: got %d, want 200: %s", one.Code, one.Body.String())
|
||||
}
|
||||
all := ask(s, http.MethodPost, "/v1/quotes", asked)
|
||||
if all.Code != http.StatusOK {
|
||||
t.Fatalf("/v1/quotes: got %d, want 200: %s", all.Code, all.Body.String())
|
||||
}
|
||||
|
||||
var best struct {
|
||||
Data SwapQuote `json:"data"`
|
||||
}
|
||||
var every struct {
|
||||
Data []SwapQuote `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(one.Body.Bytes(), &best); err != nil {
|
||||
t.Fatalf("quote body: %v", err)
|
||||
}
|
||||
if err := json.Unmarshal(all.Body.Bytes(), &every); err != nil {
|
||||
t.Fatalf("quotes body: %v", err)
|
||||
}
|
||||
if len(every.Data) == 0 {
|
||||
t.Fatal("/v1/quotes returned nothing where /v1/quote returned a price")
|
||||
}
|
||||
// The best of every is the one, or the two disagree about one pool.
|
||||
if every.Data[0].TokenOut.Amount.Cmp(best.Data.TokenOut.Amount) != 0 {
|
||||
t.Errorf("head of /v1/quotes = %s, /v1/quote = %s",
|
||||
every.Data[0].TokenOut.Amount, best.Data.TokenOut.Amount)
|
||||
}
|
||||
if every.Data[0].ProviderName == "" {
|
||||
t.Error("a quote that does not name the venue that answered it")
|
||||
}
|
||||
}
|
||||
|
||||
// A pair no pool holds is 404 and says so about the venues, never 500 about a
|
||||
// registry the caller never asked for.
|
||||
func TestUnheldPairIsNotAnUpstreamError(t *testing.T) {
|
||||
w := ask(newPoolOnlyServer(), http.MethodPost, "/v1/quotes", quoteRequest{
|
||||
ChainID: uint64(ChainIDLux),
|
||||
TokenIn: testWETH,
|
||||
TokenOut: testOther,
|
||||
Amount: "1000000000000000000",
|
||||
IsExactIn: true,
|
||||
})
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Fatalf("got %d, want 404: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if bytes.Contains(w.Body.Bytes(), []byte("no providers available")) {
|
||||
t.Errorf("a chain we settle ourselves reported an upstream registry: %s", w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// The pause/freeze surface authorised on `X-User-Role: admin` — a header the
|
||||
// caller writes — and drove state nothing read. It is gone; nothing may bring
|
||||
// back a route that answers a header.
|
||||
func TestNoSurfaceAnswersAHeaderForARole(t *testing.T) {
|
||||
s := newPoolOnlyServer()
|
||||
for _, path := range []string{
|
||||
"/v1/admin/status",
|
||||
"/v1/admin/pause",
|
||||
"/v1/admin/resume",
|
||||
"/v1/admin/pool/pause",
|
||||
"/v1/admin/pool/resume",
|
||||
"/v1/admin/pool/freeze",
|
||||
"/v1/admin/pool/status",
|
||||
} {
|
||||
for _, method := range []string{http.MethodGet, http.MethodPost} {
|
||||
r := httptest.NewRequest(method, path, bytes.NewReader([]byte(`{"poolId":"0x1","reason":"x"}`)))
|
||||
r.Header.Set("X-User-Role", "admin")
|
||||
r.Header.Set("X-User-ID", "anyone")
|
||||
w := httptest.NewRecorder()
|
||||
s.mux.ServeHTTP(w, r)
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("%s %s answered %d: %s", method, path, w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user