Repository: deepset-ai/haystack

Issue:
Auto-connect between pipeline components intermittently failing
**Describe the bug**
The connections between components when using the `.connect()` method of a pipeline are non-deterministic. 

We did find that *explicitly* defining the connection (as showcased in the documentation found [here](https://docs.haystack.deepset.ai/v2.0/docs/transformerssimilarityranker)) worked to resolve this issue, however it seems like it might be possible to ensure that if there *are* default connections that share a name, they be preferentially connected during graph construction.

Hopefully all this makes sense - if not, I'm happy to follow-up!

**Error message**
There is no specific error.

**Expected behavior**
The logic should not connect two components by incompatible IDs or should not default to a random selection/decision.

**Additional context**
Creating connections like this will lead to a semi-random incidence of the pipeline graph being constructed incorrectly:

```python
pipeline.connect("tgi_fetcher", "tgi_converter")
pipeline.connect("tgi_converter", "tgi_splitter")
pipeline.connect("tgi_splitter", "tgi_ranker")
pipeline.connect("tgi_ranker", "tgi_prompt_builder")
pipeline.connect("tgi_prompt_builder", "tgi_llm")
```

![image](https://i.imgur.com/zUBwJaB.png)

As you can hopefully see, the output of `ranker` (documents) is being connected to the `question` input of the `prompt_builder`. This is despite the fact that the `prompt_builder` has a `documents` input. 

You can get around the error by keying `documents` in your input - and swapping any potential prompt-templates, but this seems like undesirable behaviour.

```python
pipeline.connect("tgi_fetcher", "tgi_converter")
pipeline.connect("tgi_converter", "tgi_splitter")
pipeline.connect("tgi_splitter", "tgi_ranker")
pipeline.connect("tgi_ranker.documents", "tgi_prompt_builder.documents")
pipeline.connect("tgi_prompt_builder", "tgi_llm")
```

**To Reproduce**

```python
prompt_template = """
According to these documents:

{% for doc in documents %}
  {{ doc.content }}
{% endfor %}

Answer the given question: {{question}}
Answer:
"""
prompt_builder = PromptBuilder(template=prompt_template)
fetcher = LinkContentFetcher()
converter = HTMLToDocument()
splitter = DocumentSplitter(split_length=100, split_overlap=5)
llm = GPTGenerator(api_key = openai_api_key, model_name = "gpt-4")
ranker = TransformersSimilarityRanker()

pipeline = Pipeline()
pipeline.add_component(name="fetcher", instance=fetcher)
pipeline.add_component(name="converter", instance=converter)
pipeline.add_component(name="splitter", instance=splitter)
pipeline.add_component(name="ranker", instance=ranker)
pipeline.add_component(name="prompt_builder", instance=prompt_builder)
pipeline.add_component(name="llm", instance=llm)

pipeline.connect("fetcher", "converter")
pipeline.connect("converter", "splitter")
pipeline.connect("splitter", "ranker")
pipeline.connect("ranker", "prompt_builder")
pipeline.connect("prompt_builder", "llm")

question = "What is our favorite animal?"
result = pipeline.run({"prompt_builder": {"question": question},
                       "ranker": {"query": question},
                      "fetcher": {"urls": ["https://haystack.deepset.ai/advent-of-haystack/day-1#challenge"]}})

print(result['llm']['replies'][0])
```

**FAQ Check**
- [x ] Have you had a look at [our new FAQ page](https://docs.haystack.deepset.ai/docs/faq)?

**System:**
 - OS: Linux
 - GPU/CPU: CPU
 - Haystack version (commit or version number): 2.0.0b2
 - DocumentStore: N/A
 - Reader: N/A
 - Retriever: N/A


Discussion:

Patch:
diff --git a/haystack/components/writers/document_writer.py b/haystack/components/writers/document_writer.py
index f4debb17..170ee24d 100644
--- a/haystack/components/writers/document_writer.py
+++ b/haystack/components/writers/document_writer.py
@@ -15,11 +15,15 @@ class DocumentWriter:
     A component for writing documents to a DocumentStore.
     """
 
-    def __init__(self, document_store: DocumentStore, policy: DuplicatePolicy = DuplicatePolicy.FAIL):
+    def __init__(self, document_store: DocumentStore, policy: DuplicatePolicy = DuplicatePolicy.NONE):
         """
         Create a DocumentWriter component.
 
-        :param policy: The policy to use when encountering duplicate documents (default is DuplicatePolicy.FAIL).
+        :param policy: the policy to apply when a Document with the same id already exists in the DocumentStore.
+            - `DuplicatePolicy.NONE`: Default policy, behaviour depends on the Document Store.
+            - `DuplicatePolicy.SKIP`: If a Document with the same id already exists, it is skipped and not written.
+            - `DuplicatePolicy.OVERWRITE`: If a Document with the same id already exists, it is overwritten.
+            - `DuplicatePolicy.FAIL`: If a Document with the same id already exists, an error is raised.
         """
         self.document_store = document_store
         self.policy = policy
diff --git a/haystack/core/component/connection.py b/haystack/core/component/connection.py
index c84c9b5b..c3118aa4 100644
--- a/haystack/core/component/connection.py
+++ b/haystack/core/component/connection.py
@@ -116,21 +116,24 @@ class Connection:
             name_matches = [
                 (out_sock, in_sock) for out_sock, in_sock in possible_connections if in_sock.name == out_sock.name
             ]
-            if len(name_matches) != 1:
-                # TODO allow for multiple connections at once if there is no ambiguity?
-                # TODO give priority to sockets that have no default values?
-                connections_status_str = _connections_status(
-                    sender_node=sender_node,
-                    sender_sockets=sender_sockets,
-                    receiver_node=receiver_node,
-                    receiver_sockets=receiver_sockets,
-                )
-                raise PipelineConnectError(
-                    f"Cannot connect '{sender_node}' with '{receiver_node}': more than one connection is possible "
-                    "between these components. Please specify the connection name, like: "
-                    f"pipeline.connect('{sender_node}.{possible_connections[0][0].name}', "
-                    f"'{receiver_node}.{possible_connections[0][1].name}').\n{connections_status_str}"
-                )
+            if len(name_matches) == 1:
+                # Sockets match by type and name, let's use this
+                return Connection(sender_node, name_matches[0][0], receiver_node, name_matches[0][1])
+
+            # TODO allow for multiple connections at once if there is no ambiguity?
+            # TODO give priority to sockets that have no default values?
+            connections_status_str = _connections_status(
+                sender_node=sender_node,
+                sender_sockets=sender_sockets,
+                receiver_node=receiver_node,
+                receiver_sockets=receiver_sockets,
+            )
+            raise PipelineConnectError(
+                f"Cannot connect '{sender_node}' with '{receiver_node}': more than one connection is possible "
+                "between these components. Please specify the connection name, like: "
+                f"pipeline.connect('{sender_node}.{possible_connections[0][0].name}', "
+                f"'{receiver_node}.{possible_connections[0][1].name}').\n{connections_status_str}"
+            )
 
         match = possible_connections[0]
         return Connection(sender_node, match[0], receiver_node, match[1])
diff --git a/releasenotes/notes/document-writer-default-policy-693027781629fc73.yaml b/releasenotes/notes/document-writer-default-policy-693027781629fc73.yaml
new file mode 100644
index 00000000..b90629fe
--- /dev/null
+++ b/releasenotes/notes/document-writer-default-policy-693027781629fc73.yaml
@@ -0,0 +1,6 @@
+---
+enhancements:
+  - |
+    Change `DocumentWriter` default `policy` from `DuplicatePolicy.FAIL` to `DuplicatePolicy.NONE`.
+    The `DocumentStore` protocol uses the same default so that different Document Stores can choose
+    the default policy that better fit.
diff --git a/releasenotes/notes/fix-connect-with-same-name-5ce470f7f0451362.yaml b/releasenotes/notes/fix-connect-with-same-name-5ce470f7f0451362.yaml
new file mode 100644
index 00000000..529cd69a
--- /dev/null
+++ b/releasenotes/notes/fix-connect-with-same-name-5ce470f7f0451362.yaml
@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Fix `Pipeline.connect()` so it connects sockets with same name if multiple sockets with compatible types are found.


Test patch:
diff --git a/test/components/writers/test_document_writer.py b/test/components/writers/test_document_writer.py
index 9858538f..623d5ee5 100644
--- a/test/components/writers/test_document_writer.py
+++ b/test/components/writers/test_document_writer.py
@@ -16,7 +16,7 @@ class TestDocumentWriter:
             "type": "haystack.components.writers.document_writer.DocumentWriter",
             "init_parameters": {
                 "document_store": {"type": "haystack.testing.factory.MockedDocumentStore", "init_parameters": {}},
-                "policy": "FAIL",
+                "policy": "NONE",
             },
         }
 
diff --git a/test/core/pipeline/test_connections.py b/test/core/pipeline/test_connections.py
index ba4d70ca..a28a038a 100644
--- a/test/core/pipeline/test_connections.py
+++ b/test/core/pipeline/test_connections.py
@@ -386,3 +386,18 @@ def test_parse_connection():
     assert parse_connect_string("foobar") == ("foobar", None)
     assert parse_connect_string("foo.bar") == ("foo", "bar")
     assert parse_connect_string("foo.bar.baz") == ("foo", "bar.baz")
+
+
+def test_connect_with_same_socket_names():
+    SimpleComponent = factory.component_class("SimpleComponent", output_types={"documents": List})
+    ComponentWithMultipleInputs = factory.component_class(
+        "ComponentWithMultipleInputs", input_types={"question": Any, "documents": Any}
+    )
+
+    pipe = Pipeline()
+    pipe.add_component("simple", SimpleComponent())
+    pipe.add_component("multiple", ComponentWithMultipleInputs())
+
+    pipe.connect("simple", "multiple")
+
+    assert list(pipe.graph.edges) == [("simple", "multiple", "documents/documents")]