feat(hub): Direction-2 immediate-sync wait channel (v0.58.0)

GET /api/v1/wait long-poll: the box holds an authed hanging GET; the hub completes it the instant any operator intent bumps that customer's in-memory generation, then the box fires its ordinary report and the ACK delivers everything through the unchanged machinery. 240s hold with a 25s heartbeat newline defeats the nginx 60s proxy_read_timeout with no ingress annotation; WriteTimeout lifted per-connection via ResponseController. - internal/intent: per-customer generation counter + waiter registry (Bump/Wait/Close), coalescing to latest, race-closer, in-memory by design. Red-proofs: counter-vs-queue + race-closer (run-fail-reverted). - api/wait.go: the endpoint (per-customer only; global key 400; A cannot see B). - web bumps after every intent write (fire-after-commit): config CRUD, claim resend, offsite re-issue/freeze, password regen, block/unblock, floors (global bumps all config-managed), controller log-tail + log-bundle. - main.go: one intent hub shared by web+api; Close() before server.Shutdown. Pairs with controller v0.140.0 (the long-poll client). Grounding: documentation/audits/SPIKE-immediate-sync-transport-2026-07-16.md.
2026-07-16 20:44:22 +02:00
parent 10e07f5747
commit 60244727ad
11 changed files with 790 additions and 0 deletions
@@ -17,6 +17,7 @@ import (
 	"gitea.dooplex.hu/admin/felhom-hub/internal/assets"
 	"gitea.dooplex.hu/admin/felhom-hub/internal/claim"
 	"gitea.dooplex.hu/admin/felhom-hub/internal/configgen"
+	"gitea.dooplex.hu/admin/felhom-hub/internal/intent"
 	"gitea.dooplex.hu/admin/felhom-hub/internal/mailrelay"
 	"gitea.dooplex.hu/admin/felhom-hub/internal/notify"
 	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
@@ -72,6 +73,11 @@ type Handler struct {
 	// as the manual "Re-issue offsite credentials" button, so escrow invalidation + events ride
 	// along). nil = no auto re-issue; a no-op when offsite isn't provisioned/enabled for the customer.
 	offsiteReissuer func(ctx context.Context, customerID string) error
+
+	// intentHub (v0.58.0, Direction-2 immediate-sync) is the in-memory per-customer generation
+	// notifier that GET /api/v1/wait long-polls against. nil = wait endpoint returns 503 (the box
+	// falls back to the 15-min cycle). Shared with the web server, whose intent handlers Bump it.
+	intentHub *intent.Hub
 }

 // SetClaimEngine wires the customer-claim code engine (nil-safe everywhere it is used).
@@ -95,6 +101,12 @@ func (h *Handler) SetLatestVersionProvider(p LatestVersionProvider) {
 	h.latestVersion = p
 }

+// SetIntentHub wires the operator-intent notifier for GET /api/v1/wait (v0.58.0; nil-safe — an
+// unset hub makes the wait endpoint return 503).
+func (h *Handler) SetIntentHub(hub *intent.Hub) {
+	h.intentHub = hub
+}
+
 // New creates a new API handler.
 func New(store *store.Store, apiKey, resendAPIKey, fromEmail string, templateProvider ConfigTemplateProvider, logger *log.Logger) *Handler {
 	return &Handler{
@@ -180,6 +192,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	switch {
 	case r.Method == http.MethodPost && path == "/report":
 		h.handleReport(w, r)
+	// Direction-2 immediate-sync (v0.58.0): the box long-polls here; the hub completes it on any
+	// operator-intent bump for the box's customer, then the box fires its ordinary report.
+	case r.Method == http.MethodGet && path == "/wait":
+		h.handleWait(w, r)
 	case r.Method == http.MethodPost && path == "/host-report":
 		h.handleHostReport(w, r)
 	case r.Method == http.MethodPost && path == "/host-enroll":
@@ -0,0 +1,105 @@
+package api
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+)
+
+// waitMaxHold / waitHeartbeat are effectively constants — vars only so tests can shrink them.
+var (
+	// waitMaxHold is how long the hub holds a wait before completing it with the (unchanged)
+	// generation. Kept well under the ingress/NAT idle ceilings the transport spike measured
+	// (SPIKE-immediate-sync-transport-2026-07-16.md): the nginx 60 s proxy_read_timeout is defeated
+	// by the heartbeat below, and the operator NAT held an idle hold > 600 s. 240 s → a box
+	// reconnects roughly every 4 min in steady state (negligible fleet load).
+	waitMaxHold = 240 * time.Second
+	// waitHeartbeat is the spacing of keepalive newline bytes written while holding. nginx's
+	// proxy_read_timeout is measured BETWEEN successive reads from the upstream, so a byte every
+	// 25 s keeps the default 60 s from ever firing — no ingress annotation / manifest change is
+	// needed (§13 proves this live; the annotation is the documented fallback if it doesn't hold).
+	// 25 s mirrors the WG PersistentKeepalive cadence and doubles as NAT keepalive.
+	waitHeartbeat = 25 * time.Second
+)
+
+// handleWait is the Direction-2 immediate-sync long-poll (option b of the transport spike). A box
+// holds this authenticated GET carrying its last-seen generation (?gen=N); the hub completes it the
+// instant any operator intent lands for the box's customer (intent.Hub.Bump from the web handlers),
+// or after waitMaxHold, whichever comes first. The response is CONTENTLESS — a single {"gen":N}
+// line preceded by heartbeat newlines. The box reacts by firing its ordinary out-of-cycle report;
+// the report ACK then delivers config/escrow/claim/floor through the UNCHANGED machinery. The hub
+// never connects inbound and never rides state on this channel (ground truth #1).
+func (h *Handler) handleWait(w http.ResponseWriter, r *http.Request) {
+	customerID, isGlobal, ok := h.checkAuthCustomer(r)
+	if !ok {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return
+	}
+	// Per-customer only: the key RESOLVES the customer and no customer_id parameter is accepted,
+	// so customer A can never observe B's generation. A global operator key has no single customer
+	// to wait on → 400 (the controller always holds a per-customer key).
+	if isGlobal {
+		http.Error(w, "wait is per-customer (present a customer key)", http.StatusBadRequest)
+		return
+	}
+	if h.intentHub == nil {
+		http.Error(w, "wait channel unavailable", http.StatusServiceUnavailable)
+		return
+	}
+
+	var lastSeen uint64
+	if s := r.URL.Query().Get("gen"); s != "" {
+		if v, err := strconv.ParseUint(s, 10, 64); err == nil {
+			lastSeen = v
+		}
+	}
+
+	// Lift THIS connection's write deadline past the hold. The global http.Server.WriteTimeout is
+	// 60 s (main.go — deliberately untouched); http.NewResponseController overrides it per-connection
+	// (Go 1.20+). Without this the held response is cut at 60 s regardless of the ingress.
+	rc := http.NewResponseController(w)
+	_ = rc.SetWriteDeadline(time.Now().Add(waitMaxHold + 30*time.Second))
+
+	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+	w.Header().Set("X-Accel-Buffering", "no") // ask nginx not to buffer the streamed hold
+	w.Header().Set("Cache-Control", "no-store")
+	w.WriteHeader(http.StatusOK)
+	if err := rc.Flush(); err != nil {
+		return // client already gone
+	}
+
+	deadline := time.Now().Add(waitMaxHold)
+	for {
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			// Hold elapsed with no change: complete with the current generation (Scenario B).
+			h.writeWaitGen(w, rc, h.intentHub.Generation(customerID))
+			return
+		}
+		sub := remaining
+		if sub > waitHeartbeat {
+			sub = waitHeartbeat
+		}
+		gen := h.intentHub.Wait(r.Context(), customerID, lastSeen, sub)
+		if gen != lastSeen {
+			h.writeWaitGen(w, rc, gen) // intent moved → complete immediately
+			return
+		}
+		// No change within this heartbeat window: emit a keepalive byte so the ingress read-timeout
+		// never fires. A write/flush failure means the client is gone — return quietly (no log
+		// noise, no goroutine leak; the intent.Wait above already deregistered).
+		if _, err := w.Write([]byte("\n")); err != nil {
+			return
+		}
+		if err := rc.Flush(); err != nil {
+			return
+		}
+	}
+}
+
+// writeWaitGen emits the single contentless completion line and flushes.
+func (h *Handler) writeWaitGen(w http.ResponseWriter, rc *http.ResponseController, gen uint64) {
+	fmt.Fprintf(w, "{\"gen\":%d}\n", gen)
+	_ = rc.Flush()
+}
@@ -0,0 +1,183 @@
+package api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/intent"
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+// shrinkWaitTiming shrinks the hold/heartbeat for the duration of a test (restored on cleanup) so
+// long-poll tests run in milliseconds instead of the 240 s production hold.
+func shrinkWaitTiming(t *testing.T, maxHold, heartbeat time.Duration) {
+	t.Helper()
+	om, oh := waitMaxHold, waitHeartbeat
+	waitMaxHold, waitHeartbeat = maxHold, heartbeat
+	t.Cleanup(func() { waitMaxHold, waitHeartbeat = om, oh })
+}
+
+// waitWith wires an intent hub onto a test handler and returns both.
+func newWaitHandler(t *testing.T) (*Handler, *store.Store, *intent.Hub) {
+	t.Helper()
+	h, st, _ := newTestHandler(t)
+	hub := intent.New()
+	h.SetIntentHub(hub)
+	return h, st, hub
+}
+
+// doWaitAsync runs GET /wait?gen=<gen> with the given bearer in a goroutine and returns a channel
+// yielding the recorder once the handler returns.
+func doWaitAsync(h *Handler, bearer string, gen string) <-chan *httptest.ResponseRecorder {
+	out := make(chan *httptest.ResponseRecorder, 1)
+	go func() {
+		req := httptest.NewRequest(http.MethodGet, "/api/v1/wait?gen="+gen, nil)
+		if bearer != "" {
+			req.Header.Set("Authorization", "Bearer "+bearer)
+		}
+		rr := httptest.NewRecorder()
+		h.ServeHTTP(rr, req)
+		out <- rr
+	}()
+	return out
+}
+
+func recvRR(t *testing.T, ch <-chan *httptest.ResponseRecorder, d time.Duration) *httptest.ResponseRecorder {
+	t.Helper()
+	select {
+	case rr := <-ch:
+		return rr
+	case <-time.After(d):
+		t.Fatalf("wait handler did not return within %s", d)
+		return nil
+	}
+}
+
+func TestWait_Unauthorized(t *testing.T) {
+	h, _, _ := newWaitHandler(t)
+	shrinkWaitTiming(t, 50*time.Millisecond, 20*time.Millisecond)
+	rr := do(h, http.MethodGet, "/wait", "", "")
+	if rr.Code != http.StatusUnauthorized {
+		t.Fatalf("no bearer: status = %d, want 401", rr.Code)
+	}
+	rr = do(h, http.MethodGet, "/wait", "WRONGKEY", "")
+	if rr.Code != http.StatusUnauthorized {
+		t.Fatalf("wrong key: status = %d, want 401", rr.Code)
+	}
+}
+
+func TestWait_GlobalKeyRejected(t *testing.T) {
+	h, _, _ := newWaitHandler(t)
+	shrinkWaitTiming(t, 50*time.Millisecond, 20*time.Millisecond)
+	rr := do(h, http.MethodGet, "/wait", globalKey, "")
+	if rr.Code != http.StatusBadRequest {
+		t.Fatalf("global key: status = %d, want 400 (wait is per-customer)", rr.Code)
+	}
+}
+
+func TestWait_ServiceUnavailableWithoutHub(t *testing.T) {
+	h, st, _ := newTestHandler(t) // no SetIntentHub
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c", RetrievalPassword: "pw", APIKey: "CKEY", ConfigJSON: "{}"}); err != nil {
+		t.Fatalf("SaveCustomerConfig: %v", err)
+	}
+	shrinkWaitTiming(t, 50*time.Millisecond, 20*time.Millisecond)
+	rr := do(h, http.MethodGet, "/wait", "CKEY", "")
+	if rr.Code != http.StatusServiceUnavailable {
+		t.Fatalf("no hub: status = %d, want 503", rr.Code)
+	}
+}
+
+func TestWait_CompletesOnBumpWithNewGen(t *testing.T) {
+	h, st, hub := newWaitHandler(t)
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c", RetrievalPassword: "pw", APIKey: "CKEY", ConfigJSON: "{}"}); err != nil {
+		t.Fatalf("SaveCustomerConfig: %v", err)
+	}
+	shrinkWaitTiming(t, 2*time.Second, 500*time.Millisecond)
+
+	ch := doWaitAsync(h, "CKEY", "0")
+	// Let the handler register, then bump.
+	time.Sleep(30 * time.Millisecond)
+	hub.Bump("c")
+
+	rr := recvRR(t, ch, time.Second)
+	if rr.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", rr.Code)
+	}
+	if got := strings.TrimSpace(rr.Body.String()); !strings.HasSuffix(got, `{"gen":1}`) {
+		t.Fatalf("body = %q, want to end with {\"gen\":1}", got)
+	}
+}
+
+// TestWait_TimesOutWithHeartbeats proves Scenario B: with no bump, the hold completes with the
+// unchanged generation, and heartbeat newline bytes were emitted while holding (the bytes that
+// keep the nginx read-timeout from firing).
+func TestWait_TimesOutWithHeartbeats(t *testing.T) {
+	h, st, _ := newWaitHandler(t)
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c", RetrievalPassword: "pw", APIKey: "CKEY", ConfigJSON: "{}"}); err != nil {
+		t.Fatalf("SaveCustomerConfig: %v", err)
+	}
+	shrinkWaitTiming(t, 120*time.Millisecond, 25*time.Millisecond)
+
+	start := time.Now()
+	rr := do(h, http.MethodGet, "/wait", "CKEY", "")
+	elapsed := time.Since(start)
+
+	if rr.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", rr.Code)
+	}
+	if elapsed < 100*time.Millisecond {
+		t.Fatalf("returned before the hold elapsed (%s) — did it not hold?", elapsed)
+	}
+	body := rr.Body.String()
+	if !strings.Contains(body, "\n") || !strings.HasPrefix(body, "\n") {
+		t.Fatalf("expected leading heartbeat newline(s); body = %q", body)
+	}
+	if !strings.HasSuffix(strings.TrimSpace(body), `{"gen":0}`) {
+		t.Fatalf("expected final {\"gen\":0} on timeout; body = %q", body)
+	}
+}
+
+// TestWait_RaceCloserReturnsImmediately proves a bump that landed before the wait connected is not
+// lost: the box passes its last-seen gen, the hub sees the generation already advanced and returns
+// at once.
+func TestWait_RaceCloserReturnsImmediately(t *testing.T) {
+	h, st, hub := newWaitHandler(t)
+	if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c", RetrievalPassword: "pw", APIKey: "CKEY", ConfigJSON: "{}"}); err != nil {
+		t.Fatalf("SaveCustomerConfig: %v", err)
+	}
+	shrinkWaitTiming(t, 5*time.Second, 1*time.Second) // long hold; must NOT be hit
+	hub.Bump("c")                                     // gen -> 1 before the box connects
+
+	start := time.Now()
+	rr := do(h, http.MethodGet, "/wait", "CKEY", "0") // last-seen 0 != 1
+	if el := time.Since(start); el > 500*time.Millisecond {
+		t.Fatalf("race-closer should return immediately, took %s", el)
+	}
+	if got := strings.TrimSpace(rr.Body.String()); !strings.HasSuffix(got, `{"gen":1}`) {
+		t.Fatalf("body = %q, want {\"gen\":1}", got)
+	}
+}
+
+// TestWait_CustomerIsolation proves customer A's key can never observe B's generation: A holds a
+// wait, B is bumped, A must NOT complete (it times out on its own generation).
+func TestWait_CustomerIsolation(t *testing.T) {
+	h, st, hub := newWaitHandler(t)
+	for _, id := range []string{"a", "b"} {
+		if err := st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: id, RetrievalPassword: "pw", APIKey: "KEY-" + id, ConfigJSON: "{}"}); err != nil {
+			t.Fatalf("SaveCustomerConfig %s: %v", id, err)
+		}
+	}
+	shrinkWaitTiming(t, 150*time.Millisecond, 40*time.Millisecond)
+
+	ch := doWaitAsync(h, "KEY-a", "0") // A waits
+	time.Sleep(30 * time.Millisecond)
+	hub.Bump("b") // B's intent moves — must not wake A
+
+	rr := recvRR(t, ch, time.Second)
+	if got := strings.TrimSpace(rr.Body.String()); !strings.HasSuffix(got, `{"gen":0}`) {
+		t.Fatalf("A should time out on its own gen 0 (isolation); body = %q", got)
+	}
+}