perf: coalesce ring heartbeat ticks at one second

Makefile

@@ -252,6 +252,14 @@ build-storage-bench: proto
 	@mkdir -p "$(dir $(BENCH_OUTPUT))"
 	@cd storage && CGO_ENABLED=1 CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" go test $(LDFLAGS) -c -o "$(BENCH_OUTPUT)" .
 
+# Compile the ring heartbeat benchmark once so paired runs do not include Go
+# compilation. PERFLOOP_BUILD_OUTPUT_DIR is supplied by the benchmark runner.
+.PHONY: build-bench-ring-heartbeat
+build-bench-ring-heartbeat:
+	@test -n "$(PERFLOOP_BUILD_OUTPUT_DIR)" || { echo "PERFLOOP_BUILD_OUTPUT_DIR is required"; exit 1; }
+	@mkdir -p "$(PERFLOOP_BUILD_OUTPUT_DIR)"
+	@cd coordinator && go test $(LDFLAGS) -c -o "$(PERFLOOP_BUILD_OUTPUT_DIR)/ring-heartbeat.test" ./ring
+
 # Compile the topology RPC benchmark once so paired runs do not include Go
 # compilation. PERFLOOP_BUILD_OUTPUT_DIR is supplied by the benchmark runner.
 .PHONY: build-bench-cache-topology-rpc
@@ -615,6 +623,7 @@ help:
 	@echo "  bench                       - Run benchmarks"
 	@echo "  bench-integration           - Compile integration benchmarks (set INTEGRATION_BENCH_BINARY)"
 	@echo "  build-bench-compaction-serving-reads - Compile the compaction/serving benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)"
+	@echo "  build-bench-ring-heartbeat     - Compile the ring heartbeat benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)"
 	@echo "  build-bench-cache-topology-rpc - Compile the CacheService topology RPC benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)"
 	@echo "  build-bench-cache-service-delete - Compile the CacheService delete benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)"
 	@echo "  build-bench-cache-service-list-with-values - Compile the CacheService ListWithValues benchmark (set PERFLOOP_BUILD_OUTPUT_DIR)"

coordinator/ring/config.go

@@ -13,8 +13,12 @@ import (
 )
 
 const (
-	// DefaultHeartbeatPeriod is the default interval for heartbeat updates
-	DefaultHeartbeatPeriod = 500 * time.Millisecond
+	// DefaultHeartbeatPeriod is the default interval for heartbeat updates.
+	// Ring timestamps have one-second precision, so sub-second periods are coalesced.
+	DefaultHeartbeatPeriod = time.Second
+
+	// MinHeartbeatPeriod is the smallest effective interval for ring heartbeats.
+	MinHeartbeatPeriod = time.Second
 
 	// MinHeartbeatTimeout is the default timeout before marking a node unhealthy.
 	// Set long enough to handle rolling updates (60s allows for typical pod restarts).
@@ -53,9 +57,8 @@ type Config struct {
 // ApplyDefaults applies default values for any unset or invalid configuration fields.
 // This should be called before using the configuration.
 func (c *Config) ApplyDefaults() {
-	if c.HeartbeatPeriod <= 0 {
-		c.HeartbeatPeriod = DefaultHeartbeatPeriod
-	}
+	c.HeartbeatPeriod = effectiveHeartbeatPeriod(c.HeartbeatPeriod)
+
 	// HeartbeatTimeout must be at least the maximum of:
 	// - The configured value
 	// - MinHeartbeatTimeout (60s) to prevent flaky health checks
@@ -156,7 +159,19 @@ func (c *LifecyclerConfig) ApplyDefaults() {
 	}
 }
 
-// ToBasicLifecyclerConfig converts to dskit ring.BasicLifecyclerConfig
+func effectiveHeartbeatPeriod(period time.Duration) time.Duration {
+	if period <= 0 {
+		return DefaultHeartbeatPeriod
+	}
+	if period < MinHeartbeatPeriod {
+		return MinHeartbeatPeriod
+	}
+	return period
+}
+
+// ToBasicLifecyclerConfig converts to dskit ring.BasicLifecyclerConfig.
+// The heartbeat floor is applied here as well as in ApplyDefaults so direct
+// callers cannot pass a sub-second period to dskit's ticker.
 func (c *LifecyclerConfig) ToBasicLifecyclerConfig() ring.BasicLifecyclerConfig {
 	addr := c.InstanceAddr
 	if c.InstancePort > 0 {
@@ -166,7 +181,7 @@ func (c *LifecyclerConfig) ToBasicLifecyclerConfig() ring.BasicLifecyclerConfig
 	return ring.BasicLifecyclerConfig{
 		ID:                              c.InstanceID,
 		Addr:                            addr,
-		HeartbeatPeriod:                 c.RingConfig.HeartbeatPeriod,
+		HeartbeatPeriod:                 effectiveHeartbeatPeriod(c.RingConfig.HeartbeatPeriod),
 		HeartbeatTimeout:                c.RingConfig.HeartbeatTimeout,
 		TokensObservePeriod:             c.ObservePeriod,
 		NumTokens:                       c.NumTokens,

coordinator/ring/config_test.go

@@ -25,6 +25,18 @@ func TestConfig_ApplyDefaults(t *testing.T) {
 				ReplicationFactor: 1,
 			},
 		},
+		{
+			name: "sub-second heartbeat is floored",
+			input: Config{
+				HeartbeatPeriod:  500 * time.Millisecond,
+				HeartbeatTimeout: 5 * time.Second,
+			},
+			expected: Config{
+				HeartbeatPeriod:   MinHeartbeatPeriod,
+				HeartbeatTimeout:  MinHeartbeatTimeout,
+				ReplicationFactor: 1,
+			},
+		},
 		{
 			name: "short timeout is upgraded to minimum",
 			input: Config{
@@ -198,3 +210,17 @@ func TestLifecyclerConfig_ToBasicLifecyclerConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestLifecyclerConfig_ToBasicLifecyclerConfig_FloorsSubSecondHeartbeat(t *testing.T) {
+	cfg := LifecyclerConfig{
+		RingConfig: Config{
+			HeartbeatPeriod: 500 * time.Millisecond,
+		},
+	}
+
+	basicCfg := cfg.ToBasicLifecyclerConfig()
+
+	assert.Equal(t, MinHeartbeatPeriod, basicCfg.HeartbeatPeriod)
+	assert.Equal(t, 500*time.Millisecond, cfg.RingConfig.HeartbeatPeriod,
+		"conversion should not mutate an unnormalised config")
+}

coordinator/ring/heartbeat_benchmark_test.go

@@ -0,0 +1,226 @@
+// Copyright 2026 Tigris Data, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package ring
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/go-kit/log"
+	dskitring "github.com/grafana/dskit/ring"
+	"github.com/grafana/dskit/services"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/tigrisdata/ocache/coordinator/gossip"
+)
+
+const (
+	heartbeatBenchmarkMembers       = 8
+	heartbeatBenchmarkTokensPerNode = 1
+	heartbeatBenchmarkStep          = 100 * time.Millisecond
+)
+
+type heartbeatBenchmarkDelegate struct {
+	base       dskitring.InstanceRegisterDelegate
+	callbacks  atomic.Uint64
+	production ringDelegate
+}
+
+func (d *heartbeatBenchmarkDelegate) OnRingInstanceRegister(l *dskitring.BasicLifecycler, desc dskitring.Desc, exists bool, id string, instance dskitring.InstanceDesc) (dskitring.InstanceState, dskitring.Tokens) {
+	return d.base.OnRingInstanceRegister(l, desc, exists, id, instance)
+}
+
+func (d *heartbeatBenchmarkDelegate) OnRingInstanceTokens(*dskitring.BasicLifecycler, dskitring.Tokens) {
+}
+
+func (d *heartbeatBenchmarkDelegate) OnRingInstanceStopping(*dskitring.BasicLifecycler) {}
+
+func (d *heartbeatBenchmarkDelegate) OnRingInstanceHeartbeat(l *dskitring.BasicLifecycler, desc *dskitring.Desc, instance *dskitring.InstanceDesc) {
+	d.callbacks.Add(1)
+	d.production.OnRingInstanceHeartbeat(l, desc, instance)
+}
+
+type heartbeatBenchmarkFixture struct {
+	delegate     *heartbeatBenchmarkDelegate
+	reg          *prometheus.Registry
+	lifecycler   *dskitring.BasicLifecycler
+	memberlistKV *gossip.Memberlist
+	cancel       context.CancelFunc
+	stopOnce     sync.Once
+}
+
+func newHeartbeatBenchmarkFixture(b *testing.B) *heartbeatBenchmarkFixture {
+	b.Helper()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	logger := log.NewNopLogger()
+	reg := prometheus.NewRegistry()
+	nodeID := "heartbeat-benchmark-member-0"
+	memberlistKV, err := gossip.NewMemberlist(nodeID, "127.0.0.1:0", nil, logger, reg)
+	if err != nil {
+		cancel()
+		b.Fatalf("create memberlist: %v", err)
+	}
+	if err := memberlistKV.Start(ctx); err != nil {
+		cancel()
+		b.Fatalf("start memberlist: %v", err)
+	}
+
+	desc := dskitring.NewDesc()
+	registeredAt := time.Now()
+	for member := 0; member < heartbeatBenchmarkMembers; member++ {
+		desc.AddIngester(
+			fmt.Sprintf("heartbeat-benchmark-member-%d", member),
+			fmt.Sprintf("127.0.0.1:%d", 10000+member),
+			"",
+			[]uint32{uint32(member + 1)},
+			dskitring.ACTIVE,
+			registeredAt,
+		)
+	}
+	if err := memberlistKV.Client().CAS(ctx, RingKey, func(interface{}) (interface{}, bool, error) {
+		return desc, true, nil
+	}); err != nil {
+		_ = memberlistKV.Stop(context.Background())
+		cancel()
+		b.Fatalf("seed ring: %v", err)
+	}
+
+	cfg := LifecyclerConfig{
+		InstanceID:   nodeID,
+		InstanceAddr: "127.0.0.1:10000",
+		NumTokens:    heartbeatBenchmarkTokensPerNode,
+	}
+	cfg.ApplyDefaults()
+
+	delegate := &heartbeatBenchmarkDelegate{
+		base:       dskitring.NewInstanceRegisterDelegate(dskitring.ACTIVE, heartbeatBenchmarkTokensPerNode),
+		production: ringDelegate{rm: &RingManager{logger: logger}},
+	}
+	basicCfg := cfg.ToBasicLifecyclerConfig()
+	// Keep teardown from unregistering the seeded instance with a non-heartbeat CAS.
+	basicCfg.KeepInstanceInTheRingOnShutdown = true
+	lifecycler, err := dskitring.NewBasicLifecycler(
+		basicCfg,
+		RingName,
+		RingKey,
+		memberlistKV.Client(),
+		delegate,
+		logger,
+		reg,
+	)
+	if err != nil {
+		_ = memberlistKV.Stop(context.Background())
+		cancel()
+		b.Fatalf("create lifecycler: %v", err)
+	}
+	if err := services.StartAndAwaitRunning(ctx, lifecycler); err != nil {
+		_ = memberlistKV.Stop(context.Background())
+		cancel()
+		b.Fatalf("start lifecycler: %v", err)
+	}
+
+	fixture := &heartbeatBenchmarkFixture{
+		delegate:     delegate,
+		reg:          reg,
+		lifecycler:   lifecycler,
+		memberlistKV: memberlistKV,
+		cancel:       cancel,
+	}
+	b.Cleanup(fixture.stop)
+	return fixture
+}
+
+func (f *heartbeatBenchmarkFixture) stop() {
+	f.stopOnce.Do(func() {
+		_ = services.StopAndAwaitTerminated(context.Background(), f.lifecycler)
+		_ = f.memberlistKV.Stop(context.Background())
+		f.cancel()
+	})
+}
+
+type heartbeatBenchmarkCounts struct {
+	casAttempts  float64
+	casSuccesses float64
+	heartbeats   float64
+	callbacks    uint64
+}
+
+func heartbeatBenchmarkSnapshot(b *testing.B, fixture *heartbeatBenchmarkFixture) heartbeatBenchmarkCounts {
+	b.Helper()
+
+	families, err := fixture.reg.Gather()
+	if err != nil {
+		b.Fatalf("gather heartbeat metrics: %v", err)
+	}
+
+	var counts heartbeatBenchmarkCounts
+	found := make(map[string]bool, 3)
+	for _, family := range families {
+		var target *float64
+		switch family.GetName() {
+		case "memberlist_client_cas_attempt_total":
+			target = &counts.casAttempts
+		case "memberlist_client_cas_success_total":
+			target = &counts.casSuccesses
+		case "ring_member_heartbeats_total":
+			target = &counts.heartbeats
+		default:
+			continue
+		}
+		found[family.GetName()] = true
+		for _, metric := range family.GetMetric() {
+			*target += metric.GetCounter().GetValue()
+		}
+	}
+
+	for _, name := range []string{
+		"memberlist_client_cas_attempt_total",
+		"memberlist_client_cas_success_total",
+		"ring_member_heartbeats_total",
+	} {
+		if !found[name] {
+			b.Fatalf("metric %s not found", name)
+		}
+	}
+	counts.callbacks = fixture.delegate.callbacks.Load()
+	return counts
+}
+
+func BenchmarkRingHeartbeat(b *testing.B) {
+	fixture := newHeartbeatBenchmarkFixture(b)
+	before := heartbeatBenchmarkSnapshot(b, fixture)
+
+	b.ResetTimer()
+	for b.Loop() {
+		time.Sleep(heartbeatBenchmarkStep)
+	}
+	b.StopTimer()
+
+	// Stop-time lifecycle work is outside the heartbeat workload. Capture the
+	// endpoint before stopping so failed teardown CAS attempts cannot be folded
+	// into the heartbeat-attempt calculation.
+	after := heartbeatBenchmarkSnapshot(b, fixture)
+	fixture.stop()
+
+	heartbeats := after.heartbeats - before.heartbeats
+	callbacks := float64(after.callbacks - before.callbacks)
+	casAttempts := after.casAttempts - before.casAttempts
+	casSuccesses := after.casSuccesses - before.casSuccesses
+	nonHeartbeatCAS := casSuccesses - heartbeats
+	heartbeatAttempts := casAttempts - nonHeartbeatCAS
+	if heartbeats <= 0 {
+		b.Fatalf("heartbeat benchmark produced no successful heartbeats (attempts=%.0f callbacks=%.0f)", casAttempts, callbacks)
+	}
+	if heartbeatAttempts < heartbeats || callbacks < heartbeats {
+		b.Fatalf("heartbeat counters are inconsistent (attempts=%.0f successes=%.0f callbacks=%.0f heartbeats=%.0f)", heartbeatAttempts, casSuccesses, callbacks, heartbeats)
+	}
+
+	b.ReportMetric(0, "ns/op")
+	b.ReportMetric(heartbeatAttempts/heartbeats, "cas-attempts/heartbeat")
+	b.ReportMetric(callbacks/heartbeats, "heartbeat-callbacks/heartbeat")
+}

docs/cluster.md

@@ -191,7 +191,7 @@ Planned improvements:
 
 Each node monitors others via the gossip protocol:
 
-- Regular heartbeat messages (default: every 5 seconds)
+- Regular ring heartbeat messages (default: every 1 second; sub-second configured values are coalesced)
 - Configurable heartbeat timeout (default: 60 seconds)
 - Automatic topology updates propagated via memberlist
 - Epoch tracking for efficient client topology cache invalidation

docs/rfcs/RFC-007-sharded-cluster.md

@@ -231,7 +231,7 @@ dskit BasicLifecycler manages instance state transitions:
 
 #### 4.3.3 Ring Heartbeats
 
-- **Heartbeat Period**: 500ms (DefaultHeartbeatPeriod)
+- **Heartbeat Period**: 1s minimum (DefaultHeartbeatPeriod); sub-second values are coalesced because ring timestamps use Unix-second precision
 - **Heartbeat Timeout**: 60s minimum (MinHeartbeatTimeout)
 - Ring heartbeats update instance state in the KV store
 - Other nodes receive updates via memberlist gossip (not direct peer-to-peer)
@@ -263,7 +263,7 @@ Ring state is stored in a distributed KV backed by memberlist:
 | ------------------- | -------------------------------------- | --------------------------------- |
 | **Gossip Messages** | Propagate state changes between nodes  | 200ms interval, 3 nodes per cycle |
 | **Push/Pull Sync**  | Full state reconciliation              | Every 30 seconds                  |
-| **Ring Heartbeats** | Update instance liveness in KV         | Every 500ms                       |
+| **Ring Heartbeats** | Update instance liveness in KV         | Every 1s minimum; sub-second values are coalesced |
 | **KV Watcher**      | Immediate notification of ring changes | Event-driven                      |
 
 #### 4.4.2 Gossip Configuration