fix unaccepted hold invoices in the accepted handler

holdinvoice_accepted_handler waited on the hold plugin track stream and
sent a HoldInvoiceAccepted notification afterwards, without checking the
reason the stream ended. When an invoice was cancelled, the stream just closes and the handler still sent
a spurious accepted notification.

Only send the notification if the stream actually reported the invoice
as accepted.

The hold plugin has no expired state: an invoice that expires without
ever being accepted stays Unpaid and the track stream never ends, so the
handler previously hung forever, holding a task and a gRPC stream open.

Pass the invoice's expiry into the handler and time out the track stream
at that point (capped), so it gives up on expired invoices and never
sends a spurious accepted notification. When re-spawning pending
invoices at startup, derive the expiry from the decoded bolt11.

Add test_hold_invoice_expiry which creates an invoice with a short
expiry and asserts the handler gives up without sending the notification.

src/main.rs

@@ -10,7 +10,13 @@ use cln_plugin::{
     Plugin,
     options::{ConfigOption, DefaultBooleanConfigOption, StringArrayConfigOption},
 };
-use cln_rpc::{ClnRpc, model::requests::ListdatastoreRequest};
+use cln_rpc::{
+    ClnRpc,
+    model::{
+        requests::{DecodeRequest, ListdatastoreRequest},
+        responses::DecodeType,
+    },
+};
 use nostr::nips::nip47;
 use nwc::run_nwc;
 use nwc_notifications::{payment_received_handler, payment_sent_handler};
@@ -225,8 +231,47 @@ async fn load_pending_hold_invoices(plugin: Plugin<PluginState>) -> Result<(), a
         .into_inner()
         .invoices;
 
+    let mut rpc = plugin.state().rpc_lock.lock().await;
+
     for invoice in invoices {
         if invoice.state() == InvoiceState::Accepted || invoice.state() == InvoiceState::Unpaid {
+            // Bound the accepted handler by the invoice's expiry. The hold
+            // plugin has no expired state, so without this it would wait
+            // forever on an invoice that expires while Unpaid.
+            let invoice_decoded = rpc
+                .call_typed(&DecodeRequest {
+                    string: invoice.invoice.clone(),
+                })
+                .await?;
+
+            if !invoice_decoded.valid {
+                log::warn!(
+                    "Skipping hold invoice {}, could not decode it",
+                    hex::encode(&invoice.payment_hash)
+                );
+                continue;
+            }
+
+            let (created_at, expiry) = match invoice_decoded.item_type {
+                DecodeType::BOLT12_INVOICE => (
+                    invoice_decoded.invoice_created_at,
+                    invoice_decoded.invoice_relative_expiry.map(u64::from),
+                ),
+                DecodeType::BOLT11_INVOICE => {
+                    (invoice_decoded.created_at, invoice_decoded.expiry)
+                }
+                _ => continue,
+            };
+
+            let (Some(created_at), Some(expiry)) = (created_at, expiry) else {
+                log::warn!(
+                    "Skipping hold invoice {}: missing creation time or expiry",
+                    hex::encode(&invoice.payment_hash)
+                );
+                continue;
+            };
+            let expires_at = created_at.saturating_add(expiry);
+
             log::debug!(
                 "Starting holdinvoice accepted handler for {}",
                 hex::encode(&invoice.payment_hash)
@@ -234,6 +279,7 @@ async fn load_pending_hold_invoices(plugin: Plugin<PluginState>) -> Result<(), a
             tokio::spawn(holdinvoice_accepted_handler(
                 plugin.clone(),
                 invoice.payment_hash,
+                expires_at,
             ));
         }
     }

src/nwc_hold.rs

@@ -107,6 +107,8 @@ async fn make_hold_invoice(
         })?
         .into_inner();
 
+    let expires_at = Timestamp::now() + expiry;
+
     let response = nip47::MakeHoldInvoiceResponse {
         invoice: Some(holdinvoice.bolt11),
         transaction_type: nip47::TransactionType::Incoming,
@@ -114,12 +116,16 @@ async fn make_hold_invoice(
         description_hash: params.description_hash,
         amount: params.amount,
         created_at: Timestamp::now(),
-        expires_at: Timestamp::now() + expiry,
+        expires_at,
         metadata: None,
         payment_hash: params.payment_hash,
     };
 
-    tokio::spawn(holdinvoice_accepted_handler(plugin, payment_hash));
+    tokio::spawn(holdinvoice_accepted_handler(
+        plugin,
+        payment_hash,
+        expires_at.as_secs(),
+    ));
     Ok(response)
 }
 

src/nwc_notifications.rs

@@ -28,6 +28,11 @@ use crate::{
     structs::{NOT_INV_ERR, PluginState, WalletService},
 };
 
+/// Upper bound on how long a `holdinvoice_accepted_handler` waits for an
+/// invoice to be accepted, so a client cannot pin a task and a gRPC stream
+/// open forever with an absurd expiry.
+const MAX_HOLD_WAIT_SECS: u64 = 24 * 60 * 60;
+
 pub async fn payment_received_handler(
     plugin: Plugin<PluginState>,
     args: serde_json::Value,
@@ -396,6 +401,7 @@ async fn send_notification(
 pub async fn holdinvoice_accepted_handler(
     plugin: Plugin<PluginState>,
     payment_hash: Vec<u8>,
+    expires_at: u64,
 ) -> Result<(), anyhow::Error> {
     let mut hold_client = plugin.state().hold_client.lock().clone().unwrap();
 
@@ -404,13 +410,51 @@ pub async fn holdinvoice_accepted_handler(
     };
     let mut track_stream = hold_client.track(track_request).await?.into_inner();
 
-    while let Some(response) = track_stream.message().await? {
-        log::debug!("Invoice status: {}", response.state().as_str_name());
-        if response.state() == InvoiceState::Accepted {
-            break;
+    // The hold plugin has no expired state: an invoice that expires without
+    // being accepted stays in the Unpaid state and the track stream never ends
+    // on its own. Bound the wait by the invoice's expiry (capped) so we do not
+    // hold this task and gRPC stream open forever.
+    let wait_duration = expires_at
+        .saturating_sub(Timestamp::now().as_secs())
+        .min(MAX_HOLD_WAIT_SECS);
+    let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(wait_duration);
+
+    let mut accepted = false;
+    loop {
+        match tokio::time::timeout_at(deadline, track_stream.message()).await {
+            Err(_elapsed) => {
+                log::debug!(
+                    "Hold invoice {} expired before being accepted",
+                    hex::encode(&payment_hash)
+                );
+                break;
+            }
+            Ok(Err(e)) => return Err(e.into()),
+            // The stream ended without an Accepted state: the invoice was
+            // cancelled before it ever got accepted.
+            Ok(Ok(None)) => break,
+            Ok(Ok(Some(response))) => {
+                log::debug!("Invoice status: {}", response.state().as_str_name());
+                match response.state() {
+                    InvoiceState::Accepted => {
+                        accepted = true;
+                        break;
+                    }
+                    InvoiceState::Paid | InvoiceState::Cancelled => break,
+                    InvoiceState::Unpaid => (),
+                }
+            }
         }
     }
 
+    if !accepted {
+        log::debug!(
+            "Hold invoice {} was not accepted, skipping notification",
+            hex::encode(&payment_hash)
+        );
+        return Ok(());
+    }
+
     let list_request = ListRequest {
         constraint: Some(Constraint::PaymentHash(payment_hash.clone())),
     };

tests/test_cln-nip47.py

@@ -1294,9 +1294,10 @@ async def test_hold_invoice(
                 "log-level": "debug",
                 "plugin": get_plugin,
                 "important-plugin": get_hold,
+                "hold-grpc-port": node_factory.get_unused_port(),
                 "nip47-relays": url,
                 "may_reconnect": True,
-                "broken_log": r"Relay receiver exited with error",
+                "broken_log": r"Relay receiver exited with error|Connection failed",
             },
         ],
     )
@@ -1824,3 +1825,80 @@ async def test_hold_invoice(
         ):
             hold_events.append(content)
     assert len(hold_events) == 1
+
+
+@pytest.mark.asyncio
+async def test_hold_invoice_expiry(
+    node_factory,
+    get_plugin,  # noqa: F811
+    get_hold,  # noqa: F811
+    nostr_relay,
+):
+    url = nostr_relay
+    l2 = node_factory.get_node(
+        options={
+            "log-level": "debug",
+            "plugin": get_plugin,
+            "important-plugin": get_hold,
+            "hold-grpc-port": node_factory.get_unused_port(),
+            "nip47-relays": url,
+        },
+        broken_log=r"Relay receiver exited with error|Connection failed",
+    )
+    uri_res = l2.rpc.call("nip47-create", ["test1", 3010])
+    uri_str = uri_res["uri"]
+    client_pubkey = PublicKey.parse(uri_res["clientkey_public"])
+    LOGGER.info(uri_str)
+    uri = NostrWalletConnectUri.parse(uri_str)
+
+    preimage = secrets.token_hex(32)
+    payment_hash = hashlib.sha256(bytes.fromhex(preimage)).hexdigest()
+    LOGGER.info(f"preimage: {preimage}")
+    LOGGER.info(f"payment_hash: {payment_hash}")
+    content = {
+        "method": "make_hold_invoice",
+        "params": {
+            "amount": 5000,
+            "payment_hash": payment_hash,
+            "expiry": 5,
+        },
+    }
+    content = json.dumps(content)
+    keys = Keys(uri.secret())
+    encrypted_content = keys.nip04_encrypt(uri.public_key(), content)
+    event = (
+        await EventBuilder(Kind(23194), encrypted_content)
+        .tags([Tag.public_key(uri.public_key())])
+        .finalize_async(keys)
+    )
+    client = Client()
+    relay_url = RelayUrl.parse(url)
+    await client.add_relay(relay_url)
+    await client.connect()
+
+    await fetch_info_event(client, uri)
+
+    (responses1, _res) = await fetch_event_responses(
+        client, client_pubkey, 23195, client.send_event(event), 1
+    )
+    error_events = []
+    success_events = []
+    for event in responses1:
+        LOGGER.info(event)
+        content = keys.nip04_decrypt(uri.public_key(), event.content())
+        content = json.loads(content)
+        LOGGER.info(content)
+        if "result" in content and content["result"] is not None:
+            success_events.append(content)
+        if "error" in content and content["error"] is not None:
+            error_events.append(content)
+
+    assert len(success_events) == 1
+    assert len(error_events) == 0
+
+    # The hold plugin has no expired state: an invoice that expires without
+    # ever being accepted (and without being cancelled) stays `Unpaid` and the
+    # track stream never ends on its own. The handler must give up at the
+    # invoice's expiry instead of hanging, and must not send a spurious
+    # accepted notification.
+    l2.daemon.wait_for_log("was not accepted, skipping notification", timeout=20)