From a4a26fe842e4f1360619d1f2ec4bad9785fbde6a Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Wed, 26 Aug 2026 19:32:06 +0100 Subject: [PATCH] Preserve availability result delivery order Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 13 +++++++++ .../v6/internal/subsystems/registry.json | 1 + internal/hostagent/availability.go | 2 +- internal/hostagent/availability_test.go | 27 +++++++++++++++++++ .../canonical_completion_guard_test.py | 1 + .../release_control/subsystem_lookup_test.py | 1 + 6 files changed, 44 insertions(+), 1 deletion(-) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 5b530a823..018359689 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -5898,6 +5898,19 @@ path. Delivery queues, authentication failures, and persisted host-report buffers are destination-scoped; failure of one destination cannot block or replay reports to another. +Assigned availability-probe observations use the same primary-report delivery +boundary. Concurrent probe completions must allocate their monotonic sequence +and append to the pending queue in one ordered critical section; the queue must +never contain a newer sequence before an older one. A report snapshot records +the last included sequence as its delivery high-water mark. A successful +primary acknowledgement removes only observations at or below that mark, while +observations completed after the snapshot remain queued for the next report; +failed or buffered delivery retains the whole unacknowledged batch. This keeps +multi-target probe concurrency from stranding already-delivered observations +or discarding observations that the primary never received. The scheduler and +concurrent-enqueue coverage in `internal/hostagent/availability_test.go` pins +the ordering and acknowledgement contract. + Observer configuration is explicit, versioned, and file-backed. It contains no raw token values and resolves each token from a separate private absolute-path file. Proxmox registration is also destination-scoped: the primary retains its diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index 0a3e4a418..3c89fdb07 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -1581,6 +1581,7 @@ "internal/hostagent/agent_flushbuffer_test.go", "internal/hostagent/agent_metrics_test.go", "internal/hostagent/agent_new_test.go", + "internal/hostagent/availability_test.go", "internal/hostagent/command_client_test.go", "internal/hostagent/commands_deploy_test.go", "internal/hostagent/commands_host_update_test.go", diff --git a/internal/hostagent/availability.go b/internal/hostagent/availability.go index dfbc32230..320ba681f 100644 --- a/internal/hostagent/availability.go +++ b/internal/hostagent/availability.go @@ -281,9 +281,9 @@ func (m *availabilityProbeModule) runProbe(ctx context.Context, target config.Av // earliest observations rather than unbounded memory. func (m *availabilityProbeModule) enqueue(result agentshost.AvailabilityProbeResult) { m.mu.Lock() + defer m.mu.Unlock() m.sequence++ pending := pendingAvailabilityResult{sequence: m.sequence, result: result} - m.mu.Unlock() m.pending.Push(pending) } diff --git a/internal/hostagent/availability_test.go b/internal/hostagent/availability_test.go index da201c08e..e27166085 100644 --- a/internal/hostagent/availability_test.go +++ b/internal/hostagent/availability_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -312,6 +313,32 @@ func TestAvailabilityModuleQueueDropsOldestBeyondCapacity(t *testing.T) { } } +func TestAvailabilityModuleConcurrentEnqueuePreservesSequenceOrder(t *testing.T) { + module := testProbeModule(t, nil) + start := make(chan struct{}) + var workers sync.WaitGroup + for i := 0; i < availabilityPendingCapacity*4; i++ { + workers.Add(1) + go func(id int) { + defer workers.Done() + <-start + module.enqueue(agentshost.AvailabilityProbeResult{TargetID: fmt.Sprintf("target-%d", id)}) + }(i) + } + close(start) + workers.Wait() + + items := module.pending.Items() + if len(items) != availabilityPendingCapacity { + t.Fatalf("queued results = %d, want capacity %d", len(items), availabilityPendingCapacity) + } + for i := 1; i < len(items); i++ { + if items[i].sequence <= items[i-1].sequence { + t.Fatalf("queue sequence is not increasing at %d: %d followed %d", i, items[i].sequence, items[i-1].sequence) + } + } +} + func TestAvailabilityModuleStatusOnlyWhileAssigned(t *testing.T) { module := testProbeModule(t, nil) if _, ok := module.moduleStatus(); ok { diff --git a/scripts/release_control/canonical_completion_guard_test.py b/scripts/release_control/canonical_completion_guard_test.py index 26f82dd6e..3a78d874c 100644 --- a/scripts/release_control/canonical_completion_guard_test.py +++ b/scripts/release_control/canonical_completion_guard_test.py @@ -730,6 +730,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "internal/hostagent/agent_flushbuffer_test.go", "internal/hostagent/agent_metrics_test.go", "internal/hostagent/agent_new_test.go", + "internal/hostagent/availability_test.go", "internal/hostagent/command_client_test.go", "internal/hostagent/commands_deploy_test.go", "internal/hostagent/commands_host_update_test.go", diff --git a/scripts/release_control/subsystem_lookup_test.py b/scripts/release_control/subsystem_lookup_test.py index 2aea21387..4d071aca3 100644 --- a/scripts/release_control/subsystem_lookup_test.py +++ b/scripts/release_control/subsystem_lookup_test.py @@ -4600,6 +4600,7 @@ class SubsystemLookupTest(unittest.TestCase): "internal/hostagent/agent_flushbuffer_test.go", "internal/hostagent/agent_metrics_test.go", "internal/hostagent/agent_new_test.go", + "internal/hostagent/availability_test.go", "internal/hostagent/command_client_test.go", "internal/hostagent/commands_deploy_test.go", "internal/hostagent/commands_host_update_test.go",