[adapters+py] Improve transaction status reporting.

Leonid Ryzhyk · ryzhyk · commit 669d3fbfe012 · 2025-09-06T17:31:58.000Z
* Expose fewer transaction states via the API by merging
  TransactionRequested into TransactionInProgress and CommitRequested
  into CommitInProgress.

* Add Python API to retrieve transaction status.

Signed-off-by: Leonid Ryzhyk &lt;leonid@feldera.com&gt;
diff --git a/crates/adapters/src/controller/mod.rs b/crates/adapters/src/controller/mod.rs
@@ -4215,7 +4215,7 @@ impl ControllerInner {
                 self.status
                     .global_metrics
                     .transaction_status
-                    .store(TransactionStatus::None, Ordering::Release);
+                    .store(TransactionStatus::NoTransaction, Ordering::Release);
             }
             (TransactionState::Started(tid), None) => {
                 assert_ne!(tid, 0);
@@ -4227,7 +4227,7 @@ impl ControllerInner {
                 self.status
                     .global_metrics
                     .transaction_status
-                    .store(TransactionStatus::CommitRequested, Ordering::Release);
+                    .store(TransactionStatus::CommitInProgress, Ordering::Release);
             }
             (TransactionState::Committing(tid), None) => {
                 assert_ne!(tid, 0);
@@ -4251,7 +4251,7 @@ impl ControllerInner {
                 self.status
                     .global_metrics
                     .transaction_status
-                    .store(TransactionStatus::TransactionStarting, Ordering::Release);
+                    .store(TransactionStatus::TransactionInProgress, Ordering::Release);
             }
             (TransactionState::Started(tid), Some(tid2)) => {
                 assert_ne!(tid, 0);
diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -126,10 +126,8 @@ impl CompletionToken {
 #[repr(u8)]
 pub enum TransactionStatus {
     #[default]
-    None,
-    TransactionStarting,
+    NoTransaction,
     TransactionInProgress,
-    CommitRequested,
     CommitInProgress,
 }
 
@@ -273,7 +271,7 @@ impl GlobalControllerMetrics {
             state: Atomic::new(PipelineState::Paused),
             bootstrap_in_progress: AtomicBool::new(false),
             transaction_id: Atomic::new(0),
-            transaction_status: Atomic::new(TransactionStatus::None),
+            transaction_status: Atomic::new(TransactionStatus::NoTransaction),
             rss_bytes: AtomicU64::new(0),
             cpu_msecs: AtomicU64::new(0),
             uptime_msecs: AtomicU64::new(0),
@@ -676,7 +674,7 @@ impl ControllerStatus {
         self.global_metrics
             .transaction_status
             .load(Ordering::Acquire)
-            != TransactionStatus::None
+            != TransactionStatus::NoTransaction
     }
 
     pub fn request_step(&self, circuit_thread_unparker: &Unparker) {
diff --git a/python/feldera/enums.py b/python/feldera/enums.py
@@ -170,6 +170,33 @@ def __eq__(self, other):
         return self.value == other.value
 
 
+class TransactionStatus(Enum):
+    """
+    Represents the transaction handling status of a pipeline.
+    """
+
+    NoTransaction = 1
+    """There is currently no active transaction."""
+
+    TransactionInProgress = 2
+    """There is an active transaction in progress."""
+
+    CommitInProgress = 3
+    """A commit is currently in progress."""
+
+    @staticmethod
+    def from_str(value):
+        for member in TransactionStatus:
+            if member.name.lower() == value.lower():
+                return member
+        raise ValueError(
+            f"Unknown value '{value}' for enum {TransactionStatus.__name__}"
+        )
+
+    def __eq__(self, other):
+        return self.value == other.value
+
+
 class ProgramStatus(Enum):
     Pending = 1
     CompilingSql = 2
diff --git a/python/feldera/pipeline.py b/python/feldera/pipeline.py
@@ -11,7 +11,12 @@
 from queue import Queue
 
 from feldera.rest.errors import FelderaAPIError
-from feldera.enums import PipelineStatus, ProgramStatus, CheckpointStatus
+from feldera.enums import (
+    PipelineStatus,
+    ProgramStatus,
+    CheckpointStatus,
+    TransactionStatus,
+)
 from feldera.enums import StorageStatus
 from feldera.rest.pipeline import Pipeline as InnerPipeline
 from feldera.rest.feldera_client import FelderaClient
@@ -563,8 +568,10 @@ def start_transaction(self) -> int:
         """
         Start a new transaction.
 
-        Returns:
-            Transaction ID.
+        :return: Transaction ID.
+
+        :raises FelderaAPIError: If the pipeline fails to start a transaction, e.g., if the pipeline is not running or
+            there is already an active transaction.
         """
 
         return self.client.start_transaction(self.name)
@@ -576,7 +583,7 @@ def commit_transaction(
         timeout_s: Optional[float] = None,
     ):
         """
-        Commits the currently active transaction.
+        Commit the currently active transaction.
 
         :param transaction_id: If provided, the function verifies that the currently active transaction matches this ID.
             If the active transaction ID does not match, the function raises an error.
@@ -590,11 +597,36 @@ def commit_transaction(
         :raises RuntimeError: If there is currently no transaction in progress.
         :raises ValueError: If the provided `transaction_id` does not match the current transaction.
         :raises TimeoutError: If the transaction does not commit within the specified timeout (when `wait` is True).
-        :raises FelderaAPIError: If the pipeline fails to start a transaction.
+        :raises FelderaAPIError: If the pipeline fails to commit a transaction.
         """
 
         self.client.commit_transaction(self.name, transaction_id, wait, timeout_s)
 
+    def transaction_status(self) -> TransactionStatus:
+        """
+        Get pipeline's transaction handling status.
+
+        :return: Current transaction handling status of the pipeline.
+
+        :raises FelderaAPIError: If pipeline's status couldn't be read, e.g., because the pipeline is not currently running.
+        """
+
+        return self.stats().global_metrics.transaction_status
+
+    def transaction_id(self) -> Optional[int]:
+        """
+        Gets the ID of the currently active transaction or None if there is no active transaction.
+
+        :return: The ID of the transaction.
+        """
+
+        transaction_id = self.stats().global_metrics.transaction_id
+
+        if transaction_id == 0:
+            return None
+        else:
+            return transaction_id
+
     def delete(self, clear_storage: bool = False):
         """
         Deletes the pipeline.
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
@@ -565,7 +565,7 @@ def commit_transaction(
         :raises RuntimeError: If there is currently no transaction in progress.
         :raises ValueError: If the provided `transaction_id` does not match the current transaction.
         :raises TimeoutError: If the transaction does not commit within the specified timeout (when `wait` is True).
-        :raises FelderaAPIError: If the pipeline fails to start a transaction.
+        :raises FelderaAPIError: If the pipeline fails to commit a transaction.
         """
 
         # TODO: implement this without using /stats when we have a better pipeline status reporting API.
diff --git a/python/feldera/stats.py b/python/feldera/stats.py
@@ -1,5 +1,5 @@
 from typing import Mapping, Any, Optional, List
-from feldera.enums import PipelineStatus
+from feldera.enums import PipelineStatus, TransactionStatus
 from datetime import datetime
 import uuid
 
@@ -55,6 +55,8 @@ def __init__(self):
         self.total_processed_records: Optional[int] = None
         self.total_completed_records: Optional[int] = None
         self.pipeline_complete: Optional[bool] = None
+        self.transaction_status: Optional[TransactionStatus] = None
+        self.transaction_id: Optional[int] = None
 
     @classmethod
     def from_dict(cls, d: Mapping[str, Any]):
@@ -63,6 +65,7 @@ def from_dict(cls, d: Mapping[str, Any]):
         metrics.state = PipelineStatus.from_str(d["state"])
         metrics.incarnation_uuid = uuid.UUID(d["incarnation_uuid"])
         metrics.start_time = datetime.fromtimestamp(d["start_time"])
+        metrics.transaction_status = TransactionStatus.from_str(d["transaction_status"])
         return metrics
 
 
