Keep mock replication jobs on schedule and off their own node (#2410)

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.
This commit is contained in:
rcourtman
2026-10-02 22:16:29 +01:00
committed by GitHub
5 changed files with 163 additions and 7 deletions
@@ -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
@@ -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
+2
View File
@@ -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)
+53 -7
View File
@@ -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
}
+98
View File
@@ -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))
}
}