From 9f39a092b164d3cd1a0a4a111bae8616dfbb989a Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Fri, 2 Oct 2026 21:45:13 +0100 Subject: [PATCH] Keep mock replication jobs on schedule and off their own node The mock generator stamped each job's last and next sync once at startup, with the last sync up to an hour old on a 15-minute schedule. Most jobs therefore started out overdue, and on a long-running demo every job read as hours overdue in red while still reporting Healthy. The demo's Replication tab looked broken to anyone evaluating Pulse. Replication timing now follows the fixture clock. Each job keeps a stable phase inside its schedule window, so a healthy job last synced less than one interval ago and is next due in the future, and a failing job's last good sync falls behind by its failure count. The target node was picked by index and could equal the source, which showed a job replicating pve1 to pve1. The source is no longer a candidate, and an estate without a second node gets no jobs. --- .../v6/internal/subsystems/monitoring.md | 7 ++ .../subsystems/performance-and-scalability.md | 3 + internal/mock/fixture_graph.go | 2 + internal/mock/generator.go | 60 ++++++++++-- internal/mock/generator_test.go | 98 +++++++++++++++++++ 5 files changed, 163 insertions(+), 7 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index a8a2e2ebed..b845df3e01 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1483,6 +1483,13 @@ service-history reads plus denial/recovery without fabricated samples. resource timestamps between cohorts, cover every node within twenty seconds, and refresh provider-backed fixtures only once per full rotation so one demo tick cannot manufacture an estate-wide WebSocket delta. + Mock replication timing follows that same refresh instead of staying at its + startup stamp: each job keeps a stable phase inside its schedule window and + is rewritten only when that phase passes, so a healthy job always last + synced less than one interval ago with its next sync in the future, a + failing job's last good sync falls behind by its failure count, and a + long-running demo never reports healthy jobs as overdue. A mock job never + targets its own source node, and an estate without a second node has none. Mock metrics history must also stay bounded independently of estate size: eager multi-day PVE guest history is limited to a deterministic sample spread across the estate, while every omitted guest continues to receive the same diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index c84ba58888..8df6300caa 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -356,6 +356,9 @@ leaving broadcast headroom to coalesce several sampler invalidations. A cohort must not rebase unchanged guest, node, storage, or disk evidence, and the provider-backed fixture set refreshes once per full rotation rather than on every tick. +Mock replication timing is advanced in that same full-rotation refresh, and +only a job whose schedule phase has passed is rewritten, so keeping demo +replication on schedule adds no per-tick change for jobs that did not run. The browser applies resource deltas to its connection-scoped raw baseline, but canonicalizes and reconciles only changed resources plus the host-merge groups diff --git a/internal/mock/fixture_graph.go b/internal/mock/fixture_graph.go index c0d99e0cf9..a95177d919 100644 --- a/internal/mock/fixture_graph.go +++ b/internal/mock/fixture_graph.go @@ -68,6 +68,7 @@ func (g *FixtureGraph) UpdateMetrics(cfg MockConfig, now time.Time) { applyDemoScenarioGraph(g, cfg, now) syncMetricFixtureRegistriesFromGraph(*g) updateFixtureStateMetricsAt(&g.State, cfg, now) + rebaseMockReplicationJobs(g.State.ReplicationJobs, now) g.PlatformFixtures = rebasePlatformFixtures(g.PlatformFixtures, now) g.AvailabilityFixtures = rebaseAvailabilityFixtures(g.AvailabilityFixtures, now) applyDemoScenarioGraph(g, cfg, now) @@ -125,6 +126,7 @@ func (g *FixtureGraph) UpdateMetricCohort( supplementalUptimeStep: currentMockUpdateStepInt64() * supplementalTicks, }) if includeSupplemental { + rebaseMockReplicationJobs(g.State.ReplicationJobs, now) g.PlatformFixtures = rebasePlatformFixtures(g.PlatformFixtures, now) g.AvailabilityFixtures = rebaseAvailabilityFixtures(g.AvailabilityFixtures, now) g.DiscoveryFixtures = buildDiscoveryFixtures(g.State, now) diff --git a/internal/mock/generator.go b/internal/mock/generator.go index 6af546e5b7..c1d914d06d 100644 --- a/internal/mock/generator.go +++ b/internal/mock/generator.go @@ -793,6 +793,53 @@ func computeGuestCountsForNode(config MockConfig, nodeRole string, node models.N return vmCount, lxcCount } +// mockReplicationInterval matches the "*/15" schedule every mock job carries. +const mockReplicationInterval = 15 * time.Minute + +// mockReplicationTargetNode picks a replication target for a guest. Replication +// copies a guest to a different node, so the source is never a candidate; an +// estate with no second node has nothing to replicate to. +func mockReplicationTargetNode(nodes []models.Node, sourceNode string, start int) (string, bool) { + for offset := 0; offset < len(nodes); offset++ { + candidate := nodes[(start+offset)%len(nodes)].Name + if candidate != "" && candidate != sourceNode { + return candidate, true + } + } + return "", false +} + +// rebaseMockReplicationJobs keeps replication timing on the fixture clock. +// The generator stamps last and next sync once, so without a rebase every job +// on a long-running demo reads as hours overdue while still reporting ok. Each +// job keeps a stable phase inside its schedule window: a healthy job last +// synced less than one interval ago and is next due in the future, and a +// failing job's last good sync falls further behind with each failure. A job +// is rewritten only when its phase passes, so a refresh does not manufacture a +// change for jobs that did not run. +func rebaseMockReplicationJobs(jobs []models.ReplicationJob, now time.Time) { + windowSeconds := uint64(mockReplicationInterval / time.Second) + for i := range jobs { + job := &jobs[i] + phase := time.Duration(mockStableHash64("replication-phase", job.ID)%windowSeconds) * time.Second + lastRun := now.Truncate(mockReplicationInterval).Add(phase) + if lastRun.After(now) { + lastRun = lastRun.Add(-mockReplicationInterval) + } + nextSync := lastRun.Add(mockReplicationInterval) + lastSync := lastRun.Add(-time.Duration(job.FailCount) * mockReplicationInterval) + if job.LastSyncTime != nil && job.LastSyncUnix == lastSync.Unix() && job.NextSyncUnix == nextSync.Unix() { + continue + } + + job.LastSyncTime = ptrTime(lastSync) + job.LastSyncUnix = lastSync.Unix() + job.NextSyncTime = ptrTime(nextSync) + job.NextSyncUnix = nextSync.Unix() + job.LastPolled = now + } +} + func generateReplicationJobs(nodes []models.Node, vms []models.VM) []models.ReplicationJob { if len(nodes) == 0 || len(vms) == 0 { return []models.ReplicationJob{} @@ -814,10 +861,13 @@ func generateReplicationJobs(nodes []models.Node, vms []models.VM) []models.Repl instance = vm.Node } + targetNode, ok := mockReplicationTargetNode(nodes, vm.Node, (i+1)%nodeCount) + if !ok { + continue + } + jobNumber := i % 3 jobID := fmt.Sprintf("%d-%d", vm.VMID, jobNumber) - lastSync := now.Add(-time.Duration(300+rand.Intn(3600)) * time.Second) - nextSync := lastSync.Add(15 * time.Minute) durationSeconds := 90 + rand.Intn(240) durationHuman := formatSecondsAsClock(durationSeconds) status := "idle" @@ -835,7 +885,6 @@ func generateReplicationJobs(nodes []models.Node, vms []models.VM) []models.Repl status = "syncing" } - targetNode := nodes[(i+1)%nodeCount].Name rate := 80.0 + rand.Float64()*140.0 job := models.ReplicationJob{ @@ -858,12 +907,8 @@ func generateReplicationJobs(nodes []models.Node, vms []models.VM) []models.Repl State: status, Status: status, LastSyncStatus: lastStatus, - LastSyncTime: ptrTime(lastSync), - LastSyncUnix: lastSync.Unix(), LastSyncDurationSeconds: durationSeconds, LastSyncDurationHuman: durationHuman, - NextSyncTime: ptrTime(nextSync), - NextSyncUnix: nextSync.Unix(), DurationSeconds: durationSeconds, DurationHuman: durationHuman, FailCount: failCount, @@ -875,6 +920,7 @@ func generateReplicationJobs(nodes []models.Node, vms []models.VM) []models.Repl jobs = append(jobs, job) } + rebaseMockReplicationJobs(jobs, now) return jobs } diff --git a/internal/mock/generator_test.go b/internal/mock/generator_test.go index 2109ba6401..1c254a197c 100644 --- a/internal/mock/generator_test.go +++ b/internal/mock/generator_test.go @@ -1193,3 +1193,101 @@ func TestFixtureGraphIncludesBackupRunningFixture(t *testing.T) { } } } + +func replicationFixtureEstate() ([]models.Node, []models.VM) { + nodes := []models.Node{{Name: "pve1"}, {Name: "pve2"}, {Name: "pve3"}} + vms := make([]models.VM, 0, 9) + for i := 0; i < 9; i++ { + // Every guest lives on pve1, so the old (i+1) modulo target picked the + // source node for every third job. + vms = append(vms, models.VM{VMID: 100 + i, Name: "guest", Node: "pve1", Instance: "cluster"}) + } + return nodes, vms +} + +func TestGenerateReplicationJobsNeverTargetsTheSourceNode(t *testing.T) { + nodes, vms := replicationFixtureEstate() + + jobs := generateReplicationJobs(nodes, vms) + if len(jobs) == 0 { + t.Fatal("expected replication jobs for a three-node estate") + } + for _, job := range jobs { + if job.TargetNode == job.SourceNode { + t.Fatalf("job %s replicates %s to itself", job.JobID, job.SourceNode) + } + } +} + +func TestGenerateReplicationJobsNeedsASecondNode(t *testing.T) { + _, vms := replicationFixtureEstate() + + if jobs := generateReplicationJobs([]models.Node{{Name: "pve1"}}, vms); len(jobs) != 0 { + t.Fatalf("expected no replication jobs without a target node, got %d", len(jobs)) + } +} + +func TestRebaseMockReplicationJobsKeepsJobsOnSchedule(t *testing.T) { + nodes, vms := replicationFixtureEstate() + jobs := generateReplicationJobs(nodes, vms) + jobs[0].FailCount = 2 + jobs[0].LastSyncStatus = "error" + for i := 1; i < len(jobs); i++ { + jobs[i].FailCount = 0 + jobs[i].LastSyncStatus = "ok" + } + + start := time.Date(2026, 10, 2, 9, 7, 0, 0, time.UTC) + // A long-running demo used to leave every job hours overdue; walk the + // fixture clock well past one schedule window. + for _, now := range []time.Time{start, start.Add(47 * time.Minute), start.Add(26 * time.Hour)} { + rebaseMockReplicationJobs(jobs, now) + for i, job := range jobs { + if job.LastSyncTime == nil || job.NextSyncTime == nil { + t.Fatalf("job %s lost its sync times", job.JobID) + } + if job.LastSyncUnix != job.LastSyncTime.Unix() || job.NextSyncUnix != job.NextSyncTime.Unix() { + t.Fatalf("job %s unix companions drifted from its sync times", job.JobID) + } + if !job.NextSyncTime.After(now) || job.NextSyncTime.Sub(now) > mockReplicationInterval { + t.Fatalf("job %s next sync %s is not within one interval after %s", job.JobID, job.NextSyncTime, now) + } + age := now.Sub(*job.LastSyncTime) + if i == 0 { + if age < 2*mockReplicationInterval { + t.Fatalf("failing job last synced %s ago, want at least two missed intervals", age) + } + continue + } + if age < 0 || age >= mockReplicationInterval { + t.Fatalf("healthy job %s last synced %s ago, want inside one interval", job.JobID, age) + } + } + } +} + +func TestRebaseMockReplicationJobsLeavesUnchangedJobsAlone(t *testing.T) { + nodes, vms := replicationFixtureEstate() + jobs := generateReplicationJobs(nodes, vms) + start := time.Date(2026, 10, 2, 9, 0, 0, 0, time.UTC) + rebaseMockReplicationJobs(jobs, start) + before := append([]models.ReplicationJob(nil), jobs...) + + // One second later no job's phase can have passed for all of them, and a + // job whose phase has not passed must keep its polled stamp and pointers so + // the refresh does not manufacture a change for every job on every tick. + rebaseMockReplicationJobs(jobs, start.Add(time.Second)) + unchanged := 0 + for i := range jobs { + if jobs[i].LastSyncUnix != before[i].LastSyncUnix { + continue + } + unchanged++ + if !jobs[i].LastPolled.Equal(before[i].LastPolled) || jobs[i].LastSyncTime != before[i].LastSyncTime { + t.Fatalf("job %s was rewritten although its schedule phase had not passed", jobs[i].JobID) + } + } + if unchanged < len(jobs)-1 { + t.Fatalf("expected at most one job to roll over in one second, %d of %d stayed put", unchanged, len(jobs)) + } +}