Repository: PrefectHQ/prefect
Pull request: Ensure `flow.from_source` handles remote git repository updates
This PR addresses issue #11546, where the `flow.from_source(...).serve(...)` functionality crashes when new commits are added to a remote Git repository. The problem was identified as the local clone retaining a shallow copy (`--depth 1`) of the repository, causing conflicts with new commits from the remote. This change ensures that the system handles updates from the remote repository.

There are a lot of different git pull strategies we could take; this method seems to resolve the immediate problem in the least complicated way, but I understand there are pros and cons. This strategy is basically to remove the repo and retry.

### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [x] This pull request references any related issue by including "closes `<link to issue>`"
	- If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [ ] This pull request includes tests or only affects documentation.
- [x] This pull request includes a label categorizing the change e.g. `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `netlify.toml` for files that are removed or renamed


Problem statement:
flow.from_source does not work with changes in a remote git repository
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

The flow.from_source(...).serve(...) functionality together with a remote git repository does not work when new commits are added to the remote.

### Reproduction

```python3
I have started to test prefect in various ways and one test flow running is deployed and served with a local runner with the flow.from_source(...).serve(...) functionality:

    flow.from_source(
        source="git@github.com:Relativity74205/foo.git",
        entrypoint="src/foo/flow.py:main_flow",
    ).serve(
        name="foo",
        schedule=IntervalSchedule(
            interval=timedelta(hours=1),
            anchor_date=datetime(2023, 12, 1, 0),
            timezone="Europe/Berlin",
        )
    )
```

This works fine until a commit is added to the remote. Then the prefect runner is crashing with

> subprocess.CalledProcessError: Command '['git', 'pull', 'origin', '--depth', '1']' returned non-zero exit status 128.

Adding some debug code to the prefect and anyio libraries showed the problem:
1. Once a local(?) runner has been started, the local tmp directory is fixed for the duration of the runner lifetime. That is the code is pulled always to the same temporary directory:
```
        self._tmp_dir: Path = (
            Path(tempfile.gettempdir()) / "runner_storage" / str(uuid4())
        )
```
2. The first pull_code works fine with `git clone ... --depth 1`.
3. The next pull_codes work with `git pull origin --depth 1`. However, this only works when no commits are added to origin. When a commit has been added to origin the commit histories become unrelated as the local repository only has the last commit (`--depth 1`). No default git pull strategy is working in this case. If I am not mistaken only a `git pull --rebase --allow-unrelated-histories` would work in this case.
```


### Error

```python3
The WARNING log messages are custom debug messages added to the anyio by me. pull.rebase=True pull strategy was used in this case.

23:04:43.415 | INFO    | prefect.flow_runs.runner - Opening process...
23:04:44.765 | WARNING | root - ['git', 'pull', 'origin', '--depth', '1']
23:04:44.765 | WARNING | root - output=b''
23:04:44.766 | WARNING | root - errors=b'From github.com:Relativity74205/murkelhausen-data\n + 48b047a...a2aaa21 main       -> origin/main  (forced update)\nfatal: refusing to merge unrelated histories\n'
23:04:44.766 | WARNING | root - /home/arkadius/dev/murkelhausen-data
23:14:44.766 | ERROR   | prefect.flow_runs.runner - Failed to start process for flow run '0365b5e1-1785-4c4b-87f3-d57c8189a8af'.
Traceback (most recent call last):
  File "/home/arkadius/.cache/pypoetry/virtualenvs/murkelhausen-9QDjUa50-py3.11/lib/python3.11/site-packages/prefect/runner/runner.py", line 900, in _submit_run_and_capture_errors
    status_code = await self._run_process(
                  ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/arkadius/.cache/pypoetry/virtualenvs/murkelhausen-9QDjUa50-py3.11/lib/python3.11/site-packages/prefect/runner/runner.py", line 534, in _run_process
    await storage.pull_code()
  File "/home/arkadius/.cache/pypoetry/virtualenvs/murkelhausen-9QDjUa50-py3.11/lib/python3.11/site-packages/prefect/runner/storage.py", line 221, in pull_code
    await run_process(cmd, cwd=self.destination)
  File "/home/arkadius/.cache/pypoetry/virtualenvs/murkelhausen-9QDjUa50-py3.11/lib/python3.11/site-packages/anyio/_core/_subprocesses.py", line 97, in run_process
    raise CalledProcessError(cast(int, process.returncode), command, output, errors)
subprocess.CalledProcessError: Command '['git', 'pull', 'origin', '--depth', '1']' returned non-zero exit status 128.
```


### Versions

```Text
Version:             2.14.12
API version:         0.8.4
Python version:      3.11.7
Git commit:          5a2482d0
Built:               Wed, Dec 20, 2023 4:58 PM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         cloud
```


### Additional context

_No response_

Hints from issue discussion:
Hi @Relativity74205, thanks so much for the detailed writeup! I was able to reproduce this issue. 

I adapted your example to my own public repo:
```python

from prefect import flow, deploy

from prefect.client.schemas.schedules import IntervalSchedule
from datetime import timedelta, datetime

if __name__ == "__main__":

    deploy(
        flow.from_source(
            source="https://github.com/serinamarie/demo.git",
            entrypoint="extremely_tired_flow.py:tired_flow",
        ).serve(
        name="foo",
        schedule=IntervalSchedule(
            interval=timedelta(minutes=1),
            anchor_date=datetime(2023, 12, 1, 0),
            timezone="Europe/Berlin",
        )
    )
    )
```

then while this flow was being served, I made a small change to a print stmt in the flow. It caused the same error you are seeing. I'll take a look into this.

Commit message: initial commit

Patch to review:
diff --git a/src/prefect/runner/storage.py b/src/prefect/runner/storage.py
index 02968b021074..1cf67fcf49fc 100644
--- a/src/prefect/runner/storage.py
+++ b/src/prefect/runner/storage.py
@@ -1,3 +1,4 @@
+import shutil
 import subprocess
 from copy import deepcopy
 from pathlib import Path
@@ -218,35 +219,50 @@ async def pull_code(self):
             if self._include_submodules:
                 cmd += ["--recurse-submodules"]
             cmd += ["--depth", "1"]
-            await run_process(cmd, cwd=self.destination)
+            try:
+                await run_process(cmd, cwd=self.destination)
+                self._logger.debug("Successfully pulled latest changes")
+            except subprocess.CalledProcessError as exc:
+                self._logger.error(
+                    f"Failed to pull latest changes with exit code {exc}"
+                )
+                shutil.rmtree(self.destination)
+                await self._clone_repo()
+
         else:
-            self._logger.debug("Cloning repository %s", self._url)
-            # Clone the repository if it doesn't exist at the destination
+            await self._clone_repo()
 
-            repository_url = self._repository_url_with_credentials
+    async def _clone_repo(self):
+        """
+        Clones the repository into the local destination.
+        """
+        self._logger.debug("Cloning repository %s", self._url)
 
-            cmd = [
-                "git",
-                "clone",
-                repository_url,
-            ]
-            if self._branch:
-                cmd += ["--branch", self._branch]
-            if self._include_submodules:
-                cmd += ["--recurse-submodules"]
+        repository_url = self._repository_url_with_credentials
 
-            # Limit git history and set path to clone to
-            cmd += ["--depth", "1", str(self.destination)]
+        cmd = [
+            "git",
+            "clone",
+            repository_url,
+        ]
+        if self._branch:
+            cmd += ["--branch", self._branch]
+        if self._include_submodules:
+            cmd += ["--recurse-submodules"]
 
-            try:
-                await run_process(cmd)
-            except subprocess.CalledProcessError as exc:
-                # Hide the command used to avoid leaking the access token
-                exc_chain = None if self._credentials else exc
-                raise RuntimeError(
-                    f"Failed to clone repository {self._url!r} with exit code"
-                    f" {exc.returncode}."
-                ) from exc_chain
+        # Limit git history and set path to clone to
+        cmd += ["--depth", "1", str(self.destination)]
+
+        try:
+            await run_process(cmd)
+        except subprocess.CalledProcessError as exc:
+            breakpoint()
+            # Hide the command used to avoid leaking the access token
+            exc_chain = None if self._credentials else exc
+            raise RuntimeError(
+                f"Failed to clone repository {self._url!r} with exit code"
+                f" {exc.returncode}."
+            ) from exc_chain
 
     def __eq__(self, __value) -> bool:
         if isinstance(__value, GitRepository):


Code review comments:
diff_hunk: @@ -218,35 +219,50 @@ async def pull_code(self):
             if self._include_submodules:
                 cmd += ["--recurse-submodules"]
             cmd += ["--depth", "1"]
-            await run_process(cmd, cwd=self.destination)
+            try:
+                await run_process(cmd, cwd=self.destination)
+                self._logger.debug("Successfully pulled latest changes")
+            except subprocess.CalledProcessError as exc:
+                self._logger.error(
+                    f"Failed to pull latest changes with exit code {exc}"
+                )
+                shutil.rmtree(self.destination)
+                await self._clone_repo()
+
         else:
-            self._logger.debug("Cloning repository %s", self._url)
-            # Clone the repository if it doesn't exist at the destination
+            await self._clone_repo()
 
-            repository_url = self._repository_url_with_credentials
+    async def _clone_repo(self):
+        """
+        Clones the repository into the local destination.
+        """
+        self._logger.debug("Cloning repository %s", self._url)
 
-            cmd = [
-                "git",
-                "clone",
-                repository_url,
-            ]
-            if self._branch:
-                cmd += ["--branch", self._branch]
-            if self._include_submodules:
-                cmd += ["--recurse-submodules"]
+        repository_url = self._repository_url_with_credentials
 
-            # Limit git history and set path to clone to
-            cmd += ["--depth", "1", str(self.destination)]
+        cmd = [
+            "git",
+            "clone",
+            repository_url,
+        ]
+        if self._branch:
+            cmd += ["--branch", self._branch]
+        if self._include_submodules:
+            cmd += ["--recurse-submodules"]
 
-            try:
-                await run_process(cmd)
-            except subprocess.CalledProcessError as exc:
-                # Hide the command used to avoid leaking the access token
-                exc_chain = None if self._credentials else exc
-                raise RuntimeError(
-                    f"Failed to clone repository {self._url!r} with exit code"
-                    f" {exc.returncode}."
-                ) from exc_chain
+        # Limit git history and set path to clone to
+        cmd += ["--depth", "1", str(self.destination)]
+
+        try:
+            await run_process(cmd)
+        except subprocess.CalledProcessError as exc:
+            breakpoint()
path: src/prefect/runner/storage.py
text: @author:
```suggestion
```