mirror of
https://github.com/binwiederhier/ntfy.git
synced 2026-11-10 14:44:04 +08:00
Build the shared visitor config and IP limits once
Visitors are built per request, so what they share (the keying and limit config, the config-based limits of visitors without a tier, the limiter and the subscription gauge) is built once in New. Keys are built with string concatenation instead of fmt.Sprintf.
This commit is contained in:
1 parent
f90c97f3de
commit
a6cd0e4d7e
7 files changed
+49
-33
No files matched your search
+6
-6
@@ -64,9 +64,9 @@ type Server struct {
|
||||
smtpServerBackend *smtpBackend
|
||||
mailer mail.Sender
|
||||
topics map[string]*topic
|
||||
limiter *limit.Service // Decides on and counts the visitors' usage, keyed by visitor key
|
||||
subscriptions *subscriptionGauge // Open subscriptions per visitor key on this node
|
||||
ban *ban.Service // Abuse ban-feed; nil when the feature is disabled (no ban file)
|
||||
limiter *limit.Service // Decides on and counts the visitors' usage, keyed by visitor key
|
||||
visitorEnv *visitorEnv // Shared by all visitors, which are built per request
|
||||
ban *ban.Service // Abuse ban-feed; nil when the feature is disabled (no ban file)
|
||||
firebaseClient *firebaseClient
|
||||
twilio *twilio.Client
|
||||
messages int64 // Total number of messages (persisted if messageCache enabled)
|
||||
@@ -301,7 +301,6 @@ func New(conf *Config) (*Server, error) {
|
||||
messages: messages,
|
||||
messagesHistory: []int64{messages},
|
||||
messagesFlushed: messages, // The loaded total is already persisted; only new publishes are deltas
|
||||
subscriptions: newSubscriptionGauge(),
|
||||
limiter: limit.New(&limit.Config{ResetTime: conf.VisitorStatsResetTime}),
|
||||
stripe: stripe,
|
||||
streams: newStreamRegistry(),
|
||||
@@ -349,6 +348,7 @@ func New(conf *Config) (*Server, error) {
|
||||
firebaseClient = newFirebaseClient(sender, auther)
|
||||
}
|
||||
s.userManager = userManager
|
||||
s.visitorEnv = newVisitorEnv(conf, s.limiter, messageCache, userManager)
|
||||
s.firebaseClient = firebaseClient
|
||||
if pool != nil {
|
||||
metrics.SetDBPoolsFunc(s.dbPoolStats)
|
||||
@@ -2171,7 +2171,7 @@ func (s *Server) runFirebaseKeepaliver() {
|
||||
if s.firebaseClient == nil {
|
||||
return
|
||||
}
|
||||
v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0
|
||||
v := newVisitor(s.visitorEnv, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0
|
||||
for {
|
||||
select {
|
||||
case <-time.After(s.config.FirebaseKeepaliveInterval):
|
||||
@@ -2353,7 +2353,7 @@ func (s *Server) transformMatrixJSON(next handleFunc) handleFunc {
|
||||
// visitor returns the visitor for this request. It is built per request, so concurrent requests from
|
||||
// the same IP (which share a limiter key) cannot see each other's user (#1051).
|
||||
func (s *Server) visitor(ip netip.Addr, user *user.User) *visitor {
|
||||
return newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, ip, user)
|
||||
return newVisitor(s.visitorEnv, ip, user)
|
||||
}
|
||||
|
||||
func (s *Server) writeJSON(w http.ResponseWriter, v any) error {
|
||||
|
||||
@@ -346,7 +346,7 @@ func TestMaybeTruncateFCMMessage_NotTooLong(t *testing.T) {
|
||||
func TestToFirebaseSender_Abuse(t *testing.T) {
|
||||
sender := &testFirebaseSender{allowed: 2}
|
||||
client := newFirebaseClient(sender, &testAuther{})
|
||||
visitor := newVisitor(newTestConfig(t, ""), limit.New(&limit.Config{}), newSubscriptionGauge(), newMemTestCache(t), nil, netip.MustParseAddr("1.2.3.4"), nil)
|
||||
visitor := newVisitor(newVisitorEnv(newTestConfig(t, ""), limit.New(&limit.Config{}), newMemTestCache(t), nil), netip.MustParseAddr("1.2.3.4"), nil)
|
||||
|
||||
require.Nil(t, client.Send(visitor, &model.Message{Topic: "mytopic"}))
|
||||
require.Equal(t, 1, len(sender.Messages()))
|
||||
|
||||
@@ -63,7 +63,7 @@ func (s *Server) handleBillingTiersGet(w http.ResponseWriter, _ *http.Request, _
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
freeTier := vis.LimitsFor(visitorConfigFrom(s.config), nil)
|
||||
freeTier := s.visitorEnv.ipLimits
|
||||
response := []*apiAccountBillingTier{
|
||||
{
|
||||
// This is a bit of a hack: This is the "Free" tier. It has no tier code, name or price.
|
||||
|
||||
@@ -395,9 +395,10 @@ func TestPayments_Checkout_Success_And_Increase_Rate_Limits_Reset_Visitor(t *tes
|
||||
require.Equal(t, 429, rr.Code)
|
||||
|
||||
// Now let's test the message limiter by faking a ridiculously generous request limit; the
|
||||
// tier's request burst never goes below the configured one, and visitors are built per request
|
||||
s.config.VisitorRequestLimitBurst = 1000000
|
||||
s.config.VisitorRequestLimitReplenish = time.Nanosecond
|
||||
// tier's request burst never goes below the configured one, and tier limits are resolved
|
||||
// per request from the visitor config the server built at startup
|
||||
s.visitorEnv.limitsConfig.RequestLimitBurst = 1000000
|
||||
s.visitorEnv.limitsConfig.RequestLimitReplenish = time.Nanosecond
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 209; i++ {
|
||||
|
||||
@@ -5575,7 +5575,7 @@ func TestServer_SubscribeHTTP_NoWriteAfterHandlerReturn(t *testing.T) {
|
||||
// that copied the subscriber list before Unsubscribe ran. In production, this is exactly
|
||||
// how the panic occurs: the goroutine spawned by topic.Publish calls sub() after the
|
||||
// handler has already returned and Go has cleaned up the response writer.
|
||||
v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.MustParseAddr("9.9.9.9"), nil)
|
||||
v := newVisitor(s.visitorEnv, netip.MustParseAddr("9.9.9.9"), nil)
|
||||
msg := model.NewDefaultMessage("mytopic", "straggler message")
|
||||
_ = copiedSub(v, msg)
|
||||
|
||||
@@ -5595,7 +5595,7 @@ func TestServer_HandleError_SkipsWriteHeaderOnHijackedConnection(t *testing.T) {
|
||||
r, _ := http.NewRequest("GET", "/mytopic/ws", nil)
|
||||
r.Header.Set("Upgrade", "websocket")
|
||||
r.Header.Set("Connection", "Upgrade")
|
||||
v := newVisitor(s.config, s.limiter, s.subscriptions, s.messageCache, s.userManager, netip.MustParseAddr("1.2.3.4"), nil)
|
||||
v := newVisitor(s.visitorEnv, netip.MustParseAddr("1.2.3.4"), nil)
|
||||
|
||||
// Test post-upgrade errors wrapped with errWebSocketPostUpgrade (should NOT call WriteHeader)
|
||||
postUpgradeErr := &errWebSocketPostUpgrade{errors.New("websocket: close 1000 (normal)")}
|
||||
|
||||
+33
-17
@@ -28,16 +28,23 @@ const (
|
||||
// same IP cannot see each other's user (#1051). The counting happens in the limit service, keyed by
|
||||
// key, so two visitors with the same key share one set of buckets and counts.
|
||||
type visitor struct {
|
||||
*visitorEnv
|
||||
key string // The limiter key: "ip:<prefix>" or "user:<id>", see vis.Key
|
||||
ip netip.Addr // Visitor IP address
|
||||
user *user.User // Only set if authenticated user, otherwise nil
|
||||
limits *vis.Limits // Resolved from the config or the user's tier
|
||||
}
|
||||
|
||||
// visitorEnv is what every visitor of a server shares; it is built once, so building a visitor per
|
||||
// request costs little
|
||||
type visitorEnv struct {
|
||||
config *Config
|
||||
limitsConfig *vis.Config // The parts of config that keying and limit resolution need (see visitorConfigFrom)
|
||||
ipLimits *vis.Limits // The config-based limits of visitors without a tier; never modified
|
||||
limiter *limit.Service // Decides on and counts usage, keyed by key
|
||||
subscriptions *subscriptionGauge
|
||||
messageCache *message.Cache
|
||||
userManager *user.Manager // May be nil
|
||||
key string // The limiter key: "ip:<prefix>" or "user:<id>", see vis.Key
|
||||
ip netip.Addr // Visitor IP address
|
||||
user *user.User // Only set if authenticated user, otherwise nil
|
||||
limits *vis.Limits // Resolved from the config or the user's tier
|
||||
}
|
||||
|
||||
// subscriptionGauge counts the open subscriptions per visitor key on this node. It is never shared
|
||||
@@ -65,23 +72,32 @@ type visitorStats struct {
|
||||
AttachmentTotalSizeRemaining int64
|
||||
}
|
||||
|
||||
func newVisitor(conf *Config, limiter *limit.Service, subscriptions *subscriptionGauge, messageCache *message.Cache, userManager *user.Manager, ip netip.Addr, u *user.User) *visitor {
|
||||
func newVisitorEnv(conf *Config, limiter *limit.Service, messageCache *message.Cache, userManager *user.Manager) *visitorEnv {
|
||||
limitsConfig := visitorConfigFrom(conf)
|
||||
var tier *user.Tier
|
||||
if u != nil {
|
||||
tier = u.Tier // May be nil, which resolves the config-based limits
|
||||
}
|
||||
v := &visitor{
|
||||
return &visitorEnv{
|
||||
config: conf,
|
||||
limitsConfig: limitsConfig,
|
||||
ipLimits: vis.LimitsFor(limitsConfig, nil),
|
||||
limiter: limiter,
|
||||
subscriptions: subscriptions,
|
||||
subscriptions: newSubscriptionGauge(),
|
||||
messageCache: messageCache,
|
||||
userManager: userManager, // May be nil
|
||||
key: vis.Key(ip, u, limitsConfig),
|
||||
ip: ip,
|
||||
user: u,
|
||||
limits: vis.LimitsFor(limitsConfig, tier),
|
||||
}
|
||||
}
|
||||
|
||||
func newVisitor(env *visitorEnv, ip netip.Addr, u *user.User) *visitor {
|
||||
limits := env.ipLimits
|
||||
var tier *user.Tier
|
||||
if u != nil && u.Tier != nil {
|
||||
tier = u.Tier
|
||||
limits = vis.LimitsFor(env.limitsConfig, tier) // Resolved per request: the tier's fields may change at any time
|
||||
}
|
||||
v := &visitor{
|
||||
visitorEnv: env,
|
||||
key: vis.Key(ip, u, env.limitsConfig),
|
||||
ip: ip,
|
||||
user: u,
|
||||
limits: limits,
|
||||
}
|
||||
if tier != nil {
|
||||
// Carry the persisted stats into the key's daily counts, if the key is new to the limiter
|
||||
@@ -89,9 +105,9 @@ func newVisitor(conf *Config, limiter *limit.Service, subscriptions *subscriptio
|
||||
usage.Set(limit.KindMessages, u.Stats.Messages)
|
||||
usage.Set(limit.KindEmails, u.Stats.Emails)
|
||||
usage.Set(limit.KindCalls, u.Stats.Calls)
|
||||
limiter.Seed(v.key, usage)
|
||||
env.limiter.Seed(v.key, usage)
|
||||
}
|
||||
limiter.Touch(v.key) // The key is seen now, as the visitor's request is (after Seed, which only fills unknown keys)
|
||||
env.limiter.Touch(v.key) // The key is seen now, as the visitor's request is (after Seed, which only fills unknown keys)
|
||||
return v
|
||||
}
|
||||
|
||||
|
||||
+2
-3
@@ -1,7 +1,6 @@
|
||||
package visitor
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/netip"
|
||||
|
||||
"heckel.io/ntfy/v2/user"
|
||||
@@ -12,12 +11,12 @@ import (
|
||||
// tier shares the IP key with everyone behind the same prefix.
|
||||
func Key(ip netip.Addr, u *user.User, conf *Config) string {
|
||||
if u != nil && u.Tier != nil {
|
||||
return fmt.Sprintf("user:%s", u.ID)
|
||||
return "user:" + u.ID
|
||||
}
|
||||
if ip.Is4() {
|
||||
ip = netip.PrefixFrom(ip, conf.PrefixBitsIPv4).Masked().Addr()
|
||||
} else if ip.Is6() {
|
||||
ip = netip.PrefixFrom(ip, conf.PrefixBitsIPv6).Masked().Addr()
|
||||
}
|
||||
return fmt.Sprintf("ip:%s", ip.String())
|
||||
return "ip:" + ip.String()
|
||||
}
|
||||
Reference in new issue
Block a user