Harden cluster registry dissemination

Author: wolfy-j

api/cluster/raft/config.go

@@ -49,6 +49,12 @@ type Config struct {
 	BootstrapExpect int `json:"bootstrap_expect,omitempty"`
 	SnapshotRetain  int `json:"snapshot_retain"`
 	MaxPool         int `json:"max_pool"`
+	// ShutdownTransferTimeout bounds the best-effort leadership transfer
+	// attempted by Stop() when this node is leader. Zero defaults to one
+	// election timeout. The transfer is an availability optimization, not a
+	// safety requirement, so shutdown must continue if no eligible follower
+	// can take leadership promptly.
+	ShutdownTransferTimeout time.Duration `json:"shutdown_transfer_timeout"`
 	// MaxAppendEntries caps how many log entries the leader packs into a
 	// single AppendEntries RPC. The hashicorp/raft default is 64 which,
 	// when a follower restarts with an empty log and needs to catch up
@@ -98,6 +104,9 @@ func (c *Config) InitDefaults() {
 	if c.MaxPool == 0 {
 		c.MaxPool = 3
 	}
+	if c.ShutdownTransferTimeout == 0 {
+		c.ShutdownTransferTimeout = c.ElectionTimeout
+	}
 	if c.MaxAppendEntries == 0 {
 		// Default below hashicorp's 64 to cap leader memory during
 		// chaos catch-up. Followers that need 500+ entries get them

boot/components/system/cluster.example.yaml

@@ -72,3 +72,4 @@ cluster:
 #     max_append_entries: 16
 #     reconcile_debounce: 2s
 #     reconcile_timeout: 2s
+#     global_dissem_tombstone_retention: 0 # 0 keeps delete fences until restart; set e.g. 24h to bound cache tombstones.

boot/components/system/constants.go

@@ -71,4 +71,9 @@ const (
 	ClusterRaftCommitTimeout       boot.Name = "raft.commit_timeout"
 	ClusterRaftLeaderProbeInterval boot.Name = "raft.leader_probe_interval"
 	ClusterRaftLeaderProbeGrace    boot.Name = "raft.leader_probe_grace"
+	// ClusterRaftGlobalDissemTombstoneRetention bounds how long the AP
+	// global-name dissemination cache retains delete tombstones as stale-gossip
+	// fences. The Raft FSM remains authoritative; this only tunes cache memory
+	// vs stale delete repair.
+	ClusterRaftGlobalDissemTombstoneRetention boot.Name = "raft.global_dissem_tombstone_retention"
 )

boot/components/system/raft.go

@@ -67,6 +67,7 @@ func Raft() boot.Component {
 	var logger *zap.Logger
 	var handlerCfg sysraft.HandlerConfig
 	var bootstrapExpect int
+	var globalDissemTombstoneRetention time.Duration
 
 	return boot.New(boot.P{
 		Name:      RaftName,
@@ -137,6 +138,7 @@ func Raft() boot.Component {
 				ReconcileDebounce: raftCfg.GetDuration(ClusterRaftReconcileDebounce, 2*time.Second),
 				ReconcileTimeout:  raftCfg.GetDuration(ClusterRaftReconcileTimeout, 2*time.Second),
 			}
+			globalDissemTombstoneRetention = raftCfg.GetDuration(ClusterRaftGlobalDissemTombstoneRetention, 0)
 
 			// Raft rides the internode mesh exclusively. Without a connection
 			// manager (cluster transport not wired — e.g. a single-node
@@ -240,7 +242,8 @@ func Raft() boot.Component {
 			// the cache on FSM-miss so non-members resolve names locally.
 			if mem := clusterapi.GetMembership(ctx); mem != nil {
 				if memSvc, ok := mem.(*membership.Service); ok {
-					dissem := global.NewDissem(node.ID(), logger.Named("dissem"))
+					dissem := global.NewDissem(node.ID(), logger.Named("dissem"),
+						global.WithTombstoneRetention(globalDissemTombstoneRetention))
 					if err := memSvc.RegisterUserDelegate(dissem); err != nil {
 						return ctx, fmt.Errorf("raft: register globalreg dissem delegate: %w", err)
 					}

cluster/membership/membership.go

@@ -4,6 +4,7 @@ package membership
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/base64"
 	"encoding/json"
 	"errors"
@@ -128,6 +129,14 @@ type Config struct {
 	VeryVerbose  bool
 }
 
+const (
+	userBroadcastRetain       = 8192
+	userBroadcastBackpressure = userBroadcastRetain / 2
+	// Match memberlist's conservative UDP buffer. User broadcasts above this
+	// cannot fit once memberlist adds its own user-message/compound headers.
+	userBroadcastMaxPacketBytes = 1400
+)
+
 // NewService creates a new membership service.
 //
 // coll, mp, and tp wire the metrics collector and OTel providers used for
@@ -207,7 +216,7 @@ func (s *Service) Start(ctx context.Context) error {
 	mlConfig.BindAddr = s.config.BindAddr
 	mlConfig.BindPort = s.config.BindPort
 	mlConfig.Events = &eventDelegate{service: s}
-	mlConfig.Delegate = &delegate{service: s}
+	mlConfig.Delegate = newDelegate(s, mlConfig.RetransmitMult)
 
 	// Use custom transport if provided (for testing)
 	if s.transport != nil {
@@ -773,7 +782,86 @@ func (ed *eventDelegate) parseNodeMeta(meta []byte) cluster.NodeMeta {
 
 // delegate handles memberlist delegate functions
 type delegate struct {
-	service *Service
+	service    *Service
+	broadcasts memberlist.TransmitLimitedQueue
+	queuedMu   sync.Mutex
+	queued     map[string]struct{}
+}
+
+func newDelegate(service *Service, retransmitMult int) *delegate {
+	if retransmitMult <= 0 {
+		retransmitMult = 1
+	}
+	d := &delegate{
+		service: service,
+		queued:  make(map[string]struct{}),
+	}
+	d.broadcasts.NumNodes = service.broadcastNodeCount
+	d.broadcasts.RetransmitMult = retransmitMult
+	return d
+}
+
+func (s *Service) broadcastNodeCount() int {
+	if s.memberlist != nil {
+		if n := len(s.memberlist.Members()); n > 0 {
+			return n
+		}
+	}
+	s.mu.RLock()
+	n := len(s.nodes)
+	s.mu.RUnlock()
+	if n > 0 {
+		return n
+	}
+	return 1
+}
+
+type muxBroadcast struct {
+	name     string
+	msg      []byte
+	finished func(string)
+}
+
+func (b *muxBroadcast) Invalidates(other memberlist.Broadcast) bool {
+	o, ok := other.(*muxBroadcast)
+	return ok && b.name == o.name
+}
+
+func (b *muxBroadcast) Name() string { return b.name }
+func (b *muxBroadcast) Message() []byte {
+	return b.msg
+}
+func (b *muxBroadcast) Finished() {
+	if b.finished != nil {
+		b.finished(b.name)
+	}
+}
+
+var _ memberlist.NamedBroadcast = (*muxBroadcast)(nil)
+
+func (d *delegate) queueBroadcast(frame []byte) bool {
+	sum := sha256.Sum256(frame)
+	name := string(sum[:])
+	d.queuedMu.Lock()
+	if _, exists := d.queued[name]; exists {
+		d.queuedMu.Unlock()
+		return false
+	}
+	d.queued[name] = struct{}{}
+	d.queuedMu.Unlock()
+
+	d.broadcasts.QueueBroadcast(&muxBroadcast{
+		name:     name,
+		msg:      frame,
+		finished: d.forgetBroadcast,
+	})
+	return true
+}
+
+func (d *delegate) forgetBroadcast(name string) {
+	d.queuedMu.Lock()
+	delete(d.queued, name)
+	d.queuedMu.Unlock()
 }
 
 func (d *delegate) NodeMeta(limit int) []byte {
@@ -828,6 +916,38 @@ func (d *delegate) NotifyMsg(data []byte) {
 }
 
 func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
+	if d.service.tel != nil && d.service.tel.coll != nil {
+		d.service.tel.coll.CounterInc("gossip_user_getbroadcasts_calls_total", nil)
+	}
+
+	recordOut := func(out [][]byte) {
+		if d.service.tel == nil || d.service.tel.coll == nil || len(out) == 0 {
+			return
+		}
+		totalBytes := 0
+		totalCost := 0
+		for _, frame := range out {
+			totalBytes += len(frame)
+			totalCost += len(frame) + overhead
+		}
+		d.service.tel.coll.CounterAdd("gossip_user_getbroadcasts_frames_total", float64(len(out)), nil)
+		d.service.tel.coll.CounterAdd("gossip_user_getbroadcasts_bytes_total", float64(totalBytes), nil)
+		if totalCost > limit {
+			d.service.tel.coll.CounterInc("gossip_user_getbroadcasts_overshoot_total", nil)
+		}
+	}
+
+	queued := d.broadcasts.NumQueued()
+	if queued > userBroadcastRetain {
+		d.broadcasts.Prune(userBroadcastRetain)
+		queued = userBroadcastRetain
+	}
+	if queued >= userBroadcastBackpressure {
+		out := d.broadcasts.GetBroadcasts(overhead, limit)
+		recordOut(out)
+		return out
+	}
+
 	d.service.userDelegateMu.RLock()
 	dels := make([]UserDelegate, 0, len(d.service.userDelegates))
 	for _, ud := range d.service.userDelegates {
@@ -842,25 +962,31 @@ func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
 	// hogged the cycle).
 	sort.Slice(dels, func(i, j int) bool { return dels[i].Kind() < dels[j].Kind() })
 
-	if d.service.tel != nil && d.service.tel.coll != nil {
-		d.service.tel.coll.CounterInc("gossip_user_getbroadcasts_calls_total", nil)
-	}
-
-	var out [][]byte
-	totalBytes := 0
-	totalCost := 0
+	queuedCost := 0
 	muxOverhead := overhead + 5
 
 	wrap := func(ud UserDelegate, frames [][]byte) {
 		for _, f := range frames {
+			if uint64(len(f)) > uint64(^uint32(0)) {
+				if d.service.tel != nil && d.service.tel.coll != nil {
+					d.service.tel.coll.CounterInc("gossip_user_getbroadcasts_frame_dropped_total", nil)
+				}
+				continue
+			}
 			wrapped := make([]byte, 0, len(f)+5)
 			wrapped = append(wrapped, ud.Kind())
 			n := uint32(len(f))
 			wrapped = append(wrapped, byte(n), byte(n>>8), byte(n>>16), byte(n>>24))
 			wrapped = append(wrapped, f...)
-			totalCost += len(wrapped) + overhead
-			totalBytes += len(wrapped)
-			out = append(out, wrapped)
+			if len(wrapped)+overhead > userBroadcastMaxPacketBytes {
+				if d.service.tel != nil && d.service.tel.coll != nil {
+					d.service.tel.coll.CounterInc("gossip_user_getbroadcasts_frame_dropped_total", nil)
+				}
+				continue
+			}
+			if d.queueBroadcast(wrapped) {
+				queuedCost += len(wrapped) + overhead
+			}
 		}
 	}
 
@@ -883,21 +1009,19 @@ func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
 	// budget contract bounds emission to what fits the remainder. This
 	// recovers throughput when one delegate is silent.
 	for _, ud := range dels {
-		remaining := limit - totalCost
+		remaining := limit - queuedCost
 		if remaining <= muxOverhead {
 			break
 		}
 		frames := ud.GetBroadcasts(muxOverhead, remaining)
 		wrap(ud, frames)
 	}
 
-	if d.service.tel != nil && d.service.tel.coll != nil && len(out) > 0 {
-		d.service.tel.coll.CounterAdd("gossip_user_getbroadcasts_frames_total", float64(len(out)), nil)
-		d.service.tel.coll.CounterAdd("gossip_user_getbroadcasts_bytes_total", float64(totalBytes), nil)
-		if totalCost > limit {
-			d.service.tel.coll.CounterInc("gossip_user_getbroadcasts_overshoot_total", nil)
-		}
+	if d.broadcasts.NumQueued() > userBroadcastRetain {
+		d.broadcasts.Prune(userBroadcastRetain)
 	}
+	out := d.broadcasts.GetBroadcasts(overhead, limit)
+	recordOut(out)
 	return out
 }
 

cluster/membership/membership_test.go

@@ -539,7 +539,7 @@ func TestDelegate_NodeMeta_ValidJSON(t *testing.T) {
 	service, _, _, cancel := setupService(t)
 	defer cancel()
 
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	meta := d.NodeMeta(512)
 
@@ -565,7 +565,7 @@ func TestDelegate_NodeMeta_EmptyMeta(t *testing.T) {
 	}
 
 	service := NewService(config, bus, logger, nil, nil, nil)
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	meta := d.NodeMeta(512)
 
@@ -589,7 +589,7 @@ func TestDelegate_NodeMeta_ExceedsLimit(t *testing.T) {
 	}
 
 	service := NewService(config, bus, logger, nil, nil, nil)
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	meta := d.NodeMeta(10)
 
@@ -600,7 +600,7 @@ func TestDelegate_GetBroadcasts(t *testing.T) {
 	service, _, _, cancel := setupService(t)
 	defer cancel()
 
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	broadcasts := d.GetBroadcasts(10, 512)
 
@@ -611,7 +611,7 @@ func TestDelegate_LocalState(t *testing.T) {
 	service, _, _, cancel := setupService(t)
 	defer cancel()
 
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	state := d.LocalState(false)
 
@@ -623,7 +623,7 @@ func TestDelegate_MergeRemoteState(t *testing.T) {
 	service, _, _, cancel := setupService(t)
 	defer cancel()
 
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	d.MergeRemoteState([]byte(`{"key":"value"}`), true)
 }
@@ -632,7 +632,7 @@ func TestDelegate_NotifyMsg(t *testing.T) {
 	service, _, _, cancel := setupService(t)
 	defer cancel()
 
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	d.NotifyMsg([]byte("test message"))
 }
@@ -649,7 +649,7 @@ func TestDelegate_NotifyMsg_VeryVerbose(_ *testing.T) {
 	}
 
 	service := NewService(config, bus, logger, nil, nil, nil)
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	d.NotifyMsg([]byte("test message"))
 }
@@ -666,7 +666,7 @@ func TestDelegate_MergeRemoteState_VeryVerbose(_ *testing.T) {
 	}
 
 	service := NewService(config, bus, logger, nil, nil, nil)
-	d := &delegate{service: service}
+	d := newDelegate(service, memberlist.DefaultLocalConfig().RetransmitMult)
 
 	d.MergeRemoteState([]byte(`{"key":"value"}`), true)
 }