gossip: fix pending callbacks not dropping to zero

We have seen cases before where there was an issue with the pending callbacks
metric that it's stuck at some non-zero number. This commit fixes that by:

1) If the infoStore stopper() is quiescing, we should not add more items to the work
queue, because they could not be processed, and the pending callbacks metrics won't
get decremented.

2) Only increment the pending callbacks metric when we actually add items to the
work queue, and not before. Otherwise, there could be a race condition where we
increment the pending callbacks metric, and then the infostore Stopper gets quiesced
before we added any items to the work queue. This means that we won't decrement
the pending callbacks metric.

3) When the callback gets cancelled, or when the infostore stopper gets quiesced, we
decrement the pending callbacks metric by the number of items in the work queue.
Otherwise, we might not decrement the number of pending callbacks.

Fixes: #147982

Release note: None

Author: iskettaneh

pkg/gossip/gossip_test.go

@@ -1140,3 +1140,70 @@ func TestGossipBatching(t *testing.T) {
 	require.LessOrEqual(t, serverMessagesSentCount, upperBoundMessages)
 	require.LessOrEqual(t, clientMessagesSentCount, upperBoundMessages)
 }
+
+// TestCallbacksPendingMetricGoesToZeroOnStop verifies that the CallbacksPending
+// metric is correctly decremented when a callback is unregistered with pending work
+// or when the stopper is stopped.
+func TestCallbacksPendingMetricGoesToZeroOnStop(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	testCases := []struct {
+		name    string
+		cleanup func(g *Gossip, unregister func(), stopper *stop.Stopper, ctx context.Context)
+	}{
+		{
+			name: "unregister callback",
+			cleanup: func(g *Gossip, unregister func(), stopper *stop.Stopper, ctx context.Context) {
+				unregister()
+			},
+		},
+		{
+			name: "stopper shutdown",
+			cleanup: func(g *Gossip, unregister func(), stopper *stop.Stopper, ctx context.Context) {
+				stopper.Stop(ctx)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			stopper := stop.NewStopper()
+			defer stopper.Stop(ctx)
+			g := NewTest(1, stopper, metric.NewRegistry())
+
+			unregister := g.RegisterCallback("test.*", func(key string, val roachpb.Value) {
+				// Do nothing.
+			})
+
+			// Add 100 infos to the gossip that will be processed by the callback.
+			for i := 0; i < 100; i++ {
+				slice := []byte("b1")
+				require.NoError(t, g.AddInfo(fmt.Sprintf("test.key%d", i), slice, time.Hour))
+			}
+
+			// Execute the cleanup action (either unregister or stopper.Stop)
+			// We do this in a goroutine to help cause interesting potential race conditions.
+			go func() {
+				tc.cleanup(g, unregister, stopper, ctx)
+			}()
+
+			// Add another 100 infos to the gossip that will be processed by the callback.
+			// We do this in a goroutine to help cause interesting potential race conditions.
+			go func() {
+				for i := 0; i < 100; i++ {
+					slice := []byte("b2")
+					require.NoError(t, g.AddInfo(fmt.Sprintf("test.key%d", i), slice, time.Hour))
+				}
+			}()
+
+			// Wait for the pending callbacks metric to go to 0.
+			testutils.SucceedsSoon(t, func() error {
+				if g.mu.is.metrics.CallbacksPending.Value() != 0 {
+					return fmt.Errorf("CallbacksPending should be 0, got %d", g.mu.is.metrics.CallbacksPending.Value())
+				}
+				return nil
+			})
+		})
+	}
+}

pkg/gossip/infostore.go

@@ -232,11 +232,27 @@ func newInfoStore(
 	return is
 }
 
+// cleanupCallbackMetric decrements the callback metric by the number of remaining
+// items in the queue since they will never be processed. This is called when the
+// callback worker is stopped to avoid having the wrong metric value.
+// This should only be called when no new work can be added to the queue.
+func (is *infoStore) cleanupCallbackMetric(cw *callbackWork) {
+	cw.mu.Lock()
+	remainingItems := len(cw.mu.workQueue)
+	if remainingItems > 0 {
+		is.metrics.CallbacksPending.Dec(int64(remainingItems))
+	}
+	cw.mu.Unlock()
+}
+
 // launchCallbackWorker launches a worker goroutine that is responsible for
 // executing callbacks for one registered callback pattern.
 func (is *infoStore) launchCallbackWorker(ambient log.AmbientContext, cw *callbackWork) {
 	bgCtx := ambient.AnnotateCtx(context.Background())
 	_ = is.stopper.RunAsyncTask(bgCtx, "callback worker", func(ctx context.Context) {
+		// If we exit the loop, we are never going to process the work in the queues anymore, so
+		// clean up the pending callbacks metric.
+		defer is.cleanupCallbackMetric(cw)
 		for {
 			for {
 				cw.mu.Lock()
@@ -444,11 +460,19 @@ func (is *infoStore) processCallbacks(key string, content roachpb.Value, changed
 // It adds work to the callback work slices, and signals the associated callback
 // workers to execute the work.
 func (is *infoStore) runCallbacks(key string, content roachpb.Value, callbacks ...*callback) {
+	// Check if the stopper is quiescing. If so, do not add the callbacks to the
+	// callback work list because they won't be processed anyways.
+	select {
+	case <-is.stopper.ShouldQuiesce():
+		return
+	default:
+	}
+
 	// Add the callbacks to the callback work list.
 	beforeQueue := timeutil.Now()
-	is.metrics.CallbacksPending.Inc(int64(len(callbacks)))
 	for _, cb := range callbacks {
 		cb.cw.mu.Lock()
+		is.metrics.CallbacksPending.Inc(1)
 		cb.cw.mu.workQueue = append(cb.cw.mu.workQueue, callbackWorkItem{
 			schedulingTime: beforeQueue,
 			method:         cb.method,