perf(controller): repair mirror drift from watch events instead of resync

Mirror update events were filtered out, so an in-place edit of a
mirrored copy was invisible to the controller. The periodic resync did
not catch it either: NeedsSync only compares the source content hash
with the recorded annotation, and tampered mirror data leaves both
unchanged.

Changes:
- The mirror watch now passes real update events (resourceVersion
  changed, object is a managed mirror) and maps them to the source.
  Informer resync replays keep an unchanged resourceVersion and stay
  filtered.
- reconcileMirror detects in-place drift by rebuilding the expected
  mirror in memory and comparing content hashes. Metadata differences
  never count as drift, so the controller's own writes converge in one
  no-op pass.
- The resync-period default moves from 10m to 1h. Watches now cover
  source changes, mirror deletion, mirror drift, and namespace changes,
  so the resync is only a safety net for missed watch events. The chart
  default moves with it.
- Add predicate and tamper-repair regression tests; drift repair is
  verified to make no direct apiserver reads.

README.md

@@ -592,7 +592,7 @@ Complete configuration reference:
 | `controller.workerThreads` | Concurrent reconciliation workers | `5` | `10`, `20` |
 | `controller.rateLimitQPS` | API rate limit (queries per second) | `50.0` | `100.0`, `200.0` |
 | `controller.rateLimitBurst` | API burst allowance | `100` | `200`, `500` |
-| `controller.resyncPeriod` | Full cache resync period | `10m` | `30m` |
+| `controller.resyncPeriod` | Full cache resync period (safety net; drift repair is event-driven) | `1h` | `30m` |
 | `controller.verifySourceFreshness` | Verify cache against a direct API read before mirroring | `false` | `true`, `false` |
 | **Namespace Filtering** | | | |
 | `controller.excludedNamespaces` | Comma-separated namespace exclusion list | `""` | `kube-system,kube-public,kube-node-lease` |
@@ -623,7 +623,7 @@ When running the binary directly:
 - `--worker-threads int` - Concurrent workers (default: 5)
 - `--rate-limit-qps float32` - API rate limit (default: 50.0)
 - `--rate-limit-burst int` - API burst limit (default: 100)
-- `--resync-period duration` - Full cache resync period (default: 10m)
+- `--resync-period duration` - Full cache resync period, a safety net for missed watch events (default: 1h)
 - `--verify-source-freshness` - Verify cache against a direct API read before mirroring (default: false)
 
 **Namespace Filtering:**

charts/kubemirror/README.md

@@ -53,7 +53,7 @@ Defaults are taken from [`values.yaml`](values.yaml).
 | `controller.leaderElectionID` | `"kubemirror-controller-leader"` | Lease name (`--leader-election-id`). |
 | `controller.resourceTypes` | `[]` | Resource types to mirror (`--resource-types`). Empty enables auto-discovery. |
 | `controller.discoveryInterval` | `"5m"` | Auto-discovery interval (`--discovery-interval`). |
-| `controller.resyncPeriod` | `"10m"` | Cache resync period (`--resync-period`). |
+| `controller.resyncPeriod` | `"1h"` | Cache resync period, safety net only (`--resync-period`). |
 | `controller.maxTargets` | `100` | Max target namespaces per resource (`--max-targets`). |
 | `controller.workerThreads` | `5` | Concurrent reconcile workers (`--worker-threads`). |
 | `controller.rateLimitQPS` | `50.0` | API QPS limit (`--rate-limit-qps`). |

charts/kubemirror/values.yaml

@@ -54,10 +54,11 @@ controller:
   # How often to rediscover available resources in the cluster
   discoveryInterval: "5m"
 
-  # Cache resync period - how often to refresh all cached resources
+  # Cache resync period - safety net for missed watch events; mirror drift
+  # is repaired event-driven, so this can stay long
   # Higher values reduce memory churn and API load
   # Default: 10m (was 30s in earlier versions)
-  resyncPeriod: "10m"
+  resyncPeriod: "1h"
 
   # Resource limits
   maxTargets: 100

cmd/kubemirror/main.go

@@ -108,8 +108,10 @@ func main() {
 		"QPS rate limit for API server requests.")
 	flag.IntVar(&rateLimitBurst, "rate-limit-burst", 100,
 		"Burst limit for API server requests.")
-	flag.DurationVar(&resyncPeriod, "resync-period", 10*time.Minute,
+	flag.DurationVar(&resyncPeriod, "resync-period", time.Hour,
-		"Period for resyncing all resources (catches updates missed due to informer cache delays).")
+		"Period for the full informer resync. Sources, mirrors, and namespaces "+
+			"are watched, and mirror drift is repaired event-driven, so the resync "+
+			"is only a safety net for missed watch events.")
 	flag.BoolVar(&verifySourceFreshness, "verify-source-freshness", false,
 		"Verify source resource freshness by comparing cache with direct API read. "+
 			"Converges a few seconds faster when the informer cache lags behind watch "+

pkg/controller/source_reconciler.go

@@ -378,9 +378,20 @@ func (r *SourceReconciler) reconcileMirror(ctx context.Context, source runtime.O
 		}
 
 		if !needsSync {
+			// The source is unchanged since the last sync, but the mirrored
+			// copy itself may have been edited in place (its bookkeeping
+			// annotations would still match the source). Rebuild the expected
+			// mirror in memory and compare content hashes to catch that.
+			drifted, driftErr := mirrorContentDrifted(source, existing, targetNs)
+			if driftErr != nil {
+				return fmt.Errorf("failed to check mirror drift: %w", driftErr)
+			}
+			if !drifted {
 				logger.V(2).Info("mirror is up to date")
 				return nil
 			}
+			logger.Info("mirror content drifted from expected state, repairing")
+		}
 
 		// Update mirror
 		updateErr := UpdateMirror(existing, source)
@@ -421,6 +432,30 @@ func (r *SourceReconciler) reconcileMirror(ctx context.Context, source runtime.O
 	return nil
 }
 
+// mirrorContentDrifted reports whether the live mirror's content diverged from
+// what mirroring the source into targetNs should produce. It rebuilds the
+// expected mirror in memory (including transformations) and compares content
+// hashes, so metadata-only differences never count as drift. This is what
+// makes drift repair event-driven instead of depending on the periodic resync.
+func mirrorContentDrifted(source runtime.Object, existing *unstructured.Unstructured, targetNs string) (bool, error) {
+	expected, err := CreateMirror(source, targetNs)
+	if err != nil {
+		return false, fmt.Errorf("failed to build expected mirror: %w", err)
+	}
+
+	expectedHash, err := hash.ComputeContentHash(expected)
+	if err != nil {
+		return false, fmt.Errorf("failed to hash expected mirror: %w", err)
+	}
+
+	existingHash, err := hash.ComputeContentHash(existing)
+	if err != nil {
+		return false, fmt.Errorf("failed to hash existing mirror: %w", err)
+	}
+
+	return expectedHash != existingHash, nil
+}
+
 // deleteAllMirrors deletes all mirrors that this source owns across the cluster.
 // It verifies ownership (managed-by label + source-reference annotation) before
 // deleting anything to avoid destroying unrelated resources that happen to share
@@ -680,26 +715,43 @@ func (r *SourceReconciler) SetupWithManagerForResourceType(
 	mirrorObj := &unstructured.Unstructured{}
 	mirrorObj.SetGroupVersionKind(gvk)
 
-	// Create predicates to only watch mirror deletions
-	mirrorDeletePredicate := predicate.Funcs{
-		CreateFunc:  func(e event.CreateEvent) bool { return false },
-		UpdateFunc:  func(e event.UpdateEvent) bool { return false },
-		DeleteFunc:  func(e event.DeleteEvent) bool { return IsMirrorResource(e.Object) },
-		GenericFunc: func(e event.GenericEvent) bool { return false },
-	}
-
 	return ctrl.NewControllerManagedBy(mgr).
 		For(obj).
 		Named(controllerName).
-		// Watch mirror resources - when deleted, enqueue source for reconciliation
+		// Watch mirror resources - deletes trigger recreation, real writes
+		// trigger drift repair. This keeps mirrors converged without relying
+		// on the periodic resync.
 		Watches(
 			mirrorObj,
 			handler.EnqueueRequestsFromMapFunc(r.mapMirrorToSource),
-			builder.WithPredicates(mirrorDeletePredicate),
+			builder.WithPredicates(mirrorEventPredicate()),
 		).
 		Complete(r)
 }
 
+// mirrorEventPredicate gates the mirror watch. Deletes of managed mirrors
+// enqueue the source so the mirror is recreated. Updates enqueue the source
+// so in-place edits to a mirror are repaired event-driven; informer resync
+// replays carry an unchanged resourceVersion and are filtered out, so only
+// real writes pass. The controller's own mirror writes pass too and converge
+// in one extra reconcile that no-ops on the content comparison.
+func mirrorEventPredicate() predicate.Funcs {
+	return predicate.Funcs{
+		CreateFunc: func(e event.CreateEvent) bool { return false },
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			if e.ObjectOld == nil || e.ObjectNew == nil {
+				return false
+			}
+			if e.ObjectOld.GetResourceVersion() == e.ObjectNew.GetResourceVersion() {
+				return false
+			}
+			return IsMirrorResource(e.ObjectNew)
+		},
+		DeleteFunc:  func(e event.DeleteEvent) bool { return IsMirrorResource(e.Object) },
+		GenericFunc: func(e event.GenericEvent) bool { return false },
+	}
+}
+
 // mapMirrorToSource maps a mirror resource to its source for reconciliation.
 func (r *SourceReconciler) mapMirrorToSource(ctx context.Context, obj client.Object) []reconcile.Request {
 	// Only process if this is a mirror

pkg/controller/source_reconciler_test.go

@@ -18,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 
 	"github.com/lukaszraczylo/kubemirror/pkg/config"
 	"github.com/lukaszraczylo/kubemirror/pkg/constants"
@@ -1364,3 +1365,75 @@ func TestSourceReconciler_getSourceWithFreshness_DefaultNoDirectRead(t *testing.
 	assert.Equal(t, "test-secret", got.GetName())
 	assert.Zero(t, reader.gets, "default config must not verify freshness against the apiserver")
 }
+
+// TestMirrorEventPredicate locks the mirror-watch gating: real writes to a
+// mirror enqueue its source for drift repair, while informer resync replays
+// (unchanged resourceVersion), non-mirror objects, and creates stay silent.
+func TestMirrorEventPredicate(t *testing.T) {
+	source := newMirrorTestSource(t)
+	mirrorObj, err := CreateMirror(source, "target-ns")
+	require.NoError(t, err)
+	mirror, ok := mirrorObj.(*unstructured.Unstructured)
+	require.True(t, ok)
+	mirror.SetResourceVersion("100")
+
+	changed := mirror.DeepCopy()
+	changed.SetResourceVersion("101")
+
+	nonMirrorOld := source.DeepCopy()
+	nonMirrorOld.SetResourceVersion("1")
+	nonMirrorNew := source.DeepCopy()
+	nonMirrorNew.SetResourceVersion("2")
+
+	p := mirrorEventPredicate()
+
+	assert.False(t, p.Create(event.CreateEvent{Object: mirror}),
+		"mirror create must not enqueue")
+	assert.True(t, p.Update(event.UpdateEvent{ObjectOld: mirror, ObjectNew: changed}),
+		"real mirror write must enqueue the source for drift repair")
+	assert.False(t, p.Update(event.UpdateEvent{ObjectOld: mirror, ObjectNew: mirror}),
+		"informer resync replay with unchanged resourceVersion must not enqueue")
+	assert.False(t, p.Update(event.UpdateEvent{ObjectOld: nonMirrorOld, ObjectNew: nonMirrorNew}),
+		"non-mirror update must not enqueue")
+	assert.True(t, p.Delete(event.DeleteEvent{Object: mirror}),
+		"mirror delete must enqueue the source for recreation")
+	assert.False(t, p.Delete(event.DeleteEvent{Object: source}),
+		"non-mirror delete must not enqueue")
+}
+
+// TestSourceReconciler_reconcileMirror_RepairsDriftedMirror covers in-place
+// mirror tampering: the mirrored data is edited but the bookkeeping
+// annotations are left intact, so the source-hash check alone would consider
+// the mirror up to date. The reconciler must detect the content divergence
+// and rewrite the mirror — still without any direct API reads.
+func TestSourceReconciler_reconcileMirror_RepairsDriftedMirror(t *testing.T) {
+	source := newMirrorTestSource(t)
+
+	mirrorObj, err := CreateMirror(source, "target-ns")
+	require.NoError(t, err)
+	mirror, ok := mirrorObj.(*unstructured.Unstructured)
+	require.True(t, ok)
+	mirror.SetResourceVersion("100")
+
+	// Tamper with the mirrored content in place; annotations stay intact.
+	require.NoError(t, unstructured.SetNestedStringMap(mirror.Object,
+		map[string]string{"tampered-field": "dGFtcGVyZWQ="}, "data"))
+
+	mockClient := new(MockClient)
+	mockClient.On("Get", mock.Anything, types.NamespacedName{Namespace: "target-ns", Name: "test-secret"}, mock.Anything).
+		Return(nil, mirror)
+	mockClient.On("Update", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+	reader := &countingReader{}
+	r := &SourceReconciler{
+		Client:    mockClient,
+		APIReader: reader,
+		Config:    &config.Config{},
+		GVK:       source.GroupVersionKind(),
+	}
+
+	require.NoError(t, r.reconcileMirror(context.Background(), source, source, "target-ns"))
+
+	mockClient.AssertCalled(t, "Update", mock.Anything, mock.Anything, mock.Anything)
+	assert.Zero(t, reader.gets, "drift repair must not read from the apiserver directly")
+}