fix: avoid growing source panic on stale channel (#51464)

issue: #51418

- flushcommon syncmgr: skip the write-buffer panic callback for stale
growing-source meta errors while preserving failure metrics
- growing-source tests: cover stale channel and missing segment errors
without invoking the failure callback

---------

Signed-off-by: chyezh <chyezh@outlook.com>

internal/flushcommon/syncmgr/growing_source.go

@@ -564,6 +564,9 @@ func (t *GrowingSourceSyncTask) TargetOffset() int64 {
 }
 
 func (t *GrowingSourceSyncTask) HandleError(err error) {
+	if errors.IsAny(err, merr.ErrSegmentNotFound, merr.ErrChannelNotFound) {
+		return
+	}
 	if t.failureCallback != nil {
 		t.failureCallback(err)
 	}

internal/flushcommon/syncmgr/growing_source_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/errors"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
@@ -18,6 +19,7 @@ import (
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/internal/storagecommon"
 	"github.com/milvus-io/milvus/internal/storagev2/packed"
+	"github.com/milvus-io/milvus/pkg/v3/metrics"
 	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
@@ -82,6 +84,47 @@ func (s *fakeCommitGrowingFlushSource) CommitGrowingFlush(targetOffset int64) {
 	s.commits = append(s.commits, targetOffset)
 }
 
+func TestGrowingSourceSyncTaskHandleErrorSkipsFailureCallbackForStaleMetaErrors(t *testing.T) {
+	paramtable.Get().Init(paramtable.NewBaseTable())
+
+	testCases := []struct {
+		name string
+		err  error
+	}{
+		{
+			name: "channel_not_found",
+			err:  merr.WrapErrChannelNotFound("ch"),
+		},
+		{
+			name: "segment_not_found",
+			err:  merr.WrapErrSegmentNotFound(1),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			metrics.DataNodeFlushBufferCount.Reset()
+			callbackCalled := false
+			task := NewGrowingSourceSyncTask().
+				WithFailureCallback(func(error) {
+					callbackCalled = true
+					panic("failure callback should not be called")
+				})
+
+			require.NotPanics(t, func() {
+				task.HandleError(tc.err)
+				task.HandleError(tc.err)
+			})
+			require.False(t, callbackCalled)
+			require.Zero(t, testutil.ToFloat64(metrics.DataNodeFlushBufferCount.WithLabelValues(
+				paramtable.GetStringNodeID(),
+				metrics.FailLabel,
+				task.level.String(),
+			)))
+		})
+	}
+}
+
 func TestGrowingSourceSyncTaskBuildFlushConfigBM25(t *testing.T) {
 	paramtable.Get().Init(paramtable.NewBaseTable())
 	paramtable.Get().Save(paramtable.Get().DataNodeCfg.TextInlineThreshold.Key, "12345")