From 953240565f633b11ac410148e7364fb31d21c53b Mon Sep 17 00:00:00 2001 From: tinkle-community Date: Thu, 11 Jun 2026 21:45:31 +0800 Subject: [PATCH] fix(trader): stop order-sync goroutine leak and rate-limit hammering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every StartOrderSync spawned a ticker goroutine that ran forever — it survived trader stop AND deletion, so each quick-created trader left a permanent 30s Hyperliquid poll behind. Stacked leaks turned into an ~8s effective hammer that tripped Hyperliquid's 429 rate limit, which then broke the symbol board, trader creation, and order sync itself. - new trader/syncloop package: shared stoppable sync loop with exponential failure backoff (30s base, 5min cap) - all 9 exchanges' StartOrderSync now take the trader's stop channel and stop when the trader stops (close broadcast from AutoTrader.Stop) - provider/hyperliquid: GetPerpDexCoins now serves a 5min TTL cache and falls back to the stale board when the upstream returns 429, so the symbol panel keeps working through rate limiting --- provider/hyperliquid/coins.go | 54 ++++++- provider/hyperliquid/perp_dex_cache_test.go | 113 ++++++++++++++ trader/aster/trader_sync.go | 15 +- trader/auto_trader.go | 18 +-- trader/binance/order_sync.go | 15 +- trader/bitget/order_sync.go | 16 +- trader/bybit/order_sync.go | 15 +- trader/gate/order_sync.go | 15 +- trader/hyperliquid/order_sync.go | 163 ++++++++++---------- trader/kucoin/order_sync.go | 15 +- trader/lighter/order_sync.go | 22 ++- trader/okx/order_sync.go | 15 +- trader/syncloop/syncloop.go | 48 ++++++ trader/syncloop/syncloop_test.go | 78 ++++++++++ 14 files changed, 424 insertions(+), 178 deletions(-) create mode 100644 provider/hyperliquid/perp_dex_cache_test.go create mode 100644 trader/syncloop/syncloop.go create mode 100644 trader/syncloop/syncloop_test.go diff --git a/provider/hyperliquid/coins.go b/provider/hyperliquid/coins.go index 6b7bf68f..ea06a725 100644 --- a/provider/hyperliquid/coins.go +++ b/provider/hyperliquid/coins.go @@ -164,9 +164,59 @@ func fetchPerpDexCoins(ctx context.Context, client *http.Client, dex string) ([] return coins, nil } -// GetPerpDexCoins fetches current tradable USDC perp assets for a given Hyperliquid dex. +// perpDexCacheTTL bounds how often the perp-dex symbol board is re-fetched. +// The tradable symbol list changes rarely; prices/volume on the board are +// display hints, so short staleness is far better than hammering the +// Hyperliquid API (which rate-limits with 429) on every panel render. +const perpDexCacheTTL = 5 * time.Minute + +type perpDexCacheEntry struct { + coins []CoinInfo + fetchedAt time.Time +} + +type perpDexCacheStore struct { + mu sync.Mutex + entries map[string]perpDexCacheEntry +} + +var perpDexCoinCache = &perpDexCacheStore{entries: map[string]perpDexCacheEntry{}} + +// fetchPerpDexCoinsFn is swappable in tests. +var fetchPerpDexCoinsFn = fetchPerpDexCoins + +// GetPerpDexCoins returns current tradable USDC perp assets for a given +// Hyperliquid dex, served from a TTL cache. When the upstream fetch fails +// (e.g. HTTP 429 rate limiting) and stale data exists, the stale board is +// served instead of an error so the UI keeps working. func GetPerpDexCoins(ctx context.Context, dex string) ([]CoinInfo, error) { - return fetchPerpDexCoins(ctx, &http.Client{Timeout: 30 * time.Second}, dex) + perpDexCoinCache.mu.Lock() + defer perpDexCoinCache.mu.Unlock() + + entry, hasCache := perpDexCoinCache.entries[dex] + if hasCache && time.Since(entry.fetchedAt) < perpDexCacheTTL { + return copyCoins(entry.coins), nil + } + + coins, err := fetchPerpDexCoinsFn(ctx, &http.Client{Timeout: 30 * time.Second}, dex) + if err != nil { + if hasCache { + logger.Infof("⚠️ Hyperliquid perp-dex fetch failed (%v); serving cached board for dex %q from %s", + err, dex, entry.fetchedAt.Format(time.RFC3339)) + return copyCoins(entry.coins), nil + } + return nil, err + } + + perpDexCoinCache.entries[dex] = perpDexCacheEntry{coins: coins, fetchedAt: time.Now()} + return copyCoins(coins), nil +} + +// copyCoins returns a defensive copy so callers cannot mutate the cache. +func copyCoins(coins []CoinInfo) []CoinInfo { + out := make([]CoinInfo, len(coins)) + copy(out, coins) + return out } // fetchCoins fetches all default Hyperliquid crypto coins and sorts by volume diff --git a/provider/hyperliquid/perp_dex_cache_test.go b/provider/hyperliquid/perp_dex_cache_test.go new file mode 100644 index 00000000..00eaeffa --- /dev/null +++ b/provider/hyperliquid/perp_dex_cache_test.go @@ -0,0 +1,113 @@ +package hyperliquid + +import ( + "context" + "errors" + "net/http" + "testing" + "time" +) + +// withStubbedPerpDexFetch swaps the live fetch function and resets the cache, +// restoring both when the test finishes. +func withStubbedPerpDexFetch(t *testing.T, fn func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error)) { + t.Helper() + original := fetchPerpDexCoinsFn + fetchPerpDexCoinsFn = fn + perpDexCoinCache.mu.Lock() + perpDexCoinCache.entries = map[string]perpDexCacheEntry{} + perpDexCoinCache.mu.Unlock() + t.Cleanup(func() { + fetchPerpDexCoinsFn = original + perpDexCoinCache.mu.Lock() + perpDexCoinCache.entries = map[string]perpDexCacheEntry{} + perpDexCoinCache.mu.Unlock() + }) +} + +func TestGetPerpDexCoinsCachesWithinTTL(t *testing.T) { + calls := 0 + withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) { + calls++ + return []CoinInfo{{Symbol: "xyz:TSLA", MarkPrice: 400}}, nil + }) + + first, err := GetPerpDexCoins(context.Background(), "xyz") + if err != nil { + t.Fatalf("first call: %v", err) + } + second, err := GetPerpDexCoins(context.Background(), "xyz") + if err != nil { + t.Fatalf("second call: %v", err) + } + if calls != 1 { + t.Fatalf("fetch calls = %d, want 1 (second call must hit cache)", calls) + } + if len(first) != 1 || len(second) != 1 || second[0].Symbol != "xyz:TSLA" { + t.Fatalf("unexpected results: first=%v second=%v", first, second) + } +} + +func TestGetPerpDexCoinsServesStaleOnUpstreamError(t *testing.T) { + calls := 0 + withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) { + calls++ + if calls == 1 { + return []CoinInfo{{Symbol: "xyz:NVDA", MarkPrice: 1000}}, nil + } + return nil, errors.New("API returned status 429") + }) + + if _, err := GetPerpDexCoins(context.Background(), "xyz"); err != nil { + t.Fatalf("first call: %v", err) + } + + // Expire the cache so the next call must attempt a refresh. + perpDexCoinCache.mu.Lock() + entry := perpDexCoinCache.entries["xyz"] + entry.fetchedAt = time.Now().Add(-2 * perpDexCacheTTL) + perpDexCoinCache.entries["xyz"] = entry + perpDexCoinCache.mu.Unlock() + + coins, err := GetPerpDexCoins(context.Background(), "xyz") + if err != nil { + t.Fatalf("expected stale data instead of error, got: %v", err) + } + if len(coins) != 1 || coins[0].Symbol != "xyz:NVDA" { + t.Fatalf("expected stale NVDA entry, got %v", coins) + } + if calls != 2 { + t.Fatalf("fetch calls = %d, want 2 (refresh attempted)", calls) + } +} + +func TestGetPerpDexCoinsErrorsWithoutAnyCache(t *testing.T) { + withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) { + return nil, errors.New("API returned status 429") + }) + + if _, err := GetPerpDexCoins(context.Background(), "xyz"); err == nil { + t.Fatal("expected error when upstream fails and no cache exists") + } +} + +func TestGetPerpDexCoinsCachesPerDex(t *testing.T) { + withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) { + if dex == "xyz" { + return []CoinInfo{{Symbol: "xyz:AAPL"}}, nil + } + return []CoinInfo{{Symbol: "BTC"}}, nil + }) + + xyz, err := GetPerpDexCoins(context.Background(), "xyz") + if err != nil { + t.Fatalf("xyz: %v", err) + } + def, err := GetPerpDexCoins(context.Background(), "") + if err != nil { + t.Fatalf("default dex: %v", err) + } + if xyz[0].Symbol != "xyz:AAPL" || def[0].Symbol != "BTC" { + t.Fatalf("cache keys collided: xyz=%v default=%v", xyz, def) + } +} diff --git a/trader/aster/trader_sync.go b/trader/aster/trader_sync.go index 2a1c0d51..bee0ea9c 100644 --- a/trader/aster/trader_sync.go +++ b/trader/aster/trader_sync.go @@ -5,6 +5,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strings" "time" @@ -180,14 +181,8 @@ func deriveAsterOrderAction(side, positionSide string, realizedPnL float64) stri } // StartOrderSync starts background order sync task for Aster -func (t *AsterTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromAster(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ Aster order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 Aster order sync started (interval: %v)", interval) +func (t *AsterTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "Aster", func() error { + return t.SyncOrdersFromAster(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/auto_trader.go b/trader/auto_trader.go index 1faaafd4..2667849f 100644 --- a/trader/auto_trader.go +++ b/trader/auto_trader.go @@ -421,7 +421,7 @@ func (at *AutoTrader) Run() error { // Start Lighter order sync if using Lighter exchange if at.exchange == "lighter" { if lighterTrader, ok := at.trader.(*lighter.LighterTraderV2); ok && at.store != nil { - lighterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + lighterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Lighter order+position sync enabled (every 30s)") } } @@ -429,7 +429,7 @@ func (at *AutoTrader) Run() error { // Start Hyperliquid order sync if using Hyperliquid exchange if at.exchange == "hyperliquid" { if hyperliquidTrader, ok := at.trader.(*hyperliquid.HyperliquidTrader); ok && at.store != nil { - hyperliquidTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + hyperliquidTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Hyperliquid order+position sync enabled (every 30s)") } } @@ -437,7 +437,7 @@ func (at *AutoTrader) Run() error { // Start Bybit order sync if using Bybit exchange if at.exchange == "bybit" { if bybitTrader, ok := at.trader.(*bybit.BybitTrader); ok && at.store != nil { - bybitTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + bybitTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Bybit order+position sync enabled (every 30s)") } } @@ -445,7 +445,7 @@ func (at *AutoTrader) Run() error { // Start OKX order sync if using OKX exchange if at.exchange == "okx" { if okxTrader, ok := at.trader.(*okx.OKXTrader); ok && at.store != nil { - okxTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + okxTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 OKX order+position sync enabled (every 30s)") } } @@ -453,7 +453,7 @@ func (at *AutoTrader) Run() error { // Start Bitget order sync if using Bitget exchange if at.exchange == "bitget" { if bitgetTrader, ok := at.trader.(*bitget.BitgetTrader); ok && at.store != nil { - bitgetTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + bitgetTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Bitget order+position sync enabled (every 30s)") } } @@ -461,7 +461,7 @@ func (at *AutoTrader) Run() error { // Start Aster order sync if using Aster exchange if at.exchange == "aster" { if asterTrader, ok := at.trader.(*aster.AsterTrader); ok && at.store != nil { - asterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + asterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Aster order+position sync enabled (every 30s)") } } @@ -469,7 +469,7 @@ func (at *AutoTrader) Run() error { // Start Binance order sync if using Binance exchange if at.exchange == "binance" { if binanceTrader, ok := at.trader.(*binance.FuturesTrader); ok && at.store != nil { - binanceTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + binanceTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Binance order+position sync enabled (every 30s)") } } @@ -477,7 +477,7 @@ func (at *AutoTrader) Run() error { // Start Gate order sync if using Gate exchange if at.exchange == "gate" { if gateTrader, ok := at.trader.(*gate.GateTrader); ok && at.store != nil { - gateTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + gateTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 Gate order+position sync enabled (every 30s)") } } @@ -485,7 +485,7 @@ func (at *AutoTrader) Run() error { // Start KuCoin order sync if using KuCoin exchange if at.exchange == "kucoin" { if kucoinTrader, ok := at.trader.(*kucoin.KuCoinTrader); ok && at.store != nil { - kucoinTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second) + kucoinTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh) at.logInfof("🔄 KuCoin order+position sync enabled (every 30s)") } } diff --git a/trader/binance/order_sync.go b/trader/binance/order_sync.go index 76705395..01eba30c 100644 --- a/trader/binance/order_sync.go +++ b/trader/binance/order_sync.go @@ -5,6 +5,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "nofx/trader/types" "sort" "strings" @@ -349,7 +350,7 @@ func (t *FuturesTrader) determineOrderAction(side, positionSide string, realized } // StartOrderSync starts background order sync task for Binance -func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { +func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { // Run first sync immediately go func() { logger.Infof("🔄 Running initial Binance order sync...") @@ -359,13 +360,7 @@ func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, excha }() // Then run periodically - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromBinance(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ Binance order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 Binance order sync started (interval: %v)", interval) + syncloop.Run(stop, interval, "Binance", func() error { + return t.SyncOrdersFromBinance(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/bitget/order_sync.go b/trader/bitget/order_sync.go index e8f10789..3511f296 100644 --- a/trader/bitget/order_sync.go +++ b/trader/bitget/order_sync.go @@ -6,6 +6,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strconv" "strings" @@ -48,7 +49,6 @@ func (t *BitgetTrader) GetTrades(startTime time.Time, limit int) ([]BitgetTrade, return nil, fmt.Errorf("failed to get fill history: %w", err) } - // Bitget fill structure - supports both one-way and hedge mode type BitgetFill struct { TradeID string `json:"tradeId"` @@ -279,14 +279,8 @@ func (t *BitgetTrader) SyncOrdersFromBitget(traderID string, exchangeID string, } // StartOrderSync starts background order sync task for Bitget -func (t *BitgetTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromBitget(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ Bitget order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 Bitget order sync started (interval: %v)", interval) +func (t *BitgetTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "Bitget", func() error { + return t.SyncOrdersFromBitget(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/bybit/order_sync.go b/trader/bybit/order_sync.go index cf54afac..96aebd5d 100644 --- a/trader/bybit/order_sync.go +++ b/trader/bybit/order_sync.go @@ -11,6 +11,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strconv" "strings" @@ -298,14 +299,8 @@ func (t *BybitTrader) SyncOrdersFromBybit(traderID string, exchangeID string, ex } // StartOrderSync starts background order sync task for Bybit -func (t *BybitTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromBybit(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ Bybit order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 Bybit order sync started (interval: %v)", interval) +func (t *BybitTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "Bybit", func() error { + return t.SyncOrdersFromBybit(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/gate/order_sync.go b/trader/gate/order_sync.go index 85d49ebf..fd3244b1 100644 --- a/trader/gate/order_sync.go +++ b/trader/gate/order_sync.go @@ -5,6 +5,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strconv" "strings" @@ -291,14 +292,8 @@ func (t *GateTrader) SyncOrdersFromGate(traderID string, exchangeID string, exch } // StartOrderSync starts background order sync task for Gate -func (t *GateTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromGate(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ Gate order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 Gate order sync started (interval: %v)", interval) +func (t *GateTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "Gate", func() error { + return t.SyncOrdersFromGate(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/hyperliquid/order_sync.go b/trader/hyperliquid/order_sync.go index 752e5c3d..803f4957 100644 --- a/trader/hyperliquid/order_sync.go +++ b/trader/hyperliquid/order_sync.go @@ -5,6 +5,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strings" "time" @@ -44,87 +45,87 @@ func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeI syncedCount := 0 for _, trade := range trades { - // Check if trade already exists (use exchangeID which is UUID, not exchange type) - existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID) - if err == nil && existing != nil { - continue // Order already exists, skip - } + // Check if trade already exists (use exchangeID which is UUID, not exchange type) + existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID) + if err == nil && existing != nil { + continue // Order already exists, skip + } - // Normalize symbol - symbol := market.Normalize(trade.Symbol) + // Normalize symbol + symbol := market.Normalize(trade.Symbol) - // Use order action from trade (parsed from Hyperliquid Dir field) - // Dir field values: "Open Long", "Open Short", "Close Long", "Close Short" - orderAction := trade.OrderAction - positionSide := "LONG" - if strings.Contains(orderAction, "short") { - positionSide = "SHORT" - } + // Use order action from trade (parsed from Hyperliquid Dir field) + // Dir field values: "Open Long", "Open Short", "Close Long", "Close Short" + orderAction := trade.OrderAction + positionSide := "LONG" + if strings.Contains(orderAction, "short") { + positionSide = "SHORT" + } - // Create order record - use Unix milliseconds UTC - tradeTimeMs := trade.Time.UTC().UnixMilli() - orderRecord := &store.TraderOrder{ - TraderID: traderID, - ExchangeID: exchangeID, // UUID - ExchangeType: exchangeType, // Exchange type - ExchangeOrderID: trade.TradeID, - Symbol: symbol, - Side: trade.Side, - PositionSide: "BOTH", // Hyperliquid uses one-way position mode - Type: "MARKET", - OrderAction: orderAction, - Quantity: trade.Quantity, - Price: trade.Price, - Status: "FILLED", - FilledQuantity: trade.Quantity, - AvgFillPrice: trade.Price, - Commission: trade.Fee, - FilledAt: tradeTimeMs, - CreatedAt: tradeTimeMs, - UpdatedAt: tradeTimeMs, - } + // Create order record - use Unix milliseconds UTC + tradeTimeMs := trade.Time.UTC().UnixMilli() + orderRecord := &store.TraderOrder{ + TraderID: traderID, + ExchangeID: exchangeID, // UUID + ExchangeType: exchangeType, // Exchange type + ExchangeOrderID: trade.TradeID, + Symbol: symbol, + Side: trade.Side, + PositionSide: "BOTH", // Hyperliquid uses one-way position mode + Type: "MARKET", + OrderAction: orderAction, + Quantity: trade.Quantity, + Price: trade.Price, + Status: "FILLED", + FilledQuantity: trade.Quantity, + AvgFillPrice: trade.Price, + Commission: trade.Fee, + FilledAt: tradeTimeMs, + CreatedAt: tradeTimeMs, + UpdatedAt: tradeTimeMs, + } - // Insert order record - if err := orderStore.CreateOrder(orderRecord); err != nil { - logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err) - continue - } + // Insert order record + if err := orderStore.CreateOrder(orderRecord); err != nil { + logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err) + continue + } - // Create fill record - use Unix milliseconds UTC - fillRecord := &store.TraderFill{ - TraderID: traderID, - ExchangeID: exchangeID, // UUID - ExchangeType: exchangeType, // Exchange type - OrderID: orderRecord.ID, - ExchangeOrderID: trade.TradeID, - ExchangeTradeID: trade.TradeID, - Symbol: symbol, - Side: trade.Side, - Price: trade.Price, - Quantity: trade.Quantity, - QuoteQuantity: trade.Price * trade.Quantity, - Commission: trade.Fee, - CommissionAsset: "USDT", - RealizedPnL: trade.RealizedPnL, - IsMaker: false, // Hyperliquid GetTrades doesn't provide maker/taker info - CreatedAt: tradeTimeMs, - } + // Create fill record - use Unix milliseconds UTC + fillRecord := &store.TraderFill{ + TraderID: traderID, + ExchangeID: exchangeID, // UUID + ExchangeType: exchangeType, // Exchange type + OrderID: orderRecord.ID, + ExchangeOrderID: trade.TradeID, + ExchangeTradeID: trade.TradeID, + Symbol: symbol, + Side: trade.Side, + Price: trade.Price, + Quantity: trade.Quantity, + QuoteQuantity: trade.Price * trade.Quantity, + Commission: trade.Fee, + CommissionAsset: "USDT", + RealizedPnL: trade.RealizedPnL, + IsMaker: false, // Hyperliquid GetTrades doesn't provide maker/taker info + CreatedAt: tradeTimeMs, + } - if err := orderStore.CreateFill(fillRecord); err != nil { - logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err) - } + if err := orderStore.CreateFill(fillRecord); err != nil { + logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err) + } - // Create/update position record using PositionBuilder - if err := posBuilder.ProcessTrade( - traderID, exchangeID, exchangeType, - symbol, positionSide, orderAction, - trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL, - tradeTimeMs, trade.TradeID, - ); err != nil { - logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err) - } else { - logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, orderAction, trade.Quantity) - } + // Create/update position record using PositionBuilder + if err := posBuilder.ProcessTrade( + traderID, exchangeID, exchangeType, + symbol, positionSide, orderAction, + trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL, + tradeTimeMs, trade.TradeID, + ); err != nil { + logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err) + } else { + logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, orderAction, trade.Quantity) + } syncedCount++ logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f pnl=%.2f fee=%.6f action=%s", @@ -136,14 +137,8 @@ func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeI } // StartOrderSync starts background order sync task -func (t *HyperliquidTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromHyperliquid(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ Hyperliquid order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 Hyperliquid order sync started (interval: %v)", interval) +func (t *HyperliquidTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "Hyperliquid", func() error { + return t.SyncOrdersFromHyperliquid(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/kucoin/order_sync.go b/trader/kucoin/order_sync.go index a817075e..2f07a4d1 100644 --- a/trader/kucoin/order_sync.go +++ b/trader/kucoin/order_sync.go @@ -5,6 +5,7 @@ import ( "fmt" "nofx/logger" "nofx/store" + "nofx/trader/syncloop" "nofx/trader/types" "sort" "strings" @@ -399,14 +400,8 @@ func (t *KuCoinTrader) SyncOrdersFromKuCoin(traderID string, exchangeID string, } // StartOrderSync starts background order sync task for KuCoin -func (t *KuCoinTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromKuCoin(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ KuCoin order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 KuCoin order sync started (interval: %v)", interval) +func (t *KuCoinTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "KuCoin", func() error { + return t.SyncOrdersFromKuCoin(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/lighter/order_sync.go b/trader/lighter/order_sync.go index 24fab52d..675eeb6b 100644 --- a/trader/lighter/order_sync.go +++ b/trader/lighter/order_sync.go @@ -5,6 +5,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strings" "time" @@ -145,17 +146,14 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, exchangeID stri } // StartOrderSync starts background order sync task -func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromLighter(traderID, exchangeID, exchangeType, st); err != nil { - // Only log non-404 errors to reduce log spam - if !strings.Contains(err.Error(), "status 404") { - logger.Infof("⚠️ Order sync failed: %v", err) - } - } +func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "Lighter", func() error { + err := t.SyncOrdersFromLighter(traderID, exchangeID, exchangeType, st) + // A 404 just means the account has no fills yet — treat as a + // successful empty sync to avoid log spam and pointless backoff. + if err != nil && strings.Contains(err.Error(), "status 404") { + return nil } - }() - logger.Infof("🔄 Lighter order+position sync started (interval: %v)", interval) + return err + }) } diff --git a/trader/okx/order_sync.go b/trader/okx/order_sync.go index a8246ecb..88bc5823 100644 --- a/trader/okx/order_sync.go +++ b/trader/okx/order_sync.go @@ -6,6 +6,7 @@ import ( "nofx/logger" "nofx/market" "nofx/store" + "nofx/trader/syncloop" "sort" "strconv" "strings" @@ -272,14 +273,8 @@ func (t *OKXTrader) SyncOrdersFromOKX(traderID string, exchangeID string, exchan } // StartOrderSync starts background order sync task for OKX -func (t *OKXTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) { - ticker := time.NewTicker(interval) - go func() { - for range ticker.C { - if err := t.SyncOrdersFromOKX(traderID, exchangeID, exchangeType, st); err != nil { - logger.Infof("⚠️ OKX order sync failed: %v", err) - } - } - }() - logger.Infof("🔄 OKX order sync started (interval: %v)", interval) +func (t *OKXTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) { + syncloop.Run(stop, interval, "OKX", func() error { + return t.SyncOrdersFromOKX(traderID, exchangeID, exchangeType, st) + }) } diff --git a/trader/syncloop/syncloop.go b/trader/syncloop/syncloop.go new file mode 100644 index 00000000..b8ce6be9 --- /dev/null +++ b/trader/syncloop/syncloop.go @@ -0,0 +1,48 @@ +// Package syncloop runs background exchange order-sync loops with a shared +// lifecycle: loops stop when the owning trader stops, and consecutive +// failures back off exponentially so a rate-limiting exchange (HTTP 429) +// is not hammered at full frequency. +package syncloop + +import ( + "time" + + "nofx/logger" +) + +// maxBackoff caps the failure backoff so a recovered exchange is picked up +// within a few minutes at worst. +const maxBackoff = 5 * time.Minute + +// Run starts a background loop calling syncFn at the given interval until +// stop is closed. After each consecutive failure the wait doubles (capped at +// maxBackoff); the first success resets it to the base interval. +func Run(stop <-chan struct{}, interval time.Duration, name string, syncFn func() error) { + if interval <= 0 { + interval = 30 * time.Second + } + go func() { + wait := interval + timer := time.NewTimer(wait) + defer timer.Stop() + for { + select { + case <-stop: + logger.Infof("⏹ %s order sync stopped", name) + return + case <-timer.C: + if err := syncFn(); err != nil { + wait *= 2 + if wait > maxBackoff { + wait = maxBackoff + } + logger.Infof("⚠️ %s order sync failed: %v (backing off, next attempt in %v)", name, err, wait) + } else { + wait = interval + } + timer.Reset(wait) + } + } + }() + logger.Infof("🔄 %s order sync started (interval: %v)", name, interval) +} diff --git a/trader/syncloop/syncloop_test.go b/trader/syncloop/syncloop_test.go new file mode 100644 index 00000000..74f6e611 --- /dev/null +++ b/trader/syncloop/syncloop_test.go @@ -0,0 +1,78 @@ +package syncloop + +import ( + "errors" + "sync/atomic" + "testing" + "time" +) + +func TestRunStopsWhenStopChannelCloses(t *testing.T) { + stop := make(chan struct{}) + var calls atomic.Int64 + + Run(stop, 5*time.Millisecond, "test", func() error { + calls.Add(1) + return nil + }) + + // Let a few ticks happen, then stop. + time.Sleep(30 * time.Millisecond) + close(stop) + time.Sleep(20 * time.Millisecond) + after := calls.Load() + if after == 0 { + t.Fatal("sync function never ran") + } + + // No further calls after stop. + time.Sleep(40 * time.Millisecond) + if calls.Load() != after { + t.Fatalf("sync kept running after stop: %d -> %d", after, calls.Load()) + } +} + +func TestRunBacksOffOnConsecutiveFailures(t *testing.T) { + stop := make(chan struct{}) + defer close(stop) + var calls atomic.Int64 + + Run(stop, 10*time.Millisecond, "test", func() error { + calls.Add(1) + return errors.New("API returned status 429") + }) + + // With a 10ms base interval and exponential backoff (10, 20, 40, 80...), + // 100ms allows at most ~4 failing attempts. Without backoff there would + // be ~10. + time.Sleep(100 * time.Millisecond) + got := calls.Load() + if got == 0 { + t.Fatal("sync function never ran") + } + if got > 5 { + t.Fatalf("expected backoff to throttle failing sync, got %d calls in 100ms", got) + } +} + +func TestRunRecoversIntervalAfterSuccess(t *testing.T) { + stop := make(chan struct{}) + defer close(stop) + var calls atomic.Int64 + + // Fail twice, then succeed forever. + Run(stop, 5*time.Millisecond, "test", func() error { + n := calls.Add(1) + if n <= 2 { + return errors.New("transient") + } + return nil + }) + + // Failures at 5ms and 15ms (backoff 10ms), success at ~35ms (backoff 20ms), + // then the interval resets to 5ms — plenty of successful runs by 150ms. + time.Sleep(150 * time.Millisecond) + if got := calls.Load(); got < 8 { + t.Fatalf("expected interval to reset after success, got only %d calls", got) + } +}