mirror of
https://github.com/luxfi/bridge.git
synced 2026-09-24 00:26:46 +00:00
merge: z/operator-features
This commit is contained in:
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -53,6 +54,14 @@ type DepositWatcher struct {
|
||||
// slow upstream RPC can't stall the whole tick.
|
||||
perCheckTimeout time.Duration
|
||||
|
||||
// expireAfter is the age (since CreatedAt) at which a
|
||||
// user_deposit_pending swap is auto-cancelled because the user
|
||||
// never sent the deposit. Closes the final hardening-matrix gap:
|
||||
// every other pipeline stage has a terminal escape, but
|
||||
// user_deposit_pending used to grow without bound. Zero ⇒ disabled
|
||||
// (legacy "keep pending forever" behaviour, preserved for back-compat).
|
||||
expireAfter time.Duration
|
||||
|
||||
// running flips to true on Run + back to false on context cancel.
|
||||
// Exported via Running() so /health and tests can observe.
|
||||
running atomic.Bool
|
||||
@@ -63,6 +72,7 @@ type DepositWatcher struct {
|
||||
advances atomic.Uint64
|
||||
checkErrors atomic.Uint64
|
||||
listErrors atomic.Uint64
|
||||
expired atomic.Uint64
|
||||
stopOnce sync.Once
|
||||
cancelRunning context.CancelFunc
|
||||
}
|
||||
@@ -92,6 +102,25 @@ const DefaultWatcherInterval = 15 * time.Second
|
||||
// DefaultPerCheckTimeout caps each individual depositcheck.Check call.
|
||||
const DefaultPerCheckTimeout = 8 * time.Second
|
||||
|
||||
// DefaultDepositExpireAfter is the cap on how long a swap can sit in
|
||||
// user_deposit_pending before the watcher auto-cancels it. 24h is
|
||||
// generous enough to never accidentally cancel a real user (even
|
||||
// slow-to-act humans send deposits in minutes-to-hours) while bounding
|
||||
// the store growth from abandoned swap intents.
|
||||
const DefaultDepositExpireAfter = 24 * time.Hour
|
||||
|
||||
// SetExpireAfter configures the auto-cancel threshold for stale
|
||||
// user_deposit_pending swaps. After this age (since CreatedAt) a
|
||||
// swap is moved to SwapStatusCancelled with an operator-actionable
|
||||
// LastError. Zero disables (legacy: keep forever). Negative values
|
||||
// are clamped to zero.
|
||||
func (w *DepositWatcher) SetExpireAfter(d time.Duration) {
|
||||
if d < 0 {
|
||||
d = 0
|
||||
}
|
||||
w.expireAfter = d
|
||||
}
|
||||
|
||||
// Running reports whether the watcher loop is active.
|
||||
func (w *DepositWatcher) Running() bool { return w.running.Load() }
|
||||
|
||||
@@ -104,6 +133,12 @@ type WatcherStats struct {
|
||||
Advances uint64 `json:"advances"`
|
||||
CheckErrors uint64 `json:"check_errors"`
|
||||
ListErrors uint64 `json:"list_errors"`
|
||||
// Expired counts user_deposit_pending swaps that were auto-
|
||||
// cancelled because their age exceeded --deposit-expire-after.
|
||||
// Non-zero on a healthy bridge means real users abandoned their
|
||||
// quotes (or someone is spamming the create endpoint); a sudden
|
||||
// spike is a smell (UX regression on the deposit step?).
|
||||
Expired uint64 `json:"expired"`
|
||||
}
|
||||
|
||||
// Stats snapshots the current counters. Safe for concurrent reads.
|
||||
@@ -114,6 +149,7 @@ func (w *DepositWatcher) Stats() WatcherStats {
|
||||
Advances: w.advances.Load(),
|
||||
CheckErrors: w.checkErrors.Load(),
|
||||
ListErrors: w.listErrors.Load(),
|
||||
Expired: w.expired.Load(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -192,11 +228,78 @@ func (w *DepositWatcher) tick(ctx context.Context) {
|
||||
if w.logger != nil {
|
||||
w.logger.Debug("deposit watcher tick", "pending", len(swaps))
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
for _, sw := range swaps {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
// Deposit check first — a deposit that just landed should
|
||||
// advance the swap even when it's right at the edge of
|
||||
// expireAfter. checkOne is no-op-idempotent: if it advances,
|
||||
// maybeExpire's status guard prevents the cancel.
|
||||
w.checkOne(ctx, sw)
|
||||
w.maybeExpire(ctx, sw, now)
|
||||
}
|
||||
}
|
||||
|
||||
// maybeExpire auto-cancels a stale user_deposit_pending swap whose
|
||||
// CreatedAt is older than expireAfter. The status-guard inside Patch
|
||||
// keeps this safe against the race where checkOne just advanced the
|
||||
// swap on the same tick — only swaps still in user_deposit_pending
|
||||
// after the guard get cancelled.
|
||||
//
|
||||
// Disabled when expireAfter == 0 (back-compat: legacy "pending forever").
|
||||
func (w *DepositWatcher) maybeExpire(ctx context.Context, sw *Swap, now time.Time) {
|
||||
if w.expireAfter <= 0 {
|
||||
return
|
||||
}
|
||||
if sw.Status != SwapStatusUserDepositPending {
|
||||
// Already advanced (or cancelled by a prior tick) — nothing to do.
|
||||
return
|
||||
}
|
||||
if sw.CreatedAt.IsZero() {
|
||||
// Defensive: no birth timestamp ⇒ cannot compute age.
|
||||
// Skip rather than auto-cancel something we can't reason about.
|
||||
return
|
||||
}
|
||||
age := now.Sub(sw.CreatedAt)
|
||||
if age < w.expireAfter {
|
||||
return
|
||||
}
|
||||
|
||||
patched, err := w.store.Patch(ctx, sw.ID, func(s *Swap) {
|
||||
// Status guard inside the lock: don't expire a swap that
|
||||
// another tick (or a concurrent admin reset) just advanced.
|
||||
if s.Status != SwapStatusUserDepositPending {
|
||||
return
|
||||
}
|
||||
s.Status = SwapStatusCancelled
|
||||
s.LastError = fmt.Sprintf(
|
||||
"Auto-cancelled after %s of no deposit (threshold: %s) — the create-time deposit address was never funded. Create a new quote to retry.",
|
||||
age.Round(time.Second), w.expireAfter,
|
||||
)
|
||||
s.LastErrorAt = now
|
||||
})
|
||||
if err != nil {
|
||||
// Store error during expiry — log but don't count it under
|
||||
// expired/. The next tick will retry.
|
||||
if w.logger != nil {
|
||||
w.logger.Warn("expire swap",
|
||||
"swap_id", sw.ID,
|
||||
"err", err,
|
||||
)
|
||||
}
|
||||
return
|
||||
}
|
||||
if patched != nil && patched.Status == SwapStatusCancelled {
|
||||
w.expired.Add(1)
|
||||
if w.logger != nil {
|
||||
w.logger.Info("expired stale user_deposit_pending swap",
|
||||
"swap_id", sw.ID,
|
||||
"age", age.Round(time.Second),
|
||||
"threshold", w.expireAfter,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -384,3 +384,186 @@ func paramsString(p depositcheck.CheckParams) string {
|
||||
}
|
||||
|
||||
var _ = paramsString // referenced by future tests; kept to avoid an unused-fn warning
|
||||
|
||||
// =============================================================================
|
||||
// Auto-expire (--deposit-expire-after) — the last hardening pass.
|
||||
// =============================================================================
|
||||
//
|
||||
// Pattern mirrors the orphan-recovery tests in refund_driver_test.go:
|
||||
// temporarily pin the store's now func so we can seed a swap with a
|
||||
// CreatedAt back-dated by an arbitrary amount, then restore.
|
||||
|
||||
// seedAgedPendingSwap creates a SwapStatusUserDepositPending row whose
|
||||
// CreatedAt is exactly agedBy in the past. Restores the store's now
|
||||
// hook on test cleanup.
|
||||
func seedAgedPendingSwap(t *testing.T, store *InMemoryStore, agedBy time.Duration, mutate func(*Swap)) *Swap {
|
||||
t.Helper()
|
||||
realNow := store.now
|
||||
pinned := realNow().Add(-agedBy)
|
||||
store.now = func() time.Time { return pinned }
|
||||
t.Cleanup(func() { store.now = realNow })
|
||||
|
||||
sw := &Swap{
|
||||
Status: SwapStatusUserDepositPending,
|
||||
Amount: 0.5,
|
||||
SourceNetwork: "ETHEREUM_SEPOLIA",
|
||||
SourceAsset: "ETH",
|
||||
DestinationNetwork: "LUX_TESTNET",
|
||||
DestinationAsset: "LUX",
|
||||
DestinationAddress: "0xdest",
|
||||
DepositAddress: "bridge-test-wallet###0xdeposit",
|
||||
UseDepositAddress: true,
|
||||
}
|
||||
if mutate != nil {
|
||||
mutate(sw)
|
||||
}
|
||||
if err := store.Create(t.Context(), sw); err != nil {
|
||||
t.Fatalf("seed aged pending swap: %v", err)
|
||||
}
|
||||
return sw
|
||||
}
|
||||
|
||||
func TestWatcher_Expire_PastThresholdCancels(t *testing.T) {
|
||||
store := NewInMemoryStore()
|
||||
fc := newFakeChecker()
|
||||
w := NewDepositWatcher(store, fc, time.Second, nil)
|
||||
w.SetExpireAfter(1 * time.Hour)
|
||||
|
||||
sw := seedAgedPendingSwap(t, store, 2*time.Hour, nil)
|
||||
w.Tick(t.Context())
|
||||
|
||||
got, _ := store.Get(t.Context(), sw.ID)
|
||||
if got.Status != SwapStatusCancelled {
|
||||
t.Errorf("status = %q, want cancelled", got.Status)
|
||||
}
|
||||
if !strings.Contains(got.LastError, "Auto-cancelled") {
|
||||
t.Errorf("LastError = %q, want substring 'Auto-cancelled'", got.LastError)
|
||||
}
|
||||
if w.Stats().Expired != 1 {
|
||||
t.Errorf("Expired = %d, want 1", w.Stats().Expired)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWatcher_Expire_BelowThresholdPreserved(t *testing.T) {
|
||||
store := NewInMemoryStore()
|
||||
w := NewDepositWatcher(store, newFakeChecker(), time.Second, nil)
|
||||
w.SetExpireAfter(1 * time.Hour)
|
||||
|
||||
sw := seedAgedPendingSwap(t, store, 30*time.Minute, nil)
|
||||
w.Tick(t.Context())
|
||||
|
||||
got, _ := store.Get(t.Context(), sw.ID)
|
||||
if got.Status != SwapStatusUserDepositPending {
|
||||
t.Errorf("status = %q, want still user_deposit_pending", got.Status)
|
||||
}
|
||||
if w.Stats().Expired != 0 {
|
||||
t.Errorf("Expired = %d, want 0", w.Stats().Expired)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWatcher_Expire_ZeroDisables(t *testing.T) {
|
||||
store := NewInMemoryStore()
|
||||
w := NewDepositWatcher(store, newFakeChecker(), time.Second, nil)
|
||||
w.SetExpireAfter(0) // disabled
|
||||
|
||||
sw := seedAgedPendingSwap(t, store, 100*time.Hour, nil)
|
||||
w.Tick(t.Context())
|
||||
|
||||
got, _ := store.Get(t.Context(), sw.ID)
|
||||
if got.Status != SwapStatusUserDepositPending {
|
||||
t.Errorf("status = %q, want preserved (expire disabled)", got.Status)
|
||||
}
|
||||
if w.Stats().Expired != 0 {
|
||||
t.Errorf("Expired = %d, want 0 — expire was disabled", w.Stats().Expired)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWatcher_Expire_DepositConfirmedWinsOverExpiry verifies the
|
||||
// invariant: a deposit that just confirmed advances the swap even
|
||||
// when it's already past the expiry threshold. checkOne runs before
|
||||
// maybeExpire and the maybeExpire status guard then no-ops.
|
||||
func TestWatcher_Expire_DepositConfirmedWinsOverExpiry(t *testing.T) {
|
||||
store := NewInMemoryStore()
|
||||
fc := newFakeChecker()
|
||||
w := NewDepositWatcher(store, fc, time.Second, nil)
|
||||
w.SetExpireAfter(1 * time.Hour)
|
||||
|
||||
sw := seedAgedPendingSwap(t, store, 2*time.Hour, nil)
|
||||
// Same tick: the deposit lands.
|
||||
fc.setVerdict("ETHEREUM_SEPOLIA", "0xdeposit", true)
|
||||
|
||||
w.Tick(t.Context())
|
||||
|
||||
got, _ := store.Get(t.Context(), sw.ID)
|
||||
if got.Status != SwapStatusBridgeTransferPending {
|
||||
t.Errorf("status = %q, want bridge_transfer_pending — deposit confirmed should win over expiry", got.Status)
|
||||
}
|
||||
if w.Stats().Expired != 0 {
|
||||
t.Errorf("Expired = %d, want 0 — swap advanced before expiry could fire", w.Stats().Expired)
|
||||
}
|
||||
if w.Stats().Advances != 1 {
|
||||
t.Errorf("Advances = %d, want 1", w.Stats().Advances)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWatcher_Expire_StatusGuardIdempotent verifies a second tick after
|
||||
// expiry doesn't double-count (the cancelled swap is no longer in the
|
||||
// list query, but verify by direct call to maybeExpire).
|
||||
func TestWatcher_Expire_StatusGuardIdempotent(t *testing.T) {
|
||||
store := NewInMemoryStore()
|
||||
w := NewDepositWatcher(store, newFakeChecker(), time.Second, nil)
|
||||
w.SetExpireAfter(1 * time.Hour)
|
||||
|
||||
sw := seedAgedPendingSwap(t, store, 2*time.Hour, nil)
|
||||
w.Tick(t.Context())
|
||||
if w.Stats().Expired != 1 {
|
||||
t.Fatalf("first tick: Expired = %d, want 1", w.Stats().Expired)
|
||||
}
|
||||
// Direct call again — status guard inside the Patch func should make
|
||||
// this a no-op since the swap is now Cancelled.
|
||||
got, _ := store.Get(t.Context(), sw.ID)
|
||||
w.maybeExpire(t.Context(), got, time.Now().UTC())
|
||||
if w.Stats().Expired != 1 {
|
||||
t.Errorf("after direct call: Expired = %d, want still 1", w.Stats().Expired)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWatcher_Expire_ZeroCreatedAtSkipped covers the defensive path —
|
||||
// a legacy persisted swap with CreatedAt unset shouldn't be expired
|
||||
// (we can't compute its age).
|
||||
func TestWatcher_Expire_ZeroCreatedAtSkipped(t *testing.T) {
|
||||
store := NewInMemoryStore()
|
||||
w := NewDepositWatcher(store, newFakeChecker(), time.Second, nil)
|
||||
w.SetExpireAfter(1 * time.Hour)
|
||||
|
||||
sw := &Swap{
|
||||
Status: SwapStatusUserDepositPending,
|
||||
Amount: 0.5,
|
||||
SourceNetwork: "ETHEREUM_SEPOLIA",
|
||||
SourceAsset: "ETH",
|
||||
DestinationNetwork: "LUX_TESTNET",
|
||||
DestinationAsset: "LUX",
|
||||
DestinationAddress: "0xdest",
|
||||
DepositAddress: "bridge-test-wallet###0xdeposit",
|
||||
UseDepositAddress: true,
|
||||
}
|
||||
if err := store.Create(t.Context(), sw); err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
// Directly hand maybeExpire a synthetic swap with zero CreatedAt
|
||||
// to assert the defensive guard.
|
||||
syntheticAged := *sw
|
||||
syntheticAged.CreatedAt = time.Time{}
|
||||
w.maybeExpire(t.Context(), &syntheticAged, time.Now().UTC())
|
||||
if w.Stats().Expired != 0 {
|
||||
t.Errorf("Expired = %d, want 0 — zero CreatedAt should be skipped", w.Stats().Expired)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWatcher_Expire_SetClampsNegative(t *testing.T) {
|
||||
w := NewDepositWatcher(NewInMemoryStore(), newFakeChecker(), time.Second, nil)
|
||||
w.SetExpireAfter(-5 * time.Minute)
|
||||
if w.expireAfter != 0 {
|
||||
t.Errorf("expireAfter = %v, want 0 (clamped from negative)", w.expireAfter)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/zip"
|
||||
luxlog "github.com/luxfi/log"
|
||||
)
|
||||
|
||||
// rpc_proxy.go — same-origin JSON-RPC reverse proxy for the embedded SPA.
|
||||
//
|
||||
// Motivation: the public Lux RPC gateway at api.lux.network /
|
||||
// api.lux-test.network has a strict CORS allow-list — bridge.lux.network
|
||||
// is not in it, so wagmi's useBalance() never resolves on Lux chains
|
||||
// (the browser blocks every response). Proxying through the bridge
|
||||
// backend (same-origin) sidesteps the allow-list without changing
|
||||
// gateway config. The bridge already speaks to these RPCs server-side
|
||||
// for the deposit watcher, so the network path is identical; only the
|
||||
// browser-vs-bridge actor changes.
|
||||
//
|
||||
// Properties of the proxy:
|
||||
// - POST only (JSON-RPC is POST-only over HTTP).
|
||||
// - Body forwarded verbatim. No introspection / no rewriting of the
|
||||
// JSON-RPC envelope — anything that works directly against the
|
||||
// upstream works here.
|
||||
// - Upstream timeout is bounded so a slow Lux node can't hold the
|
||||
// browser tab's wagmi loop hostage.
|
||||
// - No auth header is forwarded — the upstream is a public RPC.
|
||||
// - Read-only by intent: even though POST is allowed, the upstream
|
||||
// enforces what's safe (eth_call, eth_getBalance, etc. are
|
||||
// read-only; eth_sendRawTransaction relays signed bytes that
|
||||
// anyone could broadcast directly). Adding a method allow-list
|
||||
// here would just duplicate gateway policy.
|
||||
|
||||
// DefaultRPCProxyTimeout bounds each upstream request. 12s is generous
|
||||
// enough for a slow Lux node + Cloudflare edge while still well under
|
||||
// browser-side fetch timeouts.
|
||||
const DefaultRPCProxyTimeout = 12 * time.Second
|
||||
|
||||
// rpcProxy builds a zip handler that forwards POST bodies to upstreamURL
|
||||
// verbatim. Returns nil when upstreamURL is empty (caller MUST handle
|
||||
// nil and skip route registration in that case).
|
||||
//
|
||||
// Logger may be nil; when set, failures (transport errors, non-2xx
|
||||
// upstream responses) are logged at Warn so operators see CORS-proxy
|
||||
// degradation as early as the same-origin gateway degradation would be.
|
||||
func rpcProxy(upstreamURL string, timeout time.Duration, logger luxlog.Logger) zip.Handler {
|
||||
if upstreamURL == "" {
|
||||
return nil
|
||||
}
|
||||
if timeout <= 0 {
|
||||
timeout = DefaultRPCProxyTimeout
|
||||
}
|
||||
client := &http.Client{Timeout: timeout}
|
||||
|
||||
return func(c *zip.Ctx) error {
|
||||
// c.Body() is the raw POST body from fasthttp. Build a stdlib
|
||||
// request to forward via http.Client (gives us HTTP/2 to the
|
||||
// upstream + connection pooling that fasthttp wouldn't share
|
||||
// with the framework's inbound side).
|
||||
body := c.Body()
|
||||
req, err := http.NewRequestWithContext(c.Context(), http.MethodPost, upstreamURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return c.JSON(http.StatusInternalServerError, map[string]string{
|
||||
"error": "rpc_proxy_build_request_failed",
|
||||
"detail": err.Error(),
|
||||
})
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
if logger != nil {
|
||||
logger.Warn("rpc proxy upstream error",
|
||||
"upstream", upstreamURL,
|
||||
"err", err,
|
||||
)
|
||||
}
|
||||
return c.JSON(http.StatusBadGateway, map[string]string{
|
||||
"error": "rpc_proxy_upstream_unreachable",
|
||||
"detail": err.Error(),
|
||||
})
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return c.JSON(http.StatusBadGateway, map[string]string{
|
||||
"error": "rpc_proxy_read_upstream_body_failed",
|
||||
"detail": err.Error(),
|
||||
})
|
||||
}
|
||||
// Always return JSON content-type — every upstream we care
|
||||
// about (Lux gateway, Erigon, geth) sends JSON for JSON-RPC.
|
||||
c.SetHeader("Content-Type", "application/json")
|
||||
return c.Bytes(resp.StatusCode, respBody)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user