From 41bb22a3e3da926da8a62ebc60aaef81563eb23f Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sat, 17 Oct 2026 09:04:42 +0200 Subject: [PATCH 1/2] feat: move fleet totals and browser offline notices to FTW Cloud Signed-off-by: Fredrik Ahlgren --- CLAUDE.md | 4 +- README.md | 16 +++ cmd/ftwcloudd/main.go | 32 ++++- go.mod | 2 +- go.sum | 2 + internal/fleet/fleet.go | 93 +++++++++++++++ internal/fleet/fleet_test.go | 84 ++++++++++++++ internal/push/browser_test.go | 77 ++++++++++++ internal/push/push.go | 54 ++++++--- internal/registry/fleet.go | 72 ++++++++++++ internal/registry/webpush.go | 72 ++++++++++++ internal/server/server.go | 8 +- internal/server/server_test.go | 5 + internal/uplink/uplink.go | 3 + internal/webpush/webpush.go | 193 +++++++++++++++++++++++++++++++ internal/webpush/webpush_test.go | 152 ++++++++++++++++++++++++ 16 files changed, 841 insertions(+), 28 deletions(-) create mode 100644 internal/fleet/fleet.go create mode 100644 internal/fleet/fleet_test.go create mode 100644 internal/push/browser_test.go create mode 100644 internal/registry/fleet.go create mode 100644 internal/registry/webpush.go create mode 100644 internal/webpush/webpush.go create mode 100644 internal/webpush/webpush_test.go diff --git a/CLAUDE.md b/CLAUDE.md index 2f6c4ee..aeda985 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -35,7 +35,9 @@ maintains it. - **No readings on disk.** Memory holds readings, and only while the roster's grant serves a copy; SQLite holds only the home id, the box key, the signed roster, owners' suspensions, tombstones, APNs registrations, the hashes of - extension tokens and when each away box's alert is due. Nothing logs + extension tokens when each away box's alert is due, browser notice sent markers and + anonymous daily fleet totals. Browser endpoints and box-encrypted offline + notices live only in the signed roster. Nothing logs addresses. - **A phone can only narrow, or talk to its own box.** On `/v1/app` it can say whether it is looking, an owner can suspend a roster device, and a diff --git a/README.md b/README.md index 555e5dc..ddc22db 100644 --- a/README.md +++ b/README.md @@ -364,3 +364,19 @@ with `go list -deps`, so no control code can reach this build. `ftwcloudd` is licensed under the GNU Affero General Public License, version 3 only (AGPL-3.0-only). See [LICENSE](LICENSE) and [CONTRIBUTING.md](CONTRIBUTING.md). + +Anonymous box reports use `POST /v1/fleet`. It accepts only the seven +`ftw.fleet/1` fields and reduces each report into bounded daily counts for +90 days. It stores no raw report, address or home id. An operator can read +`/fleet/stats` only on the loopback metrics listener, with +`docker exec ftw-cloud-ftwcloudd-1 ftwcloudd fleet`. These counts measure +reports, not unique boxes. + +Browser offline notices use the box's optional signed roster rows. The box +encrypts each notice and signs a short-lived VAPID header; the service holds +no browser encryption key or box signing key. It uses the APNs outage timer, +including update grace and deploy drain, and checks the current device epoch, +events scope and suspension before sending. A readable-copy grant is not +needed. The roster holds the delivery endpoints and ciphertext; a separate +per-home row holds only the current outage's sent ids. Purge removes both. +The sender refuses redirects, proxies and local destinations. diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index 0c18d6d..c6dc146 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -37,6 +37,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/apns" "github.com/srcfl/ftw-cloud/internal/appsession" + "github.com/srcfl/ftw-cloud/internal/fleet" "github.com/srcfl/ftw-cloud/internal/keys" "github.com/srcfl/ftw-cloud/internal/mcp" "github.com/srcfl/ftw-cloud/internal/metrics" @@ -49,6 +50,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/token" "github.com/srcfl/ftw-cloud/internal/uplink" + "github.com/srcfl/ftw-cloud/internal/webpush" "github.com/srcfl/ftw/go/protocol/appwire" "github.com/srcfl/ftw/go/protocol/cloudwire" "golang.org/x/crypto/acme/autocert" @@ -66,7 +68,7 @@ func main() { } return } - if len(os.Args) > 1 && os.Args[1] == "metrics" { + if len(os.Args) > 1 && (os.Args[1] == "metrics" || os.Args[1] == "fleet") { if err := printMetrics(os.Args[2:]); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) @@ -146,9 +148,13 @@ func serve(args []string) error { } else if n > 0 { log.Info("rosters restored from the registry", "homes", n, "took", time.Since(restoreStart).Round(time.Millisecond)) } + fleetSvc, err := fleet.New(reg) + if err != nil { + return err + } tokens := token.New(st, reg, log) tokens.Shards = shards - srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Log: log} + srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Fleet: fleetSvc, Log: log} // Assistants: the authorization server, with agent keys in ftwkeysd. agentKeys := keys.Dial(*keysSocket) @@ -183,7 +189,12 @@ func serve(args []string) error { default: sender = client } + webPushSvc, err := webpush.New(reg) + if err != nil { + return err + } pushSvc := push.New(reg, sender, log) + pushSvc.BrowserUnreachable = webPushSvc.Send pushSvc.WatchApp = *apnsWatch pushSvc.Owns = shards.Ours up.OnHeld = pushSvc.Held @@ -301,7 +312,7 @@ func serve(args []string) error { } } if *metricsAddr != "" { - if err := serveMetrics(ctx, *metricsAddr); err != nil { + if err := serveMetrics(ctx, *metricsAddr, fleetSvc.Stats); err != nil { return err } } @@ -402,7 +413,7 @@ func restartSpread(sockets, rate int, floor time.Duration) time.Duration { const defaultMetricsAddr = "127.0.0.1:9464" // serveMetrics answers /metrics on a loopback address until ctx ends. -func serveMetrics(ctx context.Context, addr string) error { +func serveMetrics(ctx context.Context, addr string, fleetStats ...http.HandlerFunc) error { host, _, err := net.SplitHostPort(addr) if err != nil { return fmt.Errorf("metrics-listen: %w", err) @@ -414,7 +425,12 @@ func serveMetrics(ctx context.Context, addr string) error { if err != nil { return err } - m := &http.Server{Handler: metrics.Default, ReadHeaderTimeout: 5 * time.Second, ErrorLog: log0()} + mux := http.NewServeMux() + mux.Handle("/metrics", metrics.Default) + if len(fleetStats) > 0 { + mux.HandleFunc("/fleet/stats", fleetStats[0]) + } + m := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ErrorLog: log0()} go func() { _ = m.Serve(ln) }() go func() { <-ctx.Done() @@ -429,7 +445,11 @@ func printMetrics(args []string) error { addr := fs.String("addr", defaultMetricsAddr, "where the service answers /metrics") _ = fs.Parse(args) c := &http.Client{Timeout: 5 * time.Second} - resp, err := c.Get("http://" + *addr + "/metrics") + path := "/metrics" + if len(os.Args) > 1 && os.Args[1] == "fleet" { + path = "/fleet/stats" + } + resp, err := c.Get("http://" + *addr + path) if err != nil { return err } diff --git a/go.mod b/go.mod index c461f43..af38224 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/fxamacker/cbor/v2 v2.9.4 github.com/gorilla/websocket v1.5.3 github.com/quic-go/quic-go v0.63.0 - github.com/srcfl/ftw/go/protocol v0.0.0-20261016102129-7fb84d6ed573 + github.com/srcfl/ftw/go/protocol v0.0.0-20261017070253-cbdd8a587105 golang.org/x/crypto v0.57.0 golang.org/x/net v0.58.0 golang.org/x/sys v0.48.0 diff --git a/go.sum b/go.sum index 0398da6..a605893 100644 --- a/go.sum +++ b/go.sum @@ -54,6 +54,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/srcfl/ftw/go/protocol v0.0.0-20261016102129-7fb84d6ed573 h1:1OOC4RdeERT9r6cLMTlcmTjg8+ML2kU6hCvhx/x+8xI= github.com/srcfl/ftw/go/protocol v0.0.0-20261016102129-7fb84d6ed573/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/srcfl/ftw/go/protocol v0.0.0-20261017070253-cbdd8a587105 h1:56AF3FQxXwfTVacD+AtJQM96TdzntTeQy1f4yiN54ic= +github.com/srcfl/ftw/go/protocol v0.0.0-20261017070253-cbdd8a587105/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= diff --git a/internal/fleet/fleet.go b/internal/fleet/fleet.go new file mode 100644 index 0000000..946e445 --- /dev/null +++ b/internal/fleet/fleet.go @@ -0,0 +1,93 @@ +// Package fleet reduces anonymous box reports to bounded daily counts. +package fleet + +import ( + "encoding/json" + "io" + "net/http" + "sync" + "time" + + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol/fleetwire" +) + +type Service struct { + reg *registry.Registry + now func() time.Time + mu sync.Mutex + at time.Time + tokens float64 +} + +func New(reg *registry.Registry) (*Service, error) { + if err := reg.FleetInit(); err != nil { + return nil, err + } + return &Service{reg: reg, now: time.Now, tokens: 120}, nil +} + +// ServeHTTP accepts reports only. Public requests never read totals. +func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if !s.allow() { + w.WriteHeader(http.StatusTooManyRequests) + return + } + data, err := io.ReadAll(http.MaxBytesReader(w, r.Body, fleetwire.MaxBodyBytes)) + if err != nil { + w.WriteHeader(http.StatusRequestEntityTooLarge) + return + } + report, err := fleetwire.Decode(data) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + ok, err := s.reg.FleetPut(report, s.now()) + switch { + case err != nil: + w.WriteHeader(http.StatusServiceUnavailable) + case !ok: + w.WriteHeader(http.StatusTooManyRequests) + default: + w.WriteHeader(http.StatusNoContent) + } +} + +// Stats belongs only on the operator's loopback listener. +func (s *Service) Stats(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + view, err := s.reg.FleetView() + if err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(view) +} + +// One shared budget stores no addresses and admits five reports a second. +func (s *Service) allow() bool { + s.mu.Lock() + defer s.mu.Unlock() + now := s.now() + if !s.at.IsZero() { + s.tokens = min(120, s.tokens+max(0, now.Sub(s.at).Seconds())*5) + } + s.at = now + if s.tokens < 1 { + return false + } + s.tokens-- + return true +} diff --git a/internal/fleet/fleet_test.go b/internal/fleet/fleet_test.go new file mode 100644 index 0000000..b8d8747 --- /dev/null +++ b/internal/fleet/fleet_test.go @@ -0,0 +1,84 @@ +package fleet + +import ( + "bytes" + "encoding/json" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol/fleetwire" + "net/http" + "net/http/httptest" + "path/filepath" + "sync" + "testing" + "time" +) + +func TestReportsBecomeOnlyBoundedDailyTotalsAcrossARestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "registry.db") + reg, err := registry.Open(path) + if err != nil { + t.Fatal(err) + } + svc, err := New(reg) + if err != nil { + t.Fatal(err) + } + now := time.Date(2026, 10, 17, 0, 0, 0, 0, time.UTC) + svc.now = func() time.Time { return now } + report := fleetwire.Report{Schema: fleetwire.Schema, Version: "v0.146.0", Channel: "stable", Drivers: []string{}, Battery: "none", Zone: "SE4", Age: "unknown"} + raw, _ := json.Marshal(report) + var wg sync.WaitGroup + for range 100 { + wg.Go(func() { + w := httptest.NewRecorder() + r := httptest.NewRequest("POST", "/v1/fleet", bytes.NewReader(raw)) + r.RemoteAddr = "198.51.100.8:443" + svc.ServeHTTP(w, r) + if w.Code != 204 { + t.Errorf("status %d", w.Code) + } + }) + } + wg.Wait() + if err := reg.Close(); err != nil { + t.Fatal(err) + } + reg, err = registry.Open(path) + if err != nil { + t.Fatal(err) + } + defer reg.Close() + svc, err = New(reg) + if err != nil { + t.Fatal(err) + } + view, err := reg.FleetView() + if err != nil || len(view.Days) != 1 || view.Days[0].Reports != 100 { + t.Fatal(view, err) + } + for i := 1; i < 92; i++ { + if ok, err := reg.FleetPut(report, now.AddDate(0, 0, i)); err != nil || !ok { + t.Fatal(err) + } + } + view, _ = reg.FleetView() + if len(view.Days) != 90 { + t.Fatal(len(view.Days)) + } + for _, tc := range []struct { + method, body string + code int + }{{"GET", "", 405}, {"POST", `{"home":"x"}`, 400}, {"POST", string(make([]byte, 4097)), 413}} { + w := httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest(tc.method, "/v1/fleet", bytes.NewBufferString(tc.body))) + if w.Code != tc.code { + t.Fatal(w.Code, tc) + } + } + w := httptest.NewRecorder() + svc.Stats(w, httptest.NewRequest(http.MethodGet, "/fleet/stats", nil)) + if w.Code != 200 || bytes.Contains(w.Body.Bytes(), []byte("198.51.100")) { + t.Fatal(w.Code, w.Body.String()) + } + // The only extra table holds date and aggregate JSON, not source addresses. +} diff --git a/internal/push/browser_test.go b/internal/push/browser_test.go new file mode 100644 index 0000000..ea20964 --- /dev/null +++ b/internal/push/browser_test.go @@ -0,0 +1,77 @@ +package push + +import ( + "context" + "github.com/srcfl/ftw/go/protocol/cloudwire" + "testing" + "time" +) + +func TestBrowserOfflineNoticesUseTheExistingTimerWithoutAnAPNsKey(t *testing.T) { + r := newRig(t) + r.svc.sender = nil + calls := 0 + retry := true + r.svc.BrowserUnreachable = func(context.Context, string) (int, bool) { + calls++ + if retry { + return 0, true + } + return 1, false + } + r.connected() + r.gone("") + r.clock.advance(599 * time.Second) + r.drain() + if calls != 0 { + t.Fatal("before grace") + } + r.clock.advance(2 * time.Second) + r.drain() + if calls != 1 { + t.Fatal(calls) + } + retry = false + r.clock.advance(10 * time.Second) + r.drain() + if calls != 2 { + t.Fatal("browser retry", calls) + } + r.clock.advance(time.Hour) + r.drain() + if calls != 2 { + t.Fatal("duplicate", calls) + } + r.connected() + r.gone(cloudwire.ByeUpdate) + r.clock.advance(11 * time.Minute) + r.drain() + if calls != 2 { + t.Fatal("ignored update grace") + } + r.connected() + r.clock.advance(time.Hour) + r.drain() + if calls != 2 { + t.Fatal("sent after return") + } +} + +func TestBrowserRetryDoesNotResendAnAcceptedAppleAlert(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "token") + calls := 0 + r.svc.BrowserUnreachable = func(context.Context, string) (int, bool) { calls++; return 0, calls == 1 } + r.connected() + r.gone("") + r.clock.advance(601 * time.Second) + r.drain() + if len(r.apple.notifications()) != 1 || calls != 1 { + t.Fatal("first delivery") + } + r.clock.advance(10 * time.Second) + r.drain() + if len(r.apple.notifications()) != 1 || calls != 2 { + t.Fatal("resending accepted APNs", calls) + } +} diff --git a/internal/push/push.go b/internal/push/push.go index 6313e18..122d005 100644 --- a/internal/push/push.go +++ b/internal/push/push.go @@ -181,6 +181,9 @@ type Service struct { // registered with app watch. Empty sends them nothing. The iPhone app's // is the sender's own. Set before serving. WatchApp string + // BrowserUnreachable delivers the box-encrypted browser notices on the + // same outage timer. It needs no APNs key or readable-copy grant. + BrowserUnreachable func(context.Context, string) (int, bool) // Owns reports whether this shard holds a home. Resume starts no // box.unreachable for a home another shard holds, whose rows may still // be here during a move. Nil holds every home. Set before Resume. @@ -822,51 +825,64 @@ func (s *Service) roster(home string) (*cloudwire.Roster, error) { // showing, and only to devices that may still get it; a gone token is // forgotten. func (s *Service) sendAlert(ctx context.Context, job alertJob) { - if s.sender == nil { + if s.sender == nil && (!job.unreachable || s.BrowserUnreachable == nil) { return } + if job.unreachable { + due, err := s.reg.WebPushOutage(job.home) + if err != nil { + s.retryAlert(job, nil) + return + } + if due == 0 || due > s.now().UnixMilli() { + return + } + } fresh, err := s.regs(job.home, cloudwire.PushKindAlert, job.notice) if err == nil && len(fresh) > 0 { fresh, err = s.audience(job.home, job.notice, fresh) } if err != nil { - s.log.Warn("could not read push registrations; the alert goes again", "err", err) + s.log.Warn("could not read push registrations; the alert goes again") s.retryAlert(job, nil) return } if job.tries == 0 { job.regs = fresh - if len(job.regs) == 0 { - s.told(job) - return - } - // A notice was counted when it came. if !job.notice && !s.allow(job.home, cloudwire.PushKindAlert) { pushCapped.Inc() - s.log.Warn("a home passed its push cap; push dropped", "kind", cloudwire.PushKindAlert, "cap_per_hour", MaxAlertsPerHour) s.told(job) return } } else { - // A device suspended, dropped by a roster or unregistered since - // the last try gets nothing. job.regs = slices.DeleteFunc(job.regs, func(r registry.PushReg) bool { return !slices.ContainsFunc(fresh, func(f registry.PushReg) bool { return bytes.Equal(f.Token, r.Token) }) }) - if len(job.regs) == 0 { - s.told(job) + } + job.tries++ + took := 0 + browserRetry := false + if job.unreachable && s.BrowserUnreachable != nil { + n, retry := s.BrowserUnreachable(ctx, job.home) + took += n + browserRetry = retry + } + again := []registry.PushReg{} + if s.sender != nil && len(job.regs) > 0 { + if !s.breaker.wait(ctx) { return } + n, remaining := s.deliver(ctx, job.home, cloudwire.PushKindAlert, job.payload, job.expires, job.regs) + took += n + again = remaining } - if !s.breaker.wait(ctx) { - return - } - job.tries++ - took, again := s.deliver(ctx, job.home, cloudwire.PushKindAlert, job.payload, job.expires, job.regs) - if took > 0 || len(again) == 0 { + if took > 0 || (len(again) == 0 && !browserRetry) { s.told(job) } - if len(again) > 0 { + if len(again) > 0 || browserRetry { + if again == nil { + again = []registry.PushReg{} + } s.retryAlert(job, again) } } diff --git a/internal/registry/fleet.go b/internal/registry/fleet.go new file mode 100644 index 0000000..1c962f8 --- /dev/null +++ b/internal/registry/fleet.go @@ -0,0 +1,72 @@ +package registry + +import ( + "database/sql" + "encoding/json" + "errors" + "time" + + "github.com/srcfl/ftw/go/protocol/fleetwire" +) + +// FleetInit creates only daily totals. No report or address goes on disk. +func (r *Registry) FleetInit() error { + _, err := r.db.Exec(`CREATE TABLE IF NOT EXISTS fleet_days (date TEXT PRIMARY KEY, totals BLOB NOT NULL)`) + return err +} + +// FleetPut reduces a report into today's counters in one transaction. +func (r *Registry) FleetPut(report fleetwire.Report, now time.Time) (bool, error) { + tx, err := r.db.Begin() + if err != nil { + return false, err + } + defer tx.Rollback() + date := now.UTC().Format(time.DateOnly) + day := fleetwire.NewDay(date) + var old []byte + err = tx.QueryRow(`SELECT totals FROM fleet_days WHERE date = ?`, date).Scan(&old) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return false, err + } + if len(old) > 0 { + if err := json.Unmarshal(old, &day); err != nil { + return false, err + } + } + if !day.Add(report) { + return false, nil + } + data, err := json.Marshal(day) + if err != nil { + return false, err + } + if _, err := tx.Exec(`INSERT INTO fleet_days(date, totals) VALUES (?, ?) ON CONFLICT(date) DO UPDATE SET totals=excluded.totals`, date, data); err != nil { + return false, err + } + if _, err := tx.Exec(`DELETE FROM fleet_days WHERE date NOT IN (SELECT date FROM fleet_days ORDER BY date DESC LIMIT ?)`, fleetwire.MaxDays); err != nil { + return false, err + } + return true, tx.Commit() +} + +func (r *Registry) FleetView() (fleetwire.View, error) { + view := fleetwire.View{Schema: fleetwire.StatsSchema, Meaning: fleetwire.Meaning, Days: []fleetwire.Day{}} + rows, err := r.rd.Query(`SELECT totals FROM fleet_days ORDER BY date DESC LIMIT ?`, fleetwire.MaxDays) + if err != nil { + return view, err + } + defer rows.Close() + for rows.Next() { + var data []byte + var day fleetwire.Day + if err := rows.Scan(&data); err != nil { + return view, err + } + if err := json.Unmarshal(data, &day); err != nil { + return view, err + } + view.Days = append(view.Days, day) + } + return view, rows.Err() +} diff --git a/internal/registry/webpush.go b/internal/registry/webpush.go new file mode 100644 index 0000000..15f1f5e --- /dev/null +++ b/internal/registry/webpush.go @@ -0,0 +1,72 @@ +package registry + +import ( + "database/sql" + "encoding/json" + "errors" +) + +// Browser notices live in the signed roster. This table holds only which +// notices went in the current outage, so an exit does not send them twice. +func (r *Registry) WebPushInit() error { + _, err := r.db.Exec(`CREATE TABLE IF NOT EXISTS webpush_sent (home TEXT PRIMARY KEY REFERENCES homes(home) ON DELETE CASCADE, due_ms INTEGER NOT NULL, ids BLOB NOT NULL)`) + return err +} + +// WebPushOutage is zero once the box is back. It is stable across a restart. +func (r *Registry) WebPushOutage(home string) (int64, error) { + var due int64 + err := r.rd.QueryRow(`SELECT due_ms FROM away WHERE home=?`, home).Scan(&due) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + return due, err +} + +func (r *Registry) WebPushSent(home string, due int64) ([]string, error) { + var raw []byte + err := r.rd.QueryRow(`SELECT ids FROM webpush_sent WHERE home=? AND due_ms=?`, home, due).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + var ids []string + err = json.Unmarshal(raw, &ids) + return ids, err +} + +func (r *Registry) WebPushMark(home string, due int64, id string) error { + tx, err := r.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + var raw []byte + err = tx.QueryRow(`SELECT ids FROM webpush_sent WHERE home=? AND due_ms=?`, home, due).Scan(&raw) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + var ids []string + if len(raw) > 0 { + if err = json.Unmarshal(raw, &ids); err != nil { + return err + } + } + for _, old := range ids { + if old == id { + return nil + } + } + ids = append(ids, id) + raw, err = json.Marshal(ids) + if err != nil { + return err + } + _, err = tx.Exec(`INSERT INTO webpush_sent(home,due_ms,ids) SELECT ?,?,? WHERE EXISTS(SELECT 1 FROM away WHERE home=? AND due_ms=?) ON CONFLICT(home) DO UPDATE SET due_ms=excluded.due_ms,ids=excluded.ids`, home, due, raw, home, due) + if err != nil { + return err + } + return tx.Commit() +} diff --git a/internal/server/server.go b/internal/server/server.go index 2b6a66e..4b222e3 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -37,7 +37,7 @@ import ( const SourceArchive = "/usr/share/ftw-cloud/source.tar.gz" // Routes are every path the service answers. -var Routes = append([]string{"/v1/box", "/v1/app", "/v1/app/once", "/v1/now", "/source", "/healthz", "/mcp"}, oauth.Routes...) +var Routes = append([]string{"/v1/box", "/v1/app", "/v1/app/once", "/v1/now", "/v1/fleet", "/source", "/healthz", "/mcp"}, oauth.Routes...) // devRoutes adds a local inspection route in binaries built with // -tags ftwclouddev. It is nil in every other build. @@ -57,6 +57,7 @@ type Server struct { // OAuth is the authorization server for assistants, and MCP the server // they call. Nil serves none. OAuth *oauth.Server + Fleet http.Handler MCP http.Handler Log *slog.Logger // AltSvc, when set, is the Alt-Svc header that points TCP clients at @@ -88,6 +89,11 @@ func (s *Server) Handler() http.Handler { if s.MCP != nil { mux.Handle("/mcp", s.MCP) } + if s.Fleet != nil { + mux.Handle("/v1/fleet", s.Fleet) + } else { + mux.HandleFunc("/v1/fleet", notYet) + } mux.HandleFunc("/source", source) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store") diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 9d6e9d8..7390fec 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -14,6 +14,7 @@ import ( "testing" "github.com/srcfl/ftw-cloud/internal/appsession" + "github.com/srcfl/ftw-cloud/internal/fleet" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/store" "github.com/srcfl/ftw-cloud/internal/token" @@ -37,6 +38,10 @@ func newServer(t *testing.T) (*Server, string) { s := &Server{Store: st, Registry: reg, Uplink: up, App: appsession.New(st, reg, up, noise, log), Tokens: token.New(st, reg, log), Log: log} + s.Fleet, err = fleet.New(reg) + if err != nil { + t.Fatal(err) + } // A home with readings in it, held the way a box's uplink holds them. const home = "canaryhomecanaryhomecanary" box, _ := st.For(store.Principal{Kind: store.Box, Home: home}) diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index 096aa17..4a2b729 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -869,6 +869,9 @@ func (c *boxConn) roster(env cloudwire.Envelope) error { } // A grant has three shapes: a copy with serve, a withdrawn one with its // id alone, or none at all. + if err := cloudwire.ValidateWebPush(r.WebPush); err != nil { + return fmt.Errorf("%w: browser notice rows", errProtocol) + } g := r.Grant on := len(g.ID) == cloudwire.GrantIDBytes && slices.Contains(g.Purposes, protocol.CloudPurposeServe) bare := len(g.Scopes) == 0 && len(g.Purposes) == 0 diff --git a/internal/webpush/webpush.go b/internal/webpush/webpush.go new file mode 100644 index 0000000..c332907 --- /dev/null +++ b/internal/webpush/webpush.go @@ -0,0 +1,193 @@ +// Package webpush delivers only the box's pre-encrypted offline notices. +package webpush + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "errors" + "io" + "net" + "net/http" + "net/netip" + "net/url" + "slices" + "strconv" + "strings" + "time" + + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +type Service struct { + reg *registry.Registry + client *http.Client + now func() time.Time +} + +func New(reg *registry.Registry) (*Service, error) { + if err := reg.WebPushInit(); err != nil { + return nil, err + } + // No proxy and no redirects. The resolved address itself is checked on + // every dial, so neither a URL nor DNS can reach a local service. + transport := &http.Transport{DialContext: publicDial, ForceAttemptHTTP2: true, TLSHandshakeTimeout: 5 * time.Second, ResponseHeaderTimeout: 5 * time.Second, MaxIdleConns: 8, IdleConnTimeout: time.Minute} + return &Service{reg: reg, now: time.Now, client: &http.Client{Transport: transport, Timeout: 10 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}}, nil +} + +func publicIP(ip netip.Addr) bool { + ip = ip.Unmap() + if !ip.IsGlobalUnicast() || ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() { + return false + } + for _, p := range []string{"0.0.0.0/8", "100.64.0.0/10", "192.0.0.0/24", "192.0.2.0/24", "198.18.0.0/15", "198.51.100.0/24", "203.0.113.0/24", "240.0.0.0/4", "2001:db8::/32"} { + if netip.MustParsePrefix(p).Contains(ip) { + return false + } + } + return true +} + +func publicDial(ctx context.Context, network, address string) (net.Conn, error) { + host, port, err := net.SplitHostPort(address) + if err != nil || port != "443" { + return nil, errors.New("webpush: invalid destination") + } + ips, err := net.DefaultResolver.LookupNetIP(ctx, "ip", host) + if err != nil { + return nil, errors.New("webpush: name lookup failed") + } + for _, ip := range ips { + if !publicIP(ip) { + return nil, errors.New("webpush: local destination refused") + } + } + for _, ip := range ips { + c, e := (&net.Dialer{Timeout: 5 * time.Second}).DialContext(ctx, network, net.JoinHostPort(ip.String(), port)) + if e == nil { + return c, nil + } + } + return nil, errors.New("webpush: connection failed") +} + +// deliveryURL also binds the short-lived VAPID token to this endpoint's +// origin. The push vendor verifies the box's signature; we hold no key. +func deliveryURL(row cloudwire.WebPushRow, now time.Time) bool { + u, err := url.Parse(row.Endpoint) + if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil || u.Fragment != "" || (u.Port() != "" && u.Port() != "443") { + return false + } + if ip, err := netip.ParseAddr(u.Hostname()); err == nil && !publicIP(ip) { + return false + } + auth, ok := strings.CutPrefix(row.Auth, "vapid t=") + if !ok { + return false + } + jwt, key, ok := strings.Cut(auth, ", k=") + if !ok || key == "" { + return false + } + parts := strings.Split(jwt, ".") + if len(parts) != 3 { + return false + } + raw, err := base64.RawURLEncoding.DecodeString(parts[1]) + if err != nil { + return false + } + var claims struct { + Aud string `json:"aud"` + Exp int64 `json:"exp"` + } + if json.Unmarshal(raw, &claims) != nil { + return false + } + return claims.Aud == u.Scheme+"://"+u.Host && claims.Exp > now.Unix() && claims.Exp <= now.Add(24*time.Hour).Unix() +} + +// Send shares the APNs outage timer and queue. It rechecks the signed roster, +// epochs and suspensions on every try. A copy grant is not needed: the box +// authored and encrypted these notices for the browsers that subscribed. +func (s *Service) Send(ctx context.Context, home string) (took int, retry bool) { + due, err := s.reg.WebPushOutage(home) + if err != nil { + return 0, true + } + if due == 0 || due > s.now().UnixMilli() { + return 0, false + } + h, ok, err := s.reg.Get(home) + if err != nil { + return 0, true + } + if !ok { + return 0, false + } + env, err := cloudwire.DecodeEnvelope(h.Roster) + if err != nil { + return 0, false + } + var roster cloudwire.Roster + if cloudwire.Unmarshal(env.B, &roster) != nil || cloudwire.ValidateWebPush(roster.WebPush) != nil { + return 0, false + } + if roster.TTLS <= 0 || !s.now().Before(h.RosterAt.Add(time.Duration(min(roster.TTLS, 86400))*time.Second)) { + return 0, false + } + sent, err := s.reg.WebPushSent(home, due) + if err != nil { + return 0, true + } + for _, row := range roster.WebPush { + mark := row.ID + ":" + strconv.FormatUint(row.Epoch, 10) + if slices.Contains(sent, mark) { + continue + } + eligible := slices.ContainsFunc(roster.Devices, func(d cloudwire.Device) bool { + return d.ID == row.Device && d.Epoch == row.Epoch && d.Agent == nil && slices.Contains(d.Scopes, protocol.ScopeEventsRead) + }) + if !eligible || s.reg.Suspended(home, row.Device, row.Epoch) || !deliveryURL(row, s.now()) { + continue + } + current, e := s.reg.WebPushOutage(home) + if e != nil { + return took, true + } + if current != due { + return took, false + } + req, e := http.NewRequestWithContext(ctx, http.MethodPost, row.Endpoint, bytes.NewReader(row.CT)) + if e != nil { + continue + } + req.Header.Set("Content-Type", "application/octet-stream") + req.Header.Set("Content-Encoding", "aes128gcm") + req.Header.Set("TTL", "3600") + req.Header.Set("Urgency", "high") + req.Header.Set("Authorization", row.Auth) + resp, e := s.client.Do(req) + if e != nil { + retry = true + continue + } + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1024)) + _ = resp.Body.Close() + if resp.StatusCode == 429 || resp.StatusCode >= 500 { + retry = true + continue + } + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + took++ + } + // Terminal refusals (including expired subscriptions) do not repeat. + if s.reg.WebPushMark(home, due, mark) != nil { + retry = true + } + } + return took, retry +} diff --git a/internal/webpush/webpush_test.go b/internal/webpush/webpush_test.go new file mode 100644 index 0000000..9cfd64a --- /dev/null +++ b/internal/webpush/webpush_test.go @@ -0,0 +1,152 @@ +package webpush + +import ( + "bytes" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "encoding/base64" + "encoding/json" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" + "io" + "net/http" + "net/netip" + "path/filepath" + "testing" + "time" +) + +type post func(*http.Request) (*http.Response, error) + +func (p post) RoundTrip(r *http.Request) (*http.Response, error) { return p(r) } +func auth(endpoint string, now time.Time) string { + raw, _ := json.Marshal(map[string]any{"aud": endpoint, "exp": now.Add(time.Hour).Unix()}) + return "vapid t=e30." + base64.RawURLEncoding.EncodeToString(raw) + ".signature, k=public" +} + +func TestOfflineNoticesKeepCiphertextAndRespectEveryGrantWithoutACopy(t *testing.T) { + path := filepath.Join(t.TempDir(), "registry.db") + reg, err := registry.Open(path) + if err != nil { + t.Fatal(err) + } + defer func() { reg.Close() }() + now := time.Unix(1_791_820_800, 0) + const home = "home" + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err = reg.Claim(home, elliptic.Marshal(key.Curve, key.X, key.Y), now); err != nil { + t.Fatal(err) + } + svc, err := New(reg) + if err != nil { + t.Fatal(err) + } + svc.now = func() time.Time { return now } + row := cloudwire.WebPushRow{ID: "subscription", Device: "browser", Epoch: 1, Endpoint: "https://push.example/opaque", CT: []byte("box encrypted bytes"), Auth: auth("https://push.example", now)} + device := cloudwire.Device{ID: "browser", Key: make([]byte, 32), Epoch: 1, Role: "owner", Scopes: []string{protocol.ScopeEventsRead}} + roster := cloudwire.Roster{RosterSeq: 1, TTLS: 86400, Devices: []cloudwire.Device{device}, WebPush: []cloudwire.WebPushRow{row}} + save := func() { + t.Helper() + env, err := cloudwire.Seal(key, home, cloudwire.MsgRoster, roster) + if err != nil { + t.Fatal(err) + } + raw, _ := env.Encode() + if err = reg.SetRoster(home, roster.RosterSeq, nil, raw, now); err != nil { + t.Fatal(err) + } + } + save() + status := 201 + posts := 0 + svc.client = &http.Client{Transport: post(func(r *http.Request) (*http.Response, error) { + posts++ + body, _ := io.ReadAll(r.Body) + if !bytes.Equal(body, row.CT) || r.Header.Get("Authorization") != row.Auth || r.Header.Get("Content-Encoding") != "aes128gcm" { + t.Error("notice changed") + } + return &http.Response{StatusCode: status, Body: io.NopCloser(bytes.NewReader(nil)), Header: make(http.Header)}, nil + })} + send := func(want int, retry bool) { + t.Helper() + n, again := svc.Send(context.Background(), home) + if n != want || again != retry { + t.Fatalf("sent %d retry %v, want %d %v", n, again, want, retry) + } + } + send(0, false) // online + due := now.Add(time.Minute) + reg.AwaySet(home, due) + send(0, false) // grace + now = due + status = 503 + send(0, true) + status = 201 + send(1, false) + send(0, false) + if posts != 2 { + t.Fatal(posts) + } + // The sent marker survives a process restart. + reg.Close() + reg, err = registry.Open(path) + if err != nil { + t.Fatal(err) + } + svc.reg = reg + send(0, false) + reg.AwayClear(home) + now = now.Add(time.Hour) + reg.AwaySet(home, now) + // Renew the box's VAPID token, then test epoch, scopes and suspension. + row.Auth = auth("https://push.example", now) + roster.WebPush = []cloudwire.WebPushRow{row} + roster.Devices[0].Epoch = 2 + save() + send(0, false) + roster.Devices[0].Epoch = 1 + roster.Devices[0].Scopes = nil + save() + send(0, false) + roster.Devices[0].Scopes = []string{protocol.ScopeEventsRead} + save() + reg.Suspend(home, "browser", 1, now) + send(0, false) + reg.Unsuspend(home, "browser") + send(1, false) + reg.AwayClear(home) + send(0, false) + now = now.Add(time.Hour) + reg.AwaySet(home, now) + roster.WebPush = nil + save() + send(0, false) +} + +func TestAURLOrDNSCannotReachPrivateServices(t *testing.T) { + now := time.Now() + row := cloudwire.WebPushRow{Endpoint: "https://push.example/x", Auth: auth("https://push.example", now)} + if deliveryURL(row, now.Add(2*time.Hour)) { + t.Fatal("expired authorization accepted") + } + if !deliveryURL(row, now) { + t.Fatal("valid endpoint refused") + } + for _, u := range []string{"http://push.example/x", "https://user@push.example/x", "https://push.example:444/x", "https://127.0.0.1/x", "https://169.254.169.254/latest/", "https://[::ffff:127.0.0.1]/", "https://push.example/x#part"} { + row.Endpoint = u + if deliveryURL(row, now) { + t.Fatal(u) + } + } + for _, ip := range []string{"127.0.0.1", "10.1.1.1", "169.254.169.254", "::1", "::ffff:192.168.1.2", "100.100.100.1", "fd00::1"} { + if publicIP(netip.MustParseAddr(ip)) { + t.Fatal(ip) + } + } + if _, err := publicDial(context.Background(), "tcp", "127.0.0.1:443"); err == nil { + t.Fatal("local dial succeeded") + } +} From 090735148592e19bc15f073c612b30a5acaedf27 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sat, 17 Oct 2026 09:09:15 +0200 Subject: [PATCH 2/2] fix: keep browser retries and enforce calendar retention across restarts Signed-off-by: Fredrik Ahlgren --- cmd/ftwcloudd/main.go | 1 + internal/fleet/fleet.go | 7 ++++- internal/fleet/fleet_test.go | 14 +++++++--- internal/push/browser_test.go | 50 +++++++++++++++++++++++++++++++++++ internal/push/push.go | 18 ++++++++----- internal/registry/fleet.go | 12 ++++++--- internal/server/server.go | 5 ++++ internal/webpush/webpush.go | 21 +++++++++++++++ 8 files changed, 115 insertions(+), 13 deletions(-) diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index c6dc146..5082cd9 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -195,6 +195,7 @@ func serve(args []string) error { } pushSvc := push.New(reg, sender, log) pushSvc.BrowserUnreachable = webPushSvc.Send + pushSvc.BrowserHomes = webPushSvc.Homes pushSvc.WatchApp = *apnsWatch pushSvc.Owns = shards.Ours up.OnHeld = pushSvc.Held diff --git a/internal/fleet/fleet.go b/internal/fleet/fleet.go index 946e445..d5a1102 100644 --- a/internal/fleet/fleet.go +++ b/internal/fleet/fleet.go @@ -24,6 +24,9 @@ func New(reg *registry.Registry) (*Service, error) { if err := reg.FleetInit(); err != nil { return nil, err } + if err := reg.FleetPrune(time.Now()); err != nil { + return nil, err + } return &Service{reg: reg, now: time.Now, tokens: 120}, nil } @@ -67,7 +70,7 @@ func (s *Service) Stats(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusMethodNotAllowed) return } - view, err := s.reg.FleetView() + view, err := s.reg.FleetView(s.now()) if err != nil { w.WriteHeader(http.StatusServiceUnavailable) return @@ -91,3 +94,5 @@ func (s *Service) allow() bool { s.tokens-- return true } + +func (s *Service) Prune(now time.Time) error { return s.reg.FleetPrune(now) } diff --git a/internal/fleet/fleet_test.go b/internal/fleet/fleet_test.go index b8d8747..4ee230a 100644 --- a/internal/fleet/fleet_test.go +++ b/internal/fleet/fleet_test.go @@ -23,7 +23,7 @@ func TestReportsBecomeOnlyBoundedDailyTotalsAcrossARestart(t *testing.T) { if err != nil { t.Fatal(err) } - now := time.Date(2026, 10, 17, 0, 0, 0, 0, time.UTC) + now := time.Now().UTC().Truncate(24 * time.Hour) svc.now = func() time.Time { return now } report := fleetwire.Report{Schema: fleetwire.Schema, Version: "v0.146.0", Channel: "stable", Drivers: []string{}, Battery: "none", Zone: "SE4", Age: "unknown"} raw, _ := json.Marshal(report) @@ -52,7 +52,7 @@ func TestReportsBecomeOnlyBoundedDailyTotalsAcrossARestart(t *testing.T) { if err != nil { t.Fatal(err) } - view, err := reg.FleetView() + view, err := reg.FleetView(now) if err != nil || len(view.Days) != 1 || view.Days[0].Reports != 100 { t.Fatal(view, err) } @@ -61,10 +61,18 @@ func TestReportsBecomeOnlyBoundedDailyTotalsAcrossARestart(t *testing.T) { t.Fatal(err) } } - view, _ = reg.FleetView() + view, _ = reg.FleetView(now.AddDate(0, 0, 91)) if len(view.Days) != 90 { t.Fatal(len(view.Days)) } + // A long gap expires old populated dates, even without another report. + if err := reg.FleetPrune(now.AddDate(1, 0, 0)); err != nil { + t.Fatal(err) + } + view, _ = reg.FleetView(now.AddDate(1, 0, 0)) + if len(view.Days) != 0 { + t.Fatal("old totals survived", view) + } for _, tc := range []struct { method, body string code int diff --git a/internal/push/browser_test.go b/internal/push/browser_test.go index ea20964..5bb870a 100644 --- a/internal/push/browser_test.go +++ b/internal/push/browser_test.go @@ -75,3 +75,53 @@ func TestBrowserRetryDoesNotResendAnAcceptedAppleAlert(t *testing.T) { t.Fatal("resending accepted APNs", calls) } } + +func TestABrowserOnlyHomeGetsRestartGraceWithoutAnAwayRow(t *testing.T) { + r := newRig(t) + r.svc.sender = nil + calls := 0 + r.svc.BrowserHomes = func() ([]string, error) { return []string{r.home, r.home}, nil } + r.svc.BrowserUnreachable = func(context.Context, string) (int, bool) { calls++; return 1, false } + if err := r.svc.Resume(); err != nil { + t.Fatal(err) + } + r.clock.advance(599 * time.Second) + r.drain() + if calls != 0 { + t.Fatal("before grace") + } + r.clock.advance(2 * time.Second) + r.drain() + if calls != 1 { + t.Fatal("missing or duplicate", calls) + } +} + +func TestAnAcceptedAppleAlertDoesNotLoseABrowserRetryAtRestart(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "token") + calls := 0 + r.svc.BrowserUnreachable = func(context.Context, string) (int, bool) { calls++; return 0, true } + r.connected() + r.gone("") + r.clock.advance(601 * time.Second) + r.drain() + away, _ := r.reg.AwayAll() + if len(away) != 1 || away[0].Told { + t.Fatal("pending browser delivery marked told") + } + // A new service uses the same disk state. Re-arm the browser callback + // before Resume, as main does. + r.svc = New(r.reg, nil, nil) + r.svc.now, r.svc.after = r.clock.time, r.clock.after + r.svc.BrowserUnreachable = func(context.Context, string) (int, bool) { calls++; return 1, false } + if err := r.svc.Resume(); err != nil { + t.Fatal(err) + } + r.clock.advance(time.Millisecond) + r.drain() + away, _ = r.reg.AwayAll() + if calls != 2 || !away[0].Told { + t.Fatal("retry lost across restart", calls, away) + } +} diff --git a/internal/push/push.go b/internal/push/push.go index 122d005..1bc52fa 100644 --- a/internal/push/push.go +++ b/internal/push/push.go @@ -184,6 +184,7 @@ type Service struct { // BrowserUnreachable delivers the box-encrypted browser notices on the // same outage timer. It needs no APNs key or readable-copy grant. BrowserUnreachable func(context.Context, string) (int, bool) + BrowserHomes func() ([]string, error) // Owns reports whether this shard holds a home. Resume starts no // box.unreachable for a home another shard holds, whose rows may still // be here during a move. Nil holds every home. Set before Resume. @@ -693,11 +694,19 @@ func (s *Service) Resume() error { if err != nil { return err } + if s.BrowserHomes != nil { + more, err := s.BrowserHomes() + if err != nil { + return err + } + homes = append(homes, more...) + } wait := time.Duration(cloudwire.UnreachableAfterMs) * time.Millisecond for _, home := range homes { if seen[home] || (s.Owns != nil && !s.Owns(home)) { continue } + seen[home] = true if err := s.reg.AwaySet(home, now.Add(wait)); err != nil { return err } @@ -860,11 +869,9 @@ func (s *Service) sendAlert(ctx context.Context, job alertJob) { }) } job.tries++ - took := 0 browserRetry := false if job.unreachable && s.BrowserUnreachable != nil { - n, retry := s.BrowserUnreachable(ctx, job.home) - took += n + _, retry := s.BrowserUnreachable(ctx, job.home) browserRetry = retry } again := []registry.PushReg{} @@ -872,11 +879,10 @@ func (s *Service) sendAlert(ctx context.Context, job alertJob) { if !s.breaker.wait(ctx) { return } - n, remaining := s.deliver(ctx, job.home, cloudwire.PushKindAlert, job.payload, job.expires, job.regs) - took += n + _, remaining := s.deliver(ctx, job.home, cloudwire.PushKindAlert, job.payload, job.expires, job.regs) again = remaining } - if took > 0 || (len(again) == 0 && !browserRetry) { + if len(again) == 0 && !browserRetry { s.told(job) } if len(again) > 0 || browserRetry { diff --git a/internal/registry/fleet.go b/internal/registry/fleet.go index 1c962f8..97b6c67 100644 --- a/internal/registry/fleet.go +++ b/internal/registry/fleet.go @@ -44,15 +44,15 @@ func (r *Registry) FleetPut(report fleetwire.Report, now time.Time) (bool, error if _, err := tx.Exec(`INSERT INTO fleet_days(date, totals) VALUES (?, ?) ON CONFLICT(date) DO UPDATE SET totals=excluded.totals`, date, data); err != nil { return false, err } - if _, err := tx.Exec(`DELETE FROM fleet_days WHERE date NOT IN (SELECT date FROM fleet_days ORDER BY date DESC LIMIT ?)`, fleetwire.MaxDays); err != nil { + if _, err := tx.Exec(`DELETE FROM fleet_days WHERE date < ? OR date NOT IN (SELECT date FROM fleet_days ORDER BY date DESC LIMIT ?)`, now.UTC().AddDate(0, 0, 1-fleetwire.MaxDays).Format(time.DateOnly), fleetwire.MaxDays); err != nil { return false, err } return true, tx.Commit() } -func (r *Registry) FleetView() (fleetwire.View, error) { +func (r *Registry) FleetView(now time.Time) (fleetwire.View, error) { view := fleetwire.View{Schema: fleetwire.StatsSchema, Meaning: fleetwire.Meaning, Days: []fleetwire.Day{}} - rows, err := r.rd.Query(`SELECT totals FROM fleet_days ORDER BY date DESC LIMIT ?`, fleetwire.MaxDays) + rows, err := r.rd.Query(`SELECT totals FROM fleet_days WHERE date >= ? AND date <= ? ORDER BY date DESC LIMIT ?`, now.UTC().AddDate(0, 0, 1-fleetwire.MaxDays).Format(time.DateOnly), now.UTC().Format(time.DateOnly), fleetwire.MaxDays) if err != nil { return view, err } @@ -70,3 +70,9 @@ func (r *Registry) FleetView() (fleetwire.View, error) { } return view, rows.Err() } + +// FleetPrune expires dates by calendar age, including when no reports arrive. +func (r *Registry) FleetPrune(now time.Time) error { + _, err := r.db.Exec(`DELETE FROM fleet_days WHERE date < ?`, now.UTC().AddDate(0, 0, 1-fleetwire.MaxDays).Format(time.DateOnly)) + return err +} diff --git a/internal/server/server.go b/internal/server/server.go index 4b222e3..bcbdda5 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -181,6 +181,11 @@ func (s *Server) Upkeep(ctx context.Context, every time.Duration) { case <-t.C: s.Store.Sweep() s.Uplink.Sweep() + if f, ok := s.Fleet.(interface{ Prune(time.Time) error }); ok { + if err := f.Prune(time.Now()); err != nil { + s.Log.Warn("fleet upkeep failed") + } + } if err := s.Registry.TokenExpire(time.Now()); err != nil { s.Log.Warn("token upkeep failed", "err", err) } diff --git a/internal/webpush/webpush.go b/internal/webpush/webpush.go index c332907..dae0deb 100644 --- a/internal/webpush/webpush.go +++ b/internal/webpush/webpush.go @@ -191,3 +191,24 @@ func (s *Service) Send(ctx context.Context, home string) (took int, retry bool) } return took, retry } + +// Homes includes browser-only homes in the restart grace, even if their +// box was online when the service stopped and left no away row. +func (s *Service) Homes() ([]string, error) { + homes, err := s.reg.Rosters() + if err != nil { + return nil, err + } + var out []string + for _, h := range homes { + env, err := cloudwire.DecodeEnvelope(h.Roster) + if err != nil { + continue + } + var r cloudwire.Roster + if cloudwire.Unmarshal(env.B, &r) == nil && len(r.WebPush) > 0 && cloudwire.ValidateWebPush(r.WebPush) == nil { + out = append(out, h.ID) + } + } + return out, nil +}