From 2dd05670aec03d1cd9aea6f76a4c1b49fe7f297f Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Thu, 16 Jul 2026 20:55:30 +0200 Subject: [PATCH] feat(report): Direction-2 immediate-sync wait channel client (v0.140.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit report.Waiter holds a hanging GET against hub /api/v1/wait?gen=N (same hub URL+key as the pusher). On a generation change it fires the v0.139.0 report.Trigger — nothing else; the report ACK delivers everything through the unchanged machinery. No overall client timeout (held GET); per-request ctx bounds a dead connection. First-observation records-not-fires; same-gen timeout fires nothing; errors (incl. 404 from a pre-v0.58.0 hub) back off 5s->5min while the 15-min cycle reconciles. Wired beside the trigger under the same hubPusher!=nil && Hub.Enabled gate. Red-proof: disable the baseline branch -> first observation fires (TestWaiter_FirstObservationRecordsNoFire), run-fail-reverted. Copy soften: backups_remote/escrow "néhány másodperc, legfeljebb 15 perc". Pairs with hub v0.58.0. Grounding: felhom.eu/documentation/audits/SPIKE-immediate-sync-transport-2026-07-16.md --- CHANGELOG.md | 31 +++ controller/README.md | 2 + controller/cmd/controller/main.go | 8 + controller/internal/report/waiter.go | 214 ++++++++++++++++++ controller/internal/report/waiter_test.go | 158 +++++++++++++ .../web/templates/backups_escrow.html | 2 +- .../web/templates/backups_remote.html | 2 +- 7 files changed, 415 insertions(+), 2 deletions(-) create mode 100644 controller/internal/report/waiter.go create mode 100644 controller/internal/report/waiter_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c2af9b..9425450 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,36 @@ ## Changelog +### v0.140.0 — Direction-2 immediate-sync: hub→box wait channel client (2026-07-16) + +The other half of the immediacy arc (Direction 1 = v0.139.0 box→hub trigger). An operator action on +the hub now reaches the box in **seconds** instead of on the next ~15-min cycle. Pairs with hub +v0.58.0 (the `GET /api/v1/wait` endpoint + the in-memory operator-intent notifier). Grounding: +`felhom.eu/documentation/audits/SPIKE-immediate-sync-transport-2026-07-16.md` (option b). + +- **`internal/report/waiter.go` (new) `report.Waiter`:** holds a hanging authenticated GET against + the hub's `/api/v1/wait?gen=N` (reusing the SAME hub URL + key as the pusher — no new config + keys). Its own `http.Client` has **no overall Timeout** (a held GET must stay open for the hub's + ~240 s hold) with sane connect/TLS/`ResponseHeaderTimeout` deadlines; a per-request context bounds + a black-holed connection. On a completion whose generation **differs** from the last seen, it fires + the v0.139.0 `report.Trigger` — and NOTHING else; the fired report's ACK delivers config/escrow/ + claim/floor through the UNCHANGED machinery (this adds zero delivery logic). Behaviors: + - **First observation records, never fires** (the startup report already covered current state) — + prevents a spurious echo report on every process start / config-refresh restart. Red-proof: + disable the baseline branch → `TestWaiter_FirstObservationRecordsNoFire` fires 1 (run-fail-reverted). + - **Same-generation timeout fires nothing** (the hub's hold elapsed) — not interval-shortening. + - Heartbeat newlines tolerated; the body is read only for its `{"gen":N}` line (contentless wake). + - Any error — transport, a **404 from a hub that predates the endpoint**, or a malformed body — + backs off exponentially (5 s → 5 min, reset on success) with ONE WARN per state change, and the + 15-min cycle keeps reconciling. Exits promptly on context cancel (even mid-hold). +- **`cmd/controller/main.go`:** the Waiter is constructed + started right beside the Direction-1 + trigger, gated on the SAME `hubPusher != nil && cfg.Hub.Enabled` condition (strict no-op when hub + reporting is off). One INFO line on start. +- **Copy soften (Viktor-approved):** `backups_remote.html` + `backups_escrow.html` — "ez általában + néhány **másodperc**, legfeljebb 15 perc" (was "néhány perc"). The 15-min bound stays — it is the + honest worst case when both the wait and the immediate push fail. Escrow grace window unchanged. +- **Coupling (soft):** immediacy needs hub ≥ v0.58.0; against an older hub the wait 404s and the box + degrades cleanly to the 15-min cycle. No agent coupling, no `MinAgent`. + ### v0.139.0 — immediate out-of-cycle hub report on user actions (Direction 1) (2026-07-16) Viktor's ruling: a user action with hub-side effects must round-trip in seconds, not minutes. One diff --git a/controller/README.md b/controller/README.md index 55ac2ad..946b0f9 100644 --- a/controller/README.md +++ b/controller/README.md @@ -1780,6 +1780,8 @@ Bearer token authentication, 3-attempt retry with 5-second backoff. Push status **Immediate out-of-cycle report on user actions (v0.139.0, generalizing the v0.70.0 geo push):** besides the periodic cycle, user actions with hub-side effects fire a **debounced, coalescing out-of-cycle report push** (`report.Trigger` in `internal/report/trigger.go`: buffered-1 signal channel + single worker; quiet window 2 s, min spacing 15 s, trailing-edge — a burst coalesces to ≤ 1 + ceil(burst/15 s) pushes and the LAST state always reaches the Hub). One canonical fire closure in `main.go` does the full `BuildReport`+`Claimed`+`Push`; the trigger adds NO retry of its own (the Pusher owns retries) and every failure degrades to the 15-min cycle, which stays the reconciliation backbone. Wired call sites: geo settings save/manual sync + app deploy/remove/delete (`api.Router.reportPushNow`), and via the `web.Server.SetReportTrigger` seam (`reportTriggerNow`, fired only AFTER a successful local commit): escrow recovery-code claim (the ACK hash-match flips pending→escrowed in seconds), notification-prefs save, app-email toggle, offsite target config + per-app offsite toggle, customer claim completion. `hub.enabled: false` → the seams stay nil (strict no-op). +**Direction 2 — hub→box wait channel (v0.140.0):** the reverse immediacy path, so an OPERATOR action on the hub reaches the box in seconds. `report.Waiter` (`internal/report/waiter.go`) holds a hanging authenticated `GET {hub}/api/v1/wait?gen=N` (same hub URL + key as the pusher; no new config keys) against hub ≥ v0.58.0's in-memory operator-intent generation counter. The hub completes the hold the instant any operator intent bumps that customer's generation (config save/delete, claim resend, offsite re-issue/freeze, floor, block/unblock, log-pull); on a generation **change** the Waiter fires the same Direction-1 `report.Trigger` — and nothing else, so the immediate report's ACK delivers everything through the unchanged config-refresh/escrow/claim/floor machinery (the box pulls even the wake-up; the hub never connects inbound). Its `http.Client` has **no overall timeout** (a held GET must stay open for the hub's ~240 s hold, which streams a 25 s heartbeat newline to defeat the nginx 60 s read-timeout — no ingress change needed); a per-request context bounds a dead connection. First-observation records-not-fires (no restart echo); a same-generation timeout fires nothing; any error (transport, a 404 from a pre-v0.58.0 hub, malformed body) backs off 5 s→5 min and the 15-min cycle keeps reconciling. Constructed beside the trigger under the same `hubPusher != nil && cfg.Hub.Enabled` gate. Grounding: `felhom.eu/documentation/audits/SPIKE-immediate-sync-transport-2026-07-16.md`. + #### Config apply + self-restart (`internal/api/router.go`, `internal/api/selfrestart.go`) `POST /api/config/apply` (Hub-authed) writes a new `controller.yaml`, but the new config only takes effect on **restart** — singletons such as the Cloudflare client are built once at startup (so a rotated CF API token would otherwise keep failing). Behaviour (v0.70.0): diff --git a/controller/cmd/controller/main.go b/controller/cmd/controller/main.go index dfaae73..d437804 100644 --- a/controller/cmd/controller/main.go +++ b/controller/cmd/controller/main.go @@ -811,6 +811,14 @@ func main() { } reportTrigger = report.NewTrigger(fireReport, logger) go reportTrigger.Run(ctx) + + // Direction-2 immediate-sync (v0.140.0): hold a hanging GET against the hub's wait channel + // and fire the trigger the instant operator intent moves — so an operator save round-trips in + // seconds instead of on the next ~15-min cycle. Reuses the SAME hub URL + key as the pusher; + // no new config keys. Gated on the trigger existing (hub reporting enabled). The ACK from the + // fired report delivers everything through the unchanged machinery. + waiter := report.NewWaiter(cfg.Hub.URL, cfg.Hub.APIKey, reportTrigger.Fire, logger) + go waiter.Run(ctx) } // --- Initialize API router --- diff --git a/controller/internal/report/waiter.go b/controller/internal/report/waiter.go new file mode 100644 index 0000000..767256a --- /dev/null +++ b/controller/internal/report/waiter.go @@ -0,0 +1,214 @@ +// 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 + } +} diff --git a/controller/internal/report/waiter_test.go b/controller/internal/report/waiter_test.go new file mode 100644 index 0000000..f93d5fa --- /dev/null +++ b/controller/internal/report/waiter_test.go @@ -0,0 +1,158 @@ +package report + +import ( + "context" + "fmt" + "io" + "log" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func testWaiter(t *testing.T, handler http.HandlerFunc) (*Waiter, *int32, *httptest.Server) { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + var fires int32 + w := NewWaiter(srv.URL, "KEY", func() { atomic.AddInt32(&fires, 1) }, log.New(io.Discard, "", 0)) + // Shrink tunables so the loop spins in milliseconds. + w.holdCeiling = 2 * time.Second + w.backoffMin = 10 * time.Millisecond + w.backoffMax = 40 * time.Millisecond + w.jitterMin = 3 * time.Millisecond + w.jitterMax = 6 * time.Millisecond + return w, &fires, srv +} + +// runFor runs the waiter for d, then cancels and waits for Run to return. +func runFor(w *Waiter, d time.Duration) { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { w.Run(ctx); close(done) }() + time.Sleep(d) + cancel() + <-done +} + +// genSequence returns a handler that answers the Nth request with seq[min(N-1,len-1)] (clamped), +// optionally prefixed with heartbeat newlines. +func genSequence(heartbeats int, seq ...uint64) http.HandlerFunc { + var n int32 + return func(w http.ResponseWriter, r *http.Request) { + i := int(atomic.AddInt32(&n, 1)) - 1 + if i >= len(seq) { + i = len(seq) - 1 + } + fl, _ := w.(http.Flusher) + for h := 0; h < heartbeats; h++ { + io.WriteString(w, "\n") + if fl != nil { + fl.Flush() + } + } + fmt.Fprintf(w, "{\"gen\":%d}\n", seq[i]) + } +} + +func TestWaiter_FiresOnceOnGenChange(t *testing.T) { + // baseline gen 0, then 1, then steady 1 → exactly one fire. + w, fires, _ := testWaiter(t, genSequence(0, 0, 1, 1, 1, 1)) + runFor(w, 200*time.Millisecond) + if got := atomic.LoadInt32(fires); got != 1 { + t.Fatalf("expected exactly 1 fire on a single gen change, got %d", got) + } +} + +func TestWaiter_TimeoutSameGenNoFire(t *testing.T) { + // The hub keeps returning the same generation (its hold timed out) → never fires. + w, fires, _ := testWaiter(t, genSequence(0, 7, 7, 7, 7, 7)) + runFor(w, 150*time.Millisecond) + if got := atomic.LoadInt32(fires); got != 0 { + t.Fatalf("same-gen timeouts must not fire, got %d", got) + } +} + +// TestWaiter_FirstObservationRecordsNoFire is the load-bearing property with its RED-PROOF anchor: +// a fresh process whose FIRST wait returns an already-advanced generation records it WITHOUT firing +// (the startup report already covered current state; firing would echo one extra report on every +// restart / config-refresh). +// +// RED-PROOF: drop the `!w.seen` guard in Run (fire whenever gen != lastSeen, with lastSeen starting +// at 0) — the first poll here returns gen 5 != 0 and fires once; this assertion (0 fires) then fails. +// Revert to restore green. +func TestWaiter_FirstObservationRecordsNoFire(t *testing.T) { + w, fires, _ := testWaiter(t, genSequence(0, 5, 5, 5, 5)) // existing non-zero gen, never changes + runFor(w, 150*time.Millisecond) + if got := atomic.LoadInt32(fires); got != 0 { + t.Fatalf("first observation must record-not-fire; got %d fires", got) + } +} + +func TestWaiter_HeartbeatsToleratedThenFire(t *testing.T) { + // 3 heartbeat newlines precede each completion; baseline 0 then 1 → one fire, no panic. + w, fires, _ := testWaiter(t, genSequence(3, 0, 1, 1, 1)) + runFor(w, 200*time.Millisecond) + if got := atomic.LoadInt32(fires); got != 1 { + t.Fatalf("heartbeats must be tolerated and the change fire once; got %d", got) + } +} + +func TestWaiter_MalformedCompletionNoFireNoPanic(t *testing.T) { + w, fires, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, "\n\nnot-json\n") + }) + runFor(w, 120*time.Millisecond) + if got := atomic.LoadInt32(fires); got != 0 { + t.Fatalf("a malformed completion must not fire; got %d", got) + } +} + +func TestWaiter_404TreatedAsErrorNoFire(t *testing.T) { + // A hub that predates the endpoint returns 404 — the Waiter backs off, never fires, keeps looping. + w, fires, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + }) + runFor(w, 120*time.Millisecond) + if got := atomic.LoadInt32(fires); got != 0 { + t.Fatalf("404 must be an ordinary (no-fire) error; got %d", got) + } +} + +func TestWaiter_SendsBearerAndGenParam(t *testing.T) { + var sawAuth, sawGen atomic.Value + sawAuth.Store("") + sawGen.Store("") + w, _, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) { + sawAuth.Store(r.Header.Get("Authorization")) + sawGen.Store(r.URL.Query().Get("gen")) + fmt.Fprint(w, "{\"gen\":0}\n") + }) + runFor(w, 80*time.Millisecond) + if sawAuth.Load().(string) != "Bearer KEY" { + t.Fatalf("expected Bearer KEY, got %q", sawAuth.Load()) + } + if sawGen.Load().(string) != "0" { + t.Fatalf("expected gen=0 on the first poll, got %q", sawGen.Load()) + } +} + +func TestWaiter_CtxCancelMidHoldReturnsPromptly(t *testing.T) { + // The server holds until the client disconnects; cancelling the ctx must return Run promptly. + w, _, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() // hold until the client goes away + }) + w.holdCeiling = 10 * time.Second // must NOT be what unblocks us + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { w.Run(ctx); close(done) }() + time.Sleep(40 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return promptly after ctx cancel mid-hold") + } +} diff --git a/controller/internal/web/templates/backups_escrow.html b/controller/internal/web/templates/backups_escrow.html index 0f11f69..bf28168 100644 --- a/controller/internal/web/templates/backups_escrow.html +++ b/controller/internal/web/templates/backups_escrow.html @@ -88,7 +88,7 @@ where the interim "awaiting hub confirmation" card shows until the flip. Setting expectations here avoids surprise at that card. */}}
- Mi történik ezután? A helyreállítási kód elkészült, és a rendszer elküldi a központba megerősítésre — ez általában néhány perc, legfeljebb 15 perc. Addig a Távoli mentés oldalon a „megerősítésre vár" üzenet látható; a megerősítés után indítható az első távoli mentés. + Mi történik ezután? A helyreállítási kód elkészült, és a rendszer elküldi a központba megerősítésre — ez általában néhány másodperc, legfeljebb 15 perc. Addig a Távoli mentés oldalon a „megerősítésre vár" üzenet látható; a megerősítés után indítható az első távoli mentés.