diff --git a/modules/dora/module.go b/modules/dora/module.go index 8df86bc4..1f1fa589 100644 --- a/modules/dora/module.go +++ b/modules/dora/module.go @@ -98,6 +98,7 @@ func (m *Module) PythonAPIDocs() map[string]types.ModuleDoc { "list_networks": {Signature: "list_networks() -> list[dict]", Description: "List networks with Dora explorers"}, "get_base_url": {Signature: "get_base_url(network) -> str", Description: "Get Dora base URL for a network"}, "get_network_overview": {Signature: "get_network_overview(network) -> dict", Description: "Get current and finalized epoch, slot, finality status (finalizing), validator counts, and participation when Dora provides it"}, + "get_clients": {Signature: "get_clients(network) -> list[dict]", Description: "List client nodes tracked by Dora; client_name is the ethnode instance label, status is online/offline"}, "get_validator": {Signature: "get_validator(network, index_or_pubkey) -> dict", Description: "Get validator by index or pubkey"}, "get_validators": {Signature: "get_validators(network, status=None, limit=100) -> list", Description: "List validators with optional filter"}, "get_slot": {Signature: "get_slot(network, slot_or_hash) -> dict", Description: "Get slot by number or hash"}, diff --git a/modules/dora/python/dora.py b/modules/dora/python/dora.py index 3d1dfe26..54db6a60 100644 --- a/modules/dora/python/dora.py +++ b/modules/dora/python/dora.py @@ -34,6 +34,19 @@ def get_network_overview(network: str) -> dict[str, Any]: return _runtime.invoke_data("dora.get_network_overview", {"network": network}) +def get_clients(network: str) -> list[dict[str, Any]]: + """List client nodes connected to the network's Dora explorer. + + Each entry's client_name is the per-node instance label (e.g. + 'lighthouse-geth-1') accepted by ethnode calls; status distinguishes + online from offline nodes. + """ + _require_dora_available() + data = _runtime.invoke_data("dora.get_clients", {"network": network}) + clients = data.get("clients", []) + return clients if isinstance(clients, list) else [] + + def get_validator(network: str, index_or_pubkey: str) -> dict[str, Any]: _require_dora_available() payload = _runtime.invoke_json( diff --git a/modules/ethnode/module.go b/modules/ethnode/module.go index 2bb879e4..3022355e 100644 --- a/modules/ethnode/module.go +++ b/modules/ethnode/module.go @@ -98,6 +98,7 @@ func (m *Module) PythonAPIDocs() map[string]types.ModuleDoc { // Discovery functions. "list_datasources": {Signature: "list_datasources() -> list[dict]", Description: "List available ethnode datasources"}, "list_networks": {Signature: "list_networks() -> list[dict]", Description: "List active network ids reachable for direct node access. Use instance='lb' for the load-balanced endpoint."}, + "list_instances": {Signature: "list_instances(network) -> list[dict]", Description: "List per-node instance labels (name, status) for a network via its Dora explorer; pass name as the instance argument of other ethnode calls"}, // Beacon node (CL) functions. "get_node_version": {Signature: "get_node_version(network, instance) -> dict", Description: "Get beacon node software version"}, "get_node_syncing": {Signature: "get_node_syncing(network, instance) -> dict", Description: "Get beacon node sync status"}, diff --git a/modules/ethnode/python/ethnode.py b/modules/ethnode/python/ethnode.py index aec81f86..ff993369 100644 --- a/modules/ethnode/python/ethnode.py +++ b/modules/ethnode/python/ethnode.py @@ -29,8 +29,8 @@ def list_networks() -> list[dict[str, Any]]: """List active network ids reachable for direct node access. Entries include at least name and type. The name is the network id accepted - by ethnode calls. The per-node instance label cannot be enumerated and must - be supplied by the caller. Use instance="lb" for the load-balanced endpoint. + by ethnode calls. Use list_instances(network) to enumerate per-node instance + labels, or instance="lb" for the load-balanced endpoint. """ _require_ethnode_available() data = _runtime.invoke_data("ethnode.list_networks") or {} @@ -38,6 +38,33 @@ def list_networks() -> list[dict[str, Any]]: return networks if isinstance(networks, list) else [] +def list_instances(network: str) -> list[dict[str, Any]]: + """List per-node instance labels for a network. + + Entries have name (the instance argument accepted by other ethnode calls) + and status (online/offline), sourced from the network's Dora explorer, + which tracks every node. instance="lb" (load-balanced) also always works. + """ + _require_ethnode_available() + from ethpandaops import dora as _dora + + try: + clients = _dora.get_clients(network) + except Exception as err: + raise ValueError( + f"cannot enumerate instances for {network!r}: this reads the " + f"network's Dora explorer, which is unavailable ({err}). " + f'Use instance="lb" for the load-balanced endpoint.' + ) from err + + instances = [ + {"name": c.get("client_name", ""), "status": c.get("status", "")} + for c in clients + if c.get("client_name") + ] + return sorted(instances, key=lambda entry: entry["name"]) + + def beacon_get( network: str, instance: str, diff --git a/pkg/cli/build.go b/pkg/cli/build.go index 51c59f0f..8278eb4d 100644 --- a/pkg/cli/build.go +++ b/pkg/cli/build.go @@ -105,7 +105,18 @@ func runBuild(cmd *cobra.Command, args []string) error { } if isJSON() { - return printJSON(result) + if err := printJSON(result); err != nil { + return err + } + + // Mirror the text path's exit status: a failed or cancelled build must + // be detectable from the process, not only by reading the payload. + switch result.Conclusion { + case "failure", "cancelled": + return &exitCodeError{code: 1} + } + + return nil } switch result.Conclusion { diff --git a/pkg/cli/clickhouse_cmd.go b/pkg/cli/clickhouse_cmd.go index 44089b74..01fb21f3 100644 --- a/pkg/cli/clickhouse_cmd.go +++ b/pkg/cli/clickhouse_cmd.go @@ -86,6 +86,9 @@ var clickhouseQueryRawCmd = &cobra.Command{ Short: "Execute a SQL query and return raw rows (always JSON)", Long: `Execute a SQL query and return raw rows as JSON. +Failures are also JSON on stdout ({"error": ...}, exit code 1), so piping into +a JSON parser surfaces the real error instead of a parse failure. + Keep result sets bounded: aggregate in SQL or add a LIMIT when inspecting rows. For cross-source analysis, run separate bounded queries and combine them with 'panda execute' or another client-side step instead of dumping unbounded rows diff --git a/pkg/cli/execute.go b/pkg/cli/execute.go index 92fdf1d5..a9eef97e 100644 --- a/pkg/cli/execute.go +++ b/pkg/cli/execute.go @@ -61,7 +61,7 @@ Examples: func init() { rootCmd.AddCommand(executeCmd) - executeCmd.Flags().StringVar(&executeCode, "code", "", "Python code to execute") + executeCmd.Flags().StringVarP(&executeCode, "code", "c", "", "Python code to execute") executeCmd.Flags().StringVar(&executeFile, "file", "", "Path to Python file to execute") executeCmd.Flags().IntVar(&executeTimeout, "timeout", 0, "Execution timeout in seconds (default: from config)") executeCmd.Flags().StringVar(&executeSession, "session", "", "Session ID to reuse") @@ -88,7 +88,17 @@ func runExecute(cmd *cobra.Command, _ []string) error { } if isJSON() { - return printJSON(result) + if err := printJSON(result); err != nil { + return err + } + + // Mirror the text path's exit status: a failed execution must be + // detectable from the process, not only by parsing exit_code. + if result.ExitCode != 0 { + return &exitCodeError{code: result.ExitCode} + } + + return nil } // Print stdout to stdout, stderr to stderr. diff --git a/pkg/cli/machine_output_test.go b/pkg/cli/machine_output_test.go new file mode 100644 index 00000000..29afb67f --- /dev/null +++ b/pkg/cli/machine_output_test.go @@ -0,0 +1,37 @@ +package cli + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// machineOutputFor mirrors the PersistentPreRunE decision so the rule can be +// exercised without executing a command. +func machineOutputFor(t *testing.T, format, commandName string) bool { + t.Helper() + + previous := outputFormat + t.Cleanup(func() { outputFormat = previous }) + + outputFormat = format + + return isJSON() || alwaysJSONCommands[commandName] +} + +func TestMachineOutputCoversEveryJSONSurface(t *testing.T) { + // The failure this guards is a JSON consumer merging stderr into the pipe: + // it breaks on any command rendering JSON, not just the always-JSON ones. + assert.True(t, machineOutputFor(t, "json", "query"), + "any command in JSON output mode is a machine surface") + assert.True(t, machineOutputFor(t, "text", "query-raw"), + "query-raw emits JSON regardless of --output") + assert.False(t, machineOutputFor(t, "text", "query"), + "human text output keeps cobra's stderr error line") +} + +func TestQueryRawStaysAnAlwaysJSONSurface(t *testing.T) { + // Guards the contract its help text advertises. + require.True(t, alwaysJSONCommands[clickhouseQueryRawCmd.Name()]) +} diff --git a/pkg/cli/resources_test.go b/pkg/cli/resources_test.go index 6842e64d..1895ee44 100644 --- a/pkg/cli/resources_test.go +++ b/pkg/cli/resources_test.go @@ -106,6 +106,21 @@ func TestServerErrorHintUsesExistingResourcesCommand(t *testing.T) { assert.NotContains(t, hint, "panda resources list") } +func TestServerErrorHintClassifiesJSONRPCErrors(t *testing.T) { + // Method-not-found is a permanent capability gap of the client; the hint + // must steer away from the gateway-status "retry" framing, whatever the + // HTTP status an old or new server wrapped it in. + for _, status := range []int{http.StatusBadRequest, http.StatusBadGateway} { + hint := serverErrorHint(status, "JSON-RPC error -32601: the method debug_chainConfig does not exist/is not available") + assert.Contains(t, hint, "capability gap") + assert.NotContains(t, hint, "temporarily unreachable") + } + + hint := serverErrorHint(http.StatusBadRequest, "JSON-RPC error -32000: execution reverted (data: 0x08c379a0)") + assert.Contains(t, hint, "rejected it") + assert.Contains(t, hint, "(data: ...)") +} + func TestServerErrorHintClassifiesClickHouseErrors(t *testing.T) { tests := []struct { name string diff --git a/pkg/cli/root.go b/pkg/cli/root.go index d4f0cd8c..26b0c4c9 100644 --- a/pkg/cli/root.go +++ b/pkg/cli/root.go @@ -8,6 +8,7 @@ import ( "os" "os/signal" "strings" + "sync" "syscall" "time" @@ -65,6 +66,19 @@ var skipUpdateCheckCommands = map[string]bool{ "help": true, } +// alwaysJSONCommands emit JSON regardless of --output, so they are machine +// surfaces even without the flag. +var alwaysJSONCommands = map[string]bool{ + "query-raw": true, +} + +// machineOutput reports whether this run's stdout is a JSON document destined +// for a parser. In that mode the CLI owns error rendering: stdout carries +// exactly one JSON document, stderr stays empty, and the exit code conveys +// failure — consumers routinely merge stderr into the pipe, where an +// "Error:" line or an update notice would corrupt the parse. +var machineOutput bool + var rootCmd = &cobra.Command{ Use: "panda", Short: "Ethereum network analytics CLI", @@ -85,6 +99,14 @@ var rootCmd = &cobra.Command{ outputFormat = "json" } + machineOutput = isJSON() || alwaysJSONCommands[cmd.Name()] + if machineOutput { + // Execute renders the error as JSON on stdout instead. Reached + // via cmd, not rootCmd, which cannot be named inside its own + // initializer. + cmd.Root().SilenceErrors = true + } + if shouldCheckForUpdate(cmd) { go backgroundUpdateCheck() } @@ -92,7 +114,7 @@ var rootCmd = &cobra.Command{ return nil }, PersistentPostRunE: func(cmd *cobra.Command, _ []string) error { - if !shouldCheckForUpdate(cmd) { + if machineOutput || !shouldCheckForUpdate(cmd) { return nil } @@ -111,13 +133,23 @@ func Execute() { return } + var exitErr *exitCodeError + isExitCode := errors.As(err, &exitErr) + + // In machine mode cobra's stderr line is suppressed, so render the failure + // here. An exitCodeError means the command already wrote its own JSON + // payload and is only reporting the outcome — a second document would + // break the parse it is meant to protect, so the exit code speaks alone. + if machineOutput && !isExitCode { + _ = printJSON(map[string]any{"error": err.Error()}) + } + if hint := unknownCommandHint(err); hint != "" { fmt.Fprintln(os.Stderr) fmt.Fprintln(os.Stderr, hint) } - var exitErr *exitCodeError - if errors.As(err, &exitErr) { + if isExitCode { os.Exit(exitErr.code) } @@ -130,9 +162,35 @@ func executeWithSignals() error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + // Once: wrapping is not idempotent, and every command is registered by + // now (subcommand init() order is not guaranteed relative to this file's). + attachUsageOnce.Do(func() { attachUsageToArgErrors(rootCmd) }) + return rootCmd.ExecuteContext(ctx) } +var attachUsageOnce sync.Once + +// attachUsageToArgErrors wraps every command's positional-arg validator so a +// count/shape error carries the command's usage line. SilenceUsage hides the +// usage block on runtime errors, which is right for HTTP failures but leaves +// "accepts 1 arg(s), received 0" with no path to the correct invocation. +func attachUsageToArgErrors(cmd *cobra.Command) { + if validate := cmd.Args; validate != nil { + cmd.Args = func(c *cobra.Command, args []string) error { + if err := validate(c, args); err != nil { + return fmt.Errorf("%w\n\nUsage: %s", err, c.UseLine()) + } + + return nil + } + } + + for _, sub := range cmd.Commands() { + attachUsageToArgErrors(sub) + } +} + func unknownCommandHint(err error) string { if err == nil || !strings.Contains(err.Error(), "unknown command") { return "" @@ -144,6 +202,16 @@ func unknownCommandHint(err error) string { func init() { rootCmd.SetVersionTemplate("panda version {{.Version}}\n") + // SilenceUsage suppresses the usage block everywhere, so a flag typo + // would otherwise surface as a bare "unknown flag" with no correction + // path. Attach the one-line usage to the error itself. + rootCmd.SetFlagErrorFunc(func(cmd *cobra.Command, err error) error { + return fmt.Errorf( + "%w\n\nUsage: %s\nRun '%s --help' for available flags", + err, cmd.UseLine(), cmd.CommandPath(), + ) + }) + rootCmd.AddGroup( &cobra.Group{ID: groupWorkflow, Title: "Workflow:"}, &cobra.Group{ID: groupDiscovery, Title: "Discovery:"}, diff --git a/pkg/cli/serverclient.go b/pkg/cli/serverclient.go index 73e58b7f..9c3c6c7f 100644 --- a/pkg/cli/serverclient.go +++ b/pkg/cli/serverclient.go @@ -295,6 +295,25 @@ func serverErrorHint(status int, message string) string { return "the first Prometheus argument must be a live datasource name; list them with 'panda prometheus list-datasources' or 'panda datasources --type prometheus'" } + if strings.Contains(normalized, "embedding query") || strings.Contains(normalized, "calling proxy embed") { + return "semantic search could not reach its embedding upstream; the query is fine — retry shortly, and if it persists check proxy connectivity with 'panda auth status'" + } + + if strings.Contains(normalized, "index not ready") || strings.Contains(normalized, "search index not available") { + return "the semantic search index is not serving yet (still warming after server start, or never activated because the proxy advertises no embedding model); the query is fine — retry, then check 'panda datasources'" + } + + // A JSON-RPC error rode back on a healthy node connection: the node + // processed and rejected the request, so the gateway-status hint below + // ("retry") would misread a deterministic failure as transient. + if strings.Contains(message, "JSON-RPC error -32601") { + return "the node answered but does not expose this RPC method; that is a capability gap of this client, not an outage — try another instance or client instead of retrying" + } + + if strings.Contains(message, "JSON-RPC error ") { + return "the node processed the request and rejected it; adjust the method, params, or request scope rather than retrying unchanged — any (data: ...) payload carries the node's reason (e.g. a revert)" + } + switch status { case http.StatusNotFound: return "the requested module, operation, datasource, or resource is not available on this server; check 'panda datasources' and 'panda resources'" diff --git a/pkg/embedding/remote.go b/pkg/embedding/remote.go index 8b15f6fd..c709a69a 100644 --- a/pkg/embedding/remote.go +++ b/pkg/embedding/remote.go @@ -17,7 +17,17 @@ import ( ) const ( - remoteEmbedTimeout = 2 * time.Minute + // remoteEmbedTimeout bounds index-build embed calls. It outlasts the + // proxy's 2-minute per-sub-batch upstream timeout, so for a request the + // proxy serves in one upstream call its specific error surfaces here + // rather than a generic client timeout. + remoteEmbedTimeout = 130 * time.Second + // queryEmbedTimeout bounds the interactive search query embed. One short + // query embeds in well under a second; inheriting the build timeout meant + // a hung upstream blocked every search for two minutes (observed + // 2026-07-08). Build paths keep the longer budget — item count is not the + // signal, since a warm rebuild legitimately embeds a single document. + queryEmbedTimeout = 15 * time.Second // maxBatchSize limits how many items are sent in a single embedding request. // The proxy accepts up to 500 items and sub-batches to the upstream API internally. maxBatchSize = 500 @@ -114,9 +124,12 @@ func NewRemote( protocol Protocol, ) *RemoteEmbedder { return &RemoteEmbedder{ - log: log.WithField("component", "remote-embedder"), - proxyURL: proxyURL, - httpClient: &http.Client{Timeout: remoteEmbedTimeout}, + log: log.WithField("component", "remote-embedder"), + proxyURL: proxyURL, + // Per-call deadlines are set via request contexts in doWithAuthRetry; + // the client stays unbounded so a query deadline is never silently + // stretched to the batch one. + httpClient: &http.Client{}, tokenFn: tokenFn, invalidateFn: invalidateFn, localCache: localCache, @@ -143,7 +156,7 @@ func (e *RemoteEmbedder) OnProgress(fn func(completed, total int)) { // Embed returns the L2-normalized QUERY embedding vector for a single // search-query string. func (e *RemoteEmbedder) Embed(text string) ([]float32, error) { - vectors, err := e.embedAll([]string{text}, e.queryTask()) + vectors, err := e.embedAll([]string{text}, e.queryTask(), queryEmbedTimeout) if err != nil { return nil, err } @@ -157,13 +170,13 @@ func (e *RemoteEmbedder) Embed(text string) ([]float32, error) { // EmbedBatch returns L2-normalized DOCUMENT embedding vectors for multiple texts. func (e *RemoteEmbedder) EmbedBatch(texts []string) ([][]float32, error) { - return e.embedAll(texts, e.documentTask()) + return e.embedAll(texts, e.documentTask(), remoteEmbedTimeout) } // EmbedQueryBatch returns L2-normalized QUERY embedding vectors for multiple // query-shaped texts. v1/v2 are symmetric, so this is equivalent to EmbedBatch. func (e *RemoteEmbedder) EmbedQueryBatch(texts []string) ([][]float32, error) { - return e.embedAll(texts, e.queryTask()) + return e.embedAll(texts, e.queryTask(), remoteEmbedTimeout) } func (e *RemoteEmbedder) queryTask() string { @@ -186,7 +199,7 @@ func (e *RemoteEmbedder) documentTask() string { // vectors are checked there first. Remaining misses are split into sub-batches // of maxBatchSize and sent to the proxy (which has its own Redis cache + // upstream API). -func (e *RemoteEmbedder) embedAll(texts []string, task string) ([][]float32, error) { +func (e *RemoteEmbedder) embedAll(texts []string, task string, timeout time.Duration) ([][]float32, error) { if len(texts) == 0 { return nil, nil } @@ -215,7 +228,7 @@ func (e *RemoteEmbedder) embedAll(texts []string, task string) ([][]float32, err } } - vecs, err := e.embedDirect(texts, hashes, hashToIndices, task) + vecs, err := e.embedDirect(texts, hashes, hashToIndices, task, timeout) if err != nil { return nil, err } @@ -341,7 +354,7 @@ func (e *RemoteEmbedder) embedAll(texts []string, task string) ([][]float32, err "misses": len(missItems), }).Info("Proxy cache stats") - resp, err := e.callEmbed(missItems, task) + resp, err := e.callEmbed(missItems, task, timeout) if err != nil { return nil, fmt.Errorf("embedding batch %d/%d: %w", batchNum, totalBatches, err) } @@ -447,13 +460,14 @@ func (e *RemoteEmbedder) embedDirect( hashes []string, hashToIndices map[string][]int, task string, + timeout time.Duration, ) ([][]float32, error) { items := make([]embedItem, len(texts)) for i, text := range texts { items[i] = embedItem{Hash: hashes[i], Text: text} } - resp, err := e.callEmbed(items, task) + resp, err := e.callEmbed(items, task, timeout) if err != nil { return nil, err } @@ -475,14 +489,19 @@ func (e *RemoteEmbedder) embedDirect( } // doWithAuthRetry POSTs jsonBody to the proxy path with the current auth token, -// retrying once after invalidating the token on a 401/403. -func (e *RemoteEmbedder) doWithAuthRetry(path string, jsonBody []byte) (*http.Response, error) { - send := func() (*http.Response, error) { +// retrying once after invalidating the token on a 401/403. It reads the whole +// response body so the per-call deadline covers the body too and the context +// can be released before returning. +func (e *RemoteEmbedder) doWithAuthRetry(path string, jsonBody []byte, timeout time.Duration) (int, []byte, error) { + send := func() (int, []byte, error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + req, err := http.NewRequestWithContext( - context.Background(), http.MethodPost, e.proxyURL+path, bytes.NewReader(jsonBody), + ctx, http.MethodPost, e.proxyURL+path, bytes.NewReader(jsonBody), ) if err != nil { - return nil, err + return 0, nil, err } req.Header.Set("Content-Type", "application/json") @@ -493,22 +512,32 @@ func (e *RemoteEmbedder) doWithAuthRetry(path string, jsonBody []byte) (*http.Re } } - return e.httpClient.Do(req) + resp, err := e.httpClient.Do(req) + if err != nil { + return 0, nil, err + } + defer func() { _ = resp.Body.Close() }() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return 0, nil, err + } + + return resp.StatusCode, body, nil } - resp, err := send() + status, body, err := send() if err != nil { - return nil, err + return 0, nil, err } - if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && e.invalidateFn != nil { - _ = resp.Body.Close() + if (status == http.StatusUnauthorized || status == http.StatusForbidden) && e.invalidateFn != nil { e.invalidateFn() return send() } - return resp, nil + return status, body, nil } // verifyEmbeddingSpace fails fast when the proxy's advertised embedding space @@ -555,20 +584,17 @@ func (e *RemoteEmbedder) checkCached(hashes []string, task string) ([]embedResul return nil, fmt.Errorf("marshaling check request: %w", err) } - resp, err := e.doWithAuthRetry(e.checkPath(), reqBody) + status, body, err := e.doWithAuthRetry(e.checkPath(), reqBody, remoteEmbedTimeout) if err != nil { return nil, fmt.Errorf("calling embed check: %w", err) } - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(resp.Body) - - return nil, fmt.Errorf("embed check returned status %d: %s", resp.StatusCode, string(body)) + if status != http.StatusOK { + return nil, fmt.Errorf("embed check returned status %d: %s", status, string(body)) } var checkResp embedCheckResponse - if err := json.NewDecoder(resp.Body).Decode(&checkResp); err != nil { + if err := json.Unmarshal(body, &checkResp); err != nil { return nil, fmt.Errorf("decoding check response: %w", err) } @@ -579,7 +605,7 @@ func (e *RemoteEmbedder) checkCached(hashes []string, task string) ([]embedResul return checkResp.Cached, nil } -func (e *RemoteEmbedder) callEmbed(items []embedItem, task string) (*embedResponse, error) { +func (e *RemoteEmbedder) callEmbed(items []embedItem, task string, timeout time.Duration) (*embedResponse, error) { req := embedRequest{Items: items} if e.protocol == ProtocolV3 { req.Task = task @@ -590,20 +616,17 @@ func (e *RemoteEmbedder) callEmbed(items []embedItem, task string) (*embedRespon return nil, fmt.Errorf("marshaling embed request: %w", err) } - resp, err := e.doWithAuthRetry(e.embedPath(), reqBody) + status, body, err := e.doWithAuthRetry(e.embedPath(), reqBody, timeout) if err != nil { return nil, fmt.Errorf("calling proxy embed: %w", err) } - defer func() { _ = resp.Body.Close() }() - - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(resp.Body) - return nil, fmt.Errorf("proxy embed returned status %d: %s", resp.StatusCode, string(body)) + if status != http.StatusOK { + return nil, fmt.Errorf("proxy embed returned status %d: %s", status, string(body)) } var embedResp embedResponse - if err := json.NewDecoder(resp.Body).Decode(&embedResp); err != nil { + if err := json.Unmarshal(body, &embedResp); err != nil { return nil, fmt.Errorf("decoding embed response: %w", err) } diff --git a/pkg/embedding/remote_test.go b/pkg/embedding/remote_test.go index 75fca919..0cf51f8d 100644 --- a/pkg/embedding/remote_test.go +++ b/pkg/embedding/remote_test.go @@ -9,6 +9,7 @@ import ( "net/http/httptest" "sync/atomic" "testing" + "time" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" @@ -626,3 +627,48 @@ func TestRemoteEmbedder_V1IgnoresSpaceEcho(t *testing.T) { require.NoError(t, err) assert.Equal(t, []float32{1, 0}, vec) } + +// TestRemoteEmbedder_HonorsPerCallDeadline pins the timeout routing: each +// embed path is bounded by the deadline it was handed, so an index build that +// happens to embed exactly one document (the warm-rebuild case) is not +// clamped to the much tighter interactive-query deadline. +func TestRemoteEmbedder_HonorsPerCallDeadline(t *testing.T) { + t.Parallel() + + const upstreamLatency = 300 * time.Millisecond + + srv := newMockProxy(t, func(w http.ResponseWriter, r *http.Request) { + var req embedRequest + require.NoError(t, json.NewDecoder(r.Body).Decode(&req)) + + time.Sleep(upstreamLatency) + + results := make([]embedResult, 0, len(req.Items)) + for _, item := range req.Items { + results = append(results, embedResult{Hash: item.Hash, Vector: []float32{1}}) + } + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(embedResponse{Dimensions: 8, Results: results}) + }, nil) + + embedder := NewRemote(logrus.New(), srv.URL, func() string { return "" }, nil, nil, "", 8, ProtocolV3) + + assert.Less(t, queryEmbedTimeout, remoteEmbedTimeout, + "an interactive query must be bounded more tightly than an index build") + + // A single-document build with a build-sized budget completes, even though + // it is one item — item count must not select the deadline. + vecs, err := embedder.embedAll([]string{"one document"}, taskDocument, 10*upstreamLatency) + require.NoError(t, err) + require.Len(t, vecs, 1) + + // The caller's deadline is authoritative even for a one-item request: if + // the deadline were re-derived from item count, this single-item call + // would silently get the query budget and succeed instead of expiring. + start := time.Now() + _, err = embedder.embedAll([]string{"one document"}, taskDocument, upstreamLatency/6) + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.Less(t, time.Since(start), upstreamLatency) +} diff --git a/pkg/server/api.go b/pkg/server/api.go index 7e9b138b..5d0e19e6 100644 --- a/pkg/server/api.go +++ b/pkg/server/api.go @@ -174,6 +174,26 @@ func (s *service) handleAPIDatasources(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, serverapi.DatasourcesResponse{Datasources: all}) } +// searchErrorStatus classifies a search-service failure. An unreachable +// embedding upstream, a still-warming index, or an index that never activated +// is a retryable service condition, not a caller error — 400 would pin the +// failure on the query. Filter-value rejections ("unknown category/tag/...") +// stay 400, since only the caller can fix those. +func searchErrorStatus(err error) int { + message := err.Error() + for _, serviceCondition := range []string{ + "embedding query", // embedding upstream unreachable + "index not ready", // index still building after start + "search index not available", // search runtime never activated + } { + if strings.Contains(message, serviceCondition) { + return http.StatusServiceUnavailable + } + } + + return http.StatusBadRequest +} + func (s *service) handleAPISearchExamples(w http.ResponseWriter, r *http.Request) { if s.searchService == nil { writeAPIError(w, http.StatusServiceUnavailable, "search service is unavailable") @@ -194,7 +214,7 @@ func (s *service) handleAPISearchExamples(w http.ResponseWriter, r *http.Request resp, err := s.searchService.SearchExamples(query, r.URL.Query().Get("category"), r.URL.Query().Get("dataset"), limit) if err != nil { - writeAPIError(w, http.StatusBadRequest, err.Error()) + writeAPIError(w, searchErrorStatus(err), err.Error()) return } @@ -221,7 +241,7 @@ func (s *service) handleAPISearchRunbooks(w http.ResponseWriter, r *http.Request resp, err := s.searchService.SearchRunbooks(query, r.URL.Query().Get("tag"), limit) if err != nil { - writeAPIError(w, http.StatusBadRequest, err.Error()) + writeAPIError(w, searchErrorStatus(err), err.Error()) return } @@ -250,7 +270,7 @@ func (s *service) handleAPISearchEIPs(w http.ResponseWriter, r *http.Request) { resp, err := s.searchService.SearchEIPs(query, q.Get("status"), q.Get("category"), q.Get("type"), limit) if err != nil { - writeAPIError(w, http.StatusBadRequest, err.Error()) + writeAPIError(w, searchErrorStatus(err), err.Error()) return } @@ -277,7 +297,7 @@ func (s *service) handleAPISearchSpecs(w http.ResponseWriter, r *http.Request) { resp, err := s.searchService.SearchSpecs(query, r.URL.Query().Get("fork"), limit) if err != nil { - writeAPIError(w, http.StatusBadRequest, err.Error()) + writeAPIError(w, searchErrorStatus(err), err.Error()) return } diff --git a/pkg/server/api_test.go b/pkg/server/api_test.go new file mode 100644 index 00000000..fb5dbd7d --- /dev/null +++ b/pkg/server/api_test.go @@ -0,0 +1,65 @@ +package server + +import ( + "errors" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestSearchErrorStatusClassifiesServiceConditions(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + want int + }{ + { + name: "embed upstream unreachable is a service condition", + err: errors.New(`search failed: embedding query: calling proxy embed: Post "https://proxy/embed": context deadline exceeded`), + want: http.StatusServiceUnavailable, + }, + { + name: "warming index is a service condition", + err: errors.New("search failed: example index not ready"), + want: http.StatusServiceUnavailable, + }, + { + name: "other search failures stay caller errors", + err: errors.New("search failed: unknown category"), + want: http.StatusBadRequest, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, searchErrorStatus(tt.err)) + }) + } +} + +func TestSearchErrorStatusCoversInactiveIndexes(t *testing.T) { + t.Parallel() + + // searchsvc reports a never-activated index differently from a warming + // one; both are service conditions the caller cannot fix by rewording. + for _, message := range []string{ + "example search index not available", + "runbook search index not available", + "EIP search index not available", + "consensus specs search index not available", + } { + assert.Equal(t, http.StatusServiceUnavailable, searchErrorStatus(errors.New(message)), message) + } + + // Filter-value rejections remain the caller's to fix. + for _, message := range []string{ + `unknown category: "bogus". Available categories: a, b`, + `unknown tag: "bogus". Available tags: a, b`, + } { + assert.Equal(t, http.StatusBadRequest, searchErrorStatus(errors.New(message)), message) + } +} diff --git a/pkg/server/operations_dora.go b/pkg/server/operations_dora.go index 32828bed..3f835ee9 100644 --- a/pkg/server/operations_dora.go +++ b/pkg/server/operations_dora.go @@ -27,6 +27,8 @@ func (s *service) handleDoraOperation(operationID string, w http.ResponseWriter, s.handleDoraBaseURL(w, r) case "dora.get_network_overview": s.handleDoraNetworkOverview(w, r) + case "dora.get_clients": + s.handleDoraClients(w, r) case "dora.get_validator": s.handleDoraDataGetPassthrough(w, r, "index_or_pubkey", "/api/v1/validator/%s") case "dora.get_validators": @@ -97,6 +99,42 @@ func (s *service) handleDoraBaseURL(w http.ResponseWriter, r *http.Request) { }) } +// handleDoraClients returns Dora's consensus client node list. Unlike the +// data-enveloped v1 endpoints, /api/v1/clients/consensus responds with a bare +// {clients, count} object. Each client_name is the per-node instance label +// accepted by ethnode operations. +func (s *service) handleDoraClients(w http.ResponseWriter, r *http.Request) { + req, err := decodeOperationRequest(r) + if err != nil { + writeAPIError(w, http.StatusBadRequest, err.Error()) + return + } + + baseURL, status, err := s.doraBaseURL(req.Args) + if err != nil { + writeAPIError(w, status, err.Error()) + return + } + + payload, status, err := s.doraAPIGet(r.Context(), baseURL, "/api/v1/clients/consensus", nil) + if err != nil { + writeAPIError(w, status, err.Error()) + return + } + + clients, ok := payload["clients"].([]any) + if !ok { + writeAPIError(w, http.StatusBadGateway, "Dora client list is missing the clients field") + return + } + + writeOperationResponse(s.log, w, http.StatusOK, operations.Response{ + Kind: operations.ResultKindObject, + Data: map[string]any{"clients": clients, "count": len(clients)}, + Meta: map[string]any{"network": optionalStringArg(req.Args, "network")}, + }) +} + func (s *service) handleDoraNetworkOverview(w http.ResponseWriter, r *http.Request) { req, err := decodeOperationRequest(r) if err != nil { diff --git a/pkg/server/operations_dora_test.go b/pkg/server/operations_dora_test.go index bc5291fc..ff63a81a 100644 --- a/pkg/server/operations_dora_test.go +++ b/pkg/server/operations_dora_test.go @@ -144,6 +144,62 @@ func TestDoraDataPassthroughEscapesIdentifier(t *testing.T) { assert.Contains(t, rec.Body.String(), `"slot":"escaped"`) } +func TestDoraClientsReturnsBareClientList(t *testing.T) { + t.Parallel() + + doraServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v1/clients/consensus": + _, _ = w.Write([]byte(`{"clients":[ + {"client_name":"lighthouse-geth-1","client_type":"lighthouse","status":"online","head_slot":3210}, + {"client_name":"nimbus-besu-1","client_type":"nimbus","status":"offline","head_slot":0} + ],"count":2}`)) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(doraServer.Close) + + svc := newDoraOperationService(doraServer.Client(), doraServer.URL) + rec := httptest.NewRecorder() + + handled := svc.handleDoraOperation("dora.get_clients", rec, newDoraOpRequest(t, "testnet")) + require.True(t, handled) + require.Equal(t, http.StatusOK, rec.Code) + + var response operations.Response + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &response)) + + data, ok := response.Data.(map[string]any) + require.True(t, ok) + assert.Equal(t, float64(2), data["count"]) + + clients, ok := data["clients"].([]any) + require.True(t, ok) + require.Len(t, clients, 2) + + first, ok := clients[0].(map[string]any) + require.True(t, ok) + assert.Equal(t, "lighthouse-geth-1", first["client_name"]) + assert.Equal(t, "online", first["status"]) +} + +func TestDoraClientsRejectsNonJSONBody(t *testing.T) { + t.Parallel() + + doraServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`rate limited`)) + })) + t.Cleanup(doraServer.Close) + + svc := newDoraOperationService(doraServer.Client(), doraServer.URL) + rec := httptest.NewRecorder() + + handled := svc.handleDoraOperation("dora.get_clients", rec, newDoraOpRequest(t, "testnet")) + require.True(t, handled) + assert.Equal(t, http.StatusBadGateway, rec.Code) +} + func newDoraOperationService(httpClient *http.Client, doraURL string) *service { log := logrus.New() log.SetOutput(io.Discard) diff --git a/pkg/server/operations_ethnode.go b/pkg/server/operations_ethnode.go index fcb2aad8..e90cbfa0 100644 --- a/pkg/server/operations_ethnode.go +++ b/pkg/server/operations_ethnode.go @@ -18,6 +18,10 @@ import ( var ethnodeSegmentPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*[a-z0-9]$|^[a-z0-9]$`) +// maxRPCErrorDataBytes caps the JSON-RPC error `data` payload echoed into an +// error message. It is generous enough for revert reasons and short structs. +const maxRPCErrorDataBytes = 512 + func (s *service) handleEthNodeOperation(operationID string, w http.ResponseWriter, r *http.Request) bool { switch operationID { case "ethnode.list_datasources": @@ -612,16 +616,32 @@ func (s *service) ethNodeExecutionRPCRaw( var rpcResp struct { Error *struct { - Code int `json:"code"` - Message string `json:"message"` + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` } `json:"error"` } if err := json.Unmarshal(data, &rpcResp); err != nil { return nil, "", http.StatusBadGateway, fmt.Errorf("invalid JSON-RPC response: %w", err) } + // A JSON-RPC error object means the node processed and rejected the + // request (unsupported method, bad params, node-side limits, reverts) — + // not a gateway failure. 502 would misread it as transient and its data + // field often carries the answer itself (e.g. a revert reason). if rpcResp.Error != nil { - return nil, "", http.StatusBadGateway, fmt.Errorf("JSON-RPC error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message) + message := fmt.Sprintf("JSON-RPC error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message) + if data := rpcResp.Error.Data; len(data) > 0 && !bytes.Equal(data, []byte("null")) { + // Bounded: some clients return whole traces here, and an error + // message is not a data channel. + if len(data) > maxRPCErrorDataBytes { + data = append(data[:maxRPCErrorDataBytes:maxRPCErrorDataBytes], []byte("… (truncated)")...) + } + + message += fmt.Sprintf(" (data: %s)", data) + } + + return nil, "", http.StatusBadRequest, fmt.Errorf("%s", message) } contentType := responseHeaders.Get("Content-Type") diff --git a/pkg/server/operations_ethnode_test.go b/pkg/server/operations_ethnode_test.go index 201c6f84..a0b860e5 100644 --- a/pkg/server/operations_ethnode_test.go +++ b/pkg/server/operations_ethnode_test.go @@ -71,6 +71,34 @@ func TestEthNodeCuratedBeaconPathEscapesStateIdentifier(t *testing.T) { assert.Empty(t, transport.last.URL.RawQuery) } +func TestEthNodeExecutionRPCErrorIsNotAGatewayFailure(t *testing.T) { + t.Parallel() + + transport := &recordingTransport{ + body: `{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"the method debug_chainConfig does not exist/is not available","data":"0x08c379a0"}}`, + contentType: "application/json", + } + svc := newEthNodeOperationService(true) + svc.proxyService.(*ethNodeOperationProxy).url = "https://proxy.example" + svc.httpClient = &http.Client{Transport: transport} + + rec := httptest.NewRecorder() + handled := svc.handleEthNodeOperation("ethnode.execution_rpc", rec, newEthNodeOpRequestWithArgs(t, map[string]any{ + "network": "testnet", + "instance": "node-a", + "method": "debug_chainConfig", + "params": []any{}, + })) + + require.True(t, handled) + // The node answered and rejected the request; 502 would misread the + // deterministic failure as a transient gateway error. + require.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "JSON-RPC error -32601") + assert.Contains(t, rec.Body.String(), "does not exist/is not available") + assert.Contains(t, rec.Body.String(), "0x08c379a0") +} + func newEthNodeOperationService(ethnodeAvailable bool) *service { log := logrus.New() log.SetOutput(io.Discard) diff --git a/sandbox/Dockerfile b/sandbox/Dockerfile index f75c8229..2e8b18c1 100644 --- a/sandbox/Dockerfile +++ b/sandbox/Dockerfile @@ -9,7 +9,7 @@ # hard error rather than an unlocked source compile, and no C toolchain is # needed in either stage. -FROM python:3.11-slim@sha256:a3ab0b966bc4e91546a033e22093cb840908979487a9fc0e6e38295747e49ac0 AS builder +FROM python:3.12-slim@sha256:57cd7c3a7a273101a6485ba99423ee568157882804b1124b4dd04266317710de AS builder # uv (build-time only; never copied into the final image) COPY --from=ghcr.io/astral-sh/uv:0.11.17@sha256:03bdc89bb9798628846e60c3a9ad19006c8c3c724ccd2985a33145c039a0577b /uv /uvx /bin/ @@ -37,7 +37,7 @@ COPY modules/*/python/*.py /opt/ethpandaops-pkg/ethpandaops/ # so no toolchain is required to build it. RUN uv pip install --python /usr/local/bin/python3 --prefix /install --no-cache --no-deps /opt/ethpandaops-pkg -FROM python:3.11-slim@sha256:a3ab0b966bc4e91546a033e22093cb840908979487a9fc0e6e38295747e49ac0 +FROM python:3.12-slim@sha256:57cd7c3a7a273101a6485ba99423ee568157882804b1124b4dd04266317710de # librsvg provides rsvg-convert, which the chartkit library uses to rasterise its # self-contained SVG charts. The chart font (Inter) is embedded as a data URI; the extra diff --git a/sandbox/requirements.txt b/sandbox/requirements.txt index 20d2da59..13432d5a 100644 --- a/sandbox/requirements.txt +++ b/sandbox/requirements.txt @@ -1,5 +1,5 @@ # This file was autogenerated by uv via the following command: -# uv pip compile sandbox/requirements.in -o sandbox/requirements.txt --generate-hashes --universal --python-version 3.11 --prerelease=disallow +# uv pip compile sandbox/requirements.in -o sandbox/requirements.txt --generate-hashes --universal --python-version 3.12 --prerelease=disallow annotated-types==0.7.0 \ --hash=sha256:1f02e8b43a8fbbc3f3e0d4f0f4bfc8131bcb4eebe8849b8e5c773f3a1c582a53 \ --hash=sha256:aff07c09a53a08bc8cfccb9c85b05f1aa9a2a6f23728d790723543408344ce89