diff --git a/cmd/pad/cmd_server.go b/cmd/pad/cmd_server.go index e35b013c0..11b5dc37e 100644 --- a/cmd/pad/cmd_server.go +++ b/cmd/pad/cmd_server.go @@ -975,13 +975,8 @@ func serveCmd() *cobra.Command { bypassEnv := os.Getenv("PAD_BYPASS_SETUP_TOKEN") bypassSetupToken := bypassEnv == "true" || bypassEnv == "1" srv.SetBypassSetupToken(bypassSetupToken && !cfg.IsCloudServer()) - srv.SetPushGateway(cfg.PushGatewayURL) - if cfg.PushGatewayURL != "" { - pushSecret := resolvePushGatewaySecret(cfg) - if pushSecret == "" { - return fmt.Errorf("PAD_PUSH_GATEWAY_URL is set but none of PAD_PUSH_GATEWAY_SECRET, PAD_CLOUD_OUTBOUND_SECRET or PAD_CLOUD_SECRET supplies a usable secret") - } - srv.SetPushGatewayClient(notify.NewGateway(cfg.PushGatewayURL, pushSecret)) + if err := configurePush(srv, cfg); err != nil { + return err } if userCount, ucErr := s.UserCount(); ucErr != nil { slog.Warn("could not check user count for bootstrap token wiring", "error", ucErr) @@ -1528,6 +1523,28 @@ func bootPlanBackfill(s *store.Store, cloud bool) error { return nil } +// configurePush wires the push gateway from cfg and, when one is configured, +// starts the push dispatcher (TASK-3939 B2c). A gateway URL with no usable +// secret is refused before anything starts. With no URL the server offers no +// push and no dispatcher runs: no new device can register. Rows are still +// queued for devices registered while a gateway was configured; they stay +// pending, and once a gateway is configured again the dispatcher expires any +// past the expiry age instead of sending them. Shutdown stops the dispatcher +// with the server's other background loops. +func configurePush(srv *server.Server, cfg *config.Config) error { + srv.SetPushGateway(cfg.PushGatewayURL) + if cfg.PushGatewayURL == "" { + return nil + } + pushSecret := resolvePushGatewaySecret(cfg) + if pushSecret == "" { + return fmt.Errorf("PAD_PUSH_GATEWAY_URL is set but none of PAD_PUSH_GATEWAY_SECRET, PAD_CLOUD_OUTBOUND_SECRET or PAD_CLOUD_SECRET supplies a usable secret") + } + srv.SetPushGatewayClient(notify.NewGateway(cfg.PushGatewayURL, pushSecret)) + srv.StartPushDispatch() + return nil +} + // resolvePushGatewaySecret is the secret to send the push gateway: the // explicit PAD_PUSH_GATEWAY_SECRET, else what pad already sends pad-cloud // (PAD_CLOUD_OUTBOUND_SECRET, else the last PAD_CLOUD_SECRET entry). Empty diff --git a/cmd/pad/task3939c_configure_push_test.go b/cmd/pad/task3939c_configure_push_test.go new file mode 100644 index 000000000..327b09591 --- /dev/null +++ b/cmd/pad/task3939c_configure_push_test.go @@ -0,0 +1,75 @@ +package main + +import ( + "os" + "strings" + "testing" + + "github.com/PerpetualSoftware/pad/internal/config" + "github.com/PerpetualSoftware/pad/internal/server" + "github.com/PerpetualSoftware/pad/internal/store/storetest" +) + +// TASK-3939 B2c: server startup starts the push dispatcher exactly when a +// gateway is configured with a usable secret, and refuses a gateway URL with +// no secret before anything starts. +// +// Kills: dropping StartPushDispatch (the URL+secret leg); starting the +// dispatcher before the secret check (the no-secret leg); starting it outside +// the gateway block (the no-URL leg). +func TestConfigurePush_StartsTheDispatcherOnlyWithAGateway(t *testing.T) { + newSrv := func(t *testing.T) *server.Server { + srv := server.New(storetest.NewSQLite(t)) + t.Cleanup(srv.Stop) + return srv + } + + t.Run("url and secret", func(t *testing.T) { + srv := newSrv(t) + cfg := &config.Config{PushGatewayURL: "https://push.example.test", PushGatewaySecret: "s3cret-for-the-gateway"} + if err := configurePush(srv, cfg); err != nil { + t.Fatalf("configurePush = %v, want nil", err) + } + if !srv.PushDispatchRunning() { + t.Fatal("a configured gateway did not start the push dispatcher") + } + }) + + t.Run("no url", func(t *testing.T) { + srv := newSrv(t) + if err := configurePush(srv, &config.Config{PushGatewaySecret: "s3cret-for-the-gateway"}); err != nil { + t.Fatalf("configurePush = %v, want nil", err) + } + if srv.PushDispatchRunning() { + t.Fatal("the push dispatcher started with no gateway configured") + } + }) + + t.Run("url without a secret", func(t *testing.T) { + srv := newSrv(t) + err := configurePush(srv, &config.Config{PushGatewayURL: "https://push.example.test"}) + if err == nil || !strings.Contains(err.Error(), "PAD_PUSH_GATEWAY_URL is set") { + t.Fatalf("configurePush = %v, want the missing-secret refusal", err) + } + if srv.PushDispatchRunning() { + t.Fatal("the push dispatcher started before the missing-secret refusal") + } + }) +} + +// CONVE-19: the wiring is a claim. configurePush is tested above; this pins +// that server startup calls it and returns its error, and that nothing else in +// startup starts the dispatcher around it. +func TestServerStartCallsConfigurePush(t *testing.T) { + src, err := os.ReadFile("cmd_server.go") + if err != nil { + t.Fatal(err) + } + body := string(src) + if !strings.Contains(body, "if err := configurePush(srv, cfg); err != nil {\n\t\t\t\treturn err\n\t\t\t}") { + t.Error("server start no longer calls configurePush(srv, cfg) and returns its error") + } + if n := strings.Count(body, "StartPushDispatch()"); n != 1 { + t.Errorf("cmd_server.go calls StartPushDispatch() %d times, want exactly 1 (inside configurePush)", n) + } +} diff --git a/internal/server/handlers_push_devices.go b/internal/server/handlers_push_devices.go index 48677349a..510ee4eeb 100644 --- a/internal/server/handlers_push_devices.go +++ b/internal/server/handlers_push_devices.go @@ -36,8 +36,8 @@ func (s *Server) SetPushGateway(url string) { } // SetPushGatewayClient sets the client the push dispatcher sends through -// (TASK-3939). Called once at startup, before the server accepts connections. -// Nothing starts a dispatcher from it yet. +// (TASK-3939). Called once at startup, before the server accepts connections, +// and before StartPushDispatch (cmd/pad configurePush). func (s *Server) SetPushGatewayClient(g *notify.Gateway) { s.pushGateway = g } diff --git a/internal/server/push_dispatch.go b/internal/server/push_dispatch.go index fa94d63b3..28e4011ef 100644 --- a/internal/server/push_dispatch.go +++ b/internal/server/push_dispatch.go @@ -17,10 +17,10 @@ import ( // The push dispatcher (TASK-3939 part B1): claims settled user_notifications // rows, runs every privacy gate (push_gates.go) on each, groups the survivors // per (user, item), sends one content-free push per live device, and settles -// the rows. StartPushDispatch is NOT called -// from cmd/pad yet. Part B2a adds the 30/h cap and the summary push, the retry -// matrix (which results retry, skip or delete a device), and the expire and -// retention sweeps. +// the rows. Part B2a added the 30/h cap and the summary push, the retry matrix +// (which results retry, skip or delete a device), and the expire and retention +// sweeps. Server startup (cmd/pad configurePush, TASK-3939 B2c) calls +// StartPushDispatch when PAD_PUSH_GATEWAY_URL is set, with the default timings. // // LEASE INVARIANT (r6-Q1): lease >= sendTimeout + leaseMargin. A claim is // stamped with second precision (up to 1s behind the real clock), and a Send @@ -248,6 +248,14 @@ func (s *Server) StartPushDispatch() { }() } +// PushDispatchRunning reports whether the dispatch loop has been started and +// not stopped. Read-only; startup's wiring test reads it (TASK-3939 B2c). +func (s *Server) PushDispatchRunning() bool { + s.pushDispatch.mu.Lock() + defer s.pushDispatch.mu.Unlock() + return s.pushDispatch.running +} + // stopPushDispatch signals the loop to exit. Safe when it never started. func (s *Server) stopPushDispatch() { d := &s.pushDispatch