fix: failed to keep s3 bucket retain

api/core/v1alpha1/logset_types.go

@@ -85,20 +85,31 @@ func (l *LogSetSpec) GetStoreFailureTimeout() metav1.Duration {
 }
 
 func (l *LogSetSpec) GetPVCRetentionPolicy() PVCRetentionPolicy {
-	if l.PVCRetentionPolicy == nil {
-		l.setDefaultRetentionPolicy()
+	if l.PVCRetentionPolicy != nil {
+		return *l.PVCRetentionPolicy
 	}
-	return *l.PVCRetentionPolicy
+	// inherit from s3 policy if only s3 is set (e.g. old objects without pvcRetentionPolicy)
+	if l.SharedStorage.S3 != nil && l.SharedStorage.S3.S3RetentionPolicy != nil {
+		return *l.SharedStorage.S3.S3RetentionPolicy
+	}
+	return PVCRetentionPolicyDelete
 }
 
 func (l *LogSetSpec) GetS3RetentionPolicy() *PVCRetentionPolicy {
 	if l.SharedStorage.S3 == nil {
 		return nil
 	}
-	if l.SharedStorage.S3.S3RetentionPolicy == nil {
-		l.setDefaultRetentionPolicy()
+	if l.SharedStorage.S3.S3RetentionPolicy != nil {
+		p := *l.SharedStorage.S3.S3RetentionPolicy
+		return &p
+	}
+	// inherit from pvc policy if only pvc is set (e.g. old objects without s3RetentionPolicy)
+	if l.PVCRetentionPolicy != nil {
+		p := *l.PVCRetentionPolicy
+		return &p
 	}
-	return l.SharedStorage.S3.S3RetentionPolicy
+	defaultPolicy := PVCRetentionPolicyDelete
+	return &defaultPolicy
 }
 
 type InitialConfig struct {

api/core/v1alpha1/logset_types_test.go

@@ -1,4 +1,4 @@
-// Copyright 2025 Matrix Origin
+// Copyright 2025-2026 Matrix Origin
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -105,3 +105,112 @@ func TestSetDefaultRetentionPolicy(t *testing.T) {
 		}
 	}
 }
+
+func TestGetRetentionPolicy(t *testing.T) {
+	del := PVCRetentionPolicyDelete
+	retain := PVCRetentionPolicyRetain
+	s3Tpl := func(policy *PVCRetentionPolicy) SharedStorageProvider {
+		return SharedStorageProvider{S3: &S3Provider{S3RetentionPolicy: policy}}
+	}
+
+	testCases := []struct {
+		name      string
+		logset    LogSetSpec
+		pvcPolicy PVCRetentionPolicy
+		s3Policy  *PVCRetentionPolicy
+	}{
+		{
+			name:      "both nil with s3 returns delete",
+			logset:    LogSetSpec{SharedStorage: s3Tpl(nil)},
+			pvcPolicy: del,
+			s3Policy:  &del,
+		},
+		{
+			name:      "pvc only delete inherits to s3 getter",
+			logset:    LogSetSpec{PVCRetentionPolicy: &del, SharedStorage: s3Tpl(nil)},
+			pvcPolicy: del,
+			s3Policy:  &del,
+		},
+		{
+			name:      "s3 only retain inherits to pvc getter",
+			logset:    LogSetSpec{SharedStorage: s3Tpl(&retain)},
+			pvcPolicy: retain,
+			s3Policy:  &retain,
+		},
+		{
+			name: "both set delete and retain independently",
+			logset: LogSetSpec{
+				PVCRetentionPolicy: &del,
+				SharedStorage:      s3Tpl(&retain),
+			},
+			pvcPolicy: del,
+			s3Policy:  &retain,
+		},
+		{
+			name: "both set retain and delete independently",
+			logset: LogSetSpec{
+				PVCRetentionPolicy: &retain,
+				SharedStorage:      s3Tpl(&del),
+			},
+			pvcPolicy: retain,
+			s3Policy:  &del,
+		},
+		{
+			name: "both set retain",
+			logset: LogSetSpec{
+				PVCRetentionPolicy: &retain,
+				SharedStorage:      s3Tpl(&retain),
+			},
+			pvcPolicy: retain,
+			s3Policy:  &retain,
+		},
+		{
+			name: "both set delete",
+			logset: LogSetSpec{
+				PVCRetentionPolicy: &del,
+				SharedStorage:      s3Tpl(&del),
+			},
+			pvcPolicy: del,
+			s3Policy:  &del,
+		},
+		{
+			name:      "no s3 returns nil s3 policy",
+			logset:    LogSetSpec{PVCRetentionPolicy: &retain},
+			pvcPolicy: retain,
+			s3Policy:  nil,
+		},
+		{
+			name:      "no s3 and no pvc returns delete",
+			logset:    LogSetSpec{},
+			pvcPolicy: del,
+			s3Policy:  nil,
+		},
+	}
+
+	for _, c := range testCases {
+		t.Run(c.name, func(t *testing.T) {
+			spec := c.logset
+			origPVC := spec.PVCRetentionPolicy
+			var origS3 *PVCRetentionPolicy
+			if spec.SharedStorage.S3 != nil {
+				origS3 = spec.SharedStorage.S3.S3RetentionPolicy
+			}
+
+			assert.Equal(t, c.pvcPolicy, spec.GetPVCRetentionPolicy())
+
+			s3Policy := spec.GetS3RetentionPolicy()
+			if c.s3Policy == nil {
+				assert.Nil(t, s3Policy)
+			} else {
+				assert.NotNil(t, s3Policy)
+				assert.Equal(t, *c.s3Policy, *s3Policy)
+			}
+
+			// getters must not mutate spec
+			assert.Equal(t, origPVC, spec.PVCRetentionPolicy)
+			if spec.SharedStorage.S3 != nil {
+				assert.Equal(t, origS3, spec.SharedStorage.S3.S3RetentionPolicy)
+			}
+		})
+	}
+}

go.mod

@@ -23,7 +23,6 @@ require (
 	github.com/openkruise/kruise-api v1.4.0
 	github.com/prometheus/client_golang v1.17.0
 	github.com/samber/lo v1.38.1
-	github.com/stretchr/testify v1.9.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.24.0
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
@@ -161,6 +160,7 @@ require (
 	github.com/shirou/gopsutil/v3 v3.23.12 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/stretchr/testify v1.9.0 // indirect
 	github.com/tidwall/btree v1.6.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect

pkg/controllers/bucketclaim/controller.go

@@ -1,4 +1,4 @@
-// Copyright 2025 Matrix Origin
+// Copyright 2025-2026 Matrix Origin
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -71,6 +71,16 @@ func (bca *Actor) Finalize(ctx *recon.Context[*v1alpha1.BucketClaim]) (bool, err
 		return false, ctx.Update(bucket)
 	}
 
+	// if bucket is Released (s3RetentionPolicy=Retain), skip data deletion and release the finalizer directly
+	if bucket.Status.State == v1alpha1.StatusReleased {
+		ctx.Log.Info(fmt.Sprintf("skip data deletion for released bucket %v (s3RetentionPolicy=Retain)", client.ObjectKeyFromObject(ctx.Obj)))
+		controllerutil.RemoveFinalizer(bucket, v1alpha1.BucketDataFinalizer)
+		if err := ctx.Update(bucket); err != nil {
+			return false, err
+		}
+		return true, nil
+	}
+
 	// if AnnAnyInstanceRunning is not set, indicates that there is no running pod instance found for its mo cluster
 	// so there is no need to start a job to finalize bucket data.
 	// NOTE: generally LogSet start successfully is a precondition of starting DN and CN, if no running pod found means that

pkg/controllers/bucketclaim/controller_test.go

@@ -1,4 +1,4 @@
-// Copyright 2025 Matrix Origin
+// Copyright 2025-2026 Matrix Origin
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,12 +15,71 @@
 package bucketclaim
 
 import (
+	"context"
 	"testing"
+	"time"
 
+	"github.com/golang/mock/gomock"
+	"github.com/matrixorigin/controller-runtime/pkg/fake"
 	"github.com/matrixorigin/matrixone-operator/api/core/v1alpha1"
-	"github.com/stretchr/testify/assert"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	. "github.com/onsi/gomega"
 )
 
+func TestActor_Finalize_ReleasedSkipsS3CleanupJob(t *testing.T) {
+	g := NewGomegaWithT(t)
+
+	now := metav1.NewTime(time.Now())
+	bucket := &v1alpha1.BucketClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-bucket",
+			Namespace:         "default",
+			Finalizers:        []string{v1alpha1.BucketDataFinalizer},
+			DeletionTimestamp: &now,
+			Annotations: map[string]string{
+				v1alpha1.AnnAnyInstanceRunning: "true",
+			},
+		},
+		Spec: v1alpha1.BucketClaimSpec{
+			S3: &v1alpha1.S3Provider{Path: "minio-bucket/test"},
+		},
+		Status: v1alpha1.BucketClaimStatus{
+			State: v1alpha1.StatusReleased,
+		},
+	}
+
+	scheme := runtime.NewScheme()
+	utilruntime.Must(v1alpha1.AddToScheme(scheme))
+	utilruntime.Must(batchv1.AddToScheme(scheme))
+	utilruntime.Must(corev1.AddToScheme(scheme))
+
+	cli := fake.KubeClientBuilder().WithScheme(scheme).WithObjects(bucket).Build()
+
+	mockCtrl := gomock.NewController(t)
+	eventEmitter := fake.NewMockEventEmitter(mockCtrl)
+	ctx := fake.NewContext(bucket, cli, eventEmitter)
+
+	actor := New()
+	ok, err := actor.Finalize(ctx)
+	g.Expect(err).To(Succeed())
+	g.Expect(ok).To(BeTrue())
+
+	updated := &v1alpha1.BucketClaim{}
+	err = cli.Get(context.TODO(), client.ObjectKeyFromObject(bucket), updated)
+	g.Expect(apierrors.IsNotFound(err)).To(BeTrue())
+
+	jobs := &batchv1.JobList{}
+	g.Expect(cli.List(context.TODO(), jobs, client.InNamespace(bucket.Namespace))).To(Succeed())
+	g.Expect(jobs.Items).To(BeEmpty())
+}
+
 func TestParseEndpoint(t *testing.T) {
 	s3 := v1alpha1.S3Provider{
 		Path:     "minio-mo/test",
@@ -56,6 +115,7 @@ else
 fi
 `
 	endpoint, err := parseEntrypoint(&s3)
-	assert.Nil(t, err)
-	assert.Equal(t, expect, endpoint)
+	g := NewGomegaWithT(t)
+	g.Expect(err).To(Succeed())
+	g.Expect(endpoint).To(Equal(expect))
 }

test/e2e/bucket_test.go

@@ -1,4 +1,4 @@
-// Copyright 2025 Matrix Origin
+// Copyright 2025-2026 Matrix Origin
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -104,6 +104,107 @@ var _ = Describe("Matrix BucketClaim test", func() {
 		}, waitBucketStatusTimeout, time.Second*2).Should(Succeed())
 	})
 
+	It("Should keep S3 data when deleting a Released BucketClaim with pvc=Delete and s3=Retain", func() {
+		By("create logset with pvc=Delete and s3=Retain")
+		minioSecret := e2eutil.MinioSecret(env.Namespace)
+		Expect(kubeCli.Create(ctx, minioSecret)).To(Succeed())
+
+		minioProvider := e2eutil.MinioShareStorage(minioSecret.Name)
+		policyDelete := v1alpha1.PVCRetentionPolicyDelete
+		policyRetain := v1alpha1.PVCRetentionPolicyRetain
+		minioProvider.S3.S3RetentionPolicy = &policyRetain
+		ls := e2eutil.NewLogSetTpl(env.Namespace, fmt.Sprintf("%s:%s", moImageRepo, moVersion))
+		ls.Spec.PVCRetentionPolicy = &policyDelete
+		ls.Spec.SharedStorage = minioProvider
+		Expect(kubeCli.Create(ctx, ls)).To(Succeed())
+
+		var bucket *v1alpha1.BucketClaim
+		var err error
+		Eventually(func() error {
+			bucket, err = v1alpha1.ClaimedBucket(kubeCli, minioProvider.S3)
+			if err != nil || bucket == nil {
+				return fmt.Errorf("wait bucket creating for logset %v, %v", client.ObjectKeyFromObject(ls), err)
+			}
+			expectedStatus := v1alpha1.BucketClaimStatus{
+				BindTo: v1alpha1.BucketBindToMark(ls.ObjectMeta),
+				State:  v1alpha1.StatusInUse,
+			}
+			if !reflect.DeepEqual(expectedStatus, bucket.Status) {
+				return fmt.Errorf("bucket status is not inuse, current %v", bucket.Status)
+			}
+			return nil
+		}, waitBucketStatusTimeout, time.Second*2).Should(Succeed())
+
+		By("put object to s3 path")
+		object, err := e2eminio.PutObject(minioProvider.S3.Path)
+		Expect(err).Should(BeNil())
+		exist, err := e2eminio.IsObjectExist(object)
+		Expect(err).Should(BeNil())
+		Expect(exist).Should(BeTrue())
+
+		By("wait logset available to set any-instance-running annotation")
+		Eventually(func() error {
+			if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(ls), ls); err != nil {
+				return err
+			}
+			if len(ls.Status.AvailableStores) > 0 {
+				return nil
+			}
+			return fmt.Errorf("wait logset pod in running state")
+		}, createLogSetTimeout, pollInterval).Should(Succeed())
+		Eventually(func() error {
+			if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(bucket), bucket); err != nil {
+				return err
+			}
+			if bucket.Annotations[v1alpha1.AnnAnyInstanceRunning] == "" {
+				return fmt.Errorf("wait any-instance-running annotation on bucket")
+			}
+			return nil
+		}, waitBucketStatusTimeout, time.Second*2).Should(Succeed())
+
+		By("tear down logset cluster")
+		Expect(kubeCli.Delete(ctx, ls)).To(Succeed())
+		Eventually(func() error {
+			return waitLogSetDeleted(ls)
+		}, teardownClusterTimeout, pollInterval).Should(Succeed())
+
+		By("bucket should in released state")
+		Eventually(func() error {
+			if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(bucket), bucket); err != nil {
+				return err
+			}
+			if bucket.DeletionTimestamp != nil {
+				return fmt.Errorf("bucket should not be deleted before explicit bucketclaim deletion")
+			}
+			expectedStatus := v1alpha1.BucketClaimStatus{
+				BindTo: "",
+				State:  v1alpha1.StatusReleased,
+			}
+			if !reflect.DeepEqual(expectedStatus, bucket.Status) {
+				return fmt.Errorf("bucket status is not released, current %v", bucket.Status)
+			}
+			return nil
+		}, waitBucketStatusTimeout, time.Second*2).Should(Succeed())
+
+		By("delete released bucket claim with finalizer intact")
+		Expect(bucket.Finalizers).To(ContainElement(v1alpha1.BucketDataFinalizer))
+		Expect(kubeCli.Delete(ctx, bucket)).To(Succeed())
+
+		By("wait bucket claim been deleted")
+		Eventually(func() error {
+			err := kubeCli.Get(ctx, client.ObjectKeyFromObject(bucket), bucket)
+			if apierrors.IsNotFound(err) {
+				return nil
+			}
+			return fmt.Errorf("bucket should be deleted")
+		}, waitBucketStatusTimeout, time.Second*2).Should(Succeed())
+
+		By("s3 object should still exist")
+		exist, err = e2eminio.IsObjectExist(object)
+		Expect(err).Should(BeNil())
+		Expect(exist).Should(BeTrue())
+	})
+
 	It("Should bucket been deleted use delete retain policy", func() {
 		By("create logset cluster with minio provider")
 		minioSecret := e2eutil.MinioSecret(env.Namespace)