Repository: PrefectHQ/prefect
Issue category: Feature Request

Add a record to the flow run logs when an `on_rollback` hook is called
Presently, there is no way to know that a rollback happened when I am on a flow run page. We need to include a log when an `on_rollback` hook is called. It should be consistent with the log message when an `on_failure` hook is called.


Functions to edit:
src/prefect/transactions.py:Transaction.commit
src/prefect/transactions.py:Transaction.rollback

Patch:
diff --git a/src/prefect/transactions.py b/src/prefect/transactions.py
index 088a99aadb8e..610da0177216 100644
--- a/src/prefect/transactions.py
+++ b/src/prefect/transactions.py
@@ -187,8 +187,16 @@ def commit(self) -> bool:
 
             for hook in self.on_commit_hooks:
                 hook_name = _get_hook_name(hook)
+                if self.logger:
+                    self.logger.info(f"Running commit hook {hook_name!r}")
+
                 hook(self)
 
+                if self.logger:
+                    self.logger.info(
+                        f"Commit hook {hook_name!r} finished running successfully"
+                    )
+
             if self.store and self.key:
                 self.store.write(key=self.key, value=self._staged_value)
             self.state = TransactionState.COMMITTED
@@ -235,8 +243,16 @@ def rollback(self) -> bool:
         try:
             for hook in reversed(self.on_rollback_hooks):
                 hook_name = _get_hook_name(hook)
+                if self.logger:
+                    self.logger.info(f"Running rollback hook {hook_name!r}")
+
                 hook(self)
 
+                if self.logger:
+                    self.logger.info(
+                        f"Rollback hook {hook_name!r} finished running successfully"
+                    )
+
             self.state = TransactionState.ROLLED_BACK
 
             for child in reversed(self.children):

Test patch: diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index bc582191018e..02bb90bd9f0f 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -5113,6 +5113,26 @@ def txn_flow():
     assert "whoops!" in caplog.text
 
 
+def test_rollback_hook_execution_and_completion_are_logged(caplog):
+    @task
+    def foo():
+        pass
+
+    @foo.on_rollback
+    def rollback(txn):
+        pass
+
+    @flow
+    def txn_flow():
+        with transaction():
+            foo()
+            raise ValueError("txn failed")
+
+    txn_flow(return_state=True)
+    assert "Running rollback hook 'rollback'" in caplog.text
+    assert "Rollback hook 'rollback' finished running successfully" in caplog.text
+
+
 def test_commit_errors_are_logged(caplog):
     @task
     def foo():
@@ -5131,3 +5151,22 @@ def txn_flow():
     assert "An error was encountered while running commit hook" in caplog.text
     assert "RuntimeError" in caplog.text
     assert "whoops!" in caplog.text
+
+
+def test_commit_hook_execution_and_completion_are_logged(caplog):
+    @task
+    def foo():
+        pass
+
+    @foo.on_commit
+    def commit(txn):
+        pass
+
+    @flow
+    def txn_flow():
+        with transaction():
+            foo()
+
+    txn_flow(return_state=True)
+    assert "Running commit hook 'commit'" in caplog.text
+    assert "Commit hook 'commit' finished running successfully" in caplog.text