feat: M6 stage B - cluster heartbeat, new-node bootstrap auto-pull, peer-first install (e2e verified)

internal/cluster/registry.go

@@ -11,6 +11,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -311,3 +312,94 @@ func extractFrpcBytes(tarGz []byte) ([]byte, error) {
 		}
 	}
 }
+
+// Discover periodically polls registered peers for their cached versions and
+// refreshes the node registry (M6 stage B heartbeats). Run in a goroutine.
+func (r *Registry) Discover(ctx context.Context, interval time.Duration) {
+	tick := time.NewTicker(interval)
+	defer tick.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+			peers := r.PeerList()
+			for _, p := range peers {
+				ni, err := r.queryNode(ctx, p)
+				if err != nil {
+					continue // peer unreachable; keep last known info
+				}
+				r.UpdateNodeInfo(ni)
+			}
+		}
+	}
+}
+
+// queryNode asks one peer for its node info (cached frpc versions).
+func (r *Registry) queryNode(ctx context.Context, p *PeerTarget) (*NodeInfo, error) {
+	url := fmt.Sprintf("http://%s/api/manager/cluster/nodes", p.Addr)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+	if p.User != "" || p.Pass != "" {
+		req.SetBasicAuth(p.User, p.Pass)
+	}
+	cli := &http.Client{Timeout: 10 * time.Second}
+	resp, err := cli.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		return nil, fmt.Errorf("node query HTTP %d", resp.StatusCode)
+	}
+	var payload struct {
+		Nodes []struct {
+			Addr    string   `json:"addr"`
+			Cache   []string `json:"cache"`
+			Version string   `json:"version"`
+		} `json:"nodes"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil {
+		return nil, err
+	}
+	if len(payload.Nodes) == 0 {
+		return nil, fmt.Errorf("peer returned no nodes")
+	}
+	self := payload.Nodes[0]
+	return &NodeInfo{ID: self.Addr, Addr: self.Addr, Version: self.Version, Cache: self.Cache}, nil
+}
+
+// SyncFromPeers ensures the locally cached frpc set covers the latest version
+// each known peer offers (new-member auto propagation). It leaves version
+// moves to EnsureVersion on demand; here we just warm the cache.
+func (r *Registry) SyncFromPeers(ctx context.Context) error {
+	// First refresh node info from registered peers so the cache set reflects
+	// what neighbors actually offer (a fresh join may have no info yet).
+	for _, p := range r.PeerList() {
+		if ni, err := r.queryNode(ctx, p); err == nil {
+			r.UpdateNodeInfo(ni)
+		}
+	}
+	for _, n := range r.NodeList() {
+		if n.Version == "" {
+			continue
+		}
+		if r.HasVersion(n.Version) {
+			continue
+		}
+		// EnsureVersion internally iterates registered peers (and falls back
+		// to the external URL), so no peer selection is needed here.
+		if _, err := r.EnsureVersion(ctx, n.Version); err != nil {
+			continue
+		}
+	}
+	return nil
+}
+
+// SelfInfo returns this node's own info for reporting.
+func (r *Registry) SelfInfo() NodeInfo {
+	return NodeInfo{Addr: "self", Version: "", Cache: r.CachedVersions()}
+}