fix: stop flusher setup when collection observation fails (#51458)

issue: #51436

### What this PR does

- Propagates a CreateCollection RecoveryStorage observation failure from
the flusher component to WAL dispatch.
- Stops before creating a DataSyncService when the recovery state was
not recorded.
- Adds coverage for the canceled observation path, including the absence
of schema lookup.

### Verification

- `go test -v -tags dynamic,test -gcflags="all=-N -l" -count=1
./internal/streamingnode/server/flusher/flusherimpl -run
"TestWALFlusher.*CreateCollection"`
- `make static-check`
- `gofumpt -l` on changed Go files
- `git diff --check`

Signed-off-by: sijie-ni-0214 <sijie.ni@zilliz.com>

internal/streamingnode/server/flusher/flusherimpl/flusher_components.go

@@ -38,13 +38,15 @@ type flusherComponents struct {
 }
 
 // WhenCreateCollection handles the create collection message.
-func (impl *flusherComponents) WhenCreateCollection(ctx context.Context, createCollectionMsg message.ImmutableCreateCollectionMessageV1) {
+func (impl *flusherComponents) WhenCreateCollection(ctx context.Context, createCollectionMsg message.ImmutableCreateCollectionMessageV1) error {
 	// because we need to get the schema from the recovery storage, we need to observe the message at recovery storage first.
-	impl.rs.ObserveMessage(ctx, createCollectionMsg)
+	if err := impl.rs.ObserveMessage(ctx, createCollectionMsg); err != nil {
+		return err
+	}
 	if _, ok := impl.dataServices[createCollectionMsg.VChannel()]; ok {
 		impl.logger.Info(ctx, "the data sync service of current vchannel is built, skip it", mlog.FieldVChannel(createCollectionMsg.VChannel()))
 		// May repeated consumed, so we ignore the message.
-		return
+		return nil
 	}
 	if createCollectionMsg.TimeTick() <= impl.recoveryCheckPointTimeTick {
 		// It should already be recovered from the recovery storage.
@@ -54,7 +56,7 @@ func (impl *flusherComponents) WhenCreateCollection(ctx context.Context, createC
 			mlog.FieldVChannel(createCollectionMsg.VChannel()),
 			mlog.Uint64("timeTick", createCollectionMsg.TimeTick()),
 			mlog.Uint64("recoveryCheckPointTimeTick", impl.recoveryCheckPointTimeTick))
-		return
+		return nil
 	}
 
 	msgChan := make(chan *msgstream.MsgPack, 10)
@@ -101,6 +103,7 @@ func (impl *flusherComponents) WhenCreateCollection(ctx context.Context, createC
 		nil,
 	)
 	impl.addNewDataSyncService(ctx, createCollectionMsg, msgChan, ds)
+	return nil
 }
 
 // WhenDropCollection handles the drop collection message.

internal/streamingnode/server/flusher/flusherimpl/wal_flusher.go

@@ -289,7 +289,9 @@ func (impl *WALFlusherImpl) dispatch(msg message.ImmutableMessage) (err error) {
 			impl.logger.DPanic(ctx, "the message type is not CreateCollectionMessage", mlog.Err(err))
 			return nil
 		}
-		impl.flusherComponents.WhenCreateCollection(ctx, createCollectionMsg)
+		if err := impl.flusherComponents.WhenCreateCollection(ctx, createCollectionMsg); err != nil {
+			return err
+		}
 	case message.MessageTypeDropCollection:
 		// defer to remove the data sync service from the components.
 		// TODO: Current drop collection message will be handled by the underlying data sync service.

internal/streamingnode/server/flusher/flusherimpl/wal_flusher_test.go

@@ -148,6 +148,20 @@ func TestWALFlusher_DispatchDefersAckSyncUpDropCollectionObserve(t *testing.T) {
 	require.ErrorContains(t, flusher.dispatch(msg), "observe failed")
 }
 
+func TestWALFlusher_DispatchStopsCreateCollectionOnObserveError(t *testing.T) {
+	rs := mock_recovery.NewMockRecoveryStorage(t)
+	flusher := newTestWALFlusher(rs)
+
+	msg := message.CreateTestCreateCollectionMessage(t, 2, 100, rmq.NewRmqID(100)).
+		IntoImmutableMessage(rmq.NewRmqID(101))
+	rs.EXPECT().ObserveMessage(mock.Anything, mock.Anything).Return(context.Canceled).Twice()
+
+	err := flusher.dispatch(msg)
+	require.ErrorIs(t, err, context.Canceled)
+	assert.Empty(t, flusher.flusherComponents.dataServices)
+	rs.AssertNotCalled(t, "GetSchema", mock.Anything, mock.Anything, mock.Anything)
+}
+
 func TestWALFlusher_DispatchObservesAckSyncUpTruncateCollectionBeforeHandling(t *testing.T) {
 	rs := mock_recovery.NewMockRecoveryStorage(t)
 	rs.EXPECT().ObserveMessage(mock.Anything, mock.Anything).Return(errors.New("observe failed")).Once()