webui4frpc/internal/cluster/token_transport.go

// Token transport: moves the Token between nodes over HTTP POST
// /api/manager/cluster/token (Basic Auth, same creds). Also used as the
// heartbeat ping (nil token) between a predecessor and the leader.
package cluster

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"
)

// TokenTransport delivers tokens to a peer's token endpoint.
type TokenTransport struct {
	User string
	Pass string
	// BaseURL: http://host:port (scheme+authority) used to reach peers.
	BaseURL string
}

// SendTo returns a send func bound to this transport: it POSTs the token to
// next (a node ID) using that node's recorded address from the ring state.
// If tk is nil it is a heartbeat ping (no body). Returns error on failure.
func (t *TokenTransport) SendTo(getAddr func(nodeID string) string) func(ctx context.Context, next string, tk *Token) error {
	return func(ctx context.Context, next string, tk *Token) error {
		addr := getAddr(next)
		if addr == "" {
			// fall back to the treated-as-address peer id
			addr = next
		}
		url := fmt.Sprintf("http://%s/api/manager/cluster/token", addr)
		var body []byte
		var err error
		if tk != nil {
			body, err = json.Marshal(tk)
			if err != nil {
				return err
			}
		}
		req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
		if err != nil {
			return err
		}
		req.Header.Set("Content-Type", "application/json")
		if t.User != "" || t.Pass != "" {
			req.SetBasicAuth(t.User, t.Pass)
		}
		cli := &http.Client{Timeout: 5 * time.Second}
		start := time.Now()
		resp, err := cli.Do(req)
		if err != nil {
			log.Printf("token-send %s: size=%dB elapsed=%v err=%v", addr, len(body), time.Since(start).Round(time.Millisecond), err)
			return err
		}
		defer resp.Body.Close()
		log.Printf("token-send %s: size=%dB elapsed=%v -> %d", addr, len(body), time.Since(start).Round(time.Millisecond), resp.StatusCode)
		if resp.StatusCode >= 400 {
			return fmt.Errorf("token POST %s -> HTTP %d", url, resp.StatusCode)
		}
		return nil
	}
}

// HeartbeatPing is a lightweight alive check WITHOUT body (nil token); it is
// what the predecessor sends the leader. Response 2xx means alive.
func (t *TokenTransport) HeartbeatPing(ctx context.Context, addr string) error {
	url := fmt.Sprintf("http://%s/api/manager/cluster/token", addr)
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	if t.User != "" || t.Pass != "" {
		req.SetBasicAuth(t.User, t.Pass)
	}
	cli := &http.Client{Timeout: 1500 * time.Millisecond}
	resp, err := cli.Do(req)
	if err != nil {
		return err
	}
	_ = resp.Body.Close()
	if resp.StatusCode >= 400 {
		return fmt.Errorf("heartbeat HTTP %d", resp.StatusCode)
	}
	return nil
}