refactor: drop resource cache staleness workarounds

.github/workflows/ci.yaml

@@ -1,6 +1,6 @@
 # THIS FILE WAS AUTOMATICALLY GENERATED BY KRES, PLEASE DO NOT EDIT.
 #
-# Generated on 2026-07-15T13:55:49Z by kres d66531d.
+# Generated on 2026-08-03T16:48:09Z by kres 93d28ca.
 
 concurrency:
   group: ${{ github.head_ref || github.run_id }}

go.mod

@@ -320,3 +320,5 @@ require (
 	sigs.k8s.io/structured-merge-diff/v6 v6.4.0 // indirect
 	sigs.k8s.io/yaml v1.6.0 // indirect
 )
+
+replace github.com/cosi-project/state-etcd => github.com/smira/state-etcd v0.0.0-20260803154600-abed9e9567f0

go.sum

@@ -121,8 +121,6 @@ github.com/coreos/go-systemd/v22 v22.7.0 h1:LAEzFkke61DFROc7zNLX/WA2i5J8gYqe0rSj
 github.com/coreos/go-systemd/v22 v22.7.0/go.mod h1:xNUYtjHu2EDXbsxz1i41wouACIwT7Ybq9o0BQhMwD0w=
 github.com/cosi-project/runtime v1.16.2 h1:sgjOMrnaElrja93kkJJJIQYwd45PhpGl9cBci9yqn5c=
 github.com/cosi-project/runtime v1.16.2/go.mod h1:+GrSnmJjMfWMe6NubevwwXQf/v7afddDLeCbLonvvps=
-github.com/cosi-project/state-etcd v0.7.0 h1:7EYGKbGClLA0h910oMWxtw4NZh9qt6GebaGxqGm7zts=
-github.com/cosi-project/state-etcd v0.7.0/go.mod h1:hplOOGHR1Rt/T8fiuFcP16thoQDvprqP/aateuCv36o=
 github.com/cosi-project/state-sqlite v0.4.0 h1:SwL5/LlAwnMomRjMM72igPA1F6W/88zbae4cFxqz5vw=
 github.com/cosi-project/state-sqlite v0.4.0/go.mod h1:V20oy2Sfxla0zZ+SJSgjV20feg2xGARlvVPL4Z4KfRo=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
@@ -490,6 +488,8 @@ github.com/siderolabs/tcpproxy v0.1.0/go.mod h1:onn6CPPj/w1UNqQ0U97oRPF0CqbrgEAp
 github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
 github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=
+github.com/smira/state-etcd v0.0.0-20260803154600-abed9e9567f0 h1:6fUPxDRfNOymYwKosfYAj/6ZWbMbrO60COp4iQ9X1/A=
+github.com/smira/state-etcd v0.0.0-20260803154600-abed9e9567f0/go.mod h1:+I+tyOd7uBofnxloBn66Qc3P+oleRgD6qOeO+rkF/pU=
 github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
 github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
 github.com/sosodev/duration v1.4.0 h1:35ed0KiVFriGHHzZZJaZLgmTEEICIyt8Sx0RQfj9IjE=

internal/backend/runtime/omni/controllers/cleanup/id.go

@@ -33,9 +33,7 @@ func (h *SameIDHandler[I, O]) FinalizerRemoval(ctx context.Context, r controller
 		resource.VersionUndefined,
 	)
 
-	// Use GetUncached to bypass the controller runtime cache. A cached miss here would cause the handler to skip cleanup permanently (the finalizer gets removed
-	// and the orphaned resource is never retried).
-	res, err := r.GetUncached(ctx, md)
+	res, err := r.Get(ctx, md)
 	if err != nil {
 		if state.IsNotFoundError(err) {
 			return nil
@@ -101,9 +99,7 @@ func IDHandleFunc[I, O generic.ResourceWithRD](getIDFunc GetIDFunc[I], blockIfOw
 			resource.VersionUndefined,
 		)
 
-		// Use GetUncached to bypass the controller runtime cache. A cached miss here would cause the handler to skip cleanup permanently (the finalizer gets removed
-		// and the orphaned resource is never retried).
-		res, err := r.GetUncached(ctx, md)
+		res, err := r.Get(ctx, md)
 		if err != nil {
 			if state.IsNotFoundError(err) {
 				return nil

internal/backend/runtime/omni/controllers/omni/machine_set_node.go

@@ -223,7 +223,7 @@ func (ctrl *MachineSetNodeController) MapInput(
 }
 
 func (ctrl *MachineSetNodeController) getAllMachineSetNodes(ctx context.Context, r controller.QRuntime, opts ...state.ListOption) (safe.List[*omni.MachineSetNode], error) {
-	items, err := r.ListUncached(
+	items, err := r.List(
 		ctx, resource.NewMetadata(resources.DefaultNamespace, omni.MachineSetNodeType, "", resource.VersionUndefined),
 		opts...,
 	)

internal/backend/runtime/omni/controllers/omni/machineconfig/reconciliation_context.go

@@ -32,7 +32,6 @@ import (
 	"github.com/siderolabs/omni/client/pkg/omni/resources/infra"
 	"github.com/siderolabs/omni/client/pkg/omni/resources/omni"
 	siderolinkres "github.com/siderolabs/omni/client/pkg/omni/resources/siderolink"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 	"github.com/siderolabs/omni/internal/backend/talos/lifecycle"
 )
 
@@ -267,8 +266,7 @@ func BuildReconciliationContext(ctx context.Context, r controller.Reader,
 		return nil, errors.New("failed to get machine set name from the machine config resource")
 	}
 
-	// A stale allow value could apply a config while updates are blocked.
-	machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, uncached.Reader(r), machineSetName)
+	machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, r, machineSetName)
 	if err != nil && !state.IsNotFoundError(err) {
 		return nil, err
 	}

internal/backend/runtime/omni/controllers/omni/machineconfig/status.go

@@ -43,7 +43,6 @@ import (
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/helpers"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/mappers"
 	talosutils "github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/talos"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 	"github.com/siderolabs/omni/internal/backend/runtime/talos"
 	"github.com/siderolabs/omni/internal/backend/talos/lifecycle"
 )
@@ -895,11 +894,9 @@ func (ctrl *StatusController) etcdForfeitHook(rc *ReconciliationContext, cluster
 	}
 }
 
-// listClusterMachines lists the cluster's ClusterMachines uncached, so callers see the freshest finalizer set.
+// listClusterMachines lists the cluster's ClusterMachines.
 func (ctrl *StatusController) listClusterMachines(ctx context.Context, r controller.ReaderWriter, clusterName string) ([]resource.Resource, error) {
-	qruntime := r.(controller.QRuntime) //nolint:forcetypeassert,errcheck
-
-	list, err := qruntime.ListUncached(
+	list, err := r.List(
 		ctx,
 		omni.NewClusterMachine("").Metadata(),
 		state.WithLabelQuery(resource.LabelEqual(omni.LabelCluster, clusterName)),
@@ -1253,9 +1250,7 @@ func (ctrl *StatusController) shouldReset(
 		return false, fmt.Errorf("finalizer: failed to get cluster status '%s': %w", clusterName, err)
 	}
 
-	// Read the Machine uncached: a cached read can briefly still report a Machine that is being torn
-	// down or destroyed as present and running, so the reset decision below could be made on stale data.
-	machine, err := safe.ReaderGetByID[*omni.Machine](ctx, uncached.Reader(r), machineID)
+	machine, err := safe.ReaderGetByID[*omni.Machine](ctx, r, machineID)
 	if err != nil {
 		if state.IsNotFoundError(err) {
 			return false, nil
@@ -1459,15 +1454,12 @@ func (ctrl *StatusController) acquireConfigUpdateLock(ctx context.Context, r con
 		return errors.New("failed to get machine set name from the cluster machine")
 	}
 
-	// A stale strategy could exceed the configured update parallelism.
-	machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, uncached.Reader(r), machineSetName)
+	machineSetConfigStatus, err := safe.ReaderGetByID[*omni.MachineSetConfigStatus](ctx, r, machineSetName)
 	if err != nil {
 		return err
 	}
 
-	qruntime := r.(controller.QRuntime) //nolint:forcetypeassert,errcheck
-
-	clusterMachines, err := qruntime.ListUncached(
+	clusterMachines, err := r.List(
 		ctx,
 		omni.NewClusterMachine("").Metadata(),
 		state.WithLabelQuery(resource.LabelEqual(omni.LabelMachineSet, machineSetName)),

internal/backend/runtime/omni/controllers/omni/maintenance_config_status.go

@@ -41,7 +41,6 @@ import (
 	siderolinkres "github.com/siderolabs/omni/client/pkg/omni/resources/siderolink"
 	"github.com/siderolabs/omni/client/pkg/siderolink"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/imagefactoryauth"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 	"github.com/siderolabs/omni/internal/backend/runtime/talos"
 )
 
@@ -278,8 +277,7 @@ func (helper *maintenanceConfigStatusControllerHelper) transform(ctx context.Con
 	}
 
 	// wait for the machine's incoming config to be extracted before applying anything, otherwise we could overwrite (and lose) the config the machine arrived with before it is preserved.
-	// read uncached, to avoid acting on a stale (or not yet visible) extraction status.
-	extractionStatus, err := safe.ReaderGetByID[*omni.MachineConfigExtractionStatus](ctx, uncached.Reader(r), link.Metadata().ID())
+	extractionStatus, err := safe.ReaderGetByID[*omni.MachineConfigExtractionStatus](ctx, r, link.Metadata().ID())
 	if err != nil && !state.IsNotFoundError(err) {
 		return err
 	}
@@ -288,8 +286,7 @@ func (helper *maintenanceConfigStatusControllerHelper) transform(ctx context.Con
 		return xerrors.NewTaggedf[qtransform.SkipReconcileTag]("machine config has not been extracted yet")
 	}
 
-	// read machine config patches uncached, to avoid acting on a stale (or not yet visible) just-extracted preserved config patch
-	machinePatches, err := getMachinePatches(ctx, uncached.Reader(r), link.Metadata().ID())
+	machinePatches, err := getMachinePatches(ctx, r, link.Metadata().ID())
 	if err != nil {
 		return fmt.Errorf("error collecting machine config patches: %w", err)
 	}

internal/backend/runtime/omni/controllers/omni/node_unique_token_cleanup.go

@@ -21,7 +21,6 @@ import (
 	"github.com/siderolabs/omni/client/pkg/cosi/helpers"
 	"github.com/siderolabs/omni/client/pkg/omni/resources"
 	"github.com/siderolabs/omni/client/pkg/omni/resources/siderolink"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 )
 
 // NodeUniqueTokenCleanupControllerName is the name of the NodeUniqueTokenCleanupController.
@@ -102,10 +101,7 @@ func (ctrl *NodeUniqueTokenCleanupController) Reconcile(ctx context.Context,
 }
 
 func (ctrl *NodeUniqueTokenCleanupController) reconcileRunning(ctx context.Context, r controller.QRuntime, logger *zap.Logger, nodeUniqueToken *siderolink.NodeUniqueToken) error {
-	// Read the link uncached. The cached input replica is fed by a separate watch stream and can lag
-	// real state, so a link that already exists may not yet be visible when the token's reconcile fires.
-	// Acting on a stale "no link" view here would wrongly tear down a token that actually has a link.
-	link, err := safe.ReaderGetByID[*siderolink.Link](ctx, uncached.Reader(r), nodeUniqueToken.Metadata().ID())
+	link, err := safe.ReaderGetByID[*siderolink.Link](ctx, r, nodeUniqueToken.Metadata().ID())
 	if err != nil && !state.IsNotFoundError(err) {
 		return err
 	}

internal/backend/runtime/omni/controllers/omni/node_unique_token_cleanup_test.go

@@ -43,8 +43,7 @@ func (suite *MachineStatusSnapshotControllerSuite) TestNodeUniqueTokenCleanup()
 
 	link := siderolink.NewLink(token.Metadata().ID(), &specs.SiderolinkSpec{})
 
-	// Create the link before recreating the token. The controller reads the link uncached, so an
-	// existing link is always observed and the token is never treated as orphaned, even on a busy runner.
+	// Create the link before recreating the token, so the token is never treated as orphaned.
 	suite.Require().NoError(suite.state.Create(ctx, link))
 	suite.Require().NoError(suite.state.Create(ctx, token))
 

internal/backend/runtime/omni/controllers/omni/redactedmachineconfig/redacted_cluster_machine_config.go

@@ -24,7 +24,6 @@ import (
 	"github.com/siderolabs/omni/client/pkg/omni/resources"
 	"github.com/siderolabs/omni/client/pkg/omni/resources/omni"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/helpers"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 )
 
 // ControllerName is the name of Controller.
@@ -198,7 +197,7 @@ func (ctrl *Controller) reconcileRunning(ctx context.Context, r controller.Reade
 }
 
 func (ctrl *Controller) reconcileTearingDown(ctx context.Context, r controller.ReaderWriter, cmc *omni.ClusterMachineConfigStatus) error {
-	list, err := uncached.ReaderWriter(r).List(ctx, omni.NewMachineConfigDiff("").Metadata(), state.WithLabelQuery(resource.LabelEqual(omni.LabelMachine, cmc.Metadata().ID())))
+	list, err := r.List(ctx, omni.NewMachineConfigDiff("").Metadata(), state.WithLabelQuery(resource.LabelEqual(omni.LabelMachine, cmc.Metadata().ID())))
 	if err != nil {
 		return fmt.Errorf("failed to list diffs for machine config %q: %w", cmc.Metadata().ID(), err)
 	}

internal/backend/runtime/omni/controllers/omni/schematic/configuration.go

@@ -32,7 +32,6 @@ import (
 	"github.com/siderolabs/omni/internal/backend/kernelargs"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/helpers"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/set"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 )
 
 // ConfigurationControllerName is the name of the SchematicConfiguration controller.
@@ -333,7 +332,7 @@ func newMachineCustomization(ctx context.Context, r controller.Reader, ms *omni.
 		return mc, nil
 	}
 
-	machineExtensions, err := safe.ReaderGetByID[*omni.MachineExtensions](ctx, uncached.Reader(r), ms.Metadata().ID())
+	machineExtensions, err := safe.ReaderGetByID[*omni.MachineExtensions](ctx, r, ms.Metadata().ID())
 	if err != nil && !state.IsNotFoundError(err) {
 		return mc, err
 	}
@@ -412,11 +411,7 @@ func determineKernelArgs(ctx context.Context, machineStatus *omni.MachineStatus,
 		return machineStatus.TypedSpec().Value.Schematic.KernelArgs, nil
 	}
 
-	// Read the KernelArgs resource with the given ID, bypassing the resource cache.
-	// We need this to avoid stale reads of the KernelArgs resource: there can be cases where the omni.KernelArgsInitialized annotation is present in the MachineStatus,
-	// but the KernelArgs resource is not yet visible due to the resource cache, which can cause unwanted Talos upgrades through a schematic id update.
-	kernelArgs, err := safe.ReaderGetByID[*omni.KernelArgs](ctx, uncached.Reader(r), machineStatus.Metadata().ID())
-
+	kernelArgs, err := safe.ReaderGetByID[*omni.KernelArgs](ctx, r, machineStatus.Metadata().ID())
 	if err != nil && !state.IsNotFoundError(err) {
 		return nil, err
 	}

internal/backend/runtime/omni/controllers/omni/secrets/rotation_status.go

@@ -33,7 +33,6 @@ import (
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/mappers"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/secretrotation"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/sequence"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 )
 
 // RotationStatusControllerName is the name of the SecretRotationStatusController.
@@ -184,13 +183,6 @@ func (s *Rotator) createInitialStage(currentPhase specs.SecretRotationSpec_Phase
 		version, _ = rotationStatus.Metadata().Annotations().Get(omni.RotateKubernetesCAVersion)
 		shouldRotateKubernetesCA := rotateKubernetesCA != nil && version != rotateKubernetesCA.Metadata().Version().String()
 
-		// We should use an uncached reader if we need to do a rotation. This is because a rotation is a one-time operation, therefore
-		// we cannot tolerate stale reads during it - eventual consistency doesn't work for us here.
-		// We have observed a flake in the unit tests, caused by a stale read of ClusterMachineStatus resources.
-		if shouldRotateTalosCA || shouldRotateKubernetesCA {
-			r = uncached.ReaderWriter(r)
-		}
-
 		rotationStatus.TypedSpec().Value.Component = specs.SecretRotationSpec_NONE
 		rotationStatus.TypedSpec().Value.Phase = currentPhase
 		rotationStatus.TypedSpec().Value.Status = ""
@@ -310,7 +302,7 @@ func (s *Rotator) addRotationStage(previousPhase, currentPhase specs.SecretRotat
 	sequenceContext sequence.Context[*omni.ClusterSecrets, *omni.ClusterSecretsRotationStatus],
 ) error {
 	return func(ctx context.Context, logger *zap.Logger, sequenceContext sequence.Context[*omni.ClusterSecrets, *omni.ClusterSecretsRotationStatus]) error {
-		r := uncached.ReaderWriter(sequenceContext.Runtime)
+		r := sequenceContext.Runtime
 		clusterSecrets := sequenceContext.Input
 		rotationStatus := sequenceContext.Output
 		rotationStatus.TypedSpec().Value.Phase = currentPhase

internal/backend/runtime/omni/controllers/omni/secrets/secrets.go

@@ -30,7 +30,6 @@ import (
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/etcdbackup"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/etcdbackup/store"
 	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/omni/internal/mappers"
-	"github.com/siderolabs/omni/internal/backend/runtime/omni/controllers/uncached"
 )
 
 // Controller creates omni.ClusterSecrets for each inputed omni.Cluster.
@@ -87,19 +86,7 @@ func NewSecretsController(etcdBackupStoreFactory store.Factory) *Controller {
 					return err
 				}
 
-				// Here, we need to use uncached reader, since once Secrets is created, it will never be attempted again,
-				// because of this, we cannot tolerate a stale read - later reconciliation will not put things back in order.
-				// The precise sequence of events we want to avoid is:
-				//
-				// t0: ImportedClusterSecrets is created
-				// t1: Control Plane MachineSet is created
-				// t2: The controller wakes up due to MachineSet was mapped to Cluster:
-				// - it can read the MachineSet and proceeds
-				// - despite ImportedClusterSecrets was created before, it is not yet visible to the controller due to the controller runtime cache
-				// - the controller proceeds to create a new bundle
-				// t3: ImportedClusterSecrets notification wakes up the controller, but the bundle is already there, the controller does not do anything
-
-				ics, err := safe.ReaderGetByID[*omni.ImportedClusterSecrets](ctx, uncached.Reader(r), cluster.Metadata().ID())
+				ics, err := safe.ReaderGetByID[*omni.ImportedClusterSecrets](ctx, r, cluster.Metadata().ID())
 				if err != nil && !state.IsNotFoundError(err) {
 					return err
 				}

internal/backend/runtime/omni/omni.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"os"
 	"slices"
+	"testing"
 	"time"
 
 	"github.com/cosi-project/runtime/pkg/controller"
@@ -398,6 +399,16 @@ func NewRuntime(cfg *config.Params, talosClientFactory *talos.ClientFactory, dns
 
 // RuntimeCacheOptions returns the COSI runtime cache options.
 func RuntimeCacheOptions() []options.Option {
+	if testing.Testing() {
+		// In unit tests, no resources are marked as cached, which effectively disables the resource cache.
+		//
+		// The in-memory state used by unit tests does not (yet) provide the cross-resource watch event ordering
+		// the cache needs to stay consistent, so unit tests exercise controller logic against the state directly.
+		// The cache is exercised by the integration tests, which run against the etcd-backed state.
+		// Remove this once the in-memory state provides the same ordering guarantees as the etcd one.
+		return nil
+	}
+
 	return []options.Option{
 		safe.WithResourceCache[*omni.BackupData](),
 		safe.WithResourceCache[*omni.ConfigPatch](),