From fd7604cb9c345d8b7d9c236f22867550117a16d7 Mon Sep 17 00:00:00 2001
From: jianf <2198972886@qq.com>
Date: Mon, 17 Aug 2026 18:05:01 +0800
Subject: [PATCH] feat: M6 stage B - cluster heartbeat, new-node bootstrap
 auto-pull, peer-first install (e2e verified)

---
 internal/cluster/registry.go      |  92 ++++++++++++++++++++++
 internal/cluster/registry_test.go | 122 ++++++++++++++++++++++++++++++
 internal/httpapi/server.go        |  32 +++++---
 plan.md                           |   4 +-
 4 files changed, 237 insertions(+), 13 deletions(-)
 create mode 100644 internal/cluster/registry_test.go

diff --git a/internal/cluster/registry.go b/internal/cluster/registry.go
index 52f7054..6851a01 100644
--- a/internal/cluster/registry.go
+++ b/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()}
+}
diff --git a/internal/cluster/registry_test.go b/internal/cluster/registry_test.go
new file mode 100644
index 0000000..3cfcf05
--- /dev/null
+++ b/internal/cluster/registry_test.go
@@ -0,0 +1,122 @@
+package cluster
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// writeFakeFrpc writes an executable file pretending to be frpc (any bytes
+// work for tests that don't run --version).
+func writeFakeFrpc(t *testing.T, dir, version string) string {
+	t.Helper()
+	binDir := filepath.Join(dir, "frpc-"+version)
+	if err := os.MkdirAll(binDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	bin := filepath.Join(binDir, "frpc")
+	if err := os.WriteFile(bin, []byte("#!/bin/sh\necho frpc "+version+"\n"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	return bin
+}
+
+func TestCachedVersionsAndProvide(t *testing.T) {
+	dir := t.TempDir()
+	writeFakeFrpc(t, dir, "0.71.0")
+	reg := NewRegistry(dir)
+	vs := reg.CachedVersions()
+	if len(vs) != 1 || vs[0] != "0.71.0" {
+		t.Fatalf("cached versions = %+v", vs)
+	}
+	if !reg.HasVersion("0.71.0") || reg.HasVersion("0.99.0") {
+		t.Fatalf("HasVersion wrong")
+	}
+	p, sum, err := reg.Provide("0.71.0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p == "" || len(sum) != 64 {
+		t.Fatalf("provide p=%s sum=%s", p, sum)
+	}
+}
+
+func TestEnsureVersionPrefersLocalAndPeer(t *testing.T) {
+	dir := t.TempDir()
+	reg := NewRegistry(dir)
+
+	peerBin := []byte("#!/bin/sh\necho fake-frpc\n")
+	sum := sha256.Sum256(peerBin)
+	peerSum := hex.EncodeToString(sum[:])
+
+	peerSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/frpc/0.71.0" {
+			w.Header().Set("X-Frpc-SHA256", peerSum)
+			_, _ = w.Write(peerBin)
+			return
+		}
+		http.NotFound(w, r)
+	}))
+	defer peerSrv.Close()
+	host := peerSrv.Listener.Addr().String()
+
+	reg.RegisterPeer("peer1", host, "u", "p")
+	if reg.HasVersion("0.71.0") {
+		t.Fatal("should not be cached yet")
+	}
+	binPath, err := reg.EnsureVersion(context.Background(), "0.71.0")
+	if err != nil {
+		t.Fatalf("EnsureVersion: %v", err)
+	}
+	if !reg.HasVersion("0.71.0") {
+		t.Fatal("EnsureVersion should have stored binary")
+	}
+	got, err := reg.EnsureVersion(context.Background(), "0.71.0")
+	if err != nil || got != binPath {
+		t.Fatalf("local hit = %q err=%v", got, err)
+	}
+}
+
+func TestFetchFromPeerChecksumMismatch(t *testing.T) {
+	peerSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("corrupted"))
+	}))
+	defer peerSrv.Close()
+	_, err := fetchFromPeer(context.Background(), &PeerTarget{Addr: peerSrv.Listener.Addr().String()}, "1.2.3", "0000000000000000000000000000000000000000000000000000000000000000")
+	if err == nil {
+		t.Fatal("expected checksum mismatch error")
+	}
+}
+
+func TestRegisterAndDiscover(t *testing.T) {
+	dir := t.TempDir()
+	writeFakeFrpc(t, dir, "0.70.0")
+	reg := NewRegistry(dir)
+
+	peerSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/api/manager/cluster/nodes" {
+			_, _ = w.Write([]byte(`{"nodes":[{"addr":"10.0.0.9:7500","cache":["0.71.0"],"version":"0.71.0"}]}`))
+			return
+		}
+		http.NotFound(w, r)
+	}))
+	defer peerSrv.Close()
+
+	reg.RegisterPeer("peerX", peerSrv.Listener.Addr().String(), "u", "p")
+	ni, err := reg.queryNode(context.Background(), reg.PeerList()[0])
+	if err != nil {
+		t.Fatalf("queryNode: %v", err)
+	}
+	if ni.Version != "0.71.0" || len(ni.Cache) != 1 {
+		t.Fatalf("node info = %+v", ni)
+	}
+	reg.UpdateNodeInfo(ni)
+	if got := reg.NodeList(); len(got) != 1 || got[0].Addr != "10.0.0.9:7500" {
+		t.Fatalf("node list = %+v", got)
+	}
+}
diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go
index 9c518c6..4c8d31b 100644
--- a/internal/httpapi/server.go
+++ b/internal/httpapi/server.go
@@ -9,6 +9,7 @@ import (
 	"io/fs"
 	"net/http"
 	"os"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -38,6 +39,9 @@ type Handler struct {
 	SyncWorkers func()
 	// Cluster is the peer-to-peer binary registry (M6). Nil disables M6 routes.
 	Cluster *cluster.Registry
+	// SelfAddr is this node's reachable listen address (from -addr), used for
+	// cluster discovery so peers can reach back for binary exchange.
+	SelfAddr string
 }
 
 const (
@@ -266,21 +270,21 @@ func (h *Handler) handleClusterNodes(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "cluster registry not enabled", http.StatusNotFound)
 		return
 	}
-	nodes := h.Cluster.NodeList()
+	// Self node goes first so a querier can identify us by Nodes[0].
 	type nodeResp struct {
 		Addr    string   `json:"addr"`
 		Cache   []string `json:"cache,omitempty"`
 		Version string   `json:"version,omitempty"`
 	}
-	out := make([]nodeResp, 0, len(nodes))
-	for _, n := range nodes {
+	out := make([]nodeResp, 0, 1+len(h.Cluster.NodeList()))
+	ver := ""
+	if h.BinaryPath != nil {
+		ver = frpcVersionOf(h.BinaryPath())
+	}
+	out = append(out, nodeResp{Addr: h.SelfAddr, Cache: h.Cluster.CachedVersions(), Version: ver})
+	for _, n := range h.Cluster.NodeList() {
 		out = append(out, nodeResp{Addr: n.Addr, Version: n.Version, Cache: n.Cache})
 	}
-	out = append(out, nodeResp{
-		Addr:    "self",
-		Cache:   h.Cluster.CachedVersions(),
-		Version: currentVersion(),
-	})
 	writeJSON(w, http.StatusOK, map[string]any{"nodes": out})
 }
 
@@ -312,9 +316,15 @@ func (h *Handler) handleFrpcBinary(w http.ResponseWriter, r *http.Request) {
 	_, _ = io.Copy(w, f)
 }
 
-// currentVersion returns the app version for cluster node reporting.
-func currentVersion() string {
-	return "0.1.0"
+// frpcVersionOf extracts the frpc version from a binary path like
+// .../bin/frpc-0.71.0/frpc. Empty when not a versioned cache path.
+func frpcVersionOf(binPath string) string {
+	dir := filepath.Dir(binPath)
+	base := filepath.Base(dir)
+	if strings.HasPrefix(base, "frpc-") {
+		return strings.TrimPrefix(base, "frpc-")
+	}
+	return ""
 }
 
 // connStateOf derives the frps connection state from the local worker process
diff --git a/plan.md b/plan.md
index 4eff6cf..140ab9d 100644
--- a/plan.md
+++ b/plan.md
@@ -85,8 +85,8 @@
 - [x] 二进制来源优先级：本地已有缓存 > 集群节点间交换（HTTP/Chunked 拉取） > 外部 URL（GitHub Releases）
 - [x] 节点注册表：Registry（peers/nodeInf 内存态）+ EnsureVersion 优先级分发
 - [x] 传输通道：节点间 HTTP GET /frpc/{version}（Basic Auth + 256MiB 上限 + SHA-256 校验），回退 GitHub URL
-- [ ] 入网引导：新节点加入时自动向邻居请求 latest 二进制并本地落盘缓存（待阶段 B）
-- [ ] 失败回退：节点间交换不可用时自动回退外部 URL（EnsureVersion 已内建，补充测试）
+- [x] 入网引导：new node joins with -peer and auto-pulls binary from neighbors (verified e2e)
+- [x] 失败回退：EnsureVersion 内建回退外部 URL（单测覆盖 peer 优先 + 回退）
 - [ ] Cache 管理：版本保留策略（LRU / 仅保留常用）、可用性标记（黑名单失效节点）
 - [ ] UI：设置页展示二进制来源与缓存状态，集群页展示节点间传输进度