From abcfd88cd8b7419cb03a3ef4a600c1f13cd8828d Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Sat, 31 Oct 2026 18:29:29 +0000 Subject: [PATCH] Rename topic.Topic to topic.State, so the live topic can take the name later The package comment now says the server's live topics are meant to move into the topic package. --- server/rate_visitor.go | 6 +-- server/rate_visitor_test.go | 20 +++++----- server/server_test.go | 4 +- topic/service.go | 31 ++++++++------- topic/service_test.go | 78 ++++++++++++++++++------------------- topic/types.go | 4 +- 6 files changed, 74 insertions(+), 69 deletions(-) diff --git a/server/rate_visitor.go b/server/rate_visitor.go index 22cbe6f2..071e42fb 100644 --- a/server/rate_visitor.go +++ b/server/rate_visitor.go @@ -53,7 +53,7 @@ func (s *Server) rateVisitor(t *topic) *visitor { if !s.config.VisitorSubscriberRateLimiting || !isUnifiedPushTopic(t.ID) { return nil } - state := s.topicService.Get(t.ID) + state := s.topicService.State(t.ID) if state == nil || state.RateVisitor == nil { return nil } else if t.SubscribersCount() == 0 && time.Since(state.SubscribedAt) > rateVisitorLiveFor { @@ -79,7 +79,7 @@ func (s *Server) unifiedPushTopicAbandoned(r *http.Request, t *topic) bool { if !isUnifiedPushTopic(t.ID) || t.SubscribersCount() > 0 { return false } - state := s.topicService.Get(t.ID) + state := s.topicService.State(t.ID) if state == nil || !topicAbandoned(state) { return false } @@ -168,7 +168,7 @@ func (c *rateVisitorUserCache) Refresh(ids []string) error { // topicAbandoned returns true if the topic's newest observation (a subscriber, or its first // refused push) is older than unifiedPushAbandonedAfter -func topicAbandoned(state *tp.Topic) bool { +func topicAbandoned(state *tp.State) bool { newest := state.FirstSeenAt if state.SubscribedAt.After(newest) { newest = state.SubscribedAt diff --git a/server/rate_visitor_test.go b/server/rate_visitor_test.go index dff8c337..599b71e0 100644 --- a/server/rate_visitor_test.go +++ b/server/rate_visitor_test.go @@ -31,7 +31,7 @@ func TestServer_UnifiedPush_RateVisitorOnOtherNode(t *testing.T) { }) require.Equal(t, 200, rr.Code) waitForWithMaxWait(t, 3*syncWait, func() bool { - return s2.topicService.Get("upAAAAAAAAAAAA") != nil + return s2.topicService.State("upAAAAAAAAAAAA") != nil }) response := request(t, s2, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil) require.Equal(t, 200, response.Code) @@ -64,7 +64,7 @@ func TestServer_UnifiedPush_RateVisitorOnOtherNodeUsesTierAfterRefresh(t *testin require.Equal(t, 200, rr.Code) require.Equal(t, "user:"+phil.ID, rateVisitorKey(s1, "upAAAAAAAAAAAA")) // The subscribe primes the cache waitForWithMaxWait(t, 3*syncWait, func() bool { - return s2.topicService.Get("upAAAAAAAAAAAA") != nil + return s2.topicService.State("upAAAAAAAAAAAA") != nil }) require.Equal(t, "ip:1.2.3.4", rateVisitorKey(s2, "upAAAAAAAAAAAA")) s2.execManager() // Refreshes the rate visitors' users @@ -93,7 +93,7 @@ func TestServer_UnifiedPush_StreamOnOtherNodeKeepsRateVisitor(t *testing.T) { return to != nil && to.SubscribersCount() == 1 }) waitForWithMaxWait(t, 3*syncWait, func() bool { - return s2.topicService.Get("upAAAAAAAAAAAA") != nil + return s2.topicService.State("upAAAAAAAAAAAA") != nil }) // A day later, the stream is still open on s1 @@ -102,7 +102,7 @@ func TestServer_UnifiedPush_StreamOnOtherNodeKeepsRateVisitor(t *testing.T) { require.Equal(t, 507, request(t, s2, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil).Code) s1.execManager() waitForWithMaxWait(t, 3*syncWait, func() bool { - return time.Since(s2.topicService.Get("upAAAAAAAAAAAA").SubscribedAt) < time.Hour + return time.Since(s2.topicService.State("upAAAAAAAAAAAA").SubscribedAt) < time.Hour }) response := request(t, s2, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil) require.Equal(t, 200, response.Code) @@ -172,7 +172,7 @@ func TestServer_UnifiedPush_AbandonedTopicReadsRowBeforeAnswering404(t *testing. execUserDB(t, c, fmt.Sprintf(`INSERT INTO topic (topic, first_seen_at, subscribed_at, rate_visitor_user_id, rate_visitor_ip, updated_at) VALUES ('upAAAAAAAAAAAA', %d, %d, '', '1.2.3.4', 1)`, now-2*int64(oneDay/time.Second), now-2*int64(oneDay/time.Second))) response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil) require.Equal(t, 507, response.Code) - require.WithinDuration(t, time.Now().Add(-2*oneDay), s.topicService.Get("upAAAAAAAAAAAA").SubscribedAt, time.Minute) + require.WithinDuration(t, time.Now().Add(-2*oneDay), s.topicService.State("upAAAAAAAAAAAA").SubscribedAt, time.Minute) // Known as abandoned, and so is the row: 404 ageTopic(s, "upAAAAAAAAAAAA", 6*oneDay) @@ -230,11 +230,11 @@ func TestServer_UnifiedPush_RefusedPushDoesNotReviveRateVisitor(t *testing.T) { }) require.Equal(t, 200, rr.Code) ageTopic(s, "upAAAAAAAAAAAA", 25*time.Hour) - subscribed := s.topicService.Get("upAAAAAAAAAAAA").SubscribedAt + subscribed := s.topicService.State("upAAAAAAAAAAAA").SubscribedAt for i := 0; i < 3; i++ { require.Equal(t, 507, request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil).Code) } - require.Equal(t, subscribed, s.topicService.Get("upAAAAAAAAAAAA").SubscribedAt) + require.Equal(t, subscribed, s.topicService.State("upAAAAAAAAAAAA").SubscribedAt) s.Stop() // Writes the topic var subscribedAt int64 queryUserDB(t, c, `SELECT subscribed_at FROM topic WHERE topic = 'upAAAAAAAAAAAA'`, &subscribedAt) @@ -249,7 +249,7 @@ func TestServer_UnifiedPush_NonUnifiedPushTopicIsNotAbandoned(t *testing.T) { s := newTestServer(t, c) response := request(t, s, "PUT", "/mytopic?up=1", "some message", nil) require.Equal(t, 507, response.Code) - require.Nil(t, s.topicService.Get("mytopic")) + require.Nil(t, s.topicService.State("mytopic")) } func TestServer_MatrixGateway_AbandonedTopicRejectsPushKey(t *testing.T) { @@ -485,9 +485,9 @@ func rateVisitorKey(s *Server, topic string) string { // ageTopic makes the topic look as if its subscriber activity and first observation were d // earlier than they are; an unknown topic becomes one first seen d ago func ageTopic(s *Server, topic string, d time.Duration) { - t := s.topicService.Get(topic) + t := s.topicService.State(topic) if t == nil { - t = &tp.Topic{FirstSeenAt: time.Now()} + t = &tp.State{FirstSeenAt: time.Now()} } t.FirstSeenAt = t.FirstSeenAt.Add(-d) if !t.SubscribedAt.IsZero() { diff --git a/server/server_test.go b/server/server_test.go index a7770e16..6baf11d6 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -1978,7 +1978,7 @@ func TestServer_MatrixGateway_Push_Failure_NoSubscriber_After13Hours(t *testing. response := request(t, s, "POST", "/_matrix/push/v1/notify", notification, nil) require.Equal(t, 507, response.Code) require.Equal(t, 50701, toHTTPError(t, response.Body.String()).Code) - require.Nil(t, s.topicService.Get("mytopic")) + require.Nil(t, s.topicService.State("mytopic")) // Fake: This topic has been around for 13 hours without a rate visitor s.topics["mytopic"].lastAccess = time.Now().Add(-13 * time.Hour) @@ -3502,7 +3502,7 @@ func TestServer_SubscriberRateLimiting_PollKeepsPollingSubscriberAlive(t *testin ageTopic(s, "upAAAAAAAAAAAA", 23*time.Hour) rr = request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, subscriberFn) require.Equal(t, 200, rr.Code) - require.WithinDuration(t, time.Now(), s.topicService.Get("upAAAAAAAAAAAA").SubscribedAt, time.Minute) + require.WithinDuration(t, time.Now(), s.topicService.State("upAAAAAAAAAAAA").SubscribedAt, time.Minute) } func TestServer_SubscriberRateLimiting_OverlappingStreamsKeepTopicAlive(t *testing.T) { diff --git a/topic/service.go b/topic/service.go index 5fd89604..9865c0ae 100644 --- a/topic/service.go +++ b/topic/service.go @@ -7,13 +7,18 @@ // With a store, a sync worker writes the topics changed here (a dirty set, so a round costs // what changed, not the table) every syncInterval and reads back the rows changed since the last // round, applying the same merge rules in memory; at startup it loads every row. A publish -// never reads the database (Get is memory only); Refresh is the one cold-path read, for a +// never reads the database (State is memory only); Refresh is the one cold-path read, for a // decision too final to make from a stale row. Rows not written for pruneAfter are pruned from // the table and from memory alike. // // The worker is the service's own, not the usage counters' (package limit): a stuck topic sync // must not stall the daily limits. Timestamps are the node's clock (unix seconds), so the nodes // must keep their clocks in sync (NTP); a skew beyond readOverlap may skip rows in the read-back. +// +// The live topics of a node (its subscribers and the delivery to them, server.topic and the +// server's topics map today) are meant to move into this package too, as topic.Topic and +// topic.Subscriber, with Subscribe and Unsubscribe calling SetRateVisitor, Touch and Keepalive +// themselves. Until then, put new per-topic state here rather than in the server. package topic import ( @@ -185,19 +190,19 @@ func (s *Service) Note(topic string) { s.noted++ } -// Get returns the topic's state as this node knows it, from memory, or nil if it is unknown -func (s *Service) Get(topic string) *Topic { +// State returns the topic's state as this node knows it, from memory, or nil if it is unknown +func (s *Service) State(topic string) *State { s.mu.Lock() defer s.mu.Unlock() - return s.getNoLock(topic) + return s.stateNoLock(topic) } // Refresh reads the topic's row from the store (a primary key read, with the caller's context), -// merges it into memory, and returns the topic's state like Get. It is for a decision too final -// to make from a stale row, on a cold path only. Without a store it is Get. -func (s *Service) Refresh(ctx context.Context, topic string) (*Topic, error) { +// merges it into memory, and returns the topic's state like State. It is for a decision too final +// to make from a stale row, on a cold path only. Without a store it is State. +func (s *Service) Refresh(ctx context.Context, topic string) (*State, error) { if s.store == nil { - return s.Get(topic), nil + return s.State(topic), nil } r, err := s.store.readOne(ctx, topic) if err != nil { @@ -208,7 +213,7 @@ func (s *Service) Refresh(ctx context.Context, topic string) (*Topic, error) { if r != nil { s.applyNoLock(r) } - return s.getNoLock(topic), nil + return s.stateNoLock(topic), nil } // RateVisitorUserIDs returns the distinct user IDs of the rate visitors of the topics with @@ -240,7 +245,7 @@ func (s *Service) Size() int { // SetForTest replaces the topic's state with the given one, as if it had been observed here, and // has it written. It exists so tests can age a topic; it is not used in production code. -func (s *Service) SetForTest(topic string, t *Topic) { +func (s *Service) SetForTest(topic string, t *State) { now := s.now().Unix() s.mu.Lock() defer s.mu.Unlock() @@ -488,13 +493,13 @@ func (s *Service) entryNoLock(topic string, now int64) *entry { return e } -// getNoLock returns the topic's state, or nil if it is unknown. The caller must hold s.mu. -func (s *Service) getNoLock(topic string) *Topic { +// stateNoLock returns the topic's state, or nil if it is unknown. The caller must hold s.mu. +func (s *Service) stateNoLock(topic string) *State { e, ok := s.entries[topic] if !ok { return nil } - t := &Topic{ + t := &State{ FirstSeenAt: time.Unix(e.firstSeenAt, 0).UTC(), } if e.subscribedAt != 0 { diff --git a/topic/service_test.go b/topic/service_test.go index c7e824db..600e64aa 100644 --- a/topic/service_test.go +++ b/topic/service_test.go @@ -52,7 +52,7 @@ func startTestService(t *testing.T, store *Store, now func() time.Time) *Service func TestService_SetTouchNote(t *testing.T) { s, clock := newTestServiceWithClock(t) - require.Nil(t, s.Get("upA")) + require.Nil(t, s.State("upA")) // A refused push makes the topic known: first seen now, never subscribed, no rate visitor; a // second one changes nothing @@ -60,30 +60,30 @@ func TestService_SetTouchNote(t *testing.T) { noted := clock.now() clock.advance(time.Hour) s.Note("upA") - require.Equal(t, &Topic{FirstSeenAt: noted}, s.Get("upA")) + require.Equal(t, &State{FirstSeenAt: noted}, s.State("upA")) // A subscriber becomes the rate visitor, and counts as subscriber activity; the latest one wins s.SetRateVisitor("upA", &RateVisitor{IP: "1.2.3.4"}) s.SetRateVisitor("upA", &RateVisitor{UserID: "u_1", IP: "1.2.3.4"}) set := clock.now() - require.Equal(t, &Topic{FirstSeenAt: noted, SubscribedAt: set, RateVisitor: userRateVisitor()}, s.Get("upA")) + require.Equal(t, &State{FirstSeenAt: noted, SubscribedAt: set, RateVisitor: userRateVisitor()}, s.State("upA")) // A touch keeps the rate visitor and moves the subscriber activity clock.advance(time.Minute) s.Touch("upA") touched := clock.now() - require.Equal(t, &Topic{FirstSeenAt: noted, SubscribedAt: touched, RateVisitor: userRateVisitor()}, s.Get("upA")) + require.Equal(t, &State{FirstSeenAt: noted, SubscribedAt: touched, RateVisitor: userRateVisitor()}, s.State("upA")) // A refused push to a known topic changes nothing, however old its subscriber activity clock.advance(48 * time.Hour) s.Note("upA") - require.Equal(t, &Topic{FirstSeenAt: noted, SubscribedAt: touched, RateVisitor: userRateVisitor()}, s.Get("upA")) + require.Equal(t, &State{FirstSeenAt: noted, SubscribedAt: touched, RateVisitor: userRateVisitor()}, s.State("upA")) // A touch or a subscribe of an unknown topic makes it known, first seen and subscribed now s.Touch("upB") - require.Equal(t, &Topic{FirstSeenAt: clock.now(), SubscribedAt: clock.now()}, s.Get("upB")) + require.Equal(t, &State{FirstSeenAt: clock.now(), SubscribedAt: clock.now()}, s.State("upB")) s.SetRateVisitor("upC", &RateVisitor{IP: "1.2.3.4"}) - require.Equal(t, &Topic{FirstSeenAt: clock.now(), SubscribedAt: clock.now(), RateVisitor: ipRateVisitor()}, s.Get("upC")) + require.Equal(t, &State{FirstSeenAt: clock.now(), SubscribedAt: clock.now(), RateVisitor: ipRateVisitor()}, s.State("upC")) require.Equal(t, 3, s.Size()) } @@ -93,15 +93,15 @@ func TestService_Keepalive_Throttled(t *testing.T) { set := clock.now() clock.advance(keepaliveInterval - time.Second) s.Keepalive("upA") - require.Equal(t, set, s.Get("upA").SubscribedAt) + require.Equal(t, set, s.State("upA").SubscribedAt) clock.advance(time.Second) s.Keepalive("upA") - require.Equal(t, clock.now(), s.Get("upA").SubscribedAt) - require.Equal(t, ipRateVisitor(), s.Get("upA").RateVisitor) + require.Equal(t, clock.now(), s.State("upA").SubscribedAt) + require.Equal(t, ipRateVisitor(), s.State("upA").RateVisitor) // A keepalive of an unknown topic is a touch s.Keepalive("upB") - require.Equal(t, &Topic{FirstSeenAt: clock.now(), SubscribedAt: clock.now()}, s.Get("upB")) + require.Equal(t, &State{FirstSeenAt: clock.now(), SubscribedAt: clock.now()}, s.State("upB")) } func TestService_Note_Bounded(t *testing.T) { @@ -112,12 +112,12 @@ func TestService_Note_Bounded(t *testing.T) { } require.Equal(t, maxNoted, s.Size()) s.Note("upOneTooMany") - require.Nil(t, s.Get("upOneTooMany")) + require.Nil(t, s.State("upOneTooMany")) s.Touch("up000000000000") s.Note("upOneTooMany") - require.NotNil(t, s.Get("upOneTooMany")) + require.NotNil(t, s.State("upOneTooMany")) s.Note("upTwoTooMany") - require.Nil(t, s.Get("upTwoTooMany")) + require.Nil(t, s.State("upTwoTooMany")) } func TestService_Prune(t *testing.T) { @@ -128,19 +128,19 @@ func TestService_Prune(t *testing.T) { s.Touch("upB") clock.advance(pruneAfter - 30*time.Minute) s.prune(clock.now().Unix() - int64(pruneAfter/time.Second)) - require.Nil(t, s.Get("upA")) - require.NotNil(t, s.Get("upB")) + require.Nil(t, s.State("upA")) + require.NotNil(t, s.State("upB")) // A pruned topic may be noted again for i := 0; i < maxNoted; i++ { s.Note(fmt.Sprintf("up%012d", i)) } s.Note("upOneTooMany") - require.Nil(t, s.Get("upOneTooMany")) + require.Nil(t, s.State("upOneTooMany")) clock.advance(pruneAfter + time.Hour) s.prune(clock.now().Unix() - int64(pruneAfter/time.Second)) s.Note("upOneTooMany") - require.NotNil(t, s.Get("upOneTooMany")) + require.NotNil(t, s.State("upOneTooMany")) } func TestService_RateVisitorUserIDs(t *testing.T) { @@ -159,15 +159,15 @@ func TestService_RateVisitorUserIDs(t *testing.T) { func TestService_SetForTest(t *testing.T) { s, clock := newTestServiceWithClock(t) s.SetRateVisitor("upA", ipRateVisitor()) - aged := &Topic{ + aged := &State{ FirstSeenAt: clock.now().Add(-48 * time.Hour), SubscribedAt: clock.now().Add(-25 * time.Hour), RateVisitor: ipRateVisitor(), } s.SetForTest("upA", aged) - require.Equal(t, aged, s.Get("upA")) - s.SetForTest("upB", &Topic{FirstSeenAt: clock.now().Add(-8 * 24 * time.Hour)}) - require.Equal(t, &Topic{FirstSeenAt: clock.now().Add(-8 * 24 * time.Hour)}, s.Get("upB")) + require.Equal(t, aged, s.State("upA")) + s.SetForTest("upB", &State{FirstSeenAt: clock.now().Add(-8 * 24 * time.Hour)}) + require.Equal(t, &State{FirstSeenAt: clock.now().Add(-8 * 24 * time.Hour)}, s.State("upB")) } func TestService_Persist_SharedAcrossNodes(t *testing.T) { @@ -181,13 +181,13 @@ func TestService_Persist_SharedAcrossNodes(t *testing.T) { set := clock.now() a.sync() b.sync() - require.Equal(t, &Topic{FirstSeenAt: set, SubscribedAt: set, RateVisitor: userRateVisitor()}, b.Get("upA")) + require.Equal(t, &State{FirstSeenAt: set, SubscribedAt: set, RateVisitor: userRateVisitor()}, b.State("upA")) clock.advance(time.Hour) b.Touch("upA") b.sync() a.sync() - require.Equal(t, &Topic{FirstSeenAt: set, SubscribedAt: clock.now(), RateVisitor: userRateVisitor()}, a.Get("upA")) - require.Equal(t, a.Get("upA"), b.Get("upA")) + require.Equal(t, &State{FirstSeenAt: set, SubscribedAt: clock.now(), RateVisitor: userRateVisitor()}, a.State("upA")) + require.Equal(t, a.State("upA"), b.State("upA")) }) } @@ -200,8 +200,8 @@ func TestService_Persist_SurvivesRestart(t *testing.T) { s.Note("upB") s.Close() s = startTestService(t, newStore(), clock.now) - require.Equal(t, &Topic{FirstSeenAt: clock.now(), SubscribedAt: clock.now(), RateVisitor: ipRateVisitor()}, s.Get("upA")) - require.Equal(t, &Topic{FirstSeenAt: clock.now()}, s.Get("upB")) + require.Equal(t, &State{FirstSeenAt: clock.now(), SubscribedAt: clock.now(), RateVisitor: ipRateVisitor()}, s.State("upA")) + require.Equal(t, &State{FirstSeenAt: clock.now()}, s.State("upB")) require.Equal(t, 2, s.Size()) }) } @@ -222,7 +222,7 @@ func TestService_Persist_NoteDoesNotReviveRateVisitor(t *testing.T) { a.sync() b.sync() for _, s := range []*Service{a, b} { - require.Equal(t, &Topic{FirstSeenAt: set, SubscribedAt: set, RateVisitor: userRateVisitor()}, s.Get("upA")) + require.Equal(t, &State{FirstSeenAt: set, SubscribedAt: set, RateVisitor: userRateVisitor()}, s.State("upA")) } }) } @@ -259,10 +259,10 @@ func TestService_Persist_ReadBackKeepsUnwrittenRateVisitor(t *testing.T) { b.SetRateVisitor("upA", &RateVisitor{IP: "5.6.7.8"}) _, err := b.read(context.Background(), 0) require.Nil(t, err) - require.Equal(t, &RateVisitor{IP: "5.6.7.8"}, b.Get("upA").RateVisitor) + require.Equal(t, &RateVisitor{IP: "5.6.7.8"}, b.State("upA").RateVisitor) b.sync() a.sync() - require.Equal(t, &RateVisitor{IP: "5.6.7.8"}, a.Get("upA").RateVisitor) + require.Equal(t, &RateVisitor{IP: "5.6.7.8"}, a.State("upA").RateVisitor) }) } @@ -282,7 +282,7 @@ func TestService_Persist_MergesFirstSeenAndSubscribed(t *testing.T) { a.sync() b.sync() for _, s := range []*Service{a, b} { - require.Equal(t, &Topic{FirstSeenAt: noted, SubscribedAt: touched}, s.Get("upA")) + require.Equal(t, &State{FirstSeenAt: noted, SubscribedAt: touched}, s.State("upA")) } }) } @@ -297,11 +297,11 @@ func TestService_Persist_Refresh(t *testing.T) { a.SetRateVisitor("upA", ipRateVisitor()) set := clock.now() a.sync() - require.Nil(t, b.Get("upA")) + require.Nil(t, b.State("upA")) topic, err := b.Refresh(context.Background(), "upA") require.Nil(t, err) - require.Equal(t, &Topic{FirstSeenAt: set, SubscribedAt: set, RateVisitor: ipRateVisitor()}, topic) - require.Equal(t, topic, b.Get("upA")) + require.Equal(t, &State{FirstSeenAt: set, SubscribedAt: set, RateVisitor: ipRateVisitor()}, topic) + require.Equal(t, topic, b.State("upA")) // A fresher row replaces what b knows clock.advance(time.Hour) @@ -310,7 +310,7 @@ func TestService_Persist_Refresh(t *testing.T) { topic, err = b.Refresh(context.Background(), "upA") require.Nil(t, err) require.Equal(t, clock.now(), topic.SubscribedAt) - require.Equal(t, clock.now(), b.Get("upA").SubscribedAt) + require.Equal(t, clock.now(), b.State("upA").SubscribedAt) // An unknown topic stays unknown; a failed read is an error topic, err = b.Refresh(context.Background(), "upB") @@ -351,8 +351,8 @@ func TestService_Persist_PrunesTableAndMemory(t *testing.T) { clock.advance(pruneAfter + 2*time.Hour) s.Note("upB") s.sync() // Prunes: an hour has passed since the last one - require.Nil(t, s.Get("upA")) - require.NotNil(t, s.Get("upB")) + require.Nil(t, s.State("upA")) + require.NotNil(t, s.State("upB")) require.Equal(t, []string{"upB"}, topicsOf(readAll(t, store, 0))) // A topic with an unwritten change is never pruned from memory, however old its last write @@ -366,7 +366,7 @@ func TestService_Persist_PrunesTableAndMemory(t *testing.T) { s.mu.Unlock() clock.advance(pruneInterval) s.sync() - require.NotNil(t, s.Get("upC")) + require.NotNil(t, s.State("upC")) }) } @@ -376,7 +376,7 @@ func TestService_WithoutStore(t *testing.T) { s.Note("upA") topic, err := s.Refresh(context.Background(), "upA") require.Nil(t, err) - require.Equal(t, s.Get("upA"), topic) + require.Equal(t, s.State("upA"), topic) s.Close() } diff --git a/topic/types.go b/topic/types.go index 57b3efd8..57f948f2 100644 --- a/topic/types.go +++ b/topic/types.go @@ -17,8 +17,8 @@ type RateVisitor struct { IP string // The masked IP, never empty } -// Topic is a topic's cluster-wide state as this node knows it (see Service.Get) -type Topic struct { +// State is a topic's cluster-wide state as this node knows it (see Service.State) +type State struct { FirstSeenAt time.Time // Earliest observation on any node: a refused push, or the first subscribe SubscribedAt time.Time // Last subscriber activity on any node; zero if there was none yet (the topic was only pushed to) RateVisitor *RateVisitor // Who pays for the pushes; nil if nobody who may pay subscribed yet