Skip to content

Commit 4d3905a

Browse files
0xSpuddy0xSpuddy
authored
Merge pull request #28 from tellor-io/auto-bridge-prs
Auto bridge prs
2 parents ba24baa + 771dbb7 commit 4d3905a

11 files changed

Lines changed: 474 additions & 54 deletions

File tree

‎cmd/main.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -131,11 +131,16 @@ func init() {
131131
rootCmd.Flags().BoolVar(&testMode, "test", false, "Test mode: verify price feed configurations and calculate medians without starting daemon")
132132
rootCmd.Flags().StringVar(&testQueryID, "test-query-id", "", "With --test, only run this custom query id (64-char hex); skips exchange/market tests. Exits non-zero if the query fails.")
133133
// Automatic Unbonding flags
134-
rootCmd.Flags().Uint32("auto-unbonding-frequency", 0, "Enable automatic unbonding every N days (0 = disabled, 1 - 21 days = valid")
134+
rootCmd.Flags().Uint32("auto-unbonding-frequency", 0, "Enable automatic unbonding every N days (0 = disabled, 1 - 21 days = valid)")
135135
rootCmd.Flags().Uint32("auto-unbonding-amount", 0, "Amount of tokens in loya to unbond each unbonding transaction (0 = disabled)")
136136
rootCmd.Flags().String("auto-unbonding-max-stake-percentage", "0.0", "Maximum percentage of stake to unbond each unbonding transaction (0 = disabled, 1.0 = 100%). If unbonding amount exceeds this percentage, we will skip the unbonding transaction until it exceeds this percentage again.")
137137
rootCmd.Flags().Duration("refresh-gas-estimates-interval", 12*time.Hour, "Interval for resetting cached gas estimates and gas-adjustment levels (<=0 disables)")
138138

139+
// Auto-bridge: keep wallet at a fixed balance by bridging the excess to Ethereum
140+
rootCmd.Flags().Uint64(daemonflags.FlagAutoBalanceToKeep, 0, "Keep this amount of loya in the wallet; bridge any excess to Ethereum at --auto-balance-execution-time (0 = disabled)")
141+
rootCmd.Flags().String(daemonflags.FlagAutoBalanceExecutionTime, "00:00", "UTC time to execute the auto-balance bridge (HH:MM, e.g. '03:00')")
142+
rootCmd.Flags().String(daemonflags.FlagAutoBalanceBridgeToEthAddr, "", "Ethereum address to bridge excess tokens to (required when auto-balance-to-keep > 0)")
143+
139144
// Marking required flags
140145
if err := rootCmd.MarkFlagRequired(flags.FlagHome); err != nil {
141146
panic(err)

‎env.example‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,9 @@ ETH_RPC_URL_FALLBACK=https://eth-mainnet.g.alchemy.com/v2/YOUR_ALCHEMY_API_KEY
1717
REPORTERS_VALIDATOR_ADDRESS=tellorvaloper1...
1818
WITHDRAW_FREQUENCY=43200
1919

20+
# Auto-balance is configured with CLI flags:
21+
# --auto-balance-to-keep, --auto-balance-execution-time, --auto-balance-bridge-to-eth-addr
22+
2023
# Custom query API keys
2124
# These are used by generated custom_query_config.toml entries that reference ${VAR}.
2225
CMC_PRO_API_KEY=your_coinmarketcap_api_key

‎flags/flags.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,10 @@ const (
1717
FlagPriceDaemonLoopDelayMs = "price-daemon-loop-delay-ms"
1818

1919
FlagKeyringBackend = "keyring-backend"
20+
21+
FlagAutoBalanceToKeep = "auto-balance-to-keep"
22+
FlagAutoBalanceExecutionTime = "auto-balance-execution-time"
23+
FlagAutoBalanceBridgeToEthAddr = "auto-balance-bridge-to-eth-addr"
2024
)
2125

2226
// Shared flags contains configuration flags shared by all daemons.

‎pricefeed/client/client_test.go‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@ import (
1212

1313
"github.com/stretchr/testify/mock"
1414
"github.com/stretchr/testify/require"
15-
"github.com/tellor-io/layer-daemons/appconfig"
1615
pricefeed_constants "github.com/tellor-io/layer-daemons/constants"
1716
daemonflags "github.com/tellor-io/layer-daemons/flags"
1817
"github.com/tellor-io/layer-daemons/mocks"
@@ -23,7 +22,6 @@ import (
2322
daemonserver "github.com/tellor-io/layer-daemons/server"
2423
servertypes "github.com/tellor-io/layer-daemons/server/types/daemons"
2524
pricefeed_types "github.com/tellor-io/layer-daemons/server/types/pricefeed"
26-
"github.com/tellor-io/layer-daemons/testutil/appoptions"
2725
"github.com/tellor-io/layer-daemons/testutil/client"
2826
"github.com/tellor-io/layer-daemons/testutil/constants"
2927
daemontestutils "github.com/tellor-io/layer-daemons/testutil/daemons"
@@ -281,7 +279,6 @@ func TestStart_InvalidConfig(t *testing.T) {
281279
func TestStop(t *testing.T) {
282280
// Setup daemon and grpc servers.
283281
daemonFlags := daemonflags.GetDefaultDaemonFlags()
284-
appFlags := appconfig.GetFlagValuesFromOptions(appoptions.GetDefaultTestAppOptions("", nil))
285282

286283
// Configure and run daemon server.
287284
daemonServer := daemonserver.NewServer(
@@ -302,18 +299,21 @@ func TestStop(t *testing.T) {
302299
// pricetypes.RegisterQueryServer(grpcServer, &pricesQueryServer)
303300

304301
// Start gRPC server with cleanup.
302+
ls, err := net.Listen("tcp", "127.0.0.1:0")
303+
require.NoError(t, err)
304+
grpcAddr := ls.Addr().String()
305305
defer grpcServer.Stop()
306306
go func() {
307-
ls, err := net.Listen("tcp", appFlags.GrpcAddress)
308-
require.NoError(t, err)
309-
err = grpcServer.Serve(ls)
310-
require.NoError(t, err)
307+
if serveErr := grpcServer.Serve(ls); serveErr != nil {
308+
// Ignore error on shutdown.
309+
_ = serveErr
310+
}
311311
}()
312312

313313
client := StartNewClient(
314314
grpc_util.Ctx,
315315
daemonFlags,
316-
appFlags.GrpcAddress,
316+
grpcAddr,
317317
log.NewNopLogger(),
318318
&daemontypes.GrpcClientImpl{},
319319
[]types.MarketParam{},

‎reporter/client/auto_balance_test.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package client
2+
3+
import (
4+
"testing"
5+
6+
"github.com/stretchr/testify/require"
7+
bridgetypes "github.com/tellor-io/layer/x/bridge/types"
8+
oracletypes "github.com/tellor-io/layer/x/oracle/types"
9+
10+
"cosmossdk.io/math"
11+
)
12+
13+
func TestNormalizeAutoBalanceEthAddr(t *testing.T) {
14+
addr, err := normalizeAutoBalanceEthAddr("0x0000000000000000000000000000000000000001")
15+
require.NoError(t, err)
16+
require.Equal(t, "0000000000000000000000000000000000000001", addr)
17+
18+
addr, err = normalizeAutoBalanceEthAddr("0000000000000000000000000000000000000001")
19+
require.NoError(t, err)
20+
require.Equal(t, "0000000000000000000000000000000000000001", addr)
21+
22+
addr, err = normalizeAutoBalanceEthAddr("")
23+
require.NoError(t, err)
24+
require.Empty(t, addr)
25+
26+
_, err = normalizeAutoBalanceEthAddr("not-an-address")
27+
require.Error(t, err)
28+
}
29+
30+
func TestParseAutoBalanceExecutionTime(t *testing.T) {
31+
hour, minute, err := parseAutoBalanceExecutionTime("03:05")
32+
require.NoError(t, err)
33+
require.Equal(t, 3, hour)
34+
require.Equal(t, 5, minute)
35+
36+
_, _, err = parseAutoBalanceExecutionTime("24:00")
37+
require.Error(t, err)
38+
39+
_, _, err = parseAutoBalanceExecutionTime("03")
40+
require.Error(t, err)
41+
}
42+
43+
func TestIsBridgeDepositReportMsg(t *testing.T) {
44+
require.True(t, isBridgeDepositReportMsg(&oracletypes.MsgSubmitValue{}))
45+
require.False(t, isBridgeDepositReportMsg(&bridgetypes.MsgWithdrawTokens{}))
46+
}
47+
48+
func TestShouldSkipAutoUnbond(t *testing.T) {
49+
reporterStake := math.LegacyNewDec(1_000)
50+
unbondAmount := math.NewInt(100)
51+
tenPercent, err := math.LegacyNewDecFromStr("0.10")
52+
require.NoError(t, err)
53+
fivePercent, err := math.LegacyNewDecFromStr("0.05")
54+
require.NoError(t, err)
55+
56+
require.False(t, shouldSkipAutoUnbond(reporterStake, math.LegacyZeroDec(), unbondAmount), "zero max percentage disables the cap")
57+
require.False(t, shouldSkipAutoUnbond(reporterStake, tenPercent, unbondAmount))
58+
require.True(t, shouldSkipAutoUnbond(reporterStake, fivePercent, unbondAmount))
59+
}

‎reporter/client/broadcast_message.go‎

Lines changed: 69 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,23 @@ import (
1313
"github.com/cosmos/cosmos-sdk/telemetry"
1414
)
1515

16+
// cycle list
17+
// const (
18+
// ethQueryData = "0x00000000000000000000000000000000000000000000000000000000000000400000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000953706F745072696365000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000C0000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000000800000000000000000000000000000000000000000000000000000000000000003657468000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000037573640000000000000000000000000000000000000000000000000000000000"
19+
// btcQueryData = "0x00000000000000000000000000000000000000000000000000000000000000400000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000953706F745072696365000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000C0000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000000800000000000000000000000000000000000000000000000000000000000000003627463000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000037573640000000000000000000000000000000000000000000000000000000000"
20+
// trbQueryData = "0x00000000000000000000000000000000000000000000000000000000000000400000000000000000000000000000000000000000000000000000000000000080000000000000000000000000000000000000000000000000000000000000000953706F745072696365000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000C0000000000000000000000000000000000000000000000000000000000000004000000000000000000000000000000000000000000000000000000000000000800000000000000000000000000000000000000000000000000000000000000003747262000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000037573640000000000000000000000000000000000000000000000000000000000"
21+
// )
22+
23+
// var (
24+
// eth, _ = utils.QueryBytesFromString(ethQueryData)
25+
// btc, _ = utils.QueryBytesFromString(btcQueryData)
26+
// trb, _ = utils.QueryBytesFromString(trbQueryData)
27+
// )
28+
1629
const (
17-
bridgeDepositMaxRetries = 10 // Bridge deposits have ~1 hour window, so more retries are acceptable
30+
bridgeDepositMaxRetries = 10 // Bridge deposits have ~1 hour window, so more retries are acceptable
31+
maxConcurrentTxs = 10 // Cap concurrent broadcast goroutines to prevent unbounded growth
32+
txBroadcastTimeout = 15 * time.Second // sign + broadcast + wait for block inclusion
1833
)
1934

2035
func (c *Client) GenerateDepositMessages(ctx context.Context) error {
@@ -32,11 +47,38 @@ func (c *Client) GenerateDepositMessages(ctx context.Context) error {
3247
}
3348

3449
telemetry.IncrCounterWithLabels([]string{"daemon_bridge_deposit", "found"}, 1, []metrics.Label{{Name: "chain_id", Value: c.cosmosCtx.ChainID}})
35-
c.trySend(ctx, TxChannelInfo{Msg: msg, isBridge: true, NumRetries: bridgeDepositMaxRetries, QueryMetaId: 0})
50+
c.txChan <- TxChannelInfo{Msg: msg, isBridge: true, NumRetries: bridgeDepositMaxRetries, QueryMetaId: 0}
3651

3752
return nil
3853
}
3954

55+
// func (c *Client) generateExternalMessages(ctx context.Context, filepath string, bg *sync.WaitGroup) error {
56+
// defer bg.Done()
57+
// jsonFile, err := os.ReadFile(filepath)
58+
// if err != nil {
59+
// if errors.Is(err, os.ErrNotExist) {
60+
// return nil
61+
// }
62+
// return fmt.Errorf("error reading from file: %w", err)
63+
// }
64+
// if err := os.Remove(filepath); err != nil {
65+
// return fmt.Errorf("error deleting transactions file: %w", err)
66+
// }
67+
// tx, err := c.cosmosCtx.TxConfig.TxJSONDecoder()(jsonFile)
68+
// if err != nil {
69+
// return fmt.Errorf("error decoding json file: %w", err)
70+
// }
71+
// msgs := tx.GetMsgs()
72+
73+
// resp, err := c.sendTx(ctx, msgs...)
74+
// if err != nil {
75+
// return fmt.Errorf("error sending tx: %w", err)
76+
// }
77+
// fmt.Println("response after external message", resp.TxResult.Code)
78+
79+
// return nil
80+
// }
81+
4082
func (c *Client) GenerateAndBroadcastSpotPriceReport(ctx context.Context, qd []byte, querymeta *oracletypes.QueryMeta) error {
4183
encodedValue, rawPrice, err := c.median(qd)
4284
if err != nil {
@@ -85,12 +127,12 @@ func (c *Client) GenerateAndBroadcastSpotPriceReport(ctx context.Context, qd []b
85127
Value: encodedValue,
86128
}
87129

88-
c.trySend(ctx, TxChannelInfo{
130+
c.txChan <- TxChannelInfo{
89131
Msg: msg,
90132
isBridge: false,
91133
NumRetries: 0,
92134
QueryMetaId: querymeta.Id,
93-
})
135+
}
94136

95137
// Mark as committed immediately to prevent duplicate processing
96138
mutex.Lock()
@@ -153,32 +195,40 @@ func (c *Client) HandleBridgeDepositTxInChannel(ctx context.Context, data TxChan
153195
}
154196

155197
func (c *Client) BroadcastTxMsgToChain(ctx context.Context) {
156-
defer c.broadcastWg.Wait()
157-
198+
semaphore := make(chan struct{}, maxConcurrentTxs)
158199
for {
159200
select {
160201
case <-ctx.Done():
161-
c.logger.Debug("BroadcastTxMsgToChain: context canceled, exiting")
202+
c.logger.Debug("BroadcastTxMsgToChain: context canceled")
162203
return
163204
case obj, ok := <-c.txChan:
164205
if !ok {
165-
c.logger.Debug("BroadcastTxMsgToChain: channel closed, exiting")
206+
c.logger.Debug("BroadcastTxMsgToChain: tx channel closed")
166207
return
167208
}
168-
// submit transaction in goroutine with proper tracking
209+
169210
c.broadcastWg.Add(1)
211+
// submit transaction in goroutine without waiting for completion
170212
go func(txInfo TxChannelInfo) {
171213
defer c.broadcastWg.Done()
172-
txCtx, cancel := context.WithTimeout(ctx, 4500*time.Millisecond)
214+
215+
select {
216+
case semaphore <- struct{}{}:
217+
defer func() { <-semaphore }()
218+
case <-ctx.Done():
219+
return
220+
}
221+
222+
txCtx, cancel := context.WithTimeout(ctx, txBroadcastTimeout)
173223
defer cancel()
174224

175-
if !txInfo.isBridge {
176-
_, err := c.sendTx(txCtx, txInfo.QueryMetaId, false, txInfo.Msg)
225+
if txInfo.isBridge && isBridgeDepositReportMsg(txInfo.Msg) {
226+
c.HandleBridgeDepositTxInChannel(txCtx, txInfo)
227+
} else {
228+
_, err := c.sendTx(txCtx, txInfo.QueryMetaId, txInfo.isBridge, txInfo.Msg)
177229
if err != nil {
178230
c.logger.Error(fmt.Sprintf("Error sending tx: %v", err))
179231
}
180-
} else {
181-
c.HandleBridgeDepositTxInChannel(txCtx, txInfo)
182232
}
183233
}(obj)
184234

@@ -187,3 +237,8 @@ func (c *Client) BroadcastTxMsgToChain(ctx context.Context) {
187237
}
188238
}
189239
}
240+
241+
func isBridgeDepositReportMsg(msg interface{}) bool {
242+
_, ok := msg.(*oracletypes.MsgSubmitValue)
243+
return ok
244+
}

0 commit comments

Comments
 (0)