[Embedded] Improve startup diagnosis & error msg for different ports

integration_embedded/test_client.py

@@ -180,6 +180,32 @@ def test_embedded_startup_with_blocked_grpc_port(tmp_path_factory: pytest.TempPa
         client.close()
 
 
+def test_embedded_startup_with_data_from_other_port(
+    tmp_path_factory: pytest.TempPathFactory,
+) -> None:
+    data_path = tmp_path_factory.mktemp("data")
+    binary_path = tmp_path_factory.mktemp("bin")
+    client = weaviate.connect_to_embedded(
+        port=8180,
+        grpc_port=50180,
+        environment_variables={"DISABLE_TELEMETRY": "true"},
+        persistence_data_path=data_path,
+        binary_path=binary_path,
+    )
+    client.close()
+
+    # The Raft state saved on port 8180 does not list the node on port 8181 as a voter,
+    # so the new node never elects a leader and the readiness check times out.
+    with pytest.raises(weaviate.exceptions.WeaviateStartUpError, match="persistence_data_path"):
+        weaviate.connect_to_embedded(
+            port=8181,
+            grpc_port=50181,
+            environment_variables={"DISABLE_TELEMETRY": "true"},
+            persistence_data_path=data_path,
+            binary_path=binary_path,
+        )
+
+
 def test_embedded_create_restore_backup(tmp_path_factory: pytest.TempPathFactory) -> None:
     num_objects = 10
     collection_name = "BackedUp"

weaviate/connect/helpers.py

@@ -263,6 +263,7 @@ def connect_to_embedded(
         additional_config: This includes many additional, rarely used config options. use wvc.init.AdditionalConfig() to configure.
         version: Weaviate version to be used for the embedded instance.
         persistence_data_path: Directory where the files making up the database are stored.
+            Data in this directory belongs to the `port` that created it, so each separate instance needs its own directory.
             When the XDG_DATA_HOME env variable is set, the default value is: `XDG_DATA_HOME/weaviate/`
             Otherwise it is: `~/.local/share/weaviate`
         binary_path: Directory where to download the binary. If deleted, the client will download the binary again.
@@ -568,6 +569,7 @@ def use_async_with_embedded(
         additional_config: This includes many additional, rarely used config options. use wvc.init.AdditionalConfig() to configure.
         version: Weaviate version to be used for the embedded instance.
         persistence_data_path: Directory where the files making up the database are stored.
+            Data in this directory belongs to the `port` that created it, so each separate instance needs its own directory.
             When the XDG_DATA_HOME env variable is set, the default value is: `XDG_DATA_HOME/weaviate/`
             Otherwise it is: `~/.local/share/weaviate`
         binary_path: Directory where to download the binary. If deleted, the client will download the binary again.

weaviate/connect/v4.py

@@ -983,7 +983,10 @@ def connect(self, force: bool = False) -> None:
             except WeaviateStartUpError as e:
                 self.embedded_db.stop()
                 self._connected = False
-                raise e
+                hint = self.embedded_db.existing_data_hint()
+                if hint is None:
+                    raise e
+                raise WeaviateStartUpError(f"{e.message} {hint}") from e
 
         # do it after all other init checks so as not to break all the tests
         if self._weaviate_version.is_lower_than(1, 27, patch=0):
@@ -1187,7 +1190,10 @@ async def connect(self, force: bool = False) -> None:
             except WeaviateStartUpError as e:
                 self.embedded_db.stop()
                 self._connected = False
-                raise e
+                hint = self.embedded_db.existing_data_hint()
+                if hint is None:
+                    raise e
+                raise WeaviateStartUpError(f"{e.message} {hint}") from e
 
         # do it after all other init checks so as not to break all the tests
         if self._weaviate_version.is_lower_than(1, 27, 0):

weaviate/embedded.py

@@ -92,6 +92,8 @@ def __init__(self, options: EmbeddedOptions) -> None:
         self.options = options
         self.grpc_port: int = options.grpc_port
         self.process: Optional[subprocess.Popen[bytes]] = None
+        # Data path that held Raft state before the most recent launch, set by `start()`.
+        self._existing_data_path: Optional[str] = None
         self.ensure_paths_exist()
         self.check_supported_platform()
         self._parsed_weaviate_version = ""
@@ -286,6 +288,12 @@ def start(self) -> None:
         if self.options.additional_env_vars is not None:
             my_env.update(self.options.additional_env_vars)
 
+        # Read the path from the final environment, which may override the option. Check before
+        # launching, because Weaviate creates `raft/` during its own startup.
+        data_path = my_env["PERSISTENCE_DATA_PATH"]
+        has_raft_state = os.path.isdir(os.path.join(data_path, "raft"))
+        self._existing_data_path = data_path if has_raft_state else None
+
         # filter warning about running processes.
         with warnings.catch_warnings():
             warnings.simplefilter("ignore", ResourceWarning)
@@ -307,6 +315,23 @@ def start(self) -> None:
         logger.info(f"Started {self.options.binary_path}: process ID {self.process.pid}")
         self.wait_till_listening()
 
+    def existing_data_hint(self) -> Optional[str]:
+        """Return a hint to append to a startup timeout, or None if the hint does not apply.
+
+        The hint applies when the data path held data before the most recent `start()`. Weaviate
+        uses the HTTP port as its node name, so data created with a different `port` blocks
+        startup. The client cannot tell which port created the data, so the hint names it as a
+        possible cause.
+        """
+        if self._existing_data_path is None:
+            return None
+        return (
+            f"The data path `{self._existing_data_path}` already holds data, which may have been "
+            "created with a different `port`. To run a separate instance, pass a different "
+            "`persistence_data_path` and ports. To use the existing data, pass the `port` it was "
+            "created with."
+        )
+
     @abstractmethod
     def is_listening(self) -> bool:
         raise NotImplementedError()