fix: apply denylist retry to pack_writer writeLog and binlog import (#48402)

### Summary
Follow-up to #48152 which applied denylist retry to parquet/json/csv
imports but missed two other paths.

- **fix(High)**: `pack_writer.go` `writeLog` now skips retry only for
non-retryable errors (permission denied, bucket not found, invalid
credentials, etc.), matching the denylist strategy in
`retryable_reader.go`.
- **fix(Medium)**: Binlog import's `WithDownloader` callbacks now use
`multiReadWithRetry`, skipping retry only for non-retryable errors.
Previously all transient failures were not retried.
- **fix(Low)**: `IsMilvusError` in `merr/utils.go` switched from
`errors.Cause` (root only) to `errors.As` (full chain traversal).

### Out of Scope
- `pack_writer_v2.go` / `pack_writer_v3.go` — same retry pattern but
different code path (multi-part upload); separate fix.
- `writeDelta` — no retry wrapper; separate concern.

issue: #48153

---------

Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

.gitignore

@@ -130,5 +130,7 @@ WARP.md
 **/.gocache/
 docs/plans/
 
+.worktrees/
+
 # agent context
-.context/
+.context/

internal/flushcommon/syncmgr/pack_writer.go

@@ -30,6 +30,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metautil"
 	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -100,8 +101,16 @@ func (bw *BulkPackWriter) writeLog(ctx context.Context, blob *storage.Blob,
 	root, p string, pack *SyncPack,
 ) (*datapb.Binlog, error) {
 	key := path.Join(bw.chunkManager.RootPath(), root, p)
-	err := retry.Do(ctx, func() error {
-		return bw.chunkManager.Write(ctx, key, blob.Value)
+	err := retry.Handle(ctx, func() (bool, error) {
+		err := bw.chunkManager.Write(ctx, key, blob.Value)
+		if err == nil {
+			return false, nil
+		}
+		err = storage.ToMilvusIoError(key, err)
+		if merr.IsNonRetryableErr(err) {
+			return false, err
+		}
+		return true, err
 	}, bw.writeRetryOpts...)
 	if err != nil {
 		return nil, err

internal/flushcommon/syncmgr/pack_writer_test.go

@@ -21,8 +21,13 @@ import (
 	"fmt"
 	"reflect"
 	"testing"
+	"time"
 
+	"github.com/cockroachdb/errors"
+	"github.com/minio/minio-go/v7"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
@@ -32,7 +37,9 @@ import (
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 )
 
 func TestBulkPackWriter_Write(t *testing.T) {
@@ -161,3 +168,53 @@ func TestBulkPackWriter_Write(t *testing.T) {
 		})
 	}
 }
+
+func TestBulkPackWriter_WriteLog_NonRetryableError(t *testing.T) {
+	paramtable.Get().Init(paramtable.NewBaseTable())
+
+	mc := metacache.NewMockMetaCache(t)
+	mc.EXPECT().Collection().Return(int64(1)).Maybe()
+
+	cm := mocks.NewChunkManager(t)
+	cm.EXPECT().RootPath().Return("files").Maybe()
+	// Return a permission-denied error — should NOT be retried
+	callCount := 0
+	cm.EXPECT().Write(mock.Anything, mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, key string, data []byte) error {
+			callCount++
+			// Simulate MinIO AccessDenied — maps to ErrIoPermissionDenied via ToMilvusIoError
+			return minio.ErrorResponse{Code: "AccessDenied"}
+		})
+
+	schema := &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{FieldID: common.RowIDField, DataType: schemapb.DataType_Int64},
+			{FieldID: common.TimeStampField, DataType: schemapb.DataType_Int64},
+			{FieldID: 100, Name: "pk", DataType: schemapb.DataType_Int64, IsPrimaryKey: true},
+			{
+				FieldID: 101, Name: "vec", DataType: schemapb.DataType_FloatVector,
+				TypeParams: []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "4"}},
+			},
+		},
+	}
+
+	bw := &BulkPackWriter{
+		metaCache:      mc,
+		schema:         schema,
+		chunkManager:   cm,
+		allocator:      allocator.NewLocalAllocator(10000, 100000),
+		writeRetryOpts: []retry.Option{retry.AttemptAlways(), retry.MaxSleepTime(10 * time.Second)},
+	}
+
+	// Use a timeout context so the test doesn't hang if retry loop is infinite
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	blob := &storage.Blob{Value: []byte("data"), RowNum: 1}
+	_, err := bw.writeLog(ctx, blob, "insert_log", "1/2/3/100/1", nil)
+	require.Error(t, err)
+	// Must stop after exactly 1 attempt — not retried
+	assert.Equal(t, 1, callCount, "non-retryable error should not be retried")
+	assert.True(t, merr.IsNonRetryableErr(err))
+	assert.True(t, errors.Is(err, merr.ErrIoPermissionDenied))
+}

internal/flushcommon/syncmgr/task_test.go

@@ -43,6 +43,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metricsinfo"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/retry"
@@ -463,7 +464,7 @@ func (s *SyncTaskSuite) TestRunError() {
 		handler := func(_ error) { flag = true }
 		s.chunkManager.ExpectedCalls = nil
 		s.chunkManager.EXPECT().RootPath().Return("files")
-		s.chunkManager.EXPECT().Write(mock.Anything, mock.Anything, mock.Anything).Return(retry.Unrecoverable(errors.New("mocked")))
+		s.chunkManager.EXPECT().Write(mock.Anything, mock.Anything, mock.Anything).Return(merr.WrapErrIoPermissionDenied("mocked-key", errors.New("mocked")))
 		task := s.getSuiteSyncTask(new(SyncPack).WithInsertData([]*storage.InsertData{s.getInsertBuffer()})).
 			WithFailureCallback(handler).
 			WithWriteRetryOptions(retry.AttemptAlways(), retry.MaxSleepTime(10*time.Second))

internal/util/importutilv2/binlog/reader.go

@@ -33,6 +33,8 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
+	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
 
@@ -44,10 +46,11 @@ type reader struct {
 	storageVersion int64
 	importEz       string
 
-	fileSize   *atomic.Int64
-	bufferSize int
-	deleteData map[any]typeutil.Timestamp // pk2ts
-	insertLogs map[int64][]string         // fieldID (or fieldGroupID if storage v2) -> binlogs
+	fileSize      *atomic.Int64
+	bufferSize    int
+	retryAttempts uint
+	deleteData    map[any]typeutil.Timestamp // pk2ts
+	insertLogs    map[int64][]string         // fieldID (or fieldGroupID if storage v2) -> binlogs
 
 	filters []Filter
 	dr      storage.DeserializeReader[*storage.Value]
@@ -83,6 +86,7 @@ func NewReader(ctx context.Context,
 		bufferSize:     bufferSize,
 		storageConfig:  storageConfig,
 		importEz:       importEz,
+		retryAttempts:  paramtable.Get().CommonCfg.StorageReadRetryAttempts.GetAsUint(),
 	}
 	err := r.init(paths, tsStart, tsEnd)
 	if err != nil {
@@ -124,7 +128,7 @@ func (r *reader) init(paths []string, tsStart, tsEnd uint64) error {
 		storage.WithVersion(r.storageVersion),
 		storage.WithBufferSize(32 * 1024 * 1024),
 		storage.WithDownloader(func(ctx context.Context, paths []string) ([][]byte, error) {
-			return r.cm.MultiRead(ctx, paths)
+			return r.multiReadWithRetry(ctx, paths)
 		}),
 		storage.WithStorageConfig(r.storageConfig),
 	}
@@ -182,7 +186,7 @@ func (r *reader) readDelete(deltaLogs []string, tsStart, tsEnd uint64) (map[any]
 	v1opts := []storage.RwOption{
 		storage.WithVersion(storage.StorageV1),
 		storage.WithDownloader(func(ctx context.Context, paths []string) ([][]byte, error) {
-			return r.cm.MultiRead(ctx, paths)
+			return r.multiReadWithRetry(ctx, paths)
 		}),
 	}
 	v2opts := []storage.RwOption{
@@ -265,6 +269,30 @@ func (r *reader) readDelete(deltaLogs []string, tsStart, tsEnd uint64) (map[any]
 	return deleteData, nil
 }
 
+// multiReadWithRetry wraps MultiRead with denylist retry: retries all errors
+// except permanent/validation ones (permission denied, bucket not found, etc.),
+// matching the strategy used by parquet/json/csv imports via RetryableReader.
+func (r *reader) multiReadWithRetry(ctx context.Context, paths []string) ([][]byte, error) {
+	var result [][]byte
+	representative := ""
+	if len(paths) > 0 {
+		representative = paths[0]
+	}
+	err := retry.Handle(ctx, func() (bool, error) {
+		var e error
+		result, e = r.cm.MultiRead(ctx, paths)
+		if e == nil {
+			return false, nil
+		}
+		e = storage.ToMilvusIoError(representative, e)
+		if merr.IsNonRetryableErr(e) {
+			return false, e
+		}
+		return true, e
+	}, retry.Attempts(r.retryAttempts))
+	return result, err
+}
+
 func (r *reader) Read() (*storage.InsertData, error) {
 	insertData, err := storage.NewInsertDataWithFunctionOutputField(r.schema)
 	if err != nil {

internal/util/importutilv2/binlog/reader_test.go

@@ -36,6 +36,7 @@ import (
 	"github.com/milvus-io/milvus/internal/util/testutil"
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/testutils"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -831,3 +832,44 @@ func (suite *ReaderSuite) TestZeroDeltaRead() {
 func TestBinlogReader(t *testing.T) {
 	suite.Run(t, new(ReaderSuite))
 }
+
+func TestMultiReadWithRetry_NonRetryableError(t *testing.T) {
+	paramtable.Init()
+	ctx := context.Background()
+
+	cm := mocks.NewChunkManager(t)
+	callCount := 0
+	cm.EXPECT().MultiRead(mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, paths []string) ([][]byte, error) {
+			callCount++
+			return nil, merr.WrapErrIoPermissionDenied("test/path", fmt.Errorf("access denied"))
+		})
+
+	r := &reader{ctx: ctx, cm: cm, retryAttempts: 3}
+	_, err := r.multiReadWithRetry(ctx, []string{"test/path"})
+	assert.Error(t, err)
+	assert.True(t, merr.IsNonRetryableErr(err))
+	assert.Equal(t, 1, callCount, "non-retryable error should not be retried")
+}
+
+func TestMultiReadWithRetry_RetryableError(t *testing.T) {
+	paramtable.Init()
+	ctx := context.Background()
+
+	cm := mocks.NewChunkManager(t)
+	callCount := 0
+	cm.EXPECT().MultiRead(mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, paths []string) ([][]byte, error) {
+			callCount++
+			if callCount < 3 {
+				return nil, merr.WrapErrIoFailed("test/path", fmt.Errorf("transient error"))
+			}
+			return [][]byte{[]byte("data")}, nil
+		})
+
+	r := &reader{ctx: ctx, cm: cm, retryAttempts: 3}
+	result, err := r.multiReadWithRetry(ctx, []string{"test/path"})
+	assert.NoError(t, err)
+	assert.Equal(t, [][]byte{[]byte("data")}, result)
+	assert.Equal(t, 3, callCount, "retryable error should be retried until success")
+}

pkg/util/merr/utils.go

@@ -92,12 +92,8 @@ func IsNonRetryableErr(err error) bool {
 }
 
 func IsMilvusError(err error) bool {
-	if err == nil {
-		return false
-	}
-	cause := errors.Cause(err)
-	_, ok := cause.(milvusError)
-	return ok
+	var me milvusError
+	return errors.As(err, &me)
 }
 
 func IsCanceledOrTimeout(err error) bool {

pkg/util/merr/utils_test.go

@@ -74,3 +74,24 @@ func TestIsNonRetryableErr_WrappedErrors(t *testing.T) {
 	wrappedRetryable := fmt.Errorf("network issue: %w", ErrIoUnexpectEOF)
 	assert.False(t, IsNonRetryableErr(wrappedRetryable))
 }
+
+func TestIsMilvusError_WrappedChain(t *testing.T) {
+	// Direct milvus error
+	assert.True(t, IsMilvusError(ErrCollectionNotFound))
+
+	// Wrapped via errors.Wrap — milvusError is root, both impls handle this
+	assert.True(t, IsMilvusError(errors.Wrap(ErrCollectionNotFound, "context")))
+
+	// Combined: plain error first, milvusError second.
+	// errors.Cause: multiErrors has no Cause() method, returns multiErrors itself — type
+	// assertion to milvusError fails. errors.As: multiErrors.Unwrap() returns errs[1] directly
+	// (ErrCollectionNotFound), so errors.As finds it.
+	// Known limitation: Combine(milvusErr, plain) is NOT covered — Unwrap exposes errs[1:] only,
+	// so milvusError at index 0 would not be found by errors.As either.
+	combined := Combine(errors.New("plain"), ErrCollectionNotFound)
+	assert.True(t, IsMilvusError(combined), "milvusError at tail of Combine chain should be detected")
+
+	// Non-milvus error
+	assert.False(t, IsMilvusError(errors.New("plain error")))
+	assert.False(t, IsMilvusError(nil))
+}