From 3809f777109914ebaeb6437c42f67baeeacc8d2a Mon Sep 17 00:00:00 2001 From: Kristoffer Risa Date: Sun, 30 Aug 2026 07:02:39 +0000 Subject: [PATCH] feat: reconnect the live stream with bounded backoff `powerctl live` now survives a transient network failure instead of exiting, so it can be left running as a monitor. Subscribe stays a single-attempt primitive; SubscribeWithReconnect layers the retry above it. Backoff starts at 5s and doubles to a 5-minute ceiling with equal jitter, and no more than 10 connections are opened in a rolling hour -- Tibber allows 20, and a stream that keeps failing must not spend the allowance a manual restart needs. Errors are classified before retrying. A wrong answer here is expensive in both directions: dropping a recoverable stream, or burning the hourly budget on a request that can never succeed. Read errors and abnormal closes retry; close code 4403, a refused handshake, an unknown home ID and other GraphQL errors are fatal and exit 1. Fatal errors are tagged where they are raised rather than matched on message text later. Adds --no-reconnect for the previous behaviour. Reconnect notices go to stderr, keeping stdout a clean data channel for --format json. Also fixes an adjacent bug the tests exposed: on Ctrl+C the deferred close handshake waited up to five seconds for a reply the server may never send. Closes #22 --- ARCHITECTURE.md | 47 +++- README.md | 22 +- internal/api/livestream_test.go | 261 +++++++++++++++++++ internal/api/reconnect.go | 249 ++++++++++++++++++ internal/api/reconnect_test.go | 439 ++++++++++++++++++++++++++++++++ internal/api/websocket.go | 37 ++- internal/commands/live.go | 31 ++- 7 files changed, 1067 insertions(+), 19 deletions(-) create mode 100644 internal/api/livestream_test.go create mode 100644 internal/api/reconnect.go create mode 100644 internal/api/reconnect_test.go diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 56f5893..cd86fbc 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -22,7 +22,10 @@ powerctl-cli/ │ │ ├── client_test.go │ │ ├── queries.go # GraphQL query definitions │ │ ├── websocket.go # WebSocket for live streaming -│ │ └── websocket_test.go +│ │ ├── websocket_test.go +│ │ ├── reconnect.go # Backoff, hourly budget, error taxonomy +│ │ ├── reconnect_test.go +│ │ └── livestream_test.go # End-to-end against a local WebSocket server │ ├── commands/ │ │ ├── root.go # Root command, global flags │ │ ├── config.go # `powerctl config` - setup wizard @@ -94,12 +97,43 @@ format: "markdown" # Default output format - Protocol: `graphql-transport-ws` - Requires `User-Agent: powerctl-cli/1.0` — Tibber rejects the default Go client -- Reconnection: none. A read or parse error ends the stream and exits 1 - Heartbeat: none. The connection lives as long as the server keeps it open -- Graceful shutdown on SIGINT/SIGTERM +- Graceful shutdown on SIGINT/SIGTERM, including mid-backoff - A GraphQL error payload or a null `liveMeasurement` becomes an error rather than a nil measurement handed to the formatter (see #13) +#### Reconnection (`reconnect.go`) + +`Subscribe` stays a single-attempt primitive; `SubscribeWithReconnect` layers +the retry on top, so the connection code stays testable without a retry loop +running through it. + +- Backoff: 5s, doubling to a 5-minute ceiling, with equal jitter (half the delay + fixed, half random) +- Budget: at most 10 connection attempts per rolling hour. Tibber allows 20, and + a failing stream must not spend the allowance a manual restart needs. Hitting + the cap ends the stream with `ErrReconnectBudget` +- Reset: a connection that delivered data and stayed up for a minute returns the + backoff to 5s +- `--no-reconnect` skips the layer entirely and calls `Subscribe` directly + +Whether an error is retried is the substance of it, since a wrong answer either +drops a recoverable stream or burns the hourly budget on a request that can +never succeed: + +| Condition | Behaviour | +|-----------|-----------| +| Read error, abnormal or server-initiated close | Retry | +| Handshake refused with HTTP 401/403 | Fatal, exit 1 | +| Close code `4403 Invalid token` | Fatal, exit 1 | +| `connection_error` in place of `connection_ack` | Fatal, exit 1 | +| Unknown home ID, other GraphQL errors (see #13) | Fatal, exit 1 | +| Ack timeout (10s, internal deadline) | Retry | +| `ctx` cancelled by SIGINT/SIGTERM | Clean exit 0 | + +Fatal errors are tagged at the point they are raised (`markFatal`) rather than +matched on message text later. + ### Commands (`internal/commands/`) | Command | Input | Output | Exit Codes | @@ -110,7 +144,7 @@ format: "markdown" # Default output format | `home` | - | Home info | 0=OK, 1=Error | | `prices` | - | Price list | 0=OK, 1=Error | | `consumption`| `--resolution`, `--last`, `--home-id` | Consumption history | 0=OK, 1=Error | -| `live` | `--home-id` | Stream | 0=Clean exit, 1=Error | +| `live` | `--home-id`, `--no-reconnect` | Stream | 0=Clean exit, 1=Error | | `version` | - | Version, commit, build date | 0=OK | ### Output Formatters (`internal/output/`) @@ -180,6 +214,7 @@ Authorization: Bearer | Category | Strategy | |----------|----------| | Network errors | Log and exit with code 1 | +| Dropped live stream | Reconnect with bounded backoff; exit 1 once the hourly budget is spent | | Auth errors | Clear message: "Invalid token" | | No Pulse | Exit 1 with "Pulse not enabled" | | Parse errors | Log raw response, exit 1 | @@ -210,7 +245,9 @@ windows/amd64. Release builds do not use this — see below. internal/ ├── api/ │ ├── client_test.go # Mock HTTP responses -│ └── websocket_test.go # Live payload parsing and error mapping +│ ├── websocket_test.go # Live payload parsing and error mapping +│ ├── reconnect_test.go # Backoff and budget, on a faked clock +│ └── livestream_test.go # Reconnect over a real socket, via httptest ├── config/ │ └── config_test.go # Config resolution order └── output/ diff --git a/README.md b/README.md index aa4141f..42b4914 100644 --- a/README.md +++ b/README.md @@ -173,6 +173,19 @@ powerctl live --home-id 96a14971-525a-4420-aae9-e5aefaf46a81 `powerctl consumption` takes the same flag. +A dropped connection is retried automatically, so the stream can be left +running. The wait starts at 5 seconds and doubles to a 5-minute ceiling, and no +more than 10 connections are opened in any rolling hour — Tibber allows 20, and +a manual restart needs some left. Progress notes go to stderr, so `--format +json` output stays pipeable. + +Failures a reconnect cannot fix — an invalid token, an unknown home ID — still +exit 1 immediately instead of retrying. To get that for every error: + +```bash +powerctl live --no-reconnect +``` + ### Output Formats Default output is beautiful colored CLI. Change format with `--format`: @@ -247,9 +260,12 @@ make lint # Run linter (requires golangci-lint) - The ID passed to `--home-id` is not on your account. List valid IDs with `powerctl home` **Live stream disconnects** -- Rate limit is 20 connections/hour -- There is no automatic reconnect — the command exits with code 1 and the error. - Re-run it, or wrap it in a supervisor if you need it to stay up +- Dropped connections are retried automatically — watch stderr for + `Reconnecting in ...` +- Rate limit is 20 connections/hour. `powerctl` spends at most 10 of them, then + stops with "reconnect budget exhausted" rather than taking the rest; wait for + the hour to roll over before restarting +- `--no-reconnect` restores the old behaviour of exiting 1 on the first error ## Contributing diff --git a/internal/api/livestream_test.go b/internal/api/livestream_test.go new file mode 100644 index 0000000..2f823d0 --- /dev/null +++ b/internal/api/livestream_test.go @@ -0,0 +1,261 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/websocket" + + "github.com/kristofferrisa/powerctl-cli/internal/models" +) + +// liveServer is a minimal graphql-transport-ws server standing in for Tibber, so +// the reconnect path can be exercised over a real socket rather than a stub. +type liveServer struct { + *httptest.Server + + mu sync.Mutex + conns int +} + +// startLiveServer accepts WebSocket connections and hands each one to session, +// already past connection_init, with the 1-based connection number. +func startLiveServer(t *testing.T, session func(ctx context.Context, conn *websocket.Conn, n int)) *liveServer { + t.Helper() + + s := &liveServer{} + s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + Subprotocols: []string{"graphql-transport-ws"}, + }) + if err != nil { + return + } + defer conn.CloseNow() + + s.mu.Lock() + s.conns++ + n := s.conns + s.mu.Unlock() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + if _, _, err := conn.Read(ctx); err != nil { // connection_init + return + } + session(ctx, conn, n) + })) + t.Cleanup(s.Close) + return s +} + +func (s *liveServer) connections() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.conns +} + +// client returns a LiveClient pointed at this server. +func (s *liveServer) client() *LiveClient { + c := NewLiveClient("test-token", "test-home") + c.endpoint = "ws" + strings.TrimPrefix(s.URL, "http") + return c +} + +func send(ctx context.Context, conn *websocket.Conn, msgType string, payload string) error { + data, err := json.Marshal(wsMessage{Type: msgType, ID: "1", Payload: json.RawMessage(payload)}) + if err != nil { + return err + } + return conn.Write(ctx, websocket.MessageText, data) +} + +func ackAndAwaitSubscribe(ctx context.Context, conn *websocket.Conn) error { + if err := send(ctx, conn, "connection_ack", ""); err != nil { + return err + } + _, _, err := conn.Read(ctx) // subscribe + return err +} + +const measurementPayload = `{"data":{"liveMeasurement":{"timestamp":"2026-01-01T12:00:00Z","power":1234}}}` + +// fastPolicy retries almost immediately so the tests do not wait on real backoff. +func fastPolicy() ReconnectPolicy { + p := DefaultReconnectPolicy() + p.BaseDelay = time.Millisecond + p.MaxDelay = time.Millisecond + return p +} + +func TestSubscribeWithReconnect_RecoversFromDroppedConnection(t *testing.T) { + server := startLiveServer(t, func(ctx context.Context, conn *websocket.Conn, n int) { + if err := ackAndAwaitSubscribe(ctx, conn); err != nil { + return + } + if err := send(ctx, conn, "next", measurementPayload); err != nil { + return + } + if n == 1 { + conn.CloseNow() // the blip: no close handshake, just a dead socket + return + } + _ = send(ctx, conn, "complete", "") + }) + + var measurements []*models.LiveMeasurement + err := server.client().SubscribeWithReconnect(context.Background(), fastPolicy(), + func(m *models.LiveMeasurement) error { + measurements = append(measurements, m) + return nil + }) + + if err != nil { + t.Fatalf("expected the stream to recover, got %v", err) + } + if server.connections() != 2 { + t.Errorf("expected 2 connections, got %d", server.connections()) + } + if len(measurements) != 2 { + t.Fatalf("expected 2 measurements across the reconnect, got %d", len(measurements)) + } + for i, m := range measurements { + if m == nil || m.Power != 1234 { + t.Errorf("measurement %d: expected 1234 W, got %v", i, m) + } + } +} + +func TestSubscribeWithReconnect_DoesNotRetryFatalErrors(t *testing.T) { + tests := []struct { + name string + session func(ctx context.Context, conn *websocket.Conn, n int) + wantErr string + }{ + { + name: "rejected token", + session: func(ctx context.Context, conn *websocket.Conn, _ int) { + _ = send(ctx, conn, "connection_error", `{"message":"Invalid token"}`) + }, + wantErr: "connection rejected", + }, + { + name: "token refused with close code 4403", + session: func(_ context.Context, conn *websocket.Conn, _ int) { + _ = conn.Close(statusInvalidToken, "Invalid token") + }, + wantErr: "4403", + }, + { + name: "unknown home ID", + session: func(ctx context.Context, conn *websocket.Conn, _ int) { + if err := ackAndAwaitSubscribe(ctx, conn); err != nil { + return + } + _ = send(ctx, conn, "next", `{"errors":[{"message":"home not found"}],"data":{"liveMeasurement":null}}`) + }, + wantErr: "invalid or non-existing home ID", + }, + { + name: "subscription error", + session: func(ctx context.Context, conn *websocket.Conn, _ int) { + if err := ackAndAwaitSubscribe(ctx, conn); err != nil { + return + } + _ = send(ctx, conn, "error", `[{"message":"query is not valid"}]`) + }, + wantErr: "subscription error", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := startLiveServer(t, tt.session) + + err := server.client().SubscribeWithReconnect(context.Background(), fastPolicy(), + func(*models.LiveMeasurement) error { return nil }) + + if err == nil { + t.Fatal("expected an error, got nil") + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Errorf("expected an error containing %q, got %q", tt.wantErr, err) + } + if errors.Is(err, ErrReconnectBudget) { + t.Error("retried a fatal error until the connection budget ran out") + } + if server.connections() != 1 { + t.Errorf("expected 1 connection attempt, got %d", server.connections()) + } + }) + } +} + +func TestSubscribeWithReconnect_StopsWhenCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + server := startLiveServer(t, func(sctx context.Context, conn *websocket.Conn, _ int) { + if err := ackAndAwaitSubscribe(sctx, conn); err != nil { + return + } + for { + if err := send(sctx, conn, "next", measurementPayload); err != nil { + return + } + time.Sleep(time.Millisecond) + } + }) + + received := 0 + start := time.Now() + err := server.client().SubscribeWithReconnect(ctx, fastPolicy(), + func(*models.LiveMeasurement) error { + received++ + if received == 3 { + cancel() // stand-in for Ctrl+C + } + return nil + }) + + // Ctrl+C must not wait on a close handshake the server may never answer. + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Errorf("expected a prompt exit on cancellation, took %v", elapsed) + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if server.connections() != 1 { + t.Errorf("expected no reconnect after cancellation, got %d connections", server.connections()) + } +} + +func TestSubscribe_NoReconnectStopsOnFirstError(t *testing.T) { + server := startLiveServer(t, func(ctx context.Context, conn *websocket.Conn, _ int) { + if err := ackAndAwaitSubscribe(ctx, conn); err != nil { + return + } + if err := send(ctx, conn, "next", measurementPayload); err != nil { + return + } + conn.CloseNow() + }) + + // What `powerctl live --no-reconnect` does: one connection, then out. + err := server.client().Subscribe(context.Background(), func(*models.LiveMeasurement) error { return nil }) + + if err == nil { + t.Fatal("expected a read error, got nil") + } + if server.connections() != 1 { + t.Errorf("expected 1 connection, got %d", server.connections()) + } +} diff --git a/internal/api/reconnect.go b/internal/api/reconnect.go new file mode 100644 index 0000000..c8c9e5e --- /dev/null +++ b/internal/api/reconnect.go @@ -0,0 +1,249 @@ +package api + +import ( + "context" + "errors" + "fmt" + "math/rand/v2" + "time" + + "github.com/coder/websocket" + + "github.com/kristofferrisa/powerctl-cli/internal/models" +) + +// Reconnect defaults. MaxPerHour is the one that matters: Tibber allows 20 +// WebSocket connections per hour per token, so a stream that keeps failing has +// to leave headroom rather than spend the whole allowance and lock the user out +// for the rest of the hour. +const ( + defaultBaseDelay = 5 * time.Second + defaultMaxDelay = 5 * time.Minute + defaultMaxPerHour = 10 + defaultHealthyAfter = time.Minute +) + +// statusInvalidToken is the WebSocket close code Tibber sends for a rejected token. +const statusInvalidToken websocket.StatusCode = 4403 + +// ErrReconnectBudget reports that the per-hour connection cap was reached. +// Retrying past it would burn the Tibber quota that a manual restart needs. +var ErrReconnectBudget = errors.New("reconnect budget exhausted") + +// ReconnectPolicy bounds how a dropped live stream is retried. +type ReconnectPolicy struct { + // BaseDelay is the wait before the first retry. It doubles on each + // consecutive failure, up to MaxDelay. + BaseDelay time.Duration + + // MaxDelay caps the backoff. + MaxDelay time.Duration + + // MaxPerHour caps connection attempts in any rolling hour, counting the + // first one. Reaching it ends the stream with ErrReconnectBudget. + MaxPerHour int + + // HealthyAfter is how long a connection must stay up, having delivered at + // least one measurement, before the backoff resets to BaseDelay. + HealthyAfter time.Duration + + // Notify, if set, runs before each backoff wait. It must not write to + // stdout: that stream carries measurement data. + Notify func(attempt int, delay time.Duration, cause error) + + // Test seams. Left nil, they resolve to the real clock and jittered delays. + now func() time.Time + sleep func(context.Context, time.Duration) error + jitter func(time.Duration) time.Duration +} + +// DefaultReconnectPolicy returns the policy used by `powerctl live`. +func DefaultReconnectPolicy() ReconnectPolicy { + return ReconnectPolicy{ + BaseDelay: defaultBaseDelay, + MaxDelay: defaultMaxDelay, + MaxPerHour: defaultMaxPerHour, + HealthyAfter: defaultHealthyAfter, + } +} + +func (p ReconnectPolicy) withDefaults() ReconnectPolicy { + if p.BaseDelay <= 0 { + p.BaseDelay = defaultBaseDelay + } + if p.MaxDelay <= 0 { + p.MaxDelay = defaultMaxDelay + } + if p.MaxDelay < p.BaseDelay { + p.MaxDelay = p.BaseDelay + } + if p.MaxPerHour <= 0 { + p.MaxPerHour = defaultMaxPerHour + } + if p.HealthyAfter <= 0 { + p.HealthyAfter = defaultHealthyAfter + } + if p.now == nil { + p.now = time.Now + } + if p.sleep == nil { + p.sleep = sleepCtx + } + if p.jitter == nil { + p.jitter = equalJitter + } + return p +} + +// SubscribeWithReconnect streams live measurements, reconnecting on transient +// failures within the bounds of p. It returns nil when the server ends the +// subscription, ctx's error when the caller cancels, and otherwise the failure +// that stopped it: a fatal error, or ErrReconnectBudget. +func (c *LiveClient) SubscribeWithReconnect(ctx context.Context, p ReconnectPolicy, handler func(*models.LiveMeasurement) error) error { + return retryLoop(ctx, p, func(ctx context.Context) (bool, error) { + delivered := false + err := c.Subscribe(ctx, func(m *models.LiveMeasurement) error { + delivered = true + return handler(m) + }) + return delivered, err + }) +} + +// attemptFunc runs one connection attempt, reporting whether it delivered at +// least one measurement before failing. +type attemptFunc func(ctx context.Context) (delivered bool, err error) + +func retryLoop(ctx context.Context, p ReconnectPolicy, attempt attemptFunc) error { + p = p.withDefaults() + + var window []time.Time // connection attempts inside the last hour + consecutive := 0 + + for { + started := p.now() + window = append(pruneWindow(window, started), started) + + delivered, err := attempt(ctx) + if err == nil { + return nil + } + // Cancellation is the caller's decision, so it outranks whatever error + // the attempt returned while unwinding. + if ctx.Err() != nil { + return err + } + if !retryable(err) { + return err + } + + // A connection that carried data for a while was not part of this run of + // failures, so the next retry starts over at BaseDelay. + if delivered && p.now().Sub(started) >= p.HealthyAfter { + consecutive = 0 + } + consecutive++ + + delay := p.jitter(backoff(p.BaseDelay, p.MaxDelay, consecutive-1)) + + // Check the budget as it will stand once the wait ends: waiting can age an + // attempt out of the window and free a slot, but when it cannot there is + // no reason to pause before saying so. + if spent := len(pruneWindow(window, p.now().Add(delay))); spent >= p.MaxPerHour { + return fmt.Errorf("%w: %d connection attempts in the last hour (Tibber allows 20); last error: %v", + ErrReconnectBudget, spent, err) + } + + if p.Notify != nil { + p.Notify(consecutive, delay, err) + } + if err := p.sleep(ctx, delay); err != nil { + return err + } + } +} + +// retryable reports whether reconnecting could plausibly fix err. Cancellation +// is not considered here; retryLoop checks the context directly, so an internal +// deadline (the ack timeout) stays retryable. +// +// Being too permissive here is expensive: an invalid token or an unknown home ID +// fails identically on every attempt, so retrying only spends the hourly budget +// before handing the user the same message. +func retryable(err error) bool { + if err == nil { + return false + } + if IsFatal(err) { + return false + } + if websocket.CloseStatus(err) == statusInvalidToken { + return false + } + return true +} + +// fatalError marks a stream failure that reconnecting cannot fix. +type fatalError struct{ err error } + +func (e *fatalError) Error() string { return e.err.Error() } +func (e *fatalError) Unwrap() error { return e.err } + +func markFatal(err error) error { + if err == nil { + return nil + } + return &fatalError{err: err} +} + +// IsFatal reports whether err ends the live stream for good. +func IsFatal(err error) bool { + var fe *fatalError + return errors.As(err, &fe) +} + +// backoff returns base doubled n times, capped at ceiling. +func backoff(base, ceiling time.Duration, n int) time.Duration { + d := base + for i := 0; i < n; i++ { + d *= 2 + if d >= ceiling || d <= 0 { // d <= 0 means the doubling overflowed + return ceiling + } + } + return d +} + +// equalJitter keeps half the delay and randomises the rest, so several clients +// dropped by the same outage spread their retries instead of arriving together. +func equalJitter(d time.Duration) time.Duration { + half := d / 2 + if half <= 0 { + return d + } + return half + time.Duration(rand.Int64N(int64(half))) +} + +// sleepCtx waits for d, returning early if ctx is cancelled. Ctrl+C during a +// backoff wait has to exit now, not five minutes from now. +func sleepCtx(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + +// pruneWindow drops attempts older than an hour. Entries are appended in order, +// so the stale ones are always a prefix. +func pruneWindow(window []time.Time, now time.Time) []time.Time { + cutoff := now.Add(-time.Hour) + i := 0 + for i < len(window) && !window[i].After(cutoff) { + i++ + } + return window[i:] +} diff --git a/internal/api/reconnect_test.go b/internal/api/reconnect_test.go new file mode 100644 index 0000000..322eccc --- /dev/null +++ b/internal/api/reconnect_test.go @@ -0,0 +1,439 @@ +package api + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/coder/websocket" +) + +// fakeClock drives retryLoop without real waiting: sleeping advances the clock, +// so the rolling-hour window can be exercised in a few microseconds. +type fakeClock struct { + t time.Time + delays []time.Duration +} + +func newFakeClock() *fakeClock { + return &fakeClock{t: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)} +} + +func (c *fakeClock) now() time.Time { return c.t } + +func (c *fakeClock) advance(d time.Duration) { c.t = c.t.Add(d) } + +func (c *fakeClock) sleep(_ context.Context, d time.Duration) error { + c.delays = append(c.delays, d) + c.advance(d) + return nil +} + +// testPolicy is the default policy wired to a fake clock, with jitter removed so +// delays are exact. +func testPolicy(c *fakeClock) ReconnectPolicy { + p := DefaultReconnectPolicy() + p.now = c.now + p.sleep = c.sleep + p.jitter = func(d time.Duration) time.Duration { return d } + return p +} + +// alwaysFails is an attempt that never succeeds. It aborts the test rather than +// spinning forever if the hourly budget stops capping attempts, so a regression +// there fails fast instead of hanging until the go test timeout. +func alwaysFails(t *testing.T, limit int, count *int) attemptFunc { + t.Helper() + return func(context.Context) (bool, error) { + *count++ + if *count > limit { + t.Fatalf("made %d attempts with no cap; the hourly budget is not holding", *count) + } + return false, errors.New("read error: connection reset") + } +} + +func TestRetryLoop_RetriesTransientFailures(t *testing.T) { + clock := newFakeClock() + attempts := 0 + + err := retryLoop(context.Background(), testPolicy(clock), func(context.Context) (bool, error) { + attempts++ + if attempts < 3 { + return false, errors.New("read error: connection reset") + } + return true, nil + }) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if attempts != 3 { + t.Errorf("expected 3 attempts, got %d", attempts) + } +} + +func TestRetryLoop_StopsOnFatalError(t *testing.T) { + tests := []struct { + name string + err error + }{ + { + name: "invalid home ID", + err: markFatal(errors.New("invalid or non-existing home ID")), + }, + { + name: "graphql error payload", + err: markFatal(errors.New("subscription error: something is wrong")), + }, + { + name: "invalid token close code", + err: fmt.Errorf("read error: %w", websocket.CloseError{Code: statusInvalidToken, Reason: "Invalid token"}), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + clock := newFakeClock() + attempts := 0 + + err := retryLoop(context.Background(), testPolicy(clock), func(context.Context) (bool, error) { + attempts++ + return false, tt.err + }) + + if !errors.Is(err, tt.err) { + t.Fatalf("expected %v, got %v", tt.err, err) + } + if attempts != 1 { + t.Errorf("expected the stream to stop after 1 attempt, got %d", attempts) + } + if len(clock.delays) != 0 { + t.Errorf("expected no backoff waits, got %v", clock.delays) + } + }) + } +} + +func TestRetryLoop_StaysUnderHourlyBudget(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + attempts := 0 + + err := retryLoop(context.Background(), policy, alwaysFails(t, policy.MaxPerHour, &attempts)) + + if !errors.Is(err, ErrReconnectBudget) { + t.Fatalf("expected ErrReconnectBudget, got %v", err) + } + if attempts != policy.MaxPerHour { + t.Errorf("expected %d attempts, got %d", policy.MaxPerHour, attempts) + } + // The point of the cap: Tibber allows 20 connections per hour and a manual + // restart needs some of them left. + if attempts >= 20 { + t.Errorf("attempts %d would exhaust Tibber's hourly connection quota", attempts) + } + if elapsed := clock.now().Sub(newFakeClock().now()); elapsed > time.Hour { + t.Errorf("budget should be spent inside one hour, took %v", elapsed) + } +} + +func TestRetryLoop_BudgetRecoversAfterAnHour(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + attempts := 0 + + err := retryLoop(context.Background(), policy, func(context.Context) (bool, error) { + attempts++ + // Every connection dies after two hours, so no two attempts ever share a + // rolling-hour window and the budget never runs out. + clock.advance(2 * time.Hour) + if attempts == 4 { + return true, nil + } + return true, errors.New("read error: connection reset") + }) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if attempts != 4 { + t.Errorf("expected 4 attempts, got %d", attempts) + } +} + +func TestRetryLoop_BackoffDoublesAndCaps(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + policy.MaxPerHour = 8 + + attempts := 0 + _ = retryLoop(context.Background(), policy, alwaysFails(t, policy.MaxPerHour, &attempts)) + + want := []time.Duration{ + 5 * time.Second, + 10 * time.Second, + 20 * time.Second, + 40 * time.Second, + 80 * time.Second, + 160 * time.Second, + defaultMaxDelay, + } + if len(clock.delays) != len(want) { + t.Fatalf("expected %d waits, got %d: %v", len(want), len(clock.delays), clock.delays) + } + for i, w := range want { + if clock.delays[i] != w { + t.Errorf("wait %d: expected %v, got %v", i, w, clock.delays[i]) + } + } +} + +func TestRetryLoop_HealthyConnectionResetsBackoff(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + attempts := 0 + + _ = retryLoop(context.Background(), policy, func(context.Context) (bool, error) { + attempts++ + switch attempts { + case 3: + // Ran fine for a day before dropping, so this is a fresh outage + // rather than a continuation of the previous run of failures. + clock.advance(24 * time.Hour) + return true, errors.New("read error: connection reset") + case 4: + return true, nil + default: + return false, errors.New("read error: connection reset") + } + }) + + want := []time.Duration{5 * time.Second, 10 * time.Second, 5 * time.Second} + if len(clock.delays) != len(want) { + t.Fatalf("expected waits %v, got %v", want, clock.delays) + } + for i, w := range want { + if clock.delays[i] != w { + t.Errorf("wait %d: expected %v, got %v", i, w, clock.delays[i]) + } + } +} + +func TestRetryLoop_HealthyConnectionWithoutDataDoesNotReset(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + attempts := 0 + + _ = retryLoop(context.Background(), policy, func(context.Context) (bool, error) { + attempts++ + // Connects and stays open, but never delivers a measurement. That is not + // a healthy stream, so the backoff must keep growing. + clock.advance(24 * time.Hour) + if attempts == 3 { + return false, nil + } + return false, errors.New("read error: connection reset") + }) + + want := []time.Duration{5 * time.Second, 10 * time.Second} + if len(clock.delays) != len(want) { + t.Fatalf("expected waits %v, got %v", want, clock.delays) + } + for i, w := range want { + if clock.delays[i] != w { + t.Errorf("wait %d: expected %v, got %v", i, w, clock.delays[i]) + } + } +} + +func TestRetryLoop_CancelledContextStopsImmediately(t *testing.T) { + t.Run("during an attempt", func(t *testing.T) { + clock := newFakeClock() + ctx, cancel := context.WithCancel(context.Background()) + attempts := 0 + + err := retryLoop(ctx, testPolicy(clock), func(context.Context) (bool, error) { + attempts++ + cancel() + return true, fmt.Errorf("read error: %w", context.Canceled) + }) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if attempts != 1 { + t.Errorf("expected 1 attempt, got %d", attempts) + } + if len(clock.delays) != 0 { + t.Errorf("expected no backoff wait after cancellation, got %v", clock.delays) + } + }) + + t.Run("during a backoff wait", func(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + // Ctrl+C has to land during the wait, not five minutes later. + policy.sleep = func(context.Context, time.Duration) error { return context.Canceled } + attempts := 0 + + err := retryLoop(context.Background(), policy, func(context.Context) (bool, error) { + attempts++ + return false, errors.New("read error: connection reset") + }) + + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if attempts != 1 { + t.Errorf("expected 1 attempt, got %d", attempts) + } + }) +} + +func TestRetryLoop_NotifiesBeforeEachWait(t *testing.T) { + clock := newFakeClock() + policy := testPolicy(clock) + policy.MaxPerHour = 3 + + type notice struct { + attempt int + delay time.Duration + cause string + } + var notices []notice + policy.Notify = func(attempt int, delay time.Duration, cause error) { + notices = append(notices, notice{attempt, delay, cause.Error()}) + } + + attempts := 0 + _ = retryLoop(context.Background(), policy, alwaysFails(t, policy.MaxPerHour, &attempts)) + + want := []notice{ + {1, 5 * time.Second, "read error: connection reset"}, + {2, 10 * time.Second, "read error: connection reset"}, + } + if len(notices) != len(want) { + t.Fatalf("expected %d notices, got %d: %v", len(want), len(notices), notices) + } + for i, w := range want { + if notices[i] != w { + t.Errorf("notice %d: expected %+v, got %+v", i, w, notices[i]) + } + } +} + +func TestSleepCtx_ReturnsWhenCancelled(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + start := time.Now() + if err := sleepCtx(ctx, time.Hour); !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("expected an immediate return, waited %v", elapsed) + } +} + +func TestRetryable(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"read error", errors.New("read error: connection reset by peer"), true}, + {"abnormal close", fmt.Errorf("read error: %w", websocket.CloseError{Code: websocket.StatusAbnormalClosure}), true}, + {"going away", fmt.Errorf("read error: %w", websocket.CloseError{Code: websocket.StatusGoingAway}), true}, + {"ack timeout", fmt.Errorf("failed to read ack: %w", context.DeadlineExceeded), true}, + {"invalid token", fmt.Errorf("read error: %w", websocket.CloseError{Code: statusInvalidToken, Reason: "Invalid token"}), false}, + {"invalid home ID", markFatal(errors.New("invalid or non-existing home ID")), false}, + {"wrapped fatal", fmt.Errorf("stream: %w", markFatal(errors.New("subscription error"))), false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := retryable(tt.err); got != tt.want { + t.Errorf("retryable(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +func TestBackoff(t *testing.T) { + tests := []struct { + n int + want time.Duration + }{ + {0, 5 * time.Second}, + {1, 10 * time.Second}, + {4, 80 * time.Second}, + {6, 5 * time.Minute}, + {100, 5 * time.Minute}, // must cap rather than overflow + } + + for _, tt := range tests { + if got := backoff(defaultBaseDelay, defaultMaxDelay, tt.n); got != tt.want { + t.Errorf("backoff(n=%d) = %v, want %v", tt.n, got, tt.want) + } + } +} + +func TestEqualJitter_StaysWithinHalfAndFull(t *testing.T) { + const d = 40 * time.Second + + for i := 0; i < 200; i++ { + got := equalJitter(d) + if got < d/2 || got >= d { + t.Fatalf("equalJitter(%v) = %v, want [%v, %v)", d, got, d/2, d) + } + } + + if got := equalJitter(0); got != 0 { + t.Errorf("equalJitter(0) = %v, want 0", got) + } + if got := equalJitter(time.Nanosecond); got != time.Nanosecond { + t.Errorf("equalJitter(1ns) = %v, want 1ns", got) + } +} + +func TestPruneWindow(t *testing.T) { + now := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + window := []time.Time{ + now.Add(-90 * time.Minute), // stale + now.Add(-time.Hour), // exactly an hour old: stale + now.Add(-30 * time.Minute), + now, + } + + got := pruneWindow(window, now) + if len(got) != 2 { + t.Fatalf("expected 2 attempts left in the window, got %d", len(got)) + } + if !got[0].Equal(now.Add(-30*time.Minute)) || !got[1].Equal(now) { + t.Errorf("wrong entries kept: %v", got) + } +} + +func TestReconnectPolicy_WithDefaults(t *testing.T) { + p := ReconnectPolicy{}.withDefaults() + + if p.BaseDelay != defaultBaseDelay || p.MaxDelay != defaultMaxDelay { + t.Errorf("expected default delays, got base=%v max=%v", p.BaseDelay, p.MaxDelay) + } + if p.MaxPerHour != defaultMaxPerHour || p.HealthyAfter != defaultHealthyAfter { + t.Errorf("expected default budget, got %d per hour, healthy after %v", p.MaxPerHour, p.HealthyAfter) + } + if p.now == nil || p.sleep == nil || p.jitter == nil { + t.Error("expected the clock and jitter seams to be filled in") + } + + // A ceiling below the base would make the backoff shrink. + clamped := ReconnectPolicy{BaseDelay: time.Minute, MaxDelay: time.Second}.withDefaults() + if clamped.MaxDelay != time.Minute { + t.Errorf("expected MaxDelay clamped up to BaseDelay, got %v", clamped.MaxDelay) + } +} diff --git a/internal/api/websocket.go b/internal/api/websocket.go index 0c72124..dcb3b21 100644 --- a/internal/api/websocket.go +++ b/internal/api/websocket.go @@ -27,13 +27,16 @@ const ( type LiveClient struct { token string homeID string + // endpoint is WebSocketEndpoint unless a test points it at a local server. + endpoint string } // NewLiveClient creates a new WebSocket client for live data func NewLiveClient(token, homeID string) *LiveClient { return &LiveClient{ - token: token, - homeID: homeID, + token: token, + homeID: homeID, + endpoint: WebSocketEndpoint, } } @@ -50,14 +53,28 @@ func (c *LiveClient) Subscribe(ctx context.Context, handler func(*models.LiveMea headers := http.Header{} headers.Set("User-Agent", UserAgent) - conn, _, err := websocket.Dial(ctx, WebSocketEndpoint, &websocket.DialOptions{ + conn, resp, err := websocket.Dial(ctx, c.endpoint, &websocket.DialOptions{ Subprotocols: []string{"graphql-transport-ws"}, HTTPHeader: headers, }) if err != nil { + // A refused handshake is a credentials or endpoint problem rather than a + // blip, and would be refused identically on every reconnect. + if resp != nil && (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) { + return markFatal(fmt.Errorf("failed to connect: %w", err)) + } return fmt.Errorf("failed to connect: %w", err) } - defer func() { _ = conn.Close(websocket.StatusNormalClosure, "") }() // (websocket.StatusNormalClosure, "") + defer func() { + if ctx.Err() != nil { + // The caller is already shutting down. A close handshake would make + // Ctrl+C wait up to five seconds for a reply the server may never + // send, so drop the socket instead. + _ = conn.CloseNow() + return + } + _ = conn.Close(websocket.StatusNormalClosure, "") + }() // Send connection_init initPayload, _ := json.Marshal(map[string]string{ @@ -112,13 +129,15 @@ func (c *LiveClient) Subscribe(ctx context.Context, handler func(*models.LiveMea case "next": measurement, err := c.parsePayload(msg.Payload) if err != nil { - return err + // The subscription itself is wrong (bad home ID, GraphQL + // error): a fresh connection gets the same payload back. + return markFatal(err) } if err := handler(measurement); err != nil { - return err + return markFatal(err) } case "error": - return fmt.Errorf("subscription error: %s", string(msg.Payload)) + return markFatal(fmt.Errorf("subscription error: %s", string(msg.Payload))) case "complete": return nil } @@ -148,6 +167,10 @@ func (c *LiveClient) waitForAck(ctx context.Context, conn *websocket.Conn) error return fmt.Errorf("failed to parse ack: %w", err) } + if msg.Type == "connection_error" { + return markFatal(fmt.Errorf("connection rejected: %s", string(msg.Payload))) + } + if msg.Type != "connection_ack" { return fmt.Errorf("expected connection_ack, got %s", msg.Type) } diff --git a/internal/commands/live.go b/internal/commands/live.go index 77b7625..58ca853 100644 --- a/internal/commands/live.go +++ b/internal/commands/live.go @@ -6,6 +6,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/spf13/cobra" @@ -14,7 +15,8 @@ import ( ) var ( - liveHomeID string + liveHomeID string + liveNoReconnect bool ) var liveCmd = &cobra.Command{ @@ -23,7 +25,13 @@ var liveCmd = &cobra.Command{ Long: `Stream live power consumption data from your Tibber Pulse. Requires a Tibber Pulse device connected to your home. -Press Ctrl+C to stop the stream.`, +Press Ctrl+C to stop the stream. + +A dropped connection is retried with a growing delay, so the stream survives a +transient network failure. Errors that a reconnect cannot fix - an invalid token +or an unknown home ID - still exit immediately. Tibber allows 20 WebSocket +connections per hour, and reconnects stay well inside that; use --no-reconnect +to exit on the first error instead.`, Run: func(cmd *cobra.Command, args []string) { if err := cfg.Validate(); err != nil { exitWithError("%v", err) @@ -69,7 +77,7 @@ Press Ctrl+C to stop the stream.`, fmt.Fprintf(os.Stderr, "Connecting to live stream...\n") - err := liveClient.Subscribe(ctx, func(m *models.LiveMeasurement) error { + handler := func(m *models.LiveMeasurement) error { // Clear screen for markdown, just print for JSON if cfg.Format == "json" { fmt.Println(formatter.FormatLiveMeasurement(m)) @@ -79,7 +87,21 @@ Press Ctrl+C to stop the stream.`, fmt.Println(formatter.FormatLiveMeasurement(m)) } return nil - }) + } + + var err error + if liveNoReconnect { + err = liveClient.Subscribe(ctx, handler) + } else { + policy := api.DefaultReconnectPolicy() + policy.Notify = func(attempt int, delay time.Duration, cause error) { + // stderr only - stdout carries the measurements, which + // --format json callers pipe straight into jq. + fmt.Fprintf(os.Stderr, "Connection lost: %v\nReconnecting in %s (attempt %d)...\n", + cause, delay.Round(time.Second), attempt) + } + err = liveClient.SubscribeWithReconnect(ctx, policy, handler) + } if err != nil && ctx.Err() == nil { exitWithError("Stream error: %v", err) @@ -89,5 +111,6 @@ Press Ctrl+C to stop the stream.`, func init() { liveCmd.Flags().StringVar(&liveHomeID, "home-id", "", "specific home ID to monitor") + liveCmd.Flags().BoolVar(&liveNoReconnect, "no-reconnect", false, "exit on the first stream error instead of reconnecting") rootCmd.AddCommand(liveCmd) }