mirror of
https://github.com/luxfi/dex.git
synced 2026-09-24 00:26:31 +00:00
lx: fix data race in Reconciler settle/Submit paths
settle() held r.mu across user callbacks (onFailed, onSettled), which meant the callback's writes to caller-captured variables raced with the caller's later reads despite a sleep. SubmitDEXTrade/SubmitCEXTrade also returned the live merged trade pointer, and the background settle goroutine mutated Status/SettledAt/FailureReason on that same pointer under the lock — racing with any caller that read fields after return (TestReconcilerMatchCEXFirst). Two minimal, orthogonal fixes: 1. settle() captures the callback funcs + the failure reason under r.mu, releases the lock, then invokes the user callbacks outside the lock. Foreign code never runs while we hold the Reconciler's mutex, which is also good production practice (avoids re-entrant lock-ordering hazards if a callback calls back into the Reconciler). 2. SubmitDEXTrade/SubmitCEXTrade return a value-copy snapshot of the merged trade when settlement begins. The live (mutable) trade is owned by the background settle goroutine and stays inside the Reconciler — callers must use GetStatus for current state. The pending (unmatched) return path is unchanged for backward compat. Tests use sync/atomic primitives for the captured callback state so the test-side read is happens-before-ordered with the callback write. Verified with: go test -race -count=50 -run TestReconcilerBridgeSettlementFails ./pkg/lx/ go test -race -count=20 -run TestReconciler ./pkg/lx/ go test -race -count=1 ./pkg/lx/...
This commit is contained in:
+43
-16
@@ -193,6 +193,13 @@ func reconciliationKey(symbol string, side Side, price float64, size float64) st
|
||||
}
|
||||
|
||||
// SubmitDEXTrade records a DEX trade for reconciliation.
|
||||
//
|
||||
// When the trade matches a pending counterpart and settlement begins, the
|
||||
// returned *CrossVenueTrade is a value-copy snapshot of the merged trade
|
||||
// at match time. The live, mutable trade is owned by the background settle
|
||||
// goroutine and stays inside the Reconciler — callers must use GetStatus
|
||||
// for the current state. Pending (unmatched) submissions return the
|
||||
// caller-supplied pointer unchanged for backward compatibility.
|
||||
func (r *Reconciler) SubmitDEXTrade(trade *CrossVenueTrade) (*CrossVenueTrade, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
@@ -206,9 +213,12 @@ func (r *Reconciler) SubmitDEXTrade(trade *CrossVenueTrade) (*CrossVenueTrade, e
|
||||
delete(r.pendingCEX, key)
|
||||
r.active[merged.ID] = merged
|
||||
|
||||
// Snapshot before handing the live pointer to settle().
|
||||
snapshot := *merged
|
||||
|
||||
// Kick off settlement
|
||||
go r.settle(merged)
|
||||
return merged, nil
|
||||
return &snapshot, nil
|
||||
}
|
||||
|
||||
// No match yet — queue for later
|
||||
@@ -218,6 +228,8 @@ func (r *Reconciler) SubmitDEXTrade(trade *CrossVenueTrade) (*CrossVenueTrade, e
|
||||
}
|
||||
|
||||
// SubmitCEXTrade records a CEX trade for reconciliation.
|
||||
//
|
||||
// See SubmitDEXTrade for the snapshot semantics of the return value.
|
||||
func (r *Reconciler) SubmitCEXTrade(trade *CrossVenueTrade) (*CrossVenueTrade, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
@@ -236,8 +248,11 @@ func (r *Reconciler) SubmitCEXTrade(trade *CrossVenueTrade) (*CrossVenueTrade, e
|
||||
delete(r.pendingDEX, key)
|
||||
r.active[merged.ID] = merged
|
||||
|
||||
// Snapshot before handing the live pointer to settle().
|
||||
snapshot := *merged
|
||||
|
||||
go r.settle(merged)
|
||||
return merged, nil
|
||||
return &snapshot, nil
|
||||
}
|
||||
|
||||
// No match yet — queue
|
||||
@@ -306,27 +321,39 @@ func (r *Reconciler) settle(trade *CrossVenueTrade) {
|
||||
err = nil
|
||||
}
|
||||
|
||||
// Mutate internal state under the lock, then invoke user callbacks
|
||||
// outside the lock so foreign code never runs while we hold r.mu.
|
||||
// This avoids lock-ordering surprises if a callback re-enters the
|
||||
// Reconciler, and ensures the callback's own synchronization (if any)
|
||||
// is the source of happens-before for downstream readers.
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
var (
|
||||
onFailed func(*CrossVenueTrade, string)
|
||||
onSettled func(*CrossVenueTrade) error
|
||||
reason string
|
||||
)
|
||||
if err != nil {
|
||||
reason = err.Error()
|
||||
trade.Status = ReconFailed
|
||||
trade.FailureReason = err.Error()
|
||||
trade.FailureReason = reason
|
||||
delete(r.active, trade.ID)
|
||||
r.failed[trade.ID] = trade
|
||||
if r.onFailed != nil {
|
||||
r.onFailed(trade, err.Error())
|
||||
}
|
||||
return
|
||||
onFailed = r.onFailed
|
||||
} else {
|
||||
now := time.Now()
|
||||
trade.Status = ReconSettled
|
||||
trade.SettledAt = &now
|
||||
delete(r.active, trade.ID)
|
||||
r.completed[trade.ID] = trade
|
||||
onSettled = r.onSettled
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
trade.Status = ReconSettled
|
||||
trade.SettledAt = &now
|
||||
delete(r.active, trade.ID)
|
||||
r.completed[trade.ID] = trade
|
||||
if r.onSettled != nil {
|
||||
_ = r.onSettled(trade)
|
||||
if onFailed != nil {
|
||||
onFailed(trade, reason)
|
||||
}
|
||||
if onSettled != nil {
|
||||
_ = onSettled(trade)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package lx
|
||||
|
||||
import (
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -246,10 +247,10 @@ func TestReconcilerNoMatchDifferentSide(t *testing.T) {
|
||||
func TestReconcilerSettlementCallbacks(t *testing.T) {
|
||||
recon := NewReconciler(nil)
|
||||
|
||||
var settledTrade *CrossVenueTrade
|
||||
var settledTrade atomic.Pointer[CrossVenueTrade]
|
||||
recon.SetCallbacks(
|
||||
func(trade *CrossVenueTrade) error {
|
||||
settledTrade = trade
|
||||
settledTrade.Store(trade)
|
||||
return nil
|
||||
},
|
||||
nil,
|
||||
@@ -277,20 +278,24 @@ func TestReconcilerSettlementCallbacks(t *testing.T) {
|
||||
// Wait for settlement
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
require.NotNil(t, settledTrade)
|
||||
assert.Equal(t, ReconSettled, settledTrade.Status)
|
||||
assert.NotNil(t, settledTrade.SettledAt)
|
||||
got := settledTrade.Load()
|
||||
require.NotNil(t, got)
|
||||
// Read trade state through the reconciler's mutex.
|
||||
st, ok := recon.GetStatus(got.ID)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, ReconSettled, st.Status)
|
||||
assert.NotNil(t, st.SettledAt)
|
||||
}
|
||||
|
||||
func TestReconcilerBridgeSettlementFails(t *testing.T) {
|
||||
// Bridge settlement without a bridge should fail
|
||||
recon := NewReconciler(nil) // nil bridge
|
||||
|
||||
var failedReason string
|
||||
var failedReason atomic.Value // string, written from settle goroutine
|
||||
recon.SetCallbacks(
|
||||
nil,
|
||||
func(trade *CrossVenueTrade, reason string) {
|
||||
failedReason = reason
|
||||
failedReason.Store(reason)
|
||||
},
|
||||
)
|
||||
|
||||
@@ -316,7 +321,8 @@ func TestReconcilerBridgeSettlementFails(t *testing.T) {
|
||||
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
assert.Contains(t, failedReason, "bridge required")
|
||||
got, _ := failedReason.Load().(string)
|
||||
assert.Contains(t, got, "bridge required")
|
||||
stats := recon.Stats()
|
||||
assert.Equal(t, 1, stats["failed"])
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user