Merge reviewed candidate 20260919T140941Z-core-runtime (pulse ee7ecbd928)

Integrate the four user-facing core repairs: drop replayed unified metric samples (#1966), treat all local RepoDigests as the current image (#2110), honour the removal block on forked re-enroll (#2113), and stop closing writeCh on metrics-store shutdown (#2079 finding 1), with focused regressions and contract sections.

Change-source: pulse-maintainer

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 `<base>-<hex>` 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`.

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 `<base>-<hex>` 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`.

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.

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)
+	}
+}

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{

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 {

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 == "" {

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)
+	}
+}

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) {

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")
+	}
+}

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 {

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)
 	}

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

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())
+	}
+}

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 {

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()
+}