Let the tail carry its own stop channel: Done and Close

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) {

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.

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()

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 {

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()