fix: keep Goat RPC and the network loop out of SQLite write locks

crates/client/src/graphs/graph_query.rs

@@ -455,7 +455,7 @@ pub struct UserInput {
     pub amount_sats: String,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct BridgeInRequestEvent {
     pub id: String,
     #[serde(rename = "transactionHash")]

crates/engine/src/metrics.rs

@@ -7,6 +7,8 @@ use prometheus_client::metrics::family::Family;
 use prometheus_client::metrics::gauge::Gauge;
 use prometheus_client::metrics::histogram::{Histogram, exponential_buckets};
 use prometheus_client::registry::Registry;
+use std::collections::HashSet;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
@@ -38,6 +40,9 @@ pub struct EngineMetrics {
     outbox_publish_total: Family<OutcomeLabels, Counter>,
     job_runs_total: Family<KindOutcomeLabels, Counter>,
     rows: Family<TableStateLabels, Gauge>,
+    /// Every table/state the row gauge has reported, so a state that empties
+    /// is set back to zero.
+    known_rows: Arc<Mutex<HashSet<TableStateLabels>>>,
     acks_dropped_total: Counter,
     poisoned_total: Counter,
 }
@@ -58,6 +63,7 @@ impl EngineMetrics {
             outbox_publish_total: Family::default(),
             job_runs_total: Family::default(),
             rows: Family::default(),
+            known_rows: Arc::default(),
             acks_dropped_total: Counter::default(),
             poisoned_total: Counter::default(),
         }
@@ -132,10 +138,21 @@ impl EngineMetrics {
             .inc();
     }
 
-    pub fn set_rows(&self, table: &str, state: &str, count: i64) {
-        self.rows
-            .get_or_create(&TableStateLabels { table: table.to_string(), state: state.to_string() })
-            .set(count);
+    /// Publish one refresh of the row counts. The count query only returns
+    /// states that have rows: a state it no longer returns goes to zero
+    /// instead of keeping its last value.
+    pub fn set_row_counts<'a>(&self, counts: impl IntoIterator<Item = (&'a str, &'a str, i64)>) {
+        let mut known = self.known_rows.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let mut seen = HashSet::new();
+        for (table, state, count) in counts {
+            let labels = TableStateLabels { table: table.to_string(), state: state.to_string() };
+            self.rows.get_or_create(&labels).set(count);
+            known.insert(labels.clone());
+            seen.insert(labels);
+        }
+        for labels in known.iter().filter(|labels| !seen.contains(*labels)) {
+            self.rows.get_or_create(labels).set(0);
+        }
     }
 
     pub fn record_ack_dropped(&self) {
@@ -146,3 +163,26 @@ impl EngineMetrics {
         self.poisoned_total.inc();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn rows(metrics: &EngineMetrics, table: &str, state: &str) -> i64 {
+        metrics
+            .rows
+            .get_or_create(&TableStateLabels { table: table.to_string(), state: state.to_string() })
+            .get()
+    }
+
+    #[test]
+    fn a_state_that_empties_goes_back_to_zero() {
+        let metrics = EngineMetrics::new();
+        metrics.set_row_counts([("job", "pending", 3), ("job", "running", 1)]);
+        metrics.set_row_counts([("job", "pending", 2)]);
+        assert_eq!(rows(&metrics, "job", "pending"), 2);
+        assert_eq!(rows(&metrics, "job", "running"), 0);
+        metrics.set_row_counts([]);
+        assert_eq!(rows(&metrics, "job", "pending"), 0);
+    }
+}

crates/engine/src/runtime.rs

@@ -436,11 +436,11 @@ async fn run_recheck(inner: Arc<EngineInner>, cancel: CancellationToken) {
                     }
                 }
                 match storage.engine_state_counts().await {
-                    Ok(counts) => {
-                        for count in counts {
-                            inner.metrics.set_rows(&count.table, &count.state, count.count);
-                        }
-                    }
+                    Ok(counts) => inner.metrics.set_row_counts(
+                        counts
+                            .iter()
+                            .map(|count| (count.table.as_str(), count.state.as_str(), count.count)),
+                    ),
                     Err(error) => {
                         warn!(event = "engine_recheck", error = %error, "row count refresh failed")
                     }

node/src/middleware/publisher.rs

@@ -6,7 +6,7 @@ use futures::future::BoxFuture;
 use libp2p::request_response::{OutboundRequestId, ResponseChannel};
 use libp2p::{
     PeerId, Swarm,
-    gossipsub::{IdentTopic, MessageId},
+    gossipsub::{self, IdentTopic, MessageId},
 };
 use std::collections::HashMap;
 use tokio::sync::{mpsc, oneshot};
@@ -65,6 +65,13 @@ pub enum PublishCommand {
         channel: ResponseChannel<GetGraphResponse>,
         response: GetGraphResponse,
     },
+    /// The forwarding verdict on a gossip message whose inbox write ran off the
+    /// network loop; only the swarm can report it.
+    GossipValidation {
+        id: MessageId,
+        propagation_source: PeerId,
+        acceptance: gossipsub::MessageAcceptance,
+    },
 }
 
 impl PublishCommand {
@@ -88,6 +95,14 @@ impl PublishCommand {
                 // The requester may have given up meanwhile; nothing to do then.
                 let _ = swarm.behaviour_mut().get_graph.send_response(channel, response);
             }
+            PublishCommand::GossipValidation { id, propagation_source, acceptance } => {
+                crate::p2p_msg_handler::report_gossip_validation(
+                    swarm,
+                    &id,
+                    &propagation_source,
+                    acceptance,
+                );
+            }
         }
     }
 }
@@ -119,6 +134,21 @@ impl NetworkPublisher {
             .map_err(|_| anyhow::anyhow!("network publisher stopped"))
     }
 
+    /// Hand the swarm the verdict on a gossip message recorded off the loop.
+    pub async fn report_validation(
+        &self,
+        id: MessageId,
+        propagation_source: PeerId,
+        acceptance: gossipsub::MessageAcceptance,
+    ) -> anyhow::Result<()> {
+        let send =
+            self.0.send(PublishCommand::GossipValidation { id, propagation_source, acceptance });
+        tokio::time::timeout(PUBLISH_TIMEOUT, send)
+            .await
+            .map_err(|_| anyhow::anyhow!("network task did not take the gossip verdict in time"))?
+            .map_err(|_| anyhow::anyhow!("network publisher stopped"))
+    }
+
     /// Ask `peer` for a graph definition over the direct protocol.
     pub async fn request_graph(
         &self,

node/src/p2p_admission/mod.rs

@@ -83,6 +83,9 @@ pub enum DropReason {
     /// room, by rows or by bytes. Distinct from `ImmediateRateLimited`: that one
     /// is a sender exceeding its rate, this one is the node falling behind.
     ControlQueueFull,
+    /// The bounded queue in front of the inbox writer had no room, by rows or
+    /// by bytes: the database is behind. The sender's outbox sends again.
+    InboxQueueFull,
     SenderQuota,
     GlobalQuota,
     /// Not a message drop: a `NodeInfo` from a sender that is not registered was
@@ -114,6 +117,7 @@ impl DropReason {
             Self::UnexpectedBinaryKind => "unexpected_binary_kind",
             Self::ImmediateRateLimited => "immediate_rate_limited",
             Self::ControlQueueFull => "control_queue_full",
+            Self::InboxQueueFull => "inbox_queue_full",
             Self::SenderQuota => "sender_quota",
             Self::GlobalQuota => "global_quota",
             Self::NodeTableFull => "node_table_full",
@@ -137,6 +141,7 @@ impl DropReason {
                 | Self::ForwardRate
                 | Self::ImmediateRateLimited
                 | Self::ControlQueueFull
+                | Self::InboxQueueFull
                 | Self::SenderQuota
                 | Self::GlobalQuota
         )