From b9580317355e4bc9db0738f0ea553f651865d6e7 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Thu, 27 Aug 2026 21:15:14 +0500 Subject: [PATCH 1/2] fix(observability): start correlation grace after queue --- internal/fleetobserve/observe.go | 6 +++++- internal/fleetobserve/observe_test.go | 20 ++++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/internal/fleetobserve/observe.go b/internal/fleetobserve/observe.go index e485a62..8970ed9 100644 --- a/internal/fleetobserve/observe.go +++ b/internal/fleetobserve/observe.go @@ -593,7 +593,11 @@ func summarizeQueue(snapshot queueintent.Snapshot, platform config.Config, now t } summary.ByPriority[intent.Priority]++ summary.ByScaleSet[intent.ScaleSetName]++ - correlationAge := now.Sub(intent.QueueTime) + correlationSince := intent.QueueTime + if intent.State != queueintent.StateQueued { + correlationSince = intent.StateEnteredAt + } + correlationAge := now.Sub(correlationSince) if !strings.Contains(intent.Repository, "/") { summary.UnboundRepository++ if intent.State != queueintent.StateQueued && correlationAge >= queueCorrelationGracePeriod { diff --git a/internal/fleetobserve/observe_test.go b/internal/fleetobserve/observe_test.go index 503e0b9..90a9c81 100644 --- a/internal/fleetobserve/observe_test.go +++ b/internal/fleetobserve/observe_test.go @@ -274,6 +274,7 @@ func TestQueueSummarySeparatesTransientAndPersistentCorrelationGaps(t *testing.T persistent.Key = "persistent" persistent.JobID = "6c3077ba-3664-5824-b2cf-e22a31b25f44" persistent.QueueTime = observationTime.Add(-queueCorrelationGracePeriod) + persistent.StateEnteredAt = observationTime.Add(-queueCorrelationGracePeriod) summary, err := summarizeQueue(queueintent.Snapshot{Active: []queueintent.Intent{transient, persistent}}, testPlatform(t), observationTime) if err != nil { t.Fatal(err) @@ -304,6 +305,25 @@ func TestQueueSummaryDoesNotPageOnRehydratedJobWaitingForCapacity(t *testing.T) } } +func TestQueueSummaryStartsCorrelationGraceWhenCapacityWaitEnds(t *testing.T) { + assigned := queueintent.Intent{ + Key: "assigned-rehydrated", ScaleSetID: 11, JobID: "6c3077ba-3664-5824-b2cf-e22a31b25f44", + ScaleSetName: "nddev-linux-integration", Repository: "owner", + WorkflowRef: "authoritative-rehydration", EventName: "push", + QueueTime: observationTime.Add(-10 * queueCorrelationGracePeriod), + State: queueintent.StateAssigned, Priority: 1, + StateEnteredAt: observationTime.Add(-queueCorrelationGracePeriod + time.Second), UpdatedAt: observationTime.Add(-time.Second), + ExpiresAt: observationTime.Add(time.Minute), + } + summary, err := summarizeQueue(queueintent.Snapshot{Active: []queueintent.Intent{assigned}}, testPlatform(t), observationTime) + if err != nil { + t.Fatal(err) + } + if summary.MissingWorkflowRunIDBeyondGrace != 0 || summary.UnboundRepositoryBeyondGrace != 0 { + t.Fatalf("newly assigned correlation classification = %#v", summary) + } +} + func TestCollectorMarksRunningIntentWithoutExecutionLeaseUnhealthy(t *testing.T) { collector := healthyCollector(t) collector.Journal = func(context.Context) (providerjournal.Journal, error) { From 261864dcfa5687cab4e1b9286f47d9a191f496e3 Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Thu, 27 Aug 2026 21:16:47 +0500 Subject: [PATCH 2/2] fix(observability): page correlation only after running --- internal/fleetobserve/observe.go | 4 ++-- internal/fleetobserve/observe_test.go | 5 +++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/internal/fleetobserve/observe.go b/internal/fleetobserve/observe.go index 8970ed9..d2f22f9 100644 --- a/internal/fleetobserve/observe.go +++ b/internal/fleetobserve/observe.go @@ -600,7 +600,7 @@ func summarizeQueue(snapshot queueintent.Snapshot, platform config.Config, now t correlationAge := now.Sub(correlationSince) if !strings.Contains(intent.Repository, "/") { summary.UnboundRepository++ - if intent.State != queueintent.StateQueued && correlationAge >= queueCorrelationGracePeriod { + if intent.State == queueintent.StateRunning && correlationAge >= queueCorrelationGracePeriod { summary.UnboundRepositoryBeyondGrace++ } } @@ -617,7 +617,7 @@ func summarizeQueue(snapshot queueintent.Snapshot, platform config.Config, now t // without a workflow-run identity; the authenticated runner claim // supplies that identity after assignment. Keep the raw gap visible, // but page only after the intent has actually left the queue. - if intent.State != queueintent.StateQueued && correlationAge >= queueCorrelationGracePeriod { + if intent.State == queueintent.StateRunning && correlationAge >= queueCorrelationGracePeriod { summary.MissingWorkflowRunIDBeyondGrace++ } } diff --git a/internal/fleetobserve/observe_test.go b/internal/fleetobserve/observe_test.go index 90a9c81..5eeb582 100644 --- a/internal/fleetobserve/observe_test.go +++ b/internal/fleetobserve/observe_test.go @@ -275,6 +275,7 @@ func TestQueueSummarySeparatesTransientAndPersistentCorrelationGaps(t *testing.T persistent.JobID = "6c3077ba-3664-5824-b2cf-e22a31b25f44" persistent.QueueTime = observationTime.Add(-queueCorrelationGracePeriod) persistent.StateEnteredAt = observationTime.Add(-queueCorrelationGracePeriod) + persistent.State = queueintent.StateRunning summary, err := summarizeQueue(queueintent.Snapshot{Active: []queueintent.Intent{transient, persistent}}, testPlatform(t), observationTime) if err != nil { t.Fatal(err) @@ -305,14 +306,14 @@ func TestQueueSummaryDoesNotPageOnRehydratedJobWaitingForCapacity(t *testing.T) } } -func TestQueueSummaryStartsCorrelationGraceWhenCapacityWaitEnds(t *testing.T) { +func TestQueueSummaryDoesNotPageAssignedCapacityWait(t *testing.T) { assigned := queueintent.Intent{ Key: "assigned-rehydrated", ScaleSetID: 11, JobID: "6c3077ba-3664-5824-b2cf-e22a31b25f44", ScaleSetName: "nddev-linux-integration", Repository: "owner", WorkflowRef: "authoritative-rehydration", EventName: "push", QueueTime: observationTime.Add(-10 * queueCorrelationGracePeriod), State: queueintent.StateAssigned, Priority: 1, - StateEnteredAt: observationTime.Add(-queueCorrelationGracePeriod + time.Second), UpdatedAt: observationTime.Add(-time.Second), + StateEnteredAt: observationTime.Add(-10 * queueCorrelationGracePeriod), UpdatedAt: observationTime.Add(-time.Second), ExpiresAt: observationTime.Add(time.Minute), } summary, err := summarizeQueue(queueintent.Snapshot{Active: []queueintent.Intent{assigned}}, testPlatform(t), observationTime)