diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md
index d70c06430f..3be657745e 100644
--- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md
+++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md
@@ -8071,3 +8071,37 @@ This does not change token scopes, agent admission, or existing agent cleanup.
Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and
`TestInitializeSystemSettingsPreservesExistingBytes`, `TestInitializeSystemSettingsMissing`,
`TestInitializeSystemSettingsReadError`.
+
+### Unified metric replay guard does not change agent admission
+
+The unified metric sync runs on host-agent ingest boundaries and on read-side
+registry rebuilds. A per-series replay guard now drops exact timestamp+value
+repeats before the metrics batch is enqueued, reducing WAL churn without
+changing host admission, token binding or removal-block behaviour. Focused proof
+lives in `internal/monitoring/monitor_host_agents_test.go`
+(`TestDedupeUnifiedMetricWritesDropsExactReplays`).
+
+### Docker update preflight accepts any local RepoDigest for the planned image
+
+The typed container-update preflight resolves the container image's full
+RepoDigest set and admits the planned digest when it matches any entry rather
+than only the first. A correctly planned update therefore stays actionable when
+the local image carries more than one valid RepoDigest. The shared resolution
+path is covered by `TestAgent_getImageRepoDigests_MultipleDigestsForOneImage`
+and `TestRegistryChecker_MultipleLocalRepoDigestsSuppressFalseUpdate`; this is
+source-level proof, not installed acceptance.
+
+### Forked re-enrollment consults the base-identity removal block
+
+Host report admission resolves the reporting identity before it consults the
+removal block. When a colliding live record with the same base ID and an older
+token is still present in the unified read model, identity resolution forks the
+report onto a derived `-` identity. The removal block is keyed on the
+base machine identity, so the derived ID bypassed it and the removed machine
+was silently re-admitted under a new identity; each remove/reinstall cycle added
+another derived record. `ApplyHostReport` now also consults the removal block on
+the base identity when the presenting token had no prior binding, so a fresh
+install token clears the block and heals to the base identity, while a token
+already bound to its own derived identity (an established distinct host sharing
+a hostname or machine ID, #1753) is left untouched. Focused proof lives in
+`internal/monitoring/monitor_host_agents_test.go`.
diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md
index e3897ec5e2..8fa50f318f 100644
--- a/docs/release-control/v6/internal/subsystems/monitoring.md
+++ b/docs/release-control/v6/internal/subsystems/monitoring.md
@@ -4065,3 +4065,39 @@ provider keys, canonical IDs, JSON projection, temperature/size/cadence, separat
nodes/controller members, and confirmed inventory removal.
`TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata.
This is synthetic runtime evidence, not USB hardware or reporter acceptance.
+
+### Unified metric replay guard bounds registry-rebuild writes
+
+Read-side registry rebuilds (/api/state, websocket hydrate) re-run the unified
+metric sync as often as every two seconds, and storage observations are anchored
+to the provider's LastSeen. Before this change each rebuild re-issued the same
+rows as SQLite UPDATEs and committed a fresh transaction for no new data,
+churning the WAL and driving the disk-write amplification in #1966. Monitoring
+now tracks the last sample handed to the store per (resource type, resource id,
+metric type) series and drops exact timestamp+value repeats before the batch is
+enqueued; a corrected value at the same observation time is still written, so a
+real change is never masked. The guard is bounded so a long-lived monitor with
+churning resource IDs cannot pin memory. Focused proof lives in
+`internal/monitoring/monitor_host_agents_test.go`
+(`TestDedupeUnifiedMetricWritesDropsExactReplays`).
+
+### Docker image update comparison accepts every local RepoDigest
+
+Docker records every RepoDigest an image is known by, and the digest for a
+tag's manifest is not always the first entry. Update detection now compares the
+whole local RepoDigest set against the registry's resolved and index digests,
+so a current image whose manifest appears under a later RepoDigest is not
+reported as outdated. `TestRegistryChecker_MultipleLocalRepoDigestsSuppressFalseUpdate`,
+`TestAgent_getImageRepoDigests_MultipleDigestsForOneImage` and the multi-digest
+`TestRegistryChecker_DigestsDiffer` cases pin the comparison. This is synthetic
+registry-transport evidence, not reporter acceptance.
+
+### Forked host reports cannot bypass a base-identity removal block
+
+Host report admission resolves identity before consulting removal blocks, so a
+report that forked onto a derived `-` identity could bypass a block
+keyed on the base machine identity and be silently admitted. Monitoring now
+consults the base identity as well when the presenting token had no prior
+binding: a fresh install token clears the block and heals to the base identity,
+and a token already bound to its own derived identity is left alone (#1753).
+Focused proof lives in `internal/monitoring/monitor_host_agents_test.go`.
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 ef912446fc..c40cb46b1b 100644
--- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md
+++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md
@@ -3168,3 +3168,14 @@ The existing 500-node mixed-endpoint workload and latency budgets remain intact.
`pkg/metrics/store_additional_test.go` holds every history connection to
prove availability without relying on favourable scheduling, and covers
committed-data visibility, write rejection, clear and pool shutdown.
+
+### Metrics-store shutdown never closes the ingestion channel
+
+The metrics store's ingestion worker must not close its write channel on
+shutdown. A concurrent writer that passed the `stopping` check before `Close`,
+and the `WriteBatchSync`/`WriteBatchBounded` paths that do not consult
+`stopping` at all, may still send to that channel. Closing it races those sends
+and panics the process during shutdown. The worker instead drains already-queued
+requests without closing the channel; a late write lands in the buffered channel
+and is discarded with the store. `pkg/metrics/store_additional_test.go` pins the
+post-shutdown enqueue and the concurrent-close race.
diff --git a/internal/dockeragent/agent_collect_test.go b/internal/dockeragent/agent_collect_test.go
index dd156afa3b..74b2f553f4 100644
--- a/internal/dockeragent/agent_collect_test.go
+++ b/internal/dockeragent/agent_collect_test.go
@@ -802,3 +802,29 @@ func TestAgentClose(t *testing.T) {
t.Fatal("expected docker Close to be called")
}
}
+
+func TestAgentCollectImageRepoDigestsIncludesEveryLocalDigest(t *testing.T) {
+ agent := &Agent{
+ docker: &fakeDockerClient{
+ imageInspectWithRawFn: func(context.Context, string) (imagetypes.InspectResponse, []byte, error) {
+ return imagetypes.InspectResponse{
+ RepoDigests: []string{
+ "docker.io/library/postgres@sha256:44c4",
+ "docker.io/library/postgres@sha256:cf78",
+ },
+ Architecture: "amd64",
+ Os: "linux",
+ }, nil, nil
+ },
+ },
+ logger: zerolog.New(io.Discard),
+ }
+
+ digests, arch, os, _ := agent.getImageRepoDigests(context.Background(), "image-id", "postgres:16.15-alpine3.24")
+ if len(digests) != 2 || digests[0] != "sha256:44c4" || digests[1] != "sha256:cf78" {
+ t.Fatalf("digests = %v, want both local RepoDigests", digests)
+ }
+ if arch != "amd64" || os != "linux" {
+ t.Fatalf("platform = %q/%q, want amd64/linux", arch, os)
+ }
+}
diff --git a/internal/dockeragent/agent_image_digest_test.go b/internal/dockeragent/agent_image_digest_test.go
index f48c12cc58..ff96df38f4 100644
--- a/internal/dockeragent/agent_image_digest_test.go
+++ b/internal/dockeragent/agent_image_digest_test.go
@@ -77,6 +77,42 @@ func TestAgent_getImageRepoDigest_FallbackToFirst(t *testing.T) {
}
}
+func TestAgent_getImageRepoDigests_MultipleDigestsForOneImage(t *testing.T) {
+ agent := &Agent{
+ docker: &fakeDockerClient{
+ imageInspectWithRawFn: func(ctx context.Context, imageID string) (image.InspectResponse, []byte, error) {
+ return image.InspectResponse{
+ RepoDigests: []string{
+ "docker.io/library/postgres@sha256:44c4",
+ "docker.io/library/postgres@sha256:cf78",
+ "docker.io/other/postgres@sha256:beef",
+ },
+ Architecture: "amd64",
+ Os: "linux",
+ }, nil, nil
+ },
+ },
+ logger: zerolog.New(io.Discard),
+ }
+
+ digests, arch, os, variant := agent.getImageRepoDigests(context.Background(), "image-id", "postgres:16.15-alpine3.24")
+ want := []string{"sha256:44c4", "sha256:cf78", "sha256:beef"}
+ if len(digests) != len(want) {
+ t.Fatalf("digests = %v, want %v", digests, want)
+ }
+ for i := range want {
+ if digests[i] != want[i] {
+ t.Fatalf("digests = %v, want %v", digests, want)
+ }
+ }
+ if arch != "amd64" || os != "linux" || variant != "" {
+ t.Fatalf("platform = %q/%q/%q, want amd64/linux/", arch, os, variant)
+ }
+ if got, _, _, _ := agent.getImageRepoDigest(context.Background(), "image-id", "postgres:16.15-alpine3.24"); got != "sha256:44c4" {
+ t.Fatalf("primary digest = %q, want sha256:44c4", got)
+ }
+}
+
func TestAgent_getImageRepoDigest_InvalidRepoDigest(t *testing.T) {
agent := &Agent{
docker: &fakeDockerClient{
diff --git a/internal/dockeragent/collect.go b/internal/dockeragent/collect.go
index d2e0356f6a..28c8b92874 100644
--- a/internal/dockeragent/collect.go
+++ b/internal/dockeragent/collect.go
@@ -775,7 +775,11 @@ func (a *Agent) collectContainer(ctx context.Context, summary containertypes.Sum
// Get the actual manifest digest (RepoDigest) from the image for accurate comparison.
// The ImageID is a local content-addressable ID that differs from the registry manifest digest.
// We also get the architecture details to correctly resolve manifest lists from the registry.
- digestForComparison, arch, os, variant := a.getImageRepoDigest(containerCtx, summary.ImageID, summary.Image)
+ currentDigests, arch, os, variant := a.getImageRepoDigests(containerCtx, summary.ImageID, summary.Image)
+ digestForComparison := ""
+ if len(currentDigests) > 0 {
+ digestForComparison = currentDigests[0]
+ }
var imageToCheck string
// Always prefer the image name from inspect config as it's the authoritative source
@@ -808,7 +812,11 @@ func (a *Agent) collectContainer(ctx context.Context, summary containertypes.Sum
Str("variant", variant).
Msg("Checking update for container")
- result := a.registryChecker.CheckImageUpdate(ctx, imageToCheck, digestForComparison, arch, os, variant)
+ // Pass the whole local digest set. Docker records every RepoDigest
+ // the image is known by, and a tag's manifest digest is often not
+ // the first entry, so comparing only one digest reports a current
+ // image as outdated (#2110).
+ result := a.registryChecker.CheckImageUpdate(ctx, imageToCheck, strings.Join(currentDigests, ","), arch, os, variant)
if result != nil {
container.UpdateStatus = &agentsdocker.UpdateStatus{
UpdateAvailable: result.UpdateAvailable,
@@ -956,6 +964,20 @@ func extractHealthcheckTargets(test []string) []string {
// getImageRepoDigest retrieves the RepoDigest for an image and its platform details.
// It returns the digest, architecture, OS, and variant.
func (a *Agent) getImageRepoDigest(ctx context.Context, imageID, imageName string) (string, string, string, string) {
+ digests, arch, os, variant := a.getImageRepoDigests(ctx, imageID, imageName)
+ if len(digests) == 0 {
+ return "", arch, os, variant
+ }
+ return digests[0], arch, os, variant
+}
+
+// getImageRepoDigests returns every manifest digest the local image is known
+// by, with digests whose repository matches the requested image reference
+// first, plus the platform details. A single image can carry several
+// RepoDigests when it was pulled under more than one reference (for example a
+// tag and a platform manifest), so an update check must treat the whole set as
+// the local identity rather than trusting one arbitrary entry (#2110).
+func (a *Agent) getImageRepoDigests(ctx context.Context, imageID, imageName string) ([]string, string, string, string) {
imageInspect, _, err := a.docker.ImageInspectWithRaw(ctx, imageID)
if err != nil {
a.logger.Debug().
@@ -963,7 +985,7 @@ func (a *Agent) getImageRepoDigest(ctx context.Context, imageID, imageName strin
Str("imageID", imageID).
Str("imageName", imageName).
Msg("Failed to inspect image for RepoDigest")
- return "", "", "", ""
+ return nil, "", "", ""
}
arch := imageInspect.Architecture
@@ -972,32 +994,29 @@ func (a *Agent) getImageRepoDigest(ctx context.Context, imageID, imageName strin
if len(imageInspect.RepoDigests) == 0 {
// Locally built images won't have RepoDigests
- return "", arch, os, variant
+ return nil, arch, os, variant
}
- // Try to find a RepoDigest that matches the image reference
+ matched := make([]string, 0, len(imageInspect.RepoDigests))
+ unmatched := make([]string, 0, len(imageInspect.RepoDigests))
// RepoDigests format: "registry/repo@sha256:..."
for _, repoDigest := range imageInspect.RepoDigests {
- // Extract just the digest part (after @)
- if idx := strings.LastIndex(repoDigest, "@"); idx >= 0 {
- repoRef := repoDigest[:idx] // e.g., "docker.io/library/nginx"
- digest := repoDigest[idx+1:] // e.g., "sha256:abc..."
-
- // Check if this RepoDigest matches our image reference
- // Normalize both for comparison
- if matchesImageReference(imageName, repoRef) {
- return digest, arch, os, variant
- }
+ idx := strings.LastIndex(repoDigest, "@")
+ if idx < 0 {
+ continue
+ }
+ digest := strings.TrimSpace(repoDigest[idx+1:])
+ if digest == "" {
+ continue
+ }
+ if matchesImageReference(imageName, repoDigest[:idx]) {
+ matched = append(matched, digest)
+ } else {
+ unmatched = append(unmatched, digest)
}
}
- // If no exact match, return the first RepoDigest's digest
- // This handles cases where the image was pulled with a different tag
- if idx := strings.LastIndex(imageInspect.RepoDigests[0], "@"); idx >= 0 {
- return imageInspect.RepoDigests[0][idx+1:], arch, os, variant
- }
-
- return "", arch, os, variant
+ return append(matched, unmatched...), arch, os, variant
}
func addrString(addr netip.Addr) string {
diff --git a/internal/dockeragent/container_update_typed.go b/internal/dockeragent/container_update_typed.go
index b67f331f50..9d727f12bb 100644
--- a/internal/dockeragent/container_update_typed.go
+++ b/internal/dockeragent/container_update_typed.go
@@ -63,9 +63,24 @@ func (a *Agent) TypedContainerUpdatePreflight(ctx context.Context, runtime, cont
if inspect.Config != nil {
imageName = inspect.Config.Image
}
- repoDigest, _, _, _ = a.getImageRepoDigest(ctx, localImageID, imageName)
- repoDigest = strings.TrimSpace(repoDigest)
- matchesRepoDigest = strings.EqualFold(repoDigest, expectedImageDigest)
+ // An image can carry several RepoDigests (#2110), so accept the
+ // planned digest when it matches any of them rather than only the
+ // first entry.
+ repoDigests, _, _, _ := a.getImageRepoDigests(ctx, localImageID, imageName)
+ for _, candidate := range repoDigests {
+ candidate = strings.TrimSpace(candidate)
+ if candidate == "" {
+ continue
+ }
+ if repoDigest == "" {
+ repoDigest = candidate
+ }
+ if strings.EqualFold(candidate, expectedImageDigest) {
+ repoDigest = candidate
+ matchesRepoDigest = true
+ break
+ }
+ }
}
if !matchesLocalID && !matchesRepoDigest {
if repoDigest == "" {
diff --git a/internal/dockeragent/container_update_typed_test.go b/internal/dockeragent/container_update_typed_test.go
index 646bed64e6..a25b94be31 100644
--- a/internal/dockeragent/container_update_typed_test.go
+++ b/internal/dockeragent/container_update_typed_test.go
@@ -125,3 +125,29 @@ func TestTypedContainerUpdateDelegatesToProductionRecreatePath(t *testing.T) {
}
<-cleanupDone
}
+
+func TestTypedContainerUpdatePreflightAcceptsAnyLocalRepoDigest(t *testing.T) {
+ agent := &Agent{
+ runtime: RuntimeDocker,
+ docker: &fakeDockerClient{
+ containerInspectFn: func(context.Context, string) (containertypes.InspectResponse, error) {
+ return baseInspect(), nil
+ },
+ imageInspectWithRawFn: func(context.Context, string) (image.InspectResponse, []byte, error) {
+ return image.InspectResponse{
+ RepoDigests: []string{
+ "docker.io/library/nginx@sha256:registry-old",
+ "docker.io/library/nginx@sha256:registry-new",
+ },
+ }, nil, nil
+ },
+ },
+ logger: zerolog.Nop(),
+ }
+
+ // The planned digest is the second local RepoDigest; the preflight must
+ // still admit the update rather than refuse on a first-entry mismatch.
+ if err := agent.TypedContainerUpdatePreflight(context.Background(), string(RuntimeDocker), "container-id", "sha256:registry-new"); err != nil {
+ t.Fatalf("preflight refused a planned digest that matches a local RepoDigest: %v", err)
+ }
+}
diff --git a/internal/dockeragent/registry.go b/internal/dockeragent/registry.go
index 6b800a53d6..c9bf2b3517 100644
--- a/internal/dockeragent/registry.go
+++ b/internal/dockeragent/registry.go
@@ -182,11 +182,18 @@ func (r *RegistryChecker) CheckImageUpdate(ctx context.Context, image, currentDi
return nil
}
+ // currentDigest may be a comma-separated set of the local image's
+ // RepoDigests; report the primary one while comparing against all of them.
+ primaryCurrentDigest := currentDigest
+ if idx := strings.Index(currentDigest, ","); idx >= 0 {
+ primaryCurrentDigest = strings.TrimSpace(currentDigest[:idx])
+ }
+
// Skip digest-pinned images (image@sha256:...)
if registry == "" {
return &ImageUpdateResult{
Image: image,
- CurrentDigest: currentDigest,
+ CurrentDigest: primaryCurrentDigest,
UpdateAvailable: false,
CheckedAt: time.Now(),
Error: "digest-pinned image",
@@ -202,7 +209,7 @@ func (r *RegistryChecker) CheckImageUpdate(ctx context.Context, image, currentDi
if cached.err != "" {
return &ImageUpdateResult{
Image: image,
- CurrentDigest: currentDigest,
+ CurrentDigest: primaryCurrentDigest,
UpdateAvailable: false,
CheckedAt: time.Now(),
Error: cached.err,
@@ -210,7 +217,7 @@ func (r *RegistryChecker) CheckImageUpdate(ctx context.Context, image, currentDi
}
return &ImageUpdateResult{
Image: image,
- CurrentDigest: currentDigest,
+ CurrentDigest: primaryCurrentDigest,
LatestDigest: cached.latestDigest,
UpdateAvailable: r.digestsDiffer(currentDigest, cached.comparisonDigests),
CheckedAt: time.Now(),
@@ -231,7 +238,7 @@ func (r *RegistryChecker) CheckImageUpdate(ctx context.Context, image, currentDi
return &ImageUpdateResult{
Image: image,
- CurrentDigest: currentDigest,
+ CurrentDigest: primaryCurrentDigest,
UpdateAvailable: false,
CheckedAt: time.Now(),
Error: err.Error(),
@@ -268,7 +275,7 @@ func (r *RegistryChecker) CheckImageUpdate(ctx context.Context, image, currentDi
return &ImageUpdateResult{
Image: image,
- CurrentDigest: currentDigest,
+ CurrentDigest: primaryCurrentDigest,
LatestDigest: publicLatestDigest,
UpdateAvailable: updateAvailable,
CheckedAt: time.Now(),
@@ -281,20 +288,42 @@ func (r *RegistryChecker) digestsDiffer(current, latest string) bool {
return false
}
- // Normalize digests - lowercase and remove "sha256:" prefix
- normCurrent := strings.ToLower(strings.TrimPrefix(current, "sha256:"))
-
- // latest may contain multiple comma-separated digests (resolved + head)
+ // latest may contain multiple comma-separated digests (resolved + head).
+ // current may also hold several local RepoDigests for one image (#2110).
+ latestSet := make(map[string]struct{})
for _, l := range strings.Split(latest, ",") {
- normLatest := strings.ToLower(strings.TrimPrefix(strings.TrimSpace(l), "sha256:"))
- if normCurrent == normLatest {
+ if normalized := normalizeComparisonDigest(l); normalized != "" {
+ latestSet[normalized] = struct{}{}
+ }
+ }
+ if len(latestSet) == 0 {
+ return false
+ }
+
+ compared := false
+ for _, c := range strings.Split(current, ",") {
+ normalized := normalizeComparisonDigest(c)
+ if normalized == "" {
+ continue
+ }
+ compared = true
+ if _, ok := latestSet[normalized]; ok {
return false // Match found
}
}
+ if !compared {
+ return false
+ }
return true // No match found
}
+// normalizeComparisonDigest lowercases a digest and removes the "sha256:"
+// prefix so registry and RepoDigest spellings compare equal.
+func normalizeComparisonDigest(digest string) string {
+ return strings.ToLower(strings.TrimPrefix(strings.TrimSpace(digest), "sha256:"))
+}
+
// fetchDigest retrieves the digest for an image from the registry.
// Returns the resolved platform-specific digest AND the raw HEAD digest (which might be a manifest list).
func (r *RegistryChecker) fetchDigest(ctx context.Context, registry, repository, tag, arch, goos, variant string) (string, string, error) {
diff --git a/internal/dockeragent/registry_manifest_test.go b/internal/dockeragent/registry_manifest_test.go
index 79379826e9..c2010b36c4 100644
--- a/internal/dockeragent/registry_manifest_test.go
+++ b/internal/dockeragent/registry_manifest_test.go
@@ -123,3 +123,43 @@ func TestRegistryChecker_MultiArchDigestLayersStayConsistentAcrossCache(t *testi
t.Fatalf("expected one HEAD and one GET before the cache hit, got %d requests", requestCount)
}
}
+
+// Issue #2110: a local image can carry several RepoDigests, for example the
+// tag's manifest digest and a platform manifest digest. The update check must
+// treat any of them as the current image rather than reporting a false update
+// just because the first entry differs from the registry digest.
+func TestRegistryChecker_MultipleLocalRepoDigestsSuppressFalseUpdate(t *testing.T) {
+ checker := NewRegistryChecker(zerolog.Nop())
+ checker.httpClient = &http.Client{
+ Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+ if req.Method == http.MethodHead {
+ return newStringResponse(http.StatusOK, map[string]string{
+ "Content-Type": "application/vnd.docker.distribution.manifest.v2+json",
+ "Docker-Content-Digest": "sha256:cf78",
+ }, ""), nil
+ }
+ return newStringResponse(http.StatusOK, nil, "{}"), nil
+ }),
+ }
+
+ result := checker.CheckImageUpdate(
+ context.Background(),
+ "example.test/postgres:16.15-alpine3.24",
+ "sha256:44c4,sha256:cf78",
+ "amd64",
+ "linux",
+ "",
+ )
+ if result == nil {
+ t.Fatal("expected an update result")
+ }
+ if result.UpdateAvailable {
+ t.Fatalf("a local RepoDigest matching the registry digest must suppress the update: %#v", result)
+ }
+ if result.CurrentDigest != "sha256:44c4" {
+ t.Fatalf("primary current digest = %q, want %q", result.CurrentDigest, "sha256:44c4")
+ }
+ if result.LatestDigest != "sha256:cf78" {
+ t.Fatalf("latest digest = %q, want %q", result.LatestDigest, "sha256:cf78")
+ }
+}
diff --git a/internal/dockeragent/registry_test.go b/internal/dockeragent/registry_test.go
index 7b0a7007a3..7624a32d35 100644
--- a/internal/dockeragent/registry_test.go
+++ b/internal/dockeragent/registry_test.go
@@ -207,6 +207,24 @@ func TestRegistryChecker_DigestsDiffer(t *testing.T) {
latest: "def456,abc123",
want: true, // No match found
},
+ {
+ name: "multiple current digests match second latest",
+ current: "sha256:44c4,sha256:cf78",
+ latest: "sha256:cf78",
+ want: false, // One local RepoDigest is current
+ },
+ {
+ name: "multiple current digests no match",
+ current: "sha256:44c4,sha256:aaaa",
+ latest: "sha256:cf78",
+ want: true, // No local RepoDigest is current
+ },
+ {
+ name: "empty entries in current set are ignored",
+ current: " ,sha256:cf78",
+ latest: "sha256:cf78",
+ want: false,
+ },
}
for _, tt := range tests {
diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go
index 49d0cf35bb..4be2bf5093 100644
--- a/internal/monitoring/monitor.go
+++ b/internal/monitoring/monitor.go
@@ -1247,6 +1247,14 @@ type Monitor struct {
// that already hold m.mu can consult it without lock re-entrancy.
licenseCheckerMu sync.RWMutex
licenseChecker func(feature string) bool
+
+ // Unified metric sync replay guard. The sync runs on both ingest
+ // boundaries and read-side registry rebuilds, so one observation is often
+ // replayed many times between polls. Without this every replay re-writes
+ // the same metrics row (an UPDATE in SQLite) and churns the WAL for no new
+ // data, which is the dominant write amplification in #1966.
+ unifiedMetricSyncMu sync.Mutex
+ unifiedMetricSyncLast map[unifiedMetricSampleKey]unifiedMetricSample
}
func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) {
@@ -5171,6 +5179,63 @@ func recordSupplementalResourceChanges(store ResourceStoreInterface, changes []u
}
}
+// unifiedMetricSampleKey identifies one persisted unified-metric series. The
+// metrics store keys rows by (resource type, resource id, metric type, tier,
+// timestamp), so the replay guard must match at least those coordinates.
+type unifiedMetricSampleKey struct {
+ resourceType string
+ resourceID string
+ metricType string
+}
+
+// unifiedMetricSample is the last sample handed to the metrics store for a
+// series. Timestamps are compared at the store's second resolution so a
+// sub-second replay of the same observation is still recognised as redundant.
+type unifiedMetricSample struct {
+ timestampUnix int64
+ value float64
+}
+
+// unifiedMetricSyncMaxSeries bounds the replay guard so long-lived monitors
+// with churning resource IDs cannot pin memory. The map is reset wholesale
+// when the cap is crossed; the worst case is one extra write per series.
+const unifiedMetricSyncMaxSeries = 100000
+
+// dedupeUnifiedMetricWrites drops exact repeats of a sample already handed to
+// the metrics store. The unified sync runs on ingest boundaries and on
+// read-side registry rebuilds (which fire as often as every two seconds), so
+// the same observation is replayed many times between polls. Each replay would
+// otherwise UPDATE the same row and commit a fresh transaction, churning the
+// SQLite WAL with no new data (#1966). A changed value at the same timestamp is
+// still written so corrections are not lost.
+func (m *Monitor) dedupeUnifiedMetricWrites(writes []metrics.WriteMetric) []metrics.WriteMetric {
+ if len(writes) == 0 {
+ return writes
+ }
+
+ m.unifiedMetricSyncMu.Lock()
+ defer m.unifiedMetricSyncMu.Unlock()
+ if m.unifiedMetricSyncLast == nil || len(m.unifiedMetricSyncLast) > unifiedMetricSyncMaxSeries {
+ m.unifiedMetricSyncLast = make(map[unifiedMetricSampleKey]unifiedMetricSample, len(writes))
+ }
+
+ out := writes[:0]
+ for _, write := range writes {
+ key := unifiedMetricSampleKey{
+ resourceType: write.ResourceType,
+ resourceID: write.ResourceID,
+ metricType: write.MetricType,
+ }
+ sample := unifiedMetricSample{timestampUnix: write.Timestamp.Unix(), value: write.Value}
+ if previous, ok := m.unifiedMetricSyncLast[key]; ok && previous == sample {
+ continue
+ }
+ m.unifiedMetricSyncLast[key] = sample
+ out = append(out, write)
+ }
+ return out
+}
+
func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
return
@@ -5276,6 +5341,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
appendStoreWrite("agent", targetID, "diskwrite", metric.Value)
}
}
+ storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
if len(storeWrites) > 0 {
m.metricsStore.WriteBatchBounded(storeWrites)
}
@@ -5391,6 +5457,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
appendStoreWrite("vm", targetID, "diskwrite", metric.Value)
}
}
+ storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
if len(storeWrites) > 0 {
m.metricsStore.WriteBatchBounded(storeWrites)
}
@@ -5490,6 +5557,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
appendStoreWrite("storage", targetID, "avail", float64(free), observedAt)
}
}
+ storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
if len(storeWrites) > 0 {
m.metricsStore.WriteBatchBounded(storeWrites)
}
@@ -5697,6 +5765,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)
}
}
+ storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
if len(storeWrites) > 0 {
m.metricsStore.WriteBatchBounded(storeWrites)
}
diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go
index a8acef02b2..4bfe270a24 100644
--- a/internal/monitoring/monitor_agents.go
+++ b/internal/monitoring/monitor_agents.go
@@ -3083,6 +3083,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
reportMachineID := strings.TrimSpace(report.Host.MachineID)
identifier := baseIdentifier
+ hadPriorHostBinding := false
if tokenRecord != nil && strings.TrimSpace(tokenRecord.ID) != "" {
tokenID := strings.TrimSpace(tokenRecord.ID)
bindingKey := hostTokenMachineBindingKey(tokenID, hostname, reportMachineID)
@@ -3099,6 +3100,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
// the unbound path so this machine gets its own identity slot.
boundID = ""
}
+ hadPriorHostBinding = boundID != ""
if boundID != "" {
m.hostTokenBindings[bindingKey] = boundID
}
@@ -3196,6 +3198,22 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config.
tokenID = strings.TrimSpace(tokenRecord.ID)
}
blocked, wasRemoved := m.lookupRemovedHostAgent(identifier, hostname, report.Host.MachineID, tokenID)
+ if !wasRemoved && identifier != baseIdentifier && !hadPriorHostBinding {
+ // Identity resolution can fork a re-enrolling machine onto a derived
+ // suffix before the removal block is consulted: the fork keys off a
+ // colliding live record with the same base ID and an older token, and
+ // the block is keyed on the identity the operator removed (the base
+ // machine identity). Consult the base identity too, so the block is
+ // honoured (rejected, or cleared for a fresh install token) instead of
+ // silently admitting the machine under a derived identity (#2113).
+ //
+ // A token that already had its own binding was forked previously and is
+ // an established distinct host sharing a hostname/machine ID, so the
+ // removed host's block must not capture it (#1753).
+ if baseBlocked, baseRemoved := m.lookupRemovedHostAgent(baseIdentifier, hostname, report.Host.MachineID, tokenID); baseRemoved {
+ blocked, wasRemoved = baseBlocked, true
+ }
+ }
if wasRemoved && removedHostAgentAllowsFreshReenroll(blocked, identifier, report, tokenRecord) {
// A token minted after the host was removed means the user generated a
// fresh install command for this machine: that is explicit re-enroll
diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go
index f8d32ea0e8..05bb0a1e33 100644
--- a/internal/monitoring/monitor_host_agents_test.go
+++ b/internal/monitoring/monitor_host_agents_test.go
@@ -5884,3 +5884,134 @@ func TestAgentReportsPreserveExplicitDiskIncludes(t *testing.T) {
check(t, host.Disks)
})
}
+
+// Issue #1966: read-side registry rebuilds re-run the unified metric sync as
+// often as every two seconds, so one poll observation was re-written many
+// times between polls. The store upserts on (resource, metric, tier,
+// timestamp), so each replay committed a fresh transaction for no new data and
+// churned the SQLite WAL. The replay guard drops exact timestamp+value repeats
+// before the batch is enqueued while still writing a corrected value.
+func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) {
+ monitor := &Monitor{}
+ observedAt := time.Date(2026, time.September, 19, 12, 0, 0, 0, time.UTC)
+ write := metrics.WriteMetric{
+ ResourceType: "storage",
+ ResourceID: "pool:tank",
+ MetricType: "usage",
+ Value: 62,
+ Timestamp: observedAt,
+ Tier: metrics.TierRaw,
+ }
+
+ first := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write})
+ if len(first) != 1 {
+ t.Fatalf("first write count = %d, want 1", len(first))
+ }
+ // The same observation replayed by a read-side registry rebuild must not
+ // reach the store again: that UPDATE is the write amplification in #1966.
+ replay := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write})
+ if len(replay) != 0 {
+ t.Fatalf("replayed write count = %d, want 0", len(replay))
+ }
+
+ // A corrected value at the same observation time is still written.
+ corrected := write
+ corrected.Value = 63
+ if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{corrected}); len(got) != 1 {
+ t.Fatalf("corrected write count = %d, want 1", len(got))
+ }
+
+ // A newer observation for the same series is written.
+ advanced := write
+ advanced.Value = 63
+ advanced.Timestamp = observedAt.Add(time.Minute)
+ if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{advanced}); len(got) != 1 {
+ t.Fatalf("advanced write count = %d, want 1", len(got))
+ }
+}
+
+// Issue #2113: a removed host agent re-enrolls under a derived suffixed
+// identity because identity resolution forks the machine before the removal
+// block is consulted. The block is keyed on the base machine identity, so the
+// forked ID bypasses it and the machine is silently admitted instead of being
+// rejected or cleared. This reproduces the fork trigger (a live base-identity
+// record still present in the unified read model with an older token) and
+// asserts a fresh install token heals to the base identity.
+func TestApplyHostReportHonoursRemovalBlockWhenIdentityWouldFork(t *testing.T) {
+ now := time.Now().UTC()
+ const (
+ baseID = "bf3f9ff3273346d9944c49160dc57288"
+ hostname = "truenas-iscsi.local"
+ machineID = "bf3f9ff3273346d9944c49160dc57288"
+ )
+
+ live := models.Host{
+ ID: baseID,
+ MachineID: machineID,
+ Hostname: hostname,
+ Platform: "linux",
+ Status: "online",
+ LastSeen: now,
+ IntervalSeconds: 30,
+ TokenID: "token-old",
+ }
+ state := models.NewState()
+ state.Hosts = []models.Host{live}
+
+ registry := unifiedresources.NewRegistry(nil)
+ registry.IngestSnapshot(state.GetSnapshot())
+ adapter := unifiedresources.NewMonitorAdapter(registry)
+
+ monitor := &Monitor{
+ state: state,
+ resourceStore: adapter,
+ alertManager: alerts.NewManager(),
+ hostTokenBindings: map[string]string{"token-old": baseID},
+ removedHostAgents: make(map[string]time.Time),
+ rateTracker: NewRateTracker(),
+ config: &config.Config{},
+ }
+ t.Cleanup(func() { monitor.alertManager.Stop() })
+
+ // The operator removed the host while the live record was still present in
+ // the read model: the tombstone is keyed on the base identity.
+ removedAt := now.Add(-2 * time.Hour)
+ monitor.state.AddRemovedHostAgent(models.RemovedHostAgent{
+ ID: baseID,
+ Hostname: hostname,
+ MachineID: machineID,
+ TokenID: "token-old",
+ RemovedAt: removedAt,
+ })
+ monitor.removedHostAgents[baseID] = removedAt
+
+ report := agentshost.Report{
+ Agent: agentshost.AgentInfo{ID: baseID, Type: "unified", IntervalSeconds: 30},
+ Host: agentshost.HostInfo{
+ ID: baseID,
+ Hostname: hostname,
+ MachineID: machineID,
+ Platform: "linux",
+ },
+ Timestamp: now,
+ }
+ // The fresh token post-dates the removal, but the colliding live record is
+ // still reporting, so the stale-identity reuse path declines and identity
+ // resolution forks.
+ freshToken := &config.APITokenRecord{
+ ID: "token-new",
+ Name: "reinstall",
+ CreatedAt: now.Add(-time.Hour),
+ }
+
+ host, err := monitor.ApplyHostReport(report, freshToken)
+ if err != nil {
+ t.Fatalf("fresh re-enroll should clear the block, got %v", err)
+ }
+ if host.ID != baseID {
+ t.Fatalf("expected re-enroll to heal to base identity %q, got derived %q", baseID, host.ID)
+ }
+ if len(monitor.state.GetRemovedHostAgents()) != 0 {
+ t.Fatalf("expected removal block to be cleared, still have %+v", monitor.state.GetRemovedHostAgents())
+ }
+}
diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go
index b478da9c2c..e998c8629d 100644
--- a/pkg/metrics/store.go
+++ b/pkg/metrics/store.go
@@ -1842,12 +1842,21 @@ func (s *Store) backgroundWorker() {
if batch := s.drainBuffer(); len(batch) > 0 {
remaining = append(remaining, writeRequest{metrics: batch})
}
- close(s.writeCh)
- for req := range s.writeCh {
- remaining = append(remaining, req)
+ // Do NOT close writeCh. Concurrent writers (WriteWithTier after
+ // its stopping check, and WriteBatchSync/WriteBatchBounded which
+ // never check stopping) may still be sending; closing the channel
+ // races those sends and panics the process on shutdown. Drain what
+ // is already queued instead, and let any late write land in the
+ // buffered channel to be discarded with the store.
+ for {
+ select {
+ case req := <-s.writeCh:
+ remaining = append(remaining, req)
+ default:
+ s.processWriteRequests(remaining)
+ return
+ }
}
- s.processWriteRequests(remaining)
- return
case req, ok := <-s.writeCh:
if !ok {
diff --git a/pkg/metrics/store_additional_test.go b/pkg/metrics/store_additional_test.go
index 2f8c2d7b79..83eca705bd 100644
--- a/pkg/metrics/store_additional_test.go
+++ b/pkg/metrics/store_additional_test.go
@@ -1455,3 +1455,90 @@ func TestStoreStatsReaderObservesCommittedDataAndCloses(t *testing.T) {
t.Fatal("main pool remained open after store shutdown")
}
}
+
+// Regression coverage for the shutdown send-on-closed-channel review of
+// pkg/metrics/store.go. The ingestion worker used to close writeCh on
+// <-stopCh. A writer that passed the stopping check just before Close, or a
+// WriteBatchSync/WriteBatchBounded caller (which never checks stopping), could
+// then send on the closed channel and panic the process during shutdown.
+
+func newShutdownRaceStore(t *testing.T) *Store {
+ t.Helper()
+ dir := t.TempDir()
+ cfg := DefaultConfig(dir)
+ cfg.DBPath = filepath.Join(dir, "metrics-shutdown-race.db")
+ cfg.FlushInterval = time.Hour
+ store, err := NewStore(cfg)
+ if err != nil {
+ t.Fatalf("NewStore returned error: %v", err)
+ }
+ return store
+}
+
+func shutdownRaceMetric() bufferedMetric {
+ return bufferedMetric{
+ resourceType: "vm",
+ resourceID: "shutdown-race",
+ metricType: "cpu",
+ value: 1,
+ timestamp: time.Unix(1_700_000_000, 0),
+ tier: TierRaw,
+ }
+}
+
+// A send that reaches the ingestion channel after the worker has finished
+// shutdown must not panic. Before the fix the worker closed writeCh, so this
+// deterministic post-Close enqueue hit a closed channel.
+func TestEnqueueWriteAfterShutdownDoesNotPanic(t *testing.T) {
+ store := newShutdownRaceStore(t)
+ if err := store.Close(); err != nil {
+ t.Fatalf("Close returned error: %v", err)
+ }
+
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatalf("enqueueWrite panicked after shutdown: %v", r)
+ }
+ }()
+ store.enqueueWrite(writeRequest{metrics: []bufferedMetric{shutdownRaceMetric()}})
+}
+
+// Concurrent writers racing a Close exercise the real window: WriteBatchSync
+// and WriteBatchBounded do not consult stopping, and WriteWithTier checks it
+// before releasing bufferMu. None of them may panic.
+func TestConcurrentWritesDuringShutdownDoNotPanic(t *testing.T) {
+ store := newShutdownRaceStore(t)
+
+ var wg sync.WaitGroup
+ stop := make(chan struct{})
+
+ for i := 0; i < 8; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ store.WriteWithTier("vm", "shutdown-race", "cpu", 1, time.Unix(1_700_000_000, 0), TierRaw)
+ store.WriteBatchBounded([]WriteMetric{{
+ ResourceType: "vm",
+ ResourceID: "shutdown-race",
+ MetricType: "cpu",
+ Value: 1,
+ Timestamp: time.Unix(1_700_000_000, 0),
+ Tier: TierRaw,
+ }})
+ }
+ }()
+ }
+
+ time.Sleep(50 * time.Millisecond)
+ if err := store.Close(); err != nil {
+ t.Fatalf("Close returned error: %v", err)
+ }
+ close(stop)
+ wg.Wait()
+}