Improved Sources and max_data_age

app.go

@@ -174,7 +174,7 @@ func NewApp(
 		}
 	}()
 
-	tokenBridgeClient := tokenbridgeclient.StartNewClient(ctx, logger, tokenDepositsCache, tokenBridgeTipsCache)
+	tokenBridgeClient := tokenbridgeclient.StartNewClient(ctx, logger, tokenDepositsCache, tokenBridgeTipsCache, chainId)
 	appInstance.TokenBridgeClient = tokenBridgeClient
 
 	// Start the Metrics Daemon.

cmd/chain_id.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	rpchttp "github.com/cometbft/cometbft/rpc/client/http"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/cosmos/cosmos-sdk/client/grpc/cmtservice"
+)
+
+// detectChainID queries both the gRPC endpoint and the CometBFT RPC endpoint
+// for the chain ID, validates that they agree, and returns the chain ID.
+// Returns an error if either endpoint is unreachable or if they return
+// different chain IDs.
+func detectChainID(ctx context.Context, grpcAddr, nodeRPCAddr string) (string, error) {
+	detectCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
+	defer cancel()
+
+	grpcChainID, err := chainIDFromGRPC(detectCtx, grpcAddr)
+	if err != nil {
+		return "", fmt.Errorf("failed to detect chain ID via gRPC (%s): %w", grpcAddr, err)
+	}
+
+	rpcChainID, err := chainIDFromRPC(detectCtx, nodeRPCAddr)
+	if err != nil {
+		return "", fmt.Errorf("failed to detect chain ID via node RPC (%s): %w", nodeRPCAddr, err)
+	}
+
+	if grpcChainID != rpcChainID {
+		return "", fmt.Errorf("chain ID mismatch: gRPC returned %q, node RPC returned %q", grpcChainID, rpcChainID)
+	}
+
+	return grpcChainID, nil
+}
+
+func chainIDFromGRPC(ctx context.Context, grpcAddr string) (string, error) {
+	conn, err := grpc.DialContext(ctx, grpcAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return "", fmt.Errorf("dial: %w", err)
+	}
+	defer conn.Close()
+
+	resp, err := cmtservice.NewServiceClient(conn).GetNodeInfo(ctx, &cmtservice.GetNodeInfoRequest{})
+	if err != nil {
+		return "", fmt.Errorf("GetNodeInfo: %w", err)
+	}
+
+	return resp.DefaultNodeInfo.Network, nil
+}
+
+func chainIDFromRPC(ctx context.Context, nodeRPCAddr string) (string, error) {
+	rpcClient, err := rpchttp.New(nodeRPCAddr, "/websocket")
+	if err != nil {
+		return "", fmt.Errorf("create client: %w", err)
+	}
+
+	status, err := rpcClient.Status(ctx)
+	if err != nil {
+		return "", fmt.Errorf("status: %w", err)
+	}
+
+	return status.NodeInfo.Network, nil
+}

cmd/main.go

@@ -45,23 +45,22 @@ var rootCmd = &cobra.Command{
 
 		// Check if test mode is enabled
 		if testMode {
-			if err := runTestMode(homePath, logger); err != nil {
+			if err := runTestMode(homePath, logger, testQueryID); err != nil {
 				fmt.Printf("Test mode failed: %v\n", err)
 				os.Exit(1)
 			}
 			os.Exit(0)
 		}
+		if testQueryID != "" {
+			fmt.Fprintf(os.Stderr, "Error: --test-query-id requires --test\n")
+			os.Exit(1)
+		}
 
 		// Normal daemon mode - validate required flags
-		chainId := viper.GetString(flags.FlagChainID)
 		grpcAddr := viper.GetString(flags.FlagGRPC)
 		from := viper.GetString(flags.FlagFrom)
 		node := viper.GetString(flags.FlagNode)
 
-		if chainId == "" {
-			fmt.Printf("Error: --chain-id is required in reporter mode\n")
-			os.Exit(1)
-		}
 		if grpcAddr == "" {
 			fmt.Printf("Error: --grpc is required in reporter mode\n")
 			os.Exit(1)
@@ -79,6 +78,13 @@ var rootCmd = &cobra.Command{
 		ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 		defer stop()
 
+		chainId, err := detectChainID(ctx, grpcAddr, node)
+		if err != nil {
+			fmt.Printf("Error: could not detect chain ID: %v\n", err)
+			os.Exit(1)
+		}
+		logger.Info("Detected chain ID", "chain_id", chainId)
+
 		// Pass prometheusPort and signal context to NewApp
 		appInstance := daemons.NewApp(ctx, logger, chainId, grpcAddr, homePath, prometheusPort)
 
@@ -94,6 +100,7 @@ var rootCmd = &cobra.Command{
 var (
 	prometheusPort int
 	testMode       bool
+	testQueryID    string
 )
 
 func main() {
@@ -108,7 +115,6 @@ func init() {
 	rootCmd.Flags().String(flags.FlagHome, appconfig.DefaultNodeHome, "Node home directory")
 	rootCmd.Flags().String(flags.FlagFrom, "", "Name of the key to use")
 	rootCmd.Flags().String(flags.FlagGRPC, "0.0.0.0:9090", "Address to listen on")
-	rootCmd.Flags().String(flags.FlagChainID, "layer", "Chain ID")
 	rootCmd.Flags().String(flags.FlagKeyringBackend, flags.DefaultKeyringBackend, "Select keyring's backend (os|file|kwallet|pass|test|memory)")
 	rootCmd.Flags().String(flags.FlagLogLevel, zerolog.InfoLevel.String(), "The logging level (trace|debug|info|warn|error|fatal|panic|disabled or '*:<level>,<key>:<level>')")
 	rootCmd.Flags().String(flags.FlagBroadcastMode, flags.BroadcastSync, "Transaction broadcasting mode (sync|async)")
@@ -123,6 +129,7 @@ func init() {
 
 	// Test mode flag
 	rootCmd.Flags().BoolVar(&testMode, "test", false, "Test mode: verify price feed configurations and calculate medians without starting daemon")
+	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.")
 	// Automatic Unbonding flags
 	rootCmd.Flags().Uint32("auto-unbonding-frequency", 0, "Enable automatic unbonding every N days (0 = disabled, 1 - 21 days = valid")
 	rootCmd.Flags().Uint32("auto-unbonding-amount", 0, "Amount of tokens in loya to unbond each unbonding transaction (0 = disabled)")
@@ -133,7 +140,7 @@ func init() {
 	if err := rootCmd.MarkFlagRequired(flags.FlagHome); err != nil {
 		panic(err)
 	}
-	// Note: --from, --grpc, --chain-id, and --node are only required in normal mode, not test mode
+	// Note: --from, --grpc, and --node are only required in normal mode, not test mode
 	// We'll validate them in the Run function instead
 
 	// Try to load .env from current directory, or parent directory if not found.

cmd/test_mode.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/tellor-io/layer-daemons/configs"
@@ -14,6 +15,7 @@ import (
 	libtime "github.com/tellor-io/layer-daemons/lib/time"
 	handler "github.com/tellor-io/layer-daemons/pricefeed/client/queryhandler"
 	"github.com/tellor-io/layer-daemons/pricefeed/client/types"
+	serverdaemon "github.com/tellor-io/layer-daemons/server/types/daemons"
 	pricefeedservertypes "github.com/tellor-io/layer-daemons/server/types/pricefeed"
 	daemontypes "github.com/tellor-io/layer-daemons/types"
 
@@ -27,10 +29,31 @@ type exchangeTestResult struct {
 	Error   string
 }
 
-// runTestMode loads all price feed configurations and tests them
-func runTestMode(homePath string, logger log.Logger) error {
+// runTestMode loads all price feed configurations and tests them.
+// If isolatedQueryID is non-empty, only that custom query is run (no exchange/market tests).
+func runTestMode(homePath string, logger log.Logger, isolatedQueryID string) error {
 	logger.Info("Starting test mode - verifying price feed configurations")
 
+	if isolatedQueryID != "" {
+		logger.Info("Isolated custom query test (--test-query-id); skipping exchange/market tests")
+		customQueries, err := customquery.BuildQueryEndpoints(homePath, "config", "custom_query_config.toml")
+		if err != nil {
+			return fmt.Errorf("load custom queries: %w", err)
+		}
+		id := strings.ToLower(strings.TrimSpace(isolatedQueryID))
+		qc, ok := customQueries[id]
+		if !ok {
+			return fmt.Errorf("custom query id not found in config: %s", id)
+		}
+		priceCache := pricefeedservertypes.NewMarketToExchangePrices(5 * time.Minute)
+		populateTestModePriceCacheFromExchanges(homePath, priceCache, logger)
+		if err := testCustomQuery(id, qc, priceCache, logger); err != nil {
+			return fmt.Errorf("custom query test failed: %w", err)
+		}
+		logger.Info("Isolated custom query test succeeded", "query_id", id)
+		return nil
+	}
+
 	// Load configurations
 	logger.Info("Loading market parameters...")
 	marketParams := configs.ReadMarketParamsConfigFile(homePath)
@@ -61,8 +84,10 @@ func runTestMode(homePath string, logger log.Logger) error {
 	// Test custom queries
 	if len(customQueries) > 0 {
 		logger.Info("Testing custom queries...")
+		priceCache := pricefeedservertypes.NewMarketToExchangePrices(5 * time.Minute)
+		populateTestModePriceCacheFromExchanges(homePath, priceCache, logger)
 		for queryId, queryConfig := range customQueries {
-			if err := testCustomQuery(queryId, queryConfig, logger); err != nil {
+			if err := testCustomQuery(queryId, queryConfig, priceCache, logger); err != nil {
 				logger.Error("Failed to test custom query", "query_id", queryId, "error", err)
 			}
 		}
@@ -270,17 +295,17 @@ func queryExchangeForMarket(
 	}
 }
 
-// testCustomQuery tests a single custom query configuration
-func testCustomQuery(queryId string, queryConfig customquery.QueryConfig, logger log.Logger) error {
+// testCustomQuery tests a single custom query configuration.
+// priceCache must already be populated (e.g. populateTestModePriceCacheFromExchanges) so handlers
+// that resolve USD via other markets use live exchange medians instead of synthetic values.
+func testCustomQuery(queryId string, queryConfig customquery.QueryConfig, priceCache *pricefeedservertypes.MarketToExchangePrices, logger log.Logger) error {
 	logger.Info("Testing custom query", "query_id", queryId)
 
-	// Create an empty price cache for custom queries that may need it
-	priceCache := pricefeedservertypes.NewMarketToExchangePrices(5 * time.Minute)
-
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
 	results, err := customquery.FetchPrice(ctx, queryConfig, priceCache)
+	logCustomQuerySourceResults(logger, queryId, results)
 	if err != nil {
 		logger.Warn("  ✗ Custom query failed",
 			"query_id", queryId,
@@ -292,7 +317,127 @@ func testCustomQuery(queryId string, queryConfig customquery.QueryConfig, logger
 	logger.Info("  ✓ Custom query succeeded",
 		"query_id", queryId,
 		"encoded_value", results.EncodedValue,
+		"success_rate", results.SuccessRate,
 	)
 
 	return nil
 }
+
+// logCustomQuerySourceResults prints one line per configured endpoint (contract, RPC, or combined).
+// On failure, FetchPrice may still return results with RawResults populated for diagnostics.
+func logCustomQuerySourceResults(logger log.Logger, queryId string, fr *customquery.FetchPriceResult) {
+	if fr == nil || len(fr.RawResults) == 0 {
+		logger.Info("custom query per-source results", "query_id", queryId, "count", 0, "note", "no per-endpoint results")
+		return
+	}
+	for i, r := range fr.RawResults {
+		if r.Err != nil {
+			logger.Info("custom query per-source result",
+				"query_id", queryId,
+				"source_index", i,
+				"source_id", r.SourceId,
+				"endpoint_id", r.EndpointID,
+				"market_id", r.MarketId,
+				"ok", false,
+				"error", r.Err.Error(),
+			)
+			continue
+		}
+		logger.Info("custom query per-source result",
+			"query_id", queryId,
+			"source_index", i,
+			"source_id", r.SourceId,
+			"endpoint_id", r.EndpointID,
+			"market_id", r.MarketId,
+			"ok", true,
+			"price", r.Value,
+		)
+	}
+}
+
+// populateTestModePriceCacheFromExchanges fills the cache with the same per-exchange prices used
+// in market-param tests, so custom-query handlers that call GetValidMedianPrices see real sources.
+func populateTestModePriceCacheFromExchanges(
+	homePath string,
+	cache *pricefeedservertypes.MarketToExchangePrices,
+	logger log.Logger,
+) {
+	marketParams := configs.ReadMarketParamsConfigFile(homePath)
+	exchangeConfigs := configs.ReadExchangeQueryConfigFile(homePath)
+	var updates []*serverdaemon.MarketPriceUpdate
+	for i := range marketParams {
+		u := marketPriceUpdateFromLiveExchanges(&marketParams[i], exchangeConfigs, logger)
+		if u != nil {
+			updates = append(updates, u)
+		}
+	}
+	if len(updates) > 0 {
+		cache.UpdatePrices(updates)
+	}
+	if len(updates) == 0 {
+		logger.Warn("No markets had enough live exchange prices to populate --test reference cache; custom queries that need USD-via may fail")
+	} else {
+		logger.Info("Populated --test price reference cache from live exchanges",
+			"markets_with_sufficient_exchanges", len(updates),
+			"total_market_params", len(marketParams),
+		)
+	}
+}
+
+// marketPriceUpdateFromLiveExchanges returns a single-market update when at least MinExchanges
+// configured sources succeed; otherwise nil.
+func marketPriceUpdateFromLiveExchanges(
+	marketParam *types.MarketParam,
+	exchangeConfigs map[types.ExchangeId]*types.ExchangeQueryConfig,
+	logger log.Logger,
+) *serverdaemon.MarketPriceUpdate {
+	var exchangeConfigJson types.ExchangeConfigJson
+	if err := json.Unmarshal([]byte(marketParam.ExchangeConfigJson), &exchangeConfigJson); err != nil {
+		logger.Debug("Skipping market for test cache: invalid exchange config JSON",
+			"pair", marketParam.Pair, "error", err)
+		return nil
+	}
+	now := time.Now()
+	var exchangePrices []*serverdaemon.ExchangePrice
+	for _, exchangeConfigJsonItem := range exchangeConfigJson.Exchanges {
+		exchangeId := exchangeConfigJsonItem.ExchangeName
+		exchangeDetails, exists := constants.StaticExchangeDetails[exchangeId]
+		if !exists {
+			continue
+		}
+		exchangeQueryConfig, hasConfig := exchangeConfigs[exchangeId]
+		if !hasConfig {
+			continue
+		}
+		result := queryExchangeForMarket(
+			exchangeId,
+			exchangeDetails,
+			*exchangeQueryConfig,
+			*marketParam,
+			exchangeConfigJsonItem,
+			logger,
+		)
+		if !result.Success {
+			continue
+		}
+		t := now
+		exchangePrices = append(exchangePrices, &serverdaemon.ExchangePrice{
+			ExchangeId:     exchangeId,
+			Price:          result.Price,
+			LastUpdateTime: &t,
+		})
+	}
+	if len(exchangePrices) < int(marketParam.MinExchanges) {
+		logger.Debug("Skipping market for test cache: insufficient live exchange prices",
+			"pair", marketParam.Pair,
+			"market_id", marketParam.Id,
+			"have", len(exchangePrices),
+			"need", marketParam.MinExchanges,
+		)
+		return nil
+	}
+	return &serverdaemon.MarketPriceUpdate{
+		MarketId:       marketParam.Id,
+		ExchangePrices: exchangePrices,
+	}
+}