felhom-controller/controller/internal/report/waiter.go

// waiter.go — the Direction-2 immediate-sync long-poll client (v0.140.0).
//
// The box holds a hanging authenticated GET against the hub's /api/v1/wait, carrying its last-seen
// operator-intent generation. The hub completes the hold the instant that generation moves (an
// operator saved config, re-issued offsite, changed a floor, …) or after its hold window. On a
// CHANGE the Waiter simply fires the v0.139.0 report.Trigger — the immediate report's ACK then
// delivers config/escrow/claim/floor through the UNCHANGED machinery (this file adds NO delivery
// logic). On a timeout (same generation) it fires nothing and reconnects. Every failure degrades to
// the ~15-min scheduled cycle, which stays the reconciliation backbone.
//
// The wait response is CONTENTLESS — the Waiter reads only a generation number from it (a
// {"gen":N} line preceded by heartbeat newlines) and never interprets anything else. Grounding:
// felhom.eu/documentation/audits/SPIKE-immediate-sync-transport-2026-07-16.md (option b).
package report

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"math/rand"
	"net"
	"net/http"
	"strings"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/logx"
)

const (
	// waiterHoldCeiling bounds ONE held request from the client side. The hub holds ~240 s; this
	// margin lets a healthy completion arrive while capping a black-holed connection (hub died
	// mid-hold, NAT dropped the mapping) so the loop can back off instead of hanging forever.
	waiterHoldCeiling = 300 * time.Second
	waiterBackoffMin  = 5 * time.Second
	waiterBackoffMax  = 5 * time.Minute
	// Reconnect jitter after a clean completion — spreads fleet reconnects so they don't align.
	waiterJitterMin = 1 * time.Second
	waiterJitterMax = 3 * time.Second
)

// Waiter long-polls the hub wait channel and fires onWake when the operator-intent generation
// advances. Construct with NewWaiter; Run owns all state (single goroutine).
type Waiter struct {
	url    string // "{hub}/api/v1/wait"
	apiKey string
	client *http.Client
	onWake func()
	logger *log.Logger

	// Tunables — production defaults set by NewWaiter; tests shrink them.
	holdCeiling time.Duration
	backoffMin  time.Duration
	backoffMax  time.Duration
	jitterMin   time.Duration
	jitterMax   time.Duration

	// Run-goroutine-only state.
	lastSeen uint64
	seen     bool
	failing  bool
}

// NewWaiter builds a Waiter against the hub. The HTTP client deliberately has NO overall Timeout (a
// held GET must be able to stay open for the hub's full hold) — only sane connect/TLS/header
// deadlines, so a dead hub is detected quickly while a healthy hold is never cut.
func NewWaiter(hubURL, apiKey string, onWake func(), logger *log.Logger) *Waiter {
	client := &http.Client{
		Timeout: 0, // NO total timeout — the whole point is to hold
		Transport: &http.Transport{
			DialContext:           (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
			TLSHandshakeTimeout:   10 * time.Second,
			ResponseHeaderTimeout: 30 * time.Second, // the hub flushes headers immediately, before holding
			ExpectContinueTimeout: 1 * time.Second,
			IdleConnTimeout:       90 * time.Second,
		},
	}
	return &Waiter{
		url:         strings.TrimRight(hubURL, "/") + "/api/v1/wait",
		apiKey:      apiKey,
		client:      client,
		onWake:      onWake,
		logger:      logger,
		holdCeiling: waiterHoldCeiling,
		backoffMin:  waiterBackoffMin,
		backoffMax:  waiterBackoffMax,
		jitterMin:   waiterJitterMin,
		jitterMax:   waiterJitterMax,
	}
}

// Run is the loop; main.go starts it under the process context. Exits promptly on ctx cancel.
func (w *Waiter) Run(ctx context.Context) {
	logx.Infof(w.logger, "[report] hub wait channel active (hold ≤240s)")
	backoff := w.backoffMin
	for {
		if ctx.Err() != nil {
			return
		}
		gen, err := w.pollOnce(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return // shutdown, not a real failure
			}
			if !w.failing {
				w.failing = true
				logx.Warnf(w.logger, "[report] wait channel error: %v — backing off (the 15-min cycle still reconciles)", err)
			}
			if !w.sleep(ctx, backoff) {
				return
			}
			backoff *= 2
			if backoff > w.backoffMax {
				backoff = w.backoffMax
			}
			continue
		}
		// Success: reset failure state + backoff.
		if w.failing {
			w.failing = false
			logx.Infof(w.logger, "[report] wait channel recovered")
		}
		backoff = w.backoffMin

		switch {
		case !w.seen:
			// First observation records the baseline WITHOUT firing — the startup report already
			// covered current state; firing here would echo one extra report on every process start.
			w.seen = true
			w.lastSeen = gen
			logx.Debugf(w.logger, "[report] wait baseline generation=%d", gen)
		case gen != w.lastSeen:
			w.lastSeen = gen
			logx.Debugf(w.logger, "[report] wait woke: generation=%d — firing out-of-cycle report", gen)
			if w.onWake != nil {
				w.onWake()
			}
		default:
			// Same generation → the hub's hold simply timed out; nothing to do (Scenario B).
			w.lastSeen = gen
		}

		if !w.sleep(ctx, w.jitter()) {
			return
		}
	}
}

// pollOnce performs one held GET and returns the generation the hub reported. A non-2xx status
// (incl. 404 from a hub that predates the endpoint), a transport error, or an unparsable body all
// return an error so the caller backs off.
func (w *Waiter) pollOnce(ctx context.Context) (uint64, error) {
	reqCtx, cancel := context.WithTimeout(ctx, w.holdCeiling)
	defer cancel()

	req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, fmt.Sprintf("%s?gen=%d", w.url, w.lastSeen), nil)
	if err != nil {
		return 0, err
	}
	if w.apiKey != "" {
		req.Header.Set("Authorization", "Bearer "+w.apiKey)
	}
	resp, err := w.client.Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return 0, fmt.Errorf("wait status %d", resp.StatusCode)
	}

	// The hold streams heartbeat newlines then a single {"gen":N} line. Read line-wise, ignore
	// blanks, parse the first non-blank line as the completion.
	sc := bufio.NewScanner(resp.Body)
	for sc.Scan() {
		line := strings.TrimSpace(sc.Text())
		if line == "" {
			continue // heartbeat
		}
		var r struct {
			Gen uint64 `json:"gen"`
		}
		if err := json.Unmarshal([]byte(line), &r); err != nil {
			return 0, fmt.Errorf("malformed wait completion %q: %w", line, err)
		}
		return r.Gen, nil
	}
	if err := sc.Err(); err != nil {
		return 0, err
	}
	return 0, fmt.Errorf("wait closed without a completion line")
}

// jitter returns a random reconnect delay in [jitterMin, jitterMax].
func (w *Waiter) jitter() time.Duration {
	span := w.jitterMax - w.jitterMin
	if span <= 0 {
		return w.jitterMin
	}
	return w.jitterMin + time.Duration(rand.Int63n(int64(span)))
}

// sleep waits d or until ctx is cancelled; false = cancelled (caller returns promptly).
func (w *Waiter) sleep(ctx context.Context, d time.Duration) bool {
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}