Files
zeekayandHanzo Dev 45bff3f5b6 fix(dex): close price sources synchronously; build pkg/dpdk and pkg/orderbook on linux
`go test ./...` failed three ways on the runner. Two were build failures that
cannot happen on a mac, and one was a race that only loses under load.

pkg/dpdk — `undefined: syscall.SO_NOSIGPIPE`. DarwinOptimizations guarded itself
with `if runtime.GOOS != "darwin" { return }`, but SO_NOSIGPIPE is a COMPILE-time
constant that exists only on Darwin; a runtime check cannot save a reference the
linux build still has to resolve. Split into _darwin.go / _nodarwin.go. Verified
by cross-compiling both ways -- it failed for GOOS=linux before, both pass now.

pkg/orderbook — the cgo preamble uses uint64_t while including only <stdlib.h>.
macOS pulls stdint.h in transitively and glibc does not, so this built here and
never on the runner. Added the include.

pkg/price — TestXChainSourceStalePrice: "Expected price to be stale".

Not a flaky test; Close() was not a barrier. It closed the done channel and
returned WITHOUT waiting for the polling goroutine. A loop selecting on done
against a ticker picks a ready case AT RANDOM, so a poll could still be running
after Close() returned, and poll writes last[symbol] = time.Now(). The test
closes the source, backdates that timestamp by 10s, and asserts staleness -- an
in-flight poll overwrote it. Wide window on a loaded runner, near-zero on a
laptop, which is why it reproduces 0/30 here.

All EIGHT sources had it, so the fix is one shared `lifecycle` (embedded, so
`s.done` still resolves) whose stop() joins before returning and is safe to call
twice. Oracle had the same defect written as

    // Give time for goroutines to exit
    time.Sleep(100 * time.Millisecond)

which could not work either -- calcAverages blocks on a 1s ticker, so it ran up
to a full second after Stop() returned. Now on the same lifecycle, and its loops
select on done instead of polling a flag, so they leave promptly rather than
after one more tick. Nothing restarts a stopped source (checked: 0 tests), and
the other sources already could not.

    go build ./...                        OK
    GOOS=linux go build ./...             OK   (failed before)
    go vet price/dpdk/orderbook           OK
    go test ./pkg/price/ -race            ok, 3/3 runs
    go test -run StalePrice -count=30     ok

Co-authored-by: Hanzo Dev <[email protected]>
2026-08-02 03:19:43 -07:00

299 lines
6.0 KiB
Go

package price
import (
"context"
"sync"
"time"
)
// XChainSource provides prices from Lux X-Chain DEX orderbooks.
// Native exchange chain with on-chain matching and settlement.
type XChainSource struct {
rpcURL string
wsURL string
markets map[string]XMarket
prices map[string]*Data
last map[string]time.Time
books map[string]*XOrderbook
interval time.Duration
healthy bool
mu sync.RWMutex
lifecycle
polling bool
}
// XMarket defines an X-Chain trading pair.
type XMarket struct {
BaseAsset string
QuoteAsset string
MarketID string
MinSize float64
TickSize float64
}
// XOrderbook tracks X-Chain orderbook state.
type XOrderbook struct {
Bids []XLevel
Asks []XLevel
Block uint64
Time time.Time
Spread float64
}
// XLevel represents a price level.
type XLevel struct {
Price float64
Size float64
Count int
}
// NewXChainSource creates an X-Chain price source.
func NewXChainSource(rpcURL, wsURL string) *XChainSource {
return &XChainSource{
rpcURL: rpcURL,
wsURL: wsURL,
markets: xchainMarkets(),
prices: make(map[string]*Data),
last: make(map[string]time.Time),
books: make(map[string]*XOrderbook),
interval: 50 * time.Millisecond, // Faster polling - native chain
healthy: true,
lifecycle: newLifecycle(),
}
}
func xchainMarkets() map[string]XMarket {
return map[string]XMarket{
"LUX-USDC": {
BaseAsset: "LUX",
QuoteAsset: "USDC",
MarketID: "lux-usdc-v1",
MinSize: 0.01,
TickSize: 0.0001,
},
"LUX-USDT": {
BaseAsset: "LUX",
QuoteAsset: "USDT",
MarketID: "lux-usdt-v1",
MinSize: 0.01,
TickSize: 0.0001,
},
"ETH-LUX": {
BaseAsset: "ETH",
QuoteAsset: "LUX",
MarketID: "eth-lux-v1",
MinSize: 0.001,
TickSize: 0.001,
},
"BTC-LUX": {
BaseAsset: "BTC",
QuoteAsset: "LUX",
MarketID: "btc-lux-v1",
MinSize: 0.0001,
TickSize: 0.01,
},
"AVAX-LUX": {
BaseAsset: "AVAX",
QuoteAsset: "LUX",
MarketID: "avax-lux-v1",
MinSize: 0.1,
TickSize: 0.0001,
},
}
}
// Start begins polling X-Chain orderbooks.
func (s *XChainSource) Start() error {
s.mu.Lock()
if s.polling {
s.mu.Unlock()
return nil
}
s.polling = true
s.mu.Unlock()
s.run(s.loop)
return nil
}
func (s *XChainSource) loop() {
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
for {
select {
case <-s.done:
return
case <-ticker.C:
s.poll()
}
}
}
func (s *XChainSource) poll() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
var wg sync.WaitGroup
for symbol, market := range s.markets {
wg.Add(1)
go func(sym string, m XMarket) {
defer wg.Done()
s.fetch(ctx, sym, m)
}(symbol, market)
}
wg.Wait()
}
func (s *XChainSource) fetch(ctx context.Context, symbol string, market XMarket) {
// In production: query X-Chain VM for orderbook state
// For now, simulate realistic orderbook
book := s.simulate(symbol)
s.mu.Lock()
defer s.mu.Unlock()
s.books[symbol] = book
mid := (book.Bids[0].Price + book.Asks[0].Price) / 2
s.prices[symbol] = &Data{
Symbol: symbol,
Price: mid,
Bid: book.Bids[0].Price,
Ask: book.Asks[0].Price,
Volume: s.totalVolume(book),
Confidence: 0.99, // Native chain - highest confidence
Timestamp: time.Now(),
Source: "x-chain",
}
s.last[symbol] = time.Now()
s.healthy = true
}
func (s *XChainSource) simulate(symbol string) *XOrderbook {
base := map[string]float64{
"LUX-USDC": 10.0,
"LUX-USDT": 10.0,
"ETH-LUX": 300.0,
"BTC-LUX": 6000.0,
"AVAX-LUX": 3.5,
}
b, ok := base[symbol]
if !ok {
b = 1.0
}
// Slight variation
now := time.Now().UnixNano()
v := float64(now%500) / 100000.0
mid := b * (1.0 + v)
spread := mid * 0.0002 // 2 bps spread
return &XOrderbook{
Bids: []XLevel{
{Price: mid - spread/2, Size: 1000.0, Count: 5},
{Price: mid - spread, Size: 2000.0, Count: 8},
{Price: mid - spread*2, Size: 5000.0, Count: 12},
},
Asks: []XLevel{
{Price: mid + spread/2, Size: 1000.0, Count: 5},
{Price: mid + spread, Size: 2000.0, Count: 8},
{Price: mid + spread*2, Size: 5000.0, Count: 12},
},
Block: uint64(time.Now().Unix()),
Time: time.Now(),
Spread: spread,
}
}
func (s *XChainSource) totalVolume(book *XOrderbook) float64 {
var vol float64
for _, l := range book.Bids {
vol += l.Size
}
for _, l := range book.Asks {
vol += l.Size
}
return vol
}
// Price returns the latest price for a symbol.
func (s *XChainSource) Price(symbol string) (*Data, error) {
s.mu.RLock()
defer s.mu.RUnlock()
p, ok := s.prices[symbol]
if !ok {
return nil, ErrNotFound
}
if time.Since(s.last[symbol]) > 2*time.Second {
copy := *p
copy.Stale = true
return &copy, nil
}
copy := *p
return &copy, nil
}
// Prices returns prices for multiple symbols.
func (s *XChainSource) Prices(symbols []string) (map[string]*Data, error) {
result := make(map[string]*Data)
for _, sym := range symbols {
if p, err := s.Price(sym); err == nil {
result[sym] = p
}
}
return result, nil
}
// Orderbook returns the current orderbook for a symbol.
func (s *XChainSource) Orderbook(symbol string) (*XOrderbook, error) {
s.mu.RLock()
defer s.mu.RUnlock()
book, ok := s.books[symbol]
if !ok {
return nil, ErrNotFound
}
return book, nil
}
// Subscribe is a no-op (uses polling).
func (s *XChainSource) Subscribe(symbol string) error { return nil }
// Unsubscribe is a no-op.
func (s *XChainSource) Unsubscribe(symbol string) error { return nil }
// Healthy returns health status.
func (s *XChainSource) Healthy() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.healthy
}
// Name returns "x-chain".
func (s *XChainSource) Name() string { return "x-chain" }
// Weight returns 2.0 (native DEX chain - highest weight).
func (s *XChainSource) Weight() float64 { return 2.0 }
// Close stops the source.
func (s *XChainSource) Close() error {
s.stop()
s.mu.Lock()
s.polling = false
s.mu.Unlock()
return nil
}
// Markets returns available markets.
func (s *XChainSource) Markets() map[string]XMarket {
return s.markets
}