From 49b2a779fcec5d36ad3c7eef1a8dd6368a89c09f Mon Sep 17 00:00:00 2001 From: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> Date: Sat, 15 Aug 2026 07:14:09 -0700 Subject: [PATCH 1/4] fix(cache): restore Dataset to Bound after CacheRuntime recovers from an outage Motivation: A CacheRuntime-backed Dataset that flips to Failed during a transient runtime outage (e.g. a worker pod restart) never returns to Bound, even after the runtime becomes fully Ready again. Only deleting and recreating the CacheRuntime restored the correct state. Approach: CacheEngine.Sync only refreshed cache states when the runtime was ready, without touching the Dataset's phase. The Bound phase was otherwise only ever set once, by BindToDataset during initial Setup, which does not run again on later reconciles, so the Failed phase was a one-way trap. Sync now checks the Dataset's current phase whenever the runtime is ready and, if it is Failed, restores it to Bound via the existing UpdateDatasetStatus helper before falling back to the regular cache-states sync. Validation: - go build ./... - go vet ./pkg/ddc/cache/... - gofmt -l pkg/ddc/cache/engine/sync.go pkg/ddc/cache/engine/sync_test.go (no output) - golangci-lint run ./pkg/ddc/cache/... -> 0 issues - go test ./pkg/ddc/cache/engine/... -run TestCacheEngine --ginkgo.focus="left Failed by a previous outage" -v -> PASS - Confirmed the new test is a genuine regression test: reverting only sync.go while keeping the new test makes it fail with Failed != Bound - Full unfocused suite in this package shows pre-existing, order-dependent flaky failures in ufs_test.go/dataset_test.go/fileutils_test.go that reproduce identically on unmodified master, unrelated to this change Report: https://github.com/fluid-cloudnative/fluid/issues/6160 Signed-off-by: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> --- pkg/ddc/cache/engine/sync.go | 28 ++++++++++--- pkg/ddc/cache/engine/sync_test.go | 65 +++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 6 deletions(-) diff --git a/pkg/ddc/cache/engine/sync.go b/pkg/ddc/cache/engine/sync.go index c7c45c6c24e..54caa777c14 100644 --- a/pkg/ddc/cache/engine/sync.go +++ b/pkg/ddc/cache/engine/sync.go @@ -90,12 +90,28 @@ func (e *CacheEngine) Sync(ctx cruntime.ReconcileRequestContext) (err error) { if err != nil { return err } - } else if permitSyncEngineStatus { - // sync dataset cache states when runtime is ready and sync permitted - e.Log.Info("sync dataset cache states") - err = e.syncDatasetCacheStates(ctx, runtime, runtimeClass) - if err != nil { - return err + } else { + dataset, getErr := utils.GetDataset(e.Client, e.name, e.namespace) + if getErr != nil { + return getErr + } + + if dataset.Status.Phase == datav1alpha1.FailedDatasetPhase { + // the runtime recovered from a previous outage but the dataset was left in Failed + // phase because the phase is otherwise only restored to Bound by the mount flow, + // which does not run on a normal reconcile. Restore it here. + e.Log.Info("runtime is ready again, restoring dataset phase from Failed to Bound") + err = e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass) + if err != nil { + return err + } + } else if permitSyncEngineStatus { + // sync dataset cache states when runtime is ready and sync permitted + e.Log.Info("sync dataset cache states") + err = e.syncDatasetCacheStates(ctx, runtime, runtimeClass) + if err != nil { + return err + } } } diff --git a/pkg/ddc/cache/engine/sync_test.go b/pkg/ddc/cache/engine/sync_test.go index 509a417dbb6..e1cbb850017 100644 --- a/pkg/ddc/cache/engine/sync_test.go +++ b/pkg/ddc/cache/engine/sync_test.go @@ -338,6 +338,71 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test }) }) + Context("when runtime is ready but dataset was left Failed by a previous outage", func() { + BeforeEach(func() { + dataset.Status.Phase = datav1alpha1.FailedDatasetPhase + + masterReplicas := int32(1) + masterSts := &workloadv1alpha1.AdvancedStatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-runtime-master", Namespace: "default"}, + Spec: workloadv1alpha1.AdvancedStatefulSetSpec{ + Replicas: &masterReplicas, + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "master", Image: "test-master:latest"}}, + }, + }, + }, + Status: workloadv1alpha1.AdvancedStatefulSetStatus{ReadyReplicas: 1, CurrentReplicas: 1, AvailableReplicas: 1}, + } + + workerReplicas := int32(2) + workerSts := &workloadv1alpha1.AdvancedStatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-runtime-worker", Namespace: "default"}, + Spec: workloadv1alpha1.AdvancedStatefulSetSpec{ + Replicas: &workerReplicas, + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "worker", Image: "test-worker:latest"}}, + }, + }, + }, + Status: workloadv1alpha1.AdvancedStatefulSetStatus{ReadyReplicas: 2, CurrentReplicas: 2, AvailableReplicas: 2}, + } + + clientDs := &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-runtime-client", Namespace: "default"}, + Spec: appsv1.DaemonSetSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "client", Image: "test-client:latest"}}, + }, + }, + }, + Status: appsv1.DaemonSetStatus{NumberReady: 0, DesiredNumberScheduled: 0}, + } + + engine.Client = fake.NewClientBuilder(). + WithScheme(CacheEngineTestScheme). + WithObjects(dataset, runtimeObj, runtimeClass, masterSts, workerSts, clientDs). + WithStatusSubresource(dataset, runtimeObj). + Build() + }) + + It("should restore the dataset phase to Bound", func() { + err := engine.Sync(ctx) + Expect(err).NotTo(HaveOccurred()) + + updatedDataset := &datav1alpha1.Dataset{} + err = engine.Client.Get(context.Background(), types.NamespacedName{ + Name: "test-runtime", + Namespace: "default", + }, updatedDataset) + Expect(err).NotTo(HaveOccurred()) + Expect(updatedDataset.Status.Phase).To(Equal(datav1alpha1.BoundDatasetPhase)) + }) + }) + Context("when runtime is ready with ReportSummary configured", func() { var patches *gomonkey.Patches From 2e2b62a127c4ef73e6d984b5907f301785efb8aa Mon Sep 17 00:00:00 2001 From: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> Date: Sun, 16 Aug 2026 23:04:16 -0700 Subject: [PATCH 2/4] fix(cache): keep dataset phase-restore cheap and rate-limited UpdateDatasetStatus now short-circuits when the dataset is already in the requested phase, and only execs into the master pod for cache states (GetCacheStates) when phase == Bound and the sync limiter permits it. Previously the Failed->Bound restore path in Sync() called this unconditionally, bypassing the same rate limiter that bounds every other engine RPC. Addresses review feedback from cheyang on PR #6162. Signed-off-by: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> --- pkg/ddc/cache/engine/dataset.go | 15 ++++++++++++-- pkg/ddc/cache/engine/sync.go | 4 +++- pkg/ddc/cache/engine/sync_test.go | 33 ++++++++++++++++++++++++++++++- 3 files changed, 48 insertions(+), 4 deletions(-) diff --git a/pkg/ddc/cache/engine/dataset.go b/pkg/ddc/cache/engine/dataset.go index 35abdd1dd0a..6b076ee0c3f 100644 --- a/pkg/ddc/cache/engine/dataset.go +++ b/pkg/ddc/cache/engine/dataset.go @@ -45,8 +45,19 @@ func (e *CacheEngine) BindToDataset(runtime *datav1alpha1.CacheRuntime, runtimeC func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (err error) { var cacheStates common.CacheStateList - // only update cache states for BoundDatasetPhase - if phase == datav1alpha1.BoundDatasetPhase { + current, err := utils.GetDataset(e.Client, e.name, e.namespace) + if err != nil { + return err + } + if current.Status.Phase == phase { + // already in the desired phase, nothing to do + return nil + } + + // GetCacheStates execs into the master pod with a floor of MinExecutionTimeoutSeconds, + // so keep it behind the same rate limiter that bounds other engine RPCs, and only + // attempt it for BoundDatasetPhase. + if phase == datav1alpha1.BoundDatasetPhase && e.permitSync() { e.Log.V(1).Info("Start to update cache states") cacheStates, err = e.GetCacheStates(runtime, runtimeClass) if err != nil { diff --git a/pkg/ddc/cache/engine/sync.go b/pkg/ddc/cache/engine/sync.go index 54caa777c14..f92b5171e33 100644 --- a/pkg/ddc/cache/engine/sync.go +++ b/pkg/ddc/cache/engine/sync.go @@ -99,7 +99,9 @@ func (e *CacheEngine) Sync(ctx cruntime.ReconcileRequestContext) (err error) { if dataset.Status.Phase == datav1alpha1.FailedDatasetPhase { // the runtime recovered from a previous outage but the dataset was left in Failed // phase because the phase is otherwise only restored to Bound by the mount flow, - // which does not run on a normal reconcile. Restore it here. + // which does not run on a normal reconcile. Restore it here. UpdateDatasetStatus + // keeps this cheap: it only execs into the master pod for cache states when the + // sync limiter permits it. e.Log.Info("runtime is ready again, restoring dataset phase from Failed to Bound") err = e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass) if err != nil { diff --git a/pkg/ddc/cache/engine/sync_test.go b/pkg/ddc/cache/engine/sync_test.go index e1cbb850017..856f15fb938 100644 --- a/pkg/ddc/cache/engine/sync_test.go +++ b/pkg/ddc/cache/engine/sync_test.go @@ -22,6 +22,7 @@ import ( "time" "github.com/fluid-cloudnative/fluid/pkg/common" + "github.com/fluid-cloudnative/fluid/pkg/utils" "github.com/agiledragon/gomonkey/v2" "github.com/go-logr/logr" @@ -295,7 +296,7 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test engine.Client = fake.NewClientBuilder(). WithScheme(scheme). WithObjects(dataset, runtimeObj, runtimeClass, configMap, masterSts, workerSts, clientDs). - WithStatusSubresource(runtimeObj). + WithStatusSubresource(dataset, runtimeObj). Build() }) @@ -341,6 +342,12 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test Context("when runtime is ready but dataset was left Failed by a previous outage", func() { BeforeEach(func() { dataset.Status.Phase = datav1alpha1.FailedDatasetPhase + dataset.Status.Conditions = []datav1alpha1.DatasetCondition{ + { + Type: datav1alpha1.DatasetReady, + Status: corev1.ConditionFalse, + }, + } masterReplicas := int32(1) masterSts := &workloadv1alpha1.AdvancedStatefulSet{ @@ -400,6 +407,30 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test }, updatedDataset) Expect(err).NotTo(HaveOccurred()) Expect(updatedDataset.Status.Phase).To(Equal(datav1alpha1.BoundDatasetPhase)) + + idx, cond := utils.GetDatasetCondition(updatedDataset.Status.Conditions, datav1alpha1.DatasetReady) + Expect(idx).NotTo(Equal(-1)) + Expect(cond.Status).To(Equal(corev1.ConditionTrue)) + }) + + Context("and the sync limiter is closed", func() { + BeforeEach(func() { + engine.syncRetryDuration = defaultSyncRetryDuration + engine.timeOfLastSync = time.Now() + }) + + It("should still restore the dataset phase to Bound without fetching cache states", func() { + err := engine.Sync(ctx) + Expect(err).NotTo(HaveOccurred()) + + updatedDataset := &datav1alpha1.Dataset{} + err = engine.Client.Get(context.Background(), types.NamespacedName{ + Name: "test-runtime", + Namespace: "default", + }, updatedDataset) + Expect(err).NotTo(HaveOccurred()) + Expect(updatedDataset.Status.Phase).To(Equal(datav1alpha1.BoundDatasetPhase)) + }) }) }) From 426ed336a4be5b41677444b7146da617635f96a0 Mon Sep 17 00:00:00 2001 From: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> Date: Mon, 17 Aug 2026 00:51:30 -0700 Subject: [PATCH 3/4] fix(cache): move permitSync check out of UpdateDatasetStatus UpdateDatasetStatus called e.permitSync() internally to decide whether to fetch cache states. Replace that with a fetchCacheStates parameter so the rate-limiting decision stays in Sync(), the only place permitSync is now called from. Addresses review feedback from xliuqq on PR #6162. Signed-off-by: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> --- pkg/ddc/cache/engine/dataset.go | 13 ++++++++----- pkg/ddc/cache/engine/sync.go | 9 ++++----- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/pkg/ddc/cache/engine/dataset.go b/pkg/ddc/cache/engine/dataset.go index 6b076ee0c3f..8ebcc5a11ec 100644 --- a/pkg/ddc/cache/engine/dataset.go +++ b/pkg/ddc/cache/engine/dataset.go @@ -39,10 +39,13 @@ func (e *CacheEngine) BindToDataset(runtime *datav1alpha1.CacheRuntime, runtimeC return } - return e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass) + return e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass, true) } -func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (err error) { +// UpdateDatasetStatus transitions the Dataset to phase. fetchCacheStates controls whether it +// execs into the master pod for fresh cache states (only relevant for BoundDatasetPhase); the +// caller decides this so that rate-limiting via permitSync stays a concern of Sync alone. +func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass, fetchCacheStates bool) (err error) { var cacheStates common.CacheStateList current, err := utils.GetDataset(e.Client, e.name, e.namespace) @@ -55,9 +58,9 @@ func (e *CacheEngine) UpdateDatasetStatus(phase datav1alpha1.DatasetPhase, runti } // GetCacheStates execs into the master pod with a floor of MinExecutionTimeoutSeconds, - // so keep it behind the same rate limiter that bounds other engine RPCs, and only - // attempt it for BoundDatasetPhase. - if phase == datav1alpha1.BoundDatasetPhase && e.permitSync() { + // so callers keep it behind the same rate limiter that bounds other engine RPCs, and it's + // only attempted for BoundDatasetPhase. + if phase == datav1alpha1.BoundDatasetPhase && fetchCacheStates { e.Log.V(1).Info("Start to update cache states") cacheStates, err = e.GetCacheStates(runtime, runtimeClass) if err != nil { diff --git a/pkg/ddc/cache/engine/sync.go b/pkg/ddc/cache/engine/sync.go index f92b5171e33..d870f24b68e 100644 --- a/pkg/ddc/cache/engine/sync.go +++ b/pkg/ddc/cache/engine/sync.go @@ -86,7 +86,7 @@ func (e *CacheEngine) Sync(ctx cruntime.ReconcileRequestContext) (err error) { if !runtimeReady { // update dataset status when runtime not ready - err = e.UpdateDatasetStatus(datav1alpha1.FailedDatasetPhase, runtime, runtimeClass) + err = e.UpdateDatasetStatus(datav1alpha1.FailedDatasetPhase, runtime, runtimeClass, permitSyncEngineStatus) if err != nil { return err } @@ -99,11 +99,10 @@ func (e *CacheEngine) Sync(ctx cruntime.ReconcileRequestContext) (err error) { if dataset.Status.Phase == datav1alpha1.FailedDatasetPhase { // the runtime recovered from a previous outage but the dataset was left in Failed // phase because the phase is otherwise only restored to Bound by the mount flow, - // which does not run on a normal reconcile. Restore it here. UpdateDatasetStatus - // keeps this cheap: it only execs into the master pod for cache states when the - // sync limiter permits it. + // which does not run on a normal reconcile. Restore it here, keeping the cache-states + // exec behind the same sync limiter as syncDatasetCacheStates below. e.Log.Info("runtime is ready again, restoring dataset phase from Failed to Bound") - err = e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass) + err = e.UpdateDatasetStatus(datav1alpha1.BoundDatasetPhase, runtime, runtimeClass, permitSyncEngineStatus) if err != nil { return err } From 1594e4f1a0d7c4a179c25b2a1b760e4e7e0a75a0 Mon Sep 17 00:00:00 2001 From: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> Date: Mon, 17 Aug 2026 01:03:20 -0700 Subject: [PATCH 4/4] fix(cache): make the closed-limiter regression test actually assert the skip The "sync limiter is closed" test only checked the restored phase, so it stayed green even with cheyang's rate-limiter-bypass bug reintroduced. Patch GetCacheStates directly (NewCacheFileUtil isn't reachable in this Context since no ReportSummary entry is configured) and assert it's never called while the limiter is closed. Signed-off-by: Pujitha Paladugu <10557236+pujitha24@users.noreply.github.com> --- pkg/ddc/cache/engine/sync_test.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/pkg/ddc/cache/engine/sync_test.go b/pkg/ddc/cache/engine/sync_test.go index 856f15fb938..a3ee3d571c9 100644 --- a/pkg/ddc/cache/engine/sync_test.go +++ b/pkg/ddc/cache/engine/sync_test.go @@ -19,6 +19,7 @@ package engine import ( "context" "os" + "reflect" "time" "github.com/fluid-cloudnative/fluid/pkg/common" @@ -414,9 +415,28 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test }) Context("and the sync limiter is closed", func() { + var patches *gomonkey.Patches + var getCacheStatesCalled bool + BeforeEach(func() { engine.syncRetryDuration = defaultSyncRetryDuration engine.timeOfLastSync = time.Now() + + getCacheStatesCalled = false + // Patched at the GetCacheStates level, not NewCacheFileUtil: this Context has no + // ReportSummary execution entries configured, so a real call would fail before ever + // reaching the exec layer. The point here is only whether GetCacheStates is invoked at all. + patches = gomonkey.ApplyMethod(reflect.TypeOf(engine), "GetCacheStates", + func(_ *CacheEngine, _ *datav1alpha1.CacheRuntime, _ *datav1alpha1.CacheRuntimeClass) (common.CacheStateList, error) { + getCacheStatesCalled = true + return common.CacheStateList{}, nil + }) + }) + + AfterEach(func() { + if patches != nil { + patches.Reset() + } }) It("should still restore the dataset phase to Bound without fetching cache states", func() { @@ -430,6 +450,7 @@ var _ = Describe("CacheEngine Sync Tests", Label("pkg.ddc.cache.engine.sync_test }, updatedDataset) Expect(err).NotTo(HaveOccurred()) Expect(updatedDataset.Status.Phase).To(Equal(datav1alpha1.BoundDatasetPhase)) + Expect(getCacheStatesCalled).To(BeFalse(), "GetCacheStates should be skipped while the sync limiter is closed") }) }) })