Merge main into feat/forwarding-array-upstream

Resolves conflict in src/ctx.rs — both sides added independent tokio
tests (forwarding fail-over on this branch, default-pool upstream path
on main from #103). Keep both.

src/api.rs

@@ -201,6 +201,7 @@ struct LanStatsResponse {
 struct QueriesStats {
     total: u64,
     forwarded: u64,
+    upstream: u64,
     recursive: u64,
     coalesced: u64,
     cached: u64,
@@ -548,6 +549,7 @@ async fn stats(State(ctx): State<Arc<ServerCtx>>) -> Json<StatsResponse> {
         queries: QueriesStats {
             total: snap.total,
             forwarded: snap.forwarded,
+            upstream: snap.upstream,
             recursive: snap.recursive,
             coalesced: snap.coalesced,
             cached: snap.cached,

src/ctx.rs

@@ -266,7 +266,7 @@ pub async fn resolve_query(
                 .await
                 {
                     Ok(resp_wire) => match cache_and_parse(ctx, &qname, qtype, &resp_wire) {
-                        Ok(resp) => (resp, QueryPath::Forwarded, DnssecStatus::Indeterminate),
+                        Ok(resp) => (resp, QueryPath::Upstream, DnssecStatus::Indeterminate),
                         Err(e) => {
                             error!("{} | {:?} {} | PARSE ERROR | {}", src_addr, qtype, qname, e);
                             (
@@ -1293,4 +1293,29 @@ mod tests {
             other => panic!("expected A record, got {:?}", other),
         }
     }
+
+    #[tokio::test]
+    async fn pipeline_default_pool_reports_upstream_path() {
+        let mut upstream_resp = DnsPacket::new();
+        upstream_resp.header.response = true;
+        upstream_resp.header.rescode = ResultCode::NOERROR;
+        upstream_resp.answers.push(DnsRecord::A {
+            domain: "example.com".to_string(),
+            addr: Ipv4Addr::new(93, 184, 216, 34),
+            ttl: 300,
+        });
+        let upstream_addr = crate::testutil::mock_upstream(upstream_resp).await;
+
+        let ctx = crate::testutil::test_ctx().await;
+        ctx.upstream_pool
+            .lock()
+            .unwrap()
+            .set_primary(vec![Upstream::Udp(upstream_addr)]);
+        let ctx = Arc::new(ctx);
+
+        let (resp, path) = resolve_in_test(&ctx, "example.com", QueryType::A).await;
+        assert_eq!(path, QueryPath::Upstream);
+        assert_eq!(resp.header.rescode, ResultCode::NOERROR);
+        assert_eq!(resp.answers.len(), 1);
+    }
 }

src/stats.rs

@@ -90,6 +90,7 @@ fn linux_rss() -> usize {
 pub struct ServerStats {
     queries_total: u64,
     queries_forwarded: u64,
+    queries_upstream: u64,
     queries_recursive: u64,
     queries_coalesced: u64,
     queries_cached: u64,
@@ -127,7 +128,10 @@ impl Transport {
 pub enum QueryPath {
     Local,
     Cached,
+    /// Matched a `[[forwarding]]` suffix rule.
     Forwarded,
+    /// Resolved via the default `[upstream]` pool (no suffix match).
+    Upstream,
     Recursive,
     Coalesced,
     Blocked,
@@ -141,6 +145,7 @@ impl QueryPath {
             QueryPath::Local => "LOCAL",
             QueryPath::Cached => "CACHED",
             QueryPath::Forwarded => "FORWARD",
+            QueryPath::Upstream => "UPSTREAM",
             QueryPath::Recursive => "RECURSIVE",
             QueryPath::Coalesced => "COALESCED",
             QueryPath::Blocked => "BLOCKED",
@@ -156,6 +161,8 @@ impl QueryPath {
             Some(QueryPath::Cached)
         } else if s.eq_ignore_ascii_case("FORWARD") {
             Some(QueryPath::Forwarded)
+        } else if s.eq_ignore_ascii_case("UPSTREAM") {
+            Some(QueryPath::Upstream)
         } else if s.eq_ignore_ascii_case("RECURSIVE") {
             Some(QueryPath::Recursive)
         } else if s.eq_ignore_ascii_case("COALESCED") {
@@ -183,6 +190,7 @@ impl ServerStats {
         ServerStats {
             queries_total: 0,
             queries_forwarded: 0,
+            queries_upstream: 0,
             queries_recursive: 0,
             queries_coalesced: 0,
             queries_cached: 0,
@@ -204,6 +212,7 @@ impl ServerStats {
             QueryPath::Local => self.queries_local += 1,
             QueryPath::Cached => self.queries_cached += 1,
             QueryPath::Forwarded => self.queries_forwarded += 1,
+            QueryPath::Upstream => self.queries_upstream += 1,
             QueryPath::Recursive => self.queries_recursive += 1,
             QueryPath::Coalesced => self.queries_coalesced += 1,
             QueryPath::Blocked => self.queries_blocked += 1,
@@ -232,6 +241,7 @@ impl ServerStats {
             uptime_secs: self.uptime_secs(),
             total: self.queries_total,
             forwarded: self.queries_forwarded,
+            upstream: self.queries_upstream,
             recursive: self.queries_recursive,
             coalesced: self.queries_coalesced,
             cached: self.queries_cached,
@@ -253,10 +263,11 @@ impl ServerStats {
         let secs = uptime.as_secs() % 60;
 
         log::info!(
-            "STATS | uptime {}h{}m{}s | total {} | fwd {} | recursive {} | coalesced {} | cached {} | local {} | override {} | blocked {} | errors {}",
+            "STATS | uptime {}h{}m{}s | total {} | fwd {} | upstream {} | recursive {} | coalesced {} | cached {} | local {} | override {} | blocked {} | errors {}",
             hours, mins, secs,
             self.queries_total,
             self.queries_forwarded,
+            self.queries_upstream,
             self.queries_recursive,
             self.queries_coalesced,
             self.queries_cached,
@@ -272,6 +283,7 @@ pub struct StatsSnapshot {
     pub uptime_secs: u64,
     pub total: u64,
     pub forwarded: u64,
+    pub upstream: u64,
     pub recursive: u64,
     pub coalesced: u64,
     pub cached: u64,