fix: reconcile missed market entry fills#147
Merged
anyongjin merged 1 commit intoJul 18, 2026
Merged
Conversation
Restart the private trade watcher and reconcile only stale, submitted market entry orders with positive exchange fills.
|
CodeAnt AI is reviewing your PR. |
Thanks for using CodeAnt! 🎉We're free for open-source projects. if you're enjoying it, help us grow by sharing. Share on X · |
📝 WalkthroughWalkthrough
ChangesLive order management
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant WatchMyTrades
participant watchMyTradesFn
participant handleMyTrade
WatchMyTrades->>watchMyTradesFn: Start trade stream
watchMyTradesFn-->>WatchMyTrades: Return trades or error
WatchMyTrades->>handleMyTrade: Handle non-open trade
WatchMyTrades->>watchMyTradesFn: Retry after error or closure
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
codeant-ai
Bot
added
the
size:L
Up to standards ✅🟢 Issues
|
| Metric | Results |
|---|---|
| Complexity | 17 |
| Duplication | 2 |
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
|
PR Summary by QodoFix missed market entry fills reconciliation after private WS outages
AI Description
Diagram
High-Level Assessment
Files changed (1)
|
![codeant-ai[bot]](https://avatars.githubusercontent.com/in/646884?s=60&v=4){kind=small}
Comment on lines
+1304
to
+1322
| for { | ||
| out, err := watch(map[string]interface{}{ | ||
| banexg.ParamAccount: o.Account, | ||
| }) | ||
| if err != nil { | ||
| log.Error("WatchMyTrades fail, retrying", zap.String("acc", o.Account), zap.Error(err)) | ||
| time.Sleep(retryDelay) | ||
| continue | ||
| } | ||
| o.handleMyTrade(trade) | ||
| for trade := range out { | ||
| orm.AddDumpRow(orm.DumpWsMyTrade, trade.Symbol+trade.ID, trade) | ||
| if trade.State == banexg.OdStatusOpen { | ||
| continue | ||
| } | ||
| o.handleMyTrade(trade) | ||
| } | ||
| log.Warn("WatchMyTrades stream closed, retrying", zap.String("acc", o.Account), | ||
| zap.Duration("after", retryDelay)) | ||
| time.Sleep(retryDelay) |
There was a problem hiding this comment.
Suggestion: The watcher loop has no shutdown condition, so after the app context is canceled it can keep reconnecting and sleeping forever instead of exiting. Add a context/done check before retry sleeps and before starting a new subscription attempt so the goroutine terminates cleanly. [resource leak]
Severity Level: Major ⚠️
- ❌ Trade watcher goroutine never stops on global shutdown.
- ⚠️ Process holds WebSocket connections after StopAll cancel.
- ⚠️ Resource leak complicates graceful robot shutdown procedures.Steps of Reproduction ✅
1. In `biz/biz.go:57-59`, the live bootstrap initializes `core.Ctx` and `core.StopAll` via
`ctx, cancel := context.WithCancel(context.Background()); core.Ctx = ctx; core.StopAll =
cancel`, establishing a global cancellation mechanism for goroutines.
2. A live trading run calls `biz.StartLiveOdMgr()` from `live/crypto_trader.go:377`, which
iterates configured accounts and starts per-account managers; in
`biz/odmgr_live.go:3293-3303` each manager invokes `odMgr.WatchMyTrades()` to monitor the
account trade stream.
3. `WatchMyTrades()` in `biz/odmgr_live.go:1290-1324` launches a goroutine that executes
an infinite `for` loop starting at line 1304, repeatedly calling `watch(...)`, handling
trades, logging, and sleeping on both subscription errors (`time.Sleep(retryDelay)` at
line 1310) and closed streams (`time.Sleep(retryDelay)` at line 1322) without ever
checking `core.Ctx.Done()` or any other shutdown signal.
4. When `core.StopAll()` is invoked (for example from `biz/odmgr_local.go:544` or other
shutdown paths), `core.Ctx` is canceled and other workers like
`WsDataLoader.downloadWorker` and `splitWorker` in `data/ws_loader.go:344-383` exit via
`select { case <-core.Ctx.Done(): return }`, but `WatchMyTrades` continues its
reconnect-and-sleep loop indefinitely, leaving the goroutine and its WebSocket resources
running past logical shutdown until process termination.
(Use Cmd/Ctrl + Click for best experience)
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** biz/odmgr_live.go
**Line:** 1304:1322
**Comment:**
*Resource Leak: The watcher loop has no shutdown condition, so after the app context is canceled it can keep reconnecting and sleeping forever instead of exiting. Add a context/done check before retry sleeps and before starting a new subscription attempt so the goroutine terminates cleanly.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix
Comment on lines
+1372
to
+1382
| changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus | ||
| lock.Unlock() | ||
| if err != nil { | ||
| log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), | ||
| zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) | ||
| continue | ||
| } | ||
| if changed { | ||
| log.Info("reconciled pending market entry", zap.String("acc", o.Account), | ||
| zap.String("key", od.Key()), zap.String("orderId", orderID), | ||
| zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average)) |
There was a problem hiding this comment.
Suggestion: You unlock the order and then read od.Enter fields for logging, which races with other goroutines that mutate the same order under its lock. Capture the values you want to log before unlocking (or re-lock for read) to avoid concurrent unsynchronized access. [race condition]
Severity Level: Major ⚠️
- ❌ Data race possible when reconciling stale market entries.
- ⚠️ Logs may show inconsistent filled and average values.
- ⚠️ Race detector could flag goroutine at reconciliation logging.Steps of Reproduction ✅
1. The live scheduler in `live/common.go:305-309` periodically obtains a `LiveOrderMgr`
via `biz.GetLiveOdMgr(account)` and calls `_, err := odMgr.SyncLocalOrders()`, driving the
reconciliation path on real trading accounts.
2. Inside `SyncLocalOrders()` (`biz/odmgr_live.go:131-166`), open orders are collected
into `pendingEntries` and `o.reconcilePendingMarketEntries(pendingEntries)` is invoked at
line 164 to repair stale market entry orders whose WebSocket fills were missed.
3. In `reconcilePendingMarketEntries()` (`biz/odmgr_live.go:131-185`), for each eligible
order, the function acquires `lock := od.Lock()`, inspects `od.Enter`, unlocks to call
`exg.Default.FetchOrder`, and then re-locks; while holding the lock, it calls `err =
o.applyAuthoritativeEnterOrder(od, res)` at line 1371 and computes `changed :=
enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus` at line 1372.
4. The code then executes `lock.Unlock()` at line 1373 and immediately performs logging at
line 1382: `log.Info("reconciled pending market entry", ..., zap.Float64("filled",
od.Enter.Filled), zap.Float64("average", od.Enter.Average))`, reading `od.Enter` fields
without any lock, which can race with other goroutines like `ConsumeOrderQueue()`
(`biz/odmgr_live.go:1206-1258`) and `handleMyTrade()` (`biz/odmgr_live.go:1386-1481`) that
concurrently call `od.Lock()` and mutate `od.Enter` / `od.Status`; running `go test ./biz
-race` while these paths operate on the same order ID would surface a data race at this
logging site.
(Use Cmd/Ctrl + Click for best experience)
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** biz/odmgr_live.go
**Line:** 1372:1382
**Comment:**
*Race Condition: You unlock the order and then read `od.Enter` fields for logging, which races with other goroutines that mutate the same order under its lock. Capture the values you want to log before unlocking (or re-lock for read) to avoid concurrent unsynchronized access.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix
|
CodeAnt AI finished reviewing your PR. |
Code Review by Qodo
1. Racy reconcile log reads
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
biz/odmgr_live.go (2)
157-172: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winPotential data race and stale state on
openOdsafter network I/O.
openOdsis obtained and copied while holdinglock, but then the lock is released to perform network I/O inreconcilePendingMarketEntries. Once reconciliation completes, the lock is re-acquired, and the originalopenOdsslice is iterated.This introduces two significant risks:
- Data Race: If
ormo.GetOpenODsreturns a reference to a shared slice (rather than a deep copy of the slice header), another goroutine might modify the underlying array while the lock is released, causing a race during the iteration at line 172.- Stale State: Reconciliation or other concurrent operations might have closed or created orders during the network calls. Using the old
openOdslist means the subsequent sync logic operates on stale data.To prevent this, re-fetch the open orders after reconciliation completes.
🔒️ Proposed fix to re-fetch open orders
openOds, lock := ormo.GetOpenODs(o.Account) lock.Lock() pendingEntries := make([]*ormo.InOutOrder, 0, len(openOds)) for _, od := range openOds { pendingEntries = append(pendingEntries, od) } lock.Unlock() o.reconcilePendingMarketEntries(pendingEntries) + // Re-fetch open orders as the state might have changed during reconciliation + openOds, lock = ormo.GetOpenODs(o.Account) // 按symbol分组本地订单 lock.Lock() // 过滤重复订单 var duplicateOdNum = 0 var checkKeys = make(map[string]bool) odMap := make(map[string]map[bool][]*ormo.InOutOrder) for _, od := range openOds {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@biz/odmgr_live.go` around lines 157 - 172, Re-fetch the current open orders after reconcilePendingMarketEntries returns, before the second lock acquisition and the duplicate-order grouping loop. Replace the stale openOds value used by the subsequent synchronization logic with the newly retrieved snapshot, preserving lock protection while iterating it.
157-172: 🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy liftData race on shared
openOdsslice after network I/O.By copying
openOdsintopendingEntriesunderlock.Lock(), the code correctly acknowledges thatopenOdsis a shared reference that must be protected. However, after releasing the lock for network I/O and re-acquiring it at line 167, the code iterates over the originalopenOdsslice (line 172).If another goroutine adds or removes an order (e.g., via a swap-and-pop removal) while the lock is released, the underlying array is mutated in place. Because your local
openOdsslice header retains the old length, iterating it will result in a data race and process corrupted/stale elements.You must re-fetch the open orders list after the reconciliation to safely proceed with up-to-date state.
🔒️ Proposed fix
openOds, lock := ormo.GetOpenODs(o.Account) lock.Lock() pendingEntries := make([]*ormo.InOutOrder, 0, len(openOds)) for _, od := range openOds { pendingEntries = append(pendingEntries, od) } lock.Unlock() o.reconcilePendingMarketEntries(pendingEntries) + // Re-fetch open orders as the global list may have changed during network I/O + openOds, lock = ormo.GetOpenODs(o.Account) // 按symbol分组本地订单 lock.Lock() // 过滤重复订单 var duplicateOdNum = 0 var checkKeys = make(map[string]bool) odMap := make(map[string]map[bool][]*ormo.InOutOrder) for _, od := range openOds {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@biz/odmgr_live.go` around lines 157 - 172, Re-fetch the open orders list after reconcilePendingMarketEntries returns and before the second lock-protected iteration. Replace the stale openOds reference with the newly fetched list, acquiring its corresponding lock before building odMap and processing orders; keep the initial snapshot used for pendingEntries unchanged.
🧹 Nitpick comments (1)
biz/odmgr_live.go (1)
1330-1385: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReduce cognitive complexity by extracting the loop body.
As highlighted by SonarCloud, this method has a high cognitive complexity. Extracting the reconciliation logic for a single order into a separate helper method will make the code easier to read, test, and maintain.
♻️ Proposed refactor
func (o *LiveOrderMgr) reconcilePendingMarketEntries(ods []*ormo.InOutOrder) { cutoff := btime.UTCStamp() - pendingMarketEntryReconcileAfter.Milliseconds() for _, od := range ods { - if od == nil { - continue - } - lock := od.Lock() - enter := od.Enter - if enter == nil || enter.OrderID == "" || enter.OrderType != banexg.OdTypeMarket || - enter.Status == ormo.OdStatusClosed || enter.Filled > AmtDust || enter.UpdateAt > cutoff { - lock.Unlock() - continue - } - orderID, symbol := enter.OrderID, od.Symbol - lock.Unlock() - - res, err := exg.Default.FetchOrder(symbol, orderID, map[string]interface{}{ - banexg.ParamAccount: o.Account, - }) - if err != nil { - log.Warn("reconcile pending market entry fetch fail", zap.String("acc", o.Account), - zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) - continue - } - if res == nil || res.ID == "" || res.ID != orderID { - log.Warn("reconcile pending market entry invalid response", zap.String("acc", o.Account), - zap.String("key", od.Key()), zap.String("orderId", orderID)) - continue - } - if res.Filled <= AmtDust { - continue - } - - lock = od.Lock() - enter = od.Enter - if enter == nil || enter.OrderID != orderID || enter.OrderType != banexg.OdTypeMarket || - enter.Status == ormo.OdStatusClosed || enter.Filled > AmtDust { - lock.Unlock() - continue - } - beforeFilled, beforeStatus := enter.Filled, od.Status - err = o.applyAuthoritativeEnterOrder(od, res) - changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus - lock.Unlock() - if err != nil { - log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), - zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) - continue - } - if changed { - log.Info("reconciled pending market entry", zap.String("acc", o.Account), - zap.String("key", od.Key()), zap.String("orderId", orderID), - zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average)) - } + o.reconcileSingleMarketEntry(od, cutoff) } } + +func (o *LiveOrderMgr) reconcileSingleMarketEntry(od *ormo.InOutOrder, cutoff int64) { + if od == nil { + return + } + lock := od.Lock() + enter := od.Enter + if enter == nil || enter.OrderID == "" || enter.OrderType != banexg.OdTypeMarket || + enter.Status == ormo.OdStatusClosed || enter.Filled > AmtDust || enter.UpdateAt > cutoff { + lock.Unlock() + return + } + orderID, symbol := enter.OrderID, od.Symbol + lock.Unlock() + + res, err := exg.Default.FetchOrder(symbol, orderID, map[string]interface{}{ + banexg.ParamAccount: o.Account, + }) + if err != nil { + log.Warn("reconcile pending market entry fetch fail", zap.String("acc", o.Account), + zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) + return + } + if res == nil || res.ID == "" || res.ID != orderID { + log.Warn("reconcile pending market entry invalid response", zap.String("acc", o.Account), + zap.String("key", od.Key()), zap.String("orderId", orderID)) + return + } + if res.Filled <= AmtDust { + return + } + + lock = od.Lock() + enter = od.Enter + if enter == nil || enter.OrderID != orderID || enter.OrderType != banexg.OdTypeMarket || + enter.Status == ormo.OdStatusClosed || enter.Filled > AmtDust { + lock.Unlock() + return + } + beforeFilled, beforeStatus := enter.Filled, od.Status + err = o.applyAuthoritativeEnterOrder(od, res) + changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus + lock.Unlock() + + if err != nil { + log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), + zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) + return + } + if changed { + log.Info("reconciled pending market entry", zap.String("acc", o.Account), + zap.String("key", od.Key()), zap.String("orderId", orderID), + zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average)) + } +}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@biz/odmgr_live.go` around lines 1330 - 1385, Reduce the cognitive complexity of LiveOrderMgr.reconcilePendingMarketEntries by extracting the per-order reconciliation logic from its loop into a dedicated helper method. Keep nil handling and iteration in reconcilePendingMarketEntries, move the locking, fetch validation, authoritative update, and related logging into the helper, and preserve all existing behavior.Source: Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@biz/odmgr_live.go`:
- Around line 1372-1383: In the reconciliation flow around the changed
calculation, capture od.Enter.Filled and od.Enter.Average into local variables
while lock is held, before lock.Unlock(), and use those locals in the subsequent
“reconciled pending market entry” log. Keep the existing changed check and
logging behavior unchanged.
- Around line 1290-1325: Add a shutdown signal for the goroutine started by
LiveOrderMgr.WatchMyTrades, check it before reconnecting and retry sleeping, and
exit when shutdown is requested. Update the manager cleanup path, including
CleanUp, to trigger and clear the watcher’s stop state so CleanUpOdMgr stops the
loop and prevents the goroutine from leaking.
---
Outside diff comments:
In `@biz/odmgr_live.go`:
- Around line 157-172: Re-fetch the current open orders after
reconcilePendingMarketEntries returns, before the second lock acquisition and
the duplicate-order grouping loop. Replace the stale openOds value used by the
subsequent synchronization logic with the newly retrieved snapshot, preserving
lock protection while iterating it.
- Around line 157-172: Re-fetch the open orders list after
reconcilePendingMarketEntries returns and before the second lock-protected
iteration. Replace the stale openOds reference with the newly fetched list,
acquiring its corresponding lock before building odMap and processing orders;
keep the initial snapshot used for pendingEntries unchanged.
---
Nitpick comments:
In `@biz/odmgr_live.go`:
- Around line 1330-1385: Reduce the cognitive complexity of
LiveOrderMgr.reconcilePendingMarketEntries by extracting the per-order
reconciliation logic from its loop into a dedicated helper method. Keep nil
handling and iteration in reconcilePendingMarketEntries, move the locking, fetch
validation, authoritative update, and related logging into the helper, and
preserve all existing behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
1290
to
1325
| func (o *LiveOrderMgr) WatchMyTrades() { | ||
| out, err := exg.Default.WatchMyTrades(map[string]interface{}{ | ||
| banexg.ParamAccount: o.Account, | ||
| }) | ||
| if err != nil { | ||
| log.Error("WatchMyTrades fail", zap.String("acc", o.Account), zap.Error(err)) | ||
| return | ||
| } | ||
| if !atomic.CompareAndSwapInt32(&o.isWatchMyTrade, 0, 1) { | ||
| return | ||
| } | ||
| watch := o.watchMyTradesFn | ||
| if watch == nil { | ||
| watch = exg.Default.WatchMyTrades | ||
| } | ||
| retryDelay := o.myTradeWatchRetry | ||
| if retryDelay <= 0 { | ||
| retryDelay = defaultMyTradeWatchRetry | ||
| } | ||
| go func() { | ||
| defer func() { | ||
| atomic.StoreInt32(&o.isWatchMyTrade, 0) | ||
| }() | ||
| for trade := range out { | ||
| orm.AddDumpRow(orm.DumpWsMyTrade, trade.Symbol+trade.ID, trade) | ||
| if trade.State == banexg.OdStatusOpen { | ||
| defer atomic.StoreInt32(&o.isWatchMyTrade, 0) | ||
| for { | ||
| out, err := watch(map[string]interface{}{ | ||
| banexg.ParamAccount: o.Account, | ||
| }) | ||
| if err != nil { | ||
| log.Error("WatchMyTrades fail, retrying", zap.String("acc", o.Account), zap.Error(err)) | ||
| time.Sleep(retryDelay) | ||
| continue | ||
| } | ||
| o.handleMyTrade(trade) | ||
| for trade := range out { | ||
| orm.AddDumpRow(orm.DumpWsMyTrade, trade.Symbol+trade.ID, trade) | ||
| if trade.State == banexg.OdStatusOpen { | ||
| continue | ||
| } | ||
| o.handleMyTrade(trade) | ||
| } | ||
| log.Warn("WatchMyTrades stream closed, retrying", zap.String("acc", o.Account), | ||
| zap.Duration("after", retryDelay)) | ||
| time.Sleep(retryDelay) | ||
| } | ||
| }() | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Check for shutdown/cleanup mechanisms in LiveOrderMgr.
# Look for CleanUp or similar methods to see how other goroutines are stopped.
ast-grep run --pattern $'func (o *LiveOrderMgr) CleanUp() {
$$$
}' --lang go biz/odmgr_live.go
# Check if there is a context or a stopped flag on LiveOrderMgr
rg -A 5 -B 5 "type LiveOrderMgr struct" biz/odmgr_live.goRepository: banbox/banbot
Length of output: 642
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the relevant file structure.
ast-grep outline biz/odmgr_live.go --view expanded | sed -n '1,220p'
# Find shutdown/cleanup-related methods and flags in the file.
rg -n "func \\(o \\*LiveOrderMgr\\) (CleanUp|Close|Stop|Shutdown|Release)|isWatchMyTrade|done|closed|context\\.Context" biz/odmgr_live.go
# Show the LiveOrderMgr struct and nearby methods around WatchMyTrades.
sed -n '1,260p' biz/odmgr_live.go
echo '--- WATCH SECTION ---'
sed -n '1260,1365p' biz/odmgr_live.goRepository: banbox/banbot
Length of output: 20908
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect shutdown path and any cleanup side effects around LiveOrderMgr.
sed -n '3260,3310p' biz/odmgr_live.go
echo '--- OnEnvEnd / shutdown references ---'
rg -n "OnEnvEnd|CleanUp\\(|StartLiveOdMgr|WatchMyTrades\\(|WatchLeverages\\(|TrialUnMatchesForever\\(" biz/odmgr_live.go
echo '--- CleanUp call sites in repo ---'
rg -n "\.CleanUp\\(" biz
echo '--- WatchMyTrades call sites in repo ---'
rg -n "WatchMyTrades\\(" bizRepository: banbox/banbot
Length of output: 2108
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,150p' biz/odmgr.go
echo '---'
sed -n '150,260p' biz/odmgr.goRepository: banbox/banbot
Length of output: 7577
Add a stop signal to WatchMyTrades
WatchMyTrades() keeps reconnecting forever, and CleanUp() is currently a no-op, so CleanUpOdMgr() won’t stop this goroutine. Add a shutdown flag/context check in the loop and clear it from cleanup to avoid leaking the watcher after env end.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@biz/odmgr_live.go` around lines 1290 - 1325, Add a shutdown signal for the
goroutine started by LiveOrderMgr.WatchMyTrades, check it before reconnecting
and retry sleeping, and exit when shutdown is requested. Update the manager
cleanup path, including CleanUp, to trigger and clear the watcher’s stop state
so CleanUpOdMgr stops the loop and prevents the goroutine from leaking.
Comment on lines
+1372
to
+1383
| changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus | ||
| lock.Unlock() | ||
| if err != nil { | ||
| log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), | ||
| zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) | ||
| continue | ||
| } | ||
| if changed { | ||
| log.Info("reconciled pending market entry", zap.String("acc", o.Account), | ||
| zap.String("key", od.Key()), zap.String("orderId", orderID), | ||
| zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average)) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Data race on od.Enter fields during logging.
Reading od.Enter.Filled and od.Enter.Average for logging after releasing the lock can result in a data race if another goroutine modifies them concurrently. Capture these values into local variables before unlocking to ensure thread safety.
🐛 Proposed fix
beforeFilled, beforeStatus := enter.Filled, od.Status
err = o.applyAuthoritativeEnterOrder(od, res)
changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus
+ newFilled := enter.Filled
+ newAverage := enter.Average
lock.Unlock()
if err != nil {
log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account),
zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err))
continue
}
if changed {
log.Info("reconciled pending market entry", zap.String("acc", o.Account),
zap.String("key", od.Key()), zap.String("orderId", orderID),
- zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average))
+ zap.Float64("filled", newFilled), zap.Float64("average", newAverage))
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus | |
| lock.Unlock() | |
| if err != nil { | |
| log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), | |
| zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) | |
| continue | |
| } | |
| if changed { | |
| log.Info("reconciled pending market entry", zap.String("acc", o.Account), | |
| zap.String("key", od.Key()), zap.String("orderId", orderID), | |
| zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average)) | |
| } | |
| changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus | |
| newFilled := enter.Filled | |
| newAverage := enter.Average | |
| lock.Unlock() | |
| if err != nil { | |
| log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), | |
| zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) | |
| continue | |
| } | |
| if changed { | |
| log.Info("reconciled pending market entry", zap.String("acc", o.Account), | |
| zap.String("key", od.Key()), zap.String("orderId", orderID), | |
| zap.Float64("filled", newFilled), zap.Float64("average", newAverage)) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@biz/odmgr_live.go` around lines 1372 - 1383, In the reconciliation flow
around the changed calculation, capture od.Enter.Filled and od.Enter.Average
into local variables while lock is held, before lock.Unlock(), and use those
locals in the subsequent “reconciled pending market entry” log. Keep the
existing changed check and logging behavior unchanged.
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+1363
to
+1383
| lock = od.Lock() | ||
| enter = od.Enter | ||
| if enter == nil || enter.OrderID != orderID || enter.OrderType != banexg.OdTypeMarket || | ||
| enter.Status == ormo.OdStatusClosed || enter.Filled > AmtDust { | ||
| lock.Unlock() | ||
| continue | ||
| } | ||
| beforeFilled, beforeStatus := enter.Filled, od.Status | ||
| err = o.applyAuthoritativeEnterOrder(od, res) | ||
| changed := enter.Filled > beforeFilled+AmtDust || od.Status != beforeStatus | ||
| lock.Unlock() | ||
| if err != nil { | ||
| log.Error("reconcile pending market entry apply fail", zap.String("acc", o.Account), | ||
| zap.String("key", od.Key()), zap.String("orderId", orderID), zap.Error(err)) | ||
| continue | ||
| } | ||
| if changed { | ||
| log.Info("reconciled pending market entry", zap.String("acc", o.Account), | ||
| zap.String("key", od.Key()), zap.String("orderId", orderID), | ||
| zap.Float64("filled", od.Enter.Filled), zap.Float64("average", od.Enter.Average)) | ||
| } |
There was a problem hiding this comment.
1. Racy reconcile log reads 🐞 Bug ☼ Reliability
reconcilePendingMarketEntries logs od.Enter.Filled/Average after releasing od.Lock(), which can race with concurrent order updates and trigger Go’s race detector / produce inconsistent logs. Capture the values while holding the order lock (or re-lock for reading) before logging.
Agent Prompt
### Issue description
`reconcilePendingMarketEntries` unlocks the order and then reads `od.Enter.Filled` and `od.Enter.Average` for logging. Other goroutines mutate these fields under `od.Lock()` (e.g., trade handling / order queue), so the unlocked reads can race.
### Issue Context
Order mutation is synchronized via `InOutOrder.Lock()`; logging should not read mutable fields after releasing that lock.
### Fix Focus Areas
- biz/odmgr_live.go[1330-1384]
### Suggested fix
While holding `od.Lock()` (before `lock.Unlock()`), copy any values you need for logging into local variables (e.g., `filled := enter.Filled`, `avg := enter.Average`, maybe `status := od.Status`). After unlocking, log using the copied locals instead of reading `od.Enter.*`.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
1302
to
1323
| go func() { | ||
| defer func() { | ||
| atomic.StoreInt32(&o.isWatchMyTrade, 0) | ||
| }() | ||
| for trade := range out { | ||
| orm.AddDumpRow(orm.DumpWsMyTrade, trade.Symbol+trade.ID, trade) | ||
| if trade.State == banexg.OdStatusOpen { | ||
| defer atomic.StoreInt32(&o.isWatchMyTrade, 0) | ||
| for { | ||
| out, err := watch(map[string]interface{}{ | ||
| banexg.ParamAccount: o.Account, | ||
| }) | ||
| if err != nil { | ||
| log.Error("WatchMyTrades fail, retrying", zap.String("acc", o.Account), zap.Error(err)) | ||
| time.Sleep(retryDelay) | ||
| continue | ||
| } | ||
| o.handleMyTrade(trade) | ||
| for trade := range out { | ||
| orm.AddDumpRow(orm.DumpWsMyTrade, trade.Symbol+trade.ID, trade) | ||
| if trade.State == banexg.OdStatusOpen { | ||
| continue | ||
| } | ||
| o.handleMyTrade(trade) | ||
| } | ||
| log.Warn("WatchMyTrades stream closed, retrying", zap.String("acc", o.Account), | ||
| zap.Duration("after", retryDelay)) | ||
| time.Sleep(retryDelay) | ||
| } |
There was a problem hiding this comment.
2. Watchmytrades ignores shutdown 🐞 Bug ☼ Reliability
The new WatchMyTrades retry loop never checks core.Ctx.Done() and uses uninterruptible time.Sleep, so it can keep reconnecting and logging retries indefinitely after shutdown is requested. Add context cancellation checks around the watch call, channel processing, and retry sleep.
Agent Prompt
### Issue description
`WatchMyTrades` now runs an infinite reconnect loop, but it does not observe `core.Ctx.Done()` and sleeps via `time.Sleep`, which cannot be cancelled. This makes the goroutine continue retrying during shutdown.
### Issue Context
Other long-running goroutines in this file (e.g., `ConsumeOrderQueue`) exit when `core.Ctx.Done()` is closed.
### Fix Focus Areas
- biz/odmgr_live.go[1206-1224]
- biz/odmgr_live.go[1290-1324]
### Suggested fix
- Add `select { case <-core.Ctx.Done(): return ... }` checks in the outer `for` loop.
- Replace `time.Sleep(retryDelay)` with a cancellable wait:
- `select { case <-core.Ctx.Done(): return; case <-time.After(retryDelay): }`
- Optionally, process trades with a `select` on `<-core.Ctx.Done()` and `<-out` so shutdown can stop even if the stream stays open.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
User description
Summary
Fixes the live-order state gap where a Bybit market entry is filled on the exchange but remains
filled=0/average=0and appears as not entered locally after a private WebSocket outage.WatchMyTradesafter subscription errors or a closed stream.Filled=0throughFetchOrder.Validation
go test ./biz -run 'Test(ReconcilePendingMarketEntries|WatchMyTradesRestartsAfterClosedStream)' -count=1 -vgo test ./biz -count=1No test files are included in this PR.
CodeAnt-AI Description
Restore missed market entry fills after a private trade stream outage and keep the stream reconnecting automatically
What Changed
Impact
✅ Fewer missed entry fills✅ Fewer orders stuck in pending state✅ Faster recovery after trade stream drops💡 Usage Guide
Checking Your Pull Request
Every time you make a pull request, our system automatically looks through it. We check for security issues, mistakes in how you're setting up your infrastructure, and common code problems. We do this to make sure your changes are solid and won't cause any trouble later.
Talking to CodeAnt AI
Got a question or need a hand with something in your pull request? You can easily get in touch with CodeAnt AI right here. Just type the following in a comment on your pull request, and replace "Your question here" with whatever you want to ask:
This lets you have a chat with CodeAnt AI about your pull request, making it easier to understand and improve your code.
Example
Preserve Org Learnings with CodeAnt
You can record team preferences so CodeAnt AI applies them in future reviews. Reply directly to the specific CodeAnt AI suggestion (in the same thread) and replace "Your feedback here" with your input:
This helps CodeAnt AI learn and adapt to your team's coding style and standards.
Example
Retrigger review
Ask CodeAnt AI to review the PR again, by typing:
Check Your Repository Health
To analyze the health of your code repository, visit our dashboard at https://app.codeant.ai. This tool helps you identify potential issues and areas for improvement in your codebase, ensuring your repository maintains high standards of code health.
Summary by CodeRabbit