rdma: synchronize publication acceptance with the drain boundary

Service the overflow list during normal operation: the worker
publishes and clears it after every channel job, so a burst that
exceeds the channel buffer drains as soon as the sink recovers
instead of accumulating until shutdown.

Make dispatch and the drain sweep share one critical section. An
append either lands before the sweep and is drained, or runs
after the worker exited and publishes inline; the check-then-send
window that could strand a record between the two is closed.
Request publications use the same boundary with drop semantics:
once the drain finished, no owner can guarantee the sinks are
still open, so the record is dropped rather than published.

Never close the metrics datapoint channel. The forwarder exits
through the canceled context after draining the buffer, the
closed flag turns late producers into drops, and no send can race
a closure. Implement the new Manager method in the test mock.

embedgw/embedgw.go

@@ -1243,6 +1243,13 @@ Loop:
 	}
 
 	if metricsManager != nil {
+		// Cancel the metrics context first: the forwarder exits
+		// through it, draining the buffered datapoints, and
+		// Close then only waits for the forwarder and closes the
+		// publishers. The channel itself never closes, so late
+		// producers (a handler outliving the HTTP shutdown
+		// timeout) drop their datapoint instead of panicking.
+		metricsStop()
 		metricsManager.Close()
 	}
 

metrics/metrics.go

@@ -245,15 +245,14 @@ func (m *manager) add(key string, value int64, tags ...Tag) {
 	}
 }
 
-// Close closes metrics channels, waits for data to complete, closes all plugins
+// Close stops the manager: producers drop new datapoints, the
+// forwarder drains the buffered ones and exits through the
+// canceled context, and the publishers flush and close. The
+// datapoint channel itself is never closed - a producer racing
+// the closure would panic - so the closed flag and the context
+// cancellation carry the shutdown instead.
 func (m *manager) Close() {
-	// Stop accepting new datapoints before closing the channel:
-	// producers check the flag and drop their update, so only a
-	// producer already between the check and the send can race,
-	// and that producer recovers instead of panicking.
 	m.closed.Store(true)
-	// drain the datapoint channels
-	close(m.addDataChan)
 	m.wg.Wait()
 
 	// close all publishers
@@ -269,12 +268,36 @@ type publisher interface {
 }
 
 func (m *manager) addForwarder(addChan <-chan datapoint) {
-	for data := range addChan {
-		for _, s := range m.publishers {
-			s.Add(data.key, data.value, data.tags...)
+	defer m.wg.Done()
+	for {
+		select {
+		case data, ok := <-addChan:
+			if !ok {
+				return
+			}
+			for _, s := range m.publishers {
+				s.Add(data.key, data.value, data.tags...)
+			}
+		case <-m.ctx.Done():
+			// The channel is never closed (producers race its
+			// closure otherwise); termination is the context.
+			// Drain whatever the buffer still holds so late
+			// datapoints are not lost, then exit.
+			for {
+				select {
+				case data, ok := <-addChan:
+					if !ok {
+						return
+					}
+					for _, s := range m.publishers {
+						s.Add(data.key, data.value, data.tags...)
+					}
+				default:
+					return
+				}
+			}
 		}
 	}
-	m.wg.Done()
 }
 
 type datapoint struct {

rdma/rcroutes/ops_linux.go

@@ -323,29 +323,35 @@ func newOpsTracker() *opsTracker {
 					return
 				}
 				job.emit.publish(job.err, job.byt)
+				// Service the overflow list after every
+				// channel job: bursts that exceed the
+				// buffer publish as soon as the sink
+				// recovers instead of waiting for
+				// shutdown.
+				t.serviceOverflow()
 			case <-t.drain:
-				// Drain mode: empty the channel, then the
-				// overflow list, then exit. Producers past
-				// this point append to overflow; the final
-				// sweep below runs only what arrived before
-				// the drain signal, and a producer racing
-				// the sweep re-enqueues through dispatch's
-				// post-drain path.
+				// Drain mode, single critical section with
+				// dispatch: holding pubmu across the
+				// channel-and-list sweep closes the
+				// accept-vs-drain race - a dispatch that
+				// appended before this point is drained,
+				// one that runs after sees done and
+				// publishes inline.
+				t.pubmu.Lock()
 				for {
 					select {
 					case job, ok := <-t.pubq:
 						if !ok {
+							t.pubmu.Unlock()
 							return
 						}
 						job.emit.publish(job.err, job.byt)
 					default:
-						t.pubmu.Lock()
-						pending := t.overflow
-						t.overflow = nil
-						t.pubmu.Unlock()
-						for _, job := range pending {
+						for _, job := range t.overflow {
 							job.emit.publish(job.err, job.byt)
 						}
+						t.overflow = nil
+						t.pubmu.Unlock()
 						return
 					}
 				}
@@ -355,6 +361,17 @@ func newOpsTracker() *opsTracker {
 	return t
 }
 
+// serviceOverflow publishes and clears the overflow list if the
+// worker can take it. Called by the worker only.
+func (t *opsTracker) serviceOverflow() {
+	t.pubmu.Lock()
+	defer t.pubmu.Unlock()
+	for _, job := range t.overflow {
+		job.emit.publish(job.err, job.byt)
+	}
+	t.overflow = nil
+}
+
 // Shutdown drains pending publications and stops the worker. The
 // gateway must call this BEFORE closing the operational sinks: a
 // queued publication that runs after its logger closed is lost.
@@ -383,16 +400,22 @@ func (t *opsTracker) Shutdown() {
 // publications are dropped by publishRequest before reaching
 // here.
 func (t *opsTracker) dispatch(job pubJob) {
+	// Fast path: the worker is alive. The pubmu critical section
+	// is the accept-vs-drain boundary: the drain sweep holds the
+	// same lock, so an append either lands before the sweep (and
+	// is drained) or after done closed (and runs inline).
+	t.pubmu.Lock()
 	select {
 	case <-t.done:
+		t.pubmu.Unlock()
 		job.emit.publish(job.err, job.byt)
 		return
 	default:
 	}
 	select {
 	case t.pubq <- job:
+		t.pubmu.Unlock()
 	default:
-		t.pubmu.Lock()
 		t.overflow = append(t.overflow, job)
 		t.pubmu.Unlock()
 	}
@@ -630,13 +653,6 @@ func (t *opsTracker) publishRequest(ctx fiber.Ctx, acct auth.Account,
 	if t == nil {
 		return
 	}
-	select {
-	case <-t.done:
-		// The worker already exited through the drain; sinks
-		// are closing. Drop the record.
-		return
-	default:
-	}
 	acct.Access = strings.Clone(acct.Access)
 	emit := &opsEmitter{
 		ops:    t.loadOps(),
@@ -648,7 +664,32 @@ func (t *opsTracker) publishRequest(ctx fiber.Ctx, acct auth.Account,
 		isPut:  isPut,
 		start:  time.Now(),
 	}
-	t.dispatch(pubJob{emit: emit, err: err})
+	// The accept-vs-drain boundary decides: a record accepted
+	// before the drain sweep is published by the worker; one
+	// that arrives after is dropped here (not published inline),
+	// because a request publication has no owner left to
+	// guarantee its sinks are still open.
+	t.dispatchOrDrop(pubJob{emit: emit, err: err})
+}
+
+// dispatchOrDrop is dispatch with request-publication semantics:
+// after the worker exited through the drain the job is dropped
+// instead of published inline.
+func (t *opsTracker) dispatchOrDrop(job pubJob) {
+	t.pubmu.Lock()
+	select {
+	case <-t.done:
+		t.pubmu.Unlock()
+		return
+	default:
+	}
+	select {
+	case t.pubq <- job:
+		t.pubmu.Unlock()
+	default:
+		t.overflow = append(t.overflow, job)
+		t.pubmu.Unlock()
+	}
 }
 
 // regionFromCtx reads the region the gateway middleware stored on

s3api/controllers/base_test.go

@@ -282,7 +282,9 @@ func (m *mockEvSender) Close() error                               { return nil
 type mockMetricsManager struct{}
 
 func (m *mockMetricsManager) Send(_ fiber.Ctx, _ error, _ string, _ int64, _ int) {}
-func (m *mockMetricsManager) Close()                                              {}
+func (m *mockMetricsManager) SendWithBucket(_ fiber.Ctx, _ error, _ string, _ int64, _ int, _ string) {
+}
+func (m *mockMetricsManager) Close() {}
 
 func TestProcessController(t *testing.T) {
 	payload, err := xml.Marshal(s3response.Bucket{