Enforce Component names to be unique

pkg/grpc/api_test.go

@@ -4091,7 +4091,7 @@ func TestUnlock(t *testing.T) {
 
 func TestMetadata(t *testing.T) {
 	compStore := compstore.New()
-	compStore.AddComponent(componentsV1alpha1.Component{
+	require.NoError(t, compStore.AddPendingComponentForCommit(componentsV1alpha1.Component{
 		ObjectMeta: metaV1.ObjectMeta{
 			Name: "MockComponent1Name",
 		},
@@ -4107,8 +4107,9 @@ func TestMetadata(t *testing.T) {
 				},
 			},
 		},
-	})
-	compStore.AddComponent(componentsV1alpha1.Component{
+	}))
+	require.NoError(t, compStore.CommitPendingComponent())
+	require.NoError(t, compStore.AddPendingComponentForCommit(componentsV1alpha1.Component{
 		ObjectMeta: metaV1.ObjectMeta{
 			Name: "MockComponent2Name",
 		},
@@ -4124,7 +4125,8 @@ func TestMetadata(t *testing.T) {
 				},
 			},
 		},
-	})
+	}))
+	require.NoError(t, compStore.CommitPendingComponent())
 	compStore.SetSubscriptions([]runtimePubsub.Subscription{
 		{
 			PubsubName:      "test",

pkg/grpc/universalapi/api_metadata_test.go

@@ -42,7 +42,8 @@ func TestGetMetadata(t *testing.T) {
 	})
 
 	compStore := compstore.New()
-	compStore.AddComponent(fakeComponent)
+	require.NoError(t, compStore.AddPendingComponentForCommit(fakeComponent))
+	require.NoError(t, compStore.CommitPendingComponent())
 	compStore.SetSubscriptions([]runtimePubsub.Subscription{
 		{
 			PubsubName:      "test",

pkg/http/api_test.go

@@ -1739,7 +1739,7 @@ func TestV1MetadataEndpoint(t *testing.T) {
 	fakeServer := newFakeHTTPServer()
 
 	compStore := compstore.New()
-	compStore.AddComponent(componentsV1alpha1.Component{
+	require.NoError(t, compStore.AddPendingComponentForCommit(componentsV1alpha1.Component{
 		ObjectMeta: metaV1.ObjectMeta{
 			Name: "MockComponent1Name",
 		},
@@ -1755,8 +1755,9 @@ func TestV1MetadataEndpoint(t *testing.T) {
 				},
 			},
 		},
-	})
-	compStore.AddComponent(componentsV1alpha1.Component{
+	}))
+	require.NoError(t, compStore.CommitPendingComponent())
+	require.NoError(t, compStore.AddPendingComponentForCommit(componentsV1alpha1.Component{
 		ObjectMeta: metaV1.ObjectMeta{
 			Name: "MockComponent2Name",
 		},
@@ -1772,7 +1773,8 @@ func TestV1MetadataEndpoint(t *testing.T) {
 				},
 			},
 		},
-	})
+	}))
+	require.NoError(t, compStore.CommitPendingComponent())
 	compStore.SetSubscriptions([]runtimePubsub.Subscription{
 		{
 			PubsubName:      "test",

pkg/runtime/channels/channels.go

@@ -208,7 +208,7 @@ func (c *Channels) buildHTTPPipelineForSpec(spec *config.PipelineSpec, targetPip
 	}
 
 	for _, handlerSpec := range spec.Handlers {
-		comp, exists := c.compStore.GetComponent(handlerSpec.Type, handlerSpec.Name)
+		comp, exists := c.compStore.GetComponent(handlerSpec.Name)
 		if !exists {
 			// Log the error but continue with initializing the pipeline
 			log.Errorf("couldn't find middleware component defined in configuration with name %s and type %s",

pkg/runtime/channels/channels_test.go

@@ -74,24 +74,26 @@ func TestMiddlewareBuildPipeline(t *testing.T) {
 	})
 
 	compStore := compstore.New()
-	compStore.AddComponent(componentsapi.Component{
+	require.NoError(t, compStore.AddPendingComponentForCommit(componentsapi.Component{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "mymw1",
 		},
 		Spec: componentsapi.ComponentSpec{
 			Type:    "middleware.http.fakemw",
 			Version: "v1",
 		},
-	})
-	compStore.AddComponent(componentsapi.Component{
+	}))
+	require.NoError(t, compStore.CommitPendingComponent())
+	require.NoError(t, compStore.AddPendingComponentForCommit(componentsapi.Component{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "mymw2",
 		},
 		Spec: componentsapi.ComponentSpec{
 			Type:    "middleware.http.fakemw",
 			Version: "v1",
 		},
-	})
+	}))
+	require.NoError(t, compStore.CommitPendingComponent())
 
 	t.Run("all components exists", func(t *testing.T) {
 		ch := &Channels{
@@ -135,17 +137,18 @@ func TestMiddlewareBuildPipeline(t *testing.T) {
 
 	testInitFail := func(ignoreErrors bool) func(t *testing.T) {
 		compStore := compstore.New()
-		compStore.AddComponent(componentsapi.Component{
+		require.NoError(t, compStore.AddPendingComponentForCommit(componentsapi.Component{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: "mymw",
 			},
 			Spec: componentsapi.ComponentSpec{
 				Type:    "middleware.http.fakemw",
 				Version: "v1",
 			},
-		})
+		}))
+		require.NoError(t, compStore.CommitPendingComponent())
 
-		compStore.AddComponent(componentsapi.Component{
+		require.NoError(t, compStore.AddPendingComponentForCommit(componentsapi.Component{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: "failmw",
 			},
@@ -157,7 +160,8 @@ func TestMiddlewareBuildPipeline(t *testing.T) {
 					{Name: "fail", Value: commonapi.DynamicValue{JSON: v1.JSON{Raw: []byte("true")}}},
 				},
 			},
-		})
+		}))
+		require.NoError(t, compStore.CommitPendingComponent())
 		return func(t *testing.T) {
 			ch := &Channels{
 				compStore: compStore,

pkg/runtime/compstore/components.go

@@ -13,31 +13,73 @@ limitations under the License.
 
 package compstore
 
-import compsv1alpha1 "github.com/dapr/dapr/pkg/apis/components/v1alpha1"
+import (
+	"errors"
+	"fmt"
 
-func (c *ComponentStore) GetComponent(componentType, name string) (compsv1alpha1.Component, bool) {
+	compsv1alpha1 "github.com/dapr/dapr/pkg/apis/components/v1alpha1"
+)
+
+func (c *ComponentStore) GetComponent(name string) (compsv1alpha1.Component, bool) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 	for i, comp := range c.components {
-		if comp.Spec.Type == componentType && comp.ObjectMeta.Name == name {
+		if comp.ObjectMeta.Name == name {
 			return c.components[i], true
 		}
 	}
 	return compsv1alpha1.Component{}, false
 }
 
-func (c *ComponentStore) AddComponent(component compsv1alpha1.Component) {
+func (c *ComponentStore) AddPendingComponentForCommit(component compsv1alpha1.Component) error {
+	c.compPendingLock.Lock()
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
-	for i, comp := range c.components {
-		if comp.Spec.Type == component.Spec.Type && comp.ObjectMeta.Name == component.Name {
-			c.components[i] = component
-			return
+	if c.compPending != nil {
+		c.compPendingLock.Unlock()
+		return errors.New("pending component not yet committed")
+	}
+
+	for _, existing := range c.components {
+		if existing.Name == component.Name {
+			c.compPendingLock.Unlock()
+			return fmt.Errorf("component %s already exists", existing.Name)
 		}
 	}
 
-	c.components = append(c.components, component)
+	c.compPending = &component
+
+	return nil
+}
+
+func (c *ComponentStore) DropPendingComponent() error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.compPending == nil {
+		return errors.New("no pending component to drop")
+	}
+
+	c.compPending = nil
+	c.compPendingLock.Unlock()
+
+	return nil
+}
+
+func (c *ComponentStore) CommitPendingComponent() error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if c.compPending == nil {
+		return errors.New("no pending component to commit")
+	}
+
+	c.components = append(c.components, *c.compPending)
+	c.compPending = nil
+	c.compPendingLock.Unlock()
+
+	return nil
 }
 
 func (c *ComponentStore) ListComponents() []compsv1alpha1.Component {
@@ -48,12 +90,12 @@ func (c *ComponentStore) ListComponents() []compsv1alpha1.Component {
 	return comps
 }
 
-func (c *ComponentStore) DeleteComponent(compType, name string) {
+func (c *ComponentStore) DeleteComponent(name string) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
 	for i, comp := range c.components {
-		if comp.Spec.Type == compType && comp.ObjectMeta.Name == name {
+		if comp.ObjectMeta.Name == name {
 			c.components = append(c.components[:i], c.components[i+1:]...)
 			return
 		}

pkg/runtime/compstore/compstore.go

@@ -51,6 +51,9 @@ type ComponentStore struct {
 	components              []compsv1alpha1.Component
 	subscriptions           []rtpubsub.Subscription
 	httpEndpoints           []httpEndpointV1alpha1.HTTPEndpoint
+
+	compPendingLock sync.Mutex
+	compPending     *compsv1alpha1.Component
 }
 
 func New() *ComponentStore {

pkg/runtime/processor/binding/init_test.go

@@ -104,7 +104,7 @@ func TestInitBindings(t *testing.T) {
 		assert.NoError(t, err)
 
 		output := compapi.Component{}
-		output.ObjectMeta.Name = "testinput"
+		output.ObjectMeta.Name = "testoutput"
 		output.Spec.Type = "bindings.testoutput"
 		err = proc.Init(context.TODO(), output)
 		assert.NoError(t, err)

pkg/runtime/processor/binding/send_test.go

@@ -153,7 +153,7 @@ func TestStartReadingFromBindings(t *testing.T) {
 		}
 
 		b.compStore.AddInputBinding("test", m)
-		b.compStore.AddComponent(componentsV1alpha1.Component{
+		require.NoError(t, b.compStore.AddPendingComponentForCommit(componentsV1alpha1.Component{
 			ObjectMeta: metav1.ObjectMeta{
 				Name: "test",
 			},
@@ -168,7 +168,8 @@ func TestStartReadingFromBindings(t *testing.T) {
 					},
 				},
 			},
-		})
+		}))
+		require.NoError(t, b.compStore.CommitPendingComponent())
 		err := b.StartReadingFromBindings(context.Background())
 
 		assert.NoError(t, err)

pkg/runtime/processor/processor.go

@@ -15,6 +15,7 @@
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"sync"
@@ -216,13 +217,15 @@
 		return err
 	}
 
-	if err := m.Init(ctx, comp); err != nil {
+	if err := p.compStore.AddPendingComponentForCommit(comp); err != nil {
 		return err
 	}
 
-	p.compStore.AddComponent(comp)
+	if err := m.Init(ctx, comp); err != nil {
+		return errors.Join(err, p.compStore.DropPendingComponent())
+	}
 
-	return nil
+	return p.compStore.CommitPendingComponent()
 }
 
 // Close closes the component.
@@ -239,7 +242,7 @@
 		return err
 	}
 
-	p.compStore.DeleteComponent(comp.Spec.Type, comp.Name)
+	p.compStore.DeleteComponent(comp.Name)
 
 	return nil
 }