Bound each shutdown stage by the stop deadline (closes #100)

Shutdown passes its context to every stage: the streamer, the four
handlers, the WHOIS fetcher, the database maintainer and the HTTP server
each stop waiting when it ends. A handler then skips its last write and
logs how many updates it did not write, and the HTTP server closes the
connections still open. The database is checkpointed and closed last,
and left open once the deadline has passed, since a write may still
hold it. The stop hook gives Shutdown 8 seconds and fx 9, inside the 10
seconds upaas and docker stop allow, so a stop signal exits 0.

Model: opus-5-5

internal/streamer/streamer.go

@@ -253,9 +253,10 @@ func (s *Streamer) Start(ctx context.Context) error {
 
 // Stop ends the stream and stops the handler workers. It returns once each
 // worker has handed its handler every message left on its queue and then the
-// STATE messages held for it (see queueMessage). This method is safe to call
+// STATE messages held for it (see queueMessage), or once ctx ends, and then
+// logs how many messages were not handed over. This method is safe to call
 // multiple times.
-func (s *Streamer) Stop() {
+func (s *Streamer) Stop(ctx context.Context) {
 	s.mu.Lock()
 	if s.cancel == nil {
 		// Not started, or already stopped: closing the queues again would panic.
@@ -264,28 +265,56 @@ func (s *Streamer) Stop() {
 		return
 	}
 
-	// Once ctx is cancelled the read loop queues and holds nothing more (see
-	// dispatchMessage).
+	// Once the stream's context is cancelled the read loop queues and holds
+	// nothing more (see dispatchMessage).
 	s.cancel()
 	s.cancel = nil
 
+	notHandedOver := 0
+
 	for _, info := range s.handlers {
 		// The held messages came after everything on the queue. Each waits
-		// for the worker, which does not take s.mu, to make room.
-		for _, msg := range info.heldState {
-			info.queue <- msg
+		// for the worker, which does not take s.mu, to make room, until ctx
+		// ends.
+		for len(info.heldState) > 0 && ctx.Err() == nil {
+			select {
+			case info.queue <- info.heldState[0]:
+				info.heldState = info.heldState[1:]
+			case <-ctx.Done():
+			}
 		}
 
+		notHandedOver += len(info.heldState)
 		info.heldState = nil
 
 		// The worker returns once it has emptied the closed queue.
 		close(info.queue)
 	}
 
+	handlers := s.handlers
 	s.running = false
 	s.mu.Unlock()
 	s.metrics.SetConnected(false)
-	s.workers.Wait()
+
+	workersDone := make(chan struct{})
+
+	go func() {
+		s.workers.Wait()
+		close(workersDone)
+	}()
+
+	select {
+	case <-workersDone:
+	case <-ctx.Done():
+		for _, info := range handlers {
+			notHandedOver += len(info.queue)
+		}
+	}
+
+	if notHandedOver > 0 {
+		s.logger.Error("Stop deadline passed before the handlers took every "+
+			"message", "messages", notHandedOver)
+	}
 }
 
 // IsRunning reports whether the streamer is currently connected and processing

internal/streamer/streamer_internal_test.go

@@ -377,7 +377,7 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 	stopped := make(chan struct{})
 
 	go func() {
-		s.Stop()
+		s.Stop(t.Context())
 		close(stopped)
 	}()
 
@@ -412,6 +412,48 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 	}
 }
 
+// TestStopReturnsByItsDeadline stops a streamer whose handler has a full queue
+// and a STATE message held for it, and a worker that stays busy until the test
+// ends, as one does whose handler waits for a database write. Stop must return
+// within its deadline plus a small margin instead of waiting for room on the
+// queue and for the worker.
+func TestStopReturnsByItsDeadline(t *testing.T) {
+	t.Parallel()
+
+	const (
+		deadline = 100 * time.Millisecond
+		margin   = 2 * time.Second
+	)
+
+	s := newStreamerWithFullStateQueue("")
+	s.handlers[0].heldState = []*ristypes.RISMessage{
+		{Type: ristypes.MessageTypeState},
+	}
+	// Stop cancels the stream Start would have started; there is none here.
+	s.cancel = func() {}
+
+	release := make(chan struct{})
+	defer close(release)
+
+	s.workers.Go(func() { <-release })
+
+	ctx, cancel := context.WithTimeout(t.Context(), deadline)
+	defer cancel()
+
+	stopped := make(chan struct{})
+
+	go func() {
+		s.Stop(ctx)
+		close(stopped)
+	}()
+
+	select {
+	case <-stopped:
+	case <-time.After(deadline + margin):
+		t.Fatalf("Stop did not return within %s of its deadline", margin)
+	}
+}
+
 // TestStopBeforeMessageReachesQueues stops the streamer after the read loop
 // has checked for cancellation but before it hands the message to the handler
 // queues. That is the gap a stop from another goroutine can land in, and it
@@ -429,7 +471,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 	s := New(logger.New(), metrics.New())
 	s.url = srv.URL
 	s.RegisterHandler(updateHandler{})
-	s.RegisterRawHandler(func(string) { s.Stop() })
+	s.RegisterRawHandler(func(string) { s.Stop(t.Context()) })
 
 	// Start would run the stream in the background, where the test cannot
 	// wait for it. Setting cancel as Start does lets Stop cancel the stream
@@ -443,7 +485,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 	}
 
 	// A second Stop must not close the queues again.
-	s.Stop()
+	s.Stop(t.Context())
 }
 
 // TestReconnectWhenStreamEnds has the server end the first response
@@ -676,7 +718,7 @@ func checkReconnect(
 	if err != nil {
 		t.Fatalf("Start returned error: %v", err)
 	}
-	defer s.Stop()
+	defer s.Stop(t.Context())
 
 	select {
 	case <-handler.handled: