feat(push): start the push dispatcher at server startup when a gateway is set (TASK-3939 B2c) (#2286)

* feat(push): start the push dispatcher at server startup when a gateway is set (TASK-3939 B2c)

Server startup now calls configurePush, which sets the gateway URL and, when
one is configured, refuses a missing secret, sets the gateway client and
starts the dispatcher. With no URL nothing starts: no device can register, so
nothing is queued. Shutdown already stops the loop with the server's other
background workers. The comments that said nothing starts the dispatcher are
corrected.

Claude-Session: https://claude.ai/code/session_01D6M9onL7HhMovW8DZbs6f5

* docs(push): configurePush's comment says what happens to queued rows with no gateway (TASK-3939 B2c r1 L1)

Claude-Session: https://claude.ai/code/session_01D6M9onL7HhMovW8DZbs6f5
This commit is contained in:
xarmian
2026-10-22 16:17:56 -04:00
committed by GitHub
parent b039f111eb
commit eaa1b0da16
4 changed files with 113 additions and 13 deletions
+24 -7
View File
@@ -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
+75
View File
@@ -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)
}
}
+2 -2
View File
@@ -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
}
+12 -4
View File
@@ -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