Document commit SHA support for GitDagBundle tracking_ref

providers/git/docs/bundles/index.rst

@@ -41,3 +41,36 @@ Example of using the GitDagBundle:
          }
      }
     ]'
+
+``tracking_ref`` accepts a branch, tag, or full commit SHA. Setting it to a commit SHA pins the
+bundle to that exact commit:
+
+.. code-block:: bash
+
+    export AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST='[
+     {
+         "name": "my-git-repo",
+         "classpath": "airflow.providers.git.bundles.git.GitDagBundle",
+         "kwargs": {
+             "repo_url": "https://github.com/org/repo.git",
+             "tracking_ref": "a3d1850dd1aa1919a61620aa39f202185c9321c0",
+             "subdir": "dags"
+         }
+     }
+    ]'
+
+Branches move as new commits are pushed, so combined with ``refresh_interval`` they pick up new code
+without a restart. Tags and commit SHAs are static (assuming tags aren't moved), pinning the bundle
+to known-good code — but changing a SHA-pinned ``tracking_ref`` is a ``dag_bundle_config_list``
+config change, not a ref move, so it only takes effect once the Dag processor is restarted and
+reloads the configuration. If ``[dag_processor] disable_bundle_versioning`` (or the
+``disable_bundle_versioning`` Dag parameter) is set, workers also resolve code from their own
+``tracking_ref`` rather than a recorded bundle version, so they need the updated configuration too.
+
+.. note::
+
+    Rolling back a SHA-pinned ``tracking_ref`` after a restart is reliable, since the commit's
+    objects are already present in the bundle's local storage. Promoting to a *new* SHA can fail
+    to check out that commit unless the bundle's local storage is cleared first (for example, a
+    fresh pod, or manually deleting the bundle's directory) — see
+    `GH-71388 <https://github.com/apache/airflow/issues/71388>`_ for the underlying limitation.

providers/git/src/airflow/providers/git/bundles/git.py

@@ -45,7 +45,7 @@ class GitDagBundle(BaseDagBundle):
     Instead of cloning the repository every time, we clone the repository once into a bare repo from the source
     and then do a clone for each version from there.
 
-    :param tracking_ref: Branch or tag for this DAG bundle
+    :param tracking_ref: Branch, tag, or commit SHA for this DAG bundle
     :param subdir: Subdirectory within the repository where the DAGs are stored (Optional)
     :param git_conn_id: Connection ID for SSH/token based connection to the repository (Optional)
     :param repo_url: Explicit Git repository URL to override the connection's host. (Optional)
@@ -212,6 +212,9 @@ def _initialize(self):
                 raise RuntimeError("Error cloning repository") from e
             except InvalidGitRepositoryError as e:
                 raise RuntimeError(f"Invalid git repository at {self.repo_path}") from e
+            # If tracking_ref was just repointed to a SHA that predates this working clone's
+            # last fetch, this checkout fails until the clone is fetched or storage is cleared;
+            # tracked at https://github.com/apache/airflow/issues/71388
             self.repo.git.checkout(self.tracking_ref)
             self._log.debug("bundle initialize", version=self.version)
             if self.version:

providers/git/tests/unit/git/bundles/test_git.py

@@ -701,6 +701,95 @@ def test_refresh_tag_moved_forward_and_backward(self, mock_githook, git_repo):
         files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
         assert {"test_dag.py"} == files_in_repo
 
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_tracking_ref_commit_sha_rollback(self, mock_githook, git_repo):
+        """Ensure tracking_ref accepts a full commit SHA, and rolling back to an
+        already-local SHA succeeds after a config-driven bundle re-creation.
+        """
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+        first_commit = repo.head.commit
+
+        file_path = repo_path / "new_test.py"
+        with open(file_path, "w") as f:
+            f.write("hello world")
+        repo.index.add([file_path])
+        second_commit = repo.index.commit("Another commit")
+
+        # Initial deploy pinned to the second commit's SHA; both commits are already
+        # fetched into local storage since the origin repo had them at clone time.
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref=second_commit.hexsha)
+        bundle.initialize()
+        assert _version_str(bundle.get_current_version()) == second_commit.hexsha
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"test_dag.py", "new_test.py"} == files_in_repo
+        assert_repo_is_closed(bundle)
+
+        # Rollback: config change re-creates the bundle pointed back at the first commit's
+        # SHA. That commit's objects are already in local storage, so it succeeds.
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref=first_commit.hexsha)
+        bundle.initialize()
+        assert _version_str(bundle.get_current_version()) == first_commit.hexsha
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"test_dag.py"} == files_in_repo
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_tracking_ref_commit_sha_promote_fails_without_clearing_storage(self, mock_githook, git_repo):
+        """A SHA created after the bundle's local storage was first populated can't be
+        promoted to in-place: the working clone never fetches it before checkout.
+
+        This documents a known limitation rather than desired behavior -- it should start
+        passing once the fix tracked at https://github.com/apache/airflow/issues/71388 lands,
+        at which point this test should be updated to assert success instead.
+        """
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+        first_commit = repo.head.commit
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref=first_commit.hexsha)
+        bundle.initialize()
+        assert _version_str(bundle.get_current_version()) == first_commit.hexsha
+        assert_repo_is_closed(bundle)
+
+        # Created after the bundle's local storage already exists.
+        file_path = repo_path / "new_test.py"
+        with open(file_path, "w") as f:
+            f.write("hello world")
+        repo.index.add([file_path])
+        second_commit = repo.index.commit("Another commit")
+
+        # Promote in-place: config change re-creates the bundle against the same local
+        # storage. The new commit's objects were never fetched into the working clone.
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref=second_commit.hexsha)
+        with pytest.raises(GitCommandError, match="reference is not a tree|unable to read tree"):
+            bundle.initialize()
+
+    @mock.patch("airflow.providers.git.bundles.git.GitHook")
+    def test_tracking_ref_commit_sha_promote_succeeds_with_fresh_storage(self, mock_githook, git_repo):
+        """Promoting a SHA-pinned tracking_ref to a new commit succeeds once local storage
+        is cleared (e.g. a fresh pod), since that re-clones from the updated bare mirror.
+        """
+        repo_path, repo = git_repo
+        mock_githook.return_value.repo_url = repo_path
+        first_commit = repo.head.commit
+
+        bundle = GitDagBundle(name="test", git_conn_id=CONN_HTTPS, tracking_ref=first_commit.hexsha)
+        bundle.initialize()
+        assert_repo_is_closed(bundle)
+
+        file_path = repo_path / "new_test.py"
+        with open(file_path, "w") as f:
+            f.write("hello world")
+        repo.index.add([file_path])
+        second_commit = repo.index.commit("Another commit")
+
+        # Different bundle name -> fresh local storage, simulating a freshly started pod.
+        bundle = GitDagBundle(name="test-fresh", git_conn_id=CONN_HTTPS, tracking_ref=second_commit.hexsha)
+        bundle.initialize()
+        assert _version_str(bundle.get_current_version()) == second_commit.hexsha
+        files_in_repo = {f.name for f in bundle.path.iterdir() if f.is_file()}
+        assert {"test_dag.py", "new_test.py"} == files_in_repo
+
     @mock.patch("airflow.providers.git.bundles.git.GitHook")
     def test_refresh_after_force_push_does_not_reclone(self, mock_githook, git_repo):
         """Refresh after force-push must fetch+reset, never clone."""