diff --git a/message/cache.go b/message/cache.go index 1c73a45d..385399d4 100644 --- a/message/cache.go +++ b/message/cache.go @@ -405,6 +405,7 @@ func (c *Cache) Tail() (*Tail, error) { t := &Tail{ cache: c, holes: make(map[int64]time.Time), + done: make(chan struct{}), } now := time.Now() for _, rowID := range slices.Backward(rowIDs) { diff --git a/message/tail.go b/message/tail.go index a73cf413..bdc8cba1 100644 --- a/message/tail.go +++ b/message/tail.go @@ -3,6 +3,7 @@ package message import ( "database/sql" "encoding/json" + "sync" "sync/atomic" "time" @@ -25,6 +26,8 @@ type Tail struct { cursor int64 // Highest row id read so far holes map[int64]time.Time // Skipped row ids below the cursor, and when they were skipped lastRead atomic.Int64 // Unix nanoseconds of the last successful read (see LastRead); 0 before the first + done chan struct{} // Closed by Close, to stop the loop calling Next + closed sync.Once // Guards closing done } // LastRead returns when the tail last read the database successfully; zero before the first @@ -37,6 +40,18 @@ func (t *Tail) LastRead() time.Time { return time.Unix(0, nanos) } +// Done returns a channel that is closed when the tail is closed, so the loop calling Next can stop +func (t *Tail) Done() <-chan struct{} { + return t.done +} + +// Close tells the loop calling Next to stop (see Done); it may be called more than once +func (t *Tail) Close() { + t.closed.Do(func() { + close(t.done) + }) +} + // Next returns the published messages written since the last call, in write order, reading // pages until it is caught up or has read tailMaxPages. It reads the primary: a replica may lag, // and what it does not show yet would be skipped as a hole. diff --git a/message/tail_test.go b/message/tail_test.go index 39b2ab58..cffc62e7 100644 --- a/message/tail_test.go +++ b/message/tail_test.go @@ -159,6 +159,26 @@ func TestTail_LastRead(t *testing.T) { require.Equal(t, read, tail.LastRead()) } +func TestTail_Close(t *testing.T) { + // Done tells the loop calling Next to stop; Close may be called more than once + forEachBackend(t, func(t *testing.T, s *message.Cache) { + tail, err := s.Tail() + require.Nil(t, err) + select { + case <-tail.Done(): + t.Fatal("done before Close") + default: + } + tail.Close() + tail.Close() + select { + case <-tail.Done(): + case <-time.After(time.Second): + t.Fatal("not done after Close") + } + }) +} + func requireTail(t *testing.T, tail *message.Tail, expected ...string) { t.Helper() messages, err := tail.Next() diff --git a/server/server.go b/server/server.go index 94593ebe..efedf285 100644 --- a/server/server.go +++ b/server/server.go @@ -78,7 +78,6 @@ type Server struct { metricsHandler http.Handler // Handles /metrics if enable-metrics set, and listen-metrics-http not set closeChan chan bool tail *message.Tail // Reads what was written, see runTail; nil without a message cache - tailClose chan struct{} // Stops runTail; nil without a message cache stopOnce sync.Once stopped atomic.Bool // Set by Stop; read by Run, which must not start listeners afterwards mu sync.RWMutex @@ -360,7 +359,6 @@ func New(conf *Config) (*Server, error) { return nil, err } s.tail = tail - s.tailClose = make(chan struct{}) go s.runTail() } return s, nil @@ -550,8 +548,8 @@ func (s *Server) stop() { if s.httpClusterServer != nil { s.httpClusterServer.Close() } - if s.tailClose != nil { - close(s.tailClose) // Before the databases close under it + if s.tail != nil { + s.tail.Close() // Before the databases close under it } s.closeDatabases() if s.ban != nil { diff --git a/server/server_tail.go b/server/server_tail.go index c5a39826..0700efb3 100644 --- a/server/server_tail.go +++ b/server/server_tail.go @@ -25,7 +25,7 @@ func (s *Server) runTail() { for { select { case <-time.After(tailInterval): - case <-s.tailClose: + case <-s.tail.Done(): return } messages, err := s.tail.Next()