internal/home/cluster_latency_cadence_test.go
Rama main · réplica de lectura del repositorio original.
package home
import (
"context"
"encoding/json"
"fmt"
"strings"
"testing"
"time"
)
func TestClusterLatencyProbeCadenceKeepsEvidenceFresh(t *testing.T) {
for _, test := range []struct {
name string
jitter []time.Duration
}{
{name: "regular ticks", jitter: []time.Duration{0}},
{name: "progressively early ticks", jitter: []time.Duration{0, -time.Nanosecond, -2 * time.Nanosecond, -3 * time.Nanosecond}},
{name: "alternating jitter", jitter: []time.Duration{200 * time.Nanosecond, -100 * time.Nanosecond, 50 * time.Nanosecond, -200 * time.Nanosecond}},
} {
t.Run(test.name, func(t *testing.T) {
testClusterLatencyProbeCadence(t, test.jitter)
})
}
}
func testClusterLatencyProbeCadence(t *testing.T, jitter []time.Duration) {
t.Helper()
history := &clusterLatencyHistory{}
node := clusterNode{IP: "candidate", Port: 8327}
address := clusterNodeAddress(node)
start := time.Unix(4000, 0)
var uploadedExpiry time.Time
var previousObservedAt time.Time
for cycle := range 8 {
reportAt := start.Add(time.Duration(cycle)*time.Minute + jitter[cycle%len(jitter)])
// Home receives the previous probe on the next report, not when the probe completes.
uploadAt := reportAt.Add(100 * time.Millisecond)
samples := history.report(reportAt)
if cycle == 0 {
if len(samples) != 0 {
t.Fatal("first report contained an unmeasured candidate")
}
} else {
if len(samples) != 1 || !samples[0].Valid || samples[0].Sequence != int64(cycle) {
t.Fatalf("cycle %d: missing fresh candidate evidence: %+v", cycle, samples)
}
sample := samples[0]
if !sample.ObservedAt.Equal(previousObservedAt) || sample.AgeMS != reportAt.Sub(previousObservedAt).Milliseconds() {
t.Fatalf("cycle %d: report did not preserve probe age: %+v", cycle, sample)
}
if !uploadedExpiry.IsZero() && !uploadAt.Before(uploadedExpiry) {
t.Fatalf("cycle %d: Home evidence expired at %v before upload at %v", cycle, uploadedExpiry, uploadAt)
}
// Home uses the older of observed_at and receipt time minus age_ms.
sampledAt := uploadAt.Add(-time.Duration(sample.AgeMS) * time.Millisecond)
if sample.ObservedAt.Before(sampledAt) {
sampledAt = sample.ObservedAt
}
uploadedExpiry = sampledAt.Add(clusterLatencyTTL)
}
if !history.probeDue(node, reportAt) {
t.Fatalf("cycle %d: ticker jitter or probe duration skipped a report cycle", cycle)
}
// Vary report and probe latency without sleeping or relying on platform timers.
reportDelay := time.Duration(1+cycle%3) * 100 * time.Millisecond
probeDuration := time.Duration(1+cycle%2) * time.Second
completedAt := reportAt.Add(reportDelay + probeDuration)
history.observeProbe(node, 20*time.Millisecond, true, reportAt, completedAt)
entry := history.entries[address]
if entry.nextProbeCycle != int64(cycle+1) || !entry.nextProbeAt.IsZero() || entry.probeBackoff != 0 {
t.Fatalf("cycle %d: incorrect successful probe schedule: %+v", cycle, entry)
}
if history.probeDue(node, completedAt) {
t.Fatalf("cycle %d: another probe is allowed within the same report cycle", cycle)
}
previousObservedAt = completedAt
}
}
func TestClusterLatencyProbeCyclesSurviveSkippedReports(t *testing.T) {
history := &clusterLatencyHistory{}
node := clusterNode{IP: "candidate", Port: 8327}
start := time.Unix(4500, 0)
history.observeProbe(node, time.Millisecond, true, start, start.Add(time.Second))
// Dropped ticks advance to their absolute cycle instead of shifting the baseline.
reportAt := start.Add(5*time.Minute - time.Nanosecond)
if !history.probeDue(node, reportAt) {
t.Fatal("skipped reports prevented a new probe")
}
history.observeProbe(node, time.Millisecond, true, reportAt, reportAt.Add(time.Second))
entry := history.entries[clusterNodeAddress(node)]
if entry.nextProbeCycle != 6 || !history.probeBaseline.Equal(start) {
t.Fatalf("skipped reports shifted the cycle baseline: %+v", entry)
}
if history.probeDue(node, start.Add(5*time.Minute+time.Nanosecond)) {
t.Fatal("the same cycle allowed another probe after a slightly early tick")
}
if !history.probeDue(node, start.Add(6*time.Minute-2*time.Nanosecond)) {
t.Fatal("an early next tick skipped a probe after dropped reports")
}
}
func TestClusterLatencyProbeFailureScheduleAndRecovery(t *testing.T) {
history := &clusterLatencyHistory{}
node := clusterNode{IP: "candidate", Port: 8327}
address := clusterNodeAddress(node)
reportAt := time.Unix(5000, 0)
for _, backoff := range []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute, 10 * time.Minute, 10 * time.Minute} {
completedAt := reportAt.Add(clusterLatencyTimeout)
history.observeProbe(node, 0, false, reportAt, completedAt)
entry := history.entries[address]
if entry.sample.Valid || entry.probeBackoff != backoff || !entry.nextProbeAt.Equal(completedAt.Add(backoff)) {
t.Fatalf("failure backoff changed: %+v, want %v after completion", entry, backoff)
}
if history.probeDue(node, entry.nextProbeAt.Add(-time.Nanosecond)) || !history.probeDue(node, entry.nextProbeAt) {
t.Fatal("report cycle rounding bypassed or extended failure backoff")
}
reportAt = entry.nextProbeAt
}
history.observeProbe(node, time.Millisecond, true, reportAt, reportAt.Add(time.Second))
entry := history.entries[address]
if !entry.sample.Valid || entry.probeBackoff != 0 || !entry.nextProbeAt.IsZero() || history.probeDue(node, reportAt) || !history.probeDue(node, reportAt.Add(time.Minute-time.Nanosecond)) {
t.Fatalf("successful probe did not restore report cadence: %+v", entry)
}
}
func TestClusterLatencyReportsRespectScheduledProbeBudget(t *testing.T) {
target, targetCommands := newRedisCommandTestClient(t, func(args []string) string {
if strings.EqualFold(args[0], "PING") && len(args) == 2 {
return fmt.Sprintf("$%d\r\n%s\r\n", len(args[1]), args[1])
}
return balanceTestRedisReply(args)
})
node := clusterNode{IP: target.homeCfg.Host, Port: target.homeCfg.Port}
client, commands := newRedisCommandTestClient(t, func(args []string) string {
if strings.EqualFold(args[0], "CLUSTER") {
raw := fmt.Sprintf(`{"ok":true,"probe":{"ip":%q,"port":%d}}`, node.IP, node.Port)
return fmt.Sprintf("$%d\r\n%s\r\n", len(raw), raw)
}
return balanceTestRedisReply(args)
})
enableBalanceTestClient(client)
client.balance.latencySupported = true
client.clusterNodes = []clusterNode{node}
history := &clusterLatencyHistory{}
// A slightly early next tick is a new cycle; repeated ticks in that cycle must not probe twice.
start := time.Now().Add(-3 * time.Minute)
for _, reportAt := range []time.Time{start, start.Add(time.Minute - time.Nanosecond), start.Add(time.Minute), start.Add(2*time.Minute - 2*time.Nanosecond)} {
if errReport := client.reportClusterLatency(context.Background(), history, reportAt); errReport != nil {
t.Fatal(errReport)
}
}
pings := 0
for _, args := range targetCommands.All() {
if strings.EqualFold(args[0], "PING") {
pings++
}
}
if pings != 6 {
t.Fatalf("PING count = %d, want exactly three probes (warmup plus measured PING)", pings)
}
var sequences []int64
for _, args := range commands.All() {
if len(args) != 4 || args[2] != "LATENCY" {
continue
}
var report struct {
Samples []clusterLatencySample `json:"samples"`
}
if errDecode := json.Unmarshal([]byte(args[3]), &report); errDecode != nil {
t.Fatal(errDecode)
}
if len(report.Samples) == 0 {
sequences = append(sequences, 0)
} else {
sequences = append(sequences, report.Samples[0].Sequence)
}
}
if len(sequences) != 4 || sequences[0] != 0 || sequences[1] != 1 || sequences[2] != 2 || sequences[3] != 2 || history.entries[clusterNodeAddress(node)].sample.Sequence != 3 {
t.Fatalf("reports did not upload the preceding probe before probing again: %v", sequences)
}
}