diff --git a/internal/cluster/ring_engine.go b/internal/cluster/ring_engine.go index 03bd4e1..821d327 100644 --- a/internal/cluster/ring_engine.go +++ b/internal/cluster/ring_engine.go @@ -6,6 +6,8 @@ import ( "context" "log" "time" + + "webui4frpc/internal/store" ) // Phase constants for the two-round cycle. @@ -159,7 +161,10 @@ func (e *Engine) phase2(ctx context.Context, tk *Token) error { } e.state.AddTopology(claimed, e.ID) } - e.state = tk.State + // Publish our updated state back into the token so the next node carries + // the fresh topology + pending set (do NOT revert local state to the + // incoming snapshot — that would discard the claim we just made). + tk.State = e.state tk.Passed = append(tk.Passed, e.ID) return nil } @@ -359,6 +364,12 @@ func (e *Engine) AdoptState(s State) { } } +// SubmitTask adds a new forward request to pending; it rides the next token +// round and is claimed by the lowest-load member. +func (e *Engine) SubmitTask(local store.Local, remote store.Remote, link store.Link) *Task { + return e.state.AddPending(local, remote, link) +} + // IsLeader reports whether this node is the current ring leader. func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID } diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 781b811..e531cbb 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -97,6 +97,7 @@ func NewServeMux(h *Handler) (http.Handler, error) { mux.HandleFunc(apiPrefix+"/cluster/token", auth(h.handleClusterToken)) mux.HandleFunc(apiPrefix+"/cluster/ring", auth(h.handleClusterRing)) mux.HandleFunc(apiPrefix+"/cluster/join", auth(h.handleClusterJoin)) + mux.HandleFunc(apiPrefix+"/cluster/task", auth(h.handleClusterTask)) // M6: peer-to-peer binary exchange endpoint (Basic Auth, same creds). // Not under /api so peers hit it directly; auth still applied. @@ -442,7 +443,33 @@ func (h *Handler) handleClusterJoin(w http.ResponseWriter, r *http.Request) { } } -// frpcVersionOf extracts the frpc version from a binary path like +// handleClusterTask accepts a new forward request (intermediate config: +// local/remote/link) and submits it to the ring as a pending task. +func (h *Handler) handleClusterTask(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + methodNotAllowed(w) + return + } + if h.Ring == nil { + http.Error(w, "ring engine not enabled", http.StatusNotFound) + return + } + var req struct { + Local store.Local `json:"local"` + Remote store.Remote `json:"remote"` + Link store.Link `json:"link"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "parse task: "+err.Error(), http.StatusBadRequest) + return + } + tk := h.Ring.SubmitTask(req.Local, req.Remote, req.Link) + writeJSON(w, http.StatusOK, map[string]any{"task": tk}) +} + +// handleClusterJoin accepts a newcomer join request: the target node inserts +// the newcomer after itself in the ring and returns the updated ring state +// for the newcomer to adopt. // .../bin/frpc-0.71.0/frpc. Empty when not a versioned cache path. func frpcVersionOf(binPath string) string { dir := filepath.Dir(binPath)