Implement WaitForTx RPC method

.changes/20260921_cardano_rpc_chain_follower_payload.yml

@@ -0,0 +1,8 @@
+project: cardano-rpc
+pr: 1351
+kind:
+  - breaking
+description: |
+  `ChainChange` and `ChainFollower` in `Cardano.Rpc.Server.NodeKernelAccess` now take the type of the per-block payload as a parameter. `withFollower` keeps giving the raw block bytes together with the parsed block, as `ChainFollower (ByteString, BlockInMode)`, and the new `withBlockFollower` gives only the parsed block, as `ChainFollower BlockInMode`. Code that names the old unparameterised `ChainChange` or `ChainFollower` types must add the payload type. The follower API now lives in the internal module `Cardano.Rpc.Server.NodeKernelAccess.Internal.Follower` and is re-exported from `Cardano.Rpc.Server.NodeKernelAccess`.
+
+  `ChainFollower` also has a new `tryNextChange` field that returns the next change without blocking, and `NodeKernelAccess` now also exports the STM reads `readChainTipPoint`, `readMempoolSnapshot` and `readCurrentEra`, so callers can wait for the chain tip or mempool to change. The new `positionFollowerAtCurrentTip` moves a freshly opened follower to the current chain tip and returns that point.

.changes/20260921_cardano_rpc_wait_for_tx.yml

@@ -0,0 +1,6 @@
+project: cardano-rpc
+pr: 1351
+kind:
+  - feature
+description: |
+  Add the `WaitForTx` method to the UTxO RPC submit service, which streams the progress of the given transactions until they are confirmed on chain. A `WaitForTx` stream that sends no update for 10 minutes ends with `DEADLINE_EXCEEDED`. Clients can reopen it, but the new stream does not report confirmations that occurred before it opened. A request with more than 500 transaction references is rejected with `INVALID_ARGUMENT`.

.changes/20261021_cardano_rpc_watch_mempool_ticket_cursor.yml

@@ -0,0 +1,6 @@
+project: cardano-rpc
+pr: 1351
+kind:
+  - breaking
+description: |
+  `MempoolWatchSnapshot` in `Cardano.Rpc.Server.NodeKernelAccess` no longer has the `mempoolWatchTicketNumbers` and `mempoolWatchSlotNo` fields; only `mempoolWatchTxsAfter` remains, so read new entries through it. `nextMempoolWatchSnapshot` and the next-snapshot callback of `watchMempoolStream` now take the last seen `TicketNo` instead of the previous snapshot, and return once the mempool holds an entry newer than that ticket. A waiting `WatchMempool` stream therefore wakes only when a transaction newer than the last one it sent arrives, so removals and slot-only changes no longer wake it. Its messages are unchanged.

cardano-rpc/README.md

@@ -214,7 +214,7 @@ A predicate naming more than 20,000 address terms (repeated addresses count) is
 |--------|--------|
 | [SubmitTx](https://utxorpc.org/submit/spec/#submittx) | ✅ Supported |
 | [EvalTx](https://utxorpc.org/submit/spec/#evaltx) | ✅ Supported |
-| [WaitForTx](https://utxorpc.org/submit/spec/#waitfortx) | ⬜ Not supported |
+| [WaitForTx](https://utxorpc.org/submit/spec/#waitfortx) | ✅ Supported |
 | [ReadMempool](https://utxorpc.org/submit/spec/#readmempool) | ✅ Supported |
 | [WatchMempool](https://utxorpc.org/submit/spec/#watchmempool) | ✅ Supported |
 

cardano-rpc/cardano-rpc.cabal

@@ -87,7 +87,9 @@ library
     Cardano.Rpc.Server.Internal.UtxoRpc.Type.Tx
     Cardano.Rpc.Server.Internal.UtxoRpc.Type.TxEval
     Cardano.Rpc.Server.Internal.UtxoRpc.Type.TxOutput
+    Cardano.Rpc.Server.Internal.UtxoRpc.WaitForTx
     Cardano.Rpc.Server.NodeKernelAccess
+    Cardano.Rpc.Server.NodeKernelAccess.Internal.Follower
 
   other-modules:
     Cardano.Rpc.Server.Internal.Orphans
@@ -244,4 +246,5 @@ test-suite cardano-rpc-test
     Test.Cardano.Rpc.Script
     Test.Cardano.Rpc.TxOutput
     Test.Cardano.Rpc.Type
+    Test.Cardano.Rpc.WaitForTxStream
     Test.Cardano.Rpc.WatchMempoolStream

cardano-rpc/src/Cardano/Rpc/Server.hs

@@ -49,6 +49,7 @@ import Cardano.Rpc.Server.Internal.UtxoRpc.Mempool
 import Cardano.Rpc.Server.Internal.UtxoRpc.Query
 import Cardano.Rpc.Server.Internal.UtxoRpc.Submit
 import Cardano.Rpc.Server.Internal.UtxoRpc.Sync
+import Cardano.Rpc.Server.Internal.UtxoRpc.WaitForTx
 import Cardano.Rpc.Server.NodeKernelAccess
   ( NodeKernelAccess
   , mkNodeKernelAccess
@@ -100,7 +101,7 @@ methodsUtxoRpcSubmit =
   Method (mkNonStreaming $ wrapInSpan TraceRpcEvalTxSpan . evalTxMethod)
     . Method (mkNonStreaming $ wrapInSpan TraceRpcReadMempoolSpan . readMempoolMethod)
     . Method (mkNonStreaming $ wrapInSpan TraceRpcSubmitSpan . submitTxMethod)
-    . UnsupportedMethod -- waitForTx
+    . Method (mkServerStreaming $ \req -> wrapInSpan TraceRpcWaitForTxSpan . waitForTxMethod req)
     . Method (mkServerStreaming $ \req -> wrapInSpan TraceRpcWatchMempoolSpan . watchMempoolMethod req)
     $ NoMoreMethods
 

cardano-rpc/src/Cardano/Rpc/Server/Internal/Tracing.hs

@@ -94,6 +94,8 @@ data TraceRpcSubmit
     TraceRpcEvalTxSpan TraceSpanEvent
   | -- | Mempool read span
     TraceRpcReadMempoolSpan TraceSpanEvent
+  | -- | Transaction confirmation wait span
+    TraceRpcWaitForTxSpan TraceSpanEvent
   | -- | Mempool watch span
     TraceRpcWatchMempoolSpan TraceSpanEvent
   deriving Show
@@ -106,6 +108,8 @@ instance Pretty TraceRpcSubmit where
     TraceRpcEvalTxSpan (SpanEnd _) -> "Finished eval tx method"
     TraceRpcReadMempoolSpan (SpanBegin _) -> "Started read mempool method"
     TraceRpcReadMempoolSpan (SpanEnd _) -> "Finished read mempool method"
+    TraceRpcWaitForTxSpan (SpanBegin _) -> "Started wait for tx method"
+    TraceRpcWaitForTxSpan (SpanEnd _) -> "Finished wait for tx method"
     TraceRpcWatchMempoolSpan (SpanBegin _) -> "Started watch mempool method"
     TraceRpcWatchMempoolSpan (SpanEnd _) -> "Finished watch mempool method"
     TraceRpcEvalTxDecodingError e -> "Failed to decode transaction for evaluation: " <> pshow e

cardano-rpc/src/Cardano/Rpc/Server/Internal/UtxoRpc/Mempool.hs

@@ -52,19 +52,16 @@ watchMempoolMethod request send = do
 -- | The @WatchMempool@ streaming loop. Emits every mempool entry newer than
 -- the last one seen, oldest first, filtered by the predicate.
 --
--- Removals and slot-only changes (no new ticket) produce no message: the
--- proto has no representation for eviction, and a max-ticket comparison
--- upstream would otherwise miss exactly these transitions (see
--- 'Cardano.Rpc.Server.NodeKernelAccess.nextMempoolWatchSnapshot').
+-- Removals produce no message: the proto has no representation for eviction.
 watchMempoolStream
   :: forall m
    . MonadIO m
   => m MempoolWatchSnapshot
   -- ^ Read the current mempool watch snapshot, without blocking - used
   -- once, for the initial baseline
-  -> (MempoolWatchSnapshot -> m MempoolWatchSnapshot)
-  -- ^ Block until the snapshot differs from the given one, then return the
-  -- new one
+  -> (Consensus.TicketNo -> m MempoolWatchSnapshot)
+  -- ^ Block until an entry newer than the given ticket exists, then return
+  -- the snapshot
   -> Proto U5c.TxPredicate
   -- ^ Predicate filtering which new entries are sent
   -> (NextElem (Proto U5c.WatchMempoolResponse) -> IO ())
@@ -75,7 +72,7 @@ watchMempoolStream readSnapshot nextSnapshot predicate send = do
   go Consensus.zeroTicketNo initial
  where
   go :: Consensus.TicketNo -> MempoolWatchSnapshot -> m ()
-  go lastSeenTicket snapshot@MempoolWatchSnapshot{mempoolWatchTxsAfter = txsAfter} = do
+  go lastSeenTicket MempoolWatchSnapshot{mempoolWatchTxsAfter = txsAfter} = do
     let newEntries = txsAfter lastSeenTicket
     forM_ newEntries $ \(txInMode, _ticketNo) ->
       forM_ (txInModeToTxInMempool txInMode) $ \txInMempool ->
@@ -85,4 +82,4 @@ watchMempoolStream readSnapshot nextSnapshot predicate send = do
     let lastSeenTicket' = case newEntries of
           [] -> lastSeenTicket
           _ -> snd (last newEntries)
-    nextSnapshot snapshot >>= go lastSeenTicket'
+    nextSnapshot lastSeenTicket' >>= go lastSeenTicket'

cardano-rpc/src/Cardano/Rpc/Server/Internal/UtxoRpc/Sync.hs

@@ -251,7 +251,7 @@ followTipStream
   :: forall m
    . HasCallStack
   => MonadIO m
-  => ChainFollower
+  => ChainFollower (ByteString, BlockInMode)
   -- ^ Follower to stream changes from
   -> m (Maybe (Proto U5c.BlockRef))
   -- ^ Read the current chain tip, projected into a @BlockRef@