mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-10-25 22:10:25 +00:00
feat(server): push dispatch timings are configurable, and a bad value refuses startup (TASK-3939) (#2288)
* feat(server): push dispatch timings are configurable, and a bad value refuses startup (TASK-3939) Claude-Session: https://claude.ai/code/session_01D6M9onL7HhMovW8DZbs6f5 * docs(deployment): push gateway env vars, dispatch timings and the cleartext warning (TASK-3939) Claude-Session: https://claude.ai/code/session_01D6M9onL7HhMovW8DZbs6f5 * fix(server): the lease refusal names the cause, and an overflowing margin is refused (TASK-3939) SetPushDispatchConfig wraps errPushLeaseTooShort with the resolved lease, send timeout and margin; configurePush names both knobs and appends it. A margin within sendTimeout of MaxInt64 is refused before the sum can wrap. Whitespace-only values stay malformed and are now documented and tested; values are otherwise uncapped. Claude-Session: https://claude.ai/code/session_01D6M9onL7HhMovW8DZbs6f5
This commit is contained in:
+22
-1
@@ -1525,13 +1525,34 @@ func bootPlanBackfill(s *store.Store, cloud bool) error {
|
||||
|
||||
// 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
|
||||
// secret is refused before anything starts. So is a malformed or non-positive
|
||||
// PAD_PUSH_DISPATCH_* knob, or a combination whose lease is shorter than the
|
||||
// send timeout plus margin, or whose margin overflows that sum (TASK-3939
|
||||
// part C, rulings 77a8fae7 and 37e3eafa; the refusal names both
|
||||
// PAD_PUSH_DISPATCH_LEASE and _LEASE_MARGIN and states the resolved values,
|
||||
// since either can be the cause): a bad knob is
|
||||
// an operator error whether or not a gateway is set, so it is checked first
|
||||
// and never falls back to the default. The knobs are applied before
|
||||
// StartPushDispatch because the interval is read once, at Start. 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 {
|
||||
timings, err := cfg.PushDispatchTimings()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Applied here, before any Start and whether or not a gateway is set, so a
|
||||
// combination the setter refuses (lease shorter than the send timeout plus
|
||||
// margin) refuses startup the same way a malformed value does.
|
||||
// sendTimeout is 0 (keep notify.SendTimeout): it has no knob.
|
||||
if err := srv.SetPushDispatchConfig(timings.Interval, timings.Settle, timings.Lease, timings.LeaseMargin, 0); err != nil {
|
||||
// The setter's error carries the resolved lease and margin, which is
|
||||
// what tells the operator which of the two knobs to change.
|
||||
return fmt.Errorf("PAD_PUSH_DISPATCH_LEASE must be at least the send timeout plus PAD_PUSH_DISPATCH_LEASE_MARGIN; change either knob (%w)", err)
|
||||
}
|
||||
srv.SetPushGateway(cfg.PushGatewayURL)
|
||||
if cfg.PushGatewayURL == "" {
|
||||
return nil
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/PerpetualSoftware/pad/internal/config"
|
||||
"github.com/PerpetualSoftware/pad/internal/server"
|
||||
@@ -73,3 +74,102 @@ func TestServerStartCallsConfigurePush(t *testing.T) {
|
||||
t.Errorf("cmd_server.go calls StartPushDispatch() %d times, want exactly 1 (inside configurePush)", n)
|
||||
}
|
||||
}
|
||||
|
||||
// TASK-3939 part C (ruling 77a8fae7): the PAD_PUSH_DISPATCH_* knobs reach the
|
||||
// dispatcher through configurePush, and the interval is applied before Start.
|
||||
//
|
||||
// Kills: dropping any one knob's assignment (the four expectations differ);
|
||||
// applying the config after StartPushDispatch (StartedInterval would be the
|
||||
// 5s default, the ticker having been created before the knob landed).
|
||||
func TestConfigurePush_AppliesEveryTimingKnobBeforeStart(t *testing.T) {
|
||||
srv := server.New(storetest.NewSQLite(t))
|
||||
t.Cleanup(srv.Stop)
|
||||
cfg := &config.Config{
|
||||
PushGatewayURL: "https://push.example.test", PushGatewaySecret: "s3cret-for-the-gateway",
|
||||
PushDispatchIntervalRaw: "2s", PushDispatchSettleRaw: "3s",
|
||||
PushDispatchLeaseRaw: "90s", PushDispatchLeaseMarginRaw: "20s",
|
||||
}
|
||||
if err := configurePush(srv, cfg); err != nil {
|
||||
t.Fatalf("configurePush = %v, want nil", err)
|
||||
}
|
||||
got := srv.PushDispatchApplied()
|
||||
want := server.PushDispatchApplied{
|
||||
Interval: 2 * time.Second, Settle: 3 * time.Second, Lease: 90 * time.Second,
|
||||
LeaseMargin: 20 * time.Second, StartedInterval: 2 * time.Second,
|
||||
}
|
||||
if got != want {
|
||||
t.Fatalf("dispatcher timings = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Unset knobs keep the defaults (5s / 10s / 60s / 15s) through configurePush.
|
||||
func TestConfigurePush_UnsetKnobsKeepTheDefaults(t *testing.T) {
|
||||
srv := server.New(storetest.NewSQLite(t))
|
||||
t.Cleanup(srv.Stop)
|
||||
if err := configurePush(srv, &config.Config{PushGatewayURL: "https://push.example.test", PushGatewaySecret: "s3cret-for-the-gateway"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := server.PushDispatchApplied{Interval: 5 * time.Second, Settle: 10 * time.Second, Lease: 60 * time.Second, LeaseMargin: 15 * time.Second, StartedInterval: 5 * time.Second}
|
||||
if got := srv.PushDispatchApplied(); got != want {
|
||||
t.Fatalf("dispatcher timings = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// A bad knob refuses startup, names its variable, and starts no dispatcher.
|
||||
// It refuses with or without a gateway URL: a malformed config is an operator
|
||||
// error either way.
|
||||
//
|
||||
// Kills: falling back to the default on a parse error; ignoring
|
||||
// SetPushDispatchConfig's lease-too-short error.
|
||||
func TestConfigurePush_BadKnobRefusesStartup(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
mutate func(*config.Config)
|
||||
wantVar string
|
||||
// wantDetail are substrings that tell the two lease causes apart: the
|
||||
// name alone cannot, since PAD_PUSH_DISPATCH_LEASE is a prefix of
|
||||
// PAD_PUSH_DISPATCH_LEASE_MARGIN (TASK-3939 part C, ruling 37e3eafa).
|
||||
wantDetail []string
|
||||
}{
|
||||
{"malformed interval", func(c *config.Config) { c.PushDispatchIntervalRaw = "soon" }, "PAD_PUSH_DISPATCH_INTERVAL", nil},
|
||||
{"whitespace-only interval", func(c *config.Config) { c.PushDispatchIntervalRaw = " " }, "PAD_PUSH_DISPATCH_INTERVAL", nil},
|
||||
{"malformed settle", func(c *config.Config) { c.PushDispatchSettleRaw = "10" }, "PAD_PUSH_DISPATCH_SETTLE", nil},
|
||||
{"zero lease", func(c *config.Config) { c.PushDispatchLeaseRaw = "0s" }, "PAD_PUSH_DISPATCH_LEASE", nil},
|
||||
{"negative margin", func(c *config.Config) { c.PushDispatchLeaseMarginRaw = "-1s" }, "PAD_PUSH_DISPATCH_LEASE_MARGIN", nil},
|
||||
{"lease shorter than send timeout plus margin", func(c *config.Config) { c.PushDispatchLeaseRaw = "30s" }, "PAD_PUSH_DISPATCH_LEASE", []string{"lease 30s", "margin 15s"}},
|
||||
{"margin that outgrows the default lease", func(c *config.Config) { c.PushDispatchLeaseMarginRaw = "50s" }, "PAD_PUSH_DISPATCH_LEASE_MARGIN", []string{"lease 1m0s", "margin 50s"}},
|
||||
{"margin of 2562047h", func(c *config.Config) { c.PushDispatchLeaseMarginRaw = "2562047h" }, "PAD_PUSH_DISPATCH_LEASE_MARGIN", []string{"lease 1m0s"}},
|
||||
// 2562047h47m16s + the 25s send timeout exceeds MaxInt64 and wraps
|
||||
// negative; 2562047h alone does not.
|
||||
{"margin that overflows the sum", func(c *config.Config) { c.PushDispatchLeaseMarginRaw = "2562047h47m16s" }, "PAD_PUSH_DISPATCH_LEASE_MARGIN", []string{"lease 1m0s"}},
|
||||
}
|
||||
for _, withURL := range []bool{true, false} {
|
||||
for _, tc := range cases {
|
||||
name := tc.name + " no url"
|
||||
if withURL {
|
||||
name = tc.name + " with url"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
srv := server.New(storetest.NewSQLite(t))
|
||||
t.Cleanup(srv.Stop)
|
||||
cfg := &config.Config{PushGatewaySecret: "s3cret-for-the-gateway"}
|
||||
if withURL {
|
||||
cfg.PushGatewayURL = "https://push.example.test"
|
||||
}
|
||||
tc.mutate(cfg)
|
||||
err := configurePush(srv, cfg)
|
||||
if err == nil || !strings.Contains(err.Error(), tc.wantVar) {
|
||||
t.Fatalf("configurePush = %v, want a refusal naming %s", err, tc.wantVar)
|
||||
}
|
||||
for _, d := range tc.wantDetail {
|
||||
if !strings.Contains(err.Error(), d) {
|
||||
t.Errorf("refusal %q lacks %q", err, d)
|
||||
}
|
||||
}
|
||||
if srv.PushDispatchRunning() {
|
||||
t.Fatal("the push dispatcher started despite the refusal")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1470,6 +1470,31 @@ Email enables sending workspace invitation links. Without it, users can still jo
|
||||
| `PAD_EMAIL_FROM` | `noreply@getpad.dev` | Sender email address |
|
||||
| `PAD_EMAIL_FROM_NAME` | `Pad` | Sender display name |
|
||||
|
||||
### Push notifications
|
||||
|
||||
Mobile push goes out through a push gateway. With no `PAD_PUSH_GATEWAY_URL` this server offers no push and runs no dispatcher.
|
||||
|
||||
| Variable | Default | Description |
|
||||
|----------|---------|-------------|
|
||||
| `PAD_PUSH_GATEWAY_URL` | — | Base URL of the push gateway. Setting it turns push on and starts the dispatcher. |
|
||||
| `PAD_PUSH_GATEWAY_SECRET` | see below | Secret sent to the gateway in the `X-Cloud-Secret` header. If unset, the first non-empty of `PAD_CLOUD_OUTBOUND_SECRET`, then the last entry of `PAD_CLOUD_SECRET`. Startup refuses a gateway URL with no usable secret. |
|
||||
| `PAD_PUSH_DISPATCH_INTERVAL` | `5s` | How often the dispatcher looks for notifications to push. Read once at startup. |
|
||||
| `PAD_PUSH_DISPATCH_SETTLE` | `10s` | How long a notification must sit before it is pushed. |
|
||||
| `PAD_PUSH_DISPATCH_LEASE` | `60s` | How long a claimed notification stays reserved to this instance before another may reclaim it. |
|
||||
| `PAD_PUSH_DISPATCH_LEASE_MARGIN` | `15s` | Safety margin the lease must keep beyond the send timeout. |
|
||||
|
||||
The four timing knobs are Go durations (`5s`, `1m30s`) and are environment variables only. Startup refuses a value that is not a valid duration, is zero or negative, or that breaks the lease rule below, with a message naming the variable. A value of only whitespace is malformed, not unset: only an empty or absent variable is unset. It never falls back to the default silently, and it refuses whether or not a gateway URL is set.
|
||||
|
||||
Values are not capped: a large explicit value is your choice (a huge settle simply means nothing is pushed, a huge interval means the dispatcher effectively never runs). The one bound is the lease rule, which also refuses a margin so large that adding the send timeout would overflow.
|
||||
|
||||
**Lease rule.** The lease must be at least the send timeout plus the lease margin (`PAD_PUSH_DISPATCH_LEASE >= 25s + PAD_PUSH_DISPATCH_LEASE_MARGIN`). A lease shorter than that would let a second instance reclaim a notification while the first is still sending it, and push it twice. With the defaults, 60s >= 25s + 15s. Raising the margin past 35s therefore needs a longer lease too. The refusal message names both `PAD_PUSH_DISPATCH_LEASE` and `PAD_PUSH_DISPATCH_LEASE_MARGIN` and states the lease, send timeout and margin it resolved (defaults included), so you can see which to change.
|
||||
|
||||
**The send timeout is fixed at 25s and has no setting.** It is the whole-request timeout of the gateway's HTTP client, built into the binary. A setting for it would only change the number the lease rule checks, not the timeout the client uses, so the rule would certify a value that is never applied.
|
||||
|
||||
**Use `https://` for the gateway URL.** Over plain `http://`, the `X-Cloud-Secret` header and the notification payloads travel in cleartext. Use `http://` only on a network you control, such as a private Docker network between the two containers. Startup accepts `http://` on purpose: the URL is set by the operator, at the same trust as `PAD_CLOUD_URL`.
|
||||
|
||||
The secret is shared with pad-cloud unless you set `PAD_PUSH_GATEWAY_SECRET` separately. If the gateway is not pad-cloud, set it, so the cloud secret is not sent to another service.
|
||||
|
||||
### Password recovery (when email is not configured)
|
||||
|
||||
Without an email provider, the web "Forgot password" flow can't send a reset
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/BurntSushi/toml"
|
||||
)
|
||||
@@ -95,8 +96,20 @@ type Config struct {
|
||||
// (PAD_PUSH_GATEWAY_SECRET, TASK-3939). Empty falls back to the cloud
|
||||
// outbound secret: see resolvePushGatewaySecret in cmd/pad.
|
||||
PushGatewaySecret string `toml:"push_gateway_secret"`
|
||||
MCPPublicURL string `toml:"mcp_public_url"` // Canonical URL clients paste into their MCP client (e.g. https://mcp.getpad.dev — matches mcp.stripe.com / mcp.linear.app convention). Published verbatim as RFC 9728 `resource` and used as the OAuth audience binding; pad mounts the transport internally at /mcp and pad-cloud's nginx router transparently rewrites mcp.* root → /mcp so external clients see a single canonical URL regardless of internal path.
|
||||
AuthServerURL string `toml:"auth_server_url"` // Canonical URL of the OAuth authorization server (TASK-951), e.g. https://app.getpad.dev. Embedded in protected-resource metadata's authorization_servers field.
|
||||
// The push dispatch timing knobs (PAD_PUSH_DISPATCH_INTERVAL, _SETTLE,
|
||||
// _LEASE, _LEASE_MARGIN; TASK-3939 part C, ruling 77a8fae7), kept as the
|
||||
// raw env strings. They are parsed by PushDispatchTimings at server
|
||||
// startup, not in Load, so a malformed value refuses `pad server start`
|
||||
// rather than breaking every other pad command. Empty means unset: the
|
||||
// dispatcher keeps its default; a value of only whitespace is set and malformed,
|
||||
// not unset. Env only; there is no config.toml key.
|
||||
PushDispatchIntervalRaw string `toml:"-"`
|
||||
PushDispatchSettleRaw string `toml:"-"`
|
||||
PushDispatchLeaseRaw string `toml:"-"`
|
||||
PushDispatchLeaseMarginRaw string `toml:"-"`
|
||||
|
||||
MCPPublicURL string `toml:"mcp_public_url"` // Canonical URL clients paste into their MCP client (e.g. https://mcp.getpad.dev — matches mcp.stripe.com / mcp.linear.app convention). Published verbatim as RFC 9728 `resource` and used as the OAuth audience binding; pad mounts the transport internally at /mcp and pad-cloud's nginx router transparently rewrites mcp.* root → /mcp so external clients see a single canonical URL regardless of internal path.
|
||||
AuthServerURL string `toml:"auth_server_url"` // Canonical URL of the OAuth authorization server (TASK-951), e.g. https://app.getpad.dev. Embedded in protected-resource metadata's authorization_servers field.
|
||||
// OpenAIAppsChallenge is the domain-verification token OpenAI's plugin
|
||||
// dashboard issues, served verbatim at /.well-known/openai-apps-challenge
|
||||
// on cloud (TASK-3321 G4). Validated in ResolveMCPEndpoints.
|
||||
@@ -464,6 +477,10 @@ func Load() (*Config, error) {
|
||||
if v := os.Getenv("PAD_PUSH_GATEWAY_SECRET"); v != "" {
|
||||
cfg.PushGatewaySecret = v
|
||||
}
|
||||
cfg.PushDispatchIntervalRaw = os.Getenv("PAD_PUSH_DISPATCH_INTERVAL")
|
||||
cfg.PushDispatchSettleRaw = os.Getenv("PAD_PUSH_DISPATCH_SETTLE")
|
||||
cfg.PushDispatchLeaseRaw = os.Getenv("PAD_PUSH_DISPATCH_LEASE")
|
||||
cfg.PushDispatchLeaseMarginRaw = os.Getenv("PAD_PUSH_DISPATCH_LEASE_MARGIN")
|
||||
if v := os.Getenv("PAD_MCP_PUBLIC_URL"); v != "" {
|
||||
cfg.MCPPublicURL = v
|
||||
}
|
||||
@@ -891,3 +908,41 @@ func (c *Config) EnsureEncryptionKey(allowGenerate bool) error {
|
||||
func (c *Config) LogFile() string {
|
||||
return filepath.Join(c.DataDir, "logs", "server.log")
|
||||
}
|
||||
|
||||
// PushDispatchTimings is the parsed push dispatch timing knobs. A zero field
|
||||
// means the knob was unset and the dispatcher keeps its default.
|
||||
type PushDispatchTimings struct {
|
||||
Interval, Settle, Lease, LeaseMargin time.Duration
|
||||
}
|
||||
|
||||
// PushDispatchTimings parses the four PAD_PUSH_DISPATCH_* knobs (TASK-3939
|
||||
// part C, ruling 77a8fae7). A knob that is set but is not a Go duration, or is
|
||||
// not positive, is an error naming its variable; it never falls back to the
|
||||
// default. A value of only whitespace is malformed, not unset (only the empty
|
||||
// string is unset), so it refuses too (TASK-3939 part C, ruling 37e3eafa).
|
||||
// The message is fixed: it does not echo the value.
|
||||
func (c *Config) PushDispatchTimings() (PushDispatchTimings, error) {
|
||||
var t PushDispatchTimings
|
||||
for _, k := range []struct {
|
||||
env, raw string
|
||||
dst *time.Duration
|
||||
}{
|
||||
{"PAD_PUSH_DISPATCH_INTERVAL", c.PushDispatchIntervalRaw, &t.Interval},
|
||||
{"PAD_PUSH_DISPATCH_SETTLE", c.PushDispatchSettleRaw, &t.Settle},
|
||||
{"PAD_PUSH_DISPATCH_LEASE", c.PushDispatchLeaseRaw, &t.Lease},
|
||||
{"PAD_PUSH_DISPATCH_LEASE_MARGIN", c.PushDispatchLeaseMarginRaw, &t.LeaseMargin},
|
||||
} {
|
||||
if k.raw == "" {
|
||||
continue
|
||||
}
|
||||
d, err := time.ParseDuration(strings.TrimSpace(k.raw))
|
||||
if err != nil {
|
||||
return PushDispatchTimings{}, fmt.Errorf("%s is not a valid duration (use a Go duration such as 5s or 1m30s)", k.env)
|
||||
}
|
||||
if d <= 0 {
|
||||
return PushDispatchTimings{}, fmt.Errorf("%s must be greater than zero", k.env)
|
||||
}
|
||||
*k.dst = d
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func loadWithPushEnv(t *testing.T, env map[string]string) *Config {
|
||||
t.Helper()
|
||||
t.Setenv("PAD_DATA_DIR", t.TempDir())
|
||||
for _, k := range []string{"PAD_PUSH_DISPATCH_INTERVAL", "PAD_PUSH_DISPATCH_SETTLE", "PAD_PUSH_DISPATCH_LEASE", "PAD_PUSH_DISPATCH_LEASE_MARGIN"} {
|
||||
t.Setenv(k, env[k])
|
||||
}
|
||||
cfg, err := Load()
|
||||
if err != nil {
|
||||
t.Fatalf("Load = %v, want nil: a bad push knob must not break Load", err)
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// TASK-3939 part C: each env knob is read into its own timing field. Kills a
|
||||
// knob read from the wrong variable or dropped from the parse.
|
||||
func TestPushDispatchTimings_EachKnobIsParsed(t *testing.T) {
|
||||
cfg := loadWithPushEnv(t, map[string]string{
|
||||
"PAD_PUSH_DISPATCH_INTERVAL": "2s",
|
||||
"PAD_PUSH_DISPATCH_SETTLE": "3s",
|
||||
"PAD_PUSH_DISPATCH_LEASE": "1m30s",
|
||||
"PAD_PUSH_DISPATCH_LEASE_MARGIN": "20s",
|
||||
})
|
||||
got, err := cfg.PushDispatchTimings()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := PushDispatchTimings{Interval: 2 * time.Second, Settle: 3 * time.Second, Lease: 90 * time.Second, LeaseMargin: 20 * time.Second}
|
||||
if got != want {
|
||||
t.Fatalf("PushDispatchTimings = %+v, want %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPushDispatchTimings_UnsetIsZero(t *testing.T) {
|
||||
got, err := loadWithPushEnv(t, nil).PushDispatchTimings()
|
||||
if err != nil || got != (PushDispatchTimings{}) {
|
||||
t.Fatalf("PushDispatchTimings = %+v, %v, want zero and nil", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A set knob that is malformed or not positive is an error naming its
|
||||
// variable, never a silent default.
|
||||
func TestPushDispatchTimings_BadValueNamesTheVariable(t *testing.T) {
|
||||
vars := []string{"PAD_PUSH_DISPATCH_INTERVAL", "PAD_PUSH_DISPATCH_SETTLE", "PAD_PUSH_DISPATCH_LEASE", "PAD_PUSH_DISPATCH_LEASE_MARGIN"}
|
||||
for _, v := range vars {
|
||||
for _, bad := range []string{"soon", "10", "0", "0s", "-5s", " "} {
|
||||
t.Run(v+"="+bad, func(t *testing.T) {
|
||||
cfg := loadWithPushEnv(t, map[string]string{v: bad})
|
||||
_, err := cfg.PushDispatchTimings()
|
||||
if err == nil || !strings.Contains(err.Error(), v+" ") {
|
||||
t.Fatalf("PushDispatchTimings = %v, want an error naming %s", err, v)
|
||||
}
|
||||
if strings.Contains(err.Error(), bad+")") || strings.HasSuffix(err.Error(), bad) {
|
||||
t.Errorf("error echoes the value: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,9 @@ package server
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -20,7 +22,9 @@ import (
|
||||
// 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.
|
||||
// StartPushDispatch when PAD_PUSH_GATEWAY_URL is set, after applying the
|
||||
// PAD_PUSH_DISPATCH_* timing knobs through SetPushDispatchConfig (TASK-3939
|
||||
// part C); an unset knob keeps its default below.
|
||||
//
|
||||
// LEASE INVARIANT (r6-Q1): lease >= sendTimeout + leaseMargin. A claim is
|
||||
// stamped with second precision (up to 1s behind the real clock), and a Send
|
||||
@@ -29,7 +33,9 @@ import (
|
||||
// three ways: the setter refuses a violating config, all gateway I/O runs
|
||||
// under a deadline of start+lease-leaseMargin, and a chunk only starts when
|
||||
// sendTimeout is still left before that deadline (otherwise its rows are
|
||||
// unclaimed, no attempt burned).
|
||||
// unclaimed, no attempt burned). Startup applies the operator's knobs through
|
||||
// the setter, so a lease too short for sendTimeout (fixed at notify.SendTimeout,
|
||||
// not configurable) plus the margin refuses startup (TASK-3939 part C).
|
||||
//
|
||||
// SENT BUT UNRECORDED (TASK-4032): a failed 'sent' record is not merely logged.
|
||||
// The push went out, so the row would be reclaimed after the lease and pushed
|
||||
@@ -83,8 +89,11 @@ type pushDispatchConfig struct {
|
||||
now func() time.Time // test seam; nil = time.Now
|
||||
stop chan struct{}
|
||||
running bool
|
||||
instanceID string
|
||||
lastPrune time.Time // when the retention sweep last ran
|
||||
// startedInterval is the interval the ticker was last created with, 0 if
|
||||
// the loop never started (TASK-3939 part C: proves the knob reached Start).
|
||||
startedInterval time.Duration
|
||||
instanceID string
|
||||
lastPrune time.Time // when the retention sweep last ran
|
||||
// unrecorded holds rows the gateway accepted whose 'sent' record failed
|
||||
// (TASK-4032); see parkUnrecorded.
|
||||
unrecorded map[string]unrecordedRow
|
||||
@@ -116,11 +125,15 @@ type pushDispatchSettings struct {
|
||||
}
|
||||
|
||||
// SetPushDispatchConfig overrides the timings; zero keeps the current value.
|
||||
// Server startup calls it from the PAD_PUSH_DISPATCH_* knobs (cmd/pad
|
||||
// configurePush, TASK-3939 part C) and passes 0 for sendTimeout, which has no
|
||||
// knob because notify.SendTimeout is baked into the gateway's HTTP client.
|
||||
// interval is read once, by StartPushDispatch (the ticker is fixed at Start,
|
||||
// so a later change needs a restart); settle, lease, leaseMargin and
|
||||
// sendTimeout are read on every tick.
|
||||
// A combination violating the lease invariant is refused whole and nothing
|
||||
// changes.
|
||||
// changes; the error wraps errPushLeaseTooShort and states the RESOLVED lease,
|
||||
// send timeout and margin, so a caller can tell which knob caused it.
|
||||
func (s *Server) SetPushDispatchConfig(interval, settle, lease, leaseMargin, sendTimeout time.Duration) error {
|
||||
cur := s.resolvePushDispatchSettings()
|
||||
pick := func(v, d time.Duration) time.Duration {
|
||||
@@ -134,9 +147,15 @@ func (s *Server) SetPushDispatchConfig(interval, settle, lease, leaseMargin, sen
|
||||
lease: pick(lease, cur.lease), leaseMargin: pick(leaseMargin, cur.leaseMargin),
|
||||
sendTimeout: pick(sendTimeout, cur.sendTimeout),
|
||||
}
|
||||
if next.lease < next.sendTimeout+next.leaseMargin {
|
||||
// sendTimeout + leaseMargin wraps negative when the margin is within
|
||||
// sendTimeout of MaxInt64 (ParseDuration accepts up to about 2562047h), and
|
||||
// the wrapped sum would pass the comparison below. Such a margin exceeds
|
||||
// every representable lease, so it is the same violation (TASK-3939 part C,
|
||||
// ruling 37e3eafa). Every other use of these durations is time.Time.Add or
|
||||
// NewTicker, which cannot wrap at this magnitude, so no cap is needed.
|
||||
if next.leaseMargin > math.MaxInt64-next.sendTimeout || next.lease < next.sendTimeout+next.leaseMargin {
|
||||
slog.Warn("push dispatch: config refused", "reason", "lease shorter than send timeout plus margin")
|
||||
return errPushLeaseTooShort
|
||||
return fmt.Errorf("%w: lease %s is shorter than send timeout %s plus margin %s", errPushLeaseTooShort, next.lease, next.sendTimeout, next.leaseMargin)
|
||||
}
|
||||
d := &s.pushDispatch
|
||||
d.mu.Lock()
|
||||
@@ -213,6 +232,7 @@ func (s *Server) StartPushDispatch() {
|
||||
d.stop = make(chan struct{})
|
||||
d.running = true
|
||||
stop, tick := d.stop, d.tick
|
||||
d.startedInterval = cfg.interval
|
||||
d.mu.Unlock()
|
||||
|
||||
s.bg.Add(1)
|
||||
@@ -256,6 +276,26 @@ func (s *Server) PushDispatchRunning() bool {
|
||||
return s.pushDispatch.running
|
||||
}
|
||||
|
||||
// PushDispatchApplied is what the dispatcher will use (or is using). Interval
|
||||
// is the stored setting; StartedInterval is the interval the ticker was
|
||||
// actually created with, 0 if the loop never started.
|
||||
type PushDispatchApplied struct {
|
||||
Interval, Settle, Lease, LeaseMargin, StartedInterval time.Duration
|
||||
}
|
||||
|
||||
// PushDispatchApplied reads the dispatcher's timings, filling defaults for
|
||||
// any unset one. Read-only; startup's wiring test reads it to prove the knobs
|
||||
// reached the dispatcher, and the interval before Start (TASK-3939 part C,
|
||||
// ruling 77a8fae7). A test-only seam, read at single-threaded startup: it
|
||||
// takes the lock twice (resolve, then startedInterval), so a concurrent
|
||||
// reconfigure could give a mixed snapshot.
|
||||
func (s *Server) PushDispatchApplied() PushDispatchApplied {
|
||||
cur := s.resolvePushDispatchSettings()
|
||||
s.pushDispatch.mu.Lock()
|
||||
defer s.pushDispatch.mu.Unlock()
|
||||
return PushDispatchApplied{cur.interval, cur.settle, cur.lease, cur.leaseMargin, s.pushDispatch.startedInterval}
|
||||
}
|
||||
|
||||
// stopPushDispatch signals the loop to exit. Safe when it never started.
|
||||
func (s *Server) stopPushDispatch() {
|
||||
d := &s.pushDispatch
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -120,6 +121,18 @@ func TestPushDispatch_LeaseExceedsSendTimeout(t *testing.T) {
|
||||
if err := srv.SetPushDispatchConfig(0, 0, cfg.sendTimeout+cfg.leaseMargin, 0, 0); err != nil {
|
||||
t.Fatalf("a lease exactly at the invariant was refused: %v", err)
|
||||
}
|
||||
// A margin near MaxInt64 would wrap sendTimeout+margin negative and pass
|
||||
// the comparison; it must refuse and change nothing (TASK-3939 part C,
|
||||
// ruling 37e3eafa).
|
||||
for _, margin := range []time.Duration{2562047 * time.Hour, time.Duration(math.MaxInt64) - cfg.sendTimeout + 1, time.Duration(math.MaxInt64)} {
|
||||
if err := srv.SetPushDispatchConfig(0, 0, 0, margin, 0); !errors.Is(err, errPushLeaseTooShort) {
|
||||
t.Fatalf("margin %v: err = %v, want refused", margin, err)
|
||||
}
|
||||
}
|
||||
// The error states the resolved values.
|
||||
if err := srv.SetPushDispatchConfig(0, 0, 30*time.Second, 0, 0); err == nil || !strings.Contains(err.Error(), "lease 30s") || !strings.Contains(err.Error(), "margin 15s") {
|
||||
t.Fatalf("err = %v, want the resolved lease and margin", err)
|
||||
}
|
||||
// Raising the send timeout past the lease is the same violation.
|
||||
if err := srv.SetPushDispatchConfig(0, 0, 0, 0, time.Hour); !errors.Is(err, errPushLeaseTooShort) {
|
||||
t.Fatalf("send timeout raised past the lease: err = %v", err)
|
||||
|
||||
Reference in New Issue
Block a user