perf: Subscriber performance

.golangci.yml

@@ -15,6 +15,7 @@ linters:
     - cyclop
     - forcetypeassert
     - tagliatelle
+    - varnamelen
 
     # deprecated
     - interfacer

CONTRIBUTING.md

@@ -5,7 +5,7 @@
 When you open a Pull Request to the project, you agree to license your code under the [GNU AFFERO GENERAL PUBLIC LICENSE](LICENSE)
 and to transfer the copyright on the submitted code to [Kévin Dunglas](https://dunglas.fr).
 
-Be sure to you have the right to do that (if you are a professional, ask your company)!
+Be sure to have the right to do that (if you are a professional, ask your company)!
 
 If you include code from another project, please mention it in the Pull Request description and credit the original author.
 
@@ -59,7 +59,7 @@ To test the legacy server:
 
 Go to `http://localhost:3000` and enjoy!
 
-When you send a PR, just make sure that:
+When you send a PR, make sure that:
 
 * You add valid test cases.
 * Tests are green.
@@ -73,7 +73,7 @@ To debug potential deadlocks:
 
 1. Install `go-deadlock`: `./tests/use-go-deadlock.sh`
 2. Run the tests in race mode: `go test -race ./... -v`
-3. To stress test the app, run the load test (see `docs/load-testing.md`)
+3. To stress-test the app, run the load test (see `docs/load-testing.md`)
 4. Be sure to remove `go-deadlock` before committing
 
 ## Spec

authorization.go

@@ -60,7 +60,7 @@ func authorize(r *http.Request, jwtConfig *jwtConfig, publishOrigins []string) (
 	cookie, err := r.Cookie("mercureAuthorization")
 	if err != nil {
 		// Anonymous
-		return nil, nil //nolint:nilerr
+		return nil, nil //nolint:nilerr,nilnil
 	}
 
 	// CSRF attacks cannot occur when using safe methods

bolt_transport.go

@@ -39,7 +39,7 @@ type BoltTransport struct {
 }
 
 // NewBoltTransport create a new boltTransport.
-func NewBoltTransport(u *url.URL, l Logger, tss *TopicSelectorStore) (Transport, error) {
+func NewBoltTransport(u *url.URL, l Logger, tss *TopicSelectorStore) (Transport, error) { //nolint:ireturn
 	var err error
 	q := u.Query()
 	bucketName := defaultBoltBucketName
@@ -190,6 +190,23 @@ func (t *BoltTransport) AddSubscriber(s *Subscriber) error {
 		t.dispatchHistory(s, toSeq)
 	}
 
+	s.Ready()
+
+	return nil
+}
+
+// RemoveSubscriber removes a new subscriber from the transport.
+func (t *BoltTransport) RemoveSubscriber(s *Subscriber) error {
+	select {
+	case <-t.closed:
+		return ErrClosedTransport
+	default:
+	}
+
+	t.Lock()
+	delete(t.subscribers, s)
+	t.Unlock()
+
 	return nil
 }
 

bolt_transport_test.go

@@ -35,9 +35,8 @@ func TestBoltTransportHistory(t *testing.T) {
 		})
 	}
 
-	s := NewSubscriber("8", transport.logger, &TopicSelectorStore{})
-	s.Topics = topics
-	go s.start()
+	s := NewSubscriber("8", transport.logger)
+	s.SetTopics(topics, nil)
 
 	require.Nil(t, transport.AddSubscriber(s))
 
@@ -66,9 +65,8 @@ func TestBoltTransportRetrieveAllHistory(t *testing.T) {
 		})
 	}
 
-	s := NewSubscriber(EarliestLastEventID, transport.logger, &TopicSelectorStore{})
-	s.Topics = topics
-	go s.start()
+	s := NewSubscriber(EarliestLastEventID, transport.logger)
+	s.SetTopics(topics, nil)
 	require.Nil(t, transport.AddSubscriber(s))
 
 	var count int
@@ -78,9 +76,10 @@ func TestBoltTransportRetrieveAllHistory(t *testing.T) {
 		count++
 		assert.Equal(t, strconv.Itoa(count), u.ID)
 		if count == 10 {
-			return
+			break
 		}
 	}
+	assert.Equal(t, 10, count)
 }
 
 func TestBoltTransportHistoryAndLive(t *testing.T) {
@@ -96,9 +95,8 @@ func TestBoltTransportHistoryAndLive(t *testing.T) {
 		})
 	}
 
-	s := NewSubscriber("8", transport.logger, &TopicSelectorStore{})
-	s.Topics = topics
-	go s.start()
+	s := NewSubscriber("8", transport.logger)
+	s.SetTopics(topics, nil)
 	require.Nil(t, transport.AddSubscriber(s))
 
 	var wg sync.WaitGroup
@@ -179,8 +177,7 @@ func TestBoltTransportDoNotDispatchUntilListen(t *testing.T) {
 	defer os.Remove("test.db")
 	assert.Implements(t, (*Transport)(nil), transport)
 
-	s := NewSubscriber("", transport.logger, &TopicSelectorStore{})
-	go s.start()
+	s := NewSubscriber("", transport.logger)
 	require.Nil(t, transport.AddSubscriber(s))
 
 	var wg sync.WaitGroup
@@ -204,9 +201,8 @@ func TestBoltTransportDispatch(t *testing.T) {
 	defer os.Remove("test.db")
 	assert.Implements(t, (*Transport)(nil), transport)
 
-	s := NewSubscriber("", transport.logger, &TopicSelectorStore{})
-	s.Topics = []string{"https://example.com/foo"}
-	go s.start()
+	s := NewSubscriber("", transport.logger)
+	s.SetTopics([]string{"https://example.com/foo"}, nil)
 
 	require.Nil(t, transport.AddSubscriber(s))
 
@@ -222,9 +218,8 @@ func TestBoltTransportClosed(t *testing.T) {
 	defer os.Remove("test.db")
 	assert.Implements(t, (*Transport)(nil), transport)
 
-	s := NewSubscriber("", transport.logger, &TopicSelectorStore{})
-	s.Topics = []string{"https://example.com/foo"}
-	go s.start()
+	s := NewSubscriber("", transport.logger)
+	s.SetTopics([]string{"https://example.com/foo"}, nil)
 	require.Nil(t, transport.AddSubscriber(s))
 
 	require.Nil(t, transport.Close())
@@ -242,13 +237,10 @@ func TestBoltCleanDisconnectedSubscribers(t *testing.T) {
 	defer transport.Close()
 	defer os.Remove("test.db")
 
-	tss := &TopicSelectorStore{}
-	s1 := NewSubscriber("", transport.logger, tss)
-	go s1.start()
+	s1 := NewSubscriber("", transport.logger)
 	require.Nil(t, transport.AddSubscriber(s1))
 
-	s2 := NewSubscriber("", transport.logger, tss)
-	go s2.start()
+	s2 := NewSubscriber("", transport.logger)
 	require.Nil(t, transport.AddSubscriber(s2))
 
 	assert.Len(t, transport.subscribers, 2)
@@ -272,13 +264,10 @@ func TestBoltGetSubscribers(t *testing.T) {
 	defer transport.Close()
 	defer os.Remove("test.db")
 
-	tss := &TopicSelectorStore{}
-	s1 := NewSubscriber("", transport.logger, tss)
-	go s1.start()
+	s1 := NewSubscriber("", transport.logger)
 	require.Nil(t, transport.AddSubscriber(s1))
 
-	s2 := NewSubscriber("", transport.logger, tss)
-	go s2.start()
+	s2 := NewSubscriber("", transport.logger)
 	require.Nil(t, transport.AddSubscriber(s2))
 
 	lastEventID, subscribers, err := transport.GetSubscribers()

caddy/caddy.go

@@ -90,6 +90,9 @@ type Mercure struct {
 	// Maximum cache cost, defaults to 100MB, set to -1 to disable the cache. See https://github.com/dgraph-io/ristretto for details.
 	CacheMaxCost *int64 `json:"cache_max_cost,omitempty"`
 
+	// Triggers use of LRU topic selector cache and avoidance of select priority queue (recommend 10,000 - 1,000,000)
+	LRUShardSize *int64 `json:"lru_shard_size,omitempty"`
+
 	hub    *mercure.Hub
 	logger *zap.Logger
 }
@@ -136,9 +139,18 @@ func (m *Mercure) Provision(ctx caddy.Context) error { //nolint:funlen
 		mc = *m.CacheMaxCost
 	}
 
-	tss, err := mercure.NewTopicSelectorStore(nc, mc)
-	if err != nil {
-		return err //nolint:wrapcheck
+	var err error
+	var tss *mercure.TopicSelectorStore
+	if m.LRUShardSize == nil {
+		tss, err = mercure.NewTopicSelectorStore(nc, mc)
+		if err != nil {
+			return err //nolint:wrapcheck
+		}
+	} else {
+		tss, err = mercure.NewTopicSelectorStoreLRU(*m.LRUShardSize, mercure.DefaultTopicSelectorStoreLRUShardCount)
+		if err != nil {
+			return err //nolint:wrapcheck
+		}
 	}
 
 	m.logger = ctx.Logger(m)
@@ -359,6 +371,18 @@ func (m *Mercure) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { //nolint:fu
 				}
 
 				m.CacheMaxCost = &v
+
+			case "lru_cache":
+				if !d.NextArg() {
+					return d.ArgErr()
+				}
+
+				v, err := strconv.ParseInt(d.Val(), 10, 64)
+				if err != nil {
+					return err //nolint:wrapcheck
+				}
+
+				m.LRUShardSize = &v
 			}
 		}
 	}

caddy/go.mod

@@ -7,6 +7,7 @@ replace github.com/dunglas/mercure => ../
 require (
 	github.com/caddyserver/caddy/v2 v2.4.3
 	github.com/caddyserver/certmagic v0.14.1 // indirect
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/dunglas/mercure v0.13.0
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect