From e8455e84b7994797782d6f97eb69a2d1143df4f8 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Thu, 11 Jun 2026 08:33:00 +0100 Subject: [PATCH] fix(resources): pin canonical host IDs to durable identity so they survive restarts Canonical IDs for merged-source hosts (PVE node + pulse-agent) were minted from whichever identity keys the creating record happened to carry: the agent record knows the machine ID, the Proxmox node record only knows cluster+hostname. The registry rebuilds from scratch every tick, so a boot window where the agent had not checked in yet minted a cluster-keyed ID (agent-7a62... for delly) while steady state minted a machine-keyed one (agent-bdd4...). Every restart re-ran the race, fragmenting the resource_changes journal into per-boot eras (9.4k vs 6.1k rows for the same host) and silently truncating report availability and UI timelines. Fix, in the layer that owns identity: - Persist identity pins (canonical_id <-> machine_id/dmi/cluster/hostname) in the previously schema-only resource_identities table, written by the store-backed registry after monitor-adapter rebuilds, diff-aware so steady-state ticks cost no writes. - Complete weak incoming identities from the pins before matching and ID derivation, so a node-only boot window derives the same machine-keyed canonical ID as steady state. Derivation itself is unchanged; ephemeral nil-store registries behave exactly as before. - Expand change-journal reads (Get/Count families, SQLite and memory) to the full era set recomputed from the pinned identity keys, healing historical journals at query time with no row migration. Reads keyed by a stale era ID resolve to the same merged timeline. Regression tests cover both ingest orders, restart simulation via the monitor adapter, era ID derivation, and era-merged journal reads on both store implementations. Contracts updated: unified-resources obligation 25 (durable identity pins), monitoring obligation 10 (adapter rebuild persistence). --- .../v6/internal/subsystems/monitoring.md | 7 + .../internal/subsystems/unified-resources.md | 23 ++ .../unifiedresources/canonical_id_pins.go | 189 +++++++++++ internal/unifiedresources/ids.go | 80 +++++ internal/unifiedresources/ids_test.go | 47 +++ internal/unifiedresources/monitor_adapter.go | 2 + .../monitor_adapter_read_state_test.go | 66 ++++ internal/unifiedresources/registry.go | 25 +- internal/unifiedresources/registry_test.go | 73 ++++ internal/unifiedresources/store.go | 316 ++++++++++++++++-- internal/unifiedresources/store_test.go | 83 +++++ 11 files changed, 879 insertions(+), 32 deletions(-) create mode 100644 internal/unifiedresources/canonical_id_pins.go diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index dad486372..99814c6ef 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -311,6 +311,13 @@ truth for live infrastructure data. SMB/NFS shares must enter unified resources as `network-share` records with the TrueNAS share facet, not as generic storage rows or Docker/container compatibility records. +10. Keep the monitor-adapter rebuild lifecycle persisting canonical identity + pins. `internal/unifiedresources/monitor_adapter.go` calls + `PersistIdentityPins` after snapshot rebuilds and supplemental-record + ingestion so canonical host IDs stay stable across restarts (see the + unified-resources contract's durable identity-pin obligation). Rebuild + paths added to the adapter must keep that persistence step; ephemeral + snapshot-bridge adapters stay read-only. ## Current State diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 546c3f84d..005d668ba 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -949,6 +949,29 @@ AI-only summary payloads, or page-local heuristics. `verificationOutcome.status` presentation for `verified`, `unverified`, `failed`, and `unknown` so resource history, future action audit consumers, and tests share one bounded vocabulary. +25. Keep canonical host IDs stable across restarts through durable + identity pins. Canonical ID derivation hashes the strongest identity + key available at mint time, and merged-source hosts (Proxmox node + + pulse-agent) expose different identity subsets depending on which + records are present when a registry rebuilds: agent records carry the + machine ID, node records only cluster+hostname. The store-backed + registry therefore persists identity pins + (`canonical_id` ↔ machine ID / DMI UUID / cluster / hostname) in the + `resource_identities` table (`ResourceStore.UpsertResourceIdentityPins` + / `ListResourceIdentityPins`, written diff-aware after monitor-adapter + rebuilds) and completes weak incoming identities from those pins + before matching and ID derivation + (`internal/unifiedresources/canonical_id_pins.go`). Change-journal + reads (`GetRecentChanges*` / `CountRecentChanges*`) must expand a + canonical ID to the era set recomputable from its pinned identity keys + (`ResourceIdentityPin.EraIDs`) so journal rows recorded under an + earlier era's ID stay part of the resource's timeline; reads keyed by + a stale era ID resolve to the same merged timeline. New ingest paths + must not mint host canonical IDs from snapshot-content-dependent + identity subsets without consulting the pins, and pin writes stay on + the durable store-backed registry (ephemeral per-request registries + consult, never write). Regression coverage: + `internal/unifiedresources/canonical_id_pins_test.go`. ## Current State diff --git a/internal/unifiedresources/canonical_id_pins.go b/internal/unifiedresources/canonical_id_pins.go new file mode 100644 index 000000000..19ffb902f --- /dev/null +++ b/internal/unifiedresources/canonical_id_pins.go @@ -0,0 +1,189 @@ +package unifiedresources + +import ( + "log" + "strings" +) + +// identityPinIndex is the in-memory lookup over the durable identity pins. +// It answers one question at ingest time: "have we ever durably assigned a +// canonical ID to the physical host this (possibly weak) identity refers to, +// and what strong identity keys did we record for it?" +type identityPinIndex struct { + byCanonicalID map[string]ResourceIdentityPin + byMachineID map[string]ResourceIdentityPin + byDMIUUID map[string]ResourceIdentityPin + byClusterHost map[string]ResourceIdentityPin + // byHostname holds pins per normalized hostname. Hostnames are not + // unique across machines, so lookups through this map require the + // bucket to be unambiguous. + byHostname map[string][]ResourceIdentityPin +} + +func newIdentityPinIndex(pins []ResourceIdentityPin) *identityPinIndex { + index := &identityPinIndex{ + byCanonicalID: make(map[string]ResourceIdentityPin, len(pins)), + byMachineID: make(map[string]ResourceIdentityPin), + byDMIUUID: make(map[string]ResourceIdentityPin), + byClusterHost: make(map[string]ResourceIdentityPin), + byHostname: make(map[string][]ResourceIdentityPin), + } + for _, pin := range pins { + pin = pin.normalized() + if pin.CanonicalID == "" || !pin.hasStrongKey() { + continue + } + index.byCanonicalID[pin.CanonicalID] = pin + if pin.MachineID != "" { + index.byMachineID[pin.MachineID] = pin + } + if pin.DMIUUID != "" { + index.byDMIUUID[pin.DMIUUID] = pin + } + if pin.ClusterName != "" && pin.Hostname != "" { + index.byClusterHost[clusterHostPinKey(pin.ClusterName, pin.Hostname)] = pin + } + if pin.Hostname != "" { + index.byHostname[pin.Hostname] = append(index.byHostname[pin.Hostname], pin) + } + } + return index +} + +func clusterHostPinKey(clusterName, hostname string) string { + return strings.ToLower(strings.TrimSpace(clusterName)) + "\x00" + NormalizeHostname(hostname) +} + +// find resolves the pin for an incoming identity, strongest key first. A pin +// found through a weaker key is rejected when the incoming identity carries a +// strong key that contradicts the pin (a different machine claiming the same +// cluster slot or hostname must mint fresh, not absorb the old host). +func (index *identityPinIndex) find(identity ResourceIdentity) (ResourceIdentityPin, bool) { + if index == nil { + return ResourceIdentityPin{}, false + } + machineID := strings.TrimSpace(identity.MachineID) + dmiUUID := strings.TrimSpace(identity.DMIUUID) + if machineID != "" { + if pin, ok := index.byMachineID[machineID]; ok { + return pin, true + } + } + if dmiUUID != "" { + if pin, ok := index.byDMIUUID[dmiUUID]; ok && pinCompatible(pin, machineID, "") { + return pin, true + } + } + clusterName := strings.TrimSpace(identity.ClusterName) + for _, hostname := range identity.Hostnames { + normalized := NormalizeHostname(hostname) + if normalized == "" { + continue + } + if clusterName != "" { + if pin, ok := index.byClusterHost[clusterHostPinKey(clusterName, normalized)]; ok && pinCompatible(pin, machineID, dmiUUID) { + return pin, true + } + } + bucket := index.byHostname[normalized] + if len(bucket) == 1 && pinCompatible(bucket[0], machineID, dmiUUID) { + return bucket[0], true + } + } + return ResourceIdentityPin{}, false +} + +// pinCompatible reports whether an incoming identity's strong keys are +// consistent with the pin. Empty incoming keys never contradict; the pin's +// own empty keys never contradict either. +func pinCompatible(pin ResourceIdentityPin, machineID, dmiUUID string) bool { + if machineID != "" && pin.MachineID != "" && machineID != pin.MachineID { + return false + } + if dmiUUID != "" && pin.DMIUUID != "" && dmiUUID != pin.DMIUUID { + return false + } + return true +} + +func (rr *ResourceRegistry) loadIdentityPins() { + rr.identityPins = newIdentityPinIndex(nil) + if rr.store == nil { + return + } + pins, err := rr.store.ListResourceIdentityPins() + if err != nil { + log.Printf("unifiedresources: failed to load identity pins from store: %v", err) + return + } + rr.identityPins = newIdentityPinIndex(pins) +} + +// completeIdentityFromPins fills the machine-level identity keys a weak +// incoming identity lacks, from the durable pin for the same physical host. +// This runs before identity matching and canonical-ID derivation, so a boot +// window where the agent has not checked in yet (the Proxmox node record only +// knows cluster+hostname) still derives the same canonical ID as a steady +// state rebuild that knows the machine ID. Only missing fields are filled; an +// incoming identity that already carries a machine ID is never overridden. +func (rr *ResourceRegistry) completeIdentityFromPins(resourceType ResourceType, identity ResourceIdentity) ResourceIdentity { + if CanonicalResourceType(resourceType) != ResourceTypeAgent { + return identity + } + if strings.TrimSpace(identity.MachineID) != "" && strings.TrimSpace(identity.DMIUUID) != "" { + return identity + } + pin, ok := rr.identityPins.find(identity) + if !ok { + return identity + } + if strings.TrimSpace(identity.MachineID) == "" { + identity.MachineID = pin.MachineID + } + if strings.TrimSpace(identity.DMIUUID) == "" { + identity.DMIUUID = pin.DMIUUID + } + return identity +} + +// PersistIdentityPins writes the identity pins for the registry's current +// host resources into the resource store. Only new or changed pins are +// written, so steady-state rebuild ticks cost no writes. Call this after a +// rebuild on the durable store-backed registry; ephemeral per-request +// registries consult pins but do not write them. +func (rr *ResourceRegistry) PersistIdentityPins() { + if rr.store == nil { + return + } + + rr.mu.RLock() + var pins []ResourceIdentityPin + for _, resource := range rr.resources { + pin, ok := identityPinForResource(resource) + if !ok { + continue + } + if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known && existing == pin { + continue + } + pins = append(pins, pin) + } + rr.mu.RUnlock() + + if len(pins) == 0 { + return + } + if err := rr.store.UpsertResourceIdentityPins(pins); err != nil { + log.Printf("unifiedresources: failed to persist identity pins: %v", err) + return + } + refreshed, err := rr.store.ListResourceIdentityPins() + if err != nil { + log.Printf("unifiedresources: failed to reload identity pins after persist: %v", err) + return + } + index := newIdentityPinIndex(refreshed) + rr.mu.Lock() + rr.identityPins = index + rr.mu.Unlock() +} diff --git a/internal/unifiedresources/ids.go b/internal/unifiedresources/ids.go index 151e3473a..12d209630 100644 --- a/internal/unifiedresources/ids.go +++ b/internal/unifiedresources/ids.go @@ -32,3 +32,83 @@ func CanonicalResourceID(id string) string { } return trimmed } + +// ResourceIdentityPin is the durable identity→canonical-ID record persisted in +// the resource store (resource_identities table). It exists because canonical +// ID derivation keys off the strongest identity field known at mint time, and +// merged-source hosts (PVE node + pulse-agent) expose different identity +// subsets depending on which records are present when the registry rebuilds: +// the agent record knows the machine ID, the Proxmox node record only knows +// cluster+hostname. Pins let a rebuild that only sees the weak subset recover +// the strong keys, so the same physical host derives the same canonical ID in +// every boot, in every registry instance. +type ResourceIdentityPin struct { + CanonicalID string + ResourceType ResourceType + MachineID string + DMIUUID string + ClusterName string + // Hostname is the normalized primary hostname (NormalizeHostname). + Hostname string +} + +func (p ResourceIdentityPin) normalized() ResourceIdentityPin { + p.CanonicalID = CanonicalResourceID(p.CanonicalID) + p.ResourceType = CanonicalResourceType(p.ResourceType) + p.MachineID = strings.TrimSpace(p.MachineID) + p.DMIUUID = strings.TrimSpace(p.DMIUUID) + p.ClusterName = strings.TrimSpace(p.ClusterName) + p.Hostname = NormalizeHostname(p.Hostname) + return p +} + +func (p ResourceIdentityPin) hasStrongKey() bool { + return p.MachineID != "" || p.DMIUUID != "" || (p.ClusterName != "" && p.Hostname != "") +} + +// EraIDs returns every canonical ID this pin's identity keys derive under the +// historical chooseNewID ladder (machine > dmi > cluster+hostname > hostname). +// Change-journal rows written in boots that only knew a weaker key sit under +// the weaker key's hash; expanding a read to the full era set merges those +// journal eras without rewriting history. +func (p ResourceIdentityPin) EraIDs() []string { + p = p.normalized() + ids := make([]string, 0, 5) + if p.CanonicalID != "" { + ids = append(ids, p.CanonicalID) + } + if p.MachineID != "" { + ids = append(ids, buildHashID(p.ResourceType, "machine:"+p.MachineID)) + } + if p.DMIUUID != "" { + ids = append(ids, buildHashID(p.ResourceType, "dmi:"+p.DMIUUID)) + } + if p.Hostname != "" { + if p.ClusterName != "" { + ids = append(ids, buildHashID(p.ResourceType, fmt.Sprintf("cluster:%s:%s", p.ClusterName, p.Hostname))) + } + ids = append(ids, buildHashID(p.ResourceType, "hostname:"+p.Hostname)) + } + return uniqueTrimmed(ids...) +} + +// identityPinForResource derives the persistable pin for a canonical resource. +// Only host resources with at least one strong identity key are pinned; a +// hostname-only host has a single possible derivation and needs no pin. +func identityPinForResource(resource *Resource) (ResourceIdentityPin, bool) { + if resource == nil || CanonicalResourceType(resource.Type) != ResourceTypeAgent { + return ResourceIdentityPin{}, false + } + pin := ResourceIdentityPin{ + CanonicalID: resource.ID, + ResourceType: ResourceTypeAgent, + MachineID: resource.Identity.MachineID, + DMIUUID: resource.Identity.DMIUUID, + ClusterName: resource.Identity.ClusterName, + Hostname: firstIdentityHostname(resource.Identity), + }.normalized() + if pin.CanonicalID == "" || !pin.hasStrongKey() { + return ResourceIdentityPin{}, false + } + return pin, true +} diff --git a/internal/unifiedresources/ids_test.go b/internal/unifiedresources/ids_test.go index 4c1196a12..ec40e3fca 100644 --- a/internal/unifiedresources/ids_test.go +++ b/internal/unifiedresources/ids_test.go @@ -40,3 +40,50 @@ func TestSourceSpecificIDCanonicalizesSourceIDWhitespace(t *testing.T) { t.Fatalf("SourceSpecificID should trim source ID whitespace: got %q want %q", got, want) } } + +func TestResourceIdentityPinEraIDs(t *testing.T) { + pin := ResourceIdentityPin{ + CanonicalID: buildHashID(ResourceTypeAgent, "machine:machine-1"), + ResourceType: ResourceTypeAgent, + MachineID: "machine-1", + DMIUUID: "dmi-1", + ClusterName: "homelab", + Hostname: "delly.lan", + } + + got := pin.EraIDs() + want := []string{ + buildHashID(ResourceTypeAgent, "machine:machine-1"), + buildHashID(ResourceTypeAgent, "dmi:dmi-1"), + buildHashID(ResourceTypeAgent, "cluster:homelab:delly"), + buildHashID(ResourceTypeAgent, "hostname:delly"), + } + if len(got) != len(want) { + t.Fatalf("expected %d era IDs, got %d: %v", len(want), len(got), got) + } + for _, id := range want { + found := false + for _, eraID := range got { + if eraID == id { + found = true + break + } + } + if !found { + t.Fatalf("expected era set to include %q, got %v", id, got) + } + } +} + +func TestResourceIdentityPinEraIDsSkipsWeakOnlyKeys(t *testing.T) { + pin := ResourceIdentityPin{ + CanonicalID: "agent-custom", + ResourceType: ResourceTypeAgent, + Hostname: "delly", + } + got := pin.EraIDs() + want := []string{"agent-custom", buildHashID(ResourceTypeAgent, "hostname:delly")} + if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Fatalf("expected era IDs %v, got %v", want, got) + } +} diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index e4e62d36d..1037bfda0 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -114,6 +114,7 @@ func (a *MonitorAdapter) replaceRegistry(snapshot models.StateSnapshot, recordsB occurredAt = &occ } recordRegistryChanges(registry.store, before, rebuilt.List(), rebuiltAt, occurredAt, SourcePulseDiff, "") + rebuilt.PersistIdentityPins() a.mu.Lock() a.registry = rebuilt @@ -294,6 +295,7 @@ func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records registry.IngestRecords(source, records) rebuiltAt := time.Now().UTC() recordRegistryChanges(registry.store, before, registry.List(), rebuiltAt, nil, SourcePlatformEvent, changeSourceAdapterForDataSource(source)) + registry.PersistIdentityPins() a.mu.Lock() a.lastRebuiltAt = rebuiltAt a.mu.Unlock() diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go index 96d7e7e47..cf59f44e1 100644 --- a/internal/unifiedresources/monitor_adapter_read_state_test.go +++ b/internal/unifiedresources/monitor_adapter_read_state_test.go @@ -336,3 +336,69 @@ func TestMonitorAdapterGetRecentChangesForwardsToStore(t *testing.T) { t.Fatalf("Kind = %q, want %q", changes[0].Kind, ChangeCommandExecuted) } } + +// The monitor adapter is the durable store-backed registry owner, so its +// rebuild paths must persist canonical identity pins: a later boot window +// (agent not yet checked in) relies on them to derive the same canonical host +// ID it used in steady state. +func TestMonitorAdapterRebuildPersistsIdentityPins(t *testing.T) { + store := NewMemoryStore() + adapter := NewMonitorAdapter(NewRegistry(store)) + + const machineID = "7d465a78-test-machine-id" + adapter.PopulateFromSnapshot(models.StateSnapshot{ + Nodes: []models.Node{{ + ID: "homelab-delly", + Name: "delly", + Instance: "homelab", + ClusterName: "homelab", + Status: "online", + }}, + Hosts: []models.Host{{ + ID: machineID, + MachineID: machineID, + Hostname: "delly", + LinkedNodeID: "homelab-delly", + }}, + LastUpdate: time.Now().UTC(), + }) + + pins, err := store.ListResourceIdentityPins() + if err != nil { + t.Fatalf("list identity pins: %v", err) + } + steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID) + found := false + for _, pin := range pins { + if pin.CanonicalID == steadyID && pin.MachineID == machineID { + found = true + break + } + } + if !found { + t.Fatalf("expected snapshot rebuild to persist a machine-keyed identity pin, got %+v", pins) + } + + // A restart boot window rebuilds from a snapshot that does not contain + // the agent host yet; the pinned identity must keep the canonical ID. + adapter.PopulateFromSnapshot(models.StateSnapshot{ + Nodes: []models.Node{{ + ID: "homelab-delly", + Name: "delly", + Instance: "homelab", + ClusterName: "homelab", + Status: "online", + }}, + LastUpdate: time.Now().UTC(), + }) + + for _, resource := range adapter.GetAll() { + if resource.Proxmox != nil && resource.Proxmox.NodeName == "delly" { + if resource.ID != steadyID { + t.Fatalf("boot-window rebuild minted %q, want pinned %q", resource.ID, steadyID) + } + return + } + } + t.Fatalf("expected delly node resource in boot-window rebuild") +} diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index f67a39f0d..058c05091 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -40,14 +40,15 @@ type IngestRecord struct { // ResourceRegistry merges resources from multiple sources. type ResourceRegistry struct { - mu sync.RWMutex - resources map[string]*Resource - bySource map[DataSource]map[string]string - matcher *IdentityMatcher - store ResourceStore - links []ResourceLink - exclusions map[string]struct{} - pbsBackups []models.PBSBackup + mu sync.RWMutex + resources map[string]*Resource + bySource map[DataSource]map[string]string + matcher *IdentityMatcher + store ResourceStore + links []ResourceLink + exclusions map[string]struct{} + identityPins *identityPinIndex + pbsBackups []models.PBSBackup // Cached typed view indexes. Invalidated on ingest, rebuilt lazily on // first access. Protected by mu — callers hold RLock to read, and the @@ -92,6 +93,7 @@ func NewRegistry(store ResourceStore) *ResourceRegistry { rr.bySource[SourceAvailability] = make(map[string]string) rr.loadOverrides() + rr.loadIdentityPins() return rr } @@ -2187,8 +2189,13 @@ func (rr *ResourceRegistry) ingest(source DataSource, sourceID string, resource rr.bySource[source] = make(map[string]string) } - resource.Identity = identity resource.Type = CanonicalResourceType(resource.Type) + // Complete weak host identities from the durable pins before matching and + // ID derivation, so canonical IDs do not depend on which sources happen to + // be present in this rebuild (boot windows ingest the Proxmox node record + // before the agent has checked in). + identity = rr.completeIdentityFromPins(resource.Type, identity) + resource.Identity = identity resource.Sources = []DataSource{source} resource.SourceStatus = map[DataSource]SourceStatus{ source: {Status: "online", LastSeen: resource.LastSeen}, diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index e43c74826..ed4634584 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -4378,3 +4378,76 @@ func TestMergeMetric_StaleSourceDoesNotClobberLive(t *testing.T) { t.Fatalf("expected live proxmox CPU to survive stale agent merge, got %+v", got) } } + +// The merged-source host shape from the homelab canonical-ID bug: a PVE node +// record that only knows cluster+hostname, and a pulse-agent record that +// knows the machine ID. Whichever record mints the canonical resource decides +// which identity key gets hashed, so without durable identity pins the +// canonical ID flips between boot windows (node-only) and steady state +// (agent present), fragmenting the change journal into per-boot eras. +func mergedHostNodeResource() Resource { + return Resource{ + Type: ResourceTypeAgent, + Name: "delly", + Status: StatusOnline, + Proxmox: &ProxmoxData{SourceID: "homelab-delly", NodeName: "delly", ClusterName: "homelab"}, + } +} + +func mergedHostAgentResource(machineID string) Resource { + return Resource{ + Type: ResourceTypeAgent, + Name: "delly", + Status: StatusOnline, + Agent: &AgentData{AgentID: machineID, Hostname: "delly", MachineID: machineID}, + } +} + +func TestMergedHostCanonicalIDStableAcrossRestartsAndIngestOrders(t *testing.T) { + store := NewMemoryStore() + + const machineID = "7d465a78-test-machine-id" + nodeIdentity := ResourceIdentity{Hostnames: []string{"delly"}, ClusterName: "homelab"} + agentIdentity := ResourceIdentity{MachineID: machineID, Hostnames: []string{"delly"}} + // IngestSnapshot merges the linked host's identity into the node record + // when the agent is present in the snapshot. + steadyIdentity := mergeIdentity(nodeIdentity, agentIdentity) + + steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID) + bootEraID := buildHashID(ResourceTypeAgent, "cluster:homelab:delly") + if steadyID == bootEraID { + t.Fatalf("test setup broken: era IDs must differ") + } + + // Steady-state boot: node record carries the merged identity and mints + // the machine-keyed canonical ID; the agent record merges into it. + steady := NewRegistry(store) + if id := steady.ingest(SourceProxmox, "homelab-delly", mergedHostNodeResource(), steadyIdentity); id != steadyID { + t.Fatalf("steady-state node ingest minted %q, want machine-keyed %q", id, steadyID) + } + if id := steady.ingest(SourceAgent, machineID, mergedHostAgentResource(machineID), agentIdentity); id != steadyID { + t.Fatalf("steady-state agent ingest resolved %q, want %q", id, steadyID) + } + steady.PersistIdentityPins() + + // Restart into a boot window: the agent has not checked in yet, so the + // node record only knows cluster+hostname. Before identity pins this + // minted bootEraID and the change journal fragmented into a second era. + bootWindow := NewRegistry(store) + if id := bootWindow.ingest(SourceProxmox, "homelab-delly", mergedHostNodeResource(), nodeIdentity); id != steadyID { + t.Fatalf("boot-window node ingest minted %q, want pinned %q", id, steadyID) + } + + // Restart with the opposite ingest order: agent record first into an + // empty registry, node record after. + reversed := NewRegistry(store) + if id := reversed.ingest(SourceAgent, machineID, mergedHostAgentResource(machineID), agentIdentity); id != steadyID { + t.Fatalf("agent-first ingest minted %q, want %q", id, steadyID) + } + if id := reversed.ingest(SourceProxmox, "homelab-delly", mergedHostNodeResource(), nodeIdentity); id != steadyID { + t.Fatalf("node-after-agent ingest resolved %q, want %q", id, steadyID) + } + if got := len(reversed.List()); got != 1 { + t.Fatalf("expected one merged resource after reversed-order ingest, got %d", got) + } +} diff --git a/internal/unifiedresources/store.go b/internal/unifiedresources/store.go index eb140d72a..deec107cd 100644 --- a/internal/unifiedresources/store.go +++ b/internal/unifiedresources/store.go @@ -24,6 +24,10 @@ type ResourceStore interface { AddExclusion(exclusion ResourceExclusion) error GetLinks() ([]ResourceLink, error) GetExclusions() ([]ResourceExclusion, error) + // Identity pins keep canonical IDs for merged-source hosts stable across + // restarts. See ResourceIdentityPin. + UpsertResourceIdentityPins(pins []ResourceIdentityPin) error + ListResourceIdentityPins() ([]ResourceIdentityPin, error) RecordChange(change ResourceChange) error GetRecentChanges(canonicalID string, since time.Time, limit int) ([]ResourceChange, error) GetRecentChangesFiltered(canonicalID string, since time.Time, limit int, filters ResourceChangeFilters) ([]ResourceChange, error) @@ -95,6 +99,12 @@ type SQLiteResourceStore struct { resourceChangesHasSource bool resourceChangesObservedAtNeedsFallback bool mu sync.Mutex + + // identityPinCache caches the (tiny) resource_identities table for + // change-journal era expansion. Invalidated on pin upserts. + identityPinMu sync.Mutex + identityPinCache []ResourceIdentityPin + identityPinFresh bool } const ( @@ -432,6 +442,9 @@ func (s *SQLiteResourceStore) initSchema() error { if err := s.ensureResourceChangesIndexes(); err != nil { return err } + if err := s.migrateResourceIdentitiesSchema(); err != nil { + return err + } if err := s.migrateActionAuditsSchema(); err != nil { return err } @@ -441,6 +454,33 @@ func (s *SQLiteResourceStore) initSchema() error { return nil } +// migrateResourceIdentitiesSchema upgrades the resource_identities table from +// its original schema-only shape (no code ever wrote it before identity pins) +// to the pin layout: a cluster_name column and a unique canonical_id index so +// pins upsert one row per canonical resource. +func (s *SQLiteResourceStore) migrateResourceIdentitiesSchema() error { + columns, err := s.tableColumns("resource_identities") + if err != nil { + return err + } + if _, ok := columns["cluster_name"]; !ok { + if _, err := s.db.Exec("ALTER TABLE resource_identities ADD COLUMN cluster_name TEXT"); err != nil { + return fmt.Errorf("add resource_identities.cluster_name column: %w", err) + } + } + // The table shipped empty for every deployment, but stay defensive: drop + // duplicate canonical_id rows before enforcing uniqueness. + if _, err := s.db.Exec(`DELETE FROM resource_identities WHERE id NOT IN ( + SELECT MAX(id) FROM resource_identities GROUP BY canonical_id + )`); err != nil { + return fmt.Errorf("dedupe resource_identities canonical rows: %w", err) + } + if _, err := s.db.Exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_resource_identities_canonical_unique ON resource_identities(canonical_id)"); err != nil { + return fmt.Errorf("ensure resource_identities canonical unique index: %w", err) + } + return nil +} + // migrateActionAuditsSchema adds the verification_outcome_json column to // older action_audits tables so the VerificationOutcome field on // ActionAuditRecord persists across restarts. Records written before the @@ -846,6 +886,151 @@ func (s *SQLiteResourceStore) Close() error { return nil } +func (s *SQLiteResourceStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error { + if len(pins) == 0 { + return nil + } + + s.mu.Lock() + defer s.mu.Unlock() + + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("begin identity pin upsert: %w", err) + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + + for _, pin := range pins { + pin = pin.normalized() + if pin.CanonicalID == "" || !pin.hasStrongKey() { + continue + } + // A strong identity key can only belong to one canonical resource. + // When a key moves (host reinstalled, cluster slot re-occupied by a + // different machine), the new pin claims it and stale rows go. + if _, err := tx.Exec(`DELETE FROM resource_identities WHERE canonical_id != ? AND ( + (machine_id IS NOT NULL AND machine_id = ?) + OR (dmi_uuid IS NOT NULL AND dmi_uuid = ?) + OR (cluster_name IS NOT NULL AND cluster_name != '' AND cluster_name = ? AND primary_hostname = ?) + )`, + pin.CanonicalID, + nullIfEmptyArg(pin.MachineID), + nullIfEmptyArg(pin.DMIUUID), + pin.ClusterName, + pin.Hostname, + ); err != nil { + return fmt.Errorf("clear conflicting identity pins for %q: %w", pin.CanonicalID, err) + } + if _, err := tx.Exec(`INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(canonical_id) DO UPDATE SET + resource_type = excluded.resource_type, + machine_id = COALESCE(excluded.machine_id, resource_identities.machine_id), + dmi_uuid = COALESCE(excluded.dmi_uuid, resource_identities.dmi_uuid), + cluster_name = COALESCE(NULLIF(excluded.cluster_name, ''), resource_identities.cluster_name), + primary_hostname = COALESCE(NULLIF(excluded.primary_hostname, ''), resource_identities.primary_hostname), + updated_at = CURRENT_TIMESTAMP`, + pin.CanonicalID, + string(pin.ResourceType), + nullIfEmptyArg(pin.MachineID), + nullIfEmptyArg(pin.DMIUUID), + pin.ClusterName, + pin.Hostname, + ); err != nil { + return fmt.Errorf("upsert identity pin for %q: %w", pin.CanonicalID, err) + } + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit identity pin upsert: %w", err) + } + committed = true + + s.identityPinMu.Lock() + s.identityPinFresh = false + s.identityPinMu.Unlock() + return nil +} + +func nullIfEmptyArg(value string) any { + if strings.TrimSpace(value) == "" { + return nil + } + return strings.TrimSpace(value) +} + +func (s *SQLiteResourceStore) ListResourceIdentityPins() ([]ResourceIdentityPin, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.queryResourceIdentityPins() +} + +func (s *SQLiteResourceStore) queryResourceIdentityPins() ([]ResourceIdentityPin, error) { + rows, err := s.db.Query(`SELECT canonical_id, resource_type, COALESCE(machine_id, ''), COALESCE(dmi_uuid, ''), COALESCE(cluster_name, ''), COALESCE(primary_hostname, '') FROM resource_identities`) + if err != nil { + return nil, fmt.Errorf("query identity pins: %w", err) + } + defer rows.Close() + + var pins []ResourceIdentityPin + for rows.Next() { + var pin ResourceIdentityPin + var resourceType string + if err := rows.Scan(&pin.CanonicalID, &resourceType, &pin.MachineID, &pin.DMIUUID, &pin.ClusterName, &pin.Hostname); err != nil { + return nil, fmt.Errorf("scan identity pin row: %w", err) + } + pin.ResourceType = ResourceType(resourceType) + pins = append(pins, pin.normalized()) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate identity pin rows: %w", err) + } + return pins, nil +} + +// resourceChangeIDSet expands a canonical ID to the set of IDs its journal +// rows may have been recorded under. Canonical IDs minted before the host's +// strongest identity key was known hash a weaker key (cluster+hostname during +// a boot window vs machine ID in steady state); the pinned identity makes +// every era's ID recomputable, so reads merge the eras without rewriting +// history. Unknown IDs expand to themselves. +func (s *SQLiteResourceStore) resourceChangeIDSet(canonicalID string) []string { + canonicalID = CanonicalResourceID(canonicalID) + if canonicalID == "" { + return nil + } + + s.identityPinMu.Lock() + if !s.identityPinFresh { + pins, err := s.queryResourceIdentityPins() + if err == nil { + s.identityPinCache = pins + s.identityPinFresh = true + } + } + pins := s.identityPinCache + s.identityPinMu.Unlock() + + return expandResourceChangeIDs(canonicalID, pins) +} + +func expandResourceChangeIDs(canonicalID string, pins []ResourceIdentityPin) []string { + for _, pin := range pins { + eraIDs := pin.EraIDs() + for _, eraID := range eraIDs { + if eraID == canonicalID { + return eraIDs + } + } + } + return []string{canonicalID} +} + func (s *SQLiteResourceStore) RecordChange(change ResourceChange) error { s.mu.Lock() defer s.mu.Unlock() @@ -921,7 +1106,7 @@ func (s *SQLiteResourceStore) GetRecentChangesFiltered(canonicalID string, since conditions := []string{} canonicalID = CanonicalResourceID(canonicalID) if canonicalID != "" { - conditions, args = appendRecentChangeResourceCondition(conditions, args, canonicalID, filters.IncludeRelated) + conditions, args = appendRecentChangeResourceCondition(conditions, args, s.resourceChangeIDSet(canonicalID), filters.IncludeRelated) } else { conditions = append(conditions, observedAtExpr+" >= ?") args = append(args, since) @@ -1047,7 +1232,7 @@ func (s *SQLiteResourceStore) CountRecentChanges(canonicalID string, since time. func (s *SQLiteResourceStore) CountRecentChangesFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (int, error) { query, args := buildRecentChangeCountQuery( - canonicalID, + s.resourceChangeIDSet(canonicalID), since, filters, "SELECT COUNT(*) FROM resource_changes", @@ -1072,7 +1257,7 @@ func (s *SQLiteResourceStore) CountRecentChangesByKind(canonicalID string, since func (s *SQLiteResourceStore) CountRecentChangesByKindFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeKind]int, error) { query, args := buildRecentChangeCountQuery( - canonicalID, + s.resourceChangeIDSet(canonicalID), since, filters, "SELECT COALESCE(kind, ''), COUNT(*) FROM resource_changes", @@ -1116,7 +1301,7 @@ func (s *SQLiteResourceStore) CountRecentChangesBySourceType(canonicalID string, func (s *SQLiteResourceStore) CountRecentChangesBySourceTypeFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeSourceType]int, error) { sourceTypeExpr := s.resourceChangesSourceTypeExpr() query, args := buildRecentChangeCountQuery( - canonicalID, + s.resourceChangeIDSet(canonicalID), since, filters, "SELECT "+sourceTypeExpr+", COUNT(*) FROM resource_changes", @@ -1160,7 +1345,7 @@ func (s *SQLiteResourceStore) CountRecentChangesBySourceAdapter(canonicalID stri func (s *SQLiteResourceStore) CountRecentChangesBySourceAdapterFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeSourceAdapter]int, error) { sourceAdapterExpr := s.resourceChangesSourceAdapterExpr() query, args := buildRecentChangeCountQuery( - canonicalID, + s.resourceChangeIDSet(canonicalID), since, filters, "SELECT "+sourceAdapterExpr+", COUNT(*) FROM resource_changes", @@ -2005,15 +2190,78 @@ type MemoryStore struct { exportAudits []ExportAuditRecord resourceOperatorState map[string]ResourceOperatorState loopReports map[string]LoopReport + identityPins map[string]ResourceIdentityPin } func NewMemoryStore() *MemoryStore { return &MemoryStore{ resourceOperatorState: make(map[string]ResourceOperatorState), loopReports: make(map[string]LoopReport), + identityPins: make(map[string]ResourceIdentityPin), } } +func (m *MemoryStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error { + m.mu.Lock() + defer m.mu.Unlock() + for _, pin := range pins { + pin = pin.normalized() + if pin.CanonicalID == "" || !pin.hasStrongKey() { + continue + } + // Mirror the SQLite conflict semantics: a strong identity key belongs + // to exactly one canonical resource. + for canonicalID, existing := range m.identityPins { + if canonicalID == pin.CanonicalID { + continue + } + if (pin.MachineID != "" && existing.MachineID == pin.MachineID) || + (pin.DMIUUID != "" && existing.DMIUUID == pin.DMIUUID) || + (pin.ClusterName != "" && existing.ClusterName == pin.ClusterName && existing.Hostname == pin.Hostname) { + delete(m.identityPins, canonicalID) + } + } + if existing, ok := m.identityPins[pin.CanonicalID]; ok { + if pin.MachineID == "" { + pin.MachineID = existing.MachineID + } + if pin.DMIUUID == "" { + pin.DMIUUID = existing.DMIUUID + } + if pin.ClusterName == "" { + pin.ClusterName = existing.ClusterName + } + if pin.Hostname == "" { + pin.Hostname = existing.Hostname + } + } + m.identityPins[pin.CanonicalID] = pin + } + return nil +} + +func (m *MemoryStore) ListResourceIdentityPins() ([]ResourceIdentityPin, error) { + m.mu.RLock() + defer m.mu.RUnlock() + pins := make([]ResourceIdentityPin, 0, len(m.identityPins)) + for _, pin := range m.identityPins { + pins = append(pins, pin) + } + return pins, nil +} + +func (m *MemoryStore) resourceChangeIDSetLocked(canonicalID string) []string { + canonicalID = CanonicalResourceID(canonicalID) + if canonicalID == "" { + return nil + } + pins := make([]ResourceIdentityPin, 0, len(m.identityPins)) + for _, pin := range m.identityPins { + pins = append(pins, pin) + } + return expandResourceChangeIDs(canonicalID, pins) +} + func (m *MemoryStore) AddLink(link ResourceLink) error { m.mu.Lock() defer m.mu.Unlock() @@ -2078,10 +2326,11 @@ func (m *MemoryStore) GetRecentChangesFiltered(canonicalID string, since time.Ti m.mu.RLock() defer m.mu.RUnlock() canonicalID = CanonicalResourceID(canonicalID) + idSet := m.resourceChangeIDSetLocked(canonicalID) var out []ResourceChange for i := len(m.changes) - 1; i >= 0; i-- { change := m.changes[i] - if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) { + if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) { continue } if !since.IsZero() && change.ObservedAt.Before(since) { @@ -2106,9 +2355,10 @@ func (m *MemoryStore) CountRecentChangesFiltered(canonicalID string, since time. m.mu.RLock() defer m.mu.RUnlock() canonicalID = CanonicalResourceID(canonicalID) + idSet := m.resourceChangeIDSetLocked(canonicalID) count := 0 for _, change := range m.changes { - if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) { + if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) { continue } if !since.IsZero() && change.ObservedAt.Before(since) { @@ -2130,9 +2380,10 @@ func (m *MemoryStore) CountRecentChangesByKindFiltered(canonicalID string, since m.mu.RLock() defer m.mu.RUnlock() canonicalID = CanonicalResourceID(canonicalID) + idSet := m.resourceChangeIDSetLocked(canonicalID) counts := make(map[ChangeKind]int) for _, change := range m.changes { - if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) { + if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) { continue } if !since.IsZero() && change.ObservedAt.Before(since) { @@ -2157,9 +2408,10 @@ func (m *MemoryStore) CountRecentChangesBySourceTypeFiltered(canonicalID string, m.mu.RLock() defer m.mu.RUnlock() canonicalID = CanonicalResourceID(canonicalID) + idSet := m.resourceChangeIDSetLocked(canonicalID) counts := make(map[ChangeSourceType]int) for _, change := range m.changes { - if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) { + if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) { continue } if !since.IsZero() && change.ObservedAt.Before(since) { @@ -2184,9 +2436,10 @@ func (m *MemoryStore) CountRecentChangesBySourceAdapterFiltered(canonicalID stri m.mu.RLock() defer m.mu.RUnlock() canonicalID = CanonicalResourceID(canonicalID) + idSet := m.resourceChangeIDSetLocked(canonicalID) counts := make(map[ChangeSourceAdapter]int) for _, change := range m.changes { - if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) { + if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) { continue } if !since.IsZero() && change.ObservedAt.Before(since) { @@ -2203,14 +2456,13 @@ func (m *MemoryStore) CountRecentChangesBySourceAdapterFiltered(canonicalID stri return counts, nil } -func buildRecentChangeCountQuery(canonicalID string, since time.Time, filters ResourceChangeFilters, selectClause string, observedAtExpr string, sourceTypeExpr string, sourceAdapterExpr string) (string, []any) { +func buildRecentChangeCountQuery(canonicalIDs []string, since time.Time, filters ResourceChangeFilters, selectClause string, observedAtExpr string, sourceTypeExpr string, sourceAdapterExpr string) (string, []any) { query := selectClause args := []any{} conditions := []string{observedAtExpr + " >= ?"} args = append(args, since) - canonicalID = CanonicalResourceID(canonicalID) - if canonicalID != "" { - conditions, args = appendRecentChangeResourceCondition(conditions, args, canonicalID, filters.IncludeRelated) + if len(canonicalIDs) > 0 { + conditions, args = appendRecentChangeResourceCondition(conditions, args, canonicalIDs, filters.IncludeRelated) } if len(filters.Kinds) > 0 { placeholders := make([]string, 0, len(filters.Kinds)) @@ -2240,26 +2492,44 @@ func buildRecentChangeCountQuery(canonicalID string, since time.Time, filters Re return query, args } -func appendRecentChangeResourceCondition(conditions []string, args []any, canonicalID string, includeRelated bool) ([]string, []any) { - if !includeRelated { - return append(conditions, "canonical_id = ?"), append(args, canonicalID) +func appendRecentChangeResourceCondition(conditions []string, args []any, canonicalIDs []string, includeRelated bool) ([]string, []any) { + placeholders := make([]string, len(canonicalIDs)) + for i, id := range canonicalIDs { + placeholders[i] = "?" + args = append(args, id) } - return append(conditions, `(canonical_id = ? OR EXISTS ( + inClause := "(" + strings.Join(placeholders, ", ") + ")" + if !includeRelated { + return append(conditions, "canonical_id IN "+inClause), args + } + for _, id := range canonicalIDs { + args = append(args, id) + } + return append(conditions, `(canonical_id IN `+inClause+` OR EXISTS ( SELECT 1 FROM json_each(CASE WHEN json_valid(resource_changes.related_resources) THEN resource_changes.related_resources ELSE '[]' END) - WHERE TRIM(json_each.value) = ? - ))`), append(args, canonicalID, canonicalID) + WHERE TRIM(json_each.value) IN `+inClause+` + ))`), args } -func changeMatchesResource(change ResourceChange, canonicalID string, includeRelated bool) bool { - if CanonicalResourceID(change.ResourceID) == canonicalID { +func changeMatchesResource(change ResourceChange, canonicalIDs []string, includeRelated bool) bool { + matches := func(id string) bool { + id = CanonicalResourceID(id) + for _, canonicalID := range canonicalIDs { + if id == canonicalID { + return true + } + } + return false + } + if matches(change.ResourceID) { return true } if !includeRelated { return false } for _, relatedID := range change.RelatedResources { - if CanonicalResourceID(relatedID) == canonicalID { + if matches(relatedID) { return true } } diff --git a/internal/unifiedresources/store_test.go b/internal/unifiedresources/store_test.go index e7762e709..e103d39ce 100644 --- a/internal/unifiedresources/store_test.go +++ b/internal/unifiedresources/store_test.go @@ -2308,3 +2308,86 @@ func TestSQLiteResourceStore_ResourceOperatorState_RoundTrips(t *testing.T) { t.Errorf("clear must be idempotent; got %v", err) } } + +func TestResourceChangeReadsMergeCanonicalIDEras(t *testing.T) { + const machineID = "7d465a78-test-machine-id" + steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID) + bootEraID := buildHashID(ResourceTypeAgent, "cluster:homelab:delly") + + sqliteStore, err := NewSQLiteResourceStore(t.TempDir(), "") + if err != nil { + t.Fatalf("open sqlite store: %v", err) + } + defer sqliteStore.Close() + + stores := map[string]ResourceStore{ + "memory": NewMemoryStore(), + "sqlite": sqliteStore, + } + + for name, store := range stores { + t.Run(name, func(t *testing.T) { + if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{ + CanonicalID: steadyID, + ResourceType: ResourceTypeAgent, + MachineID: machineID, + ClusterName: "homelab", + Hostname: "delly", + }}); err != nil { + t.Fatalf("upsert pin: %v", err) + } + + base := time.Date(2026, 6, 1, 12, 0, 0, 0, time.UTC) + record := func(id, resourceID string, at time.Time) { + t.Helper() + if err := store.RecordChange(ResourceChange{ + ID: id, + ResourceID: resourceID, + ObservedAt: at, + Kind: ChangeStateTransition, + SourceType: SourcePulseDiff, + Confidence: ConfidenceHigh, + }); err != nil { + t.Fatalf("record change %s: %v", id, err) + } + } + record("change-boot-era", bootEraID, base) + record("change-steady-era", steadyID, base.Add(time.Hour)) + record("change-unrelated", "agent-unrelated", base.Add(2*time.Hour)) + + changes, err := store.GetRecentChanges(steadyID, time.Time{}, 0) + if err != nil { + t.Fatalf("get changes by steady ID: %v", err) + } + if len(changes) != 2 { + t.Fatalf("expected both journal eras under the steady ID, got %d changes", len(changes)) + } + + // Reads keyed by the old era ID resolve through the pin too, so + // callers holding a stale canonical ID see the full timeline. + changes, err = store.GetRecentChanges(bootEraID, time.Time{}, 0) + if err != nil { + t.Fatalf("get changes by boot-era ID: %v", err) + } + if len(changes) != 2 { + t.Fatalf("expected both journal eras under the boot-era ID, got %d changes", len(changes)) + } + + count, err := store.CountRecentChanges(steadyID, time.Time{}) + if err != nil { + t.Fatalf("count changes: %v", err) + } + if count != 2 { + t.Fatalf("expected era-merged count 2, got %d", count) + } + + unrelated, err := store.GetRecentChanges("agent-unrelated", time.Time{}, 0) + if err != nil { + t.Fatalf("get unrelated changes: %v", err) + } + if len(unrelated) != 1 { + t.Fatalf("expected unrelated resource to keep its own timeline, got %d changes", len(unrelated)) + } + }) + } +}