bootnode/fleet/luxd.go

package fleet

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

// Prober fills in one node's live self-report. It never fails: a node that
// does not answer is a node whose fields stay null, which is the truth the
// dashboard has to render.
type Prober interface {
	Probe(ctx context.Context, base string, n *Node)
}

// RPC talks to luxd over its /v1 HTTP surface.
//
// /v1 is the only surface luxd serves — /ext was removed, and the collector
// that still called it read 404 on every request and reported a fleet of
// zeros. Chain RPC lives under /v1/bc/<chain>, so the C-Chain is
// /v1/bc/C/rpc (never /v1/C/rpc, which is not a route).
type RPC struct{ http *http.Client }

// NewRPC returns a prober with luxd-appropriate timeouts: a node that is wedged
// must not hold the whole overview hostage.
func NewRPC() *RPC {
	return &RPC{http: &http.Client{
		Timeout:   6 * time.Second,
		Transport: &http.Transport{DialContext: (&net.Dialer{Timeout: 2500 * time.Millisecond}).DialContext},
	}}
}

// call posts one JSON-RPC request and decodes result into out.
func (r *RPC) call(ctx context.Context, base, path, method string, params, out any) error {
	body, err := json.Marshal(map[string]any{
		"jsonrpc": "2.0", "id": 1, "method": method, "params": params,
	})
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+path, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := r.http.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s %s: HTTP %d", method, path, resp.StatusCode)
	}
	var env struct {
		Result json.RawMessage `json:"result"`
		Error  *struct {
			Message string `json:"message"`
		} `json:"error"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&env); err != nil {
		return err
	}
	if env.Error != nil {
		return fmt.Errorf("%s: %s", method, env.Error.Message)
	}
	if out == nil || len(env.Result) == 0 {
		return nil
	}
	return json.Unmarshal(env.Result, out)
}

// Probe fills n from the node at base. Every call is independent and
// best-effort: a partial answer is still worth rendering.
func (r *RPC) Probe(ctx context.Context, base string, n *Node) {
	const info = "/v1/info"

	var (
		wg   sync.WaitGroup
		ver  struct{ Version, GitCommit, DatabaseVersion string }
		nid  struct{ NodeID string }
		nip  struct{ IP string }
		peer struct {
			Peers []json.RawMessage `json:"peers"`
		}
		health  *bool
		boot    = make([]*bool, len(TrackedChains))
		cHeight *int64
		pHeight *int64
		verOK   bool
		nidOK   bool
		nipOK   bool
		peerOK  bool
	)

	run := func(fn func()) { wg.Add(1); go func() { defer wg.Done(); fn() }() }

	run(func() { verOK = r.call(ctx, base, info, "info.getNodeVersion", nil, &ver) == nil })
	run(func() { nidOK = r.call(ctx, base, info, "info.getNodeID", nil, &nid) == nil })
	run(func() { nipOK = r.call(ctx, base, info, "info.getNodeIP", nil, &nip) == nil })
	run(func() { peerOK = r.call(ctx, base, info, "info.peers", map[string]any{}, &peer) == nil })
	run(func() { health = r.health(ctx, base) })

	for i, chain := range TrackedChains {
		run(func() {
			var res struct {
				IsBootstrapped bool `json:"isBootstrapped"`
			}
			if r.call(ctx, base, info, "info.isBootstrapped", map[string]any{"chain": chain}, &res) == nil {
				boot[i] = &res.IsBootstrapped
			}
		})
	}
	run(func() {
		var hex string
		if r.call(ctx, base, "/v1/bc/C/rpc", "eth_blockNumber", []any{}, &hex) != nil {
			return
		}
		if v, err := strconv.ParseInt(strings.TrimPrefix(hex, "0x"), 16, 64); err == nil {
			cHeight = &v
		}
	})
	run(func() {
		var res struct {
			Height numeric `json:"height"`
		}
		if r.call(ctx, base, "/v1/bc/P", "platform.getHeight", nil, &res) != nil {
			return
		}
		v := int64(res.Height)
		pHeight = &v
	})
	wg.Wait()

	if verOK && ver.Version != "" {
		n.RPCReachable = true
		n.Version = strPtr(ver.Version)
		n.GitCommit = strPtr(ver.GitCommit)
		n.DBVersion = strPtr(ver.DatabaseVersion)
	}
	if nidOK {
		n.NodeID = strPtr(nid.NodeID)
	}
	if nipOK {
		n.IP = strPtr(hostOnly(nip.IP))
	}
	if peerOK {
		n.Peers = ptr(len(peer.Peers))
	}
	n.Healthy = health

	for i, chain := range TrackedChains {
		ch := ChainHeight{Chain: chain, Bootstrapped: boot[i]}
		switch chain {
		case "C":
			ch.Height = cHeight
		case "P":
			ch.Height = pHeight
		}
		n.Chains[i] = ch
	}
}

// health reads luxd's own health verdict. Some builds fail to encode the full
// reply and answer 503 with a valid body, so 503 is read, not discarded; any
// other outcome is unknown rather than unhealthy.
func (r *RPC) health(ctx context.Context, base string) *bool {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+"/v1/health", nil)
	if err != nil {
		return nil
	}
	resp, err := r.http.Do(req)
	if err != nil {
		return nil
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusServiceUnavailable {
		return nil
	}
	var res struct {
		Healthy *bool `json:"healthy"`
	}
	if json.NewDecoder(resp.Body).Decode(&res) != nil {
		return nil
	}
	return res.Healthy
}

func hostOnly(addr string) string {
	if host, _, err := net.SplitHostPort(addr); err == nil {
		return host
	}
	return addr
}

// numeric accepts a JSON number or a decimal string — luxd quotes some
// integers (platform.getHeight) and not others.
type numeric int64

func (n *numeric) UnmarshalJSON(b []byte) error {
	v, err := strconv.ParseInt(string(bytes.Trim(b, `"`)), 10, 64)
	if err != nil {
		return err
	}
	*n = numeric(v)
	return nil
}