diff --git a/limit/service.go b/limit/service.go index 9f6102db..95cfe431 100644 --- a/limit/service.go +++ b/limit/service.go @@ -60,7 +60,6 @@ const ( // Service decides on usage from memory; see the package comment type Service struct { - conf *Config now func() time.Time // Injectable clock, for tests counters [shardCount]*counterShard // Daily counts, kept until their period ends buckets [shardCount]*bucketShard // Token buckets, per node, evicted when idle and full @@ -74,16 +73,16 @@ type Service struct { wg sync.WaitGroup } -// New creates the service -func New(conf *Config) *Service { - return newService(conf, time.Now) +// New creates the service; the store shares the daily counts and keeps them across restarts, nil +// for memory only +func New(store *Store) *Service { + return newService(store, time.Now) } -func newService(conf *Config, now func() time.Time) *Service { +func newService(store *Store, now func() time.Time) *Service { s := &Service{ - conf: conf, now: now, - store: conf.Store, + store: store, interval: syncInterval, } s.ctx, s.cancel = context.WithCancel(context.Background()) diff --git a/limit/service_test.go b/limit/service_test.go index cb7d649c..44d3c971 100644 --- a/limit/service_test.go +++ b/limit/service_test.go @@ -36,7 +36,7 @@ func (c *testClock) advance(d time.Duration) { func newTestServiceWithClock(t *testing.T) (*Service, *testClock) { t.Helper() clock := &testClock{t: time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC)} - return newService(&Config{}, clock.now), clock + return newService(nil, clock.now), clock } func newTestService(t *testing.T) *Service { @@ -348,7 +348,7 @@ func TestService_Prune_DropsEndedPeriods(t *testing.T) { } func TestService_Allow_Concurrent(t *testing.T) { - s := New(&Config{}) + s := New(nil) rule := Rule{ Kind: KindMessages, Burst: 500, @@ -383,7 +383,7 @@ func TestService_Allow_Concurrent(t *testing.T) { } func BenchmarkAllow_HotKey(b *testing.B) { - s := New(&Config{}) + s := New(nil) rule := Rule{ Kind: KindRequests, Burst: 60, @@ -398,7 +398,7 @@ func BenchmarkAllow_HotKey(b *testing.B) { } func BenchmarkAllow_ManyKeys(b *testing.B) { - s := New(&Config{}) + s := New(nil) rule := Rule{ Kind: KindRequests, Burst: 60, @@ -417,7 +417,7 @@ func BenchmarkAllow_ManyKeys(b *testing.B) { } func BenchmarkAllow_Parallel(b *testing.B) { - s := New(&Config{}) + s := New(nil) rule := Rule{ Kind: KindRequests, Burst: 60, @@ -623,7 +623,7 @@ func TestService_ResetAt(t *testing.T) { // on its own: the tests sync by hand. It is closed when the test ends. func startTestService(t *testing.T, store *Store, now func() time.Time) *Service { t.Helper() - s := newService(&Config{Store: store}, now) + s := newService(store, now) s.interval = time.Hour require.Nil(t, s.Start(context.Background())) t.Cleanup(s.Close) @@ -701,7 +701,7 @@ func TestService_Persist_UpdatedAtIsTheNodesTime(t *testing.T) { require.True(t, s.Allow("user:u_1", Rule{Kind: KindMessages}, 1).OK) s.sync() var updatedAt int64 - require.Nil(t, s.conf.Store.db.QueryRow(`SELECT updated_at FROM usage WHERE key = 'user:u_1'`).Scan(&updatedAt)) + require.Nil(t, s.store.db.QueryRow(`SELECT updated_at FROM usage WHERE key = 'user:u_1'`).Scan(&updatedAt)) require.Equal(t, ahead.Unix(), updatedAt) }) } @@ -731,7 +731,7 @@ func TestService_Persist_TwoNodesConverge(t *testing.T) { b.sync() require.False(t, b.Allow("k", rule, 1).OK) require.Equal(t, int64(6), b.Usage("k").Count(KindMessages)) - require.Equal(t, int64(6), usageRowOf(t, a.conf.Store, "k").counts[0]) + require.Equal(t, int64(6), usageRowOf(t, a.store, "k").counts[0]) }) } @@ -745,13 +745,13 @@ func TestService_Persist_ThresholdPromotes(t *testing.T) { } require.True(t, s.Allow("ip:1.2.3.4", rule, 24).OK) s.sync() - require.Equal(t, usageRow{}, usageRowOf(t, s.conf.Store, "ip:1.2.3.4")) + require.Equal(t, usageRow{}, usageRowOf(t, s.store, "ip:1.2.3.4")) require.True(t, s.Allow("ip:1.2.3.4", rule, 1).OK) s.sync() - require.Equal(t, int64(25), usageRowOf(t, s.conf.Store, "ip:1.2.3.4").counts[0]) + require.Equal(t, int64(25), usageRowOf(t, s.store, "ip:1.2.3.4").counts[0]) require.True(t, s.Allow("ip:1.2.3.4", rule, 1).OK) s.sync() - require.Equal(t, int64(26), usageRowOf(t, s.conf.Store, "ip:1.2.3.4").counts[0]) + require.Equal(t, int64(26), usageRowOf(t, s.store, "ip:1.2.3.4").counts[0]) }) } @@ -772,7 +772,7 @@ func TestService_Persist_ReadBackPromotes(t *testing.T) { a.sync() // Reads b's row require.Equal(t, int64(40), a.Usage("ip:1.2.3.4").Count(KindMessages)) a.sync() // Writes its own 10 - require.Equal(t, int64(40), usageRowOf(t, a.conf.Store, "ip:1.2.3.4").counts[0]) + require.Equal(t, int64(40), usageRowOf(t, a.store, "ip:1.2.3.4").counts[0]) b.sync() require.Equal(t, int64(40), b.Usage("ip:1.2.3.4").Count(KindMessages)) }) @@ -869,7 +869,7 @@ func TestService_Persist_StartupWithoutStore(t *testing.T) { require.Nil(t, err) _, err = store.db.Exec(`ALTER TABLE usage RENAME TO usage_down`) require.Nil(t, err) - s := newService(&Config{Store: store}, time.Now) + s := newService(store, time.Now) s.interval = 50 * time.Millisecond t.Cleanup(s.Close) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) @@ -948,7 +948,7 @@ func TestService_Allow_DenyRacesRollover(t *testing.T) { // roll the counter over; run with -race var ticks atomic.Int64 start := time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC) - s := newService(&Config{}, func() time.Time { + s := newService(nil, func() time.Time { return start.Add(time.Duration(ticks.Add(1)) * time.Hour) }) rule := Rule{ @@ -978,7 +978,7 @@ func TestService_Close_CancelsStalledLoad(t *testing.T) { conn, err := pool.Conn(context.Background()) // Holds the only connection: the load waits for it require.Nil(t, err) defer conn.Close() - s := newService(&Config{Store: store}, time.Now) + s := newService(store, time.Now) ctx, cancel := context.WithCancel(context.Background()) cancel() require.Nil(t, s.Start(ctx)) @@ -1021,7 +1021,7 @@ var ( func newBenchService(b *testing.B) *Service { b.Helper() - s := newService(&Config{}, time.Now) + s := newService(nil, time.Now) for i := 0; i < benchIdleKeys; i++ { s.Allow(fmt.Sprintf("ip:10.%d.%d.%d", i>>16, (i>>8)&255, i&255), benchRequestRule, 1) } diff --git a/limit/types.go b/limit/types.go index f89af1d4..91714384 100644 --- a/limit/types.go +++ b/limit/types.go @@ -53,11 +53,6 @@ type Usage struct { counts [numKinds]int64 } -// Config configures the service -type Config struct { - Store *Store // Where the daily counts are shared and kept across restarts; nil for memory only -} - // counter is one key's daily usage in its current period type counter struct { periodStart int64 // When the period started (unix seconds, see periodStart) diff --git a/server/server.go b/server/server.go index 34ac55d1..cb568273 100644 --- a/server/server.go +++ b/server/server.go @@ -358,13 +358,9 @@ func New(conf *Config) (*Server, error) { firebaseClient = newFirebaseClient(sender, auther) } s.userManager = userManager - s.limiter = limit.New(&limit.Config{ - Store: createUsageStore(pool, userDB), - }) + s.limiter = limit.New(createUsageStore(pool, userDB)) s.visitors = newVisitorEnv(conf, s.limiter, messageCache, userManager) - s.topicService = tp.New(&tp.Config{ - Store: createTopicStore(pool, userDB), - }) + s.topicService = tp.New(createTopicStore(pool, userDB)) s.firebaseClient = firebaseClient if pool != nil { metrics.SetDBPoolsFunc(s.dbPoolStats) diff --git a/server/server_firebase_test.go b/server/server_firebase_test.go index e0e7cea0..4f7be92f 100644 --- a/server/server_firebase_test.go +++ b/server/server_firebase_test.go @@ -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(newVisitorEnv(newTestConfig(t, ""), limit.New(&limit.Config{}), newMemTestCache(t), nil), netip.MustParseAddr("1.2.3.4"), nil) + visitor := newVisitor(newVisitorEnv(newTestConfig(t, ""), limit.New(nil), 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())) diff --git a/server/server_rate_visitor_test.go b/server/server_rate_visitor_test.go index 2ba15144..1b8c8fd0 100644 --- a/server/server_rate_visitor_test.go +++ b/server/server_rate_visitor_test.go @@ -426,7 +426,7 @@ func ageTopic(t *testing.T, topic string, d time.Duration, servers ...*Server) { if c.DatabaseURL != "" { store = tp.NewPostgresStore(userDB) } - topicService := tp.New(&tp.Config{Store: store}) + topicService := tp.New(store) require.Nil(t, topicService.Start(context.Background())) s.topicService = topicService t.Cleanup(func() { diff --git a/topic/service.go b/topic/service.go index e9f977bd..9fa5524d 100644 --- a/topic/service.go +++ b/topic/service.go @@ -59,9 +59,10 @@ type Service struct { mu sync.Mutex // Protects entries, pending and noted } -// New creates the service -func New(conf *Config) *Service { - return newService(conf.Store, time.Now) +// New creates the service; the store shares the rows with the other nodes and keeps them across +// restarts, nil for memory only +func New(store *Store) *Service { + return newService(store, time.Now) } func newService(store *Store, now func() time.Time) *Service { diff --git a/topic/service_test.go b/topic/service_test.go index 2cb295ea..4fb118d1 100644 --- a/topic/service_test.go +++ b/topic/service_test.go @@ -458,7 +458,7 @@ func TestService_Persist_PrunesTableAndMemory(t *testing.T) { } func TestService_WithoutStore(t *testing.T) { - s := New(&Config{}) + s := New(nil) require.Nil(t, s.Start(context.Background())) s.Note("upA") topic, err := s.Refresh(context.Background(), "upA") diff --git a/topic/types.go b/topic/types.go index 8f426b66..a64177a2 100644 --- a/topic/types.go +++ b/topic/types.go @@ -1,10 +1,5 @@ package topic -// Config configures the service -type Config struct { - Store *Store // Where the rows are shared with the other nodes and kept across restarts; nil for memory only -} - // State is one topic's row of the topic table, as this node knows it. Times are unix seconds. type State struct { Topic string